halkyon-days/serverpod-offline-sync
Watch
1
0
Fork
You've already forked serverpod-offline-sync
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix: Correct checkpoint progress and add delta DST coverage (#158)

Browse source
This commit is contained in:
Marcelo Mendonça Soares 2026-10-07 15:49:15 -03:00 • committed by GitHub
commit f59a90bc97
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
13 changed files with 1829 additions and 163 deletions
Download patch file Download diff file Expand all files Collapse all files

97
test/serverpod_offline_sync_test_server/test/dst/README.md
View file

@ -2,7 +2,7 @@
These suites drive several replicas through randomized operations and
adversarial delivery, then check properties the engine must hold. A run is
described by its seed, rounds, workload profile, and graph width. Those inputs
described by its seed, rounds, workload profile, graph width, and delivery mode. Those inputs
replay exactly within the same source/SDK revision.
They complement the `integration/` suites rather than replacing them.
@ -32,6 +32,7 @@ directory are untagged and run with ordinary `dart test`.
| `DST_SEED_BASE` | unix seconds | First seed; successive seeds increment |
| `DST_PROFILE` | sparse | `sparse`, `populated`, or `mixed` (even seeds populated, odd sparse) |
| `DST_GRAPH_WIDTH` | 2 | Rows per table in each populated graph, minimum 2 |
| `DST_DELIVERY` | full | `full` exports all retained facts; `delta` uses the receiver's committed checkpoint vector |
| `DST_DEBUG_TABLE` | unset | Print every merge touching this table, in delivery order |
A divergence is usually explained by which facts a replica had merged when it
@ -47,7 +48,7 @@ which defects those fixed seeds reach.
Nothing is lost to reproducibility: a seed determines its simulation entirely
and a failure prints its replay command. Tests are named by position rather
than by seed, so the suite stays stable while the schedules underneath it vary.
Pin the seed, rounds, profile, and graph width to re-run an exact sweep.
Pin the seed, rounds, profile, graph width, and delivery mode to re-run an exact sweep.
The consequence to expect is that a run can fail for a defect unrelated to the
change that triggered it. That is the suite doing its job; take the seed from
@ -102,27 +103,39 @@ Reorders, delays, redelivers already-merged batches, and isolates incoming
delivery to selected replicas for a few rounds. Sources can still send during
this receive isolation; this is not a bidirectional network partition.
Two moves are deliberately unavailable, both because the merge contract states
that input arrives as a causally complete snapshot of the sender
(`docs/foreign-key-invariants.md`):
`full` preserves the original merge convergence experiment: collection exports
all retained facts, and a receiver's already-delivered keys decide whether another
complete export is worth scheduling. `delta` reads each receiver's production
`createSyncSinceHlc` vector under its seeded clock and passes it to the sender's
`collectPendingChanges`. Vectors are scoped by receiver, space, and author.
- **It never drops permanently.** Dropping manufactures failures outside the
contract instead of finding real ones.
- **It never splits a collected batch.** An earlier version of this harness
split batches at a random pivot to vary framing. That delivers, for example, a
delete without its insert; the engine ignores the orphaned change, the harness
records it as delivered, and the fact is lost - surfacing as a bogus
convergence failure. Chunking in the real protocol sits *below* the merge
(`chunked()` emits frames that `collectNextBatch` reassembles until
`OfflineSyncEndOfBatch`), so the whole cycle is the causal unit and splitting here
models nothing real.
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.
Delay, reorder, and redelivery are the honest moves - and redelivery is how
idempotence gets probed.
Neither mode permanently drops facts or splits a collected batch. Wire chunking
belongs below this boundary: the protocol reassembles chunks through
`OfflineSyncEndOfBatch` before merging. Delta replay selects a previously merged
batch from the same source/target/space edge. Captured payloads are serialized and
decoded afresh for each delivery, so later local writes or input mutation cannot
change the replay. Full mode retains its original complete-export resends.
When the schedule ends, quiescence drains every pending batch and collects only
facts a receiver has not yet seen. It stops introducing deliberate duplicates,
so stable replicas cannot exhaust the convergence limit merely through resends.
Quiescence clears receive isolation, drains pending batches, and stops scheduling
deliberate replays. In delta mode it requires empty checkpoint-based collections
on every directed peer/space exchange. There is no full-history repair before
checking convergence or accepted authored facts. The existing export-to-empty
mirror check runs afterward and never repairs the simulated replicas.
```sh
DST_DELIVERY=delta DST_SEED_BASE=114 DST_SEEDS=1 DST_ROUNDS=200 DST_PROFILE=populated dart test -P dst
```
The modes need not author identical operations for an identical seed: delivery
changes visible state and subsequent generator choices. Fixed collection
regressions compare full and delta receivers of the same authored scenario.
## Layout
@ -197,16 +210,34 @@ The populated profile creates every declared table and FK edge, closes the
person/company/town cycles, and drives unique conflicts, a real tuple exchange,
restore/redelete, FK retarget/detach, and a constrained refusal. These are actual
ORM transactions, each checked immediately for structure, authored preservation,
and causal monotonicity. Complete exports distribute the populated spaces before
random scheduling starts. The sparse profile retains empty-world exploration.
and causal monotonicity. The selected delivery mode distributes the populated
spaces before random scheduling starts; bootstrap naturally includes previously
unknown facts. The sparse profile retains empty-world exploration.
Every run emits one `DST_METRICS` JSON record, including failed runs. It separates
Every run emits one `DST_METRICS` JSON record, including failed runs. Its
`topology` identifies `convergence` or `cross_space`. It separates
setup attempts from scheduled commits and records attempted, committed, rejected,
skipped, unexpected failures, and committed transactions whose oracle failed.
A run failing during setup reports zero scheduled activity. Paths retain table
and action; network observations include merge counts, duplicate batches, maximum
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.
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.
Counters carry `setup.`, `scheduled.`, or `drain.` prefixes according to when the
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,
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
appearances. FK edges and cycles are **authored graph** observations. Unique
projection coverage reports columns, not a claim that every overlapping index
@ -219,13 +250,12 @@ also require all declared authored FK edges and the mandatory semantic paths.
Setup commits alone cannot satisfy the scheduled-activity gate.
CI alternates sparse and populated profiles across four consecutive seeds at
200 rounds in both topologies: 4,800 scheduled operation attempts, versus the
previous 6,000 attempts spread across fifty shallow 20-round worlds. This keeps
comparable attempt volume while exploring ten times the history depth and
connected graphs. Each topology runs in its own sixty-minute CI job with four
ten-minute per-seed ceilings; a failure in one job does not cancel the other.
Both jobs also run the ownership-collision controls. The local `melos test-dst`
command runs both topologies together, so it covers the combined CI matrix.
200 rounds in both topologies and both delivery modes: four separate jobs, each
with 2,400 scheduled operation attempts and four ten-minute per-seed ceilings
within a sixty-minute job. A failure in one job does not cancel another.
All jobs also run the ownership-collision controls. The local `melos test-dst`
command runs both topologies for the selected `DST_DELIVERY` mode; run it once
with `full` and once with `delta` to cover both modes.
Known engine failures remain failures; CI is intentionally not a PR merge gate.
```sh
@ -234,5 +264,8 @@ DST_SEED_BASE=62 DST_SEEDS=1 DST_ROUNDS=200 DST_PROFILE=populated DST_GRAPH_WIDT
```
The oracle remains bounded: it does not independently arbitrate all unique
winners or FK fixed points. Collector concurrency/checkpoint lifecycle, crash
recovery, space grant/revoke, and transport framing remain outside this schedule.
winners or FK fixed points. Concurrent collection, continuous-session send cursors, crash recovery, space
grant/revoke, and transport framing remain outside this schedule. Delta mode
exercises production resume-vector creation, peer-handshake vector normalization,
checkpoint filtering, and the inbound merge primitive that atomically persists
per-author received progress; it does not simulate the full sync driver.

219
test/serverpod_offline_sync_test_server/test/dst/dst_adversary_test.dart
View file

@ -1,3 +1,4 @@
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';
@ -56,6 +57,195 @@ void main() {
}
},
);
group('Given a delta receiver isolated while a parent and child are authored,', () {
final ids = DstIds(DstRandom(917));
final space = ids.next();
final clock = DstClock();
late DstReplica source;
late DstReplica target;
late DstAdversary adversary;
late City parent;
late List<String> initialCheckpoints;
setUpAll(() async {
source = await DstReplica.create(
name: 'source',
spaceUuids: [space],
nodeUuid: ids.next(),
clock: clock.clock,
);
target = await DstReplica.create(
name: 'target',
spaceUuids: [space],
nodeUuid: ids.next(),
clock: clock.clock,
);
initialCheckpoints = (await target.checkpoints(
space,
)).map((hlc) => hlc.toString()).toList();
adversary = DstAdversary(_DelayedNewestFirstSchedule(), [
source,
target,
], delivery: DstDeliveryMode.delta);
parent = City(id: ids.next(), name: 'parent');
});
group('when complete batches queue and the newest one arrives first,', () {
late List<String> queuedCheckpoints;
late String expected;
final observed = <String>[];
late CrdtMergeSet caughtUp;
setUpAll(() async {
Future<void> observe(DstReplica replica) async {
if (identical(replica, target)) {
observed.add((await DstSnapshot.capture(replica)).renderSpace(space));
}
}
await source.withReplicaClock(
() => source.session.db.transactionForUser(space, (tx) async {
await City.db.insertRow(source.session, parent, transaction: tx);
}),
);
await adversary.step(observe);
await source.withReplicaClock(
() => source.session.db.transactionForUser(space, (tx) async {
await Town.db.insertRow(
source.session,
Town(id: ids.next(), name: 'child', cityId: parent.id),
transaction: tx,
);
}),
);
await adversary.step(observe);
queuedCheckpoints = (await target.checkpoints(
space,
)).map((hlc) => hlc.toString()).toList();
expected = (await DstSnapshot.capture(source)).renderSpace(space);
adversary.phase = DstNetworkPhase.drain;
await adversary.quiesce(observe);
caughtUp = await source.collect(
space,
checkpoints: await target.checkpoints(space),
);
});
test(
'then queuing advances no checkpoint and every delivered snapshot retains the child.',
() {
expect(queuedCheckpoints, initialCheckpoints);
expect(observed, hasLength(greaterThanOrEqualTo(3)));
expect(observed, everyElement(expected));
expect(caughtUp, isEmpty);
expect(adversary.metrics['scheduled.merges'], isNull);
expect(adversary.metrics['drain.merges'], greaterThanOrEqualTo(3));
},
);
});
});
group(
'Given a captured insertion batch and a delta adversary that requests replays,',
() {
final ids = DstIds(DstRandom(918));
final space = ids.next();
final clock = DstClock();
late DstReplica source;
late DstReplica target;
late City row;
late DstDelivery captured;
late CrdtMergeSet original;
late DstAdversary adversary;
setUpAll(() async {
source = await DstReplica.create(
name: 'source',
spaceUuids: [space],
nodeUuid: ids.next(),
clock: clock.clock,
);
target = await DstReplica.create(
name: 'target',
spaceUuids: [space],
nodeUuid: ids.next(),
clock: clock.clock,
);
row = City(id: ids.next(), name: 'original');
await source.withReplicaClock(
() => source.session.db.transactionForUser(space, (tx) async {
await City.db.insertRow(source.session, row, transaction: tx);
}),
);
original = await source.collect(space);
captured = DstDelivery(
source: source,
target: target,
spaceUuid: space,
changes: original,
);
adversary = DstAdversary(_RepeatedDeliverySchedule(), [
source,
target,
], delivery: DstDeliveryMode.delta)..phase = DstNetworkPhase.setup;
await adversary.quiesce((_) async {});
});
group('when the source changes and repeated partial deliveries drain,', () {
late String actual;
late String expected;
late String frozenName;
late String mutatedName;
late int partialBeforeDrain;
setUpAll(() async {
(original.single as CrdtMergeInsert).data = row.copyWith(
name: 'mutated input',
);
final decoded = captured.changes.single as CrdtMergeInsert
..data = row.copyWith(name: 'mutated decoded copy');
mutatedName = decoded.databaseColumns['name']! as String;
frozenName =
(captured.changes.single as CrdtMergeInsert).databaseColumns['name']!
as String;
await source.withReplicaClock(
() => source.session.db.transactionForUser(space, (tx) async {
await City.db.updateRow(
source.session,
row.copyWith(name: 'changed'),
columns: (t) => [t.name],
transaction: tx,
);
}),
);
adversary.phase = DstNetworkPhase.scheduled;
await adversary.step((_) async {});
await adversary.step((_) async {});
await adversary.step((_) async {});
partialBeforeDrain = adversary.scheduledPartialBatches;
adversary.phase = DstNetworkPhase.drain;
await adversary.quiesce((_) async {});
actual = (await DstSnapshot.capture(target)).renderSpace(space);
expected = (await DstSnapshot.capture(source)).renderSpace(space);
});
test(
'then the captured payload is immutable and scheduled partial merges survive exact replays.',
() {
expect(mutatedName, 'mutated decoded copy');
expect(frozenName, 'original');
expect(actual, expected);
expect(partialBeforeDrain, greaterThan(0));
expect(adversary.scheduledPartialBatches, partialBeforeDrain);
expect(adversary.metrics['explicitReplays'], greaterThan(0));
expect(adversary.metrics['scheduled.emptyCollections'], greaterThan(0));
expect(adversary.metrics['setup.partialBatches'], isNull);
},
);
});
},
);
}
/// A concrete network schedule: always collect the first replica, always
@ -73,3 +263,32 @@ class _RepeatedDeliverySchedule extends DstRandom {
@override
T pick<T>(List<T> items) => items.first;
}
/// Isolate the target at each step, collect from the source, and drain newest
/// batches first. Every queued collection must use committed receiver progress.
class _DelayedNewestFirstSchedule extends DstRandom {
_DelayedNewestFirstSchedule() : super(917);
bool _selectingPartition = false;
@override
bool chance(double probability) {
if (probability == DstAdversary.receiveIsolationProbability) {
_selectingPartition = true;
return true;
}
return probability == DstAdversary.collectionProbability;
}
@override
int between(int min, int max) => max;
@override
T pick<T>(List<T> items) {
if (_selectingPartition) {
_selectingPartition = false;
return items.last;
}
return items.first is DstDelivery ? items.last : items.first;
}
}

