test: Add a DST and tests reproducing found failures (#110)
Co-authored-by: Marcelo Soares <marcelo.me.soares@gmail.com>
This commit is contained in:
parent
fbad5c61fe
commit
c78af247fe
8 changed files with 22825 additions and 184 deletions
|
|
@ -1,3 +1,4 @@
|
|||
import 'package:clock/clock.dart';
|
||||
import 'package:serverpod_database/serverpod_database.dart';
|
||||
|
||||
import '../generated/protocol.dart';
|
||||
|
|
@ -20,7 +21,7 @@ Future<CrdtSyncIntegrityViolation> recordCrdtSyncIntegrityViolation(
|
|||
transaction: transaction,
|
||||
);
|
||||
|
||||
final now = DateTime.now().toUtc();
|
||||
final now = clock.now().toUtc();
|
||||
final newViolation = existing != null
|
||||
? existing.copyWith(
|
||||
crdtDataRowId: violation.crdtDataRowId ?? existing.crdtDataRowId,
|
||||
|
|
|
|||
|
|
@ -1,5 +1,6 @@
|
|||
import 'dart:async';
|
||||
|
||||
import 'package:clock/clock.dart';
|
||||
import 'package:serverpod_database/serverpod_database.dart';
|
||||
import 'package:uuid/uuid.dart';
|
||||
|
||||
|
|
@ -463,7 +464,7 @@ class CrdtSync {
|
|||
if (role.canWrite) return;
|
||||
|
||||
final firstChange = changes.first;
|
||||
final now = DateTime.now().toUtc();
|
||||
final now = clock.now().toUtc();
|
||||
final violation = CrdtSyncIntegrityViolation(
|
||||
type: CrdtSyncViolationType.unauthorizedWrite,
|
||||
domainTableName: firstChange.tableName,
|
||||
|
|
@ -963,7 +964,7 @@ class CrdtSync {
|
|||
session,
|
||||
pending.ownerScopeId,
|
||||
);
|
||||
final now = DateTime.now().toUtc();
|
||||
final now = clock.now().toUtc();
|
||||
final violation = CrdtSyncIntegrityViolation(
|
||||
type: pending.type,
|
||||
domainTableName: pending.tableName,
|
||||
|
|
|
|||
|
|
@ -316,6 +316,7 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
|
|||
{insert.uuidRowId},
|
||||
savepoint,
|
||||
safeInsert.attempts,
|
||||
mergeCache: (fields: context.fields, node: remoteNode),
|
||||
);
|
||||
return insertedRow;
|
||||
},
|
||||
|
|
@ -897,7 +898,7 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
|
|||
CrdtNode? node,
|
||||
Hlc? hlc,
|
||||
}) async {
|
||||
final now = DateTime.now().toUtc();
|
||||
final now = clock.now().toUtc();
|
||||
throw CrdtSyncIntegrityViolationException(
|
||||
CrdtSyncIntegrityViolation(
|
||||
type: CrdtSyncViolationType.ownershipCollision,
|
||||
|
|
|
|||
|
|
@ -276,12 +276,26 @@ class CrdtForeignKeyProjector {
|
|||
await _upsertProjections(projectionWrites, transaction);
|
||||
}
|
||||
|
||||
/// Records the foreign key attempts carried by an incoming insert.
|
||||
///
|
||||
/// Foreign key columns keep their attempted value in [CrdtDataForeignKey],
|
||||
/// which hangs off a [CrdtDataField], so an insert has to materialize field
|
||||
/// metadata for them even though no update has happened yet.
|
||||
///
|
||||
/// [mergeCache] is the merge context's field cache and the node authoring the
|
||||
/// batch. The cache is loaded once before a batch is applied, so without this
|
||||
/// it would not know about the fields created here, and a later change to the
|
||||
/// same column in the same batch would take the "no field yet" path and
|
||||
/// insert a second row for the same `(rowId, columnId)`.
|
||||
///
|
||||
/// The local write path passes nothing: it has no batch to cache for.
|
||||
Future<void> recordInsertAttempts(
|
||||
String tableName,
|
||||
Set<UuidValue> rowIds,
|
||||
Transaction transaction,
|
||||
ForeignKeyAttemptsByField attemptedValues,
|
||||
) async {
|
||||
ForeignKeyAttemptsByField attemptedValues, {
|
||||
MergeFieldCache? mergeCache,
|
||||
}) async {
|
||||
if (rowIds.isEmpty) return;
|
||||
|
||||
final foreignKeyColumns = _foreignKeys.foreignKeyColumnsFor(tableName);
|
||||
|
|
@ -324,7 +338,10 @@ class CrdtForeignKeyProjector {
|
|||
columnNames: columnIds.keys.toSet(),
|
||||
transaction: transaction,
|
||||
);
|
||||
final fieldsToInsert = <CrdtDataField>[];
|
||||
// Field and key travel together so the write-back cannot pair them up
|
||||
// wrongly; zipping two lists by index would quietly depend on `insert`
|
||||
// returning rows in the order it was given them.
|
||||
final pending = <({CrdtDataField field, MergeFieldKey key})>[];
|
||||
for (final crdtRow in crdtRows) {
|
||||
final values = valuesByRowId[crdtRow.uuidRowId];
|
||||
if (values == null) continue;
|
||||
|
|
@ -335,24 +352,35 @@ class CrdtForeignKeyProjector {
|
|||
continue;
|
||||
}
|
||||
|
||||
fieldsToInsert.add(
|
||||
CrdtDataField(
|
||||
pending.add((
|
||||
field: CrdtDataField(
|
||||
rowId: crdtRow.id!,
|
||||
columnId: columnId,
|
||||
nodeId: crdtRow.nodeId,
|
||||
hlcDatetime: crdtRow.hlcDatetime,
|
||||
hlcCounter: crdtRow.hlcCounter,
|
||||
),
|
||||
);
|
||||
key: (tableName, crdtRow.uuidRowId, columnName),
|
||||
));
|
||||
}
|
||||
}
|
||||
|
||||
if (fieldsToInsert.isNotEmpty) {
|
||||
await CrdtDataField.db.insert(
|
||||
if (pending.isNotEmpty) {
|
||||
final insertedFields = await CrdtDataField.db.insert(
|
||||
_context.databaseSession,
|
||||
fieldsToInsert,
|
||||
[for (final entry in pending) entry.field],
|
||||
transaction: transaction,
|
||||
);
|
||||
if (mergeCache != null) {
|
||||
for (final (index, field) in insertedFields.indexed) {
|
||||
// The node is attached because a cached field is read back through
|
||||
// `CrdtDataField.hlc`, which resolves the node's uuid and throws
|
||||
// without it.
|
||||
mergeCache.fields[pending[index].key] = field.copyWith(
|
||||
node: mergeCache.node,
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
await recordAttemptsForRows(
|
||||
|
|
@ -971,7 +999,7 @@ class CrdtForeignKeyProjector {
|
|||
for (final child
|
||||
in state.rowsByTable[edge.childTableName] ??
|
||||
const <_ProjectedForeignKeyRow>[]) {
|
||||
if (finalHidden.contains(child.key)) continue;
|
||||
final isHidden = finalHidden.contains(child.key);
|
||||
|
||||
final attemptedValue = _attemptedValue(child, edge, state);
|
||||
final currentVisibleValue = child.values[edge.childColumn].toUuidValue();
|
||||
|
|
@ -1015,6 +1043,31 @@ class CrdtForeignKeyProjector {
|
|||
}
|
||||
}
|
||||
|
||||
// A hidden row is repaired like any other, so replicas that hid it by
|
||||
// different routes still agree about its columns. Two values it must
|
||||
// not materialize, though:
|
||||
//
|
||||
// - one whose parent row is absent entirely, which the physical
|
||||
// foreign key constraint rejects - that is why a row whose parent is
|
||||
// missing is hidden with its safe value left in place;
|
||||
// - one on a unique-indexed column that it would be adopting *back*
|
||||
// after a repair. A hidden row is invisible to unique conflict
|
||||
// resolution but still occupies the physical unique index, so
|
||||
// re-adopting the value when the parent returns wedges every later
|
||||
// legitimate claim on it.
|
||||
if (isHidden && desiredVisibleValue != null) {
|
||||
if (_parentRowForValue(edge, desiredVisibleValue, state) == null) {
|
||||
continue;
|
||||
}
|
||||
if (overrideReason == null &&
|
||||
_uniqueResolver.isUniqueIndexedColumn(
|
||||
edge.childTableName,
|
||||
edge.childColumn,
|
||||
)) {
|
||||
continue;
|
||||
}
|
||||
}
|
||||
|
||||
if (!currentVisibleValue.sameUuidValue(desiredVisibleValue)) {
|
||||
child.values[edge.childColumn] = desiredVisibleValue;
|
||||
changedUpdatesByRow.putIfAbsent(child.key, () => {})[edge.childColumn] =
|
||||
|
|
|
|||
|
|
@ -23,6 +23,18 @@ typedef MergeContext = ({
|
|||
Map<MergeRowKey, DomainRowOwner> domainOwners,
|
||||
});
|
||||
|
||||
/// The merge context's field cache, with the node authoring the batch.
|
||||
///
|
||||
/// The two are one parameter because they are only meaningful together: a
|
||||
/// cached field is read back through `CrdtDataField.hlc`, which resolves the
|
||||
/// node's uuid and throws without it. Separate optional parameters would allow
|
||||
/// a cache with no node, whose entries throw when read.
|
||||
@internal
|
||||
typedef MergeFieldCache = ({
|
||||
Map<MergeFieldKey, CrdtDataField> fields,
|
||||
CrdtNode node,
|
||||
});
|
||||
|
||||
/// Remote nodes and scope-node rows loaded for a merge.
|
||||
@internal
|
||||
typedef MergeNodes = ({
|
||||
|
|
|
|||
|
|
@ -378,6 +378,13 @@ class CrdtUniqueConflictResolver {
|
|||
);
|
||||
}
|
||||
|
||||
/// Whether [columnName] participates in any unique index of [tableName].
|
||||
///
|
||||
/// Foreign key projection needs this to know when materializing a value onto
|
||||
/// a hidden row would occupy the physical unique index.
|
||||
bool isUniqueIndexedColumn(String tableName, String columnName) =>
|
||||
_isUniqueIndexedColumn(tableName, columnName);
|
||||
|
||||
bool _isUniqueIndexedColumn(String tableName, String columnName) {
|
||||
final tableDefinition = _context.tableDefinitionsByName[tableName];
|
||||
if (tableDefinition == null) return false;
|
||||
|
|
|
|||
|
|
@ -1,3 +1,4 @@
|
|||
import 'package:clock/clock.dart';
|
||||
import 'package:meta/meta.dart';
|
||||
import 'package:serverpod_database/serverpod_database.dart';
|
||||
import 'package:serverpod_serialization/serverpod_serialization.dart';
|
||||
|
|
|
|||
|
|
@ -75,7 +75,7 @@ class Hlc implements Comparable<Hlc> {
|
|||
factory Hlc.zero(UuidValue nodeId) => Hlc(DateTime.utc(1970), 0, nodeId);
|
||||
|
||||
/// Instantiates an Hlc using the wall clock.
|
||||
factory Hlc.now(UuidValue nodeId) => Hlc(DateTime.now(), 0, nodeId);
|
||||
factory Hlc.now(UuidValue nodeId) => Hlc(clock.now(), 0, nodeId);
|
||||
|
||||
/// Parses the node segment of an HLC string: a UUID, or legacy `uuid:workerId`.
|
||||
static UuidValue parseNodeSegment(String segment) {
|
||||
|
|
|
|||
Loading…
Reference in a new issue