Struct ConnectedWorkerRegistry 

pub struct ConnectedWorkerRegistry { /* private fields */ }

Cloneable registry of currently connected worker streams.

Implementations

impl ConnectedWorkerRegistry

pub fn set_advertised_capacity( &self, worker_id: WorkerId, max_concurrency: u32, ) -> Result<bool, ServerError>

Record the capacity a worker announced after registering, and WAKE the selection waits.

The liminal counterpart of the gRPC RegisterWorker.max_concurrency field. That transport’s registration frame is a published wire type with no capacity field, so a liminal worker states its configured max_concurrency in the announcement it publishes on the reserved capabilities channel one frame later, and this applies it to the live handle selection reads.

Until this lands, the worker’s capacity is None and selection treats it as full — so this call is what makes a liminal worker dispatchable at all, which is why it wakes the parked selections the way a completion does. A worker parked on could otherwise sit until an unrelated registry change happened to fire.

Returns false when the worker is no longer registered (a disconnect racing the announcement — benign).

Errors

Returns ServerError::LockPoisoned if the registry lock is poisoned, or ServerError::Wire if the worker announced a capacity of zero — refused by name here exactly as the registration funnel refuses it, so the later channel cannot admit what the earlier one rejects.

pub fn in_flight_for_worker( &self, worker_id: WorkerId, ) -> Result<u32, ServerError>

How many dispatches worker_id is currently holding, as selection sees it.

Exposed so a test can hold this against the heartbeat tracker’s own per-worker count: the two are written by one call, and a divergence between them is the drift this projection would otherwise be able to hide.

Errors

Returns ServerError::LockPoisoned if the registry lock is poisoned.

impl ConnectedWorkerRegistry

pub fn ineligible_workers_over_tiers( &self, namespace: &str, task_queue: &str, activity_type: &str, tiers: &[Option<String>], ) -> Result<usize, ServerError>

