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