Apalis

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

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: "test@example.com".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: "user@example.com".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

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)

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

Known Issues in the Apalis Skill (v1.0.0-rc.7)

This document records real bugs, limitations, and inaccuracies found while testing the skill against apalis 1.0.0-rc.7 (May 2026). Each entry includes a reproduction, the fix, and the rationale.

---

Issue 1: Custom FromRequest extractors don't compile with .build()

Severity: High

Description: Implementing FromRequest for a custom extractor works in isolation, but using that extractor as a handler parameter with WorkerBuilder::build() fails with deep type inference errors. The built-in extractors (Data<T>, TaskId<RandomId>, Attempt, WorkerContext) work correctly.

Reproduction:

use apalis_core::service_fn::FromRequest;
use apalis::prelude::*;

struct ApiClient(reqwest::Client);

impl<Args, Ctx> FromRequest<Args, Ctx> for ApiClient {
    async fn from_request(req: &Task<Args, Ctx, ()>) -> Result<Self, BoxDynError> {
        Ok(ApiClient(reqwest::Client::new()))
    }
}

// This fails to compile:
WorkerBuilder::new("worker")
    .backend(storage)
    .build(|job: MyJob, api: ApiClient| async move { Ok(()) })

Error: Deep type mismatch in IntoWorkerService bounds — the custom FromRequest impl doesn't satisfy the trait solver constraints that .build() requires.

Workaround: Use Data<T> for custom shared state instead:

WorkerBuilder::new("worker")
    .data(reqwest::Client::new())
    .backend(storage)
    .build(|job: MyJob, client: Data<reqwest::Client>| async move {
        client.get("https://api.example.com").send().await?;
        Ok::<_, BoxDynError>(())
    })

Likely cause: This is likely a bug in the v1.0.0-rc.7 IntoWorkerService macro/impl that doesn't correctly handle non-built-in FromRequest extractors. May be fixed in a future rc release.

---

Issue 2: MemoryStorage does not implement Clone

Severity: Medium

Description: MemoryStorage<T> does not implement Clone, which means it cannot be shared between a producer and a consumer when using Monitor::register() closures or web framework integration.

Impact: You cannot do this:

let storage = MemoryStorage::new();
storage.push(job).await?;
Monitor::new().register(|_| {
    WorkerBuilder::new("w").backend(storage).build(handler) // ERROR: moved
})

Workaround: Create MemoryStorage inside the closure for the worker, or use a Clone-able backend (Redis, Postgres, SQLite, MySQL) when you need to share:

// Option A: Create inside closure
Monitor::new().register(|_| {
    WorkerBuilder::new("w")
        .backend(MemoryStorage::new())
        .build(handler)
})

// Option B: Use a Clone-able backend
let storage = RedisStorage::new(conn);
let mut s = storage.clone();
s.push(job).await?;
Monitor::new().register(|_| {
    WorkerBuilder::new("w").backend(storage).build(handler)
})

---

Issue 3: SqliteStorage requires 3 generic parameters

Severity: Medium

Description: SqliteStorage::new() and SqliteStorage::setup() now require 3 generic parameters: <T, Codec, Fetcher>. The old 1-parameter form doesn't work.

Old API (broken):

let storage = SqliteStorage::<MyJob>::new(pool);
SqliteStorage::<MyJob>::setup(pool).await?;

New API (correct):

let storage = SqliteStorage::<MyJob, (), ()>::new(&pool);
SqliteStorage::<(), (), ()>::setup(&pool).await?;

Key changes:

• 3 type params: SqliteStorage<T, Codec, Fetcher>
• new() takes &Pool<Sqlite> (reference), not owned Pool
• setup() takes &Pool<Sqlite> (reference)

---

Issue 4: PostgresStorage and MysqlStorage take pool references

Severity: Medium

Description: SQL storage constructors changed from taking owned pools to references. This is a breaking change from patterns documented elsewhere.

Correct API:

let storage = PostgresStorage::new(&pool);  // &pool, not pool
let storage = MysqlStorage::new(&pool);     // &pool, not pool
let storage = SqliteStorage::<T, (), ()>::new(&pool); // &pool, not pool

---

Issue 5: RetryPolicy is not in the prelude

Severity: Low

Description: RetryPolicy must be imported explicitly — it is not re-exported from apalis::prelude::*.

Fix:

use apalis::layers::retry::RetryPolicy;

---

Issue 6: Monitor::register() now takes a closure

Severity: High

Description: Monitor::register() changed from accepting a worker directly to accepting a closure |_runs: usize| -> Worker. This closure enables worker restarts.

Old API (broken):

Monitor::new().register(WorkerBuilder::new("w").backend(s).build(handler))

New API (correct):

Monitor::new().register(|_| {
    WorkerBuilder::new("w").backend(storage).build(handler)
})

---

Issue 7: Request<T, Ctx> renamed to Task<Args, Ctx, IdType>

Severity: High

Description: The job wrapper type was renamed and now has 3 type parameters instead of 2.

Old: Request<T, Ctx> New: Task<Args, Ctx, IdType> where IdType defaults to the backend's ID type.

This affects any code that directly manipulates the task/request type (custom layers, FromRequest impls, etc.).

