isothrum_runner/
events.rs

1//! The event stream: what an operation does, as JSON values an embedder shows or a command line
2//! prints one per line (`isothrum … --json`). It evolves from Isoloom's lifecycle events (the
3//! `event` tag, `step`/`log`/`check`/`result`), adding the run, the operation, a sequence number,
4//! the time on every event, and its own version `v`, independent of the manifest's.
5//!
6//! Rules: a secret's value never appears (masked before the event is built); `log` lines are
7//! decoded lossily; an operation ends with exactly one `result` or `error`.
8
9mod emitter;
10
11use std::io::Write;
12use std::sync::{Mutex, PoisonError};
13
14use isoloom_manifest::OutputRole;
15use serde::{Deserialize, Serialize};
16
17pub(crate) use emitter::Emitter;
18
19use crate::error::ErrorCode;
20use crate::state::{Op, RunState};
21
22/// The event stream's version.
23pub const EVENTS_VERSION: u32 = 0;
24
25/// Where a step stands.
26#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
27#[serde(rename_all = "snake_case")]
28#[non_exhaustive]
29pub enum StepState {
30    /// Started.
31    Running,
32    /// Finished and succeeded.
33    Done,
34    /// Finished and failed (an `error` follows unless the step's failure is tolerated).
35    Failed,
36    /// Not run: an earlier step failed.
37    Skipped,
38}
39
40/// Which output of a tool a line came from.
41#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
42#[serde(rename_all = "snake_case")]
43#[non_exhaustive]
44pub enum Stream {
45    /// Its standard output.
46    Stdout,
47    /// Its standard error.
48    Stderr,
49}
50
51/// Where a ready condition stands.
52#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
53#[serde(rename_all = "snake_case")]
54#[non_exhaustive]
55pub enum ReadyState {
56    /// Not yet; polled again.
57    Waiting,
58    /// It holds.
59    Ok,
60    /// It didn't hold in time.
61    Failed,
62}
63
64/// What happened.
65#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, schemars::JsonSchema)]
66#[serde(tag = "event", rename_all = "snake_case")]
67#[non_exhaustive]
68pub enum EventKind {
69    /// The run moved from one state to another.
70    State {
71        /// Before.
72        from: RunState,
73        /// After.
74        to: RunState,
75    },
76    /// A step started, finished, failed or was skipped.
77    Step {
78        /// The step's id.
79        step: String,
80        /// Where it stands.
81        state: StepState,
82        /// The manifest's line for it, when it starts.
83        #[serde(default, skip_serializing_if = "Option::is_none")]
84        note: Option<String>,
85    },
86    /// A line a step's tool printed.
87    Log {
88        /// The step's id.
89        step: String,
90        /// Which output.
91        stream: Stream,
92        /// The line, without its line break.
93        line: String,
94    },
95    /// A ready condition was checked.
96    Ready {
97        /// The condition (`step_ok up`).
98        condition: String,
99        /// Where it stands.
100        state: ReadyState,
101        /// What was seen.
102        detail: String,
103    },
104    /// An output's value, once the run is up.
105    Output {
106        /// The output's name.
107        name: String,
108        /// What it is for.
109        role: OutputRole,
110        /// Its value (`***` when sensitive), or null when it can't be read.
111        value: Option<String>,
112        /// Whether the value is sensitive.
113        sensitive: bool,
114    },
115    /// One check's result.
116    Check {
117        /// The check runner's id.
118        runner: String,
119        /// Where it ran from.
120        from: String,
121        /// The check's name.
122        name: String,
123        /// Whether it passed.
124        ok: bool,
125        /// Why it failed (empty when it passed).
126        detail: String,
127    },
128    /// Something the embedder may show, that didn't fail the operation.
129    Warning {
130        /// What kind.
131        code: WarningCode,
132        /// The text.
133        message: String,
134    },
135    /// The operation's outcome, when it ran to its end.
136    Result {
137        /// Whether it reached its goal (false: a check failed).
138        ok: bool,
139        /// What it found or produced.
140        data: serde_json::Value,
141    },
142    /// Why the operation failed.
143    Error {
144        /// The stable code.
145        code: ErrorCode,
146        /// The message.
147        message: String,
148        /// The step that failed, when one did.
149        #[serde(default, skip_serializing_if = "Option::is_none")]
150        step: Option<String>,
151    },
152}
153
154/// One event of an operation.
155#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, schemars::JsonSchema)]
156#[non_exhaustive]
157pub struct Event {
158    /// The stream's version ([`EVENTS_VERSION`]).
159    pub v: u32,
160    /// Its place in the operation, from 1.
161    pub seq: u64,
162    /// When (`YYYY-MM-DDTHH:MM:SSZ`, UTC).
163    pub time: String,
164    /// The run's name.
165    pub run: String,
166    /// The operation.
167    pub op: Op,
168    /// What happened.
169    #[serde(flatten)]
170    pub kind: EventKind,
171}
172
173/// Where an operation's events go. Implemented for closures (`|e: &Event| …`).
174pub trait Events: Send + Sync {
175    /// Receives one event.
176    fn emit(&self, event: &Event);
177}
178
179impl<F: Fn(&Event) + Send + Sync> Events for F {
180    fn emit(&self, event: &Event) {
181        self(event);
182    }
183}
184
185/// Events written as JSON, one per line, flushed each time.
186#[derive(Debug)]
187pub struct JsonLines<W>(Mutex<W>);
188
189impl<W: Write + Send> JsonLines<W> {
190    /// Writes to `w`.
191    pub fn new(w: W) -> JsonLines<W> {
192        JsonLines(Mutex::new(w))
193    }
194}
195
196impl<W: Write + Send> Events for JsonLines<W> {
197    fn emit(&self, event: &Event) {
198        let mut w = self.0.lock().unwrap_or_else(PoisonError::into_inner);
199        let written = serde_json::to_string(event)
200            .map_err(std::io::Error::from)
201            .and_then(|text| writeln!(w, "{text}"))
202            .and_then(|()| w.flush());
203        // A closed output (a pipe whose reader left) mustn't stop the operation: the run's own log
204        // keeps every event.
205        if let Err(e) = written {
206            tracing::debug!(error = %e, "an event wasn't written");
207        }
208    }
209}
210
211/// What kind of warning a `warning` event is, the same in the Rust API and the event stream.
212#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize, schemars::JsonSchema)]
213#[serde(rename_all = "snake_case")]
214#[non_exhaustive]
215pub enum WarningCode {
216    /// A step whose failure the manifest tolerates failed.
217    StepTolerated,
218    /// An embedder's hook failed.
219    HookFailed,
220    /// A check runner failed or ran out of time: its checks that didn't report fail.
221    CheckRunnerFailed,
222    /// The clean-up of a check runner that ran out of time failed.
223    CheckCleanupFailed,
224    /// What an earlier run left is removed before a start.
225    LeftoversRemoved,
226    /// A VM (or its folder) an earlier run left is removed.
227    OrphanRemoved,
228    /// The run last ran on another Vagrant provider: that run is cleared first.
229    OtherProvider,
230    /// A VM's saved state was cut off: it boots fresh.
231    SavedStateDiscarded,
232    /// A start failed on stale state: cleared, then started again.
233    Recovering,
234    /// An apply failed and is tried again.
235    Retrying,
236}
237
238/// Events kept in memory, for an embedder that reads them later (and for tests).
239#[derive(Debug, Default)]
240pub struct Recorder(Mutex<Vec<Event>>);
241
242impl Recorder {
243    /// An empty recorder.
244    #[must_use]
245    pub fn new() -> Recorder {
246        Recorder::default()
247    }
248
249    /// Every event so far, in order.
250    pub fn events(&self) -> Vec<Event> {
251        self.0.lock().unwrap_or_else(PoisonError::into_inner).clone()
252    }
253}
254
255impl Events for Recorder {
256    fn emit(&self, event: &Event) {
257        self.0.lock().unwrap_or_else(PoisonError::into_inner).push(event.clone());
258    }
259}
260
261/// The JSON Schema of one event.
262#[must_use]
263pub fn events_schema() -> serde_json::Value {
264    let mut schema = serde_json::to_value(schemars::schema_for!(Event)).unwrap_or_default();
265    if let Some(fields) = schema.as_object_mut() {
266        fields.insert("title".to_owned(), serde_json::json!("Isothrum runner event, version 0"));
267    }
268    schema
269}
270
271#[cfg(test)]
272mod tests {
273    use super::*;
274
275    fn event(kind: EventKind) -> Event {
276        Event {
277            v: 0,
278            seq: 1,
279            time: "2026-10-15T12:00:00Z".into(),
280            run: "flag-portal".into(),
281            op: Op::Up,
282            kind,
283        }
284    }
285
286    #[test]
287    fn an_event_has_the_documented_shape() {
288        let e = event(EventKind::Step {
289            step: "up".into(),
290            state: StepState::Running,
291            note: Some("Starting…".into()),
292        });
293        assert_eq!(
294            serde_json::to_value(&e).unwrap(),
295            serde_json::json!({"v":0,"seq":1,"time":"2026-10-15T12:00:00Z","run":"flag-portal","op":"up","event":"step","step":"up","state":"running","note":"Starting…"})
296        );
297        let error = event(EventKind::Error {
298            code: ErrorCode::PortInUse,
299            message: "m".into(),
300            step: None,
301        });
302        let back: Event = serde_json::from_value(serde_json::to_value(&error).unwrap()).unwrap();
303        assert_eq!(back, error);
304    }
305
306    #[test]
307    fn json_lines_are_one_event_per_line() {
308        let sink = JsonLines::new(Vec::new());
309        sink.emit(&event(EventKind::State {
310            from: RunState::Absent,
311            to: RunState::Starting,
312        }));
313        sink.emit(&event(EventKind::Result {
314            ok: true,
315            data: serde_json::json!({}),
316        }));
317        let text = String::from_utf8(sink.0.into_inner().unwrap()).unwrap();
318        let lines: Vec<Event> = text.lines().map(|l| serde_json::from_str(l).unwrap()).collect();
319        assert_eq!(lines.len(), 2);
320    }
321
322    #[test]
323    fn the_schema_names_every_event_kind() {
324        let schema = events_schema().to_string();
325        for kind in ["state", "step", "log", "ready", "output", "check", "warning", "result", "error"] {
326            assert!(schema.contains(&format!("\"{kind}\"")), "{kind}");
327        }
328    }
329}
isothrum_runner/events.rs

isothrum_runner/
events.rs