2
test/serverpod_offline_sync_test_server/test/dst/dst_convergence_test.dart
View file

@ -32,11 +32,13 @@ void main() {
rounds: config.rounds,
profile: config.profile,
graphWidth: config.graphWidth,
delivery: config.delivery,
run: () => runDstSimulation(
seed: seed,
rounds: config.rounds,
profile: config.profile,
graphWidth: config.graphWidth,
delivery: config.delivery,
topology: DstTopology.singleSpace,
),
);

2
test/serverpod_offline_sync_test_server/test/dst/dst_cross_space_test.dart
View file

@ -38,11 +38,13 @@ void main() {
rounds: config.rounds,
profile: config.profile,
graphWidth: config.graphWidth,
delivery: config.delivery,
run: () => runDstSimulation(
seed: seed,
rounds: config.rounds,
profile: config.profile,
graphWidth: config.graphWidth,
delivery: config.delivery,
topology: DstTopology.overlappingSpaces,
),
);

490
test/serverpod_offline_sync_test_server/test/dst/dst_delta_collection_test.dart Normal file
View file

@ -0,0 +1,490 @@
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_random.dart';
import 'framework/dst_snapshot.dart';
import 'framework/dst_world.dart';
void main() {
initTestClientSession(createSessionPerTest: false);
group('Given a city already acknowledged by a receiver,', () {
final ids = DstIds(DstRandom(420));
final clock = DstClock();
final space = ids.next();
late DstReplica source;
late DstReplica target;
late DstReplica reference;
late City city;
setUpAll(() async {
source = await _replica('source', [space], ids, clock);
target = await _replica('target', [space], ids, clock);
reference = await _replica('reference', [space], ids, clock);
city = City(id: ids.next(), name: 'initial');
await source.withReplicaClock(
() => source.session.db.transactionForUser(space, (tx) async {
await City.db.insertRow(source.session, city, transaction: tx);
}),
);
await target.merge(await source.collect(space), space);
await reference.merge(await source.collect(space), space);
});
group('when its name changes and later the city is deleted,', () {
late CrdtMergeSet update;
late CrdtMergeSet deletion;
late CrdtMergeSet caughtUp;
late String updatedTarget;
late String updatedReference;
late String deletedTarget;
late String deletedReference;
late String replayedTarget;
setUpAll(() async {
await source.withReplicaClock(
() => source.session.db.transactionForUser(space, (tx) async {
await City.db.updateRow(
source.session,
city.copyWith(name: 'changed'),
columns: (t) => [t.name],
transaction: tx,
);
}),
);
update = await source.collect(
space,
checkpoints: await target.checkpoints(space),
);
await target.merge(update, space);
await reference.merge(await source.collect(space), space);
updatedTarget = (await DstSnapshot.capture(target)).renderSpace(space);
updatedReference = (await DstSnapshot.capture(reference)).renderSpace(space);
await source.withReplicaClock(
() => source.session.db.transactionForUser(space, (tx) async {
await City.db.deleteRow(source.session, city, transaction: tx);
}),
);
deletion = await source.collect(
space,
checkpoints: await target.checkpoints(space),
);
await target.merge(deletion, space);
await reference.merge(await source.collect(space), space);
deletedTarget = (await DstSnapshot.capture(target)).renderSpace(space);
deletedReference = (await DstSnapshot.capture(reference)).renderSpace(space);
await target.merge(update, space);
await target.merge(deletion, space);
replayedTarget = (await DstSnapshot.capture(target)).renderSpace(space);
caughtUp = await source.collect(
space,
checkpoints: await target.checkpoints(space),
);
});
test('then the update batch omits the acknowledged insertion.', () {
expect(update, hasLength(1));
expect(update.single, isA<CrdtMergeUpdate>());
expect((update.single as CrdtMergeUpdate).value, 'changed');
expect(updatedTarget, updatedReference);
});
test('then the deletion batch contains only the new tombstone.', () {
expect(deletion, hasLength(1));
expect(deletion.single, isA<CrdtMergeDelete>());
expect(deletedTarget, deletedReference);
expect(deletedTarget, contains('HIDDEN'));
});
test('then replay changes neither the final state nor the empty catch-up.', () {
expect(replayedTarget, deletedTarget);
expect(caughtUp, isEmpty);
});
});
});
group('Given one author writing cities in two spaces,', () {
final ids = DstIds(DstRandom(421));
final clock = DstClock();
final firstSpace = ids.next();
final secondSpace = ids.next();
late DstReplica source;
late DstReplica target;
setUpAll(() async {
source = await _replica('source', [firstSpace, secondSpace], ids, clock);
target = await _replica('target', [firstSpace, secondSpace], ids, clock);
for (final space in [firstSpace, secondSpace]) {
await source.withReplicaClock(
() => source.session.db.transactionForUser(space, (tx) async {
await City.db.insertRow(
source.session,
City(id: ids.next(), name: space.toString()),
transaction: tx,
);
}),
);
}
});
group('when only the second space is acknowledged,', () {
late CrdtMergeSet firstPending;
late CrdtMergeSet secondPending;
setUpAll(() async {
await target.merge(await source.collect(secondSpace), secondSpace);
firstPending = await source.collect(
firstSpace,
checkpoints: await target.checkpoints(firstSpace),
);
secondPending = await source.collect(
secondSpace,
checkpoints: await target.checkpoints(secondSpace),
);
});
test('then the earlier first-space insertion is still pending.', () {
expect(firstPending, hasLength(1));
expect(firstPending.single.uuidSpaceId, firstSpace);
expect(secondPending, isEmpty);
});
});
});
group('Given a relay with a newer fact from one author,', () {
final ids = DstIds(DstRandom(422));
final clock = DstClock();
final space = ids.next();
late DstReplica older;
late DstReplica newer;
late DstReplica relay;
late DstReplica target;
late UuidValue olderRow;
setUpAll(() async {
older = await _replica('older', [space], ids, clock);
newer = await _replica('newer', [space], ids, clock);
relay = await _replica('relay', [space], ids, clock);
target = await _replica('target', [space], ids, clock);
olderRow = ids.next();
await older.withReplicaClock(
() => older.session.db.transactionForUser(space, (tx) async {
await City.db.insertRow(
older.session,
City(id: olderRow, name: 'older offline fact'),
transaction: tx,
);
}),
);
clock.advance(const Duration(seconds: 1));
await newer.withReplicaClock(
() => newer.session.db.transactionForUser(space, (tx) async {
await City.db.insertRow(
newer.session,
City(id: ids.next(), name: 'newer fact'),
transaction: tx,
);
}),
);
await relay.merge(await newer.collect(space), space);
await target.merge(await relay.collect(space), space);
});
group('when the older author reconnects through the relay,', () {
late CrdtMergeSet pending;
late String actual;
late String expected;
setUpAll(() async {
await relay.merge(await older.collect(space), space);
pending = await relay.collect(
space,
checkpoints: await target.checkpoints(space),
);
await target.merge(pending, space);
actual = (await DstSnapshot.capture(target)).renderSpace(space);
expected = (await DstSnapshot.capture(relay)).renderSpace(space);
});
test(
'then the older fact crosses both hops without resending the newer fact.',
() {
expect(pending, hasLength(1));
expect(pending.single.uuidRowId, olderRow);
expect(pending.single.uuidNodeId, older.nodeUuid);
expect(actual, expected);
},
);
});
});
group('Given a receiver that already holds a parent city,', () {
final ids = DstIds(DstRandom(423));
final clock = DstClock();
final space = ids.next();
late DstReplica source;
late DstReplica target;
late City parent;
setUpAll(() async {
source = await _replica('source', [space], ids, clock);
target = await _replica('target', [space], ids, clock);
parent = City(id: ids.next(), name: 'parent');
await source.withReplicaClock(
() => source.session.db.transactionForUser(space, (tx) async {
await City.db.insertRow(source.session, parent, transaction: tx);
}),
);
await target.merge(await source.collect(space), space);
});
group(
'when a child arrives alone followed by a new parent and child together,',
() {
late CrdtMergeSet childOnly;
late CrdtMergeSet together;
late DstSnapshot actual;
late DstSnapshot expected;
setUpAll(() async {
await source.withReplicaClock(
() => source.session.db.transactionForUser(space, (tx) async {
await Town.db.insertRow(
source.session,
Town(id: ids.next(), name: 'first child', cityId: parent.id),
transaction: tx,
);
}),
);
childOnly = await source.collect(
space,
checkpoints: await target.checkpoints(space),
);
await target.merge(childOnly, space);
await source.withReplicaClock(
() => source.session.db.transactionForUser(space, (tx) async {
final nextParent = await City.db.insertRow(
source.session,
City(id: ids.next(), name: 'second parent'),
transaction: tx,
);
await Town.db.insertRow(
source.session,
Town(id: ids.next(), name: 'second child', cityId: nextParent.id),
transaction: tx,
);
}),
);
together = await source.collect(
space,
checkpoints: await target.checkpoints(space),
);
await target.merge(together, space);
actual = await DstSnapshot.capture(target);
expected = await DstSnapshot.capture(source);
});
test('then each partial batch retains exactly the needed new facts.', () {
expect(childOnly, hasLength(1));
expect(childOnly.single.tableName, 'town');
expect(together, hasLength(2));
expect(
together.map((change) => change.tableName),
unorderedEquals(['city', 'town']),
);
expect(actual.renderSpace(space), expected.renderSpace(space));
expect(DstOracle.invariants(actual), isEmpty);
});
},
);
});
group('Given two disconnected replicas claiming the same unique name,', () {
final ids = DstIds(DstRandom(424));
final clock = DstClock();
final space = ids.next();
late DstReplica source;
late DstReplica other;
late DstReplica delta;
late DstReplica full;
late Unique row;
setUpAll(() async {
source = await _replica('source', [space], ids, clock);
other = await _replica('other', [space], ids, clock);
delta = await _replica('delta', [space], ids, clock);
full = await _replica('full', [space], ids, clock);
row = Unique(id: ids.next(), name: 'claim');
await source.withReplicaClock(
() => source.session.db.transactionForUser(space, (tx) async {
await Unique.db.insertRow(source.session, row, transaction: tx);
}),
);
clock.advance(const Duration(milliseconds: 1));
await other.withReplicaClock(
() => other.session.db.transactionForUser(space, (tx) async {
await Unique.db.insertRow(
other.session,
Unique(id: ids.next(), name: 'claim'),
transaction: tx,
);
}),
);
for (final receiver in [delta, full]) {
await receiver.merge(await source.collect(space), space);
await receiver.merge(await other.collect(space), space);
}
});
group('when one claim is deleted, restored, and deleted again through deltas,', () {
final batches = <CrdtMergeSet>[];
final actual = <String>[];
final expected = <String>[];
late DstSnapshot finalSnapshot;
setUpAll(() async {
// Each transition is a real local transaction with a complete collection
// boundary; only the delta receiver omits already acknowledged facts.
await source.withReplicaClock(
() => source.session.db.transactionForUser(space, (tx) async {
await Unique.db.deleteRow(source.session, row, transaction: tx);
}),
);
Future<void> deliver() async {
final batch = await source.collect(
space,
checkpoints: await delta.checkpoints(space),
);
batches.add(batch);
await delta.merge(batch, space);
await full.merge(await source.collect(space), space);
actual.add((await DstSnapshot.capture(delta)).renderSpace(space));
expected.add((await DstSnapshot.capture(full)).renderSpace(space));
}
await deliver();
final hidden = (await DstTable.unique.model.find(
source.session,
includeHidden: true,
spaceUuid: space,
)).single;
await source.withReplicaClock(
() => source.session.db.transactionForUser(space, (tx) async {
await source.session.db.insertRow(hidden, transaction: tx);
}),
);
await deliver();
await source.withReplicaClock(
() => source.session.db.transactionForUser(space, (tx) async {
await Unique.db.deleteRow(source.session, row, transaction: tx);
}),
);
await deliver();
finalSnapshot = await DstSnapshot.capture(delta);
});
test('then each projection agrees with the full-history receiver.', () {
expect(batches, hasLength(3));
expect(batches.first.single, isA<CrdtMergeDelete>());
expect(batches.last.single, isA<CrdtMergeDelete>());
expect(actual, expected);
expect(finalSnapshot.tombstones['unique/${row.id}']!.clFlag, 4);
expect(DstOracle.invariants(finalSnapshot), isEmpty);
});
});
});
group('Given a batch that collides with a row owned by another space,', () {
final ids = DstIds(DstRandom(425));
final clock = DstClock();
final owner = ids.next();
final incoming = ids.next();
final collisionId = ids.next();
final innocentId = ids.next();
late DstReplica source;
late DstReplica target;
late List<String> before;
late CrdtMergeSet batch;
setUpAll(() async {
source = await _replica('source', [incoming], ids, clock);
target = await _replica('target', [owner, incoming], ids, clock);
await target.withReplicaClock(
() => target.session.db.transactionForUser(owner, (tx) async {
await City.db.insertRow(
target.session,
City(id: collisionId, name: 'owner'),
transaction: tx,
);
}),
);
await source.withReplicaClock(
() => source.session.db.transactionForUser(incoming, (tx) async {
await City.db.insertRow(
source.session,
City(id: innocentId, name: 'must roll back'),
transaction: tx,
);
await City.db.insertRow(
source.session,
City(id: collisionId, name: 'collision'),
transaction: tx,
);
}),
);
final checkpoints = await target.checkpoints(incoming);
before = checkpoints.map((hlc) => hlc.toString()).toList();
batch = await source.collect(incoming, checkpoints: checkpoints);
});
group('when the receiver rejects the complete batch,', () {
late Object failure;
late List<String> after;
late CrdtMergeSet retry;
late City? rolledBack;
setUpAll(() async {
failure = await target
.merge(batch, incoming)
.then<Object>(
(_) => 'unexpected success',
onError: (Object error) => error,
);
after = (await target.checkpoints(
incoming,
)).map((hlc) => hlc.toString()).toList();
retry = await source.collect(
incoming,
checkpoints: await target.checkpoints(incoming),
);
rolledBack = await City.db.findById(target.session, innocentId);
});
test(
'then no received progress or partial row survives and the entire batch remains pending.',
() {
expect(batch, hasLength(2));
expect(failure, isA<OfflineSyncIntegrityViolationException>());
expect(after, before);
expect(rolledBack, isNull);
expect(retry.map(dstChangeKey), batch.map(dstChangeKey));
},
);
});
});
}
Future<DstReplica> _replica(
String name,
List<UuidValue> spaces,
DstIds ids,
DstClock clock,
) => DstReplica.create(
name: name,
spaceUuids: spaces,
nodeUuid: ids.next(),
clock: clock.clock,
);

