laterite_core/queue/
mod.rs

1//! The queue: work pushed now and run by a worker, with retries, a failed
2//! list and a lifetime per job.
3//!
4//! A job is a serde struct that implements [`Job`]; a module registers it
5//! by type, and a payload names it. Pushing records the job; a worker claims
6//! it, runs it under its timeout, deletes it on success, releases it with
7//! its backoff on failure, and moves it to the failed list once its tries or
8//! its retry window are spent.
9//!
10//! ```ignore
11//! use laterite_core::queue::{Job, JobCx, JobError, QueueRegistry};
12//!
13//! #[derive(Serialize, Deserialize)]
14//! struct SendDigest { user_id: i64 }
15//!
16//! #[async_trait]
17//! impl Job for SendDigest {
18//!     const NAME: &'static str = "acme.blog::send_digest";
19//!     async fn run(&self, cx: &JobCx) -> Result<(), JobError> {
20//!         digest::send(cx.db(), self.user_id).await?;
21//!         Ok(())
22//!     }
23//! }
24//!
25//! fn register(&self, registry: &mut Registry) {
26//!     registry.add_job::<SendDigest>();
27//! }
28//!
29//! ctx.queue().push(&SendDigest { user_id: 7 }).await?;
30//! ```
31//!
32//! Jobs live in a [`QueueDriver`]: the database by default, or Redis with
33//! the `redis` store. `queue.mode = inline` runs a job as it is pushed, for
34//! tests and for an application without a worker.
35
36use std::collections::HashMap;
37use std::future::Future;
38use std::pin::Pin;
39use std::sync::Arc;
40use std::time::Duration;
41
42use async_trait::async_trait;
43use chrono::{DateTime, SecondsFormat, Utc};
44use serde::de::DeserializeOwned;
45use serde::{Deserialize, Serialize};
46
47use crate::cache::Cache;
48use crate::config::QueueConfig;
49use crate::error::{CoreError, CoreResult};
50use crate::events::{Event, Events};
51use crate::parameters::Parameters;
52use crate::registry::Registry;
53use crate::{Db, Module, ModuleId};
54
55mod database;
56pub mod migrations;
57#[cfg(feature = "redis")]
58pub mod redis;
59mod worker;
60
61pub use database::DatabaseQueue;
62pub use worker::{WorkOptions, Worker};
63
64/// The module that owns the `jobs` and `failed_jobs` tables, listed among
65/// an application's built-ins.
66pub struct QueueModule;
67
68impl Module for QueueModule {
69    fn id(&self) -> ModuleId {
70        ModuleId::new(migrations::MODULE_ID)
71    }
72    fn migrations(&self) -> crate::MigrationSet {
73        migrations::migrations()
74    }
75}
76
77/// The queue a job lands on when it names none.
78pub const DEFAULT_QUEUE: &str = "default";
79
80/// Why a run did not succeed.
81#[derive(Debug, Clone, PartialEq, Eq)]
82pub enum JobError {
83    /// Try again after the backoff, while tries and the retry window allow.
84    Retry(String),
85    /// Move to the failed list now, whatever tries remain.
86    Fail(String),
87}
88
89impl JobError {
90    /// A failure worth another attempt.
91    pub fn retry(message: impl std::fmt::Display) -> Self {
92        Self::Retry(message.to_string())
93    }
94
95    /// A failure no attempt would mend.
96    pub fn fail(message: impl std::fmt::Display) -> Self {
97        Self::Fail(message.to_string())
98    }
99
100    pub fn message(&self) -> &str {
101        match self {
102            Self::Retry(m) | Self::Fail(m) => m,
103        }
104    }
105}
106
107impl std::fmt::Display for JobError {
108    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
109        f.write_str(self.message())
110    }
111}
112
113impl std::error::Error for JobError {}
114
115impl From<CoreError> for JobError {
116    fn from(e: CoreError) -> Self {
117        Self::Retry(e.to_string())
118    }
119}
120
121impl From<sqlx::Error> for JobError {
122    fn from(e: sqlx::Error) -> Self {
123        Self::Retry(e.to_string())
124    }
125}
126
127/// What a running job sees: the application's services and its own attempt.
128#[derive(Clone)]
129pub struct JobCx {
130    db: Db,
131    events: Events,
132    cache: Cache,
133    parameters: Parameters,
134    id: Option<i64>,
135    attempt: u32,
136    queue: String,
137}
138
139impl JobCx {
140    pub fn db(&self) -> &Db {
141        &self.db
142    }
143    pub fn events(&self) -> &Events {
144        &self.events
145    }
146    pub fn cache(&self) -> &Cache {
147        &self.cache
148    }
149    pub fn parameters(&self) -> &Parameters {
150        &self.parameters
151    }
152    /// The job's row id; `None` when it runs inline.
153    pub fn id(&self) -> Option<i64> {
154        self.id
155    }
156    /// Which attempt this is, from 1.
157    pub fn attempt(&self) -> u32 {
158        self.attempt
159    }
160    pub fn queue(&self) -> &str {
161        &self.queue
162    }
163}
164
165/// Work to run later. The struct is the payload: whatever it carries is
166/// stored as JSON and read back by the worker.
167#[async_trait]
168pub trait Job: Serialize + DeserializeOwned + Send + Sync + 'static {
169    /// The job's name, unique across the application: `acme.blog::send_digest`.
170    const NAME: &'static str;
171
172    async fn run(&self, cx: &JobCx) -> Result<(), JobError>;
173
174    /// The queue the job lands on. [`DEFAULT_QUEUE`] unless overridden.
175    fn queue(&self) -> String {
176        DEFAULT_QUEUE.to_string()
177    }
178
179    /// How many attempts before the job fails; `0` means as many as the
180    /// retry window allows. Three unless overridden.
181    fn tries(&self) -> u32 {
182        3
183    }
184
185    /// How long to wait before attempt `attempt + 1`, after attempt
186    /// `attempt` failed. Ten seconds, doubling each time, unless overridden.
187    fn backoff(&self, attempt: u32) -> Duration {
188        Duration::from_secs(10u64 << attempt.saturating_sub(1).min(8))
189    }
190
191    /// How long one attempt may run. A minute unless overridden; `None`
192    /// means no limit.
193    fn timeout(&self) -> Option<Duration> {
194        Some(Duration::from_secs(60))
195    }
196
197    /// Keep retrying until this moment, whatever `tries` says. `None`
198    /// unless overridden.
199    fn retry_until(&self) -> Option<DateTime<Utc>> {
200        None
201    }
202
203    /// A key no two pending jobs may share: a push while one with the same
204    /// key waits is dropped. `None` unless overridden.
205    fn unique_key(&self) -> Option<String> {
206        None
207    }
208}
209
210/// What the worker reads from a payload before running it.
211#[derive(Debug, Clone)]
212pub struct JobMeta {
213    pub tries: u32,
214    pub timeout: Option<Duration>,
215    pub retry_until: Option<DateTime<Utc>>,
216}
217
218type RunFn = dyn Fn(serde_json::Value, JobCx) -> Pin<Box<dyn Future<Output = Result<(), JobError>> + Send>>
219    + Send
220    + Sync;
221type MetaFn = dyn Fn(&serde_json::Value) -> Result<JobMeta, String> + Send + Sync;
222type BackoffFn = dyn Fn(&serde_json::Value, u32) -> Duration + Send + Sync;
223
224/// A registered job: its name and how to run a payload of it. Built by
225/// [`JobReg::of`] and contributed through [`QueueRegistry::add_job`].
226#[derive(Clone)]
227pub struct JobReg {
228    name: &'static str,
229    run: Arc<RunFn>,
230    meta: Arc<MetaFn>,
231    backoff: Arc<BackoffFn>,
232}
233
234impl std::fmt::Debug for JobReg {
235    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
236        f.debug_struct("JobReg").field("name", &self.name).finish()
237    }
238}
239
240impl JobReg {
241    /// The registration for `T`.
242    pub fn of<T: Job>() -> Self {
243        Self {
244            name: T::NAME,
245            run: Arc::new(|payload, cx| {
246                Box::pin(async move {
247                    let job: T = serde_json::from_value(payload)
248                        .map_err(|e| JobError::Fail(format!("payload does not decode: {e}")))?;
249                    job.run(&cx).await
250                })
251            }),
252            meta: Arc::new(|payload| {
253                let job: T = serde_json::from_value(payload.clone())
254                    .map_err(|e| format!("payload does not decode: {e}"))?;
255                Ok(JobMeta {
256                    tries: job.tries(),
257                    timeout: job.timeout(),
258                    retry_until: job.retry_until(),
259                })
260            }),
261            backoff: Arc::new(|payload, attempt| {
262                serde_json::from_value::<T>(payload.clone())
263                    .map(|job| job.backoff(attempt))
264                    .unwrap_or(Duration::from_secs(10))
265            }),
266        }
267    }
268
269    pub fn name(&self) -> &'static str {
270        self.name
271    }
272}
273
274/// `registry.add_job::<T>()`: the typed way to contribute a job.
275pub trait QueueRegistry {
276    fn add_job<T: Job>(&mut self);
277}
278
279impl QueueRegistry for Registry {
280    fn add_job<T: Job>(&mut self) {
281        self.add(JobReg::of::<T>());
282    }
283}
284
285/// A job as it waits or runs.
286#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
287pub struct Queued {
288    pub id: i64,
289    pub queue: String,
290    pub name: String,
291    pub payload: serde_json::Value,
292    /// How many times a worker has claimed it, this claim included.
293    pub attempts: u32,
294    pub available_at: String,
295    pub reserved_at: Option<String>,
296    pub reserved_by: Option<String>,
297    pub created_at: String,
298}
299
300/// A job on the failed list.
301#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
302pub struct FailedJob {
303    pub id: i64,
304    pub queue: String,
305    pub name: String,
306    pub payload: serde_json::Value,
307    pub error: String,
308    pub failed_at: String,
309}
310
311/// How many jobs a queue holds.
312#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
313pub struct QueueCount {
314    pub queue: String,
315    /// Waiting, including those released for a later attempt.
316    pub pending: u64,
317    /// Claimed by a worker now.
318    pub reserved: u64,
319}
320
321/// A pushed job's fate.
322#[derive(Debug, Clone, Copy, PartialEq, Eq)]
323pub enum Pushed {
324    /// Stored under this id.
325    Queued(i64),
326    /// Ran at once (`queue.mode = inline`).
327    Ran,
328    /// Dropped: a job with the same unique key is already waiting.
329    Duplicate,
330}
331
332/// Where jobs wait. Every method is one storage operation; the worker's
333/// rules live in [`Worker`].
334#[async_trait]
335pub trait QueueDriver: Send + Sync {
336    fn name(&self) -> &'static str;
337    /// Stores a job; `None` when `unique_key` names one already waiting.
338    async fn push(
339        &self,
340        queue: &str,
341        name: &str,
342        payload: &serde_json::Value,
343        available_at: DateTime<Utc>,
344        unique_key: Option<&str>,
345    ) -> CoreResult<Option<i64>>;
346    /// Claims the oldest job due on one of `queues`, or one whose claim is
347    /// older than `retry_after`, for `worker`. Its `attempts` count this claim.
348    async fn claim(
349        &self,
350        queues: &[String],
351        retry_after: Duration,
352        worker: &str,
353    ) -> CoreResult<Option<Queued>>;
354    /// Puts a claimed job back, due at `available_at`.
355    async fn release(&self, id: i64, available_at: DateTime<Utc>) -> CoreResult<()>;
356    /// Removes a finished job.
357    async fn delete(&self, id: i64) -> CoreResult<()>;
358    /// Moves a job to the failed list; returns the failed row's id.
359    async fn fail(&self, job: &Queued, error: &str) -> CoreResult<i64>;
360    async fn counts(&self) -> CoreResult<Vec<QueueCount>>;
361    async fn failed(&self) -> CoreResult<Vec<FailedJob>>;
362    /// Puts a failed job (or every one, with `None`) back on its queue;
363    /// returns how many moved.
364    async fn retry(&self, id: Option<i64>) -> CoreResult<usize>;
365    /// Drops a failed job (or every one, with `None`); returns how many went.
366    async fn forget(&self, id: Option<i64>) -> CoreResult<usize>;
367    /// Drops every waiting job on `queue`, or on every queue with `None`.
368    async fn clear(&self, queue: Option<&str>) -> CoreResult<usize>;
369    /// Checks the store answers; a driver with nothing to check says yes.
370    async fn ping(&self) -> CoreResult<()> {
371        Ok(())
372    }
373}
374
375/// How the queue runs.
376#[derive(Debug, Clone, Copy, PartialEq, Eq)]
377pub enum Mode {
378    /// Workers in this process take the jobs.
379    Worker,
380    /// Another process (`queue work`) takes the jobs.
381    External,
382    /// A job runs as it is pushed; nothing is stored.
383    Inline,
384}
385
386impl Mode {
387    pub fn parse(s: &str) -> CoreResult<Self> {
388        match s {
389            "worker" => Ok(Self::Worker),
390            "external" => Ok(Self::External),
391            "inline" => Ok(Self::Inline),
392            other => Err(CoreError::Config(format!(
393                "queue.mode `{other}` is not one of worker, external, inline"
394            ))),
395        }
396    }
397}
398
399/// The services a job runs against.
400#[derive(Clone)]
401pub struct Services {
402    pub db: Db,
403    pub events: Events,
404    pub cache: Cache,
405    pub parameters: Parameters,
406}
407
408struct Inner {
409    driver: Box<dyn QueueDriver>,
410    jobs: HashMap<&'static str, JobReg>,
411    services: Services,
412    mode: Mode,
413    queues: Vec<String>,
414    retry_after: Duration,
415    poll: Duration,
416    workers: usize,
417    /// Pushes wake an idle worker in this process.
418    wake: tokio::sync::Notify,
419}
420
421/// The queue handle. Cheap to clone: every clone is the same queue.
422#[derive(Clone)]
423pub struct Queue {
424    inner: Arc<Inner>,
425}
426
427impl std::fmt::Debug for Queue {
428    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
429        f.debug_struct("Queue")
430            .field("driver", &self.inner.driver.name())
431            .field("mode", &self.inner.mode)
432            .field("jobs", &self.inner.jobs.len())
433            .finish_non_exhaustive()
434    }
435}
436
437/// What [`Queue::new`] takes beside the driver and the jobs.
438#[derive(Debug, Clone)]
439pub struct Options {
440    pub mode: Mode,
441    /// The queues a worker takes from, in priority order.
442    pub queues: Vec<String>,
443    /// After this long a claimed job counts as abandoned and is claimed again.
444    pub retry_after: Duration,
445    /// How long an idle worker waits before looking again.
446    pub poll: Duration,
447    /// How many workers [`Queue::spawn_workers`] starts.
448    pub workers: usize,
449}
450
451impl Options {
452    /// The options `[queue]` configures.
453    pub fn from_config(config: &QueueConfig) -> CoreResult<Self> {
454        let mut queues = config.queues.clone();
455        if queues.is_empty() {
456            queues.push(DEFAULT_QUEUE.to_string());
457        }
458        Ok(Self {
459            mode: Mode::parse(&config.mode)?,
460            queues,
461            retry_after: Duration::from_secs(config.retry_after_seconds.max(1)),
462            poll: Duration::from_secs(config.poll_seconds.max(1)),
463            workers: config.workers,
464        })
465    }
466}
467
468impl Default for Options {
469    fn default() -> Self {
470        Self {
471            mode: Mode::Worker,
472            queues: vec![DEFAULT_QUEUE.to_string()],
473            retry_after: Duration::from_secs(90),
474            poll: Duration::from_secs(1),
475            workers: 2,
476        }
477    }
478}
479
480impl Queue {
481    /// A queue over `driver`, knowing `jobs`.
482    pub fn new(
483        driver: impl QueueDriver + 'static,
484        jobs: impl IntoIterator<Item = JobReg>,
485        services: Services,
486        options: Options,
487    ) -> Self {
488        let jobs = jobs.into_iter().map(|j| (j.name, j)).collect();
489        Self {
490            inner: Arc::new(Inner {
491                driver: Box::new(driver),
492                jobs,
493                services,
494                mode: options.mode,
495                queues: options.queues,
496                retry_after: options.retry_after,
497                poll: options.poll,
498                workers: options.workers,
499                wake: tokio::sync::Notify::new(),
500            }),
501        }
502    }
503
504    /// A queue that runs every job as it is pushed, over no storage: for
505    /// tests and tools.
506    pub fn inline(jobs: impl IntoIterator<Item = JobReg>, services: Services) -> Self {
507        Self::new(
508            NoQueue,
509            jobs,
510            services,
511            Options {
512                mode: Mode::Inline,
513                ..Options::default()
514            },
515        )
516    }
517
518    pub fn mode(&self) -> Mode {
519        self.inner.mode
520    }
521
522    pub fn driver_name(&self) -> &'static str {
523        self.inner.driver.name()
524    }
525
526    /// The names of every registered job, sorted.
527    pub fn job_names(&self) -> Vec<&'static str> {
528        let mut names: Vec<&'static str> = self.inner.jobs.keys().copied().collect();
529        names.sort_unstable();
530        names
531    }
532
533    pub(crate) fn job(&self, name: &str) -> Option<&JobReg> {
534        self.inner.jobs.get(name)
535    }
536
537    pub(crate) fn services(&self) -> &Services {
538        &self.inner.services
539    }
540
541    pub(crate) fn driver(&self) -> &dyn QueueDriver {
542        self.inner.driver.as_ref()
543    }
544
545    pub(crate) fn queues(&self) -> &[String] {
546        &self.inner.queues
547    }
548
549    pub(crate) fn retry_after(&self) -> Duration {
550        self.inner.retry_after
551    }
552
553    pub(crate) fn poll(&self) -> Duration {
554        self.inner.poll
555    }
556
557    pub(crate) fn wake(&self) -> &tokio::sync::Notify {
558        &self.inner.wake
559    }
560
561    fn check<T: Job>(&self) -> CoreResult<()> {
562        if self.inner.jobs.contains_key(T::NAME) {
563            Ok(())
564        } else {
565            Err(CoreError::Validation(format!(
566                "job `{}` is not registered; add `registry.add_job::<{}>()` to a module",
567                T::NAME,
568                std::any::type_name::<T>()
569            )))
570        }
571    }
572
573    /// Pushes `job`, due now.
574    pub async fn push<T: Job>(&self, job: &T) -> CoreResult<Pushed> {
575        self.push_at(job, Utc::now()).await
576    }
577
578    /// Pushes `job`, due after `delay`.
579    pub async fn push_in<T: Job>(&self, job: &T, delay: Duration) -> CoreResult<Pushed> {
580        let delay = chrono::Duration::from_std(delay).unwrap_or(chrono::Duration::MAX);
581        self.push_at(job, Utc::now() + delay).await
582    }
583
584    /// Pushes `job`, due at `when`.
585    pub async fn push_at<T: Job>(&self, job: &T, when: DateTime<Utc>) -> CoreResult<Pushed> {
586        self.check::<T>()?;
587        if self.inner.mode == Mode::Inline {
588            let cx = self.context(None, 1, &job.queue());
589            return match job.run(&cx).await {
590                Ok(()) => Ok(Pushed::Ran),
591                Err(e) => Err(CoreError::Internal(format!(
592                    "job `{}` failed: {e}",
593                    T::NAME
594                ))),
595            };
596        }
597        let payload = serde_json::to_value(job)
598            .map_err(|e| CoreError::Internal(format!("job `{}` cannot serialize: {e}", T::NAME)))?;
599        let unique = job.unique_key();
600        let stored = self
601            .inner
602            .driver
603            .push(&job.queue(), T::NAME, &payload, when, unique.as_deref())
604            .await?;
605        match stored {
606            Some(id) => {
607                self.inner.wake.notify_waiters();
608                tracing::debug!(job = T::NAME, id, "job queued");
609                Ok(Pushed::Queued(id))
610            }
611            None => Ok(Pushed::Duplicate),
612        }
613    }
614
615    /// Pushes a job by its registered name with a payload already encoded,
616    /// due now: what a schedule entry does. Refused when no job has the name.
617    pub async fn push_named(
618        &self,
619        name: &str,
620        payload: &serde_json::Value,
621        queue: &str,
622    ) -> CoreResult<Pushed> {
623        let Some(reg) = self.inner.jobs.get(name) else {
624            return Err(CoreError::Validation(format!(
625                "job `{name}` is not registered"
626            )));
627        };
628        if self.inner.mode == Mode::Inline {
629            let cx = self.context(None, 1, queue);
630            return match (reg.run)(payload.clone(), cx).await {
631                Ok(()) => Ok(Pushed::Ran),
632                Err(e) => Err(CoreError::Internal(format!("job `{name}` failed: {e}"))),
633            };
634        }
635        let stored = self
636            .inner
637            .driver
638            .push(queue, name, payload, Utc::now(), None)
639            .await?;
640        match stored {
641            Some(id) => {
642                self.inner.wake.notify_waiters();
643                Ok(Pushed::Queued(id))
644            }
645            None => Ok(Pushed::Duplicate),
646        }
647    }
648
649    pub(crate) fn context(&self, id: Option<i64>, attempt: u32, queue: &str) -> JobCx {
650        let s = &self.inner.services;
651        JobCx {
652            db: s.db.clone(),
653            events: s.events.clone(),
654            cache: s.cache.clone(),
655            parameters: s.parameters.clone(),
656            id,
657            attempt,
658            queue: queue.to_string(),
659        }
660    }
661
662    /// How many jobs each queue holds.
663    pub async fn counts(&self) -> CoreResult<Vec<QueueCount>> {
664        self.inner.driver.counts().await
665    }
666
667    /// The failed list, newest first.
668    pub async fn failed(&self) -> CoreResult<Vec<FailedJob>> {
669        self.inner.driver.failed().await
670    }
671
672    /// Puts one failed job, or every one, back on its queue.
673    pub async fn retry(&self, id: Option<i64>) -> CoreResult<usize> {
674        let moved = self.inner.driver.retry(id).await?;
675        if moved > 0 {
676            self.inner.wake.notify_waiters();
677        }
678        Ok(moved)
679    }
680
681    /// Drops one failed job, or every one.
682    pub async fn forget(&self, id: Option<i64>) -> CoreResult<usize> {
683        self.inner.driver.forget(id).await
684    }
685
686    /// Drops every waiting job on `queue`, or everywhere.
687    pub async fn clear(&self, queue: Option<&str>) -> CoreResult<usize> {
688        self.inner.driver.clear(queue).await
689    }
690
691    /// Runs jobs on this task until `stop` says so, or as `options` end it.
692    /// Returns how many jobs it processed.
693    pub async fn work(
694        &self,
695        options: WorkOptions,
696        stop: tokio::sync::watch::Receiver<bool>,
697    ) -> CoreResult<usize> {
698        Worker::new(self.clone(), options).run(stop).await
699    }
700
701    /// Starts the configured number of workers, each taking from the
702    /// configured queues, until `stop` says so.
703    pub fn spawn_workers(
704        &self,
705        stop: tokio::sync::watch::Receiver<bool>,
706    ) -> Vec<tokio::task::JoinHandle<()>> {
707        (0..self.inner.workers)
708            .map(|n| {
709                let queue = self.clone();
710                let stop = stop.clone();
711                tokio::spawn(async move {
712                    let options = WorkOptions::default().name(format!("worker-{n}"));
713                    if let Err(e) = queue.work(options, stop).await {
714                        tracing::error!(worker = n, error = %e, "worker stopped");
715                    }
716                })
717            })
718            .collect()
719    }
720}
721
722/// A driver holding nothing, behind [`Queue::inline`].
723struct NoQueue;
724
725#[async_trait]
726impl QueueDriver for NoQueue {
727    fn name(&self) -> &'static str {
728        "inline"
729    }
730    async fn push(
731        &self,
732        _queue: &str,
733        _name: &str,
734        _payload: &serde_json::Value,
735        _available_at: DateTime<Utc>,
736        _unique_key: Option<&str>,
737    ) -> CoreResult<Option<i64>> {
738        Ok(None)
739    }
740    async fn claim(
741        &self,
742        _queues: &[String],
743        _retry_after: Duration,
744        _worker: &str,
745    ) -> CoreResult<Option<Queued>> {
746        Ok(None)
747    }
748    async fn release(&self, _id: i64, _available_at: DateTime<Utc>) -> CoreResult<()> {
749        Ok(())
750    }
751    async fn delete(&self, _id: i64) -> CoreResult<()> {
752        Ok(())
753    }
754    async fn fail(&self, _job: &Queued, _error: &str) -> CoreResult<i64> {
755        Ok(0)
756    }
757    async fn counts(&self) -> CoreResult<Vec<QueueCount>> {
758        Ok(Vec::new())
759    }
760    async fn failed(&self) -> CoreResult<Vec<FailedJob>> {
761        Ok(Vec::new())
762    }
763    async fn retry(&self, _id: Option<i64>) -> CoreResult<usize> {
764        Ok(0)
765    }
766    async fn forget(&self, _id: Option<i64>) -> CoreResult<usize> {
767        Ok(0)
768    }
769    async fn clear(&self, _queue: Option<&str>) -> CoreResult<usize> {
770        Ok(0)
771    }
772}
773
774/// Announced when a job reaches the failed list.
775#[derive(Debug, Clone, Serialize, Deserialize)]
776pub struct JobFailed {
777    pub id: i64,
778    pub name: String,
779    pub queue: String,
780    pub attempts: u32,
781    pub error: String,
782}
783
784impl Event for JobFailed {
785    const NAME: &'static str = "queue.job_failed";
786}
787
788/// The queue a deployment configured: the database by default, Redis for
789/// the `redis` store.
790pub async fn from_config(
791    config: &QueueConfig,
792    jobs: impl IntoIterator<Item = JobReg>,
793    services: Services,
794) -> CoreResult<Queue> {
795    let options = Options::from_config(config)?;
796    match config.store.as_str() {
797        "auto" | "database" => Ok(Queue::new(
798            DatabaseQueue::new(services.db.clone()),
799            jobs,
800            services,
801            options,
802        )),
803        #[cfg(feature = "redis")]
804        "redis" => {
805            let url = config.url.as_deref().ok_or_else(|| {
806                CoreError::Config("queue.store = \"redis\" needs queue.url".to_string())
807            })?;
808            let driver = redis::connect(url, &config.prefix).await?;
809            driver.ping().await?;
810            Ok(Queue::new(driver, jobs, services, options))
811        }
812        #[cfg(not(feature = "redis"))]
813        "redis" => Err(CoreError::Config(
814            "queue.store = \"redis\" needs the `redis` feature of laterite-core".to_string(),
815        )),
816        other => Err(CoreError::Config(format!(
817            "queue.store `{other}` is not one of auto, database, redis"
818        ))),
819    }
820}
821
822pub fn time_text(at: DateTime<Utc>) -> String {
823    at.to_rfc3339_opts(SecondsFormat::Micros, true)
824}
825
826#[cfg(test)]
827mod tests {
828    use super::*;
829
830    #[test]
831    fn the_default_backoff_doubles_from_ten_seconds() {
832        #[derive(Serialize, Deserialize)]
833        struct Noop;
834        #[async_trait]
835        impl Job for Noop {
836            const NAME: &'static str = "test::noop";
837            async fn run(&self, _cx: &JobCx) -> Result<(), JobError> {
838                Ok(())
839            }
840        }
841        assert_eq!(Noop.backoff(1), Duration::from_secs(10));
842        assert_eq!(Noop.backoff(2), Duration::from_secs(20));
843        assert_eq!(Noop.backoff(4), Duration::from_secs(80));
844        assert_eq!(Noop.backoff(40), Duration::from_secs(2560));
845        assert_eq!(Noop.tries(), 3);
846    }
847
848    #[test]
849    fn a_mode_parses_or_is_refused() {
850        assert_eq!(Mode::parse("inline").unwrap(), Mode::Inline);
851        assert!(Mode::parse("sometimes").is_err());
852    }
853}
laterite_core/queue/mod.rs

laterite_core/queue/
mod.rs
