processkit 2.2.2

//! [`ProcessGroup`] — a kill-on-drop container for a tree of child processes.

use std::time::Duration;

use tokio::process::{Child, Command};

use crate::error::{Error, Result};
#[cfg(feature = "limits")]
use crate::limits::{LimitKind, LimitReason, ResourceLimits};
use crate::mechanism::Mechanism;
#[cfg(feature = "process-control")]
use crate::signal::Signal;
#[cfg(feature = "stats")]
use crate::stats::ProcessGroupStats;
use crate::sys::Job;

/// Tuning for a [`ProcessGroup`] — graceful-shutdown timing and (with the
/// `limits` feature) resource limits.
///
/// On the Unix graceful path ([`ProcessGroup::shutdown`]): give the tree
/// `shutdown_timeout` to exit after `SIGTERM`, then `SIGKILL` survivors if
/// `escalate_to_kill` is set. On Windows the job kill is atomic, so
/// `shutdown_timeout` is ignored; `escalate_to_kill` is still honored — `false`
/// preserves survivors (the handle closes without `KILL_ON_JOB_CLOSE`).
#[cfg_attr(
    feature = "limits",
    doc = "",
    doc = "[`limits`](Self::limits) caps the whole tree's memory, process count, and CPU;",
    doc = "it is applied at group creation and only where a real container exists (Windows",
    doc = "Job Object or Linux cgroup v2) — see [`ResourceLimits`]."
)]
#[derive(Debug, Clone)]
#[non_exhaustive]
pub struct ProcessGroupOptions {
    /// How long to wait after `SIGTERM` before escalating. Default: 2 seconds.
    pub shutdown_timeout: Duration,
    /// Whether to `SIGKILL` processes that outlive `shutdown_timeout`.
    /// Default: `true`.
    pub escalate_to_kill: bool,
    /// Whole-tree resource caps applied at creation. Default: no limits.
    #[cfg(feature = "limits")]
    pub limits: ResourceLimits,
}

impl Default for ProcessGroupOptions {
    fn default() -> Self {
        Self {
            shutdown_timeout: Duration::from_secs(2),
            escalate_to_kill: true,
            #[cfg(feature = "limits")]
            limits: ResourceLimits::default(),
        }
    }
}

impl ProcessGroupOptions {
    /// How long to wait after `SIGTERM` before escalating to `SIGKILL` (Unix;
    /// default 2 seconds). See [`shutdown`](ProcessGroup::shutdown).
    #[must_use]
    pub fn shutdown_timeout(mut self, timeout: Duration) -> Self {
        self.shutdown_timeout = timeout;
        self
    }

    /// Whether to `SIGKILL` processes that outlive the shutdown grace window
    /// (default `true`).
    #[must_use]
    pub fn escalate_to_kill(mut self, escalate: bool) -> Self {
        self.escalate_to_kill = escalate;
        self
    }
}

#[cfg(feature = "limits")]
impl ProcessGroupOptions {
    /// Cap the tree's total memory at `bytes`. See [`ResourceLimits`] for platform
    /// support.
    #[must_use]
    pub fn max_memory(mut self, bytes: u64) -> Self {
        self.limits.max_memory = Some(bytes);
        self
    }

    /// Cap the number of live processes in the tree at `n`.
    #[must_use]
    pub fn max_processes(mut self, n: u32) -> Self {
        self.limits.max_processes = Some(n);
        self
    }

    /// Cap the tree's CPU at `cores` cores' worth (`0.5` = half a core, `2.0` = two
    /// cores). See [`ResourceLimits::cpu_quota`] for the Windows approximation.
    #[must_use]
    pub fn cpu_quota(mut self, cores: f64) -> Self {
        self.limits.cpu_quota = Some(cores);
        self
    }
}

