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: Preserve space state across export round trips (#148)

Browse source
This commit is contained in:
Marcelo Mendonça Soares 2026-09-18 15:33:24 -03:00 • committed by GitHub
commit fc6aaa7b4c
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
4 changed files with 448 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

11
packages/serverpod_offline_sync/lib/src/database/merge.dart
View file

@ -159,6 +159,7 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
// Field metadata for merged inserts is recorded per table and node rather
// than per row: every query behind it already takes a set of row ids.
final pendingAttempts = _PendingInsertAttempts();
final deferredDeletes = <CrdtMergeDelete>[];
for (final operation in operations) {
if (!_context.isCrdtTrackedTableName(operation.tableName)) {
@ -191,6 +192,13 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
transaction,
);
case final CrdtMergeDelete delete:
// A higher delete/restore generation can have an older HLC than a
// concurrent reinsertion. Bootstrap must retain it until this batch
// has inserted the row it belongs to.
if (!context.rows.containsKey((delete.tableName, delete.uuidRowId))) {
deferredDeletes.add(delete);
continue;
}
await _applyMergeDelete(
delete,
nodesByUuid,
@ -204,6 +212,9 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
}
await _flushInsertAttempts(pendingAttempts, context, transaction);
for (final delete in deferredDeletes) {
await _applyMergeDelete(delete, nodesByUuid, context, transaction);
}
if (mayAffectProjection) {
await _foreignKeyProjector.project(

15
packages/serverpod_offline_sync/lib/src/database/merge_utils/unique_resolver.dart
View file

@ -162,6 +162,21 @@ class CrdtUniqueConflictResolver {
if (rowKey.$1 == tableName) rowKey,
]..sort(compareMergeRowKeys);
// A null tuple makes no unique claim, but still needs to shed any old
// release. Restore candidates first; per-index winners then retain their
// canonical tuples so they remain identifiable through those indexes.
for (final rowKey in tableRowKeys) {
if (!resolveVisibleConflicts && !hidden.contains(rowKey)) continue;
for (final uniqueIndex in uniqueIndexes) {
_materializeCanonicalClaim(
rowKey: rowKey,
uniqueIndex: uniqueIndex,
claimByField: claimByField,
valuesByRow: valuesByRow,
);
}
}
for (final uniqueIndex in uniqueIndexes) {
for (final rowKey in tableRowKeys) {
if (!hidden.contains(rowKey)) continue;

321
test/serverpod_offline_sync_test_server/test/integration/sync/sync_nullable_unique_projection_test.dart Normal file
View file

@ -0,0 +1,321 @@
import 'package:serverpod_offline_sync_server/serverpod_offline_sync_server.dart';
import 'package:serverpod_offline_sync_test_client/serverpod_offline_sync_test_client.dart';
import 'package:test/test.dart';
import '../test_tools/client_session.dart';
import '../test_tools/crdt_probes.dart';
import '../test_tools/sync_topology.dart';
void main() {
initTestClientSession();
group(
'Given a visible composite unique loser with a released name,',
() {
late SyncNode source;
late UniqueMixedFk loser;
late Hlc insertionHlc;
late CrdtMergeUpdate detach;
setUp(() async {
source = (
raw: testSession,
offlineSync: session,
sync: OfflineSyncEngine(
syncTables: testSyncTables,
serializationManager: testSession.db.serializationManager,
),
);
final parent = Person(id: const Uuid().v7obj(), name: 'parent');
final winner = UniqueMixedFk(
id: const Uuid().v7obj(),
name: 'claim',
parentId: parent.id,
);
await session.db.transactionForUser(testCrdtUserId, (tx) async {
await Person.db.insertRow(session, parent, transaction: tx);
await UniqueMixedFk.db.insertRow(session, winner, transaction: tx);
});
loser = UniqueMixedFk(
id: const Uuid().v7obj(),
name: 'claim',
parentId: parent.id,
);
final node = const Uuid().v7obj();
final timestamp = (await rowHlc(winner.id!)).datetime.add(
const Duration(milliseconds: 1),
);
await session.db.mergeChanges(
[
CrdtMergeInsert(
uuidSpaceId: testCrdtUserId,
tableName: UniqueMixedFk.t.tableName,
uuidRowId: loser.id!,
uuidNodeId: node,
hlcDatetime: timestamp,
hlcCounter: 0,
data: loser,
),
],
spaceId: testCrdtUserId,
);
insertionHlc = await rowHlc(loser.id!);
final projected = await UniqueMixedFk.db.findById(testSession, loser.id!);
expect(
projected!.name,
'claim__conflict__${loser.id!.uuid}',
);
detach = CrdtMergeUpdate(
uuidSpaceId: testCrdtUserId,
tableName: UniqueMixedFk.t.tableName,
uuidRowId: loser.id!,
uuidNodeId: node,
hlcDatetime: timestamp,
hlcCounter: 2,
columnName: UniqueMixedFk.t.parentId.columnName,
value: null,
);
});
group('when a remote write detaches its parent,', () {
setUp(() async {
await session.db.mergeChanges([detach], spaceId: testCrdtUserId);
});
test(
'then the name returns to its authored value without a new clock.',
() async {
final row = await UniqueMixedFk.db.findById(testSession, loser.id!);
expect(row!.name, 'claim');
expect(row.parentId, isNull);
expect(
await attemptedValue(rowId: loser.id!, columnName: 'name'),
isNull,
);
final changes = await source.sync
.collectPendingChanges(
source.raw,
checkpointsBySpaceUuid: {testCrdtUserId: const []},
)
.toList();
final insert = changes.inserts.singleWhere(
(c) => c.uuidRowId == loser.id,
);
expect((insert.data as UniqueMixedFk).name, 'claim');
expect(insert.hlc, insertionHlc);
expect(
changes.updates.where(
(c) => c.uuidRowId == loser.id && c.columnName == 'name',
),
isEmpty,
);
},
);
test('then an empty replica reproduces its domain and visibility.', () async {
final mirror = await syncNode(
await createAdditionalTestSession(),
testSyncTables,
);
await pushChanges(source, mirror);
final original = await UniqueMixedFk.db.findById(source.raw, loser.id!);
final copied = await UniqueMixedFk.db.findById(mirror.raw, loser.id!);
expect(copied!.name, original!.name);
expect(copied.parentId, original.parentId);
expect(
await UniqueMixedFk.db.findById(source.offlineSync, loser.id!),
isNotNull,
);
expect(
await rowHlc(loser.id!, databaseSession: source.offlineSync),
insertionHlc,
);
expect(
await attemptedValue(
rowId: loser.id!,
columnName: 'name',
databaseSession: source.offlineSync,
),
isNull,
);
expect(
await UniqueMixedFk.db.findById(mirror.offlineSync, loser.id!),
isNotNull,
);
expect(
await rowHlc(loser.id!, databaseSession: mirror.offlineSync),
insertionHlc,
);
expect(
await attemptedValue(
rowId: loser.id!,
columnName: 'name',
databaseSession: mirror.offlineSync,
),
isNull,
);
});
});
},
);
group(
'Given a deleted composite unique loser with a released name,',
() {
late SyncNode source;
late UniqueMixedFk loser;
late Hlc insertionHlc;
late CrdtMergeUpdate detach;
setUp(() async {
source = (
raw: testSession,
offlineSync: session,
sync: OfflineSyncEngine(
syncTables: testSyncTables,
serializationManager: testSession.db.serializationManager,
),
);
final parent = Person(id: const Uuid().v7obj(), name: 'parent');
final winner = UniqueMixedFk(
id: const Uuid().v7obj(),
name: 'claim',
parentId: parent.id,
);
await session.db.transactionForUser(testCrdtUserId, (tx) async {
await Person.db.insertRow(session, parent, transaction: tx);
await UniqueMixedFk.db.insertRow(session, winner, transaction: tx);
});
loser = UniqueMixedFk(
id: const Uuid().v7obj(),
name: 'claim',
parentId: parent.id,
);
final node = const Uuid().v7obj();
final timestamp = (await rowHlc(winner.id!)).datetime.add(
const Duration(milliseconds: 1),
);
await session.db.mergeChanges(
[
CrdtMergeInsert(
uuidSpaceId: testCrdtUserId,
tableName: UniqueMixedFk.t.tableName,
uuidRowId: loser.id!,
uuidNodeId: node,
hlcDatetime: timestamp,
hlcCounter: 0,
data: loser,
),
CrdtMergeDelete(
uuidSpaceId: testCrdtUserId,
tableName: UniqueMixedFk.t.tableName,
uuidRowId: loser.id!,
uuidNodeId: node,
hlcDatetime: timestamp,
hlcCounter: 1,
clFlag: 2,
reason: CrdtDataDeletedReason.userDelete,
),
],
spaceId: testCrdtUserId,
);
insertionHlc = await rowHlc(loser.id!);
final projected = await UniqueMixedFk.db.findById(testSession, loser.id!);
expect(
projected!.name,
'claim__hidden__${loser.id!.uuid}',
);
detach = CrdtMergeUpdate(
uuidSpaceId: testCrdtUserId,
tableName: UniqueMixedFk.t.tableName,
uuidRowId: loser.id!,
uuidNodeId: node,
hlcDatetime: timestamp,
hlcCounter: 2,
columnName: UniqueMixedFk.t.parentId.columnName,
value: null,
);
});
group('when a remote write detaches its parent,', () {
setUp(() async {
await session.db.mergeChanges([detach], spaceId: testCrdtUserId);
});
test(
'then the name returns to its authored value without a new clock.',
() async {
final row = await UniqueMixedFk.db.findById(testSession, loser.id!);
expect(row!.name, 'claim');
expect(row.parentId, isNull);
expect(
await attemptedValue(rowId: loser.id!, columnName: 'name'),
isNull,
);
final changes = await source.sync
.collectPendingChanges(
source.raw,
checkpointsBySpaceUuid: {testCrdtUserId: const []},
)
.toList();
final insert = changes.inserts.singleWhere(
(c) => c.uuidRowId == loser.id,
);
expect((insert.data as UniqueMixedFk).name, 'claim');
expect(insert.hlc, insertionHlc);
expect(
changes.updates.where(
(c) => c.uuidRowId == loser.id && c.columnName == 'name',
),
isEmpty,
);
},
);
test('then an empty replica reproduces its domain and visibility.', () async {
final mirror = await syncNode(
await createAdditionalTestSession(),
testSyncTables,
);
await pushChanges(source, mirror);
final original = await UniqueMixedFk.db.findById(source.raw, loser.id!);
final copied = await UniqueMixedFk.db.findById(mirror.raw, loser.id!);
expect(copied!.name, original!.name);
expect(copied.parentId, original.parentId);
expect(
await UniqueMixedFk.db.findById(source.offlineSync, loser.id!),
isNull,
);
expect(
await rowHlc(loser.id!, databaseSession: source.offlineSync),
insertionHlc,
);
expect(
await attemptedValue(
rowId: loser.id!,
columnName: 'name',
databaseSession: source.offlineSync,
),
isNull,
);
expect(
await UniqueMixedFk.db.findById(mirror.offlineSync, loser.id!),
isNull,
);
expect(
await rowHlc(loser.id!, databaseSession: mirror.offlineSync),
insertionHlc,
);
expect(
await attemptedValue(
rowId: loser.id!,
columnName: 'name',
databaseSession: mirror.offlineSync,
),
isNull,
);
});
});
},
);
}

101
test/serverpod_offline_sync_test_server/test/integration/sync/sync_reinsert_tombstone_bootstrap_test.dart Normal file
View file

@ -0,0 +1,101 @@
import 'package:clock/clock.dart';
import 'package:serverpod_offline_sync_server/serverpod_offline_sync_server.dart';
import 'package:serverpod_offline_sync_test_client/serverpod_offline_sync_test_client.dart';
import 'package:test/test.dart';
import '../test_tools/client_session.dart';
import '../test_tools/crdt_probes.dart';
import '../test_tools/sync_topology.dart';
void main() {
initTestClientSession();
group(
'Given a newer reinsertion hidden by an older higher-generation deletion,',
() {
late SyncNode peer;
late Unique row;
late Hlc reinsertHlc;
late CrdtMergeDelete deletion;
setUp(() async {
final author = await syncNode(await createAdditionalTestSession(), [Unique.t]);
peer = await syncNode(await createAdditionalTestSession(), [Unique.t]);
row = Unique(id: const Uuid().v7obj(), name: 'original');
final start = DateTime.now().toUtc();
await withClock(Clock.fixed(start), () async {
await author.offlineSync.db.transactionForUser(testCrdtUserId, (tx) async {
await Unique.db.insertRow(author.offlineSync, row, transaction: tx);
await Unique.db.deleteRow(author.offlineSync, row, transaction: tx);
});
await pushChanges(author, peer);
});
await withClock(Clock.fixed(start.add(const Duration(seconds: 1))), () async {
await author.offlineSync.db.transactionForUser(testCrdtUserId, (tx) async {
await Unique.db.insertRow(author.offlineSync, row, transaction: tx);
await Unique.db.deleteRow(author.offlineSync, row, transaction: tx);
});
});
await withClock(Clock.fixed(start.add(const Duration(seconds: 2))), () async {
row = row.copyWith(name: 'peer reinsertion');
await peer.offlineSync.db.transactionForUser(
testCrdtUserId,
(tx) => Unique.db.insertRow(peer.offlineSync, row, transaction: tx),
);
reinsertHlc = await rowHlc(row.id!, databaseSession: peer.offlineSync);
await pushChanges(author, peer);
});
final changes = await peer.sync
.collectPendingChanges(
peer.raw,
checkpointsBySpaceUuid: {testCrdtUserId: const []},
)
.toList();
deletion = changes.deletes.single;
expect(deletion.clFlag, 4);
expect(deletion.hlc.compareTo(reinsertHlc), isNegative);
expect(await Unique.db.findById(peer.offlineSync, row.id!), isNull);
});
group('when an empty replica merges its complete export,', () {
late SyncNode mirror;
setUp(() async {
mirror = await syncNode(await createAdditionalTestSession(), [Unique.t]);
await pushChanges(peer, mirror);
});
test('then the higher-generation tombstone and hidden row survive.', () async {
expect(await Unique.db.findById(mirror.offlineSync, row.id!), isNull);
final copied = await Unique.db.findById(mirror.raw, row.id!);
expect(copied!.name, 'peer reinsertion__hidden__${row.id!.uuid}');
final changes = await mirror.sync
.collectPendingChanges(
mirror.raw,
checkpointsBySpaceUuid: {testCrdtUserId: const []},
)
.toList();
expect(changes.deletes.single.toJson(), deletion.toJson());
});
test('then the newer authored name and insertion clock survive.', () async {
final changes = await mirror.sync
.collectPendingChanges(
mirror.raw,
checkpointsBySpaceUuid: {testCrdtUserId: const []},
)
.toList();
expect((changes.inserts.single.data as Unique).name, 'peer reinsertion');
expect(changes.inserts.single.hlc, reinsertHlc);
final attempted = await attemptedValue(
rowId: row.id!,
columnName: 'name',
databaseSession: mirror.offlineSync,
);
expect(attempted!.value, 'peer reinsertion');
expect(attempted.projectionReason, CrdtProjectionReason.hiddenUniqueRelease);
});
});
},
);
}