diff --git a/lib/features/rides/ride_state_machine.dart b/lib/features/rides/ride_state_machine.dart index cf2e610..7168f79 100644 --- a/lib/features/rides/ride_state_machine.dart +++ b/lib/features/rides/ride_state_machine.dart @@ -138,6 +138,13 @@ class RideStateMachine extends StateNotifier { // operation (try/finally) and it is intentionally NOT cleared anywhere else. bool _isFinishing = false; + // RID-3A: re-entrancy guard for telemetry batch persistence. Raised + // synchronously before the first await inside _flushBatch(); its lifetime is + // owned solely by the in-flight persistence operation (try/finally) and it is + // intentionally NOT cleared anywhere else, because clearing it while a + // persistence await is outstanding would re-open the shared-buffer race. + bool _isFlushingTelemetryBatch = false; + RideStateMachine(this.db, this.telemetryService, this.recordsService, this.audioAlertService, this.ref) : super(RideSession()) { _telemetrySub = telemetryService.telemetryStream.listen(_onTelemetryUpdate); @@ -528,16 +535,40 @@ class RideStateMachine extends StateNotifier { } Future _flushBatch() async { - if (_telemetryBatch.isNotEmpty) { + // RID-3A: telemetry batch persistence with exclusive snapshot ownership. + // + // The guard is raised synchronously, before the first `await`, so a + // competing legitimate flush (the 3 s periodic timer against the flush + // awaited by finishRide) returns without taking ownership instead of + // snapshotting the very same samples. + // + // The snapshot is removed from the live buffer BEFORE the persistence await, + // so the in-flight operation owns those samples exclusively and telemetry + // arriving during the await accumulates in a fresh live buffer. Nothing is + // removed from the live buffer after the await - that post-await + // `removeRange` is what previously allowed an overlapping flush to discard + // newer samples it had never written, while re-inserting samples another + // flush had already persisted. + // + // 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: clear inserted samples and update persisted count - _telemetryBatch.removeRange(0, batchToInsert.length); + // 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( @@ -546,6 +577,8 @@ class RideStateMachine extends StateNotifier { severity: EventSeverity.warning, ); } catch (_) {} + } finally { + _isFlushingTelemetryBatch = false; } } diff --git a/test/rid3a_telemetry_batch_ownership_test.dart b/test/rid3a_telemetry_batch_ownership_test.dart new file mode 100644 index 0000000..6bd75bb --- /dev/null +++ b/test/rid3a_telemetry_batch_ownership_test.dart @@ -0,0 +1,492 @@ +// RID-3A regression coverage: telemetry batch ownership / exactly-once buffer. +// +// Converted from the RID-3 forensic reproduction. Before this fix, two +// overlapping legitimate flushes (the 3 s periodic timer against the flush +// awaited by finishRide) shared one mutable buffer that was cleared AFTER the +// persistence await by count rather than identity. For 30 accepted samples that +// produced 30 rows representing only 20 distinct samples: indices 0-9 twice, +// indices 10-19 destroyed, indices 20-29 once - while the counters reported a +// perfect 30/30. +// +// Race windows are controlled with Completer gates on db.batch, not sleeps. The +// only real-time waits are for the production Timer.periodic(3 s) to fire, +// because that interval is a hardcoded constant with no injectable seam and +// creating one would be a production change. +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/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'; + +/// Drift's generated row class for telemetry_samples is also named +/// TelemetrySample, colliding with the telemetry model. +typedef TelemetryRow = dbgen.TelemetrySample; + +/// Gates, counts and can fail db.batch - the persistence API used by +/// RideStateMachine._flushBatch() (ride_state_machine.dart:560). +class _Rid3aAppDatabase extends AppDatabase { + _Rid3aAppDatabase() : 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++; + // Gate FIRST, then optionally fail, so a flush can be parked in-flight and + // made to fail on release - required by the ordered-restoration test. + if (gateBatch) { + final gate = Completer(); + batchGates.add(gate); + await gate.future; + } + if (batchFailuresRemaining > 0) { + batchFailuresRemaining--; + throw StateError('RID-3A simulated telemetry persistence failure'); + } + return super.batch(runInBatch); + } +} + +/// FieldTestLogger runs its OWN 3 s periodic flush() -> _db.batch +/// (field_test_logger.dart:138, :592). Those batches are indistinguishable from +/// telemetry _flushBatch() batches at the executor level, so the logger flush is +/// neutralised here to keep db.batch accounting unambiguous. Field-test event +/// persistence is not under test in RID-3A. +class _Rid3aFieldTestLogger extends FieldTestLogger { + _Rid3aFieldTestLogger(super.db); + + int stopSessionInvocations = 0; + + @override + Future flush() async {} + + @override + Future stopSession() async { + stopSessionInvocations++; + return super.stopSession(); + } +} + +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(); + + /// The production flush interval is Timer.periodic(3 s) + /// (ride_state_machine.dart:145). Waiting slightly longer guarantees exactly + /// one tick deterministically. + const tickWait = Duration(milliseconds: 3300); + + late _Rid3aAppDatabase db; + late _Rid3aFieldTestLogger logger; + late FakeTelemetryService fakeTelemetry; + late MockAudioAlertService mockAudio; + late TelemetryValidator validator; + late SharedPreferences prefs; + + final sampleBase = DateTime(2026, 9, 10, 12); + + setUp(() async { + SharedPreferences.setMockInitialValues({}); + prefs = await SharedPreferences.getInstance(); + db = _Rid3aAppDatabase(); + logger = _Rid3aFieldTestLogger(db); + validator = TelemetryValidator(); + fakeTelemetry = FakeTelemetryService( + _NullLocationService(), + _NullSensorFusionService(), + _NullCompassService(), + logger, + validator, + ); + mockAudio = MockAudioAlertService(); + }); + + 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), + ], + ); + addTearDown(container.dispose); + return container; + } + + Future startRide(ProviderContainer container) async { + final notifier = container.read(rideStateMachineProvider.notifier); + await notifier.startRide(); + return container.read(rideStateMachineProvider).dbId!; + } + + /// Emits [count] uniquely identifiable samples from index [startIndex]. + /// Identity is carried in `speed` (1000 + index) so every persisted row traces + /// back to exactly one injected sample, independent of DateTime precision. + 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; + counts[(speed - 1000.0).round()] = + (counts[(speed - 1000.0).round()] ?? 0) + 1; + } + return counts; + } + + List duplicatedIn(Map counts) => + counts.entries.where((e) => e.value > 1).map((e) => e.key).toList() + ..sort(); + + List missingOf(int total, Map counts) => [ + for (var i = 0; i < total; i++) + if ((counts[i] ?? 0) == 0) i, + ]; + + group('RID-3A: telemetry batch ownership', () { + test('single flush persists every sample exactly once, in order', + () 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); + + // One legitimate flush source, no competition. + await notifier.finishRide(); + await pumpEventQueue(times: 30); + + final rows = await persistedRows(); + final counts = countsByIndex(rows); + final session = container.read(rideStateMachineProvider); + + expect(db.batchInvocations, 1); + expect(rows, hasLength(10), reason: 'N accepted -> N durable rows'); + expect(counts.length, 10, reason: 'N distinct identities'); + expect(duplicatedIn(counts), isEmpty, reason: '0 duplicates'); + expect(missingOf(10, counts), isEmpty, reason: '0 missing'); + expect(rows.map((r) => r.speed).toList(), + [for (var i = 0; i < 10; i++) 1000.0 + i], + reason: 'correct chronological order'); + expect(rows.map((r) => r.rideId).toSet(), {rideId}, + reason: 'correct ride association'); + expect(session.totalSamplesReceived, 10); + expect(session.totalSamplesPersisted, 10, + reason: 'buffer empty after a successful final flush'); + }); + + test('overlapping periodic and finishRide flushes persist all 30 accepted ' + 'samples exactly once', () async { + final container = buildContainer(); + final notifier = container.read(rideStateMachineProvider.notifier); + final rideId = await startRide(container); + + // Batch A: indices 0..9. + emitSamples(0, 10); + await pumpEventQueue(times: 30); + + db.gateBatch = true; + // Tick 1: the periodic flush takes exclusive ownership of 0..9, removes + // them from the live buffer, and parks inside db.batch. + await Future.delayed(tickWait); + expect(db.batchInvocations, 1, reason: 'periodic flush parked in db.batch'); + final flushA = db.batchGates.single; + + // Newer telemetry (indices 10..29) arrives while A is still in flight. + // Injected BEFORE the competing flush so the live buffer is provably + // non-empty - otherwise the competing flush's exclusion could be + // attributed to an empty buffer rather than to the ownership guard. + emitSamples(10, 20); + 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. + 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'); + + // Release A: it persists exactly its owned snapshot (0..9) and nothing + // else is removed from the live buffer. + db.gateBatch = false; + 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'); + + // Legitimate later flush (the periodic timer) persists 10..29. + await Future.delayed(tickWait); + await pumpEventQueue(times: 40); + + final rows = await persistedRows(); + final counts = countsByIndex(rows); + final session = container.read(rideStateMachineProvider); + + debugPrint('==== RID-3A OVERLAP (after fix) ===='); + debugPrint('batch invocations : ${db.batchInvocations}'); + debugPrint('accepted samples : ${session.totalSamplesReceived}'); + debugPrint('durable rows : ${rows.length}'); + debugPrint('distinct identities : ${counts.length}'); + debugPrint('duplicated indices : ${duplicatedIn(counts)}'); + debugPrint('missing indices : ${missingOf(30, counts)}'); + debugPrint('received / persisted : ' + '${session.totalSamplesReceived} / ${session.totalSamplesPersisted}'); + debugPrint('rideIds : ${rows.map((r) => r.rideId).toSet()}'); + + // Previously: 30 rows / 20 distinct, indices 0-9 twice, 10-19 destroyed. + 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 preserved'); + expect(rows.map((r) => r.rideId).toSet(), {rideId}, + reason: 'every sample against the correct ride'); + expect(session.totalSamplesPersisted, 30, + reason: 'counters agree with distinct durable samples; nothing left ' + 'intentionally buffered after the final flush'); + }); + + test('telemetry arriving during an in-flight flush is persisted exactly ' + 'once, after the owned batch', () async { + final container = buildContainer(); + final rideId = await startRide(container); + + // Batch A: indices 0..4. + emitSamples(0, 5); + await pumpEventQueue(times: 30); + + db.gateBatch = true; + await Future.delayed(tickWait); + expect(db.batchInvocations, 1); + final flushA = db.batchGates.single; + + // Batch B: indices 5..9, arriving while A's persistence is parked. + emitSamples(5, 5); + await pumpEventQueue(times: 30); + expect(container.read(rideStateMachineProvider).totalSamplesReceived, 10); + + db.gateBatch = false; + flushA.complete(); + await pumpEventQueue(times: 40); + + // Legitimate later flush persists B. + await Future.delayed(tickWait); + await pumpEventQueue(times: 40); + + final rows = await persistedRows(); + final counts = countsByIndex(rows); + final session = container.read(rideStateMachineProvider); + + debugPrint('==== RID-3A DURING-FLUSH ARRIVAL (after fix) ===='); + debugPrint('durable rows : ${rows.length}'); + debugPrint('distinct identities : ${counts.length}'); + debugPrint('duplicated : ${duplicatedIn(counts)}'); + debugPrint('missing : ${missingOf(10, counts)}'); + debugPrint('order : ${rows.map((r) => r.speed).toList()}'); + + expect(rows, hasLength(10)); + expect(duplicatedIn(counts), isEmpty, reason: 'A and B each exactly once'); + expect(missingOf(10, counts), isEmpty, reason: 'nothing missing'); + expect(rows.map((r) => r.speed).toList(), + [for (var i = 0; i < 10; i++) 1000.0 + i], + reason: 'A persisted before B'); + expect(rows.map((r) => r.rideId).toSet(), {rideId}); + expect(session.totalSamplesPersisted, 10); + }); + + test('a failed flush restores its snapshot AHEAD of newer samples and a ' + 'later flush persists A B C D E F exactly once each', () async { + final container = buildContainer(); + final rideId = await startRide(container); + + // A B C = indices 0, 1, 2. + emitSamples(0, 3); + await pumpEventQueue(times: 30); + + db.gateBatch = true; + db.batchFailuresRemaining = 1; + await Future.delayed(tickWait); + expect(db.batchInvocations, 1); + final failingFlush = db.batchGates.single; + + // D E F = indices 3, 4, 5 arrive while the doomed persistence is parked. + emitSamples(3, 3); + await pumpEventQueue(times: 30); + expect(container.read(rideStateMachineProvider).totalSamplesReceived, 6); + + db.gateBatch = false; + failingFlush.complete(); // db.batch now throws for this snapshot + await pumpEventQueue(times: 40); + + var session = container.read(rideStateMachineProvider); + expect(session.totalSamplesPersisted, 0, + reason: 'a failed persistence must not increment the success counter'); + expect(await persistedRows(), isEmpty, reason: 'nothing durable yet'); + expect(session.state, RideState.riding, + reason: 'telemetry recording continues so a later flush can recover'); + + // Existing periodic timer provides the legitimate retry. No automatic + // retry worker is introduced. + await Future.delayed(tickWait); + await pumpEventQueue(times: 40); + + final rows = await persistedRows(); + final counts = countsByIndex(rows); + session = container.read(rideStateMachineProvider); + + debugPrint('==== RID-3A FAILURE + ORDERED RESTORATION (after fix) ===='); + debugPrint('batch invocations : ${db.batchInvocations}'); + debugPrint('durable rows : ${rows.length}'); + debugPrint('distinct identities : ${counts.length}'); + debugPrint('duplicated : ${duplicatedIn(counts)}'); + debugPrint('missing : ${missingOf(6, counts)}'); + debugPrint('order : ${rows.map((r) => r.speed).toList()}'); + debugPrint('received / persisted : ' + '${session.totalSamplesReceived} / ${session.totalSamplesPersisted}'); + + expect(rows, hasLength(6)); + expect(duplicatedIn(counts), isEmpty, reason: 'retry causes no duplicates'); + expect(missingOf(6, counts), isEmpty, reason: 'no rows missing'); + expect(rows.map((r) => r.speed).toList(), + [1000.0, 1001.0, 1002.0, 1003.0, 1004.0, 1005.0], + reason: 'failed snapshot A B C restored AHEAD of newer D E F, both ' + 'internal orders preserved'); + expect(rows.map((r) => r.rideId).toSet(), {rideId}); + expect(session.totalSamplesPersisted, 6, + reason: 'counter reflects only the successfully persisted snapshot'); + }); + }); +}