---

Issue 8: TaskId requires explicit type parameter

Severity: Low

Description: TaskId is now generic over the ID type. You must specify TaskId<RandomId> instead of just TaskId.

Fix:

use apalis_core::task::task_id::RandomId;
// In handler:
task_id: TaskId<RandomId>

---

Issue 9: WorkerContext::id() renamed to WorkerContext::name()

Severity: Low

Description: The method for getting the worker name changed from id() to name().

Fix:

ctx.name() // instead of ctx.id()

---

Issue 10: Error::Abort and Error::Failed replaced by standalone error types

Severity: High

Description: The old Error enum with Error::Abort and Error::Failed variants no longer exists. Instead, use standalone error types:

• AbortError — permanently fails the job
• RetryAfterError — retries after a delay
• DeferredError — instant retry

New API:

use apalis_core::error::{AbortError, RetryAfterError, DeferredError};

// Abort:
Err(AbortError::new(std::io::Error::new(std::io::ErrorKind::Other, "Critical")).into())

// Retry after delay:
Err(RetryAfterError::new(err, Duration::from_secs(5)).into())

// Instant retry:
Err(DeferredError::new(err).into())

---

Issue 11: storage.schedule() removed

Severity: Medium

Description: The Storage::schedule() method was removed. Use TaskBuilder + push_task() instead.

Old API (broken):

storage.schedule(job, timestamp).await?;

New API (correct):

let task = TaskBuilder::new(job).run_at_timestamp(timestamp).build();
storage.push_task(task).await?;

---

Issue 12: Backend crate names changed

Severity: High

Description: Backend crates are now separate crates, not apalis-sql with feature flags.

Old: apalis-sql = { version = "0.7", features = ["sqlite", "migrate"] } New:

apalis-sqlite = "1.0.0-rc.7"
apalis-postgres = { version = "1.0.0-rc.7", features = ["migrate"] }
apalis-mysql = "1.0.0-rc.7"

Similarly, message queue backends are all at 1.0.0-rc.* — there is no 0.7 release for apalis-amqp, apalis-nats, apalis-pgmq, or apalis-rsmq.

---

Summary Table

---

Environment Used for Verification

• Rustc: 1.95.0 (May 2026)
• Apalis crates tested: apalis 1.0.0-rc.7, apalis-redis 1.0.0-rc.7, apalis-sqlite 1.0.0-rc.7, apalis-postgres 1.0.0-rc.7
• External services: Redis (Docker), PostgreSQL (Docker), SQLite (:memory:)
• Test project: /test-rc7/ with 20 verified test binaries

apalis-skills

A comprehensive skill that enables LLM-based coding assistants to accurately use and build solutions with the [Apalis](https://github.com/apalis-dev/apalis) Rust library — a robust, tower-native background job processing and message queue framework.

Overview

Apalis is a Rust-first framework for building reliable, scalable background job processors and message-driven systems. It provides durable job queues, real-time message consumption, retry logic, observability, and graceful shutdown — all built on the tower::Service trait, giving access to the entire tower middleware ecosystem.

• Website: apalis.dev
• GitHub: github.com/apalis-dev/apalis
• Target version: v1.0.0-rc.7
• License: MIT OR Apache-2.0

Installation

npx skills add melonask/apalis-skills

What This Skill Covers

Core Concepts

• Backend, TaskSink, Metrics, Expose traits and when to use each
• Task<Args, Ctx, IdType> job wrapper with Parts metadata
• Job lifecycle states (Pending, Scheduled, Running, Done, Failed, Retry, Killed)
• Monitor orchestration and graceful shutdown
• WorkerBuilder API and FromRequest extractors (Data<T>, TaskId<RandomId>, Attempt, WorkerContext)
• TaskBuilder fluent API for constructing tasks with scheduling and metadata

9 Backends

• Storage (durable): Redis, PostgreSQL, SQLite, MySQL, in-Memory
• MessageQueue (fire-and-forget): AMQP/RabbitMQ, NATS, PGMQ, RSMQ
• Each with Cargo.toml dependencies, connection setup, and push/consume examples

Middleware Layers

• Tracing (default), Retry with configurable policy, Timeout, Catch-panic, Rate limiting, Filtering
• Custom tower layer creation
• Recommended layer composition order for production

Advanced Patterns

• Multiple job types with separate workers and storages
• Axum integration with shared storage
• Scheduled tasks via TaskBuilder + push_task()
• Stream-as-backend (any Stream can be a job source)
• MakeShared for sharing backends across workers
• Expose traits for programmatic observability
• Polling strategies (Interval, Backoff, Multi, Stream)
• Graceful shutdown coordination with custom signals

Error Handling

• AbortError — permanently fail (no retry)
• RetryAfterError — retry after specified delay
• DeferredError — instant retry
• BoxDynError — treated as transient (retryable)

File Structure

apalis/
├── SKILL.md                          # Main skill file (core concepts, patterns, error handling)
├── README.md                         # This file
├── known-issues.md                   # Verified bugs, limitations, and workarounds for v1.0.0-rc.7
└── references/
    ├── backends.md                   # All 9 backends: setup, config, and usage examples
    ├── middleware-layers.md          # Built-in layers, custom layers, Data<T> extraction
    └── patterns-advanced.md          # Web integration, scheduling, observability, shutdown

Quick Example

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: "user@example.com".into(), subject: "Hi".into() }).await?;

    Monitor::new()
        .register(|_| {
            WorkerBuilder::new("emailer")
                .backend(MemoryStorage::new())
                .build(send_email)
        })
        .run()
        .await?;

    Ok(())
}

