llm_optimizer_processor/
error.rs

1//! Error types for the stream processor
2//!
3//! This module provides comprehensive error handling for all processor operations
4//! including windowing, aggregation, state management, and watermark generation.
5
6use thiserror::Error;
7
8/// Main processor error type
9#[derive(Error, Debug)]
10pub enum ProcessorError {
11    /// Window-related errors
12    #[error("window error: {0}")]
13    Window(#[from] WindowError),
14
15    /// Aggregation-related errors
16    #[error("aggregation error: {0}")]
17    Aggregation(#[from] AggregationError),
18
19    /// State backend errors
20    #[error("state error: {0}")]
21    State(#[from] StateError),
22
23    /// Watermark generation errors
24    #[error("watermark error: {0}")]
25    Watermark(#[from] WatermarkError),
26
27    /// Configuration errors
28    #[error("configuration error: {source}")]
29    Configuration {
30        source: Box<dyn std::error::Error + Send + Sync>,
31    },
32
33    /// Execution errors
34    #[error("execution error: {source}")]
35    Execution {
36        source: Box<dyn std::error::Error + Send + Sync>,
37    },
38
39    /// Serialization/deserialization errors
40    #[error("serialization error: {0}")]
41    Serialization(String),
42
43    /// I/O errors
44    #[error("I/O error: {0}")]
45    Io(#[from] std::io::Error),
46
47    /// Kafka-related errors
48    #[error("kafka error: {source}")]
49    Kafka {
50        source: Box<dyn std::error::Error + Send + Sync>,
51    },
52
53    /// Generic error for unexpected conditions
54    #[error("unexpected error: {0}")]
55    Unexpected(String),
56}
57
58/// Window assignment and management errors
59#[derive(Error, Debug)]
60pub enum WindowError {
61    /// Event timestamp is out of valid range
62    #[error("invalid event timestamp: {timestamp}, reason: {reason}")]
63    InvalidTimestamp { timestamp: i64, reason: String },
64
65    /// Window size is invalid
66    #[error("invalid window size: {size}ms, must be greater than 0")]
67    InvalidWindowSize { size: u64 },
68
69    /// Slide size is invalid for sliding windows
70    #[error("invalid slide size: {slide}ms, must be greater than 0 and less than or equal to window size {window}ms")]
71    InvalidSlideSize { slide: u64, window: u64 },
72
73    /// Gap size is invalid for session windows
74    #[error("invalid gap size: {gap}ms, must be greater than 0")]
75    InvalidGapSize { gap: u64 },
76
77    /// Event arrived too late after watermark
78    #[error("late event: event timestamp {event_time} is before watermark {watermark}, late by {late_by}ms")]
79    LateEvent {
80        event_time: i64,
81        watermark: i64,
82        late_by: i64,
83    },
84
85    /// Window has already been finalized
86    #[error("window already closed: window_id={window_id}, close_time={close_time}")]
87    WindowClosed { window_id: String, close_time: i64 },
88
89    /// Cannot assign event to window
90    #[error("window assignment failed: {reason}")]
91    AssignmentFailed { reason: String },
92
93    /// Window merge operation failed
94    #[error("window merge failed: {reason}")]
95    MergeFailed { reason: String },
96}
97
98/// Aggregation computation errors
99#[derive(Error, Debug)]
100pub enum AggregationError {
101    /// Numeric overflow during aggregation
102    #[error("numeric overflow in {operation}: {details}")]
103    NumericOverflow { operation: String, details: String },
104
105    /// Division by zero in average or similar calculations
106    #[error("division by zero in {operation}")]
107    DivisionByZero { operation: String },
108
109    /// Invalid metric value (NaN, Inf, etc.)
110    #[error("invalid metric value: {value}, reason: {reason}")]
111    InvalidValue { value: f64, reason: String },
112
113    /// Percentile calculation failed
114    #[error("percentile calculation failed: percentile={percentile}, sample_count={sample_count}, reason: {reason}")]
115    PercentileFailed {
116        percentile: u8,
117        sample_count: usize,
118        reason: String,
119    },
120
121    /// Histogram bin configuration error
122    #[error("invalid histogram configuration: {reason}")]
123    HistogramConfig { reason: String },
124
125    /// Insufficient data for aggregation
126    #[error("insufficient data for {aggregation_type}: need at least {required} samples, got {actual}")]
127    InsufficientData {
128        aggregation_type: String,
129        required: usize,
130        actual: usize,
131    },
132
133    /// Aggregation state is corrupted
134    #[error("corrupted aggregation state for {aggregation_type}: {details}")]
135    CorruptedState {
136        aggregation_type: String,
137        details: String,
138    },
139
140    /// Type mismatch in aggregation
141    #[error("type mismatch: expected {expected}, got {actual}")]
142    TypeMismatch { expected: String, actual: String },
143}
144
145/// State backend operation errors
146#[derive(Error, Debug)]
147pub enum StateError {
148    /// State backend not initialized
149    #[error("state backend not initialized: {backend_type}")]
150    NotInitialized { backend_type: String },
151
152    /// State key not found
153    #[error("state key not found: {key}")]
154    KeyNotFound { key: String },
155
156    /// State serialization failed
157    #[error("state serialization failed for key '{key}': {reason}")]
158    SerializationFailed { key: String, reason: String },
159
160    /// State deserialization failed
161    #[error("state deserialization failed for key '{key}': {reason}")]
162    DeserializationFailed { key: String, reason: String },
163
164    /// State backend storage error
165    #[error("storage error in {backend_type}: {details}")]
166    StorageError {
167        backend_type: String,
168        details: String,
169    },
170
171    /// Checkpoint creation failed
172    #[error("checkpoint failed at {checkpoint_id}: {reason}")]
173    CheckpointFailed {
174        checkpoint_id: String,
175        reason: String,
176    },
177
178    /// Checkpoint restoration failed
179    #[error("restore failed from checkpoint {checkpoint_id}: {reason}")]
180    RestoreFailed {
181        checkpoint_id: String,
182        reason: String,
183    },
184
185    /// State TTL expired
186    #[error("state expired for key '{key}': ttl={ttl}ms, age={age}ms")]
187    StateExpired { key: String, ttl: u64, age: u64 },
188
189    /// Transaction failed
190    #[error("transaction failed: {operation}, reason: {reason}")]
191    TransactionFailed { operation: String, reason: String },
192
193    /// Concurrent modification detected
194    #[error("concurrent modification detected for key '{key}'")]
195    ConcurrentModification { key: String },
196
197    /// State backend is full
198    #[error("state backend full: used {used} bytes of {capacity} bytes")]
199    BackendFull { used: u64, capacity: u64 },
200}
201
202/// Watermark generation and propagation errors
203#[derive(Error, Debug)]
204pub enum WatermarkError {
205    /// Watermark went backwards
206    #[error("watermark regression: new watermark {new_watermark} is before current {current_watermark}")]
207    WatermarkRegression {
208        current_watermark: i64,
209        new_watermark: i64,
210    },
211
212    /// Source is idle for too long
213    #[error("source idle timeout: source_id={source_id}, idle_duration={idle_duration}ms, timeout={timeout}ms")]
214    SourceIdle {
215        source_id: String,
216        idle_duration: u64,
217        timeout: u64,
218    },
219
220    /// Invalid watermark configuration
221    #[error("invalid watermark configuration: {reason}")]
222    InvalidConfig { reason: String },
223
224    /// Watermark delay is negative
225    #[error("invalid watermark delay: {delay}ms, must be non-negative")]
226    InvalidDelay { delay: i64 },
227
228    /// Missing timestamp in event
229    #[error("event missing timestamp: event_id={event_id}")]
230    MissingTimestamp { event_id: String },
231
232    /// Clock skew detected
233    #[error("clock skew detected: event time {event_time} is in the future, ahead by {ahead_by}ms")]
234    ClockSkew { event_time: i64, ahead_by: i64 },
235
236    /// Watermark alignment failed across partitions
237    #[error("watermark alignment failed: {reason}")]
238    AlignmentFailed { reason: String },
239}
240
241/// Result type alias for processor operations
242pub type Result<T> = std::result::Result<T, ProcessorError>;
243
244/// Result type alias for window operations
245pub type WindowResult<T> = std::result::Result<T, WindowError>;
246
247/// Result type alias for aggregation operations
248pub type AggregationResult<T> = std::result::Result<T, AggregationError>;
249
250/// Result type alias for state operations
251pub type StateResult<T> = std::result::Result<T, StateError>;
252
253/// Result type alias for watermark operations
254pub type WatermarkResult<T> = std::result::Result<T, WatermarkError>;
255
256impl From<bincode::Error> for ProcessorError {
257    fn from(err: bincode::Error) -> Self {
258        ProcessorError::Serialization(err.to_string())
259    }
260}
261
262impl From<serde_json::Error> for ProcessorError {
263    fn from(err: serde_json::Error) -> Self {
264        ProcessorError::Serialization(err.to_string())
265    }
266}
267
268impl From<anyhow::Error> for ProcessorError {
269    fn from(err: anyhow::Error) -> Self {
270        ProcessorError::Unexpected(err.to_string())
271    }
272}
273
274#[cfg(test)]
275mod tests {
276    use super::*;
277
278    #[test]
279    fn test_window_error_display() {
280        let err = WindowError::InvalidWindowSize { size: 0 };
281        assert!(err.to_string().contains("invalid window size"));
282    }
283
284    #[test]
285    fn test_aggregation_error_display() {
286        let err = AggregationError::DivisionByZero {
287            operation: "average".to_string(),
288        };
289        assert!(err.to_string().contains("division by zero"));
290    }
291
292    #[test]
293    fn test_state_error_display() {
294        let err = StateError::KeyNotFound {
295            key: "test_key".to_string(),
296        };
297        assert!(err.to_string().contains("not found"));
298    }
299
300    #[test]
301    fn test_watermark_error_display() {
302        let err = WatermarkError::WatermarkRegression {
303            current_watermark: 1000,
304            new_watermark: 900,
305        };
306        assert!(err.to_string().contains("regression"));
307    }
308
309    #[test]
310    fn test_processor_error_from_window_error() {
311        let window_err = WindowError::InvalidWindowSize { size: 0 };
312        let processor_err: ProcessorError = window_err.into();
313        assert!(matches!(processor_err, ProcessorError::Window(_)));
314    }
315}
llm_optimizer_processor/error.rs

llm_optimizer_processor/
error.rs
