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

rustlavel_queue/
lib.rs
