xtask/

watchdog.rs

//! Runs a command that is expected to keep talking, and diagnoses it if it stops.
//!
//! A coverage run hung for over an hour: two web integration tests stopped
//! producing output while the job's heartbeat kept pulsing, so nothing looked
//! wrong until the deadline killed it — with no indication of which tests were
//! responsible. Diagnosing it meant catching the pod live and hand-running
//! `pg_stat_activity` and `/proc` inspection before it disappeared. The next
//! occurrence should arrive already diagnosed.
//!
//! Only for commands whose silence is itself a symptom. The musl link is quiet
//! for twelve minutes and is perfectly healthy; wrapping that would be a way to
//! kill good builds.

use std::io::{BufRead, BufReader};
use std::process::{Child, Command, Stdio};
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::{Duration, Instant};

use anyhow::{Context, Result, bail};

/// How long a watched command may produce nothing before it is declared stuck.
///
/// Generous on purpose: with several test threads, other tests keep completing
/// even while one is slow, so real silence this long means nothing is
/// progressing. The hang this exists for was silent for eleven minutes.
pub const STALL: Duration = Duration::from_secs(300);

/// Runs a command, forwarding its output, and fails if it goes quiet.
pub fn run_watched(program: &str, args: &[&str], env: &[(&str, &str)]) -> Result<()> {
    run_watched_for(program, args, env, STALL)
}

/// As [`run_watched`], with the silence budget supplied.
///
/// Exists so the stall path itself is testable: a test cannot wait out the real
/// five minutes, and a guard that has never fired is not known to work.
pub fn run_watched_for(
    program: &str,
    args: &[&str],
    env: &[(&str, &str)],
    stall: Duration,
) -> Result<()> {
    let mut command = Command::new(program);
    command
        .args(args)
        .stdout(Stdio::piped())
        .stderr(Stdio::piped());
    for (key, value) in env {
        command.env(key, value);
    }
    let mut child = command
        .spawn()
        .with_context(|| format!("failed to run {program}"))?;

    let last_output = Arc::new(AtomicU64::new(0));
    let started = Instant::now();

    let mut pumps = Vec::new();
    if let Some(stdout) = child.stdout.take() {
        pumps.push(pump(stdout, Arc::clone(&last_output), started, false));
    }
    if let Some(stderr) = child.stderr.take() {
        pumps.push(pump(stderr, Arc::clone(&last_output), started, true));
    }

    let status = loop {
        if let Some(status) = child.try_wait().context("waiting on the child")? {
            break status;
        }
        let quiet = started.elapsed().as_secs() - last_output.load(Ordering::Relaxed);
        if quiet > stall.as_secs() {
            diagnose(&child, quiet);
            let _ = child.kill();
            let _ = child.wait();
            bail!(
                "{program} produced no output for {quiet}s — see the diagnostics above. \
                 This is the silent-hang failure mode, not a slow build."
            );
        }
        std::thread::sleep(Duration::from_millis(500));
    };

    for pump in pumps {
        let _ = pump.join();
    }
    if !status.success() {
        bail!("{program} {} exited with {status}", args.join(" "));
    }
    Ok(())
}

/// Forwards a stream line by line, recording when each line arrived.
fn pump<R: std::io::Read + Send + 'static>(
    stream: R,
    last_output: Arc<AtomicU64>,
    started: Instant,
    is_stderr: bool,
) -> std::thread::JoinHandle<()> {
    std::thread::spawn(move || {
        for line in BufReader::new(stream).lines().map_while(Result::ok) {
            last_output.store(started.elapsed().as_secs(), Ordering::Relaxed);
            if is_stderr {
                eprintln!("{line}");
            } else {
                println!("{line}");
            }
        }
    })
}

/// Everything worth knowing about a stuck run, captured before it is killed.
///
/// Each of these had to be gathered by hand from a live pod last time, racing
/// the job's deadline. The database view is the one that mattered: connections
/// sitting `idle / ClientRead` while the client waited on the socket is what
/// identified a desynchronised connection rather than a slow query.
fn diagnose(child: &Child, quiet: u64) {
    eprintln!("\n>> STALL: no output for {quiet}s — collecting diagnostics\n");

    if let Ok(url) = std::env::var("DATABASE_URL") {
        eprintln!(">> database activity:");
        report(
            "psql",
            &[
                &url,
                "-tAc",
                "select pid, state, wait_event_type, wait_event, \
                 left(replace(query, chr(10), ' '), 80) from pg_stat_activity \
                 where backend_type = 'client backend'",
            ],
        );
        eprintln!(">> locks not granted:");
        report(
            "psql",
            &[
                &url,
                "-tAc",
                "select pid, relation::regclass, mode from pg_locks where not granted",
            ],
        );
    }

    // Thread states of the stuck process tree: futex_wait means a lock,
    // epoll_wait means it is waiting on a socket that will never answer.
    let pid = child.id();
    eprintln!(">> thread states under pid {pid}:");
    for entry in std::fs::read_dir(format!("/proc/{pid}/task"))
        .into_iter()
        .flatten()
        .flatten()
    {
        let wchan = std::fs::read_to_string(entry.path().join("wchan")).unwrap_or_default();
        if !wchan.is_empty() {
            eprintln!("   tid {:?}: {wchan}", entry.file_name());
        }
    }
    eprintln!(">> processes:");
    report("ps", &["-eo", "pid,ppid,etime,stat,pcpu,args"]);
    eprintln!();
}

/// Best effort: a diagnostic that cannot run must not mask the stall itself.
fn report(program: &str, args: &[&str]) {
    match Command::new(program).args(args).output() {
        Ok(out) => {
            for line in String::from_utf8_lossy(&out.stdout).lines().take(40) {
                eprintln!("   {line}");
            }
        }
        Err(err) => eprintln!("   ({program} unavailable: {err})"),
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    /// A command that keeps talking must not be killed, however long it runs.
    #[test]
    fn a_chatty_command_is_left_alone() {
        // Prints for longer than a stall check cycle, in small gaps.
        let out = run_watched(
            "bash",
            &["-c", "for i in 1 2 3 4 5; do echo line $i; sleep 0.3; done"],
            &[],
        );
        assert!(
            out.is_ok(),
            "a command producing output must not be stalled: {out:?}"
        );
    }

    /// The guard must actually fire, kill the child, and say so.
    #[test]
    fn a_silent_command_is_stalled_and_killed() {
        let started = Instant::now();
        let err = run_watched_for(
            "bash",
            &["-c", "echo starting; sleep 120"],
            &[],
            Duration::from_secs(2),
        )
        .expect_err("a command silent past its budget must fail");

        let text = format!("{err}");
        assert!(text.contains("no output for"), "{text}");
        assert!(
            started.elapsed() < Duration::from_secs(30),
            "the child was not killed promptly — it ran for {:?}",
            started.elapsed()
        );
    }

    /// A command that fails must still report its failure, not a stall.
    #[test]
    fn a_failing_command_reports_its_exit_status() {
        let err = run_watched("bash", &["-c", "echo working; exit 3"], &[])
            .expect_err("a non-zero exit must be an error");
        let text = format!("{err}");
        assert!(text.contains("exited with"), "{text}");
        assert!(
            !text.contains("no output for"),
            "a failing command must not be misreported as a stall: {text}"
        );
    }
}