diff --git a/crates/tracedecay/tests/common/mod.rs b/crates/tracedecay/tests/common/mod.rs index 28bbc596c4..2f0774b287 100644 --- a/crates/tracedecay/tests/common/mod.rs +++ b/crates/tracedecay/tests/common/mod.rs @@ -657,6 +657,9 @@ pub fn http_agent_with_timeout(timeout: Duration) -> ureq::Agent { /// panic while the child is still running, `Drop` force-stops and reaps it. pub struct TestChildProcess { child: Child, + /// Unix socket this child bound. Retired once the child is reaped so a + /// descriptor a subprocess inherited cannot keep the path connectable. + owned_unix_endpoint: Option, } /// Daemon-specific name retained for test fixtures that keep a daemon alive. @@ -664,7 +667,39 @@ pub type DaemonProcess = TestChildProcess; impl TestChildProcess { pub fn new(child: Child) -> Self { - Self { child } + Self { + child, + owned_unix_endpoint: None, + } + } + + /// Record the Unix socket this child bound. + /// + /// `SIGKILL` skips the daemon's own endpoint cleanup (`drop(listener)` then + /// unlink). A subprocess that inherited the listen descriptor across `fork` + /// still accepts on that path after the owner is reaped, which is a leaked + /// descriptor, not a live daemon. Retiring the path with the owner is what + /// the graceful shutdown already does, and what a forced stop must do too. + #[cfg(unix)] + pub fn own_unix_endpoint(&mut self, path: PathBuf) { + self.owned_unix_endpoint = Some(path); + } + + /// Unlink the recorded endpoint, but only after a stop this call forced. + /// + /// A child that exited on its own already ran its own endpoint cleanup, so + /// the path is either gone or has since been rebound by a replacement. The + /// restart journeys reassign the guard (`daemon = spawn_...`), which drops + /// the predecessor *after* the successor has bound the same path; unlinking + /// there would retire a live daemon's socket. + fn retire_owned_unix_endpoint(&mut self, forced: bool) { + let path = self.owned_unix_endpoint.take(); + if !forced { + return; + } + if let Some(path) = path { + let _ = std::fs::remove_file(path); + } } pub fn id(&self) -> u32 { @@ -752,9 +787,13 @@ impl TestChildProcess { /// Force-stops the daemon and reaps its process before returning. /// /// `Child::kill` maps to `SIGKILL` on Unix and the platform termination - /// primitive elsewhere, keeping fault-injection tests portable. + /// primitive elsewhere, keeping fault-injection tests portable. The owned + /// Unix endpoint, when recorded, is unlinked only after that reap. pub fn kill_and_wait(&mut self) -> std::io::Result { - terminate_and_reap(&mut self.child) + let forced = self.is_running(); + let status = terminate_and_reap(&mut self.child)?; + self.retire_owned_unix_endpoint(forced); + Ok(status) } fn drain_stderr(&mut self) { @@ -778,15 +817,30 @@ impl TestChildProcess { impl Drop for TestChildProcess { fn drop(&mut self) { - let _ = terminate_and_reap(&mut self.child); + let forced = self.is_running(); + if terminate_and_reap(&mut self.child).is_ok() { + self.retire_owned_unix_endpoint(forced); + } } } -/// PID-directed stop: survives `process_group(0)` / `setsid` detachment. +/// Stop the child and the process group it leads. +/// +/// Test daemons are started with `process_group(0)`, so the spawned pid is the +/// group leader and is outside the test's group. `Child::kill` signals only +/// that pid. A child still between `fork` and `exec` inherits the listen +/// descriptor and stays in the leader's group, so the path keeps accepting +/// after the leader is reaped. Signal that group first, while the pid is still +/// the live leader, then reap the leader. Do not signal after `try_wait` has +/// reaped: the pid can be recycled. A child that left the group (its own +/// `process_group(0)`) is not this signal's target; unlinking the owned socket +/// retires the path those descriptors were bound to. fn terminate_and_reap(child: &mut Child) -> std::io::Result { + let pid = child.id(); if let Ok(Some(status)) = child.try_wait() { return Ok(status); } + signal_owned_process_group(pid); if let Err(kill_err) = child.kill() { if let Some(status) = child.try_wait()? { @@ -798,6 +852,23 @@ fn terminate_and_reap(child: &mut Child) -> std::io::Result { child.wait() } +#[cfg(unix)] +fn signal_owned_process_group(pid: u32) { + let Ok(pid) = i32::try_from(pid) else { + return; + }; + if pid <= 0 { + return; + } + // Safety: `pid` is the still-live child returned by `Child::id` before + // `try_wait` reaps it, and `process_group(0)` made that pid the group id. + // A non-positive argument would signal this process's own group. + let _ = unsafe { libc::kill(-pid, libc::SIGKILL) }; +} + +#[cfg(not(unix))] +fn signal_owned_process_group(_pid: u32) {} + /// Detach a test child from the test process group. /// /// Nextest (and other harness timeouts) signal the test's process group. @@ -1089,13 +1160,6 @@ pub fn spawn_tracedecay_daemon_with( spawn_tracedecay_daemon_process(&home, &binary, configure) } -/// How long a replacement daemon waits for a stopped predecessor's endpoint to -/// stop accepting before reporting it as still live. -/// -/// Generous on purpose: the wait only costs time when a predecessor is -/// genuinely still reachable, and a real leak still fails rather than hangs. -const PREDECESSOR_DAEMON_VACATE_TIMEOUT: Duration = Duration::from_secs(10); - fn spawn_tracedecay_daemon_process( home: &Path, binary: &Path, @@ -1118,42 +1182,20 @@ fn spawn_tracedecay_daemon_process( }) .is_some_and(|address| TcpStream::connect(address).is_ok()) }; - // Stopping a predecessor daemon is asynchronous with respect to its - // endpoint: `kill` plus `wait` reaps the PID the harness spawned, but the - // kernel keeps the listening socket alive while *any* duplicate of that - // descriptor survives, including one a subprocess inherited across `fork` - // and still holds because it has not reached its own `exec` yet. Asserting - // instantaneously therefore reports an ordinary teardown tail as a live - // daemon, which is what `init_project_fixture` journeys (spawn, init, drop, - // spawn again) hit on a loaded runner. Wait a bounded time for the endpoint - // to stop accepting; a daemon that keeps accepting still fails with the - // same refusal. - poll_until( - Instant::now() + PREDECESSOR_DAEMON_VACATE_TIMEOUT, - Duration::from_millis(25), - || { - #[cfg(unix)] - let live = std::os::unix::net::UnixStream::connect(&socket_path).is_ok(); - #[cfg(not(unix))] - let live = portable_daemon_connectable(); - (!live).then_some(()) - }, - || { - #[cfg(unix)] - { - format!( - "refusing to replace a live test daemon at {}", - socket_path.display() - ) - } - #[cfg(not(unix))] - { - format!( - "refusing to replace a live test daemon recorded at {}", - authority_path.display() - ) - } - }, + // A reaped predecessor has already unlinked its socket. A path that still + // accepts is a daemon this call did not stop, and replacing it would bind + // over a live owner. + #[cfg(unix)] + assert!( + std::os::unix::net::UnixStream::connect(&socket_path).is_err(), + "refusing to replace a live test daemon at {}", + socket_path.display() + ); + #[cfg(not(unix))] + assert!( + !portable_daemon_connectable(), + "refusing to replace a live test daemon recorded at {}", + authority_path.display() ); let mut command = Command::new(binary); @@ -1169,6 +1211,8 @@ fn spawn_tracedecay_daemon_process( detach_from_test_process_group(&mut command); let child = command.spawn().expect("tracedecay daemon should start"); let mut daemon = DaemonProcess::new(child); + #[cfg(unix)] + daemon.own_unix_endpoint(socket_path.clone()); let deadline = Instant::now() + Duration::from_secs(10); poll_until( diff --git a/crates/tracedecay/tests/transport_acceptance_suite/daemon_fault_harness_test.rs b/crates/tracedecay/tests/transport_acceptance_suite/daemon_fault_harness_test.rs index dcf97a9b1d..447b47d06f 100644 --- a/crates/tracedecay/tests/transport_acceptance_suite/daemon_fault_harness_test.rs +++ b/crates/tracedecay/tests/transport_acceptance_suite/daemon_fault_harness_test.rs @@ -321,6 +321,75 @@ fn configured_daemon_can_be_killed_and_reaped() { } } +/// A listen descriptor inherited across `fork` must not outlive its owner. +/// +/// `branch_search_serves_a_committed_generation_behind_dirty_worktree_state` +/// drops the init daemon and immediately spawns the next one. Killing only the +/// leader leaves the inherited listener accepting on the same path, so the +/// next spawn reports a live daemon. The owner is the group leader; retiring +/// its socket with it makes that path refuse the moment the owner is reaped. +#[cfg(unix)] +#[test] +fn reaped_owner_releases_an_inherited_listen_socket() { + use std::os::unix::process::CommandExt; + use std::process::Stdio; + use std::time::{Duration, Instant}; + + let home = tempdir_or_panic(); + let socket_path = home.path().join("inherited-listen.sock"); + let script = r#" +import os, socket, sys, time +sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) +sock.bind(sys.argv[1]) +sock.listen(1) +if os.fork() == 0: + time.sleep(60) +else: + time.sleep(60) +"#; + let mut command = std::process::Command::new("python3"); + command + .arg("-c") + .arg(script) + .arg(&socket_path) + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .process_group(0); + let child = command.spawn().expect("python listener should start"); + let mut owner = common::TestChildProcess::new(child); + owner.own_unix_endpoint(socket_path.clone()); + + let deadline = Instant::now() + Duration::from_secs(5); + loop { + if std::os::unix::net::UnixStream::connect(&socket_path).is_ok() { + break; + } + if let Some(status) = owner + .try_wait() + .expect("listener status should be readable") + { + panic!("listener exited before accepting: {status}"); + } + assert!( + Instant::now() < deadline, + "inherited listener never accepted on {}", + socket_path.display() + ); + std::thread::sleep(Duration::from_millis(20)); + } + + drop(owner); + assert!( + std::os::unix::net::UnixStream::connect(&socket_path).is_err(), + "an inherited listen descriptor must not keep the owner's path accepting" + ); + assert!( + !socket_path.exists(), + "reaping the owner must unlink the socket it bound" + ); +} + #[cfg(all(unix, tracedecay_observation_fault_harness, feature = "test-transport"))] async fn assert_daemon_crash_stage( barrier_stage: &str,