fix: make ride start idempotent and harden sync enqueue
This commit is contained in:
Developer
parent
6386dc8697
commit
bd5a70e862
|
|
@ -121,6 +121,14 @@ class RideStateMachine extends StateNotifier<RideSession> {
|
|||
final Map<String, DateTime> _lastEventTimes = {};
|
||||
static const Duration _eventCooldown = Duration(seconds: 5);
|
||||
|
||||
// RID-1: re-entrancy guard for ride start. Raised synchronously before the
|
||||
// first await inside startRide() so concurrent callers cannot each create
|
||||
// their own ride row and cloudRideId. Its lifetime is owned exclusively by
|
||||
// the in-flight start operation (try/finally); it is intentionally NOT
|
||||
// cleared anywhere else, because clearing it while a start is still awaiting
|
||||
// would re-open the duplicate-start race.
|
||||
bool _isStarting = false;
|
||||
|
||||
RideStateMachine(this.db, this.telemetryService, this.recordsService, this.audioAlertService, this.ref) : super(RideSession()) {
|
||||
_telemetrySub = telemetryService.telemetryStream.listen(_onTelemetryUpdate);
|
||||
|
||||
|
|
@ -199,9 +207,32 @@ class RideStateMachine extends StateNotifier<RideSession> {
|
|||
);
|
||||
}
|
||||
|
||||
/// RID-1: idempotent ride start.
|
||||
///
|
||||
/// The guard is raised synchronously, before the first `await`, so that
|
||||
/// concurrent callers — the 20 Hz motion auto-start in [_onTelemetryUpdate]
|
||||
/// and rapid taps on the dashboard Start buttons — collapse into a single
|
||||
/// start operation instead of each minting its own ride row, `cloudRideId`
|
||||
/// and orphaned open ride.
|
||||
///
|
||||
/// The guard is released in `finally` so a genuinely failed start never
|
||||
/// leaves the machine permanently unable to record, and a later legitimate
|
||||
/// retry still succeeds. Frozen semantics (auto-start threshold, PR gating,
|
||||
/// kickstand isolation, recovery, `cloudRideId` generation) are untouched.
|
||||
Future<void> startRide() async {
|
||||
if (_isStarting) return;
|
||||
if (state.state == RideState.riding) return;
|
||||
_isStarting = true;
|
||||
try {
|
||||
await _performStartRide();
|
||||
} finally {
|
||||
_isStarting = false;
|
||||
}
|
||||
}
|
||||
|
||||
/// Performs the actual ride start. Guarded exclusively by [startRide] and
|
||||
/// must never be invoked directly.
|
||||
Future<void> _performStartRide() async {
|
||||
try {
|
||||
ref.read(fieldTestLoggerProvider).log(
|
||||
eventType: FieldTestEventType.rideStartRequested,
|
||||
|
|
|
|||
|
|
@ -1,3 +1,5 @@
|
|||
import 'dart:async';
|
||||
|
||||
import 'package:flutter/foundation.dart';
|
||||
import 'package:flutter_riverpod/flutter_riverpod.dart';
|
||||
import '../../features/rides/ride_state_machine.dart';
|
||||
|
|
@ -24,10 +26,39 @@ class SyncObserver {
|
|||
if (_lastEnqueuedRideId != completedRideId) {
|
||||
_lastEnqueuedRideId = completedRideId;
|
||||
debugPrint('[SyncObserver] Ride #$completedRideId completed, enqueuing to SyncManager');
|
||||
_syncManager.enqueueRide(completedRideId);
|
||||
// RID-1: the enqueue Future is now handled explicitly instead of being
|
||||
// left naked. SyncManager.enqueueRide can throw (SyncOutbox raises
|
||||
// StateError at capacity and ArgumentError on invalid input), and an
|
||||
// unhandled async exception here silently discarded the completed
|
||||
// ride's durable sync intent with no diagnostic.
|
||||
//
|
||||
// _enqueueCompletedRide is still invoked synchronously at this point,
|
||||
// so the callee's synchronous prefix runs before this method returns
|
||||
// and the outbox's in-memory mutation stays observable to callers.
|
||||
unawaited(_enqueueCompletedRide(completedRideId));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Awaits the enqueue and reports failure instead of swallowing it.
|
||||
///
|
||||
/// Diagnostics intentionally carry only the exception's runtime type, never
|
||||
/// its message or payload, so tokens, UIDs and ride content cannot leak into
|
||||
/// logs. Output is bounded to at most one line per completed ride because
|
||||
/// [_lastEnqueuedRideId] suppresses repeat notifications for the same ride.
|
||||
///
|
||||
/// Durability note: if this enqueue fails the automatic sync intent for that
|
||||
/// ride is lost. No retry or reconciliation subsystem is added here by
|
||||
/// design; see the RID-1 report for the documented follow-up risk.
|
||||
Future<void> _enqueueCompletedRide(int rideId) async {
|
||||
try {
|
||||
await _syncManager.enqueueRide(rideId);
|
||||
} catch (error) {
|
||||
debugPrint(
|
||||
'[SyncObserver] Enqueue failed for ride #$rideId class=${error.runtimeType}',
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Riverpod provider establishing the active SyncObserver listener
|
||||
|
|
|
|||
|
|
@ -0,0 +1,333 @@
|
|||
// RID-1 regression coverage: idempotent ride start.
|
||||
//
|
||||
// These tests reproduce the verified duplicate-ride defect deterministically.
|
||||
// The race window is controlled with a Completer-gated database rather than
|
||||
// sleeps, so no test here is timing-fragile.
|
||||
//
|
||||
// The gate/failure seam lives entirely in a test-only AppDatabase subclass
|
||||
// injected through the existing AppDatabase.forTesting constructor. No
|
||||
// production architecture was changed to make these tests possible.
|
||||
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:motorcycle_dash/database/database.dart' hide TelemetrySample;
|
||||
import 'package:motorcycle_dash/models/telemetry_sample.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';
|
||||
|
||||
/// Deterministic seam over the production database.
|
||||
///
|
||||
/// * [startRideInvocations] proves "at most one db.startRide() may be active".
|
||||
/// * [gate] parks the first insert in flight so concurrent callers can be
|
||||
/// exercised without any timing assumption.
|
||||
/// * [failuresRemaining] injects a genuine start failure and then recovers.
|
||||
class _Rid1AppDatabase extends AppDatabase {
|
||||
_Rid1AppDatabase() : super.forTesting(NativeDatabase.memory());
|
||||
|
||||
int startRideInvocations = 0;
|
||||
Completer<void>? gate;
|
||||
int failuresRemaining = 0;
|
||||
|
||||
@override
|
||||
Future<int> startRide({
|
||||
DateTime? startTime,
|
||||
int? motorcycleId,
|
||||
int? motorcycleYear,
|
||||
String? motorcycleMake,
|
||||
String? motorcycleModel,
|
||||
String? motorcycleVariant,
|
||||
}) async {
|
||||
startRideInvocations++;
|
||||
if (failuresRemaining > 0) {
|
||||
failuresRemaining--;
|
||||
throw StateError('RID-1 simulated database start failure');
|
||||
}
|
||||
final pendingGate = gate;
|
||||
if (pendingGate != null) await pendingGate.future;
|
||||
return super.startRide(
|
||||
startTime: startTime,
|
||||
motorcycleId: motorcycleId,
|
||||
motorcycleYear: motorcycleYear,
|
||||
motorcycleMake: motorcycleMake,
|
||||
motorcycleModel: motorcycleModel,
|
||||
motorcycleVariant: motorcycleVariant,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
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();
|
||||
|
||||
late _Rid1AppDatabase db;
|
||||
late FakeTelemetryService fakeTelemetry;
|
||||
late MockAudioAlertService mockAudio;
|
||||
late FieldTestLogger logger;
|
||||
late TelemetryValidator validator;
|
||||
late SharedPreferences prefs;
|
||||
|
||||
setUp(() async {
|
||||
SharedPreferences.setMockInitialValues({});
|
||||
prefs = await SharedPreferences.getInstance();
|
||||
db = _Rid1AppDatabase();
|
||||
logger = FieldTestLogger(db);
|
||||
validator = TelemetryValidator();
|
||||
fakeTelemetry = FakeTelemetryService(
|
||||
_NullLocationService(),
|
||||
_NullSensorFusionService(),
|
||||
_NullCompassService(),
|
||||
logger,
|
||||
validator,
|
||||
);
|
||||
mockAudio = MockAudioAlertService();
|
||||
});
|
||||
|
||||
tearDown(() async {
|
||||
// Never leave a parked gate behind or the suite would hang.
|
||||
final pendingGate = db.gate;
|
||||
if (pendingGate != null && !pendingGate.isCompleted) pendingGate.complete();
|
||||
db.gate = null;
|
||||
fakeTelemetry.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;
|
||||
}
|
||||
|
||||
TelemetrySample movingSample(int index, {double speed = 25.0}) {
|
||||
// 50 ms spacing mirrors the authoritative 20 Hz acquisition cadence.
|
||||
return TelemetrySample(
|
||||
timestamp:
|
||||
DateTime(2026, 9, 10, 12).add(Duration(milliseconds: index * 50)),
|
||||
latitude: 34.2570,
|
||||
longitude: -85.1647,
|
||||
speed: speed,
|
||||
heading: 90.0,
|
||||
gpsAccuracy: 3.0,
|
||||
);
|
||||
}
|
||||
|
||||
group('RID-1: idempotent ride start', () {
|
||||
test('20 Hz ready-state auto-start creates exactly one ride', () async {
|
||||
final container = buildContainer();
|
||||
final notifier = container.read(rideStateMachineProvider.notifier);
|
||||
final gate = Completer<void>();
|
||||
db.gate = gate;
|
||||
|
||||
notifier.setReady();
|
||||
expect(container.read(rideStateMachineProvider).state, RideState.ready);
|
||||
|
||||
// One second of authoritative 20 Hz telemetry. None of the resulting
|
||||
// startRide() triggers is awaited, exactly as the production
|
||||
// motion auto-start path does.
|
||||
for (var i = 0; i < 20; i++) {
|
||||
fakeTelemetry.emitSample(movingSample(i));
|
||||
}
|
||||
|
||||
// Deliver every queued stream event. The first db.startRide() is
|
||||
// deterministically parked on the gate, so this is the real race window.
|
||||
await pumpEventQueue(times: 50);
|
||||
|
||||
expect(db.startRideInvocations, 1,
|
||||
reason: 'at most one db.startRide() operation may be active');
|
||||
|
||||
gate.complete();
|
||||
await pumpEventQueue(times: 50);
|
||||
|
||||
final rides = await db.select(db.rides).get();
|
||||
expect(rides, hasLength(1), reason: 'exactly one ride row');
|
||||
expect(rides.map((r) => r.cloudRideId).toSet(), hasLength(1),
|
||||
reason: 'exactly one cloudRideId minted');
|
||||
|
||||
final session = container.read(rideStateMachineProvider);
|
||||
expect(session.state, RideState.riding);
|
||||
expect(session.dbId, rides.single.id,
|
||||
reason: 'session.dbId must resolve to that single row');
|
||||
});
|
||||
|
||||
test('two concurrent startRide() calls create exactly one ride', () async {
|
||||
final container = buildContainer();
|
||||
final notifier = container.read(rideStateMachineProvider.notifier);
|
||||
final gate = Completer<void>();
|
||||
db.gate = gate;
|
||||
|
||||
// Exactly what two rapid taps on the dashboard Start button do: fire
|
||||
// startRide() twice without awaiting.
|
||||
final first = notifier.startRide();
|
||||
final second = notifier.startRide();
|
||||
|
||||
await pumpEventQueue(times: 50);
|
||||
expect(db.startRideInvocations, 1, reason: 'exactly one database start');
|
||||
|
||||
gate.complete();
|
||||
await Future.wait([first, second]);
|
||||
|
||||
final rides = await db.select(db.rides).get();
|
||||
expect(rides, hasLength(1));
|
||||
expect(rides.map((r) => r.cloudRideId).toSet(), hasLength(1));
|
||||
|
||||
final session = container.read(rideStateMachineProvider);
|
||||
expect(session.state, RideState.riding,
|
||||
reason: 'one resulting riding session');
|
||||
expect(session.dbId, rides.single.id);
|
||||
});
|
||||
|
||||
test('failed database start does not report riding, clears the guard, '
|
||||
'and a later legitimate retry succeeds', () async {
|
||||
final container = buildContainer();
|
||||
final notifier = container.read(rideStateMachineProvider.notifier);
|
||||
db.failuresRemaining = 1;
|
||||
|
||||
await expectLater(notifier.startRide(), throwsStateError);
|
||||
|
||||
var session = container.read(rideStateMachineProvider);
|
||||
expect(session.state, isNot(RideState.riding),
|
||||
reason: 'state must not falsely become riding');
|
||||
expect(session.dbId, isNull);
|
||||
expect(await db.select(db.rides).get(), isEmpty,
|
||||
reason: 'a failed start must not leave a partial row');
|
||||
|
||||
// Retry the genuine start. Succeeding here proves the in-flight guard was
|
||||
// released by the finally block rather than staying stuck.
|
||||
await notifier.startRide();
|
||||
|
||||
session = container.read(rideStateMachineProvider);
|
||||
expect(session.state, RideState.riding);
|
||||
expect(session.dbId, isNotNull);
|
||||
|
||||
final rides = await db.select(db.rides).get();
|
||||
expect(rides, hasLength(1), reason: 'no duplicate rows from the retry');
|
||||
expect(db.startRideInvocations, 2);
|
||||
});
|
||||
|
||||
test('after a concurrent-start burst only one ride is left open, so '
|
||||
'recovery cannot manufacture phantom rides', () async {
|
||||
final container = buildContainer();
|
||||
final notifier = container.read(rideStateMachineProvider.notifier);
|
||||
final gate = Completer<void>();
|
||||
db.gate = gate;
|
||||
|
||||
notifier.setReady();
|
||||
for (var i = 0; i < 20; i++) {
|
||||
fakeTelemetry.emitSample(movingSample(i));
|
||||
}
|
||||
await pumpEventQueue(times: 50);
|
||||
gate.complete();
|
||||
await pumpEventQueue(times: 50);
|
||||
|
||||
final rides = await db.select(db.rides).get();
|
||||
|
||||
// Mirrors the selection criterion used by _recoverInterruptedRides()
|
||||
// (rides whose endTime is null). Recovery behaviour itself is unchanged.
|
||||
final openRides = rides.where((r) => r.endTime == null).toList();
|
||||
expect(openRides, hasLength(1),
|
||||
reason: 'exactly one open active ride after the burst');
|
||||
expect(rides, hasLength(1),
|
||||
reason: 'no additional empty orphan ride rows');
|
||||
});
|
||||
|
||||
test('frozen auto-start threshold is preserved at 1.5 MPH', () async {
|
||||
final container = buildContainer();
|
||||
final notifier = container.read(rideStateMachineProvider.notifier);
|
||||
notifier.setReady();
|
||||
|
||||
// Below the frozen threshold: must not auto-start.
|
||||
fakeTelemetry.emitSample(movingSample(0, speed: 1.4));
|
||||
await pumpEventQueue(times: 20);
|
||||
expect(container.read(rideStateMachineProvider).state, RideState.ready);
|
||||
expect(db.startRideInvocations, 0);
|
||||
|
||||
// At the frozen threshold: must auto-start.
|
||||
fakeTelemetry.emitSample(movingSample(1, speed: 1.5));
|
||||
await pumpEventQueue(times: 50);
|
||||
expect(container.read(rideStateMachineProvider).state, RideState.riding);
|
||||
expect(db.startRideInvocations, 1);
|
||||
});
|
||||
});
|
||||
}
|
||||
|
|
@ -0,0 +1,185 @@
|
|||
// RID-1 regression coverage: SyncObserver completed-ride enqueue error handling.
|
||||
//
|
||||
// Before this hardening the observer called SyncManager.enqueueRide() as a
|
||||
// naked Future. Because SyncOutbox.enqueue() throws StateError at capacity and
|
||||
// ArgumentError on invalid input, a failing enqueue surfaced as an unhandled
|
||||
// async exception and silently discarded the ride's durable sync intent.
|
||||
//
|
||||
// flutter_test fails a test on any unhandled async error, so these tests are a
|
||||
// genuine regression: they fail against the pre-fix observer.
|
||||
import 'package:flutter/foundation.dart';
|
||||
import 'package:flutter_test/flutter_test.dart';
|
||||
import 'package:shared_preferences/shared_preferences.dart';
|
||||
import 'package:motorcycle_dash/models/recording_health_status.dart';
|
||||
import 'package:motorcycle_dash/features/rides/ride_state_machine.dart';
|
||||
import 'package:motorcycle_dash/services/sync/sync_manager.dart';
|
||||
import 'package:motorcycle_dash/services/sync/sync_models.dart';
|
||||
import 'package:motorcycle_dash/services/sync/sync_observer.dart';
|
||||
import 'package:motorcycle_dash/services/sync/sync_outbox.dart';
|
||||
|
||||
/// SyncManager seam that can fail the next N enqueue attempts, then delegate to
|
||||
/// the real implementation. Test-only; no production change was required.
|
||||
class _FailingSyncManager extends SyncManager {
|
||||
_FailingSyncManager(super.outbox);
|
||||
|
||||
int enqueueAttempts = 0;
|
||||
int failuresRemaining = 0;
|
||||
|
||||
/// Payload that must never reach diagnostics.
|
||||
static const String secretPayload = 'RID-1-SENSITIVE-TOKEN-do-not-log';
|
||||
|
||||
@override
|
||||
Future<SyncItem> enqueueRide(
|
||||
int localRideId, {
|
||||
String? targetProviderId,
|
||||
bool autoProcess = true,
|
||||
}) async {
|
||||
enqueueAttempts++;
|
||||
if (failuresRemaining > 0) {
|
||||
failuresRemaining--;
|
||||
throw StateError(secretPayload);
|
||||
}
|
||||
return super.enqueueRide(
|
||||
localRideId,
|
||||
targetProviderId: targetProviderId,
|
||||
autoProcess: autoProcess,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
void main() {
|
||||
TestWidgetsFlutterBinding.ensureInitialized();
|
||||
|
||||
late SharedPreferences prefs;
|
||||
late List<String> logLines;
|
||||
late DebugPrintCallback originalDebugPrint;
|
||||
|
||||
setUp(() async {
|
||||
SharedPreferences.setMockInitialValues({});
|
||||
prefs = await SharedPreferences.getInstance();
|
||||
logLines = [];
|
||||
originalDebugPrint = debugPrint;
|
||||
debugPrint = (String? message, {int? wrapWidth}) {
|
||||
if (message != null) logLines.add(message);
|
||||
};
|
||||
});
|
||||
|
||||
tearDown(() {
|
||||
debugPrint = originalDebugPrint;
|
||||
});
|
||||
|
||||
RideSession riding(int dbId) => RideSession(
|
||||
dbId: dbId,
|
||||
state: RideState.riding,
|
||||
healthStatus: RecordingHealthStatus.recordingActive,
|
||||
);
|
||||
|
||||
RideSession finishedOf(RideSession previous) => previous.copyWith(
|
||||
state: RideState.finished,
|
||||
healthStatus: RecordingHealthStatus.recordingComplete,
|
||||
);
|
||||
|
||||
List<String> failureLines() =>
|
||||
logLines.where((l) => l.contains('Enqueue failed')).toList();
|
||||
|
||||
group('RID-1: SyncObserver enqueue hardening', () {
|
||||
test('enqueue failure is handled and does not escape as an unhandled '
|
||||
'async error', () async {
|
||||
final outbox = SyncOutbox(prefs);
|
||||
final manager = _FailingSyncManager(outbox)..failuresRemaining = 1;
|
||||
addTearDown(manager.dispose);
|
||||
final observer = SyncObserver(manager);
|
||||
|
||||
final previous = riding(900);
|
||||
observer.onRideSessionChanged(previous, finishedOf(previous));
|
||||
await pumpEventQueue(times: 20);
|
||||
|
||||
expect(manager.enqueueAttempts, 1);
|
||||
expect(outbox.count, 0,
|
||||
reason: 'a failed enqueue must not create an outbox item');
|
||||
expect(failureLines(), hasLength(1),
|
||||
reason: 'failure must be reported, not silently swallowed');
|
||||
});
|
||||
|
||||
test('diagnostics carry only the exception class, never its payload',
|
||||
() async {
|
||||
final outbox = SyncOutbox(prefs);
|
||||
final manager = _FailingSyncManager(outbox)..failuresRemaining = 1;
|
||||
addTearDown(manager.dispose);
|
||||
final observer = SyncObserver(manager);
|
||||
|
||||
final previous = riding(901);
|
||||
observer.onRideSessionChanged(previous, finishedOf(previous));
|
||||
await pumpEventQueue(times: 20);
|
||||
|
||||
expect(failureLines().single, contains('StateError'));
|
||||
expect(logLines.join('\n'),
|
||||
isNot(contains(_FailingSyncManager.secretPayload)),
|
||||
reason: 'no token/secret/data may leak into diagnostics');
|
||||
});
|
||||
|
||||
test('repeat completion notifications keep enqueue and logging bounded '
|
||||
'to one', () async {
|
||||
final outbox = SyncOutbox(prefs);
|
||||
final manager = _FailingSyncManager(outbox)..failuresRemaining = 5;
|
||||
addTearDown(manager.dispose);
|
||||
final observer = SyncObserver(manager);
|
||||
|
||||
final previous = riding(902);
|
||||
final completed = finishedOf(previous);
|
||||
observer.onRideSessionChanged(previous, completed);
|
||||
observer.onRideSessionChanged(previous, completed);
|
||||
observer.onRideSessionChanged(completed, completed);
|
||||
await pumpEventQueue(times: 20);
|
||||
|
||||
expect(manager.enqueueAttempts, 1,
|
||||
reason: 'no duplicate enqueue behaviour introduced');
|
||||
expect(failureLines(), hasLength(1), reason: 'bounded error logging');
|
||||
});
|
||||
|
||||
test('a later legitimate completion still enqueues after a failure',
|
||||
() async {
|
||||
final outbox = SyncOutbox(prefs);
|
||||
final manager = _FailingSyncManager(outbox)..failuresRemaining = 1;
|
||||
addTearDown(manager.dispose);
|
||||
final observer = SyncObserver(manager);
|
||||
|
||||
final failed = riding(903);
|
||||
observer.onRideSessionChanged(failed, finishedOf(failed));
|
||||
await pumpEventQueue(times: 20);
|
||||
expect(outbox.count, 0);
|
||||
|
||||
final later = riding(904);
|
||||
observer.onRideSessionChanged(later, finishedOf(later));
|
||||
await pumpEventQueue(times: 20);
|
||||
|
||||
expect(manager.enqueueAttempts, 2);
|
||||
expect(outbox.getItem(904, 'fake'), isNotNull,
|
||||
reason: 'the observer must stay functional after a handled failure');
|
||||
// Durable persistence is what the outbox guarantees. The item is
|
||||
// subsequently marked failed by processQueue() because this unit
|
||||
// registers no 'fake' provider - that is existing SyncManager behaviour
|
||||
// and outside RID-1 scope, so only durable presence is asserted here.
|
||||
expect(prefs.getString(outbox.storageKey), contains('"localRideId":904'),
|
||||
reason: 'the recovered enqueue must reach durable storage');
|
||||
});
|
||||
|
||||
test('successful enqueue remains observable synchronously', () {
|
||||
// Existing contract relied on by cloud_sync_hardening_test.dart, which
|
||||
// asserts outbox.count without awaiting. The RID-1 wrapper must not
|
||||
// delay the callee's synchronous prefix.
|
||||
final outbox = SyncOutbox(prefs);
|
||||
final manager = SyncManager(outbox);
|
||||
addTearDown(manager.dispose);
|
||||
final observer = SyncObserver(manager);
|
||||
|
||||
final previous = riding(905);
|
||||
observer.onRideSessionChanged(previous, finishedOf(previous));
|
||||
|
||||
expect(outbox.count, equals(1));
|
||||
|
||||
observer.onRideSessionChanged(previous, finishedOf(previous));
|
||||
expect(outbox.count, equals(1));
|
||||
});
|
||||
});
|
||||
}
|
||||
Loading…
Reference in New Issue