How many workers that selection COULD have chosen are currently excluded from it by the liveness verdict (#197 R3).

tiers is the ordered sequence of node filters the selection actually walked, and taking it as an argument — rather than re-deriving one here — is the whole point of the method. Selection over a Pinned{L} namespace walks the required labels and NEVER spills to a None any-node tier; a census taken over the row’s own node instead would pass None, match every worker in the pool, and report an unlabelled worker that was never a candidate as the reason the row found nobody. The refusal built on that count then tells an operator not to start the labelled worker that is the only remedy.

Counted as a UNION over the tiers and over DISTINCT workers: the pool is walked once and a worker admissible to more than one tier is counted once, so the number is a headcount rather than a sum of overlapping matches.

Read under the SAME lock and with the SAME worker_matches_node filter Self::select_and_reserve applies, so a refusal quoting this count describes the fleet selection actually saw.

Errors

Returns ServerError::LockPoisoned if the registry lock is poisoned.

pub fn pool_census( &self, namespace: &str, task_queue: &str, activity_type: &str, node: Option<&str>, ) -> Result<PoolCensus, ServerError>

Census the live fleet for one dispatch address (R1).

Taken under the SAME lock discipline as Self::select_and_reserve and read immediately after a selection miss, so the taxonomy verdict describes the fleet selection actually saw. Three nested counts — the pool, the activity coverage within it, the node coverage within that — are what separate NO_LIVE_POLLERS from POLLERS_INCOMPATIBLE.

Two further counts split the compatible workers by dispatch eligibility, which is what separates a pool that is genuinely served from one whose workers are all excluded, and — within that — an exclusion that clears itself from one that does not (POLLERS_UNREACHABLE). They are taken here, under this same lock, rather than by a second read: eligibility can change between two acquisitions, and a verdict assembled from two readings would describe a fleet that never existed at one instant.

last_compatible_poller_age is zero while a compatible worker is connected, the elapsed time since the most recent compatible departure when one has left, and None when this server has never had one.

Errors

Returns ServerError::LockPoisoned if the registry lock is poisoned.

impl ConnectedWorkerRegistry

pub fn select_and_reserve( &self, namespace: &str, task_queue: &str, activity_type: &str, node: Option<&str>, ) -> Result<Option<(WorkerHandle, DispatchReservation)>, ServerError>

Select a worker for a dispatch AND hold one of its capacity slots, both under one acquisition of the registry lock.

Why selection alone was not enough

Selection filters out a worker already at its advertised concurrency, but the increment that makes a worker reach that concurrency used to happen much later — at HeartbeatTracker::track_task, once the pending entry, the superseded-attempt release and the lease record were all done. Everything between was a window in which the projection still read zero. A fan of N simultaneous dispatches onto one worker therefore had all N threads read the same pre-dispatch count, all N pass the capacity filter, and all N push — the filter was real but it was reading a number nobody had claimed yet. The worker then refused the surplus, and a refusal resolves in the TRANSPORT domain, whose ledger is deliberately bounded: routine over-dispatch would spend that budget on a worker that was merely busy and dead-letter the surplus. The saturation the fix exists to survive would have produced the outage the fix exists to prevent.

Claiming the slot here closes the window by construction: the choice and the count that justifies the next choice are one critical section, so the second caller of a full worker sees a full worker.

The returned DispatchReservation releases the slot when it is dropped, which is what makes every early exit between here and tracking safe without any of them knowing about capacity. Hold it across track_task — which takes the durable count over — and drop it after; the overlap is one extra held slot for the width of that call, which can only park a dispatch that would otherwise have raced, never admit one.

Errors

Returns ServerError::LockPoisoned if the registry lock is poisoned.

impl ConnectedWorkerRegistry

pub fn with_metrics(metrics: Metrics) -> Self

Build a registry that records connected-worker gauge updates.

pub fn with_capacity_wake(self, wake: Arc<Notify>) -> Self

Share the dispatcher wake this registry pulses when a capacity slot is FREED.

The busy answer a full pool gives is attempt-neutral, so a parked row is waiting for capacity — and capacity-freeing is an event this server produces, not a deadline it should guess at. Without this the only thing re-offering a busy-parked row was its durable visible_after, which rides the operator’s FAILURE backoff: raising that legitimately (to slow genuine retries) parked healthy work behind it for the same duration.

Pulsed by record_dispatch_finished, which every slot-free goes through — a completed dispatch’s untrack and a released reservation alike.

pub fn admission_audit(&self) -> &AdmissionAudit

The admission audit every registration transport names its refusals through. Clones of a registry share one, so a worker refused over gRPC and then over liminal is one site, not two.

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

Attach the WS3 cluster-event publisher so worker topology changes are pushed to the dashboard. Pure builder addition.

pub fn with_worker_deployment_store( self, store: Arc<dyn WorkerDeploymentStore>, ) -> Self

Attach the durable worker-deployment store used for association lookup.

pub fn with_namespace_minting( self, store: Arc<dyn NamespaceStore>, policy: AutoCreate, ) -> Self

Install the minted-on-use namespace hook (Control-Plane Phase 1).

After a registration is authorized and its namespace set scoped, each authorized namespace is durably recorded (AutoCreate::Open) or gated (AutoCreate::Closed) through store. Without this builder the registry never touches the namespace registry, so registration stays byte-identical to before the registry existed. Pure builder addition, mirroring Self::with_cluster_publisher.

When a cluster publisher has already been attached (Self::with_cluster_publisher, called first on the boot path), it is threaded into the minter so a first worker-mint emits the live namespace created delta to the ops console (S8). Order-independence is not assumed: callers wire the publisher before minting on the boot path.

pub fn with_namespace_routing(self, routing: NamespaceRouting) -> Self

Thread the boot’s namespace-mint routing context into the registry’s minter, so a worker registering for a namespace whose registry shard this node does not own mints through the shard’s owner instead of being refused NotOwner forever.

The SECOND of the two minter construction sites (the first is ServerState::namespace_minter, which serves the gRPC and HTTP start seams). Called after Self::with_namespace_minting on the boot path; a no-op when no minter is installed, and never called at all off-cluster, so default/test registries stay byte-identical.

pub async fn accept_registration( &self, guard: &NamespaceGuard, caller: &CallerIdentity, registration: &ProtoRegisterWorker, sender: WorkerTaskSender, ) -> Result<WorkerRegistration, ServerError>

Authorize a worker registration and insert it into the connected-worker registry.

Errors

Returns ServerError if namespace authorization fails or the registry lock is poisoned.

pub async fn admit_delivery( &self, guard: &NamespaceGuard, caller: &CallerIdentity, registration: &ProtoRegisterWorker, delivery: WorkerDelivery, intervention_capabilities: InterventionCapabilities, ) -> Result<WorkerRegistration, ServerError>

The ONE admission every transport’s registration passes through, in this order: the guard’s worker-registration policy, then EACH namespace in the worker’s set authorized against the caller, then the auth-scoped mint-or-gate, then the placement-admission gate against the worker’s advertised node, then the insert. The delivery leg is the only thing a transport contributes.

It exists because the liminal transport used to insert into the registry directly — register_delivery from its connection callback — so a worker dialling the liminal listener was registered into any namespace it named with none of the gates above, while the identical worker over gRPC was refused. Two entry points, one set of rules: the rules live here, and a transport that bypasses them has to be written to do so on purpose.

Errors

Returns the guard’s namespace denial, the mint-or-gate refusal, the placement refusal, or the registry’s own insert error, exactly as the gRPC path always has.

pub async fn resolve_instance_identity( &self, instance: Option<&ProtoWorkerInstanceIdentity>, ) -> Result<Option<WorkerInstanceIdentity>, ServerError>

Resolve the wire’s optional deployment/instance identity into the registry’s own, marking whether the named deployment is known to the deployment store (or unchecked when no store is configured).

Errors

Returns the deployment store’s read error.

pub fn register<'a>( &self, namespace: impl Into<String>, activity_types: impl IntoIterator<Item = &'a String>, sender: WorkerTaskSender, max_concurrency: u32, ) -> Result<WorkerRegistration, ServerError>

