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#[derive(Debug, Error)]
8pub enum FaucetError {
9    #[error("HTTP error: {0}")]
10    Http(#[from] reqwest::Error),
11
12    /// An HTTP response with a non-success status code.
13    ///
14    /// Contains the status code, URL, and (truncated) response body for
15    /// debugging.  Whether this error is retriable depends on the status code
16    /// — see [`FaucetError::is_retriable`].
17    #[error("HTTP {status} from {url}: {body}")]
18    HttpStatus {
19        status: u16,
20        url: String,
21        body: String,
22    },
23
24    #[error("JSON error: {0}")]
25    Json(#[from] serde_json::Error),
26
27    #[error("JSONPath error: {0}")]
28    JsonPath(String),
29
30    #[error("Auth error: {0}")]
31    Auth(String),
32
33    /// The server responded with HTTP 429 Too Many Requests.
34    /// The inner value is the duration to wait before retrying,
35    /// parsed from the `Retry-After` response header (default: 60 s).
36    #[error("Rate limited: retry after {0:?}")]
37    RateLimited(Duration),
38
39    /// A URL could not be constructed or parsed.
40    #[error("URL error: {0}")]
41    Url(String),
42
43    /// A record transform could not be compiled or applied (e.g. invalid regex).
44    #[error("Transform error: {0}")]
45    Transform(String),
46
47    /// A configuration or validation error (e.g. invalid endpoint, missing descriptor).
48    #[error("Config error: {0}")]
49    Config(String),
50
51    /// A source operation failed (e.g. database query error, file read error).
52    #[error("Source error: {0}")]
53    Source(String),
54
55    /// A sink operation failed (e.g. BigQuery insert error).
56    #[error("Sink error: {0}")]
57    Sink(String),
58
59    /// A data-quality check failed under an `abort` policy.
60    #[error("Quality check '{check}' failed: {message}")]
61    QualityFailure { check: String, message: String },
62
63    /// An incoming page's shape diverged from the destination schema under an
64    /// `on_drift: fail` (or `on_incompatible: fail`) policy.
65    #[error("Schema drift on columns {columns:?}: {message}")]
66    SchemaDrift {
67        columns: Vec<String>,
68        message: String,
69    },
70
71    /// A record breached the pipeline's data contract under an
72    /// `on_breach: fail` policy. `version` is the contract version the
73    /// record was validated against.
74    #[error("Contract v{version} violated: {message}")]
75    ContractViolation { version: String, message: String },
76
77    /// A state-store operation failed (read/write/delete of a replication
78    /// bookmark, checkpoint, or other persisted pipeline state).
79    #[error("State error: {0}")]
80    State(String),
81
82    /// The resilience circuit breaker opened after repeated failures; the run
83    /// is aborted fast. `cooldown` is advisory for the orchestration layer
84    /// (e.g. `faucet schedule` delays re-entry by this duration).
85    #[error("Circuit open after {failures} consecutive failures; cooldown {cooldown:?}")]
86    CircuitOpen { failures: u32, cooldown: Duration },
87
88    /// A custom error from a third-party connector.
89    ///
90    /// Use this to wrap your own error types without losing the error chain:
91    /// ```rust
92    /// use faucet_core::FaucetError;
93    /// let err = FaucetError::Custom(Box::new(std::io::Error::new(
94    ///     std::io::ErrorKind::Other,
95    ///     "my connector failed",
96    /// )));
97    /// ```
98    #[error("Connector error: {0}")]
99    Custom(#[from] Box<dyn std::error::Error + Send + Sync>),
100}
101
102impl FaucetError {
103    /// Whether this error is transient and the request should be retried.
104    ///
105    /// Retriable errors:
106    /// - Network / connection errors (`Http` from reqwest)
107    /// - Server errors (5xx status codes)
108    /// - Rate limiting (429 — handled separately with `Retry-After`)
109    ///
110    /// Non-retriable errors:
111    /// - Client errors (4xx except 429)
112    /// - JSON parse / JSONPath / auth / transform errors
113    pub fn is_retriable(&self) -> bool {
114        match self {
115            // reqwest errors: connection timeouts, DNS failures, etc. are retriable.
116            FaucetError::Http(e) => {
117                // If it's a status error that leaked through, check the code.
118                if let Some(status) = e.status() {
119                    status.is_server_error()
120                } else {
121                    // Connection errors, timeouts, etc.
122                    true
123                }
124            }
125            // 5xx are retriable; 429 (Too Many Requests) is too — sources that
126            // surface a rate-limit as a plain HttpStatus rather than the
127            // dedicated RateLimited variant (XML, GraphQL) would otherwise abort
128            // on the first 429 (audit #146 H3).
129            FaucetError::HttpStatus { status, .. } => *status >= 500 || *status == 429,
130            FaucetError::RateLimited(_) => true,
131            _ => false,
132        }
133    }
134}
135
136#[cfg(test)]
137mod tests {
138    use super::*;
139
140    #[test]
141    fn http_status_5xx_is_retriable() {
142        let err = FaucetError::HttpStatus {
143            status: 500,
144            url: "https://example.com".into(),
145            body: "Internal Server Error".into(),
146        };
147        assert!(err.is_retriable());
148
149        let err = FaucetError::HttpStatus {
150            status: 503,
151            url: "https://example.com".into(),
152            body: "".into(),
153        };
154        assert!(err.is_retriable());
155    }
156
157    #[test]
158    fn http_status_4xx_is_not_retriable() {
159        let err = FaucetError::HttpStatus {
160            status: 400,
161            url: "https://example.com".into(),
162            body: "Bad Request".into(),
163        };
164        assert!(!err.is_retriable());
165
166        let err = FaucetError::HttpStatus {
167            status: 404,
168            url: "https://example.com".into(),
169            body: "".into(),
170        };
171        assert!(!err.is_retriable());
172    }
173
174    #[test]
175    fn http_status_429_is_retriable() {
176        // H3 (audit #146): a 429 surfaced as a plain HttpStatus (XML/GraphQL
177        // sources) must be retriable, not aborted on the first hit.
178        let err = FaucetError::HttpStatus {
179            status: 429,
180            url: "https://example.com".into(),
181            body: "Too Many Requests".into(),
182        };
183        assert!(err.is_retriable());
184    }
185
186    #[test]
187    fn rate_limited_is_retriable() {
188        let err = FaucetError::RateLimited(Duration::from_secs(30));
189        assert!(err.is_retriable());
190    }
191
192    #[test]
193    fn json_error_is_not_retriable() {
194        let serde_err = serde_json::from_str::<serde_json::Value>("not json").unwrap_err();
195        let err = FaucetError::Json(serde_err);
196        assert!(!err.is_retriable());
197    }
198
199    #[test]
200    fn jsonpath_error_is_not_retriable() {
201        let err = FaucetError::JsonPath("bad path".into());
202        assert!(!err.is_retriable());
203    }
204
205    #[test]
206    fn auth_error_is_not_retriable() {
207        let err = FaucetError::Auth("invalid token".into());
208        assert!(!err.is_retriable());
209    }
210
211    #[test]
212    fn url_error_is_not_retriable() {
213        let err = FaucetError::Url("bad url".into());
214        assert!(!err.is_retriable());
215    }
216
217    #[test]
218    fn transform_error_is_not_retriable() {
219        let err = FaucetError::Transform("bad regex".into());
220        assert!(!err.is_retriable());
221    }
222
223    #[test]
224    fn http_status_display_includes_url_and_body() {
225        let err = FaucetError::HttpStatus {
226            status: 422,
227            url: "https://api.example.com/test".into(),
228            body: "Unprocessable Entity".into(),
229        };
230        let msg = err.to_string();
231        assert!(msg.contains("422"));
232        assert!(msg.contains("https://api.example.com/test"));
233        assert!(msg.contains("Unprocessable Entity"));
234    }
235
236    #[test]
237    fn config_error_is_not_retriable() {
238        let err = FaucetError::Config("bad endpoint".into());
239        assert!(!err.is_retriable());
240    }
241
242    #[test]
243    fn config_error_display() {
244        let err = FaucetError::Config("missing descriptor".into());
245        assert_eq!(err.to_string(), "Config error: missing descriptor");
246    }
247
248    #[test]
249    fn source_error_is_not_retriable() {
250        let err = FaucetError::Source("query failed".into());
251        assert!(!err.is_retriable());
252    }
253
254    #[test]
255    fn source_error_display() {
256        let err = FaucetError::Source("connection refused".into());
257        assert_eq!(err.to_string(), "Source error: connection refused");
258    }
259
260    #[test]
261    fn custom_error_is_not_retriable() {
262        let err = FaucetError::Custom(Box::new(std::io::Error::other("custom failure")));
263        assert!(!err.is_retriable());
264    }
265
266    #[test]
267    fn custom_error_display() {
268        let err = FaucetError::Custom(Box::new(std::io::Error::other("custom failure")));
269        assert_eq!(err.to_string(), "Connector error: custom failure");
270    }
271
272    #[test]
273    fn custom_error_from_boxed() {
274        let io_err = std::io::Error::other("file missing");
275        let boxed: Box<dyn std::error::Error + Send + Sync> = Box::new(io_err);
276        let err: FaucetError = boxed.into();
277        assert!(matches!(err, FaucetError::Custom(_)));
278    }
279
280    #[test]
281    fn sink_error_is_not_retriable() {
282        let err = FaucetError::Sink("BigQuery insert failed".into());
283        assert!(!err.is_retriable());
284    }
285
286    #[test]
287    fn sink_error_display() {
288        let err = FaucetError::Sink("connection refused".into());
289        assert_eq!(err.to_string(), "Sink error: connection refused");
290    }
291
292    #[test]
293    fn quality_failure_is_not_retriable_and_displays() {
294        let err = FaucetError::QualityFailure {
295            check: "not_null".into(),
296            message: "field 'user_id' was null".into(),
297        };
298        assert!(!err.is_retriable());
299        let s = err.to_string();
300        assert!(s.contains("not_null"));
301        assert!(s.contains("user_id"));
302    }
303
304    #[test]
305    fn schema_drift_is_not_retriable_and_displays() {
306        let err = FaucetError::SchemaDrift {
307            columns: vec!["email".into(), "score".into()],
308            message: "2 new columns".into(),
309        };
310        assert!(!err.is_retriable());
311        let s = err.to_string();
312        assert!(s.contains("email"));
313        assert!(s.contains("2 new columns"));
314    }
315
316    #[test]
317    fn circuit_open_is_not_retriable_and_displays() {
318        let err = FaucetError::CircuitOpen {
319            failures: 5,
320            cooldown: std::time::Duration::from_secs(60),
321        };
322        assert!(!err.is_retriable());
323        let s = err.to_string();
324        assert!(s.contains("5"));
325        assert!(s.contains("Circuit open"));
326    }
327}
faucet_core/error.rs

faucet_core/
error.rs
