From e8ae524fe929df7db03439241e3d17a542b542e0 Mon Sep 17 00:00:00 2001 From: Brackyt <60280126+Brackyt@users.noreply.github.com> Date: Wed, 5 Aug 2026 16:09:29 +0200 Subject: [PATCH 1/3] fix(sync): refuse HISTORY_END trim when gate drops leave an empty buffer Empty drop-only ACKs were advancing the strap cursor and auto-continue while decoded_onehz stayed frozen, permanently deleting unread 1 Hz flash after reconnect. Also require durable rows before lastTrimAdvanced. --- lib/ble/ble_engine.dart | 104 ++++++++++++++++++++++++++++++++--- lib/ble/ble_state.dart | 26 ++++++++- test/ble_safe_trim_test.dart | 68 +++++++++++++++++++++++ 3 files changed, 189 insertions(+), 9 deletions(-) diff --git a/lib/ble/ble_engine.dart b/lib/ble/ble_engine.dart index d21d41d..a59131f 100644 --- a/lib/ble/ble_engine.dart +++ b/lib/ble/ble_engine.dart @@ -694,6 +694,12 @@ class BleEngine { // said" apart from "we correctly, silently rejected some as implausible // (stale-clock block) and never tallied them." See _handleSyncMarker. int _burstDroppedAtStart = 0; + // Consecutive HISTORY_END refuses where the burst banked nothing durable + // but the RecordGate dropped samples. After a short streak we re-issue + // SET_CLOCK and bounce — the usual cause is a bad post-reconnect window + // that would otherwise loop empty drop-ACKs (or, after the refuse guard, + // re-deliver forever without ever correcting the clock). + int _noDurableTrimRefuseStreak = 0; // Band-truth reconciliation: `expectedPacketCount` mismatches are advisory // (see the comment at the validation site — treating a single mismatch as // fatal was actively harmful and was reverted), but a mismatch that keeps @@ -1158,6 +1164,7 @@ class BleEngine { _autoContinue.end(); _lastBackfillAt = 0; _successfulBursts = 0; + _noDurableTrimRefuseStreak = 0; _lastHpsTerminal = null; _sessionPacketCounts = _SessionPacketCounts.zero; _sessionGapSummary = _SessionGapSummary.zero; @@ -1893,8 +1900,9 @@ class BleEngine { // Drop records whose unix is implausible vs wall-clock and (when known) the // strap's own GET_DATA_RANGE window — a previous owner's wandering-clock // pollution. Records with no decodable ts are kept (can't gate them). - // Rejected records are neither stored nor counted; the ACK still walks the - // band's cursor. + // Rejected records are neither stored nor counted. Mixed bursts (some + // rows banked) may still ACK; a drop-only empty burst must not — see + // TrimAckVerdict.blockedNoDurableProgress. // Past this point [sample] is non-null — undecodable records returned above. if (!_recordGate.admit( sample.tsEpoch, @@ -2233,6 +2241,51 @@ class BleEngine { ); } return; + case TrimAckVerdict.blockedNoDurableProgress: + // Gate rejected every historical sample this burst and nothing was + // banked (no raws, no archives). Echoing the token would trim flash + // we never stored — the HR-gap / frozen-cursor failure mode after a + // reconnect with a bad plausibility window. Keep the chunk on the + // band; after a short streak, re-correlate the clock and bounce. + _noDurableTrimRefuseStreak++; + _log( + '[SYNC] HISTORY_END token=$tokenHex has no durable rows but the ' + 'plausibility gate dropped samples this burst — NOT ACKing ' + '(streak=$_noDurableTrimRefuseStreak). The band keeps the chunk; ' + 'a SET_CLOCK/reconnect may clear a poisoned gate window.', + ); + await _bestEffortLedgerWrite(() => LocalDb.upsertSyncLedgerEntry( + chunkId: 'batch:$tokenHex', + kind: 'historical_batch', + status: 'trim_refused', + lastError: 'no_durable_progress', + metaPatch: { + 'batch_id': batchId, + 'records': d.records, + 'no_durable_refuse_streak': _noDurableTrimRefuseStreak, + }, + )); + if (_noDurableTrimRefuseStreak >= 3 && !_sessionIsStale(session)) { + _log( + '[SYNC] $_noDurableTrimRefuseStreak consecutive no-durable trim ' + 'refuses — defensive SET_CLOCK + bounce so the next session can ' + 're-admit the re-delivered chunk.', + ); + _noDurableTrimRefuseStreak = 0; + try { + await setClock(); + } catch (e) { + _log('[SYNC] defensive SET_CLOCK after no-durable refuse failed: $e'); + } + if (!_sessionIsStale(session)) { + unawaited( + _teardownSession(intentional: false).then((_) { + _setPhase(BleConnState.idle); + }), + ); + } + } + return; } } @@ -2347,16 +2400,42 @@ class BleEngine { } else { _burstMismatchStreak = 0; } - _successfulBursts++; - _mergeValidatedBurst(d); final r = d.bufferedRecTsRange; + final droppedThisBurstForLog = droppedThisBurst; + final hadDurableRows = + d.bufferedRecords > 0 || d.bufferedArchives > 0; _log( '[SYNC] HistoryEnd batch=${m.batchId} records=${d.records} ' 'expected=${m.expectedPacketCount} actual=${d.currentBurstPacketCount} ' 'historical=${d.currentBurstHistoricalPacketCount} ' 'traffic=${d.currentBurstTrafficCount} token=$tokenHex ' + 'dropped_this_burst=$droppedThisBurstForLog ' + 'durable_buffered=${d.bufferedRecords}+${d.bufferedArchives} ' 'recTs=${r == null ? "none" : "${r.$1}..${r.$2}"}', ); + // NO-PROGRESS GATE: refuse trim when this burst banked nothing durable + // but the RecordGate dropped samples. Echoing would delete flash we + // never stored (and used to also flip lastTrimAdvanced, feeding + // auto-continue while the cursor stayed frozen). + final progressVerdict = TrimAckPolicy.evaluate( + sessionCurrent: !_sessionIsStale(session), + burstDiscarded: d.burstDiscarded, + commitDurable: true, + hadDurableRows: hadDurableRows, + droppedThisBurst: droppedThisBurst, + ); + if (progressVerdict != TrimAckVerdict.send) { + await _refuseHistoryEndTrim( + progressVerdict, + d: d, + session: session, + tokenHex: tokenHex, + batchId: m.batchId, + ); + return; + } + _successfulBursts++; + _mergeValidatedBurst(d); // SAFE-TRIM INVARIANT: persist decoded+raw AND the continuation cursor // DURABLY (one transaction) BEFORE the ACK. The band trims its flash only // once the ACK is link-layer confirmed, so a crash before the ACK @@ -2372,6 +2451,8 @@ class BleEngine { sessionCurrent: !_sessionIsStale(session), burstDiscarded: d.burstDiscarded, commitDurable: durable, + hadDurableRows: hadDurableRows, + droppedThisBurst: droppedThisBurst, ); if (verdict != TrimAckVerdict.send) { await _refuseHistoryEndTrim( @@ -2451,6 +2532,7 @@ class BleEngine { return; } _chunkFailures.recordSuccess(tokenHex); + _noDurableTrimRefuseStreak = 0; d.noteBatchAcked(); // ACKed and KEEP listening await _bestEffortLedgerWrite(() => LocalDb.upsertSyncLedgerEntry( status: 'acknowledged', @@ -3165,6 +3247,7 @@ class DrainController { bool _linkDown = false; int get bufferedRecords => _raws.length; + int get bufferedArchives => _archives.length; int get lastProgressMs => _lastProgressAt.millisecondsSinceEpoch; /// Min/max real record time (rec_ts) currently buffered for this batch — lets @@ -3189,7 +3272,10 @@ class DrainController { } // Trim-advance tracking for the stuck/continuation detectors: a HISTORY_END - // whose 8-byte token differs from the last one means the cursor moved. + // whose 8-byte token differs from the last one means the cursor moved — but + // only when the burst also banked durable rows. An empty token-only ACK + // (console / drop-only) used to flip this true and feed auto-continue while + // the durable frontier stayed frozen. String? _lastAckedToken; bool lastTrimAdvanced = false; int consecutiveValidationFailures = 0; @@ -3337,11 +3423,15 @@ class DrainController { .join(); final previousAckedToken = _lastAckedToken; final previousTrimAdvanced = lastTrimAdvanced; - lastTrimAdvanced = tokenHex != null && tokenHex != _lastAckedToken; - if (tokenHex != null) _lastAckedToken = tokenHex; final raws = List.from(_raws); final samples = List.from(_samples); final archives = List.from(_archives); + final hadDurable = raws.isNotEmpty || archives.isNotEmpty; + // Token changed AND we actually banked something — empty ACKs must not + // look like cursor progress to auto-continue / stuck-strap. + lastTrimAdvanced = + tokenHex != null && tokenHex != _lastAckedToken && hadDurable; + if (tokenHex != null) _lastAckedToken = tokenHex; _raws.clear(); _samples.clear(); _archives.clear(); diff --git a/lib/ble/ble_state.dart b/lib/ble/ble_state.dart index 039dc37..6099c43 100644 --- a/lib/ble/ble_state.dart +++ b/lib/ble/ble_state.dart @@ -187,7 +187,9 @@ class DrainStopEvaluator { /// - plausibility gate: reject records whose embedded unix time is /// implausible vs wall-clock / the strap's own GET_DATA_RANGE window /// (a previous owner's wandering-clock pollution). Rejected records are -/// neither stored nor counted; the batch ACK still walks the cursor. +/// neither stored nor counted. A burst that banks at least one durable +/// row may still ACK (cursor walks past mixed pollution); a drop-only +/// empty burst must NOT ACK — see [TrimAckVerdict.blockedNoDurableProgress]. /// - frontier: track the highest plausible rec_ts admitted so far — the /// durable high-water the StuckStrapDetector / BackfillContinuation read. /// - drop counter: how many records the gate rejected (diagnostics). @@ -319,6 +321,15 @@ enum TrimAckVerdict { /// The durable commit did NOT complete. The records are not in the database, /// so the band must keep them. blockedCommitFailed, + + /// This burst's HISTORY_END would trim flash we neither decoded nor archived: + /// the plausibility gate rejected every historical record (`droppedThisBurst > + /// 0`) and the buffer is empty. Echoing the token used to "walk the cursor" + /// past wandering-clock pollution, but after a reconnect / bad session window + /// the same path permanently deleted good samples (HR frozen until a manual + /// reconnect). Refuse the trim so the band re-delivers; the engine should + /// re-correlate the clock and retry. + blockedNoDurableProgress, } /// THE gate on the one irreversible act in the whole offload protocol: echoing @@ -334,7 +345,7 @@ enum TrimAckVerdict { /// and the band was then told to trim — those records existed nowhere, /// permanently and silently. /// -/// Pure and total: the engine supplies three observations, this decides. The +/// Pure and total: the engine supplies observations, this decides. The /// engine must call it AGAIN after the commit await (the commit can take /// seconds on a large batch — long enough for the session to die under it). class TrimAckPolicy { @@ -346,10 +357,18 @@ class TrimAckPolicy { /// [commitDurable] — the atomic commit completed (pass `true` when asking /// the pre-commit question "should I even commit this /// token?"). + /// [hadDurableRows] — this burst buffered at least one raw/sample or archive + /// row to bank before ACK. Pass `true` when unknown + /// (pre-commit stale/discard checks only). + /// [droppedThisBurst] — RecordGate rejects during this burst. Combined with + /// `!hadDurableRows`, refuses trim so gate-only bursts + /// cannot delete flash we never stored. static TrimAckVerdict evaluate({ required bool sessionCurrent, required bool burstDiscarded, required bool commitDurable, + bool hadDurableRows = true, + int droppedThisBurst = 0, }) { // Order is deliberate: a stale session must be refused before anything // else touches the (new) link, and a poisoned burst must be refused before @@ -357,6 +376,9 @@ class TrimAckPolicy { if (!sessionCurrent) return TrimAckVerdict.blockedStaleSession; if (burstDiscarded) return TrimAckVerdict.blockedDiscardedBurst; if (!commitDurable) return TrimAckVerdict.blockedCommitFailed; + if (!hadDurableRows && droppedThisBurst > 0) { + return TrimAckVerdict.blockedNoDurableProgress; + } return TrimAckVerdict.send; } } diff --git a/test/ble_safe_trim_test.dart b/test/ble_safe_trim_test.dart index 65000cc..ce07601 100644 --- a/test/ble_safe_trim_test.dart +++ b/test/ble_safe_trim_test.dart @@ -153,6 +153,9 @@ void main() { log: (_) {}, ); + // Empty token-only commits do not count as trim advance (would feed + // auto-continue while the durable frontier stayed frozen). + d.onHistoricalRecord(_raw(0), _sample(0)); expect(await d.commit(_tokenA), isTrue); expect(d.lastTrimAdvanced, isTrue); @@ -171,6 +174,19 @@ void main() { expect(d.lastTrimAdvanced, isTrue); }); + test('an empty buffer commit does not claim trim advanced', () async { + final d = _drainWith((raws, samples, token, {archives}) async {}); + expect(await d.commit(_tokenA), isTrue); + expect(d.lastTrimAdvanced, isFalse); + }); + + test('archive-only commit still counts as trim advanced', () async { + final d = _drainWith((raws, samples, token, {archives}) async {}); + d.onUndecodableRecord(_archive(1)); + expect(await d.commit(_tokenA), isTrue); + expect(d.lastTrimAdvanced, isTrue); + }); + test('a successful commit clears the buffer and reports durable', () async { final d = _drainWith((raws, samples, token, {archives}) async {}); d.onHistoricalRecord(_raw(1), _sample(1)); @@ -247,6 +263,58 @@ void main() { TrimAckVerdict.send, ); }); + + test('drop-only empty burst blocks the ACK (no durable progress)', () { + expect( + TrimAckPolicy.evaluate( + sessionCurrent: true, + burstDiscarded: false, + commitDurable: true, + hadDurableRows: false, + droppedThisBurst: 12, + ), + TrimAckVerdict.blockedNoDurableProgress, + ); + }); + + test('empty burst with zero drops may still ACK (console-only)', () { + expect( + TrimAckPolicy.evaluate( + sessionCurrent: true, + burstDiscarded: false, + commitDurable: true, + hadDurableRows: false, + droppedThisBurst: 0, + ), + TrimAckVerdict.send, + ); + }); + + test('archive-or-raw banked rows still ACK even when some were dropped', () { + expect( + TrimAckPolicy.evaluate( + sessionCurrent: true, + burstDiscarded: false, + commitDurable: true, + hadDurableRows: true, + droppedThisBurst: 5, + ), + TrimAckVerdict.send, + ); + }); + + test('commit failure outranks the no-durable-progress rule', () { + expect( + TrimAckPolicy.evaluate( + sessionCurrent: true, + burstDiscarded: false, + commitDurable: false, + hadDurableRows: false, + droppedThisBurst: 9, + ), + TrimAckVerdict.blockedCommitFailed, + ); + }); }); group('P0 — a discarded burst poisons its HISTORY_END token', () { From 95d7c243ab4b8394f0deabdccd37f426cb2d0377 Mon Sep 17 00:00:00 2001 From: abdulsaheel Date: Fri, 7 Aug 2026 00:42:52 +0530 Subject: [PATCH 2/3] sync: let the no-durable-progress refusal loop escape, and stop it being silent Refusing a HISTORY_END whose burst banked nothing is right, and this PR gets that part right -- echoing the token lets the band delete flash we never stored, the one irreversible act in this protocol. The problem is everything AFTER the refusal. PARTIALLY REFUTING the review note first: this branch DOES already have a refusal counter and a SET_CLOCK+bounce remedy at >= 3. The claim of "no counter, no eventual path" is out of date. What is real is that neither could actually fire. 1. THE COUNTER COULD NEVER CLIMB. `_noDurableTrimRefuseStreak` was zeroed in the per-connection setup block -- but the streak's own remedy is a SET_CLOCK + BOUNCE, i.e. a reconnect, and the idle watchdog also tears the session down after a burst that banked nothing. Every cycle reset the count, so the ">= 3 then run the remedy" branch was unreachable in the field and the loop ran forever at streak 1. The engine already documents this exact distinction one block up: "marginal-radio + post-bond-loop are NOT reset here -- they count consecutive bad cycles across reconnects". This is the same class of counter and belongs on the same side of that line. 2. A REMEDY THAT KEEPS FAILING WAS INVISIBLE. `EmptySyncTracker` (which sets `syncClockLost`) and `StuckStrapDetector` are BOTH only evaluated inside `_onOffloadFinished`, and that needs a HISTORY_COMPLETE this loop never reaches, because the band keeps re-delivering the same un-trimmed chunk. So a band with a lost RTC -- every record below the plausibility floor -- sat in refuse -> bounce -> refuse indefinitely behind a debug log, with no user signal at all. Both now live in `NoDurableProgressEscalation` (lib/sync/sync_policy.dart), pure and unit-testable, alongside BondRefusalGiveUp which it deliberately mirrors -- the engine wires, the policy decides. Two thresholds, because there are two questions: N consecutive refusals => try the remedy; M remedies that did NOT work => surface `syncClockLost`. Counting spans reconnects and clears only on a successful trim ACK, which is the only thing that proves the condition is actually over. It still NEVER ACKs. "Keep the data" stays the answer -- what changes is that a persistently failing remedy becomes visible instead of silent. 7 tests, mutation-verified. Suite 1208 passing; the 6 failures in notification_dedupe_test are pre-existing and reproduce on origin/main unmodified. --- lib/ble/ble_engine.dart | 54 ++++++-- lib/sync/sync_policy.dart | 62 ++++++++++ test/no_durable_progress_escalation_test.dart | 116 ++++++++++++++++++ 3 files changed, 221 insertions(+), 11 deletions(-) create mode 100644 test/no_durable_progress_escalation_test.dart diff --git a/lib/ble/ble_engine.dart b/lib/ble/ble_engine.dart index a59131f..2fe5c7d 100644 --- a/lib/ble/ble_engine.dart +++ b/lib/ble/ble_engine.dart @@ -699,7 +699,14 @@ class BleEngine { // SET_CLOCK and bounce — the usual cause is a bad post-reconnect window // that would otherwise loop empty drop-ACKs (or, after the refuse guard, // re-deliver forever without ever correcting the clock). - int _noDurableTrimRefuseStreak = 0; + // + // NOT reset per connection (unlike _emptySync / _stuckStrap). The remedy IS + // a reconnect, so a per-connection reset made the escalation unreachable: + // the idle watchdog tears the session down after a refuse-only burst, the + // next connect zeroed the count, and the loop ran forever at streak 1. + // See NoDurableProgressEscalation for the full derivation. + final NoDurableProgressEscalation _noDurableProgress = + NoDurableProgressEscalation(); // Band-truth reconciliation: `expectedPacketCount` mismatches are advisory // (see the comment at the validation site — treating a single mismatch as // fatal was actively harmful and was reverted), but a mismatch that keeps @@ -1164,7 +1171,11 @@ class BleEngine { _autoContinue.end(); _lastBackfillAt = 0; _successfulBursts = 0; - _noDurableTrimRefuseStreak = 0; + // _noDurableProgress is deliberately NOT reset here — like _marginalRadio + // and _postBondLoop above, it counts consecutive bad cycles ACROSS + // reconnects. Resetting per connection made the escalation unreachable, + // because the escalation's own remedy is a reconnect. It clears on a + // successful trim ACK, the only thing that proves the condition is over. _lastHpsTerminal = null; _sessionPacketCounts = _SessionPacketCounts.zero; _sessionGapSummary = _SessionGapSummary.zero; @@ -2247,11 +2258,12 @@ class BleEngine { // we never stored — the HR-gap / frozen-cursor failure mode after a // reconnect with a bad plausibility window. Keep the chunk on the // band; after a short streak, re-correlate the clock and bounce. - _noDurableTrimRefuseStreak++; + final runRemedy = _noDurableProgress.trimRefused(); _log( '[SYNC] HISTORY_END token=$tokenHex has no durable rows but the ' 'plausibility gate dropped samples this burst — NOT ACKing ' - '(streak=$_noDurableTrimRefuseStreak). The band keeps the chunk; ' + '(refusals=${_noDurableProgress.refusals} ' + 'remedies=${_noDurableProgress.remedies}). The band keeps the chunk; ' 'a SET_CLOCK/reconnect may clear a poisoned gate window.', ); await _bestEffortLedgerWrite(() => LocalDb.upsertSyncLedgerEntry( @@ -2262,16 +2274,34 @@ class BleEngine { metaPatch: { 'batch_id': batchId, 'records': d.records, - 'no_durable_refuse_streak': _noDurableTrimRefuseStreak, + 'no_durable_refuse_streak': _noDurableProgress.refusals, + 'no_durable_remedy_cycles': _noDurableProgress.remedies, }, )); - if (_noDurableTrimRefuseStreak >= 3 && !_sessionIsStale(session)) { + if (runRemedy && !_sessionIsStale(session)) { _log( - '[SYNC] $_noDurableTrimRefuseStreak consecutive no-durable trim ' - 'refuses — defensive SET_CLOCK + bounce so the next session can ' - 're-admit the re-delivered chunk.', + '[SYNC] no-durable trim refuses hit the remedy threshold — ' + 'defensive SET_CLOCK + bounce so the next session can re-admit ' + 'the re-delivered chunk.', ); - _noDurableTrimRefuseStreak = 0; + if (_noDurableProgress.shouldSurfaceGiveUp()) { + // The remedy has now failed repeatedly. We still do NOT ACK — + // trimming flash we never banked is unrecoverable, so "keep the + // data" stays the right answer. What changes is that this stops + // being INVISIBLE: previously it refused, bounced and retried + // forever behind a debug log, because the two detectors that would + // otherwise catch it (`EmptySyncTracker` -> `syncClockLost`, + // `StuckStrapDetector`) are only evaluated in + // `_onOffloadFinished`, and HISTORY_COMPLETE never arrives here. + state.syncClockLost = true; + onState(state); + _log( + '[SYNC] ${_noDurableProgress.remedies} SET_CLOCK+bounce remedies ' + 'have not cleared the no-durable-progress loop — surfacing ' + 'syncClockLost. The chunk is still SAFE on the band; we are ' + 'refusing the trim, not losing data.', + ); + } try { await setClock(); } catch (e) { @@ -2532,7 +2562,9 @@ class BleEngine { return; } _chunkFailures.recordSuccess(tokenHex); - _noDurableTrimRefuseStreak = 0; + // A trim ACK is the only real proof the no-durable-progress condition is + // over — records were banked and the band may advance. + _noDurableProgress.trimAcked(); d.noteBatchAcked(); // ACKed and KEEP listening await _bestEffortLedgerWrite(() => LocalDb.upsertSyncLedgerEntry( status: 'acknowledged', diff --git a/lib/sync/sync_policy.dart b/lib/sync/sync_policy.dart index 1bd6ff0..abda54a 100644 --- a/lib/sync/sync_policy.dart +++ b/lib/sync/sync_policy.dart @@ -617,3 +617,65 @@ bool shouldRenotifyStaleness( if (lastNotifiedAt == null) return true; return now.difference(lastNotifiedAt) >= renotifyAfter; } + +/// Escalation for a `TrimAckVerdict.blockedNoDurableProgress` loop. +/// +/// Refusing the trim is ALWAYS the right call: echoing the token would let the +/// band delete flash we never banked, the one irreversible act in this +/// protocol. But refusing forever means sync silently stops making progress, +/// and nothing surfaced it — `EmptySyncTracker` (→ `syncClockLost`) and +/// `StuckStrapDetector` are both only evaluated in `_onOffloadFinished`, which +/// needs a HISTORY_COMPLETE that this loop never reaches, because the band +/// keeps re-delivering the same un-trimmed chunk. +/// +/// Two thresholds, because there are two different questions: +/// * [remedyAfter] refusals in a row ⇒ try the remedy (SET_CLOCK + bounce). +/// The usual cause is a poisoned post-reconnect plausibility window. +/// * [giveUpAfter] remedies that did NOT work ⇒ stop assuming the next +/// SET_CLOCK will fix it and tell the user. Still never ACKs. +/// +/// Counting must span RECONNECTS. The remedy IS a reconnect, so a +/// per-connection reset makes the escalation unreachable — the session is torn +/// down at streak 1 and the next connect zeroes it. Only a successful trim ACK +/// proves the condition is over ([trimAcked]). +class NoDurableProgressEscalation { + final int remedyAfter; + final int giveUpAfter; + + NoDurableProgressEscalation({this.remedyAfter = 3, this.giveUpAfter = 3}); + + int _refusals = 0; + int _remedies = 0; + bool _gaveUp = false; + + int get refusals => _refusals; + int get remedies => _remedies; + bool get gaveUp => _gaveUp; + + /// Feed one refused HISTORY_END. Returns true when the caller should run the + /// remedy (and resets the refusal streak so the next run needs a fresh one). + bool trimRefused() { + _refusals++; + if (_refusals < remedyAfter) return false; + _refusals = 0; + _remedies++; + return true; + } + + /// True EXACTLY ONCE, on the remedy that crosses [giveUpAfter] — the caller + /// then surfaces `syncClockLost`. Latched so it can't re-notify every cycle. + bool shouldSurfaceGiveUp() { + if (_gaveUp || _remedies < giveUpAfter) return false; + _gaveUp = true; + return true; + } + + /// A trim ACK: records were banked and the band may advance. The only real + /// proof the condition is over — clears the streak, the remedy count AND the + /// give-up latch, so a later run can escalate again. + void trimAcked() { + _refusals = 0; + _remedies = 0; + _gaveUp = false; + } +} diff --git a/test/no_durable_progress_escalation_test.dart b/test/no_durable_progress_escalation_test.dart new file mode 100644 index 0000000..f07b2dc --- /dev/null +++ b/test/no_durable_progress_escalation_test.dart @@ -0,0 +1,116 @@ +// P0 REGRESSION — the no-durable-progress refusal loop must be able to escape, +// and must never be silent. +// +// Refusing a HISTORY_END whose burst banked nothing is correct: echoing the +// token lets the band delete flash we never stored, the one irreversible act +// in this protocol. The bug was everything AFTER the refusal. +// +// 1. The refusal streak lived in the engine and was zeroed in the +// per-connection setup block. But the streak's own remedy is a +// SET_CLOCK + BOUNCE — a reconnect — and the idle watchdog also tears the +// session down after a burst that banked nothing. So the count could never +// climb: every cycle reset it, and the ">= 3 then run the remedy" branch +// was unreachable in the field. The engine already documents this exact +// distinction one block up ("marginal-radio + post-bond-loop are NOT reset +// here — they count consecutive bad cycles across reconnects"). +// +// 2. Nothing surfaced a remedy that kept failing. `EmptySyncTracker` (which +// sets `syncClockLost`) and `StuckStrapDetector` are BOTH only evaluated +// inside `_onOffloadFinished`, and that needs a HISTORY_COMPLETE which this +// loop never reaches — the band keeps re-delivering the same un-trimmed +// chunk. A band with a lost RTC could therefore sit in refuse → bounce → +// refuse forever behind a debug log. +// +// The escalation still NEVER ACKs. "Keep the data" stays the answer; what +// changes is that a persistently failing remedy becomes visible. + +import 'package:flutter_test/flutter_test.dart'; +import 'package:openstrap_edge/sync/sync_policy.dart'; + +void main() { + test('a refusal streak below the threshold does not run the remedy', () { + final e = NoDurableProgressEscalation(remedyAfter: 3); + expect(e.trimRefused(), isFalse); + expect(e.trimRefused(), isFalse); + expect(e.remedies, 0); + }); + + test('the remedy fires on the Nth consecutive refusal', () { + final e = NoDurableProgressEscalation(remedyAfter: 3); + e.trimRefused(); + e.trimRefused(); + expect(e.trimRefused(), isTrue); + expect(e.remedies, 1); + expect(e.refusals, 0, reason: 'next remedy needs a fresh streak'); + }); + + test( + 'counting SURVIVES the remedy — this is the whole bug. The remedy is a ' + 'reconnect, so a per-connection reset made escalation unreachable.', + () { + final e = NoDurableProgressEscalation(remedyAfter: 3, giveUpAfter: 3); + // Three full cycles, each ending in the reconnect that used to zero it. + for (var cycle = 0; cycle < 3; cycle++) { + expect(e.trimRefused(), isFalse); + expect(e.trimRefused(), isFalse); + expect(e.trimRefused(), isTrue, reason: 'cycle $cycle remedy'); + } + expect(e.remedies, 3); + expect( + e.shouldSurfaceGiveUp(), + isTrue, + reason: 'three failed remedies must become user-visible', + ); + }, + ); + + test('give-up is latched — it must not re-notify on every later cycle', () { + final e = NoDurableProgressEscalation(remedyAfter: 1, giveUpAfter: 2); + e.trimRefused(); + e.trimRefused(); + expect(e.shouldSurfaceGiveUp(), isTrue); + expect(e.shouldSurfaceGiveUp(), isFalse); + e.trimRefused(); + expect(e.shouldSurfaceGiveUp(), isFalse); + expect(e.gaveUp, isTrue); + }); + + test('give-up does not fire before the remedy has actually failed', () { + final e = NoDurableProgressEscalation(remedyAfter: 3, giveUpAfter: 3); + e.trimRefused(); + e.trimRefused(); + e.trimRefused(); // 1 remedy + expect(e.shouldSurfaceGiveUp(), isFalse); + }); + + test( + 'a successful trim ACK clears everything, including the give-up latch', + () { + final e = NoDurableProgressEscalation(remedyAfter: 1, giveUpAfter: 1); + e.trimRefused(); + expect(e.shouldSurfaceGiveUp(), isTrue); + + e.trimAcked(); + expect(e.refusals, 0); + expect(e.remedies, 0); + expect(e.gaveUp, isFalse); + + // A later run must be able to escalate again. + expect(e.trimRefused(), isTrue); + expect(e.shouldSurfaceGiveUp(), isTrue); + }, + ); + + test('an interleaved ACK breaks the streak — only CONSECUTIVE refusals count', + () { + final e = NoDurableProgressEscalation(remedyAfter: 3); + e.trimRefused(); + e.trimRefused(); + e.trimAcked(); // real progress happened + expect( + e.trimRefused(), + isFalse, + reason: 'the streak restarted; two old refusals must not carry over', + ); + }); +} From e32a2f94421a48306cac545df654d85b55bef711 Mon Sep 17 00:00:00 2001 From: Brackyt <60280126+Brackyt@users.noreply.github.com> Date: Fri, 7 Aug 2026 00:01:49 +0200 Subject: [PATCH 3/3] fix(sync): close DrainController safe-trim holes for archive-only / unbuffered drains Reject onRecordsBatch without onCommit at construction, buffer only when onCommit is wired, and refuse HISTORY_END when supportsSafeTrim is false so archive-only or fire-and-forget paths cannot ACK flash that was never banked. --- lib/ble/ble_engine.dart | 50 ++++++++++++++++++++++---- test/ble_safe_trim_test.dart | 70 ++++++++++++++++++++++++++++++++++++ 2 files changed, 114 insertions(+), 6 deletions(-) diff --git a/lib/ble/ble_engine.dart b/lib/ble/ble_engine.dart index 2fe5c7d..ccaa775 100644 --- a/lib/ble/ble_engine.dart +++ b/lib/ble/ble_engine.dart @@ -2443,6 +2443,15 @@ class BleEngine { 'durable_buffered=${d.bufferedRecords}+${d.bufferedArchives} ' 'recTs=${r == null ? "none" : "${r.$1}..${r.$2}"}', ); + // Non-trimmable wiring (no onCommit): unbuffered fire-and-forget cannot + // prove durability before ACK. Production always sets onCommitBatch. + if (!d.supportsSafeTrim) { + _log( + '[SYNC] HISTORY_END token=$tokenHex refused — drain has no onCommit ' + '(non-trimmable / test-only wiring); band keeps the chunk.', + ); + return; + } // NO-PROGRESS GATE: refuse trim when this burst banked nothing durable // but the RecordGate dropped samples. Echoing would delete flash we // never stored (and used to also flip lastTrimAdvanced, feeding @@ -3259,7 +3268,26 @@ class DrainController { required this.onCommit, required this.onArchive, required this.log, - }); + }) { + // Safe-trim requires the atomic commit sink: onRecordsBatch alone cannot + // persist archives or the trim cursor. Production always wires onCommit; + // forbid the latent hole where archive-only chunks would ACK and trim + // flash that was never stored. + if (onRecordsBatch != null && onCommit == null) { + throw ArgumentError( + 'DrainController: onRecordsBatch without onCommit cannot persist ' + 'archives or the trim cursor — buffered historical drains require ' + 'onCommit', + ); + } + } + + /// Whether HISTORY_END may be ACKed under the safe-trim invariant. + /// + /// Only [onCommit] can bank raws + archives + cursor in one transaction. + /// Both-sinks-null (unbuffered / test-only) fire-and-forgets via [onRecord] + /// and must not trim. + bool get supportsSafeTrim => onCommit != null; final List _raws = []; final List _samples = []; @@ -3323,7 +3351,9 @@ class DrainController { /// Bursts poisoned this connection (diagnostics). int get poisonedBursts => _trimGuard.poisonedBursts; - bool get _buffering => onCommit != null || onRecordsBatch != null; + /// Buffer only when the atomic commit path exists. Unbuffered mode + /// (test-only) must not look like it banked durable rows for trim. + bool get _buffering => onCommit != null; int get currentBurstPacketCount => burstStats.totalTrafficPacketCount; int get currentBurstTrafficCount => burstStats.totalTrafficPacketCount; int get currentBurstHistoricalPacketCount => burstStats.historicalPacketCount; @@ -3468,10 +3498,18 @@ class DrainController { _samples.clear(); _archives.clear(); try { - if (onCommit != null) { - await onCommit!(raws, samples, tokenHex, archives: archives); - } else if (onRecordsBatch != null && raws.isNotEmpty) { - await onRecordsBatch!(raws, samples); + // Defense in depth (constructor already rejects onRecordsBatch-only): + // never report durable success for buffered content without onCommit. + if (raws.isNotEmpty || archives.isNotEmpty || tokenHex != null) { + final commit = onCommit; + if (commit == null) { + throw StateError( + 'DrainController.commit requires onCommit to persist buffered ' + 'rows / archives / trim cursor (had raws=${raws.length}, ' + 'archives=${archives.length}, token=${tokenHex != null})', + ); + } + await commit(raws, samples, tokenHex, archives: archives); } return true; } catch (e) { diff --git a/test/ble_safe_trim_test.dart b/test/ble_safe_trim_test.dart index ce07601..a56c1bd 100644 --- a/test/ble_safe_trim_test.dart +++ b/test/ble_safe_trim_test.dart @@ -317,6 +317,76 @@ void main() { }); }); + group('P0 — DrainController wiring must not allow latent safe-trim holes', () { + test('onRecordsBatch without onCommit is rejected at construction', () { + expect( + () => DrainController( + onRecord: (sample, raw) async {}, + onRecordsBatch: (raws, samples) async {}, + onCommit: null, + onArchive: null, + log: (_) {}, + ), + throwsA(isA()), + ); + }); + + test('supportsSafeTrim is true only when onCommit is wired', () { + final withCommit = _drainWith((raws, samples, token, {archives}) async {}); + expect(withCommit.supportsSafeTrim, isTrue); + + final unbuffered = DrainController( + onRecord: (sample, raw) async {}, + onRecordsBatch: null, + onCommit: null, + onArchive: null, + log: (_) {}, + ); + expect(unbuffered.supportsSafeTrim, isFalse); + }); + + test('token commit without onCommit fails closed (no false durable)', () async { + final d = DrainController( + onRecord: (sample, raw) async {}, + onRecordsBatch: null, + onCommit: null, + onArchive: null, + log: (_) {}, + ); + expect(await d.commit(_tokenA), isFalse); + expect(d.lastTrimAdvanced, isFalse); + }); + + test('archive-only + onCommit still persists before success', () async { + final seen = []; + final ok = _drainWith((raws, samples, token, {archives}) async { + seen.addAll((archives ?? const []).map((a) => a.hex)); + }); + ok.onUndecodableRecord(_archive(9)); + expect(await ok.commit(_tokenA), isTrue); + expect(seen, [_hex(9)]); + expect(ok.bufferedArchives, 0); + expect(ok.lastTrimAdvanced, isTrue); + }); + + test('unbuffered mode does not treat fire-and-forget rows as buffered', () { + var wrote = 0; + final d = DrainController( + onRecord: (sample, raw) async { + wrote++; + }, + onRecordsBatch: null, + onCommit: null, + onArchive: null, + log: (_) {}, + ); + d.onHistoricalRecord(_raw(1), _sample(1)); + expect(d.bufferedRecords, 0); + expect(d.supportsSafeTrim, isFalse); + expect(wrote, 1); + }); + }); + group('P0 — a discarded burst poisons its HISTORY_END token', () { test('discardOpenChunk marks the open burst un-ACKable', () async { final d = _drainWith((raws, samples, token, {archives}) async {});