Report the notifier's health by heartbeat, not per-worker threads (#166)

The ops dashboard marked subscription-notifier DOWN ~2/3 of the time after the
notifier was gated to a single leader worker. It judged the daemon by whether
its thread was present on whichever worker answered /api/v2/metrics, but only
the leader runs it — so polls landing on the other two workers saw no thread and
reported DOWN for a perfectly healthy notifier. Per-worker thread checks can't
describe a single-leader daemon, and worse, they could no longer distinguish a
real crash from a normal non-leader.

Report it by liveness instead: the notifier stamps a heartbeat (timestamp +
expected interval) into the shared metrics store each loop iteration, including
once at startup so the dashboard shows it up immediately after a restart. Since
the beat lives in the shared SQLite store, any worker's metrics response carries
it, so the dashboard sees the notifier's health whichever worker it polls.

- metrics.py: both stores record/expose heartbeats (SQLite reuses the meta
  table, so every worker's beats share one file); snapshot() converts each beat
  to a server-computed age so readers needn't reconcile clock skew.
- notify.py: beat at loop start and every wake.
- dashboard.py: judge heartbeat daemons by freshness (up to 2 missed intervals
  before DOWN, with beat age shown); neighbor-warmer keeps its per-worker thread
  check, which is correct since every worker runs it.
- Tests: heartbeat store/snapshot shape + cross-worker sharing; dashboard
  fresh/stale/missing rendering and the unchanged warmer check. 194 passed.

Co-authored-by: root <root@vmi3417050.contaboserver.net>

metrics.py

@@ -126,6 +126,9 @@ class _MemStore:
         # epoch-minute -> {key: count}, pruned as minutes roll out of the window.
         self._recent_in: "dict[int, dict[str, int]]" = {}
         self._recent_out: "dict[int, dict[str, int]]" = {}
+        # name -> {"ts": last-beat epoch, "interval_s": expected cadence}. Lets the
+        # dashboard judge a background daemon by liveness, not per-worker thread presence.
+        self._heartbeats: "dict[str, dict[str, float]]" = {}
 
     def _bump_recent(self, buckets, key) -> None:
         """Count one event for ``key`` in the current-minute bucket (caller holds lock)."""
@@ -166,6 +169,10 @@ class _MemStore:
             row[outcome] = row.get(outcome, 0) + 1
             self._bump_recent(self._recent_out, source)
 
+    def record_heartbeat(self, name, ts, interval_s) -> None:
+        with self._lock:
+            self._heartbeats[name] = {"ts": ts, "interval_s": interval_s}
+
     def read(self) -> dict:
         with self._lock:
             inbound = {k: dict(v) for k, v in self._inbound.items()}
@@ -176,6 +183,7 @@ class _MemStore:
                 "outbound": outbound,
                 "inbound_recent": self._recent_totals(self._recent_in),
                 "outbound_recent": self._recent_totals(self._recent_out),
+                "heartbeats": {k: dict(v) for k, v in self._heartbeats.items()},
             }
 
 
@@ -258,6 +266,30 @@ class _SqliteStore:
             )
             self._prune(minute)
 
+    def record_heartbeat(self, name, ts, interval_s) -> None:
+        # Reuse the meta table (two keys per daemon) so every worker's beats land in the
+        # one shared file — the dashboard then sees the notifier's liveness whichever
+        # worker answers its poll, even though only the leader worker runs it.
+        with self._lock:
+            self._db.execute(
+                "INSERT INTO meta(key, value) VALUES(?, ?) "
+                "ON CONFLICT(key) DO UPDATE SET value = excluded.value",
+                (f"hb_ts:{name}", ts),
+            )
+            self._db.execute(
+                "INSERT INTO meta(key, value) VALUES(?, ?) "
+                "ON CONFLICT(key) DO UPDATE SET value = excluded.value",
+                (f"hb_int:{name}", interval_s),
+            )
+
+    def _read_heartbeats(self) -> "dict[str, dict[str, float]]":
+        beats: "dict[str, dict[str, float]]" = {}
+        for key, value in self._db.execute(
+                "SELECT key, value FROM meta WHERE key LIKE 'hb_ts:%' OR key LIKE 'hb_int:%'"):
+            field, _, name = key.partition(":")
+            beats.setdefault(name, {})["ts" if field == "hb_ts" else "interval_s"] = value
+        return beats
+
     def read(self) -> dict:
         cutoff = _now_minute() - WINDOW_MINUTES + 1
         with self._lock:
