running_process_platform_internal/

sync_spawn_group.rs

use std::io;
use std::process::{Command, Stdio};
use std::sync::{Arc, Mutex};
use std::thread;
use std::time::{Duration, Instant};

const DEFAULT_KILL_DRAIN_TIMEOUT: Duration = Duration::from_secs(2);

fn kill_drain_deadline() -> Instant {
    let timeout = crate::env_vars::KILL_DRAIN_TIMEOUT_MS.millis_or(DEFAULT_KILL_DRAIN_TIMEOUT);
    Instant::now() + timeout
}

fn poll_until<T>(
    deadline: Instant,
    interval: Duration,
    mut poll: impl FnMut() -> io::Result<Option<T>>,
) -> io::Result<Option<T>> {
    loop {
        if let Some(value) = poll()? {
            return Ok(Some(value));
        }
        let now = Instant::now();
        if now >= deadline {
            return Ok(None);
        }
        thread::sleep(interval.min(deadline.saturating_duration_since(now)));
    }
}

trait UnixChild: Send {
    fn kill(&mut self) -> io::Result<()>;
    fn wait(&mut self) -> io::Result<i32>;
    fn try_wait(&mut self) -> io::Result<Option<i32>>;
}

impl UnixChild for std::process::Child {
    fn kill(&mut self) -> io::Result<()> {
        std::process::Child::kill(self)
    }

    fn wait(&mut self) -> io::Result<i32> {
        std::process::Child::wait(self).map(crate::platform::process::exit_code)
    }

    fn try_wait(&mut self) -> io::Result<Option<i32>> {
        Ok(std::process::Child::try_wait(self)?.map(crate::platform::process::exit_code))
    }
}

impl crate::platform::process::DaemonChildControl for std::process::Child {
    fn kill(&mut self) -> io::Result<()> {
        std::process::Child::kill(self)
    }

    fn wait(&mut self) -> io::Result<i32> {
        std::process::Child::wait(self).map(crate::platform::process::exit_code)
    }

    fn try_wait(&mut self) -> io::Result<Option<i32>> {
        Ok(std::process::Child::try_wait(self)?.map(crate::platform::process::exit_code))
    }
}

pub struct SpawnedInner {
    child: Arc<Mutex<Option<Box<dyn UnixChild>>>>,
    pgid: i32,
    retain_exit_identity: bool,
}

impl SpawnedInner {
    // WNOWAIT observes exit without releasing the leader's PID. New spawn-mode
    // handles retain that identity until group control is finished. If another
    // reaper has consumed it, waitid fails and control must fail closed.
    fn observe_owned_exit(&self) -> io::Result<Option<i32>> {
        super::observe_owned_child_exit(self.pgid)
    }

    pub fn kill(&self) -> io::Result<()> {
        // Try the child first, then the process group, to make sure
        // any siblings spawned inside go down too.
        let mut guard = self.child.lock().expect("child mutex poisoned");
        if self.retain_exit_identity {
            if guard.is_none() {
                return Ok(());
            }
            self.observe_owned_exit()?;
        }
        if let Some(child) = guard.as_mut() {
            let _ = child.kill();
        }
        drop(guard);
        let _ = crate::platform::process::unix_signal_process_group(
            self.pgid,
            crate::platform::process::UnixSignalKind::Kill,
        );
        Ok(())
    }

    pub fn wait(&self) -> io::Result<i32> {
        if self.retain_exit_identity {
            loop {
                if let Some(code) = self.try_wait()? {
                    return Ok(code);
                }
                thread::sleep(Duration::from_millis(10));
            }
        }
        let mut guard = self.child.lock().expect("child mutex poisoned");
        let Some(child) = guard.as_mut() else {
            return Err(io::Error::other("child handle absent"));
        };
        child.wait()
    }

    pub fn try_wait(&self) -> io::Result<Option<i32>> {
        let mut guard = self.child.lock().expect("child mutex poisoned");
        if self.retain_exit_identity {
            return self.observe_owned_exit();
        }
        let Some(child) = guard.as_mut() else {
            return Ok(None);
        };
        child.try_wait()
    }

