From d31f88e97d17011e61bc93def3ca3c261890be25 Mon Sep 17 00:00:00 2001 From: Developer Date: Thu, 10 Sep 2026 14:31:33 -0400 Subject: [PATCH] fix: make telemetry flush completion dispose-safe --- lib/features/rides/ride_state_machine.dart | 107 +++- test/rid3b2_flush_dispose_safety_test.dart | 659 +++++++++++++++++++++ 2 files changed, 745 insertions(+), 21 deletions(-) create mode 100644 test/rid3b2_flush_dispose_safety_test.dart diff --git a/lib/features/rides/ride_state_machine.dart b/lib/features/rides/ride_state_machine.dart index 9019c30..2f568d9 100644 --- a/lib/features/rides/ride_state_machine.dart +++ b/lib/features/rides/ride_state_machine.dart @@ -1,6 +1,7 @@ import 'dart:async'; import 'package:wakelock_plus/wakelock_plus.dart'; import 'package:flutter_riverpod/flutter_riverpod.dart'; +import 'package:flutter/foundation.dart' show debugPrint; import '../../database/database.dart' hide TelemetrySample; import 'package:drift/drift.dart' as drift; import 'package:geolocator/geolocator.dart'; @@ -172,7 +173,7 @@ class RideStateMachine extends StateNotifier { _telemetrySub = telemetryService.telemetryStream.listen(_onTelemetryUpdate); // Write to DB every 3 seconds for tight crash-durability - _batchWriteTimer = Timer.periodic(const Duration(seconds: 3), (_) => _flushBatch()); + _batchWriteTimer = Timer.periodic(const Duration(seconds: 3), (_) => _runScheduledFlush()); // Watchdog timer (every 2 seconds) checks if telemetry pipeline is alive _watchdogTimer = Timer.periodic(const Duration(seconds: 2), (_) => _checkWatchdog()); @@ -684,27 +685,61 @@ class RideStateMachine extends StateNotifier { final completion = Completer(); _activeTelemetryFlush = completion.future; try { - await db.batch((batch) { - batch.insertAll(db.telemetrySamples, batchToInsert); - }); - // Success: account only for the owned snapshot that is now durable. - state = state.copyWith(totalSamplesPersisted: state.totalSamplesPersisted + batchToInsert.length); - return true; - } catch (e) { - // Persistence failed, so the owned snapshot is not durable. Restore it to - // the FRONT of the live buffer, preserving both the snapshot's internal - // order and the order of samples that arrived during the await, so a later - // legitimate flush can retry them. No automatic retry worker is added. - _telemetryBatch.insertAll(0, batchToInsert); - // Keep in queue for retry on next flush interval + // ---- PRE-COMMIT PHASE --------------------------------------------------- + // This is the ONLY block that may fail persistence. RID-3A ownership, + // ordering and front-restoration semantics are unchanged. try { - ref.read(fieldTestLoggerProvider).log( - eventType: FieldTestEventType.databaseQueueWarning, - message: 'Database batch write retry: $e', - severity: EventSeverity.warning, - ); - } catch (_) {} - return false; + await db.batch((batch) { + batch.insertAll(db.telemetrySamples, batchToInsert); + }); + } catch (e) { + // Persistence failed BEFORE the commit point, so the owned snapshot is + // not durable. Restore it to the FRONT of the live buffer, preserving + // both the snapshot's internal order and the order of samples that + // arrived during the await, so a later legitimate flush can retry them. + // No automatic retry worker is added. + _telemetryBatch.insertAll(0, batchToInsert); + // Keep in queue for retry on next flush interval. A notifier that has + // begun disposal must not touch ref, so the diagnostic is skipped in that + // case; it remains bounded either way. + if (mounted) { + try { + ref.read(fieldTestLoggerProvider).log( + eventType: FieldTestEventType.databaseQueueWarning, + message: 'Database batch write retry: $e', + severity: EventSeverity.warning, + ); + } catch (_) {} + } + return false; + } + + // ---- COMMIT POINT REACHED ----------------------------------------------- + // The successful return of db.batch(...) above is the telemetry durability + // commit point. The owned snapshot is now durable: nothing below may restore + // it to the live buffer or change the durability verdict. + + // Post-commit bookkeeping is notifier-owned, so it is conditional on the + // notifier still being alive and isolated in its own guard. `mounted` is + // StateNotifier's real property, so this boundary is identical in debug and + // release - correctness never depends on a debug assertion firing. After + // disposal there is no state read, no state assignment and no ref access. + // + // Accepted RID-3B-2 counter contract: when disposal races a committed flush, + // the durable row count may exceed the dead notifier's final in-memory + // totalSamplesPersisted value. That is a terminal counter snapshot on a + // disposed notifier, NOT a persistence failure, and no live consumer may + // rely on it. Committed samples are never restored to make the counter look + // consistent, and no database counter or schema field is added. + if (mounted) { + try { + state = state.copyWith(totalSamplesPersisted: state.totalSamplesPersisted + batchToInsert.length); + } catch (_) { + // Post-commit bookkeeping failure must never be classified as a + // persistence failure and must never restore the committed snapshot. + } + } + return true; } finally { _isFlushingTelemetryBatch = false; if (!completion.isCompleted) completion.complete(); @@ -712,6 +747,28 @@ class RideStateMachine extends StateNotifier { } } + /// RID-3B-2 timer Future ownership. + /// + /// A Timer.periodic callback is synchronous, so the Future returned by + /// _flushBatch() would otherwise be discarded and any rejection would surface + /// as an uncaught zone error attributed to the timer callback. The Future is + /// explicitly owned here instead. + /// + /// This is a defensive ownership boundary only: no second scheduler, no new + /// timer, and no change to the 3-second crash-durability cadence. The timer + /// callback itself stays synchronous, so cancellation semantics are unchanged. + /// + /// Diagnostics are bounded to runtimeType only. The exception object is never + /// logged because its message may carry implementation or database detail, and + /// no stack trace or telemetry payload is recorded. + void _runScheduledFlush() { + unawaited( + _flushBatch().catchError((Object error, StackTrace stackTrace) { + debugPrint('Scheduled telemetry flush failed: ${error.runtimeType}'); + }), + ); + } + Future _flushBatch() async { // RID-3A telemetry batch ownership (exclusive snapshot, clear-before-await, // front-restoration on failure, success-only counter) now lives in @@ -738,6 +795,14 @@ class RideStateMachine extends StateNotifier { // this guard so its existing every-tick cadence is unchanged. await _persistOwnedTelemetrySnapshot(); + // RID-3B-2: the notifier may have been disposed while this telemetry flush + // was in flight, because Timer.cancel() cannot abort an already-started async + // callback. The database commit above is already authoritative and is not + // revisited; everything below is notifier-owned, so return without reading + // state or ref. Incremental stats are never fabricated after disposal, and + // stats persistence is never moved ahead of the telemetry commit. + if (!mounted) return; + // Incrementally persist ongoing ride summary stats to SQLite so crashes/terminations never lose stats if (state.dbId != null && state.state == RideState.riding) { try { diff --git a/test/rid3b2_flush_dispose_safety_test.dart b/test/rid3b2_flush_dispose_safety_test.dart new file mode 100644 index 0000000..e119ec9 --- /dev/null +++ b/test/rid3b2_flush_dispose_safety_test.dart @@ -0,0 +1,659 @@ +// RID-3B-2 permanent regression suite: timer Future ownership, post-dispose-safe +// flush completion, and the telemetry persistence commit-point contract. +// +// Concerns covered: +// A. telemetry persistence commit is separated from notifier bookkeeping +// B. an in-flight flush is safe when the notifier becomes disposed +// C. the Timer.periodic flush Future is explicitly owned and contained +// +// Concern D (graceful buffered disposal) is DELIBERATELY DEFERRED and is locked +// as a documented residual by RID3B2-T7. That test is not an endorsement of data +// loss; it exists so this patch cannot silently pretend concern D was solved. +import 'dart:async'; + +import 'package:flutter_test/flutter_test.dart'; +import 'package:flutter_riverpod/flutter_riverpod.dart'; +import 'package:geolocator/geolocator.dart'; +import 'package:shared_preferences/shared_preferences.dart'; +import 'package:drift/native.dart'; +import 'package:drift/drift.dart' as drift; +import 'package:motorcycle_dash/database/database.dart' + hide TelemetrySample, FieldTestSummary, FieldTestEvent; +import 'package:motorcycle_dash/database/database.dart' as dbgen + show TelemetrySample; +import 'package:motorcycle_dash/models/telemetry_sample.dart'; +import 'package:motorcycle_dash/models/field_test_event.dart'; +import 'package:motorcycle_dash/services/telemetry_service.dart'; +import 'package:motorcycle_dash/services/location_service.dart'; +import 'package:motorcycle_dash/services/sensor_fusion_service.dart'; +import 'package:motorcycle_dash/services/compass_service.dart'; +import 'package:motorcycle_dash/services/audio_alert_service.dart'; +import 'package:motorcycle_dash/services/field_test/field_test_logger.dart'; +import 'package:motorcycle_dash/services/validation/telemetry_validator.dart'; +import 'package:motorcycle_dash/features/rides/ride_state_machine.dart'; +import 'package:motorcycle_dash/features/records/records_service.dart'; + +typedef TelemetryRow = dbgen.TelemetrySample; + +/// Counts every `batch` call and can park or fail it deterministically, so a +/// flush can be held exactly at the durability commit point without any +/// sleep-only race. +class _Rid3b2AppDatabase extends AppDatabase { + _Rid3b2AppDatabase() : super.forTesting(NativeDatabase.memory()); + + int batchInvocations = 0; + bool gateBatch = false; + final List> batchGates = []; + int batchFailuresRemaining = 0; + + @override + Future batch( + FutureOr Function(drift.Batch batch) runInBatch) async { + batchInvocations++; + if (gateBatch) { + final gate = Completer(); + batchGates.add(gate); + await gate.future; + } + if (batchFailuresRemaining > 0) { + batchFailuresRemaining--; + throw StateError('RID-3B-2 simulated pre-commit persistence failure'); + } + return super.batch(runInBatch); + } +} + +/// Captures log messages so the classification text emitted by the PRE-COMMIT +/// catch block can be inspected directly. That message is emitted only on a +/// genuine persistence failure, so its absence proves a committed snapshot was +/// never misclassified. Its own periodic flush() -> _db.batch is neutralised so +/// batch accounting refers only to telemetry persistence. +class _Rid3b2FieldTestLogger extends FieldTestLogger { + _Rid3b2FieldTestLogger(super.db); + + final List loggedMessages = []; + + int get retryWarnings => loggedMessages + .where((m) => m.contains('Database batch write retry')) + .length; + + @override + Future flush() async {} + + @override + void log({ + required FieldTestEventType eventType, + required EventSeverity severity, + required String message, + int? rideId, + DateTime? timestamp, + DateTime? telemetryTimestamp, + int? gpsAgeMs, + double? speed, + double? latitude, + double? longitude, + double? gpsAccuracy, + double? leanAngle, + double? longitudinalG, + double? lateralG, + double? overallConfidence, + Map? metadata, + }) { + loggedMessages.add(message); + super.log( + eventType: eventType, + severity: severity, + message: message, + rideId: rideId, + timestamp: timestamp, + telemetryTimestamp: telemetryTimestamp, + gpsAgeMs: gpsAgeMs, + speed: speed, + latitude: latitude, + longitude: longitude, + gpsAccuracy: gpsAccuracy, + leanAngle: leanAngle, + longitudinalG: longitudinalG, + lateralG: lateralG, + overallConfidence: overallConfidence, + metadata: metadata, + ); + } +} + + +class _Rid3b2AudioAlertService extends Fake implements AudioAlertService { + @override + Future speak(String text, {bool force = false}) async {} +} + +class _Rid3b2TelemetryService extends TelemetryService { + _Rid3b2TelemetryService(super.loc, super.fusion, super.compass, super.logger, + super.validator); + + final StreamController _controller = + StreamController.broadcast(); + + @override + Stream get telemetryStream => _controller.stream; + + void emitSample(TelemetrySample sample) => _controller.add(sample); + + @override + void dispose() => _controller.close(); +} + +class _NullLocationService extends Fake implements LocationService { + @override + void Function(Position position, bool accepted, String? rejectionReason)? + onGpsDiagnostic; + @override + Stream getFilteredPositionStream() => const Stream.empty(); +} + +class _NullSensorFusionService extends Fake implements SensorFusionService { + @override + Stream get leanAngleStream => const Stream.empty(); + @override + Stream get pitchAngleStream => const Stream.empty(); + @override + Stream get gForceStream => const Stream.empty(); + @override + double get latestRollRateDegPerSec => 0.0; + @override + double get latestRawAccelX => 0.0; + @override + double get latestRawAccelY => 0.0; + @override + double get latestRawAccelZ => 0.0; + @override + double get latestRawGyroX => 0.0; + @override + double get latestRawGyroY => 0.0; + @override + double get latestRawGyroZ => 0.0; +} + +class _NullCompassService extends Fake implements CompassService { + @override + Stream get headingStream => const Stream.empty(); +} + +void main() { + TestWidgetsFlutterBinding.ensureInitialized(); + + /// The production batch-write timer is a real 3-second Timer.periodic. Waiting + /// slightly longer than one interval makes the scheduled flush deterministic + /// without mocking the cadence, which is a frozen crash-durability contract. + const tickWait = Duration(milliseconds: 3300); + + late _Rid3b2AppDatabase db; + late _Rid3b2FieldTestLogger logger; + late _Rid3b2TelemetryService telemetry; + late TelemetryValidator validator; + late SharedPreferences prefs; + + final sampleBase = DateTime(2026, 9, 10, 12); + + setUp(() async { + SharedPreferences.setMockInitialValues({}); + prefs = await SharedPreferences.getInstance(); + db = _Rid3b2AppDatabase(); + logger = _Rid3b2FieldTestLogger(db); + validator = TelemetryValidator(); + telemetry = _Rid3b2TelemetryService(_NullLocationService(), + _NullSensorFusionService(), _NullCompassService(), logger, validator); + }); + + tearDown(() async { + for (final gate in db.batchGates) { + if (!gate.isCompleted) gate.complete(); + } + db.batchGates.clear(); + db.gateBatch = false; + db.batchFailuresRemaining = 0; + telemetry.dispose(); + logger.dispose(); + await db.close(); + }); + + ProviderContainer buildContainer({bool registerTeardown = true}) { + final container = ProviderContainer(overrides: [ + sharedPreferencesProvider.overrideWithValue(prefs), + databaseProvider.overrideWithValue(db), + telemetryServiceProvider.overrideWithValue(telemetry), + audioAlertServiceProvider + .overrideWithValue(_Rid3b2AudioAlertService()), + fieldTestLoggerProvider.overrideWithValue(logger), + ]); + if (registerTeardown) addTearDown(container.dispose); + return container; + } + + Future startRide(ProviderContainer container) async { + await container.read(rideStateMachineProvider.notifier).startRide(); + return container.read(rideStateMachineProvider).dbId!; + } + + /// Emits uniquely identifiable samples. The identity is encoded in `speed` so + /// durable rows can be accounted exactly once, and so chronological ordering is + /// observable from the autoincrement row order. + void emitSamples(int startIndex, int count) { + for (var i = 0; i < count; i++) { + final index = startIndex + i; + telemetry.emitSample(TelemetrySample( + timestamp: sampleBase.add(Duration(milliseconds: index * 50)), + latitude: 34.0 + (index * 0.0001), + longitude: -85.0, + speed: 1000.0 + index, + heading: 90.0, + gpsAccuracy: 3.0, + )); + } + } + + Future> persistedRows() => + (db.select(db.telemetrySamples) + ..orderBy([(t) => drift.OrderingTerm(expression: t.id)])) + .get(); + + Map countsByIndex(List rows) { + final counts = {}; + for (final row in rows) { + final speed = row.speed; + if (speed == null || speed < 1000) continue; + final index = (speed - 1000.0).round(); + counts[index] = (counts[index] ?? 0) + 1; + } + return counts; + } + + List duplicatedIn(Map counts) => + counts.entries.where((e) => e.value > 1).map((e) => e.key).toList() + ..sort(); + + List identitiesOf(List rows) => + rows.map((r) => (r.speed! - 1000.0).round()).toList(); + + /// Runs [body] inside a guarded zone and returns every error that escaped as an + /// uncaught asynchronous error. RID-3B-2 requires the scheduled flush Future to + /// be explicitly owned, so a regression surfaces here as a non-empty list + /// rather than as an opaque test failure. The container must be created inside + /// the zone for its timer-owned Futures to report there. + Future> captureUncaught(Future Function() body) async { + final uncaught = []; + await runZonedGuarded(body, (error, stackTrace) => uncaught.add(error)); + return uncaught; + } + + group('RID-3B-2 flush/dispose safety', () { + test('RID3B2-T1 scheduled flush Future is owned and contained', () async { + var rowsAfterFailure = -1; + var batchAfterFailure = -1; + var retryAfterFailure = -1; + var rowsAfterRetry = -1; + var counterAfterRetry = -1; + + // A genuinely reachable rejection path: the scheduled flush encounters a + // pre-commit persistence failure. The ownership wrapper must keep the + // Future owned so nothing escapes to the zone, and the timer must keep + // working on the following interval. No production seam is added solely to + // manufacture an impossible error. + final uncaught = await captureUncaught(() async { + final container = buildContainer(registerTeardown: false); + await startRide(container); + emitSamples(0, 6); + await pumpEventQueue(times: 30); + + db.batchFailuresRemaining = 1; + await Future.delayed(tickWait); + await pumpEventQueue(times: 40); + + rowsAfterFailure = (await persistedRows()).length; + batchAfterFailure = db.batchInvocations; + retryAfterFailure = logger.retryWarnings; + + // The restored snapshot must be retried by the next scheduled tick, + // proving the ownership wrapper did not break the timer. + await Future.delayed(tickWait); + await pumpEventQueue(times: 40); + + rowsAfterRetry = (await persistedRows()).length; + counterAfterRetry = + container.read(rideStateMachineProvider).totalSamplesPersisted; + container.dispose(); + }); + + expect(uncaught, isEmpty, + reason: 'the scheduled flush Future must be explicitly owned so no ' + 'rejection reaches the zone: $uncaught'); + expect(rowsAfterFailure, 0, reason: 'a failed attempt commits nothing'); + expect(batchAfterFailure, 1); + expect(retryAfterFailure, 1, + reason: 'the bounded pre-commit diagnostic is still emitted'); + expect(rowsAfterRetry, 6, reason: 'the timer retries and succeeds'); + expect(counterAfterRetry, 6, + reason: 'the counter accounts only for durable samples'); + }); + test('RID3B2-T2 dispose during parked db.batch commits without an ' + 'uncaught error', () async { + var rowsWhileParked = -1; + var counterWhileParked = -1; + var rowsAfterRelease = -1; + var distinct = -1; + var duplicates = []; + var batchCalls = -1; + var retryWarnings = -1; + var mountedAfterDispose = true; + + final uncaught = await captureUncaught(() async { + final container = buildContainer(registerTeardown: false); + final notifier = container.read(rideStateMachineProvider.notifier); + await startRide(container); + emitSamples(0, 10); + await pumpEventQueue(times: 30); + + // Park the flush exactly at the durability commit point. + db.gateBatch = true; + await Future.delayed(tickWait); + final parked = db.batchGates.single; + rowsWhileParked = (await persistedRows()).length; + counterWhileParked = + container.read(rideStateMachineProvider).totalSamplesPersisted; + + // Timer.cancel() cannot abort an already-started async callback, so this + // continuation legitimately outlives the notifier. + container.dispose(); + mountedAfterDispose = notifier.mounted; + + db.gateBatch = false; + parked.complete(); + await pumpEventQueue(times: 60); + + final rows = await persistedRows(); + final counts = countsByIndex(rows); + rowsAfterRelease = rows.length; + distinct = counts.length; + duplicates = duplicatedIn(counts); + batchCalls = db.batchInvocations; + retryWarnings = logger.retryWarnings; + }); + + expect(rowsWhileParked, 0, reason: 'nothing durable while parked'); + expect(counterWhileParked, 0, reason: 'counter not yet advanced'); + expect(mountedAfterDispose, isFalse, + reason: 'StateNotifier.mounted is the real disposal signal, so the ' + 'boundary cannot depend on a debug assertion'); + expect(uncaught, isEmpty, + reason: 'no post-dispose state or ref access may raise: $uncaught'); + expect(rowsAfterRelease, 10, reason: 'the committed snapshot is durable'); + expect(distinct, 10); + expect(duplicates, isEmpty, reason: 'persisted exactly once'); + expect(batchCalls, 1, reason: 'no retry ownership taken after disposal'); + expect(retryWarnings, 0, + reason: 'a committed write must never be reported as a persistence ' + 'failure, and no ref access may occur after disposal'); + }); + test('RID3B2-T3 a committed snapshot is never restored as a persistence ' + 'failure', () async { + var rowsPhase1 = -1; + var counterPhase1 = -1; + var rowsFinal = -1; + var duplicates = []; + var identities = []; + var batchCalls = -1; + var retryWarnings = -1; + + final uncaught = await captureUncaught(() async { + final container = buildContainer(registerTeardown: false); + await startRide(container); + + // Phase 1: a normal flush commits while the notifier is alive. + emitSamples(0, 10); + await pumpEventQueue(times: 30); + await Future.delayed(tickWait); + await pumpEventQueue(times: 40); + rowsPhase1 = (await persistedRows()).length; + counterPhase1 = + container.read(rideStateMachineProvider).totalSamplesPersisted; + + // Phase 2: disposal races a second committed flush. + emitSamples(10, 5); + await pumpEventQueue(times: 30); + db.gateBatch = true; + await Future.delayed(tickWait); + final parked = db.batchGates.single; + container.dispose(); + db.gateBatch = false; + parked.complete(); + await pumpEventQueue(times: 60); + + final rows = await persistedRows(); + final counts = countsByIndex(rows); + rowsFinal = rows.length; + duplicates = duplicatedIn(counts); + identities = counts.keys.toList()..sort(); + batchCalls = db.batchInvocations; + retryWarnings = logger.retryWarnings; + }); + + expect(rowsPhase1, 10); + expect(counterPhase1, 10, + reason: 'while mounted the counter increments normally'); + expect(uncaught, isEmpty, reason: '$uncaught'); + expect(rowsFinal, 15, reason: 'both committed snapshots are durable'); + expect(identities, List.generate(15, (i) => i), + reason: 'every identity present, none missing'); + expect(duplicates, isEmpty, + reason: 'no identity persisted twice: the phase-2 snapshot was not ' + 'restored to the live buffer and never retried'); + // Direct private-buffer inspection is unavailable, so non-restoration is + // proved through observable batch invocation and durable identity + // accounting: exactly one write per owned snapshot, and zero emissions of + // the diagnostic that only the pre-commit restoration path produces. + expect(batchCalls, 2, + reason: 'absence of a third invocation proves no retry ownership'); + expect(retryWarnings, 0, + reason: 'the pre-commit catch is the only emitter of this diagnostic'); + }); + test('RID3B2-T4 dead-notifier counter divergence is terminal, not a ' + 'persistence failure', () async { + var counterBeforeDispose = -1; + var durableRows = -1; + var batchCalls = -1; + var retryWarnings = -1; + var mountedAfterDispose = true; + + final uncaught = await captureUncaught(() async { + final container = buildContainer(registerTeardown: false); + final notifier = container.read(rideStateMachineProvider.notifier); + await startRide(container); + emitSamples(0, 10); + await pumpEventQueue(times: 30); + + db.gateBatch = true; + await Future.delayed(tickWait); + final parked = db.batchGates.single; + + // The last in-memory counter value this notifier will ever hold. + counterBeforeDispose = + container.read(rideStateMachineProvider).totalSamplesPersisted; + + container.dispose(); + mountedAfterDispose = notifier.mounted; + + db.gateBatch = false; + parked.complete(); + await pumpEventQueue(times: 60); + + durableRows = (await persistedRows()).length; + batchCalls = db.batchInvocations; + retryWarnings = logger.retryWarnings; + }); + + expect(uncaught, isEmpty, + reason: 'no attempt may be made to update dead notifier state: ' + '$uncaught'); + expect(mountedAfterDispose, isFalse); + expect(durableRows, 10, + reason: 'the database commit is authoritative and complete'); + expect(counterBeforeDispose, 0, + reason: 'the terminal in-memory counter never advanced past its ' + 'pre-commit value'); + // LOCKED RID-3B-2 CONTRACT (terminal dead-notifier counter divergence): + // durable DB rows may exceed the disposed notifier's final in-memory + // totalSamplesPersisted value. This is NOT a persistence failure, it does + // not alter the durability verdict, no live consumer may rely on that + // disposed counter, and committed samples are never restored to make it + // look consistent. No database counter or schema field is added. + expect(durableRows, greaterThan(counterBeforeDispose), + reason: 'the accepted divergence is explicitly locked here'); + expect(retryWarnings, 0, + reason: 'the durability verdict remains success, so the pre-commit ' + 'persistence-failure path must not be entered'); + expect(batchCalls, 1, reason: 'the committed snapshot is never retried'); + }); + + test('RID3B2-T5 a pre-commit failure still restores the snapshot and ' + 'retries exactly once', () async { + final container = buildContainer(); + await startRide(container); + emitSamples(0, 5); + await pumpEventQueue(times: 30); + + db.gateBatch = true; + await Future.delayed(tickWait); + final parked = db.batchGates.single; + + // Samples arriving DURING the await must end up behind the restored + // snapshot, which is what preserves RID-3A chronological ordering. + emitSamples(5, 3); + await pumpEventQueue(times: 20); + + db.batchFailuresRemaining = 1; + db.gateBatch = false; + parked.complete(); + await pumpEventQueue(times: 40); + + final rowsAfterFailure = await persistedRows(); + expect(rowsAfterFailure, isEmpty, + reason: 'a pre-commit failure commits nothing'); + expect(logger.retryWarnings, 1, + reason: 'the bounded pre-commit diagnostic is emitted while mounted'); + expect(container.read(rideStateMachineProvider).totalSamplesPersisted, 0, + reason: 'success-only counter accounting is preserved'); + + // A later legitimate flush retries the restored snapshot. + await Future.delayed(tickWait); + await pumpEventQueue(times: 40); + + final rows = await persistedRows(); + final counts = countsByIndex(rows); + expect(rows, hasLength(8), reason: 'every identity is eventually durable'); + expect(duplicatedIn(counts), isEmpty, reason: 'persisted exactly once'); + expect(counts.keys.toList()..sort(), List.generate(8, (i) => i), + reason: 'no identity missing'); + expect(identitiesOf(rows), List.generate(8, (i) => i), + reason: 'front-restoration preserved chronological ordering'); + expect(container.read(rideStateMachineProvider).totalSamplesPersisted, 8); + }); + test('RID3B2-T6 post-commit bookkeeping trouble is not classified as a ' + 'persistence failure', () async { + var rowsAfterFirst = -1; + var batchAfterFirst = -1; + var retryAfterFirst = -1; + var rowsFinal = -1; + var duplicates = []; + var batchFinal = -1; + + // StateNotifier.set state collects listener errors and rethrows + // StateNotifierListenerError, so a listener that throws on the post-commit + // counter update exercises "bookkeeping failed AFTER the commit point" + // while the notifier stays alive. Riverpod guards listener errors, so this + // locks the structural guarantee rather than relying on a manufactured + // production seam. + final uncaught = await captureUncaught(() async { + final container = buildContainer(registerTeardown: false); + await startRide(container); + emitSamples(0, 5); + await pumpEventQueue(times: 30); + + container.listen(rideStateMachineProvider, (prev, next) { + if (next.totalSamplesPersisted > (prev?.totalSamplesPersisted ?? 0)) { + throw StateError( + 'RID-3B-2 simulated post-commit bookkeeping failure'); + } + }, fireImmediately: false); + + db.gateBatch = true; + await Future.delayed(tickWait); + final parked = db.batchGates.single; + db.gateBatch = false; + parked.complete(); + await pumpEventQueue(times: 60); + + rowsAfterFirst = (await persistedRows()).length; + batchAfterFirst = db.batchInvocations; + retryAfterFirst = logger.retryWarnings; + + // A further scheduled tick must not re-persist the committed snapshot. + await Future.delayed(tickWait); + await pumpEventQueue(times: 60); + + final rows = await persistedRows(); + rowsFinal = rows.length; + duplicates = duplicatedIn(countsByIndex(rows)); + batchFinal = db.batchInvocations; + container.dispose(); + }); + + expect(uncaught.length, lessThanOrEqualTo(1), + reason: 'only the deliberately injected listener error may surface, ' + 'never a post-commit misclassification: $uncaught'); + expect(rowsAfterFirst, 5, reason: 'the commit point was reached'); + expect(retryAfterFirst, 0, + reason: 'the persistence catch is no longer entered because of ' + 'notifier bookkeeping'); + expect(batchAfterFirst, 1); + expect(rowsFinal, 5, reason: 'no duplicate persistence on a later tick'); + expect(duplicates, isEmpty, + reason: 'the snapshot-restoration path was not invoked'); + expect(batchFinal, 1, + reason: 'the committed snapshot was never returned to the live buffer, ' + 'so the next tick had nothing to own'); + }); + test('RID3B2-T7 buffered telemetry at graceful dispose is a documented ' + 'residual (concern D deferred)', () async { + final container = buildContainer(registerTeardown: false); + final notifier = container.read(rideStateMachineProvider.notifier); + final rideId = await startRide(container); + + emitSamples(0, 10); + await pumpEventQueue(times: 30); + final before = container.read(rideStateMachineProvider); + expect(before.totalSamplesReceived, 10); + expect(before.totalSamplesPersisted, 0, + reason: 'still buffered and not yet owned by a flush'); + expect(await persistedRows(), isEmpty); + expect(db.batchInvocations, 0, reason: 'no flush has run yet'); + + container.dispose(); + await pumpEventQueue(times: 30); + + // DEFERRED CONCERN D - LOCKED RESIDUAL, NOT AN ENDORSEMENT OF DATA LOSS. + // dispose() is synchronous and starts no flush, so telemetry that is still + // buffered and not yet owned is abandoned. RID-3B-2 deliberately does not + // solve this; these assertions exist so the patch cannot silently pretend + // it did. A later lifecycle ticket must decide the graceful-shutdown policy + // (owner-invoked async drain, or documented bounded loss). + expect(db.batchInvocations, 0, reason: 'dispose() must not start a flush'); + expect(await persistedRows(), isEmpty, + reason: 'these samples are NOT guaranteed durable'); + expect(notifier.mounted, isFalse); + final rideRow = + (await db.select(db.rides).get()).singleWhere((r) => r.id == rideId); + expect(rideRow.endTime, isNull, + reason: 'the ride is also left open by the deferred concern'); + }); + }); +}