Background Jobs

Persistent job queue backed by PostgreSQL, MySQL, or SQLite

Background jobs let you defer work to run outside the request cycle. Sending emails, processing uploads, generating reports — anything that shouldn't block an HTTP response.

Rapina's job system uses your existing PostgreSQL, MySQL, or SQLite database as the queue. No Redis, no RabbitMQ, no extra infrastructure. Jobs are rows in a rapina_jobs table and are claimed by in-process workers using a strategy suited to each database.

This page covers setup, defining jobs, enqueuing, running the worker, and the retry system.

Cron Scheduler vs Background Jobs

tl;dr: Use Background Jobs for durable, transactional work that must complete reliably. Use the Cron Scheduler for lightweight, periodic tasks that are safe to miss if the server restarts.

Prerequisites

Enable one of Rapina's database features. Background jobs support PostgreSQL, MySQL 8.0+, and SQLite 3.35+:

[dependencies]
rapina = { version = "0.13.1", features = ["postgres"] }
# Or use `features = ["mysql"]` or `features = ["sqlite"]`.

You also need a database connection configured in your app — see the Database page.

Setup

Run the CLI command from your project root:

rapina jobs init

This adds the framework's create_rapina_jobs migration to your src/migrations/mod.rs. If the file doesn't exist yet, it creates one. If the migration is already configured, it skips silently.

The result looks like this:

use rapina::jobs::create_rapina_jobs;

mod m20260315_000001_create_users;

rapina::migrations! {
    create_rapina_jobs,
    m20260315_000001_create_users,
}

The framework migration uses a zero timestamp (m00000000_000000_) so it always sorts before your application migrations, regardless of their dates.

Next time your app starts and runs migrations, the rapina_jobs table will be created.

Defining a Job

Use the #[job] macro to define a handler. The first argument is always the payload — a struct that implements Serialize + DeserializeOwned. Remaining arguments are dependency-injected from AppState via State<T> or Db.

use rapina::prelude::*;

#[derive(Serialize, Deserialize)]
pub struct WelcomeEmailPayload {
    pub email: String,
}

#[job(queue = "emails", max_retries = 5)]
async fn send_welcome_email(
    payload: WelcomeEmailPayload,
    mailer: State<Mailer>,
) -> JobResult {
    mailer.send(&payload.email).await?;
    Ok(())
}

The macro generates a send_welcome_email(payload) -> JobRequest helper used for enqueuing.

Enqueuing Jobs

Use the Jobs extractor in HTTP handlers to dispatch jobs.

#[post("/users")]
async fn create_user(body: Json<CreateUserRequest>, jobs: Jobs) -> Result<StatusCode> {
    jobs.enqueue(send_welcome_email(WelcomeEmailPayload {
        email: body.email.clone(),
    })).await?;

    Ok(StatusCode::CREATED)
}

For transactional enqueue — where the job row commits atomically with your business logic — use enqueue_with:

let txn = db.conn().begin().await?;
let user = User::insert(&txn, &body).await?;
jobs.enqueue_with(&txn, send_welcome_email(WelcomeEmailPayload {
    email: user.email.clone(),
})).await?;
txn.commit().await?;

If the transaction rolls back, the job is never created.

Starting the Worker

Call .jobs() on the application builder before .listen(). One worker spawns per subscribed queue, in-process alongside the HTTP server, and each shuts down gracefully on SIGINT/SIGTERM, finishing its current batch before stopping.

use rapina::jobs::JobConfig;

Rapina::new()
    .with_database(db_config).await?
    .jobs(JobConfig::default())
    .listen("127.0.0.1:3000")
    .await

All options have sensible defaults. Override only what you need:

JobConfig::default()
    .poll_interval(Duration::from_secs(2))
    .batch_size(20)
    .queues(["default", "emails", "heavy"])
    .job_timeout(Duration::from_secs(60))

Per-queue overrides

The options above apply to every queue. When one queue needs different pacing, register a QueueConfig for it. Any field you leave unset inherits the global value:

use rapina::jobs::{JobConfig, QueueConfig};

JobConfig::default()
    .batch_size(20)
    .queues(["default", "emails"])
    .queue_configs([
        QueueConfig::new("browser")
            .batch_size(1)
            .job_timeout(Duration::from_secs(300)),
    ])

