aion_server/worker/supervisor/

fleet.rs

//! The managed-worker supervisor: desired state in, running processes out.
//!
//! Desired state is the DURABLE record (`WorkerDeployment.desired`); actual
//! state is a live [`InstanceHandle`]. This type is the only thing that reads
//! one and writes the other, and every answer it gives is a join of the two —
//! never one standing in for the other.
//!
//! Supervision is COMMISSIONED, not defaulted. A server whose operator has not
//! written a `[worker_supervision]` policy supervises nothing and says so with
//! the remedy attached; it does not pick a backoff on the operator's behalf
//! (ADR-001).

use std::collections::BTreeMap;
use std::sync::{Arc, Mutex, OnceLock};

use aion_core::ClusterEvent;
use aion_store::{
    DesiredState, WorkerArtifactRef, WorkerDeployment, WorkerDeploymentListing,
    WorkerDeploymentStore,
};

use crate::cluster_publisher::ClusterEventPublisher;
use crate::worker::auto_provision::AutoWorkerOutcome;

use super::error::SupervisionError;
use super::executable::ManagedExecutable;
use super::instance::{InstanceConfig, InstanceHandle, InstanceSnapshot};
use super::policy::{SupervisionPolicy, UNCOMMISSIONED_REMEDY};
use super::status::{ManagedWorkerReport, ManagedWorkerState, ManagedWorkerStatus};

/// What an operator commissioned: how to supervise, and what to launch.
///
/// Both halves arrive together and neither can be swapped afterwards. They
/// belong to the same decision — a restart discipline is meaningless without
/// knowing what is being restarted — and holding them in one write-once cell
/// is what makes "commissioned" a single, checkable fact.
#[derive(Clone, Debug)]
struct Commission {
    policy: SupervisionPolicy,
    executable: ManagedExecutable,
}

/// How far a convergence may go when a live instance already exists.
///
/// The distinction is the staleness law: a supervised worker replays its
/// record's argv VERBATIM, so a record whose argv was rewritten underneath a
/// running instance is being served by a launch nobody deployed. Only the
/// caller knows which of the two happened, so only the caller may choose.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum Convergence {
    /// Leave an already-live instance running: it is serving this record's argv.
    Idempotent,
    /// Replace any live instance, because the record's argv was rewritten.
    Replacing,
}

/// What [`WorkerSupervisor::shutdown`] observed, by name on both sides.
#[derive(Debug, Default)]
pub struct FleetShutdownReport {
    /// Deployments proven stopped: each stop returned only after a
    /// signal-zero probe found the worker's process group empty.
    pub stopped: Vec<String>,
    /// One entry per deployment that could NOT be proven stopped, carrying
    /// the observation that contradicted it.
    pub failures: Vec<SupervisionError>,
}

/// Server-owned supervision of the managed worker fleet.
pub struct WorkerSupervisor {
    store: Arc<dyn WorkerDeploymentStore>,
    commission: OnceLock<Commission>,
    instances: Mutex<BTreeMap<String, InstanceHandle>>,
    publisher: ClusterEventPublisher,
    /// The LATEST auto-provision decision per task queue, so the status
    /// surface can say why a queue has no worker long after the deploy or the
    /// boot that decided it. Keyed by queue and therefore bounded by what has
    /// been deployed, never by how often.
    auto_provision: Mutex<BTreeMap<String, AutoWorkerOutcome>>,
    /// One async gate per deployment NAME, so every lifecycle verb's
    /// read-record → tear-down → spawn sequence is a single critical section.
    ///
    /// The instances map alone is not enough. It makes two workers for one name
    /// impossible, but it is released between the durable read and the spawn,
    /// and two interleaved converges of one name could therefore leave the
    /// EARLIER argv running: read(argv1) … read(argv2), replace→spawn(argv2),
    /// replace→spawn(argv1). Nothing would notice — `reconcile` leaves live
    /// instances alone — so the superseded launch survives until the next
    /// restart, which is exactly the staleness `Convergence::Replacing` exists
    /// to prevent.
    ///
    /// Keyed by name and therefore bounded by the deployment count. Entries
    /// are never removed: a `tokio::sync::Mutex` is a few words, and reclaiming
    /// them would need a second lock ordering to get wrong.
    gates: Mutex<BTreeMap<String, Arc<tokio::sync::Mutex<()>>>>,
}

