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;
74
75#[cfg(test)]
76mod tests_support;
77
78pub use cron::{Cron, Weekday};
79pub use database::{CreateQueueTables, DatabaseQueue};
80pub use fake::{FakeQueue, fake};
81pub use job::{
82    BoxFuture, DEFAULT_QUEUE, DEFAULT_RETRY_AFTER, DEFAULT_TRIES, FailedJob, Job, JobRegistry,
83    QueuedJob, ReservedJob,
84};
85pub use memory::MemoryQueue;
86pub use plugin::QueueDashboard;
87pub use queue::{Queue, QueueExt};
88pub use schedule::{ScheduledEvent, Scheduler, TickReport, schedule};
89pub use worker::{Outcome, Shutdown, Worker, WorkerOptions, WorkerStats, run_pool};
90
91pub use rustlavel_core::{Error, Result};
92
93/// What a job file imports.
94pub mod prelude {
95    pub use crate::{
96        Cron, DatabaseQueue, FailedJob, FakeQueue, Job, JobRegistry, MemoryQueue, Queue, QueueExt,
97        QueuedJob, ReservedJob, Scheduler, Shutdown, Weekday, Worker, run_pool,
98    };
99    pub use rustlavel_core::{Error, Json, Result};
100    pub use std::future::Future;
101    pub use std::sync::Arc;
102    pub use std::time::Duration;
103}
104
105#[cfg(test)]
106mod tests {
107    use crate::prelude::*;
108    use crate::tests_support::{CountingJob, FailingJob, counter, registry};
109    use crate::worker::Outcome;
110
111    /// The whole lifecycle, in the order an application meets it: define,
112    /// dispatch, work, retry, bury.
113    #[tokio::test]
114    async fn a_job_travels_from_dispatch_through_a_worker_to_completion() {
115        let queue: Arc<dyn Queue> = Arc::new(MemoryQueue::new());
116        let worker = Worker::new(Arc::clone(&queue), Arc::new(registry()))
117            .poll_interval(Duration::from_millis(5));
118
119        let tally = counter();
120        queue.dispatch(&CountingJob::new(tally, 10)).await.unwrap();
121        queue.dispatch(&CountingJob::new(tally, 5)).await.unwrap();
122        queue.dispatch(&FailingJob::new(counter(), 1)).await.unwrap();
123
124        let shutdown = Shutdown::new();
125        let stats = worker.max_jobs(3).run(shutdown).await.unwrap();
126
127        assert_eq!(stats.processed, 2);
128        assert_eq!(stats.failed, 1);
129        assert_eq!(tally.total(), 15);
130        assert_eq!(queue.size("default").await.unwrap(), 0);
131        assert_eq!(queue.failed_jobs().await.unwrap().len(), 1);
132    }
133
134    /// The same job, the same worker, against a fake: nothing runs, everything
135    /// is recorded.
136    #[tokio::test]
137    async fn the_fake_stands_in_for_the_real_thing_in_an_application_test() {
138        let queue = crate::fake();
139        let tally = counter();
140
141        queue.dispatch(&CountingJob::new(tally, 10)).await.unwrap();
142
143        queue.assert_pushed("counting").assert_pushed_times("counting", 1);
144        assert_eq!(tally.runs(), 0);
145    }
146
147    #[tokio::test]
148    async fn a_delayed_job_waits_and_a_worker_finds_nothing_meanwhile() {
149        let queue: Arc<dyn Queue> = Arc::new(MemoryQueue::new());
150        let worker = Worker::new(Arc::clone(&queue), Arc::new(registry()));
151
152        queue
153            .dispatch_later(Duration::from_secs(3600), &CountingJob::new(counter(), 1))
154            .await
155            .unwrap();
156
157        assert_eq!(worker.run_once().await.unwrap(), None);
158        assert_eq!(queue.size("default").await.unwrap(), 1);
159    }
160
161    #[tokio::test]
162    async fn the_scheduler_and_the_worker_meet_in_the_middle() {
163        let memory = Arc::new(MemoryQueue::new());
164        let queue: Arc<dyn Queue> = memory.clone();
165        let tally = counter();
166
167        let mut plan = Scheduler::new(Arc::clone(&queue));
168        plan.job(CountingJob::new(tally, 3)).daily().at("13:00");
169
170        plan.tick(crate::time::to_unix(2026, 8, 29, 13, 0, 0)).await.unwrap();
171
172        let worker = Worker::new(Arc::clone(&queue), Arc::new(registry()))
173            .poll_interval(Duration::from_millis(5))
174            .max_jobs(1);
175
176        // The scheduler spawns its dispatch, so the worker may look before the
177        // job lands; it is running a loop for exactly this reason.
178        let shutdown = Shutdown::new();
179        let signal = shutdown.clone();
180        tokio::spawn(async move {
181            tokio::time::sleep(Duration::from_secs(5)).await;
182            signal.signal();
183        });
184
185        let stats = worker.run(shutdown).await.unwrap();
186
187        assert_eq!(stats.processed, 1);
188        assert_eq!(tally.total(), 3);
189    }
190
191    #[tokio::test]
192    async fn a_worker_survives_whatever_a_job_does_to_it() {
193        use crate::tests_support::PanickingJob;
194
195        let queue: Arc<dyn Queue> = Arc::new(MemoryQueue::new());
196        let worker = Worker::new(Arc::clone(&queue), Arc::new(registry()));
197
198        queue.dispatch(&PanickingJob::new(counter(), 1)).await.unwrap();
199        let healthy = counter();
200        queue.dispatch(&CountingJob::new(healthy, 1)).await.unwrap();
201
202        assert_eq!(worker.run_once().await.unwrap(), Some(Outcome::Failed));
203        assert_eq!(worker.run_once().await.unwrap(), Some(Outcome::Processed));
204        assert_eq!(healthy.runs(), 1);
205    }
206}
rustlavel_queue/lib.rs

rustlavel_queue/
lib.rs
