eska_tool_runtime/
capture.rs

1//! Bounded raw process capture shares the same ownership and cancellation rules as private IPC.
2
3use crate::{
4    Component, Error, Operation, QueryPolicy,
5    process::OwnedChild,
6    query::{Stream, reader},
7};
8#[cfg(unix)]
9const TERMINATE: i32 = signal_hook::consts::signal::SIGTERM;
10#[cfg(not(unix))]
11const TERMINATE: i32 = 0;
12
13use std::{
14    process::{Command, Output, Stdio},
15    sync::mpsc,
16    thread,
17    time::Instant,
18};
19
20/// Capture one owned subprocess, invoking native cleanup before the final kill/reap on failure.
21///
22/// # Errors
23/// Returns process, stream-limit, timeout or cancellation failures; the exact child is reaped.
24pub fn capture(
25    component: Component,
26    command: &mut Command,
27    scope: &Operation,
28    policy: QueryPolicy,
29    on_failure: impl FnOnce(),
30) -> Result<Output, Error> {
31    run(component, command, Some(scope), policy, on_failure)
32}
33
34/// Queries on Windows preserve their existing console controller rather than installing another.
35pub fn run(
36    component: Component,
37    command: &mut Command,
38    scope: Option<&Operation>,
39    policy: QueryPolicy,
40    on_failure: impl FnOnce(),
41) -> Result<Output, Error> {
42    if scope.is_some_and(Operation::is_cancelled) {
43        return Err(Error::Cancelled);
44    }
45    let child = command
46        .stdin(Stdio::null())
47        .stdout(Stdio::piped())
48        .stderr(Stdio::piped())
49        .spawn()
50        .map_err(|source| Error::Process { component, source })?;
51    let mut child = OwnedChild(child);
52    let result = receive(component, &mut child, scope, policy);
53    if result.is_err() && !matches!(child.0.try_wait(), Ok(Some(_))) {
54        on_failure();
55    }
56    result
57}
58
59/// Drain both pipes concurrently; a graceful owner can clean descendants before its own exit.
60fn receive(
61    component: Component,
62    child: &mut OwnedChild,
63    scope: Option<&Operation>,
64    policy: QueryPolicy,
65) -> Result<Output, Error> {
66    let stdout = child.0.stdout.take().ok_or(Error::InvalidResponse)?;
67    let stderr = child.0.stderr.take().ok_or(Error::InvalidResponse)?;
68    let (sender, receiver) = mpsc::channel();
69    reader(
70        component,
71        stdout,
72        Stream::Stdout,
73        sender.clone(),
74        policy.output_limit,
75    )?;
76    reader(
77        component,
78        stderr,
79        Stream::Stderr,
80        sender,
81        policy.output_limit,
82    )?;
83    let started = Instant::now();
84    let mut stopping = None;
85    let mut status = None;
86    let mut stdout = None;
87    let mut stderr = None;
88    loop {
89        while let Ok((stream, result)) = receiver.try_recv() {
90            match stream {
91                Stream::Stdout => stdout = Some(result?),
92                Stream::Stderr => stderr = Some(result?),
93            }
94        }
95        let cancelled = scope.is_some_and(Operation::is_cancelled);
96        if stopping.is_none() && (cancelled || started.elapsed() >= policy.timeout) {
97            if policy.cancellation_grace.is_zero() {
98                return Err(reason(cancelled));
99            }
100            signal(
101                child,
102                scope
103                    .and_then(Operation::received_signal)
104                    .unwrap_or(TERMINATE),
105            )?;
106            stopping = Some((Instant::now(), cancelled));
107        }
108        if status.is_none() {
109            status = child
110                .0
111                .try_wait()
112                .map_err(|source| Error::Process { component, source })?;
113        }
114        if let Some((started, cancelled)) = stopping {
115            if status.is_some() || started.elapsed() >= policy.cancellation_grace {
116                return Err(reason(cancelled));
117            }
118        } else if let Some(status) = status
119            && stdout.is_some()
120            && stderr.is_some()
121        {
122            return Ok(Output {
123                status,
124                stdout: stdout.take().ok_or(Error::InvalidResponse)?,
125                stderr: stderr.take().ok_or(Error::InvalidResponse)?,
126            });
127        }
128        thread::sleep(std::time::Duration::from_millis(10));
129    }
130}
131
132/// Preserve the initiating failure even when the child exits successfully during cleanup.
133const fn reason(cancelled: bool) -> Error {
134    if cancelled {
135        Error::Cancelled
136    } else {
137        Error::Timeout
138    }
139}
140
141/// Signal only the owned process; its native owner handles any container-side PID.
142#[cfg(unix)]
143fn signal(child: &OwnedChild, signal: i32) -> Result<(), Error> {
144    use nix::{
145        errno::Errno,
146        sys::signal::{Signal, kill},
147        unistd::Pid,
148    };
149    let pid =
150        i32::try_from(child.0.id()).map_err(|error| Error::Signal(std::io::Error::other(error)))?;
151    let signal = Signal::try_from(signal).map_err(|error| Error::Signal(error.into()))?;
152    match kill(Pid::from_raw(pid), signal) {
153        Ok(()) | Err(Errno::ESRCH) => Ok(()),
154        Err(error) => Err(Error::Signal(error.into())),
155    }
156}
157
158/// Windows timeout fallback retains the previous direct-child termination behavior.
159#[cfg(not(unix))]
160fn signal(child: &mut OwnedChild, _: i32) -> Result<(), Error> {
161    child.0.kill().map_err(Error::Signal)
162}
eska_tool_runtime/capture.rs

eska_tool_runtime/
capture.rs
