fix: Capture outbound changes in a consistent snapshot (#166)

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

@@ -60,7 +60,8 @@ membership relation and a sync protocol that iterates spaces.
    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
@@ -640,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`.

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

@@ -126,8 +126,14 @@ class OfflineSyncEngine {
 
   /// Streams pending changes for every space in [checkpointsBySpaceUuid].
   ///
-  /// Changes are emitted in insert, update, then delete order. Domain row and
+  /// Changes are captured in one consistent transaction, including domain
-  /// column payloads are resolved incrementally as each change is yielded.
+  /// payloads, then emitted in insert, update, and delete order. The transaction
+  /// is released before yielding, so consumer backpressure or consumer writes
+  /// cannot keep the snapshot (or SQLite's write lock) open.
+  ///
+  /// Consume this stream outside application transactions, using a session
+  /// without an ambient transaction. Collection opens its own transaction to
+  /// capture committed data; it does not join a caller's transaction.
   ///
   /// Foreign-key columns with an active projection override are sent with their
   /// durable [CrdtDataAttemptedValue.value], not the visible value stored
@@ -150,10 +156,15 @@ class OfflineSyncEngine {
   }) async* {
     if (checkpointsBySpaceUuid.isEmpty) return;
 
+    // Resolve stable space identities before opening the snapshot. A wrapped
+    // database may initialize its CRDT registry on this first query, which can
+    // itself need a write transaction.
     final spaces = await OfflineSyncSpace.db.find(
       session,
       where: (t) => t.uuidSpaceId.inSet(checkpointsBySpaceUuid.keys.toSet()),
     );
+    if (spaces.isEmpty) return;
+
     final spaceUuidById = {
       for (final space in spaces) space.id!: space.uuidSpaceId,
     };
@@ -163,11 +174,19 @@ class OfflineSyncEngine {
     };
 
     try {
-      await for (final change in _streamPendingChanges(
+      final changes = await session.db.transaction(
+        (transaction) => _streamPendingChanges(
           session,
           spaceUuidById,
           checkpointsBySpaceId,
-      )) {
+          transaction,
+        ).toList(),
+        settings: const TransactionSettings(
+          isolationLevel: IsolationLevel.repeatableRead,
+        ),
+      );
+
+      for (final change in changes) {
         yield change;
       }
     } on PendingOutboundIntegrityViolation catch (violation) {
@@ -548,13 +567,31 @@ class OfflineSyncEngine {
     DatabaseSession session,
     Map<int, UuidValue> spaceUuidById,
     Map<int, List<Hlc>> checkpointsBySpaceId,
+    Transaction transaction,
   ) async* {
-    // Domain ownership is immutable while a collection runs, so read each
+    // Cache ownership within this collection's consistent snapshot.
-    // row's owner at most once across all three streams.
     final ownerCache = DomainRowOwnerCache();
-    yield* _streamInserts(session, spaceUuidById, checkpointsBySpaceId, ownerCache);
+    yield* _streamInserts(
-    yield* _streamUpdates(session, spaceUuidById, checkpointsBySpaceId, ownerCache);
+      session,
-    yield* _streamDeletes(session, spaceUuidById, checkpointsBySpaceId, ownerCache);
+      spaceUuidById,
+      checkpointsBySpaceId,
+      ownerCache,
+      transaction,
+    );
+    yield* _streamUpdates(
+      session,
+      spaceUuidById,
+      checkpointsBySpaceId,
+      ownerCache,
+      transaction,
+    );
+    yield* _streamDeletes(
+      session,
+      spaceUuidById,
+      checkpointsBySpaceId,
+      ownerCache,
+      transaction,
+    );
   }
 
   Stream<CrdtMergeInsert> _streamInserts(
@@ -562,9 +599,11 @@ class OfflineSyncEngine {
     Map<int, UuidValue> spaceUuidById,
     Map<int, List<Hlc>> checkpointsBySpaceId,
     DomainRowOwnerCache ownerCache,
+    Transaction transaction,
   ) async* {
     final rows = await CrdtDataRow.db.find(
       session,
+      transaction: transaction,
       where: (t) => _rowHlcAfterFilter(t, checkpointsBySpaceId),
       include: CrdtDataRow.include(
         tbl: CrdtSchemaTable.include(),
@@ -575,6 +614,7 @@ class OfflineSyncEngine {
     final attemptedValueFieldsByRowId = await _loadAttemptedValueFields(
       session,
       rows,
+      transaction,
     );
 
     for (final row in rows) {
@@ -597,6 +637,7 @@ class OfflineSyncEngine {
         attemptedValueFieldsByRowId[row.id!],
         spaceId,
         ownerCache,
+        transaction,
       );
       if (!domainRow.exists) {
         _throwPendingIntegrityViolation(
@@ -642,9 +683,11 @@ class OfflineSyncEngine {
     Map<int, UuidValue> spaceUuidById,
     Map<int, List<Hlc>> checkpointsBySpaceId,
     DomainRowOwnerCache ownerCache,
+    Transaction transaction,
   ) async* {
     final fields = await CrdtDataField.db.find(
       session,
+      transaction: transaction,
       where: (t) => _fieldHlcAfterFilter(t, checkpointsBySpaceId),
       include: CrdtDataField.include(
         row: CrdtDataRow.include(tbl: CrdtSchemaTable.include()),
@@ -674,6 +717,7 @@ class OfflineSyncEngine {
         field.attemptedValue,
         spaceId,
         ownerCache,
+        transaction,
       );
       if (!columnValue.exists) {
         _throwPendingIntegrityViolation(
@@ -720,9 +764,11 @@ class OfflineSyncEngine {
     Map<int, UuidValue> spaceUuidById,
     Map<int, List<Hlc>> checkpointsBySpaceId,
     DomainRowOwnerCache ownerCache,
+    Transaction transaction,
   ) async* {
     final tombstones = await CrdtDataDeleted.db.find(
       session,
+      transaction: transaction,
       where: (t) => _tombstoneHlcAfterFilter(t, checkpointsBySpaceId),
       include: CrdtDataDeleted.include(
         row: CrdtDataRow.include(tbl: CrdtSchemaTable.include()),
@@ -743,6 +789,7 @@ class OfflineSyncEngine {
         tableName,
         tombstone.row!.uuidRowId,
         ownerCache,
+        transaction,
       );
       if (owner.exists && owner.spaceId != spaceId) {
         _throwPendingIntegrityViolation(
@@ -828,6 +875,7 @@ class OfflineSyncEngine {
     List<CrdtDataField>? attemptedValueFields,
     int spaceId,
     DomainRowOwnerCache ownerCache,
+    Transaction transaction,
   ) async {
     final cols = table.columns
         .map(
@@ -842,9 +890,16 @@ class OfflineSyncEngine {
       'SELECT $cols FROM "$escapedTableName" '
       'WHERE "id" = $encodedRowId AND "spaceId" = $encodedSpaceId '
       'LIMIT 1',
+      transaction: transaction,
     );
     if (result.isEmpty) {
-      final owner = await _readDomainRowOwner(session, tableName, rowId, ownerCache);
+      final owner = await _readDomainRowOwner(
+        session,
+        tableName,
+        rowId,
+        ownerCache,
+        transaction,
+      );
       return (exists: owner.exists, ownerSpaceId: owner.spaceId, row: null);
     }
     ownerCache[(tableName, rowId)] = (exists: true, spaceId: spaceId);
@@ -893,9 +948,16 @@ class OfflineSyncEngine {
     CrdtDataAttemptedValue? attempted,
     int spaceId,
     DomainRowOwnerCache ownerCache,
+    Transaction transaction,
   ) async {
     if (attempted != null) {
-      final owner = await _readDomainRowOwner(session, tableName, rowId, ownerCache);
+      final owner = await _readDomainRowOwner(
+        session,
+        tableName,
+        rowId,
+        ownerCache,
+        transaction,
+      );
       if (!owner.exists || owner.spaceId != spaceId) {
         return (exists: owner.exists, ownerSpaceId: owner.spaceId, value: null);
       }
@@ -917,6 +979,7 @@ class OfflineSyncEngine {
       'FROM "$escapedTableName" '
       'WHERE "id" = $encodedRowId AND "spaceId" = $encodedSpaceId '
       'LIMIT 1',
+      transaction: transaction,
     );
     if (result.isNotEmpty) {
       ownerCache[(tableName, rowId)] = (exists: true, spaceId: spaceId);
@@ -931,7 +994,13 @@ class OfflineSyncEngine {
       );
     }
 
-    final owner = await _readDomainRowOwner(session, tableName, rowId, ownerCache);
+    final owner = await _readDomainRowOwner(
+      session,
+      tableName,
+      rowId,
+      ownerCache,
+      transaction,
+    );
     return (exists: owner.exists, ownerSpaceId: owner.spaceId, value: null);
   }
 
@@ -940,6 +1009,7 @@ class OfflineSyncEngine {
     String tableName,
     UuidValue rowId,
     DomainRowOwnerCache ownerCache,
+    Transaction transaction,
   ) async {
     final cached = ownerCache[(tableName, rowId)];
     if (cached != null) return cached;
@@ -950,6 +1020,7 @@ class OfflineSyncEngine {
       'SELECT "spaceId" FROM "$escapedTableName" '
       'WHERE "id" = $encodedRowId '
       'LIMIT 1',
+      transaction: transaction,
     );
     final owner = result.isEmpty
         ? (exists: false, spaceId: null)
@@ -1031,12 +1102,14 @@ class OfflineSyncEngine {
   Future<Map<int, List<CrdtDataField>>> _loadAttemptedValueFields(
     DatabaseSession session,
     List<CrdtDataRow> rows,
+    Transaction transaction,
   ) async {
     final rowIds = {for (final row in rows) ?row.id};
     if (rowIds.isEmpty) return {};
 
     final fields = await CrdtDataField.db.find(
       session,
+      transaction: transaction,
       where: (t) => t.rowId.inSet(rowIds) & t.attemptedValue.id.notEquals(null),
       include: CrdtDataField.include(
         column: CrdtSchemaColumn.include(),

test/serverpod_offline_sync_test_server/test/dst/README.md

@@ -113,8 +113,15 @@ Causal completeness is relative to the receiver: prerequisites must already be
 committed there or accompany the incoming fact. Each delta is collected against
 committed receiver progress, never a queued batch's anticipated progress. Pending
 batches can overlap and arrive in any order without relying on one another.
-Collection and merge run sequentially; this does not simulate concurrent writes
+Both modes inject two or three seeded operations during capture or between
-during collection or the continuous sync driver's in-session send cursor.
+emitted changes. Capture starts an independent writer after insert metadata is
+read; SQLite queues it until capture completes. A pause of up to 500 ms lets an
+unprotected writer commit before payload reads resume. All writes finish before
+collection returns. Setup and drain do not inject writes, and the driver's
+in-session send cursor is not simulated.
+
+`dst_outbound_snapshot_test.dart` covers both schedules and detects checkpoint
+gaps, including when collection buffers without a transaction.
 
 Neither mode permanently drops facts or splits a collected batch. Wire chunking
 belongs below this boundary: the protocol reassembles chunks through
@@ -225,8 +232,9 @@ batch size, total delivered changes, and receive-isolation events.
 Network observations also report checkpoint collections, collections with a
 nonzero remote-author checkpoint, empty collections, partial batches, explicit
 replays, and fresh/repeated delivered change keys. A partial batch is nonempty,
-uses committed remote progress, and is smaller than the unchanged sender's full
+uses committed remote progress, and is smaller than the sender's full export
-export. Full exports in delta mode measure omissions only and are never queued.
+measured before that collection's injected writes. Full exports in delta mode
+measure omissions only and are never queued.
 Fresh means the key was not previously delivered to this receiver; it does not
 claim the fact was absent from local authoring. A repeated key has the same
 identity, not necessarily identical serialized insertion payload.
@@ -235,7 +243,12 @@ Counters carry `setup.`, `scheduled.`, or `drain.` prefixes according to when th
 collection or delivery occurred. Exact replays do not count as new partial
 batches. Delta stress runs (100+ rounds) require at least one scheduled partial
 batch; setup and final draining cannot satisfy that gate. Shorter runs remain
-smoke checks. The soak script retains `METRICS.jsonl` even for passing runs,
+smoke checks. `collectionInterleavings` and `captureInterleavings` count schedules;
+their `*InterleavedCommits` counters count eventual commits. Yield counters also
+record the emitted change kind (`.insert`, `.update`, `.delete`). In either
+delivery mode, runs of 100+ rounds require scheduled commits from both types.
+
+The soak script retains `METRICS.jsonl` even for passing runs,
 adding a `runId` that joins each record to the run's `id` in `RUNS.tsv`.
 
 Coverage counts distinct field/tombstone events rather than repeated snapshot

test/serverpod_offline_sync_test_server/test/dst/dst_outbound_snapshot_test.dart

@@ -0,0 +1,307 @@
+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 '../integration/test_tools/client_session.dart';
+import 'framework/dst_adversary.dart';
+import 'framework/dst_random.dart';
+import 'framework/dst_snapshot.dart';
+import 'framework/dst_world.dart';
+
+void main() {
+  initTestClientSession(createSessionPerTest: false);
+
+  group('Given a delta peer missing a city and a source with a pending insert,', () {
+    final ids = DstIds(DstRandom(165));
+    final clock = DstClock();
+    final space = ids.next();
+    late DstReplica source;
+    late DstReplica target;
+    late City b;
+
+    setUpAll(() async {
+      source = await _replica('source', space, ids, clock);
+      target = await _replica('target', space, ids, clock);
+      b = City(id: ids.next(), name: 'B0');
+      await source.withReplicaClock(
+        () => source.session.db.transactionForUser(
+          space,
+          (tx) => City.db.insertRow(source.session, b, transaction: tx),
+        ),
+      );
+    });
+
+    group(
+      'when a mixed write starts during metadata capture and delta delivery drains,',
+      () {
+        late Map<String, int> metrics;
+        late String expected;
+        late String received;
+        late List<String> names;
+
+        setUpAll(() async {
+          final adversary = DstAdversary(
+            _CaptureSchedule(),
+            [source, target],
+            delivery: DstDeliveryMode.delta,
+            onCollecting: (replica, collectedSpace) async {
+              await replica.session.db.transactionForUser(collectedSpace, (tx) async {
+                await City.db.insertRow(
+                  replica.session,
+                  City(id: ids.next(), name: 'A'),
+                  transaction: tx,
+                );
+                await City.db.updateRow(
+                  replica.session,
+                  b.copyWith(name: 'B1'),
+                  columns: (t) => [t.name],
+                  transaction: tx,
+                );
+                await City.db.insertRow(
+                  replica.session,
+                  City(id: ids.next(), name: 'C'),
+                  transaction: tx,
+                );
+              });
+              return 1;
+            },
+          );
+
+          await adversary.step((_) async {});
+          adversary.phase = DstNetworkPhase.drain;
+          await adversary.quiesce((_) async {});
+          metrics = adversary.metrics;
+          expected = (await DstSnapshot.capture(source)).renderSpace(space);
+          received = (await DstSnapshot.capture(target)).renderSpace(space);
+          names = (await City.db.find(target.session)).map((row) => row.name).toList()
+            ..sort();
+        });
+
+        test('then every city converges without a full-history repair.', () {
+          expect(names, ['A', 'B1', 'C']);
+          expect(received, expected);
+        });
+
+        test('then coverage records a committed write attempted during capture.', () {
+          expect(metrics['scheduled.captureInterleavings'], 1);
+          expect(metrics['scheduled.captureInterleavedCommits'], 1);
+          expect(metrics['collectionInterleavings'], isNull);
+          expect(metrics['drain.captureInterleavings'], isNull);
+          expect(metrics['fullCollections'], isNull);
+        });
+      },
+    );
+  });
+
+  group('Given a delta peer that has not received an authored city,', () {
+    final ids = DstIds(DstRandom(163));
+    final clock = DstClock();
+    final space = ids.next();
+    late DstReplica source;
+    late DstReplica target;
+    late City b;
+
+    setUpAll(() async {
+      source = await _replica('source', space, ids, clock);
+      target = await _replica('target', space, ids, clock);
+      b = City(id: ids.next(), name: 'B0');
+      await source.withReplicaClock(
+        () => source.session.db.transactionForUser(
+          space,
+          (tx) => City.db.insertRow(source.session, b, transaction: tx),
+        ),
+      );
+    });
+
+    group(
+      'when a mixed transaction commits at the first collector yield and delta delivery drains,',
+      () {
+        late Map<String, int> metrics;
+        late String expected;
+        late String received;
+        late List<String> names;
+        late CrdtMergeSet remaining;
+
+        setUpAll(() async {
+          final adversary = DstAdversary(
+            _FirstYieldSchedule(),
+            [source, target],
+            delivery: DstDeliveryMode.delta,
+            onCollecting: (replica, collectedSpace) async {
+              await replica.session.db.transactionForUser(collectedSpace, (tx) async {
+                await City.db.insertRow(
+                  replica.session,
+                  City(id: ids.next(), name: 'A'),
+                  transaction: tx,
+                );
+                await City.db.updateRow(
+                  replica.session,
+                  b.copyWith(name: 'B1'),
+                  columns: (t) => [t.name],
+                  transaction: tx,
+                );
+                await City.db.insertRow(
+                  replica.session,
+                  City(id: ids.next(), name: 'C'),
+                  transaction: tx,
+                );
+              });
+              return 1;
+            },
+          );
+
+          await adversary.step((_) async {});
+          adversary.phase = DstNetworkPhase.drain;
+          await adversary.quiesce((_) async {});
+          metrics = adversary.metrics;
+          expected = (await DstSnapshot.capture(source)).renderSpace(space);
+          received = (await DstSnapshot.capture(target)).renderSpace(space);
+          names = (await City.db.find(target.session)).map((row) => row.name).toList()
+            ..sort();
+          remaining = await source.collect(
+            space,
+            checkpoints: await target.checkpoints(space),
+          );
+        });
+
+        test('then every inserted city converges without a full-history repair.', () {
+          expect(names, ['A', 'B1', 'C']);
+          expect(received, expected);
+          expect(remaining, isEmpty);
+        });
+
+        test(
+          'then coverage records a committed interleaving during scheduled collection.',
+          () {
+            expect(metrics['scheduled.collectionInterleavings'], 1);
+            expect(metrics['scheduled.collectionInterleavedCommits.insert'], 1);
+            expect(metrics['drain.collectionInterleavings'], isNull);
+            expect(metrics['fullCollections'], isNull);
+          },
+        );
+      },
+    );
+  });
+
+  group('Given synchronized cities and one unacknowledged name update,', () {
+    final ids = DstIds(DstRandom(164));
+    final clock = DstClock();
+    final space = ids.next();
+    late DstReplica source;
+    late DstReplica target;
+    late City b;
+    late City deleted;
+
+    setUpAll(() async {
+      source = await _replica('source', space, ids, clock);
+      target = await _replica('target', space, ids, clock);
+      b = City(id: ids.next(), name: 'B0');
+      deleted = City(id: ids.next(), name: 'deleted');
+      await source.withReplicaClock(
+        () => source.session.db.transactionForUser(space, (tx) async {
+          await City.db.insertRow(source.session, b, transaction: tx);
+          await City.db.insertRow(source.session, deleted, transaction: tx);
+        }),
+      );
+      await target.merge(await source.collect(space), space);
+      await source.withReplicaClock(
+        () => source.session.db.transactionForUser(
+          space,
+          (tx) => City.db.updateRow(
+            source.session,
+            b.copyWith(name: 'B1'),
+            columns: (t) => [t.name],
+            transaction: tx,
+          ),
+        ),
+      );
+    });
+
+    group(
+      'when an update-delete transaction commits at the first collector yield and delta delivery drains,',
+      () {
+        late Map<String, int> metrics;
+        late String expected;
+        late String received;
+        late List<String> names;
+        late CrdtMergeSet remaining;
+
+        setUpAll(() async {
+          final adversary = DstAdversary(
+            _FirstYieldSchedule(),
+            [source, target],
+            delivery: DstDeliveryMode.delta,
+            onCollecting: (replica, collectedSpace) async {
+              await replica.session.db.transactionForUser(collectedSpace, (tx) async {
+                await City.db.updateRow(
+                  replica.session,
+                  b.copyWith(name: 'B2'),
+                  columns: (t) => [t.name],
+                  transaction: tx,
+                );
+                await City.db.deleteRow(replica.session, deleted, transaction: tx);
+              });
+              return 1;
+            },
+          );
+
+          await adversary.step((_) async {});
+          adversary.phase = DstNetworkPhase.drain;
+          await adversary.quiesce((_) async {});
+          metrics = adversary.metrics;
+          expected = (await DstSnapshot.capture(source)).renderSpace(space);
+          received = (await DstSnapshot.capture(target)).renderSpace(space);
+          names = (await City.db.find(target.session)).map((row) => row.name).toList()
+            ..sort();
+          remaining = await source.collect(
+            space,
+            checkpoints: await target.checkpoints(space),
+          );
+        });
+
+        test('then the latest name converges without a full-history repair.', () {
+          expect(names, ['B2']);
+          expect(received, expected);
+          expect(remaining, isEmpty);
+        });
+
+        test('then coverage records a committed interleaving after an update.', () {
+          expect(metrics['scheduled.collectionInterleavings'], 1);
+          expect(metrics['scheduled.collectionInterleavedCommits.update'], 1);
+          expect(metrics['drain.collectionInterleavings'], isNull);
+          expect(metrics['fullCollections'], isNull);
+        });
+      },
+    );
+  });
+}
+
+Future<DstReplica> _replica(String name, UuidValue space, DstIds ids, DstClock clock) =>
+    DstReplica.create(
+      name: name,
+      spaceUuids: [space],
+      nodeUuid: ids.next(),
+      clock: clock.clock,
+    );
+
+class _FirstYieldSchedule extends DstRandom {
+  _FirstYieldSchedule() : super(163);
+
+  @override
+  bool chance(double probability) =>
+      probability == DstAdversary.collectionProbability ||
+      probability == DstAdversary.collectionInterleavingProbability;
+
+  @override
+  int between(int min, int max) => min;
+
+  @override
+  T pick<T>(List<T> values) => values.first;
+}
+
+class _CaptureSchedule extends _FirstYieldSchedule {
+  @override
+  bool chance(double probability) =>
+      probability == DstAdversary.captureInterleavingProbability ||
+      super.chance(probability);
+}

test/serverpod_offline_sync_test_server/test/dst/framework/dst_adversary.dart

@@ -61,14 +61,22 @@ class DstAdversary {
     this.random,
     this.replicas, {
     this.delivery = DstDeliveryMode.full,
+    this.onCollecting,
   });
 
   final DstDeliveryMode delivery;
+
+  /// Attempts scheduled application writes during capture or at a collector
+  /// yield. Returns commits, so skipped writes cannot count as races.
+  final Future<int> Function(DstReplica source, UuidValue spaceUuid)? onCollecting;
+
   DstNetworkPhase phase = DstNetworkPhase.scheduled;
 
   static const receiveIsolationProbability = 0.2;
   static const collectionProbability = 0.8;
   static const resendProbability = 0.15;
+  static const collectionInterleavingProbability = 0.35;
+  static const captureInterleavingProbability = 0.5;
 
   /// The simulation's randomness.
   final DstRandom random;
@@ -178,18 +186,22 @@ class DstAdversary {
     UuidValue spaceUuid, {
     bool allowResend = true,
   }) async {
-    final changes = await source.collect(spaceUuid);
+    if (delivery == DstDeliveryMode.delta) {
-    if (changes.isEmpty && delivery == DstDeliveryMode.full) return;
+      for (final target in replicas) {
+        if (identical(target, source)) continue;
+        if (!target.spaceUuids.contains(spaceUuid)) continue;
+        await _collectDelta(source, target, spaceUuid, allowResend);
+      }
+      return;
+    }
+
+    final changes = await _collectPending(source, spaceUuid);
+    if (changes.isEmpty) return;
 
     for (final target in replicas) {
       if (identical(target, source)) continue;
       if (!target.spaceUuids.contains(spaceUuid)) continue;
 
-      if (delivery == DstDeliveryMode.delta) {
-        await _collectDelta(source, target, spaceUuid, changes.length, allowResend);
-        continue;
-      }
-
       _count('fullCollections');
       final delivered = _deliveredKeys.putIfAbsent(target.name, () => {});
       // Occasionally resend what the target already merged. Redelivery must be
@@ -220,7 +232,6 @@ class DstAdversary {
     DstReplica source,
     DstReplica target,
     UuidValue spaceUuid,
-    int fullSize,
     bool allowResend,
   ) async {
     final edge = '${source.name}|${target.name}|$spaceUuid';
@@ -238,10 +249,15 @@ class DstAdversary {
     );
     if (advanced) _count('advancedCheckpointCollections');
 
-    // No local writes or merges interleave the full-size observation above
+    // Measure each receiver's baseline before its collection, since an earlier
-    // and this collection. The full export measures omission only; it is never
+    // receiver's collection may have interleaved a write. This full export is
-    // delivered as a repair in delta mode.
+    // observation only; delta mode never delivers it as a repair.
-    final changes = await source.collect(spaceUuid, checkpoints: checkpoints);
+    final fullSize = (await source.collect(spaceUuid)).length;
+    final changes = await _collectPending(
+      source,
+      spaceUuid,
+      checkpoints: checkpoints,
+    );
     if (changes.isEmpty) {
       _count('emptyCollections');
       return;
@@ -259,6 +275,52 @@ class DstAdversary {
     );
   }
 
+  Future<CrdtMergeSet> _collectPending(
+    DstReplica source,
+    UuidValue spaceUuid, {
+    List<Hlc> checkpoints = const [],
+  }) async {
+    final interleave = onCollecting;
+    if (phase != DstNetworkPhase.scheduled ||
+        interleave == null ||
+        !random.chance(collectionInterleavingProbability)) {
+      return source.collect(spaceUuid, checkpoints: checkpoints);
+    }
+
+    if (random.chance(captureInterleavingProbability)) {
+      return source.collect(
+        spaceUuid,
+        checkpoints: checkpoints,
+        onCapture: () async {
+          _count('captureInterleavings');
+          final committed = await interleave(source, spaceUuid);
+          if (committed > 0) _count('captureInterleavedCommits', committed);
+        },
+      );
+    }
+
+    final yieldIndex = random.between(1, 3);
+    var emitted = 0;
+    return source.collect(
+      spaceUuid,
+      checkpoints: checkpoints,
+      onChange: (change) async {
+        if (++emitted != yieldIndex) return;
+        _count('collectionInterleavings');
+        final committed = await interleave(source, spaceUuid);
+        if (committed > 0) {
+          _count('collectionInterleavedCommits', committed);
+          final kind = switch (change) {
+            CrdtMergeInsert() => 'insert',
+            CrdtMergeUpdate() => 'update',
+            CrdtMergeDelete() => 'delete',
+          };
+          _count('collectionInterleavedCommits.$kind', committed);
+        }
+      },
+    );
+  }
+
   Future<void> _deliverOne(Future<void> Function(DstReplica) onMerged) async {
     final deliverable = [
       for (final delivery in _pending)

test/serverpod_offline_sync_test_server/test/dst/framework/dst_runner.dart

@@ -196,8 +196,24 @@ Future<DstRunReport> runDstSimulation({
 
   final operations = DstOperations(random, ids);
   operations.oracle.accept(await DstSnapshot.capture(replicas.first));
-  final adversary = DstAdversary(random, replicas, delivery: delivery)
+  final adversary = DstAdversary(
-    ..phase = DstNetworkPhase.setup;
+    random,
+    replicas,
+    delivery: delivery,
+    onCollecting: (replica, space) async {
+      var committed = 0;
+      // Multiple operations can cross change kinds (insert/update/delete)
+      // before the collector resumes. A lone insertion would not expose a
+      // later change advancing the checkpoint past the missing insertion.
+      final attempts = random.between(2, 3);
+      for (var index = 0; index < attempts; index++) {
+        final outcome = await operations.step(replica, space);
+        simulationClock.advance(Duration(milliseconds: random.between(1, 40)));
+        if (outcome == DstOperationOutcome.applied) committed++;
+      }
+      return committed;
+    },
+  )..phase = DstNetworkPhase.setup;
   var schedulingStarted = false;
   var setupAttempted = 0;
   var setupCommitted = 0;
@@ -330,6 +346,18 @@ Future<DstRunReport> runDstSimulation({
         'Delta stress run delivered no scheduled partial batches: ${report.toJson()}',
       );
     }
+    if (rounds >= 100 &&
+        (adversary.metrics['scheduled.collectionInterleavedCommits'] ?? 0) == 0) {
+      throw StateError(
+        'Stress run committed no writes during collection: ${report.toJson()}',
+      );
+    }
+    if (rounds >= 100 &&
+        (adversary.metrics['scheduled.captureInterleavedCommits'] ?? 0) == 0) {
+      throw StateError(
+        'Stress run committed no writes attempted during capture: ${report.toJson()}',
+      );
+    }
     if (resolvedProfile == DstProfile.populated) {
       final transitions = operations.coverage.transitions;
       final missing = [

test/serverpod_offline_sync_test_server/test/dst/framework/dst_world.dart

@@ -1,3 +1,5 @@
+import 'dart:async';
+
 import 'package:clock/clock.dart';
 // Imported with `show` because the barrel below re-exports overlapping names.
 import 'package:serverpod_database/serverpod_database.dart'
@@ -7,6 +9,7 @@ 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 '../../integration/test_tools/client_session.dart';
+import '../../integration/test_tools/outbound_capture_database.dart';
 import 'dst_authored.dart';
 import 'dst_coverage.dart';
 import 'dst_random.dart';
@@ -138,6 +141,16 @@ class DstReplica {
   /// This replica's sync engine, used to collect outbound changes.
   final OfflineSyncEngine sync;
 
+  late final _captureDatabase = OutboundCaptureDatabase(
+    rawSession.db,
+    syncTables: dstSyncTables,
+  );
+
+  late final _captureSession = OfflineSyncDatabaseSession(
+    _captureDatabase,
+    syncTables: dstSyncTables,
+  );
+
   /// The spaces the adversary exchanges for this replica.
   final List<UuidValue> spaceUuids;
 
@@ -181,21 +194,54 @@ class DstReplica {
   /// An empty vector preserves the full-history convergence experiment. Delta
   /// callers pass the receiver's production handshake vector, including one
   /// entry per known author in this space, never progress from queued batches.
+  /// [onChange] lets the scheduler commit an application operation at a real
+  /// collector yield, before requesting the next change from the same stream.
+  /// [onCapture] starts an independent writer after insert metadata is read.
+  /// It finishes before collection returns, even when SQLite queues it behind
+  /// capture, so later scheduling never overlaps its random or clock state.
   Future<CrdtMergeSet> collect(
     UuidValue spaceUuid, {
     List<Hlc> checkpoints = const [],
+    Future<void> Function(CrdtMergeChange)? onChange,
+    Future<void> Function()? onCapture,
   }) async {
     return withReplicaClock(
-      () => sync
+      () async {
-          .collectPendingChanges(
+        Future<void>? captureWrite;
-            rawSession,
+        if (onCapture != null) {
+          final writerZone = Zone.current;
+          _captureDatabase.afterInsertMetadata = () async {
+            captureWrite = writerZone.run(onCapture);
+            // An unprotected writer can finish before payload reads resume.
+            // A snapshot queues it until capture releases the write lock.
+            await Future.any([
+              captureWrite!,
+              Future<void>.delayed(const Duration(milliseconds: 500)),
+            ]);
+          };
+        }
+
+        final changes = <CrdtMergeChange>[];
+        try {
+          await for (final change in sync.collectPendingChanges(
+            onCapture == null ? rawSession : _captureSession,
             checkpointsBySpaceUuid: {
               spaceUuid: OfflineSyncSpaceState.normalizeCheckpoints(
                 checkpoints,
               ).values.toList(),
             },
-          )
+          )) {
-          .toList(),
+            changes.add(change);
+            await onChange?.call(change);
+          }
+        } finally {
+          if (onCapture != null) {
+            _captureDatabase.afterInsertMetadata = null;
+            await captureWrite;
+          }
+        }
+        return changes;
+      },
     );
   }
 

test/serverpod_offline_sync_test_server/test/integration/sync/outbound_snapshot_concurrency_test.dart

@@ -0,0 +1,197 @@
+import 'dart:async';
+
+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/outbound_capture_database.dart';
+import '../test_tools/sync_topology.dart';
+
+void main() {
+  initTestClientSession(createSessionPerTest: false);
+
+  group('Given an author with a pending person and an independent writer,', () {
+    late SyncNode author;
+    late SyncNode peer;
+    late Person b;
+    late OutboundCaptureDatabase captureDatabase;
+    late OfflineSyncDatabaseSession collector;
+
+    setUpAll(() async {
+      author = await syncNode(await createAdditionalTestSession(), [Person.t]);
+      peer = await syncNode(await createAdditionalTestSession(), [Person.t]);
+      b = Person(id: const Uuid().v7obj(), name: 'B0');
+      await author.offlineSync.db.transactionForUser(
+        testCrdtUserId,
+        (tx) => Person.db.insertRow(author.offlineSync, b, transaction: tx),
+      );
+      captureDatabase = OutboundCaptureDatabase(author.raw.db, syncTables: [Person.t]);
+      collector = OfflineSyncDatabaseSession(captureDatabase, syncTables: [Person.t]);
+      await captureDatabase.initialize();
+    });
+
+    group('when a mixed write starts between metadata and payload capture,', () {
+      late List<CrdtMergeChange> firstPass;
+      late List<String> receivedNames;
+
+      setUpAll(() async {
+        final writerZone = Zone.current;
+        Future<void>? write;
+        captureDatabase.afterInsertMetadata = () async {
+          write = writerZone.run(
+            () => author.offlineSync.db.transactionForUser(testCrdtUserId, (tx) async {
+              await Person.db.insertRow(
+                author.offlineSync,
+                Person(id: const Uuid().v7obj(), name: 'A'),
+                transaction: tx,
+              );
+              await Person.db.updateRow(
+                author.offlineSync,
+                b.copyWith(name: 'B1'),
+                columns: (t) => [t.name],
+                transaction: tx,
+              );
+              await Person.db.insertRow(
+                author.offlineSync,
+                Person(id: const Uuid().v7obj(), name: 'C'),
+                transaction: tx,
+              );
+            }),
+          );
+
+          // Let an unprotected writer finish before payload capture resumes.
+          // SQLite's snapshot transaction instead keeps this writer queued.
+          await Future.any([
+            write!,
+            Future<void>.delayed(const Duration(milliseconds: 500)),
+          ]);
+        };
+
+        firstPass = await author.sync
+            .collectPendingChanges(
+              collector,
+              checkpointsBySpaceUuid: {testCrdtUserId: const []},
+            )
+            .toList();
+        await write;
+        await peer.sync.mergeInboundBatch(
+          peer.raw,
+          spaceId: testCrdtUserId,
+          mergeSet: firstPass,
+        );
+        final since = await peer.sync.createSyncSinceHlc(
+          peer.raw,
+          spaceId: testCrdtUserId,
+        );
+        final nextPass = await author.sync
+            .collectPendingChanges(
+              author.raw,
+              checkpointsBySpaceUuid: {testCrdtUserId: since.nodeCheckpoints},
+            )
+            .toList();
+        await peer.sync.mergeInboundBatch(
+          peer.raw,
+          spaceId: testCrdtUserId,
+          mergeSet: nextPass,
+        );
+        receivedNames = (await Person.db.find(
+          peer.offlineSync,
+        )).map((row) => row.name).toList()..sort();
+      });
+
+      test(
+        'then the captured payload retains the value from its metadata snapshot.',
+        () {
+          expect(firstPass, hasLength(1));
+          expect(((firstPass.single as CrdtMergeInsert).data as Person).name, 'B0');
+        },
+      );
+
+      test('then checkpoint delivery preserves every person and the latest name.', () {
+        expect(receivedNames, ['A', 'B1', 'C']);
+      });
+    });
+  });
+
+  group(
+    'Given an author with a pending person to synchronize,',
+    () {
+      late SyncNode author;
+      late OutboundCaptureDatabase captureDatabase;
+      late OfflineSyncDatabaseSession collector;
+
+      setUpAll(() async {
+        author = await syncNode(await createAdditionalTestSession(), [Person.t]);
+        await author.offlineSync.db.transactionForUser(
+          testCrdtUserId,
+          (tx) => Person.db.insertRow(
+            author.offlineSync,
+            Person(id: const Uuid().v7obj(), name: 'B0'),
+            transaction: tx,
+          ),
+        );
+        captureDatabase = OutboundCaptureDatabase(
+          author.raw.db,
+          syncTables: [Person.t],
+        );
+        collector = OfflineSyncDatabaseSession(captureDatabase, syncTables: [Person.t]);
+        await captureDatabase.initialize();
+      });
+
+      group(
+        'when an application transaction queues behind capture and reads before inserting,',
+        () {
+          Object? writeError;
+          late List<String> visibleNames;
+          late List<String> committedNames;
+
+          setUpAll(() async {
+            final writerZone = Zone.current;
+            Future<void>? write;
+            visibleNames = [];
+            captureDatabase.afterInsertMetadata = () async {
+              write = writerZone.run(
+                () => author.offlineSync.db
+                    .transactionForUser(testCrdtUserId, (tx) async {
+                      visibleNames = (await Person.db.find(
+                        author.offlineSync,
+                        transaction: tx,
+                      )).map((row) => row.name).toList();
+                      await Person.db.insertRow(
+                        author.offlineSync,
+                        Person(id: const Uuid().v7obj(), name: 'A'),
+                        transaction: tx,
+                      );
+                    })
+                    .then<void>((_) {}, onError: (Object error) => writeError = error),
+              );
+
+              await Future<void>.delayed(const Duration(milliseconds: 100));
+            };
+
+            await author.sync
+                .collectPendingChanges(
+                  collector,
+                  checkpointsBySpaceUuid: {testCrdtUserId: const []},
+                )
+                .toList();
+            await write;
+            committedNames = (await Person.db.find(
+              author.offlineSync,
+            )).map((row) => row.name).toList()..sort();
+          });
+
+          test(
+            'then the visible read and write commit after capture releases its lock.',
+            () {
+              expect(writeError, isNull);
+              expect(visibleNames, ['B0']);
+              expect(committedNames, ['A', 'B0']);
+            },
+          );
+        },
+      );
+    },
+  );
+}

test/serverpod_offline_sync_test_server/test/integration/sync/outbound_snapshot_test.dart

@@ -0,0 +1,271 @@
+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/sync_topology.dart';
+
+void main() {
+  initTestClientSession(createSessionPerTest: false);
+
+  group('Given an author with one pending person insert,', () {
+    late SyncNode author;
+    late SyncNode peer;
+    late Person b;
+
+    setUpAll(() async {
+      author = await syncNode(await createAdditionalTestSession(), [Person.t]);
+      peer = await syncNode(await createAdditionalTestSession(), [Person.t]);
+      b = Person(id: const Uuid().v7obj(), name: 'B0');
+      await author.offlineSync.db.transactionForUser(
+        testCrdtUserId,
+        (tx) => Person.db.insertRow(author.offlineSync, b, transaction: tx),
+      );
+    });
+
+    group(
+      'when an insert-update-insert transaction commits after collection first yields and delivery resumes,',
+      () {
+        late List<String> collectedInserts;
+        late List<String> authoredNames;
+        late List<String> receivedNames;
+        late List<String> recoveredNames;
+
+        setUpAll(() async {
+          final firstPass = <CrdtMergeChange>[];
+          var committed = false;
+          await for (final change in author.sync.collectPendingChanges(
+            author.raw,
+            checkpointsBySpaceUuid: {testCrdtUserId: const []},
+          )) {
+            firstPass.add(change);
+            if (committed) continue;
+            committed = true;
+
+            await author.offlineSync.db.transactionForUser(testCrdtUserId, (tx) async {
+              await Person.db.insertRow(
+                author.offlineSync,
+                Person(id: const Uuid().v7obj(), name: 'A'),
+                transaction: tx,
+              );
+              await Person.db.updateRow(
+                author.offlineSync,
+                b.copyWith(name: 'B1'),
+                columns: (t) => [t.name],
+                transaction: tx,
+              );
+              await Person.db.insertRow(
+                author.offlineSync,
+                Person(id: const Uuid().v7obj(), name: 'C'),
+                transaction: tx,
+              );
+            });
+          }
+
+          await peer.sync.mergeInboundBatch(
+            peer.raw,
+            spaceId: testCrdtUserId,
+            mergeSet: firstPass,
+          );
+          final secondPass = await _deliverPending(author, peer);
+          final thirdPass = await _deliverPending(author, peer);
+          collectedInserts = [
+            for (final change in [...firstPass, ...secondPass, ...thirdPass])
+              if (change is CrdtMergeInsert) (change.data as Person).name,
+          ];
+          authoredNames = await _personNames(author);
+          receivedNames = await _personNames(peer);
+
+          await pushChanges(author, peer);
+          recoveredNames = await _personNames(peer);
+        });
+
+        test('then every committed insert is collected across the passes.', () {
+          expect(collectedInserts, unorderedEquals(['B0', 'A', 'C']));
+        });
+
+        test('then the peer has every committed person after checkpoint delivery.', () {
+          expect(authoredNames, ['A', 'B1', 'C']);
+          expect(receivedNames, authoredNames);
+        });
+
+        test('then a full-history resend restores all committed people.', () {
+          expect(recoveredNames, authoredNames);
+        });
+      },
+    );
+  });
+
+  group('Given two synchronized people and one pending name update,', () {
+    late SyncNode author;
+    late SyncNode peer;
+    late Person b;
+    late Person deleted;
+    late List<Hlc> checkpoints;
+
+    setUpAll(() async {
+      author = await syncNode(await createAdditionalTestSession(), [Person.t]);
+      peer = await syncNode(await createAdditionalTestSession(), [Person.t]);
+      b = Person(id: const Uuid().v7obj(), name: 'B0');
+      deleted = Person(id: const Uuid().v7obj(), name: 'deleted');
+      await author.offlineSync.db.transactionForUser(testCrdtUserId, (tx) async {
+        await Person.db.insertRow(author.offlineSync, b, transaction: tx);
+        await Person.db.insertRow(author.offlineSync, deleted, transaction: tx);
+      });
+      await pushChanges(author, peer);
+      checkpoints = (await peer.sync.createSyncSinceHlc(
+        peer.raw,
+        spaceId: testCrdtUserId,
+      )).nodeCheckpoints;
+
+      await author.offlineSync.db.transactionForUser(
+        testCrdtUserId,
+        (tx) => Person.db.updateRow(
+          author.offlineSync,
+          b.copyWith(name: 'B1'),
+          columns: (t) => [t.name],
+          transaction: tx,
+        ),
+      );
+    });
+
+    group(
+      'when an update-delete transaction commits after collection first yields and delivery resumes,',
+      () {
+        late List<String> authoredNames;
+        late List<String> receivedNames;
+        late List<String> recoveredNames;
+
+        setUpAll(() async {
+          final firstPass = <CrdtMergeChange>[];
+          var committed = false;
+          await for (final change in author.sync.collectPendingChanges(
+            author.raw,
+            checkpointsBySpaceUuid: {testCrdtUserId: checkpoints},
+          )) {
+            firstPass.add(change);
+            if (committed) continue;
+            committed = true;
+
+            await author.offlineSync.db.transactionForUser(testCrdtUserId, (tx) async {
+              await Person.db.updateRow(
+                author.offlineSync,
+                b.copyWith(name: 'B2'),
+                columns: (t) => [t.name],
+                transaction: tx,
+              );
+              await Person.db.deleteRow(author.offlineSync, deleted, transaction: tx);
+            });
+          }
+
+          await peer.sync.mergeInboundBatch(
+            peer.raw,
+            spaceId: testCrdtUserId,
+            mergeSet: firstPass,
+          );
+          await _deliverPending(author, peer);
+          authoredNames = await _personNames(author);
+          receivedNames = await _personNames(peer);
+
+          await pushChanges(author, peer);
+          recoveredNames = await _personNames(peer);
+        });
+
+        test('then the peer retains the latest committed name.', () {
+          expect(authoredNames, ['B2']);
+          expect(receivedNames, authoredNames);
+        });
+
+        test('then a full-history resend restores the latest committed name.', () {
+          expect(recoveredNames, authoredNames);
+        });
+      },
+    );
+  });
+
+  group('Given pending people collected through a fresh sync wrapper,', () {
+    late SyncNode author;
+    late OfflineSyncDatabaseSession collector;
+
+    setUpAll(() async {
+      author = await syncNode(await createAdditionalTestSession(), [Person.t]);
+      await author.offlineSync.db.transactionForUser(testCrdtUserId, (tx) async {
+        await Person.db.insertRow(
+          author.offlineSync,
+          Person(id: const Uuid().v7obj(), name: 'A'),
+          transaction: tx,
+        );
+        await Person.db.insertRow(
+          author.offlineSync,
+          Person(id: const Uuid().v7obj(), name: 'B'),
+          transaction: tx,
+        );
+      });
+      collector = OfflineSyncDatabaseSession.wraps(author.raw, syncTables: [Person.t]);
+    });
+
+    group('when the consumer cancels after one change and writes another person,', () {
+      late List<String> names;
+
+      setUpAll(() async {
+        await author.sync
+            .collectPendingChanges(
+              collector,
+              checkpointsBySpaceUuid: {testCrdtUserId: const []},
+            )
+            .first;
+        await author.offlineSync.db.transactionForUser(
+          testCrdtUserId,
+          (tx) => Person.db.insertRow(
+            author.offlineSync,
+            Person(id: const Uuid().v7obj(), name: 'C'),
+            transaction: tx,
+          ),
+        );
+        final changes = await author.sync
+            .collectPendingChanges(
+              collector,
+              checkpointsBySpaceUuid: {testCrdtUserId: const []},
+            )
+            .toList();
+        names =
+            changes
+                .whereType<CrdtMergeInsert>()
+                .map((change) => (change.data as Person).name)
+                .toList()
+              ..sort();
+      });
+
+      test(
+        'then the write and next collection complete without a retained transaction.',
+        () {
+          expect(names, ['A', 'B', 'C']);
+        },
+      );
+    });
+  });
+}
+
+Future<List<CrdtMergeChange>> _deliverPending(SyncNode author, SyncNode peer) async {
+  final since = await peer.sync.createSyncSinceHlc(
+    peer.raw,
+    spaceId: testCrdtUserId,
+  );
+  final changes = await author.sync
+      .collectPendingChanges(
+        author.raw,
+        checkpointsBySpaceUuid: {testCrdtUserId: since.nodeCheckpoints},
+      )
+      .toList();
+  await peer.sync.mergeInboundBatch(
+    peer.raw,
+    spaceId: testCrdtUserId,
+    mergeSet: changes,
+  );
+  return changes;
+}
+
+Future<List<String>> _personNames(SyncNode node) async {
+  final rows = await Person.db.find(node.offlineSync);
+  return rows.map((row) => row.name).toList()..sort();
+}

test/serverpod_offline_sync_test_server/test/integration/test_tools/outbound_capture_database.dart

@@ -0,0 +1,45 @@
+import 'package:serverpod_database/serverpod_database.dart' as database;
+import 'package:serverpod_offline_sync/serverpod_offline_sync.dart';
+
+/// Pauses real SQLite collection after the metadata query has completed.
+/// No query results or transaction behavior are replaced.
+class OutboundCaptureDatabase extends OfflineSyncDatabase {
+  OutboundCaptureDatabase(super.delegate, {required super.syncTables});
+
+  Future<void> Function()? afterInsertMetadata;
+
+  @override
+  Future<List<T>> find<T extends database.TableRow>({
+    database.Expression? where,
+    int? limit,
+    int? offset,
+    database.Column? orderBy,
+    List<database.Column>? orderByList,
+    bool orderDescending = false,
+    database.Transaction? transaction,
+    database.Include? include,
+    database.LockMode? lockMode,
+    database.LockBehavior? lockBehavior,
+  }) async {
+    final rows = await super.find<T>(
+      where: where,
+      limit: limit,
+      offset: offset,
+      orderBy: orderBy,
+      orderByList: orderByList,
+      orderDescending: orderDescending,
+      transaction: transaction,
+      include: include,
+      lockMode: lockMode,
+      lockBehavior: lockBehavior,
+    );
+
+    if (T == CrdtDataRow) {
+      final afterMetadata = afterInsertMetadata;
+      afterInsertMetadata = null;
+      await afterMetadata?.call();
+    }
+
+    return rows;
+  }
+}