Insert an already-authorized worker stream into the default task queue of a single namespace, with no node affinity.

Convenience over Self::register_namespaces for callers that serve one namespace and do not select a task queue (notably tests of the default pool).

Errors

Returns ServerError::LockPoisoned if the registry lock is poisoned.

pub fn register_pool<'a>( &self, pool: PoolAddress, activity_types: impl IntoIterator<Item = &'a String>, sender: WorkerTaskSender, max_concurrency: u32, ) -> Result<WorkerRegistration, ServerError>

Insert an already-authorized worker stream into one explicit worker pool (single namespace + task queue), with no node affinity.

Errors

Returns ServerError::LockPoisoned if the registry lock is poisoned.

pub fn register_namespaces<'a>( &self, namespaces: impl IntoIterator<Item = String>, task_queue: impl Into<String>, node: Option<String>, activity_types: impl IntoIterator<Item = &'a String>, sender: WorkerTaskSender, max_concurrency: u32, ) -> Result<WorkerRegistration, ServerError>

Insert an already-authorized worker stream serving a SET of namespaces under one task_queue, with an optional node locality affinity.

The worker is indexed under one (namespace, task_queue, activity_type) key per namespace in its set, so a dispatch in any of those namespaces can reach it. node is recorded on the handle and used only as a within-pool filter at selection time — it is NOT part of PoolAddress.

Errors

Returns ServerError::LockPoisoned if the registry lock is poisoned.

pub fn register_delivery<'a>( &self, namespaces: impl IntoIterator<Item = String>, task_queue: impl Into<String>, node: Option<String>, activity_types: impl IntoIterator<Item = &'a String>, delivery: WorkerDelivery, options: RegistrationOptions, ) -> Result<WorkerRegistration, ServerError>

Insert an already-authorized worker serving a SET of namespaces under one task_queue and optional node, delivered to through an explicit WorkerDelivery transport, carrying the RegistrationOptions it registered with.

This is the one transport-agnostic registration: Self::register_namespaces is the gRPC façade over it (it wraps the stream sender in WorkerDelivery::Grpc). Selection (select_worker/workers_for) is identical across transports; only the held delivery differs.

Errors

Returns ServerError::LockPoisoned if the registry lock is poisoned.

pub fn worker_arrival(&self) -> WorkerArrival

Subscribe to the next change in what dispatch selection can see: a new worker registers, or a reachability verdict is published (Self::set_dispatch_ineligible). Both are ways a pool that had no selectable worker gains one, and a wait that only woke on the first would sleep through a pool whose workers are all excluded — those workers are already registered, so no registration is coming for them.

Take the subscription BEFORE you read the registry, and await it only after the read has missed. That ordering is the whole contract; see WorkerArrival for why a subscription taken after the read loses the arrival that landed during it.

Callers must re-check the registry after waking: the newly arrived worker may not serve the namespace or activity type the caller needs, and a republished verdict may have restored nobody.

The returned value carries its own #[must_use] message: a subscription constructed and dropped is a subscription thrown away.

