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::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}
104
105impl<T: Sink + ?Sized> Sink for std::sync::Arc<T> {
106    fn event(&self, event: &Event) {
107        (**self).event(event);
108    }
109    fn warning(&self, line: &str) {
110        (**self).warning(line);
111    }
112    fn output(&self, stderr: bool, bytes: &[u8]) {
113        (**self).output(stderr, bytes);
114    }
115}
116
117/// Prints events and warnings as `qk:` lines on stderr, each in one write so
118/// task processes sharing stderr cannot split them, and passes output through.
119pub struct Lines;
120
121impl Sink for Lines {
122    fn event(&self, event: &Event) {
123        let line = event.to_string();
124        if !line.is_empty() {
125            write_all(true, format!("{line}\n").as_bytes());
126        }
127    }
128
129    fn warning(&self, line: &str) {
130        write_all(true, format!("{line}\n").as_bytes());
131    }
132
133    fn output(&self, stderr: bool, bytes: &[u8]) {
134        write_all(stderr, bytes);
135    }
136}
137
138pub fn write_all(stderr: bool, bytes: &[u8]) {
139    let _ = if stderr {
140        io::stderr().lock().write_all(bytes)
141    } else {
142        io::stdout().lock().write_all(bytes)
143    };
144}
145
146static SINK: OnceLock<Box<dyn Sink>> = OnceLock::new();
147
148/// Installs the sink for the rest of the process; only the first one counts.
149pub fn install(sink: Box<dyn Sink>) {
150    let _ = SINK.set(sink);
151}
152
153fn sink() -> &'static dyn Sink {
154    SINK.get().map_or(&Lines, |sink| sink.as_ref())
155}
156
157pub fn event(event: Event) {
158    sink().event(&event);
159}
160
161pub fn warning(line: &str) {
162    sink().warning(line);
163}
164
165pub fn output(stderr: bool, bytes: &[u8]) {
166    if !bytes.is_empty() {
167        sink().output(stderr, bytes);
168    }
169}
qk_executor/report.rs

qk_executor/
report.rs
