Lines
96.45 %
Functions
91.67 %
Branches
100 %
//! 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:");
"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()
{
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.
fn a_silent_command_is_stalled_and_killed() {
let err = run_watched_for(
&["-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}");
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.
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");
assert!(text.contains("exited with"), "{text}");
!text.contains("no output for"),
"a failing command must not be misreported as a stall: {text}"