/// A container that ties the lifetime of a child-process tree to its own.
///
/// Every process spawned into the group — and everything *those* processes
/// spawn — is killed when the group is dropped (kill-on-close), so an exiting or
/// panicking owner never leaks subprocesses. The containment mechanism is
/// platform-specific and observable via [`mechanism`](Self::mechanism).
///
/// Dropping the group performs an immediate **hard** kill. For a graceful
/// `SIGTERM` → wait → `SIGKILL` teardown (Unix), call
/// [`shutdown`](Self::shutdown) instead — `Drop` cannot `await`, so the graceful
/// tier lives in that async method.
///
/// The drop guarantee covers every exit that runs destructors (returns,
/// panics with unwinding). If the owner dies **abruptly** — `SIGKILL`,
/// `std::process::abort` — `Drop` never runs: on Windows the kernel still
/// kills the tree (the job handle closes with the process), elsewhere that
/// hardening is the opt-in
/// [`Command::kill_on_parent_death`](crate::Command::kill_on_parent_death)
/// (Linux, direct child only; unavailable on macOS/BSD).
pub struct ProcessGroup {
    job: Job,
    options: ProcessGroupOptions,
}

// Manual: `Job` is an opaque OS handle.
impl std::fmt::Debug for ProcessGroup {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("ProcessGroup")
            .field("mechanism", &self.mechanism())
            .field("options", &self.options)
            .finish_non_exhaustive()
    }
}

impl ProcessGroup {
    /// Create an empty group with [default options](ProcessGroupOptions).
    ///
    /// # Errors
    ///
    /// [`Error::Io`] if the OS rejects creating the group's containment primitive
    /// (a Job Object on Windows, a cgroup on Linux). The default options set no
    /// resource caps, so no limit-enforcement failure can arise.
    pub fn new() -> Result<Self> {
        Self::with_options(ProcessGroupOptions::default())
    }

