rustlavel_queue/
lib.rs

1//! rustlavel-queue: background jobs and scheduling.
2//!
3//! Work that should not happen while someone is waiting for a response —
4//! sending mail, resizing an image, calling a slow API — goes on a queue and is
5//! run by a worker. Work that should happen at a particular time goes on a
6//! schedule.
7//!
8//! ```ignore
9//! // Define a job.
10//! struct SendWelcomeEmail { user_id: i64 }
11//!
12//! impl Job for SendWelcomeEmail {
13//!     const NAME: &'static str = "send-welcome-email";
14//!
15//!     fn payload(&self) -> Json {
16//!         Json::object([("user_id", Json::from(self.user_id))])
17//!     }
18//!
19//!     fn from_payload(payload: &Json) -> Result<Self> {
20//!         Ok(SendWelcomeEmail {
21//!             user_id: payload.get("user_id").and_then(Json::as_i64).unwrap_or_default(),
22//!         })
23//!     }
24//!
25//!     fn handle(&self) -> impl Future<Output = Result<()>> + Send {
26//!         async move { Ok(()) }
27//!     }
28//! }
29//!
30//! // Dispatch it from a handler.
31//! queue.dispatch(&SendWelcomeEmail { user_id }).await?;
32//!
33//! // Run it, in `queue:work`.
34//! let mut jobs = JobRegistry::new();
35//! jobs.register::<SendWelcomeEmail>();
36//!
37//! let shutdown = Shutdown::new();
38//! shutdown.on_ctrl_c();
39//! run_pool(Worker::new(queue, Arc::new(jobs)), 4, shutdown).await?;
40//! ```
41//!
42//! ## The one thing that is not like Laravel
43//!
44//! Laravel serialises the job object and asks PHP to bring the class back. A
45//! compiled language cannot resurrect a type from a string, so the way back
46//! from a stored name to running code is stated once, at boot, in a
47//! [`JobRegistry`]. That is the only line of ceremony the design adds, and it
48//! buys a `queue:work` that fails to compile against a renamed job rather than
49//! failing at three in the morning against a class that is no longer there.
50//!
51//! ## What is in here
52//!
53//! * [`Job`] and [`JobRegistry`] — defining work and finding it again.
54//! * [`Queue`] and [`QueueExt`] — the driver contract and its conveniences.
55//! * [`MemoryQueue`], [`DatabaseQueue`], [`FakeQueue`] — the three drivers.
56//! * [`Worker`], [`run_pool`], [`Shutdown`] — running jobs, and stopping well.
57//! * [`Scheduler`] and [`Cron`] — running work at a time rather than on demand.
58//! * [`QueueDashboard`] — the plugin that exposes it all over HTTP.
59//!
60//! Everything the queue does is reported on `rustlavel_core::events` as
61//! `queue.pushed`, `queue.processed` and `queue.failed`, so Telescope shows
62//! jobs without this crate knowing Telescope exists.
63
64pub mod cron;
65pub mod database;
66pub mod fake;
67pub mod job;
68pub mod memory;
69pub mod plugin;
70pub mod queue;
71pub mod schedule;
72pub mod time;
73pub mod worker;
74pub mod chain;
75pub mod progress;
76
77#[cfg(test)]
78mod tests_support;
79
80pub use chain::Chain;
81pub use cron::{Cron, Weekday};
82pub use database::{CreateJobProgressTable, CreateQueueTables, DatabaseProgress, DatabaseQueue};
83pub use fake::{FakeQueue, fake};
84pub use job::{
85    BoxFuture, DEFAULT_QUEUE, DEFAULT_RETRY_AFTER, DEFAULT_TRIES, FailedJob, Job, JobRegistry,
86    QueuedJob, ReservedJob,
87};
88pub use memory::MemoryQueue;
89pub use progress::{JobContext, MemoryProgress, Progress, ProgressStore};
90pub use plugin::QueueDashboard;
91pub use queue::{Queue, QueueExt};
92pub use schedule::{ScheduledEvent, Scheduler, TickReport, schedule};
93pub use worker::{Outcome, Shutdown, Worker, WorkerOptions, WorkerStats, run_pool};
94
95pub use rustlavel_core::{Error, Result};
96
97/// What a job file imports.
98pub mod prelude {
99    pub use crate::{
100        Cron, DatabaseQueue, FailedJob, FakeQueue, Job, JobRegistry, MemoryQueue, Queue,
101        QueueDashboard, QueueExt, QueuedJob, ReservedJob, Scheduler, Shutdown, Weekday, Worker,
102        run_pool,
103    };
104    pub use rustlavel_core::{Error, Json, Result};
105    pub use std::future::Future;
106    pub use std::sync::Arc;
107    pub use std::time::Duration;
108}
109
110#[cfg(test)]
111mod tests {
112    use crate::prelude::*;
113    use crate::tests_support::{CountingJob, FailingJob, counter, registry};
114    use crate::worker::Outcome;
115
116    /// The whole lifecycle, in the order an application meets it: define,
117    /// dispatch, work, retry, bury.
118    #[tokio::test]
119    async fn a_job_travels_from_dispatch_through_a_worker_to_completion() {
120        let queue: Arc<dyn Queue> = Arc::new(MemoryQueue::new());
121        let worker = Worker::new(Arc::clone(&queue), Arc::new(registry()))
122            .poll_interval(Duration::from_millis(5));
123
124        let tally = counter();
125        queue.dispatch(&CountingJob::new(tally, 10)).await.unwrap();
126        queue.dispatch(&CountingJob::new(tally, 5)).await.unwrap();
127        queue.dispatch(&FailingJob::new(counter(), 1)).await.unwrap();
128
129        let shutdown = Shutdown::new();
130        let stats = worker.max_jobs(3).run(shutdown).await.unwrap();
131
132        assert_eq!(stats.processed, 2);
133        assert_eq!(stats.failed, 1);
134        assert_eq!(tally.total(), 15);
135        assert_eq!(queue.size("default").await.unwrap(), 0);
136        assert_eq!(queue.failed_jobs().await.unwrap().len(), 1);
137    }
138
139    /// The same job, the same worker, against a fake: nothing runs, everything
140    /// is recorded.
141    #[tokio::test]
142    async fn the_fake_stands_in_for_the_real_thing_in_an_application_test() {
143        let queue = crate::fake();
144        let tally = counter();
145
146        queue.dispatch(&CountingJob::new(tally, 10)).await.unwrap();
147
148        queue.assert_pushed("counting").assert_pushed_times("counting", 1);
149        assert_eq!(tally.runs(), 0);
150    }
151
152    #[tokio::test]
153    async fn a_delayed_job_waits_and_a_worker_finds_nothing_meanwhile() {
154        let queue: Arc<dyn Queue> = Arc::new(MemoryQueue::new());
155        let worker = Worker::new(Arc::clone(&queue), Arc::new(registry()));
156
157        queue
158            .dispatch_later(Duration::from_secs(3600), &CountingJob::new(counter(), 1))
159            .await
160            .unwrap();
161
162        assert_eq!(worker.run_once().await.unwrap(), None);
163        assert_eq!(queue.size("default").await.unwrap(), 1);
164    }
165
166    #[tokio::test]
167    async fn the_scheduler_and_the_worker_meet_in_the_middle() {
168        let memory = Arc::new(MemoryQueue::new());
169        let queue: Arc<dyn Queue> = memory.clone();
170        let tally = counter();
171
172        let mut plan = Scheduler::new(Arc::clone(&queue));
173        plan.job(CountingJob::new(tally, 3)).daily().at("13:00");
174
175        plan.tick(crate::time::to_unix(2026, 8, 29, 13, 0, 0)).await.unwrap();
176
177        let worker = Worker::new(Arc::clone(&queue), Arc::new(registry()))
178            .poll_interval(Duration::from_millis(5))
179            .max_jobs(1);
180
181        // The scheduler spawns its dispatch, so the worker may look before the
182        // job lands; it is running a loop for exactly this reason.
183        let shutdown = Shutdown::new();
184        let signal = shutdown.clone();
185        tokio::spawn(async move {
186            tokio::time::sleep(Duration::from_secs(5)).await;
187            signal.signal();
188        });
189
190        let stats = worker.run(shutdown).await.unwrap();
191
192        assert_eq!(stats.processed, 1);
193        assert_eq!(tally.total(), 3);
194    }
195
196    #[tokio::test]
197    async fn a_worker_survives_whatever_a_job_does_to_it() {
198        use crate::tests_support::PanickingJob;
199
200        let queue: Arc<dyn Queue> = Arc::new(MemoryQueue::new());
201        let worker = Worker::new(Arc::clone(&queue), Arc::new(registry()));
202
203        queue.dispatch(&PanickingJob::new(counter(), 1)).await.unwrap();
204        let healthy = counter();
205        queue.dispatch(&CountingJob::new(healthy, 1)).await.unwrap();
206
207        assert_eq!(worker.run_once().await.unwrap(), Some(Outcome::Failed));
208        assert_eq!(worker.run_once().await.unwrap(), Some(Outcome::Processed));
209        assert_eq!(healthy.runs(), 1);
210    }
211}
rustlavel_queue/lib.rs

rustlavel_queue/
lib.rs
