rustlavel_queue/
queue.rs

1//! The [`Queue`] contract every driver implements.
2//!
3//! Dyn-compatible on purpose: an application picks its driver at boot and the
4//! rest of the framework holds an `Arc<dyn Queue>` without knowing which one it
5//! got. That is why every method returns a [`BoxFuture`] rather than being an
6//! `async fn` — `async fn` in a trait is not dyn-compatible — and why the trait
7//! speaks in [`QueuedJob`] rather than in a generic `J: Job`.
8//!
9//! The generic conveniences that cannot be dyn-compatible (`dispatch`, which
10//! takes a concrete job type) live in [`QueueExt`], blanket-implemented for
11//! every `Queue` including `dyn Queue`, so callers never notice the split.
12
13use crate::job::{BoxFuture, FailedJob, Job, QueuedJob, ReservedJob};
14use rustlavel_core::events::{self, Event};
15use rustlavel_core::{Json, Result};
16use std::time::Duration;
17
18/// A queue backend.
19pub trait Queue: Send + Sync + 'static {
20    /// The driver's name. It appears in `queue.*` events and error messages,
21    /// which is how Telescope shows which backend handled a job.
22    fn driver(&self) -> &'static str;
23
24    /// Store a job. Returns the driver's identifier for it.
25    ///
26    /// The envelope carries its own queue and delay, so this one primitive is
27    /// all a driver has to implement.
28    fn push(&self, job: QueuedJob) -> BoxFuture<'_, Result<String>>;
29
30    /// Store a job on a named queue, whatever the job itself asked for.
31    fn push_on<'a>(&'a self, queue: &'a str, job: QueuedJob) -> BoxFuture<'a, Result<String>> {
32        self.push(job.on_queue(queue))
33    }
34
35    /// Store a job that only becomes visible to a worker after `delay`.
36    fn later<'a>(&'a self, delay: Duration, job: QueuedJob) -> BoxFuture<'a, Result<String>> {
37        self.push(job.with_delay(delay))
38    }
39
40    /// Reserve the next available job, or `None` when there is nothing ready.
41    ///
42    /// A delayed job whose time has not come is not "available"; neither is one
43    /// another worker already holds. Reserving must be atomic across workers —
44    /// see the database driver for how that is done and why.
45    fn pop<'a>(&'a self, queue: &'a str) -> BoxFuture<'a, Result<Option<ReservedJob>>>;
46
47    /// How many jobs are waiting on a queue, including delayed ones.
48    fn size<'a>(&'a self, queue: &'a str) -> BoxFuture<'a, Result<u64>>;
49
50    /// Finish with a reserved job: it succeeded and is gone.
51    fn delete<'a>(&'a self, job: &'a ReservedJob) -> BoxFuture<'a, Result<()>>;
52
53    /// Put a reserved job back, available again after `delay`.
54    ///
55    /// The attempt count is kept, which is what makes a retry a retry rather
56    /// than a fresh job that will fail forever.
57    fn release<'a>(&'a self, job: &'a ReservedJob, delay: Duration) -> BoxFuture<'a, Result<()>>;
58
59    /// Move a reserved job to the dead-letter store with the error that killed
60    /// it. It must no longer be on the queue afterwards.
61    fn fail<'a>(&'a self, job: &'a ReservedJob, error: &'a str) -> BoxFuture<'a, Result<()>>;
62
63    /// Everything in the dead-letter store, oldest first.
64    fn failed_jobs(&self) -> BoxFuture<'_, Result<Vec<FailedJob>>>;
65
66    /// Throw away every job waiting on a queue. Returns how many went.
67    fn clear<'a>(&'a self, queue: &'a str) -> BoxFuture<'a, Result<u64>>;
68}
69
70/// The conveniences that take a concrete job type, and so cannot live on a
71/// dyn-compatible trait. Blanket-implemented, including for `dyn Queue`.
72pub trait QueueExt: Queue {
73    /// `dispatch(SendWelcomeEmail { user_id })` — the everyday call.
74    fn dispatch<'a, J: Job>(&'a self, job: &'a J) -> BoxFuture<'a, Result<String>> {
75        self.push(job.to_queued())
76    }
77
78    /// Dispatch onto a named queue instead of the job's own.
79    fn dispatch_on<'a, J: Job>(
80        &'a self,
81        queue: &'a str,
82        job: &'a J,
83    ) -> BoxFuture<'a, Result<String>> {
84        self.push_on(queue, job.to_queued())
85    }
86
87    /// Dispatch, but not before `delay` has passed.
88    fn dispatch_later<'a, J: Job>(
89        &'a self,
90        delay: Duration,
91        job: &'a J,
92    ) -> BoxFuture<'a, Result<String>> {
93        self.later(delay, job.to_queued())
94    }
95}
96
97impl<T: Queue + ?Sized> QueueExt for T {}
98
99/// Report a push on the instrumentation bus.
100///
101/// Guarded by `has_subscribers` so an application with no Telescope never pays
102/// for building an event nobody reads. Drivers call this from `push`; the
103/// worker reports `queue.processed` and `queue.failed` itself.
104pub(crate) fn record_pushed(driver: &'static str, job: &QueuedJob, id: &str) {
105    if !events::has_subscribers() {
106        return;
107    }
108
109    Event::new("queue.pushed")
110        .with("job", job.name.as_str())
111        .with("queue", job.queue.as_str())
112        .with("id", id)
113        .with("driver", driver)
114        .with("delay_seconds", job.delay.as_secs())
115        .dispatch();
116}
117
118/// The JSON body a dashboard or `queue:status` shows.
119pub async fn stats(queue: &dyn Queue, names: &[String]) -> Result<Json> {
120    let mut sizes = Vec::new();
121    for name in names {
122        sizes.push((name.clone(), Json::from(queue.size(name).await?)));
123    }
124
125    let failed = queue.failed_jobs().await?;
126
127    Ok(Json::object([
128        ("driver", Json::from(queue.driver())),
129        ("queues", Json::Object(sizes.into_iter().collect())),
130        ("failed", Json::Array(failed.iter().map(FailedJob::to_json).collect())),
131    ]))
132}
rustlavel_queue/queue.rs

rustlavel_queue/
queue.rs
