Struct ShuffleReceiver 

pub struct ShuffleReceiver { /* private fields */ }

Inbound side of the shuffle fabric: a Tonic ShuffleTransport server that surfaces every received frame, attributed to its peer, on the bounded queue.

Implementations

impl ShuffleReceiver

pub async fn bind( local_id: ShufflePeerId, addr: SocketAddr, receiver_incarnation: Uuid, ) -> Result<Self>

Bind on addr and start serving; the resolved address is at Self::local_addr.

Errors

Returns io::Error on bind failure.

pub fn install_process_lease_deadline( &self, deadline: Arc<LeaseDeadline>, ) -> Result<()>

Bind inbound admission to this process’s renewable cluster lease.

This must be installed before an assignment certificate is activated.

Errors

Returns an error for an expired lease, a different previously installed deadline, or an already-active assignment.

pub fn install_assignment_fence( &self, fence: &CheckpointAssignmentFence, owners: &[ShufflePeerId], ) -> Result<bool>

Install the exact assignment certificate accepted by inbound streams. Existing streams are rejected on their next frame; the new delivery domain starts at zero.

Errors

Returns an error for a malformed certificate, a same-version certificate conflict, or a certificate that does not bind this exact receiver process.

pub fn suspend_assignment_fence(&self)

Temporarily close inbound admission after a transient durable-authority read failure. Pending handshakes are discarded, while committed delivery expectations remain bound to the retained certificate for exact same-version reactivation.

pub fn invalidate_assignment_fence(&self)

Reject all streams while an adopted owner map is awaiting its owner-complete process certificate. Pending tokens and sequence expectations cannot cross this boundary.

pub fn assignment_version(&self) -> u64

Assignment scope currently accepted for inbound streams and staged frames.

pub fn active_assignment_digest(&self) -> Option<[u8; 32]>

Digest of the exact assignment certificate currently active for inbound streams.

pub fn set_recovery_gen(&self, gen: u64)

Advance the inbound recovery scope. Existing streams are rejected; the next exact handshake re-baselines sequence after the coordinated rewind.

pub fn recovery_gen(&self) -> u64

Recovery scope currently accepted for inbound streams and staged frames.

pub const fn incarnation(&self) -> Uuid

Process incarnation bound into every accepted stream handshake.

pub const fn local_id(&self) -> ShufflePeerId

Node id bound into every accepted stream handshake.

pub fn delivery_loss_incidents(&self) -> Arc<AtomicU64>

Cumulative delivery-loss incidents. A value above the recovered floor means an epoch must not seal; exact missing-frame magnitude is diagnostic and never trusted for this non-wrapping correctness signal.

pub fn recovered_delivery_loss_incidents(&self) -> Arc<AtomicU64>

Delivery-loss incidents covered by a completed coordinated rewind. The callback uses this durable in-process floor so repaired incidents do not fault restored state again.

pub fn has_unrecovered_delivery_loss(&self) -> bool

Whether an admitted delivery loss has not yet been repaired by coordinated recovery.

pub fn complete_recovery(&self, gen: u64) -> bool

Mark the loss cutoff captured when gen began as repaired. Loss detected after the generation advanced remains above this floor and still faults the next checkpoint.

Returns false when gen is not the currently prepared recovery generation.

pub async fn bind_with_kv( local_id: ShufflePeerId, addr: SocketAddr, kv: Arc<dyn ClusterKv>, receiver_incarnation: Uuid, ) -> Result<Self>

Bind and publish the listener’s address into kv for peer discovery.

Errors

Returns io::Error on bind failure.

pub fn local_addr(&self) -> SocketAddr

Local socket address the server is bound to.

pub async fn recv(&self) -> Option<ReceivedShuffle>

Await the next (peer_id, msg); None once the server stops and the queue drains. Concurrent callers serialise via rx_returned; cancellation-safe.

pub fn drain_available(&self) -> Vec<ReceivedShuffle>

Drain every currently available admitted message without blocking; empty if a recv() holds the receiver.

pub fn drain_checkpointed_data_for(&self, stage: &str) -> Vec<ReceivedBatch>

Non-blocking drain of checkpointed operator batches for stage.

pub fn drain_staged_barriers(&self) -> Vec<ReceivedShuffle>

Take the barriers stashed by Self::drain_checkpointed_data_for.

pub fn has_staged_checkpoint_barriers(&self) -> bool

Whether an in-band barrier currently blocks normal holdover draining.

pub fn stage_checkpointed_inbound(&self) -> bool

Move available inbound frames into the bounded holdover through the first barrier. Data remains keyed for its owning stage.

pub fn retire_checkpoint_barriers( &self, attempt: CheckpointAttempt, assignment_digest: [u8; 32], ) -> Result<()>

Retire barrier markers through a certified terminal checkpoint without dropping data.

Errors

Rejects a noncanonical attempt and exact markers whose assignment digest differs from the durable terminal outcome.

pub fn stash_barrier(&self, barrier: ReceivedShuffle)

Re-stash a peer barrier pulled while aligning a different checkpoint, so a lagging node still sees it when it reaches that checkpoint.

pub fn drain_checkpointed_staged(&self) -> Vec<(String, ReceivedBatch)>

Drain every staged checkpointed operator batch.

pub fn drain_checkpointed_holdover( &self, ) -> Result<Vec<(String, ReceivedBatch)>>

Drain only the checkpointed batches already in the holdover. Unlike Self::drain_checkpointed_staged, this does not consume the live receive queue, whose data/barrier order must remain visible to checkpoint alignment.

Errors

Returns a typed cancellation while assignment consumption is suspended, or a process lease error after this receiver loses execution authority. An error observed before extraction leaves the holdover untouched; the caller revalidates authority immediately after a successful transfer.

pub fn drain_all_staged(&self) -> Vec<(String, ReceivedBatch)>

Empty the per-stage holdover, returning every buffered (stage, batch).

Trait Implementations

impl Debug for ShuffleReceiver

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl Drop for ShuffleReceiver

fn drop(&mut self)

Executes the destructor for this type.

fn pin_drop(self: Pin<&mut Self>)

🔬This is a nightly-only experimental API. (pin_ergonomics)

Execute the destructor for this type, but different to Drop::drop, it requires self to be pinned.