xy/crates/xy-supervisor/src/supervisor.rs

use crate::{
    backoff::Backoff,
    child::ChildHandle,
    logs::LogSink,
    policy::{RestartDecision, decide},
    retry_window::RetryWindow,
};
use std::time::{Duration, Instant};
use tokio::sync::{mpsc, oneshot, watch};
use tokio::time::sleep;
use tracing::{debug, info, warn};
use xy_protocol::{ServerConfig, ServerState};

pub enum SupervisorCmd {
    Start {
        ack: oneshot::Sender<StartAck>,
    },
    Stop {
        ack: oneshot::Sender<StopAck>,
    },
    Restart {
        ack: oneshot::Sender<()>,
    },
    Reconfigure {
        new: Box<ServerConfig>,
        ack: oneshot::Sender<()>,
    },
    Shutdown {
        ack: oneshot::Sender<()>,
    },
}

#[derive(Debug, PartialEq, Eq)]
pub enum StartAck {
    Started,
    AlreadyRunning,
    SpawnFailed(String),
}

#[derive(Debug, PartialEq, Eq)]
pub enum StopAck {
    Stopped,
    NotRunning,
}

#[derive(Debug, Clone)]
pub struct WaitProgress {
    pub description: String,
    pub started_at: Instant,
    pub timeout: Duration,
}

#[derive(Debug, Clone)]
pub struct Status {
    pub state: ServerState,
    pub pid: Option<u32>,
    pub port: u16,
    pub started_at: Option<Instant>,
    pub restart_count: u32,
    pub last_exit: Option<i32>,
    pub waiting: Option<WaitProgress>,
}

#[derive(Clone)]
pub struct SupervisorHandle {
    pub name: String,
    pub tx: mpsc::Sender<SupervisorCmd>,
    pub status: watch::Receiver<Status>,
    pub log_sink: LogSink,
}

#[async_trait::async_trait]
pub trait Spawner: Send + 'static {
    type Child: ChildHandle;
    async fn spawn(&self, cfg: &ServerConfig, sink: LogSink) -> std::io::Result<Self::Child>;
}

pub struct SupervisorTask<S: Spawner> {
    cfg: ServerConfig,
    log_sink: LogSink,
    spawner: S,
    status_tx: watch::Sender<Status>,
    cmd_rx: mpsc::Receiver<SupervisorCmd>,
    backoff: Backoff,
    retry_window: RetryWindow,
    restart_count: u32,
    last_exit: Option<i32>,
    started_at: Option<Instant>,
    current_pid: Option<u32>,
    waiting: Option<WaitProgress>,
}

impl<S: Spawner> SupervisorTask<S> {
    pub fn new(
        cfg: ServerConfig,
        log_sink: LogSink,
        spawner: S,
        status_tx: watch::Sender<Status>,
        cmd_rx: mpsc::Receiver<SupervisorCmd>,
    ) -> Self {
        let backoff = Backoff::new(cfg.restart.backoff_initial, cfg.restart.backoff_max);
        let retry_window =
            RetryWindow::new(Duration::from_secs(60), cfg.restart.max_retries_per_minute);

        Self {
            cfg,
            log_sink,
            spawner,
            status_tx,
            cmd_rx,
            backoff,
            retry_window,
            restart_count: 0,
            last_exit: None,
            started_at: None,
            current_pid: None,
            waiting: None,
        }
    }

    fn set_state(&mut self, s: ServerState) {
        let _ = self.status_tx.send(Status {
            state: s,
            pid: self.current_pid,
            port: self.cfg.port,
            started_at: self.started_at,
            restart_count: self.restart_count,
            last_exit: self.last_exit,
            waiting: self.waiting.clone(),
        });
    }