That spawns three workers. default and emails claim up to 20 jobs per cycle with a 30s lease; browser claims one at a time with a five-minute lease, while still polling on the global 5s interval because poll_interval was not overridden.

This is the knob for the job_timeout sizing problem described below: rather than raising the global timeout to cover your slowest handler, give the slow queue batch_size(1) and a long lease, and leave everything else fast.

Each worker polls, claims and reaps only its own queue, so queues are isolated: a backlog on one does not delay another, and a worker never touches a lease belonging to a queue it does not subscribe to. That last property is what makes it safe to run workers for different queues as separate processes against the same database.

Registering a config for a queue that queues does not list is harmless. The entry is simply unused, which lets one JobConfig be narrowed per process with different queues() calls.

Running Workers as a Separate Process

The worker above runs inside the HTTP app. To scale workers independently, or to give one queue a process of its own, call run_worker from a second binary. It shares nothing with the API but the database: no listener, no port, no healthcheck.

// src/bin/worker.rs
use rapina::database::DatabaseConfig;
use rapina::jobs::{JobConfig, QueueConfig, run_worker};
use rapina::prelude::*;
use std::time::Duration;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    load_dotenv();
    TracingConfig::new().init();

    let db = DatabaseConfig::from_env()?.connect().await?;

    run_worker(
        db,
        JobConfig::default()
            .queues(["browser"])
            .queue_configs([QueueConfig::new("browser")
                .batch_size(1)
                .job_timeout(Duration::from_secs(300))]),
    )
    .await;

    Ok(())
}

cargo run --bin worker        # locally
cargo build --bin worker      # for a worker service image

run_worker returns once every worker has drained, so SIGINT/SIGTERM behaves exactly as it does in-process: each worker finishes the batch it is executing, then stops.

Because each worker claims and reaps only its own queue, the API and any number of worker processes can point at the same database without contending over leases. A typical split gives the API the fast queues through .jobs() and one process per heavy queue.

Two constraints are worth knowing before you deploy this:

Handlers must be linked into the binary that runs them. #[job] registers handlers with inventory at link time, so the dispatch table is whatever the final executable contains, not whatever the crate declares. A binary that does not compile your #[job] functions finds no handler for any job_type and permanently fails every job it claims. In a project whose jobs live in modules of main.rs, a file under src/bin/ is its own crate root and sees none of them; expose them through a src/lib.rs (or a workspace crate) that both binaries use, which is also what lets your HTTP handlers keep calling the typed enqueue helpers.

Only Db resolves in handler arguments. run_worker builds its state from the connection you pass it, so a handler taking State<T> has nothing to receive and fails every attempt until its retries are exhausted. Keep those jobs on the in-process worker started by .jobs(), where the full AppState is available.

Job Lifecycle

pending → running → completed
                  ↘ failed   (or back to pending if retries remain)

The worker transitions each claimed job from pending to running while preventing duplicate claims. PostgreSQL uses a CTE with FOR UPDATE SKIP LOCKED. MySQL uses a transaction that selects rows with FOR UPDATE SKIP LOCKED, updates them, and then fetches the updated rows because MySQL does not support UPDATE ... RETURNING. SQLite uses one UPDATE ... WHERE id IN (...) ... RETURNING statement and relies on its single-writer model.

On completion the job moves to completed or failed.

A worker that dies mid-job (crash, SIGKILL, power loss) leaves the row in running with a stale locked_until. On each poll cycle the worker reaps expired leases in its own queue only: jobs with retry budget left return to pending and are claimed again on the next cycle (immediately, with no backoff delay), with the crashed run counted toward attempts exactly like an errored run, and jobs past max_retries are marked failed with a lease-expired last_error. Recovery latency is bounded by job_timeout + poll_interval. A panicking handler does not stop the worker: the panic is caught, logged at error level, and processed through the same retry path as an error. Under panic = abort there is nothing to catch, the process dies and the job is recovered like any other crash.

