From a3b87e8764b30662e8a3373cb79a77708df2779f Mon Sep 17 00:00:00 2001 From: Dimitri Kennedy Date: Sat, 24 Oct 2026 17:22:33 -0400 Subject: [PATCH 1/3] test(runtime-core): size release-check scans for load and pin the drop, kill-anchor and retired-format guards - Unit-test builds give the stop paths' whole-host lsof scan a named LOADED_HOST_SCAN_WAIT (60 s) instead of the 5 s production budget, which expired under load and failed release-check fixtures as provider_timeout_uncertain before their result under test. - Name Reservation::drop's lease wait (RELEASE_LEASE_WAIT), document the dead-owner recovery fallback, and prove a receipt left by an expired wait is unchanged and recoverable. - A claim writer that panics inside the lease is reported as that panic, not as a lease overlap by the writers after it. - Pin the once-only SIGKILL grace anchor and make the upgraded-node fixture differ from its live child only in boot format. - Cover the IPv6 branch of the peer probe's address rewrite. Refs HACK-1457 Co-Authored-By: Claude Opus 5.5 --- packages/runtime-core/src/node/supervisor.rs | 60 ++++- packages/runtime-core/src/node/transport.rs | 22 +- .../src/provider/graph/dependency_slots.rs | 242 ++++++++++++++++-- .../runtime-core/src/provider/lifecycle.rs | 8 + .../src/provider/lifecycle/interrupted.rs | 13 +- .../src/provider/test_executable.rs | 7 + tests/native-networks-fixture.test.ts | 44 +++- 7 files changed, 361 insertions(+), 35 deletions(-) diff --git a/packages/runtime-core/src/node/supervisor.rs b/packages/runtime-core/src/node/supervisor.rs index d5298929c..4d86f7db3 100644 --- a/packages/runtime-core/src/node/supervisor.rs +++ b/packages/runtime-core/src/node/supervisor.rs @@ -579,6 +579,13 @@ mod tests { } impl Job { fn new(timeout_ms: u64) -> Self { + Self::spawn(timeout_ms, true) + } + /// `leader: false` spawns the child in this test process's group instead of its own. + /// Its identity then records `process_group = pid`, the leadership `verify` requires, + /// so every group signal is authorized but finds no such group (`ESRCH`, tolerated): + /// the child outlives each SIGKILL, as a kill that never takes effect would. + fn spawn(timeout_ms: u64, leader: bool) -> Self { let root = crate::private_test_root( &std::env::temp_dir().canonicalize().unwrap(), "hkl-node-run", @@ -586,14 +593,16 @@ mod tests { let store = Store::open(&root, true).unwrap(); let id = "a".repeat(64); super::super::test_support::job(&store, &id, "running", timeout_ms); - let mut child = Command::new("/bin/sh") + let mut command = Command::new("/bin/sh"); + command .args(["-c", "trap '' TERM; echo ready; read line; exit 3"]) .stdin(Stdio::piped()) .stdout(Stdio::piped()) - .stderr(Stdio::piped()) - .process_group(0) - .spawn() - .unwrap(); + .stderr(Stdio::piped()); + if leader { + command.process_group(0); + } + let mut child = command.spawn().unwrap(); let stdin = child.stdin.take(); // Exactly the handshake line: run_child later drains the rest of stdout. let mut ready = [0_u8; 6]; @@ -604,7 +613,9 @@ mod tests { .read_exact(&mut ready) .unwrap(); assert_eq!(&ready, b"ready\n"); - let identity = super::super::ownership::observe(child.id() as i32).unwrap(); + let mut identity = super::super::ownership::observe(child.id() as i32).unwrap(); + assert_eq!(identity.process_group == identity.pid, leader); + identity.process_group = identity.pid; Self { root, store, @@ -800,4 +811,41 @@ mod tests { drop(job.stdin.take()); assert_eq!(job.child.wait().unwrap().code(), Some(3)); } + + /// Hang guard for a stop whose SIGKILL never takes effect. The correct path quarantines + /// about `TERMINATION_GRACE` after the first SIGKILL; only a grace that is re-anchored + /// on every later SIGKILL, and so never expires, should reach this. + const UNSETTLED_KILL_HANG_GUARD: Duration = Duration::from_secs(60); + + /// HACK-1457: the grace after this owner's first SIGKILL is anchored once. A child that + /// outlives every SIGKILL is re-authorized and signalled again on each later pass; if + /// each of those re-anchored the grace, the termination deadline would never pass. + #[test] + fn repeated_kills_of_a_child_that_never_exits_keep_the_first_kill_anchor() { + use super::super::ownership::Observed; + let mut job = Job::spawn(50, false); + let pid = job.pid(); + let recorded = job.identity.clone(); + let begun = Instant::now(); + let mut checks = 0; + let failure = job + .run(&mut |observed| { + assert_eq!(observed, pid); + assert!( + begun.elapsed() < UNSETTLED_KILL_HANG_GUARD, + "E_TEST_KILL_GRACE_REANCHORED" + ); + checks += 1; + Ok(Observed::Present(recorded.clone())) + }) + .unwrap_err(); + assert_eq!(failure.message, "Child termination deadline exceeded."); + // The SIGTERM check, the first SIGKILL, and at least the SIGKILL of the pass that + // quarantined: a re-anchoring regression had a later SIGKILL to re-anchor on. + assert!(checks >= 3, "E_TEST_KILL_NOT_REPEATED"); + // Every group signal found no group: the child still waits for its input. + assert!(!exit_published(pid)); + drop(job.stdin.take()); + assert_eq!(job.child.wait().unwrap().code(), Some(3)); + } } diff --git a/packages/runtime-core/src/node/transport.rs b/packages/runtime-core/src/node/transport.rs index 0e9474c3e..ff6c01da6 100644 --- a/packages/runtime-core/src/node/transport.rs +++ b/packages/runtime-core/src/node/transport.rs @@ -338,13 +338,27 @@ mod tests { .spawn() .unwrap(); let pid = child.id() as i32; + // The receipt differs from this live child's identity only in its boot format + // (HACK-1457): same PID, UID, group and start, so nothing but the retired format + // can be why this identity never authorizes a signal. + let live = super::super::ownership::observe(pid).unwrap(); let legacy = crate::node::ProcessIdentity { - pid, - uid: unsafe { libc::geteuid() }, - process_group: pid, - start: 1, boot: "1700000000:250000".into(), + ..live.clone() }; + assert_ne!(legacy.boot, live.boot); + // The only signal authority any owner has is `verify`, and it refuses this + // identity for its format alone: against the live child, and against an observer + // that still reports the retired format, which every other field matches. + for observed in [&live, &legacy] { + assert_eq!( + super::super::ownership::verify(&legacy, observed) + .unwrap_err() + .message, + "Owned child identity uses a retired boot format; no signal sent.", + "E_TEST_RETIRED_FORMAT_NOT_REFUSED" + ); + } let id = "b".repeat(64); super::super::test_support::job(&store, &id, "running", 60_000); store diff --git a/packages/runtime-core/src/provider/graph/dependency_slots.rs b/packages/runtime-core/src/provider/graph/dependency_slots.rs index 6fa96d81f..293690e95 100644 --- a/packages/runtime-core/src/provider/graph/dependency_slots.rs +++ b/packages/runtime-core/src/provider/graph/dependency_slots.rs @@ -423,30 +423,66 @@ fn remove_record(directory: &Path, record: &Record) -> Result<(), CandidateError .and_then(|f| f.sync_all()) .map_err(state::io) } -impl Drop for Reservation { - fn drop(&mut self) { +/// How long dropping an unadmitted [`Reservation`] waits for the provider lease to remove +/// its own receipt. Drop cannot report failure, so this bounds how long an abandoned start +/// blocks its caller. When the lease stays busy past it, or the lease, home or receipt +/// cannot be verified, the receipt is left exactly as written: its slots stay reserved +/// (never double-assigned), and the fallback that releases it is the explicit +/// [`recover_orphan`] for a never-admitted run, or dead-owner graph cleanup +/// ([`recover_cleaned`]) once this process is gone. Both require the recorded owner to +/// be dead, so a live process never loses its own reservation to them. +const RELEASE_LEASE_WAIT: Duration = Duration::from_secs(5); + +/// What [`Reservation::release_within`] did with an unadmitted reservation's receipt. +#[derive(Debug, PartialEq, Eq)] +enum Release { + /// Admitted or already released: the receipt is not this guard's to remove. + Retained, + /// The lease was taken and an exact, unbound receipt was removed. + Removed, + /// The lease was taken but the home, a socket or the receipt did not verify. + Refused, + /// The lease was still busy when the wait expired; the receipt was left for recovery. + LeaseBusy, + /// The lease could not be opened or verified; the receipt was left for recovery. + LeaseUnavailable, +} + +impl Reservation { + /// Remove this unadmitted reservation's receipt under the provider lease, waiting at + /// most `wait` for that lease. `LeaseBusy` and `LeaseUnavailable` never touch the receipt. + fn release_within(&mut self, wait: Duration) -> Release { if !self.releasable { - return; + return Release::Retained; } - let deadline = Instant::now() + Duration::from_secs(5); + let deadline = Instant::now() + wait; loop { match state::Lock::acquire_existing(&self.provider) { Ok(_lock) => { - if fs::metadata(&self.home) + let removed = fs::metadata(&self.home) .is_ok_and(|m| (m.dev(), m.ino()) == self.home_identity) - { - let _ = self.cancel_locked(); - } - return; + && self.cancel_locked().is_ok(); + self.releasable = false; + return if removed { + Release::Removed + } else { + Release::Refused + }; } Err(e) if e.code == "provider_busy" && Instant::now() < deadline => { std::thread::sleep(Duration::from_millis(20)) } - Err(_) => return, + Err(e) if e.code == "provider_busy" => return Release::LeaseBusy, + Err(_) => return Release::LeaseUnavailable, } } } } +impl Drop for Reservation { + fn drop(&mut self) { + let _ = self.release_within(RELEASE_LEASE_WAIT); + } +} fn fingerprint(record: &Record) -> Result { Ok(format!( "{:x}", @@ -777,13 +813,157 @@ mod tests { other ); } + /// A candidate whose provider holds an owner record, plus the short-home alias it names. + struct LeaseFixture { + _root: super::super::tests::Fixture, + candidate: Candidate, + provider: PathBuf, + short_home: PathBuf, + token: String, + } + impl LeaseFixture { + fn new() -> Self { + let root = super::super::tests::Fixture::new(); + let candidate = Candidate::discover(&root.0).unwrap(); + let provider = candidate.state_root.join("run/smolvm"); + state::private_directory(&provider.join("home")).unwrap(); + drop(state::Lock::acquire(&provider).unwrap()); + let token = super::super::probes::token().unwrap(); + let short_home = Path::new("/private/tmp").join(format!("hkl-{}", &token[..12])); + std::os::unix::fs::symlink(provider.join("home"), &short_home).unwrap(); + let owner: state::Owner = serde_json::from_value(serde_json::json!({ + "version":1,"checkout":candidate.checkout,"token":token, + "machine":format!("hack-{}",&token[..12]),"short_home":short_home, + "created":true,"phase":"running","process":null, + "dependency_sockets":{"slots":2},"storage":null,"overlay":null, + "guest_boot_id":null,"previous_guest_boot_id":null,"daemon_pid":null, + "daemon_start":null,"rootfs_digest":null + })) + .unwrap(); + state::write(&provider.join("owner.json"), &owner).unwrap(); + Self { + _root: root, + candidate, + provider, + short_home, + token, + } + } + /// An unadmitted reservation of `slot` whose recorded owner is a dead process, + /// as it is for recovery once the process that dropped it has exited. + fn reservation(&self, run: char, slot: u8) -> (Reservation, PathBuf) { + let mut record = fixture_record(run, slot); + record.owner = self.token.clone(); + record.process.pid = i32::MAX; + assert!(!identity::alive(record.process.pid).unwrap()); + let directory = root(&self.candidate); + store(&directory, &record); + let m = fs::metadata(&self.short_home).unwrap(); + let receipt = directory.join(format!("{}.json", record.run)); + let reservation = Reservation { + record, + directory, + provider: self.provider.clone(), + home: self.short_home.clone(), + home_identity: (m.dev(), m.ino()), + releasable: true, + }; + (reservation, receipt) + } + } + impl Drop for LeaseFixture { + fn drop(&mut self) { + let _ = fs::remove_file(&self.short_home); + } + } + /// HACK-1457: when the lease stays busy past the release wait, dropping an unadmitted + /// reservation leaves its receipt exactly as written, still reserving its slot, and + /// dead-owner recovery (the core of `recover_orphan` and `recover_cleaned`) releases it. + #[test] + fn a_release_that_outlasts_a_busy_lease_leaves_a_recoverable_receipt() { + let fixture = LeaseFixture::new(); + // A free lease: the guard removes its own unbound receipt. + let (mut free, free_receipt) = fixture.reservation('1', 0); + assert_eq!( + free.release_within(Duration::ZERO), + Release::Removed, + "E_TEST_RELEASE_NOT_REMOVED" + ); + assert!(absent(&free_receipt).unwrap(), "E_TEST_RELEASE_NOT_REMOVED"); + + let (mut busy, receipt) = fixture.reservation('2', 1); + let written = fs::read(&receipt).unwrap(); + let held = state::Lock::acquire_existing(&fixture.provider).unwrap(); + let wait = Duration::from_millis(100); + let started = Instant::now(); + assert_eq!( + busy.release_within(wait), + Release::LeaseBusy, + "E_TEST_RELEASE_IGNORED_BUSY_LEASE" + ); + assert!(started.elapsed() >= wait, "E_TEST_RELEASE_WAIT_NOT_HONORED"); + assert_eq!( + fs::read(&receipt).ok().as_deref(), + Some(&written[..]), + "E_TEST_RECEIPT_CHANGED" + ); + let reserved = records(&root(&fixture.candidate), &fixture.token, 2).unwrap(); + assert_eq!( + reserved, + vec![busy.record.clone()], + "E_TEST_SLOT_NOT_RESERVED" + ); + drop(held); + + let lease = state::Lock::acquire_existing(&fixture.provider).unwrap(); + recover_record(&fixture.candidate, &busy.record, None).unwrap(); + assert!(absent(&receipt).unwrap(), "E_TEST_RECEIPT_NOT_RECOVERED"); + drop(lease); + // A late retry never removes or recreates a receipt it no longer matches. + assert_eq!(busy.release_within(wait), Release::Refused); + assert!(absent(&receipt).unwrap()); + } /// Hang guard for the claim-writer test, not a correctness bound. A writer can /// wait behind seven fsync-backed critical sections whose latency is unbounded on /// a loaded host (HACK-1430: a 3 s per-writer budget expired at load average /// ~190), so only a lease that is never released should reach this limit. const CLAIM_LEASE_HANG_GUARD: Duration = Duration::from_secs(120); + /// Counts one writer inside the claim lease. Declared after the lease guard, it drops + /// first, so a writer that panics inside the lease has left the count before the next + /// writer can take the lease, and that writer does not report the panic as an overlap. + struct InsideLease<'a>(&'a std::sync::atomic::AtomicUsize); + impl Drop for InsideLease<'_> { + fn drop(&mut self) { + self.0.fetch_sub(1, std::sync::atomic::Ordering::SeqCst); + } + } + /// The panic message a writer thread ended with, or a closed placeholder. + fn panic_message(payload: &(dyn std::any::Any + Send)) -> &str { + payload + .downcast_ref::<&str>() + .copied() + .or_else(|| payload.downcast_ref::().map(String::as_str)) + .unwrap_or("E_TEST_CLAIM_WRITER_PANIC_UNPRINTABLE") + } #[test] fn concurrent_claim_writers_serialize_without_overlapping_slots() { + let slots = claim_writers(None).unwrap_or_else(|failure| panic!("{failure}")); + assert_eq!(slots, (0..8).collect()); + } + /// HACK-1457: a writer that panics inside the lease is reported as that panic. Before, + /// it left the inside count raised, and the next writer failed as a lease overlap. + #[test] + fn a_writer_panic_inside_the_lease_names_itself_not_an_overlap() { + let failure = claim_writers(Some(0)).unwrap_err(); + assert_eq!( + failure, "E_TEST_CLAIM_WRITER_PANICKED (writer 0: E_TEST_INJECTED_WRITER_FAULT)", + "E_TEST_CLAIM_FAILURE_MISATTRIBUTED" + ); + } + /// Run eight contending claim writers; `fault` names a writer that panics inside the + /// lease. Returns the slots claimed, or every writer panic in writer order, so a + /// failure names its own cause rather than a symptom it left for other writers. + fn claim_writers(fault: Option) -> Result, String> { use std::sync::Arc; use std::sync::atomic::{AtomicUsize, Ordering::SeqCst}; const WRITERS: u8 = 8; @@ -815,21 +995,29 @@ mod tests { assert!(Instant::now() < deadline, "E_TEST_CLAIM_LEASE_HANG"); std::thread::sleep(Duration::from_millis(1)); } - Err(e) => panic!("unexpected reservation lock refusal: {}", e.code), + Err(e) => panic!("E_TEST_CLAIM_LEASE_REFUSED: {}", e.code), } }; assert!(was_refused, "E_TEST_CLAIM_WRITER_UNCONTENDED"); // The lease, not the slot selection, must exclude other writers. - assert_eq!(inside.fetch_add(1, SeqCst), 0, "E_TEST_CLAIM_LEASE_OVERLAP"); + let entered = inside.fetch_add(1, SeqCst); + let in_lease = InsideLease(&inside); + assert_eq!(entered, 0, "E_TEST_CLAIM_LEASE_OVERLAP"); + if fault == Some(index) { + panic!("E_TEST_INJECTED_WRITER_FAULT"); + } let mut record = fixture_record(char::from(b'1' + index), 0); - let existing = records(&directory, &record.owner, WRITERS).unwrap(); + let existing = records(&directory, &record.owner, WRITERS) + .unwrap_or_else(|e| panic!("E_TEST_CLAIM_RECORDS_REFUSED: {}", e.code)); let used = existing .iter() .flat_map(|r| r.slots.values().copied()) .collect(); - record.slots = select(&BTreeSet::from([0]), &used, WRITERS, true).unwrap(); + record.slots = select(&BTreeSet::from([0]), &used, WRITERS, true) + .unwrap_or_else(|e| panic!("E_TEST_CLAIM_SELECT_REFUSED: {}", e.code)); store(&directory, &record); - assert_eq!(inside.fetch_sub(1, SeqCst), 1, "E_TEST_CLAIM_LEASE_OVERLAP"); + assert_eq!(inside.load(SeqCst), 1, "E_TEST_CLAIM_LEASE_OVERLAP"); + drop(in_lease); drop(lock); record.slots[&0] })); @@ -846,16 +1034,30 @@ mod tests { std::thread::sleep(Duration::from_millis(1)); } drop(held); - let slots = threads - .into_iter() - .map(|t| t.join().unwrap()) - .collect::>(); + let mut slots = BTreeSet::new(); + let mut panics = Vec::new(); + for (index, thread) in threads.into_iter().enumerate() { + match thread.join() { + Ok(slot) => { + slots.insert(slot); + } + Err(payload) => { + panics.push(format!("writer {index}: {}", panic_message(&*payload))) + } + } + } + if !panics.is_empty() { + return Err(format!( + "E_TEST_CLAIM_WRITER_PANICKED ({})", + panics.join("; ") + )); + } assert!(all_refused, "E_TEST_CLAIM_WRITERS_NOT_REFUSED"); - assert_eq!(slots, (0..WRITERS).collect()); assert_eq!( records(&directory, &"a".repeat(32), WRITERS).unwrap().len(), usize::from(WRITERS) ); + Ok(slots) } #[test] fn legacy_exact_selection_cannot_steal_unbound_durable_reservation() { diff --git a/packages/runtime-core/src/provider/lifecycle.rs b/packages/runtime-core/src/provider/lifecycle.rs index 41458e9a3..f69acf4a7 100644 --- a/packages/runtime-core/src/provider/lifecycle.rs +++ b/packages/runtime-core/src/provider/lifecycle.rs @@ -1998,7 +1998,15 @@ impl Drop for VmLock { } /// How long `down`, `recover` and other stop paths wait for `lsof` over the owned disks. +#[cfg(not(test))] const STOP_HANDLE_WAIT: Duration = Duration::from_secs(5); +/// The unit-test build of [`STOP_HANDLE_WAIT`]. Release-check fixtures prove what the scan +/// reports (status 1, no output), not how fast it reports it, and they run beside other +/// tests' whole-host scans; on a loaded host one scan took longer than 5 s and the +/// fixture failed as `provider_timeout_uncertain` before its result under test (HACK-1457). +/// Tests of the timeout itself pass their own bound to [`lock_absent_disks_within`]. +#[cfg(test)] +const STOP_HANDLE_WAIT: Duration = super::test_executable::LOADED_HOST_SCAN_WAIT; /// Keep this guard through alias restoration and the stopped receipt. fn lock_absent_disks(candidate: &Candidate, owner: &Owner) -> Result { diff --git a/packages/runtime-core/src/provider/lifecycle/interrupted.rs b/packages/runtime-core/src/provider/lifecycle/interrupted.rs index f0b4986f0..c3bf4f4e4 100644 --- a/packages/runtime-core/src/provider/lifecycle/interrupted.rs +++ b/packages/runtime-core/src/provider/lifecycle/interrupted.rs @@ -302,8 +302,19 @@ mod tests { assert!(owner.storage.is_none() && owner.overlay.is_none()); // Recording identifies the disks now, which these non-ext4 fixture files fail; the next // boot's allocation and format checks are what should decide instead. + // The error is the identification itself (a 4-byte fixture has no superblock to read), + // not an earlier release check such as a slow disk-handle scan (HACK-1457). let mut recording = pool.owner("booting", true, false); - assert!(finish_absent(&pool.candidate, &mut recording, "recovered-unclean", true).is_err()); + let error = + finish_absent(&pool.candidate, &mut recording, "recovered-unclean", true).unwrap_err(); + assert_eq!( + error.code, "provider_state", + "E_TEST_RECORDING_REFUSED_EARLY" + ); + assert!( + recording.storage.is_none(), + "E_TEST_RECORDING_REFUSED_EARLY" + ); } #[test] diff --git a/packages/runtime-core/src/provider/test_executable.rs b/packages/runtime-core/src/provider/test_executable.rs index 6acd16d09..567382d8e 100644 --- a/packages/runtime-core/src/provider/test_executable.rs +++ b/packages/runtime-core/src/provider/test_executable.rs @@ -44,6 +44,13 @@ pub(super) fn sleeping_executable(binary: &Path) { fs::set_permissions(binary, fs::Permissions::from_mode(0o700)).unwrap(); } +/// Test bound for one whole-host `lsof` scan. `lsof` walks every process on the host, so +/// on a loaded machine with other tests scanning at the same time one scan can take well +/// over its production budget (HACK-1457: over 5 s at load average ~80-120 with process +/// spawn churn). Only a scan that never finishes should reach this; it stays well inside +/// [`ISOLATED_HANG_GUARD`], so an isolated fixture still reports its own failure. +pub(super) const LOADED_HOST_SCAN_WAIT: Duration = Duration::from_secs(60); + /// Hang guard for [`run_isolated`]: the child launches the whole test binary on a host that /// may be heavily loaded, so this bounds a hang rather than the fixture's own timing. pub(super) const ISOLATED_HANG_GUARD: Duration = Duration::from_secs(120); diff --git a/tests/native-networks-fixture.test.ts b/tests/native-networks-fixture.test.ts index 146b5c0f9..1f33eae0d 100644 --- a/tests/native-networks-fixture.test.ts +++ b/tests/native-networks-fixture.test.ts @@ -524,9 +524,10 @@ test("peer probe attributes a failure to one URL and phase in closed stderr evid /** * Runs the peer probe with `Bun.dns.lookup` replaced by a stub that answers - * every name with 127.0.0.1, after `delayMs` for names starting with `slow`. - * The probe names use RFC 6761 `.invalid`, so only the stubbed answer can - * reach the local listener: a second, real resolution fails (HACK-1428). + * every name with `answer` (default 127.0.0.1, family 4), after `delayMs` for + * names starting with `slow`. The probe names use RFC 6761 `.invalid`, so only + * the stubbed answer can reach the local listener: a second, real resolution + * fails (HACK-1428). */ async function runStubbedPeerProbe(opts: { readonly root: string; @@ -534,8 +535,12 @@ async function runStubbedPeerProbe(opts: { readonly alias: string; readonly timeoutMs: number; readonly delayMs: number; + readonly answer?: { readonly address: string; readonly family: 4 | 6 }; }): Promise<{ readonly code: number; readonly stderr: string }> { - const stub = `Bun.dns.lookup=async(h)=>{if(h.startsWith("slow"))await Bun.sleep(${opts.delayMs});return [{address:"127.0.0.1",family:4}]};`; + const answer = JSON.stringify([ + opts.answer ?? { address: "127.0.0.1", family: 4 }, + ]); + const stub = `Bun.dns.lookup=async(h)=>{if(h.startsWith("slow"))await Bun.sleep(${opts.delayMs});return ${answer}};`; const child = Bun.spawn( [ process.execPath, @@ -599,6 +604,37 @@ test("peer probe fetches the address it just resolved", async () => { } }); +// HACK-1457: an IPv6-only answer is bracketed into the URL host. Without the +// brackets the URL rejects the bare address, keeps the unresolvable name, and the +// fetch fails. +test("peer probe fetches an IPv6-only answer through a bracketed host", async () => { + const root = await mkdtemp(join(tmpdir(), "hack-network-probe-")); + const hosts: (string | null)[] = []; + const server = Bun.serve({ + hostname: "::1", + port: 0, + fetch(request) { + hosts.push(request.headers.get("host")); + return new Response("primary:peer"); + }, + }); + try { + const result = await runStubbedPeerProbe({ + root, + origin: `http://probe-1457.invalid:${server.port}`, + alias: `http://probe-1457-alias.invalid:${server.port}/a`, + timeoutMs: 4000, + delayMs: 0, + answer: { address: "::1", family: 6 }, + }); + expect(result).toEqual({ code: 0, stderr: "" }); + expect(hosts).toEqual([`[::1]:${server.port}`, `[::1]:${server.port}`]); + } finally { + server.stop(true); + await rm(root, { recursive: true, force: true }); + } +}); + // HACK-1428: resolution time is charged against the URL's one shared budget. test("peer probe charges resolution time against the shared per-URL budget", async () => { const root = await mkdtemp(join(tmpdir(), "hack-network-probe-")); From 8d591dbf8c7f067674ae9b59ab7e2c2349087c3f Mon Sep 17 00:00:00 2001 From: Dimitri Kennedy Date: Sat, 24 Oct 2026 18:31:24 -0400 Subject: [PATCH 2/3] test(runtime-core): give private_child's fixture-shell waits a named hang guard A 2 s wait for a spawned /bin/sh to exit or print its marker expired at load average ~230 during the HACK-1457 load runs. It is a hang guard, not a correctness bound, so it becomes FIXTURE_SHELL_HANG_GUARD (60 s) with a closed E_TEST_FIXTURE_SHELL_HANG message. Refs HACK-1457 Co-Authored-By: Claude Opus 5.5 --- .../src/provider/lifecycle/private_child.rs | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/packages/runtime-core/src/provider/lifecycle/private_child.rs b/packages/runtime-core/src/provider/lifecycle/private_child.rs index 83964eb55..e0fa4cfee 100644 --- a/packages/runtime-core/src/provider/lifecycle/private_child.rs +++ b/packages/runtime-core/src/provider/lifecycle/private_child.rs @@ -510,6 +510,10 @@ printf 'relay-guest-absent-v1\n' #[cfg(test)] mod tests { use super::*; + /// Hang guard for a fixture shell to exit or print its marker, not a correctness bound: + /// on a loaded host spawning `/bin/sh` alone took longer than the former 2 s (HACK-1457, + /// load average ~230). Only a fixture that never exits or never prints should reach it. + const FIXTURE_SHELL_HANG_GUARD: Duration = Duration::from_secs(60); #[test] fn marker_is_exact_canonical_and_port_bound() { assert_eq!( @@ -661,9 +665,9 @@ mod tests { ) .unwrap(); if command == "exit 7" { - let deadline = Instant::now() + Duration::from_secs(2); + let deadline = Instant::now() + FIXTURE_SHELL_HANG_GUARD; while handle.poll_exit().unwrap().is_none() { - assert!(Instant::now() < deadline); + assert!(Instant::now() < deadline, "E_TEST_FIXTURE_SHELL_HANG"); std::thread::sleep(Duration::from_millis(2)); } assert_eq!(handle.status.unwrap().code(), Some(7)); @@ -708,12 +712,12 @@ mod tests { assert_eq!(handle.boot, "boot"); assert_eq!(handle.slot, 7); assert_eq!(handle.stop_user(), "1001:1002"); - let deadline = Instant::now() + Duration::from_secs(2); + let deadline = Instant::now() + FIXTURE_SHELL_HANG_GUARD; let ready = loop { if let Some(ready) = handle.poll_ready().unwrap() { break ready; } - assert!(Instant::now() < deadline); + assert!(Instant::now() < deadline, "E_TEST_FIXTURE_SHELL_HANG"); std::thread::sleep(Duration::from_millis(2)); }; assert_eq!(ready.pid, 42); From 6508480b22aa0ea84e3b0767daeb642078e2e113 Mon Sep 17 00:00:00 2001 From: Dimitri Kennedy Date: Sat, 24 Oct 2026 22:18:08 -0400 Subject: [PATCH 3/3] test(runtime-core): bound relay, endpoint and CPU tests by named guards and pick unallocated ports The second round of HACK-1457 load fixes. The CPU counter test now spins for 40 ms of thread CPU and brackets the native read between an inner and an outer process-clock window, so parallel CPU use cannot fail it. HTTPS recovery and discovery fixtures take a port below the kernel's ephemeral range, which no port-0 bind or outgoing connection can take before the fixture needs it. host_endpoint, relay_loop and https_recovery tests replace their fixed 1-6 s waits with LOADED_EVENT_HANG_GUARD, MAX_CONNECT_BUDGET or ISOLATED_HANG_GUARD; expiry tests still force an expired deadline. A new test covers the Reservation Drop wiring. Refs HACK-1457 Co-Authored-By: Claude Opus 5.5 --- .../src/provider/graph/dependency_slots.rs | 21 +++ .../src/provider/host_endpoint/tests.rs | 144 ++++++++++-------- .../src/provider/https_recovery/tests.rs | 16 +- .../provider/relay_loop/admission_tests.rs | 81 ++++------ .../src/provider/relay_loop/tests.rs | 38 +++-- .../runtime-core/src/provider/resources.rs | 55 +++++-- .../provider/shared_https_recovery/tests.rs | 4 +- .../src/provider/test_executable.rs | 17 +++ .../runtime-core/src/provider/test_socket.rs | 97 ++++++++++++ 9 files changed, 313 insertions(+), 160 deletions(-) diff --git a/packages/runtime-core/src/provider/graph/dependency_slots.rs b/packages/runtime-core/src/provider/graph/dependency_slots.rs index 293690e95..b73112bf4 100644 --- a/packages/runtime-core/src/provider/graph/dependency_slots.rs +++ b/packages/runtime-core/src/provider/graph/dependency_slots.rs @@ -923,6 +923,27 @@ mod tests { assert_eq!(busy.release_within(wait), Release::Refused); assert!(absent(&receipt).unwrap()); } + /// Dropping the guard is what releases an abandoned start: `Drop` runs the bounded + /// release, and an admitted reservation's receipt stays for its running graph. + #[test] + fn dropping_an_unadmitted_reservation_releases_its_receipt_under_a_free_lease() { + let fixture = LeaseFixture::new(); + let (abandoned, abandoned_receipt) = fixture.reservation('3', 0); + drop(abandoned); + assert!( + absent(&abandoned_receipt).unwrap(), + "E_TEST_DROP_NOT_RELEASED" + ); + let (mut admitted, admitted_receipt) = fixture.reservation('4', 1); + let written = fs::read(&admitted_receipt).unwrap(); + admitted.admitted(); + drop(admitted); + assert_eq!( + fs::read(&admitted_receipt).ok().as_deref(), + Some(&written[..]), + "E_TEST_DROP_RELEASED_ADMITTED" + ); + } /// Hang guard for the claim-writer test, not a correctness bound. A writer can /// wait behind seven fsync-backed critical sections whose latency is unbounded on /// a loaded host (HACK-1430: a 3 s per-writer budget expired at load average diff --git a/packages/runtime-core/src/provider/host_endpoint/tests.rs b/packages/runtime-core/src/provider/host_endpoint/tests.rs index 8776eb135..c99a32494 100644 --- a/packages/runtime-core/src/provider/host_endpoint/tests.rs +++ b/packages/runtime-core/src/provider/host_endpoint/tests.rs @@ -9,6 +9,11 @@ use std::{ time::Instant, }; +#[cfg(target_os = "macos")] +use crate::provider::test_executable::{ + ISOLATED_HANG_GUARD, LOADED_EVENT_HANG_GUARD, MAX_CONNECT_BUDGET, +}; + #[cfg(target_os = "macos")] fn fixture() -> (TcpListener, HostEndpoint) { let listener = TcpListener::bind("127.0.0.1:0").unwrap(); @@ -112,9 +117,8 @@ fn pinned_discovery_child() { } } let executable = std::env::current_exe().unwrap().canonicalize().unwrap(); - let reserved = TcpListener::bind("127.0.0.1:0").unwrap(); - let port = reserved.local_addr().unwrap().port(); - drop(reserved); + // The listener children bind this port long after it is chosen; never an ephemeral one. + let port = crate::provider::test_socket::unallocated_tcp_port(); let spawn = || { OwnedChild( Command::new(&executable) @@ -133,7 +137,8 @@ fn pinned_discovery_child() { ) }; let await_discovery = |child: &mut OwnedChild| { - let deadline = Instant::now() + Duration::from_secs(5); + // The child is a whole test binary starting on a possibly loaded host. + let deadline = Instant::now() + LOADED_EVENT_HANG_GUARD; loop { if let Ok(found) = HostEndpoint::discover(&executable, port) { assert_eq!(found.0, child.0.id() as i32); @@ -203,7 +208,7 @@ fn identity_notifications_require_authenticated_checks_and_ignore_revocation_or_ })); let (authority, session) = authorized(&endpoint); let mut pending = endpoint - .begin_connect(Duration::from_secs(1), &session) + .begin_connect(MAX_CONNECT_BUDGET, &session) .unwrap(); pending.deadline = Instant::now(); assert_eq!( @@ -216,14 +221,14 @@ fn identity_notifications_require_authenticated_checks_and_ignore_revocation_or_ assert_eq!(failures.load(Ordering::Relaxed), 0); assert!( endpoint - .begin_connect(Duration::from_secs(1), &session) + .begin_connect(MAX_CONNECT_BUDGET, &session) .is_err() ); assert_eq!(failures.load(Ordering::Relaxed), 1); authority.revoke(); assert!( endpoint - .begin_connect(Duration::from_secs(1), &session) + .begin_connect(MAX_CONNECT_BUDGET, &session) .is_err() ); assert_eq!(failures.load(Ordering::Relaxed), 1); @@ -253,9 +258,7 @@ fn pending_accepted_identity_notification_child() { counter.fetch_add(1, Ordering::Relaxed); })); let (_authority, session) = authorized(&endpoint); - let mut pending = endpoint - .begin_connect(Duration::from_secs(3), &session) - .unwrap(); + let mut pending = begin_unexpiring(&endpoint, &session); let mut peer = accept_pending(&listener); let peer_port = pending .stream @@ -280,7 +283,8 @@ fn pending_accepted_identity_notification_child() { assert!(pending.progress().is_err()); assert_eq!(failures.load(Ordering::Relaxed), 1); peer.set_nonblocking(false).unwrap(); - peer.set_read_timeout(Some(Duration::from_secs(1))).unwrap(); + peer.set_read_timeout(Some(LOADED_EVENT_HANG_GUARD)) + .unwrap(); assert_eq!(peer.read(&mut [0; 1]).unwrap(), 0); } @@ -293,7 +297,8 @@ fn discovery_listener_child() { .parse() .unwrap(); let _listener = TcpListener::bind((std::net::Ipv4Addr::LOCALHOST, port)).unwrap(); - thread::sleep(Duration::from_secs(30)); + // Outlive the parent's whole watchdog; the parent kills this child when done. + thread::sleep(ISOLATED_HANG_GUARD); } #[cfg(target_os = "macos")] @@ -303,7 +308,7 @@ fn accepted_stream_preserves_payload_and_half_close() { listener.set_nonblocking(true).unwrap(); thread::scope(|scope| { let server = scope.spawn(|| { - let deadline = Instant::now() + Duration::from_secs(3); + let deadline = Instant::now() + LOADED_EVENT_HANG_GUARD; let mut peer = loop { match listener.accept() { Ok((peer, _)) => break peer, @@ -315,15 +320,16 @@ fn accepted_stream_preserves_payload_and_half_close() { } }; peer.set_nonblocking(false).unwrap(); - peer.set_read_timeout(Some(Duration::from_secs(2))).unwrap(); + peer.set_read_timeout(Some(LOADED_EVENT_HANG_GUARD)) + .unwrap(); let mut body = String::new(); peer.read_to_string(&mut body).unwrap(); assert_eq!(body, "synthetic endpoint payload"); peer.write_all(b"reply").unwrap(); }); - let mut stream = endpoint.connect(Duration::from_secs(2)).unwrap(); + let mut stream = endpoint.connect(MAX_CONNECT_BUDGET).unwrap(); stream - .set_read_timeout(Some(Duration::from_secs(2))) + .set_read_timeout(Some(LOADED_EVENT_HANG_GUARD)) .unwrap(); stream.write_all(b"synthetic endpoint payload").unwrap(); stream.shutdown(std::net::Shutdown::Write).unwrap(); @@ -346,12 +352,18 @@ fn pending_accept_times_out_without_releasing_payload() { .code, "host_endpoint_timeout" ); - assert!(start.elapsed() < Duration::from_secs(2)); + // Not a latency bound: returning before the largest budget the API accepts still + // shows the 80 ms budget was honored rather than replaced by that maximum, and it + // leaves room for a loaded host's scheduling delay (HACK-1457). + assert!( + start.elapsed() < MAX_CONNECT_BUDGET, + "E_TEST_CONNECT_BUDGET_NOT_HONORED" + ); listener.set_nonblocking(true).unwrap(); let (mut stream, _) = listener.accept().unwrap(); stream.set_nonblocking(false).unwrap(); stream - .set_read_timeout(Some(Duration::from_secs(1))) + .set_read_timeout(Some(LOADED_EVENT_HANG_GUARD)) .unwrap(); let mut data = Vec::new(); stream.read_to_end(&mut data).unwrap(); @@ -452,12 +464,13 @@ fn precheck_replacement_child() { let (listener, endpoint) = fixture(); let address = listener.local_addr().unwrap(); let mut replacement = None; - let result = endpoint.connect_after_check(Duration::from_secs(1), || { + let result = endpoint.connect_after_check(MAX_CONNECT_BUDGET, || { drop(listener); // Parallel subprocess creation can briefly retain a forked copy of the // closed listener until exec. Require a real replacement before connect; // a collision or timeout must never count as the expected identity refusal. - let deadline = Instant::now() + Duration::from_millis(250); + // Waiting longer than the connect budget would expire the connect anyway. + let deadline = Instant::now() + MAX_CONNECT_BUDGET; replacement = Some(loop { match TcpListener::bind(address) { Ok(listener) => break listener, @@ -476,7 +489,8 @@ fn precheck_replacement_child() { replacement.set_nonblocking(true).unwrap(); let mut peer = accept_pending(&replacement); peer.set_nonblocking(false).unwrap(); - peer.set_read_timeout(Some(Duration::from_secs(1))).unwrap(); + peer.set_read_timeout(Some(LOADED_EVENT_HANG_GUARD)) + .unwrap(); let mut data = Vec::new(); peer.read_to_end(&mut data).unwrap(); assert!(data.is_empty()); @@ -484,7 +498,7 @@ fn precheck_replacement_child() { #[cfg(target_os = "macos")] fn accept_pending(listener: &TcpListener) -> TcpStream { - let deadline = Instant::now() + Duration::from_secs(1); + let deadline = Instant::now() + LOADED_EVENT_HANG_GUARD; loop { match listener.accept() { Ok((stream, _)) => return stream, @@ -497,6 +511,18 @@ fn accept_pending(listener: &TcpListener) -> TcpStream { } } +/// Begin a pending connection for a test whose result is not its deadline's expiry. The +/// deadline moves out to a hang guard, a value the API itself refuses (it accepts at most +/// 5 s), so a loaded host's scheduling delay cannot expire it before the test's next check +/// and turn the result under test into `host_endpoint_timeout` (HACK-1457). Expiry tests +/// set an expired deadline themselves. +#[cfg(target_os = "macos")] +fn begin_unexpiring(endpoint: &HostEndpoint, session: &AuthorizedSession) -> PendingConnection { + let mut pending = endpoint.begin_connect(MAX_CONNECT_BUDGET, session).unwrap(); + pending.deadline = Instant::now() + LOADED_EVENT_HANG_GUARD; + pending +} + #[cfg(target_os = "macos")] fn authorization() -> (super::super::relay_auth::Authority, AuthorizedSession) { use super::super::relay_auth::{Authority, Binding, Credential}; @@ -524,9 +550,7 @@ fn authorization() -> (super::super::relay_auth::Authority, AuthorizedSession) { fn pending_connection_releases_only_after_accept_and_is_nonblocking() { let (listener, endpoint) = fixture(); let (_authority, session) = authorization(); - let mut pending = endpoint - .begin_connect(Duration::from_secs(2), &session) - .unwrap(); + let mut pending = begin_unexpiring(&endpoint, &session); assert!(pending.progress().unwrap().is_none()); listener.set_nonblocking(true).unwrap(); let mut peer = accept_pending(&listener); @@ -541,7 +565,8 @@ fn pending_connection_releases_only_after_accept_and_is_nonblocking() { .unwrap() .unwrap(); peer.set_nonblocking(false).unwrap(); - peer.set_read_timeout(Some(Duration::from_secs(1))).unwrap(); + peer.set_read_timeout(Some(LOADED_EVENT_HANG_GUARD)) + .unwrap(); let mut data = [0; 8]; peer.read_exact(&mut data).unwrap(); assert_eq!(&data, b"verified"); @@ -556,7 +581,7 @@ fn revoked_connection_never_starts_and_pending_revoke_closes_without_data() { authority.revoke(); assert!( endpoint - .begin_connect(Duration::from_secs(1), &session) + .begin_connect(MAX_CONNECT_BUDGET, &session) .is_err() ); assert_eq!( @@ -564,15 +589,14 @@ fn revoked_connection_never_starts_and_pending_revoke_closes_without_data() { std::io::ErrorKind::WouldBlock ); let (authority, session) = authorization(); - let mut pending = endpoint - .begin_connect(Duration::from_secs(1), &session) - .unwrap(); + let mut pending = begin_unexpiring(&endpoint, &session); let mut peer = accept_pending(&listener); authority.revoke(); assert_eq!(pending.progress().err().unwrap().code, "relay_auth_refused"); assert!(pending.stream.is_none()); peer.set_nonblocking(false).unwrap(); - peer.set_read_timeout(Some(Duration::from_secs(1))).unwrap(); + peer.set_read_timeout(Some(LOADED_EVENT_HANG_GUARD)) + .unwrap(); assert_eq!(peer.read(&mut [0]).unwrap(), 0); } @@ -583,9 +607,7 @@ fn pending_timeout_and_drop_release_the_socket_without_data() { listener.set_nonblocking(true).unwrap(); let (_authority, session) = authorization(); for expire in [true, false] { - let mut pending = endpoint - .begin_connect(Duration::from_secs(1), &session) - .unwrap(); + let mut pending = begin_unexpiring(&endpoint, &session); let mut peer = accept_pending(&listener); if expire { pending.deadline = Instant::now(); @@ -597,7 +619,8 @@ fn pending_timeout_and_drop_release_the_socket_without_data() { } drop(pending); peer.set_nonblocking(false).unwrap(); - peer.set_read_timeout(Some(Duration::from_secs(1))).unwrap(); + peer.set_read_timeout(Some(LOADED_EVENT_HANG_GUARD)) + .unwrap(); assert_eq!(peer.read(&mut [0]).unwrap(), 0); } assert!(endpoint.begin_connect(Duration::ZERO, &session).is_err()); @@ -621,7 +644,7 @@ fn pending_replacement_cannot_release_stream() { /// [`HostEndpoint::discover`]. Close/rebind and discovery fixtures therefore run isolated. #[cfg(target_os = "macos")] fn isolated_endpoint_fixture(test: &str) { - use crate::provider::test_executable::{ISOLATED_HANG_GUARD, run_isolated}; + use crate::provider::test_executable::run_isolated; run_isolated(test, ISOLATED_HANG_GUARD); } #[cfg(target_os = "macos")] @@ -631,13 +654,11 @@ fn pending_replacement_child() { let (listener, endpoint) = fixture(); let (_authority, session) = authorization(); let address = listener.local_addr().unwrap(); - let mut pending = endpoint - .begin_connect(Duration::from_secs(1), &session) - .unwrap(); + let mut pending = begin_unexpiring(&endpoint, &session); // The nonblocking handshake may still be in flight. Let it reach the original listener // first: completed later, it lands on the replacement, which the endpoint refuses but // which leaves that connection queued there. - let deadline = Instant::now() + Duration::from_secs(1); + let deadline = Instant::now() + LOADED_EVENT_HANG_GUARD; while pending.stream.as_ref().unwrap().peer_addr().is_err() { assert!( Instant::now() < deadline, @@ -668,7 +689,8 @@ fn pending_socket_survives_closed_peer_with_default_sigpipe() { .args(["--exact", "provider::host_endpoint::tests::pending_socket_survives_closed_peer_with_default_sigpipe", "--test-threads=1"]) .env_clear().env(HELPER,mode) .stdin(std::process::Stdio::null()).stdout(std::process::Stdio::null()).spawn().unwrap(); - let deadline = Instant::now() + Duration::from_secs(6); + // The child launches the whole test binary on a possibly loaded host. + let deadline = Instant::now() + ISOLATED_HANG_GUARD; loop { if let Some(status) = child.try_wait().unwrap() { if mode == "unprotected" { @@ -694,9 +716,7 @@ fn pending_socket_survives_closed_peer_with_default_sigpipe() { let (listener, endpoint) = fixture(); listener.set_nonblocking(true).unwrap(); let (_authority, session) = authorization(); - let mut pending = endpoint - .begin_connect(Duration::from_secs(2), &session) - .unwrap(); + let mut pending = begin_unexpiring(&endpoint, &session); let peer = accept_pending(&listener); let stream = pending.progress().unwrap().unwrap(); if std::env::var(HELPER).unwrap() == "unprotected" { @@ -739,7 +759,7 @@ fn pending_socket_survives_closed_peer_with_default_sigpipe() { unsafe { libc::signal(libc::SIGPIPE, libc::SIG_DFL) }, libc::SIG_ERR ); - let deadline = Instant::now() + Duration::from_secs(1); + let deadline = Instant::now() + LOADED_EVENT_HANG_GUARD; loop { let byte = [42_u8]; // SAFETY: live stream descriptor and readable one-byte buffer. A raw write @@ -762,9 +782,7 @@ fn revocation_readiness_wakes_handler_and_closes_pending_socket() { listener.set_nonblocking(true).unwrap(); let (authority, session) = authorization(); let watch = session.watch_revocation().unwrap(); - let pending = endpoint - .begin_connect(Duration::from_secs(2), &session) - .unwrap(); + let pending = begin_unexpiring(&endpoint, &session); let mut peer = accept_pending(&listener); thread::scope(|scope| { let (ready_tx, ready_rx) = std::sync::mpsc::channel(); @@ -777,16 +795,18 @@ fn revocation_readiness_wakes_handler_and_closes_pending_socket() { ready_tx.send(()).unwrap(); // SAFETY: a live owned watch keeps its initialized pollfd valid through // this bounded call; the kernel does not retain the pointer. - assert_eq!(unsafe { libc::poll(&mut fd, 1, 2000) }, 1); + let guard = LOADED_EVENT_HANG_GUARD.as_millis() as libc::c_int; + assert_eq!(unsafe { libc::poll(&mut fd, 1, guard) }, 1); assert!(watch.is_revoked().unwrap()); drop(pending); }); - ready_rx.recv_timeout(Duration::from_secs(1)).unwrap(); + ready_rx.recv_timeout(LOADED_EVENT_HANG_GUARD).unwrap(); authority.revoke(); handler.join().unwrap(); }); peer.set_nonblocking(false).unwrap(); - peer.set_read_timeout(Some(Duration::from_secs(1))).unwrap(); + peer.set_read_timeout(Some(LOADED_EVENT_HANG_GUARD)) + .unwrap(); assert_eq!(peer.read(&mut [0]).unwrap(), 0); assert!(session.with_active(|| ()).is_err()); } @@ -796,9 +816,7 @@ fn revocation_readiness_wakes_handler_and_closes_pending_socket() { fn pending_accept_after_backend_fin_preserves_request_direction() { let (listener, endpoint) = fixture(); let (_authority, session) = authorization(); - let mut pending = endpoint - .begin_connect(Duration::from_secs(2), &session) - .unwrap(); + let mut pending = begin_unexpiring(&endpoint, &session); listener.set_nonblocking(true).unwrap(); let mut peer = accept_pending(&listener); peer.set_nonblocking(false).unwrap(); @@ -806,7 +824,7 @@ fn pending_accept_after_backend_fin_preserves_request_direction() { // Observe FIN before identity inspection, without releasing caller payload. let raw = pending.stream.as_mut().unwrap(); raw.set_nonblocking(false).unwrap(); - raw.set_read_timeout(Some(Duration::from_secs(1))).unwrap(); + raw.set_read_timeout(Some(LOADED_EVENT_HANG_GUARD)).unwrap(); assert_eq!(raw.read(&mut [0]).unwrap(), 0); raw.set_nonblocking(true).unwrap(); let mut stream = pending @@ -817,7 +835,8 @@ fn pending_accept_after_backend_fin_preserves_request_direction() { .with_active(|| stream.write_all(b"request")) .unwrap() .unwrap(); - peer.set_read_timeout(Some(Duration::from_secs(1))).unwrap(); + peer.set_read_timeout(Some(LOADED_EVENT_HANG_GUARD)) + .unwrap(); let mut received = [0; 7]; peer.read_exact(&mut received).unwrap(); assert_eq!(&received, b"request"); @@ -835,19 +854,18 @@ fn backend_fin_does_not_bypass_listener_replacement() { fn backend_fin_replacement_child() { let (listener, endpoint) = fixture(); let (_authority, session) = authorization(); - let mut pending = endpoint - .begin_connect(Duration::from_secs(2), &session) - .unwrap(); + let mut pending = begin_unexpiring(&endpoint, &session); listener.set_nonblocking(true).unwrap(); let mut peer = accept_pending(&listener); peer.set_nonblocking(false).unwrap(); - peer.set_read_timeout(Some(Duration::from_secs(1))).unwrap(); + peer.set_read_timeout(Some(LOADED_EVENT_HANG_GUARD)) + .unwrap(); peer.shutdown(std::net::Shutdown::Write).unwrap(); let port = listener.local_addr().unwrap().port(); drop(listener); // Parallel subprocess fixtures can briefly inherit the closing listener // during fork/exec. Require a real replacement within a bounded window. - let deadline = Instant::now() + Duration::from_millis(500); + let deadline = Instant::now() + LOADED_EVENT_HANG_GUARD; let _replacement = loop { match TcpListener::bind(("127.0.0.1", port)) { Ok(listener) => break listener, @@ -869,9 +887,7 @@ fn backend_fin_replacement_child() { fn closed_accepted_descriptor_cannot_prove_peer_identity() { let (listener, endpoint) = fixture(); let (_authority, session) = authorization(); - let mut pending = endpoint - .begin_connect(Duration::from_secs(2), &session) - .unwrap(); + let mut pending = begin_unexpiring(&endpoint, &session); listener.set_nonblocking(true).unwrap(); let peer = accept_pending(&listener); drop(peer); diff --git a/packages/runtime-core/src/provider/https_recovery/tests.rs b/packages/runtime-core/src/provider/https_recovery/tests.rs index 55c6788ff..c02b4dfd4 100644 --- a/packages/runtime-core/src/provider/https_recovery/tests.rs +++ b/packages/runtime-core/src/provider/https_recovery/tests.rs @@ -271,7 +271,7 @@ fn rejects_a_live_inherited_unix_listener_at_the_pinned_inode() { // ConnectionRefused. Poll the unchanged classifier for a bounded time and // record the raw observation on the final attempt so a consistent platform // errno can be identified later without relaxing the product classifier. - let deadline = Instant::now() + Duration::from_secs(1); + let deadline = Instant::now() + crate::provider::test_executable::LOADED_EVENT_HANG_GUARD; while socket_absent(&stale).is_err() { if Instant::now() >= deadline { let observed = UnixStream::connect(&stale) @@ -419,9 +419,9 @@ fn complete_fixture() -> (Fixture, Candidate, Alias) { receipt["caSha256"] = serde_json::json!(digest(&der)); receipt["authority"]["socket"] = hostname_authority::managed::inspect(&candidate).unwrap()["socket"].clone(); - let free = std::net::TcpListener::bind((std::net::Ipv4Addr::LOCALHOST, 0)).unwrap(); - receipt["listener"]["port"] = serde_json::json!(free.local_addr().unwrap().port()); - drop(free); + // Recovery requires this recorded port free much later; never an ephemeral one. + receipt["listener"]["port"] = + serde_json::json!(crate::provider::test_socket::unallocated_tcp_port()); config["binding"]["httpsPort"] = receipt["listener"]["port"].clone(); fs::write(&path, serde_json::to_vec(&config).unwrap()).unwrap(); fs::write(&receipt_path, serde_json::to_vec(&receipt).unwrap()).unwrap(); @@ -564,9 +564,8 @@ fn successful_port_guards_own_all_four_allocated_addresses_during_observation() IpAddr::V6(_) => IpAddr::V6(std::net::Ipv6Addr::LOCALHOST), }; let reachable = std::net::SocketAddr::new(local_ip, address.port()); - let peer = - std::net::TcpStream::connect_timeout(&reachable, std::time::Duration::from_secs(1)) - .unwrap(); + let guard = crate::provider::test_executable::LOADED_EVENT_HANG_GUARD; + let peer = std::net::TcpStream::connect_timeout(&reachable, guard).unwrap(); let mut readiness = libc::pollfd { fd: listener.as_raw_fd(), events: libc::POLLIN, @@ -574,7 +573,8 @@ fn successful_port_guards_own_all_four_allocated_addresses_during_observation() }; // SAFETY: readiness is one live pollfd holding our live listener. Observe // its asynchronous accept readiness once with a finite timeout, no retry. - assert_eq!(unsafe { libc::poll(&raw mut readiness, 1, 1000) }, 1); + let guard = guard.as_millis() as libc::c_int; + assert_eq!(unsafe { libc::poll(&raw mut readiness, 1, guard) }, 1); assert_eq!(readiness.revents, libc::POLLIN); let (accepted, peer_address) = listener.accept().unwrap(); assert_eq!(peer_address, peer.local_addr().unwrap()); diff --git a/packages/runtime-core/src/provider/relay_loop/admission_tests.rs b/packages/runtime-core/src/provider/relay_loop/admission_tests.rs index 6b6b3060c..43dfe0106 100644 --- a/packages/runtime-core/src/provider/relay_loop/admission_tests.rs +++ b/packages/runtime-core/src/provider/relay_loop/admission_tests.rs @@ -1,6 +1,9 @@ use super::tests::{accept, fixture, guest, reactor, session, transport_pair}; use super::*; use crate::provider::relay_auth::Credential; +use crate::provider::test_executable::{ + ISOLATED_HANG_GUARD, LOADED_EVENT_HANG_GUARD, MAX_CONNECT_BUDGET, run_isolated, +}; use std::{net::TcpListener, thread}; fn authenticate(client: &mut UnixStream, credential: &Credential) -> Traffic { @@ -35,7 +38,7 @@ fn fragmented_handshake_streams_and_shares_flow_capacity() { let mut reactor = reactor(2); let (server, mut client) = transport_pair(); reactor - .admit(server, &endpoint, &authority, Duration::from_secs(2)) + .admit(server, &endpoint, &authority, MAX_CONNECT_BUDGET) .unwrap(); assert_eq!(reactor.connections(), 1); thread::scope(|scope| { @@ -54,7 +57,7 @@ fn fragmented_handshake_streams_and_shares_flow_capacity() { let traffic = authenticate(&mut client, &credential); guest(&mut client, traffic, 128 * 1024, 11); }); - let deadline = Instant::now() + Duration::from_secs(4); + let deadline = Instant::now() + LOADED_EVENT_HANG_GUARD; while reactor.connections() > 0 { assert!(Instant::now() < deadline); reactor.tick(Duration::from_millis(10)).unwrap(); @@ -71,16 +74,26 @@ fn partial_hello_has_fixed_deadline_and_never_connects() { let (listener, endpoint, authority, credential) = fixture(); let mut reactor = reactor(1); let (server, mut client) = transport_pair(); + let admitted = Instant::now(); reactor - .admit(server, &endpoint, &authority, Duration::from_millis(40)) + .admit(server, &endpoint, &authority, MAX_CONNECT_BUDGET) .unwrap(); + let returned = Instant::now(); let (_, hello) = credential.begin().unwrap(); client.write_all(&hello[..1]).unwrap(); ticks(&mut reactor, 2); let deadline = reactor.handshakes[0].deadline; + // The deadline comes from the admitted budget alone. + assert!( + admitted + MAX_CONNECT_BUDGET <= deadline && deadline <= returned + MAX_CONNECT_BUDGET, + "E_TEST_HANDSHAKE_DEADLINE_NOT_FROM_BUDGET" + ); client.write_all(&hello[1..2]).unwrap(); ticks(&mut reactor, 2); assert_eq!(reactor.handshakes[0].deadline, deadline); + // Expire that fixed deadline instead of racing a short budget against a loaded + // host's scheduling delay during the partial steps above (HACK-1457). + reactor.handshakes[0].deadline = Instant::now(); reactor.tick(Duration::from_secs(1)).unwrap(); assert_eq!(reactor.connections(), 0); assert_eq!(reactor.stats().timed_out, 1); @@ -94,7 +107,7 @@ fn bad_proof_and_truncated_hello_close_without_upstream() { let mut reactor = reactor(1); let (server, mut client) = transport_pair(); reactor - .admit(server, &endpoint, &authority, Duration::from_secs(1)) + .admit(server, &endpoint, &authority, MAX_CONNECT_BUDGET) .unwrap(); let (_, hello) = credential.begin().unwrap(); if bad_proof { @@ -120,7 +133,7 @@ fn pending_and_active_share_capacity_and_cancellation_releases_it() { let mut reactor = reactor(2); let (server, _client) = transport_pair(); let id = reactor - .admit(server, &endpoint, &authority, Duration::from_secs(1)) + .admit(server, &endpoint, &authority, MAX_CONNECT_BUDGET) .unwrap(); let (_, authorized) = session(&authority, &credential); let (server, _client2) = transport_pair(); @@ -149,7 +162,7 @@ fn shared_revocation_wakes_all_pending_handshake_stages() { for phase in 0..4 { let (server, mut client) = transport_pair(); reactor - .admit(server, &endpoint, &authority, Duration::from_secs(2)) + .admit(server, &endpoint, &authority, MAX_CONNECT_BUDGET) .unwrap(); if phase > 0 { let (start, hello) = credential.begin().unwrap(); @@ -243,52 +256,10 @@ fn listener_accepts_one_and_preserves_owned_path() { fn authenticated_admission_rechecks_replaced_endpoint() { // The close/rebind fixture must not be inherited by unrelated parallel forks. // Keep its sockets in an isolated child and require the complete assertion path. - use std::process::{Child, Command, Stdio}; - struct OwnedChild(Child); - impl Drop for OwnedChild { - fn drop(&mut self) { - let _ = self.0.kill(); - let _ = self.0.wait(); - } - } - let mut child = OwnedChild( - Command::new(std::env::current_exe().unwrap()) - .args([ - "--ignored", - "--exact", - "provider::relay_loop::admission_tests::replaced_endpoint_child", - ]) - .stdin(Stdio::null()) - .stdout(Stdio::piped()) - .stderr(Stdio::inherit()) - .spawn() - .unwrap(), - ); - let deadline = Instant::now() + Duration::from_secs(5); - loop { - if let Some(status) = child.0.try_wait().unwrap() { - assert!(status.success()); - break; - } - assert!( - Instant::now() < deadline, - "endpoint replacement child exceeded its watchdog" - ); - thread::sleep(Duration::from_millis(1)); - } - let mut output = String::new(); - child - .0 - .stdout - .take() - .unwrap() - .take(4097) - .read_to_string(&mut output) - .unwrap(); - assert!(output.len() <= 4096); - assert!( - output.contains("test result: ok. 1 passed"), - "isolated assertion did not run" + // The watchdog bounds launching the whole test binary on a loaded host (HACK-1457). + run_isolated( + "provider::relay_loop::admission_tests::replaced_endpoint_child", + ISOLATED_HANG_GUARD, ); } #[test] @@ -302,13 +273,13 @@ fn replaced_endpoint_child() { let mut reactor = reactor(1); let (server, mut client) = transport_pair(); reactor - .admit(server, &endpoint, &authority, Duration::from_secs(2)) + .admit(server, &endpoint, &authority, MAX_CONNECT_BUDGET) .unwrap(); thread::scope(|scope| { let client = scope.spawn(|| { let _ = authenticate(&mut client, &credential); }); - let deadline = Instant::now() + Duration::from_secs(3); + let deadline = Instant::now() + LOADED_EVENT_HANG_GUARD; while reactor.connections() > 0 { assert!(Instant::now() < deadline); reactor.tick(Duration::from_millis(10)).unwrap(); @@ -366,7 +337,7 @@ fn buffered_hello_keeps_original_deadline_and_wrong_proof_never_connects() { server, &endpoint, &authority, - Instant::now() + Duration::from_secs(1), + Instant::now() + MAX_CONNECT_BUDGET, &hello, ) .unwrap(); diff --git a/packages/runtime-core/src/provider/relay_loop/tests.rs b/packages/runtime-core/src/provider/relay_loop/tests.rs index 43fe9af54..917297aea 100644 --- a/packages/runtime-core/src/provider/relay_loop/tests.rs +++ b/packages/runtime-core/src/provider/relay_loop/tests.rs @@ -1,5 +1,6 @@ use super::*; use crate::provider::relay_auth::{Authority, Binding, Credential}; +use crate::provider::test_executable::{LOADED_EVENT_HANG_GUARD, MAX_CONNECT_BUDGET}; use std::{net::TcpListener, thread}; pub(super) fn fixture() -> (TcpListener, HostEndpoint, Authority, Credential) { let listener = TcpListener::bind("127.0.0.1:0").unwrap(); @@ -32,25 +33,27 @@ pub(super) fn session( let (session, proof) = server.finish(&proof).unwrap(); (client.accept(&proof).unwrap().into_traffic(), session) } +/// A reactor whose flows neither expire on the connect budget nor go idle while a loaded +/// host delays the test's own threads (HACK-1457). Tests of either deadline force it. pub(super) fn reactor(max_flows: usize) -> RelayLoop { RelayLoop::new(Limits { max_flows, - connect_timeout: Duration::from_secs(2), - idle_timeout: Duration::from_secs(3), + connect_timeout: MAX_CONNECT_BUDGET, + idle_timeout: LOADED_EVENT_HANG_GUARD, }) .unwrap() } pub(super) fn accept(listener: &TcpListener) -> TcpStream { - let deadline = Instant::now() + Duration::from_secs(2); + let deadline = Instant::now() + LOADED_EVENT_HANG_GUARD; loop { match listener.accept() { Ok((stream, _)) => { stream.set_nonblocking(false).unwrap(); stream - .set_read_timeout(Some(Duration::from_secs(2))) + .set_read_timeout(Some(LOADED_EVENT_HANG_GUARD)) .unwrap(); stream - .set_write_timeout(Some(Duration::from_secs(2))) + .set_write_timeout(Some(LOADED_EVENT_HANG_GUARD)) .unwrap(); return stream; } @@ -65,19 +68,19 @@ pub(super) fn accept(listener: &TcpListener) -> TcpStream { pub(super) fn transport_pair() -> (UnixStream, UnixStream) { let (server, client) = UnixStream::pair().unwrap(); client - .set_read_timeout(Some(Duration::from_secs(3))) + .set_read_timeout(Some(LOADED_EVENT_HANG_GUARD)) .unwrap(); client - .set_write_timeout(Some(Duration::from_secs(3))) + .set_write_timeout(Some(LOADED_EVENT_HANG_GUARD)) .unwrap(); (server, client) } pub(super) fn guest(transport: &mut UnixStream, mut traffic: Traffic, bytes: usize, seed: usize) { transport - .set_read_timeout(Some(Duration::from_secs(3))) + .set_read_timeout(Some(LOADED_EVENT_HANG_GUARD)) .unwrap(); transport - .set_write_timeout(Some(Duration::from_secs(3))) + .set_write_timeout(Some(LOADED_EVENT_HANG_GUARD)) .unwrap(); let mut writer = transport.try_clone().unwrap(); thread::scope(|scope| { @@ -135,8 +138,9 @@ fn multiple_streaming_flows_preserve_eof_and_bound_queues() { let mut peer = accept(&listener); peers.push(scope.spawn(move || { peer.set_nonblocking(false).unwrap(); - peer.set_read_timeout(Some(Duration::from_secs(3))).unwrap(); - peer.set_write_timeout(Some(Duration::from_secs(3))) + peer.set_read_timeout(Some(LOADED_EVENT_HANG_GUARD)) + .unwrap(); + peer.set_write_timeout(Some(LOADED_EVENT_HANG_GUARD)) .unwrap(); let mut total = 0; let mut body = [0; MAX_DATA]; @@ -153,7 +157,7 @@ fn multiple_streaming_flows_preserve_eof_and_bound_queues() { })); guests.push(scope.spawn(move || guest(&mut transport, client, 262144, flow * 17))); } - let deadline = Instant::now() + Duration::from_secs(5); + let deadline = Instant::now() + LOADED_EVENT_HANG_GUARD; while relay.active() > 0 { assert!(Instant::now() < deadline); relay.tick(Duration::from_millis(100)).unwrap(); @@ -208,7 +212,9 @@ fn idle_and_pending_deadlines_retire_without_payload() { let (_, session) = session(&authority, &credential); let (server, mut guest) = transport_pair(); relay.add(server, &endpoint, session).unwrap(); - relay.flows[0].last_activity = Instant::now() - Duration::from_secs(4); + relay.flows[0].last_activity = Instant::now() + .checked_sub(relay.limits.idle_timeout + Duration::from_secs(1)) + .unwrap(); relay.tick(Duration::from_secs(1)).unwrap(); assert_eq!(relay.stats.timed_out, 1); assert_eq!(guest.read(&mut [0]).unwrap(), 0); @@ -428,7 +434,7 @@ fn peer_fin_keeps_the_request_direction_open() { .unwrap(); assert_eq!(request, b"after response FIN"); }); - let deadline = Instant::now() + Duration::from_secs(3); + let deadline = Instant::now() + LOADED_EVENT_HANG_GUARD; while relay.active() > 0 { assert!(Instant::now() < deadline); relay.tick(Duration::from_millis(100)).unwrap(); @@ -446,7 +452,7 @@ fn unaccepted_peer_expires_on_the_connection_budget() { let (_, session) = session(&authority, &credential); let (server, mut guest) = transport_pair(); relay.add(server, &endpoint, session).unwrap(); - let deadline = Instant::now() + Duration::from_secs(1); + let deadline = Instant::now() + LOADED_EVENT_HANG_GUARD; while relay.active() > 0 { assert!(Instant::now() < deadline); relay.tick(Duration::from_millis(100)).unwrap(); @@ -467,7 +473,7 @@ fn truncated_request_is_reset_not_graceful_backend_eof() { .write_all(&client.send.encode(Frame::Data(b"partial request")).unwrap()) .unwrap(); guest.shutdown(Shutdown::Write).unwrap(); - let deadline = Instant::now() + Duration::from_secs(1); + let deadline = Instant::now() + LOADED_EVENT_HANG_GUARD; while relay.active() > 0 { assert!(Instant::now() < deadline); relay.tick(Duration::from_millis(20)).unwrap(); diff --git a/packages/runtime-core/src/provider/resources.rs b/packages/runtime-core/src/provider/resources.rs index e92e0fc40..a1d9fb9f4 100644 --- a/packages/runtime-core/src/provider/resources.rs +++ b/packages/runtime-core/src/provider/resources.rs @@ -193,11 +193,13 @@ mod tests { #[test] fn owned_tree_includes_live_child_and_refuses_stale_root() { let mut child = std::process::Command::new("/bin/bash") - .args(["-c", "/bin/sleep 30 & wait"]) + .args(["-c", "/bin/sleep 120 & wait"]) .spawn() .unwrap(); let root = identity::observe(child.id() as i32).unwrap(); - let deadline = std::time::Instant::now() + std::time::Duration::from_secs(2); + // A hang guard for the fork, exec and first accounting sample on a loaded host. + let deadline = + std::time::Instant::now() + crate::provider::test_executable::LOADED_EVENT_HANG_GUARD; let result = loop { if let Ok(tree) = observe(&root) { // A two-PID snapshot can still catch the forked shell before exec, @@ -248,8 +250,9 @@ mod tests { #[test] fn cpu_counter_advances_in_nanoseconds_under_work() { let root = identity::observe(std::process::id() as i32).unwrap(); - // Observe only this process's counter so unrelated parallel test children do not matter. - let read = || { + // The native counter under test: this whole process, in Mach absolute time converted + // to nanoseconds by `cpu_nanoseconds`. + let native = || { let mut value = std::mem::MaybeUninit::::zeroed(); assert_eq!( unsafe { @@ -260,7 +263,8 @@ mod tests { let value = unsafe { value.assume_init() }; cpu_nanoseconds(value.ri_user_time + value.ri_system_time).unwrap() }; - let cpu_clock = || { + // An independent clock for the same whole-process quantity, at microsecond resolution. + let process_clock = || { let mut usage = std::mem::MaybeUninit::::zeroed(); assert_eq!( unsafe { libc::getrusage(libc::RUSAGE_SELF, usage.as_mut_ptr()) }, @@ -270,18 +274,41 @@ mod tests { ((usage.ru_utime.tv_sec + usage.ru_stime.tv_sec) as u64) * 1_000_000_000 + ((usage.ru_utime.tv_usec + usage.ru_stime.tv_usec) as u64) * 1000 }; - let before = read(); - let clock_before = cpu_clock(); - let start = std::time::Instant::now(); - while start.elapsed() < std::time::Duration::from_millis(40) { + let thread_clock = || { + let mut now = libc::timespec { + tv_sec: 0, + tv_nsec: 0, + }; + assert_eq!( + unsafe { libc::clock_gettime(libc::CLOCK_THREAD_CPUTIME_ID, &mut now) }, + 0 + ); + now.tv_sec as u64 * 1_000_000_000 + now.tv_nsec as u64 + }; + // Parallel tests in this process also use CPU, so no single read pair is simultaneous. + // Bracket the native window between an inner and an outer process-clock window instead: + // every counter is monotonic, so inner <= native <= outer whatever runs in between. + // And spin for 40 ms of this thread's CPU, not wall time, so a loaded host cannot + // shrink the measured work (HACK-1457: 40 ms of wall time measured 35 ms native and + // 72 ms by the clock read just after it). + const WORK: u64 = 40_000_000; + let outer_before = process_clock(); + let native_before = native(); + let inner_before = process_clock(); + let spin_start = thread_clock(); + while thread_clock() - spin_start < WORK { std::hint::black_box(123u64.wrapping_mul(456)); } - let native_delta = read() - before; - let clock_delta = cpu_clock() - clock_before; - assert!(native_delta > 1_000_000); + let inner_delta = process_clock() - inner_before; + let native_delta = native() - native_before; + let outer_delta = process_clock() - outer_before; + assert!(inner_delta >= WORK, "E_TEST_CPU_WORK_NOT_MEASURED"); + // Half or double absorbs the two counters' different accounting; an unconverted Mach + // tick count (about 1/41 of nanoseconds on Apple silicon) or a doubled conversion + // still fails. assert!( - native_delta > clock_delta / 2 && native_delta < clock_delta * 2, - "native={native_delta}, clock={clock_delta}" + native_delta > inner_delta / 2 && native_delta < outer_delta * 2, + "E_TEST_CPU_COUNTER_UNITS" ); } } diff --git a/packages/runtime-core/src/provider/shared_https_recovery/tests.rs b/packages/runtime-core/src/provider/shared_https_recovery/tests.rs index 99d1785bb..cb9f764ea 100644 --- a/packages/runtime-core/src/provider/shared_https_recovery/tests.rs +++ b/packages/runtime-core/src/provider/shared_https_recovery/tests.rs @@ -62,9 +62,7 @@ fn fixture(present_socket: bool) -> Fixture { let frontend_sha = executable(&frontend); let runtime_sha = executable(&runtime); let caddy_sha = executable(&caddy); - let reserved = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); - let port = reserved.local_addr().unwrap().port(); - drop(reserved); + let port = crate::provider::test_socket::unallocated_tcp_port(); let generation = "a".repeat(32); let run = "b".repeat(32); let mut lease = Lease { diff --git a/packages/runtime-core/src/provider/test_executable.rs b/packages/runtime-core/src/provider/test_executable.rs index 567382d8e..fb85242d2 100644 --- a/packages/runtime-core/src/provider/test_executable.rs +++ b/packages/runtime-core/src/provider/test_executable.rs @@ -51,6 +51,23 @@ pub(super) fn sleeping_executable(binary: &Path) { /// [`ISOLATED_HANG_GUARD`], so an isolated fixture still reports its own failure. pub(super) const LOADED_HOST_SCAN_WAIT: Duration = Duration::from_secs(60); +/// Hang guard for a test that waits for an event it has already caused: a queued +/// connection, bytes or a close it wrote, a relay draining flows whose peers it drives, a +/// child it spawned reaching exec. It bounds a hang, not latency. On a loaded host, 1-5 s +/// waits for such events were too short even though the event did arrive (HACK-1457). +/// Only an event that never happens reaches it. +// The host-endpoint, relay and resource fixtures that use it are macOS-only. +#[cfg_attr(not(target_os = "macos"), allow(dead_code))] +pub(super) const LOADED_EVENT_HANG_GUARD: Duration = Duration::from_secs(60); + +/// The largest connect or admission budget that `HostEndpoint` and `RelayLoop` accept (they +/// refuse zero and anything over 5 s). A test whose result is not that budget's expiry +/// passes it, so a loaded host's scheduling delay before the next progress check cannot +/// turn the result into a timeout (HACK-1457). Tests of expiry still force an expired +/// deadline themselves, and the 5 s bound of the APIs is unchanged. +#[cfg_attr(not(target_os = "macos"), allow(dead_code))] +pub(super) const MAX_CONNECT_BUDGET: Duration = Duration::from_secs(5); + /// Hang guard for [`run_isolated`]: the child launches the whole test binary on a host that /// may be heavily loaded, so this bounds a hang rather than the fixture's own timing. pub(super) const ISOLATED_HANG_GUARD: Duration = Duration::from_secs(120); diff --git a/packages/runtime-core/src/provider/test_socket.rs b/packages/runtime-core/src/provider/test_socket.rs index 0aadc6530..e86bce497 100644 --- a/packages/runtime-core/src/provider/test_socket.rs +++ b/packages/runtime-core/src/provider/test_socket.rs @@ -51,6 +51,63 @@ pub(super) fn bound_unlistened(path: &Path) -> OwnedFd { socket } +/// Lowest port [`unallocated_tcp_port`] picks from. Its range ends where the kernel's +/// ephemeral range starts, and it must stay a range the kernel never assigns by itself. +#[cfg(target_os = "macos")] +const UNALLOCATED_PORT_FLOOR: u16 = 20_000; + +/// Return a TCP port that no wildcard or loopback socket holds now, chosen below the +/// kernel's ephemeral range, for a fixture that records a port and later needs it free. +/// +/// Do not get such a port by binding port 0 and dropping the listener. The kernel hands +/// out the same range to every other port-0 bind and outgoing connection on the host, so +/// under parallel load another socket can take the port before the fixture needs it +/// (HACK-1457: a full HTTPS recovery fixture refused its own recorded port at load ~200). +/// The kernel never assigns a port below its ephemeral range by itself, so only an +/// explicit bind of the same number can take this one. +#[cfg(target_os = "macos")] +pub(super) fn unallocated_tcp_port() -> u16 { + use std::net::{Ipv4Addr, Ipv6Addr, TcpListener}; + let mut first: libc::c_int = 0; + let mut length = std::mem::size_of_val(&first); + // SAFETY: the name is NUL-terminated and the output is one live, correctly sized integer. + let read = unsafe { + libc::sysctlbyname( + c"net.inet.ip.portrange.first".as_ptr(), + (&raw mut first).cast(), + &raw mut length, + std::ptr::null_mut(), + 0, + ) + }; + let ceiling = u16::try_from(first).unwrap_or(0); + assert!( + read == 0 && length == std::mem::size_of_val(&first) && ceiling > UNALLOCATED_PORT_FLOOR, + "E_TEST_EPHEMERAL_RANGE_UNKNOWN" + ); + let span = u32::from(ceiling - UNALLOCATED_PORT_FLOOR); + // Spread parallel fixtures over the range; the bind probes below decide. + let seed = std::process::id() + .wrapping_mul(7919) + .wrapping_add(crate::next_test_scratch() as u32) + .wrapping_add( + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .subsec_nanos(), + ); + (0..span) + .map(|step| UNALLOCATED_PORT_FLOOR + ((seed.wrapping_add(step) % span) as u16)) + .find(|&port| { + // Without SO_REUSEADDR a wildcard bind fails if any socket holds the port. + TcpListener::bind((Ipv4Addr::UNSPECIFIED, port)).is_ok() + && TcpListener::bind((Ipv6Addr::UNSPECIFIED, port)).is_ok() + && TcpListener::bind((Ipv4Addr::LOCALHOST, port)).is_ok() + && TcpListener::bind((Ipv6Addr::LOCALHOST, port)).is_ok() + }) + .expect("E_TEST_NO_UNALLOCATED_PORT") +} + #[cfg(test)] mod tests { use super::*; @@ -80,4 +137,44 @@ mod tests { drop(copy); std::fs::remove_dir_all(root).unwrap(); } + + #[cfg(target_os = "macos")] + #[test] + fn an_unallocated_port_is_free_and_outside_the_kernel_assigned_range() { + use std::net::{Ipv4Addr, Ipv6Addr, TcpListener}; + let mut first: libc::c_int = 0; + let mut length = std::mem::size_of_val(&first); + // SAFETY: NUL-terminated name; one live, correctly sized output integer. + assert_eq!( + unsafe { + libc::sysctlbyname( + c"net.inet.ip.portrange.first".as_ptr(), + (&raw mut first).cast(), + &raw mut length, + std::ptr::null_mut(), + 0, + ) + }, + 0 + ); + for _ in 0..8 { + let port = unallocated_tcp_port(); + assert!( + (UNALLOCATED_PORT_FLOOR..u16::try_from(first).unwrap()).contains(&port), + "E_TEST_PORT_IN_KERNEL_RANGE" + ); + // Free on all four addresses that a later wildcard and loopback claim needs. + let held = [ + TcpListener::bind((Ipv4Addr::UNSPECIFIED, port)), + TcpListener::bind((Ipv4Addr::LOCALHOST, port)), + ]; + assert!(held.iter().all(Result::is_ok), "E_TEST_PORT_NOT_FREE"); + drop(held); + let held = [ + TcpListener::bind((Ipv6Addr::UNSPECIFIED, port)), + TcpListener::bind((Ipv6Addr::LOCALHOST, port)), + ]; + assert!(held.iter().all(Result::is_ok), "E_TEST_PORT_NOT_FREE"); + } + } }