Trigger Phrases

This skill activates when the user's request involves any of the following:

• Background workers or job processors in Rust
• Task queues or message queues (Redis, PostgreSQL, SQLite, MySQL, AMQP, NATS, PGMQ, RSMQ)
• Asynchronous job processing (emails, file processing, reports, webhooks)
• Retry logic, timeouts, rate limiting, or job scheduling in Rust
• Monitoring workers and job queues
• Integrating background processing into web frameworks (Axum, Actix-Web)
• Building message consumers or event-driven systems
• Graceful shutdown of worker processes
• Tower middleware with job processing
• The apalis crate or any of its backend crates

Key Design Decisions

• SKILL.md under 500 lines — follows progressive disclosure; detailed docs live in references/
• Trigger-optimized description — uses a "pushy" description that lists many trigger phrases to maximize accurate activation
• Code verified against v1.0.0-rc.7 — all examples tested with 20 binaries in a real project using Redis, PostgreSQL, and SQLite backends
• Known issues documented — 12 verified issues with workarounds, including a custom FromRequest limitation

License

This skill is provided as-is for educational and development purposes. Apalis is licensed under MIT OR Apache-2.0.

Apalis Backends Reference

This file covers all available backends in the apalis ecosystem, including setup, configuration, and usage patterns for each.

Choosing a Backend

---

MemoryStorage (Built-in)

The in-memory backend. Best for testing, prototyping, and lightweight workloads where durability is not needed. Jobs are lost on process restart.

Setup

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

Usage

use apalis::prelude::*;

#[derive(Debug, Clone)]
struct MyJob { id: u64 }

let mut storage = MemoryStorage::new();
storage.push(MyJob { id: 1 }).await.unwrap();

WorkerBuilder::new("memory-worker")
    .backend(storage)
    .build_fn(handle_job)

Important Notes

• MemoryStorage is generic over the job type: MemoryStorage<YourJobType>
• Jobs do NOT survive process restarts
• Suitable for single-worker, single-process scenarios
• MemoryStorage is not Clone — for use with Monitor::register, create the storage inside the closure

---

RedisStorage

Durable job queue backed by Redis. Provides persistence, atomic operations, and supports scheduling. Requires a Redis server.

Setup

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

Connection and Usage

use apalis::prelude::*;
use apalis_redis::RedisStorage;
use serde::{Deserialize, Serialize};

#[derive(Debug, Serialize, Deserialize)]
struct EmailJob { to: String, body: String }

#[tokio::main]
async fn main() -> Result<(), BoxDynError> {
    let redis_url = std::env::var("REDIS_URL").expect("REDIS_URL must be set");
    let conn = apalis_redis::connect(redis_url.as_str()).await?;

    let storage: RedisStorage<EmailJob> = RedisStorage::new(conn);

    let mut s = storage.clone();
    s.push(EmailJob {
        to: "user@example.com".into(),
        body: "Hello!".into(),
    }).await?;

    WorkerBuilder::new("redis-email-worker")
        .backend(storage)
        .build_fn(|job: EmailJob| async move {
            println!("Sending to: {}", job.to);
            Ok::<_, BoxDynError>(())
        })
        .run()
        .await;

    Ok(())
}

Configuration Tips

