openzeppelin_relayer/jobs/handlers/
mod.rs

1use eyre::Report;
2use tracing::{debug, error, warn};
3
4use crate::{
5    models::{ApiError, RepositoryError},
6    observability::request_id::get_request_id,
7    queues::{HandlerError, WorkerContext},
8};
9
10mod transaction_request_handler;
11pub use transaction_request_handler::*;
12
13mod transaction_submission_handler;
14pub use transaction_submission_handler::*;
15
16mod notification_handler;
17pub use notification_handler::*;
18
19mod transaction_status_handler;
20pub use transaction_status_handler::*;
21
22mod token_swap_request_handler;
23pub use token_swap_request_handler::*;
24
25mod transaction_cleanup_handler;
26pub use transaction_cleanup_handler::*;
27
28mod system_cleanup_handler;
29pub use system_cleanup_handler::*;
30
31// Handles job results for simple handlers (no transaction state management).
32//
33// Used by: notification_handler, solana_swap_request_handler, transaction_cleanup_handler
34//
35// # Retry Strategy
36// - On success: Job completes
37// - On error: Retry until max_attempts reached
38// - At max_attempts: Abort job
39mod relayer_health_check_handler;
40pub use relayer_health_check_handler::*;
41
42/// Returns `true` when retrying the job can never succeed, so it should be
43/// aborted immediately instead of consuming the retry budget.
44///
45/// A `NotFound` for the entity a job operates on (e.g. a transaction that was
46/// deleted or expired from the repository) will never reappear by re-running
47/// the same job. On standard SQS queues this is critical: each retry
48/// re-enqueues a fresh message with `ApproximateReceiveCount` reset to 1, so
49/// the attempt counter never advances and the job would otherwise loop
50/// forever, also bypassing the SQS dead-letter redrive policy.
51fn is_terminal_error(err: &Report) -> bool {
52    if let Some(api_err) = err.downcast_ref::<ApiError>() {
53        if matches!(api_err, ApiError::NotFound(_)) {
54            return true;
55        }
56    }
57    if let Some(repo_err) = err.downcast_ref::<RepositoryError>() {
58        if matches!(repo_err, RepositoryError::NotFound(_)) {
59            return true;
60        }
61    }
62    false
63}
64
65pub fn handle_result(
66    result: Result<(), Report>,
67    ctx: &WorkerContext,
68    job_type: &str,
69    max_attempts: usize,
70) -> Result<(), HandlerError> {
71    if result.is_ok() {
72        debug!(
73            job_type = %job_type,
74            request_id = ?get_request_id(),
75            "request handled successfully"
76        );
77        return Ok(());
78    }
79
80    let err = result.as_ref().unwrap_err();
81    warn!(
82        job_type = %job_type,
83        request_id = ?get_request_id(),
84        error = %err,
85        attempt = %ctx.attempt,
86        max_attempts = %max_attempts,
87        "request failed"
88    );
89
90    if is_terminal_error(err) {
91        error!(
92            job_type = %job_type,
93            request_id = ?get_request_id(),
94            error = %err,
95            "terminal error (entity not found), aborting job without retry"
96        );
97        return Err(HandlerError::Abort(
98            "Terminal error: target entity not found, not retryable".into(),
99        ));
100    }
101
102    if ctx.attempt >= max_attempts {
103        error!(
104            job_type = %job_type,
105            request_id = ?get_request_id(),
106            max_attempts = %max_attempts,
107            "max attempts reached, failing job"
108        );
109        return Err(HandlerError::Abort("Failed to handle request".into()));
110    }
111
112    Err(HandlerError::Retry(
113        "Failed to handle request. Retrying".into(),
114    ))
115}
116
117#[cfg(test)]
118mod tests {
119    use super::*;
120
121    #[test]
122    fn test_handle_result_success() {
123        let result: Result<(), Report> = Ok(());
124        let ctx = WorkerContext::new(0, "test-task".into());
125
126        let handled = handle_result(result, &ctx, "test_job", 3);
127        assert!(handled.is_ok());
128    }
129
130    #[test]
131    fn test_handle_result_retry() {
132        let result: Result<(), Report> = Err(Report::msg("Test error"));
133        let ctx = WorkerContext::new(0, "test-task".into());
134
135        let handled = handle_result(result, &ctx, "test_job", 3);
136
137        assert!(handled.is_err());
138        assert!(matches!(handled, Err(HandlerError::Retry(_))));
139    }
140
141    #[test]
142    fn test_handle_result_abort() {
143        let result: Result<(), Report> = Err(Report::msg("Test error"));
144        let ctx = WorkerContext::new(3, "test-task".into());
145
146        let handled = handle_result(result, &ctx, "test_job", 3);
147
148        assert!(handled.is_err());
149        assert!(matches!(handled, Err(HandlerError::Abort(_))));
150    }
151
152    #[test]
153    fn test_handle_result_max_attempts_exceeded() {
154        let result: Result<(), Report> = Err(Report::msg("Test error"));
155        let ctx = WorkerContext::new(5, "test-task".into());
156
157        let handled = handle_result(result, &ctx, "test_job", 3);
158
159        assert!(handled.is_err());
160        assert!(matches!(handled, Err(HandlerError::Abort(_))));
161    }
162
163    #[test]
164    fn test_handle_result_terminal_api_not_found_aborts_at_first_attempt() {
165        // A missing entity will never appear by retrying the same job, so a
166        // NotFound must abort immediately instead of burning the whole retry
167        // budget (or looping forever on standard SQS queues where attempt stays 0).
168        let result: Result<(), Report> = Err(Report::new(crate::models::ApiError::NotFound(
169            "Transaction with ID 4af0a83b not found".into(),
170        )));
171        let ctx = WorkerContext::new(0, "test-task".into());
172
173        let handled = handle_result(result, &ctx, "Transaction Request", 5);
174
175        assert!(matches!(handled, Err(HandlerError::Abort(_))));
176    }
177
178    #[test]
179    fn test_handle_result_terminal_repository_not_found_aborts_at_first_attempt() {
180        let result: Result<(), Report> = Err(Report::new(
181            crate::models::RepositoryError::NotFound("relayer xyz not found".into()),
182        ));
183        let ctx = WorkerContext::new(0, "test-task".into());
184
185        let handled = handle_result(result, &ctx, "Transaction Request", 5);
186
187        assert!(matches!(handled, Err(HandlerError::Abort(_))));
188    }
189
190    #[test]
191    fn test_handle_result_non_terminal_error_still_retries() {
192        // Regression guard: transient errors must keep retrying below max.
193        let result: Result<(), Report> = Err(Report::new(
194            crate::models::RepositoryError::ConnectionError("redis timeout".into()),
195        ));
196        let ctx = WorkerContext::new(0, "test-task".into());
197
198        let handled = handle_result(result, &ctx, "Transaction Request", 5);
199
200        assert!(matches!(handled, Err(HandlerError::Retry(_))));
201    }
202}
openzeppelin_relayer/jobs/handlers/mod.rs

openzeppelin_relayer/jobs/handlers/
mod.rs
