From a70fd4ad607ec5bcb0a2b3e888ee03bc1d6a51c8 Mon Sep 17 00:00:00 2001 From: datagutt Date: Mon, 21 Sep 2026 03:20:06 +0200 Subject: [PATCH 1/4] fix(ci): move off yanked chacha20 0.10.1 `cargo deny check advisories` fails on main with `error[yanked]: detected yanked crate` for chacha20 0.10.1, pulled in transitively by rand 0.10.2 -> network-sim (a dev-dependency), because deny.toml sets `yanked = "deny"`. 0.10.2 is published and not yanked, so `cargo update -p chacha20` is the remedy the tool itself suggests; the advisory gate is not relaxed. Same fix as CERALIVE/srtla-send-rs 9c6e35c. --- Cargo.lock | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) 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", From a93b274413db3796722caadf942921385471b359 Mon Sep 17 00:00:00 2001 From: datagutt Date: Mon, 21 Sep 2026 03:20:07 +0200 Subject: [PATCH 2/4] fix(network-sim): bound namespace-process teardown to the namespace's own PIDs `NamespaceProcess::kill` signalled `-`, assuming the tracked `sudo ip netns exec` child led its own process group. It does not: `spawn_with_env` uses a plain `Command::spawn()`, so the wrapper inherits the harness's group and its PID is not a PGID. `kill -- -` therefore fails ESRCH, the error was discarded by `let _ =`, and the trailing `self.child.wait()` could block forever on a process nothing had signalled. Teardown now signals the exact PIDs `ip netns pids` reports, polls for their disappearance under bounded TERM-then-KILL grace periods, and is idempotent across the explicit `kill()` in `stop()` plus `Drop`. This signals every process in the namespace, not only the one being killed. That is correct for whole-stack teardown, the only caller today, and is documented on `kill()` as unsuitable for stopping one process mid-test while another in the same namespace must keep running. Pinned by `kill_returns_when_wrapper_and_inner_process_have_mismatched_groups`, which reproduces the three-group shape without needing privileges. Found and fixed by CERALIVE/srtla-send-rs 4f258c7 (Andres Cera); this is the kill logic from that commit. Its companion change to a bounded 64 KiB log tail is deliberately not taken: netns_stall_gate asserts on mid-run log lines and several netns tests scan the whole run for panics, both of which a tail would silently drop at RUST_LOG=debug. --- crates/network-sim/src/harness.rs | 239 ++++++++++++++++++++++++++---- 1 file changed, 213 insertions(+), 26 deletions(-) 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 // --------------------------------------------------------------------------- From 4a1ba9f7ed861326383ec1f22f695eab84ded2be Mon Sep 17 00:00:00 2001 From: datagutt Date: Mon, 21 Sep 2026 03:20:10 +0200 Subject: [PATCH 3/4] build: gate mimalloc behind a default-on feature mimalloc was an unconditional dependency and global allocator. It is now optional behind a `mimalloc` feature that is in `default`, so the shipped binary links it exactly as before, while `--no-default-features` builds the binary and the test harness against the system allocator for profiling and allocator-sensitive tooling. Verified with nm: the default build carries mi_malloc symbols, the --no-default-features build carries none. Ported from CERALIVE/srtla-send-rs f0e33ab (Andres Cera). --- Cargo.toml | 6 +++++- src/lib.rs | 7 ++++--- src/main.rs | 6 ++++-- 3 files changed, 13 insertions(+), 6 deletions(-) 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/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; From bb10171bec76f5d3c8036823f536848d630a02d6 Mon Sep 17 00:00:00 2001 From: datagutt Date: Mon, 21 Sep 2026 03:20:12 +0200 Subject: [PATCH 4/4] feat(srtla_send): exit cleanly on SIGTERM/SIGINT The sender only handled SIGHUP, so SIGTERM and SIGINT killed it through the default signal action and a supervisor saw a signal death rather than a clean stop. Both now return `Ok(())` from the event loop, so the process logs the shutdown and exits 0. There is nothing to release on the way out (the control socket path is only unlinked at startup), so this carries no shutdown-hook machinery. `tests/signal_shutdown.rs` spawns the real binary and pins a prompt exit code 0 for both signals. The handlers are registered after the "listening for SRT" line is logged, so the test waits for the handler to appear in `/proc//status` (SigCgt) before signalling instead of racing the default action. Signal half of CERALIVE/srtla-send-rs d03056a (Andres Cera); its empty-start-waits-for-SIGHUP behaviour is not taken. --- src/sender/mod.rs | 14 +++ tests/signal_shutdown.rs | 182 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 196 insertions(+) create mode 100644 tests/signal_shutdown.rs 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); +}