fix: Rebuild projections once per context after the registry changed

Every ephemeral OfflineSyncDatabase shares one context but initializes a
recorder of its own, and each rebuilt the projections of every space
when the context's first initialize had changed the registry: on every
session and every merge, until the process stopped (upstream #150). A
new device, or one whose app changed its synced tables, spent about 3 s
of every inbound merge on it with a library of 8,000 cards, a server
after such a migration as much on every request.

The context now runs the rebuild once (Once): wrappers initializing
while it runs await the same run, later ones skip it, and a failed run
is tried again by the next.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Tizian 「ティツィアーン」 2026-10-17 16:51:28 +02:00
commit f062725a16
Signed by: raisondetredev
GPG key ID: 07F6FCF405488B64
3 changed files with 83 additions and 3 deletions

View file

@ -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!);
} }

View 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;
}
}
}

View 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);
});
});
}