faucet_core/
error.rs

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

faucet_core/
error.rs
