dataflow_rs/engine/
error.rs

1use chrono::Utc;
2use serde::{Deserialize, Serialize};
3use thiserror::Error;
4
5/// Main error type for the dataflow engine
6///
7/// `#[non_exhaustive]`: a downstream `match` needs a wildcard arm. Adding it here
8/// makes every *future* variant additive rather than a breaking change, which
9/// matters because this enum is the crate's error channel and will keep growing.
10#[derive(Debug, Error, Clone, Serialize, Deserialize)]
11#[non_exhaustive]
12pub enum DataflowError {
13    /// Validation errors occurring during rule evaluation
14    #[error("Validation error: {0}")]
15    Validation(String),
16
17    /// Errors during function execution
18    #[error("Function execution error: {context}")]
19    FunctionExecution {
20        context: String,
21        #[source]
22        #[serde(skip)]
23        source: Option<Box<DataflowError>>,
24    },
25
26    /// Workflow-related errors
27    #[error("Workflow error: {0}")]
28    Workflow(String),
29
30    /// Task-related errors
31    #[error("Task error: {0}")]
32    Task(String),
33
34    /// Function not found errors
35    #[error("Function not found: {0}")]
36    FunctionNotFound(String),
37
38    /// JSON serialization/deserialization errors
39    #[error("Deserialization error: {0}")]
40    Deserialization(String),
41
42    /// I/O errors (file reading, etc.)
43    #[error("IO error: {0}")]
44    Io(String),
45
46    /// JSONLogic/DataLogic evaluation errors
47    #[error("Logic evaluation error: {0}")]
48    LogicEvaluation(String),
49
50    /// HTTP request errors
51    #[error("HTTP error: {status} - {message}")]
52    Http { status: u16, message: String },
53
54    /// Timeout errors
55    #[error("Timeout error: {0}")]
56    Timeout(String),
57
58    /// Any other errors
59    #[error("Unknown error: {0}")]
60    Unknown(String),
61
62    /// A failure the service layer classifies itself.
63    ///
64    /// The engine never interprets `kind`: it carries it to [`ErrorInfo::code`]
65    /// and otherwise treats this exactly like any other error —
66    /// `continue_on_error`, the audit-trail entry and the `Result::Err`
67    /// short-circuit are unchanged. No built-in returns this variant.
68    ///
69    /// `Display` renders `message` alone, so `to_string()` is always safe to hand
70    /// to an untrusted caller. `detail` is reachable only through `Debug`,
71    /// [`DataflowError::detail`] and [`ErrorInfo::detail`].
72    ///
73    /// Build it with [`DataflowError::service`].
74    #[error("{message}")]
75    Service {
76        /// Stable, service-owned classification, e.g. `"circuit_open"`.
77        kind: String,
78        /// Caller-safe text.
79        message: String,
80        /// Operator-only text: logged and kept on the trace, not intended for an
81        /// untrusted caller.
82        #[serde(default, skip_serializing_if = "Option::is_none")]
83        detail: Option<String>,
84        /// Retryability, declared by the service rather than inferred from the
85        /// variant.
86        retryable: bool,
87    },
88}
89
90impl DataflowError {
91    /// Creates a new function execution error with context
92    pub fn function_execution<S: Into<String>>(context: S, source: Option<DataflowError>) -> Self {
93        DataflowError::FunctionExecution {
94            context: context.into(),
95            source: source.map(Box::new),
96        }
97    }
98
99    /// Creates a new HTTP error
100    pub fn http<S: Into<String>>(status: u16, message: S) -> Self {
101        DataflowError::Http {
102            status,
103            message: message.into(),
104        }
105    }
106
107    /// Convert from std::io::Error
108    pub fn from_io(err: std::io::Error) -> Self {
109        DataflowError::Io(err.to_string())
110    }
111
112    /// Convert from serde_json::Error
113    pub fn from_serde(err: serde_json::Error) -> Self {
114        DataflowError::Deserialization(err.to_string())
115    }
116
117    /// Determines if this error is retryable (worth retrying)
118    ///
119    /// Retryable errors are typically transient infrastructure failures that might succeed on retry.
120    /// Non-retryable errors are typically data validation, logic, or configuration errors that
121    /// will consistently fail on retry.
122    pub fn retryable(&self) -> bool {
123        match self {
124            // Retryable errors - infrastructure/transient failures
125            DataflowError::Http { status, .. } => {
126                // Retry on server errors (5xx) and specific client errors that might be transient
127                *status >= 500 || *status == 429 || *status == 408 || *status == 0
128                // 0 means connection error
129            }
130            DataflowError::Timeout(_) => true,
131            DataflowError::Io(_) => true,
132            DataflowError::FunctionExecution { source, .. } => {
133                // Inherit retryability from the source error if present
134                source.as_ref().map(|e| e.retryable()).unwrap_or(false)
135            }
136
137            // Non-retryable errors - data/logic/configuration issues
138            DataflowError::Validation(_) => false,
139            DataflowError::LogicEvaluation(_) => false,
140            DataflowError::Deserialization(_) => false,
141            DataflowError::Workflow(_) => false,
142            DataflowError::Task(_) => false,
143            DataflowError::FunctionNotFound(_) => false,
144            DataflowError::Unknown(_) => false,
145
146            // Declared by the service rather than inferred from the variant.
147            DataflowError::Service { retryable, .. } => *retryable,
148        }
149    }
150
151    /// The service-owned classification, or `None` for every engine-owned
152    /// variant.
153    ///
154    /// This is the whole point of [`DataflowError::Service`]: a handler can
155    /// classify its own failures with a stable code the engine never interprets.
156    pub fn kind(&self) -> Option<&str> {
157        match self {
158            DataflowError::Service { kind, .. } => Some(kind),
159            // Same inheritance as `retryable()`: a `Service` error wrapped for
160            // context via `FunctionExecution` must not lose its classification.
161            DataflowError::FunctionExecution { source, .. } => {
162                source.as_deref().and_then(DataflowError::kind)
163            }
164            _ => None,
165        }
166    }
167
168    /// Operator-only detail, if this is a [`DataflowError::Service`] carrying one.
169    ///
170    /// Never included in `Display`, so `to_string()` stays safe to hand to an
171    /// untrusted caller. Reachable through `Debug`, this method, and
172    /// [`ErrorInfo::detail`].
173    pub fn detail(&self) -> Option<&str> {
174        match self {
175            DataflowError::Service { detail, .. } => detail.as_deref(),
176            DataflowError::FunctionExecution { source, .. } => {
177                source.as_deref().and_then(DataflowError::detail)
178            }
179            _ => None,
180        }
181    }
182
183    /// Start building a service-classified error.
184    ///
185    /// `kind` is the stable code the service will switch on; `message` is the
186    /// caller-safe text. Defaults: no detail, `retryable: false`.
187    ///
188    /// ```
189    /// use dataflow_rs::DataflowError;
190    ///
191    /// let e = DataflowError::service("circuit_open", "upstream unavailable")
192    ///     .detail("connector 'billing' breaker open")
193    ///     .retryable(true)
194    ///     .build();
195    ///
196    /// assert_eq!(e.kind(), Some("circuit_open"));
197    /// assert_eq!(e.detail(), Some("connector 'billing' breaker open"));
198    /// assert!(e.retryable());
199    /// // Display carries only the caller-safe text.
200    /// assert_eq!(e.to_string(), "upstream unavailable");
201    /// ```
202    pub fn service(kind: impl Into<String>, message: impl Into<String>) -> ServiceErrorBuilder {
203        ServiceErrorBuilder {
204            kind: kind.into(),
205            message: message.into(),
206            detail: None,
207            retryable: false,
208        }
209    }
210}
211
212/// Builder for [`DataflowError::Service`]. Mirrors [`ErrorInfoBuilder`].
213#[must_use = "ServiceErrorBuilder must be `.build()` to produce a DataflowError"]
214pub struct ServiceErrorBuilder {
215    kind: String,
216    message: String,
217    detail: Option<String>,
218    retryable: bool,
219}
220
221impl ServiceErrorBuilder {
222    /// Attach operator-only detail — logged and kept on the trace, not intended
223    /// for an untrusted caller.
224    pub fn detail(mut self, detail: impl Into<String>) -> Self {
225        self.detail = Some(detail.into());
226        self
227    }
228
229    /// Declare retryability. Defaults to `false`.
230    ///
231    /// Carriage for consumers: no engine code path acts on `retryable()`.
232    pub fn retryable(mut self, retryable: bool) -> Self {
233        self.retryable = retryable;
234        self
235    }
236
237    pub fn build(self) -> DataflowError {
238        DataflowError::Service {
239            kind: self.kind,
240            message: self.message,
241            detail: self.detail,
242            retryable: self.retryable,
243        }
244    }
245}
246
247/// Type alias for Result with DataflowError
248pub type Result<T> = std::result::Result<T, DataflowError>;
249
250/// Structured error information for error tracking in messages
251///
252/// `#[non_exhaustive]`: construct through [`ErrorInfo::builder`],
253/// [`ErrorInfo::new`], [`ErrorInfo::simple`] or [`ErrorInfo::simple_ref`], which
254/// are the documented paths. Field reads and `..` patterns are unaffected, and
255/// future field additions stay non-breaking.
256#[derive(Debug, Clone, Serialize, Deserialize)]
257#[non_exhaustive]
258pub struct ErrorInfo {
259    /// Error code (e.g., "WORKFLOW_ERROR", "TASK_ERROR", "VALIDATION_ERROR")
260    pub code: String,
261
262    /// Human-readable error message
263    pub message: String,
264
265    /// Optional path to the error location (e.g., "workflow.id", "task.id", "data.field")
266    pub path: Option<String>,
267
268    /// ID of the workflow where the error occurred (if available)
269    #[serde(skip_serializing_if = "Option::is_none")]
270    pub workflow_id: Option<String>,
271
272    /// ID of the task where the error occurred (if available)
273    #[serde(skip_serializing_if = "Option::is_none")]
274    pub task_id: Option<String>,
275
276    /// Timestamp when the error occurred
277    #[serde(skip_serializing_if = "Option::is_none")]
278    pub timestamp: Option<String>,
279
280    /// Whether a retry was attempted
281    #[serde(skip_serializing_if = "Option::is_none")]
282    pub retry_attempted: Option<bool>,
283
284    /// Number of retries attempted
285    #[serde(skip_serializing_if = "Option::is_none")]
286    pub retry_count: Option<u32>,
287
288    /// Operator-only detail lifted from [`DataflowError::Service`].
289    ///
290    /// Present on the persisted trace; a service must **not** surface it to an
291    /// untrusted caller. `None` for every engine-owned error, and omitted from
292    /// the serialized form when absent, so the JSON shape is unchanged unless a
293    /// `Service` error carried one.
294    #[serde(default, skip_serializing_if = "Option::is_none")]
295    pub detail: Option<String>,
296}
297
298/// Code recorded for a [`DataflowError::Service`] — also the historical
299/// `TASK_ERROR` fallback for every other variant. Shared with
300/// `workflow_executor`'s per-task error recording, which needs the identical
301/// "Service contributes its own kind verbatim; everything else stays
302/// `TASK_ERROR`" rule.
303///
304/// `kind` is passed through **verbatim** rather than upper-cased: the service owns
305/// it and will switch on the recorded `code`, so making the two different strings
306/// would force every consumer to know the transform. An empty `kind` falls back to
307/// the historical `TASK_ERROR` so no `ErrorInfo` ever carries an empty code.
308pub(crate) fn service_error_code(error: &DataflowError) -> String {
309    match error.kind() {
310        Some(kind) if !kind.is_empty() => kind.to_string(),
311        _ => "TASK_ERROR".to_string(),
312    }
313}
314
315impl ErrorInfo {
316    /// Create a new error info entry with all fields
317    pub fn new(workflow_id: Option<String>, task_id: Option<String>, error: DataflowError) -> Self {
318        Self {
319            code: match &error {
320                DataflowError::Validation(_) => "VALIDATION_ERROR".to_string(),
321                DataflowError::Workflow(_) => "WORKFLOW_ERROR".to_string(),
322                DataflowError::Task(_) => "TASK_ERROR".to_string(),
323                DataflowError::FunctionNotFound(_) => "FUNCTION_NOT_FOUND".to_string(),
324                DataflowError::FunctionExecution { .. } => "FUNCTION_ERROR".to_string(),
325                DataflowError::LogicEvaluation(_) => "LOGIC_ERROR".to_string(),
326                DataflowError::Http { .. } => "HTTP_ERROR".to_string(),
327                DataflowError::Timeout(_) => "TIMEOUT_ERROR".to_string(),
328                DataflowError::Io(_) => "IO_ERROR".to_string(),
329                DataflowError::Deserialization(_) => "DESERIALIZATION_ERROR".to_string(),
330                DataflowError::Unknown(_) => "UNKNOWN_ERROR".to_string(),
331                // Same "empty falls back to TASK_ERROR" rule as
332                // `service_error_code`, inlined rather than routed back through
333                // it — `error` is already known to be `Service` here, so a
334                // second match on `error.kind()` would just re-derive `kind`.
335                DataflowError::Service { kind, .. } => {
336                    if kind.is_empty() {
337                        "TASK_ERROR".to_string()
338                    } else {
339                        kind.clone()
340                    }
341                }
342            },
343            detail: error.detail().map(str::to_string),
344            message: error.to_string(),
345            path: None,
346            workflow_id,
347            task_id,
348            timestamp: Some(Utc::now().to_rfc3339()),
349            retry_attempted: Some(false),
350            retry_count: Some(0),
351        }
352    }
353
354    /// Create a simple error info with just code, message, and optional path
355    pub fn simple(code: String, message: String, path: Option<String>) -> Self {
356        Self {
357            code,
358            message,
359            path,
360            workflow_id: None,
361            task_id: None,
362            timestamp: Some(Utc::now().to_rfc3339()),
363            retry_attempted: None,
364            retry_count: None,
365            detail: None,
366        }
367    }
368
369    /// Create a simple error info from references (avoids cloning when possible)
370    pub fn simple_ref(code: &str, message: &str, path: Option<&str>) -> Self {
371        Self {
372            code: code.to_string(),
373            message: message.to_string(),
374            path: path.map(|s| s.to_string()),
375            workflow_id: None,
376            task_id: None,
377            timestamp: Some(Utc::now().to_rfc3339()),
378            retry_attempted: None,
379            retry_count: None,
380            detail: None,
381        }
382    }
383
384    /// Mark that a retry was attempted
385    pub fn with_retry(mut self) -> Self {
386        self.retry_attempted = Some(true);
387        self.retry_count = Some(self.retry_count.unwrap_or(0) + 1);
388        self
389    }
390
391    /// Create a builder for ErrorInfo
392    pub fn builder(code: impl Into<String>, message: impl Into<String>) -> ErrorInfoBuilder {
393        ErrorInfoBuilder::new(code, message)
394    }
395}
396
397/// Builder for creating ErrorInfo instances with a fluent API
398#[must_use = "ErrorInfoBuilder must be `.build()` to produce an ErrorInfo"]
399pub struct ErrorInfoBuilder {
400    code: String,
401    message: String,
402    path: Option<String>,
403    workflow_id: Option<String>,
404    task_id: Option<String>,
405    timestamp: Option<String>,
406    retry_attempted: Option<bool>,
407    retry_count: Option<u32>,
408    detail: Option<String>,
409}
410
411impl ErrorInfoBuilder {
412    /// Create a new ErrorInfoBuilder with required fields
413    pub fn new(code: impl Into<String>, message: impl Into<String>) -> Self {
414        Self {
415            code: code.into(),
416            message: message.into(),
417            path: None,
418            workflow_id: None,
419            task_id: None,
420            timestamp: Some(Utc::now().to_rfc3339()),
421            retry_attempted: None,
422            retry_count: None,
423            detail: None,
424        }
425    }
426
427    /// Set the error path
428    pub fn path(mut self, path: impl Into<String>) -> Self {
429        self.path = Some(path.into());
430        self
431    }
432
433    /// Set the workflow ID
434    pub fn workflow_id(mut self, id: impl Into<String>) -> Self {
435        self.workflow_id = Some(id.into());
436        self
437    }
438
439    /// Set the task ID
440    pub fn task_id(mut self, id: impl Into<String>) -> Self {
441        self.task_id = Some(id.into());
442        self
443    }
444
445    /// Set custom timestamp (defaults to now if not set)
446    pub fn timestamp(mut self, timestamp: impl Into<String>) -> Self {
447        self.timestamp = Some(timestamp.into());
448        self
449    }
450
451    /// Mark as retry attempted
452    pub fn retry_attempted(mut self, attempted: bool) -> Self {
453        self.retry_attempted = Some(attempted);
454        self
455    }
456
457    /// Set retry count
458    pub fn retry_count(mut self, count: u32) -> Self {
459        self.retry_count = Some(count);
460        self
461    }
462
463    /// Attach operator-only detail. Not surfaced by `Display` anywhere; a service
464    /// must not pass it to an untrusted caller.
465    pub fn detail(mut self, detail: impl Into<String>) -> Self {
466        self.detail = Some(detail.into());
467        self
468    }
469
470    /// Build the ErrorInfo instance
471    pub fn build(self) -> ErrorInfo {
472        ErrorInfo {
473            code: self.code,
474            message: self.message,
475            path: self.path,
476            workflow_id: self.workflow_id,
477            task_id: self.task_id,
478            timestamp: self.timestamp,
479            retry_attempted: self.retry_attempted,
480            retry_count: self.retry_count,
481            detail: self.detail,
482        }
483    }
484}
485
486#[cfg(test)]
487mod tests {
488    use super::*;
489
490    #[test]
491    fn test_retryable_errors() {
492        // Test retryable errors
493        assert!(
494            DataflowError::Http {
495                status: 500,
496                message: "Internal Server Error".to_string()
497            }
498            .retryable()
499        );
500        assert!(
501            DataflowError::Http {
502                status: 502,
503                message: "Bad Gateway".to_string()
504            }
505            .retryable()
506        );
507        assert!(
508            DataflowError::Http {
509                status: 503,
510                message: "Service Unavailable".to_string()
511            }
512            .retryable()
513        );
514        assert!(
515            DataflowError::Http {
516                status: 429,
517                message: "Too Many Requests".to_string()
518            }
519            .retryable()
520        );
521        assert!(
522            DataflowError::Http {
523                status: 408,
524                message: "Request Timeout".to_string()
525            }
526            .retryable()
527        );
528        assert!(
529            DataflowError::Http {
530                status: 0,
531                message: "Connection Error".to_string()
532            }
533            .retryable()
534        );
535        assert!(DataflowError::Timeout("Connection timeout".to_string()).retryable());
536        assert!(DataflowError::Io("Network error".to_string()).retryable());
537    }
538
539    #[test]
540    fn test_non_retryable_errors() {
541        // Test non-retryable errors
542        assert!(
543            !DataflowError::Http {
544                status: 400,
545                message: "Bad Request".to_string()
546            }
547            .retryable()
548        );
549        assert!(
550            !DataflowError::Http {
551                status: 401,
552                message: "Unauthorized".to_string()
553            }
554            .retryable()
555        );
556        assert!(
557            !DataflowError::Http {
558                status: 403,
559                message: "Forbidden".to_string()
560            }
561            .retryable()
562        );
563        assert!(
564            !DataflowError::Http {
565                status: 404,
566                message: "Not Found".to_string()
567            }
568            .retryable()
569        );
570        assert!(!DataflowError::Validation("Invalid input".to_string()).retryable());
571        assert!(!DataflowError::LogicEvaluation("Invalid logic".to_string()).retryable());
572        assert!(!DataflowError::Deserialization("Invalid JSON".to_string()).retryable());
573        assert!(!DataflowError::Workflow("Invalid workflow".to_string()).retryable());
574        assert!(!DataflowError::Unknown("Unknown error".to_string()).retryable());
575    }
576
577    #[test]
578    fn test_function_execution_error_retryability() {
579        // Test that function execution errors inherit retryability from source
580        let retryable_source = DataflowError::Http {
581            status: 500,
582            message: "Server Error".to_string(),
583        };
584        let non_retryable_source = DataflowError::Validation("Invalid data".to_string());
585
586        let retryable_func_error =
587            DataflowError::function_execution("HTTP call failed", Some(retryable_source));
588        let non_retryable_func_error =
589            DataflowError::function_execution("Validation failed", Some(non_retryable_source));
590        let no_source_func_error = DataflowError::function_execution("Unknown failure", None);
591
592        assert!(retryable_func_error.retryable());
593        assert!(!non_retryable_func_error.retryable());
594        assert!(!no_source_func_error.retryable());
595    }
596
597    #[test]
598    fn test_error_info_builder() {
599        // Test basic builder
600        let error = ErrorInfo::builder("TEST_ERROR", "Test message").build();
601        assert_eq!(error.code, "TEST_ERROR");
602        assert_eq!(error.message, "Test message");
603        assert!(error.timestamp.is_some());
604        assert!(error.path.is_none());
605
606        // Test full builder
607        let error = ErrorInfo::builder("VALIDATION_ERROR", "Field validation failed")
608            .path("data.email")
609            .workflow_id("workflow_1")
610            .task_id("validate_email")
611            .retry_attempted(true)
612            .retry_count(2)
613            .build();
614
615        assert_eq!(error.code, "VALIDATION_ERROR");
616        assert_eq!(error.message, "Field validation failed");
617        assert_eq!(error.path, Some("data.email".to_string()));
618        assert_eq!(error.workflow_id, Some("workflow_1".to_string()));
619        assert_eq!(error.task_id, Some("validate_email".to_string()));
620        assert_eq!(error.retry_attempted, Some(true));
621        assert_eq!(error.retry_count, Some(2));
622    }
623
624    #[test]
625    fn test_error_info_new_from_dataflow_error() {
626        // Test ErrorInfo::new generates correct codes for each error type
627        let test_cases = vec![
628            (
629                DataflowError::Validation("test".to_string()),
630                "VALIDATION_ERROR",
631            ),
632            (
633                DataflowError::Workflow("test".to_string()),
634                "WORKFLOW_ERROR",
635            ),
636            (DataflowError::Task("test".to_string()), "TASK_ERROR"),
637            (
638                DataflowError::FunctionNotFound("test".to_string()),
639                "FUNCTION_NOT_FOUND",
640            ),
641            (
642                DataflowError::function_execution("test", None),
643                "FUNCTION_ERROR",
644            ),
645            (
646                DataflowError::LogicEvaluation("test".to_string()),
647                "LOGIC_ERROR",
648            ),
649            (DataflowError::http(404, "Not Found"), "HTTP_ERROR"),
650            (DataflowError::Timeout("test".to_string()), "TIMEOUT_ERROR"),
651            (DataflowError::Io("test".to_string()), "IO_ERROR"),
652            (
653                DataflowError::Deserialization("test".to_string()),
654                "DESERIALIZATION_ERROR",
655            ),
656            (DataflowError::Unknown("test".to_string()), "UNKNOWN_ERROR"),
657        ];
658
659        for (error, expected_code) in test_cases {
660            let info = ErrorInfo::new(
661                Some("workflow_1".to_string()),
662                Some("task_1".to_string()),
663                error,
664            );
665            assert_eq!(info.code, expected_code);
666            assert_eq!(info.workflow_id, Some("workflow_1".to_string()));
667            assert_eq!(info.task_id, Some("task_1".to_string()));
668            assert!(info.timestamp.is_some());
669            assert_eq!(info.retry_attempted, Some(false));
670            assert_eq!(info.retry_count, Some(0));
671        }
672    }
673
674    #[test]
675    fn test_error_info_simple_constructors() {
676        // Test simple constructor
677        let error = ErrorInfo::simple(
678            "CUSTOM_ERROR".to_string(),
679            "Custom message".to_string(),
680            Some("data.field".to_string()),
681        );
682        assert_eq!(error.code, "CUSTOM_ERROR");
683        assert_eq!(error.message, "Custom message");
684        assert_eq!(error.path, Some("data.field".to_string()));
685        assert!(error.workflow_id.is_none());
686        assert!(error.task_id.is_none());
687        assert!(error.timestamp.is_some());
688
689        // Test simple_ref constructor
690        let error = ErrorInfo::simple_ref("REF_ERROR", "Ref message", Some("data.path"));
691        assert_eq!(error.code, "REF_ERROR");
692        assert_eq!(error.message, "Ref message");
693        assert_eq!(error.path, Some("data.path".to_string()));
694
695        // Test simple_ref with None path
696        let error = ErrorInfo::simple_ref("NO_PATH", "No path message", None);
697        assert!(error.path.is_none());
698    }
699
700    #[test]
701    fn test_error_info_with_retry() {
702        let error = ErrorInfo::simple_ref("TEST", "Test", None);
703        assert!(error.retry_attempted.is_none());
704        assert!(error.retry_count.is_none());
705
706        let error = error.with_retry();
707        assert_eq!(error.retry_attempted, Some(true));
708        assert_eq!(error.retry_count, Some(1));
709
710        let error = error.with_retry();
711        assert_eq!(error.retry_attempted, Some(true));
712        assert_eq!(error.retry_count, Some(2));
713    }
714
715    #[test]
716    fn test_error_display_messages() {
717        // Test that error display messages are correct
718        assert_eq!(
719            DataflowError::Validation("test".to_string()).to_string(),
720            "Validation error: test"
721        );
722        assert_eq!(
723            DataflowError::Workflow("test".to_string()).to_string(),
724            "Workflow error: test"
725        );
726        assert_eq!(
727            DataflowError::Task("test".to_string()).to_string(),
728            "Task error: test"
729        );
730        assert_eq!(
731            DataflowError::FunctionNotFound("test".to_string()).to_string(),
732            "Function not found: test"
733        );
734        assert_eq!(
735            DataflowError::http(404, "Not Found").to_string(),
736            "HTTP error: 404 - Not Found"
737        );
738        assert_eq!(
739            DataflowError::Timeout("test".to_string()).to_string(),
740            "Timeout error: test"
741        );
742    }
743
744    #[test]
745    fn test_error_conversions() {
746        // Test from_serde (we can't easily create a real serde error, but we can test the conversion works)
747        let json_str = "invalid json";
748        let serde_result: std::result::Result<serde_json::Value, _> =
749            serde_json::from_str(json_str);
750        if let Err(e) = serde_result {
751            let dataflow_err = DataflowError::from_serde(e);
752            assert!(matches!(dataflow_err, DataflowError::Deserialization(_)));
753        }
754    }
755
756    #[test]
757    fn service_error_carries_kind_detail_and_declared_retryability() {
758        let e = DataflowError::service("circuit_open", "upstream unavailable")
759            .detail("connector 'billing' breaker open")
760            .retryable(true)
761            .build();
762
763        assert_eq!(e.kind(), Some("circuit_open"));
764        assert_eq!(e.detail(), Some("connector 'billing' breaker open"));
765        assert!(e.retryable());
766    }
767
768    #[test]
769    fn function_execution_inherits_kind_detail_and_retryable_from_a_wrapped_service_source() {
770        // `DataflowError::function_execution` is the documented way to add
771        // context to a downstream error. `retryable()` already recursed into
772        // `source`; `kind()`/`detail()` must do the same, or wrapping a
773        // `Service` error for context silently downgrades it to a generic
774        // `TASK_ERROR` with no detail.
775        let inner = DataflowError::service("circuit_open", "upstream unavailable")
776            .detail("connector 'billing' breaker open since 12:04")
777            .retryable(true)
778            .build();
779        let wrapped = DataflowError::function_execution("calling billing connector", Some(inner));
780
781        assert_eq!(wrapped.kind(), Some("circuit_open"));
782        assert_eq!(
783            wrapped.detail(),
784            Some("connector 'billing' breaker open since 12:04")
785        );
786        assert!(wrapped.retryable());
787
788        // `ErrorInfo::new`'s Service arm goes through the same `kind()` call,
789        // so the wrapped error's code makes it through end to end too.
790        assert_eq!(service_error_code(&wrapped), "circuit_open");
791    }
792
793    #[test]
794    fn service_display_hides_the_detail_but_debug_shows_it() {
795        let e = DataflowError::service("circuit_open", "upstream unavailable")
796            .detail("SECRET-TOPOLOGY")
797            .build();
798
799        // `to_string()` is always safe to hand to an untrusted caller.
800        assert_eq!(e.to_string(), "upstream unavailable");
801        assert!(!e.to_string().contains("SECRET-TOPOLOGY"));
802        // The operator channel is reachable through Debug.
803        assert!(format!("{e:?}").contains("SECRET-TOPOLOGY"));
804    }
805
806    #[test]
807    fn service_retryability_is_independent_of_every_other_field() {
808        let yes = DataflowError::service("k", "m").retryable(true).build();
809        let no = DataflowError::service("k", "m").retryable(false).build();
810        assert!(yes.retryable());
811        assert!(!no.retryable());
812        // Defaults to false.
813        assert!(!DataflowError::service("k", "m").build().retryable());
814    }
815
816    #[test]
817    fn kind_and_detail_are_none_for_every_engine_owned_variant() {
818        let variants = [
819            DataflowError::Validation("v".into()),
820            DataflowError::FunctionExecution {
821                context: "c".into(),
822                source: None,
823            },
824            DataflowError::LogicEvaluation("l".into()),
825            DataflowError::Deserialization("d".into()),
826            DataflowError::Workflow("w".into()),
827            DataflowError::Task("t".into()),
828            DataflowError::FunctionNotFound("f".into()),
829            DataflowError::Http {
830                status: 500,
831                message: "h".into(),
832            },
833            DataflowError::Timeout("to".into()),
834            DataflowError::Io("io".into()),
835            DataflowError::Unknown("u".into()),
836        ];
837        assert_eq!(variants.len(), 11, "one assertion per engine-owned variant");
838        for v in &variants {
839            assert_eq!(v.kind(), None, "kind() for {v:?}");
840            assert_eq!(v.detail(), None, "detail() for {v:?}");
841        }
842    }
843
844    #[test]
845    fn service_error_code_passes_kind_through_verbatim() {
846        // The recorded decision: verbatim, not upper-cased. A service switching
847        // on the recorded `code` should not have to know a transform.
848        let e = DataflowError::service("circuit_open", "m").build();
849        let info = ErrorInfo::new(None, None, e);
850        assert_eq!(info.code, "circuit_open");
851    }
852
853    #[test]
854    fn an_empty_kind_falls_back_rather_than_recording_an_empty_code() {
855        let e = DataflowError::service("", "m").build();
856        let info = ErrorInfo::new(None, None, e);
857        assert_eq!(info.code, "TASK_ERROR");
858        assert!(!info.code.is_empty());
859    }
860
861    #[test]
862    fn a_non_ascii_kind_is_neither_panicked_on_nor_mangled() {
863        let e = DataflowError::service("limite_dépassé", "m").build();
864        let info = ErrorInfo::new(None, None, e);
865        // Verbatim, so the exact string round-trips including the accent.
866        assert_eq!(info.code, "limite_dépassé");
867    }
868
869    #[test]
870    fn error_info_lifts_the_detail_and_omits_it_when_absent() {
871        let with = ErrorInfo::new(
872            None,
873            None,
874            DataflowError::service("k", "m").detail("op only").build(),
875        );
876        assert_eq!(with.detail.as_deref(), Some("op only"));
877        assert!(serde_json::to_string(&with).unwrap().contains("detail"));
878
879        // Absent on a Service without one, and on every engine-owned variant.
880        let without = ErrorInfo::new(None, None, DataflowError::service("k", "m").build());
881        assert_eq!(without.detail, None);
882        assert!(!serde_json::to_string(&without).unwrap().contains("detail"));
883
884        let engine_owned = ErrorInfo::new(None, None, DataflowError::Task("t".into()));
885        assert_eq!(engine_owned.detail, None);
886        assert!(
887            !serde_json::to_string(&engine_owned)
888                .unwrap()
889                .contains("detail"),
890            "the JSON shape is unchanged for every pre-existing error"
891        );
892    }
893
894    #[test]
895    fn a_service_error_round_trips_through_serde_with_and_without_detail() {
896        for e in [
897            DataflowError::service("k", "m")
898                .detail("d")
899                .retryable(true)
900                .build(),
901            DataflowError::service("k", "m").build(),
902        ] {
903            let json = serde_json::to_string(&e).unwrap();
904            let back: DataflowError = serde_json::from_str(&json).unwrap();
905            assert_eq!(back.kind(), e.kind());
906            assert_eq!(back.detail(), e.detail());
907            assert_eq!(back.retryable(), e.retryable());
908        }
909
910        // `detail: None` emits no key.
911        let bare = DataflowError::service("k", "m").build();
912        assert!(!serde_json::to_string(&bare).unwrap().contains("detail"));
913    }
914}
dataflow_rs/engine/error.rs

dataflow_rs/engine/
error.rs
