Rust · Tokio · MIT licensed

Background jobs at Rust speed.

KioMQ is an all-in-one task queue and orchestration library for Rust. Enqueue work, process it across a pool of workers, and keep it durable with Redis — all inside the Tokio runtime you already have.

$cargo add kiomq

kiodsh queues & progress

soon kiodsh, the terminal dashboard in this recording, is coming soon.

Why KioMQ

The building blocks for background work

Inspired by BullMQ's ergonomics, implemented as an embeddable Rust library — no broker to babysit, no sidecar to deploy.

Async & sync processors

Async for I/O-bound work; blocking closures run on spawn_blocking so CPU-heavy jobs never stall the runtime.

Configurable concurrency

Each worker processes up to concurrency jobs at once — defaults to the host's logical CPU count.

Event-driven idle workers

Lock-free atomics plus Notify mean an empty queue costs almost nothing. No polling loops burning cores.

Bulk enqueue

Push thousands of jobs in one round trip with bulk_add and bulk_add_only.

Priorities, delays & cron

Run a job now, in N milliseconds, on a cron schedule, or ahead of everything else by priority score.

Retries & stalled detection

Exponential or fixed backoff, per-job attempt limits, and lock renewal that recovers jobs from dead workers.

Events & metrics built in

Subscribe to job-state events, stream progress updates, and read Tokio runtime plus per-worker timings.

Pluggable stores

Start with InMemoryStore, graduate to RedisStore when you need durability and many machines.

Quick start

A working queue in about ten lines

Create a store, wrap it in a Queue, hand a closure to a Worker. That's the whole model — the same code scales from a unit test to a Redis-backed fleet.

No macros to learn. Processors are ordinary async or sync closures.
Typed end to end. Job data, return value, and progress are your own types.
Panic-safe. A panicking job fails and retries; the process stays up.
Swap the store, keep the code. In-memory in tests, Redis in production.

Read the quick start

use std::sync::Arc;
use kiomq::{InMemoryStore, Job, KioError, Queue, Worker, WorkerOpts};

#[tokio::main]
async fn main() -> kiomq::KioResult<()> {
    let store: InMemoryStore<u64, u64, ()> = InMemoryStore::new(None, "demo");
    let queue = Queue::new(store, None).await?;

    // A processor is just a closure: (store, job) -> Result<R, KioError>
    let processor = |_store: Arc<_>, job: Job<u64, u64, ()>| async move {
        Ok::<u64, KioError>(job.data.unwrap_or_default() * 2)
    };

    let worker = Worker::new_async(&queue, processor, Some(WorkerOpts::default()))?;
    worker.run()?;

    queue
        .bulk_add_only((0..10u64).map(|i| (format!("job-{i}"), None, i)))
        .await?;

    let metrics = queue.current_metrics.clone();
    while !metrics.all_jobs_completed() {
        tokio::task::yield_now().await;
    }

    worker.close();
    Ok(())
}
use std::sync::Arc;
use kiomq::{InMemoryStore, Job, KioError, Queue, Worker, WorkerOpts};

#[tokio::main]
async fn main() -> kiomq::KioResult<()> {
    let store: InMemoryStore<u64, u64, ()> = InMemoryStore::new(None, "demo-sync");
    let queue = Queue::new(store, None).await?;

    // Sync processors run on a blocking thread via spawn_blocking —
    // use them for hashing, image work, blocking FFI, etc.
    let processor = |_store: Arc<_>, job: Job<u64, u64, ()>| {
        Ok::<u64, KioError>(job.data.unwrap_or_default() * 2)
    };

    let worker = Worker::new_sync(&queue, processor, Some(WorkerOpts::default()))?;
    worker.run()?;

    queue.add_job("compute", 42u64, None).await?;

    let metrics = queue.current_metrics.clone();
    while !metrics.all_jobs_completed() {
        tokio::task::yield_now().await;
    }

    worker.close();
    Ok(())
}
use kiomq::{Config, KioResult, Queue, RedisStore, SharedRedis};