impl WorkerSupervisor {
    /// 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.
    #[must_use]
    pub fn new(store: Arc<dyn WorkerDeploymentStore>, publisher: ClusterEventPublisher) -> Self {
        Self {
            store,
            commission: OnceLock::new(),
            instances: Mutex::new(BTreeMap::new()),
            publisher,
            auto_provision: Mutex::new(BTreeMap::new()),
            gates: Mutex::new(BTreeMap::new()),
        }
    }

    /// The per-name gate, created on first use.
    ///
    /// A poisoned gate map is a typed error rather than a silent free-for-all:
    /// proceeding without the gate is exactly the interleaving it exists to
    /// prevent, and doing that quietly would make the race intermittent instead
    /// of impossible.
    fn gate(&self, name: &str) -> Result<Arc<tokio::sync::Mutex<()>>, SupervisionError> {
        let mut gates = self
            .gates
            .lock()
            .map_err(|poison| SupervisionError::StatePoisoned {
                detail: poison.to_string(),
            })?;
        Ok(Arc::clone(
            gates
                .entry(name.to_owned())
                .or_insert_with(|| Arc::new(tokio::sync::Mutex::new(()))),
        ))
    }

    /// 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 record_auto_provision(&self, outcomes: &[AutoWorkerOutcome]) {
        match self.auto_provision.lock() {
            Ok(mut log) => {
                for outcome in outcomes {
                    drop(log.insert(outcome.task_queue.clone(), outcome.clone()));
                }
            }
            Err(poison) => tracing::error!(
                detail = %poison,
                "the auto-provision log is poisoned; this server's managed-worker report will \
                 not say why a queue has no built-in agent worker"
            ),
        }
    }

    /// The latest auto-provision decision per queue, ordered by queue.
    ///
    /// A poisoned log reports EMPTY rather than refusing the whole status read:
    /// the fleet join is the operator's primary answer and must keep arriving.
    /// The poison itself was already logged where it was found.
    fn auto_provision_log(&self) -> Vec<AutoWorkerOutcome> {
        self.auto_provision
            .lock()
            .map(|log| log.values().cloned().collect())
            .unwrap_or_default()
    }

    /// 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 commission(&self, policy: SupervisionPolicy, executable: ManagedExecutable) -> bool {
        self.commission
            .set(Commission { policy, executable })
            .is_ok()
    }

    /// The installed policy, if this server has one.
    #[must_use]
    pub fn policy(&self) -> Option<SupervisionPolicy> {
        self.commission.get().map(|commission| commission.policy)
    }

    /// Whether an operator commission is installed.
    #[must_use]
    pub fn is_commissioned(&self) -> bool {
        self.commission.get().is_some()
    }

    fn require_commission(&self) -> Result<&Commission, SupervisionError> {
        self.commission
            .get()
            .ok_or(SupervisionError::NotCommissioned)
    }

    fn instances(
        &self,
    ) -> Result<std::sync::MutexGuard<'_, BTreeMap<String, InstanceHandle>>, SupervisionError> {
        self.instances
            .lock()
            .map_err(|poison| SupervisionError::StatePoisoned {
                detail: poison.to_string(),
            })
    }

    async fn record(&self, name: &str) -> Result<WorkerDeployment, SupervisionError> {
        self.store
            .get_worker_deployment(name)
            .await
            .map_err(|source| SupervisionError::Store { source })?
            .ok_or_else(|| SupervisionError::UnknownDeployment {
                name: name.to_owned(),
            })
    }

    async fn listing(&self) -> Result<WorkerDeploymentListing, SupervisionError> {
        self.store
            .list_worker_deployments()
            .await
            .map_err(|source| SupervisionError::Store { source })
    }

    /// 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 start(&self, name: &str) -> Result<ManagedWorkerStatus, SupervisionError> {
        let gate = self.gate(name)?;
        let held = gate.lock().await;
        let started = self.start_gated(name).await;
        drop(held);
        started
    }

    /// [`Self::start`] with the per-name gate ALREADY held by the caller.
    async fn start_gated(&self, name: &str) -> Result<ManagedWorkerStatus, SupervisionError> {
        let commission = self.require_commission()?.clone();
        let record = self.record(name).await?;
        let record = if record.desired == DesiredState::Running {
            record
        } else {
            let record = self
                .store
                .set_desired_state(name, DesiredState::Running)
                .await
                .map_err(|source| SupervisionError::Store { source })?
                .ok_or_else(|| SupervisionError::UnknownDeployment {
                    name: name.to_owned(),
                })?;
            self.publish_desired_state(&record);
            record
        };

        let snapshot = self.ensure_instance(&record, &commission)?;
        Ok(status_of(&record, snapshot, self.is_commissioned()))
    }

    /// 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 converge(
        &self,
        name: &str,
        mode: Convergence,
    ) -> Result<ManagedWorkerStatus, SupervisionError> {
        let gate = self.gate(name)?;
        let held = gate.lock().await;
        let converged = self.converge_gated(name, mode).await;
        drop(held);
        converged
    }

    /// [`Self::converge`] with the per-name gate ALREADY held by the caller.
    ///
    /// Everything from the durable read to the spawn happens inside that gate:
    /// two concurrent converges of one name are serialized, so the launch left
    /// running is always the one the LAST durable write named.
    async fn converge_gated(
        &self,
        name: &str,
        mode: Convergence,
    ) -> Result<ManagedWorkerStatus, SupervisionError> {
        let record = self.record(name).await?;
        if record.desired == DesiredState::Stopped {
            self.tear_down(name).await?;
            return Ok(status_of(&record, None, self.is_commissioned()));
        }
        // Commission is required only to RUN something. Checking it after the
        // stopped arm above is deliberate: an uncommissioned server must still
        // be able to converge a deployment DOWN.
        let commission = self.require_commission()?.clone();
        if mode == Convergence::Replacing {
            self.tear_down(name).await?;
        }
        let snapshot = self.ensure_instance(&record, &commission)?;
        Ok(status_of(&record, snapshot, self.is_commissioned()))
    }

    /// 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 forget(&self, name: &str) -> Result<(), SupervisionError> {
        let gate = self.gate(name)?;
        let held = gate.lock().await;
        let forgotten = self.tear_down(name).await;
        drop(held);
        forgotten
    }

    /// Remove any live instance for `name` and prove its process group empty.
    /// Absent instances are a no-op, which is what makes every caller idempotent.
    ///
    /// The pid and process group are read BEFORE the stop, because the stop
    /// consumes the handle: a failure to prove the group empty is the one
    /// moment those two numbers matter, and they are gone by the time the
    /// failure exists unless they are captured first.
    async fn tear_down(&self, name: &str) -> Result<(), SupervisionError> {
        let handle = self.instances()?.remove(name);
        let Some(handle) = handle else {
            return Ok(());
        };
        let identity = handle.snapshot().ok().map(|snapshot| ProcessIdentity {
            pid: snapshot.pid,
            process_group: snapshot.process_group,
        });
        handle
            .stop(name)
            .await
            .map_err(|error| attach_process_identity(error, identity))
    }

    /// Ensure a live instance exists for `record`, returning its snapshot.
    ///
    /// The lock is held across the liveness test and the insert so two callers
    /// cannot both observe "not live" and spawn two workers for one deployment.
    fn ensure_instance(
        &self,
        record: &WorkerDeployment,
        commission: &Commission,
    ) -> Result<Option<InstanceSnapshot>, SupervisionError> {
        let mut instances = self.instances()?;
        let live = instances
            .get(&record.name)
            .is_some_and(|handle| !handle.is_finished());
        if !live {
            let handle = InstanceHandle::start(self.instance_config(record, commission));
            drop(instances.insert(record.name.clone(), handle));
        }
        instances
            .get(&record.name)
            .map(InstanceHandle::snapshot)
            .transpose()
    }

    /// 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 stop(&self, name: &str) -> Result<ManagedWorkerStatus, SupervisionError> {
        let gate = self.gate(name)?;
        let held = gate.lock().await;
        let stopped = self.stop_gated(name).await;
        drop(held);
        stopped
    }

    /// [`Self::stop`] with the per-name gate ALREADY held by the caller.
    async fn stop_gated(&self, name: &str) -> Result<ManagedWorkerStatus, SupervisionError> {
        let record = self
            .store
            .set_desired_state(name, DesiredState::Stopped)
            .await
            .map_err(|source| SupervisionError::Store { source })?
            .ok_or_else(|| SupervisionError::UnknownDeployment {
                name: name.to_owned(),
            })?;
        self.publish_desired_state(&record);
        self.tear_down(name).await?;
        Ok(status_of(&record, None, self.is_commissioned()))
    }

    /// 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 restart(&self, name: &str) -> Result<ManagedWorkerStatus, SupervisionError> {
        // ONE gate for the whole stop-then-start, taken here and held across
        // both halves: a restart that released it between them would let
        // another caller's converge spawn into the gap and then be torn down
        // by this restart's own start.
        let gate = self.gate(name)?;
        let held = gate.lock().await;
        let restarted = self.restart_gated(name).await;
        drop(held);
        restarted
    }

    /// [`Self::restart`] with the per-name gate ALREADY held by the caller.
    async fn restart_gated(&self, name: &str) -> Result<ManagedWorkerStatus, SupervisionError> {
        self.require_commission()?;
        self.record(name).await?;
        self.tear_down(name).await?;
        self.start_gated(name).await
    }

    /// 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 reconcile(&self) -> Result<usize, SupervisionError> {
        drop(self.require_commission()?.clone());
        let listing = self.listing().await?;
        let mut supervised = 0_usize;
        for record in listing
            .deployments
            .iter()
            .filter(|record| record.desired == DesiredState::Running)
        {
            match self.converge(&record.name, Convergence::Idempotent).await {
                Ok(_) => supervised = supervised.saturating_add(1),
                // One record that cannot be converged must not abandon the
                // rest of the fleet: a boot that stops at the first failure
                // leaves every later deployment unsupervised and silent.
                Err(error) => tracing::error!(
                    worker = record.name.as_str(),
                    %error,
                    "worker deployment could not be converged; it is not being supervised"
                ),
            }
        }
        for poisoned in &listing.undecodable {
            tracing::error!(
                worker = poisoned.name.as_str(),
                error = poisoned.error.as_str(),
                "worker deployment record could not be decoded; it is not being supervised"
            );
        }
        Ok(supervised)
    }

    /// 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 report(&self) -> Result<ManagedWorkerReport, SupervisionError> {
        let listing = self.listing().await?;
        let commissioned = self.is_commissioned();
        let mut workers = Vec::with_capacity(listing.deployments.len());
        {
            let instances = self.instances()?;
            for record in &listing.deployments {
                let snapshot = instances
                    .get(&record.name)
                    .map(InstanceHandle::snapshot)
                    .transpose()?;
                workers.push(status_of(record, snapshot, commissioned));
            }
        }
        Ok(ManagedWorkerReport {
            commissioned,
            remedy: (!commissioned).then(|| UNCOMMISSIONED_REMEDY.to_owned()),
            workers,
            undecodable: listing
                .undecodable
                .iter()
                .map(|poisoned| poisoned.name.clone())
                .collect(),
            auto_provision: self.auto_provision_log(),
        })
    }

    /// 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.
    pub async fn shutdown(&self) -> FleetShutdownReport {
        let handles = match self.instances() {
            Ok(mut instances) => std::mem::take(&mut *instances),
            Err(error) => {
                return FleetShutdownReport {
                    stopped: Vec::new(),
                    failures: vec![error],
                };
            }
        };
        let mut report = FleetShutdownReport {
            stopped: Vec::new(),
            failures: Vec::new(),
        };
        for (name, handle) in handles {
            match handle.stop(&name).await {
                Ok(()) => report.stopped.push(name),
                Err(error) => report.failures.push(error),
            }
        }
        report
    }

    /// Publish one durable desired-state write onto the cluster channel — the
    /// SAME event the worker-deployment desired-state endpoint publishes for
    /// its writes, so the live feed cannot tell the transports apart.
    ///
    /// Called exactly where a write happened, never speculatively: an event
    /// here is proof of a durable change. The emitted event's return value is
    /// dropped deliberately — a channel with no subscribers is the calm state,
    /// not a failure.
    fn publish_desired_state(&self, record: &WorkerDeployment) {
        let name = record.name.clone();
        let desired_state = record.desired;
        drop(
            self.publisher
                .emit(|meta| ClusterEvent::WorkerDeploymentDesiredStateChanged {
                    meta,
                    name,
                    desired_state,
                }),
        );
    }

    fn instance_config(
        &self,
        record: &WorkerDeployment,
        commission: &Commission,
    ) -> InstanceConfig {
        let WorkerArtifactRef::Builtin { verb } = &record.artifact;
        InstanceConfig {
            name: record.name.clone(),
            verb: verb.clone(),
            executable: commission.executable.clone(),
            policy: commission.policy,
            store: Arc::clone(&self.store),
        }
    }
}

