From 4fd431f90790d80e1f3ef3d475f3388cf25d8720 Mon Sep 17 00:00:00 2001 From: malice Date: Tue, 14 Jul 2026 10:38:07 -0600 Subject: [PATCH] fix(reminders): pace the reminder roll-call so N fires don't burst (#130) The ReminderScheduler is the third fire-broadcast exit and the only one that does not pass through the EventBus, so FirePacer (which paces the Central and native fire-event exits to <=1/60s) never sees it. Its tick is a roll-call: every eligible row is its own broadcast, dispatched in a plain `for` loop with no gap. Unpaced, N eligible fires produce N back-to-back mesh transmissions; the only downstream protection is RadioSendQueue's ~2.2-2.6s per-transport inter-packet jitter, which prevents packet collision but still lets a roll-call monopolise the mesh. Not currently firing in production (every overdue fire is filtered by terminate_when, so the eligible set is 0) -- this is fire-season hardening against a latent burst, not a live incident. Adds `spacing_seconds` (adapter_config, default 60 to match FirePacer) enforcing a minimum gap between consecutive SUCCESSFUL reminder deliveries. Deliberately a pure spacing change: * WHAT gets broadcast is untouched; nothing is dropped. * The ok-gated last_broadcast_at stamp still uses the tick's `now`. * A failed dispatch sent no packet, so it does not arm the gap. * Rows filtered by terminate_when/render never burn a spacing slot. * A lone eligible fire has nothing to pace against -> zero added latency. * The wait is interruptible by stop(): a 15-fire roll-call holds tick_once() for ~14 min and stop() awaits the tick task, so a plain sleep would stall shutdown. Chose in-loop spacing over routing reminders through FirePacer itself: reminders re-derive their targets from live DB state every tick and only clear a row via last_broadcast_at after a confirmed send, so enqueuing into a 60s-drain FIFO would re-enqueue the same fire on every intervening tick -- the queue would grow faster than it drains. pacer.py, consumer.py, store.py and main.py are untouched. Tests fake the clock end-to-end, so 60s spacing costs the suite nothing. Co-authored-by: Matt Johnson Co-authored-by: Claude Opus 4.8 (1M context) --- work/meshai/adapter_config/defaults.py | 15 + .../notifications/reminders/__init__.py | 90 ++++- work/tests/test_fire_reminder_pacing.py | 354 ++++++++++++++++++ 3 files changed, 458 insertions(+), 1 deletion(-) create mode 100644 work/tests/test_fire_reminder_pacing.py diff --git a/work/meshai/adapter_config/defaults.py b/work/meshai/adapter_config/defaults.py index b14f9b0..3cb5ff4 100644 --- a/work/meshai/adapter_config/defaults.py +++ b/work/meshai/adapter_config/defaults.py @@ -510,6 +510,11 @@ REGISTRY: dict[tuple[str, str], dict[str, Any]] = { "type": "json", "description": "Stop reminding when any of these conditions is true.", }, + ("reminders_wfigs", "spacing_seconds"): { + "default": 60, # matches FirePacer's <=1/60s cadence + "type": "int", + "description": "Minimum seconds between consecutive reminder broadcasts. A roll-call of N eligible fires is spread over N intervals instead of firing back-to-back. 0 disables spacing.", + }, ("reminders_swpc", "cadence_kind"): { "default": "interval", @@ -531,6 +536,11 @@ REGISTRY: dict[tuple[str, str], dict[str, Any]] = { "type": "json", "description": "Stop reminding when any of these conditions is true.", }, + ("reminders_swpc", "spacing_seconds"): { + "default": 60, + "type": "int", + "description": "Minimum seconds between consecutive reminder broadcasts. 0 disables spacing.", + }, ("reminders_itd_511_work_zone", "cadence_kind"): { "default": "clock", @@ -562,6 +572,11 @@ REGISTRY: dict[tuple[str, str], dict[str, Any]] = { "type": "json", "description": "Stop reminding when any of these conditions is true.", }, + ("reminders_itd_511_work_zone", "spacing_seconds"): { + "default": 60, + "type": "int", + "description": "Minimum seconds between consecutive reminder broadcasts. 0 disables spacing.", + }, # NWS dedup-window relaxation (separate from reminders by design). ("nws", "duplicate_allowed_after_seconds"): { diff --git a/work/meshai/notifications/reminders/__init__.py b/work/meshai/notifications/reminders/__init__.py index 4ce2840..30b6067 100644 --- a/work/meshai/notifications/reminders/__init__.py +++ b/work/meshai/notifications/reminders/__init__.py @@ -29,6 +29,27 @@ When a reminder fires: touched here (only the handler sets it on first sight). Quiet hours are NOT respected (Phase-2 deleted that concept). + +Output spacing +-------------- +A reminder tick is a ROLL-CALL: every eligible row for an adapter is a +separate broadcast. Without spacing, N eligible fires produce N back-to-back +``_dispatch()`` calls in one tick -- the reminder path does NOT go through the +EventBus, so ``FirePacer`` (which paces the Central and native fire-event +exits to <=1/60s) never sees it. The only downstream protection is +``RadioSendQueue``'s per-transport inter-packet jitter (~2.2-2.6 s), which +stops packets colliding but does not stop a roll-call from monopolising the +mesh for N x ~2.4 s. + +``spacing_seconds`` (adapter_config, default 60 -- matching FirePacer's +cadence) enforces a minimum gap between consecutive *successful* reminder +deliveries, so a roll-call of N fires is spread over N intervals. It is a +pure spacing change: WHAT gets broadcast, and the ``ok``-gated +``last_broadcast_at`` stamp, are untouched. + +Spacing is measured from the last actual delivery, so a lone eligible fire +(nothing to pace against) goes out with ZERO added latency. The wait is +interruptible by ``stop()`` -- a long roll-call must not stall shutdown. """ from __future__ import annotations @@ -47,6 +68,13 @@ logger = logging.getLogger(__name__) _TICK_SECONDS = 60.0 +# Fallback minimum gap between consecutive reminder broadcasts, used when +# adapter_config has no `spacing_seconds` for the adapter. Mirrors the +# FirePacer interval (notifications/pipeline/pacer.py) so the reminder exit +# and the event exits pace at the same rate. +_DEFAULT_SPACING_SECONDS = 60.0 + + # ============================================================================ # Public scheduler # ============================================================================ @@ -64,6 +92,11 @@ class ReminderScheduler: self._tick = tick_seconds self._task: Optional[asyncio.Task] = None self._stop: Optional[asyncio.Event] = None + # Timestamp of the last SUCCESSFUL reminder delivery, used to space + # consecutive broadcasts. Deliberately shared across adapters (the + # mesh is one shared medium) and across ticks. None => nothing to + # pace against yet, so the next send goes out immediately. + self._last_dispatch_at: Optional[float] = None async def start(self) -> None: if self._task is not None and not self._task.done(): @@ -140,6 +173,38 @@ class ReminderScheduler: logger.debug("reminder %s: unknown cadence_kind=%r", adapter, cfg.cadence_kind) return 0 + async def _space(self, cfg: "_ReminderConfig") -> bool: + """Wait out the inter-broadcast gap before the next dispatch. + + Returns True to proceed with the send, False if `stop()` was signalled + while waiting (caller must abandon the rest of the roll-call). + + No wait happens when nothing has been sent yet (`_last_dispatch_at is + None`) or when the gap has already elapsed on its own -- so a lone + eligible row is never delayed. This is what turns a burst of N + back-to-back sends into N sends spaced `spacing_seconds` apart. + """ + spacing = cfg.spacing_seconds + if spacing <= 0 or self._last_dispatch_at is None: + return True + remaining = spacing - (self._clock() - self._last_dispatch_at) + if remaining <= 0: + return True + # Interruptible wait: a roll-call of N rows holds tick_once() for + # N * spacing seconds, and stop() awaits the tick task -- so a plain + # sleep here would stall shutdown for minutes. Race the sleep against + # the stop event when we have one (production). Tests construct the + # scheduler without start(), leaving _stop None, and get the injected + # self._sleep so they can fake time with zero suite slowdown. + if self._stop is not None: + try: + await asyncio.wait_for(self._stop.wait(), timeout=remaining) + return False # stop signalled -- abandon the roll-call + except asyncio.TimeoutError: + return True # gap elapsed normally + await self._sleep(remaining) + return True + async def _tick_interval(self, adapter: str, cfg: "_ReminderConfig", now: float) -> int: cadence = int(cfg.cadence_value) if cfg.cadence_value else 0 if cadence <= 0: @@ -153,10 +218,15 @@ class ReminderScheduler: wire = self._render(adapter, r, prefix="Active") if not wire: continue + # Space AFTER the terminate/render filters: a row that is skipped + # never transmits, so it must not consume a spacing slot. + if not await self._space(cfg): + break ok = await self._dispatch(adapter, r, wire) if ok: self._stamp_broadcast(adapter, r, now) fired += 1 + self._last_dispatch_at = self._clock() return fired async def _tick_clock(self, adapter: str, cfg: "_ReminderConfig", now: float) -> int: @@ -200,10 +270,13 @@ class ReminderScheduler: continue wire = self._render(adapter, r, prefix="Active") if not wire: continue + if not await self._space(cfg): + break ok = await self._dispatch(adapter, r, wire) if ok: self._stamp_broadcast(adapter, r, now) fired += 1 + self._last_dispatch_at = self._clock() return fired # ---- table-specific helpers -------------------------------------- @@ -377,7 +450,7 @@ class ReminderScheduler: class _ReminderConfig: """Snapshot of the reminders_ config rows.""" __slots__ = ("cadence_kind", "cadence_value", "channels", "terminate_when", - "dow_mask", "timezone") + "dow_mask", "timezone", "spacing_seconds") def __init__(self, **kw): for k in self.__slots__: @@ -392,6 +465,20 @@ class _ReminderConfig: ck = _safe_get(adapter_config, ac_adapter, "cadence_kind") if ck is None: return None + # spacing_seconds: minimum gap between consecutive reminder + # broadcasts. Absent/garbage config must never mean "unpaced" -- + # fall back to the FirePacer-matching default. An explicit 0 (or + # negative) is honored as "spacing disabled". + _sp = _safe_get(adapter_config, ac_adapter, "spacing_seconds") + try: + spacing = (_DEFAULT_SPACING_SECONDS if _sp is None + else float(_sp)) + except (TypeError, ValueError): + logger.warning( + "reminders %s: bad spacing_seconds=%r; using default %.0fs", + adapter, _sp, _DEFAULT_SPACING_SECONDS) + spacing = _DEFAULT_SPACING_SECONDS + return cls( cadence_kind=ck, cadence_value=_safe_get(adapter_config, ac_adapter, "cadence_value"), @@ -399,6 +486,7 @@ class _ReminderConfig: terminate_when=_safe_get(adapter_config, ac_adapter, "terminate_when") or [], dow_mask=_safe_get(adapter_config, ac_adapter, "dow_mask"), timezone=_safe_get(adapter_config, ac_adapter, "timezone"), + spacing_seconds=spacing, ) except Exception: logger.exception("reminders: config load failed for %s", adapter) diff --git a/work/tests/test_fire_reminder_pacing.py b/work/tests/test_fire_reminder_pacing.py new file mode 100644 index 0000000..c2ff045 --- /dev/null +++ b/work/tests/test_fire_reminder_pacing.py @@ -0,0 +1,354 @@ +"""Reminder-path output spacing. + +The ReminderScheduler is the THIRD fire-broadcast exit, and the only one that +does not pass through the EventBus -- so FirePacer (which paces the Central +and native fire-event exits to <=1/60s) never sees it. Its tick is a ROLL-CALL: +every eligible row is its own broadcast, dispatched in a plain `for` loop. +Unpaced, N eligible fires => N back-to-back mesh transmissions. + +These tests pin the spacing behavior: + * N eligible fires do NOT emit back-to-back (spacing waited between sends), + * the gap equals the configured spacing_seconds, + * a LONE eligible fire is not delayed at all (nothing to pace against), + * spacing_seconds=0 restores the old unpaced behavior (kill switch), + * rows filtered out by terminate_when do not consume a spacing slot, + * the `ok`-gated count + last_broadcast_at stamp semantics are UNCHANGED, + * a failed dispatch does not arm the spacing gap (no packet went out). + +Time is faked end-to-end (injected clock + sleep), so the suite pays no +wall-clock cost for a 60s default spacing. +""" +from __future__ import annotations + +import asyncio +import time +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from meshai.notifications.reminders import ReminderScheduler +from meshai.persistence import get_db + + +# ---------- helpers -------------------------------------------------------- + + +def _seed_fire(conn, *, irwin_id, last_broadcast_at, current_contained_pct=10, + last_event_at=None, name="Test Fire"): + if last_event_at is None: + last_event_at = int(time.time()) + conn.execute( + "INSERT OR REPLACE INTO fires(irwin_id, incident_name, incident_type, " + "current_acres, current_contained_pct, lat, lon, county, state, " + "declared_at, last_event_at, first_broadcast_at, last_broadcast_at) " + "VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)", + (irwin_id, name, "WF", 500, current_contained_pct, + 42.5, -114.5, "Cassia", "ID", + last_broadcast_at, last_event_at, last_broadcast_at, last_broadcast_at), + ) + + +def _enable_wfigs_reminders(spacing_seconds=None): + """Enable wfigs reminders; optionally override spacing_seconds.""" + conn = get_db() + for col in ("default_json", "value_json"): + conn.execute( + f"UPDATE adapter_config SET {col}='true' " + "WHERE adapter='reminders_wfigs' AND key='enabled'" + ) + if spacing_seconds is not None: + for col in ("default_json", "value_json"): + conn.execute( + f"UPDATE adapter_config SET {col}=? " + "WHERE adapter='reminders_wfigs' AND key='spacing_seconds'", + (str(int(spacing_seconds)),), + ) + from meshai.adapter_config import adapter_config as _ac + _ac.invalidate() + + +class _FakeTime: + """Injected clock + sleep. `sleep()` advances the clock instead of waiting, + and records every gap it was asked to wait, so a test can assert on the + exact spacing without any wall-clock cost.""" + + def __init__(self, start=1_780_000_000.0): + self.now = start + self.sleeps: list[float] = [] + + def clock(self) -> float: + return self.now + + async def sleep(self, seconds: float) -> None: + self.sleeps.append(seconds) + self.now += seconds + + +@pytest.fixture +def mock_dispatcher(): + d = MagicMock() + d.dispatch_scheduled_broadcast = AsyncMock(return_value=True) + d.dispatch_scheduled_fire_broadcast = AsyncMock(return_value=True) + return d + + +def _sched(dispatcher, ft: _FakeTime) -> ReminderScheduler: + return ReminderScheduler(dispatcher, clock=ft.clock, sleep=ft.sleep) + + +# ============================================================================ +# The burst: N eligible fires must NOT go out back-to-back +# ============================================================================ + + +def test_roll_call_of_n_fires_is_not_back_to_back(mock_dispatcher): + """5 eligible fires => 5 sends, each separated by the configured spacing. + + This is THE regression guard: unpaced, this loop fired all 5 with zero + gaps. Asserting on the send timestamps (not just the sleep calls) proves + the gap is real from the dispatcher's point of view. + """ + ft = _FakeTime() + conn = get_db() + _enable_wfigs_reminders(spacing_seconds=60) + for i in range(5): + _seed_fire(conn, irwin_id=f"BURST{i}", + last_broadcast_at=ft.now - 9 * 3600, + last_event_at=int(ft.now)) + + sent_at: list[float] = [] + mock_dispatcher.dispatch_scheduled_fire_broadcast = AsyncMock( + side_effect=lambda **kw: sent_at.append(ft.now) or True + ) + + fired = asyncio.run(_sched(mock_dispatcher, ft).tick_once()) + + assert fired == 5, "all 5 eligible fires must still broadcast -- none dropped" + assert len(sent_at) == 5 + gaps = [b - a for a, b in zip(sent_at, sent_at[1:])] + assert gaps == [60.0, 60.0, 60.0, 60.0], ( + f"expected a 60s gap between every consecutive send, got {gaps}" + ) + assert ft.sleeps == [60.0] * 4, ( + "spacing must be waited between sends (N-1 waits for N sends), " + f"got {ft.sleeps}" + ) + + +def test_spacing_seconds_is_honored_from_config(mock_dispatcher): + """The gap tracks adapter_config, so it is tunable without a deploy.""" + ft = _FakeTime() + conn = get_db() + _enable_wfigs_reminders(spacing_seconds=15) + for i in range(3): + _seed_fire(conn, irwin_id=f"CFG{i}", + last_broadcast_at=ft.now - 9 * 3600, + last_event_at=int(ft.now)) + + fired = asyncio.run(_sched(mock_dispatcher, ft).tick_once()) + + assert fired == 3 + assert ft.sleeps == [15.0, 15.0] + + +def test_spacing_zero_disables_pacing(mock_dispatcher): + """spacing_seconds=0 is the documented kill switch -> no waits at all.""" + ft = _FakeTime() + conn = get_db() + _enable_wfigs_reminders(spacing_seconds=0) + for i in range(4): + _seed_fire(conn, irwin_id=f"OFF{i}", + last_broadcast_at=ft.now - 9 * 3600, + last_event_at=int(ft.now)) + + fired = asyncio.run(_sched(mock_dispatcher, ft).tick_once()) + + assert fired == 4 + assert ft.sleeps == [] + + +# ============================================================================ +# No added latency when there is nothing to pace against +# ============================================================================ + + +def test_single_eligible_fire_is_not_delayed(mock_dispatcher): + """A lone fire has nothing to pace against -> ZERO added latency. + + Spacing must never become a fixed startup tax on the first broadcast. + """ + ft = _FakeTime() + conn = get_db() + _enable_wfigs_reminders(spacing_seconds=60) + _seed_fire(conn, irwin_id="LONE", last_broadcast_at=ft.now - 9 * 3600, + last_event_at=int(ft.now)) + + t_before = ft.now + fired = asyncio.run(_sched(mock_dispatcher, ft).tick_once()) + + assert fired == 1 + assert ft.sleeps == [], "a lone reminder must not wait for anything" + assert ft.now == t_before, "no simulated time may pass before a lone send" + mock_dispatcher.dispatch_scheduled_fire_broadcast.assert_called_once() + + +def test_elapsed_gap_is_credited_not_re_waited(mock_dispatcher): + """If the spacing gap already elapsed on its own, do not wait again. + + Spacing is measured from the last delivery, so a fire arriving well after + the previous one goes out immediately. + """ + ft = _FakeTime() + conn = get_db() + _enable_wfigs_reminders(spacing_seconds=60) + _seed_fire(conn, irwin_id="ELAPSED", last_broadcast_at=ft.now - 9 * 3600, + last_event_at=int(ft.now)) + + sch = _sched(mock_dispatcher, ft) + # Pretend we delivered something 10 minutes ago -- far past the 60s gap. + sch._last_dispatch_at = ft.now - 600 + + fired = asyncio.run(sch.tick_once()) + + assert fired == 1 + assert ft.sleeps == [], "an already-elapsed gap must not be re-waited" + + +# ============================================================================ +# Spacing must not change WHAT is broadcast -- only its timing +# ============================================================================ + + +def test_terminated_rows_do_not_consume_a_spacing_slot(mock_dispatcher): + """Rows killed by terminate_when never transmit, so they must not add gaps. + + Seeds 2 sendable fires around 2 that terminate (tombstoned / 100% contained). + Correct behavior: 2 sends, exactly 1 gap between them. + """ + ft = _FakeTime() + conn = get_db() + _enable_wfigs_reminders(spacing_seconds=60) + old = ft.now - 9 * 3600 + _seed_fire(conn, irwin_id="LIVE1", last_broadcast_at=old, + last_event_at=int(ft.now)) + _seed_fire(conn, irwin_id="CONTAINED", last_broadcast_at=old, + current_contained_pct=100, last_event_at=int(ft.now)) + _seed_fire(conn, irwin_id="LIVE2", last_broadcast_at=old, + last_event_at=int(ft.now)) + _seed_fire(conn, irwin_id="TOMB", last_broadcast_at=old, + last_event_at=int(ft.now)) + conn.execute("UPDATE fires SET tombstoned_at=? WHERE irwin_id='TOMB'", + (ft.now,)) + + fired = asyncio.run(_sched(mock_dispatcher, ft).tick_once()) + + assert fired == 2, "only the 2 non-terminated fires may broadcast" + assert ft.sleeps == [60.0], ( + "2 sends => exactly 1 gap; a skipped row must not burn a spacing slot" + ) + pks = {c.kwargs["source_event_pk"] + for c in mock_dispatcher.dispatch_scheduled_fire_broadcast.call_args_list} + assert pks == {"LIVE1", "LIVE2"} + + +def test_stamp_and_count_semantics_unchanged_on_success(mock_dispatcher): + """The post-success last_broadcast_at stamp still lands, still uses the + tick's `now` (NOT the post-spacing send time) -- unchanged semantics.""" + ft = _FakeTime() + conn = get_db() + _enable_wfigs_reminders(spacing_seconds=60) + for i in range(2): + _seed_fire(conn, irwin_id=f"STAMP{i}", + last_broadcast_at=ft.now - 9 * 3600, + last_event_at=int(ft.now)) + tick_now = ft.now + + fired = asyncio.run(_sched(mock_dispatcher, ft).tick_once()) + + assert fired == 2 + for i in range(2): + row = conn.execute( + "SELECT last_broadcast_at FROM fires WHERE irwin_id=?", + (f"STAMP{i}",), + ).fetchone() + assert row["last_broadcast_at"] == tick_now, ( + "stamp must still use the tick's `now`, not the delayed send time" + ) + + +def test_failed_dispatch_does_not_stamp_or_arm_the_gap(mock_dispatcher): + """A dispatch returning False sent no packet: it must not stamp + last_broadcast_at, must not count, and must not open a spacing gap.""" + ft = _FakeTime() + conn = get_db() + _enable_wfigs_reminders(spacing_seconds=60) + _seed_fire(conn, irwin_id="FAIL1", last_broadcast_at=ft.now - 9 * 3600, + last_event_at=int(ft.now)) + mock_dispatcher.dispatch_scheduled_fire_broadcast = AsyncMock( + return_value=False + ) + + fired = asyncio.run(_sched(mock_dispatcher, ft).tick_once()) + + assert fired == 0 + assert ft.sleeps == [], "a send that never landed must not arm the gap" + row = conn.execute( + "SELECT last_broadcast_at FROM fires WHERE irwin_id='FAIL1'" + ).fetchone() + assert row["last_broadcast_at"] == ft.now - 9 * 3600, ( + "a failed dispatch must not stamp last_broadcast_at" + ) + + +def test_nothing_is_dropped_in_a_large_roll_call(mock_dispatcher): + """Spacing spreads a roll-call out; it never drops a fire.""" + ft = _FakeTime() + conn = get_db() + _enable_wfigs_reminders(spacing_seconds=60) + n = 15 + for i in range(n): + _seed_fire(conn, irwin_id=f"MANY{i:02d}", + last_broadcast_at=ft.now - 9 * 3600, + last_event_at=int(ft.now)) + + fired = asyncio.run(_sched(mock_dispatcher, ft).tick_once()) + + assert fired == n + assert mock_dispatcher.dispatch_scheduled_fire_broadcast.call_count == n + assert ft.sleeps == [60.0] * (n - 1) + pks = {c.kwargs["source_event_pk"] + for c in mock_dispatcher.dispatch_scheduled_fire_broadcast.call_args_list} + assert pks == {f"MANY{i:02d}" for i in range(n)} + + +# ============================================================================ +# Shutdown must not be held hostage by a long roll-call +# ============================================================================ + + +def test_stop_interrupts_the_spacing_wait(mock_dispatcher): + """A 15-fire roll-call at 60s spacing holds tick_once() for ~14 minutes, + and stop() awaits the tick task -- so the spacing wait MUST abort when the + stop event is set, or shutdown hangs. + """ + conn = get_db() + _enable_wfigs_reminders(spacing_seconds=60) + now = 1_780_000_000 + for i in range(5): + _seed_fire(conn, irwin_id=f"STOP{i}", last_broadcast_at=now - 9 * 3600, + last_event_at=now) + + async def _run() -> int: + sch = ReminderScheduler(mock_dispatcher, clock=lambda: now) + # Simulate the running state start() sets up, then ask it to stop. + # The spacing wait races the stop event, so it must return promptly + # rather than sleeping out the full 60s gap. + sch._stop = asyncio.Event() + sch._stop.set() + return await asyncio.wait_for(sch.tick_once(), timeout=5.0) + + fired = asyncio.run(_run()) + + # The first fire has nothing to pace against and goes out; the roll-call is + # then abandoned at the first spacing wait because stop was signalled. + assert fired == 1