Struct SyncClient 

pub struct SyncClient { /* private fields */ }

Sync client — manages the WebSocket connection to Origin.

The client runs as a background Tokio task. It:

1. Connects to Origin via WebSocket
2. Sends handshake with JWT + vector clock + shape subscriptions
3. Pushes accumulated CRDT deltas
4. Receives shape snapshots and incremental deltas
5. Handles rejections via CompensationRegistry
6. Auto-reconnects with exponential backoff on disconnect

Implementations

impl SyncClient

pub async fn build_delta_pushes( &self, pending: &[PendingDelta], peer_id: u64, ) -> Vec<DeltaPushMsg>

Build DeltaPush messages from pending deltas.

Respects the flow control window: returns at most next_batch_size() deltas. Each message includes a CRC32C checksum of the delta payload for integrity verification at Origin.

peer_id is passed in rather than held on the client: it changes when a collision refusal rotates this replica’s identity, and a delta pushed under the previous one is refused again.

pub async fn record_push(&self, mutation_ids: &[u64])

Record that deltas were pushed (update flow control in-flight tracking).

pub async fn handle_delta_ack(&self, ack: &DeltaAckMsg)

Process a DeltaAck from Origin.

pub async fn handle_delta_reject(&self, reject: &DeltaRejectMsg)

Process a DeltaReject from Origin.

impl SyncClient

pub async fn build_handshake(&self, identity: &LiteIdentity) -> HandshakeMsg

Build a handshake message from current state and the instance identity.

The identity is a parameter rather than client state because it changes while the client is alive: a reported fork replaces it wholesale, and a peer-id collision rotates part of it. A handshake built from a cached copy would re-present the identity Origin just rejected.

pub async fn handle_handshake_ack(&self, ack: &HandshakeAckMsg) -> bool

Process a handshake acknowledgment from Origin.

impl SyncClient

pub fn build_ping(&self) -> Option<SyncFrame>

Build a ping frame.

pub fn backoff_duration(&self, attempt: u32) -> Duration

Calculate backoff duration for reconnection attempt N.

pub async fn update_pending_stats(&self, count: usize, bytes: usize)

Update pending queue stats in the flow controller. Called from the push loop after reading pending deltas.

pub async fn is_queue_full(&self) -> bool

Check if the pending queue is at capacity (flow control).

pub async fn sync_metrics(&self) -> SyncMetricsSnapshot

Get a snapshot of sync metrics for monitoring/health.

pub async fn reset_flow_control(&self)

Reset flow controller on reconnect.

impl SyncClient

pub async fn handle_shape_snapshot(&self, msg: &ShapeSnapshotMsg)

Process a ShapeSnapshot from Origin.

Marks the shape as snapshot-loaded and re-bases the sequence tracker so that gap detection restarts from the snapshot LSN rather than any stale watermark from before the resync. Also clears the resync gate so a future gap on this or another shape can trigger a new request.

pub async fn handle_shape_delta(&self, msg: &ShapeDeltaMsg)

Process a ShapeDelta from Origin.

pub async fn handle_clock_sync(&self, msg: &VectorClockSyncMsg)

Process a VectorClockSync from Origin.

pub async fn check_sequence_gap( &self, shape_id: &str, lsn: u64, ) -> Option<ResyncRequestMsg>

Check an incoming ShapeDelta for sequence gaps.

For each shape, we track the last LSN received. If the incoming LSN is not contiguous (gap > 1), this indicates missing deltas in the stream. Returns Some(ResyncRequestMsg) if a gap is detected, None otherwise.

pub async fn reset_sequence_tracking(&self)

Reset sequence tracking state on reconnect.

pub async fn set_pending_resync(&self, msg: ResyncRequestMsg)

Store a pending re-sync request (set by gap detection in receive loop).

pub async fn take_pending_resync(&self) -> Option<ResyncRequestMsg>

Take the pending re-sync request (consumed by delta push loop).

pub async fn set_pending_array_ack(&self, msg: ArrayAckMsg)

Merge an ArrayAck into the pending-ack map.

Per-array, only the ack with the highest HLC is kept — ack_hlc_bytes is stored in the same 18-byte big-endian layout used by nodedb_array::sync::Hlc, so byte-wise comparison gives the correct temporal ordering. This means no ack is silently lost: if two acks arrive for the same array before the push loop drains them, the one with the higher frontier is retained, which is exactly what Origin needs to advance its GC cursor.

pub async fn drain_pending_array_acks(&self) -> Vec<ArrayAckMsg>

Drain all pending ArrayAcks (consumed by the push loop).

Returns all per-array acks accumulated since the last drain. The map is cleared so new acks can accumulate for the next tick.

pub async fn reset_array_catchups(&self, requests: Vec<ArrayCatchupRequestMsg>)

Replace every pending ArrayCatchupRequest with requests.

Called once per session after the handshake: the new round supersedes any request the previous session did not send.

pub async fn queue_array_catchup(&self, request: ArrayCatchupRequestMsg)

Queue an ArrayCatchupRequest for the next push tick, replacing an unsent request for the same array.

pub async fn drain_array_catchups(&self) -> Vec<ArrayCatchupRequestMsg>

