diff --git a/.config/nextest.toml b/.config/nextest.toml index 0580f3f5a..c55b0bd0d 100644 --- a/.config/nextest.toml +++ b/.config/nextest.toml @@ -15,6 +15,11 @@ leak-timeout = "500ms" filter = "package(fabro-workflow)" slow-timeout = { period = "2s", terminate-after = 3 } + # Real descendant regressions include bounded reaping and process probes. + # Leave room for their own watchdogs to run fail-safe fixture cleanup. + [[profile.default.overrides]] + filter = "package(fabro-proc) & binary(lifecycle)" + slow-timeout = { period = "10s", terminate-after = 3 } [profile.e2e] # E2E (ignored) tests: flag SLOW after 10s, hard-kill after 30s diff --git a/Cargo.lock b/Cargo.lock index 19f57246b..cbd70bee2 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2919,6 +2919,10 @@ dependencies = [ "cc", "libc", "tempfile", + "thiserror 2.0.18", + "tokio", + "tokio-util", + "tracing", ] [[package]] diff --git a/lib/apps/fabro-server/src/git_checkout.rs b/lib/apps/fabro-server/src/git_checkout.rs index 9a2e8c6f0..d2856be42 100644 --- a/lib/apps/fabro-server/src/git_checkout.rs +++ b/lib/apps/fabro-server/src/git_checkout.rs @@ -4,10 +4,12 @@ use std::time::Duration; use base64::Engine as _; use base64::engine::general_purpose::STANDARD as BASE64_STANDARD; +use fabro_proc::ProcessError; use fabro_store::KeyedMutex; use fabro_types::{GitHubRepositorySlug, GitRunTarget}; +use tokio::fs; use tokio::process::Command; -use tokio::{fs, time}; +use tokio_util::sync::CancellationToken; const GIT_CLONE_TIMEOUT: Duration = Duration::from_mins(2); const GIT_FETCH_TIMEOUT: Duration = Duration::from_mins(1); @@ -483,18 +485,28 @@ async fn run_git_plan(plan: GitCommandPlan) -> Result, GitCommandError> if let Some(current_dir) = plan.current_dir.as_ref() { command.current_dir(current_dir); } - command.kill_on_drop(true); - - let output = time::timeout(plan.timeout, command.output()) - .await - .map_err(|_| GitCommandError::Timeout { + let output = fabro_proc::capture( + &mut command, + Some(plan.timeout), + &CancellationToken::new(), + None, + ) + .await + .map_err(|err| match err { + ProcessError::TimedOut => GitCommandError::Timeout { command: safe_command_label(&plan), timeout_secs: plan.timeout.as_secs(), - })? - .map_err(|err| GitCommandError::Spawn { + }, + ProcessError::Io(source) => GitCommandError::Spawn { command: safe_command_label(&plan), - source: err, - })?; + source, + }, + ProcessError::Cancelled => GitCommandError::Spawn { + command: safe_command_label(&plan), + source: std::io::Error::new(std::io::ErrorKind::Interrupted, err), + }, + })? + .output; if output.status.success() { return Ok(output.stdout); @@ -546,13 +558,107 @@ mod tests { reason = "Git checkout unit tests build local git repositories and inspect temp files synchronously." )] + use std::error::Error; use std::fs; use std::path::Path; use tempfile::TempDir; + use tokio::{task, time}; use super::*; + fn shell_plan(script: &str, directory: &Path) -> GitCommandPlan { + GitCommandPlan { + program: "sh".to_string(), + args: vec!["-c".to_string(), script.to_string()], + env: vec![ + ("HOME".to_string(), directory.display().to_string()), + ("GIT_CONFIG_NOSYSTEM".to_string(), "1".to_string()), + ("GIT_CONFIG_GLOBAL".to_string(), "/dev/null".to_string()), + ], + current_dir: Some(directory.to_path_buf()), + timeout: Duration::from_secs(5), + sensitive_values: vec![], + } + } + + #[tokio::test] + async fn git_plan_capture_preserves_bytes_environment_and_diagnostics() { + let temp = TempDir::new().unwrap(); + let mut plan = shell_plan( + "printf '%s' \"$TEST_CAPTURE_VALUE\"; head -c 200000 /dev/zero; printf '\\377'", + temp.path(), + ); + plan.env.push(( + "TEST_CAPTURE_VALUE".to_string(), + "literal $(command)".to_string(), + )); + let output = run_git_plan(plan).await.unwrap(); + assert!(output.starts_with(b"literal $(command)")); + assert_eq!(output.len(), 200_019); + assert_eq!(output.last(), Some(&255)); + + let mut plan = shell_plan( + "printf stdout; printf '%s' \"$TEST_CAPTURE_VALUE\" >&2; exit 7", + temp.path(), + ); + let secret = "fixture-secret-value"; + plan.env + .push(("TEST_CAPTURE_VALUE".to_string(), secret.to_string())); + plan.sensitive_values.push(secret.to_string()); + let error = run_git_plan(plan).await.unwrap_err(); + let message = error.to_string(); + assert!(!message.contains(secret)); + assert!(message.ends_with("REDACTED")); + + let mut plan = shell_plan("exit 0", temp.path()); + plan.program = "/nonexistent/fabro-test-git".to_string(); + let error = run_git_plan(plan).await.unwrap_err(); + assert!(Error::source(&error).unwrap().is::()); + } + + #[cfg(unix)] + #[tokio::test] + async fn git_plan_deadline_includes_held_pipe() { + struct Cleanup(TempDir); + impl Drop for Cleanup { + fn drop(&mut self) { + if let Some(pid) = fs::read_to_string(self.0.path().join("leader.pid")) + .ok() + .and_then(|value| value.trim().parse().ok()) + { + fabro_proc::sigkill_process_group(pid); + } + } + } + let fixture = Cleanup(TempDir::new().unwrap()); + let mut plan = shell_plan( + "echo $$ > leader.pid; sleep 60 & echo $! > helper.pid; exit 0", + fixture.0.path(), + ); + plan.timeout = Duration::from_millis(200); + let error = time::timeout(Duration::from_secs(8), run_git_plan(plan)) + .await + .unwrap() + .unwrap_err(); + assert!(matches!(error, GitCommandError::Timeout { .. })); + let helper: u32 = fs::read_to_string(fixture.0.path().join("helper.pid")) + .unwrap() + .trim() + .parse() + .unwrap(); + time::timeout(Duration::from_secs(5), async { + while task::spawn_blocking(move || fabro_proc::process_running_strict(helper)) + .await + .unwrap() + { + time::sleep(Duration::from_millis(20)).await; + } + }) + .await + .unwrap(); + } + fn repository_slug(value: &str) -> GitHubRepositorySlug { GitHubRepositorySlug::try_new(value).expect("slug should parse") } diff --git a/lib/apps/fabro-server/src/run_manifest.rs b/lib/apps/fabro-server/src/run_manifest.rs index dbad49938..f671d11e4 100644 --- a/lib/apps/fabro-server/src/run_manifest.rs +++ b/lib/apps/fabro-server/src/run_manifest.rs @@ -17,6 +17,7 @@ use fabro_graphviz::render::apply_direction; use fabro_llm::FabroClient; use fabro_llm::lithos_catalog::Catalog; use fabro_llm::probe::{self, ModelTestStatus}; +use fabro_proc::ProcessError; use fabro_sandbox::{ CloneRequest, ProviderAccess, RunSandbox, SandboxSpec, sandbox_spec_for_environment, }; @@ -42,7 +43,9 @@ use fabro_workflow::workflow_bundle::{BundledWorkflow, ParsedWorkflowConfig, Wor use futures_util::stream::{self, StreamExt}; use lithos_llm::catalog::ProviderId; use tokio::process::Command; +#[cfg(test)] use tokio::time; +use tokio_util::sync::CancellationToken; use crate::run_compiler; use crate::server::AppState; @@ -882,13 +885,19 @@ async fn check_git_remote_ref( /// failure to its most useful message: stderr, then stdout, then the exit /// status. async fn run_ls_remote(mut command: Command) -> std::result::Result<(), String> { - // Dropping a timed-out `Command::output` future does not stop the child - // unless kill-on-drop is enabled. - command.kill_on_drop(true); - let output = time::timeout(Duration::from_secs(10), command.output()) - .await - .map_err(|_| "git ls-remote timed out after 10s".to_string())? - .map_err(|err| format!("Failed to run git ls-remote: {err}"))?; + let output = fabro_proc::capture( + &mut command, + Some(Duration::from_secs(10)), + &CancellationToken::new(), + None, + ) + .await + .map_err(|err| match err { + ProcessError::TimedOut => "git ls-remote timed out after 10s".to_string(), + ProcessError::Io(source) => format!("Failed to run git ls-remote: {source}"), + ProcessError::Cancelled => "git ls-remote cancelled".to_string(), + })? + .output; if output.status.success() { return Ok(()); @@ -1661,6 +1670,32 @@ mod tests { use super::*; + #[tokio::test] + async fn ls_remote_capture_keeps_diagnostic_precedence_and_unlimited_output() { + for (script, expected) in [ + ("printf stdout; printf stderr >&2; exit 7", "stderr"), + ("printf stdout; exit 7", "stdout"), + ] { + let mut command = Command::new("sh"); + command.args(["-c", script]); + assert_eq!(super::run_ls_remote(command).await.unwrap_err(), expected); + } + let mut command = Command::new("sh"); + command.args(["-c", "head -c 200000 /dev/zero | tr '\\0' x >&2; exit 7"]); + assert_eq!( + super::run_ls_remote(command).await.unwrap_err().len(), + 200_000 + ); + let mut command = Command::new("sh"); + command.args(["-c", "exit 7"]); + assert!( + super::run_ls_remote(command) + .await + .unwrap_err() + .starts_with("git ls-remote exited with status") + ); + } + fn minimal_manifest() -> types::RunManifest { types::RunManifest { args: None, diff --git a/lib/foundation/fabro-proc/Cargo.toml b/lib/foundation/fabro-proc/Cargo.toml index 4ecdb8ab8..47335363e 100644 --- a/lib/foundation/fabro-proc/Cargo.toml +++ b/lib/foundation/fabro-proc/Cargo.toml @@ -4,11 +4,17 @@ edition.workspace = true version.workspace = true publish = false license.workspace = true -description = "Safe wrappers for process management primitives (signals, pre-exec hooks, title rewriting)" +description = "Local subprocess ownership, capture, and OS process primitives" [lints] workspace = true +[dependencies] +tokio.workspace = true +tokio-util.workspace = true +thiserror.workspace = true +tracing.workspace = true + [target.'cfg(unix)'.dependencies] libc = "0.2" diff --git a/lib/foundation/fabro-proc/src/lib.rs b/lib/foundation/fabro-proc/src/lib.rs index 5fbadb7bf..7dcc9fe97 100644 --- a/lib/foundation/fabro-proc/src/lib.rs +++ b/lib/foundation/fabro-proc/src/lib.rs @@ -1,3 +1,7 @@ +//! Local subprocess ownership and capture, plus OS process primitives. +//! Finite commands use [`capture`]; command and credential policy stay with +//! callers. Process ownership remains private to capture. + #![allow( unsafe_code, reason = "This crate wraps low-level OS or FFI APIs that require unsafe code." @@ -24,3 +28,6 @@ pub use signal::{ sigkill, sigkill_process_group, sigterm, sigterm_process_group, sigusr1, sigusr2, }; pub use title::{init as title_init, set as title_set}; + +mod process; +pub use process::{CapturedOutput, ProcessError, capture}; diff --git a/lib/foundation/fabro-proc/src/process.rs b/lib/foundation/fabro-proc/src/process.rs new file mode 100644 index 000000000..7c476ee46 --- /dev/null +++ b/lib/foundation/fabro-proc/src/process.rs @@ -0,0 +1,280 @@ +//! Finite local commands: the leader and required output share one deadline. +//! +//! Capture and child waiting are scoped futures, never spawned. Cancellation +//! drops the readers and kills the owned group before reaping the direct child. +use std::future::{self, Future}; +use std::io; +use std::process::{ExitStatus, Output, Stdio}; +use std::time::Duration; + +use tokio::io::{AsyncRead, AsyncReadExt}; +use tokio::process::{Child, Command}; +use tokio::time; +use tokio_util::sync::CancellationToken; + +/// Infrastructure failure; command arguments, environment and output are +/// absent. +#[derive(thiserror::Error)] +pub enum ProcessError { + #[error("process operation cancelled")] + Cancelled, + #[error("process operation timed out")] + TimedOut, + #[error("process I/O failed")] + Io(#[from] io::Error), +} + +impl std::fmt::Debug for ProcessError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + std::fmt::Display::fmt(self, f) + } +} + +/// Owns a child and its original process group through output completion. +/// +/// On Unix, drop synchronously requests group KILL. Tokio's child reaper owns +/// the remaining direct-child wait (its orphan queue on Unix); no application +/// cleanup task or pipe task is detached. Orderly paths await reaping. Drop +/// reaping needs a live Tokio runtime and is best effort, not a bounded +/// promise. Non-Unix cleanup covers only the direct child. Escaped groups, +/// runtime/process destruction and uninterruptible OS failures are outside this +/// contract. +/// +/// Normal completion, including nonzero exit, preserves helpers that closed +/// inherited pipes. Dropping an unfinished owner always requests cleanup. +#[must_use] +struct ProcessOwner { + child: Child, + group: Option, + armed: bool, +} + +// Bounds direct-child reaping after immediate termination. Output readers +// are dropped when capture fails, so there is no separate drain allowance. +const CLEANUP_ALLOWANCE: Duration = Duration::from_secs(2); + +impl ProcessOwner { + /// Spawn a prepared command in a fresh group. All command policy stays with + /// the caller; only group setup and kill-on-drop are supplied here. + fn spawn(command: &mut Command) -> Result { + #[cfg(unix)] + command.process_group(0); + let child = command.kill_on_drop(true).spawn()?; + let group = child.id(); + tracing::debug!(pid = group, "Spawned owned local process"); + Ok(Self { + child, + group, + armed: true, + }) + } + + /// Wait for the leader and both readers under one deadline. Dropping the + /// completion future closes the pipes; the owner remains armed until the + /// command completes normally or termination and reaping succeed. + async fn complete( + mut self, + output: impl Future>, + timeout: Option, + cancel: &CancellationToken, + ) -> Result { + let deadline = async { + match timeout { + Some(duration) => time::sleep(duration).await, + None => future::pending().await, + } + }; + let result = tokio::select! { + result = async { + let (status, ()) = tokio::try_join!(self.child.wait(), output)?; + Ok(status) + } => result, + () = cancel.cancelled() => Err(ProcessError::Cancelled), + () = deadline => Err(ProcessError::TimedOut), + }; + match result { + Ok(status) => { + self.armed = false; + tracing::debug!("Local process and output completed"); + Ok(status) + } + Err(failure) => { + tracing::debug!(reason = %failure, "Stopping local process"); + if let Err(error) = self.terminate().await { + // Preserve an original I/O failure and leave drop's + // fallback armed. Cleanup failure is never called success. + tracing::warn!(error_kind = ?error.kind(), "Local process cleanup failed"); + return Err(match failure { + ProcessError::Io(_) => failure, + _ => ProcessError::Io(error), + }); + } + self.armed = false; + Err(failure) + } + } + } + + async fn terminate(&mut self) -> io::Result<()> { + #[cfg(unix)] + if let Some(group) = self.group { + kill_group(group)?; + } + self.child.start_kill()?; + time::timeout(CLEANUP_ALLOWANCE, self.child.wait()) + .await + .map_err(|_| io::Error::new(io::ErrorKind::TimedOut, "process reaping timed out"))??; + Ok(()) + } +} + +impl Drop for ProcessOwner { + fn drop(&mut self) { + if self.armed { + #[cfg(unix)] + if let Some(group) = self.group { + if let Err(error) = kill_group(group) { + tracing::warn!(error_kind = ?error.kind(), "Dropped local process group cleanup failed"); + } + } + // Also covers non-Unix and a failed group signal. Tokio retains + // direct-child reaping responsibility when Child is dropped. + let _ = self.child.start_kill(); + } + } +} + +#[cfg(unix)] +fn kill_group(group: u32) -> io::Result<()> { + let group = i32::try_from(group) + .ok() + .filter(|group| *group > 0) + .ok_or_else(|| { + io::Error::new(io::ErrorKind::InvalidInput, "invalid owned process group") + })?; + // SAFETY: positive saved child id names the fresh group established at spawn; + // negation targets that group, never the caller's group or all processes. + if unsafe { libc::kill(-group, libc::SIGKILL) } == 0 { + return Ok(()); + } + let error = io::Error::last_os_error(); + if error.raw_os_error() == Some(libc::ESRCH) { + Ok(()) + } else { + Err(error) + } +} + +/// Raw capture with per-stream completeness, retaining std's output shape. +/// Deliberately does not implement Debug: output may contain credentials. +pub struct CapturedOutput { + pub output: Output, + pub stdout_truncated: bool, + pub stderr_truncated: bool, +} + +/// Capture raw bytes with optional per-stream prefix retention. Reaching the +/// cap continues draining and is not process failure. `None` retains unlimited +/// bytes. Like Tokio Command::output, stdin configuration is preserved (a piped +/// stdin is closed before waiting); stdout/stderr are captured concurrently. +/// +/// Timeout/cancellation covers both leader exit and EOF. On failure, readers +/// close immediately, the owned group is killed, and direct-child reaping gets +/// up to two seconds. Dropping this future also requests group KILL; Tokio +/// owns eventual direct-child reaping while its runtime remains alive. On +/// non-Unix platforms only the direct child is killed. Normal completion, +/// including nonzero exit, preserves helpers that closed inherited pipes. +pub async fn capture( + command: &mut Command, + timeout: Option, + cancel: &CancellationToken, + prefix_cap: Option, +) -> Result { + command.stdout(Stdio::piped()).stderr(Stdio::piped()); + let mut owner = ProcessOwner::spawn(command)?; + let stdout = owner.child.stdout.take(); + let stderr = owner.child.stderr.take(); + let mut stdout_bytes = Vec::new(); + let mut stderr_bytes = Vec::new(); + let mut stdout_truncated = false; + let mut stderr_truncated = false; + let status = owner + .complete( + async { + tokio::try_join!( + drain(stdout, &mut stdout_bytes, &mut stdout_truncated, prefix_cap), + drain(stderr, &mut stderr_bytes, &mut stderr_truncated, prefix_cap), + )?; + Ok(()) + }, + timeout, + cancel, + ) + .await?; + Ok(CapturedOutput { + output: Output { + status, + stdout: stdout_bytes, + stderr: stderr_bytes, + }, + stdout_truncated, + stderr_truncated, + }) +} + +async fn drain( + reader: Option, + bytes: &mut Vec, + truncated: &mut bool, + cap: Option, +) -> io::Result<()> { + if let Some(mut reader) = reader { + let mut chunk = vec![0; 8192]; + loop { + let count = reader.read(&mut chunk).await?; + if count == 0 { + break; + } + let retained = count.min(cap.unwrap_or(usize::MAX).saturating_sub(bytes.len())); + bytes.extend_from_slice(&chunk[..retained]); + *truncated |= retained < count; + } + } + Ok(()) +} + +#[cfg(all(test, unix))] +mod tests { + use std::error::Error; + + use super::*; + + #[tokio::test] + async fn read_failure_preserves_source_and_reaps_child() { + let mut command = Command::new("sleep"); + command + .arg("60") + .stdout(Stdio::null()) + .stderr(Stdio::null()); + let owner = ProcessOwner::spawn(&mut command).unwrap(); + let pid = owner.child.id().unwrap(); + let output = async { + Err(io::Error::new( + io::ErrorKind::ConnectionReset, + "reader failed", + )) + }; + let error = owner + .complete( + output, + Some(Duration::from_secs(5)), + &CancellationToken::new(), + ) + .await + .unwrap_err(); + assert!(error.source().unwrap().is::()); + assert_eq!(error.to_string(), "process I/O failed"); + assert_eq!(format!("{error:?}"), "process I/O failed"); + assert!(!crate::process_exists(pid)); + } +} diff --git a/lib/foundation/fabro-proc/tests/lifecycle.rs b/lib/foundation/fabro-proc/tests/lifecycle.rs new file mode 100644 index 000000000..f439d2cc3 --- /dev/null +++ b/lib/foundation/fabro-proc/tests/lifecycle.rs @@ -0,0 +1,244 @@ +#![cfg(unix)] +use std::process::Stdio; +use std::time::Duration; + +use fabro_proc::{self as process, ProcessError}; +use tokio::process::Command; +use tokio::{fs, task, time}; +use tokio_util::sync::CancellationToken; + +struct Fixture(tempfile::TempDir); +impl Fixture { + fn new() -> Self { + Self(tempfile::tempdir().expect("fixture directory")) + } + fn command(&self, script: &str) -> Command { + let mut command = Command::new("sh"); + command.args(["-c", script]).current_dir(self.0.path()); + command + } + async fn ready(&self) { + time::timeout(Duration::from_secs(5), async { + loop { + if fs::read_to_string(self.0.path().join("helper.pid")) + .await + .ok() + .and_then(|s| s.trim().parse::().ok()) + .is_some() + { + break; + } + time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("helper readiness"); + } + async fn pid(&self, file: &str) -> u32 { + fs::read_to_string(self.0.path().join(file)) + .await + .expect("fixture process observation") + .trim() + .parse() + .expect("fixture process observation") + } + async fn stopped(&self) { + for name in ["leader.pid", "helper.pid"] { + let pid = self.pid(name).await; + time::timeout(Duration::from_secs(5), async { + while task::spawn_blocking(move || process::process_running_strict(pid)) + .await + .expect("fixture process observation") + { + time::sleep(Duration::from_millis(20)).await; + } + }) + .await + .expect("owned process must stop"); + } + // Direct child must also be reaped, not merely a zombie. + let leader = self.pid("leader.pid").await; + time::timeout(Duration::from_secs(5), async { + while process::process_exists(leader) { + time::sleep(Duration::from_millis(20)).await; + } + }) + .await + .expect("direct child must be reaped while runtime lives"); + } +} +impl Drop for Fixture { + #[expect( + clippy::disallowed_methods, + reason = "Fail-safe test cleanup must run synchronously on panic" + )] + fn drop(&mut self) { + for name in ["leader.pid", "helper.pid"] { + if let Some(pid) = std::fs::read_to_string(self.0.path().join(name)) + .ok() + .and_then(|s| s.trim().parse().ok()) + { + if name == "leader.pid" { + process::sigkill_process_group(pid); + } else { + process::sigkill(pid); + } + } + } + } +} +const HELD: &str = "echo $$ > leader.pid; sleep 60 & echo $! > helper.pid; printf partial; exit 0"; +const WAIT: &str = "echo $$ > leader.pid; sleep 60 >/dev/null 2>&1 & echo $! > helper.pid; wait"; +const IGNORE_TERM: &str = "echo $$ > leader.pid; trap 'exit 0' TERM; sh -c 'trap \"\" TERM; echo $$ > helper.pid; exec sleep 60' & wait"; + +#[tokio::test] +async fn abort_execution_and_held_pipe_drain_reaps_and_kills_group() { + for script in [WAIT, HELD] { + let fixture = Fixture::new(); + let mut command = fixture.command(script); + let task = tokio::spawn(async move { + process::capture(&mut command, None, &CancellationToken::new(), None).await + }); + fixture.ready().await; + if script == HELD { + let leader = fixture.pid("leader.pid").await; + time::timeout(Duration::from_secs(5), async { + while process::process_exists(leader) { + time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .unwrap(); + } + task.abort(); + assert!(matches!(task.await, Err(error) if error.is_cancelled())); + fixture.stopped().await; + } +} + +#[tokio::test] +async fn timeout_and_cancellation_stop_real_descendants() { + for cancel in [false, true] { + let fixture = Fixture::new(); + let mut command = fixture.command(IGNORE_TERM); + let token = CancellationToken::new(); + let run = process::capture(&mut command, Some(Duration::from_millis(500)), &token, None); + let trigger = async { + fixture.ready().await; + if cancel { + token.cancel(); + } + }; + let (result, ()) = + time::timeout(Duration::from_secs(8), async { tokio::join!(run, trigger) }) + .await + .unwrap(); + assert!(matches!( + (cancel, result), + (true, Err(ProcessError::Cancelled)) | (false, Err(ProcessError::TimedOut)) + )); + fixture.stopped().await; + } +} + +#[tokio::test] +async fn normal_exit_preserves_helpers_with_closed_pipes() { + for code in [0, 7] { + let fixture = Fixture::new(); + let mut command = fixture.command(&format!("echo $$ > leader.pid; sleep 60 /dev/null 2>&1 & echo $! > helper.pid; exit {code}")); + let output = process::capture( + &mut command, + Some(Duration::from_secs(5)), + &CancellationToken::new(), + None, + ) + .await + .unwrap(); + assert_eq!(output.output.status.code(), Some(code)); + assert!(process::process_exists(fixture.pid("helper.pid").await)); + } +} + +#[tokio::test] +async fn prefix_cap_drains_both_streams_and_preserves_raw_bytes() { + for cap in [None, Some(0), Some(257)] { + let mut command = Command::new("sh"); + command.args([ + "-c", + "head -c 200000 /dev/zero & head -c 200000 /dev/zero >&2 & wait; printf '\\377'", + ]); + let output = time::timeout( + Duration::from_secs(8), + process::capture(&mut command, None, &CancellationToken::new(), cap), + ) + .await + .unwrap() + .unwrap(); + assert!(output.output.status.success()); + assert_eq!(output.output.stdout.len(), cap.unwrap_or(200_001)); + assert_eq!(output.output.stderr.len(), cap.unwrap_or(200_000)); + assert_eq!(output.stdout_truncated, cap.is_some()); + assert_eq!(output.stderr_truncated, cap.is_some()); + if cap.is_none() { + assert_eq!(output.output.stdout.last(), Some(&255)); + } + } + let mut command = Command::new("sh"); + command.args(["-c", "printf abcd; printf x >&2"]); + let output = process::capture(&mut command, None, &CancellationToken::new(), Some(2)) + .await + .unwrap(); + assert!(output.stdout_truncated); + assert!(!output.stderr_truncated); +} + +#[tokio::test] +async fn spawn_failure_preserves_source_without_rendering_command_details() { + use std::error::Error; + let mut command = Command::new("/nonexistent/fabro-test-executable"); + let error = process::capture(&mut command, None, &CancellationToken::new(), None) + .await + .err() + .unwrap(); + assert!(error.source().unwrap().is::()); + assert_eq!(error.to_string(), "process I/O failed"); + assert_eq!(format!("{error:?}"), "process I/O failed"); +} + +#[tokio::test] +async fn capture_preserves_prepared_stdin_and_literal_arguments() { + let fixture = Fixture::new(); + let input_path = fixture.0.path().join("input"); + fs::write(&input_path, b"configured stdin\n").await.unwrap(); + let input = fs::File::open(&input_path).await.unwrap().into_std().await; + let mut command = Command::new("sh"); + command + .args([ + "-c", + "cat; printf '%s' \"$1\"", + "fixture", + "$(must-not-run)", + ]) + .stdin(Stdio::from(input)); + let result = process::capture( + &mut command, + Some(Duration::from_secs(5)), + &CancellationToken::new(), + None, + ) + .await + .unwrap(); + assert_eq!(result.output.stdout, b"configured stdin\n$(must-not-run)"); + let mut command = Command::new("cat"); + command.stdin(Stdio::piped()); + let result = process::capture( + &mut command, + Some(Duration::from_secs(5)), + &CancellationToken::new(), + None, + ) + .await + .unwrap(); + assert!(result.output.stdout.is_empty()); + assert!(result.output.status.success()); +}