CR: fix/heartbeat-retry-1022

modules/common/biometrics/manifest.v1.json

@@ -635,7 +635,7 @@
   },
   "sourceHashes": {
     "modules/common/biometrics/signal.py": "0b1447c1162245232c053aeeb5cb3e75e7da3392c38a43c465f28596f084b32f",
-    "modules/piezo-processor/main.py": "58e13cbf5fdde9e80c6d62f78713ecead4960b1cb2411caf41b6b60ebb5472c7",
+    "modules/piezo-processor/main.py": "11841a735be9ac73642edd32af8e7b1981ed944e4e65778ed95e06460284b4ba",
     "src/lib/sleep-stages.ts": "b1800e49ce8525d314333da6a29a63131a002d160bb294fa97b24df6e655e228",
     "modules/piezo-processor/runtime_adapter.py": "0a22a2fa05aae6569bfc4b53fba80d75e89f702f021ccb61f82ce64e284c5bb8",
     "modules/common/biometrics/runtime.py": "dbe327ce6e6ae315cad83439903e4c5d6a5c594468086739504caa8ed028a944",

modules/piezo-processor/main.py

@@ -276,6 +276,7 @@ def write_heartbeats(holder: "DBHolder", side: str, chunk: Chunk) -> bool:
 
 # Minutes held for a retry after a failed write; older ones are dropped.
 HEARTBEAT_RETRY_MAX = 30
+HEARTBEAT_RETRY_INTERVAL_S = 5.0
 # A single-sleeper minute waits for the other side's tracker to finish it, but
 # no longer than this past the minute's end (s) — a stuck tracker can't hold
 # the stream back.
@@ -299,10 +300,13 @@ class BeatFront:
     interval spans an absence, and stores the minute in progress."""
 
     def __init__(self, holder: "DBHolder", mode: SingleSleeperMode,
-                 present: Callable[[str], bool] = lambda side: True):
+                 present: Callable[[str], bool] = lambda side: True,
+                 clock: Callable[[], float] = time.monotonic):
         self._holder = holder
         self._mode = mode
         self._present = present
+        self._clock = clock
+        self._retry_after = 0.0
         self._was_present = {"left": False, "right": False}
         self.trackers = {"left": BeatTracker("left"), "right": BeatTracker("right")}
         self._latest_ts: Optional[float] = None
@@ -347,7 +351,15 @@ def gap(self) -> None:
         for t in self.trackers.values():
             t.gap()
 
-    def flush(self) -> None:
+    def tick(self) -> None:
+        """Retry on the reader thread even when no piezo records arrive."""
+        self._drain()
+
+    def flush(self, *, shutdown: bool = False) -> None:
+        # Shutdown gets one final bounded attempt even inside the backoff.
+        # _drain still stops at the first failure; ingestion never bypasses it.
+        if shutdown:
+            self._retry_after = 0.0
         self._write({s: t.take_chunks(flush_before=float("inf"))
                      for s, t in self.trackers.items()}, force=True)
 
@@ -362,11 +374,21 @@ def _store_from(self, side: str, source: str, chunk: Chunk) -> None:
         self._store(side, chunk)
 
     def _store(self, side: str, chunk: Chunk) -> None:
-        if not write_heartbeats(self._holder, side, chunk):
-            self._retry.append((side, chunk))
-            if len(self._retry) > HEARTBEAT_RETRY_MAX:
-                dropped = self._retry.pop(0)
-                log.warning("Dropping unwritten heartbeat minute %s/%d", dropped[0], dropped[1].start)
+        self._retry.append((side, chunk))
+
+    def _drain(self) -> None:
+        # One failed transaction per call, even when new minutes arrive during
+        # an outage. Back off from completion (busy_timeout may itself take 5s).
+        if self._clock() >= self._retry_after:
+            while self._retry:
+                side, chunk = self._retry[0]
+                if not write_heartbeats(self._holder, side, chunk):
+                    self._retry_after = self._clock() + HEARTBEAT_RETRY_INTERVAL_S
+                    break
+                self._retry.pop(0)
+        while len(self._retry) > HEARTBEAT_RETRY_MAX:
+            dropped = self._retry.pop(0)
+            log.warning("Dropping unwritten heartbeat minute %s/%d", dropped[0], dropped[1].start)
 
     def _side_done(self, side: str, start: int) -> bool:
         """True once `side`'s tracker can no longer produce minute `start`."""
@@ -379,22 +401,21 @@ def _side_done(self, side: str, start: int) -> bool:
         return end is not None and end >= start + 60 + STEP_S
 
     def _write(self, chunks: dict, force: bool = False) -> None:
-        retry, self._retry = self._retry, []
-        for side, c in retry:
-            self._store(side, c)
         home = self._mode.home_side()
         if home is None:
             # Leaving single-sleeper mode: held minutes go out as they were
             # queued, before any per-side minute.
             self._release(force=True)
-            for side, cs in chunks.items():
-                for c in cs:
-                    self._store_from(side, side, c)
+            for side, c in sorted(((side, c) for side, cs in chunks.items() for c in cs),
+                                  key=lambda item: item[1].start):
+                self._store_from(side, side, c)
+            self._drain()
             return
         for side, cs in chunks.items():
             for c in cs:
                 self._pending.setdefault(c.start, (home, {}))[1][side] = c
         self._release(force)
+        self._drain()
 
     def _release(self, force: bool) -> None:
         """Store held minutes, oldest first, once neither tracker can still
@@ -517,8 +538,12 @@ def main() -> None:
 
     report_health("healthy", "piezo-processor v2 started")
 
+    def poll() -> None:
+        vitals_router.tick()
+        beat_front.tick()
+
     try:
-        for record in follower.read_records(on_poll=vitals_router.tick):
+        for record in follower.read_records(on_poll=poll):
             rtype = record.get("type")
 
             # Surface genuinely-new firmware types once (blanketReadings, log,
@@ -575,7 +600,7 @@ def main() -> None:
         report_health("down", str(e))
         sys.exit(1)
     finally:
-        beat_front.flush()
+        beat_front.flush(shutdown=True)
         vitals_router.flush()
         db_holder.conn.close()
         log.info("Shutdown complete")

modules/piezo-processor/test_main.py

@@ -2167,11 +2167,13 @@ def test_a_failed_write_is_retried(self):
         from beats import Chunk
         conn = TestWriteVitalsResilience()._make_db()
         holder = main.DBHolder(conn)
-        front = main.BeatFront(holder, _Mode(None))
+        now = [0.0]
+        front = main.BeatFront(holder, _Mode(None), clock=lambda: now[0])
         front._write({"left": [Chunk(60, [0, 1000], 0.8)], "right": []})
         assert holder.write_failures == 1
         conn.execute("CREATE TABLE heartbeats (id INTEGER PRIMARY KEY AUTOINCREMENT, side TEXT, "
                      "timestamp INTEGER, beats TEXT, quality REAL, UNIQUE(side, timestamp))")
+        now[0] += main.HEARTBEAT_RETRY_INTERVAL_S
         front._write({"left": [], "right": []})
         assert _hb_rows(conn) == [("left", 60, [0, 1000], 0.8)]
         assert holder.write_failures == 0
@@ -2306,3 +2308,122 @@ def test_no_hint_without_a_heart_rate(self, monkeypatch):
         assert hints == []
 
 
+
+class TestHeartbeatRetryBackoff:
+    @pytest.mark.parametrize('message', ['database is locked', 'disk I/O error'])
+    def test_outage_backoff_preserves_order_and_limits_reconnects(self, monkeypatch, message):
+        import sqlite3
+        import main
+        from beats import Chunk
+        from unittest.mock import MagicMock
+        now = [100.0]
+        conn = MagicMock()
+        conn.execute.side_effect = sqlite3.OperationalError(message)
+        holder = main.DBHolder(conn)
+        reconnect = MagicMock()
+        monkeypatch.setattr(main, '_reconnect_db', reconnect)
+        front = main.BeatFront(holder, _Mode(None), clock=lambda: now[0])
+        chunks = [Chunk(t, [0, 1000], .8) for t in range(60, 601, 60)]
+        front._write({'left': chunks})
+        assert conn.execute.call_count == 1
+        assert holder.write_failures == 1
+        for _ in range(100):
+            front._write({})
+        assert conn.execute.call_count == 1
+        # Even forced shutdown flush must respect backoff.
+        front.flush()
+        assert conn.execute.call_count == 1
+        for attempt in range(2, main._DB_RECONNECT_THRESHOLD + 1):
+            now[0] += main.HEARTBEAT_RETRY_INTERVAL_S
+            front._write({})
+            assert conn.execute.call_count == attempt
+        assert reconnect.call_count == 1
+        assert holder.write_failures == 0
+        assert [c for _, c in front._retry] == chunks
+        conn.execute.side_effect = None
+        now[0] += main.HEARTBEAT_RETRY_INTERVAL_S
+        front._write({'right': [Chunk(660, [0], .7)]})
+        successful = conn.execute.call_args_list[main._DB_RECONNECT_THRESHOLD:]
+        assert [call.args[1][1] for call in successful] == list(range(60, 661, 60))
+        assert front._retry == []
+        assert holder.write_failures == 0
+
+    def test_failure_after_success_stops_and_capacity_stays_bounded(self, monkeypatch):
+        import main
+        from beats import Chunk
+        from unittest.mock import Mock
+        conn, holder = _hb_db()
+        writer = Mock(side_effect=[True, False])
+        monkeypatch.setattr(main, 'write_heartbeats', writer)
+        front = main.BeatFront(holder, _Mode(None), clock=lambda: 0)
+        front._write({'left': [Chunk(t, [0], .8) for t in range(35)]})
+        assert writer.call_count == 2
+        assert len(front._retry) == main.HEARTBEAT_RETRY_MAX
+        assert [c.start for _, c in front._retry] == list(range(5, 35))
+        front._write({})
+        assert writer.call_count == 2
+        conn.close()
+
+class TestHeartbeatLifecycleRetries:
+    @pytest.mark.parametrize('recovers', [True, False])
+    def test_shutdown_gets_one_final_attempt_inside_backoff(self, monkeypatch, recovers):
+        import main
+        from beats import Chunk
+        from unittest.mock import Mock
+        _, holder = _hb_db()
+        writer = Mock(return_value=False)
+        monkeypatch.setattr(main, 'write_heartbeats', writer)
+        front = main.BeatFront(holder, _Mode(None), clock=lambda: 0)
+        chunks = [Chunk(60, [0], .8), Chunk(120, [0], .8)]
+        front._write({'left': chunks})
+        writer.return_value = recovers
+        front.flush(shutdown=True)
+        assert writer.call_count == (3 if recovers else 2)
+        assert [c for _, c in front._retry] == ([] if recovers else chunks)
+        assert [call.args[2].start for call in writer.call_args_list] == ([60, 60, 120] if recovers else [60, 60])
+        holder.conn.close()
+
+    def test_idle_poll_retries_only_after_deadline(self, monkeypatch):
+        import main
+        from beats import Chunk
+        from unittest.mock import Mock
+        _, holder = _hb_db()
+        writer = Mock(return_value=False)
+        monkeypatch.setattr(main, 'write_heartbeats', writer)
+        now = [0.0]
+        front = main.BeatFront(holder, _Mode(None), clock=lambda: now[0])
+        front._write({'left': [Chunk(60, [0], .8)]})
+        writer.return_value = True
+        now[0] = main.HEARTBEAT_RETRY_INTERVAL_S - .01
+        front.tick()
+        assert writer.call_count == 1
+        now[0] += .01
+        front.tick()
+        assert writer.call_count == 2
+        assert front._retry == []
+        holder.conn.close()
+
+    def test_main_wires_idle_poll_and_shutdown_flush(self, monkeypatch, tmp_path):
+        import main
+        from unittest.mock import MagicMock
+        monkeypatch.setattr(main, 'BIOMETRICS_DB', tmp_path / 'biometrics.db')
+        conn = MagicMock()
+        monkeypatch.setattr(main, 'open_biometrics_db', lambda: conn)
+        for name in ['PumpGate', 'FrzHealthPumpState', 'SideProcessor', 'SingleSleeperMode',
+                     'BedPresence', 'RuntimeCoordinator', 'report_health']:
+            monkeypatch.setattr(main, name, MagicMock())
+        front, vitals = MagicMock(), MagicMock()
+        monkeypatch.setattr(main, 'BeatFront', MagicMock(return_value=front))
+        monkeypatch.setattr(main, 'SingleSleeperVitals', MagicMock(return_value=vitals))
+        follower = MagicMock()
+        def read_records(*, on_poll):
+            on_poll()
+            return iter(())
+        follower.read_records.side_effect = read_records
+        monkeypatch.setattr(main, 'create_follower', MagicMock(return_value=follower))
+        main.main()
+        front.tick.assert_called_once_with()
+        vitals.tick.assert_called_once_with()
+        front.flush.assert_called_once_with(shutdown=True)
+        vitals.flush.assert_called_once_with()
+        conn.close.assert_called_once_with()

src/demo/biometrics/worker-build.json

@@ -1,5 +1,5 @@
 {
-  "buildHash": "d94729b469ff0890dd9ca4531b00679147ebddd89c2574542a1e358a8511dca4",
+  "buildHash": "8f3bfabff0661ae6f38f316217c3b77746b4ef3240614ea2883dfb0127bfa0c1",
   "identity": "esbuild-browser-evaluator-bundle",
   "simulated": true
 }

src/demo/tests/biometrics.referenceVitals.test.ts

@@ -9,16 +9,17 @@ function required<T>(handler: T | undefined): T {
 
 const NOW = Date.UTC(2026, 9, 10, 12)
 
-describe('demo biometrics.getReferenceVitals', () => {
-  beforeEach(() => {
-    vi.useFakeTimers({ toFake: ['Date'] })
-    vi.setSystemTime(NOW)
-  })
+// Every case needs a completed synthetic night, even when CI runs at bedtime.
+beforeEach(() => {
+  vi.useFakeTimers({ toFake: ['Date'] })
+  vi.setSystemTime(NOW)
+})
 
-  afterEach(() => {
-    vi.useRealTimers()
-  })
+afterEach(() => {
+  vi.useRealTimers()
+})
 
+describe('demo biometrics.getReferenceVitals', () => {
   it('returns the Watch overlay for the latest night, ascending on a fixed five-minute grid', async () => {
     const night = demoNights('left')[0]
     const startDate = new Date(night.vitals[0].t)
@@ -77,8 +78,6 @@ describe('demo reference samples and stages', () => {
 })
 
 it.each(['hrv_sdnn', 'breathing_rate'] as const)('keeps %s samples stable across overlapping windows and distinct from pod values', async (metric) => {
-  vi.useFakeTimers({ toFake: ['Date'] })
-  vi.setSystemTime(NOW)
   const night = demoNights('left')[0]
   const input = { side: 'left' as const, startDate: night.enteredBedAt, endDate: night.leftBedAt ?? new Date(), metric }
   const all = await required(biometrics.getReferenceSamples)(input)
@@ -95,5 +94,4 @@ it.each(['hrv_sdnn', 'breathing_rate'] as const)('keeps %s samples stable across
   }
   expect(all.slice(1).some((r, i) => r.timestamp.getTime() - all[i].timestamp.getTime() === 300_000)).toBe(true)
   expect(await required(biometrics.getReferenceSamples)({ ...input, startDate: all[1].timestamp, endDate: all[1].timestamp })).toEqual([all[1]])
-  vi.useRealTimers()
 })