perf: Batch projection reads and engine writes (#157)

.github/copilot-instructions.md

@@ -22,6 +22,16 @@
 - Remove unnecessary comments from production code.
 - Ensure linting passes before considering the task finished.
 
+## Performance Changes
+
+- Do not infer application bulk-write intent by inspecting or comparing domain
+  row values. Applications express uniform bulk edits explicitly through
+  `updateWhere`; the engine must not rediscover that intent from equal values.
+- Engine-owned planning may drive batching, including projected explicit-null
+  column sets, query predicates, schema identities, and merge operation boundaries.
+  Preserve domain inspection required for CRDT correctness and dependency traversal.
+- Compare proposed gains against the best existing API for the operation.
+
 ## Pubspec Dependency Management
 
 - Do not add any new dependencies, unless explicitly asked.

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

@@ -1,68 +1,5 @@
 part of 'recorder.dart';
 
-/// Field metadata for merged inserts, waiting to be written as one batch.
-///
-/// Every query behind [CrdtForeignKeyProjector.recordInsertAttempts] already
-/// takes a set of row ids, so a batch of inserts into one table costs the same
-/// as one insert. Rows are grouped by node as well as table because the merge
-/// cache stores each field with the node that authored it.
-class _PendingInsertAttempts {
-  final Map<(String, int), ({Set<UuidValue> rowIds, CrdtNode node})> _byGroup = {};
-  final Map<String, ProjectionAttemptsByField> _attemptsByTable = {};
-  final Set<MergeRowKey> _rows = {};
-
-  bool get isEmpty => _byGroup.isEmpty;
-
-  /// Whether metadata for [rowKey] is still unwritten.
-  bool holds(MergeRowKey rowKey) => _rows.contains(rowKey);
-
-  void add({
-    required String tableName,
-    required UuidValue rowId,
-    required CrdtNode node,
-    required ProjectionAttemptsByField attempts,
-  }) {
-    _byGroup
-        .putIfAbsent(
-          (tableName, node.id!),
-          () => (rowIds: <UuidValue>{}, node: node),
-        )
-        .rowIds
-        .add(rowId);
-    _rows.add((tableName, rowId));
-    if (attempts.isEmpty) return;
-    _attemptsByTable
-        .putIfAbsent(tableName, () => <MergeFieldKey, ProjectionAttempt>{})
-        .addAll(attempts);
-  }
-
-  /// Empties the collector and returns what it held.
-  List<
-    ({
-      String tableName,
-      Set<UuidValue> rowIds,
-      CrdtNode node,
-      ProjectionAttemptsByField attempts,
-    })
-  >
-  take() {
-    final groups = [
-      for (final MapEntry(key: key, value: group) in _byGroup.entries)
-        (
-          tableName: key.$1,
-          rowIds: group.rowIds,
-          node: group.node,
-          attempts:
-              _attemptsByTable[key.$1] ?? const <MergeFieldKey, ProjectionAttempt>{},
-        ),
-    ];
-    _byGroup.clear();
-    _attemptsByTable.clear();
-    _rows.clear();
-    return groups;
-  }
-}
-
 /// Adds merge-specific behavior to [CrdtMutationRecorder].
 extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
   /// Merges remote CRDT changes into the current database.
@@ -142,62 +79,61 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
             transaction,
           );
 
-    // Field metadata for merged inserts is recorded per table and node rather
+    final writes = _MergeWriteBatch(
-    // than per row: every query behind it already takes a set of row ids.
+      session: _session,
-    final pendingAttempts = _PendingInsertAttempts();
+      projector: _foreignKeyProjector,
+      fields: context.fields,
+      transaction: transaction,
+    );
     final deferredDeletes = <CrdtMergeDelete>[];
 
-    for (final operation in operations) {
+    await writes.apply(
-      if (!_context.isCrdtTrackedTableName(operation.tableName)) {
+      operations.where(
-        continue;
+        (operation) => _context.isCrdtTrackedTableName(operation.tableName),
-      }
+      ),
-      // A later operation on a row resolves its field metadata through the
+      (operation) async {
-      // merge cache, so anything still pending for that row is written first.
+        switch (operation) {
-      if (pendingAttempts.holds((operation.tableName, operation.uuidRowId))) {
+          case final CrdtMergeInsert insert:
-        await _flushInsertAttempts(pendingAttempts, context, transaction);
+            await _applyMergeInsert(
-      }
+              insert,
+              nodesByUuid,
+              context,
+              authoredOverlays,
+              batch.plan,
+              writes,
+              presence,
+              transaction,
+            );
+          case final CrdtMergeUpdate update:
+            await _applyMergeUpdate(
+              update,
+              nodesByUuid,
+              context,
+              authoredOverlays,
+              writes,
+              transaction,
+            );
+          case final CrdtMergeDelete delete:
+            // A higher delete/restore generation can have an older HLC than a
+            // concurrent reinsertion. Bootstrap must retain it until this batch
+            // has inserted the row it belongs to.
+            if (!context.rows.containsKey((delete.tableName, delete.uuidRowId))) {
+              deferredDeletes.add(delete);
+              return;
+            }
+            await _applyMergeDelete(
+              delete,
+              nodesByUuid,
+              context,
+              transaction,
+            );
+            // A delete can hide the row, and a resurrecting one can bring it
+            // back; either way what was known about it no longer holds.
+            presence?.forget((delete.tableName, delete.uuidRowId));
+        }
+      },
+    );
 
-      switch (operation) {
-        case final CrdtMergeInsert insert:
-          await _applyMergeInsert(
-            insert,
-            nodesByUuid,
-            context,
-            authoredOverlays,
-            batch.plan,
-            pendingAttempts,
-            presence,
-            transaction,
-          );
-        case final CrdtMergeUpdate update:
-          await _applyMergeUpdate(
-            update,
-            nodesByUuid,
-            context,
-            authoredOverlays,
-            transaction,
-          );
-        case final CrdtMergeDelete delete:
-          // A higher delete/restore generation can have an older HLC than a
-          // concurrent reinsertion. Bootstrap must retain it until this batch
-          // has inserted the row it belongs to.
-          if (!context.rows.containsKey((delete.tableName, delete.uuidRowId))) {
-            deferredDeletes.add(delete);
-            continue;
-          }
-          await _applyMergeDelete(
-            delete,
-            nodesByUuid,
-            context,
-            transaction,
-          );
-          // A delete can hide the row, and a resurrecting one can bring it
-          // back; either way what was known about it no longer holds.
-          presence?.forget((delete.tableName, delete.uuidRowId));
-      }
-    }
-
-    await _flushInsertAttempts(pendingAttempts, context, transaction);
     for (final delete in deferredDeletes) {
       await _applyMergeDelete(delete, nodesByUuid, context, transaction);
     }
@@ -471,7 +407,7 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
     MergeContext context,
     Map<MergeFieldKey, Object?> authoredOverlays,
     ProjectionPlan? batchPlan,
