queue-runtime 0.2.0

Multi-provider queue runtime for Queue-Keeper
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148

//! Error types for queue operations.

use crate::message::Timestamp;
use chrono::Duration;
use thiserror::Error;

/// Comprehensive error type for all queue operations
#[derive(Debug, Error)]
pub enum QueueError {
    #[error("Queue not found: {queue_name}")]
    QueueNotFound { queue_name: String },

    #[error("Message not found: {receipt}")]
    MessageNotFound { receipt: String },

    #[error("Invalid or expired receipt handle: {receipt}")]
    InvalidReceipt { receipt: String },

    #[error("Session '{session_id}' is locked until {locked_until}")]
    SessionLocked {
        session_id: String,
        locked_until: Timestamp,
    },

    #[error("Session '{session_id}' not found or expired")]
    SessionNotFound { session_id: String },

    #[error("Operation timed out after {duration:?}")]
    Timeout { duration: Duration },

    #[error("Connection failed: {message}")]
    ConnectionFailed { message: String },

    #[error("Authentication failed: {message}")]
    AuthenticationFailed { message: String },

    #[error("Permission denied for operation: {operation}")]
    PermissionDenied { operation: String },

    #[error("Message too large: {size} bytes (max: {max_size})")]
    MessageTooLarge { size: usize, max_size: usize },

    #[error("Batch size {size} exceeds maximum {max_size}")]
    BatchTooLarge { size: usize, max_size: usize },

    #[error("Provider error ({provider}): {code} - {message}")]
    ProviderError {
        provider: String,
        code: String,
        message: String,
    },

    #[error("Serialization failed: {0}")]
    SerializationError(#[from] SerializationError),

    #[error("Configuration error: {0}")]
    ConfigurationError(#[from] ConfigurationError),

    #[error("Validation error: {0}")]
    ValidationError(#[from] ValidationError),
}

impl QueueError {
    /// Check if error is transient and should be retried
    pub fn is_transient(&self) -> bool {
        match self {
            Self::QueueNotFound { .. } => false,
            Self::MessageNotFound { .. } => false,
            Self::InvalidReceipt { .. } => false,
            Self::SessionLocked { .. } => true,
            Self::SessionNotFound { .. } => false,
            Self::Timeout { .. } => true,
            Self::ConnectionFailed { .. } => true,
            Self::AuthenticationFailed { .. } => false,
            Self::PermissionDenied { .. } => false,
            Self::MessageTooLarge { .. } => false,
            Self::BatchTooLarge { .. } => false,
            Self::ProviderError { .. } => true, // Provider-specific errors are usually transient
            Self::SerializationError(_) => false,
            Self::ConfigurationError(_) => false,
            Self::ValidationError(_) => false,
        }
    }

    /// Check if error should be retried
    pub fn should_retry(&self) -> bool {
        self.is_transient()
    }

    /// Get suggested retry delay
    pub fn retry_after(&self) -> Option<Duration> {
        match self {
            Self::SessionLocked { .. } => Some(Duration::seconds(5)),
            Self::Timeout { .. } => Some(Duration::seconds(1)),
            Self::ConnectionFailed { .. } => Some(Duration::seconds(5)),
            _ => None,
        }
    }
}

/// Errors during message serialization/deserialization
#[derive(Debug, Error)]
pub enum SerializationError {
    #[error("JSON serialization failed: {0}")]
    JsonError(#[from] serde_json::Error),

    #[error("Message body is not valid UTF-8")]
    InvalidUtf8,

    #[error("Message attribute '{key}' has invalid value")]
    InvalidAttribute { key: String },

    #[error("Message exceeds size limit: {size} bytes")]
    MessageTooLarge { size: usize },
}

/// Configuration errors
#[derive(Debug, Error)]
pub enum ConfigurationError {
    #[error("Invalid configuration: {message}")]
    Invalid { message: String },

    #[error("Missing required configuration: {key}")]
    Missing { key: String },

    #[error("Configuration parsing failed: {message}")]
    Parsing { message: String },

    #[error("Unsupported provider: {provider} - {message}")]
    UnsupportedProvider { provider: String, message: String },
}

/// Validation errors
#[derive(Debug, Error)]
pub enum ValidationError {
    #[error("Required field missing: {field}")]
    Required { field: String },

    #[error("Invalid format for {field}: {message}")]
    InvalidFormat { field: String, message: String },

    #[error("Value out of range for {field}: {message}")]
    OutOfRange { field: String, message: String },
}

#[cfg(test)]
#[path = "error_tests.rs"]
mod tests;