diff --git a/lib/features/dashboard/widgets/ride_status_chip.dart b/lib/features/dashboard/widgets/ride_status_chip.dart index cd2e366..5e9b341 100644 --- a/lib/features/dashboard/widgets/ride_status_chip.dart +++ b/lib/features/dashboard/widgets/ride_status_chip.dart @@ -7,8 +7,12 @@ import '../../../models/recording_health_status.dart'; /// Authoritative ride status badge/chip. /// -/// Subscribes strictly to [rideStateMachineProvider.select((s) => s.state)] -/// to prevent rebuilding on 20 Hz telemetry ticks. +/// Subscribes strictly to +/// [rideStateMachineProvider.select((s) => (s.state, s.healthStatus))] - the +/// narrowest selection that can render a truthful status claim. RideState decides +/// which lifecycle branch applies; the RID-3C-1 health status decides the claim. +/// Both fields change only on real lifecycle or health transitions, so this still +/// never rebuilds on 20 Hz telemetry ticks. class RideStatusChip extends ConsumerWidget { const RideStatusChip({super.key}); diff --git a/lib/services/field_test/field_test_logger.dart b/lib/services/field_test/field_test_logger.dart index 0f547e3..25b9ab7 100644 --- a/lib/services/field_test/field_test_logger.dart +++ b/lib/services/field_test/field_test_logger.dart @@ -72,6 +72,11 @@ class FieldTestLogger { Timer? _flushTimer; bool _isFlushing = false; + /// Batch A (A1): the currently in-flight flush, published so an explicit + /// [stopSession] can await the real owner instead of observing a no-op flush. + /// A coordination primitive only - not a second queue and not a second logger. + Future? _activeFlush; + // Requirement 6: Memory / Queue / Database Observability Metrics int _peakQueueDepth = 0; int _totalEnqueued = 0; @@ -136,10 +141,53 @@ class FieldTestLogger { FieldTestLogger(this._db) { _flushTimer = Timer.periodic(const Duration(seconds: 3), (_) { - flush(); + _runContainedFlush(); }); } + /// Batch A (A2): explicit ownership of the flush Future. + /// + /// A Timer.periodic callback is synchronous, and dispose() cannot await, so the + /// Future returned by [flush] would otherwise be discarded and any rejection + /// would surface as an uncaught zone error. Diagnostics are bounded to + /// runtimeType only - never the exception message, a stack trace, or any event + /// payload. No new timer, no new queue and no blocking shutdown is introduced. + void _runContainedFlush() { + unawaited( + flush().catchError((Object error, StackTrace stackTrace) { + debugPrint( + '[FieldTestLogger] event flush failed class=${error.runtimeType}', + ); + }), + ); + } + + /// Batch A (A1): bounded drain used by [stopSession] so an explicit stop cannot + /// return before its own event tail is durable. + /// + /// Awaits any in-flight owner first, then flushes what remains. Bounded rather + /// than convergent: unlike RID-3B-1 there is no event-acceptance cutoff here, so + /// anything still outstanding after the bound is simply retained in memory in + /// chronological order and retried by the periodic timer - never discarded. + static const int _maxStopDrainPasses = 3; + + Future _drainEventBatchForStop() async { + for (var pass = 0; pass < _maxStopDrainPasses; pass++) { + final active = _activeFlush; + if (active != null) { + try { + await active; + } catch (_) { + // The owner retains and restores its own snapshot on failure; ownership + // is re-checked on the next pass. + } + } + if (_eventBatch.isEmpty) return; + if (_isFlushing) continue; + await flush(); + } + } + AppIdentity? _sessionIdentity; AppIdentity get sessionIdentity => _sessionIdentity ?? AppIdentity.current(); @@ -177,8 +225,10 @@ class FieldTestLogger { rideId: rideId, ); - // Final database flush before generating summary - await flush(); + // Final database flush before generating summary. Awaits any in-flight + // periodic owner first, so a stop inside a batch interval still persists its + // tail before returning successfully (Batch A / A1). + await _drainEventBatchForStop(); final summary = await generateAndSaveSummary(rideId); _activeRideId = null; @@ -579,28 +629,73 @@ class FieldTestLogger { } } - /// Flushes buffered events to SQLite in a single transaction + /// Flushes buffered events to SQLite in a single transaction. + /// + /// Batch A (A1) ownership contract - the RID-3A/RID-3B-2 principles applied + /// with the smallest logger-specific implementation (no duplicated ride-state + /// machinery, no new queue, no new timer, no second logger, no persistence + /// redesign): + /// - the owned snapshot is taken and removed from the live buffer BEFORE the + /// await, so exactly one owner exists per event + /// - ownership is guaranteed by the synchronous [_isFlushing] guard, raised + /// before the first await; concurrent callers return without taking a + /// second snapshot + /// - the successful return of `_db.batch(...)` is the durability COMMIT POINT + /// - a PRE-commit failure restores the owned events to the FRONT of the live + /// buffer, preserving their chronological precedence over events that + /// arrived during the await, so they remain retryable + /// - a POST-commit bookkeeping failure can never restore or duplicate events + /// that are already durable + /// + /// Invariant: every accepted event is either durably persisted exactly once, or + /// retained in memory in original chronological order for retry. A pre-commit + /// persistence failure never silently discards an event. Future flush() async { if (_isFlushing || _eventBatch.isEmpty) return; _isFlushing = true; final stopwatch = Stopwatch()..start(); + final toWrite = List.from(_eventBatch); + _eventBatch.clear(); + // Publish this exact in-flight operation so an explicit stopSession() can + // await the real owner instead of observing a no-op flush and returning + // before its own tail is durable. + final completion = Completer(); + _activeFlush = completion.future; try { - final toWrite = List.from(_eventBatch); - _eventBatch.clear(); - - await _db.batch((batch) { - batch.insertAll(_db.fieldTestEvents, toWrite); - }); + // ---- PRE-COMMIT PHASE: the only block that may fail persistence. ---- + try { + await _db.batch((batch) { + batch.insertAll(_db.fieldTestEvents, toWrite); + }); + } catch (e) { + // Not durable, so restore the owned events to the FRONT of the live + // buffer. Owned A B C plus arrivals D E during the await must become + // A B C D E - never D E A B C - so chronological order survives retry. + _eventBatch.insertAll(0, toWrite); + _failedFlushCount++; + // Bounded diagnostic: operation, runtimeType and event count only. The + // exception object is never logged because its message may carry + // implementation or payload detail. + if (kDebugMode) { + debugPrint( + '[FieldTestLogger] event batch write failed ' + 'class=${e.runtimeType} events=${toWrite.length}', + ); + } + return; + } + // ---- COMMIT POINT REACHED: the owned events are durable. ---- + // Post-commit bookkeeping only. Nothing below may restore the snapshot or + // cause a second write of events that are already durable. _totalFlushed += toWrite.length; stopwatch.stop(); _lastFlushDurationMs = stopwatch.elapsedMilliseconds; - } catch (e) { - _failedFlushCount++; - if (kDebugMode) print('[FieldTestLogger] Flush error: $e'); } finally { _isFlushing = false; + if (!completion.isCompleted) completion.complete(); + _activeFlush = null; } } @@ -826,7 +921,14 @@ class FieldTestLogger { void dispose() { _flushTimer?.cancel(); - flush(); + // Batch A (A2): dispose() is NOT an awaitable OS durability boundary and this + // makes NO persistence guarantee. The flush is best-effort only, but it is + // explicitly owned so no rejection escapes as an unhandled async error, and + // flush()'s own guards keep a pre-commit failure safe: owned events are + // restored to the in-memory buffer rather than discarded, and nothing mutates + // state after the await. stopSession() remains the real awaitable + // logger-finalization path. + _runContainedFlush(); _timelineStreamController.close(); } } diff --git a/test/rid3c3a0_field_test_logger_durability_test.dart b/test/rid3c3a0_field_test_logger_durability_test.dart new file mode 100644 index 0000000..77bced7 --- /dev/null +++ b/test/rid3c3a0_field_test_logger_durability_test.dart @@ -0,0 +1,423 @@ +// Batch A (A1/A2) permanent regression suite: FieldTestLogger persistence +// ownership and durability. +// +// Locks the invariant that every event accepted into FieldTestLogger is either +// durably persisted EXACTLY ONCE, or retained in memory in original chronological +// order for retry. A pre-commit persistence failure must never silently discard +// an event, and a post-commit bookkeeping failure must never requeue durable +// events. Also locks async containment of the periodic timer and dispose() flush, +// and the runtimeType-only diagnostic (no exception-message disclosure). +import 'dart:async'; + +import 'package:flutter/foundation.dart'; +import 'package:flutter_test/flutter_test.dart'; +import 'package:drift/native.dart'; +import 'package:drift/drift.dart' as drift; +import 'package:motorcycle_dash/database/database.dart' + hide FieldTestEvent, FieldTestSummary, TelemetrySample; +import 'package:motorcycle_dash/database/database.dart' as dbgen + show FieldTestEvent; +import 'package:motorcycle_dash/models/field_test_event.dart'; +import 'package:motorcycle_dash/services/field_test/field_test_logger.dart'; + +typedef EventRow = dbgen.FieldTestEvent; + +/// Counts every `batch` call and can park or fail it deterministically, so a flush +/// can be held exactly at its durability commit point without a sleep-only race. +class _BatchAAppDatabase extends AppDatabase { + _BatchAAppDatabase() : super.forTesting(NativeDatabase.memory()); + + int batchInvocations = 0; + bool gateBatch = false; + final List> batchGates = []; + int batchFailuresRemaining = 0; + + @override + Future batch( + FutureOr Function(drift.Batch batch) runInBatch) async { + batchInvocations++; + if (gateBatch) { + final gate = Completer(); + batchGates.add(gate); + await gate.future; + } + if (batchFailuresRemaining > 0) { + batchFailuresRemaining--; + throw StateError('BatchA simulated pre-commit persistence failure'); + } + return super.batch(runInBatch); + } +} + +void main() { + TestWidgetsFlutterBinding.ensureInitialized(); + + /// The production logger flush timer is a real 3-second Timer.periodic. + const tickWait = Duration(milliseconds: 3300); + + late _BatchAAppDatabase db; + late FieldTestLogger logger; + var seq = 0; + + /// Captures debugPrint output so the bounded diagnostic can be asserted to + /// contain no exception message. Restored in tearDown. + final capturedPrints = []; + late DebugPrintCallback originalDebugPrint; + + setUp(() async { + db = _BatchAAppDatabase(); + logger = FieldTestLogger(db); + seq = 0; + capturedPrints.clear(); + originalDebugPrint = debugPrint; + debugPrint = (String? message, {int? wrapWidth}) { + if (message != null) capturedPrints.add(message); + }; + }); + + tearDown(() async { + debugPrint = originalDebugPrint; + for (final gate in db.batchGates) { + if (!gate.isCompleted) gate.complete(); + } + db.batchGates.clear(); + db.gateBatch = false; + db.batchFailuresRemaining = 0; + logger.dispose(); + await db.close(); + }); + + /// Enqueues a uniquely identifiable event. Identity lives in the message ONLY + /// for test accounting; the assertions never treat message text as a diagnostic. + void logEvent(String tag, {EventSeverity severity = EventSeverity.info}) { + logger.log( + eventType: FieldTestEventType.sampleAccepted, + severity: severity, + message: 'EVT-$tag-${seq++}', + rideId: 1, + timestamp: DateTime(2026, 9, 10, 12).add(Duration(milliseconds: seq * 50)), + ); + } + + Future> persistedEvents() => + (db.select(db.fieldTestEvents) + ..orderBy([(e) => drift.OrderingTerm(expression: e.id)])) + .get(); + + /// Extracts the ordered identity tags from durable rows. + List tagsOf(List rows) => rows + .map((r) => r.message.split('-').skip(1).take(2).join('-')) + .toList(); + + Map tagCounts(List rows) { + final counts = {}; + for (final tag in tagsOf(rows)) { + counts[tag] = (counts[tag] ?? 0) + 1; + } + return counts; + } + + List duplicatedTags(List rows) => + tagCounts(rows).entries.where((e) => e.value > 1).map((e) => e.key).toList() + ..sort(); + + /// Runs [body] in a guarded zone and returns every uncaught async error, so + /// timer/dispose Future ownership regressions surface as data, not as an opaque + /// failure. + Future> captureUncaught(Future Function() body) async { + final uncaught = []; + await runZonedGuarded(body, (error, stackTrace) => uncaught.add(error)); + return uncaught; + } + + group('Batch A: FieldTestLogger durability', () { + test('F1 a successful flush persists owned events exactly once', () async { + logEvent('A'); + logEvent('B'); + logEvent('C'); + expect(logger.currentQueueDepth, 3); + + await logger.flush(); + + final rows = await persistedEvents(); + expect(rows, hasLength(3)); + expect(duplicatedTags(rows), isEmpty); + expect(logger.currentQueueDepth, 0, + reason: 'the owned snapshot is removed before the await'); + expect(logger.totalFlushed, 3, reason: 'success-only accounting'); + expect(logger.failedFlushCount, 0); + expect(db.batchInvocations, 1); + }); + + test('F2 a pre-commit failure restores the owned snapshot', () async { + logEvent('A'); + logEvent('B'); + logEvent('C'); + + db.batchFailuresRemaining = 1; + await logger.flush(); + + expect(await persistedEvents(), isEmpty, + reason: 'a failed attempt commits nothing'); + expect(logger.currentQueueDepth, 3, + reason: 'owned events are retained for retry, never discarded'); + expect(logger.failedFlushCount, 1, + reason: 'existing failure accounting preserved'); + expect(logger.totalFlushed, 0); + }); + + test('F3 arrivals during a failed flush keep order A B C D E', () async { + logEvent('A'); + logEvent('B'); + logEvent('C'); + + db.gateBatch = true; + final pending = logger.flush(); + await pumpEventQueue(times: 20); + expect(db.batchInvocations, 1, reason: 'flush parked at the commit point'); + + // D and E arrive DURING the await and must end up behind the restored + // snapshot: A B C D E, never D E A B C. + logEvent('D'); + logEvent('E'); + + db.batchFailuresRemaining = 1; + db.gateBatch = false; + db.batchGates.single.complete(); + await pending; + + expect(await persistedEvents(), isEmpty); + expect(logger.currentQueueDepth, 5, + reason: 'owned A B C restored in front of arrivals D E'); + + // Order is proven behaviourally by the retry's durable row order. + await logger.flush(); + final rows = await persistedEvents(); + expect(tagsOf(rows), ['A-0', 'B-1', 'C-2', 'D-3', 'E-4'], + reason: 'chronological order survives restoration and retry'); + expect(duplicatedTags(rows), isEmpty); + }); + + test('F4 failure then retry yields no duplicates and nothing missing', + () async { + for (final tag in ['A', 'B', 'C', 'D']) { + logEvent(tag); + } + + db.batchFailuresRemaining = 1; + await logger.flush(); + expect(await persistedEvents(), isEmpty); + expect(logger.currentQueueDepth, 4); + + await logger.flush(); + final rows = await persistedEvents(); + expect(rows, hasLength(4), reason: 'every event eventually durable'); + expect(duplicatedTags(rows), isEmpty, reason: 'exactly once'); + expect(tagCounts(rows).keys.toList()..sort(), ['A-0', 'B-1', 'C-2', 'D-3'], + reason: 'nothing missing'); + expect(logger.failedFlushCount, 1); + expect(logger.totalFlushed, 4); + }); + + test('F5 arrivals during a successful flush are left for the next flush', + () async { + logEvent('A'); + logEvent('B'); + logEvent('C'); + + db.gateBatch = true; + final pending = logger.flush(); + await pumpEventQueue(times: 20); + logEvent('D'); + logEvent('E'); + db.gateBatch = false; + db.batchGates.single.complete(); + await pending; + + var rows = await persistedEvents(); + expect(tagsOf(rows), ['A-0', 'B-1', 'C-2'], + reason: 'only the owned snapshot commits'); + expect(logger.currentQueueDepth, 2, reason: 'D E remain buffered'); + + await logger.flush(); + rows = await persistedEvents(); + expect(tagsOf(rows), ['A-0', 'B-1', 'C-2', 'D-3', 'E-4']); + expect(duplicatedTags(rows), isEmpty, reason: 'exactly once overall'); + expect(db.batchInvocations, 2); + expect(logger.totalFlushed, 5); + }); + + test('F6 concurrent flush callers yield exactly one owner per event', + () async { + logEvent('A'); + logEvent('B'); + logEvent('C'); + + db.gateBatch = true; + final first = logger.flush(); + await pumpEventQueue(times: 20); + expect(db.batchInvocations, 1); + + // Concurrent callers must not take a second snapshot of the same events. + final others = [logger.flush(), logger.flush(), logger.flush()]; + await pumpEventQueue(times: 20); + expect(db.batchInvocations, 1, + reason: 'the synchronous ownership guard is raised before the await'); + + db.gateBatch = false; + db.batchGates.single.complete(); + await first; + await Future.wait(others); + + final rows = await persistedEvents(); + expect(tagsOf(rows), ['A-0', 'B-1', 'C-2']); + expect(duplicatedTags(rows), isEmpty); + expect(db.batchInvocations, 1); + }); + test('F7 stopSession coordinates safely with an in-flight periodic flush', + () async { + logger.startSession(1); + logEvent('A'); + logEvent('B'); + logEvent('C'); + + db.gateBatch = true; + final periodic = logger.flush(); + await pumpEventQueue(times: 20); + expect(db.batchInvocations, 1, reason: 'the periodic owner holds A B C'); + + // Tail arrives while another owner is in flight. + logEvent('D'); + + db.gateBatch = false; + final stop = logger.stopSession(); + for (final gate in db.batchGates) { + if (!gate.isCompleted) gate.complete(); + } + await periodic; + await stop; + + final rows = (await persistedEvents()) + .where((r) => r.message.startsWith('EVT-')) + .toList(); + expect(tagsOf(rows), ['A-0', 'B-1', 'C-2', 'D-3'], + reason: 'exactly once each, in chronological order, tail preserved'); + expect(duplicatedTags(rows), isEmpty); + expect(logger.currentQueueDepth, 0, + reason: 'stopSession drained the tail it owned'); + }); + + test('F8 failure accounting is preserved across mixed outcomes', () async { + logEvent('A'); + db.batchFailuresRemaining = 1; + await logger.flush(); + expect(logger.failedFlushCount, 1); + expect(logger.totalFlushed, 0, reason: 'success-only accounting'); + + logEvent('B'); + await logger.flush(); + expect(logger.failedFlushCount, 1, reason: 'no additional failure'); + expect(logger.totalFlushed, 2, reason: 'A retried plus B'); + expect(await persistedEvents(), hasLength(2)); + }); + + test('F9 the failure diagnostic discloses no exception message', () async { + logEvent('A'); + db.batchFailuresRemaining = 1; + await logger.flush(); + + final diagnostics = + capturedPrints.where((m) => m.contains('[FieldTestLogger]')).toList(); + expect(diagnostics, isNotEmpty, reason: 'a bounded diagnostic is emitted'); + for (final message in diagnostics) { + expect(message, contains('class=StateError'), + reason: 'runtimeType only, per the MotorSignal precedent'); + expect(message, contains('events=1'), reason: 'bounded event count'); + expect(message, isNot(contains('simulated')), + reason: 'the exception message must never be disclosed'); + expect(message, isNot(contains('BatchA')), + reason: 'no exception text or payload detail'); + } + }); + + test('F10 post-commit bookkeeping cannot requeue durable events', () async { + logEvent('A'); + logEvent('B'); + await logger.flush(); + + expect(await persistedEvents(), hasLength(2)); + expect(logger.currentQueueDepth, 0); + expect(logger.failedFlushCount, 0, + reason: 'nothing after the commit point is a persistence failure'); + expect(logger.totalFlushed, 2); + + // Restoration lives only in the pre-commit catch, so no later pass may + // re-write events that are already durable. + await logger.flush(); + await logger.flush(); + expect(await persistedEvents(), hasLength(2), reason: 'never duplicated'); + expect(db.batchInvocations, 1, reason: 'an empty buffer is a no-op'); + }); + + test('F11 an empty flush is a no-op', () async { + await logger.flush(); + expect(db.batchInvocations, 0); + expect(logger.totalFlushed, 0); + expect(logger.failedFlushCount, 0); + expect(logger.currentQueueDepth, 0); + }); + + test('F12 a sub-interval stopSession persists its tail before returning', + () async { + logger.startSession(1); + logEvent('A'); + logEvent('B'); + expect(db.batchInvocations, 0, reason: 'no periodic tick has run yet'); + + await logger.stopSession(); + + final rows = (await persistedEvents()) + .where((r) => r.message.startsWith('EVT-')) + .toList(); + expect(tagsOf(rows), ['A-0', 'B-1'], + reason: 'the tail is durable before stopSession returned'); + expect(logger.currentQueueDepth, 0); + }); + + test('F13 timer and dispose flush failures raise no uncaught error', + () async { + final uncaught = await captureUncaught(() async { + // Periodic-timer path: a real failure inside a timer-owned flush. + logEvent('A'); + db.batchFailuresRemaining = 1; + await Future.delayed(tickWait); + await pumpEventQueue(times: 40); + + // dispose() path: an unawaited, explicitly owned flush that fails. + final localLogger = FieldTestLogger(db); + localLogger.log( + eventType: FieldTestEventType.sampleAccepted, + severity: EventSeverity.info, + message: 'EVT-DISPOSE-0', + rideId: 1, + ); + db.batchFailuresRemaining = 1; + localLogger.dispose(); + await pumpEventQueue(times: 40); + }); + + expect(uncaught, isEmpty, + reason: 'both the timer and dispose flush Futures must be explicitly ' + 'owned so no rejection reaches the zone: $uncaught'); + // A pre-commit failure during dispose retains rather than discards, and + // commits nothing. + expect( + (await persistedEvents()) + .where((r) => r.message.startsWith('EVT-')) + .toList(), + isEmpty, + reason: 'failed attempts commit nothing'); + expect(logger.failedFlushCount, greaterThanOrEqualTo(1)); + }); + }); +}