From 516d7d0b86feb42ee466b957f1d84de1836cdd93 Mon Sep 17 00:00:00 2001 From: Anders Olsson Date: Fri, 7 Aug 2026 16:17:20 +0200 Subject: [PATCH] fix(daemon): keep serving while a gated server waits The autostart loop awaited each Start ack, and a gated start does not ack until its wait-for condition resolves. Because the accept loop and the signal handler are installed after that loop, a server gated on a socket that is not there yet left the daemon deaf for the whole gate: `xy list` hung in the backlog and SIGTERM went unhandled. Nothing consumed the ack, so drop the await and let the gates run concurrently with the accept loop. Also clear started_at before publishing Waiting. A crash-and-regate kept the dead process's start instant, so `xy list` showed a growing uptime for a server that had been down for minutes -- in the one state whose whole purpose is to explain what is going on. And log the spawn error when a wait-for command cannot be run at all, so a misspelled binary is distinguishable from a condition that is merely not met yet. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01EGntTHCW3sEPy1VBRopNNp --- crates/xy-supervisor/src/ready.rs | 22 +++++- crates/xy-supervisor/src/supervisor.rs | 103 +++++++++++++++++++++++++ crates/xy/src/daemon/mod.rs | 14 ++-- 3 files changed, 129 insertions(+), 10 deletions(-) diff --git a/crates/xy-supervisor/src/ready.rs b/crates/xy-supervisor/src/ready.rs index d7e8c01..ee94deb 100644 --- a/crates/xy-supervisor/src/ready.rs +++ b/crates/xy-supervisor/src/ready.rs @@ -1,5 +1,6 @@ use std::process::Stdio; use std::time::Duration; +use tracing::debug; use xy_protocol::WaitCondition; pub(crate) async fn is_ready(condition: &WaitCondition, budget: Duration) -> bool { @@ -21,7 +22,16 @@ pub(crate) async fn is_ready(condition: &WaitCondition, budget: Duration) -> boo match tokio::time::timeout(budget, cmd.status()).await { Ok(Ok(status)) => status.success(), - _ => false, + Ok(Err(err)) => { + debug!( + command = %command.display(), + error = %err, + "wait-for command could not be run", + ); + + false + } + Err(_) => false, } } } @@ -108,6 +118,16 @@ mod tests { assert!(!is_ready(&condition, ms(500)).await); } + #[tokio::test] + async fn a_command_that_cannot_be_run_is_not_ready() { + let condition = WaitCondition::Command { + command: PathBuf::from("/definitely/not/here/docker"), + args: vec![], + }; + + assert!(!is_ready(&condition, ms(500)).await); + } + #[tokio::test] async fn a_command_that_outlasts_its_budget_is_not_ready() { let condition = WaitCondition::Command { diff --git a/crates/xy-supervisor/src/supervisor.rs b/crates/xy-supervisor/src/supervisor.rs index 227934d..c05c456 100644 --- a/crates/xy-supervisor/src/supervisor.rs +++ b/crates/xy-supervisor/src/supervisor.rs @@ -346,6 +346,8 @@ impl SupervisorTask { timeout: wait.timeout, }); + self.started_at = None; + self.set_state(ServerState::Waiting); let outcome = loop { @@ -903,4 +905,105 @@ mod tests { 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(); + } } diff --git a/crates/xy/src/daemon/mod.rs b/crates/xy/src/daemon/mod.rs index 1fe727b..99260ed 100644 --- a/crates/xy/src/daemon/mod.rs +++ b/crates/xy/src/daemon/mod.rs @@ -97,16 +97,12 @@ pub async fn run(paths: Paths) -> Result<()> { ) .await; - let (ack_tx, ack_rx) = oneshot::channel(); + let (ack_tx, _ack_rx) = oneshot::channel(); - if handle - .tx - .send(SupervisorCmd::Start { ack: ack_tx }) - .await - .is_ok() - { - let _ = ack_rx.await; - } + // Deliberately not awaiting the ack: a gated start blocks until its + // wait-for condition resolves, and awaiting here would keep the accept + // loop and the signal handler below from ever being installed. + let _ = handle.tx.send(SupervisorCmd::Start { ack: ack_tx }).await; } let registry_for_shutdown = registry.clone();