fix: Refresh node clocks across sync wrappers (#152)

packages/serverpod_offline_sync/lib/src/database/database.dart

@@ -168,7 +168,7 @@ class OfflineSyncDatabase implements Database {
         ));
     try {
       await transactionForUser<void>(effectiveSpaceId, (tx) async {
-        await _recorder.lockCurrentUser(tx);
+        await _recorder.lockAndRefreshCurrentNodeHlc(tx);
         await _recorder.mergeChanges(mergeSet, tx);
       });
     } on OfflineSyncIntegrityViolationException catch (exception) {
@@ -308,6 +308,18 @@ class OfflineSyncDatabase implements Database {
     return _stripSpaceIdFromSpaceScopedRead([result], include, transaction).single;
   }
 
+  Future<R> _runTrackedWrite<R>(
+    Transaction? transaction,
+    TransactionFunction<R> action,
+  ) => DatabaseUtil.runInTransactionOrSavepoint(
+    _delegate,
+    transaction,
+    (tx) => _recorder.withCurrentNodeHlc(tx, (tx) async {
+      await _recorder.lockAndRefreshCurrentNodeHlc(tx);
+      return action(tx);
+    }),
+  );
+
   @override
   Future<List<T>> insert<T extends TableRow>(
     List<T> rows, {
@@ -325,8 +337,7 @@ class OfflineSyncDatabase implements Database {
         noReturn: noReturn,
       );
     }
-    return DatabaseUtil.runInTransactionOrSavepoint(
+    return _runTrackedWrite(
-      _delegate,
       transaction,
       (tx) async {
         final prepared = _prepareRowsForInsert(rows, tx);
@@ -426,8 +437,7 @@ class OfflineSyncDatabase implements Database {
         noReturn: noReturn,
       );
     }
-    return DatabaseUtil.runInTransactionOrSavepoint(
+    return _runTrackedWrite(
-      _delegate,
       transaction,
       (tx) async {
         final prepared = _prepareRowsForInsert(rows, tx);
@@ -641,8 +651,7 @@ class OfflineSyncDatabase implements Database {
         noReturn: noReturn,
       );
     }
-    return DatabaseUtil.runInTransactionOrSavepoint(
+    return _runTrackedWrite(
-      _delegate,
       transaction,
       (tx) async {
         final plannedUpdates = await _recorder.planLocalUpdates(rows, columns, tx);
@@ -681,8 +690,7 @@ class OfflineSyncDatabase implements Database {
         transaction: transaction,
       );
     }
-    return DatabaseUtil.runInTransactionOrSavepoint(
+    return _runTrackedWrite(
-      _delegate,
       transaction,
       (tx) async {
         final plannedUpdates = await _recorder.planLocalUpdates([row], columns, tx);
@@ -786,8 +794,7 @@ class OfflineSyncDatabase implements Database {
     }
 
     _assertNoSpaceIdColumnValues<T>(columnValues);
-    return DatabaseUtil.runInTransactionOrSavepoint(
+    return _runTrackedWrite(
-      _delegate,
       transaction,
       (tx) async {
         final result = await _delegate.updateWhere<T>(
@@ -882,8 +889,7 @@ class OfflineSyncDatabase implements Database {
       );
     }
 
-    return DatabaseUtil.runInTransactionOrSavepoint(
+    return _runTrackedWrite(
-      _delegate,
       transaction,
       (tx) async {
         final rows = await _delegate.find<T>(
@@ -956,7 +962,9 @@ class OfflineSyncDatabase implements Database {
     // [transaction] before CRDT tables exist. Callers that need CRDT state
     // ([transactionForUser], mutating ORM methods) initialize explicitly.
     return _delegate.transaction(
-      transactionFunction,
+      _recorder.hasPersistentSpace
+          ? (tx) => _recorder.withCurrentNodeHlc(tx, transactionFunction)
+          : transactionFunction,
       settings: settings,
     );
   }
@@ -984,7 +992,7 @@ class OfflineSyncDatabase implements Database {
         try {
           spaceForTransaction[tx] = space;
           userForTransaction[tx] = userId;
-          return await transactionFunction(tx);
+          return await _recorder.withCurrentNodeHlc(tx, transactionFunction);
         } finally {
           spaceForTransaction.remove(tx);
           userForTransaction.remove(tx);

packages/serverpod_offline_sync/lib/src/database/merge.dart

@@ -65,20 +65,6 @@ class _PendingInsertAttempts {
 
 /// Adds merge-specific behavior to [CrdtMutationRecorder].
 extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
-  /// Locks the current user row so merges can serialize with other work.
-  Future<void> lockCurrentUser(Transaction transaction) async {
-    final user = _context.effectiveSpaceFor(transaction);
-    // Use a row lock without fetching the record since the merge path only
-    // needs serialization against concurrent work for the same user.
-    await OfflineSyncSpace.db.lockRows(
-      _session,
-      where: (t) => t.id.equals(user.id),
-      transaction: transaction,
-      lockMode: LockMode.forUpdate,
-      lockBehavior: LockBehavior.wait,
-    );
-  }
-
   /// Merges remote CRDT changes into the current database.
   Future<void> mergeChanges(
     CrdtMergeSet mergeSet,

packages/serverpod_offline_sync/lib/src/database/merge_utils/recorder_context.dart

@@ -21,6 +21,15 @@ enum ForeignKeyTargetPresence {
   hidden,
 }
 
+class _CurrentNodeHlc {
+  _CurrentNodeHlc(this.node);
+
+  final CrdtNode node;
+  final managers = <int, HlcManager>{};
+  bool active = false;
+  Future<void>? observed;
+}
+
 /// Shared state and database access for the CRDT recorder and its helpers.
 ///
 /// Owns per-recorder space/HLC management and the low-level domain and CRDT
@@ -54,10 +63,55 @@ class CrdtRecorderContext {
     databaseSession,
   );
 
-  final Map<UuidValue, HlcManager> _hlcManagers = {};
+  // A node clock belongs to the active transaction, not a wrapper or a space.
+  // Nested wrappers share it; completing or rolling back the scope drops it.
+  static final _currentNodes = Expando<Map<int, _CurrentNodeHlc>>();
 
-  /// Clears cached [HlcManager] instances for this recorder session.
+  _CurrentNodeHlc _currentNodeFor(OfflineSyncSpace space, Transaction transaction) =>
-  void clearHlcManagers() => _hlcManagers.clear();
+      (_currentNodes[transaction] ??= {})[space.currentNodeId!] ??= _CurrentNodeHlc(
+        space.currentNode!.copyWith(),
+      );
+
+  Future<R> withCurrentNodeHlc<R>(
+    Transaction transaction,
+    TransactionFunction<R> action,
+  ) async {
+    final space = effectiveSpaceFor(transaction);
+    final nodeId = space.currentNodeId!;
+    final currentNode = _currentNodeFor(space, transaction);
+    if (currentNode.active) return action(transaction);
+
+    currentNode.active = true;
+    try {
+      return await action(transaction);
+    } finally {
+      final nodes = _currentNodes[transaction]!..remove(nodeId);
+      if (nodes.isEmpty) _currentNodes[transaction] = null;
+    }
+  }
+
+  Future<void> lockAndRefreshCurrentNodeHlc(Transaction transaction) {
+    final currentNode = _currentNodeFor(effectiveSpaceFor(transaction), transaction);
+    return currentNode.observed ??= _lockAndRefreshCurrentNodeHlc(
+      currentNode,
+      transaction,
+    );
+  }
+
+  Future<void> _lockAndRefreshCurrentNodeHlc(
+    _CurrentNodeHlc currentNode,
+    Transaction transaction,
+  ) async {
+    final nodeId = currentNode.node.id!;
+    final node = await CrdtNode.db.findById(
+      databaseSession,
+      nodeId,
+      transaction: transaction,
+      lockMode: LockMode.forUpdate,
+    );
+    if (node == null) throw StateError('Current CRDT node $nodeId is missing.');
+    currentNode.node.lastHlc = node.lastHlc;
+  }
 
   /// CRDT schema ids by table name: `tableName -> (tableId, columnsByName)`.
   Map<String, (int, Map<String, CrdtSchemaColumn>)> get schema =>
@@ -792,9 +846,9 @@ WHERE "id" IN (${rowIds.sqlLiteralList()})
 
   HlcManager hlcManagerFor(Transaction transaction) {
     final user = effectiveSpaceFor(transaction);
-    return _hlcManagers.putIfAbsent(
+    final currentNode = _currentNodeFor(user, transaction);
-      user.uuidSpaceId,
+    return currentNode.managers[user.id!] ??= HlcManager.forSpace(
-      () => HlcManager.forSpace(user),
+      user.copyWith(currentNode: currentNode.node),
     );
   }
 
@@ -802,11 +856,12 @@ WHERE "id" IN (${rowIds.sqlLiteralList()})
     HlcManager hlcManager,
     Transaction transaction,
   ) async {
-    await CrdtNode.db.updateRow(
+    await CrdtNode.db.update(
       databaseSession,
-      hlcManager.getNode(),
+      [hlcManager.getNode()],
       columns: (t) => [t.lastHlc],
       transaction: transaction,
+      noReturn: true,
     );
   }
 

packages/serverpod_offline_sync/lib/src/database/recorder.dart

@@ -234,13 +234,12 @@ class CrdtMutationRecorder {
 
   /// Initializes the CRDT recorder.
   ///
-  /// Clears this recorder's live space/HLC caches, then ensures the shared
+  /// Clears this recorder's live space cache, then ensures the shared
   /// [OfflineSyncDatabaseContext] metadata is loaded. The shared schema is loaded only
   /// once per process and is not reloaded here; see [OfflineSyncDatabaseContext] for
   /// the cache lifetime assumptions.
   Future<void> initialize() async {
     _context.spaceManager.clearCache();
-    _context.clearHlcManagers();
     _ensureInitializedFuture = null;
     _isInitialized = false;
 
@@ -327,6 +326,19 @@ class CrdtMutationRecorder {
     return _context.spaceManager.getOrCreate(userId);
   }
 
+  /// Whether a plain transaction can share an already initialized client node.
+  bool get hasPersistentSpace => _isInitialized && persistentUserId != null;
+
+  /// Shares the current-node clock within an atomic scope without reading it yet.
+  Future<R> withCurrentNodeHlc<R>(
+    Transaction transaction,
+    TransactionFunction<R> action,
+  ) => _context.withCurrentNodeHlc(transaction, action);
+
+  /// Locks the node and refreshes its clock before the first tracked write or merge.
+  Future<void> lockAndRefreshCurrentNodeHlc(Transaction transaction) =>
+      _context.lockAndRefreshCurrentNodeHlc(transaction);
+
   /// Records the latest acknowledged sync checkpoint for [otherNodeId].
   Future<void> recordSyncCheckpoint(
     UuidValue userId,
@@ -1079,6 +1091,10 @@ class CrdtMutationRecorder {
       null,
       transaction,
     );
+    await _context.persistCurrentNodeHlc(
+      _context.hlcManagerFor(transaction),
+      transaction,
+    );
   }
 
   Future<void> _softDeleteRowsByTable(

packages/serverpod_offline_sync/lib/src/hlc/hlc.dart

@@ -103,7 +103,7 @@ class Hlc implements Comparable<Hlc> {
 
   /// Increments the current timestamp for transmission to another system.
   Hlc increment() {
-    final localWallTime = clock.now().toUtc();
+    final localWallTime = clock.now().toUtcMillisecond();
     final datetimeNew = localWallTime.isAfter(datetime) ? localWallTime : datetime;
     final counterNew = datetimeNew == datetime ? counter + 1 : 0;
 

packages/serverpod_offline_sync/lib/src/managers/hlc.dart

@@ -10,7 +10,7 @@ class HlcManager {
     this.uuidSpaceId,
     this.normalizedSpaceId,
     this.normalizedNodeId,
-    this.lastHlc,
+    this._node,
   );
 
   /// Creates a new [HlcManager] for the current node of [space].
@@ -19,7 +19,7 @@ class HlcManager {
       space.uuidSpaceId,
       space.id!,
       space.currentNodeId!,
-      space.currentNode!.lastHlc ?? Hlc.zero(space.currentNode!.uuidNodeId),
+      space.currentNode!,
     );
   }
 
@@ -35,8 +35,12 @@ class HlcManager {
   /// The normalized node ID of the current node.
   final int normalizedNodeId;
 
+  final CrdtNode _node;
+
   /// The last HLC timestamp for the current node.
-  Hlc lastHlc;
+  Hlc get lastHlc => _node.lastHlc ?? Hlc.zero(_node.uuidNodeId);
+
+  set lastHlc(Hlc value) => _node.lastHlc = value;
 
   /// Returns the next HLC timestamp for the current node.
   Hlc increment() {

test/serverpod_offline_sync_test_server/test/integration/database/cached_clock_read_scope_test.dart

@@ -0,0 +1,98 @@
+import 'package:serverpod_offline_sync_server/serverpod_offline_sync_server.dart';
+import 'package:serverpod_offline_sync_test_client/serverpod_offline_sync_test_client.dart';
+import 'package:test/test.dart';
+
+import '../../../../../benchmark/utils/query_counter.dart';
+import '../test_tools/client_session.dart';
+
+void main() {
+  initTestClientSession(createSessionPerTest: false);
+
+  group('Given an initialized client wrapper with a persistent space,', () {
+    late QueryCountingDatabase database;
+    late OfflineSyncDatabaseSession app;
+    late Unique row;
+
+    setUpAll(() async {
+      database = QueryCountingDatabase((await createAdditionalTestSession()).db);
+      app = OfflineSyncDatabaseSession(
+        database,
+        syncTables: testSyncTables,
+        persistentUserId: testCrdtUserId,
+      );
+      await app.db.initialize();
+
+      row = await Unique.db.insertRow(app, Unique(name: 'client read'));
+    });
+
+    group('when a plain transaction only reads a row,', () {
+      late Unique? found;
+      late int queries;
+      late int rowsRead;
+
+      setUpAll(() async {
+        final queriesBefore = database.queryCount;
+        final rowsBefore = database.rowsRead;
+
+        found = await app.db.transaction(
+          (tx) => Unique.db.findById(app, row.id!, transaction: tx),
+        );
+
+        queries = database.queryCount - queriesBefore;
+        rowsRead = database.rowsRead - rowsBefore;
+      });
+
+      test('then the read uses only membership and domain queries.', () {
+        expect(found?.name, 'client read');
+        expect(queries, 3);
+        expect(rowsRead, 2);
+      });
+    });
+  });
+
+  group('Given an initialized wrapper with a prepared space,', () {
+    late QueryCountingDatabase database;
+    late OfflineSyncDatabaseSession app;
+    late Unique row;
+
+    setUpAll(() async {
+      database = QueryCountingDatabase((await createAdditionalTestSession()).db);
+      app = OfflineSyncDatabaseSession(database, syncTables: testSyncTables);
+      await app.db.initialize();
+
+      row = await app.db.transactionForUser(
+        testCrdtUserId,
+        (tx) => Unique.db.insertRow(
+          app,
+          Unique(name: 'space read'),
+          transaction: tx,
+        ),
+      );
+    });
+
+    group('when a space-bound transaction only reads a row,', () {
+      late Unique? found;
+      late int queries;
+      late int rowsRead;
+
+      setUpAll(() async {
+        final queriesBefore = database.queryCount;
+        final rowsBefore = database.rowsRead;
+
+        found = await app.db.transactionForUser(
+          testCrdtUserId,
+          (tx) => Unique.db.findById(app, row.id!, transaction: tx),
+        );
+
+        queries = database.queryCount - queriesBefore;
+        rowsRead = database.rowsRead - rowsBefore;
+      });
+
+      test('then the read uses only membership and domain queries.', () {
+        expect(found?.name, 'space read');
+        expect(queries, 3);
+        expect(rowsRead, 2);
+      });
+    });
+  });
+}

test/serverpod_offline_sync_test_server/test/integration/server/cached_clock_concurrency_test.dart

@@ -0,0 +1,131 @@
+import 'dart:async';
+
+import 'package:clock/clock.dart';
+import 'package:serverpod/serverpod.dart';
+import 'package:serverpod_offline_sync_server/serverpod_offline_sync_server.dart';
+import 'package:serverpod_offline_sync_test_server/src/generated/protocol.dart'
+    as server;
+import 'package:test/test.dart';
+
+import '../test_tools/client_session.dart';
+import '../test_tools/serverpod_test_tools.dart';
+
+void main() {
+  withServerpod(
+    'Server session node clock serialization',
+    rollbackDatabase: RollbackDatabase.disabled,
+    (sessionBuilder, _) {
+      final raw = sessionBuilder.build();
+
+      group('Given two server wrappers with cached clocks sharing a node,', () {
+        late OfflineSyncDatabaseSession first;
+        late OfflineSyncDatabaseSession second;
+        late UuidValue space;
+        late DateTime start;
+
+        setUpAll(() async {
+          first = OfflineSyncDatabaseSession.wraps(raw, syncTables: [server.Unique.t]);
+          second = OfflineSyncDatabaseSession.wraps(
+            sessionBuilder.build(),
+            syncTables: [server.Unique.t],
+          );
+          await first.db.initialize();
+          await second.db.initialize();
+
+          space = const Uuid().v7obj();
+          start = DateTime.fromMillisecondsSinceEpoch(
+            DateTime.now().millisecondsSinceEpoch,
+            isUtc: true,
+          );
+
+          await withClock(Clock.fixed(start), () async {
+            await first.db.transactionForUser(space, (tx) async {
+              await server.Unique.db.insertRow(
+                first,
+                server.Unique(id: const Uuid().v7obj(), name: 'prime first'),
+                transaction: tx,
+              );
+            });
+
+            await second.db.transactionForUser(space, (tx) async {
+              await server.Unique.db.insertRow(
+                second,
+                server.Unique(id: const Uuid().v7obj(), name: 'prime second'),
+                transaction: tx,
+              );
+            });
+          });
+        });
+
+        tearDownAll(() async {
+          await first.clearUserTables();
+        });
+
+        group('when a stale wrapper writes while a newer node clock is committing,', () {
+          late Hlc firstClock;
+          late Hlc secondClock;
+          late Hlc persistedClock;
+
+          setUpAll(() async {
+            final firstWritten = Completer<void>();
+            final releaseFirst = Completer<void>();
+
+            final firstRow = server.Unique(
+              id: const Uuid().v7obj(),
+              name: 'future first',
+            );
+            final secondRow = server.Unique(
+              id: const Uuid().v7obj(),
+              name: 'later second',
+            );
+
+            final writingFirst = withClock(
+              Clock.fixed(start.add(const Duration(seconds: 10))),
+              () => first.db.transactionForUser(space, (tx) async {
+                await server.Unique.db.insertRow(first, firstRow, transaction: tx);
+                firstWritten.complete();
+                await releaseFirst.future;
+              }),
+            );
+            await firstWritten.future;
+
+            final writingSecond = withClock(
+              Clock.fixed(start),
+              () => second.db.transactionForUser(space, (tx) async {
+                await server.Unique.db.insertRow(second, secondRow, transaction: tx);
+              }),
+            );
+
+            releaseFirst.complete();
+            await Future.wait([writingFirst, writingSecond]);
+
+            firstClock = (await CrdtDataRow.db.findFirstRow(
+              raw,
+              where: (t) => t.uuidRowId.equals(firstRow.id),
+              include: CrdtDataRow.include(node: CrdtNode.include()),
+            ))!.hlc;
+            secondClock = (await CrdtDataRow.db.findFirstRow(
+              raw,
+              where: (t) => t.uuidRowId.equals(secondRow.id),
+              include: CrdtDataRow.include(node: CrdtNode.include()),
+            ))!.hlc;
+
+            persistedClock = (await OfflineSyncSpace.db.findFirstRow(
+              raw,
+              where: (t) => t.uuidSpaceId.equals(space),
+              include: OfflineSyncSpace.include(currentNode: CrdtNode.include()),
+            ))!.currentNode!.lastHlc!;
+          });
+
+          test(
+            'then the later transaction observes the committed clock without regression.',
+            () {
+              expect(secondClock.compareTo(firstClock), greaterThan(0));
+              expect(persistedClock, secondClock);
+            },
+          );
+        });
+      });
+    },
+  );
+}

test/serverpod_offline_sync_test_server/test/integration/sync/sync_cached_clock_receive_test.dart

@@ -0,0 +1,676 @@
+import 'package:clock/clock.dart';
+import 'package:serverpod_database/serverpod_database.dart' show Database;
+import 'package:serverpod_offline_sync_server/serverpod_offline_sync_server.dart';
+import 'package:serverpod_offline_sync_test_client/serverpod_offline_sync_test_client.dart';
+import 'package:test/test.dart';
+
+import '../test_tools/client_session.dart';
+import '../test_tools/crdt_probes.dart';
+import '../test_tools/sync_topology.dart';
+
+void main() {
+  initTestClientSession(createSessionPerTest: false);
+
+  group(
+    'Given a cached app clock and a newer insertion received through another wrapper,',
+    () {
+      late DateTime start;
+      late SyncNode app;
+      late SyncNode existingPeer;
+      late SyncNode emptyPeer;
+      late Unique remote;
+      late Hlc insertionClock;
+
+      setUpAll(() async {
+        start = DateTime.fromMillisecondsSinceEpoch(
+          DateTime.now().millisecondsSinceEpoch,
+          isUtc: true,
+        );
+
+        app = await syncNode(await createAdditionalTestSession(), testSyncTables);
+        existingPeer = await syncNode(
+          await createAdditionalTestSession(),
+          testSyncTables,
+        );
+        emptyPeer = await syncNode(await createAdditionalTestSession(), testSyncTables);
+
+        remote = Unique(id: const Uuid().v7obj(), name: 'remote');
+
+        await withClock(Clock.fixed(start), () async {
+          await app.offlineSync.db.transactionForUser(testCrdtUserId, (tx) async {
+            await Unique.db.insertRow(
+              app.offlineSync,
+              Unique(id: const Uuid().v7obj(), name: 'prime app clock'),
+              transaction: tx,
+            );
+          });
+
+          final insertion = CrdtMergeInsert(
+            uuidSpaceId: testCrdtUserId,
+            tableName: Unique.t.tableName,
+            uuidRowId: remote.id!,
+            uuidNodeId: const Uuid().v7obj(),
+            hlcDatetime: start.add(const Duration(seconds: 10)),
+            hlcCounter: 0,
+            data: remote,
+          );
+          await existingPeer.offlineSync.db.mergeChanges([
+            insertion,
+          ], spaceId: testCrdtUserId);
+          await app.sync.wrapDatabase(app.raw.db).mergeChanges([
+            insertion,
+          ], spaceId: testCrdtUserId);
+        });
+
+        insertionClock = await rowHlc(remote.id!, databaseSession: app.offlineSync);
+      });
+
+      group(
+        'when the app authors an edit and a new row and round-trips with both peers,',
+        () {
+          late CrdtDataField authored;
+          late Hlc newRowClock;
+          late Hlc persistedClock;
+          late Unique? existingValue;
+          late CrdtDataField? existingField;
+          late Unique? bootstrapValue;
+          late CrdtDataField? bootstrapField;
+          late Unique? roundTripped;
+
+          setUpAll(() async {
+            final fresh = Unique(
+              id: const Uuid().v7obj(),
+              name: 'causally later insert',
+            );
+
+            await withClock(Clock.fixed(start), () async {
+              await app.offlineSync.db.transactionForUser(testCrdtUserId, (tx) async {
+                await Unique.db.updateRow(
+                  app.offlineSync,
+                  remote.copyWith(name: 'local edit'),
+                  columns: (t) => [t.name],
+                  transaction: tx,
+                );
+                await Unique.db.insertRow(app.offlineSync, fresh, transaction: tx);
+              });
+
+              authored = (await CrdtDataField.db.findFirstRow(
+                app.raw,
+                where: (t) =>
+                    t.row.uuidRowId.equals(remote.id) & t.column.name.equals('name'),
+                include: CrdtDataField.include(node: CrdtNode.include()),
+              ))!;
+              newRowClock = await rowHlc(fresh.id!, databaseSession: app.offlineSync);
+
+              final space = (await OfflineSyncSpace.db.findFirstRow(
+                app.raw,
+                where: (t) => t.uuidSpaceId.equals(testCrdtUserId),
+                include: OfflineSyncSpace.include(currentNode: CrdtNode.include()),
+              ))!;
+              persistedClock = space.currentNode!.lastHlc!;
+
+              await pushChanges(app, existingPeer);
+              await pushChanges(app, emptyPeer);
+
+              existingValue = await Unique.db.findById(
+                existingPeer.offlineSync,
+                remote.id!,
+              );
+              existingField = await CrdtDataField.db.findFirstRow(
+                existingPeer.raw,
+                where: (t) =>
+                    t.row.uuidRowId.equals(remote.id) & t.column.name.equals('name'),
+                include: CrdtDataField.include(node: CrdtNode.include()),
+              );
+
+              bootstrapValue = await Unique.db.findById(
+                emptyPeer.offlineSync,
+                remote.id!,
+              );
+              bootstrapField = await CrdtDataField.db.findFirstRow(
+                emptyPeer.raw,
+                where: (t) =>
+                    t.row.uuidRowId.equals(remote.id) & t.column.name.equals('name'),
+                include: CrdtDataField.include(node: CrdtNode.include()),
+              );
+
+              await pushChanges(existingPeer, app);
+              await pushChanges(emptyPeer, app);
+              roundTripped = await Unique.db.findById(app.offlineSync, remote.id!);
+            });
+          });
+
+          test('then the authored field clock follows the received row clock.', () {
+            expect(authored.hlc.compareTo(insertionClock), greaterThan(0));
+            expect(persistedClock.compareTo(authored.hlc), greaterThanOrEqualTo(0));
+          });
+
+          test('then an unrelated insertion also observes the received clock.', () {
+            expect(newRowClock.compareTo(insertionClock), greaterThan(0));
+            expect(persistedClock.compareTo(newRowClock), greaterThanOrEqualTo(0));
+          });
+
+          test(
+            'then the existing peer accepts the local edit and its authored clock.',
+            () {
+              expect(existingValue?.name, 'local edit');
+              expect(existingField?.hlc, authored.hlc);
+              expect(roundTripped?.name, 'local edit');
+            },
+          );
+
+          test(
+            'then the empty bootstrap replica preserves the edit and its field clock.',
+            () {
+              expect(bootstrapValue?.name, 'local edit');
+              expect(bootstrapField?.hlc, authored.hlc);
+              expect(roundTripped?.name, 'local edit');
+            },
+          );
+        },
+      );
+    },
+  );
+
+  group('Given two spaces sharing a node and a cached app clock,', () {
+    late SyncNode app;
+    late DateTime start;
+    late Unique local;
+    late Hlc receivedFieldClock;
+
+    setUpAll(() async {
+      app = await syncNode(await createAdditionalTestSession(), testSyncTables);
+
+      start = DateTime.fromMillisecondsSinceEpoch(
+        DateTime.now().millisecondsSinceEpoch,
+        isUtc: true,
+      );
+
+      local = Unique(id: const Uuid().v7obj(), name: 'first space');
+      final otherSpace = const Uuid().v7obj();
+      final otherRow = Unique(id: const Uuid().v7obj(), name: 'second space');
+
+      final remoteNode = const Uuid().v7obj();
+      receivedFieldClock = Hlc(
+        start.add(const Duration(seconds: 10)),
+        41,
+        remoteNode,
+      );
+
+      await withClock(Clock.fixed(start), () async {
+        await app.offlineSync.db.transactionForUser(testCrdtUserId, (tx) async {
+          await Unique.db.insertRow(app.offlineSync, local, transaction: tx);
+        });
+        await app.offlineSync.db.transactionForUser(otherSpace, (tx) async {
+          await Unique.db.insertRow(app.offlineSync, otherRow, transaction: tx);
+        });
+
+        await app.sync.wrapDatabase(app.raw.db).mergeChanges([
+          CrdtMergeUpdate(
+            uuidSpaceId: otherSpace,
+            tableName: Unique.t.tableName,
+            uuidRowId: otherRow.id!,
+            uuidNodeId: remoteNode,
+            hlcDatetime: receivedFieldClock.datetime,
+            hlcCounter: receivedFieldClock.counter,
+            columnName: 'name',
+            value: 'remote second space',
+          ),
+        ], spaceId: otherSpace);
+      });
+    });
+
+    group(
+      'when the app edits its row after the other space receives a newer field,',
+      () {
+        late CrdtDataField authored;
+        late Hlc persistedClock;
+
+        setUpAll(() async {
+          await withClock(Clock.fixed(start), () async {
+            await app.offlineSync.db.transactionForUser(testCrdtUserId, (tx) async {
+              await Unique.db.updateRow(
+                app.offlineSync,
+                local.copyWith(name: 'after other space'),
+                columns: (t) => [t.name],
+                transaction: tx,
+              );
+            });
+          });
+
+          authored = (await CrdtDataField.db.findFirstRow(
+            app.raw,
+            where: (t) =>
+                t.row.uuidRowId.equals(local.id) & t.column.name.equals('name'),
+            include: CrdtDataField.include(node: CrdtNode.include()),
+          ))!;
+
+          final space = (await OfflineSyncSpace.db.findFirstRow(
+            app.raw,
+            where: (t) => t.uuidSpaceId.equals(testCrdtUserId),
+            include: OfflineSyncSpace.include(currentNode: CrdtNode.include()),
+          ))!;
+          persistedClock = space.currentNode!.lastHlc!;
+        });
+
+        test(
+          'then the local field and persisted node observe the other space clock.',
+          () {
+            expect(authored.hlc.compareTo(receivedFieldClock), greaterThan(0));
+            expect(persistedClock.compareTo(authored.hlc), greaterThanOrEqualTo(0));
+          },
+        );
+      },
+    );
+  });
+
+  group('Given two wrappers with cached clocks for the same database,', () {
+    late SyncNode app;
+    late OfflineSyncDatabaseSession second;
+    late DateTime start;
+    late Unique row;
+
+    setUpAll(() async {
+      app = await syncNode(await createAdditionalTestSession(), testSyncTables);
+      second = OfflineSyncDatabaseSession.wraps(app.raw, syncTables: testSyncTables);
+      await second.db.initialize();
+
+      start = DateTime.fromMillisecondsSinceEpoch(
+        DateTime.now().millisecondsSinceEpoch,
+        isUtc: true,
+      );
+      row = Unique(id: const Uuid().v7obj(), name: 'first wrapper');
+
+      await withClock(Clock.fixed(start), () async {
+        await app.offlineSync.db.transactionForUser(testCrdtUserId, (tx) async {
+          await Unique.db.insertRow(app.offlineSync, row, transaction: tx);
+        });
+
+        await second.db.transactionForUser(testCrdtUserId, (tx) async {
+          await Unique.db.updateRow(
+            second,
+            row.copyWith(name: 'prime second wrapper'),
+            columns: (t) => [t.name],
+            transaction: tx,
+          );
+        });
+      });
+    });
+
+    group(
+      'when both wrappers write in the same transaction as wall time moves back,',
+      () {
+        late Hlc firstClock;
+        late Hlc secondClock;
+        late Hlc persistedClock;
+
+        setUpAll(() async {
+          await app.offlineSync.db.transactionForUser(testCrdtUserId, (tx) async {
+            await withClock(
+              Clock.fixed(start.add(const Duration(seconds: 10))),
+              () async {
+                await Unique.db.updateRow(
+                  app.offlineSync,
+                  row.copyWith(name: 'future first wrapper'),
+                  columns: (t) => [t.name],
+                  transaction: tx,
+                );
+              },
+            );
+            firstClock = (await CrdtDataField.db.findFirstRow(
+              app.raw,
+              where: (t) =>
+                  t.row.uuidRowId.equals(row.id) & t.column.name.equals('name'),
+              include: CrdtDataField.include(node: CrdtNode.include()),
+              transaction: tx,
+            ))!.hlc;
+
+            await withClock(Clock.fixed(start), () async {
+              await Unique.db.updateRow(
+                second,
+                row.copyWith(name: 'later second wrapper'),
+                columns: (t) => [t.name],
+                transaction: tx,
+              );
+            });
+            secondClock = (await CrdtDataField.db.findFirstRow(
+              app.raw,
+              where: (t) =>
+                  t.row.uuidRowId.equals(row.id) & t.column.name.equals('name'),
+              include: CrdtDataField.include(node: CrdtNode.include()),
+              transaction: tx,
+            ))!.hlc;
+          });
+
+          final space = (await OfflineSyncSpace.db.findFirstRow(
+            app.raw,
+            where: (t) => t.uuidSpaceId.equals(testCrdtUserId),
+            include: OfflineSyncSpace.include(currentNode: CrdtNode.include()),
+          ))!;
+          persistedClock = space.currentNode!.lastHlc!;
+        });
+
+        test(
+          'then both wrappers advance one transaction-visible clock monotonically.',
+          () {
+            expect(secondClock.compareTo(firstClock), greaterThan(0));
+            expect(persistedClock, secondClock);
+          },
+        );
+      },
+    );
+  });
+
+  group('Given a committed local deletion with no subsequent writes,', () {
+    late SyncNode app;
+    late DateTime start;
+    late Hlc deletionClock;
+    late Hlc persistedDeletionClock;
+
+    setUpAll(() async {
+      app = await syncNode(await createAdditionalTestSession(), testSyncTables);
+
+      start = DateTime.fromMillisecondsSinceEpoch(
+        DateTime.now().millisecondsSinceEpoch,
+        isUtc: true,
+      );
+      final row = Unique(id: const Uuid().v7obj(), name: 'deleted');
+
+      await withClock(Clock.fixed(start), () async {
+        await app.offlineSync.db.transactionForUser(testCrdtUserId, (tx) async {
+          await Unique.db.insertRow(app.offlineSync, row, transaction: tx);
+          await Unique.db.deleteRow(app.offlineSync, row, transaction: tx);
+        });
+      });
+
+      deletionClock = (await CrdtDataDeleted.db.findFirstRow(
+        app.raw,
+        where: (t) => t.row.uuidRowId.equals(row.id),
+        include: CrdtDataDeleted.include(node: CrdtNode.include()),
+      ))!.hlc;
+      persistedDeletionClock = (await OfflineSyncSpace.db.findFirstRow(
+        app.raw,
+        where: (t) => t.uuidSpaceId.equals(testCrdtUserId),
+        include: OfflineSyncSpace.include(currentNode: CrdtNode.include()),
+      ))!.currentNode!.lastHlc!;
+    });
+
+    group('when a new wrapper inserts another row at the same wall time,', () {
+      late Hlc newClock;
+
+      setUpAll(() async {
+        final second = OfflineSyncDatabaseSession.wraps(
+          app.raw,
+          syncTables: testSyncTables,
+        );
+        final row = Unique(id: const Uuid().v7obj(), name: 'after deletion');
+
+        await withClock(Clock.fixed(start), () async {
+          await second.db.transactionForUser(testCrdtUserId, (tx) async {
+            await Unique.db.insertRow(second, row, transaction: tx);
+          });
+        });
+
+        newClock = await rowHlc(row.id!, databaseSession: app.offlineSync);
+      });
+
+      test(
+        'then the persisted deletion clock orders the new write after the tombstone.',
+        () {
+          expect(persistedDeletionClock, deletionClock);
+          expect(newClock.compareTo(deletionClock), greaterThan(0));
+        },
+      );
+    });
+  });
+
+  group('Given a persistent wrapper inside a caller-owned transaction,', () {
+    late OfflineSyncDatabaseSession app;
+    late Database rawDatabase;
+    late DateTime start;
+    late Hlc committedClock;
+
+    setUpAll(() async {
+      final raw = await createAdditionalTestSession();
+      rawDatabase = raw.db;
+      app = OfflineSyncDatabaseSession.wraps(
+        raw,
+        syncTables: testSyncTables,
+        persistentUserId: testCrdtUserId,
+      );
+      await app.db.initialize();
+
+      start = DateTime.fromMillisecondsSinceEpoch(
+        DateTime.now().millisecondsSinceEpoch,
+        isUtc: true,
+      );
+      final prime = Unique(
+        id: const Uuid().v7obj(),
+        name: 'committed before savepoint',
+      );
+
+      await withClock(Clock.fixed(start), () async {
+        await Unique.db.insertRow(app, prime);
+      });
+
+      committedClock = await rowHlc(prime.id!, databaseSession: app);
+    });
+
+    group(
+      'when the caller rolls back a savepoint containing a future-dated write,',
+      () {
+        late Hlc expectedNext;
+        late Hlc nextClock;
+        late Hlc persistedClock;
+        late Unique? rolledBackRow;
+
+        setUpAll(() async {
+          final rolledBack = Unique(
+            id: const Uuid().v7obj(),
+            name: 'rollback savepoint',
+          );
+          final next = Unique(
+            id: const Uuid().v7obj(),
+            name: 'after savepoint rollback',
+          );
+
+          await rawDatabase.transaction((tx) async {
+            final savepoint = await tx.createSavepoint();
+            await withClock(
+              Clock.fixed(start.add(const Duration(seconds: 30))),
+              () async {
+                await Unique.db.insertRow(app, rolledBack, transaction: tx);
+              },
+            );
+            await savepoint.rollback();
+
+            await withClock(Clock.fixed(start), () async {
+              expectedNext = committedClock.increment();
+              await Unique.db.insertRow(app, next, transaction: tx);
+            });
+          });
+
+          nextClock = await rowHlc(next.id!, databaseSession: app);
+          rolledBackRow = await Unique.db.findById(app, rolledBack.id!);
+
+          final space = (await OfflineSyncSpace.db.findFirstRow(
+            app,
+            where: (t) => t.uuidSpaceId.equals(testCrdtUserId),
+            include: OfflineSyncSpace.include(currentNode: CrdtNode.include()),
+          ))!;
+          persistedClock = space.currentNode!.lastHlc!;
+        });
+
+        test(
+          'then the next write reloads the clock surviving the caller rollback.',
+          () {
+            expect(rolledBackRow, isNull);
+            expect(nextClock, expectedNext);
+            expect(persistedClock, expectedNext);
+          },
+        );
+      },
+    );
+  });
+
+  group(
+    'Given a client wrapper with a committed clock before explicit cancellation,',
+    () {
+      late OfflineSyncDatabaseSession app;
+      late DateTime start;
+      late Hlc committedClock;
+
+      setUpAll(() async {
+        app = OfflineSyncDatabaseSession.wraps(
+          await createAdditionalTestSession(),
+          syncTables: testSyncTables,
+          persistentUserId: testCrdtUserId,
+        );
+        await app.db.initialize();
+
+        start = DateTime.fromMillisecondsSinceEpoch(
+          DateTime.now().millisecondsSinceEpoch,
+          isUtc: true,
+        );
+        final prime = Unique(id: const Uuid().v7obj(), name: 'before cancellation');
+
+        await withClock(Clock.fixed(start), () async {
+          await Unique.db.insertRow(app, prime);
+        });
+
+        committedClock = await rowHlc(prime.id!, databaseSession: app);
+      });
+
+      group(
+        'when the caller cancels a future-dated transaction and then writes again,',
+        () {
+          String? returned;
+          Object? failure;
+          late Hlc persistedAfterCancel;
+          late Hlc expectedNext;
+          late Hlc nextClock;
+          late Unique? cancelledRow;
+
+          setUpAll(() async {
+            final cancelled = Unique(id: const Uuid().v7obj(), name: 'cancelled');
+            final next = Unique(id: const Uuid().v7obj(), name: 'after cancellation');
+
+            try {
+              returned = await withClock(
+                Clock.fixed(start.add(const Duration(seconds: 30))),
+                () => app.db.transaction((tx) async {
+                  await Unique.db.insertRow(app, cancelled, transaction: tx);
+                  await tx.cancel();
+                  return 'cancelled';
+                }),
+              );
+            } on Object catch (error) {
+              failure = error;
+            }
+
+            cancelledRow = await Unique.db.findById(app, cancelled.id!);
+            persistedAfterCancel = (await OfflineSyncSpace.db.findFirstRow(
+              app,
+              where: (t) => t.uuidSpaceId.equals(testCrdtUserId),
+              include: OfflineSyncSpace.include(currentNode: CrdtNode.include()),
+            ))!.currentNode!.lastHlc!;
+
+            await withClock(Clock.fixed(start), () async {
+              expectedNext = committedClock.increment();
+              await Unique.db.insertRow(app, next);
+            });
+            nextClock = await rowHlc(next.id!, databaseSession: app);
+          });
+
+          test(
+            'then cancellation returns normally and discards the transaction clock.',
+            () {
+              expect(failure, isNull);
+              expect(returned, 'cancelled');
+              expect(cancelledRow, isNull);
+              expect(persistedAfterCancel, committedClock);
+              expect(nextClock, expectedNext);
+            },
+          );
+        },
+      );
+    },
+  );
+
+  group('Given a wrapper with a committed local clock,', () {
+    late SyncNode app;
+    late DateTime start;
+    late Hlc committedClock;
+
+    setUpAll(() async {
+      app = await syncNode(await createAdditionalTestSession(), testSyncTables);
+
+      start = DateTime.fromMillisecondsSinceEpoch(
+        DateTime.now().millisecondsSinceEpoch,
+        isUtc: true,
+      );
+      final prime = Unique(id: const Uuid().v7obj(), name: 'committed');
+
+      await withClock(Clock.fixed(start), () async {
+        await app.offlineSync.db.transactionForUser(testCrdtUserId, (tx) async {
+          await Unique.db.insertRow(app.offlineSync, prime, transaction: tx);
+        });
+      });
+
+      committedClock = await rowHlc(prime.id!, databaseSession: app.offlineSync);
+    });
+
+    group('when a future-dated transaction rolls back before another local write,', () {
+      late Object? failure;
+      late Hlc afterRollback;
+      late Hlc nextClock;
+      late Hlc expectedNext;
+      late Unique? rolledBackRow;
+
+      setUpAll(() async {
+        final rolledBack = Unique(id: const Uuid().v7obj(), name: 'rolled back');
+        final next = Unique(id: const Uuid().v7obj(), name: 'after rollback');
+
+        try {
+          await withClock(
+            Clock.fixed(start.add(const Duration(seconds: 30))),
+            () async {
+              await app.offlineSync.db.transactionForUser(testCrdtUserId, (tx) async {
+                await Unique.db.insertRow(app.offlineSync, rolledBack, transaction: tx);
+                throw Exception('rollback clock');
+              });
+            },
+          );
+        } on Exception catch (error) {
+          failure = error;
+        }
+
+        final space = (await OfflineSyncSpace.db.findFirstRow(
+          app.raw,
+          where: (t) => t.uuidSpaceId.equals(testCrdtUserId),
+          include: OfflineSyncSpace.include(currentNode: CrdtNode.include()),
+        ))!;
+        afterRollback = space.currentNode!.lastHlc!;
+        rolledBackRow = await Unique.db.findById(app.offlineSync, rolledBack.id!);
+
+        await withClock(Clock.fixed(start), () async {
+          expectedNext = committedClock.increment();
+          await app.offlineSync.db.transactionForUser(testCrdtUserId, (tx) async {
+            await Unique.db.insertRow(app.offlineSync, next, transaction: tx);
+          });
+          nextClock = await rowHlc(next.id!, databaseSession: app.offlineSync);
+        });
+      });
+
+      test(
+        'then rollback leaves neither a persisted clock change nor a poisoned cache.',
+        () {
+          expect(failure, isA<Exception>());
+          expect(rolledBackRow, isNull);
+          expect(afterRollback, committedClock);
+          expect(nextClock, expectedNext);
+        },
+      );
+    });
+  });
+}

test/serverpod_offline_sync_test_server/test/integration/sync/sync_clock_precision_test.dart

@@ -0,0 +1,98 @@
+import 'package:clock/clock.dart';
+import 'package:serverpod_offline_sync_server/serverpod_offline_sync_server.dart';
+import 'package:serverpod_offline_sync_test_client/serverpod_offline_sync_test_client.dart';
+import 'package:test/test.dart';
+
+import '../test_tools/client_session.dart';
+import '../test_tools/crdt_probes.dart';
+import '../test_tools/sync_topology.dart';
+
+void main() {
+  initTestClientSession(createSessionPerTest: false);
+
+  group('Given a received insertion with a counter in the current millisecond,', () {
+    late SyncNode app;
+    late SyncNode peer;
+    late DateTime start;
+    late Unique row;
+    late Hlc insertionClock;
+
+    setUpAll(() async {
+      app = await syncNode(await createAdditionalTestSession(), testSyncTables);
+      peer = await syncNode(await createAdditionalTestSession(), testSyncTables);
+
+      start = DateTime.fromMillisecondsSinceEpoch(
+        DateTime.now().millisecondsSinceEpoch,
+        isUtc: true,
+      );
+
+      row = Unique(id: const Uuid().v7obj(), name: 'remote within millisecond');
+      final insertion = CrdtMergeInsert(
+        uuidSpaceId: testCrdtUserId,
+        tableName: Unique.t.tableName,
+        uuidRowId: row.id!,
+        uuidNodeId: const Uuid().v7obj(),
+        hlcDatetime: start,
+        hlcCounter: 41,
+        data: row,
+      );
+
+      await withClock(Clock.fixed(start), () async {
+        await app.offlineSync.db.mergeChanges([insertion], spaceId: testCrdtUserId);
+        await peer.offlineSync.db.mergeChanges([insertion], spaceId: testCrdtUserId);
+      });
+
+      insertionClock = await rowHlc(row.id!, databaseSession: app.offlineSync);
+    });
+
+    group('when the same wrapper edits the row half a millisecond later,', () {
+      late Hlc authoredClock;
+      late Hlc persistedClock;
+      late Unique? peerValue;
+      late Unique? roundTripped;
+
+      setUpAll(() async {
+        await withClock(
+          Clock.fixed(start.add(const Duration(microseconds: 500))),
+          () async {
+            await app.offlineSync.db.transactionForUser(testCrdtUserId, (tx) async {
+              await Unique.db.updateRow(
+                app.offlineSync,
+                row.copyWith(name: 'local within millisecond'),
+                columns: (t) => [t.name],
+                transaction: tx,
+              );
+            });
+
+            authoredClock = (await CrdtDataField.db.findFirstRow(
+              app.raw,
+              where: (t) =>
+                  t.row.uuidRowId.equals(row.id) & t.column.name.equals('name'),
+              include: CrdtDataField.include(node: CrdtNode.include()),
+            ))!.hlc;
+            persistedClock = (await OfflineSyncSpace.db.findFirstRow(
+              app.raw,
+              where: (t) => t.uuidSpaceId.equals(testCrdtUserId),
+              include: OfflineSyncSpace.include(currentNode: CrdtNode.include()),
+            ))!.currentNode!.lastHlc!;
+
+            await pushChanges(app, peer);
+            peerValue = await Unique.db.findById(peer.offlineSync, row.id!);
+
+            await pushChanges(peer, app);
+            roundTripped = await Unique.db.findById(app.offlineSync, row.id!);
+          },
+        );
+      });
+
+      test('then the counter advances and the edit survives the peer round trip.', () {
+        expect(authoredClock.compareTo(insertionClock), greaterThan(0));
+        expect(authoredClock.datetime, start);
+        expect(authoredClock.counter, 42);
+        expect(persistedClock, authoredClock);
+        expect(peerValue?.name, 'local within millisecond');
+        expect(roundTripped?.name, 'local within millisecond');
+      });
+    });
+  });
+}