    pub fn shutdown(&mut self) {
        self.shutdown_with_deadline(kill_drain_deadline());
    }

    fn shutdown_with_deadline(&mut self, deadline: Instant) {
        let identity_owned = !self.retain_exit_identity || self.observe_owned_exit().is_ok();
        let group_signaled = identity_owned && crate::platform::process::unix_signal_process_group(
            self.pgid,
            crate::platform::process::UnixSignalKind::Kill,
        )
        .is_ok();
        let Some(mut child) = self.child.lock().expect("child mutex poisoned").take() else {
            return;
        };
        if !group_signaled && identity_owned {
            let _ = child.kill();
        }
        match poll_until(deadline, Duration::from_millis(10), || child.try_wait()) {
            Ok(Some(_)) => {}
            Ok(None) | Err(_) => spawn_background_reaper(child),
        }
    }
}

impl crate::platform::process::SpawnedChildControl for SpawnedInner {
    #[cfg(feature = "independent-spawn")]
    fn retain_exit_identity(&mut self) {
        self.retain_exit_identity = true;
    }
    fn kill(&mut self) -> io::Result<()> {
        SpawnedInner::kill(self)
    }

    fn wait(&mut self) -> io::Result<i32> {
        SpawnedInner::wait(self)
    }

    fn try_wait(&mut self) -> io::Result<Option<i32>> {
        SpawnedInner::try_wait(self)
    }

    fn shutdown(&mut self) {
        SpawnedInner::shutdown(self);
    }
}

impl Drop for SpawnedInner {
    fn drop(&mut self) {
        // Detached handles do not invoke shutdown. Release their wait ownership
        // without killing the process, including a retained terminal leader.
        if self.retain_exit_identity {
            if let Some(mut child) = self.child.lock().expect("child mutex poisoned").take() {
                if !matches!(child.try_wait(), Ok(Some(_))) {
                    spawn_background_reaper(child);
                }
            }
        }
    }
}

fn spawn_background_reaper(mut child: Box<dyn UnixChild>) {
    thread::spawn(move || {
        // Once ownership is off the caller's teardown path, a blocking wait is
        // the most reliable terminal policy: it reaps exactly once without a
        // retry loop, spinning, or retaining the shared child mutex.
        let _ = child.wait();
    });
}

fn slot_to_stdio(slot: &crate::platform::process::StdioSource<'_>) -> io::Result<Stdio> {
    match slot {
        crate::platform::process::StdioSource::Null => Ok(Stdio::null()),
        crate::platform::process::StdioSource::Parent => Ok(Stdio::inherit()),
        crate::platform::process::StdioSource::File(file) => Ok(Stdio::from(file.try_clone()?)),
        crate::platform::process::StdioSource::Pipe => Ok(Stdio::piped()),
    }
}

fn daemon_slot_to_stdio(
    slot: &crate::platform::process::DaemonStdioSource<'_>,
) -> io::Result<Stdio> {
    match slot {
        crate::platform::process::DaemonStdioSource::Null => Ok(Stdio::null()),
        crate::platform::process::DaemonStdioSource::File(file) => {
            Ok(Stdio::from(file.try_clone()?))
        }
    }
}

pub fn spawn_sync_daemon(
    command: &mut Command,
    stdio: crate::platform::process::DaemonStdio<'_>,
    environment: crate::platform::process::SyncEnvironment,
    _breakaway: bool,
) -> io::Result<crate::platform::process::DaemonChild> {
    spawn_sync_daemon_inner(command, stdio, environment, None)
}

pub fn spawn_sync_daemon_with_inheritance(
    command: &mut Command,
    stdio: crate::platform::process::DaemonStdio<'_>,
    environment: crate::platform::process::SyncEnvironment,
    _breakaway: bool,
    inheritance: crate::platform::process::DaemonExecInheritance,
) -> io::Result<crate::platform::process::DaemonChild> {
    spawn_sync_daemon_inner(command, stdio, environment, Some(inheritance))
}