297
test/serverpod_offline_sync_test_server/test/dst/dst_delta_unique_reclaim_test.dart Normal file
View file

@ -0,0 +1,297 @@
import 'package:serverpod_database/serverpod_database.dart'
show DatabaseUniqueViolationException;
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_random.dart';
import 'framework/dst_snapshot.dart';
import 'framework/dst_world.dart';
void main() {
initTestClientSession(createSessionPerTest: false);
group('Given two synchronized UUID claims with one projected conflict,', () {
late DstIds ids;
late DstClock clock;
late UuidValue space;
late UuidValue contested;
late UuidValue replacement;
late DstReplica source;
late DstReplica peer;
late UniqueUuid winner;
late UniqueUuid loser;
late DstSnapshot before;
setUp(() async {
ids = DstIds(DstRandom(427));
clock = DstClock();
space = ids.next();
contested = ids.next();
replacement = ids.next();
source = await DstReplica.create(
name: 'source',
spaceUuids: [space],
nodeUuid: ids.next(),
clock: clock.clock,
);
peer = await DstReplica.create(
name: 'peer',
spaceUuids: [space],
nodeUuid: ids.next(),
clock: clock.clock,
);
winner = UniqueUuid(id: ids.next(), value: contested);
loser = UniqueUuid(id: ids.next(), value: contested);
await source.withReplicaClock(
() => source.session.db.transactionForUser(space, (tx) async {
await UniqueUuid.db.insertRow(source.session, winner, transaction: tx);
}),
);
clock.advance(const Duration(milliseconds: 1));
await peer.withReplicaClock(
() => peer.session.db.transactionForUser(space, (tx) async {
await UniqueUuid.db.insertRow(peer.session, loser, transaction: tx);
}),
);
await source.merge(await peer.collect(space), space);
await peer.merge(await source.collect(space), space);
before = await DstSnapshot.capture(source);
});
group(
'when the winner locally changes to an uncontested UUID and sends only its delta,',
() {
late DstSnapshot local;
late DstSnapshot remote;
late CrdtMergeSet delta;
setUp(() async {
await source.withReplicaClock(
() => source.session.db.transactionForUser(space, (tx) async {
await UniqueUuid.db.updateRow(
source.session,
winner.copyWith(value: replacement),
columns: (t) => [t.value],
transaction: tx,
);
}),
);
local = await DstSnapshot.capture(source);
delta = await source.collect(
space,
checkpoints: await peer.checkpoints(space),
);
await peer.merge(delta, space);
remote = await DstSnapshot.capture(peer);
});
test('then the old loser reclaims its authored UUID on both replicas.', () {
final key = ('unique_uuid', loser.id!, 'value');
expect(before.projections[key], isNotNull);
expect(delta, hasLength(1));
expect(delta.single, isA<CrdtMergeUpdate>());
expect(local.projections[key], isNull);
expect(remote.projections[key], isNull);
expect(local.renderSpace(space), remote.renderSpace(space));
});
},
);
group('when the winner changes in a batch update without returned rows,', () {
late DstSnapshot local;
late DstSnapshot remote;
late CrdtMergeSet delta;
setUp(() async {
await source.withReplicaClock(
() => source.session.db.transactionForUser(space, (tx) async {
await UniqueUuid.db.update(
source.session,
[winner.copyWith(value: replacement)],
columns: (t) => [t.value],
noReturn: true,
transaction: tx,
);
}),
);
local = await DstSnapshot.capture(source);
delta = await source.collect(space, checkpoints: await peer.checkpoints(space));
await peer.merge(delta, space);
remote = await DstSnapshot.capture(peer);
});
test('then the released claimant reclaims its value before delta catch-up.', () {
final key = ('unique_uuid', loser.id!, 'value');
expect(before.projections[key], isNotNull);
expect(delta, hasLength(1));
expect(delta.single.uuidRowId, winner.id);
expect(local.projections[key], isNull);
expect(remote.projections[key], isNull);
expect(dstValue(local.authoredValue(key)), dstValue(before.authoredValue(key)));
expect(local.fieldHlc(key), before.fieldHlc(key));
expect(local.renderSpace(space), remote.renderSpace(space));
});
});
group('when a limited predicate update changes only the first claim,', () {
late DstSnapshot local;
late DstSnapshot remote;
late CrdtMergeSet delta;
setUp(() async {
await source.withReplicaClock(
() => source.session.db.transactionForUser(space, (tx) async {
await UniqueUuid.db.updateWhere(
source.session,
columnValues: (t) => [t.value(replacement)],
where: (t) => t.id.inSet(<UuidValue>{winner.id!, loser.id!}),
orderBy: (t) => t.id,
limit: 1,
noReturn: true,
transaction: tx,
);
}),
);
local = await DstSnapshot.capture(source);
delta = await source.collect(space, checkpoints: await peer.checkpoints(space));
await peer.merge(delta, space);
remote = await DstSnapshot.capture(peer);
});
test('then the released claimant reclaims its value before delta catch-up.', () {
final key = ('unique_uuid', loser.id!, 'value');
expect(before.projections[key], isNotNull);
expect(delta, hasLength(1));
expect(delta.single.uuidRowId, winner.id);
expect(local.projections[key], isNull);
expect(remote.projections[key], isNull);
expect(dstValue(local.authoredValue(key)), dstValue(before.authoredValue(key)));
expect(local.fieldHlc(key), before.fieldHlc(key));
expect(local.renderSpace(space), remote.renderSpace(space));
});
});
group('when a predicate update skips the winner and selects the next claim,', () {
late List<UniqueUuid> updated;
late DstSnapshot local;
late DstSnapshot remote;
late CrdtMergeSet delta;
setUp(() async {
updated = await source.withReplicaClock(
() => source.session.db.transactionForUser(space, (tx) async {
return UniqueUuid.db.updateWhere(
source.session,
columnValues: (t) => [t.value(replacement)],
where: (t) => t.id.inSet(<UuidValue>{winner.id!, loser.id!}),
orderBy: (t) => t.id,
offset: 1,
limit: 1,
transaction: tx,
);
}),
);
local = await DstSnapshot.capture(source);
delta = await source.collect(space, checkpoints: await peer.checkpoints(space));
await peer.merge(delta, space);
remote = await DstSnapshot.capture(peer);
});
test('then exactly the selected loser changes and is returned.', () {
final loserKey = ('unique_uuid', loser.id!, 'value');
final winnerKey = ('unique_uuid', winner.id!, 'value');
expect(before.projections[loserKey], isNotNull);
expect(updated, hasLength(1));
expect(updated.single.id, loser.id);
expect(updated.single.value, replacement);
expect(delta, hasLength(1));
expect(delta.single.uuidRowId, loser.id);
expect(local.projections[loserKey], isNull);
expect(dstValue(local.authoredValue(winnerKey)), dstValue(contested));
expect(local.fieldHlc(winnerKey), before.fieldHlc(winnerKey));
expect(local.renderSpace(space), remote.renderSpace(space));
});
});
group('when an upsert changes the winning claim,', () {
late DstSnapshot local;
late DstSnapshot remote;
late CrdtMergeSet delta;
setUp(() async {
await source.withReplicaClock(
() => source.session.db.transactionForUser(space, (tx) async {
await UniqueUuid.db.upsertRow(
source.session,
winner.copyWith(value: replacement),
conflictColumns: (t) => [t.id],
updateColumns: (t) => [t.value],
transaction: tx,
);
}),
);
local = await DstSnapshot.capture(source);
delta = await source.collect(space, checkpoints: await peer.checkpoints(space));
await peer.merge(delta, space);
remote = await DstSnapshot.capture(peer);
});
test('then the released claimant reclaims its value before delta catch-up.', () {
final key = ('unique_uuid', loser.id!, 'value');
expect(before.projections[key], isNotNull);
expect(delta, hasLength(1));
expect(delta.single.uuidRowId, winner.id);
expect(local.projections[key], isNull);
expect(remote.projections[key], isNull);
expect(dstValue(local.authoredValue(key)), dstValue(before.authoredValue(key)));
expect(local.fieldHlc(key), before.fieldHlc(key));
expect(local.renderSpace(space), remote.renderSpace(space));
});
});
group('when a predicate update fails its physical unique constraint,', () {
late Object failure;
late DstSnapshot after;
late List<String> checkpointBefore;
late List<String> checkpointAfter;
setUp(() async {
checkpointBefore = (await source.checkpoints(
space,
)).map((hlc) => hlc.toString()).toList();
failure = await source
.withReplicaClock(
() => source.session.db.transactionForUser(space, (tx) async {
await UniqueUuid.db.updateWhere(
source.session,
columnValues: (t) => [t.value(replacement)],
where: (t) => t.id.inSet(<UuidValue>{winner.id!, loser.id!}),
transaction: tx,
);
}),
)
.then<Object>(
(_) => 'unexpected success',
onError: (Object error) => error,
);
after = await DstSnapshot.capture(source);
checkpointAfter = (await source.checkpoints(
space,
)).map((hlc) => hlc.toString()).toList();
});
test(
'then pre-write closure capture changes no domain, authored metadata, or progress.',
() {
expect(before.projections[('unique_uuid', loser.id!, 'value')], isNotNull);
expect(failure, isA<DatabaseUniqueViolationException>());
expect(after.renderSpace(space), before.renderSpace(space));
expect(after.renderRawMetadata(), before.renderRawMetadata());
expect(checkpointAfter, checkpointBefore);
},
);
});
});
}

