diff --git a/src/core/actor.rs b/src/core/actor.rs index 0be8adb..afe63df 100644 --- a/src/core/actor.rs +++ b/src/core/actor.rs @@ -382,7 +382,7 @@ async fn run_final_command_for_tui_inner( state: &Arc>, cancel_rx: oneshot::Receiver<()>, ) -> anyhow::Result<()> { - let mut final_cmd = match crate::core::command::spawn_captured_group(command) { + let mut final_cmd = match crate::core::command::spawn_group(command, false) { Ok(final_cmd) => final_cmd, Err(error) => { mark_final_failed_with_message( diff --git a/src/core/command.rs b/src/core/command.rs index 91087d2..5055efa 100644 --- a/src/core/command.rs +++ b/src/core/command.rs @@ -1,5 +1,4 @@ use command_group::{AsyncCommandGroup, AsyncGroupChild}; -use tokio::process::Child; use tokio::task::JoinHandle; use std::sync::{Arc, Mutex}; @@ -9,14 +8,9 @@ use crate::core::server::{LOG_CAPACITY, build_command}; use crate::core::state::RingBuffer; /// A spawned final command plus its captured output log. -pub struct FinalCommand { - pub child: Child, - #[allow(dead_code)] // read by AppState in Task 9 - pub log: Arc>, - pub readers: Vec>, -} - -/// A captured final command spawned as a process group for TUI cancellation. +/// +/// Always a process group, so the command and every descendant it spawned can +/// be killed together on shutdown. pub struct FinalCommandGroup { child: AsyncGroupChild, pub log: Arc>, @@ -28,60 +22,30 @@ impl FinalCommandGroup { self.child.wait().await } + /// Kill the whole process group, descendants included. pub async fn cancel(&mut self) -> std::io::Result<()> { self.child.kill().await } } -/// Spawn the final command (NOT as a process group — matches today's `Command::spawn`). -pub fn spawn(command: &str) -> anyhow::Result { - spawn_inner(command, true) -} - -#[allow(dead_code)] // used by TUI command execution in Task 7 -pub fn spawn_captured(command: &str) -> anyhow::Result { - spawn_inner(command, false) -} - -#[allow(dead_code)] // used by TUI command execution -pub fn spawn_captured_group(command: &str) -> anyhow::Result { +/// Spawn the final command as a process group. +/// +/// `tee_output` mirrors the child's output to the real stdout/stderr for plain +/// mode; the TUI captures only and renders the log itself. +pub fn spawn_group(command: &str, tee_output: bool) -> anyhow::Result { let mut cmd = build_command(command)?; let mut child = cmd.group_spawn()?; let log = Arc::new(Mutex::new(RingBuffer::new(LOG_CAPACITY))); let mut readers = Vec::new(); if let Some(stdout) = child.inner().stdout.take() { - readers.push(spawn_reader(stdout, Arc::clone(&log), false, false)); - } - if let Some(stderr) = child.inner().stderr.take() { - readers.push(spawn_reader(stderr, Arc::clone(&log), true, false)); - } - - Ok(FinalCommandGroup { - child, - log, - readers, - }) -} - -fn spawn_inner(command: &str, tee_output: bool) -> anyhow::Result { - let mut cmd = build_command(command)?; - if !tee_output { - cmd.kill_on_drop(true); - } - - let mut child = cmd.spawn()?; - let log = Arc::new(Mutex::new(RingBuffer::new(LOG_CAPACITY))); - let mut readers = Vec::new(); - - if let Some(stdout) = child.stdout.take() { readers.push(spawn_reader(stdout, Arc::clone(&log), false, tee_output)); } - if let Some(stderr) = child.stderr.take() { + if let Some(stderr) = child.inner().stderr.take() { readers.push(spawn_reader(stderr, Arc::clone(&log), true, tee_output)); } - Ok(FinalCommand { + Ok(FinalCommandGroup { child, log, readers, @@ -94,8 +58,8 @@ mod tests { #[tokio::test] async fn spawn_captured_records_output_lines() { - let mut command = spawn_captured("sh -c 'echo out; echo err >&2'").unwrap(); - let status = command.child.wait().await.unwrap(); + let mut command = spawn_group("sh -c 'echo out; echo err >&2'", false).unwrap(); + let status = command.wait().await.unwrap(); for reader in command.readers { let _ = reader.await; } diff --git a/src/runner/plain.rs b/src/runner/plain.rs index 5985464..690fa64 100644 --- a/src/runner/plain.rs +++ b/src/runner/plain.rs @@ -1,68 +1,117 @@ use anyhow::Context; +use tokio::sync::watch; use std::time::Duration; -use crate::config::Config; +use crate::config::{Config, Server}; use crate::core::Engine; use crate::core::state::ServerStatus; /// Drive the engine in plain-log mode. -/// Exit semantics match the legacy tool exactly. pub async fn run(config: Config, max_attempts: u8) -> anyhow::Result<()> { + // Registered before the servers start, so a Ctrl+C arriving during startup + // is already observable. + let mut shutdown = shutdown_signal(); + let Config { servers, command } = config; let mut engine = Engine::start(&servers, max_attempts)?; + let mut ready = vec![false; servers.len()]; - let final_result = loop { - let mut ready = true; - - for server in &servers { - match engine.probe(server).await { - Ok(ServerStatus::Running) => {} - Ok(_) => ready = false, - Err(e) => { - engine.stop_all().await?; - return Err(e); - } - } - } + loop { + // `None` means Ctrl+C won the race and the probe round was cancelled. + let round = tokio::select! { + result = probe_pending(&mut engine, &servers, &mut ready) => Some(result), + _ = shutdown.changed() => None, + }; - if ready { - break tokio::select! { - result = run_final_command(&command) => result, - _ = tokio::signal::ctrl_c() => { - if let Err(e) = engine.stop_all().await { - eprintln!("Error stopping servers: {}", e); - std::process::exit(1); - } - std::process::exit(0); - } - }; + match round { + None => return stop_for_shutdown(&mut engine).await, + Some(Err(error)) => { + engine.stop_all().await?; + return Err(error); + } + Some(Ok(true)) => { + let final_result = run_final_command(&command, &mut shutdown).await; + engine.stop_all().await?; + return final_result; + } + Some(Ok(false)) => {} } tokio::select! { _ = tokio::time::sleep(Duration::from_secs(1)) => {} - _ = tokio::signal::ctrl_c() => { - if let Err(e) = engine.stop_all().await { - eprintln!("Error stopping servers: {}", e); - std::process::exit(1); - } - std::process::exit(0); - } + _ = shutdown.changed() => return stop_for_shutdown(&mut engine).await, } - }; + } +} + +/// Watch channel that flips to `true` on the first Ctrl+C. +/// +/// A dedicated task owns one long-lived `ctrl_c()` future. Building a fresh one +/// per `select!` iteration would drop any signal delivered while that future was +/// not being polled — during a health probe, for instance — and because tokio +/// installs a handler the default terminate action is gone too, so the signal +/// would be lost entirely rather than killing the process. +fn shutdown_signal() -> watch::Receiver { + let (tx, rx) = watch::channel(false); + tokio::spawn(async move { + if tokio::signal::ctrl_c().await.is_ok() { + let _ = tx.send(true); + } + }); + rx +} + +/// Probe every server that is not ready yet and report whether all are up. +/// +/// Servers already known to be ready are skipped: re-probing them keeps +/// incrementing their attempt counter, so a healthy server could exhaust its +/// attempts — and get blamed in the error — while a different, slower server +/// was the one actually holding up the run. +async fn probe_pending( + engine: &mut Engine, + servers: &[Server], + ready: &mut [bool], +) -> anyhow::Result { + for (index, server) in servers.iter().enumerate() { + if ready[index] { + continue; + } + + if engine.probe(server).await? == ServerStatus::Running { + ready[index] = true; + } + } - engine.stop_all().await?; - final_result + Ok(ready.iter().all(|ready| *ready)) } -async fn run_final_command(command: &str) -> anyhow::Result<()> { - let mut final_cmd = crate::core::command::spawn(command) +/// Ctrl+C path: stop every server, then exit successfully. +async fn stop_for_shutdown(engine: &mut Engine) -> anyhow::Result<()> { + engine.stop_all().await.context("Error stopping servers") +} + +async fn run_final_command( + command: &str, + shutdown: &mut watch::Receiver, +) -> anyhow::Result<()> { + let mut final_cmd = crate::core::command::spawn_group(command, true) .context(format!("Could not start process {}", command))?; log::info!("Running command {}", command); - // Wait for the process to finish. - let status = final_cmd.child.wait().await?; + let status = tokio::select! { + status = final_cmd.wait() => status?, + _ = shutdown.changed() => { + // Kill the command and everything it spawned. Exiting the process + // here instead would leave the whole group orphaned. + let _ = final_cmd.cancel().await; + for reader in final_cmd.readers { + let _ = reader.await; + } + return Ok(()); + } + }; for reader in final_cmd.readers { let _ = reader.await; diff --git a/tests/cli.rs b/tests/cli.rs index 7de6b19..716f0b7 100644 --- a/tests/cli.rs +++ b/tests/cli.rs @@ -400,6 +400,193 @@ fn rejects_unreasonably_large_timeout() { )); } +#[cfg(unix)] +#[test] +fn blames_the_server_that_never_became_ready() { + use std::fs; + + let suffix = std::process::id(); + let config = format!("/tmp/server-runner-blame-{suffix}.yaml"); + let _cleanup = RemoveFileOnDrop(config.clone()); + let port = 8128; + + fs::write( + &config, + format!( + "servers:\n - name: \"Ready Server\"\n url: \"http://127.0.0.1:{port}\"\n command: \"python3 -m http.server {port} --bind 127.0.0.1\"\n timeout: 1\n - name: \"Never Ready\"\n url: \"http://127.0.0.1:9996\"\n command: \"sleep 30\"\n timeout: 1\ncommand: \"echo done\"\n" + ), + ) + .unwrap(); + + // A ready server must not keep burning attempts. Otherwise it exhausts them + // first and gets blamed for a run that a different server was holding up. + let mut command = Command::cargo_bin("server-runner").unwrap(); + + command + .arg("-c") + .arg(&config) + .arg("-a") + .arg("8") + .assert() + .failure() + .stderr(predicate::str::contains( + "Could not connect to server Never Ready", + )) + .stderr(predicate::str::contains("Ready Server after").not()); + + assert_port_released(&format!("127.0.0.1:{port}")); +} + +#[cfg(unix)] +#[test] +fn stops_final_command_descendants_on_ctrl_c() { + use std::fs; + + let suffix = std::process::id(); + let marker = format!("/tmp/server-runner-final-descendant-{suffix}"); + let config = format!("/tmp/server-runner-final-descendant-{suffix}.yaml"); + let script = format!("/tmp/server-runner-final-descendant-{suffix}.sh"); + let _ = fs::remove_file(&marker); + let _cleanup_config = RemoveFileOnDrop(config.clone()); + let _cleanup_script = RemoveFileOnDrop(script.clone()); + let _cleanup_marker = RemoveFileOnDrop(marker.clone()); + let port = 8129; + let addr = format!("127.0.0.1:{port}"); + + // The descendant outlives the test by far, so if Ctrl+C fails to kill the + // group it is still running when we check rather than having exited on its + // own. The runner also blocks on its output readers until the group dies, + // so an uncancelled group shows up as a runner that never exits. + fs::write( + &script, + format!("#!/bin/sh\nsleep 300 /dev/null 2>&1 &\necho $! > {marker}\nwait\n"), + ) + .unwrap(); + + fs::write( + &config, + format!( + "servers:\n - name: \"Output Server\"\n url: \"http://127.0.0.1:{port}\"\n command: \"python3 -m http.server {port} --bind 127.0.0.1\"\n timeout: 1\ncommand: \"sh {script}\"\n" + ), + ) + .unwrap(); + + let mut child = std::process::Command::new(assert_cmd::cargo::cargo_bin("server-runner")) + .arg("-c") + .arg(&config) + .arg("-a") + .arg("20") + .spawn() + .unwrap(); + + assert_port_opens(&addr); + let pid = wait_for_marker(&marker); + + let interrupt = std::process::Command::new("kill") + .arg("-INT") + .arg(child.id().to_string()) + .status() + .unwrap(); + assert!(interrupt.success()); + + let status = wait_with_timeout(&mut child, Duration::from_secs(10)) + .expect("runner did not exit promptly after Ctrl+C"); + assert!(status.success()); + + thread::sleep(Duration::from_millis(250)); + let orphaned = pid_alive(&pid); + if orphaned { + let _ = std::process::Command::new("kill") + .arg("-9") + .arg(&pid) + .status(); + } + + assert!(!orphaned, "final command descendant {pid} survived Ctrl+C"); + assert_port_released(&addr); +} + +#[cfg(unix)] +#[test] +fn honours_ctrl_c_while_a_probe_is_in_flight() { + let addr = "127.0.0.1:8130"; + + let mut child = std::process::Command::new(assert_cmd::cargo::cargo_bin("server-runner")) + .arg("-c") + .arg("tests/hanging_url.yaml") + .arg("-a") + .arg("20") + .spawn() + .unwrap(); + + // The port accepts but never answers, so the runner's second probe blocks + // for the full 10s timeout. Wait past the first one-second retry gap so the + // signal lands squarely inside that probe rather than in the gap. + assert_port_opens(addr); + thread::sleep(Duration::from_secs(3)); + + let interrupt = std::process::Command::new("kill") + .arg("-INT") + .arg(child.id().to_string()) + .status() + .unwrap(); + assert!(interrupt.success()); + + let status = wait_with_timeout(&mut child, Duration::from_secs(5)) + .expect("runner ignored Ctrl+C delivered during an in-flight probe"); + assert!(status.success()); + + assert_port_released(addr); +} + +#[cfg(unix)] +fn wait_for_marker(path: &str) -> String { + for _ in 0..100 { + if let Ok(contents) = std::fs::read_to_string(path) { + let pid = contents.trim().to_string(); + if !pid.is_empty() { + return pid; + } + } + + thread::sleep(Duration::from_millis(50)); + } + + panic!("final command never recorded its descendant pid in {path}"); +} + +#[cfg(unix)] +fn pid_alive(pid: &str) -> bool { + std::process::Command::new("kill") + .arg("-0") + .arg(pid) + .stdout(std::process::Stdio::null()) + .stderr(std::process::Stdio::null()) + .status() + .map(|status| status.success()) + .unwrap_or(false) +} + +#[cfg(unix)] +fn wait_with_timeout( + child: &mut std::process::Child, + timeout: Duration, +) -> Option { + let deadline = std::time::Instant::now() + timeout; + + while std::time::Instant::now() < deadline { + match child.try_wait() { + Ok(Some(status)) => return Some(status), + Ok(None) => thread::sleep(Duration::from_millis(50)), + Err(_) => return None, + } + } + + let _ = child.kill(); + let _ = child.wait(); + None +} + fn assert_port_released(addr: &str) { if TcpListener::bind(addr).is_ok() { return; diff --git a/tests/hanging_server.py b/tests/hanging_server.py new file mode 100644 index 0000000..a6a7497 --- /dev/null +++ b/tests/hanging_server.py @@ -0,0 +1,14 @@ +import socket + +# Accepts connections but never answers, so a readiness probe against this +# port blocks for the full configured timeout. Used to check that Ctrl+C is +# still honoured while a probe is in flight. +server = socket.socket() +server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) +server.bind(("127.0.0.1", 8130)) +server.listen(5) + +accepted = [] +while True: + connection, _ = server.accept() + accepted.append(connection) diff --git a/tests/hanging_url.yaml b/tests/hanging_url.yaml new file mode 100644 index 0000000..bb78420 --- /dev/null +++ b/tests/hanging_url.yaml @@ -0,0 +1,6 @@ +servers: + - name: "Hanging Server" + url: "http://127.0.0.1:8130" + command: "python3 tests/hanging_server.py" + timeout: 10 +command: "echo done"