493 lines
19 KiB
Dart
493 lines
19 KiB
Dart
// RID-3A regression coverage: telemetry batch ownership / exactly-once buffer.
|
|
//
|
|
// Converted from the RID-3 forensic reproduction. Before this fix, two
|
|
// overlapping legitimate flushes (the 3 s periodic timer against the flush
|
|
// awaited by finishRide) shared one mutable buffer that was cleared AFTER the
|
|
// persistence await by count rather than identity. For 30 accepted samples that
|
|
// produced 30 rows representing only 20 distinct samples: indices 0-9 twice,
|
|
// indices 10-19 destroyed, indices 20-29 once - while the counters reported a
|
|
// perfect 30/30.
|
|
//
|
|
// Race windows are controlled with Completer gates on db.batch, not sleeps. The
|
|
// only real-time waits are for the production Timer.periodic(3 s) to fire,
|
|
// because that interval is a hardcoded constant with no injectable seam and
|
|
// creating one would be a production change.
|
|
import 'dart:async';
|
|
|
|
import 'package:flutter/foundation.dart';
|
|
import 'package:flutter_test/flutter_test.dart';
|
|
import 'package:flutter_riverpod/flutter_riverpod.dart';
|
|
import 'package:geolocator/geolocator.dart';
|
|
import 'package:shared_preferences/shared_preferences.dart';
|
|
import 'package:drift/native.dart';
|
|
import 'package:drift/drift.dart' as drift;
|
|
import 'package:motorcycle_dash/database/database.dart'
|
|
hide TelemetrySample, FieldTestSummary, FieldTestEvent;
|
|
import 'package:motorcycle_dash/database/database.dart' as dbgen
|
|
show TelemetrySample;
|
|
import 'package:motorcycle_dash/models/telemetry_sample.dart';
|
|
import 'package:motorcycle_dash/models/field_test_summary.dart';
|
|
import 'package:motorcycle_dash/services/telemetry_service.dart';
|
|
import 'package:motorcycle_dash/services/location_service.dart';
|
|
import 'package:motorcycle_dash/services/sensor_fusion_service.dart';
|
|
import 'package:motorcycle_dash/services/compass_service.dart';
|
|
import 'package:motorcycle_dash/services/audio_alert_service.dart';
|
|
import 'package:motorcycle_dash/services/field_test/field_test_logger.dart';
|
|
import 'package:motorcycle_dash/services/validation/telemetry_validator.dart';
|
|
import 'package:motorcycle_dash/features/rides/ride_state_machine.dart';
|
|
import 'package:motorcycle_dash/features/records/records_service.dart';
|
|
|
|
/// Drift's generated row class for telemetry_samples is also named
|
|
/// TelemetrySample, colliding with the telemetry model.
|
|
typedef TelemetryRow = dbgen.TelemetrySample;
|
|
|
|
/// Gates, counts and can fail db.batch - the persistence API used by
|
|
/// RideStateMachine._flushBatch() (ride_state_machine.dart:560).
|
|
class _Rid3aAppDatabase extends AppDatabase {
|
|
_Rid3aAppDatabase() : super.forTesting(NativeDatabase.memory());
|
|
|
|
int batchInvocations = 0;
|
|
bool gateBatch = false;
|
|
final List<Completer<void>> batchGates = [];
|
|
int batchFailuresRemaining = 0;
|
|
|
|
@override
|
|
Future<void> batch(
|
|
FutureOr<void> Function(drift.Batch batch) runInBatch) async {
|
|
batchInvocations++;
|
|
// Gate FIRST, then optionally fail, so a flush can be parked in-flight and
|
|
// made to fail on release - required by the ordered-restoration test.
|
|
if (gateBatch) {
|
|
final gate = Completer<void>();
|
|
batchGates.add(gate);
|
|
await gate.future;
|
|
}
|
|
if (batchFailuresRemaining > 0) {
|
|
batchFailuresRemaining--;
|
|
throw StateError('RID-3A simulated telemetry persistence failure');
|
|
}
|
|
return super.batch(runInBatch);
|
|
}
|
|
}
|
|
|
|
/// FieldTestLogger runs its OWN 3 s periodic flush() -> _db.batch
|
|
/// (field_test_logger.dart:138, :592). Those batches are indistinguishable from
|
|
/// telemetry _flushBatch() batches at the executor level, so the logger flush is
|
|
/// neutralised here to keep db.batch accounting unambiguous. Field-test event
|
|
/// persistence is not under test in RID-3A.
|
|
class _Rid3aFieldTestLogger extends FieldTestLogger {
|
|
_Rid3aFieldTestLogger(super.db);
|
|
|
|
int stopSessionInvocations = 0;
|
|
|
|
@override
|
|
Future<void> flush() async {}
|
|
|
|
@override
|
|
Future<FieldTestSummary> stopSession() async {
|
|
stopSessionInvocations++;
|
|
return super.stopSession();
|
|
}
|
|
}
|
|
|
|
class MockAudioAlertService extends Fake implements AudioAlertService {
|
|
final List<String> spokenMessages = [];
|
|
@override
|
|
Future<void> speak(String text, {bool force = false}) async {
|
|
spokenMessages.add(text);
|
|
}
|
|
}
|
|
|
|
class FakeTelemetryService extends TelemetryService {
|
|
final StreamController<TelemetrySample> _controller =
|
|
StreamController<TelemetrySample>.broadcast();
|
|
|
|
FakeTelemetryService(
|
|
super.loc,
|
|
super.fusion,
|
|
super.compass,
|
|
super.logger,
|
|
super.validator,
|
|
);
|
|
|
|
@override
|
|
Stream<TelemetrySample> get telemetryStream => _controller.stream;
|
|
|
|
void emitSample(TelemetrySample sample) => _controller.add(sample);
|
|
|
|
@override
|
|
void dispose() => _controller.close();
|
|
}
|
|
|
|
class _NullLocationService extends Fake implements LocationService {
|
|
@override
|
|
void Function(Position position, bool accepted, String? rejectionReason)?
|
|
onGpsDiagnostic;
|
|
@override
|
|
Stream<Position> getFilteredPositionStream() => const Stream.empty();
|
|
}
|
|
|
|
class _NullSensorFusionService extends Fake implements SensorFusionService {
|
|
@override
|
|
Stream<double> get leanAngleStream => const Stream.empty();
|
|
@override
|
|
Stream<double> get pitchAngleStream => const Stream.empty();
|
|
@override
|
|
Stream<GForceData> get gForceStream => const Stream.empty();
|
|
@override
|
|
double get latestRollRateDegPerSec => 0.0;
|
|
@override
|
|
double get latestRawAccelX => 0.0;
|
|
@override
|
|
double get latestRawAccelY => 0.0;
|
|
@override
|
|
double get latestRawAccelZ => 0.0;
|
|
@override
|
|
double get latestRawGyroX => 0.0;
|
|
@override
|
|
double get latestRawGyroY => 0.0;
|
|
@override
|
|
double get latestRawGyroZ => 0.0;
|
|
}
|
|
|
|
class _NullCompassService extends Fake implements CompassService {
|
|
@override
|
|
Stream<double> get headingStream => const Stream.empty();
|
|
}
|
|
|
|
void main() {
|
|
TestWidgetsFlutterBinding.ensureInitialized();
|
|
|
|
/// The production flush interval is Timer.periodic(3 s)
|
|
/// (ride_state_machine.dart:145). Waiting slightly longer guarantees exactly
|
|
/// one tick deterministically.
|
|
const tickWait = Duration(milliseconds: 3300);
|
|
|
|
late _Rid3aAppDatabase db;
|
|
late _Rid3aFieldTestLogger logger;
|
|
late FakeTelemetryService fakeTelemetry;
|
|
late MockAudioAlertService mockAudio;
|
|
late TelemetryValidator validator;
|
|
late SharedPreferences prefs;
|
|
|
|
final sampleBase = DateTime(2026, 9, 10, 12);
|
|
|
|
setUp(() async {
|
|
SharedPreferences.setMockInitialValues({});
|
|
prefs = await SharedPreferences.getInstance();
|
|
db = _Rid3aAppDatabase();
|
|
logger = _Rid3aFieldTestLogger(db);
|
|
validator = TelemetryValidator();
|
|
fakeTelemetry = FakeTelemetryService(
|
|
_NullLocationService(),
|
|
_NullSensorFusionService(),
|
|
_NullCompassService(),
|
|
logger,
|
|
validator,
|
|
);
|
|
mockAudio = MockAudioAlertService();
|
|
});
|
|
|
|
tearDown(() async {
|
|
for (final g in db.batchGates) {
|
|
if (!g.isCompleted) g.complete();
|
|
}
|
|
db.batchGates.clear();
|
|
db.gateBatch = false;
|
|
fakeTelemetry.dispose();
|
|
logger.dispose();
|
|
await db.close();
|
|
});
|
|
|
|
ProviderContainer buildContainer() {
|
|
final container = ProviderContainer(
|
|
overrides: [
|
|
sharedPreferencesProvider.overrideWithValue(prefs),
|
|
databaseProvider.overrideWithValue(db),
|
|
telemetryServiceProvider.overrideWithValue(fakeTelemetry),
|
|
audioAlertServiceProvider.overrideWithValue(mockAudio),
|
|
fieldTestLoggerProvider.overrideWithValue(logger),
|
|
],
|
|
);
|
|
addTearDown(container.dispose);
|
|
return container;
|
|
}
|
|
|
|
Future<int> startRide(ProviderContainer container) async {
|
|
final notifier = container.read(rideStateMachineProvider.notifier);
|
|
await notifier.startRide();
|
|
return container.read(rideStateMachineProvider).dbId!;
|
|
}
|
|
|
|
/// Emits [count] uniquely identifiable samples from index [startIndex].
|
|
/// Identity is carried in `speed` (1000 + index) so every persisted row traces
|
|
/// back to exactly one injected sample, independent of DateTime precision.
|
|
void emitSamples(int startIndex, int count) {
|
|
for (var i = 0; i < count; i++) {
|
|
final index = startIndex + i;
|
|
fakeTelemetry.emitSample(
|
|
TelemetrySample(
|
|
timestamp: sampleBase.add(Duration(milliseconds: index * 50)),
|
|
latitude: 34.0 + (index * 0.0001),
|
|
longitude: -85.0,
|
|
speed: 1000.0 + index,
|
|
heading: 90.0,
|
|
gpsAccuracy: 3.0,
|
|
),
|
|
);
|
|
}
|
|
}
|
|
|
|
Future<List<TelemetryRow>> persistedRows() => (db.select(db.telemetrySamples)
|
|
..orderBy([(t) => drift.OrderingTerm(expression: t.id)]))
|
|
.get();
|
|
|
|
Map<int, int> countsByIndex(List<TelemetryRow> rows) {
|
|
final counts = <int, int>{};
|
|
for (final row in rows) {
|
|
final speed = row.speed;
|
|
if (speed == null || speed < 1000) continue;
|
|
counts[(speed - 1000.0).round()] =
|
|
(counts[(speed - 1000.0).round()] ?? 0) + 1;
|
|
}
|
|
return counts;
|
|
}
|
|
|
|
List<int> duplicatedIn(Map<int, int> counts) =>
|
|
counts.entries.where((e) => e.value > 1).map((e) => e.key).toList()
|
|
..sort();
|
|
|
|
List<int> missingOf(int total, Map<int, int> counts) => [
|
|
for (var i = 0; i < total; i++)
|
|
if ((counts[i] ?? 0) == 0) i,
|
|
];
|
|
|
|
group('RID-3A: telemetry batch ownership', () {
|
|
test('single flush persists every sample exactly once, in order',
|
|
() async {
|
|
final container = buildContainer();
|
|
final notifier = container.read(rideStateMachineProvider.notifier);
|
|
final rideId = await startRide(container);
|
|
|
|
emitSamples(0, 10);
|
|
await pumpEventQueue(times: 30);
|
|
expect(container.read(rideStateMachineProvider).totalSamplesReceived, 10);
|
|
|
|
// One legitimate flush source, no competition.
|
|
await notifier.finishRide();
|
|
await pumpEventQueue(times: 30);
|
|
|
|
final rows = await persistedRows();
|
|
final counts = countsByIndex(rows);
|
|
final session = container.read(rideStateMachineProvider);
|
|
|
|
expect(db.batchInvocations, 1);
|
|
expect(rows, hasLength(10), reason: 'N accepted -> N durable rows');
|
|
expect(counts.length, 10, reason: 'N distinct identities');
|
|
expect(duplicatedIn(counts), isEmpty, reason: '0 duplicates');
|
|
expect(missingOf(10, counts), isEmpty, reason: '0 missing');
|
|
expect(rows.map((r) => r.speed).toList(),
|
|
[for (var i = 0; i < 10; i++) 1000.0 + i],
|
|
reason: 'correct chronological order');
|
|
expect(rows.map((r) => r.rideId).toSet(), {rideId},
|
|
reason: 'correct ride association');
|
|
expect(session.totalSamplesReceived, 10);
|
|
expect(session.totalSamplesPersisted, 10,
|
|
reason: 'buffer empty after a successful final flush');
|
|
});
|
|
|
|
test('overlapping periodic and finishRide flushes persist all 30 accepted '
|
|
'samples exactly once', () async {
|
|
final container = buildContainer();
|
|
final notifier = container.read(rideStateMachineProvider.notifier);
|
|
final rideId = await startRide(container);
|
|
|
|
// Batch A: indices 0..9.
|
|
emitSamples(0, 10);
|
|
await pumpEventQueue(times: 30);
|
|
|
|
db.gateBatch = true;
|
|
// Tick 1: the periodic flush takes exclusive ownership of 0..9, removes
|
|
// them from the live buffer, and parks inside db.batch.
|
|
await Future<void>.delayed(tickWait);
|
|
expect(db.batchInvocations, 1, reason: 'periodic flush parked in db.batch');
|
|
final flushA = db.batchGates.single;
|
|
|
|
// Newer telemetry (indices 10..29) arrives while A is still in flight.
|
|
// Injected BEFORE the competing flush so the live buffer is provably
|
|
// non-empty - otherwise the competing flush's exclusion could be
|
|
// attributed to an empty buffer rather than to the ownership guard.
|
|
emitSamples(10, 20);
|
|
await pumpEventQueue(times: 40);
|
|
expect(container.read(rideStateMachineProvider).totalSamplesReceived, 30);
|
|
|
|
// Competing legitimate flush: finishRide() awaits _flushBatch(). It must
|
|
// take no ownership of the samples A already owns, and must not touch the
|
|
// newer ones either.
|
|
final finishing = notifier.finishRide();
|
|
await pumpEventQueue(times: 30);
|
|
expect(db.batchInvocations, 1,
|
|
reason: 'a competing flush while the guard is raised must take no '
|
|
'ownership of the live buffer');
|
|
|
|
// Release A: it persists exactly its owned snapshot (0..9) and nothing
|
|
// else is removed from the live buffer.
|
|
db.gateBatch = false;
|
|
flushA.complete();
|
|
await finishing;
|
|
await pumpEventQueue(times: 40);
|
|
|
|
final rowsAfterA = await persistedRows();
|
|
final countsAfterA = countsByIndex(rowsAfterA);
|
|
expect(duplicatedIn(countsAfterA), isEmpty);
|
|
expect(missingOf(10, countsAfterA), isEmpty,
|
|
reason: 'A persisted its owned snapshot');
|
|
expect(countsAfterA.length, 10,
|
|
reason: 'A must not have persisted samples it did not own');
|
|
|
|
// Legitimate later flush (the periodic timer) persists 10..29.
|
|
await Future<void>.delayed(tickWait);
|
|
await pumpEventQueue(times: 40);
|
|
|
|
final rows = await persistedRows();
|
|
final counts = countsByIndex(rows);
|
|
final session = container.read(rideStateMachineProvider);
|
|
|
|
debugPrint('==== RID-3A OVERLAP (after fix) ====');
|
|
debugPrint('batch invocations : ${db.batchInvocations}');
|
|
debugPrint('accepted samples : ${session.totalSamplesReceived}');
|
|
debugPrint('durable rows : ${rows.length}');
|
|
debugPrint('distinct identities : ${counts.length}');
|
|
debugPrint('duplicated indices : ${duplicatedIn(counts)}');
|
|
debugPrint('missing indices : ${missingOf(30, counts)}');
|
|
debugPrint('received / persisted : '
|
|
'${session.totalSamplesReceived} / ${session.totalSamplesPersisted}');
|
|
debugPrint('rideIds : ${rows.map((r) => r.rideId).toSet()}');
|
|
|
|
// Previously: 30 rows / 20 distinct, indices 0-9 twice, 10-19 destroyed.
|
|
expect(rows, hasLength(30), reason: '30 accepted -> 30 durable rows');
|
|
expect(counts.length, 30, reason: '30 distinct identities');
|
|
expect(duplicatedIn(counts), isEmpty, reason: '0 duplicates');
|
|
expect(missingOf(30, counts), isEmpty, reason: '0 missing');
|
|
expect(rows.map((r) => r.speed).toList(),
|
|
[for (var i = 0; i < 30; i++) 1000.0 + i],
|
|
reason: 'chronological identity order preserved');
|
|
expect(rows.map((r) => r.rideId).toSet(), {rideId},
|
|
reason: 'every sample against the correct ride');
|
|
expect(session.totalSamplesPersisted, 30,
|
|
reason: 'counters agree with distinct durable samples; nothing left '
|
|
'intentionally buffered after the final flush');
|
|
});
|
|
|
|
test('telemetry arriving during an in-flight flush is persisted exactly '
|
|
'once, after the owned batch', () async {
|
|
final container = buildContainer();
|
|
final rideId = await startRide(container);
|
|
|
|
// Batch A: indices 0..4.
|
|
emitSamples(0, 5);
|
|
await pumpEventQueue(times: 30);
|
|
|
|
db.gateBatch = true;
|
|
await Future<void>.delayed(tickWait);
|
|
expect(db.batchInvocations, 1);
|
|
final flushA = db.batchGates.single;
|
|
|
|
// Batch B: indices 5..9, arriving while A's persistence is parked.
|
|
emitSamples(5, 5);
|
|
await pumpEventQueue(times: 30);
|
|
expect(container.read(rideStateMachineProvider).totalSamplesReceived, 10);
|
|
|
|
db.gateBatch = false;
|
|
flushA.complete();
|
|
await pumpEventQueue(times: 40);
|
|
|
|
// Legitimate later flush persists B.
|
|
await Future<void>.delayed(tickWait);
|
|
await pumpEventQueue(times: 40);
|
|
|
|
final rows = await persistedRows();
|
|
final counts = countsByIndex(rows);
|
|
final session = container.read(rideStateMachineProvider);
|
|
|
|
debugPrint('==== RID-3A DURING-FLUSH ARRIVAL (after fix) ====');
|
|
debugPrint('durable rows : ${rows.length}');
|
|
debugPrint('distinct identities : ${counts.length}');
|
|
debugPrint('duplicated : ${duplicatedIn(counts)}');
|
|
debugPrint('missing : ${missingOf(10, counts)}');
|
|
debugPrint('order : ${rows.map((r) => r.speed).toList()}');
|
|
|
|
expect(rows, hasLength(10));
|
|
expect(duplicatedIn(counts), isEmpty, reason: 'A and B each exactly once');
|
|
expect(missingOf(10, counts), isEmpty, reason: 'nothing missing');
|
|
expect(rows.map((r) => r.speed).toList(),
|
|
[for (var i = 0; i < 10; i++) 1000.0 + i],
|
|
reason: 'A persisted before B');
|
|
expect(rows.map((r) => r.rideId).toSet(), {rideId});
|
|
expect(session.totalSamplesPersisted, 10);
|
|
});
|
|
|
|
test('a failed flush restores its snapshot AHEAD of newer samples and a '
|
|
'later flush persists A B C D E F exactly once each', () async {
|
|
final container = buildContainer();
|
|
final rideId = await startRide(container);
|
|
|
|
// A B C = indices 0, 1, 2.
|
|
emitSamples(0, 3);
|
|
await pumpEventQueue(times: 30);
|
|
|
|
db.gateBatch = true;
|
|
db.batchFailuresRemaining = 1;
|
|
await Future<void>.delayed(tickWait);
|
|
expect(db.batchInvocations, 1);
|
|
final failingFlush = db.batchGates.single;
|
|
|
|
// D E F = indices 3, 4, 5 arrive while the doomed persistence is parked.
|
|
emitSamples(3, 3);
|
|
await pumpEventQueue(times: 30);
|
|
expect(container.read(rideStateMachineProvider).totalSamplesReceived, 6);
|
|
|
|
db.gateBatch = false;
|
|
failingFlush.complete(); // db.batch now throws for this snapshot
|
|
await pumpEventQueue(times: 40);
|
|
|
|
var session = container.read(rideStateMachineProvider);
|
|
expect(session.totalSamplesPersisted, 0,
|
|
reason: 'a failed persistence must not increment the success counter');
|
|
expect(await persistedRows(), isEmpty, reason: 'nothing durable yet');
|
|
expect(session.state, RideState.riding,
|
|
reason: 'telemetry recording continues so a later flush can recover');
|
|
|
|
// Existing periodic timer provides the legitimate retry. No automatic
|
|
// retry worker is introduced.
|
|
await Future<void>.delayed(tickWait);
|
|
await pumpEventQueue(times: 40);
|
|
|
|
final rows = await persistedRows();
|
|
final counts = countsByIndex(rows);
|
|
session = container.read(rideStateMachineProvider);
|
|
|
|
debugPrint('==== RID-3A FAILURE + ORDERED RESTORATION (after fix) ====');
|
|
debugPrint('batch invocations : ${db.batchInvocations}');
|
|
debugPrint('durable rows : ${rows.length}');
|
|
debugPrint('distinct identities : ${counts.length}');
|
|
debugPrint('duplicated : ${duplicatedIn(counts)}');
|
|
debugPrint('missing : ${missingOf(6, counts)}');
|
|
debugPrint('order : ${rows.map((r) => r.speed).toList()}');
|
|
debugPrint('received / persisted : '
|
|
'${session.totalSamplesReceived} / ${session.totalSamplesPersisted}');
|
|
|
|
expect(rows, hasLength(6));
|
|
expect(duplicatedIn(counts), isEmpty, reason: 'retry causes no duplicates');
|
|
expect(missingOf(6, counts), isEmpty, reason: 'no rows missing');
|
|
expect(rows.map((r) => r.speed).toList(),
|
|
[1000.0, 1001.0, 1002.0, 1003.0, 1004.0, 1005.0],
|
|
reason: 'failed snapshot A B C restored AHEAD of newer D E F, both '
|
|
'internal orders preserved');
|
|
expect(rows.map((r) => r.rideId).toSet(), {rideId});
|
|
expect(session.totalSamplesPersisted, 6,
|
|
reason: 'counter reflects only the successfully persisted snapshot');
|
|
});
|
|
});
|
|
}
|