84
test/serverpod_offline_sync_test_server/test/dst/dst_local_checkpoint_test.dart Normal file
View file

@ -0,0 +1,84 @@
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_random.dart';
import 'framework/dst_world.dart';
void main() {
initTestClientSession(createSessionPerTest: false);
group('Given two local writes committed within the same clock millisecond,', () {
final ids = DstIds(DstRandom(426));
final clock = DstClock();
final space = ids.next();
late DstReplica author;
late DstReplica peer;
late Hlc latestAuthored;
setUpAll(() async {
author = await DstReplica.create(
name: 'author',
spaceUuids: [space],
nodeUuid: ids.next(),
clock: clock.clock,
);
peer = await DstReplica.create(
name: 'peer',
spaceUuids: [space],
nodeUuid: ids.next(),
clock: clock.clock,
);
for (final name in ['first', 'second']) {
await author.withReplicaClock(
() => author.session.db.transactionForUser(space, (tx) async {
await City.db.insertRow(
author.session,
City(id: ids.next(), name: name),
transaction: tx,
);
}),
);
}
final changes = await author.collect(space);
latestAuthored = changes.maxHlc!;
await peer.merge(changes, space);
});
group(
'when the author reconnects before its wall clock advances and after it moves backward,',
() {
late Hlc sameMillisecond;
late Hlc backwardClock;
late CrdtMergeSet sameMillisecondEcho;
late CrdtMergeSet backwardClockEcho;
setUpAll(() async {
var checkpoints = await author.checkpoints(space);
sameMillisecond = checkpoints.singleWhere(
(hlc) => hlc.nodeId == author.nodeUuid,
);
sameMillisecondEcho = await peer.collect(space, checkpoints: checkpoints);
clock.advance(const Duration(seconds: -1));
checkpoints = await author.checkpoints(space);
backwardClock = checkpoints.singleWhere(
(hlc) => hlc.nodeId == author.nodeUuid,
);
backwardClockEcho = await peer.collect(space, checkpoints: checkpoints);
});
test('then both resume vectors cover the persisted logical counter.', () {
expect(latestAuthored.counter, greaterThan(0));
expect(sameMillisecond, greaterThanOrEqualTo(latestAuthored));
expect(backwardClock, greaterThanOrEqualTo(latestAuthored));
});
test("then neither resume causes the peer to echo the author's own facts.", () {
expect(sameMillisecondEcho, isEmpty);
expect(backwardClockEcho, isEmpty);
});
},
);
});
}

