dataflow_rs/base/
error.rs

1//! The error type and the wire shape of a recorded error.
2
3use super::AuthoredPath;
4use serde::ser::SerializeStruct;
5use serde::{Deserialize, Deserializer, Serialize, Serializer};
6use std::borrow::Cow;
7use std::fmt;
8use std::str::FromStr;
9use std::sync::Arc;
10
11mod error_code_serde {
12    use super::{Cow, Deserialize, Deserializer, FromStr, Serializer};
13    use datalogic_rs::ErrorCode;
14
15    pub(super) fn serialize<S: Serializer>(code: &ErrorCode, s: S) -> Result<S::Ok, S::Error> {
16        s.serialize_str(code.as_str())
17    }
18
19    pub(super) fn deserialize<'de, D: Deserializer<'de>>(d: D) -> Result<ErrorCode, D::Error> {
20        let text = Cow::<str>::deserialize(d)?;
21        ErrorCode::from_str(&text).map_err(serde::de::Error::custom)
22    }
23}
24
25/// One step in a compiled expression's structural evaluation path, from
26/// root to failing node.
27///
28/// Mirrors [`datalogic_rs::PathStep`] with `PartialEq + Eq` so a host can
29/// persist, compare, or inspect where inside a rule an evaluation failed
30/// without re-parsing a display string.
31#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
32#[non_exhaustive]
33pub struct ExprPathStep {
34    /// Compile-time node id, matching [`datalogic_rs::Error::node_ids`].
35    pub node_id: u32,
36    /// Operator name at this node, when one applies.
37    #[serde(default, skip_serializing_if = "Option::is_none")]
38    pub operator: Option<String>,
39    /// Zero-based position within the parent node's argument list.
40    #[serde(default, skip_serializing_if = "Option::is_none")]
41    pub arg_index: Option<u32>,
42    /// JSONLogic-flavoured pointer from the root to this node (e.g. `"/if/0/>/0"`).
43    pub json_pointer: String,
44}
45
46impl From<datalogic_rs::PathStep> for ExprPathStep {
47    fn from(step: datalogic_rs::PathStep) -> Self {
48        Self {
49            node_id: step.node_id,
50            operator: step.operator,
51            arg_index: step.arg_index,
52            json_pointer: step.json_pointer,
53        }
54    }
55}
56
57impl From<&datalogic_rs::PathStep> for ExprPathStep {
58    fn from(step: &datalogic_rs::PathStep) -> Self {
59        Self {
60            node_id: step.node_id,
61            operator: step.operator.clone(),
62            arg_index: step.arg_index,
63            json_pointer: step.json_pointer.clone(),
64        }
65    }
66}
67
68/// Structured budget exhaustion details from expression evaluation or run
69/// task accounting.
70#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
71#[non_exhaustive]
72pub struct BudgetCause {
73    /// The operation or task count when the ceiling was crossed.
74    pub used: u64,
75    /// The configured ceiling.
76    pub limit: u64,
77}
78
79impl BudgetCause {
80    /// Budget exhaustion where `used` crossed `limit`.
81    pub const fn new(used: u64, limit: u64) -> Self {
82        Self { used, limit }
83    }
84}
85
86/// Structured diagnostic details for an expression compilation, evaluation
87/// or budget failure.
88///
89/// Carried on [`Error::expr_cause`] and [`ErrorRecord::cause`] for host
90/// diagnostics without changing the sanitized wire serialization of [`Error`]
91/// or [`ErrorRecord`].
92#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
93#[non_exhaustive]
94pub struct ExprCause {
95    /// Machine-stable error code from the expression engine.
96    #[serde(with = "error_code_serde")]
97    pub code: datalogic_rs::ErrorCode,
98    /// The operator that raised the error, when known.
99    #[serde(default, skip_serializing_if = "Option::is_none")]
100    pub operator: Option<String>,
101    /// RFC 6901 JSON Pointer to the failing sub-expression in the authored
102    /// rule, when known.
103    #[serde(default, skip_serializing_if = "Option::is_none")]
104    pub expr_path: Option<String>,
105    /// Structural steps from the root of the compiled expression to the
106    /// failing node.
107    #[serde(default, skip_serializing_if = "Vec::is_empty")]
108    pub steps: Vec<ExprPathStep>,
109    /// Budget exhaustion details when `code` is
110    /// [`ErrorCode::BudgetExceeded`](datalogic_rs::ErrorCode::BudgetExceeded).
111    #[serde(default, skip_serializing_if = "Option::is_none")]
112    pub budget: Option<BudgetCause>,
113    /// Operations spent by this evaluation before failing, when metered.
114    #[serde(default, skip_serializing_if = "Option::is_none")]
115    pub ops: Option<u64>,
116}
117
118impl ExprCause {
119    /// Operations spent by this evaluation before failing, when metered.
120    pub const fn ops_spent(&self) -> Option<u64> {
121        self.ops
122    }
123
124    /// Extracts structured cause metadata from a [`datalogic_rs::Error`] and,
125    /// when available, the compiled [`datalogic_rs::Logic`] it failed in.
126    pub(crate) fn from_datalogic(
127        e: &datalogic_rs::Error,
128        logic: Option<&datalogic_rs::Logic>,
129        ops: Option<u64>,
130    ) -> Self {
131        let (expr_path, steps) = match logic {
132            Some(logic) => {
133                let resolved = e.resolve_path(logic);
134                let expr_path = resolved.last().map(|s| s.json_pointer.clone());
135                let steps = resolved.into_iter().map(ExprPathStep::from).collect();
136                (expr_path, steps)
137            }
138            None => (None, Vec::new()),
139        };
140        let budget = match e.kind {
141            datalogic_rs::ErrorKind::BudgetExceeded { budget, spent } => {
142                Some(BudgetCause::new(spent, budget))
143            }
144            _ => None,
145        };
146        Self {
147            code: e.code(),
148            operator: e.operator().map(str::to_owned),
149            expr_path,
150            steps,
151            budget,
152            ops: ops.or_else(|| budget.as_ref().map(|b| b.used)),
153        }
154    }
155
156    /// Cause for an expression whose evaluated value did not convert to its
157    /// target type `T`.
158    pub(crate) fn type_mismatch(ops: Option<u64>) -> Self {
159        Self {
160            code: datalogic_rs::ErrorCode::TypeError,
161            operator: None,
162            expr_path: Some(String::new()),
163            steps: Vec::new(),
164            budget: None,
165            ops,
166        }
167    }
168
169    /// Cause for a run-level operation or task budget exceeded outside a
170    /// single expression's own ceiling.
171    pub(crate) fn budget_exceeded(used: u64, limit: u64, ops: Option<u64>) -> Self {
172        Self {
173            code: datalogic_rs::ErrorCode::BudgetExceeded,
174            operator: None,
175            expr_path: None,
176            steps: Vec::new(),
177            budget: Some(BudgetCause::new(used, limit)),
178            ops,
179        }
180    }
181}
182
183/// Which workflow execution site produced an [`Error`] or [`ErrorRecord`].
184#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
185#[serde(rename_all = "snake_case")]
186#[non_exhaustive]
187pub enum SiteKind {
188    /// A workflow's top-level `condition`.
189    WorkflowCondition,
190    /// A task group's `condition`.
191    GroupCondition,
192    /// A task's `condition`.
193    TaskCondition,
194    /// A `filter` built-in's `condition`.
195    FilterCondition,
196    /// A `validation` built-in's rule `logic`.
197    ValidationRule,
198    /// A `validation` built-in's rule `message`.
199    ValidationMessage,
200    /// A `map` built-in's mapping destination `path`.
201    MappingPath,
202    /// A `map` built-in's mapping `logic`.
203    MappingLogic,
204    /// A workflow's `loop.over`.
205    LoopOver,
206    /// A task's `for_each.over`.
207    ForEachOver,
208    /// A task's function input or execution.
209    Task,
210}
211
212/// Where in a workflow a runtime [`Error`] or [`ErrorRecord`] originated.
213#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
214#[non_exhaustive]
215pub struct ErrorSite {
216    /// Which kind of site failed.
217    pub kind: SiteKind,
218    /// The workflow's id.
219    pub workflow_id: Arc<str>,
220    /// The task's id, when the site belongs to a task.
221    #[serde(default, skip_serializing_if = "Option::is_none")]
222    pub task_id: Option<Arc<str>>,
223    /// The group's id, when the site is a group condition.
224    #[serde(default, skip_serializing_if = "Option::is_none")]
225    pub group_id: Option<Arc<str>>,
226    /// Zero-based rule index when the site is a `validation` rule.
227    #[serde(default, skip_serializing_if = "Option::is_none")]
228    pub rule_index: Option<usize>,
229    /// Zero-based mapping index when the site is a `map` mapping.
230    #[serde(default, skip_serializing_if = "Option::is_none")]
231    pub mapping_index: Option<usize>,
232    /// Target context path when the site is a `map` mapping with a known destination.
233    #[serde(default, skip_serializing_if = "Option::is_none")]
234    pub target_path: Option<Arc<str>>,
235    /// Where the site sits in the authored workflow definition.
236    #[serde(default, skip_serializing_if = "Option::is_none")]
237    pub authored_path: Option<AuthoredPath>,
238    /// The loop sweep counter, inside a workflow's loop.
239    #[serde(default, skip_serializing_if = "Option::is_none")]
240    pub loop_counter: Option<i64>,
241}
242
243impl ErrorSite {
244    /// A site of `kind` in `workflow_id`.
245    pub fn new(kind: SiteKind, workflow_id: Arc<str>) -> Self {
246        Self {
247            kind,
248            workflow_id,
249            task_id: None,
250            group_id: None,
251            rule_index: None,
252            mapping_index: None,
253            target_path: None,
254            authored_path: None,
255            loop_counter: None,
256        }
257    }
258
259    /// Sets the task id on this site.
260    #[must_use]
261    pub fn with_task(mut self, task_id: Arc<str>) -> Self {
262        self.task_id = Some(task_id);
263        self
264    }
265
266    /// Sets the group id on this site.
267    #[must_use]
268    pub fn with_group(mut self, group_id: Arc<str>) -> Self {
269        self.group_id = Some(group_id);
270        self
271    }
272
273    /// Sets the authored path on this site.
274    #[must_use]
275    pub fn with_path(mut self, path: AuthoredPath) -> Self {
276        self.authored_path = Some(path);
277        self
278    }
279
280    /// Sets the loop sweep counter on this site.
281    #[must_use]
282    pub fn with_loop_counter(mut self, loop_counter: Option<i64>) -> Self {
283        self.loop_counter = loop_counter;
284        self
285    }
286}
287
288/// Why a run, a compile or a function failed.
289#[derive(Clone)]
290pub struct Error {
291    kind: ErrorKind,
292    message: Cow<'static, str>,
293    retryable: bool,
294    source: Option<Arc<dyn std::error::Error + Send + Sync>>,
295    expr_cause: Option<Arc<ExprCause>>,
296    site: Option<Arc<ErrorSite>>,
297}
298
299/// The class of an [`Error`]. It decides the recorded code.
300#[derive(Clone, Debug, PartialEq, Eq)]
301#[non_exhaustive]
302pub enum ErrorKind {
303    /// A configuration is invalid.
304    Config,
305    /// An expression or definition did not compile.
306    Compile,
307    /// An expression failed to evaluate.
308    Eval,
309    /// The operation budget was exceeded.
310    Budget,
311    /// A deadline passed.
312    TimedOut,
313    /// The run was cancelled.
314    Cancelled,
315    /// A function reported its own code.
316    Service {
317        /// The code the function chose.
318        code: Cow<'static, str>,
319    },
320}
321
322/// The code of a task's own failure: a `Service` error with no code, and a
323/// 5xx status's failure.
324pub(crate) const TASK_ERROR: &str = "TASK_ERROR";
325/// The record a 5xx status adds.
326pub(crate) const TASK_STATUS_ERROR: &str = "TASK_STATUS_ERROR";
327/// The record of a failed validation rule, and a built-in's refused input.
328pub(crate) const VALIDATION_ERROR: &str = "VALIDATION_ERROR";
329/// The record of a `map` mapping that failed, with its destination as the
330/// record's `path`.
331pub(crate) const MAPPING_ERROR: &str = "MAPPING_ERROR";
332/// The record of a built-in's expression that failed to evaluate: a
333/// `validation` rule, a `filter` condition.
334pub(crate) const EVALUATION_ERROR: &str = "EVALUATION_ERROR";
335
336impl Error {
337    pub(crate) fn new(kind: ErrorKind, message: impl Into<Cow<'static, str>>) -> Self {
338        Self {
339            retryable: kind == ErrorKind::TimedOut,
340            kind,
341            message: message.into(),
342            source: None,
343            expr_cause: None,
344            site: None,
345        }
346    }
347
348    /// An error carrying a function's own code.
349    pub fn service(
350        code: impl Into<Cow<'static, str>>,
351        message: impl Into<Cow<'static, str>>,
352    ) -> Self {
353        Self {
354            kind: ErrorKind::Service { code: code.into() },
355            message: message.into(),
356            retryable: false,
357            source: None,
358            expr_cause: None,
359            site: None,
360        }
361    }
362
363    /// An error indicating an operation, task, or custom run budget was exceeded.
364    pub fn budget(message: impl Into<Cow<'static, str>>) -> Self {
365        Self::new(ErrorKind::Budget, message)
366    }
367
368    /// The class of the error.
369    pub fn kind(&self) -> &ErrorKind {
370        &self.kind
371    }
372
373    /// The human-readable message.
374    pub fn message(&self) -> &str {
375        &self.message
376    }
377
378    /// Overrides whether a retry may succeed.
379    #[must_use]
380    pub fn with_retryable(mut self, yes: bool) -> Self {
381        self.retryable = yes;
382        self
383    }
384
385    /// Attaches the error that caused this one.
386    #[must_use]
387    pub fn with_source(mut self, source: impl std::error::Error + Send + Sync + 'static) -> Self {
388        self.source = Some(Arc::new(source));
389        self
390    }
391
392    /// The underlying [`datalogic_rs::Error`], when this error came from
393    /// expression compilation or evaluation.
394    pub fn datalogic_error(&self) -> Option<&datalogic_rs::Error> {
395        self.source
396            .as_deref()
397            .and_then(|s| s.downcast_ref::<datalogic_rs::Error>())
398    }
399
400    /// Attaches structured expression diagnostics to this error.
401    #[must_use]
402    pub fn with_expr_cause(mut self, cause: ExprCause) -> Self {
403        self.expr_cause = Some(Arc::new(cause));
404        self
405    }
406
407    /// Structured expression diagnostics, when this error came from an
408    /// expression or a budget check.
409    pub fn expr_cause(&self) -> Option<&ExprCause> {
410        self.expr_cause.as_deref()
411    }
412
413    pub(crate) fn expr_cause_arc(&self) -> Option<Arc<ExprCause>> {
414        self.expr_cause.clone()
415    }
416
417    /// Attaches the workflow site where this error happened.
418    #[must_use]
419    pub fn with_site(mut self, site: ErrorSite) -> Self {
420        self.site = Some(Arc::new(site));
421        self
422    }
423
424    /// Where in the workflow this error happened, when known.
425    pub fn site(&self) -> Option<&ErrorSite> {
426        self.site.as_deref()
427    }
428
429    pub(crate) fn site_arc(&self) -> Option<Arc<ErrorSite>> {
430        self.site.clone()
431    }
432
433    /// May a retry succeed? A timeout says yes unless told otherwise; everything
434    /// else says no.
435    pub fn is_retryable(&self) -> bool {
436        self.retryable
437    }
438
439    /// Prefixes the message with `context`, keeping kind, retryability,
440    /// source, structured cause and workflow site.
441    #[must_use]
442    pub fn wrap(mut self, context: impl fmt::Display) -> Self {
443        self.message = Cow::Owned(format!("{context}: {}", self.message));
444        self
445    }
446
447    /// Replaces the human-readable message while keeping kind, retryability,
448    /// source, structured cause and workflow site.
449    pub(crate) fn with_message(mut self, message: impl Into<Cow<'static, str>>) -> Self {
450        self.message = message.into();
451        self
452    }
453
454    /// The recorded code for this error.
455    pub fn code(&self) -> &str {
456        match &self.kind {
457            ErrorKind::Service { code } if !code.is_empty() => code,
458            _ => self.code_static(),
459        }
460    }
461
462    /// The static code for any non-service error, or for a service error with
463    /// an empty code.
464    pub(crate) fn code_static(&self) -> &'static str {
465        match &self.kind {
466            ErrorKind::Eval => "LOGIC_ERROR",
467            ErrorKind::Budget => "BUDGET_EXCEEDED",
468            ErrorKind::TimedOut => "TIMEOUT_ERROR",
469            ErrorKind::Cancelled => "CANCELLED",
470            ErrorKind::Config => "CONFIG_ERROR",
471            ErrorKind::Compile => "COMPILATION_ERROR",
472            ErrorKind::Service { .. } => TASK_ERROR,
473        }
474    }
475
476    /// The text of this error's record: what v3's `Display` prints for the
477    /// variant `kind` corresponds to, which the recorder wraps as
478    /// "Task {id} error: {text}". A kind v3 has no variant for, and a service
479    /// error (v3's display is the bare message), give the message alone.
480    pub(crate) fn record_text(&self) -> Cow<'_, str> {
481        let prefix = match &self.kind {
482            ErrorKind::Eval => "Logic evaluation error: ",
483            ErrorKind::TimedOut => "Timeout error: ",
484            ErrorKind::Budget
485            | ErrorKind::Config
486            | ErrorKind::Compile
487            | ErrorKind::Cancelled
488            | ErrorKind::Service { .. } => return Cow::Borrowed(&self.message),
489        };
490        Cow::Owned(format!("{prefix}{}", self.message))
491    }
492}
493
494impl fmt::Display for Error {
495    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
496        f.write_str(&self.message)
497    }
498}
499
500impl fmt::Debug for Error {
501    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
502        let mut s = f.debug_struct("Error");
503        s.field("kind", &self.kind)
504            .field("message", &self.message)
505            .field("retryable", &self.is_retryable())
506            .field("source", &self.source);
507        if let Some(cause) = &self.expr_cause {
508            s.field("expr_cause", cause);
509        }
510        if let Some(site) = &self.site {
511            s.field("site", site);
512        }
513        s.finish_non_exhaustive()
514    }
515}
516
517impl std::error::Error for Error {
518    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
519        self.source.as_deref().map(|s| s as _)
520    }
521}
522
523/// The result of a fallible v4 operation.
524pub type Result<T> = std::result::Result<T, Error>;
525
526/// One error record, written as v3 wrote one: the same keys, in the same
527/// order, each optional one absent when unset. The engine stamps every
528/// record a task adds with the workflow's and the task's ids and the run's
529/// timestamp, keeping any the record already carries; a workflow's own
530/// failure carries no task id.
531#[derive(Clone, Debug, Serialize, Deserialize)]
532#[cfg_attr(feature = "ts", derive(ts_rs::TS))]
533#[non_exhaustive]
534pub struct ErrorRecord {
535    /// The error code.
536    pub code: String,
537    /// The human-readable message.
538    pub message: String,
539    /// Where the error happened; `null` when not known, and never absent.
540    pub path: Option<String>,
541    /// The workflow it happened in, shared with the program's own id and
542    /// the audit entry's, so a run allocates it once.
543    #[serde(skip_serializing_if = "Option::is_none")]
544    #[cfg_attr(feature = "ts", ts(optional))]
545    pub workflow_id: Option<Arc<str>>,
546    /// The task it happened in, shared as `workflow_id` is.
547    #[serde(skip_serializing_if = "Option::is_none")]
548    #[cfg_attr(feature = "ts", ts(optional))]
549    pub task_id: Option<Arc<str>>,
550    /// When it happened, shared with every record of the run.
551    #[serde(skip_serializing_if = "Option::is_none")]
552    #[cfg_attr(feature = "ts", ts(optional))]
553    pub timestamp: Option<Arc<str>>,
554    /// Operator-only detail; never to be shown to an untrusted caller.
555    #[serde(default, skip_serializing_if = "Option::is_none")]
556    #[cfg_attr(feature = "ts", ts(optional))]
557    pub detail: Option<String>,
558    /// The fan-out element the error came from.
559    #[serde(default, skip_serializing_if = "Option::is_none")]
560    #[cfg_attr(feature = "ts", ts(optional))]
561    pub element_index: Option<usize>,
562    /// Structured expression diagnostics, carried in-memory for host reports
563    /// and omitted from the sanitized wire representation.
564    #[serde(skip)]
565    #[cfg_attr(feature = "ts", ts(skip))]
566    pub cause: Option<Arc<ExprCause>>,
567    /// Where in the workflow the error happened, carried in-memory for host
568    /// reports and omitted from the sanitized wire representation.
569    #[serde(skip)]
570    #[cfg_attr(feature = "ts", ts(skip))]
571    pub site: Option<Arc<ErrorSite>>,
572}
573
574impl PartialEq for ErrorRecord {
575    fn eq(&self, other: &Self) -> bool {
576        self.code == other.code
577            && self.message == other.message
578            && self.path == other.path
579            && self.workflow_id == other.workflow_id
580            && self.task_id == other.task_id
581            && self.timestamp == other.timestamp
582            && self.detail == other.detail
583            && self.element_index == other.element_index
584    }
585}
586
587impl Eq for ErrorRecord {}
588
589impl ErrorRecord {
590    /// A record with a code and a message and nothing else.
591    pub fn new(code: impl Into<String>, message: impl Into<String>) -> Self {
592        Self {
593            code: code.into(),
594            message: message.into(),
595            path: None,
596            workflow_id: None,
597            task_id: None,
598            timestamp: None,
599            detail: None,
600            element_index: None,
601            cause: None,
602            site: None,
603        }
604    }
605
606    /// Structured expression diagnostics, when this record came from an
607    /// expression or a budget failure.
608    pub fn cause(&self) -> Option<&ExprCause> {
609        self.cause.as_deref()
610    }
611
612    /// Attaches structured expression diagnostics to this record.
613    #[must_use]
614    pub fn with_cause(mut self, cause: ExprCause) -> Self {
615        self.cause = Some(Arc::new(cause));
616        self
617    }
618
619    /// Where in the workflow this error happened, when known.
620    pub fn site(&self) -> Option<&ErrorSite> {
621        self.site.as_deref()
622    }
623
624    /// Attaches the workflow site where this error happened.
625    #[must_use]
626    pub fn with_site(mut self, site: ErrorSite) -> Self {
627        self.site = Some(Arc::new(site));
628        self
629    }
630
631    /// Fills what the record leaves unset: the workflow's id, the task's id
632    /// (`None` for a workflow's own failure), the run's timestamp `now`,
633    /// and in a fan-out the element's index. Every record the engine keeps
634    /// is stamped here, so a consumer can group records by task. The ids and
635    /// the timestamp are the program's and the run's own `Arc<str>`, shared
636    /// rather than copied.
637    pub(crate) fn stamp(
638        &mut self,
639        workflow_id: &Arc<str>,
640        task_id: Option<&Arc<str>>,
641        now: &Arc<str>,
642        element_index: Option<usize>,
643    ) {
644        if self.workflow_id.is_none() {
645            self.workflow_id = Some(Arc::clone(workflow_id));
646        }
647        if self.task_id.is_none() {
648            self.task_id = task_id.map(Arc::clone);
649        }
650        if self.timestamp.is_none() {
651            self.timestamp = Some(Arc::clone(now));
652        }
653        if self.element_index.is_none() {
654            self.element_index = element_index;
655        }
656    }
657}
658
659impl From<&Error> for ErrorRecord {
660    fn from(error: &Error) -> Self {
661        let mut record = Self::new(error.code(), error.message());
662        record.cause = error.expr_cause_arc();
663        record.site = error.site_arc();
664        record
665    }
666}
667
668/// Writes the error as `{"code", "message", "retryable"}`: [`Error::code`],
669/// [`Error::message`] and [`Error::is_retryable`], in that order.
670impl Serialize for Error {
671    fn serialize<S: Serializer>(&self, s: S) -> std::result::Result<S::Ok, S::Error> {
672        let mut o = s.serialize_struct("Error", 3)?;
673        o.serialize_field("code", self.code())?;
674        o.serialize_field("message", self.message())?;
675        o.serialize_field("retryable", &self.is_retryable())?;
676        o.end()
677    }
678}
679
680// `DataflowError`, not `Error`: a TypeScript `Error` would shadow the
681// global one wherever it is imported.
682#[cfg(feature = "ts")]
683crate::base::ts::ts_decl!(
684    Error,
685    "DataflowError",
686    [
687        " An error as JSON: its code, its message, and whether a retry may",
688        " succeed.",
689    ],
690    |_cfg| {
691        use crate::base::ts::{Prop, object};
692        object(&[
693            Prop::new("code", "string", &[" The recorded code."]),
694            Prop::new("message", "string", &[" The human-readable message."]),
695            Prop::new("retryable", "boolean", &[" May a retry succeed?"]),
696        ])
697    }
698);
699
700#[cfg(test)]
701mod tests {
702    use super::*;
703
704    #[test]
705    fn record_text_prefixes_two_kinds_and_is_the_message_otherwise() {
706        // The record texts are 3.x's, byte for byte.
707        let m = "the message";
708        for (kind, expected) in [
709            (ErrorKind::Eval, "Logic evaluation error: the message"),
710            (ErrorKind::TimedOut, "Timeout error: the message"),
711            (ErrorKind::Budget, m),
712            (ErrorKind::Service { code: "X".into() }, m),
713            (ErrorKind::Config, m),
714            (ErrorKind::Compile, m),
715            (ErrorKind::Cancelled, m),
716        ] {
717            assert_eq!(
718                Error::new(kind.clone(), m).record_text(),
719                expected,
720                "{kind:?}"
721            );
722        }
723    }
724
725    fn kinds() -> Vec<(ErrorKind, &'static str)> {
726        vec![
727            (ErrorKind::Eval, "LOGIC_ERROR"),
728            (ErrorKind::Budget, "BUDGET_EXCEEDED"),
729            (ErrorKind::TimedOut, "TIMEOUT_ERROR"),
730            (ErrorKind::Cancelled, "CANCELLED"),
731            (ErrorKind::Config, "CONFIG_ERROR"),
732            (ErrorKind::Compile, "COMPILATION_ERROR"),
733            (
734                ErrorKind::Service {
735                    code: "X_FAIL".into(),
736                },
737                "X_FAIL",
738            ),
739            (ErrorKind::Service { code: "".into() }, "TASK_ERROR"),
740        ]
741    }
742
743    #[test]
744    fn code_gives_each_row_of_the_table() {
745        for (kind, code) in kinds() {
746            assert_eq!(Error::new(kind, "m").code(), code);
747        }
748    }
749
750    #[test]
751    fn only_a_timeout_is_retryable_by_default() {
752        for (kind, _) in kinds() {
753            let timed_out = kind == ErrorKind::TimedOut;
754            assert_eq!(Error::new(kind, "m").is_retryable(), timed_out);
755        }
756        assert!(Error::service("A", "m").with_retryable(true).is_retryable());
757        assert!(
758            !Error::new(ErrorKind::TimedOut, "m")
759                .with_retryable(false)
760                .is_retryable()
761        );
762    }
763
764    #[test]
765    fn wrap_keeps_kind_retryable_and_source() {
766        let source = std::io::Error::other("disk");
767        let error = Error::new(ErrorKind::Eval, "bad")
768            .with_retryable(true)
769            .with_source(source)
770            .wrap("task t");
771        assert_eq!(error.to_string(), "task t: bad");
772        assert_eq!(error.kind(), &ErrorKind::Eval);
773        assert!(error.is_retryable());
774        let source = std::error::Error::source(&error).expect("source kept");
775        assert_eq!(source.to_string(), "disk");
776    }
777
778    #[test]
779    fn debug_shows_kind_message_and_retryable() {
780        let text = format!("{:?}", Error::new(ErrorKind::Budget, "over"));
781        assert!(text.contains("Budget") && text.contains("over") && text.contains("retryable"));
782    }
783
784    #[test]
785    fn a_stored_record_with_3x_retry_keys_reads_and_writes_without_them() {
786        let stored =
787            r#"{"code":"X","message":"m","path":null,"retry_attempted":true,"retry_count":2}"#;
788        let record: ErrorRecord = serde_json::from_str(stored).unwrap();
789        assert_eq!(
790            serde_json::to_string(&record).unwrap(),
791            r#"{"code":"X","message":"m","path":null}"#
792        );
793    }
794
795    #[test]
796    fn stamp_fills_only_what_the_record_leaves_unset() {
797        let mut record = ErrorRecord::new("X", "m");
798        record.task_id = Some(Arc::from("own"));
799        record.stamp(
800            &Arc::from("w"),
801            Some(&Arc::from("t")),
802            &Arc::from("now"),
803            Some(3),
804        );
805        assert_eq!(
806            serde_json::to_string(&record).unwrap(),
807            r#"{"code":"X","message":"m","path":null,"workflow_id":"w","task_id":"own","timestamp":"now","element_index":3}"#
808        );
809        let mut workflow_own = ErrorRecord::new("X", "m");
810        workflow_own.stamp(&Arc::from("w"), None, &Arc::from("now"), None);
811        assert_eq!(workflow_own.task_id, None);
812    }
813
814    #[test]
815    fn a_record_from_an_error_carries_only_code_and_message() {
816        let record = ErrorRecord::from(&Error::service("X", "boom"));
817        assert_eq!(
818            serde_json::to_string(&record).unwrap(),
819            r#"{"code":"X","message":"boom","path":null}"#
820        );
821    }
822}
dataflow_rs/base/error.rs

dataflow_rs/base/
error.rs
