Struct WorkerSupervisor 

pub struct WorkerSupervisor { /* private fields */ }

Server-owned supervision of the managed worker fleet.

Implementations

impl WorkerSupervisor

pub fn new( store: Arc<dyn WorkerDeploymentStore>, publisher: ClusterEventPublisher, ) -> Self

Build an UNCOMMISSIONED supervisor over the durable deployment store, publishing desired-state changes onto the deployment-global cluster channel.

The publisher lives HERE — not in the transports — so every desired-state write the supervisor makes (Self::start, Self::stop, and Self::restart through them) reaches the console’s live feed identically whichever transport asked, exactly as the worker-deployment desired-state endpoint publishes its own writes.

Nothing is supervised until Self::commission installs an operator policy: construction is not commissioning, so a server that boots without the config section never spawns anything.

pub fn record_auto_provision(&self, outcomes: &[AutoWorkerOutcome])

Record what auto-provision decided about each task queue, replacing any earlier decision for the same queue.

A failure to take the lock is logged rather than propagated: losing the REPORT of a provisioning decision must not turn a successful deploy into a refusal, and the decision itself has already been logged and returned to its caller by the time this is called.

pub fn commission( &self, policy: SupervisionPolicy, executable: ManagedExecutable, ) -> bool

Install the operator’s supervision policy and the executable a builtin deployment means on this server.

Returns false when a commission was already installed, in which case the existing one stands: the restart discipline of a running fleet is not something a later caller gets to swap out underneath it.

pub fn policy(&self) -> Option<SupervisionPolicy>

The installed policy, if this server has one.

pub fn is_commissioned(&self) -> bool

Whether an operator commission is installed.

pub async fn start( &self, name: &str, ) -> Result<ManagedWorkerStatus, SupervisionError>

Start (or adopt) supervision of one deployment, recording the intent durably first. A durable desired-state flip is published onto the cluster channel, so the console’s live feed sees it whichever transport asked.

Idempotent: a deployment already being supervised is reported as it is, without a second process (and, already desiring Running, without a second event).

Errors

Returns SupervisionError::NotCommissioned when no policy is installed — carrying the remedy — SupervisionError::UnknownDeployment for a name with no durable record, and SupervisionError::Store when the record cannot be read or the intent cannot be written.

pub async fn converge( &self, name: &str, mode: Convergence, ) -> Result<ManagedWorkerStatus, SupervisionError>

Bring ONE deployment’s live state to its already-durable desired state, writing no durable intent and publishing no desired-state event.

This is what a transport calls immediately after it has written the record itself. A durable desired = running is an INSTRUCTION, and the node that accepted it acts on it now rather than at the next boot reconcile — but the write and its event already happened at the transport, so re-writing them here would double-publish a change that occurred once. Self::start and Self::stop remain the verbs for a caller that is asking for the intent to CHANGE.

mode decides what an already-live instance means: Convergence::Idempotent leaves it running, Convergence::Replacing tears it down first because the record it is replaying was rewritten under it.

Errors

Returns SupervisionError::UnknownDeployment for a name with no durable record, SupervisionError::Store when the record cannot be read, SupervisionError::NotCommissioned when the record wants to run and no policy is installed, and SupervisionError::StopIncomplete when a replaced or stopped instance cannot be proven gone.

pub async fn forget(&self, name: &str) -> Result<(), SupervisionError>

Drop one deployment out of supervision entirely, proving any live instance stopped first, without touching durable state.

For the transport that has just DELETED the record: a supervised process whose record no longer exists is an orphan nothing can report on, stop, or restart — report() joins the durable listing, so it would vanish from every surface while still holding its port and its process group.

Errors

Returns SupervisionError::StopIncomplete when the instance’s process group cannot be proven empty, SupervisionError::TaskLost when its supervision task ended abnormally, and SupervisionError::StatePoisoned when the instance map was poisoned.

pub async fn stop( &self, name: &str, ) -> Result<ManagedWorkerStatus, SupervisionError>

Stop one deployment and record the intent durably. The durable desired-state write is published onto the cluster channel, so the console’s live feed sees it whichever transport asked.

The returned status is written only once the process group has been probed empty; a group that survives the termination ladder produces SupervisionError::StopIncomplete instead, so a caller can never read “stopped” off bookkeeping alone.

Errors

Returns SupervisionError::UnknownDeployment for an absent record, SupervisionError::Store when the intent cannot be written, SupervisionError::StopIncomplete when the group cannot be proven empty, and SupervisionError::TaskLost when the supervision task ended abnormally.

pub async fn restart( &self, name: &str, ) -> Result<ManagedWorkerStatus, SupervisionError>

Stop and start one deployment, leaving desired state at Running.

Publishes exactly what it durably writes: a restart of an already-Running deployment changes no desired state and emits no desired-state event; one that flips it inherits Self::start’s.

Errors

Returns the same failures as Self::stop and Self::start.

pub async fn reconcile(&self) -> Result<usize, SupervisionError>

Bring the fleet to its durable desired state.

Called at boot and safe to call again: deployments already supervised are left alone. Returns the number of deployments now supervised.

Each record is brought up under its OWN per-name gate, and its record is re-read inside it: a reconcile running beside a concurrent PUT must not spawn the argv it happened to list before that write landed.

Errors

Returns SupervisionError::NotCommissioned when no policy is installed, and SupervisionError::Store when the durable listing cannot be read.

pub async fn report(&self) -> Result<ManagedWorkerReport, SupervisionError>

Join durable intent with live supervision for every deployment.

Errors

Returns SupervisionError::Store when the durable listing cannot be read, and SupervisionError::StatePoisoned when a status cell was poisoned by a panicking holder.

pub async fn shutdown(&self) -> FleetShutdownReport

Stop every supervised instance, for server shutdown.

Durable desired state is deliberately NOT changed: a server going down is not an operator asking for the fleet to stay down, and the next boot reconciles it back up. The report names both sides: every instance proven stopped (its process group observed empty) and one failure per instance that could not be — an empty failure list is the proof that no worker was orphaned, and the stopped names let the shutdown outcome record say WHICH workers went down rather than a count.