diff --git a/nvx-rs/src/host/cli/store/launch.rs b/nvx-rs/src/host/cli/store/launch.rs index d0a5c04e6..8d93f101d 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, deadline)?; + 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/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/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/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 new file mode 100644 index 000000000..04aaadc86 --- /dev/null +++ b/nvx-rs/src/host/cli/store/tests/recovery.rs @@ -0,0 +1,423 @@ +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()); + } + + #[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() { + 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(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(); + #[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); + } + } + } +} 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..13e6437d9 100644 --- a/nvx-rs/src/host/platform/linux/mod.rs +++ b/nvx-rs/src/host/platform/linux/mod.rs @@ -14,6 +14,8 @@ 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, listen_ready, }; diff --git a/nvx-rs/src/host/platform/linux/tests.rs b/nvx-rs/src/host/platform/linux/tests.rs index d977adcae..04ab1e6b6 100644 --- a/nvx-rs/src/host/platform/linux/tests.rs +++ b/nvx-rs/src/host/platform/linux/tests.rs @@ -6,13 +6,71 @@ 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}; 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/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.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 42e283d69..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,17 +32,352 @@ 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()) }; + 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()) } + }; + 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) { + 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 + ); +} + +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 server = StaleServer::new().unwrap(); + server.accept().unwrap(); + let deadline = Instant::now() + Duration::from_secs(2); + 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(); + } + 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] +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!( @@ -49,7 +391,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 }; diff --git a/nvx-rs/src/host/platform/windows/mod.rs b/nvx-rs/src/host/platform/windows/mod.rs index eef623a5c..dca1e75a0 100644 --- a/nvx-rs/src/host/platform/windows/mod.rs +++ b/nvx-rs/src/host/platform/windows/mod.rs @@ -22,6 +22,8 @@ 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, listen_ready, }; 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};