perf: Read the domain rows a collection sends in batches

Collecting pending changes read each change's domain row on its own: one
statement per insert, and one per update for the column it carries. A
device that had never synced a space of about 100,600 field rows made the
server run some 60,000 of them in the collection's one transaction, 143
to 198 s, and since nothing is sent before everything is collected, the
client's connection closed first and the next attempt started over.

Before walking the inserts, and again before walking the updates, the
collection now reads the rows they name ahead, 500 to a statement per
table and space, with the same column expressions, in the same snapshot:
SELECT those columns WHERE "spaceId" = the space AND "id" IN (the rows).
The raw rows are kept for the collection and each found row's owner is
recorded in the owner cache; an insert decodes its row and restores its
attempted values from them as before, and an update takes its column
from them, so the updates of a row collected as an insert read nothing
more. A row the read-ahead didn't find (missing, or owned by another
space) takes the per-row path unchanged, which tells those cases apart;
both end the collection with an integrity violation, so they are left
unbatched. Deletes still look up their owners row by row, but find in
the owner cache every row the same collection read ahead. Chunking and
streaming are unchanged.

Measured on embedded PostgreSQL, a first collection of 30,040 rows and
90,054 updates (90,067 field rows) took 51.5 to 53.6 s before and takes
16.9 to 17.1 s now, its 120,194 changes identical on the wire, byte for
byte, to what the per-row path produces from the same database. On SQLite,
3,594 changes of a fixture with structured columns, blobs, deletes and
attempted foreign keys on inserts and updates are byte for byte the same
before and after. The new test collects 1,200 rows, 600 of them renamed,
for a fresh peer (1,800 reads of the table before, 3 now), and 600 renames
since a peer's checkpoint (600 reads before, 2 now), and checks every
payload against the author's rows and the peer's rows after the merge.
The core, server, integration and simulation suites pass, but for one
cross-space seed with delta delivery (DST_SEED_BASE=1794152224), whose
local update of an organization finds no row before this change as after.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Signed-off-by: Tizian 「ティツィアーン」 <raisondetredev@pm.me>

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

@@ -120,6 +120,7 @@ class OfflineSyncDatabase implements Database {
     UuidValue? userId,
     bool once = false,
     OfflineSyncOnMergeSuccess? onMergeSuccess,
+    OfflineSyncOnInboundMerge? onInboundMerge,
   }) async* {
     await _ensureInitialized();
     final effectiveUserId = await _requireUserId(userId);
@@ -129,6 +130,7 @@ class OfflineSyncDatabase implements Database {
       inbound: inbound,
       once: once,
       onMergeSuccess: onMergeSuccess,
+      onInboundMerge: onInboundMerge,
       mode: mode,
     );
   }

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

@@ -36,7 +36,7 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
     // Nothing in this batch touches a foreign key or a unique index, so no
     // projection decision can change: skip both passes instead of loading
     // state for tables that cannot produce a candidate or a conflict.
-    final mayAffectProjection = _foreignKeyProjector.mergeOperationsMayAffectProjection(
+    var mayAffectProjection = _foreignKeyProjector.mergeOperationsMayAffectProjection(
       operations,
     );
     final authoredOverlays = <MergeFieldKey, Object?>{};
@@ -48,6 +48,14 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
     );
     final nodesByUuid = remoteNodes.nodesByUuid;
     final context = await _loadMergeContext(mergeSet, transaction);
+    // New rows whose parents are there and which nothing refers to yet (a
+    // rating, an edit, a tag given, a card added) can't change a projection
+    // decision either, though they carry foreign keys: proven so, neither
+    // pass runs, and the batch still resolves their parents in one go.
+    final unchangedByInserts =
+        mayAffectProjection &&
+        await _mergeLeavesProjectionUnchanged(operations, context, transaction);
+    if (unchangedByInserts) mayAffectProjection = false;
 
     // Deletes carry no authored overlay but still change visibility, so the
     // seed is every touched row, not just the ones with overlays.
@@ -59,7 +67,7 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
       touchedRows.add((operation.tableName, operation.uuidRowId));
     }
 
-    final batch = mayAffectProjection
+    final batch = mayAffectProjection || unchangedByInserts
         ? await _planBatchInserts(
             operations,
             nodesByUuid,
@@ -67,6 +75,7 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
             touchedTables,
             touchedRows,
             transaction,
+            plan: mayAffectProjection,
           )
         : (plan: null, pending: const <PendingProjectionRow>[]);
     // Every merged insert has to know whether the rows its foreign keys name
@@ -83,13 +92,28 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
       session: _session,
       projector: _foreignKeyProjector,
       fields: context.fields,
+      rows: context.rows,
       transaction: transaction,
     );
     final deferredDeletes = <CrdtMergeDelete>[];
+    // Without the pass, new rows nothing else in the batch touches are
+    // written many to a statement instead of one by one in savepoints.
+    final writtenInChunks = mayAffectProjection
+        ? const <MergeRowKey>{}
+        : await _applyNewRowInsertsInChunks(
+            operations,
+            nodesByUuid,
+            context,
+            writes,
+            presence,
+            transaction,
+          );
 
     await writes.apply(
       operations.where(
-        (operation) => _context.isCrdtTrackedTableName(operation.tableName),
+        (operation) =>
+            _context.isCrdtTrackedTableName(operation.tableName) &&
+            !writtenInChunks.contains((operation.tableName, operation.uuidRowId)),
       ),
       (operation) async {
         switch (operation) {
@@ -362,8 +386,11 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
     MergeContext context,
     Set<String> seedTables,
     Set<MergeRowKey> seedRows,
-    Transaction transaction,
-  ) async {
+    Transaction transaction, {
+    // Whether to plan them: without, only the rows are listed, for their
+    // parents to be resolved together.
+    bool plan = true,
+  }) async {
     final pending = <PendingProjectionRow>[];
     final seenRowKeys = <MergeRowKey>{};
 
@@ -391,7 +418,7 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
         hidden: false,
       ));
     }
-    if (pending.isEmpty) return (plan: null, pending: pending);
+    if (pending.isEmpty || !plan) return (plan: null, pending: pending);
 
     return (
       plan: await _foreignKeyProjector.project(
@@ -404,6 +431,300 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
     );
   }
 
