diff --git a/lib/ble/ble_engine.dart b/lib/ble/ble_engine.dart index d21d41d..ccaa775 100644 --- a/lib/ble/ble_engine.dart +++ b/lib/ble/ble_engine.dart @@ -694,6 +694,19 @@ 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). + // + // 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 @@ -1158,6 +1171,11 @@ class BleEngine { _autoContinue.end(); _lastBackfillAt = 0; _successfulBursts = 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; @@ -1893,8 +1911,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 +2252,70 @@ 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. + final runRemedy = _noDurableProgress.trimRefused(); + _log( + '[SYNC] HISTORY_END token=$tokenHex has no durable rows but the ' + 'plausibility gate dropped samples this burst — NOT ACKing ' + '(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( + chunkId: 'batch:$tokenHex', + kind: 'historical_batch', + status: 'trim_refused', + lastError: 'no_durable_progress', + metaPatch: { + 'batch_id': batchId, + 'records': d.records, + 'no_durable_refuse_streak': _noDurableProgress.refusals, + 'no_durable_remedy_cycles': _noDurableProgress.remedies, + }, + )); + if (runRemedy && !_sessionIsStale(session)) { + _log( + '[SYNC] no-durable trim refuses hit the remedy threshold — ' + 'defensive SET_CLOCK + bounce so the next session can re-admit ' + 'the re-delivered chunk.', + ); + 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) { + _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 +2430,51 @@ 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}"}', ); + // 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 + // 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 +2490,8 @@ class BleEngine { sessionCurrent: !_sessionIsStale(session), burstDiscarded: d.burstDiscarded, commitDurable: durable, + hadDurableRows: hadDurableRows, + droppedThisBurst: droppedThisBurst, ); if (verdict != TrimAckVerdict.send) { await _refuseHistoryEndTrim( @@ -2451,6 +2571,9 @@ class BleEngine { return; } _chunkFailures.recordSuccess(tokenHex); + // 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', @@ -3145,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 = []; @@ -3165,6 +3307,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 +3332,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; @@ -3205,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; @@ -3337,19 +3485,31 @@ 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(); 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/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/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/ble_safe_trim_test.dart b/test/ble_safe_trim_test.dart index 65000cc..a56c1bd 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,128 @@ 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 — 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', () { 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', + ); + }); +}