onWelcome method

Future<bool> onWelcome(
  1. RelayWelcomeMessage message
)

Imports the room state of a welcome, then queues everything the relay does not hold.

The state is merged (merge: true) into the document: on a reconnect the document already holds local state that must not be clobbered. History is kept (pruneHistory: false) so this client can later upload a snapshot covering it.

A welcome carries the whole room — the snapshot plus the log after it — so its version vector is exactly what the relay has. Whatever the document holds beyond that is queued and pushed, whoever wrote it. That covers three cases with one rule: unacknowledged changes that survived the reconnect, changes restored from storage after a restart (which reach the document as imported ones, so nothing else would ever push them), and changes of another peer the relay lost.

One limit comes with the version vector: it cannot describe a hole in the middle of one peer's sequence, only how far that peer got.

Re-delivering a change the relay already had is harmless: the relay appends it and every peer discards it as known.

Returns whether the room state went in; on false the join is abandoned and nothing is pushed.

A state this client cannot take in becomes a SyncFault. Reported on CRDTSocketClient.faults, never thrown: this runs inside the callback that reads the socket, where a throw reaches no catch and no onError and ends up as an uncaught error in the zone.

A relay checks the protocol version and nothing else, so it is this path, not the CRDT-aware one, that meets unreadable state most often.

Implementation

Future<bool> onWelcome(RelayWelcomeMessage message) async {
  final Snapshot? snapshot;
  final List<Change> changes;

  try {
    snapshot = message.snapshot != null
        ? Snapshot.fromBytes(base64Decode(message.snapshot!))
        : null;
    changes = [
      for (final blob in message.changes)
        Change.fromBytes(base64Decode(blob)),
    ];

    document.import(
      snapshot: snapshot,
      changes: changes,
      merge: true,
      pruneHistory: false,
    );
  } catch (error, stackTrace) {
    client.reportSyncFault(
      SyncFault(
        reason: 'Could not import the room state the relay served',
        error: error,
        stackTrace: stackTrace,
      ),
    );
    return false;
  }

  _seqTracker.markThrough(message.seq);
  _handshaken = true;
  _queue.resetInFlight();

  _queueUnknownToRelay(snapshot, changes);

  await flush();

  if (message.compact) {
    await uploadSnapshot(message.seq);
  }
  return true;
}