From 0282a2972fd2a36cecac107e2d5e733d776f91bf Mon Sep 17 00:00:00 2001 From: Developer Date: Fri, 18 Sep 2026 18:34:58 -0400 Subject: [PATCH] fix(telemetry): accurately track dropped telemetry samples under queue bounds in ride state and field test summary --- lib/features/rides/ride_state_machine.dart | 404 ++++++++----- .../field_test/field_test_logger.dart | 28 +- test/field_test_logger_test.dart | 120 +++- .../rid3a_telemetry_batch_ownership_test.dart | 538 ++++++++++++++++-- 4 files changed, 871 insertions(+), 219 deletions(-) diff --git a/lib/features/rides/ride_state_machine.dart b/lib/features/rides/ride_state_machine.dart index 2df4ae7..5681796 100644 --- a/lib/features/rides/ride_state_machine.dart +++ b/lib/features/rides/ride_state_machine.dart @@ -36,7 +36,7 @@ class RideSession { final double maxAccelG; final double maxBrakingG; final DateTime? startTime; - + // Phase 4 Stats final int movingTimeSeconds; final int stoppedTimeSeconds; @@ -105,7 +105,8 @@ class RideSession { elevationLoss: elevationLoss ?? this.elevationLoss, averageSpeedMph: averageSpeedMph ?? this.averageSpeedMph, totalSamplesReceived: totalSamplesReceived ?? this.totalSamplesReceived, - totalSamplesPersisted: totalSamplesPersisted ?? this.totalSamplesPersisted, + totalSamplesPersisted: + totalSamplesPersisted ?? this.totalSamplesPersisted, ); } } @@ -116,20 +117,20 @@ class RideStateMachine extends StateNotifier { final RecordsService recordsService; final AudioAlertService audioAlertService; final Ref ref; - + StreamSubscription? _telemetrySub; - + TelemetrySample? _lastSample; double _accumulatedDistance = 0; // meters internally double _movingTimeAccumulator = 0.0; // seconds internally double _stoppedTimeAccumulator = 0.0; // seconds internally - + // Batching final List _telemetryBatch = []; Timer? _batchWriteTimer; Timer? _watchdogTimer; DateTime? _lastTelemetryReceivedTime; - + // Event Debouncing (Store last trigger time per event type) final Map _lastEventTimes = {}; static const Duration _eventCooldown = Duration(seconds: 5); @@ -181,14 +182,30 @@ class RideStateMachine extends StateNotifier { // retryable rather than transitioning to finished. static const int _maxFinalDrainPasses = 4; - RideStateMachine(this.db, this.telemetryService, this.recordsService, this.audioAlertService, this.ref) : super(RideSession()) { + // Explicit per-ride counter of telemetry samples discarded due to queue capacity bounds. + int _telemetrySamplesLost = 0; + int get telemetrySamplesLost => _telemetrySamplesLost; + + RideStateMachine( + this.db, + this.telemetryService, + this.recordsService, + this.audioAlertService, + this.ref, + ) : super(RideSession()) { _telemetrySub = telemetryService.telemetryStream.listen(_onTelemetryUpdate); - + // Write to DB every 3 seconds for tight crash-durability - _batchWriteTimer = Timer.periodic(const Duration(seconds: 3), (_) => _runScheduledFlush()); + _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()); + _watchdogTimer = Timer.periodic( + const Duration(seconds: 2), + (_) => _checkWatchdog(), + ); // Recover any unfinished rides from previous sessions _recoverInterruptedRides(); @@ -278,25 +295,34 @@ class RideStateMachine extends StateNotifier { // is also what prevents monotonic freeze-blindness: when samples stop, the // monotonic reference stops advancing, but this term still goes stale. final accepted = _lastAcceptedTelemetryTime; - final telemetryAgeMs = - accepted == null ? null : now.difference(accepted).inMilliseconds; + final telemetryAgeMs = accepted == null + ? null + : now.difference(accepted).inMilliseconds; // Fail closed on a backwards clock jump: negative age is not freshness. - final telemetryActive = telemetryAgeMs != null && + final telemetryActive = + telemetryAgeMs != null && telemetryAgeMs >= 0 && telemetryAgeMs <= sensorBound; // B. GPS and D. IMU evidence: monotonic, presence-proven. - final gpsOperational = - _isFresh(_monotonicAgeMs(_lastGpsEvidenceMonotonicMs), sensorBound); - final sensorActive = - _isFresh(_monotonicAgeMs(_lastImuEvidenceMonotonicMs), sensorBound); + final gpsOperational = _isFresh( + _monotonicAgeMs(_lastGpsEvidenceMonotonicMs), + sensorBound, + ); + final sensorActive = _isFresh( + _monotonicAgeMs(_lastImuEvidenceMonotonicMs), + sensorBound, + ); // E. Database evidence: advances ONLY after a real successful telemetry // commit (the RID-3B-2 commit point). A scheduled flush is not evidence, and // a failed pre-commit batch does not advance it. - final dbOperational = !_dbPersistenceFailed && - _isFresh(_monotonicAgeMs(_lastDbCommitMonotonicMs), - RecordingHealthContract.databaseWriteStaleAfterMs); + final dbOperational = + !_dbPersistenceFailed && + _isFresh( + _monotonicAgeMs(_lastDbCommitMonotonicMs), + RecordingHealthContract.databaseWriteStaleAfterMs, + ); final previous = state.health; final health = previous.copyWith( @@ -335,11 +361,14 @@ class RideStateMachine extends StateNotifier { if (stalled && previousMisses == 0) { // First detection of this stall episode: logged once, as before. try { - ref.read(fieldTestLoggerProvider).log( - eventType: FieldTestEventType.telemetryGap, - severity: EventSeverity.warning, - message: 'Watchdog: Telemetry pipeline stalled ($elapsedSec seconds without samples)', - ); + ref + .read(fieldTestLoggerProvider) + .log( + eventType: FieldTestEventType.telemetryGap, + severity: EventSeverity.warning, + message: + 'Watchdog: Telemetry pipeline stalled ($elapsedSec seconds without samples)', + ); } catch (_) {} } @@ -357,24 +386,30 @@ class RideStateMachine extends StateNotifier { Future _recoverInterruptedRides() async { try { - final unclosedRides = await (db.select(db.rides) - ..where((r) => r.endTime.isNull()) - ).get(); + final unclosedRides = await (db.select( + db.rides, + )..where((r) => r.endTime.isNull())).get(); for (final r in unclosedRides) { // Find last recorded telemetry sample for this ride to determine actual end time - final lastSample = await (db.select(db.telemetrySamples) - ..where((t) => t.rideId.equals(r.id)) - ..orderBy([(t) => drift.OrderingTerm(expression: t.timestamp, mode: drift.OrderingMode.desc)]) - ..limit(1) - ).getSingleOrNull(); + final lastSample = + await (db.select(db.telemetrySamples) + ..where((t) => t.rideId.equals(r.id)) + ..orderBy([ + (t) => drift.OrderingTerm( + expression: t.timestamp, + mode: drift.OrderingMode.desc, + ), + ]) + ..limit(1)) + .getSingleOrNull(); - final calculatedEndTime = lastSample?.timestamp ?? r.startTime.add(Duration(seconds: r.movingTime + r.stoppedTime)); + final calculatedEndTime = + lastSample?.timestamp ?? + r.startTime.add(Duration(seconds: r.movingTime + r.stoppedTime)); await (db.update(db.rides)..where((t) => t.id.equals(r.id))).write( - RidesCompanion( - endTime: drift.Value(calculatedEndTime), - ), + RidesCompanion(endTime: drift.Value(calculatedEndTime)), ); try { @@ -382,7 +417,8 @@ class RideStateMachine extends StateNotifier { logger.log( eventType: FieldTestEventType.databaseRecovery, severity: EventSeverity.critical, - message: 'Interrupted ride #${r.id} recovered after unexpected exit', + message: + 'Interrupted ride #${r.id} recovered after unexpected exit', rideId: r.id, ); await logger.generateAndSaveSummary(r.id, wasRecovered: true); @@ -425,11 +461,13 @@ class RideStateMachine extends StateNotifier { /// must never be invoked directly. Future _performStartRide() async { try { - ref.read(fieldTestLoggerProvider).log( - eventType: FieldTestEventType.rideStartRequested, - severity: EventSeverity.info, - message: 'Ride start requested by user or motion trigger', - ); + ref + .read(fieldTestLoggerProvider) + .log( + eventType: FieldTestEventType.rideStartRequested, + severity: EventSeverity.info, + message: 'Ride start requested by user or motion trigger', + ); } catch (_) {} final defaultBike = ref.read(defaultMotorcycleProvider); @@ -448,6 +486,7 @@ class RideStateMachine extends StateNotifier { // IMU or the 3-second batch; health warms up automatically as real evidence // arrives and _evaluateRecordingHealth promotes the status. _resetHealthEvidence(); + _telemetrySamplesLost = 0; state = state.copyWith( dbId: id, state: RideState.riding, @@ -516,11 +555,13 @@ class RideStateMachine extends StateNotifier { WakelockPlus.disable().catchError((_) {}); } catch (_) {} try { - ref.read(fieldTestLoggerProvider).log( - eventType: FieldTestEventType.appPaused, - severity: EventSeverity.info, - message: 'Ride session paused', - ); + ref + .read(fieldTestLoggerProvider) + .log( + eventType: FieldTestEventType.appPaused, + severity: EventSeverity.info, + message: 'Ride session paused', + ); } catch (_) {} } @@ -528,7 +569,10 @@ class RideStateMachine extends StateNotifier { // RID-3C-1: resuming does not by itself prove recording health. Evidence ages // out monotonically across the pause, so the next accepted sample // re-establishes truthfully whether the ride is active or degraded. - state = state.copyWith(state: RideState.riding, healthStatus: RecordingHealthStatus.recordingArmed); + state = state.copyWith( + state: RideState.riding, + healthStatus: RecordingHealthStatus.recordingArmed, + ); _lastSample = null; // reset to avoid jumping _lastTelemetryReceivedTime = DateTime.now(); // RID-3C-1: re-publish truthful health for the resumed session. @@ -537,11 +581,13 @@ class RideStateMachine extends StateNotifier { WakelockPlus.enable().catchError((_) {}); } catch (_) {} try { - ref.read(fieldTestLoggerProvider).log( - eventType: FieldTestEventType.appResumed, - severity: EventSeverity.info, - message: 'Ride session resumed', - ); + ref + .read(fieldTestLoggerProvider) + .log( + eventType: FieldTestEventType.appResumed, + severity: EventSeverity.info, + message: 'Ride session resumed', + ); } catch (_) {} } @@ -581,15 +627,19 @@ class RideStateMachine extends StateNotifier { /// [finishRide] and must never be invoked directly. Future _performFinishRide() async { if (state.dbId != null) { - state = state.copyWith(healthStatus: RecordingHealthStatus.recordingFinalizing); + state = state.copyWith( + healthStatus: RecordingHealthStatus.recordingFinalizing, + ); try { - ref.read(fieldTestLoggerProvider).log( - eventType: FieldTestEventType.rideFinishRequested, - severity: EventSeverity.info, - message: 'Ride finish requested for ride #${state.dbId}', - rideId: state.dbId, - ); + ref + .read(fieldTestLoggerProvider) + .log( + eventType: FieldTestEventType.rideFinishRequested, + severity: EventSeverity.info, + message: 'Ride finish requested for ride #${state.dbId}', + rideId: state.dbId, + ); } catch (_) {} // RID-3B-1: the explicit-Stop durability barrier replaces the single @@ -597,7 +647,7 @@ class RideStateMachine extends StateNotifier { // 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!, RidesCompanion( @@ -616,12 +666,14 @@ class RideStateMachine extends StateNotifier { ); try { - ref.read(fieldTestLoggerProvider).log( - eventType: FieldTestEventType.rideFinalized, - severity: EventSeverity.info, - message: 'Ride record #${state.dbId} finalized in database', - rideId: state.dbId, - ); + ref + .read(fieldTestLoggerProvider) + .log( + eventType: FieldTestEventType.rideFinalized, + severity: EventSeverity.info, + message: 'Ride record #${state.dbId} finalized in database', + rideId: state.dbId, + ); } catch (_) {} // Check for personal records @@ -634,7 +686,7 @@ class RideStateMachine extends StateNotifier { longestRideMiles: state.tripDistanceMiles, greatestElevationGain: state.elevationGain, ); - + await recordsService.checkAndSaveRecords(runStats, (recordName, value) { audioAlertService.speak('New record: $recordName'); }); @@ -644,11 +696,13 @@ class RideStateMachine extends StateNotifier { try { final logger = ref.read(fieldTestLoggerProvider); if (logger.isSessionActive) { + logger.telemetrySamplesLost = _telemetrySamplesLost; await logger.stopSession(); logger.log( eventType: FieldTestEventType.rideSummaryCreated, severity: EventSeverity.info, - message: 'Field test summary generated and saved for ride #${state.dbId}', + message: + 'Field test summary generated and saved for ride #${state.dbId}', rideId: state.dbId, ); } @@ -713,23 +767,29 @@ class RideStateMachine extends StateNotifier { try { if (sample.quality != null) { - ref.read(fieldTestLoggerProvider).processTelemetrySample(sample, sample.quality!); + ref + .read(fieldTestLoggerProvider) + .processTelemetrySample(sample, sample.quality!); } } catch (_) {} } - + static const int _maxBatchCapacity = 1000; void _queueForBatch(TelemetrySample sample) { if (state.dbId == null) return; - + // Bound queue to prevent memory leaks during extended DB locks if (_telemetryBatch.length >= _maxBatchCapacity) { _telemetryBatch.removeAt(0); + _telemetrySamplesLost++; try { - ref.read(fieldTestLoggerProvider).log( + final logger = ref.read(fieldTestLoggerProvider); + logger.recordTelemetryLoss(1); + logger.log( eventType: FieldTestEventType.databaseQueueOverflow, - message: 'Telemetry batch buffer reached capacity (1000); dropping oldest sample', + message: + 'Telemetry batch buffer reached capacity (1000); dropping oldest sample', severity: EventSeverity.warning, ); } catch (_) {} @@ -752,12 +812,14 @@ class RideStateMachine extends StateNotifier { accelZ: drift.Value(sample.accelZ), longitudinalG: drift.Value(sample.longitudinalG), lateralG: drift.Value(sample.lateralG), - ) + ), ); - state = state.copyWith(totalSamplesReceived: state.totalSamplesReceived + 1); + 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 @@ -840,7 +902,9 @@ class RideStateMachine extends StateNotifier { // 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); + state = state.copyWith( + healthStatus: RecordingHealthStatus.recordingActive, + ); throw StateError( 'Ride finalization aborted: telemetry durability could not be established', ); @@ -873,6 +937,28 @@ class RideStateMachine extends StateNotifier { // arrived during the await, so a later legitimate flush can retry them. // No automatic retry worker is added. _telemetryBatch.insertAll(0, batchToInsert); + // Enforce live queue capacity after failed-snapshot restoration by + // dropping the excess oldest samples (from the front), preserving the + // existing oldest-first overflow policy and queue ordering. + final excess = _telemetryBatch.length - _maxBatchCapacity; + if (excess > 0) { + _telemetryBatch.removeRange(0, excess); + _telemetrySamplesLost += excess; + if (mounted) { + try { + final logger = ref.read(fieldTestLoggerProvider); + logger.recordTelemetryLoss(excess); + for (var i = 0; i < excess; i++) { + logger.log( + eventType: FieldTestEventType.databaseQueueOverflow, + message: + 'Telemetry batch buffer reached capacity (1000); dropping oldest sample', + severity: EventSeverity.warning, + ); + } + } catch (_) {} + } + } // RID-3C-1: a pre-commit failure means this telemetry is NOT durable, so // database health evidence must not advance and is marked non-operational // until a real commit succeeds. @@ -882,11 +968,13 @@ class RideStateMachine extends StateNotifier { // 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, - ); + ref + .read(fieldTestLoggerProvider) + .log( + eventType: FieldTestEventType.databaseQueueWarning, + message: 'Database batch write retry: $e', + severity: EventSeverity.warning, + ); } catch (_) {} } return false; @@ -917,8 +1005,11 @@ class RideStateMachine extends StateNotifier { _lastDbCommitMonotonicMs = _latestObservedMonotonicMs; _dbPersistenceFailed = false; state = state.copyWith( - totalSamplesPersisted: state.totalSamplesPersisted + batchToInsert.length, - health: state.health.copyWith(lastDatabaseWriteTime: DateTime.now()), + totalSamplesPersisted: + state.totalSamplesPersisted + batchToInsert.length, + health: state.health.copyWith( + lastDatabaseWriteTime: DateTime.now(), + ), ); _evaluateRecordingHealth(); } catch (_) { @@ -993,7 +1084,9 @@ class RideStateMachine extends StateNotifier { // Incrementally persist ongoing ride summary stats to SQLite so crashes/terminations never lose stats if (state.dbId != null && state.state == RideState.riding) { try { - await (db.update(db.rides)..where((t) => t.id.equals(state.dbId!))).write( + await (db.update( + db.rides, + )..where((t) => t.id.equals(state.dbId!))).write( RidesCompanion( distanceMiles: drift.Value(state.tripDistanceMiles), topSpeedMph: drift.Value(state.topSpeedMph), @@ -1031,10 +1124,12 @@ class RideStateMachine extends StateNotifier { newTopSpeed = sample.speed!; } } - + // Continuous sub-second time accumulation across 20Hz frames if (_lastSample != null && sample.speed != null) { - final dt = sample.timestamp.difference(_lastSample!.timestamp).inMilliseconds / 1000.0; + final dt = + sample.timestamp.difference(_lastSample!.timestamp).inMilliseconds / + 1000.0; if (dt > 0 && dt < 2.0) { if (sample.speed! > 1.8) { _movingTimeAccumulator += dt; @@ -1049,23 +1144,34 @@ class RideStateMachine extends StateNotifier { // Speed-gate dynamic motorcycle metrics (require speed >= 3.0 MPH to prevent stationary kickstand / dismount artifacts from registering as personal riding records) final isDynamicallyMoving = sample.speed != null && sample.speed! >= 3.0; - if (isDynamicallyMoving && sample.leanAngle != null && sample.leanAngle!.abs() <= 65.0) { - if (sample.leanAngle! < 0 && sample.leanAngle!.abs() > state.maxLeanLeft) { + if (isDynamicallyMoving && + sample.leanAngle != null && + sample.leanAngle!.abs() <= 65.0) { + if (sample.leanAngle! < 0 && + sample.leanAngle!.abs() > state.maxLeanLeft) { newMaxLeanLeft = sample.leanAngle!.abs(); - } else if (sample.leanAngle! > 0 && sample.leanAngle! > state.maxLeanRight) { + } else if (sample.leanAngle! > 0 && + sample.leanAngle! > state.maxLeanRight) { newMaxLeanRight = sample.leanAngle!; } } - if (isDynamicallyMoving && sample.longitudinalG != null && sample.longitudinalG!.abs() <= 2.5) { - if (sample.longitudinalG! > 0 && sample.longitudinalG! > state.maxAccelG) { + if (isDynamicallyMoving && + sample.longitudinalG != null && + sample.longitudinalG!.abs() <= 2.5) { + if (sample.longitudinalG! > 0 && + sample.longitudinalG! > state.maxAccelG) { newMaxAccelG = sample.longitudinalG!; - } else if (sample.longitudinalG! < 0 && sample.longitudinalG!.abs() > state.maxBrakingG) { + } else if (sample.longitudinalG! < 0 && + sample.longitudinalG!.abs() > state.maxBrakingG) { newMaxBrakingG = sample.longitudinalG!.abs(); } } - if (sample.latitude != null && sample.longitude != null && _lastSample?.latitude != null && _lastSample?.longitude != null) { + if (sample.latitude != null && + sample.longitude != null && + _lastSample?.latitude != null && + _lastSample?.longitude != null) { final distMeters = Geolocator.distanceBetween( _lastSample!.latitude!, _lastSample!.longitude!, @@ -1074,39 +1180,42 @@ class RideStateMachine extends StateNotifier { ); // Accumulate distance ONLY if moving and GPS accuracy is reasonable (suppresses stationary jitter at stop lights) - final isMoving = (sample.speed != null && sample.speed! >= 1.0) || (_lastSample!.speed != null && _lastSample!.speed! >= 1.0); - final hasGoodGps = sample.gpsAccuracy != null && sample.gpsAccuracy! <= 35.0; + final isMoving = + (sample.speed != null && sample.speed! >= 1.0) || + (_lastSample!.speed != null && _lastSample!.speed! >= 1.0); + final hasGoodGps = + sample.gpsAccuracy != null && sample.gpsAccuracy! <= 35.0; if (isMoving && hasGoodGps && distMeters > 0.3 && distMeters < 150.0) { _accumulatedDistance += distMeters; } - + // Elevation tracking if (sample.altitude != null && _lastSample!.altitude != null) { - final elevationDelta = sample.altitude! - _lastSample!.altitude!; - // Threshold filter to prevent phantom gain (e.g. > 1 meter change) - if (elevationDelta.abs() > 1.0 && elevationDelta.abs() < 50.0) { - if (elevationDelta > 0) { - newElevationGain += elevationDelta; - } else { - newElevationLoss += elevationDelta.abs(); - } - } + final elevationDelta = sample.altitude! - _lastSample!.altitude!; + // Threshold filter to prevent phantom gain (e.g. > 1 meter change) + if (elevationDelta.abs() > 1.0 && elevationDelta.abs() < 50.0) { + if (elevationDelta > 0) { + newElevationGain += elevationDelta; + } else { + newElevationLoss += elevationDelta.abs(); + } + } } } - + double tripDistanceMiles = _accumulatedDistance / 1609.34; - + // Calculate avg speed (only moving time) double avgSpeed = 0; if (newMovingTime > 0) { - // miles per hour - avgSpeed = (tripDistanceMiles / newMovingTime) * 3600; + // miles per hour + avgSpeed = (tripDistanceMiles / newMovingTime) * 3600; } - + _detectEvents(sample); _lastSample = sample; - + state = state.copyWith( topSpeedMph: newTopSpeed, maxLeanLeft: newMaxLeanLeft, @@ -1121,7 +1230,7 @@ class RideStateMachine extends StateNotifier { averageSpeedMph: avgSpeed, ); } - + void _detectEvents(TelemetrySample sample) { if (state.dbId == null) return; final now = DateTime.now(); @@ -1131,32 +1240,48 @@ class RideStateMachine extends StateNotifier { final lastTime = _lastEventTimes[type]; if (lastTime == null || now.difference(lastTime) > _eventCooldown) { _lastEventTimes[type] = now; - db.into(db.rideEvents).insert(RideEventsCompanion.insert( - rideId: state.dbId!, - timestamp: now, - eventType: type, - value: value, - latitude: drift.Value(sample.latitude), - longitude: drift.Value(sample.longitude), - )); + db + .into(db.rideEvents) + .insert( + RideEventsCompanion.insert( + rideId: state.dbId!, + timestamp: now, + eventType: type, + value: value, + latitude: drift.Value(sample.latitude), + longitude: drift.Value(sample.longitude), + ), + ); } } } if (sample.longitudinalG != null) { - checkAndLogEvent('Hard Braking', sample.longitudinalG!, sample.longitudinalG! < -0.8); - checkAndLogEvent('Hard Acceleration', sample.longitudinalG!, sample.longitudinalG! > 0.6); + checkAndLogEvent( + 'Hard Braking', + sample.longitudinalG!, + sample.longitudinalG! < -0.8, + ); + checkAndLogEvent( + 'Hard Acceleration', + sample.longitudinalG!, + sample.longitudinalG! > 0.6, + ); } - + if (sample.leanAngle != null) { - checkAndLogEvent('Extreme Lean', sample.leanAngle!, sample.leanAngle!.abs() > 45.0); + checkAndLogEvent( + 'Extreme Lean', + sample.leanAngle!, + sample.leanAngle!.abs() > 45.0, + ); } - + if (sample.speed != null) { checkAndLogEvent('High Speed', sample.speed!, sample.speed! > 100.0); } } - + @override void dispose() { _telemetrySub?.cancel(); @@ -1166,10 +1291,17 @@ class RideStateMachine extends StateNotifier { } } -final rideStateMachineProvider = StateNotifierProvider((ref) { - final db = ref.watch(databaseProvider); - final telemetryService = ref.watch(telemetryServiceProvider); - final recordsService = ref.watch(recordsServiceProvider.notifier); - final audioAlertService = ref.watch(audioAlertServiceProvider); - return RideStateMachine(db, telemetryService, recordsService, audioAlertService, ref); -}); +final rideStateMachineProvider = + StateNotifierProvider((ref) { + final db = ref.watch(databaseProvider); + final telemetryService = ref.watch(telemetryServiceProvider); + final recordsService = ref.watch(recordsServiceProvider.notifier); + final audioAlertService = ref.watch(audioAlertServiceProvider); + return RideStateMachine( + db, + telemetryService, + recordsService, + audioAlertService, + ref, + ); + }); diff --git a/lib/services/field_test/field_test_logger.dart b/lib/services/field_test/field_test_logger.dart index 25b9ab7..354c96c 100644 --- a/lib/services/field_test/field_test_logger.dart +++ b/lib/services/field_test/field_test_logger.dart @@ -93,6 +93,20 @@ class FieldTestLogger { int get totalEnqueued => _totalEnqueued; int get totalFlushed => _totalFlushed; int get droppedEventsCount => _droppedEventsCount; + int _telemetrySamplesLost = 0; + int get telemetrySamplesLost => _telemetrySamplesLost; + + /// Records the loss of telemetry samples due to buffer overflow or queue bounds. + void recordTelemetryLoss(int count) { + if (count > 0) { + _telemetrySamplesLost += count; + } + } + + /// Sets or replaces the explicit telemetry loss counter for the active session. + set telemetrySamplesLost(int value) { + _telemetrySamplesLost = value < 0 ? 0 : value; + } String? get oldestDroppedItem => _oldestDroppedItem; String? get newestDroppedItem => _newestDroppedItem; int get lastFlushDurationMs => _lastFlushDurationMs; @@ -267,6 +281,7 @@ class FieldTestLogger { _oldestDroppedItem = null; _newestDroppedItem = null; _failedFlushCount = 0; + _telemetrySamplesLost = 0; _recentSampleRateHz = 0.0; _recentSampleCount = 0; @@ -700,7 +715,12 @@ class FieldTestLogger { } /// Generates forensic summary from recorded ride telemetry and events - Future generateAndSaveSummary(int rideId, {bool wasRecovered = false}) async { + Future generateAndSaveSummary( + int rideId, { + int? telemetrySamplesLost, + bool wasRecovered = false, + }) async { + final effectiveTelemetryLost = telemetrySamplesLost ?? _telemetrySamplesLost; final ride = await (_db.select(_db.rides)..where((r) => r.id.equals(rideId))).getSingleOrNull(); final startTime = ride?.startTime ?? DateTime.now(); final endTime = ride?.endTime ?? DateTime.now(); @@ -815,7 +835,7 @@ class FieldTestLogger { timestampGaps: _telemetryGapCount, gpsFixCount: gpsFixCount, totalGpsAccuracy: accuracySum, - databaseSamplesLost: _droppedEventsCount, + databaseSamplesLost: effectiveTelemetryLost, imuSaturationCount: _imuSaturationCount, isMountCalibrated: true, ); @@ -852,7 +872,7 @@ class FieldTestLogger { bestZeroTo60: best0to60, bestZeroTo100: best0to100, databaseSamplesWritten: totalSamples, - databaseSamplesLost: _droppedEventsCount, + databaseSamplesLost: effectiveTelemetryLost, eventCount: events.length, overallConfidence: qualityReport.overallScore / 100.0, rideQualityScore: qualityReport.overallScore, @@ -896,7 +916,7 @@ class FieldTestLogger { bestZeroTo60: drift.Value(best0to60), bestZeroTo100: drift.Value(best0to100), databaseSamplesWritten: drift.Value(totalSamples), - databaseSamplesLost: drift.Value(_droppedEventsCount), + databaseSamplesLost: drift.Value(effectiveTelemetryLost), eventCount: drift.Value(events.length), overallConfidence: drift.Value(qualityReport.overallScore / 100.0), rideQualityScore: drift.Value(qualityReport.overallScore), diff --git a/test/field_test_logger_test.dart b/test/field_test_logger_test.dart index 18fc532..3a071e6 100644 --- a/test/field_test_logger_test.dart +++ b/test/field_test_logger_test.dart @@ -1,4 +1,4 @@ -import 'package:flutter_test/flutter_test.dart'; +import 'package:flutter_test/flutter_test.dart'; import 'package:drift/native.dart'; import 'package:shared_preferences/shared_preferences.dart'; @@ -29,36 +29,118 @@ void main() { expect(logger.isSessionActive, isTrue); expect(logger.activeRideId, 10); - expect(logger.timelineEvents.any((e) => e.eventType == FieldTestEventType.rideStarted), isTrue); + expect( + logger.timelineEvents.any( + (e) => e.eventType == FieldTestEventType.rideStarted, + ), + isTrue, + ); final summary = await logger.stopSession(); expect(logger.isSessionActive, isFalse); expect(summary.rideId, 10); }); - test('Queue drops lowest-priority events when buffer capacity is reached', () { - logger.startSession(11); + test( + 'Queue drops lowest-priority events when buffer capacity is reached', + () { + logger.startSession(11); - // Fill queue with 1,000 info events - for (int i = 0; i < 1005; i++) { + // Fill queue with 1,000 info events + for (int i = 0; i < 1005; i++) { + logger.log( + eventType: FieldTestEventType.databaseFlush, + severity: EventSeverity.info, + message: 'Flush event $i', + rideId: 11, + ); + } + + // Add a critical event logger.log( - eventType: FieldTestEventType.databaseFlush, - severity: EventSeverity.info, - message: 'Flush event $i', + eventType: FieldTestEventType.databaseRecovery, + severity: EventSeverity.critical, + message: 'Critical DB Recovery', rideId: 11, ); - } - // Add a critical event - logger.log( - eventType: FieldTestEventType.databaseRecovery, - severity: EventSeverity.critical, - message: 'Critical DB Recovery', - rideId: 11, - ); + expect(logger.droppedEventsCount, greaterThanOrEqualTo(5)); + expect( + logger.timelineEvents.any( + (e) => e.severity == EventSeverity.critical, + ), + isTrue, + ); + }, + ); - expect(logger.droppedEventsCount, greaterThanOrEqualTo(5)); - expect(logger.timelineEvents.any((e) => e.severity == EventSeverity.critical), isTrue); + test( + 'Dropped diagnostic events alone do not inflate databaseSamplesLost', + () async { + logger.startSession(12); + + // Saturate diagnostic event queue so diagnostic events are dropped + for (int i = 0; i < 1010; i++) { + logger.log( + eventType: FieldTestEventType.databaseFlush, + severity: EventSeverity.info, + message: 'Flush info event $i', + rideId: 12, + ); + } + + expect(logger.droppedEventsCount, greaterThanOrEqualTo(10)); + // Zero telemetry lost recorded + expect(logger.telemetrySamplesLost, 0); + + final summary = await logger.stopSession(); + // databaseSamplesLost MUST be 0 despite droppedEventsCount > 0 + expect(summary.databaseSamplesLost, 0); + }, + ); + + test( + 'Recorded telemetry loss is preserved across diagnostic event drops and reflected in summary', + () async { + logger.startSession(13); + + // Record 101 telemetry drops + logger.recordTelemetryLoss(101); + expect(logger.telemetrySamplesLost, 101); + + // Saturate diagnostic event queue with warning burst + for (int i = 0; i < 1020; i++) { + logger.log( + eventType: FieldTestEventType.databaseQueueOverflow, + severity: EventSeverity.warning, + message: 'Overflow warning $i', + rideId: 13, + ); + } + + expect(logger.droppedEventsCount, greaterThanOrEqualTo(20)); + // Telemetry loss tally must NOT be corrupted by diagnostic event drops + expect(logger.telemetrySamplesLost, 101); + + final summary = await logger.stopSession(); + expect(summary.databaseSamplesLost, 101); + }, + ); + + test('Starting a new session resets telemetry loss tally', () async { + logger.startSession(14); + logger.recordTelemetryLoss(55); + expect(logger.telemetrySamplesLost, 55); + + final summary1 = await logger.stopSession(); + expect(summary1.databaseSamplesLost, 55); + + // Start new session #15 + logger.startSession(15); + expect(logger.telemetrySamplesLost, 0); + + final summary2 = await logger.stopSession(); + expect(summary2.databaseSamplesLost, 0); }); }); } diff --git a/test/rid3a_telemetry_batch_ownership_test.dart b/test/rid3a_telemetry_batch_ownership_test.dart index 3ce5161..2bc617f 100644 --- a/test/rid3a_telemetry_batch_ownership_test.dart +++ b/test/rid3a_telemetry_batch_ownership_test.dart @@ -23,10 +23,12 @@ 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 +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/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'; @@ -35,6 +37,7 @@ 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/models/recording_health_status.dart'; import 'package:motorcycle_dash/features/records/records_service.dart'; /// Drift's generated row class for telemetry_samples is also named @@ -53,7 +56,8 @@ class _Rid3aAppDatabase extends AppDatabase { @override Future batch( - FutureOr Function(drift.Batch batch) runInBatch) async { + 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. @@ -79,6 +83,51 @@ class _Rid3aFieldTestLogger extends FieldTestLogger { _Rid3aFieldTestLogger(super.db); int stopSessionInvocations = 0; + int overflowEventCount = 0; + final List loggedEventTypes = []; + + @override + void log({ + required FieldTestEventType eventType, + required EventSeverity severity, + required String message, + DateTime? timestamp, + int? rideId, + DateTime? telemetryTimestamp, + int? gpsAgeMs, + double? speed, + double? latitude, + double? longitude, + double? gpsAccuracy, + double? leanAngle, + double? longitudinalG, + double? lateralG, + double? overallConfidence, + Map? metadata, + }) { + loggedEventTypes.add(eventType); + if (eventType == FieldTestEventType.databaseQueueOverflow) { + overflowEventCount++; + } + super.log( + eventType: eventType, + severity: severity, + message: message, + timestamp: timestamp, + rideId: rideId, + telemetryTimestamp: telemetryTimestamp, + gpsAgeMs: gpsAgeMs, + speed: speed, + latitude: latitude, + longitude: longitude, + gpsAccuracy: gpsAccuracy, + leanAngle: leanAngle, + longitudinalG: longitudinalG, + lateralG: lateralG, + overallConfidence: overallConfidence, + metadata: metadata, + ); + } @override Future flush() async {} @@ -122,7 +171,7 @@ class FakeTelemetryService extends TelemetryService { class _NullLocationService extends Fake implements LocationService { @override void Function(Position position, bool accepted, String? rejectionReason)? - onGpsDiagnostic; + onGpsDiagnostic; @override Stream getFilteredPositionStream() => const Stream.empty(); } @@ -238,9 +287,18 @@ void main() { } } - Future> persistedRows() => (db.select(db.telemetrySamples) - ..orderBy([(t) => drift.OrderingTerm(expression: t.id)])) - .get(); + Future> persistedRows() => + (db.select(db.telemetrySamples)..orderBy([ + (t) => drift.OrderingTerm( + expression: t.timestamp, + mode: drift.OrderingMode.asc, + ), + (t) => drift.OrderingTerm( + expression: t.id, + mode: drift.OrderingMode.asc, + ), + ])) + .get(); Map countsByIndex(List rows) { final counts = {}; @@ -258,13 +316,12 @@ void main() { ..sort(); List missingOf(int total, Map counts) => [ - for (var i = 0; i < total; i++) - if ((counts[i] ?? 0) == 0) i, - ]; + 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 { + 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); @@ -286,14 +343,18 @@ void main() { 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(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'); + expect( + session.totalSamplesPersisted, + 10, + reason: 'buffer empty after a successful final flush', + ); }); test('overlapping periodic and finishRide flushes persist all 30 accepted ' @@ -310,7 +371,11 @@ void main() { // 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'); + 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. @@ -327,8 +392,11 @@ void main() { // drains the newer samples only after A has committed. final finishing = notifier.finishRide(); await pumpEventQueue(times: 30); - expect(db.batchInvocations, 1, - reason: 'while the owner is parked, no second flush may take ownership'); + expect( + db.batchInvocations, + 1, + reason: 'while the owner is parked, no second flush may take ownership', + ); // 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 @@ -339,10 +407,16 @@ void main() { 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'); + 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 @@ -350,13 +424,21 @@ void main() { // 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'); + 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', + ); // Release the barrier's drain and let finishRide complete. It persists // 10..29 itself, so no later periodic flush is required. @@ -376,23 +458,36 @@ void main() { 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()}'); + 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'); + 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 ' @@ -434,11 +529,15 @@ void main() { debugPrint('order : ${rows.map((r) => r.speed).toList()}'); expect(rows, hasLength(10)); - expect(duplicatedIn(counts), isEmpty, reason: 'A and B each exactly once'); + 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.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); }); @@ -468,11 +567,17 @@ void main() { await pumpEventQueue(times: 40); var session = container.read(rideStateMachineProvider); - expect(session.totalSamplesPersisted, 0, - reason: 'a failed persistence must not increment the success counter'); + 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'); + 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. @@ -490,19 +595,332 @@ void main() { debugPrint('duplicated : ${duplicatedIn(counts)}'); debugPrint('missing : ${missingOf(6, counts)}'); debugPrint('order : ${rows.map((r) => r.speed).toList()}'); - debugPrint('received / persisted : ' - '${session.totalSamplesReceived} / ${session.totalSamplesPersisted}'); + debugPrint( + 'received / persisted : ' + '${session.totalSamplesReceived} / ${session.totalSamplesPersisted}', + ); expect(rows, hasLength(6)); - expect(duplicatedIn(counts), isEmpty, reason: 'retry causes no duplicates'); + 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.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'); + expect( + session.totalSamplesPersisted, + 6, + reason: 'counter reflects only the successfully persisted snapshot', + ); }); + + test('failed flush restoration above 1000 capacity bounds queue to 1000 ' + 'by dropping oldest samples', () async { + final container = buildContainer(); + final notifier = container.read(rideStateMachineProvider.notifier); + final rideId = await startRide(container); + + // 1. Emit 600 samples (indices 0..599). + emitSamples(0, 600); + await pumpEventQueue(times: 30); + expect( + container.read(rideStateMachineProvider).totalSamplesReceived, + 600, + ); + + // 2. Start flush of those 600 samples, but gate db.batch. + db.gateBatch = true; + db.batchFailuresRemaining = 1; + final finishAttempt = notifier.finishRide(); + await pumpEventQueue(times: 10); + expect(db.batchInvocations, 1); + final failingFlush = db.batchGates.single; + + // 3. While write is pending, emit 500 new samples (indices 600..1099). + // Total received = 1100. + emitSamples(600, 500); + await pumpEventQueue(times: 30); + expect( + container.read(rideStateMachineProvider).totalSamplesReceived, + 1100, + ); + + // 4. Fail the write. Snapshot of 600 is restored to front of the 500 new samples. + // Combined = 1100 > 1000. + db.gateBatch = false; + failingFlush.complete(); + await expectLater(finishAttempt, throwsStateError); + await pumpEventQueue(times: 30); + + // Assert ride remains open and no final summary produced after failed Stop + final intermediateSession = container.read(rideStateMachineProvider); + expect( + intermediateSession.state, + RideState.riding, + reason: 'ride remains riding on stop failure', + ); + expect( + intermediateSession.healthStatus, + RecordingHealthStatus.recordingActive, + reason: 'recordingActive restored on stop failure', + ); + final intermediateRides = await db.select(db.rides).get(); + expect( + intermediateRides.singleWhere((r) => r.id == rideId).endTime, + isNull, + reason: 'no endTime written to DB on stop failure', + ); + expect( + logger.stopSessionInvocations, + 0, + reason: 'no logger stopSession called on stop failure', + ); + expect( + await db.select(db.fieldTestSummaries).get(), + isEmpty, + reason: 'no final summary produced on stop failure', + ); + expect( + notifier.telemetrySamplesLost, + 100, + reason: '100 samples lost on restoration excess', + ); + expect(logger.telemetrySamplesLost, 100); + + // 5. Emit 1 more sample (index 1100). + emitSamples(1100, 1); + await pumpEventQueue(times: 30); + expect( + container.read(rideStateMachineProvider).totalSamplesReceived, + 1101, + ); + expect( + notifier.telemetrySamplesLost, + 101, + reason: '101 samples lost after 1 drop on arrival', + ); + expect(logger.telemetrySamplesLost, 101); + + // 6. Now do a successful finish flush to persist whatever remains in queue. + await notifier.finishRide(); + await pumpEventQueue(times: 30); + + final rows = await persistedRows(); + final counts = countsByIndex(rows); + final session = container.read(rideStateMachineProvider); + + debugPrint('==== RID-3A OVERFLOW REGRESSION ===='); + debugPrint('durable rows : ${rows.length}'); + debugPrint('distinct identities : ${counts.length}'); + debugPrint('overflow events : ${logger.overflowEventCount}'); + debugPrint( + 'first row speed : ${rows.isNotEmpty ? rows.first.speed : "empty"}', + ); + debugPrint( + 'last row speed : ${rows.isNotEmpty ? rows.last.speed : "empty"}', + ); + + // Must be capped at 1000. + expect(rows, hasLength(1000)); + // First 101 oldest samples (indices 0..100) dropped (100 on restoration + 1 on sample 1100). + // Verify complete retained sample identity sequence in exact order. + expect( + rows.map((r) => r.speed).toList(), + [for (var i = 101; i <= 1100; i++) 1000.0 + i], + reason: 'complete sequence of 1000 retained samples in exact order', + ); + expect(duplicatedIn(counts), isEmpty); + expect(logger.overflowEventCount, 101); + expect(session.totalSamplesPersisted, 1000); + expect(rows.map((r) => r.rideId).toSet(), {rideId}); + + final summaries = await db.select(db.fieldTestSummaries).get(); + expect(summaries, hasLength(1)); + final summary = summaries.single; + expect( + summary.databaseSamplesLost, + 101, + reason: 'databaseSamplesLost reports exact dropped telemetry', + ); + expect(summary.databaseSamplesWritten, 1000); + expect(notifier.telemetrySamplesLost, 101); + expect(logger.telemetrySamplesLost, 101); + }); + + test('repeated flush failures keep queue bounded to exactly 1000 capacity', () async { + final container = buildContainer(); + final notifier = container.read(rideStateMachineProvider.notifier); + final rideId = await startRide(container); + + // Cycle 1: Emit 800 samples (0..799). + emitSamples(0, 800); + await pumpEventQueue(times: 30); + + db.gateBatch = true; + db.batchFailuresRemaining = 1; + final finish1 = notifier.finishRide(); + await pumpEventQueue(times: 10); + expect(db.batchInvocations, 1); + final flush1 = db.batchGates.single; + + // 300 arrive during await (800..1099). Total received: 1100. + emitSamples(800, 300); + await pumpEventQueue(times: 30); + + db.gateBatch = false; + flush1.complete(); + await expectLater(finish1, throwsStateError); + await pumpEventQueue(times: 30); + + // Assert ride remains open and no final summary produced after fail 1 + final sessionAfterFail1 = container.read(rideStateMachineProvider); + expect(sessionAfterFail1.state, RideState.riding); + expect( + sessionAfterFail1.healthStatus, + RecordingHealthStatus.recordingActive, + ); + final ridesAfterFail1 = await db.select(db.rides).get(); + expect( + ridesAfterFail1.singleWhere((r) => r.id == rideId).endTime, + isNull, + ); + expect(logger.stopSessionInvocations, 0); + expect(await db.select(db.fieldTestSummaries).get(), isEmpty); + + // First failure: 800 + 300 = 1100 -> 100 oldest dropped (0..99). Retained: 100..1099. + expect(logger.overflowEventCount, 100); + expect(notifier.telemetrySamplesLost, 100); + expect(logger.telemetrySamplesLost, 100); + + // Cycle 2: Flush 2 starts, owns all 1000 currently in queue. + db.batchGates.clear(); + db.gateBatch = true; + db.batchFailuresRemaining = 1; + final finish2 = notifier.finishRide(); + await pumpEventQueue(times: 10); + expect(db.batchInvocations, 2); + final flush2 = db.batchGates.single; + + // 200 arrive during await (1100..1299). Total received: 1300. + emitSamples(1100, 200); + await pumpEventQueue(times: 30); + + db.gateBatch = false; + flush2.complete(); + await expectLater(finish2, throwsStateError); + await pumpEventQueue(times: 30); + + // Assert ride remains open and no final summary produced after fail 2 + final sessionAfterFail2 = container.read(rideStateMachineProvider); + expect(sessionAfterFail2.state, RideState.riding); + expect( + sessionAfterFail2.healthStatus, + RecordingHealthStatus.recordingActive, + ); + final ridesAfterFail2 = await db.select(db.rides).get(); + expect( + ridesAfterFail2.singleWhere((r) => r.id == rideId).endTime, + isNull, + ); + expect(logger.stopSessionInvocations, 0); + expect(await db.select(db.fieldTestSummaries).get(), isEmpty); + + // Second failure: 1000 + 200 = 1200 -> 200 oldest dropped (100..299). Retained: 300..1299. + expect(logger.overflowEventCount, 300); + expect(notifier.telemetrySamplesLost, 300); + expect(logger.telemetrySamplesLost, 300); + + // Subsequent 50 arrivals (1300..1349). + emitSamples(1300, 50); + await pumpEventQueue(times: 30); + // Each drops 1 oldest (300..349). Retained: 350..1349 (1000 samples). + expect(logger.overflowEventCount, 350); + expect(notifier.telemetrySamplesLost, 350); + expect(logger.telemetrySamplesLost, 350); + + // Successful finalization persists retained 1000. + await notifier.finishRide(); + await pumpEventQueue(times: 30); + + final rows = await persistedRows(); + final counts = countsByIndex(rows); + final session = container.read(rideStateMachineProvider); + + debugPrint('==== RID-3A REPEATED FAILURES ===='); + debugPrint('durable rows : ${rows.length}'); + debugPrint('distinct identities : ${counts.length}'); + debugPrint('overflow events : ${logger.overflowEventCount}'); + debugPrint( + 'first row speed : ${rows.isNotEmpty ? rows.first.speed : "empty"}', + ); + debugPrint( + 'last row speed : ${rows.isNotEmpty ? rows.last.speed : "empty"}', + ); + + expect(rows, hasLength(1000)); + // Verify complete retained sample identity sequence in exact order. + expect( + rows.map((r) => r.speed).toList(), + [for (var i = 350; i <= 1349; i++) 1000.0 + i], + reason: + 'complete sequence of 1000 retained samples in exact order after repeated drops', + ); + expect(duplicatedIn(counts), isEmpty); + expect(logger.overflowEventCount, 350); + expect(session.totalSamplesPersisted, 1000); + expect(rows.map((r) => r.rideId).toSet(), {rideId}); + + final summaries = await db.select(db.fieldTestSummaries).get(); + expect(summaries, hasLength(1)); + final summary = summaries.single; + expect( + summary.databaseSamplesLost, + 350, + reason: 'databaseSamplesLost reflects 350 total dropped telemetry', + ); + expect(summary.databaseSamplesWritten, 1000); + expect(notifier.telemetrySamplesLost, 350); + expect(logger.telemetrySamplesLost, 350); + }); + + test( + 'clean ride with zero telemetry loss produces databaseSamplesLost == 0 and new ride resets counter', + () async { + final container = buildContainer(); + final notifier = container.read(rideStateMachineProvider.notifier); + final logger = container.read(fieldTestLoggerProvider); + final rideId = await startRide(container); + + // Emit 50 samples (< 1000 buffer capacity) + emitSamples(0, 50); + await pumpEventQueue(times: 30); + + expect(notifier.telemetrySamplesLost, 0); + expect(logger.telemetrySamplesLost, 0); + + // Finish cleanly + await notifier.finishRide(); + await pumpEventQueue(times: 30); + + final summaries = await db.select(db.fieldTestSummaries).get(); + expect(summaries, hasLength(1)); + expect(summaries.single.databaseSamplesLost, 0); + expect(summaries.single.databaseSamplesWritten, 50); + + // Start new ride: counters must reset to 0 + final rideId2 = await startRide(container); + expect(rideId2, isNot(rideId)); + expect(notifier.telemetrySamplesLost, 0); + expect(logger.telemetrySamplesLost, 0); + }, + ); }); }