bio_tools/
run.rs

1//! Reusable execution of third-party command-line tools.
2//!
3//! [`CommandSpec`] describes an invocation without tying it to a particular
4//! biology tool. [`CommandRunner`] turns that description into a
5//! [`std::process::Command`], captures bounded output, enforces an optional
6//! timeout, and applies the selected ExitPolicy. Rust applications can
7//! construct a [`CommandSpec`] directly or implement [`ToolInvocation`] for
8//! their own tool-specific configuration type.
9
10use std::{
11    collections::BTreeMap,
12    error::Error,
13    ffi::{OsStr, OsString},
14    fmt,
15    fs::File,
16    io::{self, Read, Write},
17    path::PathBuf,
18    process::{Command, ExitStatus, Stdio},
19    thread::{self, JoinHandle},
20    time::{Duration, Instant},
21};
22
23pub use crate::run_log::RunLogSpec;
24use crate::run_log::{ActiveRunLog, append_log_error};
25
26/// Whether a non-zero exit status is returned to the caller or treated as an
27/// execution error.
28#[derive(Clone, Copy, Debug, Eq, PartialEq)]
29pub enum ExitPolicy {
30    /// Return every completed process to the caller, irrespective of its exit
31    /// status. This is useful for --help and --version probes.
32    AllowFailure,
33    /// Convert a non-zero exit status into RunError::ExitFailure.
34    RequireSuccess,
35}
36
37/// Whether the child inherits the parent process's environment before command
38/// specific values are applied.
39#[derive(Clone, Copy, Debug, Eq, PartialEq)]
40pub enum EnvironmentPolicy {
41    Inherit,
42    Clear,
43}
44
45/// Maximum bytes retained from the tail of each output stream.
46///
47/// Readers continue draining both pipes after a limit is reached. This avoids
48/// deadlocking a verbose child while keeping application memory bounded.
49#[derive(Clone, Copy, Debug, Eq, PartialEq)]
50pub struct CaptureLimits {
51    pub stdout: usize,
52    pub stderr: usize,
53}
54
55impl CaptureLimits {
56    pub const fn new(stdout: usize, stderr: usize) -> Self {
57        Self { stdout, stderr }
58    }
59}
60
61impl Default for CaptureLimits {
62    fn default() -> Self {
63        Self::new(100_000, 100_000)
64    }
65}
66
67/// A tool invocation independent of any particular application or tool
68/// registry.
69#[derive(Clone, Debug, Eq, PartialEq)]
70pub struct CommandSpec {
71    pub program: OsString,
72    pub arguments: Vec<OsString>,
73    pub current_dir: Option<PathBuf>,
74    pub timeout: Option<Duration>,
75    pub stdin: Option<Vec<u8>>,
76    pub environment_policy: EnvironmentPolicy,
77    pub environment: BTreeMap<OsString, OsString>,
78    pub exit_policy: ExitPolicy,
79    pub capture_limits: CaptureLimits,
80    pub run_log: Option<RunLogSpec>,
81}
82
83impl CommandSpec {
84    pub fn new(program: impl Into<OsString>) -> Self {
85        Self {
86            program: program.into(),
87            arguments: Vec::new(),
88            current_dir: None,
89            timeout: Some(Duration::from_secs(120)),
90            stdin: None,
91            environment_policy: EnvironmentPolicy::Inherit,
92            environment: BTreeMap::new(),
93            exit_policy: ExitPolicy::RequireSuccess,
94            capture_limits: CaptureLimits::default(),
95            run_log: None,
96        }
97    }
98
99    pub fn arg(mut self, argument: impl Into<OsString>) -> Self {
100        self.arguments.push(argument.into());
101        self
102    }
103
104    pub fn args<I, S>(mut self, arguments: I) -> Self
105    where
106        I: IntoIterator<Item = S>,
107        S: Into<OsString>,
108    {
109        self.arguments.extend(arguments.into_iter().map(Into::into));
110        self
111    }
112
113    pub fn current_dir(mut self, path: impl Into<PathBuf>) -> Self {
114        self.current_dir = Some(path.into());
115        self
116    }
117
118    pub fn timeout(mut self, timeout: impl Into<Option<Duration>>) -> Self {
119        self.timeout = timeout.into();
120        self
121    }
122
123    pub fn stdin(mut self, input: impl Into<Vec<u8>>) -> Self {
124        self.stdin = Some(input.into());
125        self
126    }
127
128    pub fn environment_policy(mut self, policy: EnvironmentPolicy) -> Self {
129        self.environment_policy = policy;
130        self
131    }
132
133    pub fn env(mut self, name: impl Into<OsString>, value: impl Into<OsString>) -> Self {
134        self.environment.insert(name.into(), value.into());
135        self
136    }
137
138    pub fn envs<I, K, V>(mut self, variables: I) -> Self
139    where
140        I: IntoIterator<Item = (K, V)>,
141        K: Into<OsString>,
142        V: Into<OsString>,
143    {
144        self.environment.extend(
145            variables
146                .into_iter()
147                .map(|(name, value)| (name.into(), value.into())),
148        );
149        self
150    }
151
152    pub fn exit_policy(mut self, policy: ExitPolicy) -> Self {
153        self.exit_policy = policy;
154        self
155    }
156
157    pub fn capture_limits(mut self, limits: CaptureLimits) -> Self {
158        self.capture_limits = limits;
159        self
160    }
161
162    pub fn run_log(mut self, settings: RunLogSpec) -> Self {
163        self.run_log = Some(settings);
164        self
165    }
166
167    /// A shell-free, diagnostic representation of the argument vector.
168    pub fn display_command(&self) -> String {
169        std::iter::once(self.program.as_os_str())
170            .chain(self.arguments.iter().map(OsString::as_os_str))
171            .map(display_argument)
172            .collect::<Vec<_>>()
173            .join(" ")
174    }
175}
176
177/// Implemented by tool-specific types that can describe their execution as a
178/// generic CommandSpec.
179pub trait ToolInvocation {
180    fn command_spec(&self) -> CommandSpec;
181}
182
183impl ToolInvocation for CommandSpec {
184    fn command_spec(&self) -> CommandSpec {
185        self.clone()
186    }
187}
188
189/// Captured output from a child process.
190#[derive(Debug)]
191pub struct CommandOutput {
192    pub status: ExitStatus,
193    pub stdout: Vec<u8>,
194    pub stderr: Vec<u8>,
195    pub elapsed: Duration,
196    pub run_log_dir: Option<PathBuf>,
197}
198
199impl CommandOutput {
200    pub fn return_code(&self) -> Option<i32> {
201        self.status.code()
202    }
203
204    pub fn stdout_lossy(&self) -> String {
205        String::from_utf8_lossy(&self.stdout).into_owned()
206    }
207
208    pub fn stderr_lossy(&self) -> String {
209        String::from_utf8_lossy(&self.stderr).into_owned()
210    }
211
212    /// The most useful tail of failed-process output, preferring stderr.
213    pub fn diagnostic(&self, maximum_characters: usize) -> String {
214        let stderr = String::from_utf8_lossy(&self.stderr);
215        let stdout = String::from_utf8_lossy(&self.stdout);
216        let detail = if stderr.trim().is_empty() {
217            stdout.trim()
218        } else {
219            stderr.trim()
220        };
221        tail_characters(detail, maximum_characters)
222    }
223}
224
225/// Failures that occur before or while executing a tool command.
226#[derive(Debug)]
227pub enum RunError {
228    EmptyProgram,
229    Start {
230        program: OsString,
231        source: io::Error,
232    },
233    Input {
234        source: io::Error,
235    },
236    Output {
237        stream: &'static str,
238        source: io::Error,
239    },
240    Wait {
241        source: io::Error,
242    },
243    Log {
244        action: &'static str,
245        source: io::Error,
246    },
247    Timeout {
248        timeout: Duration,
249        output: CommandOutput,
250    },
251    ExitFailure {
252        output: CommandOutput,
253    },
254}
255
256impl RunError {
257    pub fn output(&self) -> Option<&CommandOutput> {
258        match self {
259            Self::Timeout { output, .. } | Self::ExitFailure { output } => Some(output),
260            _ => None,
261        }
262    }
263
264    pub fn io_error_kind(&self) -> Option<io::ErrorKind> {
265        match self {
266            Self::Start { source, .. }
267            | Self::Input { source }
268            | Self::Output { source, .. }
269            | Self::Wait { source }
270            | Self::Log { source, .. } => Some(source.kind()),
271            _ => None,
272        }
273    }
274}
275
276impl fmt::Display for RunError {
277    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
278        match self {
279            Self::EmptyProgram => formatter.write_str("No executable was provided."),
280            Self::Start { source, .. } => {
281                write!(formatter, "The tool could not be started: {source}.")
282            }
283            Self::Input { source } => {
284                write!(
285                    formatter,
286                    "The tool's standard input could not be written: {source}."
287                )
288            }
289            Self::Output { stream, source } => {
290                write!(
291                    formatter,
292                    "The tool's {stream} could not be read: {source}."
293                )
294            }
295            Self::Wait { source } => {
296                write!(formatter, "The tool could not be waited on: {source}.")
297            }
298            Self::Log { action, source } => {
299                write!(
300                    formatter,
301                    "The tool run could not be audited while {action}: {source}."
302                )
303            }
304            Self::Timeout { timeout, .. } => write!(
305                formatter,
306                "The tool exceeded the {}-second request limit.",
307                display_seconds(*timeout)
308            ),
309            Self::ExitFailure { output } => {
310                let code = output
311                    .return_code()
312                    .map(|code| code.to_string())
313                    .unwrap_or_else(|| "no exit code".to_owned());
314                let detail = output.diagnostic(2_000);
315                write!(
316                    formatter,
317                    "The tool exited with code {code}: {}",
318                    if detail.is_empty() {
319                        "No diagnostic output was produced."
320                    } else {
321                        &detail
322                    }
323                )
324            }
325        }
326    }
327}
328
329impl Error for RunError {
330    fn source(&self) -> Option<&(dyn Error + 'static)> {
331        match self {
332            Self::Start { source, .. }
333            | Self::Input { source }
334            | Self::Output { source, .. }
335            | Self::Wait { source }
336            | Self::Log { source, .. } => Some(source),
337            _ => None,
338        }
339    }
340}
341
342/// Stateless command executor. The type gives applications a stable place to
343/// add execution policy while keeping individual tool descriptions as data.
344#[derive(Clone, Debug)]
345pub struct CommandRunner {
346    poll_interval: Duration,
347}
348
349impl Default for CommandRunner {
350    fn default() -> Self {
351        Self {
352            poll_interval: Duration::from_millis(10),
353        }
354    }
355}
356
357impl CommandRunner {
358    pub fn new() -> Self {
359        Self::default()
360    }
361
362    pub fn run_tool<T: ToolInvocation>(&self, tool: &T) -> Result<CommandOutput, RunError> {
363        self.run(&tool.command_spec())
364    }
365
366    pub fn run(&self, spec: &CommandSpec) -> Result<CommandOutput, RunError> {
367        if spec.program.is_empty() {
368            return Err(RunError::EmptyProgram);
369        }
370
371        let run_log = spec
372            .run_log
373            .as_ref()
374            .map(|settings| ActiveRunLog::start(spec, settings))
375            .transpose()
376            .map_err(|source| RunError::Log {
377                action: "preparing the log",
378                source,
379            })?;
380
381        let mut command = Command::new(&spec.program);
382        command
383            .args(&spec.arguments)
384            .stdout(Stdio::piped())
385            .stderr(Stdio::piped());
386
387        if let Some(current_dir) = &spec.current_dir {
388            command.current_dir(current_dir);
389        }
390
391        if spec.environment_policy == EnvironmentPolicy::Clear {
392            command.env_clear();
393        }
394
395        command.envs(&spec.environment);
396
397        if spec.stdin.is_some() {
398            command.stdin(Stdio::piped());
399        } else {
400            command.stdin(Stdio::null());
401        }
402
403        let stdout_log = run_log
404            .as_ref()
405            .map(ActiveRunLog::stdout_file)
406            .transpose()
407            .map_err(|source| RunError::Log {
408                action: "opening stdout.txt",
409                source,
410            })?;
411        let stderr_log = run_log
412            .as_ref()
413            .map(ActiveRunLog::stderr_file)
414            .transpose()
415            .map_err(|source| RunError::Log {
416                action: "opening stderr.txt",
417                source,
418            })?;
419
420        let started = Instant::now();
421        let mut child = match command.spawn() {
422            Ok(child) => child,
423            Err(source) => {
424                if let Some(run_log) = &run_log
425                    && let Err(log_error) = run_log.record_start_error(&source)
426                {
427                    append_log_error(run_log.directory(), "recording start failure", &log_error);
428                }
429                return Err(RunError::Start {
430                    program: spec.program.clone(),
431                    source,
432                });
433            }
434        };
435
436        let stdout = child.stdout.take().expect("stdout was configured as piped");
437        let stderr = child.stderr.take().expect("stderr was configured as piped");
438        let stdout_limit = spec.capture_limits.stdout;
439        let stderr_limit = spec.capture_limits.stderr;
440        let stdout_reader = thread::spawn(move || read_tail(stdout, stdout_limit, stdout_log));
441        let stderr_reader = thread::spawn(move || read_tail(stderr, stderr_limit, stderr_log));
442
443        let input_writer = spec.stdin.as_ref().map(|input| {
444            let mut stdin = child.stdin.take().expect("stdin was configured as piped");
445            let input = input.clone();
446            thread::spawn(move || match stdin.write_all(&input) {
447                Err(error) if error.kind() == io::ErrorKind::BrokenPipe => Ok(()),
448                result => result,
449            })
450        });
451
452        let deadline = spec
453            .timeout
454            .and_then(|timeout| started.checked_add(timeout));
455
456        let (status, timed_out) = loop {
457            match child.try_wait() {
458                Ok(Some(status)) => break (status, false),
459                Ok(None) => {}
460                Err(source) => {
461                    let _ = child.kill();
462                    let _ = child.wait();
463                    return Err(RunError::Wait { source });
464                }
465            }
466
467            if deadline.is_some_and(|deadline| Instant::now() >= deadline) {
468                child.kill().map_err(|source| RunError::Wait { source })?;
469                let status = child.wait().map_err(|source| RunError::Wait { source })?;
470                break (status, true);
471            }
472
473            let sleep_for = deadline
474                .map(|deadline| deadline.saturating_duration_since(Instant::now()))
475                .map(|remaining| remaining.min(self.poll_interval))
476                .unwrap_or(self.poll_interval);
477            thread::sleep(sleep_for);
478        };
479
480        join_input(input_writer)?;
481        let stdout = join_output(stdout_reader, "standard output")?;
482        let stderr = join_output(stderr_reader, "standard error")?;
483
484        let output = CommandOutput {
485            status,
486            stdout,
487            stderr,
488            elapsed: started.elapsed(),
489            run_log_dir: run_log
490                .as_ref()
491                .map(|run_log| run_log.directory().to_owned()),
492        };
493
494        if let Some(run_log) = &run_log {
495            run_log
496                .finish(&output, timed_out)
497                .map_err(|source| RunError::Log {
498                    action: "finalizing the log",
499                    source,
500                })?;
501        }
502
503        if timed_out {
504            return Err(RunError::Timeout {
505                timeout: spec.timeout.unwrap_or_default(),
506                output,
507            });
508        }
509        if spec.exit_policy == ExitPolicy::RequireSuccess && !output.status.success() {
510            return Err(RunError::ExitFailure { output });
511        }
512        Ok(output)
513    }
514}
515
516/// Run one command with the default runner.
517pub fn run(spec: &CommandSpec) -> Result<CommandOutput, RunError> {
518    CommandRunner::default().run(spec)
519}
520
521fn read_tail(
522    mut reader: impl Read,
523    limit: usize,
524    mut full_log: Option<File>,
525) -> io::Result<Vec<u8>> {
526    let mut retained = Vec::with_capacity(limit.min(8 * 1024));
527    let mut buffer = [0_u8; 8 * 1024];
528    loop {
529        let count = reader.read(&mut buffer)?;
530        if count == 0 {
531            if let Some(log) = &mut full_log {
532                log.flush()?;
533            }
534            return Ok(retained);
535        }
536        if let Some(log) = &mut full_log {
537            log.write_all(&buffer[..count])?;
538        }
539        if limit == 0 {
540            continue;
541        }
542        let chunk = &buffer[..count];
543        if chunk.len() >= limit {
544            retained.clear();
545            retained.extend_from_slice(&chunk[chunk.len() - limit..]);
546            continue;
547        }
548        let overflow = retained
549            .len()
550            .saturating_add(chunk.len())
551            .saturating_sub(limit);
552        if overflow > 0 {
553            retained.drain(..overflow);
554        }
555        retained.extend_from_slice(chunk);
556    }
557}
558
559fn join_input(handle: Option<JoinHandle<io::Result<()>>>) -> Result<(), RunError> {
560    let Some(handle) = handle else {
561        return Ok(());
562    };
563    handle
564        .join()
565        .unwrap_or_else(|_| Err(io::Error::other("standard-input worker panicked")))
566        .map_err(|source| RunError::Input { source })
567}
568
569fn join_output(
570    handle: JoinHandle<io::Result<Vec<u8>>>,
571    stream: &'static str,
572) -> Result<Vec<u8>, RunError> {
573    handle
574        .join()
575        .unwrap_or_else(|_| Err(io::Error::other("output worker panicked")))
576        .map_err(|source| RunError::Output { stream, source })
577}
578
579fn tail_characters(value: &str, maximum: usize) -> String {
580    value
581        .chars()
582        .rev()
583        .take(maximum)
584        .collect::<String>()
585        .chars()
586        .rev()
587        .collect()
588}
589
590fn display_seconds(duration: Duration) -> String {
591    let seconds = duration.as_secs_f64();
592    if seconds.fract() == 0.0 {
593        format!("{seconds:.0}")
594    } else {
595        format!("{seconds:.3}")
596            .trim_end_matches('0')
597            .trim_end_matches('.')
598            .to_owned()
599    }
600}
601
602fn display_argument(argument: &OsStr) -> String {
603    let argument = argument.to_string_lossy();
604    if argument.is_empty() || argument.chars().any(char::is_whitespace) {
605        format!("{:?}", argument.as_ref())
606    } else {
607        argument.into_owned()
608    }
609}
bio_tools/run.rs

bio_tools/
run.rs
