Spaces:
Running
Running
| """ | |
| Layer 1 — stale awareness. Answers one question: is every source and every | |
| handoff where it should be right now? It reads; it never fixes anything. | |
| A source is judged against its own rhythm (registry.py), on two clocks: | |
| - fetch clock: when did we last successfully fetch it? | |
| - data clock: how new is the newest period it gave us, allowing for the | |
| source's legitimate latency? | |
| A fetch that succeeds daily but keeps returning the same old period is stale on | |
| the data clock even though the fetch clock looks fine. | |
| """ | |
| import sqlite3 | |
| from datetime import datetime, timedelta, timezone | |
| from intel.db import now_iso | |
| from intel.handoff import ACTIVE_LAYERS | |
| # A source is 'late' after 1.5× its cadence and 'stale' after 3×. Generous on | |
| # purpose: a monthly source a few days late is normal, three months late isn't. | |
| LATE_FACTOR = 1.5 | |
| STALE_FACTOR = 3.0 | |
| # A handoff to an active layer that sits unreceived this long is overdue. | |
| HANDOFF_OVERDUE = timedelta(hours=24) | |
| STATUS_ORDER = ["never", "failing", "stale", "late", "fresh"] | |
| def _parse(ts: str | None) -> datetime | None: | |
| if not ts: | |
| return None | |
| dt = datetime.fromisoformat(ts) | |
| return dt if dt.tzinfo else dt.replace(tzinfo=timezone.utc) | |
| def record_attempt(con: sqlite3.Connection, source_id: str, ok: bool, | |
| latest_data_period: str | None = None, error: str | None = None) -> None: | |
| """Every ingester calls this once per run, success or failure.""" | |
| now = now_iso() | |
| if ok: | |
| con.execute( | |
| """update sources set last_attempt_at = ?, last_success_at = ?, last_error = null, | |
| latest_data_period = coalesce(?, latest_data_period) where id = ?""", | |
| (now, now, latest_data_period, source_id), | |
| ) | |
| else: | |
| con.execute("update sources set last_attempt_at = ?, last_error = ? where id = ?", (now, error, source_id)) | |
| con.commit() | |
| def _grade(age: timedelta, cadence: timedelta) -> str: | |
| if age > cadence * STALE_FACTOR: | |
| return "stale" | |
| if age > cadence * LATE_FACTOR: | |
| return "late" | |
| return "fresh" | |
| def source_status(row: sqlite3.Row, now: datetime) -> dict: | |
| cadence = timedelta(hours=row["expected_cadence_hours"]) | |
| latency = timedelta(hours=row["latency_hours"]) | |
| last_success = _parse(row["last_success_at"]) | |
| last_attempt = _parse(row["last_attempt_at"]) | |
| out = { | |
| "source": row["id"], "family": row["family"], | |
| "last_success_at": row["last_success_at"], "latest_data_period": row["latest_data_period"], | |
| "last_error": row["last_error"], | |
| } | |
| if last_success is None: | |
| out["status"] = "failing" if last_attempt else "never" | |
| out["reason"] = f"never succeeded: {row['last_error']}" if last_attempt else "never fetched" | |
| return out | |
| fetch_grade = _grade(now - last_success, cadence) | |
| data_grade = "fresh" | |
| period = _parse(row["latest_data_period"]) | |
| if period is not None: | |
| data_grade = _grade(max(now - period - latency, timedelta(0)), cadence) | |
| # A success clears last_error, so an error present means the latest attempt | |
| # failed — the source is failing now, even if its last good data is still | |
| # within cadence. (Comparing timestamps instead misses a failure in the same | |
| # second as the success.) | |
| failing = bool(row["last_error"]) | |
| grade = min(fetch_grade, data_grade, key=STATUS_ORDER.index) | |
| if failing: | |
| out["status"], out["reason"] = "failing", f"last attempt failed: {row['last_error']}" | |
| else: | |
| out["status"] = grade | |
| if grade == "fresh": | |
| out["reason"] = "on schedule" | |
| elif STATUS_ORDER.index(data_grade) < STATUS_ORDER.index(fetch_grade): | |
| out["reason"] = f"fetched fine but newest data is {row['latest_data_period']}" | |
| else: | |
| out["reason"] = f"last successful fetch {row['last_success_at']}" | |
| return out | |
| def report(con: sqlite3.Connection, now: datetime | None = None) -> dict: | |
| now = now or datetime.now(timezone.utc) | |
| sources = [source_status(r, now) for r in con.execute("select * from sources order by id")] | |
| overdue, queued = [], [] | |
| for h in con.execute("select id, run_id, from_layer, to_layer, produced_at from handoffs where received_at is null"): | |
| item = dict(h) | |
| if h["to_layer"] not in ACTIVE_LAYERS: | |
| queued.append(item) | |
| elif now - _parse(h["produced_at"]) > HANDOFF_OVERDUE: | |
| overdue.append(item) | |
| healthy = all(s["status"] == "fresh" for s in sources) and not overdue | |
| return { | |
| "checked_at": now.isoformat(timespec="seconds"), | |
| "healthy": healthy, | |
| "sources": sources, | |
| "handoffs_overdue": overdue, | |
| "handoffs_queued_for_unbuilt_layers": len(queued), | |
| } | |