qk_executor/
report.rs

1//! Where a run's progress, warnings and task output go. The runner and the
2//! cache report events; the CLI installs a sink that renders them for the
3//! output style. Without one, events print as `qk:` status lines on stderr.
4
5use std::io::{self, Write};
6use std::sync::{Mutex, OnceLock};
7
8use crate::Outcome;
9
10/// How a cacheable task's key was answered.
11#[derive(Clone, Copy, Debug, PartialEq, Eq)]
12pub enum Cached {
13    Hit,
14    RemoteHit,
15    Miss,
16}
17
18#[derive(Clone, Debug, PartialEq, Eq)]
19pub enum Event {
20    /// A run with this many tasks begins.
21    Planned {
22        tasks: usize,
23    },
24    /// A finite task starts; `checking` when it consults the cache first.
25    Started {
26        id: String,
27        checking: bool,
28    },
29    /// A continuous task starts; it runs until its dependents are done.
30    StartedContinuous {
31        id: String,
32    },
33    Cache {
34        id: String,
35        cached: Cached,
36    },
37    Finished {
38        id: String,
39        outcome: Outcome,
40    },
41    /// Not run because a dependency failed.
42    Skipped {
43        id: String,
44    },
45    Stopping {
46        id: String,
47    },
48    Stopped {
49        id: String,
50    },
51}
52
53impl std::fmt::Display for Event {
54    /// The `qk:` status line for the event, or nothing.
55    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
56        match self {
57            Self::Planned { .. } => Ok(()),
58            Self::Started { id, checking: true } => write!(f, "qk: checking {id}"),
59            Self::Started {
60                id,
61                checking: false,
62            } => write!(f, "qk: running {id}"),
63            Self::StartedContinuous { id } => write!(f, "qk: started {id} (continuous)"),
64            Self::Cache {
65                id,
66                cached: Cached::Hit,
67            } => write!(f, "qk: cache hit {id}"),
68            Self::Cache {
69                id,
70                cached: Cached::RemoteHit,
71            } => write!(f, "qk: remote cache hit {id}"),
72            Self::Cache {
73                id,
74                cached: Cached::Miss,
75            } => write!(f, "qk: cache miss {id}"),
76            Self::Finished {
77                id,
78                outcome: Outcome::Success,
79            } => write!(f, "qk: finished {id}"),
80            Self::Finished {
81                id,
82                outcome: Outcome::Failed(code),
83            } => {
84                write!(f, "qk: failed {id} (exit {code})")
85            }
86            Self::Finished {
87                id,
88                outcome: Outcome::Cancelled,
89            } => write!(f, "qk: cancelled {id}"),
90            Self::Skipped { id } => write!(f, "qk: skipped {id} (dependency failed)"),
91            Self::Stopping { id } => write!(f, "qk: stopping {id} (no longer needed)"),
92            Self::Stopped { id } => write!(f, "qk: stopped {id}"),
93        }
94    }
95}
96
97pub trait Sink: Send + Sync {
98    fn event(&self, event: &Event);
99    /// A `qk:` line about something that went wrong or was skipped.
100    fn warning(&self, line: &str);
101    /// Task output, to stdout or stderr.
102    fn output(&self, stderr: bool, bytes: &[u8]);
103    /// Ends one atomic output batch, including any incremental replay chunks.
104    fn output_finished(&self) {}
105}
106
107impl<T: Sink + ?Sized> Sink for std::sync::Arc<T> {
108    fn event(&self, event: &Event) {
109        (**self).event(event);
110    }
111    fn warning(&self, line: &str) {
112        (**self).warning(line);
113    }
114    fn output(&self, stderr: bool, bytes: &[u8]) {
115        (**self).output(stderr, bytes);
116    }
117    fn output_finished(&self) {
118        (**self).output_finished();
119    }
120}
121
122/// Prints events and warnings as `qk:` lines on stderr, each in one write so
123/// task processes sharing stderr cannot split them, and passes output through.
124pub struct Lines;
125
126impl Sink for Lines {
127    fn event(&self, event: &Event) {
128        let line = event.to_string();
129        if !line.is_empty() {
130            write_all(true, format!("{line}\n").as_bytes());
131        }
132    }
133
134    fn warning(&self, line: &str) {
135        write_all(true, format!("{line}\n").as_bytes());
136    }
137
138    fn output(&self, stderr: bool, bytes: &[u8]) {
139        write_all(stderr, bytes);
140    }
141}
142
143pub fn write_all(stderr: bool, bytes: &[u8]) {
144    let _ = if stderr {
145        io::stderr().lock().write_all(bytes)
146    } else {
147        io::stdout().lock().write_all(bytes)
148    };
149}
150
151static SINK: OnceLock<Box<dyn Sink>> = OnceLock::new();
152
153/// Installs the sink for the rest of the process; only the first one counts.
154pub fn install(sink: Box<dyn Sink>) {
155    let _ = SINK.set(sink);
156}
157
158fn sink() -> &'static dyn Sink {
159    SINK.get().map_or(&Lines, |sink| sink.as_ref())
160}
161
162pub fn event(event: Event) {
163    let _guard = OUTPUT.lock().unwrap();
164    sink().event(&event);
165}
166
167pub fn warning(line: &str) {
168    let _guard = OUTPUT.lock().unwrap();
169    sink().warning(line);
170}
171
172// Held logs replay in bounded chunks while keeping each task's complete output
173// together, including its header and GitHub Actions group delimiters.
174static OUTPUT: Mutex<()> = Mutex::new(());
175
176pub(crate) fn output_batch<T>(consume: impl FnOnce(&mut dyn FnMut(bool, &[u8])) -> T) -> T {
177    let _guard = OUTPUT.lock().unwrap();
178    let result = consume(&mut |stderr, bytes| {
179        if !bytes.is_empty() {
180            sink().output(stderr, bytes);
181        }
182    });
183    sink().output_finished();
184    result
185}
186
187pub fn output(stderr: bool, bytes: &[u8]) {
188    output_batch(|output| output(stderr, bytes));
189}
qk_executor/report.rs

qk_executor/
report.rs
