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: Allow only one releasable column per unique index (#91)

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

23
docs/unique-constraint-invariants.md
View file

@ -31,14 +31,20 @@ For each unique index conflict group:
full HLC ordering as the rest of CRDT merge. This includes node id as the HLC
tie-break. Row id is only a final fallback for identical HLCs.
3. Leave the winner's unique value unchanged.
4. Rewrite each loser's visible unique-indexed value to a deterministic
conflict-free value:
4. Rewrite each loser's visible releasable unique-indexed value to a
deterministic conflict-free value:
- nullable unique values may be released to `null`;
- text values receive a stable `__conflict__<rowId>` suffix;
- UUID values receive a deterministic synthetic UUID.
5. Keep all rows visible. Application code can decide whether and how to surface
the changed visible value as a business conflict.
For a composite unique index, fixed partition columns such as `scopeId` are not
released, and not every remaining indexed column has to be releasable. The index
is supported when at least one non-`scopeId` column can be released
deterministically. Stable discriminator columns may remain unchanged; changing
the releasable column or columns is enough to make the unique tuple conflict-free.
The rewritten value is a materialized data value, not a synced conflict class.
The core CRDT protocol does not need to expose unique conflict objects to users.
@ -183,22 +189,27 @@ visibility and FK values, without also needing row aliasing and field folding.
when the loser is released, then the materialized UUID is deterministic from
the table, column, original value, and loser row id.
7. Given a scoped composite unique index contains a stable non-releasable
discriminator and a releasable text column, when a loser is released, then the
discriminator remains unchanged and the text column receives the deterministic
conflict suffix.
### Deferred Merge And Overwrite Policies
7. Given overwrite is enabled for a unique index, when a loser has visible
8. Given overwrite is enabled for a unique index, when a loser has visible
restrict children, then FK projection retargets those children through the
row alias or blocks the projection deterministically; no visible orphan is
produced.
8. Given merge is enabled for a unique index, when loser rows contain newer
9. Given merge is enabled for a unique index, when loser rows contain newer
non-key field values, then field folding follows documented per-column
policies and every replica computes the same winner row.
9. Given a row participates in multiple unique conflict groups, when merge or
10. Given a row participates in multiple unique conflict groups, when merge or
overwrite projection runs, then row aliases are deterministic and do not form
cycles.
10. Given FK repair changes a unique-indexed visible value, when combined unique
11. Given FK repair changes a unique-indexed visible value, when combined unique
and FK projection runs, then the fixed-point result matches a full
recomputation oracle.

32
packages/serverpod_offline_sync_client/lib/src/database/merge_utils/unique_resolver.dart
View file

@ -71,12 +71,12 @@ class CrdtUniqueConflictResolver {
required _MergeContext context,
}) {
var uniqueHlc = insert.hlc;
for (final column in uniqueIndex.columns) {
for (final columnName in uniqueIndex.indexedColumns) {
uniqueHlc = uniqueHlc.maxBetween(
context.incomingFieldHlcs[(
insert.tableName,
insert.uuidRowId,
column.columnName,
columnName,
)],
);
}
@ -120,7 +120,7 @@ class CrdtUniqueConflictResolver {
final uniqueColumnNamesToRead = {
for (final uniqueIndex in _recorder._uniqueIndexesForTable(tableDefinition))
for (final column in uniqueIndex.columns) column.columnName,
...uniqueIndex.indexedColumns,
};
if (uniqueColumnNamesToRead.isEmpty) {
return updates;
@ -168,7 +168,7 @@ class CrdtUniqueConflictResolver {
values,
conflict.uniqueIndex,
);
for (final columnName in conflict.uniqueIndex.columnNames) {
for (final columnName in conflict.uniqueIndex.releaseColumnNames) {
if (releasedValues.containsKey(columnName)) {
resolvedUpdates[columnName] = releasedValues[columnName];
values[columnName] = releasedValues[columnName];
@ -198,12 +198,12 @@ class CrdtUniqueConflictResolver {
final released = Map<String, Object?>.from(data);
final values = {
for (final column in uniqueIndex.columns)
column.columnName: released[column.columnName],
for (final columnName in uniqueIndex.indexedColumns)
columnName: released[columnName],
};
if (values.values.any((value) => value == null)) return released;
for (final column in uniqueIndex.columns) {
for (final column in uniqueIndex.releaseColumns) {
released[column.columnName] = _recorder._conflictFreeValue(
column,
values[column.columnName],
@ -221,11 +221,10 @@ class CrdtUniqueConflictResolver {
required _UniqueIndexConflictRelease uniqueIndex,
required Transaction transaction,
}) async {
final columnNames = uniqueIndex.columnNames.toList();
final valuesByRowId = await _recorder._readDomainColumnValues(
tableName,
{rowId},
columnNames,
uniqueIndex.indexedColumns,
transaction,
);
final values = valuesByRowId[rowId];
@ -255,9 +254,8 @@ class CrdtUniqueConflictResolver {
final uniqueIndexesByConflictId = <UuidValue, Set<_UniqueIndexConflictRelease>>{};
for (final uniqueIndex in uniqueIndexes) {
final columnValues = {
for (final column in uniqueIndex.columns)
if (values.containsKey(column.columnName))
column.columnName: values[column.columnName],
for (final columnName in uniqueIndex.indexedColumns)
if (values.containsKey(columnName)) columnName: values[columnName],
};
if (uniqueIndex.scoped) {
columnValues['scopeId'] = _recorder
@ -265,7 +263,7 @@ class CrdtUniqueConflictResolver {
.normalizedScopeId;
}
final expectedValueCount =
uniqueIndex.columns.length + (uniqueIndex.scoped ? 1 : 0);
uniqueIndex.indexedColumns.length + (uniqueIndex.scoped ? 1 : 0);
if (columnValues.length != expectedValueCount) continue;
if (columnValues.values.any((value) => value == null)) continue;
@ -314,7 +312,7 @@ class CrdtUniqueConflictResolver {
required Transaction transaction,
}) async {
final uniqueColumnNames = {
for (final column in uniqueIndex.columns) column.columnName,
...uniqueIndex.indexedColumns,
};
final missingColumnIds = <int>{};
final maxFieldHlc = <Hlc>[];
@ -374,12 +372,12 @@ class CrdtUniqueConflictResolver {
if (tableDefinition == null) return false;
return _recorder
._uniqueIndexesForTable(tableDefinition)
.any((i) => i.columns.any((c) => c.columnName == columnName));
.any((i) => i.indexedColumns.contains(columnName));
}
}
extension on _UniqueIndexConflictRelease {
Set<String> get columnNames => {
for (final column in columns) column.columnName,
Set<String> get releaseColumnNames => {
for (final column in releaseColumns) column.columnName,
};
}

112
packages/serverpod_offline_sync_client/lib/src/database/recorder.dart
View file

@ -37,21 +37,16 @@ typedef _ForeignKeyEdge = ({
typedef _ForeignKeyValueKey = (String tableName, String columnName, String value);
typedef _UniqueIndexConflictRelease = ({
List<_UniqueColumnConflictRelease> columns,
List<String> indexedColumns,
List<_UniqueColumnConflictRelease> releaseColumns,
bool scoped,
});
typedef _UniqueColumnConflictRelease = ({
String columnName,
_UniqueConflictReleaseKind kind,
CrdtUniqueConflictReleaseKind kind,
});
enum _UniqueConflictReleaseKind {
setNull,
textSuffix,
syntheticUuid,
}
enum _ForeignKeyTargetPresence {
absent,
visible,
@ -1113,12 +1108,12 @@ WHERE c."id" IN ($whereRowIds)
final uniqueIndexes = _uniqueIndexesForTable(tableDefinition);
for (final uniqueIndex in uniqueIndexes) {
final columnNames = uniqueIndex.columnNames.toList();
final releaseColumnNames = uniqueIndex.releaseColumnNames.toList();
final valuesByRowId = await _readDomainColumnValues(
tableName,
rowIds,
columnNames,
uniqueIndex.indexedColumns,
transaction,
);
final updatedRowIds = <UuidValue>{};
@ -1127,7 +1122,7 @@ WHERE c."id" IN ($whereRowIds)
if (values.values.any((value) => value == null)) continue;
final updates = <String, Object?>{};
for (final column in uniqueIndex.columns) {
for (final column in uniqueIndex.releaseColumns) {
updates[column.columnName] = _conflictFreeValue(
column,
values[column.columnName],
@ -1145,7 +1140,7 @@ WHERE c."id" IN ($whereRowIds)
await recordFieldsUpdatedByTable(
tableName,
updatedRowIds,
columnNames,
releaseColumnNames,
transaction,
);
}
@ -1290,13 +1285,13 @@ WHERE "id" IN (${_sqlLiteralList(rowIds)})
String releaseSuffix,
) {
switch (column.kind) {
case _UniqueConflictReleaseKind.setNull:
case CrdtUniqueConflictReleaseKind.setNull:
return null;
case _UniqueConflictReleaseKind.textSuffix:
case CrdtUniqueConflictReleaseKind.textSuffix:
if (value is String) {
return '${value}__${releaseSuffix}__${conflictingId.uuid}';
}
case _UniqueConflictReleaseKind.syntheticUuid:
case CrdtUniqueConflictReleaseKind.syntheticUuid:
if (value != null) {
return const Uuid().v5obj(
Namespace.oid.value,
@ -1414,88 +1409,39 @@ List<_UniqueIndexConflictRelease> _syncableUniqueIndexesForTable(
TableDefinition table,
Set<String> syncTableNames,
) {
final uniqueIndexes = table.indexes
.where(
(index) =>
index.isUnique &&
!index.isPrimary &&
index.elements.every(
(element) => element.type == IndexElementDefinitionType.column,
) &&
(_isScopedUniqueIndex(index) ||
isCrdtAllowedForeignKeyOnlyUniqueIndex(
table,
index,
syncTableNames,
)),
)
.toList();
final columnsByName = {for (final column in table.columns) column.name: column};
return [
for (final index in uniqueIndexes)
for (final index in crdtSyncableUniqueIndexesForTable(table, syncTableNames))
_uniqueIndexConflictReleaseForIndex(table, columnsByName, index),
];
}
bool _isScopedUniqueIndex(IndexDefinition index) {
return index.elements.any(
(element) =>
element.type == IndexElementDefinitionType.column &&
element.definition == 'scopeId',
);
}
_UniqueIndexConflictRelease _uniqueIndexConflictReleaseForIndex(
TableDefinition table,
Map<String, ColumnDefinition> columnsByName,
IndexDefinition index,
) {
final columnNames = index.elements.map((element) => element.definition).toList();
final indexedColumns = [
for (final columnName in columnNames)
if (columnName != 'scopeId') columnName,
];
final releaseColumns = <_UniqueColumnConflictRelease>[];
for (final columnName in indexedColumns) {
final column = columnsByName[columnName];
if (column == null) {
throw StateError('No column definition found for ${table.name}.$columnName.');
}
final kind = crdtUniqueConflictReleaseKindForColumn(table, column);
if (kind != null) {
releaseColumns.add((columnName: columnName, kind: kind));
}
}
return (
scoped: columnNames.contains('scopeId'),
columns: [
for (final columnName in columnNames)
if (columnName != 'scopeId')
_uniqueConflictReleaseForColumn(
table,
columnsByName[columnName],
columnName,
),
],
);
}
_UniqueColumnConflictRelease _uniqueConflictReleaseForColumn(
TableDefinition table,
ColumnDefinition? column,
String columnName,
) {
if (column == null) {
throw StateError('No column definition found for ${table.name}.$columnName.');
}
if (column.isNullable) {
return (columnName: columnName, kind: _UniqueConflictReleaseKind.setNull);
}
if (column.columnType == ColumnType.text) {
return (columnName: columnName, kind: _UniqueConflictReleaseKind.textSuffix);
}
if (column.columnType == ColumnType.uuid && !_isForeignKeyColumn(table, columnName)) {
return (columnName: columnName, kind: _UniqueConflictReleaseKind.syntheticUuid);
}
throw StateError(
'CRDT soft deletes cannot release unique conflicts for ${table.name}.$columnName. '
'Only nullable, text, and non-FK UUID unique columns are supported.',
);
}
bool _isForeignKeyColumn(TableDefinition table, String columnName) {
return table.foreignKeys.any(
(fk) => fk.columns.length == 1 && fk.columns.single == columnName,
indexedColumns: indexedColumns,
releaseColumns: releaseColumns,
);
}

20
packages/serverpod_offline_sync_client/lib/src/database/schema.dart
View file

@ -58,6 +58,19 @@ class CrdtSchemaRegistry {
);
}
final nonReleasableUniqueIndexes = crdtNonReleasableUniqueIndexViolations(
syncTables,
tableDefinitionsByName,
);
if (nonReleasableUniqueIndexes.isNotEmpty) {
throw StateError(
'CRDT unique conflict resolution requires at least one releasable '
'non-scope column for ${nonReleasableUniqueIndexes.length} unique '
'index(es): ${nonReleasableUniqueIndexes.join(', ')}. '
'Only nullable, text, and non-FK UUID unique columns are supported.',
);
}
final tablesMissingScopeIdRelation = _missingCrdtScopeRelations(
syncTables,
tableDefinitionsByName,
@ -217,12 +230,7 @@ bool _isForbiddenGlobalUniqueIndex(
Set<String> syncTableNames,
) {
if (!index.isUnique || index.isPrimary) return false;
final includesScopeId = index.elements.any(
(element) =>
element.type == IndexElementDefinitionType.column &&
element.definition == 'scopeId',
);
if (includesScopeId) return false;
if (isCrdtScopedUniqueIndex(index)) return false;
return !isCrdtAllowedForeignKeyOnlyUniqueIndex(
tableDefinition,

106
packages/serverpod_offline_sync_client/lib/src/database/unique_index_utils.dart
View file

@ -29,3 +29,109 @@ bool isCrdtAllowedForeignKeyOnlyUniqueIndex(
return indexColumns.every(coveredForeignKeyColumns.contains);
}
@internal
bool isCrdtScopedUniqueIndex(IndexDefinition index) {
return index.elements.any(
(element) =>
element.type == IndexElementDefinitionType.column &&
element.definition == 'scopeId',
);
}
@internal
Iterable<IndexDefinition> crdtSyncableUniqueIndexesForTable(
TableDefinition table,
Set<String> syncTableNames,
) {
return table.indexes.where(
(index) =>
index.isUnique &&
!index.isPrimary &&
index.elements.every(
(element) => element.type == IndexElementDefinitionType.column,
) &&
(isCrdtScopedUniqueIndex(index) ||
isCrdtAllowedForeignKeyOnlyUniqueIndex(
table,
index,
syncTableNames,
)),
);
}
@internal
bool isCrdtForeignKeyColumn(TableDefinition table, String columnName) {
return table.foreignKeys.any(
(fk) => fk.columns.length == 1 && fk.columns.single == columnName,
);
}
/// How a unique-indexed column can be released to a conflict-free value, or
/// null when the column cannot be released deterministically.
@internal
enum CrdtUniqueConflictReleaseKind {
setNull,
textSuffix,
syntheticUuid,
}
@internal
CrdtUniqueConflictReleaseKind? crdtUniqueConflictReleaseKindForColumn(
TableDefinition table,
ColumnDefinition column,
) {
if (column.isNullable) return CrdtUniqueConflictReleaseKind.setNull;
if (column.columnType == ColumnType.text) {
return CrdtUniqueConflictReleaseKind.textSuffix;
}
if (column.columnType == ColumnType.uuid &&
!isCrdtForeignKeyColumn(table, column.name)) {
return CrdtUniqueConflictReleaseKind.syntheticUuid;
}
return null;
}
@internal
List<String> crdtNonReleasableUniqueIndexViolations(
List<Table> syncTables,
Map<String, TableDefinition> tableDefinitionsByName,
) {
final syncTableNames = {for (final table in syncTables) table.tableName};
return [
for (final tableName in syncTableNames)
if (tableDefinitionsByName.containsKey(tableName))
for (final index in crdtSyncableUniqueIndexesForTable(
tableDefinitionsByName[tableName]!,
syncTableNames,
))
if (!_hasReleasableNonScopeUniqueColumn(
tableDefinitionsByName[tableName]!,
index,
))
'$tableName.${index.indexName}',
];
}
bool _hasReleasableNonScopeUniqueColumn(
TableDefinition table,
IndexDefinition index,
) {
final columnsByName = {
for (final column in table.columns) column.name: column,
};
for (final element in index.elements) {
if (element.type != IndexElementDefinitionType.column) continue;
if (element.definition == 'scopeId') continue;
final column = columnsByName[element.definition];
if (column == null) {
throw StateError(
'No column definition found for ${table.name}.${element.definition}.',
);
}
if (crdtUniqueConflictReleaseKindForColumn(table, column) != null) {
return true;
}
}
return false;
}

1
test/serverpod_offline_sync_test_server/lib/src/demo_sync_tables.dart
View file

@ -24,6 +24,7 @@ final demoSyncTables = <Table>[
Types.t,
Unique.t,
UniqueComposite.t,
UniqueDiscriminator.t,
UniqueSetNullChild.t,
UniqueUuid.t,
];

1
test/serverpod_offline_sync_test_server/lib/src/endpoints/demo_debug_endpoint.dart
View file

@ -295,6 +295,7 @@ class DemoDebugEndpoint extends Endpoint {
await add<Unique>();
await add<UniqueUuid>();
await add<UniqueComposite>();
await add<UniqueDiscriminator>();
await add<UniqueSetNullChild>();
await add<Types>();
await add<FkChainRoot>();

9
test/serverpod_offline_sync_test_server/lib/src/models/unique_discriminator.spy.yaml Normal file
View file

@ -0,0 +1,9 @@
class: UniqueDiscriminator
table: unique_discriminator
database: all
fields:
id: UuidValue?, defaultPersist=random_v7
### Owner scope of this row. Maintained by the CRDT sync layer.
scopeId: int?, relation(parent=crdt_scopes, onDelete=Cascade)
categoryId: int
name: String, unique(per=[scopeId, categoryId])

8
test/serverpod_offline_sync_test_server/lib/src/models/unique_no_release.spy.yaml Normal file
View file

@ -0,0 +1,8 @@
class: UniqueNoRelease
table: unique_no_release
database: all
fields:
id: UuidValue?, defaultPersist=random_v7
### Owner scope of this row. Maintained by the CRDT sync layer.
scopeId: int?, relation(parent=crdt_scopes, onDelete=Cascade)
categoryId: int, unique(per=[scopeId])

88
test/serverpod_offline_sync_test_server/test/integration/merge/update_test.dart
View file

@ -421,6 +421,94 @@ void main() {
},
);
group(
'Given a remote update that loses a composite unique conflict with a stable discriminator, ',
() {
late UniqueDiscriminator loser;
late CrdtMergeUpdate remoteUpdate;
setUp(() async {
await session.db.transactionForUser(testCrdtUserId, (tx) async {
await UniqueDiscriminator.db.insertRow(
session,
UniqueDiscriminator(
id: const Uuid().v7obj(),
categoryId: 7,
name: 'shared-name',
),
transaction: tx,
);
loser = await UniqueDiscriminator.db.insertRow(
session,
UniqueDiscriminator(
id: const Uuid().v7obj(),
categoryId: 7,
name: 'original-name',
),
transaction: tx,
);
});
final loserCrdtRow = await CrdtDataRow.db.findFirstRow(
session,
where: (t) => t.uuidRowId.equals(loser.id),
include: CrdtDataRow.include(node: CrdtNode.include()),
);
remoteUpdate = CrdtMergeUpdate(
uuidScopeId: testCrdtUserId,
tableName: UniqueDiscriminator.t.tableName,
uuidRowId: loser.id!,
uuidNodeId: const Uuid().v7obj(),
hlcDatetime: loserCrdtRow!.hlc.datetime.advance(),
hlcCounter: loserCrdtRow.hlc.counter,
columnName: UniqueDiscriminator.t.name.columnName,
value: 'shared-name',
);
mergeSet = [remoteUpdate];
});
group('when merging, ', () {
setUp(() async {
await session.db.mergeChanges(
mergeSet,
scopeId: testCrdtUserId,
);
});
test(
'then the losing row keeps the discriminator and receives a conflict-free name.',
() async {
final row = await UniqueDiscriminator.db.findById(session, loser.id!);
expect(row, isNotNull);
expect(row!.categoryId, 7);
expect(row.name, 'shared-name__conflict__${loser.id!.uuid}');
},
);
test(
'then the stable discriminator does not receive CRDT field metadata.',
() async {
final fields = await CrdtDataField.db.find(
session,
where: (t) => t.row.uuidRowId.equals(loser.id),
include: CrdtDataField.include(
column: CrdtSchemaColumn.include(),
node: CrdtNode.include(),
),
);
expect(fields, hasLength(1));
expect(fields.first.column!.name, UniqueDiscriminator.t.name.columnName);
expect(fields.first.hlc, remoteUpdate.hlc);
},
);
});
},
);
group(
'Given a remote update that loses a UUID unique conflict, ',
() {

64
test/serverpod_offline_sync_test_server/test/integration/schema/schema_test.dart
View file

@ -297,6 +297,70 @@ void main() {
},
);
test(
'Given a CRDT database with a scoped composite unique index that has a stable discriminator and a releasable column, '
'when the database is initialized, '
'then the unique index metadata is accepted.',
() async {
final crdtSession = CrdtDatabaseSession.wraps(
testSession,
syncTables: [UniqueDiscriminator.t],
);
await expectLater(crdtSession.db.initialize(), completes);
},
);
test(
'Given a CRDT schema registry with a scoped unique index that has no releasable non-scope column, '
'when the registry is created, '
'then an error is thrown.',
() {
expect(
() => CrdtSchemaRegistry(
session,
syncTables: [UniqueNoRelease.t],
),
throwsA(
isA<StateError>().having(
(e) => e.message,
'message',
contains(
'CRDT unique conflict resolution requires at least one '
'releasable non-scope column',
),
),
),
);
},
);
test(
'Given a CRDT database with a scoped unique index that has no releasable non-scope column, '
'when the database is initialized, '
'then an error is thrown.',
() async {
final crdtSession = CrdtDatabaseSession.wraps(
testSession,
syncTables: [UniqueNoRelease.t],
);
await expectLater(
crdtSession.db.initialize(),
throwsA(
isA<StateError>().having(
(e) => e.message,
'message',
contains(
'CRDT unique conflict resolution requires at least one '
'releasable non-scope column',
),
),
),
);
},
);
test(
'Given a CRDT schema registry with a synced table that has the scopeId cascade relation to crdt_scopes, '
'when the registry is created, '

1
test/serverpod_offline_sync_test_server/test/integration/test_tools/client_session.dart
View file

@ -98,6 +98,7 @@ Future<void> _initialize() async {
Town.t,
Unique.t,
UniqueComposite.t,
UniqueDiscriminator.t,
UniqueSetNullChild.t,
UniqueUuid.t,
],