feat(supervisor): gate spawns on the wait-for condition

crates/xy-supervisor/src/supervisor.rs

@@ -43,6 +43,13 @@ pub enum StopAck {
     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,
@@ -51,6 +58,7 @@ pub struct Status {
     pub started_at: Option<Instant>,
     pub restart_count: u32,
     pub last_exit: Option<i32>,
+    pub waiting: Option<WaitProgress>,
 }
 
 #[derive(Clone)]
@@ -79,6 +87,7 @@ pub struct SupervisorTask<S: Spawner> {
     last_exit: Option<i32>,
     started_at: Option<Instant>,
     current_pid: Option<u32>,
+    waiting: Option<WaitProgress>,
 }
 
 impl<S: Spawner> SupervisorTask<S> {
@@ -105,6 +114,7 @@ impl<S: Spawner> SupervisorTask<S> {
             last_exit: None,
             started_at: None,
             current_pid: None,
+            waiting: None,
         }
     }
 
@@ -116,6 +126,7 @@ impl<S: Spawner> SupervisorTask<S> {
             started_at: self.started_at,
             restart_count: self.restart_count,
             last_exit: self.last_exit,
+            waiting: self.waiting.clone(),
         });
     }
 
@@ -137,11 +148,22 @@ impl<S: Spawner> SupervisorTask<S> {
                                         child = Some(c);
                                         let _ = ack.send(StartAck::Started);
                                     }
-                                    Err(err) => {
+                                    Err(StartFailure::Spawn(err)) => {
                                         warn!(name = %self.cfg.name, error = %err, "spawn failed");
                                         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.set_state(ServerState::Failed);
+                                        let _ = ack.send(StartAck::SpawnFailed(
+                                            "wait-for timed out".to_string(),
+                                        ));
+                                    }
+                                    Err(StartFailure::Cancelled) => {
+                                        self.set_state(ServerState::Stopped);
+                                    }
+                                    Err(StartFailure::Shutdown) => return,
                                 }
                             }
                         }
@@ -161,10 +183,18 @@ impl<S: Spawner> SupervisorTask<S> {
 
                             match self.do_start(StartCause::Restart).await {
                                 Ok(c) => child = Some(c),
-                                Err(err) => {
+                                Err(StartFailure::Spawn(err)) => {
                                     warn!(name = %self.cfg.name, error = %err, "restart spawn failed");
                                     self.set_state(ServerState::Failed);
                                 }
+                                Err(StartFailure::WaitTimedOut) => {
+                                    warn!(name = %self.cfg.name, "wait-for timed out");
+                                    self.set_state(ServerState::Failed);
+                                }
+                                Err(StartFailure::Cancelled) => {
+                                    self.set_state(ServerState::Stopped);
+                                }
+                                Err(StartFailure::Shutdown) => return,
                             }
 
                             let _ = ack.send(());
@@ -267,10 +297,18 @@ impl<S: Spawner> SupervisorTask<S> {
                                 Action::RetryNow => {
                                     match self.do_start(StartCause::Restart).await {
                                         Ok(c) => child = Some(c),
-                                        Err(err) => {
+                                        Err(StartFailure::Spawn(err)) => {
                                             warn!(name = %self.cfg.name, error = %err, "restart spawn failed");
                                             self.set_state(ServerState::Failed);
                                         }
+                                        Err(StartFailure::WaitTimedOut) => {
+                                            warn!(name = %self.cfg.name, "wait-for timed out");
+                                            self.set_state(ServerState::Failed);
+                                        }
+                                        Err(StartFailure::Cancelled) => {
+                                            self.set_state(ServerState::Stopped);
+                                        }
+                                        Err(StartFailure::Shutdown) => return,
                                     }
                                 }
                                 Action::Cancel => {
@@ -286,10 +324,85 @@ impl<S: Spawner> SupervisorTask<S> {
         }
     }
 
-    async fn do_start(&mut self, cause: StartCause) -> std::io::Result<S::Child> {
+    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.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;
+                        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?;
+        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);
@@ -339,6 +452,20 @@ enum StartCause {
     Restart,
 }
 
+enum StartFailure {
+    Spawn(std::io::Error),
+    WaitTimedOut,
+    Cancelled,
+    Shutdown,
+}
+
+enum GateOutcome {
+    Ready,
+    TimedOut,
+    Cancelled,
+    Shutdown,
+}
+
 pub struct RealSpawner;
 
 #[async_trait::async_trait]
@@ -355,9 +482,10 @@ 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};
+    use xy_protocol::{RestartConfig, RestartPolicy, StopConfig, WaitCondition, WaitForConfig};
 
     struct QueueSpawner {
         queue: Arc<Mutex<Vec<MockChild>>>,
@@ -394,6 +522,12 @@ mod tests {
         }
     }
 
+    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();
@@ -409,6 +543,7 @@ mod tests {
             started_at: None,
             restart_count: 0,
             last_exit: None,
+            waiting: None,
         }
     }
 
@@ -551,4 +686,142 @@ mod tests {
         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_or_touching_the_retry_budget() {
+        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;
+
+        assert_eq!(
+            status_rx.borrow().restart_count,
+            0,
+            "waiting must not consume the restart budget"
+        );
+
+        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();
+    }
 }

crates/xy/src/daemon/mod.rs

@@ -46,6 +46,7 @@ pub fn spawn_supervisor(paths: &Paths, cfg: ServerConfig) -> Result<SupervisorHa
         started_at: None,
         restart_count: 0,
         last_exit: None,
+        waiting: None,
     };
 
     let (status_tx, status_rx) = watch::channel(initial_status);