204
test/serverpod_offline_sync_test_server/test/dst/dst_relay_checkpoint_test.dart Normal file
View file

@ -0,0 +1,204 @@
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_random.dart';
import 'framework/dst_world.dart';
void main() {
initTestClientSession(createSessionPerTest: false);
group("Given a relay with its own older fact and another author's newer fact,", () {
late DstIds ids;
late DstClock clock;
late UuidValue space;
late DstReplica author;
late DstReplica relay;
late DstReplica receiver;
late CrdtMergeSet initial;
Future<void> insert(DstReplica replica, String name) => replica.withReplicaClock(
() => replica.session.db.transactionForUser(space, (tx) async {
await City.db.insertRow(
replica.session,
City(id: ids.next(), name: name),
transaction: tx,
);
}),
);
setUp(() async {
ids = DstIds(DstRandom(990));
clock = DstClock();
space = ids.next();
author = await DstReplica.create(
name: 'author',
spaceUuids: [space],
nodeUuid: ids.next(),
clock: clock.clock,
);
relay = await DstReplica.create(
name: 'relay',
spaceUuids: [space],
nodeUuid: ids.next(),
clock: clock.clock,
);
receiver = await DstReplica.create(
name: 'receiver',
spaceUuids: [space],
nodeUuid: ids.next(),
clock: clock.clock,
);
await insert(relay, 'relay fact');
clock.advance(const Duration(seconds: 1));
await insert(author, 'author fact');
await relay.merge(await author.collect(space), space);
initial = await relay.collect(space);
await receiver.merge(initial, space);
});
group('when the receiver reconnects after the production inbound merge,', () {
late List<Hlc> vector;
late CrdtMergeSet pending;
setUp(() async {
vector = await receiver.checkpoints(space);
pending = await relay.collect(space, checkpoints: vector);
});
test(
'then each author has its own checkpoint and the relay has nothing to resend.',
() {
expect(initial, hasLength(2));
expect(initial.maxHlc!.nodeId, author.nodeUuid);
expect(vector.map((hlc) => hlc.nodeId).toSet(), hasLength(3));
expect(vector, hasLength(3));
expect(
vector.singleWhere((hlc) => hlc.nodeId == relay.nodeUuid),
initial.singleWhere((change) => change.uuidNodeId == relay.nodeUuid).hlc,
);
expect(pending, isEmpty);
},
);
});
group('when a stale session echoes a receiver fact before another local write,', () {
late CrdtMergeSet echo;
late CrdtMergeSet pending;
late List<Hlc> vector;
setUp(() async {
final stale = await receiver.checkpoints(space);
clock.advance(const Duration(seconds: 1));
await insert(receiver, 'receiver fact one');
await relay.merge(
await receiver.collect(space, checkpoints: await relay.checkpoints(space)),
space,
);
echo = await relay.collect(space, checkpoints: stale);
await receiver.merge(echo, space);
clock.advance(const Duration(seconds: 1));
await insert(receiver, 'receiver fact two');
await relay.merge(
await receiver.collect(space, checkpoints: await relay.checkpoints(space)),
space,
);
vector = await receiver.checkpoints(space);
pending = await relay.collect(space, checkpoints: vector);
});
test(
'then reconnecting neither resends the relay fact nor echoes either local fact.',
() {
expect(echo, hasLength(1));
expect(echo.single.uuidNodeId, receiver.nodeUuid);
expect(vector, hasLength(3));
expect(vector.map((hlc) => hlc.nodeId).toSet(), hasLength(3));
expect(pending, isEmpty);
},
);
});
group("when the checkpoint API is given a relayed author under the peer's id,", () {
late Object failure;
late List<Hlc> before;
late List<Hlc> after;
setUp(() async {
before = await receiver.checkpoints(space);
failure = await receiver.session.db
.recordSyncCheckpoint(relay.nodeUuid, initial.maxHlc!, userId: space)
.then<Object>(
(_) => 'unexpected success',
onError: (Object error) => error,
);
after = await receiver.checkpoints(space);
});
test('then the invalid checkpoint is rejected without changing progress.', () {
expect(initial.maxHlc!.nodeId, author.nodeUuid);
expect(failure, isA<ArgumentError>());
expect(after, before);
});
});
group(
'when a stored relay checkpoint carries the wrong author from older bookkeeping,',
() {
late List<Hlc> beforeRepair;
late List<Hlc> afterRepair;
late CrdtMergeSet repair;
late CrdtMergeSet pending;
late Hlc? persisted;
setUp(() async {
final nodes = await OfflineSyncSpaceNode.db.find(
receiver.rawSession,
include: OfflineSyncSpaceNode.include(node: CrdtNode.include()),
);
final relayNode = nodes.singleWhere(
(node) => node.node!.uuidNodeId == relay.nodeUuid,
);
await OfflineSyncSpaceNode.db.updateRow(
receiver.rawSession,
relayNode.copyWith(lastReceivedHlc: initial.maxHlc),
columns: (t) => [t.lastReceivedHlc],
);
beforeRepair = await receiver.checkpoints(space);
repair = await relay.collect(space, checkpoints: beforeRepair);
await receiver.merge(repair, space);
afterRepair = await receiver.checkpoints(space);
persisted = (await OfflineSyncSpaceNode.db.findById(
receiver.rawSession,
relayNode.id!,
))!.lastReceivedHlc;
pending = await relay.collect(space, checkpoints: afterRepair);
});
test(
'then one safe retransmission repairs the persisted author tag and settles.',
() {
final relayHlc = initial
.singleWhere((change) => change.uuidNodeId == relay.nodeUuid)
.hlc;
expect(initial.maxHlc, greaterThan(relayHlc));
expect(beforeRepair, hasLength(3));
expect(
beforeRepair.singleWhere((hlc) => hlc.nodeId == relay.nodeUuid),
Hlc.zero(relay.nodeUuid),
);
expect(repair, hasLength(1));
expect(repair.single.uuidNodeId, relay.nodeUuid);
expect(persisted, relayHlc);
expect(
afterRepair.singleWhere((hlc) => hlc.nodeId == relay.nodeUuid),
relayHlc,
);
expect(pending, isEmpty);
},
);
},
);
});
}

