From 5e0db4c19172ce0ef2e68378ee342e8985b0ff96 Mon Sep 17 00:00:00 2001 From: Developer Date: Tue, 13 Oct 2026 22:05:25 -0400 Subject: [PATCH] fix(sync): harden cloud confirmation states, R2 edge 404 handling, and in-flight cancellation --- cloudflare/motorsignal-r2-worker/src/index.ts | 2 +- lib/services/sync/firebase_sync_provider.dart | 42 ++-- lib/services/sync/r2_upload_service.dart | 33 ++- lib/services/sync/sync_manager.dart | 61 +++++- lib/services/sync/sync_provider.dart | 14 ++ test/cloud_confirmation_contract_test.dart | 196 ++++++++++++++++++ test/completed_ride_sync_recovery_test.dart | 157 ++++++++++++++ 7 files changed, 483 insertions(+), 22 deletions(-) diff --git a/cloudflare/motorsignal-r2-worker/src/index.ts b/cloudflare/motorsignal-r2-worker/src/index.ts index e79051b..1523f28 100644 --- a/cloudflare/motorsignal-r2-worker/src/index.ts +++ b/cloudflare/motorsignal-r2-worker/src/index.ts @@ -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'); diff --git a/lib/services/sync/firebase_sync_provider.dart b/lib/services/sync/firebase_sync_provider.dart index 0a82075..1ceb393 100644 --- a/lib/services/sync/firebase_sync_provider.dart +++ b/lib/services/sync/firebase_sync_provider.dart @@ -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(); diff --git a/lib/services/sync/r2_upload_service.dart b/lib/services/sync/r2_upload_service.dart index 21f7131..83f7f0c 100644 --- a/lib/services/sync/r2_upload_service.dart +++ b/lib/services/sync/r2_upload_service.dart @@ -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, diff --git a/lib/services/sync/sync_manager.dart b/lib/services/sync/sync_manager.dart index c228d74..989dea9 100644 --- a/lib/services/sync/sync_manager.dart +++ b/lib/services/sync/sync_manager.dart @@ -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( diff --git a/lib/services/sync/sync_provider.dart b/lib/services/sync/sync_provider.dart index 3d84bc9..f36599f 100644 --- a/lib/services/sync/sync_provider.dart +++ b/lib/services/sync/sync_provider.dart @@ -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( diff --git a/test/cloud_confirmation_contract_test.dart b/test/cloud_confirmation_contract_test.dart index 71dcadf..0f69f19 100644 --- a/test/cloud_confirmation_contract_test.dart +++ b/test/cloud_confirmation_contract_test.dart @@ -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( + '404 Not Found

404 Not Found

cloudflare', + 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); + }, + ); }); } diff --git a/test/completed_ride_sync_recovery_test.dart b/test/completed_ride_sync_recovery_test.dart index 4d0395a..5cbce4e 100644 --- a/test/completed_ride_sync_recovery_test.dart +++ b/test/completed_ride_sync_recovery_test.dart @@ -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.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'); + }, + ); }); }