diff --git a/app/main.py b/app/main.py index f8194cb4..0027bcb3 100644 --- a/app/main.py +++ b/app/main.py @@ -962,6 +962,22 @@ 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 +#: 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] = {} + + async def _check_stale_workers() -> None: """Mark workers as offline if stale, and purge never-enrolled workers offline > 1 hour. @@ -973,6 +989,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 +1010,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 +1023,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 +1035,31 @@ 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 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 + 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) + ] + _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) @@ -4602,21 +4636,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..2a77542c 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,87 @@ 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, clear_raises=False): + sends = [] + + async def _capture_send(title, message, **kw): + sends.append(title) + return 1 + + record = AsyncMock(return_value=record_returns) + 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), + 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() + + 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()