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,
@@ -343,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 {
@@ -443,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 {
@@ -663,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 {
@@ -703,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 {
@@ -808,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 {
@@ -917,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/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/engine.dart

@@ -62,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 =
@@ -87,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}) {
@@ -186,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,
@@ -193,9 +214,7 @@ class OfflineSyncEngine {
           checkpointsBySpaceId,
           transaction,
         ).toList(),
-        settings: const TransactionSettings(
+        settings: _collectionSettings,
-          isolationLevel: IsolationLevel.repeatableRead,
-        ),
       );
 
       for (final change in changes) {
@@ -206,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

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

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,
+  ),
+);

test/serverpod_offline_sync_test_server/test/integration/sync/streamed_collection_test.dart

@@ -0,0 +1,184 @@
+import 'dart:async';
+import 'dart:typed_data';
+
+import 'package:serverpod_offline_sync_server/serverpod_offline_sync_server.dart';
+import 'package:serverpod_offline_sync_test_client/serverpod_offline_sync_test_client.dart';
+import 'package:test/test.dart';
+
+import '../test_tools/client_session.dart';
+import '../test_tools/query_log_database.dart';
+import '../test_tools/sync_topology.dart';
+
+const _heldCount = 600;
+const _newCount = 600;
+const _renamedCount = 300;
+
+/// A peer that synced [_heldCount] rows long ago collects what happened
+/// since: [_newCount] rows added and [_renamedCount] of the old ones renamed.
+/// The inserts are handed out one by one once their rows are read; the
+/// renames' rows are read only after the last insert, so the query log tells
+/// whether a change reached the consumer while the collection still ran.
+void main() {
+  initTestClientSession(createSessionPerTest: false);
+
+  for (final streamed in [true, false]) {
+    group(
+      'Given a peer $_newCount inserts and $_renamedCount renames behind, '
+      'collected ${streamed ? 'as they are read' : 'whole'},',
+      () {
+        late SyncNode author;
+        late SyncNode peer;
+        late QueryLogDatabase queryLog;
+        late OfflineSyncDatabaseSession collector;
+        late OfflineSyncEngine engine;
+        late List<Hlc> checkpoints;
+
+        setUpAll(() async {
+          author = await syncNode(await createAdditionalTestSession(), [Types.t]);
+          peer = await syncNode(await createAdditionalTestSession(), [Types.t]);
+          final held = await _insertTypes(author, 0, _heldCount);
+          await pushChanges(author, peer);
+          checkpoints = (await peer.sync.createSyncSinceHlc(
+            peer.raw,
+            spaceId: testCrdtUserId,
+          )).nodeCheckpoints;
+          await _insertTypes(author, _heldCount, _newCount);
+          await _rename(author, held.take(_renamedCount));
+          queryLog = QueryLogDatabase(author.raw.db, syncTables: [Types.t]);
+          collector = OfflineSyncDatabaseSession(queryLog, syncTables: [Types.t]);
+          await queryLog.initialize();
+          engine = OfflineSyncEngine(
+            syncTables: [Types.t],
+            serializationManager: author.raw.db.serializationManager,
+            streamCollectedChanges: streamed,
+          );
+        });
+
+        group('when the peer collects since its checkpoint,', () {
+          late int readsAtFirstChange;
+          late int readsInAll;
+          late List<CrdtMergeChange> changes;
+          late Map<UuidValue, Map<String, dynamic>> authored;
+          late Map<UuidValue, Map<String, dynamic>> received;
+
+          setUpAll(() async {
+            queryLog.rawQueries.clear();
+            int? atFirst;
+            changes = [];
+            await for (final change in engine.collectPendingChanges(
+              collector,
+              checkpointsBySpaceUuid: {testCrdtUserId: checkpoints},
+            )) {
+              atFirst ??= queryLog.readsOf('types').length;
+              changes.add(change);
+            }
+            readsAtFirstChange = atFirst!;
+            readsInAll = queryLog.readsOf('types').length;
+            await peer.offlineSync.db.mergeChanges(
+              _overTheWire(author, changes),
+              spaceId: testCrdtUserId,
+            );
+            authored = await _rowsById(author);
+            received = await _rowsById(peer);
+          });
+
+          if (streamed) {
+            test('then the first change arrives before the renamed rows are read.', () {
+              expect(readsAtFirstChange, lessThan(readsInAll));
+            });
+          } else {
+            test('then the first change arrives once every row is read.', () {
+              expect(readsAtFirstChange, readsInAll);
+            });
+          }
+
+          test('then every insert and every rename is collected.', () {
+            expect(changes.whereType<CrdtMergeInsert>(), hasLength(_newCount));
+            expect(changes.whereType<CrdtMergeUpdate>(), hasLength(_renamedCount));
+          });
+
+          test("then the peer holds exactly the author's rows.", () {
+            expect(received, authored);
+          });
+        });
+
+        if (streamed) {
+          group('when the consumer stops after the first change,', () {
+            test('then the collection ends without an error.', () async {
+              final first = await engine
+                  .collectPendingChanges(
+                    collector,
+                    checkpointsBySpaceUuid: {testCrdtUserId: checkpoints},
+                  )
+                  .first;
+              expect(first, isA<CrdtMergeInsert>());
+              // The collection's transaction ends: the next one isn't held up.
+              final again = await engine
+                  .collectPendingChanges(
+                    collector,
+                    checkpointsBySpaceUuid: {testCrdtUserId: checkpoints},
+                  )
+                  .length;
+              expect(again, _newCount + _renamedCount);
+            });
+          });
+        }
+      },
+    );
+  }
+}
+
+Future<List<Types>> _insertTypes(SyncNode node, int from, int count) =>
+    node.offlineSync.db.transactionForUser(
+      testCrdtUserId,
+      (tx) => Types.db.insert(
+        node.offlineSync,
+        [
+          for (var i = from; i < from + count; i++)
+            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(),
+              jsonbDocument: SyncDocument(
+                title: 'JSONB $i',
+                enabled: i.isEven,
+                numbers: [i],
+              ),
+            ),
+        ],
+        transaction: tx,
+      ),
+    );
+
+Future<void> _rename(SyncNode node, Iterable<Types> rows) =>
+    node.offlineSync.db.transactionForUser(
+      testCrdtUserId,
+      (tx) => Types.db.update(
+        node.offlineSync,
+        [for (final row in rows) row.copyWith(aText: 'renamed ${row.anInt}')],
+        columns: (t) => [t.aText],
+        transaction: tx,
+      ),
+    );
+
+/// The changes as the peer decodes them from the sync stream.
+List<CrdtMergeChange> _overTheWire(SyncNode node, List<CrdtMergeChange> changes) {
+  final protocol = node.raw.db.serializationManager;
+  return [
+    for (final change in changes)
+      protocol.decodeWithType(protocol.encodeWithTypeForProtocol(change))!
+          as CrdtMergeChange,
+  ];
+}
+
+/// The model's JSON, without the space it is stored in locally.
+Map<String, dynamic> _payload(Types row) => row.toJson()..remove('spaceId');
+
+Future<Map<UuidValue, Map<String, dynamic>>> _rowsById(SyncNode node) async {
+  final rows = await Types.db.find(node.offlineSync);
+  return {for (final row in rows) row.id!: _payload(row)};
+}