# Qrush
[](https://crates.io/crates/qrush)
[](https://docs.rs/qrush)
[](LICENSE)
[](https://crates.io/crates/qrush)
A lightweight, production-ready job queue and task scheduler for Rust applications built on Redis and Tokio. The core is web-framework agnostic, and the optional built-in dashboard works with **either Actix Web or Axum**. Qrush provides both integrated and separate process modes, making it suitable for everything from simple background tasks to large-scale distributed systems.
## Features
- 🚀 **Dual Deployment Modes**: Integrated (single process) or separate worker process
- 🧩 **Framework Choice**: Optional dashboard for Actix Web *or* Axum; the queue/worker core needs neither
- ⚡ **High Performance**: Built on Redis and Tokio for maximum throughput
- 📅 **Cron Scheduling**: Full cron expression support for recurring tasks
- ⏱️ **Delayed Jobs**: Schedule jobs to run after a specified delay
- 📊 **Built-in Metrics UI**: Real-time dashboard for monitoring queues, jobs, and workers
- 🔒 **Security**: Optional Basic Auth for metrics endpoints
- 🎯 **Type-Safe**: Leverages Rust's type system for safe job handling
- 🔄 **Graceful Shutdown**: Clean worker shutdown with configurable grace periods
- 📈 **Scalable**: Support for multiple queues with different priorities and concurrency levels
## Feature Flags
The built-in dashboard is optional and works with **either Actix or Axum** —
pick the one that matches your app.
| `dashboard-actix` | ❌ | Metrics dashboard served with Actix Web (`qrush::routes::metrics_route`). Pulls in Actix Web, Tera, and the web stack. |
| `dashboard-axum` | ❌ | Metrics dashboard served with Axum (`qrush::routes::axum_route`). Pulls in Axum, Tera, and the web stack. |
| `dashboard` | ❌ | Back-compat alias for `dashboard-actix`. |
**Library-only usage (default).** No dashboard framework is enabled by default,
so a plain dependency gives you `enqueue` + workers with no web stack:
```toml
[dependencies]
qrush = "2.0.2"
```
To mount the dashboard, opt into one framework:
```toml
# Actix
qrush = { version = "2.0.2", features = ["dashboard-actix"] }
# Axum
qrush = { version = "2.0.2", features = ["dashboard-axum"] }
```
### Migrating from 1.x to 2.0
In 1.x the dashboard was Actix-only and enabled by default. In 2.0 it is
framework-selectable and **off by default**. Nothing else changed — the
route-wiring function and all queue/worker/cron APIs are the same.
| Dashboard default | on (Actix) | off |
| Enable Actix dashboard | (default) | `features = ["dashboard-actix"]` |
| Enable Axum dashboard | not available | `features = ["dashboard-axum"]` |
```toml
# 1.x
qrush = "1.0.1"
# 2.0 — Actix (equivalent to the old default; no code changes needed)
qrush = { version = "2.0.2", features = ["dashboard-actix"] }
```
If you only used `enqueue` + workers (no dashboard), a plain `qrush = "2.0.2"`
now pulls in **less** — the web stack is no longer compiled by default. See the
[CHANGELOG](CHANGELOG.md) for the full list of changes.
## Quick Start
### Installation
Add to your `Cargo.toml`:
```toml
[dependencies]
qrush = "2.0.2"
tokio = { version = "1", features = ["rt-multi-thread", "macros"] }
serde = { version = "1", features = ["derive"] }
async-trait = "0.1"
anyhow = "1"
futures = "0.3"
```
> `qrush` bundles its own Redis client (with cluster support), so you don't need
> to depend on `redis` directly unless you use it yourself.
### Basic Usage (Integrated Mode)
```rust
use qrush::job::Job;
use qrush::queue::{enqueue, enqueue_in};
use qrush::config::QueueConfig;
use qrush::registry::register_job;
use async_trait::async_trait;
use serde::{Deserialize, Serialize};
use futures::future::BoxFuture;
use anyhow::Result;
#[derive(Clone, Serialize, Deserialize)]
pub struct EmailJob {
pub to: String,
pub subject: String,
}
#[async_trait]
impl Job for EmailJob {
async fn perform(&self) -> Result<()> {
println!("Sending email to {}: {}", self.to, self.subject);
// Your email sending logic here
Ok(())
}
fn name(&self) -> &'static str { "EmailJob" }
fn queue(&self) -> &'static str { "default" }
}
impl EmailJob {
pub fn name() -> &'static str { "EmailJob" }
pub fn handler(payload: String) -> BoxFuture<'static, Result<Box<dyn Job>>> {
Box::pin(async move {
let job: EmailJob = serde_json::from_str(&payload)?;
Ok(Box::new(job) as Box<dyn Job>)
})
}
}
#[tokio::main]
async fn main() -> Result<()> {
// Set Redis URL
std::env::set_var("REDIS_URL", "redis://127.0.0.1:6379");
// Register job
register_job(EmailJob::name(), EmailJob::handler);
// Initialize queues
let queues = vec![
QueueConfig::new("default", 5, 0),
];
QueueConfig::initialize(
"redis://127.0.0.1:6379".to_string(),
queues
).await?;
// Enqueue a job
enqueue(EmailJob {
to: "user@example.com".to_string(),
subject: "Hello!".to_string(),
}).await?;
// Keep running
tokio::signal::ctrl_c().await?;
Ok(())
}
```
## Architecture
QRush supports two deployment modes:
### Integrated Mode
Workers run in the same process as your application. Perfect for small to medium applications.
```
┌─────────────────────┐
│ Application │
│ (Single Process) │
│ │
│ • HTTP Server │
│ • Enqueue Jobs │
│ • Process Jobs │ ← Workers here
└─────────────────────┘
```
### Separate Process Mode
Workers run in a dedicated process. Recommended for production environments.
```
┌─────────────────────┐ ┌─────────────────────┐
│ Web Server │ │ qrush-engine │
│ (cargo run) │ │ (separate process) │
│ │ │ │
│ • HTTP Server │ │ • Worker Pools │
│ • Enqueue Jobs ─────┼──Redis──┼─▶ Process Jobs │
│ • Serve Routes │ │ • Cron Scheduler │
└─────────────────────┘ └─────────────────────┘
```
## Documentation
### Integrated Mode
See [Part 1: Integrated Mode](#integrated-mode-detailed) below for complete setup instructions.
### Separate Process Mode
See [Part 2: Separate Process Mode](#separate-process-mode-detailed) below for production deployment.
## API Reference
### Core Traits
- `Job`: Implement this trait for your job types. Only `perform`, `name`, and
`queue` are required; the `before`/`after`/`on_error`/`always`
[lifecycle hooks](#job-lifecycle-hooks) are optional overrides.
- `CronJob`: Implement for recurring scheduled jobs
### Core Functions
- `enqueue(job) -> QrushResult<String>`: Enqueue a job immediately; returns the job ID
- `enqueue_in(job, delay_secs) -> QrushResult<String>`: Enqueue a job with a [delay](#delayed-jobs); returns the job ID
- `register_job(name, handler)`: Register a job handler
- `QueueConfig::initialize(redis_url, queues)`: Start worker pools **and** the cron scheduler
- `set_basic_auth(Some(QrushBasicAuthConfig { .. }))`: [Protect the dashboard](#securing-the-dashboard-basic-auth) with HTTP Basic Auth
Failed jobs are [retried automatically](#retries--dead-letter-queue) with
exponential backoff and moved to a dead-letter queue after `MAX_RETRIES` (3).
### Cron Scheduling
All under `qrush::cron::cron_scheduler::CronScheduler` (see [Cron Jobs](#cron-jobs)):
- `register_cron_job(job) -> Result<()>`: Persist a schedule to Redis
- `list_cron_jobs() -> Result<Vec<CronJobMeta>>`: List registered cron jobs
- `run_now(cron_id) -> Result<String>`: Enqueue a cron job immediately
- `toggle_cron_job(cron_id, enabled) -> Result<()>`: Pause / resume a schedule
- `delete_cron_job(cron_id) -> Result<()>`: Remove a schedule
### Errors
The public API returns `QrushResult<T>` (`Result<T, QrushError>`). `QrushError`
distinguishes `Redis`, `Serialization`, and `Config` failures, and implements
`std::error::Error`, so it still propagates through `?` in `anyhow`-based code.
### Engine Runtime
- `qrush::engine::run_engine(redis_url, queues, shutdown_grace_secs)`: Run worker process
- `qrush::engine::parse_queues(spec)`: Parse queue specification string
### Command-Line Interface
The crate also ships reference binaries — `qrush` (a management CLI with
`start`/`stop`/`status`/`stats`/`queues`/`jobs` subcommands) and `qrush-engine`
(the worker process) — that you can adapt for your own app. See
[`src/bin/cli.md`](src/bin/cli.md) for the full CLI guide.
## Examples
### Runnable dashboard examples
The repo ships a complete, runnable dashboard example for each framework. With a
Redis instance available (`REDIS_URL`, defaults to `redis://127.0.0.1:6379`):
```sh
# Actix — serves http://127.0.0.1:8080/qrush/metrics
cargo run --example actix_dashboard --features dashboard-actix
# Axum — serves http://127.0.0.1:8080/qrush/metrics
cargo run --example axum_dashboard --features dashboard-axum
```
### Job Lifecycle Hooks
Beyond `perform`, the [`Job`](#core-traits) trait exposes optional hooks that
wrap each execution. All are `async` and have default no-op implementations, so
you only override the ones you need:
| `before` | Before `perform` | `async fn before(&self) -> Result<()>` | Return `Err` to **skip** the job — it is marked `skipped` (a terminal, non-failure state) and `perform` never runs. |
| `perform` | The actual work | `async fn perform(&self) -> Result<()>` | Return `Err` to trigger [retry / dead-letter](#retries--dead-letter-queue). |
| `after` | After a **successful** `perform` | `async fn after(&self)` | Skipped if `perform` errored. |
| `on_error` | After a **failed** `perform` | `async fn on_error(&self, err: &anyhow::Error)` | Runs before the retry is scheduled. Good for logging/alerting. |
| `always` | After every attempt that ran `perform` | `async fn always(&self)` | Runs on both success and failure (but not when `before` skipped the job). |
```rust
use qrush::job::Job;
use async_trait::async_trait;
use serde::{Deserialize, Serialize};
use anyhow::{bail, Result};
#[derive(Clone, Serialize, Deserialize)]
pub struct ChargeCard {
pub user_id: String,
pub amount_cents: u64,
}
#[async_trait]
impl Job for ChargeCard {
// Guard: bail out early (job is marked `skipped`, not `failed`).
async fn before(&self) -> Result<()> {
if self.amount_cents == 0 {
bail!("nothing to charge — skipping");
}
Ok(())
}
async fn perform(&self) -> Result<()> {
println!("Charging {} cents to {}", self.amount_cents, self.user_id);
// ... call payment gateway; return Err to retry ...
Ok(())
}
async fn after(&self) {
println!("charge succeeded — sending receipt");
}
async fn on_error(&self, err: &anyhow::Error) {
eprintln!("charge failed, will retry: {err}");
}
async fn always(&self) {
println!("charge attempt finished (success or failure)");
}
fn name(&self) -> &'static str { "ChargeCard" }
fn queue(&self) -> &'static str { "critical" }
}
```
### Delayed Jobs
`enqueue_in(job, delay_secs)` runs a job after a delay instead of immediately.
It returns the job ID and is otherwise identical to `enqueue` — same job type,
same worker, same retry semantics.
```rust
use qrush::queue::{enqueue, enqueue_in};
// Run now.
let id = enqueue(EmailJob {
to: "user@example.com".into(),
subject: "Welcome!".into(),
}).await?;
// Run in 10 minutes (600 seconds).
let id = enqueue_in(EmailJob {
to: "user@example.com".into(),
subject: "Don't forget to verify your email".into(),
}, 600).await?;
```
Delayed jobs sit in a Redis sorted set keyed by their run-at timestamp; a
dedicated delayed-worker pool (started by `QueueConfig::initialize`) promotes
them onto their queue once due. Precision is bounded by the poll interval, so
treat the delay as "at least N seconds", not an exact wall-clock alarm.
### Retries & Dead-Letter Queue
When `perform` returns `Err`, QRush retries the job automatically — you don't
schedule retries yourself:
1. `on_error` is called, and the error string is stored on the job.
2. The job's retry counter increments. While it's `<= 3` (`MAX_RETRIES`), the
job is re-queued with **exponential backoff plus jitter**
(`10s * 2^retries`, jittered to avoid thundering-herd retries) and its status
becomes `retrying`.
3. After the 3rd retry is exhausted, the job moves to the **dead-letter queue**
(`status = dead`) instead of being dropped. Inspect and requeue dead jobs
from the dashboard at `/qrush/metrics/extras/dead` (or the dead-jobs view).
Job status values you'll see in Redis / on the dashboard:
| `pending` | Enqueued, waiting for a worker |
| `delayed` | Scheduled via `enqueue_in`, not yet due |
| `retrying` | Failed once or more; waiting for its backoff to elapse |
| `skipped` | `before()` returned `Err`; terminal, treated as a non-failure |
| `success` | `perform()` completed successfully |
| `dead` | Retries exhausted; parked in the dead-letter queue |
| `failed` | Could not run at all (e.g. no handler registered for the job name) |
> Retries and the dead-letter queue are handled by the **worker** process, so
> they apply wherever `QueueConfig::initialize` runs — the app in integrated
> mode, or the engine binary in [separate process mode](#separate-process-mode-detailed).
### Cron Jobs
A cron job is a regular [`Job`](#basic-usage-integrated-mode) that runs on a schedule instead of
being enqueued by hand. The work still lives in `Job::perform`; `CronJob` only
adds *when* to run it. Follow these three steps.
#### Step 1 — Define the job and its `perform()`
This is identical to any other QRush job: implement `Job` (the work + a handler
so a worker can rebuild it from Redis).
```rust
use qrush::job::Job;
use qrush::registry::register_job;
use async_trait::async_trait;
use serde::{Deserialize, Serialize};
use futures::future::BoxFuture;
use anyhow::Result;
#[derive(Clone, Serialize, Deserialize)]
pub struct EmailJob {
pub to: String,
pub subject: String,
}
#[async_trait]
impl Job for EmailJob {
// 👇 This is the execution — it runs every time the schedule fires.
async fn perform(&self) -> Result<()> {
println!("Sending email to {}: {}", self.to, self.subject);
// Your recurring work goes here.
Ok(())
}
fn name(&self) -> &'static str { "EmailJob" }
fn queue(&self) -> &'static str { "default" }
}
impl EmailJob {
pub fn name() -> &'static str { "EmailJob" }
// Lets a worker rebuild the job from its stored payload.
pub fn handler(payload: String) -> BoxFuture<'static, Result<Box<dyn Job>>> {
Box::pin(async move {
let job: EmailJob = serde_json::from_str(&payload)?;
Ok(Box::new(job) as Box<dyn Job>)
})
}
}
```
#### Step 2 — Add the schedule (`CronJob`)
Attach a cron expression and a unique id to the same type.
```rust
use qrush::cron::cron_job::CronJob;
#[async_trait]
impl CronJob for EmailJob {
// 6-field: sec min hour day month weekday. See "Cron Expressions" below.
fn cron_expression(&self) -> &'static str { "0 0 * * * *" } // every hour
fn cron_id(&self) -> &'static str { "hourly_email" } // must be unique
}
```
#### Step 3 — Register and start it in `main`
The job above is framework-agnostic; only `main` differs. In **both** frameworks
the order is the same:
1. `set_redis_url(...)` — required before any Redis call.
2. `register_job(...)` — so a worker can run the job.
3. `CronScheduler::register_cron_job(...)` — saves the schedule to Redis.
4. `QueueConfig::initialize(...)` — starts the workers **and** the cron scheduler.
> ⚠️ The cron scheduler only runs where `QueueConfig::initialize` is called. In
> [Separate Process Mode](#separate-process-mode-detailed) that's the engine
> binary, not the web server — put steps 2–4 there.
**Actix** (`features = ["dashboard-actix"]`):
```rust
use qrush::config::{set_redis_url, QueueConfig};
use qrush::registry::register_job;
use qrush::cron::cron_scheduler::CronScheduler;
use qrush::routes::metrics_route::qrush_metrics_routes;
use actix_web::{web, App, HttpServer};
#[actix_web::main]
async fn main() -> anyhow::Result<()> {
let redis_url = std::env::var("REDIS_URL")
.unwrap_or_else(|_| "redis://127.0.0.1:6379".to_string());
set_redis_url(redis_url.clone())?; // 1
register_job(EmailJob::name(), EmailJob::handler); // 2
let job = EmailJob { // 3
to: "user@example.com".into(),
subject: "Hourly report".into(),
};
CronScheduler::register_cron_job(job).await?;
let queues = vec![QueueConfig::new("default", 5, 0)]; // 4
QueueConfig::initialize(redis_url, queues).await?;
// Serve the dashboard at http://127.0.0.1:8080/qrush/metrics
HttpServer::new(|| {
App::new().service(web::scope("/qrush").configure(qrush_metrics_routes))
})
.bind("0.0.0.0:8080")?
.run()
.await?;
Ok(())
}
```
**Axum** (`features = ["dashboard-axum"]`):
```rust
use qrush::config::{set_redis_url, QueueConfig};
use qrush::registry::register_job;
use qrush::cron::cron_scheduler::CronScheduler;
use qrush::routes::axum_route::qrush_metrics_router;
use axum::Router;
#[tokio::main]
async fn main() -> anyhow::Result<()> {
let redis_url = std::env::var("REDIS_URL")
.unwrap_or_else(|_| "redis://127.0.0.1:6379".to_string());
set_redis_url(redis_url.clone())?; // 1
register_job(EmailJob::name(), EmailJob::handler); // 2
let job = EmailJob { // 3
to: "user@example.com".into(),
subject: "Hourly report".into(),
};
CronScheduler::register_cron_job(job).await?;
let queues = vec![QueueConfig::new("default", 5, 0)]; // 4
QueueConfig::initialize(redis_url, queues).await?;
// Serve the dashboard at http://127.0.0.1:8080/qrush/metrics
let app = Router::new().nest("/qrush", qrush_metrics_router());
let listener = tokio::net::TcpListener::bind("0.0.0.0:8080").await?;
axum::serve(listener, app).await?;
Ok(())
}
```
**That's it.** When the schedule fires, QRush enqueues the job onto its `queue()`
and a worker runs `perform()`. Watch it (and manage schedules) on the dashboard
at `/qrush/metrics/extras/cron`.
> **On restart:** `register_cron_job` errors if a job with the same `cron_id`
> already exists in Redis, so the `?` above would abort a second boot. The
> schedule already survives restarts, so either skip re-registering, treat the
> duplicate as non-fatal (log and continue instead of `?`), or call
> `CronScheduler::delete_cron_job("hourly_email")` first to re-seed it.
#### Managing cron jobs
Manage schedules from the dashboard at `/qrush/metrics/extras/cron`, or
programmatically via `CronScheduler`:
```rust
use qrush::cron::cron_scheduler::CronScheduler;
CronScheduler::list_cron_jobs().await?; // -> Vec<CronJobMeta>
CronScheduler::run_now("hourly_email").await?; // enqueue once, right now
CronScheduler::toggle_cron_job("hourly_email", false).await?; // pause
CronScheduler::toggle_cron_job("hourly_email", true).await?; // resume
CronScheduler::delete_cron_job("hourly_email").await?; // remove entirely
```
To register a job that starts **paused**, override `enabled()` on the `CronJob`
impl (it defaults to `true`); enable it later from the dashboard or with
`toggle_cron_job`:
```rust
fn enabled(&self) -> bool { false }
```
A disabled job stays registered but is skipped and removed from the run schedule
until re-enabled.
### Multiple Queues
```rust
let queues = vec![
QueueConfig::new("default", 5, 0), // 5 workers, priority 0
QueueConfig::new("critical", 10, 0), // 10 workers, priority 0
QueueConfig::new("low", 2, 1), // 2 workers, priority 1
];
```
### Metrics UI
> Requires a dashboard feature — `dashboard-actix` or `dashboard-axum` (not
> enabled by default). See [Feature Flags](#feature-flags).
Access the built-in metrics dashboard at `/qrush/metrics`:
- Queue statistics and job counts
- Worker status and health
- Cron job management
- Job retry and deletion
- CSV export
## Requirements
- Rust 1.89.0 or later
- Redis 6.0 or later
- Tokio runtime (multi-threaded)
## Environment Variables
QRush itself only reads `REDIS_URL` (and only where you pass it — most APIs take
the URL explicitly). The other variables below are conventions used by the
example binaries; **your** code decides whether to read them.
```bash
# Read by qrush where a Redis URL is expected
REDIS_URL=redis://127.0.0.1:6379
# Conventions (you read these yourself — see the sections linked)
QRUSH_BASIC_AUTH=admin:password # dashboard auth — you parse it and call set_basic_auth()
RUST_LOG=info,qrush=info # tracing filter, honored by tracing_subscriber
```
> ⚠️ Setting `QRUSH_BASIC_AUTH` alone does **nothing** — the crate never reads
> it. Dashboard auth is configured programmatically; see
> [Securing the Dashboard](#securing-the-dashboard-basic-auth).
# Detailed Documentation
## Integrated Mode (Detailed)
**Use this mode when:** You want a simple setup with workers running in the same process as your web server.
### 1. Add Dependencies
```toml
[dependencies]
# Pick the dashboard framework you use: "dashboard-actix" or "dashboard-axum"
qrush = { version = "2.0.2", features = ["dashboard-actix"] }
actix-web = "4" # or: axum = "0.8"
tokio = { version = "1", features = ["rt-multi-thread", "macros"] }
serde = { version = "1", features = ["derive"] }
async-trait = "0.1"
anyhow = "1"
futures = "0.3"
```
### 2. Define a Job
```rust
use qrush::job::Job;
use async_trait::async_trait;
use serde::{Deserialize, Serialize};
use futures::future::BoxFuture;
use anyhow::Result;
#[derive(Clone, Serialize, Deserialize)]
pub struct NotifyUser {
pub user_id: String,
pub message: String,
}
#[async_trait]
impl Job for NotifyUser {
async fn perform(&self) -> Result<()> {
println!("Notify {} -> {}", self.user_id, self.message);
Ok(())
}
fn name(&self) -> &'static str { "NotifyUser" }
fn queue(&self) -> &'static str { "default" }
}
impl NotifyUser {
pub fn name() -> &'static str { "NotifyUser" }
pub fn handler(payload: String) -> BoxFuture<'static, Result<Box<dyn Job>>> {
Box::pin(async move {
let job: NotifyUser = serde_json::from_str(&payload)?;
Ok(Box::new(job) as Box<dyn Job>)
})
}
}
```
### 3. Initialize QRush
The queue/worker setup is identical for both frameworks — only the dashboard
wiring differs. The dashboard mounts at `/qrush/metrics/...` in both cases.
**Actix** (`features = ["dashboard-actix"]`):
```rust
use qrush::config::{QueueConfig, set_redis_url};
use qrush::registry::register_job;
use qrush::routes::metrics_route::qrush_metrics_routes;
use actix_web::{web, App, HttpServer};
#[actix_web::main]
async fn main() -> std::io::Result<()> {
// Set Redis URL
let redis_url = std::env::var("REDIS_URL")
.unwrap_or_else(|_| "redis://127.0.0.1:6379".to_string());
set_redis_url(redis_url.clone())?;
// Register jobs
register_job(NotifyUser::name(), NotifyUser::handler);
// Initialize queues
let queues = vec![
QueueConfig::new("default", 5, 0),
];
QueueConfig::initialize(redis_url, queues).await?;
// Start web server
HttpServer::new(|| {
App::new()
.service(web::scope("/qrush").configure(qrush_metrics_routes))
})
.bind("0.0.0.0:8080")?
.run()
.await
}
```
**Axum** (`features = ["dashboard-axum"]`):
```rust
use qrush::config::{QueueConfig, set_redis_url};
use qrush::registry::register_job;
use qrush::routes::axum_route::qrush_metrics_router;
use axum::Router;
#[tokio::main]
async fn main() -> anyhow::Result<()> {
// Set Redis URL
let redis_url = std::env::var("REDIS_URL")
.unwrap_or_else(|_| "redis://127.0.0.1:6379".to_string());
set_redis_url(redis_url.clone())?;
// Register jobs
register_job(NotifyUser::name(), NotifyUser::handler);
// Initialize queues
let queues = vec![
QueueConfig::new("default", 5, 0),
];
QueueConfig::initialize(redis_url, queues).await?;
// Mount the dashboard under /qrush
let app = Router::new().nest("/qrush", qrush_metrics_router());
let listener = tokio::net::TcpListener::bind("0.0.0.0:8080").await?;
axum::serve(listener, app).await?;
Ok(())
}
```
### 4. Enqueue Jobs
```rust
use qrush::queue::{enqueue, enqueue_in};
// Immediate
enqueue(NotifyUser {
user_id: "123".to_string(),
message: "Hello".to_string(),
}).await?;
// Delayed (300 seconds)
enqueue_in(NotifyUser {
user_id: "123".to_string(),
message: "Reminder".to_string(),
}, 300).await?;
```
---
## Separate Process Mode (Detailed)
**Use this mode when:** You want production-ready separation with workers in a dedicated process.
### 1. Create Engine Binary
Create `src/bin/qrush_engine.rs`:
```rust
use qrush::engine::{run_engine, parse_queues};
use qrush::config::set_redis_url;
use qrush::registry::register_job;
use qrush::cron::cron_scheduler::CronScheduler;
// Your job types
use your_app::jobs::NotifyUser;
#[tokio::main(flavor = "multi_thread")]
async fn main() -> anyhow::Result<()> {
dotenvy::dotenv().ok();
// Setup logging
tracing_subscriber::fmt::init();
// Get Redis URL
let redis_url = std::env::var("REDIS_URL")
.unwrap_or_else(|_| "redis://127.0.0.1:6379".to_string());
// Set Redis URL (needed before registering cron jobs)
set_redis_url(redis_url.clone())?;
// Register jobs
register_job(NotifyUser::name(), NotifyUser::handler);
// Register cron jobs (optional)
let daily_report = DailyReportJob { /* ... */ };
CronScheduler::register_cron_job(daily_report).await?;
// Parse queues from a compact spec. Each entry is
// "name[:concurrency[:priority]]" (comma-separated). Omitted fields default
// to concurrency=5, priority=0 — so "default,critical:10:0" is valid too.
let queues = parse_queues("default:5:0,critical:10:0");
// Run engine
run_engine(redis_url, queues, 5).await
}
```
### 2. Update Cargo.toml
The engine binary above uses `tracing_subscriber` for logging and `dotenvy`
to load `.env`, so add them alongside the `[[bin]]` entry:
```toml
[dependencies]
tracing-subscriber = { version = "0.3", features = ["env-filter"] }
dotenvy = "0.15"
[[bin]]
name = "qrush_engine"
path = "src/bin/qrush_engine.rs"
```
### 3. Web Server (No Workers)
In your web server `main.rs`, only register jobs for enqueueing — do **not**
call `QueueConfig::initialize` (that starts workers; the engine process owns
them here). Use whichever framework you enabled. To also expose the dashboard,
mount it exactly as shown in [Integrated Mode → Initialize QRush](#3-initialize-qrush).
**Actix** (`features = ["dashboard-actix"]`):
```rust
use qrush::config::set_redis_url;
use qrush::registry::register_job;
use qrush::routes::metrics_route::qrush_metrics_routes;
use actix_web::{web, App, HttpServer};
#[actix_web::main]
async fn main() -> std::io::Result<()> {
let redis_url = std::env::var("REDIS_URL")
.unwrap_or_else(|_| "redis://127.0.0.1:6379".to_string());
set_redis_url(redis_url).expect("failed to set Redis URL");
// Register jobs (for serialization only — no worker initialization).
register_job(NotifyUser::name(), NotifyUser::handler);
HttpServer::new(|| {
App::new().service(web::scope("/qrush").configure(qrush_metrics_routes))
})
.bind("0.0.0.0:8080")?
.run()
.await
}
```
**Axum** (`features = ["dashboard-axum"]`):
```rust
use qrush::config::set_redis_url;
use qrush::registry::register_job;
use qrush::routes::axum_route::qrush_metrics_router;
use axum::Router;
#[tokio::main]
async fn main() -> anyhow::Result<()> {
let redis_url = std::env::var("REDIS_URL")
.unwrap_or_else(|_| "redis://127.0.0.1:6379".to_string());
set_redis_url(redis_url)?;
// Register jobs (for serialization only — no worker initialization).
register_job(NotifyUser::name(), NotifyUser::handler);
let app = Router::new().nest("/qrush", qrush_metrics_router());
let listener = tokio::net::TcpListener::bind("0.0.0.0:8080").await?;
axum::serve(listener, app).await?;
Ok(())
}
```
### 4. Run Both Processes
**Terminal 1 - Web Server:**
```bash
export REDIS_URL=redis://127.0.0.1:6379
cargo run
```
**Terminal 2 - Worker Engine:**
```bash
export REDIS_URL=redis://127.0.0.1:6379
cargo run --bin qrush_engine
```
---
## Cron Expressions
QRush accepts both **6-field** (`sec min hour day month weekday`) and **5-field**
(`min hour day month weekday`) expressions. A 5-field expression defaults seconds
to `0`, so `*/5 * * * *` and `0 */5 * * * *` are equivalent.
Common examples:
| `"* * * * *"` | Every minute (5-field) |
| `"0 * * * * *"` | Every minute (6-field) |
| `"0 */5 * * * *"` | Every 5 minutes |
| `"0 0 * * * *"` | Every hour |
| `"0 0 0 * * *"` | Daily at midnight |
| `"0 30 9 * * *"` | Daily at 09:30 |
| `"0 0 0 * * 1"` | Every Monday at midnight |
| `"0 0 9 * * MON-FRI"` | Weekdays at 09:00 |
| `"0 0 0 1 * *"` | First day of every month |
| `"0 0 12 1 JAN *"` | Jan 1st at noon |
Each field supports the usual operators:
- `*` — any value
- `a` — an exact value
- `a,b,c` — a list
- `a-b` — an inclusive range
- `*/n` — a step over the whole range (e.g. `*/15` in minutes)
- `a-b/n` — a step within a range
- **Names**: months `JAN`–`DEC`, weekdays `SUN`–`SAT` (case-insensitive).
For the weekday field, both `0` and `7` mean Sunday.
**Timezone.** Expressions evaluate in UTC by default. Override per job with
`fn timezone(&self) -> &'static str` on the `CronJob` impl, returning any IANA
name (e.g. `"Asia/Kolkata"`, `"America/New_York"`) — so `"0 0 9 * * *"` fires at
09:00 in that zone, DST included.
```rust
#[async_trait]
impl CronJob for EmailJob {
fn cron_expression(&self) -> &'static str { "0 0 9 * * *" } // 9 AM…
fn cron_id(&self) -> &'static str { "morning_email" }
fn timezone(&self) -> &'static str { "Asia/Kolkata" } // …IST
}
```
**Precision & missed runs.** The scheduler ticks every ~5 seconds, so a job fires
within a few seconds of its scheduled time (don't rely on sub-5s precision). If
the scheduler was down when a run was due, that run fires once on the next tick
and is then re-anchored to its next future slot — missed cycles are **not**
backfilled one-per-cycle. Claiming is atomic in Redis, so running multiple engine
processes will **not** double-fire the same job.
## Metrics Endpoints
Paths assume the dashboard is mounted at `/qrush` (as in the examples). The
Actix and Axum adapters expose the **same** routes:
| `GET /qrush/metrics` | Dashboard overview |
| `GET /qrush/metrics/health` | Health check (returns `healthy`) |
| `GET /qrush/metrics/queues/{queue}` | Per-queue details |
| `GET /qrush/metrics/queues/{queue}/export` | Export a queue's jobs as CSV |
| `GET /qrush/metrics/extras/summary` | Aggregate metrics summary |
| `GET /qrush/metrics/extras/delayed` | Delayed (scheduled-later) jobs |
| `GET /qrush/metrics/extras/scheduled` | Scheduled jobs |
| `GET /qrush/metrics/extras/retry` | Jobs waiting to retry |
| `GET /qrush/metrics/extras/failed` | Failed jobs |
| `GET /qrush/metrics/extras/dead` | Dead-letter queue |
| `GET /qrush/metrics/extras/cron` | Cron job management |
| `POST /qrush/metrics/jobs/action` | Job actions (retry / delete) |
| `POST /qrush/metrics/cron/action` | Cron actions (run-now / toggle / delete) |
## Securing the Dashboard (Basic Auth)
The dashboard is **open by default**. To require HTTP Basic Auth, register
credentials with `set_basic_auth` **before** you start the web server. Once
credentials are set, the built-in middleware (already wired into both the Actix
and Axum routers) enforces them on every `/qrush/metrics/...` request using a
constant-time credential comparison.
```rust
use qrush::config::{set_basic_auth, QrushBasicAuthConfig};
// Read from the environment (recommended) — the crate does NOT do this for you.
if let Ok(raw) = std::env::var("QRUSH_BASIC_AUTH") {
if let Some((username, password)) = raw.split_once(':') {
set_basic_auth(Some(QrushBasicAuthConfig {
username: username.to_string(),
password: password.to_string(),
}));
}
}
// ...then mount the dashboard and start the server as usual.
```
- Call `set_basic_auth` once, during startup, before serving requests.
- Passing `None` (or never calling it) leaves the dashboard open.
- There's no env-var auto-wiring: `QRUSH_BASIC_AUTH` is only a naming
convention — you read it and call `set_basic_auth` yourself, as above.
- Basic Auth sends credentials base64-encoded, not encrypted. Terminate TLS in
front of the dashboard (reverse proxy) for anything internet-facing.
## Production Tips
- Use separate process mode for production
- Protect the dashboard with [Basic Auth](#securing-the-dashboard-basic-auth) (and put TLS in front of it)
- Configure appropriate queue concurrency based on your workload
- Monitor Redis memory usage
- Use graceful shutdown for zero-downtime deployments
- Scale workers horizontally by running multiple engine processes
## License
This project is licensed under the MIT License - see the [LICENSE](LICENSE) file for details.
## Contributing
Contributions are welcome! Please feel free to submit a Pull Request.
## Support
- **Documentation**: [docs.rs/qrush](https://docs.rs/qrush)
- **Issues**: [GitHub Issues](https://github.com/srotas-space/qrush/issues)
- **Discussions**: [GitHub Discussions](https://github.com/srotas-space/qrush/discussions)
---
Made with ❤️ by [Srotas Space](https://open-source.srotas.space)