From 19ad9d6b8b68a22e4534b86f4622770e2a3ff753 Mon Sep 17 00:00:00 2001 From: Vladyslav Nikonov Date: Thu, 6 Aug 2026 21:35:53 +0300 Subject: [PATCH 1/4] feat(agent): stream package operation output over per-operation event channel For each executed package operation, the broker now creates a dedicated local named pipe (Devolutions.Now.PackageBroker.Operation.) before returning the execution response and advertises it via the event_channel field of the operation submission. The channel streams one-way server-to-client event frames per the event-channel protocol v1.0 (now-policy-api 0.3): HELLO on connect, STDOUT/STDERR data frames when CaptureOutput is requested, STATUS_UPDATED on every status transition, and FINISH when the operation reaches a terminal status. Output is chunked without splitting UTF-8 characters across frames. Slow or absent clients never stall the operation: data is dropped past a bounded per-stream budget and reported via overflow frames. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- devolutions-agent/src/broker/event_channel.rs | 866 ++++++++++++++++++ devolutions-agent/src/broker/executor/mod.rs | 13 + .../src/broker/executor/windows/mod.rs | 13 +- .../src/broker/executor/windows/process.rs | 147 ++- devolutions-agent/src/broker/mod.rs | 1 + .../src/broker/operation_tracker.rs | 41 +- devolutions-agent/src/broker/server/mod.rs | 186 +++- 7 files changed, 1221 insertions(+), 46 deletions(-) create mode 100644 devolutions-agent/src/broker/event_channel.rs diff --git a/devolutions-agent/src/broker/event_channel.rs b/devolutions-agent/src/broker/event_channel.rs new file mode 100644 index 000000000..13138699e --- /dev/null +++ b/devolutions-agent/src/broker/event_channel.rs @@ -0,0 +1,866 @@ +//! Per-operation event channel. +//! +//! For each executed operation the broker opens a dedicated local named pipe and +//! streams `NOW_BROKER` event frames to the client (see the `event_channel` module +//! of `now-policy-api`): a `Hello` frame first, then `Stdout`/`Stderr` data frames +//! (only when the request opted in via `CaptureOutput`), a `StatusUpdated` frame on +//! every status transition, and a final `Finish` frame once the operation reaches a +//! terminal status. +//! +//! Writing is strictly best-effort: a client that never connects, connects late, +//! reads too slowly, or disconnects early must never stall or fail the operation. +//! Producers push into a bounded in-memory queue; when the per-stream byte budget +//! is exhausted, data is dropped and accounted for with `StdoutOverflow` / +//! `StderrOverflow` frames. + +use std::collections::VecDeque; +use std::mem; +use std::sync::{Arc, Mutex}; + +use now_policy_api::event_channel::{EventFrame, MAX_EVENT_FRAME_BODY_BYTES}; +use tokio::sync::Notify; +use tokio_util::sync::CancellationToken; + +/// Prefix of per-operation event pipe names (without the `\\.\pipe\` namespace). +pub const OPERATION_PIPE_PREFIX: &str = "Devolutions.Now.PackageBroker.Operation."; + +/// Maximum bytes of not-yet-written output buffered per stream (stdout / stderr). +/// +/// When the budget is exhausted (client absent or reading too slowly), further +/// output is dropped and reported through overflow frames. +const PER_STREAM_BUDGET_BYTES: usize = 256 * 1024; + +/// Bare pipe name for an operation's event channel, as returned to clients in the +/// `EventChannel` descriptor (no `\\.\pipe\` prefix, matching the protocol samples). +pub fn operation_pipe_name(operation_id: &str) -> String { + format!("{OPERATION_PIPE_PREFIX}{operation_id}") +} + +/// UTF-8-safe splitter turning a stream of raw bytes into `String` chunks. +/// +/// Bytes may arrive split in the middle of a multi-byte UTF-8 sequence; the +/// incomplete trailing sequence is buffered until more bytes arrive. Invalid +/// sequences are replaced with U+FFFD (lossy). Yielded chunks never exceed +/// [`MAX_EVENT_FRAME_BODY_BYTES`] and never split a character across chunks. +#[derive(Debug, Default)] +pub struct Utf8StreamChunker { + /// Incomplete trailing UTF-8 sequence carried over to the next push (< 4 bytes). + pending: Vec, +} + +impl Utf8StreamChunker { + /// Feed raw bytes; returns zero or more complete UTF-8 chunks. + pub fn push(&mut self, bytes: &[u8]) -> Vec { + let mut buffer = mem::take(&mut self.pending); + buffer.extend_from_slice(bytes); + + let mut decoded = String::new(); + let mut rest: &[u8] = &buffer; + loop { + match core::str::from_utf8(rest) { + Ok(valid) => { + decoded.push_str(valid); + break; + } + Err(error) => { + let (valid, after_valid) = rest.split_at(error.valid_up_to()); + // INVARIANT: `valid` covers `error.valid_up_to()` bytes, which + // `from_utf8` guarantees to be valid UTF-8. + decoded.push_str(core::str::from_utf8(valid).expect("validated prefix")); + match error.error_len() { + Some(invalid_len) => { + decoded.push(char::REPLACEMENT_CHARACTER); + rest = &after_valid[invalid_len..]; + } + None => { + // Unexpected end of input: possibly a character split + // across reads; wait for more bytes. + self.pending = after_valid.to_vec(); + break; + } + } + } + } + } + + split_at_char_boundaries(decoded, MAX_EVENT_FRAME_BODY_BYTES) + } + + /// Flush the buffered incomplete sequence (if any) at end of stream. + pub fn flush(&mut self) -> Option { + if self.pending.is_empty() { + return None; + } + self.pending.clear(); + // The pending bytes can no longer be completed: they decode lossily to a + // single replacement character. + Some(char::REPLACEMENT_CHARACTER.to_string()) + } +} + +/// Split `text` into chunks of at most `max_bytes` bytes, cutting only at +/// character boundaries. +fn split_at_char_boundaries(text: String, max_bytes: usize) -> Vec { + if text.is_empty() { + return Vec::new(); + } + if text.len() <= max_bytes { + return vec![text]; + } + + let mut chunks = Vec::new(); + let mut rest = text.as_str(); + while rest.len() > max_bytes { + let mut cut = max_bytes; + while !rest.is_char_boundary(cut) { + cut -= 1; + } + chunks.push(rest[..cut].to_owned()); + rest = &rest[cut..]; + } + if !rest.is_empty() { + chunks.push(rest.to_owned()); + } + chunks +} + +/// Which output stream a data chunk belongs to. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum OutputStream { + Stdout, + Stderr, +} + +/// State of one output stream inside the queue. +#[derive(Debug, Default)] +struct StreamState { + chunker: Utf8StreamChunker, + /// Bytes of queued (not yet consumed) data frames for this stream. + queued_bytes: usize, + /// Bytes dropped since the last overflow frame was queued. + skipped_bytes: u32, +} + +#[derive(Debug)] +struct QueueState { + items: VecDeque, + stdout: StreamState, + stderr: StreamState, + /// Set once `finish` was queued; no further items are accepted. + closed: bool, +} + +/// Shared bounded event queue between producers (executor, tracker) and the +/// pipe writer task. +#[derive(Debug)] +struct EventQueue { + state: Mutex, + /// Signalled whenever an item is queued or the queue is closed. + notify: Notify, + /// Fired when the queue is closed (operation reached a terminal status). + finished: CancellationToken, +} + +/// Cloneable producer handle for an operation's event channel. +/// +/// All methods are non-blocking and infallible: they enqueue frames for the +/// writer task (or drop data once the bounded budget is exhausted) and are safe +/// to call from blocking threads. +#[derive(Debug, Clone)] +pub struct OperationEventSink { + queue: Arc, +} + +impl OperationEventSink { + fn new() -> Self { + Self { + queue: Arc::new(EventQueue { + state: Mutex::new(QueueState { + items: VecDeque::new(), + stdout: StreamState::default(), + stderr: StreamState::default(), + closed: false, + }), + notify: Notify::new(), + finished: CancellationToken::new(), + }), + } + } + + /// Queue raw stdout bytes (chunked into UTF-8-safe data frames). + pub fn stdout(&self, bytes: &[u8]) { + self.push_output(OutputStream::Stdout, bytes); + } + + /// Queue raw stderr bytes (chunked into UTF-8-safe data frames). + pub fn stderr(&self, bytes: &[u8]) { + self.push_output(OutputStream::Stderr, bytes); + } + + /// Queue a `StatusUpdated` notification frame. + pub fn status_updated(&self) { + let mut state = self.queue.state.lock().expect("event queue lock poisoned"); + if state.closed { + return; + } + state.items.push_back(EventFrame::StatusUpdated); + drop(state); + self.queue.notify.notify_one(); + } + + /// Queue the final `Finish` frame and close the queue. + /// + /// Flushes buffered incomplete UTF-8 sequences and pending overflow + /// accounting first, so `Finish` is always the last frame. + pub fn finish(&self) { + let mut state = self.queue.state.lock().expect("event queue lock poisoned"); + if state.closed { + return; + } + for stream in [OutputStream::Stdout, OutputStream::Stderr] { + let stream_state = match stream { + OutputStream::Stdout => &mut state.stdout, + OutputStream::Stderr => &mut state.stderr, + }; + let tail = stream_state.chunker.flush(); + if let Some(tail) = tail { + Self::enqueue_chunk(&mut state, stream, tail); + } + let stream_state = match stream { + OutputStream::Stdout => &mut state.stdout, + OutputStream::Stderr => &mut state.stderr, + }; + let skipped = mem::take(&mut stream_state.skipped_bytes); + if skipped > 0 { + state.items.push_back(overflow_frame(stream, skipped)); + } + } + state.items.push_back(EventFrame::Finish); + state.closed = true; + drop(state); + self.queue.finished.cancel(); + self.queue.notify.notify_one(); + } + + fn push_output(&self, stream: OutputStream, bytes: &[u8]) { + let mut state = self.queue.state.lock().expect("event queue lock poisoned"); + if state.closed { + return; + } + let stream_state = match stream { + OutputStream::Stdout => &mut state.stdout, + OutputStream::Stderr => &mut state.stderr, + }; + let chunks = stream_state.chunker.push(bytes); + let mut queued_any = false; + for chunk in chunks { + Self::enqueue_chunk(&mut state, stream, chunk); + queued_any = true; + } + drop(state); + if queued_any { + self.queue.notify.notify_one(); + } + } + + /// Queue one data chunk, respecting the per-stream byte budget. + /// + /// A chunk that does not fit is dropped in full and accounted in the skipped + /// counter; once room is available again, an overflow frame is queued before + /// the next data frame so the client learns how many bytes it missed. + fn enqueue_chunk(state: &mut QueueState, stream: OutputStream, chunk: String) { + let stream_state = match stream { + OutputStream::Stdout => &mut state.stdout, + OutputStream::Stderr => &mut state.stderr, + }; + if stream_state.queued_bytes + chunk.len() > PER_STREAM_BUDGET_BYTES { + let len = u32::try_from(chunk.len()).unwrap_or(u32::MAX); + stream_state.skipped_bytes = stream_state.skipped_bytes.saturating_add(len); + return; + } + let skipped = mem::take(&mut stream_state.skipped_bytes); + stream_state.queued_bytes += chunk.len(); + if skipped > 0 { + state.items.push_back(overflow_frame(stream, skipped)); + } + state.items.push_back(data_frame(stream, chunk)); + } + + /// Wait for and take the next queued frame; `None` once the queue is closed + /// and fully drained. + async fn next_frame(&self) -> Option { + loop { + let notified = self.queue.notify.notified(); + { + let mut state = self.queue.state.lock().expect("event queue lock poisoned"); + if let Some(frame) = state.items.pop_front() { + match &frame { + EventFrame::Stdout(data) => state.stdout.queued_bytes -= data.len(), + EventFrame::Stderr(data) => state.stderr.queued_bytes -= data.len(), + _ => {} + } + // Wake any further waiter (Notify stores a single permit). + if !state.items.is_empty() { + self.queue.notify.notify_one(); + } + return Some(frame); + } + if state.closed { + return None; + } + } + notified.await; + } + } + + /// Completes once [`OperationEventSink::finish`] was called. + async fn finished(&self) { + self.queue.finished.cancelled().await; + } +} + +fn data_frame(stream: OutputStream, chunk: String) -> EventFrame { + match stream { + OutputStream::Stdout => EventFrame::Stdout(chunk), + OutputStream::Stderr => EventFrame::Stderr(chunk), + } +} + +fn overflow_frame(stream: OutputStream, bytes_skipped: u32) -> EventFrame { + match stream { + OutputStream::Stdout => EventFrame::StdoutOverflow { bytes_skipped }, + OutputStream::Stderr => EventFrame::StderrOverflow { bytes_skipped }, + } +} + +#[cfg(windows)] +pub use windows_channel::open_operation_channel; + +#[cfg(windows)] +mod windows_channel { + use std::time::Duration; + + use anyhow::Context as _; + use now_policy_api::event_channel::{EVENT_CHANNEL_VERSION_MAJOR, EVENT_CHANNEL_VERSION_MINOR, EventFrame}; + use now_policy_api::{EventChannel, EventChannelKind}; + use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _}; + use tokio::net::windows::named_pipe::{NamedPipeServer, ServerOptions}; + use tracing::{debug, warn}; + use win_api_wrappers::identity::sid::Sid; + use win_api_wrappers::security::acl::{Acl, ExplicitAccess, InheritableAcl, InheritableAclKind, Trustee}; + use win_api_wrappers::security::attributes::SecurityAttributesInit; + use windows::Win32::Foundation::GENERIC_ALL; + use windows::Win32::Security; + use windows::Win32::Security::Authorization::SET_ACCESS; + use windows::Win32::Storage::FileSystem::FILE_GENERIC_READ; + + use super::{OperationEventSink, operation_pipe_name}; + + /// How long the channel stays available for a late client connection after the + /// operation finished without any client having connected. + const NEVER_CONNECTED_LINGER: Duration = Duration::from_secs(60); + + /// After the final `Finish` frame is written, how long the writer waits for the + /// client to drain the pipe and close its end before tearing the pipe down. + const CLIENT_DRAIN_TIMEOUT: Duration = Duration::from_secs(30); + + /// Create the event channel for an operation. + /// + /// The named pipe instance is created (with its security descriptor) before this + /// function returns, so the path in the returned descriptor is immediately + /// connectable. A background task then serves at most one client, best-effort. + /// + /// `client_sid` is the authenticated requesting user; only that user (plus + /// SYSTEM and Administrators) can open the pipe. + pub fn open_operation_channel( + operation_id: &str, + client_sid: &Sid, + ) -> anyhow::Result<(OperationEventSink, EventChannel)> { + let pipe_name = operation_pipe_name(operation_id); + let pipe_path = format!(r"\\.\pipe\{pipe_name}"); + + let security_attributes = build_channel_security_attributes(client_sid) + .context("failed to build event channel security attributes")?; + + // SAFETY: `create_with_security_attributes_raw` requires a pointer to a valid + // `SECURITY_ATTRIBUTES` living across the call. `security_attributes` owns the + // structure and its descriptor and outlives the call; `CreateNamedPipeW` copies + // the descriptor and does not retain the pointer. + let server = unsafe { + ServerOptions::new() + .access_inbound(false) + .first_pipe_instance(true) + .max_instances(1) + .create_with_security_attributes_raw(&pipe_path, security_attributes.as_mut_ptr().cast()) + } + .with_context(|| format!("failed to create event channel pipe '{pipe_path}'"))?; + + let sink = OperationEventSink::new(); + let descriptor = EventChannel { + kind: EventChannelKind::LocalPipe, + path: pipe_name, + }; + + let writer_sink = sink.clone(); + let operation_id = operation_id.to_owned(); + tokio::spawn(async move { + run_channel(server, writer_sink, &operation_id).await; + }); + + Ok((sink, descriptor)) + } + + /// Serve one client connection on the operation's event pipe, best-effort. + async fn run_channel(server: NamedPipeServer, sink: OperationEventSink, operation_id: &str) { + // Wait for a client; once the operation finishes without one, linger only + // for a bounded grace period so late clients can still fetch the frames. + let connected = tokio::select! { + result = server.connect() => result.is_ok(), + () = sink.finished() => { + matches!( + tokio::time::timeout(NEVER_CONNECTED_LINGER, server.connect()).await, + Ok(Ok(())) + ) + } + }; + if !connected { + debug!(operation_id, "No client connected to the event channel"); + return; + } + + debug!(operation_id, "Client connected to the event channel"); + let mut server = server; + + let hello = EventFrame::Hello { + version_major: EVENT_CHANNEL_VERSION_MAJOR, + version_minor: EVENT_CHANNEL_VERSION_MINOR, + }; + if let Err(error) = write_frame(&mut server, &hello).await { + warn!(operation_id, %error, "Failed to write event channel hello frame"); + return; + } + + while let Some(frame) = sink.next_frame().await { + if let Err(error) = write_frame(&mut server, &frame).await { + // Best-effort: the client disconnected or stopped reading; the + // operation itself is unaffected. + debug!(operation_id, %error, "Stopped writing event channel frames"); + return; + } + } + + // All frames (ending with `Finish`) were handed to the pipe; give the + // client a bounded amount of time to drain and close its end so buffered + // data is not discarded by our handle closing first. + let _ = tokio::time::timeout(CLIENT_DRAIN_TIMEOUT, async { + let mut scratch = [0u8; 16]; + loop { + match server.read(&mut scratch).await { + Ok(0) | Err(_) => break, + Ok(_) => {} + } + } + }) + .await; + debug!(operation_id, "Event channel closed"); + } + + async fn write_frame(server: &mut NamedPipeServer, frame: &EventFrame) -> anyhow::Result<()> { + let bytes = frame.encode().context("failed to encode event frame")?; + server.write_all(&bytes).await.context("failed to write event frame")?; + server.flush().await.context("failed to flush event frame")?; + Ok(()) + } + + /// Security descriptor for a per-operation event pipe: + /// - SYSTEM: full control + /// - Administrators: full control + /// - requesting client user: read (the channel is one-way, server to client) + fn build_channel_security_attributes( + client_sid: &Sid, + ) -> anyhow::Result { + let system_sid = + Sid::from_well_known(Security::WinLocalSystemSid, None).context("failed to create SYSTEM SID")?; + let admins_sid = Sid::from_well_known(Security::WinBuiltinAdministratorsSid, None) + .context("failed to create Administrators SID")?; + + let entries = [ + ExplicitAccess { + access_permissions: GENERIC_ALL.0, + access_mode: SET_ACCESS, + inheritance: Security::ACE_FLAGS(0), + trustee: Trustee::Sid(system_sid), + }, + ExplicitAccess { + access_permissions: GENERIC_ALL.0, + access_mode: SET_ACCESS, + inheritance: Security::ACE_FLAGS(0), + trustee: Trustee::Sid(admins_sid), + }, + ExplicitAccess { + access_permissions: FILE_GENERIC_READ.0, + access_mode: SET_ACCESS, + inheritance: Security::ACE_FLAGS(0), + trustee: Trustee::Sid(client_sid.clone()), + }, + ]; + + let empty_acl = Acl::new().context("failed to create empty ACL")?; + let dacl = empty_acl.set_entries(&entries).context("failed to set ACL entries")?; + + Ok(SecurityAttributesInit { + dacl: Some(InheritableAcl { + kind: InheritableAclKind::Protected, + acl: dacl, + }), + ..Default::default() + } + .init()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn collect(chunker: &mut Utf8StreamChunker, parts: &[&[u8]]) -> Vec { + let mut out = Vec::new(); + for part in parts { + out.extend(chunker.push(part)); + } + if let Some(tail) = chunker.flush() { + out.push(tail); + } + out + } + + #[test] + fn chunker_passes_ascii_through() { + let mut chunker = Utf8StreamChunker::default(); + assert_eq!(collect(&mut chunker, &[b"hello world"]), ["hello world"]); + } + + #[test] + fn chunker_reassembles_char_split_across_pushes() { + let bytes = "héllo π".as_bytes(); + for split in 1..bytes.len() { + let mut chunker = Utf8StreamChunker::default(); + let chunks = collect(&mut chunker, &[&bytes[..split], &bytes[split..]]); + assert_eq!(chunks.concat(), "héllo π", "split at {split}"); + } + } + + #[test] + fn chunker_replaces_invalid_bytes() { + let mut chunker = Utf8StreamChunker::default(); + let chunks = collect(&mut chunker, &[&[b'a', 0xff, 0xfe, b'b']]); + assert_eq!(chunks.concat(), "a\u{FFFD}\u{FFFD}b"); + } + + #[test] + fn chunker_flushes_incomplete_trailing_sequence_as_replacement() { + let mut chunker = Utf8StreamChunker::default(); + // First two bytes of a three-byte character. + let chunks = collect(&mut chunker, &[&[b'x', 0xE2, 0x82]]); + assert_eq!(chunks.concat(), "x\u{FFFD}"); + } + + #[test] + fn chunker_respects_max_frame_body_size_and_char_boundaries() { + // 'é' is 2 bytes; an odd max forces the boundary check to back off. + let text = "é".repeat(MAX_EVENT_FRAME_BODY_BYTES); // 2 * MAX bytes total. + let mut chunker = Utf8StreamChunker::default(); + let chunks = chunker.push(text.as_bytes()); + assert!(chunks.len() >= 2); + for chunk in &chunks { + assert!(chunk.len() <= MAX_EVENT_FRAME_BODY_BYTES); + assert!(chunk.chars().all(|c| c == 'é')); + } + assert_eq!(chunks.concat(), text); + } + + #[test] + fn split_at_char_boundaries_never_splits_chars() { + let text = "aπ".repeat(10); + let chunks = split_at_char_boundaries(text.clone(), 4); + assert!(chunks.iter().all(|c| c.len() <= 4)); + assert_eq!(chunks.concat(), text); + } + + #[tokio::test] + async fn sink_orders_data_status_and_finish() { + let sink = OperationEventSink::new(); + sink.stdout(b"out"); + sink.status_updated(); + sink.stderr(b"err"); + sink.finish(); + + let mut frames = Vec::new(); + while let Some(frame) = sink.next_frame().await { + frames.push(frame); + } + assert_eq!( + frames, + [ + EventFrame::Stdout("out".to_owned()), + EventFrame::StatusUpdated, + EventFrame::Stderr("err".to_owned()), + EventFrame::Finish, + ] + ); + } + + #[tokio::test] + async fn sink_ignores_events_after_finish() { + let sink = OperationEventSink::new(); + sink.finish(); + sink.stdout(b"late"); + sink.status_updated(); + sink.finish(); + + assert_eq!(sink.next_frame().await, Some(EventFrame::Finish)); + assert_eq!(sink.next_frame().await, None); + } + + #[tokio::test] + async fn sink_accounts_overflow_when_budget_is_exhausted() { + let sink = OperationEventSink::new(); + let chunk = vec![b'a'; 64 * 1024]; + let pushes = 64; // 4 MiB total, way over the 256 KiB budget. + for _ in 0..pushes { + sink.stdout(&chunk); + } + sink.finish(); + + let mut received = 0usize; + let mut skipped = 0u64; + let mut saw_finish = false; + while let Some(frame) = sink.next_frame().await { + match frame { + EventFrame::Stdout(data) => received += data.len(), + EventFrame::StdoutOverflow { bytes_skipped } => skipped += u64::from(bytes_skipped), + EventFrame::Finish => saw_finish = true, + other => panic!("unexpected frame: {other:?}"), + } + } + + assert!(saw_finish); + assert!(skipped > 0, "expected dropped bytes"); + assert!(received <= PER_STREAM_BUDGET_BYTES); + assert_eq!(received as u64 + skipped, (chunk.len() * pushes) as u64); + } + + #[tokio::test] + async fn sink_emits_overflow_frame_before_next_data_frame() { + let sink = OperationEventSink::new(); + let big = vec![b'a'; PER_STREAM_BUDGET_BYTES]; // Fills the budget exactly (4 frames). + sink.stdout(&big); + sink.stdout(b"dropped"); + + // Drain the queued data to free budget, then push more data. + let mut drained = 0; + while drained < PER_STREAM_BUDGET_BYTES { + match sink.next_frame().await { + Some(EventFrame::Stdout(data)) => drained += data.len(), + other => panic!("unexpected frame: {other:?}"), + } + } + sink.stdout(b"fresh"); + sink.finish(); + + assert_eq!( + sink.next_frame().await, + Some(EventFrame::StdoutOverflow { bytes_skipped: 7 }) + ); + assert_eq!(sink.next_frame().await, Some(EventFrame::Stdout("fresh".to_owned()))); + assert_eq!(sink.next_frame().await, Some(EventFrame::Finish)); + assert_eq!(sink.next_frame().await, None); + } +} + +#[cfg(all(test, windows))] +mod pipe_tests { + use now_policy_api::EventChannelKind; + use now_policy_api::event_channel::{ + EVENT_CHANNEL_VERSION_MAJOR, EVENT_CHANNEL_VERSION_MINOR, EventFrame, EventFrameDecoder, + }; + use tokio::io::AsyncReadExt as _; + use win_api_wrappers::identity::sid::Sid; + use win_api_wrappers::process::Process; + use windows::Win32::Security::TOKEN_QUERY; + + use super::*; + + fn current_user_sid() -> Sid { + Process::current_process() + .token(TOKEN_QUERY) + .expect("open current process token") + .sid_and_attributes() + .expect("query token user SID") + .sid + } + + fn test_operation_id(tag: &str) -> String { + format!("test-{tag}-{}", uuid::Uuid::new_v4()) + } + + async fn read_all_frames(pipe_name: &str) -> Vec { + let path = format!(r"\\.\pipe\{pipe_name}"); + let mut client = tokio::net::windows::named_pipe::ClientOptions::new() + .write(false) + .open(&path) + .expect("connect to event channel pipe"); + + let mut decoder = EventFrameDecoder::new(); + let mut frames = Vec::new(); + let mut buffer = [0u8; 4096]; + loop { + let read = match client.read(&mut buffer).await { + Ok(0) => break, + Ok(read) => read, + Err(_) => break, + }; + decoder.extend(&buffer[..read]); + while let Some(frame) = decoder.next_frame().expect("valid frame stream") { + let finish = frame == EventFrame::Finish; + frames.push(frame); + if finish { + return frames; + } + } + } + assert!(!decoder.has_buffered_data(), "stream truncated mid-frame"); + frames + } + + fn assert_hello_first(frames: &[EventFrame]) { + assert_eq!( + frames.first(), + Some(&EventFrame::Hello { + version_major: EVENT_CHANNEL_VERSION_MAJOR, + version_minor: EVENT_CHANNEL_VERSION_MINOR, + }) + ); + } + + #[tokio::test] + async fn channel_streams_hello_data_status_and_finish() { + let operation_id = test_operation_id("stream"); + let (sink, descriptor) = open_operation_channel(&operation_id, ¤t_user_sid()).expect("open channel"); + assert_eq!(descriptor.kind, EventChannelKind::LocalPipe); + assert_eq!(descriptor.path, operation_pipe_name(&operation_id)); + assert!( + !descriptor.path.contains('\\'), + "descriptor path must be the bare pipe name" + ); + + let reader = tokio::spawn({ + let pipe_name = descriptor.path.clone(); + async move { read_all_frames(&pipe_name).await } + }); + + sink.status_updated(); // Running. + sink.stdout("hello π".as_bytes()); + sink.stderr(b"warning"); + sink.status_updated(); // Completed. + sink.finish(); + + let frames = reader.await.expect("reader task"); + assert_hello_first(&frames); + assert_eq!( + frames[1..], + [ + EventFrame::StatusUpdated, + EventFrame::Stdout("hello π".to_owned()), + EventFrame::Stderr("warning".to_owned()), + EventFrame::StatusUpdated, + EventFrame::Finish, + ] + ); + } + + #[tokio::test] + async fn channel_without_client_does_not_block_operation() { + let operation_id = test_operation_id("noclient"); + let (sink, _descriptor) = open_operation_channel(&operation_id, ¤t_user_sid()).expect("open channel"); + + // No client ever connects; all sink calls must return immediately. + sink.status_updated(); + sink.stdout(&vec![b'x'; 1024 * 1024]); + sink.finish(); + } + + #[tokio::test] + async fn channel_serves_frames_to_late_client_after_finish() { + let operation_id = test_operation_id("late"); + let (sink, descriptor) = open_operation_channel(&operation_id, ¤t_user_sid()).expect("open channel"); + + sink.stdout(b"already done"); + sink.status_updated(); + sink.finish(); + + // Client connects only after the operation finished (within the linger window). + let frames = read_all_frames(&descriptor.path).await; + assert_hello_first(&frames); + assert_eq!( + frames[1..], + [ + EventFrame::Stdout("already done".to_owned()), + EventFrame::StatusUpdated, + EventFrame::Finish, + ] + ); + } + + #[tokio::test] + async fn slow_reader_gets_overflow_frames_and_operation_is_unaffected() { + let operation_id = test_operation_id("slow"); + let (sink, descriptor) = open_operation_channel(&operation_id, ¤t_user_sid()).expect("open channel"); + + // Connect but do not read yet. + let path = format!(r"\\.\pipe\{}", descriptor.path); + let mut client = tokio::net::windows::named_pipe::ClientOptions::new() + .write(false) + .open(&path) + .expect("connect to event channel pipe"); + + // Push far more than the per-stream budget plus any pipe buffering. + let chunk = vec![b'a'; 64 * 1024]; + let pushes = 256; // 16 MiB. + for _ in 0..pushes { + sink.stdout(&chunk); + } + sink.finish(); + + // Now read everything. + let mut decoder = EventFrameDecoder::new(); + let mut received = 0u64; + let mut skipped = 0u64; + let mut saw_finish = false; + let mut buffer = [0u8; 4096]; + 'outer: loop { + let read = match client.read(&mut buffer).await { + Ok(0) => break, + Ok(read) => read, + Err(_) => break, + }; + decoder.extend(&buffer[..read]); + while let Some(frame) = decoder.next_frame().expect("valid frame stream") { + match frame { + EventFrame::Hello { .. } => {} + EventFrame::Stdout(data) => received += data.len() as u64, + EventFrame::StdoutOverflow { bytes_skipped } => skipped += u64::from(bytes_skipped), + EventFrame::Finish => { + saw_finish = true; + break 'outer; + } + other => panic!("unexpected frame: {other:?}"), + } + } + } + + assert!(saw_finish, "finish frame must arrive even for slow readers"); + assert!(skipped > 0, "slow reader must observe overflow"); + assert_eq!(received + skipped, (chunk.len() * pushes) as u64); + } +} diff --git a/devolutions-agent/src/broker/executor/mod.rs b/devolutions-agent/src/broker/executor/mod.rs index f7cd0c39a..61f0287b6 100644 --- a/devolutions-agent/src/broker/executor/mod.rs +++ b/devolutions-agent/src/broker/executor/mod.rs @@ -9,6 +9,8 @@ use tokio_util::sync::CancellationToken; use tracing::info; use win_api_wrappers::identity::sid::Sid; +use crate::broker::event_channel::OperationEventSink; + mod output; #[cfg(windows)] @@ -63,6 +65,12 @@ pub struct ExecutionContext { /// after a grace period) and reports the cancellation through an error chain /// containing [`OperationCanceled`]. pub cancel_token: CancellationToken, + /// Per-operation event channel sink, when one was opened for this operation. + /// + /// The executor forwards the main (and pre-operation) command's stdout/stderr + /// chunks to the sink when [`ExecutionContext::capture_output`] is true. All + /// sink calls are non-blocking and best-effort. + pub event_sink: Option, } /// Marker error reported by executors when an operation was terminated because its @@ -144,6 +152,11 @@ impl CommandExecutor for DryRunExecutor { elevation = %ctx.elevation, "Dry-run: would execute plan" ); + if ctx.capture_output + && let Some(sink) = &ctx.event_sink + { + sink.stdout(format!("dry-run: would execute {:?}\n", ctx.command).as_bytes()); + } Ok(ExecutionOutput::default()) } } diff --git a/devolutions-agent/src/broker/executor/windows/mod.rs b/devolutions-agent/src/broker/executor/windows/mod.rs index a6da675e9..1f77ee71a 100644 --- a/devolutions-agent/src/broker/executor/windows/mod.rs +++ b/devolutions-agent/src/broker/executor/windows/mod.rs @@ -29,7 +29,7 @@ mod process; mod token; use privileges::SharedPrivileges; -use process::create_process; +use process::{OutputCapture, create_process}; use token::{detect_running_as_system, find_user_session, get_elevated_token}; /// Windows command executor using `win-api-wrappers` safe abstractions. @@ -367,7 +367,7 @@ fn run_plan( token, &kill_cmd, session_id, - false, + OutputCapture::disabled(), requires_elevation, None, Some(&ctx.cancel_token), @@ -386,7 +386,7 @@ fn run_plan( token, command.args(), session_id, - ctx.capture_output, + OutputCapture::new(ctx.capture_output, ctx.event_sink.as_ref()), requires_elevation, None, Some(&ctx.cancel_token), @@ -407,7 +407,7 @@ fn run_plan( token, command.args(), session_id, - ctx.capture_output, + OutputCapture::new(ctx.capture_output, ctx.event_sink.as_ref()), requires_elevation, process_started, Some(&ctx.cancel_token), @@ -427,7 +427,7 @@ fn run_plan( token, command.args(), session_id, - false, + OutputCapture::disabled(), requires_elevation, None, Some(&ctx.cancel_token), @@ -1953,6 +1953,7 @@ mod tests { scope: Some(Scope::User), capture_output: false, cancel_token: tokio_util::sync::CancellationToken::new(), + event_sink: None, }; let error = reject_unsupported_vcpkg_elevation(&ctx).expect_err("elevated vcpkg should fail"); @@ -1977,6 +1978,7 @@ mod tests { scope: Some(Scope::User), capture_output: false, cancel_token: tokio_util::sync::CancellationToken::new(), + event_sink: None, }; let executor = WindowsExecutor { is_system: true }; @@ -2005,6 +2007,7 @@ mod tests { scope: Some(Scope::User), capture_output: false, cancel_token: tokio_util::sync::CancellationToken::new(), + event_sink: None, }; let error = execute_as_current_user(&ctx, None).expect_err("mismatched client SID should fail"); diff --git a/devolutions-agent/src/broker/executor/windows/process.rs b/devolutions-agent/src/broker/executor/windows/process.rs index 6d928a472..c73ac1975 100644 --- a/devolutions-agent/src/broker/executor/windows/process.rs +++ b/devolutions-agent/src/broker/executor/windows/process.rs @@ -19,6 +19,7 @@ use windows::Win32::System::Threading::{ }; use windows::Win32::UI::WindowsAndMessaging::SW_HIDE; +use crate::broker::event_channel::{OperationEventSink, OutputStream}; use crate::broker::executor::{ ExecutionOutput, MAX_CAPTURED_OUTPUT_BYTES, OperationCanceled, ProcessStartedCallback, tail_utf8, }; @@ -32,12 +33,39 @@ const CANCEL_GRACE_PERIOD: Duration = Duration::from_secs(60); /// Granularity of the wait loop used to observe cancellation requests. const WAIT_SLICE_MS: u32 = 500; +/// Output capture configuration for [`create_process`]. +#[derive(Clone, Copy)] +pub(super) struct OutputCapture<'a> { + /// Whether the child's stdout/stderr are redirected and captured at all. + pub(super) enabled: bool, + /// Event sink receiving raw output chunks as they arrive (when capturing). + pub(super) event_sink: Option<&'a OperationEventSink>, +} + +impl<'a> OutputCapture<'a> { + pub(super) fn disabled() -> Self { + Self { + enabled: false, + event_sink: None, + } + } + + pub(super) fn new(enabled: bool, event_sink: Option<&'a OperationEventSink>) -> Self { + Self { + enabled, + event_sink: if enabled { event_sink } else { None }, + } + } +} + /// Create a process under the given token and wait for exit. /// /// This is the unified process-creation path used by both SYSTEM and current-user modes. /// The process always runs with no visible window (`STARTF_USESHOWWINDOW` + `SW_HIDE`, the /// same approach `devolutions-session` uses). When `capture` is true, the child's -/// stdout+stderr are redirected into a single pipe and returned (tail-truncated to +/// stdout and stderr are redirected into two separate pipes; raw chunks are forwarded to +/// `event_sink` (when present) as they arrive, and a tail-truncated combined copy is +/// returned in [`ExecutionOutput`] (limited to /// [`crate::broker::executor::MAX_CAPTURED_OUTPUT_BYTES`]); otherwise no output is captured. /// /// Returns the process exit code and (when captured) its output. @@ -50,7 +78,7 @@ pub(super) fn create_process( token: &Token, command: &[String], session_id: u32, - capture: bool, + capture: OutputCapture<'_>, requires_elevation: bool, process_started: Option, cancel: Option<&CancellationToken>, @@ -64,6 +92,8 @@ pub(super) fn create_process( return Err(anyhow::Error::new(OperationCanceled)); } + let event_sink = capture.event_sink; + let capture = capture.enabled; let cmd_line = CommandLine::new(command.to_vec()); debug!(session_id, capture, "Building process creation parameters"); @@ -102,9 +132,11 @@ pub(super) fn create_process( ..Default::default() } .init(); - let (output_read, held_output_write, held_stdin_read) = if capture { + let (stdout_read, stderr_read, held_write_ends, held_stdin_read) = if capture { let (out_read, out_write) = - Pipe::new_anonymous(Some(&inheritable), 0).context("failed to create output capture pipe")?; + Pipe::new_anonymous(Some(&inheritable), 0).context("failed to create stdout capture pipe")?; + let (err_read, err_write) = + Pipe::new_anonymous(Some(&inheritable), 0).context("failed to create stderr capture pipe")?; // Empty stdin (write end closed immediately so the child reads EOF). let (in_read, in_write) = Pipe::new_anonymous(Some(&inheritable), 0).context("failed to create stdin pipe")?; drop(in_write); @@ -112,11 +144,16 @@ pub(super) fn create_process( startup_info.flags = STARTF_USESTDHANDLES | STARTF_USESHOWWINDOW; startup_info.std_input = in_read.handle.raw(); startup_info.std_output = out_write.handle.raw(); - startup_info.std_error = out_write.handle.raw(); - - (Some(out_read), Some(out_write), Some(in_read)) + startup_info.std_error = err_write.handle.raw(); + + ( + Some(out_read), + Some(err_read), + Some((out_write, err_write)), + Some(in_read), + ) } else { - (None, None, None) + (None, None, None, None) }; // `CREATE_NEW_PROCESS_GROUP` makes the child's PID a process group ID so a @@ -160,8 +197,8 @@ pub(super) fn create_process( process_started(started_at); } - // Close our copies of the child's handles so the read end observes EOF on exit. - drop(held_output_write); + // Close our copies of the child's handles so the read ends observe EOF on exit. + drop(held_write_ends); drop(held_stdin_read); info!( @@ -171,28 +208,11 @@ pub(super) fn create_process( "Process spawned, waiting for exit" ); - // Drain the pipe on a separate thread so a child producing more output than the pipe - // buffer can hold does not deadlock against our wait-for-exit. - let reader = output_read.map(|mut pipe| { - std::thread::spawn(move || { - let mut buffer = Vec::new(); - let mut chunk = [0u8; 8192]; - loop { - match pipe.read(&mut chunk) { - Ok(0) => break, - Ok(read) => { - buffer.extend_from_slice(&chunk[..read]); - if buffer.len() > MAX_CAPTURED_OUTPUT_BYTES { - let excess = buffer.len() - MAX_CAPTURED_OUTPUT_BYTES; - buffer.drain(..excess); - } - } - Err(_) => break, - } - } - buffer - }) - }); + // Drain each pipe on a separate thread so a child producing more output than a pipe + // buffer can hold does not deadlock against our wait-for-exit. Chunks are forwarded + // to the event sink as they arrive; a bounded tail is kept for diagnostics. + let stdout_reader = stdout_read.map(|pipe| spawn_output_reader(pipe, event_sink.cloned(), OutputStream::Stdout)); + let stderr_reader = stderr_read.map(|pipe| spawn_output_reader(pipe, event_sink.cloned(), OutputStream::Stderr)); let deadline = Instant::now() + OperationTracker::operation_timeout(); // INVARIANT: whenever the loop breaks, the process has exited (or was terminated), @@ -229,11 +249,10 @@ pub(super) fn create_process( } }; - // Join the reader thread on every outcome so it is never left detached. - let stdout = match reader { - Some(handle) => tail_utf8(&handle.join().unwrap_or_default()), - None => String::new(), - }; + // Join the reader threads on every outcome so they are never left detached. + let stdout_tail = stdout_reader.map(|handle| handle.join().unwrap_or_default()); + let stderr_tail = stderr_reader.map(|handle| handle.join().unwrap_or_default()); + let stdout = combined_output_tail(stdout_tail, stderr_tail); match outcome { WaitOutcome::Canceled => Err(anyhow::Error::new(OperationCanceled)), @@ -266,6 +285,60 @@ enum WaitOutcome { TimedOut, } +/// Spawn a thread draining one output pipe until EOF. +/// +/// Raw chunks are forwarded to the event sink (when present) as they arrive; the +/// thread returns a tail-bounded copy of the stream for diagnostics. +fn spawn_output_reader( + mut pipe: Pipe, + sink: Option, + stream: OutputStream, +) -> std::thread::JoinHandle> { + std::thread::spawn(move || { + let mut buffer = Vec::new(); + let mut chunk = [0u8; 8192]; + loop { + match pipe.read(&mut chunk) { + Ok(0) => break, + Ok(read) => { + if let Some(sink) = &sink { + match stream { + OutputStream::Stdout => sink.stdout(&chunk[..read]), + OutputStream::Stderr => sink.stderr(&chunk[..read]), + } + } + buffer.extend_from_slice(&chunk[..read]); + if buffer.len() > MAX_CAPTURED_OUTPUT_BYTES { + let excess = buffer.len() - MAX_CAPTURED_OUTPUT_BYTES; + buffer.drain(..excess); + } + } + Err(_) => break, + } + } + buffer + }) +} + +/// Combine the captured stdout and stderr tails into the single diagnostic string +/// kept in [`ExecutionOutput::stdout`] (used e.g. for pre-command failure notes). +fn combined_output_tail(stdout_tail: Option>, stderr_tail: Option>) -> String { + let mut combined = stdout_tail.unwrap_or_default(); + if let Some(stderr_tail) = stderr_tail + && !stderr_tail.is_empty() + { + if !combined.is_empty() && !combined.ends_with(b"\n") { + combined.push(b'\n'); + } + combined.extend_from_slice(&stderr_tail); + } + if combined.is_empty() { + String::new() + } else { + tail_utf8(&combined) + } +} + /// Resolve an executable name to its full path using the given environment's PATH. /// /// Handles both absolute paths and bare names (e.g., `winget.exe`). diff --git a/devolutions-agent/src/broker/mod.rs b/devolutions-agent/src/broker/mod.rs index d3ecf844f..a551c0a0d 100644 --- a/devolutions-agent/src/broker/mod.rs +++ b/devolutions-agent/src/broker/mod.rs @@ -6,6 +6,7 @@ pub(crate) mod auth; pub mod command_builder; pub mod evaluator; +pub mod event_channel; pub mod executor; pub mod operation_tracker; pub mod pipe; diff --git a/devolutions-agent/src/broker/operation_tracker.rs b/devolutions-agent/src/broker/operation_tracker.rs index 3dac38b5b..147c7a2df 100644 --- a/devolutions-agent/src/broker/operation_tracker.rs +++ b/devolutions-agent/src/broker/operation_tracker.rs @@ -10,10 +10,12 @@ use std::time::Duration; use anyhow::{Context as _, bail}; use chrono::{DateTime, Utc}; -use now_policy_api::{OperationStatus, PackageRequest, ResourceId}; +use now_policy_api::{EventChannel, OperationStatus, PackageRequest, ResourceId}; use sha2::{Digest as _, Sha256}; use tokio_util::sync::CancellationToken; +use crate::broker::event_channel::OperationEventSink; + /// How long completed/failed operation results are retained for status queries. const RESULT_RETENTION: Duration = Duration::from_secs(5 * 60); // 5 minutes. @@ -41,6 +43,12 @@ pub struct TrackedOperation { pub expires_at: Option>, /// Fired when cancellation is requested, observed by the executor. pub cancel_token: CancellationToken, + /// Per-operation event channel sink used to notify the client of status + /// transitions and operation completion. + pub event_sink: Option, + /// Descriptor of the per-operation event channel, returned to the client + /// (also on idempotent request resubmission). + pub event_channel: Option, } /// Thread-safe operation tracker. @@ -123,17 +131,34 @@ impl OperationTracker { owner_key: owner_key.to_owned(), expires_at: None, cancel_token: CancellationToken::new(), + event_sink: None, + event_channel: None, }, ); Ok((operation_id, true)) } + /// Attach the per-operation event channel (sink + client-facing descriptor). + /// + /// Called right after registration, before the execution task is spawned, so + /// every status transition is observed on the channel. + pub fn set_event_channel(&self, operation_id: &str, sink: OperationEventSink, descriptor: EventChannel) { + let mut state = self.state.lock().expect("tracker lock poisoned"); + if let Some(op) = state.operations.get_mut(operation_id) { + op.event_sink = Some(sink); + op.event_channel = Some(descriptor); + } + } + /// Mark an operation as Running (process launched). pub fn mark_running(&self, request_id: &str, started_at: DateTime) { let mut state = self.state.lock().expect("tracker lock poisoned"); if let Some(op) = state.operations.get_mut(request_id) { op.status = OperationStatus::Running; op.started_at = Some(started_at); + if let Some(sink) = &op.event_sink { + sink.status_updated(); + } } } @@ -158,6 +183,7 @@ impl OperationTracker { op.note = Some(note); op.completed_at = Some(now); op.expires_at = Some(now + chrono::Duration::from_std(RESULT_RETENTION).expect("valid duration")); + notify_terminal(op); } } @@ -170,6 +196,7 @@ impl OperationTracker { op.note = Some(note); op.completed_at = Some(now); op.expires_at = Some(now + chrono::Duration::from_std(RESULT_RETENTION).expect("valid duration")); + notify_terminal(op); } } @@ -182,6 +209,7 @@ impl OperationTracker { op.note = Some(note); op.completed_at = Some(now); op.expires_at = Some(now + chrono::Duration::from_std(RESULT_RETENTION).expect("valid duration")); + notify_terminal(op); } } @@ -201,6 +229,9 @@ impl OperationTracker { if !op.status.is_terminal() { op.status = OperationStatus::Canceling; op.cancel_token.cancel(); + if let Some(sink) = &op.event_sink { + sink.status_updated(); + } } Some(op.clone()) @@ -261,3 +292,11 @@ fn request_fingerprint(request: &PackageRequest) -> anyhow::Result { let bytes = serde_json::to_vec(request).context("failed to serialize package request for idempotency")?; Ok(hex::encode(Sha256::digest(bytes))) } + +/// Notify the event channel that the operation reached a terminal status. +fn notify_terminal(op: &TrackedOperation) { + if let Some(sink) = &op.event_sink { + sink.status_updated(); + sink.finish(); + } +} diff --git a/devolutions-agent/src/broker/server/mod.rs b/devolutions-agent/src/broker/server/mod.rs index ee058559e..7baf78852 100644 --- a/devolutions-agent/src/broker/server/mod.rs +++ b/devolutions-agent/src/broker/server/mod.rs @@ -253,6 +253,7 @@ impl BrokerState { scope: request.options.scope, capture_output: request.capture_output, cancel_token: tokio_util::sync::CancellationToken::new(), + event_sink: None, }; let owner_key = request.client_owner_key(); @@ -260,12 +261,45 @@ impl BrokerState { .tracker .register(&owner_key, &request, generated_operation_id) .map_err(|error| error_response(ErrorCode::Conflict, format!("{error:#}")))?; + // On idempotent resubmission, return the originally created channel descriptor. + #[cfg_attr( + not(windows), + expect(unused_mut, reason = "only mutated on Windows where event channels exist") + )] + let mut event_channel = self + .tracker + .get(&operation_id) + .and_then(|operation| operation.event_channel); if is_new_operation { // The executor observes the tracked operation's cancel token so a later // cancel request can terminate the spawned process. if let Some(tracked) = self.tracker.get(&operation_id) { context.cancel_token = tracked.cancel_token; } + + // Open the per-operation event channel before returning the response so + // the advertised pipe is immediately connectable. Best-effort: the + // operation proceeds without a channel if creation fails. + #[cfg(windows)] + { + let operation_key = operation_id.to_string(); + match crate::broker::event_channel::open_operation_channel(&operation_key, user_sid) { + Ok((sink, descriptor)) => { + self.tracker + .set_event_channel(&operation_key, sink.clone(), descriptor.clone()); + context.event_sink = Some(sink); + event_channel = Some(descriptor); + } + Err(error) => { + warn!( + operation_id = %operation_key, + error = format!("{error:#}"), + "Failed to open per-operation event channel; continuing without it" + ); + } + } + } + execution::spawn_execution( Arc::clone(&self.executor), self.tracker.clone(), @@ -283,8 +317,7 @@ impl BrokerState { operation_id, status, submitted_at, - // The per-operation event channel is not implemented yet. - event_channel: None, + event_channel, }) } else { None @@ -820,8 +853,16 @@ mod tests { } async fn submit_operation(state: &BrokerState, request: &PackageRequest) -> OperationSubmission { + // Use the real test-process user SID: the per-operation event pipe ACL only + // admits the requesting client user (plus SYSTEM/Administrators). + let user_sid = win_api_wrappers::process::Process::current_process() + .token(windows::Win32::Security::TOKEN_QUERY) + .expect("open current process token") + .sid_and_attributes() + .expect("query token user SID") + .sid; let response = state - .execute(request.clone(), &test_sid()) + .execute(request.clone(), &user_sid) .await .expect("execute request accepted"); response.operation.expect("operation submitted") @@ -955,4 +996,143 @@ mod tests { .expect_err("foreign operation is rejected"); assert_eq!(error.code, ErrorCode::NotFound); } + + // ─── Event channel ─────────────────────────────────────────────────────── + + use now_policy_api::event_channel::{ + EVENT_CHANNEL_VERSION_MAJOR, EVENT_CHANNEL_VERSION_MINOR, EventFrame, EventFrameDecoder, + }; + + /// Executor that emits output through the operation's event sink, honoring `capture_output`. + struct StreamingExecutor; + + #[async_trait] + impl CommandExecutor for StreamingExecutor { + async fn execute( + &self, + ctx: &ExecutionContext, + process_started: Option, + ) -> anyhow::Result { + if let Some(process_started) = process_started { + process_started(Utc::now()); + } + if ctx.capture_output + && let Some(sink) = &ctx.event_sink + { + sink.stdout("installing π...\n".as_bytes()); + sink.stderr(b"warning: low disk space\n"); + } + Ok(ExecutionOutput { + exit_code: 0, + stdout: String::new(), + started_at: Some(Utc::now()), + }) + } + } + + /// Connect to an operation's event pipe (bare name) and read frames until `Finish`. + async fn read_channel_frames(pipe_name: &str) -> Vec { + use tokio::io::AsyncReadExt as _; + + let path = format!(r"\\.\pipe\{pipe_name}"); + let mut client = tokio::net::windows::named_pipe::ClientOptions::new() + .write(false) + .open(&path) + .expect("connect to event channel pipe"); + + let mut decoder = EventFrameDecoder::new(); + let mut frames = Vec::new(); + let mut buffer = [0u8; 4096]; + loop { + let read = match client.read(&mut buffer).await { + Ok(0) => break, + Ok(read) => read, + Err(_) => break, + }; + decoder.extend(&buffer[..read]); + while let Some(frame) = decoder.next_frame().expect("valid frame stream") { + let finish = frame == EventFrame::Finish; + frames.push(frame); + if finish { + return frames; + } + } + } + panic!("event channel stream ended without a Finish frame"); + } + + #[tokio::test] + async fn execute_returns_connectable_event_channel_with_status_frames_only() { + let state = state_with_executor(Arc::new(InstantExecutor)); + let request = request(); // capture_output: false. + let operation = submit_operation(&state, &request).await; + + let channel = operation.event_channel.expect("event channel advertised"); + assert_eq!(channel.kind, api::EventChannelKind::LocalPipe); + assert_eq!( + channel.path, + format!("Devolutions.Now.PackageBroker.Operation.{}", operation.operation_id) + ); + + wait_for_status(&state, &request, &operation.operation_id).await; + + let frames = read_channel_frames(&channel.path).await; + assert_eq!( + frames.first(), + Some(&EventFrame::Hello { + version_major: EVENT_CHANNEL_VERSION_MAJOR, + version_minor: EVENT_CHANNEL_VERSION_MINOR, + }) + ); + assert_eq!(frames.last(), Some(&EventFrame::Finish)); + assert!( + frames[1..frames.len() - 1] + .iter() + .all(|frame| *frame == EventFrame::StatusUpdated), + "CaptureOutput=false must yield status frames only: {frames:?}" + ); + assert!( + frames[1..].contains(&EventFrame::StatusUpdated), + "at least the terminal transition must be signalled" + ); + } + + #[tokio::test] + async fn execute_streams_captured_output_over_event_channel() { + let state = state_with_executor(Arc::new(StreamingExecutor)); + let mut request = request(); + request.capture_output = true; + let operation = submit_operation(&state, &request).await; + let channel = operation.event_channel.expect("event channel advertised"); + + wait_for_status(&state, &request, &operation.operation_id).await; + + let frames = read_channel_frames(&channel.path).await; + assert_eq!( + frames, + [ + EventFrame::Hello { + version_major: EVENT_CHANNEL_VERSION_MAJOR, + version_minor: EVENT_CHANNEL_VERSION_MINOR, + }, + EventFrame::StatusUpdated, // Running. + EventFrame::Stdout("installing π...\n".to_owned()), + EventFrame::Stderr("warning: low disk space\n".to_owned()), + EventFrame::StatusUpdated, // Completed. + EventFrame::Finish, + ] + ); + } + + #[tokio::test] + async fn resubmitted_request_reuses_the_same_event_channel() { + let state = state_with_executor(Arc::new(StuckExecutor)); + let request = request(); + let first = submit_operation(&state, &request).await; + let second = submit_operation(&state, &request).await; + + assert_eq!(second.operation_id, first.operation_id); + assert_eq!(second.event_channel, first.event_channel); + assert!(first.event_channel.is_some()); + } } From 29f25bfe41d6c356c19de0b208fd91a114cdab3d Mon Sep 17 00:00:00 2001 From: Vladyslav Nikonov Date: Thu, 6 Aug 2026 22:10:10 +0300 Subject: [PATCH 2/4] refactor(agent): address event channel review feedback - Create the event pipe duplex (client still read-only via DACL) so the post-Finish drain read actually blocks until the client closes its end, instead of erroring immediately on an outbound-only pipe. - Drop the per-frame flush: named pipe writes go straight to the kernel buffer and tokio's named pipe flush is a no-op. - Take the event sink out of the tracked operation on terminal transitions so queued frames are released once the writer task is done, rather than living until result eviction. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- devolutions-agent/src/broker/event_channel.rs | 13 ++++++++++--- devolutions-agent/src/broker/operation_tracker.rs | 8 ++++++-- 2 files changed, 16 insertions(+), 5 deletions(-) diff --git a/devolutions-agent/src/broker/event_channel.rs b/devolutions-agent/src/broker/event_channel.rs index 13138699e..50aeabf5f 100644 --- a/devolutions-agent/src/broker/event_channel.rs +++ b/devolutions-agent/src/broker/event_channel.rs @@ -386,9 +386,13 @@ mod windows_channel { // `SECURITY_ATTRIBUTES` living across the call. `security_attributes` owns the // structure and its descriptor and outlives the call; `CreateNamedPipeW` copies // the descriptor and does not retain the pointer. + // + // The pipe is created duplex even though the protocol is one-way (server to + // client): the DACL grants the client read access only, so nothing can be + // written back, while the server end keeps read access so the post-`Finish` + // drain wait below can block until the client closes its end. let server = unsafe { ServerOptions::new() - .access_inbound(false) .first_pipe_instance(true) .max_instances(1) .create_with_security_attributes_raw(&pipe_path, security_attributes.as_mut_ptr().cast()) @@ -451,7 +455,9 @@ mod windows_channel { // All frames (ending with `Finish`) were handed to the pipe; give the // client a bounded amount of time to drain and close its end so buffered - // data is not discarded by our handle closing first. + // data is not discarded by our handle closing first. The client is + // restricted to read access by the DACL, so this read only ever completes + // when the client closes the pipe. let _ = tokio::time::timeout(CLIENT_DRAIN_TIMEOUT, async { let mut scratch = [0u8; 16]; loop { @@ -467,8 +473,9 @@ mod windows_channel { async fn write_frame(server: &mut NamedPipeServer, frame: &EventFrame) -> anyhow::Result<()> { let bytes = frame.encode().context("failed to encode event frame")?; + // No explicit flush: named pipe writes go straight to the kernel pipe + // buffer, and tokio's named pipe `flush` is a no-op anyway. server.write_all(&bytes).await.context("failed to write event frame")?; - server.flush().await.context("failed to flush event frame")?; Ok(()) } diff --git a/devolutions-agent/src/broker/operation_tracker.rs b/devolutions-agent/src/broker/operation_tracker.rs index 147c7a2df..10d62ad78 100644 --- a/devolutions-agent/src/broker/operation_tracker.rs +++ b/devolutions-agent/src/broker/operation_tracker.rs @@ -294,8 +294,12 @@ fn request_fingerprint(request: &PackageRequest) -> anyhow::Result { } /// Notify the event channel that the operation reached a terminal status. -fn notify_terminal(op: &TrackedOperation) { - if let Some(sink) = &op.event_sink { +/// +/// The sink is taken out of the tracked operation: terminal statuses are final, so no further +/// frames will ever be emitted, and dropping the tracker's handle lets the queued frames be +/// released as soon as the pipe writer task is done instead of living until eviction. +fn notify_terminal(op: &mut TrackedOperation) { + if let Some(sink) = op.event_sink.take() { sink.status_updated(); sink.finish(); } From 9ec8e7a801fc195b0a78b87ecbcaf4e6b992c51e Mon Sep 17 00:00:00 2001 From: Vladyslav Nikonov Date: Mon, 10 Aug 2026 19:47:41 +0300 Subject: [PATCH 3/4] refactor(agent): harden event channel for client disconnect and slow readers Raise the per-stream buffer budget to 1 MiB (stdout and stderr each) and switch overflow handling to drop-oldest semantics: when a new chunk does not fit, the oldest queued data frames of the same stream are evicted and replaced in place by an overflow frame accounting the skipped bytes (adjacent overflow markers merge), so clients always receive the most recent output. Treat a broken event pipe as normal client teardown: the writer task now abandons the queue on any write failure, logging a single debug message and turning all further sink calls into silent no-ops, releasing buffered memory immediately. The operation and the child process are never affected; final state remains available via QueryStatus. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- devolutions-agent/src/broker/event_channel.rs | 269 +++++++++++++----- 1 file changed, 201 insertions(+), 68 deletions(-) diff --git a/devolutions-agent/src/broker/event_channel.rs b/devolutions-agent/src/broker/event_channel.rs index 50aeabf5f..9995634eb 100644 --- a/devolutions-agent/src/broker/event_channel.rs +++ b/devolutions-agent/src/broker/event_channel.rs @@ -10,8 +10,9 @@ //! Writing is strictly best-effort: a client that never connects, connects late, //! reads too slowly, or disconnects early must never stall or fail the operation. //! Producers push into a bounded in-memory queue; when the per-stream byte budget -//! is exhausted, data is dropped and accounted for with `StdoutOverflow` / -//! `StderrOverflow` frames. +//! is exhausted, the oldest queued data is dropped and accounted for with +//! `StdoutOverflow` / `StderrOverflow` frames marking the gap. When the client +//! disconnects, the queue is abandoned and producers become no-ops. use std::collections::VecDeque; use std::mem; @@ -26,9 +27,10 @@ pub const OPERATION_PIPE_PREFIX: &str = "Devolutions.Now.PackageBroker.Operation /// Maximum bytes of not-yet-written output buffered per stream (stdout / stderr). /// -/// When the budget is exhausted (client absent or reading too slowly), further -/// output is dropped and reported through overflow frames. -const PER_STREAM_BUDGET_BYTES: usize = 256 * 1024; +/// When the budget is exhausted (client absent or reading too slowly), the oldest +/// queued data is dropped and reported through overflow frames, so a client always +/// receives the most recent output. +const PER_STREAM_BUDGET_BYTES: usize = 1024 * 1024; /// Bare pipe name for an operation's event channel, as returned to clients in the /// `EventChannel` descriptor (no `\\.\pipe\` prefix, matching the protocol samples). @@ -137,8 +139,6 @@ struct StreamState { chunker: Utf8StreamChunker, /// Bytes of queued (not yet consumed) data frames for this stream. queued_bytes: usize, - /// Bytes dropped since the last overflow frame was queued. - skipped_bytes: u32, } #[derive(Debug)] @@ -146,10 +146,20 @@ struct QueueState { items: VecDeque, stdout: StreamState, stderr: StreamState, - /// Set once `finish` was queued; no further items are accepted. + /// Set once `finish` was queued (or the channel was abandoned); no further + /// items are accepted. closed: bool, } +impl QueueState { + fn stream_mut(&mut self, stream: OutputStream) -> &mut StreamState { + match stream { + OutputStream::Stdout => &mut self.stdout, + OutputStream::Stderr => &mut self.stderr, + } + } +} + /// Shared bounded event queue between producers (executor, tracker) and the /// pipe writer task. #[derive(Debug)] @@ -210,30 +220,18 @@ impl OperationEventSink { /// Queue the final `Finish` frame and close the queue. /// - /// Flushes buffered incomplete UTF-8 sequences and pending overflow - /// accounting first, so `Finish` is always the last frame. + /// Flushes buffered incomplete UTF-8 sequences first, so `Finish` is always + /// the last frame. pub fn finish(&self) { let mut state = self.queue.state.lock().expect("event queue lock poisoned"); if state.closed { return; } for stream in [OutputStream::Stdout, OutputStream::Stderr] { - let stream_state = match stream { - OutputStream::Stdout => &mut state.stdout, - OutputStream::Stderr => &mut state.stderr, - }; - let tail = stream_state.chunker.flush(); + let tail = state.stream_mut(stream).chunker.flush(); if let Some(tail) = tail { Self::enqueue_chunk(&mut state, stream, tail); } - let stream_state = match stream { - OutputStream::Stdout => &mut state.stdout, - OutputStream::Stderr => &mut state.stderr, - }; - let skipped = mem::take(&mut stream_state.skipped_bytes); - if skipped > 0 { - state.items.push_back(overflow_frame(stream, skipped)); - } } state.items.push_back(EventFrame::Finish); state.closed = true; @@ -242,16 +240,25 @@ impl OperationEventSink { self.queue.notify.notify_one(); } + /// Abandon the queue: the client is gone, so buffered frames are released and + /// all further producer calls become no-ops. + fn abandon(&self) { + let mut state = self.queue.state.lock().expect("event queue lock poisoned"); + state.items.clear(); + state.stdout = StreamState::default(); + state.stderr = StreamState::default(); + state.closed = true; + drop(state); + self.queue.finished.cancel(); + self.queue.notify.notify_one(); + } + fn push_output(&self, stream: OutputStream, bytes: &[u8]) { let mut state = self.queue.state.lock().expect("event queue lock poisoned"); if state.closed { return; } - let stream_state = match stream { - OutputStream::Stdout => &mut state.stdout, - OutputStream::Stderr => &mut state.stderr, - }; - let chunks = stream_state.chunker.push(bytes); + let chunks = state.stream_mut(stream).chunker.push(bytes); let mut queued_any = false; for chunk in chunks { Self::enqueue_chunk(&mut state, stream, chunk); @@ -265,27 +272,67 @@ impl OperationEventSink { /// Queue one data chunk, respecting the per-stream byte budget. /// - /// A chunk that does not fit is dropped in full and accounted in the skipped - /// counter; once room is available again, an overflow frame is queued before - /// the next data frame so the client learns how many bytes it missed. + /// When the chunk does not fit, the *oldest* queued data frames of the same + /// stream are evicted to make room, so a slow client always receives the most + /// recent output. Each gap is marked in place with an overflow frame carrying + /// the number of skipped bytes. fn enqueue_chunk(state: &mut QueueState, stream: OutputStream, chunk: String) { - let stream_state = match stream { - OutputStream::Stdout => &mut state.stdout, - OutputStream::Stderr => &mut state.stderr, - }; - if stream_state.queued_bytes + chunk.len() > PER_STREAM_BUDGET_BYTES { - let len = u32::try_from(chunk.len()).unwrap_or(u32::MAX); - stream_state.skipped_bytes = stream_state.skipped_bytes.saturating_add(len); - return; - } - let skipped = mem::take(&mut stream_state.skipped_bytes); - stream_state.queued_bytes += chunk.len(); - if skipped > 0 { - state.items.push_back(overflow_frame(stream, skipped)); + // INVARIANT: chunk.len() <= MAX_EVENT_FRAME_BODY_BYTES <= PER_STREAM_BUDGET_BYTES, + // so evicting queued data always frees enough room. + while state.stream_mut(stream).queued_bytes + chunk.len() > PER_STREAM_BUDGET_BYTES { + if !Self::evict_oldest_data_frame(state, stream) { + // Defensive: nothing left to evict; drop the new chunk instead. + let bytes_skipped = u32::try_from(chunk.len()).unwrap_or(u32::MAX); + state.items.push_back(overflow_frame(stream, bytes_skipped)); + return; + } } + state.stream_mut(stream).queued_bytes += chunk.len(); state.items.push_back(data_frame(stream, chunk)); } + /// Evict the oldest queued data frame of `stream`, marking the gap with an + /// overflow frame (merged with an immediately preceding one when present). + /// + /// Returns false when no data frame of that stream is queued. + fn evict_oldest_data_frame(state: &mut QueueState, stream: OutputStream) -> bool { + let position = state.items.iter().position(|frame| { + matches!( + (stream, frame), + (OutputStream::Stdout, EventFrame::Stdout(_)) | (OutputStream::Stderr, EventFrame::Stderr(_)) + ) + }); + let Some(position) = position else { + return false; + }; + + let Some(evicted) = state.items.remove(position) else { + return false; + }; + let evicted_len = match &evicted { + EventFrame::Stdout(data) | EventFrame::Stderr(data) => data.len(), + // INVARIANT: `position` was found by matching a data frame of `stream`. + _ => unreachable!("evicted frame is a data frame"), + }; + state.stream_mut(stream).queued_bytes -= evicted_len; + let skipped = u32::try_from(evicted_len).unwrap_or(u32::MAX); + + // Merge into an overflow frame already sitting at the gap, if any. + let merged = position > 0 + && match (stream, state.items.get_mut(position - 1)) { + (OutputStream::Stdout, Some(EventFrame::StdoutOverflow { bytes_skipped })) + | (OutputStream::Stderr, Some(EventFrame::StderrOverflow { bytes_skipped })) => { + *bytes_skipped = bytes_skipped.saturating_add(skipped); + true + } + _ => false, + }; + if !merged { + state.items.insert(position, overflow_frame(stream, skipped)); + } + true + } + /// Wait for and take the next queued frame; `None` once the queue is closed /// and fully drained. async fn next_frame(&self) -> Option { @@ -345,7 +392,7 @@ mod windows_channel { use now_policy_api::{EventChannel, EventChannelKind}; use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _}; use tokio::net::windows::named_pipe::{NamedPipeServer, ServerOptions}; - use tracing::{debug, warn}; + use tracing::debug; use win_api_wrappers::identity::sid::Sid; use win_api_wrappers::security::acl::{Acl, ExplicitAccess, InheritableAcl, InheritableAclKind, Trustee}; use win_api_wrappers::security::attributes::SecurityAttributesInit; @@ -440,15 +487,26 @@ mod windows_channel { version_minor: EVENT_CHANNEL_VERSION_MINOR, }; if let Err(error) = write_frame(&mut server, &hello).await { - warn!(operation_id, %error, "Failed to write event channel hello frame"); + sink.abandon(); + debug!( + operation_id, + %error, + "Event channel client disconnected, continuing without streaming" + ); return; } while let Some(frame) = sink.next_frame().await { if let Err(error) = write_frame(&mut server, &frame).await { - // Best-effort: the client disconnected or stopped reading; the - // operation itself is unaffected. - debug!(operation_id, %error, "Stopped writing event channel frames"); + // Best-effort: an abrupt client disconnect is the normal teardown + // path (clients close the pipe on cancel/completion); release the + // queue and let the operation run to completion unaffected. + sink.abandon(); + debug!( + operation_id, + %error, + "Event channel client disconnected, continuing without streaming" + ); return; } } @@ -633,7 +691,7 @@ mod tests { async fn sink_accounts_overflow_when_budget_is_exhausted() { let sink = OperationEventSink::new(); let chunk = vec![b'a'; 64 * 1024]; - let pushes = 64; // 4 MiB total, way over the 256 KiB budget. + let pushes = 64; // 4 MiB total, way over the 1 MiB budget. for _ in 0..pushes { sink.stdout(&chunk); } @@ -658,35 +716,79 @@ mod tests { } #[tokio::test] - async fn sink_emits_overflow_frame_before_next_data_frame() { + async fn sink_drops_oldest_data_and_keeps_most_recent() { let sink = OperationEventSink::new(); - let big = vec![b'a'; PER_STREAM_BUDGET_BYTES]; // Fills the budget exactly (4 frames). - sink.stdout(&big); - sink.stdout(b"dropped"); - - // Drain the queued data to free budget, then push more data. - let mut drained = 0; - while drained < PER_STREAM_BUDGET_BYTES { - match sink.next_frame().await { - Some(EventFrame::Stdout(data)) => drained += data.len(), - other => panic!("unexpected frame: {other:?}"), - } - } - sink.stdout(b"fresh"); + let frame_len = 64 * 1024; + // Fill the budget exactly, then push one more chunk of a distinct byte. + let old = vec![b'o'; PER_STREAM_BUDGET_BYTES]; + sink.stdout(&old); + sink.stdout(&vec![b'n'; frame_len]); sink.finish(); + let mut frames = Vec::new(); + while let Some(frame) = sink.next_frame().await { + frames.push(frame); + } + + // The oldest frame was evicted: overflow first, then the surviving data. assert_eq!( - sink.next_frame().await, - Some(EventFrame::StdoutOverflow { bytes_skipped: 7 }) + frames.first(), + Some(&EventFrame::StdoutOverflow { + bytes_skipped: u32::try_from(frame_len).expect("frame length fits in u32"), + }) ); - assert_eq!(sink.next_frame().await, Some(EventFrame::Stdout("fresh".to_owned()))); - assert_eq!(sink.next_frame().await, Some(EventFrame::Finish)); + match frames.get(1) { + Some(EventFrame::Stdout(data)) => assert!(data.bytes().all(|b| b == b'o')), + other => panic!("unexpected frame: {other:?}"), + } + match frames.iter().rev().nth(1) { + Some(EventFrame::Stdout(data)) => { + assert!(data.bytes().all(|b| b == b'n'), "newest data must survive"); + } + other => panic!("unexpected frame: {other:?}"), + } + assert_eq!(frames.last(), Some(&EventFrame::Finish)); + } + + #[tokio::test] + async fn sink_merges_consecutive_overflow_frames() { + let sink = OperationEventSink::new(); + let frame_len = 64 * 1024; + sink.stdout(&vec![b'o'; PER_STREAM_BUDGET_BYTES]); + // Two more chunks evict two oldest frames; the gap marker must merge. + sink.stdout(&vec![b'n'; frame_len]); + sink.stdout(&vec![b'n'; frame_len]); + sink.finish(); + + let mut overflow_frames = 0; + let mut skipped = 0u64; + while let Some(frame) = sink.next_frame().await { + if let EventFrame::StdoutOverflow { bytes_skipped } = frame { + overflow_frames += 1; + skipped += u64::from(bytes_skipped); + } + } + assert_eq!(overflow_frames, 1, "consecutive gaps must merge into one frame"); + assert_eq!(skipped, 2 * frame_len as u64); + } + + #[tokio::test] + async fn sink_ignores_events_after_abandon() { + let sink = OperationEventSink::new(); + sink.stdout(b"buffered"); + sink.abandon(); + sink.stdout(b"late"); + sink.status_updated(); + sink.finish(); + assert_eq!(sink.next_frame().await, None); } } #[cfg(all(test, windows))] mod pipe_tests { + use std::time::Duration; + use now_policy_api::EventChannelKind; use now_policy_api::event_channel::{ EVENT_CHANNEL_VERSION_MAJOR, EVENT_CHANNEL_VERSION_MINOR, EventFrame, EventFrameDecoder, @@ -819,6 +921,37 @@ mod pipe_tests { ); } + #[tokio::test] + async fn abrupt_client_disconnect_does_not_affect_producers() { + let operation_id = test_operation_id("disconnect"); + let (sink, descriptor) = open_operation_channel(&operation_id, ¤t_user_sid()).expect("open channel"); + + // Connect, read a little, then abruptly drop the pipe mid-stream (the + // normal client teardown on cancel/completion). + let path = format!(r"\\.\pipe\{}", descriptor.path); + let mut client = tokio::net::windows::named_pipe::ClientOptions::new() + .write(false) + .open(&path) + .expect("connect to event channel pipe"); + sink.stdout(b"first"); + let mut buffer = [0u8; 256]; + let read = client.read(&mut buffer).await.expect("read some frames"); + assert!(read > 0); + drop(client); + + // Producers must keep working as silent no-ops; nothing may panic or block. + for _ in 0..64 { + sink.stdout(&vec![b'x'; 64 * 1024]); + sink.status_updated(); + } + sink.finish(); + + // The writer task abandons the queue once the broken pipe is observed. + tokio::time::timeout(Duration::from_secs(10), sink.finished()) + .await + .expect("sink must be closed after finish"); + } + #[tokio::test] async fn slow_reader_gets_overflow_frames_and_operation_is_unaffected() { let operation_id = test_operation_id("slow"); From 0d740a95e96b3255699a1593436dd19ba7125c33 Mon Sep 17 00:00:00 2001 From: Vladyslav Nikonov Date: Mon, 10 Aug 2026 20:26:31 +0300 Subject: [PATCH 4/4] fix(agent): bound event channel writes and overflow accounting Address review feedback on the event channel: - Apply a bounded deadline (30 s) to each frame write so a connected client that stops reading cannot pin the writer task, pipe handle, and queued data forever once the kernel pipe buffer fills; on expiry the sink is abandoned like on a disconnect. - Account dropped bytes in a per-stream pending-overflow counter instead of inserting overflow frames into the queue on every eviction. The counter is reported lazily as a single overflow frame right before the next data frame of that stream (or before Finish), so evictions strictly shrink the queue and gaps coalesce even when stdout and stderr output interleaves. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- devolutions-agent/src/broker/event_channel.rs | 127 ++++++++++++++---- 1 file changed, 100 insertions(+), 27 deletions(-) diff --git a/devolutions-agent/src/broker/event_channel.rs b/devolutions-agent/src/broker/event_channel.rs index 9995634eb..255a6480b 100644 --- a/devolutions-agent/src/broker/event_channel.rs +++ b/devolutions-agent/src/broker/event_channel.rs @@ -12,7 +12,8 @@ //! Producers push into a bounded in-memory queue; when the per-stream byte budget //! is exhausted, the oldest queued data is dropped and accounted for with //! `StdoutOverflow` / `StderrOverflow` frames marking the gap. When the client -//! disconnects, the queue is abandoned and producers become no-ops. +//! disconnects or stops reading, the queue is abandoned and producers become +//! no-ops. use std::collections::VecDeque; use std::mem; @@ -139,6 +140,12 @@ struct StreamState { chunker: Utf8StreamChunker, /// Bytes of queued (not yet consumed) data frames for this stream. queued_bytes: usize, + /// Bytes dropped by evictions that were not yet reported to the client. + /// + /// Emitted lazily as a single overflow frame right before the next data + /// frame of this stream (or before `Finish`), so evictions never grow the + /// queue and consecutive gaps coalesce regardless of stream interleaving. + pending_overflow: u64, } #[derive(Debug)] @@ -274,16 +281,17 @@ impl OperationEventSink { /// /// When the chunk does not fit, the *oldest* queued data frames of the same /// stream are evicted to make room, so a slow client always receives the most - /// recent output. Each gap is marked in place with an overflow frame carrying - /// the number of skipped bytes. + /// recent output. Dropped bytes accumulate in the stream's `pending_overflow` + /// counter and are reported lazily by `next_frame`, so evictions strictly + /// shrink the queue. fn enqueue_chunk(state: &mut QueueState, stream: OutputStream, chunk: String) { // INVARIANT: chunk.len() <= MAX_EVENT_FRAME_BODY_BYTES <= PER_STREAM_BUDGET_BYTES, // so evicting queued data always frees enough room. while state.stream_mut(stream).queued_bytes + chunk.len() > PER_STREAM_BUDGET_BYTES { if !Self::evict_oldest_data_frame(state, stream) { // Defensive: nothing left to evict; drop the new chunk instead. - let bytes_skipped = u32::try_from(chunk.len()).unwrap_or(u32::MAX); - state.items.push_back(overflow_frame(stream, bytes_skipped)); + let stream_state = state.stream_mut(stream); + stream_state.pending_overflow = stream_state.pending_overflow.saturating_add(chunk.len() as u64); return; } } @@ -291,8 +299,8 @@ impl OperationEventSink { state.items.push_back(data_frame(stream, chunk)); } - /// Evict the oldest queued data frame of `stream`, marking the gap with an - /// overflow frame (merged with an immediately preceding one when present). + /// Evict the oldest queued data frame of `stream`, accounting the dropped + /// bytes in the stream's `pending_overflow` counter. /// /// Returns false when no data frame of that stream is queued. fn evict_oldest_data_frame(state: &mut QueueState, stream: OutputStream) -> bool { @@ -314,25 +322,23 @@ impl OperationEventSink { // INVARIANT: `position` was found by matching a data frame of `stream`. _ => unreachable!("evicted frame is a data frame"), }; - state.stream_mut(stream).queued_bytes -= evicted_len; - let skipped = u32::try_from(evicted_len).unwrap_or(u32::MAX); - - // Merge into an overflow frame already sitting at the gap, if any. - let merged = position > 0 - && match (stream, state.items.get_mut(position - 1)) { - (OutputStream::Stdout, Some(EventFrame::StdoutOverflow { bytes_skipped })) - | (OutputStream::Stderr, Some(EventFrame::StderrOverflow { bytes_skipped })) => { - *bytes_skipped = bytes_skipped.saturating_add(skipped); - true - } - _ => false, - }; - if !merged { - state.items.insert(position, overflow_frame(stream, skipped)); - } + let stream_state = state.stream_mut(stream); + stream_state.queued_bytes -= evicted_len; + stream_state.pending_overflow = stream_state.pending_overflow.saturating_add(evicted_len as u64); true } + /// Take the pending overflow of `stream` as a protocol frame, if any. + fn take_pending_overflow(state: &mut QueueState, stream: OutputStream) -> Option { + let stream_state = state.stream_mut(stream); + if stream_state.pending_overflow == 0 { + return None; + } + let bytes_skipped = u32::try_from(stream_state.pending_overflow).unwrap_or(u32::MAX); + stream_state.pending_overflow = stream_state.pending_overflow.saturating_sub(u64::from(bytes_skipped)); + Some(overflow_frame(stream, bytes_skipped)) + } + /// Wait for and take the next queued frame; `None` once the queue is closed /// and fully drained. async fn next_frame(&self) -> Option { @@ -340,6 +346,20 @@ impl OperationEventSink { let notified = self.queue.notify.notified(); { let mut state = self.queue.state.lock().expect("event queue lock poisoned"); + // Report accumulated gaps right before the data they precede + // (or before `Finish`, when no data of that stream survived). + let overflow = match state.items.front() { + Some(EventFrame::Stdout(_)) => Self::take_pending_overflow(&mut state, OutputStream::Stdout), + Some(EventFrame::Stderr(_)) => Self::take_pending_overflow(&mut state, OutputStream::Stderr), + Some(EventFrame::Finish) => Self::take_pending_overflow(&mut state, OutputStream::Stdout) + .or_else(|| Self::take_pending_overflow(&mut state, OutputStream::Stderr)), + _ => None, + }; + if let Some(frame) = overflow { + // The frame at the queue front is left in place for the next call. + self.queue.notify.notify_one(); + return Some(frame); + } if let Some(frame) = state.items.pop_front() { match &frame { EventFrame::Stdout(data) => state.stdout.queued_bytes -= data.len(), @@ -411,6 +431,11 @@ mod windows_channel { /// client to drain the pipe and close its end before tearing the pipe down. const CLIENT_DRAIN_TIMEOUT: Duration = Duration::from_secs(30); + /// How long a single frame write may remain pending before the client is + /// considered stalled and the channel is abandoned. Writes only block once + /// the kernel pipe buffer is full, i.e. the client stopped reading. + const WRITE_STALL_TIMEOUT: Duration = Duration::from_secs(30); + /// Create the event channel for an operation. /// /// The named pipe instance is created (with its security descriptor) before this @@ -491,7 +516,7 @@ mod windows_channel { debug!( operation_id, %error, - "Event channel client disconnected, continuing without streaming" + "Event channel client disconnected or stalled, continuing without streaming" ); return; } @@ -505,7 +530,7 @@ mod windows_channel { debug!( operation_id, %error, - "Event channel client disconnected, continuing without streaming" + "Event channel client disconnected or stalled, continuing without streaming" ); return; } @@ -529,11 +554,19 @@ mod windows_channel { debug!(operation_id, "Event channel closed"); } + /// Write one frame with a bounded deadline. + /// + /// A connected client that stops reading would otherwise leave the write + /// pending forever once the kernel pipe buffer fills, retaining the writer + /// task, pipe handle, and queued data indefinitely. async fn write_frame(server: &mut NamedPipeServer, frame: &EventFrame) -> anyhow::Result<()> { let bytes = frame.encode().context("failed to encode event frame")?; // No explicit flush: named pipe writes go straight to the kernel pipe // buffer, and tokio's named pipe `flush` is a no-op anyway. - server.write_all(&bytes).await.context("failed to write event frame")?; + tokio::time::timeout(WRITE_STALL_TIMEOUT, server.write_all(&bytes)) + .await + .map_err(|_| anyhow::anyhow!("client stopped reading (write pending for {WRITE_STALL_TIMEOUT:?})"))? + .context("failed to write event frame")?; Ok(()) } @@ -755,7 +788,8 @@ mod tests { let sink = OperationEventSink::new(); let frame_len = 64 * 1024; sink.stdout(&vec![b'o'; PER_STREAM_BUDGET_BYTES]); - // Two more chunks evict two oldest frames; the gap marker must merge. + // Two more chunks evict two oldest frames; the gaps must coalesce into a + // single overflow report. sink.stdout(&vec![b'n'; frame_len]); sink.stdout(&vec![b'n'; frame_len]); sink.finish(); @@ -772,6 +806,45 @@ mod tests { assert_eq!(skipped, 2 * frame_len as u64); } + #[tokio::test] + async fn sink_bounds_overflow_frames_with_interleaved_streams() { + let sink = OperationEventSink::new(); + let frame_len = 64 * 1024; + // Fill both stream budgets, then keep alternating: every push evicts, and + // the queue must not grow by an overflow frame per eviction. + let evictions_per_stream = 32; + sink.stdout(&vec![b'a'; PER_STREAM_BUDGET_BYTES]); + sink.stderr(&vec![b'b'; PER_STREAM_BUDGET_BYTES]); + for _ in 0..evictions_per_stream { + sink.stdout(&vec![b'a'; frame_len]); + sink.stderr(&vec![b'b'; frame_len]); + } + sink.finish(); + + let mut stdout_overflow_frames = 0u32; + let mut stderr_overflow_frames = 0u32; + let mut stdout_skipped = 0u64; + let mut stderr_skipped = 0u64; + while let Some(frame) = sink.next_frame().await { + match frame { + EventFrame::StdoutOverflow { bytes_skipped } => { + stdout_overflow_frames += 1; + stdout_skipped += u64::from(bytes_skipped); + } + EventFrame::StderrOverflow { bytes_skipped } => { + stderr_overflow_frames += 1; + stderr_skipped += u64::from(bytes_skipped); + } + _ => {} + } + } + + assert_eq!(stdout_overflow_frames, 1, "interleaved gaps must still coalesce"); + assert_eq!(stderr_overflow_frames, 1, "interleaved gaps must still coalesce"); + assert_eq!(stdout_skipped, (evictions_per_stream * frame_len) as u64); + assert_eq!(stderr_skipped, (evictions_per_stream * frame_len) as u64); + } + #[tokio::test] async fn sink_ignores_events_after_abandon() { let sink = OperationEventSink::new();