    pub async fn run(mut self) {
        let mut child: Option<S::Child> = None;

        loop {
            tokio::select! {
                cmd = self.cmd_rx.recv() => {
                    let Some(cmd) = cmd else { break; };

                    match cmd {
                        SupervisorCmd::Start { ack } => {
                            if child.is_some() {
                                let _ = ack.send(StartAck::AlreadyRunning);
                            } else {
                                match self.do_start(StartCause::Initial).await {
                                    Ok(c) => {
                                        child = Some(c);
                                        let _ = ack.send(StartAck::Started);
                                    }
                                    Err(StartFailure::Spawn(err)) => {
                                        warn!(name = %self.cfg.name, error = %err, "spawn failed");
                                        self.started_at = None;
                                        self.set_state(ServerState::Failed);
                                        let _ = ack.send(StartAck::SpawnFailed(err.to_string()));
                                    }
                                    Err(StartFailure::WaitTimedOut) => {
                                        warn!(name = %self.cfg.name, "wait-for timed out");
                                        self.started_at = None;
                                        self.set_state(ServerState::Failed);
                                        let _ = ack.send(StartAck::SpawnFailed(
                                            "wait-for timed out".to_string(),
                                        ));
                                    }
                                    Err(StartFailure::Cancelled) => {
                                        self.started_at = None;
                                        self.set_state(ServerState::Stopped);
                                    }
                                    Err(StartFailure::Shutdown) => return,
                                }
                            }
                        }
                        SupervisorCmd::Stop { ack } => {
                            if let Some(c) = child.take() {
                                self.do_stop(c).await;
                                let _ = ack.send(StopAck::Stopped);
                            } else {
                                let _ = ack.send(StopAck::NotRunning);
                            }
                        }
                        SupervisorCmd::Restart { ack } => {
                            if let Some(c) = child.take() {
                                self.set_state(ServerState::Restarting);
                                self.do_stop(c).await;
                            }

                            match self.do_start(StartCause::Restart).await {
                                Ok(c) => child = Some(c),
                                Err(StartFailure::Spawn(err)) => {
                                    warn!(name = %self.cfg.name, error = %err, "restart spawn failed");
                                    self.started_at = None;
                                    self.set_state(ServerState::Failed);
                                }
                                Err(StartFailure::WaitTimedOut) => {
                                    warn!(name = %self.cfg.name, "wait-for timed out");
                                    self.started_at = None;
                                    self.set_state(ServerState::Failed);
                                }
                                Err(StartFailure::Cancelled) => {
                                    self.started_at = None;
                                    self.set_state(ServerState::Stopped);
                                }
                                Err(StartFailure::Shutdown) => return,
                            }

                            let _ = ack.send(());
                        }
                        SupervisorCmd::Reconfigure { new, ack } => {
                            self.cfg = *new;
                            self.backoff =
                                Backoff::new(self.cfg.restart.backoff_initial, self.cfg.restart.backoff_max);
                            self.retry_window = RetryWindow::new(
                                Duration::from_secs(60),
                                self.cfg.restart.max_retries_per_minute,
                            );

                            let _ = ack.send(());
                        }
                        SupervisorCmd::Shutdown { ack } => {
                            if let Some(c) = child.take() {
                                self.do_stop(c).await;
                            }

                            let _ = ack.send(());
                            return;
                        }
                    }
                }
                code = wait_child(&mut child) => {
                    child = None;

                    self.last_exit = code;
                    self.current_pid = None;

                    let now = Instant::now();

                    self.retry_window.record(now);

                    let cap = self.retry_window.cap_reached(now);
                    let decision = decide(self.cfg.restart.policy, code, cap);

                    debug!(name = %self.cfg.name, ?code, ?decision, "child exited");

                    match decision {
                        RestartDecision::StayStopped => {
                            self.started_at = None;
                            self.set_state(ServerState::Stopped);
                        }
                        RestartDecision::MarkFailed => {
                            self.started_at = None;
                            self.set_state(ServerState::Failed);
                        }
                        RestartDecision::Restart => {
                            self.set_state(ServerState::Restarting);

                            let delay = self.backoff.next();

                            enum Action {
                                RetryNow,
                                Cancel,
                                Exit,
                            }

                            let mut delay_fut = std::pin::pin!(sleep(delay));

                            let action = tokio::select! {
                                _ = &mut delay_fut => Action::RetryNow,
                                cmd = self.cmd_rx.recv() => match cmd {
                                    None => Action::Exit,
                                    Some(SupervisorCmd::Stop { ack }) => {
                                        let _ = ack.send(StopAck::NotRunning);
                                        Action::Cancel
                                    }
                                    Some(SupervisorCmd::Shutdown { ack }) => {
                                        let _ = ack.send(());
                                        return;
                                    }
                                    Some(SupervisorCmd::Start { ack }) => {
                                        let _ = ack.send(StartAck::Started);
                                        Action::RetryNow
                                    }
                                    Some(SupervisorCmd::Restart { ack }) => {
                                        let _ = ack.send(());
                                        Action::RetryNow
                                    }
                                    Some(SupervisorCmd::Reconfigure { new, ack }) => {
                                        self.cfg = *new;
                                        self.backoff = Backoff::new(
                                            self.cfg.restart.backoff_initial,
                                            self.cfg.restart.backoff_max,
                                        );
                                        self.retry_window = RetryWindow::new(
                                            Duration::from_secs(60),
                                            self.cfg.restart.max_retries_per_minute,
                                        );
                                        let _ = ack.send(());
                                        Action::RetryNow
                                    }
                                },
                            };

                            match action {
                                Action::RetryNow => {
                                    match self.do_start(StartCause::Restart).await {
                                        Ok(c) => child = Some(c),
                                        Err(StartFailure::Spawn(err)) => {
                                            warn!(name = %self.cfg.name, error = %err, "restart spawn failed");
                                            self.started_at = None;
                                            self.set_state(ServerState::Failed);
                                        }
                                        Err(StartFailure::WaitTimedOut) => {
                                            warn!(name = %self.cfg.name, "wait-for timed out");
                                            self.started_at = None;
                                            self.set_state(ServerState::Failed);
                                        }
                                        Err(StartFailure::Cancelled) => {
                                            self.started_at = None;
                                            self.set_state(ServerState::Stopped);
                                        }
                                        Err(StartFailure::Shutdown) => return,
                                    }
                                }
                                Action::Cancel => {
                                    self.started_at = None;
                                    self.set_state(ServerState::Stopped);
                                }
                                Action::Exit => return,
                            }
                        }
                    }
                }
            }
        }
    }

