diff --git a/lib/features/rides/ride_state_machine.dart b/lib/features/rides/ride_state_machine.dart index 7168f79..9019c30 100644 --- a/lib/features/rides/ride_state_machine.dart +++ b/lib/features/rides/ride_state_machine.dart @@ -145,6 +145,29 @@ class RideStateMachine extends StateNotifier { // persistence await is outstanding would re-open the shared-buffer race. bool _isFlushingTelemetryBatch = false; + // RID-3B-1: coordination primitive exposing the in-flight telemetry + // persistence operation, so an explicit Stop can await the current owner + // instead of observing a no-op flush. This is NOT a second buffer and NOT a + // serialization framework: it publishes exactly one Future for exactly one + // owned snapshot and is cleared when that snapshot's attempt settles. + Future? _activeTelemetryFlush; + + // RID-3B-1: late private telemetry-acceptance cutoff (invariant I3). Owned + // solely by the in-flight finish operation and reopened in its finally path, + // so a failed-but-retryable ride keeps recording and the next ride is + // unaffected. Strictly private: no public RideState or RideSession change. + bool _telemetryAcceptanceClosed = false; + + // RID-3B-1: bound on the convergent final drain. Telemetry arrives at 20 Hz + // (one sample per 50 ms) while each pass awaits a single db.batch round-trip, + // so the samples that can accumulate during a pass are normally 0 or 1 and + // convergence is reached in 1-2 passes. Four passes tolerate a persistence + // latency of roughly 150 ms per pass while still bounding finalization to at + // most four batch writes. The bound never fabricates durability: if telemetry + // is still outstanding when it is exhausted, finalization FAILS and remains + // retryable rather than transitioning to finished. + static const int _maxFinalDrainPasses = 4; + RideStateMachine(this.db, this.telemetryService, this.recordsService, this.audioAlertService, this.ref) : super(RideSession()) { _telemetrySub = telemetryService.telemetryStream.listen(_onTelemetryUpdate); @@ -375,6 +398,13 @@ class RideStateMachine extends StateNotifier { await _performFinishRide(); } finally { _isFinishing = false; + // RID-3B-1: the telemetry-acceptance cutoff is owned solely by this + // in-flight finish operation, so it is reopened here. A failed but + // retryable ride therefore keeps recording, and the next ride is + // unaffected. Deliberately NOT cleared from reset(): source analysis shows + // reset() has no callers, and clearing finish-owned state elsewhere would + // weaken the barrier. + _telemetryAcceptanceClosed = false; } } @@ -393,7 +423,11 @@ class RideStateMachine extends StateNotifier { ); } catch (_) {} - await _flushBatch(); // write remaining points + // RID-3B-1: the explicit-Stop durability barrier replaces the single + // tolerant flush. It awaits any in-flight periodic flush owner, drains to + // convergence, closes telemetry acceptance, and performs a final strict + // drain - throwing if durability cannot be established (invariant I2). + await _establishTelemetryDurabilityForFinish(); await db.endRide( state.dbId!, @@ -451,14 +485,19 @@ class RideStateMachine extends StateNotifier { } } catch (_) {} - state = state.copyWith( - state: RideState.finished, - healthStatus: RecordingHealthStatus.recordingComplete, - ); try { WakelockPlus.disable().catchError((_) {}); } catch (_) {} audioAlertService.speak('Session ended'); + + // RID-3B-1: the sync-visible finished edge is the LAST side effect of + // finalization. It is emitted only after telemetry durability, endTime, + // personal records and the forensic summary are complete, so SyncObserver + // can keep observing only this edge (invariant I1). + state = state.copyWith( + state: RideState.finished, + healthStatus: RecordingHealthStatus.recordingComplete, + ); } void reset() { @@ -480,7 +519,17 @@ class RideStateMachine extends StateNotifier { if (state.state != RideState.riding) return; - if (state.healthStatus != RecordingHealthStatus.recordingActive) { + // RID-3B-1 (invariant I3): telemetry accepted after the finalization + // boundary is never attributed to the finishing ride. The gate is private + // and is reopened by finishRide()'s finally path, so a retryable ride and + // every subsequent ride continue to record normally. + if (_telemetryAcceptanceClosed) return; + + // RID-3B-1: incoming telemetry must not clobber recordingFinalizing back to + // recordingActive while finalization is in progress. Minimum private + // condition: the existing RID-2 in-flight finish flag. + if (!_isFinishing && + state.healthStatus != RecordingHealthStatus.recordingActive) { state = state.copyWith(healthStatus: RecordingHealthStatus.recordingActive); } @@ -534,8 +583,143 @@ class RideStateMachine extends StateNotifier { state = state.copyWith(totalSamplesReceived: state.totalSamplesReceived + 1); } + /// RID-3A ownership semantics, plus RID-3B-1 active-flush publication. + /// + /// Takes exclusive ownership of the current buffer contents, removes them from + /// the live buffer BEFORE awaiting persistence, and restores them to the FRONT + /// on failure so chronological identity order survives. Returns true only when + /// the owned snapshot is durable (or there was nothing to own); it never + /// reports success for a failed persistence. + /// + /// Tolerant by design for the periodic path: a persistence failure is + /// retained, restored and logged with the existing bounded warning, never + /// thrown into the timer callback. + /// RID-3B-1: finish-only strict drain. + /// + /// Awaits any active telemetry flush owner first - looping, because a periodic + /// tick may take ownership again while waiting - then takes ownership under + /// RID-3A's guard and persists. Returns false whenever durability could not be + /// established, so finishRide can distinguish success from failure. It never + /// reports success for a failed persistence and never silently swallows a + /// durability failure. No automatic retry worker is added. + Future _drainTelemetryBatchStrictly() async { + while (_isFlushingTelemetryBatch) { + final active = _activeTelemetryFlush; + if (active == null) break; + try { + await active; + } catch (_) { + // The owner retains, restores and logs its own failure; ownership is + // re-checked by the loop condition. + } + } + if (_telemetryBatch.isEmpty) return true; + // Another owner is still active and could not be awaited, so durability is + // not established and must not be assumed. + if (_isFlushingTelemetryBatch) return false; + return _persistOwnedTelemetrySnapshot(); + } + + /// RID-3B-1: the explicit-Stop durability barrier (invariants I1-I3). + /// + /// Awaits any in-flight periodic flush, drains the buffer to convergence within + /// a bounded policy while telemetry capture continues, closes acceptance, then + /// performs one final strict drain for anything queued between the last + /// convergence check and the cutoff assignment. + /// + /// Throws if durability cannot be established, so the caller aborts + /// finalization without writing endTime, evaluating personal records, + /// generating a forensic summary, announcing completion, or emitting the + /// sync-visible finished edge. + Future _establishTelemetryDurabilityForFinish() async { + var durable = true; + var passes = 0; + + // Bounded convergent drain. Telemetry keeps being accepted during this phase + // so existing capture semantics are preserved; the final few samples are + // never intentionally dropped to simplify finalization. + while (passes < _maxFinalDrainPasses && + (_telemetryBatch.isNotEmpty || _activeTelemetryFlush != null)) { + passes++; + if (!await _drainTelemetryBatchStrictly()) { + durable = false; + break; + } + } + + // Convergence must actually have been reached. The cap never fabricates + // durability: telemetry still outstanding means finalization fails. + if (durable && + (_telemetryBatch.isNotEmpty || _activeTelemetryFlush != null)) { + durable = false; + } + + if (durable) { + // I3: close acceptance, then one final strict drain. + _telemetryAcceptanceClosed = true; + durable = await _drainTelemetryBatchStrictly(); + } + + if (!durable) { + // I2: restore a truthful recoverable recording status and propagate, so + // RID-2's existing UI boundary containment handles the rejection and the + // ride remains retryable. RID-3A has already restored any failed owned + // snapshot to the front of the buffer in correct order. + state = state.copyWith(healthStatus: RecordingHealthStatus.recordingActive); + throw StateError( + 'Ride finalization aborted: telemetry durability could not be established', + ); + } + } + + Future _persistOwnedTelemetrySnapshot() async { + if (_isFlushingTelemetryBatch || _telemetryBatch.isEmpty) return true; + + _isFlushingTelemetryBatch = true; + final batchToInsert = List.from(_telemetryBatch); + _telemetryBatch.clear(); + // Publish this exact in-flight operation so an explicit Stop can await the + // current owner instead of observing a no-op flush. One Future for one owned + // snapshot - a coordination primitive, not a second buffering system. + 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 + try { + ref.read(fieldTestLoggerProvider).log( + eventType: FieldTestEventType.databaseQueueWarning, + message: 'Database batch write retry: $e', + severity: EventSeverity.warning, + ); + } catch (_) {} + return false; + } finally { + _isFlushingTelemetryBatch = false; + if (!completion.isCompleted) completion.complete(); + _activeTelemetryFlush = null; + } + } + Future _flushBatch() async { - // RID-3A: telemetry batch persistence with exclusive snapshot ownership. + // RID-3A telemetry batch ownership (exclusive snapshot, clear-before-await, + // front-restoration on failure, success-only counter) now lives in + // _persistOwnedTelemetrySnapshot(), unchanged in behaviour. RID-3B-1 also + // publishes the in-flight operation there so an explicit Stop can await the + // current owner instead of observing a no-op flush. The periodic path stays + // tolerant and never throws into the timer callback; timer Future ownership + // and teardown safety are RID-3B-2 scope and are deliberately unchanged. // // The guard is raised synchronously, before the first `await`, so a // competing legitimate flush (the 3 s periodic timer against the flush @@ -552,35 +736,7 @@ class RideStateMachine extends StateNotifier { // // The incremental ride-stats persistence below is deliberately left outside // this guard so its existing every-tick cadence is unchanged. - if (!_isFlushingTelemetryBatch && _telemetryBatch.isNotEmpty) { - _isFlushingTelemetryBatch = true; - final batchToInsert = List.from(_telemetryBatch); - _telemetryBatch.clear(); - 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); - } 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 chronological identity order survives and a later legitimate flush - // can retry them. No automatic retry worker is introduced. - _telemetryBatch.insertAll(0, batchToInsert); - // Keep in queue for retry on next flush interval - try { - ref.read(fieldTestLoggerProvider).log( - eventType: FieldTestEventType.databaseQueueWarning, - message: 'Database batch write retry: $e', - severity: EventSeverity.warning, - ); - } catch (_) {} - } finally { - _isFlushingTelemetryBatch = false; - } - } + await _persistOwnedTelemetrySnapshot(); // Incrementally persist ongoing ride summary stats to SQLite so crashes/terminations never lose stats if (state.dbId != null && state.state == RideState.riding) { diff --git a/test/rid3a_telemetry_batch_ownership_test.dart b/test/rid3a_telemetry_batch_ownership_test.dart index 6bd75bb..3ce5161 100644 --- a/test/rid3a_telemetry_batch_ownership_test.dart +++ b/test/rid3a_telemetry_batch_ownership_test.dart @@ -321,32 +321,48 @@ void main() { await pumpEventQueue(times: 40); expect(container.read(rideStateMachineProvider).totalSamplesReceived, 30); - // Competing legitimate flush: finishRide() awaits _flushBatch(). It must - // take no ownership of the samples A already owns, and must not touch the - // newer ones either. + // Competing legitimate flush: finishRide() now awaits the active owner + // (RID-3B-1) instead of observing a no-op. While flush A is still parked it + // must take no ownership of A's snapshot and issue no batch of its own; it + // drains the newer samples only after A has committed. final finishing = notifier.finishRide(); await pumpEventQueue(times: 30); expect(db.batchInvocations, 1, - reason: 'a competing flush while the guard is raised must take no ' - 'ownership of the live buffer'); + reason: 'while the owner is parked, no second flush may take ownership'); - // Release A: it persists exactly its owned snapshot (0..9) and nothing - // else is removed from the live buffer. - db.gateBatch = false; + // Release ONLY flush A, leaving the gate armed so the RID-3B-1 durability + // barrier's own drain parks as the next gate. That yields a deterministic + // observation point at which flush A's committed write is isolated: A has + // persisted and nothing else has been written yet. flushA.complete(); - await finishing; await pumpEventQueue(times: 40); - final rowsAfterA = await persistedRows(); - final countsAfterA = countsByIndex(rowsAfterA); - expect(duplicatedIn(countsAfterA), isEmpty); - expect(missingOf(10, countsAfterA), isEmpty, - reason: 'A persisted its owned snapshot'); - expect(countsAfterA.length, 10, - reason: 'A must not have persisted samples it did not own'); + final rowsOwnedByA = await persistedRows(); + final countsOwnedByA = countsByIndex(rowsOwnedByA); + expect(duplicatedIn(countsOwnedByA), isEmpty, + reason: 'no duplication from overlapping ownership'); + expect(missingOf(10, countsOwnedByA), isEmpty, + reason: 'A persisted its whole owned snapshot'); + // The real RID-3A ownership property: flush A's write contains exactly its + // own identity set, and none of the newer samples - those remain buffered + // for later ownership. This is observed BEFORE finishRide can complete the + // durability barrier, not after `await finishing`, because RID-3B-1 now + // intentionally drains all accepted pre-boundary telemetry before + // finalizing. The previous "only 10 rows after await finishing" expectation + // depended on the finish flush being a no-op, which RID-3B-1 removed. + expect(countsOwnedByA.keys.toList()..sort(), + [for (var i = 0; i < 10; i++) i], + reason: 'flush A owns exactly indices 0-9'); + expect(countsOwnedByA.keys.where((i) => i >= 10).toList(), isEmpty, + reason: 'no newer sample was included in flush A write'); + expect(db.batchGates, hasLength(2), + reason: 'the barrier now owns the newer batch and is parked'); - // Legitimate later flush (the periodic timer) persists 10..29. - await Future.delayed(tickWait); + // Release the barrier's drain and let finishRide complete. It persists + // 10..29 itself, so no later periodic flush is required. + db.gateBatch = false; + db.batchGates.last.complete(); + await finishing; await pumpEventQueue(times: 40); final rows = await persistedRows(); diff --git a/test/rid3b1_stop_durability_barrier_test.dart b/test/rid3b1_stop_durability_barrier_test.dart new file mode 100644 index 0000000..a7d9a56 --- /dev/null +++ b/test/rid3b1_stop_durability_barrier_test.dart @@ -0,0 +1,731 @@ +// RID-3B-1 regression coverage: explicit-Stop telemetry durability barrier. +// +// Invariants under test: +// I1 - finished is emitted only after every sample accepted before the +// finalization boundary is durably persisted exactly once. +// I2 - if durability cannot be established, finalization fails: no endTime, +// no records, no summary, no finished edge, retry remains possible. +// I3 - telemetry accepted after the boundary is not attributed to the ride. +// I4 - these apply to explicit Stop only (dispose/process death are RID-3B-2). +// +// Ordering is MEASURED, not inferred: the db.batch / endRide / records / +// stopSession overrides and the Riverpod finished-edge listener all append to one +// shared ordered event log. Race windows use Completer gates, never sleeps. +import 'dart:async'; + +import 'package:flutter/foundation.dart'; +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_summary.dart'; +import 'package:motorcycle_dash/models/recording_health_status.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; + +class _Rid3b1AppDatabase extends AppDatabase { + _Rid3b1AppDatabase(this.eventLog) : super.forTesting(NativeDatabase.memory()); + + final List eventLog; + int batchInvocations = 0; + int endRideInvocations = 0; + bool gateBatch = false; + final List> batchGates = []; + int batchFailuresRemaining = 0; + + @override + Future batch( + FutureOr Function(drift.Batch batch) runInBatch) async { + final index = batchInvocations++; + if (gateBatch) { + final gate = Completer(); + batchGates.add(gate); + eventLog.add('batch$index:parked'); + await gate.future; + } + if (batchFailuresRemaining > 0) { + batchFailuresRemaining--; + eventLog.add('batch$index:FAILED'); + throw StateError('RID-3B-1 simulated telemetry persistence failure'); + } + await super.batch(runInBatch); + eventLog.add('batch$index:committed'); + } + + @override + Future endRide(int rideId, RidesCompanion data) async { + endRideInvocations++; + eventLog.add('endRide'); + return super.endRide(rideId, data); + } +} + +/// Neutralises FieldTestLogger's own 3 s periodic flush() -> _db.batch +/// (field_test_logger.dart:138, :592) so db.batch accounting refers only to +/// telemetry persistence. FieldTestLogger internals are NOT modified. +class _Rid3b1FieldTestLogger extends FieldTestLogger { + _Rid3b1FieldTestLogger(super.db, this.eventLog); + + final List eventLog; + int stopSessionInvocations = 0; + + @override + Future flush() async {} + + @override + Future stopSession() async { + stopSessionInvocations++; + eventLog.add('stopSession'); + return super.stopSession(); + } +} + +class _Rid3b1RecordsService extends RecordsService { + _Rid3b1RecordsService(super.prefs, this.eventLog); + + final List eventLog; + int checkInvocations = 0; + + @override + Future checkAndSaveRecords( + PersonalRecords newStats, + void Function(String, double)? onNewRecord, { + double confidence = 1.0, + }) async { + checkInvocations++; + eventLog.add('records'); + return super.checkAndSaveRecords(newStats, onNewRecord, + confidence: confidence); + } +} + +class MockAudioAlertService extends Fake implements AudioAlertService { + final List spokenMessages = []; + @override + Future speak(String text, {bool force = false}) async { + spokenMessages.add(text); + } +} + +class FakeTelemetryService extends TelemetryService { + final StreamController _controller = + StreamController.broadcast(); + + FakeTelemetryService( + super.loc, + super.fusion, + super.compass, + super.logger, + super.validator, + ); + + @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(); + + const tickWait = Duration(milliseconds: 3300); + + late _Rid3b1AppDatabase db; + late _Rid3b1FieldTestLogger logger; + late _Rid3b1RecordsService records; + late FakeTelemetryService fakeTelemetry; + late MockAudioAlertService mockAudio; + late TelemetryValidator validator; + late SharedPreferences prefs; + late List eventLog; + late int finishedTransitions; + + final sampleBase = DateTime(2026, 9, 10, 12); + + setUp(() async { + SharedPreferences.setMockInitialValues({}); + prefs = await SharedPreferences.getInstance(); + eventLog = []; + db = _Rid3b1AppDatabase(eventLog); + logger = _Rid3b1FieldTestLogger(db, eventLog); + records = _Rid3b1RecordsService(prefs, eventLog); + validator = TelemetryValidator(); + fakeTelemetry = FakeTelemetryService( + _NullLocationService(), + _NullSensorFusionService(), + _NullCompassService(), + logger, + validator, + ); + mockAudio = MockAudioAlertService(); + finishedTransitions = 0; + }); + + tearDown(() async { + for (final g in db.batchGates) { + if (!g.isCompleted) g.complete(); + } + db.batchGates.clear(); + db.gateBatch = false; + fakeTelemetry.dispose(); + logger.dispose(); + await db.close(); + }); + + ProviderContainer buildContainer() { + final container = ProviderContainer( + overrides: [ + sharedPreferencesProvider.overrideWithValue(prefs), + databaseProvider.overrideWithValue(db), + telemetryServiceProvider.overrideWithValue(fakeTelemetry), + audioAlertServiceProvider.overrideWithValue(mockAudio), + fieldTestLoggerProvider.overrideWithValue(logger), + recordsServiceProvider.overrideWith((ref) => records), + ], + ); + addTearDown(container.dispose); + container.listen(rideStateMachineProvider, (prev, next) { + if (prev?.state != RideState.finished && + next.state == RideState.finished) { + finishedTransitions++; + eventLog.add('FINISHED-EDGE(sync-notify)'); + } + }); + return container; + } + + Future startRide(ProviderContainer container) async { + final notifier = container.read(rideStateMachineProvider.notifier); + await notifier.startRide(); + return container.read(rideStateMachineProvider).dbId!; + } + + /// Unique identity: sample index N carries speed 1000 + N. + void emitSamples(int startIndex, int count) { + for (var i = 0; i < count; i++) { + final index = startIndex + i; + fakeTelemetry.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 idx = (speed - 1000.0).round(); + counts[idx] = (counts[idx] ?? 0) + 1; + } + return counts; + } + + List duplicatedIn(Map c) => + c.entries.where((e) => e.value > 1).map((e) => e.key).toList()..sort(); + + List missingOf(int total, Map c) => + [for (var i = 0; i < total; i++) if ((c[i] ?? 0) == 0) i]; + + int sessionEndedCount() => + mockAudio.spokenMessages.where((m) => m == 'Session ended').length; + + void releaseAllGates() { + db.gateBatch = false; + for (final g in db.batchGates) { + if (!g.isCompleted) g.complete(); + } + } + + group('RID-3B-1: explicit-Stop durability barrier', () { + test('A normal Stop makes telemetry durable before endTime, records, ' + 'summary and the finished edge', () async { + final container = buildContainer(); + final notifier = container.read(rideStateMachineProvider.notifier); + final rideId = await startRide(container); + + emitSamples(0, 10); + await pumpEventQueue(times: 30); + expect(container.read(rideStateMachineProvider).totalSamplesReceived, 10); + + await notifier.finishRide(); + await pumpEventQueue(times: 30); + + final rows = await persistedRows(); + final counts = countsByIndex(rows); + final session = container.read(rideStateMachineProvider); + final rideRow = + (await db.select(db.rides).get()).singleWhere((r) => r.id == rideId); + + debugPrint('==== RID-3B-1 A (normal stop) ===='); + debugPrint('event log : $eventLog'); + debugPrint('rows : ${rows.length} distinct: ${counts.length}'); + debugPrint('counters : ${session.totalSamplesReceived} / ' + '${session.totalSamplesPersisted}'); + + // I1 ordering, measured from the shared event log. + final commit = eventLog.indexOf('batch0:committed'); + final endRideAt = eventLog.indexOf('endRide'); + final recordsAt = eventLog.indexOf('records'); + final stopAt = eventLog.indexOf('stopSession'); + final edge = eventLog.indexOf('FINISHED-EDGE(sync-notify)'); + expect(commit, greaterThanOrEqualTo(0)); + expect(commit, lessThan(endRideAt), reason: 'durability before endTime'); + expect(endRideAt, lessThan(recordsAt), reason: 'endTime before records'); + expect(recordsAt, lessThan(stopAt), reason: 'records before summary'); + expect(stopAt, lessThan(edge), reason: 'summary before the finished edge'); + + expect(rows, hasLength(10)); + expect(counts.length, 10); + expect(duplicatedIn(counts), isEmpty); + expect(missingOf(10, counts), isEmpty); + expect(rows.map((r) => r.rideId).toSet(), {rideId}); + expect(rows.map((r) => r.speed).toList(), + [for (var i = 0; i < 10; i++) 1000.0 + i]); + expect(rideRow.endTime, isNotNull); + expect(finishedTransitions, 1, reason: 'exactly one finished edge'); + expect(sessionEndedCount(), 1); + expect(session.state, RideState.finished); + expect(session.healthStatus, RecordingHealthStatus.recordingComplete); + expect(session.totalSamplesPersisted, 10); + expect(await db.select(db.fieldTestSummaries).get(), hasLength(1)); + expect(records.checkInvocations, 1); + expect(logger.stopSessionInvocations, 1); + }); + + test('B finishRide waits for an in-flight periodic flush and emits no ' + 'finished edge until every sample is durable', () async { + final container = buildContainer(); + final notifier = container.read(rideStateMachineProvider.notifier); + final rideId = await startRide(container); + + emitSamples(0, 10); + await pumpEventQueue(times: 30); + + // The production periodic flush takes ownership of 0..9 and parks. + db.gateBatch = true; + await Future.delayed(tickWait); + expect(db.batchInvocations, 1, reason: 'periodic flush owns batch A'); + expect(db.batchGates, hasLength(1)); + + // Newer telemetry accumulates while A is in flight. + emitSamples(10, 5); + await pumpEventQueue(times: 30); + expect(container.read(rideStateMachineProvider).totalSamplesReceived, 15); + + final finishing = notifier.finishRide(); + await pumpEventQueue(times: 40); + + // B.4: finishRide MUST NOT finish while the periodic owner is parked. + expect(container.read(rideStateMachineProvider).state, + isNot(RideState.finished)); + expect(finishedTransitions, 0, reason: 'no sync-visible edge yet'); + expect(eventLog.contains('endRide'), isFalse, + reason: 'no endTime before durability'); + expect(eventLog.contains('records'), isFalse); + expect(eventLog.contains('stopSession'), isFalse); + expect(await persistedRows(), isEmpty); + + // B.5/B.6: release A; the barrier then drains the remainder itself. + releaseAllGates(); + await finishing; + await pumpEventQueue(times: 40); + + final rows = await persistedRows(); + final counts = countsByIndex(rows); + final session = container.read(rideStateMachineProvider); + debugPrint('==== RID-3B-1 B (in-flight periodic flush) ===='); + debugPrint('event log : $eventLog'); + debugPrint('rows : ${rows.length} distinct: ${counts.length}'); + debugPrint('counters : ${session.totalSamplesReceived} / ' + '${session.totalSamplesPersisted}'); + + // B.7: the finished edge only after ALL telemetry is durable. + final edge = eventLog.indexOf('FINISHED-EDGE(sync-notify)'); + for (final entry in eventLog) { + if (entry.endsWith(':committed')) { + expect(eventLog.indexOf(entry), lessThan(edge), + reason: 'every commit precedes the finished edge'); + } + } + expect(rows, hasLength(15)); + expect(counts.length, 15); + expect(duplicatedIn(counts), isEmpty); + expect(missingOf(15, counts), isEmpty); + expect(rows.map((r) => r.rideId).toSet(), {rideId}); + expect(rows.map((r) => r.speed).toList(), + [for (var i = 0; i < 15; i++) 1000.0 + i]); + expect(finishedTransitions, 1); + expect(session.totalSamplesPersisted, 15); + }); + + test('C telemetry arriving during finalization is retained, persisted ' + 'exactly once, and durable before the finished edge', () async { + final container = buildContainer(); + final notifier = container.read(rideStateMachineProvider.notifier); + final rideId = await startRide(container); + + emitSamples(0, 5); + await pumpEventQueue(times: 30); + + db.gateBatch = true; + final finishing = notifier.finishRide(); + await pumpEventQueue(times: 20); + expect(db.batchInvocations, 1, reason: 'first drain pass parked'); + + // Arrives before the acceptance cutoff, so it must be retained. + emitSamples(5, 5); + await pumpEventQueue(times: 30); + expect(container.read(rideStateMachineProvider).totalSamplesReceived, 10); + + releaseAllGates(); + await finishing; + await pumpEventQueue(times: 40); + + final rows = await persistedRows(); + final counts = countsByIndex(rows); + final session = container.read(rideStateMachineProvider); + debugPrint('==== RID-3B-1 C (during-finalization arrival) ===='); + debugPrint('event log : $eventLog'); + debugPrint('rows : ${rows.length} distinct: ${counts.length}'); + + expect(rows, hasLength(10), reason: 'nothing dropped to simplify finish'); + expect(counts.length, 10); + expect(duplicatedIn(counts), isEmpty); + expect(missingOf(10, counts), isEmpty); + expect(rows.map((r) => r.rideId).toSet(), {rideId}); + expect(rows.map((r) => r.speed).toList(), + [for (var i = 0; i < 10; i++) 1000.0 + i]); + expect(eventLog.indexOf('batch1:committed'), + lessThan(eventLog.indexOf('FINISHED-EDGE(sync-notify)')), + reason: 'the during-finalization batch is durable before the edge'); + expect(finishedTransitions, 1); + expect(session.totalSamplesPersisted, 10); + }); + + test('E final drain failure aborts finalization: no endTime, records, ' + 'summary, announcement or finished edge', () async { + final container = buildContainer(); + final notifier = container.read(rideStateMachineProvider.notifier); + final rideId = await startRide(container); + + emitSamples(0, 10); + await pumpEventQueue(times: 30); + + db.batchFailuresRemaining = 1; + Object? escaped; + try { + await notifier.finishRide(); + } catch (e) { + escaped = e; + } + await pumpEventQueue(times: 40); + + final session = container.read(rideStateMachineProvider); + final rows = await persistedRows(); + final rideRow = + (await db.select(db.rides).get()).singleWhere((r) => r.id == rideId); + + debugPrint('==== RID-3B-1 E (final drain failure) ===='); + debugPrint('event log : $eventLog'); + debugPrint('escaped : ${escaped.runtimeType}'); + debugPrint('state : ${session.state} / ${session.healthStatus}'); + debugPrint('counters : ${session.totalSamplesReceived} / ' + '${session.totalSamplesPersisted}'); + + expect(escaped, isA(), + reason: 'the durability failure must propagate from finishRide'); + expect(session.state, isNot(RideState.finished), reason: 'I2'); + expect(session.healthStatus, RecordingHealthStatus.recordingActive, + reason: 'truthful recoverable recording status'); + expect(rideRow.endTime, isNull, reason: 'I2: no endTime written'); + expect(db.endRideInvocations, 0); + expect(records.checkInvocations, 0, reason: 'I2: no records pass'); + expect(logger.stopSessionInvocations, 0, reason: 'I2: no logger stop'); + expect(await db.select(db.fieldTestSummaries).get(), isEmpty, + reason: 'I2: no forensic summary'); + expect(finishedTransitions, 0, reason: 'I2: no sync-visible edge'); + expect(eventLog.contains('FINISHED-EDGE(sync-notify)'), isFalse); + expect(sessionEndedCount(), 0, reason: 'I2: no Session ended'); + expect(rows, isEmpty); + expect(session.totalSamplesPersisted, 0); + expect(session.totalSamplesReceived, 10, + reason: 'RID-3A retained the failed snapshot for retry'); + expect(session.dbId, rideId, reason: 'ride remains retryable'); + }); + + test('F after a failed finalization an explicit retry finalizes exactly ' + 'once with every accepted sample durable', () async { + final container = buildContainer(); + final notifier = container.read(rideStateMachineProvider.notifier); + final rideId = await startRide(container); + final cloudRideIdBefore = (await db.select(db.rides).get()) + .singleWhere((r) => r.id == rideId) + .cloudRideId; + + emitSamples(0, 10); + await pumpEventQueue(times: 30); + + db.batchFailuresRemaining = 1; + try { + await notifier.finishRide(); + } catch (_) {} + await pumpEventQueue(times: 40); + expect(finishedTransitions, 0); + expect(await persistedRows(), isEmpty); + + // Explicit retry, with no injected failure remaining. + await notifier.finishRide(); + await pumpEventQueue(times: 40); + + final rows = await persistedRows(); + final counts = countsByIndex(rows); + final session = container.read(rideStateMachineProvider); + final rides = await db.select(db.rides).get(); + final rideRow = rides.singleWhere((r) => r.id == rideId); + + debugPrint('==== RID-3B-1 F (failure then retry) ===='); + debugPrint('event log : $eventLog'); + debugPrint('rows : ${rows.length} distinct: ${counts.length}'); + debugPrint('counters : ${session.totalSamplesReceived} / ' + '${session.totalSamplesPersisted}'); + + expect(rows, hasLength(10)); + expect(counts.length, 10); + expect(duplicatedIn(counts), isEmpty, reason: 'retry causes no duplicates'); + expect(missingOf(10, counts), isEmpty, reason: 'retry loses nothing'); + expect(rows.map((r) => r.speed).toList(), + [for (var i = 0; i < 10; i++) 1000.0 + i], + reason: 'chronological order preserved across the failure'); + expect(db.endRideInvocations, 1, reason: 'exactly one endTime write'); + expect(rideRow.endTime, isNotNull); + expect(records.checkInvocations, 1); + expect(logger.stopSessionInvocations, 1); + expect(await db.select(db.fieldTestSummaries).get(), hasLength(1)); + expect(sessionEndedCount(), 1); + expect(finishedTransitions, 1); + expect(session.state, RideState.finished); + expect(rides, hasLength(1), reason: 'no duplicate ride row'); + expect(rides.single.cloudRideId, cloudRideIdBefore, + reason: 'no new cloudRideId minted'); + expect(session.totalSamplesPersisted, 10); + }); + + test('D telemetry after the boundary is not attributed to the finished ' + 'ride, and acceptance reopens for the next ride', () async { + final container = buildContainer(); + final notifier = container.read(rideStateMachineProvider.notifier); + final rideId = await startRide(container); + + emitSamples(0, 5); + await pumpEventQueue(times: 30); + await notifier.finishRide(); + await pumpEventQueue(times: 40); + + final receivedAtFinish = + container.read(rideStateMachineProvider).totalSamplesReceived; + expect(receivedAtFinish, 5); + + // I3: telemetry after the finalization boundary never enters the finished + // ride's buffer. Observed indirectly, since the cutoff instant itself is + // private and widening the public API to expose it is not permitted. + emitSamples(5, 5); + await pumpEventQueue(times: 40); + expect(container.read(rideStateMachineProvider).totalSamplesReceived, + receivedAtFinish, + reason: 'no post-boundary sample accepted into the finishing ride'); + var rows = await persistedRows(); + expect(rows, hasLength(5)); + expect(rows.map((r) => r.rideId).toSet(), {rideId}); + + // The cutoff is owned solely by the in-flight finish operation, so it must + // have reopened: the NEXT ride records normally and its samples are + // attributed only to the new ride (no cross-ride leakage). + final rideId2 = await startRide(container); + expect(rideId2, isNot(rideId)); + emitSamples(100, 5); + await pumpEventQueue(times: 40); + expect(container.read(rideStateMachineProvider).totalSamplesReceived, 5, + reason: 'acceptance reopened for the new ride'); + await notifier.finishRide(); + await pumpEventQueue(times: 40); + + rows = await persistedRows(); + final ride1Rows = rows.where((r) => r.rideId == rideId).toList(); + final ride2Rows = rows.where((r) => r.rideId == rideId2).toList(); + debugPrint('==== RID-3B-1 D (after-boundary attribution) ===='); + debugPrint('ride1 rows : ${ride1Rows.length}'); + debugPrint('ride2 rows : ${ride2Rows.length}'); + debugPrint('ride2 ids : ${ride2Rows.map((r) => r.speed).toList()}'); + + expect(ride1Rows, hasLength(5), + reason: 'the finished ride gained no later samples'); + expect(ride2Rows, hasLength(5), reason: 'the next ride recorded normally'); + expect(ride2Rows.map((r) => r.speed).toList(), + [for (var i = 0; i < 5; i++) 1100.0 + i]); + expect(duplicatedIn(countsByIndex(rows)), isEmpty); + expect(finishedTransitions, 2, reason: 'one finished edge per ride'); + }); + + test('H telemetry arriving during finalization does not clobber ' + 'recordingFinalizing', () async { + final container = buildContainer(); + final notifier = container.read(rideStateMachineProvider.notifier); + await startRide(container); + + emitSamples(0, 5); + await pumpEventQueue(times: 30); + + db.gateBatch = true; + final finishing = notifier.finishRide(); + await pumpEventQueue(times: 20); + + final healthWhileFinalizing = + container.read(rideStateMachineProvider).healthStatus; + + emitSamples(5, 5); + await pumpEventQueue(times: 30); + final healthAfterArrival = + container.read(rideStateMachineProvider).healthStatus; + + releaseAllGates(); + await finishing; + await pumpEventQueue(times: 40); + + debugPrint('==== RID-3B-1 H (healthStatus) ===='); + debugPrint('while finalizing : $healthWhileFinalizing'); + debugPrint('after arrival : $healthAfterArrival'); + debugPrint('final : ' + '${container.read(rideStateMachineProvider).healthStatus}'); + + expect(healthWhileFinalizing, RecordingHealthStatus.recordingFinalizing); + expect(healthAfterArrival, RecordingHealthStatus.recordingFinalizing, + reason: 'incoming telemetry must not reset recordingFinalizing'); + expect(container.read(rideStateMachineProvider).healthStatus, + RecordingHealthStatus.recordingComplete); + }); + + test('G RID-3A exactly-once accounting is preserved under the durability ' + 'barrier (30-sample overlap shape)', () async { + final container = buildContainer(); + final notifier = container.read(rideStateMachineProvider.notifier); + final rideId = await startRide(container); + + emitSamples(0, 10); + await pumpEventQueue(times: 30); + + // Periodic flush owns 0..9 and parks. + db.gateBatch = true; + await Future.delayed(tickWait); + expect(db.batchInvocations, 1); + + // 20 newer samples accumulate while the owner is in flight. + emitSamples(10, 20); + await pumpEventQueue(times: 40); + expect(container.read(rideStateMachineProvider).totalSamplesReceived, 30); + + // finishRide must await the owner rather than observe a no-op flush. + final finishing = notifier.finishRide(); + await pumpEventQueue(times: 40); + expect(container.read(rideStateMachineProvider).state, + isNot(RideState.finished)); + expect(finishedTransitions, 0); + + releaseAllGates(); + await finishing; + await pumpEventQueue(times: 40); + + final rows = await persistedRows(); + final counts = countsByIndex(rows); + final session = container.read(rideStateMachineProvider); + debugPrint('==== RID-3B-1 G (RID-3A accounting under barrier) ===='); + debugPrint('event log : $eventLog'); + debugPrint('accepted : ${session.totalSamplesReceived}'); + debugPrint('rows : ${rows.length}'); + debugPrint('distinct : ${counts.length}'); + debugPrint('duplicated: ${duplicatedIn(counts)}'); + debugPrint('missing : ${missingOf(30, counts)}'); + debugPrint('counters : ${session.totalSamplesReceived} / ' + '${session.totalSamplesPersisted}'); + + expect(rows, hasLength(30), reason: '30 accepted -> 30 durable rows'); + expect(counts.length, 30, reason: '30 distinct identities'); + expect(duplicatedIn(counts), isEmpty, reason: '0 duplicates'); + expect(missingOf(30, counts), isEmpty, reason: '0 missing'); + expect(rows.map((r) => r.speed).toList(), + [for (var i = 0; i < 30; i++) 1000.0 + i], + reason: 'chronological identity order'); + expect(rows.map((r) => r.rideId).toSet(), {rideId}); + expect(session.totalSamplesPersisted, 30, + reason: 'counter integrity: accepted distinct == durable distinct'); + expect(finishedTransitions, 1); + expect((await db.select(db.rides).get()), hasLength(1)); + }); + }); +}