diff --git a/lib/ble/ble_engine.dart b/lib/ble/ble_engine.dart index f6184d9..6a42086 100644 --- a/lib/ble/ble_engine.dart +++ b/lib/ble/ble_engine.dart @@ -155,6 +155,33 @@ bool burstPacketCountMatches({ }) => expectedPacketCount == actualBurstPacketCount + droppedThisBurst; +/// Honest burst-completeness signal for TELEMETRY ONLY — this NEVER gates the +/// commit/ACK decision (see the log-only call site). +/// +/// [receivedTrafficCount] is every frame we actually received this burst, ALL +/// types (historical R24 data + interleaved console/event/unknown) — i.e. +/// [BurstStats.totalTrafficPacketCount], NOT the banked historical subset. The +/// band's [expectedPacketCount] (num_packets) likewise counts every frame it +/// transmitted, so comparing the two all-types totals is type-agnostic and +/// interleaving-immune: benign console/event frames riding along cannot fake a +/// shortfall the way comparing against the R24-only subset did. +/// +/// [droppedThisBurst] (RecordGate plausibility rejections this burst) is added +/// back because the band counted those frames but they never entered +/// [receivedTrafficCount]. A POSITIVE result is frames the band counted that we +/// did NOT count as valid received traffic — i.e. missing OR corrupted traffic +/// (would-flag / potential loss): CRC-failed frames also never enter +/// [receivedTrafficCount], so a positive shortfall cannot by itself prove a +/// frame never arrived. Zero is complete; negative just means we tallied more +/// than expected (retried/duplicate frames), which is not loss. +@visibleForTesting +int burstPacketShortfall({ + required int expectedPacketCount, + required int receivedTrafficCount, + int droppedThisBurst = 0, +}) => + expectedPacketCount - (receivedTrafficCount + droppedThisBurst); + /// Fired for every LIVE high-rate frame (0x28/0x2B/0x33). These are EPHEMERAL — /// they are NOT persisted to raw_records (that bloated storage ~50x and stalled /// derivation). The caller routes them to an in-memory sink for the live UI / @@ -660,6 +687,28 @@ class BleEngine { @visibleForTesting void debugBeginConnectSetup() => _connectSetup = true; + /// Feed a decoded control frame straight into the state absorber. + /// + /// The response-driven clock policy (trust verdict, bounded SET_CLOCK + /// re-issue) lives on the far side of a real radio, so without this the only + /// coverage possible was of the pure [ClockPolicy] predicates — never of the + /// engine wiring that decides whether to act on them. + @visibleForTesting + void debugAbsorbDecoded(Decoded d) => _absorbState(d); + + /// Age the clock-suspicion start past [ClockPolicy.suspectGraceSeconds]. + /// + /// The grace window is 12 real hours off a monotonic stopwatch, so the only + /// alternative to a seam here is not covering post-grace behaviour at all — + /// and post-grace is precisely where history un-defers onto a strap RTC we + /// have just concluded is the fast one. + @visibleForTesting + void debugExpireClockSuspicion() { + if (_phoneClockSuspectSince == null) return; + _phoneClockSuspectSince = + _monotonicSecs() - ClockPolicy.suspectGraceSeconds - 1; + } + /// Told by AppState on every foreground/background transition. Drives the /// connection interval — see [desiredLinkPriority]. void setBackground(bool value) { @@ -820,6 +869,31 @@ class BleEngine { // Lifetime count of GET_CLOCK `clock_epoch` reads rejected by the same gate // (ClockPolicy.acceptsClockRead) — see the clock_epoch handler below. int _corruptClockReadCount = 0; + // True when the last GET_CLOCK showed a plausible strap RTC reading > 1 day in + // the FUTURE relative to the phone — the phone clock is likely wrong (slow), so + // history offload is DEFERRED (not drained-and-trimmed) until the clocks agree. + // See ClockPolicy.phoneClockSuspect and _startHistoricalRefresh. + bool _phoneClockSuspect = false; + /// MONOTONIC seconds ([_monotonicSecs]) at which the suspicion started — not a + /// wall `DateTime`. The whole point of this state is that the wall clock is + /// not trusted: timing the grace window off `DateTime.now()` lets the very + /// jump we are waiting for (the phone stepping forward over NTP, possibly + /// still >1 day behind the strap) expire the window instantly and hand back + /// permission to drain-and-trim under a clock we still don't trust. + double? _phoneClockSuspectSince; + bool get historyPausedForClock => _deferForClock; + /// Defer history only while the disagreement is still young. A slow phone + /// re-syncs over NTP in minutes; one that persists past the grace window is a + /// strap RTC running fast, and deferring forever would stall sync for good. + bool get _deferForClock => + _phoneClockSuspect && + !ClockPolicy.suspectGraceExpired( + _phoneClockSuspectSince, _monotonicSecs()); + int _clockPausedOffloads = 0; // diagnostics: offloads deferred for this reason + /// Completes when the `clock_epoch` for the GET_CLOCK issued by [_readClock] + /// has been absorbed, so the clock gates read THIS session's verdict instead + /// of whatever the last connection left behind. + Completer? _clockReadPending; DateTime? _bondTime; // when the handshake completed (bond confirmed) DateTime? _armTime; // when live (R10/R11) streams were last armed // Run-state for a chain of auto-continued offload rounds: how many @@ -1311,10 +1385,19 @@ class BleEngine { _clockCorrectTries = 0; // fresh retry budget for this connection // Drop the previous session's clock correlation so an alarm armed before // THIS session's GET_CLOCK reply lands falls back to the raw wall epoch - // (drift 0) instead of the stale strap-RTC frame. setClock()→getClock() - // below repopulates it for this connection. + // (drift 0) instead of the stale strap-RTC frame. The reads below + // repopulate it for this connection. _clockRef = null; - await setClock(); + // READ BEFORE WRITE. This used to be an unconditional SET_CLOCK, which is + // precisely the write [ClockPolicy.phoneClockSuspect] says we must never + // make: on a phone running >1 day slow it stamps that slow time onto a + // CORRECT strap RTC — and worse, it destroys the evidence, because the + // read-back then "agrees" and every later suspect-clock gate sees a + // healthy pair. Read first; skip the write while the PHONE is the suspect + // one. Unset/behind/garbage-low RTCs are unaffected (not suspect) and are + // still corrected here and by the clock_epoch handler's bounded re-issue. + await _readClock(); + if (!_deferForClock) await setClock(); _lastClockVerifyAt = DateTime.now(); // Per-connection policy reset. Marginal-radio + post-bond-loop are NOT reset // here — they count consecutive bad cycles across reconnects and self-reset on @@ -1393,9 +1476,25 @@ class BleEngine { ); _setPhase(BleConnState.listening); _log('Connected + subscribed — listening (history + live).'); - _setOffloadActive(true); - _lastBackfillAt = _wallSecs(); - await sendInit(); // triggers the historical offload flood + // INIT seq4 IS SEND_HISTORICAL_DATA, so it needs the SAME data-safety gate + // as _startHistoricalRefresh — without it every fresh connection drains + // and trims under exactly the untrustworthy phone clock we refuse to drain + // under there, which is the common case (a dead-battery reboot lands a bad + // clock and a reconnect together). + final drainOnInit = !_deferForClock; + if (!drainOnInit) { + _clockPausedOffloads++; + _log( + '[SYNC] INIT drain DEFERRED — phone clock appears wrong relative to ' + 'the strap RTC; not draining history until they agree ' + '(deferred_total=$_clockPausedOffloads).', + ); + } + _setOffloadActive(drainOnInit); + // Only a real drain spends the backfill floor; a deferred one leaves it + // open so a foreground trigger can retry as soon as the phone corrects. + if (drainOnInit) _lastBackfillAt = _wallSecs(); + await sendInit(drain: drainOnInit); // seq4 triggers the offload flood return true; } catch (e) { _log('connect setup failed: $e'); @@ -1508,13 +1607,22 @@ class BleEngine { )) { return false; } + // Spend the floor OPTIMISTICALLY so two triggers racing into the await + // below can't both slip past `shouldRun`, then hand it back if the refresh + // asked the strap for nothing. Without the hand-back, a refresh deferred + // for a suspect clock bought the next attempt a full backfill interval of + // silence — so a phone that corrected itself seconds later still sat + // blocked, which is exactly the window the deferral is short enough to + // ride out. + final floorBefore = _lastBackfillAt; _lastBackfillAt = _wallSecs(); - await _startHistoricalRefresh( + final sent = await _startHistoricalRefresh( trigger: trigger, reason: trigger.name, refreshRange: true, ); - return true; + if (!sent) _lastBackfillAt = floorBefore; + return sent; } /// Foreground catch-up pull: the app came back to the foreground on a healthy @@ -1543,18 +1651,24 @@ class BleEngine { /// This keeps periodic sync, manual resync, workout-end backfill, and future /// callers on the same protocol path instead of each open-coding their own /// "maybe just send 0x16" behavior. - Future _startHistoricalRefresh({ + /// + /// Returns whether `SEND_HISTORICAL_DATA` actually went out. Callers use it to + /// decide whether the attempt was worth spending a rate-limit floor on — a + /// refresh that dropped out at one of the gates below asked the strap for + /// nothing, so it must not buy the next real attempt fifteen minutes of + /// silence. + Future _startHistoricalRefresh({ required BackfillTrigger trigger, required String reason, bool refreshRange = true, }) async { final d = _drain; - if (_session?.connected != true || d == null) return; + if (_session?.connected != true || d == null) return false; if (_offloadActive && !d._complete) { _log( '[SYNC] refresh($reason) dropped — strap is already transmitting history.', ); - return; + return false; } d.rearm(); _setOffloadActive(true); @@ -1565,6 +1679,26 @@ class BleEngine { // has time to emit the range response before we request another drain. await Future.delayed(const Duration(milliseconds: 120)); } + // Data-safety gate: never drain-and-trim history under an untrustworthy phone + // clock. Poll the strap RTC and compare; if the phone clock looks slow (strap + // plausible but > 1 day ahead), DEFER — draining now would drop the strap's + // real records as "future" and the ACK would trim them off the band forever. + // The strap retains everything; we drain on a later refresh once the clocks + // agree (the phone's clock almost always self-corrects via NTP). SET_CLOCK is + // deliberately NOT issued here — pushing the strap back to the slow phone + // would corrupt a correct RTC (see ClockPolicy.phoneClockSuspect). + await _readClock(); + if (_session?.connected != true) return false; + if (_deferForClock) { + _clockPausedOffloads++; + _log( + '[SYNC] refresh($reason) DEFERRED — phone clock appears wrong relative ' + 'to the strap RTC; not draining history until they agree ' + '(deferred_total=$_clockPausedOffloads).', + ); + _setOffloadActive(false); + return false; + } final wait = HistoricalSyncCommandPolicy.waitSeconds( _lastHistoricalSendAt, _wallSecs(), @@ -1575,11 +1709,19 @@ class BleEngine { 'for the 0x16 floor.', ); await Future.delayed(Duration(milliseconds: (wait * 1000).ceil())); - if (_session?.connected != true) return; + if (_session?.connected != true) return false; } _log('[SYNC] refresh($reason) — sending SEND_HISTORICAL_DATA.'); - await _send(Cmd.sendHistoricalData, const [0x00]); + // `_send` swallows write failures and reports them as false. Claiming + // success anyway leaves the strap with no request, `_offloadActive` stuck + // true — so later refreshes bounce off the "already transmitting" guard — + // and both rate-limit floors spent on a command that never left the phone. + if (!await _send(Cmd.sendHistoricalData, const [0x00])) { + _setOffloadActive(false); + return false; + } _lastHistoricalSendAt = _wallSecs(); + return true; } Future _subscribe( @@ -2209,6 +2351,33 @@ class BleEngine { if (f.containsKey('clock_epoch')) { final dev = f['clock_epoch'] as int; final wall = DateTime.now().millisecondsSinceEpoch ~/ 1000; + // Assess phone-clock trust from the RAW read, before the alarm-safety gate + // below diverts a future reading. A plausible strap RTC that reads > 1 day + // ahead of the phone means the phone clock is likely slow — history offload + // then DEFERS (see _startHistoricalRefresh) instead of dropping the strap's + // real records as "future" and trimming them off the band. Cleared the + // moment a read agrees (the phone almost always self-corrects via NTP). + final wasSuspect = _phoneClockSuspect; + _phoneClockSuspect = ClockPolicy.phoneClockSuspect(dev, wall); + if (_phoneClockSuspect && !wasSuspect) { + _phoneClockSuspectSince = _monotonicSecs(); + } else if (!_phoneClockSuspect) { + _phoneClockSuspectSince = null; + } + // Release any gate waiting on THIS read (see [_readClock]). Done as soon + // as the verdict above is settled, before the alarm-correlation work + // below, because the verdict is all a gate is waiting for. + final pendingRead = _clockReadPending; + if (pendingRead != null && !pendingRead.isCompleted) { + pendingRead.complete(); + } + if (_phoneClockSuspect != wasSuspect) { + _log(_phoneClockSuspect + ? '[SYNC] Phone clock appears wrong: strap RTC=$dev is > 1 day ahead ' + 'of phone wall=$wall — DEFERRING history offload until they agree.' + : '[SYNC] Phone/strap clocks agree again (strap=$dev wall=$wall) — ' + 'history offload may resume.'); + } // SANITY GATE, mirroring the one `range_newest` gets below. An // implausibly far-future `clock_epoch` yields a large NEGATIVE driftSec, // and setAlarm arms at `when - driftSec` — years out, where the alarm @@ -2228,29 +2397,55 @@ class BleEngine { } else { _clockRef = ClockRef(device: dev, wall: wall); _log('Clock correlated: device=$dev wall=$wall (drift=${wall - dev}s).'); - // Re-issue SET_CLOCK if the strap RTC has drifted > 1 day or is unset — - // but BOUND the retries: setClock() reads the clock back, so an - // unbounded re-issue on a firmware that never latches either payload - // form would spin SET_CLOCK/GET_CLOCK forever. Historical records carry - // their own embedded unix time regardless, so giving up after a few - // tries is safe. - if (ClockPolicy.shouldSetClock(dev, wall)) { - if (_clockCorrectTries < 3) { - _clockCorrectTries++; - _log( - 'Clock drift over policy — re-issuing SET_CLOCK ' - '(attempt $_clockCorrectTries/3).', - ); - unawaited(setClock()); - } else { - _log( - 'Clock still off after 3 SET_CLOCK attempts — giving up; ' - 'firmware may not accept our payload length.', - ); - } + } + // CORRECTION RUNS ON THE RAW READ, outside the correlation gate above. + // + // It used to be nested inside the accepted-read branch, which quietly + // made a fast strap RTC unfixable: `acceptsClockRead` rejects anything + // past `wall + kFutureMargin` and `phoneClockSuspect` trips past that + // SAME margin, so the one reading that means "the strap clock is ahead" + // could never reach the one code path that fixes it. History would + // un-defer at grace expiry — having concluded the STRAP is the fast one — + // straight back onto an uncorrected fast RTC, where the record gate + // rejects every future-stamped record and the offload can never bank + // anything. + // + // Rejecting the read for CORRELATION is still right (a junk value would + // arm alarms years out). Rejecting it for CORRECTION never was: SET_CLOCK + // writes real wall time, which is the correct outcome whether the read + // was junk or the RTC is genuinely ahead, and the retry budget is bounded + // at 3 either way. + if (ClockPolicy.shouldSetClock(dev, wall)) { + if (_deferForClock) { + // While the phone is still the suspect party, writing our wall clock + // onto a strap that may well be RIGHT corrupts a correct RTC and + // destroys the evidence — the read-back then "agrees" forever. Hold + // off until the phone corrects (gate clears) or the grace expires + // (the strap is the fast one, and the branch below fixes it). + _log( + 'Clock drift over policy but the PHONE clock is the suspect one ' + '(strap=$dev wall=$wall) — NOT writing SET_CLOCK yet.', + ); + } else if (_clockCorrectTries < 3) { + // BOUND the retries: setClock() reads the clock back and this handler + // re-issues on drift, so an unbounded loop would spin + // SET_CLOCK/GET_CLOCK forever on firmware that never latches. + // Historical records carry their own embedded unix time regardless, + // so giving up after a few tries is safe. + _clockCorrectTries++; + _log( + 'Clock drift over policy — re-issuing SET_CLOCK ' + '(attempt $_clockCorrectTries/3).', + ); + unawaited(setClock()); } else { - _clockCorrectTries = 0; // latched — reset for the next drift episode + _log( + 'Clock still off after 3 SET_CLOCK attempts — giving up; ' + 'firmware may not accept our payload length.', + ); } + } else { + _clockCorrectTries = 0; // latched — reset for the next drift episode } } if (f.containsKey('range_oldest') && f.containsKey('range_newest')) { @@ -2596,6 +2791,20 @@ class BleEngine { expectedPacketCount: expected, droppedThisBurst: droppedThisBurst, ); + // Honest, LOG-ONLY completeness signal (never gates the ACK). Compares + // num_packets against the ALL-TYPES received total (currentBurstTrafficCount), + // not the banked R24 subset — see burstPacketShortfall. Only a POSITIVE + // shortfall means frames the band counted that we did not count as valid + // received traffic (missing OR CRC-corrupted — potential loss); this is + // the signal we want visible in telemetry BEFORE ever wiring a FAIL gate + // (which needs its own design + field validation to avoid re-flood). + final shortfall = expected == null + ? 0 + : burstPacketShortfall( + expectedPacketCount: expected, + receivedTrafficCount: d.currentBurstTrafficCount, + droppedThisBurst: droppedThisBurst, + ); // ADVISORY ONLY, never a gate: `expectedPacketCount`'s exact semantics // (which transport packet types the band itself counts — command // responses interleaved with the burst? retried/duplicate frames?) are @@ -2635,11 +2844,27 @@ class BleEngine { 'traffic_burst_packets': d.currentBurstTrafficCount, 'burst_validation_failures': d.consecutiveValidationFailures, 'burst_breakdown': d.currentBurstBreakdown, + 'burst_shortfall': shortfall, }, )); } else { _burstMismatchStreak = 0; } + // Would-flag: the correct-signal completeness diagnostic. LOG-ONLY — the + // commit + verbatim-token ACK below are unchanged. A positive shortfall + // is the honest missing/corrupted-traffic telemetry we want to watch + // before a later, field-validated FAIL gate ever acts on it. + if (shortfall > 0) { + _log( + '[SYNC] burst completeness would-flag (LOG-ONLY, commit+ACK ' + 'unchanged): expected=$expected ' + 'received=${d.currentBurstTrafficCount} ' + 'dropped_this_burst=$droppedThisBurst shortfall=$shortfall ' + '(all-types received total — frames the band counted that we did ' + 'not; missing or CRC-corrupted, potential loss; groundwork for a ' + 'future FAIL gate, NOT gating today)', + ); + } final r = d.bufferedRecTsRange; final droppedThisBurstForLog = droppedThisBurst; final hadDurableRows = @@ -2939,10 +3164,15 @@ class BleEngine { inner.map((b) => b.toRadixString(16).padLeft(2, '0')).join(); // ── high-level flows ───────────────────────────────────────────────────────────── - Future sendInit() async { - _log('Sending 5-packet INIT…'); + /// [drain] false sends the first FOUR packets only: seq4 is + /// SEND_HISTORICAL_DATA (the flash drain), and it is skipped when the phone + /// clock is suspect — see _doConnect and [ClockPolicy.phoneClockSuspect]. + Future sendInit({bool drain = true}) async { + final pkts = + drain ? initPackets : initPackets.take(initPackets.length - 1).toList(); + _log('Sending ${pkts.length}-packet INIT…'); try { - for (final pkt in initPackets) { + for (final pkt in pkts) { await _write(pkt); await Future.delayed(const Duration(milliseconds: 120)); } @@ -3065,6 +3295,46 @@ class BleEngine { /// verify drift and re-correlate the strap-RTC ↔ wall clock. Future getClock() => _send(Cmd.getClock, const []); + /// GET_CLOCK, awaited to the *response* rather than to the write. + /// + /// Both clock gates (the connect-path SET_CLOCK decision and the history + /// drain in [_startHistoricalRefresh]) used to send GET_CLOCK, sleep a fixed + /// 120 ms, then read [_phoneClockSuspect]. That flag is cross-session state, + /// so a reply slower than the sleep — routine on a busy link mid-offload — + /// let the gate answer with the PREVIOUS connection's verdict, or with the + /// process default (`false`) on the very first connect. Both directions are + /// wrong: a stale `false` permits the drain-and-trim the gate exists to + /// prevent, and a stale `true` blocks a link whose clocks now agree. + /// + /// Returns whether a fresh reply landed. A timeout deliberately does NOT + /// change either gate's decision: an unanswered GET_CLOCK is not evidence + /// about the phone, and failing closed would mean a strap whose reply we + /// never see is a strap we never SET_CLOCK (it ships RTC-unset) and never + /// sync. Callers proceed on the last known verdict; the log line is the + /// signal that the read never landed. + Future _readClock() async { + final pending = _clockReadPending = Completer(); + await _send(Cmd.getClock, const []); + try { + await pending.future.timeout(_clockReadTimeout); + return true; + } on TimeoutException { + _log( + '[SYNC] GET_CLOCK went unanswered for ${_clockReadTimeout.inSeconds}s ' + '— clock verdict is UNVERIFIED for this read; proceeding on the last ' + 'known state (phone_clock_suspect=$_phoneClockSuspect).', + ); + return false; + } finally { + if (identical(_clockReadPending, pending)) _clockReadPending = null; + } + } + + /// How long [_readClock] waits for `clock_epoch`. A connected-link round trip + /// is tens of milliseconds; this is sized to survive a burst of historical + /// frames queued ahead of the response, not to be a plausible steady state. + static const Duration _clockReadTimeout = Duration(seconds: 3); + /// On-device wake alarm (SET_ALARM_TIME = 0x42) — the RICH 20-byte form that /// actually FIRES on WHOOP 4.0: /// ``` diff --git a/lib/compute/derivation_engine.dart b/lib/compute/derivation_engine.dart index 03bfc53..af384ae 100644 --- a/lib/compute/derivation_engine.dart +++ b/lib/compute/derivation_engine.dart @@ -697,7 +697,21 @@ import 'substrate.dart'; // floor, and now bills through the same per-sample gate, resting floor and // gap cap as the re-score. Already-stored sessions are left alone — they are // not re-derived — so the change applies from this version forward. -const int kAlgoVersion = 62; +// v63 - RR BEATS ARE FETCHED BY rec_ts, NOT BY COUNTER SPAN. +// The derivation pages 1 Hz frames ordered by rec_ts and then pulled that +// page's RR beats with `decodedRrByCounterRange(first.counter, last.counter)`. +// The strap's counter resets on every reboot, so the moment a page straddled +// one the span was inverted or nonsensical and the query returned nothing: +// the page decoded with an EMPTY beat list, and every beat-derived figure for +// that stretch — RMSSD, SDNN, the HRV curve, and the readiness that leans on +// them — silently came back absent or computed off whatever beats survived on +// the other pages. Both tables are keyed by rec_ts now, so the lookup uses the +// page's own rec_ts bounds and pulls exactly its beats. +// +// Days already finalized at v62 hold those RR-less results permanently — they +// are never revisited at the same version — so this needs the bump to be +// re-derived onto real beats. +const int kAlgoVersion = 63; // Fold idempotency, the minimum-nights warm-up, and legacy-payload handling // all live in SleepProfilePolicy (pure, unit-tested) — see @@ -1836,13 +1850,16 @@ class DerivationEngine { rangePages: rangePages, rangeRows: rangeRows, ); - final firstCounter = (decodedRows.first['counter'] as num?)?.toInt(); - final lastCounter = (decodedRows.last['counter'] as num?)?.toInt(); - final rrRows = firstCounter == null || lastCounter == null + // The page is ordered rec_ts ASC, so first = min second, last = max. + // decoded_rr shares the rec_ts key, so this pulls exactly the page's + // beats — no counter span (which broke across the strap's reboot reset). + final firstRecTs = (decodedRows.first['rec_ts'] as num?)?.toInt(); + final lastRecTs = (decodedRows.last['rec_ts'] as num?)?.toInt(); + final rrRows = firstRecTs == null || lastRecTs == null ? const >[] - : await LocalDb.decodedRrByCounterRange( - fromCounter: firstCounter, - toCounter: lastCounter, + : await LocalDb.decodedRrByRecTsRange( + fromRecTs: firstRecTs, + toRecTs: lastRecTs, ); worker.send({'type': 'page', 'frames': decodedRows, 'rr': rrRows}); final last = decodedRows.last; diff --git a/lib/compute/derive_prepare.dart b/lib/compute/derive_prepare.dart index cd02dc7..2e016d3 100644 --- a/lib/compute/derive_prepare.dart +++ b/lib/compute/derive_prepare.dart @@ -431,11 +431,14 @@ class _PrepareAccumulator { List> rrRows, ) { if (frames.isEmpty) return; - final rrByCounter = >>{}; + // Associate beats to frames by rec_ts (their shared key). The strap's counter + // resets on reboot, so grouping by counter mis-joined two seconds that reused + // one counter within a page. + final rrByRecTs = >>{}; for (final row in rrRows) { - final counter = _num(row['counter'])?.toInt(); - if (counter == null) continue; - rrByCounter.putIfAbsent(counter, () => >[]).add(row); + final recTs = _num(row['rec_ts'])?.toInt(); + if (recTs == null) continue; + rrByRecTs.putIfAbsent(recTs, () => >[]).add(row); } for (final row in frames) { final recTs = _num(row['rec_ts'])?.toInt(); @@ -454,9 +457,7 @@ class _PrepareAccumulator { // tsSec is what lets `Substrate.fromJson` tell "absent" (empty ⇒ // zero-filled) from "present but zero". skinContact.add(_num(row['skin_contact'])?.toInt() ?? 0); - final counter = _num(row['counter'])?.toInt(); - if (counter == null) continue; - final beats = rrByCounter[counter]; + final beats = rrByRecTs[recTs]; if (beats == null) continue; for (final beat in beats) { final rr = _num(beat['rr_ms'])?.toDouble(); diff --git a/lib/data/db.dart b/lib/data/db.dart index 2ca1326..0bcf23a 100644 --- a/lib/data/db.dart +++ b/lib/data/db.dart @@ -1,8 +1,8 @@ // Local raw-first storage (SQLite via sqflite). // // Durable storage layers: -// decoded_onehz — canonical per-second decoded substrate, deduped by rec_ts. -// decoded_rr — sparse RR beats for that substrate, deduped by (rr_ts_ms, beat_index). +// decoded_onehz — canonical per-second decoded substrate, keyed by rec_ts. +// decoded_rr — sparse RR beats for that substrate, keyed by (rec_ts, beat_index). // samples — legacy header cache kept only for backward-compat fallback. // // `counter` (u32 @[3:7]) is still kept as the strap's record id, but analytics @@ -95,7 +95,7 @@ class LocalDb { /// pass it: sqflite throws `ArgumentError('onCreate must be null if no /// version is specified')` BEFORE opening anything when `onCreate` is given /// without `version` (sqflite_common database_mixin.dart). - static const int schemaVersion = 31; + static const int schemaVersion = 33; /// SQLite caps host parameters per statement (`SQLITE_MAX_VARIABLE_NUMBER` — /// only 999 on the builds shipped with older Android/iOS). Any `IN (?, ?, …)` @@ -326,7 +326,11 @@ class LocalDb { await _createCycleSymptom(db); } if (oldV < 19) { - await _createDecodedStore(db); + // The v17 step (or a v11-16 origin) may leave OLD counter-keyed decoded + // tables here; the backfill below writes through the rec_ts-keyed + // _queueDecodedOneHz, so convert to the current schema first (preserving + // any existing rows), then reconstruct the rest from raw_records. + await _rekeyDecodedStoreByRecTs(db); await _backfillDecodedStore(db); await _dropRawStore(db); await _ensureSessionSchema(db); // adds hrr_bpm @@ -430,6 +434,47 @@ class LocalDb { // output is untouched and the edits replay over it. await _createSleepNap(db); } + if (oldV < 32) { + // Re-key raw_archive off the volatile `counter` onto frame `hex`. + // `counter INTEGER PRIMARY KEY` + IGNORE silently DROPPED a distinct + // undecodable frame whenever a post-reboot counter (reset to ~0) + // collided with a still-present pre-reboot row — data loss in the + // "never lose" table. Rebuild keyed by content. Existing rows have + // unique counters, so the copy loses nothing; at most it collapses an + // exact-duplicate hex, which is the dedup we want. + // + // raw_archive is normally created lazily in onOpen (_repairOpenSchema), + // NOT in this ladder, so on an old DB it may not exist yet here — in + // which case there is nothing to migrate and a fresh (hex-keyed) create + // is all that's needed. DROP the old index name before the fresh CREATE + // so it can't collide on the name the rename carried onto the aside + // table (the leaked-`_new`-index footgun documented on the decoded + // rebuild). + final hasArchive = (await db.rawQuery( + "SELECT 1 FROM sqlite_master WHERE type='table' AND name='raw_archive'", + )).isNotEmpty; + if (hasArchive) { + await db.execute('ALTER TABLE raw_archive RENAME TO _raw_archive_old'); + await db.execute('DROP INDEX IF EXISTS idx_raw_archive_captured'); + await _createRawArchive(db); + await db.execute( + 'INSERT OR IGNORE INTO raw_archive ' + '(hex, counter, packet_type, rec_ts, captured_at, reason) ' + 'SELECT hex, counter, packet_type, rec_ts, captured_at, reason ' + 'FROM _raw_archive_old', + ); + await db.execute('DROP TABLE _raw_archive_old'); + } else { + await _createRawArchive(db); + } + } + if (oldV < 33) { + // RE-KEY the decoded ledger off the volatile record `counter` onto + // rec_ts. The counter resets to ~0 on every reboot, so counter-as-PK + // let a post-reboot second REPLACE-evict a pre-reboot one — silently, + // unrecoverably deleting a 1 Hz row (raw_records is dropped). + await _rekeyDecodedStoreByRecTs(db); + } }, onOpen: (db) async { await _repairOpenSchema(db); @@ -1199,82 +1244,121 @@ class LocalDb { } final db = await instance; - await db.transaction((txn) async { - // Read the existing high-water THROUGH the txn — never via the global db - // handle, which would deadlock against this same open transaction. - var maxCounter = await _cursorIntVia(txn, 'counter_hw') ?? 0; - var maxRecTs = await _cursorIntVia(txn, 'rec_ts_hw') ?? 0; - // CHUNKED BATCH: sqflite serialises an ENTIRE batch's operations+args into - // ONE platform-channel message, and the native side builds a single - // ArrayList of every argument. A large backlog offload (raws in the - // hundreds-of-thousands) blew the native heap in SqlCommand.getSqlArguments - // → OutOfMemoryError (Crashlytics 0.9.13). Committing in bounded chunks - // flushes and frees each message's args. These commits all happen INSIDE - // the single `db.transaction` below, so the safe-trim invariant holds: the - // whole offload (raw_archive + samples + decoded_onehz + decoded_rr + - // cursor) is still one atomic transaction — every row is durable before the - // caller echoes the HISTORY_END trim token, or none is. - const chunkOps = 4000; - var batch = txn.batch(); - var ops = 0; - Future flushChunk() async { - if (ops == 0) return; - await batch.commit(noResult: true); - batch = txn.batch(); - ops = 0; - } + // POWER-LOSS DURABILITY WINDOW. This is the ACK-gating commit: once it + // returns, the caller writes the BLE batch-ACK and the band trims its flash. + // Under WAL + synchronous=NORMAL (the default this connection opens with) a + // commit is durable only at the next checkpoint — so a kernel panic / + // battery-yank AFTER the ACK but BEFORE the -wal is checkpointed loses these + // just-committed rows from the phone while they are already gone from the + // band. Raise durability to FULL (fsync AT commit) for THIS commit only, + // leaving every other path at NORMAL — they are all recomputable and FULL + // everywhere is brutally slow. `synchronous` is per-connection and CANNOT be + // changed mid-transaction, so it is set on the connection BEFORE + // db.transaction opens and reset AFTER it commits. The reset lives in a + // finally: a leaked FULL from a throwing commit would fsync every subsequent + // write on this connection forever. `PRAGMA synchronous=FULL/NORMAL` returns + // NO rows → execute() (not rawQuery), kept non-fatal like the open-time + // PRAGMAs so a PRAGMA throw can never fail a durable commit. Every ACK-gating + // commit — the foreground drain AND the headless iOS-restore recovery drain + // (background_sync.dart) — funnels through here, so this one choke point + // covers them all. `synchronous` is per-connection, so the bracket is only + // safe because these drains never OVERLAP on a shared connection: the offload + // processor is single-flight (ble_engine.dart) and BandOwnership makes the + // headless drain yield when the foreground owns the band. Do not add a second + // concurrent caller of commitSyncBatch on the main-isolate connection without + // reinstating that serialization — a mid-window reset would silently + // downgrade this commit back to NORMAL. + try { + await db.execute('PRAGMA synchronous=FULL'); + } catch (_) { + /* durability upgrade is best-effort — NORMAL still commits correctly */ + } + try { + await db.transaction((txn) async { + // Read the existing high-water THROUGH the txn — never via the global db + // handle, which would deadlock against this same open transaction. + var maxCounter = await _cursorIntVia(txn, 'counter_hw') ?? 0; + var maxRecTs = await _cursorIntVia(txn, 'rec_ts_hw') ?? 0; + // CHUNKED BATCH: sqflite serialises an ENTIRE batch's operations+args into + // ONE platform-channel message, and the native side builds a single + // ArrayList of every argument. A large backlog offload (raws in the + // hundreds-of-thousands) blew the native heap in SqlCommand.getSqlArguments + // → OutOfMemoryError (Crashlytics 0.9.13). Committing in bounded chunks + // flushes and frees each message's args. These commits all happen INSIDE + // the single `db.transaction` below, so the safe-trim invariant holds: the + // whole offload (raw_archive + samples + decoded_onehz + decoded_rr + + // cursor) is still one atomic transaction — every row is durable before the + // caller echoes the HISTORY_END trim token, or none is. + const chunkOps = 4000; + var batch = txn.batch(); + var ops = 0; + Future flushChunk() async { + if (ops == 0) return; + await batch.commit(noResult: true); + batch = txn.batch(); + ops = 0; + } - // SAFE-TRIM INVARIANT: archive the undecodable records in the SAME - // transaction as the raw records + trim cursor, so they are durably set - // aside BEFORE the caller writes the batch-ACK that lets the band trim. - if (archives != null) { - for (final a in archives) { - batch.insert('raw_archive', { - 'counter': a.counter, - 'hex': a.hex, - 'packet_type': a.packetType, - 'rec_ts': a.recTs, - 'captured_at': a.capturedAt, - 'reason': a.reason, - }, conflictAlgorithm: ConflictAlgorithm.ignore); - if (++ops >= chunkOps) await flushChunk(); + // SAFE-TRIM INVARIANT: archive the undecodable records in the SAME + // transaction as the raw records + trim cursor, so they are durably set + // aside BEFORE the caller writes the batch-ACK that lets the band trim. + if (archives != null) { + for (final a in archives) { + batch.insert('raw_archive', { + 'counter': a.counter, + 'hex': a.hex, + 'packet_type': a.packetType, + 'rec_ts': a.recTs, + 'captured_at': a.capturedAt, + 'reason': a.reason, + }, conflictAlgorithm: ConflictAlgorithm.ignore); + if (++ops >= chunkOps) await flushChunk(); + } } - } - for (var i = 0; i < raws.length; i++) { - final raw = raws[i]; - final recTs = _recTsFor(raw); - final sample = samples[i]; - if (sample != null) { - batch.insert('samples', { - 'counter': raw.counter, - ...sample.toDbMap(), - }, conflictAlgorithm: ConflictAlgorithm.ignore); - ops++; + for (var i = 0; i < raws.length; i++) { + final raw = raws[i]; + final recTs = _recTsFor(raw); + final sample = samples[i]; + if (sample != null) { + batch.insert('samples', { + 'counter': raw.counter, + ...sample.toDbMap(), + }, conflictAlgorithm: ConflictAlgorithm.ignore); + ops++; + } + ops += _queueDecodedOneHz(batch, raw, sample); + if (raw.counter > maxCounter) maxCounter = raw.counter; + if (recTs > maxRecTs) maxRecTs = recTs; + if (ops >= chunkOps) await flushChunk(); } - ops += _queueDecodedOneHz(batch, raw, sample); - if (raw.counter > maxCounter) maxCounter = raw.counter; - if (recTs > maxRecTs) maxRecTs = recTs; - if (ops >= chunkOps) await flushChunk(); - } - checkpoint( - 'decoded_archive_queued raws=${raws.length} ' - 'archives=${archives?.length ?? 0}', - ); - await flushChunk(); - checkpoint('decoded_archive_committed'); - await setCursor('counter_hw', '$maxCounter', txn: txn); - await setCursor('rec_ts_hw', '$maxRecTs', txn: txn); - if (trimToken != null) await setCursor('strap_trim', trimToken, txn: txn); - if (extraCursors != null) { - for (final e in extraCursors.entries) { - await setCursor(e.key, e.value, txn: txn); + checkpoint( + 'decoded_archive_queued raws=${raws.length} ' + 'archives=${archives?.length ?? 0}', + ); + await flushChunk(); + checkpoint('decoded_archive_committed'); + await setCursor('counter_hw', '$maxCounter', txn: txn); + await setCursor('rec_ts_hw', '$maxRecTs', txn: txn); + if (trimToken != null) await setCursor('strap_trim', trimToken, txn: txn); + if (extraCursors != null) { + for (final e in extraCursors.entries) { + await setCursor(e.key, e.value, txn: txn); + } } + checkpoint( + 'cursor_advanced counter_hw=$maxCounter rec_ts_hw=$maxRecTs ' + 'trim=${trimToken != null}', + ); + }); + } finally { + // ALWAYS restore NORMAL — even if the commit threw — so a leaked FULL does + // not fsync every subsequent write on this connection. Non-fatal. + try { + await db.execute('PRAGMA synchronous=NORMAL'); + } catch (_) { + /* non-fatal — see open-time PRAGMA discipline */ } - checkpoint( - 'cursor_advanced counter_hw=$maxCounter rec_ts_hw=$maxRecTs ' - 'trim=${trimToken != null}', - ); - }); + } await _writeCaptureFreshness(raws); } @@ -2072,10 +2156,19 @@ class LocalDb { // analytics: one row per real second (`rec_ts`) plus sparse RR beats for that // second. raw_records stays as the replay/debug ledger and upgrade fallback. static Future _createDecodedStore(Database db) async { + // KEYED BY rec_ts, NOT the band's record `counter`. The strap resets its + // per-record counter to ~0 on every reboot, so `counter INTEGER PRIMARY KEY` + // let a post-reboot record (counter=c, rec_ts=T2) REPLACE-evict a still-present + // pre-reboot row (counter=c, rec_ts=T1) — silently and UNRECOVERABLY deleting + // T1's only decoded 1 Hz row (raw_records is DROPped, so this store is the sole + // system of record). rec_ts is unique per real second, so newest-wins REPLACE + // on rec_ts is safe. `counter` is demoted to a NOT NULL forensic column (also + // the keyset-cursor tiebreak in decodedOneHzBatchByRecTsRange, which never + // fires now that rec_ts is unique). await db.execute(''' CREATE TABLE IF NOT EXISTS decoded_onehz ( - counter INTEGER PRIMARY KEY, - rec_ts INTEGER NOT NULL, + rec_ts INTEGER PRIMARY KEY, + counter INTEGER NOT NULL, hr INTEGER NOT NULL, ax REAL NOT NULL, ay REAL NOT NULL, @@ -2085,40 +2178,25 @@ class LocalDb { skin_temp_raw INTEGER NOT NULL ) '''); + // Forensic-only lookup by the raw counter; not on any read path. await db.execute( - 'CREATE INDEX IF NOT EXISTS idx_decoded_onehz_rects ON decoded_onehz(rec_ts, counter)', - ); - await db.execute( - 'CREATE UNIQUE INDEX IF NOT EXISTS idx_decoded_onehz_rec_ts_unique ' - 'ON decoded_onehz(rec_ts)', + 'CREATE INDEX IF NOT EXISTS idx_decoded_onehz_counter ON decoded_onehz(counter)', ); + // decoded_rr shares the rec_ts key with its parent: PRIMARY KEY (rec_ts, + // beat_index). Parent and child now delete/replace by the SAME key, so no + // orphan guard is needed. rr_ts_ms (= rec_ts*1000) stays as the per-beat + // timestamp the compute worker reads. No secondary index: the rec_ts-range + // read path is served by the PK, and the old UNIQUE(rr_ts_ms, beat_index) is + // now implied by the PK (rr_ts_ms is rec_ts*1000). await db.execute(''' CREATE TABLE IF NOT EXISTS decoded_rr ( - counter INTEGER NOT NULL, + rec_ts INTEGER NOT NULL, beat_index INTEGER NOT NULL, rr_ts_ms INTEGER NOT NULL, rr_ms INTEGER NOT NULL, - PRIMARY KEY (counter, beat_index) + PRIMARY KEY (rec_ts, beat_index) ) '''); - await db.execute( - 'CREATE UNIQUE INDEX IF NOT EXISTS idx_decoded_rr_ts_beat_unique ' - 'ON decoded_rr(rr_ts_ms, beat_index)', - ); - // idx_decoded_rr_ts(rr_ts_ms) was a strict prefix of the unique index - // above, so SQLite could already serve every rr_ts_ms lookup and ordering - // from it. The narrower index only added a second b-tree to maintain on - // the hottest write path in the app. - await db.execute('DROP INDEX IF EXISTS idx_decoded_rr_ts'); - // idx_decoded_rr_counter(counter, beat_index) was an EXACT duplicate of the - // index `PRIMARY KEY (counter, beat_index)` already creates - // (sqlite_autoindex_decoded_rr_1) — same table, same columns, same order. - // Measured on a 3-day fill: both b-trees 3,264,512 bytes, i.e. ~1.09 MB/day - // of pure duplication, plus a second b-tree write per beat on the hottest - // insert path in the app. After dropping it the planner still serves - // `counter` lookups and (counter, beat_index) ordering from the auto-index - // — pinned by test/db_storage_hygiene_test.dart, same as the drop above. - await db.execute('DROP INDEX IF EXISTS idx_decoded_rr_counter'); } /// Rebuild the decoded substrate into noop-style canonical time-keyed rows: @@ -2126,12 +2204,32 @@ class LocalDb { /// (second, beat_index). Older duplicate counters remain in raw_records for /// forensics, but analytics no longer sees them. static Future _rebuildCanonicalDecodedStore(Database db) async { - // Guarantee the source tables exist before we SELECT from them. On upgrade - // paths from before the decoded store landed, decoded_onehz/decoded_rr were - // never created in the migration chain, so this rebuild threw "no such table: - // decoded_onehz" — failing openDatabase on every launch (stuck at loading). - // Creating them (empty) here makes the dedup/rebuild a safe no-op in that case. - await _createDecodedStore(db); + // FROZEN v17 step: it dedups the OLD counter-keyed decoded tables by rec_ts + // via a `decoded_rr.counter` join. If the store is ALREADY rec_ts-keyed (the + // ladder created it fresh at v11 with the current schema, so decoded_rr has + // no `counter` column), it is already canonical — this rebuild is impossible + // and unnecessary, so skip it. A genuinely old (counter-keyed) store still + // gets the original rebuild here, and the v33 re-key converts it afterward. + final rrCols = await db.rawQuery('PRAGMA table_info(decoded_rr)'); + if (rrCols.isNotEmpty && !rrCols.any((c) => c['name'] == 'counter')) return; + // Guarantee the OLD-schema source tables exist before we SELECT from them. + // On upgrade paths from before the decoded store landed they were never + // created, so this rebuild threw "no such table" and bricked openDatabase. + await db.execute(''' + CREATE TABLE IF NOT EXISTS decoded_onehz ( + counter INTEGER PRIMARY KEY, rec_ts INTEGER NOT NULL, + hr INTEGER NOT NULL, ax REAL NOT NULL, ay REAL NOT NULL, az REAL NOT NULL, + spo2_red_raw INTEGER NOT NULL, spo2_ir_raw INTEGER NOT NULL, + skin_temp_raw INTEGER NOT NULL + ) + '''); + await db.execute(''' + CREATE TABLE IF NOT EXISTS decoded_rr ( + counter INTEGER NOT NULL, beat_index INTEGER NOT NULL, + rr_ts_ms INTEGER NOT NULL, rr_ms INTEGER NOT NULL, + PRIMARY KEY (counter, beat_index) + ) + '''); await db.execute('DROP TABLE IF EXISTS _decoded_onehz_new'); await db.execute('DROP TABLE IF EXISTS _decoded_rr_new'); // Drop any leftover temp-named indexes BEFORE recreating them. SQLite index @@ -2216,6 +2314,75 @@ class LocalDb { await db.execute('ALTER TABLE _decoded_rr_new RENAME TO decoded_rr'); } + /// v33: re-key the decoded store off the volatile record `counter` onto rec_ts. + /// + /// Rebuilds BOTH decoded tables FROM THE EXISTING decoded tables ONLY. It must + /// NOT backfill from raw_records (that table is DROPped — a raw-backfill would + /// zero the store, total loss). Existing rows have a unique rec_ts, so the copy + /// loses nothing; any pre-fix duplicate counters collapse to newest-wins per + /// rec_ts. Idempotent: a crash mid-migration re-runs cleanly (the temp tables + /// are dropped up front, and every write is INSERT OR REPLACE keyed on identity). + /// + /// All copies are INSERT ... SELECT (server-side, ZERO host-bound variables), + /// so the iOS SQLITE_MAX_VARIABLE_NUMBER (999) never applies — no chunking is + /// needed. Mirrors [_rebuildCanonicalDecodedStore]'s rename-aside shape. + static Future _rekeyDecodedStoreByRecTs(Database db) async { + // The source tables may not exist on a pre-decoded-store upgrade path; a + // create (new schema, IF NOT EXISTS) makes the copy a safe no-op there. On a + // normal path the OLD-schema tables already exist and this is a no-op — the + // columns we SELECT (rec_ts, counter, hr, …; beat_index, rr_ts_ms, rr_ms) + // are present in both the old and new decoded schemas. + await _createDecodedStore(db); + await db.execute('DROP TABLE IF EXISTS _decoded_onehz_v33'); + await db.execute('DROP TABLE IF EXISTS _decoded_rr_v33'); + await db.execute(''' + CREATE TABLE _decoded_onehz_v33 ( + rec_ts INTEGER PRIMARY KEY, + counter INTEGER NOT NULL, + hr INTEGER NOT NULL, + ax REAL NOT NULL, + ay REAL NOT NULL, + az REAL NOT NULL, + spo2_red_raw INTEGER NOT NULL, + spo2_ir_raw INTEGER NOT NULL, + skin_temp_raw INTEGER NOT NULL + ) + '''); + await db.execute(''' + CREATE TABLE _decoded_rr_v33 ( + rec_ts INTEGER NOT NULL, + beat_index INTEGER NOT NULL, + rr_ts_ms INTEGER NOT NULL, + rr_ms INTEGER NOT NULL, + PRIMARY KEY (rec_ts, beat_index) + ) + '''); + // Deterministic newest-wins: ORDER BY rec_ts, counter so INSERT OR REPLACE on + // the rec_ts PK keeps the highest-counter (latest-offloaded) row per second. + await db.execute( + 'INSERT OR REPLACE INTO _decoded_onehz_v33 ' + '(rec_ts, counter, hr, ax, ay, az, spo2_red_raw, spo2_ir_raw, skin_temp_raw) ' + 'SELECT rec_ts, counter, hr, ax, ay, az, spo2_red_raw, spo2_ir_raw, skin_temp_raw ' + 'FROM decoded_onehz ORDER BY rec_ts ASC, counter ASC', + ); + // rec_ts derived from rr_ts_ms (= rec_ts*1000 by construction). Pre-fix orphan + // beats (owning row evicted) re-home onto their real second here. + await db.execute( + 'INSERT OR REPLACE INTO _decoded_rr_v33 (rec_ts, beat_index, rr_ts_ms, rr_ms) ' + 'SELECT rr_ts_ms / 1000, beat_index, rr_ts_ms, rr_ms ' + 'FROM decoded_rr ORDER BY rr_ts_ms ASC, beat_index ASC', + ); + await db.execute('DROP TABLE IF EXISTS decoded_rr'); + await db.execute('DROP TABLE IF EXISTS decoded_onehz'); + await db.execute('ALTER TABLE _decoded_onehz_v33 RENAME TO decoded_onehz'); + await db.execute('ALTER TABLE _decoded_rr_v33 RENAME TO decoded_rr'); + // The rec_ts PK auto-indexes; add back the forensic counter index (the temp + // tables carried no named secondary indexes, so nothing leaked onto rename). + await db.execute( + 'CREATE INDEX IF NOT EXISTS idx_decoded_onehz_counter ON decoded_onehz(counter)', + ); + } + // raw_records — keyed by the band's per-record u32 `counter` (the natural // idempotency key; re-draining the same flash region inserts nothing new). Only // the 1 Hz historical substrate (0x2F / R24) is persisted here — LIVE high-rate @@ -2363,71 +2530,37 @@ class LocalDb { } } - /// THE orphan guard for an INSERT-OR-REPLACE into `decoded_onehz`. - /// - /// Queue this onto [batch] IMMEDIATELY BEFORE writing the row for [counter] @ - /// [recTs] — every write path into `decoded_onehz` must go through it, or it - /// strands `decoded_rr` beats (see [_queueDecodedOneHz] for the full - /// derivation of both eviction cases). Returns the number of ops queued. - static int _queueOrphanGuard( - Batch batch, { - required int counter, - required int recTs, - }) { - batch.rawDelete( - 'DELETE FROM decoded_rr WHERE ' - // (a) UNIQUE(rec_ts) eviction — the LOSER counter's beats. - 'counter IN ' - '(SELECT counter FROM decoded_onehz WHERE rec_ts = ? AND counter != ?) ' - // (b) counter-PK eviction — stale-timestamped beats under OUR counter. - 'OR (counter = ? AND rr_ts_ms != ?)', - [recTs, counter, counter, recTs * 1000], - ); - return 1; - } - - /// Queues the decoded_onehz + decoded_rr (+ orphan-guard delete) writes for - /// one raw onto [batch]. Returns the number of batch operations added, so a - /// caller committing a large offload can chunk the batch to bound the native - /// argument-list size (see [commitSyncBatch]). + /// Queues the decoded_onehz + decoded_rr writes for one raw onto [batch]. + /// Returns the number of batch operations added, so a caller committing a + /// large offload can chunk the batch to bound the native argument-list size + /// (see [commitSyncBatch]). static int _queueDecodedOneHz(Batch batch, RawRecord raw, Sample? sample) { final decoded = _decodeOneHzSample(raw, preferred: sample); if (decoded == null) return 0; - final recTs = raw.recTs ?? decoded.tsEpoch; + // `??` substitutes on NULL only, and `rec_ts` is the primary key now. The + // legacy `raw_records.rec_ts` column is `NOT NULL DEFAULT 0`, so every + // undated row [_backfillDecodedStore] replays arrives here as an explicit + // 0 — which under the old counter PK coexisted harmlessly and under this + // one REPLACE-evicts all the others down to a single row. Same `> 0` + // fallback [_recTsFor] uses (inlined: `decoded` already carries the + // timestamp, so going through it would re-decode the hex for nothing). + final rawRecTs = raw.recTs; + final recTs = + (rawRecTs != null && rawRecTs > 0) ? rawRecTs : decoded.tsEpoch; // TIME-KEYED, NEWEST-WINS (noop/WHOOP-4 model: dedupe records by their - // embedded timestamp, not by a counter). decoded_onehz has a UNIQUE(rec_ts) - // index and decoded_rr a UNIQUE(rr_ts_ms, beat_index). We use REPLACE, not - // IGNORE: the strap's record `counter` RESETS to ~0 on every reboot, so a - // post-reboot record whose second already had a row would be SILENTLY DROPPED - // under IGNORE — quarantining everything after a reboot (observed: whole days - // present in raw_records but absent from the decoded substrate the engine - // reads → "not worn / metrics still computing / strain –"). REPLACE lets the - // freshly-offloaded record for a given second win, which is what we want. + // embedded timestamp, not by the volatile counter). decoded_onehz is keyed + // by rec_ts and decoded_rr by (rec_ts, beat_index). We use REPLACE, not + // IGNORE: a freshly-offloaded record for a given second should win over a + // stale one. Because rec_ts is the key, the strap's per-reboot counter reset + // can no longer make one second's record evict another's (the pre-fix + // counter-PK eviction that silently, unrecoverably deleted 1 Hz rows). // - // ORPHAN GUARD: decoded_rr rows are keyed by their record's own counter. When - // the REPLACE below evicts a DIFFERENT counter's row for this second, that - // loser's RR beats would stay behind under a counter with no decoded_onehz - // row — invisible to the counter-joined prune (permanent leak). The winner's - // REPLACE on UNIQUE(rr_ts_ms, beat_index) only overwrites overlapping beat - // indexes, so delete the evicted counter's beats explicitly, in the same - // batch/transaction (mirrors the v17 rebuild's decoded_onehz join). - // - // …AND the COUNTER-PK eviction, which the guard used to miss entirely. - // `decoded_onehz` is `counter INTEGER PRIMARY KEY` as well as - // UNIQUE(rec_ts), and (per the comment above) the strap's counter RESETS to - // ~0 on every reboot — so this same REPLACE also silently DELETES the row - // of an OLDER SECOND that happened to reuse this counter. That older - // second's beats live under OUR counter carrying ITS rr_ts_ms, and only the - // overlapping beat_indexes get overwritten below: any beat at an index past - // the new record's beat count SURVIVES, still stamped days earlier. Neither - // prune path can ever see it (the counter-join finds a fresh rec_ts; the - // orphan sweep finds the counter present), so a later page's RR series was - // polluted with beats from another day — silently wrecking RMSSD/HRV. - // Drop every beat under this counter that is not stamped with THIS second. - var ops = _queueOrphanGuard(batch, counter: raw.counter, recTs: recTs); + // Clear this second's RR beats before reinserting so a SHRINKING beat count + // can't strand stale high-index beats — the parent+child share the rec_ts + // key, so this single DELETE replaces the old counter-based orphan guard. batch.insert('decoded_onehz', { - 'counter': raw.counter, 'rec_ts': recTs, + 'counter': raw.counter, 'hr': decoded.hr, 'ax': decoded.ax ?? 0, 'ay': decoded.ay ?? 0, @@ -2436,12 +2569,14 @@ class LocalDb { 'spo2_ir_raw': decoded.spo2IrRaw ?? 0, 'skin_temp_raw': decoded.skinTempRaw ?? 0, }, conflictAlgorithm: ConflictAlgorithm.replace); - ops++; // the decoded_onehz insert + var ops = 1; // the decoded_onehz insert + batch.rawDelete('DELETE FROM decoded_rr WHERE rec_ts = ?', [recTs]); + ops++; for (var i = 0; i < decoded.rrIntervalsMs.length; i++) { final rr = decoded.rrIntervalsMs[i]; if (rr <= 0) continue; batch.insert('decoded_rr', { - 'counter': raw.counter, + 'rec_ts': recTs, 'beat_index': i, 'rr_ts_ms': recTs * 1000, 'rr_ms': rr, @@ -2543,12 +2678,19 @@ class LocalDb { /// Durable archive for historical records we received but could not decode /// (unknown/unsupported version). NEVER pruned — the whole point is that a /// future firmware's records survive until we understand the format. Keyed by - /// counter so a re-flood after a missed ACK dedups (IGNORE on conflict). + /// frame `hex` (content identity), like `events`/`band_events` — NOT by + /// `counter`. The strap resets its record counter to ~0 on every reboot, so + /// two DISTINCT undecodable frames from different boots can collide on a + /// reused counter; a `counter`-PK + IGNORE silently DROPPED the second, in + /// the one table whose whole purpose is to never lose a frame. Hashing on the + /// bytes means an identical re-flood (missed-ACK redelivery) still dedups, + /// while genuinely distinct frames both survive a counter collision. `counter` + /// is retained as a plain forensic column. static Future _createRawArchive(Database db) async { await db.execute(''' CREATE TABLE IF NOT EXISTS raw_archive ( - counter INTEGER PRIMARY KEY, - hex TEXT NOT NULL, + hex TEXT PRIMARY KEY, + counter INTEGER, packet_type INTEGER NOT NULL, rec_ts INTEGER, captured_at INTEGER NOT NULL, @@ -3061,92 +3203,28 @@ class LocalDb { ); } - /// How many times [decodedRrByCounterRange]'s degraded counter-span fallback - /// hit its row cap and therefore returned an INCOMPLETE set of beats. Any - /// value above zero means some window's HRV was computed from truncated - /// input; it should stay at zero in normal operation. - static int decodedRrFallbackTruncations = 0; - - /// Sparse RR beats for one contiguous decoded 1 Hz page. - /// - /// [fromCounter] / [toCounter] are the page's FIRST and LAST row counters, as - /// returned by [decodedOneHzBatchByRecTsRange] (which orders `rec_ts ASC, - /// counter ASC`). They are page ENDPOINTS, **not** a monotonic counter span: - /// the strap's counter resets to ~0 on every reboot, so a page straddling a - /// reboot has first = a pre-reboot high and last = a post-reboot low. The old - /// `WHERE counter >= ? AND counter <= ?` then read `>= 1200000 AND <= 5` and - /// returned ZERO rows — the entire page's RR beats vanished with no error, so - /// that window silently produced no RMSSD/HRV at all. + /// Sparse RR beats for one contiguous decoded 1 Hz page, by its rec_ts window. /// - /// Selection is therefore by the page's real TIME window, resolved from those - /// two endpoint counters. `decoded_onehz` is UNIQUE(rec_ts), so - /// `[first.rec_ts, last.rec_ts]` contains exactly the page's rows — no - /// over-fetch — and the join to `decoded_onehz` additionally keeps orphaned - /// beats (whose owning row was evicted) out of the read path. - /// - /// When the endpoints are NOT real rows the caller is asking for a plain - /// counter span (e.g. `0 .. 1<<30` = "everything"); that falls back to a - /// NORMALIZED counter range so an inverted pair still can't return nothing. - static Future>> decodedRrByCounterRange({ - required int fromCounter, - required int toCounter, + /// [fromRecTs] / [toRecTs] are the page's first and last record seconds (the + /// page is ordered `rec_ts ASC`, so first = min, last = max). decoded_rr shares + /// the rec_ts key with decoded_onehz, so `[fromRecTs, toRecTs]` on the PK + /// contains exactly the page's beats — bounded, indexed, and immune to the + /// strap's reboot counter reset (the old counter-span read could degenerate to + /// `counter >= high AND counter <= low` = zero rows, silently dropping a whole + /// page's RR). + static Future>> decodedRrByRecTsRange({ + required int fromRecTs, + required int toRecTs, }) async { final db = await instance; - final bounds = (await db.rawQuery( - 'SELECT COUNT(*) AS n, MIN(rec_ts) AS lo, MAX(rec_ts) AS hi ' - 'FROM decoded_onehz WHERE counter IN (?, ?)', - [fromCounter, toCounter], - )).first; - final n = (bounds['n'] as num?)?.toInt() ?? 0; - final want = fromCounter == toCounter ? 1 : 2; - if (n == want) { - return db.rawQuery( - 'SELECT rr.counter AS counter, rr.beat_index AS beat_index, ' - ' rr.rr_ts_ms AS rr_ts_ms, rr.rr_ms AS rr_ms ' - 'FROM decoded_rr rr ' - 'JOIN decoded_onehz d ON d.counter = rr.counter ' - 'WHERE d.rec_ts >= ? AND d.rec_ts <= ? ' - 'ORDER BY d.rec_ts ASC, rr.beat_index ASC', - [bounds['lo'], bounds['hi']], - ); - } - final lo = fromCounter <= toCounter ? fromCounter : toCounter; - final hi = fromCounter <= toCounter ? toCounter : fromCounter; - // BOUNDED. This branch is reached when an endpoint row is not in - // `decoded_onehz` — a prune, or an import's REPLACE + orphan-guard DELETE - // landing between the frame-page read and this call. The caller's counters - // are then just a span, and because the strap's counter resets on reboot a - // reboot-straddling page degenerates to `0 .. ~1200000`, i.e. effectively - // the whole table. Unbounded, that is a hundreds-of-MB platform-heap read - // on the same Java heap that OOMed the import path. A page is 2000 frames - // and a second rarely carries more than a handful of beats, so this cap is - // orders of magnitude above any legitimate page — reaching it means the - // degraded path is being used for a range it was never meant to serve. - const fallbackBeatCap = 200000; - final rows = await db.query( - 'decoded_rr', - columns: ['counter', 'beat_index', 'rr_ts_ms', 'rr_ms'], - where: 'counter >= ? AND counter <= ?', - whereArgs: [lo, hi], - orderBy: 'counter ASC, beat_index ASC', - limit: fallbackBeatCap, + final lo = fromRecTs <= toRecTs ? fromRecTs : toRecTs; + final hi = fromRecTs <= toRecTs ? toRecTs : fromRecTs; + return db.rawQuery( + 'SELECT rec_ts, beat_index, rr_ts_ms, rr_ms FROM decoded_rr ' + 'WHERE rec_ts >= ? AND rec_ts <= ? ' + 'ORDER BY rec_ts ASC, beat_index ASC', + [lo, hi], ); - // Never truncate silently — a short read here means missing beats, which - // shows up downstream as understated HRV rather than as an error. db.dart - // deliberately takes no telemetry dependency, so the fact is recorded as a - // plain counter the Diagnostics screen can surface. - // - // A static field is sound HERE specifically: every sqflite call needs the - // root isolate's platform channel, and this method's only caller - // (`DerivationEngine._prepare`) reads on the main isolate and ships each - // page to the compute worker with `worker.send`. Increments therefore land - // in the same isolate that reads them. Move this read into an isolate and - // the counter silently stops working — pass the count back over the port - // instead of reaching for a static. - if (rows.length >= fallbackBeatCap) { - decodedRrFallbackTruncations++; - } - return rows; } // ── VERSIONED DERIVED STORE I/O (day_result; main isolate only) ───────────── @@ -3589,20 +3667,20 @@ class LocalDb { where: 'rec_ts >= ? AND rec_ts < ?', whereArgs: [startSec, endSec], onPage: (page) async { - final counters = [ + final recTsList = [ for (final row in page) - if (row['counter'] != null) row['counter'], + if (row['rec_ts'] != null) row['rec_ts'], ]; - if (counters.isEmpty) return; - // CHUNKED `IN (…)`: even one page's counters can approach + if (recTsList.isEmpty) return; + // CHUNKED `IN (…)`: even one page's seconds can approach // SQLITE_MAX_VARIABLE_NUMBER, and a full day is 86,400 — two orders // of magnitude past it, so one giant statement could never bind. - // (This never surfaced only because the missing `version:` above - // aborted the export earlier.) - for (final chunk in _sqlVarChunks(counters)) { + // Keyed on rec_ts (decoded_rr's key), which pulls exactly this page's + // beats — a counter `IN` could over-match a reboot-reused counter. + for (final chunk in _sqlVarChunks(recTsList)) { final placeholders = List.filled(chunk.length, '?').join(','); final rr = await src.rawQuery( - 'SELECT * FROM decoded_rr WHERE counter IN ($placeholders)', + 'SELECT * FROM decoded_rr WHERE rec_ts IN ($placeholders)', chunk, ); if (rr.isEmpty) continue; @@ -3712,8 +3790,7 @@ class LocalDb { final (startSec, endSec) = _localDayWindow(dayId); deleted += await txn.delete( 'decoded_rr', - where: - 'counter IN (SELECT counter FROM decoded_onehz WHERE rec_ts >= ? AND rec_ts < ?)', + where: 'rec_ts >= ? AND rec_ts < ?', whereArgs: [startSec, endSec], ); deleted += await txn.delete( @@ -3958,6 +4035,7 @@ class LocalDb { ops = 0; } + final rows = >[]; for (final r in page) { final row = { for (final e in r.entries) @@ -3970,21 +4048,58 @@ class LocalDb { )) { continue; // locally finalized — never overwritten by an import } - // ORPHAN GUARD ON THE IMPORT PATH. A plain replace-insert into - // decoded_onehz bypasses _queueDecodedOneHz entirely, so a - // foreign row colliding on UNIQUE(rec_ts) (different counter) or - // on the `counter` PRIMARY KEY (different second) evicted a local - // row and stranded its decoded_rr beats — the exact leak the - // ingest path is guarded against, wide open here. Queue the SAME - // guard, in the same batch/transaction, right before the row. - if (t == 'decoded_onehz') { - final counter = (row['counter'] as num?)?.toInt(); - final recTs = (row['rec_ts'] as num?)?.toInt(); - if (counter == null || recTs == null) continue; - ops += _queueOrphanGuard(batch, counter: counter, recTs: recTs); + // A LEGACY export's decoded_rr carries no rec_ts column; derive + // it from rr_ts_ms (= rec_ts*1000) so the NOT NULL PK column is + // always populated. SQLite storage classes are per VALUE, not per + // column, so a foreign export can hand back a String where + // INTEGER is declared — a bare `as num` there throws inside the + // transaction and takes the whole restore down with it. Leave a + // non-numeric value alone and let the row fail its own NOT NULL + // check instead of aborting every other row's import. + if (t == 'decoded_rr' && row['rec_ts'] == null) { + final rrTsMs = row['rr_ts_ms']; + if (rrTsMs is num) row['rec_ts'] = rrTsMs.toInt() ~/ 1000; } + rows.add(row); + } + // REPLACE the beat set for a colliding second, don't patch it. + // decoded_rr is keyed by (rec_ts, beat_index), so a row-by-row + // replace-insert only overwrites the indices the foreign export + // actually reaches: importing [500] over a local [700, 710, 720] + // leaves beats 1 and 2 behind and hands that second a spliced + // foreign/local RR series — silently wrong RMSSD, out of a restore. + // [_queueDecodedOneHz] guards the identical hazard on the write + // path with a DELETE ahead of its inserts. + // + // Here the delete has to TRAIL the inserts and be bounded by the + // highest index this page carried, because a second's beats can + // straddle a page boundary: a leading `DELETE WHERE rec_ts = ?` on + // page 2 would wipe the beats page 1 just imported. Trailing + + // bounded is idempotent across the split — page 1 inserts 0,1 and + // clears >1; page 2 inserts 2,3 and clears >3 — and beat_index is + // dense by construction, so "everything past the last one" is + // exactly the stale local tail. + final highestBeat = {}; + for (final row in rows) { batch.insert(t, row, conflictAlgorithm: ConflictAlgorithm.replace); copied++; + if (t == 'decoded_rr') { + final recTs = row['rec_ts']; + final idx = row['beat_index']; + if (recTs != null && idx is num) { + final n = idx.toInt(); + final prev = highestBeat[recTs]; + if (prev == null || n > prev) highestBeat[recTs] = n; + } + } + if (++ops >= chunkOps) await flush(); + } + for (final e in highestBeat.entries) { + batch.delete( + 'decoded_rr', + where: 'rec_ts = ? AND beat_index > ?', + whereArgs: [e.key, e.value], + ); if (++ops >= chunkOps) await flush(); } await flush(); @@ -5630,10 +5745,12 @@ class LocalDb { // caller's `if (deleted > 0) log(...)` never fired even on a real prune. int deleted = 0; await db.transaction((txn) async { + // decoded_rr shares the rec_ts key, so a plain rec_ts range delete covers + // every beat in the window — no counter subquery, no orphan sweep (there + // are no counter-orphans once parent and child are keyed the same way). deleted += await txn.delete( 'decoded_rr', - where: - 'counter IN (SELECT counter FROM decoded_onehz WHERE rec_ts < ?)', + where: 'rec_ts < ?', whereArgs: [cutoffSec], ); deleted += await txn.delete( @@ -5641,18 +5758,6 @@ class LocalDb { where: 'rec_ts < ?', whereArgs: [cutoffSec], ); - // ORPHAN SWEEP: pre-guard builds could leave decoded_rr beats whose - // owning counter lost a rec_ts collision (REPLACE evicted its - // decoded_onehz row) — the counter-joined delete above never selects - // those. Their rr_ts_ms is the colliding second, so once the window is - // pruned they're strictly before the cutoff; delete any beat in the - // pruned window whose counter no longer exists in decoded_onehz. - deleted += await txn.delete( - 'decoded_rr', - where: - 'rr_ts_ms < ? AND counter NOT IN (SELECT counter FROM decoded_onehz)', - whereArgs: [cutoffSec * 1000], - ); deleted += await txn.delete('samples', where: 'ts < ?', whereArgs: [cutoffSec]); deleted += diff --git a/lib/sync/sync_policy.dart b/lib/sync/sync_policy.dart index 858aa4a..fbcd4be 100644 --- a/lib/sync/sync_policy.dart +++ b/lib/sync/sync_policy.dart @@ -137,6 +137,46 @@ class ClockPolicy { return drift > 86400 || deviceClock < kMinPlausibleUnix; } + /// True when the strap RTC reads a PLAUSIBLE absolute time but sits more than + /// [kFutureMargin] in the FUTURE relative to the phone — the signature of a + /// PHONE clock running slow (dead-battery reboot, bad NTP, a manual set-back). + /// + /// This is the one clock-disagreement we must NOT act on destructively. We + /// cannot prove which clock is right, but both wrong moves are unsafe: + /// - draining now drops the strap's (correctly-stamped, real-now) records as + /// "implausibly future", and a mixed-burst ACK then TRIMS them off the + /// band — permanent, silent loss; and + /// - SET_CLOCK-ing the strap backward to match the slow phone would corrupt + /// a correct RTC. + /// So the caller DEFERS history offload until the clocks agree (the phone + /// clock almost always self-corrects via NTP within minutes; the strap keeps + /// every record until then). The `>= kMinPlausibleUnix` guard excludes an + /// unset/garbage-low RTC (that is a strap problem [shouldSetClock] fixes, not + /// a phone problem); the strap-BEHIND case is a plausible-past time that is + /// not dropped as future and is corrected forward by [shouldSetClock]. + /// How long a suspect-clock disagreement may defer history before we stop + /// believing the phone is the wrong one. A phone that rebooted with a dead + /// battery re-syncs over NTP within minutes, so a disagreement that survives + /// this long is a strap RTC that is genuinely running fast — not a slow + /// phone. Past this the gate stops deferring and the strap clock is corrected + /// normally, so a bad strap RTC cannot stall history forever. + static const int suspectGraceSeconds = 12 * 3600; + + /// True once a suspect-clock state has persisted past [suspectGraceSeconds]. + /// + /// Both arguments are MONOTONIC seconds (a `Stopwatch`), never wall clock. + /// The state being timed is "we do not trust `DateTime.now()`", so timing it + /// with `DateTime.now()` is self-defeating: a phone that steps forward a day + /// over NTP — while possibly still more than a day behind the strap — would + /// instantly age the suspicion past the grace window and re-authorize the + /// drain-and-trim this gate exists to hold back. + static bool suspectGraceExpired(double? sinceSecs, double nowSecs) => + sinceSecs != null && nowSecs - sinceSecs >= suspectGraceSeconds; + + static bool phoneClockSuspect(int deviceClock, int wallNow) => + deviceClock >= kMinPlausibleUnix && + deviceClock > wallNow + kFutureMargin; + /// Salvage an implausible record time using the strap↔wall clock offset /// (device→wall = [clockWall] - [deviceClock]). A wandering/unset RTC offsets /// EVERY record in a session by the same amount, so shifting by that offset diff --git a/pubspec.lock b/pubspec.lock index 3301a54..566a91a 100644 --- a/pubspec.lock +++ b/pubspec.lock @@ -892,10 +892,10 @@ packages: dependency: transitive description: name: meta - sha256: "23f08335362185a5ea2ad3a4e597f1375e78bce8a040df5c600c8d3552ef2394" + sha256: "1741988757a65eb6b36abe716829688cf01910bbf91c34354ff7ec1c3de2b349" url: "https://pub.dev" source: hosted - version: "1.17.0" + version: "1.18.0" mgrs_dart: dependency: transitive description: @@ -1328,7 +1328,7 @@ packages: source: hosted version: "2.4.2+3" sqflite_common: - dependency: transitive + dependency: "direct dev" description: name: sqflite_common sha256: "1581ffbf7a0e333b380d6a30737d78516b826cb35beb7fb0bf8a3ea0c678b465" @@ -1411,26 +1411,26 @@ packages: dependency: "direct dev" description: name: test - sha256: "280d6d890011ca966ad08df7e8a4ddfab0fb3aa49f96ed6de56e3521347a9ae7" + sha256: "8d9ceddbab833f180fbefed08afa76d7c03513dfdba87ffcec2718b02bbcbf20" url: "https://pub.dev" source: hosted - version: "1.30.0" + version: "1.31.0" test_api: dependency: transitive description: name: test_api - sha256: "8161c84903fd860b26bfdefb7963b3f0b68fee7adea0f59ef805ecca346f0c7a" + sha256: "949a932224383300f01be9221c39180316445ecb8e7547f70a41a35bf421fb9e" url: "https://pub.dev" source: hosted - version: "0.7.10" + version: "0.7.11" test_core: dependency: transitive description: name: test_core - sha256: "0381bd1585d1a924763c308100f2138205252fb90c9d4eeaf28489ee65ccde51" + sha256: "1991d4cfe85d5043241acac92962c3977c8d2f2add1ee73130c7b286417d1d34" url: "https://pub.dev" source: hosted - version: "0.6.16" + version: "0.6.17" timezone: dependency: "direct main" description: diff --git a/pubspec.yaml b/pubspec.yaml index f334e21..386f996 100644 --- a/pubspec.yaml +++ b/pubspec.yaml @@ -260,6 +260,13 @@ dev_dependencies: # exportDaysDb can run without a platform plugin (transitive via # path_provider; declared directly since test/ now imports it). path_provider_platform_interface: ^2.1.0 + # ack_commit_sync_full_test wraps the ffi factory in SqfliteDatabaseFactoryLogger + # to spy the PRAGMA synchronous=FULL/NORMAL bracket around the ACK-gating commit + # (transitive via sqflite_common_ffi; declared directly since test/ now imports it). + # PINNED, not a caret range: that constructor is @experimental upstream, so a + # minor bump is allowed to change or withdraw it and would break the test on + # someone else's `pub upgrade` rather than on a deliberate one here. + sqflite_common: 2.5.8 flutter_launcher_icons: android: "launcher_icon" diff --git a/test/ack_commit_sync_full_test.dart b/test/ack_commit_sync_full_test.dart new file mode 100644 index 0000000..64c2197 --- /dev/null +++ b/test/ack_commit_sync_full_test.dart @@ -0,0 +1,98 @@ +// POWER-LOSS DURABILITY of the ACK-gating commit. `commitSyncBatch` is the one +// commit the safe-trim invariant hangs on: it must be durable (fsynced) BEFORE +// the caller writes the BLE batch-ACK that lets the band trim its flash. The DB +// otherwise runs WAL + synchronous=NORMAL (durable only at a checkpoint), so +// this path raises synchronous=FULL for its single transaction and restores +// NORMAL afterward — leaving every other (recomputable) path fast. This test +// pins that bracket: FULL is set around the commit, NORMAL is restored after, +// AND the restore still happens when the commit THROWS (a leaked FULL would +// fsync every subsequent write on the connection forever). +// +// We spy the real SQL stream via SqfliteDatabaseFactoryLogger (synchronous is +// per-connection and invisible from a second connection, so the log is the only +// honest observation point) and also read `PRAGMA synchronous` on LocalDb's own +// connection — the same one commitSyncBatch uses — to confirm the resting value. + +import 'package:flutter_test/flutter_test.dart'; +import 'package:path/path.dart' as p; +import 'package:sqflite_common/sqflite_logger.dart'; +import 'package:sqflite_common_ffi/sqflite_ffi.dart'; +import 'package:openstrap_edge/data/db.dart'; +import 'package:openstrap_edge/data/models.dart'; + +void main() { + // Every `synchronous=…` statement executed on any connection, in order. + final syncStmts = []; + + setUpAll(() async { + sqfliteFfiInit(); + // ignore: experimental_member_use — stable enough to spy SQL in a test. + databaseFactory = SqfliteDatabaseFactoryLogger( + databaseFactoryFfi, + options: SqfliteLoggerOptions( + log: (event) { + if (event is SqfliteLoggerSqlEvent) { + final sql = event.sql.toLowerCase(); + if (sql.contains('pragma synchronous=')) syncStmts.add(sql); + } + }, + ), + ); + LocalDb.dbName = 'openstrap_ack_sync_full_test.db'; + final dir = await databaseFactory.getDatabasesPath(); + await databaseFactory.deleteDatabase(p.join(dir, LocalDb.dbName)); + // Force the open now so its onConfigure PRAGMAs aren't counted in per-test + // windows — each test clears syncStmts against an already-open connection. + await LocalDb.instance; + }); + + tearDownAll(() async { + await LocalDb.close(); + final dir = await databaseFactory.getDatabasesPath(); + await databaseFactory.deleteDatabase(p.join(dir, LocalDb.dbName)); + }); + + Future restingSynchronous() async { + final db = await LocalDb.instance; + final rows = await db.rawQuery('PRAGMA synchronous'); + return rows.first.values.first as int; // FULL=2, NORMAL=1 + } + + RawRecord recAt(int counter) => RawRecord( + counter: counter, + packetType: 0x2F, + hex: '2f18aabbccdd', + capturedAt: 1750000000000 + counter, + recTs: 1750000000 + counter, + ); + + test('commitSyncBatch brackets synchronous=FULL and restores NORMAL', () async { + expect(await restingSynchronous(), 1, reason: 'connection opens at NORMAL'); + + syncStmts.clear(); + await LocalDb.commitSyncBatch( + [recAt(5001)], + [Sample(tsEpoch: 1750005001, counter: 5001, hr: 60)], + trimToken: 'deadbeef', + ); + + expect(syncStmts, ['pragma synchronous=full', 'pragma synchronous=normal'], + reason: 'FULL is set before the commit and NORMAL restored right after'); + expect(await restingSynchronous(), 1, reason: 'connection left at NORMAL'); + }); + + test('synchronous is restored to NORMAL even when the commit throws', () async { + syncStmts.clear(); + // raws non-empty but samples empty → samples[i] throws RangeError INSIDE the + // db.transaction, after FULL is set. The finally must still restore NORMAL. + await expectLater( + LocalDb.commitSyncBatch([recAt(6001)], const []), + throwsA(isA()), + ); + + expect(syncStmts, ['pragma synchronous=full', 'pragma synchronous=normal'], + reason: 'a thrown commit must not leak FULL'); + expect(await restingSynchronous(), 1, + reason: 'FULL did not leak past the throwing commit'); + }); +} diff --git a/test/ble_clock_gate_test.dart b/test/ble_clock_gate_test.dart new file mode 100644 index 0000000..ef30d75 --- /dev/null +++ b/test/ble_clock_gate_test.dart @@ -0,0 +1,120 @@ +// Regression tests for the phone-clock trust gate — the thing standing between +// a phone whose wall clock is a day slow and permanent, silent data loss. +// +// The failure it guards: when the phone reads slow, the strap's correctly +// stamped records look "implausibly future", the record gate drops them, and +// the HISTORY_END ACK then trims them off the band's flash for good. So while +// the phone is the suspect party we neither drain history nor push our wall +// clock onto the strap. +// +// Both halves of that refusal were reachable around, which is what these cover: +// the SET_CLOCK half was handed straight back by the drift-correction retry +// sitting behind the gate's own read-back. + +import 'package:flutter_test/flutter_test.dart'; +import 'package:openstrap_edge/ble/ble_engine.dart'; +import 'package:openstrap_protocol/openstrap_protocol.dart'; + +void main() { + TestWidgetsFlutterBinding.ensureInitialized(); + + BleEngine newEngine(List logs) => BleEngine( + onRecord: (sample, raw) async {}, + onState: (_) {}, + log: logs.add, + ); + + /// A GET_CLOCK response carrying [strapEpoch] as the strap RTC. + Decoded clockReply(int strapEpoch) => + Decoded('cmd_response', {'clock_epoch': strapEpoch}); + + int wallNow() => DateTime.now().millisecondsSinceEpoch ~/ 1000; + + group('P0 — a slow phone never gets to write its clock onto the strap', () { + test('a plausible strap RTC >1d ahead defers history and blocks SET_CLOCK', + () { + final logs = []; + final engine = newEngine(logs); + + // Strap two days ahead of us and plausible => the PHONE is the suspect + // one. shouldSetClock is also true here (drift > 1 day), which is exactly + // the collision: the drift correction wants to write, the trust gate says + // it must not. + engine.debugAbsorbDecoded(clockReply(wallNow() + 2 * 86400)); + + expect(engine.historyPausedForClock, isTrue, + reason: 'history must defer rather than drain-and-trim'); + expect( + logs.where((l) => l.contains('re-issuing SET_CLOCK')), + isEmpty, + reason: 'writing our slow wall clock onto a correct RTC would corrupt ' + 'it AND destroy the evidence — the read-back then agrees forever', + ); + // Two independent gates have to stay lined up for that to hold: the + // corrupt-read ceiling and the phone-suspect threshold both key off + // kFutureMargin, so today the read is rejected before the SET_CLOCK + // retry is even reached. This asserts the OUTCOME, not which gate got + // there first — so loosening either one still trips the test. + expect(engine.clockRef, isNull, + reason: 'a read we do not trust must not become the alarm ' + 'correlation either'); + }); + + test('a strap RTC BEHIND us is a strap problem and is still corrected', () { + final logs = []; + final engine = newEngine(logs); + + // Two days behind: drifted or unset. Nothing suspicious about the phone, + // so the correction must still run — the gate is not allowed to become a + // blanket "never SET_CLOCK". + engine.debugAbsorbDecoded(clockReply(wallNow() - 2 * 86400)); + + expect(engine.historyPausedForClock, isFalse); + expect(logs.any((l) => l.contains('re-issuing SET_CLOCK')), isTrue); + }); + + test('a fast strap RTC IS corrected once the grace window expires', () { + final logs = []; + final engine = newEngine(logs); + + engine.debugAbsorbDecoded(clockReply(wallNow() + 2 * 86400)); + expect(engine.historyPausedForClock, isTrue); + expect(logs.any((l) => l.contains('re-issuing SET_CLOCK')), isFalse, + reason: 'during grace the phone is still the suspect party'); + + // Twelve hours on and the reading has not budged: the phone would have + // re-synced over NTP long ago, so it is the STRAP that runs fast. + engine.debugExpireClockSuspicion(); + engine.debugAbsorbDecoded(clockReply(wallNow() + 2 * 86400)); + + expect(engine.historyPausedForClock, isFalse, + reason: 'history must stop deferring or sync stalls for good'); + expect( + logs.any((l) => l.contains('re-issuing SET_CLOCK')), + isTrue, + reason: 'OLD BEHAVIOUR: correction was nested inside the ' + 'acceptsClockRead branch, which rejects on the SAME margin that ' + 'flags the strap as fast — so history un-deferred straight back ' + 'onto an uncorrected fast RTC whose records the gate then dropped', + ); + // The two decisions are now independent, which is the whole point: this + // same read is still refused as an alarm correlation (a far-future value + // would arm the alarm years out) while being acted on as a correction. + expect(logs.any((l) => l.contains('corrupt strap RTC read')), isTrue); + expect(engine.clockRef, isNull); + }); + + test('the gate lifts the moment a read shows the clocks agreeing', () { + final logs = []; + final engine = newEngine(logs); + + engine.debugAbsorbDecoded(clockReply(wallNow() + 2 * 86400)); + expect(engine.historyPausedForClock, isTrue); + + // The phone corrected itself over NTP; the next read agrees. + engine.debugAbsorbDecoded(clockReply(wallNow())); + expect(engine.historyPausedForClock, isFalse, + reason: 'sync must resume without waiting out the grace window'); + }); + }); +} diff --git a/test/ble_engine_test.dart b/test/ble_engine_test.dart index 6acce80..3234f63 100644 --- a/test/ble_engine_test.dart +++ b/test/ble_engine_test.dart @@ -1,5 +1,6 @@ import 'package:flutter_test/flutter_test.dart'; import 'package:openstrap_edge/ble/ble_engine.dart'; +import 'package:openstrap_protocol/openstrap_protocol.dart' as proto; void main() { group('historical burst packet accounting', () { @@ -122,6 +123,103 @@ void main() { }); }); + group('burst completeness shortfall (log-only would-flag signal)', () { + test('no shortfall when all-types received total equals num_packets', () { + // Band sent 49 frames (30 R24 + 17 console + 2 event); we received all. + final received = countBurstTrafficPackets( + dataPacketCountsByRevision: const {24: 30}, + consoleCount: 17, + eventCount: 2, + ); + expect( + burstPacketShortfall( + expectedPacketCount: 49, + receivedTrafficCount: received, + ), + 0, + ); + }); + + test( + 'interleaved console/event frames do NOT false-positive: comparing ' + 'against the all-types received total (not the banked R24 subset) ' + 'keeps shortfall at zero', + () { + final received = countBurstTrafficPackets( + dataPacketCountsByRevision: const {24: 15}, + consoleCount: 37, + eventCount: 2, + ); + // Banked R24 subset alone is 15 — comparing THAT to num_packets=54 + // would fabricate a 39-frame "loss". The correct all-types total is 54. + expect(received, 54); + expect( + burstPacketShortfall( + expectedPacketCount: 54, + receivedTrafficCount: received, + ), + 0, + ); + }, + ); + + test( + 'positive shortfall flags missing-or-corrupted traffic (band counted ' + 'more than we did)', () { + final received = countBurstTrafficPackets( + dataPacketCountsByRevision: const {24: 20}, + consoleCount: 3, + ); + // Band reported 30, we counted 23 all-types, nothing gate-dropped → 7 + // frames the band sent that we did not count (never arrived or CRC-failed). + expect( + burstPacketShortfall( + expectedPacketCount: 30, + receivedTrafficCount: received, + ), + 7, + ); + }); + + test('gate-dropped records are added back so they never read as loss', () { + // 26 all-types received, 24 legitimately gate-dropped, band expected 50 → + // fully explained, no true loss. + expect( + burstPacketShortfall( + expectedPacketCount: 50, + receivedTrafficCount: 26, + droppedThisBurst: 24, + ), + 0, + ); + }); + + test('negative shortfall (retries/dupes counted extra) is not loss', () { + expect( + burstPacketShortfall( + expectedPacketCount: 26, + receivedTrafficCount: 28, + ), + lessThan(0), + ); + }); + + test('shortfall==0 is exactly burstPacketCountMatches', () { + const expected = 50, received = 26, dropped = 24; + final matches = burstPacketCountMatches( + expectedPacketCount: expected, + actualBurstPacketCount: received, + droppedThisBurst: dropped, + ); + final shortfall = burstPacketShortfall( + expectedPacketCount: expected, + receivedTrafficCount: received, + droppedThisBurst: dropped, + ); + expect(matches, (shortfall == 0)); + }); + }); + group('maintenance traffic gating', () { test('maintenance traffic is paused while offload is active', () { expect(shouldPauseMaintenanceTraffic(offloadActive: true), isTrue); @@ -131,4 +229,15 @@ void main() { expect(shouldPauseMaintenanceTraffic(offloadActive: false), isFalse); }); }); + + group('INIT drain gate', () { + // sendInit(drain: false) drops the LAST init packet to skip the flash + // drain under a suspect phone clock. That is only correct while the last + // packet actually IS SEND_HISTORICAL_DATA — if INIT is ever reordered this + // must fail rather than silently skip an unrelated command (and let the + // drain through under the bad clock). + test('the last INIT packet is the historical drain', () { + expect(proto.initPackets.last, proto.cmdSendHistorical(4)); + }); + }); } diff --git a/test/db_integrity_test.dart b/test/db_integrity_test.dart index 69b360b..06a9cdf 100644 --- a/test/db_integrity_test.dart +++ b/test/db_integrity_test.dart @@ -1,12 +1,10 @@ // DB integrity regressions, run against the REAL LocalDb over sqflite_ffi: // -// 1. decoded_rr ORPHAN GUARD — a post-reboot rec_ts collision (two counters, -// one second) must not strand the evicted counter's RR beats under a -// counter with no decoded_onehz row (the counter-joined prune can never -// select those → permanent leak, and the loser's extra beat indexes would -// survive the winner's UNIQUE(rr_ts_ms, beat_index) REPLACE). -// 2. prune ORPHAN SWEEP — pre-existing orphans (written by pre-guard builds) -// are cleaned by pruneRawBeforeRecTs once their window is pruned. +// 1. decoded_rr REC_TS KEY — a post-reboot rec_ts collision (two counters, one +// second) keeps exactly the winner's beats; the write path DELETEs the +// second's beats before reinserting, so a shrinking beat count strands none. +// 2. prune BY REC_TS — pruneDecodedBeforeRecTs deletes decoded_rr by its rec_ts +// key (no counter subquery, no orphan sweep) and keeps recent beats. // 3. importFromDbFile FINALIZED protection — a foreign export never overwrites // a locally-finalized (day_id, algo_version) day_result row; non-finalized // rows keep the merge-REPLACE behavior. @@ -62,14 +60,13 @@ void main() { expect(health['ok'], isTrue, reason: '$health'); }); - test('rec_ts collision leaves no orphaned decoded_rr beats', () async { + test('rec_ts collision leaves exactly the winner\'s beats', () async { const ts = 1780000000; - // Pre-reboot record: high counter, THREE beats. + // Two records for the SAME second, different counters (a reboot straddle). + // THREE beats then TWO. REPLACE on the rec_ts key keeps the newest row; the + // write path DELETEs the second's beats before reinserting, so no stale + // high-index beat survives. await LocalDb.insertRecord(_raw(ts, 100), _sample(ts, 100, [800, 810, 820])); - // Post-reboot record for the SAME second: counter reset low, TWO beats. - // REPLACE on UNIQUE(rec_ts) evicts counter 100's decoded_onehz row; without - // the guard its beats (incl. beat_index 2, which the winner's two-beat - // REPLACE never touches) would be orphaned forever. await LocalDb.insertRecord(_raw(ts, 5), _sample(ts, 5, [900, 910])); final db = await LocalDb.instance; @@ -77,30 +74,30 @@ void main() { expect(onehz.length, 1); expect(onehz.first['counter'], 5); // newest-wins - // No RR beats survive under the evicted counter… - final loserBeats = - await db.query('decoded_rr', where: 'counter = ?', whereArgs: [100]); - expect(loserBeats, isEmpty); - // …and globally: zero orphans (every beat's counter owns a decoded row). + // Only the winner's two beats remain, keyed by rec_ts. + final beats = await db.query('decoded_rr', + where: 'rec_ts = ?', whereArgs: [ts], orderBy: 'beat_index ASC'); + expect([for (final b in beats) b['rr_ms']], [900, 910]); + + // Globally: zero orphans (every beat's rec_ts owns a decoded row). final orphans = await db.rawQuery( 'SELECT COUNT(*) c FROM decoded_rr ' - 'WHERE counter NOT IN (SELECT counter FROM decoded_onehz)', + 'WHERE rec_ts NOT IN (SELECT rec_ts FROM decoded_onehz)', ); expect(orphans.first['c'], 0); - // The RR read path (decodedRrByCounterRange, joined to frames by counter in - // derive_prepare.addDecodedPage) sees ONLY the winner's beats. - final rr = await LocalDb.decodedRrByCounterRange(fromCounter: 0, toCounter: 1 << 30); + // The RR read path sees ONLY the winner's beats. + final rr = await LocalDb.decodedRrByRecTsRange(fromRecTs: ts, toRecTs: ts); expect([for (final r in rr) r['rr_ms']], [900, 910]); - expect({for (final r in rr) r['counter']}, {5}); }); - test('prune sweeps pre-existing decoded_rr orphans', () async { + test('prune deletes decoded_rr by rec_ts, keeps recent beats', () async { const oldTs = 1700000000; // strictly before the cutoff below final db = await LocalDb.instance; - // Simulate a pre-guard leak: an RR beat whose counter has no decoded row. + // An old beat (its owning row absent — e.g. a leftover) is still deleted by + // the plain rec_ts-range prune; no counter subquery, no orphan sweep needed. await db.insert('decoded_rr', { - 'counter': 999999, + 'rec_ts': oldTs, 'beat_index': 0, 'rr_ts_ms': oldTs * 1000, 'rr_ms': 850, @@ -111,9 +108,9 @@ void main() { final deleted = await LocalDb.pruneDecodedBeforeRecTs(oldTs + 1000); - final orphan = await db.query('decoded_rr', where: 'counter = ?', whereArgs: [999999]); - expect(orphan, isEmpty, reason: 'orphan sweep must clean the leaked beat'); - final kept = await db.query('decoded_rr', where: 'counter = ?', whereArgs: [7]); + final old = await db.query('decoded_rr', where: 'rec_ts = ?', whereArgs: [oldTs]); + expect(old, isEmpty, reason: 'the pruned window\'s beats must be deleted'); + final kept = await db.query('decoded_rr', where: 'rec_ts = ?', whereArgs: [keepTs]); expect(kept.length, 1); // used to always come back 0 even when rows genuinely got pruned - none // of the txn.delete() counts were ever added up. diff --git a/test/db_migration_ladder_test.dart b/test/db_migration_ladder_test.dart index 4b471d5..1a30878 100644 --- a/test/db_migration_ladder_test.dart +++ b/test/db_migration_ladder_test.dart @@ -22,6 +22,7 @@ import 'package:path/path.dart' as p; import 'package:sqflite_common_ffi/sqflite_ffi.dart'; import 'package:openstrap_edge/data/db.dart'; import 'package:openstrap_edge/data/journal_fields.dart'; +import 'package:openstrap_edge/data/models.dart'; /// The pre-v3 raw_records shape: keyed by frame hex, NO rec_ts column. const _legacyRawDdl = ''' @@ -46,6 +47,29 @@ const _v6RawDdl = ''' ) '''; +/// The origin/main (pre-v33) COUNTER-keyed decoded store, exactly as a user who +/// installed at v19..31 has it — and by then raw_records is already DROPPED, so +/// the v33 rekey is the ONLY copy of their 1 Hz data (no raw-backfill safety +/// net). This is the highest-risk path the re-key touches. +const _counterKeyedDecodedDdl = [ + ''' + CREATE TABLE decoded_onehz ( + counter INTEGER PRIMARY KEY, rec_ts INTEGER NOT NULL, + hr INTEGER NOT NULL, ax REAL NOT NULL, ay REAL NOT NULL, az REAL NOT NULL, + spo2_red_raw INTEGER NOT NULL, spo2_ir_raw INTEGER NOT NULL, + skin_temp_raw INTEGER NOT NULL) +''', + 'CREATE UNIQUE INDEX idx_decoded_onehz_rec_ts_unique ON decoded_onehz(rec_ts)', + ''' + CREATE TABLE decoded_rr ( + counter INTEGER NOT NULL, beat_index INTEGER NOT NULL, + rr_ts_ms INTEGER NOT NULL, rr_ms INTEGER NOT NULL, + PRIMARY KEY (counter, beat_index)) +''', + 'CREATE UNIQUE INDEX idx_decoded_rr_ts_beat_unique ' + 'ON decoded_rr(rr_ts_ms, beat_index)', +]; + /// The v5-era derived tables, so step 9's derived_day → day_result copy is real. const _v5DerivedDdl = [ ''' @@ -417,4 +441,170 @@ void main() { expect(health['ok'], isTrue, reason: '$health'); }, ); + + test( + 'v33 re-key converts a v31 COUNTER-keyed decoded store to rec_ts LOSSLESSLY ' + '— raw_records is already dropped, so the rekey is the only copy', + () async { + const name = 'migrate_from_v31_counterkeyed_test.db'; + created.add(name); + await _seedOldDb( + name, + 31, + [..._counterKeyedDecodedDdl, ..._v5DerivedDdl], + seedRows: (db) async { + // Three distinct seconds, distinct counters (valid old-schema data). + for (final r in const [ + [100, 1785000000, 60], + [101, 1785000001, 61], + [102, 1785000002, 62], + ]) { + await db.insert('decoded_onehz', { + 'counter': r[0], 'rec_ts': r[1], 'hr': r[2], + 'ax': 0.0, 'ay': 0.0, 'az': 0.0, + 'spo2_red_raw': 0, 'spo2_ir_raw': 0, 'skin_temp_raw': 0, + }); + } + // Beats under counter 100 (two) and 102 (one). + await db.insert('decoded_rr', { + 'counter': 100, 'beat_index': 0, + 'rr_ts_ms': 1785000000 * 1000, 'rr_ms': 800, + }); + await db.insert('decoded_rr', { + 'counter': 100, 'beat_index': 1, + 'rr_ts_ms': 1785000000 * 1000, 'rr_ms': 810, + }); + await db.insert('decoded_rr', { + 'counter': 102, 'beat_index': 0, + 'rr_ts_ms': 1785000002 * 1000, 'rr_ms': 900, + }); + }, + ); + + expect(await _openThroughLocalDb(name), LocalDb.schemaVersion); + final db = await LocalDb.instance; + + // Every second survives; counter is preserved as the forensic column. + final oh = await db.query('decoded_onehz', orderBy: 'rec_ts ASC'); + expect([for (final r in oh) r['rec_ts']], + [1785000000, 1785000001, 1785000002]); + expect([for (final r in oh) r['counter']], [100, 101, 102]); + // PK is now rec_ts, not counter. + final ohInfo = await db.rawQuery('PRAGMA table_info(decoded_onehz)'); + expect(ohInfo.firstWhere((c) => c['name'] == 'rec_ts')['pk'], 1); + expect(ohInfo.firstWhere((c) => c['name'] == 'counter')['pk'], 0); + + // Beats re-home onto their real second; decoded_rr loses its counter col. + final rr = + await db.query('decoded_rr', orderBy: 'rec_ts ASC, beat_index ASC'); + expect([for (final r in rr) r['rec_ts']], + [1785000000, 1785000000, 1785000002]); + expect([for (final r in rr) r['rr_ms']], [800, 810, 900]); + final rrInfo = await db.rawQuery('PRAGMA table_info(decoded_rr)'); + expect(rrInfo.any((c) => c['name'] == 'counter'), isFalse); + + // No stranded beats, no cross-stamped timestamps, no leaked temp tables. + expect( + (await db.rawQuery( + 'SELECT COUNT(*) c FROM decoded_rr WHERE rr_ts_ms != rec_ts * 1000', + )).first['c'], + 0, + ); + expect( + await db.rawQuery( + "SELECT name FROM sqlite_master WHERE name LIKE '%\\_v33' ESCAPE '\\' " + "OR name LIKE '%\\_new' ESCAPE '\\'", + ), + isEmpty, + ); + + final health = await LocalDb.schemaHealth(); + expect(health['ok'], isTrue, reason: '$health'); + }, + ); + + test( + 'v31→v32 re-keys raw_archive off the volatile counter onto frame hex ' + 'without losing a distinct frame, collapsing only exact-duplicate hex', + () async { + const name = 'migrate_from_v31_rawarchive_test.db'; + created.add(name); + // The pre-v32 raw_archive shape: keyed by the strap counter, which resets + // to ~0 on reboot — so a post-reboot frame reusing a live counter was + // silently IGNORE-dropped in the one table meant to never lose a frame. + await _seedOldDb( + name, + 31, + const [ + ''' + CREATE TABLE raw_archive ( + counter INTEGER PRIMARY KEY, + hex TEXT NOT NULL, + packet_type INTEGER NOT NULL, + rec_ts INTEGER, + captured_at INTEGER NOT NULL, + reason TEXT NOT NULL + ) + ''', + 'CREATE INDEX idx_raw_archive_captured ON raw_archive(captured_at DESC)', + ], + seedRows: (db) async { + Future row(int counter, String hex) => db.insert('raw_archive', { + 'counter': counter, + 'hex': hex, + 'packet_type': 0x2F, + 'captured_at': 1750000000000 + counter, + 'reason': 'undecodable_rec_v99', + }); + // Three distinct frames (distinct counter AND hex) — none may be lost. + await row(1, 'aa01'); + await row(2, 'bb02'); + await row(3, 'cc03'); + // Two rows the OLD counter-PK allowed but that carry IDENTICAL bytes; + // the content re-key must collapse them to one (the dedup we want). + await row(10, 'ff06'); + await row(11, 'ff06'); + }, + ); + + expect(await _openThroughLocalDb(name), LocalDb.schemaVersion); + + // 5 old rows → 4: the three distinct frames survive, the duplicate-hex + // pair collapses to one. Nothing distinct was lost. + final stats = await LocalDb.rawArchiveStats(); + expect(stats['count'], 4); + + // The migrated table is now hex-PK, proven end-to-end through the REAL + // ladder (not just a fresh onCreate): two DISTINCT frames that reuse ONE + // counter both survive — the exact loss the old counter-PK caused. + await LocalDb.archiveRawRecord(ArchiveRecord( + counter: 1, // reuses a counter already present from the seed + hex: 'dd04', + packetType: 0x2F, + capturedAt: 1750000500000, + reason: 'undecodable_post_reboot', + )); + await LocalDb.archiveRawRecord(ArchiveRecord( + counter: 1, // SAME counter, DIFFERENT bytes + hex: 'ee05', + packetType: 0x2F, + capturedAt: 1750000600000, + reason: 'undecodable_post_reboot', + )); + expect((await LocalDb.rawArchiveStats())['count'], 6); + + // …and an identical re-flood still dedups on content. + await LocalDb.archiveRawRecord(ArchiveRecord( + counter: 999, // different counter, but bytes already archived + hex: 'dd04', + packetType: 0x2F, + capturedAt: 1750000700000, + reason: 'undecodable_post_reboot', + )); + expect((await LocalDb.rawArchiveStats())['count'], 6); + + final health = await LocalDb.schemaHealth(); + expect(health['ok'], isTrue, reason: '$health'); + }, + ); } diff --git a/test/db_p0_fixes_test.dart b/test/db_p0_fixes_test.dart index 1fa07cd..cfde5f3 100644 --- a/test/db_p0_fixes_test.dart +++ b/test/db_p0_fixes_test.dart @@ -88,49 +88,60 @@ void main() { if (await tmp.exists()) await tmp.delete(recursive: true); }); - // ── fix 3 ──────────────────────────────────────────────────────────────── + // ── fix 3: THE re-key regression ───────────────────────────────────────── test( - 'a REUSED counter (reboot reset) leaves no stale-timestamped decoded_rr ' - 'beats behind — the counter-PK eviction is guarded too', + 'a REUSED counter (reboot reset) KEEPS both seconds — the counter-PK ' + 'eviction that silently, unrecoverably deleted a 1 Hz row is gone', () async { - const older = 1785000000; - const newer = 1785000600; // a DIFFERENT second, same counter + const older = 1785000000; // pre-reboot, counter 777 + const newer = 1785000600; // post-reboot, SAME counter 777 const counter = 777; await LocalDb.insertRecord( _raw(older, counter), _sample(older, counter, [800, 810, 820]), // THREE beats ); - // Post-reboot the counter is handed out again, now for a later second. - // The INSERT-OR-REPLACE evicts the older second's decoded_onehz row via - // the `counter` PRIMARY KEY; only beat_index 0 and 1 are overwritten, so - // beat_index 2 used to SURVIVE still stamped with `older` — invisible to - // both prune paths, and it polluted every later RR read of that counter. + // Pre-fix: `counter` was decoded_onehz's PRIMARY KEY, so this REPLACE + // evicted the `older` row entirely — and raw_records is dropped, so that + // 1 Hz second was gone for good. Under the rec_ts key both seconds live. await LocalDb.insertRecord( _raw(newer, counter), _sample(newer, counter, [900, 910]), // TWO beats ); final db = await LocalDb.instance; - final beats = await db.query( - 'decoded_rr', - where: 'counter = ?', - whereArgs: [counter], - orderBy: 'beat_index ASC', + final rows = await db.query( + 'decoded_onehz', + where: 'rec_ts IN (?, ?)', + whereArgs: [older, newer], + orderBy: 'rec_ts ASC', ); - expect(beats, hasLength(2), reason: 'the third beat must not survive'); - expect( - beats.every((b) => b['rr_ts_ms'] == newer * 1000), - isTrue, - reason: 'no beat may carry the evicted second\'s timestamp: $beats', + expect(rows, hasLength(2), + reason: 'the pre-reboot second must NOT be evicted'); + expect([for (final r in rows) r['counter']], [counter, counter]); + + // Each second keeps its own beats, keyed by its rec_ts. + final oldBeats = await db.query('decoded_rr', + where: 'rec_ts = ?', whereArgs: [older], orderBy: 'beat_index ASC'); + expect([for (final b in oldBeats) b['rr_ms']], [800, 810, 820]); + final newBeats = await db.query('decoded_rr', + where: 'rec_ts = ?', whereArgs: [newer], orderBy: 'beat_index ASC'); + expect([for (final b in newBeats) b['rr_ms']], [900, 910]); + + // Re-offloading `older` with FEWER beats must not strand the third — + // the write path DELETEs the second's beats before reinserting. + await LocalDb.insertRecord( + _raw(older, counter), + _sample(older, counter, [850]), // ONE beat now ); - expect([for (final b in beats) b['rr_ms']], [900, 910]); + final reBeats = await db.query('decoded_rr', + where: 'rec_ts = ?', whereArgs: [older], orderBy: 'beat_index ASC'); + expect([for (final b in reBeats) b['rr_ms']], [850], + reason: 'stale high-index beats must not survive a shrink'); - // And globally: no orphans, no cross-second contamination. + // No beat carries the wrong second's timestamp, anywhere. final stale = await db.rawQuery( - 'SELECT COUNT(*) c FROM decoded_rr rr ' - 'JOIN decoded_onehz d ON d.counter = rr.counter ' - 'WHERE rr.rr_ts_ms != d.rec_ts * 1000', + 'SELECT COUNT(*) c FROM decoded_rr WHERE rr_ts_ms != rec_ts * 1000', ); expect(stale.first['c'], 0); }, @@ -138,8 +149,8 @@ void main() { // ── fix 4 ──────────────────────────────────────────────────────────────── test( - 'decodedRrByCounterRange returns a page spanning a counter RESET — the ' - 'endpoints are page bounds, not a monotonic counter span', + 'decodedRrByRecTsRange returns a page spanning a counter RESET — the ' + 'window is by rec_ts, immune to the reboot counter reset', () async { const t0 = 1785100000; // Pre-reboot: high counter. Post-reboot: the counter restarts near zero, @@ -161,16 +172,19 @@ void main() { toRecTs: t0 + 1, ); expect(page, hasLength(2)); - final first = (page.first['counter'] as num).toInt(); - final last = (page.last['counter'] as num).toInt(); - expect(first, 1200000); - expect(last, 5, reason: 'the page really does end on a LOWER counter'); - - final rr = await LocalDb.decodedRrByCounterRange( - fromCounter: first, - toCounter: last, + final first = (page.first['rec_ts'] as num).toInt(); + final last = (page.last['rec_ts'] as num).toInt(); + expect(first, t0); + expect(last, t0 + 1); + // Sanity: the page really does end on a LOWER counter (reboot straddle). + expect((page.first['counter'] as num).toInt(), 1200000); + expect((page.last['counter'] as num).toInt(), 5); + + final rr = await LocalDb.decodedRrByRecTsRange( + fromRecTs: first, + toRecTs: last, ); - // `counter >= 1200000 AND counter <= 5` used to match nothing at all — + // A counter span (`>= 1200000 AND <= 5`) used to match nothing at all — // the window silently produced no RR beats, so no RMSSD/HRV, no error. expect(rr, hasLength(4)); expect([for (final r in rr) r['rr_ms']], [800, 805, 900, 905]); @@ -318,27 +332,28 @@ void main() { // ── fix 8 ──────────────────────────────────────────────────────────────── test( - 'importFromDbFile routes decoded_onehz through the orphan guard', + 'importFromDbFile merges a LEGACY (counter-keyed, no rec_ts) decoded export ' + 'cleanly under the rec_ts key — foreign-wins, no stranded beats', () async { final db = await LocalDb.instance; await db.delete('decoded_onehz'); await db.delete('decoded_rr'); - const collideTs = 1786000000; // rec_ts collision, different counter - const reuseCounter = 8003; // counter collision, different rec_ts - const localReuseTs = 1786000500; - const foreignReuseTs = 1786009999; + const collideTs = 1786000000; // rec_ts present locally AND in the foreign + const t2 = 1786000500; // local-only second + const t3 = 1786009999; // foreign-only second await LocalDb.insertRecord( _raw(collideTs, 8002), _sample(collideTs, 8002, [700, 710, 720]), ); await LocalDb.insertRecord( - _raw(localReuseTs, reuseCounter), - _sample(localReuseTs, reuseCounter, [600, 610, 620]), + _raw(t2, 8003), + _sample(t2, 8003, [600, 610]), ); - // A foreign export that collides both ways. + // A foreign export in the OLD schema: decoded_rr has a `counter` column + // and NO rec_ts (the import must derive rec_ts from rr_ts_ms). final dir = await databaseFactory.getDatabasesPath(); final srcPath = p.join(dir, 'p0_foreign_export.db'); await databaseFactory.deleteDatabase(srcPath); @@ -378,38 +393,46 @@ void main() { } } - await foreign(8001, collideTs, [500]); // same second, other counter - await foreign(reuseCounter, foreignReuseTs, [400]); // same counter, other second + // Same second as local, but with FEWER beats than the 3 already stored. + // An equal-or-larger foreign set hides the bug: every local beat_index + // gets overwritten and a row-by-row replace-insert looks correct. Only a + // shrinking set exposes the stale local tail (beats 1 and 2) that a + // merge-without-clear leaves spliced onto the foreign series. + await foreign(8001, collideTs, [500]); + await foreign(9999, t3, [400]); await src.close(); await LocalDb.importFromDbFile(srcPath); - // (a) UNIQUE(rec_ts) eviction: the local counter's beats went with it. - expect( - await db.query('decoded_rr', where: 'counter = ?', whereArgs: [8002]), - isEmpty, - reason: 'the evicted counter\'s beats must not be stranded', - ); - // (b) counter-PK eviction: no beat under the reused counter still carries - // the local second's timestamp. - final reused = await db.query( - 'decoded_rr', - where: 'counter = ?', - whereArgs: [reuseCounter], - ); - expect(reused, hasLength(1)); - expect(reused.first['rr_ts_ms'], foreignReuseTs * 1000); + // Foreign wins the rec_ts collision; the other two seconds are untouched. + final onehz = await db.query('decoded_onehz', orderBy: 'rec_ts ASC'); + expect([for (final r in onehz) r['rec_ts']], [collideTs, t2, t3]); + final collided = + onehz.firstWhere((r) => r['rec_ts'] == collideTs); + expect(collided['counter'], 8001, reason: 'foreign row won'); + expect(collided['hr'], 61); + + // The collided second's beats are EXACTLY the foreign set. Not a merge: + // the local [700, 710, 720] must be gone, tail included. + final b1 = await db.query('decoded_rr', + where: 'rec_ts = ?', whereArgs: [collideTs], orderBy: 'beat_index ASC'); + expect([for (final b in b1) b['rr_ms']], [500], + reason: 'stale local beats 1-2 would splice a foreign/local RR ' + 'series into one second and silently corrupt its RMSSD'); + // The foreign-only second imported with rec_ts derived from rr_ts_ms. + final b3 = await db.query('decoded_rr', + where: 'rec_ts = ?', whereArgs: [t3]); + expect([for (final b in b3) b['rr_ms']], [400]); + expect(b3.first['rr_ts_ms'], t3 * 1000); // Nothing orphaned, nothing cross-stamped, anywhere. final orphans = await db.rawQuery( 'SELECT COUNT(*) c FROM decoded_rr ' - 'WHERE counter NOT IN (SELECT counter FROM decoded_onehz)', + 'WHERE rec_ts NOT IN (SELECT rec_ts FROM decoded_onehz)', ); expect(orphans.first['c'], 0); final stale = await db.rawQuery( - 'SELECT COUNT(*) c FROM decoded_rr rr ' - 'JOIN decoded_onehz d ON d.counter = rr.counter ' - 'WHERE rr.rr_ts_ms != d.rec_ts * 1000', + 'SELECT COUNT(*) c FROM decoded_rr WHERE rr_ts_ms != rec_ts * 1000', ); expect(stale.first['c'], 0); }, diff --git a/test/db_paged_import_export_test.dart b/test/db_paged_import_export_test.dart index 4575824..ae968ca 100644 --- a/test/db_paged_import_export_test.dart +++ b/test/db_paged_import_export_test.dart @@ -169,10 +169,10 @@ void main() { )).first['c']; expect(distinct, rows); - // Nothing stranded: the orphan guard still runs per row under paging. + // Nothing stranded: every beat's rec_ts owns a decoded_onehz row. final orphans = (await db.rawQuery( 'SELECT COUNT(*) c FROM decoded_rr ' - 'WHERE counter NOT IN (SELECT counter FROM decoded_onehz)', + 'WHERE rec_ts NOT IN (SELECT rec_ts FROM decoded_onehz)', )).first['c']; expect(orphans, 0); }); @@ -277,9 +277,4 @@ void main() { } }); }); - - test('the degraded RR fallback truncation counter starts clean', () async { - // A non-zero value means some window computed HRV from truncated beats. - expect(LocalDb.decodedRrFallbackTruncations, 0); - }); } diff --git a/test/db_storage_hygiene_test.dart b/test/db_storage_hygiene_test.dart index 56f3465..16e980d 100644 --- a/test/db_storage_hygiene_test.dart +++ b/test/db_storage_hygiene_test.dart @@ -1,7 +1,6 @@ // Storage hygiene: -// 1. decoded_rr's rr_ts_ms lookups are still index-served after dropping -// idx_decoded_rr_ts, which was a strict prefix of the (rr_ts_ms, -// beat_index) unique index and only added write cost. +// 1. decoded_rr is keyed (rec_ts, beat_index) and carries no redundant +// secondary index; rec_ts-range reads are served by the PK auto-index. // 2. Superseded generations of the recomputable per-day intermediates are // pruned. They are keyed (day_id, algo_version), so every kAlgoVersion // bump wrote a whole new generation beside the old one and nothing @@ -28,85 +27,40 @@ void main() { await databaseFactory.deleteDatabase(p.join(dir, LocalDb.dbName)); }); - test('the redundant single-column rr index is gone', () async { + test('decoded_rr carries no redundant secondary index', () async { final db = await LocalDb.instance; - final idx = (await db.rawQuery( - "SELECT name FROM sqlite_master WHERE type='index' AND tbl_name='decoded_rr'", - )).map((r) => r['name'] as String?).whereType().toList(); - expect(idx, isNot(contains('idx_decoded_rr_ts'))); - expect(idx, contains('idx_decoded_rr_ts_beat_unique')); - }); - - test('an existing duplicate-of-primary-key rr index is dropped on open', () async { - // idx_decoded_rr_counter(counter, beat_index) duplicated, column for column, - // the index PRIMARY KEY (counter, beat_index) already creates. Measured on a - // 3-day fill: both b-trees 3,264,512 bytes — ~1.09 MB/day of pure - // duplication plus a second b-tree write per beat on the hottest insert - // path in the app. - // - // PLANTED FIRST, then reopened. A fresh database never creates the index - // any more, so simply asserting it is absent asserts nothing — deleting the - // DROP leaves the test green. The installs that have the index are the ones - // that were created before it stopped being written, and the only thing - // that removes it for them is `_repairOpenSchema` on the next open. That is - // the path this reproduces. - var db = await LocalDb.instance; - await db.execute( - 'CREATE INDEX IF NOT EXISTS idx_decoded_rr_counter ' - 'ON decoded_rr(counter, beat_index)', - ); + final idx = await _rrIndexes(db); + // The (rec_ts, beat_index) PRIMARY KEY auto-indexes; nothing else should be + // maintained on this hot insert path. expect( - await _rrIndexes(db), - contains('idx_decoded_rr_counter'), - reason: 'the fixture must actually plant the index', + idx.where((n) => !n.startsWith('sqlite_autoindex')), + isEmpty, + reason: 'unexpected secondary index on decoded_rr: $idx', ); - - await LocalDb.close(); - db = await LocalDb.instance; - expect(await _rrIndexes(db), isNot(contains('idx_decoded_rr_counter'))); }); - test('counter lookups are still index-served without it', () async { - // Dropping an index is only safe if the planner has another. The PK's - // auto-index covers exactly the same columns in the same order. + test('rec_ts-range reads on decoded_rr are served by the PK auto-index', () async { + // decoded_rr shares the rec_ts key with decoded_onehz, so the derive read + // path (decodedRrByRecTsRange) is a PK range scan — never a full-table read. final db = await LocalDb.instance; - for (final sql in const [ - 'EXPLAIN QUERY PLAN SELECT * FROM decoded_rr WHERE counter = 42 ' - 'ORDER BY beat_index', - 'EXPLAIN QUERY PLAN SELECT * FROM decoded_rr WHERE counter BETWEEN 1 AND 9', - ]) { - final detail = (await db.rawQuery( - sql, - )).map((r) => r['detail'].toString()).join(' | '); - expect( - detail.toUpperCase(), - contains('USING'), - reason: 'planner fell back to a full scan: $detail', - ); - expect( - detail, - contains('sqlite_autoindex_decoded_rr_1'), - reason: 'expected the primary key auto-index: $detail', - ); - } - }); - - test('rr_ts_ms range scans are still served by an index', () async { - final db = await LocalDb.instance; - final plan = await db.rawQuery( - 'EXPLAIN QUERY PLAN SELECT * FROM decoded_rr WHERE rr_ts_ms < 1000 ' - 'ORDER BY rr_ts_ms ASC, beat_index ASC', - ); - final detail = plan.map((r) => r['detail'].toString()).join(' | '); + final detail = (await db.rawQuery( + 'EXPLAIN QUERY PLAN SELECT * FROM decoded_rr WHERE rec_ts BETWEEN 1 AND 9 ' + 'ORDER BY rec_ts ASC, beat_index ASC', + )).map((r) => r['detail'].toString()).join(' | '); + // Assert the SHAPE of the plan, not its wording. `sqlite_autoindex_ + // decoded_rr_1` is an internal name SQLite is free to change, and the + // property under test is only "seek, don't scan, and don't sort" — which + // SEARCH + no temp b-tree says on every version. + final plan = detail.toUpperCase(); expect( - detail, - contains('idx_decoded_rr_ts_beat_unique'), - reason: 'planner fell back to a scan: $detail', + plan, + contains('SEARCH'), + reason: 'planner fell back to a full scan: $detail', ); expect( - detail.toUpperCase(), + plan, isNot(contains('USE TEMP B-TREE')), - reason: 'ordering should come from the index: $detail', + reason: 'ordering should come from the PK: $detail', ); }); diff --git a/test/local_persistence_test.dart b/test/local_persistence_test.dart index 380bcbf..835b86d 100644 --- a/test/local_persistence_test.dart +++ b/test/local_persistence_test.dart @@ -243,9 +243,9 @@ void main() { expect(frames.first['hr'], 61); expect(frames.first['spo2_red_raw'], 1234); - final rr = await LocalDb.decodedRrByCounterRange( - fromCounter: 424242, - toCounter: 424242, + final rr = await LocalDb.decodedRrByRecTsRange( + fromRecTs: startSec, + toRecTs: startSec, ); expect(rr, hasLength(2)); expect(rr.first['rr_ms'], 980); diff --git a/test/raw_archive_test.dart b/test/raw_archive_test.dart index 5b35d66..4f9ffdc 100644 --- a/test/raw_archive_test.dart +++ b/test/raw_archive_test.dart @@ -75,9 +75,9 @@ void main() { expect(await LocalDb.getCursorInt('counter_hw'), 1001); }); - test('re-flood of the same archived counter dedups (IGNORE on counter PK)', () async { + test('identical re-flood dedups on frame hex (missed-ACK redelivery)', () async { final archive = ArchiveRecord( - counter: 2002, // same counter as above + counter: 2002, // same counter AND same bytes as above hex: '2f63deadbeef', packetType: 0x2F, capturedAt: 1750000099999, @@ -89,7 +89,7 @@ void main() { trimToken: 'aabbccddeeff0022', archives: [archive], ); - // Still one archived row — the re-delivery did not duplicate it. + // Still one archived row — same bytes, so the redelivery deduped. final stats = await LocalDb.rawArchiveStats(); expect(stats['count'], 1); // …but the trim cursor still advanced (this chunk was ACK-safe). @@ -108,4 +108,31 @@ void main() { expect(stats['count'], 2); expect((stats['by_reason'] as Map)['undecodable_rec_v112'], 1); }); + + test('two DISTINCT frames sharing a reused counter BOTH survive', () async { + // The strap resets its record counter to ~0 on reboot, so a post-reboot + // frame can reuse a pre-reboot counter while carrying different bytes. + // Under the old `counter INTEGER PRIMARY KEY` + IGNORE, the second frame + // was silently DROPPED — permanent loss in the table that exists precisely + // to never lose a frame. Content-keyed, both must survive. + final before = (await LocalDb.rawArchiveStats())['count'] as int; + const reusedCounter = 4004; + await LocalDb.archiveRawRecord(ArchiveRecord( + counter: reusedCounter, + hex: '2f63aaaaaaaa', // pre-reboot frame + packetType: 0x2F, + capturedAt: 1750000200000, + reason: 'undecodable_pre_reboot', + )); + await LocalDb.archiveRawRecord(ArchiveRecord( + counter: reusedCounter, // SAME counter, DIFFERENT bytes + hex: '2f63bbbbbbbb', // post-reboot frame + packetType: 0x2F, + capturedAt: 1750000300000, + reason: 'undecodable_post_reboot', + )); + final after = (await LocalDb.rawArchiveStats())['count'] as int; + // Pre-fix: +1 (second dropped by counter-PK IGNORE). Post-fix: +2. + expect(after - before, 2); + }); } diff --git a/test/sync_policy_test.dart b/test/sync_policy_test.dart index 33f2e45..f6c952c 100644 --- a/test/sync_policy_test.dart +++ b/test/sync_policy_test.dart @@ -77,6 +77,53 @@ void main() { expect(ClockPolicy.shouldSetClock(wall + 86400 + 1, wall), isTrue); expect(ClockPolicy.shouldSetClock(1000, wall), isTrue); // frozen/unset }); + + test('stops deferring once the disagreement outlives the grace window', () { + // MONOTONIC seconds — an arbitrary stopwatch origin, not an epoch. + const t0 = 1234.0; + const hour = 3600.0; + expect(ClockPolicy.suspectGraceExpired(null, t0), isFalse); + expect(ClockPolicy.suspectGraceExpired(t0, t0), isFalse); + // a slow phone re-syncs over NTP well inside this + expect(ClockPolicy.suspectGraceExpired(t0, t0 + hour), isFalse); + // still disagreeing after the window => the strap rtc is the fast one, + // so history must stop deferring instead of stalling forever + expect(ClockPolicy.suspectGraceExpired(t0, t0 + 13 * hour), isTrue); + }); + + test('a forward wall-clock jump cannot expire the grace window early', () { + // The regression: the window used to be measured with DateTime.now(), so + // the phone stepping its clock forward — the very event this state is + // waiting on, and one that can leave it STILL more than a day behind the + // strap — aged the suspicion instantly and re-authorised the + // drain-and-trim. Read monotonically, a wall jump is simply invisible: + // only real elapsed time moves this forward. + const startedAt = 500.0; + const aMinuteOfRealTimeLater = 560.0; // wall may have jumped days + expect( + ClockPolicy.suspectGraceExpired(startedAt, aMinuteOfRealTimeLater), + isFalse, + reason: 'a minute of real time is a minute, whatever the wall says', + ); + }); + + test('flags a slow PHONE clock: a plausible strap RTC > 1d in the future', () { + // Clocks agree → not suspect. + expect(ClockPolicy.phoneClockSuspect(wall, wall), isFalse); + // Strap up to +1 day ahead is within margin → not suspect. + expect(ClockPolicy.phoneClockSuspect(wall + kFutureMargin, wall), isFalse); + // Plausible strap RTC > 1 day ahead → the phone is likely slow → DEFER + // offload (the P1: draining would drop-then-trim real records). + expect( + ClockPolicy.phoneClockSuspect(wall + kFutureMargin + 1, wall), isTrue); + expect(ClockPolicy.phoneClockSuspect(wall + 2 * 86400, wall), isTrue); + // Strap BEHIND the phone is a plausible-past time — not dropped as future, + // and corrected forward by shouldSetClock — so NOT a phone problem. + expect(ClockPolicy.phoneClockSuspect(wall - 2 * 86400, wall), isFalse); + // An unset/garbage-low RTC is a STRAP problem (shouldSetClock), not the + // phone — must not trip the phone-suspect defer. + expect(ClockPolicy.phoneClockSuspect(1000, wall), isFalse); + }); }); group('BackfillPolicy', () {