waitUntilCaughtUp method

Future<void> waitUntilCaughtUp({ Duration? timeout, void onProgress(SyncProgress progress)?, })

Returns a Future that completes once the local commit id has caught up to the remote commit id as it stood at the time of this call. Captures the remote commit id reported by the first SyncProgress event that fires after the call and completes the future once a subsequent event reports localCommitId >= snapshotServerCommitId. New remote changes that arrive after the call do NOT extend the wait.

If local is already caught up at the time of the next sync iteration (the iteration short-circuits with "server and local are in sync", emitting a SyncStatus.success event with both commit ids null), the future completes immediately.

On a SyncService that emits per-batch SyncStatus.inProgress events (see SyncServiceImpl from at_client 3.x onward), completion can fire mid-iteration as soon as localCommitId crosses the captured snapshot — no need to wait for the entire pull/push run to finish.

onProgress, when supplied, is invoked for every SyncProgress event observed while waiting (including the SyncStatus.inProgress per-batch events on the pull and push paths). Use it to drive "syncing N of M" UIs — typically by reading progress.localCommitId against progress.serverCommitId. The callback fires before the catch-up completion check, so apps see the same final event the future completes on. Exceptions thrown by onProgress are caught and logged; they do not affect the completion logic.

Transient sync failures while waiting are tolerated — the returned future does not reject on a SyncStatus.failure event; callers bound the wait by passing timeout. With no timeout and a stalled sync (e.g. network loss for the duration), the future hangs.

Throws TimeoutException when timeout elapses before catch-up.

Implementation

Future<void> waitUntilCaughtUp({
  Duration? timeout,
  void Function(SyncProgress progress)? onProgress,
}) {
  final completer = Completer<void>();
  int? snapshotServerCommitId;
  late final SyncProgressListener listener;

  void detach() {
    // Defer removal so we don't mutate the listener list while the
    // implementation is iterating it on the same call frame.
    Future.microtask(() => removeProgressListener(listener));
  }

  listener = _ClosureProgressListener((progress) {
    // Surface every event to the caller's progress callback first,
    // before the catch-up completion check, so apps see the same
    // final event we use to fulfil the future. Defensive try/catch:
    // a misbehaving callback must not derail the completion logic.
    if (onProgress != null) {
      try {
        onProgress(progress);
      } catch (e, st) {
        _waitLogger
            .warning('onProgress callback threw and was swallowed: $e\n$st');
      }
    }
    if (completer.isCompleted) return;
    // Don't complete while there are still pending client→server
    // pushes in the local sync queue. commit-id equality alone is
    // not sufficient: writes that arrived in the queue AFTER the
    // sync round started won't have been pushed in the round we're
    // currently observing, even though localCommitId may have
    // caught up to the round's snapshot serverCommitId. A
    // null `pendingPushCount` (older event shape, or queue not
    // opened yet) is treated as zero pending — i.e. don't block
    // on it.
    final pending = progress.pendingPushCount ?? 0;
    if (pending > 0) {
      _waitLogger.finer('waitUntilCaughtUp: $pending pending pushes — '
          'not caught up yet');
      return;
    }
    if (snapshotServerCommitId == null) {
      // The "server and local are in sync" early-exit fires a
      // SyncStatus.success event with both commit ids null — treat
      // that as "already caught up at call time" and complete.
      if (progress.syncStatus == SyncStatus.success &&
          progress.serverCommitId == null &&
          progress.localCommitId == null) {
        detach();
        completer.complete();
        return;
      }
      // Wait for an event that actually carries commit ids.
      if (progress.serverCommitId == null) return;
      snapshotServerCommitId = progress.serverCommitId;
    }
    _waitLogger.finer('serverCommitId (target): $snapshotServerCommitId'
        ' localCommitId (actual): ${progress.localCommitId}');
    final local = progress.localCommitId;
    if (local != null && local >= snapshotServerCommitId!) {
      detach();
      completer.complete();
    }
  });

  addProgressListener(listener);
  // Trigger an iteration so a progress event is guaranteed to fire
  // even when nothing else is currently driving sync.
  sync();

  if (timeout == null) return completer.future;
  return completer.future.timeout(timeout, onTimeout: () {
    detach();
    throw TimeoutException(
      'waitUntilCaughtUp timed out after $timeout',
      timeout,
    );
  });
}