feat: Write a row in its own space, and a new one where the host says

A client with a persistent user wrote every change in its personal
space: an update or delete of a row in a shared space it is a member of
found no row and threw, and a new row could reach a shared space only
through transactionForUser, which a host calls at each write itself.

A write without a transaction now acts in the space it touches, through
transactionForUser, which checks membership and role as before: an
update or delete in the space its rows are stored in, a new row in the
one `newRowSpace` names (null, or no hook, for the personal space). One
call touches one space and is refused, with a StateError, where it
would touch two. A write within a transaction acts in its space, as
before; the server, without a persistent user, is unchanged.

Tested on the client: a writing member updates and deletes a shared
row, by row and by condition; a reading member is refused; a call
naming rows of two spaces is refused; new rows go where the hook says,
refused for a space only read or not a member's; a transaction of the
caller's own still decides. The integration suite passes (994 tests).

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

@@ -85,6 +85,19 @@ class OfflineSyncDatabase implements Database {
 
   final CrdtMutationRecorder _recorder;
 
+  /// Which space a new row goes in, for a client with a persistent user
+  /// writing without a transaction: a space the user is a member of, or
+  /// null for their personal space. Without it, new rows go in the
+  /// personal space.
+  ///
+  /// A write without a transaction acts in the space it touches: rows
+  /// updated or deleted in their own space, new rows in the one this
+  /// names, each through [transactionForUser], which checks that the user
+  /// may write there. One call touches one space, and is refused
+  /// ([StateError]) where it would touch two. A write within a transaction
+  /// acts in that transaction's space, as before.
+  FutureOr<UuidValue?> Function(TableRow row)? newRowSpace;
+
   late final _sync = OfflineSyncEngine(
     syncTables: _syncTables,
     serializationManager: serializationManager,
@@ -120,6 +133,7 @@ class OfflineSyncDatabase implements Database {
     UuidValue? userId,
     bool once = false,
     OfflineSyncOnMergeSuccess? onMergeSuccess,
+    OfflineSyncOnInboundMerge? onInboundMerge,
   }) async* {
     await _ensureInitialized();
     final effectiveUserId = await _requireUserId(userId);
@@ -129,6 +143,7 @@ class OfflineSyncDatabase implements Database {
       inbound: inbound,
       once: once,
       onMergeSuccess: onMergeSuccess,
+      onInboundMerge: onInboundMerge,
       mode: mode,
     );
   }
@@ -341,6 +356,21 @@ class OfflineSyncDatabase implements Database {
         noReturn: noReturn,
       );
     }
