diff --git a/modules/common/biometrics/manifest.v1.json b/modules/common/biometrics/manifest.v1.json index a477d1cab..41c441405 100644 --- a/modules/common/biometrics/manifest.v1.json +++ b/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", diff --git a/modules/piezo-processor/main.py b/modules/piezo-processor/main.py index 3480d4afd..cacd80499 100644 --- a/modules/piezo-processor/main.py +++ b/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") diff --git a/modules/piezo-processor/test_main.py b/modules/piezo-processor/test_main.py index d53947cef..70d1db44b 100644 --- a/modules/piezo-processor/test_main.py +++ b/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() diff --git a/src/demo/biometrics/worker-build.json b/src/demo/biometrics/worker-build.json index 48adb7b77..50cbcdd1c 100644 --- a/src/demo/biometrics/worker-build.json +++ b/src/demo/biometrics/worker-build.json @@ -1,5 +1,5 @@ { - "buildHash": "d94729b469ff0890dd9ca4531b00679147ebddd89c2574542a1e358a8511dca4", + "buildHash": "8f3bfabff0661ae6f38f316217c3b77746b4ef3240614ea2883dfb0127bfa0c1", "identity": "esbuild-browser-evaluator-bundle", "simulated": true } diff --git a/src/demo/tests/biometrics.referenceVitals.test.ts b/src/demo/tests/biometrics.referenceVitals.test.ts index 15f0bd655..1932f3bd7 100644 --- a/src/demo/tests/biometrics.referenceVitals.test.ts +++ b/src/demo/tests/biometrics.referenceVitals.test.ts @@ -9,16 +9,17 @@ function required(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() })