@@ -286,6 +318,7 @@ class _SqliteStore:
                 "outbound": outbound,
                 "inbound_recent": recent_in,
                 "outbound_recent": recent_out,
+                "heartbeats": self._read_heartbeats(),
             }
 
 
@@ -325,16 +358,33 @@ def record_outbound(phase: str, outcome: str) -> None:
         pass
 
 
+def record_heartbeat(name: str, interval_s: float) -> None:
+    """Stamp ``name``'s liveness. ``interval_s`` is the daemon's expected beat cadence, so
+    a reader can tell fresh from stale without knowing the schedule. Best-effort."""
+    try:
+        _store.record_heartbeat(name, time.time(), interval_s)
+    except Exception:  # noqa: BLE001 - instrumentation must never break the daemon
+        pass
+
+
 def snapshot() -> dict:
     """A JSON-serializable copy of all counters plus process facts."""
     try:
         data = _store.read()
     except Exception:  # noqa: BLE001 - a locked/broken store must not 500 the dashboard
         data = {"started_at": time.time(), "inbound": {}, "outbound": {},
-                "inbound_recent": {}, "outbound_recent": {}}
+                "inbound_recent": {}, "outbound_recent": {}, "heartbeats": {}}
     inbound = data["inbound"]
     outbound = data["outbound"]
     started_at = data["started_at"]
+    # Convert each heartbeat's absolute timestamp to an age here, on the server clock, so
+    # a reader (dashboard) never has to reconcile clock skew to judge freshness.
+    now = time.time()
+    heartbeats = {
+        name: {"age_s": round(now - hb["ts"], 1), "interval_s": hb.get("interval_s")}
+        for name, hb in (data.get("heartbeats") or {}).items()
+        if hb.get("ts") is not None
+    }
     return {
         "pid": os.getpid(),
         "started_at": started_at,
@@ -347,4 +397,5 @@ def snapshot() -> dict:
         "window_minutes": WINDOW_MINUTES,
         "inbound_recent": data["inbound_recent"],
         "outbound_recent": data["outbound_recent"],
+        "heartbeats": heartbeats,
     }

notify.py

@@ -32,6 +32,7 @@ from sqlalchemy import delete, select
 import climate
 import grading
 import grid
+import metrics
 import push
 from db import sync_session_maker
 from models import AccessToken, Notification, PushSubscription, Subscription
@@ -291,8 +292,14 @@ def run_pass() -> int:
 
 
 def run_loop():
+    # Beat once at startup so the dashboard shows the notifier alive immediately after a
+    # (re)start, without waiting a full INTERVAL for the first pass. Then beat every wake:
+    # healthy => a fresh beat at most INTERVAL apart, which is how the dashboard tells this
+    # single-leader daemon apart from a genuinely dead one (per-worker thread checks can't).
+    metrics.record_heartbeat("subscription-notifier", INTERVAL)
     # Wait first, then run — gives the app a moment to finish booting.
     while not _STOP.wait(INTERVAL):
+        metrics.record_heartbeat("subscription-notifier", INTERVAL)
         try:
             run_pass()
         except Exception:  # noqa: BLE001 - a bad pass must never kill the loop

tests/test_dashboard.py

