haematite-cli 0.21.8

//! The unreaped leader reserves its process-group id until every signal is sent.

use super::termination::End;
use rustix::process::{Pid, Signal, WaitId, WaitIdOptions, kill_process_group, waitid};
use tokio::process::Child;
use tokio::task::JoinHandle;

pub(super) struct Contained {
    child: Child,
    pid: Pid,
    reaped: bool,
    signalled: bool,
}

impl Contained {
    pub fn new(child: Child) -> anyhow::Result<Self> {
        let pid = child
            .id()
            .and_then(|pid| i32::try_from(pid).ok())
            .and_then(Pid::from_raw)
            .ok_or_else(|| anyhow::anyhow!("ACP child has no usable process-group id"))?;
        Ok(Self {
            child,
            pid,
            reaped: false,
            signalled: false,
        })
    }
    pub const fn child(&mut self) -> &mut Child {
        &mut self.child
    }
    pub fn observe_exit(&self) -> JoinHandle<anyhow::Result<End>> {
        let pid = self.pid;
        tokio::task::spawn_blocking(move || {
            let status = loop {
                match waitid(
                    WaitId::Pid(pid),
                    WaitIdOptions::EXITED | WaitIdOptions::NOWAIT,
                ) {
                    Ok(Some(status)) => break status,
                    Ok(None) => anyhow::bail!("ACP exit observer returned no status"),
                    Err(rustix::io::Errno::INTR) => {}
                    Err(error) => anyhow::bail!("ACP exit observer failed: {error}"),
                }
            };
            status
                .exit_status()
                .map(End::exited)
                .or_else(|| status.terminating_signal().map(End::signalled))
                .ok_or_else(|| anyhow::anyhow!("ACP exit has no status or signal"))
        })
    }
    fn signal(&self, signal: Signal) -> anyhow::Result<()> {
        match kill_process_group(self.pid, signal) {
            Ok(()) | Err(rustix::io::Errno::SRCH) => Ok(()),
            Err(rustix::io::Errno::PERM) if no_living_group(self.pid)? => Ok(()),
            Err(error) => Err(anyhow::anyhow!(
                "acp_group_signal_failed: {signal:?}: {error}"
            )),
        }
    }
    pub fn kill(&mut self) -> anyhow::Result<()> {
        if self.signalled {
            return Ok(());
        }
        if let Err(error) = self.signal(Signal::KILL) {
            return match self.child.start_kill() {
                Ok(()) => Err(error),
                Err(leader) => Err(anyhow::anyhow!(
                    "{error}; acp_leader_kill_failed: {}",
                    leader.kind()
                )),
            };
        }
        self.signalled = true;
        Ok(())
    }
    pub async fn shutdown(
        &mut self,
        exited: &mut Option<JoinHandle<anyhow::Result<End>>>,
    ) -> Vec<anyhow::Result<()>> {
        if self.signalled {
            return Vec::new();
        }
        let mut results = vec![self.signal(Signal::TERM)];
        if exited.is_some() {
            match tokio::time::timeout(std::time::Duration::from_secs(2), observe(exited)).await {
                Ok(status) => results.push(status.map(|_| ())),
                Err(_) => eprintln!("haematite assistant: acp_exit_grace_elapsed; sending KILL"),
            }
        }
        results.push(self.kill());
        if exited.is_some() {
            results.push(observe(exited).await.map(|_| ()));
        }
        results
    }
    pub async fn reap(&mut self) -> anyhow::Result<()> {
        self.child.wait().await?;
        self.reaped = true;
        Ok(())
    }
}

impl Drop for Contained {
    fn drop(&mut self) {
        if !self.reaped
            && !self.signalled
            && let Err(error) = kill_process_group(self.pid, Signal::KILL)
            && error != rustix::io::Errno::SRCH
        {
            eprintln!("haematite assistant group cleanup failed: {error}");
        }
    }
}

fn no_living_group(pid: Pid) -> anyhow::Result<bool> {
    // Darwin reports EPERM when a reserved group contains only zombies.
    let mut command = std::process::Command::new("/bin/ps");
    command
        .args(["-g", &pid.as_raw_nonzero().to_string(), "-o", "pid=,stat="])
        .stdin(std::process::Stdio::null())
        .stdout(std::process::Stdio::piped())
        .stderr(std::process::Stdio::piped());
    let output = crate::process_spawn::locked(|| command.spawn())?.wait_with_output()?;
    anyhow::ensure!(
        matches!(output.status.code(), Some(0 | 1)),
        "ACP group membership unreadable"
    );
    let text = std::str::from_utf8(&output.stdout)?;
    for line in text.lines() {
        let fields: Vec<_> = line.split_whitespace().collect();
        anyhow::ensure!(
            fields.len() == 2 && fields[0].parse::<u32>().is_ok(),
            "ACP group membership invalid"
        );
        if !fields[1].starts_with('Z') {
            return Ok(false);
        }
    }
    Ok(true)
}

pub(super) async fn observe(
    task: &mut Option<JoinHandle<anyhow::Result<End>>>,
) -> anyhow::Result<End> {
    let result = task
        .as_mut()
        .ok_or_else(|| anyhow::anyhow!("acp_exit_already_observed"))?
        .await;
    task.take();
    let status = result.map_err(|_| anyhow::anyhow!("acp_exit_observer_panicked"))??;
    if status.status != 0 {
        eprintln!(
            "haematite assistant: acp_harness_exited: status {}; {}",
            status.status, status.reason
        );
    }
    Ok(status)
}