dataflow_rs/base/
error.rs

1//! The error type and the wire shape of a recorded error.
2
3use serde::ser::SerializeStruct;
4use serde::{Deserialize, Serialize, Serializer};
5use std::borrow::Cow;
6use std::fmt;
7use std::sync::Arc;
8
9/// Why a run, a compile or a function failed.
10pub struct Error {
11    kind: ErrorKind,
12    message: Cow<'static, str>,
13    retryable: bool,
14    source: Option<Box<dyn std::error::Error + Send + Sync>>,
15}
16
17/// The class of an [`Error`]. It decides the recorded code.
18#[derive(Clone, Debug, PartialEq, Eq)]
19#[non_exhaustive]
20pub enum ErrorKind {
21    /// A configuration is invalid.
22    Config,
23    /// An expression or definition did not compile.
24    Compile,
25    /// An expression failed to evaluate.
26    Eval,
27    /// The operation budget was exceeded.
28    Budget,
29    /// A deadline passed.
30    TimedOut,
31    /// The run was cancelled.
32    Cancelled,
33    /// A function reported its own code.
34    Service {
35        /// The code the function chose.
36        code: Cow<'static, str>,
37    },
38}
39
40/// The code of a task's own failure: a `Service` error with no code, and a
41/// 5xx status's failure.
42pub(crate) const TASK_ERROR: &str = "TASK_ERROR";
43/// The record a 5xx status adds.
44pub(crate) const TASK_STATUS_ERROR: &str = "TASK_STATUS_ERROR";
45/// The record of a failed validation rule, and a built-in's refused input.
46pub(crate) const VALIDATION_ERROR: &str = "VALIDATION_ERROR";
47/// The record of a `map` mapping that failed, with its destination as the
48/// record's `path`.
49pub(crate) const MAPPING_ERROR: &str = "MAPPING_ERROR";
50/// The record of a built-in's expression that failed to evaluate: a
51/// `validation` rule, a `filter` condition.
52pub(crate) const EVALUATION_ERROR: &str = "EVALUATION_ERROR";
53
54impl Error {
55    pub(crate) fn new(kind: ErrorKind, message: impl Into<Cow<'static, str>>) -> Self {
56        Self {
57            retryable: kind == ErrorKind::TimedOut,
58            kind,
59            message: message.into(),
60            source: None,
61        }
62    }
63
64    /// An error carrying a function's own code.
65    pub fn service(
66        code: impl Into<Cow<'static, str>>,
67        message: impl Into<Cow<'static, str>>,
68    ) -> Self {
69        Self {
70            kind: ErrorKind::Service { code: code.into() },
71            message: message.into(),
72            retryable: false,
73            source: None,
74        }
75    }
76
77    /// The class of the error.
78    pub fn kind(&self) -> &ErrorKind {
79        &self.kind
80    }
81
82    /// The human-readable message.
83    pub fn message(&self) -> &str {
84        &self.message
85    }
86
87    /// Overrides whether a retry may succeed.
88    #[must_use]
89    pub fn with_retryable(mut self, yes: bool) -> Self {
90        self.retryable = yes;
91        self
92    }
93
94    /// Attaches the error that caused this one.
95    #[must_use]
96    pub fn with_source(mut self, source: impl std::error::Error + Send + Sync + 'static) -> Self {
97        self.source = Some(Box::new(source));
98        self
99    }
100
101    /// May a retry succeed? A timeout says yes unless told otherwise; everything
102    /// else says no.
103    pub fn is_retryable(&self) -> bool {
104        self.retryable
105    }
106
107    /// Prefixes the message with `context`, keeping kind, retryability and source.
108    #[must_use]
109    pub fn wrap(mut self, context: impl fmt::Display) -> Self {
110        self.message = Cow::Owned(format!("{context}: {}", self.message));
111        self
112    }
113
114    /// The recorded code for this error.
115    pub fn code(&self) -> &str {
116        match &self.kind {
117            ErrorKind::Service { code } if !code.is_empty() => code,
118            _ => self.code_static(),
119        }
120    }
121
122    /// The static code for any non-service error, or for a service error with
123    /// an empty code.
124    pub(crate) fn code_static(&self) -> &'static str {
125        match &self.kind {
126            ErrorKind::Eval => "LOGIC_ERROR",
127            ErrorKind::Budget => "BUDGET_EXCEEDED",
128            ErrorKind::TimedOut => "TIMEOUT_ERROR",
129            ErrorKind::Cancelled => "CANCELLED",
130            ErrorKind::Config => "CONFIG_ERROR",
131            ErrorKind::Compile => "COMPILATION_ERROR",
132            ErrorKind::Service { .. } => TASK_ERROR,
133        }
134    }
135
136    /// The text of this error's record: what v3's `Display` prints for the
137    /// variant `kind` corresponds to, which the recorder wraps as
138    /// "Task {id} error: {text}". A kind v3 has no variant for, and a service
139    /// error (v3's display is the bare message), give the message alone.
140    pub(crate) fn record_text(&self) -> Cow<'_, str> {
141        let prefix = match &self.kind {
142            ErrorKind::Eval => "Logic evaluation error: ",
143            ErrorKind::TimedOut => "Timeout error: ",
144            ErrorKind::Budget
145            | ErrorKind::Config
146            | ErrorKind::Compile
147            | ErrorKind::Cancelled
148            | ErrorKind::Service { .. } => return Cow::Borrowed(&self.message),
149        };
150        Cow::Owned(format!("{prefix}{}", self.message))
151    }
152}
153
154impl fmt::Display for Error {
155    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
156        f.write_str(&self.message)
157    }
158}
159
160impl fmt::Debug for Error {
161    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
162        f.debug_struct("Error")
163            .field("kind", &self.kind)
164            .field("message", &self.message)
165            .field("retryable", &self.is_retryable())
166            .field("source", &self.source)
167            .finish_non_exhaustive()
168    }
169}
170
171impl std::error::Error for Error {
172    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
173        self.source.as_deref().map(|s| s as _)
174    }
175}
176
177/// The result of a fallible v4 operation.
178pub type Result<T> = std::result::Result<T, Error>;
179
180/// One error record, written as v3 wrote one: the same keys, in the same
181/// order, each optional one absent when unset. The engine stamps every
182/// record a task adds with the workflow's and the task's ids and the run's
183/// timestamp, keeping any the record already carries; a workflow's own
184/// failure carries no task id.
185#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
186#[cfg_attr(feature = "ts", derive(ts_rs::TS))]
187#[non_exhaustive]
188pub struct ErrorRecord {
189    /// The error code.
190    pub code: String,
191    /// The human-readable message.
192    pub message: String,
193    /// Where the error happened; `null` when not known, and never absent.
194    pub path: Option<String>,
195    /// The workflow it happened in, shared with the program's own id and
196    /// the audit entry's, so a run allocates it once.
197    #[serde(skip_serializing_if = "Option::is_none")]
198    #[cfg_attr(feature = "ts", ts(optional))]
199    pub workflow_id: Option<Arc<str>>,
200    /// The task it happened in, shared as `workflow_id` is.
201    #[serde(skip_serializing_if = "Option::is_none")]
202    #[cfg_attr(feature = "ts", ts(optional))]
203    pub task_id: Option<Arc<str>>,
204    /// When it happened, shared with every record of the run.
205    #[serde(skip_serializing_if = "Option::is_none")]
206    #[cfg_attr(feature = "ts", ts(optional))]
207    pub timestamp: Option<Arc<str>>,
208    /// Operator-only detail; never to be shown to an untrusted caller.
209    #[serde(default, skip_serializing_if = "Option::is_none")]
210    #[cfg_attr(feature = "ts", ts(optional))]
211    pub detail: Option<String>,
212    /// The fan-out element the error came from.
213    #[serde(default, skip_serializing_if = "Option::is_none")]
214    #[cfg_attr(feature = "ts", ts(optional))]
215    pub element_index: Option<usize>,
216}
217
218impl ErrorRecord {
219    /// A record with a code and a message and nothing else.
220    pub fn new(code: impl Into<String>, message: impl Into<String>) -> Self {
221        Self {
222            code: code.into(),
223            message: message.into(),
224            path: None,
225            workflow_id: None,
226            task_id: None,
227            timestamp: None,
228            detail: None,
229            element_index: None,
230        }
231    }
232
233    /// Fills what the record leaves unset: the workflow's id, the task's id
234    /// (`None` for a workflow's own failure), the run's timestamp `now`,
235    /// and in a fan-out the element's index. Every record the engine keeps
236    /// is stamped here, so a consumer can group records by task. The ids and
237    /// the timestamp are the program's and the run's own `Arc<str>`, shared
238    /// rather than copied.
239    pub(crate) fn stamp(
240        &mut self,
241        workflow_id: &Arc<str>,
242        task_id: Option<&Arc<str>>,
243        now: &Arc<str>,
244        element_index: Option<usize>,
245    ) {
246        if self.workflow_id.is_none() {
247            self.workflow_id = Some(Arc::clone(workflow_id));
248        }
249        if self.task_id.is_none() {
250            self.task_id = task_id.map(Arc::clone);
251        }
252        if self.timestamp.is_none() {
253            self.timestamp = Some(Arc::clone(now));
254        }
255        if self.element_index.is_none() {
256            self.element_index = element_index;
257        }
258    }
259}
260
261impl From<&Error> for ErrorRecord {
262    fn from(error: &Error) -> Self {
263        Self::new(error.code(), error.message())
264    }
265}
266
267/// Writes the error as `{"code", "message", "retryable"}`: [`Error::code`],
268/// [`Error::message`] and [`Error::is_retryable`], in that order.
269impl Serialize for Error {
270    fn serialize<S: Serializer>(&self, s: S) -> std::result::Result<S::Ok, S::Error> {
271        let mut o = s.serialize_struct("Error", 3)?;
272        o.serialize_field("code", self.code())?;
273        o.serialize_field("message", self.message())?;
274        o.serialize_field("retryable", &self.is_retryable())?;
275        o.end()
276    }
277}
278
279// `DataflowError`, not `Error`: a TypeScript `Error` would shadow the
280// global one wherever it is imported.
281#[cfg(feature = "ts")]
282crate::base::ts::ts_decl!(
283    Error,
284    "DataflowError",
285    [
286        " An error as JSON: its code, its message, and whether a retry may",
287        " succeed.",
288    ],
289    |_cfg| {
290        use crate::base::ts::{Prop, object};
291        object(&[
292            Prop::new("code", "string", &[" The recorded code."]),
293            Prop::new("message", "string", &[" The human-readable message."]),
294            Prop::new("retryable", "boolean", &[" May a retry succeed?"]),
295        ])
296    }
297);
298
299#[cfg(test)]
300mod tests {
301    use super::*;
302
303    #[test]
304    fn record_text_prefixes_two_kinds_and_is_the_message_otherwise() {
305        // The record texts are 3.x's, byte for byte.
306        let m = "the message";
307        for (kind, expected) in [
308            (ErrorKind::Eval, "Logic evaluation error: the message"),
309            (ErrorKind::TimedOut, "Timeout error: the message"),
310            (ErrorKind::Budget, m),
311            (ErrorKind::Service { code: "X".into() }, m),
312            (ErrorKind::Config, m),
313            (ErrorKind::Compile, m),
314            (ErrorKind::Cancelled, m),
315        ] {
316            assert_eq!(
317                Error::new(kind.clone(), m).record_text(),
318                expected,
319                "{kind:?}"
320            );
321        }
322    }
323
324    fn kinds() -> Vec<(ErrorKind, &'static str)> {
325        vec![
326            (ErrorKind::Eval, "LOGIC_ERROR"),
327            (ErrorKind::Budget, "BUDGET_EXCEEDED"),
328            (ErrorKind::TimedOut, "TIMEOUT_ERROR"),
329            (ErrorKind::Cancelled, "CANCELLED"),
330            (ErrorKind::Config, "CONFIG_ERROR"),
331            (ErrorKind::Compile, "COMPILATION_ERROR"),
332            (
333                ErrorKind::Service {
334                    code: "X_FAIL".into(),
335                },
336                "X_FAIL",
337            ),
338            (ErrorKind::Service { code: "".into() }, "TASK_ERROR"),
339        ]
340    }
341
342    #[test]
343    fn code_gives_each_row_of_the_table() {
344        for (kind, code) in kinds() {
345            assert_eq!(Error::new(kind, "m").code(), code);
346        }
347    }
348
349    #[test]
350    fn only_a_timeout_is_retryable_by_default() {
351        for (kind, _) in kinds() {
352            let timed_out = kind == ErrorKind::TimedOut;
353            assert_eq!(Error::new(kind, "m").is_retryable(), timed_out);
354        }
355        assert!(Error::service("A", "m").with_retryable(true).is_retryable());
356        assert!(
357            !Error::new(ErrorKind::TimedOut, "m")
358                .with_retryable(false)
359                .is_retryable()
360        );
361    }
362
363    #[test]
364    fn wrap_keeps_kind_retryable_and_source() {
365        let source = std::io::Error::other("disk");
366        let error = Error::new(ErrorKind::Eval, "bad")
367            .with_retryable(true)
368            .with_source(source)
369            .wrap("task t");
370        assert_eq!(error.to_string(), "task t: bad");
371        assert_eq!(error.kind(), &ErrorKind::Eval);
372        assert!(error.is_retryable());
373        let source = std::error::Error::source(&error).expect("source kept");
374        assert_eq!(source.to_string(), "disk");
375    }
376
377    #[test]
378    fn debug_shows_kind_message_and_retryable() {
379        let text = format!("{:?}", Error::new(ErrorKind::Budget, "over"));
380        assert!(text.contains("Budget") && text.contains("over") && text.contains("retryable"));
381    }
382
383    #[test]
384    fn a_stored_record_with_3x_retry_keys_reads_and_writes_without_them() {
385        let stored =
386            r#"{"code":"X","message":"m","path":null,"retry_attempted":true,"retry_count":2}"#;
387        let record: ErrorRecord = serde_json::from_str(stored).unwrap();
388        assert_eq!(
389            serde_json::to_string(&record).unwrap(),
390            r#"{"code":"X","message":"m","path":null}"#
391        );
392    }
393
394    #[test]
395    fn stamp_fills_only_what_the_record_leaves_unset() {
396        let mut record = ErrorRecord::new("X", "m");
397        record.task_id = Some(Arc::from("own"));
398        record.stamp(
399            &Arc::from("w"),
400            Some(&Arc::from("t")),
401            &Arc::from("now"),
402            Some(3),
403        );
404        assert_eq!(
405            serde_json::to_string(&record).unwrap(),
406            r#"{"code":"X","message":"m","path":null,"workflow_id":"w","task_id":"own","timestamp":"now","element_index":3}"#
407        );
408        let mut workflow_own = ErrorRecord::new("X", "m");
409        workflow_own.stamp(&Arc::from("w"), None, &Arc::from("now"), None);
410        assert_eq!(workflow_own.task_id, None);
411    }
412
413    #[test]
414    fn a_record_from_an_error_carries_only_code_and_message() {
415        let record = ErrorRecord::from(&Error::service("X", "boom"));
416        assert_eq!(
417            serde_json::to_string(&record).unwrap(),
418            r#"{"code":"X","message":"boom","path":null}"#
419        );
420    }
421}
dataflow_rs/base/error.rs

dataflow_rs/base/
error.rs
