perf: Write a merge's new rows in chunks when the pass is left out

A merge wrote each inserted row one by one: a savepoint for the domain
row, another for its tracker, two statements each, and then read every
tracker back to record attempted values. A device's first push of a
library of 8,000 cards (24,650 changes) took the server 34 to 37 s, nearly
all of it in those round trips.

A batch the projection pass leaves out (nothing in it touches a foreign
key or a unique index, or its inserts are proven to change no decision)
now writes its new rows many to a statement, up to 1,000 at a time, in
one savepoint: each table's domain rows in one insert, parents' tables
first, then their trackers. A row qualifies when the batch inserts it once
and touches it no other way, no tracker exists for it, nothing earlier in
the batch names it, and every row of the batch it names was written
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, the presence it claimed for its rows is
forgotten, and it and every row after it are left to the per-row path,
which tells those cases apart. Attempted values read the trackers the
merge holds instead of querying them.

The proof for merged inserts also accepts a parent of the row's own
table inserted earlier in the batch, so a library's nested studysets no
longer send its first push through the pass; a chunk ends before a row
whose parent of its own table it holds. A row naming itself, or a parent
inserted after it, still takes the pass.

Measured on PostgreSQL, the 24,650 changes merge in about 5 s instead of
34 to 37 s, and with the studysets nested instead of 41 s. A device on
SQLite, whose adapter still inserts row by row, receives the library in
44 s instead of 60 s. The new tests cover a batch spanning chunks, a chunk
failing on a row whose tracker was lost and falling back, and a tree of
one table's rows inserted parents first (the shortcut, a chunk per level)
and children first (the pass). The core, server, integration and
simulation suites pass, but for two PostgreSQL tests that can't start
their database while another embedded PostgreSQL runs.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

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,
+    Transaction transaction, {
-  ) async {
+    // 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(
+        _newMergeRow(tableName, rowId, remoteNode, incomingHlc, transaction),
-          spaceId: _context.hlcManagerFor(transaction).normalizedSpaceId,
-          tblId: tableId,
-          uuidRowId: rowId,
-          nodeId: remoteNode.id!,
-          hlcDatetime: incomingHlc.datetime,
-          hlcCounter: incomingHlc.counter,
-        ),
         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, {
+          withExplicitInsertNulls(spaceScopedRow, explicitNulls),
-            for (final column in spaceScopedRow.table.crdtSyncableColumns)
-              if (column.hasDefault && data[column.columnName] == null)
-                column.columnName,
-          }),
           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 _databaseContext.rebuildProjectionsOnce(
-      await _rebuildProjectionsForAllSpaces();
+      _rebuildProjectionsForAllSpaces,
-    }
+    );
     if (persistentUserId != null) {
       await _context.spaceManager.getOrCreate(persistentUserId!);
     }
@@ -1232,12 +1244,19 @@ class CrdtMutationRecorder {
       null,
       CrdtDataDeletedReason.userDelete,
     );
+    // 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,
+      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,6 +27,12 @@ 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});
 
@@ -275,6 +281,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 +376,7 @@ class OfflineSyncEngine {
           batch,
           outboundSpaces,
           onMergeSuccess,
+          onInboundMerge,
         );
 
         if (once) {
@@ -429,6 +437,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 +471,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);

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();
+      });
+    });
+  });
+}