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

feat: Add an example app and misc fixes (#81)

Browse source
This commit is contained in:
Marcelo Mendonça Soares 2026-06-23 22:53:47 -03:00 • committed by GitHub
commit de7e22d91a
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
5 changed files with 9138 additions and 61 deletions
Download patch file Download diff file Expand all files Collapse all files

7
packages/serverpod_offline_sync_client/lib/src/crdt/client_sync.dart
View file

@ -34,8 +34,11 @@ class CrdtSyncClient {
/// Pushes local pending changes for [session] to the remote peer once.
///
/// The [session] must be wrapped in a [CrdtDatabaseSession].
Future<void> syncOnce(DatabaseSession session) async {
await _startSyncSession(session, once: true).done;
Future<void> syncOnce(
DatabaseSession session, {
CrdtSyncOnMergeSuccess? onMergeSuccess,
}) async {
await _startSyncSession(session, once: true, onMergeSuccess: onMergeSuccess).done;
}
/// Keeps synchronizing [session] with the remote peer until cancelled through

35
packages/serverpod_offline_sync_client/lib/src/crdt/sync.dart
View file

@ -340,6 +340,17 @@ class CrdtSync {
yield CrdtSyncClose();
await inboundIterator.moveAndThrowIfNot<CrdtSyncClose>();
sessionCompleted = true;
// Keep reading until the peer closes its side so the underlying
// transport subscription reaches "done" instead of being left paused.
// A paused inbound controller stalls the peer's stream teardown for
// several seconds (the transport's close timeout), which lands on the
// critical path of the next sync round over a shared connection.
//
// The drain runs detached: awaiting it here would deadlock the
// symmetric close handshake, since the peer only closes its side once
// our own outbound stream closes, which happens after this generator
// returns.
unawaited(_drainUntilDone(inboundIterator));
return;
}
@ -348,8 +359,10 @@ class CrdtSync {
}
} finally {
// Cancelling inbound on normal completion races with WebSocket stream
// teardown and produces "connection closed" errors on the peer. Keep
// cleanup for abnormal exits so listener cancellation can unblock.
// teardown and produces "connection closed" errors on the peer. On normal
// completion the inbound is instead drained to "done" (see above). Keep
// forced cancellation for abnormal exits so listener cancellation can
// unblock.
if (!sessionCompleted) {
// Best-effort cleanup, since the transport will close the socket anyway.
const waitTimeout = Duration(milliseconds: 200);
@ -358,6 +371,24 @@ class CrdtSync {
}
}
/// Drains [iterator] until the peer closes the stream.
///
/// Used to settle the inbound transport after the `once` close handshake so
/// its controller reaches "done" with an active listener instead of being
/// torn down while paused. Trailing events (idle timeouts, late frames) are
/// discarded; errors are swallowed since the transport is closing anyway.
static Future<void> _drainUntilDone(
StreamIterator<CrdtSyncStreamEvent> iterator,
) async {
try {
while (await iterator.moveNext()) {
// Discard whatever the peer sends before it closes its side.
}
} on Object catch (_) {
// Best-effort: the transport is shutting down.
}
}
void _validateSyncTablesHash(String syncTablesHash) {
if (syncTablesHash != currentSyncTablesHash) {
throw SyncTablesHashMismatchException(

47
packages/serverpod_offline_sync_client/lib/src/database/database.dart
View file

@ -55,6 +55,8 @@ class CrdtDatabase implements Database {
final Duration _continuousSyncInterval;
final CrdtMutationRecorder _recorder;
bool _initialized = false;
Future<void>? _initializing;
late final CrdtSync _sync = CrdtSync(
syncTables: _syncTables,
@ -65,7 +67,27 @@ class CrdtDatabase implements Database {
/// Initializes the CRDT database.
Future<void> initialize() async {
await _recorder.initialize();
final initializing = _recorder.initialize().then((_) {
_initialized = true;
});
_initializing = initializing;
try {
await initializing;
} finally {
if (identical(_initializing, initializing)) {
_initializing = null;
}
}
}
Future<void> _ensureInitialized() async {
if (_initialized) return;
final initializing = _initializing;
if (initializing != null) {
await initializing;
return;
}
await initialize();
}
/// The hash describing the synchronized schema configured for this database.
@ -73,6 +95,7 @@ class CrdtDatabase implements Database {
/// Returns the current node identifier for the effective user.
Future<UuidValue> currentNodeId({UuidValue? userId}) async {
await _ensureInitialized();
final effectiveUserId = await _requireUserId(userId);
final user = await _recorder.getOrCreateScope(effectiveUserId);
return user.currentNode!.uuidNodeId;
@ -85,6 +108,7 @@ class CrdtDatabase implements Database {
bool once = false,
CrdtSyncOnMergeSuccess? onMergeSuccess,
}) async* {
await _ensureInitialized();
final effectiveUserId = await _requireUserId(userId);
yield* _sync.sync(
_delegate.session,
@ -101,6 +125,7 @@ class CrdtDatabase implements Database {
Hlc syncedHlc, {
UuidValue? userId,
}) async {
await _ensureInitialized();
final effectiveUserId = await _requireUserId(userId);
await _recorder.recordSyncCheckpoint(
effectiveUserId,
@ -119,6 +144,7 @@ class CrdtDatabase implements Database {
UuidValue? scopeId,
}) async {
if (mergeSet.isEmpty) return;
await _ensureInitialized();
final effectiveScopeId =
scopeId ??
@ -181,6 +207,7 @@ class CrdtDatabase implements Database {
LockMode? lockMode,
LockBehavior? lockBehavior,
}) async {
await _ensureInitialized();
final result = await _delegate.find<T>(
where: _whereVisibleWithTombstone<T>(where, include, transaction),
limit: limit,
@ -206,6 +233,7 @@ class CrdtDatabase implements Database {
LockMode? lockMode,
LockBehavior? lockBehavior,
}) async {
await _ensureInitialized();
final table = serializationManager.getTableForType(T);
final where = table?.id.equals(id);
final result = await _delegate.findFirstRow<T>(
@ -231,6 +259,7 @@ class CrdtDatabase implements Database {
LockMode? lockMode,
LockBehavior? lockBehavior,
}) async {
await _ensureInitialized();
final result = await _delegate.findFirstRow<T>(
where: _whereVisibleWithTombstone<T>(where, include, transaction),
offset: offset,
@ -255,6 +284,7 @@ class CrdtDatabase implements Database {
bool ignoreConflicts = false,
}) async {
if (rows.isEmpty) return [];
await _ensureInitialized();
return DatabaseUtil.runInTransactionOrSavepoint(
_delegate,
transaction,
@ -278,6 +308,7 @@ class CrdtDatabase implements Database {
T row, {
Transaction? transaction,
}) async {
await _ensureInitialized();
return DatabaseUtil.runInTransactionOrSavepoint(
_delegate,
transaction,
@ -323,6 +354,7 @@ class CrdtDatabase implements Database {
Expression? updateWhere,
Transaction? transaction,
}) async {
await _ensureInitialized();
if (T == CrdtSyncIntegrityViolation) {
return _delegate.upsertRow<T>(
row,
@ -343,6 +375,7 @@ class CrdtDatabase implements Database {
Transaction? transaction,
}) async {
if (rows.isEmpty) return [];
await _ensureInitialized();
return DatabaseUtil.runInTransactionOrSavepoint(
_delegate,
transaction,
@ -369,6 +402,7 @@ class CrdtDatabase implements Database {
List<Column>? columns,
Transaction? transaction,
}) async {
await _ensureInitialized();
return DatabaseUtil.runInTransactionOrSavepoint(
_delegate,
transaction,
@ -423,6 +457,7 @@ class CrdtDatabase implements Database {
required List<ColumnValue> columnValues,
Transaction? transaction,
}) async {
await _ensureInitialized();
final table = serializationManager.getTableForType(T);
if (table == null) return null;
@ -447,6 +482,7 @@ class CrdtDatabase implements Database {
bool orderDescending = false,
Transaction? transaction,
}) async {
await _ensureInitialized();
if (_recorder.isCrdtTracked<T>()) {
_assertNoScopeIdColumnValues<T>(columnValues);
}
@ -487,6 +523,7 @@ class CrdtDatabase implements Database {
Transaction? transaction,
}) async {
if (rows.isEmpty) return [];
await _ensureInitialized();
return deleteWhere<T>(
where: rows.first.table.id.inSet(
rows.map((row) => row.id).castToIdType().toSet(),
@ -505,6 +542,7 @@ class CrdtDatabase implements Database {
T row, {
Transaction? transaction,
}) async {
await _ensureInitialized();
final deletedRows = await deleteWhere<T>(
where: row.table.id.equals(row.id),
transaction: transaction,
@ -527,6 +565,7 @@ class CrdtDatabase implements Database {
bool orderDescending = false,
Transaction? transaction,
}) async {
await _ensureInitialized();
if (!_recorder.isCrdtTracked<T>()) {
return _delegate.deleteWhere<T>(
where: where,
@ -566,6 +605,7 @@ class CrdtDatabase implements Database {
bool useCache = true,
Transaction? transaction,
}) async {
await _ensureInitialized();
return _delegate.count<T>(
where: _whereVisibleWithTombstone<T>(where, null, transaction),
limit: limit,
@ -581,6 +621,7 @@ class CrdtDatabase implements Database {
required Transaction transaction,
LockBehavior lockBehavior = LockBehavior.wait,
}) async {
await _ensureInitialized();
return _delegate.lockRows<T>(
where: _whereVisibleWithTombstone<T>(where, null, transaction)!,
lockMode: lockMode,
@ -594,6 +635,9 @@ class CrdtDatabase implements Database {
TransactionFunction<R> transactionFunction, {
TransactionSettings? settings,
}) async {
// Do not initialize CRDT here. Serverpod runs database migrations inside
// [transaction] before CRDT tables exist. Callers that need CRDT state
// ([transactionForUser], mutating ORM methods) initialize explicitly.
return _delegate.transaction(
transactionFunction,
settings: settings,
@ -610,6 +654,7 @@ class CrdtDatabase implements Database {
TransactionFunction<R> transactionFunction, {
TransactionSettings? settings,
}) async {
await _ensureInitialized();
// Ensure that the user exists with a node before starting the transaction.
final user = await _recorder.getOrCreateScope(userId);

45
packages/serverpod_offline_sync_client/lib/src/database/merge_utils/foreign_key_projector.dart
View file

@ -239,7 +239,15 @@ extension _CrdtForeignKeyProjector on CrdtMutationRecorder {
attempts[fieldKey] = (value: attemptedValue, reason: null);
if (attemptedValue == null) continue;
if (await _foreignKeyTargetVisible(edge, attemptedValue, transaction)) {
final targetPresence = await _lookupForeignKeyTargetPresence(
parentTableName: edge.parentTableName,
parentColumn: edge.parentColumn,
value: attemptedValue,
transaction: transaction,
);
if (targetPresence == _ForeignKeyTargetPresence.visible ||
(targetPresence == _ForeignKeyTargetPresence.hidden &&
edge.action == ForeignKeyAction.cascade)) {
continue;
}
@ -1053,25 +1061,6 @@ extension _CrdtForeignKeyProjector on CrdtMutationRecorder {
return (valid: true, value: defaultValue);
}
Future<bool> _foreignKeyTargetVisible(
_ForeignKeyEdge edge,
UuidValue value,
Transaction transaction,
) async {
final rowIds = await _findVisibleDomainRowIdsWhere(
tableName: edge.parentTableName,
predicates: [
_domainColumnPredicate(
'scopeId',
_getHlcManager(transaction).normalizedScopeId,
),
_domainColumnPredicate(edge.parentColumn, value),
],
transaction: transaction,
);
return rowIds.isNotEmpty;
}
Future<_ForeignKeyDefaultProjection> _defaultProjectionValueFromDatabase(
_ForeignKeyEdge edge,
Transaction transaction,
@ -1081,14 +1070,16 @@ extension _CrdtForeignKeyProjector on CrdtMutationRecorder {
return (valid: edge.childNullable, value: null);
}
final targetVisible = await _foreignKeyTargetVisible(
edge,
defaultValue,
transaction,
final targetVisible = await _lookupForeignKeyTargetPresence(
parentTableName: edge.parentTableName,
parentColumn: edge.parentColumn,
value: defaultValue,
transaction: transaction,
);
if (!targetVisible) return (valid: false, value: null);
return (valid: true, value: defaultValue);
if (targetVisible == _ForeignKeyTargetPresence.visible) {
return (valid: true, value: defaultValue);
}
return (valid: false, value: null);
}
UuidValue? _parentReferenceValue(

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

@ -52,6 +52,12 @@ enum _UniqueConflictReleaseKind {
syntheticUuid,
}
enum _ForeignKeyTargetPresence {
absent,
visible,
hidden,
}
/// Persists additional CRDT rows after a mutating ORM operation completes.
///
/// Callbacks receive the underlying database (not the CRDT proxy) and the
@ -73,11 +79,7 @@ class CrdtMutationRecorder {
late final DatabaseSession _dbSession = _db.session;
late final _session = CrdtDatabaseSession(
_db,
syncTables: syncTables,
persistentUserId: persistentUserId,
);
late final _session = BasicDatabaseSession(_db);
late final CrdtScopeManager _scopeManager = CrdtScopeManager(_dbSession);
@ -1092,6 +1094,35 @@ WHERE (${predicates.join(') AND (')})
};
}
Future<_ForeignKeyTargetPresence> _lookupForeignKeyTargetPresence({
required String parentTableName,
required String parentColumn,
required UuidValue value,
required Transaction transaction,
}) async {
final (tableId, _) = _schema[parentTableName]!;
final scopeId = _getHlcManager(transaction).normalizedScopeId;
final result = await _db.unsafeQuery(
'''
SELECT CASE
WHEN r."id" IS NULL OR r."visibility" <= $crdtRowLastVisibleVisibilityIndex THEN 1
ELSE 0
END AS visible
FROM "${_escapeIdentifier(parentTableName)}" d
LEFT JOIN "crdt_data_rows" r
ON r."scopeId" = $scopeId AND r."tblId" = $tableId AND r."uuidRowId" = d."id"
WHERE (${_domainColumnPredicate('scopeId', scopeId)})
AND (${_domainColumnPredicate(parentColumn, value)})
LIMIT 1
''',
transaction: transaction,
);
if (result.isEmpty) return _ForeignKeyTargetPresence.absent;
return (result.first[0] as num) == 1
? _ForeignKeyTargetPresence.visible
: _ForeignKeyTargetPresence.hidden;
}
Future<void> _updateDomainRows(
String tableName,
Set<UuidValue> rowIds,