fix: make field test event persistence ownership-safe
This commit is contained in:
parent
c227244d5f
commit
304d2db108
|
|
@ -7,8 +7,12 @@ import '../../../models/recording_health_status.dart';
|
||||||
|
|
||||||
/// Authoritative ride status badge/chip.
|
/// Authoritative ride status badge/chip.
|
||||||
///
|
///
|
||||||
/// Subscribes strictly to [rideStateMachineProvider.select((s) => s.state)]
|
/// Subscribes strictly to
|
||||||
/// to prevent rebuilding on 20 Hz telemetry ticks.
|
/// [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 {
|
class RideStatusChip extends ConsumerWidget {
|
||||||
const RideStatusChip({super.key});
|
const RideStatusChip({super.key});
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -72,6 +72,11 @@ class FieldTestLogger {
|
||||||
Timer? _flushTimer;
|
Timer? _flushTimer;
|
||||||
bool _isFlushing = false;
|
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<void>? _activeFlush;
|
||||||
|
|
||||||
// Requirement 6: Memory / Queue / Database Observability Metrics
|
// Requirement 6: Memory / Queue / Database Observability Metrics
|
||||||
int _peakQueueDepth = 0;
|
int _peakQueueDepth = 0;
|
||||||
int _totalEnqueued = 0;
|
int _totalEnqueued = 0;
|
||||||
|
|
@ -136,10 +141,53 @@ class FieldTestLogger {
|
||||||
|
|
||||||
FieldTestLogger(this._db) {
|
FieldTestLogger(this._db) {
|
||||||
_flushTimer = Timer.periodic(const Duration(seconds: 3), (_) {
|
_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<void> _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? _sessionIdentity;
|
||||||
AppIdentity get sessionIdentity => _sessionIdentity ?? AppIdentity.current();
|
AppIdentity get sessionIdentity => _sessionIdentity ?? AppIdentity.current();
|
||||||
|
|
||||||
|
|
@ -177,8 +225,10 @@ class FieldTestLogger {
|
||||||
rideId: rideId,
|
rideId: rideId,
|
||||||
);
|
);
|
||||||
|
|
||||||
// Final database flush before generating summary
|
// Final database flush before generating summary. Awaits any in-flight
|
||||||
await flush();
|
// 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);
|
final summary = await generateAndSaveSummary(rideId);
|
||||||
_activeRideId = null;
|
_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<void> flush() async {
|
Future<void> flush() async {
|
||||||
if (_isFlushing || _eventBatch.isEmpty) return;
|
if (_isFlushing || _eventBatch.isEmpty) return;
|
||||||
_isFlushing = true;
|
_isFlushing = true;
|
||||||
final stopwatch = Stopwatch()..start();
|
final stopwatch = Stopwatch()..start();
|
||||||
|
final toWrite = List<FieldTestEventsCompanion>.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<void>();
|
||||||
|
_activeFlush = completion.future;
|
||||||
|
|
||||||
try {
|
try {
|
||||||
final toWrite = List<FieldTestEventsCompanion>.from(_eventBatch);
|
// ---- PRE-COMMIT PHASE: the only block that may fail persistence. ----
|
||||||
_eventBatch.clear();
|
try {
|
||||||
|
await _db.batch((batch) {
|
||||||
await _db.batch((batch) {
|
batch.insertAll(_db.fieldTestEvents, toWrite);
|
||||||
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;
|
_totalFlushed += toWrite.length;
|
||||||
stopwatch.stop();
|
stopwatch.stop();
|
||||||
_lastFlushDurationMs = stopwatch.elapsedMilliseconds;
|
_lastFlushDurationMs = stopwatch.elapsedMilliseconds;
|
||||||
} catch (e) {
|
|
||||||
_failedFlushCount++;
|
|
||||||
if (kDebugMode) print('[FieldTestLogger] Flush error: $e');
|
|
||||||
} finally {
|
} finally {
|
||||||
_isFlushing = false;
|
_isFlushing = false;
|
||||||
|
if (!completion.isCompleted) completion.complete();
|
||||||
|
_activeFlush = null;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -826,7 +921,14 @@ class FieldTestLogger {
|
||||||
|
|
||||||
void dispose() {
|
void dispose() {
|
||||||
_flushTimer?.cancel();
|
_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();
|
_timelineStreamController.close();
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -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<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('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 = <String>[];
|
||||||
|
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<List<EventRow>> persistedEvents() =>
|
||||||
|
(db.select(db.fieldTestEvents)
|
||||||
|
..orderBy([(e) => drift.OrderingTerm(expression: e.id)]))
|
||||||
|
.get();
|
||||||
|
|
||||||
|
/// Extracts the ordered identity tags from durable rows.
|
||||||
|
List<String> tagsOf(List<EventRow> rows) => rows
|
||||||
|
.map((r) => r.message.split('-').skip(1).take(2).join('-'))
|
||||||
|
.toList();
|
||||||
|
|
||||||
|
Map<String, int> tagCounts(List<EventRow> rows) {
|
||||||
|
final counts = <String, int>{};
|
||||||
|
for (final tag in tagsOf(rows)) {
|
||||||
|
counts[tag] = (counts[tag] ?? 0) + 1;
|
||||||
|
}
|
||||||
|
return counts;
|
||||||
|
}
|
||||||
|
|
||||||
|
List<String> duplicatedTags(List<EventRow> 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<List<Object>> captureUncaught(Future<void> Function() body) async {
|
||||||
|
final uncaught = <Object>[];
|
||||||
|
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<void>.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));
|
||||||
|
});
|
||||||
|
});
|
||||||
|
}
|
||||||
Loading…
Reference in New Issue