rx_runner/
lib.rs

1//! Runtime-neutral child-process execution for polling event loops.
2//!
3//! `rx-runner` tags stdout/stderr lines, bounds retained output, exposes
4//! non-blocking polling and cancellation, and reaps unfinished children on
5//! drop. It uses only the Rust standard library.
6//!
7//! ```no_run
8//! use std::time::Duration;
9//! use rx_runner::CommandSpec;
10//!
11//! # fn main() -> std::io::Result<()> {
12//! let mut process = CommandSpec::new("rustc").arg("--version").spawn()?;
13//! loop {
14//!     let update = process.poll()?;
15//!     for line in update.lines {
16//!         println!("{:?}: {}", line.stream, line.text);
17//!     }
18//!     if update.exit.is_some() {
19//!         break;
20//!     }
21//!     std::thread::sleep(Duration::from_millis(20));
22//! }
23//! # Ok(())
24//! # }
25//! ```
26
27use std::collections::VecDeque;
28use std::ffi::{OsStr, OsString};
29use std::fmt;
30use std::io::{self, BufRead, BufReader, Read};
31use std::path::{Path, PathBuf};
32use std::process::{Child, Command, ExitStatus, Stdio};
33use std::sync::{Arc, Mutex, MutexGuard};
34use std::thread::{self, JoinHandle};
35use std::time::{Duration, Instant};
36
37/// Default number of output lines retained between polls.
38pub const DEFAULT_OUTPUT_CAPACITY: usize = 1_024;
39const OUTPUT_DRAIN_GRACE: Duration = Duration::from_millis(50);
40
41/// A program invocation with separate arguments and process options.
42#[derive(Debug, Clone, PartialEq, Eq, Hash)]
43pub struct CommandSpec {
44    program: OsString,
45    arguments: Vec<OsString>,
46    working_directory: Option<PathBuf>,
47    output_capacity: usize,
48}
49
50impl CommandSpec {
51    /// Create a specification for `program`.
52    pub fn new(program: impl Into<OsString>) -> Self {
53        Self {
54            program: program.into(),
55            arguments: Vec::new(),
56            working_directory: None,
57            output_capacity: DEFAULT_OUTPUT_CAPACITY,
58        }
59    }
60
61    /// Append one argument.
62    #[must_use]
63    pub fn arg(mut self, arg: impl Into<OsString>) -> Self {
64        self.arguments.push(arg.into());
65        self
66    }
67
68    /// Append multiple arguments.
69    #[must_use]
70    pub fn args<I, S>(mut self, args: I) -> Self
71    where
72        I: IntoIterator<Item = S>,
73        S: Into<OsString>,
74    {
75        self.arguments.extend(args.into_iter().map(Into::into));
76        self
77    }
78
79    /// Set the child working directory.
80    #[must_use]
81    pub fn current_dir(mut self, path: impl Into<PathBuf>) -> Self {
82        self.working_directory = Some(path.into());
83        self
84    }
85
86    /// Set the maximum number of unread lines retained between polls.
87    ///
88    /// A capacity of zero discards all lines while still counting them in
89    /// [`ProcessUpdate::dropped_lines`].
90    #[must_use]
91    pub fn output_capacity(mut self, lines: usize) -> Self {
92        self.output_capacity = lines;
93        self
94    }
95
96    /// Return the executable name or path.
97    pub fn program(&self) -> &OsStr {
98        &self.program
99    }
100
101    /// Iterate over the separate process arguments.
102    pub fn arguments(&self) -> impl Iterator<Item = &OsStr> {
103        self.arguments.iter().map(OsString::as_os_str)
104    }
105
106    /// Return the configured child working directory.
107    pub fn working_directory(&self) -> Option<&Path> {
108        self.working_directory.as_deref()
109    }
110
111    /// Spawn the child with piped stdout and stderr.
112    ///
113    /// # Errors
114    ///
115    /// Returns an I/O error when the process or output-reader threads cannot
116    /// be spawned.
117    pub fn spawn(&self) -> io::Result<RunningProcess> {
118        RunningProcess::spawn(self)
119    }
120}
121
122/// Child output source.
123#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
124pub enum OutputStream {
125    /// Standard output.
126    Stdout,
127    /// Standard error.
128    Stderr,
129}
130
131/// One decoded child-output line.
132#[derive(Debug, Clone, PartialEq, Eq, Hash)]
133pub struct OutputLine {
134    /// Stream that produced the line.
135    pub stream: OutputStream,
136    /// UTF-8 text, with invalid bytes replaced and line endings removed.
137    pub text: String,
138}
139
140/// Terminal child status.
141#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
142pub struct ProcessExit {
143    /// Platform exit code, or `None` when terminated by a signal.
144    pub code: Option<i32>,
145    /// Whether the platform exit status reports success.
146    pub success: bool,
147    /// Wall-clock duration since spawn.
148    pub elapsed: Duration,
149}
150
151/// Output and optional terminal status produced by one poll.
152#[derive(Debug, Clone, PartialEq, Eq)]
153pub struct ProcessUpdate {
154    /// Lines retained since the previous poll.
155    pub lines: Vec<OutputLine>,
156    /// Lines dropped because the configured capacity was exceeded.
157    pub dropped_lines: usize,
158    /// Terminal status, returned exactly once.
159    pub exit: Option<ProcessExit>,
160}
161
162#[derive(Debug)]
163struct OutputBuffer {
164    capacity: usize,
165    lines: VecDeque<OutputLine>,
166    dropped_lines: usize,
167}
168
169impl OutputBuffer {
170    fn new(capacity: usize) -> Self {
171        Self {
172            capacity,
173            lines: VecDeque::with_capacity(capacity),
174            dropped_lines: 0,
175        }
176    }
177
178    fn push(&mut self, line: OutputLine) {
179        if self.capacity == 0 {
180            self.dropped_lines += 1;
181            return;
182        }
183        if self.lines.len() == self.capacity {
184            self.lines.pop_front();
185            self.dropped_lines += 1;
186        }
187        self.lines.push_back(line);
188    }
189
190    fn drain(&mut self) -> (Vec<OutputLine>, usize) {
191        let lines = self.lines.drain(..).collect();
192        let dropped_lines = std::mem::take(&mut self.dropped_lines);
193        (lines, dropped_lines)
194    }
195}
196
197/// A running child process with non-blocking output/status polling.
198pub struct RunningProcess {
199    child: Child,
200    output: Arc<Mutex<OutputBuffer>>,
201    readers: Vec<JoinHandle<()>>,
202    started_at: Instant,
203    exit: Option<ProcessExit>,
204    exit_detected_at: Option<Instant>,
205    exit_reported: bool,
206}
207
208impl fmt::Debug for RunningProcess {
209    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
210        formatter
211            .debug_struct("RunningProcess")
212            .field("id", &self.child.id())
213            .field("finished", &self.exit.is_some())
214            .finish()
215    }
216}
217
218impl RunningProcess {
219    fn spawn(spec: &CommandSpec) -> io::Result<Self> {
220        let mut command = Command::new(&spec.program);
221        command
222            .args(&spec.arguments)
223            .stdout(Stdio::piped())
224            .stderr(Stdio::piped());
225        if let Some(path) = &spec.working_directory {
226            command.current_dir(path);
227        }
228
229        let mut child = command.spawn()?;
230        let stdout = child
231            .stdout
232            .take()
233            .ok_or_else(|| io::Error::other("child stdout was not piped"))?;
234        let stderr = child
235            .stderr
236            .take()
237            .ok_or_else(|| io::Error::other("child stderr was not piped"))?;
238        let output = Arc::new(Mutex::new(OutputBuffer::new(spec.output_capacity)));
239
240        let stdout_reader = match spawn_reader(stdout, OutputStream::Stdout, Arc::clone(&output)) {
241            Ok(reader) => reader,
242            Err(error) => {
243                let _ = child.kill();
244                let _ = child.wait();
245                return Err(error);
246            }
247        };
248        let stderr_reader = match spawn_reader(stderr, OutputStream::Stderr, Arc::clone(&output)) {
249            Ok(reader) => reader,
250            Err(error) => {
251                let _ = child.kill();
252                let _ = child.wait();
253                let _ = stdout_reader.join();
254                return Err(error);
255            }
256        };
257
258        Ok(Self {
259            child,
260            output,
261            readers: vec![stdout_reader, stderr_reader],
262            started_at: Instant::now(),
263            exit: None,
264            exit_detected_at: None,
265            exit_reported: false,
266        })
267    }
268
269    /// Return the platform process identifier.
270    pub fn id(&self) -> u32 {
271        self.child.id()
272    }
273
274    /// Drain retained output and check for process completion without waiting.
275    ///
276    /// # Errors
277    ///
278    /// Returns an I/O error when process status cannot be read or the shared
279    /// output buffer is poisoned.
280    pub fn poll(&mut self) -> io::Result<ProcessUpdate> {
281        if self.exit.is_none()
282            && let Some(status) = self.child.try_wait()?
283        {
284            self.finish(status);
285        }
286
287        let readers_finished = self.readers.iter().all(JoinHandle::is_finished);
288        if readers_finished {
289            self.join_readers();
290        }
291        let (lines, dropped_lines) = self.output()?.drain();
292        let drain_grace_elapsed = self
293            .exit_detected_at
294            .is_some_and(|detected_at| detected_at.elapsed() >= OUTPUT_DRAIN_GRACE);
295        let exit = if self.exit_reported || !readers_finished && !drain_grace_elapsed {
296            None
297        } else {
298            self.exit_reported = self.exit.is_some();
299            self.exit
300        };
301
302        Ok(ProcessUpdate {
303            lines,
304            dropped_lines,
305            exit,
306        })
307    }
308
309    /// Terminate and reap the child, returning its terminal status.
310    ///
311    /// # Errors
312    ///
313    /// Returns an I/O error when process status cannot be read or the child
314    /// cannot be terminated and reaped.
315    pub fn cancel(&mut self) -> io::Result<ProcessExit> {
316        if let Some(exit) = self.exit {
317            self.exit_reported = true;
318            return Ok(exit);
319        }
320
321        let status = match self.child.try_wait()? {
322            Some(status) => status,
323            None => {
324                self.child.kill()?;
325                self.child.wait()?
326            }
327        };
328        self.finish(status);
329        if self.readers.iter().all(JoinHandle::is_finished) {
330            self.join_readers();
331        }
332        self.exit_reported = true;
333        self.exit
334            .ok_or_else(|| io::Error::other("cancelled process has no exit status"))
335    }
336
337    fn finish(&mut self, status: ExitStatus) {
338        self.exit_detected_at = Some(Instant::now());
339        self.exit = Some(ProcessExit {
340            code: status.code(),
341            success: status.success(),
342            elapsed: self.started_at.elapsed(),
343        });
344    }
345
346    fn join_readers(&mut self) {
347        for reader in self.readers.drain(..) {
348            let _ = reader.join();
349        }
350    }
351
352    fn output(&self) -> io::Result<MutexGuard<'_, OutputBuffer>> {
353        self.output
354            .lock()
355            .map_err(|_| io::Error::other("process output buffer is poisoned"))
356    }
357}
358
359impl Drop for RunningProcess {
360    fn drop(&mut self) {
361        if self.exit.is_none() {
362            let _ = self.child.kill();
363            let _ = self.child.wait();
364        }
365        if self.readers.iter().all(JoinHandle::is_finished) {
366            self.join_readers();
367        }
368    }
369}
370
371fn spawn_reader<R>(
372    reader: R,
373    stream: OutputStream,
374    output: Arc<Mutex<OutputBuffer>>,
375) -> io::Result<JoinHandle<()>>
376where
377    R: Read + Send + 'static,
378{
379    thread::Builder::new()
380        .name(format!("rx-runner-{stream:?}"))
381        .spawn(move || {
382            let mut reader = BufReader::new(reader);
383            let mut bytes = Vec::new();
384            loop {
385                bytes.clear();
386                let count = match reader.read_until(b'\n', &mut bytes) {
387                    Ok(count) => count,
388                    Err(_) => break,
389                };
390                if count == 0 {
391                    break;
392                }
393                while matches!(bytes.last(), Some(b'\n' | b'\r')) {
394                    bytes.pop();
395                }
396                let line = OutputLine {
397                    stream,
398                    text: String::from_utf8_lossy(&bytes).into_owned(),
399                };
400                let Ok(mut buffer) = output.lock() else {
401                    break;
402                };
403                buffer.push(line);
404            }
405        })
406}
407
408#[cfg(test)]
409mod tests {
410    use super::*;
411
412    #[cfg(unix)]
413    fn fixture(script: &str) -> CommandSpec {
414        CommandSpec::new("sh").args(["-c", script])
415    }
416
417    #[cfg(windows)]
418    fn fixture(script: &str) -> CommandSpec {
419        CommandSpec::new("cmd").args(["/C", script])
420    }
421
422    fn collect(mut process: RunningProcess) -> io::Result<ProcessUpdate> {
423        let mut lines = Vec::new();
424        let mut dropped_lines = 0;
425        for _ in 0..200 {
426            let update = process.poll()?;
427            lines.extend(update.lines);
428            dropped_lines += update.dropped_lines;
429            if update.exit.is_some() {
430                return Ok(ProcessUpdate {
431                    lines,
432                    dropped_lines,
433                    exit: update.exit,
434                });
435            }
436            thread::sleep(Duration::from_millis(5));
437        }
438        Err(io::Error::new(
439            io::ErrorKind::TimedOut,
440            "fixture process did not exit",
441        ))
442    }
443
444    #[test]
445    fn command_spec_preserves_separate_arguments_and_directory() {
446        let spec = CommandSpec::new("tool")
447            .args(["one", "two words"])
448            .current_dir("workspace")
449            .output_capacity(12);
450
451        assert_eq!(spec.program(), OsStr::new("tool"));
452        assert_eq!(
453            spec.arguments().collect::<Vec<_>>(),
454            vec![OsStr::new("one"), OsStr::new("two words")]
455        );
456        assert_eq!(spec.working_directory(), Some(Path::new("workspace")));
457        assert_eq!(spec.output_capacity, 12);
458    }
459
460    #[test]
461    fn captures_stdout_and_stderr_lines() {
462        #[cfg(unix)]
463        let spec = fixture("printf 'out\\n'; printf 'err\\n' >&2");
464        #[cfg(windows)]
465        let spec = fixture("echo out & echo err 1>&2");
466
467        let update = collect(spec.spawn().expect("spawn fixture")).expect("collect fixture");
468
469        assert!(update.exit.is_some_and(|exit| exit.success));
470        assert!(update.lines.contains(&OutputLine {
471            stream: OutputStream::Stdout,
472            text: "out".to_string(),
473        }));
474        assert!(update.lines.contains(&OutputLine {
475            stream: OutputStream::Stderr,
476            text: "err".to_string(),
477        }));
478    }
479
480    #[test]
481    fn bounded_output_reports_dropped_lines() {
482        #[cfg(unix)]
483        let spec = fixture("printf '1\\n2\\n3\\n4\\n'").output_capacity(2);
484        #[cfg(windows)]
485        let spec = fixture("(echo 1 & echo 2 & echo 3 & echo 4)").output_capacity(2);
486
487        let update = collect(spec.spawn().expect("spawn fixture")).expect("collect fixture");
488
489        assert_eq!(update.lines.len(), 2);
490        assert_eq!(update.dropped_lines, 2);
491    }
492
493    #[cfg(unix)]
494    #[test]
495    fn cancel_terminates_and_reaps_running_process() {
496        let mut process = fixture("exec sleep 30").spawn().expect("spawn fixture");
497
498        let exit = process.cancel().expect("cancel fixture");
499
500        assert!(!exit.success);
501        assert!(
502            process
503                .poll()
504                .expect("poll cancelled process")
505                .exit
506                .is_none()
507        );
508    }
509
510    #[test]
511    fn terminal_exit_is_reported_once() {
512        #[cfg(unix)]
513        let spec = fixture("exit 7");
514        #[cfg(windows)]
515        let spec = fixture("exit /B 7");
516        let mut process = spec.spawn().expect("spawn fixture");
517
518        let update = collect_until_exit(&mut process).expect("collect terminal exit");
519
520        assert_eq!(update.exit.and_then(|exit| exit.code), Some(7));
521        assert!(process.poll().expect("poll after exit").exit.is_none());
522    }
523
524    #[cfg(unix)]
525    #[test]
526    fn exit_is_reported_when_descendant_keeps_output_pipe_open() {
527        let mut process = fixture("(sleep 1) &").spawn().expect("spawn fixture");
528
529        for _ in 0..50 {
530            if process.poll().expect("poll fixture").exit.is_some() {
531                return;
532            }
533            thread::sleep(Duration::from_millis(5));
534        }
535
536        panic!("direct child exit was hidden by a descendant output pipe");
537    }
538
539    fn collect_until_exit(process: &mut RunningProcess) -> io::Result<ProcessUpdate> {
540        for _ in 0..200 {
541            let update = process.poll()?;
542            if update.exit.is_some() {
543                return Ok(update);
544            }
545            thread::sleep(Duration::from_millis(5));
546        }
547        Err(io::Error::new(
548            io::ErrorKind::TimedOut,
549            "fixture process did not exit",
550        ))
551    }
552}
rx_runner/lib.rs

rx_runner/
lib.rs
