apalis

name

apalis

description

Comprehensive guide for building background job processing, task queues, and message-driven systems with the Apalis Rust library. Use this skill whenever the user wants to: create background workers or job processors in Rust; set up task queues with Redis, PostgreSQL, SQLite, MySQL, AMQP/RabbitMQ, NATS, PGMQ, or RSMQ; process jobs asynchronously (emails, file processing, reports, webhooks); implement retry logic, timeouts, rate limiting, or job scheduling; monitor workers and job queues; integrate background processing into web frameworks (Axum, Actix-Web, Warp); build reliable message consumers or event-driven systems; handle graceful shutdown of worker processes; add middleware to job handlers (tracing, Prometheus, Sentry, panic catching); schedule jobs for future execution; use the tower middleware ecosystem with job processing; or create custom backends. Make sure to use this skill when the user mentions apalis, background jobs, task queue, job processing, message queue, worker service, async job runner, background processing in Rust, or any reliable async task execution pattern in Rust.

Apalis — Rust Background Job & Task Processing Framework

Apalis is a robust, tower-native framework for processing background jobs, tasks, and messages in Rust. It provides durable job queues, real-time message consumption, retry logic, observability, and graceful shutdown — all built on the tower::Service trait, which means the entire tower middleware ecosystem is available. Apalis provides granular backend traits (Backend, TaskSink, Metrics, Expose) that let you depend only on the capabilities you actually use.

Crate Architecture

Crate	Purpose
apalis	Main crate — worker, monitor, layers, error handling, common types
apalis-core	Core abstractions: Backend, TaskSink, Metrics, Expose, Worker, Task
apalis-redis	Redis-based durable job queue
apalis-postgres	PostgreSQL-based durable job queue
apalis-sqlite	SQLite-based durable job queue
apalis-mysql	MySQL-based durable job queue
apalis-amqp	AMQP (RabbitMQ) message queue
apalis-nats	NATS message queue
apalis-pgmq	PGMQ (PostgreSQL native message queue)
apalis-rsmq	Redis Simple Message Queue
apalis-sql	Shared SQL utilities for PostgreSQL, SQLite, MySQL backends
apalis-board	Web dashboard for monitoring and managing queues
apalis-workflow	Sequential and DAG workflow support

Quick Reference: Which Guide Do I Need?

• Setting up a specific backend (Redis, PostgreSQL, etc.) -> Read references/backends.md
• Middleware, layers, retry, tracing, Prometheus -> Read references/middleware-layers.md
• Advanced patterns: multiple job types, web framework integration, scheduling, custom backends -> Read references/patterns-advanced.md

Core Architecture

Apalis is built around tower::Service. Every job handler is a tower Service, enabling the full middleware ecosystem. The execution flow:

Monitor (coordinates workers + shutdown)
  |
  +-> Worker 1          Worker 2          ...
  |     |                  |
  |   [Layers]           [Layers]        <- tower middleware stack
  |     |                  |
  |   Service            Service         <- async handler function
  |     |                  |
  |   Backend            Backend         <- polls/dequeues jobs (Stream)
  |
  [Shutdown Signal]

Backend Trait System

v1 granularises backend capabilities into focused, composable traits:

• Backend — minimal contract: polling, heartbeating, middleware
• BackendExt — serialization: codec, compact representation, encoded polling
• TaskSink — task enqueueing: push, push_bulk, push_stream, push_task
• Metrics — observability: global(), fetch_by_queue()
• Expose — super-trait combining Metrics + ListWorkers + ListQueues + ListTasks + ListAllTasks
• MakeShared — connection sharing across multiple workers

Task<Args, Ctx, IdType> — The Job Wrapper

Every job is wrapped in a Task containing the payload and metadata:

pub struct Task<Args, Ctx, IdType> {
    args: Args,              // The actual job payload
    parts: Parts<Ctx>,       // Metadata: task_id, attempts, extensions, context
}

Job States

pub enum Status {
    Pending,     // Ready to be processed
    Scheduled,   // Scheduled for later execution
    Running,     // Currently being processed
    Done,        // Successfully completed
    Failed,      // Failed (with last_error stored)
    Retry,       // Awaiting retry
    Killed,      // Manually killed
}

Quick Start: Minimal Working Example

[dependencies]
apalis = "1.0.0-rc.7"
tokio = { version = "1", features = ["full"] }

use apalis::prelude::*;

#[derive(Debug, Clone)]
struct Email {
    to: String,
    subject: String,
}

async fn send_email(email: Email) {
    println!("Sending email to: {}", email.to);
}

#[tokio::main]
async fn main() -> Result<(), BoxDynError> {
    let mut storage = MemoryStorage::new();

    storage.push(Email {
        to: "[e-mail]".into(),
        subject: "Hello".into(),
    }).await?;

    WorkerBuilder::new("email-worker")
        .backend(storage)
        .build(send_email)
        .run()
        .await?;

    Ok(())
}

