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..4ed30ef4c 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,22 @@ 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 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 `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..431c3798f 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()?; 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,30 +1068,46 @@ 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) -> Result<(), IsolationSessionError> { 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), + ) + .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); } - Self::join(&self.stdin, Self::JOIN_MS); + Self::join(self.stdin.as_ref(), Self::JOIN_MS); self.close_handles(); self.finished = true; + Ok(()) } } @@ -1110,11 +1117,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..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 @@ -6,24 +6,34 @@ //! 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::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, 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) fn duplicate_handle(handle: HANDLE) -> Result { let mut duplicated = HANDLE::default(); @@ -45,12 +55,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 +124,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 +134,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 +144,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 +163,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 +180,76 @@ 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, +) -> Result { + if relays.is_empty() || relays.iter().all(|relay| relay_is_finished(relay)) { + 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 Ok(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(); + } + return Ok(OutputRelayDrain::GaveUp); + } + + let remaining_idle = idle_window.saturating_sub(idle_elapsed); + let remaining_cap = absolute_cap.saturating_sub(total_elapsed); + 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() + ))); + } + } + } +} + +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 +381,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 +457,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 +484,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 +505,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 +532,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 +547,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 +561,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 +593,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 +612,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 +632,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 +659,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 +693,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 +711,233 @@ 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(); + 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 result = drain_output_relays_within( + &[&relay], + std::time::Duration::from_millis(100), + std::time::Duration::from_secs(2), + ) + .unwrap(); + let elapsed = started.elapsed(); + + writer.join().unwrap(); + drop(dest_write); + let output = reader.join().unwrap(); + + assert_eq!(result, OutputRelayDrain::Drained); + 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 result = drain_output_relays_within( + &[&relay], + std::time::Duration::from_millis(30), + std::time::Duration::from_millis(500), + ) + .unwrap(); + 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!( + 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 result = + drain_output_relays_within(&[&relay], Duration::from_millis(300), cap).unwrap(); + 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!(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/core/mxc_common/process_util.rs b/src/mxc-sdk/src/core/mxc_common/process_util.rs index 749f38882..8efc26f98 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, allow it to drain before +/// cancelling at the timeout. pub struct InterruptiblePipeReader(Arc); impl InterruptiblePipeReader {