Logan
7e1b01a275
### Firestore read reductions
**1. `doc_get_cached()` in `firestore.py` — new 5-min TTL cache**
One place, benefits everything. System and node config documents almost never change during a monitoring session.
**2. System doc: 4 reads → 1 per call**
| Before | After |
|---|---|
| `upload.py` — `doc_get("systems")` for ai_flags | `doc_get_cached` |
| `transcription.py` — `get_vocabulary()` → `doc_get("systems")` | cache hit |
| `intelligence.py` — `get_vocabulary()` → `doc_get("systems")` | cache hit |
| `intelligence.py` — `doc_get("systems")` again for ten_codes | eliminated (reads same cached doc) |
**3. Node doc: cached in `_on_call_start` and `intelligence.py`**
The node is read every call event to get `assigned_system_id` and lat/lon for geocoding. Both now use the cache — node assignments and positions essentially never change at runtime.
**4. Node sweeper: 30s → 90s interval**
The sweeper was doing a full node collection scan 3× more often than necessary — the offline threshold is already 90s. Cuts sweeper reads by 66%.
**5. Vocabulary induction: scans all-time calls → last 7 days**
Previously fetched every ended call for a system (could be thousands). Now scoped to the last 7 days.
> **Note:** The vocabulary induction query `(system_id == X, ended_at >= cutoff)` needs a Firestore
> composite index on `(system_id ASC, ended_at ASC)`. When the induction loop first fires it will log
> an error with a Firebase Console link to create it in one click.
58 lines
1.8 KiB
Python
58 lines
1.8 KiB
Python
import asyncio
|
|
from datetime import datetime, timezone, timedelta
|
|
from app.config import settings
|
|
from app.internal.logger import logger
|
|
from app.internal import firestore as fstore
|
|
|
|
SWEEP_INTERVAL = 90 # seconds — matches node_offline_threshold; no gain in checking faster
|
|
|
|
|
|
async def sweeper_loop():
|
|
"""
|
|
Periodically check for nodes that haven't checked in recently
|
|
and mark them offline in Firestore.
|
|
"""
|
|
logger.info("Node sweeper started.")
|
|
while True:
|
|
await asyncio.sleep(SWEEP_INTERVAL)
|
|
try:
|
|
await _sweep()
|
|
except Exception as e:
|
|
logger.error(f"Sweeper error: {e}")
|
|
|
|
|
|
async def _sweep():
|
|
threshold = datetime.now(timezone.utc) - timedelta(seconds=settings.node_offline_threshold)
|
|
|
|
def _query():
|
|
from app.internal.firestore import db
|
|
return [
|
|
doc.to_dict()
|
|
for doc in db.collection("nodes").stream()
|
|
]
|
|
|
|
nodes = await asyncio.to_thread(_query)
|
|
for node in nodes:
|
|
status = node.get("status", "offline")
|
|
if status == "offline":
|
|
continue
|
|
|
|
last_seen_raw = node.get("last_seen")
|
|
if not last_seen_raw:
|
|
continue
|
|
|
|
# last_seen may be a Firestore Timestamp, a datetime, or an ISO string
|
|
if isinstance(last_seen_raw, str):
|
|
last_seen = datetime.fromisoformat(last_seen_raw)
|
|
else:
|
|
last_seen = last_seen_raw
|
|
if last_seen.tzinfo is None:
|
|
last_seen = last_seen.replace(tzinfo=timezone.utc)
|
|
|
|
if last_seen < threshold:
|
|
node_id = node.get("node_id")
|
|
await fstore.doc_update("nodes", node_id, {"status": "offline"})
|
|
logger.info(f"Node {node_id} marked offline (last seen: {last_seen.isoformat()})")
|
|
from app.routers.tokens import release_token
|
|
await release_token(node_id)
|