From 7b591439c1f305f38e04bc8d88d65ab8c09bdd78 Mon Sep 17 00:00:00 2001 From: emi Date: Tue, 21 Jul 2026 15:52:28 +0000 Subject: [PATCH] Fix Postgres metrics: dropped counters + invisible notifier heartbeat (#8) --- core/metrics.py | 158 +++++++++++++++++++++++++++++++------ notifications/notify.py | 12 ++- tests/core/test_metrics.py | 86 ++++++++++++++++++++ 3 files changed, 229 insertions(+), 27 deletions(-) diff --git a/core/metrics.py b/core/metrics.py index f882f11..a12cd77 100644 --- a/core/metrics.py +++ b/core/metrics.py @@ -311,6 +311,29 @@ _METRICS_TABLES = ( "outbound_minute", "event_cume", "event_minute", ) +# Postgres advisory-lock key that serializes the once-per-deploy metrics reset +# (the TRUNCATE below). MUST differ from core/singleton.py's NOTIFIER_LOCK_KEY +# (1), which the notifier leader holds session-long — they share one advisory +# lock space, so a collision would deadlock the notifier against the reset. +_METRICS_BOOT_LOCK_KEY = 0x6D657472 # "metr" + + +def _boot_token() -> float: + """A value identical for every uvicorn worker of one container boot and + different across deploys, so the since-start reset fires exactly once per + deployment even though (with the lazy store below) workers connect to + Postgres seconds or minutes apart. PID 1's start time (`/proc/1/stat` field + 22, ticks since host boot) is shared by all workers in the container and + changes when the container is recreated. Falls back to a per-process value + if `/proc` is unavailable (non-Linux) — then each worker resets once, the + pre-lazy behavior.""" + try: + with open("/proc/1/stat", "rb") as f: + # comm (field 2) may contain spaces/parens, so split after the last ')'. + return float(f.read().rsplit(b")", 1)[1].split()[19]) + except Exception: # noqa: BLE001 - no /proc: degrade to per-process reset + return float(os.getpid()) + # Postgres DDL: same shape as the SQLite schema below, but UNLOGGED (fast, non- # durable — metrics are since-start and disposable) and one statement per execute # (psycopg's extended protocol won't run a multi-statement script). @@ -353,13 +376,26 @@ class _SqliteStore: for stmt in _PG_METRICS_SCHEMA: self._db.execute(stmt) # SQLite is wiped each boot by an external `ExecStartPre rm` of the file; - # Postgres has no file to remove, so reset the since-start counters here. - # Guarded to exactly the metrics tables and only this (Postgres) path. - self._db.execute("TRUNCATE " + ", ".join(_METRICS_TABLES)) - self._db.execute( - "INSERT INTO meta(key, value) VALUES('started_at', %s) ON CONFLICT DO NOTHING", - (time.time(),), - ) + # Postgres has no file to remove, so reset the since-start counters here — + # but exactly ONCE per deployment, not once per worker. With the lazy store + # (see _LazyPgStore) workers connect at different times, so an unguarded + # per-store TRUNCATE would let a late worker wipe counts the others already + # tallied. Gate it on a per-boot token under an advisory lock: the first + # worker of a fresh boot finds a stale/absent token and resets; the rest + # find this boot's token and skip. Guarded to exactly the metrics tables. + token = _boot_token() + with self._db.transaction(): + self._db.execute("SELECT pg_advisory_xact_lock(%s)", (_METRICS_BOOT_LOCK_KEY,)) + row = self._db.execute( + "SELECT value FROM meta WHERE key = 'boot_token'").fetchone() + if row is None or row[0] != token: + self._db.execute("TRUNCATE " + ", ".join(_METRICS_TABLES)) + self._db.execute( + "INSERT INTO meta(key, value) VALUES ('boot_token', %s) " + "ON CONFLICT (key) DO UPDATE SET value = excluded.value", (token,)) + self._db.execute( + "INSERT INTO meta(key, value) VALUES ('started_at', %s) " + "ON CONFLICT (key) DO UPDATE SET value = excluded.value", (time.time(),)) else: # check_same_thread=False: the connection is shared across a worker's threads, # serialized by self._lock. isolation_level=None: autocommit each statement. @@ -417,14 +453,19 @@ class _SqliteStore: def record_inbound(self, category, bucket) -> None: minute = _now_minute() with self._lock: + # Qualify the existing value as .count, not a bare `count`: + # Postgres rejects the unqualified form in ON CONFLICT DO UPDATE as + # ambiguous (the target row and the `excluded` pseudo-row both have a + # `count`), which silently dropped every traffic counter on the + # Postgres deployments. SQLite accepts the qualified form too. self._db.execute( f"INSERT INTO inbound_cume(category, bucket, count) VALUES({self._ph}, {self._ph}, 1) " - "ON CONFLICT(category, bucket) DO UPDATE SET count = count + 1", + "ON CONFLICT(category, bucket) DO UPDATE SET count = inbound_cume.count + 1", (category, bucket), ) self._db.execute( f"INSERT INTO inbound_minute(minute, category, count) VALUES({self._ph}, {self._ph}, 1) " - "ON CONFLICT(minute, category) DO UPDATE SET count = count + 1", + "ON CONFLICT(minute, category) DO UPDATE SET count = inbound_minute.count + 1", (minute, category), ) self._prune(minute) @@ -434,12 +475,12 @@ class _SqliteStore: with self._lock: self._db.execute( f"INSERT INTO outbound_cume(source, outcome, count) VALUES({self._ph}, {self._ph}, 1) " - "ON CONFLICT(source, outcome) DO UPDATE SET count = count + 1", + "ON CONFLICT(source, outcome) DO UPDATE SET count = outbound_cume.count + 1", (source, outcome), ) self._db.execute( f"INSERT INTO outbound_minute(minute, source, count) VALUES({self._ph}, {self._ph}, 1) " - "ON CONFLICT(minute, source) DO UPDATE SET count = count + 1", + "ON CONFLICT(minute, source) DO UPDATE SET count = outbound_minute.count + 1", (minute, source), ) self._prune(minute) @@ -457,12 +498,12 @@ class _SqliteStore: referrer = REFERRER_OTHER self._db.execute( f"INSERT INTO event_cume(event, referrer, day, count) VALUES({self._ph}, {self._ph}, {self._ph}, 1) " - "ON CONFLICT(event, referrer, day) DO UPDATE SET count = count + 1", + "ON CONFLICT(event, referrer, day) DO UPDATE SET count = event_cume.count + 1", (event, referrer, day), ) self._db.execute( f"INSERT INTO event_minute(minute, event, count) VALUES({self._ph}, {self._ph}, 1) " - "ON CONFLICT(minute, event) DO UPDATE SET count = count + 1", + "ON CONFLICT(minute, event) DO UPDATE SET count = event_minute.count + 1", (minute, event), ) self._prune(minute) @@ -536,17 +577,90 @@ class _SqliteStore: } +def _empty_snapshot() -> dict: + """The shape `read()` returns with nothing recorded yet — used while the lazy + Postgres store hasn't connected, and by `snapshot()` when the store is + momentarily unreadable, so callers never KeyError on a missing counter.""" + return {"started_at": time.time(), "inbound": {}, "outbound": {}, + "inbound_recent": {}, "outbound_recent": {}, "heartbeats": {}, + "events": {}, "events_recent": {}} + + +class _LazyPgStore: + """Postgres is the intended cross-worker store, but the app and db containers + start together, so on a cold boot the app imports this module — and used to + build the store eagerly — before Postgres is accepting connections. That build + then failed and (see `_make_store`) fell back to a per-process in-memory store + for the WHOLE container lifetime: metrics fragmented across workers and the + subscription-notifier heartbeat invisible (it lived in one worker's memory, + never in the shared table). Fix: don't connect at import. Connect on first use + — which is always after the app's own DB init, so Postgres is up — and if it + still isn't ready, retry on a short backoff instead of downgrading forever. + Once connected, delegate straight through.""" + + _RETRY_INTERVAL_S = 5.0 + + def __init__(self, dsn: str) -> None: + self._dsn = dsn + self._real: "_SqliteStore | None" = None + self._next_try = 0.0 + self._lock = threading.Lock() + + def _get(self) -> "_SqliteStore | None": + real = self._real + if real is not None: + return real + now = time.monotonic() + with self._lock: + if self._real is not None: + return self._real + if now < self._next_try: + return None + try: + self._real = _SqliteStore(self._dsn, postgres=True) + except Exception: # noqa: BLE001 - Postgres not ready yet; retry later + self._next_try = now + self._RETRY_INTERVAL_S + return None + return self._real + + # Writes are best-effort until connected — the pre-DB-ready window is seconds, + # and the module-level record_* already swallow errors. A dropped write there + # is the same guarantee the old eager store gave once it fell back. + def record_inbound(self, category, bucket) -> None: + s = self._get() + if s is not None: + s.record_inbound(category, bucket) + + def record_outbound(self, source, outcome) -> None: + s = self._get() + if s is not None: + s.record_outbound(source, outcome) + + def record_event(self, event, referrer, day) -> None: + s = self._get() + if s is not None: + s.record_event(event, referrer, day) + + def record_heartbeat(self, name, ts, interval_s) -> None: + s = self._get() + if s is not None: + s.record_heartbeat(name, ts, interval_s) + + def read(self) -> dict: + s = self._get() + return s.read() if s is not None else _empty_snapshot() + + def _make_store(): """Pick the store from the environment. A ``postgresql://`` URL in THERMOGRAPH_DATABASE_URL routes counters into a shared Postgres store (prod / - multi-worker) and takes precedence; otherwise a path in ``THERMOGRAPH_METRICS_DB`` - selects the shared SQLite store (needed when running multiple workers); otherwise - in-memory. A store that fails to open falls back — metrics must never break boot.""" + multi-worker) and takes precedence — via `_LazyPgStore`, so a not-yet-ready DB + at import never permanently downgrades to in-memory. Otherwise a path in + ``THERMOGRAPH_METRICS_DB`` selects the shared SQLite store (needed when running + multiple workers); otherwise in-memory. Non-Postgres opens still fall back — + metrics must never break boot.""" if IS_POSTGRES: - try: - return _SqliteStore(_PG_DSN, postgres=True) - except Exception: # noqa: BLE001 - never let instrumentation break startup - pass + return _LazyPgStore(_PG_DSN) path = os.environ.get("THERMOGRAPH_METRICS_DB", "").strip() if path: try: @@ -633,9 +747,7 @@ def snapshot() -> dict: 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": {}, "heartbeats": {}, - "events": {}, "events_recent": {}} + data = _empty_snapshot() inbound = data["inbound"] outbound = data["outbound"] started_at = data["started_at"] diff --git a/notifications/notify.py b/notifications/notify.py index 08c8bde..b24d981 100644 --- a/notifications/notify.py +++ b/notifications/notify.py @@ -356,15 +356,19 @@ def run_loop(): # 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. + # Wait first, then run — gives the app a moment to finish booting. The whole + # body is guarded: this is a daemon thread, so ANY escaping exception (a bad + # pass, a homepage refresh, a Discord post) would silently kill it for the + # process's lifetime — no notifier, no heartbeat, no restart. Swallow per + # iteration instead; the next wake retries and keeps the heartbeat fresh. 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 + _maybe_refresh_homepage() + _maybe_post_discord() + except Exception: # noqa: BLE001 - a bad iteration must never kill the loop pass - _maybe_refresh_homepage() - _maybe_post_discord() def start(): diff --git a/tests/core/test_metrics.py b/tests/core/test_metrics.py index acbb69f..5a4a743 100644 --- a/tests/core/test_metrics.py +++ b/tests/core/test_metrics.py @@ -316,3 +316,89 @@ def test_event_rate_limit_per_ip(): for _ in range(m._EVENT_RATE_PER_MIN + 25): m.record_event("home.locate", ip="203.0.113.9") assert m.snapshot()["events_total"] == m._EVENT_RATE_PER_MIN + + +# --- Postgres store: lazy connect, no permanent in-memory downgrade ------------- +# Regression for the cold-boot race where the app+db containers start together, the +# app imported metrics before Postgres was accepting connections, the eager store +# build failed, and the process fell back to a per-worker in-memory store for its +# whole lifetime — fragmenting counters and hiding the notifier heartbeat. + +class _FakeStore: + """Stand-in for a connected `_SqliteStore`: records heartbeats to memory and + `read()` surfaces them in the canonical snapshot shape.""" + + def __init__(self, *args, **kwargs): + self.beats = {} + + def record_inbound(self, *a): + pass + + def record_outbound(self, *a): + pass + + def record_event(self, *a): + pass + + def record_heartbeat(self, name, ts, interval_s): + self.beats[name] = {"ts": ts, "interval_s": interval_s} + + def read(self): + return {"started_at": 0.0, "inbound": {}, "outbound": {}, + "inbound_recent": {}, "outbound_recent": {}, + "heartbeats": dict(self.beats), "events": {}, "events_recent": {}} + + +def test_postgres_url_uses_lazy_store_and_never_downgrades(monkeypatch): + monkeypatch.setenv("THERMOGRAPH_DATABASE_URL", "postgresql://u:p@127.0.0.1:1/x") + m = _fresh() + # Postgres is the intended store, and it is NOT connected at import time. + assert isinstance(m._store, m._LazyPgStore) + assert m._store._real is None + + # Boot-time DB blip: the underlying store can't be built. Metrics must not crash, + # must not fabricate a heartbeat, and must NOT downgrade to an in-memory store. + def _boom(*a, **k): + raise RuntimeError("postgres not ready yet") + monkeypatch.setattr(m, "_SqliteStore", _boom) + m.record_heartbeat("subscription-notifier", 900) + assert m.snapshot()["heartbeats"] == {} + assert isinstance(m._store, m._LazyPgStore) # still the PG store, not _MemStore + + +def test_lazy_store_connects_on_first_use_and_delegates(monkeypatch): + monkeypatch.setenv("THERMOGRAPH_DATABASE_URL", "postgresql://u:p@127.0.0.1:1/x") + m = _fresh() + monkeypatch.setattr(m, "_SqliteStore", _FakeStore) + # First use (as the notifier's first beat is, after the app's own DB init) connects. + m.record_heartbeat("subscription-notifier", 900) + assert m._store._real is not None + assert "subscription-notifier" in m.snapshot()["heartbeats"] + + +def test_lazy_store_retries_after_a_failed_connect(monkeypatch): + monkeypatch.setenv("THERMOGRAPH_DATABASE_URL", "postgresql://u:p@127.0.0.1:1/x") + m = _fresh() + calls = {"n": 0} + + def _flaky(*a, **k): + calls["n"] += 1 + if calls["n"] == 1: + raise RuntimeError("not ready") + return _FakeStore() + monkeypatch.setattr(m, "_SqliteStore", _flaky) + + m.record_heartbeat("x", 900) # first attempt fails, backoff armed + assert m._store._real is None + m.record_heartbeat("x", 900) # within backoff: no second attempt + assert calls["n"] == 1 + m._store._next_try = 0.0 # backoff elapsed + m.record_heartbeat("subscription-notifier", 900) + assert calls["n"] == 2 + assert m._store._real is not None + + +def test_boot_token_is_stable_within_a_process(): + m = _fresh() + assert isinstance(m._boot_token(), float) + assert m._boot_token() == m._boot_token()