fix(alerts): one offline push per episode, not one per phone nap

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":

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):

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()