fn spawn_sync_daemon_inner(
    command: &mut Command,
    stdio: crate::platform::process::DaemonStdio<'_>,
    environment: crate::platform::process::SyncEnvironment,
    inheritance: Option<crate::platform::process::DaemonExecInheritance>,
) -> io::Result<crate::platform::process::DaemonChild> {
    apply_environment(command, environment);
    command
        .stdin(Stdio::null())
        .stdout(daemon_slot_to_stdio(&stdio.stdout)?)
        .stderr(daemon_slot_to_stdio(&stdio.stderr)?);

    match inheritance {
        Some(inheritance) => {
            crate::platform::process::configure_sync_daemon_command_with_inheritance(
                command,
                inheritance,
            )?;
        }
        None => crate::platform::process::configure_sync_daemon_command(command)?,
    }

    let child = crate::platform::ape::spawn_std(command, |command| command.spawn())?;
    let pid = child.id();
    Ok(crate::platform::process::DaemonChild {
        pid,
        inner: Box::new(child),
    })
}

pub fn spawn_sync(
    command: &mut Command,
    stdio: crate::platform::process::SpawnStdio<'_>,
    environment: crate::platform::process::SyncEnvironment,
) -> io::Result<crate::platform::process::SpawnedChild> {
    spawn_sync_inner(command, stdio, environment, false)
}

#[cfg(feature = "independent-spawn")]
pub(crate) fn spawn_sync_owned_daemon(command: &mut Command, stdio: crate::platform::process::SpawnStdio<'_>, environment: crate::platform::process::SyncEnvironment) -> io::Result<crate::platform::process::SpawnedChild> {
    spawn_sync_inner(command, stdio, environment, true)
}

fn spawn_sync_inner(command: &mut Command, stdio: crate::platform::process::SpawnStdio<'_>, environment: crate::platform::process::SyncEnvironment, detached: bool) -> io::Result<crate::platform::process::SpawnedChild> {
    apply_environment(command, environment);
    command.stdin(slot_to_stdio(&stdio.stdin)?);
    command.stdout(slot_to_stdio(&stdio.stdout)?);
    command.stderr(slot_to_stdio(&stdio.stderr)?);

    if detached { crate::platform::process::configure_sync_daemon_command(command)?; }
    else { crate::platform::process::configure_sync_contained_command(command)?; }

    let mut child = crate::platform::ape::spawn_std(command, |command| command.spawn())?;
    let pid = child.id();
    let pgid = pid as i32;

    let stdin = child.stdin.take();
    let stdout = child.stdout.take();
    let stderr = child.stderr.take();

    let child: Arc<Mutex<Option<Box<dyn UnixChild>>>> = Arc::new(Mutex::new(Some(Box::new(child))));

    // Drain watcher: wait for exit, then sleep `drain_timeout`. We
    // don't proactively close anything on Unix — Rust's ChildStdin/etc.
    // own their fds; once the child exits and the kernel ref-counts
    // its copies to zero, parent reads will EOF naturally.
    if let Some(timeout) = stdio.drain_timeout {
        let child_clone = Arc::clone(&child);
        thread::spawn(move || {
            // Borrow child for try_wait.  We do a polling loop so
            // shutdown() taking the inner Child during Drop doesn't
            // wedge us.
            loop {
                {
                    let mut guard = child_clone.lock().expect("child mutex poisoned");
                    match guard.as_mut() {
                        Some(c) => match c.try_wait() {
                            Ok(Some(_)) => break,
                            Ok(None) => {}
                            Err(_) => break,
                        },
                        None => return,
                    }
                }
                // #199: intentional — try_wait poll on the contained
                // child, 50ms cadence inside a bounded outer drain
                // loop. waitpid(WNOHANG)-equivalent semantics.
                thread::sleep(std::time::Duration::from_millis(50));
            }
            // #199: intentional — post-mortem pipe drain. Children's
            // write-ends of the captured stdio pipes are still being
            // closed by the kernel after exit; this gives readers a
            // chance to see the final bytes before the watcher
            // releases its keep-alive.
            thread::sleep(timeout);
        });
    }

    Ok(crate::platform::process::SpawnedChild {
        kill_on_drop: true,
        stdin,
        stdout,
        stderr,
        pid,
        inner: Box::new(SpawnedInner { child, pgid, retain_exit_identity: false }),
    })
}

