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) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01EGntTHCW3sEPy1VBRopNNp
This commit is contained in:
@@ -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 {
|
||||
|
||||
@@ -346,6 +346,8 @@ impl<S: Spawner> SupervisorTask<S> {
|
||||
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();
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user