    async fn await_ready(&mut self) -> GateOutcome {
        let Some(wait) = self.cfg.wait_for.clone() else {
            return GateOutcome::Ready;
        };

        let started_at = Instant::now();

        self.waiting = Some(WaitProgress {
            description: crate::ready::describe(&wait.condition),
            started_at,
            timeout: wait.timeout,
        });

        self.started_at = None;

        self.set_state(ServerState::Waiting);

        let outcome = loop {
            if crate::ready::is_ready(&wait.condition, wait.interval).await {
                break GateOutcome::Ready;
            }

            if started_at.elapsed() >= wait.timeout {
                break GateOutcome::TimedOut;
            }

            let mut tick = std::pin::pin!(sleep(wait.interval));

            let interrupted = tokio::select! {
                _ = &mut tick => None,
                cmd = self.cmd_rx.recv() => match cmd {
                    None => Some(GateOutcome::Shutdown),
                    Some(SupervisorCmd::Stop { ack }) => {
                        let _ = ack.send(StopAck::NotRunning);
                        Some(GateOutcome::Cancelled)
                    }
                    Some(SupervisorCmd::Shutdown { ack }) => {
                        let _ = ack.send(());
                        Some(GateOutcome::Shutdown)
                    }
                    Some(SupervisorCmd::Start { ack }) => {
                        let _ = ack.send(StartAck::Started);
                        None
                    }
                    Some(SupervisorCmd::Restart { ack }) => {
                        let _ = ack.send(());
                        None
                    }
                    Some(SupervisorCmd::Reconfigure { new, ack }) => {
                        self.cfg = *new;
                        self.backoff =
                            Backoff::new(self.cfg.restart.backoff_initial, self.cfg.restart.backoff_max);
                        self.retry_window = RetryWindow::new(
                            Duration::from_secs(60),
                            self.cfg.restart.max_retries_per_minute,
                        );
                        let _ = ack.send(());
                        None
                    }
                },
            };

            if let Some(outcome) = interrupted {
                break outcome;
            }
        };

        self.waiting = None;

        outcome
    }

