perf: Read the domain rows a collection sends in batches

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

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

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

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

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

@@ -39,6 +39,11 @@ 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.
@@ -150,8 +155,9 @@ class OfflineSyncEngine {
   /// scoped by those internal ids. Node ids are stable per replica and may
   /// appear in multiple spaces, so checkpoint filtering must compare both
   /// `spaceId` and `uuidNodeId`. This still runs one query per change kind for
-  /// the whole pass. Per-row ownership and integrity checks resolve against
+  /// the whole pass, and reads the domain rows those changes name a batch of
-  /// each row's own space.
+  /// 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
@@ -581,13 +587,16 @@ class OfflineSyncEngine {
     Map<int, List<Hlc>> checkpointsBySpaceId,
     Transaction transaction,
   ) async* {
-    // Cache ownership within this collection's consistent snapshot.
+    // Cache ownership and domain rows within this collection's consistent
+    // snapshot.
     final ownerCache = DomainRowOwnerCache();
+    final domainRows = _DomainRowReads();
     yield* _streamInserts(
       session,
       spaceUuidById,
       checkpointsBySpaceId,
       ownerCache,
+      domainRows,
       transaction,
     );
     yield* _streamUpdates(
@@ -595,6 +604,7 @@ class OfflineSyncEngine {
       spaceUuidById,
       checkpointsBySpaceId,
       ownerCache,
+      domainRows,
       transaction,
     );
     yield* _streamDeletes(
@@ -611,6 +621,7 @@ class OfflineSyncEngine {
     Map<int, UuidValue> spaceUuidById,
     Map<int, List<Hlc>> checkpointsBySpaceId,
     DomainRowOwnerCache ownerCache,
+    _DomainRowReads domainRows,
     Transaction transaction,
   ) async* {
     final rows = await CrdtDataRow.db.find(
@@ -629,6 +640,19 @@ class OfflineSyncEngine {
       transaction,
     );
 
+    await _readDomainRowsAhead(
+      session,
+      [
+        for (final row in rows)
+          if (_syncTablesByName.containsKey(row.tbl!.name) &&
+              _classNamesByTableName.containsKey(row.tbl!.name))
+            (row.tbl!.name, row.spaceId, row.uuidRowId),
+      ],
+      ownerCache,
+      domainRows,
+      transaction,
+    );
+
     for (final row in rows) {
       final tableName = row.tbl!.name;
       if (!_syncTablesByName.containsKey(tableName)) continue;
@@ -649,6 +673,7 @@ class OfflineSyncEngine {
         attemptedValueFieldsByRowId[row.id!],
         spaceId,
         ownerCache,
+        domainRows,
         transaction,
       );
       if (!domainRow.exists) {
@@ -695,6 +720,7 @@ class OfflineSyncEngine {
     Map<int, UuidValue> spaceUuidById,
     Map<int, List<Hlc>> checkpointsBySpaceId,
     DomainRowOwnerCache ownerCache,
+    _DomainRowReads domainRows,
     Transaction transaction,
   ) async* {
     final fields = await CrdtDataField.db.find(
@@ -709,14 +735,27 @@ class OfflineSyncEngine {
       ),
     );
 
+    bool isUpdate(CrdtDataField field) =>
+        _syncTablesByName.containsKey(field.row!.tbl!.name) &&
+        !(field.hlcDatetime == field.row!.hlcDatetime &&
+            field.hlcCounter == field.row!.hlcCounter &&
+            field.nodeId == field.row!.nodeId);
+
+    await _readDomainRowsAhead(
+      session,
+      [
+        for (final field in fields)
+          if (isUpdate(field))
+            (field.row!.tbl!.name, field.row!.spaceId, field.row!.uuidRowId),
+      ],
+      ownerCache,
+      domainRows,
+      transaction,
+    );
+
     for (final field in fields) {
+      if (!isUpdate(field)) continue;
       final tableName = field.row!.tbl!.name;
-      if (!_syncTablesByName.containsKey(tableName)) continue;
-      if (field.hlcDatetime == field.row!.hlcDatetime &&
-          field.hlcCounter == field.row!.hlcCounter &&
-          field.nodeId == field.row!.nodeId) {
-        continue;
-      }
 
       final spaceId = field.row!.spaceId;
       final spaceUuid = spaceUuidById[spaceId]!;
@@ -729,6 +768,7 @@ class OfflineSyncEngine {
         field.attemptedValue,
         spaceId,
         ownerCache,
+        domainRows,
         transaction,
       );
       if (!columnValue.exists) {
@@ -878,6 +918,8 @@ class OfflineSyncEngine {
   /// targeted SQL projection also avoids fetching and serializing unrelated
   /// columns. This is the inverse of inbound FK materialization: the wire payload
   /// carries attempted facts, not locally projected visible values.
+  ///
+  /// A row in [domainRows] for [spaceId] is not read again.
   Future<({bool exists, int? ownerSpaceId, dynamic row})> _fetchDomainRow(
     DatabaseSession session,
     String tableName,
@@ -887,36 +929,37 @@ class OfflineSyncEngine {
     List<CrdtDataField>? attemptedValueFields,
     int spaceId,
     DomainRowOwnerCache ownerCache,
+    _DomainRowReads domainRows,
     Transaction transaction,
   ) async {
-    final cols = table.columns
+    final Map<String, dynamic> rawColumns;
-        .map(
+    final readAhead = domainRows[(tableName, rowId)];
-          (column) =>
+    if (readAhead != null && readAhead.spaceId == spaceId) {
-              '${_outboundColumnExpression(session, column)} AS "${column.columnName.escapeIdentifier()}"',
+      rawColumns = readAhead.columns;
-        )
+    } else {
-        .join(', ');
+      final encodedRowId = rowId.sqlLiteral();
-    final encodedRowId = rowId.sqlLiteral();
+      final encodedSpaceId = spaceId.sqlLiteral();
-    final encodedSpaceId = spaceId.sqlLiteral();
+      final escapedTableName = tableName.escapeIdentifier();
-    final escapedTableName = tableName.escapeIdentifier();
+      final result = await session.db.unsafeQuery(
-    final result = await session.db.unsafeQuery(
+        'SELECT ${_outboundColumns(session, table)} FROM "$escapedTableName" '
-      'SELECT $cols FROM "$escapedTableName" '
+        'WHERE "id" = $encodedRowId AND "spaceId" = $encodedSpaceId '
-      'WHERE "id" = $encodedRowId AND "spaceId" = $encodedSpaceId '
+        'LIMIT 1',
-      'LIMIT 1',
+        transaction: transaction,
-      transaction: transaction,
-    );
-    if (result.isEmpty) {
-      final owner = await _readDomainRowOwner(
-        session,
-        tableName,
-        rowId,
-        ownerCache,
-        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)
@@ -960,6 +1003,7 @@ class OfflineSyncEngine {
     CrdtDataAttemptedValue? attempted,
     int spaceId,
     DomainRowOwnerCache ownerCache,
+    _DomainRowReads domainRows,
     Transaction transaction,
   ) async {
     if (attempted != null) {
@@ -980,12 +1024,25 @@ class OfflineSyncEngine {
       );
     }
 
-    final encodedRowId = rowId.sqlLiteral();
-    final encodedSpaceId = spaceId.sqlLiteral();
-    final escapedTableName = tableName.escapeIdentifier();
     final column = _syncTablesByName[tableName]!.columns.singleWhere(
       (c) => c.columnName == columnName,
     );
+    final readAhead = domainRows[(tableName, rowId)];
+    if (readAhead != null && readAhead.spaceId == spaceId) {
+      return (
+        exists: true,
+        ownerSpaceId: spaceId,
+        value: _decodeColumnValue(
+          tableName,
+          columnName,
+          _decodeStructuredValue(session, column, readAhead.columns[columnName]),
+        ),
+      );
+    }
+
+    final encodedRowId = rowId.sqlLiteral();
+    final encodedSpaceId = spaceId.sqlLiteral();
+    final escapedTableName = tableName.escapeIdentifier();
     final result = await session.db.unsafeQuery(
       'SELECT ${_outboundColumnExpression(session, column)} '
       'FROM "$escapedTableName" '
@@ -1016,6 +1073,58 @@ class OfflineSyncEngine {
     return (exists: owner.exists, ownerSpaceId: owner.spaceId, value: null);
   }
 
+  /// Reads the domain rows [rowKeys] name into [domainRows], those of one
+  /// table and space [_domainRowReadBatchSize] to a statement, and records
+  /// the owner of each row found. Rows already read are skipped; a row not
+  /// found is left to the per-row reads, which tell its cases apart.
+  Future<void> _readDomainRowsAhead(
+    DatabaseSession session,
+    Iterable<(String, int, UuidValue)> rowKeys,
+    DomainRowOwnerCache ownerCache,
+    _DomainRowReads domainRows,
+    Transaction transaction,
+  ) async {
+    final rowIdsByTableAndSpace = <(String, int), Set<UuidValue>>{};
+    for (final (tableName, spaceId, rowId) in rowKeys) {
+      if (domainRows.containsKey((tableName, rowId))) continue;
+      rowIdsByTableAndSpace.putIfAbsent((tableName, spaceId), () => {}).add(rowId);
+    }
+
+    for (final MapEntry(key: (tableName, spaceId), value: rowIds)
+        in rowIdsByTableAndSpace.entries) {
+      final columns = _outboundColumns(session, _syncTablesByName[tableName]!);
+      final escapedTableName = tableName.escapeIdentifier();
+      final encodedSpaceId = spaceId.sqlLiteral();
+      final ids = rowIds.toList();
+      for (var start = 0; start < ids.length; start += _domainRowReadBatchSize) {
+        final batch = ids.skip(start).take(_domainRowReadBatchSize);
+        final result = await session.db.unsafeQuery(
+          'SELECT $columns FROM "$escapedTableName" '
+          'WHERE "spaceId" = $encodedSpaceId AND "id" IN (${batch.sqlLiteralList()})',
+          transaction: transaction,
+        );
+        for (final row in result) {
+          final rawColumns = row.toColumnMap();
+          final rowId = (rawColumns['id'] as Object?).toUuidValue()!;
+          domainRows[(tableName, rowId)] = (spaceId: spaceId, columns: rawColumns);
+          ownerCache[(tableName, rowId)] = (exists: true, spaceId: spaceId);
+        }
+      }
+    }
+  }
+
+  /// Rows of one table and space read ahead in one statement at most: bounds
+  /// the size of each statement.
+  static const _domainRowReadBatchSize = 500;
+
+  /// The outbound column expressions of [table], each named for its column.
+  String _outboundColumns(DatabaseSession session, Table table) => table.columns
+      .map(
+        (column) =>
+            '${_outboundColumnExpression(session, column)} AS "${column.columnName.escapeIdentifier()}"',
+      )
+      .join(', ');
+
   Future<DomainRowOwner> _readDomainRowOwner(
     DatabaseSession session,
     String tableName,

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

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

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

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