From d250bca506c906e0bca9c32e29ca27d255bebf33 Mon Sep 17 00:00:00 2001 From: Federico Kamelhar Date: Wed, 30 Sep 2026 17:47:45 -0400 Subject: [PATCH] fix(cli): start sandbox exec without waiting for piped stdin EOF With a non-terminal stdin, sandbox exec read stdin to EOF before it sent the exec request. A pipe that never closes (CI runners, supervisors, agent harnesses) blocked the CLI forever in read(2) without the gateway ever seeing the request, and a slow producer delayed the command until EOF. Collect piped stdin on a detached reader thread for at most 200 ms. Input that reaches EOF within that window still travels in the single request that older gateways need. If the pipe is still open, start the command through the streaming RPC and forward the collected prefix plus the rest of stdin as it arrives, closing remote stdin at EOF. The 4 MiB cap covers the prefix and the streamed remainder together. Closes #3993 Signed-off-by: Federico Kamelhar --- crates/openshell-cli/src/run.rs | 307 +++++++++++++++++++---- docs/how-it-works/sandboxes/overview.mdx | 8 +- 2 files changed, 264 insertions(+), 51 deletions(-) diff --git a/crates/openshell-cli/src/run.rs b/crates/openshell-cli/src/run.rs index 02c806c421..c9dd405f8b 100644 --- a/crates/openshell-cli/src/run.rs +++ b/crates/openshell-cli/src/run.rs @@ -1852,6 +1852,86 @@ fn local_terminal_size() -> Option<(u32, u32)> { const MAX_EXEC_REQUEST_BYTES: usize = 1024 * 1024; const MAX_EXEC_STDIN_BYTES: usize = 4 * 1024 * 1024; +/// How long `sandbox exec` waits for piped stdin to reach EOF before it starts +/// the command and streams the rest of the input as it arrives. +/// +/// Small pipes such as `echo x | openshell sandbox exec ...` close well within +/// this window and keep using the single-request path that older gateways +/// need. A pipe that stays open (a CI runner, a supervisor, a harness that +/// never closes stdin) must not block the command: after the grace period the +/// command starts and stdin is forwarded until EOF or until the command exits. +const EXEC_STDIN_UNARY_GRACE: Duration = Duration::from_millis(200); + +/// Piped stdin as collected by [`collect_piped_stdin`]. +enum PipedStdin { + /// stdin reached EOF within the grace period; `prefix` holds all of it. + Complete(Vec), + /// stdin is still open. `prefix` is what arrived so far; `rest` delivers + /// the remaining chunks until EOF. + Open { + prefix: Vec, + rest: tokio::sync::mpsc::Receiver>>, + }, +} + +/// Read a pipe on a detached OS thread and hand its bytes over in chunks. +/// +/// A plain `std::thread` rather than `spawn_blocking`, so runtime shutdown +/// never waits on a thread parked in `read(2)`. The thread exits at EOF, on a +/// read error, or when the receiver is dropped. +fn spawn_piped_stdin_reader( + mut reader: impl Read + Send + 'static, +) -> tokio::sync::mpsc::Receiver>> { + let (tx, rx) = tokio::sync::mpsc::channel::>>(64); + std::thread::spawn(move || { + let mut buf = [0u8; 4096]; + loop { + match reader.read(&mut buf) { + Ok(0) => return, + Err(error) if error.kind() == ErrorKind::Interrupted => {} + Err(error) => { + let _ = tx.blocking_send(Err(error)); + return; + } + Ok(n) => { + if tx.blocking_send(Ok(buf[..n].to_vec())).is_err() { + return; + } + } + } + } + }); + rx +} + +/// Collect piped stdin until EOF or until `grace` elapses, whichever comes +/// first. Input beyond `limit` bytes is rejected with the upload hint. +async fn collect_piped_stdin( + mut rx: tokio::sync::mpsc::Receiver>>, + grace: Duration, + limit: usize, +) -> Result { + let deadline = tokio::time::Instant::now() + grace; + let mut prefix = Vec::new(); + loop { + match tokio::time::timeout_at(deadline, rx.recv()).await { + Ok(Some(Ok(chunk))) => { + prefix.extend_from_slice(&chunk); + if prefix.len() > limit { + return Err(piped_stdin_limit_error()); + } + } + Ok(Some(Err(error))) => return Err(error).into_diagnostic(), + Ok(None) => return Ok(PipedStdin::Complete(prefix)), + Err(_elapsed) => return Ok(PipedStdin::Open { prefix, rest: rx }), + } + } +} + +fn piped_stdin_limit_error() -> miette::Report { + miette::miette!("piped stdin exceeds the 4 MiB limit; use `sandbox upload` for larger input") +} + /// Execute a command in a running sandbox via gRPC, streaming output to the terminal. /// /// Returns the remote command's exit code, or an error if the event stream @@ -1902,24 +1982,24 @@ pub async fn sandbox_exec_grpc( // interactive RPC closes the SSH channel when stdin reaches EOF. Retain // the existing 4 MiB input cap because the supervisor's process stdin // queue is unbounded; larger input should use file upload instead. - let stdin_prefix = if stdin_is_terminal { - Vec::new() + // + // Never block on stdin EOF before starting the command: a pipe that stays + // open (CI runners, harnesses) would otherwise hang the exec forever + // without the gateway ever seeing the request. After a short grace period + // the command starts and the remaining input streams until EOF. + let (stdin_prefix, stdin_rest) = if stdin_is_terminal { + (Vec::new(), None) } else { - tokio::task::spawn_blocking(|| { - let mut prefix = Vec::new(); - std::io::stdin() - .take((MAX_EXEC_STDIN_BYTES + 1) as u64) - .read_to_end(&mut prefix) - .into_diagnostic()?; - if prefix.len() > MAX_EXEC_STDIN_BYTES { - return Err(miette::miette!( - "piped stdin exceeds the 4 MiB limit; use `sandbox upload` for larger input" - )); - } - Ok::<_, miette::Report>(prefix) - }) - .await - .into_diagnostic()?? + match collect_piped_stdin( + spawn_piped_stdin_reader(std::io::stdin()), + EXEC_STDIN_UNARY_GRACE, + MAX_EXEC_STDIN_BYTES, + ) + .await? + { + PipedStdin::Complete(prefix) => (prefix, None), + PipedStdin::Open { prefix, rest } => (prefix, Some(rest)), + } }; let (cols, rows) = if tty { @@ -1952,7 +2032,10 @@ pub async fn sandbox_exec_grpc( "exec command or environment exceeds the gateway's 1 MiB message limit" )); } - if (tty && stdin_is_terminal) || request.encoded_len() > MAX_EXEC_REQUEST_BYTES { + if (tty && stdin_is_terminal) + || stdin_rest.is_some() + || request.encoded_len() > MAX_EXEC_REQUEST_BYTES + { return sandbox_exec_streaming_grpc( client, &sandbox, @@ -1964,6 +2047,7 @@ pub async fn sandbox_exec_grpc( tty, stdin_is_terminal, std::mem::take(&mut request.stdin), + stdin_rest, ) .await; } @@ -2350,6 +2434,7 @@ async fn sandbox_exec_streaming_grpc( tty: bool, stdin_is_terminal: bool, stdin_prefix: Vec, + stdin_rest: Option>>>, ) -> Result { #[cfg(unix)] use openshell_core::proto::ExecSandboxWindowResize; @@ -2407,42 +2492,88 @@ async fn sandbox_exec_streaming_grpc( // closes (blocking_send returns Err) or stdin hits EOF. let stdin_tx = input_tx.clone(); let (stdin_result_tx, mut stdin_result_rx) = tokio::sync::oneshot::channel(); - std::thread::spawn(move || { - let mut stdin = std::io::stdin().lock(); - let mut buf = [0u8; 4096]; - let result = (|| { - for chunk in stdin_prefix.chunks(buf.len()) { - if stdin_tx - .blocking_send(ExecSandboxInput { - payload: Some(exec_sandbox_input::Payload::Stdin(chunk.to_vec())), - }) - .is_err() - { - return Ok(()); + if let Some(mut rest) = stdin_rest { + // Piped stdin that was still open when the command started: the + // reader thread from `spawn_piped_stdin_reader` already owns stdin, + // so forward its chunks here after the collected prefix. The 4 MiB + // cap covers the prefix and the streamed remainder together. + tokio::spawn(async move { + let mut forwarded = 0usize; + let result = async { + for chunk in stdin_prefix.chunks(4096) { + forwarded += chunk.len(); + if stdin_tx + .send(ExecSandboxInput { + payload: Some(exec_sandbox_input::Payload::Stdin(chunk.to_vec())), + }) + .await + .is_err() + { + return Ok(()); + } } + while let Some(chunk) = rest.recv().await { + let chunk = chunk?; + forwarded += chunk.len(); + if forwarded > MAX_EXEC_STDIN_BYTES { + return Err(std::io::Error::new( + ErrorKind::InvalidInput, + piped_stdin_limit_error().to_string(), + )); + } + if stdin_tx + .send(ExecSandboxInput { + payload: Some(exec_sandbox_input::Payload::Stdin(chunk)), + }) + .await + .is_err() + { + return Ok(()); + } + } + Ok(()) } - loop { - match stdin.read(&mut buf) { - Ok(0) => return Ok(()), - Err(error) if error.kind() == ErrorKind::Interrupted => {} - Err(error) => return Err(error), - Ok(n) => { - if stdin_tx - .blocking_send(ExecSandboxInput { - payload: Some(exec_sandbox_input::Payload::Stdin( - buf[..n].to_vec(), - )), - }) - .is_err() - { - return Ok(()); + .await; + let _ = stdin_result_tx.send(result); + }); + } else { + std::thread::spawn(move || { + let mut stdin = std::io::stdin().lock(); + let mut buf = [0u8; 4096]; + let result = (|| { + for chunk in stdin_prefix.chunks(buf.len()) { + if stdin_tx + .blocking_send(ExecSandboxInput { + payload: Some(exec_sandbox_input::Payload::Stdin(chunk.to_vec())), + }) + .is_err() + { + return Ok(()); + } + } + loop { + match stdin.read(&mut buf) { + Ok(0) => return Ok(()), + Err(error) if error.kind() == ErrorKind::Interrupted => {} + Err(error) => return Err(error), + Ok(n) => { + if stdin_tx + .blocking_send(ExecSandboxInput { + payload: Some(exec_sandbox_input::Payload::Stdin( + buf[..n].to_vec(), + )), + }) + .is_err() + { + return Ok(()); + } } } } - } - })(); - let _ = stdin_result_tx.send(result); - }); + })(); + let _ = stdin_result_tx.send(result); + }); + } // SIGWINCH handler: forward terminal resize events. #[cfg(unix)] @@ -8206,4 +8337,82 @@ mod tests { let log = log_line("OCSF", "ocsf", message, "sandbox", &[]); assert!(format_log_line(&log).ends_with(message)); } + + use std::io::Write as _; + use std::time::{Duration, Instant}; + + fn exec_stdin_runtime() -> tokio::runtime::Runtime { + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("runtime") + } + + #[test] + fn piped_stdin_that_closes_quickly_goes_in_one_request() { + let (reader, mut writer) = std::io::pipe().expect("pipe"); + writer.write_all(b"hi\n").unwrap(); + drop(writer); + let collected = exec_stdin_runtime().block_on(super::collect_piped_stdin( + super::spawn_piped_stdin_reader(reader), + Duration::from_secs(5), + super::MAX_EXEC_STDIN_BYTES, + )); + match collected.expect("collect") { + super::PipedStdin::Complete(prefix) => assert_eq!(prefix, b"hi\n"), + super::PipedStdin::Open { .. } => panic!("closed pipe must complete"), + } + } + + #[test] + fn piped_stdin_that_stays_open_does_not_block_the_command() { + let (reader, mut writer) = std::io::pipe().expect("pipe"); + writer.write_all(b"early").unwrap(); + let grace = Duration::from_millis(100); + let started = Instant::now(); + let runtime = exec_stdin_runtime(); + let collected = runtime.block_on(super::collect_piped_stdin( + super::spawn_piped_stdin_reader(reader), + grace, + super::MAX_EXEC_STDIN_BYTES, + )); + let elapsed = started.elapsed(); + assert!(elapsed >= grace, "must wait the grace period: {elapsed:?}"); + assert!( + elapsed < Duration::from_secs(3), + "must not wait for EOF: {elapsed:?}" + ); + let super::PipedStdin::Open { prefix, mut rest } = collected.expect("collect") else { + panic!("an open pipe must start the command before EOF"); + }; + assert_eq!(prefix, b"early"); + // The remainder keeps flowing after the command has started. + writer.write_all(b"late").unwrap(); + drop(writer); + let next = runtime + .block_on(rest.recv()) + .expect("late chunk") + .expect("read"); + assert_eq!(next, b"late"); + assert!( + runtime.block_on(rest.recv()).is_none(), + "EOF closes the channel" + ); + } + + #[test] + fn piped_stdin_over_the_limit_is_rejected_with_the_upload_hint() { + let (reader, mut writer) = std::io::pipe().expect("pipe"); + writer.write_all(&[0u8; 8]).unwrap(); + drop(writer); + let error = exec_stdin_runtime() + .block_on(super::collect_piped_stdin( + super::spawn_piped_stdin_reader(reader), + Duration::from_secs(5), + 4, + )) + .err() + .expect("over the limit must fail"); + assert!(error.to_string().contains("sandbox upload"), "{error}"); + } } diff --git a/docs/how-it-works/sandboxes/overview.mdx b/docs/how-it-works/sandboxes/overview.mdx index 8194b0df5b..db157750b5 100644 --- a/docs/how-it-works/sandboxes/overview.mdx +++ b/docs/how-it-works/sandboxes/overview.mdx @@ -325,8 +325,12 @@ The CLI sends small piped input in one request for compatibility with older gateways. It streams larger input in bounded frames, including for commands without a TTY, so input is not limited by the gateway's per-message request size. Piped input is limited to 4 MiB; use `sandbox upload` for larger files. -The CLI closes remote stdin when the pipe reaches EOF and continues -reading command output until the command finishes. +The CLI waits at most 200 ms for piped stdin to reach EOF. If the pipe is +still open after that, for example under a CI runner or a harness that never +closes stdin, the command starts anyway and the remaining input streams to it +as it arrives. The CLI closes remote stdin when the pipe reaches EOF and +continues reading command output until the command finishes. Redirect stdin +from `/dev/null` when a command needs no input. The command's exit code is propagated to the CLI, so `exec` works in scripts that check return codes. Large stdout and stderr streams are delivered before a successful exit. A slow