Core Patterns at a Glance

1. Define a Task and Handler

Tasks are plain Rust structs. Handlers are plain async functions — no macros required. The first argument is the job type, followed by any number of Data<T> extractors (up to 8).

use apalis::prelude::*;

#[derive(Debug, Clone)]
struct GenerateReport { user_id: u64 }

#[derive(Clone)]
struct DbPool(String);

async fn generate_report(job: GenerateReport, pool: Data<DbPool>) {
    println!("Generating report for user: {}, pool: {}", job.user_id, pool.0);
}

2. Build a Worker with Layers and Shared State

Use WorkerBuilder to construct a worker. The backend must be set first, before layers and data. Layers wrap the handler (like middleware in web frameworks). Data<T> injects shared state that handlers extract.

use apalis::prelude::*;
use apalis::layers::retry::RetryPolicy;
use std::time::Duration;

WorkerBuilder::new("report-worker")
    .backend(storage)
    .enable_tracing()
    .retry(RetryPolicy::default())
    .timeout(Duration::from_secs(60))
    .catch_panic()
    .data(DbPool("postgres://localhost".into()))
    .build(generate_report)
    .run()
    .await?;

3. Run Multiple Job Types

Each job type gets its own worker and storage. All workers register with a single Monitor. Monitor::register takes a closure |_runs: usize| that produces the worker — this enables restarts.

use apalis::prelude::*;

#[derive(Debug, Clone)]
struct SendEmail { to: String, subject: String }

#[derive(Debug, Clone)]
struct GenerateReport { user_id: u64 }

async fn handle_email(job: SendEmail) { println!("Email to: {}", job.to); }
async fn handle_report(job: GenerateReport) { println!("Report for: {}", job.user_id); }

#[tokio::main]
async fn main() -> Result<(), BoxDynError> {
    Monitor::new()
        .register(|_| {
            WorkerBuilder::new("emails")
                .backend(MemoryStorage::new())
                .build(handle_email)
        })
        .register(|_| {
            WorkerBuilder::new("reports")
                .backend(MemoryStorage::new())
                .build(handle_report)
        })
        .run()
        .await?;

    Ok(())
}

4. Schedule a Job for Later

Use TaskBuilder to build a scheduled task, then push it with push_task:

use apalis::prelude::*;

let scheduled = TaskBuilder::new(Email {
    to: "[e-mail]".into(),
    subject: "Reminder".into(),
})
.run_in_seconds(3600)
.build();

storage.push_task(scheduled).await?;

TaskBuilder methods: run_at_timestamp(u64), run_at_time(SystemTime), run_after(Duration), run_in_seconds(u64), run_in_minutes(u64), run_in_hours(u64).

5. Push Jobs from Web Handlers

Apalis workers run alongside web servers. Push jobs from HTTP handlers, then the background worker picks them up:

// In an Axum handler
async fn create_user(
    axum::Json(payload): axum::Json<CreateUser>,
    storage: axum::Extension<MemoryStorage<CreateUser>>,
) -> &'static str {
    let mut s = storage.into_inner();
    s.push(payload.into_inner()).await.unwrap();
    "User created and welcome email queued"
}

6. Error Handling in Handlers

Handlers can return (), Result<(), BoxDynError>, or any type implementing IntoResponse. To control retry vs abort behavior, return specific error types:

use apalis_core::error::{AbortError, RetryAfterError};
use std::time::Duration;

async fn fragile_handler(job: MyJob) -> Result<(), BoxDynError> {
    if job.should_abort {
        // Permanently failed — will NOT be retried
        let err = std::io::Error::new(std::io::ErrorKind::Other, "Critical failure");
        return Err(AbortError::new(err).into());
    }
    // Transient failure — WILL be retried (with optional delay)
    let err = std::io::Error::new(std::io::ErrorKind::Other, "Temporary issue");
    Err(RetryAfterError::new(err, Duration::from_secs(5)).into())
}

A third variant, DeferredError, causes an instant retry without delay.

Error Handling Guide

Error Type	Behavior	When to Use
AbortError	Permanently failed, no retry	Invalid data, missing resources, business rule violations
RetryAfterError	Retried after specified delay	Transient failures: network errors, rate limits
DeferredError	Instant retry	Recoverable conditions where delay is unnecessary
BoxDynError	Treated as transient (retryable)	Any standard error — becomes retryable

Monitor and Shutdown

The Monitor orchestrates all workers. It handles graceful shutdown, configurable shutdown timeouts, and coordinates worker events. register() takes a closure that receives the run count.

use apalis::prelude::*;
use std::time::Duration;

let monitor = Monitor::new()
    .shutdown_timeout(Duration::from_secs(30))
    .with_terminator(async {
        tokio::signal::ctrl_c().await.unwrap();
    })
    .register(|_| {
        WorkerBuilder::new("worker")
            .backend(MemoryStorage::new())
            .build(handler)
    });

