From 2d361f90fde00dc0ddcee35cd2bfe919bbc931f8 Mon Sep 17 00:00:00 2001 From: BatmanByte <300328404+BatmanByte@users.noreply.github.com> Date: Sat, 25 Jul 2026 12:44:04 +0800 Subject: [PATCH 01/12] feat(exec): drain guest output before attach Start draining an execution's stdout and stderr as soon as its guest-side state is created. Retain a bounded one-megabyte replay buffer so commands are not blocked when no client attaches, and report overwritten output explicitly when a client attaches late. The dropped control event carries stdout and stderr byte counts so the host only resets the decoder that lost bytes. This keeps an intact stream from receiving a spurious replacement character. --- src/boxlite/src/portal/interfaces/exec.rs | 56 ++++++ src/boxlite/tests/run_main_command.rs | 86 +++++++++ src/guest/src/service/exec/mod.rs | 1 + src/guest/src/service/exec/output.rs | 225 ++++++++++++++++++++++ src/guest/src/service/exec/state.rs | 156 +++++---------- src/shared/proto/boxlite/v1/service.proto | 6 + 6 files changed, 418 insertions(+), 112 deletions(-) create mode 100644 src/guest/src/service/exec/output.rs diff --git a/src/boxlite/src/portal/interfaces/exec.rs b/src/boxlite/src/portal/interfaces/exec.rs index b04135b9d..51cd0c92f 100644 --- a/src/boxlite/src/portal/interfaces/exec.rs +++ b/src/boxlite/src/portal/interfaces/exec.rs @@ -414,6 +414,23 @@ impl ExecProtocol { tracing::trace!(len = chunk.data.len(), "Received exec stderr"); stderr.send_bytes(chunk.data); } + Some(exec_output::Event::Dropped(dropped)) => { + tracing::warn!( + stdout_bytes = dropped.stdout_bytes, + stderr_bytes = dropped.stderr_bytes, + "Guest output buffer dropped older output" + ); + if dropped.stdout_bytes > 0 { + stdout.flush(); + } + if dropped.stderr_bytes > 0 { + stderr.flush(); + } + let _ = stderr.tx.send(format!( + "[boxlite] output dropped (stdout: {} bytes, stderr: {} bytes)\n", + dropped.stdout_bytes, dropped.stderr_bytes + )); + } None => {} } } @@ -1167,6 +1184,45 @@ mod tests { assert!(stderr_rx.try_recv().is_err()); } + #[test] + fn output_drop_only_flushes_the_affected_utf8_decoder() { + use boxlite_shared::{OutputDropped, Stderr as StderrMsg, exec_output}; + + let (stdout_tx, mut stdout_rx) = mpsc::unbounded_channel::(); + let (stderr_tx, mut stderr_rx) = mpsc::unbounded_channel::(); + let mut stdout = DecodedStream::new(stdout_tx); + let mut stderr = DecodedStream::new(stderr_tx); + + let output = |event| ExecOutput { event: Some(event) }; + ExecProtocol::route_output( + output(exec_output::Event::Stderr(StderrMsg { data: vec![0xE2] })), + &mut stdout, + &mut stderr, + ); + ExecProtocol::route_output( + output(exec_output::Event::Dropped(OutputDropped { + stdout_bytes: 1, + stderr_bytes: 0, + })), + &mut stdout, + &mut stderr, + ); + ExecProtocol::route_output( + output(exec_output::Event::Stderr(StderrMsg { + data: vec![0x94, 0x80], + })), + &mut stdout, + &mut stderr, + ); + + assert!(stdout_rx.try_recv().is_err()); + assert_eq!( + stderr_rx.try_recv().ok(), + Some("[boxlite] output dropped (stdout: 1 bytes, stderr: 0 bytes)\n".to_string()) + ); + assert_eq!(stderr_rx.try_recv().ok(), Some("─".to_string())); + } + /// Flushing a DecodedStream must drain held-over bytes, leave the /// decoder in a valid drained state, and be idempotent. The attach loop /// flushes both streams on every exit path (clean EOF, transport error, diff --git a/src/boxlite/tests/run_main_command.rs b/src/boxlite/tests/run_main_command.rs index 2654c8706..71ab5fd3e 100644 --- a/src/boxlite/tests/run_main_command.rs +++ b/src/boxlite/tests/run_main_command.rs @@ -63,6 +63,92 @@ async fn attached_stdout(opts: BoxOptions) -> String { stdout } +#[tokio::test] +async fn main_command_exits_after_large_output_without_attach() { + let home = boxlite_test_utils::home::PerTestBoxHome::new(); + let runtime = boxlite::BoxliteRuntime::new(boxlite::runtime::options::BoxliteOptions { + home_dir: home.path.clone(), + image_registries: common::test_registries(), + }) + .expect("create runtime"); + + let handle = runtime + .create( + main_command_opts(&["sh", "-c", "head -c 1048576 /dev/zero; exit 23"], false), + None, + ) + .await + .expect("create box"); + + let completed = tokio::time::timeout(std::time::Duration::from_secs(30), async { + handle.start().await.expect("start box"); + loop { + if handle.info().status == boxlite::BoxStatus::Stopped { + return true; + } + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + } + }) + .await + .unwrap_or(false); + + let _ = handle.stop().await; + let _ = runtime.remove(handle.id().as_str(), true).await; + let _ = runtime.shutdown(Some(common::TEST_SHUTDOWN_TIMEOUT)).await; + + assert!( + completed, + "a main command that has no Attach consumer must still drain and exit" + ); +} + +#[tokio::test] +async fn late_attach_reports_output_dropped() { + let home = boxlite_test_utils::home::PerTestBoxHome::new(); + let runtime = boxlite::BoxliteRuntime::new(boxlite::runtime::options::BoxliteOptions { + home_dir: home.path.clone(), + image_registries: common::test_registries(), + }) + .expect("create runtime"); + + let handle = runtime + .create( + main_command_opts(&["sh", "-c", "head -c 2097152 /dev/zero; sleep 30"], false), + None, + ) + .await + .expect("create box"); + + handle.start().await.expect("start box"); + tokio::time::sleep(std::time::Duration::from_secs(1)).await; + + let mut execution = handle + .attach(None) + .await + .expect("attach to the main command"); + let mut stderr = execution.stderr().expect("stderr stream"); + let dropped = tokio::time::timeout(std::time::Duration::from_secs(5), async { + while let Some(chunk) = stderr.next().await { + if chunk.contains("[boxlite] output dropped") { + return Some(chunk); + } + } + None + }) + .await + .unwrap_or(None); + + let _ = handle.stop().await; + let _ = runtime.remove(handle.id().as_str(), true).await; + let _ = runtime.shutdown(Some(common::TEST_SHUTDOWN_TIMEOUT)).await; + + let dropped = dropped.expect("late attach must report overwritten output"); + assert!( + dropped.contains("stdout:") && dropped.contains("stderr: 0 bytes"), + "stdout-only loss must preserve the stderr decoder: {dropped:?}" + ); +} + /// `tty: true` must give the *main command* a real terminal. /// /// `test -t 0` asks the kernel, so this cannot pass unless init's fd 0 really diff --git a/src/guest/src/service/exec/mod.rs b/src/guest/src/service/exec/mod.rs index aa5b0c040..29119c348 100644 --- a/src/guest/src/service/exec/mod.rs +++ b/src/guest/src/service/exec/mod.rs @@ -20,6 +20,7 @@ pub(in crate::service) mod error; #[cfg(target_os = "linux")] pub mod exec_handle; pub(in crate::service) mod executor; +mod output; pub(in crate::service) mod registry; pub(in crate::service) mod state; mod timeout; diff --git a/src/guest/src/service/exec/output.rs b/src/guest/src/service/exec/output.rs new file mode 100644 index 000000000..2de51035c --- /dev/null +++ b/src/guest/src/service/exec/output.rs @@ -0,0 +1,225 @@ +use crate::service::exec::exec_handle::{ExecStderr, ExecStdout}; +use async_stream::stream; +use boxlite_shared::{exec_output, ExecOutput, OutputDropped, Stderr, Stdout}; +use futures::{Stream, StreamExt}; +use std::collections::VecDeque; +use std::pin::Pin; +use std::sync::Arc; +use tokio::sync::{watch, Mutex}; +use tonic::Status; + +const BUFFER_CAPACITY_BYTES: usize = 1024 * 1024; + +pub(crate) type OutputStream = Pin> + Send>>; + +#[derive(Clone)] +pub(crate) struct OutputManager { + inner: Arc>, + updated: watch::Sender<()>, +} + +struct OutputState { + entries: VecDeque, + buffered_bytes: usize, + oldest_sequence: u64, + next_sequence: u64, + pending_dropped: DroppedBytes, + open_readers: usize, + attached: bool, +} + +struct OutputEntry { + sequence: u64, + output: ExecOutput, + source: OutputSource, + byte_len: usize, +} + +#[derive(Clone, Copy)] +enum OutputSource { + Stdout, + Stderr, +} + +#[derive(Default)] +struct DroppedBytes { + stdout: u64, + stderr: u64, +} + +impl DroppedBytes { + fn record(&mut self, source: OutputSource, byte_len: usize) { + let byte_len = byte_len as u64; + match source { + OutputSource::Stdout => self.stdout += byte_len, + OutputSource::Stderr => self.stderr += byte_len, + } + } + + fn take(&mut self) -> Self { + std::mem::take(self) + } +} + +impl OutputManager { + pub(crate) fn new(stdout: Option, stderr: Option) -> Self { + let open_readers = usize::from(stdout.is_some()) + usize::from(stderr.is_some()); + let (updated, _) = watch::channel(()); + let manager = Self { + inner: Arc::new(Mutex::new(OutputState { + entries: VecDeque::new(), + buffered_bytes: 0, + oldest_sequence: 0, + next_sequence: 0, + pending_dropped: DroppedBytes::default(), + open_readers, + attached: false, + })), + updated, + }; + + if let Some(stdout) = stdout { + manager.spawn_stdout(stdout); + } + if let Some(stderr) = stderr { + manager.spawn_stderr(stderr); + } + + manager + } + + pub(crate) async fn attach(&self) -> Result { + { + let mut state = self.inner.lock().await; + if state.attached { + return Err(Status::already_exists("Already attached")); + } + state.attached = true; + } + + let manager = self.clone(); + let output = stream! { + let mut next_sequence = 0; + let mut updates = manager.updated.subscribe(); + + loop { + enum Next { + Item(ExecOutput), + Wait, + Done, + } + + let next = { + let mut state = manager.inner.lock().await; + + if next_sequence < state.oldest_sequence { + next_sequence = state.oldest_sequence; + Next::Item(dropped_output(state.pending_dropped.take())) + } else if next_sequence < state.next_sequence { + let index = (next_sequence - state.oldest_sequence) as usize; + let entry = state.entries.get(index).expect("ring sequence must exist"); + next_sequence += 1; + Next::Item(entry.output.clone()) + } else if state.open_readers == 0 { + Next::Done + } else { + Next::Wait + } + }; + + match next { + Next::Item(item) => yield Ok(item), + Next::Done => break, + Next::Wait => { + if updates.changed().await.is_err() { + break; + } + } + } + } + }; + + Ok(Box::pin(output)) + } + + fn spawn_stdout(&self, stdout: ExecStdout) { + let manager = self.clone(); + tokio::spawn(async move { + manager.drain(stdout, OutputSource::Stdout).await; + }); + } + + fn spawn_stderr(&self, stderr: ExecStderr) { + let manager = self.clone(); + tokio::spawn(async move { + manager.drain(stderr, OutputSource::Stderr).await; + }); + } + + async fn drain(&self, mut stream: S, source: OutputSource) + where + S: Stream> + Unpin, + { + while let Some(data) = stream.next().await { + self.push(source, data).await; + } + self.reader_finished().await; + } + + async fn push(&self, source: OutputSource, data: Vec) { + let byte_len = data.len(); + let output = match source { + OutputSource::Stdout => ExecOutput { + event: Some(exec_output::Event::Stdout(Stdout { data })), + }, + OutputSource::Stderr => ExecOutput { + event: Some(exec_output::Event::Stderr(Stderr { data })), + }, + }; + + let mut state = self.inner.lock().await; + let sequence = state.next_sequence; + state.next_sequence += 1; + + while state.buffered_bytes + byte_len > BUFFER_CAPACITY_BYTES { + let Some(removed) = state.entries.pop_front() else { + state.pending_dropped.record(source, byte_len); + state.oldest_sequence = state.next_sequence; + break; + }; + state.buffered_bytes -= removed.byte_len; + state + .pending_dropped + .record(removed.source, removed.byte_len); + state.oldest_sequence = removed.sequence + 1; + } + + if byte_len <= BUFFER_CAPACITY_BYTES { + state.buffered_bytes += byte_len; + state.entries.push_back(OutputEntry { + sequence, + output, + source, + byte_len, + }); + } + drop(state); + self.updated.send_replace(()); + } + + async fn reader_finished(&self) { + let mut state = self.inner.lock().await; + state.open_readers -= 1; + drop(state); + self.updated.send_replace(()); + } +} + +fn dropped_output(dropped: DroppedBytes) -> ExecOutput { + ExecOutput { + event: Some(exec_output::Event::Dropped(OutputDropped { + stdout_bytes: dropped.stdout, + stderr_bytes: dropped.stderr, + })), + } +} diff --git a/src/guest/src/service/exec/state.rs b/src/guest/src/service/exec/state.rs index 36977e68a..58a88acfe 100644 --- a/src/guest/src/service/exec/state.rs +++ b/src/guest/src/service/exec/state.rs @@ -1,12 +1,12 @@ use crate::service::exec::error::ExecutionError; use crate::service::exec::exec_handle::ExecHandle; +use crate::service::exec::output::OutputManager; use boxlite_shared::ExecOutput; -use futures::Stream; +use futures::{Stream, StreamExt as _}; use std::os::unix::io::AsRawFd; use std::sync::Arc; use tokio::sync::{mpsc, Mutex}; use tokio::task::{AbortHandle, JoinHandle}; -use tracing::info; /// Abstraction for checking container init health. /// @@ -26,6 +26,7 @@ pub(crate) trait InitHealthCheck: Send + Sync { struct Inner { /// The process handle (owns pid, pty_controller, stdin, stdout, stderr) handle: Option, + output: OutputManager, /// Abort handles for stdin forwarding tasks that may own the taken stdin FD. input_tasks: Vec, /// Stdout/stderr forwarding tasks (set on attach) @@ -54,7 +55,6 @@ pub(crate) struct ExecutionExit { /// Execution state. /// /// Handle owns pid, pty_controller, stdin, stdout, stderr. -/// stdin is taken on send_input(), stdout/stderr are taken on attach(). #[derive(Clone)] pub(crate) struct ExecutionState { inner: Arc>, @@ -65,26 +65,28 @@ pub(crate) struct ExecutionState { } impl ExecutionState { - fn from_inner(inner: Inner, exit: crate::reaper::ExitSlot) -> Self { + fn from_handle( + mut handle: ExecHandle, + init_health: Option>>, + exit: crate::reaper::ExitSlot, + ) -> Self { Self { - inner: Arc::new(Mutex::new(inner)), - exit, - } - } - - /// Create new execution state for a guest-side process. - pub(super) fn new(handle: ExecHandle, exit: crate::reaper::ExitSlot) -> Self { - Self::from_inner( - Inner { + inner: Arc::new(Mutex::new(Inner { + output: OutputManager::new(handle.stdout(), handle.stderr()), handle: Some(handle), input_tasks: Vec::new(), output_tasks: Vec::new(), released: false, timed_out: false, - init_health: None, - }, + init_health, + })), exit, - ) + } + } + + /// Create new execution state for a guest-side process. + pub(super) fn new(handle: ExecHandle, exit: crate::reaper::ExitSlot) -> Self { + Self::from_handle(handle, None, exit) } #[cfg(test)] @@ -104,17 +106,7 @@ impl ExecutionState { init_health: Arc>, exit: crate::reaper::ExitSlot, ) -> Self { - Self::from_inner( - Inner { - handle: Some(handle), - input_tasks: Vec::new(), - output_tasks: Vec::new(), - released: false, - timed_out: false, - init_health: Some(init_health), - }, - exit, - ) + Self::from_handle(handle, Some(init_health), exit) } /// Create execution state for the container's init process itself. @@ -123,17 +115,7 @@ impl ExecutionState { /// reparents to guest main (the boxlite-guest agent process), which owns /// `waitpid(-1)`. See `wait_process`. pub(crate) fn new_init_session(handle: ExecHandle, exit: crate::reaper::ExitSlot) -> Self { - Self::from_inner( - Inner { - handle: Some(handle), - input_tasks: Vec::new(), - output_tasks: Vec::new(), - released: false, - timed_out: false, - init_health: None, - }, - exit, - ) + Self::from_handle(handle, None, exit) } /// Check if the container init process died. @@ -288,88 +270,38 @@ impl ExecutionState { } /// Attach to execution output. - /// - /// Takes stdout/stderr from handle and starts forwarding tasks. - /// Returns stream of output chunks. pub async fn attach( &self, - exec_id: &str, + _exec_id: &str, ) -> Result, ExecutionError> { - use boxlite_shared::{exec_output, Stderr, Stdout}; - use futures::StreamExt; - - let (tx, rx) = mpsc::channel(100); - - // Take stdout/stderr from handle - let (stdout, stderr) = { - let mut inner = self.inner.lock().await; - - if !inner.output_tasks.is_empty() { - return Err(ExecutionError::AlreadyAttached); + let output = { + let inner = self.inner.lock().await; + if inner.released { + return Err(ExecutionError::HandleUnavailable); } - - let handle = inner - .handle - .as_mut() - .ok_or(ExecutionError::HandleUnavailable)?; - - let stdout = handle.stdout(); - let stderr = handle.stderr(); - - (stdout, stderr) + inner.output.clone() }; - - // Spawn forwarding tasks - let mut tasks = Vec::new(); - - // Spawn stdout forwarding task - let exec_id_string = exec_id.to_string(); - if let Some(mut stdout) = stdout { - let tx = tx.clone(); - let handle = tokio::spawn(async move { - while let Some(chunk) = stdout.next().await { - let msg = ExecOutput { - event: Some(exec_output::Event::Stdout(Stdout { data: chunk })), - }; - if tx.send(msg).await.is_err() { - break; - } - } - info!(execution = ?exec_id_string, "Stdout forwarding task ended"); - }); - tasks.push(handle); - } - - // Spawn stderr forwarding task - let exec_id_string = exec_id.to_string(); - if let Some(mut stderr) = stderr { - let tx = tx.clone(); - let handle = tokio::spawn(async move { - while let Some(chunk) = stderr.next().await { - let msg = ExecOutput { - event: Some(exec_output::Event::Stderr(Stderr { data: chunk })), - }; - if tx.send(msg).await.is_err() { - break; - } - } - info!(execution = ?exec_id_string, "Stderr forwarding task ended"); - }); - tasks.push(handle); - } - - // Store tasks - { - let mut inner = self.inner.lock().await; - if inner.released { - for task in tasks { - task.abort(); + let mut output = output + .attach() + .await + .map_err(|_| ExecutionError::AlreadyAttached)?; + let (tx, rx) = mpsc::channel(100); + let task = tokio::spawn(async move { + while let Some(message) = output.next().await { + match message { + Ok(message) if tx.send(message).await.is_err() => break, + Ok(_) => {} + Err(_) => break, } - } else { - inner.output_tasks = tasks; } - } + }); + let mut inner = self.inner.lock().await; + if inner.released { + task.abort(); + return Err(ExecutionError::HandleUnavailable); + } + inner.output_tasks.push(task); Ok(rx) } diff --git a/src/shared/proto/boxlite/v1/service.proto b/src/shared/proto/boxlite/v1/service.proto index 39c682ea2..5acb22eb8 100644 --- a/src/shared/proto/boxlite/v1/service.proto +++ b/src/shared/proto/boxlite/v1/service.proto @@ -447,9 +447,15 @@ message ExecOutput { oneof event { Stdout stdout = 1; Stderr stderr = 2; + OutputDropped dropped = 3; } } +message OutputDropped { + uint64 stdout_bytes = 1; + uint64 stderr_bytes = 2; +} + message Stdout { bytes data = 1; } From d336396757ecee448ae6033e7ae98c52d89bb2d7 Mon Sep 17 00:00:00 2001 From: BatmanByte <300328404+BatmanByte@users.noreply.github.com> Date: Sun, 26 Jul 2026 18:12:17 +0800 Subject: [PATCH 02/12] fix(exec): use reactor-backed stdio --- src/guest/src/container/command.rs | 9 +- src/guest/src/container/lifecycle.rs | 2 +- src/guest/src/service/exec/exec_handle.rs | 152 ++++++++++++++++------ src/guest/src/service/exec/executor.rs | 6 +- 4 files changed, 120 insertions(+), 49 deletions(-) diff --git a/src/guest/src/container/command.rs b/src/guest/src/container/command.rs index 7cd2b62a7..b664df39c 100644 --- a/src/guest/src/container/command.rs +++ b/src/guest/src/container/command.rs @@ -307,12 +307,7 @@ impl ContainerCommand { tracing::debug!(pid = pid.as_raw(), "spawned with pipes"); // Non-PTY mode: stdout and stderr are separate pipes - Ok(ExecHandle::new( - pid, - stdin_write, - stdout_read, - Some(stderr_read), - )) + ExecHandle::new(pid, stdin_write, stdout_read, Some(stderr_read)) } /// Build phase of PTY spawn: zygote IPC only, no console-socket handshake. @@ -515,7 +510,7 @@ pub(crate) fn create_pty_child( let (stdin, stdout) = reconcile_pty_fds(&pty_master)?; // PTY mode: stderr is None (merged into stdout) - let mut child = ExecHandle::new(pid, stdin, stdout, None); + let mut child = ExecHandle::new(pid, stdin, stdout, None)?; let pty_controller = pty_master_to_file(pty_master); child.set_pty(pty_controller, config); diff --git a/src/guest/src/container/lifecycle.rs b/src/guest/src/container/lifecycle.rs index c2a7bb251..46d07eb4c 100644 --- a/src/guest/src/container/lifecycle.rs +++ b/src/guest/src/container/lifecycle.rs @@ -339,7 +339,7 @@ impl Container { stdin, stdout, stderr, - } => ExecHandle::new(pid, stdin, stdout, Some(stderr)), + } => ExecHandle::new(pid, stdin, stdout, Some(stderr))?, // Mirrors the tenant PTY path: the master becomes stdin+stdout and // is retained for window-size ioctls, so ResizeTty reaches the main // command exactly as it reaches an exec. diff --git a/src/guest/src/service/exec/exec_handle.rs b/src/guest/src/service/exec/exec_handle.rs index 775f9de2d..35cddc644 100644 --- a/src/guest/src/service/exec/exec_handle.rs +++ b/src/guest/src/service/exec/exec_handle.rs @@ -7,26 +7,25 @@ use boxlite_shared::errors::{BoxliteError, BoxliteResult}; use futures::stream::{Stream, StreamExt}; use nix::sys::signal::Signal; use nix::unistd::Pid; -use std::os::unix::io::OwnedFd; +use std::io; +use std::os::fd::{AsRawFd, OwnedFd}; use std::pin::Pin; use std::task::{Context, Poll}; -use tokio::io::AsyncWriteExt; +use tokio::io::{unix::AsyncFd, Interest}; /// Stdin writer for executed process /// /// Async wrapper around file descriptor for writing to process stdin. pub struct ExecStdin { - inner: tokio::fs::File, + inner: AsyncFd, } impl ExecStdin { /// Create from file descriptor - pub fn new(fd: OwnedFd) -> Self { - use std::os::fd::{FromRawFd, IntoRawFd}; - let std_file = unsafe { std::fs::File::from_raw_fd(fd.into_raw_fd()) }; - Self { - inner: tokio::fs::File::from_std(std_file), - } + pub fn new(fd: OwnedFd) -> BoxliteResult { + Ok(Self { + inner: async_fd(fd, "stdin")?, + }) } /// Write all data to stdin @@ -35,34 +34,69 @@ impl ExecStdin { /// /// - I/O error (pipe closed, etc.) pub async fn write_all(&mut self, data: &[u8]) -> BoxliteResult<()> { - self.inner - .write_all(data) - .await - .map_err(|e| BoxliteError::Internal(format!("Failed to write to stdin: {}", e))) + let mut written = 0; + while written < data.len() { + let count = self + .inner + .async_io(Interest::WRITABLE, |fd| write_fd(fd, &data[written..])) + .await + .map_err(|error| { + BoxliteError::Internal(format!("Failed to write to stdin: {error}")) + })?; + if count == 0 { + return Err(BoxliteError::Internal( + "Failed to write to stdin: write returned zero bytes".into(), + )); + } + written += count; + } + Ok(()) } } +fn async_fd(fd: OwnedFd, stream_name: &str) -> BoxliteResult> { + set_nonblocking(&fd)?; + AsyncFd::new(fd).map_err(|error| { + BoxliteError::Internal(format!( + "Failed to register {stream_name} with Tokio I/O reactor: {error}" + )) + }) +} + +fn set_nonblocking(fd: &OwnedFd) -> BoxliteResult<()> { + let flags = nix::fcntl::fcntl(fd.as_raw_fd(), nix::fcntl::FcntlArg::F_GETFL) + .map_err(|error| BoxliteError::Internal(format!("Failed to read fd flags: {error}")))?; + let mut flags = nix::fcntl::OFlag::from_bits_truncate(flags); + flags.insert(nix::fcntl::OFlag::O_NONBLOCK); + nix::fcntl::fcntl(fd.as_raw_fd(), nix::fcntl::FcntlArg::F_SETFL(flags)).map_err(|error| { + BoxliteError::Internal(format!("Failed to set fd non-blocking: {error}")) + })?; + Ok(()) +} + +fn read_fd(fd: &OwnedFd, buffer: &mut [u8]) -> io::Result { + nix::unistd::read(fd.as_raw_fd(), buffer).map_err(Into::into) +} + +fn write_fd(fd: &OwnedFd, buffer: &[u8]) -> io::Result { + nix::unistd::write(fd, buffer).map_err(Into::into) +} + // Shared output stream implementation struct OutputStream { inner: Pin> + Send>>, } impl OutputStream { - fn new(fd: OwnedFd) -> Self { + fn new(fd: OwnedFd) -> BoxliteResult { use async_stream::stream; - use std::os::fd::{FromRawFd, IntoRawFd}; - use tokio::io::AsyncReadExt; - // Convert OwnedFd to tokio file - let std_file = unsafe { std::fs::File::from_raw_fd(fd.into_raw_fd()) }; - let file = tokio::fs::File::from_std(std_file); - let mut reader = tokio::io::BufReader::new(file); + let reader = async_fd(fd, "output")?; - // Read chunks as they arrive (works for both PTY and pipes) let stream = stream! { let mut buf = [0u8; 1024]; loop { - match reader.read(&mut buf).await { + match reader.async_io(Interest::READABLE, |fd| read_fd(fd, &mut buf)).await { Ok(0) => break, // EOF Ok(n) => yield buf[..n].to_vec(), Err(_) => break, @@ -70,9 +104,9 @@ impl OutputStream { } }; - Self { + Ok(Self { inner: Box::pin(stream), - } + }) } } @@ -93,10 +127,10 @@ pub struct ExecStdout { impl ExecStdout { /// Create from file descriptor - pub fn new(fd: OwnedFd) -> Self { - Self { - inner: OutputStream::new(fd), - } + pub fn new(fd: OwnedFd) -> BoxliteResult { + Ok(Self { + inner: OutputStream::new(fd)?, + }) } } @@ -117,10 +151,10 @@ pub struct ExecStderr { impl ExecStderr { /// Create from file descriptor - pub fn new(fd: OwnedFd) -> Self { - Self { - inner: OutputStream::new(fd), - } + pub fn new(fd: OwnedFd) -> BoxliteResult { + Ok(Self { + inner: OutputStream::new(fd)?, + }) } } @@ -265,17 +299,22 @@ impl ExecHandle { /// * `stdin` - Stdin file descriptor /// * `stdout` - Stdout file descriptor /// * `stderr` - Stderr file descriptor, or `None` in PTY mode (merged into stdout) - pub fn new(pid: Pid, stdin: OwnedFd, stdout: OwnedFd, stderr: Option) -> Self { - Self { + pub fn new( + pid: Pid, + stdin: OwnedFd, + stdout: OwnedFd, + stderr: Option, + ) -> BoxliteResult { + Ok(Self { pid, - stdin: Some(ExecStdin::new(stdin)), - stdout: Some(ExecStdout::new(stdout)), + stdin: Some(ExecStdin::new(stdin)?), + stdout: Some(ExecStdout::new(stdout)?), // In PTY mode, stderr is None because stdout/stderr are merged // at the PTY level (single reader from PTY master) - stderr: stderr.map(ExecStderr::new), + stderr: stderr.map(ExecStderr::new).transpose()?, pty_controller: None, pty_config: None, - } + }) } /// Set PTY controller and config @@ -504,3 +543,40 @@ mod process_group_tests { ); } } + +#[cfg(test)] +mod tests { + use super::*; + use futures::StreamExt; + use std::time::Duration; + + #[test] + fn output_read_does_not_occupy_blocking_pool() { + let runtime = tokio::runtime::Builder::new_multi_thread() + .worker_threads(1) + .max_blocking_threads(1) + .enable_all() + .build() + .unwrap(); + let (read_fd, _write_fd) = nix::unistd::pipe().unwrap(); + let mut stdout = ExecStdout::new(read_fd).unwrap(); + + runtime.block_on(async { + let output_read = tokio::spawn(async move { stdout.next().await }); + tokio::time::sleep(Duration::from_millis(50)).await; + + let temporary_file = tempfile::NamedTempFile::new().unwrap(); + std::fs::write(temporary_file.path(), b"ready").unwrap(); + let read_file = tokio::fs::read(temporary_file.path()); + + assert!( + tokio::time::timeout(Duration::from_millis(100), read_file) + .await + .is_ok(), + "an idle output stream must not occupy a blocking-pool worker" + ); + + output_read.abort(); + }); + } +} diff --git a/src/guest/src/service/exec/executor.rs b/src/guest/src/service/exec/executor.rs index b4640baf9..9beaba65f 100644 --- a/src/guest/src/service/exec/executor.rs +++ b/src/guest/src/service/exec/executor.rs @@ -168,12 +168,12 @@ fn spawn_with_pipes(req: &ExecRequest) -> BoxliteResult { let pid = child.id(); // Non-PTY mode: stdout and stderr are separate pipes - Ok(ExecHandle::new( + ExecHandle::new( Pid::from_raw(pid as i32), stdin_write, stdout_read, Some(stderr_read), - )) + ) } /// Spawn process with PTY (interactive mode). @@ -265,7 +265,7 @@ fn spawn_with_pty(req: &ExecRequest, config: PtyConfig) -> BoxliteResult Date: Mon, 27 Jul 2026 13:20:30 +0800 Subject: [PATCH 03/12] fix(exec): address output drain review Track the stream cursor separately from ring retention so OutputDropped only reports bytes that have not been handed to the Attach stream. Clean up and reap a spawned process whenever its I/O handle setup or PTY handoff fails, including the container-backed paths. Replace the late-attach test's timing sleep with a guest-side completion marker. The marker is created only after the main command's pipe writes have completed, making the overflow assertion independent of host scheduling. --- src/boxlite/tests/run_main_command.rs | 36 +++++++++++- src/guest/src/container/command.rs | 48 ++++++++++++++-- src/guest/src/service/exec/exec_handle.rs | 5 +- src/guest/src/service/exec/executor.rs | 67 ++++++++++++++++++++--- src/guest/src/service/exec/output.rs | 63 +++++++++++++++++++-- 5 files changed, 192 insertions(+), 27 deletions(-) diff --git a/src/boxlite/tests/run_main_command.rs b/src/boxlite/tests/run_main_command.rs index 71ab5fd3e..458ca34d5 100644 --- a/src/boxlite/tests/run_main_command.rs +++ b/src/boxlite/tests/run_main_command.rs @@ -8,7 +8,7 @@ mod common; -use boxlite::{BoxOptions, RootfsSpec}; +use boxlite::{BoxCommand, BoxOptions, LiteBox, RootfsSpec}; use tokio_stream::StreamExt; /// Create a box whose main command is `cmd`, optionally on a PTY. @@ -63,6 +63,29 @@ async fn attached_stdout(opts: BoxOptions) -> String { stdout } +async fn wait_for_file(handle: &LiteBox, path: &str) { + tokio::time::timeout(std::time::Duration::from_secs(30), async { + loop { + let execution = handle + .exec(BoxCommand::new("test").args(["-e", path])) + .await + .expect("check main command marker"); + if execution + .wait() + .await + .expect("wait for marker check") + .exit_code + == 0 + { + break; + } + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + } + }) + .await + .expect("main command must reach marker"); +} + #[tokio::test] async fn main_command_exits_after_large_output_without_attach() { let home = boxlite_test_utils::home::PerTestBoxHome::new(); @@ -113,14 +136,21 @@ async fn late_attach_reports_output_dropped() { let handle = runtime .create( - main_command_opts(&["sh", "-c", "head -c 2097152 /dev/zero; sleep 30"], false), + main_command_opts( + &[ + "sh", + "-c", + "head -c 2097152 /dev/zero; touch /tmp/main-output-ready; sleep 30", + ], + false, + ), None, ) .await .expect("create box"); handle.start().await.expect("start box"); - tokio::time::sleep(std::time::Duration::from_secs(1)).await; + wait_for_file(&handle, "/tmp/main-output-ready").await; let mut execution = handle .attach(None) diff --git a/src/guest/src/container/command.rs b/src/guest/src/container/command.rs index b664df39c..58926a0db 100644 --- a/src/guest/src/container/command.rs +++ b/src/guest/src/container/command.rs @@ -307,7 +307,13 @@ impl ContainerCommand { tracing::debug!(pid = pid.as_raw(), "spawned with pipes"); // Non-PTY mode: stdout and stderr are separate pipes - ExecHandle::new(pid, stdin_write, stdout_read, Some(stderr_read)) + match ExecHandle::new(pid, stdin_write, stdout_read, Some(stderr_read)) { + Ok(handle) => Ok(handle), + Err(error) => { + terminate_process(pid); + Err(error) + } + } } /// Build phase of PTY spawn: zygote IPC only, no console-socket handshake. @@ -488,7 +494,13 @@ impl SpawnedPty { /// Call this AFTER releasing the container mutex — the handshake /// does not need serialization. pub(crate) fn finish(self) -> BoxliteResult { - let pty_master = self.socket.receive_pty_master()?; + let pty_master = match self.socket.receive_pty_master() { + Ok(pty_master) => pty_master, + Err(error) => { + terminate_process(self.pid); + return Err(error); + } + }; create_pty_child(self.pid, pty_master, self.config) } } @@ -505,18 +517,42 @@ pub(crate) fn create_pty_child( pty_master: OwnedFd, config: PtyConfig, ) -> BoxliteResult { - set_pty_window_size(&pty_master, &config)?; - crate::service::exec::tty::apply_modes(&pty_master, &config.modes)?; - let (stdin, stdout) = reconcile_pty_fds(&pty_master)?; + if let Err(error) = set_pty_window_size(&pty_master, &config) { + terminate_process(pid); + return Err(error); + } + if let Err(error) = crate::service::exec::tty::apply_modes(&pty_master, &config.modes) { + terminate_process(pid); + return Err(error); + } + let (stdin, stdout) = match reconcile_pty_fds(&pty_master) { + Ok(fds) => fds, + Err(error) => { + terminate_process(pid); + return Err(error); + } + }; // PTY mode: stderr is None (merged into stdout) - let mut child = ExecHandle::new(pid, stdin, stdout, None)?; + let mut child = match ExecHandle::new(pid, stdin, stdout, None) { + Ok(handle) => handle, + Err(error) => { + terminate_process(pid); + return Err(error); + } + }; let pty_controller = pty_master_to_file(pty_master); child.set_pty(pty_controller, config); Ok(child) } +fn terminate_process(pid: Pid) { + let _fence = crate::reaper::reap_fence(); + let _ = nix::sys::signal::kill(pid, nix::sys::signal::Signal::SIGKILL); + let _ = nix::sys::wait::waitpid(pid, None); +} + /// Set PTY terminal window size via ioctl. fn set_pty_window_size(pty_master: &OwnedFd, config: &PtyConfig) -> BoxliteResult<()> { use nix::pty::Winsize; diff --git a/src/guest/src/service/exec/exec_handle.rs b/src/guest/src/service/exec/exec_handle.rs index 35cddc644..29af33d49 100644 --- a/src/guest/src/service/exec/exec_handle.rs +++ b/src/guest/src/service/exec/exec_handle.rs @@ -558,10 +558,9 @@ mod tests { .enable_all() .build() .unwrap(); - let (read_fd, _write_fd) = nix::unistd::pipe().unwrap(); - let mut stdout = ExecStdout::new(read_fd).unwrap(); - runtime.block_on(async { + let (read_fd, _write_fd) = nix::unistd::pipe().unwrap(); + let mut stdout = ExecStdout::new(read_fd).unwrap(); let output_read = tokio::spawn(async move { stdout.next().await }); tokio::time::sleep(Duration::from_millis(50)).await; diff --git a/src/guest/src/service/exec/executor.rs b/src/guest/src/service/exec/executor.rs index 9beaba65f..f87932107 100644 --- a/src/guest/src/service/exec/executor.rs +++ b/src/guest/src/service/exec/executor.rs @@ -161,19 +161,25 @@ fn spawn_with_pipes(req: &ExecRequest) -> BoxliteResult { cmd.stderr(std::process::Stdio::from_raw_fd(stderr_write.into_raw_fd())); } - let child = cmd + let mut child = cmd .spawn() .map_err(|e| BoxliteError::Internal(format!("Failed to spawn '{}': {}", req.program, e)))?; let pid = child.id(); // Non-PTY mode: stdout and stderr are separate pipes - ExecHandle::new( + match ExecHandle::new( Pid::from_raw(pid as i32), stdin_write, stdout_read, Some(stderr_read), - ) + ) { + Ok(handle) => Ok(handle), + Err(error) => { + terminate_child(&mut child); + Err(error) + } + } } /// Spawn process with PTY (interactive mode). @@ -242,7 +248,7 @@ fn spawn_with_pty(req: &ExecRequest, config: PtyConfig) -> BoxliteResult BoxliteResult fd, + Err(error) => { + terminate_child(&mut child); + return Err(BoxliteError::Internal(format!( + "Failed to dup PTY for stdin: {error}" + ))); + } + }; + let stdout_fd = match dup(master.as_raw_fd()) { + Ok(fd) => fd, + Err(error) => { + terminate_child(&mut child); + return Err(BoxliteError::Internal(format!( + "Failed to dup PTY for stdout: {error}" + ))); + } + }; let stdin = unsafe { OwnedFd::from_raw_fd(stdin_fd) }; let stdout = unsafe { OwnedFd::from_raw_fd(stdout_fd) }; // PTY mode: stderr is None (merged into stdout) - let mut handle = ExecHandle::new(Pid::from_raw(pid as i32), stdin, stdout, None)?; + let mut handle = match ExecHandle::new(Pid::from_raw(pid as i32), stdin, stdout, None) { + Ok(handle) => handle, + Err(error) => { + terminate_child(&mut child); + return Err(error); + } + }; // Keep master FD for resize operations let pty_controller = { @@ -277,3 +303,26 @@ fn spawn_with_pty(req: &ExecRequest, config: PtyConfig) -> BoxliteResult, } struct OutputEntry { @@ -74,6 +75,7 @@ impl OutputManager { pending_dropped: DroppedBytes::default(), open_readers, attached: false, + attachment_next_sequence: None, })), updated, }; @@ -95,6 +97,7 @@ impl OutputManager { return Err(Status::already_exists("Already attached")); } state.attached = true; + state.attachment_next_sequence = Some(0); } let manager = self.clone(); @@ -114,12 +117,19 @@ impl OutputManager { if next_sequence < state.oldest_sequence { next_sequence = state.oldest_sequence; + state.attachment_next_sequence = Some(next_sequence); Next::Item(dropped_output(state.pending_dropped.take())) } else if next_sequence < state.next_sequence { let index = (next_sequence - state.oldest_sequence) as usize; - let entry = state.entries.get(index).expect("ring sequence must exist"); + let output = state + .entries + .get(index) + .expect("ring sequence must exist") + .output + .clone(); next_sequence += 1; - Next::Item(entry.output.clone()) + state.attachment_next_sequence = Some(next_sequence); + Next::Item(output) } else if state.open_readers == 0 { Next::Done } else { @@ -183,14 +193,18 @@ impl OutputManager { while state.buffered_bytes + byte_len > BUFFER_CAPACITY_BYTES { let Some(removed) = state.entries.pop_front() else { - state.pending_dropped.record(source, byte_len); + if state.entry_is_unread(sequence) { + state.pending_dropped.record(source, byte_len); + } state.oldest_sequence = state.next_sequence; break; }; state.buffered_bytes -= removed.byte_len; - state - .pending_dropped - .record(removed.source, removed.byte_len); + if state.entry_is_unread(removed.sequence) { + state + .pending_dropped + .record(removed.source, removed.byte_len); + } state.oldest_sequence = removed.sequence + 1; } @@ -215,6 +229,13 @@ impl OutputManager { } } +impl OutputState { + fn entry_is_unread(&self, sequence: u64) -> bool { + self.attachment_next_sequence + .is_none_or(|next_sequence| sequence >= next_sequence) + } +} + fn dropped_output(dropped: DroppedBytes) -> ExecOutput { ExecOutput { event: Some(exec_output::Event::Dropped(OutputDropped { @@ -223,3 +244,33 @@ fn dropped_output(dropped: DroppedBytes) -> ExecOutput { })), } } + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn dropped_counts_only_entries_the_attachment_has_not_received() { + let manager = OutputManager::new(None, None); + manager + .push(OutputSource::Stdout, b"already sent".to_vec()) + .await; + + let mut output = manager.attach().await.unwrap(); + assert!(matches!( + output.next().await.unwrap().unwrap().event, + Some(exec_output::Event::Stdout(_)) + )); + + for _ in 0..=BUFFER_CAPACITY_BYTES / 1024 { + manager.push(OutputSource::Stderr, vec![0; 1024]).await; + } + + let dropped = output.next().await.unwrap().unwrap(); + let Some(exec_output::Event::Dropped(dropped)) = dropped.event else { + panic!("the unread stderr must be reported as dropped"); + }; + assert_eq!(dropped.stdout_bytes, 0); + assert!(dropped.stderr_bytes > 0); + } +} From 9eeffa258547c276b74d42a9dae560a4f357b4db Mon Sep 17 00:00:00 2001 From: BatmanByte <300328404+BatmanByte@users.noreply.github.com> Date: Mon, 27 Jul 2026 15:22:25 +0800 Subject: [PATCH 04/12] test(exec): synchronize output reader probe Replace the timing-based wait in the AsyncFd regression test with a signal emitted after the output stream first registers pending. The blocking-pool probe now starts only after the reader has been polled, avoiding scheduler-dependent false passes. --- src/guest/src/service/exec/exec_handle.rs | 18 ++++++++++++++++-- 1 file changed, 16 insertions(+), 2 deletions(-) diff --git a/src/guest/src/service/exec/exec_handle.rs b/src/guest/src/service/exec/exec_handle.rs index 29af33d49..4347212b1 100644 --- a/src/guest/src/service/exec/exec_handle.rs +++ b/src/guest/src/service/exec/exec_handle.rs @@ -548,6 +548,7 @@ mod process_group_tests { mod tests { use super::*; use futures::StreamExt; + use std::task::Poll; use std::time::Duration; #[test] @@ -561,8 +562,21 @@ mod tests { runtime.block_on(async { let (read_fd, _write_fd) = nix::unistd::pipe().unwrap(); let mut stdout = ExecStdout::new(read_fd).unwrap(); - let output_read = tokio::spawn(async move { stdout.next().await }); - tokio::time::sleep(Duration::from_millis(50)).await; + let (pending_tx, pending_rx) = tokio::sync::oneshot::channel(); + let output_read = tokio::spawn(async move { + let mut pending_tx = Some(pending_tx); + futures::future::poll_fn(move |cx| { + let next = Pin::new(&mut stdout).poll_next(cx); + if matches!(next, Poll::Pending) { + if let Some(tx) = pending_tx.take() { + let _ = tx.send(()); + } + } + next + }) + .await + }); + pending_rx.await.unwrap(); let temporary_file = tempfile::NamedTempFile::new().unwrap(); std::fs::write(temporary_file.path(), b"ready").unwrap(); From 5991c1f34e808841f340cccad1c773e089952d31 Mon Sep 17 00:00:00 2001 From: BatmanByte <300328404+BatmanByte@users.noreply.github.com> Date: Mon, 27 Jul 2026 16:29:16 +0800 Subject: [PATCH 05/12] fix(ci): remove unused test import --- src/guest/src/service/exec/exec_handle.rs | 1 - 1 file changed, 1 deletion(-) diff --git a/src/guest/src/service/exec/exec_handle.rs b/src/guest/src/service/exec/exec_handle.rs index 4347212b1..714e081e0 100644 --- a/src/guest/src/service/exec/exec_handle.rs +++ b/src/guest/src/service/exec/exec_handle.rs @@ -547,7 +547,6 @@ mod process_group_tests { #[cfg(test)] mod tests { use super::*; - use futures::StreamExt; use std::task::Poll; use std::time::Duration; From 3cbb59e3d4dbcc4337582ee37ec4c22bad5d174b Mon Sep 17 00:00:00 2001 From: BatmanByte <300328404+BatmanByte@users.noreply.github.com> Date: Mon, 27 Jul 2026 17:33:40 +0800 Subject: [PATCH 06/12] refactor(exec): name attach response stream --- src/guest/src/service/exec/output.rs | 4 ++-- src/guest/src/service/exec/state.rs | 1 + 2 files changed, 3 insertions(+), 2 deletions(-) diff --git a/src/guest/src/service/exec/output.rs b/src/guest/src/service/exec/output.rs index 663fb0732..18b777e7b 100644 --- a/src/guest/src/service/exec/output.rs +++ b/src/guest/src/service/exec/output.rs @@ -10,7 +10,7 @@ use tonic::Status; const BUFFER_CAPACITY_BYTES: usize = 1024 * 1024; -pub(crate) type OutputStream = Pin> + Send>>; +pub(crate) type AttachStream = Pin> + Send>>; #[derive(Clone)] pub(crate) struct OutputManager { @@ -90,7 +90,7 @@ impl OutputManager { manager } - pub(crate) async fn attach(&self) -> Result { + pub(crate) async fn attach(&self) -> Result { { let mut state = self.inner.lock().await; if state.attached { diff --git a/src/guest/src/service/exec/state.rs b/src/guest/src/service/exec/state.rs index 58a88acfe..9b88a5185 100644 --- a/src/guest/src/service/exec/state.rs +++ b/src/guest/src/service/exec/state.rs @@ -1,5 +1,6 @@ use crate::service::exec::error::ExecutionError; use crate::service::exec::exec_handle::ExecHandle; +<<<<<<< HEAD use crate::service::exec::output::OutputManager; use boxlite_shared::ExecOutput; use futures::{Stream, StreamExt as _}; From dc0f396a1869dae0946699fd97a35bfa83ac2364 Mon Sep 17 00:00:00 2001 From: BatmanByte <300328404+BatmanByte@users.noreply.github.com> Date: Wed, 29 Jul 2026 19:48:47 +0800 Subject: [PATCH 07/12] fix(exec): make output gaps stream-local New guests report per-stream offsets and final byte counts so Attach can detect both mid-stream and terminal output loss. Keep the former dropped event as a deprecated decode-only wire variant for hosts attaching to existing guests.\n\nEmit each stream's terminal frame once its own final entry has been sent, rather than waiting for its sibling pipe. Also reap an init if its stdio cannot register with Tokio's reactor, matching the other spawn paths. --- src/boxlite/src/portal/interfaces/exec.rs | 283 +++++++++++++++++++--- src/boxlite/tests/run_main_command.rs | 8 +- src/guest/src/container/command.rs | 2 +- src/guest/src/container/lifecycle.rs | 65 ++++- src/guest/src/service/exec/output.rs | 240 ++++++++++++------ src/shared/proto/boxlite/v1/service.proto | 8 +- 6 files changed, 493 insertions(+), 113 deletions(-) diff --git a/src/boxlite/src/portal/interfaces/exec.rs b/src/boxlite/src/portal/interfaces/exec.rs index 51cd0c92f..15a97f299 100644 --- a/src/boxlite/src/portal/interfaces/exec.rs +++ b/src/boxlite/src/portal/interfaces/exec.rs @@ -337,8 +337,8 @@ impl ExecProtocol { // cut, doubling visible columns and desyncing TUI cursor // math (see https://github.com/.../issues/...). Holding // the trailing partial across chunks fixes this. - let mut stdout = DecodedStream::new(stdout_tx); - let mut stderr = DecodedStream::new(stderr_tx); + let mut stdout = OutputTracker::new(stdout_tx); + let mut stderr = OutputTracker::new(stderr_tx); loop { // Use select! to handle cancellation while streaming @@ -375,7 +375,8 @@ impl ExecProtocol { // synthesized "Attach stream error: …" line. stdout.flush(); stderr.flush(); - let _ = stderr.tx.send(format!("Attach stream error: {}", e)); + let _ = + stderr.stream.tx.send(format!("Attach stream error: {}", e)); break; } None => { @@ -404,37 +405,69 @@ impl ExecProtocol { }); } - fn route_output(output: ExecOutput, stdout: &mut DecodedStream, stderr: &mut DecodedStream) { + fn route_output(output: ExecOutput, stdout: &mut OutputTracker, stderr: &mut OutputTracker) { match output.event { Some(exec_output::Event::Stdout(chunk)) => { tracing::trace!(len = chunk.data.len(), "Received exec stdout"); - stdout.send_bytes(chunk.data); + if let Some(lost_bytes) = + stdout.receive(chunk.offset, chunk.data, chunk.total_bytes) + { + Self::report_gap(stderr, "stdout", lost_bytes); + } } Some(exec_output::Event::Stderr(chunk)) => { tracing::trace!(len = chunk.data.len(), "Received exec stderr"); - stderr.send_bytes(chunk.data); + if let Some(lost_bytes) = + stderr.receive(chunk.offset, chunk.data, chunk.total_bytes) + { + Self::report_gap(stderr, "stderr", lost_bytes); + } } Some(exec_output::Event::Dropped(dropped)) => { - tracing::warn!( - stdout_bytes = dropped.stdout_bytes, - stderr_bytes = dropped.stderr_bytes, - "Guest output buffer dropped older output" + Self::report_legacy_drop( + stdout, + stderr, + dropped.stdout_bytes, + dropped.stderr_bytes, ); - if dropped.stdout_bytes > 0 { - stdout.flush(); - } - if dropped.stderr_bytes > 0 { - stderr.flush(); - } - let _ = stderr.tx.send(format!( - "[boxlite] output dropped (stdout: {} bytes, stderr: {} bytes)\n", - dropped.stdout_bytes, dropped.stderr_bytes - )); } None => {} } } + fn report_gap(stderr: &mut OutputTracker, source: &str, lost_bytes: u64) { + tracing::warn!( + source, + lost_bytes, + "Guest output buffer dropped older output" + ); + let _ = stderr.stream.tx.send(format!( + "[boxlite] {source} output dropped {lost_bytes} bytes\n" + )); + } + + fn report_legacy_drop( + stdout: &mut OutputTracker, + stderr: &mut OutputTracker, + stdout_bytes: u64, + stderr_bytes: u64, + ) { + tracing::warn!( + stdout_bytes, + stderr_bytes, + "Legacy guest output buffer dropped older output" + ); + if stdout_bytes > 0 { + stdout.flush(); + } + if stderr_bytes > 0 { + stderr.flush(); + } + let _ = stderr.stream.tx.send(format!( + "[boxlite] output dropped (stdout: {stdout_bytes} bytes, stderr: {stderr_bytes} bytes)\n" + )); + } + fn spawn_wait( mut client: ExecutionClient, execution_id: String, @@ -736,6 +769,78 @@ impl DecodedStream { } } +struct OutputTracker { + expected_offset: u64, + stream: DecodedStream, +} + +impl OutputTracker { + fn new(tx: mpsc::UnboundedSender) -> Self { + Self { + expected_offset: 0, + stream: DecodedStream::new(tx), + } + } + + fn receive( + &mut self, + offset: Option, + data: Vec, + total_bytes: Option, + ) -> Option { + if let Some(total_bytes) = total_bytes { + let Some(offset) = offset else { + tracing::warn!(total_bytes, "Exec output end frame has no offset"); + return None; + }; + if !data.is_empty() || offset != total_bytes { + tracing::warn!(offset, total_bytes, "Invalid exec output end frame"); + return None; + } + if total_bytes < self.expected_offset { + tracing::warn!( + expected_offset = self.expected_offset, + total_bytes, + "Exec output end frame moved backwards" + ); + return None; + } + + let lost_bytes = total_bytes - self.expected_offset; + self.expected_offset = total_bytes; + self.stream.flush(); + return (lost_bytes > 0).then_some(lost_bytes); + } + + let Some(offset) = offset else { + self.expected_offset += data.len() as u64; + self.stream.send_bytes(data); + return None; + }; + + if offset < self.expected_offset { + tracing::warn!( + expected_offset = self.expected_offset, + offset, + "Exec output chunk moved backwards" + ); + return None; + } + + let lost_bytes = offset - self.expected_offset; + if lost_bytes > 0 { + self.stream.flush(); + } + self.expected_offset = offset + data.len() as u64; + self.stream.send_bytes(data); + (lost_bytes > 0).then_some(lost_bytes) + } + + fn flush(&mut self) { + self.stream.flush(); + } +} + // ============================================================================ // UNIT TESTS // ============================================================================ @@ -1163,16 +1268,20 @@ mod tests { let (stdout_tx, mut stdout_rx) = mpsc::unbounded_channel::(); let (stderr_tx, mut stderr_rx) = mpsc::unbounded_channel::(); - let mut stdout = DecodedStream::new(stdout_tx); - let mut stderr = DecodedStream::new(stderr_tx); - - let mk_stdout = |bytes: Vec| ExecOutput { - event: Some(exec_output::Event::Stdout(StdoutMsg { data: bytes })), + let mut stdout = OutputTracker::new(stdout_tx); + let mut stderr = OutputTracker::new(stderr_tx); + + let mk_stdout = |offset, bytes: Vec| ExecOutput { + event: Some(exec_output::Event::Stdout(StdoutMsg { + data: bytes, + offset: Some(offset), + total_bytes: None, + })), }; // "─" split into [E2] and [94 80] across two messages. - ExecProtocol::route_output(mk_stdout(vec![0xE2]), &mut stdout, &mut stderr); - ExecProtocol::route_output(mk_stdout(vec![0x94, 0x80]), &mut stdout, &mut stderr); + ExecProtocol::route_output(mk_stdout(0, vec![0xE2]), &mut stdout, &mut stderr); + ExecProtocol::route_output(mk_stdout(1, vec![0x94, 0x80]), &mut stdout, &mut stderr); // First message: holdover only, no emission. // Second message: complete "─" emitted. @@ -1185,17 +1294,67 @@ mod tests { } #[test] - fn output_drop_only_flushes_the_affected_utf8_decoder() { + fn output_gap_only_flushes_the_affected_utf8_decoder() { + use boxlite_shared::{Stderr as StderrMsg, Stdout as StdoutMsg, exec_output}; + + let (stdout_tx, mut stdout_rx) = mpsc::unbounded_channel::(); + let (stderr_tx, mut stderr_rx) = mpsc::unbounded_channel::(); + let mut stdout = OutputTracker::new(stdout_tx); + let mut stderr = OutputTracker::new(stderr_tx); + + let output = |event| ExecOutput { event: Some(event) }; + ExecProtocol::route_output( + output(exec_output::Event::Stderr(StderrMsg { + data: vec![0xE2], + offset: Some(0), + total_bytes: None, + })), + &mut stdout, + &mut stderr, + ); + ExecProtocol::route_output( + output(exec_output::Event::Stdout(StdoutMsg { + data: b"after-gap".to_vec(), + offset: Some(1), + total_bytes: None, + })), + &mut stdout, + &mut stderr, + ); + ExecProtocol::route_output( + output(exec_output::Event::Stderr(StderrMsg { + data: vec![0x94, 0x80], + offset: Some(1), + total_bytes: None, + })), + &mut stdout, + &mut stderr, + ); + + assert_eq!(stdout_rx.try_recv().ok(), Some("after-gap".to_string())); + assert_eq!( + stderr_rx.try_recv().ok(), + Some("[boxlite] stdout output dropped 1 bytes\n".to_string()) + ); + assert_eq!(stderr_rx.try_recv().ok(), Some("─".to_string())); + } + + #[test] + fn legacy_drop_only_flushes_the_affected_utf8_decoder() { use boxlite_shared::{OutputDropped, Stderr as StderrMsg, exec_output}; let (stdout_tx, mut stdout_rx) = mpsc::unbounded_channel::(); let (stderr_tx, mut stderr_rx) = mpsc::unbounded_channel::(); - let mut stdout = DecodedStream::new(stdout_tx); - let mut stderr = DecodedStream::new(stderr_tx); + let mut stdout = OutputTracker::new(stdout_tx); + let mut stderr = OutputTracker::new(stderr_tx); let output = |event| ExecOutput { event: Some(event) }; ExecProtocol::route_output( - output(exec_output::Event::Stderr(StderrMsg { data: vec![0xE2] })), + output(exec_output::Event::Stderr(StderrMsg { + data: vec![0xE2], + offset: None, + total_bytes: None, + })), &mut stdout, &mut stderr, ); @@ -1210,6 +1369,8 @@ mod tests { ExecProtocol::route_output( output(exec_output::Event::Stderr(StderrMsg { data: vec![0x94, 0x80], + offset: None, + total_bytes: None, })), &mut stdout, &mut stderr, @@ -1223,6 +1384,66 @@ mod tests { assert_eq!(stderr_rx.try_recv().ok(), Some("─".to_string())); } + #[test] + fn output_end_reports_a_missing_tail() { + use boxlite_shared::{Stdout as StdoutMsg, exec_output}; + + let (stdout_tx, mut stdout_rx) = mpsc::unbounded_channel::(); + let (stderr_tx, mut stderr_rx) = mpsc::unbounded_channel::(); + let mut stdout = OutputTracker::new(stdout_tx); + let mut stderr = OutputTracker::new(stderr_tx); + + let output = |event| ExecOutput { event: Some(event) }; + ExecProtocol::route_output( + output(exec_output::Event::Stdout(StdoutMsg { + data: b"hello".to_vec(), + offset: Some(0), + total_bytes: None, + })), + &mut stdout, + &mut stderr, + ); + ExecProtocol::route_output( + output(exec_output::Event::Stdout(StdoutMsg { + data: Vec::new(), + offset: Some(10), + total_bytes: Some(10), + })), + &mut stdout, + &mut stderr, + ); + + assert_eq!(stdout_rx.try_recv().ok(), Some("hello".to_string())); + assert_eq!( + stderr_rx.try_recv().ok(), + Some("[boxlite] stdout output dropped 5 bytes\n".to_string()) + ); + } + + #[test] + fn legacy_output_without_offsets_remains_contiguous() { + use boxlite_shared::{Stdout as StdoutMsg, exec_output}; + + let (stdout_tx, mut stdout_rx) = mpsc::unbounded_channel::(); + let (stderr_tx, mut stderr_rx) = mpsc::unbounded_channel::(); + let mut stdout = OutputTracker::new(stdout_tx); + let mut stderr = OutputTracker::new(stderr_tx); + + let output = |data| ExecOutput { + event: Some(exec_output::Event::Stdout(StdoutMsg { + data, + offset: None, + total_bytes: None, + })), + }; + ExecProtocol::route_output(output(b"hello ".to_vec()), &mut stdout, &mut stderr); + ExecProtocol::route_output(output(b"world".to_vec()), &mut stdout, &mut stderr); + + assert_eq!(stdout_rx.try_recv().ok(), Some("hello ".to_string())); + assert_eq!(stdout_rx.try_recv().ok(), Some("world".to_string())); + assert!(stderr_rx.try_recv().is_err()); + } + /// Flushing a DecodedStream must drain held-over bytes, leave the /// decoder in a valid drained state, and be idempotent. The attach loop /// flushes both streams on every exit path (clean EOF, transport error, diff --git a/src/boxlite/tests/run_main_command.rs b/src/boxlite/tests/run_main_command.rs index 458ca34d5..4a79e6592 100644 --- a/src/boxlite/tests/run_main_command.rs +++ b/src/boxlite/tests/run_main_command.rs @@ -126,7 +126,7 @@ async fn main_command_exits_after_large_output_without_attach() { } #[tokio::test] -async fn late_attach_reports_output_dropped() { +async fn late_attach_reports_output_gap() { let home = boxlite_test_utils::home::PerTestBoxHome::new(); let runtime = boxlite::BoxliteRuntime::new(boxlite::runtime::options::BoxliteOptions { home_dir: home.path.clone(), @@ -159,7 +159,7 @@ async fn late_attach_reports_output_dropped() { let mut stderr = execution.stderr().expect("stderr stream"); let dropped = tokio::time::timeout(std::time::Duration::from_secs(5), async { while let Some(chunk) = stderr.next().await { - if chunk.contains("[boxlite] output dropped") { + if chunk.contains("[boxlite] stdout output dropped") { return Some(chunk); } } @@ -174,8 +174,8 @@ async fn late_attach_reports_output_dropped() { let dropped = dropped.expect("late attach must report overwritten output"); assert!( - dropped.contains("stdout:") && dropped.contains("stderr: 0 bytes"), - "stdout-only loss must preserve the stderr decoder: {dropped:?}" + dropped.contains("stdout output dropped"), + "late attach must report the stdout gap: {dropped:?}" ); } diff --git a/src/guest/src/container/command.rs b/src/guest/src/container/command.rs index 58926a0db..c3e9cd63c 100644 --- a/src/guest/src/container/command.rs +++ b/src/guest/src/container/command.rs @@ -547,7 +547,7 @@ pub(crate) fn create_pty_child( Ok(child) } -fn terminate_process(pid: Pid) { +pub(super) fn terminate_process(pid: Pid) { let _fence = crate::reaper::reap_fence(); let _ = nix::sys::signal::kill(pid, nix::sys::signal::Signal::SIGKILL); let _ = nix::sys::wait::waitpid(pid, None); diff --git a/src/guest/src/container/lifecycle.rs b/src/guest/src/container/lifecycle.rs index 46d07eb4c..cebb580fa 100644 --- a/src/guest/src/container/lifecycle.rs +++ b/src/guest/src/container/lifecycle.rs @@ -339,7 +339,7 @@ impl Container { stdin, stdout, stderr, - } => ExecHandle::new(pid, stdin, stdout, Some(stderr))?, + } => init_pipe_handle(pid, stdin, stdout, stderr)?, // Mirrors the tenant PTY path: the master becomes stdin+stdout and // is retained for window-size ioctls, so ResizeTty reaches the main // command exactly as it reaches an exec. @@ -540,6 +540,21 @@ impl Container { } } +fn init_pipe_handle( + pid: nix::unistd::Pid, + stdin: std::os::fd::OwnedFd, + stdout: std::os::fd::OwnedFd, + stderr: std::os::fd::OwnedFd, +) -> BoxliteResult { + match ExecHandle::new(pid, stdin, stdout, Some(stderr)) { + Ok(handle) => Ok(handle), + Err(error) => { + super::command::terminate_process(pid); + Err(error) + } + } +} + // ==================== // Init Health Check // ==================== @@ -580,3 +595,51 @@ impl Drop for Container { tracing::debug!(container_id = %self.id, "Container cleanup complete"); } } + +#[cfg(test)] +mod tests { + use super::*; + use nix::sys::signal::kill; + use nix::unistd::{pipe, Pid}; + use std::os::fd::OwnedFd; + use std::process::{Child, Command}; + + struct ChildGuard(Child); + + impl Drop for ChildGuard { + fn drop(&mut self) { + let _ = self.0.kill(); + let _ = self.0.wait(); + } + } + + #[test] + fn init_pipe_handle_failure_reaps_init() { + let child = ChildGuard( + Command::new("/bin/sh") + .args(["-c", "sleep 30"]) + .spawn() + .unwrap(), + ); + let pid = Pid::from_raw(child.0.id() as i32); + let (_stdin_read, stdin_write) = pipe().unwrap(); + let stdout: OwnedFd = std::fs::File::open("/proc/self/stat").unwrap().into(); + let (stderr_read, _stderr_write) = pipe().unwrap(); + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_io() + .build() + .unwrap(); + + let result = + runtime.block_on(async { init_pipe_handle(pid, stdin_write, stdout, stderr_read) }); + + assert!( + result.is_err(), + "regular files cannot register with AsyncFd" + ); + assert!( + kill(pid, None).is_err(), + "failed init handle must not leave a child" + ); + } +} diff --git a/src/guest/src/service/exec/output.rs b/src/guest/src/service/exec/output.rs index 18b777e7b..e611b315c 100644 --- a/src/guest/src/service/exec/output.rs +++ b/src/guest/src/service/exec/output.rs @@ -1,6 +1,6 @@ use crate::service::exec::exec_handle::{ExecStderr, ExecStdout}; use async_stream::stream; -use boxlite_shared::{exec_output, ExecOutput, OutputDropped, Stderr, Stdout}; +use boxlite_shared::{exec_output, ExecOutput, Stderr, Stdout}; use futures::{Stream, StreamExt}; use std::collections::VecDeque; use std::pin::Pin; @@ -23,16 +23,14 @@ struct OutputState { buffered_bytes: usize, oldest_sequence: u64, next_sequence: u64, - pending_dropped: DroppedBytes, - open_readers: usize, attached: bool, - attachment_next_sequence: Option, + stdout: StreamState, + stderr: StreamState, } struct OutputEntry { sequence: u64, output: ExecOutput, - source: OutputSource, byte_len: usize, } @@ -42,29 +40,17 @@ enum OutputSource { Stderr, } -#[derive(Default)] -struct DroppedBytes { - stdout: u64, - stderr: u64, -} - -impl DroppedBytes { - fn record(&mut self, source: OutputSource, byte_len: usize) { - let byte_len = byte_len as u64; - match source { - OutputSource::Stdout => self.stdout += byte_len, - OutputSource::Stderr => self.stderr += byte_len, - } - } - - fn take(&mut self) -> Self { - std::mem::take(self) - } +struct StreamState { + enabled: bool, + finished: bool, + total_bytes: u64, + last_sequence: Option, } impl OutputManager { pub(crate) fn new(stdout: Option, stderr: Option) -> Self { - let open_readers = usize::from(stdout.is_some()) + usize::from(stderr.is_some()); + let stdout_enabled = stdout.is_some(); + let stderr_enabled = stderr.is_some(); let (updated, _) = watch::channel(()); let manager = Self { inner: Arc::new(Mutex::new(OutputState { @@ -72,10 +58,19 @@ impl OutputManager { buffered_bytes: 0, oldest_sequence: 0, next_sequence: 0, - pending_dropped: DroppedBytes::default(), - open_readers, attached: false, - attachment_next_sequence: None, + stdout: StreamState { + enabled: stdout_enabled, + finished: !stdout_enabled, + total_bytes: 0, + last_sequence: None, + }, + stderr: StreamState { + enabled: stderr_enabled, + finished: !stderr_enabled, + total_bytes: 0, + last_sequence: None, + }, })), updated, }; @@ -97,12 +92,13 @@ impl OutputManager { return Err(Status::already_exists("Already attached")); } state.attached = true; - state.attachment_next_sequence = Some(0); } let manager = self.clone(); let output = stream! { let mut next_sequence = 0; + let mut stdout_end_sent = false; + let mut stderr_end_sent = false; let mut updates = manager.updated.subscribe(); loop { @@ -113,12 +109,18 @@ impl OutputManager { } let next = { - let mut state = manager.inner.lock().await; + let state = manager.inner.lock().await; if next_sequence < state.oldest_sequence { next_sequence = state.oldest_sequence; - state.attachment_next_sequence = Some(next_sequence); - Next::Item(dropped_output(state.pending_dropped.take())) + } + + if state.stdout.ready_to_end(next_sequence) && !stdout_end_sent { + stdout_end_sent = true; + Next::Item(end_output(OutputSource::Stdout, state.stdout.total_bytes)) + } else if state.stderr.ready_to_end(next_sequence) && !stderr_end_sent { + stderr_end_sent = true; + Next::Item(end_output(OutputSource::Stderr, state.stderr.total_bytes)) } else if next_sequence < state.next_sequence { let index = (next_sequence - state.oldest_sequence) as usize; let output = state @@ -128,9 +130,10 @@ impl OutputManager { .output .clone(); next_sequence += 1; - state.attachment_next_sequence = Some(next_sequence); Next::Item(output) - } else if state.open_readers == 0 { + } else if (!state.stdout.enabled || stdout_end_sent) + && (!state.stderr.enabled || stderr_end_sent) + { Next::Done } else { Next::Wait @@ -173,38 +176,28 @@ impl OutputManager { while let Some(data) = stream.next().await { self.push(source, data).await; } - self.reader_finished().await; + self.reader_finished(source).await; } async fn push(&self, source: OutputSource, data: Vec) { let byte_len = data.len(); - let output = match source { - OutputSource::Stdout => ExecOutput { - event: Some(exec_output::Event::Stdout(Stdout { data })), - }, - OutputSource::Stderr => ExecOutput { - event: Some(exec_output::Event::Stderr(Stderr { data })), - }, - }; let mut state = self.inner.lock().await; let sequence = state.next_sequence; state.next_sequence += 1; + let stream = state.stream_mut(source); + stream.enabled = true; + let offset = stream.total_bytes; + stream.total_bytes += byte_len as u64; + stream.last_sequence = Some(sequence); + let output = data_output(source, data, offset); while state.buffered_bytes + byte_len > BUFFER_CAPACITY_BYTES { let Some(removed) = state.entries.pop_front() else { - if state.entry_is_unread(sequence) { - state.pending_dropped.record(source, byte_len); - } state.oldest_sequence = state.next_sequence; break; }; state.buffered_bytes -= removed.byte_len; - if state.entry_is_unread(removed.sequence) { - state - .pending_dropped - .record(removed.source, removed.byte_len); - } state.oldest_sequence = removed.sequence + 1; } @@ -213,7 +206,6 @@ impl OutputManager { state.entries.push_back(OutputEntry { sequence, output, - source, byte_len, }); } @@ -221,56 +213,154 @@ impl OutputManager { self.updated.send_replace(()); } - async fn reader_finished(&self) { + async fn reader_finished(&self, source: OutputSource) { let mut state = self.inner.lock().await; - state.open_readers -= 1; + state.stream_mut(source).finished = true; drop(state); self.updated.send_replace(()); } } impl OutputState { - fn entry_is_unread(&self, sequence: u64) -> bool { - self.attachment_next_sequence - .is_none_or(|next_sequence| sequence >= next_sequence) + fn stream_mut(&mut self, source: OutputSource) -> &mut StreamState { + match source { + OutputSource::Stdout => &mut self.stdout, + OutputSource::Stderr => &mut self.stderr, + } } } -fn dropped_output(dropped: DroppedBytes) -> ExecOutput { - ExecOutput { - event: Some(exec_output::Event::Dropped(OutputDropped { - stdout_bytes: dropped.stdout, - stderr_bytes: dropped.stderr, - })), +impl StreamState { + fn ready_to_end(&self, next_sequence: u64) -> bool { + self.enabled + && self.finished + && self + .last_sequence + .is_none_or(|last_sequence| next_sequence > last_sequence) } } +fn data_output(source: OutputSource, data: Vec, offset: u64) -> ExecOutput { + let event = match source { + OutputSource::Stdout => exec_output::Event::Stdout(Stdout { + data, + offset: Some(offset), + total_bytes: None, + }), + OutputSource::Stderr => exec_output::Event::Stderr(Stderr { + data, + offset: Some(offset), + total_bytes: None, + }), + }; + ExecOutput { event: Some(event) } +} + +fn end_output(source: OutputSource, total_bytes: u64) -> ExecOutput { + let event = match source { + OutputSource::Stdout => exec_output::Event::Stdout(Stdout { + data: Vec::new(), + offset: Some(total_bytes), + total_bytes: Some(total_bytes), + }), + OutputSource::Stderr => exec_output::Event::Stderr(Stderr { + data: Vec::new(), + offset: Some(total_bytes), + total_bytes: Some(total_bytes), + }), + }; + ExecOutput { event: Some(event) } +} + #[cfg(test)] mod tests { use super::*; #[tokio::test] - async fn dropped_counts_only_entries_the_attachment_has_not_received() { + async fn completed_stream_emits_total_bytes_after_replayed_data() { let manager = OutputManager::new(None, None); manager .push(OutputSource::Stdout, b"already sent".to_vec()) .await; let mut output = manager.attach().await.unwrap(); - assert!(matches!( - output.next().await.unwrap().unwrap().event, - Some(exec_output::Event::Stdout(_)) - )); + let first = output.next().await.unwrap().unwrap(); + let Some(exec_output::Event::Stdout(first)) = first.event else { + panic!("the replayed stdout data must be sent first"); + }; + assert_eq!(first.data, b"already sent"); + assert_eq!(first.offset, Some(0)); + assert_eq!(first.total_bytes, None); - for _ in 0..=BUFFER_CAPACITY_BYTES / 1024 { - manager.push(OutputSource::Stderr, vec![0; 1024]).await; - } + let end = output.next().await.unwrap().unwrap(); + let Some(exec_output::Event::Stdout(end)) = end.event else { + panic!("the stdout end frame must follow its replayed data"); + }; + assert!(end.data.is_empty()); + assert_eq!(end.offset, Some(b"already sent".len() as u64)); + assert_eq!(end.total_bytes, Some(b"already sent".len() as u64)); + } + + #[tokio::test] + async fn closed_stdout_emits_its_end_before_stderr_closes() { + let (stdout_read, stdout_write) = nix::unistd::pipe().unwrap(); + let (stderr_read, stderr_write) = nix::unistd::pipe().unwrap(); + let manager = OutputManager::new( + Some(ExecStdout::new(stdout_read).unwrap()), + Some(ExecStderr::new(stderr_read).unwrap()), + ); + + nix::unistd::write(&stdout_write, b"out").unwrap(); + drop(stdout_write); + + let mut output = manager.attach().await.unwrap(); + let first = tokio::time::timeout(std::time::Duration::from_secs(1), output.next()) + .await + .expect("stdout data must arrive") + .unwrap() + .unwrap(); + let Some(exec_output::Event::Stdout(first)) = first.event else { + panic!("the first event must be stdout data"); + }; + assert_eq!(first.data, b"out"); + + let end = tokio::time::timeout(std::time::Duration::from_secs(1), output.next()) + .await + .expect("stdout end must not wait for stderr EOF") + .unwrap() + .unwrap(); + let Some(exec_output::Event::Stdout(end)) = end.event else { + panic!("stdout must emit its own end frame"); + }; + assert_eq!(end.total_bytes, Some(3)); + + drop(stderr_write); + } + + #[tokio::test] + async fn end_frames_survive_when_all_data_is_evicted() { + let manager = OutputManager::new(None, None); + manager.push(OutputSource::Stdout, b"lost".to_vec()).await; + manager + .push(OutputSource::Stderr, vec![0; BUFFER_CAPACITY_BYTES + 1]) + .await; + + let mut output = manager.attach().await.unwrap(); + let stdout_end = output.next().await.unwrap().unwrap(); + let Some(exec_output::Event::Stdout(stdout_end)) = stdout_end.event else { + panic!("stdout end frame must survive its evicted data"); + }; + assert!(stdout_end.data.is_empty()); + assert_eq!(stdout_end.total_bytes, Some(4)); - let dropped = output.next().await.unwrap().unwrap(); - let Some(exec_output::Event::Dropped(dropped)) = dropped.event else { - panic!("the unread stderr must be reported as dropped"); + let stderr_end = output.next().await.unwrap().unwrap(); + let Some(exec_output::Event::Stderr(stderr_end)) = stderr_end.event else { + panic!("stderr end frame must survive its evicted data"); }; - assert_eq!(dropped.stdout_bytes, 0); - assert!(dropped.stderr_bytes > 0); + assert!(stderr_end.data.is_empty()); + assert_eq!( + stderr_end.total_bytes, + Some((BUFFER_CAPACITY_BYTES + 1) as u64) + ); } } diff --git a/src/shared/proto/boxlite/v1/service.proto b/src/shared/proto/boxlite/v1/service.proto index 5acb22eb8..deb6753e4 100644 --- a/src/shared/proto/boxlite/v1/service.proto +++ b/src/shared/proto/boxlite/v1/service.proto @@ -444,13 +444,15 @@ message AttachRequest { } message ExecOutput { + // New guests mark each stream's end with an empty event carrying total_bytes. oneof event { Stdout stdout = 1; Stderr stderr = 2; - OutputDropped dropped = 3; + OutputDropped dropped = 3 [deprecated = true]; } } +// Sent only by guests predating per-stream offsets. message OutputDropped { uint64 stdout_bytes = 1; uint64 stderr_bytes = 2; @@ -458,10 +460,14 @@ message OutputDropped { message Stdout { bytes data = 1; + optional uint64 offset = 2; + optional uint64 total_bytes = 3; } message Stderr { bytes data = 1; + optional uint64 offset = 2; + optional uint64 total_bytes = 3; } // SendInput: client streaming stdin From d165e21b553a4db7eabf837997152a72c97e7790 Mon Sep 17 00:00:00 2001 From: BatmanByte <300328404+BatmanByte@users.noreply.github.com> Date: Wed, 29 Jul 2026 19:59:09 +0800 Subject: [PATCH 08/12] fix(exec): remove unpublished dropped output event OutputDropped was introduced only by this unmerged PR and has no released guest compatibility requirement. Remove the deprecated field and host decoder so the Attach protocol is exclusively per-stream offsets plus final byte counts. --- src/boxlite/src/portal/interfaces/exec.rs | 75 ----------------------- src/boxlite/tests/run_main_command.rs | 2 +- src/deps/libkrun-sys/build.rs | 5 +- src/guest/src/service/exec/exec_handle.rs | 7 ++- src/guest/src/service/exec/output.rs | 32 +++++++++- src/guest/src/service/exec/registry.rs | 3 +- src/guest/src/service/exec/state.rs | 37 ++++++----- src/guest/src/service/ssh/bridge.rs | 3 +- src/shared/proto/boxlite/v1/service.proto | 9 +-- 9 files changed, 64 insertions(+), 109 deletions(-) diff --git a/src/boxlite/src/portal/interfaces/exec.rs b/src/boxlite/src/portal/interfaces/exec.rs index 15a97f299..5bd7da4b8 100644 --- a/src/boxlite/src/portal/interfaces/exec.rs +++ b/src/boxlite/src/portal/interfaces/exec.rs @@ -423,14 +423,6 @@ impl ExecProtocol { Self::report_gap(stderr, "stderr", lost_bytes); } } - Some(exec_output::Event::Dropped(dropped)) => { - Self::report_legacy_drop( - stdout, - stderr, - dropped.stdout_bytes, - dropped.stderr_bytes, - ); - } None => {} } } @@ -446,28 +438,6 @@ impl ExecProtocol { )); } - fn report_legacy_drop( - stdout: &mut OutputTracker, - stderr: &mut OutputTracker, - stdout_bytes: u64, - stderr_bytes: u64, - ) { - tracing::warn!( - stdout_bytes, - stderr_bytes, - "Legacy guest output buffer dropped older output" - ); - if stdout_bytes > 0 { - stdout.flush(); - } - if stderr_bytes > 0 { - stderr.flush(); - } - let _ = stderr.stream.tx.send(format!( - "[boxlite] output dropped (stdout: {stdout_bytes} bytes, stderr: {stderr_bytes} bytes)\n" - )); - } - fn spawn_wait( mut client: ExecutionClient, execution_id: String, @@ -1339,51 +1309,6 @@ mod tests { assert_eq!(stderr_rx.try_recv().ok(), Some("─".to_string())); } - #[test] - fn legacy_drop_only_flushes_the_affected_utf8_decoder() { - use boxlite_shared::{OutputDropped, Stderr as StderrMsg, exec_output}; - - let (stdout_tx, mut stdout_rx) = mpsc::unbounded_channel::(); - let (stderr_tx, mut stderr_rx) = mpsc::unbounded_channel::(); - let mut stdout = OutputTracker::new(stdout_tx); - let mut stderr = OutputTracker::new(stderr_tx); - - let output = |event| ExecOutput { event: Some(event) }; - ExecProtocol::route_output( - output(exec_output::Event::Stderr(StderrMsg { - data: vec![0xE2], - offset: None, - total_bytes: None, - })), - &mut stdout, - &mut stderr, - ); - ExecProtocol::route_output( - output(exec_output::Event::Dropped(OutputDropped { - stdout_bytes: 1, - stderr_bytes: 0, - })), - &mut stdout, - &mut stderr, - ); - ExecProtocol::route_output( - output(exec_output::Event::Stderr(StderrMsg { - data: vec![0x94, 0x80], - offset: None, - total_bytes: None, - })), - &mut stdout, - &mut stderr, - ); - - assert!(stdout_rx.try_recv().is_err()); - assert_eq!( - stderr_rx.try_recv().ok(), - Some("[boxlite] output dropped (stdout: 1 bytes, stderr: 0 bytes)\n".to_string()) - ); - assert_eq!(stderr_rx.try_recv().ok(), Some("─".to_string())); - } - #[test] fn output_end_reports_a_missing_tail() { use boxlite_shared::{Stdout as StdoutMsg, exec_output}; diff --git a/src/boxlite/tests/run_main_command.rs b/src/boxlite/tests/run_main_command.rs index 4a79e6592..0719beb2d 100644 --- a/src/boxlite/tests/run_main_command.rs +++ b/src/boxlite/tests/run_main_command.rs @@ -106,7 +106,7 @@ async fn main_command_exits_after_large_output_without_attach() { let completed = tokio::time::timeout(std::time::Duration::from_secs(30), async { handle.start().await.expect("start box"); loop { - if handle.info().status == boxlite::BoxStatus::Stopped { + if handle.info().await.expect("get box info").status == boxlite::BoxStatus::Stopped { return true; } tokio::time::sleep(std::time::Duration::from_millis(100)).await; diff --git a/src/deps/libkrun-sys/build.rs b/src/deps/libkrun-sys/build.rs index 925a23d55..e1b8bd797 100644 --- a/src/deps/libkrun-sys/build.rs +++ b/src/deps/libkrun-sys/build.rs @@ -476,7 +476,10 @@ impl LibFixup { let mut cmd = Command::new("patchelf"); cmd.args(["--add-needed", LIBC, lib_path_str]); run_command(&mut cmd, &format!("add {} dependency", LIBC)); - println!("cargo:warning=Added {} dependency to {}", LIBC, lib_path_str); + println!( + "cargo:warning=Added {} dependency to {}", + LIBC, lib_path_str + ); } /// Extract SONAME from versioned library filename. diff --git a/src/guest/src/service/exec/exec_handle.rs b/src/guest/src/service/exec/exec_handle.rs index 714e081e0..456cb800f 100644 --- a/src/guest/src/service/exec/exec_handle.rs +++ b/src/guest/src/service/exec/exec_handle.rs @@ -482,8 +482,8 @@ mod process_group_tests { assert!(checked_process_group_target(Pid::from_raw(0), Pid::from_raw(0)).is_err()); } - #[test] - fn process_group_kill_reaches_a_background_descendant() { + #[tokio::test] + async fn process_group_kill_reaches_a_background_descendant() { let mut command = Command::new("/bin/sh"); command .arg("-c") @@ -520,7 +520,8 @@ mod process_group_tests { let (_stdin_peer, stdin) = nix::unistd::pipe().unwrap(); let (stdout, _stdout_peer) = nix::unistd::pipe().unwrap(); let (stderr, _stderr_peer) = nix::unistd::pipe().unwrap(); - let handle = ExecHandle::new(leader, stdin, stdout, Some(stderr)); + let handle = ExecHandle::new(leader, stdin, stdout, Some(stderr)) + .expect("test pipe must register with Tokio"); handle.kill_process_group(Signal::SIGTERM).unwrap(); std::thread::sleep(Duration::from_millis(25)); diff --git a/src/guest/src/service/exec/output.rs b/src/guest/src/service/exec/output.rs index e611b315c..05a30d248 100644 --- a/src/guest/src/service/exec/output.rs +++ b/src/guest/src/service/exec/output.rs @@ -4,8 +4,9 @@ use boxlite_shared::{exec_output, ExecOutput, Stderr, Stdout}; use futures::{Stream, StreamExt}; use std::collections::VecDeque; use std::pin::Pin; -use std::sync::Arc; +use std::sync::{Arc, Mutex as StdMutex}; use tokio::sync::{watch, Mutex}; +use tokio::task::JoinHandle; use tonic::Status; const BUFFER_CAPACITY_BYTES: usize = 1024 * 1024; @@ -16,6 +17,7 @@ pub(crate) type AttachStream = Pin>, updated: watch::Sender<()>, + drain_tasks: Arc>>>, } struct OutputState { @@ -73,6 +75,7 @@ impl OutputManager { }, })), updated, + drain_tasks: Arc::new(StdMutex::new(Vec::new())), }; if let Some(stdout) = stdout { @@ -85,6 +88,21 @@ impl OutputManager { manager } + pub(crate) async fn shutdown_drains(&self) { + let tasks = std::mem::take( + &mut *self + .drain_tasks + .lock() + .expect("output drain task lock poisoned"), + ); + for task in &tasks { + task.abort(); + } + for task in tasks { + let _ = task.await; + } + } + pub(crate) async fn attach(&self) -> Result { { let mut state = self.inner.lock().await; @@ -157,16 +175,24 @@ impl OutputManager { fn spawn_stdout(&self, stdout: ExecStdout) { let manager = self.clone(); - tokio::spawn(async move { + let task = tokio::spawn(async move { manager.drain(stdout, OutputSource::Stdout).await; }); + self.drain_tasks + .lock() + .expect("output drain task lock poisoned") + .push(task); } fn spawn_stderr(&self, stderr: ExecStderr) { let manager = self.clone(); - tokio::spawn(async move { + let task = tokio::spawn(async move { manager.drain(stderr, OutputSource::Stderr).await; }); + self.drain_tasks + .lock() + .expect("output drain task lock poisoned") + .push(task); } async fn drain(&self, mut stream: S, source: OutputSource) diff --git a/src/guest/src/service/exec/registry.rs b/src/guest/src/service/exec/registry.rs index 824661031..0c8b638dd 100644 --- a/src/guest/src/service/exec/registry.rs +++ b/src/guest/src/service/exec/registry.rs @@ -136,7 +136,8 @@ mod release_tests { let (_stdin_peer, stdin) = pipe().unwrap(); let (stdout, _stdout_peer) = pipe().unwrap(); let (stderr, _stderr_peer) = pipe().unwrap(); - let handle = ExecHandle::new(Pid::from_raw(pid), stdin, stdout, Some(stderr)); + let handle = ExecHandle::new(Pid::from_raw(pid), stdin, stdout, Some(stderr)) + .expect("test pipe must register with Tokio"); let exit = ExitSlot::settled_for_test(ExitStatus::Code(7)); if is_init { ExecutionState::new_init_session(handle, exit) diff --git a/src/guest/src/service/exec/state.rs b/src/guest/src/service/exec/state.rs index 9b88a5185..ea06fa1d9 100644 --- a/src/guest/src/service/exec/state.rs +++ b/src/guest/src/service/exec/state.rs @@ -1,6 +1,5 @@ use crate::service::exec::error::ExecutionError; use crate::service::exec::exec_handle::ExecHandle; -<<<<<<< HEAD use crate::service::exec::output::OutputManager; use boxlite_shared::ExecOutput; use futures::{Stream, StreamExt as _}; @@ -288,11 +287,9 @@ impl ExecutionState { .map_err(|_| ExecutionError::AlreadyAttached)?; let (tx, rx) = mpsc::channel(100); let task = tokio::spawn(async move { - while let Some(message) = output.next().await { - match message { - Ok(message) if tx.send(message).await.is_err() => break, - Ok(_) => {} - Err(_) => break, + while let Some(Ok(message)) = output.next().await { + if tx.send(message).await.is_err() { + break; } } }); @@ -314,20 +311,27 @@ impl ExecutionState { /// clone still exists. Forwarders are aborted because stdin may already have /// moved out of the handle into one of those tasks. pub(super) async fn release_resources(&self) -> bool { - let mut inner = self.inner.lock().await; - if inner.released { - return false; - } - inner.released = true; + let (output, input_tasks, output_tasks) = { + let mut inner = self.inner.lock().await; + if inner.released { + return false; + } + inner.released = true; + let output = inner.output.clone(); + let input_tasks = std::mem::take(&mut inner.input_tasks); + let output_tasks = std::mem::take(&mut inner.output_tasks); + drop(inner.handle.take()); + drop(inner.init_health.take()); + (output, input_tasks, output_tasks) + }; - for task in inner.input_tasks.drain(..) { + for task in input_tasks { task.abort(); } - for task in inner.output_tasks.drain(..) { + for task in output_tasks { task.abort(); } - drop(inner.handle.take()); - drop(inner.init_health.take()); + output.shutdown_drains().await; true } @@ -410,7 +414,8 @@ mod release_tests { pty_controller.as_raw_fd(), ]; - let mut handle = ExecHandle::new(Pid::from_raw(42_424), stdin, stdout, Some(stderr)); + let mut handle = ExecHandle::new(Pid::from_raw(42_424), stdin, stdout, Some(stderr)) + .expect("test pipe must register with Tokio"); handle.set_pty( std::fs::File::from(pty_controller), PtyConfig { diff --git a/src/guest/src/service/ssh/bridge.rs b/src/guest/src/service/ssh/bridge.rs index 59f2cb228..7d8a9aca5 100644 --- a/src/guest/src/service/ssh/bridge.rs +++ b/src/guest/src/service/ssh/bridge.rs @@ -885,7 +885,8 @@ mod tests { let (_stdin_peer, stdin) = pipe().unwrap(); let (stdout, _stdout_peer) = pipe().unwrap(); let (stderr, _stderr_peer) = pipe().unwrap(); - let handle = ExecHandle::new(Pid::from_raw(pid), stdin, stdout, Some(stderr)); + let handle = ExecHandle::new(Pid::from_raw(pid), stdin, stdout, Some(stderr)) + .expect("test pipe must register with Tokio"); let (exit, exit_tx) = ExitSlot::pending_for_test(); let state = ExecutionState::new_for_test(handle, exit); registry.register(execution_id.into(), state).await; diff --git a/src/shared/proto/boxlite/v1/service.proto b/src/shared/proto/boxlite/v1/service.proto index deb6753e4..5f31c0037 100644 --- a/src/shared/proto/boxlite/v1/service.proto +++ b/src/shared/proto/boxlite/v1/service.proto @@ -444,20 +444,13 @@ message AttachRequest { } message ExecOutput { - // New guests mark each stream's end with an empty event carrying total_bytes. + // An empty stdout/stderr event with total_bytes marks that stream's end. oneof event { Stdout stdout = 1; Stderr stderr = 2; - OutputDropped dropped = 3 [deprecated = true]; } } -// Sent only by guests predating per-stream offsets. -message OutputDropped { - uint64 stdout_bytes = 1; - uint64 stderr_bytes = 2; -} - message Stdout { bytes data = 1; optional uint64 offset = 2; From dc31d495ee88a3cb55b7bb7e5d17f9482f99a4b1 Mon Sep 17 00:00:00 2001 From: BatmanByte <300328404+BatmanByte@users.noreply.github.com> Date: Thu, 30 Jul 2026 00:07:37 +0800 Subject: [PATCH 09/12] fix(exec): report output reader failures --- src/boxlite/src/portal/interfaces/exec.rs | 101 ++++++++++---- src/guest/src/service/exec/exec_handle.rs | 49 +++++-- src/guest/src/service/exec/mod.rs | 4 +- src/guest/src/service/exec/output.rs | 129 +++++++++++------- src/guest/src/service/exec/state.rs | 9 +- src/guest/src/service/ssh/bridge.rs | 118 ++++++++++++++-- .../src/service/ssh/reverse_streamlocal.rs | 31 +++-- 7 files changed, 329 insertions(+), 112 deletions(-) diff --git a/src/boxlite/src/portal/interfaces/exec.rs b/src/boxlite/src/portal/interfaces/exec.rs index 5bd7da4b8..f87771d0e 100644 --- a/src/boxlite/src/portal/interfaces/exec.rs +++ b/src/boxlite/src/portal/interfaces/exec.rs @@ -409,32 +409,59 @@ impl ExecProtocol { match output.event { Some(exec_output::Event::Stdout(chunk)) => { tracing::trace!(len = chunk.data.len(), "Received exec stdout"); - if let Some(lost_bytes) = - stdout.receive(chunk.offset, chunk.data, chunk.total_bytes) - { - Self::report_gap(stderr, "stdout", lost_bytes); - } + Self::forward_output( + stdout, + "stdout", + chunk.offset, + chunk.data, + chunk.total_bytes, + ); } Some(exec_output::Event::Stderr(chunk)) => { tracing::trace!(len = chunk.data.len(), "Received exec stderr"); - if let Some(lost_bytes) = - stderr.receive(chunk.offset, chunk.data, chunk.total_bytes) - { - Self::report_gap(stderr, "stderr", lost_bytes); - } + Self::forward_output( + stderr, + "stderr", + chunk.offset, + chunk.data, + chunk.total_bytes, + ); } None => {} } } - fn report_gap(stderr: &mut OutputTracker, source: &str, lost_bytes: u64) { + fn forward_output( + output: &mut OutputTracker, + source: &str, + offset: Option, + data: Vec, + total_bytes: Option, + ) { + match output.receive(offset, data, total_bytes) { + ReceivedOutput::Data { data, lost_bytes } => { + if let Some(lost_bytes) = lost_bytes { + Self::report_gap(output, source, lost_bytes); + } + output.stream.send_bytes(data); + } + ReceivedOutput::End { lost_bytes } => { + if let Some(lost_bytes) = lost_bytes { + Self::report_gap(output, source, lost_bytes); + } + } + ReceivedOutput::Ignore => {} + } + } + + fn report_gap(output: &mut OutputTracker, source: &str, lost_bytes: u64) { tracing::warn!( source, lost_bytes, "Guest output buffer dropped older output" ); - let _ = stderr.stream.tx.send(format!( - "[boxlite] {source} output dropped {lost_bytes} bytes\n" + let _ = output.stream.tx.send(format!( + "[boxlite] {source} output dropped {lost_bytes} bytes\r\n" )); } @@ -744,6 +771,17 @@ struct OutputTracker { stream: DecodedStream, } +enum ReceivedOutput { + Data { + data: Vec, + lost_bytes: Option, + }, + End { + lost_bytes: Option, + }, + Ignore, +} + impl OutputTracker { fn new(tx: mpsc::UnboundedSender) -> Self { Self { @@ -757,15 +795,15 @@ impl OutputTracker { offset: Option, data: Vec, total_bytes: Option, - ) -> Option { + ) -> ReceivedOutput { if let Some(total_bytes) = total_bytes { let Some(offset) = offset else { tracing::warn!(total_bytes, "Exec output end frame has no offset"); - return None; + return ReceivedOutput::Ignore; }; if !data.is_empty() || offset != total_bytes { tracing::warn!(offset, total_bytes, "Invalid exec output end frame"); - return None; + return ReceivedOutput::Ignore; } if total_bytes < self.expected_offset { tracing::warn!( @@ -773,19 +811,23 @@ impl OutputTracker { total_bytes, "Exec output end frame moved backwards" ); - return None; + return ReceivedOutput::Ignore; } let lost_bytes = total_bytes - self.expected_offset; self.expected_offset = total_bytes; self.stream.flush(); - return (lost_bytes > 0).then_some(lost_bytes); + return ReceivedOutput::End { + lost_bytes: (lost_bytes > 0).then_some(lost_bytes), + }; } let Some(offset) = offset else { self.expected_offset += data.len() as u64; - self.stream.send_bytes(data); - return None; + return ReceivedOutput::Data { + data, + lost_bytes: None, + }; }; if offset < self.expected_offset { @@ -794,7 +836,7 @@ impl OutputTracker { offset, "Exec output chunk moved backwards" ); - return None; + return ReceivedOutput::Ignore; } let lost_bytes = offset - self.expected_offset; @@ -802,8 +844,10 @@ impl OutputTracker { self.stream.flush(); } self.expected_offset = offset + data.len() as u64; - self.stream.send_bytes(data); - (lost_bytes > 0).then_some(lost_bytes) + ReceivedOutput::Data { + data, + lost_bytes: (lost_bytes > 0).then_some(lost_bytes), + } } fn flush(&mut self) { @@ -1301,11 +1345,11 @@ mod tests { &mut stderr, ); - assert_eq!(stdout_rx.try_recv().ok(), Some("after-gap".to_string())); assert_eq!( - stderr_rx.try_recv().ok(), - Some("[boxlite] stdout output dropped 1 bytes\n".to_string()) + stdout_rx.try_recv().ok(), + Some("[boxlite] stdout output dropped 1 bytes\r\n".to_string()) ); + assert_eq!(stdout_rx.try_recv().ok(), Some("after-gap".to_string())); assert_eq!(stderr_rx.try_recv().ok(), Some("─".to_string())); } @@ -1340,9 +1384,10 @@ mod tests { assert_eq!(stdout_rx.try_recv().ok(), Some("hello".to_string())); assert_eq!( - stderr_rx.try_recv().ok(), - Some("[boxlite] stdout output dropped 5 bytes\n".to_string()) + stdout_rx.try_recv().ok(), + Some("[boxlite] stdout output dropped 5 bytes\r\n".to_string()) ); + assert!(stderr_rx.try_recv().is_err()); } #[test] diff --git a/src/guest/src/service/exec/exec_handle.rs b/src/guest/src/service/exec/exec_handle.rs index 456cb800f..edc349615 100644 --- a/src/guest/src/service/exec/exec_handle.rs +++ b/src/guest/src/service/exec/exec_handle.rs @@ -75,20 +75,30 @@ fn set_nonblocking(fd: &OwnedFd) -> BoxliteResult<()> { } fn read_fd(fd: &OwnedFd, buffer: &mut [u8]) -> io::Result { - nix::unistd::read(fd.as_raw_fd(), buffer).map_err(Into::into) + loop { + match nix::unistd::read(fd.as_raw_fd(), buffer) { + Err(nix::errno::Errno::EINTR) => continue, + result => return result.map_err(Into::into), + } + } } fn write_fd(fd: &OwnedFd, buffer: &[u8]) -> io::Result { - nix::unistd::write(fd, buffer).map_err(Into::into) + loop { + match nix::unistd::write(fd, buffer) { + Err(nix::errno::Errno::EINTR) => continue, + result => return result.map_err(Into::into), + } + } } // Shared output stream implementation struct OutputStream { - inner: Pin> + Send>>, + inner: Pin>> + Send>>, } impl OutputStream { - fn new(fd: OwnedFd) -> BoxliteResult { + fn new(fd: OwnedFd, pty_eio_is_eof: bool) -> BoxliteResult { use async_stream::stream; let reader = async_fd(fd, "output")?; @@ -98,8 +108,12 @@ impl OutputStream { loop { match reader.async_io(Interest::READABLE, |fd| read_fd(fd, &mut buf)).await { Ok(0) => break, // EOF - Ok(n) => yield buf[..n].to_vec(), - Err(_) => break, + Ok(n) => yield Ok(buf[..n].to_vec()), + Err(error) if pty_eio_is_eof && error.raw_os_error() == Some(nix::libc::EIO) => break, + Err(error) => { + yield Err(error); + break; + } } } }; @@ -111,7 +125,7 @@ impl OutputStream { } impl Stream for OutputStream { - type Item = Vec; + type Item = io::Result>; fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { self.inner.as_mut().poll_next(cx) @@ -129,13 +143,19 @@ impl ExecStdout { /// Create from file descriptor pub fn new(fd: OwnedFd) -> BoxliteResult { Ok(Self { - inner: OutputStream::new(fd)?, + inner: OutputStream::new(fd, false)?, + }) + } + + fn new_pty(fd: OwnedFd) -> BoxliteResult { + Ok(Self { + inner: OutputStream::new(fd, true)?, }) } } impl Stream for ExecStdout { - type Item = Vec; + type Item = io::Result>; fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { self.inner.poll_next_unpin(cx) @@ -153,13 +173,13 @@ impl ExecStderr { /// Create from file descriptor pub fn new(fd: OwnedFd) -> BoxliteResult { Ok(Self { - inner: OutputStream::new(fd)?, + inner: OutputStream::new(fd, false)?, }) } } impl Stream for ExecStderr { - type Item = Vec; + type Item = io::Result>; fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { self.inner.poll_next_unpin(cx) @@ -305,10 +325,15 @@ impl ExecHandle { stdout: OwnedFd, stderr: Option, ) -> BoxliteResult { + let stdout = if stderr.is_some() { + ExecStdout::new(stdout)? + } else { + ExecStdout::new_pty(stdout)? + }; Ok(Self { pid, stdin: Some(ExecStdin::new(stdin)?), - stdout: Some(ExecStdout::new(stdout)?), + stdout: Some(stdout), // In PTY mode, stderr is None because stdout/stderr are merged // at the PTY level (single reader from PTY master) stderr: stderr.map(ExecStderr::new).transpose()?, diff --git a/src/guest/src/service/exec/mod.rs b/src/guest/src/service/exec/mod.rs index 29119c348..616bf31c9 100644 --- a/src/guest/src/service/exec/mod.rs +++ b/src/guest/src/service/exec/mod.rs @@ -65,7 +65,7 @@ impl GuestServer { pub(crate) async fn attach_execution( &self, exec_id: &str, - ) -> Result, ExecutionError> { + ) -> Result>, ExecutionError> { info!(execution_id = %exec_id, "attach request"); self.execution(exec_id).await?.attach(exec_id).await } @@ -195,7 +195,7 @@ impl Execution for GuestServer { let rx = self .attach_execution(&request.into_inner().execution_id) .await?; - let stream = ReceiverStream::new(rx).map(Ok); + let stream = ReceiverStream::new(rx); Ok(Response::new(Box::pin(stream) as Self::AttachStream)) } diff --git a/src/guest/src/service/exec/output.rs b/src/guest/src/service/exec/output.rs index 05a30d248..68d96b11e 100644 --- a/src/guest/src/service/exec/output.rs +++ b/src/guest/src/service/exec/output.rs @@ -8,6 +8,7 @@ use std::sync::{Arc, Mutex as StdMutex}; use tokio::sync::{watch, Mutex}; use tokio::task::JoinHandle; use tonic::Status; +use tracing::error; const BUFFER_CAPACITY_BYTES: usize = 1024 * 1024; @@ -25,6 +26,7 @@ struct OutputState { buffered_bytes: usize, oldest_sequence: u64, next_sequence: u64, + failure: Option, attached: bool, stdout: StreamState, stderr: StreamState, @@ -36,7 +38,7 @@ struct OutputEntry { byte_len: usize, } -#[derive(Clone, Copy)] +#[derive(Clone, Copy, Debug)] enum OutputSource { Stdout, Stderr, @@ -60,6 +62,7 @@ impl OutputManager { buffered_bytes: 0, oldest_sequence: 0, next_sequence: 0, + failure: None, attached: false, stdout: StreamState { enabled: stdout_enabled, @@ -79,10 +82,10 @@ impl OutputManager { }; if let Some(stdout) = stdout { - manager.spawn_stdout(stdout); + manager.spawn(stdout, OutputSource::Stdout); } if let Some(stderr) = stderr { - manager.spawn_stderr(stderr); + manager.spawn(stderr, OutputSource::Stderr); } manager @@ -122,6 +125,7 @@ impl OutputManager { loop { enum Next { Item(ExecOutput), + Error(Status), Wait, Done, } @@ -129,43 +133,46 @@ impl OutputManager { let next = { let state = manager.inner.lock().await; - if next_sequence < state.oldest_sequence { - next_sequence = state.oldest_sequence; - } - - if state.stdout.ready_to_end(next_sequence) && !stdout_end_sent { - stdout_end_sent = true; - Next::Item(end_output(OutputSource::Stdout, state.stdout.total_bytes)) - } else if state.stderr.ready_to_end(next_sequence) && !stderr_end_sent { - stderr_end_sent = true; - Next::Item(end_output(OutputSource::Stderr, state.stderr.total_bytes)) - } else if next_sequence < state.next_sequence { - let index = (next_sequence - state.oldest_sequence) as usize; - let output = state - .entries - .get(index) - .expect("ring sequence must exist") - .output - .clone(); - next_sequence += 1; - Next::Item(output) - } else if (!state.stdout.enabled || stdout_end_sent) - && (!state.stderr.enabled || stderr_end_sent) - { - Next::Done + if let Some(failure) = &state.failure { + Next::Error(Status::internal(failure.clone())) } else { - Next::Wait + if next_sequence < state.oldest_sequence { + next_sequence = state.oldest_sequence; + } + if state.stdout.ready_to_end(next_sequence) && !stdout_end_sent { + stdout_end_sent = true; + Next::Item(end_output(OutputSource::Stdout, state.stdout.total_bytes)) + } else if state.stderr.ready_to_end(next_sequence) && !stderr_end_sent { + stderr_end_sent = true; + Next::Item(end_output(OutputSource::Stderr, state.stderr.total_bytes)) + } else if next_sequence < state.next_sequence { + let index = (next_sequence - state.oldest_sequence) as usize; + let output = state + .entries + .get(index) + .expect("ring sequence must exist") + .output + .clone(); + next_sequence += 1; + Next::Item(output) + } else if (!state.stdout.enabled || stdout_end_sent) + && (!state.stderr.enabled || stderr_end_sent) + { + Next::Done + } else { + Next::Wait + } } }; match next { Next::Item(item) => yield Ok(item), - Next::Done => break, - Next::Wait => { - if updates.changed().await.is_err() { - break; - } + Next::Error(status) => { + yield Err(status); + break; } + Next::Done => break, + Next::Wait => updates.changed().await.expect("output manager must outlive attach stream"), } } }; @@ -173,21 +180,13 @@ impl OutputManager { Ok(Box::pin(output)) } - fn spawn_stdout(&self, stdout: ExecStdout) { - let manager = self.clone(); - let task = tokio::spawn(async move { - manager.drain(stdout, OutputSource::Stdout).await; - }); - self.drain_tasks - .lock() - .expect("output drain task lock poisoned") - .push(task); - } - - fn spawn_stderr(&self, stderr: ExecStderr) { + fn spawn(&self, stream: S, source: OutputSource) + where + S: Stream>> + Send + Unpin + 'static, + { let manager = self.clone(); let task = tokio::spawn(async move { - manager.drain(stderr, OutputSource::Stderr).await; + manager.drain(stream, source).await; }); self.drain_tasks .lock() @@ -197,10 +196,17 @@ impl OutputManager { async fn drain(&self, mut stream: S, source: OutputSource) where - S: Stream> + Unpin, + S: Stream>> + Unpin, { - while let Some(data) = stream.next().await { - self.push(source, data).await; + while let Some(item) = stream.next().await { + match item { + Ok(data) => self.push(source, data).await, + Err(error) => { + error!(?source, %error, "execution output reader failed"); + self.reader_failed(source, error).await; + return; + } + } } self.reader_finished(source).await; } @@ -245,6 +251,15 @@ impl OutputManager { drop(state); self.updated.send_replace(()); } + + async fn reader_failed(&self, source: OutputSource, error: std::io::Error) { + let mut state = self.inner.lock().await; + let stream = state.stream_mut(source); + stream.finished = true; + state.failure = Some(format!("failed to read {source:?}: {error}")); + drop(state); + self.updated.send_replace(()); + } } impl OutputState { @@ -389,4 +404,20 @@ mod tests { Some((BUFFER_CAPACITY_BYTES + 1) as u64) ); } + + #[tokio::test] + async fn reader_failure_is_reported_to_attach() { + let manager = OutputManager::new(None, None); + manager + .reader_failed( + OutputSource::Stdout, + std::io::Error::other("simulated pipe failure"), + ) + .await; + + let mut output = manager.attach().await.unwrap(); + let error = output.next().await.unwrap().unwrap_err(); + assert_eq!(error.code(), tonic::Code::Internal); + assert!(error.message().contains("simulated pipe failure")); + } } diff --git a/src/guest/src/service/exec/state.rs b/src/guest/src/service/exec/state.rs index ea06fa1d9..36d2b1ff3 100644 --- a/src/guest/src/service/exec/state.rs +++ b/src/guest/src/service/exec/state.rs @@ -7,6 +7,7 @@ use std::os::unix::io::AsRawFd; use std::sync::Arc; use tokio::sync::{mpsc, Mutex}; use tokio::task::{AbortHandle, JoinHandle}; +use tonic::Status; /// Abstraction for checking container init health. /// @@ -272,8 +273,8 @@ impl ExecutionState { /// Attach to execution output. pub async fn attach( &self, - _exec_id: &str, - ) -> Result, ExecutionError> { + exec_id: &str, + ) -> Result>, ExecutionError> { let output = { let inner = self.inner.lock().await; if inner.released { @@ -286,12 +287,14 @@ impl ExecutionState { .await .map_err(|_| ExecutionError::AlreadyAttached)?; let (tx, rx) = mpsc::channel(100); + let execution_id = exec_id.to_owned(); let task = tokio::spawn(async move { - while let Some(Ok(message)) = output.next().await { + while let Some(message) = output.next().await { if tx.send(message).await.is_err() { break; } } + tracing::info!(%execution_id, "execution output forwarding ended"); }); let mut inner = self.inner.lock().await; diff --git a/src/guest/src/service/ssh/bridge.rs b/src/guest/src/service/ssh/bridge.rs index 7d8a9aca5..5d021202a 100644 --- a/src/guest/src/service/ssh/bridge.rs +++ b/src/guest/src/service/ssh/bridge.rs @@ -511,14 +511,14 @@ fn streamlocal_launch(container_id: &str, socket_path: String) -> InternalHelper } struct PreparedStreamlocalOutput { - output: mpsc::Receiver, + output: mpsc::Receiver>, buffered_stdout: Vec, } enum OutputPumpStart { Attach, AwaitActivation { - output: mpsc::Receiver, + output: mpsc::Receiver>, buffered_stdout: Vec, activate_rx: oneshot::Receiver<()>, }, @@ -559,9 +559,15 @@ async fn await_streamlocal_ready( let mut parser = StreamlocalReadyParser::default(); loop { - let message = output.recv().await.ok_or_else(|| { - BridgeError::Exec("streamlocal helper exited before readiness".into()) - })?; + let message = output + .recv() + .await + .ok_or_else(|| { + BridgeError::Exec("streamlocal helper exited before readiness".into()) + })? + .map_err(|error| { + BridgeError::Exec(format!("streamlocal output failed: {error}")) + })?; match message.event { Some(boxlite_shared::exec_output::Event::Stdout(stdout)) => { if let Some(buffered_stdout) = parser.push_stdout(&stdout.data)? { @@ -655,6 +661,8 @@ fn spawn_output_pump( let _ = session_handle.data(channel_id, buffered_stdout).await; } + let mut stdout_offset = 0; + let mut stderr_offset = 0; loop { tokio::select! { // A resolved oneshot receiver must not be polled again by the @@ -664,15 +672,58 @@ fn spawn_output_pump( _ = &mut cancel_rx => return, message = output.recv() => { match message { - Some(message) => match message.event { + Some(Ok(message)) => match message.event { Some(boxlite_shared::exec_output::Event::Stdout(stdout)) => { - let _ = session_handle.data(channel_id, stdout.data).await; + if let Some(lost_bytes) = output_gap( + &mut stdout_offset, + stdout.offset, + stdout.data.len(), + stdout.total_bytes, + ) { + let _ = session_handle + .data( + channel_id, + output_gap_message("stdout", lost_bytes), + ) + .await; + } + if !stdout.data.is_empty() { + let _ = session_handle.data(channel_id, stdout.data).await; + } } Some(boxlite_shared::exec_output::Event::Stderr(stderr)) => { - let _ = session_handle.extended_data(channel_id, 1, stderr.data).await; + if let Some(lost_bytes) = output_gap( + &mut stderr_offset, + stderr.offset, + stderr.data.len(), + stderr.total_bytes, + ) { + let _ = session_handle + .extended_data( + channel_id, + 1, + output_gap_message("stderr", lost_bytes), + ) + .await; + } + if !stderr.data.is_empty() { + let _ = session_handle.extended_data(channel_id, 1, stderr.data).await; + } } None => {} }, + Some(Err(error)) => { + warn!(%error, %execution_id, "SSH output stream failed"); + terminate_process_group(server.clone(), execution_id.clone()).await; + finish_channel( + &session_handle, + channel_id, + completion, + ExitNotification::Status(INDETERMINATE_EXIT_STATUS), + ) + .await; + return; + } None => break, } } @@ -702,6 +753,46 @@ fn spawn_output_pump( }) } +fn output_gap( + expected_offset: &mut u64, + offset: Option, + data_len: usize, + total_bytes: Option, +) -> Option { + if let Some(total_bytes) = total_bytes { + if offset != Some(total_bytes) || total_bytes < *expected_offset { + warn!( + ?offset, + total_bytes, expected_offset, "invalid SSH execution output end frame" + ); + return None; + } + let lost_bytes = total_bytes - *expected_offset; + *expected_offset = total_bytes; + return (lost_bytes > 0).then_some(lost_bytes); + } + + let Some(offset) = offset else { + *expected_offset += data_len as u64; + return None; + }; + if offset < *expected_offset { + warn!( + offset, + expected_offset, "SSH execution output chunk moved backwards" + ); + return None; + } + + let lost_bytes = offset - *expected_offset; + *expected_offset = offset + data_len as u64; + (lost_bytes > 0).then_some(lost_bytes) +} + +fn output_gap_message(source: &str, lost_bytes: u64) -> Vec { + format!("[boxlite] {source} output dropped {lost_bytes} bytes\r\n").into_bytes() +} + /// Keep one lifecycle owner for every SSH-created execution. /// /// The output task may finish early because Attach failed or the SSH channel @@ -1046,6 +1137,17 @@ mod tests { assert!(!ChannelCompletion::Forwarding.sends_exit_notification()); } + #[test] + fn output_gap_detects_mid_stream_and_terminal_loss() { + let mut offset = 0; + assert_eq!(output_gap(&mut offset, Some(0), 3, None), None); + assert_eq!(offset, 3); + assert_eq!(output_gap(&mut offset, Some(5), 2, None), Some(2)); + assert_eq!(offset, 7); + assert_eq!(output_gap(&mut offset, Some(10), 0, Some(10)), Some(3)); + assert_eq!(offset, 10); + } + #[test] fn streamlocal_readiness_prefix_can_be_fragmented() { let magic = crate::service::ssh::streamlocal::STREAMLOCAL_READY_MAGIC; diff --git a/src/guest/src/service/ssh/reverse_streamlocal.rs b/src/guest/src/service/ssh/reverse_streamlocal.rs index 80412f783..5fe2e630e 100644 --- a/src/guest/src/service/ssh/reverse_streamlocal.rs +++ b/src/guest/src/service/ssh/reverse_streamlocal.rs @@ -513,7 +513,7 @@ struct RunningHelper { execution_id: String, stdin: mpsc::Sender, stdin_task: JoinHandle<()>, - output: mpsc::Receiver, + output: mpsc::Receiver>, buffered_stdout: Vec, } @@ -765,10 +765,10 @@ fn spawn_listener( } output = helper.output.recv() => { match output { - Some(ExecOutput { event: Some(exec_output::Event::Stderr(stderr)) }) => { + Some(Ok(ExecOutput { event: Some(exec_output::Event::Stderr(stderr)) })) => { debug!(bytes = stderr.data.len(), "reverse streamlocal helper stderr"); } - Some(ExecOutput { event: Some(exec_output::Event::Stdout(stdout)) }) => { + Some(Ok(ExecOutput { event: Some(exec_output::Event::Stdout(stdout)) })) => { if !append_stopped_marker_prefix( &mut helper.buffered_stdout, &stdout.data, @@ -777,7 +777,11 @@ fn spawn_listener( break End::HelperEnded; } } - Some(ExecOutput { event: None }) => {} + Some(Ok(ExecOutput { event: None })) => {} + Some(Err(error)) => { + warn!(%error, "reverse streamlocal helper output failed"); + break End::HelperEnded; + } None => break End::HelperEnded, } } @@ -923,7 +927,7 @@ impl MarkerParser { } async fn read_marker( - output: &mut mpsc::Receiver, + output: &mut mpsc::Receiver>, marker: &[u8], initial_stdout: Vec, max_trailing_bytes: usize, @@ -936,9 +940,13 @@ async fn read_marker( } } loop { - let message = output.recv().await.ok_or_else(|| { - "reverse streamlocal helper exited before control marker".to_string() - })?; + let message = output + .recv() + .await + .ok_or_else(|| { + "reverse streamlocal helper exited before control marker".to_string() + })? + .map_err(|error| format!("reverse streamlocal helper output failed: {error}"))?; match message.event { Some(exec_output::Event::Stdout(stdout)) => { if let Some(buffered) = parser.push(marker, &stdout.data, max_trailing_bytes)? { @@ -976,7 +984,7 @@ fn spawn_failed_helper_cleanup( server: Arc, registry: ExecutionRegistry, execution_id: String, - output: Option>, + output: Option>>, stdin_task: Option>, ) { spawn_execution_cleanup(server, registry, execution_id, output, stdin_task, true); @@ -986,7 +994,7 @@ fn spawn_execution_cleanup( server: Arc, registry: ExecutionRegistry, execution_id: String, - output: Option>, + output: Option>>, stdin_task: Option>, force_termination: bool, ) { @@ -994,6 +1002,9 @@ fn spawn_execution_cleanup( let output_task = output.map(|mut output| { tokio::spawn(async move { while let Some(message) = output.recv().await { + let Ok(message) = message else { + break; + }; if let Some(exec_output::Event::Stderr(stderr)) = message.event { debug!( bytes = stderr.data.len(), From b5d053eeca3112b93b33a06494d229ab8cbcd7fb Mon Sep 17 00:00:00 2001 From: BatmanByte <300328404+BatmanByte@users.noreply.github.com> Date: Thu, 30 Jul 2026 01:25:29 +0800 Subject: [PATCH 10/12] fix(exec): drain buffered output before attach failure --- src/guest/src/service/exec/output.rs | 81 +++++++++++++++++----------- 1 file changed, 50 insertions(+), 31 deletions(-) diff --git a/src/guest/src/service/exec/output.rs b/src/guest/src/service/exec/output.rs index 68d96b11e..d8b35d297 100644 --- a/src/guest/src/service/exec/output.rs +++ b/src/guest/src/service/exec/output.rs @@ -26,12 +26,17 @@ struct OutputState { buffered_bytes: usize, oldest_sequence: u64, next_sequence: u64, - failure: Option, + failure: Option, attached: bool, stdout: StreamState, stderr: StreamState, } +struct ReaderFailure { + sequence: u64, + message: String, +} + struct OutputEntry { sequence: u64, output: ExecOutput, @@ -133,35 +138,37 @@ impl OutputManager { let next = { let state = manager.inner.lock().await; - if let Some(failure) = &state.failure { - Next::Error(Status::internal(failure.clone())) + if next_sequence < state.oldest_sequence { + next_sequence = state.oldest_sequence; + } + let failure = state + .failure + .as_ref() + .filter(|failure| next_sequence >= failure.sequence); + if let Some(failure) = failure { + Next::Error(Status::internal(failure.message.clone())) + } else if state.stdout.ready_to_end(next_sequence) && !stdout_end_sent { + stdout_end_sent = true; + Next::Item(end_output(OutputSource::Stdout, state.stdout.total_bytes)) + } else if state.stderr.ready_to_end(next_sequence) && !stderr_end_sent { + stderr_end_sent = true; + Next::Item(end_output(OutputSource::Stderr, state.stderr.total_bytes)) + } else if next_sequence < state.next_sequence { + let index = (next_sequence - state.oldest_sequence) as usize; + let output = state + .entries + .get(index) + .expect("ring sequence must exist") + .output + .clone(); + next_sequence += 1; + Next::Item(output) + } else if (!state.stdout.enabled || stdout_end_sent) + && (!state.stderr.enabled || stderr_end_sent) + { + Next::Done } else { - if next_sequence < state.oldest_sequence { - next_sequence = state.oldest_sequence; - } - if state.stdout.ready_to_end(next_sequence) && !stdout_end_sent { - stdout_end_sent = true; - Next::Item(end_output(OutputSource::Stdout, state.stdout.total_bytes)) - } else if state.stderr.ready_to_end(next_sequence) && !stderr_end_sent { - stderr_end_sent = true; - Next::Item(end_output(OutputSource::Stderr, state.stderr.total_bytes)) - } else if next_sequence < state.next_sequence { - let index = (next_sequence - state.oldest_sequence) as usize; - let output = state - .entries - .get(index) - .expect("ring sequence must exist") - .output - .clone(); - next_sequence += 1; - Next::Item(output) - } else if (!state.stdout.enabled || stdout_end_sent) - && (!state.stderr.enabled || stderr_end_sent) - { - Next::Done - } else { - Next::Wait - } + Next::Wait } }; @@ -256,7 +263,10 @@ impl OutputManager { let mut state = self.inner.lock().await; let stream = state.stream_mut(source); stream.finished = true; - state.failure = Some(format!("failed to read {source:?}: {error}")); + state.failure = Some(ReaderFailure { + sequence: state.next_sequence, + message: format!("failed to read {source:?}: {error}"), + }); drop(state); self.updated.send_replace(()); } @@ -406,8 +416,11 @@ mod tests { } #[tokio::test] - async fn reader_failure_is_reported_to_attach() { + async fn reader_failure_follows_buffered_output() { let manager = OutputManager::new(None, None); + manager + .push(OutputSource::Stdout, b"before failure".to_vec()) + .await; manager .reader_failed( OutputSource::Stdout, @@ -416,6 +429,12 @@ mod tests { .await; let mut output = manager.attach().await.unwrap(); + let first = output.next().await.unwrap().unwrap(); + let Some(exec_output::Event::Stdout(first)) = first.event else { + panic!("buffered stdout must arrive before the reader failure"); + }; + assert_eq!(first.data, b"before failure"); + let error = output.next().await.unwrap().unwrap_err(); assert_eq!(error.code(), tonic::Code::Internal); assert!(error.message().contains("simulated pipe failure")); From 93cb54a6bf85b433bed2617a960c0fdc35dcf73c Mon Sep 17 00:00:00 2001 From: BatmanByte <300328404+BatmanByte@users.noreply.github.com> Date: Thu, 30 Jul 2026 10:45:55 +0800 Subject: [PATCH 11/12] fix(exec): preserve first output failure --- src/guest/src/service/exec/output.rs | 46 +++++++-- src/guest/src/service/ssh/bridge.rs | 143 ++++++++++++++++++++++----- 2 files changed, 159 insertions(+), 30 deletions(-) diff --git a/src/guest/src/service/exec/output.rs b/src/guest/src/service/exec/output.rs index d8b35d297..47c2ac870 100644 --- a/src/guest/src/service/exec/output.rs +++ b/src/guest/src/service/exec/output.rs @@ -261,12 +261,13 @@ impl OutputManager { async fn reader_failed(&self, source: OutputSource, error: std::io::Error) { let mut state = self.inner.lock().await; - let stream = state.stream_mut(source); - stream.finished = true; - state.failure = Some(ReaderFailure { - sequence: state.next_sequence, - message: format!("failed to read {source:?}: {error}"), - }); + state.stream_mut(source).finished = true; + if state.failure.is_none() { + state.failure = Some(ReaderFailure { + sequence: state.next_sequence, + message: format!("failed to read {source:?}: {error}"), + }); + } drop(state); self.updated.send_replace(()); } @@ -439,4 +440,37 @@ mod tests { assert_eq!(error.code(), tonic::Code::Internal); assert!(error.message().contains("simulated pipe failure")); } + + #[tokio::test] + async fn first_reader_failure_stops_output_before_later_reader_failure() { + let manager = OutputManager::new(None, None); + manager + .push(OutputSource::Stdout, b"before failure".to_vec()) + .await; + manager + .reader_failed( + OutputSource::Stdout, + std::io::Error::other("stdout failure"), + ) + .await; + manager + .push(OutputSource::Stderr, b"after failure".to_vec()) + .await; + manager + .reader_failed( + OutputSource::Stderr, + std::io::Error::other("stderr failure"), + ) + .await; + + let mut output = manager.attach().await.unwrap(); + let first = output.next().await.unwrap().unwrap(); + let Some(exec_output::Event::Stdout(first)) = first.event else { + panic!("buffered stdout must arrive before the first reader failure"); + }; + assert_eq!(first.data, b"before failure"); + + let error = output.next().await.unwrap().unwrap_err(); + assert!(error.message().contains("stdout failure")); + } } diff --git a/src/guest/src/service/ssh/bridge.rs b/src/guest/src/service/ssh/bridge.rs index 5d021202a..796abb5b9 100644 --- a/src/guest/src/service/ssh/bridge.rs +++ b/src/guest/src/service/ssh/bridge.rs @@ -198,6 +198,7 @@ impl ChannelBridge { OutputPumpStart::AwaitActivation { output: ready.output, buffered_stdout: ready.buffered_stdout, + stdout_offset: ready.stdout_offset, activate_rx, }, Some(activate_tx), @@ -513,6 +514,7 @@ fn streamlocal_launch(container_id: &str, socket_path: String) -> InternalHelper struct PreparedStreamlocalOutput { output: mpsc::Receiver>, buffered_stdout: Vec, + stdout_offset: u64, } enum OutputPumpStart { @@ -520,6 +522,7 @@ enum OutputPumpStart { AwaitActivation { output: mpsc::Receiver>, buffered_stdout: Vec, + stdout_offset: u64, activate_rx: oneshot::Receiver<()>, }, } @@ -527,10 +530,16 @@ enum OutputPumpStart { #[derive(Default)] struct StreamlocalReadyParser { matched: usize, + stdout_offset: u64, } impl StreamlocalReadyParser { - fn push_stdout(&mut self, data: &[u8]) -> Result>, BridgeError> { + fn push_stdout( + &mut self, + offset: Option, + data: &[u8], + ) -> Result>, BridgeError> { + self.stdout_offset = offset.unwrap_or(self.stdout_offset) + data.len() as u64; let magic = crate::service::ssh::streamlocal::STREAMLOCAL_READY_MAGIC; let prefix_bytes = (magic.len() - self.matched).min(data.len()); if data[..prefix_bytes] != magic[self.matched..self.matched + prefix_bytes] { @@ -570,10 +579,13 @@ async fn await_streamlocal_ready( })?; match message.event { Some(boxlite_shared::exec_output::Event::Stdout(stdout)) => { - if let Some(buffered_stdout) = parser.push_stdout(&stdout.data)? { + if let Some(buffered_stdout) = + parser.push_stdout(stdout.offset, &stdout.data)? + { return Ok(PreparedStreamlocalOutput { output, buffered_stdout, + stdout_offset: parser.stdout_offset, }); } } @@ -610,7 +622,7 @@ fn spawn_output_pump( mut cancel_rx: oneshot::Receiver<()>, ) -> JoinHandle<()> { tokio::spawn(async move { - let (mut output, buffered_stdout) = match output_start { + let (mut output, buffered_stdout, mut stdout_offset) = match output_start { OutputPumpStart::Attach => { let attached = tokio::select! { _ = &mut cancel_rx => return, @@ -626,7 +638,7 @@ fn spawn_output_pump( Err(_) => Err(ExecutionError::Io("attach timed out".into())), }; match attached { - Ok(output) => (output, Vec::new()), + Ok(output) => (output, Vec::new(), 0), Err(error) => { warn!(%error, %execution_id, "SSH attach failed"); terminate_process_group(server.clone(), execution_id.clone()).await; @@ -644,6 +656,7 @@ fn spawn_output_pump( OutputPumpStart::AwaitActivation { output, buffered_stdout, + stdout_offset, mut activate_rx, } => { let activated = tokio::select! { @@ -653,7 +666,7 @@ fn spawn_output_pump( if activated.is_err() { return; } - (output, buffered_stdout) + (output, buffered_stdout, stdout_offset) } }; @@ -661,7 +674,6 @@ fn spawn_output_pump( let _ = session_handle.data(channel_id, buffered_stdout).await; } - let mut stdout_offset = 0; let mut stderr_offset = 0; loop { tokio::select! { @@ -680,12 +692,28 @@ fn spawn_output_pump( stdout.data.len(), stdout.total_bytes, ) { - let _ = session_handle - .data( - channel_id, - output_gap_message("stdout", lost_bytes), - ) - .await; + match output_gap_disposition(completion) { + OutputGapDisposition::Report => { + let _ = session_handle + .data( + channel_id, + output_gap_message("stdout", lost_bytes), + ) + .await; + } + OutputGapDisposition::Abort => { + warn!(%execution_id, lost_bytes, "SSH forwarding output gap"); + terminate_process_group(server.clone(), execution_id.clone()).await; + finish_channel( + &session_handle, + channel_id, + completion, + ExitNotification::Status(INDETERMINATE_EXIT_STATUS), + ) + .await; + return; + } + } } if !stdout.data.is_empty() { let _ = session_handle.data(channel_id, stdout.data).await; @@ -698,13 +726,29 @@ fn spawn_output_pump( stderr.data.len(), stderr.total_bytes, ) { - let _ = session_handle - .extended_data( - channel_id, - 1, - output_gap_message("stderr", lost_bytes), - ) - .await; + match output_gap_disposition(completion) { + OutputGapDisposition::Report => { + let _ = session_handle + .extended_data( + channel_id, + 1, + output_gap_message("stderr", lost_bytes), + ) + .await; + } + OutputGapDisposition::Abort => { + warn!(%execution_id, lost_bytes, "SSH forwarding output gap"); + terminate_process_group(server.clone(), execution_id.clone()).await; + finish_channel( + &session_handle, + channel_id, + completion, + ExitNotification::Status(INDETERMINATE_EXIT_STATUS), + ) + .await; + return; + } + } } if !stderr.data.is_empty() { let _ = session_handle.extended_data(channel_id, 1, stderr.data).await; @@ -789,6 +833,19 @@ fn output_gap( (lost_bytes > 0).then_some(lost_bytes) } +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +enum OutputGapDisposition { + Report, + Abort, +} + +fn output_gap_disposition(completion: ChannelCompletion) -> OutputGapDisposition { + match completion { + ChannelCompletion::Session => OutputGapDisposition::Report, + ChannelCompletion::Forwarding => OutputGapDisposition::Abort, + } +} + fn output_gap_message(source: &str, lost_bytes: u64) -> Vec { format!("[boxlite] {source} output dropped {lost_bytes} bytes\r\n").into_bytes() } @@ -1153,9 +1210,13 @@ mod tests { let magic = crate::service::ssh::streamlocal::STREAMLOCAL_READY_MAGIC; let mut parser = StreamlocalReadyParser::default(); - assert_eq!(parser.push_stdout(&magic[..3]).unwrap(), None); - assert_eq!(parser.push_stdout(&magic[3..11]).unwrap(), None); - assert_eq!(parser.push_stdout(&magic[11..]).unwrap(), Some(Vec::new())); + assert_eq!(parser.push_stdout(Some(0), &magic[..3]).unwrap(), None); + assert_eq!(parser.push_stdout(Some(3), &magic[3..11]).unwrap(), None); + assert_eq!( + parser.push_stdout(Some(11), &magic[11..]).unwrap(), + Some(Vec::new()) + ); + assert_eq!(parser.stdout_offset, magic.len() as u64); } #[test] @@ -1166,15 +1227,49 @@ mod tests { let mut parser = StreamlocalReadyParser::default(); assert_eq!( - parser.push_stdout(&frame).unwrap(), + parser.push_stdout(Some(0), &frame).unwrap(), Some(b"socket payload".to_vec()) ); + assert_eq!(parser.stdout_offset, frame.len() as u64); } #[test] fn streamlocal_readiness_rejects_non_helper_output() { let mut parser = StreamlocalReadyParser::default(); - assert!(parser.push_stdout(b"not readiness").is_err()); + assert!(parser.push_stdout(Some(0), b"not readiness").is_err()); + } + + #[test] + fn streamlocal_ready_prefix_does_not_create_an_output_gap() { + let magic = crate::service::ssh::streamlocal::STREAMLOCAL_READY_MAGIC; + let mut parser = StreamlocalReadyParser::default(); + assert_eq!( + parser.push_stdout(Some(0), magic).unwrap(), + Some(Vec::new()) + ); + + let mut stdout_offset = parser.stdout_offset; + assert_eq!( + output_gap( + &mut stdout_offset, + Some(magic.len() as u64), + b"socket payload".len(), + None, + ), + None + ); + } + + #[test] + fn forwarding_output_gaps_abort_instead_of_writing_to_the_byte_stream() { + assert_eq!( + output_gap_disposition(ChannelCompletion::Session), + OutputGapDisposition::Report + ); + assert_eq!( + output_gap_disposition(ChannelCompletion::Forwarding), + OutputGapDisposition::Abort + ); } #[test] From 213ef8b7052478a1f7caf78d62b12983028a68a9 Mon Sep 17 00:00:00 2001 From: BatmanByte <300328404+BatmanByte@users.noreply.github.com> Date: Thu, 30 Jul 2026 13:59:52 +0800 Subject: [PATCH 12/12] test(exec): read output gaps from their source stream --- src/boxlite/tests/run_main_command.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/boxlite/tests/run_main_command.rs b/src/boxlite/tests/run_main_command.rs index 0719beb2d..44e49a64d 100644 --- a/src/boxlite/tests/run_main_command.rs +++ b/src/boxlite/tests/run_main_command.rs @@ -156,9 +156,9 @@ async fn late_attach_reports_output_gap() { .attach(None) .await .expect("attach to the main command"); - let mut stderr = execution.stderr().expect("stderr stream"); + let mut stdout = execution.stdout().expect("stdout stream"); let dropped = tokio::time::timeout(std::time::Duration::from_secs(5), async { - while let Some(chunk) = stderr.next().await { + while let Some(chunk) = stdout.next().await { if chunk.contains("[boxlite] stdout output dropped") { return Some(chunk); }