• Set REDIS_URL environment variable (e.g., redis://127.0.0.1:6379)
• Jobs are stored as JSON in Redis hashes and sorted sets
• RedisStorage implements Clone

---

PostgresStorage

Durable job queue backed by PostgreSQL. Supports both standard polling and trigger-based (NOTIFY/LISTEN) modes for lower latency.

Setup

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

Connection and Usage

use apalis::prelude::*;
use apalis_postgres::PostgresStorage;
use apalis_postgres::PgPool;
use serde::{Deserialize, Serialize};

#[derive(Debug, Serialize, Deserialize, Clone)]
struct ReportJob { user_id: u64 }

#[tokio::main]
async fn main() -> Result<(), BoxDynError> {
    let database_url = std::env::var("DATABASE_URL").expect("DATABASE_URL must be set");
    let pool = PgPool::connect(&database_url).await?;

    // Run migrations to create required tables
    PostgresStorage::<(), (), ()>::setup(&pool).await?;

    let storage: PostgresStorage<ReportJob> = PostgresStorage::new(&pool);

    let mut s = storage.clone();
    s.push(ReportJob { user_id: 42 }).await?;

    WorkerBuilder::new("pg-report-worker")
        .backend(storage)
        .build_fn(|job: ReportJob| async move {
            println!("Generating report for user: {}", job.user_id);
            Ok::<_, BoxDynError>(())
        })
        .run()
        .await;

    Ok(())
}

Named Queues

Use new_with_config to separate job types into different tables:

use apalis_sql::Config;

let email_storage = PostgresStorage::new_with_config(&pool, &Config::new("email_jobs"));
let report_storage = PostgresStorage::new_with_config(&pool, &Config::new("report_jobs"));

Important Notes

• Always call PostgresStorage::<(), (), ()>::setup(&pool) before creating storage — this runs database migrations
• The migrate feature is required for setup() to work
• PostgresStorage::new() takes &PgPool (a reference), not PgPool
• PostgresStorage implements Clone

---

SqliteStorage

Durable job queue backed by SQLite. Best for embedded or single-process applications. Supports standard polling and event-driven (hook) modes.

Setup

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

Connection and Usage

use apalis::prelude::*;
use apalis_sqlite::SqliteStorage;
use apalis_sqlite::SqlitePool;
use serde::{Deserialize, Serialize};

#[derive(Debug, Serialize, Deserialize, Clone)]
struct Task { name: String }

#[tokio::main]
async fn main() -> Result<(), BoxDynError> {
    let pool = SqlitePool::connect("sqlite::memory:").await?;
    SqliteStorage::<(), (), ()>::setup(&pool).await?;

    let mut storage = SqliteStorage::<Task, (), ()>::new(&pool);
    storage.push(Task { name: "backup".into() }).await?;

    WorkerBuilder::new("sqlite-worker")
        .backend(storage)
        .build_fn(|job: Task| async move {
            println!("Running task: {}", job.name);
            Ok::<_, BoxDynError>(())
        })
        .run()
        .await?;

    Ok(())
}

Important Notes

• Always call SqliteStorage::<(), (), ()>::setup(&pool) before creating storage
• SqliteStorage now takes 3 generic parameters: SqliteStorage<T, Codec, Fetcher>. Use SqliteStorage::<T, (), ()> for the standard setup
• SqliteStorage::new() takes &Pool<Sqlite> (a reference)
• SQLite has limited concurrency — keep worker count low (1-2)

---

MysqlStorage

Durable job queue backed by MySQL. Similar API to PostgresStorage.

Setup

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

Usage

use apalis::prelude::*;
use apalis_mysql::MysqlStorage;
use apalis_mysql::MysqlPool;
use serde::{Deserialize, Serialize};

#[derive(Debug, Serialize, Deserialize, Clone)]
struct Job { data: String }

#[tokio::main]
async fn main() -> Result<(), BoxDynError> {
    let pool = MysqlPool::connect("mysql://user:pass@localhost/db").await?;
    MysqlStorage::<(), (), ()>::setup(&pool).await?;

    let storage: MysqlStorage<Job> = MysqlStorage::new(&pool);

    let mut s = storage.clone();
    s.push(Job { data: "test".into() }).await?;

    WorkerBuilder::new("mysql-worker")
        .backend(storage)
        .build_fn(|job: Job| async move {
            println!("Processing: {}", job.data);
            Ok::<_, BoxDynError>(())
        })
        .run()
        .await;

    Ok(())
}

---

AmqpBackend (RabbitMQ)

Message queue backend for AMQP (RabbitMQ). Fire-and-forget messaging with ack/nack support. Best for event-driven architectures.

Setup

[dependencies]
apalis = "1.0.0-rc.7"
apalis-amqp = "1.0.0-rc.7"
tokio = { version = "1", features = ["full"] }
serde = { version = "1", features = ["derive"] }
lapin = "2"

Connection and Usage

use apalis::prelude::*;
use apalis_amqp::AmqpBackend;
use serde::{Deserialize, Serialize};

#[derive(Debug, Serialize, Deserialize)]
struct Event { event_type: String, payload: String }

#[tokio::main]
async fn main() -> Result<(), BoxDynError> {
    let amqp_url = std::env::var("AMQP_URL").expect("AMQP_URL must be set");
    let conn = lapin::Connection::connect(&amqp_url, Default::default()).await?;
    let channel = conn.create_channel().await.unwrap();

    let storage = AmqpBackend::new_with_config(channel, "my_queue");

    storage.enqueue(Event {
        event_type: "user.created".into(),
        payload: r#"{"user_id": 123}"#.into(),
    }).await.unwrap();

    WorkerBuilder::new("amqp-consumer")
        .backend(storage)
        .build_fn(|event: Event| async move {
            println!("Received event: {}", event.event_type);
            Ok::<_, BoxDynError>(())
        })
        .run()
        .await;

    Ok(())
}

Important Notes

• AMQP backends implement MessageQueue, not Storage — no scheduling, retry, or status tracking
• RabbitMQ handles message durability and acknowledgments
• Failed jobs are nack'd back to the queue (based on RabbitMQ requeue policy)

---

NatsBackend

Message queue backend for NATS. Lightweight, high-throughput pub/sub messaging.

Setup

[dependencies]
apalis = "1.0.0-rc.7"
apalis-nats = "1.0.0-rc.7"
tokio = { version = "1", features = ["full"] }
serde = { version = "1", features = ["derive"] }
async-nats = "0.38"

Usage

use apalis::prelude::*;
use apalis_nats::NatsBackend;
use serde::{Deserialize, Serialize};

#[derive(Debug, Serialize, Deserialize)]
struct Notification { user_id: u64, message: String }

#[tokio::main]
async fn main() -> Result<(), BoxDynError> {
    let nats_url = std::env::var("NATS_URL").unwrap_or_else(|_| "nats://127.0.0.1:4222".into());
    let client = async_nats::connect(nats_url).await?;

    let storage = NatsBackend::new_with_config(client, "notifications");

    storage.enqueue(Notification {
        user_id: 1,
        message: "Welcome!".into(),
    }).await.unwrap();

    WorkerBuilder::new("nats-worker")
        .backend(storage)
        .build_fn(|notification: Notification| async move {
            println!("Notifying user {}: {}", notification.user_id, notification.message);
            Ok::<_, BoxDynError>(())
        })
        .run()
        .await;

    Ok(())
}

---

PgmqBackend

PostgreSQL-native message queue. Uses PGMQ extension for lightweight messaging without a separate broker.

Setup

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

Usage

use apalis::prelude::*;
use apalis_pgmq::PgmqBackend;
use serde::{Deserialize, Serialize};

#[derive(Debug, Serialize, Deserialize)]
struct LogEntry { level: String, message: String }

#[tokio::main]
async fn main() -> Result<(), BoxDynError> {
    let pool = sqlx::PgPool::connect("postgres://user:pass@localhost/db").await?;
    let storage = PgmqBackend::new(pool, "log_queue");

    storage.enqueue(LogEntry {
        level: "INFO".into(),
        message: "Application started".into(),
    }).await.unwrap();

    Monitor::new()
        .register(|_| {
            WorkerBuilder::new("pgmq-worker")
                .backend(storage)
                .build_fn(|entry: LogEntry| async move { Ok(()) })
        })
        .run()
        .await?;

    Ok(())
}

---

RsmqBackend

Redis Simple Message Queue backend. Lightweight message queue backed by Redis.

Setup

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

Usage

use apalis::prelude::*;
use apalis_rsmq::RsmqBackend;
use serde::{Deserialize, Serialize};

#[derive(Debug, Serialize, Deserialize)]
struct Message { content: String }

#[tokio::main]
async fn main() -> Result<(), BoxDynError> {
    let client = rsmq_async::Rsmq::new(Default::default()).await.unwrap();
    let storage = RsmqBackend::new(client, "my_queue");

    storage.enqueue(Message { content: "Hello RSMQ".into() }).await.unwrap();

    Monitor::new()
        .register(|_| {
            WorkerBuilder::new("rsmq-worker")
                .backend(storage)
                .build_fn(|msg: Message| async move { Ok(()) })
        })
        .run()
        .await?;

    Ok(())
}

---

Storage vs MessageQueue: Quick Decision

Apalis Middleware & Layers Reference

Apalis leverages the tower::Service trait for its handler abstraction, which means the entire tower middleware ecosystem is available. Layers wrap around the handler function, allowing you to add cross-cutting concerns (tracing, retry, timeout, metrics) without modifying handler code.

How Layers Work

Layers are applied in the WorkerBuilder and wrap the handler in order. The backend must be set first. The execution flows from outer to inner:

Request from Backend
    |
[Layer 1: TraceLayer]        <- outermost (added first)
    |
[Layer 2: RetryLayer]
    |
[Layer 3: TimeoutLayer]
    |
[Layer 4: CatchPanicLayer]   <- innermost (added last)
    |
Handler Function

The response (and errors) flow back through the layers in reverse. A RetryLayer would see errors from inner layers and decide whether to retry the entire call chain.

Adding Layers to a Worker

Builder Methods (WorkerBuilderExt)

WorkerBuilderExt Convenience Methods

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

WorkerBuilder::new("worker")
    .backend(storage)
    .concurrency(2)
    .enable_tracing()
    .retry(RetryPolicy::default())
    .timeout(Duration::from_secs(30))
    .catch_panic()
    .rate_limit(100, Duration::from_secs(1))
    .data(shared_state)
    .build(handler)

Using .chain() with tower::ServiceBuilder

use tower::ServiceBuilder;

WorkerBuilder::new("worker")
    .chain(|builder| {
        builder
            .enable_tracing()
            .retry(RetryPolicy::default())
            .timeout(Duration::from_secs(30))
    })
    .backend(storage)
    .build(handler)

---

Built-in Layers

TraceLayer (default feature)

Provides structured tracing for every job execution. Enabled by default.

WorkerBuilder::new("worker")
    .backend(storage)
    .enable_tracing()
    .build(handler)

RetryLayer (default feature)

Automatically retries failed jobs based on a configurable policy.

use apalis::layers::retry::RetryPolicy;
use apalis::layers::retry::RetryLayer;

// Default retry policy
WorkerBuilder::new("worker")
    .backend(storage)
    .retry(RetryPolicy::default())
    .build(handler)

Configure retry with a fixed number of retries and a conditional:

use apalis::prelude::*;
use apalis::layers::retry::RetryPolicy;

WorkerBuilder::new("worker")
    .backend(storage)
    .retry(
        RetryPolicy::retries(3)
            .retry_if(|e: &BoxDynError| e.downcast_ref::<apalis::layers::catch_panic::PanicError>().is_none()),
    )
    .build(handler)

Key points:

• Only RetryAfterError triggers retry with a delay. Other errors retry instantly.
• AbortError permanently fails the job.
• Retry count is tracked per-job via the Attempt field in Parts.

TimeoutLayer (default feature)

Cancels long-running jobs after a specified duration.

use std::time::Duration;

WorkerBuilder::new("worker")
    .backend(storage)
    .timeout(Duration::from_secs(60))
    .build(slow_handler)

CatchPanicLayer (default feature)

Converts panics into errors, preventing the worker from crashing. Should be the innermost layer.

use apalis::layers::catch_panic::CatchPanicLayer;

WorkerBuilder::new("worker")
    .backend(storage)
    .layer(CatchPanicLayer::new())
    .build(panic_prone_handler)

With a custom panic handler:

use apalis::layers::catch_panic::CatchPanicLayer;
use apalis::layers::catch_panic::PanicError;

WorkerBuilder::new("worker")
    .backend(storage)
    .layer(CatchPanicLayer::with_panic_handler(|e| {
        println!("Caught panic: {:?}", e);
        PanicError("Custom handler".to_string())
    }))
    .build(handler)

RateLimitLayer (default feature)

Limits the rate of job processing.

use std::time::Duration;

WorkerBuilder::new("worker")
    .backend(storage)
    .rate_limit(100, Duration::from_secs(1))
    .build(handler)

FilterLayer (feature: filter)

Filters jobs based on a predicate. Jobs that don't match are rejected. Uses tower's FilterLayer, so the closure must return Result<Task, BoxError>.

WorkerBuilder::new("worker")
    .backend(storage)
    .filter(|req: Task<MyJob, _, _>| {
        if req.args.priority > 5 {
            Ok(req)
        } else {
            Err(Box::new(std::io::Error::new(std::io::ErrorKind::Other, "Low priority")) as BoxDynError)
        }
    })
    .build(handler)

Note: The filter closure receives Task<Args, Ctx, IdType>, not a direct reference to the job payload. Access the payload via req.args.

ErrorHandlingLayer

Built into apalis-core. Wraps the handler service and converts errors into BoxDynError.

use apalis_core::error::ErrorHandlingLayer;

WorkerBuilder::new("worker")
    .layer(ErrorHandlingLayer::new())
    .backend(storage)
    .build(handler)

---

Composing Layers: Recommended Order

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

WorkerBuilder::new("production-worker")
    .backend(storage)
    .concurrency(4)
    .enable_tracing()
    .retry(RetryPolicy::default())
    .timeout(Duration::from_secs(60))
    .catch_panic()
    .data(shared_state)
    .build(handler)

Why this order?

• TraceLayer on the outside sees every attempt (including retries)
• RetryLayer wraps TimeoutLayer, so timeouts are treated as failures eligible for retry
• CatchPanicLayer is closest to the handler so it catches panics before they reach retry logic

---

Custom Tower Layers

Any tower::Layer compatible middleware works. Here is an example logging layer:

use std::task::{Context, Poll};
use tower::{Layer, Service};
use apalis_core::task::Task;

#[derive(Clone)]
struct LoggingService<S> {
    inner: S,
}

impl<S, Args, Ctx, IdType> Service<Task<Args, Ctx, IdType>> for LoggingService<S>
where
    S: Service<Task<Args, Ctx, IdType>, Error = BoxDynError>,
{
    type Response = S::Response;
    type Error = BoxDynError;
    type Future = S::Future;

    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
        self.inner.poll_ready(cx)
    }

    fn call(&mut self, req: Task<Args, Ctx, IdType>) -> Self::Future {
        println!("Processing task: {:?}", req.parts.task_id);
        self.inner.call(req)
    }
}

