xtask/
heartbeat.rs

1//! Keeps a job's log stream alive across long silent phases.
2//!
3//! Forgejo cancels a task whose runner reports nothing for `ZOMBIE_TASK_TIMEOUT`
4//! (about 10 minutes). Phases here are legitimately quiet for longer: an LTO
5//! link printed nothing for ~11 minutes and had its task cancelled *after* the
6//! build had already succeeded, and an instrumented test run is ~24 minutes of
7//! near-silence.
8//!
9//! In-process, unlike the shell wrapper it replaces: the thing that knows which
10//! phase is running is the thing running it, so the pulse can name the phase
11//! without a file passed between two scripts.
12
13use std::sync::atomic::{AtomicBool, Ordering};
14use std::sync::{Arc, Mutex};
15use std::time::Duration;
16
17const PULSE: Duration = Duration::from_secs(60);
18
19/// Announces phases and pulses between them. Stops on drop.
20pub struct Heartbeat {
21    phase: Arc<Mutex<String>>,
22    stop: Arc<AtomicBool>,
23    handle: Option<std::thread::JoinHandle<()>>,
24}
25
26impl Heartbeat {
27    pub fn start() -> Self {
28        let phase = Arc::new(Mutex::new(String::from("starting")));
29        let stop = Arc::new(AtomicBool::new(false));
30
31        let ticker_phase = Arc::clone(&phase);
32        let ticker_stop = Arc::clone(&stop);
33        let handle = std::thread::spawn(move || {
34            // Polled in short slices rather than one long sleep so teardown does
35            // not wait out a whole minute.
36            let mut waited = Duration::ZERO;
37            while !ticker_stop.load(Ordering::Relaxed) {
38                std::thread::sleep(Duration::from_millis(200));
39                waited += Duration::from_millis(200);
40                if waited >= PULSE {
41                    waited = Duration::ZERO;
42                    let current = ticker_phase
43                        .lock()
44                        .map(|p| p.clone())
45                        .unwrap_or_else(|_| "unknown".into());
46                    println!(">> …still in: {current}");
47                }
48            }
49        });
50
51        Self {
52            phase,
53            stop,
54            handle: Some(handle),
55        }
56    }
57
58    /// Names the phase now running, and says so in the log.
59    pub fn phase(&self, name: &str) {
60        println!(">> {name}");
61        if let Ok(mut phase) = self.phase.lock() {
62            *phase = name.to_owned();
63        }
64    }
65}
66
67impl Drop for Heartbeat {
68    fn drop(&mut self) {
69        self.stop.store(true, Ordering::Relaxed);
70        if let Some(handle) = self.handle.take() {
71            let _ = handle.join();
72        }
73    }
74}
xtask/heartbeat.rs

xtask/
heartbeat.rs