pub fn workers_for( &self, namespace: &str, task_queue: &str, activity_type: &str, node: Option<&str>, ) -> Result<Vec<WorkerHandle>, ServerError>

Return a snapshot of the DISPATCH-ELIGIBLE workers registered for the (namespace, task_queue, activity_type) pool, ordered by worker id and then rotated so each call starts from the next worker in the pool. The rotation cursor is per triple, so each pool round-robins independently.

When node is Some, the result is filtered to workers whose advertised node equals it — a dispatch pinned to a node reaches only workers on that node (NODE affinity = require). When node is None, the behaviour is exactly the unpinned pool: every worker in the (namespace, task_queue) pool is a candidate regardless of locality. node is a within-pool filter, NOT part of the pool key, so the per-triple rotation cursor is shared across pinned and unpinned lookups of the same pool.

Candidates, ordering, eligibility and the cursor all come from [eligible_candidates_in_rotation], which Self::select_and_reserve reads too: ONE derivation and ONE cursor, so the gRPC push dispatcher and every other dispatch path rotate over the same workers in the same order and cannot drift about who is dispatchable or whose turn it is.

Errors

Returns ServerError::LockPoisoned if the registry lock is poisoned.

pub fn all_workers(&self) -> Result<Vec<WorkerHandle>, ServerError>

Return a snapshot of every connected worker stream.

Errors

Returns ServerError::LockPoisoned if the registry lock is poisoned.

pub fn worker_by_id( &self, worker_id: WorkerId, ) -> Result<Option<WorkerHandle>, ServerError>

Return the handle for a worker by id, or None when it is not registered.

The intervention router resolves the owning worker of a target attempt by id (NOI-6): the attempt-owner back-index stores a WorkerId, and the router reads back the live handle to gate on its advertised capabilities and select its delivery. A None result means the owner disconnected — the router treats that as the attempt-scoped no-op.

Errors

Returns ServerError::LockPoisoned if the registry lock is poisoned.

pub fn set_intervention_capabilities( &self, worker_id: WorkerId, capabilities: &InterventionCapabilities, ) -> Result<bool, ServerError>

Replace the advertised intervention capabilities of a registered worker (NOI-6). The liminal registration frame cannot carry capabilities, so a liminal agent worker announces them on the reserved capabilities channel right after registering, and this applies the announcement to the live handle the intervention router gates on. Returns false when the worker is no longer registered (a disconnect racing the announcement — benign).

Errors

Returns ServerError::LockPoisoned if the registry lock is poisoned.

pub fn broadcast_drain(&self) -> Result<usize, ServerError>

Broadcast a graceful drain request to every connected worker stream. Workers are removed from routing before any transport signal is attempted. A liminal worker has no drain control frame, so it is force-fenced and deregistered with an error-level identity-bearing log instead of being silently skipped.

Errors

Returns ServerError::LockPoisoned if the registry lock is poisoned.

pub fn drain_worker(&self, worker_id: WorkerId) -> Result<bool, ServerError>

Stop assigning work to one worker and request a graceful transport drain.

Returns false if the worker is not registered or if its transport has no drain channel. In the latter case the worker is deregistered immediately, after an error-level log names the worker and its transport limitation.

Errors

Returns ServerError::LockPoisoned if the registry lock is poisoned.

pub fn set_dispatch_ineligible( &self, unreachable: BTreeMap<WorkerId, DispatchExclusion>, ) -> Result<(), ServerError>

Publish the liveness probe’s reachability verdict: the set of workers the server cannot currently reach on the push leg, which [eligible_candidates_in_rotation] then skips for BOTH selectors.

Replaces the whole set rather than toggling one worker, so the published verdict is always exactly one round’s evidence and a worker can never be left excluded by a stale entry nobody cleared.

Publishing WAKES the selection wait (WorkerArrival). A dispatch that found no eligible worker is blocked on this verdict every bit as squarely as on a registration, and the workers it needs are already registered — so a park that only woke on registrations would sleep through their recovery. A publication landing between a dispatch’s census and its park is retained, because the dispatch holds a WorkerArrival taken before the census; this method owes nothing to that ordering beyond firing.

