faucet_core/
error.rs

1//! Error types for faucet-stream.
2
3use std::time::Duration;
4use thiserror::Error;
5
6/// What a [`FaucetError::PolicyViolation`] was decided on (#1051).
7#[derive(Debug, Clone, Default, PartialEq, Eq)]
8pub struct ViolationFacts {
9    /// The label the rule matched on the column.
10    pub label: Option<String>,
11    /// The sink id the value was headed for (`dlq` for the dead-letter queue).
12    pub sink: Option<String>,
13    /// That sink's connector kind.
14    pub sink_kind: Option<String>,
15}
16
17/// All possible errors returned by faucet-stream.
18///
19/// `#[non_exhaustive]`: this enum demonstrably grows (`QualityFailure`,
20/// `CircuitOpen`, `SchemaDrift`, `ContractViolation` all arrived post-1.0) and
21/// it is the one type every third-party connector author touches. Marking it
22/// keeps the next variant a **minor** release instead of a forced major for
23/// `faucet-core` and everything downstream. Match with a `_` arm.
24#[derive(Debug, Error)]
25#[non_exhaustive]
26pub enum FaucetError {
27    #[error("HTTP error: {0}")]
28    Http(#[from] reqwest::Error),
29
30    /// An HTTP response with a non-success status code.
31    ///
32    /// Contains the status code, URL, and (truncated) response body for
33    /// debugging.  Whether this error is retriable depends on the status code
34    /// — see [`FaucetError::is_retriable`].
35    #[error("HTTP {status} from {url}: {body}")]
36    HttpStatus {
37        status: u16,
38        url: String,
39        body: String,
40    },
41
42    #[error("JSON error: {0}")]
43    Json(#[from] serde_json::Error),
44
45    #[error("JSONPath error: {0}")]
46    JsonPath(String),
47
48    #[error("Auth error: {0}")]
49    Auth(String),
50
51    /// The server responded with HTTP 429 Too Many Requests.
52    /// The inner value is the duration to wait before retrying,
53    /// parsed from the `Retry-After` response header (default: 60 s).
54    #[error("Rate limited: retry after {0:?}")]
55    RateLimited(Duration),
56
57    /// A URL could not be constructed or parsed.
58    #[error("URL error: {0}")]
59    Url(String),
60
61    /// A record transform could not be compiled or applied (e.g. invalid regex).
62    #[error("Transform error: {0}")]
63    Transform(String),
64
65    /// A configuration or validation error (e.g. invalid endpoint, missing descriptor).
66    #[error("Config error: {0}")]
67    Config(String),
68
69    /// A source operation failed (e.g. database query error, file read error).
70    #[error("Source error: {0}")]
71    Source(String),
72
73    /// A sink operation failed (e.g. BigQuery insert error).
74    #[error("Sink error: {0}")]
75    Sink(String),
76
77    /// A data-quality check failed under an `abort` policy.
78    #[error("Quality check '{check}' failed: {message}")]
79    QualityFailure { check: String, message: String },
80
81    /// An incoming page's shape diverged from the destination schema under an
82    /// `on_drift: fail` (or `on_incompatible: fail`) policy.
83    #[error("Schema drift on columns {columns:?}: {message}")]
84    SchemaDrift {
85        columns: Vec<String>,
86        message: String,
87    },
88
89    /// A data-flow policy rule was violated (#702): at run time a labelled
90    /// value headed for a sink the rule does not allow, or a static check
91    /// refused the run before it started.
92    #[error("Policy `{rule}` violated on column `{column}`: {message}")]
93    PolicyViolation {
94        rule: String,
95        column: String,
96        message: String,
97        /// The label, sink and sink kind the verdict was decided on, when
98        /// known (#1051).
99        facts: Option<Box<ViolationFacts>>,
100    },
101
102    /// A run budget ceiling was crossed (#703): `budget` names the ceiling
103    /// (`max_records` / `max_bytes` / `max_duration_secs`), `limit` its value
104    /// and `actual` what the run reached or would have reached. A records /
105    /// bytes crossing refuses the whole page **before** it is written, so
106    /// the bookmark never advances past it; a duration crossing cancels the
107    /// run cooperatively at the next page boundary.
108    #[error("Budget `{budget}` exceeded: limit {limit}, actual {actual}")]
109    BudgetExceeded {
110        budget: String,
111        limit: u64,
112        actual: u64,
113    },
114
115    /// A sink accounted for fewer rows of a page than it was offered (#879):
116    /// its write returned `accepted` for a page of `offered` rows, without
117    /// per-row outcomes or a coalescing report
118    /// ([`Sink::write_batch_reported`](crate::Sink::write_batch_reported))
119    /// saying where the rest went. The page is not checkpointed, so a resumed
120    /// run reads it again.
121    #[error(
122        "Rows unaccounted: the sink accepted {accepted} of {offered} offered rows and reported no per-row failure or coalescing for the rest"
123    )]
124    RowsUnaccounted { offered: u64, accepted: u64 },
125
126    /// A run's learned column profile drifted from its baseline under a
127    /// `profiling.on_drift: fail` policy (#708). Raised after the run's data
128    /// is written: the failure marks the run, it does not undo it.
129    #[error("Profile drift on columns {columns:?}: {message}")]
130    ProfileDrift {
131        columns: Vec<String>,
132        message: String,
133    },
134
135    /// A record breached the pipeline's data contract under an
136    /// `on_breach: fail` policy. `version` is the contract version the
137    /// record was validated against.
138    #[error("Contract v{version} violated: {message}")]
139    ContractViolation { version: String, message: String },
140
141    /// A state-store operation failed (read/write/delete of a replication
142    /// bookmark, checkpoint, or other persisted pipeline state).
143    #[error("State error: {0}")]
144    State(String),
145
146    /// A stored state value this release or this source cannot read (#736):
147    /// written by a newer faucet (a newer state format or bookmark schema) or
148    /// by a different source. Raised before anything is read from the source,
149    /// instead of guessing at the shape — which would silently re-sync or skip.
150    #[error(
151        "state '{key}' is incompatible: found {found}, expected {expected} — it was written by a \
152         newer faucet or by a different source; run the release that wrote it, or reset the row \
153         (`faucet state reset`) after confirming where it should resume"
154    )]
155    StateIncompatible {
156        /// The state key.
157        key: String,
158        /// What is stored.
159        found: String,
160        /// What this release and source read.
161        expected: String,
162    },
163
164    /// The resilience circuit breaker opened after repeated failures; the run
165    /// is aborted fast. `cooldown` is advisory for the orchestration layer
166    /// (e.g. `faucet schedule` delays re-entry by this duration).
167    #[error("Circuit open after {failures} consecutive failures; cooldown {cooldown:?}")]
168    CircuitOpen { failures: u32, cooldown: Duration },
169
170    /// An operation did not finish within its time bound (a publish waiting
171    /// for broker confirms, a flush). Its outcome is unknown, so it is
172    /// retriable: the resilience policy classifies it as
173    /// [`RetryClass::Timeout`](crate::resilience::RetryClass::Timeout).
174    #[error("{operation} timed out after {after:?}")]
175    Timeout {
176        /// What was being waited on, e.g. `"nats publish"`.
177        operation: String,
178        /// The bound that elapsed.
179        after: Duration,
180    },
181
182    /// A backend call was abandoned because the run was cancelled
183    /// ([`CallContext::call`](crate::io::CallContext::call)). Not retriable:
184    /// the caller is stopping.
185    #[error("{operation} cancelled")]
186    Cancelled {
187        /// What was being waited on, e.g. `"page call"`.
188        operation: String,
189    },
190
191    /// A custom error from a third-party connector.
192    ///
193    /// Use this to wrap your own error types without losing the error chain:
194    /// ```rust
195    /// use faucet_core::FaucetError;
196    /// let err = FaucetError::Custom(Box::new(std::io::Error::new(
197    ///     std::io::ErrorKind::Other,
198    ///     "my connector failed",
199    /// )));
200    /// ```
201    #[error("Connector error: {0}")]
202    Custom(#[from] Box<dyn std::error::Error + Send + Sync>),
203}
204
205impl FaucetError {
206    /// A storage client's failure, typed by its HTTP status when it has one:
207    /// a 429 or 5xx becomes [`HttpStatus`](Self::HttpStatus) (so the
208    /// resilience policy retries it) with `message` as the body; anything
209    /// else, including a failure with no response, is a
210    /// [`Sink`](Self::Sink) error carrying `message`.
211    pub fn sink_status(status: Option<u16>, url: impl Into<String>, message: String) -> Self {
212        match status {
213            Some(status) if status == 429 || status >= 500 => FaucetError::HttpStatus {
214                status,
215                url: url.into(),
216                body: message,
217            },
218            _ => FaucetError::Sink(message),
219        }
220    }
221
222    /// Whether this error is transient and the request should be retried.
223    ///
224    /// Retriable errors:
225    /// - Network / connection errors (`Http` from reqwest)
226    /// - Server errors (5xx status codes)
227    /// - Rate limiting (429 — handled separately with `Retry-After`)
228    ///
229    /// Non-retriable errors:
230    /// - Client errors (4xx except 429)
231    /// - JSON parse / JSONPath / auth / transform errors
232    pub fn is_retriable(&self) -> bool {
233        match self {
234            // reqwest errors: connection timeouts, DNS failures, etc. are retriable.
235            FaucetError::Http(e) => {
236                // If it's a status error that leaked through, check the code.
237                if let Some(status) = e.status() {
238                    status.is_server_error()
239                } else {
240                    // Connection errors, timeouts, etc.
241                    true
242                }
243            }
244            // 5xx are retriable; 429 (Too Many Requests) is too — sources that
245            // surface a rate-limit as a plain HttpStatus rather than the
246            // dedicated RateLimited variant (XML, GraphQL) would otherwise abort
247            // on the first 429 (audit #146 H3).
248            FaucetError::HttpStatus { status, .. } => *status >= 500 || *status == 429,
249            FaucetError::RateLimited(_) => true,
250            FaucetError::Timeout { .. } => true,
251            _ => false,
252        }
253    }
254}
255
256#[cfg(test)]
257mod tests {
258    use super::*;
259
260    #[test]
261    fn rows_unaccounted_message_reads_as_one_sentence() {
262        let msg = FaucetError::RowsUnaccounted {
263            offered: 10,
264            accepted: 9,
265        }
266        .to_string();
267        assert!(!msg.contains("  "), "{msg}");
268        assert!(msg.contains("accepted 9 of 10 offered rows and reported no per-row failure"));
269    }
270
271    #[test]
272    fn sink_status_types_only_retryable_statuses() {
273        for status in [429, 500, 503] {
274            let e = FaucetError::sink_status(Some(status), "s3://b/k", "m".into());
275            assert!(
276                matches!(e, FaucetError::HttpStatus { status: s, ref url, ref body }
277                    if s == status && url == "s3://b/k" && body == "m")
278            );
279            assert!(e.is_retriable());
280        }
281        for status in [None, Some(403), Some(404)] {
282            let e = FaucetError::sink_status(status, "u", "m".into());
283            assert!(matches!(e, FaucetError::Sink(ref m) if m == "m"), "{e:?}");
284        }
285    }
286
287    #[test]
288    fn http_status_5xx_is_retriable() {
289        let err = FaucetError::HttpStatus {
290            status: 500,
291            url: "https://example.com".into(),
292            body: "Internal Server Error".into(),
293        };
294        assert!(err.is_retriable());
295
296        let err = FaucetError::HttpStatus {
297            status: 503,
298            url: "https://example.com".into(),
299            body: "".into(),
300        };
301        assert!(err.is_retriable());
302    }
303
304    #[test]
305    fn http_status_4xx_is_not_retriable() {
306        let err = FaucetError::HttpStatus {
307            status: 400,
308            url: "https://example.com".into(),
309            body: "Bad Request".into(),
310        };
311        assert!(!err.is_retriable());
312
313        let err = FaucetError::HttpStatus {
314            status: 404,
315            url: "https://example.com".into(),
316            body: "".into(),
317        };
318        assert!(!err.is_retriable());
319    }
320
321    #[test]
322    fn http_status_429_is_retriable() {
323        // H3 (audit #146): a 429 surfaced as a plain HttpStatus (XML/GraphQL
324        // sources) must be retriable, not aborted on the first hit.
325        let err = FaucetError::HttpStatus {
326            status: 429,
327            url: "https://example.com".into(),
328            body: "Too Many Requests".into(),
329        };
330        assert!(err.is_retriable());
331    }
332
333    #[test]
334    fn cancelled_is_not_retriable() {
335        let err = FaucetError::Cancelled {
336            operation: "page call".into(),
337        };
338        assert!(!err.is_retriable());
339        assert_eq!(err.to_string(), "page call cancelled");
340    }
341
342    #[test]
343    fn timeout_is_retriable_and_names_the_operation() {
344        let err = FaucetError::Timeout {
345            operation: "nats publish".into(),
346            after: Duration::from_secs(30),
347        };
348        assert!(err.is_retriable());
349        assert_eq!(err.to_string(), "nats publish timed out after 30s");
350    }
351
352    #[test]
353    fn rate_limited_is_retriable() {
354        let err = FaucetError::RateLimited(Duration::from_secs(30));
355        assert!(err.is_retriable());
356    }
357
358    #[test]
359    fn json_error_is_not_retriable() {
360        let serde_err = serde_json::from_str::<serde_json::Value>("not json").unwrap_err();
361        let err = FaucetError::Json(serde_err);
362        assert!(!err.is_retriable());
363    }
364
365    #[test]
366    fn jsonpath_error_is_not_retriable() {
367        let err = FaucetError::JsonPath("bad path".into());
368        assert!(!err.is_retriable());
369    }
370
371    #[test]
372    fn auth_error_is_not_retriable() {
373        let err = FaucetError::Auth("invalid token".into());
374        assert!(!err.is_retriable());
375    }
376
377    #[test]
378    fn url_error_is_not_retriable() {
379        let err = FaucetError::Url("bad url".into());
380        assert!(!err.is_retriable());
381    }
382
383    #[test]
384    fn transform_error_is_not_retriable() {
385        let err = FaucetError::Transform("bad regex".into());
386        assert!(!err.is_retriable());
387    }
388
389    #[test]
390    fn http_status_display_includes_url_and_body() {
391        let err = FaucetError::HttpStatus {
392            status: 422,
393            url: "https://api.example.com/test".into(),
394            body: "Unprocessable Entity".into(),
395        };
396        let msg = err.to_string();
397        assert!(msg.contains("422"));
398        assert!(msg.contains("https://api.example.com/test"));
399        assert!(msg.contains("Unprocessable Entity"));
400    }
401
402    #[test]
403    fn config_error_is_not_retriable() {
404        let err = FaucetError::Config("bad endpoint".into());
405        assert!(!err.is_retriable());
406    }
407
408    #[test]
409    fn config_error_display() {
410        let err = FaucetError::Config("missing descriptor".into());
411        assert_eq!(err.to_string(), "Config error: missing descriptor");
412    }
413
414    #[test]
415    fn source_error_is_not_retriable() {
416        let err = FaucetError::Source("query failed".into());
417        assert!(!err.is_retriable());
418    }
419
420    #[test]
421    fn source_error_display() {
422        let err = FaucetError::Source("connection refused".into());
423        assert_eq!(err.to_string(), "Source error: connection refused");
424    }
425
426    #[test]
427    fn custom_error_is_not_retriable() {
428        let err = FaucetError::Custom(Box::new(std::io::Error::other("custom failure")));
429        assert!(!err.is_retriable());
430    }
431
432    #[test]
433    fn custom_error_display() {
434        let err = FaucetError::Custom(Box::new(std::io::Error::other("custom failure")));
435        assert_eq!(err.to_string(), "Connector error: custom failure");
436    }
437
438    #[test]
439    fn custom_error_from_boxed() {
440        let io_err = std::io::Error::other("file missing");
441        let boxed: Box<dyn std::error::Error + Send + Sync> = Box::new(io_err);
442        let err: FaucetError = boxed.into();
443        assert!(matches!(err, FaucetError::Custom(_)));
444    }
445
446    #[test]
447    fn sink_error_is_not_retriable() {
448        let err = FaucetError::Sink("BigQuery insert failed".into());
449        assert!(!err.is_retriable());
450    }
451
452    #[test]
453    fn sink_error_display() {
454        let err = FaucetError::Sink("connection refused".into());
455        assert_eq!(err.to_string(), "Sink error: connection refused");
456    }
457
458    #[test]
459    fn quality_failure_is_not_retriable_and_displays() {
460        let err = FaucetError::QualityFailure {
461            check: "not_null".into(),
462            message: "field 'user_id' was null".into(),
463        };
464        assert!(!err.is_retriable());
465        let s = err.to_string();
466        assert!(s.contains("not_null"));
467        assert!(s.contains("user_id"));
468    }
469
470    #[test]
471    fn schema_drift_is_not_retriable_and_displays() {
472        let err = FaucetError::SchemaDrift {
473            columns: vec!["email".into(), "score".into()],
474            message: "2 new columns".into(),
475        };
476        assert!(!err.is_retriable());
477        let s = err.to_string();
478        assert!(s.contains("email"));
479        assert!(s.contains("2 new columns"));
480    }
481
482    #[test]
483    fn circuit_open_is_not_retriable_and_displays() {
484        let err = FaucetError::CircuitOpen {
485            failures: 5,
486            cooldown: std::time::Duration::from_secs(60),
487        };
488        assert!(!err.is_retriable());
489        let s = err.to_string();
490        assert!(s.contains("5"));
491        assert!(s.contains("Circuit open"));
492    }
493}
faucet_core/error.rs

faucet_core/
error.rs
