openzeppelin_relayer/queues/
worker_types.rs

1//! Worker types for the queue abstraction.
2use serde::{Deserialize, Serialize};
3use std::fmt;
4use std::sync::Arc;
5
6use crate::queues::QueueType;
7
8/// Handle to a running worker task.
9#[derive(Debug)]
10pub enum WorkerHandle {
11    Apalis(Box<dyn std::any::Any + Send>),
12    Tokio(tokio::task::JoinHandle<()>),
13}
14
15/// Queue health status information.
16#[derive(Debug, Clone, Serialize, Deserialize)]
17pub struct QueueHealth {
18    pub queue_type: QueueType,
19    /// Backlog depth (undelivered messages). `None` means the depth is
20    /// **unavailable** (e.g. the backlog read failed or the backend cannot
21    /// report it), which must NOT be conflated with an empty queue (`Some(0)`).
22    pub messages_visible: Option<u64>,
23    pub messages_in_flight: u64,
24    pub messages_dlq: u64,
25    pub backend: String,
26    pub is_healthy: bool,
27}
28
29/// Backend-neutral context passed to all job handlers.
30#[derive(Debug, Clone)]
31pub struct WorkerContext {
32    pub attempt: usize,
33    pub task_id: String,
34}
35
36impl WorkerContext {
37    pub fn new(attempt: usize, task_id: String) -> Self {
38        Self { attempt, task_id }
39    }
40}
41
42/// Backend-neutral handler error for retry control.
43#[derive(Debug)]
44pub enum HandlerError {
45    Retry(String),
46    Abort(String),
47}
48
49impl fmt::Display for HandlerError {
50    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
51        match self {
52            Self::Retry(msg) => write!(f, "Retry: {msg}"),
53            Self::Abort(msg) => write!(f, "Abort: {msg}"),
54        }
55    }
56}
57
58impl std::error::Error for HandlerError {}
59
60impl From<HandlerError> for apalis::prelude::Error {
61    fn from(err: HandlerError) -> Self {
62        match err {
63            HandlerError::Retry(msg) => apalis::prelude::Error::Failed(Arc::new(msg.into())),
64            HandlerError::Abort(msg) => apalis::prelude::Error::Abort(Arc::new(msg.into())),
65        }
66    }
67}
68
69#[cfg(test)]
70mod tests {
71    use super::*;
72
73    #[test]
74    fn test_worker_context_new() {
75        let ctx = WorkerContext::new(3, "task-abc".to_string());
76        assert_eq!(ctx.attempt, 3);
77        assert_eq!(ctx.task_id, "task-abc");
78    }
79
80    #[test]
81    fn test_handler_error_retry_display() {
82        let err = HandlerError::Retry("connection timeout".to_string());
83        assert_eq!(err.to_string(), "Retry: connection timeout");
84    }
85
86    #[test]
87    fn test_handler_error_abort_display() {
88        let err = HandlerError::Abort("invalid payload".to_string());
89        assert_eq!(err.to_string(), "Abort: invalid payload");
90    }
91
92    #[test]
93    fn test_handler_error_retry_into_apalis_failed() {
94        let err = HandlerError::Retry("temp failure".to_string());
95        let apalis_err: apalis::prelude::Error = err.into();
96        assert!(
97            matches!(apalis_err, apalis::prelude::Error::Failed(_)),
98            "Retry should map to Failed"
99        );
100    }
101
102    #[test]
103    fn test_handler_error_abort_into_apalis_abort() {
104        let err = HandlerError::Abort("permanent failure".to_string());
105        let apalis_err: apalis::prelude::Error = err.into();
106        assert!(
107            matches!(apalis_err, apalis::prelude::Error::Abort(_)),
108            "Abort should map to Abort"
109        );
110    }
111}
openzeppelin_relayer/queues/worker_types.rs

openzeppelin_relayer/queues/
worker_types.rs
