fix: Fix missing entries when syncing multiple nodes (#58)

packages/serverpod_offline_sync_client/lib/src/crdt/sync.dart

@@ -7,6 +7,7 @@ import 'package:uuid/uuid.dart';
 import '../database/database.dart';
 import '../managers/user.dart';
 import '../protocol/protocol.dart';
+import '../utils/case_when.dart' show Case;
 import 'exceptions.dart';
 import 'merge.dart';
 
@@ -122,44 +123,50 @@ class CrdtSync {
         '${uuid.v5(Namespace.oid.value, canonicalSignature)}';
   }
 
-  /// Streams local CRDT changes since the last sync checkpoint with [otherNodeId].
+  /// Streams local CRDT changes that a peer node has not seen yet.
   ///
   /// Changes are emitted in insert, update, then delete order. Domain row and
   /// column payloads are resolved incrementally as each change is yielded.
   ///
-  /// When [sinceHlc] is provided, it is used instead of the stored checkpoint.
+  /// All changes for nodes that are not present in the [nodeCheckpoints] list
+  /// are collected and emitted. Passing an empty list will collect all changes.
   Stream<CrdtMergeChange> collectPendingChanges(
     DatabaseSession session, {
     required UuidValue userId,
-    required UuidValue otherNodeId,
+    required UuidValue peerNodeId,
-    Hlc? sinceHlc,
+    required List<Hlc> nodeCheckpoints,
   }) async* {
     final crdtUser = await CrdtUserManager(session).getOrCreate(userId);
-    final syncCheckpoint =
+    yield* _streamInserts(session, crdtUser, nodeCheckpoints);
-        sinceHlc ??
+    yield* _streamUpdates(session, crdtUser, nodeCheckpoints);
-        await _syncCheckpointForNode(
+    yield* _streamDeletes(session, crdtUser, nodeCheckpoints);
-          session,
-          crdtUser,
-          otherNodeId,
-        );
-
-    yield* _streamInserts(session, crdtUser, syncCheckpoint);
-    yield* _streamUpdates(session, crdtUser, syncCheckpoint);
-    yield* _streamDeletes(session, crdtUser, syncCheckpoint);
   }
 
   /// Creates the [CrdtSyncSinceHlc] checkpoint for the post-connect handshake.
   ///
-  /// [CrdtSyncSinceHlc.lastSyncFromPeer] reflects the latest change this node
+  /// [CrdtSyncSinceHlc.nodeCheckpoints] reflects the latest change this node
-  /// has received from [peerNodeId], tagged with the peer's node id.
+  /// has received from each known node, tagged with the source node id.
   Future<CrdtSyncSinceHlc> createSyncSinceHlc(
     DatabaseSession session, {
     required UuidValue userId,
     required UuidValue peerNodeId,
   }) async {
     final crdtUser = await CrdtUserManager(session).getOrCreate(userId);
-    final lastSeenHlc = await _syncCheckpointForNode(session, crdtUser, peerNodeId);
+
-    return CrdtSyncSinceHlc(lastSyncFromPeer: lastSeenHlc);
+    final nodes = await CrdtNode.db.find(
+      session,
+      where: (t) =>
+          t.userId.equals(crdtUser.id) &
+          t.uuidNodeId.notEquals(crdtUser.currentNode!.uuidNodeId),
+    );
+
+    return CrdtSyncSinceHlc(
+      nodeCheckpoints: [
+        // The local node is always included to avoid collecting its own changes.
+        Hlc.now(crdtUser.currentNode!.uuidNodeId),
+        for (final node in nodes) node.lastReceivedHlc ?? Hlc.zero(node.uuidNodeId),
+      ],
+    );
   }
 
   /// Merges a remote [mergeSet] and records the sync checkpoint for [otherNodeId].
@@ -190,13 +197,13 @@ class CrdtSync {
     DatabaseSession session, {
     required UuidValue userId,
     required UuidValue peerNodeId,
-    required Hlc sinceHlc,
+    required List<Hlc> nodeCheckpoints,
   }) async* {
     yield* collectPendingChanges(
           session,
           userId: userId,
-          otherNodeId: peerNodeId,
+          peerNodeId: peerNodeId,
-          sinceHlc: sinceHlc,
+          nodeCheckpoints: nodeCheckpoints,
         )
         .chunked(_syncBatchSize)
         .map(
@@ -209,9 +216,9 @@ class CrdtSync {
   ///
   /// Both peers send [CrdtSyncConnect], read the peer connect frame, then send
   /// and read [CrdtSyncSinceHlc] once. After that, both sides exchange framed
-  /// data batches using (1) the peer's authoritative since HLC for the first
+  /// data batches using the peer's per-node checkpoints. Sent changes advance
-  /// send, and (2) [CrdtSyncSinceHlc.lastSyncFromPeer] for subsequent sends
+  /// the corresponding in-memory checkpoints for subsequent sends within the
-  /// within the session.
+  /// session.
   ///
   /// If a batch fails to merge, the exception propagates and the stream closes.
   /// The next sync attempt resumes from the last persisted checkpoint on the
@@ -258,27 +265,26 @@ class CrdtSync {
       );
 
       final peerSinceHlc = await inboundIterator.moveAndThrowIfNot<CrdtSyncSinceHlc>();
-      if (peerSinceHlc.lastSyncFromPeer.nodeId != localNodeId) {
+      final nodeCheckpoints = {
-        throw CrdtSyncInvalidSinceHlcException(
+        for (final checkpoint in peerSinceHlc.nodeCheckpoints)
-          receivedNodeId: peerSinceHlc.lastSyncFromPeer.nodeId,
+          checkpoint.nodeId: checkpoint,
-          expectedNodeId: localNodeId,
+      };
-        );
-      }
-
-      var lastSentToPeer = peerSinceHlc.lastSyncFromPeer;
 
       while (true) {
         final pendingLocalChanges = collectPendingChanges(
           session,
           userId: userId,
-          otherNodeId: peerNodeId,
+          peerNodeId: peerNodeId,
-          sinceHlc: lastSentToPeer,
+          nodeCheckpoints: nodeCheckpoints.values.toList(),
         );
 
         var hasChanges = false;
         await for (final changes in pendingLocalChanges.chunked(_syncBatchSize)) {
           hasChanges = true;
-          lastSentToPeer = lastSentToPeer.maxBetween(changes.maxHlc);
+          for (final change in changes) {
+            final lastCheckpoint = nodeCheckpoints[change.uuidNodeId];
+            nodeCheckpoints[change.uuidNodeId] = change.hlc.maxBetween(lastCheckpoint);
+          }
           yield CrdtSyncMergeChunk(changes: changes);
         }
         if (hasChanges || once) {
@@ -302,7 +308,7 @@ class CrdtSync {
 
         if (hasChanges || inboundMergeSet.isNotEmpty) {
           await onMergeSuccess?.call(
-            lastSentToPeer.maxBetween(lastReceivedFromPeer),
+            nodeCheckpoints.values.max.maxBetween(lastReceivedFromPeer),
           );
         }
 
@@ -348,27 +354,14 @@ class CrdtSync {
     return crdtDb;
   }
 
-  Future<Hlc> _syncCheckpointForNode(
-    DatabaseSession session,
-    CrdtUser crdtUser,
-    UuidValue otherNodeId,
-  ) async {
-    final otherNode = await CrdtNode.db.findFirstRow(
-      session,
-      where: (t) => t.userId.equals(crdtUser.id) & t.uuidNodeId.equals(otherNodeId),
-    );
-
-    return otherNode?.lastReceivedHlc ?? Hlc.zero(otherNodeId);
-  }
-
   Stream<CrdtMergeInsert> _streamInserts(
     DatabaseSession session,
     CrdtUser crdtUser,
-    Hlc lastSyncHlc,
+    List<Hlc> nodeCheckpoints,
   ) async* {
     final rows = await CrdtDataRow.db.find(
       session,
-      where: (t) => _rowHlcAfterFilter(t, crdtUser, lastSyncHlc),
+      where: (t) => _rowHlcAfterFilter(t, crdtUser, nodeCheckpoints),
       include: CrdtDataRow.include(
         tbl: CrdtSchemaTable.include(),
         node: CrdtNode.include(),
@@ -406,11 +399,11 @@ class CrdtSync {
   Stream<CrdtMergeUpdate> _streamUpdates(
     DatabaseSession session,
     CrdtUser crdtUser,
-    Hlc lastSyncHlc,
+    List<Hlc> nodeCheckpoints,
   ) async* {
     final fields = await CrdtDataField.db.find(
       session,
-      where: (t) => _fieldHlcAfterFilter(t, crdtUser, lastSyncHlc),
+      where: (t) => _fieldHlcAfterFilter(t, crdtUser, nodeCheckpoints),
       include: CrdtDataField.include(
         row: CrdtDataRow.include(tbl: CrdtSchemaTable.include()),
         column: CrdtSchemaColumn.include(),
@@ -445,11 +438,11 @@ class CrdtSync {
   Stream<CrdtMergeDelete> _streamDeletes(
     DatabaseSession session,
     CrdtUser crdtUser,
-    Hlc lastSyncHlc,
+    List<Hlc> nodeCheckpoints,
   ) async* {
     final tombstones = await CrdtDataDeleted.db.find(
       session,
-      where: (t) => _tombstoneHlcAfterFilter(t, crdtUser, lastSyncHlc),
+      where: (t) => _tombstoneHlcAfterFilter(t, crdtUser, nodeCheckpoints),
       include: CrdtDataDeleted.include(
         row: CrdtDataRow.include(tbl: CrdtSchemaTable.include()),
         node: CrdtNode.include(),
@@ -475,32 +468,41 @@ class CrdtSync {
   Expression _rowHlcAfterFilter(
     CrdtDataRowTable t,
     CrdtUser crdtUser,
-    Hlc lastSyncHlc,
+    List<Hlc> nodeCheckpoints,
   ) =>
       t.userId.equals(crdtUser.id) &
-      ((t.hlcDatetime > lastSyncHlc.datetime) |
+      t.node.afterAnyCheckpointFilter(
-          (t.hlcDatetime.equals(lastSyncHlc.datetime) &
+        t.node.uuidNodeId,
-              (t.hlcCounter > lastSyncHlc.counter)));
+        t.hlcDatetime,
+        t.hlcCounter,
+        nodeCheckpoints,
+      );
 
   Expression _fieldHlcAfterFilter(
     CrdtDataFieldTable t,
     CrdtUser crdtUser,
-    Hlc lastSyncHlc,
+    List<Hlc> nodeCheckpoints,
   ) =>
       t.row.userId.equals(crdtUser.id) &
-      ((t.hlcDatetime > lastSyncHlc.datetime) |
+      t.node.afterAnyCheckpointFilter(
-          (t.hlcDatetime.equals(lastSyncHlc.datetime) &
+        t.node.uuidNodeId,
-              (t.hlcCounter > lastSyncHlc.counter)));
+        t.hlcDatetime,
+        t.hlcCounter,
+        nodeCheckpoints,
+      );
 
   Expression _tombstoneHlcAfterFilter(
     CrdtDataDeletedTable t,
     CrdtUser crdtUser,
-    Hlc lastSyncHlc,
+    List<Hlc> nodeCheckpoints,
   ) =>
       t.row.userId.equals(crdtUser.id) &
-      ((t.hlcDatetime > lastSyncHlc.datetime) |
+      t.node.afterAnyCheckpointFilter(
-          (t.hlcDatetime.equals(lastSyncHlc.datetime) &
+        t.node.uuidNodeId,
-              (t.hlcCounter > lastSyncHlc.counter)));
+        t.hlcDatetime,
+        t.hlcCounter,
+        nodeCheckpoints,
+      );
 
   Future<dynamic> _fetchDomainRow(
     DatabaseSession session,
@@ -625,3 +627,26 @@ class CrdtSync {
     return entries;
   }
 }
+
+extension on CrdtNodeTable {
+  Expression afterAnyCheckpointFilter(
+    ColumnUuid uuidNodeId,
+    ColumnDateTime hlcDatetime,
+    ColumnInt hlcCounter,
+    List<Hlc> nodeCheckpoints,
+  ) {
+    if (nodeCheckpoints.isEmpty) return Constant.bool(true);
+
+    final caseExpression = Case();
+    for (final checkpoint in nodeCheckpoints) {
+      caseExpression.when(
+        uuidNodeId.equals(checkpoint.nodeId),
+        then:
+            (hlcDatetime > checkpoint.datetime) |
+            (hlcDatetime.equals(checkpoint.datetime) &
+                (hlcCounter > checkpoint.counter)),
+      );
+    }
+    return caseExpression.orElse(Constant.bool(true));
+  }
+}

packages/serverpod_offline_sync_client/lib/src/utils/case_when.dart

@@ -0,0 +1,79 @@
+import 'package:serverpod_database/serverpod_database.dart';
+
+/// A builder for a SQL CASE expression.
+///
+/// Creates a searched CASE expression when [_column] is omitted and a
+/// simple CASE expression when it is provided.
+class Case {
+  /// Creates a new [Case] expression builder.
+  Case([this._column]);
+  final Column? _column;
+  final List<_WhenThen> _whenThenExpressions = [];
+
+  /// Adds a WHEN and THEN expression.
+  void when(Expression expression, {required Expression then}) {
+    _whenThenExpressions.add(_WhenThen(expression, then));
+  }
+
+  /// Completes the CASE expression with an ELSE expression.
+  Expression orElse(Expression expression) {
+    return _CaseExpression(
+      _column,
+      List.unmodifiable(_whenThenExpressions),
+      expression,
+    );
+  }
+}
+
+class _CaseExpression extends Expression<void> {
+  _CaseExpression(
+    this._column,
+    this._whenThenExpressions,
+    this._elseExpression,
+  ) : super(null);
+  final Column? _column;
+  final List<_WhenThen> _whenThenExpressions;
+  final Expression _elseExpression;
+
+  @override
+  List<Column> get columns => [
+    if (_column != null) _column,
+    ..._whenThenExpressions.expand(
+      (expressions) => [
+        ...expressions.when.columns,
+        ...expressions.then.columns,
+      ],
+    ),
+    ..._elseExpression.columns,
+  ];
+
+  @override
+  Iterable<Expression> get depthFirst sync* {
+    yield* super.depthFirst;
+    for (final expressions in _whenThenExpressions) {
+      yield* expressions.when.depthFirst;
+      yield* expressions.then.depthFirst;
+    }
+    yield* _elseExpression.depthFirst;
+  }
+
+  @override
+  String toString() {
+    var expression = 'CASE';
+    if (_column != null) {
+      expression += ' $_column';
+    }
+    for (final expressions in _whenThenExpressions) {
+      // No need to use a string buffer here as the expression is not complex.
+      // ignore: use_string_buffers
+      expression += ' WHEN ${expressions.when} THEN ${expressions.then}';
+    }
+    return '$expression ELSE $_elseExpression END';
+  }
+}
+
+class _WhenThen {
+  _WhenThen(this.when, this.then);
+  final Expression when;
+  final Expression then;
+}

packages/serverpod_offline_sync_server/lib/src/models/sync/since_hlc.spy.yaml

@@ -3,5 +3,10 @@
 class: CrdtSyncSinceHlc
 extends: CrdtSyncStreamEvent
 fields:
-  ### The last HLC timestamp that the sender has received from the other peer.
+  ### Per-node checkpoints that describe which changes the sender already has.
-  lastSyncFromPeer: Hlc
+  ###
+  ### Each checkpoint is an [Hlc] whose `nodeId` is the source node being
+  ### checkpointed. The receiver uses this list to compute a diff without
+  ### relying on a single global HLC, which would miss concurrent changes from
+  ### nodes unknown at the time of the previous sync.
+  nodeCheckpoints: List<Hlc>

packages/serverpod_offline_sync_shared/lib/src/hlc/hlc.dart

@@ -196,3 +196,9 @@ extension BaseHlcToHlc on BaseHlc {
   /// Converts the [BaseHlc] to an [Hlc] with the given [nodeId].
   Hlc toHlcForNode(UuidValue nodeId) => Hlc(hlcDatetime, hlcCounter, nodeId);
 }
+
+/// Extensions for an [Iterable] of [Hlc].
+extension HlcMax on Iterable<Hlc> {
+  /// The greatest HLC represented by the HLCs in this iterable.
+  Hlc get max => reduce((a, b) => a > b ? a : b);
+}

test/serverpod_offline_sync_test_server/test/integration/sync/sync_flow_test.dart

@@ -537,6 +537,88 @@ void main() {
     },
   );
 
+  withServerpod(
+    'Given a server and two client CRDT sessions for the same user that are initialized with the same sync tables',
+    rollbackDatabase: RollbackDatabase.disabled,
+    (sessionBuilder, _) {
+      final rawServerSession = sessionBuilder.build();
+      late CrdtDatabaseSession secondClientSession;
+
+      rawServerSession.serverpod
+        ..initializeCrdtSync(syncTables: serverSyncTables)
+        ..authenticationHandler = (session, token) async => AuthenticationInfo(
+          testCrdtUserId.toString(),
+          <Scope>{},
+          authId: const Uuid().v4(),
+        );
+
+      setUp(() async {
+        testClient = client.Client(
+          'http://localhost:${rawServerSession.server.port}',
+        )..authKeyProvider = TestClientAuthKeyProvider();
+
+        clientSession = CrdtDatabaseSession.wraps(
+          testSession,
+          syncTables: clientSyncTables,
+          persistentUserId: testCrdtUserId,
+        );
+        await clientSession.db.initialize();
+
+        secondClientSession = CrdtDatabaseSession.wraps(
+          await createAdditionalTestSession(),
+          syncTables: clientSyncTables,
+          persistentUserId: testCrdtUserId,
+        );
+        await secondClientSession.db.initialize();
+
+        serverSession = CrdtDatabaseSession.wraps(
+          rawServerSession,
+          syncTables: serverSyncTables,
+        );
+        await serverSession.db.initialize();
+      });
+
+      tearDown(() async {
+        await serverSession.clearUserTables();
+      });
+
+      test(
+        'when the first node uploads an offline deletion after the second node synchronized a newer change '
+        'then the second node receives the deletion on its next sync.',
+        () async {
+          final rowToDelete = await client.Unique.db.insertRow(
+            clientSession,
+            client.Unique(name: 'deleted-by-first-node'),
+          );
+          await testClient.crdt.syncOnce(clientSession);
+          await testClient.crdt.syncOnce(secondClientSession);
+
+          // After both nodes have synchronized, the first node deletes the row.
+          await client.Unique.db.deleteRow(clientSession, rowToDelete);
+
+          // The second node inserts a new row and synchronizes it.
+          await client.Unique.db.insertRow(
+            secondClientSession,
+            client.Unique(name: 'inserted-by-second-node'),
+          );
+          await testClient.crdt.syncOnce(secondClientSession);
+
+          // The first node synchronizes the deletion with the server.
+          await testClient.crdt.syncOnce(clientSession);
+
+          // This last sync from the second node will only receive the deletion
+          // from the first node if checkpoints are tracked per known node.
+          await testClient.crdt.syncOnce(secondClientSession);
+          final deletedRow = await client.Unique.db.findById(
+            secondClientSession,
+            rowToDelete.id!,
+          );
+          expect(deletedRow, isNull);
+        },
+      );
+    },
+  );
+
   withServerpod(
     'Given a server and client CRDT session that are initialized with the same sync tables and a continuous sync interval',
     rollbackDatabase: RollbackDatabase.disabled,

test/serverpod_offline_sync_test_server/test/integration/sync/sync_roundtrip_test.dart

@@ -65,8 +65,9 @@ void main() {
         final mergeSet = await crdtSync
             .collectPendingChanges(
               testSession,
-              otherNodeId: const Uuid().v7obj(),
+              peerNodeId: const Uuid().v7obj(),
               userId: testCrdtUserId,
+              nodeCheckpoints: const [],
             )
             .toList();
 
@@ -117,7 +118,7 @@ void main() {
               testSession,
               userId: testCrdtUserId,
               peerNodeId: const Uuid().v7obj(),
-              sinceHlc: Hlc.zero(const Uuid().v7obj()),
+              nodeCheckpoints: const [],
             )
             .toList();
 
@@ -185,8 +186,9 @@ void main() {
         final mergeSet = await crdtSync
             .collectPendingChanges(
               testSession,
-              otherNodeId: const Uuid().v7obj(),
+              peerNodeId: const Uuid().v7obj(),
               userId: testCrdtUserId,
+              nodeCheckpoints: const [],
             )
             .toList();
 
@@ -208,6 +210,27 @@ void main() {
       },
     );
   });
+
+  test(
+    'Given a CRDT node without local changes '
+    'when synchronization checkpoints are created '
+    'then one fresh checkpoint for the local node is included.',
+    () async {
+      final user = await CrdtUserManager(testSession).getOrCreate(testCrdtUserId);
+      final sinceHlc = await crdtSync.createSyncSinceHlc(
+        testSession,
+        userId: testCrdtUserId,
+        peerNodeId: const Uuid().v7obj(),
+      );
+
+      expect(sinceHlc.nodeCheckpoints, hasLength(1));
+      expect(sinceHlc.nodeCheckpoints.single.nodeId, user.currentNode!.uuidNodeId);
+      expect(
+        sinceHlc.nodeCheckpoints.single,
+        greaterThan(Hlc.zero(sinceHlc.nodeCheckpoints.single.nodeId)),
+      );
+    },
+  );
 }
 
 extension on List<int> {

test/serverpod_offline_sync_test_server/test/integration/test_tools/client_session.dart

@@ -14,6 +14,7 @@ late CrdtDatabaseSession _crdtSession;
 late Directory _tempDir;
 late UuidValue _testCrdtUserId;
 late bool _withPersistentUser;
+var _additionalTestSessionCount = 0;
 
 /// A fresh [ClientDatabaseSession] for each test that is automatically closed
 /// and removed once no longer needed.
@@ -63,6 +64,17 @@ void initTestClientSession({bool withPersistentUser = false}) {
   });
 }
 
+/// Creates an additional client database session in the shared test temporary
+/// directory and closes it when the current test completes.
+Future<ClientDatabaseSession> createAdditionalTestSession() async {
+  final additionalSession = await _testClient.createSession(
+    p.join(_tempDir.path, 'test-${++_additionalTestSessionCount}.db'),
+    isDebugMode: true,
+  );
+  addTearDown(additionalSession.close);
+  return additionalSession;
+}
+
 Future<void> _initialize() async {
   _crdtSession = CrdtDatabaseSession.wraps(
     testSession,