Outbox ordering for entity repositories and payload-carrying upload crash reports

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

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<TAppDatabase extends SynchronizerDb> {
   final SyncLogger _logger;
   final SyncCrashReporter? _crashReporter;
 
+  static String _payloadExcerpt(Map<String, dynamic> 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<TAppDatabase extends SynchronizerDb> {
             '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<TAppDatabase extends SynchronizerDb> {
     PendingLocalChange localChange,
     SyncTypeHandler<dynamic, dynamic, dynamic> 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<TAppDatabase extends SynchronizerDb> {
 
     // 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);

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';
+}

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';

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';
+}

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;
 }

packages/drift_sync_core/lib/src/sync_entity_repository.dart

@@ -39,60 +39,78 @@ abstract class SyncEntityRepository<TAppDatabase extends SynchronizerDb,
     }
   }
 
+  /// Outbox ordering: the pending change is enqueued BEFORE the remote
+  /// attempt, so no failure mode (server rejection, crash mid-request) can
+  /// leave a local record invisible to the sync loop. Success removes the
+  /// queued row; failure records the error on it, handing retry/backoff/
+  /// quarantine to the synchronizer.
   Future<(TEntity, DataDestination)> 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<void> _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<void> _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<void> _concludeEntityChanges(TEntity entity) async {
-    await db.concludeEntityLocalChanges(
-      syncHandler.entityType,
-      syncHandler.getServerId(entity),
-      Operation.put,
-    );
-    await syncHandler.upsertLocal(entity);
+  Future<void> _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<TAppDatabase extends SynchronizerDb,
       // Graceful fallback to local storage when network is unavailable
       return null;
     }
-    // All other exceptions are logged by the adapter and will propagate
+    // Other exceptions propagate to put/post, which record them on the
+    // already-enqueued pending change.
   }
 
+  /// Same outbox ordering as [put]/[post]: the delete change is enqueued
+  /// (replacing any queued put for the same entity) before the remote
+  /// attempt. A record the server never knew about concludes immediately.
   Future<DataDestination> 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<void> _handleDeleteStorage(TEntity entity, bool synced) async {
-    await db.transaction(() async {
-      await syncHandler.deleteLocal(entity);
-      if (!synced) {
-        await _createDeletePendingChange(entity);
-      } else {
-        await _concludeDeleteChanges(entity);
-      }
-    });
-  }
-
-  Future<void> _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<void> _concludeDeleteChanges(TEntity entity) async {
@@ -156,6 +185,7 @@ abstract class SyncEntityRepository<TAppDatabase extends SynchronizerDb,
       // Graceful fallback to local storage when network is unavailable
       return false;
     }
-    // All other exceptions are logged by the adapter and will propagate
+    // Other exceptions propagate to delete, which records them on the
+    // already-enqueued pending change.
   }
 }

packages/drift_sync_core/pubspec.yaml

@@ -1,6 +1,6 @@
 name: drift_sync_core
 description: Offline-first synchronization engine for Drift databases. Three-phase reconciliation, typed outcomes, pluggable transport.
-version: 0.2.0
+version: 0.3.3
 homepage: https://github.com/whilesmartflutter/drift_sync
 repository: https://github.com/whilesmartflutter/drift_sync
 issue_tracker: https://github.com/whilesmartflutter/drift_sync/issues

packages/drift_sync_core/test/_fakes.dart

@@ -169,6 +169,7 @@ class FakeHandler extends SyncTypeHandler<TestEntity, String, int> {
 
   // Behavior queues — each call dequeues one entry; empty = use default.
   final List<Object> putRemoteThrows = [];
+  final List<Object> deleteRemoteThrows = [];
   final List<Object> getAllRemoteThrows = [];
   final List<Object> assignClientIdThrows = [];
   final List<TestEntity> assignedIds = [];
@@ -266,6 +267,7 @@ class FakeHandler extends SyncTypeHandler<TestEntity, String, int> {
   @override
   Future<void> deleteRemote(TestEntity entity) async {
     deletedRemote.add(entity);
+    if (deleteRemoteThrows.isNotEmpty) throw deleteRemoteThrows.removeAt(0);
     if (entity.id != null) remoteItems.remove(entity.id);
   }