fallow-process 3.11.0

Shared child-process lifecycle and process-tree cleanup for fallow
Documentation
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
//! RAII wrapper around `std::process::Child` that registers the child's
//! PID with the process-wide signal registry on spawn and deregisters
//! on drop or explicit consume (`wait_with_output`, `wait`).
//!
//! Storage model: the wrapper owns the `Child` outright. Regular subprocesses
//! register their PID. Top-level process-tree subprocesses share an owned
//! process group or Windows Job Object handle with the signal registry and
//! timeout watchers. Nested subprocesses inherit that outer tree and retain a
//! direct-process terminator for their local timeout.
//!
//! Why PID-based and not Child-based: the wrapper needs to call
//! `Child::wait_with_output(self)` which consumes the Child by value.
//! If the registry also held the Child, there would be no clean way to
//! transfer ownership for the wait while still letting the signal
//! handler kill it. Storing the PID sidesteps the problem entirely:
//! kill-by-PID is a side channel that does not interfere with wait.
//!
//! Known race (small window, low consequence): a child that completes
//! naturally is reaped inside `wait_with_output` BEFORE we deregister
//! its PID from the registry. If a signal arrives in the microseconds-
//! wide window between `wait_with_output` returning and `deregister`
//! running, the drain snapshots a now-recycled PID and sends `kill -9`
//! to whatever process the kernel assigned that PID to. The window is
//! small (one async-write to a Mutex), the consequence is one stray
//! SIGKILL during shutdown, and recovery requires a more invasive
//! design (an `Arc<Mutex<Option<Child>>>` shared with the registry).
//! Documented here so future maintainers don't re-derive the trade-off.

use std::io;
use std::process::{
    Child, ChildStderr, ChildStdin, ChildStdout, Command, ExitStatus, Output, Stdio,
};
use std::sync::Arc;

use super::registry;

/// RAII handle wrapping a spawned `Child` with registry tracking.
pub struct ScopedChild {
    /// `None` after the wrapper has consumed the child (`wait_with_output`,
    /// `wait`). Drop checks this and reaps non-blockingly if the child
    /// is still here.
    inner: Option<Child>,
    /// Registry key. `None` after deregister so Drop does not redo it.
    id: Option<u64>,
    /// Shared ownership of the reserved process-tree identity, when requested.
    process_tree: Option<registry::ProcessTreeHandle>,
    /// PID of a child that inherited an already-managed outer process tree.
    inherited_tree_pid: Option<u32>,
}

/// Cloneable process-tree termination capability for timeout and I/O guards.
#[derive(Clone)]
pub struct ProcessTreeTerminator {
    target: TerminationTarget,
    direct_pid: u32,
}

#[derive(Clone)]
enum TerminationTarget {
    ProcessTree(registry::ProcessTreeHandle),
    Process(u32),
}

impl ProcessTreeTerminator {
    pub fn terminate(&self) -> io::Result<()> {
        let result = match &self.target {
            TerminationTarget::ProcessTree(process_tree) => process_tree.terminate(),
            TerminationTarget::Process(pid) => {
                registry::kill_pid(*pid);
                Ok(())
            }
        };
        if result.is_err() {
            registry::kill_pid(self.direct_pid);
        }
        result
    }

    #[cfg(all(test, windows))]
    fn is_alive(&self) -> bool {
        match &self.target {
            TerminationTarget::ProcessTree(process_tree) => process_tree.is_alive(),
            TerminationTarget::Process(pid) => registry::pid_is_alive(*pid),
        }
    }
}

impl ScopedChild {
    /// Spawn the command and register the resulting child's PID.
    pub fn spawn(command: &mut Command) -> io::Result<Self> {
        let child = command.spawn()?;
        let id = registry::register(child.id());
        Ok(Self {
            inner: Some(child),
            id: Some(id),
            process_tree: None,
            inherited_tree_pid: None,
        })
    }

    /// Spawn a subprocess wrapper in a dedicated process tree and register the
    /// tree for process-wide signal cleanup.
    pub fn spawn_process_tree(command: &mut Command) -> io::Result<Self> {
        if crate::process_tree::inherits_managed_process_tree() {
            let child = command.spawn()?;
            let pid = child.id();
            let id = registry::register(pid);
            return Ok(Self {
                inner: Some(child),
                id: Some(id),
                process_tree: None,
                inherited_tree_pid: Some(pid),
            });
        }

        crate::process_tree::configure_std_command(command);
        let mut child = command.spawn()?;
        let process_tree = match crate::process_tree::ProcessTree::for_std_child(&child) {
            Ok(process_tree) => Arc::new(process_tree),
            Err(error) => {
                terminate_failed_setup(&mut child);
                return Err(error);
            }
        };
        let id = registry::register_process_tree(Arc::clone(&process_tree));
        Ok(Self {
            inner: Some(child),
            id: Some(id),
            process_tree: Some(process_tree),
            inherited_tree_pid: None,
        })
    }

