dataflow_rs/run/
report.rs

1//! What a finished run reports.
2
3use super::RunStatus;
4use crate::base::{Error, ErrorKind, WorkflowIx};
5use crate::observe::{Selection, Trace, WorkflowExit, WorkflowRecord};
6use crate::program::Program;
7use crate::state::Stats;
8use serde::ser::SerializeStruct;
9use serde::{Serialize, Serializer};
10
11/// One workflow considered by a run, as the engine records it.
12#[derive(Debug)]
13pub(crate) struct Record {
14    pub(crate) ix: WorkflowIx,
15    pub(crate) selection: Selection,
16    pub(crate) sweeps: u32,
17    pub(crate) exit: Option<WorkflowExit>,
18}
19
20/// What the run's arena and view did, counted.
21#[doc(hidden)]
22#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
23#[non_exhaustive]
24pub struct ArenaStats {
25    /// Views built.
26    pub views: u32,
27    /// Checkpoints taken.
28    pub checkpoints: u32,
29    /// The largest view spine, in bytes, right after a build.
30    pub live_max: usize,
31    /// The largest run-arena bytes in use.
32    pub peak_arena: usize,
33}
34
35/// The report of one run: how it ended, what each workflow did, and its
36/// trace when one was asked for.
37pub struct Run {
38    program: Program,
39    status: RunStatus,
40    records: Vec<Record>,
41    stats: Stats,
42    /// Boxed: the report is the run future's output, and most runs are
43    /// not traced.
44    trace: Option<Box<Trace>>,
45}
46
47impl Run {
48    pub(crate) fn new(
49        program: Program,
50        status: RunStatus,
51        records: Vec<Record>,
52        stats: Stats,
53        trace: Option<Box<Trace>>,
54    ) -> Self {
55        Self {
56            program,
57            status,
58            records,
59            stats,
60            trace,
61        }
62    }
63
64    /// The run's trace, when [`RunOptions::trace`](crate::RunOptions::trace())
65    /// asked for one: on every status, the step that stopped the run
66    /// included. A run stopped between ops made no step for the stop, and
67    /// one stopped as a sweep starts only the engine's step for the counter
68    /// it wrote; either way the trace replays to the message's context.
69    pub fn trace(&self) -> Option<&Trace> {
70        self.trace.as_deref()
71    }
72
73    /// How the run ended.
74    pub fn status(&self) -> &RunStatus {
75        &self.status
76    }
77
78    /// Every workflow considered, in `WorkflowIx` order, up to and including
79    /// the one that stopped the run.
80    pub fn workflows(&self) -> impl Iterator<Item = WorkflowRecord<'_>> {
81        self.records.iter().map(|r| WorkflowRecord {
82            id: &self.program.workflows()[r.ix.index() as usize].id,
83            index: r.ix,
84            selection: &r.selection,
85            sweeps: r.sweeps,
86            exit: r.exit,
87        })
88    }
89
90    /// How many workflows were selected and ran.
91    pub fn matched(&self) -> usize {
92        self.records
93            .iter()
94            .filter(|r| matches!(r.selection, Selection::Ran))
95            .count()
96    }
97
98    /// May running the message again succeed? True for a timeout and for a
99    /// stop whose error says so.
100    pub fn retryable(&self) -> bool {
101        match &self.status {
102            RunStatus::Stopped(e) => e.is_retryable(),
103            RunStatus::TimedOut => true,
104            _ => false,
105        }
106    }
107
108    /// Did the run end before every workflow was considered?
109    pub fn partial(&self) -> bool {
110        matches!(
111            self.status,
112            RunStatus::Stopped(_) | RunStatus::TimedOut | RunStatus::Cancelled
113        )
114    }
115
116    /// The run's trace, taken out of the report, which keeps none after.
117    /// Take it before [`into_result`](Self::into_result) when the trace is
118    /// wanted on every status: `into_result` drops the report, the trace
119    /// with it, on exactly the runs that stopped.
120    pub fn take_trace(&mut self) -> Option<Trace> {
121        self.trace.take().map(|trace| *trace)
122    }
123
124    /// `Ok(self)` unless the run was stopped, timed out or cancelled, which
125    /// become the error, the report and its trace dropped; take the trace
126    /// first with [`take_trace`](Self::take_trace) to keep it.
127    pub fn into_result(self) -> crate::Result<Self> {
128        match self.status {
129            RunStatus::Stopped(e) => Err(e),
130            RunStatus::TimedOut => Err(Error::new(ErrorKind::TimedOut, "the run timed out")),
131            RunStatus::Cancelled => Err(Error::new(ErrorKind::Cancelled, "the run was cancelled")),
132            RunStatus::Completed | RunStatus::CompletedWithErrors => Ok(self),
133        }
134    }
135
136    /// The run's arena counters.
137    #[doc(hidden)]
138    pub fn arena_stats(&self) -> ArenaStats {
139        ArenaStats {
140            views: self.stats.views,
141            checkpoints: self.stats.checkpoints,
142            live_max: self.stats.live_max,
143            peak_arena: self.stats.peak_arena,
144        }
145    }
146}
147
148/// Writes a run as `{"status", "error"?, "retryable", "partial", "matched",
149/// "workflows"}`: `error` only for a stopped run, `workflows` the
150/// records of [`Run::workflows`]. Neither the trace nor the arena counters
151/// are part of it. A `Run` has no `Deserialize`, since it holds its
152/// program.
153impl Serialize for Run {
154    fn serialize<S: Serializer>(&self, s: S) -> Result<S::Ok, S::Error> {
155        /// The records, written as a sequence.
156        struct Records<'r>(&'r Run);
157        impl Serialize for Records<'_> {
158            fn serialize<S: Serializer>(&self, s: S) -> Result<S::Ok, S::Error> {
159                s.collect_seq(self.0.workflows())
160            }
161        }
162
163        let mut o = s.serialize_struct("Run", 6)?;
164        o.serialize_field("status", &self.status)?;
165        match &self.status {
166            RunStatus::Stopped(e) => o.serialize_field("error", e)?,
167            _ => o.skip_field("error")?,
168        }
169        o.serialize_field("retryable", &self.retryable())?;
170        o.serialize_field("partial", &self.partial())?;
171        o.serialize_field("matched", &self.matched())?;
172        o.serialize_field("workflows", &Records(self))?;
173        o.end()
174    }
175}
176
177#[cfg(feature = "ts")]
178crate::base::ts::ts_decl!(
179    Run,
180    "Run",
181    [
182        " A run as JSON: how it ended and what each workflow did. Neither the trace",
183        " nor the arena counters are part of it.",
184    ],
185    |cfg| {
186        use crate::base::ts::{Prop, object};
187        object(&[
188            Prop::new("status", RunStatus::name(cfg), &[" How the run ended."]),
189            Prop::optional(
190                "error",
191                Error::name(cfg),
192                &[" The error that stopped the run; present only when `status` is `stopped`."],
193            ),
194            Prop::new(
195                "retryable",
196                "boolean",
197                &[
198                    " May running the message again succeed? True for a timeout and for a",
199                    " stop whose error says so.",
200                ],
201            ),
202            Prop::new(
203                "partial",
204                "boolean",
205                &[" Did the run end before every workflow was considered?"],
206            ),
207            Prop::new(
208                "matched",
209                "number",
210                &[" How many workflows were selected and ran."],
211            ),
212            Prop::new(
213                "workflows",
214                Vec::<WorkflowRecord<'static>>::name(cfg),
215                &[
216                    " Every workflow considered, in program order, up to and including the",
217                    " one that stopped the run.",
218                ],
219            ),
220        ])
221    }
222);
223
224impl std::fmt::Debug for Run {
225    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
226        f.debug_struct("Run")
227            .field("status", &self.status)
228            .field("records", &self.records)
229            .finish_non_exhaustive()
230    }
231}
232
233#[cfg(test)]
234mod tests {
235    use super::*;
236    use crate::Registry;
237    use crate::def::{ParseOptions, parse};
238    use crate::program::CompileOptions;
239    use serde_json::json;
240
241    fn program() -> Program {
242        let compiler = Registry::default().compiler(CompileOptions::default());
243        let ws = ["a", "b", "c", "d", "e"].map(|id| {
244            let json = json!({"id": id, "tasks": [
245                {"id": "t", "function": {"name": "log", "input": {"message": "hi"}}}
246            ]});
247            let def = parse(&json, ParseOptions::default()).def.unwrap();
248            compiler.compile_one(&def).unwrap().workflow
249        });
250        compiler.link(ws)
251    }
252
253    fn record(i: u32, selection: Selection, exit: Option<WorkflowExit>) -> Record {
254        Record {
255            ix: WorkflowIx::new(i),
256            selection,
257            sweeps: u32::from(exit.is_some()),
258            exit,
259        }
260    }
261
262    fn records() -> Vec<Record> {
263        vec![
264            record(0, Selection::Ran, Some(WorkflowExit::Completed)),
265            record(1, Selection::SkippedByCondition, None),
266            record(
267                2,
268                Selection::ConditionFailed(Error::new(ErrorKind::Eval, "x")),
269                None,
270            ),
271            record(3, Selection::SkippedByRollout, None),
272            record(4, Selection::Ran, Some(WorkflowExit::Stopped)),
273        ]
274    }
275
276    fn run(status: RunStatus) -> Run {
277        Run::new(program(), status, records(), Stats::default(), None)
278    }
279
280    #[test]
281    fn a_run_reports_its_workflows_and_their_exits() {
282        let r = run(RunStatus::Completed);
283        assert_eq!(r.matched(), 2);
284        assert!(!r.retryable() && !r.partial());
285        let seen: Vec<_> = r.workflows().map(|w| (w.id, w.sweeps, w.exit)).collect();
286        assert_eq!(seen[0], ("a", 1, Some(WorkflowExit::Completed)));
287        assert_eq!(seen[1], ("b", 0, None));
288        assert_eq!(seen[4], ("e", 1, Some(WorkflowExit::Stopped)));
289        assert!(matches!(
290            r.workflows().nth(2).unwrap().selection,
291            Selection::ConditionFailed(_)
292        ));
293        assert!(r.into_result().is_ok());
294        assert!(run(RunStatus::CompletedWithErrors).into_result().is_ok());
295        assert!(!run(RunStatus::CompletedWithErrors).partial());
296
297        let r = run(RunStatus::TimedOut);
298        assert!(r.retryable() && r.partial());
299        let e = r.into_result().unwrap_err();
300        assert_eq!(e.code(), "TIMEOUT_ERROR");
301
302        let r = run(RunStatus::Cancelled);
303        assert!(!r.retryable() && r.partial());
304        assert_eq!(r.into_result().unwrap_err().code(), "CANCELLED");
305
306        let r = run(RunStatus::Stopped(
307            Error::service("X", "m").with_retryable(true),
308        ));
309        assert!(r.retryable() && r.partial());
310        assert_eq!(r.into_result().unwrap_err().code(), "X");
311        let r = run(RunStatus::Stopped(Error::service("X", "m")));
312        assert!(!r.retryable() && r.partial());
313    }
314
315    #[test]
316    fn arena_stats_come_from_the_state_stats() {
317        let stats = Stats {
318            views: 2,
319            checkpoints: 3,
320            live_max: 40,
321            peak_arena: 50,
322            ..Stats::default()
323        };
324        let r = Run::new(program(), RunStatus::Completed, Vec::new(), stats, None);
325        let a = r.arena_stats();
326        assert_eq!(
327            (a.views, a.checkpoints, a.live_max, a.peak_arena),
328            (2, 3, 40, 50)
329        );
330    }
331}
dataflow_rs/run/report.rs

dataflow_rs/run/
report.rs
