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

faucet_core/
error.rs