fn apply_environment(
    command: &mut Command,
    environment: crate::platform::process::SyncEnvironment,
) {
    let crate::platform::process::SyncEnvironment::Explicit(base) = environment else {
        return;
    };

    // `env_clear` also clears Command's mutation map. Preserve additions,
    // overrides, and removals so they are replayed after the selected base.
    let explicit: Vec<_> = command
        .get_envs()
        .map(|(key, value)| (key.to_os_string(), value.map(std::ffi::OsStr::to_os_string)))
        .collect();
    command.env_clear();
    command.envs(base);
    for (key, value) in explicit {
        match value {
            Some(value) => {
                command.env(key, value);
            }
            None => {
                command.env_remove(key);
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::atomic::{AtomicUsize, Ordering};
    use std::sync::{mpsc, Condvar};

    struct FakeChild {
        wait_gate: Arc<(Mutex<bool>, Condvar)>,
        waits: Arc<AtomicUsize>,
        kills: Arc<AtomicUsize>,
    }

    impl UnixChild for FakeChild {
        fn kill(&mut self) -> io::Result<()> {
            self.kills.fetch_add(1, Ordering::SeqCst);
            Ok(())
        }

        fn wait(&mut self) -> io::Result<i32> {
            self.waits.fetch_add(1, Ordering::SeqCst);
            let (lock, condvar) = &*self.wait_gate;
            let mut released = lock.lock().expect("wait gate mutex poisoned");
            while !*released {
                released = condvar.wait(released).expect("wait gate mutex poisoned");
            }
            Ok(0)
        }

        fn try_wait(&mut self) -> io::Result<Option<i32>> {
            self.waits.fetch_add(1, Ordering::SeqCst);
            let released = *self.wait_gate.0.lock().expect("wait gate mutex poisoned");
            Ok(released.then_some(0))
        }
    }

    struct BlockedFixture {
        inner: SpawnedInner,
        child: Arc<Mutex<Option<Box<dyn UnixChild>>>>,
        wait_gate: Arc<(Mutex<bool>, Condvar)>,
        waits: Arc<AtomicUsize>,
        kills: Arc<AtomicUsize>,
    }

    fn blocked_inner() -> BlockedFixture {
        let wait_gate = Arc::new((Mutex::new(false), Condvar::new()));
        let waits = Arc::new(AtomicUsize::new(0));
        let kills = Arc::new(AtomicUsize::new(0));
        let child: Arc<Mutex<Option<Box<dyn UnixChild>>>> =
            Arc::new(Mutex::new(Some(Box::new(FakeChild {
                wait_gate: Arc::clone(&wait_gate),
                waits: Arc::clone(&waits),
                kills: Arc::clone(&kills),
            }))));
        BlockedFixture {
            inner: SpawnedInner {
                child: Arc::clone(&child),
                pgid: i32::MAX,
                retain_exit_identity: false,
            },
            child,
            wait_gate,
            waits,
            kills,
        }
    }

    fn release_wait(wait_gate: &Arc<(Mutex<bool>, Condvar)>) {
        let (lock, condvar) = &**wait_gate;
        *lock.lock().expect("wait gate mutex poisoned") = true;
        condvar.notify_all();
    }

    struct ShutdownOnDrop {
        inner: Option<SpawnedInner>,
        deadline: Instant,
    }

    impl Drop for ShutdownOnDrop {
        fn drop(&mut self) {
            self.inner
                .as_mut()
                .expect("test wrapper missing inner")
                .shutdown_with_deadline(self.deadline);
        }
    }

    #[test]
    fn drop_is_bounded_when_child_wait_does_not_complete() {
        // Regression for #619: SpawnedChild::drop delegates directly to
        // SpawnedInner::shutdown, modeled by this wrapper around the fake child.
        let BlockedFixture {
            inner, wait_gate, ..
        } = blocked_inner();
        let (tx, rx) = mpsc::channel();
        let started = Instant::now();
        let worker = thread::spawn(move || {
            drop(ShutdownOnDrop {
                inner: Some(inner),
                deadline: Instant::now() + Duration::from_millis(50),
            });
            let _ = tx.send(started.elapsed());
        });

        // The property is causal, not a stopwatch reading: Drop must return
        // WITHOUT waiting for the child, so it must report back before the
        // gate is released. A blocked Drop cannot, whatever the machine load.
        //
        // Asserting a wall-clock bound instead conflated that with "finished
        // inside 100ms", which a loaded runner broke by 0.2ms. The window
        // below is generous because it only bounds how long a *failure* takes
        // to detect: a correct Drop returns at its 50ms deadline and never
        // approaches it.
        let timely = rx.recv_timeout(Duration::from_secs(5));
        release_wait(&wait_gate);
        let returned_before_release = timely.is_ok();
        let elapsed = timely
            .or_else(|_| rx.recv_timeout(Duration::from_secs(5)))
            .expect("shutdown did not unblock even after releasing fake child");
        worker.join().expect("shutdown worker panicked");
        assert!(
            returned_before_release,
            "Drop blocked in child.wait() until the fake child was released              (took {elapsed:?}); its deadline should have bounded it"
        );
    }

    #[test]
    fn shutdown_does_not_hold_child_mutex_while_reaping() {
        let BlockedFixture {
            mut inner,
            child,
            wait_gate,
            waits,
            ..
        } = blocked_inner();
        let worker = thread::spawn(move || {
            inner.shutdown_with_deadline(Instant::now() + Duration::from_millis(50));
        });
        let deadline = Instant::now() + Duration::from_secs(1);
        while waits.load(Ordering::SeqCst) == 0 && Instant::now() < deadline {
            thread::yield_now();
        }
        // `waits` counts every try_wait poll as well as the final blocking
        // wait: shutdown polls every 10ms until its 50ms deadline, then hands
        // the child to a background reaper. A test thread descheduled for a
        // poll interval or two observes more than one call, so the property
        // is "reaping has started", not "exactly one call so far".
        assert!(
            waits.load(Ordering::SeqCst) >= 1,
            "fake wait never started"
        );

        let child_mutex_available = child.try_lock().is_ok();
        release_wait(&wait_gate);
        worker.join().expect("shutdown worker panicked");
        assert!(
            child_mutex_available,
            "shutdown held the child mutex across reaping"
        );
    }

    struct ReadyChild {
        polls: Arc<AtomicUsize>,
        waits: Arc<AtomicUsize>,
    }

    impl UnixChild for ReadyChild {
        fn kill(&mut self) -> io::Result<()> {
            Ok(())
        }

        fn wait(&mut self) -> io::Result<i32> {
            self.waits.fetch_add(1, Ordering::SeqCst);
            Ok(0)
        }

        fn try_wait(&mut self) -> io::Result<Option<i32>> {
            self.polls.fetch_add(1, Ordering::SeqCst);
            Ok(Some(0))
        }
    }

    #[test]
    fn shutdown_reaps_ready_child_exactly_once() {
        let polls = Arc::new(AtomicUsize::new(0));
        let waits = Arc::new(AtomicUsize::new(0));
        let child: Arc<Mutex<Option<Box<dyn UnixChild>>>> =
            Arc::new(Mutex::new(Some(Box::new(ReadyChild {
                polls: Arc::clone(&polls),
                waits: Arc::clone(&waits),
            }))));
        let mut inner = SpawnedInner {
            child,
            pgid: i32::MAX,
            retain_exit_identity: false,
        };

        inner.shutdown_with_deadline(Instant::now() + Duration::from_secs(1));

        assert_eq!(polls.load(Ordering::SeqCst), 1);
        assert_eq!(waits.load(Ordering::SeqCst), 0);
    }

    #[test]
    fn shutdown_falls_back_to_direct_kill_when_group_signal_fails() {
        let BlockedFixture {
            mut inner,
            wait_gate,
            kills,
            ..
        } = blocked_inner();
        release_wait(&wait_gate);

        inner.shutdown_with_deadline(Instant::now() + Duration::from_secs(1));

        assert_eq!(kills.load(Ordering::SeqCst), 1);
    }
}