diff --git a/lib/features/rides/ride_state_machine.dart b/lib/features/rides/ride_state_machine.dart index ddce8be..e9c7592 100644 --- a/lib/features/rides/ride_state_machine.dart +++ b/lib/features/rides/ride_state_machine.dart @@ -121,6 +121,14 @@ class RideStateMachine extends StateNotifier { final Map _lastEventTimes = {}; static const Duration _eventCooldown = Duration(seconds: 5); + // RID-1: re-entrancy guard for ride start. Raised synchronously before the + // first await inside startRide() so concurrent callers cannot each create + // their own ride row and cloudRideId. Its lifetime is owned exclusively by + // the in-flight start operation (try/finally); it is intentionally NOT + // cleared anywhere else, because clearing it while a start is still awaiting + // would re-open the duplicate-start race. + bool _isStarting = false; + RideStateMachine(this.db, this.telemetryService, this.recordsService, this.audioAlertService, this.ref) : super(RideSession()) { _telemetrySub = telemetryService.telemetryStream.listen(_onTelemetryUpdate); @@ -199,9 +207,32 @@ class RideStateMachine extends StateNotifier { ); } + /// RID-1: idempotent ride start. + /// + /// The guard is raised synchronously, before the first `await`, so that + /// concurrent callers — the 20 Hz motion auto-start in [_onTelemetryUpdate] + /// and rapid taps on the dashboard Start buttons — collapse into a single + /// start operation instead of each minting its own ride row, `cloudRideId` + /// and orphaned open ride. + /// + /// The guard is released in `finally` so a genuinely failed start never + /// leaves the machine permanently unable to record, and a later legitimate + /// retry still succeeds. Frozen semantics (auto-start threshold, PR gating, + /// kickstand isolation, recovery, `cloudRideId` generation) are untouched. Future startRide() async { + if (_isStarting) return; if (state.state == RideState.riding) return; + _isStarting = true; + try { + await _performStartRide(); + } finally { + _isStarting = false; + } + } + /// Performs the actual ride start. Guarded exclusively by [startRide] and + /// must never be invoked directly. + Future _performStartRide() async { try { ref.read(fieldTestLoggerProvider).log( eventType: FieldTestEventType.rideStartRequested, diff --git a/lib/services/sync/sync_observer.dart b/lib/services/sync/sync_observer.dart index b6959c6..a5c5010 100644 --- a/lib/services/sync/sync_observer.dart +++ b/lib/services/sync/sync_observer.dart @@ -1,3 +1,5 @@ +import 'dart:async'; + import 'package:flutter/foundation.dart'; import 'package:flutter_riverpod/flutter_riverpod.dart'; import '../../features/rides/ride_state_machine.dart'; @@ -24,10 +26,39 @@ class SyncObserver { if (_lastEnqueuedRideId != completedRideId) { _lastEnqueuedRideId = completedRideId; debugPrint('[SyncObserver] Ride #$completedRideId completed, enqueuing to SyncManager'); - _syncManager.enqueueRide(completedRideId); + // RID-1: the enqueue Future is now handled explicitly instead of being + // left naked. SyncManager.enqueueRide can throw (SyncOutbox raises + // StateError at capacity and ArgumentError on invalid input), and an + // unhandled async exception here silently discarded the completed + // ride's durable sync intent with no diagnostic. + // + // _enqueueCompletedRide is still invoked synchronously at this point, + // so the callee's synchronous prefix runs before this method returns + // and the outbox's in-memory mutation stays observable to callers. + unawaited(_enqueueCompletedRide(completedRideId)); } } } + + /// Awaits the enqueue and reports failure instead of swallowing it. + /// + /// Diagnostics intentionally carry only the exception's runtime type, never + /// its message or payload, so tokens, UIDs and ride content cannot leak into + /// logs. Output is bounded to at most one line per completed ride because + /// [_lastEnqueuedRideId] suppresses repeat notifications for the same ride. + /// + /// Durability note: if this enqueue fails the automatic sync intent for that + /// ride is lost. No retry or reconciliation subsystem is added here by + /// design; see the RID-1 report for the documented follow-up risk. + Future _enqueueCompletedRide(int rideId) async { + try { + await _syncManager.enqueueRide(rideId); + } catch (error) { + debugPrint( + '[SyncObserver] Enqueue failed for ride #$rideId class=${error.runtimeType}', + ); + } + } } /// Riverpod provider establishing the active SyncObserver listener diff --git a/test/rid1_ride_start_idempotency_test.dart b/test/rid1_ride_start_idempotency_test.dart new file mode 100644 index 0000000..9195cfc --- /dev/null +++ b/test/rid1_ride_start_idempotency_test.dart @@ -0,0 +1,333 @@ +// RID-1 regression coverage: idempotent ride start. +// +// These tests reproduce the verified duplicate-ride defect deterministically. +// The race window is controlled with a Completer-gated database rather than +// sleeps, so no test here is timing-fragile. +// +// The gate/failure seam lives entirely in a test-only AppDatabase subclass +// injected through the existing AppDatabase.forTesting constructor. No +// production architecture was changed to make these tests possible. +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:motorcycle_dash/database/database.dart' hide TelemetrySample; +import 'package:motorcycle_dash/models/telemetry_sample.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'; + +/// Deterministic seam over the production database. +/// +/// * [startRideInvocations] proves "at most one db.startRide() may be active". +/// * [gate] parks the first insert in flight so concurrent callers can be +/// exercised without any timing assumption. +/// * [failuresRemaining] injects a genuine start failure and then recovers. +class _Rid1AppDatabase extends AppDatabase { + _Rid1AppDatabase() : super.forTesting(NativeDatabase.memory()); + + int startRideInvocations = 0; + Completer? gate; + int failuresRemaining = 0; + + @override + Future startRide({ + DateTime? startTime, + int? motorcycleId, + int? motorcycleYear, + String? motorcycleMake, + String? motorcycleModel, + String? motorcycleVariant, + }) async { + startRideInvocations++; + if (failuresRemaining > 0) { + failuresRemaining--; + throw StateError('RID-1 simulated database start failure'); + } + final pendingGate = gate; + if (pendingGate != null) await pendingGate.future; + return super.startRide( + startTime: startTime, + motorcycleId: motorcycleId, + motorcycleYear: motorcycleYear, + motorcycleMake: motorcycleMake, + motorcycleModel: motorcycleModel, + motorcycleVariant: motorcycleVariant, + ); + } +} + +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(); + + late _Rid1AppDatabase db; + late FakeTelemetryService fakeTelemetry; + late MockAudioAlertService mockAudio; + late FieldTestLogger logger; + late TelemetryValidator validator; + late SharedPreferences prefs; + + setUp(() async { + SharedPreferences.setMockInitialValues({}); + prefs = await SharedPreferences.getInstance(); + db = _Rid1AppDatabase(); + logger = FieldTestLogger(db); + validator = TelemetryValidator(); + fakeTelemetry = FakeTelemetryService( + _NullLocationService(), + _NullSensorFusionService(), + _NullCompassService(), + logger, + validator, + ); + mockAudio = MockAudioAlertService(); + }); + + tearDown(() async { + // Never leave a parked gate behind or the suite would hang. + final pendingGate = db.gate; + if (pendingGate != null && !pendingGate.isCompleted) pendingGate.complete(); + db.gate = null; + fakeTelemetry.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; + } + + TelemetrySample movingSample(int index, {double speed = 25.0}) { + // 50 ms spacing mirrors the authoritative 20 Hz acquisition cadence. + return TelemetrySample( + timestamp: + DateTime(2026, 9, 10, 12).add(Duration(milliseconds: index * 50)), + latitude: 34.2570, + longitude: -85.1647, + speed: speed, + heading: 90.0, + gpsAccuracy: 3.0, + ); + } + + group('RID-1: idempotent ride start', () { + test('20 Hz ready-state auto-start creates exactly one ride', () async { + final container = buildContainer(); + final notifier = container.read(rideStateMachineProvider.notifier); + final gate = Completer(); + db.gate = gate; + + notifier.setReady(); + expect(container.read(rideStateMachineProvider).state, RideState.ready); + + // One second of authoritative 20 Hz telemetry. None of the resulting + // startRide() triggers is awaited, exactly as the production + // motion auto-start path does. + for (var i = 0; i < 20; i++) { + fakeTelemetry.emitSample(movingSample(i)); + } + + // Deliver every queued stream event. The first db.startRide() is + // deterministically parked on the gate, so this is the real race window. + await pumpEventQueue(times: 50); + + expect(db.startRideInvocations, 1, + reason: 'at most one db.startRide() operation may be active'); + + gate.complete(); + await pumpEventQueue(times: 50); + + final rides = await db.select(db.rides).get(); + expect(rides, hasLength(1), reason: 'exactly one ride row'); + expect(rides.map((r) => r.cloudRideId).toSet(), hasLength(1), + reason: 'exactly one cloudRideId minted'); + + final session = container.read(rideStateMachineProvider); + expect(session.state, RideState.riding); + expect(session.dbId, rides.single.id, + reason: 'session.dbId must resolve to that single row'); + }); + + test('two concurrent startRide() calls create exactly one ride', () async { + final container = buildContainer(); + final notifier = container.read(rideStateMachineProvider.notifier); + final gate = Completer(); + db.gate = gate; + + // Exactly what two rapid taps on the dashboard Start button do: fire + // startRide() twice without awaiting. + final first = notifier.startRide(); + final second = notifier.startRide(); + + await pumpEventQueue(times: 50); + expect(db.startRideInvocations, 1, reason: 'exactly one database start'); + + gate.complete(); + await Future.wait([first, second]); + + final rides = await db.select(db.rides).get(); + expect(rides, hasLength(1)); + expect(rides.map((r) => r.cloudRideId).toSet(), hasLength(1)); + + final session = container.read(rideStateMachineProvider); + expect(session.state, RideState.riding, + reason: 'one resulting riding session'); + expect(session.dbId, rides.single.id); + }); + + test('failed database start does not report riding, clears the guard, ' + 'and a later legitimate retry succeeds', () async { + final container = buildContainer(); + final notifier = container.read(rideStateMachineProvider.notifier); + db.failuresRemaining = 1; + + await expectLater(notifier.startRide(), throwsStateError); + + var session = container.read(rideStateMachineProvider); + expect(session.state, isNot(RideState.riding), + reason: 'state must not falsely become riding'); + expect(session.dbId, isNull); + expect(await db.select(db.rides).get(), isEmpty, + reason: 'a failed start must not leave a partial row'); + + // Retry the genuine start. Succeeding here proves the in-flight guard was + // released by the finally block rather than staying stuck. + await notifier.startRide(); + + session = container.read(rideStateMachineProvider); + expect(session.state, RideState.riding); + expect(session.dbId, isNotNull); + + final rides = await db.select(db.rides).get(); + expect(rides, hasLength(1), reason: 'no duplicate rows from the retry'); + expect(db.startRideInvocations, 2); + }); + + test('after a concurrent-start burst only one ride is left open, so ' + 'recovery cannot manufacture phantom rides', () async { + final container = buildContainer(); + final notifier = container.read(rideStateMachineProvider.notifier); + final gate = Completer(); + db.gate = gate; + + notifier.setReady(); + for (var i = 0; i < 20; i++) { + fakeTelemetry.emitSample(movingSample(i)); + } + await pumpEventQueue(times: 50); + gate.complete(); + await pumpEventQueue(times: 50); + + final rides = await db.select(db.rides).get(); + + // Mirrors the selection criterion used by _recoverInterruptedRides() + // (rides whose endTime is null). Recovery behaviour itself is unchanged. + final openRides = rides.where((r) => r.endTime == null).toList(); + expect(openRides, hasLength(1), + reason: 'exactly one open active ride after the burst'); + expect(rides, hasLength(1), + reason: 'no additional empty orphan ride rows'); + }); + + test('frozen auto-start threshold is preserved at 1.5 MPH', () async { + final container = buildContainer(); + final notifier = container.read(rideStateMachineProvider.notifier); + notifier.setReady(); + + // Below the frozen threshold: must not auto-start. + fakeTelemetry.emitSample(movingSample(0, speed: 1.4)); + await pumpEventQueue(times: 20); + expect(container.read(rideStateMachineProvider).state, RideState.ready); + expect(db.startRideInvocations, 0); + + // At the frozen threshold: must auto-start. + fakeTelemetry.emitSample(movingSample(1, speed: 1.5)); + await pumpEventQueue(times: 50); + expect(container.read(rideStateMachineProvider).state, RideState.riding); + expect(db.startRideInvocations, 1); + }); + }); +} diff --git a/test/rid1_sync_observer_hardening_test.dart b/test/rid1_sync_observer_hardening_test.dart new file mode 100644 index 0000000..9ff1f06 --- /dev/null +++ b/test/rid1_sync_observer_hardening_test.dart @@ -0,0 +1,185 @@ +// RID-1 regression coverage: SyncObserver completed-ride enqueue error handling. +// +// Before this hardening the observer called SyncManager.enqueueRide() as a +// naked Future. Because SyncOutbox.enqueue() throws StateError at capacity and +// ArgumentError on invalid input, a failing enqueue surfaced as an unhandled +// async exception and silently discarded the ride's durable sync intent. +// +// flutter_test fails a test on any unhandled async error, so these tests are a +// genuine regression: they fail against the pre-fix observer. +import 'package:flutter/foundation.dart'; +import 'package:flutter_test/flutter_test.dart'; +import 'package:shared_preferences/shared_preferences.dart'; +import 'package:motorcycle_dash/models/recording_health_status.dart'; +import 'package:motorcycle_dash/features/rides/ride_state_machine.dart'; +import 'package:motorcycle_dash/services/sync/sync_manager.dart'; +import 'package:motorcycle_dash/services/sync/sync_models.dart'; +import 'package:motorcycle_dash/services/sync/sync_observer.dart'; +import 'package:motorcycle_dash/services/sync/sync_outbox.dart'; + +/// SyncManager seam that can fail the next N enqueue attempts, then delegate to +/// the real implementation. Test-only; no production change was required. +class _FailingSyncManager extends SyncManager { + _FailingSyncManager(super.outbox); + + int enqueueAttempts = 0; + int failuresRemaining = 0; + + /// Payload that must never reach diagnostics. + static const String secretPayload = 'RID-1-SENSITIVE-TOKEN-do-not-log'; + + @override + Future enqueueRide( + int localRideId, { + String? targetProviderId, + bool autoProcess = true, + }) async { + enqueueAttempts++; + if (failuresRemaining > 0) { + failuresRemaining--; + throw StateError(secretPayload); + } + return super.enqueueRide( + localRideId, + targetProviderId: targetProviderId, + autoProcess: autoProcess, + ); + } +} + +void main() { + TestWidgetsFlutterBinding.ensureInitialized(); + + late SharedPreferences prefs; + late List logLines; + late DebugPrintCallback originalDebugPrint; + + setUp(() async { + SharedPreferences.setMockInitialValues({}); + prefs = await SharedPreferences.getInstance(); + logLines = []; + originalDebugPrint = debugPrint; + debugPrint = (String? message, {int? wrapWidth}) { + if (message != null) logLines.add(message); + }; + }); + + tearDown(() { + debugPrint = originalDebugPrint; + }); + + RideSession riding(int dbId) => RideSession( + dbId: dbId, + state: RideState.riding, + healthStatus: RecordingHealthStatus.recordingActive, + ); + + RideSession finishedOf(RideSession previous) => previous.copyWith( + state: RideState.finished, + healthStatus: RecordingHealthStatus.recordingComplete, + ); + + List failureLines() => + logLines.where((l) => l.contains('Enqueue failed')).toList(); + + group('RID-1: SyncObserver enqueue hardening', () { + test('enqueue failure is handled and does not escape as an unhandled ' + 'async error', () async { + final outbox = SyncOutbox(prefs); + final manager = _FailingSyncManager(outbox)..failuresRemaining = 1; + addTearDown(manager.dispose); + final observer = SyncObserver(manager); + + final previous = riding(900); + observer.onRideSessionChanged(previous, finishedOf(previous)); + await pumpEventQueue(times: 20); + + expect(manager.enqueueAttempts, 1); + expect(outbox.count, 0, + reason: 'a failed enqueue must not create an outbox item'); + expect(failureLines(), hasLength(1), + reason: 'failure must be reported, not silently swallowed'); + }); + + test('diagnostics carry only the exception class, never its payload', + () async { + final outbox = SyncOutbox(prefs); + final manager = _FailingSyncManager(outbox)..failuresRemaining = 1; + addTearDown(manager.dispose); + final observer = SyncObserver(manager); + + final previous = riding(901); + observer.onRideSessionChanged(previous, finishedOf(previous)); + await pumpEventQueue(times: 20); + + expect(failureLines().single, contains('StateError')); + expect(logLines.join('\n'), + isNot(contains(_FailingSyncManager.secretPayload)), + reason: 'no token/secret/data may leak into diagnostics'); + }); + + test('repeat completion notifications keep enqueue and logging bounded ' + 'to one', () async { + final outbox = SyncOutbox(prefs); + final manager = _FailingSyncManager(outbox)..failuresRemaining = 5; + addTearDown(manager.dispose); + final observer = SyncObserver(manager); + + final previous = riding(902); + final completed = finishedOf(previous); + observer.onRideSessionChanged(previous, completed); + observer.onRideSessionChanged(previous, completed); + observer.onRideSessionChanged(completed, completed); + await pumpEventQueue(times: 20); + + expect(manager.enqueueAttempts, 1, + reason: 'no duplicate enqueue behaviour introduced'); + expect(failureLines(), hasLength(1), reason: 'bounded error logging'); + }); + + test('a later legitimate completion still enqueues after a failure', + () async { + final outbox = SyncOutbox(prefs); + final manager = _FailingSyncManager(outbox)..failuresRemaining = 1; + addTearDown(manager.dispose); + final observer = SyncObserver(manager); + + final failed = riding(903); + observer.onRideSessionChanged(failed, finishedOf(failed)); + await pumpEventQueue(times: 20); + expect(outbox.count, 0); + + final later = riding(904); + observer.onRideSessionChanged(later, finishedOf(later)); + await pumpEventQueue(times: 20); + + expect(manager.enqueueAttempts, 2); + expect(outbox.getItem(904, 'fake'), isNotNull, + reason: 'the observer must stay functional after a handled failure'); + // Durable persistence is what the outbox guarantees. The item is + // subsequently marked failed by processQueue() because this unit + // registers no 'fake' provider - that is existing SyncManager behaviour + // and outside RID-1 scope, so only durable presence is asserted here. + expect(prefs.getString(outbox.storageKey), contains('"localRideId":904'), + reason: 'the recovered enqueue must reach durable storage'); + }); + + test('successful enqueue remains observable synchronously', () { + // Existing contract relied on by cloud_sync_hardening_test.dart, which + // asserts outbox.count without awaiting. The RID-1 wrapper must not + // delay the callee's synchronous prefix. + final outbox = SyncOutbox(prefs); + final manager = SyncManager(outbox); + addTearDown(manager.dispose); + final observer = SyncObserver(manager); + + final previous = riding(905); + observer.onRideSessionChanged(previous, finishedOf(previous)); + + expect(outbox.count, equals(1)); + + observer.onRideSessionChanged(previous, finishedOf(previous)); + expect(outbox.count, equals(1)); + }); + }); +}