Never publish this fork

Halkyon Learn uses this fork as a git dependency, from
https://code.halkyon.app/project-halkyon/serverpod-offline-sync. The package name
serverpod_offline_sync belongs to its author on pub.dev, so `dart pub publish`
should refuse outright rather than depend on pub.dev refusing the upload.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01MMm7usAo6q9ZZRoPkWjoxi

.github/copilot-instructions.md

@@ -22,6 +22,16 @@
 - Remove unnecessary comments from production code.
 - Ensure linting passes before considering the task finished.
 
+## Performance Changes
+
+- Do not infer application bulk-write intent by inspecting or comparing domain
+  row values. Applications express uniform bulk edits explicitly through
+  `updateWhere`; the engine must not rediscover that intent from equal values.
+- Engine-owned planning may drive batching, including projected explicit-null
+  column sets, query predicates, schema identities, and merge operation boundaries.
+  Preserve domain inspection required for CRDT correctness and dependency traversal.
+- Compare proposed gains against the best existing API for the operation.
+
 ## Pubspec Dependency Management
 
 - Do not add any new dependencies, unless explicitly asked.

.github/workflows/ci.yaml

@@ -98,8 +98,23 @@ jobs:
       #     file-path: coverage-results.txt
       #     comment-tag: coverage-results
 
+  browser_tests:
+    name: dart test (Chrome)
+    runs-on: ubuntu-latest
+    timeout-minutes: 10
+    steps:
+      - uses: actions/checkout@v4
+      - uses: ./.github/actions/setup
+
+      - name: Prepare SQLite web assets
+        run: bash scripts/setup_sqlite_web_assets.sh
+
+      - name: Run browser regressions
+        working-directory: test/serverpod_offline_sync_test_server
+        run: dart test test/browser --platform chrome --concurrency=1
+
   simulation:
-    name: deterministic simulation (${{ matrix.topology }})
+    name: deterministic simulation (${{ matrix.topology }}, ${{ matrix.delivery }})
     runs-on: ubuntu-latest
     timeout-minutes: 60
     # Deliberately not on pull requests. The suite seeds from the clock, so it
@@ -115,6 +130,7 @@ jobs:
         # Keep the topology files below and shared ownership controls exhaustive
         # when adding a new test file tagged dst.
         topology: [convergence, cross_space]
+        delivery: [full, delta]
     steps:
       - uses: actions/checkout@v4
       - uses: ./.github/actions/setup
@@ -135,6 +151,7 @@ jobs:
           DST_ROUNDS: 200
           DST_PROFILE: mixed
           DST_GRAPH_WIDTH: 2
+          DST_DELIVERY: ${{ matrix.delivery }}
 
   benchmarks:
     name: run benchmarks

.gitignore

@@ -22,3 +22,7 @@ coverage-results.txt
 
 # Deterministic simulation hunt results
 .dst-soak/
+
+# Downloaded by scripts/setup_sqlite_web_assets.sh.
+test/serverpod_offline_sync_test_server/test/browser/sqlite3.wasm
+test/serverpod_offline_sync_test_server/test/browser/db_worker.js

CHANGELOG.md

@@ -1,3 +1,26 @@
+## 0.0.10
+
+- fix: Prevents concurrent edits from being skipped during sync.
+- fix: Prevents database operations on the web from hanging inside transactions.
+- fix: Respects space access changes made in the same transaction.
+- fix: Prevents sync from stalling after reconnects, including when the device
+  clock moves backward.
+- fix: Prevents previously synced data from being repeatedly resent after reconnects.
+- fix: Rejects checkpoints attributed to the wrong node with `ArgumentError`.
+- fix: Keeps unique-conflict results consistent across devices after updates
+  and upserts.
+- fix: Keeps foreign key defaults consistent across devices when an upsert
+  supplies null, including after a previously referenced record is restored.
+- perf: Improves performance when applying synced changes.
+- perf: Improves conflict resolution performance.
+- chore: Requires Serverpod `>=4.0.4 <4.1.0`.
+
+## 0.0.9
+
+- fix: Refreshes node clocks across sync wrappers.
+- fix: Repairs foreign keys on columns that also carry a unique claim.
+- chore: Release the Serverpod dependency version to `^4.0.0`.
+
 ## 0.0.8
 
 - fix: BREAKING. Reject unique-text values ending in `__conflict__<UUID>`,

README.md

@@ -101,17 +101,17 @@ Add the packages to the server and client `pubspec.yaml` files.
 ```yaml
 # your_project_client/pubspec.yaml
 dependencies:
-  serverpod_offline_sync_client: 0.0.8
+  serverpod_offline_sync_client: 0.0.10
 ```
 
 ```yaml
 # your_project_server/pubspec.yaml
 dependencies:
-  serverpod_offline_sync_server: 0.0.8
+  serverpod_offline_sync_server: 0.0.10
 ```
 
 > [!NOTE]
-> Version `0.0.8` requires Serverpod `4.0.0`.
+> Version `0.0.10` requires Serverpod `>=4.0.4 <4.1.0`.
 
 After adding the dependencies, list the `serverpod_offline_sync` module on the
 `generator.yaml` file and enable the experimental `database: sync` option:

analysis_options.yaml

@@ -107,7 +107,6 @@ linter:
     - use_colored_box
     - use_decorated_box
     - use_enums
-    - use_if_null_to_convert_nulls_to_bools
     - use_is_even_rather_than_modulo
     - use_named_constants
     - use_raw_strings

benchmark/pubspec.yaml

@@ -11,7 +11,7 @@ dependencies:
   benchmark_harness: ^2.4.0
   intl: ^0.20.2
   path: ^1.9.1
-  serverpod_database: 4.0.0
+  serverpod_database: '>=4.0.4 <4.1.0'
   serverpod_offline_sync:
     path: ../packages/serverpod_offline_sync
   serverpod_offline_sync_server:

benchmark/utils/tables.dart