-    _PendingInsertAttempts pendingAttempts,
+    _MergeWriteBatch writes,
     CrdtForeignKeyPresenceCache? presence,
     Transaction transaction,
   ) async {
@@ -525,7 +461,7 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
         // same-space recovery), so same-merge field updates resolve ownership
         // from the cache instead of re-reading the domain row.
         context.domainOwners[rowKey] = (exists: true, spaceId: mergingSpaceId);
-        pendingAttempts.add(
+        writes.recordInsertAttempts(
           tableName: insert.tableName,
           rowId: insert.uuidRowId,
           node: remoteNode,
@@ -616,36 +552,12 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
     return (domain: plannedDomain, attempts: attempts);
   }
 
-  /// Writes the field metadata collected for merged inserts, one call per
-  /// table and node.
-  ///
-  /// Grouped by node because the merge cache pairs every field it stores with
-  /// the node that authored it, which is how the cached field resolves its own
-  /// clock.
-  Future<void> _flushInsertAttempts(
-    _PendingInsertAttempts pending,
-    MergeContext context,
-    Transaction transaction,
-  ) async {
-    if (pending.isEmpty) return;
-
-    final groups = pending.take();
-    for (final group in groups) {
-      await _foreignKeyProjector.recordInsertAttempts(
-        group.tableName,
-        group.rowIds,
-        transaction,
-        group.attempts,
-        mergeCache: (fields: context.fields, node: group.node),
-      );
-    }
-  }
-
   Future<void> _applyMergeUpdate(
     CrdtMergeUpdate update,
     Map<UuidValue, CrdtNode> remoteNodes,
     MergeContext context,
     Map<MergeFieldKey, Object?> authoredOverlays,
+    _MergeWriteBatch writes,
     Transaction transaction,
   ) async {
     final rowKey = (update.tableName, update.uuidRowId);
@@ -673,6 +585,7 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
       remoteNode: remoteNode,
       incomingHlc: update.hlc,
       fields: context.fields,
+      writes: writes,
       transaction: transaction,
     );
 
@@ -780,8 +693,9 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
     CrdtNode remoteNode,
     Hlc incomingHlc,
     CrdtDataField? currentField,
-    Transaction transaction,
+    Transaction transaction, {
-  ) async {
+    _MergeWriteBatch? writes,
+  }) async {
     // Projection can create field metadata for a row this merge context loaded
     // before the row existed, so an unseen field may still be persisted. Adopt
     // it instead of inserting a second row onto the row/column unique index.
@@ -794,6 +708,8 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
         );
 
     if (existingField == null) {
+      // Dependent metadata needs the field identity, so new fields cannot
+      // wait in the batch of existing field-clock updates.
       final insertedField = await CrdtDataField.db.insertRow(
         _session,
         CrdtDataField(
@@ -822,12 +738,16 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
       hlcCounter: incomingHlc.counter,
     );
 
-    await CrdtDataField.db.update(
+    if (writes != null) {
-      _session,
+      writes.recordFieldUpdate(updatedField);
-      [updatedField],
+    } else {
-      columns: (t) => [t.nodeId, t.hlcDatetime, t.hlcCounter],
+      await CrdtDataField.db.update(
-      transaction: transaction,
+        _session,
-    );
+        [updatedField],
+        columns: (t) => [t.nodeId, t.hlcDatetime, t.hlcCounter],
+        transaction: transaction,
+      );
+    }
     return updatedField;
   }
 
@@ -1018,6 +938,7 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
     required Map<MergeFieldKey, CrdtDataField> fields,
     required Transaction transaction,
     CrdtSchemaColumn? schemaColumn,
+    _MergeWriteBatch? writes,
   }) async {
     final resolvedSchemaColumn =
         schemaColumn ?? _context.schemaColumn(tableName, columnName);
@@ -1035,6 +956,7 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
       incomingHlc,
       currentField,
       transaction,
+      writes: writes,
     );
     return true;
   }

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

