From c125b71fbd6630ed9c645ed124dceb7e9b5a42b7 Mon Sep 17 00:00:00 2001 From: GeiserX <9169332+GeiserX@users.noreply.github.com> Date: Wed, 19 Aug 2026 13:38:31 +0200 Subject: [PATCH 1/2] fix(alerts): a Doze-napping phone worker pushed an offline alert per nap MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Wiring Telegram made this visible within the hour: seven identical 'worker went offline' pushes in ~80 minutes for one Android phone. The phone was flapping on Android Doze maintenance windows — one heartbeat every ~12 minutes — and the heartbeat route cleared the stored alert on the FIRST beat after offline. Every nap therefore became a brand-new alerted episode: mark offline -> push -> single beat -> clear (re-arm) -> mark offline -> push, forever. The 24h record dedupe never got to hold because its row kept being deleted. Clearing now belongs to the stale-worker sweep alone, and only after the worker stays online for two consecutive sweeps (~4 minutes). A single Doze beat never re-arms — the record cooldown holds and the episode costs ONE push — while a genuinely recovered server re-arms almost immediately. The bell keeps showing the alert through a flap, which is the true state of an unreliable worker. Streaks are in-memory on purpose: a UI restart forgets them, which at worst re-arms one alert early, and every missed clear self-heals within 24h because the record cooldown is measured from the lingering row's own created_at. Pinned end to end in TestFlapDamping (one push per episode, sustained recovery re-arms, streak resets on every flap) and the route test now asserts the heartbeat clears NOTHING; mutation-verified — collapsing the gate to first-sweep clearing turns three tests red. --- app/main.py | 74 ++++++++++++-------- tests/conftest.py | 1 + tests/test_worker_offline_alerts.py | 103 ++++++++++++++++++++++------ 3 files changed, 130 insertions(+), 48 deletions(-) diff --git a/app/main.py b/app/main.py index f8194cb4..24a57bee 100644 --- a/app/main.py +++ b/app/main.py @@ -962,6 +962,19 @@ async def _run_vacuum() -> None: logger.warning("Database VACUUM error: %s", exc) +# A recovery re-arms the offline alert only after this many CONSECUTIVE sweeps +# online. One heartbeat is not a recovery: an Android phone in Doze wakes every +# ~12 minutes, heartbeats once and sleeps again — and when that single beat +# cleared the alert row immediately, every nap became a "new" offline episode +# with its own push. Seven identical Telegram messages in ~80 minutes, live. +# Two sweeps (~4 minutes online) is enough to tell a recovery from a blip while +# a genuinely recovered server re-arms almost immediately. In-memory on +# purpose: a UI restart forgets streaks, which at worst re-arms one alert +# early — the 24h record cooldown still caps the damage. +_SUSTAINED_RECOVERY_SWEEPS = 2 +_worker_online_streak: dict[str, int] = {} + + async def _check_stale_workers() -> None: """Mark workers as offline if stale, and purge never-enrolled workers offline > 1 hour. @@ -973,6 +986,9 @@ async def _check_stale_workers() -> None: re-enroll — a permanent fleet lockout after a reboot/maintenance window. Only a worker that never completed enrollment is purged automatically; removing an enrolled worker is a deliberate action via the UI. + + This sweep is also the ONLY place the offline alert is cleared — see the + sustained-recovery branch. The heartbeat route deliberately does not clear. """ try: workers = await database.list_workers() @@ -991,6 +1007,7 @@ async def _check_stale_workers() -> None: cid = w.get("client_id") or w["name"] last = datetime.fromisoformat(last_hb).replace(tzinfo=UTC) if w["status"] == "online" and last < cutoff: + _worker_online_streak.pop(cid, None) # Conditional on the heartbeat the decision was based on: a # recovery heartbeat landing between this sweep's read and its # write must win, or the worker would be alerted offline at its @@ -1003,6 +1020,7 @@ async def _check_stale_workers() -> None: # Same alert/push/recovery pattern as a failing collector. await _raise_worker_offline_alert(w) elif w["status"] == "offline": + _worker_online_streak.pop(cid, None) if last < purge_cutoff and not w.get("api_key_enc"): await database.delete_worker(w["id"]) logger.info("Purged stale unenrolled worker '%s' (offline since %s)", w["name"], last_hb) @@ -1014,18 +1032,26 @@ async def _check_stale_workers() -> None: # retried instead of being lost to the already-offline # state, which used to make the miss permanent. await _raise_worker_offline_alert(w) - elif w["status"] == "online" and any( - a.get("kind") == "worker" and a.get("client_id") == cid for a in _collector_alerts - ): - # Reconciliation: an online worker with a lingering worker - # alert means a recovery clear failed or was missed (the - # heartbeat route treats that clear as best-effort so a DB - # hiccup can never fail a heartbeat). Finish the job here. - await database.clear_alerts("worker", cid) - _collector_alerts = [ - a for a in _collector_alerts if not (a.get("kind") == "worker" and a.get("client_id") == cid) - ] - logger.info("Cleared lingering offline alert for recovered worker '%s'", w["name"]) + elif w["status"] == "online": + # Sustained recovery, and the ONLY place the offline alert is + # cleared. Clearing re-arms the next push, so a clear on the + # FIRST heartbeat turned every Doze nap of a phone worker into + # a fresh alerted episode (~5 pushes/hour, live). The clear + # fires once when the streak reaches the threshold; the + # bell-presence fallback re-runs it if that clear failed. A + # clear missed by both paths self-heals within 24h — the + # record cooldown is measured from the lingering row's own + # created_at, so it ages out of the dedupe window on its own. + streak = _worker_online_streak.get(cid, 0) + 1 + _worker_online_streak[cid] = streak + bell_lingers = any(a.get("kind") == "worker" and a.get("client_id") == cid for a in _collector_alerts) + if streak == _SUSTAINED_RECOVERY_SWEEPS or (streak > _SUSTAINED_RECOVERY_SWEEPS and bell_lingers): + await database.clear_alerts("worker", cid) + _collector_alerts = [ + a for a in _collector_alerts if not (a.get("kind") == "worker" and a.get("client_id") == cid) + ] + if bell_lingers: + logger.info("Cleared offline alert for recovered worker '%s'", w["name"]) except Exception as exc: logger.warning("Stale worker check error for worker '%s': %s", w.get("name", w.get("id")), exc) @@ -4602,21 +4628,15 @@ async def api_worker_heartbeat(request: Request, body: WorkerHeartbeat) -> dict[ ) metrics.record_heartbeat(body.name) if previous is not None and previous[0] == "offline": - # Recovered — drop the stored alert so the NEXT outage notifies again - # instead of being deduped inside the cooldown window, and prune the - # bell entry now rather than at the next hourly rebuild. Keyed by - # client_id (the identity), never the cosmetic display name. Best - # effort on purpose: a DB hiccup here must never fail a HEARTBEAT — - # the stale-worker sweep reconciles any clear this misses. - try: - await database.clear_alerts("worker", cid) - except Exception as exc: # noqa: BLE001 - logger.warning("Could not clear the offline alert for recovered worker %s (sweep will): %s", cid, exc) - else: - global _collector_alerts - _collector_alerts = [ - a for a in _collector_alerts if not (a.get("kind") == "worker" and a.get("client_id") == cid) - ] + # Deliberately NO alert clearing here. This used to drop the stored + # alert on the first heartbeat after offline — which re-armed the push + # on every single beat, so a phone worker waking from Android Doze + # every ~12 minutes produced an identical offline push per nap (seven + # in 80 minutes, live). The stale-worker sweep now owns clearing, and + # only after the worker stays online for _SUSTAINED_RECOVERY_SWEEPS + # consecutive sweeps — a real recovery re-arms in ~4 minutes, a Doze + # blip never does, and the 24h record cooldown keeps one push per + # episode. logger.info("Worker '%s' is back online", previous[1]) resp: dict[str, Any] = {"status": "ok", "worker_id": worker_id} if state == "enroll": diff --git a/tests/conftest.py b/tests/conftest.py index 89711289..c40d872e 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -133,6 +133,7 @@ def _reset_process_wide_caches(): """ for module_name, attr in ( ("app.main", "_net_baselines"), + ("app.main", "_worker_online_streak"), ("app.credential_test", "_last_attempt"), ): with contextlib.suppress(Exception): diff --git a/tests/test_worker_offline_alerts.py b/tests/test_worker_offline_alerts.py index 600f81d2..84448cb1 100644 --- a/tests/test_worker_offline_alerts.py +++ b/tests/test_worker_offline_alerts.py @@ -74,8 +74,10 @@ def _run(coro): def _isolate_bell(): before = main._collector_alerts main._collector_alerts = [] + main._worker_online_streak.clear() yield main._collector_alerts = before + main._worker_online_streak.clear() class TestOfflineTransition: @@ -149,16 +151,19 @@ def test_a_still_offline_worker_retries_the_durable_alert(self): r.delete.assert_not_awaited() def test_an_online_worker_with_a_lingering_alert_is_reconciled(self): - """A recovery clear the heartbeat route missed (it is best-effort so a - DB hiccup can never fail a heartbeat) is finished by the sweep.""" + """The sweep owns clearing, and only after a SUSTAINED recovery: the + first online sweep arms the streak, the second clears row + bell.""" from datetime import UTC, datetime fresh = _worker_row(last_heartbeat=datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%S")) main._collector_alerts = [ {"kind": "worker", "platform": "watchtower", "client_id": "cid-watchtower", "error": "x"} ] - r = self._sweep(rows=[fresh]) - r.clear.assert_awaited_once_with("worker", "cid-watchtower") + first = self._sweep(rows=[fresh]) + first.clear.assert_not_awaited() # one online sweep is not a recovery yet + assert main._collector_alerts # bell stays honest meanwhile + second = self._sweep(rows=[fresh]) + second.clear.assert_awaited_once_with("worker", "cid-watchtower") assert main._collector_alerts == [] @@ -251,28 +256,22 @@ async def _capture_clear(kind, subject): ) return cleared, result - def test_recovery_clears_the_alert_and_the_bell_by_identity(self): + def test_a_recovery_heartbeat_never_clears_the_alert(self): + """The route clearing on the FIRST heartbeat was the flap-spam engine: + a phone in Android Doze wakes ~every 12 minutes, beats once, sleeps — + and each beat re-armed the push, one identical Telegram message per + nap. The route now only logs; the sweep clears after a sustained + recovery.""" main._collector_alerts = [ {"kind": "worker", "platform": "watchtower", "client_id": "cid-watchtower", "error": "offline"}, {"kind": "collector", "platform": "honeygain", "error": "kept"}, ] - cleared, _ = self._heartbeat(previous=("offline", "watchtower")) - assert cleared == [("worker", "cid-watchtower")] - kinds = [(a["kind"], a.get("client_id")) for a in main._collector_alerts] - assert ("worker", "cid-watchtower") not in kinds - assert ("collector", None) in kinds # untouched - - def test_a_failed_clear_never_fails_the_heartbeat(self): - """Heartbeats are the fleet's lifeline: the clear is best-effort, the - bell entry stays (memory and disk must not diverge), and the sweep's - reconciliation finishes the job.""" - main._collector_alerts = [ - {"kind": "worker", "platform": "watchtower", "client_id": "cid-watchtower", "error": "offline"} - ] - cleared, result = self._heartbeat(previous=("offline", "watchtower"), clear_raises=True) - assert result["status"] == "ok" # the heartbeat itself succeeded + cleared, result = self._heartbeat(previous=("offline", "watchtower")) + assert result["status"] == "ok" assert cleared == [] - assert len(main._collector_alerts) == 1 # NOT pruned while the row survives + kinds = [(a["kind"], a.get("client_id")) for a in main._collector_alerts] + assert ("worker", "cid-watchtower") in kinds # bell stays until the sweep decides + assert ("collector", None) in kinds def test_an_online_worker_heartbeat_clears_nothing(self): # Negative control: no transition, no clearing. @@ -286,3 +285,65 @@ def test_an_online_worker_heartbeat_clears_nothing(self): def test_a_new_worker_heartbeat_clears_nothing(self): cleared, _ = self._heartbeat(previous=None) assert cleared == [] + + +class TestFlapDamping: + """The OPPO incident, pinned end to end: a Doze-napping phone worker must + cost ONE push per episode, not one per nap.""" + + def _sweep(self, *, rows, record_returns=True): + sends = [] + + async def _capture_send(title, message, **kw): + sends.append(title) + return 1 + + record = AsyncMock(return_value=record_returns) + clear = AsyncMock() + with ( + patch("app.main.database.list_workers", new_callable=AsyncMock, return_value=rows), + patch("app.main.database.mark_worker_offline_if_unchanged", new_callable=AsyncMock, return_value=True), + patch("app.main.database.delete_worker", new_callable=AsyncMock), + patch("app.main.database.record_alert", record), + patch("app.main.database.clear_alerts", clear), + patch("app.main.notify.send", _capture_send), + ): + _run(_check_stale_workers()) + return SimpleNamespace(record=record, clear=clear, sends=sends) + + def _fresh_row(self): + from datetime import UTC, datetime + + return _worker_row(last_heartbeat=datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%S")) + + def test_a_single_doze_beat_does_not_rearm_the_push(self): + # Episode starts: stale online worker -> offline, one push. + first = self._sweep(rows=[_worker_row()]) + assert len(first.sends) == 1 + first.clear.assert_not_awaited() + + # The phone wakes once (route logs, clears nothing — pinned above), + # and the next sweep sees it online: streak 1, still no clear. + blip = self._sweep(rows=[self._fresh_row()]) + blip.clear.assert_not_awaited() + + # It dozes off again: the record cooldown still holds (deduped), so + # the re-offline transition pushes NOTHING. One episode, one push. + again = self._sweep(rows=[_worker_row()], record_returns=None) + assert again.sends == [] + + def test_a_sustained_recovery_rearms_the_next_episode(self): + self._sweep(rows=[_worker_row()]) # episode 1: push + self._sweep(rows=[self._fresh_row()]) # online sweep 1 — no clear yet + second = self._sweep(rows=[self._fresh_row()]) # online sweep 2 — clears + second.clear.assert_awaited_once_with("worker", "cid-watchtower") + # Cleared row -> a NEW offline episode records fresh and pushes again. + episode2 = self._sweep(rows=[_worker_row()]) + assert len(episode2.sends) == 1 + + def test_the_streak_resets_on_every_flap(self): + self._sweep(rows=[_worker_row()]) # offline + self._sweep(rows=[self._fresh_row()]) # online (streak 1) + self._sweep(rows=[_worker_row()], record_returns=None) # flap: offline again + after_flap = self._sweep(rows=[self._fresh_row()]) # online (streak must be 1 again) + after_flap.clear.assert_not_awaited() From eeb5c6946781c080090438ba32a2c0d84bd63acb Mon Sep 17 00:00:00 2001 From: GeiserX <9169332+GeiserX@users.noreply.github.com> Date: Wed, 19 Aug 2026 13:44:33 +0200 Subject: [PATCH 2/2] fix(alerts): a failed durable clear must not lose its retry to the bell rebuild MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CodeRabbit's one finding on #348, and it closed the edge the first cut had documented as acceptable: the retry was gated on the in-memory bell entry, which the hourly rebuild derives from OFFLINE workers only — so an online worker whose durable clear failed lost both the bell entry and, with it, the retry, suppressing the next episode's push until the row aged out of the 24h window. The retry state now lives in the streak itself: at or past the threshold the clear runs every sweep until it SUCCEEDS (a raise lands in the per-worker handler and the streak stays armed), and only success parks the streak on a sentinel that the next offline episode resets. Regression tests: a failed clear is retried on the next sweep with the bell entry gone, and a successful clear is never repeated while online. Mutation-verified: parking the sentinel before the clear turns the retry test red. --- app/main.py | 28 ++++++++++++++++++---------- tests/test_worker_offline_alerts.py | 26 ++++++++++++++++++++++++-- 2 files changed, 42 insertions(+), 12 deletions(-) diff --git a/app/main.py b/app/main.py index 24a57bee..0027bcb3 100644 --- a/app/main.py +++ b/app/main.py @@ -972,6 +972,9 @@ async def _run_vacuum() -> None: # purpose: a UI restart forgets streaks, which at worst re-arms one alert # early — the 24h record cooldown still caps the damage. _SUSTAINED_RECOVERY_SWEEPS = 2 +#: Sentinel streak value: this recovery's durable clear SUCCEEDED — stop +#: clearing until the next offline episode resets the streak. +_STREAK_CLEARED = -1 _worker_online_streak: dict[str, int] = {} @@ -1036,21 +1039,26 @@ async def _check_stale_workers() -> None: # Sustained recovery, and the ONLY place the offline alert is # cleared. Clearing re-arms the next push, so a clear on the # FIRST heartbeat turned every Doze nap of a phone worker into - # a fresh alerted episode (~5 pushes/hour, live). The clear - # fires once when the streak reaches the threshold; the - # bell-presence fallback re-runs it if that clear failed. A - # clear missed by both paths self-heals within 24h — the - # record cooldown is measured from the lingering row's own - # created_at, so it ages out of the dedupe window on its own. - streak = _worker_online_streak.get(cid, 0) + 1 + # a fresh alerted episode (~5 pushes/hour, live). The retry + # state lives in the streak itself: at or past the threshold + # the clear runs every sweep until it SUCCEEDS (a raise lands + # in this loop's per-worker handler and the streak stays + # armed), and only success parks the streak on the sentinel — + # so a failed durable clear cannot be stranded by the hourly + # bell rebuild dropping the in-memory entry. + streak = _worker_online_streak.get(cid, 0) + if streak == _STREAK_CLEARED: + continue + streak += 1 _worker_online_streak[cid] = streak - bell_lingers = any(a.get("kind") == "worker" and a.get("client_id") == cid for a in _collector_alerts) - if streak == _SUSTAINED_RECOVERY_SWEEPS or (streak > _SUSTAINED_RECOVERY_SWEEPS and bell_lingers): + if streak >= _SUSTAINED_RECOVERY_SWEEPS: await database.clear_alerts("worker", cid) + had_bell = any(a.get("kind") == "worker" and a.get("client_id") == cid for a in _collector_alerts) _collector_alerts = [ a for a in _collector_alerts if not (a.get("kind") == "worker" and a.get("client_id") == cid) ] - if bell_lingers: + _worker_online_streak[cid] = _STREAK_CLEARED + if had_bell: logger.info("Cleared offline alert for recovered worker '%s'", w["name"]) except Exception as exc: logger.warning("Stale worker check error for worker '%s': %s", w.get("name", w.get("id")), exc) diff --git a/tests/test_worker_offline_alerts.py b/tests/test_worker_offline_alerts.py index 84448cb1..2a77542c 100644 --- a/tests/test_worker_offline_alerts.py +++ b/tests/test_worker_offline_alerts.py @@ -291,7 +291,7 @@ class TestFlapDamping: """The OPPO incident, pinned end to end: a Doze-napping phone worker must cost ONE push per episode, not one per nap.""" - def _sweep(self, *, rows, record_returns=True): + def _sweep(self, *, rows, record_returns=True, clear_raises=False): sends = [] async def _capture_send(title, message, **kw): @@ -299,7 +299,7 @@ async def _capture_send(title, message, **kw): return 1 record = AsyncMock(return_value=record_returns) - clear = AsyncMock() + clear = AsyncMock(side_effect=RuntimeError("db locked") if clear_raises else None) with ( patch("app.main.database.list_workers", new_callable=AsyncMock, return_value=rows), patch("app.main.database.mark_worker_offline_if_unchanged", new_callable=AsyncMock, return_value=True), @@ -347,3 +347,25 @@ def test_the_streak_resets_on_every_flap(self): self._sweep(rows=[_worker_row()], record_returns=None) # flap: offline again after_flap = self._sweep(rows=[self._fresh_row()]) # online (streak must be 1 again) after_flap.clear.assert_not_awaited() + + def test_a_failed_durable_clear_is_retried_next_sweep(self): + """The retry must not depend on the in-memory bell entry: the hourly + rebuild derives worker entries from OFFLINE workers only, so it drops + the bell for an online worker whose durable clear failed — and a + bell-gated retry would then never run, suppressing the next episode's + push until the row aged out of the 24h window.""" + self._sweep(rows=[_worker_row()]) # offline episode + self._sweep(rows=[self._fresh_row()]) # online sweep 1 — arming + failing = self._sweep(rows=[self._fresh_row()], clear_raises=True) + failing.clear.assert_awaited_once() # threshold reached, clear attempted + main._collector_alerts = [] # hourly rebuild dropped the bell entry + retry = self._sweep(rows=[self._fresh_row()]) + retry.clear.assert_awaited_once_with("worker", "cid-watchtower") + + def test_a_successful_clear_is_not_repeated_while_online(self): + self._sweep(rows=[_worker_row()]) + self._sweep(rows=[self._fresh_row()]) + cleared = self._sweep(rows=[self._fresh_row()]) # clears, parks on the sentinel + cleared.clear.assert_awaited_once() + later = self._sweep(rows=[self._fresh_row()]) + later.clear.assert_not_awaited()