@@ -19,7 +19,6 @@ List<Table> get benchmarkSyncTables => [
   FkChainSetNullSetNullChild.t,
   Organization.t,
   Person.t,
-  RequiredSetNullChild.t,
   RestrictChild.t,
   Town.t,
   Types.t,

docs/foreign-key-invariants.md

@@ -21,12 +21,16 @@ not new user field updates emitted opportunistically during merge.
 - Supported foreign keys are single-column child references to single-column
   parent references. Schemas with composite foreign keys must fail recorder
   initialization instead of silently bypassing projection.
-- Merge input is causally complete: the sync protocol delivers each batch as a
-  full causal snapshot of the sender, so every merged FK attempt's parent facts
-  precede or accompany the child fact. Input that violates this precondition
-  (for example, a non-nullable FK attempt whose parent fact is absent and whose
-  default target is missing) is outside the merge contract and fails the merge
-  transaction atomically against the database foreign key constraints.
+- Serverpod-generated schemas guarantee that `SET NULL` uses nullable columns
+  and `SET DEFAULT` uses columns with declared database defaults.
+- Merge input is causally complete relative to the receiver: every merged FK
+  attempt's parent facts must already be committed at the receiver or accompany
+  the child fact in the same complete batch. A checkpoint-filtered delta can
+  omit acknowledged facts; it need not repeat the sender's full history. Input
+  that violates this precondition (for example, a non-nullable FK attempt whose
+  parent fact is absent and whose default target is missing) is outside the
+  merge contract and fails the merge transaction atomically against the database
+  foreign key constraints.
 - Local `ON DELETE` actions are authored CRDT facts. `RESTRICT` and `NO ACTION`
   reject the local delete while visible children exist; `SET NULL` and
   `SET DEFAULT` update child FK fields and their ordinary `CrdtDataField` HLCs;
@@ -49,15 +53,15 @@ soft-deleted instead of removed:
 - `ON DELETE RESTRICT` and `NO ACTION`: the delete fails if any visible child
   still references the parent. No parent tombstone is recorded for the failed
   operation.
-- `ON DELETE SET NULL`: visible child FK columns are updated to `null` when the
-  FK is nullable, and those child FK fields receive ordinary CRDT field updates.
+- `ON DELETE SET NULL`: visible child FK columns are updated to `null`, and
+  those child FK fields receive ordinary CRDT field updates.
 - `ON DELETE SET DEFAULT`: visible child FK columns are updated to the column
   default when the default is legal, and those child FK fields receive ordinary
-  CRDT field updates. A nullable null default is legal; a non-null default must
-  remain visible in the same space after the delete. A default in the same
-  delete batch is unavailable. If a visible child needs this repair, an invalid
-  default rejects the transaction, including any earlier repairs or tombstones
-  in that transaction. An unused invalid default does not itself block a delete.
+  CRDT field updates. A concrete UUID default must remain visible in the same
+  space after the delete. A default in the same delete batch is unavailable.
+  If a visible child needs this repair, an invalid default rejects the
+  transaction, including any earlier repairs or tombstones in that transaction.
+  An unused invalid default does not itself block a delete.
 - `ON DELETE CASCADE`: visible cascade descendants receive synced
   user-delete tombstones. Local cascade descendants are not hidden only as
   `foreignKeyCascade` projection rows.
@@ -78,9 +82,8 @@ a column.
 
 For each child FK whose attempted value points to a hidden or missing parent:
 
-- `ON DELETE SET NULL`: the visible FK value is `null` when the FK is nullable.
-  If the FK is non-nullable, this action cannot repair the row and must fall
-  through to the deterministic blocking policy.
+- `ON DELETE SET NULL`: the visible FK value is `null`. The schema guarantees
+  that the FK is nullable.
 - `ON DELETE SET DEFAULT`: the visible FK value is the column default only if
   that default points to a visible parent or is a legal nullable default. If the
   default target is hidden or missing, this action cannot repair the row and
@@ -90,7 +93,7 @@ For each child FK whose attempted value points to a hidden or missing parent:
   projection.
 - `ON DELETE CASCADE`: the child can be hidden only if the whole cascade closure
   is valid. If any descendant blocks the cascade through `RESTRICT`, `NO ACTION`,
-  invalid `SET NULL`, or invalid `SET DEFAULT`, the ancestor delete loses and
+  or an unavailable `SET DEFAULT` target, the ancestor delete loses and
   the cascade is not partially applied.
 
 This makes mixed chains atomic for visibility. For example, if deleting `A`

docs/outbound-collection-consistency.md

@@ -1,104 +0,0 @@
-# Outbound collection consistency (follow-up)
-
-## Summary
-
-Outbound change collection reads CRDT metadata and the domain data it describes
-in **separate, non-atomic queries**. Under concurrent domain writes — real on
-the server, where co-members of a shared space write while a sync runs — this
-race can make an otherwise-healthy stream sync **fail and record a durable
-integrity violation** for a condition that the very next round would resolve on
-its own. Failing the session for a transient, self-correcting race is the design
-weakness to fix; no data is ever applied incorrectly, so this is a
-robustness/liveness problem, not a corruption one.
-
-This is **not yet implemented** — it is tracked design debt. The single-pass
-multi-space collection (`collectPendingChanges`) did not introduce the race
-but widened its window, which is what surfaced it.
-
-## The mechanism
-
-Collection is *snapshot-then-fetch*, and lock-free by design (the per-space
-`transactionForUser` lock guards the inbound merge, not outbound collection):
-
-1. `_streamInserts` / `_streamUpdates` / `_streamDeletes` read the CRDT metadata
-   (`CrdtDataRow` / `CrdtDataField` / tombstones) as a query result — a snapshot
-   at some point `t0`.
-2. The domain row/column values are then read in **later, separate** queries
-   (`_fetchDomainRow`, `_fetchOwnedColumnValue`, `_readDomainRowOwner`).
-
-Between (1) and (2), and across the three independently-snapshotted queries, the
-domain can change underneath the collection.
-
-## Failure modes
-
-1. **Concurrent delete → false-positive `missingDomainRow`.** A row whose CRDT
-   insert metadata was snapshotted, but whose domain row is deleted before the
-   domain read, makes `_fetchDomainRow` miss. Today that records a durable
-   `missingDomainRow` violation and **fails the session** — even though the next
-   round would simply collect the tombstone and converge.
-
-2. **FK target inserted after the insert snapshot → dangling reference.** The
-   insert snapshot is taken at `t0`, but FK *column values* are read at fetch
-   time (later). If a referencing row's FK points to row `B` that was inserted
-   **after** `t0`, then `B` is not in the collected insert batch, yet the
-   reference to it is shipped. The same applies to an FK **update** collected in
-   the updates pass whose target was inserted after the inserts pass: the edge is
-   sent without the node. The peer then either churns through FK projection
-   (`SET NULL` / `SET DEFAULT` / `RESTRICT` with the durable `attemptedValue`)
-   until a later round delivers `B`, or trips an integrity violation and fails.
-   Either way, a benign ordering skew degrades the stream.
-
-3. **Value ahead of its HLC (self-correcting, listed for completeness).** If a
-   column is bumped from HLC `h` to `h2` between the metadata read and the domain
-   read, the batch ships `(value@h2, HLC=h)`. This converges: our own
-   `CrdtDataField` is now at `h2`, our send checkpoint only advanced to `h`, so
-   the next round re-ships `(value@h2, HLC=h2)` and every replica ratifies it.
-   No permanent divergence — but it shares the same root cause as (1) and (2).
-
-## Why "fail the session" is the wrong response
-
-All three are **transient and self-correcting**: the next collection round picks
-up the tombstone, the late insert, or the higher HLC, and replicas converge.
-Tearing down the stream — and persisting an integrity-violation record that reads
-like corruption — for a race that resolves itself next round converts an
-ordering skew into an outage plus a misleading audit trail. A durable
-`offline_sync_integrity_violations` entry should mean a *genuine* invariant breach
-(an ownership collision, real corruption), not "two reads happened to straddle a
-concurrent commit."
-
-## Directions
-
-- **Read a consistent snapshot (primary).** Run the whole collection — the three
-  metadata queries *and* the domain reads — inside one read-only,
-  snapshot-consistent transaction (repeatable-read). On Postgres this is MVCC:
-  **a read snapshot, not a write lock**, so it adds no cross-space write
-  contention (and is therefore the right answer to "do we have to lock all
-  spaces?" — no, we take a snapshot, not a lock). A single snapshot closes the
-  race outright:
-  - The delete in (1) is invisible if it commits after the snapshot, and fully
-    visible (metadata + domain) if before — never half-seen.
-  - It makes each batch **referentially closed by construction**: the domain's
-    own FK constraint guarantees that wherever `A.fk = B` is visible in the
-    snapshot, `B` is too, so `B` is collected alongside `A`. (2) cannot happen.
-
-- **Tolerate, don't fail (fallback / defense in depth).** Where a consistent
-  snapshot is impractical, distinguish a *transient* missing row / missing
-  referent (present in CRDT metadata, not yet consistently readable, or arriving
-  next round) from a *true* violation, and **defer the racy change to the next
-  round** instead of failing the session and recording a durable violation. The
-  peer's FK projection already tolerates a transiently-missing referent — the
-  collector should match that posture rather than failing on it.
-
-- **Shrink the window.** Independent of the above, replace row-by-row domain
-  reads with set-based queries (`WHERE id IN (…)`) per table, narrowing the gap
-  between the metadata snapshot and the domain reads. Cheap, and reduces exposure
-  even before a full snapshot lands.
-
-## Priority
-
-Correctness is preserved today (convergence + fail-safe: no wrong data is ever
-applied), so this does not block current use. But because it can fail a healthy
-long-lived continuous sync under concurrent shared-space writes — the exact
-workload shared spaces enable — it should be addressed before such deployments
-are relied upon. The consistent-snapshot approach is the smallest change that
-removes all three failure modes at once.

docs/projection-model.md

@@ -9,7 +9,6 @@ Related documents:
 
 - [Foreign-key projection invariants](foreign-key-invariants.md)
 - [Unique-constraint invariants](unique-constraint-invariants.md)
-- [Outbound collection consistency](outbound-collection-consistency.md)
 
 ## Decisions
 
@@ -177,10 +176,24 @@ value. Reinsertion follows the full-row reauthoring rule above.
 
 ## Outbound and rebuild behavior
 
+Collection captures pending insert, update, and delete metadata, attempted
+values, domain payloads, and ownership checks in one repeatable-read
+transaction. It releases the transaction before emitting the captured changes.
+Concurrent commits outside that snapshot wait for a later pass. Consume the
+collector outside application transactions, with no ambient session transaction.
+
+The complete pass is buffered, so memory use grows with its payloads and the
+first change is emitted only after capture completes. SQLite writers wait
+during capture. Bounded wire chunks do not bound capture memory.
+
 Outbound collection substitutes `CrdtDataAttemptedValue.value` whenever it
 exists; otherwise it reads the domain value. It never exports an FK candidate,
 unique loser value, or temporary parking value as an authored update.
 
+Snapshot collection cannot recover changes already skipped by an older
+collector's checkpoint. Recovery requires an explicit full-history collection
+using `checkpointsBySpaceUuid: {space: []}`, merged into the affected peer.
+
 An empty replica merging a complete space export must reproduce normalized
 domain rows, hidden state, authored values/HLCs, and derived projection. A
 rebuild may rewrite `projectionReason`, but it must preserve

docs/shared-spaces.md

@@ -52,15 +52,16 @@ membership relation and a sync protocol that iterates spaces.
 2. **Smallest protocol delta.** The exchanged frames are today's, plus a
    per-cycle space-set exchange and a space tag on each frame. `collectNextBatch`
    demultiplexes a cycle's combined batch, `collectPendingChanges` collects every
-   active space in one pass, and the per-space merge plus `recordSyncCheckpoint`
-   run once per inbound space group.
+   active space in one pass, and `mergeInboundBatch` applies each inbound space
+   group while atomically persisting received progress for every author.
 3. **Chunked streaming.** Outbound changes are collected in a single pass over
    the active spaces and streamed in `syncBatchSize` chunks, so the wire payload
    stays bounded regardless of space count. The collection step itself
    materializes a cycle's pending rows across the active spaces (one
    `spaceId IN (…)` query per change kind), so peak memory tracks a cycle's
    pending rows rather than being held strictly one space at a time. See
-   `outbound-collection-consistency.md`.
+   [outbound and rebuild behavior](projection-model.md#outbound-and-rebuild-behavior)
+   for snapshot guarantees and capture costs.
 4. **Membership cannot be forged.** The server computes its own space set from
    `offline_sync_space_members` (plus the implicit personal space) and never widens it
    from anything the client sends. A space the user is not a member of is never
@@ -128,7 +129,7 @@ They do not map one-to-one and neither replaces the other. One member may have
 many nodes (one per replica/database install). The **server is a node** in every
 space it syncs but is not a member row. A freshly invited member has **zero
 space-node rows** until their first device syncs. Nodes and space-node rows are
-created implicitly by sync (`getOrCreate`, `recordSyncCheckpoint`); membership
+created implicitly by space initialization and inbound merge; membership
 rows are managed through `session.offlineSync.spaces` from application endpoints.
 Removing the `nodes` list would break checkpointing and causal filtering — it is
 the chain topology, not the access list.
@@ -316,10 +317,11 @@ The shared establishment exchanges the space set so a follower never acts before
 it knows the agreed set. Both cadences handshake initial and newly added spaces
 through the loop's own `SinceHlc` exchange.
 
-`collectPendingChanges`, `mergeInboundBatch`, `recordSyncCheckpoint`, and the
+`collectPendingChanges`, `mergeInboundBatch`, and the
 ownership-violation internals (`_streamInserts`/`Updates`/`Deletes`,
 `_fetchDomainRow`, the durable-violation path) are space-parameterized and merge
-each inbound space group in its own `transactionForUser`/lock; any space's merge
+each inbound space group in its own `transactionForUser`/lock, including its
+per-author checkpoints; any space's merge
 failure (for example an ownership collision or unauthorized role write) records
 the durable violation and fails the session, preserving today's fail-fast
 semantics.
@@ -639,11 +641,3 @@ role enforcement above.
    Purge-vs-keep policy, and whether revoked or demoted unsynced changes are
    surfaced, dropped, or kept pending locally, need their own design. (Carried
    from `row-ownership.md` open question 3.)
-2. **Outbound collection consistency.** Collection reads CRDT metadata and the
-   domain data separately and lock-free, so a concurrent delete — or an FK
-   reference to a row inserted after the insert snapshot — can fail a healthy
-   stream sync and record a durable violation for a race that the next round
-   would resolve. The single-pass multi-space collection widened this window.
-   Failing the session for a transient, self-correcting race is the weakness to
-   fix; the leading direction is a read-only snapshot (MVCC, not a lock). See
-   `outbound-collection-consistency.md`.

example/offline_sync_demo/lib/src/snapshot.dart

@@ -319,7 +319,7 @@ IconData iconForTable(String table, {required bool hidden}) {
     'town' => Icons.map,
     'organization' => Icons.apartment,
     'company' => Icons.business,
-    'restrict_child' || 'required_set_null_child' => Icons.link,
+    'restrict_child' => Icons.link,
     'unique' || 'unique_uuid' || 'unique_composite' => Icons.key,
     'types' => Icons.data_object,
     _ when table.startsWith('fk_chain') => Icons.account_tree,

example/offline_sync_demo/lib/src/table_ops.dart

@@ -160,14 +160,6 @@ final Map<String, TableOps> demoTableOps = {
       create: (id, label, foreignKeys) =>
           RestrictChild(id: id, name: label, parentId: foreignKeys['parentId']),
     ),
-    _ops<RequiredSetNullChild>(
-      canCreateRoot: false,
-      create: (id, label, foreignKeys) {
-        final parentId = foreignKeys['parentId'];
-        if (parentId == null) return null;
-        return RequiredSetNullChild(id: id, name: label, parentId: parentId);
-      },
-    ),
     _ops<Unique>(
       create: (id, label, foreignKeys) => Unique(id: id, name: label),
     ),

example/offline_sync_demo/pubspec.yaml

@@ -16,7 +16,7 @@ dependencies:
   http: ^1.5.0
   path: ^1.9.1
   path_provider: ^2.1.5
-  serverpod_database: 4.0.0
+  serverpod_database: '>=4.0.4 <4.1.0'
   serverpod_offline_sync_client:
     path: ../../packages/serverpod_offline_sync_client
   serverpod_offline_sync_test_client:

legacy/drift_offline_sync/pubspec.yaml

@@ -1,6 +1,6 @@
 name: drift_offline_sync
 description: A Drift plugin to build offline-first applications with CRDT support.
-version: 4.0.0-beta.1
+version: 0.0.1
 repository: https://github.com/marcelomendoncasoares/drift_offline_sync
 
 environment:

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

@@ -133,7 +133,11 @@ class OfflineSyncDatabase implements Database {
     );
   }
 
-  /// Records the latest acknowledged sync checkpoint for [otherNodeId].
+  /// Records acknowledged per-author progress for [otherNodeId].
+  ///
+  /// [syncedHlc] must carry [otherNodeId] as its author. Throws [ArgumentError]
+  /// otherwise. A correctly tagged checkpoint replaces previously stored
+  /// progress tagged with another author; valid progress advances monotonically.
   Future<void> recordSyncCheckpoint(
     UuidValue otherNodeId,
     Hlc syncedHlc, {
@@ -487,9 +491,15 @@ class OfflineSyncDatabase implements Database {
           tx,
           projectionUnchanged: projection.projectionUnchanged,
           domainBeforeUpsert: projection.domain,
-          upsertRows: updateColumns == null && projection.domain.isNotEmpty
-              ? prepared.rows
-              : const [],
+          upsertInputs: updateColumns == null && projection.domain.isNotEmpty
+              ? _recorder.upsertInputsByRow(
+                  suppliedRows: prepared.rows,
+                  databaseRows: values,
+                  updatedRows: updatedRows,
+                  conflictColumns: conflictColumns,
+                )
+              : const {},
+          upsertConflictColumns: conflictColumns.map((c) => c.columnName).toSet(),
         );
         if (noReturn) return <T>[];
         _stripStampedRows(result, prepared);
@@ -670,6 +680,7 @@ class OfflineSyncDatabase implements Database {
           columns,
           tx,
           projectionUnchanged: plannedUpdates.projectionUnchanged,
+          projectionRows: plannedUpdates.projectionRows,
         );
         return noReturn ? <T>[] : updatedRows;
       },
@@ -707,6 +718,7 @@ class OfflineSyncDatabase implements Database {
           columns,
           tx,
           projectionUnchanged: plannedUpdates.projectionUnchanged,
+          projectionRows: plannedUpdates.projectionRows,
         );
         return updatedRow;
       },
@@ -797,28 +809,42 @@ class OfflineSyncDatabase implements Database {
     return _runTrackedWrite(
       transaction,
       (tx) async {
-        final result = await _delegate.updateWhere<T>(
-          columnValues: columnValues,
-          where: (await _whereVisibleWithTombstone<T>(
-            where,
-            null,
-            tx,
-            membershipWide: false,
-          ))!,
+        final scopedWhere = (await _whereVisibleWithTombstone<T>(
+          where,
+          null,
+          tx,
+          membershipWide: false,
+        ))!;
+        final columns = columnValues.map((e) => e.column).toList();
+        final plan = await _recorder.planLocalUpdateWhere<T>(
+          columns,
+          scopedWhere,
+          tx,
           limit: limit,
           offset: offset,
           orderBy: orderBy,
           orderByList: orderByList,
+        );
+        final selectedWhere = plan == null
+            ? scopedWhere
+            : serializationManager.getTableForType(T)!.id.inSet(plan.rowIds);
+        final result = await _delegate.updateWhere<T>(
+          columnValues: columnValues,
+          where: selectedWhere,
+          limit: plan == null ? limit : null,
+          offset: plan == null ? offset : null,
+          orderBy: orderBy,
+          orderByList: orderByList,
           transaction: tx,
         );
 
-        final columns = columnValues.map((e) => e.column).toList();
         _recorder.validateAuthoredRows(result, columns);
         await _recorder.afterUpdate(
           result,
           columns,
           tx,
           authoredColumnValues: true,
+          projectionRows: plan?.projectionRows ?? const {},
         );
         if (noReturn) return <T>[];
         result.forEach(_stripSpaceId);
@@ -1039,15 +1065,18 @@ class OfflineSyncDatabase implements Database {
       OfflineSyncSpace.db.find(
         _delegate.session,
         where: (t) => t.uuidSpaceId.equals(userId),
+        transaction: transaction,
       ),
       OfflineSyncSpaceMember.db
           .find(
             _delegate.session,
             where: (t) => t.userUuid.equals(userId),
             include: OfflineSyncSpaceMember.include(space: OfflineSyncSpace.include()),
+            transaction: transaction,
           )
           .then((memberships) => [for (final member in memberships) member.space!]),
     ]);
+
     return {
       for (final spaces in spaceGroups)
         for (final space in spaces) space.id!,

packages/serverpod_offline_sync/lib/src/database/merge.dart

@@ -1,68 +1,5 @@
 part of 'recorder.dart';
 
-/// Field metadata for merged inserts, waiting to be written as one batch.
-///
-/// Every query behind [CrdtForeignKeyProjector.recordInsertAttempts] already
-/// takes a set of row ids, so a batch of inserts into one table costs the same
-/// as one insert. Rows are grouped by node as well as table because the merge
-/// cache stores each field with the node that authored it.
-class _PendingInsertAttempts {
-  final Map<(String, int), ({Set<UuidValue> rowIds, CrdtNode node})> _byGroup = {};
-  final Map<String, ProjectionAttemptsByField> _attemptsByTable = {};
-  final Set<MergeRowKey> _rows = {};
-
-  bool get isEmpty => _byGroup.isEmpty;
-
-  /// Whether metadata for [rowKey] is still unwritten.
-  bool holds(MergeRowKey rowKey) => _rows.contains(rowKey);
-
-  void add({
-    required String tableName,
-    required UuidValue rowId,
-    required CrdtNode node,
-    required ProjectionAttemptsByField attempts,
-  }) {
-    _byGroup
-        .putIfAbsent(
-          (tableName, node.id!),
-          () => (rowIds: <UuidValue>{}, node: node),
-        )
-        .rowIds
-        .add(rowId);
-    _rows.add((tableName, rowId));
-    if (attempts.isEmpty) return;
-    _attemptsByTable
-        .putIfAbsent(tableName, () => <MergeFieldKey, ProjectionAttempt>{})
-        .addAll(attempts);
-  }
-
-  /// Empties the collector and returns what it held.
-  List<
-    ({
-      String tableName,
-      Set<UuidValue> rowIds,
-      CrdtNode node,
-      ProjectionAttemptsByField attempts,
-    })
-  >
-  take() {
-    final groups = [
-      for (final MapEntry(key: key, value: group) in _byGroup.entries)
-        (
-          tableName: key.$1,
-          rowIds: group.rowIds,
-          node: group.node,
-          attempts:
-              _attemptsByTable[key.$1] ?? const <MergeFieldKey, ProjectionAttempt>{},
-        ),
-    ];
-    _byGroup.clear();
-    _attemptsByTable.clear();
-    _rows.clear();
-    return groups;
-  }
-}
-
 /// Adds merge-specific behavior to [CrdtMutationRecorder].
 extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
   /// Merges remote CRDT changes into the current database.
@@ -142,62 +79,61 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
             transaction,
           );
 
-    // Field metadata for merged inserts is recorded per table and node rather
-    // than per row: every query behind it already takes a set of row ids.
-    final pendingAttempts = _PendingInsertAttempts();
+    final writes = _MergeWriteBatch(
+      session: _session,
+      projector: _foreignKeyProjector,
+      fields: context.fields,
+      transaction: transaction,
+    );
     final deferredDeletes = <CrdtMergeDelete>[];
 
-    for (final operation in operations) {
-      if (!_context.isCrdtTrackedTableName(operation.tableName)) {
-        continue;
-      }
-      // A later operation on a row resolves its field metadata through the
-      // merge cache, so anything still pending for that row is written first.
-      if (pendingAttempts.holds((operation.tableName, operation.uuidRowId))) {
-        await _flushInsertAttempts(pendingAttempts, context, transaction);
-      }
+    await writes.apply(
+      operations.where(
+        (operation) => _context.isCrdtTrackedTableName(operation.tableName),
+      ),
+      (operation) async {
+        switch (operation) {
+          case final CrdtMergeInsert insert:
+            await _applyMergeInsert(
+              insert,
+              nodesByUuid,
+              context,
+              authoredOverlays,
+              batch.plan,
+              writes,
+              presence,
+              transaction,
+            );
+          case final CrdtMergeUpdate update:
+            await _applyMergeUpdate(
+              update,
+              nodesByUuid,
+              context,
+              authoredOverlays,
+              writes,
+              transaction,
+            );
+          case final CrdtMergeDelete delete:
+            // A higher delete/restore generation can have an older HLC than a
+            // concurrent reinsertion. Bootstrap must retain it until this batch
+            // has inserted the row it belongs to.
+            if (!context.rows.containsKey((delete.tableName, delete.uuidRowId))) {
+              deferredDeletes.add(delete);
+              return;
+            }
+            await _applyMergeDelete(
+              delete,
+              nodesByUuid,
+              context,
+              transaction,
+            );
+            // A delete can hide the row, and a resurrecting one can bring it
+            // back; either way what was known about it no longer holds.
+            presence?.forget((delete.tableName, delete.uuidRowId));
+        }
+      },
+    );
 
-      switch (operation) {
-        case final CrdtMergeInsert insert:
-          await _applyMergeInsert(
-            insert,
-            nodesByUuid,
-            context,
-            authoredOverlays,
-            batch.plan,
-            pendingAttempts,
-            presence,
-            transaction,
-          );
-        case final CrdtMergeUpdate update:
-          await _applyMergeUpdate(
-            update,
-            nodesByUuid,
-            context,
-            authoredOverlays,
-            transaction,
-          );
-        case final CrdtMergeDelete delete:
-          // A higher delete/restore generation can have an older HLC than a
-          // concurrent reinsertion. Bootstrap must retain it until this batch
-          // has inserted the row it belongs to.
-          if (!context.rows.containsKey((delete.tableName, delete.uuidRowId))) {
-            deferredDeletes.add(delete);
-            continue;
-          }
-          await _applyMergeDelete(
-            delete,
-            nodesByUuid,
-            context,
-            transaction,
-          );
-          // A delete can hide the row, and a resurrecting one can bring it
-          // back; either way what was known about it no longer holds.
-          presence?.forget((delete.tableName, delete.uuidRowId));
-      }
-    }
-
-    await _flushInsertAttempts(pendingAttempts, context, transaction);
     for (final delete in deferredDeletes) {
       await _applyMergeDelete(delete, nodesByUuid, context, transaction);
     }
@@ -259,8 +195,11 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
         in maxIncomingHlcByNode.entries) {
       final remoteNode = remoteNodes.spaceNodesByUuid[nodeId];
       if (remoteNode == null) continue;
+      final previous = remoteNode.lastReceivedHlc;
       final updatedSpaceNode = remoteNode.copyWith(
-        lastReceivedHlc: incomingHlc.maxBetween(remoteNode.lastReceivedHlc),
+        lastReceivedHlc: incomingHlc.maxBetween(
+          previous?.nodeId == nodeId ? previous : null,
+        ),
       );
       if (updatedSpaceNode.lastReceivedHlc == remoteNode.lastReceivedHlc) {
         continue;
@@ -471,7 +410,7 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
     MergeContext context,
     Map<MergeFieldKey, Object?> authoredOverlays,
     ProjectionPlan? batchPlan,
-    _PendingInsertAttempts pendingAttempts,
+    _MergeWriteBatch writes,
     CrdtForeignKeyPresenceCache? presence,
     Transaction transaction,
   ) async {
@@ -525,7 +464,7 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
         // same-space recovery), so same-merge field updates resolve ownership
         // from the cache instead of re-reading the domain row.
         context.domainOwners[rowKey] = (exists: true, spaceId: mergingSpaceId);
-        pendingAttempts.add(
+        writes.recordInsertAttempts(
           tableName: insert.tableName,
           rowId: insert.uuidRowId,
           node: remoteNode,
@@ -616,36 +555,12 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
     return (domain: plannedDomain, attempts: attempts);
   }
 
-  /// Writes the field metadata collected for merged inserts, one call per
-  /// table and node.
-  ///
-  /// Grouped by node because the merge cache pairs every field it stores with
-  /// the node that authored it, which is how the cached field resolves its own
-  /// clock.
-  Future<void> _flushInsertAttempts(
-    _PendingInsertAttempts pending,
-    MergeContext context,
-    Transaction transaction,
-  ) async {
-    if (pending.isEmpty) return;
-
-    final groups = pending.take();
-    for (final group in groups) {
-      await _foreignKeyProjector.recordInsertAttempts(
-        group.tableName,
-        group.rowIds,
-        transaction,
-        group.attempts,
-        mergeCache: (fields: context.fields, node: group.node),
-      );
-    }
-  }
-
   Future<void> _applyMergeUpdate(
     CrdtMergeUpdate update,
     Map<UuidValue, CrdtNode> remoteNodes,
     MergeContext context,
     Map<MergeFieldKey, Object?> authoredOverlays,
+    _MergeWriteBatch writes,
     Transaction transaction,
   ) async {
     final rowKey = (update.tableName, update.uuidRowId);
@@ -673,6 +588,7 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
       remoteNode: remoteNode,
       incomingHlc: update.hlc,
       fields: context.fields,
+      writes: writes,
       transaction: transaction,
     );
 
@@ -780,8 +696,9 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
     CrdtNode remoteNode,
     Hlc incomingHlc,
     CrdtDataField? currentField,
-    Transaction transaction,
-  ) async {
+    Transaction transaction, {
+    _MergeWriteBatch? writes,
+  }) async {
     // Projection can create field metadata for a row this merge context loaded
     // before the row existed, so an unseen field may still be persisted. Adopt
     // it instead of inserting a second row onto the row/column unique index.
@@ -794,6 +711,8 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
         );
 
     if (existingField == null) {
+      // Dependent metadata needs the field identity, so new fields cannot
+      // wait in the batch of existing field-clock updates.
       final insertedField = await CrdtDataField.db.insertRow(
         _session,
         CrdtDataField(
@@ -822,12 +741,16 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
       hlcCounter: incomingHlc.counter,
     );
 
-    await CrdtDataField.db.update(
-      _session,
-      [updatedField],
-      columns: (t) => [t.nodeId, t.hlcDatetime, t.hlcCounter],
-      transaction: transaction,
-    );
+    if (writes != null) {
+      writes.recordFieldUpdate(updatedField);
+    } else {
+      await CrdtDataField.db.update(
+        _session,
+        [updatedField],
+        columns: (t) => [t.nodeId, t.hlcDatetime, t.hlcCounter],
+        transaction: transaction,
+      );
+    }
     return updatedField;
   }
 
@@ -1018,6 +941,7 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
     required Map<MergeFieldKey, CrdtDataField> fields,
     required Transaction transaction,
     CrdtSchemaColumn? schemaColumn,
+    _MergeWriteBatch? writes,
   }) async {
     final resolvedSchemaColumn =
         schemaColumn ?? _context.schemaColumn(tableName, columnName);
@@ -1035,6 +959,7 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
       incomingHlc,
       currentField,
       transaction,
+      writes: writes,
     );
     return true;
   }

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

@@ -94,7 +94,7 @@ TableRow<UuidValue?> withExplicitInsertNulls(
 
 /// Keeps per-row default semantics when a batch contains projected nulls.
 /// PostgreSQL builds a batch using the first row's table, so rows with different
-/// null overrides must be written separately within the caller's transaction.
+/// null overrides use separate consecutive batches in the caller's transaction.
 @internal
 Future<List<T>> insertWithExplicitNulls<T extends TableRow>(
   Database database,
@@ -112,15 +112,30 @@ Future<List<T>> insertWithExplicitNulls<T extends TableRow>(
       noReturn: noReturn,
     );
   }
-  return [
-    for (final row in rows)
-      ...(await database.insert<TableRow<UuidValue?>>(
-        [withExplicitInsertNulls(row, explicitNulls[row.id] ?? const {})],
-        transaction: transaction,
-        ignoreConflicts: ignoreConflicts,
-        noReturn: noReturn,
-      )).cast<T>(),
-  ];
+  final inserted = <T>[];
+  var start = 0;
+  while (start < rows.length) {
+    final columns = explicitNulls[rows[start].id] ?? const <String>{};
+    var end = start + 1;
+    while (end < rows.length) {
+      final next = explicitNulls[rows[end].id] ?? const <String>{};
+      if (columns.length != next.length || !columns.containsAll(next)) break;
+      end++;
+    }
+
+    final result = await database.insert<TableRow<UuidValue?>>(
+      [
+        for (var index = start; index < end; index++)
+          withExplicitInsertNulls(rows[index], columns),
+      ],
+      transaction: transaction,
+      ignoreConflicts: ignoreConflicts,
+      noReturn: noReturn,
+    );
+    inserted.addAll(result.cast<T>());
+    start = end;
+  }
+  return inserted;
 }
 
 class _ExplicitNullRow implements TableRow<UuidValue?> {

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

@@ -11,6 +11,8 @@ import 'recorder_context.dart';
 import 'types.dart';
 import 'unique_resolver.dart';
 
+part 'projection_reads.dart';
+
 typedef _ProjectedForeignKeyRow = ({
   MergeRowKey key,
   CrdtDataRow crdtRow,
@@ -538,13 +540,11 @@ class CrdtForeignKeyProjector {
 
       switch (edge.action) {
         case ForeignKeyAction.setNull:
-          if (edge.childNullable) {
-            safeData[edge.childColumn] = null;
-            attempts[fieldKey] = (
-              value: attemptedValue,
-              reason: CrdtProjectionReason.foreignKeySetNull,
-            );
-          }
+          safeData[edge.childColumn] = null;
+          attempts[fieldKey] = (
+            value: attemptedValue,
+            reason: CrdtProjectionReason.foreignKeySetNull,
+          );
         case ForeignKeyAction.setDefault:
           final defaultProjection = await _defaultProjectionValueFromDatabase(
             edge,
@@ -880,10 +880,11 @@ class CrdtForeignKeyProjector {
     Set<String>? seedTables,
     Set<MergeRowKey>? seedRows,
   }) async {
-    final rows = <MergeRowKey, _ProjectedForeignKeyRow>{};
-    final fieldIds = <MergeFieldKey, int>{};
-    final attemptedValues = <MergeFieldKey, CrdtDataAttemptedValue>{};
-    final fieldHlcs = <MergeFieldKey, Hlc>{};
+    final reader = _ProjectionReader(_context, transaction);
+    final rows = reader.rows;
+    final fieldIds = reader.fieldIds;
+    final attemptedValues = reader.attemptedValues;
+    final fieldHlcs = reader.fieldHlcs;
 
     final tablesToLoad = _tablesToLoad(
       seedTables: seedTables,
@@ -897,18 +898,10 @@ class CrdtForeignKeyProjector {
     );
 
     if (seedRows == null) {
-      for (final tableName in tablesToLoad) {
-        await _loadTableRowsInto(
-          tableName: tableName,
-          rowIds: null,
-          columnNames: columnsByTable[tableName] ?? const <String>{},
-          rows: rows,
-          fieldIds: fieldIds,
-          attemptedValues: attemptedValues,
-          fieldHlcs: fieldHlcs,
-          transaction: transaction,
-        );
-      }
+      await reader.loadRows(
+        requestedRows: {for (final table in tablesToLoad) table: null},
+        columnsByTable: columnsByTable,
+      );
     } else {
       await _loadRowClosureInto(
         tablesToLoad: tablesToLoad,
@@ -916,11 +909,7 @@ class CrdtForeignKeyProjector {
         pendingInserts: pendingInserts,
         authoredOverlays: authoredOverlays,
         columnsByTable: columnsByTable,
-        rows: rows,
-        fieldIds: fieldIds,
-        attemptedValues: attemptedValues,
-        fieldHlcs: fieldHlcs,
-        transaction: transaction,
+        reader: reader,
       );
     }
 
@@ -1105,78 +1094,6 @@ class CrdtForeignKeyProjector {
     return childRowsByAttempt;
   }
 
-  /// Loads [rowIds] of [tableName], or every row of it when [rowIds] is null.
-  ///
-  /// Returns the row ids that exist, so a closure pass can expand from them.
-  Future<Set<UuidValue>> _loadTableRowsInto({
-    required String tableName,
-    required Set<UuidValue>? rowIds,
-    required Set<String> columnNames,
-    required Map<MergeRowKey, _ProjectedForeignKeyRow> rows,
-    required Map<MergeFieldKey, int> fieldIds,
-    required Map<MergeFieldKey, CrdtDataAttemptedValue> attemptedValues,
-    required Map<MergeFieldKey, Hlc> fieldHlcs,
-    required Transaction transaction,
-  }) async {
-    if (rowIds != null && rowIds.isEmpty) return const {};
-
-    final (tableId, _) = _context.schema[tableName]!;
-    final userId = _context.hlcManagerFor(transaction).normalizedSpaceId;
-    final crdtRows = await CrdtDataRow.db.find(
-      _context.databaseSession,
-      where: (t) {
-        final inSpace = t.spaceId.equals(userId) & t.tblId.equals(tableId);
-        return rowIds == null ? inSpace : inSpace & t.uuidRowId.inSet(rowIds);
-      },
-      include: CrdtDataRow.include(
-        node: CrdtNode.include(),
-        deleted: CrdtDataDeleted.include(node: CrdtNode.include()),
-      ),
-      orderBy: (t) => t.uuidRowId,
-      transaction: transaction,
-    );
-    if (crdtRows.isEmpty) return const {};
-
-    final loadedIds = {for (final row in crdtRows) row.uuidRowId};
-    final valuesByRowId = columnNames.isEmpty
-        ? <UuidValue, Map<String, Object?>>{}
-        : await _context.readDomainColumnValues(
-            tableName,
-            loadedIds,
-            columnNames.toList(),
-            transaction,
-          );
-
-    for (final row in crdtRows) {
-      final key = (tableName, row.uuidRowId);
-      rows[key] = (
-        key: key,
-        crdtRow: row,
-        values: Map<String, Object?>.from(valuesByRowId[row.uuidRowId] ?? const {}),
-      );
-    }
-
-    if (columnNames.isNotEmpty) {
-      final loadedFields = await _loadFields(
-        tableName: tableName,
-        rowIds: loadedIds,
-        columnNames: columnNames,
-        transaction: transaction,
-      );
-      for (final field in loadedFields) {
-        final fieldKey = (tableName, field.row!.uuidRowId, field.column!.name);
-        fieldIds[fieldKey] = field.id!;
-        fieldHlcs[fieldKey] = field.hlc;
-        final attempted = field.attemptedValue;
-        if (attempted != null) {
-          attemptedValues[fieldKey] = attempted;
-        }
-      }
-    }
-
-    return loadedIds;
-  }
-
   /// Loads the rows the seeds can reach instead of every row of their tables.
   ///
   /// Defaults are loaded as evaluation context. Their implicit dependents join
@@ -1199,12 +1116,13 @@ class CrdtForeignKeyProjector {
     required List<PendingProjectionRow> pendingInserts,
     required Map<MergeFieldKey, Object?> authoredOverlays,
     required Map<String, Set<String>> columnsByTable,
-    required Map<MergeRowKey, _ProjectedForeignKeyRow> rows,
-    required Map<MergeFieldKey, int> fieldIds,
-    required Map<MergeFieldKey, CrdtDataAttemptedValue> attemptedValues,
-    required Map<MergeFieldKey, Hlc> fieldHlcs,
-    required Transaction transaction,
+    required _ProjectionReader reader,
   }) async {
+    final transaction = reader._transaction;
+    final rows = reader.rows;
+    final attemptedValues = reader.attemptedValues;
+    final hasAttemptedClaims = await reader.hasAttemptedClaims(tablesToLoad);
+
     final unwrittenValues = _unwrittenValues(pendingInserts, authoredOverlays);
 
     final requested = <String, Set<UuidValue>>{};
@@ -1245,20 +1163,10 @@ class CrdtForeignKeyProjector {
         final wave = queued;
         queued = <String, Set<UuidValue>>{};
 
-        final loadedNow = <String, Set<UuidValue>>{};
-        for (final MapEntry(key: tableName, value: ids) in wave.entries) {
-          final loaded = await _loadTableRowsInto(
-            tableName: tableName,
-            rowIds: ids,
-            columnNames: columnsByTable[tableName] ?? const <String>{},
-            rows: rows,
-            fieldIds: fieldIds,
-            attemptedValues: attemptedValues,
-            fieldHlcs: fieldHlcs,
-            transaction: transaction,
-          );
-          if (loaded.isNotEmpty) loadedNow[tableName] = loaded;
-        }
+        final loadedNow = await reader.loadRows(
+          requestedRows: wave,
+          columnsByTable: columnsByTable,
+        );
 
         for (final MapEntry(key: tableName, value: ids) in frontier.entries) {
           (loadedNow[tableName] ??= <UuidValue>{}).addAll(ids);
@@ -1268,6 +1176,7 @@ class CrdtForeignKeyProjector {
 
         await _expandRowClosure(
           loadedNow: loadedNow,
+          hasAttemptedClaims: hasAttemptedClaims,
           tablesToLoad: tablesToLoad,
           rows: rows,
           unwrittenValues: unwrittenValues,
@@ -1535,6 +1444,7 @@ class CrdtForeignKeyProjector {
     required Map<MergeRowKey, Map<String, Object?>> unwrittenValues,
     required Map<MergeFieldKey, CrdtDataAttemptedValue> attemptedValues,
     required void Function(String tableName, Iterable<UuidValue> ids) enqueue,
+    required bool hasAttemptedClaims,
     required Transaction transaction,
   }) async {
     // Both the stored and the unwritten value matter: the walk has to reach the
@@ -1553,14 +1463,19 @@ class CrdtForeignKeyProjector {
       }..remove(null);
     }
 
+    final claims = _ProjectionClaimReads(
+      _context,
+      transaction,
+      includeAttemptedClaims: hasAttemptedClaims,
+    );
+
     for (final edge in _foreignKeys.edges) {
-      await _expandToChildren(
+      _expandToChildren(
         edge: edge,
         loadedNow: loadedNow,
         tablesToLoad: tablesToLoad,
         valuesFor: valuesFor,
-        enqueue: enqueue,
-        transaction: transaction,
+        collectClaims: claims.add,
       );
       await _expandToParents(
         edge: edge,
@@ -1574,27 +1489,27 @@ class CrdtForeignKeyProjector {
       );
     }
 
-    await _expandUniqueClaimants(
+    _expandUniqueClaimants(
       loadedNow: loadedNow,
       attemptedValues: attemptedValues,
       valuesFor: valuesFor,
-      enqueue: enqueue,
-      transaction: transaction,
+      collectClaims: claims.add,
     );
+
+    await claims.enqueueMatches(enqueue);
   }
 
   /// Enqueues the children of the parents loaded in this wave.
   ///
   /// Hiding or restoring a parent decides its children's fate, so they join the
   /// walk whether they still name it or only remember it as an attempt.
-  Future<void> _expandToChildren({
+  void _expandToChildren({
     required ForeignKeyEdge edge,
     required Map<String, Set<UuidValue>> loadedNow,
     required Set<String> tablesToLoad,
     required Set<Object?> Function(MergeRowKey, String) valuesFor,
-    required void Function(String tableName, Iterable<UuidValue> ids) enqueue,
-    required Transaction transaction,
-  }) async {
+    required void Function(String, Map<String, Set<Object?>>) collectClaims,
+  }) {
     final parentIds = loadedNow[edge.parentTableName];
     if (parentIds == null || parentIds.isEmpty) return;
     if (!tablesToLoad.contains(edge.childTableName)) return;
@@ -1611,12 +1526,7 @@ class CrdtForeignKeyProjector {
     }
     if (references.isEmpty) return;
 
-    await _enqueueRowsClaiming(
-      tableName: edge.childTableName,
-      valuesByColumn: {edge.childColumn: references},
-      enqueue: enqueue,
-      transaction: transaction,
-    );
+    collectClaims(edge.childTableName, {edge.childColumn: references});
   }
 
   /// Enqueues the parents of the children loaded in this wave.
@@ -1686,13 +1596,12 @@ class CrdtForeignKeyProjector {
   /// own unique index for it. A row that wants a value it does not hold is
   /// already loaded, because holding a projection means holding an attempted
   /// value.
-  Future<void> _expandUniqueClaimants({
+  void _expandUniqueClaimants({
     required Map<String, Set<UuidValue>> loadedNow,
     required Map<MergeFieldKey, CrdtDataAttemptedValue> attemptedValues,
     required Set<Object?> Function(MergeRowKey, String) valuesFor,
-    required void Function(String tableName, Iterable<UuidValue> ids) enqueue,
-    required Transaction transaction,
-  }) async {
+    required void Function(String, Map<String, Set<Object?>>) collectClaims,
+  }) {
     for (final MapEntry(key: tableName, value: rowIds) in loadedNow.entries) {
       for (final uniqueIndex in _uniqueResolver.uniqueIndexesFor(tableName)) {
         final claimedByColumn = <String, Set<Object?>>{};
@@ -1713,46 +1622,11 @@ class CrdtForeignKeyProjector {
           continue;
         }
 
-        await _enqueueRowsClaiming(
-          tableName: tableName,
-          valuesByColumn: claimedByColumn,
-          enqueue: enqueue,
-          transaction: transaction,
-        );
+        collectClaims(tableName, claimedByColumn);
       }
     }
   }
 
-  /// Enqueues every row of [tableName] that holds, or is owed, these values.
-  ///
-  /// A row that was repaired or released away from a value no longer names it
-  /// in the domain column, so the attempted value is the only record left of
-  /// the claim it is waiting to take back.
-  Future<void> _enqueueRowsClaiming({
-    required String tableName,
-    required Map<String, Set<Object?>> valuesByColumn,
-    required void Function(String tableName, Iterable<UuidValue> ids) enqueue,
-    required Transaction transaction,
-  }) async {
-    enqueue(
-      tableName,
-      await _context.findDomainRowIdsWhereColumnsIn(
-        tableName: tableName,
-        valuesByColumn: valuesByColumn,
-        transaction: transaction,
-      ),
-    );
-    enqueue(
-      tableName,
-      await _context.findRowIdsHoldingAttemptedValues(
-        tableName: tableName,
-        columnNames: valuesByColumn.keys.toSet(),
-        values: {for (final values in valuesByColumn.values) ...values},
-        transaction: transaction,
-      ),
-    );
-  }
-
   Future<List<CrdtDataField>> _loadFields({
     required String tableName,
     required Set<UuidValue> rowIds,
@@ -1932,7 +1806,7 @@ class CrdtForeignKeyProjector {
   ) {
     switch (edge.action) {
       case ForeignKeyAction.setNull:
-        return edge.childNullable;
+        return true;
       case ForeignKeyAction.setDefault:
         final defaultValue = edge.defaultValue.toUuidValue();
         if (defaultValue == null) return edge.childNullable;
@@ -2017,7 +1891,7 @@ class CrdtForeignKeyProjector {
             case ForeignKeyAction.noAction:
               return true;
             case ForeignKeyAction.setNull:
-              if (!edge.childNullable) return true;
+              break;
             case ForeignKeyAction.setDefault:
               if (!_defaultProjectionValue(edge, state, finalHidden).valid) {
                 return true;
@@ -2248,10 +2122,8 @@ class CrdtForeignKeyProjector {
             _targetHiddenOrMissing(edge, attemptedValue, state, finalHidden)) {
           switch (edge.action) {
             case ForeignKeyAction.setNull:
-              if (edge.childNullable) {
-                desiredVisibleValue = null;
-                overrideReason = CrdtProjectionReason.foreignKeySetNull;
-              }
+              desiredVisibleValue = null;
+              overrideReason = CrdtProjectionReason.foreignKeySetNull;
             case ForeignKeyAction.setDefault:
               final defaultProjection = _defaultProjectionValue(
                 edge,