motorsignal
/
MotorSignal-App
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix(sync): harden cloud confirmation states, R2 edge 404 handling, and in-flight cancellation

This commit is contained in:
Developer 2026-10-13 22:05:25 -04:00
parent d431f11a13
commit 5e0db4c191
7 changed files with 483 additions and 22 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
cloudflare/motorsignal-r2-worker/src/index.ts
View File

@ -26,7 +26,7 @@ const CORS_HEADERS = {
'Access-Control-Allow-Origin': '*',
'Access-Control-Allow-Methods': 'GET, POST, PUT, DELETE, OPTIONS',
'Access-Control-Allow-Headers': 'Content-Type, Authorization, X-Remote-Ride-Id, X-Bundle-Sha256, X-App-Version',
'Access-Control-Expose-Headers': 'ETag, Content-Disposition, X-MotorSignal-Ride-Id, X-MotorSignal-Sha256',
'Access-Control-Expose-Headers': 'ETag, Content-Disposition, X-MotorSignal-Ride-Id, X-MotorSignal-Sha256, X-MotorSignal-Size',
};
const FIREBASE_JWKS = new URL('https://www.googleapis.com/service_accounts/v1/jwk/securetoken@system.gserviceaccount.com');

42
lib/services/sync/firebase_sync_provider.dart
View File

