fix(satpass): persist consolidation schedule — survive reboot (4b)

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: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

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(

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

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"
 

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;

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"