Drain every pending ArrayCatchupRequest in array-name order (consumed by the push loop).

impl SyncClient

pub async fn queue_row_push_reject(&self, msg: &RowPushMsg, error: &NodeDbError)

Queue the refusal of msg, which failed to apply with error.

pub async fn drain_row_push_rejects(&self) -> Vec<RowPushRejectMsg>

Take every queued refusal.

pub async fn requeue_row_push_rejects(&self, unsent: Vec<RowPushRejectMsg>)

Put back unsent, drained refusals a failed send did not deliver, ahead of any queued since. Refusals past the cap are logged and not sent.

impl SyncClient

pub fn new(config: SyncConfig) -> Self

Create a new sync client (does not connect yet).

The client holds no copy of the instance’s identity. lite_id, epoch, and the Loro peer id are read from the delegate at the moment each frame is built, because all three change while a client is alive — a peer-id rotation or a fork recovery replaces them — and a cached copy would keep stamping outbound frames with the identity Origin just refused.

pub fn with_flow_control( config: SyncConfig, flow_config: FlowControlConfig, ) -> Self

Create a new sync client with custom flow control config.

pub async fn state(&self) -> SyncState

Current connection state.

pub async fn set_state(&self, new_state: SyncState)

Set the connection state.

pub fn set_compensation_handler(&self, handler: Arc<dyn CompensationHandler>)

Register a compensation handler.

pub fn shapes(&self) -> &Arc<Mutex<ShapeManager>>

Access the shape manager (for subscribing/unsubscribing).

pub fn clock(&self) -> &Arc<Mutex<VectorClock>>

Access the vector clock.

pub fn compensation(&self) -> &Arc<CompensationRegistry>

Access the compensation registry.

pub fn config(&self) -> &SyncConfig

Access config.

pub fn flow(&self) -> &Arc<Mutex<FlowController>>

Access the flow controller.

pub fn metrics(&self) -> &Arc<SyncMetrics>

Access the sync metrics.

pub async fn producer_id(&self) -> u64

Producer ID assigned by Origin, or 0 if the handshake has not yet completed.

pub async fn accepted_epoch(&self) -> u64

Accepted epoch echoed by Origin, or 0 if the handshake has not yet completed.

pub async fn load_producer_state(&self, producer_id: u64, accepted_epoch: u64)

Load producer state (producer_id + accepted_epoch) from previously persisted values. Called on reconnect so the client knows its identity before the next handshake.

pub fn is_fenced(&self) -> bool

Whether Origin fenced this producer.

When true, the sync loop should disconnect and reconnect. The epoch is only bumped on db-open (via LiteIdentity), so reconnecting alone does not change the epoch. A fenced producer requires the operator to restart the process to mint a fresh epoch.

pub fn set_fenced(&self)

Mark this producer as fenced by Origin.

Also unsets the push_paused_for_auth flag so the disconnect path is not confused with an auth-pause: fencing is a permanent producer-epoch rejection, not a token issue.

pub fn clear_fenced(&self)

Clear the fenced flag. Called on reconnect so the client can attempt re-registration; if Origin still fences it the flag is set again.

pub async fn admit_row_push( &self, peer_id: u64, collection: &str, sequence: u64, ) -> bool

Returns true when this row push should be applied.

Every client → server engine path is protected by Origin’s sync_admit gate (producer/epoch/seq); there is no equivalent for server → client RowPush frames. Without this gate, a re-delivered frame is applied unconditionally: a duplicate delivery of a stale post-image can resurrect a row a later, un-replayed write already deleted, and an out-of-order pair can leave the older post-image as the final state.

sequence == 0 is the unsequenced sentinel — it mirrors how Origin treats producer_id == 0 as unidentified in sync_gate.rs, marking DDL-managed system rows (retention policies, alerts) that were never assigned a per-collection sequence. Such a frame carries no ordering information to dedup against, so it is always applied and never allowed to move the watermark (doing so would let a later legitimately-sequenced frame for the same collection be skipped as “stale”).

impl SyncClient

pub async fn should_refresh_token(&self) -> bool

Check if the JWT token needs proactive refresh (at 80% of lifetime).

Returns true if a refresh should be initiated. Called from the ping loop to piggyback on the keepalive timer.

pub async fn is_refresh_backoff_elapsed(&self) -> bool

Check if a token refresh attempt is currently allowed given the backoff.

Returns false if the minimum retry interval since the last attempt has not elapsed yet. Called by push_control_messages before invoking initiate_token_refresh.

pub async fn initiate_token_refresh(&self) -> Option<TokenRefreshMsg>

Initiate a token refresh via the token provider.

Marks a refresh as in-flight, calls the provider with a 30-second timeout, and returns the new token message on success. On failure (provider returns None or times out) the token_refresh_pending flag is cleared, push_paused_for_auth is kept true (push remains paused), and the backoff doubles for the next attempt.

pub async fn handle_token_refresh_ack(&self, ack: &TokenRefreshAckMsg)

Handle a TokenRefreshAck from Origin.

pub async fn pause_for_auth(&self)

Pause delta push due to auth failure. Called when Origin rejects with PermissionDenied, indicating the token has expired.

pub async fn is_push_paused_for_auth(&self) -> bool

Check if push is paused for auth.