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

test: Harden DST authored-state checks and workloads (#123)

Browse source
This commit is contained in:
Marcelo Mendonça Soares 2026-09-17 10:39:07 -03:00 • committed by GitHub
commit ad3e1a49e4
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
10 changed files with 10312 additions and 458 deletions
Download patch file Download diff file Expand all files Collapse all files

1
packages/serverpod_offline_sync/lib/serverpod_offline_sync.dart
View file

@ -1,6 +1,7 @@
export 'src/crdt/extensions.dart';
export 'src/crdt/merge.dart';
export 'src/database/database.dart';
export 'src/database/exceptions.dart';
export 'src/database/recorder.dart';
export 'src/database/schema.dart';
export 'src/database/session.dart'

19
packages/serverpod_offline_sync/lib/src/database/database.dart
View file

@ -432,7 +432,7 @@ class OfflineSyncDatabase implements Database {
(tx) async {
final prepared = _prepareRowsForInsert(rows, tx);
final values = _recorder.withForeignKeyInsertDefaults(prepared.rows);
final projectionUnchanged = await _recorder.prepareLocalUpsert(
final projection = await _recorder.prepareLocalUpsert(
values,
conflictColumns,
updateColumns,
@ -475,7 +475,8 @@ class OfflineSyncDatabase implements Database {
updatedRows,
updateColumns,
tx,
projectionUnchanged: projectionUnchanged,
projectionUnchanged: projection.projectionUnchanged,
domainBeforeUpsert: projection.domain,
);
if (noReturn) return <T>[];
_stripStampedRows(result, prepared);
@ -519,11 +520,24 @@ class OfflineSyncDatabase implements Database {
if (deletedRowIds.contains(row.id)) row,
];
if (rowsToReinsert.isEmpty) return [];
// Hidden targets were filtered out of the delegate's upsert results, so
// its duplicate-target check cannot see these restorations. Check the
// submitted identities before turning them into ordinary updates.
if (rowsToReinsert.map((row) => row.id).toSet().length != rowsToReinsert.length) {
throw DatabaseQueryException(
'ON CONFLICT DO UPDATE command cannot affect row a second time',
code: switch (dialect) {
DatabaseDialect.postgres => PgErrorCode.cardinalityViolation,
DatabaseDialect.sqlite => SqliteErrorCode.integrityConstraintViolation,
},
);
}
final plannedReinserts = await _recorder.planLocalUpdates(
rowsToReinsert,
null,
transaction,
restoring: true,
);
final reinsertedRows = await _delegate.update<T>(
plannedReinserts.rows,
@ -789,6 +803,7 @@ class OfflineSyncDatabase implements Database {
);
final columns = columnValues.map((e) => e.column).toList();
_recorder.validateAuthoredRows(result, columns);
await _recorder.afterUpdate(result, columns, tx);
if (noReturn) return <T>[];
result.forEach(_stripSpaceId);

22
packages/serverpod_offline_sync/lib/src/database/exceptions.dart Normal file
View file

@ -0,0 +1,22 @@
/// A caller tried to author a unique value reserved for sync projection.
class OfflineSyncReservedValueException implements Exception {
/// Creates an exception identifying the rejected field and value.
const OfflineSyncReservedValueException({
required this.tableName,
required this.columnName,
required this.value,
});
/// The synchronized table containing the unique column.
final String tableName;
/// The unique column being authored.
final String columnName;
/// The rejected text or UUID value, rendered as a string.
final String value;
@override
String toString() =>
'Reserved generated unique value for $tableName.$columnName: $value';
}

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

@ -87,6 +87,29 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
if (mergeSet.isEmpty) return;
final operations = mergeSet.causallyOrderedChanges;
// Validate remote facts before touching rows or progress, including stale
// and repeated deliveries. Generated alternatives are never authored facts.
for (final operation in operations) {
if (!_context.isCrdtTrackedTableName(operation.tableName) ||
!_uniqueResolver.hasReservedUniqueColumns(operation.tableName)) {
continue;
}
switch (operation) {
case CrdtMergeInsert():
for (final MapEntry(key: column, value: value)
in operation.databaseColumns.entries) {
_uniqueResolver.validateAuthoredValue(operation.tableName, column, value);
}
case CrdtMergeUpdate():
_uniqueResolver.validateAuthoredValue(
operation.tableName,
operation.columnName,
operation.value,
);
case CrdtMergeDelete():
break;
}
}
// Nothing in this batch touches a foreign key or a unique index, so no
// projection decision can change: skip both passes instead of loading
// state for tables that cannot produce a candidate or a conflict.

37
packages/serverpod_offline_sync/lib/src/database/merge_utils/database_helpers.dart
View file

@ -1,5 +1,3 @@
import 'dart:typed_data';
import 'package:meta/meta.dart';
import 'package:serverpod_database/serverpod_database.dart';
import 'package:serverpod_serialization/serverpod_serialization.dart';
@ -62,41 +60,14 @@ int compareMergeRowKeys(MergeRowKey left, MergeRowKey right) {
return left.$2.uuid.compareTo(right.$2.uuid);
}
/// Canonicalizes a domain/attempted value so UUID blobs round-trip as
/// [UuidValue] rather than as SQLite byte lists.
///
/// Only values already shaped like a UUID are converted. Anything else is
/// returned unchanged, so an ordinary string or number never pays for a thrown
/// [FormatException] on the projection and outbound paths.
/// Decodes UUID columns using their schema, preserving text and binary values
/// even when their contents could also represent a UUID.
@internal
Object? canonicalDomainValue(Object? value) {
if (value is UuidValue) return value;
if (value is Uint8List) {
return value.length == 16 ? UuidValue.fromByteList(value) : value;
}
if (value is String) {
return _looksLikeUuid(value) ? UuidValue.withValidation(value) : value;
}
Object? canonicalDomainValue(Object? value, ColumnDefinition? column) {
if (column?.columnType == ColumnType.uuid) return value.toUuidValue();
return value;
}
/// Whether [value] has the 8-4-4-4-12 shape of a canonical UUID string.
bool _looksLikeUuid(String value) {
if (value.length != 36) return false;
for (var index = 0; index < 36; index++) {
final code = value.codeUnitAt(index);
if (index == 8 || index == 13 || index == 18 || index == 23) {
if (code != 0x2d) return false;
continue;
}
final isDigit = code >= 0x30 && code <= 0x39;
final isLower = code >= 0x61 && code <= 0x66;
final isUpper = code >= 0x41 && code <= 0x46;
if (!isDigit && !isLower && !isUpper) return false;
}
return true;
}
/// Extensions on [UuidValue?] used for UUID comparison.
@internal
extension UuidValueComparisonExtension on UuidValue? {

91
packages/serverpod_offline_sync/lib/src/database/merge_utils/foreign_key_projector.dart
View file

@ -306,7 +306,14 @@ class CrdtForeignKeyProjector {
if (projectionValuesEqual(attempt.value, values[columnName])) continue;
projectionWrites.add(
(fieldId: fieldId, value: attempt.value, reason: attempt.reason),
(
fieldId: fieldId,
value: canonicalDomainValue(
attempt.value,
_context.columnsByTableAndName[tableName]?[columnName],
),
reason: attempt.reason,
),
);
}
}
@ -606,36 +613,6 @@ class CrdtForeignKeyProjector {
};
}
/// Fields that currently carry a sparse attempted-value override.
///
/// Reinsert must not bump these field HLCs: the authored fact is unchanged
/// and the domain difference is projection.
Future<Set<MergeFieldKey>> findActiveAttemptedFields({
required String tableName,
required Set<UuidValue> rowIds,
required Transaction transaction,
}) async {
if (rowIds.isEmpty) return const {};
final columnNames = {
..._foreignKeys.foreignKeyColumnsFor(tableName),
..._uniqueResolver.uniqueColumnNamesFor(tableName),
};
if (columnNames.isEmpty) return const {};
final fields = await _loadFields(
tableName: tableName,
rowIds: rowIds,
columnNames: columnNames,
transaction: transaction,
);
return {
for (final field in fields)
if (field.attemptedValue != null)
(tableName, field.row!.uuidRowId, field.column!.name),
};
}
/// Proves that an ordinary local write leaves the existing projection fixed.
///
/// Only new, unreferenced rows and additions to null FKs qualify. All parents
@ -689,6 +666,7 @@ class CrdtForeignKeyProjector {
inserting || writtenColumns == null || writtenColumns.contains(column)
? supplied[column]
: previous[column],
_context.columnsByTableAndName[tableName]?[column],
),
};
if (!inserting) {
@ -779,6 +757,8 @@ class CrdtForeignKeyProjector {
/// authored value of existing rows before planning. When [materialize] is
/// false, only the in-memory plan is computed so a later physical insert is
/// not blocked by this rebuild.
/// Local writes preserve proposed visible unique values for database
/// constraint enforcement; only merge/rebuild passes arbitrate new conflicts.
///
/// [seedTables] names the tables this pass changed. Only their foreign key
/// components are loaded, which is equivalent to a full pass because no
@ -791,6 +771,8 @@ class CrdtForeignKeyProjector {
Set<String>? seedTables,
Set<MergeRowKey>? seedRows,
bool materialize = true,
bool localWrite = false,
Set<MergeRowKey> restoringRows = const {},
}) async {
final state = await _loadProjectionState(
transaction,
@ -834,6 +816,8 @@ class CrdtForeignKeyProjector {
finalHidden: finalHidden,
authoredOverlays: authoredOverlays,
materialize: materialize,
localWrite: localWrite,
restoringRows: restoringRows,
transaction: transaction,
);
}
@ -1557,9 +1541,15 @@ class CrdtForeignKeyProjector {
// parent a row points at now and the one it is about to point at.
Set<Object?> valuesFor(MergeRowKey rowKey, String columnName) {
return <Object?>{
rows[rowKey]?.values[columnName],
canonicalDomainValue(
rows[rowKey]?.values[columnName],
_context.columnsByTableAndName[rowKey.$1]?[columnName],
),
if (unwrittenValues[rowKey]?.containsKey(columnName) ?? false)
unwrittenValues[rowKey]![columnName],
canonicalDomainValue(
unwrittenValues[rowKey]![columnName],
_context.columnsByTableAndName[rowKey.$1]?[columnName],
),
}..remove(null);
}
@ -2101,6 +2091,8 @@ class CrdtForeignKeyProjector {
required Set<MergeRowKey> finalHidden,
required Map<MergeFieldKey, Object?> authoredOverlays,
required bool materialize,
required bool localWrite,
required Set<MergeRowKey> restoringRows,
required Transaction transaction,
}) async {
final authoredByField = _authoredValuesByField(state, authoredOverlays);
@ -2119,6 +2111,25 @@ class CrdtForeignKeyProjector {
}
}
if (localWrite) {
// Keep locally observable unique values until the physical write has
// passed its constraints. An unchanged projected field carries its
// current domain value; a selected field carries the caller's value.
// Resolving a new conflict here would hide a violation from the database.
for (final row in state.rows.values) {
if (state.pendingInsertKeys.contains(row.key)) continue;
for (final column in _uniqueResolver.uniqueColumnNamesFor(row.key.$1)) {
final key = (row.key.$1, row.key.$2, column);
row.values[column] = canonicalDomainValue(
authoredOverlays.containsKey(key)
? authoredOverlays[key]
: state.originalDomain[row.key]?[column],
_context.columnsByTableAndName[row.key.$1]?[column],
);
}
}
}
final uniqueReasons = _uniqueResolver.planUniqueProjection(
valuesByRow: {
for (final row in state.rows.values) row.key: row.values,
@ -2126,10 +2137,13 @@ class CrdtForeignKeyProjector {
crdtRows: {
for (final row in state.rows.values) row.key: row.crdtRow,
},
hidden: finalHidden,
// Restores must pass the physical unique constraint before their
// tombstones are lifted. Releasing these inputs would hide collisions.
hidden: finalHidden.difference(restoringRows),
authoredByField: authoredByField,
claimByField: claimByField,
fieldHlcs: state.fieldHlcs,
resolveVisibleConflicts: !localWrite,
);
final terminalReasons = <MergeFieldKey, CrdtProjectionReason>{
@ -2185,6 +2199,7 @@ class CrdtForeignKeyProjector {
authoredOverlays.containsKey(fieldKey)
? authoredOverlays[fieldKey]
: state.attemptedValues[fieldKey]?.value ?? domainValue,
_context.columnsByTableAndName[row.key.$1]?[columnName],
);
}
}
@ -2517,7 +2532,11 @@ class CrdtForeignKeyProjector {
updates.entries.toList()..sort((a, b) => a.key.compareTo(b.key)),
);
final signature = sortedUpdates.entries
.map((entry) => '${entry.key}:${entry.value.sqlLiteral()}')
.map(
(entry) =>
'${entry.key}:'
'${_context.encodeDomainColumnValue(rowKey.$1, entry.key, entry.value)}',
)
.join('\x1f');
final groupKey = (rowKey.$1, signature);
@ -2687,7 +2706,7 @@ class CrdtForeignKeyProjector {
final attempted = CrdtDataAttemptedValue(
id: current?.id,
fieldId: write.fieldId,
value: canonicalDomainValue(write.value),
value: write.value,
projectionReason: write.reason,
);
if (current == null) {

68
packages/serverpod_offline_sync/lib/src/database/merge_utils/recorder_context.dart
View file

@ -278,6 +278,32 @@ WHERE r."spaceId" = $spaceId AND r."tblId" = $childTableId
);
}
/// Reauthors existing fields at their row's newly persisted insertion HLC.
///
/// This runs before reprojection so unique claims use their new age. Updating
/// in place preserves attempted values whose foreign key cascades on field
/// deletion. [rows] carries the newly stamped metadata returned by the row
/// update, so no field or domain reads are needed. Ordinary field records can
/// be removed once projection is complete.
Future<void> resetReinsertedFieldClocks(
List<CrdtDataRow> rows,
Transaction transaction,
) async {
for (final row in rows) {
await CrdtDataField.db.updateWhere(
databaseSession,
columnValues: (t) => [
t.nodeId(row.nodeId),
t.hlcDatetime(row.hlcDatetime),
t.hlcCounter(row.hlcCounter),
],
where: (t) => t.rowId.equals(row.id),
transaction: transaction,
noReturn: true,
);
}
}
Future<void> recordFieldsUpdatedByTable(
String tableName,
Set<UuidValue> rowIds,
@ -651,6 +677,16 @@ WHERE (${domainColumnPredicate('spaceId', spaceId)})
};
}
/// Encodes model or merge values using the same column-aware encoder as ORM
/// writes. The generated column retains semantics such as DateTime, Duration,
/// binary data, and structured JSON that a SQL storage type alone cannot.
String encodeDomainColumnValue(String tableName, String columnName, Object? value) {
final column = syncTableByName[tableName]!.columns.singleWhere(
(column) => column.columnName == columnName,
);
return ValueEncoder.instance.encodeColumnValue(column, value);
}
Future<void> updateDomainRows(
String tableName,
Set<UuidValue> rowIds,
@ -659,18 +695,12 @@ WHERE (${domainColumnPredicate('spaceId', spaceId)})
) async {
if (rowIds.isEmpty || updates.isEmpty) return;
final columns = {
for (final column in syncTableByName[tableName]!.columns)
column.columnName: column,
};
final assignments = updates.entries
.map((entry) {
final value = ValueEncoder.instance.encodeColumnValue(
columns[entry.key]!,
entry.value,
);
return '"${entry.key.escapeIdentifier()}" = $value';
})
.map(
(e) =>
'"${e.key.escapeIdentifier()}" = '
'${encodeDomainColumnValue(tableName, e.key, e.value)}',
)
.join(', ');
await database.unsafeExecute(
@ -735,10 +765,13 @@ WHERE "id" IN (${rowIds.sqlLiteralList()})
}
case CrdtUniqueConflictReleaseKind.syntheticUuid:
if (value != null) {
return const Uuid().v5obj(
final hash = const Uuid().v5(
Namespace.oid.value,
'$tableName.${column.columnName}:${value}__${releaseSuffix}__$conflictingId',
);
// Keep the deterministic payload, but mark these custom UUIDs with
// version 8 so authored values can be rejected without domain reads.
return UuidValue.fromString(hash.replaceRange(14, 15, '8'));
}
}
@ -879,9 +912,16 @@ WHERE c."id" IN ($whereRowIds)
);
if (result.isEmpty) return null;
final foreignKeyIndex = result.first[0] as int;
return (
foreignKeyIndex: result.first[0] as int,
value: result.first[1] as Object,
foreignKeyIndex: foreignKeyIndex,
// Raw SQLite UUID columns are blobs; report the authored identifier.
value: canonicalDomainValue(
result.first[1],
columnsByTableAndName[childTableName]?[foreignKeys[foreignKeyIndex]
.columns
.single],
)!,
);
}

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

@ -1,9 +1,11 @@
import 'package:meta/meta.dart';
import 'package:serverpod_database/serverpod_database.dart' show ColumnType;
import 'package:serverpod_serialization/serverpod_serialization.dart';
import '../../crdt/extensions.dart';
import '../../generated/protocol.dart';
import '../../hlc/hlc.dart';
import '../exceptions.dart';
import '../unique_index_utils.dart';
import 'database_helpers.dart';
import 'recorder_context.dart';
@ -21,6 +23,73 @@ class CrdtUniqueConflictResolver {
final CrdtRecorderContext _context;
static final _reservedTextSuffix = RegExp(
'__(?:conflict|hidden|park)__[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-'
r'[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$',
);
final _uniqueTextColumns = <String, Set<String>>{};
final _syntheticUuidColumns = <String, Set<String>>{};
Set<String> _textColumnsFor(String tableName) => _uniqueTextColumns.putIfAbsent(
tableName,
() => {
for (final column in uniqueColumnNamesFor(tableName))
if (_context.columnsByTableAndName[tableName]?[column]?.columnType ==
ColumnType.text)
column,
},
);
Set<String> _uuidColumnsFor(String tableName) => _syntheticUuidColumns.putIfAbsent(
tableName,
() => {
for (final index in uniqueIndexesFor(tableName))
for (final column in index.releaseColumns)
if (column.kind == CrdtUniqueConflictReleaseKind.syntheticUuid)
column.columnName,
},
);
/// Whether authored values need a generated-namespace check.
bool hasReservedUniqueColumns(String tableName) =>
_textColumnsFor(tableName).isNotEmpty || _uuidColumnsFor(tableName).isNotEmpty;
/// Rejects authored values in the namespace owned by deterministic projection.
/// This checks the schema and input only, never the currently occupied names.
void validateAuthoredValue(String tableName, String columnName, Object? value) {
if (!isReservedValue(tableName, columnName, value)) return;
throw OfflineSyncReservedValueException(
tableName: tableName,
columnName: columnName,
value: value.toString(),
);
}
/// Generated text suffixes and version-8 UUIDs belong only to projection.
/// UUID checks apply only where the schema uses synthetic UUID release,
/// leaving primary keys, foreign keys, and nullable UUID columns unrestricted.
bool isReservedValue(String tableName, String columnName, Object? value) {
if (value is String &&
value.contains('__') &&
_textColumnsFor(tableName).contains(columnName) &&
_reservedTextSuffix.hasMatch(value)) {
return true;
}
if ((value is UuidValue || value is String) &&
_uuidColumnsFor(tableName).contains(columnName)) {
final uuid = value.toString();
return uuid.length == 36 && uuid[14] == '8';
}
return false;
}
/// Validates a group of newly authored field facts.
void validateAuthoredFields(Map<MergeFieldKey, Object?> values) {
for (final MapEntry(key: key, value: value) in values.entries) {
validateAuthoredValue(key.$1, key.$3, value);
}
}
/// Whether [tableName] has any synchronized unique index.
bool tableHasUniqueIndexes(String tableName) =>
uniqueIndexesFor(tableName).isNotEmpty;
@ -77,6 +146,7 @@ class CrdtUniqueConflictResolver {
required Map<MergeFieldKey, Object?> authoredByField,
required Map<MergeFieldKey, Object?> claimByField,
required Map<MergeFieldKey, Hlc> fieldHlcs,
bool resolveVisibleConflicts = true,
}) {
final reasons = <MergeFieldKey, CrdtProjectionReason>{};
final tableNames = {for (final rowKey in valuesByRow.keys) rowKey.$1};
@ -107,6 +177,9 @@ class CrdtUniqueConflictResolver {
);
}
// Local writes retain their proposed unique values so the database
// enforces uniqueness. Hidden rows still release their old values.
if (!resolveVisibleConflicts) continue;
final groups = <String, List<MergeRowKey>>{};
for (final rowKey in tableRowKeys) {
if (hidden.contains(rowKey)) continue;
@ -250,6 +323,7 @@ class CrdtUniqueConflictResolver {
@internal
String canonicalProjectionValue(Object? value) {
if (value == null) return '';
if (value is String) return value;
final uuid = tryUuidValue(value);
if (uuid != null) return uuid.uuid;
return value.toString();
@ -259,6 +333,7 @@ String canonicalProjectionValue(Object? value) {
@internal
bool projectionValuesEqual(Object? left, Object? right) {
if (left == null || right == null) return left == right;
if (left is String && right is String) return left == right;
if (left is Map && right is Map) {
return left.length == right.length &&
left.entries.every(

224
packages/serverpod_offline_sync/lib/src/database/recorder.dart
View file

@ -372,8 +372,10 @@ class CrdtMutationRecorder {
);
}
/// Returns whether primary-key upserts preserve the existing projection.
Future<bool> prepareLocalUpsert<T extends TableRow>(
/// Captures the materialized values before an upsert so its accepted writes
/// can replace old attempted values without authoring unchanged projections.
Future<({bool projectionUnchanged, Map<MergeRowKey, Map<String, Object?>> domain})>
prepareLocalUpsert<T extends TableRow>(
List<T> rows,
List<Column> conflictColumns,
List<Column>? updateColumns,
@ -384,15 +386,17 @@ class CrdtMutationRecorder {
for (final row in rows)
if (row.id case final UuidValue rowId) rowId,
};
if (conflictColumns.length == 1 &&
conflictColumns.single.columnName == 'id' &&
rowIds.length == rows.length) {
final targetsPrimaryKey =
conflictColumns.length == 1 && conflictColumns.single.columnName == 'id';
const emptyDomain = <MergeRowKey, Map<String, Object?>>{};
if (targetsPrimaryKey && rowIds.length == rows.length) {
final stored = await _context.findCrdtRows(tableName, rowIds, transaction);
if (stored.isEmpty &&
!_foreignKeys.tablesWithDefaultDependencies.contains(tableName)) {
// The full pass has no persisted seed or default to load. Classification
// after the physical upsert still reads metadata again.
return false;
validateAuthoredRows(rows, null);
return (projectionUnchanged: false, domain: emptyDomain);
}
if (stored.length == rowIds.length &&
await _foreignKeyProjector.canLeaveLocalProjectionUnchanged(
@ -401,20 +405,82 @@ class CrdtMutationRecorder {
inserting: false,
columns: updateColumns,
)) {
return true;
validateAuthoredRows(rows, updateColumns);
return (projectionUnchanged: true, domain: emptyDomain);
}
}
await projectCurrent(tableName, rowIds, transaction);
return false;
if (!targetsPrimaryKey && _foreignKeyProjector.needsProjection(tableName, null)) {
final suppliedValues = [for (final row in rows) row.toJsonForDatabase() as Map];
rowIds.addAll(
await _context.findDomainRowIdsWhereColumnsIn(
tableName: tableName,
valuesByColumn: {
for (final column in conflictColumns)
if (column.columnName != 'spaceId')
column.columnName: {
for (final values in suppliedValues)
canonicalDomainValue(
values[column.columnName],
_context.columnsByTableAndName[tableName]?[column.columnName],
),
},
},
transaction: transaction,
),
);
}
final projected = await _foreignKeyProjector.project(
transaction,
seedTables: {tableName},
// Other conflict targets can update an existing row with a different id.
seedRows: {for (final rowId in rowIds) (tableName, rowId)},
);
if (_uniqueResolver.hasReservedUniqueColumns(tableName)) {
final written = updateColumns?.map((column) => column.columnName).toSet();
for (final row in rows) {
final supplied = _authoredValuesFromRow(row, null);
for (final MapEntry(key: column, value: value) in supplied.entries) {
if (!_uniqueResolver.isReservedValue(tableName, column, value)) continue;
final before =
projected.domain[(tableName, row.id)] ??
projected.domain.entries
.where(
(entry) =>
entry.key.$1 == tableName &&
conflictColumns.every(
(key) =>
key.columnName == 'spaceId' ||
(key.columnName == 'id'
? entry.key.$2 == row.id
: projectionValuesEqual(
entry.value[key.columnName],
supplied[key.columnName],
)),
),
)
.firstOrNull
?.value;
if (before != null &&
((written != null && !written.contains(column)) ||
projectionValuesEqual(before[column], value))) {
continue;
}
// A new claim must fail before the physical upsert can collide with
// another row's generated value. Unchanged echoes are checked after
// the write, when we know whether it updated or restored the row.
_uniqueResolver.validateAuthoredValue(tableName, column, value);
}
}
}
return (projectionUnchanged: false, domain: projected.domain);
}
/// Plans FK/unique projection for rows that are about to be inserted.
///
/// Releases hidden unique claims first and returns copies whose unique/FK
/// columns already hold the planned domain values, so the physical write
/// cannot violate an immediate unique index. Authored values that differ
/// from the planned domain are returned so they can be stored as attempted
/// values after the insert.
/// Releases hidden unique claims first and plans foreign key repairs while
/// preserving submitted unique values for database constraint enforcement.
/// Authored values that differ from the planned domain are returned so they
/// can be stored as attempted values after the insert.
Future<({List<T> rows, ProjectionAttemptsByField attempts})>
planLocalInserts<T extends TableRow>(
List<T> rows,
@ -428,6 +494,8 @@ class CrdtMutationRecorder {
return (rows: rows, attempts: unplanned);
}
validateAuthoredRows(rows, null);
if (await _foreignKeyProjector.canLeaveLocalProjectionUnchanged(
rows,
transaction,
@ -459,6 +527,7 @@ class CrdtMutationRecorder {
final planned = await _foreignKeyProjector.project(
transaction,
pendingInserts: pending,
localWrite: true,
seedTables: {tableName},
seedRows: {for (final row in pending) (tableName, row.rowId)},
);
@ -496,8 +565,9 @@ class CrdtMutationRecorder {
planLocalUpdates<T extends TableRow>(
List<T> rows,
List<Column>? columns,
Transaction transaction,
) async {
Transaction transaction, {
bool restoring = false,
}) async {
if (rows.isEmpty) return (rows: rows, projectionUnchanged: true);
final tableName = rows.first.table.tableName;
if (!_context.isCrdtTrackedTableName(tableName)) {
@ -524,9 +594,7 @@ class CrdtMutationRecorder {
row,
columns,
).entries)
(tableName, row.id as UuidValue, columnName): canonicalDomainValue(
value,
),
(tableName, row.id as UuidValue, columnName): value,
};
var overlays = authored;
if (columns == null) {
@ -548,9 +616,15 @@ class CrdtMutationRecorder {
fieldKey: value,
};
}
_uniqueResolver.validateAuthoredFields(overlays);
final planned = await _foreignKeyProjector.project(
transaction,
authoredOverlays: overlays,
localWrite: true,
restoringRows: {
if (restoring)
for (final row in rows) (tableName, row.id as UuidValue),
},
seedTables: {tableName},
seedRows: {
for (final row in rows)
@ -569,6 +643,30 @@ class CrdtMutationRecorder {
);
}
/// Validates unprojected ORM writes, or projected inserts with their attempts.
void validateAuthoredRows<T extends TableRow>(
List<T> rows,
List<Column>? columns, {
ProjectionAttemptsByField attempts = const {},
}) {
if (rows.isEmpty) return;
final tableName = rows.first.table.tableName;
if (!_uniqueResolver.hasReservedUniqueColumns(tableName)) return;
for (final row in rows) {
for (final MapEntry(key: columnName, value: value) in _authoredValuesFromRow(
row,
columns,
).entries) {
final key = (tableName, row.id, columnName);
_uniqueResolver.validateAuthoredValue(
tableName,
columnName,
attempts.containsKey(key) ? attempts[key]!.value : value,
);
}
}
}
Map<String, Object?> _authoredValuesFromRow<T extends TableRow>(
T row,
List<Column>? columns,
@ -580,7 +678,11 @@ class CrdtMutationRecorder {
column.columnName,
];
return {
for (final columnName in columnNames) columnName: json[columnName],
for (final columnName in columnNames)
columnName: canonicalDomainValue(
json[columnName],
_context.columnsByTableAndName[row.table.tableName]?[columnName],
),
};
}
@ -659,6 +761,7 @@ class CrdtMutationRecorder {
Transaction transaction, {
ProjectionAttemptsByField attempts = const {},
}) async {
validateAuthoredRows(insertedRows, null, attempts: attempts);
await _forTrackedRows(insertedRows, transaction, (
tableName,
rowIds,
@ -700,39 +803,36 @@ class CrdtMutationRecorder {
rowIds,
hlcManager,
) async {
final crdtDataRows = await _context.findRequiredCrdtRows(
tableName,
rowIds,
'reinserted',
final crdtDataRows = await _touchCrdtRows(
await _context.findRequiredCrdtRows(
tableName,
rowIds,
'reinserted',
transaction,
),
hlcManager,
transaction,
);
await _touchCrdtRows(crdtDataRows, hlcManager, transaction);
// A reinsert authors the complete row at its new insertion timestamp.
// Retain attempted values while projecting, but discard their old ages.
await _context.resetReinsertedFieldClocks(crdtDataRows, transaction);
await _context.markCrdtRowsDeleted(
crdtDataRows,
false,
CrdtDataDeletedReason.userReinsert,
transaction,
);
// Every field to skip carries an attempted value: a repaired foreign key
// and a restored authored value are both subsets of that set, over a
// subset of its columns.
final skippedFields = await _foreignKeyProjector.findActiveAttemptedFields(
tableName: tableName,
rowIds: rowIds,
transaction: transaction,
);
// Reinsert is a full-row passthrough. Projected unique/FK columns must
// keep their original claim HLC so restoration can reclaim or lose
// without authoring a newer unique claim.
await _recordUpdatedFields(
reinsertedRows,
crdtDataRows,
null,
transaction,
skippedFields: skippedFields,
);
await _maybeProject(tableName, rowIds, null, transaction);
// Only projected fields need a metadata row to hold their authored value.
// All other fields now inherit the insertion timestamp implicitly.
await CrdtDataField.db.deleteWhere(
_session,
where: (t) =>
t.rowId.inSet(crdtDataRows.map((row) => row.id!).toSet()) &
t.attemptedValue.id.equals(null),
transaction: transaction,
noReturn: true,
);
});
}
@ -742,6 +842,7 @@ class CrdtMutationRecorder {
List<Column>? columns,
Transaction transaction, {
bool projectionUnchanged = false,
Map<MergeRowKey, Map<String, Object?>> domainBeforeUpsert = const {},
}) async {
await _foreignKeyProjector.assertVisibleTargets(updatedRows, columns, transaction);
@ -756,6 +857,24 @@ class CrdtMutationRecorder {
'updated',
transaction,
);
// Only rows returned by the physical upsert were accepted by updateWhere.
// Explicit columns author even an unchanged null; full-row passthrough
// keeps the claim behind an unchanged displayed alternative.
final authoredOverlays = <MergeFieldKey, Object?>{
for (final row in updatedRows)
if (domainBeforeUpsert.containsKey((tableName, row.id)))
for (final MapEntry(key: columnName, value: value)
in _authoredValuesFromRow(row, columns).entries)
if ((domainBeforeUpsert[(tableName, row.id)]?.containsKey(columnName) ??
false) &&
(columns != null ||
!projectionValuesEqual(
domainBeforeUpsert[(tableName, row.id)]![columnName],
value,
)))
(tableName, row.id as UuidValue, columnName): value,
};
_uniqueResolver.validateAuthoredFields(authoredOverlays);
final implicitForeignKeyRepairFields = columns == null
? await _foreignKeyProjector.findImplicitRepairFields(
tableName: tableName,
@ -768,7 +887,9 @@ class CrdtMutationRecorder {
crdtDataRows,
columns,
transaction,
skippedFields: implicitForeignKeyRepairFields,
skippedFields: implicitForeignKeyRepairFields.difference(
authoredOverlays.keys.toSet(),
),
);
if (projectionUnchanged &&
await _foreignKeyProjector.canLeaveLocalProjectionUnchanged(
@ -785,6 +906,7 @@ class CrdtMutationRecorder {
rowIds,
updatedColumnNames,
transaction,
authoredOverlays: authoredOverlays,
);
});
}
@ -793,13 +915,15 @@ class CrdtMutationRecorder {
String tableName,
Set<UuidValue> rowIds,
Set<String>? columnNames,
Transaction transaction,
) async {
Transaction transaction, {
Map<MergeFieldKey, Object?> authoredOverlays = const {},
}) async {
if (_foreignKeyProjector.needsProjection(tableName, columnNames)) {
await _foreignKeyProjector.project(
transaction,
seedTables: {tableName},
seedRows: {for (final rowId in rowIds) (tableName, rowId)},
authoredOverlays: authoredOverlays,
);
}
}
@ -843,14 +967,14 @@ class CrdtMutationRecorder {
);
}
Future<void> _touchCrdtRows(
Future<List<CrdtDataRow>> _touchCrdtRows(
List<CrdtDataRow> rows,
HlcManager hlcManager,
Transaction transaction,
) async {
if (rows.isEmpty) return;
if (rows.isEmpty) return [];
await CrdtDataRow.db.update(
return CrdtDataRow.db.update(
_session,
[for (final row in rows) _context.withNextHlc(row, hlcManager)],
columns: (t) => [t.nodeId, t.hlcDatetime, t.hlcCounter],

4
packages/serverpod_offline_sync/lib/src/sync/engine.dart
View file

@ -900,7 +900,7 @@ class OfflineSyncEngine {
return (
exists: true,
ownerSpaceId: spaceId,
value: canonicalDomainValue(attempted.value),
value: attempted.value,
);
}
@ -1208,7 +1208,7 @@ extension on Map<String, dynamic> {
for (final field in attemptedValueFields) {
final attempted = field.attemptedValue;
if (attempted == null) continue;
this[field.column!.name] = canonicalDomainValue(attempted.value);
this[field.column!.name] = attempted.value;
}
}
}