#[derive(Clone)]
struct LoggingLayer;

impl<S> Layer<S> for LoggingLayer {
    type Service = LoggingService<S>;

    fn layer(&self, inner: S) -> Self::Service {
        LoggingService { inner }
    }
}

// Usage
WorkerBuilder::new("worker")
    .backend(storage)
    .layer(LoggingLayer)
    .build(handler)

---

Data<T> Extraction Pattern

Shared state is injected into handlers via Data<T>. Use WorkerBuilder::data() to provide the state, and Data<T> as a handler parameter to extract it.

use apalis::prelude::*;

#[derive(Clone)]
struct AppState {
    db: String,
    email_client: String,
}

async fn handler(job: MyJob, state: Data<AppState>) {
    println!("DB: {}, API: {}", state.db, state.email_client);
}

// Setup
let state = AppState { db: "postgres://localhost".into(), email_client: "https://api.example.com".into() };
WorkerBuilder::new("worker")
    .data(state)
    .backend(storage)
    .build(handler)

Important rules for `Data<T>`:

• The type must implement Clone (use Arc for expensive resources)
• Extract up to 8 Data<T> parameters per handler
• Data<T> is extracted from the Extensions in the task Task

---

Feature Flags Summary

# Minimal (most features enabled by default)
apalis = "1.0.0-rc.7"