    /// Create an empty group with the given options.
    ///
    /// # Errors
    ///
    /// [`Error::Io`] if the OS rejects creating the group's containment primitive.
    #[cfg_attr(
        feature = "limits",
        doc = "",
        doc = "With the `limits` feature, if `options.limits` sets any cap it is enforced",
        doc = "now. When the active mechanism can't honor a requested limit (no",
        doc = "cgroup/Job Object, or a Linux cgroup whose controllers can't be enabled —",
        doc = "see [`ResourceLimits`] for the cgroup-v2 real-root requirement) this",
        doc = "returns [`Error::ResourceLimit`] — rather than handing back an unbounded",
        doc = "group — and an invalid cap value returns it too, with",
        doc = "[`LimitReason::Invalid`](crate::LimitReason::Invalid)."
    )]
    pub fn with_options(options: ProcessGroupOptions) -> Result<Self> {
        #[cfg(feature = "limits")]
        let job = {
            validate_limits(&options.limits)?;
            Job::new(&options.limits).map_err(|source| {
                if options.limits.any() {
                    // A real signal from the backend, not a guess: every
                    // backend reports `ErrorKind::Unsupported` exactly when no
                    // whole-tree container mechanism exists at all on this
                    // platform (macOS/BSD's POSIX-only fallback, a Linux host
                    // with no cgroup v2 mounted) — the same convention
                    // `map_unsupported` relies on for the signal/suspend/resume
                    // paths. Every other failure (Linux delegation/
                    // subtree_control rejected, a Windows Job Object call
                    // failing) means a capable mechanism exists but this
                    // request could not be applied to it.
                    let reason = if source.kind() == std::io::ErrorKind::Unsupported {
                        LimitReason::Unsupported
                    } else {
                        LimitReason::Unenforceable
                    };
                    Error::ResourceLimit {
                        kind: first_requested_kind(&options.limits),
                        reason,
                        detail: source.to_string(),
                    }
                } else {
                    Error::Io(source)
                }
            })?
        };
        #[cfg(not(feature = "limits"))]
        let job = Job::new().map_err(Error::Io)?;
        Ok(Self { job, options })
    }

    /// Spawn `cmd` as a member of this group.
    ///
    /// The returned [`Child`] — and any process it later spawns — belongs to the
    /// group and is reaped when the group is killed or dropped. The caller is
    /// responsible for configuring `cmd`'s stdio; the group only handles
    /// containment. To build a capture-wired `tokio::process::Command` from a
    /// [`Command`](crate::Command), use its `to_tokio_command()` bridge (a hidden
    /// raw-tokio helper), or construct the `tokio::process::Command` directly.
    ///
    /// **Windows:** to make containment race-free the child is created
    /// `CREATE_SUSPENDED`, assigned to the job, then resumed. This **overwrites**
    /// any process-creation flags the caller set on `cmd` (e.g.
    /// `CREATE_NO_WINDOW`) — Win32 exposes no way to read them back and OR the
    /// suspend bit in. The `Command`-driven launch paths (run helpers,
    /// [`start`](Self::start), pipelines) don't have this limitation: their
    /// [`Command::create_no_window`](crate::Command::create_no_window) flag
    /// travels alongside the OS command and is OR'd in. Only this raw escape
    /// hatch forces `CREATE_SUSPENDED` alone.
    /// **Unix:** the group likewise installs a `pre_exec` hook on `cmd` to join
    /// the cgroup / process group.
    ///
    /// These mutations make `cmd` **single-use**: the spawn appends a `pre_exec`
    /// hook (Unix) and re-sets the creation flags (Windows), which would stack if
    /// the same command were spawned twice. **`spawn` takes `cmd` by value** so
    /// that reuse is a compile error rather than a silent hook-stacking footgun —
    /// build a fresh `Command` per spawn. (The crate's own run helpers
    /// already rebuild the OS command per run, so this only ever concerned direct
    /// `spawn` callers.)
    ///
    /// # Errors
    ///
    /// [`Error::Spawn`] if the OS refuses to start `cmd` — the working directory
    /// is bad, permission is denied, and so on. (This raw path reports every
    /// launch failure as [`Error::Spawn`]; the `Command`-driven run helpers, by
    /// contrast, translate a not-found program into [`Error::NotFound`].)
    pub fn spawn(&self, mut cmd: Command) -> Result<Child> {
        self.spawn_with_options(&mut cmd, &crate::sys::SpawnOptions::default())
    }

    /// `spawn`, carrying the per-spawn knobs a raw `tokio::process::Command`
    /// can't (extra Windows creation flags; the setsid/process-group
    /// coordination). The `Command`-driven launch path.
    pub(crate) fn spawn_with_options(
        &self,
        cmd: &mut Command,
        opts: &crate::sys::SpawnOptions,
    ) -> Result<Child> {
        let child = self.job.spawn(cmd, opts).map_err(|source| Error::Spawn {
            program: program_name(cmd),
            source,
        })?;
        Ok(child)
    }

    /// Attach an already-started [`Child`] to this group.
    ///
    /// Only the child itself is moved into the group; processes it has *already*
    /// spawned keep their original containment (future forks are captured).
    ///
    /// On the POSIX process-group mechanism, a child that has already `exec`'d
    /// cannot be re-grouped (POSIX forbids it), so it is tracked
    /// *individually*: the child itself is signalled/killed with the group,
    /// but — unlike on Windows/cgroup — its future forks are not captured.
    /// The caller keeps the [`Child`] handle and is responsible for reaping:
    /// an adopted child that exited but was never awaited probes as alive, so
    /// a graceful [`shutdown`](Self::shutdown) can wait out its full timeout
    /// on the zombie before escalating.
    ///
    /// **Reap promptly (pid-reuse hazard).** An individually-tracked adopted child
    /// is remembered by **pid**. If you let it exit *and be reaped* elsewhere
    /// without dropping/tearing down this group, that pid can be recycled by the OS
    /// to an **unrelated** process — and this group's later teardown would then
    /// signal that stranger. The crate has no start-time identity to detect the
    /// reuse (and macOS's small pid space makes it likelier). Reap an adopted child
    /// through this group's lifetime, or tear the group down when done, so a stale
    /// pid is never carried across a reuse.
    ///
    /// On the containment backends, adopting a child that has already **exited
    /// but not yet been reaped** is a successful no-op (`Ok`) — there is nothing
    /// left to contain — while an **already-reaped** child (one that was
    /// `wait`ed, so its handle/pid is gone) errors, since there is no longer
    /// anything to reference.
    ///
    /// # Errors
    ///
    /// [`Error::Io`] if `child` has already been reaped (awaited), leaving no
    /// live handle/pid to reference. Adopting an exited-but-unreaped child is a
    /// successful no-op.
    #[cfg(feature = "process-control")]
    pub fn adopt(&self, child: &Child) -> Result<()> {
        self.job.adopt(child).map_err(Error::Io)?;
        #[cfg(feature = "tracing")]
        tracing::trace!(
            target: "processkit",
            mechanism = ?self.mechanism(),
            pid = ?child.id(),
            "adopted an externally spawned child"
        );
        Ok(())
    }

    /// Immediately hard-kill every process currently in the group. Idempotent;
    /// the group remains usable for further spawns afterwards.
    ///
    /// This is an unconditional **hard** kill (`SIGKILL` / `cgroup.kill` /
    /// `TerminateJobObject`), not a graceful `SIGTERM` — for a `SIGTERM` → grace →
    /// `SIGKILL` teardown use [`shutdown`](Self::shutdown) /
    /// [`shutdown_ref`](Self::shutdown_ref). The name mirrors the underlying
    /// `Job::kill_all` it delegates to.
    ///
    /// On the legacy per-pid kill fallback (a Linux kernel without `cgroup.kill`,
    /// pre-5.14), a tree that won't drain within the bounded sweep — a fork bomb
    /// still out-spawning, or un-reapable `D`-state zombies — surfaces as an `Err`
    /// rather than a false success; the atomic backends (`cgroup.kill`, Windows
    /// Job Object) don't need to.
    ///
    /// **Limitation (process-group mechanism only).** On macOS/BSD and the Linux
    /// process-group fallback, a member that changed its real/saved uid (a
    /// `sudo`/setuid child) can reject `SIGKILL` with `EPERM` and survive; this is
    /// **not** surfaced as an `Err` here, because on those platforms `EPERM` is
    /// indistinguishable from a `killpg` against an unreaped zombie (a normal,
    /// harmless case), and reporting it would falsely fail an ordinary teardown.
    /// A privileged child under the process-group mechanism can therefore outlive
    /// `kill_all` — the atomic mechanisms (`cgroup.kill`, Job Object) have no such
    /// gap.
    ///
    /// # Errors
    ///
    /// [`Error::Io`], only on the legacy per-pid kill fallback (a pre-5.14 Linux
    /// kernel without `cgroup.kill`), when the tree won't drain within the
    /// bounded sweep. The atomic backends (`cgroup.kill`, Windows Job Object)
    /// never fail here, and a setuid member that rejects `SIGKILL` under the
    /// process-group mechanism is deliberately *not* reported (see above).
    pub fn kill_all(&self) -> Result<()> {
        #[cfg(feature = "tracing")]
        tracing::debug!(
            target: "processkit",
            mechanism = ?self.mechanism(),
            "hard-killing every process in the group"
        );
        self.job.kill_all().map_err(Error::Io)?;
        Ok(())
    }

    /// Broadcast `sig` to every process in the group.
    ///
    /// Best-effort: a member that has already exited is skipped, and an empty
    /// group succeeds trivially.
    ///
    /// # Platform support
    ///
    /// - **Linux (cgroup or process-group fallback), macOS/BSD** — any signal,
    ///   attempted for every live member of the tree.
    /// - **Windows** — only [`Signal::Kill`]; any other signal — including
    ///   [`Signal::Other`] — returns [`Error::Unsupported`].
    ///
    /// `SIGKILL` ([`Signal::Kill`], or `Other(libc::SIGKILL)`) is routed through
    /// the same whole-tree hard kill as [`kill_all`](Self::kill_all)
    /// on every backend (`cgroup.kill` / `killpg` / Job Object terminate), so it
    /// cannot miss a process forked mid-broadcast. Other signals are a per-member
    /// broadcast.
    ///
    /// **Limitation (process-group mechanism only).** On macOS/BSD and the Linux
    /// process-group fallback, delivery failures such as `EPERM` are not surfaced:
    /// signalling is best-effort and returns `Ok`. As with
    /// [`kill_all`](Self::kill_all), `EPERM` cannot be distinguished from signalling
    /// an unreaped zombie, so reporting it would falsely fail a normal operation.
    /// The Linux cgroup mechanism does surface delivery failures for signals other
    /// than `SIGKILL`.
    ///
    /// # Errors
    ///
    /// [`Error::Unsupported`] on Windows for any signal other than
    /// [`Signal::Kill`] (Job Objects have no POSIX signals). On the Linux cgroup
    /// mechanism, [`Error::Io`] if the OS rejects delivering the signal to a
    /// member. Delivery errors are deliberately not reported by the process-group
    /// mechanism (see above).
    #[cfg(feature = "process-control")]
    pub fn signal(&self, sig: Signal) -> Result<()> {
        self.job
            .signal(sig)
            .map_err(|source| map_unsupported(source, format!("signal({sig:?})")))
    }

    /// Suspend (freeze) every process in the group.
    ///
    /// # Platform support
    ///
    /// - **Linux cgroup** — one `cgroup.freeze` write covering the whole subtree
    ///   (kernel ≥ 5.2; older kernels fall back to per-process `SIGSTOP`). The
    ///   freeze is applied by the kernel shortly after the write returns, not
    ///   instantaneously.
    /// - **Linux process-group fallback, macOS/BSD** — `SIGSTOP` to every
    ///   group; an individually-tracked adopted child (see
    ///   [`adopt`](Self::adopt)) is frozen alone — its own descendants keep
    ///   running.
    /// - **Windows** — suspends every thread of every member process. Best-effort
    ///   and not atomic: threads spawned mid-walk can be missed, and Windows keeps
    ///   per-thread suspend *counts*, so nested `suspend` calls stack — N suspends
    ///   need N [`resume`](Self::resume)s. On Unix suspend/resume are idempotent
    ///   (level-triggered).
    ///
    /// A suspended tree can still be hard-killed
    /// ([`kill_all`](Self::kill_all), or dropping the group) — SIGKILL,
    /// `cgroup.kill`, and `TerminateJobObject` all act on frozen processes. The
    /// graceful [`shutdown`](Self::shutdown), however, starts with a `SIGTERM`
    /// that a frozen tree cannot act on until thawed, so it waits out
    /// `shutdown_timeout` and then escalates; call [`resume`](Self::resume) first
    /// for a clean graceful shutdown.
    ///
    /// **Spawning into a suspended group is platform-divergent.** Under the
    /// Linux cgroup mechanism the freeze is *group state*: a child spawned (or
    /// adopted) while the group is suspended joins the frozen cgroup and
    /// **starts frozen** — it does not run until [`resume`](Self::resume).
    /// Worse, the *spawn call itself* can block until then: the child joins
    /// the cgroup before `exec`, so it can freeze before the spawn handshake
    /// completes and [`start`](Self::start) never returns. The Windows and
    /// POSIX process-group mechanisms freeze only the members present at the
    /// call, so a later spawn runs normally. Don't start new work in a
    /// suspended group (e.g. a
    /// [`Supervisor::with_runner(&group)`](crate::Supervisor::with_runner)
    /// restarting into it) — resume first.
    ///
    /// On **Unix**, a graceful [`shutdown`](Self::shutdown) of a suspended group
    /// cannot drain (C7): frozen members don't run their `SIGTERM` handlers (and a
    /// `SIGSTOP`'d member keeps the signal pending), so the graceful phase burns
    /// its full `shutdown_timeout` and then hard-kills — or, under
    /// `escalate_to_kill = false`, spares the still-frozen survivors.
    /// [`resume`](Self::resume) before a graceful shutdown if you want the members
    /// to actually handle the signal. (On **Windows** the point is moot: a graceful
    /// shutdown is a prompt hard kill regardless — there's no soft-signal tier and
    /// no grace wait — so a suspended group is torn down at once, not after the
    /// timeout.)
    ///
    /// # Errors
    ///
    /// [`Error::Unsupported`] if the active mechanism cannot freeze the tree;
    /// otherwise [`Error::Io`] if the OS rejects the freeze / `SIGSTOP`.
    #[cfg(feature = "process-control")]
    pub fn suspend(&self) -> Result<()> {
        self.job
            .suspend()
            .map_err(|source| map_unsupported(source, "suspend"))
    }

    /// Resume a tree suspended by [`suspend`](Self::suspend).
    ///
    /// See [`suspend`](Self::suspend) for the platform matrix and the Windows
    /// suspend-count nesting caveat.
    ///
    /// # Errors
    ///
    /// [`Error::Unsupported`] if the active mechanism cannot thaw the tree;
    /// otherwise [`Error::Io`] if the OS rejects the resume / `SIGCONT`.
    #[cfg(feature = "process-control")]
    pub fn resume(&self) -> Result<()> {
        self.job
            .resume()
            .map_err(|source| map_unsupported(source, "resume"))
    }

    /// The pids of the processes currently in the group.
    ///
    /// A point-in-time snapshot: a returned pid may belong to a process that
    /// exits (or is reaped) immediately afterwards, and a process spawned during
    /// the call may be missing. Useful for diagnostics, dashboards, and targeted
    /// per-pid action.
    ///
    /// # Platform support
    ///
    /// - **Windows** — every pid assigned to the Job Object (the whole tree).
    /// - **Linux cgroup** — every pid in the cgroup (`cgroup.procs`, whole tree).
    /// - **Linux process-group fallback, macOS/BSD** — the tracked **group
    ///   leaders**, plus any individually-tracked adopted child (one pid per
    ///   spawned/adopted child); descendants inside the groups are contained
    ///   but not enumerated. An exited child still counts as a member until it
    ///   is reaped (awaited): the liveness probe sees the not-yet-collected
    ///   process.
    ///
    /// # Errors
    ///
    /// [`Error::Io`] if the group's membership cannot be read (e.g. a failed
    /// `cgroup.procs` read or Job Object query).
    #[cfg(feature = "process-control")]
    pub fn members(&self) -> Result<Vec<u32>> {
        let pids = self.job.members().map_err(Error::Io)?;
        Ok(pids)
    }

    /// Gracefully tear the group down, consuming it.
    ///
    /// On Unix: `SIGTERM` the tree, wait up to `shutdown_timeout`, then `SIGKILL`
    /// survivors when `escalate_to_kill` is set. On Windows the kill is atomic.
    /// Dropping the group instead (without calling this) performs only the hard
    /// kill.
    ///
    /// **Reap your children, or the grace is wasted (POSIX process-group
    /// mechanism only).** On the [`Mechanism::ProcessGroup`](crate::Mechanism)
    /// fallback (macOS/the BSDs, and Linux without a usable cgroup), liveness is
    /// probed by signalling the group id, and an **unreaped zombie still answers**
    /// — its process-group entry survives until the child is `wait`ed. So a child
    /// that exits promptly on `SIGTERM` but whose [`RunningProcess`](crate::RunningProcess)
    /// handle was dropped without being awaited (or is still held un-awaited) reads
    /// as alive for the full `shutdown_timeout`, and `shutdown` then burns the
    /// whole grace plus a pointless `SIGKILL` escalation. Await each child you
    /// start into the group (any consuming verb, or `wait`) so its handle reaps it.
    /// The Windows Job Object and Linux cgroup mechanisms are immune (a process
    /// leaves `cgroup.procs` / the job on *exit*, before reaping).
    ///
    /// When `escalate_to_kill` is set, the final hard kill can surface the same
    /// undrained-tree `Err` as [`kill_all`](Self::kill_all) on the legacy pre-5.14
    /// per-pid fallback.
    ///
    /// Holding the group behind a shared handle (an `Arc`, a long-lived
    /// supervisor) that can't be moved out by value? Use the borrowing twin
    /// [`shutdown_ref`](Self::shutdown_ref) — same teardown, `&self`.
    ///
    /// # Errors
    ///
    /// [`Error::Io`] if the graceful teardown fails — including the same
    /// undrained-tree failure as [`kill_all`](Self::kill_all) on the legacy
    /// pre-5.14 per-pid fallback, when `escalate_to_kill` performs the final hard
    /// kill.
    pub async fn shutdown(self) -> Result<()> {
        self.shutdown_ref().await
    }

    /// Gracefully tear the group down **without consuming it** — the borrowing
    /// twin of [`shutdown`](Self::shutdown), for a group held behind a shared
    /// handle (an `Arc`, a supervisor) that cannot be moved out by value to call
    /// the consuming form.
    ///
    /// Identical teardown to [`shutdown`](Self::shutdown): on Unix, `SIGTERM` the
    /// tree, wait up to the configured
    /// [`shutdown_timeout`](ProcessGroupOptions::shutdown_timeout), then `SIGKILL`
    /// survivors when [`escalate_to_kill`](ProcessGroupOptions::escalate_to_kill)
    /// is set; on Windows the kill is atomic and the timeout is ignored. The group
    /// stays usable afterwards (a re-`shutdown_ref` on an already-drained tree is a
    /// near no-op). Spawning or adopting a new child **re-arms** `Drop`'s kill
    /// backstop for the whole group, so a straggler started after — or
    /// *concurrently with* — this shutdown is still torn down on `Drop`: a
    /// non-escalating shutdown that is still in flight when the child joins cannot
    /// silently strip the newcomer of its backstop (its spare is keyed to a
    /// generation the join bumps). A group left untouched keeps the survivors an
    /// [`escalate_to_kill`](ProcessGroupOptions::escalate_to_kill)` = false`
    /// shutdown chose to spare.
    ///
    /// The same reaping caveat as [`shutdown`](Self::shutdown) applies on the
    /// POSIX process-group mechanism: await each child you started into the group,
    /// or an unreaped zombie reads as alive for the whole grace.
    ///
    /// # Errors
    ///
    /// [`Error::Io`] if the graceful teardown fails (see
    /// [`shutdown`](Self::shutdown) — the same undrained-tree failure on the
    /// legacy per-pid fallback applies).
    pub async fn shutdown_ref(&self) -> Result<()> {
        #[cfg(feature = "tracing")]
        tracing::debug!(
            target: "processkit",
            mechanism = ?self.mechanism(),
            timeout_ms = self.options.shutdown_timeout.as_millis() as u64,
            escalate = self.options.escalate_to_kill,
            "graceful shutdown: TERM, grace, then KILL"
        );
        self.job
            .graceful_shutdown(
                crate::sys::SIGTERM_RAW,
                self.options.shutdown_timeout,
                self.options.escalate_to_kill,
            )
            .await
            .map_err(Error::Io)?;
        Ok(())
    }

    /// Gracefully tear the tree down **without consuming** the group — the
    /// run-level graceful-timeout path holds an `Arc`/`Weak`, not an owned group.
    /// Sends `signal`, waits up to `grace`, then `SIGKILL`s survivors. On Windows
    /// the signal/grace are ignored (atomic job kill). Best-effort: the caller
    /// reaps the child and the group's `Drop` backstops any straggler.
    pub(crate) async fn graceful_terminate(&self, grace: Duration, signal: i32) -> Result<()> {
        self.job
            .graceful_shutdown(signal, grace, true)
            .await
            .map_err(Error::Io)?;
        Ok(())
    }

    /// Snapshot the group's resource usage (active process count and, where the
    /// platform supports it, total CPU time and peak memory). See
    /// [`ProcessGroupStats`].
    ///
    /// # Errors
    ///
    /// [`Error::Io`] if the platform's resource query fails.
    #[cfg(feature = "stats")]
    pub fn stats(&self) -> Result<ProcessGroupStats> {
        let stats = self.job.stats().map_err(Error::Io)?;
        Ok(stats)
    }

    /// Sample [`stats`](Self::stats) on an interval as a
    /// [`Stream`](tokio_stream::Stream) of snapshots. The first sample is taken
    /// immediately; the series ends on the first failure. A zero `every` is
    /// clamped to 1 ms.
    #[cfg(feature = "stats")]
    pub fn sample_stats(&self, every: Duration) -> crate::stats::StatsSampler<'_> {
        crate::stats::StatsSampler::new(self, every)
    }

    /// The containment mechanism actually in effect (see [`Mechanism`]).
    pub fn mechanism(&self) -> Mechanism {
        self.job.mechanism()
    }
}