/// The two numbers an operator needs to find a process the server could not
/// prove stopped, captured before the stop consumes the handle that holds them.
struct ProcessIdentity {
    pid: Option<u32>,
    process_group: Option<i32>,
}

/// Name the surviving process group in a stop failure.
///
/// Only [`SupervisionError::StopIncomplete`] is rewritten, and only when the
/// identity was actually read: every other failure is about something else, and
/// an unknown pid is reported as unknown rather than as a number nobody
/// measured.
fn attach_process_identity(
    error: SupervisionError,
    identity: Option<ProcessIdentity>,
) -> SupervisionError {
    match (error, identity) {
        (SupervisionError::StopIncomplete { name, detail }, Some(identity)) => {
            let pid = identity
                .pid
                .map_or_else(|| String::from("unknown"), |pid| pid.to_string());
            let group = identity
                .process_group
                .map_or_else(|| String::from("unknown"), |group| group.to_string());
            SupervisionError::StopIncomplete {
                name,
                detail: format!("{detail} (last observed pid {pid}, process group {group})"),
            }
        }
        (error, _) => error,
    }
}

/// Join one durable record with one live snapshot.
fn status_of(
    record: &WorkerDeployment,
    snapshot: Option<InstanceSnapshot>,
    commissioned: bool,
) -> ManagedWorkerStatus {
    // Three different absences, three different answers. An instance that
    // exists but has not spoken yet is STARTING; a server that could not have
    // supervised is UNSUPERVISED; only genuinely nothing-to-run is STOPPED.
    // Collapsing any of them into another is how a status surface comes to
    // report a terminal state about work that is under way.
    let supervised = snapshot.is_some();
    let snapshot = snapshot.unwrap_or_default();
    let state = snapshot.state.unwrap_or({
        if supervised {
            ManagedWorkerState::Starting
        } else if record.desired == DesiredState::Running && !commissioned {
            ManagedWorkerState::Uncommissioned
        } else {
            ManagedWorkerState::Stopped
        }
    });
    ManagedWorkerStatus {
        name: record.name.clone(),
        task_queue: record.task_queue.clone(),
        desired: record.desired,
        state,
        pid: snapshot.pid,
        process_group: snapshot.process_group,
        restarts: snapshot.restarts,
        last_exit: snapshot.last_exit,
        last_error: snapshot.last_error,
        deployed_binary: record.binary.clone(),
        spawn_binary: snapshot.spawn_binary,
    }
}