    /// Return a cloneable handle that terminates the owned tree, or the direct
    /// child when it inherited a tree from a Fallow parent.
    pub fn process_tree_terminator(&self) -> Option<ProcessTreeTerminator> {
        let direct_pid = self.inner.as_ref().map(Child::id)?;
        if let Some(process_tree) = self.process_tree.as_ref() {
            return Some(ProcessTreeTerminator {
                target: TerminationTarget::ProcessTree(Arc::clone(process_tree)),
                direct_pid,
            });
        }
        self.inherited_tree_pid.map(|pid| ProcessTreeTerminator {
            target: TerminationTarget::Process(pid),
            direct_pid,
        })
    }

    /// OS-level process id of the underlying child. Returns `0` if the
    /// child has been consumed.
    pub fn id(&self) -> u32 {
        self.inner.as_ref().map_or(0, Child::id)
    }

    /// Take the child's stdin handle, if it was piped. Same semantics
    /// as `Child::stdin.take()`. Returns `None` if stdin was not piped
    /// or the child has been consumed.
    pub fn take_stdin(&mut self) -> Option<ChildStdin> {
        self.inner.as_mut().and_then(|c| c.stdin.take())
    }

    /// Take the child's stdout handle, if it was piped. Same semantics as
    /// `Child::stdout.take()`. Returns `None` if stdout was not piped or the
    /// child has been consumed. Used by long-lived readers (e.g. the audit
    /// base-file `cat-file --batch` reader) that drive both pipes while leaving
    /// the wrapper owning the child for registry tracking and the terminal wait.
    pub fn take_stdout(&mut self) -> Option<ChildStdout> {
        self.inner.as_mut().and_then(|c| c.stdout.take())
    }

    /// Take the child's stderr handle, if it was piped. Same semantics as
    /// `Child::stderr.take()`.
    pub fn take_stderr(&mut self) -> Option<ChildStderr> {
        self.inner.as_mut().and_then(|child| child.stderr.take())
    }

    /// Consume self and wait for the child to exit, collecting stdout
    /// and stderr. The signal handler may have already killed the
    /// child via the registered termination target; in that case wait returns
    /// normally with a non-zero status.
    #[expect(
        clippy::expect_used,
        reason = "ScopedChild owns inner until one terminal wait method consumes it"
    )]
    pub fn wait_with_output(mut self) -> io::Result<Output> {
        let child = self.inner.take().expect("inner already taken");
        let id = self.id.take();
        let result = child.wait_with_output();
        if let Some(id) = id {
            registry::deregister(id);
        }
        result
    }

    /// Wait for the child to exit, returning the status. Same signal-cleanup
    /// semantics as `wait_with_output`.
    #[expect(
        clippy::expect_used,
        reason = "ScopedChild owns inner until one terminal wait method consumes it"
    )]
    pub fn wait(mut self) -> io::Result<ExitStatus> {
        let mut child = self.inner.take().expect("inner already taken");
        let id = self.id.take();
        let result = child.wait();
        if let Some(id) = id {
            registry::deregister(id);
        }
        result
    }
}

fn terminate_failed_setup(child: &mut Child) {
    let _ = crate::process_tree::cleanup_std_child(None, child);
}

impl Drop for ScopedChild {
    fn drop(&mut self) {
        if let Some(mut child) = self.inner.take() {
            let running = !matches!(child.try_wait(), Ok(Some(_)));
            if running {
                let _ = crate::process_tree::cleanup_std_child(
                    self.process_tree.as_deref(),
                    &mut child,
                );
            }
        }
        if let Some(id) = self.id.take() {
            registry::deregister(id);
        }
    }
}

/// Convenience: spawn and wait for exit, returning the status.
pub fn status(command: &mut Command) -> io::Result<ExitStatus> {
    let scoped = ScopedChild::spawn(command)?;
    scoped.wait()
}

/// Convenience: spawn and collect full output (stdout + stderr).
///
/// Mirrors `Command::output` semantics by unconditionally setting
/// stdout / stderr to piped and stdin to null. Callers that need
/// different stdio (e.g. inherited stdin for interactive prompts)
/// must use `ScopedChild::spawn` directly and drive the wait
/// themselves.
pub fn output(command: &mut Command) -> io::Result<Output> {
    command
        .stdin(Stdio::null())
        .stdout(Stdio::piped())
        .stderr(Stdio::piped());
    ScopedChild::spawn(command)?.wait_with_output()
}