/// Best-effort program name for error messages.
fn program_name(cmd: &Command) -> String {
    cmd.as_std().get_program().to_string_lossy().into_owned()
}

/// Map a backend `ErrorKind::Unsupported` to the typed [`Error::Unsupported`],
/// passing every other IO failure through unchanged. Unambiguous here: on the
/// signal/suspend/resume paths the only producer of `Unsupported` is the
/// backends' own "this platform can't do that" reporting.
#[cfg(feature = "process-control")]
fn map_unsupported(source: std::io::Error, operation: impl Into<String>) -> Error {
    if source.kind() == std::io::ErrorKind::Unsupported {
        Error::Unsupported {
            operation: operation.into(),
        }
    } else {
        Error::Io(source)
    }
}

/// Reject nonsensical limit values before touching the OS, so a typo surfaces as a
/// clear [`Error::ResourceLimit`] (`reason: Invalid`) rather than an opaque
/// kernel error.
#[cfg(feature = "limits")]
fn validate_limits(limits: &ResourceLimits) -> Result<()> {
    if limits.max_memory == Some(0) {
        return Err(Error::ResourceLimit {
            kind: LimitKind::Memory,
            reason: LimitReason::Invalid,
            detail: "max_memory must be greater than 0".into(),
        });
    }
    if limits.max_processes == Some(0) {
        return Err(Error::ResourceLimit {
            kind: LimitKind::Processes,
            reason: LimitReason::Invalid,
            detail: "max_processes must be greater than 0".into(),
        });
    }
    if let Some(cores) = limits.cpu_quota
        && !(cores.is_finite() && cores > 0.0)
    {
        return Err(Error::ResourceLimit {
            kind: LimitKind::Cpu,
            reason: LimitReason::Invalid,
            detail: "cpu_quota must be a finite value greater than 0".into(),
        });
    }
    Ok(())
}