monitor.run().await?;

The Shutdown signal can be shared across the application:

let shutdown = Shutdown::new();
// In another thread/task:
shutdown.start_shutdown();
// Workers check: shutdown.is_shutting_down()

FromRequest Extractor Pattern

Handlers can extract data from the task using the FromRequest trait. Built-in extractors include:

• Data<T> — Shared state injected via WorkerBuilder::data()
• TaskId<IdType> — The unique job identifier (requires explicit type parameter)
• Attempt — Current retry attempt count
• WorkerContext — Runtime worker context (use ctx.name() for worker name)

use apalis::prelude::*;
use apalis_core::task::task_id::RandomId;

async fn handler(
    job: MyJob,
    pool: Data<DbPool>,
    task_id: TaskId<RandomId>,
    attempt: Attempt,
) {
    println!("Processing job {} (attempt {})", task_id, attempt.current());
}

Observability: Metrics, Workers, Tasks

Storage backends that implement the expose traits allow programmatic inspection. Import the required traits or use apalis::prelude::* which re-exports Metrics, ListWorkers, ListTasks, ListQueues.

use apalis::prelude::*;

// Global statistics — returns Vec<Statistic>
let stats = storage.global().await?;
// Statistic { title: "PENDING_JOBS", stat_type: Number, value: "42", ... }

// List all registered workers across queues
let workers = storage.list_all_workers().await?;
// Vec<RunningWorker> { id, queue, backend, started_at, last_heartbeat, layers }

// List tasks with filtering and pagination
use apalis_core::backend::ListTasks;
let filter = Filter { status: None, page: 1, page_size: Some(50) };
let pending = storage.list_tasks("MyJob", &filter).await?;

This enables building custom dashboards or alerting systems without apalis-board.

Key Design Principles

1. Macro-free handlers — Handlers are plain async functions. No proc macros, no attribute magic.

2. Tower-native — Every handler is a tower::Service. The entire tower middleware ecosystem (tracing, retry, timeout, rate limiting) works out of the box.

3. Stream-based backends — Any type implementing Stream<Item = Result<Option<T>, Error>> can be a backend. This includes channels, database listeners, WebSocket connections, or custom sources.

4. Type-safe jobs — Job types are statically typed through generics. Each storage type is parameterized by its job type, preventing accidental mixing.

5. Granular backend traits — Backend, TaskSink, Metrics, Expose are separate traits. Depend only on what you use — a function that only pushes tasks bounds on TaskSink, not the full Backend.

6. Shared connections — Use MakeShared to share a single backend connection across multiple workers.

Common Pitfalls

1. Missing Clone on shared data — Any type passed via Data<T> must implement Clone. Wrap expensive resources (connection pools) in Arc if needed.

2. Blocking in async handlers — Never use .block_on() or synchronous I/O inside an async handler. Use tokio::task::spawn_blocking for CPU-bound or blocking operations.

3. Forgetting the retry layer — If you want automatic retry, you must add .retry(RetryPolicy::default()) to the worker. Without it, errors just mark the job as failed permanently.

4. Not setting concurrency — Default concurrency is 1. For high-throughput workloads, set .concurrency(n) and .parallelize(tokio::spawn).

5. Ignoring shutdown timeout — If jobs take a long time and the process receives a shutdown signal, in-flight jobs may be interrupted. Set an appropriate shutdown_timeout on the Monitor.

6. TaskId must be explicit — TaskId is generic over IdType. Use TaskId<RandomId> for the default ID type.

Feature Flags (apalis crate)

Feature	Default	Description
tracing	yes	Structured tracing for every job execution
retry	yes	Retry failed jobs with configurable policy
timeout	yes	Time out long-running jobs
limit	yes	Rate limit job processing
catch-panic	yes	Catch panics and convert to errors
filter	no	Filter jobs based on a predicate
sentry	no	Sentry exception and performance monitoring
prometheus	no	Prometheus metrics export
opentelemetry	no	OpenTelemetry metrics

Reference Files

• references/backends.md — Setup and configuration for all 9 backends (Memory, Redis, PostgreSQL, SQLite, MySQL, AMQP, NATS, PGMQ, RSMQ), including Cargo.toml dependencies, connection setup, and push/consume examples
• references/middleware-layers.md — All built-in layers (tracing, retry, timeout, catch-panic, rate limit, filter, Prometheus, Sentry, error handling), custom layer creation, tower integration, and Data<T> extraction
• references/patterns-advanced.md — Multiple job types, web framework integration (Axum, Actix-Web), scheduled tasks, stream-as-backend, MakeShared pattern, observability traits, custom backend implementation, TaskBuilder, FromRequest custom extractors, graceful shutdown patterns