Qrush

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.

Feature	Default	Description
`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:

[dependencies]
qrush = "2.0.0"

To mount the dashboard, opt into one framework:

# Actix
qrush = { version = "2.0.0", features = ["dashboard-actix"] }

# Axum
qrush = { version = "2.0.0", 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.

	1.x	2.0
Dashboard default	on (Actix)	off
Enable Actix dashboard	(default)	`features = ["dashboard-actix"]`
Enable Axum dashboard	not available	`features = ["dashboard-axum"]`

# 1.x
qrush = "1.0.1"

# 2.0 — Actix (equivalent to the old default; no code changes needed)
qrush = { version = "2.0.0", features = ["dashboard-actix"] }

If you only used enqueue + workers (no dashboard), a plain qrush = "2.0.0" now pulls in less — the web stack is no longer compiled by default. See the CHANGELOG for the full list of changes.

Quick Start

Installation

Add to your Cargo.toml:

[dependencies]
qrush = "2.0.0"
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)

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 below for complete setup instructions.

Separate Process Mode

See Part 2: Separate Process Mode below for production deployment.

API Reference

Core Traits

Job: Implement this trait for your job types
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; returns the job ID
register_job(name, handler): Register a job handler
QueueConfig::initialize(redis_url, queues): Start worker pools

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 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):

# 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

Cron Jobs

use qrush::cron::cron_job::CronJob;
use qrush::cron::cron_scheduler::CronScheduler;

#[async_trait]
impl CronJob for EmailJob {
    fn cron_expression(&self) -> &'static str {
        "0 0 * * * *"  // Every hour
    }
    fn cron_id(&self) -> &'static str { "hourly_email" }
}

// Register cron job
let job = EmailJob { /* ... */ };
CronScheduler::register_cron_job(job).await?;

Multiple Queues

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.

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

# Required
REDIS_URL=redis://127.0.0.1:6379

# Optional
QRUSH_BASIC_AUTH=admin:password  # Basic auth for metrics
RUST_LOG=info,qrush=info         # Logging level

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

[dependencies]
# Pick the dashboard framework you use: "dashboard-actix" or "dashboard-axum"
qrush = { version = "2.0.0", 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

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"]):

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"]):

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

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:

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: "name:concurrency:priority"
    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:

[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.

Actix (features = ["dashboard-actix"]):

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"]):

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:

export REDIS_URL=redis://127.0.0.1:6379
cargo run

Terminal 2 - Worker Engine:

export REDIS_URL=redis://127.0.0.1:6379
cargo run --bin qrush_engine

Cron Expressions

QRush uses 6-field cron expressions (seconds, minutes, hours, day, month, weekday):

"0 * * * * *" - Every minute
"0 */5 * * * *" - Every 5 minutes
"0 0 * * * *" - Every hour
"0 0 0 * * *" - Daily at midnight
"0 0 0 * * 1" - Every Monday at midnight
"0 0 0 1 * *" - First day of month at midnight

Metrics Endpoints

GET /qrush/metrics - Dashboard overview
GET /qrush/metrics/queues/{queue} - Queue details
GET /qrush/metrics/extras/cron - Cron job management
GET /qrush/metrics/extras/workers - Worker status
GET /qrush/metrics/health - Health check
POST /qrush/metrics/jobs/action - Job actions (retry/delete)

Production Tips

Use separate process mode for production
Set QRUSH_BASIC_AUTH to protect metrics endpoints
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 file for details.

Contributing

Contributions are welcome! Please feel free to submit a Pull Request.

Support

Documentation: docs.rs/qrush
Issues: GitHub Issues
Discussions: GitHub Discussions

Made with ❤️ by Srotas Space

qrush 2.0.0

Qrush

Features

Feature Flags

Migrating from 1.x to 2.0

Quick Start

Installation

Basic Usage (Integrated Mode)

Architecture

Integrated Mode

Separate Process Mode

Documentation

Integrated Mode

Separate Process Mode

API Reference

Core Traits

Core Functions

Errors

Engine Runtime

Command-Line Interface

Examples

Runnable dashboard examples

Cron Jobs

Multiple Queues

Metrics UI

Requirements

Environment Variables

Detailed Documentation

Integrated Mode (Detailed)

1. Add Dependencies

2. Define a Job

3. Initialize QRush

4. Enqueue Jobs

Separate Process Mode (Detailed)

1. Create Engine Binary

2. Update Cargo.toml

3. Web Server (No Workers)

4. Run Both Processes

Cron Expressions

Metrics Endpoints

Production Tips

License

Contributing

Support