/// Which limit an enforcement failure (as opposed to a `validate_limits`
/// rejection) should be attributed to, when the backend's error can't be
/// pinned to a single one: the **first** requested limit in `max_memory`,
/// `max_processes`, `cpu_quota` order — see [`LimitKind`]'s doc for why this
/// fixed tie-break is honest rather than arbitrary. `limits.any()` is a
/// precondition (checked by the caller), so at least one arm always matches.
#[cfg(feature = "limits")]
fn first_requested_kind(limits: &ResourceLimits) -> LimitKind {
    if limits.max_memory.is_some() {
        LimitKind::Memory
    } else if limits.max_processes.is_some() {
        LimitKind::Processes
    } else {
        LimitKind::Cpu
    }
}

#[cfg(all(test, feature = "limits"))]
mod tests {
    use super::*;

    #[test]
    fn builders_set_limits() {
        let opts = ProcessGroupOptions::default()
            .max_memory(1024)
            .max_processes(8)
            .cpu_quota(0.5);
        assert_eq!(opts.limits.max_memory, Some(1024));
        assert_eq!(opts.limits.max_processes, Some(8));
        assert_eq!(opts.limits.cpu_quota, Some(0.5));
        assert!(opts.limits.any());
    }

    #[test]
    fn default_options_have_no_limits() {
        let opts = ProcessGroupOptions::default();
        assert!(!opts.limits.any());
    }

