fix: Rebuild projections once per context after the registry changed

Every ephemeral OfflineSyncDatabase shares one context but initializes a
recorder of its own, and each rebuilt the projections of every space
when the context's first initialize had changed the registry: on every
session and every merge, until the process stopped (upstream #150). A
new device, or one whose app changed its synced tables, spent about 3 s
of every inbound merge on it with a library of 8,000 cards, a server
after such a migration as much on every request.

The context now runs the rebuild once (Once): wrappers initializing
while it runs await the same run, later ones skip it, and a failed run
is tried again by the next.

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

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
@@ -362,8 +371,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 +403,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 +416,49 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
     );
   }
 
+  /// Whether this batch leaves every projection decision as it is though it
+  /// touches foreign keys: it only inserts new rows, each once, and updates
+  /// columns no projection reads, and the inserts are proven
+  /// ([CrdtForeignKeyProjector.canLeaveProjectionUnchangedByMergedInserts]).
+  /// A delete, or an insert onto a row there already, 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>{};
+    for (final operation in operations) {
+      if (!_context.isCrdtTrackedTableName(operation.tableName)) continue;
+      switch (operation) {
+        case CrdtMergeDelete():
+          return false;
+        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;
+      }
+    }
+    return _foreignKeyProjector.canLeaveProjectionUnchangedByMergedInserts(
+      insertsByTable,
+      positions,
+      transaction,
+    );
+  }
+
   Future<void> _applyMergeInsert(
     CrdtMergeInsert insert,
     Map<UuidValue, CrdtNode> remoteNodes,

packages/serverpod_offline_sync/lib/src/database/merge_utils/foreign_key_projector.dart

@@ -756,6 +756,99 @@ 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 (and proven alike; one inserted after its child 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);
+          // A row of the same table inserted by the same merge would have
+          // to be proven first; the pass orders that.
+          if (id == null || (edge.parentTableName == tableName && ids.contains(id))) {
+            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.

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!);
     }

packages/serverpod_offline_sync/lib/src/utils/once.dart

@@ -0,0 +1,21 @@
+import 'package:meta/meta.dart';
+
+/// Runs an action once for everyone who asks: a caller while it runs awaits
+/// the same run, and one after it is done gets it done; only a run that
+/// failed is run again by the next caller.
+@internal
+class Once {
+  Future<void>? _run;
+
+  /// Runs [action] unless it ran or runs already, and completes with that
+  /// run.
+  Future<void> run(Future<void> Function() action) async {
+    final run = _run ??= action();
+    try {
+      await run;
+    } catch (_) {
+      if (identical(_run, run)) _run = null;
+      rethrow;
+    }
+  }
+}

packages/serverpod_offline_sync/test/utils/once_test.dart

@@ -0,0 +1,47 @@
+import 'dart:async';
+
+import 'package:serverpod_offline_sync/src/utils/once.dart';
+import 'package:test/test.dart';
+
+void main() {
+  group('Given an action run once,', () {
+    late Once once;
+    late int runs;
+    late Completer<void> gate;
+
+    setUp(() {
+      once = Once();
+      runs = 0;
+      gate = Completer<void>();
+    });
+
+    Future<void> action() async {
+      runs++;
+      await gate.future;
+    }
+
+    test('when asked while it runs and after, then it runs once and every '
+        'caller awaits that run.', () async {
+      final first = once.run(action);
+      final meanwhile = once.run(action);
+      gate.complete();
+      await Future.wait([first, meanwhile]);
+      await once.run(action);
+
+      expect(runs, 1);
+    });
+
+    test('when its run fails, then the next caller runs it again.', () async {
+      final failing = once.run(() async {
+        runs++;
+        throw StateError('failed');
+      });
+      await expectLater(failing, throwsStateError);
+
+      gate.complete();
+      await once.run(action);
+
+      expect(runs, 2);
+    });
+  });
+}

test/serverpod_offline_sync_test_server/test/integration/merge/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);
+      });
+    });
+  });
+}