Spaces:
Running
Running
File size: 4,786 Bytes
b7c46f4 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 | """
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),
}
|