openzeppelin_relayer/queues/
mod.rs

1//! Queue backend abstraction layer.
2//!
3//! This module provides a backend-agnostic interface for job queue operations.
4//! Implementations can use Redis/Apalis, AWS SQS, or GCP Pub/Sub as the backend.
5//!
6//! # Environment Variables
7//!
8//! - `QUEUE_BACKEND`: Backend to use ("redis", "sqs", or "pubsub" / "gcp-pubsub",
9//!   default: "redis")
10//!
11//! # Example
12//!
13//! ```ignore
14//! // Create backend from environment
15//! let backend = create_queue_backend(redis_connections).await?;
16//!
17//! // Produce a job
18//! backend.produce_transaction_request(job, None).await?;
19//!
20//! // Initialize workers on the pipeline runtime
21//! let workers = backend.initialize_workers(app_state, handle).await?;
22//! ```
23
24use async_trait::async_trait;
25use std::sync::Arc;
26
27use crate::{
28    config::ServerConfig,
29    jobs::{
30        Job, NotificationSend, RelayerHealthCheck, TokenSwapRequest, TransactionRequest,
31        TransactionSend, TransactionStatusCheck,
32    },
33    models::DefaultAppState,
34    utils::RedisConnections,
35};
36use actix_web::web::ThinData;
37
38pub mod cron;
39pub mod errors;
40pub mod pubsub;
41pub mod queue_type;
42pub mod redis;
43pub mod retry_config;
44pub mod sqs;
45pub mod swap_filter;
46pub mod worker_types;
47
48pub use errors::QueueBackendError;
49pub use queue_type::QueueType;
50pub use redis::queue::Queue;
51pub use retry_config::{backoff_config_for_queue, retry_delay_secs, status_check_retry_delay_secs};
52pub use swap_filter::filter_relayers_for_swap;
53pub use worker_types::{HandlerError, QueueHealth, WorkerContext, WorkerHandle};
54
55/// Supported queue backend implementations.
56#[derive(Debug, Clone, Copy, PartialEq, Eq)]
57pub enum QueueBackendType {
58    Redis,
59    Sqs,
60    PubSub,
61}
62
63impl QueueBackendType {
64    pub const fn as_str(self) -> &'static str {
65        match self {
66            Self::Redis => "redis",
67            Self::Sqs => "sqs",
68            Self::PubSub => "pubsub",
69        }
70    }
71}
72
73impl std::fmt::Display for QueueBackendType {
74    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
75        f.write_str(self.as_str())
76    }
77}
78
79/// Queue backend abstraction trait.
80///
81/// This trait defines the interface for job queue operations that can be
82/// implemented by different backends (Redis/Apalis, AWS SQS, etc.).
83///
84/// The trait is designed to be backend-agnostic, with no Redis or SQS-specific
85/// types in the interface.
86#[async_trait]
87pub trait QueueBackend: Send + Sync {
88    /// Produces a transaction request job to the queue.
89    ///
90    /// # Arguments
91    /// * `job` - The job to enqueue
92    /// * `scheduled_on` - Optional Unix timestamp for delayed execution
93    ///
94    /// # Returns
95    /// Result with job ID on success, or QueueBackendError on failure
96    async fn produce_transaction_request(
97        &self,
98        job: Job<TransactionRequest>,
99        scheduled_on: Option<i64>,
100    ) -> Result<String, QueueBackendError>;
101
102    /// Produces a transaction submission job to the queue.
103    async fn produce_transaction_submission(
104        &self,
105        job: Job<TransactionSend>,
106        scheduled_on: Option<i64>,
107    ) -> Result<String, QueueBackendError>;
108
109    /// Produces a transaction status check job to the queue.
110    async fn produce_transaction_status_check(
111        &self,
112        job: Job<TransactionStatusCheck>,
113        scheduled_on: Option<i64>,
114    ) -> Result<String, QueueBackendError>;
115
116    /// Produces a notification send job to the queue.
117    async fn produce_notification(
118        &self,
119        job: Job<NotificationSend>,
120        scheduled_on: Option<i64>,
121    ) -> Result<String, QueueBackendError>;
122
123    /// Produces a token swap request job to the queue.
124    async fn produce_token_swap_request(
125        &self,
126        job: Job<TokenSwapRequest>,
127        scheduled_on: Option<i64>,
128    ) -> Result<String, QueueBackendError>;
129
130    /// Produces a relayer health check job to the queue.
131    async fn produce_relayer_health_check(
132        &self,
133        job: Job<RelayerHealthCheck>,
134        scheduled_on: Option<i64>,
135    ) -> Result<String, QueueBackendError>;
136
137    /// Initializes and starts all worker tasks for this backend.
138    ///
139    /// Workers will poll their respective queues and process jobs using
140    /// the provided application state.
141    ///
142    /// # Arguments
143    /// * `app_state` - Application state containing handlers and configuration
144    /// * `handle` - Handle to the multi-thread pipeline runtime. Workers MUST be
145    ///   spawned via `handle.spawn` (not bare `tokio::spawn`) so background work
146    ///   is distributed across the pipeline runtime's worker threads instead of
147    ///   landing on the actix `System` arbiter's single thread.
148    ///
149    /// # Returns
150    /// Vector of worker handles that can be used to monitor or stop workers
151    async fn initialize_workers(
152        &self,
153        app_state: Arc<ThinData<DefaultAppState>>,
154        handle: tokio::runtime::Handle,
155    ) -> Result<Vec<WorkerHandle>, QueueBackendError>;
156
157    /// Performs a health check on all queues.
158    ///
159    /// Returns health status for each queue, including message counts
160    /// and backend-specific health indicators.
161    async fn health_check(&self) -> Result<Vec<QueueHealth>, QueueBackendError>;
162
163    /// Returns the backend type identifier.
164    fn backend_type(&self) -> QueueBackendType;
165
166    /// Signals all workers to shut down gracefully.
167    ///
168    /// The default implementation is a no-op. Redis, SQS, and Pub/Sub backends all
169    /// override this to broadcast a shutdown signal (via a `watch` channel) into
170    /// their respective polling loops / cron tasks / Apalis Monitor signal futures,
171    /// so a programmatic shutdown (no OS signal) still stops them promptly.
172    fn shutdown(&self) {}
173}
174
175/// Enum-based queue backend storage, following the codebase convention
176/// used by `SignerRepositoryStorage`, `NetworkRepositoryStorage`, etc.
177///
178/// Provides static dispatch over the concrete backend implementations
179/// instead of `dyn QueueBackend` trait objects.
180#[derive(Clone)]
181pub enum QueueBackendStorage {
182    Redis(Box<redis::backend::RedisBackend>),
183    Sqs(sqs::backend::SqsBackend),
184    PubSub(Box<pubsub::backend::PubSubBackend>),
185}
186
187impl std::fmt::Debug for QueueBackendStorage {
188    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
189        match self {
190            Self::Redis(b) => std::fmt::Debug::fmt(b, f),
191            Self::Sqs(b) => std::fmt::Debug::fmt(b, f),
192            Self::PubSub(b) => std::fmt::Debug::fmt(b, f),
193        }
194    }
195}
196
197impl QueueBackendStorage {
198    /// Returns a reference to the underlying `Queue` when the backend is Redis.
199    ///
200    /// Returns `None` for non-Redis backends (e.g. SQS, Pub/Sub) that do not use
201    /// a `Queue`.
202    pub fn queue(&self) -> Option<&Queue> {
203        match self {
204            Self::Redis(b) => Some(b.queue()),
205            Self::Sqs(_) | Self::PubSub(_) => None,
206        }
207    }
208
209    /// Returns the Redis connection pools when the backend is Redis.
210    ///
211    /// Delegates to `Queue::redis_connections()`. Returns `None` for non-Redis backends.
212    pub fn redis_connections(&self) -> Option<Arc<RedisConnections>> {
213        self.queue().map(|q| q.redis_connections())
214    }
215}
216
217#[async_trait]
218impl QueueBackend for QueueBackendStorage {
219    async fn produce_transaction_request(
220        &self,
221        job: Job<TransactionRequest>,
222        scheduled_on: Option<i64>,
223    ) -> Result<String, QueueBackendError> {
224        match self {
225            Self::Redis(b) => b.produce_transaction_request(job, scheduled_on).await,
226            Self::Sqs(b) => b.produce_transaction_request(job, scheduled_on).await,
227            Self::PubSub(b) => b.produce_transaction_request(job, scheduled_on).await,
228        }
229    }
230
231    async fn produce_transaction_submission(
232        &self,
233        job: Job<TransactionSend>,
234        scheduled_on: Option<i64>,
235    ) -> Result<String, QueueBackendError> {
236        match self {
237            Self::Redis(b) => b.produce_transaction_submission(job, scheduled_on).await,
238            Self::Sqs(b) => b.produce_transaction_submission(job, scheduled_on).await,
239            Self::PubSub(b) => b.produce_transaction_submission(job, scheduled_on).await,
240        }
241    }
242
243    async fn produce_transaction_status_check(
244        &self,
245        job: Job<TransactionStatusCheck>,
246        scheduled_on: Option<i64>,
247    ) -> Result<String, QueueBackendError> {
248        match self {
249            Self::Redis(b) => b.produce_transaction_status_check(job, scheduled_on).await,
250            Self::Sqs(b) => b.produce_transaction_status_check(job, scheduled_on).await,
251            Self::PubSub(b) => b.produce_transaction_status_check(job, scheduled_on).await,
252        }
253    }
254
255    async fn produce_notification(
256        &self,
257        job: Job<NotificationSend>,
258        scheduled_on: Option<i64>,
259    ) -> Result<String, QueueBackendError> {
260        match self {
261            Self::Redis(b) => b.produce_notification(job, scheduled_on).await,
262            Self::Sqs(b) => b.produce_notification(job, scheduled_on).await,
263            Self::PubSub(b) => b.produce_notification(job, scheduled_on).await,
264        }
265    }
266
267    async fn produce_token_swap_request(
268        &self,
269        job: Job<TokenSwapRequest>,
270        scheduled_on: Option<i64>,
271    ) -> Result<String, QueueBackendError> {
272        match self {
273            Self::Redis(b) => b.produce_token_swap_request(job, scheduled_on).await,
274            Self::Sqs(b) => b.produce_token_swap_request(job, scheduled_on).await,
275            Self::PubSub(b) => b.produce_token_swap_request(job, scheduled_on).await,
276        }
277    }
278
279    async fn produce_relayer_health_check(
280        &self,
281        job: Job<RelayerHealthCheck>,
282        scheduled_on: Option<i64>,
283    ) -> Result<String, QueueBackendError> {
284        match self {
285            Self::Redis(b) => b.produce_relayer_health_check(job, scheduled_on).await,
286            Self::Sqs(b) => b.produce_relayer_health_check(job, scheduled_on).await,
287            Self::PubSub(b) => b.produce_relayer_health_check(job, scheduled_on).await,
288        }
289    }
290
291    async fn initialize_workers(
292        &self,
293        app_state: Arc<ThinData<DefaultAppState>>,
294        handle: tokio::runtime::Handle,
295    ) -> Result<Vec<WorkerHandle>, QueueBackendError> {
296        match self {
297            Self::Redis(b) => b.initialize_workers(app_state, handle).await,
298            Self::Sqs(b) => b.initialize_workers(app_state, handle).await,
299            Self::PubSub(b) => b.initialize_workers(app_state, handle).await,
300        }
301    }
302
303    async fn health_check(&self) -> Result<Vec<QueueHealth>, QueueBackendError> {
304        match self {
305            Self::Redis(b) => b.health_check().await,
306            Self::Sqs(b) => b.health_check().await,
307            Self::PubSub(b) => b.health_check().await,
308        }
309    }
310
311    fn backend_type(&self) -> QueueBackendType {
312        match self {
313            Self::Redis(b) => b.backend_type(),
314            Self::Sqs(b) => b.backend_type(),
315            Self::PubSub(b) => b.backend_type(),
316        }
317    }
318
319    fn shutdown(&self) {
320        match self {
321            Self::Redis(b) => b.shutdown(),
322            Self::Sqs(b) => b.shutdown(),
323            Self::PubSub(b) => b.shutdown(),
324        }
325    }
326}
327
328/// Creates a queue backend based on the QUEUE_BACKEND environment variable.
329///
330/// # Arguments
331/// * `redis_connections` - Redis connection pools (used by Redis backend, ignored by SQS)
332///
333/// # Environment Variables
334/// - `QUEUE_BACKEND`: Backend to use ("redis" or "sqs", default: "redis")
335///
336/// # Returns
337/// Arc-wrapped `QueueBackendStorage` implementing QueueBackend
338///
339/// # Errors
340/// Returns QueueBackendError::ConfigError if:
341/// - QUEUE_BACKEND contains an unsupported value
342/// - Required backend-specific configuration is missing
343pub async fn create_queue_backend(
344    redis_connections: Arc<RedisConnections>,
345) -> Result<Arc<QueueBackendStorage>, QueueBackendError> {
346    let backend_type = ServerConfig::get_queue_backend();
347
348    let storage = match backend_type.to_lowercase().as_str() {
349        "redis" => {
350            let backend = redis::backend::RedisBackend::new(redis_connections).await?;
351            QueueBackendStorage::Redis(Box::new(backend))
352        }
353        "sqs" => {
354            let backend = sqs::backend::SqsBackend::new().await?;
355            QueueBackendStorage::Sqs(backend)
356        }
357        "pubsub" | "gcp-pubsub" => {
358            let backend = pubsub::backend::PubSubBackend::new(redis_connections).await?;
359            QueueBackendStorage::PubSub(Box::new(backend))
360        }
361        other => {
362            return Err(QueueBackendError::ConfigError(format!(
363                "Unsupported QUEUE_BACKEND value: {other}. Must be 'redis', 'sqs', or 'pubsub' (alias 'gcp-pubsub')"
364            )));
365        }
366    };
367
368    Ok(Arc::new(storage))
369}
370
371#[cfg(test)]
372mod tests {
373    use super::*;
374
375    #[test]
376    fn test_queue_type_enum_values() {
377        // Ensure all QueueType variants are covered
378        let types = vec![
379            QueueType::TransactionRequest,
380            QueueType::TransactionSubmission,
381            QueueType::StatusCheck,
382            QueueType::StatusCheckEvm,
383            QueueType::StatusCheckStellar,
384            QueueType::Notification,
385            QueueType::TokenSwapRequest,
386            QueueType::RelayerHealthCheck,
387        ];
388
389        for queue_type in types {
390            assert!(!queue_type.queue_name().is_empty());
391            assert!(!queue_type.redis_namespace().is_empty());
392        }
393    }
394
395    #[test]
396    fn test_queue_type_visibility_timeouts_in_range() {
397        // All visibility timeouts should be within SQS limits (0-43200).
398        let all_types = [
399            QueueType::TransactionRequest,
400            QueueType::TransactionSubmission,
401            QueueType::StatusCheck,
402            QueueType::StatusCheckEvm,
403            QueueType::StatusCheckStellar,
404            QueueType::Notification,
405            QueueType::TokenSwapRequest,
406            QueueType::RelayerHealthCheck,
407        ];
408        for qt in all_types {
409            let vt = qt.visibility_timeout_secs();
410            assert!(vt > 0, "{qt}: visibility timeout must be > 0");
411            assert!(
412                vt <= 43200,
413                "{qt}: visibility timeout {vt}s exceeds SQS max (43200s)"
414            );
415        }
416    }
417
418    #[test]
419    fn test_queue_type_polling_intervals_appropriate() {
420        // Status check should poll most frequently
421        assert_eq!(QueueType::StatusCheck.default_wait_time_secs(), 5);
422
423        // Others should be slower
424        assert!(QueueType::TransactionRequest.default_wait_time_secs() >= 5);
425        assert!(QueueType::TransactionSubmission.default_wait_time_secs() >= 5);
426        assert!(QueueType::Notification.default_wait_time_secs() >= 10);
427    }
428
429    #[test]
430    fn test_queue_backend_error_variants() {
431        let errors = vec![
432            QueueBackendError::RedisError("test".to_string()),
433            QueueBackendError::SqsError("test".to_string()),
434            QueueBackendError::SerializationError("test".to_string()),
435            QueueBackendError::ConfigError("test".to_string()),
436            QueueBackendError::QueueNotFound("test".to_string()),
437            QueueBackendError::WorkerInitError("test".to_string()),
438            QueueBackendError::QueueError("test".to_string()),
439        ];
440
441        for error in errors {
442            let error_str = error.to_string();
443            assert!(!error_str.is_empty());
444        }
445    }
446
447    #[test]
448    fn test_queue_backend_type_string_representations() {
449        assert_eq!(QueueBackendType::Redis.as_str(), "redis");
450        assert_eq!(QueueBackendType::Sqs.as_str(), "sqs");
451        assert_eq!(QueueBackendType::PubSub.as_str(), "pubsub");
452        assert_eq!(QueueBackendType::Redis.to_string(), "redis");
453        assert_eq!(QueueBackendType::Sqs.to_string(), "sqs");
454        assert_eq!(QueueBackendType::PubSub.to_string(), "pubsub");
455    }
456
457    // ── create_queue_backend dispatch ──────────────────────────────
458    //
459    // Serializes the env mutation these tests need. The Pub/Sub path fast-fails
460    // on a missing PUBSUB_PROJECT_ID *before* any network/auth, so both the
461    // alias-routing and unsupported-value checks stay deterministic and offline.
462    use std::sync::Mutex as StdMutex;
463    static BACKEND_ENV_LOCK: StdMutex<()> = StdMutex::new(());
464
465    fn dummy_redis_connections() -> Arc<RedisConnections> {
466        // deadpool builds lazily, so this never opens a connection.
467        let pool = deadpool_redis::Config::from_url("redis://127.0.0.1:6379")
468            .builder()
469            .expect("pool builder")
470            .max_size(1)
471            .runtime(deadpool_redis::Runtime::Tokio1)
472            .build()
473            .expect("pool build");
474        Arc::new(RedisConnections::new_single_pool(Arc::new(pool)))
475    }
476
477    #[tokio::test]
478    async fn test_create_queue_backend_rejects_unsupported_value() {
479        let _lock = BACKEND_ENV_LOCK.lock().unwrap_or_else(|p| p.into_inner());
480        let prev = std::env::var("QUEUE_BACKEND").ok();
481
482        std::env::set_var("QUEUE_BACKEND", "kafka-not-real");
483        let result = create_queue_backend(dummy_redis_connections()).await;
484
485        match prev {
486            Some(v) => std::env::set_var("QUEUE_BACKEND", v),
487            None => std::env::remove_var("QUEUE_BACKEND"),
488        }
489
490        let err = result
491            .expect_err("unsupported backend must error")
492            .to_string();
493        assert!(
494            err.contains("Unsupported QUEUE_BACKEND"),
495            "expected unsupported error, got: {err}"
496        );
497        // The error must advertise pubsub as a valid option.
498        assert!(err.contains("pubsub"), "error should list pubsub: {err}");
499    }
500
501    // No regression: adding the Pub/Sub backend must not remove redis/sqs
502    // from the dispatch — both are still recognized as valid backend selectors.
503    #[tokio::test]
504    async fn test_redis_and_sqs_backends_still_recognized() {
505        let _lock = BACKEND_ENV_LOCK.lock().unwrap_or_else(|p| p.into_inner());
506        let prev = std::env::var("QUEUE_BACKEND").ok();
507
508        std::env::set_var("QUEUE_BACKEND", "definitely-not-a-backend");
509        let err = create_queue_backend(dummy_redis_connections())
510            .await
511            .expect_err("unsupported value must error")
512            .to_string();
513
514        match prev {
515            Some(v) => std::env::set_var("QUEUE_BACKEND", v),
516            None => std::env::remove_var("QUEUE_BACKEND"),
517        }
518
519        // The unsupported-value error still lists redis and sqs alongside pubsub,
520        // proving the pre-existing backends were not dropped from the dispatch.
521        assert!(err.contains("redis"), "redis must remain selectable: {err}");
522        assert!(err.contains("sqs"), "sqs must remain selectable: {err}");
523        assert!(err.contains("pubsub"), "pubsub must be selectable: {err}");
524    }
525
526    #[tokio::test]
527    async fn test_create_queue_backend_routes_pubsub_aliases() {
528        let _lock = BACKEND_ENV_LOCK.lock().unwrap_or_else(|p| p.into_inner());
529        let prev_backend = std::env::var("QUEUE_BACKEND").ok();
530        let prev_project = std::env::var("PUBSUB_PROJECT_ID").ok();
531
532        // Force the Pub/Sub path to fast-fail before any network call.
533        std::env::remove_var("PUBSUB_PROJECT_ID");
534
535        for alias in ["pubsub", "gcp-pubsub"] {
536            std::env::set_var("QUEUE_BACKEND", alias);
537            let err = create_queue_backend(dummy_redis_connections())
538                .await
539                .expect_err("missing PUBSUB_PROJECT_ID must error")
540                .to_string();
541            // Routed to the Pub/Sub branch (NOT the catch-all "Unsupported").
542            assert!(
543                !err.contains("Unsupported QUEUE_BACKEND"),
544                "alias '{alias}' should route to the Pub/Sub backend, got: {err}"
545            );
546            assert!(
547                err.contains("PUBSUB_PROJECT_ID"),
548                "alias '{alias}' should fail fast on missing project id, got: {err}"
549            );
550        }
551
552        match prev_backend {
553            Some(v) => std::env::set_var("QUEUE_BACKEND", v),
554            None => std::env::remove_var("QUEUE_BACKEND"),
555        }
556        if let Some(v) = prev_project {
557            std::env::set_var("PUBSUB_PROJECT_ID", v);
558        }
559    }
560}
openzeppelin_relayer/queues/mod.rs

openzeppelin_relayer/queues/
mod.rs