#[cfg(test)]
#[expect(
    clippy::expect_used,
    reason = "test setup failures should fail at the exact setup operation"
)]
mod tests {
    use super::*;

    #[cfg(unix)]
    fn assert_deregistered(id: u64) {
        registry::deregister(id);
    }

    #[test]
    #[cfg(unix)]
    fn scoped_child_drop_deregisters() {
        let mut cmd = Command::new("true");
        let child = ScopedChild::spawn(&mut cmd).expect("spawn true");
        let id = child.id.expect("freshly spawned wrapper has an id");
        assert!(id > 0);
        drop(child);
        assert_deregistered(id);
    }

    #[test]
    #[cfg(unix)]
    fn scoped_child_drop_terminates_and_reaps_a_running_child() {
        let mut command = Command::new("sleep");
        command.arg("30");
        let child = ScopedChild::spawn(&mut command).expect("spawn sleep");
        let pid = child.id();

        drop(child);

        assert!(
            !registry::pid_is_alive(pid),
            "running child {pid} survived ScopedChild::drop"
        );
    }

    #[test]
    #[cfg(unix)]
    fn scoped_child_wait_deregisters_and_succeeds() {
        let mut cmd = Command::new("true");
        let child = ScopedChild::spawn(&mut cmd).expect("spawn true");
        let id = child.id.expect("freshly spawned wrapper has an id");
        let status = child.wait().expect("wait true");
        assert!(status.success());
        assert_deregistered(id);
    }

    #[test]
    #[cfg(unix)]
    fn output_helper_collects_stdout() {
        let mut cmd = Command::new("echo");
        cmd.arg("hello")
            .stdout(Stdio::piped())
            .stderr(Stdio::piped());
        let output = output(&mut cmd).expect("echo");
        assert!(output.status.success());
        assert_eq!(output.stdout, b"hello\n");
    }

    #[cfg(any(unix, windows))]
    #[test]
    fn nested_managed_process_tree_terminates_inherited_child() {
        const HELPER_ENV: &str = "FALLOW_NESTED_PROCESS_TREE_TEST";
        const ROOT_ENV: &str = "FALLOW_NESTED_PROCESS_TREE_ROOT";
        const TEST_NAME: &str =
            "scoped_child::tests::nested_managed_process_tree_terminates_inherited_child";

        match std::env::var(HELPER_ENV).ok().as_deref() {
            Some("nested") => {
                let root = std::env::var_os(ROOT_ENV).expect("nested helper root");
                std::fs::write(
                    std::path::Path::new(&root).join("nested.pid"),
                    std::process::id().to_string(),
                )
                .expect("write nested PID");
                std::thread::sleep(std::time::Duration::from_secs(30));
                return;
            }
            Some("outer") => {
                let root = std::env::var_os(ROOT_ENV).expect("outer helper root");
                let mut command =
                    Command::new(std::env::current_exe().expect("current test executable"));
                command
                    .args(["--exact", TEST_NAME, "--nocapture"])
                    .env(HELPER_ENV, "nested")
                    .env(ROOT_ENV, root);
                let child =
                    ScopedChild::spawn_process_tree(&mut command).expect("spawn nested helper");
                let _ = child.wait();
                return;
            }
            _ => {}
        }

        let root = tempfile::tempdir().expect("temporary nested process-tree root");
        let mut command = Command::new(std::env::current_exe().expect("current test executable"));
        command
            .args(["--exact", TEST_NAME, "--nocapture"])
            .env(HELPER_ENV, "outer")
            .env(ROOT_ENV, root.path());
        let outer = ScopedChild::spawn_process_tree(&mut command).expect("spawn outer helper");
        let terminator = outer
            .process_tree_terminator()
            .expect("outer process-tree terminator");
        let pid_path = root.path().join("nested.pid");
        let ready_deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
        while !pid_path.exists() && std::time::Instant::now() < ready_deadline {
            std::thread::sleep(std::time::Duration::from_millis(20));
        }
        let nested_pid = std::fs::read_to_string(&pid_path)
            .expect("nested PID")
            .trim()
            .parse::<u32>()
            .expect("numeric nested PID");

        terminator.terminate().expect("terminate outer tree");
        let status = outer.wait().expect("reap outer helper");
        assert!(!status.success(), "outer helper was not terminated");

        let exit_deadline = std::time::Instant::now() + std::time::Duration::from_secs(3);
        while registry::pid_is_alive(nested_pid) && std::time::Instant::now() < exit_deadline {
            std::thread::sleep(std::time::Duration::from_millis(20));
        }
        assert!(
            !registry::pid_is_alive(nested_pid),
            "nested managed child {nested_pid} survived outer cleanup"
        );
    }

