Struct ShuffleSender 

pub struct ShuffleSender { /* private fields */ }

Lazy pool of ordered streams per peer.

Implementations

impl ShuffleSender

pub fn with_send_stall_timeout(self, timeout: Option<Duration>) -> Self

Fail a data or frontier send when one of its waits lasts timeout: the wait for the peer’s or the node’s outbound byte budget, or for a slot in the peer’s stream queue. A wait ends once the streams have drained room for one frame, so a wait that long means they stopped draining. is_shuffle_send_stalled recognises the error. Checkpoint barriers are bounded by their checkpoint deadline instead. Unset, a send waits for as long as its peer stays slow.

pub fn peer_flow(&self) -> Vec<ShufflePeerFlow>

Per-peer totals, ascending by peer.

pub fn outbound_budget_used_bytes(&self) -> usize

Bytes of the node’s outbound data budget held by frames being encoded or queued, including the share already granted to a send still waiting for the rest. The separate checkpoint-barrier budget is excluded.

impl ShuffleSender

pub fn install_pipeline_identity_pair( &self, receiver: &ShuffleReceiver, identity: &PipelineIdentity, ) -> Result<bool>

Bind both directions of this process’s fabric to identity. A receiver refuses the handshake of a sender bound to a different identity or to none, so the first exchange fails before any frame crosses. Changing the identity cancels the streams and handshakes opened under the previous one; sequence domains stay scoped to the assignment.

Errors

Rejects a non-canonical identity and a sender and receiver of different processes.

impl ShuffleSender

pub fn topology_fence(&self) -> Option<ShuffleTopologyFence>

Logical generation retained by this endpoint, including while assignment intake is closed.

pub fn topology_version(&self) -> u64

Cheap graph guard. Zero denotes the legacy fabric, never an inferred topology.

pub fn install_topology_fence_pair( &self, receiver: &ShuffleReceiver, expected: Option<ShuffleTopologyFence>, target: ShuffleTopologyFence, ) -> Result<bool>

Fence both directions of one process’s fabric to an exact committed catalog generation. The trusted caller must hold intake/execution and prove predecessor retirement and Commit. This supplies no actor readiness, output permit or cluster Release.

Existing assignment locks cancel prior streams, handshakes and blocked sends before starting a new sequence domain. Identical retries preserve sequences. Process/assignment/recovery identities and unrepaired delivery-loss fences are preserved.

Errors

Rejects stale/conflicting topology, incompatible endpoints, inactive assignments, expired process leases and unrepaired delivery loss before changing either direction.

pub fn install_topology_fence_pair_for_recovery( &self, receiver: &ShuffleReceiver, expected: Option<ShuffleTopologyFence>, target: ShuffleTopologyFence, generation: u64, ) -> Result<bool>

Install held transport for an exact authorized coordinated recovery Start. The trusted caller must prove the stopped roster, selected cut and current generation. Loss covered by that generation’s prepared cutoff may remain pending during installation; this does not forgive it or authorize input/output. Only recovery completion promotes the repair floor. Later loss and an exhausted incident counter still reject installation.

Errors

Rejects a zero/stale generation or any ordinary process, assignment and topology mismatch.

pub async fn send_to_for_topology( &self, peer: ShufflePeerId, expected_assignment_version: u64, topology: Option<ShuffleTopologyFence>, message: &ShuffleMessage, ) -> Result<()>

Send under an immutable graph topology binding and exact routing assignment. Existing unbound send methods target only the legacy fabric.

Errors

Rejects stale topology before sequence allocation, as well as ordinary send failures.

pub async fn send_to_for_generation( &self, peer: ShufflePeerId, assignment_version: u64, recovery_generation: u64, topology: Option<ShuffleTopologyFence>, message: &ShuffleMessage, ) -> Result<()>

Send under the exact routing assignment, recovery generation, and graph topology. A retained pre-recovery plan cannot acquire a stream in a newer recovery generation.

Errors

Rejects a stale generation before connection or sequence allocation, plus ordinary send errors.

impl ShuffleSender