Size job_timeout for the batch's worst case, batch_size times your slowest handler, or keep batches small. You can also set both per queue with a QueueConfig so one slow queue does not force a long lease on the rest. The lease is stamped for the whole batch when it is claimed, but jobs run one at a time, so a job near the end of a large batch can start with its lease already gone and run a second time on another worker. There is no heartbeat to renew the lease mid-run.

Failed jobs are retried according to the retry_policy set on the handler.

Exponential backoff (default)

#[job(max_retries = 5, retry_policy = "exponential", retry_delay_secs = 1.0)]
async fn send_welcome_email(payload: EmailPayload) -> JobResult { ... }

Jitter is seeded from the job's UUID so concurrent failures don't retry in lockstep. Delay is capped at one week.

Fixed delay

#[job(max_retries = 10, retry_policy = "fixed", retry_delay_secs = 30.0)]
async fn sync_inventory(payload: SyncPayload) -> JobResult { ... }

Every retry waits the same retry_delay_secs. The first retry is always immediate regardless of the configured delay.

MySQL expresses retry intervals in microseconds. Stored scheduling precision still depends on the column definition: the supplied migration uses TIMESTAMP without fractional-second precision, so it does not guarantee subsecond retry timing.

No retries

#[job(max_retries = 1, retry_policy = "none")]
async fn charge_card(payload: ChargePayload) -> JobResult { ... }

The job is permanently marked failed on the first error. Use this for operations that must not be duplicated.

DI Limitations

Job handlers run outside the request cycle. Only State<T> and Db work — they source data from AppState directly. Request-bound extractors (Context, Headers, Path, Query, CurrentUser) will fail at runtime and must not be used in job handlers.

In a worker started by run_worker the list is narrower still: that process builds no AppState, so only Db resolves and a handler taking State<T> fails every attempt.

Trace Propagation

When a job is enqueued from an HTTP handler, the request's trace_id is stored on the job row. The worker restores it into its tracing span before calling the handler, so all log lines emitted during job execution are correlated with the original HTTP request.

Table Schema

The migration creates a rapina_jobs table with the following columns:

Types shown are the PostgreSQL mapping; SeaORM emits the corresponding backend types for MySQL and SQLite.

The migration uses portable JSON rather than PostgreSQL JSONB, so PostgreSQL JSONB indexing is unavailable. PostgreSQL creates a partial index on (queue, run_at) WHERE status = 'pending'; MySQL creates a regular (queue, run_at) index; SQLite skips this index because it is a single-writer database.

Types

JobStatus

The JobStatus enum represents the lifecycle of a job:

use rapina::prelude::*;

// Available when the `database` feature is enabled
let status = JobStatus::Pending;
println!("{status}"); // "pending"

let parsed: JobStatus = "running".parse().unwrap();

JobStatus implements Display, FromStr, Serialize, Deserialize, Hash, Copy, and Eq. The string representation is always lowercase.

JobRow

JobRow is a plain struct that maps directly to a row in the rapina_jobs table. It implements SeaORM's FromQueryResult so you can use it with raw queries:

use rapina::database::{Db, DbError};
use rapina::jobs::JobRow;
use rapina::sea_orm::{ConnectionTrait, FromQueryResult, Statement};

async fn failed_email_jobs(db: &Db) -> Result<Vec<JobRow>, DbError> {
    let rows = JobRow::find_by_statement(Statement::from_string(
        db.conn().get_database_backend(),
        "SELECT * FROM rapina_jobs WHERE queue = 'emails' AND status = 'failed'",
    ))
    .all(db.conn())
    .await
    .map_err(DbError::from)?;

    for row in &rows {
        let status = row.parse_status().unwrap();
        println!("{}: {} (attempts: {})", row.id, status, row.attempts);
    }

    Ok(rows)
}

The status field is a String because SeaORM's FromQueryResult derive doesn't support custom enum deserialization. Use parse_status() to get a typed JobStatus.

Manual Setup

If you prefer not to use the CLI, add the migration reference manually:

// src/migrations/mod.rs
use rapina::jobs::create_rapina_jobs;

mod m20260315_000001_create_users;

rapina::migrations! {
    create_rapina_jobs,
    m20260315_000001_create_users,
}

The create_rapina_jobs module is exported from the rapina crate, so there's no file to create in your project — just the use import and the macro entry.