#[tokio::main]
async fn main() -> KioResult<()> {
    // Config re-exports deadpool_redis::Config, so you can reuse
    // the pool configuration your app already has.
    let config = Config::default();
    let redis = SharedRedis::create(&config)?;

    let store = RedisStore::new(None, "my-queue", &redis).await?;
    let queue: Queue<(), (), (), _> = Queue::new(store, None).await?;

    // Every worker on every machine that points at this queue name
    // now shares the same job set, locks, and events.
    Ok(())
}
use kiomq::{JobDelay, JobOptions, Repeat};

// Run in five seconds.
let delayed = JobOptions {
    delay: JobDelay::TimeMilis(5_000),
    ..Default::default()
};

// Jump the queue: lower priority values run first.
let urgent = JobOptions { priority: 1, ..Default::default() };

// Every weekday at 07:30, forever.
let nightly = JobOptions {
    repeat: Some(Repeat::from_cron_str("30 7 * * 1-5")?),
    ..Default::default()
};

// Or every 30 seconds, at most 100 times.
let polling = JobOptions {
    repeat: Some(Repeat::repeat_every_for_times(30_000, Some(100))),
    ..Default::default()
};

queue.add_job("report", payload, Some(nightly)).await?;

Architecture

One queue, many workers, any store

Producers hand jobs to a queue; the store owns the state; workers reserve, execute, and report. Scale up by raising concurrency, scale out by pointing more processes at the same Redis queue.

Swap InMemoryStore for RedisStore and the same worker code runs on as many machines as you like.

Backends

Pick durability to match the workload

Stores are pluggable behind a single Store trait — the rest of your code doesn't change.

Store	Feature flag	Durable	Multi-process	Best for
`InMemoryStore`	always available	No	No	Tests, dev loops, short-lived in-process work
`RedisStore`	`redis-store` (default)	Yes	Yes	Production fleets spread across machines

Store trait — swap the backend per environment, keep the code.

Processor flavours — async futures and blocking closures.

CPU burned while queues sit empty, thanks to atomics + Notify.

MIT

Permissive licence. No CLA, no open-core tier.

Observability

You can see what the queue is doing

Metrics collection runs in the background with no extra setup, and every job state transition is an event you can subscribe to.

Job events. Completed, Failed, Stalled, Progress and more, per state or all at once.
Progress updates. Persist typed progress from inside a processor.
Process metrics. CPU, RSS, heap stats, and Tokio runtime counters, keyed by PID.
Worker metrics. Per-task poll counts and idle durations for latency profiling.

Metrics reference

use kiomq::{EventParameters, JobState};

// Listen to one state…
let id = queue.on(JobState::Completed, |evt| async move {
    if let EventParameters::Completed { job_id, .. } = evt {
        println!("job {job_id} completed");
    }
});

// …or match on every event the queue emits.
queue.on_all_events(|evt: EventParameters<u64, u8>| async move {
    match evt {
        EventParameters::Failed { job_id, reason, .. } => {
            eprintln!("job {job_id} failed: {}", reason.reason);
        }
        EventParameters::Progress { job_id, data } => {
            println!("job {job_id} is {data}% done");
        }
        _ => {}
    }
});

// Listeners are removable by id.
queue.remove_event_listener(id);
// Snapshots are keyed by PID, so a multi-process
// deployment stays distinguishable.
let snapshots = queue.fetch_proess_metrics().await?;

if let Some(m) = snapshots.values().next() {
    println!(
        "cpu: {:.1}%  mem: {} MB  tokio workers: {}",
        m.process_cpu_usage,
        m.memory_usage / 1_024 / 1_024,
        m.rt_metrics.workers_count,
    );
}

// Per-worker, per-task timing data.
for (worker_id, wm) in &queue.fetch_worker_metrics().await? {
    for task in &wm.tasks {
        println!(
            "worker {worker_id} polls={} idle={:?}",
            task.metrics.total_poll_count,
            task.metrics.total_idle_duration,
        );
    }
}

Ready to move work off the request path?

Add one dependency, keep your runtime. KioMQ is MIT licensed and developed in the open — issues and pull requests welcome.

Get started API reference Join the Discord