feat: Report merges of inbound changes apart from pushes

onMergeSuccess is called for every space a sync touched, also for one
this peer only sent changes of, so a client re-read its data after every
sync that pushed. onInboundMerge is called only for a space whose
changes from the remote peer were merged.

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

View file

@ -120,6 +120,7 @@ class OfflineSyncDatabase implements Database {
UuidValue? userId,
bool once = false,
OfflineSyncOnMergeSuccess? onMergeSuccess,
OfflineSyncOnInboundMerge? onInboundMerge,
}) async* {
await _ensureInitialized();
final effectiveUserId = await _requireUserId(userId);
@ -129,6 +130,7 @@ class OfflineSyncDatabase implements Database {
inbound: inbound,
once: once,
onMergeSuccess: onMergeSuccess,
onInboundMerge: onInboundMerge,
mode: mode,
);
}

View file

@ -43,8 +43,14 @@ class OfflineSyncClient {
Future<void> syncOnce(
DatabaseSession session, {
OfflineSyncOnMergeSuccess? onMergeSuccess,
OfflineSyncOnInboundMerge? onInboundMerge,
}) 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
@ -54,12 +60,19 @@ class OfflineSyncClient {
OfflineSyncSubscription syncContinuously(
DatabaseSession session, {
OfflineSyncOnMergeSuccess? onMergeSuccess,
}) => _startSyncSession(session, once: false, onMergeSuccess: onMergeSuccess);
OfflineSyncOnInboundMerge? onInboundMerge,
}) => _startSyncSession(
session,
once: false,
onMergeSuccess: onMergeSuccess,
onInboundMerge: onInboundMerge,
);
OfflineSyncSubscription _startSyncSession(
DatabaseSession session, {
required bool once,
OfflineSyncOnMergeSuccess? onMergeSuccess,
OfflineSyncOnInboundMerge? onInboundMerge,
}) {
// The stream will be closed using [outboundChanges.closeOrSkip].
// ignore: close_sinks
@ -78,6 +91,7 @@ class OfflineSyncClient {
inbound: remoteStream,
once: once,
onMergeSuccess: onMergeSuccess,
onInboundMerge: onInboundMerge,
mode: OfflineSyncPeerMode.follower,
)
.listen(

View file

@ -27,6 +27,12 @@ export 'space_state.dart' show OfflineSyncPeerMode;
typedef OfflineSyncOnMergeSuccess =
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.
typedef DomainRowOwner = ({bool exists, int? spaceId});
@ -275,6 +281,7 @@ class OfflineSyncEngine {
required OfflineSyncPeerMode mode,
bool once = false,
OfflineSyncOnMergeSuccess? onMergeSuccess,
OfflineSyncOnInboundMerge? onInboundMerge,
}) async* {
// Idle timeouts are a continuous-only affordance: they let an idle cycle
// settle into an empty batch without closing the stream. A `once` session
@ -369,6 +376,7 @@ class OfflineSyncEngine {
batch,
outboundSpaces,
onMergeSuccess,
onInboundMerge,
);
if (once) {
@ -429,6 +437,7 @@ class OfflineSyncEngine {
OfflineSyncCycleBatch batch,
Set<UuidValue> outboundSpaces,
OfflineSyncOnMergeSuccess? onMergeSuccess,
OfflineSyncOnInboundMerge? onInboundMerge,
) async {
if (batch.spaceSet != null) {
await spaces.adoptPeerGrants(batch.spaceSet!.spaces);
@ -462,6 +471,9 @@ class OfflineSyncEngine {
mergeSet: entry.value,
);
await _reportMerge(onMergeSuccess, spaces, spaceId, receivedHlc);
if (entry.value.isNotEmpty) {
await onInboundMerge?.call(spaceId);
}
}
for (final spaceId in outboundSpaces.difference(mergedSpaces)) {
await _reportMerge(onMergeSuccess, spaces, spaceId, null);