9
test/serverpod_offline_sync_test_server/test/dst/dst_runner_test.dart
View file

@ -36,13 +36,14 @@ void main() {
test(
'Given a property violation from seed 114 at 80 rounds, '
'when its failure is reported, '
'then the replay command preserves the seed, depth, profile, and graph width.',
'then the replay command preserves the seed, depth, profile, graph width, and delivery mode.',
() {
final failure = DstPropertyFailure(
seed: 114,
rounds: 80,
profile: DstProfile.populated,
graphWidth: 3,
delivery: DstDeliveryMode.delta,
violations: [(property: 'foreignKeyClosure', detail: 'orphan')],
);
@ -51,7 +52,7 @@ void main() {
expect(
message,
'DST property failure (seed 114)\n'
'Replay: DST_SEED_BASE=114 DST_SEEDS=1 DST_ROUNDS=80 DST_PROFILE=populated DST_GRAPH_WIDTH=3 dart test -P dst\n'
'Replay: DST_SEED_BASE=114 DST_SEEDS=1 DST_ROUNDS=80 DST_PROFILE=populated DST_GRAPH_WIDTH=3 DST_DELIVERY=delta dart test -P dst\n'
'- foreignKeyClosure: orphan\n',
);
},
@ -60,7 +61,7 @@ void main() {
test(
'Given an unexpected exception from seed 115 at 40 rounds, '
'when its failure is reported, '
'then the replay command preserves the seed, depth, profile, and graph width.',
'then the replay command preserves the seed, depth, profile, graph width, and delivery mode.',
() async {
final failure = await runWithSeedReported<Object>(
index: 1,
@ -74,7 +75,7 @@ void main() {
expect(
message,
'Bad state: Simulation 1 (seed 115) failed\n'
'Replay: DST_SEED_BASE=115 DST_SEEDS=1 DST_ROUNDS=40 DST_PROFILE=sparse DST_GRAPH_WIDTH=2 dart test -P dst\n'
'Replay: DST_SEED_BASE=115 DST_SEEDS=1 DST_ROUNDS=40 DST_PROFILE=sparse DST_GRAPH_WIDTH=2 DST_DELIVERY=full dart test -P dst\n'
'Bad state: unexpected',
);
},

200
test/serverpod_offline_sync_test_server/test/dst/framework/dst_adversary.dart
View file

@ -1,3 +1,4 @@
import 'dart:convert';
import 'dart:io';
import 'package:serverpod/serverpod.dart';
@ -6,48 +7,64 @@ import 'package:serverpod_offline_sync_server/serverpod_offline_sync_server.dart
import 'dst_random.dart';
import 'dst_world.dart';
/// One batch of changes waiting to be merged into a replica.
/// The phase in which a network action actually takes place.
enum DstNetworkPhase { setup, scheduled, drain }
/// An immutable captured batch, including the committed receiver baseline.
///
/// Generated changes and their domain payloads are mutable. Store the wire
/// representation and decode a fresh copy for each merge, so even a merge that
/// mutates its input cannot change a later replay.
class DstDelivery {
/// Creates a pending delivery.
DstDelivery({
required this.source,
required this.target,
required this.spaceUuid,
required this.changes,
});
required CrdtMergeSet changes,
List<Hlc> checkpoints = const [],
this.partial = false,
this.replay = false,
}) : payload = jsonEncode(changes),
checkpoints = List.unmodifiable(checkpoints);
/// The replica that will merge the batch.
DstDelivery._replay(DstDelivery original)
: source = original.source,
target = original.target,
spaceUuid = original.spaceUuid,
payload = original.payload,
checkpoints = original.checkpoints,
partial = original.partial,
replay = true;
final DstReplica source;
final DstReplica target;
/// The space the batch belongs to.
final UuidValue spaceUuid;
final String payload;
final List<Hlc> checkpoints;
final bool partial;
final bool replay;
/// The changes to merge.
final CrdtMergeSet changes;
CrdtMergeSet get changes => [
for (final value in jsonDecode(payload) as List)
target.rawSession.db.serializationManager.deserialize<CrdtMergeChange>(value),
];
}
/// Schedules change delivery between replicas adversarially.
/// Delays, reorders, redelivers and isolates complete collected batches.
///
/// The adversary reorders, delays, redelivers, and partitions. Two moves are
/// deliberately *not* available, both for the same reason: the merge contract
/// states that each batch arrives as a causally complete snapshot of the
/// sender (`docs/foreign-key-invariants.md`).
///
/// - **Never drops permanently.** Dropping would manufacture failures outside
/// the contract instead of finding real ones.
/// - **Never splits a collected batch.** An earlier version of this harness
/// split batches at a random pivot to vary framing. That delivers, say, a
/// delete without its insert; the engine ignores the orphaned change, the
/// harness marks it delivered, and the fact is lost forever - which then
/// surfaces as a bogus convergence failure. Chunking in the real protocol
/// sits *below* the merge (`chunked()` feeds frames that
/// `collectNextBatch` reassembles until `OfflineSyncEndOfBatch`), so the whole
/// cycle is the causal unit and splitting here models nothing real.
///
/// Delay, reorder, and redelivery are the honest moves; redelivery is how
/// idempotence gets probed.
/// Full mode retains the original convergence experiment. Delta mode always
/// collects against facts already committed at the receiver, never against a
/// pending delivery. Thus pending batches may overlap but do not depend on one
/// another arriving first. Neither mode splits batches or drops them forever.
class DstAdversary {
/// Creates an adversary over [replicas].
DstAdversary(this.random, this.replicas);
DstAdversary(
this.random,
this.replicas, {
this.delivery = DstDeliveryMode.full,
});
final DstDeliveryMode delivery;
DstNetworkPhase phase = DstNetworkPhase.scheduled;
static const receiveIsolationProbability = 0.2;
static const collectionProbability = 0.8;
@ -62,8 +79,22 @@ class DstAdversary {
final List<DstDelivery> _pending = [];
final Map<String, Set<String>> _deliveredKeys = {};
final Map<String, int> _partitionedUntil = {};
final Map<String, List<DstDelivery>> _replayable = {};
final Map<String, int> _observations = {};
var _round = 0;
void _count(String name, [int amount = 1]) {
_observations
..update(name, (value) => value + amount, ifAbsent: () => amount)
..update(
'${phase.name}.$name',
(value) => value + amount,
ifAbsent: () => amount,
);
}
int get scheduledPartialBatches => _observations['scheduled.partialBatches'] ?? 0;
/// Batches merged so far, for reporting how much work a seed actually did.
int mergeCount = 0;
int receiveIsolationEvents = 0;
@ -72,6 +103,7 @@ class DstAdversary {
int deliveredChanges = 0;
Map<String, int> get metrics => {
..._observations,
'merges': mergeCount,
'receiveIsolationEvents': receiveIsolationEvents,
'duplicateBatches': duplicateBatches,
@ -117,13 +149,14 @@ class DstAdversary {
throw StateError(
'The simulation did not quiesce after $_maxQuiescePasses passes. '
'Replicas keep producing changes for each other, which means merging a '
'batch is not idempotent: a merge is re-authoring facts instead of '
'absorbing them.',
'batch is not idempotent or checkpoint collection cannot settle. '
'Delivery mode: ${delivery.name}; metrics: $metrics',
);
}
void _partitionRandomReplica() {
receiveIsolationEvents++;
_count('receiveIsolationEvents');
final replica = random.pick(replicas);
_partitionedUntil[replica.name] = _round + random.between(1, 3);
}
@ -146,12 +179,18 @@ class DstAdversary {
bool allowResend = true,
}) async {
final changes = await source.collect(spaceUuid);
if (changes.isEmpty) return;
if (changes.isEmpty && delivery == DstDeliveryMode.full) 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
// a no-op, so this is the idempotence probe rather than wasted work.
@ -167,11 +206,59 @@ class DstAdversary {
if (fresh.isEmpty) continue;
_pending.add(
DstDelivery(target: target, spaceUuid: spaceUuid, changes: changes),
DstDelivery(
source: source,
target: target,
spaceUuid: spaceUuid,
changes: changes,
),
);
}
}
Future<void> _collectDelta(
DstReplica source,
DstReplica target,
UuidValue spaceUuid,
int fullSize,
bool allowResend,
) async {
final edge = '${source.name}|${target.name}|$spaceUuid';
final history = _replayable[edge] ?? const <DstDelivery>[];
if (allowResend && random.chance(resendProbability) && history.isNotEmpty) {
_pending.add(DstDelivery._replay(random.pick(history)));
}
final checkpoints = await target.checkpoints(spaceUuid);
_count('checkpointCollections');
final advanced = checkpoints.any(
(checkpoint) =>
checkpoint.nodeId != target.nodeUuid &&
checkpoint > Hlc.zero(checkpoint.nodeId),
);
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);
if (changes.isEmpty) {
_count('emptyCollections');
return;
}
_pending.add(
DstDelivery(
source: source,
target: target,
spaceUuid: spaceUuid,
changes: changes,
checkpoints: checkpoints,
partial: advanced && changes.length < fullSize,
),
);
}
Future<void> _deliverOne(Future<void> Function(DstReplica) onMerged) async {
final deliverable = [
for (final delivery in _pending)
@ -184,30 +271,53 @@ class DstAdversary {
final delivery = random.pick(deliverable);
_pending.remove(delivery);
final changes = delivery.changes;
try {
await delivery.target.merge(delivery.changes, delivery.spaceUuid);
await delivery.target.merge(changes, delivery.spaceUuid);
} on Exception catch (exception) {
// As with local operations, database errors arrive without engine
// frames, so the batch that caused them is described here.
final keys = delivery.changes.map(dstChangeKey).join('\n ');
final keys = changes.map(dstChangeKey).join('\n ');
throw StateError(
'Merging ${delivery.changes.length} changes for space '
'${delivery.spaceUuid} into ${delivery.target} failed: $exception\n'
'Merging ${changes.length} changes for space '
'${delivery.spaceUuid} from ${delivery.source} into ${delivery.target} failed: $exception\n'
'Mode: ${this.delivery.name}; checkpoints: ${delivery.checkpoints}; replay: ${delivery.replay}\n'
'Batch:\n $keys',
);
}
final prior = _deliveredKeys[delivery.target.name] ?? {};
if (delivery.changes.any((change) => prior.contains(dstChangeKey(change)))) {
if (changes.any((change) => prior.contains(dstChangeKey(change)))) {
duplicateBatches++;
_count('duplicateBatches');
}
if (delivery.changes.length > maxBatchSize) maxBatchSize = delivery.changes.length;
deliveredChanges += delivery.changes.length;
if (changes.length > maxBatchSize) maxBatchSize = changes.length;
final phaseMaxKey = '${phase.name}.maxBatchSize';
_observations.update(
phaseMaxKey,
(value) => value > changes.length ? value : changes.length,
ifAbsent: () => changes.length,
);
deliveredChanges += changes.length;
_count('deliveredChanges', changes.length);
mergeCount++;
_count('merges');
if (delivery.replay) _count('explicitReplays');
if (delivery.partial && !delivery.replay) _count('partialBatches');
final repeated = changes
.where((change) => prior.contains(dstChangeKey(change)))
.length;
_count('repeatedChanges', repeated);
_count('freshChanges', changes.length - repeated);
if (this.delivery == DstDeliveryMode.delta && !delivery.replay) {
final edge =
'${delivery.source.name}|${delivery.target.name}|${delivery.spaceUuid}';
_replayable.putIfAbsent(edge, () => []).add(delivery);
}
_trace(delivery);
_deliveredKeys
.putIfAbsent(delivery.target.name, () => {})
.addAll(delivery.changes.map(dstChangeKey));
.addAll(changes.map(dstChangeKey));
await onMerged(delivery.target);
}
@ -233,7 +343,11 @@ class DstAdversary {
// Printing is the point: this is an opt-in trace read from the test runner
// output while diagnosing a failing seed.
// ignore: avoid_print
print('merge -> ${delivery.target}: ${relevant.join(' || ')}');
print(
'merge ${delivery.source} -> ${delivery.target} '
'mode=${this.delivery.name} checkpoints=${delivery.checkpoints} '
'replay=${delivery.replay}: ${relevant.join(' || ')}',
);
}
static const _maxQuiescePasses = 24;

14
test/serverpod_offline_sync_test_server/test/dst/framework/dst_random.dart
View file

@ -5,6 +5,15 @@ import 'dart:typed_data';
import 'package:clock/clock.dart';
import 'package:serverpod/serverpod.dart';
/// The causal facts supplied by each simulated collection.
enum DstDeliveryMode {
/// Preserve the original full-history merge convergence experiment.
full,
/// Collect against the receiver's committed production checkpoint vector.
delta,
}
/// Seeded randomness for the deterministic simulation harness.
///
/// Every choice a simulation makes - which replica acts, which operation it
@ -144,6 +153,7 @@ class DstConfig {
required this.seedBase,
this.profile = DstProfile.sparse,
this.graphWidth = 2,
this.delivery = DstDeliveryMode.full,
});
/// Reads the sweep configuration from the environment.
@ -156,9 +166,13 @@ class DstConfig {
Platform.environment['DST_PROFILE'] ?? 'sparse',
),
graphWidth: _readInt('DST_GRAPH_WIDTH', 2),
delivery: DstDeliveryMode.values.byName(
Platform.environment['DST_DELIVERY'] ?? 'full',
),
);
}
final DstDeliveryMode delivery;
final DstProfile profile;
final int graphWidth;

45
test/serverpod_offline_sync_test_server/test/dst/framework/dst_runner.dart
View file

@ -11,7 +11,11 @@ import 'dst_world.dart';
/// The shape of one simulated deployment.
class DstTopology {
/// Creates a topology.
const DstTopology({required this.spaceCount, required this.subscriptions});
const DstTopology({
required this.name,
required this.spaceCount,
required this.subscriptions,
});
/// Two spaces, and three replicas whose subscription sets deliberately
/// differ: two hold only one space each, one holds both.
@ -20,6 +24,7 @@ class DstTopology {
/// some replica holds strictly more than another, because the claim is that
/// the extra space changes nothing about the shared one.
static const overlappingSpaces = DstTopology(
name: 'cross_space',
spaceCount: 2,
subscriptions: [
[0],
@ -30,6 +35,7 @@ class DstTopology {
/// One space shared by every replica - the classic convergence shape.
static const singleSpace = DstTopology(
name: 'convergence',
spaceCount: 1,
subscriptions: [
[0],
@ -38,6 +44,9 @@ class DstTopology {
],
);
/// The stable topology name used in reports and soak records.
final String name;
/// How many spaces exist.
final int spaceCount;
@ -54,6 +63,7 @@ class DstRunReport {
/// Creates a report.
DstRunReport({
required this.seed,
required this.topology,
required this.merges,
required this.applied,
required this.rejected,
@ -64,6 +74,7 @@ class DstRunReport {
required this.rounds,
required this.profile,
required this.graphWidth,
required this.delivery,
required this.skipped,
required this.setupAttempted,
required this.scheduledCommitted,
@ -71,7 +82,9 @@ class DstRunReport {
required this.network,
});
final String topology;
final int rounds;
final DstDeliveryMode delivery;
final DstProfile profile;
final int graphWidth;
final int skipped;
@ -84,8 +97,10 @@ class DstRunReport {
Map<String, Object> toJson() => {
'seed': seed,
'topology': topology,
'rounds': rounds,
'profile': profile.name,
'delivery': delivery.name,
'graphWidth': graphWidth,
'attempted': attempted,
'committed': applied,
@ -124,7 +139,7 @@ class DstRunReport {
@override
String toString() =>
'seed=$seed merges=$merges applied=$applied rejected=$rejected '
'seed=$seed topology=$topology merges=$merges applied=$applied rejected=$rejected '
'visible=$visibleRows hidden=$hiddenRows paths=$appliedPaths';
}
@ -139,6 +154,7 @@ Future<DstRunReport> runDstSimulation({
DstTopology topology = DstTopology.overlappingSpaces,
DstProfile profile = DstProfile.sparse,
int graphWidth = 2,
DstDeliveryMode delivery = DstDeliveryMode.full,
}) async {
if (rounds < 1 || graphWidth < 2) {
throw ArgumentError('rounds >= 1 and graphWidth >= 2 required');
@ -180,7 +196,8 @@ Future<DstRunReport> runDstSimulation({
final operations = DstOperations(random, ids);
operations.oracle.accept(await DstSnapshot.capture(replicas.first));
final adversary = DstAdversary(random, replicas);
final adversary = DstAdversary(random, replicas, delivery: delivery)
..phase = DstNetworkPhase.setup;
var schedulingStarted = false;
var setupAttempted = 0;
var setupCommitted = 0;
@ -194,6 +211,7 @@ Future<DstRunReport> runDstSimulation({
rounds: rounds,
profile: profile,
graphWidth: graphWidth,
delivery: delivery,
replica: replica,
violations: violations,
);
@ -220,6 +238,7 @@ Future<DstRunReport> runDstSimulation({
setupAttempted = operations.attempted;
setupCommitted = operations.committed;
schedulingStarted = true;
adversary.phase = DstNetworkPhase.scheduled;
for (var round = 0; round < rounds; round++) {
for (final replica in replicas) {
@ -231,6 +250,7 @@ Future<DstRunReport> runDstSimulation({
await adversary.step(checkInvariants);
}
adversary.phase = DstNetworkPhase.drain;
await adversary.quiesce(checkInvariants);
final snapshots = <DstReplica, DstSnapshot>{
@ -266,17 +286,20 @@ Future<DstRunReport> runDstSimulation({
rounds: rounds,
profile: profile,
graphWidth: graphWidth,
delivery: delivery,
violations: violations,
);
}
final report = DstRunReport(
topology: topology.name,
seed: seed,
merges: adversary.mergeCount,
applied: operations.committed,
rounds: rounds,
profile: resolvedProfile,
graphWidth: graphWidth,
delivery: delivery,
skipped: operations.skipped,
setupAttempted: setupAttempted,
scheduledCommitted: operations.committed - setupCommitted,
@ -300,6 +323,13 @@ Future<DstRunReport> runDstSimulation({
'Insufficient scheduled activity: required $requiredCommits commits; ${report.toJson()}',
);
}
if (delivery == DstDeliveryMode.delta &&
rounds >= 100 &&
adversary.scheduledPartialBatches == 0) {
throw StateError(
'Delta stress run delivered no scheduled partial batches: ${report.toJson()}',
);
}
if (resolvedProfile == DstProfile.populated) {
final transitions = operations.coverage.transitions;
final missing = [
@ -328,9 +358,11 @@ Future<DstRunReport> runDstSimulation({
print(
'DST_METRICS ${jsonEncode({
'seed': seed,
'topology': topology.name,
'rounds': rounds,
'profile': resolvedProfile.name,
'requestedProfile': profile.name,
'delivery': delivery.name,
'graphWidth': graphWidth,
'attempted': operations.attempted,
'committed': operations.committed,
@ -365,6 +397,7 @@ Future<T> runWithSeedReported<T>({
required Future<T> Function() run,
DstProfile profile = DstProfile.sparse,
int graphWidth = 2,
DstDeliveryMode delivery = DstDeliveryMode.full,
}) async {
try {
return await run();
@ -374,7 +407,7 @@ Future<T> runWithSeedReported<T>({
Error.throwWithStackTrace(
StateError(
'Simulation $index (seed $seed) failed\n'
'Replay: DST_SEED_BASE=$seed DST_SEEDS=1 DST_ROUNDS=$rounds DST_PROFILE=${profile.name} DST_GRAPH_WIDTH=$graphWidth dart test -P dst\n'
'Replay: DST_SEED_BASE=$seed DST_SEEDS=1 DST_ROUNDS=$rounds DST_PROFILE=${profile.name} DST_GRAPH_WIDTH=$graphWidth DST_DELIVERY=${delivery.name} dart test -P dst\n'
'$error',
),
stackTrace,
@ -395,8 +428,10 @@ class DstPropertyFailure implements Exception {
this.replica,
this.profile = DstProfile.sparse,
this.graphWidth = 2,
this.delivery = DstDeliveryMode.full,
});
final DstDeliveryMode delivery;
final DstProfile profile;
final int graphWidth;
@ -417,7 +452,7 @@ class DstPropertyFailure implements Exception {
final buffer = StringBuffer()
..writeln('DST property failure (seed $seed)')
..writeln(
'Replay: DST_SEED_BASE=$seed DST_SEEDS=1 DST_ROUNDS=$rounds DST_PROFILE=${profile.name} DST_GRAPH_WIDTH=$graphWidth dart test -P dst',
'Replay: DST_SEED_BASE=$seed DST_SEEDS=1 DST_ROUNDS=$rounds DST_PROFILE=${profile.name} DST_GRAPH_WIDTH=$graphWidth DST_DELIVERY=${delivery.name} dart test -P dst',
);
if (replica != null) buffer.writeln('Replica: $replica');
for (final violation in violations) {

35
test/serverpod_offline_sync_test_server/test/dst/framework/dst_world.dart
View file

@ -2,6 +2,7 @@ import 'package:clock/clock.dart';
// Imported with `show` because the barrel below re-exports overlapping names.
import 'package:serverpod_database/serverpod_database.dart'
show DatabaseSession, TableRow, Transaction;
import 'package:serverpod_offline_sync/src/sync/space_state.dart';
import 'package:serverpod_offline_sync_server/serverpod_offline_sync_server.dart';
import 'package:serverpod_offline_sync_test_client/serverpod_offline_sync_test_client.dart';
@ -167,18 +168,32 @@ class DstReplica {
);
}
/// Collects every change this replica holds for [spaceUuid].
/// Reads the production resume vector under this replica's seeded clock.
Future<List<Hlc>> checkpoints(UuidValue spaceUuid) => withReplicaClock(
() async => (await sync.createSyncSinceHlc(
rawSession,
spaceId: spaceUuid,
)).nodeCheckpoints,
);
/// Collects a complete batch after the receiver's committed [checkpoints].
///
/// The harness deliberately collects the full history rather than tracking
/// per-peer checkpoints. Redelivery is a merge the engine must absorb
/// idempotently, so letting the adversary resend is a property under test
/// rather than a defect in the harness.
Future<CrdtMergeSet> collect(UuidValue spaceUuid) async {
/// 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.
Future<CrdtMergeSet> collect(
UuidValue spaceUuid, {
List<Hlc> checkpoints = const [],
}) async {
return withReplicaClock(
() => sync
.collectPendingChanges(
rawSession,
checkpointsBySpaceUuid: {spaceUuid: const []},
checkpointsBySpaceUuid: {
spaceUuid: OfflineSyncSpaceState.normalizeCheckpoints(
checkpoints,
).values.toList(),
},
)
.toList(),
);
@ -188,7 +203,11 @@ class DstReplica {
Future<void> merge(CrdtMergeSet changes, UuidValue spaceUuid) async {
if (changes.isEmpty) return;
await withReplicaClock(
() => session.db.mergeChanges(changes, spaceId: spaceUuid),
() => sync.mergeInboundBatch(
rawSession,
spaceId: spaceUuid,
mergeSet: changes,
),
);
}