    #[cfg(windows)]
    #[test]
    fn windows_job_object_terminates_descendants_without_taskkill_lookup() {
        const HELPER_ENV: &str = "FALLOW_WINDOWS_JOB_OBJECT_TEST_ROOT";
        const TASKKILL_MARKER_ENV: &str = "FALLOW_FAKE_TASKKILL_MARKER";
        const TEST_NAME: &str = "scoped_child::tests::windows_job_object_terminates_descendants_without_taskkill_lookup";

        if let Some(root) = std::env::var_os(HELPER_ENV) {
            run_windows_job_object_helper(std::path::Path::new(&root));
            return;
        }

        let root = tempfile::tempdir().expect("temporary Windows Job Object root");
        compile_fake_taskkill(root.path(), TASKKILL_MARKER_ENV);
        std::fs::write(
            root.path().join("descendant.cmd"),
            "@echo off\r\necho ready>descendant-ready\r\nping.exe -n 30 127.0.0.1 >NUL\r\n",
        )
        .expect("write descendant script");
        std::fs::write(
            root.path().join("leader.cmd"),
            "@echo off\r\nstart \"\" /B cmd.exe /D /S /C call descendant.cmd\r\nping.exe -n 30 127.0.0.1 >NUL\r\n",
        )
        .expect("write leader script");
        let mut search_paths = vec![root.path().to_path_buf()];
        if let Some(path) = std::env::var_os("PATH") {
            search_paths.extend(std::env::split_paths(&path));
        }
        let search_path =
            std::env::join_paths(search_paths).expect("prepend fake taskkill to PATH");

        let output = Command::new(std::env::current_exe().expect("current test executable"))
            .args(["--exact", TEST_NAME, "--nocapture"])
            .current_dir(root.path())
            .env(HELPER_ENV, root.path())
            .env(TASKKILL_MARKER_ENV, root.path().join("taskkill-invoked"))
            .env("PATH", search_path)
            .env_remove("NoDefaultCurrentDirectoryInExePath")
            .output()
            .expect("run Windows Job Object helper");

        assert!(
            output.status.success(),
            "helper failed: {}",
            String::from_utf8_lossy(&output.stderr)
        );
        assert!(
            !root.path().join("taskkill-invoked").exists(),
            "cleanup executed project-local taskkill"
        );
    }

    #[cfg(windows)]
    fn compile_fake_taskkill(root: &std::path::Path, marker_env: &str) {
        let source = root.join("fake-taskkill.rs");
        let executable = root.join("taskkill.exe");
        std::fs::write(
            &source,
            format!(
                "fn main() {{ let marker = std::env::var_os({marker_env:?}).expect(\"marker path\"); std::fs::write(marker, b\"invoked\").expect(\"write marker\"); }}"
            ),
        )
        .expect("write fake taskkill source");
        let rustc = std::env::var_os("RUSTC").unwrap_or_else(|| "rustc".into());
        let output = Command::new(rustc)
            .args(["--edition=2024", "-o"])
            .arg(&executable)
            .arg(&source)
            .output()
            .expect("compile fake taskkill executable");
        assert!(
            output.status.success(),
            "fake taskkill compilation failed: {}",
            String::from_utf8_lossy(&output.stderr)
        );
    }

    #[cfg(windows)]
    fn run_windows_job_object_helper(root: &std::path::Path) {
        use std::time::{Duration, Instant};

        let mut command = Command::new("cmd.exe");
        command
            .args(["/D", "/S", "/C", "call leader.cmd"])
            .current_dir(root)
            .stdin(Stdio::null())
            .stdout(Stdio::null())
            .stderr(Stdio::null());
        let child = ScopedChild::spawn_process_tree(&mut command).expect("spawn Windows job tree");
        let terminator = child
            .process_tree_terminator()
            .expect("Windows process-tree terminator");
        let ready = root.join("descendant-ready");
        let ready_deadline = Instant::now() + Duration::from_secs(5);
        while !ready.exists() && Instant::now() < ready_deadline {
            std::thread::sleep(Duration::from_millis(20));
        }
        assert!(ready.exists(), "descendant did not start inside the job");

        let started = Instant::now();
        terminator
            .terminate()
            .expect("terminate Windows Job Object");
        let status = child.wait().expect("reap Windows job leader");
        assert!(
            !status.success(),
            "terminated job leader exited successfully"
        );
        assert!(
            started.elapsed() < Duration::from_secs(2),
            "wait remained blocked after Job Object termination"
        );

        let exit_deadline = Instant::now() + Duration::from_secs(2);
        while terminator.is_alive() && Instant::now() < exit_deadline {
            std::thread::sleep(Duration::from_millis(20));
        }
        assert!(
            !terminator.is_alive(),
            "job descendants survived termination"
        );
    }
}