matt/meshai
1
0
Fork 0
mirror of https://github.com/zvx-echo6/meshai.git synced 2026-08-26 17:31:34 +00:00
Code Issues Projects Releases Packages Wiki Activity

fix(reminders): pace the reminder roll-call so N fires don't burst (#130)

Browse source 
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 <mj@k7zvx.com>
Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
malice 2026-07-14 10:38:07 -06:00 • committed by GitHub
commit 4fd431f907
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 458 additions and 1 deletions
Download patch file Download diff file Expand all files Collapse all files

15
work/meshai/adapter_config/defaults.py
View file

@ -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"): {

90
work/meshai/notifications/reminders/__init__.py
View file

@ -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_<adapter> 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)

354
work/tests/test_fire_reminder_pacing.py Normal file
View file

@ -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