openquiver_runtime/
journal.rs

1//! What goes in the journal, and how to read it back.
2//!
3//! Writing and reading are one fact in two directions, so they are written
4//! together. A record that encodes one way and decodes another is not a bug that
5//! shows up in a test — it shows up when someone resumes a run from yesterday.
6//!
7//! The vocabulary lives here rather than in `openquiver-store` on purpose: the store
8//! deliberately does not know what an event means, which is what lets this set
9//! grow without a schema migration each time.
10
11use std::collections::BTreeMap;
12
13use openquiver_graph::NodeName;
14use serde::{Deserialize, Serialize};
15
16use crate::RunValue;
17
18pub const RUN_STARTED: &str = "run.started";
19pub const NODE_ENTERED: &str = "node.entered";
20pub const NODE_EXITED: &str = "node.exited";
21pub const PRINT: &str = "print";
22pub const RUN_FINISHED: &str = "run.finished";
23
24/// One thing a run did.
25#[derive(Debug, Clone, PartialEq)]
26pub(crate) enum Record {
27    Started(Started),
28    /// Written before the node touches anything, and paired with
29    /// [`Record::Exited`] afterwards. A node with one and not the other is the
30    /// only state a resumed run cannot decide by itself.
31    Entered(Entered),
32    Exited(Exited),
33    Printed(Printed),
34    Finished,
35}
36
37#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
38pub(crate) struct Started {
39    /// The fingerprint of the plan this run walks.
40    ///
41    /// The workflow's *name* is on the run row in the store and is not repeated
42    /// here. What is recorded is the thing a name cannot tell you: whether the
43    /// graph is still the one the run started on.
44    pub plan: String,
45}
46
47#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
48pub(crate) struct Entered {
49    pub node: NodeName,
50}
51
52#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
53pub(crate) struct Exited {
54    pub node: NodeName,
55    /// What the node put on its output pins.
56    pub outputs: BTreeMap<String, RunValue>,
57    /// Which execution output continues. `None` ends this line of execution.
58    pub exec: Option<String>,
59}
60
61#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
62pub(crate) struct Printed {
63    pub text: String,
64}
65
66/// A journal entry that could not be understood.
67#[derive(Debug, thiserror::Error)]
68pub enum JournalError {
69    #[error("this journal has a `{0}` entry, which this version does not understand")]
70    UnknownKind(String),
71    #[error("a `{kind}` entry could not be read: {source}")]
72    Malformed {
73        kind: String,
74        source: serde_json::Error,
75    },
76}
77
78impl Record {
79    pub(crate) fn kind(&self) -> &'static str {
80        match self {
81            Record::Started(_) => RUN_STARTED,
82            Record::Entered(_) => NODE_ENTERED,
83            Record::Exited(_) => NODE_EXITED,
84            Record::Printed(_) => PRINT,
85            Record::Finished => RUN_FINISHED,
86        }
87    }
88
89    pub(crate) fn payload(&self) -> Result<serde_json::Value, serde_json::Error> {
90        match self {
91            Record::Started(it) => serde_json::to_value(it),
92            Record::Entered(it) => serde_json::to_value(it),
93            Record::Exited(it) => serde_json::to_value(it),
94            Record::Printed(it) => serde_json::to_value(it),
95            Record::Finished => Ok(serde_json::Value::Object(Default::default())),
96        }
97    }
98
99    /// Reads one entry back.
100    ///
101    /// An unrecognised kind is refused rather than skipped. Skipping would mean
102    /// replaying a journal minus the part this build does not know about, and
103    /// silently continuing from a state that never existed is worse than saying
104    /// the journal is too new.
105    pub(crate) fn decode(kind: &str, payload: &serde_json::Value) -> Result<Self, JournalError> {
106        fn read<T: serde::de::DeserializeOwned>(
107            kind: &str,
108            payload: &serde_json::Value,
109        ) -> Result<T, JournalError> {
110            serde_json::from_value(payload.clone()).map_err(|source| JournalError::Malformed {
111                kind: kind.to_owned(),
112                source,
113            })
114        }
115
116        Ok(match kind {
117            RUN_STARTED => Record::Started(read(kind, payload)?),
118            NODE_ENTERED => Record::Entered(read(kind, payload)?),
119            NODE_EXITED => Record::Exited(read(kind, payload)?),
120            PRINT => Record::Printed(read(kind, payload)?),
121            RUN_FINISHED => Record::Finished,
122            other => return Err(JournalError::UnknownKind(other.to_owned())),
123        })
124    }
125}
openquiver_runtime/journal.rs

openquiver_runtime/
journal.rs
