camel_component_exec/
process.rs

1//! process.rs — OS layer: spawn, capped IO drain, process-tree kill.
2
3use bytes::Bytes;
4use std::process::Stdio;
5use tokio::io::AsyncReadExt;
6use tokio::process::Command;
7
8/// Result of a successful (possibly timed-out) execution.
9pub struct RawResult {
10    pub stdout: Bytes,
11    pub stderr: Bytes,
12    pub stdout_truncated: bool,
13    pub stderr_truncated: bool,
14    pub exit_code: Option<i32>,
15    pub timed_out: bool,
16}
17
18/// Drain a reader into Bytes, stopping storage (but continuing to drain) once
19/// `cap` is reached. Sets truncated=true if the stream exceeded the cap.
20pub async fn drain_with_cap<R: tokio::io::AsyncRead + Unpin>(r: R, cap: usize) -> (Bytes, bool) {
21    let mut buf = Vec::with_capacity(cap.min(8192));
22    let mut reader = r;
23    let mut truncated = false;
24    let mut tmp = [0u8; 8192];
25    loop {
26        match reader.read(&mut tmp).await {
27            Ok(0) => break,
28            Ok(n) => {
29                if !truncated {
30                    let space = cap - buf.len();
31                    if n <= space {
32                        buf.extend_from_slice(&tmp[..n]);
33                    } else {
34                        // Store the prefix that fits, then flip truncated.
35                        // Keep draining so the child never blocks on a full pipe.
36                        buf.extend_from_slice(&tmp[..space]);
37                        truncated = true;
38                    }
39                }
40            }
41            Err(e) => {
42                tracing::debug!(error = %e, "drain read interrupted");
43                break;
44            }
45        }
46    }
47    (Bytes::from(buf), truncated)
48}
49
50/// Spawn the child in its own process group (Unix) with given env/cwd/args.
51/// `kill_on_drop(true)` is a belt-and-suspenders guard so an accidentally-dropped
52/// Child never leaks a process (C-1 defense in depth).
53pub fn spawn(
54    exe: &std::path::Path,
55    args: &[String],
56    env: &std::collections::HashMap<String, String>,
57    cwd: &std::path::Path,
58) -> std::io::Result<tokio::process::Child> {
59    let mut cmd = Command::new(exe);
60    cmd.args(args)
61        .current_dir(cwd)
62        .env_clear()
63        .envs(env.iter())
64        .kill_on_drop(true)
65        .stdin(Stdio::piped())
66        .stdout(Stdio::piped())
67        .stderr(Stdio::piped());
68    #[cfg(unix)]
69    {
70        // process_group(0) creates a new process group for the child.
71        // This is a standard POSIX operation. The resulting Child id is
72        // always non-zero after a successful spawn, so kill_tree can safely
73        // use -pgid later.
74        cmd.process_group(0);
75    }
76    cmd.spawn()
77}
78
79/// Kill the whole process group (Unix). Best-effort on Windows (post-v1: job object).
80/// M-5: guard against pid 0/None (would signal the caller's own group if the child
81/// already exited and `id()` returned None).
82pub fn kill_tree(child: &mut tokio::process::Child) {
83    let Some(pid) = child.id() else { return };
84    #[cfg(unix)]
85    {
86        // SAFETY: sending SIGKILL to a negative pid signals the whole process group.
87        // This is a kernel syscall with no in-process memory-safety implications.
88        // pid is guaranteed non-zero (guarded above) so we cannot accidentally kill
89        // our own process group. The return value is intentionally ignored (best-effort
90        // kill — the child may already have exited).
91        unsafe {
92            libc::kill(-(pid as i32), libc::SIGKILL);
93        }
94    }
95    #[cfg(not(unix))]
96    {
97        let _ = child.start_kill();
98    }
99}
camel_component_exec/process.rs

camel_component_exec/
process.rs