pub fn new(local_id: ShufflePeerId, incarnation: Uuid) -> Self

Empty sender; peers arrive via Self::register_peer or KV discovery.

Panics

Panics when the node ID is zero or the process incarnation is nil.

pub const fn local_id(&self) -> ShufflePeerId

Node id bound into every outbound stream handshake.

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

Bind outbound 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 bind_process_lease_deadline_pair( &self, receiver: &ShuffleReceiver, deadline: Arc<LeaseDeadline>, ) -> Result<()>

Bind both directions of one local shuffle fabric to the same process lease.

Validation and installation are serialized across both handles, so an incompatible receiver cannot leave only the sender bound.

Errors

Returns an error without changing either handle when the deadline is expired, either handle is already bound to a different deadline, or an active assignment prevents a new binding.

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

Install the exact assignment certificate accepted by outbound shuffle streams. Changing scope closes all pooled streams and resets their scoped sequences.

Errors

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

pub fn suspend_assignment_fence(&self)

Temporarily close outbound admission because the durable assignment authority could not be read. Unlike invalidation, this preserves the exact certificate and sequence counters so the same durable head can resume without manufacturing a delivery gap.

pub fn invalidate_assignment_fence(&self)

Reject every stream while a newer assignment is being adopted but has not yet earned an owner-complete process certificate. The last certificate is retained only as a monotonic conflict floor; it is inactive while Self::assignment_version is zero.

pub fn assignment_version(&self) -> u64

Assignment scope currently accepted for newly enqueued frames.

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

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

pub fn set_recovery_gen(&self, gen: u64)

Advance the generation stamped onto outbound data frames. Called after a coordinated rewind so peers can discard anything this node produced before it.

pub fn recovery_gen(&self) -> u64

Recovery scope currently accepted for newly enqueued frames.

pub const fn incarnation(&self) -> Uuid

Process incarnation bound into every outbound stream handshake.

pub fn with_kv( local_id: ShufflePeerId, kv: Arc<dyn ClusterKv>, incarnation: Uuid, ) -> Self

Sender that falls back to kv discovery for peers not previously registered.

pub fn register_peer(&self, peer: ShufflePeerId, addr: SocketAddr)

Register (or update) a peer’s shuffle address.

pub async fn send_to( &self, peer: ShufflePeerId, msg: &ShuffleMessage, ) -> Result<()>

Send msg to peer, opening a client-streaming call if necessary.

Errors

Returns io::Error when the peer is unregistered/undiscoverable, the endpoint cannot be built, or the per-peer stream has shut down.

pub async fn send_to_for_assignment( &self, peer: ShufflePeerId, expected_assignment_version: u64, msg: &ShuffleMessage, ) -> Result<()>

Send only while the sender remains in expected_assignment_version.

Routing paths use this to prevent data sliced with an old ownership publication from entering a stream opened under a newer assignment scope.

Errors

Returns io::Error when the expected assignment is no longer current, in addition to the errors returned by Self::send_to.

pub async fn establish_assignment_mesh( &self, assignment_fence: &CheckpointAssignmentFence, ) -> Result<()>

Establish an exact-scope stream to every remote assignment participant without sending a frame or consuming a delivery sequence.

Errors

Returns the first peer connection or handshake error after attempting the full roster.

pub async fn fan_out_barrier( &self, peers: &[ShufflePeerId], barrier: CheckpointBarrier, assignment_fence: &CheckpointAssignmentFence, ) -> Result<()>

Ship barrier to every required peer. All peers are attempted, but any failure rejects the cut so the coordinator can abort promptly.

Errors

Returns the first peer error after attempting the full fan-out.

pub async fn fan_out_barrier_for_topology( &self, peers: &[ShufflePeerId], barrier: CheckpointBarrier, assignment_fence: &CheckpointAssignmentFence, topology: Option<ShuffleTopologyFence>, ) -> Result<()>

Fan out a barrier under an immutable graph topology binding.

Errors

Rejects a different topology or any ordinary barrier/assignment/peer failure.

Trait Implementations

impl Debug for ShuffleSender

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

Formats the value using the given formatter.