fix: make telemetry batch flush ownership safe
This commit is contained in:
Developer
parent
6161233b2d
commit
5402412ea0
|
|
@ -138,6 +138,13 @@ class RideStateMachine extends StateNotifier<RideSession> {
|
|||
// operation (try/finally) and it is intentionally NOT cleared anywhere else.
|
||||
bool _isFinishing = false;
|
||||
|
||||
// RID-3A: re-entrancy guard for telemetry batch persistence. Raised
|
||||
// synchronously before the first await inside _flushBatch(); its lifetime is
|
||||
// owned solely by the in-flight persistence operation (try/finally) and it is
|
||||
// intentionally NOT cleared anywhere else, because clearing it while a
|
||||
// persistence await is outstanding would re-open the shared-buffer race.
|
||||
bool _isFlushingTelemetryBatch = false;
|
||||
|
||||
RideStateMachine(this.db, this.telemetryService, this.recordsService, this.audioAlertService, this.ref) : super(RideSession()) {
|
||||
_telemetrySub = telemetryService.telemetryStream.listen(_onTelemetryUpdate);
|
||||
|
||||
|
|
@ -528,16 +535,40 @@ class RideStateMachine extends StateNotifier<RideSession> {
|
|||
}
|
||||
|
||||
Future<void> _flushBatch() async {
|
||||
if (_telemetryBatch.isNotEmpty) {
|
||||
// RID-3A: telemetry batch persistence with exclusive snapshot ownership.
|
||||
//
|
||||
// The guard is raised synchronously, before the first `await`, so a
|
||||
// competing legitimate flush (the 3 s periodic timer against the flush
|
||||
// awaited by finishRide) returns without taking ownership instead of
|
||||
// snapshotting the very same samples.
|
||||
//
|
||||
// The snapshot is removed from the live buffer BEFORE the persistence await,
|
||||
// so the in-flight operation owns those samples exclusively and telemetry
|
||||
// arriving during the await accumulates in a fresh live buffer. Nothing is
|
||||
// removed from the live buffer after the await - that post-await
|
||||
// `removeRange` is what previously allowed an overlapping flush to discard
|
||||
// newer samples it had never written, while re-inserting samples another
|
||||
// flush had already persisted.
|
||||
//
|
||||
// The incremental ride-stats persistence below is deliberately left outside
|
||||
// this guard so its existing every-tick cadence is unchanged.
|
||||
if (!_isFlushingTelemetryBatch && _telemetryBatch.isNotEmpty) {
|
||||
_isFlushingTelemetryBatch = true;
|
||||
final batchToInsert = List<TelemetrySamplesCompanion>.from(_telemetryBatch);
|
||||
_telemetryBatch.clear();
|
||||
try {
|
||||
await db.batch((batch) {
|
||||
batch.insertAll(db.telemetrySamples, batchToInsert);
|
||||
});
|
||||
// Success: clear inserted samples and update persisted count
|
||||
_telemetryBatch.removeRange(0, batchToInsert.length);
|
||||
// Success: account only for the owned snapshot that is now durable.
|
||||
state = state.copyWith(totalSamplesPersisted: state.totalSamplesPersisted + batchToInsert.length);
|
||||
} catch (e) {
|
||||
// Persistence failed, so the owned snapshot is not durable. Restore it
|
||||
// to the FRONT of the live buffer, preserving both the snapshot's
|
||||
// internal order and the order of samples that arrived during the await,
|
||||
// so chronological identity order survives and a later legitimate flush
|
||||
// can retry them. No automatic retry worker is introduced.
|
||||
_telemetryBatch.insertAll(0, batchToInsert);
|
||||
// Keep in queue for retry on next flush interval
|
||||
try {
|
||||
ref.read(fieldTestLoggerProvider).log(
|
||||
|
|
@ -546,6 +577,8 @@ class RideStateMachine extends StateNotifier<RideSession> {
|
|||
severity: EventSeverity.warning,
|
||||
);
|
||||
} catch (_) {}
|
||||
} finally {
|
||||
_isFlushingTelemetryBatch = false;
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -0,0 +1,492 @@
|
|||
// 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');
|
||||
});
|
||||
});
|
||||
}
|
||||
Loading…
Reference in New Issue