From 8cfc2a00d6b7d95ccc59dce4b759affb940941a0 Mon Sep 17 00:00:00 2001 From: Dimitri Kennedy Date: Sat, 24 Oct 2026 20:00:16 -0400 Subject: [PATCH 1/2] test(runtime-core): prove relay refusals without socket timeouts and reap failed relay starts HACK-1342 items 2 and 4 (item 1 is already fixed on next by #399; item 3 is left to #419). Item 2: the relay refuses an excess connection (and one whose target is down) by accepting and closing it, which can happen before the client's next syscall. Once the peer has closed, macOS rejects SO_RCVTIMEO and SO_SNDTIMEO with EINVAL, so `Relay::connect` panicked at the `set_read_timeout`/`set_write_timeout` unwraps. The relay is correct. The two refusal tests now use `Relay::connect_refused`, which sets no socket timeouts: it waits for readability with a bounded poll and then still requires EOF or a reset. A live peer or a peer that sends data still fails the check. Item 4: `Relay::start_options` now hands the spawned child to the existing `PublisherStartup` guard until readiness is proven, so a failed or panicking readiness check kills and reaps `stream-relay-host` (and removes a fresh root) instead of orphaning it. Refs HACK-1342 Co-Authored-By: Claude Opus 5.5 --- .../tests/stream_relay_contract.rs | 179 ++++++++++++++++-- 1 file changed, 158 insertions(+), 21 deletions(-) diff --git a/packages/runtime-core/tests/stream_relay_contract.rs b/packages/runtime-core/tests/stream_relay_contract.rs index 02584c656..486d18736 100644 --- a/packages/runtime-core/tests/stream_relay_contract.rs +++ b/packages/runtime-core/tests/stream_relay_contract.rs @@ -72,25 +72,19 @@ impl Relay { if let Some(reservation) = reservation { command.args(["--reservation", reservation]); } - let mut child = command - .stdin(Stdio::null()) - .stdout(Stdio::piped()) - .spawn() - .unwrap(); - let output = child.stdout.take().unwrap(); - let mut descriptor = libc::pollfd { - fd: output.as_raw_fd(), - events: libc::POLLIN, - revents: 0, + // A caller-supplied root stays the caller's; only a fresh root is removed on failure. + let mut startup = PublisherStartup { + child: None, + root: if reuse { PathBuf::new() } else { root.clone() }, }; - assert_eq!(unsafe { libc::poll(&mut descriptor, 1, 3000) }, 1); - let mut line = String::new(); - BufReader::new(output).read_line(&mut line).unwrap(); - assert_eq!(line, "ready\n"); - assert_eq!( - fs::metadata(&socket).unwrap().permissions().mode() & 0o777, - 0o700 + startup.child = Some( + command + .stdin(Stdio::null()) + .stdout(Stdio::piped()) + .spawn() + .unwrap(), ); + let child = await_bare_ready(startup, &socket); Self { child, root, @@ -98,6 +92,8 @@ impl Relay { port: None, } } + /// Connects a client the relay must keep. Setting a timeout fails if the relay already + /// closed the stream (see `connect_refused`), which is a real failure here. fn connect(&self) -> UnixStream { let stream = UnixStream::connect(&self.socket).unwrap(); stream @@ -108,6 +104,16 @@ impl Relay { .unwrap(); stream } + /// Connects a client the relay must refuse, and proves the refusal. + /// + /// The relay refuses by closing the accepted stream, which can happen before this + /// client's next syscall. Once the peer has closed, XNU's `sosetopt` rejects + /// `SO_RCVTIMEO`/`SO_SNDTIMEO` with EINVAL (both directions are shut), so this never + /// sets socket timeouts; it bounds the wait with `poll` instead. + fn connect_refused(&self) { + let stream = UnixStream::connect(&self.socket).unwrap(); + assert_refused_within(&stream, REFUSAL_DEADLINE); + } fn stop(&mut self) { if self.child.try_wait().unwrap().is_none() { unsafe { @@ -135,6 +141,79 @@ impl Drop for Relay { let _ = fs::remove_dir_all(&self.root); } } + +/// Requires a relay's bare "ready\n" line and a private socket, then hands the child over. +/// No `Relay` exists until this returns, so `startup` owns the child meanwhile: if any check +/// fails or panics, its `Drop` kills and reaps the child (and removes a fresh root) instead +/// of orphaning `stream-relay-host`. +fn await_bare_ready(mut startup: PublisherStartup, socket: &std::path::Path) -> Child { + let output = startup + .child + .as_mut() + .expect("startup owns child") + .stdout + .take() + .unwrap(); + let mut descriptor = libc::pollfd { + fd: output.as_raw_fd(), + events: libc::POLLIN, + revents: 0, + }; + assert_eq!(unsafe { libc::poll(&mut descriptor, 1, 3000) }, 1); + let mut line = String::new(); + BufReader::new(output).read_line(&mut line).unwrap(); + assert_eq!(line, "ready\n"); + assert_eq!( + fs::metadata(socket).unwrap().permissions().mode() & 0o777, + 0o700 + ); + startup.root = PathBuf::new(); + startup.child.take().expect("startup owns child") +} + +#[test] +fn failed_relay_readiness_kills_and_reaps_the_child() { + let root = PathBuf::from(format!( + "/tmp/hkr-{}-{}", + std::process::id(), + NEXT.fetch_add(1, Ordering::Relaxed) + )); + fs::create_dir(&root).unwrap(); + // A live child with the wrong readiness line fails the check without exiting on its own. + let child = Command::new("/bin/sh") + .args(["-c", "echo unready; exec sleep 30"]) + .stdin(Stdio::null()) + .stdout(Stdio::piped()) + .spawn() + .unwrap(); + let pid = child.id() as libc::pid_t; + let startup = PublisherStartup { + child: Some(child), + root: root.clone(), + }; + let socket = root.join("app.sock"); + let failed = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { + await_bare_ready(startup, &socket) + })); + let mut status = 0; + // SAFETY: a non-blocking wait on this test's own child pid. + let waited = unsafe { libc::waitpid(pid, &mut status, libc::WNOHANG) }; + let error = io::Error::last_os_error(); + if waited == 0 { + // SAFETY: the child is still ours and unreaped; stop it before failing. + unsafe { + libc::kill(pid, libc::SIGKILL); + libc::waitpid(pid, &mut status, 0); + } + let _ = fs::remove_dir_all(&root); + panic!("E_RELAY_STARTUP_ORPHAN: failed readiness left the relay child running"); + } + assert!(failed.is_err(), "wrong readiness line must fail startup"); + // ECHILD: the guard already killed and reaped it, so no zombie or orphan remains. + assert_eq!(waited, -1); + assert_eq!(error.raw_os_error(), Some(libc::ECHILD)); + assert!(!root.exists(), "failed startup must remove its fresh root"); +} #[test] fn large_stream_half_close_drains_all_buffered_bytes() { let target = TcpListener::bind("127.0.0.1:0").unwrap(); @@ -326,8 +405,7 @@ fn occupied_socket_is_not_unlinked_and_unavailable_target_fails_connection() { .status() .unwrap(); assert_eq!(status.code(), Some(73)); - let mut client = relay.connect(); - assert_refused(client.read(&mut [0])); + relay.connect_refused(); assert!(relay.child.try_wait().unwrap().is_none()); } @@ -347,8 +425,7 @@ fn connection_limit_refuses_excess_without_disrupting_existing_streams() { clients.push(client); upstreams.push(upstream); } - let mut excess = relay.connect(); - assert_refused(excess.read(&mut [0])); + relay.connect_refused(); assert!(relay.child.try_wait().unwrap().is_none()); for (client, upstream) in clients.iter_mut().zip(&mut upstreams) { upstream.write_all(b"alive").unwrap(); @@ -372,6 +449,66 @@ fn assert_refused(result: std::io::Result) { } } +/// Bounds how long a refused client waits for the relay to close it. +const REFUSAL_DEADLINE: Duration = Duration::from_secs(10); + +/// Proves `stream` is refused within `limit` without setting socket timeouts: waits for +/// readability with `poll`, then requires EOF or a reset. A live peer times out and fails; +/// a peer that sends data fails the refusal check. +fn assert_refused_within(stream: &UnixStream, limit: Duration) { + let deadline = Instant::now() + limit; + loop { + let remaining = deadline.saturating_duration_since(Instant::now()); + let mut descriptor = libc::pollfd { + fd: stream.as_raw_fd(), + events: libc::POLLIN, + revents: 0, + }; + // SAFETY: one initialized descriptor lives throughout this bounded poll. + let polled = unsafe { + libc::poll( + &mut descriptor, + 1, + remaining.as_millis().min(i32::MAX as u128) as i32, + ) + }; + if polled < 0 && io::Error::last_os_error().kind() == io::ErrorKind::Interrupted { + continue; + } + assert_eq!( + polled, 1, + "expected prompt refusal, stream stayed open for {limit:?}" + ); + break; + } + // Readable: EOF, a reset, or data is pending, so this read cannot block. + let mut reader = stream; + assert_refused(reader.read(&mut [0])); +} + +#[test] +fn refusal_check_accepts_a_peer_that_closed_before_the_check() { + // The relay can close a refused stream before the client's first syscall after connect. + let (client, relay_side) = UnixStream::pair().unwrap(); + drop(relay_side); + assert_refused_within(&client, REFUSAL_DEADLINE); +} + +#[test] +#[should_panic(expected = "expected prompt refusal")] +fn refusal_check_rejects_a_live_peer() { + let (client, _relay_side) = UnixStream::pair().unwrap(); + assert_refused_within(&client, Duration::from_millis(50)); +} + +#[test] +#[should_panic(expected = "expected prompt refusal")] +fn refusal_check_rejects_a_peer_that_sends_data() { + let (client, mut relay_side) = UnixStream::pair().unwrap(); + relay_side.write_all(b"x").unwrap(); + assert_refused_within(&client, REFUSAL_DEADLINE); +} + #[test] fn inherited_descriptors_do_not_keep_parent_channels_alive() { let target = TcpListener::bind("127.0.0.1:0").unwrap(); From 5752a3296d616139b41973982e1912b1c89a263f Mon Sep 17 00:00:00 2001 From: Dimitri Kennedy Date: Sat, 24 Oct 2026 22:10:23 -0400 Subject: [PATCH 2/2] test(runtime-core): skip stale relay scratch roots left by a reused PID (HACK-1342) Relay test roots were named /tmp/-- and created with create_dir().unwrap(), so a directory left by an earlier run whose PID is now reused failed a fresh test with EEXIST. fresh_root() skips an existing name, and every root site uses it, including the fixed-name ones. Co-Authored-By: Claude Opus 5.5 --- .../tests/stream_relay_contract.rs | 90 +++++++++---------- 1 file changed, 43 insertions(+), 47 deletions(-) diff --git a/packages/runtime-core/tests/stream_relay_contract.rs b/packages/runtime-core/tests/stream_relay_contract.rs index 486d18736..69c935954 100644 --- a/packages/runtime-core/tests/stream_relay_contract.rs +++ b/packages/runtime-core/tests/stream_relay_contract.rs @@ -19,6 +19,25 @@ use std::{ }; static NEXT: AtomicU64 = AtomicU64::new(0); const BINARY: &str = concat!(env!("OUT_DIR"), "/stream-relay-host"); + +/// Creates a new scratch root `/tmp/--`. A name left behind by +/// an earlier run whose PID is now reused is skipped, so a stale directory +/// never fails a fresh test. +fn fresh_root(prefix: &str) -> PathBuf { + loop { + let root = PathBuf::from(format!( + "/tmp/{prefix}-{}-{}", + std::process::id(), + NEXT.fetch_add(1, Ordering::Relaxed) + )); + match fs::create_dir(&root) { + Ok(()) => return root, + Err(error) if error.kind() == io::ErrorKind::AlreadyExists => continue, + Err(error) => panic!("create {}: {error}", root.display()), + } + } +} + struct Relay { child: Child, root: PathBuf, @@ -41,16 +60,7 @@ impl Relay { existing: Option, ) -> Self { let reuse = existing.is_some(); - let root = existing.unwrap_or_else(|| { - PathBuf::from(format!( - "/tmp/hkr-{}-{}", - std::process::id(), - NEXT.fetch_add(1, Ordering::Relaxed) - )) - }); - if !reuse { - fs::create_dir(&root).unwrap(); - } + let root = existing.unwrap_or_else(|| fresh_root("hkr")); fs::set_permissions(&root, fs::Permissions::from_mode(0o700)).unwrap(); let socket = root.join("app.sock"); let mut command = Command::new(BINARY); @@ -172,13 +182,23 @@ fn await_bare_ready(mut startup: PublisherStartup, socket: &std::path::Path) -> } #[test] -fn failed_relay_readiness_kills_and_reaps_the_child() { - let root = PathBuf::from(format!( - "/tmp/hkr-{}-{}", +fn fresh_root_skips_a_stale_directory_with_the_next_name() { + let stale = PathBuf::from(format!( + "/tmp/hks-{}-{}", std::process::id(), - NEXT.fetch_add(1, Ordering::Relaxed) + NEXT.load(Ordering::Relaxed) )); - fs::create_dir(&root).unwrap(); + fs::create_dir(&stale).unwrap(); + let root = fresh_root("hks"); + assert_ne!(root, stale); + assert!(root.is_dir()); + fs::remove_dir(&root).unwrap(); + fs::remove_dir(&stale).unwrap(); +} + +#[test] +fn failed_relay_readiness_kills_and_reaps_the_child() { + let root = fresh_root("hkr"); // A live child with the wrong readiness line fails the check without exiting on its own. let child = Command::new("/bin/sh") .args(["-c", "echo unready; exec sleep 30"]) @@ -528,8 +548,7 @@ fn inherited_descriptors_do_not_keep_parent_channels_alive() { #[test] fn namespace_requests_reject_invalid_identity_before_publishing() { - let root = PathBuf::from(format!("/tmp/hkr-netns-{}", std::process::id())); - fs::create_dir(&root).unwrap(); + let root = fresh_root("hkr-netns"); fs::set_permissions(&root, fs::Permissions::from_mode(0o700)).unwrap(); let path = root.join("app.sock"); for (address, option, pid, start, code) in [ @@ -695,8 +714,7 @@ fn reused_socket_rejects_old_reservation_and_expiring_partial_handshakes() { #[test] fn malformed_reservations_refuse_before_socket_creation() { - let root = PathBuf::from(format!("/tmp/hkr-reservation-{}", std::process::id())); - fs::create_dir(&root).unwrap(); + let root = fresh_root("hkr-reservation"); let socket = root.join("app.sock"); for token in [ "", @@ -786,12 +804,7 @@ fn try_frontend_publisher( control_token: Option<&str>, unix: bool, ) -> Result { - let root = PathBuf::from(format!( - "/tmp/hkp-{}-{}", - std::process::id(), - NEXT.fetch_add(1, Ordering::Relaxed) - )); - fs::create_dir(&root).unwrap(); + let root = fresh_root("hkp"); fs::set_permissions(&root, fs::Permissions::from_mode(0o700)).unwrap(); let mut command = Command::new(BINARY); command @@ -1135,8 +1148,7 @@ fn publisher_refuses_port_conflicts_and_stale_reused_upstream() { #[test] fn publisher_requires_ack_before_forwarding_and_preserves_upstream_socket() { use std::os::unix::{fs::MetadataExt, net::UnixListener}; - let root = PathBuf::from(format!("/tmp/hkp-ack-{}", std::process::id())); - fs::create_dir(&root).unwrap(); + let root = fresh_root("hkp-ack"); let path = root.join("socket"); let upstream = UnixListener::bind(&path).unwrap(); fs::set_permissions(&path, fs::Permissions::from_mode(0o700)).unwrap(); @@ -1179,8 +1191,7 @@ fn publisher_requires_ack_before_forwarding_and_preserves_upstream_socket() { #[test] fn publisher_refuses_nonprivate_or_aliased_upstream_before_listening() { use std::os::unix::{fs::symlink, net::UnixListener}; - let root = PathBuf::from(format!("/tmp/hkp-private-{}", std::process::id())); - fs::create_dir(&root).unwrap(); + let root = fresh_root("hkp-private"); let path = root.join("socket"); let _listener = UnixListener::bind(&path).unwrap(); // Hold the requested port for the whole test: a publisher that reached bind() @@ -1222,12 +1233,7 @@ fn publisher_refuses_nonprivate_or_aliased_upstream_before_listening() { #[test] fn private_control_stops_only_matching_publisher_and_preserves_replaced_path() { use std::os::unix::net::{UnixDatagram, UnixListener}; - let root = PathBuf::from(format!( - "/tmp/hkc-{}-{}", - std::process::id(), - NEXT.fetch_add(1, Ordering::Relaxed) - )); - fs::create_dir(&root).unwrap(); + let root = fresh_root("hkc"); let upstream = root.join("upstream"); let _server = UnixListener::bind(&upstream).unwrap(); fs::set_permissions(&upstream, fs::Permissions::from_mode(0o700)).unwrap(); @@ -1293,12 +1299,7 @@ fn private_control_stops_only_matching_publisher_and_preserves_replaced_path() { #[test] fn control_startup_refuses_occupied_and_nonprivate_paths_without_retaining_publisher_resources() { use std::os::unix::{fs::MetadataExt, net::UnixListener}; - let root = PathBuf::from(format!( - "/tmp/hkcf-{}-{}", - std::process::id(), - NEXT.fetch_add(1, Ordering::Relaxed) - )); - fs::create_dir(&root).unwrap(); + let root = fresh_root("hkcf"); fs::set_permissions(&root, fs::Permissions::from_mode(0o700)).unwrap(); let upstream = root.join("upstream"); let _server = UnixListener::bind(&upstream).unwrap(); @@ -1453,12 +1454,7 @@ fn unix_publication_streams_and_preserves_occupied_or_replaced_frontend() { #[test] fn unix_publication_survives_a_delayed_private_upstream_ack() { - let root = PathBuf::from(format!( - "/tmp/hkr-{}-{}", - std::process::id(), - NEXT.fetch_add(1, Ordering::Relaxed) - )); - fs::create_dir(&root).unwrap(); + let root = fresh_root("hkr"); fs::set_permissions(&root, fs::Permissions::from_mode(0o700)).unwrap(); let upstream = root.join("delayed.sock"); let listener = UnixListener::bind(&upstream).unwrap();