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

            
14
use std::io::{BufRead, BufReader};
15
use std::process::{Child, Command, Stdio};
16
use std::sync::Arc;
17
use std::sync::atomic::{AtomicU64, Ordering};
18
use std::time::{Duration, Instant};
19

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

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

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

            
34
/// As [`run_watched`], with the silence budget supplied.
35
///
36
/// Exists so the stall path itself is testable: a test cannot wait out the real
37
/// five minutes, and a guard that has never fired is not known to work.
38
3
pub fn run_watched_for(
39
3
    program: &str,
40
3
    args: &[&str],
41
3
    env: &[(&str, &str)],
42
3
    stall: Duration,
43
3
) -> Result<()> {
44
3
    let mut command = Command::new(program);
45
3
    command
46
3
        .args(args)
47
3
        .stdout(Stdio::piped())
48
3
        .stderr(Stdio::piped());
49
3
    for (key, value) in env {
50
        command.env(key, value);
51
    }
52
3
    let mut child = command
53
3
        .spawn()
54
3
        .with_context(|| format!("failed to run {program}"))?;
55

            
56
3
    let last_output = Arc::new(AtomicU64::new(0));
57
3
    let started = Instant::now();
58

            
59
3
    let mut pumps = Vec::new();
60
3
    if let Some(stdout) = child.stdout.take() {
61
3
        pumps.push(pump(stdout, Arc::clone(&last_output), started, false));
62
3
    }
63
3
    if let Some(stderr) = child.stderr.take() {
64
3
        pumps.push(pump(stderr, Arc::clone(&last_output), started, true));
65
3
    }
66

            
67
2
    let status = loop {
68
14
        if let Some(status) = child.try_wait().context("waiting on the child")? {
69
2
            break status;
70
12
        }
71
12
        let quiet = started.elapsed().as_secs() - last_output.load(Ordering::Relaxed);
72
12
        if quiet > stall.as_secs() {
73
1
            diagnose(&child, quiet);
74
1
            let _ = child.kill();
75
1
            let _ = child.wait();
76
1
            bail!(
77
                "{program} produced no output for {quiet}s — see the diagnostics above. \
78
                 This is the silent-hang failure mode, not a slow build."
79
            );
80
11
        }
81
11
        std::thread::sleep(Duration::from_millis(500));
82
    };
83

            
84
4
    for pump in pumps {
85
4
        let _ = pump.join();
86
4
    }
87
2
    if !status.success() {
88
1
        bail!("{program} {} exited with {status}", args.join(" "));
89
1
    }
90
1
    Ok(())
91
3
}
92

            
93
/// Forwards a stream line by line, recording when each line arrived.
94
6
fn pump<R: std::io::Read + Send + 'static>(
95
6
    stream: R,
96
6
    last_output: Arc<AtomicU64>,
97
6
    started: Instant,
98
6
    is_stderr: bool,
99
6
) -> std::thread::JoinHandle<()> {
100
6
    std::thread::spawn(move || {
101
7
        for line in BufReader::new(stream).lines().map_while(Result::ok) {
102
7
            last_output.store(started.elapsed().as_secs(), Ordering::Relaxed);
103
7
            if is_stderr {
104
                eprintln!("{line}");
105
7
            } else {
106
7
                println!("{line}");
107
7
            }
108
        }
109
6
    })
110
6
}
111

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

            
121
1
    if let Ok(url) = std::env::var("DATABASE_URL") {
122
1
        eprintln!(">> database activity:");
123
1
        report(
124
1
            "psql",
125
1
            &[
126
1
                &url,
127
1
                "-tAc",
128
1
                "select pid, state, wait_event_type, wait_event, \
129
1
                 left(replace(query, chr(10), ' '), 80) from pg_stat_activity \
130
1
                 where backend_type = 'client backend'",
131
1
            ],
132
1
        );
133
1
        eprintln!(">> locks not granted:");
134
1
        report(
135
1
            "psql",
136
1
            &[
137
1
                &url,
138
1
                "-tAc",
139
1
                "select pid, relation::regclass, mode from pg_locks where not granted",
140
1
            ],
141
1
        );
142
1
    }
143

            
144
    // Thread states of the stuck process tree: futex_wait means a lock,
145
    // epoll_wait means it is waiting on a socket that will never answer.
146
1
    let pid = child.id();
147
1
    eprintln!(">> thread states under pid {pid}:");
148
1
    for entry in std::fs::read_dir(format!("/proc/{pid}/task"))
149
1
        .into_iter()
150
1
        .flatten()
151
1
        .flatten()
152
    {
153
1
        let wchan = std::fs::read_to_string(entry.path().join("wchan")).unwrap_or_default();
154
1
        if !wchan.is_empty() {
155
1
            eprintln!("   tid {:?}: {wchan}", entry.file_name());
156
1
        }
157
    }
158
1
    eprintln!(">> processes:");
159
1
    report("ps", &["-eo", "pid,ppid,etime,stat,pcpu,args"]);
160
1
    eprintln!();
161
1
}
162

            
163
/// Best effort: a diagnostic that cannot run must not mask the stall itself.
164
3
fn report(program: &str, args: &[&str]) {
165
3
    match Command::new(program).args(args).output() {
166
3
        Ok(out) => {
167
23
            for line in String::from_utf8_lossy(&out.stdout).lines().take(40) {
168
23
                eprintln!("   {line}");
169
23
            }
170
        }
171
        Err(err) => eprintln!("   ({program} unavailable: {err})"),
172
    }
173
3
}
174

            
175
#[cfg(test)]
176
mod tests {
177
    use super::*;
178

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

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

            
206
1
        let text = format!("{err}");
207
1
        assert!(text.contains("no output for"), "{text}");
208
1
        assert!(
209
1
            started.elapsed() < Duration::from_secs(30),
210
            "the child was not killed promptly — it ran for {:?}",
211
            started.elapsed()
212
        );
213
1
    }
214

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