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

allsource_core/
error.rs
