From 4bf2359dda8e9bd53138df2e4f6d2db655a6bf82 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marcelo=20Mendon=C3=A7a=20Soares?= Date: Thu, 8 Oct 2026 20:32:24 -0300 Subject: [PATCH] fix: Capture outbound changes in a consistent snapshot (#166) --- docs/outbound-collection-consistency.md | 104 ------ docs/projection-model.md | 15 +- docs/shared-spaces.md | 11 +- .../lib/src/sync/engine.dart | 103 +++++- .../test/dst/README.md | 23 +- .../test/dst/dst_outbound_snapshot_test.dart | 307 ++++++++++++++++++ .../test/dst/framework/dst_adversary.dart | 86 ++++- .../test/dst/framework/dst_runner.dart | 32 +- .../test/dst/framework/dst_world.dart | 56 +++- .../outbound_snapshot_concurrency_test.dart | 197 +++++++++++ .../sync/outbound_snapshot_test.dart | 271 ++++++++++++++++ .../test_tools/outbound_capture_database.dart | 45 +++ 12 files changed, 1097 insertions(+), 153 deletions(-) delete mode 100644 docs/outbound-collection-consistency.md create mode 100644 test/serverpod_offline_sync_test_server/test/dst/dst_outbound_snapshot_test.dart create mode 100644 test/serverpod_offline_sync_test_server/test/integration/sync/outbound_snapshot_concurrency_test.dart create mode 100644 test/serverpod_offline_sync_test_server/test/integration/sync/outbound_snapshot_test.dart create mode 100644 test/serverpod_offline_sync_test_server/test/integration/test_tools/outbound_capture_database.dart diff --git a/docs/outbound-collection-consistency.md b/docs/outbound-collection-consistency.md deleted file mode 100644 index 5a15517..0000000 --- a/docs/outbound-collection-consistency.md +++ /dev/null @@ -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. diff --git a/docs/projection-model.md b/docs/projection-model.md index 7f4c806..fd5af11 100644 --- a/docs/projection-model.md +++ b/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 diff --git a/docs/shared-spaces.md b/docs/shared-spaces.md index f4a6b11..f2b9bde 100644 --- a/docs/shared-spaces.md +++ b/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`. diff --git a/packages/serverpod_offline_sync/lib/src/sync/engine.dart b/packages/serverpod_offline_sync/lib/src/sync/engine.dart index ce081b5..a561f95 100644 --- a/packages/serverpod_offline_sync/lib/src/sync/engine.dart +++ b/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 - /// column payloads are resolved incrementally as each change is yielded. + /// Changes are captured in one consistent transaction, including domain + /// 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( - session, - spaceUuidById, - checkpointsBySpaceId, - )) { + 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 spaceUuidById, Map> checkpointsBySpaceId, + Transaction transaction, ) async* { - // Domain ownership is immutable while a collection runs, so read each - // row's owner at most once across all three streams. + // Cache ownership within this collection's consistent snapshot. final ownerCache = DomainRowOwnerCache(); - yield* _streamInserts(session, spaceUuidById, checkpointsBySpaceId, ownerCache); - yield* _streamUpdates(session, spaceUuidById, checkpointsBySpaceId, ownerCache); - yield* _streamDeletes(session, spaceUuidById, checkpointsBySpaceId, ownerCache); + yield* _streamInserts( + session, + spaceUuidById, + checkpointsBySpaceId, + ownerCache, + transaction, + ); + yield* _streamUpdates( + session, + spaceUuidById, + checkpointsBySpaceId, + ownerCache, + transaction, + ); + yield* _streamDeletes( + session, + spaceUuidById, + checkpointsBySpaceId, + ownerCache, + transaction, + ); } Stream _streamInserts( @@ -562,9 +599,11 @@ class OfflineSyncEngine { Map spaceUuidById, Map> 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 spaceUuidById, Map> 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 spaceUuidById, Map> 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? 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>> _loadAttemptedValueFields( DatabaseSession session, List 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(), diff --git a/test/serverpod_offline_sync_test_server/test/dst/README.md b/test/serverpod_offline_sync_test_server/test/dst/README.md index d0624fe..b4c1b29 100644 --- a/test/serverpod_offline_sync_test_server/test/dst/README.md +++ b/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 -during collection or the continuous sync driver's in-session send cursor. +Both modes inject two or three seeded operations during capture or between +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 -export. Full exports in delta mode measure omissions only and are never queued. +uses committed remote progress, and is smaller than the sender's full export +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 diff --git a/test/serverpod_offline_sync_test_server/test/dst/dst_outbound_snapshot_test.dart b/test/serverpod_offline_sync_test_server/test/dst/dst_outbound_snapshot_test.dart new file mode 100644 index 0000000..c5104ec --- /dev/null +++ b/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 metrics; + late String expected; + late String received; + late List 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 metrics; + late String expected; + late String received; + late List 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 metrics; + late String expected; + late String received; + late List 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 _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(List values) => values.first; +} + +class _CaptureSchedule extends _FirstYieldSchedule { + @override + bool chance(double probability) => + probability == DstAdversary.captureInterleavingProbability || + super.chance(probability); +} diff --git a/test/serverpod_offline_sync_test_server/test/dst/framework/dst_adversary.dart b/test/serverpod_offline_sync_test_server/test/dst/framework/dst_adversary.dart index cfb8259..2f26f08 100644 --- a/test/serverpod_offline_sync_test_server/test/dst/framework/dst_adversary.dart +++ b/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 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 (changes.isEmpty && delivery == DstDeliveryMode.full) return; + if (delivery == DstDeliveryMode.delta) { + 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 - // and this collection. The full export measures omission only; it is never - // delivered as a repair in delta mode. - final changes = await source.collect(spaceUuid, checkpoints: checkpoints); + // Measure each receiver's baseline before its collection, since an earlier + // receiver's collection may have interleaved a write. This full export is + // observation only; delta mode never delivers it as a repair. + 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 _collectPending( + DstReplica source, + UuidValue spaceUuid, { + List 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 _deliverOne(Future Function(DstReplica) onMerged) async { final deliverable = [ for (final delivery in _pending) diff --git a/test/serverpod_offline_sync_test_server/test/dst/framework/dst_runner.dart b/test/serverpod_offline_sync_test_server/test/dst/framework/dst_runner.dart index 11c64fd..ff25ee2 100644 --- a/test/serverpod_offline_sync_test_server/test/dst/framework/dst_runner.dart +++ b/test/serverpod_offline_sync_test_server/test/dst/framework/dst_runner.dart @@ -196,8 +196,24 @@ Future runDstSimulation({ final operations = DstOperations(random, ids); operations.oracle.accept(await DstSnapshot.capture(replicas.first)); - final adversary = DstAdversary(random, replicas, delivery: delivery) - ..phase = DstNetworkPhase.setup; + final adversary = DstAdversary( + 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 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 = [ diff --git a/test/serverpod_offline_sync_test_server/test/dst/framework/dst_world.dart b/test/serverpod_offline_sync_test_server/test/dst/framework/dst_world.dart index dbe856f..9eebe38 100644 --- a/test/serverpod_offline_sync_test_server/test/dst/framework/dst_world.dart +++ b/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 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 collect( UuidValue spaceUuid, { List checkpoints = const [], + Future Function(CrdtMergeChange)? onChange, + Future Function()? onCapture, }) async { return withReplicaClock( - () => sync - .collectPendingChanges( - rawSession, + () async { + Future? captureWrite; + 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.delayed(const Duration(milliseconds: 500)), + ]); + }; + } + + final changes = []; + 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; + }, ); } diff --git a/test/serverpod_offline_sync_test_server/test/integration/sync/outbound_snapshot_concurrency_test.dart b/test/serverpod_offline_sync_test_server/test/integration/sync/outbound_snapshot_concurrency_test.dart new file mode 100644 index 0000000..e1c44e7 --- /dev/null +++ b/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 firstPass; + late List receivedNames; + + setUpAll(() async { + final writerZone = Zone.current; + Future? 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.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 visibleNames; + late List committedNames; + + setUpAll(() async { + final writerZone = Zone.current; + Future? 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((_) {}, onError: (Object error) => writeError = error), + ); + + await Future.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']); + }, + ); + }, + ); + }, + ); +} diff --git a/test/serverpod_offline_sync_test_server/test/integration/sync/outbound_snapshot_test.dart b/test/serverpod_offline_sync_test_server/test/integration/sync/outbound_snapshot_test.dart new file mode 100644 index 0000000..02e7502 --- /dev/null +++ b/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 collectedInserts; + late List authoredNames; + late List receivedNames; + late List recoveredNames; + + setUpAll(() async { + final firstPass = []; + 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 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 authoredNames; + late List receivedNames; + late List recoveredNames; + + setUpAll(() async { + final firstPass = []; + 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 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() + .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> _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> _personNames(SyncNode node) async { + final rows = await Person.db.find(node.offlineSync); + return rows.map((row) => row.name).toList()..sort(); +} diff --git a/test/serverpod_offline_sync_test_server/test/integration/test_tools/outbound_capture_database.dart b/test/serverpod_offline_sync_test_server/test/integration/test_tools/outbound_capture_database.dart new file mode 100644 index 0000000..aadf285 --- /dev/null +++ b/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 Function()? afterInsertMetadata; + + @override + Future> find({ + database.Expression? where, + int? limit, + int? offset, + database.Column? orderBy, + List? orderByList, + bool orderDescending = false, + database.Transaction? transaction, + database.Include? include, + database.LockMode? lockMode, + database.LockBehavior? lockBehavior, + }) async { + final rows = await super.find( + 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; + } +}