# Full observability stack
apalis = { version = "1.0.0-rc.7", features = ["filter", "sentry", "prometheus", "opentelemetry"] }

Apalis Advanced Patterns Reference

This document covers real-world integration patterns, advanced configurations, and production-ready setups for apalis v1.0.0-rc.7.

Multiple Job Types

Each job type requires its own worker and storage instance. Use Monitor to run all workers together. Each storage is generic over its job type, providing type safety.

use apalis::prelude::*;

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

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

#[derive(Debug, Clone)]
struct CleanupTask { days_old: u64 }

async fn handle_email(job: SendEmail) { println!("Email to: {}", job.to); }
async fn handle_report(job: GenerateReport) { println!("Report for: {}", job.user_id); }
async fn handle_cleanup(job: CleanupTask) { println!("Cleanup: {} days old", job.days_old); }

#[tokio::main]
async fn main() -> Result<(), BoxDynError> {
    Monitor::new()
        .register(|_| {
            WorkerBuilder::new("email-worker")
                .concurrency(4)
                .enable_tracing()
                .backend(MemoryStorage::new())
                .build(handle_email)
        })
        .register(|_| {
            WorkerBuilder::new("report-worker")
                .concurrency(2)
                .timeout(std::time::Duration::from_secs(300))
                .backend(MemoryStorage::new())
                .build(handle_report)
        })
        .register(|_| {
            WorkerBuilder::new("cleanup-worker")
                .concurrency(1)
                .backend(MemoryStorage::new())
                .build(handle_cleanup)
        })
        .run()
        .await?;

    Ok(())
}

