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

rustlavel_queue/
job.rs
