From d2fd8996ebe9695b0d886da6f01587a11b518682 Mon Sep 17 00:00:00 2001 From: Dom Giandinoto Date: Fri, 16 Oct 2026 15:58:22 -0700 Subject: [PATCH 1/4] init impl --- .../isolation-session/state-aware-rust.md | 18 + .../isolation_session/common/manager.rs | 124 ++++--- .../isolation_session/common/pipe_relay.rs | 332 ++++++++++++++++-- .../isolation_session/common/state_aware.rs | 83 ++++- .../src/core/mxc_common/process_util.rs | 4 + 5 files changed, 463 insertions(+), 98 deletions(-) diff --git a/docs/development/architecture/backends/isolation-session/state-aware-rust.md b/docs/development/architecture/backends/isolation-session/state-aware-rust.md index f7ac7d06c..a04d7601a 100644 --- a/docs/development/architecture/backends/isolation-session/state-aware-rust.md +++ b/docs/development/architecture/backends/isolation-session/state-aware-rust.md @@ -421,6 +421,24 @@ waiter, so the caller decides when to block. Either way, two concurrent exec calls against the same `sandboxId` are not coordinated by MXC; the OS-side service serialises (or rejects, depending on session state) at its own layer. +### Output draining after relayed execution + +After waiting for the workload and stopping stdin forwarding, `create_process` +copies remaining stdout and stderr until both finish, neither successfully +writes output for five seconds, or 60 seconds pass, whichever comes first. +These limits do not apply to SDK `Piped` execution. + +On timeout, MXC cancels reads, warns that output may be missing, and briefly +waits for the copying threads to exit. Cancellation can discard data already +read, but cannot interrupt blocked writes or flushes. Those threads may keep +pipe handles open and finish writing after cleanup returns. Descendant +processes and stdin handling are unchanged. + +Lifecycle `Relayed` execution sends warnings to host stderr and configured +diagnostic output, preserving the workload exit code if warning delivery +succeeds. Create-and-run execution stores warnings in the caller's logger; +`wxc-exec` prints them to stderr. + ### Deprovision and concurrent sandboxes `deprovision` removes only its own agent user (`deprovision_agent_user`). diff --git a/src/mxc-sdk/src/backends/isolation_session/common/manager.rs b/src/mxc-sdk/src/backends/isolation_session/common/manager.rs index 7ec4c8b79..bd2839595 100644 --- a/src/mxc-sdk/src/backends/isolation_session/common/manager.rs +++ b/src/mxc-sdk/src/backends/isolation_session/common/manager.rs @@ -7,11 +7,11 @@ //! against the local console. use std::sync::Arc; +use std::time::Duration; use crate::mxc_common::audit::{AuditEvent, AuditEventName, KillMethod, TeardownStatus}; use crate::mxc_common::logger::Logger; -use crate::mxc_common::process_util::{OwnedHandle, PipeReadCanceller}; -use crate::mxc_common::sandbox_process::StreamCloser; +use crate::mxc_common::process_util::OwnedHandle; use crate::mxc_common::state_aware_backend::{ExecHandle, ExecOutcome}; use crate::isolation_session_bindings::bindings::{ @@ -35,7 +35,8 @@ use super::error::{ }; use super::owned_thread::{self, Impersonation}; use super::pipe_relay::{ - create_relay_thread, create_relay_thread_with_stop, duplicate_handle, PipeRelayWithStopParams, + create_relay_thread, create_relay_thread_with_stop, drain_output_relays_within, + duplicate_handle, OutputRelayDrain, PipeRelay, PipeRelayWithStopParams, }; use super::process_options::{build_iso_process_options, ProcessOptions}; @@ -491,28 +492,28 @@ impl IsolationSessionManager { } scope.stdout = if stdout_handle_val != 0 { - let (thread, canceller) = unsafe { - create_relay_thread( - HANDLE(stdout_handle_val as *mut core::ffi::c_void), - wxc_stdout, - ) - } - .map_err(|e| lifecycle_err(format!("create stdout relay: {}", e)))?; - scope.output_cancellers.push(canceller); - Some(thread) + Some( + unsafe { + create_relay_thread( + HANDLE(stdout_handle_val as *mut core::ffi::c_void), + wxc_stdout, + ) + } + .map_err(|e| lifecycle_err(format!("create stdout relay: {}", e)))?, + ) } else { None }; scope.stderr = if stderr_handle_val != 0 { - let (thread, canceller) = unsafe { - create_relay_thread( - HANDLE(stderr_handle_val as *mut core::ffi::c_void), - wxc_stderr, - ) - } - .map_err(|e| lifecycle_err(format!("create stderr relay: {}", e)))?; - scope.output_cancellers.push(canceller); - Some(thread) + Some( + unsafe { + create_relay_thread( + HANDLE(stderr_handle_val as *mut core::ffi::c_void), + wxc_stderr, + ) + } + .map_err(|e| lifecycle_err(format!("create stderr relay: {}", e)))?, + ) } else { None }; @@ -581,7 +582,7 @@ impl IsolationSessionManager { let outcome = started.wait(options.timeout_ms)?; scope.stop_stdin(); - scope.finish(); + scope.finish(logger); Ok(match outcome { ExecOutcome::Exited(exit_code) => exit_code, @@ -1022,26 +1023,23 @@ impl StartedProcess { /// Owns the relay threads and the workload for the span of `create_process`. /// /// [`finish`](Self::finish) tears down a workload that has already exited. -/// Dropping without it abandons the run: the workload is terminated so the -/// output relays reach EOF. Every join is bounded, so a relay whose read never -/// returns cannot wedge the caller's thread. +/// Dropping without it cancels output reads and terminates the workload. +/// Every join is bounded; cancellation does not interrupt destination writes. struct RelayScope<'a> { process: &'a IsoSessionProcess, stop_event: HANDLE, - stdout: Option, - stderr: Option, + stdout: Option, + stderr: Option, stdin: Option, - /// Ends the output relays' reads. A descendant that inherited the write - /// ends keeps them open past the workload's exit, so EOF alone is not a - /// bound. - output_cancellers: Vec, finished: bool, } impl<'a> RelayScope<'a> { const JOIN_MS: u32 = 1000; - /// Expiry cancels the read, dropping an in-flight chunk. - const DRAIN_MS: u32 = 5000; + /// Stop waiting only after all output relays are idle for this long. + const IDLE_MS: u64 = 5000; + /// Overall cap so a pathological one-byte-forever trickle still returns. + const ABSOLUTE_CAP_MS: u64 = 60_000; fn new(process: &'a IsoSessionProcess, stop_event: HANDLE) -> Self { Self { @@ -1050,7 +1048,6 @@ impl<'a> RelayScope<'a> { stdout: None, stderr: None, stdin: None, - output_cancellers: Vec::new(), finished: false, } } @@ -1062,14 +1059,8 @@ impl<'a> RelayScope<'a> { } } - fn cancel_output_reads(&self) { - for canceller in &self.output_cancellers { - canceller.close(); - } - } - /// Whether the relay exited. An absent relay was never spawned. - fn join(relay: &Option, timeout_ms: u32) -> bool { + fn join(relay: Option<&OwnedHandle>, timeout_ms: u32) -> bool { match relay { // SAFETY: an owned thread handle held by this scope. Some(t) => (unsafe { WaitForSingleObject(t.get(), timeout_ms) }) == WAIT_OBJECT_0, @@ -1077,28 +1068,47 @@ impl<'a> RelayScope<'a> { } } - /// Each relay reads its own duplicate, so one still running does not keep - /// the pipe handles alive. + fn join_output(relay: &Option, timeout_ms: u32) -> bool { + Self::join(relay.as_ref().map(PipeRelay::thread), timeout_ms) + } + + fn output_relays(&self) -> Vec<&PipeRelay> { + [&self.stdout, &self.stderr] + .into_iter() + .filter_map(Option::as_ref) + .collect() + } + + /// Release the process's handles. A still-running relay owns its duplicates + /// until it exits. fn close_handles(&self) { let _ = self.process.Close(); } fn stop_stdin(&self) { self.signal_stop(); - Self::join(&self.stdin, Self::JOIN_MS); + Self::join(self.stdin.as_ref(), Self::JOIN_MS); } /// Teardown for a workload that has already exited. - fn finish(&mut self) { + fn finish(&mut self, mut logger: Option<&mut Logger>) { self.signal_stop(); - let stdout_drained = Self::join(&self.stdout, Self::DRAIN_MS); - let stderr_drained = Self::join(&self.stderr, Self::DRAIN_MS); - if !(stdout_drained && stderr_drained) { - self.cancel_output_reads(); - Self::join(&self.stdout, Self::JOIN_MS); - Self::join(&self.stderr, Self::JOIN_MS); + let output_relays = self.output_relays(); + let output_drain = drain_output_relays_within( + &output_relays, + Duration::from_millis(Self::IDLE_MS), + Duration::from_millis(Self::ABSOLUTE_CAP_MS), + |warning| { + if let Some(logger) = &mut logger { + logger.warning_line(warning); + } + }, + ); + if output_drain == OutputRelayDrain::GaveUp { + Self::join_output(&self.stdout, Self::JOIN_MS); + Self::join_output(&self.stderr, Self::JOIN_MS); } - Self::join(&self.stdin, Self::JOIN_MS); + Self::join(self.stdin.as_ref(), Self::JOIN_MS); self.close_handles(); self.finished = true; } @@ -1110,11 +1120,13 @@ impl Drop for RelayScope<'_> { return; } self.signal_stop(); - self.cancel_output_reads(); + for relay in self.output_relays() { + relay.cancel_read(); + } let _ = self.process.Terminate(); - Self::join(&self.stdout, Self::JOIN_MS); - Self::join(&self.stderr, Self::JOIN_MS); - Self::join(&self.stdin, Self::JOIN_MS); + Self::join_output(&self.stdout, Self::JOIN_MS); + Self::join_output(&self.stderr, Self::JOIN_MS); + Self::join(self.stdin.as_ref(), Self::JOIN_MS); self.close_handles(); } } diff --git a/src/mxc-sdk/src/backends/isolation_session/common/pipe_relay.rs b/src/mxc-sdk/src/backends/isolation_session/common/pipe_relay.rs index e179f9b54..f9dbb8b83 100644 --- a/src/mxc-sdk/src/backends/isolation_session/common/pipe_relay.rs +++ b/src/mxc-sdk/src/backends/isolation_session/common/pipe_relay.rs @@ -6,24 +6,36 @@ //! desktop-session boundary. //! //! Two relay variants: -//! - [`create_relay_thread`] — cancellable; used for stdout / stderr. Ends at -//! EOF or when its [`PipeReadCanceller`] fires. +//! - [`create_relay_thread`] — cancellable and progress-reporting; used for +//! stdout / stderr. Ends at EOF or when its +//! [`PipeReadCanceller`](crate::mxc_common::process_util::PipeReadCanceller) +//! fires. //! - [`create_relay_thread_with_stop`] — stop-event-aware; used for stdin //! in TTY (ConPTY) mode where the agent can exit while the local stdin //! handle remains open. use std::io::Read; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::Arc; +use std::time::{Duration, Instant}; use crate::mxc_common::error::WxcError; use crate::mxc_common::process_util::{InterruptiblePipeReader, OwnedHandle, PipeReadCanceller}; +use crate::mxc_common::sandbox_process::StreamCloser; use windows::Win32::Foundation::{DuplicateHandle, DUPLICATE_SAME_ACCESS, HANDLE, WAIT_OBJECT_0}; use windows::Win32::Storage::FileSystem::{FlushFileBuffers, ReadFile, WriteFile}; use windows::Win32::System::Threading::{ - CreateThread, GetCurrentProcess, WaitForMultipleObjects, THREAD_CREATION_FLAGS, + CreateThread, GetCurrentProcess, WaitForMultipleObjects, WaitForSingleObject, + THREAD_CREATION_FLAGS, }; const BUFFER_SIZE: u32 = 4096; +const DRAIN_POLL_INTERVAL: Duration = Duration::from_millis(200); + +pub(super) const OUTPUT_RELAY_TRUNCATION_WARNING: &str = + "Warning: isolation session output may be truncated because output draining reached its idle \ + or overall time limit before EOF; a descendant may still be holding the output pipe open"; pub(super) fn duplicate_handle(handle: HANDLE) -> Result { let mut duplicated = HANDLE::default(); @@ -45,12 +57,43 @@ pub(super) fn duplicate_handle(handle: HANDLE) -> Result /// Parameters for a pipe relay thread. The thread reads `reader` and writes /// every chunk to `h_write`, flushing after each write. /// +/// `progress_bytes` is bumped after each successful destination write. The +/// teardown path polls it to distinguish a healthy slow drain from a relay that +/// is waiting for EOF because a descendant inherited a write end. +/// /// # Ownership /// Ownership of this struct transfers to the relay thread, which frees it on /// exit. The caller must **not** keep or free it. Both handles are owned. struct PipeRelayParams { reader: InterruptiblePipeReader, h_write: OwnedHandle, + progress_bytes: Arc, +} + +pub(super) struct PipeRelay { + thread: OwnedHandle, + canceller: PipeReadCanceller, + progress_bytes: Arc, +} + +impl PipeRelay { + pub(super) fn thread(&self) -> &OwnedHandle { + &self.thread + } + + pub(super) fn cancel_read(&self) { + self.canceller.close(); + } + + fn progress_bytes(&self) -> u64 { + self.progress_bytes.load(Ordering::Relaxed) + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(super) enum OutputRelayDrain { + Drained, + GaveUp, } /// Thread procedure for relaying data between two handles. @@ -83,6 +126,9 @@ unsafe extern "system" fn pipe_relay_thread_proc(param: *mut core::ffi::c_void) { break; } + params + .progress_bytes + .fetch_add(bytes_written as u64, Ordering::Relaxed); let _ = FlushFileBuffers(params.h_write.get()); } @@ -90,8 +136,8 @@ unsafe extern "system" fn pipe_relay_thread_proc(param: *mut core::ffi::c_void) 0 } -/// Create a relay thread via `CreateThread`. Returns the thread HANDLE wrapped -/// in `OwnedHandle`, plus a canceller that ends the relay's read on demand. +/// Create a relay thread via `CreateThread`, returning a `PipeRelay` that +/// wraps the thread handle, a read canceller, and a shared progress counter. /// /// The thread reads a **duplicate** of `h_read` and writes a **duplicate** of /// `h_write`, so releasing either source handle cannot invalidate it, and @@ -100,12 +146,14 @@ unsafe extern "system" fn pipe_relay_thread_proc(param: *mut core::ffi::c_void) pub(super) unsafe fn create_relay_thread( h_read: HANDLE, h_write: HANDLE, -) -> Result<(OwnedHandle, PipeReadCanceller), WxcError> { +) -> Result { let reader = InterruptiblePipeReader::new(duplicate_handle(h_read)?); let canceller = reader.canceller(); + let progress_bytes = Arc::new(AtomicU64::new(0)); let raw = Box::into_raw(Box::new(PipeRelayParams { reader, h_write: duplicate_handle(h_write)?, + progress_bytes: Arc::clone(&progress_bytes), })); match unsafe { CreateThread( @@ -117,7 +165,11 @@ pub(super) unsafe fn create_relay_thread( None, ) } { - Ok(handle) => Ok((OwnedHandle::new(handle), canceller)), + Ok(handle) => Ok(PipeRelay { + thread: OwnedHandle::new(handle), + canceller, + progress_bytes, + }), Err(e) => { // The thread never started, so nothing will run the proc that // frees the box. Reclaim it here or it leaks. @@ -130,6 +182,71 @@ pub(super) unsafe fn create_relay_thread( } } +/// Wait for output relays to finish while they are still making progress. +/// +/// Each successful output write restarts the idle timeout, but not the overall +/// timeout. On timeout, reads cancel. Blocked writes or flushes cannot be +/// cancelled, so callers must also limit how long they wait for the relay +/// threads to exit. +pub(super) fn drain_output_relays_within( + relays: &[&PipeRelay], + idle_window: Duration, + absolute_cap: Duration, + on_give_up: impl FnOnce(&str), +) -> OutputRelayDrain { + if relays.is_empty() || relays.iter().all(|relay| relay_is_finished(relay)) { + return OutputRelayDrain::Drained; + } + + let started = Instant::now(); + let mut idle_started = started; + let mut last_progress = total_progress(relays); + + loop { + if relays.iter().all(|relay| relay_is_finished(relay)) { + return OutputRelayDrain::Drained; + } + + let now = Instant::now(); + let progress = total_progress(relays); + if progress != last_progress { + last_progress = progress; + idle_started = now; + } + + let idle_elapsed = now.saturating_duration_since(idle_started); + let total_elapsed = now.saturating_duration_since(started); + if idle_elapsed >= idle_window || total_elapsed >= absolute_cap { + for relay in relays { + relay.cancel_read(); + } + on_give_up(OUTPUT_RELAY_TRUNCATION_WARNING); + return OutputRelayDrain::GaveUp; + } + + let remaining_idle = idle_window.saturating_sub(idle_elapsed); + let remaining_cap = absolute_cap.saturating_sub(total_elapsed); + let sleep_for = DRAIN_POLL_INTERVAL.min(remaining_idle).min(remaining_cap); + if sleep_for.is_zero() { + std::thread::yield_now(); + } else { + std::thread::sleep(sleep_for); + } + } +} + +fn relay_is_finished(relay: &PipeRelay) -> bool { + // SAFETY: an owned thread handle held by `PipeRelay`. + (unsafe { WaitForSingleObject(relay.thread().get(), 0) }) == WAIT_OBJECT_0 +} + +fn total_progress(relays: &[&PipeRelay]) -> u64 { + relays + .iter() + .map(|relay| relay.progress_bytes()) + .sum::() +} + // ── Stop-event-aware pipe relay ──────────────────────────────────────────── // // Used for the IsolationSession stdin relay in TTY (ConPTY) mode, where the @@ -261,9 +378,8 @@ pub(super) unsafe fn create_relay_thread_with_stop( mod tests { use super::*; use crate::mxc_common::process_util::{create_std_pipes, read_from_pipe, SendOwnedHandle}; - use crate::mxc_common::sandbox_process::StreamCloser; use crate::mxc_common::string_util; - use windows::Win32::Foundation::WAIT_TIMEOUT; + use windows::Win32::Foundation::{ERROR_BROKEN_PIPE, ERROR_NO_DATA, WAIT_TIMEOUT}; use windows::Win32::System::Pipes::CreatePipe; use windows::Win32::System::Threading::{ CreateEventW, CreateProcessW, SetEvent, WaitForSingleObject, CREATE_NO_WINDOW, @@ -338,8 +454,7 @@ mod tests { let (source_read, source_write) = create_test_pipes(); let (mut dest_read, dest_write) = create_test_pipes(); - let (relay_thread, _canceller) = - unsafe { create_relay_thread(source_read.get(), dest_write.get()).unwrap() }; + let relay = unsafe { create_relay_thread(source_read.get(), dest_write.get()).unwrap() }; // Read from dest concurrently — the relay calls FlushFileBuffers after // each write, which blocks until the reader drains the pipe buffer. @@ -366,7 +481,7 @@ mod tests { // Wait for relay thread to finish unsafe { - WaitForSingleObject(relay_thread.get(), 5000); + WaitForSingleObject(relay.thread().get(), 5000); } // Close the dest write end so the reader sees EOF @@ -387,21 +502,20 @@ mod tests { let (source_read, _source_write) = create_test_pipes(); let (_dest_read, dest_write) = create_test_pipes(); - let (relay_thread, canceller) = - unsafe { create_relay_thread(source_read.get(), dest_write.get()).unwrap() }; + let relay = unsafe { create_relay_thread(source_read.get(), dest_write.get()).unwrap() }; // Parked in `ReadFile`: nothing was written and `_source_write` keeps // the pipe from reporting EOF. assert_eq!( - unsafe { WaitForSingleObject(relay_thread.get(), 250) }, + unsafe { WaitForSingleObject(relay.thread().get(), 250) }, WAIT_TIMEOUT, "the relay should still be blocked before the cancel" ); - canceller.close(); + relay.cancel_read(); assert_eq!( - unsafe { WaitForSingleObject(relay_thread.get(), 5000) }, + unsafe { WaitForSingleObject(relay.thread().get(), 5000) }, WAIT_OBJECT_0, "the cancelled relay should exit without EOF" ); @@ -415,8 +529,7 @@ mod tests { let (source_read, source_write) = create_test_pipes(); let (mut dest_read, dest_write) = create_test_pipes(); - let (relay_thread, _canceller) = - unsafe { create_relay_thread(source_read.get(), dest_write.get()).unwrap() }; + let relay = unsafe { create_relay_thread(source_read.get(), dest_write.get()).unwrap() }; drop(source_read); @@ -431,7 +544,7 @@ mod tests { drop(source_write); assert_eq!( - unsafe { WaitForSingleObject(relay_thread.get(), 5000) }, + unsafe { WaitForSingleObject(relay.thread().get(), 5000) }, WAIT_OBJECT_0, "the relay should reach EOF on its own duplicate" ); @@ -445,8 +558,7 @@ mod tests { let (source_read, mut source_write) = create_test_pipes(); let (mut dest_read, dest_write) = create_test_pipes(); - let (relay_thread, _canceller) = - unsafe { create_relay_thread(source_read.get(), dest_write.get()).unwrap() }; + let relay = unsafe { create_relay_thread(source_read.get(), dest_write.get()).unwrap() }; // Write data larger than the pipe buffer (4096 bytes default). // Use ASCII to avoid from_utf8_lossy expansion of invalid bytes. @@ -478,7 +590,7 @@ mod tests { writer.join().unwrap(); // Wait for relay thread to finish (source EOF propagated) unsafe { - WaitForSingleObject(relay_thread.get(), 5000); + WaitForSingleObject(relay.thread().get(), 5000); } // Close the dest write end so the reader sees EOF drop(dest_write); @@ -497,7 +609,7 @@ mod tests { // dest: non-inheritable (only the test reads here) let (mut dest_read, dest_write) = create_test_pipes(); - let (relay_thread, _canceller) = + let relay = unsafe { create_relay_thread(child_stdout_read.get(), dest_write.get()).unwrap() }; // Concurrent reader to avoid FlushFileBuffers deadlock @@ -517,7 +629,7 @@ mod tests { // Child exits → last write-end closes → relay sees EOF → relay exits unsafe { WaitForSingleObject(child.get(), 10000); - WaitForSingleObject(relay_thread.get(), 5000); + WaitForSingleObject(relay.thread().get(), 5000); } // Close relay's dest write so reader sees EOF @@ -544,7 +656,7 @@ mod tests { let (mut child_stdout_read, child_stdout_write) = create_std_pipes(true).unwrap(); // Relay: test input → child stdin - let (relay_thread, _canceller) = + let relay = unsafe { create_relay_thread(source_read.get(), child_stdin_write.get()).unwrap() }; // findstr /R "." echoes all non-empty lines from stdin to stdout @@ -578,7 +690,7 @@ mod tests { drop(source_write); unsafe { - WaitForSingleObject(relay_thread.get(), 5000); + WaitForSingleObject(relay.thread().get(), 5000); } // Close child_stdin_write → child sees EOF → child exits @@ -596,6 +708,172 @@ mod tests { ); } + #[test] + fn output_drain_extends_while_progress_continues() { + let (source_read, mut source_write) = create_test_pipes(); + let (mut dest_read, dest_write) = create_test_pipes(); + let relay = unsafe { create_relay_thread(source_read.get(), dest_write.get()).unwrap() }; + + let dest_send = SendOwnedHandle::take(&mut dest_read); + let reader = std::thread::spawn(move || read_from_pipe(dest_send.get())); + + let source_send = SendOwnedHandle::take(&mut source_write); + let writer = std::thread::spawn(move || { + for _ in 0..8 { + let mut written = 0u32; + unsafe { + WriteFile( + source_send.get(), + Some(b"chunk\n"), + Some(&mut written), + None, + ) + .unwrap(); + } + std::thread::sleep(std::time::Duration::from_millis(20)); + } + drop(source_send); + }); + + let started = std::time::Instant::now(); + let mut warnings = Vec::new(); + let result = drain_output_relays_within( + &[&relay], + std::time::Duration::from_millis(100), + std::time::Duration::from_secs(2), + |warning| warnings.push(warning.to_string()), + ); + let elapsed = started.elapsed(); + + writer.join().unwrap(); + drop(dest_write); + let output = reader.join().unwrap(); + + assert_eq!(result, OutputRelayDrain::Drained); + assert!(warnings.is_empty(), "a progressing relay must not warn"); + assert!( + elapsed >= std::time::Duration::from_millis(120), + "the drain returned before the slow producer finished: {elapsed:?}" + ); + assert_eq!(output, "chunk\n".repeat(8)); + } + + #[test] + fn output_drain_cancels_and_releases_handles_once_idle() { + let (source_read, source_write) = create_test_pipes(); + let (dest_read, dest_write) = create_test_pipes(); + let relay = unsafe { create_relay_thread(source_read.get(), dest_write.get()).unwrap() }; + drop(source_read); + drop(dest_write); + + let started = std::time::Instant::now(); + let mut logger = + crate::mxc_common::logger::Logger::new(crate::mxc_common::logger::Mode::Buffer); + let result = drain_output_relays_within( + &[&relay], + std::time::Duration::from_millis(30), + std::time::Duration::from_millis(500), + |warning| logger.warning_line(warning), + ); + let elapsed = started.elapsed(); + + assert_eq!(result, OutputRelayDrain::GaveUp); + assert!( + elapsed < std::time::Duration::from_secs(2), + "the drain must be bounded, took {elapsed:?}" + ); + assert_eq!(logger.take_warnings(), [OUTPUT_RELAY_TRUNCATION_WARNING]); + + assert_eq!( + unsafe { WaitForSingleObject(relay.thread().get(), 5000) }, + WAIT_OBJECT_0, + "the relay must exit while the source writer is still open" + ); + let mut written = 0; + let error = unsafe { + WriteFile( + source_write.get(), + Some(b"still open"), + Some(&mut written), + None, + ) + } + .unwrap_err(); + assert!( + error.code() == ERROR_BROKEN_PIPE.to_hresult() + || error.code() == ERROR_NO_DATA.to_hresult(), + "expected the relay's source duplicate to be closed: {error}" + ); + assert_eq!( + read_from_pipe(dest_read.get()), + "", + "destination must reach EOF" + ); + } + + #[test] + fn output_drain_honors_absolute_cap_while_progress_continues() { + let (source_read, mut source_write) = create_test_pipes(); + let (mut dest_read, dest_write) = create_test_pipes(); + let relay = unsafe { create_relay_thread(source_read.get(), dest_write.get()).unwrap() }; + drop(source_read); + drop(dest_write); + + let dest_send = SendOwnedHandle::take(&mut dest_read); + let reader = std::thread::spawn(move || read_from_pipe(dest_send.get())); + let source_send = SendOwnedHandle::take(&mut source_write); + let writer = std::thread::spawn(move || { + for _ in 0..200 { + let mut written = 0; + if let Err(error) = unsafe { + WriteFile( + source_send.get(), + Some(b"chunk\n"), + Some(&mut written), + None, + ) + } { + assert!( + error.code() == ERROR_BROKEN_PIPE.to_hresult() + || error.code() == ERROR_NO_DATA.to_hresult(), + "expected cancellation to close the reader: {error}" + ); + return; + } + std::thread::sleep(Duration::from_millis(10)); + } + }); + + let cap = Duration::from_millis(600); + let started = Instant::now(); + let mut warnings = Vec::new(); + let result = + drain_output_relays_within(&[&relay], Duration::from_millis(300), cap, |warning| { + warnings.push(warning.to_string()) + }); + let elapsed = started.elapsed(); + assert_eq!( + unsafe { WaitForSingleObject(relay.thread().get(), 5000) }, + WAIT_OBJECT_0 + ); + writer.join().unwrap(); + let output = reader.join().unwrap(); + + assert_eq!(result, OutputRelayDrain::GaveUp); + assert!( + elapsed >= cap, + "the idle timer fired during progress: {elapsed:?}" + ); + assert!( + elapsed < Duration::from_millis(1600), + "cap was exceeded: {elapsed:?}" + ); + assert_eq!(warnings, [OUTPUT_RELAY_TRUNCATION_WARNING]); + assert!(output.len() >= 12 && output.len() < 200 * 6); + assert_eq!(output, "chunk\n".repeat(output.len() / 6)); + assert_eq!(relay.progress_bytes(), output.len() as u64); + } + // ── Tests for `create_relay_thread_with_stop` ────────────────────────── // // These exercise the stop-event-aware relay variant. Cancellation (test #1) diff --git a/src/mxc-sdk/src/backends/isolation_session/common/state_aware.rs b/src/mxc-sdk/src/backends/isolation_session/common/state_aware.rs index b31c8a9b8..b27e44b65 100644 --- a/src/mxc-sdk/src/backends/isolation_session/common/state_aware.rs +++ b/src/mxc-sdk/src/backends/isolation_session/common/state_aware.rs @@ -6,7 +6,7 @@ //! `IsolationSessionManager` because the OS service may idle-restart //! between caller invocations. -use std::io::IsTerminal; +use std::io::{IsTerminal, Write}; use serde::Serialize; @@ -82,6 +82,34 @@ fn wants_interactive_console(stdio: ExecStdio, host_is_terminal: impl FnOnce() - } } +/// Relayed execution owns host stdio, unlike the piped SDK path. Its local +/// logger's warnings must reach stderr before that logger is discarded. +fn completed_relayed_exec( + exit_code: i32, + logger: &Logger, + mut stderr: impl Write, +) -> Result { + for warning in logger.warnings() { + writeln!(stderr, "{warning}").map_err(|error| { + MxcError::backend_error(format!( + "failed to report isolation session output warning: {error}; {warning}" + )) + })?; + } + + // Output was relayed inside `create_process`; null handles prevent the + // dispatcher from starting another relay. + let null = HANDLE(std::ptr::null_mut()); + Ok(ExecHandle { + stdout: null, + stderr: null, + stdin: null, + stdin_closer: None, + waiter: Box::new(move || Ok(ExecOutcome::Exited(exit_code))), + terminator: Box::new(|| Ok(())), + }) +} + impl StatefulSandboxBackend for IsolationSessionRunner { const ID_PREFIX: &'static str = sandbox_id::ID_PREFIX; const BACKEND_KEY: &'static str = "isolation_session"; @@ -323,20 +351,7 @@ impl StatefulSandboxBackend for IsolationSessionRunner { // The output relay completed inside `create_process`. The // dispatcher sees zero pipe handles, skips its own relay setup, // and gets the exit code from the waiter closure. - let null = HANDLE(std::ptr::null_mut()); - Ok(ExecHandle { - stdout: null, - stderr: null, - stdin: null, - stdin_closer: None, - // `Exited`, never `TimedOut`: a backend serving - // `ExecStdio::Relayed` reports an exit code, and the - // relay rejects anything else. - waiter: Box::new(move || Ok(ExecOutcome::Exited(exit_code))), - // Nothing to terminate: `create_process` returned only once - // the process was gone. - terminator: Box::new(|| Ok(())), - }) + completed_relayed_exec(exit_code, &logger, std::io::stderr().lock()) } ExecStdio::Piped => { let options = build_process_options( @@ -369,6 +384,7 @@ impl StatefulSandboxBackend for IsolationSessionRunner { #[cfg(test)] mod tests { + use super::super::pipe_relay::OUTPUT_RELAY_TRUNCATION_WARNING; use super::*; use crate::mxc_common::models::{ ContainerPolicy, NetworkAction, NetworkEgressPolicy, NetworkIngressPolicy, ProxyAddress, @@ -376,6 +392,43 @@ mod tests { }; use crate::mxc_common::mxc_error::MxcErrorCode; + #[test] + fn relayed_exec_reports_truncation_without_a_diagnostic_sink() { + let mut logger = Logger::new(crate::mxc_common::logger::Mode::Buffer); + logger.warning_line(OUTPUT_RELAY_TRUNCATION_WARNING); + let mut stderr = Vec::new(); + let handle = completed_relayed_exec(42, &logger, &mut stderr).unwrap(); + assert_eq!( + String::from_utf8(stderr).unwrap(), + format!("{OUTPUT_RELAY_TRUNCATION_WARNING}\n") + ); + assert!(handle.stdout.is_invalid()); + assert!(handle.stderr.is_invalid()); + assert!(handle.stdin.is_invalid()); + assert!(matches!( + (handle.waiter)().unwrap(), + ExecOutcome::Exited(42) + )); + } + + #[test] + fn relayed_exec_without_warnings_preserves_stderr() { + let logger = Logger::new(crate::mxc_common::logger::Mode::Buffer); + let mut stderr = Vec::new(); + let handle = completed_relayed_exec(0, &logger, &mut stderr).unwrap(); + assert!(stderr.is_empty()); + assert!(matches!((handle.waiter)().unwrap(), ExecOutcome::Exited(0))); + } + + #[test] + fn relayed_exec_reports_warning_delivery_failure() { + let mut logger = Logger::new(crate::mxc_common::logger::Mode::Buffer); + logger.warning_line(OUTPUT_RELAY_TRUNCATION_WARNING); + let error = completed_relayed_exec(0, &logger, &mut [][..]).unwrap_err(); + assert_eq!(error.code, MxcErrorCode::BackendError); + assert!(error.message.contains(OUTPUT_RELAY_TRUNCATION_WARNING)); + } + // ====== Wire-format constants ====== // Checked binding verifies BACKEND_KEY before delivering configuration. diff --git a/src/mxc-sdk/src/core/mxc_common/process_util.rs b/src/mxc-sdk/src/core/mxc_common/process_util.rs index 749f38882..c35ffdb06 100644 --- a/src/mxc-sdk/src/core/mxc_common/process_util.rs +++ b/src/mxc-sdk/src/core/mxc_common/process_util.rs @@ -144,6 +144,10 @@ unsafe impl Sync for CancelablePipe {} /// pipe (all write ends closed) still reads as EOF as usual. `Send` so it can /// be handed to a reader thread. Single-reader: at most one thread may `read` it /// at a time (any number of cancellers may fire concurrently). +/// +/// Cancelling can discard data that a read has received but has not yet returned +/// to the caller. When collecting remaining output, cancel only after the +/// timeout, and warn that some output may be missing. pub struct InterruptiblePipeReader(Arc); impl InterruptiblePipeReader { From 9a0c5e0e2808f7cc7c3edb7e3426c566385c84dd Mon Sep 17 00:00:00 2001 From: Dom Giandinoto Date: Tue, 20 Oct 2026 10:50:42 -0700 Subject: [PATCH 2/4] forwarding potential output-truncation warning to stderr --- .../isolation_session/common/manager.rs | 50 ++- .../isolation_session/common/one_shot.rs | 2 +- .../isolation_session/common/state_aware.rs | 308 ++++++++++++++---- 3 files changed, 292 insertions(+), 68 deletions(-) diff --git a/src/mxc-sdk/src/backends/isolation_session/common/manager.rs b/src/mxc-sdk/src/backends/isolation_session/common/manager.rs index bd2839595..879feec7f 100644 --- a/src/mxc-sdk/src/backends/isolation_session/common/manager.rs +++ b/src/mxc-sdk/src/backends/isolation_session/common/manager.rs @@ -43,6 +43,23 @@ use super::process_options::{build_iso_process_options, ProcessOptions}; // The OS account name is lifecycle addressing data, not diagnostic identity. const AGENT_IDENTITY_MARKER: &str = "isolation-session-agent"; +pub(super) struct RelayedProcessResult { + pub exit_code: i32, + pub output_may_be_truncated: bool, +} + +impl RelayedProcessResult { + fn new(outcome: ExecOutcome, output_drain: OutputRelayDrain) -> Self { + Self { + exit_code: match outcome { + ExecOutcome::Exited(exit_code) => exit_code, + ExecOutcome::TimedOut => WAIT_FOR_EXIT_TIMEOUT, + }, + output_may_be_truncated: output_drain == OutputRelayDrain::GaveUp, + } + } +} + /// Keeps the process's MTA alive for as long as the lifecycle's WinRT objects, /// which outlive the call that created them. /// `CoIncrementMTAUsage` holds a reference releasable from any thread; @@ -399,7 +416,8 @@ impl IsolationSessionManager { /// Step 3: Create a process inside the started isolation session and run it /// to completion. Output is streamed live to the calling process's stdio via - /// internal relay threads; only the exit code is returned. + /// internal relay threads; returns the exit code and whether output draining + /// stopped before EOF. /// /// This is the **relay** path, reached by `wxc-exec` and by an in-process /// caller that asked to attach. A caller that wants the streams handed back @@ -408,7 +426,7 @@ impl IsolationSessionManager { &self, options: &ProcessOptions, logger: Option<&mut Logger>, - ) -> Result { + ) -> Result { owned_thread::call(&self.impersonation, || { // Everything fallible that does not need the workload runs first, so no // failure can strand a running one. @@ -582,12 +600,8 @@ impl IsolationSessionManager { let outcome = started.wait(options.timeout_ms)?; scope.stop_stdin(); - scope.finish(logger); - - Ok(match outcome { - ExecOutcome::Exited(exit_code) => exit_code, - ExecOutcome::TimedOut => WAIT_FOR_EXIT_TIMEOUT, - }) + let output_drain = scope.finish(logger); + Ok(RelayedProcessResult::new(outcome, output_drain)) }) } @@ -1091,7 +1105,7 @@ impl<'a> RelayScope<'a> { } /// Teardown for a workload that has already exited. - fn finish(&mut self, mut logger: Option<&mut Logger>) { + fn finish(&mut self, mut logger: Option<&mut Logger>) -> OutputRelayDrain { self.signal_stop(); let output_relays = self.output_relays(); let output_drain = drain_output_relays_within( @@ -1111,6 +1125,7 @@ impl<'a> RelayScope<'a> { Self::join(self.stdin.as_ref(), Self::JOIN_MS); self.close_handles(); self.finished = true; + output_drain } } @@ -1364,6 +1379,23 @@ fn wait_with_graceful_shutdown( mod tests { use super::*; + #[test] + fn relayed_result_reports_drain_status_without_changing_exit_code() { + for (drain, truncated) in [ + (OutputRelayDrain::Drained, false), + (OutputRelayDrain::GaveUp, true), + ] { + for exit_code in [0, 42, -1] { + let result = RelayedProcessResult::new(ExecOutcome::Exited(exit_code), drain); + assert_eq!(result.exit_code, exit_code); + assert_eq!(result.output_may_be_truncated, truncated); + } + let result = RelayedProcessResult::new(ExecOutcome::TimedOut, drain); + assert_eq!(result.exit_code, WAIT_FOR_EXIT_TIMEOUT); + assert_eq!(result.output_may_be_truncated, truncated); + } + } + #[test] fn bounded_termination_requires_confirmation_after_a_timeout_sentinel() { assert!(termination_is_confirmed(0, None)); diff --git a/src/mxc-sdk/src/backends/isolation_session/common/one_shot.rs b/src/mxc-sdk/src/backends/isolation_session/common/one_shot.rs index 68161e291..5bc57f086 100644 --- a/src/mxc-sdk/src/backends/isolation_session/common/one_shot.rs +++ b/src/mxc-sdk/src/backends/isolation_session/common/one_shot.rs @@ -122,7 +122,7 @@ impl ScriptRunner for IsolationSessionRunner { } let exit_code = match manager.create_process(&options, Some(logger)) { - Ok(code) => code, + Ok(result) => result.exit_code, Err(e) => { let stopped = manager.stop_session().is_ok(); let deprovisioned = manager.deprovision_agent_user().is_ok(); diff --git a/src/mxc-sdk/src/backends/isolation_session/common/state_aware.rs b/src/mxc-sdk/src/backends/isolation_session/common/state_aware.rs index b27e44b65..715d392f5 100644 --- a/src/mxc-sdk/src/backends/isolation_session/common/state_aware.rs +++ b/src/mxc-sdk/src/backends/isolation_session/common/state_aware.rs @@ -6,13 +6,17 @@ //! `IsolationSessionManager` because the OS service may idle-restart //! between caller invocations. -use std::io::{IsTerminal, Write}; +use std::io::{self, IsTerminal}; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::Arc; +use std::time::Instant; use serde::Serialize; use crate::mxc_common::logger::Logger; use crate::mxc_common::models::{ExecutionRequest, IsolationSessionProvisionConfig}; use crate::mxc_common::mxc_error::MxcError; +use crate::mxc_common::process_util::OwnedHandle; use crate::mxc_common::sandbox_process::{PtySize, SandboxProcess}; use crate::mxc_common::state_aware_backend::{ DeprovisionResult, ExecHandle, ExecOutcome, ExecStdio, ProvisionResult, StartResult, @@ -22,10 +26,20 @@ use crate::mxc_common::validator::{ validate_state_aware_network_policy_support, NetworkPolicySupport, }; -use windows::Win32::Foundation::HANDLE; +use windows::Win32::Foundation::{ + ERROR_NOT_FOUND, ERROR_OPERATION_ABORTED, ERROR_WRITE_FAULT, HANDLE, WAIT_OBJECT_0, + WAIT_TIMEOUT, +}; +use windows::Win32::Storage::FileSystem::WriteFile; +use windows::Win32::System::Console::{GetStdHandle, STD_ERROR_HANDLE}; +use windows::Win32::System::Threading::{ + CreateThread, GetExitCodeThread, WaitForSingleObject, THREAD_CREATION_FLAGS, +}; +use windows::Win32::System::IO::CancelSynchronousIo; use super::error::map_lifecycle_error; use super::manager::{log_sandbox_torn_down, IsolationSessionManager, TeardownOutcome}; +use super::pipe_relay::{duplicate_handle, OUTPUT_RELAY_TRUNCATION_WARNING}; use super::policy::{ reject_unhonorable_environment, validate_post_provision_policy, validate_provision_policy, }; @@ -82,32 +96,55 @@ fn wants_interactive_console(stdio: ExecStdio, host_is_terminal: impl FnOnce() - } } -/// Relayed execution owns host stdio, unlike the piped SDK path. Its local -/// logger's warnings must reach stderr before that logger is discarded. -fn completed_relayed_exec( - exit_code: i32, - logger: &Logger, - mut stderr: impl Write, -) -> Result { - for warning in logger.warnings() { - writeln!(stderr, "{warning}").map_err(|error| { - MxcError::backend_error(format!( - "failed to report isolation session output warning: {error}; {warning}" - )) - })?; - } - - // Output was relayed inside `create_process`; null handles prevent the - // dispatcher from starting another relay. - let null = HANDLE(std::ptr::null_mut()); - Ok(ExecHandle { - stdout: null, - stderr: null, - stdin: null, - stdin_closer: None, - waiter: Box::new(move || Ok(ExecOutcome::Exited(exit_code))), - terminator: Box::new(|| Ok(())), - }) +struct TruncationWarningParams { + stderr: OwnedHandle, + stopped: Arc, +} + +// The thread owns the parameters and reports the write's HRESULT as its exit code. +unsafe extern "system" fn truncation_warning_thread(param: *mut core::ffi::c_void) -> u32 { + // SAFETY: start_truncation_warning transfers this box to the thread exactly once. + let params = unsafe { Box::from_raw(param as *mut TruncationWarningParams) }; + if params.stopped.load(Ordering::Acquire) { + return ERROR_OPERATION_ABORTED.to_hresult().0 as u32; + } + let message = format!("{OUTPUT_RELAY_TRUNCATION_WARNING}\n"); + let mut written = 0; + // SAFETY: the owned handle and message remain valid until WriteFile returns. + match unsafe { + WriteFile( + params.stderr.get(), + Some(message.as_bytes()), + Some(&mut written), + None, + ) + } { + Ok(()) if written as usize == message.len() => 0, + Ok(()) => ERROR_WRITE_FAULT.to_hresult().0 as u32, + Err(error) => error.code().0 as u32, + } +} + +fn start_truncation_warning(params: TruncationWarningParams) -> windows::core::Result { + let raw = Box::into_raw(Box::new(params)); + // SAFETY: the procedure takes ownership of raw only if thread creation succeeds. + match unsafe { + CreateThread( + None, + 0, + Some(truncation_warning_thread), + Some(raw.cast()), + THREAD_CREATION_FLAGS(0), + None, + ) + } { + Ok(thread) => Ok(OwnedHandle::new(thread)), + Err(error) => { + // SAFETY: no thread was created, so the box still belongs to this caller. + drop(unsafe { Box::from_raw(raw) }); + Err(error) + } + } } impl StatefulSandboxBackend for IsolationSessionRunner { @@ -344,14 +381,89 @@ impl StatefulSandboxBackend for IsolationSessionRunner { ); let mut logger = Logger::inherit_thread_diagnostic_sink(); - let exit_code = manager + let result = manager .create_process(&options, Some(&mut logger)) .map_err(map_lifecycle_error)?; + let exit_code = result.exit_code; + + if result.output_may_be_truncated { + let warning_result = (|| { + // GetStdHandle returns a borrowed handle; the worker owns a duplicate. + let stderr = + unsafe { GetStdHandle(STD_ERROR_HANDLE) }.map_err(io::Error::other)?; + let stderr = duplicate_handle(stderr).map_err(io::Error::other)?; + let stopped = Arc::new(AtomicBool::new(false)); + let thread = start_truncation_warning(TruncationWarningParams { + stderr, + stopped: Arc::clone(&stopped), + }) + .map_err(io::Error::other)?; + // SAFETY: thread owns a valid thread handle throughout the wait. + let wait = unsafe { WaitForSingleObject(thread.get(), 1000) }; + if wait == WAIT_OBJECT_0 { + let mut code = 0; + unsafe { GetExitCodeThread(thread.get(), &mut code) } + .map_err(io::Error::other)?; + return windows::core::HRESULT(code as i32) + .ok() + .map_err(io::Error::other); + } + let error = if wait == WAIT_TIMEOUT { + io::Error::new( + io::ErrorKind::TimedOut, + "stderr warning write timed out", + ) + } else { + io::Error::last_os_error() + }; + stopped.store(true, Ordering::Release); + let cancelling = Instant::now(); + loop { + // Retry to catch a worker paused between its stop check and WriteFile. + if let Err(error) = unsafe { CancelSynchronousIo(thread.get()) } { + if error.code() != ERROR_NOT_FOUND.to_hresult() { + return Err(io::Error::other(error)); + } + } + match unsafe { WaitForSingleObject(thread.get(), 10) } { + WAIT_OBJECT_0 => break, + WAIT_TIMEOUT if cancelling.elapsed().as_millis() < 1000 => continue, + WAIT_TIMEOUT => { + // The worker retains its data and handle if cancellation stalls. + return Err(io::Error::new( + io::ErrorKind::TimedOut, + "stderr warning writer did not stop after cancellation", + )); + } + _ => return Err(io::Error::last_os_error()), + } + } + Err(error) + })(); + if let Err(error) = warning_result { + logger.log_line(&format!( + "Could not write isolation session truncation warning to stderr: {error}" + )); + } + } // The output relay completed inside `create_process`. The // dispatcher sees zero pipe handles, skips its own relay setup, // and gets the exit code from the waiter closure. - completed_relayed_exec(exit_code, &logger, std::io::stderr().lock()) + let null = HANDLE(std::ptr::null_mut()); + Ok(ExecHandle { + stdout: null, + stderr: null, + stdin: null, + stdin_closer: None, + // `Exited`, never `TimedOut`: a backend serving + // `ExecStdio::Relayed` reports an exit code, and the + // relay rejects anything else. + waiter: Box::new(move || Ok(ExecOutcome::Exited(exit_code))), + // Nothing to terminate: `create_process` returned only once + // the process was gone. + terminator: Box::new(|| Ok(())), + }) } ExecStdio::Piped => { let options = build_process_options( @@ -384,49 +496,129 @@ impl StatefulSandboxBackend for IsolationSessionRunner { #[cfg(test)] mod tests { - use super::super::pipe_relay::OUTPUT_RELAY_TRUNCATION_WARNING; use super::*; use crate::mxc_common::models::{ ContainerPolicy, NetworkAction, NetworkEgressPolicy, NetworkIngressPolicy, ProxyAddress, ProxyConfig, }; use crate::mxc_common::mxc_error::MxcErrorCode; + use std::fs::File; + use std::os::windows::io::AsRawHandle; + + fn warning_test_pipe() -> (File, OwnedHandle) { + use windows::Win32::System::Pipes::CreatePipe; + + let mut read = HANDLE::default(); + let mut write = HANDLE::default(); + // SAFETY: both out pointers are valid; the handles are owned below. + unsafe { CreatePipe(&mut read, &mut write, None, 4096).unwrap() }; + ( + File::from(OwnedHandle::new(read).try_into_std_owned_handle().unwrap()), + OwnedHandle::new(write), + ) + } - #[test] - fn relayed_exec_reports_truncation_without_a_diagnostic_sink() { - let mut logger = Logger::new(crate::mxc_common::logger::Mode::Buffer); - logger.warning_line(OUTPUT_RELAY_TRUNCATION_WARNING); - let mut stderr = Vec::new(); - let handle = completed_relayed_exec(42, &logger, &mut stderr).unwrap(); + fn warning_thread_result(thread: &OwnedHandle) -> io::Result<()> { assert_eq!( - String::from_utf8(stderr).unwrap(), - format!("{OUTPUT_RELAY_TRUNCATION_WARNING}\n") + unsafe { WaitForSingleObject(thread.get(), 1000) }, + WAIT_OBJECT_0 ); - assert!(handle.stdout.is_invalid()); - assert!(handle.stderr.is_invalid()); - assert!(handle.stdin.is_invalid()); - assert!(matches!( - (handle.waiter)().unwrap(), - ExecOutcome::Exited(42) - )); + let mut code = 0; + unsafe { GetExitCodeThread(thread.get(), &mut code).unwrap() }; + windows::core::HRESULT(code as i32) + .ok() + .map_err(io::Error::other) + } + + #[test] + fn truncation_thread_writes_only_the_fixed_warning() { + use std::io::Read; + + let (mut read, write) = warning_test_pipe(); + let thread = start_truncation_warning(TruncationWarningParams { + stderr: write, + stopped: Arc::new(AtomicBool::new(false)), + }) + .unwrap(); + warning_thread_result(&thread).unwrap(); + + let mut output = String::new(); + read.read_to_string(&mut output).unwrap(); + assert_eq!(output, format!("{OUTPUT_RELAY_TRUNCATION_WARNING}\n")); + } + + #[test] + fn truncation_thread_reports_a_closed_pipe() { + let (read, write) = warning_test_pipe(); + drop(read); + let thread = start_truncation_warning(TruncationWarningParams { + stderr: write, + stopped: Arc::new(AtomicBool::new(false)), + }) + .unwrap(); + assert!(warning_thread_result(&thread).is_err()); } #[test] - fn relayed_exec_without_warnings_preserves_stderr() { - let logger = Logger::new(crate::mxc_common::logger::Mode::Buffer); - let mut stderr = Vec::new(); - let handle = completed_relayed_exec(0, &logger, &mut stderr).unwrap(); - assert!(stderr.is_empty()); - assert!(matches!((handle.waiter)().unwrap(), ExecOutcome::Exited(0))); + fn relayed_warning_to_a_full_pipe_cancels_and_releases_its_handle() { + use std::io::Read; + use windows::Win32::Foundation::ERROR_BROKEN_PIPE; + use windows::Win32::System::Pipes::PeekNamedPipe; + + let (mut read, write) = warning_test_pipe(); + let mut written = 0; + unsafe { WriteFile(write.get(), Some(&[b'x'; 4096]), Some(&mut written), None).unwrap() }; + assert_eq!(written, 4096); + let stopped = Arc::new(AtomicBool::new(false)); + let thread = start_truncation_warning(TruncationWarningParams { + stderr: write, + stopped: Arc::clone(&stopped), + }) + .unwrap(); + let started = Instant::now(); + assert_eq!( + unsafe { WaitForSingleObject(thread.get(), 50) }, + WAIT_TIMEOUT + ); + stopped.store(true, Ordering::Release); + loop { + if let Err(error) = unsafe { CancelSynchronousIo(thread.get()) } { + assert_eq!(error.code(), ERROR_NOT_FOUND.to_hresult()); + } + if unsafe { WaitForSingleObject(thread.get(), 10) } == WAIT_OBJECT_0 { + break; + } + assert!( + started.elapsed().as_secs() < 3, + "warning writer did not stop" + ); + } + assert!(warning_thread_result(&thread).is_err()); + + let mut buffered = [0; 4096]; + read.read_exact(&mut buffered).unwrap(); + assert_eq!(buffered, [b'x'; 4096]); + // SAFETY: the read handle is live. Peek cannot block if a writer leaked. + let error = + unsafe { PeekNamedPipe(HANDLE(read.as_raw_handle()), None, 0, None, None, None) } + .unwrap_err(); + assert_eq!(error.code(), ERROR_BROKEN_PIPE.to_hresult()); } #[test] - fn relayed_exec_reports_warning_delivery_failure() { - let mut logger = Logger::new(crate::mxc_common::logger::Mode::Buffer); - logger.warning_line(OUTPUT_RELAY_TRUNCATION_WARNING); - let error = completed_relayed_exec(0, &logger, &mut [][..]).unwrap_err(); - assert_eq!(error.code, MxcErrorCode::BackendError); - assert!(error.message.contains(OUTPUT_RELAY_TRUNCATION_WARNING)); + fn stopped_warning_thread_does_not_start_a_write() { + use std::io::Read; + + let (mut read, write) = warning_test_pipe(); + let thread = start_truncation_warning(TruncationWarningParams { + stderr: write, + stopped: Arc::new(AtomicBool::new(true)), + }) + .unwrap(); + assert!(warning_thread_result(&thread).is_err()); + let mut output = String::new(); + read.read_to_string(&mut output).unwrap(); + assert!(output.is_empty()); } // ====== Wire-format constants ====== From fd7d3472a230d59f7459c9a568aa032a847153e0 Mon Sep 17 00:00:00 2001 From: Dom Giandinoto Date: Tue, 20 Oct 2026 13:33:27 -0700 Subject: [PATCH 3/4] removing warning delivery, deferring to future PR --- .../isolation-session/state-aware-rust.md | 18 +- .../isolation_session/common/manager.rs | 55 +--- .../isolation_session/common/one_shot.rs | 2 +- .../isolation_session/common/pipe_relay.rs | 20 +- .../isolation_session/common/state_aware.rs | 251 +----------------- .../src/core/mxc_common/process_util.rs | 4 +- 6 files changed, 24 insertions(+), 326 deletions(-) diff --git a/docs/development/architecture/backends/isolation-session/state-aware-rust.md b/docs/development/architecture/backends/isolation-session/state-aware-rust.md index a04d7601a..4ed30ef4c 100644 --- a/docs/development/architecture/backends/isolation-session/state-aware-rust.md +++ b/docs/development/architecture/backends/isolation-session/state-aware-rust.md @@ -428,16 +428,14 @@ copies remaining stdout and stderr until both finish, neither successfully writes output for five seconds, or 60 seconds pass, whichever comes first. These limits do not apply to SDK `Piped` execution. -On timeout, MXC cancels reads, warns that output may be missing, and briefly -waits for the copying threads to exit. Cancellation can discard data already -read, but cannot interrupt blocked writes or flushes. Those threads may keep -pipe handles open and finish writing after cleanup returns. Descendant -processes and stdin handling are unchanged. - -Lifecycle `Relayed` execution sends warnings to host stderr and configured -diagnostic output, preserving the workload exit code if warning delivery -succeeds. Create-and-run execution stores warnings in the caller's logger; -`wxc-exec` prints them to stderr. +On timeout, MXC cancels reads and briefly waits for the copying threads to exit. +Cancellation can discard data already read, but cannot interrupt blocked writes +or flushes. Those threads may keep pipe handles open and finish writing after +cleanup returns. Descendant processes and stdin handling are unchanged. + +Reaching the drain limit does not change the workload's exit code or emit a +truncation warning. The drain fix reduces premature cancellation of progressing +output; it does not guarantee complete output when a limit is reached. ### Deprovision and concurrent sandboxes diff --git a/src/mxc-sdk/src/backends/isolation_session/common/manager.rs b/src/mxc-sdk/src/backends/isolation_session/common/manager.rs index 879feec7f..6ffe2a518 100644 --- a/src/mxc-sdk/src/backends/isolation_session/common/manager.rs +++ b/src/mxc-sdk/src/backends/isolation_session/common/manager.rs @@ -43,23 +43,6 @@ use super::process_options::{build_iso_process_options, ProcessOptions}; // The OS account name is lifecycle addressing data, not diagnostic identity. const AGENT_IDENTITY_MARKER: &str = "isolation-session-agent"; -pub(super) struct RelayedProcessResult { - pub exit_code: i32, - pub output_may_be_truncated: bool, -} - -impl RelayedProcessResult { - fn new(outcome: ExecOutcome, output_drain: OutputRelayDrain) -> Self { - Self { - exit_code: match outcome { - ExecOutcome::Exited(exit_code) => exit_code, - ExecOutcome::TimedOut => WAIT_FOR_EXIT_TIMEOUT, - }, - output_may_be_truncated: output_drain == OutputRelayDrain::GaveUp, - } - } -} - /// Keeps the process's MTA alive for as long as the lifecycle's WinRT objects, /// which outlive the call that created them. /// `CoIncrementMTAUsage` holds a reference releasable from any thread; @@ -416,8 +399,7 @@ impl IsolationSessionManager { /// Step 3: Create a process inside the started isolation session and run it /// to completion. Output is streamed live to the calling process's stdio via - /// internal relay threads; returns the exit code and whether output draining - /// stopped before EOF. + /// internal relay threads; only the exit code is returned. /// /// This is the **relay** path, reached by `wxc-exec` and by an in-process /// caller that asked to attach. A caller that wants the streams handed back @@ -426,7 +408,7 @@ impl IsolationSessionManager { &self, options: &ProcessOptions, logger: Option<&mut Logger>, - ) -> Result { + ) -> Result { owned_thread::call(&self.impersonation, || { // Everything fallible that does not need the workload runs first, so no // failure can strand a running one. @@ -600,8 +582,12 @@ impl IsolationSessionManager { let outcome = started.wait(options.timeout_ms)?; scope.stop_stdin(); - let output_drain = scope.finish(logger); - Ok(RelayedProcessResult::new(outcome, output_drain)) + scope.finish(); + + Ok(match outcome { + ExecOutcome::Exited(exit_code) => exit_code, + ExecOutcome::TimedOut => WAIT_FOR_EXIT_TIMEOUT, + }) }) } @@ -1105,18 +1091,13 @@ impl<'a> RelayScope<'a> { } /// Teardown for a workload that has already exited. - fn finish(&mut self, mut logger: Option<&mut Logger>) -> OutputRelayDrain { + fn finish(&mut self) { self.signal_stop(); let output_relays = self.output_relays(); let output_drain = drain_output_relays_within( &output_relays, Duration::from_millis(Self::IDLE_MS), Duration::from_millis(Self::ABSOLUTE_CAP_MS), - |warning| { - if let Some(logger) = &mut logger { - logger.warning_line(warning); - } - }, ); if output_drain == OutputRelayDrain::GaveUp { Self::join_output(&self.stdout, Self::JOIN_MS); @@ -1125,7 +1106,6 @@ impl<'a> RelayScope<'a> { Self::join(self.stdin.as_ref(), Self::JOIN_MS); self.close_handles(); self.finished = true; - output_drain } } @@ -1379,23 +1359,6 @@ fn wait_with_graceful_shutdown( mod tests { use super::*; - #[test] - fn relayed_result_reports_drain_status_without_changing_exit_code() { - for (drain, truncated) in [ - (OutputRelayDrain::Drained, false), - (OutputRelayDrain::GaveUp, true), - ] { - for exit_code in [0, 42, -1] { - let result = RelayedProcessResult::new(ExecOutcome::Exited(exit_code), drain); - assert_eq!(result.exit_code, exit_code); - assert_eq!(result.output_may_be_truncated, truncated); - } - let result = RelayedProcessResult::new(ExecOutcome::TimedOut, drain); - assert_eq!(result.exit_code, WAIT_FOR_EXIT_TIMEOUT); - assert_eq!(result.output_may_be_truncated, truncated); - } - } - #[test] fn bounded_termination_requires_confirmation_after_a_timeout_sentinel() { assert!(termination_is_confirmed(0, None)); diff --git a/src/mxc-sdk/src/backends/isolation_session/common/one_shot.rs b/src/mxc-sdk/src/backends/isolation_session/common/one_shot.rs index 5bc57f086..68161e291 100644 --- a/src/mxc-sdk/src/backends/isolation_session/common/one_shot.rs +++ b/src/mxc-sdk/src/backends/isolation_session/common/one_shot.rs @@ -122,7 +122,7 @@ impl ScriptRunner for IsolationSessionRunner { } let exit_code = match manager.create_process(&options, Some(logger)) { - Ok(result) => result.exit_code, + Ok(code) => code, Err(e) => { let stopped = manager.stop_session().is_ok(); let deprovisioned = manager.deprovision_agent_user().is_ok(); diff --git a/src/mxc-sdk/src/backends/isolation_session/common/pipe_relay.rs b/src/mxc-sdk/src/backends/isolation_session/common/pipe_relay.rs index f9dbb8b83..012ecc51c 100644 --- a/src/mxc-sdk/src/backends/isolation_session/common/pipe_relay.rs +++ b/src/mxc-sdk/src/backends/isolation_session/common/pipe_relay.rs @@ -33,10 +33,6 @@ use windows::Win32::System::Threading::{ const BUFFER_SIZE: u32 = 4096; const DRAIN_POLL_INTERVAL: Duration = Duration::from_millis(200); -pub(super) const OUTPUT_RELAY_TRUNCATION_WARNING: &str = - "Warning: isolation session output may be truncated because output draining reached its idle \ - or overall time limit before EOF; a descendant may still be holding the output pipe open"; - pub(super) fn duplicate_handle(handle: HANDLE) -> Result { let mut duplicated = HANDLE::default(); unsafe { @@ -192,7 +188,6 @@ pub(super) fn drain_output_relays_within( relays: &[&PipeRelay], idle_window: Duration, absolute_cap: Duration, - on_give_up: impl FnOnce(&str), ) -> OutputRelayDrain { if relays.is_empty() || relays.iter().all(|relay| relay_is_finished(relay)) { return OutputRelayDrain::Drained; @@ -220,7 +215,6 @@ pub(super) fn drain_output_relays_within( for relay in relays { relay.cancel_read(); } - on_give_up(OUTPUT_RELAY_TRUNCATION_WARNING); return OutputRelayDrain::GaveUp; } @@ -736,12 +730,10 @@ mod tests { }); let started = std::time::Instant::now(); - let mut warnings = Vec::new(); let result = drain_output_relays_within( &[&relay], std::time::Duration::from_millis(100), std::time::Duration::from_secs(2), - |warning| warnings.push(warning.to_string()), ); let elapsed = started.elapsed(); @@ -750,7 +742,6 @@ mod tests { let output = reader.join().unwrap(); assert_eq!(result, OutputRelayDrain::Drained); - assert!(warnings.is_empty(), "a progressing relay must not warn"); assert!( elapsed >= std::time::Duration::from_millis(120), "the drain returned before the slow producer finished: {elapsed:?}" @@ -767,13 +758,10 @@ mod tests { drop(dest_write); let started = std::time::Instant::now(); - let mut logger = - crate::mxc_common::logger::Logger::new(crate::mxc_common::logger::Mode::Buffer); let result = drain_output_relays_within( &[&relay], std::time::Duration::from_millis(30), std::time::Duration::from_millis(500), - |warning| logger.warning_line(warning), ); let elapsed = started.elapsed(); @@ -782,7 +770,6 @@ mod tests { elapsed < std::time::Duration::from_secs(2), "the drain must be bounded, took {elapsed:?}" ); - assert_eq!(logger.take_warnings(), [OUTPUT_RELAY_TRUNCATION_WARNING]); assert_eq!( unsafe { WaitForSingleObject(relay.thread().get(), 5000) }, @@ -846,11 +833,7 @@ mod tests { let cap = Duration::from_millis(600); let started = Instant::now(); - let mut warnings = Vec::new(); - let result = - drain_output_relays_within(&[&relay], Duration::from_millis(300), cap, |warning| { - warnings.push(warning.to_string()) - }); + let result = drain_output_relays_within(&[&relay], Duration::from_millis(300), cap); let elapsed = started.elapsed(); assert_eq!( unsafe { WaitForSingleObject(relay.thread().get(), 5000) }, @@ -868,7 +851,6 @@ mod tests { elapsed < Duration::from_millis(1600), "cap was exceeded: {elapsed:?}" ); - assert_eq!(warnings, [OUTPUT_RELAY_TRUNCATION_WARNING]); assert!(output.len() >= 12 && output.len() < 200 * 6); assert_eq!(output, "chunk\n".repeat(output.len() / 6)); assert_eq!(relay.progress_bytes(), output.len() as u64); diff --git a/src/mxc-sdk/src/backends/isolation_session/common/state_aware.rs b/src/mxc-sdk/src/backends/isolation_session/common/state_aware.rs index 715d392f5..b31c8a9b8 100644 --- a/src/mxc-sdk/src/backends/isolation_session/common/state_aware.rs +++ b/src/mxc-sdk/src/backends/isolation_session/common/state_aware.rs @@ -6,17 +6,13 @@ //! `IsolationSessionManager` because the OS service may idle-restart //! between caller invocations. -use std::io::{self, IsTerminal}; -use std::sync::atomic::{AtomicBool, Ordering}; -use std::sync::Arc; -use std::time::Instant; +use std::io::IsTerminal; use serde::Serialize; use crate::mxc_common::logger::Logger; use crate::mxc_common::models::{ExecutionRequest, IsolationSessionProvisionConfig}; use crate::mxc_common::mxc_error::MxcError; -use crate::mxc_common::process_util::OwnedHandle; use crate::mxc_common::sandbox_process::{PtySize, SandboxProcess}; use crate::mxc_common::state_aware_backend::{ DeprovisionResult, ExecHandle, ExecOutcome, ExecStdio, ProvisionResult, StartResult, @@ -26,20 +22,10 @@ use crate::mxc_common::validator::{ validate_state_aware_network_policy_support, NetworkPolicySupport, }; -use windows::Win32::Foundation::{ - ERROR_NOT_FOUND, ERROR_OPERATION_ABORTED, ERROR_WRITE_FAULT, HANDLE, WAIT_OBJECT_0, - WAIT_TIMEOUT, -}; -use windows::Win32::Storage::FileSystem::WriteFile; -use windows::Win32::System::Console::{GetStdHandle, STD_ERROR_HANDLE}; -use windows::Win32::System::Threading::{ - CreateThread, GetExitCodeThread, WaitForSingleObject, THREAD_CREATION_FLAGS, -}; -use windows::Win32::System::IO::CancelSynchronousIo; +use windows::Win32::Foundation::HANDLE; use super::error::map_lifecycle_error; use super::manager::{log_sandbox_torn_down, IsolationSessionManager, TeardownOutcome}; -use super::pipe_relay::{duplicate_handle, OUTPUT_RELAY_TRUNCATION_WARNING}; use super::policy::{ reject_unhonorable_environment, validate_post_provision_policy, validate_provision_policy, }; @@ -96,57 +82,6 @@ fn wants_interactive_console(stdio: ExecStdio, host_is_terminal: impl FnOnce() - } } -struct TruncationWarningParams { - stderr: OwnedHandle, - stopped: Arc, -} - -// The thread owns the parameters and reports the write's HRESULT as its exit code. -unsafe extern "system" fn truncation_warning_thread(param: *mut core::ffi::c_void) -> u32 { - // SAFETY: start_truncation_warning transfers this box to the thread exactly once. - let params = unsafe { Box::from_raw(param as *mut TruncationWarningParams) }; - if params.stopped.load(Ordering::Acquire) { - return ERROR_OPERATION_ABORTED.to_hresult().0 as u32; - } - let message = format!("{OUTPUT_RELAY_TRUNCATION_WARNING}\n"); - let mut written = 0; - // SAFETY: the owned handle and message remain valid until WriteFile returns. - match unsafe { - WriteFile( - params.stderr.get(), - Some(message.as_bytes()), - Some(&mut written), - None, - ) - } { - Ok(()) if written as usize == message.len() => 0, - Ok(()) => ERROR_WRITE_FAULT.to_hresult().0 as u32, - Err(error) => error.code().0 as u32, - } -} - -fn start_truncation_warning(params: TruncationWarningParams) -> windows::core::Result { - let raw = Box::into_raw(Box::new(params)); - // SAFETY: the procedure takes ownership of raw only if thread creation succeeds. - match unsafe { - CreateThread( - None, - 0, - Some(truncation_warning_thread), - Some(raw.cast()), - THREAD_CREATION_FLAGS(0), - None, - ) - } { - Ok(thread) => Ok(OwnedHandle::new(thread)), - Err(error) => { - // SAFETY: no thread was created, so the box still belongs to this caller. - drop(unsafe { Box::from_raw(raw) }); - Err(error) - } - } -} - impl StatefulSandboxBackend for IsolationSessionRunner { const ID_PREFIX: &'static str = sandbox_id::ID_PREFIX; const BACKEND_KEY: &'static str = "isolation_session"; @@ -381,71 +316,9 @@ impl StatefulSandboxBackend for IsolationSessionRunner { ); let mut logger = Logger::inherit_thread_diagnostic_sink(); - let result = manager + let exit_code = manager .create_process(&options, Some(&mut logger)) .map_err(map_lifecycle_error)?; - let exit_code = result.exit_code; - - if result.output_may_be_truncated { - let warning_result = (|| { - // GetStdHandle returns a borrowed handle; the worker owns a duplicate. - let stderr = - unsafe { GetStdHandle(STD_ERROR_HANDLE) }.map_err(io::Error::other)?; - let stderr = duplicate_handle(stderr).map_err(io::Error::other)?; - let stopped = Arc::new(AtomicBool::new(false)); - let thread = start_truncation_warning(TruncationWarningParams { - stderr, - stopped: Arc::clone(&stopped), - }) - .map_err(io::Error::other)?; - // SAFETY: thread owns a valid thread handle throughout the wait. - let wait = unsafe { WaitForSingleObject(thread.get(), 1000) }; - if wait == WAIT_OBJECT_0 { - let mut code = 0; - unsafe { GetExitCodeThread(thread.get(), &mut code) } - .map_err(io::Error::other)?; - return windows::core::HRESULT(code as i32) - .ok() - .map_err(io::Error::other); - } - let error = if wait == WAIT_TIMEOUT { - io::Error::new( - io::ErrorKind::TimedOut, - "stderr warning write timed out", - ) - } else { - io::Error::last_os_error() - }; - stopped.store(true, Ordering::Release); - let cancelling = Instant::now(); - loop { - // Retry to catch a worker paused between its stop check and WriteFile. - if let Err(error) = unsafe { CancelSynchronousIo(thread.get()) } { - if error.code() != ERROR_NOT_FOUND.to_hresult() { - return Err(io::Error::other(error)); - } - } - match unsafe { WaitForSingleObject(thread.get(), 10) } { - WAIT_OBJECT_0 => break, - WAIT_TIMEOUT if cancelling.elapsed().as_millis() < 1000 => continue, - WAIT_TIMEOUT => { - // The worker retains its data and handle if cancellation stalls. - return Err(io::Error::new( - io::ErrorKind::TimedOut, - "stderr warning writer did not stop after cancellation", - )); - } - _ => return Err(io::Error::last_os_error()), - } - } - Err(error) - })(); - if let Err(error) = warning_result { - logger.log_line(&format!( - "Could not write isolation session truncation warning to stderr: {error}" - )); - } - } // The output relay completed inside `create_process`. The // dispatcher sees zero pipe handles, skips its own relay setup, @@ -502,124 +375,6 @@ mod tests { ProxyConfig, }; use crate::mxc_common::mxc_error::MxcErrorCode; - use std::fs::File; - use std::os::windows::io::AsRawHandle; - - fn warning_test_pipe() -> (File, OwnedHandle) { - use windows::Win32::System::Pipes::CreatePipe; - - let mut read = HANDLE::default(); - let mut write = HANDLE::default(); - // SAFETY: both out pointers are valid; the handles are owned below. - unsafe { CreatePipe(&mut read, &mut write, None, 4096).unwrap() }; - ( - File::from(OwnedHandle::new(read).try_into_std_owned_handle().unwrap()), - OwnedHandle::new(write), - ) - } - - fn warning_thread_result(thread: &OwnedHandle) -> io::Result<()> { - assert_eq!( - unsafe { WaitForSingleObject(thread.get(), 1000) }, - WAIT_OBJECT_0 - ); - let mut code = 0; - unsafe { GetExitCodeThread(thread.get(), &mut code).unwrap() }; - windows::core::HRESULT(code as i32) - .ok() - .map_err(io::Error::other) - } - - #[test] - fn truncation_thread_writes_only_the_fixed_warning() { - use std::io::Read; - - let (mut read, write) = warning_test_pipe(); - let thread = start_truncation_warning(TruncationWarningParams { - stderr: write, - stopped: Arc::new(AtomicBool::new(false)), - }) - .unwrap(); - warning_thread_result(&thread).unwrap(); - - let mut output = String::new(); - read.read_to_string(&mut output).unwrap(); - assert_eq!(output, format!("{OUTPUT_RELAY_TRUNCATION_WARNING}\n")); - } - - #[test] - fn truncation_thread_reports_a_closed_pipe() { - let (read, write) = warning_test_pipe(); - drop(read); - let thread = start_truncation_warning(TruncationWarningParams { - stderr: write, - stopped: Arc::new(AtomicBool::new(false)), - }) - .unwrap(); - assert!(warning_thread_result(&thread).is_err()); - } - - #[test] - fn relayed_warning_to_a_full_pipe_cancels_and_releases_its_handle() { - use std::io::Read; - use windows::Win32::Foundation::ERROR_BROKEN_PIPE; - use windows::Win32::System::Pipes::PeekNamedPipe; - - let (mut read, write) = warning_test_pipe(); - let mut written = 0; - unsafe { WriteFile(write.get(), Some(&[b'x'; 4096]), Some(&mut written), None).unwrap() }; - assert_eq!(written, 4096); - let stopped = Arc::new(AtomicBool::new(false)); - let thread = start_truncation_warning(TruncationWarningParams { - stderr: write, - stopped: Arc::clone(&stopped), - }) - .unwrap(); - let started = Instant::now(); - assert_eq!( - unsafe { WaitForSingleObject(thread.get(), 50) }, - WAIT_TIMEOUT - ); - stopped.store(true, Ordering::Release); - loop { - if let Err(error) = unsafe { CancelSynchronousIo(thread.get()) } { - assert_eq!(error.code(), ERROR_NOT_FOUND.to_hresult()); - } - if unsafe { WaitForSingleObject(thread.get(), 10) } == WAIT_OBJECT_0 { - break; - } - assert!( - started.elapsed().as_secs() < 3, - "warning writer did not stop" - ); - } - assert!(warning_thread_result(&thread).is_err()); - - let mut buffered = [0; 4096]; - read.read_exact(&mut buffered).unwrap(); - assert_eq!(buffered, [b'x'; 4096]); - // SAFETY: the read handle is live. Peek cannot block if a writer leaked. - let error = - unsafe { PeekNamedPipe(HANDLE(read.as_raw_handle()), None, 0, None, None, None) } - .unwrap_err(); - assert_eq!(error.code(), ERROR_BROKEN_PIPE.to_hresult()); - } - - #[test] - fn stopped_warning_thread_does_not_start_a_write() { - use std::io::Read; - - let (mut read, write) = warning_test_pipe(); - let thread = start_truncation_warning(TruncationWarningParams { - stderr: write, - stopped: Arc::new(AtomicBool::new(true)), - }) - .unwrap(); - assert!(warning_thread_result(&thread).is_err()); - let mut output = String::new(); - read.read_to_string(&mut output).unwrap(); - assert!(output.is_empty()); - } // ====== Wire-format constants ====== diff --git a/src/mxc-sdk/src/core/mxc_common/process_util.rs b/src/mxc-sdk/src/core/mxc_common/process_util.rs index c35ffdb06..8efc26f98 100644 --- a/src/mxc-sdk/src/core/mxc_common/process_util.rs +++ b/src/mxc-sdk/src/core/mxc_common/process_util.rs @@ -146,8 +146,8 @@ unsafe impl Sync for CancelablePipe {} /// at a time (any number of cancellers may fire concurrently). /// /// Cancelling can discard data that a read has received but has not yet returned -/// to the caller. When collecting remaining output, cancel only after the -/// timeout, and warn that some output may be missing. +/// to the caller. When collecting remaining output, allow it to drain before +/// cancelling at the timeout. pub struct InterruptiblePipeReader(Arc); impl InterruptiblePipeReader { From d7b90bbfa7ba16db6435586c8bc2beba3a474f95 Mon Sep 17 00:00:00 2001 From: Dom Giandinoto Date: Tue, 20 Oct 2026 14:22:49 -0700 Subject: [PATCH 4/4] modifying the drain poll wait --- .../isolation_session/common/manager.rs | 8 +- .../isolation_session/common/pipe_relay.rs | 108 +++++++++++++++--- 2 files changed, 100 insertions(+), 16 deletions(-) diff --git a/src/mxc-sdk/src/backends/isolation_session/common/manager.rs b/src/mxc-sdk/src/backends/isolation_session/common/manager.rs index 6ffe2a518..431c3798f 100644 --- a/src/mxc-sdk/src/backends/isolation_session/common/manager.rs +++ b/src/mxc-sdk/src/backends/isolation_session/common/manager.rs @@ -582,7 +582,7 @@ impl IsolationSessionManager { let outcome = started.wait(options.timeout_ms)?; scope.stop_stdin(); - scope.finish(); + scope.finish()?; Ok(match outcome { ExecOutcome::Exited(exit_code) => exit_code, @@ -1091,14 +1091,15 @@ impl<'a> RelayScope<'a> { } /// Teardown for a workload that has already exited. - fn finish(&mut self) { + fn finish(&mut self) -> Result<(), IsolationSessionError> { self.signal_stop(); let output_relays = self.output_relays(); let output_drain = drain_output_relays_within( &output_relays, Duration::from_millis(Self::IDLE_MS), Duration::from_millis(Self::ABSOLUTE_CAP_MS), - ); + ) + .map_err(|error| lifecycle_err(format!("drain output relays: {error}")))?; if output_drain == OutputRelayDrain::GaveUp { Self::join_output(&self.stdout, Self::JOIN_MS); Self::join_output(&self.stderr, Self::JOIN_MS); @@ -1106,6 +1107,7 @@ impl<'a> RelayScope<'a> { Self::join(self.stdin.as_ref(), Self::JOIN_MS); self.close_handles(); self.finished = true; + Ok(()) } } diff --git a/src/mxc-sdk/src/backends/isolation_session/common/pipe_relay.rs b/src/mxc-sdk/src/backends/isolation_session/common/pipe_relay.rs index 012ecc51c..6cc7df1f4 100644 --- a/src/mxc-sdk/src/backends/isolation_session/common/pipe_relay.rs +++ b/src/mxc-sdk/src/backends/isolation_session/common/pipe_relay.rs @@ -23,7 +23,9 @@ use crate::mxc_common::error::WxcError; use crate::mxc_common::process_util::{InterruptiblePipeReader, OwnedHandle, PipeReadCanceller}; use crate::mxc_common::sandbox_process::StreamCloser; -use windows::Win32::Foundation::{DuplicateHandle, DUPLICATE_SAME_ACCESS, HANDLE, WAIT_OBJECT_0}; +use windows::Win32::Foundation::{ + DuplicateHandle, DUPLICATE_SAME_ACCESS, HANDLE, WAIT_OBJECT_0, WAIT_TIMEOUT, +}; use windows::Win32::Storage::FileSystem::{FlushFileBuffers, ReadFile, WriteFile}; use windows::Win32::System::Threading::{ CreateThread, GetCurrentProcess, WaitForMultipleObjects, WaitForSingleObject, @@ -188,18 +190,19 @@ pub(super) fn drain_output_relays_within( relays: &[&PipeRelay], idle_window: Duration, absolute_cap: Duration, -) -> OutputRelayDrain { +) -> Result { if relays.is_empty() || relays.iter().all(|relay| relay_is_finished(relay)) { - return OutputRelayDrain::Drained; + return Ok(OutputRelayDrain::Drained); } + let handles: Vec = relays.iter().map(|relay| relay.thread().get()).collect(); let started = Instant::now(); let mut idle_started = started; let mut last_progress = total_progress(relays); loop { if relays.iter().all(|relay| relay_is_finished(relay)) { - return OutputRelayDrain::Drained; + return Ok(OutputRelayDrain::Drained); } let now = Instant::now(); @@ -215,16 +218,22 @@ pub(super) fn drain_output_relays_within( for relay in relays { relay.cancel_read(); } - return OutputRelayDrain::GaveUp; + return Ok(OutputRelayDrain::GaveUp); } let remaining_idle = idle_window.saturating_sub(idle_elapsed); let remaining_cap = absolute_cap.saturating_sub(total_elapsed); - let sleep_for = DRAIN_POLL_INTERVAL.min(remaining_idle).min(remaining_cap); - if sleep_for.is_zero() { - std::thread::yield_now(); - } else { - std::thread::sleep(sleep_for); + let wait_for = DRAIN_POLL_INTERVAL.min(remaining_idle).min(remaining_cap); + // SAFETY: relays own all handles throughout the wait. The timeout is at most 200 ms. + match unsafe { WaitForMultipleObjects(&handles, true, wait_for.as_millis() as u32) } { + WAIT_OBJECT_0 => return Ok(OutputRelayDrain::Drained), + WAIT_TIMEOUT => {} + _ => { + return Err(WxcError::Process(format!( + "WaitForMultipleObjects for output relays failed: {}", + std::io::Error::last_os_error() + ))); + } } } } @@ -702,6 +711,76 @@ mod tests { ); } + #[test] + fn output_drain_wakes_when_all_relays_exit_between_polls() { + let (first_read, first_write) = create_test_pipes(); + let (second_read, mut second_write) = create_test_pipes(); + let (_destination_read, destination_write) = create_test_pipes(); + let first = + unsafe { create_relay_thread(first_read.get(), destination_write.get()).unwrap() }; + let second = + unsafe { create_relay_thread(second_read.get(), destination_write.get()).unwrap() }; + drop(first_write); + assert_eq!( + unsafe { WaitForSingleObject(first.thread().get(), 1000) }, + WAIT_OBJECT_0 + ); + + let source = SendOwnedHandle::take(&mut second_write); + let writer = std::thread::spawn(move || { + std::thread::sleep(Duration::from_millis(40)); + drop(source); + }); + let started = Instant::now(); + let result = drain_output_relays_within( + &[&first, &second], + Duration::from_secs(1), + Duration::from_secs(2), + ) + .unwrap(); + let elapsed = started.elapsed(); + let second_finished = unsafe { WaitForSingleObject(second.thread().get(), 0) }; + writer.join().unwrap(); + + assert_eq!(result, OutputRelayDrain::Drained); + assert_eq!(second_finished, WAIT_OBJECT_0, "must wait for both relays"); + assert!( + elapsed < DRAIN_POLL_INTERVAL, + "completion should wake the drain before the next poll: {elapsed:?}" + ); + } + + #[test] + fn output_drain_without_relays_finishes_immediately() { + assert_eq!( + drain_output_relays_within(&[], Duration::ZERO, Duration::ZERO).unwrap(), + OutputRelayDrain::Drained + ); + } + + #[test] + fn output_drain_reports_a_failed_thread_wait() { + let (source_read, _source_write) = create_test_pipes(); + let (_destination_read, destination_write) = create_test_pipes(); + let mut relay = + unsafe { create_relay_thread(source_read.get(), destination_write.get()).unwrap() }; + let thread = std::mem::replace(&mut relay.thread, OwnedHandle::new(HANDLE::default())); + let result = + drain_output_relays_within(&[&relay], Duration::from_secs(1), Duration::from_secs(2)); + relay.cancel_read(); + assert_eq!( + unsafe { WaitForSingleObject(thread.get(), 1000) }, + WAIT_OBJECT_0 + ); + assert!( + result + .unwrap_err() + .to_string() + .contains("WaitForMultipleObjects"), + "wait failures must not be reported as successful draining" + ); + } + #[test] fn output_drain_extends_while_progress_continues() { let (source_read, mut source_write) = create_test_pipes(); @@ -734,7 +813,8 @@ mod tests { &[&relay], std::time::Duration::from_millis(100), std::time::Duration::from_secs(2), - ); + ) + .unwrap(); let elapsed = started.elapsed(); writer.join().unwrap(); @@ -762,7 +842,8 @@ mod tests { &[&relay], std::time::Duration::from_millis(30), std::time::Duration::from_millis(500), - ); + ) + .unwrap(); let elapsed = started.elapsed(); assert_eq!(result, OutputRelayDrain::GaveUp); @@ -833,7 +914,8 @@ mod tests { let cap = Duration::from_millis(600); let started = Instant::now(); - let result = drain_output_relays_within(&[&relay], Duration::from_millis(300), cap); + let result = + drain_output_relays_within(&[&relay], Duration::from_millis(300), cap).unwrap(); let elapsed = started.elapsed(); assert_eq!( unsafe { WaitForSingleObject(relay.thread().get(), 5000) },