allsource_core/
error.rs

1/// AllSource error types
2#[derive(Debug, thiserror::Error)]
3pub enum AllSourceError {
4    #[error("Event not found: {0}")]
5    EventNotFound(String),
6
7    #[error("Entity not found: {0}")]
8    EntityNotFound(String),
9
10    #[error("Tenant already exists: {0}")]
11    TenantAlreadyExists(String),
12
13    #[error("Tenant not found: {0}")]
14    TenantNotFound(String),
15
16    #[error("Invalid event: {0}")]
17    InvalidEvent(String),
18
19    #[error("Invalid query: {0}")]
20    InvalidQuery(String),
21
22    #[error("Invalid input: {0}")]
23    InvalidInput(String),
24
25    #[error("Storage error: {0}")]
26    StorageError(String),
27
28    #[error("Serialization error: {0}")]
29    SerializationError(#[from] serde_json::Error),
30
31    #[error("Arrow error: {0}")]
32    ArrowError(String),
33
34    #[error("Index error: {0}")]
35    IndexError(String),
36
37    #[error("Validation error: {0}")]
38    ValidationError(String),
39
40    #[error("Concurrency error: {0}")]
41    ConcurrencyError(String),
42
43    #[error("Version conflict: expected {expected}, current {current}")]
44    VersionConflict { expected: u64, current: u64 },
45
46    #[error("Queue full: {0}")]
47    QueueFull(String),
48
49    #[error("Internal error: {0}")]
50    InternalError(String),
51
52    /// Event payload failed validation against a registered schema. The
53    /// HTTP layer maps this to 422 with a structured body so callers can
54    /// distinguish schema problems from generic validation failures.
55    #[error(
56        "Schema violation for event_type '{event_type}' (schema v{schema_version}): {errors:?}"
57    )]
58    SchemaViolation {
59        event_type: String,
60        schema_version: u32,
61        errors: Vec<String>,
62    },
63}
64
65impl AllSourceError {
66    /// Returns true for transient errors that may succeed on retry
67    /// (storage I/O, concurrency conflicts, queue pressure).
68    pub fn is_retryable(&self) -> bool {
69        matches!(
70            self,
71            AllSourceError::StorageError(_)
72                | AllSourceError::ConcurrencyError(_)
73                | AllSourceError::VersionConflict { .. }
74                | AllSourceError::QueueFull(_)
75        )
76    }
77}
78
79// Alias for domain layer convenience
80pub use AllSourceError as Error;
81
82impl From<arrow::error::ArrowError> for AllSourceError {
83    fn from(err: arrow::error::ArrowError) -> Self {
84        AllSourceError::ArrowError(err.to_string())
85    }
86}
87
88impl From<parquet::errors::ParquetError> for AllSourceError {
89    fn from(err: parquet::errors::ParquetError) -> Self {
90        AllSourceError::StorageError(err.to_string())
91    }
92}
93
94impl From<crate::infrastructure::persistence::SimdJsonError> for AllSourceError {
95    fn from(err: crate::infrastructure::persistence::SimdJsonError) -> Self {
96        AllSourceError::SerializationError(serde_json::Error::io(std::io::Error::new(
97            std::io::ErrorKind::InvalidData,
98            err.to_string(),
99        )))
100    }
101}
102
103#[cfg(feature = "postgres")]
104impl From<sqlx::Error> for AllSourceError {
105    fn from(err: sqlx::Error) -> Self {
106        AllSourceError::StorageError(format!("Database error: {err}"))
107    }
108}
109
110/// Custom Result type for AllSource operations
111pub type Result<T> = std::result::Result<T, AllSourceError>;
112
113#[cfg(feature = "server")]
114mod axum_impl {
115    use super::AllSourceError;
116    use axum::{
117        http::StatusCode,
118        response::{IntoResponse, Response},
119    };
120
121    /// Implement IntoResponse for axum error handling
122    impl IntoResponse for AllSourceError {
123        fn into_response(self) -> Response {
124            let (status, error_message) = match self {
125                AllSourceError::EventNotFound(_)
126                | AllSourceError::EntityNotFound(_)
127                | AllSourceError::TenantNotFound(_) => (StatusCode::NOT_FOUND, self.to_string()),
128                AllSourceError::InvalidEvent(_)
129                | AllSourceError::InvalidQuery(_)
130                | AllSourceError::InvalidInput(_)
131                | AllSourceError::ValidationError(_) => (StatusCode::BAD_REQUEST, self.to_string()),
132                AllSourceError::VersionConflict { expected, current } => {
133                    let body = serde_json::json!({
134                        "error": "version_conflict",
135                        "expected_version": expected,
136                        "current_version": current,
137                    });
138                    return (StatusCode::CONFLICT, axum::Json(body)).into_response();
139                }
140                AllSourceError::TenantAlreadyExists(_) | AllSourceError::ConcurrencyError(_) => {
141                    (StatusCode::CONFLICT, self.to_string())
142                }
143                AllSourceError::QueueFull(_) => (StatusCode::SERVICE_UNAVAILABLE, self.to_string()),
144                AllSourceError::StorageError(_)
145                | AllSourceError::ArrowError(_)
146                | AllSourceError::IndexError(_)
147                | AllSourceError::InternalError(_) => {
148                    (StatusCode::INTERNAL_SERVER_ERROR, self.to_string())
149                }
150                AllSourceError::SerializationError(_) => {
151                    (StatusCode::UNPROCESSABLE_ENTITY, self.to_string())
152                }
153                AllSourceError::SchemaViolation {
154                    event_type,
155                    schema_version,
156                    errors,
157                } => {
158                    let body = serde_json::json!({
159                        "error": {
160                            "code": "schema_violation",
161                            "message": format!(
162                                "Event payload does not match registered schema for '{event_type}'"
163                            ),
164                            "details": {
165                                "event_type": event_type,
166                                "schema_version": schema_version,
167                                "errors": errors,
168                            }
169                        }
170                    });
171                    return (StatusCode::UNPROCESSABLE_ENTITY, axum::Json(body)).into_response();
172                }
173            };
174
175            let body = serde_json::json!({
176                "error": error_message,
177            });
178
179            (status, axum::Json(body)).into_response()
180        }
181    }
182}
183
184#[cfg(test)]
185mod tests {
186    use super::*;
187
188    #[cfg(feature = "server")]
189    use axum::{http::StatusCode, response::IntoResponse};
190
191    #[test]
192    fn test_error_display() {
193        let err = AllSourceError::EventNotFound("event-123".to_string());
194        assert_eq!(err.to_string(), "Event not found: event-123");
195
196        let err = AllSourceError::EntityNotFound("entity-456".to_string());
197        assert_eq!(err.to_string(), "Entity not found: entity-456");
198
199        let err = AllSourceError::TenantAlreadyExists("tenant-1".to_string());
200        assert_eq!(err.to_string(), "Tenant already exists: tenant-1");
201
202        let err = AllSourceError::TenantNotFound("tenant-2".to_string());
203        assert_eq!(err.to_string(), "Tenant not found: tenant-2");
204    }
205
206    #[test]
207    fn test_error_variants() {
208        let errors: Vec<AllSourceError> = vec![
209            AllSourceError::InvalidEvent("bad event".to_string()),
210            AllSourceError::InvalidQuery("bad query".to_string()),
211            AllSourceError::InvalidInput("bad input".to_string()),
212            AllSourceError::StorageError("storage failed".to_string()),
213            AllSourceError::ArrowError("arrow failed".to_string()),
214            AllSourceError::IndexError("index failed".to_string()),
215            AllSourceError::ValidationError("validation failed".to_string()),
216            AllSourceError::ConcurrencyError("conflict".to_string()),
217            AllSourceError::QueueFull("queue full".to_string()),
218            AllSourceError::InternalError("internal error".to_string()),
219        ];
220
221        for err in errors {
222            let msg = err.to_string();
223            assert!(!msg.is_empty());
224        }
225    }
226
227    #[cfg(feature = "server")]
228    #[test]
229    fn test_into_response_not_found() {
230        let err = AllSourceError::EventNotFound("event-123".to_string());
231        let response = err.into_response();
232        assert_eq!(response.status(), StatusCode::NOT_FOUND);
233
234        let err = AllSourceError::EntityNotFound("entity-456".to_string());
235        let response = err.into_response();
236        assert_eq!(response.status(), StatusCode::NOT_FOUND);
237
238        let err = AllSourceError::TenantNotFound("tenant-1".to_string());
239        let response = err.into_response();
240        assert_eq!(response.status(), StatusCode::NOT_FOUND);
241    }
242
243    #[cfg(feature = "server")]
244    #[test]
245    fn test_into_response_bad_request() {
246        let err = AllSourceError::InvalidEvent("bad event".to_string());
247        let response = err.into_response();
248        assert_eq!(response.status(), StatusCode::BAD_REQUEST);
249
250        let err = AllSourceError::InvalidQuery("bad query".to_string());
251        let response = err.into_response();
252        assert_eq!(response.status(), StatusCode::BAD_REQUEST);
253
254        let err = AllSourceError::InvalidInput("bad input".to_string());
255        let response = err.into_response();
256        assert_eq!(response.status(), StatusCode::BAD_REQUEST);
257
258        let err = AllSourceError::ValidationError("validation failed".to_string());
259        let response = err.into_response();
260        assert_eq!(response.status(), StatusCode::BAD_REQUEST);
261    }
262
263    #[cfg(feature = "server")]
264    #[test]
265    fn test_into_response_conflict() {
266        let err = AllSourceError::TenantAlreadyExists("tenant-1".to_string());
267        let response = err.into_response();
268        assert_eq!(response.status(), StatusCode::CONFLICT);
269
270        let err = AllSourceError::ConcurrencyError("conflict".to_string());
271        let response = err.into_response();
272        assert_eq!(response.status(), StatusCode::CONFLICT);
273    }
274
275    #[cfg(feature = "server")]
276    #[test]
277    fn test_into_response_service_unavailable() {
278        let err = AllSourceError::QueueFull("queue is full".to_string());
279        let response = err.into_response();
280        assert_eq!(response.status(), StatusCode::SERVICE_UNAVAILABLE);
281    }
282
283    /// Locks the wire contract from neotoma-gaps bead t-0795 AC #3:
284    /// 422 with `{error: {code: "schema_violation", details: {event_type,
285    /// schema_version, errors}}}`. Callers parse this shape — don't break it.
286    #[cfg(feature = "server")]
287    #[tokio::test]
288    async fn test_into_response_schema_violation() {
289        use axum::body::to_bytes;
290        let err = AllSourceError::SchemaViolation {
291            event_type: "user.created".to_string(),
292            schema_version: 2,
293            errors: vec!["Missing required field: email".to_string()],
294        };
295        let response = err.into_response();
296        assert_eq!(response.status(), StatusCode::UNPROCESSABLE_ENTITY);
297
298        let body_bytes = to_bytes(response.into_body(), usize::MAX).await.unwrap();
299        let body: serde_json::Value = serde_json::from_slice(&body_bytes).unwrap();
300        assert_eq!(body["error"]["code"], "schema_violation");
301        assert_eq!(body["error"]["details"]["event_type"], "user.created");
302        assert_eq!(body["error"]["details"]["schema_version"], 2);
303        let errors = body["error"]["details"]["errors"].as_array().unwrap();
304        assert_eq!(errors.len(), 1);
305        assert!(errors[0].as_str().unwrap().contains("email"));
306    }
307
308    #[cfg(feature = "server")]
309    #[test]
310    fn test_into_response_internal_error() {
311        let err = AllSourceError::StorageError("storage error".to_string());
312        let response = err.into_response();
313        assert_eq!(response.status(), StatusCode::INTERNAL_SERVER_ERROR);
314
315        let err = AllSourceError::ArrowError("arrow error".to_string());
316        let response = err.into_response();
317        assert_eq!(response.status(), StatusCode::INTERNAL_SERVER_ERROR);
318
319        let err = AllSourceError::IndexError("index error".to_string());
320        let response = err.into_response();
321        assert_eq!(response.status(), StatusCode::INTERNAL_SERVER_ERROR);
322
323        let err = AllSourceError::InternalError("internal error".to_string());
324        let response = err.into_response();
325        assert_eq!(response.status(), StatusCode::INTERNAL_SERVER_ERROR);
326    }
327
328    #[test]
329    fn test_from_arrow_error() {
330        let arrow_err = arrow::error::ArrowError::InvalidArgumentError("test".to_string());
331        let err: AllSourceError = arrow_err.into();
332        assert!(matches!(err, AllSourceError::ArrowError(_)));
333    }
334
335    #[test]
336    fn test_from_parquet_error() {
337        let parquet_err = parquet::errors::ParquetError::General("test".to_string());
338        let err: AllSourceError = parquet_err.into();
339        assert!(matches!(err, AllSourceError::StorageError(_)));
340    }
341
342    #[test]
343    fn test_error_debug() {
344        let err = AllSourceError::EventNotFound("test".to_string());
345        let debug_str = format!("{err:?}");
346        assert!(debug_str.contains("EventNotFound"));
347    }
348
349    #[test]
350    fn test_is_retryable() {
351        // Retryable errors
352        assert!(AllSourceError::StorageError("io".to_string()).is_retryable());
353        assert!(AllSourceError::ConcurrencyError("conflict".to_string()).is_retryable());
354        assert!(AllSourceError::QueueFull("backpressure".to_string()).is_retryable());
355
356        // Non-retryable errors
357        assert!(!AllSourceError::EventNotFound("e1".to_string()).is_retryable());
358        assert!(!AllSourceError::InvalidEvent("bad".to_string()).is_retryable());
359        assert!(!AllSourceError::InvalidQuery("bad".to_string()).is_retryable());
360        assert!(!AllSourceError::ValidationError("bad".to_string()).is_retryable());
361        assert!(!AllSourceError::TenantNotFound("t1".to_string()).is_retryable());
362        assert!(!AllSourceError::TenantAlreadyExists("t1".to_string()).is_retryable());
363        assert!(!AllSourceError::InternalError("bug".to_string()).is_retryable());
364    }
365}
allsource_core/error.rs

allsource_core/
error.rs
