rustlavel_queue/
job.rs

1//! What a job is, and how a worker finds one again after it has been stored.
2//!
3//! ```ignore
4//! struct SendWelcomeEmail { user_id: i64 }
5//!
6//! impl Job for SendWelcomeEmail {
7//!     const NAME: &'static str = "send-welcome-email";
8//!
9//!     fn payload(&self) -> Json {
10//!         Json::object([("user_id", Json::from(self.user_id))])
11//!     }
12//!
13//!     fn from_payload(payload: &Json) -> Result<Self> {
14//!         Ok(SendWelcomeEmail { user_id: payload.get("user_id").and_then(Json::as_i64).unwrap_or(0) })
15//!     }
16//!
17//!     fn handle(&self) -> impl Future<Output = Result<()>> + Send {
18//!         async move { Ok(()) }
19//!     }
20//! }
21//! ```
22
23use rustlavel_core::{Error, Json, Result};
24use std::collections::HashMap;
25use std::future::Future;
26use std::pin::Pin;
27use std::sync::Arc;
28use std::time::Duration;
29
30/// The queue a job goes on when it does not ask for another.
31pub const DEFAULT_QUEUE: &str = "default";
32
33/// How many times a job is attempted before it is considered dead.
34///
35/// Three matches Laravel, and is the number that turns "the API blipped" into a
36/// non-event without turning "this job is broken" into an infinite loop.
37pub const DEFAULT_TRIES: u32 = 3;
38
39/// The base delay before a failed job is retried. Doubled on each attempt by
40/// the worker's backoff.
41pub const DEFAULT_RETRY_AFTER: Duration = Duration::from_secs(60);
42
43/// A boxed future, the shape everything dyn-compatible in this crate returns.
44pub type BoxFuture<'a, T> = Pin<Box<dyn Future<Output = T> + Send + 'a>>;
45
46/// A unit of work that can be stored and run later.
47///
48/// Deliberately *not* dyn-compatible: `NAME` and `from_payload` have no
49/// receiver, and `handle` returns an opaque future. Nothing needs a `dyn Job`,
50/// because the moment a job is queued it becomes a [`QueuedJob`] — plain data —
51/// and the way back to a type is the [`JobRegistry`].
52pub trait Job: Send + Sync + 'static {
53    /// The stable name this job is stored under.
54    ///
55    /// It travels into the database and back out again, possibly across a
56    /// deploy, so renaming it strands every job already queued. Choose it the
57    /// way you would choose a table name.
58    const NAME: &'static str;
59
60    /// Everything `from_payload` needs to rebuild this job.
61    fn payload(&self) -> Json;
62
63    /// Rebuild the job from a payload the queue handed back.
64    ///
65    /// Returning an error here fails the job rather than panicking the worker,
66    /// which is what you want when an old payload meets new code.
67    fn from_payload(payload: &Json) -> Result<Self>
68    where
69        Self: Sized;
70
71    /// Do the work.
72    ///
73    /// Declared as `-> impl Future + Send` rather than `async fn` because the
74    /// registry boxes the future and sends it to another task, and an `async
75    /// fn` *in a trait* promises nothing about `Send`. An implementation is
76    /// free to write `async fn handle(&self) -> Result<()>` instead: the
77    /// compiler can see the body, so it can prove the `Send` this demands.
78    fn handle(&self) -> impl Future<Output = Result<()>> + Send;
79
80    /// How many attempts this job gets before it is dead-lettered.
81    fn tries(&self) -> u32 {
82        DEFAULT_TRIES
83    }
84
85    /// How long to wait before the first retry. The worker doubles it per
86    /// attempt.
87    fn retry_after(&self) -> Duration {
88        DEFAULT_RETRY_AFTER
89    }
90
91    /// Which queue this job belongs on.
92    fn queue(&self) -> &'static str {
93        DEFAULT_QUEUE
94    }
95
96    /// Package the job for storage. Rarely called directly — `push` does it.
97    fn to_queued(&self) -> QueuedJob {
98        QueuedJob {
99            name: Self::NAME.to_string(),
100            payload: self.payload(),
101            queue: self.queue().to_string(),
102            max_tries: self.tries().max(1),
103            retry_after: self.retry_after(),
104            delay: Duration::ZERO,
105        }
106    }
107}
108
109/// A job as it is stored: a name, a payload, and the policy for running it.
110///
111/// Everything past the dispatch call deals in this rather than in a concrete
112/// type, which is what keeps the [`Queue`](crate::Queue) trait dyn-compatible.
113#[derive(Debug, Clone, PartialEq)]
114pub struct QueuedJob {
115    pub name: String,
116    pub payload: Json,
117    pub queue: String,
118    pub max_tries: u32,
119    pub retry_after: Duration,
120    /// How long after being pushed the job becomes visible to a worker.
121    pub delay: Duration,
122}
123
124impl QueuedJob {
125    /// An envelope built by hand, for a caller that has a name and a payload
126    /// but no Rust type — a job pushed by another service, say.
127    pub fn new(name: impl Into<String>, payload: Json) -> Self {
128        QueuedJob {
129            name: name.into(),
130            payload,
131            queue: DEFAULT_QUEUE.to_string(),
132            max_tries: DEFAULT_TRIES,
133            retry_after: DEFAULT_RETRY_AFTER,
134            delay: Duration::ZERO,
135        }
136    }
137
138    pub fn on_queue(mut self, queue: impl Into<String>) -> Self {
139        self.queue = queue.into();
140        self
141    }
142
143    pub fn with_delay(mut self, delay: Duration) -> Self {
144        self.delay = delay;
145        self
146    }
147
148    pub fn with_tries(mut self, tries: u32) -> Self {
149        self.max_tries = tries.max(1);
150        self
151    }
152
153    pub fn with_retry_after(mut self, retry_after: Duration) -> Self {
154        self.retry_after = retry_after;
155        self
156    }
157}
158
159/// A job a worker has taken off the queue and is responsible for.
160///
161/// It is not gone: until the worker deletes, releases or fails it, the job is
162/// still the queue's business, which is what makes a crashed worker recoverable.
163#[derive(Debug, Clone, PartialEq)]
164pub struct ReservedJob {
165    /// The driver's identifier for this job. Opaque to everything else.
166    pub id: String,
167    pub job: QueuedJob,
168    /// How many times this job has now been handed to a worker, counting this
169    /// time. Always at least 1.
170    pub attempts: u32,
171}
172
173impl ReservedJob {
174    /// Whether this attempt is the last one the job is allowed.
175    pub fn is_last_attempt(&self) -> bool {
176        self.attempts >= self.job.max_tries
177    }
178}
179
180/// A job that used up every attempt, kept so a human can look at it.
181#[derive(Debug, Clone, PartialEq)]
182pub struct FailedJob {
183    pub id: String,
184    pub name: String,
185    pub queue: String,
186    pub payload: Json,
187    pub attempts: u32,
188    /// What the last attempt said on its way out.
189    pub error: String,
190    /// Epoch seconds.
191    pub failed_at: i64,
192}
193
194impl FailedJob {
195    /// The shape a dashboard or `queue:failed` wants.
196    pub fn to_json(&self) -> Json {
197        Json::object([
198            ("id", Json::from(self.id.as_str())),
199            ("name", Json::from(self.name.as_str())),
200            ("queue", Json::from(self.queue.as_str())),
201            ("payload", self.payload.clone()),
202            ("attempts", Json::from(self.attempts)),
203            ("error", Json::from(self.error.as_str())),
204            ("failed_at", Json::from(self.failed_at)),
205        ])
206    }
207}
208
209/// What the registry stores: a payload in, a running job out.
210type Handler = Arc<dyn Fn(Json) -> BoxFuture<'static, Result<()>> + Send + Sync>;
211
212/// The map from a stored job name back to code that can run it.
213///
214/// Laravel needs nothing like this: it stores the serialised PHP object and
215/// asks the runtime to bring the class back to life. A compiled language cannot
216/// resurrect a type from a string — the type may not even be linked into the
217/// binary that reads the row — so the application states the connection once,
218/// at boot:
219///
220/// ```ignore
221/// let mut jobs = JobRegistry::new();
222/// jobs.register::<SendWelcomeEmail>();
223/// jobs.register::<GenerateInvoice>();
224/// ```
225///
226/// The cost is one line per job. What is bought is that `queue:work` fails to
227/// *compile* against a renamed job type, instead of failing at three in the
228/// morning against a class that is no longer there.
229#[derive(Default)]
230pub struct JobRegistry {
231    handlers: HashMap<String, Handler>,
232}
233
234impl JobRegistry {
235    pub fn new() -> Self {
236        JobRegistry::default()
237    }
238
239    /// Teach the registry about one job type.
240    ///
241    /// Registering the same name twice replaces the handler, so a test can
242    /// substitute a job without rebuilding the registry.
243    pub fn register<J: Job>(&mut self) -> &mut Self {
244        self.register_fn(J::NAME, |payload| {
245            Box::pin(async move {
246                // Rebuilding happens inside the future so a bad payload fails
247                // the job the same way a bad `handle` does, with the same
248                // retries and the same dead-letter entry.
249                let job = J::from_payload(&payload)?;
250                job.handle().await
251            })
252        })
253    }
254
255    /// Register a handler directly, for a name with no Rust type behind it —
256    /// a job pushed by another service, or a stub in a test.
257    pub fn register_fn<F>(&mut self, name: &str, handler: F) -> &mut Self
258    where
259        F: Fn(Json) -> BoxFuture<'static, Result<()>> + Send + Sync + 'static,
260    {
261        self.handlers.insert(name.to_string(), Arc::new(handler));
262        self
263    }
264
265    /// The handler for a name, if one was registered.
266    pub fn handler(&self, name: &str) -> Option<Handler> {
267        self.handlers.get(name).map(Arc::clone)
268    }
269
270    pub fn contains(&self, name: &str) -> bool {
271        self.handlers.contains_key(name)
272    }
273
274    pub fn len(&self) -> usize {
275        self.handlers.len()
276    }
277
278    pub fn is_empty(&self) -> bool {
279        self.handlers.is_empty()
280    }
281
282    /// Every registered name, sorted — for `queue:work`'s boot log and for the
283    /// error message below.
284    pub fn names(&self) -> Vec<String> {
285        let mut names: Vec<String> = self.handlers.keys().cloned().collect();
286        names.sort();
287        names
288    }
289
290    /// Run a stored job, or explain why it cannot be run.
291    ///
292    /// The error names what *is* registered: "no handler for `x`" on its own
293    /// leaves the reader guessing whether they misspelled the job or forgot
294    /// the registration line.
295    pub fn run(&self, name: &str, payload: Json) -> BoxFuture<'static, Result<()>> {
296        match self.handler(name) {
297            Some(handler) => handler(payload),
298            None => {
299                let message = format!(
300                    "no handler is registered for the job `{name}`. Add \
301                     `registry.register::<{name}>()` at boot. Registered: {}",
302                    if self.handlers.is_empty() {
303                        "nothing yet".to_string()
304                    } else {
305                        self.names().join(", ")
306                    }
307                );
308                Box::pin(async move { Err(Error::msg(message)) })
309            }
310        }
311    }
312}
313
314#[cfg(test)]
315mod tests {
316    use super::*;
317    use std::sync::atomic::{AtomicI64, Ordering};
318
319    static ADDED: AtomicI64 = AtomicI64::new(0);
320
321    struct AddToTotal {
322        amount: i64,
323    }
324
325    impl Job for AddToTotal {
326        const NAME: &'static str = "add-to-total";
327
328        fn payload(&self) -> Json {
329            Json::object([("amount", Json::from(self.amount))])
330        }
331
332        fn from_payload(payload: &Json) -> Result<Self> {
333            let amount = payload
334                .get("amount")
335                .and_then(Json::as_i64)
336                .ok_or_else(|| Error::msg("add-to-total needs an `amount`"))?;
337            Ok(AddToTotal { amount })
338        }
339
340        fn handle(&self) -> impl Future<Output = Result<()>> + Send {
341            let amount = self.amount;
342            async move {
343                ADDED.fetch_add(amount, Ordering::SeqCst);
344                Ok(())
345            }
346        }
347    }
348
349    struct Fussy;
350
351    impl Job for Fussy {
352        const NAME: &'static str = "fussy";
353
354        fn payload(&self) -> Json {
355            Json::Null
356        }
357
358        fn from_payload(_payload: &Json) -> Result<Self> {
359            Ok(Fussy)
360        }
361
362        /// Spelled as an `async fn`, which the trait also accepts.
363        async fn handle(&self) -> Result<()> {
364            Ok(())
365        }
366
367        fn tries(&self) -> u32 {
368            7
369        }
370
371        fn retry_after(&self) -> Duration {
372            Duration::from_secs(5)
373        }
374
375        fn queue(&self) -> &'static str {
376            "slow"
377        }
378    }
379
380    #[test]
381    fn a_job_packages_itself_with_its_own_policy() {
382        let queued = Fussy.to_queued();
383
384        assert_eq!(queued.name, "fussy");
385        assert_eq!(queued.queue, "slow");
386        assert_eq!(queued.max_tries, 7);
387        assert_eq!(queued.retry_after, Duration::from_secs(5));
388        assert_eq!(queued.delay, Duration::ZERO);
389    }
390
391    #[test]
392    fn the_defaults_apply_to_a_job_that_says_nothing() {
393        let queued = AddToTotal { amount: 1 }.to_queued();
394
395        assert_eq!(queued.queue, DEFAULT_QUEUE);
396        assert_eq!(queued.max_tries, DEFAULT_TRIES);
397        assert_eq!(queued.retry_after, DEFAULT_RETRY_AFTER);
398    }
399
400    #[test]
401    fn a_job_that_asks_for_zero_tries_still_runs_once() {
402        assert_eq!(QueuedJob::new("x", Json::Null).with_tries(0).max_tries, 1);
403    }
404
405    #[tokio::test]
406    async fn the_registry_runs_a_job_rebuilt_from_its_payload() {
407        ADDED.store(0, Ordering::SeqCst);
408
409        let mut registry = JobRegistry::new();
410        registry.register::<AddToTotal>();
411
412        let queued = AddToTotal { amount: 41 }.to_queued();
413        registry.run(&queued.name, queued.payload).await.unwrap();
414
415        assert_eq!(ADDED.load(Ordering::SeqCst), 41);
416    }
417
418    #[tokio::test]
419    async fn an_unknown_job_name_names_what_is_registered() {
420        let mut registry = JobRegistry::new();
421        registry.register::<AddToTotal>();
422
423        let error = registry.run("not-a-job", Json::Null).await.unwrap_err().to_string();
424
425        assert!(error.contains("no handler is registered for the job `not-a-job`"), "{error}");
426        assert!(error.contains("add-to-total"), "{error}");
427    }
428
429    #[tokio::test]
430    async fn an_unreadable_payload_fails_the_job_rather_than_the_worker() {
431        let mut registry = JobRegistry::new();
432        registry.register::<AddToTotal>();
433
434        let error = registry.run("add-to-total", Json::Null).await.unwrap_err().to_string();
435        assert!(error.contains("needs an `amount`"), "{error}");
436    }
437
438    #[tokio::test]
439    async fn a_handler_can_be_registered_without_a_type_behind_it() {
440        let mut registry = JobRegistry::new();
441        registry.register_fn("external", |payload| {
442            Box::pin(async move {
443                assert_eq!(payload.get("from").and_then(Json::as_str), Some("php"));
444                Ok(())
445            })
446        });
447
448        assert!(registry.contains("external"));
449        registry
450            .run("external", Json::object([("from", Json::from("php"))]))
451            .await
452            .unwrap();
453    }
454
455    #[test]
456    fn registering_twice_replaces_rather_than_duplicates() {
457        let mut registry = JobRegistry::new();
458        registry.register::<AddToTotal>();
459        registry.register::<AddToTotal>();
460
461        assert_eq!(registry.len(), 1);
462        assert_eq!(registry.names(), vec!["add-to-total"]);
463    }
464
465    #[test]
466    fn a_reserved_job_knows_when_it_is_out_of_attempts() {
467        let job = QueuedJob::new("x", Json::Null).with_tries(3);
468        let reserve = |attempts| ReservedJob { id: "1".into(), job: job.clone(), attempts };
469
470        assert!(!reserve(2).is_last_attempt());
471        assert!(reserve(3).is_last_attempt());
472        assert!(reserve(4).is_last_attempt());
473    }
474
475    #[test]
476    fn a_failed_job_renders_for_a_dashboard() {
477        let failed = FailedJob {
478            id: "9".into(),
479            name: "add-to-total".into(),
480            queue: "default".into(),
481            payload: Json::object([("amount", Json::from(2))]),
482            attempts: 3,
483            error: "boom".into(),
484            failed_at: 1_700_000_000,
485        };
486
487        assert_eq!(
488            failed.to_json().to_string(),
489            r#"{"attempts":3,"error":"boom","failed_at":1700000000,"id":"9","name":"add-to-total","payload":{"amount":2},"queue":"default"}"#
490        );
491    }
492}
rustlavel_queue/job.rs

rustlavel_queue/
job.rs
