fix(telemetry): accurately track dropped telemetry samples under queue bounds in ride state and field test summary

This commit is contained in:
Developer 2026-09-18 18:34:58 -04:00
parent bb509a4bac
commit 0282a2972f
4 changed files with 871 additions and 219 deletions

View File

@ -36,7 +36,7 @@ class RideSession {
final double maxAccelG; final double maxAccelG;
final double maxBrakingG; final double maxBrakingG;
final DateTime? startTime; final DateTime? startTime;
// Phase 4 Stats // Phase 4 Stats
final int movingTimeSeconds; final int movingTimeSeconds;
final int stoppedTimeSeconds; final int stoppedTimeSeconds;
@ -105,7 +105,8 @@ class RideSession {
elevationLoss: elevationLoss ?? this.elevationLoss, elevationLoss: elevationLoss ?? this.elevationLoss,
averageSpeedMph: averageSpeedMph ?? this.averageSpeedMph, averageSpeedMph: averageSpeedMph ?? this.averageSpeedMph,
totalSamplesReceived: totalSamplesReceived ?? this.totalSamplesReceived, totalSamplesReceived: totalSamplesReceived ?? this.totalSamplesReceived,
totalSamplesPersisted: totalSamplesPersisted ?? this.totalSamplesPersisted, totalSamplesPersisted:
totalSamplesPersisted ?? this.totalSamplesPersisted,
); );
} }
} }
@ -116,20 +117,20 @@ class RideStateMachine extends StateNotifier<RideSession> {
final RecordsService recordsService; final RecordsService recordsService;
final AudioAlertService audioAlertService; final AudioAlertService audioAlertService;
final Ref ref; final Ref ref;
StreamSubscription? _telemetrySub; StreamSubscription? _telemetrySub;
TelemetrySample? _lastSample; TelemetrySample? _lastSample;
double _accumulatedDistance = 0; // meters internally double _accumulatedDistance = 0; // meters internally
double _movingTimeAccumulator = 0.0; // seconds internally double _movingTimeAccumulator = 0.0; // seconds internally
double _stoppedTimeAccumulator = 0.0; // seconds internally double _stoppedTimeAccumulator = 0.0; // seconds internally
// Batching // Batching
final List<TelemetrySamplesCompanion> _telemetryBatch = []; final List<TelemetrySamplesCompanion> _telemetryBatch = [];
Timer? _batchWriteTimer; Timer? _batchWriteTimer;
Timer? _watchdogTimer; Timer? _watchdogTimer;
DateTime? _lastTelemetryReceivedTime; DateTime? _lastTelemetryReceivedTime;
// Event Debouncing (Store last trigger time per event type) // Event Debouncing (Store last trigger time per event type)
final Map<String, DateTime> _lastEventTimes = {}; final Map<String, DateTime> _lastEventTimes = {};
static const Duration _eventCooldown = Duration(seconds: 5); static const Duration _eventCooldown = Duration(seconds: 5);
@ -181,14 +182,30 @@ class RideStateMachine extends StateNotifier<RideSession> {
// retryable rather than transitioning to finished. // retryable rather than transitioning to finished.
static const int _maxFinalDrainPasses = 4; 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); _telemetrySub = telemetryService.telemetryStream.listen(_onTelemetryUpdate);
// Write to DB every 3 seconds for tight crash-durability // 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 // 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 // Recover any unfinished rides from previous sessions
_recoverInterruptedRides(); _recoverInterruptedRides();
@ -278,25 +295,34 @@ class RideStateMachine extends StateNotifier<RideSession> {
// is also what prevents monotonic freeze-blindness: when samples stop, the // is also what prevents monotonic freeze-blindness: when samples stop, the
// monotonic reference stops advancing, but this term still goes stale. // monotonic reference stops advancing, but this term still goes stale.
final accepted = _lastAcceptedTelemetryTime; final accepted = _lastAcceptedTelemetryTime;
final telemetryAgeMs = final telemetryAgeMs = accepted == null
accepted == null ? null : now.difference(accepted).inMilliseconds; ? null
: now.difference(accepted).inMilliseconds;
// Fail closed on a backwards clock jump: negative age is not freshness. // Fail closed on a backwards clock jump: negative age is not freshness.
final telemetryActive = telemetryAgeMs != null && final telemetryActive =
telemetryAgeMs != null &&
telemetryAgeMs >= 0 && telemetryAgeMs >= 0 &&
telemetryAgeMs <= sensorBound; telemetryAgeMs <= sensorBound;
// B. GPS and D. IMU evidence: monotonic, presence-proven. // B. GPS and D. IMU evidence: monotonic, presence-proven.
final gpsOperational = final gpsOperational = _isFresh(
_isFresh(_monotonicAgeMs(_lastGpsEvidenceMonotonicMs), sensorBound); _monotonicAgeMs(_lastGpsEvidenceMonotonicMs),
final sensorActive = sensorBound,
_isFresh(_monotonicAgeMs(_lastImuEvidenceMonotonicMs), sensorBound); );
final sensorActive = _isFresh(
_monotonicAgeMs(_lastImuEvidenceMonotonicMs),
sensorBound,
);
// E. Database evidence: advances ONLY after a real successful telemetry // E. Database evidence: advances ONLY after a real successful telemetry
// commit (the RID-3B-2 commit point). A scheduled flush is not evidence, and // commit (the RID-3B-2 commit point). A scheduled flush is not evidence, and
// a failed pre-commit batch does not advance it. // a failed pre-commit batch does not advance it.
final dbOperational = !_dbPersistenceFailed && final dbOperational =
_isFresh(_monotonicAgeMs(_lastDbCommitMonotonicMs), !_dbPersistenceFailed &&
RecordingHealthContract.databaseWriteStaleAfterMs); _isFresh(
_monotonicAgeMs(_lastDbCommitMonotonicMs),
RecordingHealthContract.databaseWriteStaleAfterMs,
);
final previous = state.health; final previous = state.health;
final health = previous.copyWith( final health = previous.copyWith(
@ -335,11 +361,14 @@ class RideStateMachine extends StateNotifier<RideSession> {
if (stalled && previousMisses == 0) { if (stalled && previousMisses == 0) {
// First detection of this stall episode: logged once, as before. // First detection of this stall episode: logged once, as before.
try { try {
ref.read(fieldTestLoggerProvider).log( ref
eventType: FieldTestEventType.telemetryGap, .read(fieldTestLoggerProvider)
severity: EventSeverity.warning, .log(
message: 'Watchdog: Telemetry pipeline stalled ($elapsedSec seconds without samples)', eventType: FieldTestEventType.telemetryGap,
); severity: EventSeverity.warning,
message:
'Watchdog: Telemetry pipeline stalled ($elapsedSec seconds without samples)',
);
} catch (_) {} } catch (_) {}
} }
@ -357,24 +386,30 @@ class RideStateMachine extends StateNotifier<RideSession> {
Future<void> _recoverInterruptedRides() async { Future<void> _recoverInterruptedRides() async {
try { try {
final unclosedRides = await (db.select(db.rides) final unclosedRides = await (db.select(
..where((r) => r.endTime.isNull()) db.rides,
).get(); )..where((r) => r.endTime.isNull())).get();
for (final r in unclosedRides) { for (final r in unclosedRides) {
// Find last recorded telemetry sample for this ride to determine actual end time // Find last recorded telemetry sample for this ride to determine actual end time
final lastSample = await (db.select(db.telemetrySamples) final lastSample =
..where((t) => t.rideId.equals(r.id)) await (db.select(db.telemetrySamples)
..orderBy([(t) => drift.OrderingTerm(expression: t.timestamp, mode: drift.OrderingMode.desc)]) ..where((t) => t.rideId.equals(r.id))
..limit(1) ..orderBy([
).getSingleOrNull(); (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( await (db.update(db.rides)..where((t) => t.id.equals(r.id))).write(
RidesCompanion( RidesCompanion(endTime: drift.Value(calculatedEndTime)),
endTime: drift.Value(calculatedEndTime),
),
); );
try { try {
@ -382,7 +417,8 @@ class RideStateMachine extends StateNotifier<RideSession> {
logger.log( logger.log(
eventType: FieldTestEventType.databaseRecovery, eventType: FieldTestEventType.databaseRecovery,
severity: EventSeverity.critical, severity: EventSeverity.critical,
message: 'Interrupted ride #${r.id} recovered after unexpected exit', message:
'Interrupted ride #${r.id} recovered after unexpected exit',
rideId: r.id, rideId: r.id,
); );
await logger.generateAndSaveSummary(r.id, wasRecovered: true); await logger.generateAndSaveSummary(r.id, wasRecovered: true);
@ -425,11 +461,13 @@ class RideStateMachine extends StateNotifier<RideSession> {
/// must never be invoked directly. /// must never be invoked directly.
Future<void> _performStartRide() async { Future<void> _performStartRide() async {
try { try {
ref.read(fieldTestLoggerProvider).log( ref
eventType: FieldTestEventType.rideStartRequested, .read(fieldTestLoggerProvider)
severity: EventSeverity.info, .log(
message: 'Ride start requested by user or motion trigger', eventType: FieldTestEventType.rideStartRequested,
); severity: EventSeverity.info,
message: 'Ride start requested by user or motion trigger',
);
} catch (_) {} } catch (_) {}
final defaultBike = ref.read(defaultMotorcycleProvider); final defaultBike = ref.read(defaultMotorcycleProvider);
@ -448,6 +486,7 @@ class RideStateMachine extends StateNotifier<RideSession> {
// IMU or the 3-second batch; health warms up automatically as real evidence // IMU or the 3-second batch; health warms up automatically as real evidence
// arrives and _evaluateRecordingHealth promotes the status. // arrives and _evaluateRecordingHealth promotes the status.
_resetHealthEvidence(); _resetHealthEvidence();
_telemetrySamplesLost = 0;
state = state.copyWith( state = state.copyWith(
dbId: id, dbId: id,
state: RideState.riding, state: RideState.riding,
@ -516,11 +555,13 @@ class RideStateMachine extends StateNotifier<RideSession> {
WakelockPlus.disable().catchError((_) {}); WakelockPlus.disable().catchError((_) {});
} catch (_) {} } catch (_) {}
try { try {
ref.read(fieldTestLoggerProvider).log( ref
eventType: FieldTestEventType.appPaused, .read(fieldTestLoggerProvider)
severity: EventSeverity.info, .log(
message: 'Ride session paused', eventType: FieldTestEventType.appPaused,
); severity: EventSeverity.info,
message: 'Ride session paused',
);
} catch (_) {} } catch (_) {}
} }
@ -528,7 +569,10 @@ class RideStateMachine extends StateNotifier<RideSession> {
// RID-3C-1: resuming does not by itself prove recording health. Evidence ages // RID-3C-1: resuming does not by itself prove recording health. Evidence ages
// out monotonically across the pause, so the next accepted sample // out monotonically across the pause, so the next accepted sample
// re-establishes truthfully whether the ride is active or degraded. // 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 _lastSample = null; // reset to avoid jumping
_lastTelemetryReceivedTime = DateTime.now(); _lastTelemetryReceivedTime = DateTime.now();
// RID-3C-1: re-publish truthful health for the resumed session. // RID-3C-1: re-publish truthful health for the resumed session.
@ -537,11 +581,13 @@ class RideStateMachine extends StateNotifier<RideSession> {
WakelockPlus.enable().catchError((_) {}); WakelockPlus.enable().catchError((_) {});
} catch (_) {} } catch (_) {}
try { try {
ref.read(fieldTestLoggerProvider).log( ref
eventType: FieldTestEventType.appResumed, .read(fieldTestLoggerProvider)
severity: EventSeverity.info, .log(
message: 'Ride session resumed', eventType: FieldTestEventType.appResumed,
); severity: EventSeverity.info,
message: 'Ride session resumed',
);
} catch (_) {} } catch (_) {}
} }
@ -581,15 +627,19 @@ class RideStateMachine extends StateNotifier<RideSession> {
/// [finishRide] and must never be invoked directly. /// [finishRide] and must never be invoked directly.
Future<void> _performFinishRide() async { Future<void> _performFinishRide() async {
if (state.dbId != null) { if (state.dbId != null) {
state = state.copyWith(healthStatus: RecordingHealthStatus.recordingFinalizing); state = state.copyWith(
healthStatus: RecordingHealthStatus.recordingFinalizing,
);
try { try {
ref.read(fieldTestLoggerProvider).log( ref
eventType: FieldTestEventType.rideFinishRequested, .read(fieldTestLoggerProvider)
severity: EventSeverity.info, .log(
message: 'Ride finish requested for ride #${state.dbId}', eventType: FieldTestEventType.rideFinishRequested,
rideId: state.dbId, severity: EventSeverity.info,
); message: 'Ride finish requested for ride #${state.dbId}',
rideId: state.dbId,
);
} catch (_) {} } catch (_) {}
// RID-3B-1: the explicit-Stop durability barrier replaces the single // RID-3B-1: the explicit-Stop durability barrier replaces the single
@ -597,7 +647,7 @@ class RideStateMachine extends StateNotifier<RideSession> {
// convergence, closes telemetry acceptance, and performs a final strict // convergence, closes telemetry acceptance, and performs a final strict
// drain - throwing if durability cannot be established (invariant I2). // drain - throwing if durability cannot be established (invariant I2).
await _establishTelemetryDurabilityForFinish(); await _establishTelemetryDurabilityForFinish();
await db.endRide( await db.endRide(
state.dbId!, state.dbId!,
RidesCompanion( RidesCompanion(
@ -616,12 +666,14 @@ class RideStateMachine extends StateNotifier<RideSession> {
); );
try { try {
ref.read(fieldTestLoggerProvider).log( ref
eventType: FieldTestEventType.rideFinalized, .read(fieldTestLoggerProvider)
severity: EventSeverity.info, .log(
message: 'Ride record #${state.dbId} finalized in database', eventType: FieldTestEventType.rideFinalized,
rideId: state.dbId, severity: EventSeverity.info,
); message: 'Ride record #${state.dbId} finalized in database',
rideId: state.dbId,
);
} catch (_) {} } catch (_) {}
// Check for personal records // Check for personal records
@ -634,7 +686,7 @@ class RideStateMachine extends StateNotifier<RideSession> {
longestRideMiles: state.tripDistanceMiles, longestRideMiles: state.tripDistanceMiles,
greatestElevationGain: state.elevationGain, greatestElevationGain: state.elevationGain,
); );
await recordsService.checkAndSaveRecords(runStats, (recordName, value) { await recordsService.checkAndSaveRecords(runStats, (recordName, value) {
audioAlertService.speak('New record: $recordName'); audioAlertService.speak('New record: $recordName');
}); });
@ -644,11 +696,13 @@ class RideStateMachine extends StateNotifier<RideSession> {
try { try {
final logger = ref.read(fieldTestLoggerProvider); final logger = ref.read(fieldTestLoggerProvider);
if (logger.isSessionActive) { if (logger.isSessionActive) {
logger.telemetrySamplesLost = _telemetrySamplesLost;
await logger.stopSession(); await logger.stopSession();
logger.log( logger.log(
eventType: FieldTestEventType.rideSummaryCreated, eventType: FieldTestEventType.rideSummaryCreated,
severity: EventSeverity.info, 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, rideId: state.dbId,
); );
} }
@ -713,23 +767,29 @@ class RideStateMachine extends StateNotifier<RideSession> {
try { try {
if (sample.quality != null) { if (sample.quality != null) {
ref.read(fieldTestLoggerProvider).processTelemetrySample(sample, sample.quality!); ref
.read(fieldTestLoggerProvider)
.processTelemetrySample(sample, sample.quality!);
} }
} catch (_) {} } catch (_) {}
} }
static const int _maxBatchCapacity = 1000; static const int _maxBatchCapacity = 1000;
void _queueForBatch(TelemetrySample sample) { void _queueForBatch(TelemetrySample sample) {
if (state.dbId == null) return; if (state.dbId == null) return;
// Bound queue to prevent memory leaks during extended DB locks // Bound queue to prevent memory leaks during extended DB locks
if (_telemetryBatch.length >= _maxBatchCapacity) { if (_telemetryBatch.length >= _maxBatchCapacity) {
_telemetryBatch.removeAt(0); _telemetryBatch.removeAt(0);
_telemetrySamplesLost++;
try { try {
ref.read(fieldTestLoggerProvider).log( final logger = ref.read(fieldTestLoggerProvider);
logger.recordTelemetryLoss(1);
logger.log(
eventType: FieldTestEventType.databaseQueueOverflow, 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, severity: EventSeverity.warning,
); );
} catch (_) {} } catch (_) {}
@ -752,12 +812,14 @@ class RideStateMachine extends StateNotifier<RideSession> {
accelZ: drift.Value(sample.accelZ), accelZ: drift.Value(sample.accelZ),
longitudinalG: drift.Value(sample.longitudinalG), longitudinalG: drift.Value(sample.longitudinalG),
lateralG: drift.Value(sample.lateralG), 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. /// RID-3A ownership semantics, plus RID-3B-1 active-flush publication.
/// ///
/// Takes exclusive ownership of the current buffer contents, removes them from /// Takes exclusive ownership of the current buffer contents, removes them from
@ -840,7 +902,9 @@ class RideStateMachine extends StateNotifier<RideSession> {
// RID-2's existing UI boundary containment handles the rejection and the // RID-2's existing UI boundary containment handles the rejection and the
// ride remains retryable. RID-3A has already restored any failed owned // ride remains retryable. RID-3A has already restored any failed owned
// snapshot to the front of the buffer in correct order. // snapshot to the front of the buffer in correct order.
state = state.copyWith(healthStatus: RecordingHealthStatus.recordingActive); state = state.copyWith(
healthStatus: RecordingHealthStatus.recordingActive,
);
throw StateError( throw StateError(
'Ride finalization aborted: telemetry durability could not be established', 'Ride finalization aborted: telemetry durability could not be established',
); );
@ -873,6 +937,28 @@ class RideStateMachine extends StateNotifier<RideSession> {
// arrived during the await, so a later legitimate flush can retry them. // arrived during the await, so a later legitimate flush can retry them.
// No automatic retry worker is added. // No automatic retry worker is added.
_telemetryBatch.insertAll(0, batchToInsert); _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 // 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 // database health evidence must not advance and is marked non-operational
// until a real commit succeeds. // until a real commit succeeds.
@ -882,11 +968,13 @@ class RideStateMachine extends StateNotifier<RideSession> {
// case; it remains bounded either way. // case; it remains bounded either way.
if (mounted) { if (mounted) {
try { try {
ref.read(fieldTestLoggerProvider).log( ref
eventType: FieldTestEventType.databaseQueueWarning, .read(fieldTestLoggerProvider)
message: 'Database batch write retry: $e', .log(
severity: EventSeverity.warning, eventType: FieldTestEventType.databaseQueueWarning,
); message: 'Database batch write retry: $e',
severity: EventSeverity.warning,
);
} catch (_) {} } catch (_) {}
} }
return false; return false;
@ -917,8 +1005,11 @@ class RideStateMachine extends StateNotifier<RideSession> {
_lastDbCommitMonotonicMs = _latestObservedMonotonicMs; _lastDbCommitMonotonicMs = _latestObservedMonotonicMs;
_dbPersistenceFailed = false; _dbPersistenceFailed = false;
state = state.copyWith( state = state.copyWith(
totalSamplesPersisted: state.totalSamplesPersisted + batchToInsert.length, totalSamplesPersisted:
health: state.health.copyWith(lastDatabaseWriteTime: DateTime.now()), state.totalSamplesPersisted + batchToInsert.length,
health: state.health.copyWith(
lastDatabaseWriteTime: DateTime.now(),
),
); );
_evaluateRecordingHealth(); _evaluateRecordingHealth();
} catch (_) { } catch (_) {
@ -993,7 +1084,9 @@ class RideStateMachine extends StateNotifier<RideSession> {
// Incrementally persist ongoing ride summary stats to SQLite so crashes/terminations never lose stats // Incrementally persist ongoing ride summary stats to SQLite so crashes/terminations never lose stats
if (state.dbId != null && state.state == RideState.riding) { if (state.dbId != null && state.state == RideState.riding) {
try { 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( RidesCompanion(
distanceMiles: drift.Value(state.tripDistanceMiles), distanceMiles: drift.Value(state.tripDistanceMiles),
topSpeedMph: drift.Value(state.topSpeedMph), topSpeedMph: drift.Value(state.topSpeedMph),
@ -1031,10 +1124,12 @@ class RideStateMachine extends StateNotifier<RideSession> {
newTopSpeed = sample.speed!; newTopSpeed = sample.speed!;
} }
} }
// Continuous sub-second time accumulation across 20Hz frames // Continuous sub-second time accumulation across 20Hz frames
if (_lastSample != null && sample.speed != null) { 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 (dt > 0 && dt < 2.0) {
if (sample.speed! > 1.8) { if (sample.speed! > 1.8) {
_movingTimeAccumulator += dt; _movingTimeAccumulator += dt;
@ -1049,23 +1144,34 @@ class RideStateMachine extends StateNotifier<RideSession> {
// Speed-gate dynamic motorcycle metrics (require speed >= 3.0 MPH to prevent stationary kickstand / dismount artifacts from registering as personal riding records) // 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; final isDynamicallyMoving = sample.speed != null && sample.speed! >= 3.0;
if (isDynamicallyMoving && sample.leanAngle != null && sample.leanAngle!.abs() <= 65.0) { if (isDynamicallyMoving &&
if (sample.leanAngle! < 0 && sample.leanAngle!.abs() > state.maxLeanLeft) { sample.leanAngle != null &&
sample.leanAngle!.abs() <= 65.0) {
if (sample.leanAngle! < 0 &&
sample.leanAngle!.abs() > state.maxLeanLeft) {
newMaxLeanLeft = sample.leanAngle!.abs(); 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!; newMaxLeanRight = sample.leanAngle!;
} }
} }
if (isDynamicallyMoving && sample.longitudinalG != null && sample.longitudinalG!.abs() <= 2.5) { if (isDynamicallyMoving &&
if (sample.longitudinalG! > 0 && sample.longitudinalG! > state.maxAccelG) { sample.longitudinalG != null &&
sample.longitudinalG!.abs() <= 2.5) {
if (sample.longitudinalG! > 0 &&
sample.longitudinalG! > state.maxAccelG) {
newMaxAccelG = sample.longitudinalG!; 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(); 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( final distMeters = Geolocator.distanceBetween(
_lastSample!.latitude!, _lastSample!.latitude!,
_lastSample!.longitude!, _lastSample!.longitude!,
@ -1074,39 +1180,42 @@ class RideStateMachine extends StateNotifier<RideSession> {
); );
// Accumulate distance ONLY if moving and GPS accuracy is reasonable (suppresses stationary jitter at stop lights) // 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 isMoving =
final hasGoodGps = sample.gpsAccuracy != null && sample.gpsAccuracy! <= 35.0; (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) { if (isMoving && hasGoodGps && distMeters > 0.3 && distMeters < 150.0) {
_accumulatedDistance += distMeters; _accumulatedDistance += distMeters;
} }
// Elevation tracking // Elevation tracking
if (sample.altitude != null && _lastSample!.altitude != null) { if (sample.altitude != null && _lastSample!.altitude != null) {
final elevationDelta = sample.altitude! - _lastSample!.altitude!; final elevationDelta = sample.altitude! - _lastSample!.altitude!;
// Threshold filter to prevent phantom gain (e.g. > 1 meter change) // Threshold filter to prevent phantom gain (e.g. > 1 meter change)
if (elevationDelta.abs() > 1.0 && elevationDelta.abs() < 50.0) { if (elevationDelta.abs() > 1.0 && elevationDelta.abs() < 50.0) {
if (elevationDelta > 0) { if (elevationDelta > 0) {
newElevationGain += elevationDelta; newElevationGain += elevationDelta;
} else { } else {
newElevationLoss += elevationDelta.abs(); newElevationLoss += elevationDelta.abs();
} }
} }
} }
} }
double tripDistanceMiles = _accumulatedDistance / 1609.34; double tripDistanceMiles = _accumulatedDistance / 1609.34;
// Calculate avg speed (only moving time) // Calculate avg speed (only moving time)
double avgSpeed = 0; double avgSpeed = 0;
if (newMovingTime > 0) { if (newMovingTime > 0) {
// miles per hour // miles per hour
avgSpeed = (tripDistanceMiles / newMovingTime) * 3600; avgSpeed = (tripDistanceMiles / newMovingTime) * 3600;
} }
_detectEvents(sample); _detectEvents(sample);
_lastSample = sample; _lastSample = sample;
state = state.copyWith( state = state.copyWith(
topSpeedMph: newTopSpeed, topSpeedMph: newTopSpeed,
maxLeanLeft: newMaxLeanLeft, maxLeanLeft: newMaxLeanLeft,
@ -1121,7 +1230,7 @@ class RideStateMachine extends StateNotifier<RideSession> {
averageSpeedMph: avgSpeed, averageSpeedMph: avgSpeed,
); );
} }
void _detectEvents(TelemetrySample sample) { void _detectEvents(TelemetrySample sample) {
if (state.dbId == null) return; if (state.dbId == null) return;
final now = DateTime.now(); final now = DateTime.now();
@ -1131,32 +1240,48 @@ class RideStateMachine extends StateNotifier<RideSession> {
final lastTime = _lastEventTimes[type]; final lastTime = _lastEventTimes[type];
if (lastTime == null || now.difference(lastTime) > _eventCooldown) { if (lastTime == null || now.difference(lastTime) > _eventCooldown) {
_lastEventTimes[type] = now; _lastEventTimes[type] = now;
db.into(db.rideEvents).insert(RideEventsCompanion.insert( db
rideId: state.dbId!, .into(db.rideEvents)
timestamp: now, .insert(
eventType: type, RideEventsCompanion.insert(
value: value, rideId: state.dbId!,
latitude: drift.Value(sample.latitude), timestamp: now,
longitude: drift.Value(sample.longitude), eventType: type,
)); value: value,
latitude: drift.Value(sample.latitude),
longitude: drift.Value(sample.longitude),
),
);
} }
} }
} }
if (sample.longitudinalG != null) { if (sample.longitudinalG != null) {
checkAndLogEvent('Hard Braking', sample.longitudinalG!, sample.longitudinalG! < -0.8); checkAndLogEvent(
checkAndLogEvent('Hard Acceleration', sample.longitudinalG!, sample.longitudinalG! > 0.6); 'Hard Braking',
sample.longitudinalG!,
sample.longitudinalG! < -0.8,
);
checkAndLogEvent(
'Hard Acceleration',
sample.longitudinalG!,
sample.longitudinalG! > 0.6,
);
} }
if (sample.leanAngle != null) { 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) { if (sample.speed != null) {
checkAndLogEvent('High Speed', sample.speed!, sample.speed! > 100.0); checkAndLogEvent('High Speed', sample.speed!, sample.speed! > 100.0);
} }
} }
@override @override
void dispose() { void dispose() {
_telemetrySub?.cancel(); _telemetrySub?.cancel();
@ -1166,10 +1291,17 @@ class RideStateMachine extends StateNotifier<RideSession> {
} }
} }
final rideStateMachineProvider = StateNotifierProvider<RideStateMachine, RideSession>((ref) { final rideStateMachineProvider =
final db = ref.watch(databaseProvider); StateNotifierProvider<RideStateMachine, RideSession>((ref) {
final telemetryService = ref.watch(telemetryServiceProvider); final db = ref.watch(databaseProvider);
final recordsService = ref.watch(recordsServiceProvider.notifier); final telemetryService = ref.watch(telemetryServiceProvider);
final audioAlertService = ref.watch(audioAlertServiceProvider); final recordsService = ref.watch(recordsServiceProvider.notifier);
return RideStateMachine(db, telemetryService, recordsService, audioAlertService, ref); final audioAlertService = ref.watch(audioAlertServiceProvider);
}); return RideStateMachine(
db,
telemetryService,
recordsService,
audioAlertService,
ref,
);
});

View File

@ -93,6 +93,20 @@ class FieldTestLogger {
int get totalEnqueued => _totalEnqueued; int get totalEnqueued => _totalEnqueued;
int get totalFlushed => _totalFlushed; int get totalFlushed => _totalFlushed;
int get droppedEventsCount => _droppedEventsCount; 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 oldestDroppedItem => _oldestDroppedItem;
String? get newestDroppedItem => _newestDroppedItem; String? get newestDroppedItem => _newestDroppedItem;
int get lastFlushDurationMs => _lastFlushDurationMs; int get lastFlushDurationMs => _lastFlushDurationMs;
@ -267,6 +281,7 @@ class FieldTestLogger {
_oldestDroppedItem = null; _oldestDroppedItem = null;
_newestDroppedItem = null; _newestDroppedItem = null;
_failedFlushCount = 0; _failedFlushCount = 0;
_telemetrySamplesLost = 0;
_recentSampleRateHz = 0.0; _recentSampleRateHz = 0.0;
_recentSampleCount = 0; _recentSampleCount = 0;
@ -700,7 +715,12 @@ class FieldTestLogger {
} }
/// Generates forensic summary from recorded ride telemetry and events /// Generates forensic summary from recorded ride telemetry and events
Future<FieldTestSummary> generateAndSaveSummary(int rideId, {bool wasRecovered = false}) async { Future<FieldTestSummary> 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 ride = await (_db.select(_db.rides)..where((r) => r.id.equals(rideId))).getSingleOrNull();
final startTime = ride?.startTime ?? DateTime.now(); final startTime = ride?.startTime ?? DateTime.now();
final endTime = ride?.endTime ?? DateTime.now(); final endTime = ride?.endTime ?? DateTime.now();
@ -815,7 +835,7 @@ class FieldTestLogger {
timestampGaps: _telemetryGapCount, timestampGaps: _telemetryGapCount,
gpsFixCount: gpsFixCount, gpsFixCount: gpsFixCount,
totalGpsAccuracy: accuracySum, totalGpsAccuracy: accuracySum,
databaseSamplesLost: _droppedEventsCount, databaseSamplesLost: effectiveTelemetryLost,
imuSaturationCount: _imuSaturationCount, imuSaturationCount: _imuSaturationCount,
isMountCalibrated: true, isMountCalibrated: true,
); );
@ -852,7 +872,7 @@ class FieldTestLogger {
bestZeroTo60: best0to60, bestZeroTo60: best0to60,
bestZeroTo100: best0to100, bestZeroTo100: best0to100,
databaseSamplesWritten: totalSamples, databaseSamplesWritten: totalSamples,
databaseSamplesLost: _droppedEventsCount, databaseSamplesLost: effectiveTelemetryLost,
eventCount: events.length, eventCount: events.length,
overallConfidence: qualityReport.overallScore / 100.0, overallConfidence: qualityReport.overallScore / 100.0,
rideQualityScore: qualityReport.overallScore, rideQualityScore: qualityReport.overallScore,
@ -896,7 +916,7 @@ class FieldTestLogger {
bestZeroTo60: drift.Value(best0to60), bestZeroTo60: drift.Value(best0to60),
bestZeroTo100: drift.Value(best0to100), bestZeroTo100: drift.Value(best0to100),
databaseSamplesWritten: drift.Value(totalSamples), databaseSamplesWritten: drift.Value(totalSamples),
databaseSamplesLost: drift.Value(_droppedEventsCount), databaseSamplesLost: drift.Value(effectiveTelemetryLost),
eventCount: drift.Value(events.length), eventCount: drift.Value(events.length),
overallConfidence: drift.Value(qualityReport.overallScore / 100.0), overallConfidence: drift.Value(qualityReport.overallScore / 100.0),
rideQualityScore: drift.Value(qualityReport.overallScore), rideQualityScore: drift.Value(qualityReport.overallScore),

View File

@ -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:drift/native.dart';
import 'package:shared_preferences/shared_preferences.dart'; import 'package:shared_preferences/shared_preferences.dart';
@ -29,36 +29,118 @@ void main() {
expect(logger.isSessionActive, isTrue); expect(logger.isSessionActive, isTrue);
expect(logger.activeRideId, 10); 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(); final summary = await logger.stopSession();
expect(logger.isSessionActive, isFalse); expect(logger.isSessionActive, isFalse);
expect(summary.rideId, 10); expect(summary.rideId, 10);
}); });
test('Queue drops lowest-priority events when buffer capacity is reached', () { test(
logger.startSession(11); 'Queue drops lowest-priority events when buffer capacity is reached',
() {
logger.startSession(11);
// Fill queue with 1,000 info events // Fill queue with 1,000 info events
for (int i = 0; i < 1005; i++) { 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( logger.log(
eventType: FieldTestEventType.databaseFlush, eventType: FieldTestEventType.databaseRecovery,
severity: EventSeverity.info, severity: EventSeverity.critical,
message: 'Flush event $i', message: 'Critical DB Recovery',
rideId: 11, rideId: 11,
); );
}
// Add a critical event expect(logger.droppedEventsCount, greaterThanOrEqualTo(5));
logger.log( expect(
eventType: FieldTestEventType.databaseRecovery, logger.timelineEvents.any(
severity: EventSeverity.critical, (e) => e.severity == EventSeverity.critical,
message: 'Critical DB Recovery', ),
rideId: 11, isTrue,
); );
},
);
expect(logger.droppedEventsCount, greaterThanOrEqualTo(5)); test(
expect(logger.timelineEvents.any((e) => e.severity == EventSeverity.critical), isTrue); '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);
}); });
}); });
} }

View File

@ -23,10 +23,12 @@ import 'package:drift/native.dart';
import 'package:drift/drift.dart' as drift; import 'package:drift/drift.dart' as drift;
import 'package:motorcycle_dash/database/database.dart' import 'package:motorcycle_dash/database/database.dart'
hide TelemetrySample, FieldTestSummary, FieldTestEvent; hide TelemetrySample, FieldTestSummary, FieldTestEvent;
import 'package:motorcycle_dash/database/database.dart' as dbgen import 'package:motorcycle_dash/database/database.dart'
as dbgen
show TelemetrySample; show TelemetrySample;
import 'package:motorcycle_dash/models/telemetry_sample.dart'; import 'package:motorcycle_dash/models/telemetry_sample.dart';
import 'package:motorcycle_dash/models/field_test_summary.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/telemetry_service.dart';
import 'package:motorcycle_dash/services/location_service.dart'; import 'package:motorcycle_dash/services/location_service.dart';
import 'package:motorcycle_dash/services/sensor_fusion_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/field_test/field_test_logger.dart';
import 'package:motorcycle_dash/services/validation/telemetry_validator.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/rides/ride_state_machine.dart';
import 'package:motorcycle_dash/models/recording_health_status.dart';
import 'package:motorcycle_dash/features/records/records_service.dart'; import 'package:motorcycle_dash/features/records/records_service.dart';
/// Drift's generated row class for telemetry_samples is also named /// Drift's generated row class for telemetry_samples is also named
@ -53,7 +56,8 @@ class _Rid3aAppDatabase extends AppDatabase {
@override @override
Future<void> batch( Future<void> batch(
FutureOr<void> Function(drift.Batch batch) runInBatch) async { FutureOr<void> Function(drift.Batch batch) runInBatch,
) async {
batchInvocations++; batchInvocations++;
// Gate FIRST, then optionally fail, so a flush can be parked in-flight and // 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. // made to fail on release - required by the ordered-restoration test.
@ -79,6 +83,51 @@ class _Rid3aFieldTestLogger extends FieldTestLogger {
_Rid3aFieldTestLogger(super.db); _Rid3aFieldTestLogger(super.db);
int stopSessionInvocations = 0; int stopSessionInvocations = 0;
int overflowEventCount = 0;
final List<FieldTestEventType> 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<String, dynamic>? 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 @override
Future<void> flush() async {} Future<void> flush() async {}
@ -122,7 +171,7 @@ class FakeTelemetryService extends TelemetryService {
class _NullLocationService extends Fake implements LocationService { class _NullLocationService extends Fake implements LocationService {
@override @override
void Function(Position position, bool accepted, String? rejectionReason)? void Function(Position position, bool accepted, String? rejectionReason)?
onGpsDiagnostic; onGpsDiagnostic;
@override @override
Stream<Position> getFilteredPositionStream() => const Stream.empty(); Stream<Position> getFilteredPositionStream() => const Stream.empty();
} }
@ -238,9 +287,18 @@ void main() {
} }
} }
Future<List<TelemetryRow>> persistedRows() => (db.select(db.telemetrySamples) Future<List<TelemetryRow>> persistedRows() =>
..orderBy([(t) => drift.OrderingTerm(expression: t.id)])) (db.select(db.telemetrySamples)..orderBy([
.get(); (t) => drift.OrderingTerm(
expression: t.timestamp,
mode: drift.OrderingMode.asc,
),
(t) => drift.OrderingTerm(
expression: t.id,
mode: drift.OrderingMode.asc,
),
]))
.get();
Map<int, int> countsByIndex(List<TelemetryRow> rows) { Map<int, int> countsByIndex(List<TelemetryRow> rows) {
final counts = <int, int>{}; final counts = <int, int>{};
@ -258,13 +316,12 @@ void main() {
..sort(); ..sort();
List<int> missingOf(int total, Map<int, int> counts) => [ List<int> missingOf(int total, Map<int, int> counts) => [
for (var i = 0; i < total; i++) for (var i = 0; i < total; i++)
if ((counts[i] ?? 0) == 0) i, if ((counts[i] ?? 0) == 0) i,
]; ];
group('RID-3A: telemetry batch ownership', () { group('RID-3A: telemetry batch ownership', () {
test('single flush persists every sample exactly once, in order', test('single flush persists every sample exactly once, in order', () async {
() async {
final container = buildContainer(); final container = buildContainer();
final notifier = container.read(rideStateMachineProvider.notifier); final notifier = container.read(rideStateMachineProvider.notifier);
final rideId = await startRide(container); final rideId = await startRide(container);
@ -286,14 +343,18 @@ void main() {
expect(counts.length, 10, reason: 'N distinct identities'); expect(counts.length, 10, reason: 'N distinct identities');
expect(duplicatedIn(counts), isEmpty, reason: '0 duplicates'); expect(duplicatedIn(counts), isEmpty, reason: '0 duplicates');
expect(missingOf(10, counts), isEmpty, reason: '0 missing'); expect(missingOf(10, counts), isEmpty, reason: '0 missing');
expect(rows.map((r) => r.speed).toList(), expect(rows.map((r) => r.speed).toList(), [
[for (var i = 0; i < 10; i++) 1000.0 + i], for (var i = 0; i < 10; i++) 1000.0 + i,
reason: 'correct chronological order'); ], reason: 'correct chronological order');
expect(rows.map((r) => r.rideId).toSet(), {rideId}, expect(rows.map((r) => r.rideId).toSet(), {
reason: 'correct ride association'); rideId,
}, reason: 'correct ride association');
expect(session.totalSamplesReceived, 10); expect(session.totalSamplesReceived, 10);
expect(session.totalSamplesPersisted, 10, expect(
reason: 'buffer empty after a successful final flush'); session.totalSamplesPersisted,
10,
reason: 'buffer empty after a successful final flush',
);
}); });
test('overlapping periodic and finishRide flushes persist all 30 accepted ' 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 // Tick 1: the periodic flush takes exclusive ownership of 0..9, removes
// them from the live buffer, and parks inside db.batch. // them from the live buffer, and parks inside db.batch.
await Future<void>.delayed(tickWait); await Future<void>.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; final flushA = db.batchGates.single;
// Newer telemetry (indices 10..29) arrives while A is still in flight. // 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. // drains the newer samples only after A has committed.
final finishing = notifier.finishRide(); final finishing = notifier.finishRide();
await pumpEventQueue(times: 30); await pumpEventQueue(times: 30);
expect(db.batchInvocations, 1, expect(
reason: 'while the owner is parked, no second flush may take ownership'); 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 // 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 // barrier's own drain parks as the next gate. That yields a deterministic
@ -339,10 +407,16 @@ void main() {
final rowsOwnedByA = await persistedRows(); final rowsOwnedByA = await persistedRows();
final countsOwnedByA = countsByIndex(rowsOwnedByA); final countsOwnedByA = countsByIndex(rowsOwnedByA);
expect(duplicatedIn(countsOwnedByA), isEmpty, expect(
reason: 'no duplication from overlapping ownership'); duplicatedIn(countsOwnedByA),
expect(missingOf(10, countsOwnedByA), isEmpty, isEmpty,
reason: 'A persisted its whole owned snapshot'); 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 // 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 // own identity set, and none of the newer samples - those remain buffered
// for later ownership. This is observed BEFORE finishRide can complete the // 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 // intentionally drains all accepted pre-boundary telemetry before
// finalizing. The previous "only 10 rows after await finishing" expectation // finalizing. The previous "only 10 rows after await finishing" expectation
// depended on the finish flush being a no-op, which RID-3B-1 removed. // depended on the finish flush being a no-op, which RID-3B-1 removed.
expect(countsOwnedByA.keys.toList()..sort(), expect(
[for (var i = 0; i < 10; i++) i], countsOwnedByA.keys.toList()..sort(),
reason: 'flush A owns exactly indices 0-9'); [for (var i = 0; i < 10; i++) i],
expect(countsOwnedByA.keys.where((i) => i >= 10).toList(), isEmpty, reason: 'flush A owns exactly indices 0-9',
reason: 'no newer sample was included in flush A write'); );
expect(db.batchGates, hasLength(2), expect(
reason: 'the barrier now owns the newer batch and is parked'); 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 // Release the barrier's drain and let finishRide complete. It persists
// 10..29 itself, so no later periodic flush is required. // 10..29 itself, so no later periodic flush is required.
@ -376,23 +458,36 @@ void main() {
debugPrint('distinct identities : ${counts.length}'); debugPrint('distinct identities : ${counts.length}');
debugPrint('duplicated indices : ${duplicatedIn(counts)}'); debugPrint('duplicated indices : ${duplicatedIn(counts)}');
debugPrint('missing indices : ${missingOf(30, counts)}'); debugPrint('missing indices : ${missingOf(30, counts)}');
debugPrint('received / persisted : ' debugPrint(
'${session.totalSamplesReceived} / ${session.totalSamplesPersisted}'); 'received / persisted : '
debugPrint('rideIds : ${rows.map((r) => r.rideId).toSet()}'); '${session.totalSamplesReceived} / ${session.totalSamplesPersisted}',
);
debugPrint(
'rideIds : ${rows.map((r) => r.rideId).toSet()}',
);
// Previously: 30 rows / 20 distinct, indices 0-9 twice, 10-19 destroyed. // Previously: 30 rows / 20 distinct, indices 0-9 twice, 10-19 destroyed.
expect(rows, hasLength(30), reason: '30 accepted -> 30 durable rows'); expect(rows, hasLength(30), reason: '30 accepted -> 30 durable rows');
expect(counts.length, 30, reason: '30 distinct identities'); expect(counts.length, 30, reason: '30 distinct identities');
expect(duplicatedIn(counts), isEmpty, reason: '0 duplicates'); expect(duplicatedIn(counts), isEmpty, reason: '0 duplicates');
expect(missingOf(30, counts), isEmpty, reason: '0 missing'); expect(missingOf(30, counts), isEmpty, reason: '0 missing');
expect(rows.map((r) => r.speed).toList(), expect(
[for (var i = 0; i < 30; i++) 1000.0 + i], rows.map((r) => r.speed).toList(),
reason: 'chronological identity order preserved'); [for (var i = 0; i < 30; i++) 1000.0 + i],
expect(rows.map((r) => r.rideId).toSet(), {rideId}, reason: 'chronological identity order preserved',
reason: 'every sample against the correct ride'); );
expect(session.totalSamplesPersisted, 30, expect(
reason: 'counters agree with distinct durable samples; nothing left ' rows.map((r) => r.rideId).toSet(),
'intentionally buffered after the final flush'); {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 ' 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()}'); debugPrint('order : ${rows.map((r) => r.speed).toList()}');
expect(rows, hasLength(10)); 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(missingOf(10, counts), isEmpty, reason: 'nothing missing');
expect(rows.map((r) => r.speed).toList(), expect(rows.map((r) => r.speed).toList(), [
[for (var i = 0; i < 10; i++) 1000.0 + i], for (var i = 0; i < 10; i++) 1000.0 + i,
reason: 'A persisted before B'); ], reason: 'A persisted before B');
expect(rows.map((r) => r.rideId).toSet(), {rideId}); expect(rows.map((r) => r.rideId).toSet(), {rideId});
expect(session.totalSamplesPersisted, 10); expect(session.totalSamplesPersisted, 10);
}); });
@ -468,11 +567,17 @@ void main() {
await pumpEventQueue(times: 40); await pumpEventQueue(times: 40);
var session = container.read(rideStateMachineProvider); var session = container.read(rideStateMachineProvider);
expect(session.totalSamplesPersisted, 0, expect(
reason: 'a failed persistence must not increment the success counter'); session.totalSamplesPersisted,
0,
reason: 'a failed persistence must not increment the success counter',
);
expect(await persistedRows(), isEmpty, reason: 'nothing durable yet'); expect(await persistedRows(), isEmpty, reason: 'nothing durable yet');
expect(session.state, RideState.riding, expect(
reason: 'telemetry recording continues so a later flush can recover'); session.state,
RideState.riding,
reason: 'telemetry recording continues so a later flush can recover',
);
// Existing periodic timer provides the legitimate retry. No automatic // Existing periodic timer provides the legitimate retry. No automatic
// retry worker is introduced. // retry worker is introduced.
@ -490,19 +595,332 @@ void main() {
debugPrint('duplicated : ${duplicatedIn(counts)}'); debugPrint('duplicated : ${duplicatedIn(counts)}');
debugPrint('missing : ${missingOf(6, counts)}'); debugPrint('missing : ${missingOf(6, counts)}');
debugPrint('order : ${rows.map((r) => r.speed).toList()}'); debugPrint('order : ${rows.map((r) => r.speed).toList()}');
debugPrint('received / persisted : ' debugPrint(
'${session.totalSamplesReceived} / ${session.totalSamplesPersisted}'); 'received / persisted : '
'${session.totalSamplesReceived} / ${session.totalSamplesPersisted}',
);
expect(rows, hasLength(6)); 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(missingOf(6, counts), isEmpty, reason: 'no rows missing');
expect(rows.map((r) => r.speed).toList(), expect(
[1000.0, 1001.0, 1002.0, 1003.0, 1004.0, 1005.0], rows.map((r) => r.speed).toList(),
reason: 'failed snapshot A B C restored AHEAD of newer D E F, both ' [1000.0, 1001.0, 1002.0, 1003.0, 1004.0, 1005.0],
'internal orders preserved'); 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(rows.map((r) => r.rideId).toSet(), {rideId});
expect(session.totalSamplesPersisted, 6, expect(
reason: 'counter reflects only the successfully persisted snapshot'); 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);
},
);
}); });
} }