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:845fb557db3e21cf78b152f9e39ba6a26947bab9
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:741e334d6ac4ca89f99945dda2e444152ee32994
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
8 changed files with 62 additions and 889 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -85,19 +85,6 @@ class OfflineSyncDatabase implements Database {
final CrdtMutationRecorder _recorder; final CrdtMutationRecorder _recorder;
/// Which space a new row goes in, for a client with a persistent user
/// writing without a transaction: a space the user is a member of, or
/// null for their personal space. Without it, new rows go in the
/// personal space.
///
/// A write without a transaction acts in the space it touches: rows
/// updated or deleted in their own space, new rows in the one this
/// names, each through [transactionForUser], which checks that the user
/// may write there. One call touches one space, and is refused
/// ([StateError]) where it would touch two. A write within a transaction
/// acts in that transaction's space, as before.
FutureOr<UuidValue?> Function(TableRow row)? newRowSpace;
late final _sync = OfflineSyncEngine( late final _sync = OfflineSyncEngine(
syncTables: _syncTables, syncTables: _syncTables,
serializationManager: serializationManager, serializationManager: serializationManager,
@ -356,21 +343,6 @@ class OfflineSyncDatabase implements Database {
noReturn: noReturn, noReturn: noReturn,
); );
} }
if (_routesWrite(transaction)) {
final space = await _newRowsSpace(rows);
if (space != null) {
return transactionForUser(
_recorder.persistentUserId!,
(tx) => insert<T>(
rows,
transaction: tx,
ignoreConflicts: ignoreConflicts,
noReturn: noReturn,
),
spaceId: space,
);
}
}
return _runTrackedWrite( return _runTrackedWrite(
transaction, transaction,
(tx) async { (tx) async {
@ -471,31 +443,6 @@ class OfflineSyncDatabase implements Database {
noReturn: noReturn, noReturn: noReturn,
); );
} }
if (_routesWrite(transaction)) {
final table = rows.first.table;
final ids = {
for (final row in rows)
if (row.id case final Object id) id,
};
final stored = ids.isEmpty
? null
: await _storedRowsSpace<T>(table.id.inSet(ids.castToIdType().toSet()));
final space = stored ?? await _newRowsSpace(rows);
if (space != null) {
return transactionForUser(
_recorder.persistentUserId!,
(tx) => upsert<T>(
rows,
conflictColumns: conflictColumns,
updateColumns: updateColumns,
updateWhere: updateWhere,
transaction: tx,
noReturn: noReturn,
),
spaceId: space,
);
}
}
return _runTrackedWrite( return _runTrackedWrite(
transaction, transaction,
(tx) async { (tx) async {
@ -716,23 +663,6 @@ class OfflineSyncDatabase implements Database {
noReturn: noReturn, noReturn: noReturn,
); );
} }
if (_routesWrite(transaction)) {
final space = await _storedRowsSpace<T>(
rows.first.table.id.inSet(rows.map((row) => row.id).castToIdType().toSet()),
);
if (space != null) {
return transactionForUser(
_recorder.persistentUserId!,
(tx) => update<T>(
rows,
columns: columns,
transaction: tx,
noReturn: noReturn,
),
spaceId: space,
);
}
}
return _runTrackedWrite( return _runTrackedWrite(
transaction, transaction,
(tx) async { (tx) async {
@ -773,16 +703,6 @@ class OfflineSyncDatabase implements Database {
transaction: transaction, transaction: transaction,
); );
} }
if (_routesWrite(transaction)) {
final space = await _storedRowsSpace<T>(row.table.id.equals(row.id));
if (space != null) {
return transactionForUser(
_recorder.persistentUserId!,
(tx) => updateRow<T>(row, columns: columns, transaction: tx),
spaceId: space,
);
}
}
return _runTrackedWrite( return _runTrackedWrite(
transaction, transaction,
(tx) async { (tx) async {
@ -888,26 +808,6 @@ class OfflineSyncDatabase implements Database {
} }
_assertNoSpaceIdColumnValues<T>(columnValues); _assertNoSpaceIdColumnValues<T>(columnValues);
if (_routesWrite(transaction)) {
final space = await _storedRowsSpace<T>(where);
if (space != null) {
return transactionForUser(
_recorder.persistentUserId!,
(tx) => updateWhere<T>(
columnValues: columnValues,
where: where,
limit: limit,
offset: offset,
orderBy: orderBy,
orderByList: orderByList,
orderDescending: orderDescending,
transaction: tx,
noReturn: noReturn,
),
spaceId: space,
);
}
}
return _runTrackedWrite( return _runTrackedWrite(
transaction, transaction,
(tx) async { (tx) async {
@ -1017,23 +917,6 @@ class OfflineSyncDatabase implements Database {
); );
} }
if (_routesWrite(transaction)) {
final space = await _storedRowsSpace<T>(where);
if (space != null) {
return transactionForUser(
_recorder.persistentUserId!,
(tx) => deleteWhere<T>(
where: where,
orderBy: orderBy,
orderByList: orderByList,
orderDescending: orderDescending,
transaction: tx,
noReturn: noReturn,
),
spaceId: space,
);
}
}
return _runTrackedWrite( return _runTrackedWrite(
transaction, transaction,
(tx) async { (tx) async {

57
packages/serverpod_offline_sync/lib/src/database/space.dart
View file

@ -80,63 +80,6 @@ extension _OfflineSyncDatabaseSpace on OfflineSyncDatabase {
} }
} }
/// Whether a write chooses its space itself ([OfflineSyncDatabase.newRowSpace]):
/// one without a transaction, on a client with a persistent user.
bool _routesWrite(Transaction? transaction) =>
transaction == null && _recorder.persistentUserId != null;
/// The space other than the personal one that new [rows] go in, as
/// [OfflineSyncDatabase.newRowSpace] says; null for the personal one.
/// Throws a [StateError] where the rows go in two spaces.
Future<UuidValue?> _newRowsSpace<T extends TableRow>(List<T> rows) async {
final route = newRowSpace;
if (route == null) return null;
final personal = _recorder.persistentUserId!;
final spaces = <UuidValue>{
for (final row in rows)
if (_readSpaceId(row) == null) await route(row) ?? personal,
};
if (spaces.length > 1) {
throw StateError(
'Cannot write new ${rows.first.table.tableName} rows into the spaces '
'$spaces in one call.',
);
}
final space = spaces.singleOrNull;
return space == personal ? null : space;
}
/// The space other than the personal one that the visible rows matching
/// [where] are stored in; null where they are personal, or none match.
/// Throws a [StateError] where they are stored in two spaces.
Future<UuidValue?> _storedRowsSpace<T extends TableRow>(Expression where) async {
final rows = await _delegate.find<T>(
where: await _whereVisibleWithTombstone<T>(
where,
null,
null,
membershipWide: true,
),
);
final spaceIds = {
for (final row in rows)
if (_readSpaceId(row) case final int spaceId) spaceId,
};
if (spaceIds.isEmpty) return null;
if (spaceIds.length > 1) {
throw StateError(
'Cannot write ${rows.first.table.tableName} rows of several spaces in '
'one call.',
);
}
final space = await OfflineSyncSpace.db.findById(
_delegate.session,
spaceIds.single,
);
final uuid = space?.uuidSpaceId;
return uuid == null || uuid == _recorder.persistentUserId ? null : uuid;
}
OfflineSyncSpace _requireEffectiveSpace(Transaction transaction) { OfflineSyncSpace _requireEffectiveSpace(Transaction transaction) {
return _recorder.spaceForQueries(transaction) ?? return _recorder.spaceForQueries(transaction) ??
(throw StateError( (throw StateError(

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

@ -62,16 +62,6 @@ class OfflineSyncEngine {
/// Delay between continuous sync rounds. /// Delay between continuous sync rounds.
this._continuousSyncInterval = defaultContinuousSyncInterval, this._continuousSyncInterval = defaultContinuousSyncInterval,
/// Whether [collectPendingChanges] hands out each change as soon as it is
/// read, its transaction open until the last, rather than once all of
/// them are. A large collection then starts sending at once, where
/// whole it sent nothing until its end, so a peer waiting longer than its
/// idle timeout closed the stream. Meant for PostgreSQL, whose
/// repeatable-read snapshot only reads: Serverpod's SQLite transactions
/// are write transactions, which would hold off the device's own writes
/// for as long as the peer takes to receive.
this._streamCollectedChanges = false,
}) : _syncTables = syncTables, }) : _syncTables = syncTables,
_serializationManager = serializationManager, _serializationManager = serializationManager,
_databaseContext = _databaseContext =
@ -97,7 +87,6 @@ class OfflineSyncEngine {
final OfflineSyncDatabaseContext _databaseContext; final OfflineSyncDatabaseContext _databaseContext;
final int _syncBatchSize; final int _syncBatchSize;
final Duration _continuousSyncInterval; final Duration _continuousSyncInterval;
final bool _streamCollectedChanges;
/// Wraps [database] in a CRDT-aware database using this sync context. /// Wraps [database] in a CRDT-aware database using this sync context.
OfflineSyncDatabase wrapDatabase(Database database, {UuidValue? persistentUserId}) { OfflineSyncDatabase wrapDatabase(Database database, {UuidValue? persistentUserId}) {
@ -197,16 +186,6 @@ class OfflineSyncEngine {
}; };
try { try {
if (_streamCollectedChanges) {
await for (final change in _collectedAsRead(
session,
spaceUuidById,
checkpointsBySpaceId,
)) {
yield change;
}
return;
}
final changes = await session.db.transaction( final changes = await session.db.transaction(
(transaction) => _streamPendingChanges( (transaction) => _streamPendingChanges(
session, session,
@ -214,7 +193,9 @@ class OfflineSyncEngine {
checkpointsBySpaceId, checkpointsBySpaceId,
transaction, transaction,
).toList(), ).toList(),
settings: _collectionSettings, settings: const TransactionSettings(
isolationLevel: IsolationLevel.repeatableRead,
),
); );
for (final change in changes) { for (final change in changes) {
@ -225,54 +206,6 @@ class OfflineSyncEngine {
} }
} }
static const _collectionSettings = TransactionSettings(
isolationLevel: IsolationLevel.repeatableRead,
);
/// The pending changes, each handed out as it is read within the
/// collection's one transaction, which stays open until the last is out or
/// the listener cancels; an error, a violation among them, ends the stream
/// once the transaction has rolled back.
Stream<CrdtMergeChange> _collectedAsRead(
DatabaseSession session,
Map<int, UuidValue> spaceUuidById,
Map<int, List<Hlc>> checkpointsBySpaceId,
) {
var cancelled = false;
late final StreamController<CrdtMergeChange> changes;
changes = StreamController<CrdtMergeChange>(
onListen: () {
unawaited(
session.db
.transaction(
(transaction) async {
await for (final change in _streamPendingChanges(
session,
spaceUuidById,
checkpointsBySpaceId,
transaction,
)) {
// Nobody reads on: end the collection and its transaction.
if (cancelled) return;
changes.add(change);
}
},
settings: _collectionSettings,
)
.then(
(_) => changes.close(),
onError: (Object error, StackTrace stackTrace) {
if (!cancelled) changes.addError(error, stackTrace);
return changes.close();
},
),
);
},
onCancel: () => cancelled = true,
);
return changes.stream;
}
/// Creates the [OfflineSyncSinceHlc] checkpoint for a space handshake. /// Creates the [OfflineSyncSinceHlc] checkpoint for a space handshake.
/// ///
/// [OfflineSyncSinceHlc.nodeCheckpoints] reflects the latest change this node /// [OfflineSyncSinceHlc.nodeCheckpoints] reflects the latest change this node

7
packages/serverpod_offline_sync_server/lib/src/business/offline_sync.dart
View file

@ -37,23 +37,16 @@ extension OfflineSyncInitialize on Serverpod {
/// ///
/// [continuousSyncInterval] controls how long a continuous sync session waits /// [continuousSyncInterval] controls how long a continuous sync session waits
/// after completing one sync round before checking for local changes again. /// after completing one sync round before checking for local changes again.
///
/// [streamCollectedChanges] sends the changes a peer lacks as they are read
/// rather than once all are, so a peer's first sync of a large space hears
/// from the server before its idle timeout; the server's PostgreSQL holds
/// the collection's read-only snapshot meanwhile.
void initializeOfflineSync({ void initializeOfflineSync({
required List<Table> syncTables, required List<Table> syncTables,
int syncBatchSize = OfflineSyncEngine.defaultSyncBatchSize, int syncBatchSize = OfflineSyncEngine.defaultSyncBatchSize,
Duration continuousSyncInterval = OfflineSyncEngine.defaultContinuousSyncInterval, Duration continuousSyncInterval = OfflineSyncEngine.defaultContinuousSyncInterval,
bool streamCollectedChanges = true,
}) { }) {
_offlineSyncByServerpod[this] = OfflineSyncEngine( _offlineSyncByServerpod[this] = OfflineSyncEngine(
syncTables: syncTables, syncTables: syncTables,
serializationManager: serializationManager, serializationManager: serializationManager,
syncBatchSize: syncBatchSize, syncBatchSize: syncBatchSize,
continuousSyncInterval: continuousSyncInterval, continuousSyncInterval: continuousSyncInterval,
streamCollectedChanges: streamCollectedChanges,
); );
} }
} }

217
test/serverpod_offline_sync_test_server/test/integration/crud/space_routing_test.dart
View file

@ -1,217 +0,0 @@
import 'package:serverpod_offline_sync_client/serverpod_offline_sync_client.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';
void main() {
initTestClientSession(withPersistentUser: true);
group(
'Given a persistent user who is a member of a shared space holding a row, '
'and a row of their own,',
() {
late UuidValue sharedSpaceUuid;
late OfflineSyncSpace sharedSpace;
late Person shared;
late Person own;
/// The space each person row is stored in, by name, read past the
/// engine.
Future<Map<String, int?>> storedSpaces() async => {
for (final person in await Person.db.find(testSession))
person.name: person.spaceId,
};
Future<void> joinAs(OfflineSyncSpaceRole role) async {
await OfflineSyncSpaceMember.db.insertRow(
session,
OfflineSyncSpaceMember(
spaceId: sharedSpace.id!,
userUuid: testCrdtUserId,
role: role,
),
);
}
setUp(() async {
sharedSpaceUuid = const Uuid().v7obj();
shared = await session.db.transactionForUser(
sharedSpaceUuid,
(tx) => Person.db.insertRow(
session,
Person(name: 'shared'),
transaction: tx,
),
);
sharedSpace = (await OfflineSyncSpace.db.findFirstRow(
testSession,
where: (t) => t.uuidSpaceId.equals(sharedSpaceUuid),
))!;
own = await Person.db.insertRow(session, Person(name: 'own'));
});
test(
'when a writing member updates and deletes the shared row without a '
'transaction, then it changes in its own space.',
() async {
await joinAs(OfflineSyncSpaceRole.readWrite);
final updated = await Person.db.updateRow(
session,
shared.copyWith(name: 'shared, changed'),
);
expect(updated.name, 'shared, changed');
expect((await storedSpaces())['shared, changed'], sharedSpace.id);
await Person.db.deleteRow(session, updated);
expect(await Person.db.findById(session, shared.id!), isNull);
expect(await Person.db.findById(session, own.id!), isNotNull);
},
);
test(
'when a writing member updates the shared rows by a condition, then '
'they change in their own space.',
() async {
await joinAs(OfflineSyncSpaceRole.readWrite);
await Person.db.updateWhere(
session,
columnValues: (t) => [t.name('renamed')],
where: (t) => t.name.equals('shared'),
);
await Person.db.deleteWhere(
session,
where: (t) => t.name.equals('renamed'),
);
expect(
[for (final p in await Person.db.find(session)) p.name],
['own'],
);
},
);
test(
'when a reading member updates or deletes the shared row, then it is '
'refused and stays as it was.',
() async {
await joinAs(OfflineSyncSpaceRole.readOnly);
await expectLater(
Person.db.updateRow(session, shared.copyWith(name: 'changed')),
throwsA(isA<OfflineSyncSpaceRoleException>()),
);
await expectLater(
Person.db.deleteRow(session, shared),
throwsA(isA<OfflineSyncSpaceRoleException>()),
);
expect((await Person.db.findById(session, shared.id!))?.name, 'shared');
},
);
test(
'when one call names rows of two spaces, then it is refused and '
'nothing changes.',
() async {
await joinAs(OfflineSyncSpaceRole.readWrite);
await expectLater(
Person.db.update(session, [
shared.copyWith(name: 'a'),
own.copyWith(name: 'b'),
]),
throwsA(isA<StateError>()),
);
expect(
{for (final p in await Person.db.find(session)) p.name},
{'shared', 'own'},
);
},
);
test(
'when the host routes a new row to the shared space, then it is '
'stored there; one it routes nowhere stays personal.',
() async {
await joinAs(OfflineSyncSpaceRole.readWrite);
session.db.newRowSpace = (row) =>
(row as Person).name.startsWith('to shared') ? sharedSpaceUuid : null;
await Person.db.insertRow(session, Person(name: 'to shared'));
await Person.db.insert(session, [Person(name: 'to shared, too')]);
await Person.db.insertRow(session, Person(name: 'mine'));
final spaces = await storedSpaces();
expect(spaces['to shared'], sharedSpace.id);
expect(spaces['to shared, too'], sharedSpace.id);
expect(spaces['mine'], spaces['own']);
},
);
test(
'when the host routes new rows of one call to two spaces, then it is '
'refused.',
() async {
await joinAs(OfflineSyncSpaceRole.readWrite);
session.db.newRowSpace = (row) =>
(row as Person).name == 'there' ? sharedSpaceUuid : null;
await expectLater(
Person.db.insert(session, [Person(name: 'there'), Person(name: 'here')]),
throwsA(isA<StateError>()),
);
expect(
{for (final p in await Person.db.find(session)) p.name},
{'shared', 'own'},
);
},
);
test(
'when the host routes a new row to a space the person may only read, '
'or is not a member of, then it is refused.',
() async {
await joinAs(OfflineSyncSpaceRole.readOnly);
final elsewhere = const Uuid().v7obj();
session.db.newRowSpace = (row) =>
(row as Person).name == 'read only' ? sharedSpaceUuid : elsewhere;
await expectLater(
Person.db.insertRow(session, Person(name: 'read only')),
throwsA(isA<OfflineSyncSpaceRoleException>()),
);
await expectLater(
Person.db.insertRow(session, Person(name: 'nowhere')),
throwsA(isA<OfflineSyncSpaceMembershipException>()),
);
expect(
{for (final p in await Person.db.find(session)) p.name},
{'shared', 'own'},
);
},
);
test(
'when a caller writes within a transaction of its own, then the '
'transaction decides the space, as before.',
() async {
await joinAs(OfflineSyncSpaceRole.readWrite);
session.db.newRowSpace = (_) => sharedSpaceUuid;
await session.db.transaction((tx) async {
await Person.db.insertRow(
session,
Person(name: 'in a transaction'),
transaction: tx,
);
});
final spaces = await storedSpaces();
expect(spaces['in a transaction'], spaces['own']);
},
);
},
);
}

7
test/serverpod_offline_sync_test_server/test/integration/crud/tombstone_space_test.dart
View file

@ -369,17 +369,13 @@ void main() {
); );
}); });
// Collected whole, as a device does, and sent while collected, as the
// server does: a violation ends either alike.
for (final streamed in [false, true]) {
test( test(
'when collecting pending changes${streamed ? ' as they are read' : ''}, ' 'when collecting pending changes, '
'then the sync fails and records orphan metadata.', 'then the sync fails and records orphan metadata.',
() async { () async {
final sync = OfflineSyncEngine( final sync = OfflineSyncEngine(
syncTables: [Person.t], syncTables: [Person.t],
serializationManager: session.db.serializationManager, serializationManager: session.db.serializationManager,
streamCollectedChanges: streamed,
); );
await expectLater( await expectLater(
@ -436,7 +432,6 @@ void main() {
expect(violation.occurrences, 1); expect(violation.occurrences, 1);
}, },
); );
}
}); });
group('Given update CRDT metadata whose domain row is missing,', () { group('Given update CRDT metadata whose domain row is missing,', () {

173
test/serverpod_offline_sync_test_server/test/integration/server/streamed_collection_postgres_test.dart
View file

@ -1,173 +0,0 @@
import 'dart:io';
import 'dart:typed_data';
import 'package:serverpod/serverpod.dart';
import 'package:serverpod_offline_sync_server/serverpod_offline_sync_server.dart';
import 'package:serverpod_offline_sync_test_server/src/generated/protocol.dart'
as server;
import 'package:test/test.dart';
import '../test_tools/postgres_migrations.dart';
import '../test_tools/query_log_database.dart';
import '../test_tools/serverpod_test_tools.dart';
const _heldCount = 600;
const _newCount = 600;
const _renamedCount = 300;
/// The server's collection, sent as it is read, on the database it runs on:
/// its repeatable-read snapshot stays open while the changes go out.
void main() {
final serverDirectory = Directory(
'${Directory.systemTemp.path}/offline_sync_streamed_${const Uuid().v4()}',
);
setUpAll(() => preparePostgresMigrations(serverDirectory));
tearDownAll(() async {
if (serverDirectory.existsSync()) await serverDirectory.delete(recursive: true);
});
withServerpod(
'[PostgreSQL streamed collection]',
rollbackDatabase: RollbackDatabase.disabled,
serverDirectory: serverDirectory,
configOverride: (config) => config.copyWith(
database: PostgresDatabaseConfig.embedded(
dataPath: '${serverDirectory.path}/postgres',
name: 'serverpod_test',
maxConnectionCount: 6,
),
),
(sessionBuilder, _) {
final tables = [server.Types.t];
late UuidValue space;
late OfflineSyncDatabaseSession author;
late QueryLogDatabase queryLog;
late OfflineSyncDatabaseSession collector;
late List<Hlc> checkpoints;
OfflineSyncEngine engine({required bool streamed}) => OfflineSyncEngine(
syncTables: tables,
serializationManager: author.db.serializationManager,
streamCollectedChanges: streamed,
);
Stream<CrdtMergeChange> collect({required bool streamed}) =>
engine(streamed: streamed).collectPendingChanges(
collector,
checkpointsBySpaceUuid: {space: checkpoints},
);
setUpAll(() async {
space = const Uuid().v7obj();
author = OfflineSyncDatabaseSession.wraps(
sessionBuilder.build(),
syncTables: tables,
);
await author.db.initialize();
final held = await _insertTypes(author, space, 0, _heldCount);
checkpoints = (await engine(streamed: false).createSyncSinceHlc(
author,
spaceId: space,
)).nodeCheckpoints;
await _insertTypes(author, space, _heldCount, _newCount);
await author.db.transactionForUser(
space,
(tx) => server.Types.db.update(
author,
[
for (final row in held.take(_renamedCount))
row.copyWith(aText: 'renamed ${row.anInt}'),
],
columns: (t) => [t.aText],
transaction: tx,
),
);
queryLog = QueryLogDatabase(sessionBuilder.build().db, syncTables: tables);
collector = OfflineSyncDatabaseSession(queryLog, syncTables: tables);
await queryLog.initialize();
});
group('Given changes a peer lacks, collected as they are read,', () {
late int readsAtFirstChange;
late int readsInAll;
late List<CrdtMergeChange> streamed;
setUpAll(() async {
queryLog.rawQueries.clear();
int? atFirst;
streamed = [];
var wroteMeanwhile = false;
await for (final change in collect(streamed: true)) {
atFirst ??= queryLog.readsOf('types').length;
streamed.add(change);
if (!wroteMeanwhile) {
wroteMeanwhile = true;
// Another session writes while the collection runs.
await _insertTypes(author, space, _heldCount + _newCount, 1);
}
}
readsAtFirstChange = atFirst!;
readsInAll = queryLog.readsOf('types').length;
});
test('then the first change arrives before the renamed rows are read.', () {
expect(readsAtFirstChange, lessThan(readsInAll));
});
test('then a row written meanwhile stays out of the collection.', () {
expect(streamed.whereType<CrdtMergeInsert>(), hasLength(_newCount));
expect(streamed.whereType<CrdtMergeUpdate>(), hasLength(_renamedCount));
});
test('then the changes are those a whole collection of the same rows '
'holds.', () async {
final whole = await collect(streamed: false).toList();
final protocol = author.db.serializationManager;
String encoded(CrdtMergeChange change) =>
protocol.encodeWithTypeForProtocol(change).toString();
// The whole collection also holds the row written meanwhile.
final wholeEncoded = whole.map(encoded).toList();
final streamedEncoded = streamed.map(encoded).toList();
expect(wholeEncoded, containsAllInOrder(streamedEncoded));
expect(wholeEncoded, hasLength(streamedEncoded.length + 1));
});
});
group('Given a listener that stops after the first change,', () {
test('then the collection ends and the next one runs.', () async {
final first = await collect(streamed: true).first;
expect(first, isA<CrdtMergeInsert>());
final again = await collect(streamed: true).length;
expect(again, _newCount + 1 + _renamedCount);
});
});
},
);
}
Future<List<server.Types>> _insertTypes(
OfflineSyncDatabaseSession node,
UuidValue space,
int from,
int count,
) => node.db.transactionForUser(
space,
(tx) => server.Types.db.insert(
node,
[
for (var i = from; i < from + count; i++)
server.Types(
aBool: i.isEven,
aDateTime: DateTime.utc(2026, 11, 9).add(Duration(minutes: i)),
aText: 'row $i',
anInt: i,
anInt64: BigInt.from(i) << 40,
aReal: i / 8,
aBlob: Uint8List.fromList([i % 256, 0, 255]).buffer.asByteData(),
),
],
transaction: tx,
),
);

184
test/serverpod_offline_sync_test_server/test/integration/sync/streamed_collection_test.dart
View file

@ -1,184 +0,0 @@
import 'dart:async';
import 'dart:typed_data';
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/query_log_database.dart';
import '../test_tools/sync_topology.dart';
const _heldCount = 600;
const _newCount = 600;
const _renamedCount = 300;
/// A peer that synced [_heldCount] rows long ago collects what happened
/// since: [_newCount] rows added and [_renamedCount] of the old ones renamed.
/// The inserts are handed out one by one once their rows are read; the
/// renames' rows are read only after the last insert, so the query log tells
/// whether a change reached the consumer while the collection still ran.
void main() {
initTestClientSession(createSessionPerTest: false);
for (final streamed in [true, false]) {
group(
'Given a peer $_newCount inserts and $_renamedCount renames behind, '
'collected ${streamed ? 'as they are read' : 'whole'},',
() {
late SyncNode author;
late SyncNode peer;
late QueryLogDatabase queryLog;
late OfflineSyncDatabaseSession collector;
late OfflineSyncEngine engine;
late List<Hlc> checkpoints;
setUpAll(() async {
author = await syncNode(await createAdditionalTestSession(), [Types.t]);
peer = await syncNode(await createAdditionalTestSession(), [Types.t]);
final held = await _insertTypes(author, 0, _heldCount);
await pushChanges(author, peer);
checkpoints = (await peer.sync.createSyncSinceHlc(
peer.raw,
spaceId: testCrdtUserId,
)).nodeCheckpoints;
await _insertTypes(author, _heldCount, _newCount);
await _rename(author, held.take(_renamedCount));
queryLog = QueryLogDatabase(author.raw.db, syncTables: [Types.t]);
collector = OfflineSyncDatabaseSession(queryLog, syncTables: [Types.t]);
await queryLog.initialize();
engine = OfflineSyncEngine(
syncTables: [Types.t],
serializationManager: author.raw.db.serializationManager,
streamCollectedChanges: streamed,
);
});
group('when the peer collects since its checkpoint,', () {
late int readsAtFirstChange;
late int readsInAll;
late List<CrdtMergeChange> changes;
late Map<UuidValue, Map<String, dynamic>> authored;
late Map<UuidValue, Map<String, dynamic>> received;
setUpAll(() async {
queryLog.rawQueries.clear();
int? atFirst;
changes = [];
await for (final change in engine.collectPendingChanges(
collector,
checkpointsBySpaceUuid: {testCrdtUserId: checkpoints},
)) {
atFirst ??= queryLog.readsOf('types').length;
changes.add(change);
}
readsAtFirstChange = atFirst!;
readsInAll = queryLog.readsOf('types').length;
await peer.offlineSync.db.mergeChanges(
_overTheWire(author, changes),
spaceId: testCrdtUserId,
);
authored = await _rowsById(author);
received = await _rowsById(peer);
});
if (streamed) {
test('then the first change arrives before the renamed rows are read.', () {
expect(readsAtFirstChange, lessThan(readsInAll));
});
} else {
test('then the first change arrives once every row is read.', () {
expect(readsAtFirstChange, readsInAll);
});
}
test('then every insert and every rename is collected.', () {
expect(changes.whereType<CrdtMergeInsert>(), hasLength(_newCount));
expect(changes.whereType<CrdtMergeUpdate>(), hasLength(_renamedCount));
});
test("then the peer holds exactly the author's rows.", () {
expect(received, authored);
});
});
if (streamed) {
group('when the consumer stops after the first change,', () {
test('then the collection ends without an error.', () async {
final first = await engine
.collectPendingChanges(
collector,
checkpointsBySpaceUuid: {testCrdtUserId: checkpoints},
)
.first;
expect(first, isA<CrdtMergeInsert>());
// The collection's transaction ends: the next one isn't held up.
final again = await engine
.collectPendingChanges(
collector,
checkpointsBySpaceUuid: {testCrdtUserId: checkpoints},
)
.length;
expect(again, _newCount + _renamedCount);
});
});
}
},
);
}
}
Future<List<Types>> _insertTypes(SyncNode node, int from, int count) =>
node.offlineSync.db.transactionForUser(
testCrdtUserId,
(tx) => Types.db.insert(
node.offlineSync,
[
for (var i = from; i < from + count; i++)
Types(
aBool: i.isEven,
aDateTime: DateTime.utc(2026, 11, 9).add(Duration(minutes: i)),
aText: 'row $i',
anInt: i,
anInt64: BigInt.from(i) << 40,
aReal: i / 8,
aBlob: Uint8List.fromList([i % 256, 0, 255]).buffer.asByteData(),
jsonbDocument: SyncDocument(
title: 'JSONB $i',
enabled: i.isEven,
numbers: [i],
),
),
],
transaction: tx,
),
);
Future<void> _rename(SyncNode node, Iterable<Types> rows) =>
node.offlineSync.db.transactionForUser(
testCrdtUserId,
(tx) => Types.db.update(
node.offlineSync,
[for (final row in rows) row.copyWith(aText: 'renamed ${row.anInt}')],
columns: (t) => [t.aText],
transaction: tx,
),
);
/// The changes as the peer decodes them from the sync stream.
List<CrdtMergeChange> _overTheWire(SyncNode node, List<CrdtMergeChange> changes) {
final protocol = node.raw.db.serializationManager;
return [
for (final change in changes)
protocol.decodeWithType(protocol.encodeWithTypeForProtocol(change))!
as CrdtMergeChange,
];
}
/// The model's JSON, without the space it is stored in locally.
Map<String, dynamic> _payload(Types row) => row.toJson()..remove('spaceId');
Future<Map<UuidValue, Map<String, dynamic>>> _rowsById(SyncNode node) async {
final rows = await Types.db.find(node.offlineSync);
return {for (final row in rows) row.id!: _payload(row)};
}