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

Compare commits

base: halkyon-days:f062725a16fffa1be8cf9141cdbfefae209598ea
Branches Tags
halkyon-days:halkyon/byte-payloads
halkyon-days:halkyon/space-writes
halkyon-days:halkyon/batched-row-reads
halkyon-days:halkyon/batch-inserts
halkyon-days:halkyon/fast-merge-deletes
halkyon-days:halkyon/fast-merges
halkyon-days:halkyon/fast-deletes
halkyon-days:halkyon/inbound-merges
..
compare: halkyon-days:b5450353b98240a1bd6f4471815a8be94c91ce5b
Branches Tags
halkyon-days:halkyon/space-writes
halkyon-days:halkyon/batched-row-reads
halkyon-days:halkyon/batch-inserts
halkyon-days:halkyon/fast-merge-deletes
halkyon-days:halkyon/fast-merges
halkyon-days:halkyon/fast-deletes
halkyon-days:halkyon/inbound-merges
halkyon-days:halkyon/byte-payloads
6 changed files with 8 additions and 427 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -36,7 +36,7 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
// 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.
var mayAffectProjection = _foreignKeyProjector.mergeOperationsMayAffectProjection(
final mayAffectProjection = _foreignKeyProjector.mergeOperationsMayAffectProjection(
operations,
);
final authoredOverlays = <MergeFieldKey, Object?>{};
@ -48,14 +48,6 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
);
final nodesByUuid = remoteNodes.nodesByUuid;
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
// seed is every touched row, not just the ones with overlays.
@ -67,7 +59,7 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
touchedRows.add((operation.tableName, operation.uuidRowId));
}
final batch = mayAffectProjection || unchangedByInserts
final batch = mayAffectProjection
? await _planBatchInserts(
operations,
nodesByUuid,
@ -75,7 +67,6 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
touchedTables,
touchedRows,
transaction,
plan: mayAffectProjection,
)
: (plan: null, pending: const <PendingProjectionRow>[]);
// Every merged insert has to know whether the rows its foreign keys name
@ -371,11 +362,8 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
MergeContext context,
Set<String> seedTables,
Set<MergeRowKey> seedRows,
Transaction transaction, {
// Whether to plan them: without, only the rows are listed, for their
// parents to be resolved together.
bool plan = true,
}) async {
Transaction transaction,
) async {
final pending = <PendingProjectionRow>[];
final seenRowKeys = <MergeRowKey>{};
@ -403,7 +391,7 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
hidden: false,
));
}
if (pending.isEmpty || !plan) return (plan: null, pending: pending);
if (pending.isEmpty) return (plan: null, pending: pending);
return (
plan: await _foreignKeyProjector.project(
@ -416,49 +404,6 @@ 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(
CrdtMergeInsert insert,
Map<UuidValue, CrdtNode> remoteNodes,

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

@ -756,99 +756,6 @@ 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.

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

@ -10,7 +10,6 @@ import '../hlc/hlc.dart';
import '../managers/hlc.dart';
import '../sync/engine.dart';
import '../sync/exceptions.dart';
import '../utils/once.dart';
import 'database.dart';
import 'merge_utils/database_helpers.dart';
import 'merge_utils/foreign_key_graph.dart';
@ -59,17 +58,6 @@ class OfflineSyncDatabaseContext {
@internal
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.
Future<void> initialize(DatabaseSession session) async {
if (_schema != null) return;
@ -276,9 +264,9 @@ class CrdtMutationRecorder {
Future<void> _initializeOnce() async {
await _databaseContext.initialize(_session);
await _databaseContext.rebuildProjectionsOnce(
_rebuildProjectionsForAllSpaces,
);
if (_databaseContext.registryChanged) {
await _rebuildProjectionsForAllSpaces();
}
if (persistentUserId != null) {
await _context.spaceManager.getOrCreate(persistentUserId!);
}

21
packages/serverpod_offline_sync/lib/src/utils/once.dart
View file

@ -1,21 +0,0 @@
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
View file

@ -1,47 +0,0 @@
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);
});
});
}

191
test/serverpod_offline_sync_test_server/test/integration/merge/merge_inserts_projection_test.dart
View file

@ -1,191 +0,0 @@
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);
});
});
});
}