From 3cb2af3056b8093bd07251c11e77bcf4c65c2422 Mon Sep 17 00:00:00 2001 From: Enrique Saurez Date: Tue, 10 Nov 2026 04:44:52 -0800 Subject: [PATCH 1/6] cli: recover interrupted launches under the sandbox lock Use verified endpoint and log claims for unrecorded launches and boot/PID/start identities for recorded processes. Require confirmed exit before removing capability and journal; retain indeterminate evidence and its refusal cause. Keep committed-runtime recovery separate and bounded claim rechecks intact. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: eb3cc7e0-f51e-45ed-b911-c7b214740e05 Copilot-Session: 6aaf63ff-0a40-44dd-bc07-f43344fb11e5 --- nvx-rs/src/host/cli/store/launch.rs | 123 ++++++- nvx-rs/src/host/cli/store/locked.rs | 11 +- nvx-rs/src/host/cli/store/mod.rs | 3 + nvx-rs/src/host/cli/store/tests.rs | 1 + nvx-rs/src/host/cli/store/tests/launch.rs | 17 +- nvx-rs/src/host/cli/store/tests/recovery.rs | 362 ++++++++++++++++++++ 6 files changed, 505 insertions(+), 12 deletions(-) create mode 100644 nvx-rs/src/host/cli/store/tests/recovery.rs diff --git a/nvx-rs/src/host/cli/store/launch.rs b/nvx-rs/src/host/cli/store/launch.rs index d0a5c04e6..d3cdb0100 100644 --- a/nvx-rs/src/host/cli/store/launch.rs +++ b/nvx-rs/src/host/cli/store/launch.rs @@ -1,13 +1,17 @@ -//! Locked launch transitions, without spawning, recovery, or process termination. +//! Locked launch transitions and bounded recovery of an interrupted launch. + +use std::thread; +use std::time::{Duration, Instant}; use super::StateFile; use super::files; use super::locked::LockedState; use super::record::{LaunchJournal, RuntimeRecord}; -use crate::host::process::ProcessRef; +use crate::host::cli::deadline::RETRY_INTERVAL; +use crate::host::process::{BootId, KILL_TIMEOUT, Opened, Process, ProcessRef}; use crate::host::runtime::RuntimeIdentity; use crate::host::vm::{Prepared, SavedVm}; -use crate::{Error, Result}; +use crate::{Error, ErrorKind, Result}; /// A launch whose lock stays borrowed until commit or until its journal is left behind. #[derive(Debug)] @@ -17,6 +21,77 @@ pub(crate) struct Launch<'a> { } impl LockedState { + /// Resolves only an uncommitted journal, before the next command can mutate state. + pub(crate) fn recover(&mut self, deadline: Instant) -> Result<()> { + if self.runtime()?.is_some() { + return Ok(()); + } + let Some(journal) = self.journal()? else { + return Ok(()); + }; + if journal.vm_dir != self.dir() || journal.log != self.path(StateFile::Log) { + return Err(Error::io( + "the launch journal belongs to another sandbox directory", + )); + } + match journal.process() { + Some(reference) => self.end_recorded(&journal, reference, deadline)?, + None if !self.released_claim(&journal, deadline)? => { + let process = self.endpoint_owner(&journal)?; + end_process(&process, deadline)?; + } + None => {} + } + if self.capability()?.is_some() { + files::remove(self, StateFile::Capability)?; + } + // Keep the journal until cleanup has finished, including after a partial removal. + files::remove(self, StateFile::Journal) + } + + /// Ends only the recorded process, or accepts an independently proved release of its claim. + fn end_recorded( + &self, + journal: &LaunchJournal, + reference: &ProcessRef, + deadline: Instant, + ) -> Result<()> { + let boot = BootId::current().map_err(|error| { + Error::io("cannot identify the current host boot").with_source(error) + })?; + if boot != reference.boot { + return Ok(()); + } + let unheld = match Process::open(reference) { + Ok(Opened::Running(process)) => return end_process(&process, deadline), + Ok(Opened::Exited(_)) => Error::new( + ErrorKind::StillRunning, + "the recorded VM exited, but a process may still hold its launch log", + ), + Err(error) => Error::io("cannot verify the recorded VM process").with_source(error), + }; + if self.released_claim(journal, deadline)? { + Ok(()) + } else { + Err(unheld) + } + } + + /// Rechecks transient fork-inherited descriptors for at most one second. + fn released_claim(&self, journal: &LaunchJournal, deadline: Instant) -> Result { + let until = deadline.min(Instant::now() + Duration::from_secs(1)); + loop { + if !self.claim_held(journal)? { + return Ok(true); + } + let left = until.saturating_duration_since(Instant::now()); + if left.is_zero() { + return Ok(false); + } + thread::sleep(left.min(RETRY_INTERVAL)); + } + } + /// Durably publishes the capability and complete prepared journal before the caller spawns. pub(crate) fn begin_launch( &mut self, @@ -38,6 +113,48 @@ impl LockedState { } } +/// Ends a held, verified process and requires confirmation of its exit before the deadline. +fn end_process(process: &Process, deadline: Instant) -> Result<()> { + let failed = |message: &str| { + Error::new( + ErrorKind::StillRunning, + format!( + "cannot confirm exit of VM process {}: {message}", + process.reference().pid + ), + ) + }; + if process + .wait(Instant::now()) + .map_err(|error| failed("the exit check failed").with_source(error))? + .is_some() + { + return Ok(()); + } + let until = deadline.min(Instant::now() + KILL_TIMEOUT); + if until <= Instant::now() { + return Err(failed("the recovery deadline passed")); + } + let kill_error = process.kill().err(); + let exited = process.wait(until).map_err(|error| { + let mut failure = failed("the exit wait failed"); + if let Some(kill_error) = &kill_error { + failure = failed(&format!( + "termination failed: {kill_error}; the exit wait failed" + )); + } + failure.with_source(error) + })?; + if exited.is_some() { + return Ok(()); + } + let failure = failed("the verified process has not exited"); + Err(match kill_error { + Some(error) => failure.with_source(error), + None => failure, + }) +} + impl Launch<'_> { /// Records only the caller's spawned process, durably, before changing the in-memory state. pub(crate) fn record_process(&mut self, process: &ProcessRef) -> Result<()> { diff --git a/nvx-rs/src/host/cli/store/locked.rs b/nvx-rs/src/host/cli/store/locked.rs index 68737e290..35bf46ffb 100644 --- a/nvx-rs/src/host/cli/store/locked.rs +++ b/nvx-rs/src/host/cli/store/locked.rs @@ -54,6 +54,9 @@ impl StateDir { /// access, who could hold its lock. A lock file that another transition removed or replaced /// after this opened it locks nothing, so the lock is taken again: at once the first time, /// and after a pause, until the deadline, when that happens again right after. + /// + /// Once held, the lock recovers an interrupted launch before returning. Recovery can end a + /// verified VM, but retains indeterminate records and reports the cause of its refusal. pub(crate) fn lock(&self, deadline: Deadline) -> Result { let path = self.path(StateFile::Lock); let mut retried_at_once = false; @@ -71,10 +74,14 @@ impl StateDir { } } if names_locked_file(&path, &file)? { - return Ok(LockedState { + let mut locked = LockedState { state: self.clone(), lock: file, - }); + }; + locked + .recover(deadline.instant()) + .map_err(crate::host::cli::sandbox::error_with_cause)?; + return Ok(locked); } if retried_at_once && !deadline.pause(RETRY_INTERVAL)? { return Err(busy()); diff --git a/nvx-rs/src/host/cli/store/mod.rs b/nvx-rs/src/host/cli/store/mod.rs index ac58708e0..43f2197df 100644 --- a/nvx-rs/src/host/cli/store/mod.rs +++ b/nvx-rs/src/host/cli/store/mod.rs @@ -33,6 +33,9 @@ //! Its journal also identifies the inherited log claim: held means a launch process may run, //! and released proves exit. An endpoint owner is held and rechecked before it is returned; //! Linux additionally verifies that this process holds that launch's log. +//! Taking the next lock recovers an interrupted, uncommitted launch before allowing mutation: +//! it ends only a verified process, removes its journal and capability only after proved exit, +//! and refuses indeterminate recovery without changing that evidence. #![cfg_attr( not(test), expect( diff --git a/nvx-rs/src/host/cli/store/tests.rs b/nvx-rs/src/host/cli/store/tests.rs index 5db8c5adb..4dabc42a6 100644 --- a/nvx-rs/src/host/cli/store/tests.rs +++ b/nvx-rs/src/host/cli/store/tests.rs @@ -10,6 +10,7 @@ use crate::host::store::{JOURNAL_NAME, RUNTIME_NAME, SandboxStore}; mod launch; mod liveness; +mod recovery; /// How long a test waits for a lock that it released to be free. /// diff --git a/nvx-rs/src/host/cli/store/tests/launch.rs b/nvx-rs/src/host/cli/store/tests/launch.rs index 5fd38d31a..8712ea4ff 100644 --- a/nvx-rs/src/host/cli/store/tests/launch.rs +++ b/nvx-rs/src/host/cli/store/tests/launch.rs @@ -175,16 +175,19 @@ fn an_interruption_keeps_the_journal_and_capability_without_publishing_runtime() launch.record_process(&saved.process).unwrap(); } } + let before = fs::read(state.path(StateFile::Journal)).unwrap(); + assert!(locked.begin_launch(&prepared, &runtime).is_err()); + assert_eq!(fs::read(state.path(StateFile::Journal)).unwrap(), before); drop(locked); - let mut reopened = state.lock(Deadline::after(10.0)).unwrap(); - let journal = reopened.journal().unwrap().unwrap(); + let journal: record::LaunchJournal = + serde_json::from_slice(&fs::read(state.path(StateFile::Journal)).unwrap()).unwrap(); assert_eq!(journal.process(), spawned.then_some(&saved.process)); assert_eq!(journal.capability, saved.capability); - assert_eq!(reopened.capability().unwrap(), Some(saved.capability)); - assert!(reopened.runtime().unwrap().is_none()); - let before = fs::read(state.path(StateFile::Journal)).unwrap(); - assert!(reopened.begin_launch(&prepared, &runtime).is_err()); - assert_eq!(fs::read(state.path(StateFile::Journal)).unwrap(), before); + assert_eq!( + fs::read(state.path(StateFile::Capability)).unwrap(), + saved.capability + ); + assert!(!state.path(StateFile::Runtime).exists()); } } diff --git a/nvx-rs/src/host/cli/store/tests/recovery.rs b/nvx-rs/src/host/cli/store/tests/recovery.rs new file mode 100644 index 000000000..fae235449 --- /dev/null +++ b/nvx-rs/src/host/cli/store/tests/recovery.rs @@ -0,0 +1,362 @@ +use super::launch::prepared; +use super::*; +use crate::host::cli::store::files; +use crate::host::config::ConsoleAttachment; + +#[test] +fn the_next_lock_cleans_a_pre_spawn_journal_only_after_its_claim_is_released() { + let (_directory, state) = state(); + let (prepared, runtime) = prepared(&state, ConsoleAttachment::Disconnected); + let mut locked = state.lock(Deadline::after(10.0)).unwrap(); + drop(locked.begin_launch(&prepared, &runtime).unwrap()); + drop(locked); + let journal = fs::read(state.path(StateFile::Journal)).unwrap(); + let capability = fs::read(state.path(StateFile::Capability)).unwrap(); + let started = Instant::now(); + assert!(state.lock(Deadline::after(0.05)).is_err()); + assert!(started.elapsed() < Duration::from_secs(2)); + assert_eq!(fs::read(state.path(StateFile::Journal)).unwrap(), journal); + assert_eq!( + fs::read(state.path(StateFile::Capability)).unwrap(), + capability + ); + drop(prepared); + let locked = state.lock(Deadline::after(10.0)).unwrap(); + assert!(locked.journal().unwrap().is_none()); + assert!(locked.capability().unwrap().is_none()); + assert!(state.path(StateFile::Log).exists()); +} + +#[test] +fn a_transient_inherited_log_descriptor_is_rechecked_before_refusing_recovery() { + let (_directory, state) = state(); + let (prepared, runtime) = prepared(&state, ConsoleAttachment::Disconnected); + let mut locked = state.lock(Deadline::after(10.0)).unwrap(); + drop(locked.begin_launch(&prepared, &runtime).unwrap()); + let inherited = prepared.log().try_clone().unwrap(); + drop(prepared); + drop(locked); + let release = thread::spawn(move || { + thread::sleep(Duration::from_millis(80)); + drop(inherited); + }); + let locked = state.lock(Deadline::after(10.0)).unwrap(); + release.join().unwrap(); + assert!(locked.journal().unwrap().is_none()); + assert!(locked.capability().unwrap().is_none()); +} + +#[test] +fn an_interrupted_cleanup_with_no_capability_can_finish_idempotently() { + let (_directory, state) = state(); + let (prepared, runtime) = prepared(&state, ConsoleAttachment::Disconnected); + let mut locked = state.lock(Deadline::after(10.0)).unwrap(); + drop(locked.begin_launch(&prepared, &runtime).unwrap()); + files::remove(&locked, StateFile::Capability).unwrap(); + drop(prepared); + locked + .recover(Instant::now() + Duration::from_secs(10)) + .unwrap(); + assert!(locked.journal().unwrap().is_none()); + locked.recover(Instant::now()).unwrap(); +} + +#[test] +fn replaced_logs_and_foreign_journal_paths_keep_all_evidence() { + for foreign in [false, true] { + let (_directory, state) = state(); + let (prepared, runtime) = prepared(&state, ConsoleAttachment::Disconnected); + let mut locked = state.lock(Deadline::after(10.0)).unwrap(); + drop(locked.begin_launch(&prepared, &runtime).unwrap()); + drop(prepared); + if foreign { + let mut journal = locked.journal().unwrap().unwrap(); + journal.vm_dir = state.dir().join("another-vm"); + locked.write_record(StateFile::Journal, &journal).unwrap(); + } else { + fs::rename(state.path(StateFile::Log), state.dir().join("old-log")).unwrap(); + fs::write(state.path(StateFile::Log), b"another log").unwrap(); + } + let journal = fs::read(state.path(StateFile::Journal)).unwrap(); + let capability = fs::read(state.path(StateFile::Capability)).unwrap(); + assert!(locked.recover(Instant::now()).is_err()); + assert_eq!(fs::read(state.path(StateFile::Journal)).unwrap(), journal); + assert_eq!( + fs::read(state.path(StateFile::Capability)).unwrap(), + capability + ); + drop(locked); + let (_, message) = error(state.lock(Deadline::after(0.0)).unwrap_err()); + let cause = if foreign { + "another sandbox directory" + } else { + "is not the claimed log" + }; + assert!(message.contains(cause), "{message}"); + } +} + +#[cfg(any(target_os = "linux", windows))] +mod process { + #[cfg(target_os = "linux")] + use std::process::{Child, Command, Stdio}; + + use super::*; + use crate::host::process::{BootId, Process, ProcessRef}; + use crate::host::spawn::{self, VmmCommand}; + use crate::host::vm::{Prepared, SavedVm, VmDir}; + + const HELPER: &str = "host::cli::store::tests::recovery::process::launch_helper"; + + struct Running(Option); + + impl Running { + fn process(&self) -> Process { + self.0.as_ref().unwrap().process().unwrap() + } + } + + impl Drop for Running { + fn drop(&mut self) { + if let Some(child) = self.0.take() { + child.kill_and_confirm_exit().unwrap(); + } + } + } + + #[cfg(target_os = "linux")] + struct Other(Child); + + #[cfg(target_os = "linux")] + impl Drop for Other { + fn drop(&mut self) { + if self.0.try_wait().unwrap().is_none() { + self.0.kill().unwrap(); + } + self.0.wait().unwrap(); + } + } + + fn running(state: &StateDir, prepared: &Prepared) -> Running { + fs::write( + state.dir().join("recovery-endpoint"), + &prepared.outputs().endpoint, + ) + .unwrap(); + let command = VmmCommand::new( + std::env::current_exe().unwrap(), + ["--exact", HELPER, "--ignored", "--nocapture"] + .map(Into::into) + .to_vec(), + state.dir(), + ); + let child = Running(Some( + spawn::spawn_vm( + &command, + prepared.capability().as_bytes(), + prepared.log().try_clone().unwrap(), + ) + .unwrap(), + )); + assert!(released(|| state.dir().join("recovery-ready").exists())); + child + } + + #[test] + fn an_unrecorded_vm_requires_both_endpoint_and_log_proofs_before_it_is_ended() { + let (_directory, state) = state(); + let (prepared, runtime) = prepared(&state, ConsoleAttachment::Disconnected); + let mut locked = state.lock(Deadline::after(10.0)).unwrap(); + drop(locked.begin_launch(&prepared, &runtime).unwrap()); + let child = running(&state, &prepared); + let process = child.process(); + drop(prepared); + drop(locked); + assert!(!process.has_exited().unwrap()); + let locked = state.lock(Deadline::after(10.0)).unwrap(); + assert!(process.wait(Instant::now()).unwrap().is_some()); + assert!(locked.journal().unwrap().is_none()); + assert!(locked.capability().unwrap().is_none()); + } + + #[cfg(target_os = "linux")] + #[test] + fn an_endpoint_owner_without_the_log_is_not_signaled_and_its_journal_stays() { + let (_directory, state) = state(); + let (prepared, runtime) = prepared(&state, ConsoleAttachment::Disconnected); + let mut locked = state.lock(Deadline::after(10.0)).unwrap(); + drop(locked.begin_launch(&prepared, &runtime).unwrap()); + fs::write( + state.dir().join("recovery-endpoint"), + &prepared.outputs().endpoint, + ) + .unwrap(); + let mut command = Command::new(std::env::current_exe().unwrap()); + command + .args(["--exact", HELPER, "--ignored", "--nocapture"]) + .current_dir(state.dir()) + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::null()); + let mut other = Other(crate::host::tests::spawn_locked(&mut command).unwrap()); + assert!(released(|| state.dir().join("recovery-ready").exists())); + let journal = fs::read(state.path(StateFile::Journal)).unwrap(); + let capability = fs::read(state.path(StateFile::Capability)).unwrap(); + assert!( + locked + .recover(Instant::now() + Duration::from_secs(2)) + .is_err() + ); + assert!(other.0.try_wait().unwrap().is_none()); + assert_eq!(fs::read(state.path(StateFile::Journal)).unwrap(), journal); + assert_eq!( + fs::read(state.path(StateFile::Capability)).unwrap(), + capability + ); + } + + #[test] + fn a_recorded_process_is_ended_only_after_a_matching_identity_and_confirmed_exit() { + let (_directory, state) = state(); + let (prepared, runtime) = prepared(&state, ConsoleAttachment::Disconnected); + let mut locked = state.lock(Deadline::after(10.0)).unwrap(); + let mut launch = locked.begin_launch(&prepared, &runtime).unwrap(); + let child = running(&state, &prepared); + let process = child.process(); + launch.record_process(process.reference()).unwrap(); + drop(launch); + drop(prepared); + assert!(locked.recover(Instant::now()).is_err()); + assert!(!process.has_exited().unwrap()); + assert!(locked.journal().unwrap().is_some()); + assert!(locked.capability().unwrap().is_some()); + locked + .recover(Instant::now() + Duration::from_secs(10)) + .unwrap(); + assert!(process.wait(Instant::now()).unwrap().is_some()); + assert!(locked.journal().unwrap().is_none()); + assert!(locked.capability().unwrap().is_none()); + } + + #[test] + fn incorrect_recorded_ids_and_start_times_never_signal_a_live_process_or_drop_its_claim() { + for mismatch in 0..3 { + let (_directory, state) = state(); + let (prepared, runtime) = prepared(&state, ConsoleAttachment::Disconnected); + let mut locked = state.lock(Deadline::after(10.0)).unwrap(); + let mut launch = locked.begin_launch(&prepared, &runtime).unwrap(); + let child = running(&state, &prepared); + let process = child.process(); + let mut reference = process.reference().clone(); + match mismatch { + 0 => reference.pid = 0, + 1 => { + reference.pid = std::process::id(); + reference.start_time = platform::process_start_time(reference.pid) + .unwrap() + .unwrap() + + 1; + } + 2 => reference.start_time += 1, + _ => unreachable!(), + } + launch.record_process(&reference).unwrap(); + drop(launch); + drop(prepared); + let journal = fs::read(state.path(StateFile::Journal)).unwrap(); + let capability = fs::read(state.path(StateFile::Capability)).unwrap(); + assert!( + locked + .recover(Instant::now() + Duration::from_secs(2)) + .is_err() + ); + assert!(!process.has_exited().unwrap()); + assert_eq!(fs::read(state.path(StateFile::Journal)).unwrap(), journal); + assert_eq!( + fs::read(state.path(StateFile::Capability)).unwrap(), + capability + ); + } + } + + #[test] + fn an_earlier_boot_proves_exit_without_signaling_a_process_from_this_boot() { + let (_directory, state) = state(); + let (prepared, runtime) = prepared(&state, ConsoleAttachment::Disconnected); + let mut locked = state.lock(Deadline::after(10.0)).unwrap(); + let mut launch = locked.begin_launch(&prepared, &runtime).unwrap(); + let child = running(&state, &prepared); + let process = child.process(); + let mut boot = *BootId::current().unwrap().as_bytes(); + boot[0] ^= 1; + launch + .record_process(&ProcessRef { + boot: BootId::from_bytes(boot), + ..process.reference().clone() + }) + .unwrap(); + drop(launch); + drop(prepared); + locked.recover(Instant::now()).unwrap(); + assert!(!process.has_exited().unwrap()); + assert!(locked.journal().unwrap().is_none()); + assert!(locked.capability().unwrap().is_none()); + } + + #[test] + fn a_committed_runtime_is_not_recovered_through_its_leftover_journal() { + let (_directory, state) = state(); + let (prepared, runtime) = prepared(&state, ConsoleAttachment::Disconnected); + let mut locked = state.lock(Deadline::after(10.0)).unwrap(); + let mut launch = locked.begin_launch(&prepared, &runtime).unwrap(); + let child = running(&state, &prepared); + let process = child.process(); + launch.record_process(process.reference()).unwrap(); + let journal = fs::read(state.path(StateFile::Journal)).unwrap(); + launch + .commit(&SavedVm { + vm_dir: VmDir::open(state.dir()).unwrap(), + endpoint: prepared.outputs().endpoint.clone(), + capability: *prepared.capability().as_bytes(), + process: process.reference().clone(), + runtime, + }) + .unwrap(); + files::replace(&locked, StateFile::Journal, &journal).unwrap(); + let before = fs::read(state.path(StateFile::Runtime)).unwrap(); + drop(prepared); + drop(locked); + let locked = state.lock(Deadline::after(10.0)).unwrap(); + assert!(!process.has_exited().unwrap()); + assert_eq!(fs::read(state.path(StateFile::Runtime)).unwrap(), before); + assert_eq!(fs::read(state.path(StateFile::Journal)).unwrap(), journal); + assert!(locked.capability().unwrap().is_some()); + } + + #[test] + #[ignore = "invoked as an owned subprocess by interrupted-launch recovery tests"] + fn launch_helper() { + let endpoint = fs::read_to_string("recovery-endpoint").unwrap(); + #[cfg(target_os = "linux")] + let _server = std::os::unix::net::UnixListener::bind(endpoint).unwrap(); + #[cfg(windows)] + let server = platform::tests::control_server(&endpoint); + fs::write("recovery-ready", b"").unwrap(); + #[cfg(target_os = "linux")] + thread::sleep(Duration::from_secs(30)); + #[cfg(windows)] + { + use std::io::Read; + + loop { + platform::tests::accept_control(&server); + match (&server).read(&mut [0]) { + Ok(0) => {} + Err(error) if error.kind() == std::io::ErrorKind::BrokenPipe => {} + result => panic!("unexpected endpoint probe: {result:?}"), + } + platform::tests::disconnect_control(&server); + } + } + } +} From dd1623f81fed740fdb1d1d326f09f46cb5f438aa Mon Sep 17 00:00:00 2001 From: Enrique Saurez Date: Tue, 10 Nov 2026 05:14:02 -0800 Subject: [PATCH 2/6] test: reconnect the owned Windows recovery control fixture Port the required single-instance named-pipe disconnect helper and CLI-gated test re-exports. Repeated endpoint-owner verification must reconnect after each finished client; no production API or assertion changes. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: eb3cc7e0-f51e-45ed-b911-c7b214740e05 Copilot-Session: 6aaf63ff-0a40-44dd-bc07-f43344fb11e5 --- nvx-rs/src/host/platform/tests.rs | 2 ++ nvx-rs/src/host/platform/windows/endpoint/tests.rs | 13 +++++++++++++ nvx-rs/src/host/platform/windows/tests.rs | 2 ++ 3 files changed, 17 insertions(+) diff --git a/nvx-rs/src/host/platform/tests.rs b/nvx-rs/src/host/platform/tests.rs index be5b7967a..37392e1ba 100644 --- a/nvx-rs/src/host/platform/tests.rs +++ b/nvx-rs/src/host/platform/tests.rs @@ -2,6 +2,8 @@ pub(crate) use super::linux::tests::{ check_identity, owner_only_problem, start_time_from_stat, store_dir_problem, }; +#[cfg(all(windows, feature = "cli"))] +pub(crate) use super::windows::tests::disconnect_control; #[cfg(windows)] pub(crate) use super::windows::tests::set_access; #[cfg(windows)] diff --git a/nvx-rs/src/host/platform/windows/endpoint/tests.rs b/nvx-rs/src/host/platform/windows/endpoint/tests.rs index 42e283d69..4d882574a 100644 --- a/nvx-rs/src/host/platform/windows/endpoint/tests.rs +++ b/nvx-rs/src/host/platform/windows/endpoint/tests.rs @@ -36,6 +36,19 @@ pub(crate) fn accept_control(server: &File) -> i32 { connected } +#[cfg(feature = "cli")] +pub(crate) fn disconnect_control(server: &File) { + use std::os::windows::io::{AsHandle, AsRawHandle}; + + use windows_sys::Win32::System::Pipes::DisconnectNamedPipe; + + // SAFETY: the owned synchronous pipe finished its previous client. + assert_ne!( + unsafe { DisconnectNamedPipe(server.as_handle().as_raw_handle()) }, + 0 + ); +} + #[test] fn a_duplicated_transport_keeps_the_verified_connection_without_reconnecting() { let endpoint = format!( diff --git a/nvx-rs/src/host/platform/windows/tests.rs b/nvx-rs/src/host/platform/windows/tests.rs index 7da84f781..a3aa27b6e 100644 --- a/nvx-rs/src/host/platform/windows/tests.rs +++ b/nvx-rs/src/host/platform/windows/tests.rs @@ -20,6 +20,8 @@ use windows_sys::Win32::System::Threading::{ PROCESS_QUERY_LIMITED_INFORMATION, PROCESS_SYNCHRONIZE, TerminateProcess, }; +#[cfg(feature = "cli")] +pub(crate) use super::endpoint::tests::disconnect_control; pub(crate) use super::endpoint::tests::{accept_control, control_server}; use super::identity::status_error; use super::log_claim::{local_directory, log_job_id, log_job_name, log_job_processes, object_name}; From eaf42142765dc01e7cf7bd0ab8fe8f11ab5fb94e Mon Sep 17 00:00:00 2001 From: Enrique Saurez Date: Tue, 10 Nov 2026 14:57:54 -0800 Subject: [PATCH 3/6] fix: bound interrupted-launch ownership probes by the caller deadline Share one recovery deadline across held endpoint-owner checks and platform retry waits, retaining evidence on timeout without changing legacy five-second callers. Reconnect only stale closed Windows test-pipe clients and cover both timeout and connect/drop-before-accept behavior with focused software tests. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: daabb81a-bdcc-435a-b767-7c672945cccc --- nvx-rs/src/host/cli/store/launch.rs | 2 +- nvx-rs/src/host/cli/store/liveness.rs | 27 +++- nvx-rs/src/host/cli/store/tests/liveness.rs | 71 +++++++++-- nvx-rs/src/host/cli/store/tests/recovery.rs | 63 +++++++++- nvx-rs/src/host/platform/linux/endpoint.rs | 17 ++- nvx-rs/src/host/platform/linux/mod.rs | 3 +- nvx-rs/src/host/platform/linux/tests.rs | 58 +++++++++ nvx-rs/src/host/platform/other.rs | 11 ++ nvx-rs/src/host/platform/windows/endpoint.rs | 27 +++- .../host/platform/windows/endpoint/tests.rs | 116 ++++++++++++++++-- nvx-rs/src/host/platform/windows/mod.rs | 3 +- 11 files changed, 363 insertions(+), 35 deletions(-) diff --git a/nvx-rs/src/host/cli/store/launch.rs b/nvx-rs/src/host/cli/store/launch.rs index d3cdb0100..8d93f101d 100644 --- a/nvx-rs/src/host/cli/store/launch.rs +++ b/nvx-rs/src/host/cli/store/launch.rs @@ -37,7 +37,7 @@ impl LockedState { match journal.process() { Some(reference) => self.end_recorded(&journal, reference, deadline)?, None if !self.released_claim(&journal, deadline)? => { - let process = self.endpoint_owner(&journal)?; + let process = self.endpoint_owner(&journal, deadline)?; end_process(&process, deadline)?; } None => {} diff --git a/nvx-rs/src/host/cli/store/liveness.rs b/nvx-rs/src/host/cli/store/liveness.rs index 4b584f836..d58ff09af 100644 --- a/nvx-rs/src/host/cli/store/liveness.rs +++ b/nvx-rs/src/host/cli/store/liveness.rs @@ -1,6 +1,7 @@ //! Bounded launch proofs, without recovery, record removal, or process termination. use std::io; +use std::time::Instant; use super::StateFile; use super::locked::LockedState; @@ -25,12 +26,19 @@ impl LockedState { /// Holds the server of the recorded endpoint, rechecks its ownership, and on Linux proves /// that this very process also holds the launch's log. An error is not proof of exit. - pub(crate) fn endpoint_owner(&self, journal: &LaunchJournal) -> Result { + pub(crate) fn endpoint_owner( + &self, + journal: &LaunchJournal, + deadline: Instant, + ) -> Result { if !self.claim_held(journal)? { return Err(Error::io("the launch log claim is released")); } - let process = - hold_endpoint_server(|| platform::endpoint_server_pid(&journal.endpoint), find)?; + let process = hold_endpoint_server( + deadline, + |deadline| platform::endpoint_server_pid_until(&journal.endpoint, deadline), + find, + )?; let reference = process.reference(); let matches = platform::serves_launch(reference.pid, reference.start_time, journal.log_identity) @@ -66,16 +74,23 @@ fn find(pid: u32) -> io::Result> { /// Adapted from the old store's owner check: the second endpoint lookup happens only after /// the process is held. The callbacks keep replacement races testable without test-only hooks. pub(super) fn hold_endpoint_server( - mut server_pid: impl FnMut() -> io::Result>, + deadline: Instant, + mut server_pid: impl FnMut(Instant) -> io::Result>, hold: impl FnOnce(u32) -> io::Result>, ) -> Result { - let pid = server_pid() + let mut probe = || { + if Instant::now() >= deadline { + return Err(io::ErrorKind::TimedOut.into()); + } + server_pid(deadline) + }; + let pid = probe() .map_err(|error| Error::io("cannot examine the control endpoint").with_source(error))? .ok_or_else(|| Error::io("no process serves the control endpoint"))?; let process = hold(pid) .map_err(|error| Error::io("cannot hold the control endpoint owner").with_source(error))? .ok_or_else(|| Error::io("the control endpoint owner exited before it could be held"))?; - let again = server_pid() + let again = probe() .map_err(|error| Error::io("cannot recheck the control endpoint").with_source(error))?; if again != Some(pid) { return Err(Error::io( diff --git a/nvx-rs/src/host/cli/store/tests/liveness.rs b/nvx-rs/src/host/cli/store/tests/liveness.rs index 6565bfc04..fcc522311 100644 --- a/nvx-rs/src/host/cli/store/tests/liveness.rs +++ b/nvx-rs/src/host/cli/store/tests/liveness.rs @@ -95,7 +95,11 @@ fn replacing_the_log_or_sandbox_never_proves_the_launch_exited() { .to_string() .contains("cannot check the launch log claim") ); - assert!(locked.endpoint_owner(&journal).is_err()); + assert!( + locked + .endpoint_owner(&journal, Instant::now() + Duration::from_secs(5)) + .is_err() + ); assert!(locked.journal().unwrap().is_some()); } @@ -126,7 +130,9 @@ fn the_spawned_process_inherits_the_preparations_claim() { assert!(locked.claim_held(&journal).unwrap()); #[cfg(target_os = "linux")] { - let owner = locked.endpoint_owner(&journal).unwrap(); + let owner = locked + .endpoint_owner(&journal, Instant::now() + Duration::from_secs(5)) + .unwrap(); assert_eq!(owner.reference().pid, child.0.as_ref().unwrap().id()); assert!(!owner.has_exited().unwrap()); } @@ -165,12 +171,19 @@ fn an_endpoint_owner_without_the_claimed_log_is_rejected_without_signaling_it() ); assert!(released(|| state.dir().join("holder-ready").exists())); assert!(locked.claim_held(&journal).unwrap()); - let error = locked.endpoint_owner(&journal).err().unwrap(); + let error = locked + .endpoint_owner(&journal, Instant::now() + Duration::from_secs(5)) + .err() + .unwrap(); assert!(error.to_string().contains("does not hold the launch log")); assert!(other.0.try_wait().unwrap().is_none()); drop(prepared); assert!(released(|| !locked.claim_held(&journal).unwrap())); - assert!(locked.endpoint_owner(&journal).is_err()); + assert!( + locked + .endpoint_owner(&journal, Instant::now() + Duration::from_secs(5)) + .is_err() + ); assert!(other.0.try_wait().unwrap().is_none()); } @@ -198,10 +211,13 @@ fn without_a_log_job_inherited_handles_remain_the_claim() { #[test] fn endpoint_ownership_is_rechecked_only_after_the_server_is_held() { let pid = std::process::id(); + let deadline = Instant::now() + Duration::from_secs(5); let held = std::cell::Cell::new(false); let mut lookups = 0; let owner = hold_endpoint_server( - || { + deadline, + |budget| { + assert_eq!(budget, deadline); lookups += 1; assert_eq!(held.get(), lookups == 2); Ok(Some(pid)) @@ -218,7 +234,8 @@ fn endpoint_ownership_is_rechecked_only_after_the_server_is_held() { for replacement in [None, Some(pid.wrapping_add(1))] { let mut lookups = 0; let error = hold_endpoint_server( - || { + deadline, + |_| { lookups += 1; Ok(if lookups == 1 { Some(pid) } else { replacement }) }, @@ -233,12 +250,16 @@ fn endpoint_ownership_is_rechecked_only_after_the_server_is_held() { #[test] fn missing_owners_and_lookup_or_hold_errors_are_not_exit_proofs() { - assert!(hold_endpoint_server(|| Ok(None), |_| panic!("no owner was found")).is_err()); - assert!(hold_endpoint_server(|| Ok(Some(1)), |_| Ok(None)).is_err()); + let deadline = Instant::now() + Duration::from_secs(5); + assert!( + hold_endpoint_server(deadline, |_| Ok(None), |_| panic!("no owner was found")).is_err() + ); + assert!(hold_endpoint_server(deadline, |_| Ok(Some(1)), |_| Ok(None)).is_err()); for phase in 0..3 { let mut lookups = 0; let error = hold_endpoint_server( - || { + deadline, + |_| { lookups += 1; if (phase == 0 && lookups == 1) || (phase == 2 && lookups == 2) { Err(io::Error::from(io::ErrorKind::PermissionDenied)) @@ -261,6 +282,38 @@ fn missing_owners_and_lookup_or_hold_errors_are_not_exit_proofs() { } } +#[test] +fn an_exhausted_budget_prevents_the_first_probe_and_the_held_owners_recheck() { + for after_hold in [false, true] { + let deadline = Instant::now() + + if after_hold { + Duration::from_secs(1) + } else { + Duration::ZERO + }; + let mut probes = 0; + let error = hold_endpoint_server( + deadline, + |budget| { + assert_eq!(budget, deadline); + probes += 1; + Ok(Some(std::process::id())) + }, + |_| { + let process = own_process(); + thread::sleep(deadline.saturating_duration_since(Instant::now())); + Ok(Some(process)) + }, + ) + .err() + .unwrap(); + let source = error.source().unwrap().downcast_ref::().unwrap(); + assert_eq!(source.kind(), io::ErrorKind::TimedOut); + assert_eq!(probes, usize::from(after_hold)); + assert!(!own_process().has_exited().unwrap()); + } +} + #[test] #[ignore = "invoked as an owned subprocess by the launch-claim tests"] fn log_holder_helper() { diff --git a/nvx-rs/src/host/cli/store/tests/recovery.rs b/nvx-rs/src/host/cli/store/tests/recovery.rs index fae235449..04aaadc86 100644 --- a/nvx-rs/src/host/cli/store/tests/recovery.rs +++ b/nvx-rs/src/host/cli/store/tests/recovery.rs @@ -179,6 +179,63 @@ mod process { assert!(locked.capability().unwrap().is_none()); } + #[test] + fn a_busy_endpoint_cannot_extend_recovery_or_signal_its_live_process() { + use std::error::Error as _; + + let (_directory, state) = state(); + let (prepared, runtime) = prepared(&state, ConsoleAttachment::Disconnected); + let mut locked = state.lock(Deadline::after(10.0)).unwrap(); + drop(locked.begin_launch(&prepared, &runtime).unwrap()); + fs::write(state.dir().join("recovery-small-backlog"), b"").unwrap(); + let child = running(&state, &prepared); + let process = child.process(); + #[cfg(target_os = "linux")] + let _clients = { + let mut clients = Vec::new(); + loop { + assert!(clients.len() < 16, "the helper's backlog must fill"); + match crate::host::sys::connect_nonblocking(&prepared.outputs().endpoint) { + Ok(client) => clients.push(client), + Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => break, + Err(error) => panic!("cannot fill the recovery endpoint: {error}"), + } + } + clients + }; + #[cfg(windows)] + let _client = platform::connect_endpoint( + &prepared.outputs().endpoint, + process.reference().pid, + Duration::from_secs(2), + ) + .unwrap(); + drop(prepared); + let journal = fs::read(state.path(StateFile::Journal)).unwrap(); + let capability = fs::read(state.path(StateFile::Capability)).unwrap(); + for budget in [Duration::ZERO, Duration::from_millis(50)] { + let started = Instant::now(); + let error = locked.recover(started + budget).unwrap_err(); + let source = error + .source() + .unwrap() + .downcast_ref::() + .unwrap(); + assert_eq!(source.kind(), std::io::ErrorKind::TimedOut); + assert!( + started.elapsed() < Duration::from_millis(250), + "{:?}", + started.elapsed() + ); + assert!(!process.has_exited().unwrap()); + assert_eq!(fs::read(state.path(StateFile::Journal)).unwrap(), journal); + assert_eq!( + fs::read(state.path(StateFile::Capability)).unwrap(), + capability + ); + } + } + #[cfg(target_os = "linux")] #[test] fn an_endpoint_owner_without_the_log_is_not_signaled_and_its_journal_stays() { @@ -338,7 +395,11 @@ mod process { fn launch_helper() { let endpoint = fs::read_to_string("recovery-endpoint").unwrap(); #[cfg(target_os = "linux")] - let _server = std::os::unix::net::UnixListener::bind(endpoint).unwrap(); + let server = std::os::unix::net::UnixListener::bind(endpoint).unwrap(); + #[cfg(target_os = "linux")] + if std::path::Path::new("recovery-small-backlog").exists() { + nix::sys::socket::listen(&server, nix::sys::socket::Backlog::new(1).unwrap()).unwrap(); + } #[cfg(windows)] let server = platform::tests::control_server(&endpoint); fs::write("recovery-ready", b"").unwrap(); diff --git a/nvx-rs/src/host/platform/linux/endpoint.rs b/nvx-rs/src/host/platform/linux/endpoint.rs index dc29d19f6..38bd463c3 100644 --- a/nvx-rs/src/host/platform/linux/endpoint.rs +++ b/nvx-rs/src/host/platform/linux/endpoint.rs @@ -273,8 +273,16 @@ pub(crate) fn connect_endpoint( /// Returns the process that listens on a control endpoint, or `None` if nothing listens. pub(crate) fn endpoint_server_pid(endpoint: &str) -> io::Result> { - let deadline = Instant::now() + Duration::from_secs(5); + endpoint_server_pid_until(endpoint, Instant::now() + Duration::from_secs(5)) +} + +/// Looks up the endpoint owner within the caller's absolute deadline. +pub(crate) fn endpoint_server_pid_until( + endpoint: &str, + deadline: Instant, +) -> io::Result> { loop { + time_left(Some(deadline))?; match sys::connect_nonblocking(endpoint) { Ok(stream) => return sys::peer_pid(&stream).map(Some), Err(error) @@ -286,10 +294,9 @@ pub(crate) fn endpoint_server_pid(endpoint: &str) -> io::Result> { return Ok(None); } // A full listen backlog drains as OpenVMM accepts clients. - Err(error) - if error.kind() == io::ErrorKind::WouldBlock && Instant::now() < deadline => - { - thread::sleep(Duration::from_millis(25)); + Err(error) if error.kind() == io::ErrorKind::WouldBlock => { + let left = time_left(Some(deadline))?.expect("the probe has a deadline"); + thread::sleep(left.min(Duration::from_millis(25))); } Err(error) => return Err(error), } diff --git a/nvx-rs/src/host/platform/linux/mod.rs b/nvx-rs/src/host/platform/linux/mod.rs index 516d09b16..d86aa3406 100644 --- a/nvx-rs/src/host/platform/linux/mod.rs +++ b/nvx-rs/src/host/platform/linux/mod.rs @@ -15,7 +15,8 @@ use crate::hypervisor::Hypervisor; pub(crate) use crate::host::sys::{interrupted, watch_interrupts, write_interruptibly}; pub(crate) use self::endpoint::{ - ReadyListener, connect_endpoint, control_endpoint, endpoint_server_pid, listen_ready, + ReadyListener, connect_endpoint, control_endpoint, endpoint_server_pid, + endpoint_server_pid_until, listen_ready, }; pub use self::identity::file_identity; pub(crate) use self::identity::{ diff --git a/nvx-rs/src/host/platform/linux/tests.rs b/nvx-rs/src/host/platform/linux/tests.rs index d977adcae..06d32f2f8 100644 --- a/nvx-rs/src/host/platform/linux/tests.rs +++ b/nvx-rs/src/host/platform/linux/tests.rs @@ -13,6 +13,64 @@ pub(crate) use super::process::{check_identity, start_time_from_stat}; use super::*; use crate::host::platform::Transport; +#[test] +fn a_full_control_backlog_never_replaces_the_callers_probe_deadline() { + use std::os::unix::net::UnixListener; + + use nix::sys::socket::{Backlog, listen}; + + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("control.sock"); + let endpoint = path.to_str().unwrap(); + let listener = UnixListener::bind(&path).unwrap(); + listen(&listener, Backlog::new(1).unwrap()).unwrap(); + let mut clients = Vec::new(); + loop { + assert!(clients.len() < 16, "the configured backlog must fill"); + match crate::host::sys::connect_nonblocking(endpoint) { + Ok(client) => clients.push(client), + Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => break, + Err(error) => panic!("cannot fill the control backlog: {error}"), + } + } + let started = Instant::now(); + let error = + endpoint_server_pid_until(endpoint, started + Duration::from_millis(50)).unwrap_err(); + assert_eq!(error.kind(), std::io::ErrorKind::TimedOut); + assert!( + started.elapsed() < Duration::from_millis(250), + "{:?}", + started.elapsed() + ); + for _ in &clients { + drop(listener.accept().unwrap()); + } + drop(clients); + assert_eq!( + endpoint_server_pid_until(endpoint, started) + .unwrap_err() + .kind(), + std::io::ErrorKind::TimedOut + ); + listener.set_nonblocking(true).unwrap(); + assert_eq!( + listener.accept().unwrap_err().kind(), + std::io::ErrorKind::WouldBlock + ); + assert_eq!( + endpoint_server_pid(endpoint).unwrap(), + Some(std::process::id()) + ); + drop(listener); + fs::remove_file(&path).unwrap(); + assert_eq!( + endpoint_server_pid_until(endpoint, started) + .unwrap_err() + .kind(), + std::io::ErrorKind::TimedOut + ); +} + #[test] fn a_duplicated_transport_keeps_the_verified_connection_without_reconnecting() { use std::io::Write; diff --git a/nvx-rs/src/host/platform/other.rs b/nvx-rs/src/host/platform/other.rs index b4c8586b1..c5096aa45 100644 --- a/nvx-rs/src/host/platform/other.rs +++ b/nvx-rs/src/host/platform/other.rs @@ -285,3 +285,14 @@ pub(crate) fn connect_endpoint( pub(crate) fn endpoint_server_pid(_endpoint: &str) -> io::Result> { Ok(None) } + +#[cfg(feature = "cli")] +pub(crate) fn endpoint_server_pid_until( + _endpoint: &str, + deadline: Instant, +) -> io::Result> { + if Instant::now() >= deadline { + return Err(io::ErrorKind::TimedOut.into()); + } + Err(unsupported()) +} diff --git a/nvx-rs/src/host/platform/windows/endpoint.rs b/nvx-rs/src/host/platform/windows/endpoint.rs index 5ade3b93c..943e7b681 100644 --- a/nvx-rs/src/host/platform/windows/endpoint.rs +++ b/nvx-rs/src/host/platform/windows/endpoint.rs @@ -10,7 +10,7 @@ use std::time::{Duration, Instant}; use windows_sys::Win32::Foundation::{ ERROR_BROKEN_PIPE, ERROR_FILE_NOT_FOUND, ERROR_IO_PENDING, ERROR_MORE_DATA, ERROR_NO_DATA, ERROR_OPERATION_ABORTED, ERROR_PIPE_BUSY, ERROR_PIPE_CONNECTED, ERROR_PIPE_NOT_CONNECTED, - GENERIC_READ, GENERIC_WRITE, HANDLE, WAIT_OBJECT_0, + ERROR_SEM_TIMEOUT, GENERIC_READ, GENERIC_WRITE, HANDLE, WAIT_OBJECT_0, }; use windows_sys::Win32::Storage::FileSystem::{ CreateFileW, FILE_ALL_ACCESS, FILE_FLAG_FIRST_PIPE_INSTANCE, FILE_FLAG_OVERLAPPED, @@ -246,9 +246,17 @@ pub(crate) fn connect_endpoint( /// serves the endpoint. The instance is busy while a client is attached and briefly between /// clients, which is waited out for up to five seconds. pub(crate) fn endpoint_server_pid(endpoint: &str) -> io::Result> { + endpoint_server_pid_until(endpoint, Instant::now() + Duration::from_secs(5)) +} + +/// Looks up the endpoint owner within the caller's absolute deadline. +pub(crate) fn endpoint_server_pid_until( + endpoint: &str, + deadline: Instant, +) -> io::Result> { let name = wide(endpoint.replace('/', "\\")); - let deadline = Instant::now() + Duration::from_secs(5); loop { + time_left(Some(deadline))?; // SAFETY: the name is NUL-terminated; SQOS identification prevents impersonation. let handle = unsafe { CreateFileW( @@ -271,9 +279,20 @@ pub(crate) fn endpoint_server_pid(endpoint: &str) -> io::Result> { } match last_error() { ERROR_FILE_NOT_FOUND => return Ok(None), - ERROR_PIPE_BUSY if Instant::now() < deadline => { + ERROR_PIPE_BUSY => { + let left = time_left(Some(deadline))?.expect("the probe has a deadline"); + let wait = left.as_millis().min(100) as u32; + // Zero selects the server's default timeout, not a nonblocking wait. + if wait == 0 { + return Err(io::ErrorKind::TimedOut.into()); + } // SAFETY: the name is NUL-terminated. - unsafe { WaitNamedPipeW(name.as_ptr(), 100) }; + if unsafe { WaitNamedPipeW(name.as_ptr(), wait) } == 0 { + let error = last_error(); + if error != ERROR_SEM_TIMEOUT { + return Err(io::Error::from_raw_os_error(error as i32)); + } + } } error => return Err(io::Error::from_raw_os_error(error as i32)), } diff --git a/nvx-rs/src/host/platform/windows/endpoint/tests.rs b/nvx-rs/src/host/platform/windows/endpoint/tests.rs index 4d882574a..cced44b26 100644 --- a/nvx-rs/src/host/platform/windows/endpoint/tests.rs +++ b/nvx-rs/src/host/platform/windows/endpoint/tests.rs @@ -27,16 +27,29 @@ pub(crate) fn control_server(endpoint: &str) -> File { pub(crate) fn accept_control(server: &File) -> i32 { use std::os::windows::io::{AsHandle, AsRawHandle}; - // SAFETY: the synchronous server and its handle remain valid while it accepts. - let connected = - unsafe { ConnectNamedPipe(server.as_handle().as_raw_handle(), ptr::null_mut()) }; - if connected == 0 { - assert_eq!(last_error(), ERROR_PIPE_CONNECTED); + let deadline = Instant::now() + Duration::from_secs(5); + let mut stale = 0; + loop { + // SAFETY: the synchronous server and its handle remain valid while it accepts. + let connected = + unsafe { ConnectNamedPipe(server.as_handle().as_raw_handle(), ptr::null_mut()) }; + if connected != 0 { + return connected; + } + let error = last_error(); + if error == ERROR_PIPE_CONNECTED { + return connected; + } + assert_eq!(error, ERROR_NO_DATA, "unexpected control-pipe accept error"); + stale += 1; + assert!( + stale <= 16 && Instant::now() < deadline, + "stale control-pipe connections exhausted the accept budget" + ); + disconnect_control(server); } - connected } -#[cfg(feature = "cli")] pub(crate) fn disconnect_control(server: &File) { use std::os::windows::io::{AsHandle, AsRawHandle}; @@ -49,6 +62,95 @@ pub(crate) fn disconnect_control(server: &File) { ); } +#[test] +fn a_stale_control_probe_is_disconnected_before_the_actual_client_is_accepted() { + let endpoint = format!( + r"\\.\pipe\nvx-stale-probe-{}-{}", + std::process::id(), + crate::host::store::LaunchId::generate().unwrap().as_str() + ); + let server = control_server(&endpoint); + drop(connect_endpoint(&endpoint, std::process::id(), Duration::from_secs(2)).unwrap()); + std::thread::scope(|scope| { + let (ready, accepting) = std::sync::mpsc::channel(); + let accepted = scope.spawn(move || { + ready.send(()).unwrap(); + accept_control(&server); + let mut request = [0]; + (&server).read_exact(&mut request).unwrap(); + assert_eq!(request, [7]); + (&server).write_all(&[9]).unwrap(); + }); + accepting.recv_timeout(Duration::from_secs(2)).unwrap(); + let deadline = Instant::now() + Duration::from_secs(2); + let mut client = loop { + let left = time_left(Some(deadline)).unwrap().unwrap(); + match connect_endpoint( + &endpoint, + std::process::id(), + left.min(Duration::from_millis(25)), + ) { + Ok(client) => break client, + Err(error) if error.kind() == io::ErrorKind::WouldBlock => {} + Err(error) => panic!("the actual control client failed: {error}"), + } + }; + client + .write_all(&[7], Some(Duration::from_secs(2))) + .unwrap(); + let mut response = [0]; + assert_eq!( + client + .read(&mut response, Some(Duration::from_secs(2))) + .unwrap(), + 1 + ); + assert_eq!(response, [9]); + accepted.join().unwrap(); + }); +} + +#[test] +fn a_busy_control_pipe_never_replaces_the_callers_probe_deadline() { + let endpoint = format!( + r"\\.\pipe\nvx-probe-budget-{}-{}", + std::process::id(), + crate::host::store::LaunchId::generate().unwrap().as_str() + ); + let server = control_server(&endpoint); + let client = connect_endpoint(&endpoint, std::process::id(), Duration::from_secs(2)).unwrap(); + for (budget, limit) in [ + (Duration::from_millis(50), Duration::from_millis(250)), + (Duration::from_micros(500), Duration::from_millis(40)), + ] { + let started = Instant::now(); + let deadline = started + budget; + let error = endpoint_server_pid_until(&endpoint, deadline).unwrap_err(); + assert_eq!(error.kind(), io::ErrorKind::TimedOut); + assert!(started.elapsed() < limit, "{:?}", started.elapsed()); + assert_eq!( + endpoint_server_pid_until(&endpoint, started) + .unwrap_err() + .kind(), + io::ErrorKind::TimedOut + ); + } + drop(client); + drop(server); + assert_eq!( + endpoint_server_pid_until(&endpoint, Instant::now()) + .unwrap_err() + .kind(), + io::ErrorKind::TimedOut + ); + let server = control_server(&endpoint); + assert_eq!( + endpoint_server_pid(&endpoint).unwrap(), + Some(std::process::id()) + ); + drop(server); +} + #[test] fn a_duplicated_transport_keeps_the_verified_connection_without_reconnecting() { let endpoint = format!( diff --git a/nvx-rs/src/host/platform/windows/mod.rs b/nvx-rs/src/host/platform/windows/mod.rs index eef623a5c..89222218d 100644 --- a/nvx-rs/src/host/platform/windows/mod.rs +++ b/nvx-rs/src/host/platform/windows/mod.rs @@ -23,7 +23,8 @@ use windows_sys::Win32::System::Threading::OpenProcess; use crate::hypervisor::Hypervisor; pub(crate) use self::endpoint::{ - ReadyListener, connect_endpoint, control_endpoint, endpoint_server_pid, listen_ready, + ReadyListener, connect_endpoint, control_endpoint, endpoint_server_pid, + endpoint_server_pid_until, listen_ready, }; pub use self::identity::file_identity; pub(crate) use self::identity::{ From d9e315ea314eecc519c7c3ee36f4900e6c78edd0 Mon Sep 17 00:00:00 2001 From: Enrique Saurez Date: Tue, 10 Nov 2026 15:05:57 -0800 Subject: [PATCH 4/6] fix: gate recovery probe re-exports to the CLI feature Keep the bounded primitive available to legacy platform wrappers, but expose its new recovery caller only with CLI. Import it directly in Linux platform tests. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: daabb81a-bdcc-435a-b767-7c672945cccc --- nvx-rs/src/host/platform/linux/mod.rs | 5 +++-- nvx-rs/src/host/platform/linux/tests.rs | 2 +- nvx-rs/src/host/platform/windows/mod.rs | 5 +++-- 3 files changed, 7 insertions(+), 5 deletions(-) diff --git a/nvx-rs/src/host/platform/linux/mod.rs b/nvx-rs/src/host/platform/linux/mod.rs index d86aa3406..13e6437d9 100644 --- a/nvx-rs/src/host/platform/linux/mod.rs +++ b/nvx-rs/src/host/platform/linux/mod.rs @@ -14,9 +14,10 @@ use crate::hypervisor::Hypervisor; #[cfg(feature = "cli")] pub(crate) use crate::host::sys::{interrupted, watch_interrupts, write_interruptibly}; +#[cfg(feature = "cli")] +pub(crate) use self::endpoint::endpoint_server_pid_until; pub(crate) use self::endpoint::{ - ReadyListener, connect_endpoint, control_endpoint, endpoint_server_pid, - endpoint_server_pid_until, listen_ready, + ReadyListener, connect_endpoint, control_endpoint, endpoint_server_pid, listen_ready, }; pub use self::identity::file_identity; pub(crate) use self::identity::{ diff --git a/nvx-rs/src/host/platform/linux/tests.rs b/nvx-rs/src/host/platform/linux/tests.rs index 06d32f2f8..04ab1e6b6 100644 --- a/nvx-rs/src/host/platform/linux/tests.rs +++ b/nvx-rs/src/host/platform/linux/tests.rs @@ -6,7 +6,7 @@ use std::process::{Command, Stdio}; use std::thread; use std::time::{Duration, Instant}; -use super::endpoint::UnixTransport; +use super::endpoint::{UnixTransport, endpoint_server_pid_until}; pub(crate) use super::private::{owner_only_problem, store_dir_problem}; use super::process::wait_step; pub(crate) use super::process::{check_identity, start_time_from_stat}; diff --git a/nvx-rs/src/host/platform/windows/mod.rs b/nvx-rs/src/host/platform/windows/mod.rs index 89222218d..dca1e75a0 100644 --- a/nvx-rs/src/host/platform/windows/mod.rs +++ b/nvx-rs/src/host/platform/windows/mod.rs @@ -22,9 +22,10 @@ use windows_sys::Win32::System::Threading::OpenProcess; use crate::hypervisor::Hypervisor; +#[cfg(feature = "cli")] +pub(crate) use self::endpoint::endpoint_server_pid_until; pub(crate) use self::endpoint::{ - ReadyListener, connect_endpoint, control_endpoint, endpoint_server_pid, - endpoint_server_pid_until, listen_ready, + ReadyListener, connect_endpoint, control_endpoint, endpoint_server_pid, listen_ready, }; pub use self::identity::file_identity; pub(crate) use self::identity::{ From 9c864b6ab5b23f9404e565afde4e11333a695dec Mon Sep 17 00:00:00 2001 From: Enrique Saurez Date: Tue, 10 Nov 2026 15:30:17 -0800 Subject: [PATCH 5/6] test: preserve strict duplicated-pipe ownership assertions Keep the duplicate transport test's pre-existing connection check strict rather than hiding a lost client behind recovery's stale-probe accept behavior. Limit stale acceptance to one disconnect/reaccept and avoid scoped-join hangs when the deterministic stale-probe test's real client fails. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: daabb81a-bdcc-435a-b767-7c672945cccc --- .../host/platform/windows/endpoint/tests.rs | 108 +++++++++--------- 1 file changed, 53 insertions(+), 55 deletions(-) diff --git a/nvx-rs/src/host/platform/windows/endpoint/tests.rs b/nvx-rs/src/host/platform/windows/endpoint/tests.rs index cced44b26..46890faba 100644 --- a/nvx-rs/src/host/platform/windows/endpoint/tests.rs +++ b/nvx-rs/src/host/platform/windows/endpoint/tests.rs @@ -27,27 +27,19 @@ pub(crate) fn control_server(endpoint: &str) -> File { pub(crate) fn accept_control(server: &File) -> i32 { use std::os::windows::io::{AsHandle, AsRawHandle}; - let deadline = Instant::now() + Duration::from_secs(5); - let mut stale = 0; - loop { + let connect = || { // SAFETY: the synchronous server and its handle remain valid while it accepts. - let connected = - unsafe { ConnectNamedPipe(server.as_handle().as_raw_handle(), ptr::null_mut()) }; - if connected != 0 { - return connected; - } - let error = last_error(); - if error == ERROR_PIPE_CONNECTED { - return connected; - } - assert_eq!(error, ERROR_NO_DATA, "unexpected control-pipe accept error"); - stale += 1; - assert!( - stale <= 16 && Instant::now() < deadline, - "stale control-pipe connections exhausted the accept budget" - ); + unsafe { ConnectNamedPipe(server.as_handle().as_raw_handle(), ptr::null_mut()) } + }; + let mut connected = connect(); + if connected == 0 && last_error() == ERROR_NO_DATA { disconnect_control(server); + connected = connect(); } + if connected == 0 { + assert_eq!(last_error(), ERROR_PIPE_CONNECTED); + } + connected } pub(crate) fn disconnect_control(server: &File) { @@ -71,43 +63,41 @@ fn a_stale_control_probe_is_disconnected_before_the_actual_client_is_accepted() ); let server = control_server(&endpoint); drop(connect_endpoint(&endpoint, std::process::id(), Duration::from_secs(2)).unwrap()); - std::thread::scope(|scope| { - let (ready, accepting) = std::sync::mpsc::channel(); - let accepted = scope.spawn(move || { - ready.send(()).unwrap(); - accept_control(&server); - let mut request = [0]; - (&server).read_exact(&mut request).unwrap(); - assert_eq!(request, [7]); - (&server).write_all(&[9]).unwrap(); - }); - accepting.recv_timeout(Duration::from_secs(2)).unwrap(); - let deadline = Instant::now() + Duration::from_secs(2); - let mut client = loop { - let left = time_left(Some(deadline)).unwrap().unwrap(); - match connect_endpoint( - &endpoint, - std::process::id(), - left.min(Duration::from_millis(25)), - ) { - Ok(client) => break client, - Err(error) if error.kind() == io::ErrorKind::WouldBlock => {} - Err(error) => panic!("the actual control client failed: {error}"), - } - }; - client - .write_all(&[7], Some(Duration::from_secs(2))) - .unwrap(); - let mut response = [0]; - assert_eq!( - client - .read(&mut response, Some(Duration::from_secs(2))) - .unwrap(), - 1 - ); - assert_eq!(response, [9]); - accepted.join().unwrap(); + let (ready, accepting) = std::sync::mpsc::channel(); + let accepted = std::thread::spawn(move || { + ready.send(()).unwrap(); + accept_control(&server); + let mut request = [0]; + (&server).read_exact(&mut request).unwrap(); + assert_eq!(request, [7]); + (&server).write_all(&[9]).unwrap(); }); + accepting.recv_timeout(Duration::from_secs(2)).unwrap(); + let deadline = Instant::now() + Duration::from_secs(2); + let mut client = loop { + let left = time_left(Some(deadline)).unwrap().unwrap(); + match connect_endpoint( + &endpoint, + std::process::id(), + left.min(Duration::from_millis(25)), + ) { + Ok(client) => break client, + Err(error) if error.kind() == io::ErrorKind::WouldBlock => {} + Err(error) => panic!("the actual control client failed: {error}"), + } + }; + client + .write_all(&[7], Some(Duration::from_secs(2))) + .unwrap(); + let mut response = [0]; + assert_eq!( + client + .read(&mut response, Some(Duration::from_secs(2))) + .unwrap(), + 1 + ); + assert_eq!(response, [9]); + accepted.join().unwrap(); } #[test] @@ -153,6 +143,8 @@ fn a_busy_control_pipe_never_replaces_the_callers_probe_deadline() { #[test] fn a_duplicated_transport_keeps_the_verified_connection_without_reconnecting() { + use std::os::windows::io::{AsHandle, AsRawHandle}; + let endpoint = format!( r"\\.\pipe\nvx-transport-{}-{}", std::process::id(), @@ -164,7 +156,13 @@ fn a_duplicated_transport_keeps_the_verified_connection_without_reconnecting() { drop(transport.try_clone_handle().unwrap().unwrap()); let pipe = transport.try_clone_handle().unwrap().unwrap(); drop(transport); - assert_eq!(accept_control(&server), 0); + // A replacement connection would hide a lost duplicate, so this check stays strict. + // SAFETY: the owned synchronous server remains valid while it accepts. + assert_eq!( + unsafe { ConnectNamedPipe(server.as_handle().as_raw_handle(), ptr::null_mut()) }, + 0 + ); + assert_eq!(last_error(), ERROR_PIPE_CONNECTED); // SAFETY: this creates a fresh unnamed manual-reset event. let event = owned(unsafe { CreateEventW(ptr::null(), 1, 0, ptr::null()) }).unwrap(); let mut duplicate = PipeTransport { pipe, event }; From 971103c782175dee4963005cdb7d65f4c871fc59 Mon Sep 17 00:00:00 2001 From: Enrique Saurez Date: Tue, 10 Nov 2026 15:51:58 -0800 Subject: [PATCH 6/6] test: reap owned stale-pipe fixtures on every client outcome Replace the standalone accept thread with an explicitly invoked owned test subprocess. A non-inheritable kill-on-close job and held child handle bound cleanup on connection errors, timeouts and panic without abandoning a thread. Check the original client cause and a signaled held process handle after teardown, before and after releasing the accept protocol. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: daabb81a-bdcc-435a-b767-7c672945cccc --- .../host/platform/windows/endpoint/tests.rs | 327 +++++++++++++++--- 1 file changed, 281 insertions(+), 46 deletions(-) diff --git a/nvx-rs/src/host/platform/windows/endpoint/tests.rs b/nvx-rs/src/host/platform/windows/endpoint/tests.rs index 46890faba..a29192927 100644 --- a/nvx-rs/src/host/platform/windows/endpoint/tests.rs +++ b/nvx-rs/src/host/platform/windows/endpoint/tests.rs @@ -1,7 +1,14 @@ use std::fs::File; use std::io::{Read, Write}; +use std::os::windows::io::{AsHandle, AsRawHandle}; +use std::process::{Child, Command, ExitStatus, Stdio}; use windows_sys::Win32::Storage::FileSystem::PIPE_ACCESS_DUPLEX; +use windows_sys::Win32::System::JobObjects::{ + AssignProcessToJobObject, CreateJobObjectW, JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE, + JOBOBJECT_EXTENDED_LIMIT_INFORMATION, JobObjectExtendedLimitInformation, + SetInformationJobObject, +}; use super::*; @@ -25,8 +32,6 @@ pub(crate) fn control_server(endpoint: &str) -> File { } pub(crate) fn accept_control(server: &File) -> i32 { - use std::os::windows::io::{AsHandle, AsRawHandle}; - let connect = || { // SAFETY: the synchronous server and its handle remain valid while it accepts. unsafe { ConnectNamedPipe(server.as_handle().as_raw_handle(), ptr::null_mut()) } @@ -43,8 +48,6 @@ pub(crate) fn accept_control(server: &File) -> i32 { } pub(crate) fn disconnect_control(server: &File) { - use std::os::windows::io::{AsHandle, AsRawHandle}; - use windows_sys::Win32::System::Pipes::DisconnectNamedPipe; // SAFETY: the owned synchronous pipe finished its previous client. @@ -54,50 +57,284 @@ pub(crate) fn disconnect_control(server: &File) { ); } +struct StaleServer { + child: Option, + job: Option, + directory: tempfile::TempDir, + endpoint: String, +} + +impl StaleServer { + fn new() -> io::Result { + let directory = tempfile::tempdir()?; + let endpoint = format!( + r"\\.\pipe\nvx-stale-probe-{}-{}", + std::process::id(), + crate::host::store::LaunchId::generate().unwrap().as_str() + ); + // SAFETY: this creates an unnamed, non-inheritable job for this fixture alone. + let job = owned(unsafe { CreateJobObjectW(ptr::null(), ptr::null()) }) + .ok_or_else(io::Error::last_os_error)?; + // SAFETY: this Win32 structure is valid with every field zeroed. + let mut limits: JOBOBJECT_EXTENDED_LIMIT_INFORMATION = unsafe { std::mem::zeroed() }; + limits.BasicLimitInformation.LimitFlags = JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE; + // SAFETY: the job and the initialized input buffer remain valid for the call. + if unsafe { + SetInformationJobObject( + raw(&job), + JobObjectExtendedLimitInformation, + ptr::from_ref(&limits).cast(), + std::mem::size_of_val(&limits) as u32, + ) + } == 0 + { + return Err(io::Error::last_os_error()); + } + let output = File::create(directory.path().join("helper.log"))?; + let mut command = Command::new(std::env::current_exe()?); + command + .args([ + "--exact", + "host::platform::windows::endpoint::tests::stale_probe_accept_helper", + "--ignored", + "--nocapture", + "--test-threads=1", + ]) + .env("NVX_TEST_STALE_CONTROL_ENDPOINT", &endpoint) + .current_dir(directory.path()) + .stdin(Stdio::null()) + .stdout(Stdio::from(output.try_clone()?)) + .stderr(Stdio::from(output)); + let child = crate::host::tests::spawn_locked(&mut command)?; + let server = Self { + child: Some(child), + job: Some(job), + directory, + endpoint, + }; + // SAFETY: both owned handles identify this fixture's job and its exact child. + if unsafe { + AssignProcessToJobObject( + raw(server.job.as_ref().unwrap()), + server.child.as_ref().unwrap().as_raw_handle(), + ) + } == 0 + { + return Err(io::Error::last_os_error()); + } + server.ready("ready")?; + drop(connect_endpoint( + &server.endpoint, + server.child.as_ref().unwrap().id(), + Duration::from_secs(2), + )?); + Ok(server) + } + + fn ready(&self, name: &str) -> io::Result<()> { + if crate::host::tests::released(|| self.directory.path().join(name).exists()) { + Ok(()) + } else { + Err(io::Error::new( + io::ErrorKind::TimedOut, + "the pipe helper was not ready", + )) + } + } + + fn accept(&self) -> io::Result<()> { + std::fs::write(self.directory.path().join("accept"), b"")?; + self.ready("accepting") + } + + fn wait(&mut self, deadline: Instant) -> io::Result> { + loop { + if let Some(status) = self.child.as_mut().unwrap().try_wait()? { + return Ok(Some(status)); + } + let left = deadline.saturating_duration_since(Instant::now()); + if left.is_zero() { + return Ok(None); + } + std::thread::sleep(left.min(Duration::from_millis(25))); + } + } + + fn stop(&mut self) -> io::Result<()> { + drop(self.job.take()); + if self.child.is_none() { + return Ok(()); + } + // Also owns cleanup if admission failed before the child entered the job. + let termination = self.child.as_mut().unwrap().kill(); + if self + .wait(Instant::now() + Duration::from_secs(2))? + .is_some() + { + drop(self.child.take()); + return Ok(()); + } + Err(io::Error::new( + io::ErrorKind::TimedOut, + format!("cannot confirm the pipe helper exited; termination: {termination:?}"), + )) + } + + fn finish(mut self, client: io::Result<()>) -> io::Result<()> { + if client.is_err() { + if let Err(error) = self.stop() { + eprintln!("pipe-helper cleanup also failed: {error}"); + } + return client; + } + let status = self + .wait(Instant::now() + Duration::from_secs(2))? + .ok_or_else(|| { + io::Error::new(io::ErrorKind::TimedOut, "the pipe helper did not exit") + })?; + drop(self.child.take()); + if !status.success() { + return Err(io::Error::other(format!( + "the pipe helper exited with {status}: {}", + std::fs::read_to_string(self.directory.path().join("helper.log"))? + ))); + } + Ok(()) + } +} + +impl Drop for StaleServer { + fn drop(&mut self) { + if let Err(error) = self.stop() { + eprintln!("cannot clean up the owned pipe helper: {error}"); + } + } +} + #[test] fn a_stale_control_probe_is_disconnected_before_the_actual_client_is_accepted() { - let endpoint = format!( - r"\\.\pipe\nvx-stale-probe-{}-{}", - std::process::id(), - crate::host::store::LaunchId::generate().unwrap().as_str() - ); - let server = control_server(&endpoint); - drop(connect_endpoint(&endpoint, std::process::id(), Duration::from_secs(2)).unwrap()); - let (ready, accepting) = std::sync::mpsc::channel(); - let accepted = std::thread::spawn(move || { - ready.send(()).unwrap(); - accept_control(&server); - let mut request = [0]; - (&server).read_exact(&mut request).unwrap(); - assert_eq!(request, [7]); - (&server).write_all(&[9]).unwrap(); - }); - accepting.recv_timeout(Duration::from_secs(2)).unwrap(); + let server = StaleServer::new().unwrap(); + server.accept().unwrap(); let deadline = Instant::now() + Duration::from_secs(2); - let mut client = loop { - let left = time_left(Some(deadline)).unwrap().unwrap(); - match connect_endpoint( - &endpoint, - std::process::id(), - left.min(Duration::from_millis(25)), - ) { - Ok(client) => break client, - Err(error) if error.kind() == io::ErrorKind::WouldBlock => {} - Err(error) => panic!("the actual control client failed: {error}"), + let result = (|| { + let mut client = loop { + let left = time_left(Some(deadline))?.unwrap(); + match connect_endpoint( + &server.endpoint, + server.child.as_ref().unwrap().id(), + left.min(Duration::from_millis(25)), + ) { + Ok(client) => break client, + Err(error) if error.kind() == io::ErrorKind::WouldBlock => {} + Err(error) => return Err(error), + } + }; + client.write_all(&[7], Some(Duration::from_secs(2)))?; + let mut response = [0]; + assert_eq!(client.read(&mut response, Some(Duration::from_secs(2)))?, 1); + assert_eq!(response, [9]); + Ok(()) + })(); + server.finish(result).unwrap(); +} + +#[test] +fn client_failure_timeout_and_panic_reap_the_stale_accept_helper_and_preserve_the_cause() { + for (accepting, failure) in [ + (false, "connect"), + (true, "connect"), + (false, "panic"), + (true, "panic"), + (false, "timeout"), + ] { + let server = StaleServer::new().unwrap(); + let process = server + .child + .as_ref() + .unwrap() + .as_handle() + .try_clone_to_owned() + .unwrap(); + if accepting { + server.accept().unwrap(); } - }; - client - .write_all(&[7], Some(Duration::from_secs(2))) - .unwrap(); - let mut response = [0]; - assert_eq!( - client - .read(&mut response, Some(Duration::from_secs(2))) - .unwrap(), - 1 - ); - assert_eq!(response, [9]); - accepted.join().unwrap(); + let client_error = if failure == "timeout" { + let deadline = Instant::now() + Duration::from_millis(50); + loop { + let left = match time_left(Some(deadline)) { + Ok(left) => left.unwrap(), + Err(error) => break error, + }; + match connect_endpoint( + &server.endpoint, + server.child.as_ref().unwrap().id(), + left.min(Duration::from_millis(25)), + ) { + Err(error) if error.kind() == io::ErrorKind::WouldBlock => {} + Err(error) => { + panic!("unexpected failure waiting for the busy fixture: {error}") + } + Ok(_) => panic!("the unreleased stale fixture must stay busy"), + } + } + } else { + connect_endpoint( + &format!("{}-missing", server.endpoint), + server.child.as_ref().unwrap().id(), + Duration::from_secs(2), + ) + .err() + .expect("the deliberately missing client endpoint must fail") + }; + let original = ( + client_error.kind(), + client_error.raw_os_error(), + client_error.to_string(), + ); + let started = Instant::now(); + let failed = std::panic::catch_unwind(std::panic::AssertUnwindSafe(move || { + if failure == "panic" { + let _owned = server; + std::panic::panic_any("forced stale-client panic"); + } + server.finish(Err(client_error)) + })); + if failure == "panic" { + assert_eq!( + failed.unwrap_err().downcast_ref::<&str>(), + Some(&"forced stale-client panic") + ); + } else { + let error = failed.unwrap().unwrap_err(); + assert_eq!(error.kind(), original.0); + assert_eq!(error.raw_os_error(), original.1); + assert_eq!(error.to_string(), original.2); + } + // SAFETY: this held process handle still names the exact owned helper, not a reused PID. + assert_eq!( + unsafe { WaitForSingleObject(raw(&process), 0) }, + WAIT_OBJECT_0 + ); + assert!(started.elapsed() < Duration::from_secs(5)); + } +} + +#[test] +#[ignore = "invoked as an owned subprocess by the stale-pipe tests"] +fn stale_probe_accept_helper() { + let endpoint = std::env::var("NVX_TEST_STALE_CONTROL_ENDPOINT").unwrap(); + let server = control_server(&endpoint); + std::fs::write("ready", b"").unwrap(); + assert!(crate::host::tests::released(|| std::path::Path::new( + "accept" + ) + .exists())); + std::fs::write("accepting", b"").unwrap(); + accept_control(&server); + let mut request = [0]; + (&server).read_exact(&mut request).unwrap(); + assert_eq!(request, [7]); + (&server).write_all(&[9]).unwrap(); } #[test] @@ -143,8 +380,6 @@ fn a_busy_control_pipe_never_replaces_the_callers_probe_deadline() { #[test] fn a_duplicated_transport_keeps_the_verified_connection_without_reconnecting() { - use std::os::windows::io::{AsHandle, AsRawHandle}; - let endpoint = format!( r"\\.\pipe\nvx-transport-{}-{}", std::process::id(),