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_heartbeat_tracker(self, heartbeat_tracker: HeartbeatTracker) -> Self

Share the deployment’s liveness tracker, so a dispatch this path places is counted against the worker’s advertised capacity exactly as a bridge-seam dispatch is.

Required for the capacity contract to hold on this path: the tracker’s in-flight map IS the count selection reads.

pub fn with_lease_recorder(self, lease_recorder: LeaseRecorderSeam) -> Self

Share the lease-record seam, so an accepted delivery on this path is attributed through the same recorder (and counted on the same ledger) as one on the engine-seam bridge.

pub fn with_delivery_gate(self, delivery_gate: DeliveryGate) -> Self

Share the deployment’s delivery gate, so a dispatch waiting on a blocking transport abandons promptly when the server begins draining.

pub fn with_liminal_delivery( self, liminal_delivery: Arc<dyn WorkerTaskDelivery>, ) -> Self

Install the liminal delivery arm, so a liminal-registered worker selected by this dispatcher is SERVED over its own transport rather than deregistered for lacking a gRPC sender (#52).

pub fn with_cluster_publisher( self, cluster_publisher: ClusterEventPublisher, ) -> Self

Share the deployment-global cluster-event publisher so a dispatch parked with no availability deadline on this leg is announced on the operator’s real-time channel, not only in the log (#266 T4).

pub fn with_queue_service( self, declarations: QueueDeclarationSource, state: QueueServiceState, config: QueueServiceConfig, ) -> Self

Share the queue-service seams so a park on this path reaches the same GET /queues/unserved and describe surfaces the direct path feeds.

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

Share the server drain gate.

pub fn with_completion_fences(self, completion_fences: CompletionFences) -> Self

Share the completion-generation registry used by result ingestion.

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. There is now only one walk to derive it for: since #52 R4 this dispatcher selects for BOTH transports and each chosen worker is served over the one it registered on, so “prefer labelled worker, spill to any” has a single meaning by construction rather than by two implementations agreeing:

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 the WorkerArrival it subscribed to BEFORE walking the set 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, formatter: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.