fix: make telemetry flush completion dispose-safe
This commit is contained in:
Developer
parent
94d8191b43
commit
d31f88e97d
|
|
@ -1,6 +1,7 @@
|
|||
import 'dart:async';
|
||||
import 'package:wakelock_plus/wakelock_plus.dart';
|
||||
import 'package:flutter_riverpod/flutter_riverpod.dart';
|
||||
import 'package:flutter/foundation.dart' show debugPrint;
|
||||
import '../../database/database.dart' hide TelemetrySample;
|
||||
import 'package:drift/drift.dart' as drift;
|
||||
import 'package:geolocator/geolocator.dart';
|
||||
|
|
@ -172,7 +173,7 @@ class RideStateMachine extends StateNotifier<RideSession> {
|
|||
_telemetrySub = telemetryService.telemetryStream.listen(_onTelemetryUpdate);
|
||||
|
||||
// Write to DB every 3 seconds for tight crash-durability
|
||||
_batchWriteTimer = Timer.periodic(const Duration(seconds: 3), (_) => _flushBatch());
|
||||
_batchWriteTimer = Timer.periodic(const Duration(seconds: 3), (_) => _runScheduledFlush());
|
||||
|
||||
// Watchdog timer (every 2 seconds) checks if telemetry pipeline is alive
|
||||
_watchdogTimer = Timer.periodic(const Duration(seconds: 2), (_) => _checkWatchdog());
|
||||
|
|
@ -684,27 +685,61 @@ class RideStateMachine extends StateNotifier<RideSession> {
|
|||
final completion = Completer<void>();
|
||||
_activeTelemetryFlush = completion.future;
|
||||
try {
|
||||
await db.batch((batch) {
|
||||
batch.insertAll(db.telemetrySamples, batchToInsert);
|
||||
});
|
||||
// Success: account only for the owned snapshot that is now durable.
|
||||
state = state.copyWith(totalSamplesPersisted: state.totalSamplesPersisted + batchToInsert.length);
|
||||
return true;
|
||||
} 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 a later
|
||||
// legitimate flush can retry them. No automatic retry worker is added.
|
||||
_telemetryBatch.insertAll(0, batchToInsert);
|
||||
// Keep in queue for retry on next flush interval
|
||||
// ---- PRE-COMMIT PHASE ---------------------------------------------------
|
||||
// This is the ONLY block that may fail persistence. RID-3A ownership,
|
||||
// ordering and front-restoration semantics are unchanged.
|
||||
try {
|
||||
ref.read(fieldTestLoggerProvider).log(
|
||||
eventType: FieldTestEventType.databaseQueueWarning,
|
||||
message: 'Database batch write retry: $e',
|
||||
severity: EventSeverity.warning,
|
||||
);
|
||||
} catch (_) {}
|
||||
return false;
|
||||
await db.batch((batch) {
|
||||
batch.insertAll(db.telemetrySamples, batchToInsert);
|
||||
});
|
||||
} catch (e) {
|
||||
// Persistence failed BEFORE the commit point, 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 a later legitimate flush can retry them.
|
||||
// No automatic retry worker is added.
|
||||
_telemetryBatch.insertAll(0, batchToInsert);
|
||||
// Keep in queue for retry on next flush interval. A notifier that has
|
||||
// begun disposal must not touch ref, so the diagnostic is skipped in that
|
||||
// case; it remains bounded either way.
|
||||
if (mounted) {
|
||||
try {
|
||||
ref.read(fieldTestLoggerProvider).log(
|
||||
eventType: FieldTestEventType.databaseQueueWarning,
|
||||
message: 'Database batch write retry: $e',
|
||||
severity: EventSeverity.warning,
|
||||
);
|
||||
} catch (_) {}
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
// ---- COMMIT POINT REACHED -----------------------------------------------
|
||||
// The successful return of db.batch(...) above is the telemetry durability
|
||||
// commit point. The owned snapshot is now durable: nothing below may restore
|
||||
// it to the live buffer or change the durability verdict.
|
||||
|
||||
// Post-commit bookkeeping is notifier-owned, so it is conditional on the
|
||||
// notifier still being alive and isolated in its own guard. `mounted` is
|
||||
// StateNotifier's real property, so this boundary is identical in debug and
|
||||
// release - correctness never depends on a debug assertion firing. After
|
||||
// disposal there is no state read, no state assignment and no ref access.
|
||||
//
|
||||
// Accepted RID-3B-2 counter contract: when disposal races a committed flush,
|
||||
// the durable row count may exceed the dead notifier's final in-memory
|
||||
// totalSamplesPersisted value. That is a terminal counter snapshot on a
|
||||
// disposed notifier, NOT a persistence failure, and no live consumer may
|
||||
// rely on it. Committed samples are never restored to make the counter look
|
||||
// consistent, and no database counter or schema field is added.
|
||||
if (mounted) {
|
||||
try {
|
||||
state = state.copyWith(totalSamplesPersisted: state.totalSamplesPersisted + batchToInsert.length);
|
||||
} catch (_) {
|
||||
// Post-commit bookkeeping failure must never be classified as a
|
||||
// persistence failure and must never restore the committed snapshot.
|
||||
}
|
||||
}
|
||||
return true;
|
||||
} finally {
|
||||
_isFlushingTelemetryBatch = false;
|
||||
if (!completion.isCompleted) completion.complete();
|
||||
|
|
@ -712,6 +747,28 @@ class RideStateMachine extends StateNotifier<RideSession> {
|
|||
}
|
||||
}
|
||||
|
||||
/// RID-3B-2 timer Future ownership.
|
||||
///
|
||||
/// A Timer.periodic callback is synchronous, so the Future returned by
|
||||
/// _flushBatch() would otherwise be discarded and any rejection would surface
|
||||
/// as an uncaught zone error attributed to the timer callback. The Future is
|
||||
/// explicitly owned here instead.
|
||||
///
|
||||
/// This is a defensive ownership boundary only: no second scheduler, no new
|
||||
/// timer, and no change to the 3-second crash-durability cadence. The timer
|
||||
/// callback itself stays synchronous, so cancellation semantics are unchanged.
|
||||
///
|
||||
/// Diagnostics are bounded to runtimeType only. The exception object is never
|
||||
/// logged because its message may carry implementation or database detail, and
|
||||
/// no stack trace or telemetry payload is recorded.
|
||||
void _runScheduledFlush() {
|
||||
unawaited(
|
||||
_flushBatch().catchError((Object error, StackTrace stackTrace) {
|
||||
debugPrint('Scheduled telemetry flush failed: ${error.runtimeType}');
|
||||
}),
|
||||
);
|
||||
}
|
||||
|
||||
Future<void> _flushBatch() async {
|
||||
// RID-3A telemetry batch ownership (exclusive snapshot, clear-before-await,
|
||||
// front-restoration on failure, success-only counter) now lives in
|
||||
|
|
@ -738,6 +795,14 @@ class RideStateMachine extends StateNotifier<RideSession> {
|
|||
// this guard so its existing every-tick cadence is unchanged.
|
||||
await _persistOwnedTelemetrySnapshot();
|
||||
|
||||
// RID-3B-2: the notifier may have been disposed while this telemetry flush
|
||||
// was in flight, because Timer.cancel() cannot abort an already-started async
|
||||
// callback. The database commit above is already authoritative and is not
|
||||
// revisited; everything below is notifier-owned, so return without reading
|
||||
// state or ref. Incremental stats are never fabricated after disposal, and
|
||||
// stats persistence is never moved ahead of the telemetry commit.
|
||||
if (!mounted) return;
|
||||
|
||||
// Incrementally persist ongoing ride summary stats to SQLite so crashes/terminations never lose stats
|
||||
if (state.dbId != null && state.state == RideState.riding) {
|
||||
try {
|
||||
|
|
|
|||
|
|
@ -0,0 +1,659 @@
|
|||
// RID-3B-2 permanent regression suite: timer Future ownership, post-dispose-safe
|
||||
// flush completion, and the telemetry persistence commit-point contract.
|
||||
//
|
||||
// Concerns covered:
|
||||
// A. telemetry persistence commit is separated from notifier bookkeeping
|
||||
// B. an in-flight flush is safe when the notifier becomes disposed
|
||||
// C. the Timer.periodic flush Future is explicitly owned and contained
|
||||
//
|
||||
// Concern D (graceful buffered disposal) is DELIBERATELY DEFERRED and is locked
|
||||
// as a documented residual by RID3B2-T7. That test is not an endorsement of data
|
||||
// loss; it exists so this patch cannot silently pretend concern D was solved.
|
||||
import 'dart:async';
|
||||
|
||||
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_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/features/records/records_service.dart';
|
||||
|
||||
typedef TelemetryRow = dbgen.TelemetrySample;
|
||||
|
||||
/// Counts every `batch` call and can park or fail it deterministically, so a
|
||||
/// flush can be held exactly at the durability commit point without any
|
||||
/// sleep-only race.
|
||||
class _Rid3b2AppDatabase extends AppDatabase {
|
||||
_Rid3b2AppDatabase() : 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++;
|
||||
if (gateBatch) {
|
||||
final gate = Completer<void>();
|
||||
batchGates.add(gate);
|
||||
await gate.future;
|
||||
}
|
||||
if (batchFailuresRemaining > 0) {
|
||||
batchFailuresRemaining--;
|
||||
throw StateError('RID-3B-2 simulated pre-commit persistence failure');
|
||||
}
|
||||
return super.batch(runInBatch);
|
||||
}
|
||||
}
|
||||
|
||||
/// Captures log messages so the classification text emitted by the PRE-COMMIT
|
||||
/// catch block can be inspected directly. That message is emitted only on a
|
||||
/// genuine persistence failure, so its absence proves a committed snapshot was
|
||||
/// never misclassified. Its own periodic flush() -> _db.batch is neutralised so
|
||||
/// batch accounting refers only to telemetry persistence.
|
||||
class _Rid3b2FieldTestLogger extends FieldTestLogger {
|
||||
_Rid3b2FieldTestLogger(super.db);
|
||||
|
||||
final List<String> loggedMessages = [];
|
||||
|
||||
int get retryWarnings => loggedMessages
|
||||
.where((m) => m.contains('Database batch write retry'))
|
||||
.length;
|
||||
|
||||
@override
|
||||
Future<void> flush() async {}
|
||||
|
||||
@override
|
||||
void log({
|
||||
required FieldTestEventType eventType,
|
||||
required EventSeverity severity,
|
||||
required String message,
|
||||
int? rideId,
|
||||
DateTime? timestamp,
|
||||
DateTime? telemetryTimestamp,
|
||||
int? gpsAgeMs,
|
||||
double? speed,
|
||||
double? latitude,
|
||||
double? longitude,
|
||||
double? gpsAccuracy,
|
||||
double? leanAngle,
|
||||
double? longitudinalG,
|
||||
double? lateralG,
|
||||
double? overallConfidence,
|
||||
Map<String, dynamic>? metadata,
|
||||
}) {
|
||||
loggedMessages.add(message);
|
||||
super.log(
|
||||
eventType: eventType,
|
||||
severity: severity,
|
||||
message: message,
|
||||
rideId: rideId,
|
||||
timestamp: timestamp,
|
||||
telemetryTimestamp: telemetryTimestamp,
|
||||
gpsAgeMs: gpsAgeMs,
|
||||
speed: speed,
|
||||
latitude: latitude,
|
||||
longitude: longitude,
|
||||
gpsAccuracy: gpsAccuracy,
|
||||
leanAngle: leanAngle,
|
||||
longitudinalG: longitudinalG,
|
||||
lateralG: lateralG,
|
||||
overallConfidence: overallConfidence,
|
||||
metadata: metadata,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
class _Rid3b2AudioAlertService extends Fake implements AudioAlertService {
|
||||
@override
|
||||
Future<void> speak(String text, {bool force = false}) async {}
|
||||
}
|
||||
|
||||
class _Rid3b2TelemetryService extends TelemetryService {
|
||||
_Rid3b2TelemetryService(super.loc, super.fusion, super.compass, super.logger,
|
||||
super.validator);
|
||||
|
||||
final StreamController<TelemetrySample> _controller =
|
||||
StreamController<TelemetrySample>.broadcast();
|
||||
|
||||
@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 batch-write timer is a real 3-second Timer.periodic. Waiting
|
||||
/// slightly longer than one interval makes the scheduled flush deterministic
|
||||
/// without mocking the cadence, which is a frozen crash-durability contract.
|
||||
const tickWait = Duration(milliseconds: 3300);
|
||||
|
||||
late _Rid3b2AppDatabase db;
|
||||
late _Rid3b2FieldTestLogger logger;
|
||||
late _Rid3b2TelemetryService telemetry;
|
||||
late TelemetryValidator validator;
|
||||
late SharedPreferences prefs;
|
||||
|
||||
final sampleBase = DateTime(2026, 9, 10, 12);
|
||||
|
||||
setUp(() async {
|
||||
SharedPreferences.setMockInitialValues({});
|
||||
prefs = await SharedPreferences.getInstance();
|
||||
db = _Rid3b2AppDatabase();
|
||||
logger = _Rid3b2FieldTestLogger(db);
|
||||
validator = TelemetryValidator();
|
||||
telemetry = _Rid3b2TelemetryService(_NullLocationService(),
|
||||
_NullSensorFusionService(), _NullCompassService(), logger, validator);
|
||||
});
|
||||
|
||||
tearDown(() async {
|
||||
for (final gate in db.batchGates) {
|
||||
if (!gate.isCompleted) gate.complete();
|
||||
}
|
||||
db.batchGates.clear();
|
||||
db.gateBatch = false;
|
||||
db.batchFailuresRemaining = 0;
|
||||
telemetry.dispose();
|
||||
logger.dispose();
|
||||
await db.close();
|
||||
});
|
||||
|
||||
ProviderContainer buildContainer({bool registerTeardown = true}) {
|
||||
final container = ProviderContainer(overrides: [
|
||||
sharedPreferencesProvider.overrideWithValue(prefs),
|
||||
databaseProvider.overrideWithValue(db),
|
||||
telemetryServiceProvider.overrideWithValue(telemetry),
|
||||
audioAlertServiceProvider
|
||||
.overrideWithValue(_Rid3b2AudioAlertService()),
|
||||
fieldTestLoggerProvider.overrideWithValue(logger),
|
||||
]);
|
||||
if (registerTeardown) addTearDown(container.dispose);
|
||||
return container;
|
||||
}
|
||||
|
||||
Future<int> startRide(ProviderContainer container) async {
|
||||
await container.read(rideStateMachineProvider.notifier).startRide();
|
||||
return container.read(rideStateMachineProvider).dbId!;
|
||||
}
|
||||
|
||||
/// Emits uniquely identifiable samples. The identity is encoded in `speed` so
|
||||
/// durable rows can be accounted exactly once, and so chronological ordering is
|
||||
/// observable from the autoincrement row order.
|
||||
void emitSamples(int startIndex, int count) {
|
||||
for (var i = 0; i < count; i++) {
|
||||
final index = startIndex + i;
|
||||
telemetry.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;
|
||||
final index = (speed - 1000.0).round();
|
||||
counts[index] = (counts[index] ?? 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> identitiesOf(List<TelemetryRow> rows) =>
|
||||
rows.map((r) => (r.speed! - 1000.0).round()).toList();
|
||||
|
||||
/// Runs [body] inside a guarded zone and returns every error that escaped as an
|
||||
/// uncaught asynchronous error. RID-3B-2 requires the scheduled flush Future to
|
||||
/// be explicitly owned, so a regression surfaces here as a non-empty list
|
||||
/// rather than as an opaque test failure. The container must be created inside
|
||||
/// the zone for its timer-owned Futures to report there.
|
||||
Future<List<Object>> captureUncaught(Future<void> Function() body) async {
|
||||
final uncaught = <Object>[];
|
||||
await runZonedGuarded(body, (error, stackTrace) => uncaught.add(error));
|
||||
return uncaught;
|
||||
}
|
||||
|
||||
group('RID-3B-2 flush/dispose safety', () {
|
||||
test('RID3B2-T1 scheduled flush Future is owned and contained', () async {
|
||||
var rowsAfterFailure = -1;
|
||||
var batchAfterFailure = -1;
|
||||
var retryAfterFailure = -1;
|
||||
var rowsAfterRetry = -1;
|
||||
var counterAfterRetry = -1;
|
||||
|
||||
// A genuinely reachable rejection path: the scheduled flush encounters a
|
||||
// pre-commit persistence failure. The ownership wrapper must keep the
|
||||
// Future owned so nothing escapes to the zone, and the timer must keep
|
||||
// working on the following interval. No production seam is added solely to
|
||||
// manufacture an impossible error.
|
||||
final uncaught = await captureUncaught(() async {
|
||||
final container = buildContainer(registerTeardown: false);
|
||||
await startRide(container);
|
||||
emitSamples(0, 6);
|
||||
await pumpEventQueue(times: 30);
|
||||
|
||||
db.batchFailuresRemaining = 1;
|
||||
await Future<void>.delayed(tickWait);
|
||||
await pumpEventQueue(times: 40);
|
||||
|
||||
rowsAfterFailure = (await persistedRows()).length;
|
||||
batchAfterFailure = db.batchInvocations;
|
||||
retryAfterFailure = logger.retryWarnings;
|
||||
|
||||
// The restored snapshot must be retried by the next scheduled tick,
|
||||
// proving the ownership wrapper did not break the timer.
|
||||
await Future<void>.delayed(tickWait);
|
||||
await pumpEventQueue(times: 40);
|
||||
|
||||
rowsAfterRetry = (await persistedRows()).length;
|
||||
counterAfterRetry =
|
||||
container.read(rideStateMachineProvider).totalSamplesPersisted;
|
||||
container.dispose();
|
||||
});
|
||||
|
||||
expect(uncaught, isEmpty,
|
||||
reason: 'the scheduled flush Future must be explicitly owned so no '
|
||||
'rejection reaches the zone: $uncaught');
|
||||
expect(rowsAfterFailure, 0, reason: 'a failed attempt commits nothing');
|
||||
expect(batchAfterFailure, 1);
|
||||
expect(retryAfterFailure, 1,
|
||||
reason: 'the bounded pre-commit diagnostic is still emitted');
|
||||
expect(rowsAfterRetry, 6, reason: 'the timer retries and succeeds');
|
||||
expect(counterAfterRetry, 6,
|
||||
reason: 'the counter accounts only for durable samples');
|
||||
});
|
||||
test('RID3B2-T2 dispose during parked db.batch commits without an '
|
||||
'uncaught error', () async {
|
||||
var rowsWhileParked = -1;
|
||||
var counterWhileParked = -1;
|
||||
var rowsAfterRelease = -1;
|
||||
var distinct = -1;
|
||||
var duplicates = <int>[];
|
||||
var batchCalls = -1;
|
||||
var retryWarnings = -1;
|
||||
var mountedAfterDispose = true;
|
||||
|
||||
final uncaught = await captureUncaught(() async {
|
||||
final container = buildContainer(registerTeardown: false);
|
||||
final notifier = container.read(rideStateMachineProvider.notifier);
|
||||
await startRide(container);
|
||||
emitSamples(0, 10);
|
||||
await pumpEventQueue(times: 30);
|
||||
|
||||
// Park the flush exactly at the durability commit point.
|
||||
db.gateBatch = true;
|
||||
await Future<void>.delayed(tickWait);
|
||||
final parked = db.batchGates.single;
|
||||
rowsWhileParked = (await persistedRows()).length;
|
||||
counterWhileParked =
|
||||
container.read(rideStateMachineProvider).totalSamplesPersisted;
|
||||
|
||||
// Timer.cancel() cannot abort an already-started async callback, so this
|
||||
// continuation legitimately outlives the notifier.
|
||||
container.dispose();
|
||||
mountedAfterDispose = notifier.mounted;
|
||||
|
||||
db.gateBatch = false;
|
||||
parked.complete();
|
||||
await pumpEventQueue(times: 60);
|
||||
|
||||
final rows = await persistedRows();
|
||||
final counts = countsByIndex(rows);
|
||||
rowsAfterRelease = rows.length;
|
||||
distinct = counts.length;
|
||||
duplicates = duplicatedIn(counts);
|
||||
batchCalls = db.batchInvocations;
|
||||
retryWarnings = logger.retryWarnings;
|
||||
});
|
||||
|
||||
expect(rowsWhileParked, 0, reason: 'nothing durable while parked');
|
||||
expect(counterWhileParked, 0, reason: 'counter not yet advanced');
|
||||
expect(mountedAfterDispose, isFalse,
|
||||
reason: 'StateNotifier.mounted is the real disposal signal, so the '
|
||||
'boundary cannot depend on a debug assertion');
|
||||
expect(uncaught, isEmpty,
|
||||
reason: 'no post-dispose state or ref access may raise: $uncaught');
|
||||
expect(rowsAfterRelease, 10, reason: 'the committed snapshot is durable');
|
||||
expect(distinct, 10);
|
||||
expect(duplicates, isEmpty, reason: 'persisted exactly once');
|
||||
expect(batchCalls, 1, reason: 'no retry ownership taken after disposal');
|
||||
expect(retryWarnings, 0,
|
||||
reason: 'a committed write must never be reported as a persistence '
|
||||
'failure, and no ref access may occur after disposal');
|
||||
});
|
||||
test('RID3B2-T3 a committed snapshot is never restored as a persistence '
|
||||
'failure', () async {
|
||||
var rowsPhase1 = -1;
|
||||
var counterPhase1 = -1;
|
||||
var rowsFinal = -1;
|
||||
var duplicates = <int>[];
|
||||
var identities = <int>[];
|
||||
var batchCalls = -1;
|
||||
var retryWarnings = -1;
|
||||
|
||||
final uncaught = await captureUncaught(() async {
|
||||
final container = buildContainer(registerTeardown: false);
|
||||
await startRide(container);
|
||||
|
||||
// Phase 1: a normal flush commits while the notifier is alive.
|
||||
emitSamples(0, 10);
|
||||
await pumpEventQueue(times: 30);
|
||||
await Future<void>.delayed(tickWait);
|
||||
await pumpEventQueue(times: 40);
|
||||
rowsPhase1 = (await persistedRows()).length;
|
||||
counterPhase1 =
|
||||
container.read(rideStateMachineProvider).totalSamplesPersisted;
|
||||
|
||||
// Phase 2: disposal races a second committed flush.
|
||||
emitSamples(10, 5);
|
||||
await pumpEventQueue(times: 30);
|
||||
db.gateBatch = true;
|
||||
await Future<void>.delayed(tickWait);
|
||||
final parked = db.batchGates.single;
|
||||
container.dispose();
|
||||
db.gateBatch = false;
|
||||
parked.complete();
|
||||
await pumpEventQueue(times: 60);
|
||||
|
||||
final rows = await persistedRows();
|
||||
final counts = countsByIndex(rows);
|
||||
rowsFinal = rows.length;
|
||||
duplicates = duplicatedIn(counts);
|
||||
identities = counts.keys.toList()..sort();
|
||||
batchCalls = db.batchInvocations;
|
||||
retryWarnings = logger.retryWarnings;
|
||||
});
|
||||
|
||||
expect(rowsPhase1, 10);
|
||||
expect(counterPhase1, 10,
|
||||
reason: 'while mounted the counter increments normally');
|
||||
expect(uncaught, isEmpty, reason: '$uncaught');
|
||||
expect(rowsFinal, 15, reason: 'both committed snapshots are durable');
|
||||
expect(identities, List<int>.generate(15, (i) => i),
|
||||
reason: 'every identity present, none missing');
|
||||
expect(duplicates, isEmpty,
|
||||
reason: 'no identity persisted twice: the phase-2 snapshot was not '
|
||||
'restored to the live buffer and never retried');
|
||||
// Direct private-buffer inspection is unavailable, so non-restoration is
|
||||
// proved through observable batch invocation and durable identity
|
||||
// accounting: exactly one write per owned snapshot, and zero emissions of
|
||||
// the diagnostic that only the pre-commit restoration path produces.
|
||||
expect(batchCalls, 2,
|
||||
reason: 'absence of a third invocation proves no retry ownership');
|
||||
expect(retryWarnings, 0,
|
||||
reason: 'the pre-commit catch is the only emitter of this diagnostic');
|
||||
});
|
||||
test('RID3B2-T4 dead-notifier counter divergence is terminal, not a '
|
||||
'persistence failure', () async {
|
||||
var counterBeforeDispose = -1;
|
||||
var durableRows = -1;
|
||||
var batchCalls = -1;
|
||||
var retryWarnings = -1;
|
||||
var mountedAfterDispose = true;
|
||||
|
||||
final uncaught = await captureUncaught(() async {
|
||||
final container = buildContainer(registerTeardown: false);
|
||||
final notifier = container.read(rideStateMachineProvider.notifier);
|
||||
await startRide(container);
|
||||
emitSamples(0, 10);
|
||||
await pumpEventQueue(times: 30);
|
||||
|
||||
db.gateBatch = true;
|
||||
await Future<void>.delayed(tickWait);
|
||||
final parked = db.batchGates.single;
|
||||
|
||||
// The last in-memory counter value this notifier will ever hold.
|
||||
counterBeforeDispose =
|
||||
container.read(rideStateMachineProvider).totalSamplesPersisted;
|
||||
|
||||
container.dispose();
|
||||
mountedAfterDispose = notifier.mounted;
|
||||
|
||||
db.gateBatch = false;
|
||||
parked.complete();
|
||||
await pumpEventQueue(times: 60);
|
||||
|
||||
durableRows = (await persistedRows()).length;
|
||||
batchCalls = db.batchInvocations;
|
||||
retryWarnings = logger.retryWarnings;
|
||||
});
|
||||
|
||||
expect(uncaught, isEmpty,
|
||||
reason: 'no attempt may be made to update dead notifier state: '
|
||||
'$uncaught');
|
||||
expect(mountedAfterDispose, isFalse);
|
||||
expect(durableRows, 10,
|
||||
reason: 'the database commit is authoritative and complete');
|
||||
expect(counterBeforeDispose, 0,
|
||||
reason: 'the terminal in-memory counter never advanced past its '
|
||||
'pre-commit value');
|
||||
// LOCKED RID-3B-2 CONTRACT (terminal dead-notifier counter divergence):
|
||||
// durable DB rows may exceed the disposed notifier's final in-memory
|
||||
// totalSamplesPersisted value. This is NOT a persistence failure, it does
|
||||
// not alter the durability verdict, no live consumer may rely on that
|
||||
// disposed counter, and committed samples are never restored to make it
|
||||
// look consistent. No database counter or schema field is added.
|
||||
expect(durableRows, greaterThan(counterBeforeDispose),
|
||||
reason: 'the accepted divergence is explicitly locked here');
|
||||
expect(retryWarnings, 0,
|
||||
reason: 'the durability verdict remains success, so the pre-commit '
|
||||
'persistence-failure path must not be entered');
|
||||
expect(batchCalls, 1, reason: 'the committed snapshot is never retried');
|
||||
});
|
||||
|
||||
test('RID3B2-T5 a pre-commit failure still restores the snapshot and '
|
||||
'retries exactly once', () async {
|
||||
final container = buildContainer();
|
||||
await startRide(container);
|
||||
emitSamples(0, 5);
|
||||
await pumpEventQueue(times: 30);
|
||||
|
||||
db.gateBatch = true;
|
||||
await Future<void>.delayed(tickWait);
|
||||
final parked = db.batchGates.single;
|
||||
|
||||
// Samples arriving DURING the await must end up behind the restored
|
||||
// snapshot, which is what preserves RID-3A chronological ordering.
|
||||
emitSamples(5, 3);
|
||||
await pumpEventQueue(times: 20);
|
||||
|
||||
db.batchFailuresRemaining = 1;
|
||||
db.gateBatch = false;
|
||||
parked.complete();
|
||||
await pumpEventQueue(times: 40);
|
||||
|
||||
final rowsAfterFailure = await persistedRows();
|
||||
expect(rowsAfterFailure, isEmpty,
|
||||
reason: 'a pre-commit failure commits nothing');
|
||||
expect(logger.retryWarnings, 1,
|
||||
reason: 'the bounded pre-commit diagnostic is emitted while mounted');
|
||||
expect(container.read(rideStateMachineProvider).totalSamplesPersisted, 0,
|
||||
reason: 'success-only counter accounting is preserved');
|
||||
|
||||
// A later legitimate flush retries the restored snapshot.
|
||||
await Future<void>.delayed(tickWait);
|
||||
await pumpEventQueue(times: 40);
|
||||
|
||||
final rows = await persistedRows();
|
||||
final counts = countsByIndex(rows);
|
||||
expect(rows, hasLength(8), reason: 'every identity is eventually durable');
|
||||
expect(duplicatedIn(counts), isEmpty, reason: 'persisted exactly once');
|
||||
expect(counts.keys.toList()..sort(), List<int>.generate(8, (i) => i),
|
||||
reason: 'no identity missing');
|
||||
expect(identitiesOf(rows), List<int>.generate(8, (i) => i),
|
||||
reason: 'front-restoration preserved chronological ordering');
|
||||
expect(container.read(rideStateMachineProvider).totalSamplesPersisted, 8);
|
||||
});
|
||||
test('RID3B2-T6 post-commit bookkeeping trouble is not classified as a '
|
||||
'persistence failure', () async {
|
||||
var rowsAfterFirst = -1;
|
||||
var batchAfterFirst = -1;
|
||||
var retryAfterFirst = -1;
|
||||
var rowsFinal = -1;
|
||||
var duplicates = <int>[];
|
||||
var batchFinal = -1;
|
||||
|
||||
// StateNotifier.set state collects listener errors and rethrows
|
||||
// StateNotifierListenerError, so a listener that throws on the post-commit
|
||||
// counter update exercises "bookkeeping failed AFTER the commit point"
|
||||
// while the notifier stays alive. Riverpod guards listener errors, so this
|
||||
// locks the structural guarantee rather than relying on a manufactured
|
||||
// production seam.
|
||||
final uncaught = await captureUncaught(() async {
|
||||
final container = buildContainer(registerTeardown: false);
|
||||
await startRide(container);
|
||||
emitSamples(0, 5);
|
||||
await pumpEventQueue(times: 30);
|
||||
|
||||
container.listen<RideSession>(rideStateMachineProvider, (prev, next) {
|
||||
if (next.totalSamplesPersisted > (prev?.totalSamplesPersisted ?? 0)) {
|
||||
throw StateError(
|
||||
'RID-3B-2 simulated post-commit bookkeeping failure');
|
||||
}
|
||||
}, fireImmediately: false);
|
||||
|
||||
db.gateBatch = true;
|
||||
await Future<void>.delayed(tickWait);
|
||||
final parked = db.batchGates.single;
|
||||
db.gateBatch = false;
|
||||
parked.complete();
|
||||
await pumpEventQueue(times: 60);
|
||||
|
||||
rowsAfterFirst = (await persistedRows()).length;
|
||||
batchAfterFirst = db.batchInvocations;
|
||||
retryAfterFirst = logger.retryWarnings;
|
||||
|
||||
// A further scheduled tick must not re-persist the committed snapshot.
|
||||
await Future<void>.delayed(tickWait);
|
||||
await pumpEventQueue(times: 60);
|
||||
|
||||
final rows = await persistedRows();
|
||||
rowsFinal = rows.length;
|
||||
duplicates = duplicatedIn(countsByIndex(rows));
|
||||
batchFinal = db.batchInvocations;
|
||||
container.dispose();
|
||||
});
|
||||
|
||||
expect(uncaught.length, lessThanOrEqualTo(1),
|
||||
reason: 'only the deliberately injected listener error may surface, '
|
||||
'never a post-commit misclassification: $uncaught');
|
||||
expect(rowsAfterFirst, 5, reason: 'the commit point was reached');
|
||||
expect(retryAfterFirst, 0,
|
||||
reason: 'the persistence catch is no longer entered because of '
|
||||
'notifier bookkeeping');
|
||||
expect(batchAfterFirst, 1);
|
||||
expect(rowsFinal, 5, reason: 'no duplicate persistence on a later tick');
|
||||
expect(duplicates, isEmpty,
|
||||
reason: 'the snapshot-restoration path was not invoked');
|
||||
expect(batchFinal, 1,
|
||||
reason: 'the committed snapshot was never returned to the live buffer, '
|
||||
'so the next tick had nothing to own');
|
||||
});
|
||||
test('RID3B2-T7 buffered telemetry at graceful dispose is a documented '
|
||||
'residual (concern D deferred)', () async {
|
||||
final container = buildContainer(registerTeardown: false);
|
||||
final notifier = container.read(rideStateMachineProvider.notifier);
|
||||
final rideId = await startRide(container);
|
||||
|
||||
emitSamples(0, 10);
|
||||
await pumpEventQueue(times: 30);
|
||||
final before = container.read(rideStateMachineProvider);
|
||||
expect(before.totalSamplesReceived, 10);
|
||||
expect(before.totalSamplesPersisted, 0,
|
||||
reason: 'still buffered and not yet owned by a flush');
|
||||
expect(await persistedRows(), isEmpty);
|
||||
expect(db.batchInvocations, 0, reason: 'no flush has run yet');
|
||||
|
||||
container.dispose();
|
||||
await pumpEventQueue(times: 30);
|
||||
|
||||
// DEFERRED CONCERN D - LOCKED RESIDUAL, NOT AN ENDORSEMENT OF DATA LOSS.
|
||||
// dispose() is synchronous and starts no flush, so telemetry that is still
|
||||
// buffered and not yet owned is abandoned. RID-3B-2 deliberately does not
|
||||
// solve this; these assertions exist so the patch cannot silently pretend
|
||||
// it did. A later lifecycle ticket must decide the graceful-shutdown policy
|
||||
// (owner-invoked async drain, or documented bounded loss).
|
||||
expect(db.batchInvocations, 0, reason: 'dispose() must not start a flush');
|
||||
expect(await persistedRows(), isEmpty,
|
||||
reason: 'these samples are NOT guaranteed durable');
|
||||
expect(notifier.mounted, isFalse);
|
||||
final rideRow =
|
||||
(await db.select(db.rides).get()).singleWhere((r) => r.id == rideId);
|
||||
expect(rideRow.endTime, isNull,
|
||||
reason: 'the ride is also left open by the deferred concern');
|
||||
});
|
||||
});
|
||||
}
|
||||
Loading…
Reference in New Issue