    #[test]
    fn validate_rejects_nonsense() {
        for (opts, expected_kind) in [
            (
                ProcessGroupOptions::default().max_memory(0),
                LimitKind::Memory,
            ),
            (
                ProcessGroupOptions::default().max_processes(0),
                LimitKind::Processes,
            ),
            (
                ProcessGroupOptions::default().cpu_quota(0.0),
                LimitKind::Cpu,
            ),
            (
                ProcessGroupOptions::default().cpu_quota(-1.0),
                LimitKind::Cpu,
            ),
            (
                ProcessGroupOptions::default().cpu_quota(f64::NAN),
                LimitKind::Cpu,
            ),
            (
                ProcessGroupOptions::default().cpu_quota(f64::INFINITY),
                LimitKind::Cpu,
            ),
        ] {
            // `validate_limits` classifies as `Invalid` with the specific
            // field that failed — never a guess, and never touching the OS.
            match validate_limits(&opts.limits) {
                Err(Error::ResourceLimit { kind, reason, .. }) => {
                    assert_eq!(kind, expected_kind);
                    assert_eq!(reason, LimitReason::Invalid);
                }
                other => panic!("expected ResourceLimit, got {other:?}"),
            }
            match ProcessGroup::with_options(opts) {
                Err(Error::ResourceLimit { kind, reason, .. }) => {
                    assert_eq!(kind, expected_kind);
                    assert_eq!(reason, LimitReason::Invalid);
                }
                other => panic!("expected ResourceLimit, got {other:?}"),
            }
        }
    }

    #[test]
    fn first_requested_kind_follows_the_documented_tie_break_order() {
        // max_memory wins over the others when several are set...
        let mut limits = ResourceLimits {
            max_memory: Some(1),
            max_processes: Some(1),
            cpu_quota: Some(1.0),
        };
        assert_eq!(first_requested_kind(&limits), LimitKind::Memory);

        // ...then max_processes, when max_memory is unset...
        limits.max_memory = None;
        assert_eq!(first_requested_kind(&limits), LimitKind::Processes);

        // ...and cpu_quota is the last resort.
        limits.max_processes = None;
        assert_eq!(first_requested_kind(&limits), LimitKind::Cpu);
    }
}