@@ -94,7 +94,7 @@ TableRow<UuidValue?> withExplicitInsertNulls(
 
 /// Keeps per-row default semantics when a batch contains projected nulls.
 /// PostgreSQL builds a batch using the first row's table, so rows with different
-/// null overrides must be written separately within the caller's transaction.
+/// null overrides use separate consecutive batches in the caller's transaction.
 @internal
 Future<List<T>> insertWithExplicitNulls<T extends TableRow>(
   Database database,
@@ -112,15 +112,30 @@ Future<List<T>> insertWithExplicitNulls<T extends TableRow>(
       noReturn: noReturn,
     );
   }
-  return [
+  final inserted = <T>[];
-    for (final row in rows)
+  var start = 0;
-      ...(await database.insert<TableRow<UuidValue?>>(
+  while (start < rows.length) {
-        [withExplicitInsertNulls(row, explicitNulls[row.id] ?? const {})],
+    final columns = explicitNulls[rows[start].id] ?? const <String>{};
-        transaction: transaction,
+    var end = start + 1;
-        ignoreConflicts: ignoreConflicts,
+    while (end < rows.length) {
-        noReturn: noReturn,
+      final next = explicitNulls[rows[end].id] ?? const <String>{};
-      )).cast<T>(),
+      if (columns.length != next.length || !columns.containsAll(next)) break;
-  ];
+      end++;
+    }
+
+    final result = await database.insert<TableRow<UuidValue?>>(
+      [
+        for (var index = start; index < end; index++)
+          withExplicitInsertNulls(rows[index], columns),
+      ],
+      transaction: transaction,
+      ignoreConflicts: ignoreConflicts,
+      noReturn: noReturn,
+    );
+    inserted.addAll(result.cast<T>());
+    start = end;
+  }
+  return inserted;
 }
 
 class _ExplicitNullRow implements TableRow<UuidValue?> {

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

@@ -11,6 +11,8 @@ import 'recorder_context.dart';
 import 'types.dart';
 import 'unique_resolver.dart';
 
+part 'projection_reads.dart';
+
 typedef _ProjectedForeignKeyRow = ({
   MergeRowKey key,
   CrdtDataRow crdtRow,
@@ -880,10 +882,11 @@ class CrdtForeignKeyProjector {
     Set<String>? seedTables,
     Set<MergeRowKey>? seedRows,
   }) async {
-    final rows = <MergeRowKey, _ProjectedForeignKeyRow>{};
+    final reader = _ProjectionReader(_context, transaction);
-    final fieldIds = <MergeFieldKey, int>{};
+    final rows = reader.rows;
-    final attemptedValues = <MergeFieldKey, CrdtDataAttemptedValue>{};
+    final fieldIds = reader.fieldIds;
-    final fieldHlcs = <MergeFieldKey, Hlc>{};
+    final attemptedValues = reader.attemptedValues;
+    final fieldHlcs = reader.fieldHlcs;
 
     final tablesToLoad = _tablesToLoad(
       seedTables: seedTables,
@@ -897,18 +900,10 @@ class CrdtForeignKeyProjector {
     );
 
     if (seedRows == null) {
-      for (final tableName in tablesToLoad) {
+      await reader.loadRows(
-        await _loadTableRowsInto(
+        requestedRows: {for (final table in tablesToLoad) table: null},
-          tableName: tableName,
+        columnsByTable: columnsByTable,
-          rowIds: null,
+      );
-          columnNames: columnsByTable[tableName] ?? const <String>{},
-          rows: rows,
-          fieldIds: fieldIds,
-          attemptedValues: attemptedValues,
-          fieldHlcs: fieldHlcs,
-          transaction: transaction,
-        );
-      }
     } else {
       await _loadRowClosureInto(
         tablesToLoad: tablesToLoad,
@@ -916,11 +911,7 @@ class CrdtForeignKeyProjector {
         pendingInserts: pendingInserts,
         authoredOverlays: authoredOverlays,
         columnsByTable: columnsByTable,
-        rows: rows,
+        reader: reader,
-        fieldIds: fieldIds,
-        attemptedValues: attemptedValues,
-        fieldHlcs: fieldHlcs,
-        transaction: transaction,
       );
     }
 
@@ -1105,78 +1096,6 @@ class CrdtForeignKeyProjector {
     return childRowsByAttempt;
   }
 
-  /// Loads [rowIds] of [tableName], or every row of it when [rowIds] is null.
-  ///
-  /// Returns the row ids that exist, so a closure pass can expand from them.
-  Future<Set<UuidValue>> _loadTableRowsInto({
-    required String tableName,
-    required Set<UuidValue>? rowIds,
-    required Set<String> columnNames,
-    required Map<MergeRowKey, _ProjectedForeignKeyRow> rows,
-    required Map<MergeFieldKey, int> fieldIds,
-    required Map<MergeFieldKey, CrdtDataAttemptedValue> attemptedValues,
-    required Map<MergeFieldKey, Hlc> fieldHlcs,
-    required Transaction transaction,
-  }) async {
-    if (rowIds != null && rowIds.isEmpty) return const {};
-
-    final (tableId, _) = _context.schema[tableName]!;
-    final userId = _context.hlcManagerFor(transaction).normalizedSpaceId;
-    final crdtRows = await CrdtDataRow.db.find(
-      _context.databaseSession,
-      where: (t) {
-        final inSpace = t.spaceId.equals(userId) & t.tblId.equals(tableId);
-        return rowIds == null ? inSpace : inSpace & t.uuidRowId.inSet(rowIds);
-      },
-      include: CrdtDataRow.include(
-        node: CrdtNode.include(),
-        deleted: CrdtDataDeleted.include(node: CrdtNode.include()),
-      ),
-      orderBy: (t) => t.uuidRowId,
-      transaction: transaction,
-    );
-    if (crdtRows.isEmpty) return const {};
-
-    final loadedIds = {for (final row in crdtRows) row.uuidRowId};
-    final valuesByRowId = columnNames.isEmpty
-        ? <UuidValue, Map<String, Object?>>{}
-        : await _context.readDomainColumnValues(
-            tableName,
-            loadedIds,
-            columnNames.toList(),
-            transaction,
-          );
-
-    for (final row in crdtRows) {
-      final key = (tableName, row.uuidRowId);
-      rows[key] = (
-        key: key,
-        crdtRow: row,
-        values: Map<String, Object?>.from(valuesByRowId[row.uuidRowId] ?? const {}),
-      );
-    }
-
-    if (columnNames.isNotEmpty) {
-      final loadedFields = await _loadFields(
-        tableName: tableName,
-        rowIds: loadedIds,
-        columnNames: columnNames,
-        transaction: transaction,
-      );
-      for (final field in loadedFields) {
-        final fieldKey = (tableName, field.row!.uuidRowId, field.column!.name);
-        fieldIds[fieldKey] = field.id!;
-        fieldHlcs[fieldKey] = field.hlc;
-        final attempted = field.attemptedValue;
-        if (attempted != null) {
-          attemptedValues[fieldKey] = attempted;
-        }
-      }
-    }
-
-    return loadedIds;
-  }
-
   /// Loads the rows the seeds can reach instead of every row of their tables.
   ///
   /// Defaults are loaded as evaluation context. Their implicit dependents join
@@ -1199,12 +1118,13 @@ class CrdtForeignKeyProjector {
     required List<PendingProjectionRow> pendingInserts,
     required Map<MergeFieldKey, Object?> authoredOverlays,
     required Map<String, Set<String>> columnsByTable,
-    required Map<MergeRowKey, _ProjectedForeignKeyRow> rows,
+    required _ProjectionReader reader,
-    required Map<MergeFieldKey, int> fieldIds,
-    required Map<MergeFieldKey, CrdtDataAttemptedValue> attemptedValues,
-    required Map<MergeFieldKey, Hlc> fieldHlcs,
-    required Transaction transaction,
   }) async {
+    final transaction = reader._transaction;
+    final rows = reader.rows;
+    final attemptedValues = reader.attemptedValues;
+    final hasAttemptedClaims = await reader.hasAttemptedClaims(tablesToLoad);
+
     final unwrittenValues = _unwrittenValues(pendingInserts, authoredOverlays);
 
     final requested = <String, Set<UuidValue>>{};
@@ -1245,20 +1165,10 @@ class CrdtForeignKeyProjector {
         final wave = queued;
         queued = <String, Set<UuidValue>>{};
 
-        final loadedNow = <String, Set<UuidValue>>{};
+        final loadedNow = await reader.loadRows(
-        for (final MapEntry(key: tableName, value: ids) in wave.entries) {
+          requestedRows: wave,
-          final loaded = await _loadTableRowsInto(
+          columnsByTable: columnsByTable,
-            tableName: tableName,
+        );
-            rowIds: ids,
-            columnNames: columnsByTable[tableName] ?? const <String>{},
-            rows: rows,
-            fieldIds: fieldIds,
-            attemptedValues: attemptedValues,
-            fieldHlcs: fieldHlcs,
-            transaction: transaction,
-          );
-          if (loaded.isNotEmpty) loadedNow[tableName] = loaded;
-        }
 
         for (final MapEntry(key: tableName, value: ids) in frontier.entries) {
           (loadedNow[tableName] ??= <UuidValue>{}).addAll(ids);
@@ -1268,6 +1178,7 @@ class CrdtForeignKeyProjector {
 
         await _expandRowClosure(
           loadedNow: loadedNow,
+          hasAttemptedClaims: hasAttemptedClaims,
           tablesToLoad: tablesToLoad,
           rows: rows,
           unwrittenValues: unwrittenValues,
@@ -1535,6 +1446,7 @@ class CrdtForeignKeyProjector {
     required Map<MergeRowKey, Map<String, Object?>> unwrittenValues,
     required Map<MergeFieldKey, CrdtDataAttemptedValue> attemptedValues,
     required void Function(String tableName, Iterable<UuidValue> ids) enqueue,
+    required bool hasAttemptedClaims,
     required Transaction transaction,
   }) async {
     // Both the stored and the unwritten value matter: the walk has to reach the
@@ -1553,14 +1465,19 @@ class CrdtForeignKeyProjector {
       }..remove(null);
     }
 
+    final claims = _ProjectionClaimReads(
+      _context,
+      transaction,
+      includeAttemptedClaims: hasAttemptedClaims,
+    );
+
     for (final edge in _foreignKeys.edges) {
-      await _expandToChildren(
+      _expandToChildren(
         edge: edge,
         loadedNow: loadedNow,
         tablesToLoad: tablesToLoad,
         valuesFor: valuesFor,
-        enqueue: enqueue,
+        collectClaims: claims.add,
-        transaction: transaction,
       );
       await _expandToParents(
         edge: edge,
@@ -1574,27 +1491,27 @@ class CrdtForeignKeyProjector {
       );
     }
 
-    await _expandUniqueClaimants(
+    _expandUniqueClaimants(
       loadedNow: loadedNow,
       attemptedValues: attemptedValues,
       valuesFor: valuesFor,
-      enqueue: enqueue,
+      collectClaims: claims.add,
-      transaction: transaction,
     );
+
+    await claims.enqueueMatches(enqueue);
   }
 
   /// Enqueues the children of the parents loaded in this wave.
   ///
   /// Hiding or restoring a parent decides its children's fate, so they join the
   /// walk whether they still name it or only remember it as an attempt.
-  Future<void> _expandToChildren({
+  void _expandToChildren({
     required ForeignKeyEdge edge,
     required Map<String, Set<UuidValue>> loadedNow,
     required Set<String> tablesToLoad,
     required Set<Object?> Function(MergeRowKey, String) valuesFor,
-    required void Function(String tableName, Iterable<UuidValue> ids) enqueue,
+    required void Function(String, Map<String, Set<Object?>>) collectClaims,
-    required Transaction transaction,
+  }) {
-  }) async {
     final parentIds = loadedNow[edge.parentTableName];
     if (parentIds == null || parentIds.isEmpty) return;
     if (!tablesToLoad.contains(edge.childTableName)) return;
@@ -1611,12 +1528,7 @@ class CrdtForeignKeyProjector {
     }
     if (references.isEmpty) return;
 
-    await _enqueueRowsClaiming(
+    collectClaims(edge.childTableName, {edge.childColumn: references});
-      tableName: edge.childTableName,
-      valuesByColumn: {edge.childColumn: references},
-      enqueue: enqueue,
-      transaction: transaction,
-    );
   }
 
   /// Enqueues the parents of the children loaded in this wave.
@@ -1686,13 +1598,12 @@ class CrdtForeignKeyProjector {
   /// own unique index for it. A row that wants a value it does not hold is
   /// already loaded, because holding a projection means holding an attempted
   /// value.
-  Future<void> _expandUniqueClaimants({
+  void _expandUniqueClaimants({
     required Map<String, Set<UuidValue>> loadedNow,
     required Map<MergeFieldKey, CrdtDataAttemptedValue> attemptedValues,
     required Set<Object?> Function(MergeRowKey, String) valuesFor,
-    required void Function(String tableName, Iterable<UuidValue> ids) enqueue,
+    required void Function(String, Map<String, Set<Object?>>) collectClaims,
-    required Transaction transaction,
+  }) {
-  }) async {
     for (final MapEntry(key: tableName, value: rowIds) in loadedNow.entries) {
       for (final uniqueIndex in _uniqueResolver.uniqueIndexesFor(tableName)) {
         final claimedByColumn = <String, Set<Object?>>{};
@@ -1713,46 +1624,11 @@ class CrdtForeignKeyProjector {
           continue;
         }
 
-        await _enqueueRowsClaiming(
+        collectClaims(tableName, claimedByColumn);
-          tableName: tableName,
-          valuesByColumn: claimedByColumn,
-          enqueue: enqueue,
-          transaction: transaction,
-        );
       }
     }
   }
 
-  /// Enqueues every row of [tableName] that holds, or is owed, these values.
-  ///
-  /// A row that was repaired or released away from a value no longer names it
-  /// in the domain column, so the attempted value is the only record left of
-  /// the claim it is waiting to take back.
-  Future<void> _enqueueRowsClaiming({
-    required String tableName,
-    required Map<String, Set<Object?>> valuesByColumn,
-    required void Function(String tableName, Iterable<UuidValue> ids) enqueue,
-    required Transaction transaction,
-  }) async {
-    enqueue(
-      tableName,
-      await _context.findDomainRowIdsWhereColumnsIn(
-        tableName: tableName,
-        valuesByColumn: valuesByColumn,
-        transaction: transaction,
-      ),
-    );
-    enqueue(
-      tableName,
-      await _context.findRowIdsHoldingAttemptedValues(
-        tableName: tableName,
-        columnNames: valuesByColumn.keys.toSet(),
-        values: {for (final values in valuesByColumn.values) ...values},
-        transaction: transaction,
-      ),
-    );
-  }
-
   Future<List<CrdtDataField>> _loadFields({
     required String tableName,
     required Set<UuidValue> rowIds,

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

@@ -0,0 +1,193 @@
+part of 'foreign_key_projector.dart';
+
+/// Loads metadata for one projection read phase in one space and transaction.
+/// Field identities are resolved through the scoped rows and registered columns,
+/// so equal domain UUIDs in different tables remain distinct.
+///
+/// Its maps become mutable projection state. Complete loading before applying
+/// pending inserts or authored overlays to them.
+class _ProjectionReader {
+  _ProjectionReader(this._context, this._transaction);
+
+  final CrdtRecorderContext _context;
+  final Transaction _transaction;
+  final rows = <MergeRowKey, _ProjectedForeignKeyRow>{};
+  final fieldIds = <MergeFieldKey, int>{};
+  final attemptedValues = <MergeFieldKey, CrdtDataAttemptedValue>{};
+  final fieldHlcs = <MergeFieldKey, Hlc>{};
+
+  /// Whether this space and these tables hold any attempted claims.
+  /// A false result lets the closure skip attempted-claim queries, but is
+  /// valid only until the next write phase; each projection checks again.
+  Future<bool> hasAttemptedClaims(Set<String> tablesToLoad) async {
+    final spaceId = _context.hlcManagerFor(_transaction).normalizedSpaceId;
+    return tablesToLoad.isNotEmpty &&
+        await CrdtDataAttemptedValue.db.findFirstRow(
+              _context.databaseSession,
+              where: (t) =>
+                  t.field.row.spaceId.equals(spaceId) &
+                  t.field.row.tblId.inSet({
+                    for (final table in tablesToLoad) _context.schema[table]!.$1,
+                  }),
+              transaction: _transaction,
+            ) !=
+            null;
+  }
+
+  /// Loads one closure wave, sharing metadata reads across its tables.
+  /// A null row-id set requests the whole table during a full rebuild.
+  Future<Map<String, Set<UuidValue>>> loadRows({
+    required Map<String, Set<UuidValue>?> requestedRows,
+    required Map<String, Set<String>> columnsByTable,
+  }) async {
+    final tablesById = {
+      for (final entry in requestedRows.entries)
+        if (entry.value == null || entry.value!.isNotEmpty)
+          _context.schema[entry.key]!.$1: entry.key,
+    };
+    if (tablesById.isEmpty) return {};
+
+    final spaceId = _context.hlcManagerFor(_transaction).normalizedSpaceId;
+    final crdtRows = await CrdtDataRow.db.find(
+      _context.databaseSession,
+      where: (t) =>
+          t.spaceId.equals(spaceId) &
+          tablesById.entries
+              .map((entry) {
+                final rowIds = requestedRows[entry.value];
+                final table = t.tblId.equals(entry.key);
+                return rowIds == null ? table : table & t.uuidRowId.inSet(rowIds);
+              })
+              .reduce((left, right) => left | right),
+      include: CrdtDataRow.include(
+        node: CrdtNode.include(),
+        deleted: CrdtDataDeleted.include(node: CrdtNode.include()),
+      ),
+      orderBy: (t) => t.uuidRowId,
+      transaction: _transaction,
+    );
+    final metadataByTable = <String, List<CrdtDataRow>>{};
+    for (final row in crdtRows) {
+      (metadataByTable[tablesById[row.tblId]!] ??= []).add(row);
+    }
+
+    final loadedByTable = <String, Set<UuidValue>>{};
+    for (final tableName in requestedRows.keys) {
+      final tableRows = metadataByTable[tableName];
+      if (tableRows == null) continue;
+      final loadedIds = {for (final row in tableRows) row.uuidRowId};
+      loadedByTable[tableName] = loadedIds;
+      final columns = columnsByTable[tableName] ?? const <String>{};
+      final valuesByRowId = columns.isEmpty
+          ? <UuidValue, Map<String, Object?>>{}
+          : await _context.readDomainColumnValues(
+              tableName,
+              loadedIds,
+              columns.toList(),
+              _transaction,
+            );
+      for (final row in tableRows) {
+        final key = (tableName, row.uuidRowId);
+        rows[key] = (
+          key: key,
+          crdtRow: row,
+          values: Map<String, Object?>.from(valuesByRowId[row.uuidRowId] ?? const {}),
+        );
+      }
+    }
+
+    final columnsById = <int, String>{};
+    final fieldRequests = <({Set<int> rows, Set<int> columns})>[];
+    for (final tableName in metadataByTable.keys) {
+      final columns = _context.schemaColumnIds(
+        tableName,
+        columnsByTable[tableName] ?? const <String>{},
+      );
+      if (columns.isEmpty) continue;
+      columnsById.addAll({for (final entry in columns.entries) entry.value: entry.key});
+      fieldRequests.add((
+        rows: {for (final row in metadataByTable[tableName]!) row.id!},
+        columns: columns.values.toSet(),
+      ));
+    }
+    if (fieldRequests.isEmpty) return loadedByTable;
+
+    final rowKeysById = {
+      for (final row in crdtRows) row.id!: (tablesById[row.tblId]!, row.uuidRowId),
+    };
+    final fields = await CrdtDataField.db.find(
+      _context.databaseSession,
+      where: (t) => fieldRequests
+          .map((request) {
+            return t.rowId.inSet(request.rows) & t.columnId.inSet(request.columns);
+          })
+          .reduce((left, right) => left | right),
+      include: CrdtDataField.include(
+        node: CrdtNode.include(),
+        attemptedValue: CrdtDataAttemptedValue.include(),
+      ),
+      transaction: _transaction,
+    );
+    for (final field in fields) {
+      final rowKey = rowKeysById[field.rowId]!;
+      final fieldKey = (rowKey.$1, rowKey.$2, columnsById[field.columnId]!);
+      fieldIds[fieldKey] = field.id!;
+      fieldHlcs[fieldKey] = field.hlc;
+      if (field.attemptedValue case final attempted?) {
+        attemptedValues[fieldKey] = attempted;
+      }
+    }
+    return loadedByTable;
+  }
+}
+
+/// Collects the required claim lookups for one closure wave by table.
+/// Traversal selects the claims; this collector handles grouped execution.
+/// Each composite claim remains independent when the queries are combined.
+class _ProjectionClaimReads {
+  _ProjectionClaimReads(
+    this._context,
+    this._transaction, {
+    required this._includeAttemptedClaims,
+  });
+
+  final CrdtRecorderContext _context;
+  final Transaction _transaction;
+  final bool _includeAttemptedClaims;
+  final _byTable = <String, List<Map<String, Set<Object?>>>>{};
+
+  void add(String tableName, Map<String, Set<Object?>> claim) {
+    (_byTable[tableName] ??= []).add(claim);
+  }
+
+  Future<void> enqueueMatches(
+    void Function(String tableName, Iterable<UuidValue> ids) enqueue,
+  ) async {
+    for (final MapEntry(key: tableName, value: claims) in _byTable.entries) {
+      enqueue(
+        tableName,
+        await _context.findDomainRowIdsWhereAnyColumnsIn(
+          tableName: tableName,
+          alternatives: claims,
+          transaction: _transaction,
+        ),
+      );
+      if (!_includeAttemptedClaims) continue;
+
+      enqueue(
+        tableName,
+        await _context.findRowIdsHoldingAnyAttemptedValues(
+          tableName: tableName,
+          alternatives: [
+            for (final claim in claims)
+              (
+                columnNames: claim.keys.toSet(),
+                values: {for (final values in claim.values) ...values},
+              ),
+          ],
+          transaction: _transaction,
+        ),
+      );
+    }
+  }
+}

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

@@ -531,22 +531,43 @@ WHERE r."spaceId" = $spaceId AND r."tblId" = $childTableId
     required Transaction transaction,
   }) async {
     if (columnNames.isEmpty || values.isEmpty) return const {};
+    return findRowIdsHoldingAnyAttemptedValues(
+      tableName: tableName,
+      alternatives: [(columnNames: columnNames, values: values)],
+      transaction: transaction,
+    );
+  }
 
+  /// Matches the union of claim lookups without mixing their columns and values.
+  Future<Set<UuidValue>> findRowIdsHoldingAnyAttemptedValues({
+    required String tableName,
+    required List<({Set<String> columnNames, Set<Object?> values})> alternatives,
+    required Transaction transaction,
+  }) async {
     final (tableId, columns) = schema[tableName]!;
-    final columnIds = {
+    final predicates = <String>{};
-      for (final columnName in columnNames) ?columns[columnName]?.id,
+    for (final alternative in alternatives) {
-    };
+      final columnIds = {
-    if (columnIds.isEmpty) return const {};
+        for (final columnName in alternative.columnNames) ?columns[columnName]?.id,
+      };
+      if (columnIds.isEmpty) continue;
 
-    final encodedValues = {
+      final encodedValues = {
-      for (final value in values)
+        for (final value in alternative.values)
-        if (value != null)
+          if (value != null)
-          ValueEncoder.instance.encodeColumnValue(
+            ValueEncoder.instance.encodeColumnValue(
-            CrdtDataAttemptedValue.t.value,
+              CrdtDataAttemptedValue.t.value,
-            Protocol().dynamicFieldToJson(value),
+              Protocol().dynamicFieldToJson(value),
-          ),
+            ),
-    };
+      };
-    if (encodedValues.isEmpty) return const {};
+      if (encodedValues.isEmpty) continue;
+
+      predicates.add(
+        '(f."columnId" IN (${columnIds.join(', ')}) '
+        'AND a."value" IN (${encodedValues.join(', ')}))',
+      );
+    }
+    if (predicates.isEmpty) return const {};
 
     final spaceId = hlcManagerFor(transaction).normalizedSpaceId;
     final result = await database.unsafeQuery(
@@ -557,8 +578,7 @@ JOIN "crdt_data_fields" f ON f."id" = a."fieldId"
 JOIN "crdt_data_rows" r ON r."id" = f."rowId"
 WHERE r."spaceId" = $spaceId
   AND r."tblId" = $tableId
-  AND f."columnId" IN (${columnIds.join(', ')})
+  AND (${predicates.join(' OR ')})
-  AND a."value" IN (${encodedValues.join(', ')})
 ''',
       transaction: transaction,
     );
@@ -618,22 +638,38 @@ WHERE r."spaceId" = $spaceId
     required String tableName,
     required Map<String, Set<Object?>> valuesByColumn,
     required Transaction transaction,
-  }) async {
+  }) => findDomainRowIdsWhereAnyColumnsIn(
-    if (valuesByColumn.isEmpty) return const {};
+    tableName: tableName,
-    if (valuesByColumn.values.any((values) => values.isEmpty)) return const {};
+    alternatives: [valuesByColumn],
+    transaction: transaction,
+  );
 
-    final predicates = valuesByColumn.entries
+  /// Matches each composite claim independently and unions the matching row ids.
-        .map(
+  Future<Set<UuidValue>> findDomainRowIdsWhereAnyColumnsIn({
-          (entry) =>
+    required String tableName,
-              '"${entry.key.escapeIdentifier()}" IN (${entry.value.sqlLiteralList()})',
+    required List<Map<String, Set<Object?>>> alternatives,
-        )
+    required Transaction transaction,
-        .join(') AND (');
+  }) async {
+    final predicates = <String>{};
+    for (final valuesByColumn in alternatives) {
+      if (valuesByColumn.isEmpty) continue;
+      if (valuesByColumn.values.any((values) => values.isEmpty)) continue;
+
+      final claim = valuesByColumn.entries
+          .map(
+            (entry) =>
+                '"${entry.key.escapeIdentifier()}" IN (${entry.value.sqlLiteralList()})',
+          )
+          .join(') AND (');
+      predicates.add('(($claim))');
+    }
+    if (predicates.isEmpty) return const {};
 
     final result = await database.unsafeQuery(
       '''
 SELECT "id"
 FROM "${tableName.escapeIdentifier()}"
-WHERE ($predicates)
+WHERE ${predicates.join(' OR ')}
 ''',
       transaction: transaction,
     );

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

@@ -0,0 +1,154 @@
+part of 'recorder.dart';
+
+/// Owns pending metadata for one merge and the boundaries that make it readable.
+/// Context fields advance immediately. Existing field writes wait until an
+/// insert, delete, pending insert's next operation, or successful completion.
+/// Insert attempts are flushed before another operation touches their row.
+class _MergeWriteBatch {
+  _MergeWriteBatch({
+    required this._session,
+    required this._projector,
+    required this._fields,
+    required this._transaction,
+  });
+
+  final DatabaseSession _session;
+  final CrdtForeignKeyProjector _projector;
+  final Map<MergeFieldKey, CrdtDataField> _fields;
+  final Transaction _transaction;
+  final _insertAttempts = _PendingInsertAttempts();
+  final _fieldUpdates = <int, CrdtDataField>{};
+
+  /// Successful completion persists all metadata before projection can run.
+  /// Failure leaves rollback to the enclosing transaction, without flushing.
+  ///
+  /// Update callbacks must use the merge context's current clocks and record
+  /// changes to existing fields through [recordFieldUpdate].
+  Future<void> apply(
+    Iterable<CrdtMergeChange> operations,
+    Future<void> Function(CrdtMergeChange operation) applyOperation,
+  ) async {
+    for (final operation in operations) {
+      final needsInsertMetadata = _insertAttempts.holds((
+        operation.tableName,
+        operation.uuidRowId,
+      ));
+
+      // Reinserts can advance field clocks immediately. Flush older queued
+      // updates first so the final flush cannot overwrite those newer clocks.
+      if (operation is! CrdtMergeUpdate || needsInsertMetadata) {
+        await _flushFieldUpdates();
+      }
+      if (needsInsertMetadata) {
+        await _flushInsertAttempts();
+      }
+
+      await applyOperation(operation);
+    }
+
+    await _flushFieldUpdates();
+    await _flushInsertAttempts();
+  }
+
+  void recordFieldUpdate(CrdtDataField field) {
+    _fieldUpdates[field.id!] = field;
+  }
+
+  void recordInsertAttempts({
+    required String tableName,
+    required UuidValue rowId,
+    required CrdtNode node,
+    required ProjectionAttemptsByField attempts,
+  }) {
+    _insertAttempts.add(
+      tableName: tableName,
+      rowId: rowId,
+      node: node,
+      attempts: attempts,
+    );
+  }
+
+  Future<void> _flushFieldUpdates() async {
+    if (_fieldUpdates.isEmpty) return;
+
+    await CrdtDataField.db.update(
+      _session,
+      _fieldUpdates.values.toList(),
+      columns: (t) => [t.nodeId, t.hlcDatetime, t.hlcCounter],
+      transaction: _transaction,
+      noReturn: true,
+    );
+    _fieldUpdates.clear();
+  }
+
+  Future<void> _flushInsertAttempts() async {
+    for (final group in _insertAttempts.take()) {
+      await _projector.recordInsertAttempts(
+        group.tableName,
+        group.rowIds,
+        _transaction,
+        group.attempts,
+        mergeCache: (fields: _fields, node: group.node),
+      );
+    }
+  }
+}
+
+/// Field metadata for merged inserts, waiting to be written as one batch.
+///
+/// Groups rows by table and author: cached field clocks need the node that
+/// authored each group.
+class _PendingInsertAttempts {
+  final Map<(String, int), ({Set<UuidValue> rowIds, CrdtNode node})> _byGroup = {};
+  final Map<String, ProjectionAttemptsByField> _attemptsByTable = {};
+  final Set<MergeRowKey> _rows = {};
+
+  /// Whether metadata for [rowKey] is still unwritten.
+  bool holds(MergeRowKey rowKey) => _rows.contains(rowKey);
+
+  void add({
+    required String tableName,
+    required UuidValue rowId,
+    required CrdtNode node,
+    required ProjectionAttemptsByField attempts,
+  }) {
+    _byGroup
+        .putIfAbsent(
+          (tableName, node.id!),
+          () => (rowIds: <UuidValue>{}, node: node),
+        )
+        .rowIds
+        .add(rowId);
+    _rows.add((tableName, rowId));
+    if (attempts.isEmpty) return;
+    _attemptsByTable
+        .putIfAbsent(tableName, () => <MergeFieldKey, ProjectionAttempt>{})
+        .addAll(attempts);
+  }
+
+  /// Empties the collector and returns what it held.
+  List<
+    ({
+      String tableName,
+      Set<UuidValue> rowIds,
+      CrdtNode node,
+      ProjectionAttemptsByField attempts,
+    })
+  >
+  take() {
+    final groups = [
+      for (final MapEntry(key: key, value: group) in _byGroup.entries)
+        (
+          tableName: key.$1,
+          rowIds: group.rowIds,
+          node: group.node,
+          attempts:
+              _attemptsByTable[key.$1] ?? const <MergeFieldKey, ProjectionAttempt>{},
+        ),
+    ];
+    _byGroup.clear();
+    _attemptsByTable.clear();
+    _rows.clear();
+    return groups;
+  }
+}

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

@@ -21,6 +21,7 @@ import 'schema.dart';
 import 'unique_index_utils.dart';
 
 part 'merge.dart';
+part 'merge_write_batch.dart';
 
 typedef _CrdtSchema = Map<String, (int, Map<String, CrdtSchemaColumn>)>;
 

test/serverpod_offline_sync_test_server/test/integration/merge/metadata_batch_test.dart

@@ -0,0 +1,360 @@
+import 'dart:convert';
+
+import 'package:serverpod/serverpod.dart' show DatabaseQueryException;
+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/sync_topology.dart';
+
+void main() {
+  initTestClientSession(createSessionPerTest: false);
+
+  group('Given an edited city and a town arriving with another parent,', () {
+    late SyncNode node;
+    late City originalParent;
+    late City replacementParent;
+    late Town town;
+    late Hlc originalNameClock;
+    late CrdtMergeUpdate finalName;
+    late CrdtMergeUpdate finalParent;
+    late CrdtMergeSet changes;
+
+    setUpAll(() async {
+      node = await syncNode(await createAdditionalTestSession(), testSyncTables);
+      originalParent = City(id: const Uuid().v7obj(), name: 'original parent');
+      replacementParent = City(id: const Uuid().v7obj(), name: 'replacement parent');
+      town = Town(id: const Uuid().v7obj(), name: 'town', cityId: originalParent.id);
+      await node.offlineSync.db.transactionForUser(testCrdtUserId, (tx) async {
+        await City.db.insert(node.offlineSync, [
+          originalParent,
+          replacementParent,
+        ], transaction: tx);
+        await City.db.updateRow(
+          node.offlineSync,
+          replacementParent.copyWith(name: 'locally edited'),
+          columns: (t) => [t.name],
+          transaction: tx,
+        );
+      });
+      originalNameClock = (await _fieldClock(
+        node,
+        'city',
+        replacementParent.id!,
+        'name',
+      ))!;
+      final remote = Hlc(
+        originalNameClock.datetime.add(const Duration(seconds: 1)),
+        0,
+        const Uuid().v7obj(),
+      );
+      finalName = _update(
+        'city',
+        replacementParent.id!,
+        'name',
+        'final name',
+        Hlc(remote.datetime, 1, remote.nodeId),
+      );
+      finalParent = _update(
+        'town',
+        town.id!,
+        'cityId',
+        replacementParent.id,
+        Hlc(remote.datetime, 3, remote.nodeId),
+      );
+      changes = [
+        _update('city', replacementParent.id!, 'name', 'intermediate name', remote),
+        finalName,
+        CrdtMergeInsert(
+          uuidSpaceId: testCrdtUserId,
+          tableName: 'town',
+          uuidRowId: town.id!,
+          uuidNodeId: remote.nodeId,
+          hlcDatetime: remote.datetime,
+          hlcCounter: remote.counter + 2,
+          data: town,
+        ),
+        finalParent,
+        CrdtMergeDelete(
+          uuidSpaceId: testCrdtUserId,
+          tableName: 'city',
+          uuidRowId: originalParent.id!,
+          uuidNodeId: remote.nodeId,
+          hlcDatetime: remote.datetime,
+          hlcCounter: remote.counter + 4,
+          clFlag: 2,
+          reason: CrdtDataDeletedReason.userDelete,
+        ),
+      ];
+    });
+
+    group(
+      'when one merge edits the city, inserts and reparents the town, and deletes its old parent,',
+      () {
+        late City? mergedCity;
+        late City? deletedParent;
+        late Town? mergedTown;
+        late Hlc? nameClock;
+        late Hlc? parentClock;
+        late String firstExport;
+        late String replayedExport;
+
+        setUpAll(() async {
+          await node.offlineSync.db.mergeChanges(changes, spaceId: testCrdtUserId);
+          mergedCity = await City.db.findById(node.offlineSync, replacementParent.id!);
+          deletedParent = await City.db.findById(node.offlineSync, originalParent.id!);
+          mergedTown = await Town.db.findById(node.offlineSync, town.id!);
+          nameClock = await _fieldClock(node, 'city', replacementParent.id!, 'name');
+          parentClock = await _fieldClock(node, 'town', town.id!, 'cityId');
+          firstExport = await _export(node);
+
+          await node.offlineSync.db.mergeChanges(changes, spaceId: testCrdtUserId);
+          replayedExport = await _export(node);
+        });
+
+        test(
+          'then the repeated city edit retains its newest value and authored clock.',
+          () {
+            expect(originalNameClock < finalName.hlc, isTrue);
+            expect(mergedCity?.name, 'final name');
+            expect(nameClock, finalName.hlc);
+          },
+        );
+
+        test(
+          'then the inserted town retains its new parent and authored clock after the old parent is deleted.',
+          () {
+            expect(deletedParent, isNull);
+            expect(mergedTown, isNotNull);
+            expect(mergedTown!.cityId, replacementParent.id);
+            expect(parentClock, finalParent.hlc);
+          },
+        );
+
+        test('then replay leaves the exported authored history byte stable.', () {
+          expect(replayedExport, firstExport);
+        });
+      },
+    );
+  });
+
+  group('Given a city with a local name edit,', () {
+    late SyncNode node;
+    late City city;
+    late Hlc localNameClock;
+
+    setUpAll(() async {
+      node = await syncNode(await createAdditionalTestSession(), testSyncTables);
+      city = City(id: const Uuid().v7obj(), name: 'original');
+      await node.offlineSync.db.transactionForUser(testCrdtUserId, (tx) async {
+        await City.db.insertRow(node.offlineSync, city, transaction: tx);
+        await City.db.updateRow(
+          node.offlineSync,
+          city.copyWith(name: 'locally edited'),
+          columns: (t) => [t.name],
+          transaction: tx,
+        );
+      });
+      localNameClock = (await _fieldClock(node, 'city', city.id!, 'name'))!;
+    });
+
+    group(
+      'when one batch edits and reinserts it before an intermediate edit arrives,',
+      () {
+        late Hlc reinsertClock;
+        late City? merged;
+        late Hlc? mergedClock;
+        late City? afterLateEdit;
+        late Hlc? afterLateClock;
+
+        setUpAll(() async {
+          final editClock = Hlc(
+            localNameClock.datetime.add(const Duration(seconds: 1)),
+            0,
+            const Uuid().v7obj(),
+          );
+          reinsertClock = Hlc(editClock.datetime, 2, const Uuid().v7obj());
+          await node.offlineSync.db.mergeChanges([
+            _update('city', city.id!, 'name', 'edited', editClock),
+            CrdtMergeInsert(
+              uuidSpaceId: testCrdtUserId,
+              tableName: 'city',
+              uuidRowId: city.id!,
+              uuidNodeId: reinsertClock.nodeId,
+              hlcDatetime: reinsertClock.datetime,
+              hlcCounter: reinsertClock.counter,
+              data: city.copyWith(name: 'reinserted'),
+            ),
+          ], spaceId: testCrdtUserId);
+          merged = await City.db.findById(node.offlineSync, city.id!);
+          mergedClock = await _fieldClock(node, 'city', city.id!, 'name');
+
+          await node.offlineSync.db.mergeChanges([
+            _update(
+              'city',
+              city.id!,
+              'name',
+              'late edit',
+              Hlc(editClock.datetime, 1, editClock.nodeId),
+            ),
+          ], spaceId: testCrdtUserId);
+          afterLateEdit = await City.db.findById(node.offlineSync, city.id!);
+          afterLateClock = await _fieldClock(node, 'city', city.id!, 'name');
+        });
+
+        test('then the batch preserves the reinsertion value and author clock.', () {
+          expect(merged?.name, 'reinserted');
+          expect(mergedClock, reinsertClock);
+        });
+
+        test('then the intermediate edit cannot overwrite the newer reinsertion.', () {
+          expect(afterLateEdit?.name, 'reinserted');
+          expect(afterLateClock, reinsertClock);
+        });
+      },
+    );
+  });
+
+  group('Given two cities with existing name edits,', () {
+    late SyncNode node;
+    late City first;
+    late City second;
+    late Hlc firstClock;
+    late Hlc secondClock;
+    late Hlc incoming;
+
+    setUpAll(() async {
+      node = await syncNode(await createAdditionalTestSession(), testSyncTables);
+      first = City(id: const Uuid().v7obj(), name: 'first');
+      second = City(id: const Uuid().v7obj(), name: 'second');
+      await node.offlineSync.db.transactionForUser(testCrdtUserId, (tx) async {
+        await City.db.insert(node.offlineSync, [first, second], transaction: tx);
+        await City.db.updateRow(
+          node.offlineSync,
+          first.copyWith(name: 'first edited'),
+          columns: (t) => [t.name],
+          transaction: tx,
+        );
+        await City.db.updateRow(
+          node.offlineSync,
+          second.copyWith(name: 'second edited'),
+          columns: (t) => [t.name],
+          transaction: tx,
+        );
+      });
+      firstClock = (await _fieldClock(node, 'city', first.id!, 'name'))!;
+      secondClock = (await _fieldClock(node, 'city', second.id!, 'name'))!;
+      incoming = Hlc(
+        secondClock.datetime.add(const Duration(seconds: 1)),
+        0,
+        const Uuid().v7obj(),
+      );
+    });
+
+    group('when a merge contains a valid edit followed by an invalid null name,', () {
+      Object? failure;
+      late List<String> rolledBackNames;
+      late List<Hlc?> rolledBackClocks;
+      late CrdtMergeUpdate recovery;
+      late City? recoveredCity;
+      late Hlc? recoveredClock;
+
+      setUpAll(() async {
+        try {
+          await node.offlineSync.db.mergeChanges([
+            _update(
+              'city',
+              first.id!,
+              'name',
+              'must roll back',
+              Hlc(incoming.datetime, 2, incoming.nodeId),
+            ),
+            _update(
+              'city',
+              second.id!,
+              'name',
+              null,
+              Hlc(incoming.datetime, 3, incoming.nodeId),
+            ),
+          ], spaceId: testCrdtUserId);
+        } on Exception catch (error) {
+          failure = error;
+        }
+        rolledBackNames = [
+          (await City.db.findById(node.offlineSync, first.id!))!.name,
+          (await City.db.findById(node.offlineSync, second.id!))!.name,
+        ];
+        rolledBackClocks = [
+          await _fieldClock(node, 'city', first.id!, 'name'),
+          await _fieldClock(node, 'city', second.id!, 'name'),
+        ];
+
+        recovery = _update('city', first.id!, 'name', 'recovered', incoming);
+        await node.offlineSync.db.mergeChanges([recovery], spaceId: testCrdtUserId);
+        recoveredCity = await City.db.findById(node.offlineSync, first.id!);
+        recoveredClock = await _fieldClock(node, 'city', first.id!, 'name');
+      });
+
+      test('then neither row retains values or clocks from the rejected merge.', () {
+        expect(failure, isA<DatabaseQueryException>());
+        expect(rolledBackNames, ['first edited', 'second edited']);
+        expect(rolledBackClocks, [firstClock, secondClock]);
+      });
+
+      test(
+        'then a subsequent merge can accept an edit older than the rejected edits.',
+        () {
+          expect(recoveredCity?.name, 'recovered');
+          expect(recoveredClock, recovery.hlc);
+        },
+      );
+    });
+  });
+}
+
+CrdtMergeUpdate _update(
+  String table,
+  UuidValue row,
+  String column,
+  Object? value,
+  Hlc clock,
+) {
+  return CrdtMergeUpdate(
+    uuidSpaceId: testCrdtUserId,
+    tableName: table,
+    uuidRowId: row,
+    uuidNodeId: clock.nodeId,
+    hlcDatetime: clock.datetime,
+    hlcCounter: clock.counter,
+    columnName: column,
+    value: value,
+  );
+}
+
+Future<Hlc?> _fieldClock(
+  SyncNode node,
+  String table,
+  UuidValue row,
+  String column,
+) async {
+  final field = await CrdtDataField.db.findFirstRow(
+    node.offlineSync,
+    where: (t) =>
+        t.row.tbl.name.equals(table) &
+        t.row.uuidRowId.equals(row) &
+        t.column.name.equals(column),
+    include: CrdtDataField.include(node: CrdtNode.include()),
+  );
+  return field?.hlc;
+}
+
+Future<String> _export(SyncNode node) async {
+  final changes = await node.sync
+      .collectPendingChanges(
+        node.raw,
+        checkpointsBySpaceUuid: {testCrdtUserId: const []},
+      )
+      .toList();
+  return jsonEncode(changes);
+}

test/serverpod_offline_sync_test_server/test/integration/sync/sync_projection_wave_identity_test.dart

@@ -0,0 +1,202 @@
+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/sync_topology.dart';
+
+void main() {
+  initTestClientSession(createSessionPerTest: false);
+
+  group(
+    'Given a town and organization sharing a UUID with their missing parents,',
+    () {
+      late SyncNode author;
+      late SyncNode observer;
+      late UuidValue sharedId;
+      late Town town;
+      late Organization organization;
+      late Hlc authored;
+      late Map<(String, String), Hlc> originalClocks;
+
+      setUpAll(() async {
+        author = await syncNode(
+          await createAdditionalTestSession(),
+          testSyncTables,
+        );
+        observer = await syncNode(
+          await createAdditionalTestSession(),
+          testSyncTables,
+        );
+        sharedId = const Uuid().v7obj();
+        town = Town(
+          id: sharedId,
+          name: 'town',
+          cityId: sharedId,
+          mayorId: sharedId,
+        );
+        organization = Organization(
+          id: sharedId,
+          name: 'organization',
+          cityId: sharedId,
+        );
+        authored = Hlc(DateTime.utc(2026, 9, 1), 0, const Uuid().v7obj());
+        await author.offlineSync.db.mergeChanges([
+          CrdtMergeInsert(
+            uuidSpaceId: testCrdtUserId,
+            tableName: Town.t.tableName,
+            uuidRowId: sharedId,
+            uuidNodeId: authored.nodeId,
+            hlcDatetime: authored.datetime,
+            hlcCounter: authored.counter,
+            data: town,
+          ),
+          CrdtMergeInsert(
+            uuidSpaceId: testCrdtUserId,
+            tableName: Organization.t.tableName,
+            uuidRowId: sharedId,
+            uuidNodeId: authored.nodeId,
+            hlcDatetime: authored.datetime,
+            hlcCounter: authored.counter + 1,
+            data: organization,
+          ),
+        ], spaceId: testCrdtUserId);
+        originalClocks = await _dependentClocks(author, sharedId);
+      });
+
+      group('when the city and mayor arrive together and synchronize,', () {
+        late Town? authorTown;
+        late Town? observerTown;
+        late Organization? authorOrganization;
+        late Organization? observerOrganization;
+        late Person? authorMayor;
+        late Person? observerMayor;
+        late Map<(String, String), Hlc> authorClocks;
+        late Map<(String, String), Hlc> observerClocks;
+        late int authorAttempts;
+        late int observerAttempts;
+
+        setUpAll(() async {
+          final city = City(id: sharedId, name: 'city');
+          final mayor = Person(
+            id: sharedId,
+            name: 'mayor',
+            cityId: sharedId,
+            organizationId: sharedId,
+          );
+          await author.offlineSync.db.mergeChanges([
+            CrdtMergeInsert(
+              uuidSpaceId: testCrdtUserId,
+              tableName: City.t.tableName,
+              uuidRowId: sharedId,
+              uuidNodeId: authored.nodeId,
+              hlcDatetime: authored.datetime,
+              hlcCounter: authored.counter + 2,
+              data: city,
+            ),
+            CrdtMergeInsert(
+              uuidSpaceId: testCrdtUserId,
+              tableName: Person.t.tableName,
+              uuidRowId: sharedId,
+              uuidNodeId: authored.nodeId,
+              hlcDatetime: authored.datetime,
+              hlcCounter: authored.counter + 3,
+              data: mayor,
+            ),
+          ], spaceId: testCrdtUserId);
+          await syncWithServer(author, observer);
+
+          authorTown = await Town.db.findById(author.offlineSync, sharedId);
+          observerTown = await Town.db.findById(observer.offlineSync, sharedId);
+          authorOrganization = await Organization.db.findById(
+            author.offlineSync,
+            sharedId,
+          );
+          observerOrganization = await Organization.db.findById(
+            observer.offlineSync,
+            sharedId,
+          );
+          authorMayor = await Person.db.findById(author.offlineSync, sharedId);
+          observerMayor = await Person.db.findById(
+            observer.offlineSync,
+            sharedId,
+          );
+          authorClocks = await _dependentClocks(author, sharedId);
+          observerClocks = await _dependentClocks(observer, sharedId);
+          authorAttempts = await CrdtDataAttemptedValue.db.count(
+            author.offlineSync,
+          );
+          observerAttempts = await CrdtDataAttemptedValue.db.count(
+            observer.offlineSync,
+          );
+        });
+
+        test(
+          'then both replicas restore each reference to its own parent table.',
+          () {
+            for (final value in [authorTown, observerTown]) {
+              expect(value, isNotNull);
+              expect(
+                (value!.name, value.cityId, value.mayorId),
+                ('town', sharedId, sharedId),
+              );
+            }
+            for (final value in [authorOrganization, observerOrganization]) {
+              expect(value, isNotNull);
+              expect((value!.name, value.cityId), ('organization', sharedId));
+            }
+            for (final value in [authorMayor, observerMayor]) {
+              expect(value, isNotNull);
+              expect(
+                (value!.name, value.cityId, value.organizationId),
+                ('mayor', sharedId, sharedId),
+              );
+            }
+          },
+        );
+
+        test(
+          'then restoring each dependency preserves its authored field clocks.',
+          () {
+            expect(originalClocks.keys.toSet(), {
+              ('town', 'cityId'),
+              ('town', 'mayorId'),
+              ('organization', 'cityId'),
+            });
+            expect(authorClocks, originalClocks);
+            expect(observerClocks, originalClocks);
+          },
+        );
+
+        test(
+          'then neither replica retains a repaired-away parent reference.',
+          () {
+            expect(authorAttempts, 0);
+            expect(observerAttempts, 0);
+          },
+        );
+      });
+    },
+  );
+}
+
+Future<Map<(String, String), Hlc>> _dependentClocks(
+  SyncNode node,
+  UuidValue id,
+) async {
+  final fields = await CrdtDataField.db.find(
+    node.offlineSync,
+    where: (t) =>
+        t.row.uuidRowId.equals(id) &
+        t.row.tbl.name.inSet({'town', 'organization'}) &
+        t.column.name.inSet({'cityId', 'mayorId'}),
+    include: CrdtDataField.include(
+      row: CrdtDataRow.include(tbl: CrdtSchemaTable.include()),
+      column: CrdtSchemaColumn.include(),
+      node: CrdtNode.include(),
+    ),
+  );
+  return {
+    for (final field in fields) (field.row!.tbl!.name, field.column!.name): field.hlc,
+  };
+}