Note: MemoryStorage does not implement Clone, so it cannot be shared outside the register closure. Use MemoryStorage::new() inside each closure, or use a Clone-able backend (Redis, Postgres, SQLite, MySQL) if you need to push jobs from elsewhere.

---

Web Framework Integration

Axum Integration

Run apalis workers alongside an Axum HTTP server. The worker runs in a background tokio::spawn, and a Clone-able backend is shared with route handlers via Arc.

use apalis::prelude::*;
use axum::{routing::post, Router, Extension, Json};
use apalis_redis::RedisStorage;

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

async fn handle_email(job: SendEmail) { println!("Sending to: {}", job.to); }

async fn queue_email(
    Json(payload): Json<SendEmail>,
    Extension(storage): Extension<RedisStorage<SendEmail>>,
) -> &'static str {
    let mut s = storage;
    s.push(payload).await.unwrap();
    "Email queued"
}

#[tokio::main]
async fn main() -> Result<(), BoxDynError> {
    let conn = apalis_redis::connect("redis://127.0.0.1:6379").await?;
    let storage = RedisStorage::new(conn);

    // Spawn the worker in the background
    let worker_storage = storage.clone();
    tokio::spawn(async move {
        Monitor::new()
            .register(|_| {
                WorkerBuilder::new("email-worker")
                    .concurrency(4)
                    .enable_tracing()
                    .backend(worker_storage.clone())
                    .build(handle_email)
            })
            .run()
            .await
            .unwrap();
    });

    // Run the Axum server
    let app = Router::new()
        .route("/email", post(queue_email))
        .layer(Extension(storage));

    let listener = tokio::net::TcpListener::bind("0.0.0.0:3000").await?;
    axum::serve(listener, app).await?;
    Ok(())
}

Why Redis here? MemoryStorage doesn't implement Clone, so it can't be shared between the Axum router and the worker. Redis (and other SQL backends) implement Clone, making them suitable for shared use.

---

Scheduled Tasks

Using TaskBuilder

v1.0.0-rc.7 provides TaskBuilder for scheduling jobs. Build a task with timing metadata, then push it with push_task:

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

let scheduled = TaskBuilder::new(Email {
    to: "user@example.com".into(),
    subject: "Reminder".into(),
})
.run_in_seconds(3600)
.build();

storage.push_task(scheduled).await?;

TaskBuilder Methods

Combining with Web Handlers

async fn schedule_reminder(
    Json(payload): Json<SendEmail>,
    storage: Extension<RedisStorage<SendEmail>>,
) -> &'static str {
    let task = TaskBuilder::new(payload)
        .run_in_minutes(30)
        .build();
    storage.push_task(task).await.unwrap();
    "Email scheduled for 30 minutes from now"
}

---

Observability: Metrics, Workers, Tasks

Storage backends implement expose traits (Metrics, ListWorkers, ListTasks, ListQueues) for programmatic inspection. Import via apalis::prelude::*.

use apalis::prelude::*;

// Global statistics — returns Vec<Statistic>
let stats = storage.global().await?;

// List all registered workers across queues
let workers = storage.list_all_workers().await?;

// 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?;

Building a Health Check Endpoint

async fn health_check(
    Extension(storage): Extension<PostgresStorage<ReportJob>>,
) -> Json<serde_json::Value> {
    use apalis_core::backend::Metrics;
    match storage.global().await {
        Ok(stats) => Json(serde_json::json!({ "status": "ok", "stats": stats })),
        Err(e) => Json(serde_json::json!({ "status": "error", "error": e.to_string() })),
    }
}

---

Sharing Backends

SQL and Redis backends implement Clone, so the same pool can be shared between producers and consumers:

let storage = RedisStorage::new(conn);
let storage_clone = storage.clone();

// Push jobs using one clone
let mut s = storage_clone;
s.push(MyJob { id: 1 }).await.unwrap();