+  /// Whether this batch leaves every projection decision as it is though it
+  /// touches foreign keys, besides updating columns no projection reads: it
+  /// only inserts new rows, each once, proven
+  /// ([CrdtForeignKeyProjector.canLeaveProjectionUnchangedByMergedInserts]),
+  /// or only deletes rows, none restoring one, proven
+  /// ([CrdtForeignKeyProjector.canLeaveProjectionUnchangedByMergedDeletes]).
+  /// A batch doing both, an insert onto a row there already, or a restore
+  /// takes the full pass.
+  Future<bool> _mergeLeavesProjectionUnchanged(
+    List<CrdtMergeChange> operations,
+    MergeContext context,
+    Transaction transaction,
+  ) async {
+    final insertsByTable = <String, Map<UuidValue, Map<String, Object?>>>{};
+    final positions = <MergeRowKey, int>{};
+    final deletesByTable = <String, Set<UuidValue>>{};
+    for (final operation in operations) {
+      if (!_context.isCrdtTrackedTableName(operation.tableName)) continue;
+      switch (operation) {
+        case final CrdtMergeDelete delete:
+          // An even flag deletes, an odd one restores.
+          if (delete.clFlag.isOdd || !delete.reason.isSynced) return false;
+          (deletesByTable[delete.tableName] ??= {}).add(delete.uuidRowId);
+        case final CrdtMergeUpdate update:
+          if (_foreignKeyProjector.needsProjection(update.tableName, {
+            update.columnName,
+          })) {
+            return false;
+          }
+        case final CrdtMergeInsert insert:
+          if (context.rows[(insert.tableName, insert.uuidRowId)] != null) {
+            return false;
+          }
+          final rows = insertsByTable[insert.tableName] ??= {};
+          if (rows.containsKey(insert.uuidRowId)) return false;
+          rows[insert.uuidRowId] = _sanitizeMergeRowData(
+            insert.tableName,
+            insert.databaseColumns,
+          );
+          positions[(insert.tableName, insert.uuidRowId)] = positions.length;
+      }
+    }
+    if (insertsByTable.isNotEmpty && deletesByTable.isNotEmpty) return false;
+    if (deletesByTable.isNotEmpty) {
+      return _foreignKeyProjector.canLeaveProjectionUnchangedByMergedDeletes(
+        deletesByTable,
+        transaction,
+      );
+    }
+    return _foreignKeyProjector.canLeaveProjectionUnchangedByMergedInserts(
+      insertsByTable,
+      positions,
+      transaction,
+    );
+  }
+
+  /// Writes the batch's new rows that nothing else in it touches, many to a
+  /// statement, and returns their keys for the per-row pass to skip.
+  ///
+  /// Only for a batch the projection pass skips, where each merged insert
+  /// writes its row as planned, one by one in two savepoints and two
+  /// statements of its own: nearly all the time of a first sync of thousands
+  /// of rows. A row qualifies when the batch inserts it once and touches it
+  /// no other way, nothing earlier in the batch names it, and each row of
+  /// the batch it names qualified before it. Its values are planned in the
+  /// batch's order, as the per-row path plans them. A chunk whose writes fail
+  /// (a row there already, perhaps another space's) is rolled back and left,
+  /// with every row after it, to the per-row path, which tells those apart.
+  Future<Set<MergeRowKey>> _applyNewRowInsertsInChunks(
+    List<CrdtMergeChange> operations,
+    Map<UuidValue, CrdtNode> remoteNodes,
+    MergeContext context,
+    _MergeWriteBatch writes,
+    CrdtForeignKeyPresenceCache? presence,
+    Transaction transaction,
+  ) async {
+    final touches = <MergeRowKey, int>{};
+    for (final operation in operations) {
+      if (!_context.isCrdtTrackedTableName(operation.tableName)) continue;
+      final rowKey = (operation.tableName, operation.uuidRowId);
+      touches[rowKey] = (touches[rowKey] ?? 0) + 1;
+    }
+
+    final written = <MergeRowKey>{};
+    // Rows an insert names, by foreign key or as its set-default fallback: one
+    // named before it is written would find the row missing.
+    final named = <MergeRowKey>{};
+    final mergingSpaceId = _context.hlcManagerFor(transaction).normalizedSpaceId;
+    var chunk = _NewRowChunk();
+    Future<bool> writeChunk() async {
+      if (!await _writeNewRowChunk(chunk, context, writes, transaction)) {
+        // Marked visible for their children, but not written after all.
+        if (presence != null) chunk.rowKeys.forEach(presence.forget);
+        return false;
+      }
+      written.addAll(chunk.rowKeys);
+      chunk = _NewRowChunk();
+      return true;
+    }
+
+    for (final operation in operations) {
+      if (operation is! CrdtMergeInsert ||
+          !_context.isCrdtTrackedTableName(operation.tableName)) {
+        continue;
+      }
+      final rowKey = (operation.tableName, operation.uuidRowId);
+      final data = _sanitizeMergeRowData(
+        operation.tableName,
+        operation.databaseColumns,
+      );
+      final parents = _rowsNamedByMergeInsert(operation.tableName, data);
+      // A parent named by another column could be any row: leave it all.
+      if (parents == null) break;
+      final qualifies =
+          operation.data is TableRow &&
+          touches[rowKey] == 1 &&
+          context.rows[rowKey] == null &&
+          !named.contains(rowKey) &&
+          parents.every(
+            (parent) =>
+                !touches.containsKey(parent) ||
+                written.contains(parent) ||
+                chunk.holds(parent),
+          );
+      named.addAll(parents);
+      for (final edge
+          in _foreignKeys.edgesByChildTable[operation.tableName] ??
+              const <ForeignKeyEdge>[]) {
+        if (edge.action != ForeignKeyAction.setDefault) continue;
+        if (tryUuidValue(edge.defaultValue) case final fallback?) {
+          named.add((edge.parentTableName, fallback));
+        }
+      }
+      if (!qualifies) continue;
+      // Parents go in before their children: a chunk writes each table after
+      // the tables its rows name there, so a row naming one of its own table,
+      // or one whose table has to follow its own, waits for the next chunk.
+      if (!chunk.canTake(operation.tableName, parents) && !await writeChunk()) {
+        return written;
+      }
+
+      final planned = await _planMergeInsertValues(
+        operation,
+        data,
+        null,
+        presence,
+        transaction,
+      );
+      final domain = _mergeInsertDomainRow(operation, planned.domain, mergingSpaceId);
+      chunk.add(
+        (
+          tableName: operation.tableName,
+          rowId: operation.uuidRowId,
+          node: _requireRemoteNode(remoteNodes, operation.uuidNodeId),
+          hlc: operation.hlc,
+          domainRow: domain.row,
+          explicitNulls: domain.explicitNulls,
+          attempts: planned.attempts,
+        ),
+        parents,
+      );
+      // Its children further on find it written, as one by one.
+      presence?.markVisible(rowKey);
+      if (chunk.length >= _NewRowChunk.maxRows && !await writeChunk()) {
+        return written;
+      }
+    }
+    await writeChunk();
+    return written;
+  }
+
+  /// Writes [chunk] in one savepoint, domain rows first and their trackers
+  /// after them, as the per-row path does, and records what it would.
+  /// Rolls the chunk back and returns false when a statement fails.
+  Future<bool> _writeNewRowChunk(
+    _NewRowChunk chunk,
+    MergeContext context,
+    _MergeWriteBatch writes,
+    Transaction transaction,
+  ) async {
+    if (chunk.isEmpty) return true;
+    final Map<MergeRowKey, CrdtDataRow> trackers;
+    try {
+      trackers = await DatabaseUtil.runInTransactionOrSavepoint(
+        _db,
+        transaction,
+        (savepoint) async {
+          for (final tableName in chunk.tablesInWriteOrder) {
+            final rows = chunk.rowsByTable[tableName]!;
+            await insertWithExplicitNulls<TableRow>(
+              _db,
+              [for (final row in rows) row.domainRow],
+              explicitNulls: {
+                for (final row in rows)
+                  if (row.explicitNulls.isNotEmpty) row.rowId: row.explicitNulls,
+              },
+              transaction: savepoint,
+              noReturn: true,
+            );
+          }
+          final inserted = <MergeRowKey, CrdtDataRow>{};
+          for (final MapEntry(key: tableName, value: rows)
+              in chunk.rowsByTable.entries) {
+            final trackers = await CrdtDataRow.db.insert(
+              _session,
+              [
+                for (final row in rows)
+                  _newMergeRow(tableName, row.rowId, row.node, row.hlc, savepoint),
+              ],
+              transaction: savepoint,
+            );
+            for (final tracker in trackers) {
+              inserted[(tableName, tracker.uuidRowId)] = tracker;
+            }
+          }
+          return inserted;
+        },
+      );
+    } on DatabaseQueryException {
+      return false;
+    }
+
+    final mergingSpaceId = _context.hlcManagerFor(transaction).normalizedSpaceId;
+    for (final row in chunk.rows) {
+      final rowKey = (row.tableName, row.rowId);
+      context.rows[rowKey] = trackers[rowKey]!.copyWith(node: row.node);
+      context.domainOwners[rowKey] = (exists: true, spaceId: mergingSpaceId);
+      writes.recordInsertAttempts(
+        tableName: row.tableName,
+        rowId: row.rowId,
+        node: row.node,
+        attempts: row.attempts,
+      );
+    }
+    return true;
+  }
+
+  /// The rows [data], an insert into [tableName], names by foreign key, or
+  /// null when one names its row by another column or not by a UUID.
+  Set<MergeRowKey>? _rowsNamedByMergeInsert(
+    String tableName,
+    Map<String, Object?> data,
+  ) {
+    final named = <MergeRowKey>{};
+    for (final edge
+        in _foreignKeys.edgesByChildTable[tableName] ?? const <ForeignKeyEdge>[]) {
+      final value = data[edge.childColumn];
+      if (value == null) continue;
+      if (edge.parentColumn != 'id') return null;
+      final id = tryUuidValue(value);
+      if (id == null) return null;
+      named.add((edge.parentTableName, id));
+    }
+    return named;
+  }
+
+  /// The domain row a merged insert writes with its planned values [data],
+  /// owned by [spaceId], and the columns whose null it writes though they
+  /// have a default.
+  ({TableRow row, Set<String> explicitNulls}) _mergeInsertDomainRow(
+    CrdtMergeInsert insert,
+    Map<String, Object?> data,
+    int spaceId,
+  ) {
+    final tableRow = _requireMergeInsertTableRow(insert);
+    final row = _withPlannedDomainValues(tableRow, data).copyWithSpaceId(spaceId);
+    return (
+      row: row,
+      explicitNulls: {
+        for (final column in row.table.crdtSyncableColumns)
+          if (column.hasDefault && data[column.columnName] == null) column.columnName,
+      },
+    );
+  }
+
+  /// The tracker of a row a merge inserts, not written yet.
+  CrdtDataRow _newMergeRow(
+    String tableName,
+    UuidValue rowId,
+    CrdtNode remoteNode,
+    Hlc incomingHlc,
+    Transaction transaction,
+  ) {
+    final (tableId, _) = _context.schema[tableName]!;
+    return CrdtDataRow(
+      spaceId: _context.hlcManagerFor(transaction).normalizedSpaceId,
+      tblId: tableId,
+      uuidRowId: rowId,
+      nodeId: remoteNode.id!,
+      hlcDatetime: incomingHlc.datetime,
+      hlcCounter: incomingHlc.counter,
+    );
+  }
+
   Future<void> _applyMergeInsert(
     CrdtMergeInsert insert,
     Map<UuidValue, CrdtNode> remoteNodes,
@@ -656,17 +977,9 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
     Transaction transaction,
   ) async {
     if (currentRow == null) {
-      final (tableId, _) = _context.schema[tableName]!;
       final insertedRow = await CrdtDataRow.db.insertRow(
         _session,
-        CrdtDataRow(
-          spaceId: _context.hlcManagerFor(transaction).normalizedSpaceId,
-          tblId: tableId,
-          uuidRowId: rowId,
-          nodeId: remoteNode.id!,
-          hlcDatetime: incomingHlc.datetime,
-          hlcCounter: incomingHlc.counter,
-        ),
+        _newMergeRow(tableName, rowId, remoteNode, incomingHlc, transaction),
         transaction: transaction,
       );
 
@@ -810,10 +1123,12 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
     Map<String, Object?> data,
     Transaction transaction,
   ) async {
-    final tableRow = _requireMergeInsertTableRow(insert);
-    final patchedRow = _withPlannedDomainValues(tableRow, data);
     final mergingSpaceId = _context.hlcManagerFor(transaction).normalizedSpaceId;
-    final spaceScopedRow = patchedRow.copyWithSpaceId(mergingSpaceId);
+    final (row: spaceScopedRow, :explicitNulls) = _mergeInsertDomainRow(
+      insert,
+      data,
+      mergingSpaceId,
+    );
     try {
       // Domain insert first. A CRDT tracker row for this id, written before
       // the domain row, made SQLite reject a later null unique-FK insert.
@@ -821,11 +1136,7 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
         _db,
         transaction,
         (savepoint) => _db.insertRow(
-          withExplicitInsertNulls(spaceScopedRow, {
-            for (final column in spaceScopedRow.table.crdtSyncableColumns)
-              if (column.hasDefault && data[column.columnName] == null)
-                column.columnName,
-          }),
+          withExplicitInsertNulls(spaceScopedRow, explicitNulls),
           transaction: savepoint,
         ),
       );

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

@@ -33,6 +33,15 @@ String domainColumnPredicate(
   String alias = 'd',
 }) => '$alias."${columnName.escapeIdentifier()}" = ${value.sqlLiteral()}';
 
+/// SQL predicate matching rows where [columnName] holds one of [values],
+/// which must not be empty.
+@internal
+String domainColumnInPredicate(
+  String columnName,
+  Iterable<Object?> values, {
+  String alias = 'd',
+}) => '$alias."${columnName.escapeIdentifier()}" IN (${values.sqlLiteralList()})';
+
 /// SQL predicate matching rows where [columnName] differs from [value].
 @internal
 String domainColumnNotPredicate(

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

@@ -159,6 +159,24 @@ class CrdtForeignKeyProjector {
         _foreignKeys.referencingKeysByParentTable[parentTableName] ??
         const <ReferencingForeignKey>[];
     for (final reference in foreignKeys) {
+      // A cascade by id needs no per-parent answer: ask for every visible
+      // child of the whole batch at once instead of one query per parent.
+      if (reference.action == ForeignKeyAction.cascade &&
+          reference.parentColumn == 'id') {
+        final childIds = await _context.findVisibleDomainRowIdsWhere(
+          tableName: reference.childTableName,
+          predicates: [
+            domainColumnInPredicate(reference.childColumn, parentIds),
+          ],
+          transaction: transaction,
+        );
+        if (childIds.isNotEmpty) {
+          cascadeDeletes
+              .putIfAbsent(reference.childTableName, () => {})
+              .addAll(childIds);
+        }
+        continue;
+      }
       final parentValuesById = reference.parentColumn == 'id'
           ? null
           : await _context.readDomainColumnValues(
@@ -380,7 +398,11 @@ class CrdtForeignKeyProjector {
     };
     if (attemptedColumns.isEmpty) return;
 
-    final crdtRows = await _context.findCrdtRows(tableName, rowIds, transaction);
+    // A merge holds the trackers it has just written: reading thousands of
+    // them back took seconds.
+    final crdtRows = mergeCache == null
+        ? await _context.findCrdtRows(tableName, rowIds, transaction)
+        : [for (final rowId in rowIds) ?mergeCache.rows[(tableName, rowId)]];
     if (crdtRows.isEmpty) return;
 
     final columnIds = _context.schemaColumnIds(tableName, attemptedColumns);
@@ -738,6 +760,204 @@ class CrdtForeignKeyProjector {
     );
   }
 
+  /// Proves that merged inserts of new rows, [rowsByTable] (each row's
+  /// authored values by its id, by its table; [positions] says where each
+  /// comes in the batch), leave every projection decision as it is, as
+  /// [canLeaveLocalProjectionUnchanged] proves it for local inserts: each row
+  /// references its parents by id, each parent visible or inserted earlier in
+  /// the same merge, of its own table too (and proven alike; one inserted
+  /// after its child, or the row itself, would find the child's reference
+  /// held back, which only the pass restores), nothing refers to the row yet
+  /// (no row, no attempted value), and no unique claim is contested. The
+  /// merge then writes them as authored, without the pass, which would load
+  /// every row connected to them: all of a library whose tags join its
+  /// parts.
+  Future<bool> canLeaveProjectionUnchangedByMergedInserts(
+    Map<String, Map<UuidValue, Map<String, Object?>>> rowsByTable,
+    Map<MergeRowKey, int> positions,
+    Transaction transaction,
+  ) async {
+    for (final MapEntry(key: tableName, value: rows) in rowsByTable.entries) {
+      final ids = rows.keys.toSet();
+      final childEdges =
+          _foreignKeys.edgesByChildTable[tableName] ?? const <ForeignKeyEdge>[];
+      final parentEdges =
+          _foreignKeys.edgesByParentTable[tableName] ?? const <ForeignKeyEdge>[];
+      if (childEdges.any((edge) => edge.parentColumn != 'id') ||
+          parentEdges.any((edge) => edge.parentColumn != 'id')) {
+        return false;
+      }
+      final columns = _context.columnsByTableAndName[tableName];
+      Object? valueOf(Map<String, Object?> row, String column) =>
+          canonicalDomainValue(row[column], columns?[column]);
+
+      final parentsByTable = <String, Set<UuidValue>>{};
+      for (final edge in childEdges) {
+        for (final MapEntry(key: rowId, value: row) in rows.entries) {
+          final value = valueOf(row, edge.childColumn);
+          if (value == null) {
+            if (!edge.childNullable || edge.defaultValue != null) return false;
+            continue;
+          }
+          final id = tryUuidValue(value);
+          if (id == null) return false;
+          if (positions[(edge.parentTableName, id)] case final parentAt?) {
+            if (parentAt >= positions[(tableName, rowId)]!) return false;
+            continue;
+          }
+          (parentsByTable[edge.parentTableName] ??= {}).add(id);
+        }
+      }
+      final referencingColumnsByTable = <String, Set<String>>{};
+      for (final edge in parentEdges) {
+        if (edge.action == ForeignKeyAction.setDefault &&
+            ids.contains(tryUuidValue(edge.defaultValue))) {
+          return false;
+        }
+        (referencingColumnsByTable[edge.childTableName] ??= {}).add(edge.childColumn);
+      }
+      final uniqueClaims = <Map<String, Set<Object?>>>[];
+      for (final index in _uniqueResolver.uniqueIndexesFor(tableName)) {
+        final tuples = <String>{};
+        final claims = <String, Set<Object?>>{};
+        for (final row in rows.values) {
+          final tuple = [
+            for (final column in index.indexedColumns) valueOf(row, column),
+          ];
+          if (tuple.any((value) => value == null)) continue;
+          if (!tuples.add(tuple.map(canonicalProjectionValue).join('\x1f'))) {
+            return false;
+          }
+          for (final (i, column) in index.indexedColumns.indexed) {
+            (claims[column] ??= {}).add(tuple[i]);
+          }
+        }
+        if (claims.isNotEmpty) uniqueClaims.add(claims);
+      }
+      if (!await _context.localProjectionDependenciesAreStable(
+        tableName: tableName,
+        rowIds: ids,
+        persisted: false,
+        parentsByTable: parentsByTable,
+        referencingColumnsByTable: referencingColumnsByTable,
+        uniqueClaims: uniqueClaims,
+        transaction: transaction,
+      )) {
+        return false;
+      }
+    }
+    return true;
+  }
+
+  /// Proves that a local delete of [tableName] rows leaves nothing for the
+  /// projection pass to decide: the delete's own cascade
+  /// (`applyDeleteActions`, recursively) already hid everything it hides.
+  ///
+  /// Holds when, in every table the delete can cascade into, every incoming
+  /// edge cascades by id (no SET NULL / SET DEFAULT repairs), no outgoing edge
+  /// restricts (the deleted row could be the blocker of a withheld parent
+  /// delete), no table has a unique index (a released claim could restore a
+  /// contestant) or a SET DEFAULT dependency, and no attempted value is
+  /// pending in the tables the pass would load. Otherwise the full pass runs.
+  Future<bool> canSkipProjectionAfterLocalDelete(
+    String tableName,
+    Transaction transaction,
+  ) async =>
+      _deletesOnlyCascade(tableName) &&
+      !await _hasAttemptedClaimsConnectedTo({tableName}, transaction);
+
+  /// Proves that merged deletes of [deletedByTable] (row ids by table, each
+  /// one deleting, none restoring) leave nothing for the projection pass to
+  /// decide. A device deleting a row locally tombstones everything visible
+  /// below it too ([canSkipProjectionAfterLocalDelete]), so its batch carries
+  /// those deletes along; what the pass could still decide is a row below
+  /// that the deleting device never saw, inserted here meanwhile, which it
+  /// hides by cascade. So, beyond the tables' own rules, every visible row
+  /// below a deleted one must be deleted by the same batch, found by
+  /// walking down the cascades alone, where the pass walks both ways through
+  /// everything connected.
+  Future<bool> canLeaveProjectionUnchangedByMergedDeletes(
+    Map<String, Set<UuidValue>> deletedByTable,
+    Transaction transaction,
+  ) async {
+    if (!deletedByTable.keys.every(_deletesOnlyCascade)) return false;
+    if (await _hasAttemptedClaimsConnectedTo(
+      deletedByTable.keys.toSet(),
+      transaction,
+    )) {
+      return false;
+    }
+    final deleted = {
+      for (final MapEntry(key: table, value: ids) in deletedByTable.entries)
+        for (final id in ids) (table, id),
+    };
+    var frontier = deletedByTable;
+    while (frontier.isNotEmpty) {
+      final next = <String, Set<UuidValue>>{};
+      for (final MapEntry(key: table, value: ids) in frontier.entries) {
+        if (ids.isEmpty) continue;
+        for (final edge
+            in _foreignKeys.edgesByParentTable[table] ?? const <ForeignKeyEdge>[]) {
+          for (final child in await _context.findVisibleDomainRowIdsWhere(
+            tableName: edge.childTableName,
+            predicates: [domainColumnInPredicate(edge.childColumn, ids)],
+            transaction: transaction,
+          )) {
+            if (!deleted.contains((edge.childTableName, child))) return false;
+            (next[edge.childTableName] ??= {}).add(child);
+          }
+        }
+      }
+      frontier = next;
+    }
+    return true;
+  }
+
+  /// Whether a delete of [tableName] rows can only cascade: every table the
+  /// delete can reach is reached by cascades on id alone (no SET NULL or SET
+  /// DEFAULT repairs), none restricts its parent (a deleted row could be the
+  /// blocker of a withheld parent delete), and none has a unique index (a
+  /// released claim could restore a contestant) or a SET DEFAULT dependency.
+  bool _deletesOnlyCascade(String tableName) {
+    final cascadeTables = <String>{};
+    final queue = [tableName];
+    while (queue.isNotEmpty) {
+      final table = queue.removeLast();
+      if (!cascadeTables.add(table)) continue;
+      for (final edge
+          in _foreignKeys.edgesByParentTable[table] ?? const <ForeignKeyEdge>[]) {
+        if (edge.action != ForeignKeyAction.cascade || edge.parentColumn != 'id') {
+          return false;
+        }
+        queue.add(edge.childTableName);
+      }
+    }
+    for (final table in cascadeTables) {
+      if (_uniqueResolver.tableHasUniqueIndexes(table)) return false;
+      if (_foreignKeys.tablesWithDefaultDependencies.contains(table)) return false;
+      for (final edge
+          in _foreignKeys.edgesByChildTable[table] ?? const <ForeignKeyEdge>[]) {
+        if (edge.action == ForeignKeyAction.restrict ||
+            edge.action == ForeignKeyAction.noAction) {
+          return false;
+        }
+      }
+    }
+    return true;
+  }
+
+  /// Whether an attempted value is pending in a table connected to [tables],
+  /// which the pass would load and could resolve.
+  Future<bool> _hasAttemptedClaimsConnectedTo(
+    Set<String> tables,
+    Transaction transaction,
+  ) => _ProjectionReader(_context, transaction).hasAttemptedClaims(
+    _foreignKeys
+        .connectedTables(tables)
+        .where(_context.syncTableByName.containsKey)
+        .toSet(),
+  );
+
   /// Whether a mutation of [tableName] may require FK or unique projection.
   bool needsProjection(String tableName, Set<String>? columnNames) {
     if (_foreignKeys.columnsMayAffectForeignKeys(tableName, columnNames)) {

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

@@ -23,7 +23,8 @@ typedef MergeContext = ({
   Map<MergeRowKey, DomainRowOwner> domainOwners,
 });
 
-/// The merge context's field cache, with the node authoring the batch.
+/// The merge context's field cache, with the node authoring the batch, and
+/// its trackers, which hold every row the merge has written.
 ///
 /// The two are one parameter because they are only meaningful together: a
 /// cached field is read back through `CrdtDataField.hlc`, which resolves the
@@ -33,6 +34,7 @@ typedef MergeContext = ({
 typedef MergeFieldCache = ({
   Map<MergeFieldKey, CrdtDataField> fields,
   CrdtNode node,
+  Map<MergeRowKey, CrdtDataRow> rows,
 });
 
 /// Remote nodes and space-node rows loaded for a merge.

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

@@ -9,12 +9,14 @@ class _MergeWriteBatch {
     required this._session,
     required this._projector,
     required this._fields,
+    required this._rows,
     required this._transaction,
   });
 
   final DatabaseSession _session;
   final CrdtForeignKeyProjector _projector;
   final Map<MergeFieldKey, CrdtDataField> _fields;
+  final Map<MergeRowKey, CrdtDataRow> _rows;
   final Transaction _transaction;
   final _insertAttempts = _PendingInsertAttempts();
   final _fieldUpdates = <int, CrdtDataField>{};
@@ -88,7 +90,7 @@ class _MergeWriteBatch {
         group.rowIds,
         _transaction,
         group.attempts,
-        mergeCache: (fields: _fields, node: group.node),
+        mergeCache: (fields: _fields, node: group.node, rows: _rows),
       );
     }
   }
@@ -152,3 +154,78 @@ class _PendingInsertAttempts {
     return groups;
   }
 }
+
+/// A new row a merge writes in a chunk, with its planned values.
+typedef _NewRow = ({
+  String tableName,
+  UuidValue rowId,
+  CrdtNode node,
+  Hlc hlc,
+  TableRow domainRow,
+  Set<String> explicitNulls,
+  ProjectionAttemptsByField attempts,
+});
+
+/// New rows of a merge waiting to be written together, a statement per table,
+/// each table's rows in the order they came.
+class _NewRowChunk {
+  /// Rows a chunk holds at most: bounds the size of each statement.
+  static const maxRows = 1000;
+
+  final Map<String, List<_NewRow>> rowsByTable = {};
+  final List<_NewRow> rows = [];
+  final Set<MergeRowKey> rowKeys = {};
+
+  /// The tables of rows held here that rows of the key's table name.
+  final Map<String, Set<String>> _parentTables = {};
+
+  bool get isEmpty => rows.isEmpty;
+
+  int get length => rows.length;
+
+  bool holds(MergeRowKey rowKey) => rowKeys.contains(rowKey);
+
+  /// Whether a row of [tableName] naming [parents] can join: its table can
+  /// still be written after the tables of the parents held here, so neither
+  /// one of its own table nor one whose table must follow its own.
+  bool canTake(String tableName, Iterable<MergeRowKey> parents) => parents.every(
+    (parent) =>
+        !holds(parent) || (parent.$1 != tableName && !_follows(parent.$1, tableName)),
+  );
+
+  /// Adds [row], naming [parents].
+  void add(_NewRow row, Iterable<MergeRowKey> parents) {
+    for (final parent in parents) {
+      if (holds(parent)) (_parentTables[row.tableName] ??= {}).add(parent.$1);
+    }
+    (rowsByTable[row.tableName] ??= []).add(row);
+    rows.add(row);
+    rowKeys.add((row.tableName, row.rowId));
+  }
+
+  /// The tables held here, each after the tables its rows name here, and
+  /// otherwise in the order they came.
+  List<String> get tablesInWriteOrder {
+    final ordered = <String>[];
+    void place(String tableName) {
+      if (ordered.contains(tableName)) return;
+      _parentTables[tableName]?.forEach(place);
+      ordered.add(tableName);
+    }
+
+    rowsByTable.keys.forEach(place);
+    return ordered;
+  }
+
+  /// Whether [tableName] has to be written after [other].
+  bool _follows(String tableName, String other) {
+    final seen = <String>{};
+    final queue = [tableName];
+    while (queue.isNotEmpty) {
+      final parents = _parentTables[queue.removeLast()] ?? const <String>{};
+      if (parents.contains(other)) return true;
+      queue.addAll(parents.where(seen.add));
+    }
+    return false;
+  }
+}

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

@@ -10,6 +10,7 @@ import '../hlc/hlc.dart';
 import '../managers/hlc.dart';
 import '../sync/engine.dart';
 import '../sync/exceptions.dart';
+import '../utils/once.dart';
 import 'database.dart';
 import 'merge_utils/database_helpers.dart';
 import 'merge_utils/foreign_key_graph.dart';
@@ -58,6 +59,17 @@ class OfflineSyncDatabaseContext {
   @internal
   bool get registryChanged => _registryChanged;
 
+  final _projectionRebuild = Once();
+
+  /// Runs [rebuild], the projections of every space built anew, once for
+  /// this context when [registryChanged]. Every ephemeral database shares the
+  /// context and initializes a recorder of its own: each rebuilt every space
+  /// again, for every session and every merge until the process stopped
+  /// (upstream #150), which cost a device of a large library seconds a sync.
+  @internal
+  Future<void> rebuildProjectionsOnce(Future<void> Function() rebuild) =>
+      _registryChanged ? _projectionRebuild.run(rebuild) : Future<void>.value();
+
   /// Initializes shared schema rows and caches their generated identifiers.
   Future<void> initialize(DatabaseSession session) async {
     if (_schema != null) return;
@@ -264,9 +276,9 @@ class CrdtMutationRecorder {
 
   Future<void> _initializeOnce() async {
     await _databaseContext.initialize(_session);
-    if (_databaseContext.registryChanged) {
-      await _rebuildProjectionsForAllSpaces();
-    }
+    await _databaseContext.rebuildProjectionsOnce(
+      _rebuildProjectionsForAllSpaces,
+    );
     if (persistentUserId != null) {
       await _context.spaceManager.getOrCreate(persistentUserId!);
     }
@@ -1232,12 +1244,19 @@ class CrdtMutationRecorder {
       null,
       CrdtDataDeletedReason.userDelete,
     );
-    await _maybeProject(
+    // The delete's own cascade (applyDeleteActions above) usually is the
+    // whole projection; only run the full pass when it can decide more.
+    if (!await _foreignKeyProjector.canSkipProjectionAfterLocalDelete(
       tableName,
-      deletedRows.uuidRowIds,
-      null,
       transaction,
-    );
+    )) {
+      await _maybeProject(
+        tableName,
+        deletedRows.uuidRowIds,
+        null,
+        transaction,
+      );
+    }
     await _context.persistCurrentNodeHlc(
       _context.hlcManagerFor(transaction),
       transaction,

packages/serverpod_offline_sync/lib/src/sync/client_sync.dart

@@ -43,8 +43,14 @@ class OfflineSyncClient {
   Future<void> syncOnce(
     DatabaseSession session, {
     OfflineSyncOnMergeSuccess? onMergeSuccess,
+    OfflineSyncOnInboundMerge? onInboundMerge,
   }) async {
-    await _startSyncSession(session, once: true, onMergeSuccess: onMergeSuccess).done;
+    await _startSyncSession(
+      session,
+      once: true,
+      onMergeSuccess: onMergeSuccess,
+      onInboundMerge: onInboundMerge,
+    ).done;
   }
 
   /// Keeps synchronizing [session] with the remote peer until cancelled through
@@ -54,12 +60,19 @@ class OfflineSyncClient {
   OfflineSyncSubscription syncContinuously(
     DatabaseSession session, {
     OfflineSyncOnMergeSuccess? onMergeSuccess,
-  }) => _startSyncSession(session, once: false, onMergeSuccess: onMergeSuccess);
+    OfflineSyncOnInboundMerge? onInboundMerge,
+  }) => _startSyncSession(
+    session,
+    once: false,
+    onMergeSuccess: onMergeSuccess,
+    onInboundMerge: onInboundMerge,
+  );
 
   OfflineSyncSubscription _startSyncSession(
     DatabaseSession session, {
     required bool once,
     OfflineSyncOnMergeSuccess? onMergeSuccess,
+    OfflineSyncOnInboundMerge? onInboundMerge,
   }) {
     // The stream will be closed using [outboundChanges.closeOrSkip].
     // ignore: close_sinks
@@ -78,6 +91,7 @@ class OfflineSyncClient {
             inbound: remoteStream,
             once: once,
             onMergeSuccess: onMergeSuccess,
+            onInboundMerge: onInboundMerge,
             mode: OfflineSyncPeerMode.follower,
           )
           .listen(

packages/serverpod_offline_sync/lib/src/sync/engine.dart

@@ -27,12 +27,23 @@ export 'space_state.dart' show OfflineSyncPeerMode;
 typedef OfflineSyncOnMergeSuccess =
     FutureOr<void> Function(UuidValue spaceUuid, Hlc syncedHlc);
 
+/// Callback function for when changes from the remote peer were merged into
+/// a space. Unlike [OfflineSyncOnMergeSuccess], it isn't called for a space
+/// this peer only sent changes of, so a listener re-reads its data only
+/// when there is something new to read.
+typedef OfflineSyncOnInboundMerge = FutureOr<void> Function(UuidValue spaceUuid);
+
 /// A tuple representing the ownership of a domain row.
 typedef DomainRowOwner = ({bool exists, int? spaceId});
 
 /// A cache of domain row owners by table name and row id.
 typedef DomainRowOwnerCache = Map<(String, UuidValue), DomainRowOwner>;
 
+/// Domain rows read ahead for one collection, by table name and row id: each
+/// row's raw outbound columns and the space it was read for.
+typedef _DomainRowReads =
+    Map<(String, UuidValue), ({int spaceId, Map<String, dynamic> columns})>;
+
 /// The shared CRDT synchronization logic used by both client and server nodes.
 class OfflineSyncEngine {
   /// Creates a new [OfflineSyncEngine] instance.
@@ -144,8 +155,9 @@ class OfflineSyncEngine {
   /// scoped by those internal ids. Node ids are stable per replica and may
   /// appear in multiple spaces, so checkpoint filtering must compare both
   /// `spaceId` and `uuidNodeId`. This still runs one query per change kind for
-  /// the whole pass. Per-row ownership and integrity checks resolve against
-  /// each row's own space.
+  /// the whole pass, and reads the domain rows those changes name a batch of
+  /// rows to a statement. Per-row ownership and integrity checks resolve
+  /// against each row's own space.
   ///
   /// All changes for nodes that are not present in a space's checkpoint list
   /// are collected and emitted. Passing an empty list for a space will collect
@@ -275,6 +287,7 @@ class OfflineSyncEngine {
     required OfflineSyncPeerMode mode,
     bool once = false,
     OfflineSyncOnMergeSuccess? onMergeSuccess,
+    OfflineSyncOnInboundMerge? onInboundMerge,
   }) async* {
     // Idle timeouts are a continuous-only affordance: they let an idle cycle
     // settle into an empty batch without closing the stream. A `once` session
@@ -369,6 +382,7 @@ class OfflineSyncEngine {
           batch,
           outboundSpaces,
           onMergeSuccess,
+          onInboundMerge,
         );
 
         if (once) {
@@ -429,6 +443,7 @@ class OfflineSyncEngine {
     OfflineSyncCycleBatch batch,
     Set<UuidValue> outboundSpaces,
     OfflineSyncOnMergeSuccess? onMergeSuccess,
+    OfflineSyncOnInboundMerge? onInboundMerge,
   ) async {
     if (batch.spaceSet != null) {
       await spaces.adoptPeerGrants(batch.spaceSet!.spaces);
@@ -462,6 +477,9 @@ class OfflineSyncEngine {
         mergeSet: entry.value,
       );
       await _reportMerge(onMergeSuccess, spaces, spaceId, receivedHlc);
+      if (entry.value.isNotEmpty) {
+        await onInboundMerge?.call(spaceId);
+      }
     }
     for (final spaceId in outboundSpaces.difference(mergedSpaces)) {
       await _reportMerge(onMergeSuccess, spaces, spaceId, null);
@@ -569,13 +587,16 @@ class OfflineSyncEngine {
     Map<int, List<Hlc>> checkpointsBySpaceId,
     Transaction transaction,
   ) async* {
-    // Cache ownership within this collection's consistent snapshot.
+    // Cache ownership and domain rows within this collection's consistent
+    // snapshot.
     final ownerCache = DomainRowOwnerCache();
+    final domainRows = _DomainRowReads();
     yield* _streamInserts(
       session,
       spaceUuidById,
       checkpointsBySpaceId,
       ownerCache,
+      domainRows,
       transaction,
     );
     yield* _streamUpdates(
@@ -583,6 +604,7 @@ class OfflineSyncEngine {
       spaceUuidById,
       checkpointsBySpaceId,
       ownerCache,
+      domainRows,
       transaction,
     );
     yield* _streamDeletes(
@@ -599,6 +621,7 @@ class OfflineSyncEngine {
     Map<int, UuidValue> spaceUuidById,
     Map<int, List<Hlc>> checkpointsBySpaceId,
     DomainRowOwnerCache ownerCache,
+    _DomainRowReads domainRows,
     Transaction transaction,
   ) async* {
     final rows = await CrdtDataRow.db.find(
@@ -617,6 +640,19 @@ class OfflineSyncEngine {
       transaction,
     );
 
+    await _readDomainRowsAhead(
+      session,
+      [
+        for (final row in rows)
+          if (_syncTablesByName.containsKey(row.tbl!.name) &&
+              _classNamesByTableName.containsKey(row.tbl!.name))
+            (row.tbl!.name, row.spaceId, row.uuidRowId),
+      ],
+      ownerCache,
+      domainRows,
+      transaction,
+    );
+
     for (final row in rows) {
       final tableName = row.tbl!.name;
       if (!_syncTablesByName.containsKey(tableName)) continue;
@@ -637,6 +673,7 @@ class OfflineSyncEngine {
         attemptedValueFieldsByRowId[row.id!],
         spaceId,
         ownerCache,
+        domainRows,
         transaction,
       );
       if (!domainRow.exists) {
@@ -683,6 +720,7 @@ class OfflineSyncEngine {
     Map<int, UuidValue> spaceUuidById,
     Map<int, List<Hlc>> checkpointsBySpaceId,
     DomainRowOwnerCache ownerCache,
+    _DomainRowReads domainRows,
     Transaction transaction,
   ) async* {
     final fields = await CrdtDataField.db.find(
@@ -697,14 +735,27 @@ class OfflineSyncEngine {
       ),
     );
 
+    bool isUpdate(CrdtDataField field) =>
+        _syncTablesByName.containsKey(field.row!.tbl!.name) &&
+        !(field.hlcDatetime == field.row!.hlcDatetime &&
+            field.hlcCounter == field.row!.hlcCounter &&
+            field.nodeId == field.row!.nodeId);
+
+    await _readDomainRowsAhead(
+      session,
+      [
+        for (final field in fields)
+          if (isUpdate(field))
+            (field.row!.tbl!.name, field.row!.spaceId, field.row!.uuidRowId),
+      ],
+      ownerCache,
+      domainRows,
+      transaction,
+    );
+
     for (final field in fields) {
+      if (!isUpdate(field)) continue;
       final tableName = field.row!.tbl!.name;
-      if (!_syncTablesByName.containsKey(tableName)) continue;
-      if (field.hlcDatetime == field.row!.hlcDatetime &&
-          field.hlcCounter == field.row!.hlcCounter &&
-          field.nodeId == field.row!.nodeId) {
-        continue;
-      }
 
       final spaceId = field.row!.spaceId;
       final spaceUuid = spaceUuidById[spaceId]!;
@@ -717,6 +768,7 @@ class OfflineSyncEngine {
         field.attemptedValue,
         spaceId,
         ownerCache,
+        domainRows,
         transaction,
       );
       if (!columnValue.exists) {
@@ -866,6 +918,8 @@ class OfflineSyncEngine {
   /// targeted SQL projection also avoids fetching and serializing unrelated
   /// columns. This is the inverse of inbound FK materialization: the wire payload
   /// carries attempted facts, not locally projected visible values.
+  ///
+  /// A row in [domainRows] for [spaceId] is not read again.
   Future<({bool exists, int? ownerSpaceId, dynamic row})> _fetchDomainRow(
     DatabaseSession session,
     String tableName,
@@ -875,36 +929,37 @@ class OfflineSyncEngine {
     List<CrdtDataField>? attemptedValueFields,
     int spaceId,
     DomainRowOwnerCache ownerCache,
+    _DomainRowReads domainRows,
     Transaction transaction,
   ) async {
-    final cols = table.columns
-        .map(
-          (column) =>
-              '${_outboundColumnExpression(session, column)} AS "${column.columnName.escapeIdentifier()}"',
-        )
-        .join(', ');
-    final encodedRowId = rowId.sqlLiteral();
-    final encodedSpaceId = spaceId.sqlLiteral();
-    final escapedTableName = tableName.escapeIdentifier();
-    final result = await session.db.unsafeQuery(
-      'SELECT $cols FROM "$escapedTableName" '
-      'WHERE "id" = $encodedRowId AND "spaceId" = $encodedSpaceId '
-      'LIMIT 1',
-      transaction: transaction,
-    );
-    if (result.isEmpty) {
-      final owner = await _readDomainRowOwner(
-        session,
-        tableName,
-        rowId,
-        ownerCache,
-        transaction,
+    final Map<String, dynamic> rawColumns;
+    final readAhead = domainRows[(tableName, rowId)];
+    if (readAhead != null && readAhead.spaceId == spaceId) {
+      rawColumns = readAhead.columns;
+    } else {
+      final encodedRowId = rowId.sqlLiteral();
+      final encodedSpaceId = spaceId.sqlLiteral();
+      final escapedTableName = tableName.escapeIdentifier();
+      final result = await session.db.unsafeQuery(
+        'SELECT ${_outboundColumns(session, table)} FROM "$escapedTableName" '
+        'WHERE "id" = $encodedRowId AND "spaceId" = $encodedSpaceId '
+        'LIMIT 1',
+        transaction: transaction,
       );
-      return (exists: owner.exists, ownerSpaceId: owner.spaceId, row: null);
+      if (result.isEmpty) {
+        final owner = await _readDomainRowOwner(
+          session,
+          tableName,
+          rowId,
+          ownerCache,
+          transaction,
+        );
+        return (exists: owner.exists, ownerSpaceId: owner.spaceId, row: null);
+      }
+      rawColumns = result.first.toColumnMap();
     }
     ownerCache[(tableName, rowId)] = (exists: true, spaceId: spaceId);
 
-    final rawColumns = result.first.toColumnMap();
     final columnMap =
         <String, dynamic>{
             for (final column in table.columns)
@@ -948,6 +1003,7 @@ class OfflineSyncEngine {
     CrdtDataAttemptedValue? attempted,
     int spaceId,
     DomainRowOwnerCache ownerCache,
+    _DomainRowReads domainRows,
     Transaction transaction,
   ) async {
     if (attempted != null) {
@@ -968,12 +1024,25 @@ class OfflineSyncEngine {
       );
     }
 
-    final encodedRowId = rowId.sqlLiteral();
-    final encodedSpaceId = spaceId.sqlLiteral();
-    final escapedTableName = tableName.escapeIdentifier();
     final column = _syncTablesByName[tableName]!.columns.singleWhere(
       (c) => c.columnName == columnName,
     );
+    final readAhead = domainRows[(tableName, rowId)];
+    if (readAhead != null && readAhead.spaceId == spaceId) {
+      return (
+        exists: true,
+        ownerSpaceId: spaceId,
+        value: _decodeColumnValue(
+          tableName,
+          columnName,
+          _decodeStructuredValue(session, column, readAhead.columns[columnName]),
+        ),
+      );
+    }
+
+    final encodedRowId = rowId.sqlLiteral();
+    final encodedSpaceId = spaceId.sqlLiteral();
+    final escapedTableName = tableName.escapeIdentifier();
     final result = await session.db.unsafeQuery(
       'SELECT ${_outboundColumnExpression(session, column)} '
       'FROM "$escapedTableName" '
@@ -1004,6 +1073,58 @@ class OfflineSyncEngine {
     return (exists: owner.exists, ownerSpaceId: owner.spaceId, value: null);
   }
 
+  /// Reads the domain rows [rowKeys] name into [domainRows], those of one
+  /// table and space [_domainRowReadBatchSize] to a statement, and records
+  /// the owner of each row found. Rows already read are skipped; a row not
+  /// found is left to the per-row reads, which tell its cases apart.
+  Future<void> _readDomainRowsAhead(
+    DatabaseSession session,
+    Iterable<(String, int, UuidValue)> rowKeys,
+    DomainRowOwnerCache ownerCache,
+    _DomainRowReads domainRows,
+    Transaction transaction,
+  ) async {
+    final rowIdsByTableAndSpace = <(String, int), Set<UuidValue>>{};
+    for (final (tableName, spaceId, rowId) in rowKeys) {
+      if (domainRows.containsKey((tableName, rowId))) continue;
+      rowIdsByTableAndSpace.putIfAbsent((tableName, spaceId), () => {}).add(rowId);
+    }
+
+    for (final MapEntry(key: (tableName, spaceId), value: rowIds)
+        in rowIdsByTableAndSpace.entries) {
+      final columns = _outboundColumns(session, _syncTablesByName[tableName]!);
+      final escapedTableName = tableName.escapeIdentifier();
+      final encodedSpaceId = spaceId.sqlLiteral();
+      final ids = rowIds.toList();
+      for (var start = 0; start < ids.length; start += _domainRowReadBatchSize) {
+        final batch = ids.skip(start).take(_domainRowReadBatchSize);
+        final result = await session.db.unsafeQuery(
+          'SELECT $columns FROM "$escapedTableName" '
+          'WHERE "spaceId" = $encodedSpaceId AND "id" IN (${batch.sqlLiteralList()})',
+          transaction: transaction,
+        );
+        for (final row in result) {
+          final rawColumns = row.toColumnMap();
+          final rowId = (rawColumns['id'] as Object?).toUuidValue()!;
+          domainRows[(tableName, rowId)] = (spaceId: spaceId, columns: rawColumns);
+          ownerCache[(tableName, rowId)] = (exists: true, spaceId: spaceId);
+        }
+      }
+    }
+  }
+
+  /// Rows of one table and space read ahead in one statement at most: bounds
+  /// the size of each statement.
+  static const _domainRowReadBatchSize = 500;
+
+  /// The outbound column expressions of [table], each named for its column.
+  String _outboundColumns(DatabaseSession session, Table table) => table.columns
+      .map(
+        (column) =>
+            '${_outboundColumnExpression(session, column)} AS "${column.columnName.escapeIdentifier()}"',
+      )
+      .join(', ');
+
   Future<DomainRowOwner> _readDomainRowOwner(
     DatabaseSession session,
     String tableName,

packages/serverpod_offline_sync/lib/src/utils/once.dart

@@ -0,0 +1,21 @@
+import 'package:meta/meta.dart';
+
+/// Runs an action once for everyone who asks: a caller while it runs awaits
+/// the same run, and one after it is done gets it done; only a run that
+/// failed is run again by the next caller.
+@internal
+class Once {
+  Future<void>? _run;
+
+  /// Runs [action] unless it ran or runs already, and completes with that
+  /// run.
+  Future<void> run(Future<void> Function() action) async {
+    final run = _run ??= action();
+    try {
+      await run;
+    } catch (_) {
+      if (identical(_run, run)) _run = null;
+      rethrow;
+    }
+  }
+}

packages/serverpod_offline_sync/test/utils/once_test.dart

@@ -0,0 +1,47 @@
+import 'dart:async';
+
+import 'package:serverpod_offline_sync/src/utils/once.dart';
+import 'package:test/test.dart';
+
+void main() {
+  group('Given an action run once,', () {
+    late Once once;
+    late int runs;
+    late Completer<void> gate;
+
+    setUp(() {
+      once = Once();
+      runs = 0;
+      gate = Completer<void>();
+    });
+
+    Future<void> action() async {
+      runs++;
+      await gate.future;
+    }
+
+    test('when asked while it runs and after, then it runs once and every '
+        'caller awaits that run.', () async {
+      final first = once.run(action);
+      final meanwhile = once.run(action);
+      gate.complete();
+      await Future.wait([first, meanwhile]);
+      await once.run(action);
+
+      expect(runs, 1);
+    });
+
+    test('when its run fails, then the next caller runs it again.', () async {
+      final failing = once.run(() async {
+        runs++;
+        throw StateError('failed');
+      });
+      await expectLater(failing, throwsStateError);
+
+      gate.complete();
+      await once.run(action);
+
+      expect(runs, 2);
+    });
+  });
+}

test/serverpod_offline_sync_test_server/test/integration/merge/foreign_key_invariant_test.dart

@@ -157,6 +157,182 @@ void main() {
             expect(tombstone.hlc, remoteParentDelete.hlc);
           },
         );
+
+        group('and the restrict child is then deleted locally,', () {
+          setUp(() async {
+            await session.db.transactionForUser(testCrdtUserId, (tx) async {
+              await RestrictChild.db.deleteRow(session, child, transaction: tx);
+            });
+          });
+
+          test('then the merged parent delete takes effect.', () async {
+            expect(await Person.db.findById(session, parent.id!), isNull);
+            expect(await RestrictChild.db.findById(session, child.id!), isNull);
+          });
+        });
+      });
+    },
+  );
+
+  group(
+    'Given a set-null middle with two cascade children,',
+    () {
+      late FkChainSetNullMiddle middle;
+      late FkChainSetNullCascadeChild deleted;
+      late FkChainSetNullCascadeChild sibling;
+
+      setUp(() async {
+        await session.db.transactionForUser(testCrdtUserId, (tx) async {
+          middle = await FkChainSetNullMiddle.db.insertRow(
+            session,
+            FkChainSetNullMiddle(id: const Uuid().v7obj(), name: 'middle'),
+            transaction: tx,
+          );
+          deleted = await FkChainSetNullCascadeChild.db.insertRow(
+            session,
+            FkChainSetNullCascadeChild(
+              id: const Uuid().v7obj(),
+              name: 'deleted child',
+              setNullMiddleId: middle.id,
+            ),
+            transaction: tx,
+          );
+          sibling = await FkChainSetNullCascadeChild.db.insertRow(
+            session,
+            FkChainSetNullCascadeChild(
+              id: const Uuid().v7obj(),
+              name: 'sibling',
+              setNullMiddleId: middle.id,
+            ),
+            transaction: tx,
+          );
+        });
+      });
+
+      group('when one cascade child is deleted locally,', () {
+        setUp(() async {
+          await session.db.transactionForUser(testCrdtUserId, (tx) async {
+            await FkChainSetNullCascadeChild.db.deleteRow(
+              session,
+              deleted,
+              transaction: tx,
+            );
+          });
+        });
+
+        test('then only that child is hidden, by a user delete.', () async {
+          expect(
+            await FkChainSetNullCascadeChild.db.findById(session, deleted.id!),
+            isNull,
+          );
+          final visibleSibling = await FkChainSetNullCascadeChild.db.findById(
+            session,
+            sibling.id!,
+          );
+          expect(visibleSibling, isNotNull);
+          expect(visibleSibling!.setNullMiddleId, middle.id);
+          expect(
+            await FkChainSetNullMiddle.db.findById(session, middle.id!),
+            isNotNull,
+          );
+          final tombstone = await CrdtDataDeleted.db.findFirstRow(
+            session,
+            where: (t) => t.row.uuidRowId.equals(deleted.id),
+          );
+          expect(tombstone, isNotNull);
+          expect(tombstone!.isDeleted, isTrue);
+          expect(tombstone.reason, CrdtDataDeletedReason.userDelete);
+        });
+
+        group('and the middle is then deleted by a merged remote delete,', () {
+          setUp(() async {
+            await session.db.mergeChanges(
+              [
+                _deleteChange(
+                  tableName: FkChainSetNullMiddle.t.tableName,
+                  rowId: middle.id!,
+                  after: await rowHlc(middle.id!),
+                ),
+              ],
+              spaceId: testCrdtUserId,
+            );
+          });
+
+          test('then the remaining child is hidden with it.', () async {
+            expect(
+              await FkChainSetNullMiddle.db.findById(session, middle.id!),
+              isNull,
+            );
+            expect(
+              await FkChainSetNullCascadeChild.db.findById(session, sibling.id!),
+              isNull,
+            );
+          });
+        });
+      });
+    },
+  );
+
+  group(
+    'Given a set-null middle kept by its restrict child against a merged delete,',
+    () {
+      late FkChainSetNullMiddle middle;
+      late FkChainSetNullRestrictChild blocker;
+
+      setUp(() async {
+        await session.db.transactionForUser(testCrdtUserId, (tx) async {
+          middle = await FkChainSetNullMiddle.db.insertRow(
+            session,
+            FkChainSetNullMiddle(id: const Uuid().v7obj(), name: 'kept middle'),
+            transaction: tx,
+          );
+          blocker = await FkChainSetNullRestrictChild.db.insertRow(
+            session,
+            FkChainSetNullRestrictChild(
+              id: const Uuid().v7obj(),
+              name: 'blocker',
+              setNullMiddleId: middle.id,
+            ),
+            transaction: tx,
+          );
+        });
+        await session.db.mergeChanges(
+          [
+            _deleteChange(
+              tableName: FkChainSetNullMiddle.t.tableName,
+              rowId: middle.id!,
+              after: await rowHlc(middle.id!),
+            ),
+          ],
+          spaceId: testCrdtUserId,
+        );
+        expect(
+          await FkChainSetNullMiddle.db.findById(session, middle.id!),
+          isNotNull,
+        );
+      });
+
+      group('when the restrict child is deleted locally,', () {
+        setUp(() async {
+          await session.db.transactionForUser(testCrdtUserId, (tx) async {
+            await FkChainSetNullRestrictChild.db.deleteRow(
+              session,
+              blocker,
+              transaction: tx,
+            );
+          });
+        });
+
+        test('then the merged delete of the middle takes effect.', () async {
+          expect(
+            await FkChainSetNullMiddle.db.findById(session, middle.id!),
+            isNull,
+          );
+          expect(
+            await FkChainSetNullRestrictChild.db.findById(session, blocker.id!),
+            isNull,
+          );
+        });
       });
     },
   );

test/serverpod_offline_sync_test_server/test/integration/merge/merge_deletes_projection_test.dart

@@ -0,0 +1,117 @@
+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';
+
+// Merged deletes of rows below which nothing visible stays leave the
+// projection pass out; anything else takes it.
+void main() {
+  initTestClientSession();
+
+  Future<void> merge(List<CrdtMergeChange> changes) =>
+      session.db.mergeChanges(changes, spaceId: testCrdtUserId);
+
+  Future<CrdtMergeDelete> deleteOf(String tableName, UuidValue rowId) async =>
+      CrdtMergeDelete(
+        uuidSpaceId: testCrdtUserId,
+        tableName: tableName,
+        uuidRowId: rowId,
+        uuidNodeId: const Uuid().v7obj(),
+        hlcDatetime: (await rowHlc(
+          rowId,
+        )).datetime.add(const Duration(milliseconds: 1)),
+        hlcCounter: 0,
+        clFlag: 2,
+        reason: CrdtDataDeletedReason.userDelete,
+      );
+
+  group('Given a set-null middle with two cascade children stored here,', () {
+    late FkChainSetNullMiddle middle;
+    late FkChainSetNullCascadeChild deleted;
+    late FkChainSetNullCascadeChild sibling;
+
+    setUp(() async {
+      await session.db.transactionForUser(testCrdtUserId, (tx) async {
+        middle = await FkChainSetNullMiddle.db.insertRow(
+          session,
+          FkChainSetNullMiddle(id: const Uuid().v7obj(), name: 'middle'),
+          transaction: tx,
+        );
+        deleted = await FkChainSetNullCascadeChild.db.insertRow(
+          session,
+          FkChainSetNullCascadeChild(
+            id: const Uuid().v7obj(),
+            name: 'deleted',
+            setNullMiddleId: middle.id,
+          ),
+          transaction: tx,
+        );
+        sibling = await FkChainSetNullCascadeChild.db.insertRow(
+          session,
+          FkChainSetNullCascadeChild(
+            id: const Uuid().v7obj(),
+            name: 'sibling',
+            setNullMiddleId: middle.id,
+          ),
+          transaction: tx,
+        );
+      });
+    });
+
+    group('when a remote delete of one child is merged,', () {
+      setUp(() async {
+        await merge([
+          await deleteOf(FkChainSetNullCascadeChild.t.tableName, deleted.id!),
+        ]);
+      });
+
+      test('then only that child is hidden, by its delete.', () async {
+        expect(
+          await FkChainSetNullCascadeChild.db.findById(session, deleted.id!),
+          isNull,
+        );
+        expect(
+          await FkChainSetNullCascadeChild.db.findById(session, sibling.id!),
+          isNotNull,
+        );
+        expect(
+          await FkChainSetNullMiddle.db.findById(session, middle.id!),
+          isNotNull,
+        );
+        final row = await CrdtDataRow.db.findFirstRow(
+          session,
+          where: (t) => t.uuidRowId.equals(deleted.id),
+        );
+        expect(row!.visibility, CrdtDataRowVisibility.userDelete);
+      });
+    });
+
+    group('when a remote delete of the middle alone is merged,', () {
+      setUp(() async {
+        await merge([
+          await deleteOf(FkChainSetNullMiddle.t.tableName, middle.id!),
+        ]);
+      });
+
+      test('then its children are hidden with it, by the cascade.', () async {
+        expect(
+          await FkChainSetNullMiddle.db.findById(session, middle.id!),
+          isNull,
+        );
+        for (final child in [deleted, sibling]) {
+          expect(
+            await FkChainSetNullCascadeChild.db.findById(session, child.id!),
+            isNull,
+          );
+          final row = await CrdtDataRow.db.findFirstRow(
+            session,
+            where: (t) => t.uuidRowId.equals(child.id),
+          );
+          expect(row!.visibility, CrdtDataRowVisibility.foreignKeyCascade);
+        }
+      });
+    });
+  });
+}

test/serverpod_offline_sync_test_server/test/integration/merge/merge_inserts_projection_test.dart

@@ -0,0 +1,191 @@
+import 'package:serverpod/serverpod.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';
+
+// Merged inserts of new rows whose parents are visible, which nothing
+// refers to yet, change no projection decision, and the merge leaves the
+// projection pass out for them. Anything else still takes the pass: these
+// cover both sides.
+void main() {
+  initTestClientSession();
+
+  CrdtMergeInsert insertOf(TableRow row, String tableName) {
+    final node = const Uuid().v7obj();
+    return CrdtMergeInsert(
+      uuidSpaceId: testCrdtUserId,
+      tableName: tableName,
+      uuidRowId: row.id! as UuidValue,
+      uuidNodeId: node,
+      hlcDatetime: DateTime.now().toUtc(),
+      hlcCounter: 0,
+      data: row,
+    );
+  }
+
+  Future<void> merge(List<CrdtMergeChange> changes) =>
+      session.db.mergeChanges(changes, spaceId: testCrdtUserId);
+
+  group('Given a set-null middle stored here,', () {
+    late FkChainSetNullMiddle middle;
+
+    setUp(() async {
+      middle = await session.db.transactionForUser(
+        testCrdtUserId,
+        (tx) => FkChainSetNullMiddle.db.insertRow(
+          session,
+          FkChainSetNullMiddle(id: const Uuid().v7obj(), name: 'middle'),
+          transaction: tx,
+        ),
+      );
+    });
+
+    group('when a remote insert of its cascade child is merged,', () {
+      late FkChainSetNullCascadeChild child;
+
+      setUp(() async {
+        child = FkChainSetNullCascadeChild(
+          id: const Uuid().v7obj(),
+          name: 'child',
+          setNullMiddleId: middle.id,
+        );
+        await merge([insertOf(child, FkChainSetNullCascadeChild.t.tableName)]);
+      });
+
+      test('then the child is there as authored, holding nothing back.', () async {
+        final visible = await FkChainSetNullCascadeChild.db.findById(
+          session,
+          child.id!,
+        );
+        expect(visible, isNotNull);
+        expect(visible!.setNullMiddleId, middle.id);
+        expect(
+          await attemptedValue(
+            rowId: child.id!,
+            columnName: FkChainSetNullCascadeChild.t.setNullMiddleId.columnName,
+          ),
+          isNull,
+        );
+      });
+    });
+
+    group('when the middle is deleted by a merged remote delete and then a '
+        'remote insert of its cascade child is merged,', () {
+      late FkChainSetNullCascadeChild child;
+
+      setUp(() async {
+        await merge([
+          CrdtMergeDelete(
+            uuidSpaceId: testCrdtUserId,
+            tableName: FkChainSetNullMiddle.t.tableName,
+            uuidRowId: middle.id!,
+            uuidNodeId: const Uuid().v7obj(),
+            hlcDatetime: (await rowHlc(
+              middle.id!,
+            )).datetime.add(const Duration(milliseconds: 1)),
+            hlcCounter: 0,
+            clFlag: 2,
+            reason: CrdtDataDeletedReason.userDelete,
+          ),
+        ]);
+        child = FkChainSetNullCascadeChild(
+          id: const Uuid().v7obj(),
+          name: 'child of a deleted middle',
+          setNullMiddleId: middle.id,
+        );
+        await merge([insertOf(child, FkChainSetNullCascadeChild.t.tableName)]);
+      });
+
+      test('then the child is hidden with its middle.', () async {
+        expect(
+          await FkChainSetNullMiddle.db.findById(session, middle.id!),
+          isNull,
+        );
+        expect(
+          await FkChainSetNullCascadeChild.db.findById(session, child.id!),
+          isNull,
+        );
+      });
+    });
+  });
+
+  group('Given a set-null grandchild merged before its middle, its foreign '
+      'key held back,', () {
+    late FkChainSetNullMiddle middle;
+    late FkChainSetNullSetNullChild grandchild;
+
+    setUp(() async {
+      middle = FkChainSetNullMiddle(id: const Uuid().v7obj(), name: 'late');
+      grandchild = FkChainSetNullSetNullChild(
+        id: const Uuid().v7obj(),
+        name: 'early',
+        setNullMiddleId: middle.id,
+      );
+      await merge([
+        insertOf(grandchild, FkChainSetNullSetNullChild.t.tableName),
+      ]);
+      final early = await FkChainSetNullSetNullChild.db.findById(
+        session,
+        grandchild.id!,
+      );
+      expect(early!.setNullMiddleId, isNull);
+    });
+
+    group("when the middle's insert is merged,", () {
+      setUp(() async {
+        await merge([insertOf(middle, FkChainSetNullMiddle.t.tableName)]);
+      });
+
+      test('then the grandchild points at it.', () async {
+        final visible = await FkChainSetNullSetNullChild.db.findById(
+          session,
+          grandchild.id!,
+        );
+        expect(visible!.setNullMiddleId, middle.id);
+        expect(
+          await attemptedValue(
+            rowId: grandchild.id!,
+            columnName: FkChainSetNullSetNullChild.t.setNullMiddleId.columnName,
+          ),
+          isNull,
+        );
+      });
+    });
+  });
+
+  group('Given nothing stored,', () {
+    group('when a remote insert of a middle and its cascade child is merged '
+        'in one batch,', () {
+      late FkChainSetNullMiddle middle;
+      late FkChainSetNullCascadeChild child;
+
+      setUp(() async {
+        middle = FkChainSetNullMiddle(id: const Uuid().v7obj(), name: 'new');
+        child = FkChainSetNullCascadeChild(
+          id: const Uuid().v7obj(),
+          name: 'new child',
+          setNullMiddleId: middle.id,
+        );
+        await merge([
+          insertOf(middle, FkChainSetNullMiddle.t.tableName),
+          insertOf(child, FkChainSetNullCascadeChild.t.tableName),
+        ]);
+      });
+
+      test('then both are there, the child pointing at the middle.', () async {
+        expect(
+          await FkChainSetNullMiddle.db.findById(session, middle.id!),
+          isNotNull,
+        );
+        final visible = await FkChainSetNullCascadeChild.db.findById(
+          session,
+          child.id!,
+        );
+        expect(visible!.setNullMiddleId, middle.id);
+      });
+    });
+  });
+}

test/serverpod_offline_sync_test_server/test/integration/merge/merge_new_rows_in_chunks_test.dart

@@ -0,0 +1,225 @@
+import 'dart:typed_data';
+
+import 'package:serverpod/serverpod.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';
+
+// A batch the projection pass leaves out writes its new rows many to a
+// statement, up to a thousand at a time, rather than one by one. The rows
+// and their trackers have to come out as one by one, also where a chunk
+// fails and the rows from it on are written one by one after all.
+void main() {
+  initTestClientSession();
+
+  final node = const Uuid().v7obj();
+  final authoredAt = DateTime.now().toUtc();
+  var counter = 0;
+
+  // In the order a device authored them, as a push sends them.
+  CrdtMergeInsert insertOf(TableRow row, String tableName) => CrdtMergeInsert(
+    uuidSpaceId: testCrdtUserId,
+    tableName: tableName,
+    uuidRowId: row.id! as UuidValue,
+    uuidNodeId: node,
+    hlcDatetime: authoredAt,
+    hlcCounter: counter++,
+    data: row,
+  );
+
+  /// [count] middles, each followed by a cascade child naming it.
+  (List<FkChainSetNullMiddle>, List<FkChainSetNullCascadeChild>) family(
+    int count,
+  ) {
+    final middles = [
+      for (var i = 0; i < count; i++)
+        FkChainSetNullMiddle(id: const Uuid().v7obj(), name: 'middle $i'),
+    ];
+    final children = [
+      for (final (i, middle) in middles.indexed)
+        FkChainSetNullCascadeChild(
+          id: const Uuid().v7obj(),
+          name: 'child $i',
+          setNullMiddleId: middle.id,
+        ),
+    ];
+    return (middles, children);
+  }
+
+  List<CrdtMergeChange> inserts(
+    List<FkChainSetNullMiddle> middles,
+    List<FkChainSetNullCascadeChild> children,
+  ) => [
+    for (final (i, middle) in middles.indexed) ...[
+      insertOf(middle, FkChainSetNullMiddle.t.tableName),
+      insertOf(children[i], FkChainSetNullCascadeChild.t.tableName),
+    ],
+  ];
+
+  Future<void> expectWrittenAsAuthored(
+    List<FkChainSetNullMiddle> middles,
+    List<FkChainSetNullCascadeChild> children,
+  ) async {
+    final storedMiddles = {
+      for (final middle in await FkChainSetNullMiddle.db.find(
+        session,
+        where: (t) => t.id.inSet(<UuidValue>{for (final middle in middles) middle.id!}),
+      ))
+        middle.id: middle,
+    };
+    final storedChildren = {
+      for (final child in await FkChainSetNullCascadeChild.db.find(
+        session,
+        where: (t) => t.id.inSet(<UuidValue>{for (final child in children) child.id!}),
+      ))
+        child.id: child,
+    };
+    for (final middle in middles) {
+      expect(storedMiddles[middle.id]?.name, middle.name);
+    }
+    for (final child in children) {
+      expect(storedChildren[child.id]?.setNullMiddleId, child.setNullMiddleId);
+    }
+    final trackers = await CrdtDataRow.db.find(
+      testSession,
+      where: (t) => t.uuidRowId.inSet(<UuidValue>{
+        for (final middle in middles) middle.id!,
+        for (final child in children) child.id!,
+      }),
+    );
+    expect(trackers, hasLength(middles.length + children.length));
+    expect(
+      await attemptedValue(
+        rowId: children.last.id!,
+        columnName: FkChainSetNullCascadeChild.t.setNullMiddleId.columnName,
+      ),
+      isNull,
+    );
+  }
+
+  group('Given a remote batch of 1,200 new middles, each followed by a '
+      'cascade child naming it,', () {
+    late List<FkChainSetNullMiddle> middles;
+    late List<FkChainSetNullCascadeChild> children;
+
+    setUp(() async {
+      (middles, children) = family(1200);
+      await session.db.mergeChanges(
+        inserts(middles, children),
+        spaceId: testCrdtUserId,
+      );
+    });
+
+    test('then every row is there as authored, with one tracker each.', () async {
+      await expectWrittenAsAuthored(middles, children);
+    });
+  });
+
+  group('Given a middle here whose tracker was lost,', () {
+    late List<FkChainSetNullMiddle> middles;
+    late List<FkChainSetNullCascadeChild> children;
+
+    setUp(() async {
+      (middles, children) = family(800);
+      final lost = middles[600];
+      await session.db.transactionForUser(
+        testCrdtUserId,
+        (tx) => FkChainSetNullMiddle.db.insertRow(
+          session,
+          FkChainSetNullMiddle(id: lost.id, name: 'before recovery'),
+          transaction: tx,
+        ),
+      );
+      await CrdtDataRow.db.deleteWhere(
+        testSession,
+        where: (t) => t.uuidRowId.equals(lost.id),
+      );
+    });
+
+    group('when a remote batch of new rows holding it in its second '
+        'thousand is merged,', () {
+      setUp(() async {
+        await session.db.mergeChanges(
+          inserts(middles, children),
+          spaceId: testCrdtUserId,
+        );
+      });
+
+      test('then every row is there as authored, the lost one recovered, '
+          'with one tracker each.', () async {
+        await expectWrittenAsAuthored(middles, children);
+      });
+    });
+  });
+
+  group('Given rows of one table forming a tree,', () {
+    // Each row's parent is the row half as far into the list.
+    late List<Types> tree;
+
+    setUp(() {
+      tree = [];
+      for (var i = 0; i < 40; i++) {
+        tree.add(
+          Types(
+            id: const Uuid().v7obj(),
+            aBool: true,
+            aDateTime: DateTime.utc(2026),
+            aText: 'row $i',
+            anInt: i,
+            anInt64: BigInt.from(i),
+            aReal: 0.5,
+            aBlob: ByteData(1),
+            parentId: i == 0 ? null : tree[(i - 1) ~/ 2].id,
+          ),
+        );
+      }
+    });
+
+    Future<void> expectTreeAsAuthored() async {
+      final stored = {
+        for (final row in await Types.db.find(
+          session,
+          where: (t) => t.id.inSet(<UuidValue>{for (final row in tree) row.id!}),
+        ))
+          row.id: row,
+      };
+      for (final row in tree) {
+        expect(stored[row.id]?.parentId, row.parentId);
+        expect(
+          await attemptedValue(
+            rowId: row.id!,
+            columnName: Types.t.parentId.columnName,
+          ),
+          isNull,
+        );
+      }
+    }
+
+    group('when a remote batch inserting them, parents first, is merged,', () {
+      setUp(() async {
+        await session.db.mergeChanges([
+          for (final row in tree) insertOf(row, Types.t.tableName),
+        ], spaceId: testCrdtUserId);
+      });
+
+      test('then every row is there, holding its parent.', () async {
+        await expectTreeAsAuthored();
+      });
+    });
+
+    group('when a remote batch inserting them, children first, is merged,', () {
+      setUp(() async {
+        await session.db.mergeChanges([
+          for (final row in tree.reversed) insertOf(row, Types.t.tableName),
+        ], spaceId: testCrdtUserId);
+      });
+
+      test('then every row is there, holding its parent.', () async {
+        await expectTreeAsAuthored();
+      });
+    });
+  });
+}

test/serverpod_offline_sync_test_server/test/integration/sync/outbound_domain_row_reads_test.dart

@@ -0,0 +1,210 @@
+import 'dart:typed_data';
+
+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/query_log_database.dart';
+import '../test_tools/sync_topology.dart';
+
+const _rowCount = 1200;
+const _renamedCount = 600;
+
+void main() {
+  initTestClientSession(createSessionPerTest: false);
+
+  group(
+    'Given an author with $_rowCount pending rows, $_renamedCount of them renamed,',
+    () {
+      late SyncNode author;
+      late SyncNode peer;
+      late QueryLogDatabase queryLog;
+      late OfflineSyncDatabaseSession collector;
+
+      setUpAll(() async {
+        author = await syncNode(await createAdditionalTestSession(), [Types.t]);
+        peer = await syncNode(await createAdditionalTestSession(), [Types.t]);
+        final rows = await _insertTypes(author, _rowCount);
+        await _rename(author, rows.take(_renamedCount), 'renamed');
+        queryLog = QueryLogDatabase(author.raw.db, syncTables: [Types.t]);
+        collector = OfflineSyncDatabaseSession(queryLog, syncTables: [Types.t]);
+        await queryLog.initialize();
+      });
+
+      group('when a fresh peer collects everything and merges it,', () {
+        late List<CrdtMergeChange> changes;
+        late Map<UuidValue, Map<String, dynamic>> authored;
+        late Map<UuidValue, Map<String, dynamic>> received;
+
+        setUpAll(() async {
+          queryLog.rawQueries.clear();
+          changes = await author.sync
+              .collectPendingChanges(
+                collector,
+                checkpointsBySpaceUuid: {testCrdtUserId: const []},
+              )
+              .toList();
+          await peer.offlineSync.db.mergeChanges(
+            _overTheWire(author, changes),
+            spaceId: testCrdtUserId,
+          );
+          authored = await _rowsById(author);
+          received = await _rowsById(peer);
+        });
+
+        test('then domain rows are read in batches rather than one by one.', () {
+          expect(queryLog.readsOf('types').length, lessThanOrEqualTo(3));
+        });
+
+        test('then every insert carries the row as the author has it.', () {
+          expect(changes.inserts, hasLength(_rowCount));
+          for (final insert in changes.inserts) {
+            expect(_payload(insert.data as Types), authored[insert.uuidRowId]);
+          }
+        });
+
+        test('then every rename is collected with its current value.', () {
+          final renames = changes.updates.toList();
+          expect(renames, hasLength(_renamedCount));
+          for (final rename in renames) {
+            expect(rename.columnName, 'aText');
+            expect(rename.value, authored[rename.uuidRowId]!['aText']);
+          }
+        });
+
+        test("then the peer holds exactly the author's rows.", () {
+          expect(received, authored);
+        });
+      });
+    },
+  );
+
+  group('Given a peer holding $_rowCount rows from its author,', () {
+    late SyncNode author;
+    late SyncNode peer;
+    late List<Types> rows;
+    late List<Hlc> checkpoints;
+    late QueryLogDatabase queryLog;
+    late OfflineSyncDatabaseSession collector;
+
+    setUpAll(() async {
+      author = await syncNode(await createAdditionalTestSession(), [Types.t]);
+      peer = await syncNode(await createAdditionalTestSession(), [Types.t]);
+      rows = await _insertTypes(author, _rowCount);
+      await pushChanges(author, peer);
+      checkpoints = (await peer.sync.createSyncSinceHlc(
+        peer.raw,
+        spaceId: testCrdtUserId,
+      )).nodeCheckpoints;
+      queryLog = QueryLogDatabase(author.raw.db, syncTables: [Types.t]);
+      collector = OfflineSyncDatabaseSession(queryLog, syncTables: [Types.t]);
+      await queryLog.initialize();
+    });
+
+    group(
+      'when the author renames $_renamedCount rows and the peer collects since its checkpoint,',
+      () {
+        late List<CrdtMergeChange> changes;
+        late Map<UuidValue, Map<String, dynamic>> authored;
+        late Map<UuidValue, Map<String, dynamic>> received;
+
+        setUpAll(() async {
+          await _rename(author, rows.skip(_rowCount - _renamedCount), 'renamed');
+          queryLog.rawQueries.clear();
+          changes = await author.sync
+              .collectPendingChanges(
+                collector,
+                checkpointsBySpaceUuid: {testCrdtUserId: checkpoints},
+              )
+              .toList();
+          await peer.offlineSync.db.mergeChanges(
+            _overTheWire(author, changes),
+            spaceId: testCrdtUserId,
+          );
+          authored = await _rowsById(author);
+          received = await _rowsById(peer);
+        });
+
+        test('then the renamed rows are read in batches rather than one by one.', () {
+          expect(queryLog.readsOf('types').length, lessThanOrEqualTo(2));
+        });
+
+        test('then only the renames are collected, each with its new value.', () {
+          expect(changes.inserts, isEmpty);
+          expect(changes.updates, hasLength(_renamedCount));
+          for (final rename in changes.updates) {
+            expect(rename.value, startsWith('renamed '));
+            expect(rename.value, authored[rename.uuidRowId]!['aText']);
+          }
+        });
+
+        test("then the peer holds exactly the author's rows.", () {
+          expect(received, authored);
+        });
+      },
+    );
+  });
+}
+
+Future<List<Types>> _insertTypes(SyncNode node, int count) =>
+    node.offlineSync.db.transactionForUser(
+      testCrdtUserId,
+      (tx) => Types.db.insert(
+        node.offlineSync,
+        [
+          for (var i = 0; i < count; i++)
+            Types(
+              aBool: i.isEven,
+              aDateTime: DateTime.utc(2026, 11, 8).add(Duration(minutes: i)),
+              aText: 'row $i',
+              anInt: i,
+              anInt64: BigInt.from(i) << 40,
+              aReal: i / 8,
+              aBlob: Uint8List.fromList([i % 256, 0, 255]).buffer.asByteData(),
+              anEnum: i % 3 == 0 ? null : TypesEnum.values[i % TypesEnum.values.length],
+              optionalText: i.isOdd ? 'optional $i' : null,
+              optionalUuid: i % 5 == 0 ? const Uuid().v4obj() : null,
+              jsonDocument: i % 4 == 0
+                  ? null
+                  : SyncDocument(title: 'JSON $i', enabled: i.isOdd, numbers: [i, -i]),
+              jsonbDocument: SyncDocument(
+                title: 'JSONB ☕ $i',
+                enabled: i.isEven,
+                numbers: [i],
+              ),
+              jsonbNumbers: i % 7 == 0 ? null : [i, i + 1],
+            ),
+        ],
+        transaction: tx,
+      ),
+    );
+
+Future<void> _rename(SyncNode node, Iterable<Types> rows, String prefix) =>
+    node.offlineSync.db.transactionForUser(
+      testCrdtUserId,
+      (tx) => Types.db.update(
+        node.offlineSync,
+        [for (final row in rows) row.copyWith(aText: '$prefix ${row.anInt}')],
+        columns: (t) => [t.aText],
+        transaction: tx,
+      ),
+    );
+
+/// The changes as the peer decodes them from the sync stream.
+List<CrdtMergeChange> _overTheWire(SyncNode node, List<CrdtMergeChange> changes) {
+  final protocol = node.raw.db.serializationManager;
+  return [
+    for (final change in changes)
+      protocol.decodeWithType(protocol.encodeWithTypeForProtocol(change))!
+          as CrdtMergeChange,
+  ];
+}
+
+/// The model's JSON, without the space it is stored in locally.
+Map<String, dynamic> _payload(Types row) => row.toJson()..remove('spaceId');
+
+Future<Map<UuidValue, Map<String, dynamic>>> _rowsById(SyncNode node) async {
+  final rows = await Types.db.find(node.offlineSync);
+  return {for (final row in rows) row.id!: _payload(row)};
+}

test/serverpod_offline_sync_test_server/test/integration/test_tools/query_log_database.dart

@@ -0,0 +1,30 @@
+import 'package:serverpod_database/serverpod_database.dart' as database;
+import 'package:serverpod_offline_sync/serverpod_offline_sync.dart';
+
+/// Records the SQL of every raw query run through it, so a test can tell how
+/// often collection reads a domain table. Results are not replaced.
+class QueryLogDatabase extends OfflineSyncDatabase {
+  QueryLogDatabase(super.delegate, {required super.syncTables});
+
+  final List<String> rawQueries = [];
+
+  /// The raw queries reading rows of [tableName].
+  Iterable<String> readsOf(String tableName) =>
+      rawQueries.where((query) => query.contains('FROM "$tableName"'));
+
+  @override
+  Future<database.DatabaseResult> unsafeQuery(
+    String query, {
+    int? timeoutInSeconds,
+    database.Transaction? transaction,
+    database.QueryParameters? parameters,
+  }) {
+    rawQueries.add(query);
+    return super.unsafeQuery(
+      query,
+      timeoutInSeconds: timeoutInSeconds,
+      transaction: transaction,
+      parameters: parameters,
+    );
+  }
+}