@ -265,6 +265,11 @@ class FirebaseSyncProvider implements SyncProvider {
'Firebase services are unavailable or unconfigured. Ride remains safely stored locally.',
);
}
if (ride == null) {
return SyncResult.permanentFailure(
'Local ride #${item.localRideId} not found in database.',
);
}
// Refresh the Firebase ID token before a manual upload. This prevents a
// recently completed sign-in from being rejected because the cached token
@ -280,21 +285,29 @@ class FirebaseSyncProvider implements SyncProvider {
}
try {
final ride = await (_db.select(
_db.rides,
)..where((r) => r.id.equals(item.localRideId))).getSingleOrNull();
if (ride == null) {
return SyncResult.permanentFailure(
'Local ride #${item.localRideId} not found in database.',
);
}
// The cloud identity is generated once and then reused forever. Never
// derive the remote identity from the local auto-increment database ID.
final remoteRideId = await _db.ensureCloudRideId(item.localRideId);
if (await isRideConfirmedInCloud(remoteRideId)) {
final confirmation = await verifyCloudConfirmation(remoteRideId);
if (confirmation.isConfirmed) {
return SyncResult.success(remoteId: remoteRideId);
}
if (confirmation.isInconsistent) {
return SyncResult.permanentFailure(
'Cloud data inconsistent for ride $remoteRideId: ${confirmation.message}. Re-upload aborted to prevent overwriting cloud object.',
);
}
if (confirmation.isUnauthorized) {
return SyncResult.authRequired(
message:
'Cloud verification unauthorized for ride $remoteRideId: ${confirmation.message}',
);
}
if (confirmation.isTemporarilyUnverifiable) {
return SyncResult.temporaryFailure(
'Cloud confirmation temporarily unavailable for ride $remoteRideId: ${confirmation.message}',
);
}
final appIdentity = AppIdentity.current();
final zipBytes = await _bundleWorker.generateCompressedBundle(
@ -351,9 +364,14 @@ class FirebaseSyncProvider implements SyncProvider {
);
}
final objectKey =
r2Result.objectKey ??
final expectedObjectKey =
'users/${user.uid}/rides/$remoteRideId/telemetry.zip';
final objectKey = r2Result.objectKey ?? expectedObjectKey;
if (objectKey != expectedObjectKey) {
return SyncResult.permanentFailure(
'R2 Worker returned an unexpected object key ($objectKey); expected $expectedObjectKey.',
);
}
final events = await (_db.select(
_db.rideEvents,
)..where((e) => e.rideId.equals(item.localRideId))).get();

33
lib/services/sync/r2_upload_service.dart
View File

@ -367,16 +367,35 @@ class R2UploadService {
if (status == 404) {
// Distinguish between:
// 1. Live worker without /r2/verify endpoint ("Endpoint not found") -> temporarilyUnverifiable
// 2. Telemetry bundle or ride not found -> missing
// 2. Non-JSON 404 (e.g. Cloudflare proxy / edge HTML 404) -> temporarilyUnverifiable
// 3. Telemetry bundle or ride not found from API -> missing
try {
final decoded = jsonDecode(response.body);
if (decoded is Map && decoded['error'] == 'Endpoint not found') {
return const R2VerificationResponse.temporarilyUnverifiable(
'R2 verification endpoint not yet deployed on worker.',
statusCode: 404,
);
if (decoded is Map) {
final error = decoded['error']?.toString();
final code = decoded['code']?.toString();
if (error == 'Endpoint not found') {
return const R2VerificationResponse.temporarilyUnverifiable(
'R2 verification endpoint not yet deployed on worker.',
statusCode: 404,
);
}
if (code == 'OBJECT_NOT_FOUND' ||
code == 'NOT_FOUND' ||
error == 'Telemetry bundle not found in R2' ||
error == 'Ride not found') {
return R2VerificationResponse.missing(
error ?? 'Telemetry bundle or ride not found in cloud.',
statusCode: 404,
);
}
}
} catch (_) {}
} catch (_) {
return const R2VerificationResponse.temporarilyUnverifiable(
'Non-JSON 404 received from R2 gateway; worker endpoint may be undeployed or edge routing failed.',
statusCode: 404,
);
}
return R2VerificationResponse.missing(
'Telemetry bundle or ride not found in cloud ($status).',
statusCode: status,

61
lib/services/sync/sync_manager.dart
View File

@ -72,10 +72,16 @@ class SyncManager extends ChangeNotifier with WidgetsBindingObserver {
this._authService,
bool autoStart = true,
}) : _nowProvider = nowProvider ?? (() => DateTime.now()) {
if (defaultProviderId != null) _activeProviderId = defaultProviderId;
if (defaultProviderId != null) {
_activeProviderId = defaultProviderId;
} else if (initialProviders != null && initialProviders.isNotEmpty) {
_activeProviderId = initialProviders.first.providerId;
} else {
_activeProviderId = 'fake';
}
if (initialProviders != null) {
for (final p in initialProviders) {
registerProvider(p);
_providers[p.providerId] = p;
}
}
WidgetsBinding.instance.addObserver(this);
@ -130,11 +136,16 @@ class SyncManager extends ChangeNotifier with WidgetsBindingObserver {
}
void registerProvider(SyncProvider provider) {
final wasRegistered = _providers.containsKey(provider.providerId);
_providers[provider.providerId] = provider;
if (_activeProviderId == 'none' || _activeProviderId == 'fake') {
_activeProviderId = provider.providerId;
}
notifyListeners();
if (!wasRegistered && _db != null) {
unawaited(reconcileCompletedRides(targetProviderId: provider.providerId));
unawaited(processQueue());
}
}
SyncProvider? getProvider(String providerId) => _providers[providerId];
@ -149,6 +160,30 @@ class SyncManager extends ChangeNotifier with WidgetsBindingObserver {
final user = _authService?.currentUser;
final isAuthenticated = user != null && !user.isAnonymous;
final rideOwner = await _db.getCloudOwnerUid(localRideId);
if (rideOwner != null && isAuthenticated && rideOwner != user.uid) {
final cancelledItem = SyncItem(
localRideId: localRideId,
providerId: providerId,
status: SyncStatus.cancelled,
createdAt: _now(),
lastError: 'Ride belongs to another account and cannot be uploaded',
);
await _outbox.update(cancelledItem);
notifyListeners();
return cancelledItem;
}
if (rideOwner != null && !isAuthenticated) {
final cancelledItem = SyncItem(
localRideId: localRideId,
providerId: providerId,
status: SyncStatus.cancelled,
createdAt: _now(),
lastError: 'Cannot upload another user\'s ride while unauthenticated',
);
await _outbox.update(cancelledItem);
notifyListeners();
return cancelledItem;
}
if (isAuthenticated && rideOwner == null) {
await _db.claimRideForCloudOwner(localRideId, user.uid);
}
@ -274,6 +309,22 @@ class SyncManager extends ChangeNotifier with WidgetsBindingObserver {
// Check if already tracked in outbox for this provider
final existingItem = _outbox.getItem(ride.id, providerId);
if (existingItem != null) {
if (!existingItem.status.isTerminal &&
ride.cloudOwnerUid != null &&
isAuthenticated &&
ride.cloudOwnerUid != user.uid &&
providerId == 'firebase') {
await _outbox.update(
existingItem.copyWith(
status: SyncStatus.cancelled,
lastError:
'Ride belongs to another account and cannot be uploaded',
clearNextRetry: true,
),
);
foreignOwnerSkipped++;
continue;
}
alreadyTracked++;
continue;
}
@ -611,6 +662,12 @@ class SyncManager extends ChangeNotifier with WidgetsBindingObserver {
continue;
}
// 4. In-flight cancellation protection: verify item was not cancelled while uploading
final currentItem = _outbox.getItem(item.localRideId, item.providerId);
if (currentItem?.status == SyncStatus.cancelled) {
continue;
}
if (result.success) {
await _outbox.update(
uploadingItem.copyWith(

14
lib/services/sync/sync_provider.dart
View File

@ -222,10 +222,24 @@ class FakeSyncProvider implements SyncProvider {
uploadAttemptsPerRide[item.localRideId] =
(uploadAttemptsPerRide[item.localRideId] ?? 0) + 1;
final existingRemoteId = item.remoteId;
if (existingRemoteId != null) {
final confirmation = await verifyCloudConfirmation(existingRemoteId);
if (confirmation.isConfirmed) {
return SyncResult.success(remoteId: existingRemoteId);
}
if (confirmation.isInconsistent) {
return SyncResult.permanentFailure(
'Cloud data inconsistent: ${confirmation.message}',
);
}
}
switch (behavior) {
case FakeSyncBehavior.success:
uploadedItems.add(item);
final remoteId =
existingRemoteId ??
'remote_${item.localRideId}_${DateTime.now().millisecondsSinceEpoch}';
confirmedRemoteRideIds.add(remoteId);
confirmationResults[remoteId] = CloudConfirmationResult.confirmed(

196
test/cloud_confirmation_contract_test.dart
View File

@ -12,6 +12,7 @@ import 'package:motorcycle_dash/services/auth/auth_user.dart';
import 'package:motorcycle_dash/services/sync/firebase_sync_provider.dart';
import 'package:motorcycle_dash/services/sync/r2_upload_service.dart';
import 'package:motorcycle_dash/services/sync/sync_provider.dart';
import 'package:motorcycle_dash/services/sync/sync_models.dart';
class _MockAuthService implements AuthService {
AuthUser? _currentUser;
@ -562,5 +563,200 @@ void main() {
expect(result.status, equals(CloudConfirmationStatus.confirmed));
},
);
test(
'14. Edge Routing: Non-JSON HTML 404 from proxy returns temporarilyUnverifiable instead of missing',
() async {
await fakeFirestore
.collection('users')
.doc(testUid)
.collection('rides')
.doc(remoteRideId)
.set(createFirestoreRideData());
final mockClient = http_testing.MockClient((request) async {
return http.Response(
'<html><head><title>404 Not Found</title></head><body><h1>404 Not Found</h1>cloudflare</body></html>',
404,
);
});
final r2Service = R2UploadService(client: mockClient);
final provider = FirebaseSyncProvider(
db: db,
authService: authService,
firestore: fakeFirestore,
r2UploadService: r2Service,
);
final result = await provider.verifyCloudConfirmation(remoteRideId);
expect(
result.status,
equals(CloudConfirmationStatus.temporarilyUnverifiable),
);
},
);
test(
'15. uploadRide Protection: Aborts with permanentFailure when cloud confirmation is inconsistent (never overwrites object)',
() async {
final rideId = await db.startRide(
startTime: DateTime.utc(2026, 10, 12, 10, 0, 0),
);
await db.endRide(rideId, const RidesCompanion());
await db.claimRideForCloudOwner(rideId, testUid);
final actualRemoteRideId = await db.ensureCloudRideId(rideId);
// Firestore document exists
await fakeFirestore
.collection('users')
.doc(testUid)
.collection('rides')
.doc(actualRemoteRideId)
.set(createFirestoreRideData(rideId: actualRemoteRideId));
var uploadAttempted = false;
final mockClient = http_testing.MockClient((request) async {
if (request.url.path == '/r2/verify') {
// Return inconsistent hash
return http.Response(
jsonEncode({
'verified': true,
'remoteRideId': actualRemoteRideId,
'objectKey': 'users/$testUid/rides/$actualRemoteRideId/telemetry.zip',
'size': validSize,
'sha256': 'corrupted_hash_mismatch_123',
}),
200,
);
}
if (request.url.path == '/r2/upload') {
uploadAttempted = true;
return http.Response('{}', 200);
}
return http.Response('{}', 404);
});
final r2Service = R2UploadService(client: mockClient);
final provider = FirebaseSyncProvider(
db: db,
authService: authService,
firestore: fakeFirestore,
r2UploadService: r2Service,
);
final result = await provider.uploadRide(
SyncItem(
localRideId: rideId,
providerId: 'firebase',
createdAt: DateTime.now(),
),
);
expect(result.success, isFalse);
expect(result.isRetryable, isFalse);
expect(result.errorMessage, contains('inconsistent'));
expect(uploadAttempted, isFalse, reason: 'Must NEVER attempt R2 upload when cloud data is inconsistent');
},
);
test(
'16. uploadRide Protection: Defers with temporaryFailure when cloud verification is temporarilyUnverifiable (no heavy upload)',
() async {
final rideId = await db.startRide(
startTime: DateTime.utc(2026, 10, 12, 10, 0, 0),
);
await db.endRide(rideId, const RidesCompanion());
await db.claimRideForCloudOwner(rideId, testUid);
final actualRemoteRideId = await db.ensureCloudRideId(rideId);
await fakeFirestore
.collection('users')
.doc(testUid)
.collection('rides')
.doc(actualRemoteRideId)
.set(createFirestoreRideData(rideId: actualRemoteRideId));
var uploadAttempted = false;
final mockClient = http_testing.MockClient((request) async {
if (request.url.path == '/r2/verify') {
return http.Response('Rate limit exceeded', 429);
}
if (request.url.path == '/r2/upload') {
uploadAttempted = true;
return http.Response('{}', 200);
}
return http.Response('{}', 404);
});
final r2Service = R2UploadService(client: mockClient);
final provider = FirebaseSyncProvider(
db: db,
authService: authService,
firestore: fakeFirestore,
r2UploadService: r2Service,
);
final result = await provider.uploadRide(
SyncItem(
localRideId: rideId,
providerId: 'firebase',
createdAt: DateTime.now(),
),
);
expect(result.success, isFalse);
expect(result.isRetryable, isTrue);
expect(result.errorMessage, contains('temporarily unavailable'));
expect(uploadAttempted, isFalse, reason: 'Must NOT upload bundle when verification is rate limited');
},
);
test(
'17. uploadRide Protection: Returns authRequired when cloud verification is unauthorized',
() async {
final rideId = await db.startRide(
startTime: DateTime.utc(2026, 10, 12, 10, 0, 0),
);
await db.endRide(rideId, const RidesCompanion());
await db.claimRideForCloudOwner(rideId, testUid);
final actualRemoteRideId = await db.ensureCloudRideId(rideId);
await fakeFirestore
.collection('users')
.doc(testUid)
.collection('rides')
.doc(actualRemoteRideId)
.set(createFirestoreRideData(ownerId: 'different_owner_uid', rideId: actualRemoteRideId));
var uploadAttempted = false;
final mockClient = http_testing.MockClient((request) async {
if (request.url.path == '/r2/upload') {
uploadAttempted = true;
}
return http.Response('{}', 200);
});
final r2Service = R2UploadService(client: mockClient);
final provider = FirebaseSyncProvider(
db: db,
authService: authService,
firestore: fakeFirestore,
r2UploadService: r2Service,
);
final result = await provider.uploadRide(
SyncItem(
localRideId: rideId,
providerId: 'firebase',
createdAt: DateTime.now(),
),
);
expect(result.success, isFalse);
expect(result.requiresAuth, isTrue);
expect(uploadAttempted, isFalse);
},
);
});
}

157
test/completed_ride_sync_recovery_test.dart
View File

@ -632,5 +632,162 @@ void main() {
expect(fakeProvider.uploadedItems.isEmpty, isTrue);
},
);
test(
'13. In-Flight Cancellation Protection: cancelSync preserves cancelled status even after upload completes',
() async {
authService.setAuthenticatedUser('rider_alice', 'alice@example.com');
final rideId = await db.startRide(
startTime: DateTime.utc(2026, 10, 12, 10, 0, 0),
);
await db.endRide(rideId, const RidesCompanion());
await db.claimRideForCloudOwner(rideId, 'rider_alice');
final outbox = SyncOutbox(prefs);
await outbox.update(
SyncItem(
localRideId: rideId,
providerId: 'firebase',
status: SyncStatus.queued,
createdAt: DateTime.now(),
),
);
// Simulate provider with 50ms upload latency
fakeProvider.delayMs = 50;
final manager = SyncManager(
outbox,
initialProviders: [fakeProvider],
defaultProviderId: 'firebase',
db: db,
authService: authService,
autoStart: false,
);
// Start processing queue
final processFuture = manager.processQueue();
// While upload is in flight, user cancels sync
await Future<void>.delayed(const Duration(milliseconds: 10));
await manager.cancelSync(rideId, 'firebase');
await processFuture;
final item = outbox.getItem(rideId, 'firebase');
expect(item?.status, equals(SyncStatus.cancelled),
reason: 'Item must remain cancelled and not get overwritten with synced');
},
);
test(
'14. Default Active Provider: Selects defaultProviderId or first initial provider, defaulting to fake for bare test manager',
() async {
final outbox = SyncOutbox(prefs);
final managerExplicit = SyncManager(
outbox,
defaultProviderId: 'firebase',
autoStart: false,
);
expect(managerExplicit.activeProviderId, equals('firebase'));
final customProvider = FakeSyncProvider(providerId: 'custom_gateway');
final managerWithProviders = SyncManager(
outbox,
initialProviders: [customProvider],
autoStart: false,
);
expect(managerWithProviders.activeProviderId, equals('custom_gateway'));
final managerNoProviders = SyncManager(
outbox,
autoStart: false,
);
expect(managerNoProviders.activeProviderId, equals('fake'));
},
);
test(
'15. Immediate Enqueue Ownership Protection: enqueueRide cancels foreign-owned rides immediately',
() async {
authService.setAuthenticatedUser('rider_alice', 'alice@example.com');
final rideId = await db
.into(db.rides)
.insert(
RidesCompanion.insert(
cloudRideId: const drift.Value('ride_bob_999'),
cloudOwnerUid: const drift.Value('rider_bob'),
startTime: DateTime.utc(2026, 10, 12, 11, 0, 0),
endTime: drift.Value(DateTime.utc(2026, 10, 12, 12, 0, 0)),
),
);
final outbox = SyncOutbox(prefs);
final manager = SyncManager(
outbox,
initialProviders: [fakeProvider],
defaultProviderId: 'firebase',
db: db,
authService: authService,
autoStart: false,
);
final item = await manager.enqueueRide(rideId, autoProcess: false);
expect(item.status, equals(SyncStatus.cancelled));
expect(item.lastError, contains('belongs to another account'));
expect(outbox.getItem(rideId, 'firebase')?.status, equals(SyncStatus.cancelled));
},
);
test(
'16. Failed Cloud Verification followed by Successful Retry: Recovers to synced without re-upload',
() async {
authService.setAuthenticatedUser('rider_alice', 'alice@example.com');
final rideId = await db.startRide(
startTime: DateTime.utc(2026, 10, 12, 13, 0, 0),
);
await db.endRide(rideId, const RidesCompanion());
final remoteRideId = await db.ensureCloudRideId(rideId);
// Step 1: Initial reconciliation fails with temporary verification error (e.g. HTTP 503)
fakeProvider.confirmationResults[remoteRideId] =
CloudConfirmationResult.temporarilyUnverifiable(
remoteRideId: remoteRideId,
message: 'HTTP 503 gateway timeout',
);
final outbox = SyncOutbox(prefs);
final manager = SyncManager(
outbox,
initialProviders: [fakeProvider],
defaultProviderId: 'firebase',
db: db,
authService: authService,
autoStart: false,
);
final report1 = await manager.reconcileCompletedRides();
expect(report1.unverifiablePendingCount, equals(1));
expect(outbox.getItem(rideId, 'firebase')?.status, equals(SyncStatus.retryWait));
// Step 2: Backend recovers and verifies the bundle in R2
fakeProvider.confirmationResults[remoteRideId] =
CloudConfirmationResult.confirmed(remoteRideId: remoteRideId);
fakeProvider.confirmedRemoteRideIds.add(remoteRideId);
// Process queue on retry
await manager.retryAllPending();
expect(
outbox.getItem(rideId, 'firebase')?.status,
equals(SyncStatus.synced),
);
expect(fakeProvider.uploadedItems.isEmpty, isTrue,
reason: 'Must confirm and mark synced without re-uploading the bundle');
},
);
});
}