Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
f062725a16 |
|||
|
9495f616c4 |
|||
|
b5450353b9 |
|||
|
0345f56787 |
11 changed files with 719 additions and 14 deletions
|
|
@ -120,6 +120,7 @@ class OfflineSyncDatabase implements Database {
|
||||||
UuidValue? userId,
|
UuidValue? userId,
|
||||||
bool once = false,
|
bool once = false,
|
||||||
OfflineSyncOnMergeSuccess? onMergeSuccess,
|
OfflineSyncOnMergeSuccess? onMergeSuccess,
|
||||||
|
OfflineSyncOnInboundMerge? onInboundMerge,
|
||||||
}) async* {
|
}) async* {
|
||||||
await _ensureInitialized();
|
await _ensureInitialized();
|
||||||
final effectiveUserId = await _requireUserId(userId);
|
final effectiveUserId = await _requireUserId(userId);
|
||||||
|
|
@ -129,6 +130,7 @@ class OfflineSyncDatabase implements Database {
|
||||||
inbound: inbound,
|
inbound: inbound,
|
||||||
once: once,
|
once: once,
|
||||||
onMergeSuccess: onMergeSuccess,
|
onMergeSuccess: onMergeSuccess,
|
||||||
|
onInboundMerge: onInboundMerge,
|
||||||
mode: mode,
|
mode: mode,
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -36,7 +36,7 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
|
||||||
// Nothing in this batch touches a foreign key or a unique index, so no
|
// Nothing in this batch touches a foreign key or a unique index, so no
|
||||||
// projection decision can change: skip both passes instead of loading
|
// projection decision can change: skip both passes instead of loading
|
||||||
// state for tables that cannot produce a candidate or a conflict.
|
// state for tables that cannot produce a candidate or a conflict.
|
||||||
final mayAffectProjection = _foreignKeyProjector.mergeOperationsMayAffectProjection(
|
var mayAffectProjection = _foreignKeyProjector.mergeOperationsMayAffectProjection(
|
||||||
operations,
|
operations,
|
||||||
);
|
);
|
||||||
final authoredOverlays = <MergeFieldKey, Object?>{};
|
final authoredOverlays = <MergeFieldKey, Object?>{};
|
||||||
|
|
@ -48,6 +48,14 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
|
||||||
);
|
);
|
||||||
final nodesByUuid = remoteNodes.nodesByUuid;
|
final nodesByUuid = remoteNodes.nodesByUuid;
|
||||||
final context = await _loadMergeContext(mergeSet, transaction);
|
final context = await _loadMergeContext(mergeSet, transaction);
|
||||||
|
// New rows whose parents are there and which nothing refers to yet (a
|
||||||
|
// rating, an edit, a tag given, a card added) can't change a projection
|
||||||
|
// decision either, though they carry foreign keys: proven so, neither
|
||||||
|
// pass runs, and the batch still resolves their parents in one go.
|
||||||
|
final unchangedByInserts =
|
||||||
|
mayAffectProjection &&
|
||||||
|
await _mergeLeavesProjectionUnchanged(operations, context, transaction);
|
||||||
|
if (unchangedByInserts) mayAffectProjection = false;
|
||||||
|
|
||||||
// Deletes carry no authored overlay but still change visibility, so the
|
// Deletes carry no authored overlay but still change visibility, so the
|
||||||
// seed is every touched row, not just the ones with overlays.
|
// seed is every touched row, not just the ones with overlays.
|
||||||
|
|
@ -59,7 +67,7 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
|
||||||
touchedRows.add((operation.tableName, operation.uuidRowId));
|
touchedRows.add((operation.tableName, operation.uuidRowId));
|
||||||
}
|
}
|
||||||
|
|
||||||
final batch = mayAffectProjection
|
final batch = mayAffectProjection || unchangedByInserts
|
||||||
? await _planBatchInserts(
|
? await _planBatchInserts(
|
||||||
operations,
|
operations,
|
||||||
nodesByUuid,
|
nodesByUuid,
|
||||||
|
|
@ -67,6 +75,7 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
|
||||||
touchedTables,
|
touchedTables,
|
||||||
touchedRows,
|
touchedRows,
|
||||||
transaction,
|
transaction,
|
||||||
|
plan: mayAffectProjection,
|
||||||
)
|
)
|
||||||
: (plan: null, pending: const <PendingProjectionRow>[]);
|
: (plan: null, pending: const <PendingProjectionRow>[]);
|
||||||
// Every merged insert has to know whether the rows its foreign keys name
|
// Every merged insert has to know whether the rows its foreign keys name
|
||||||
|
|
@ -362,8 +371,11 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
|
||||||
MergeContext context,
|
MergeContext context,
|
||||||
Set<String> seedTables,
|
Set<String> seedTables,
|
||||||
Set<MergeRowKey> seedRows,
|
Set<MergeRowKey> seedRows,
|
||||||
Transaction transaction,
|
Transaction transaction, {
|
||||||
) async {
|
// Whether to plan them: without, only the rows are listed, for their
|
||||||
|
// parents to be resolved together.
|
||||||
|
bool plan = true,
|
||||||
|
}) async {
|
||||||
final pending = <PendingProjectionRow>[];
|
final pending = <PendingProjectionRow>[];
|
||||||
final seenRowKeys = <MergeRowKey>{};
|
final seenRowKeys = <MergeRowKey>{};
|
||||||
|
|
||||||
|
|
@ -391,7 +403,7 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
|
||||||
hidden: false,
|
hidden: false,
|
||||||
));
|
));
|
||||||
}
|
}
|
||||||
if (pending.isEmpty) return (plan: null, pending: pending);
|
if (pending.isEmpty || !plan) return (plan: null, pending: pending);
|
||||||
|
|
||||||
return (
|
return (
|
||||||
plan: await _foreignKeyProjector.project(
|
plan: await _foreignKeyProjector.project(
|
||||||
|
|
@ -404,6 +416,49 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Whether this batch leaves every projection decision as it is though it
|
||||||
|
/// touches foreign keys: it only inserts new rows, each once, and updates
|
||||||
|
/// columns no projection reads, and the inserts are proven
|
||||||
|
/// ([CrdtForeignKeyProjector.canLeaveProjectionUnchangedByMergedInserts]).
|
||||||
|
/// A delete, or an insert onto a row there already, takes the full pass.
|
||||||
|
Future<bool> _mergeLeavesProjectionUnchanged(
|
||||||
|
List<CrdtMergeChange> operations,
|
||||||
|
MergeContext context,
|
||||||
|
Transaction transaction,
|
||||||
|
) async {
|
||||||
|
final insertsByTable = <String, Map<UuidValue, Map<String, Object?>>>{};
|
||||||
|
final positions = <MergeRowKey, int>{};
|
||||||
|
for (final operation in operations) {
|
||||||
|
if (!_context.isCrdtTrackedTableName(operation.tableName)) continue;
|
||||||
|
switch (operation) {
|
||||||
|
case CrdtMergeDelete():
|
||||||
|
return false;
|
||||||
|
case final CrdtMergeUpdate update:
|
||||||
|
if (_foreignKeyProjector.needsProjection(update.tableName, {
|
||||||
|
update.columnName,
|
||||||
|
})) {
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
case final CrdtMergeInsert insert:
|
||||||
|
if (context.rows[(insert.tableName, insert.uuidRowId)] != null) {
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
final rows = insertsByTable[insert.tableName] ??= {};
|
||||||
|
if (rows.containsKey(insert.uuidRowId)) return false;
|
||||||
|
rows[insert.uuidRowId] = _sanitizeMergeRowData(
|
||||||
|
insert.tableName,
|
||||||
|
insert.databaseColumns,
|
||||||
|
);
|
||||||
|
positions[(insert.tableName, insert.uuidRowId)] = positions.length;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return _foreignKeyProjector.canLeaveProjectionUnchangedByMergedInserts(
|
||||||
|
insertsByTable,
|
||||||
|
positions,
|
||||||
|
transaction,
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
Future<void> _applyMergeInsert(
|
Future<void> _applyMergeInsert(
|
||||||
CrdtMergeInsert insert,
|
CrdtMergeInsert insert,
|
||||||
Map<UuidValue, CrdtNode> remoteNodes,
|
Map<UuidValue, CrdtNode> remoteNodes,
|
||||||
|
|
|
||||||
|
|
@ -33,6 +33,15 @@ String domainColumnPredicate(
|
||||||
String alias = 'd',
|
String alias = 'd',
|
||||||
}) => '$alias."${columnName.escapeIdentifier()}" = ${value.sqlLiteral()}';
|
}) => '$alias."${columnName.escapeIdentifier()}" = ${value.sqlLiteral()}';
|
||||||
|
|
||||||
|
/// SQL predicate matching rows where [columnName] holds one of [values],
|
||||||
|
/// which must not be empty.
|
||||||
|
@internal
|
||||||
|
String domainColumnInPredicate(
|
||||||
|
String columnName,
|
||||||
|
Iterable<Object?> values, {
|
||||||
|
String alias = 'd',
|
||||||
|
}) => '$alias."${columnName.escapeIdentifier()}" IN (${values.sqlLiteralList()})';
|
||||||
|
|
||||||
/// SQL predicate matching rows where [columnName] differs from [value].
|
/// SQL predicate matching rows where [columnName] differs from [value].
|
||||||
@internal
|
@internal
|
||||||
String domainColumnNotPredicate(
|
String domainColumnNotPredicate(
|
||||||
|
|
|
||||||
|
|
@ -159,6 +159,24 @@ class CrdtForeignKeyProjector {
|
||||||
_foreignKeys.referencingKeysByParentTable[parentTableName] ??
|
_foreignKeys.referencingKeysByParentTable[parentTableName] ??
|
||||||
const <ReferencingForeignKey>[];
|
const <ReferencingForeignKey>[];
|
||||||
for (final reference in foreignKeys) {
|
for (final reference in foreignKeys) {
|
||||||
|
// A cascade by id needs no per-parent answer: ask for every visible
|
||||||
|
// child of the whole batch at once instead of one query per parent.
|
||||||
|
if (reference.action == ForeignKeyAction.cascade &&
|
||||||
|
reference.parentColumn == 'id') {
|
||||||
|
final childIds = await _context.findVisibleDomainRowIdsWhere(
|
||||||
|
tableName: reference.childTableName,
|
||||||
|
predicates: [
|
||||||
|
domainColumnInPredicate(reference.childColumn, parentIds),
|
||||||
|
],
|
||||||
|
transaction: transaction,
|
||||||
|
);
|
||||||
|
if (childIds.isNotEmpty) {
|
||||||
|
cascadeDeletes
|
||||||
|
.putIfAbsent(reference.childTableName, () => {})
|
||||||
|
.addAll(childIds);
|
||||||
|
}
|
||||||
|
continue;
|
||||||
|
}
|
||||||
final parentValuesById = reference.parentColumn == 'id'
|
final parentValuesById = reference.parentColumn == 'id'
|
||||||
? null
|
? null
|
||||||
: await _context.readDomainColumnValues(
|
: await _context.readDomainColumnValues(
|
||||||
|
|
@ -738,6 +756,147 @@ class CrdtForeignKeyProjector {
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Proves that merged inserts of new rows, [rowsByTable] (each row's
|
||||||
|
/// authored values by its id, by its table; [positions] says where each
|
||||||
|
/// comes in the batch), leave every projection decision as it is, as
|
||||||
|
/// [canLeaveLocalProjectionUnchanged] proves it for local inserts: each row
|
||||||
|
/// references its parents by id, each parent visible or inserted earlier in
|
||||||
|
/// the same merge (and proven alike; one inserted after its child would
|
||||||
|
/// find the child's reference held back, which only the pass restores),
|
||||||
|
/// nothing refers
|
||||||
|
/// to the row yet (no row, no attempted value), and no unique claim is
|
||||||
|
/// contested. The merge then writes them as authored, without the pass,
|
||||||
|
/// which would load every row connected to them: all of a library whose
|
||||||
|
/// tags join its parts.
|
||||||
|
Future<bool> canLeaveProjectionUnchangedByMergedInserts(
|
||||||
|
Map<String, Map<UuidValue, Map<String, Object?>>> rowsByTable,
|
||||||
|
Map<MergeRowKey, int> positions,
|
||||||
|
Transaction transaction,
|
||||||
|
) async {
|
||||||
|
for (final MapEntry(key: tableName, value: rows) in rowsByTable.entries) {
|
||||||
|
final ids = rows.keys.toSet();
|
||||||
|
final childEdges =
|
||||||
|
_foreignKeys.edgesByChildTable[tableName] ?? const <ForeignKeyEdge>[];
|
||||||
|
final parentEdges =
|
||||||
|
_foreignKeys.edgesByParentTable[tableName] ?? const <ForeignKeyEdge>[];
|
||||||
|
if (childEdges.any((edge) => edge.parentColumn != 'id') ||
|
||||||
|
parentEdges.any((edge) => edge.parentColumn != 'id')) {
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
final columns = _context.columnsByTableAndName[tableName];
|
||||||
|
Object? valueOf(Map<String, Object?> row, String column) =>
|
||||||
|
canonicalDomainValue(row[column], columns?[column]);
|
||||||
|
|
||||||
|
final parentsByTable = <String, Set<UuidValue>>{};
|
||||||
|
for (final edge in childEdges) {
|
||||||
|
for (final MapEntry(key: rowId, value: row) in rows.entries) {
|
||||||
|
final value = valueOf(row, edge.childColumn);
|
||||||
|
if (value == null) {
|
||||||
|
if (!edge.childNullable || edge.defaultValue != null) return false;
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
final id = tryUuidValue(value);
|
||||||
|
// A row of the same table inserted by the same merge would have
|
||||||
|
// to be proven first; the pass orders that.
|
||||||
|
if (id == null || (edge.parentTableName == tableName && ids.contains(id))) {
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
if (positions[(edge.parentTableName, id)] case final parentAt?) {
|
||||||
|
if (parentAt > positions[(tableName, rowId)]!) return false;
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
(parentsByTable[edge.parentTableName] ??= {}).add(id);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
final referencingColumnsByTable = <String, Set<String>>{};
|
||||||
|
for (final edge in parentEdges) {
|
||||||
|
if (edge.action == ForeignKeyAction.setDefault &&
|
||||||
|
ids.contains(tryUuidValue(edge.defaultValue))) {
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
(referencingColumnsByTable[edge.childTableName] ??= {}).add(edge.childColumn);
|
||||||
|
}
|
||||||
|
final uniqueClaims = <Map<String, Set<Object?>>>[];
|
||||||
|
for (final index in _uniqueResolver.uniqueIndexesFor(tableName)) {
|
||||||
|
final tuples = <String>{};
|
||||||
|
final claims = <String, Set<Object?>>{};
|
||||||
|
for (final row in rows.values) {
|
||||||
|
final tuple = [
|
||||||
|
for (final column in index.indexedColumns) valueOf(row, column),
|
||||||
|
];
|
||||||
|
if (tuple.any((value) => value == null)) continue;
|
||||||
|
if (!tuples.add(tuple.map(canonicalProjectionValue).join('\x1f'))) {
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
for (final (i, column) in index.indexedColumns.indexed) {
|
||||||
|
(claims[column] ??= {}).add(tuple[i]);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if (claims.isNotEmpty) uniqueClaims.add(claims);
|
||||||
|
}
|
||||||
|
if (!await _context.localProjectionDependenciesAreStable(
|
||||||
|
tableName: tableName,
|
||||||
|
rowIds: ids,
|
||||||
|
persisted: false,
|
||||||
|
parentsByTable: parentsByTable,
|
||||||
|
referencingColumnsByTable: referencingColumnsByTable,
|
||||||
|
uniqueClaims: uniqueClaims,
|
||||||
|
transaction: transaction,
|
||||||
|
)) {
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return true;
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Proves that a local delete of [tableName] rows leaves nothing for the
|
||||||
|
/// projection pass to decide: the delete's own cascade
|
||||||
|
/// (`applyDeleteActions`, recursively) already hid everything it hides.
|
||||||
|
///
|
||||||
|
/// Holds when, in every table the delete can cascade into, every incoming
|
||||||
|
/// edge cascades by id (no SET NULL / SET DEFAULT repairs), no outgoing edge
|
||||||
|
/// restricts (the deleted row could be the blocker of a withheld parent
|
||||||
|
/// delete), no table has a unique index (a released claim could restore a
|
||||||
|
/// contestant) or a SET DEFAULT dependency, and no attempted value is
|
||||||
|
/// pending in the tables the pass would load. Otherwise the full pass runs.
|
||||||
|
Future<bool> canSkipProjectionAfterLocalDelete(
|
||||||
|
String tableName,
|
||||||
|
Transaction transaction,
|
||||||
|
) async {
|
||||||
|
final cascadeTables = <String>{};
|
||||||
|
final queue = [tableName];
|
||||||
|
while (queue.isNotEmpty) {
|
||||||
|
final table = queue.removeLast();
|
||||||
|
if (!cascadeTables.add(table)) continue;
|
||||||
|
for (final edge
|
||||||
|
in _foreignKeys.edgesByParentTable[table] ?? const <ForeignKeyEdge>[]) {
|
||||||
|
if (edge.action != ForeignKeyAction.cascade || edge.parentColumn != 'id') {
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
queue.add(edge.childTableName);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
for (final table in cascadeTables) {
|
||||||
|
if (_uniqueResolver.tableHasUniqueIndexes(table)) return false;
|
||||||
|
if (_foreignKeys.tablesWithDefaultDependencies.contains(table)) return false;
|
||||||
|
for (final edge
|
||||||
|
in _foreignKeys.edgesByChildTable[table] ?? const <ForeignKeyEdge>[]) {
|
||||||
|
if (edge.action == ForeignKeyAction.restrict ||
|
||||||
|
edge.action == ForeignKeyAction.noAction) {
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
final reachable = _foreignKeys
|
||||||
|
.connectedTables({tableName})
|
||||||
|
.where(_context.syncTableByName.containsKey)
|
||||||
|
.toSet();
|
||||||
|
return !await _ProjectionReader(
|
||||||
|
_context,
|
||||||
|
transaction,
|
||||||
|
).hasAttemptedClaims(reachable);
|
||||||
|
}
|
||||||
|
|
||||||
/// Whether a mutation of [tableName] may require FK or unique projection.
|
/// Whether a mutation of [tableName] may require FK or unique projection.
|
||||||
bool needsProjection(String tableName, Set<String>? columnNames) {
|
bool needsProjection(String tableName, Set<String>? columnNames) {
|
||||||
if (_foreignKeys.columnsMayAffectForeignKeys(tableName, columnNames)) {
|
if (_foreignKeys.columnsMayAffectForeignKeys(tableName, columnNames)) {
|
||||||
|
|
|
||||||
|
|
@ -10,6 +10,7 @@ import '../hlc/hlc.dart';
|
||||||
import '../managers/hlc.dart';
|
import '../managers/hlc.dart';
|
||||||
import '../sync/engine.dart';
|
import '../sync/engine.dart';
|
||||||
import '../sync/exceptions.dart';
|
import '../sync/exceptions.dart';
|
||||||
|
import '../utils/once.dart';
|
||||||
import 'database.dart';
|
import 'database.dart';
|
||||||
import 'merge_utils/database_helpers.dart';
|
import 'merge_utils/database_helpers.dart';
|
||||||
import 'merge_utils/foreign_key_graph.dart';
|
import 'merge_utils/foreign_key_graph.dart';
|
||||||
|
|
@ -58,6 +59,17 @@ class OfflineSyncDatabaseContext {
|
||||||
@internal
|
@internal
|
||||||
bool get registryChanged => _registryChanged;
|
bool get registryChanged => _registryChanged;
|
||||||
|
|
||||||
|
final _projectionRebuild = Once();
|
||||||
|
|
||||||
|
/// Runs [rebuild], the projections of every space built anew, once for
|
||||||
|
/// this context when [registryChanged]. Every ephemeral database shares the
|
||||||
|
/// context and initializes a recorder of its own: each rebuilt every space
|
||||||
|
/// again, for every session and every merge until the process stopped
|
||||||
|
/// (upstream #150), which cost a device of a large library seconds a sync.
|
||||||
|
@internal
|
||||||
|
Future<void> rebuildProjectionsOnce(Future<void> Function() rebuild) =>
|
||||||
|
_registryChanged ? _projectionRebuild.run(rebuild) : Future<void>.value();
|
||||||
|
|
||||||
/// Initializes shared schema rows and caches their generated identifiers.
|
/// Initializes shared schema rows and caches their generated identifiers.
|
||||||
Future<void> initialize(DatabaseSession session) async {
|
Future<void> initialize(DatabaseSession session) async {
|
||||||
if (_schema != null) return;
|
if (_schema != null) return;
|
||||||
|
|
@ -264,9 +276,9 @@ class CrdtMutationRecorder {
|
||||||
|
|
||||||
Future<void> _initializeOnce() async {
|
Future<void> _initializeOnce() async {
|
||||||
await _databaseContext.initialize(_session);
|
await _databaseContext.initialize(_session);
|
||||||
if (_databaseContext.registryChanged) {
|
await _databaseContext.rebuildProjectionsOnce(
|
||||||
await _rebuildProjectionsForAllSpaces();
|
_rebuildProjectionsForAllSpaces,
|
||||||
}
|
);
|
||||||
if (persistentUserId != null) {
|
if (persistentUserId != null) {
|
||||||
await _context.spaceManager.getOrCreate(persistentUserId!);
|
await _context.spaceManager.getOrCreate(persistentUserId!);
|
||||||
}
|
}
|
||||||
|
|
@ -1232,12 +1244,19 @@ class CrdtMutationRecorder {
|
||||||
null,
|
null,
|
||||||
CrdtDataDeletedReason.userDelete,
|
CrdtDataDeletedReason.userDelete,
|
||||||
);
|
);
|
||||||
await _maybeProject(
|
// The delete's own cascade (applyDeleteActions above) usually is the
|
||||||
|
// whole projection; only run the full pass when it can decide more.
|
||||||
|
if (!await _foreignKeyProjector.canSkipProjectionAfterLocalDelete(
|
||||||
tableName,
|
tableName,
|
||||||
deletedRows.uuidRowIds,
|
|
||||||
null,
|
|
||||||
transaction,
|
transaction,
|
||||||
);
|
)) {
|
||||||
|
await _maybeProject(
|
||||||
|
tableName,
|
||||||
|
deletedRows.uuidRowIds,
|
||||||
|
null,
|
||||||
|
transaction,
|
||||||
|
);
|
||||||
|
}
|
||||||
await _context.persistCurrentNodeHlc(
|
await _context.persistCurrentNodeHlc(
|
||||||
_context.hlcManagerFor(transaction),
|
_context.hlcManagerFor(transaction),
|
||||||
transaction,
|
transaction,
|
||||||
|
|
|
||||||
|
|
@ -43,8 +43,14 @@ class OfflineSyncClient {
|
||||||
Future<void> syncOnce(
|
Future<void> syncOnce(
|
||||||
DatabaseSession session, {
|
DatabaseSession session, {
|
||||||
OfflineSyncOnMergeSuccess? onMergeSuccess,
|
OfflineSyncOnMergeSuccess? onMergeSuccess,
|
||||||
|
OfflineSyncOnInboundMerge? onInboundMerge,
|
||||||
}) async {
|
}) async {
|
||||||
await _startSyncSession(session, once: true, onMergeSuccess: onMergeSuccess).done;
|
await _startSyncSession(
|
||||||
|
session,
|
||||||
|
once: true,
|
||||||
|
onMergeSuccess: onMergeSuccess,
|
||||||
|
onInboundMerge: onInboundMerge,
|
||||||
|
).done;
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Keeps synchronizing [session] with the remote peer until cancelled through
|
/// Keeps synchronizing [session] with the remote peer until cancelled through
|
||||||
|
|
@ -54,12 +60,19 @@ class OfflineSyncClient {
|
||||||
OfflineSyncSubscription syncContinuously(
|
OfflineSyncSubscription syncContinuously(
|
||||||
DatabaseSession session, {
|
DatabaseSession session, {
|
||||||
OfflineSyncOnMergeSuccess? onMergeSuccess,
|
OfflineSyncOnMergeSuccess? onMergeSuccess,
|
||||||
}) => _startSyncSession(session, once: false, onMergeSuccess: onMergeSuccess);
|
OfflineSyncOnInboundMerge? onInboundMerge,
|
||||||
|
}) => _startSyncSession(
|
||||||
|
session,
|
||||||
|
once: false,
|
||||||
|
onMergeSuccess: onMergeSuccess,
|
||||||
|
onInboundMerge: onInboundMerge,
|
||||||
|
);
|
||||||
|
|
||||||
OfflineSyncSubscription _startSyncSession(
|
OfflineSyncSubscription _startSyncSession(
|
||||||
DatabaseSession session, {
|
DatabaseSession session, {
|
||||||
required bool once,
|
required bool once,
|
||||||
OfflineSyncOnMergeSuccess? onMergeSuccess,
|
OfflineSyncOnMergeSuccess? onMergeSuccess,
|
||||||
|
OfflineSyncOnInboundMerge? onInboundMerge,
|
||||||
}) {
|
}) {
|
||||||
// The stream will be closed using [outboundChanges.closeOrSkip].
|
// The stream will be closed using [outboundChanges.closeOrSkip].
|
||||||
// ignore: close_sinks
|
// ignore: close_sinks
|
||||||
|
|
@ -78,6 +91,7 @@ class OfflineSyncClient {
|
||||||
inbound: remoteStream,
|
inbound: remoteStream,
|
||||||
once: once,
|
once: once,
|
||||||
onMergeSuccess: onMergeSuccess,
|
onMergeSuccess: onMergeSuccess,
|
||||||
|
onInboundMerge: onInboundMerge,
|
||||||
mode: OfflineSyncPeerMode.follower,
|
mode: OfflineSyncPeerMode.follower,
|
||||||
)
|
)
|
||||||
.listen(
|
.listen(
|
||||||
|
|
|
||||||
|
|
@ -27,6 +27,12 @@ export 'space_state.dart' show OfflineSyncPeerMode;
|
||||||
typedef OfflineSyncOnMergeSuccess =
|
typedef OfflineSyncOnMergeSuccess =
|
||||||
FutureOr<void> Function(UuidValue spaceUuid, Hlc syncedHlc);
|
FutureOr<void> Function(UuidValue spaceUuid, Hlc syncedHlc);
|
||||||
|
|
||||||
|
/// Callback function for when changes from the remote peer were merged into
|
||||||
|
/// a space. Unlike [OfflineSyncOnMergeSuccess], it isn't called for a space
|
||||||
|
/// this peer only sent changes of, so a listener re-reads its data only
|
||||||
|
/// when there is something new to read.
|
||||||
|
typedef OfflineSyncOnInboundMerge = FutureOr<void> Function(UuidValue spaceUuid);
|
||||||
|
|
||||||
/// A tuple representing the ownership of a domain row.
|
/// A tuple representing the ownership of a domain row.
|
||||||
typedef DomainRowOwner = ({bool exists, int? spaceId});
|
typedef DomainRowOwner = ({bool exists, int? spaceId});
|
||||||
|
|
||||||
|
|
@ -275,6 +281,7 @@ class OfflineSyncEngine {
|
||||||
required OfflineSyncPeerMode mode,
|
required OfflineSyncPeerMode mode,
|
||||||
bool once = false,
|
bool once = false,
|
||||||
OfflineSyncOnMergeSuccess? onMergeSuccess,
|
OfflineSyncOnMergeSuccess? onMergeSuccess,
|
||||||
|
OfflineSyncOnInboundMerge? onInboundMerge,
|
||||||
}) async* {
|
}) async* {
|
||||||
// Idle timeouts are a continuous-only affordance: they let an idle cycle
|
// Idle timeouts are a continuous-only affordance: they let an idle cycle
|
||||||
// settle into an empty batch without closing the stream. A `once` session
|
// settle into an empty batch without closing the stream. A `once` session
|
||||||
|
|
@ -369,6 +376,7 @@ class OfflineSyncEngine {
|
||||||
batch,
|
batch,
|
||||||
outboundSpaces,
|
outboundSpaces,
|
||||||
onMergeSuccess,
|
onMergeSuccess,
|
||||||
|
onInboundMerge,
|
||||||
);
|
);
|
||||||
|
|
||||||
if (once) {
|
if (once) {
|
||||||
|
|
@ -429,6 +437,7 @@ class OfflineSyncEngine {
|
||||||
OfflineSyncCycleBatch batch,
|
OfflineSyncCycleBatch batch,
|
||||||
Set<UuidValue> outboundSpaces,
|
Set<UuidValue> outboundSpaces,
|
||||||
OfflineSyncOnMergeSuccess? onMergeSuccess,
|
OfflineSyncOnMergeSuccess? onMergeSuccess,
|
||||||
|
OfflineSyncOnInboundMerge? onInboundMerge,
|
||||||
) async {
|
) async {
|
||||||
if (batch.spaceSet != null) {
|
if (batch.spaceSet != null) {
|
||||||
await spaces.adoptPeerGrants(batch.spaceSet!.spaces);
|
await spaces.adoptPeerGrants(batch.spaceSet!.spaces);
|
||||||
|
|
@ -462,6 +471,9 @@ class OfflineSyncEngine {
|
||||||
mergeSet: entry.value,
|
mergeSet: entry.value,
|
||||||
);
|
);
|
||||||
await _reportMerge(onMergeSuccess, spaces, spaceId, receivedHlc);
|
await _reportMerge(onMergeSuccess, spaces, spaceId, receivedHlc);
|
||||||
|
if (entry.value.isNotEmpty) {
|
||||||
|
await onInboundMerge?.call(spaceId);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
for (final spaceId in outboundSpaces.difference(mergedSpaces)) {
|
for (final spaceId in outboundSpaces.difference(mergedSpaces)) {
|
||||||
await _reportMerge(onMergeSuccess, spaces, spaceId, null);
|
await _reportMerge(onMergeSuccess, spaces, spaceId, null);
|
||||||
|
|
|
||||||
21
packages/serverpod_offline_sync/lib/src/utils/once.dart
Normal file
21
packages/serverpod_offline_sync/lib/src/utils/once.dart
Normal file
|
|
@ -0,0 +1,21 @@
|
||||||
|
import 'package:meta/meta.dart';
|
||||||
|
|
||||||
|
/// Runs an action once for everyone who asks: a caller while it runs awaits
|
||||||
|
/// the same run, and one after it is done gets it done; only a run that
|
||||||
|
/// failed is run again by the next caller.
|
||||||
|
@internal
|
||||||
|
class Once {
|
||||||
|
Future<void>? _run;
|
||||||
|
|
||||||
|
/// Runs [action] unless it ran or runs already, and completes with that
|
||||||
|
/// run.
|
||||||
|
Future<void> run(Future<void> Function() action) async {
|
||||||
|
final run = _run ??= action();
|
||||||
|
try {
|
||||||
|
await run;
|
||||||
|
} catch (_) {
|
||||||
|
if (identical(_run, run)) _run = null;
|
||||||
|
rethrow;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
47
packages/serverpod_offline_sync/test/utils/once_test.dart
Normal file
47
packages/serverpod_offline_sync/test/utils/once_test.dart
Normal file
|
|
@ -0,0 +1,47 @@
|
||||||
|
import 'dart:async';
|
||||||
|
|
||||||
|
import 'package:serverpod_offline_sync/src/utils/once.dart';
|
||||||
|
import 'package:test/test.dart';
|
||||||
|
|
||||||
|
void main() {
|
||||||
|
group('Given an action run once,', () {
|
||||||
|
late Once once;
|
||||||
|
late int runs;
|
||||||
|
late Completer<void> gate;
|
||||||
|
|
||||||
|
setUp(() {
|
||||||
|
once = Once();
|
||||||
|
runs = 0;
|
||||||
|
gate = Completer<void>();
|
||||||
|
});
|
||||||
|
|
||||||
|
Future<void> action() async {
|
||||||
|
runs++;
|
||||||
|
await gate.future;
|
||||||
|
}
|
||||||
|
|
||||||
|
test('when asked while it runs and after, then it runs once and every '
|
||||||
|
'caller awaits that run.', () async {
|
||||||
|
final first = once.run(action);
|
||||||
|
final meanwhile = once.run(action);
|
||||||
|
gate.complete();
|
||||||
|
await Future.wait([first, meanwhile]);
|
||||||
|
await once.run(action);
|
||||||
|
|
||||||
|
expect(runs, 1);
|
||||||
|
});
|
||||||
|
|
||||||
|
test('when its run fails, then the next caller runs it again.', () async {
|
||||||
|
final failing = once.run(() async {
|
||||||
|
runs++;
|
||||||
|
throw StateError('failed');
|
||||||
|
});
|
||||||
|
await expectLater(failing, throwsStateError);
|
||||||
|
|
||||||
|
gate.complete();
|
||||||
|
await once.run(action);
|
||||||
|
|
||||||
|
expect(runs, 2);
|
||||||
|
});
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
@ -157,6 +157,182 @@ void main() {
|
||||||
expect(tombstone.hlc, remoteParentDelete.hlc);
|
expect(tombstone.hlc, remoteParentDelete.hlc);
|
||||||
},
|
},
|
||||||
);
|
);
|
||||||
|
|
||||||
|
group('and the restrict child is then deleted locally,', () {
|
||||||
|
setUp(() async {
|
||||||
|
await session.db.transactionForUser(testCrdtUserId, (tx) async {
|
||||||
|
await RestrictChild.db.deleteRow(session, child, transaction: tx);
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|
||||||
|
test('then the merged parent delete takes effect.', () async {
|
||||||
|
expect(await Person.db.findById(session, parent.id!), isNull);
|
||||||
|
expect(await RestrictChild.db.findById(session, child.id!), isNull);
|
||||||
|
});
|
||||||
|
});
|
||||||
|
});
|
||||||
|
},
|
||||||
|
);
|
||||||
|
|
||||||
|
group(
|
||||||
|
'Given a set-null middle with two cascade children,',
|
||||||
|
() {
|
||||||
|
late FkChainSetNullMiddle middle;
|
||||||
|
late FkChainSetNullCascadeChild deleted;
|
||||||
|
late FkChainSetNullCascadeChild sibling;
|
||||||
|
|
||||||
|
setUp(() async {
|
||||||
|
await session.db.transactionForUser(testCrdtUserId, (tx) async {
|
||||||
|
middle = await FkChainSetNullMiddle.db.insertRow(
|
||||||
|
session,
|
||||||
|
FkChainSetNullMiddle(id: const Uuid().v7obj(), name: 'middle'),
|
||||||
|
transaction: tx,
|
||||||
|
);
|
||||||
|
deleted = await FkChainSetNullCascadeChild.db.insertRow(
|
||||||
|
session,
|
||||||
|
FkChainSetNullCascadeChild(
|
||||||
|
id: const Uuid().v7obj(),
|
||||||
|
name: 'deleted child',
|
||||||
|
setNullMiddleId: middle.id,
|
||||||
|
),
|
||||||
|
transaction: tx,
|
||||||
|
);
|
||||||
|
sibling = await FkChainSetNullCascadeChild.db.insertRow(
|
||||||
|
session,
|
||||||
|
FkChainSetNullCascadeChild(
|
||||||
|
id: const Uuid().v7obj(),
|
||||||
|
name: 'sibling',
|
||||||
|
setNullMiddleId: middle.id,
|
||||||
|
),
|
||||||
|
transaction: tx,
|
||||||
|
);
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|
||||||
|
group('when one cascade child is deleted locally,', () {
|
||||||
|
setUp(() async {
|
||||||
|
await session.db.transactionForUser(testCrdtUserId, (tx) async {
|
||||||
|
await FkChainSetNullCascadeChild.db.deleteRow(
|
||||||
|
session,
|
||||||
|
deleted,
|
||||||
|
transaction: tx,
|
||||||
|
);
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|
||||||
|
test('then only that child is hidden, by a user delete.', () async {
|
||||||
|
expect(
|
||||||
|
await FkChainSetNullCascadeChild.db.findById(session, deleted.id!),
|
||||||
|
isNull,
|
||||||
|
);
|
||||||
|
final visibleSibling = await FkChainSetNullCascadeChild.db.findById(
|
||||||
|
session,
|
||||||
|
sibling.id!,
|
||||||
|
);
|
||||||
|
expect(visibleSibling, isNotNull);
|
||||||
|
expect(visibleSibling!.setNullMiddleId, middle.id);
|
||||||
|
expect(
|
||||||
|
await FkChainSetNullMiddle.db.findById(session, middle.id!),
|
||||||
|
isNotNull,
|
||||||
|
);
|
||||||
|
final tombstone = await CrdtDataDeleted.db.findFirstRow(
|
||||||
|
session,
|
||||||
|
where: (t) => t.row.uuidRowId.equals(deleted.id),
|
||||||
|
);
|
||||||
|
expect(tombstone, isNotNull);
|
||||||
|
expect(tombstone!.isDeleted, isTrue);
|
||||||
|
expect(tombstone.reason, CrdtDataDeletedReason.userDelete);
|
||||||
|
});
|
||||||
|
|
||||||
|
group('and the middle is then deleted by a merged remote delete,', () {
|
||||||
|
setUp(() async {
|
||||||
|
await session.db.mergeChanges(
|
||||||
|
[
|
||||||
|
_deleteChange(
|
||||||
|
tableName: FkChainSetNullMiddle.t.tableName,
|
||||||
|
rowId: middle.id!,
|
||||||
|
after: await rowHlc(middle.id!),
|
||||||
|
),
|
||||||
|
],
|
||||||
|
spaceId: testCrdtUserId,
|
||||||
|
);
|
||||||
|
});
|
||||||
|
|
||||||
|
test('then the remaining child is hidden with it.', () async {
|
||||||
|
expect(
|
||||||
|
await FkChainSetNullMiddle.db.findById(session, middle.id!),
|
||||||
|
isNull,
|
||||||
|
);
|
||||||
|
expect(
|
||||||
|
await FkChainSetNullCascadeChild.db.findById(session, sibling.id!),
|
||||||
|
isNull,
|
||||||
|
);
|
||||||
|
});
|
||||||
|
});
|
||||||
|
});
|
||||||
|
},
|
||||||
|
);
|
||||||
|
|
||||||
|
group(
|
||||||
|
'Given a set-null middle kept by its restrict child against a merged delete,',
|
||||||
|
() {
|
||||||
|
late FkChainSetNullMiddle middle;
|
||||||
|
late FkChainSetNullRestrictChild blocker;
|
||||||
|
|
||||||
|
setUp(() async {
|
||||||
|
await session.db.transactionForUser(testCrdtUserId, (tx) async {
|
||||||
|
middle = await FkChainSetNullMiddle.db.insertRow(
|
||||||
|
session,
|
||||||
|
FkChainSetNullMiddle(id: const Uuid().v7obj(), name: 'kept middle'),
|
||||||
|
transaction: tx,
|
||||||
|
);
|
||||||
|
blocker = await FkChainSetNullRestrictChild.db.insertRow(
|
||||||
|
session,
|
||||||
|
FkChainSetNullRestrictChild(
|
||||||
|
id: const Uuid().v7obj(),
|
||||||
|
name: 'blocker',
|
||||||
|
setNullMiddleId: middle.id,
|
||||||
|
),
|
||||||
|
transaction: tx,
|
||||||
|
);
|
||||||
|
});
|
||||||
|
await session.db.mergeChanges(
|
||||||
|
[
|
||||||
|
_deleteChange(
|
||||||
|
tableName: FkChainSetNullMiddle.t.tableName,
|
||||||
|
rowId: middle.id!,
|
||||||
|
after: await rowHlc(middle.id!),
|
||||||
|
),
|
||||||
|
],
|
||||||
|
spaceId: testCrdtUserId,
|
||||||
|
);
|
||||||
|
expect(
|
||||||
|
await FkChainSetNullMiddle.db.findById(session, middle.id!),
|
||||||
|
isNotNull,
|
||||||
|
);
|
||||||
|
});
|
||||||
|
|
||||||
|
group('when the restrict child is deleted locally,', () {
|
||||||
|
setUp(() async {
|
||||||
|
await session.db.transactionForUser(testCrdtUserId, (tx) async {
|
||||||
|
await FkChainSetNullRestrictChild.db.deleteRow(
|
||||||
|
session,
|
||||||
|
blocker,
|
||||||
|
transaction: tx,
|
||||||
|
);
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|
||||||
|
test('then the merged delete of the middle takes effect.', () async {
|
||||||
|
expect(
|
||||||
|
await FkChainSetNullMiddle.db.findById(session, middle.id!),
|
||||||
|
isNull,
|
||||||
|
);
|
||||||
|
expect(
|
||||||
|
await FkChainSetNullRestrictChild.db.findById(session, blocker.id!),
|
||||||
|
isNull,
|
||||||
|
);
|
||||||
|
});
|
||||||
});
|
});
|
||||||
},
|
},
|
||||||
);
|
);
|
||||||
|
|
|
||||||
|
|
@ -0,0 +1,191 @@
|
||||||
|
import 'package:serverpod/serverpod.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';
|
||||||
|
|
||||||
|
// Merged inserts of new rows whose parents are visible, which nothing
|
||||||
|
// refers to yet, change no projection decision, and the merge leaves the
|
||||||
|
// projection pass out for them. Anything else still takes the pass: these
|
||||||
|
// cover both sides.
|
||||||
|
void main() {
|
||||||
|
initTestClientSession();
|
||||||
|
|
||||||
|
CrdtMergeInsert insertOf(TableRow row, String tableName) {
|
||||||
|
final node = const Uuid().v7obj();
|
||||||
|
return CrdtMergeInsert(
|
||||||
|
uuidSpaceId: testCrdtUserId,
|
||||||
|
tableName: tableName,
|
||||||
|
uuidRowId: row.id! as UuidValue,
|
||||||
|
uuidNodeId: node,
|
||||||
|
hlcDatetime: DateTime.now().toUtc(),
|
||||||
|
hlcCounter: 0,
|
||||||
|
data: row,
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
Future<void> merge(List<CrdtMergeChange> changes) =>
|
||||||
|
session.db.mergeChanges(changes, spaceId: testCrdtUserId);
|
||||||
|
|
||||||
|
group('Given a set-null middle stored here,', () {
|
||||||
|
late FkChainSetNullMiddle middle;
|
||||||
|
|
||||||
|
setUp(() async {
|
||||||
|
middle = await session.db.transactionForUser(
|
||||||
|
testCrdtUserId,
|
||||||
|
(tx) => FkChainSetNullMiddle.db.insertRow(
|
||||||
|
session,
|
||||||
|
FkChainSetNullMiddle(id: const Uuid().v7obj(), name: 'middle'),
|
||||||
|
transaction: tx,
|
||||||
|
),
|
||||||
|
);
|
||||||
|
});
|
||||||
|
|
||||||
|
group('when a remote insert of its cascade child is merged,', () {
|
||||||
|
late FkChainSetNullCascadeChild child;
|
||||||
|
|
||||||
|
setUp(() async {
|
||||||
|
child = FkChainSetNullCascadeChild(
|
||||||
|
id: const Uuid().v7obj(),
|
||||||
|
name: 'child',
|
||||||
|
setNullMiddleId: middle.id,
|
||||||
|
);
|
||||||
|
await merge([insertOf(child, FkChainSetNullCascadeChild.t.tableName)]);
|
||||||
|
});
|
||||||
|
|
||||||
|
test('then the child is there as authored, holding nothing back.', () async {
|
||||||
|
final visible = await FkChainSetNullCascadeChild.db.findById(
|
||||||
|
session,
|
||||||
|
child.id!,
|
||||||
|
);
|
||||||
|
expect(visible, isNotNull);
|
||||||
|
expect(visible!.setNullMiddleId, middle.id);
|
||||||
|
expect(
|
||||||
|
await attemptedValue(
|
||||||
|
rowId: child.id!,
|
||||||
|
columnName: FkChainSetNullCascadeChild.t.setNullMiddleId.columnName,
|
||||||
|
),
|
||||||
|
isNull,
|
||||||
|
);
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|
||||||
|
group('when the middle is deleted by a merged remote delete and then a '
|
||||||
|
'remote insert of its cascade child is merged,', () {
|
||||||
|
late FkChainSetNullCascadeChild child;
|
||||||
|
|
||||||
|
setUp(() async {
|
||||||
|
await merge([
|
||||||
|
CrdtMergeDelete(
|
||||||
|
uuidSpaceId: testCrdtUserId,
|
||||||
|
tableName: FkChainSetNullMiddle.t.tableName,
|
||||||
|
uuidRowId: middle.id!,
|
||||||
|
uuidNodeId: const Uuid().v7obj(),
|
||||||
|
hlcDatetime: (await rowHlc(
|
||||||
|
middle.id!,
|
||||||
|
)).datetime.add(const Duration(milliseconds: 1)),
|
||||||
|
hlcCounter: 0,
|
||||||
|
clFlag: 2,
|
||||||
|
reason: CrdtDataDeletedReason.userDelete,
|
||||||
|
),
|
||||||
|
]);
|
||||||
|
child = FkChainSetNullCascadeChild(
|
||||||
|
id: const Uuid().v7obj(),
|
||||||
|
name: 'child of a deleted middle',
|
||||||
|
setNullMiddleId: middle.id,
|
||||||
|
);
|
||||||
|
await merge([insertOf(child, FkChainSetNullCascadeChild.t.tableName)]);
|
||||||
|
});
|
||||||
|
|
||||||
|
test('then the child is hidden with its middle.', () async {
|
||||||
|
expect(
|
||||||
|
await FkChainSetNullMiddle.db.findById(session, middle.id!),
|
||||||
|
isNull,
|
||||||
|
);
|
||||||
|
expect(
|
||||||
|
await FkChainSetNullCascadeChild.db.findById(session, child.id!),
|
||||||
|
isNull,
|
||||||
|
);
|
||||||
|
});
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|
||||||
|
group('Given a set-null grandchild merged before its middle, its foreign '
|
||||||
|
'key held back,', () {
|
||||||
|
late FkChainSetNullMiddle middle;
|
||||||
|
late FkChainSetNullSetNullChild grandchild;
|
||||||
|
|
||||||
|
setUp(() async {
|
||||||
|
middle = FkChainSetNullMiddle(id: const Uuid().v7obj(), name: 'late');
|
||||||
|
grandchild = FkChainSetNullSetNullChild(
|
||||||
|
id: const Uuid().v7obj(),
|
||||||
|
name: 'early',
|
||||||
|
setNullMiddleId: middle.id,
|
||||||
|
);
|
||||||
|
await merge([
|
||||||
|
insertOf(grandchild, FkChainSetNullSetNullChild.t.tableName),
|
||||||
|
]);
|
||||||
|
final early = await FkChainSetNullSetNullChild.db.findById(
|
||||||
|
session,
|
||||||
|
grandchild.id!,
|
||||||
|
);
|
||||||
|
expect(early!.setNullMiddleId, isNull);
|
||||||
|
});
|
||||||
|
|
||||||
|
group("when the middle's insert is merged,", () {
|
||||||
|
setUp(() async {
|
||||||
|
await merge([insertOf(middle, FkChainSetNullMiddle.t.tableName)]);
|
||||||
|
});
|
||||||
|
|
||||||
|
test('then the grandchild points at it.', () async {
|
||||||
|
final visible = await FkChainSetNullSetNullChild.db.findById(
|
||||||
|
session,
|
||||||
|
grandchild.id!,
|
||||||
|
);
|
||||||
|
expect(visible!.setNullMiddleId, middle.id);
|
||||||
|
expect(
|
||||||
|
await attemptedValue(
|
||||||
|
rowId: grandchild.id!,
|
||||||
|
columnName: FkChainSetNullSetNullChild.t.setNullMiddleId.columnName,
|
||||||
|
),
|
||||||
|
isNull,
|
||||||
|
);
|
||||||
|
});
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|
||||||
|
group('Given nothing stored,', () {
|
||||||
|
group('when a remote insert of a middle and its cascade child is merged '
|
||||||
|
'in one batch,', () {
|
||||||
|
late FkChainSetNullMiddle middle;
|
||||||
|
late FkChainSetNullCascadeChild child;
|
||||||
|
|
||||||
|
setUp(() async {
|
||||||
|
middle = FkChainSetNullMiddle(id: const Uuid().v7obj(), name: 'new');
|
||||||
|
child = FkChainSetNullCascadeChild(
|
||||||
|
id: const Uuid().v7obj(),
|
||||||
|
name: 'new child',
|
||||||
|
setNullMiddleId: middle.id,
|
||||||
|
);
|
||||||
|
await merge([
|
||||||
|
insertOf(middle, FkChainSetNullMiddle.t.tableName),
|
||||||
|
insertOf(child, FkChainSetNullCascadeChild.t.tableName),
|
||||||
|
]);
|
||||||
|
});
|
||||||
|
|
||||||
|
test('then both are there, the child pointing at the middle.', () async {
|
||||||
|
expect(
|
||||||
|
await FkChainSetNullMiddle.db.findById(session, middle.id!),
|
||||||
|
isNotNull,
|
||||||
|
);
|
||||||
|
final visible = await FkChainSetNullCascadeChild.db.findById(
|
||||||
|
session,
|
||||||
|
child.id!,
|
||||||
|
);
|
||||||
|
expect(visible!.setNullMiddleId, middle.id);
|
||||||
|
});
|
||||||
|
});
|
||||||
|
});
|
||||||
|
}
|
||||||
Loading…
Reference in a new issue