From 2b6aaad931b05a0bc911ad0da6a56bea65377cd6 Mon Sep 17 00:00:00 2001 From: GeiserX <9169332+GeiserX@users.noreply.github.com> Date: Wed, 19 Aug 2026 17:32:08 +0200 Subject: [PATCH] =?UTF-8?q?fix(alerts):=20cap=20offline=20pushes=20per=20w?= =?UTF-8?q?orker=20=E2=80=94=20episode=20damping=20lost=20to=20a=20rhythmi?= =?UTF-8?q?c=20phone?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The sustained-recovery damping (v1.36.1) was defeated within hours by the same phone: its NORMAL duty cycle turned out to be 7-10 minutes awake, 2-3 minutes dozing — every wake window passed the sustained bar, re-armed the alert, and the next nap pushed again. Eight pushes in two hours, live, with the damping active. A worker whose ordinary operation is rhythmic flapping can never be rate-limited by episode detection. So the cap moves to the notification itself: after an offline push for a worker, further offline pushes for THAT worker wait out a 6-hour floor, independent of episodes. The first push for a genuinely dead server is still immediate; the dashboard and bell keep tracking every episode in real time; only the phone-buzz is capped (worst case 4/day/worker, down from ~100). Recovery deliberately never resets the floor — resetting on recovery is exactly the re-arm loop this ends. A total delivery failure releases the floor so the un-dedupe retry contract stays true. Floor state is in-memory: a restart forgets it, costing at most one early push. Pinned in TestOfflinePushFloor (records-but-does-not-push inside the floor, expiry re-pushes, recovery never resets, failed delivery releases); mutation-verified — removing the gate turns the suite red. --- app/main.py | 51 +++++++++++++++++++---- tests/conftest.py | 1 + tests/test_worker_offline_alerts.py | 64 ++++++++++++++++++++++++++++- 3 files changed, 107 insertions(+), 9 deletions(-) diff --git a/app/main.py b/app/main.py index 0027bcb3..31eb3a05 100644 --- a/app/main.py +++ b/app/main.py @@ -408,6 +408,12 @@ async def _push_alert(kind: str, subject: str, title: str, message: str, alert_i return metrics.record_notify_delivery(bool(delivered)) if not delivered: + # The worker push floor is armed at spawn time, before the outcome is + # known — a TOTAL delivery failure must release it, or the un-dedupe + # retry below would be suppressed for the whole floor window and the + # "will retry on the next cycle" promise becomes false. + if kind == "worker": + _worker_last_offline_push.pop(subject, None) try: undeduped = await database.delete_alert(alert_id) except Exception as exc: # noqa: BLE001 @@ -977,6 +983,20 @@ async def _run_vacuum() -> None: _STREAK_CLEARED = -1 _worker_online_streak: dict[str, int] = {} +#: Floor between offline PUSHES for one worker, independent of episodes. +#: Sustained-recovery damping alone was defeated by a phone whose NORMAL duty +#: cycle is 7-10 minutes awake, 2-3 minutes dozing: every wake window passed +#: the sustained bar, re-armed the alert, and the next nap pushed — eight +#: pushes in two hours, live, AFTER the episode damping shipped. A worker +#: whose ordinary operation is rhythmic flapping can never be rate-limited by +#: episode detection, so the cap sits on the notification itself: the first +#: offline push is immediate, repeats for the same worker wait out the floor. +#: The bell and dashboard stay real-time; only the phone-buzz is capped. +#: In-memory on purpose — a restart forgets the floor, which at worst costs +#: one early push. +_WORKER_OFFLINE_PUSH_COOLDOWN_S = 6 * 3600 +_worker_last_offline_push: dict[str, float] = {} + async def _check_stale_workers() -> None: """Mark workers as offline if stale, and purge never-enrolled workers offline > 1 hour. @@ -1078,15 +1098,30 @@ async def _raise_worker_offline_alert(w: dict[str, Any]) -> None: "containers keep running and earning, but CashPilot cannot see or manage them until it reconnects." ) if alert_id := await database.record_alert("worker", cid, msg): - _spawn( - _push_alert( - "worker", - cid, - f"CashPilot: worker '{w['name']}' went offline", - msg, - alert_id, + # The push floor, checked only on a FRESH row (a deduped record never + # gets here): a new episode inside the floor keeps its row and bell + # entry — the dashboard tells the truth — but does not buzz the phone + # again. Deliberately never reset on recovery; resetting there is + # exactly the re-arm loop this exists to end. + now = time.monotonic() + last_push = _worker_last_offline_push.get(cid) + if last_push is None or now - last_push >= _WORKER_OFFLINE_PUSH_COOLDOWN_S: + _worker_last_offline_push[cid] = now + _spawn( + _push_alert( + "worker", + cid, + f"CashPilot: worker '{w['name']}' went offline", + msg, + alert_id, + ) + ) + else: + logger.info( + "Offline push for worker '%s' suppressed by the per-worker floor (%.0f min since last)", + w["name"], + (now - last_push) / 60, ) - ) # Into the bell NOW (the sweep runs every 2 min); the hourly collection # rebuild re-derives it from the workers table. if not any(a.get("kind") == "worker" and a.get("client_id") == cid for a in _collector_alerts): diff --git a/tests/conftest.py b/tests/conftest.py index c40d872e..03905f73 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -134,6 +134,7 @@ def _reset_process_wide_caches(): for module_name, attr in ( ("app.main", "_net_baselines"), ("app.main", "_worker_online_streak"), + ("app.main", "_worker_last_offline_push"), ("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 2a77542c..fe8056db 100644 --- a/tests/test_worker_offline_alerts.py +++ b/tests/test_worker_offline_alerts.py @@ -75,9 +75,11 @@ def _isolate_bell(): before = main._collector_alerts main._collector_alerts = [] main._worker_online_streak.clear() + main._worker_last_offline_push.clear() yield main._collector_alerts = before main._worker_online_streak.clear() + main._worker_last_offline_push.clear() class TestOfflineTransition: @@ -337,8 +339,12 @@ def test_a_sustained_recovery_rearms_the_next_episode(self): 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. + # Cleared row -> a NEW offline episode records fresh again. The PUSH + # additionally obeys the per-worker floor (TestOfflinePushFloor), so + # it fires here only once the floor has passed. + main._worker_last_offline_push["cid-watchtower"] -= main._WORKER_OFFLINE_PUSH_COOLDOWN_S + 1 episode2 = self._sweep(rows=[_worker_row()]) + episode2.record.assert_awaited_once() assert len(episode2.sends) == 1 def test_the_streak_resets_on_every_flap(self): @@ -369,3 +375,59 @@ def test_a_successful_clear_is_not_repeated_while_online(self): cleared.clear.assert_awaited_once() later = self._sweep(rows=[self._fresh_row()]) later.clear.assert_not_awaited() + + +class TestOfflinePushFloor: + """Episode damping cannot rate-limit a worker whose NORMAL operation is + rhythmic flapping — the live OPPO phone stays awake 7-10 minutes between + 2-3 minute dozes, so every wake window passes the sustained-recovery bar, + re-arms the alert, and the next nap pushed again (eight pushes in two + hours AFTER episode damping shipped). The floor caps the buzz itself: + the first push is immediate, repeats per worker wait out the floor, and + the dashboard/bell stay real-time truth.""" + + _sweep = TestFlapDamping._sweep + _fresh_row = TestFlapDamping._fresh_row + + def _cycle_to_rearmed(self): + first = self._sweep(rows=[_worker_row()]) # episode 1: pushes, arms the floor + assert len(first.sends) == 1 + self._sweep(rows=[self._fresh_row()]) + self._sweep(rows=[self._fresh_row()]) # sustained recovery: cleared + re-armed + + def test_a_new_episode_inside_the_floor_records_but_does_not_push(self): + self._cycle_to_rearmed() + second = self._sweep(rows=[_worker_row()]) + second.record.assert_awaited_once() # the row and bell stay truthful + assert second.sends == [] # the phone does not buzz again + assert any(a.get("kind") == "worker" for a in main._collector_alerts) + + def test_the_floor_expires_and_the_next_episode_pushes(self): + self._cycle_to_rearmed() + main._worker_last_offline_push["cid-watchtower"] -= main._WORKER_OFFLINE_PUSH_COOLDOWN_S + 1 + second = self._sweep(rows=[_worker_row()]) + assert len(second.sends) == 1 + + def test_recovery_never_resets_the_floor(self): + # The whole point: recovering is what re-armed the spam loop, so a + # recovery must not shorten the floor. + self._cycle_to_rearmed() + armed_at = main._worker_last_offline_push["cid-watchtower"] + self._sweep(rows=[self._fresh_row()]) + self._sweep(rows=[self._fresh_row()]) + assert main._worker_last_offline_push["cid-watchtower"] == armed_at + + def test_a_failed_delivery_releases_the_floor(self): + """The floor is armed at spawn time, before the outcome is known — a + total delivery failure must release it or the un-dedupe retry would be + suppressed for the whole floor window.""" + import time as _time + + main._worker_last_offline_push["cid-watchtower"] = _time.monotonic() + with ( + patch("app.main.notify.send", new_callable=AsyncMock, return_value=0), + patch("app.main.notify.is_enabled", return_value=True), + patch("app.main.database.delete_alert", new_callable=AsyncMock, return_value=True), + ): + _run(main._push_alert("worker", "cid-watchtower", "t", "m", 123)) + assert "cid-watchtower" not in main._worker_last_offline_push