+    if (_routesWrite(transaction)) {
+      final space = await _newRowsSpace(rows);
+      if (space != null) {
+        return transactionForUser(
+          _recorder.persistentUserId!,
+          (tx) => insert<T>(
+            rows,
+            transaction: tx,
+            ignoreConflicts: ignoreConflicts,
+            noReturn: noReturn,
+          ),
+          spaceId: space,
+        );
+      }
+    }
     return _runTrackedWrite(
       transaction,
       (tx) async {
@@ -441,6 +471,31 @@ class OfflineSyncDatabase implements Database {
         noReturn: noReturn,
       );
     }
+    if (_routesWrite(transaction)) {
+      final table = rows.first.table;
+      final ids = {
+        for (final row in rows)
+          if (row.id case final Object id) id,
+      };
+      final stored = ids.isEmpty
+          ? null
+          : await _storedRowsSpace<T>(table.id.inSet(ids.castToIdType().toSet()));
+      final space = stored ?? await _newRowsSpace(rows);
+      if (space != null) {
+        return transactionForUser(
+          _recorder.persistentUserId!,
+          (tx) => upsert<T>(
+            rows,
+            conflictColumns: conflictColumns,
+            updateColumns: updateColumns,
+            updateWhere: updateWhere,
+            transaction: tx,
+            noReturn: noReturn,
+          ),
+          spaceId: space,
+        );
+      }
+    }
     return _runTrackedWrite(
       transaction,
       (tx) async {
@@ -661,6 +716,23 @@ class OfflineSyncDatabase implements Database {
         noReturn: noReturn,
       );
     }
+    if (_routesWrite(transaction)) {
+      final space = await _storedRowsSpace<T>(
+        rows.first.table.id.inSet(rows.map((row) => row.id).castToIdType().toSet()),
+      );
+      if (space != null) {
+        return transactionForUser(
+          _recorder.persistentUserId!,
+          (tx) => update<T>(
+            rows,
+            columns: columns,
+            transaction: tx,
+            noReturn: noReturn,
+          ),
+          spaceId: space,
+        );
+      }
+    }
     return _runTrackedWrite(
       transaction,
       (tx) async {
@@ -701,6 +773,16 @@ class OfflineSyncDatabase implements Database {
         transaction: transaction,
       );
     }
+    if (_routesWrite(transaction)) {
+      final space = await _storedRowsSpace<T>(row.table.id.equals(row.id));
+      if (space != null) {
+        return transactionForUser(
+          _recorder.persistentUserId!,
+          (tx) => updateRow<T>(row, columns: columns, transaction: tx),
+          spaceId: space,
+        );
+      }
+    }
     return _runTrackedWrite(
       transaction,
       (tx) async {
@@ -806,6 +888,26 @@ class OfflineSyncDatabase implements Database {
     }
 
     _assertNoSpaceIdColumnValues<T>(columnValues);
+    if (_routesWrite(transaction)) {
+      final space = await _storedRowsSpace<T>(where);
+      if (space != null) {
+        return transactionForUser(
+          _recorder.persistentUserId!,
+          (tx) => updateWhere<T>(
+            columnValues: columnValues,
+            where: where,
+            limit: limit,
+            offset: offset,
+            orderBy: orderBy,
+            orderByList: orderByList,
+            orderDescending: orderDescending,
+            transaction: tx,
+            noReturn: noReturn,
+          ),
+          spaceId: space,
+        );
+      }
+    }
     return _runTrackedWrite(
       transaction,
       (tx) async {
@@ -915,6 +1017,23 @@ class OfflineSyncDatabase implements Database {
       );
     }
 
+    if (_routesWrite(transaction)) {
+      final space = await _storedRowsSpace<T>(where);
+      if (space != null) {
+        return transactionForUser(
+          _recorder.persistentUserId!,
+          (tx) => deleteWhere<T>(
+            where: where,
+            orderBy: orderBy,
+            orderByList: orderByList,
+            orderDescending: orderDescending,
+            transaction: tx,
+            noReturn: noReturn,
+          ),
+          spaceId: space,
+        );
+      }
+    }
     return _runTrackedWrite(
       transaction,
       (tx) async {

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/database/space.dart

@@ -80,6 +80,63 @@ extension _OfflineSyncDatabaseSpace on OfflineSyncDatabase {
     }
   }
 
+  /// Whether a write chooses its space itself ([OfflineSyncDatabase.newRowSpace]):
+  /// one without a transaction, on a client with a persistent user.
+  bool _routesWrite(Transaction? transaction) =>
+      transaction == null && _recorder.persistentUserId != null;
+
+  /// The space other than the personal one that new [rows] go in, as
+  /// [OfflineSyncDatabase.newRowSpace] says; null for the personal one.
+  /// Throws a [StateError] where the rows go in two spaces.
+  Future<UuidValue?> _newRowsSpace<T extends TableRow>(List<T> rows) async {
+    final route = newRowSpace;
+    if (route == null) return null;
+    final personal = _recorder.persistentUserId!;
+    final spaces = <UuidValue>{
+      for (final row in rows)
+        if (_readSpaceId(row) == null) await route(row) ?? personal,
+    };
+    if (spaces.length > 1) {
+      throw StateError(
+        'Cannot write new ${rows.first.table.tableName} rows into the spaces '
+        '$spaces in one call.',
+      );
+    }
+    final space = spaces.singleOrNull;
+    return space == personal ? null : space;
+  }
+
+  /// The space other than the personal one that the visible rows matching
+  /// [where] are stored in; null where they are personal, or none match.
+  /// Throws a [StateError] where they are stored in two spaces.
+  Future<UuidValue?> _storedRowsSpace<T extends TableRow>(Expression where) async {
+    final rows = await _delegate.find<T>(
+      where: await _whereVisibleWithTombstone<T>(
+        where,
+        null,
+        null,
+        membershipWide: true,
+      ),
+    );
+    final spaceIds = {
+      for (final row in rows)
+        if (_readSpaceId(row) case final int spaceId) spaceId,
+    };
+    if (spaceIds.isEmpty) return null;
+    if (spaceIds.length > 1) {
+      throw StateError(
+        'Cannot write ${rows.first.table.tableName} rows of several spaces in '
+        'one call.',
+      );
+    }
+    final space = await OfflineSyncSpace.db.findById(
+      _delegate.session,
+      spaceIds.single,
+    );
+    final uuid = space?.uuidSpaceId;
+    return uuid == null || uuid == _recorder.persistentUserId ? null : uuid;
+  }
+
   OfflineSyncSpace _requireEffectiveSpace(Transaction transaction) {
     return _recorder.spaceForQueries(transaction) ??
         (throw StateError(

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.
@@ -51,6 +62,16 @@ class OfflineSyncEngine {
 
     /// Delay between continuous sync rounds.
     this._continuousSyncInterval = defaultContinuousSyncInterval,
+
+    /// Whether [collectPendingChanges] hands out each change as soon as it is
+    /// read, its transaction open until the last, rather than once all of
+    /// them are. A large collection then starts sending at once, where
+    /// whole it sent nothing until its end, so a peer waiting longer than its
+    /// idle timeout closed the stream. Meant for PostgreSQL, whose
+    /// repeatable-read snapshot only reads: Serverpod's SQLite transactions
+    /// are write transactions, which would hold off the device's own writes
+    /// for as long as the peer takes to receive.
+    this._streamCollectedChanges = false,
   }) : _syncTables = syncTables,
        _serializationManager = serializationManager,
        _databaseContext =
@@ -76,6 +97,7 @@ class OfflineSyncEngine {
   final OfflineSyncDatabaseContext _databaseContext;
   final int _syncBatchSize;
   final Duration _continuousSyncInterval;
+  final bool _streamCollectedChanges;
 
   /// Wraps [database] in a CRDT-aware database using this sync context.
   OfflineSyncDatabase wrapDatabase(Database database, {UuidValue? persistentUserId}) {
@@ -144,8 +166,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
@@ -174,6 +197,16 @@ class OfflineSyncEngine {
     };
 
     try {
+      if (_streamCollectedChanges) {
+        await for (final change in _collectedAsRead(
+          session,
+          spaceUuidById,
+          checkpointsBySpaceId,
+        )) {
+          yield change;
+        }
+        return;
+      }
       final changes = await session.db.transaction(
         (transaction) => _streamPendingChanges(
           session,
@@ -181,9 +214,7 @@ class OfflineSyncEngine {
           checkpointsBySpaceId,
           transaction,
         ).toList(),
-        settings: const TransactionSettings(
-          isolationLevel: IsolationLevel.repeatableRead,
-        ),
+        settings: _collectionSettings,
       );
 
       for (final change in changes) {
@@ -194,6 +225,54 @@ class OfflineSyncEngine {
     }
   }
 
+  static const _collectionSettings = TransactionSettings(
+    isolationLevel: IsolationLevel.repeatableRead,
+  );
+
+  /// The pending changes, each handed out as it is read within the
+  /// collection's one transaction, which stays open until the last is out or
+  /// the listener cancels; an error, a violation among them, ends the stream
+  /// once the transaction has rolled back.
+  Stream<CrdtMergeChange> _collectedAsRead(
+    DatabaseSession session,
+    Map<int, UuidValue> spaceUuidById,
+    Map<int, List<Hlc>> checkpointsBySpaceId,
+  ) {
+    var cancelled = false;
+    late final StreamController<CrdtMergeChange> changes;
+    changes = StreamController<CrdtMergeChange>(
+      onListen: () {
+        unawaited(
+          session.db
+              .transaction(
+                (transaction) async {
+                  await for (final change in _streamPendingChanges(
+                    session,
+                    spaceUuidById,
+                    checkpointsBySpaceId,
+                    transaction,
+                  )) {
+                    // Nobody reads on: end the collection and its transaction.
+                    if (cancelled) return;
+                    changes.add(change);
+                  }
+                },
+                settings: _collectionSettings,
+              )
+              .then(
+                (_) => changes.close(),
+                onError: (Object error, StackTrace stackTrace) {
+                  if (!cancelled) changes.addError(error, stackTrace);
+                  return changes.close();
+                },
+              ),
+        );
+      },
+      onCancel: () => cancelled = true,
+    );
+    return changes.stream;
+  }
+
   /// Creates the [OfflineSyncSinceHlc] checkpoint for a space handshake.
   ///
   /// [OfflineSyncSinceHlc.nodeCheckpoints] reflects the latest change this node
@@ -275,6 +354,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 +449,7 @@ class OfflineSyncEngine {
           batch,
           outboundSpaces,
           onMergeSuccess,
+          onInboundMerge,
         );
 
         if (once) {
@@ -429,6 +510,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 +544,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 +654,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 +671,7 @@ class OfflineSyncEngine {
       spaceUuidById,
       checkpointsBySpaceId,
       ownerCache,
+      domainRows,
       transaction,
     );
     yield* _streamDeletes(
@@ -599,6 +688,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 +707,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 +740,7 @@ class OfflineSyncEngine {
         attemptedValueFieldsByRowId[row.id!],
         spaceId,
         ownerCache,
+        domainRows,
         transaction,
       );
       if (!domainRow.exists) {
@@ -683,6 +787,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 +802,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 +835,7 @@ class OfflineSyncEngine {
         field.attemptedValue,
         spaceId,
         ownerCache,
+        domainRows,
         transaction,
       );
       if (!columnValue.exists) {
@@ -866,6 +985,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 +996,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 +1070,7 @@ class OfflineSyncEngine {
     CrdtDataAttemptedValue? attempted,
     int spaceId,
     DomainRowOwnerCache ownerCache,
+    _DomainRowReads domainRows,
     Transaction transaction,
   ) async {
     if (attempted != null) {
@@ -968,12 +1091,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 +1140,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);
+    });
+  });
+}

packages/serverpod_offline_sync_server/lib/src/business/offline_sync.dart

@@ -37,16 +37,23 @@ extension OfflineSyncInitialize on Serverpod {
   ///
   /// [continuousSyncInterval] controls how long a continuous sync session waits
   /// after completing one sync round before checking for local changes again.
+  ///
+  /// [streamCollectedChanges] sends the changes a peer lacks as they are read
+  /// rather than once all are, so a peer's first sync of a large space hears
+  /// from the server before its idle timeout; the server's PostgreSQL holds
+  /// the collection's read-only snapshot meanwhile.
   void initializeOfflineSync({
     required List<Table> syncTables,
     int syncBatchSize = OfflineSyncEngine.defaultSyncBatchSize,
     Duration continuousSyncInterval = OfflineSyncEngine.defaultContinuousSyncInterval,
+    bool streamCollectedChanges = true,
   }) {
     _offlineSyncByServerpod[this] = OfflineSyncEngine(
       syncTables: syncTables,
       serializationManager: serializationManager,
       syncBatchSize: syncBatchSize,
       continuousSyncInterval: continuousSyncInterval,
+      streamCollectedChanges: streamCollectedChanges,
     );
   }
 }

test/serverpod_offline_sync_test_server/test/integration/crud/space_routing_test.dart

@@ -0,0 +1,217 @@
+import 'package:serverpod_offline_sync_client/serverpod_offline_sync_client.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';
+
+void main() {
+  initTestClientSession(withPersistentUser: true);
+
+  group(
+    'Given a persistent user who is a member of a shared space holding a row, '
+    'and a row of their own,',
+    () {
+      late UuidValue sharedSpaceUuid;
+      late OfflineSyncSpace sharedSpace;
+      late Person shared;
+      late Person own;
+
+      /// The space each person row is stored in, by name, read past the
+      /// engine.
+      Future<Map<String, int?>> storedSpaces() async => {
+        for (final person in await Person.db.find(testSession))
+          person.name: person.spaceId,
+      };
+
+      Future<void> joinAs(OfflineSyncSpaceRole role) async {
+        await OfflineSyncSpaceMember.db.insertRow(
+          session,
+          OfflineSyncSpaceMember(
+            spaceId: sharedSpace.id!,
+            userUuid: testCrdtUserId,
+            role: role,
+          ),
+        );
+      }
+
+      setUp(() async {
+        sharedSpaceUuid = const Uuid().v7obj();
+        shared = await session.db.transactionForUser(
+          sharedSpaceUuid,
+          (tx) => Person.db.insertRow(
+            session,
+            Person(name: 'shared'),
+            transaction: tx,
+          ),
+        );
+        sharedSpace = (await OfflineSyncSpace.db.findFirstRow(
+          testSession,
+          where: (t) => t.uuidSpaceId.equals(sharedSpaceUuid),
+        ))!;
+        own = await Person.db.insertRow(session, Person(name: 'own'));
+      });
+
+      test(
+        'when a writing member updates and deletes the shared row without a '
+        'transaction, then it changes in its own space.',
+        () async {
+          await joinAs(OfflineSyncSpaceRole.readWrite);
+
+          final updated = await Person.db.updateRow(
+            session,
+            shared.copyWith(name: 'shared, changed'),
+          );
+          expect(updated.name, 'shared, changed');
+          expect((await storedSpaces())['shared, changed'], sharedSpace.id);
+
+          await Person.db.deleteRow(session, updated);
+          expect(await Person.db.findById(session, shared.id!), isNull);
+          expect(await Person.db.findById(session, own.id!), isNotNull);
+        },
+      );
+
+      test(
+        'when a writing member updates the shared rows by a condition, then '
+        'they change in their own space.',
+        () async {
+          await joinAs(OfflineSyncSpaceRole.readWrite);
+
+          await Person.db.updateWhere(
+            session,
+            columnValues: (t) => [t.name('renamed')],
+            where: (t) => t.name.equals('shared'),
+          );
+          await Person.db.deleteWhere(
+            session,
+            where: (t) => t.name.equals('renamed'),
+          );
+
+          expect(
+            [for (final p in await Person.db.find(session)) p.name],
+            ['own'],
+          );
+        },
+      );
+
+      test(
+        'when a reading member updates or deletes the shared row, then it is '
+        'refused and stays as it was.',
+        () async {
+          await joinAs(OfflineSyncSpaceRole.readOnly);
+
+          await expectLater(
+            Person.db.updateRow(session, shared.copyWith(name: 'changed')),
+            throwsA(isA<OfflineSyncSpaceRoleException>()),
+          );
+          await expectLater(
+            Person.db.deleteRow(session, shared),
+            throwsA(isA<OfflineSyncSpaceRoleException>()),
+          );
+          expect((await Person.db.findById(session, shared.id!))?.name, 'shared');
+        },
+      );
+
+      test(
+        'when one call names rows of two spaces, then it is refused and '
+        'nothing changes.',
+        () async {
+          await joinAs(OfflineSyncSpaceRole.readWrite);
+
+          await expectLater(
+            Person.db.update(session, [
+              shared.copyWith(name: 'a'),
+              own.copyWith(name: 'b'),
+            ]),
+            throwsA(isA<StateError>()),
+          );
+          expect(
+            {for (final p in await Person.db.find(session)) p.name},
+            {'shared', 'own'},
+          );
+        },
+      );
+
+      test(
+        'when the host routes a new row to the shared space, then it is '
+        'stored there; one it routes nowhere stays personal.',
+        () async {
+          await joinAs(OfflineSyncSpaceRole.readWrite);
+          session.db.newRowSpace = (row) =>
+              (row as Person).name.startsWith('to shared') ? sharedSpaceUuid : null;
+
+          await Person.db.insertRow(session, Person(name: 'to shared'));
+          await Person.db.insert(session, [Person(name: 'to shared, too')]);
+          await Person.db.insertRow(session, Person(name: 'mine'));
+
+          final spaces = await storedSpaces();
+          expect(spaces['to shared'], sharedSpace.id);
+          expect(spaces['to shared, too'], sharedSpace.id);
+          expect(spaces['mine'], spaces['own']);
+        },
+      );
+
+      test(
+        'when the host routes new rows of one call to two spaces, then it is '
+        'refused.',
+        () async {
+          await joinAs(OfflineSyncSpaceRole.readWrite);
+          session.db.newRowSpace = (row) =>
+              (row as Person).name == 'there' ? sharedSpaceUuid : null;
+
+          await expectLater(
+            Person.db.insert(session, [Person(name: 'there'), Person(name: 'here')]),
+            throwsA(isA<StateError>()),
+          );
+          expect(
+            {for (final p in await Person.db.find(session)) p.name},
+            {'shared', 'own'},
+          );
+        },
+      );
+
+      test(
+        'when the host routes a new row to a space the person may only read, '
+        'or is not a member of, then it is refused.',
+        () async {
+          await joinAs(OfflineSyncSpaceRole.readOnly);
+          final elsewhere = const Uuid().v7obj();
+          session.db.newRowSpace = (row) =>
+              (row as Person).name == 'read only' ? sharedSpaceUuid : elsewhere;
+
+          await expectLater(
+            Person.db.insertRow(session, Person(name: 'read only')),
+            throwsA(isA<OfflineSyncSpaceRoleException>()),
+          );
+          await expectLater(
+            Person.db.insertRow(session, Person(name: 'nowhere')),
+            throwsA(isA<OfflineSyncSpaceMembershipException>()),
+          );
+          expect(
+            {for (final p in await Person.db.find(session)) p.name},
+            {'shared', 'own'},
+          );
+        },
+      );
+
+      test(
+        'when a caller writes within a transaction of its own, then the '
+        'transaction decides the space, as before.',
+        () async {
+          await joinAs(OfflineSyncSpaceRole.readWrite);
+          session.db.newRowSpace = (_) => sharedSpaceUuid;
+
+          await session.db.transaction((tx) async {
+            await Person.db.insertRow(
+              session,
+              Person(name: 'in a transaction'),
+              transaction: tx,
+            );
+          });
+
+          final spaces = await storedSpaces();
+          expect(spaces['in a transaction'], spaces['own']);
+        },
+      );
+    },
+  );
+}

test/serverpod_offline_sync_test_server/test/integration/crud/tombstone_space_test.dart

@@ -369,69 +369,74 @@ void main() {
       );
     });
 
-    test(
-      'when collecting pending changes, '
-      'then the sync fails and records orphan metadata.',
-      () async {
-        final sync = OfflineSyncEngine(
-          syncTables: [Person.t],
-          serializationManager: session.db.serializationManager,
-        );
+    // Collected whole, as a device does, and sent while collected, as the
+    // server does: a violation ends either alike.
+    for (final streamed in [false, true]) {
+      test(
+        'when collecting pending changes${streamed ? ' as they are read' : ''}, '
+        'then the sync fails and records orphan metadata.',
+        () async {
+          final sync = OfflineSyncEngine(
+            syncTables: [Person.t],
+            serializationManager: session.db.serializationManager,
+            streamCollectedChanges: streamed,
+          );
 
-        await expectLater(
-          sync
-              .collectPendingChanges(
-                session,
-                checkpointsBySpaceUuid: {testCrdtUserId: const []},
-              )
-              .toList(),
-          throwsA(
-            isA<OfflineSyncIntegrityViolationException>()
-                .having(
-                  (e) => e.violation.operation,
-                  'violation.operation',
-                  OfflineSyncViolationOperation.outboundInsert,
+          await expectLater(
+            sync
+                .collectPendingChanges(
+                  session,
+                  checkpointsBySpaceUuid: {testCrdtUserId: const []},
                 )
-                .having(
-                  (e) => e.violation.type,
-                  'violation.type',
-                  OfflineSyncViolationType.missingDomainRow,
-                )
-                .having(
-                  (e) => e.violation.ownerSpaceUuid,
-                  'violation.ownerSpaceUuid',
-                  isNull,
-                )
-                .having(
-                  (e) => e.violation.incomingSpaceUuid,
-                  'violation.incomingSpaceUuid',
-                  testCrdtUserId,
-                )
-                .having((e) => e.violation.id, 'violation.id', isNotNull),
-          ),
-        );
+                .toList(),
+            throwsA(
+              isA<OfflineSyncIntegrityViolationException>()
+                  .having(
+                    (e) => e.violation.operation,
+                    'violation.operation',
+                    OfflineSyncViolationOperation.outboundInsert,
+                  )
+                  .having(
+                    (e) => e.violation.type,
+                    'violation.type',
+                    OfflineSyncViolationType.missingDomainRow,
+                  )
+                  .having(
+                    (e) => e.violation.ownerSpaceUuid,
+                    'violation.ownerSpaceUuid',
+                    isNull,
+                  )
+                  .having(
+                    (e) => e.violation.incomingSpaceUuid,
+                    'violation.incomingSpaceUuid',
+                    testCrdtUserId,
+                  )
+                  .having((e) => e.violation.id, 'violation.id', isNotNull),
+            ),
+          );
 
-        final violation = await OfflineSyncIntegrityViolation.db.findFirstRow(
-          session,
-          where: (t) =>
-              t.uuidRowId.equals(person.id) &
-              t.type.equals(OfflineSyncViolationType.missingDomainRow) &
-              t.operation.equals(OfflineSyncViolationOperation.outboundInsert),
-        );
+          final violation = await OfflineSyncIntegrityViolation.db.findFirstRow(
+            session,
+            where: (t) =>
+                t.uuidRowId.equals(person.id) &
+                t.type.equals(OfflineSyncViolationType.missingDomainRow) &
+                t.operation.equals(OfflineSyncViolationOperation.outboundInsert),
+          );
 
-        expect(violation, isNotNull);
-        expect(violation!.domainTableName, Person.t.tableName);
-        expect(violation.type, OfflineSyncViolationType.missingDomainRow);
-        expect(violation.operation, OfflineSyncViolationOperation.outboundInsert);
-        expect(violation.ownerSpaceUuid, isNull);
-        expect(violation.incomingSpaceUuid, testCrdtUserId);
-        expect(violation.crdtDataRowId, crdtRow.id);
-        expect(violation.uuidNodeId, crdtRow.node!.uuidNodeId);
-        expect(violation.hlcDatetime, crdtRow.hlcDatetime);
-        expect(violation.hlcCounter, crdtRow.hlcCounter);
-        expect(violation.occurrences, 1);
-      },
-    );
+          expect(violation, isNotNull);
+          expect(violation!.domainTableName, Person.t.tableName);
+          expect(violation.type, OfflineSyncViolationType.missingDomainRow);
+          expect(violation.operation, OfflineSyncViolationOperation.outboundInsert);
+          expect(violation.ownerSpaceUuid, isNull);
+          expect(violation.incomingSpaceUuid, testCrdtUserId);
+          expect(violation.crdtDataRowId, crdtRow.id);
+          expect(violation.uuidNodeId, crdtRow.node!.uuidNodeId);
+          expect(violation.hlcDatetime, crdtRow.hlcDatetime);
+          expect(violation.hlcCounter, crdtRow.hlcCounter);
+          expect(violation.occurrences, 1);
+        },
+      );
+    }
   });
 
   group('Given update CRDT metadata whose domain row is missing,', () {

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/server/streamed_collection_postgres_test.dart

@@ -0,0 +1,173 @@
+import 'dart:io';
+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_server/src/generated/protocol.dart'
+    as server;
+import 'package:test/test.dart';
+
+import '../test_tools/postgres_migrations.dart';
+import '../test_tools/query_log_database.dart';
+import '../test_tools/serverpod_test_tools.dart';
+
+const _heldCount = 600;
+const _newCount = 600;
+const _renamedCount = 300;
+
+/// The server's collection, sent as it is read, on the database it runs on:
+/// its repeatable-read snapshot stays open while the changes go out.
+void main() {
+  final serverDirectory = Directory(
+    '${Directory.systemTemp.path}/offline_sync_streamed_${const Uuid().v4()}',
+  );
+
+  setUpAll(() => preparePostgresMigrations(serverDirectory));
+
+  tearDownAll(() async {
+    if (serverDirectory.existsSync()) await serverDirectory.delete(recursive: true);
+  });
+
+  withServerpod(
+    '[PostgreSQL streamed collection]',
+    rollbackDatabase: RollbackDatabase.disabled,
+    serverDirectory: serverDirectory,
+    configOverride: (config) => config.copyWith(
+      database: PostgresDatabaseConfig.embedded(
+        dataPath: '${serverDirectory.path}/postgres',
+        name: 'serverpod_test',
+        maxConnectionCount: 6,
+      ),
+    ),
+    (sessionBuilder, _) {
+      final tables = [server.Types.t];
+      late UuidValue space;
+      late OfflineSyncDatabaseSession author;
+      late QueryLogDatabase queryLog;
+      late OfflineSyncDatabaseSession collector;
+      late List<Hlc> checkpoints;
+
+      OfflineSyncEngine engine({required bool streamed}) => OfflineSyncEngine(
+        syncTables: tables,
+        serializationManager: author.db.serializationManager,
+        streamCollectedChanges: streamed,
+      );
+
+      Stream<CrdtMergeChange> collect({required bool streamed}) =>
+          engine(streamed: streamed).collectPendingChanges(
+            collector,
+            checkpointsBySpaceUuid: {space: checkpoints},
+          );
+
+      setUpAll(() async {
+        space = const Uuid().v7obj();
+        author = OfflineSyncDatabaseSession.wraps(
+          sessionBuilder.build(),
+          syncTables: tables,
+        );
+        await author.db.initialize();
+        final held = await _insertTypes(author, space, 0, _heldCount);
+        checkpoints = (await engine(streamed: false).createSyncSinceHlc(
+          author,
+          spaceId: space,
+        )).nodeCheckpoints;
+        await _insertTypes(author, space, _heldCount, _newCount);
+        await author.db.transactionForUser(
+          space,
+          (tx) => server.Types.db.update(
+            author,
+            [
+              for (final row in held.take(_renamedCount))
+                row.copyWith(aText: 'renamed ${row.anInt}'),
+            ],
+            columns: (t) => [t.aText],
+            transaction: tx,
+          ),
+        );
+        queryLog = QueryLogDatabase(sessionBuilder.build().db, syncTables: tables);
+        collector = OfflineSyncDatabaseSession(queryLog, syncTables: tables);
+        await queryLog.initialize();
+      });
+
+      group('Given changes a peer lacks, collected as they are read,', () {
+        late int readsAtFirstChange;
+        late int readsInAll;
+        late List<CrdtMergeChange> streamed;
+
+        setUpAll(() async {
+          queryLog.rawQueries.clear();
+          int? atFirst;
+          streamed = [];
+          var wroteMeanwhile = false;
+          await for (final change in collect(streamed: true)) {
+            atFirst ??= queryLog.readsOf('types').length;
+            streamed.add(change);
+            if (!wroteMeanwhile) {
+              wroteMeanwhile = true;
+              // Another session writes while the collection runs.
+              await _insertTypes(author, space, _heldCount + _newCount, 1);
+            }
+          }
+          readsAtFirstChange = atFirst!;
+          readsInAll = queryLog.readsOf('types').length;
+        });
+
+        test('then the first change arrives before the renamed rows are read.', () {
+          expect(readsAtFirstChange, lessThan(readsInAll));
+        });
+
+        test('then a row written meanwhile stays out of the collection.', () {
+          expect(streamed.whereType<CrdtMergeInsert>(), hasLength(_newCount));
+          expect(streamed.whereType<CrdtMergeUpdate>(), hasLength(_renamedCount));
+        });
+
+        test('then the changes are those a whole collection of the same rows '
+            'holds.', () async {
+          final whole = await collect(streamed: false).toList();
+          final protocol = author.db.serializationManager;
+          String encoded(CrdtMergeChange change) =>
+              protocol.encodeWithTypeForProtocol(change).toString();
+          // The whole collection also holds the row written meanwhile.
+          final wholeEncoded = whole.map(encoded).toList();
+          final streamedEncoded = streamed.map(encoded).toList();
+          expect(wholeEncoded, containsAllInOrder(streamedEncoded));
+          expect(wholeEncoded, hasLength(streamedEncoded.length + 1));
+        });
+      });
+
+      group('Given a listener that stops after the first change,', () {
+        test('then the collection ends and the next one runs.', () async {
+          final first = await collect(streamed: true).first;
+          expect(first, isA<CrdtMergeInsert>());
+          final again = await collect(streamed: true).length;
+          expect(again, _newCount + 1 + _renamedCount);
+        });
+      });
+    },
+  );
+}
+
+Future<List<server.Types>> _insertTypes(
+  OfflineSyncDatabaseSession node,
+  UuidValue space,
+  int from,
+  int count,
+) => node.db.transactionForUser(
+  space,
+  (tx) => server.Types.db.insert(
+    node,
+    [
+      for (var i = from; i < from + count; i++)
+        server.Types(
+          aBool: i.isEven,
+          aDateTime: DateTime.utc(2026, 11, 9).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(),
+        ),
+    ],
+    transaction: tx,
+  ),
+);