ntfs_mac_core/
runner.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright 2026 kodephp contributors
3
4//! Small subprocess runner.
5//!
6//! Wraps `std::process::Command` with:
7//!
8//! * A `timeout` guard that SIGTERMs then SIGKILLs on overrun.
9//! * Preserved stderr so [`crate::Error::CommandFailed`] can surface the
10//!   underlying tool's own diagnostics.
11//! * `CommandRunner` trait so tests can inject a fake binary (see
12//!   `#[cfg(test)]` in each module).
13
14use std::io::{BufRead, Read, Write};
15use std::os::unix::process::ExitStatusExt;
16use std::process::{Child, Command, Stdio};
17use std::sync::{Arc, Mutex};
18use std::time::{Duration, Instant};
19
20use crate::error::{Error, Result};
21
22/// Status reported for a subprocess that was killed by its timeout.
23/// Exposed so callers and tests can branch on it without magic numbers.
24pub const TIMEOUT_STATUS: i32 = 150;
25
26/// Grace given after SIGTERM before escalating to SIGKILL.
27const SIGTERM_GRACE: Duration = Duration::from_secs(1);
28
29/// Poll interval used while racing against a deadline.
30const POLL_INTERVAL: Duration = Duration::from_millis(20);
31
32/// Upper bound on waiting for the stdout/stderr readers to flush.
33const READ_DRAIN_TIMEOUT: Duration = Duration::from_millis(500);
34
35/// Output of a completed subprocess.
36#[derive(Debug, Default)]
37pub struct RunResult {
38    pub status: i32,
39    pub stdout: String,
40    pub stderr: String,
41    pub duration: Duration,
42    /// `true` when the configured timeout fired and the child was
43    /// terminated. `status` is then [`TIMEOUT_STATUS`].
44    pub timed_out: bool,
45}
46
47impl RunResult {
48    pub fn success(&self) -> bool {
49        self.status == 0
50    }
51}
52
53/// Configuration for a single subprocess.
54#[derive(Debug, Clone)]
55pub struct RunOptions {
56    /// Hard timeout. `None` disables the timeout.
57    pub timeout: Option<Duration>,
58    /// Additional environment variables (additive).
59    pub env_extra: Vec<(String, String)>,
60    /// Working directory override.
61    pub cwd: Option<std::path::PathBuf>,
62    /// Capture stdout/stderr? Defaults to `true`. When `false`, the
63    /// child inherits the parent's stdio (progress displays stay visible).
64    pub capture: bool,
65    /// Stream stdin (only meaningful when `capture` is `false`).
66    pub stdin_data: Option<String>,
67}
68
69impl Default for RunOptions {
70    fn default() -> Self {
71        Self {
72            timeout: None,
73            env_extra: Vec::new(),
74            cwd: None,
75            // Manual Default (not derive): a bool derives to `false`,
76            // but capturing is the long-standing default behaviour —
77            // error reporting depends on the captured stderr.
78            capture: true,
79            stdin_data: None,
80        }
81    }
82}
83
84/// Runs a command with [`RunOptions`] and returns [`RunResult`].
85///
86/// On timeout the child is SIGTERMed and, if it does not exit within
87/// [`SIGTERM_GRACE`], SIGKILLed. `run` itself never fails on timeout:
88/// it returns [`RunResult`] with [`RunResult::timed_out`] set and
89/// [`RunResult::status`] = [`TIMEOUT_STATUS`]. [`run_expect_success`]
90/// turns that into [`Error::CommandTimedOut`].
91pub fn run(cmd: &str, args: &[&str], opts: &RunOptions) -> Result<RunResult> {
92    let started = Instant::now();
93
94    let mut command = Command::new(cmd);
95    command.args(args).env_remove("TERM"); // Prevent tools from emitting escape codes we cannot parse.
96    if opts.capture {
97        command
98            .stdin(Stdio::piped())
99            .stdout(Stdio::piped())
100            .stderr(Stdio::piped());
101    } else {
102        // Inherit mode: the child's output goes straight to the parent's
103        // terminals (used for long-running tools with progress output
104        // such as `rsync --progress`).
105        command
106            .stdin(Stdio::inherit())
107            .stdout(Stdio::inherit())
108            .stderr(Stdio::inherit());
109    }
110    for (k, v) in &opts.env_extra {
111        command.env(k, v);
112    }
113    if let Some(cwd) = &opts.cwd {
114        command.current_dir(cwd);
115    }
116
117    let mut child = command.spawn().map_err(|io| Error::CommandFailed {
118        cmd: cmd.to_string(),
119        status: -1,
120        stderr: String::new(),
121        io: Some(io),
122    })?;
123
124    // Feed stdin if provided.
125    if let Some(data) = &opts.stdin_data {
126        use std::io::Write;
127        let _ = child.stdin.as_mut().map(|w| w.write_all(data.as_bytes()));
128        if let Some(mut stdin) = child.stdin.take() {
129            let _ = stdin.flush();
130        }
131    } else {
132        child.stdin.take();
133    }
134
135    // Collect stdout/stderr in background threads so we can race against the timeout.
136    let stdout_buf = Arc::new(Mutex::new(Vec::new()));
137    let stderr_buf = Arc::new(Mutex::new(Vec::new()));
138
139    let stdout_arc = stdout_buf.clone();
140    // Take stdout and stderr handles before moving them into threads,
141    // so `child` remains fully owned for `try_wait`/`wait`.
142    let stdout_handle = child.stdout.take();
143    let stderr_handle = child.stderr.take();
144
145    let stdout_thread = std::thread::spawn(move || {
146        if let Some(mut handle) = stdout_handle {
147            let mut out = Vec::new();
148            let _ = handle.read_to_end(&mut out);
149            // A poisoned mutex still holds valid data (the panic happened
150            // elsewhere); recovering it is strictly better than panicking.
151            let _ = stdout_arc
152                .lock()
153                .unwrap_or_else(std::sync::PoisonError::into_inner)
154                .write_all(&out);
155        }
156    });
157
158    let stderr_arc = stderr_buf.clone();
159    let stderr_thread = std::thread::spawn(move || {
160        if let Some(mut handle) = stderr_handle {
161            let mut err = Vec::new();
162            let _ = handle.read_to_end(&mut err);
163            let _ = stderr_arc
164                .lock()
165                .unwrap_or_else(std::sync::PoisonError::into_inner)
166                .write_all(&err);
167        }
168    });
169
170    // Wait with timeout.
171    let mut waited_status = None;
172    let mut timed_out = false;
173
174    if let Some(timeout) = opts.timeout {
175        let deadline = Instant::now() + timeout;
176        loop {
177            if let Ok(Some(status)) = child.try_wait() {
178                waited_status = Some(status);
179                break;
180            }
181            if Instant::now() >= deadline {
182                timed_out = true;
183                break;
184            }
185            std::thread::sleep(POLL_INTERVAL);
186        }
187    }
188
189    if timed_out {
190        terminate_gracefully(&mut child);
191        waited_status = Some(
192            child
193                .wait()
194                .unwrap_or_else(|_| std::process::ExitStatus::from_raw(137)),
195        );
196    } else if waited_status.is_none() {
197        // Wait unbounded if no timeout.
198        waited_status = Some(
199            child
200                .wait()
201                .unwrap_or_else(|_| std::process::ExitStatus::from_raw(137)),
202        );
203    }
204
205    // Drain whatever the child produced, but never let the drain itself
206    // blow past the timeout. `join()` below would otherwise block for as
207    // long as *any* descendant holds a pipe write-end: a timed-out
208    // `brew install` leaves `curl`/`tar` behind, and they keep the pipe
209    // open long after we killed `brew`. The child is already dead or
210    // dying at this point, so returning with a partial capture is the
211    // correct behaviour — the reader threads keep running detached and
212    // finish on their own when the orphan eventually exits.
213    let drain_deadline = Instant::now() + READ_DRAIN_TIMEOUT;
214    while !stdout_thread.is_finished() || !stderr_thread.is_finished() {
215        if Instant::now() >= drain_deadline {
216            break;
217        }
218        std::thread::sleep(POLL_INTERVAL);
219    }
220    if stdout_thread.is_finished() {
221        let _ = stdout_thread.join();
222    }
223    if stderr_thread.is_finished() {
224        let _ = stderr_thread.join();
225    }
226
227    let status = match waited_status {
228        Some(_s) if timed_out => TIMEOUT_STATUS,
229        Some(s) if s.success() => 0,
230        Some(s) => s.code().unwrap_or(-1),
231        None => -1,
232    };
233
234    let stdout = String::from_utf8_lossy(
235        &stdout_buf
236            .lock()
237            .unwrap_or_else(std::sync::PoisonError::into_inner),
238    )
239    .to_string();
240    let stderr = String::from_utf8_lossy(
241        &stderr_buf
242            .lock()
243            .unwrap_or_else(std::sync::PoisonError::into_inner),
244    )
245    .to_string();
246
247    Ok(RunResult {
248        status,
249        stdout,
250        stderr,
251        duration: started.elapsed(),
252        timed_out,
253    })
254}
255
256/// Terminate a timed-out child: SIGTERM first, escalating to SIGKILL
257/// only if it is still alive after [`SIGTERM_GRACE`].
258///
259/// The tools this runner supervises hold filesystem state while they run
260/// (`fsck_ntfs` checks, `newfs_ntfs` formats, `rsync`/`cp` transfer up
261/// to an hour of data), so an unannounced SIGKILL can leave a volume's
262/// journal or a transfer in a worse state than a clean shutdown. This
263/// makes the module-level contract — "SIGTERMs then SIGKILLs" — true
264/// instead of decorative.
265fn terminate_gracefully(child: &mut Child) {
266    let pid = child.id() as libc::pid_t;
267    // SAFETY: `pid` is the pid of a child we spawned moments ago and is
268    // still alive (it just failed `try_wait` above), so it cannot have
269    // been reaped or recycled by another process into an unrelated task.
270    // `SIGTERM` is a valid signal; sending it cannot invalidate any of
271    // our own allocations because `child` still owns the process handle.
272    // `libc::kill` is `unsafe` solely because signalling an arbitrary
273    // pid of unknown ownership is UB-adjacent — both preconditions here
274    // are established by construction.
275    let _ = unsafe { libc::kill(pid, libc::SIGTERM) };
276
277    let grace_deadline = Instant::now() + SIGTERM_GRACE;
278    while Instant::now() < grace_deadline {
279        match child.try_wait() {
280            // The child honoured SIGTERM; it will not be SIGKILLed.
281            Ok(Some(_)) => return,
282            Ok(None) => std::thread::sleep(POLL_INTERVAL),
283            // A wait error means there is no child left to signal.
284            Err(_) => return,
285        }
286    }
287
288    let _ = child.kill();
289}
290
291/// Convenience helper: run a command that should succeed, otherwise
292/// wrap into [`Error::CommandFailed`].
293pub fn run_expect_success(cmd: &str, args: &[&str], opts: &RunOptions) -> Result<RunResult> {
294    let out = run(cmd, args, opts)?;
295    if out.timed_out {
296        return Err(Error::CommandTimedOut {
297            cmd: cmd.to_string(),
298            timeout: opts.timeout.unwrap_or_default(),
299        });
300    }
301    if out.success() {
302        Ok(out)
303    } else {
304        Err(Error::CommandFailed {
305            cmd: cmd.to_string(),
306            status: out.status,
307            stderr: if out.stderr.trim().is_empty() {
308                String::from("<empty stderr>")
309            } else {
310                out.stderr.trim().to_string()
311            },
312            io: None,
313        })
314    }
315}
316
317/// Locate a binary on `PATH`. Returns the full path or
318/// [`Error::MissingDependency`].
319pub fn which(binary: &str) -> Result<std::path::PathBuf> {
320    let path_env = std::env::var_os("PATH").ok_or_else(|| Error::MissingDependency {
321        binary: binary.to_string(),
322        detail: "PATH environment variable is not set".into(),
323        hint: Some("Set PATH to include the directory containing the tool.".into()),
324        io: None,
325    })?;
326
327    for dir in std::env::split_paths(&path_env) {
328        let candidate = dir.join(binary);
329        if candidate.is_file() {
330            // Check executability.
331            use std::os::unix::fs::PermissionsExt;
332            let meta = std::fs::metadata(&candidate)?;
333            if meta.permissions().mode() & 0o111 != 0 {
334                return Ok(candidate);
335            }
336        }
337    }
338
339    Err(Error::MissingDependency {
340        binary: binary.to_string(),
341        detail: format!("`{binary}` not found on PATH"),
342        hint: Some(format!(
343            "Install via `brew install {binary}` (or run `./scripts/install.sh`)."
344        )),
345        io: None,
346    })
347}
348
349/// Read stdin from an interactive TTY, returning a trimmed line.
350///
351/// Used for the CLI confirmation flow. `prompt` is echoed to stderr
352/// so stdout stays clean for JSON mode.
353pub fn read_line_interactive(prompt: &str) -> Result<String> {
354    eprintln!("{prompt}");
355    let mut line = String::new();
356    if std::io::stdin().lock().read_line(&mut line).is_err() {
357        return Err(Error::Cancelled);
358    }
359    Ok(line.trim().to_string())
360}
ntfs_mac_core/runner.rs

ntfs_mac_core/
runner.rs
