diff --git a/Cargo.lock b/Cargo.lock index 8defae27..008a10f1 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -124,9 +124,9 @@ checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801" [[package]] name = "chacha20" -version = "0.10.1" +version = "0.10.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d524456ba66e72eb8b115ff89e01e497f8e6d11d78b70b1aa13c0fbd97540a81" +checksum = "65c35e4b699c7e15ccbe7ee35c005e4fc0a278d22238a2857e6ce2dadeda1b06" dependencies = [ "cfg-if", "cpufeatures", diff --git a/Cargo.toml b/Cargo.toml index c11b7448..8f81dce6 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -43,12 +43,16 @@ smallvec = "=2.0.0-alpha.12" mimalloc = { version = "0.1", default-features = false, features = [ "secure", "v3", -] } +], optional = true } [target.'cfg(unix)'.dependencies] libc = "0.2" [features] +default = ["mimalloc"] +# Default-on preserves the shipped binary's allocator; --no-default-features +# falls back to the system allocator (profiling, allocator-sensitive tooling). +mimalloc = ["dep:mimalloc"] # test_helpers exposes advance_test_clock(), which calls tokio::time::advance — # only available with tokio's test-util. Pull it here so the test-internals # surface compiles standalone (e.g. `cargo build --all-features`), not just under diff --git a/crates/network-sim/src/harness.rs b/crates/network-sim/src/harness.rs index d86360c2..b326b4a6 100644 --- a/crates/network-sim/src/harness.rs +++ b/crates/network-sim/src/harness.rs @@ -173,6 +173,19 @@ pub fn check_impairment_deps() -> std::result::Result<(), SkipReason> { // NamespaceProcess // --------------------------------------------------------------------------- +/// What teardown is allowed to signal. +/// +/// Deliberately *not* "the tracked child's process group". `spawn_with_env` +/// runs a plain `Command::spawn()`, so the `sudo ip netns exec` wrapper +/// inherits the harness's process group and its PID is therefore **not** a +/// PGID. `kill -- -` fails `ESRCH` against it. The namespace's own PID +/// list is authoritative and needs no such assumption. +enum ProcessScope { + Namespace(String), + #[cfg(all(test, target_os = "linux"))] + Pids(Vec), +} + /// A child process running inside a network namespace. /// /// Captures stdout+stderr and kills the process on drop. @@ -190,8 +203,11 @@ pub fn check_impairment_deps() -> std::result::Result<(), SkipReason> { /// finish under 64 KiB. pub struct NamespaceProcess { child: Child, - #[expect(dead_code)] label: String, + scope: ProcessScope, + /// Set once `child` has been reaped, so a repeated `kill()` (explicit + /// `stop()` followed by `Drop`) does not wait on it again. + reaped: bool, stdout: Arc>>, stderr: Arc>>, } @@ -252,6 +268,8 @@ impl NamespaceProcess { Ok(Self { child, label, + scope: ProcessScope::Namespace(ns.name.clone()), + reaped: false, stdout, stderr, }) @@ -269,36 +287,31 @@ impl NamespaceProcess { self.stderr.lock().map(|b| b.clone()).unwrap_or_default() } - /// Send SIGTERM, wait briefly, then SIGKILL if needed. + /// Send SIGTERM to the namespace's PIDs, then SIGKILL if needed. /// - /// Signals the entire process group (negative PID) so the inner process - /// receives the signal even when wrapped by `sudo ip netns exec`. + /// Every wait is bounded, and repeated calls are harmless (`stop()` kills + /// explicitly and `Drop` kills again). + /// + /// This signals **every** process in the namespace, not just this one: + /// the inner process cannot be reached through the `sudo` wrapper's PID + /// (see [`ProcessScope`]). That is fine for whole-stack teardown, which is + /// the only caller today; do not kill or drop one process mid-test while + /// another in the same namespace must keep running. pub fn kill(&mut self) { - if let Some(pid) = self.pid() { - // Signal the entire process group so the inner process receives it - let _ = Command::new("sudo") - .args(["kill", "-TERM", "--", &format!("-{pid}")]) - .output(); + if self.reaped && self.scope_pids().is_empty() { + return; } - match self.child.try_wait().ok().flatten() { - Some(_) => return, - None => { - // Wait up to 2s for graceful exit - std::thread::sleep(Duration::from_secs(2)); - if self.child.try_wait().ok().flatten().is_some() { - return; - } - } + self.signal_scope("-TERM"); + if self.wait_until_stopped(Duration::from_secs(2)) { + return; } - // Force kill the process group - if let Some(pid) = self.pid() { - let _ = Command::new("sudo") - .args(["kill", "-9", "--", &format!("-{pid}")]) - .output(); + self.signal_scope("-KILL"); + let _ = self.child.kill(); + if !self.wait_until_stopped(Duration::from_secs(2)) { + tracing::warn!(label = %self.label, "namespace process teardown exceeded grace period"); } - let _ = self.child.wait(); } /// Check if the process is still running. @@ -318,8 +331,75 @@ impl NamespaceProcess { } } - fn pid(&self) -> Option { - Some(self.child.id()) + fn signal_scope(&self, signal: &str) { + let pids = self.scope_pids(); + if pids.is_empty() { + return; + } + + let mut command = match &self.scope { + ProcessScope::Namespace(_) => { + let mut command = Command::new("sudo"); + command.args(["-n", "kill"]); + command + } + #[cfg(all(test, target_os = "linux"))] + ProcessScope::Pids(_) => Command::new("kill"), + }; + command.args([signal, "--"]); + command.args(pids.iter().map(u32::to_string)); + let _ = command.output(); + } + + /// Poll until the tracked child is reaped and nothing in scope is left + /// running, or `timeout` elapses. Returns whether everything stopped. + fn wait_until_stopped(&mut self, timeout: Duration) -> bool { + let deadline = Instant::now() + timeout; + loop { + if let Ok(Some(_)) = self.child.try_wait() { + self.reaped = true; + } + if self.reaped && self.scope_pids().is_empty() { + return true; + } + if Instant::now() >= deadline { + return false; + } + std::thread::sleep(Duration::from_millis(100)); + } + } + + fn scope_pids(&self) -> Vec { + match &self.scope { + ProcessScope::Namespace(namespace) => { + let Ok(output) = Command::new("sudo") + .args(["-n", "ip", "netns", "pids", namespace]) + .output() + else { + return Vec::new(); + }; + if !output.status.success() { + return Vec::new(); + } + String::from_utf8_lossy(&output.stdout) + .split_whitespace() + .filter_map(|pid| pid.parse().ok()) + .collect() + } + #[cfg(all(test, target_os = "linux"))] + ProcessScope::Pids(pids) => pids + .iter() + .copied() + .filter(|pid| { + Command::new("kill") + .args(["-0", "--", &pid.to_string()]) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .status() + .is_ok_and(|status| status.success()) + }) + .collect(), + } } } @@ -329,6 +409,113 @@ impl Drop for NamespaceProcess { } } +#[cfg(all(test, target_os = "linux"))] +mod namespace_process_tests { + use std::io::{BufRead, BufReader}; + use std::process::{Command, Stdio}; + use std::sync::{Arc, Mutex, mpsc}; + use std::thread; + use std::time::Duration; + + use super::{NamespaceProcess, ProcessScope}; + + fn signal_exact(signal: &str, target: &str) { + let _ = Command::new("kill").args([signal, "--", target]).status(); + } + + fn is_alive(pid: u32) -> bool { + Command::new("kill") + .args(["-0", "--", &pid.to_string()]) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .status() + .is_ok_and(|status| status.success()) + } + + /// The exact shape `sudo ip netns exec` produces: the tracked child is + /// NOT its own process-group leader, and the process that must actually + /// die lives in a third group. Signalling `-` reaches + /// neither, so the old teardown's unbounded `wait()` never returned. + #[test] + fn kill_returns_when_wrapper_and_inner_process_have_mismatched_groups() { + let mut child = Command::new("sh") + .args([ + "-c", + "setsid sh -c 'trap \"\" TERM INT; echo ready; exec sleep 30' & echo $!; wait", + ]) + .stdout(Stdio::piped()) + .stderr(Stdio::null()) + .spawn() + .expect("spawn mismatched process groups"); + let wrapper_pid = child.id(); + let mut reader = BufReader::new(child.stdout.take().expect("child stdout")); + let mut pid_line = String::new(); + reader.read_line(&mut pid_line).expect("read inner pid"); + let inner_pid = pid_line.trim().parse::().expect("parse inner pid"); + let mut readiness_line = String::new(); + reader + .read_line(&mut readiness_line) + .expect("read inner readiness"); + assert_eq!(readiness_line.trim(), "ready"); + + let wrapper_pgid = Command::new("ps") + .args(["-o", "pgid=", "-p", &wrapper_pid.to_string()]) + .output() + .expect("read wrapper pgid"); + let wrapper_pgid = String::from_utf8_lossy(&wrapper_pgid.stdout) + .trim() + .parse::() + .expect("parse wrapper pgid"); + assert_ne!( + wrapper_pid, wrapper_pgid, + "wrapper unexpectedly leads its PGID" + ); + assert_ne!( + wrapper_pgid, inner_pid, + "inner process did not create a new group" + ); + for signal in ["-TERM", "-INT", "-TERM", "-INT"] { + signal_exact(signal, &inner_pid.to_string()); + } + assert!(is_alive(inner_pid), "inner process did not ignore TERM/INT"); + + let mut process = NamespaceProcess { + child, + label: "mismatched process groups".to_string(), + scope: ProcessScope::Pids(vec![inner_pid]), + reaped: false, + stdout: Arc::new(Mutex::new(Vec::new())), + stderr: Arc::new(Mutex::new(Vec::new())), + }; + let (done_tx, done_rx) = mpsc::channel(); + let teardown = thread::spawn(move || { + // Twice, plus the drop: teardown must be idempotent, because + // `stop()` kills explicitly and `Drop` kills again. + process.kill(); + process.kill(); + drop(process); + let _ = done_tx.send(()); + }); + + let completed = done_rx.recv_timeout(Duration::from_secs(8)).is_ok(); + if !completed { + signal_exact("-KILL", &wrapper_pid.to_string()); + signal_exact("-KILL", &format!("-{inner_pid}")); + } + teardown.join().expect("teardown thread"); + + assert!( + completed, + "NamespaceProcess::kill blocked on mismatched process groups" + ); + assert!( + !is_alive(wrapper_pid), + "wrapper process leaked after teardown" + ); + assert!(!is_alive(inner_pid), "inner process leaked after teardown"); + } +} + // --------------------------------------------------------------------------- // SrtlaTestTopology // --------------------------------------------------------------------------- diff --git a/src/lib.rs b/src/lib.rs index da853ac8..2ae581ab 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -7,9 +7,10 @@ // Use mimalloc as the global allocator for tests (non-Windows only). Excluded // under miri: the batch_recv miri CI lane interprets the test binary, and miri // cannot execute mimalloc's C FFI, so those runs fall back to miri's own -// allocator instead. -#[cfg(not(windows))] -#[cfg(all(test, not(miri)))] +// allocator instead. Gated on the default-on `mimalloc` feature so +// --no-default-features also builds the test binary against the system +// allocator. +#[cfg(all(not(windows), test, not(miri), feature = "mimalloc"))] #[global_allocator] static ALLOC: mimalloc::MiMalloc = mimalloc::MiMalloc; diff --git a/src/main.rs b/src/main.rs index 3e7fe662..1f0e770c 100644 --- a/src/main.rs +++ b/src/main.rs @@ -14,8 +14,10 @@ use srtla_send::{ }; use tracing_subscriber::EnvFilter; -// Use mimalloc as the global allocator for the binary (non-Windows only) -#[cfg(not(windows))] +// Use mimalloc as the global allocator for the binary (non-Windows only). +// Gated default-on via the `mimalloc` feature; --no-default-features builds +// fall back to the system allocator. +#[cfg(all(not(windows), feature = "mimalloc"))] #[global_allocator] static ALLOC: mimalloc::MiMalloc = mimalloc::MiMalloc; diff --git a/src/sender/mod.rs b/src/sender/mod.rs index c2085f55..0a74972e 100644 --- a/src/sender/mod.rs +++ b/src/sender/mod.rs @@ -192,6 +192,12 @@ pub async fn run_sender_with_config( // Prepare SIGHUP stream (Unix only) or a never-completing future (non-Unix) #[cfg(unix)] let mut sighup = signal(SignalKind::hangup())?; + // SIGTERM/SIGINT drive a graceful `Ok(())` return so the process exits 0 + // under a supervisor instead of dying to the default signal action. + #[cfg(unix)] + let mut sigterm = signal(SignalKind::terminate())?; + #[cfg(unix)] + let mut sigint = signal(SignalKind::interrupt())?; // Main loop - run housekeeping frequently like C version // Run housekeeping once before entering the main event loop so we start in a clean state. @@ -435,6 +441,14 @@ pub async fn run_sender_with_config( ) .await; } + _ = sigterm.recv() => { + info!("received SIGTERM - shutting down"); + return Ok(()); + } + _ = sigint.recv() => { + info!("received SIGINT - shutting down"); + return Ok(()); + } } #[cfg(not(unix))] diff --git a/tests/signal_shutdown.rs b/tests/signal_shutdown.rs new file mode 100644 index 00000000..f5884ac3 --- /dev/null +++ b/tests/signal_shutdown.rs @@ -0,0 +1,182 @@ +//! Signal shutdown integration tests (Unix). +//! +//! These spawn the real `srtla_send` binary and pin that SIGTERM and SIGINT +//! produce a prompt, clean exit (code 0) rather than death by the default +//! signal action, which a supervisor reports as a failure. +#![cfg(unix)] + +use std::io::{BufRead, BufReader}; +use std::process::{Child, Command, Stdio}; +use std::sync::{Arc, Mutex}; +use std::thread; +use std::time::{Duration, Instant}; + +const BIN: &str = env!("CARGO_BIN_EXE_srtla_send"); +const START_TIMEOUT: Duration = Duration::from_secs(5); +/// Poll well past `PROMPT_EXIT` so a slow exit is reported as too slow rather +/// than as "never exited". +const EXIT_TIMEOUT: Duration = Duration::from_secs(10); +const PROMPT_EXIT: Duration = Duration::from_secs(2); + +fn free_udp_port() -> u16 { + let sock = std::net::UdpSocket::bind("127.0.0.1:0").expect("bind ephemeral udp"); + sock.local_addr().unwrap().port() +} + +struct SenderProc { + child: Child, + logs: Arc>, +} + +impl SenderProc { + fn spawn(args: &[&str]) -> Self { + let mut child = Command::new(BIN) + .args(args) + .env("RUST_LOG", "info") + .stdin(Stdio::null()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .expect("spawn srtla_send"); + + // tracing writes to stdout; panics/anyhow errors land on stderr. Pump both + // into one buffer so log assertions see everything the binary emitted. + let logs = Arc::new(Mutex::new(String::new())); + let streams: [Box; 2] = [ + Box::new(child.stdout.take().expect("capture stdout")), + Box::new(child.stderr.take().expect("capture stderr")), + ]; + for stream in streams { + let sink = logs.clone(); + thread::spawn(move || { + for line in BufReader::new(stream).lines().map_while(Result::ok) { + let mut buf = sink.lock().unwrap(); + buf.push_str(&line); + buf.push('\n'); + } + }); + } + + Self { child, logs } + } + + fn logs(&self) -> String { + self.logs.lock().unwrap().clone() + } + + fn wait_for_log(&self, needle: &str, timeout: Duration) -> bool { + let deadline = Instant::now() + timeout; + while Instant::now() < deadline { + if self.logs().contains(needle) { + return true; + } + thread::sleep(Duration::from_millis(50)); + } + false + } + + /// Block until the sender has installed its handler for `signum`. + /// + /// The handlers are registered after the "listening for SRT" line is + /// logged, so signalling straight off that line can still hit the default + /// action and kill the process. `SigCgt` in `/proc//status` is the + /// kernel's own record of caught signals, so it closes that window + /// without a guessed sleep. + #[cfg(target_os = "linux")] + fn wait_for_handler(&self, signum: u32, timeout: Duration) -> bool { + let path = format!("/proc/{}/status", self.child.id()); + let deadline = Instant::now() + timeout; + while Instant::now() < deadline { + let caught = std::fs::read_to_string(&path).ok().and_then(|status| { + let mask = status.lines().find_map(|l| l.strip_prefix("SigCgt:"))?; + u64::from_str_radix(mask.trim(), 16).ok() + }); + if caught.is_some_and(|mask| mask & (1 << (signum - 1)) != 0) { + return true; + } + thread::sleep(Duration::from_millis(20)); + } + false + } + + /// No portable way to observe the handler off Linux; give startup a moment. + #[cfg(not(target_os = "linux"))] + fn wait_for_handler(&self, _signum: u32, _timeout: Duration) -> bool { + thread::sleep(Duration::from_millis(500)); + true + } + + fn signal(&self, sig: &str) { + let status = Command::new("kill") + .args([format!("-{sig}"), self.child.id().to_string()]) + .status() + .expect("run kill"); + assert!(status.success(), "kill -{sig} failed"); + } + + fn wait_exit(&mut self, timeout: Duration) -> Option { + let deadline = Instant::now() + timeout; + while Instant::now() < deadline { + match self.child.try_wait() { + // A signal death has no exit code; report it as -1 so it can + // never be mistaken for the clean 0 these tests require. + Ok(Some(status)) => return Some(status.code().unwrap_or(-1)), + Ok(None) => thread::sleep(Duration::from_millis(20)), + Err(_) => return None, + } + } + None + } +} + +impl Drop for SenderProc { + fn drop(&mut self) { + let _ = self.child.kill(); + let _ = self.child.wait(); + } +} + +fn assert_signal_exits_cleanly(sig: &str, signum: u32) { + let dir = tempfile::tempdir().unwrap(); + let ips = dir.path().join("ips.txt"); + std::fs::write(&ips, "127.0.0.1\n").expect("write ips file"); + let port = free_udp_port().to_string(); + let mut proc = SenderProc::spawn(&[port.as_str(), "127.0.0.1", "9999", ips.to_str().unwrap()]); + + assert!( + proc.wait_for_log("listening for SRT", START_TIMEOUT), + "sender never started; logs:\n{}", + proc.logs() + ); + assert!( + proc.wait_for_handler(signum, START_TIMEOUT), + "sender never installed a SIG{sig} handler; logs:\n{}", + proc.logs() + ); + + let t0 = Instant::now(); + proc.signal(sig); + let code = proc.wait_exit(EXIT_TIMEOUT); + let elapsed = t0.elapsed(); + + assert_eq!( + code, + Some(0), + "SIG{sig} must exit 0; logs:\n{}", + proc.logs() + ); + assert!( + elapsed < PROMPT_EXIT, + "SIG{sig} must exit promptly (< {PROMPT_EXIT:?}); took {elapsed:?}" + ); +} + +#[test] +fn sigterm_exits_cleanly_and_promptly() { + assert_signal_exits_cleanly("TERM", 15); +} + +#[test] +fn sigint_exits_cleanly_and_promptly() { + assert_signal_exits_cleanly("INT", 2); +}