rullst_core/queue/
mod.rs

1//! # Rullst Queue System (`rullst::queue`)
2//!
3//! Provides a unified API for dispatching and processing background jobs.
4//!
5//! ## Drivers
6//! - **SQLite** (`queue-sqlite`, opt-in): Uses an auto-created `rullst_jobs`
7//!   table.
8//! - **Redis** (optional): Requires the `queue-redis` feature flag.
9//!
10//! ## Quick Start
11//! ```rust,no_run
12//! use rullst_core::queue::{Queue, QueueError, Worker, WorkerHandle};
13//!
14//! async fn configure_worker(queue: &Queue) -> Result<WorkerHandle, QueueError> {
15//!     queue
16//!         .dispatch("send_email", serde_json::json!({"to": "user@example.com"}))
17//!         .await?;
18//!
19//!     let mut worker = Worker::new(queue);
20//!     worker.register("send_email", |payload| async move {
21//!         println!("Sending email to: {}", payload["to"]);
22//!         Ok(())
23//!     });
24//!     worker.run()
25//! }
26//! ```
27
28#[cfg(feature = "queue-redis")]
29/// Redis queue driver implementation.
30pub mod redis;
31#[cfg(feature = "queue-sqlite")]
32/// SQLite queue driver implementation.
33pub mod sqlite;
34/// Background job worker executor.
35pub mod worker;
36
37#[cfg(test)]
38mod lease_tests;
39#[cfg(all(test, feature = "queue-sqlite", not(miri)))]
40mod recovery_tests;
41#[cfg(all(test, feature = "queue-sqlite"))]
42mod tests;
43#[cfg(test)]
44mod worker_tests;
45
46#[cfg(feature = "queue-redis")]
47pub use redis::redis_driver::RedisDriver;
48#[cfg(feature = "queue-sqlite")]
49pub use sqlite::SqliteDriver;
50pub use worker::{JobHandler, Worker, WorkerHandle};
51
52use async_trait::async_trait;
53use serde_json::Value;
54use std::sync::Arc;
55use std::time::{Duration, SystemTime, UNIX_EPOCH};
56use uuid::Uuid;
57
58const MAX_SCHEDULE_DELAY: Duration = Duration::from_secs(366 * 24 * 60 * 60);
59
60/// Stalled leases after which the built-in drivers fail a job instead of
61/// requeuing it, so a job that keeps killing its worker cannot loop forever.
62#[cfg_attr(
63    not(any(feature = "queue-sqlite", feature = "queue-redis")),
64    allow(dead_code)
65)]
66pub(crate) const DEFAULT_MAX_STALLED_LEASES: u32 = 5;
67
68// ─── Error Types ────────────────────────────────────────────────────────────
69
70/// Errors that can occur during queue operations.
71#[derive(Debug, thiserror::Error)]
72#[non_exhaustive]
73pub enum QueueError {
74    /// The underlying database or connection failed.
75    #[error("Queue driver error: {0}")]
76    Driver(String),
77    /// Serialization/deserialization of job payloads failed.
78    #[error("Queue serialization error: {0}")]
79    Serialization(String),
80    /// A job handler was not found for the given job name.
81    #[error("No handler registered for job: {0}")]
82    HandlerNotFound(String),
83    /// The job execution itself failed.
84    #[error("Job execution failed: {0}")]
85    JobFailed(String),
86    /// The worker was started outside an active Tokio runtime.
87    #[error("the queue worker requires an active Tokio runtime")]
88    RuntimeUnavailable,
89    /// A worker option would make safe processing impossible.
90    #[error("invalid queue worker configuration: {0}")]
91    InvalidConfiguration(String),
92    /// Persisting a job state transition failed.
93    #[error("job '{job_id}' could not transition via '{operation}': {message}")]
94    StateTransition {
95        /// Job whose state could not be persisted.
96        job_id: String,
97        /// Attempted transition operation.
98        operation: &'static str,
99        /// Driver failure description.
100        message: String,
101    },
102    /// A handler exceeded the configured execution deadline.
103    #[error("job '{job_id}' exceeded its {timeout_ms}ms execution timeout")]
104    JobTimedOut {
105        /// Timed-out job identifier.
106        job_id: String,
107        /// Configured timeout in milliseconds.
108        timeout_ms: u64,
109    },
110    /// A handler panicked and was isolated by the worker.
111    #[error("job '{job_id}' handler panicked")]
112    JobPanicked {
113        /// Panicking job identifier.
114        job_id: String,
115    },
116    /// An internal worker task terminated unexpectedly.
117    #[error("queue worker task failed: {0}")]
118    WorkerTask(String),
119    /// A custom queue backend does not implement an optional lifecycle action.
120    #[error("queue operation is unsupported: {0}")]
121    Unsupported(String),
122}
123
124// ─── Queued Job ─────────────────────────────────────────────────────────────
125
126/// A job that has been placed on the queue and is ready for processing.
127#[derive(Debug, Clone)]
128pub struct QueuedJob {
129    /// Unique identifier for this job instance.
130    pub id: String,
131    /// The job type name (used to look up the handler).
132    pub name: String,
133    /// The JSON payload associated with this job.
134    pub payload: Value,
135    /// Number of times this job has been attempted.
136    pub attempts: u32,
137}
138
139/// Detailed job information, used for dashboard monitoring
140#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
141pub struct QueuedJobDetail {
142    /// Unique identifier of the queued job.
143    pub id: String,
144    /// The name/type of the job.
145    pub name: String,
146    /// The JSON payload of the job as a string.
147    pub payload: String,
148    /// Current execution status of the job (e.g. "pending", "processing", "failed").
149    pub status: String,
150    /// Error message if the job failed.
151    pub error: Option<String>,
152    /// Number of processing attempts made so far.
153    pub attempts: i32,
154    /// Time when the job was originally created/pushed.
155    pub created_at: String,
156    /// Time when the job status was last updated.
157    pub updated_at: String,
158}
159
160// ─── Queue Driver Trait ─────────────────────────────────────────────────────
161
162/// Abstraction over queue storage backends.
163///
164/// Implement this trait to add support for new queue backends.
165/// The framework ships with `SqliteDriver` and (optionally) `RedisDriver`.
166#[async_trait]
167pub trait QueueDriver: Send + Sync {
168    /// Push a new job onto the queue.
169    async fn push(&self, id: &str, job_name: &str, payload: &str) -> Result<(), QueueError>;
170    /// Push a job that must not be claimed before `available_at`.
171    ///
172    /// Custom drivers remain source-compatible. Their default implementation accepts jobs that
173    /// are already due and fails closed for future delivery until the backend implements durable
174    /// scheduling.
175    async fn push_at(
176        &self,
177        id: &str,
178        job_name: &str,
179        payload: &str,
180        available_at: SystemTime,
181    ) -> Result<(), QueueError> {
182        if available_at > SystemTime::now() {
183            Err(QueueError::Unsupported(
184                "this driver cannot persist future scheduled jobs".to_string(),
185            ))
186        } else {
187            self.push(id, job_name, payload).await
188        }
189    }
190    /// Pop the next available job from the queue (FIFO).
191    ///
192    /// [`Worker`] always drives this future to completion and never cancels an
193    /// in-flight claim, so implementations may commit the claim before they
194    /// return. Graceful worker shutdown waits for that call, so it should not
195    /// block indefinitely.
196    async fn pop(&self) -> Result<Option<QueuedJob>, QueueError>;
197    /// Mark a job as successfully completed.
198    ///
199    /// Drivers may remove it immediately or retain a bounded history under an explicit policy.
200    /// This transition is not fenced by claim; [`Worker`] uses
201    /// [`Self::mark_complete_attempt`].
202    async fn mark_complete(&self, job_id: &str) -> Result<(), QueueError>;
203    /// Mark a job as failed, recording the error message.
204    async fn mark_failed(&self, job_id: &str, error: &str) -> Result<(), QueueError>;
205    /// Return an interrupted processing job to the pending state.
206    async fn requeue(&self, _job_id: &str, _reason: &str) -> Result<(), QueueError> {
207        Err(QueueError::Unsupported(
208            "this driver cannot requeue interrupted jobs".to_string(),
209        ))
210    }
211    /// Marks the exact claim returned by [`Self::pop`] as completed.
212    ///
213    /// `attempt` is that claim's [`QueuedJob::attempts`] value and acts as a
214    /// fencing token: the built-in drivers complete the job only while it is
215    /// still processing under the same attempt, so a worker whose lease was
216    /// recovered and claimed again cannot finish the newer claim. [`Worker`]
217    /// calls this method. The default delegates to [`Self::mark_complete`]
218    /// without a fence, so existing custom drivers keep their behaviour.
219    async fn mark_complete_attempt(&self, job_id: &str, _attempt: u32) -> Result<(), QueueError> {
220        self.mark_complete(job_id).await
221    }
222    /// Marks the exact claim returned by [`Self::pop`] as failed.
223    ///
224    /// Fenced like [`Self::mark_complete_attempt`]; the default delegates to
225    /// [`Self::mark_failed`].
226    async fn mark_failed_attempt(
227        &self,
228        job_id: &str,
229        _attempt: u32,
230        error: &str,
231    ) -> Result<(), QueueError> {
232        self.mark_failed(job_id, error).await
233    }
234    /// Returns the exact claim returned by [`Self::pop`] to the pending state.
235    ///
236    /// Fenced like [`Self::mark_complete_attempt`]; the default delegates to
237    /// [`Self::requeue`].
238    async fn requeue_attempt(
239        &self,
240        job_id: &str,
241        _attempt: u32,
242        reason: &str,
243    ) -> Result<(), QueueError> {
244        self.requeue(job_id, reason).await
245    }
246    /// Returns the exact claim returned by [`Self::pop`] to the pending state,
247    /// not claimable again until `delay` has passed.
248    ///
249    /// [`Worker`] uses this to hand back a job whose name it has no handler
250    /// for, so a worker that registered that name (for example a newer version
251    /// during a rolling deploy) can run it. Fenced like
252    /// [`Self::mark_complete_attempt`]. The default returns
253    /// [`QueueError::Unsupported`], and the worker then fails the job instead.
254    async fn requeue_attempt_after(
255        &self,
256        _job_id: &str,
257        _attempt: u32,
258        _reason: &str,
259        _delay: Duration,
260    ) -> Result<(), QueueError> {
261        Err(QueueError::Unsupported(
262            "this driver cannot defer claimed jobs".to_string(),
263        ))
264    }
265    /// Recover processing leases left behind by a crashed worker.
266    ///
267    /// The built-in SQLite and Redis drivers fail a job, instead of returning
268    /// it to pending, when its lease has stalled five times, and report both
269    /// requeued and failed leases in the returned count.
270    async fn recover_stalled(&self, _stale_after: std::time::Duration) -> Result<u64, QueueError> {
271        Err(QueueError::Unsupported(
272            "this driver cannot recover stalled jobs".to_string(),
273        ))
274    }
275    /// Return the count of pending jobs.
276    async fn pending_count(&self) -> Result<u64, QueueError>;
277    /// List all recent jobs for monitoring
278    async fn list_all_jobs(&self, _limit: u32) -> Result<Vec<QueuedJobDetail>, QueueError> {
279        Err(QueueError::Unsupported(
280            "this queue driver does not expose job inspection".to_string(),
281        ))
282    }
283    /// Retry a failed job
284    async fn retry_failed_job(&self, _job_id: &str) -> Result<(), QueueError> {
285        Err(QueueError::Unsupported(
286            "this queue driver does not expose failed-job retry".to_string(),
287        ))
288    }
289    /// Legacy, misnamed hook that purges failed jobs in the SQLite driver.
290    async fn purge_completed_jobs(&self) -> Result<(), QueueError> {
291        Err(QueueError::Unsupported(
292            "this queue driver does not expose failed-job purge".to_string(),
293        ))
294    }
295    /// Purge failed jobs retained by the backend.
296    async fn purge_failed_jobs(&self) -> Result<(), QueueError> {
297        self.purge_completed_jobs().await
298    }
299    /// Purge successful jobs retained by an explicit backend history policy.
300    async fn purge_completed_history(&self) -> Result<(), QueueError> {
301        Err(QueueError::Unsupported(
302            "this queue driver does not retain completed job history".to_string(),
303        ))
304    }
305}
306
307// ─── Queue Facade ───────────────────────────────────────────────────────────
308
309/// The main queue facade for dispatching background jobs.
310///
311/// Provides a driver-agnostic API. Create with `Queue::sqlite()` or `Queue::redis()`.
312pub struct Queue {
313    driver: Arc<Box<dyn QueueDriver>>,
314}
315
316impl Queue {
317    /// Create a queue backed by SQLite. The `rullst_jobs` table is auto-created.
318    #[cfg(feature = "queue-sqlite")]
319    pub async fn sqlite(database_url: impl Into<String>) -> Result<Self, QueueError> {
320        let driver = SqliteDriver::new(database_url).await?;
321        Ok(Self {
322            driver: Arc::new(Box::new(driver)),
323        })
324    }
325
326    /// Creates a SQLite queue with explicit bounded retention of successful jobs.
327    ///
328    /// Retained rows include the original payload and are visible to queue inspection. The caller
329    /// owns access control and retention policy; [`Self::sqlite`] deletes successes by default.
330    #[cfg(feature = "queue-sqlite")]
331    pub async fn sqlite_with_completed_history(
332        database_url: impl Into<String>,
333        retained_jobs: usize,
334    ) -> Result<Self, QueueError> {
335        let driver = SqliteDriver::new(database_url)
336            .await?
337            .try_with_completed_history_limit(retained_jobs)?;
338        Ok(Self {
339            driver: Arc::new(Box::new(driver)),
340        })
341    }
342
343    /// Create a queue backed by Redis. Requires the `queue-redis` feature.
344    #[cfg(feature = "queue-redis")]
345    #[cfg_attr(mutants, mutants::skip)]
346    pub fn redis(redis_url: impl Into<String>) -> Result<Self, QueueError> {
347        let driver = redis::redis_driver::RedisDriver::new(redis_url)?;
348        Ok(Self {
349            driver: Arc::new(Box::new(driver)),
350        })
351    }
352
353    /// Create a queue from any custom driver implementing `QueueDriver`.
354    pub fn custom(driver: Box<dyn QueueDriver>) -> Self {
355        Self {
356            driver: Arc::new(driver),
357        }
358    }
359
360    /// Dispatch a named job with a JSON payload onto the queue.
361    pub async fn dispatch(&self, job_name: &str, payload: Value) -> Result<String, QueueError> {
362        let id = Uuid::new_v4().to_string();
363        let payload_str = serde_json::to_string(&payload)
364            .map_err(|e| QueueError::Serialization(e.to_string()))?;
365        self.driver.push(&id, job_name, &payload_str).await?;
366        Ok(id)
367    }
368
369    /// Dispatches a durable job that cannot be claimed before `available_at`.
370    ///
371    /// The built-in SQLite and Redis drivers support millisecond scheduling for at most 366 days
372    /// ahead. Actual execution occurs on the first worker poll after the timestamp; wall-clock
373    /// precision, provider acceptance, and exactly-once delivery are not implied.
374    pub async fn dispatch_at(
375        &self,
376        job_name: &str,
377        payload: Value,
378        available_at: SystemTime,
379    ) -> Result<String, QueueError> {
380        if available_at.duration_since(UNIX_EPOCH).is_err() {
381            return Err(QueueError::InvalidConfiguration(
382                "scheduled timestamp predates the Unix epoch".to_string(),
383            ));
384        }
385        let now = SystemTime::now();
386        if available_at
387            .duration_since(now)
388            .is_ok_and(|delay| delay > MAX_SCHEDULE_DELAY)
389        {
390            return Err(QueueError::InvalidConfiguration(
391                "scheduled jobs may be dispatched at most 366 days ahead".to_string(),
392            ));
393        }
394        let id = Uuid::new_v4().to_string();
395        let payload_str = serde_json::to_string(&payload)
396            .map_err(|error| QueueError::Serialization(error.to_string()))?;
397        self.driver
398            .push_at(&id, job_name, &payload_str, available_at)
399            .await?;
400        Ok(id)
401    }
402
403    /// Return the number of pending jobs in the queue.
404    pub async fn pending_count(&self) -> Result<u64, QueueError> {
405        self.driver.pending_count().await
406    }
407
408    /// List all recent jobs for visual monitoring
409    pub async fn list_all_jobs(&self, limit: u32) -> Result<Vec<QueuedJobDetail>, QueueError> {
410        self.driver.list_all_jobs(limit).await
411    }
412
413    /// Retry a failed job in the queue
414    #[cfg_attr(mutants, mutants::skip)]
415    pub async fn retry_failed_job(&self, job_id: &str) -> Result<(), QueueError> {
416        self.driver.retry_failed_job(job_id).await
417    }
418
419    /// Purge failed jobs retained by the queue backend.
420    pub async fn purge_failed_jobs(&self) -> Result<(), QueueError> {
421        self.driver.purge_failed_jobs().await
422    }
423
424    /// Purges successful jobs retained by the queue backend.
425    pub async fn purge_completed_history(&self) -> Result<(), QueueError> {
426        self.driver.purge_completed_history().await
427    }
428
429    /// Legacy compatibility name for [`Self::purge_failed_jobs`].
430    #[deprecated(
431        since = "12.0.0",
432        note = "use purge_failed_jobs; this method has always targeted failed rows"
433    )]
434    #[cfg_attr(mutants, mutants::skip)]
435    pub async fn purge_completed_jobs(&self) -> Result<(), QueueError> {
436        self.purge_failed_jobs().await
437    }
438
439    /// Get an `Arc` reference to the internal driver (for sharing with `Worker`).
440    pub(crate) fn driver_ref(&self) -> Arc<Box<dyn QueueDriver>> {
441        Arc::clone(&self.driver)
442    }
443}
444
445#[cfg(any(feature = "queue-sqlite", feature = "queue-redis"))]
446pub(crate) fn unix_timestamp_millis_ceil(timestamp: SystemTime) -> Result<u64, QueueError> {
447    let duration = timestamp.duration_since(UNIX_EPOCH).map_err(|_| {
448        QueueError::InvalidConfiguration("timestamp predates Unix epoch".to_string())
449    })?;
450    let whole_millis = duration.as_millis();
451    let has_fractional_millisecond = duration.subsec_nanos() % 1_000_000 != 0;
452    let rounded_millis = whole_millis
453        .checked_add(u128::from(has_fractional_millisecond))
454        .ok_or_else(|| {
455            QueueError::InvalidConfiguration("timestamp exceeds queue storage range".to_string())
456        })?;
457    u64::try_from(rounded_millis).map_err(|_| {
458        QueueError::InvalidConfiguration("timestamp exceeds queue storage range".to_string())
459    })
460}
461
462#[cfg(feature = "queue-sqlite")]
463pub(crate) fn unix_timestamp_millis_floor(timestamp: SystemTime) -> Result<u64, QueueError> {
464    let duration = timestamp.duration_since(UNIX_EPOCH).map_err(|_| {
465        QueueError::InvalidConfiguration("timestamp predates Unix epoch".to_string())
466    })?;
467    u64::try_from(duration.as_millis()).map_err(|_| {
468        QueueError::InvalidConfiguration("timestamp exceeds queue storage range".to_string())
469    })
470}
rullst_core/queue/mod.rs

rullst_core/queue/
mod.rs