The wake is unconditional rather than gated on the set having shrunk, because a change-gated wake would make correctness depend on this method judging what “changed” means for a caller it cannot see — a set that shrank for a worker in some other pool is no restoration for THIS dispatch, and a set republished identically may still coincide with the registration that serves it. Every round publishes, so waking on each one is self-healing at the probe’s own cadence — it invents no clock of its own — and a publication with nobody parked costs one waiterless notify_waiters.

Errors

Returns ServerError::LockPoisoned if the registry lock is poisoned.

pub fn transports_of( &self, workers: impl IntoIterator<Item = WorkerId>, ) -> Result<BTreeMap<WorkerId, WorkerTransport>, ServerError>

The transport each named worker is delivered over, for the workers still registered (#25).

A worker absent from the result has LEFT the registry — the caller must treat that as “no transport”, never as a default one. That distinction is the whole reason this returns a map rather than a vector in the caller’s order: the liveness probe scopes its verdict by transport coverage, and a departed worker whose transport was guessed would be judged by a probe that never had a wire to it.

Read under ONE lock acquisition rather than one per worker, so the answer describes a single registry state instead of a smear across a round of registrations and disconnects.

Errors

Returns ServerError::LockPoisoned if the registry lock is poisoned.

pub fn is_dispatch_ineligible( &self, worker_id: WorkerId, ) -> Result<bool, ServerError>

Whether a worker is currently excluded from dispatch selection for unreachability.

Errors

Returns ServerError::LockPoisoned if the registry lock is poisoned.

pub fn grpc_liveness_targets( &self, ) -> Result<Vec<GrpcLivenessTarget>, ServerError>

Every gRPC-delivered worker the liveness probe must ping this round, paired with the stream sender the ping rides (#197).

The liminal half of the same census is LiminalConnectionNotifier::liveness_targets, which enumerates CONNECTIONS. This one enumerates REGISTRATIONS, because a gRPC worker’s only server-side identity is its registry handle: the stream is owned by a tonic task and reachable solely through the sender the registration carries.

Deliberately not filtered by current eligibility. A worker excluded from dispatch is precisely the worker whose next answered ping restores it, so skipping the excluded set would make exclusion permanent — the exact defect this lane exists to remove.

Errors

Returns ServerError::LockPoisoned if the registry lock is poisoned.

pub fn dispatch_ineligible( &self, ) -> Result<BTreeMap<WorkerId, DispatchExclusion>, ServerError>

The currently published exclusion set, read as a whole.

The liveness probe reads this BEFORE publishing a round’s verdict, so it can announce the workers that just LEFT the set. Without the whole previous set there is no way to name a recovery: per-worker queries can only be asked about workers the new verdict already mentions, and a recovered worker is precisely the one it does not.

Errors

Returns ServerError::LockPoisoned if the registry lock is poisoned.

pub fn is_registered(&self, worker_id: WorkerId) -> Result<bool, ServerError>

Return whether a worker stream is currently registered.

The activity dispatch path uses this after queuing a task to detect a worker whose stream tore down concurrently: a sweep that ran before the dispatch tracked its task can never complete it, so the dispatch must fail the activity itself instead of waiting forever.

Errors

Returns ServerError::LockPoisoned if the registry lock is poisoned.

pub fn deregister(&self, worker_id: WorkerId) -> Result<(), ServerError>

Remove a worker by id from every namespace/activity index it advertised.

Emits a WS3 WorkerDeathReason::Disconnect delta — the truthful default for a removed worker whose stream/registration went away. Callers that can PROVE a finer reason (a liveness-timeout sweep) call Self::deregister_with_reason instead, so the dashboard never sees a fabricated distinction.

Errors

Returns ServerError::LockPoisoned if the registry lock is poisoned.

pub fn deregister_with_reason( &self, worker_id: WorkerId, reason: WorkerDeathReason, ) -> Result<(), ServerError>

Remove a worker by id, attributing the departure to an explicit WorkerDeathReason the caller can prove at its call site (for example a heartbeat sweep passes WorkerDeathReason::Timeout).

Errors

Returns ServerError::LockPoisoned if the registry lock is poisoned.

Trait Implementations

impl Clone for ConnectedWorkerRegistry

fn clone(&self) -> ConnectedWorkerRegistry

Returns a duplicate of the value.

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

Performs copy-assignment from source.

impl Debug for ConnectedWorkerRegistry

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

Formats the value using the given formatter.

impl Default for ConnectedWorkerRegistry

fn default() -> Self

Returns the “default value” for a type.