From 351f13ca644fc491a01339a4bce90b3aa19f4719 Mon Sep 17 00:00:00 2001 From: malice Date: Sun, 5 Jul 2026 01:52:00 -0600 Subject: [PATCH] =?UTF-8?q?fix(satpass):=20persist=20consolidation=20sched?= =?UTF-8?q?ule=20=E2=80=94=20survive=20reboot=20(4b)=20(#38)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Satpass consolidations were scheduled with in-memory asyncio timers (_pending_satpass_timers) lost on restart, orphaning satpass_pending rows that never consolidated/broadcast. Persist a durable due_at and rebuild timers on startup. - v22.sql: satpass_pending.due_at INTEGER; SCHEMA_VERSION 21->22 - due_at = received_at + CONSOLIDATION_DELAY(5); the live +N*60 stagger is in-memory only (meaningless across restart) so not persisted — live call_later path unchanged, due_at is a pure reboot backstop - consumer._sweep_pending_satpass() at start(): past-due rows fire (orphans recovered), future rows re-armed for the remaining wait; skips cids the live drain path already owns (no double-schedule); per-row try/except - reuses _satpass_consolidation_fire so emit logic is identical Non-reboot behavior byte-identical. 6 new tests; suite at 10-failure baseline. Co-authored-by: Matt Johnson Co-authored-by: Claude Opus 4.8 (1M context) --- work/meshai/central/consumer.py | 63 ++++++ work/meshai/central/satpass_handler.py | 64 +++++- work/meshai/persistence/db.py | 2 +- work/meshai/persistence/migrations/v22.sql | 9 + work/tests/test_satpass_persisted_timer.py | 242 +++++++++++++++++++++ 5 files changed, 375 insertions(+), 5 deletions(-) create mode 100644 work/meshai/persistence/migrations/v22.sql create mode 100644 work/tests/test_satpass_persisted_timer.py diff --git a/work/meshai/central/consumer.py b/work/meshai/central/consumer.py index 738c064..1992a07 100644 --- a/work/meshai/central/consumer.py +++ b/work/meshai/central/consumer.py @@ -749,6 +749,63 @@ class CentralConsumer: except Exception: logger.exception("satpass consolidation failed for %s", consolidated_id) + def _sweep_pending_satpass(self, now: Optional[float] = None) -> None: + """Reconstruct consolidation timers for satpass_pending rows that a + restart orphaned. + + The live in-memory scheduler (_check_satpass_consolidation) loses its + asyncio TimerHandles when the process exits, but the satpass_pending + rows and their persisted due_at survive. Without this sweep those rows + would sit forever, never consolidated or broadcast. Run once at + startup, it re-arms a timer for each pending consolidated_id off its + durable due_at, reusing the SAME _satpass_consolidation_fire path so + the emit logic is byte-identical to normal operation. + + Idempotent and additive: it SKIPS any cid already armed by the live + path (present in _pending_satpass_timers), so it can never + double-schedule and is safe to call once alongside the module-set + drain path (which covers the live case this sweep cannot). + """ + try: + from meshai.central.satpass_handler import load_pending_schedule + schedule = load_pending_schedule() + except Exception: + logger.exception("satpass sweep: failed to load pending schedule") + return + if not schedule: + return + try: + loop = asyncio.get_event_loop() + except Exception: + logger.exception("satpass sweep: no event loop for timer reconstruction") + return + now = time.time() if now is None else now + recovered = 0 + overdue = 0 + for cid, due_at in schedule: + try: + if cid in self._pending_satpass_timers: + # Live path already armed this cid; never double-schedule. + continue + if due_at <= now: + # Orphan already past due: fire soon, with a small + # increasing stagger so a backlog doesn't emit in one burst. + delay = 0.5 + overdue * 2.0 + overdue += 1 + else: + # Not yet due: reconstruct the remaining wait exactly. + delay = due_at - now + handle = loop.call_later( + delay, self._satpass_consolidation_fire, cid) + self._pending_satpass_timers[cid] = handle + recovered += 1 + except Exception: + logger.exception("satpass sweep: failed to re-arm timer for %s", cid) + if recovered: + logger.info( + "satpass sweep: reconstructed %d consolidation timer(s) after restart", + recovered) + async def _on_message(self, msg, owned=None) -> None: """JetStream callback: normalize + emit, then ack. @@ -821,6 +878,12 @@ class CentralConsumer: logger.info("CentralConsumer started; %d subjects subscribed (drain mode active)", len(subject_owned)) + # Reboot recovery: re-arm consolidation timers for any satpass_pending + # rows the previous process left behind (in-memory TimerHandles don't + # survive a restart). Additive to the live module-set path; guarded + # against double-scheduling; a bad row never aborts startup. + self._sweep_pending_satpass() + # Schedule drain timeout: if no messages trigger drain completion # within the window (e.g. empty backlog), auto-exit drain mode. asyncio.get_event_loop().call_later( diff --git a/work/meshai/central/satpass_handler.py b/work/meshai/central/satpass_handler.py index 7367020..b62a91a 100644 --- a/work/meshai/central/satpass_handler.py +++ b/work/meshai/central/satpass_handler.py @@ -54,6 +54,19 @@ _TZ = ZoneInfo("America/Boise") # Consumer polls this after each satpass _normalize() call. _pending_consolidation_ids: set[str] = set() +# Baseline consolidation delay, in seconds, from a pending row's arrival to +# when its consolidated broadcast should fire. This is the DURABLE fire-time +# basis persisted as satpass_pending.due_at (= received_at + this). +# +# It matches the live consumer's baseline: the consumer schedules the timer +# at `5.0 + N*60` where N is the count of OTHER in-flight timers (a runtime +# anti-thundering-herd stagger). The `+N*60` term depends on transient +# in-memory scheduler state that has no meaning across a restart, so it is +# deliberately NOT persisted; only the N=0 baseline (5s) is durable. The +# live in-memory timer still drives normal operation exactly as before — +# due_at is purely the reboot-recovery backstop the in-memory timer can't be. +CONSOLIDATION_DELAY = 5 + def drain_pending_consolidation_ids() -> set[str]: """Atomically drain and return all pending consolidation IDs.""" @@ -404,14 +417,19 @@ def handle_satpass(envelope: dict, subject: str, subject=subject, handled=0, table_name="satpass_pending", table_pk=f"{consolidated_id}:{observer}") - # Accumulate into pending table + # Accumulate into pending table. due_at is the durable fire-time backstop + # (received_at + baseline delay) so a restart can reconstruct a consolidation + # timer for rows the in-memory scheduler would otherwise orphan. + due_at = now + CONSOLIDATION_DELAY conn.execute( "INSERT OR REPLACE INTO satpass_pending(" "consolidated_id, observer, sat_name, norad_id, max_elevation, " - "aos_at, los_at, aos_compass, los_compass, peak_compass, received_at) " - "VALUES (?,?,?,?,?,?,?,?,?,?,?)", + "aos_at, los_at, aos_compass, los_compass, peak_compass, received_at, " + "due_at) " + "VALUES (?,?,?,?,?,?,?,?,?,?,?,?)", (consolidated_id, observer, sat_name, norad_id, max_el, - aos_epoch, los_epoch, aos_compass, los_compass, direction, now)) + aos_epoch, los_epoch, aos_compass, los_compass, direction, now, + due_at)) # Signal consumer to schedule consolidation timer _pending_consolidation_ids.add(consolidated_id) @@ -602,6 +620,44 @@ CREATE TABLE IF NOT EXISTS satpass_pending ( los_compass TEXT, peak_compass TEXT, received_at INTEGER, + due_at INTEGER, PRIMARY KEY (consolidated_id, observer) ); """ + + +def load_pending_schedule() -> list[tuple[str, int]]: + """Return [(consolidated_id, due_at)] for every cid with pending rows. + + Used by the consumer's startup sweep to reconstruct consolidation timers + that were lost with the in-memory scheduler on restart. One entry per + distinct consolidated_id, keyed on the EARLIEST due_at across its observer + rows (MIN) so the reconstructed fire time matches the live timer, which is + armed off the first arrival and never re-armed for later observers. + + A row written before due_at existed (pre-v22, or a partial write) has + due_at IS NULL; COALESCE falls it back to received_at + baseline delay so + such a row is still recoverable rather than silently stranded. + """ + try: + conn = get_db() + except Exception: + logger.exception("satpass sweep: persistence unavailable") + return [] + rows = conn.execute( + "SELECT consolidated_id, " + "MIN(COALESCE(due_at, received_at + ?)) AS due_at " + "FROM satpass_pending GROUP BY consolidated_id", + (CONSOLIDATION_DELAY,), + ).fetchall() + out: list[tuple[str, int]] = [] + for r in rows: + try: + cid = r["consolidated_id"] + due = r["due_at"] + if cid is None or due is None: + continue + out.append((str(cid), int(due))) + except Exception: + logger.exception("satpass sweep: skipping malformed pending row") + return out diff --git a/work/meshai/persistence/db.py b/work/meshai/persistence/db.py index d16dbea..99f4ec3 100644 --- a/work/meshai/persistence/db.py +++ b/work/meshai/persistence/db.py @@ -30,7 +30,7 @@ logger = logging.getLogger(__name__) DEFAULT_DB_PATH = "/data/meshai.sqlite" MESHAI_DB_PATH_ENV = "MESHAI_DB_PATH" -SCHEMA_VERSION = 21 +SCHEMA_VERSION = 22 SCHEMA_META_TABLE = "schema_meta" MIGRATIONS_DIR = Path(__file__).parent / "migrations" diff --git a/work/meshai/persistence/migrations/v22.sql b/work/meshai/persistence/migrations/v22.sql new file mode 100644 index 0000000..ac9d3f7 --- /dev/null +++ b/work/meshai/persistence/migrations/v22.sql @@ -0,0 +1,9 @@ +-- v22: due_at column on satpass_pending. +-- Persists the epoch-second fire time (received_at + baseline consolidation +-- delay) for each pending observer row so a restart can reconstruct the +-- consolidation timers that previously lived only as in-memory asyncio +-- TimerHandles and were orphaned on reboot. Nullable; legacy pending rows +-- (there are none at rest — the table is drained per bucket) keep NULL and +-- the startup sweep falls back to received_at + baseline delay for them. + +ALTER TABLE satpass_pending ADD COLUMN due_at INTEGER; diff --git a/work/tests/test_satpass_persisted_timer.py b/work/tests/test_satpass_persisted_timer.py new file mode 100644 index 0000000..04438ad --- /dev/null +++ b/work/tests/test_satpass_persisted_timer.py @@ -0,0 +1,242 @@ +"""Tests for the satpass persisted-timer reboot-recovery fix. + +Pending satellite-pass consolidations used to be scheduled only as in-memory +asyncio TimerHandles, so a restart orphaned any satpass_pending rows: the row +survived but its timer did not, and it was never consolidated/broadcast. + +The fix persists a durable `due_at` on each pending row and adds a startup +sweep (`CentralConsumer._sweep_pending_satpass`) that reconstructs a timer for +every pending consolidated_id off its persisted due_at, reusing the existing +`_satpass_consolidation_fire` emit path. + +These tests cover: + - a PAST-due orphan is recovered (its timer fires -> consolidation invoked) + - a FUTURE-due row is scheduled, NOT fired immediately + - `due_at` is persisted on the normal ingest path + - SCHEMA_VERSION == 22 and the v22 migration applies cleanly on a fresh DB +""" +from __future__ import annotations + +import asyncio +import json +import time +import types + +import pytest + +from meshai.persistence import get_db, init_db, SCHEMA_VERSION +from meshai.adapter_config import invalidate_cache + + +# ── helpers ─────────────────────────────────────────────────────────── + +def _enable_satpass_db(norad_ids=(25544,), dry_run=True): + """Enable satpass and set opt-in norad_ids in the test DB.""" + conn = get_db() + conn.execute("UPDATE adapter_config SET value_json='true' " + "WHERE adapter='satpass' AND key='enabled'") + conn.execute("UPDATE adapter_config SET value_json=? " + "WHERE adapter='satpass' AND key='dry_run'", + (json.dumps(bool(dry_run)),)) + conn.execute("UPDATE adapter_config SET value_json=? " + "WHERE adapter='satpass' AND key='norad_ids'", + (json.dumps(list(norad_ids)),)) + invalidate_cache() + + +def _insert_pending(consolidated_id, *, due_at, observer="Boise", + norad_id=25544, received_at=None): + """Write a single satpass_pending row with an explicit due_at.""" + conn = get_db() + now = int(time.time()) if received_at is None else received_at + aos = now + 600 + los = aos + 360 + conn.execute( + "INSERT OR REPLACE INTO satpass_pending(" + "consolidated_id, observer, sat_name, norad_id, max_elevation, " + "aos_at, los_at, aos_compass, los_compass, peak_compass, received_at, " + "due_at) VALUES (?,?,?,?,?,?,?,?,?,?,?,?)", + (consolidated_id, observer, "ISS", norad_id, 72.5, + aos, los, "SW", "NE", "S", now, due_at)) + + +def _make_consumer(bus=None): + """Construct a CentralConsumer with minimal fakes (no NATS needed).""" + from meshai.central.consumer import CentralConsumer + env = types.SimpleNamespace(central=None) + return CentralConsumer(env, bus) + + +def _run(coro): + return asyncio.new_event_loop().run_until_complete(coro) + + +def _ingest_envelope(norad_id=25544, observer="Boise", max_el=72.5, + aos="2026-06-12T03:32:00Z", los="2026-06-12T03:38:00Z"): + return { + "specversion": "1.0", + "type": "central.sat.pass", + "source": "central", + "id": f"pass-{norad_id}-{aos}", + "data": { + "adapter": "n2yo_visualpasses", + "category": "pass.n2yo_visualpasses", + "data": { + "norad_id": norad_id, + "satellite_name": "ISS", + "observer_name": observer, + "max_elevation_deg": max_el, + "aos_time": aos, + "los_time": los, + "azimuth_at_peak_compass": "S", + "azimuth_at_aos_compass": "SW", + "azimuth_at_los_compass": "NE", + }, + }, + } + + +# ── schema / migration ─────────────────────────────────────────────── + +def test_schema_version_is_22(): + assert SCHEMA_VERSION == 22 + + +def test_v22_migration_applies_and_adds_due_at_column(tmp_path, monkeypatch): + """Fresh DB migrates cleanly to v22 and satpass_pending has due_at.""" + from meshai.persistence import close_thread_connection + from meshai.persistence import db as persistence_db + db = str(tmp_path / "fresh-v22.sqlite") + monkeypatch.setenv("MESHAI_DB_PATH", db) + persistence_db._initialised.clear() + close_thread_connection() + conn = init_db() + row = conn.execute("SELECT value FROM schema_meta WHERE key='version'").fetchone() + assert int(row["value"]) == 22 + cols = {r["name"] for r in conn.execute("PRAGMA table_info(satpass_pending)")} + assert "due_at" in cols + close_thread_connection() + persistence_db._initialised.discard(db) + + +# ── due_at persisted on normal ingest ──────────────────────────────── + +def test_due_at_persisted_on_normal_ingest(): + """handle_satpass writes due_at = received_at + CONSOLIDATION_DELAY.""" + from meshai.central.satpass_handler import ( + handle_satpass, CONSOLIDATION_DELAY, _parse_iso_epoch) + _enable_satpass_db(norad_ids=[25544], dry_run=True) + for attr in ("_disabled_logged", "_no_norad_ids_logged"): + if hasattr(handle_satpass, attr): + delattr(handle_satpass, attr) + + env = _ingest_envelope() + aos_epoch = _parse_iso_epoch("2026-06-12T03:32:00Z") + now = aos_epoch - 300 # inside horizon, before los + assert handle_satpass(env, "central.sat.pass.iss", now=now) is None + + conn = get_db() + row = conn.execute( + "SELECT received_at, due_at FROM satpass_pending " + "WHERE norad_id=25544").fetchone() + assert row is not None, "ingest did not write a pending row" + assert row["due_at"] is not None + assert row["due_at"] == row["received_at"] + CONSOLIDATION_DELAY + assert row["due_at"] == now + CONSOLIDATION_DELAY + + +# ── startup sweep: past-due orphan is recovered ────────────────────── + +def test_sweep_recovers_past_due_orphan(monkeypatch): + """A pending row with due_at in the PAST fires consolidation via the sweep.""" + _enable_satpass_db(norad_ids=[25544], dry_run=True) + now = int(time.time()) + cid = "25544:ORPHAN" + _insert_pending(cid, due_at=now - 100, received_at=now - 105) + + fired = [] + import meshai.central.satpass_handler as sh + real = sh.consolidate_satpass_pending + + def _spy(consolidated_id): + fired.append(consolidated_id) + return real(consolidated_id) # exercise the real path (dry-run -> None) + + monkeypatch.setattr(sh, "consolidate_satpass_pending", _spy) + + consumer = _make_consumer(bus=None) + + async def _main(): + consumer._sweep_pending_satpass(now=now) + # overdue orphan is armed at ~0.5s; give the loop time to fire it. + await asyncio.sleep(1.0) + + _run(_main()) + + assert cid in fired, "sweep did not fire consolidation for the orphaned cid" + # Orphan recovered: consolidation (dry-run) drained its pending rows. + conn = get_db() + remaining = conn.execute( + "SELECT COUNT(*) AS n FROM satpass_pending WHERE consolidated_id=?", + (cid,)).fetchone()["n"] + assert remaining == 0 + + +# ── startup sweep: future row scheduled, not fired now ─────────────── + +def test_sweep_schedules_future_row_without_firing(monkeypatch): + """A pending row with due_at in the FUTURE is armed but does not fire yet.""" + _enable_satpass_db(norad_ids=[25544], dry_run=True) + now = int(time.time()) + cid = "25544:FUTURE" + _insert_pending(cid, due_at=now + 3600, received_at=now) + + fired = [] + import meshai.central.satpass_handler as sh + monkeypatch.setattr(sh, "consolidate_satpass_pending", + lambda c: fired.append(c)) + + consumer = _make_consumer(bus=None) + + async def _main(): + consumer._sweep_pending_satpass(now=now) + await asyncio.sleep(0.3) + + _run(_main()) + + assert cid not in fired, "future row fired immediately" + assert cid in consumer._pending_satpass_timers, "future row was not armed" + # Pending row untouched (still awaiting its future fire). + conn = get_db() + remaining = conn.execute( + "SELECT COUNT(*) AS n FROM satpass_pending WHERE consolidated_id=?", + (cid,)).fetchone()["n"] + assert remaining == 1 + + +# ── sweep does not double-schedule an already-armed cid ────────────── + +def test_sweep_does_not_double_schedule(monkeypatch): + """A cid already armed by the live path is skipped by the sweep.""" + _enable_satpass_db(norad_ids=[25544], dry_run=True) + now = int(time.time()) + cid = "25544:ARMED" + _insert_pending(cid, due_at=now - 10, received_at=now - 15) + + consumer = _make_consumer(bus=None) + + fired = [] + import meshai.central.satpass_handler as sh + monkeypatch.setattr(sh, "consolidate_satpass_pending", + lambda c: fired.append(c)) + + async def _main(): + sentinel = object() + consumer._pending_satpass_timers[cid] = sentinel # live path owns it + consumer._sweep_pending_satpass(now=now) + # The sweep must not have replaced the live handle. + assert consumer._pending_satpass_timers[cid] is sentinel + await asyncio.sleep(0.1) + + _run(_main()) + assert cid not in fired, "sweep double-scheduled an already-armed cid"