927 lines
33 KiB
Dart
927 lines
33 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/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';
|
|
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/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
|
|
/// 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;
|
|
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
|
|
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.timestamp,
|
|
mode: drift.OrderingMode.asc,
|
|
),
|
|
(t) => drift.OrderingTerm(
|
|
expression: t.id,
|
|
mode: drift.OrderingMode.asc,
|
|
),
|
|
]))
|
|
.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() now awaits the active owner
|
|
// (RID-3B-1) instead of observing a no-op. While flush A is still parked it
|
|
// must take no ownership of A's snapshot and issue no batch of its own; it
|
|
// 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',
|
|
);
|
|
|
|
// 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
|
|
// observation point at which flush A's committed write is isolated: A has
|
|
// persisted and nothing else has been written yet.
|
|
flushA.complete();
|
|
await pumpEventQueue(times: 40);
|
|
|
|
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',
|
|
);
|
|
// 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
|
|
// durability barrier, not after `await finishing`, because RID-3B-1 now
|
|
// 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',
|
|
);
|
|
|
|
// Release the barrier's drain and let finishRide complete. It persists
|
|
// 10..29 itself, so no later periodic flush is required.
|
|
db.gateBatch = false;
|
|
db.batchGates.last.complete();
|
|
await finishing;
|
|
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',
|
|
);
|
|
});
|
|
|
|
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);
|
|
},
|
|
);
|
|
});
|
|
}
|