// Build worker with the other clone
WorkerBuilder::new("worker")
    .backend(storage)
    .build(handler)
    .run()
    .await;

For SQL backends, MakeShared provides connection sharing across multiple workers:

use apalis_core::backend::MakeShared;

let shared = storage.make_shared();

// Each worker gets its own connection from the shared pool
Monitor::new()
    .register(|_| {
        let conn = shared.clone();
        WorkerBuilder::new("worker-1")
            .backend(conn)
            .build(handler)
    })
    .register(|_| {
        let conn = shared.clone();
        WorkerBuilder::new("worker-2")
            .backend(conn)
            .build(handler)
    })
    .run()
    .await?;

---

Polling Strategies

Apalis provides configurable polling strategies for how workers fetch jobs from the backend. Strategies control the balance between responsiveness and resource usage.

IntervalStrategy

Poll at a fixed interval:

use apalis_core::backend::poll_strategy::strategies::IntervalStrategy;
use std::time::Duration;

let strategy = IntervalStrategy::new(Duration::from_millis(500));

BackoffStrategy

Poll at increasing intervals when no jobs are available (reduces CPU usage during idle periods):

use apalis_core::backend::poll_strategy::strategies::{IntervalStrategy, BackoffConfig, BackoffStrategy};
use std::time::Duration;

let inner = IntervalStrategy::new(Duration::from_millis(100));
let backoff_config = BackoffConfig::new(Duration::from_secs(10))
    .with_multiplier(2.0)
    .with_jitter(0.1);

let strategy = BackoffStrategy::new(inner, backoff_config);

MultiStrategy

Combine multiple strategies (races between them):

use apalis_core::backend::poll_strategy::StrategyBuilder;

let strategy = StrategyBuilder::new()
    .add(IntervalStrategy::new(Duration::from_millis(100)))
    .add(IntervalStrategy::new(Duration::from_millis(500)))
    .build();

StreamStrategy

Use an external stream as the polling source:

use apalis_core::backend::poll_strategy::strategies::StreamStrategy;

let strategy = StreamStrategy::new(my_notification_stream);

---

Graceful Shutdown Patterns

Custom Shutdown Signal

use apalis::prelude::*;

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?;

Shared Shutdown Across Application

use apalis::prelude::*;

let shutdown = Shutdown::new();
let shutdown_clone = shutdown.clone();

// Spawn monitor
tokio::spawn(async move {
    Monitor::new()
        .with_terminator(async move { shutdown_clone.start_shutdown() })
        .register(|_| {
            WorkerBuilder::new("worker")
                .backend(MemoryStorage::new())
                .build(handler)
        })
        .run()
        .await
        .unwrap();
});

// In another part of the app:
shutdown.start_shutdown();

Shutdown in Web Applications

When running workers alongside a web server, coordinate their shutdown:

let shutdown = Shutdown::new();
let web_shutdown = shutdown.clone();

// Graceful shutdown for Axum — trigger when server stops
tokio::spawn(async move {
    web_shutdown.start_shutdown();
});

// Worker respects the same shutdown signal
Monitor::new()
    .with_terminator(async move { shutdown.start_shutdown() })
    .register(|_| {
        WorkerBuilder::new("worker")
            .backend(MemoryStorage::new())
            .build(handler)
    })
    .run()
    .await?;

---

Stream as Backend

Any type that implements Stream<Item = Result<Option<T>, Error>> can serve as a backend. This includes channels, database listeners, WebSocket connections, or any async source.

use apalis::prelude::*;
use futures::channel::mpsc;

#[derive(Debug, Clone)]
struct LogEntry { message: String }

async fn handle_log(entry: LogEntry) { println!("Log: {}", entry.message); }

#[tokio::main]
async fn main() -> Result<(), BoxDynError> {
    let (tx, rx) = mpsc::channel::<Result<Option<LogEntry>, BoxDynError>>(100);

    tx.send(Ok(Some(LogEntry { message: "App started".into() }))).await?;
    tx.send(Ok(Some(LogEntry { message: "User logged in".into() }))).await?;

    WorkerBuilder::new("log-worker")
        .backend(rx)
        .build(handle_log)
        .run()
        .await?;

    Ok(())
}

---

Pushing Bulk Jobs

let jobs = vec![
    SendEmail { to: "a@b.com".into(), subject: "A".into() },
    SendEmail { to: "c@d.com".into(), subject: "C".into() },
    SendEmail { to: "e@f.com".into(), subject: "E".into() },
];

// Use push_bulk for batch operations (when available)
for job in jobs {
    storage.push(job).await.unwrap();
}

For MessageQueue backends, use enqueue directly:

// storage.enqueue(job).await.unwrap();

---

Vacuum: Cleaning Up Old Jobs

Periodically clean up completed and failed jobs to prevent database bloat:

let removed = storage.vacuum().await.unwrap();
println!("Removed {} old jobs", removed);

Run on a schedule in production:

tokio::spawn(async move {
    let mut interval = tokio::time::interval(Duration::from_secs(86400));
    loop {
        interval.tick().await;
        if let Err(e) = storage.vacuum().await {
            eprintln!("Vacuum failed: {}", e);
        }
    }
});

---

WorkerBuilder Method Reference