rskit_process/
sync.rs

1//! Blocking subprocess execution.
2
3use std::io::{ErrorKind, Read, Write};
4use std::process::{ChildStdin, Command as StdCommand, Stdio};
5use std::thread;
6use std::time::{Duration, Instant};
7
8use crate::{
9    AppError, AppResult, EnvPolicy, ErrorCode, InputPolicy, OutputPolicy, ProcessConfig, ProcessIo,
10    ProcessResult, ProcessSpec,
11};
12
13/// Execute a subprocess on the current thread using captured or inherited I/O mode.
14pub fn run(spec: &ProcessSpec, config: &ProcessConfig) -> AppResult<ProcessResult> {
15    if spec.program.as_os_str().is_empty() {
16        return Err(AppError::invalid_input("program", "must not be empty"));
17    }
18
19    match &config.io {
20        ProcessIo::Captured(io) => run_blocking(
21            spec,
22            config,
23            &io.input,
24            Some(&io.output),
25            pipe_stdin_stdio(&io.input)?,
26        ),
27        ProcessIo::Inherited(io) => run_blocking(
28            spec,
29            &inherited_config(config),
30            &io.input,
31            None,
32            stdin_stdio(&io.input),
33        ),
34        ProcessIo::Observed(_) => Err(AppError::invalid_input(
35            "process.io",
36            "observed mode requires async run_with_cancel",
37        )),
38    }
39}
40
41fn run_blocking(
42    spec: &ProcessSpec,
43    config: &ProcessConfig,
44    input: &InputPolicy,
45    output: Option<&OutputPolicy>,
46    stdin: Stdio,
47) -> AppResult<ProcessResult> {
48    let start = Instant::now();
49    let mut cmd = StdCommand::new(&spec.program);
50    cmd.args(&spec.args)
51        .stdin(stdin)
52        .stdout(stdout_stdio(output))
53        .stderr(stderr_stdio(output));
54
55    if let Some(dir) = &spec.dir {
56        cmd.current_dir(dir);
57    }
58
59    if matches!(spec.env_policy, EnvPolicy::Empty) {
60        cmd.env_clear();
61    }
62    for (key, value) in &spec.env {
63        cmd.env(key, value);
64    }
65
66    if config.signal.create_process_group {
67        crate::process_group::isolate(&mut cmd);
68    }
69
70    let mut child = cmd.spawn().map_err(|error| {
71        AppError::new(
72            ErrorCode::Internal,
73            format!("failed to spawn process: {error}"),
74        )
75    })?;
76    let pid = Some(child.id());
77
78    let max_output_bytes = output.and_then(|output| output.max_output_bytes);
79    let stdout = child
80        .stdout
81        .take()
82        .map(|stream| spawn_reader(stream, max_output_bytes));
83    let stderr = child
84        .stderr
85        .take()
86        .map(|stream| spawn_reader(stream, max_output_bytes));
87    let stdin = spawn_stdin_writer(child.stdin.take(), input);
88
89    let (exit_code, timed_out, synthetic_stderr) =
90        wait_with_timeout(&mut child, pid, config.timeout, config)?;
91    join_stdin(stdin)?;
92    let stdout_output = join_reader(stdout)?;
93    let mut stderr_output = join_reader(stderr)?;
94    if let Some(extra_stderr) = synthetic_stderr {
95        stderr_output.truncated |= append_bounded(
96            &mut stderr_output.bytes,
97            extra_stderr.as_bytes(),
98            max_output_bytes,
99        );
100    }
101
102    fn spawn_stdin_writer(
103        stdin: Option<ChildStdin>,
104        input: &InputPolicy,
105    ) -> Option<thread::JoinHandle<AppResult<()>>> {
106        let InputPolicy::Bytes(bytes) = input else {
107            return None;
108        };
109        let mut stdin = stdin?;
110        let bytes = bytes.clone();
111        Some(thread::spawn(move || match stdin.write_all(&bytes) {
112            Ok(()) => Ok(()),
113            Err(error) if error.kind() == ErrorKind::BrokenPipe => Ok(()),
114            Err(error) => Err(AppError::new(
115                ErrorCode::Internal,
116                format!("failed to write to stdin: {error}"),
117            )),
118        }))
119    }
120
121    Ok(ProcessResult::completed(
122        exit_code,
123        stdout_output.bytes,
124        stderr_output.bytes,
125        stdout_output.truncated,
126        stderr_output.truncated,
127        start.elapsed(),
128        timed_out,
129        false,
130    ))
131}
132
133fn stdin_stdio(input: &InputPolicy) -> Stdio {
134    match input {
135        InputPolicy::Closed => Stdio::null(),
136        InputPolicy::Bytes(_) => Stdio::piped(),
137        InputPolicy::Inherit => Stdio::inherit(),
138    }
139}
140
141fn pipe_stdin_stdio(input: &InputPolicy) -> AppResult<Stdio> {
142    match input {
143        InputPolicy::Closed => Ok(Stdio::null()),
144        InputPolicy::Bytes(_) => Ok(Stdio::piped()),
145        InputPolicy::Inherit => Err(AppError::invalid_input(
146            "process.io.input",
147            "inherited stdin requires inherited I/O mode; pipe-backed interactive stdin is not supported",
148        )),
149    }
150}
151
152fn inherited_config(config: &ProcessConfig) -> ProcessConfig {
153    let mut config = config.clone();
154    config.signal = config
155        .signal
156        .with_create_process_group(false)
157        .with_terminate_descendants(false);
158    config
159}
160
161fn stdout_stdio(output: Option<&OutputPolicy>) -> Stdio {
162    match output {
163        Some(output) if output.capture_stdout => Stdio::piped(),
164        Some(_) => Stdio::null(),
165        None => Stdio::inherit(),
166    }
167}
168
169fn stderr_stdio(output: Option<&OutputPolicy>) -> Stdio {
170    match output {
171        Some(output) if output.capture_stderr => Stdio::piped(),
172        Some(_) => Stdio::null(),
173        None => Stdio::inherit(),
174    }
175}
176
177fn wait_with_timeout(
178    child: &mut std::process::Child,
179    pid: Option<u32>,
180    timeout: Option<Duration>,
181    config: &ProcessConfig,
182) -> AppResult<(Option<i32>, bool, Option<String>)> {
183    let Some(timeout) = timeout else {
184        let status = child.wait().map_err(|error| {
185            AppError::new(
186                ErrorCode::Internal,
187                format!("process execution error: {error}"),
188            )
189        })?;
190        return Ok((status.code(), false, None));
191    };
192
193    let deadline = Instant::now() + timeout;
194    loop {
195        if let Some(status) = child.try_wait().map_err(|error| {
196            AppError::new(
197                ErrorCode::Internal,
198                format!("process execution error: {error}"),
199            )
200        })? {
201            return Ok((status.code(), false, None));
202        }
203        if Instant::now() >= deadline {
204            return terminate_and_wait(child, pid, config);
205        }
206        thread::sleep(Duration::from_millis(10));
207    }
208}
209
210fn terminate_and_wait(
211    child: &mut std::process::Child,
212    pid: Option<u32>,
213    config: &ProcessConfig,
214) -> AppResult<(Option<i32>, bool, Option<String>)> {
215    if !terminate_process(pid, config, libc::SIGTERM) {
216        child.kill().map_err(|error| {
217            AppError::new(
218                ErrorCode::Internal,
219                format!("failed to terminate process after timeout: {error}"),
220            )
221        })?;
222    }
223
224    let deadline = Instant::now() + config.signal.grace_period;
225    loop {
226        if let Some(status) = child.try_wait().map_err(|error| {
227            AppError::new(
228                ErrorCode::Internal,
229                format!("process execution error after timeout: {error}"),
230            )
231        })? {
232            return Ok((status.code(), true, None));
233        }
234        if Instant::now() >= deadline {
235            if !terminate_process(pid, config, libc::SIGKILL) {
236                child.kill().map_err(|error| {
237                    AppError::new(
238                        ErrorCode::Internal,
239                        format!("failed to kill process after timeout: {error}"),
240                    )
241                })?;
242            }
243            let status = child.wait().map_err(|error| {
244                AppError::new(
245                    ErrorCode::Internal,
246                    format!("process execution error after kill: {error}"),
247                )
248            })?;
249            return Ok((
250                status.code(),
251                true,
252                Some("process killed by SIGKILL after timeout".to_string()),
253            ));
254        }
255        thread::sleep(Duration::from_millis(10));
256    }
257}
258
259struct ReaderOutput {
260    bytes: Vec<u8>,
261    truncated: bool,
262}
263
264fn spawn_reader<R>(
265    mut reader: R,
266    max_bytes: Option<usize>,
267) -> thread::JoinHandle<std::io::Result<ReaderOutput>>
268where
269    R: Read + Send + 'static,
270{
271    thread::spawn(move || {
272        let mut bytes = Vec::new();
273        let mut buffer = [0_u8; 4096];
274        let mut remaining = max_bytes.unwrap_or(usize::MAX);
275        let mut truncated = false;
276        loop {
277            let read = reader.read(&mut buffer)?;
278            if read == 0 {
279                break;
280            }
281            if remaining > 0 {
282                let to_copy = remaining.min(read);
283                bytes.extend_from_slice(&buffer[..to_copy]);
284                remaining -= to_copy;
285                if to_copy < read {
286                    truncated = true;
287                }
288            } else {
289                truncated = true;
290            }
291        }
292        Ok(ReaderOutput { bytes, truncated })
293    })
294}
295
296fn join_reader(
297    handle: Option<thread::JoinHandle<std::io::Result<ReaderOutput>>>,
298) -> AppResult<ReaderOutput> {
299    match handle {
300        Some(handle) => handle
301            .join()
302            .map_err(|_| {
303                AppError::new(ErrorCode::Internal, "process output reader thread panicked")
304            })?
305            .map_err(AppError::internal),
306        None => Ok(ReaderOutput {
307            bytes: Vec::new(),
308            truncated: false,
309        }),
310    }
311}
312
313fn join_stdin(handle: Option<thread::JoinHandle<AppResult<()>>>) -> AppResult<()> {
314    match handle {
315        Some(handle) => handle
316            .join()
317            .map_err(|_| AppError::new(ErrorCode::Internal, "process stdin writer panicked"))?,
318        None => Ok(()),
319    }
320}
321
322fn append_bounded(target: &mut Vec<u8>, extra: &[u8], max_bytes: Option<usize>) -> bool {
323    let Some(limit) = max_bytes else {
324        if !target.is_empty() {
325            target.push(b'\n');
326        }
327        target.extend_from_slice(extra);
328        return false;
329    };
330    if target.len() >= limit {
331        return true;
332    }
333    if !target.is_empty() && target.len() + 1 < limit {
334        target.push(b'\n');
335    }
336    let remaining = limit.saturating_sub(target.len());
337    if extra.len() > remaining {
338        target.extend_from_slice(&extra[..remaining]);
339        true
340    } else {
341        target.extend_from_slice(extra);
342        false
343    }
344}
345
346fn terminate_process(pid: Option<u32>, config: &ProcessConfig, signal: i32) -> bool {
347    if let Some(pid) = pid {
348        #[cfg(unix)]
349        // SAFETY: `kill` targets either the child pid or the negated
350        // process-group id created by the `pre_exec` hook. ESRCH means it
351        // already exited.
352        unsafe {
353            let target =
354                if config.signal.create_process_group && config.signal.terminate_descendants {
355                    -(pid as i32)
356                } else {
357                    pid as i32
358                };
359            let result = libc::kill(target, signal);
360            if result != 0 {
361                let error = std::io::Error::last_os_error();
362                if error.raw_os_error() != Some(libc::ESRCH) {
363                    return false;
364                }
365            }
366            return true;
367        }
368    }
369    false
370}
371
372#[cfg(test)]
373mod tests {
374    use super::*;
375    use crate::{CapturedIo, ProcessIo};
376
377    #[test]
378    fn stdio_helpers_map_input_and_output_policies() {
379        assert!(pipe_stdin_stdio(&InputPolicy::Closed).is_ok());
380        assert!(pipe_stdin_stdio(&InputPolicy::Bytes(b"x".to_vec())).is_ok());
381        assert_eq!(
382            pipe_stdin_stdio(&InputPolicy::Inherit).unwrap_err().code(),
383            ErrorCode::InvalidInput
384        );
385
386        let captured = OutputPolicy::captured();
387        let _ = stdout_stdio(Some(&captured));
388        let _ = stderr_stdio(Some(&captured));
389        let discarded = OutputPolicy::observe_only();
390        let _ = stdout_stdio(Some(&discarded));
391        let _ = stderr_stdio(Some(&discarded));
392        let _ = stdout_stdio(None);
393        let _ = stderr_stdio(None);
394        let _ = stdin_stdio(&InputPolicy::Closed);
395        let _ = stdin_stdio(&InputPolicy::Bytes(Vec::new()));
396        let _ = stdin_stdio(&InputPolicy::Inherit);
397    }
398
399    #[test]
400    fn inherited_config_disables_descendant_signalling_and_join_none_is_ok() {
401        let config = ProcessConfig::default()
402            .with_io(ProcessIo::captured(CapturedIo::new()))
403            .with_timeout(None);
404        let inherited = inherited_config(&config);
405
406        assert!(!inherited.signal.create_process_group);
407        assert!(!inherited.signal.terminate_descendants);
408        assert_eq!(inherited.timeout, None);
409        join_stdin(None).unwrap();
410        let output = join_reader(None).unwrap();
411        assert!(output.bytes.is_empty());
412        assert!(!output.truncated);
413        assert!(!terminate_process(None, &config, libc::SIGTERM));
414    }
415
416    #[test]
417    fn append_bounded_preserves_separator_and_reports_truncation() {
418        let mut bytes = b"abc".to_vec();
419        assert!(!append_bounded(&mut bytes, b"def", None));
420        assert_eq!(bytes, b"abc\ndef");
421
422        let mut limited = b"abc".to_vec();
423        assert!(!append_bounded(&mut limited, b"de", Some(6)));
424        assert_eq!(limited, b"abc\nde");
425        assert!(append_bounded(&mut limited, b"f", Some(6)));
426
427        let mut already_full = b"abcdef".to_vec();
428        assert!(append_bounded(&mut already_full, b"g", Some(6)));
429        assert_eq!(already_full, b"abcdef");
430    }
431
432    #[test]
433    fn join_helpers_report_panicked_threads() {
434        let reader = std::thread::spawn(|| -> std::io::Result<ReaderOutput> {
435            panic!("reader panic");
436        });
437        let stdin = std::thread::spawn(|| -> AppResult<()> {
438            panic!("stdin panic");
439        });
440
441        let reader_error = match join_reader(Some(reader)) {
442            Ok(_) => panic!("reader panic should map to error"),
443            Err(error) => error,
444        };
445        let stdin_error = match join_stdin(Some(stdin)) {
446            Ok(_) => panic!("stdin panic should map to error"),
447            Err(error) => error,
448        };
449
450        assert_eq!(reader_error.code(), ErrorCode::Internal);
451        assert_eq!(stdin_error.code(), ErrorCode::Internal);
452    }
453
454    #[cfg(unix)]
455    #[test]
456    fn terminate_process_treats_esrch_as_already_exited() {
457        // Spawn and reap a short-lived child so the PID is guaranteed dead,
458        // then probe with signal 0 (existence check) instead of SIGTERM so
459        // PID-reuse cannot cause this test to deliver a real signal to an
460        // unrelated process.
461        let mut child = std::process::Command::new("/usr/bin/true")
462            .stdout(std::process::Stdio::null())
463            .stderr(std::process::Stdio::null())
464            .spawn()
465            .expect("spawning /usr/bin/true succeeds");
466        let pid = child.id();
467        child.wait().expect("child exits and is reaped");
468
469        let config = ProcessConfig::default();
470        assert!(terminate_process(Some(pid), &config, 0));
471    }
472}
rskit_process/sync.rs

rskit_process/
sync.rs
