perf: Send the changes a peer lacks as they are read, on the server

Collecting pending changes ran in one repeatable-read transaction and
handed nothing out until every change was read. A device's first sync of
a large space thus heard nothing from the server until the end: on a home
server, a library of 56,621 rows took about 100 s, so the client's idle
timeout (40 s, its pings unanswered meanwhile) closed the stream first,
and every attempt started over. Reading the rows in batches made that
shorter, not bounded.

An engine may now hand each change out as soon as it is read
(streamCollectedChanges), the collection's transaction staying open until
the last is out or the listener cancels; cancelling ends the collection
and its transaction, and an error, an integrity violation among them,
reaches the listener once the transaction rolled back, recorded as before.
The server's initializeOfflineSync switches it on: PostgreSQL's snapshot
only reads. A device's engine keeps it off, since Serverpod's SQLite
transactions are write transactions and would hold off the device's own
writes for as long as the peer takes to receive. A receiver merges a batch
only once its end arrives, so a stream broken halfway still merges
nothing.

The new tests collect 600 inserts and 300 renames for a peer behind: as
read, the first change arrives before the renamed rows are read; whole, once
all are; both merge into the author's rows. On embedded PostgreSQL, a row
written by another session during the streamed collection stays out of it,
the changes are a whole collection's, in order, and a listener stopping
after the first lets the next collection run. The orphan-insert violation
test runs both ways. Every PostgreSQL test file passes alone; run together,
the files bind the same port, the two already there as well.

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

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