    async fn do_start(&mut self, cause: StartCause) -> Result<S::Child, StartFailure> {
        match self.await_ready().await {
            GateOutcome::Ready => {}
            GateOutcome::TimedOut => return Err(StartFailure::WaitTimedOut),
            GateOutcome::Cancelled => return Err(StartFailure::Cancelled),
            GateOutcome::Shutdown => return Err(StartFailure::Shutdown),
        }

        self.set_state(ServerState::Starting);

        let c = self
            .spawner
            .spawn(&self.cfg, self.log_sink.clone())
            .await
            .map_err(StartFailure::Spawn)?;

        if cause == StartCause::Restart {
            self.restart_count = self.restart_count.saturating_add(1);
        }

        self.started_at = Some(Instant::now());
        self.current_pid = Some(c.pid());
        self.backoff.reset();
        self.set_state(ServerState::Running);

        info!(name = %self.cfg.name, pid = c.pid(), "started");

        Ok(c)
    }

    async fn do_stop(&mut self, mut c: S::Child) {
        self.set_state(ServerState::Stopping);

        let _ = c.terminate();

        let grace = self.cfg.stop.grace;

        match tokio::time::timeout(grace, c.wait()).await {
            Ok(_) => {}
            Err(_) => {
                let _ = c.kill();
                let _ = c.wait().await;
            }
        }

        self.current_pid = None;
        self.started_at = None;
        self.set_state(ServerState::Stopped);
    }
}

