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
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
//! Error types: [`enum@Error`] for infrastructure, [`JobError`] for handlers.
use thiserror::Error;
/// `Result` alias defaulting to this crate's [`enum@Error`].
pub type Result<T, E = Error> = std::result::Result<T, E>;
/// Infrastructure-level errors (serialization, transport, configuration).
#[derive(Debug, Error)]
pub enum Error {
/// A job payload or envelope could not be (de)serialized.
#[error("serialization error: {0}")]
Serde(#[from] serde_json::Error),
/// The transport failed.
#[error("backend error: {0}")]
Backend(#[source] Box<dyn std::error::Error + Send + Sync + 'static>),
/// A queue name that is not part of the queue set was used.
#[error("unknown queue `{0}`")]
UnknownQueue(String),
/// No handler is registered for a job type.
#[error("no handler registered for job type `{0}`")]
NoHandler(String),
/// Two handlers claimed the same `Job::NAME`.
#[error("handler for job type `{0}` registered twice")]
DuplicateHandler(String),
/// The worker or backend is shut down.
#[error("worker is shut down")]
ShutDown,
/// A consumer stream ended on its own, which means the backend went away.
#[error("consumer for queue `{0}` stopped unexpectedly")]
ConsumerStopped(String),
}
impl Error {
/// Wrap a transport error.
pub fn backend<E: std::error::Error + Send + Sync + 'static>(e: E) -> Self {
Self::Backend(Box::new(e))
}
}
/// Error returned by a [`crate::JobHandler`].
#[derive(Debug, Error)]
pub enum JobError {
/// Transient failure; the retry policy decides whether to retry.
#[error("job failed (retryable): {0}")]
Retryable(#[source] Box<dyn std::error::Error + Send + Sync + 'static>),
/// Permanent failure; go straight to dead-letter regardless of policy.
#[error("job failed (fatal): {0}")]
Fatal(#[source] Box<dyn std::error::Error + Send + Sync + 'static>),
/// Not a failure: the job could not run *yet* and must be tried again in `delay`.
///
/// The motivating case is an external API answering `429 Too Many Requests` with
/// `Retry-After: 30`: nothing went wrong. The job has to wait exactly that
/// long and then run *before* the backlog that piled up meanwhile.
///
/// Unlike [`JobError::Retryable`]:
///
/// * [`crate::Envelope::attempt`] is **unchanged**, so a deferral never burns down
/// the retry budget and a job may defer itself indefinitely;
/// * the retry policy is **not consulted**: neither its backoff (the handler
/// states the delay) nor its `max_attempts` (nothing failed, so nothing is
/// dead-lettered);
/// * [`crate::Envelope::deferrals`] is incremented and the envelope comes back with
/// the highest priority its queue supports, ahead of normally enqueued work;
/// * the worker logs it at `INFO`, not `WARN`/`ERROR`.
///
/// There is no built-in cap: a handler that wants one inspects
/// [`crate::JobContext::deferrals`] and returns [`JobError::Fatal`] instead.
#[error("job deferred for {delay:?}: {reason}")]
Deferred {
/// How long the job must wait before it is delivered again.
delay: std::time::Duration,
/// Why it was deferred, for logs. Not part of any control flow.
reason: String,
},
}
impl JobError {
/// Wrap `e` as a transient failure.
pub fn retryable<E: std::error::Error + Send + Sync + 'static>(e: E) -> Self {
Self::Retryable(Box::new(e))
}
/// Wrap `e` as a permanent failure.
pub fn fatal<E: std::error::Error + Send + Sync + 'static>(e: E) -> Self {
Self::Fatal(Box::new(e))
}
/// Transient failure with a plain message.
pub fn retryable_msg(msg: impl Into<String>) -> Self {
Self::Retryable(msg.into().into())
}
/// Permanent failure with a plain message.
pub fn fatal_msg(msg: impl Into<String>) -> Self {
Self::Fatal(msg.into().into())
}
/// Defer the job by `delay` with the generic reason `"deferred"`.
///
/// Not a failure: the attempt counter is untouched and the retry policy is not
/// consulted. See [`JobError::Deferred`].
pub fn deferred(delay: std::time::Duration) -> Self {
Self::Deferred {
delay,
reason: "deferred".to_owned(),
}
}
/// Defer the job by `delay`, recording why (`"rate limited: Retry-After 30s"`).
///
/// Not a failure: the attempt counter is untouched and the retry policy is not
/// consulted. See [`JobError::Deferred`].
pub fn deferred_msg(delay: std::time::Duration, msg: impl Into<String>) -> Self {
Self::Deferred {
delay,
reason: msg.into(),
}
}
}
/// Convenience: any `std::error::Error` becomes a retryable job error.
impl From<Box<dyn std::error::Error + Send + Sync + 'static>> for JobError {
fn from(e: Box<dyn std::error::Error + Send + Sync + 'static>) -> Self {
Self::Retryable(e)
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::time::Duration;
#[test]
fn deferred_uses_a_generic_reason() {
let err = JobError::deferred(Duration::from_secs(30));
match &err {
JobError::Deferred { delay, reason } => {
assert_eq!(*delay, Duration::from_secs(30));
assert_eq!(reason, "deferred");
}
other => panic!("unexpected error: {other}"),
}
assert_eq!(err.to_string(), "job deferred for 30s: deferred");
}
#[test]
fn deferred_msg_keeps_the_message() {
let err = JobError::deferred_msg(Duration::from_millis(1_500), "rate limited");
match &err {
JobError::Deferred { delay, reason } => {
assert_eq!(*delay, Duration::from_millis(1_500));
assert_eq!(reason, "rate limited");
}
other => panic!("unexpected error: {other}"),
}
assert_eq!(err.to_string(), "job deferred for 1.5s: rate limited");
}
#[test]
fn a_deferral_carries_no_source_error() {
use std::error::Error as _;
assert!(
JobError::deferred(Duration::from_secs(1))
.source()
.is_none()
);
assert!(JobError::retryable_msg("boom").source().is_some());
}
}