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 rows_unaccounted_message_reads_as_one_sentence() {
248        let msg = FaucetError::RowsUnaccounted {
249            offered: 10,
250            accepted: 9,
251        }
252        .to_string();
253        assert!(!msg.contains("  "), "{msg}");
254        assert!(msg.contains("accepted 9 of 10 offered rows and reported no per-row failure"));
255    }
256
257    #[test]
258    fn sink_status_types_only_retryable_statuses() {
259        for status in [429, 500, 503] {
260            let e = FaucetError::sink_status(Some(status), "s3://b/k", "m".into());
261            assert!(
262                matches!(e, FaucetError::HttpStatus { status: s, ref url, ref body }
263                    if s == status && url == "s3://b/k" && body == "m")
264            );
265            assert!(e.is_retriable());
266        }
267        for status in [None, Some(403), Some(404)] {
268            let e = FaucetError::sink_status(status, "u", "m".into());
269            assert!(matches!(e, FaucetError::Sink(ref m) if m == "m"), "{e:?}");
270        }
271    }
272
273    #[test]
274    fn http_status_5xx_is_retriable() {
275        let err = FaucetError::HttpStatus {
276            status: 500,
277            url: "https://example.com".into(),
278            body: "Internal Server Error".into(),
279        };
280        assert!(err.is_retriable());
281
282        let err = FaucetError::HttpStatus {
283            status: 503,
284            url: "https://example.com".into(),
285            body: "".into(),
286        };
287        assert!(err.is_retriable());
288    }
289
290    #[test]
291    fn http_status_4xx_is_not_retriable() {
292        let err = FaucetError::HttpStatus {
293            status: 400,
294            url: "https://example.com".into(),
295            body: "Bad Request".into(),
296        };
297        assert!(!err.is_retriable());
298
299        let err = FaucetError::HttpStatus {
300            status: 404,
301            url: "https://example.com".into(),
302            body: "".into(),
303        };
304        assert!(!err.is_retriable());
305    }
306
307    #[test]
308    fn http_status_429_is_retriable() {
309        // H3 (audit #146): a 429 surfaced as a plain HttpStatus (XML/GraphQL
310        // sources) must be retriable, not aborted on the first hit.
311        let err = FaucetError::HttpStatus {
312            status: 429,
313            url: "https://example.com".into(),
314            body: "Too Many Requests".into(),
315        };
316        assert!(err.is_retriable());
317    }
318
319    #[test]
320    fn cancelled_is_not_retriable() {
321        let err = FaucetError::Cancelled {
322            operation: "page call".into(),
323        };
324        assert!(!err.is_retriable());
325        assert_eq!(err.to_string(), "page call cancelled");
326    }
327
328    #[test]
329    fn timeout_is_retriable_and_names_the_operation() {
330        let err = FaucetError::Timeout {
331            operation: "nats publish".into(),
332            after: Duration::from_secs(30),
333        };
334        assert!(err.is_retriable());
335        assert_eq!(err.to_string(), "nats publish timed out after 30s");
336    }
337
338    #[test]
339    fn rate_limited_is_retriable() {
340        let err = FaucetError::RateLimited(Duration::from_secs(30));
341        assert!(err.is_retriable());
342    }
343
344    #[test]
345    fn json_error_is_not_retriable() {
346        let serde_err = serde_json::from_str::<serde_json::Value>("not json").unwrap_err();
347        let err = FaucetError::Json(serde_err);
348        assert!(!err.is_retriable());
349    }
350
351    #[test]
352    fn jsonpath_error_is_not_retriable() {
353        let err = FaucetError::JsonPath("bad path".into());
354        assert!(!err.is_retriable());
355    }
356
357    #[test]
358    fn auth_error_is_not_retriable() {
359        let err = FaucetError::Auth("invalid token".into());
360        assert!(!err.is_retriable());
361    }
362
363    #[test]
364    fn url_error_is_not_retriable() {
365        let err = FaucetError::Url("bad url".into());
366        assert!(!err.is_retriable());
367    }
368
369    #[test]
370    fn transform_error_is_not_retriable() {
371        let err = FaucetError::Transform("bad regex".into());
372        assert!(!err.is_retriable());
373    }
374
375    #[test]
376    fn http_status_display_includes_url_and_body() {
377        let err = FaucetError::HttpStatus {
378            status: 422,
379            url: "https://api.example.com/test".into(),
380            body: "Unprocessable Entity".into(),
381        };
382        let msg = err.to_string();
383        assert!(msg.contains("422"));
384        assert!(msg.contains("https://api.example.com/test"));
385        assert!(msg.contains("Unprocessable Entity"));
386    }
387
388    #[test]
389    fn config_error_is_not_retriable() {
390        let err = FaucetError::Config("bad endpoint".into());
391        assert!(!err.is_retriable());
392    }
393
394    #[test]
395    fn config_error_display() {
396        let err = FaucetError::Config("missing descriptor".into());
397        assert_eq!(err.to_string(), "Config error: missing descriptor");
398    }
399
400    #[test]
401    fn source_error_is_not_retriable() {
402        let err = FaucetError::Source("query failed".into());
403        assert!(!err.is_retriable());
404    }
405
406    #[test]
407    fn source_error_display() {
408        let err = FaucetError::Source("connection refused".into());
409        assert_eq!(err.to_string(), "Source error: connection refused");
410    }
411
412    #[test]
413    fn custom_error_is_not_retriable() {
414        let err = FaucetError::Custom(Box::new(std::io::Error::other("custom failure")));
415        assert!(!err.is_retriable());
416    }
417
418    #[test]
419    fn custom_error_display() {
420        let err = FaucetError::Custom(Box::new(std::io::Error::other("custom failure")));
421        assert_eq!(err.to_string(), "Connector error: custom failure");
422    }
423
424    #[test]
425    fn custom_error_from_boxed() {
426        let io_err = std::io::Error::other("file missing");
427        let boxed: Box<dyn std::error::Error + Send + Sync> = Box::new(io_err);
428        let err: FaucetError = boxed.into();
429        assert!(matches!(err, FaucetError::Custom(_)));
430    }
431
432    #[test]
433    fn sink_error_is_not_retriable() {
434        let err = FaucetError::Sink("BigQuery insert failed".into());
435        assert!(!err.is_retriable());
436    }
437
438    #[test]
439    fn sink_error_display() {
440        let err = FaucetError::Sink("connection refused".into());
441        assert_eq!(err.to_string(), "Sink error: connection refused");
442    }
443
444    #[test]
445    fn quality_failure_is_not_retriable_and_displays() {
446        let err = FaucetError::QualityFailure {
447            check: "not_null".into(),
448            message: "field 'user_id' was null".into(),
449        };
450        assert!(!err.is_retriable());
451        let s = err.to_string();
452        assert!(s.contains("not_null"));
453        assert!(s.contains("user_id"));
454    }
455
456    #[test]
457    fn schema_drift_is_not_retriable_and_displays() {
458        let err = FaucetError::SchemaDrift {
459            columns: vec!["email".into(), "score".into()],
460            message: "2 new columns".into(),
461        };
462        assert!(!err.is_retriable());
463        let s = err.to_string();
464        assert!(s.contains("email"));
465        assert!(s.contains("2 new columns"));
466    }
467
468    #[test]
469    fn circuit_open_is_not_retriable_and_displays() {
470        let err = FaucetError::CircuitOpen {
471            failures: 5,
472            cooldown: std::time::Duration::from_secs(60),
473        };
474        assert!(!err.is_retriable());
475        let s = err.to_string();
476        assert!(s.contains("5"));
477        assert!(s.contains("Circuit open"));
478    }
479}
faucet_core/error.rs

faucet_core/
error.rs
