etl_core/
error.rs

1//! Error taxonomy and per-stage error policies.
2//!
3//! Three classes of failure exist in a pipeline (see `docs/DESIGN.md`):
4//! *retryable* (transient I/O — handled by the sink retry layer),
5//! *record-level* (a bad payload or failed user code — subject to
6//! [`ErrorPolicy`]), and *fatal* (invariant violations — the pipeline
7//! stops). Record-level policies are deliberately limited to `Skip` and
8//! `Fail`; every skip is surfaced through metrics.
9
10/// What to do when a record fails in a stage.
11#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
12#[non_exhaustive]
13pub enum ErrorPolicy {
14    /// Drop the record, count it in `etl_*_dropped_total{reason}`, and
15    /// continue. Default for deserializers.
16    Skip,
17    /// Fail the batch and stop the pipeline. Default for operators.
18    #[default]
19    Fail,
20}
21
22/// Broad classification used in metrics labels (`error_type`) and by the
23/// retry layer.
24#[derive(Clone, Copy, Debug, PartialEq, Eq)]
25#[non_exhaustive]
26pub enum ErrorClass {
27    /// Transient; retrying the same operation may succeed.
28    Retryable,
29    /// Specific to one record; retrying the same record cannot succeed.
30    RecordLevel,
31    /// The component or pipeline is broken; processing must stop.
32    Fatal,
33}
34
35/// An unrecoverable pipeline failure: an invariant was violated or a
36/// `Fail`-policy stage tripped. The pipeline transitions to `Failed`, the
37/// partition watermarks stop advancing, and the process exits non-zero.
38#[derive(Debug, thiserror::Error)]
39#[error("fatal error in {component}: {reason}")]
40pub struct FatalError {
41    /// Component id where the failure originated.
42    pub component: String,
43    /// Human-readable cause.
44    pub reason: String,
45}
46
47/// A payload could not be deserialized.
48#[derive(Debug, thiserror::Error)]
49#[non_exhaustive]
50pub enum DeserError {
51    /// The payload bytes do not match the expected format.
52    #[error("malformed payload: {reason}")]
53    Malformed {
54        /// Human-readable cause, for logs and dead-record metrics.
55        reason: String,
56    },
57    /// A schema required to decode the payload is not available.
58    #[error("schema unavailable: {reason}")]
59    SchemaUnavailable {
60        /// Human-readable cause.
61        reason: String,
62    },
63    /// The payload cannot be decoded *yet*: a required resource (typically
64    /// a schema fetched from a registry) is still being obtained, and the
65    /// deserializer has already triggered the asynchronous work that will
66    /// make it available. The chain reports the batch `Blocked` at this
67    /// payload and the driver's retry loop re-pushes it — the record is
68    /// neither dropped nor counted as an error, and the stage's
69    /// [`ErrorPolicy`] does not apply.
70    ///
71    /// Contract: a deserializer must return this **before emitting any
72    /// record** for the payload; records emitted ahead of a `NotReady`
73    /// would be duplicated when the payload is replayed.
74    #[error("not ready: {reason}")]
75    NotReady {
76        /// What is being waited for.
77        reason: String,
78    },
79}
80
81/// A source failed to poll, commit, or manage its assignment.
82#[derive(Debug, thiserror::Error)]
83#[non_exhaustive]
84pub enum SourceError {
85    /// Underlying client error.
86    #[error("source error ({class:?}): {reason}")]
87    Client {
88        /// Retryable vs fatal, as judged by the connector.
89        class: ErrorClass,
90        /// Human-readable cause.
91        reason: String,
92    },
93}
94
95/// A sink failed to write a batch.
96#[derive(Debug, thiserror::Error)]
97#[non_exhaustive]
98pub enum SinkError {
99    /// Underlying client error.
100    #[error("sink error ({class:?}): {reason}")]
101    Client {
102        /// Retryable (will be retried on another replica) vs fatal.
103        class: ErrorClass,
104        /// Human-readable cause.
105        reason: String,
106    },
107}
108
109#[cfg(test)]
110mod tests {
111    use super::*;
112
113    #[test]
114    fn defaults_match_documented_policy() {
115        assert_eq!(ErrorPolicy::default(), ErrorPolicy::Fail);
116    }
117
118    #[test]
119    fn errors_render_reasons() {
120        let e = DeserError::Malformed {
121            reason: "truncated header".into(),
122        };
123        assert!(e.to_string().contains("truncated header"));
124    }
125}
etl_core/error.rs

etl_core/
error.rs
