diff --git a/packages/drift_sync_core/CHANGELOG.md b/packages/drift_sync_core/CHANGELOG.md index 8940445..de9d600 100644 --- a/packages/drift_sync_core/CHANGELOG.md +++ b/packages/drift_sync_core/CHANGELOG.md @@ -1,5 +1,41 @@ # Changelog +## 0.3.3 + +### Fixed + +* A queued payload that fails to unmarshal now throws `UnmarshalException`, + classified permanent — quarantined and surfaced on the first attempt + instead of burning the unknown-failure retry budget. +* A change deferred by `shouldPersistRemote` now records a transient + `DependencyPendingException` on the row, so sync history can show why it + is waiting instead of leaving it silently pending. + +## 0.3.2 + +### Improved + +* `upload_local_change` crash reports now include the attempt number and + a truncated excerpt of the queued payload, so serialization failures + (e.g. null type casts during unmarshal) are diagnosable from the crash + report alone. + +## 0.3.1 + +### Fixed + +* `SyncEntityRepository.put()`/`post()`/`delete()` now enqueue the pending + local change *before* attempting the remote call (outbox ordering). + Previously any non-`UnavailableException` failure — e.g. a server + validation rejection — escaped before the pending change was written, + permanently orphaning the record: saved locally but invisible to the sync + loop, never retried, never quarantined, absent from sync history. Remote + failures are now recorded on the queued change, handing retry/backoff/ + quarantine to the synchronizer, and these methods no longer rethrow. +* `delete()` of a never-synced entity (no server id) no longer calls + `deleteRemote`; the change concludes immediately, also removing any + queued put for the same entity. + ## 0.2.0 ### Behavioral change diff --git a/packages/drift_sync_core/lib/src/drift_synchronizer.dart b/packages/drift_sync_core/lib/src/drift_synchronizer.dart index 872eae3..9dfadd0 100644 --- a/packages/drift_sync_core/lib/src/drift_synchronizer.dart +++ b/packages/drift_sync_core/lib/src/drift_synchronizer.dart @@ -1,4 +1,5 @@ import 'dart:async'; +import 'dart:convert'; import 'package:drift_sync_core/drift_sync_core.dart'; import 'package:meta/meta.dart'; @@ -81,6 +82,18 @@ abstract class DriftSynchronizer { final SyncLogger _logger; final SyncCrashReporter? _crashReporter; + static String _payloadExcerpt(Map data, + {int maxLength = 2000}) { + try { + final encoded = jsonEncode(data); + return encoded.length > maxLength + ? '${encoded.substring(0, maxLength)}…' + : encoded; + } catch (_) { + return data.toString(); + } + } + /// Logs and routes the error to the crash reporter unconditionally. void _reportError( Object error, @@ -208,6 +221,11 @@ abstract class DriftSynchronizer { 'is_deleted': localChange.deleted.toString(), 'failure_class': failureClass.name, 'quarantined': quarantine.toString(), + 'attempt': '${localChange.attemptCount + 1}', + // The payload is essential for diagnosing unmarshal/serialization + // failures, where the exception alone (e.g. a null type cast) + // says nothing about which field was at fault. + 'data': _payloadExcerpt(localChange.data), }, ); await appDatabase.concludeLocalChange( @@ -226,7 +244,13 @@ abstract class DriftSynchronizer { PendingLocalChange localChange, SyncTypeHandler handler, ) async { - final entity = await handler.unmarshal(localChange.data); + final dynamic entity; + try { + entity = await handler.unmarshal(localChange.data); + } catch (e) { + // An immutable queued payload that fails to parse can never succeed. + throw UnmarshalException(localChange.entityType, e); + } if (localChange.deleted) { // For delete operations, try to use server ID if available final serverId = handler.getServerId(entity); @@ -240,10 +264,12 @@ abstract class DriftSynchronizer { // For put operations if (!await handler.shouldPersistRemote(entity)) { - _logger.info( - 'Skipping sync for ${handler.entityType}:${handler.getClientId(entity)} - dependencies not ready', + // Recorded on the row (transient) so sync history can show why the + // change is waiting instead of leaving it silently pending. + throw DependencyPendingException( + '${handler.entityType}:${handler.getClientId(entity)} ' + 'has unsynced dependencies', ); - return; } final updated = await handler.putRemote(entity); diff --git a/packages/drift_sync_core/lib/src/exceptions/dependency_pending_exception.dart b/packages/drift_sync_core/lib/src/exceptions/dependency_pending_exception.dart new file mode 100644 index 0000000..059eae1 --- /dev/null +++ b/packages/drift_sync_core/lib/src/exceptions/dependency_pending_exception.dart @@ -0,0 +1,10 @@ +/// Upload deferred because a dependency has not synced yet (e.g. a +/// transfer waiting for its legs' server ids). Transient by definition. +class DependencyPendingException implements Exception { + DependencyPendingException(this.message); + + final String message; + + @override + String toString() => 'Waiting for dependencies: $message'; +} diff --git a/packages/drift_sync_core/lib/src/exceptions/exceptions.dart b/packages/drift_sync_core/lib/src/exceptions/exceptions.dart index 6299ac4..a21735a 100644 --- a/packages/drift_sync_core/lib/src/exceptions/exceptions.dart +++ b/packages/drift_sync_core/lib/src/exceptions/exceptions.dart @@ -3,3 +3,5 @@ export 'unavailable_exception.dart'; export 'cancel_exception.dart'; export 'notfound_exception.dart'; export 'invalid_state_exception.dart'; +export 'unmarshal_exception.dart'; +export 'dependency_pending_exception.dart'; diff --git a/packages/drift_sync_core/lib/src/exceptions/unmarshal_exception.dart b/packages/drift_sync_core/lib/src/exceptions/unmarshal_exception.dart new file mode 100644 index 0000000..10f7272 --- /dev/null +++ b/packages/drift_sync_core/lib/src/exceptions/unmarshal_exception.dart @@ -0,0 +1,11 @@ +/// A queued payload failed to deserialize. The payload is immutable, so +/// this can never succeed on retry; classified permanent. +class UnmarshalException implements Exception { + UnmarshalException(this.entityType, this.cause); + + final String entityType; + final Object cause; + + @override + String toString() => 'UnmarshalException($entityType): $cause'; +} diff --git a/packages/drift_sync_core/lib/src/failure_class.dart b/packages/drift_sync_core/lib/src/failure_class.dart index 6c0702a..6c67d5a 100644 --- a/packages/drift_sync_core/lib/src/failure_class.dart +++ b/packages/drift_sync_core/lib/src/failure_class.dart @@ -21,5 +21,7 @@ typedef FailureClassifier = FailureClass Function(Object error); /// classifier (e.g. `restFailureClassifier`) to the synchronizer. FailureClass defaultFailureClassifier(Object error) { if (error is UnavailableException) return FailureClass.transient; + if (error is DependencyPendingException) return FailureClass.transient; + if (error is UnmarshalException) return FailureClass.permanent; return FailureClass.unknown; } diff --git a/packages/drift_sync_core/lib/src/sync_entity_repository.dart b/packages/drift_sync_core/lib/src/sync_entity_repository.dart index 8b365ab..b79d332 100644 --- a/packages/drift_sync_core/lib/src/sync_entity_repository.dart +++ b/packages/drift_sync_core/lib/src/sync_entity_repository.dart @@ -39,60 +39,78 @@ abstract class SyncEntityRepository put(TEntity entity) async { - // Check authorization before attempting remote operations + final pending = _pendingPut(entity); + await db.transaction(() => db.insertLocalChange(pending)); + final canSync = await requestAuthorizationService.canSync(); final serverId = syncHandler.getServerId(entity); - final remoteCreated = - (canSync && serverId != null) ? await putRemote(entity) : null; - final created = remoteCreated ?? entity; - final ds = - remoteCreated == null ? DataDestination.local : DataDestination.both; - - await _handleLocalStorage(created, remoteCreated); - return (created, ds); + + TEntity? remoteCreated; + if (canSync && serverId != null) { + try { + remoteCreated = await putRemote(entity); + } catch (error) { + await db.concludeLocalChange(pending, error: error); + return (entity, DataDestination.local); + } + } + + if (remoteCreated == null) { + return (entity, DataDestination.local); + } + await _concludePutSuccess(pending, remoteCreated); + return (remoteCreated, DataDestination.both); } Future<(TEntity, DataDestination)> post(TEntity entity) async { - // Check authorization before attempting remote operations - final canSync = await requestAuthorizationService.canSync(); - final remoteCreated = canSync ? await putRemote(entity) : null; - final created = remoteCreated ?? entity; - final ds = - remoteCreated == null ? DataDestination.local : DataDestination.both; + final pending = _pendingPut(entity); + await db.transaction(() => db.insertLocalChange(pending)); - await _handleLocalStorage(created, remoteCreated); - return (created, ds); - } + final canSync = await requestAuthorizationService.canSync(); - Future _handleLocalStorage( - TEntity entity, TEntity? remoteCreated) async { - await db.transaction(() async { - if (remoteCreated == null) { - await _createPendingChange(entity); - } else { - await _concludeEntityChanges(entity); + TEntity? remoteCreated; + if (canSync) { + try { + remoteCreated = await putRemote(entity); + } catch (error) { + await db.concludeLocalChange(pending, error: error); + return (entity, DataDestination.local); } - }); + } + + if (remoteCreated == null) { + return (entity, DataDestination.local); + } + await _concludePutSuccess(pending, remoteCreated); + return (remoteCreated, DataDestination.both); } - Future _createPendingChange(TEntity entity) async { - final localChange = PendingLocalChange.put( + PendingLocalChange _pendingPut(TEntity entity) { + return PendingLocalChange.put( entityData: syncHandler.marshal(entity), entityType: syncHandler.entityType, entityId: syncHandler.getClientId(entity), entityRev: syncHandler.getRev(entity), ); - await db.insertLocalChange(localChange); } - Future _concludeEntityChanges(TEntity entity) async { - await db.concludeEntityLocalChanges( - syncHandler.entityType, - syncHandler.getServerId(entity), - Operation.put, - ); - await syncHandler.upsertLocal(entity); + Future _concludePutSuccess( + PendingLocalChange pending, TEntity remoteCreated) async { + await db.transaction(() async { + await db.concludeLocalChange(pending, persistedToRemote: true); + await db.concludeEntityLocalChanges( + syncHandler.entityType, + syncHandler.getServerId(remoteCreated), + Operation.put, + ); + await syncHandler.upsertLocal(remoteCreated); + }); } @protected @@ -106,38 +124,49 @@ abstract class SyncEntityRepository delete(TEntity entity) async { - // Check authorization before attempting remote operations - final canSync = await requestAuthorizationService.canSync(); - final synced = canSync ? await deleteRemote(entity) : false; - final ds = synced ? DataDestination.both : DataDestination.local; - - await _handleDeleteStorage(entity, synced); - return ds; - } - - Future _handleDeleteStorage(TEntity entity, bool synced) async { - await db.transaction(() async { - await syncHandler.deleteLocal(entity); - if (!synced) { - await _createDeletePendingChange(entity); - } else { - await _concludeDeleteChanges(entity); - } - }); - } - - Future _createDeletePendingChange(TEntity entity) async { - final localChange = PendingLocalChange.delete( + final pending = PendingLocalChange.delete( entityType: syncHandler.entityType, data: syncHandler.marshal(entity), entityId: syncHandler.getClientId(entity), entityRev: syncHandler.getRev(entity), ); - await db.insertLocalChange(localChange); + await db.transaction(() async { + await syncHandler.deleteLocal(entity); + await db.insertLocalChange(pending); + }); + + if (syncHandler.getServerId(entity) == null) { + await db.concludeLocalChange(pending, persistedToRemote: true); + return DataDestination.local; + } + + final canSync = await requestAuthorizationService.canSync(); + bool synced = false; + if (canSync) { + try { + synced = await deleteRemote(entity); + } catch (error) { + await db.concludeLocalChange(pending, error: error); + return DataDestination.local; + } + } + + if (!synced) { + return DataDestination.local; + } + await db.transaction(() async { + await db.concludeLocalChange(pending, persistedToRemote: true); + await _concludeDeleteChanges(entity); + }); + return DataDestination.both; } Future _concludeDeleteChanges(TEntity entity) async { @@ -156,6 +185,7 @@ abstract class SyncEntityRepository { // Behavior queues — each call dequeues one entry; empty = use default. final List putRemoteThrows = []; + final List deleteRemoteThrows = []; final List getAllRemoteThrows = []; final List assignClientIdThrows = []; final List assignedIds = []; @@ -266,6 +267,7 @@ class FakeHandler extends SyncTypeHandler { @override Future deleteRemote(TestEntity entity) async { deletedRemote.add(entity); + if (deleteRemoteThrows.isNotEmpty) throw deleteRemoteThrows.removeAt(0); if (entity.id != null) remoteItems.remove(entity.id); } diff --git a/packages/drift_sync_core/test/orchestrator/upload_test.dart b/packages/drift_sync_core/test/orchestrator/upload_test.dart index 95411f1..c6ee0b4 100644 --- a/packages/drift_sync_core/test/orchestrator/upload_test.dart +++ b/packages/drift_sync_core/test/orchestrator/upload_test.dart @@ -120,7 +120,9 @@ void main() { reason: 'pending change still concluded (nothing to retry)'); }); - test('skips put when shouldPersistRemote returns false', () async { + test( + 'records a transient dependency-pending error when ' + 'shouldPersistRemote refuses', () async { wallet.shouldPersistRemoteResult = false; await db.insertLocalChange(_put( entityType: 'wallet', @@ -130,8 +132,29 @@ void main() { await sync.uploadLocalChanges(); expect(wallet.putRemoteCalls, isEmpty); - expect(db.allPending, hasLength(1), - reason: 'pending change preserved for next cycle'); + final row = db.allPending.single; + expect(row.error, contains('dependencies')); + expect(row.quarantinedAt, isNull, + reason: 'transient — retried once dependencies sync'); + expect(row.attemptCount, 1); + }); + + test('quarantines an unparseable queued payload on the first attempt', + () async { + await db.insertLocalChange(PendingLocalChange.put( + entityType: 'wallet', + entityId: 'w-bad', + entityRev: '1', + entityData: const {'clientId': 123}, + )); + + await sync.uploadLocalChanges(); + + expect(wallet.putRemoteCalls, isEmpty); + final row = db.allPending.single; + expect(row.error, contains('UnmarshalException')); + expect(row.quarantinedAt, isNotNull, + reason: 'an immutable payload that cannot parse never will'); }); test('returns false on UnavailableException, leaves pending intact', diff --git a/packages/drift_sync_core/test/sync_entity_repository_test.dart b/packages/drift_sync_core/test/sync_entity_repository_test.dart new file mode 100644 index 0000000..f83e8f9 --- /dev/null +++ b/packages/drift_sync_core/test/sync_entity_repository_test.dart @@ -0,0 +1,185 @@ +import 'package:drift_sync_core/drift_sync_core.dart'; +import 'package:test/test.dart'; + +import '_fakes.dart'; + +class TestRepo extends SyncEntityRepository { + const TestRepo({ + required super.syncHandler, + required super.db, + required super.requestAuthorizationService, + }); +} + +void main() { + late FakeSynchronizerDb db; + late FakeHandler handler; + late FakeAuthService auth; + late TestRepo repo; + + setUp(() { + db = FakeSynchronizerDb(); + handler = FakeHandler(entityType: 'test'); + auth = FakeAuthService(); + repo = TestRepo( + syncHandler: handler, + db: db, + requestAuthorizationService: auth, + ); + }); + + PendingLocalChange? pendingFor(String clientId) { + for (final c in db.allPending) { + if (c.entityId == clientId) return c; + } + return null; + } + + group('post', () { + test('successful remote create leaves no pending change', () async { + final (created, ds) = await repo.post(const TestEntity(clientId: 'a')); + + expect(ds, DataDestination.both); + expect(created.id, isNotNull); + expect(db.allPending, isEmpty); + expect(handler.localItems['a']?.id, created.id); + }); + + test('server rejection keeps the pending change with the error', () async { + handler.putRemoteThrows.add(Exception('422 validation failed')); + + final (created, ds) = await repo.post(const TestEntity(clientId: 'a')); + + expect(ds, DataDestination.local); + expect(created.id, isNull); + final pending = pendingFor('a'); + expect(pending, isNotNull); + expect(pending!.error, contains('422')); + expect(pending.attemptCount, 1); + expect(pending.deleted, isFalse); + }); + + test('server rejection does not rethrow', () async { + handler.putRemoteThrows.add(StateError('boom')); + + await expectLater( + repo.post(const TestEntity(clientId: 'a')), + completes, + ); + }); + + test('network unavailable enqueues without an error', () async { + handler.putRemoteThrows.add(UnavailableException()); + + final (_, ds) = await repo.post(const TestEntity(clientId: 'a')); + + expect(ds, DataDestination.local); + final pending = pendingFor('a'); + expect(pending, isNotNull); + expect(pending!.error, isNull); + expect(pending.attemptCount, 0); + }); + + test('unauthenticated enqueues without attempting remote', () async { + auth.authorized = false; + + final (_, ds) = await repo.post(const TestEntity(clientId: 'a')); + + expect(ds, DataDestination.local); + expect(handler.putRemoteCalls, isEmpty); + expect(pendingFor('a'), isNotNull); + }); + }); + + group('put', () { + test('successful remote update leaves no pending change', () async { + final (_, ds) = + await repo.put(const TestEntity(clientId: 'a', id: 7)); + + expect(ds, DataDestination.both); + expect(db.allPending, isEmpty); + }); + + test('server rejection keeps the pending change with the error', () async { + handler.putRemoteThrows.add(Exception('500 server error')); + + final (_, ds) = await repo.put(const TestEntity(clientId: 'a', id: 7)); + + expect(ds, DataDestination.local); + final pending = pendingFor('a'); + expect(pending, isNotNull); + expect(pending!.error, contains('500')); + }); + + test('entity without server id enqueues without attempting remote', + () async { + final (_, ds) = await repo.put(const TestEntity(clientId: 'a')); + + expect(ds, DataDestination.local); + expect(handler.putRemoteCalls, isEmpty); + expect(pendingFor('a'), isNotNull); + }); + }); + + group('delete', () { + test('successful remote delete leaves no pending change', () async { + handler.localItems['a'] = const TestEntity(clientId: 'a', id: 7); + + final ds = await repo.delete(const TestEntity(clientId: 'a', id: 7)); + + expect(ds, DataDestination.both); + expect(db.allPending, isEmpty); + expect(handler.localItems, isEmpty); + expect(handler.deletedRemote, hasLength(1)); + }); + + test('server rejection keeps the delete change with the error', () async { + handler.deleteRemoteThrows.add(Exception('409 conflict')); + + final ds = await repo.delete(const TestEntity(clientId: 'a', id: 7)); + + expect(ds, DataDestination.local); + final pending = pendingFor('a'); + expect(pending, isNotNull); + expect(pending!.deleted, isTrue); + expect(pending.error, contains('409')); + }); + + test('never-synced entity concludes immediately without remote call', + () async { + handler.localItems['a'] = const TestEntity(clientId: 'a'); + + final ds = await repo.delete(const TestEntity(clientId: 'a')); + + expect(ds, DataDestination.local); + expect(db.allPending, isEmpty); + expect(handler.deletedRemote, isEmpty); + expect(handler.localItems, isEmpty); + }); + + test('delete change replaces a queued put for the same entity', () async { + handler.putRemoteThrows.add(Exception('422')); + await repo.post(const TestEntity(clientId: 'a')); + expect(pendingFor('a')!.deleted, isFalse); + + await repo.delete(const TestEntity(clientId: 'a')); + + expect(db.allPending, isEmpty, + reason: 'never-synced delete concludes and removes the queued put'); + }); + + test('unauthenticated delete of synced entity stays queued', () async { + auth.authorized = false; + + final ds = await repo.delete(const TestEntity(clientId: 'a', id: 7)); + + expect(ds, DataDestination.local); + final pending = pendingFor('a'); + expect(pending, isNotNull); + expect(pending!.deleted, isTrue); + expect(pending.error, isNull); + expect(handler.deletedRemote, isEmpty); + }); + }); +}