@@ -0,0 +1,56 @@
+"""Tests for the ops dashboard's daemon health rendering (scripts/dashboard.py).
+
+The single-leader subscription notifier must be judged by heartbeat freshness, not by
+whether its thread happens to live on the worker the dashboard polled — otherwise ~2/3
+of polls (the non-leader workers) would cry DOWN for a perfectly healthy notifier.
+"""
+import importlib.util
+import os
+
+REPO = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+_spec = importlib.util.spec_from_file_location(
+    "dashboard", os.path.join(REPO, "scripts", "dashboard.py"))
+dashboard = importlib.util.module_from_spec(_spec)
+_spec.loader.exec_module(dashboard)
+
+C = dashboard.C(on=False)  # no ANSI colors, so we can assert on plain text
+
+
+def _mark(name, tnames, heartbeats):
+    return dashboard._daemon_mark(name, tnames, heartbeats, C)
+
+
+def test_notifier_fresh_heartbeat_is_up_even_when_thread_absent():
+    # The polled worker is a non-leader (no notifier thread), but a fresh shared heartbeat
+    # proves the leader is running it.
+    mark, detail = _mark("subscription-notifier", ["MainThread", "neighbor-warmer"],
+                         {"subscription-notifier": {"age_s": 120.0, "interval_s": 900}})
+    assert mark == "up"
+    assert "2m ago" in detail
+
+
+def test_notifier_stale_heartbeat_is_down():
+    # Beat older than two intervals + slack => genuinely stuck/dead, flag it.
+    mark, detail = _mark("subscription-notifier", ["MainThread"],
+                         {"subscription-notifier": {"age_s": 4_000.0, "interval_s": 900}})
+    assert mark == "DOWN"
+    assert "stale" in detail
+
+
+def test_notifier_missing_heartbeat_is_down():
+    mark, detail = _mark("subscription-notifier", ["MainThread"], {})
+    assert mark == "DOWN"
+    assert "no heartbeat" in detail
+
+
+def test_neighbor_warmer_still_uses_per_worker_thread_check():
+    # The warmer runs in every worker, so thread presence is the right signal and a
+    # heartbeat is neither written nor consulted.
+    assert _mark("neighbor-warmer", ["MainThread", "neighbor-warmer"], {})[0] == "up"
+    assert _mark("neighbor-warmer", ["MainThread"], {})[0] == "DOWN"
+
+
+def test_short_dur_formats():
+    assert dashboard._short_dur(45) == "45s"
+    assert dashboard._short_dur(120) == "2m"
+    assert dashboard._short_dur(3 * 3600 + 5 * 60) == "3h05m"

tests/test_metrics.py

@@ -199,3 +199,43 @@ def test_env_selects_sqlite_store(tmp_path, monkeypatch):
     assert isinstance(m._store, m._SqliteStore)
     m.record_inbound("page", 200)
     assert m.snapshot()["inbound"]["page"]["total"] == 1
+
+
+def test_heartbeat_snapshot_reports_age_and_interval(monkeypatch):
+    # A daemon stamps its liveness; the snapshot reports it as an age on the server clock
+    # (so a reader needn't reconcile clock skew) plus the beat's expected interval.
+    m = _fresh()
+    clock = {"now": 1_000_000.0}
+    monkeypatch.setattr(m.time, "time", lambda: clock["now"])
+    m.record_heartbeat("subscription-notifier", 900)
+    clock["now"] += 42                       # 42 s pass before the dashboard polls
+    hb = m.snapshot()["heartbeats"]["subscription-notifier"]
+    assert hb["age_s"] == 42.0
+    assert hb["interval_s"] == 900
+
+
+def test_heartbeat_absent_until_recorded():
+    # No beat yet => no entry (the dashboard renders this as DOWN / "no heartbeat").
+    m = _fresh()
+    assert m.snapshot()["heartbeats"] == {}
+
+
+def test_heartbeat_shared_across_workers_sqlite(tmp_path):
+    # Only the leader worker beats, but every worker reads the shared store — so the
+    # dashboard sees the notifier alive whichever worker happens to answer its poll.
+    m = _fresh()
+    db = str(tmp_path / "metrics.db")
+    leader = m._SqliteStore(db)
+    non_leader = m._SqliteStore(db)          # never beats (it isn't the notifier leader)
+    leader.record_heartbeat("subscription-notifier", 5_000.0, 900)
+    beats = non_leader.read()["heartbeats"]
+    assert beats["subscription-notifier"]["ts"] == 5_000.0
+    assert beats["subscription-notifier"]["interval_s"] == 900
+
+
+def test_heartbeat_survives_store_read_shape(tmp_path):
+    # The in-memory and SQLite stores expose heartbeats identically (ts + interval_s).
+    m = _fresh()
+    mem = m._MemStore()
+    mem.record_heartbeat("subscription-notifier", 10.0, 900)
+    assert mem.read()["heartbeats"]["subscription-notifier"] == {"ts": 10.0, "interval_s": 900}