mq_bridge/
errors.rs

1use thiserror::Error;
2
3/// Errors that can occur during message processing (handling or publishing).
4#[derive(Error, Debug)]
5pub enum ProcessingError {
6    /// A transient error occurred. The operation should be retried.
7    #[error("retryable error: {0}")]
8    Retryable(#[source] anyhow::Error),
9    /// A permanent error occurred. The operation should not be retried.
10    #[error("non-retryable error: {0}")]
11    NonRetryable(#[source] anyhow::Error),
12    /// A connection-level error occurred. Used to signal broker disconnects, etc.
13    #[error("connection error: {0}")]
14    Connection(#[source] anyhow::Error),
15}
16impl ProcessingError {
17    pub fn is_connection_error(&self) -> bool {
18        matches!(self, ProcessingError::Connection(_))
19    }
20}
21
22pub type HandlerError = ProcessingError;
23pub type PublisherError = ProcessingError;
24
25/// Errors that can occur when consuming messages.
26#[derive(Error, Debug)]
27pub enum ConsumerError {
28    /// A transport-level or other error occurred that should trigger a reconnect.
29    #[error("consumer connection error: {0}")]
30    Connection(#[source] anyhow::Error),
31
32    /// A consumer gap was detected: the requested events were already garbage-collected.
33    #[error("consumer gap: requested offset {requested} but earliest available is {base}")]
34    Gap { requested: u64, base: u64 },
35
36    /// The consumer has reached the end of the stream and has shut down gracefully.
37    #[error("consumer reached end of stream")]
38    EndOfStream,
39
40    /// A permanent, non-retryable error: the message cannot be processed and must
41    /// not be re-read (e.g. failed AEAD decryption/authentication of a payload).
42    #[error("permanent consumer error: {0}")]
43    Permanent(#[source] anyhow::Error),
44}
45
46impl From<anyhow::Error> for ConsumerError {
47    fn from(err: anyhow::Error) -> Self {
48        // By default, we'll treat any generic error as a connection-level, retryable error.
49        ConsumerError::Connection(err)
50    }
51}
52
53impl From<anyhow::Error> for ProcessingError {
54    fn from(err: anyhow::Error) -> Self {
55        // Default to Retryable for generic errors. Callers should use
56        // ProcessingError::NonRetryable directly for known permanent failures.
57        ProcessingError::Retryable(err)
58    }
59}
60
61#[cfg(test)]
62mod tests {
63    use super::*;
64
65    #[test]
66    fn test_is_connection_error_only_matches_connection_variant() {
67        assert!(ProcessingError::Connection(anyhow::anyhow!("offline")).is_connection_error());
68        assert!(!ProcessingError::Retryable(anyhow::anyhow!("retry")).is_connection_error());
69        assert!(!ProcessingError::NonRetryable(anyhow::anyhow!("stop")).is_connection_error());
70    }
71
72    #[test]
73    fn test_anyhow_error_conversions_use_default_variants() {
74        let consumer_error = ConsumerError::from(anyhow::anyhow!("consumer failure"));
75        assert!(matches!(consumer_error, ConsumerError::Connection(_)));
76
77        let processing_error = ProcessingError::from(anyhow::anyhow!("processing failure"));
78        assert!(matches!(processing_error, ProcessingError::Retryable(_)));
79    }
80
81    #[test]
82    fn test_consumer_error_display_messages() {
83        assert_eq!(
84            ConsumerError::Gap {
85                requested: 42,
86                base: 9
87            }
88            .to_string(),
89            "consumer gap: requested offset 42 but earliest available is 9"
90        );
91        assert_eq!(
92            ConsumerError::EndOfStream.to_string(),
93            "consumer reached end of stream"
94        );
95    }
96}
mq_bridge/errors.rs

mq_bridge/
errors.rs
