Struct ActivityDispatcher 

pub struct ActivityDispatcher { /* private fields */ }

Push dispatcher backed by the connected-worker registry.

Implementations

impl ActivityDispatcher

pub fn new(registry: ConnectedWorkerRegistry) -> Self

Build a dispatcher over the shared worker registry.

pub fn with_drain_state(self, drain_state: DrainState) -> Self

Share the server drain gate.

pub async fn dispatch( &self, activity: &ScheduledActivity, ) -> Result<(), ServerError>

Push a scheduled activity to a matching worker.

Errors

Returns a typed dispatch error if no worker is available or the selected stream is closed; returns lock poison if registry access cannot be trusted.

pub async fn dispatch_preferring( &self, activity: &ScheduledActivity, preferred: &BTreeSet<String>, ) -> Result<(), ServerError>

Dispatch activity preferring workers on one of the preferred node labels, spilling to ANY live worker when none of the preferred labels has a live worker (Control-Plane Phase 2, P2-P3 — the Prefer{L} soft spill).

This is consulted ONLY for an UNPINNED activity (activity.node == None): a per-activity authored pin always wins and is dispatched through Self::dispatch unchanged. The recorded row’s node is NEVER mutated — preference is a pure dispatch-time worker-selection optimization in this non-replayed path, exactly like the existing round-robin, so replay is untouched (CP-Phase-2 §2.4).

The prefer-then-spill tier sequence is derived ONCE, from the shared preferred_node_order, so this gRPC path and the liminal RegistryLiminalDispatch can never diverge on what “prefer labelled worker, spill to any” means:

Tier 1..N: for each preferred label (deterministic set order) try a NON-WAITING workers_for(node = Some(label)) and dispatch to the first live worker found. Tier N+1 (spill): if no preferred label has a live worker, fall back to Self::dispatch with the activity’s own (unpinned) node, so the wait-for-worker backstop and round-robin behave exactly as today. An empty preferred set is the spill case immediately.

Errors

As Self::dispatch.

pub async fn dispatch_requiring( &self, activity: &ScheduledActivity, required: &BTreeSet<String>, ) -> Result<(), ServerError>

Dispatch activity REQUIRING a worker whose advertised node is one of the required labels, WAITING when none is live and NEVER spilling to a node=None any-worker dispatch (Control-Plane Phase 2, P2-I1 — the Pinned{L} hard pin). This is the opposite of Self::dispatch_preferring: a Prefer set appends a None spill tier; a Pinned set has NO None tier and instead holds on the wait-for-worker backstop until an L-labelled worker registers.

Consulted ONLY for an UNPINNED activity (activity.node == None): a per-activity authored pin always wins and dispatches through Self::dispatch unchanged. The recorded row’s node is NEVER mutated — the required set is a pure dispatch-time worker-selection input in this non-replayed path, so replay is untouched (CP-Phase-2 §2.4).

Each retry tries every required label (deterministic [BTreeSet] order) via a NON-WAITING workers_for(node = Some(label)) and delivers to the first live worker found, preserving the round-robin exactly like Self::dispatch_to_node. When no required label has a live worker across the whole set, it awaits wait_for_worker and retries — the same isolation-stall a per-activity Some(N) pin already exhibits. An EMPTY required set can never be satisfied by any labelled worker, so it stalls (isolation > availability); the caller sets a non-empty Pinned{L} for a live pin.

Errors

As Self::dispatch.

Trait Implementations

impl Clone for ActivityDispatcher

fn clone(&self) -> ActivityDispatcher

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Debug for ActivityDispatcher

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

Formats the value using the given formatter.