async fn wait_child<C: ChildHandle>(slot: &mut Option<C>) -> Option<i32> {
    match slot.as_mut() {
        Some(c) => c.wait().await.ok().flatten(),
        None => std::future::pending().await,
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum StartCause {
    Initial,
    Restart,
}

enum StartFailure {
    Spawn(std::io::Error),
    WaitTimedOut,
    Cancelled,
    Shutdown,
}

enum GateOutcome {
    Ready,
    TimedOut,
    Cancelled,
    Shutdown,
}

pub struct RealSpawner;

#[async_trait::async_trait]
impl Spawner for RealSpawner {
    type Child = crate::child::RealChild;

    async fn spawn(&self, cfg: &ServerConfig, sink: LogSink) -> std::io::Result<Self::Child> {
        crate::child::spawn_with_logs(cfg, sink)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::child::MockChild;
    use crate::logs::{LogSink, RotatingLogWriter};
    use std::path::PathBuf;
    use std::sync::{Arc, Mutex};
    use tempfile::tempdir;
    use xy_protocol::{RestartConfig, RestartPolicy, StopConfig, WaitCondition, WaitForConfig};

    struct QueueSpawner {
        queue: Arc<Mutex<Vec<MockChild>>>,
    }

    #[async_trait::async_trait]
    impl Spawner for QueueSpawner {
        type Child = MockChild;

        async fn spawn(&self, _cfg: &ServerConfig, _sink: LogSink) -> std::io::Result<MockChild> {
            let mut q = self.queue.lock().unwrap();
            Ok(q.remove(0))
        }
    }

    fn cfg(name: &str, policy: RestartPolicy, max_retries: u32) -> ServerConfig {
        ServerConfig {
            name: name.to_string(),
            command: "/bin/true".into(),
            args: vec![],
            port: 1,
            env: Default::default(),
            working_dir: None,
            restart: RestartConfig {
                policy,
                backoff_initial: Duration::from_millis(1),
                backoff_max: Duration::from_millis(1),
                max_retries_per_minute: max_retries,
            },
            stop: StopConfig {
                grace: Duration::from_millis(50),
            },
            wait_for: None,
            stdin: Default::default(),
        }
    }

    fn cfg_with_wait(name: &str, wait: WaitForConfig) -> ServerConfig {
        let mut c = cfg(name, RestartPolicy::Never, 5);
        c.wait_for = Some(wait);
        c
    }

    fn sink(name: &str) -> LogSink {
        let dir = tempdir().unwrap();
        let writer = RotatingLogWriter::open(&dir.path().join("s.log"), 1024, 3).unwrap();
        std::mem::forget(dir);
        LogSink::new(name.to_string(), writer, 1024)
    }

    fn initial_status(cfg: &ServerConfig) -> Status {
        Status {
            state: ServerState::Stopped,
            pid: None,
            port: cfg.port,
            started_at: None,
            restart_count: 0,
            last_exit: None,
            waiting: None,
        }
    }

    async fn wait_for_pid(rx: &mut watch::Receiver<Status>, want: u32) {
        let deadline = tokio::time::Instant::now() + Duration::from_secs(2);
        loop {
            if rx.borrow().pid == Some(want) {
                return;
            }
            tokio::select! {
                _ = rx.changed() => {}
                _ = tokio::time::sleep_until(deadline) => panic!("never saw pid {want}, last={:?}", rx.borrow().pid),
            }
        }
    }

    async fn wait_for(rx: &mut watch::Receiver<Status>, want: ServerState) {
        let deadline = tokio::time::Instant::now() + Duration::from_secs(2);
        loop {
            if rx.borrow().state == want {
                return;
            }
            tokio::select! {
                _ = rx.changed() => {}
                _ = tokio::time::sleep_until(deadline) => panic!("never reached {want:?}, last={:?}", rx.borrow().state),
            }
        }
    }

    #[tokio::test]
    async fn start_runs_to_running_and_stop_to_stopped() {
        let cfg = cfg("x", RestartPolicy::Never, 5);
        let (mock, mut ctl) = MockChild::new(1);
        let queue = Arc::new(Mutex::new(vec![mock]));
        let spawner = QueueSpawner { queue };

        let (status_tx, mut status_rx) = watch::channel(initial_status(&cfg));
        let (cmd_tx, cmd_rx) = mpsc::channel(8);
        let task = SupervisorTask::new(cfg, sink("x"), spawner, status_tx, cmd_rx);
        let h = tokio::spawn(task.run());

        let (ack_tx, ack_rx) = oneshot::channel();
        cmd_tx
            .send(SupervisorCmd::Start { ack: ack_tx })
            .await
            .unwrap();
        assert_eq!(ack_rx.await.unwrap(), StartAck::Started);
        wait_for(&mut status_rx, ServerState::Running).await;

        ctl.exit_tx.take().unwrap().send(Some(0)).unwrap();
        wait_for(&mut status_rx, ServerState::Stopped).await;

        let (ack_tx, ack_rx) = oneshot::channel();
        cmd_tx
            .send(SupervisorCmd::Shutdown { ack: ack_tx })
            .await
            .unwrap();
        ack_rx.await.unwrap();
        h.await.unwrap();
    }

    #[tokio::test]
    async fn first_start_is_not_counted_as_a_restart() {
        let cfg = cfg("x", RestartPolicy::Never, 5);
        let (mock, ctl) = MockChild::new(1);
        let queue = Arc::new(Mutex::new(vec![mock]));
        let spawner = QueueSpawner { queue };

        let (status_tx, mut status_rx) = watch::channel(initial_status(&cfg));
        let (cmd_tx, cmd_rx) = mpsc::channel(8);
        let task = SupervisorTask::new(cfg, sink("x"), spawner, status_tx, cmd_rx);
        let h = tokio::spawn(task.run());

        let (ack_tx, ack_rx) = oneshot::channel();
        cmd_tx
            .send(SupervisorCmd::Start { ack: ack_tx })
            .await
            .unwrap();
        assert_eq!(ack_rx.await.unwrap(), StartAck::Started);
        wait_for(&mut status_rx, ServerState::Running).await;

        assert_eq!(
            status_rx.borrow().restart_count,
            0,
            "a server started once has not restarted"
        );

        // do_stop's post-kill wait() is unbounded, and MockChild only reports an
        // exit when its controller does; dropping it closes the channel so
        // shutdown can finish.
        drop(ctl);

        let (ack_tx, ack_rx) = oneshot::channel();
        cmd_tx
            .send(SupervisorCmd::Shutdown { ack: ack_tx })
            .await
            .unwrap();
        ack_rx.await.unwrap();
        h.await.unwrap();
    }

    #[tokio::test]
    async fn an_automatic_restart_after_a_crash_is_counted() {
        let cfg = cfg("x", RestartPolicy::Always, 5);
        let (first, mut ctl) = MockChild::new(1);
        let (second, ctl2) = MockChild::new(2);
        let queue = Arc::new(Mutex::new(vec![first, second]));
        let spawner = QueueSpawner { queue };

        let (status_tx, mut status_rx) = watch::channel(initial_status(&cfg));
        let (cmd_tx, cmd_rx) = mpsc::channel(8);
        let task = SupervisorTask::new(cfg, sink("x"), spawner, status_tx, cmd_rx);
        let h = tokio::spawn(task.run());

        let (ack_tx, ack_rx) = oneshot::channel();
        cmd_tx
            .send(SupervisorCmd::Start { ack: ack_tx })
            .await
            .unwrap();
        assert_eq!(ack_rx.await.unwrap(), StartAck::Started);
        wait_for(&mut status_rx, ServerState::Running).await;

        ctl.exit_tx.take().unwrap().send(Some(1)).unwrap();

        wait_for_pid(&mut status_rx, 2).await;

        assert_eq!(
            status_rx.borrow().restart_count,
            1,
            "one crash-and-respawn is one restart"
        );

        drop(ctl2);

        let (ack_tx, ack_rx) = oneshot::channel();
        cmd_tx
            .send(SupervisorCmd::Shutdown { ack: ack_tx })
            .await
            .unwrap();
        ack_rx.await.unwrap();
        h.await.unwrap();
    }

    #[tokio::test]
    async fn a_met_condition_spawns_immediately() {
        let tmp = tempfile::tempdir().unwrap();
        let file = tmp.path().join("sock");

        std::fs::write(&file, b"").unwrap();

        let cfg = cfg_with_wait(
            "x",
            WaitForConfig {
                condition: WaitCondition::Path(file),
                timeout: Duration::from_millis(500),
                interval: Duration::from_millis(10),
            },
        );

        let (mock, ctl) = MockChild::new(1);
        let queue = Arc::new(Mutex::new(vec![mock]));
        let spawner = QueueSpawner { queue };

        let (status_tx, mut status_rx) = watch::channel(initial_status(&cfg));
        let (cmd_tx, cmd_rx) = mpsc::channel(8);
        let task = SupervisorTask::new(cfg, sink("x"), spawner, status_tx, cmd_rx);
        let h = tokio::spawn(task.run());

        let (ack_tx, ack_rx) = oneshot::channel();
        cmd_tx
            .send(SupervisorCmd::Start { ack: ack_tx })
            .await
            .unwrap();
        assert_eq!(ack_rx.await.unwrap(), StartAck::Started);
        wait_for(&mut status_rx, ServerState::Running).await;

        drop(ctl);

        let (ack_tx, ack_rx) = oneshot::channel();
        cmd_tx
            .send(SupervisorCmd::Shutdown { ack: ack_tx })
            .await
            .unwrap();
        ack_rx.await.unwrap();
        h.await.unwrap();
    }

    #[tokio::test]
    async fn a_timed_out_gate_fails_without_spawning() {
        let cfg = cfg_with_wait(
            "x",
            WaitForConfig {
                condition: WaitCondition::Path(PathBuf::from("/definitely/not/here.sock")),
                timeout: Duration::from_millis(80),
                interval: Duration::from_millis(10),
            },
        );

        // An empty queue: any spawn attempt panics, which is the assertion that
        // a timed-out gate never reaches the spawner.
        let queue = Arc::new(Mutex::new(Vec::new()));
        let spawner = QueueSpawner { queue };

        let (status_tx, mut status_rx) = watch::channel(initial_status(&cfg));
        let (cmd_tx, cmd_rx) = mpsc::channel(8);
        let task = SupervisorTask::new(cfg, sink("x"), spawner, status_tx, cmd_rx);
        let h = tokio::spawn(task.run());

        let (ack_tx, _ack_rx) = oneshot::channel();
        cmd_tx
            .send(SupervisorCmd::Start { ack: ack_tx })
            .await
            .unwrap();

        wait_for(&mut status_rx, ServerState::Waiting).await;
        wait_for(&mut status_rx, ServerState::Failed).await;

        let (ack_tx, ack_rx) = oneshot::channel();
        cmd_tx
            .send(SupervisorCmd::Shutdown { ack: ack_tx })
            .await
            .unwrap();
        ack_rx.await.unwrap();
        h.await.unwrap();
    }

    #[tokio::test]
    async fn repeated_gate_timeouts_do_not_pollute_the_retry_window() {
        let tmp = tempfile::tempdir().unwrap();
        let file = tmp.path().join("sock");

        let mut cfg = cfg("x", RestartPolicy::Always, 2);

        cfg.wait_for = Some(WaitForConfig {
            condition: WaitCondition::Path(file.clone()),
            timeout: Duration::from_millis(20),
            interval: Duration::from_millis(5),
        });

        let (first, mut ctl) = MockChild::new(1);
        let (second, ctl2) = MockChild::new(2);
        let queue = Arc::new(Mutex::new(vec![first, second]));
        let spawner = QueueSpawner { queue };

        let (status_tx, mut status_rx) = watch::channel(initial_status(&cfg));
        let (cmd_tx, cmd_rx) = mpsc::channel(8);
        let task = SupervisorTask::new(cfg, sink("x"), spawner, status_tx, cmd_rx);
        let h = tokio::spawn(task.run());

        let (ack_tx, _ack_rx) = oneshot::channel();
        cmd_tx
            .send(SupervisorCmd::Start { ack: ack_tx })
            .await
            .unwrap();
        wait_for(&mut status_rx, ServerState::Waiting).await;
        wait_for(&mut status_rx, ServerState::Failed).await;

        let (ack_tx, _ack_rx) = oneshot::channel();
        cmd_tx
            .send(SupervisorCmd::Start { ack: ack_tx })
            .await
            .unwrap();
        wait_for(&mut status_rx, ServerState::Waiting).await;
        wait_for(&mut status_rx, ServerState::Failed).await;

        std::fs::write(&file, b"").unwrap();

        let (ack_tx, ack_rx) = oneshot::channel();
        cmd_tx
            .send(SupervisorCmd::Start { ack: ack_tx })
            .await
            .unwrap();
        assert_eq!(ack_rx.await.unwrap(), StartAck::Started);
        wait_for(&mut status_rx, ServerState::Running).await;

        ctl.exit_tx.take().unwrap().send(Some(1)).unwrap();

        wait_for_pid(&mut status_rx, 2).await;

        assert_eq!(
            status_rx.borrow().state,
            ServerState::Running,
            "the crash after two gate timeouts must still be within the retry budget"
        );

        drop(ctl2);

        let (ack_tx, ack_rx) = oneshot::channel();
        cmd_tx
            .send(SupervisorCmd::Shutdown { ack: ack_tx })
            .await
            .unwrap();
        ack_rx.await.unwrap();
        h.await.unwrap();
    }

    #[tokio::test]
    async fn stop_during_a_wait_cancels_promptly() {
        let cfg = cfg_with_wait(
            "x",
            WaitForConfig {
                condition: WaitCondition::Path(PathBuf::from("/definitely/not/here.sock")),
                timeout: Duration::from_secs(30),
                interval: Duration::from_millis(10),
            },
        );

        let queue = Arc::new(Mutex::new(Vec::new()));
        let spawner = QueueSpawner { queue };

        let (status_tx, mut status_rx) = watch::channel(initial_status(&cfg));
        let (cmd_tx, cmd_rx) = mpsc::channel(8);
        let task = SupervisorTask::new(cfg, sink("x"), spawner, status_tx, cmd_rx);
        let h = tokio::spawn(task.run());

        let (ack_tx, _ack_rx) = oneshot::channel();
        cmd_tx
            .send(SupervisorCmd::Start { ack: ack_tx })
            .await
            .unwrap();
        wait_for(&mut status_rx, ServerState::Waiting).await;

        let (ack_tx, ack_rx) = oneshot::channel();
        cmd_tx
            .send(SupervisorCmd::Stop { ack: ack_tx })
            .await
            .unwrap();

        // The 30s timeout would swamp this if the poll loop were not interruptible.
        tokio::time::timeout(Duration::from_secs(2), ack_rx)
            .await
            .expect("stop must not wait for the gate timeout")
            .unwrap();

        wait_for(&mut status_rx, ServerState::Stopped).await;

        let (ack_tx, ack_rx) = oneshot::channel();
        cmd_tx
            .send(SupervisorCmd::Shutdown { ack: ack_tx })
            .await
            .unwrap();
        ack_rx.await.unwrap();
        h.await.unwrap();
    }

    #[tokio::test]
    async fn shutdown_during_a_wait_ends_the_task_promptly() {
        let cfg = cfg_with_wait(
            "x",
            WaitForConfig {
                condition: WaitCondition::Path(PathBuf::from("/definitely/not/here.sock")),
                timeout: Duration::from_secs(30),
                interval: Duration::from_millis(10),
            },
        );

        let queue = Arc::new(Mutex::new(Vec::new()));
        let spawner = QueueSpawner { queue };

        let (status_tx, mut status_rx) = watch::channel(initial_status(&cfg));
        let (cmd_tx, cmd_rx) = mpsc::channel(8);
        let task = SupervisorTask::new(cfg, sink("x"), spawner, status_tx, cmd_rx);
        let h = tokio::spawn(task.run());

        let (ack_tx, _ack_rx) = oneshot::channel();
        cmd_tx
            .send(SupervisorCmd::Start { ack: ack_tx })
            .await
            .unwrap();

        wait_for(&mut status_rx, ServerState::Waiting).await;

        let (ack_tx, ack_rx) = oneshot::channel();
        cmd_tx
            .send(SupervisorCmd::Shutdown { ack: ack_tx })
            .await
            .unwrap();

        // The 30s gate timeout would swamp both of these if Shutdown were only
        // observed once the gate resolved.
        tokio::time::timeout(Duration::from_millis(2_000), ack_rx)
            .await
            .expect("shutdown must not wait for the gate timeout")
            .unwrap();

        tokio::time::timeout(Duration::from_millis(2_000), h)
            .await
            .expect("the supervisor task must return once shut down")
            .unwrap();
    }

    #[tokio::test]
    async fn a_server_waiting_on_its_gate_reports_no_uptime() {
        let tmp = tempfile::tempdir().unwrap();
        let file = tmp.path().join("sock");

        std::fs::write(&file, b"").unwrap();

        let mut cfg = cfg("x", RestartPolicy::Always, 5);

        cfg.wait_for = Some(WaitForConfig {
            condition: WaitCondition::Path(file.clone()),
            timeout: Duration::from_millis(2_000),
            interval: Duration::from_millis(10),
        });

        // Only one child: the gate must never clear again, so a second spawn
        // would panic on the empty queue.
        let (first, mut ctl) = MockChild::new(1);
        let queue = Arc::new(Mutex::new(vec![first]));
        let spawner = QueueSpawner { queue };

        let (status_tx, mut status_rx) = watch::channel(initial_status(&cfg));
        let (cmd_tx, cmd_rx) = mpsc::channel(8);
        let task = SupervisorTask::new(cfg, sink("x"), spawner, status_tx, cmd_rx);
        let h = tokio::spawn(task.run());

        let (ack_tx, ack_rx) = oneshot::channel();
        cmd_tx
            .send(SupervisorCmd::Start { ack: ack_tx })
            .await
            .unwrap();
        assert_eq!(ack_rx.await.unwrap(), StartAck::Started);
        wait_for(&mut status_rx, ServerState::Running).await;
        assert!(status_rx.borrow().started_at.is_some());

        std::fs::remove_file(&file).unwrap();

        ctl.exit_tx.take().unwrap().send(Some(1)).unwrap();

        wait_for(&mut status_rx, ServerState::Waiting).await;

        assert!(
            status_rx.borrow().started_at.is_none(),
            "a server waiting on its gate has no live process, so it has no uptime"
        );

        let (ack_tx, ack_rx) = oneshot::channel();
        cmd_tx
            .send(SupervisorCmd::Shutdown { ack: ack_tx })
            .await
            .unwrap();
        ack_rx.await.unwrap();
        h.await.unwrap();
    }
}