From 9307af46a8af6a123922e2787a12bf64bb0bdfaa Mon Sep 17 00:00:00 2001 From: Anders Olsson Date: Fri, 7 Aug 2026 15:46:38 +0200 Subject: [PATCH] feat(supervisor): gate spawns on the wait-for condition --- crates/xy-supervisor/src/supervisor.rs | 285 ++++++++++++++++++++++++- crates/xy/src/daemon/mod.rs | 1 + 2 files changed, 280 insertions(+), 6 deletions(-) diff --git a/crates/xy-supervisor/src/supervisor.rs b/crates/xy-supervisor/src/supervisor.rs index 094e6c0..0518dd0 100644 --- a/crates/xy-supervisor/src/supervisor.rs +++ b/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, pub restart_count: u32, pub last_exit: Option, + pub waiting: Option, } #[derive(Clone)] @@ -79,6 +87,7 @@ pub struct SupervisorTask { last_exit: Option, started_at: Option, current_pid: Option, + waiting: Option, } impl SupervisorTask { @@ -105,6 +114,7 @@ impl SupervisorTask { last_exit: None, started_at: None, current_pid: None, + waiting: None, } } @@ -116,6 +126,7 @@ impl SupervisorTask { started_at: self.started_at, restart_count: self.restart_count, last_exit: self.last_exit, + waiting: self.waiting.clone(), }); } @@ -137,11 +148,22 @@ impl SupervisorTask { 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 SupervisorTask { 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 SupervisorTask { 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 SupervisorTask { } } - async fn do_start(&mut self, cause: StartCause) -> std::io::Result { + 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 { + 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>>, @@ -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(); + } } diff --git a/crates/xy/src/daemon/mod.rs b/crates/xy/src/daemon/mod.rs index f243fa7..1fe727b 100644 --- a/crates/xy/src/daemon/mod.rs +++ b/crates/xy/src/daemon/mod.rs @@ -46,6 +46,7 @@ pub fn spawn_supervisor(paths: &Paths, cfg: ServerConfig) -> Result