Background Jobs in Rust With Queues Workers and Retries

A signup handler that sends a welcome email inline takes 40 ms when the mail provider is healthy and 30 seconds when it is not. Worse, if the process restarts between "user row committed" and "email sent", the user never gets the message and nobody knows. Background jobs fix both problems: the request returns as soon as the work is recorded, and a separate worker does the slow part with retries. The code below compiles on Rust 1.99 with edition 2024 and tokio 1.54, and the SQL targets PostgreSQL 17.

This guide builds the idea in three layers. First an in-process queue you can run in a single file, so the retry and dead-letter rules are visible. Then the same rules against a Postgres table with FOR UPDATE SKIP LOCKED, which survives restarts. Then the operational parts that decide whether the system is trustworthy: idempotency, visibility timeouts, shutdown and monitoring.

Pick the durability before the crate

Every queue answers one question: what happens to a job when the process dies? The answer sorts the options.

Start with the cheapest option whose failure you can afford. Losing a thumbnail is a shrug. Losing a payment confirmation is an incident. Most web services land on Postgres first, for one reason: the job can be inserted in the same transaction as the change that caused it. If the transaction rolls back, the job never existed, and you cannot send "order shipped" for an order that was never saved. That property is the whole argument for a database queue, and it is why the outbox pattern exists.

The rules a queue must implement

Whatever storage you pick, four rules make a job system correct, and they are cheap to state before writing any SQL.

1. A job has an identity, a kind, a payload and a counter of attempts.
2. A worker claims a job before running it, so two workers never run the same job at the same moment.
3. A failure schedules a retry later, with a delay that grows, up to a maximum number of attempts.
4. After the last attempt the job moves to a dead state where a human can inspect it. It is never silently dropped and never retried forever.

The following program implements exactly those rules in memory. It is not durable, but it is complete and runs as pasted, which makes it the best place to see the retry arithmetic before a database gets involved. The manifest needs only tokio = { version = "1.54", features = ["full"] }.

use std::collections::VecDeque;
use std::sync::{Arc, Mutex};
use std::time::Duration;
use tokio::sync::Notify;

#[derive(Debug, Clone)]
struct Job {
    id: u32,
    kind: &'static str,
    attempts: u32,
    max_attempts: u32,
}

#[derive(Default)]
struct Queue {
    ready: Mutex<VecDeque<Job>>,
    dead: Mutex<Vec<Job>>,
    wake: Notify,
}

impl Queue {
    fn push(&self, job: Job) {
        self.ready.lock().unwrap().push_back(job);
        self.wake.notify_one();
    }
    fn pop(&self) -> Option<Job> {
        self.ready.lock().unwrap().pop_front()
    }
}

async fn handle(job: &Job) -> Result<(), String> {
    match job.kind {
        "flaky" if job.attempts < 3 => Err(format!("upstream 503 on attempt {}", job.attempts)),
        "broken" => Err("invalid payload".to_string()),
        _ => Ok(()),
    }
}

fn backoff(attempts: u32) -> Duration {
    Duration::from_millis(10 * 2u64.pow(attempts))
}

async fn worker(name: &'static str, q: Arc<Queue>) {
    loop {
        let Some(mut job) = q.pop() else {
            if tokio::time::timeout(Duration::from_millis(200), q.wake.notified()).await.is_err() {
                return;
            }
            continue;
        };
        job.attempts += 1;
        match handle(&job).await {
            Ok(()) => println!("{name}: job {} ({}) done after {} attempt(s)", job.id, job.kind, job.attempts),
            Err(e) if job.attempts >= job.max_attempts => {
                println!("{name}: job {} ({}) dead: {e}", job.id, job.kind);
                q.dead.lock().unwrap().push(job);
            }
            Err(e) => {
                let delay = backoff(job.attempts);
                println!("{name}: job {} ({}) failed: {e}; retry in {delay:?}", job.id, job.kind);
                let q2 = Arc::clone(&q);
                tokio::spawn(async move {
                    tokio::time::sleep(delay).await;
                    q2.push(job);
                });
            }
        }
    }
}

#[tokio::main]
async fn main() {
    let q = Arc::new(Queue::default());
    for (id, kind) in [(1, "email"), (2, "flaky"), (3, "broken")] {
        q.push(Job { id, kind, attempts: 0, max_attempts: 4 });
    }
    worker("w1", q.clone()).await;
    println!("dead letters: {}", q.dead.lock().unwrap().len());
}

w1: job 1 (email) done after 1 attempt(s)
w1: job 2 (flaky) failed: upstream 503 on attempt 1; retry in 20ms
w1: job 3 (broken) failed: invalid payload; retry in 20ms
w1: job 3 (broken) failed: invalid payload; retry in 40ms
w1: job 2 (flaky) failed: upstream 503 on attempt 2; retry in 40ms
w1: job 2 (flaky) done after 3 attempt(s)
w1: job 3 (broken) failed: invalid payload; retry in 80ms
w1: job 3 (broken) dead: invalid payload
dead letters: 1

Read the output as a story. Job 1 succeeds on the first try. Job 2 fails twice with a transient error and succeeds on attempt three, which is the case retries exist for. Job 3 fails every time, so after four attempts it lands in dead, which is the case dead letters exist for. The delay doubles with each attempt (20, 40, 80 ms), and the Mutex guards are dropped before any .await, which matters: holding a std::sync::MutexGuard across an await point makes the future non-Send and the compiler refuses to spawn it.

Two details in worker are worth copying. The let Some(mut job) = q.pop() else { ... } form keeps the happy path unindented. And the idle wait is a Notify with a timeout rather than a sleep loop, so a pushed job wakes a worker immediately instead of after a polling interval. Notify::notify_one stores a permit when nobody is waiting, so a push that lands between pop returning None and notified() being awaited is not lost.

A durable queue in Postgres

The in-memory version forgets everything on restart. Moving the same rules into a table makes them durable. The table needs the same fields the struct had, plus a run_at so a retry can be scheduled in the future and a locked_at so a crashed worker's claim can expire.

CREATE TABLE jobs (
    id           bigserial PRIMARY KEY,
    kind         text        NOT NULL,
    payload      jsonb       NOT NULL,
    status       text        NOT NULL DEFAULT 'queued'
                 CHECK (status IN ('queued', 'running', 'done', 'dead')),
    attempts     int         NOT NULL DEFAULT 0,
    max_attempts int         NOT NULL DEFAULT 5,
    run_at       timestamptz NOT NULL DEFAULT now(),
    locked_at    timestamptz,
    last_error   text,
    created_at   timestamptz NOT NULL DEFAULT now()
);

CREATE INDEX jobs_ready ON jobs (run_at) WHERE status = 'queued';

The partial index matters more than it looks. A busy queue accumulates millions of done rows, and an index over all of them makes every claim scan dead weight. Indexing only queued rows keeps the claim query fast no matter how large the table grows, as long as you also delete or archive finished jobs on a schedule.

Enqueue inside the transaction that needs it

use serde_json::json;
use sqlx::{Postgres, Transaction};

pub async fn register_user(
    tx: &mut Transaction<'_, Postgres>,
    email: &str,
) -> Result<i64, sqlx::Error> {
    let user_id: i64 = sqlx::query_scalar("INSERT INTO users (email) VALUES ($1) RETURNING id")
        .bind(email)
        .fetch_one(&mut **tx)
        .await?;

    sqlx::query("INSERT INTO jobs (kind, payload) VALUES ('welcome_email', $1)")
        .bind(json!({ "user_id": user_id }))
        .execute(&mut **tx)
        .await?;

    Ok(user_id)
}

The caller commits the transaction. If the commit fails, both rows vanish together. If it succeeds, both exist, and the job will run even if the server dies one millisecond later. Note the payload carries user_id, not the user's email: the job loads fresh data when it runs, so an email change between enqueue and execution is respected.

Claim with SKIP LOCKED

Several workers polling one table would normally queue up behind each other's row locks. FOR UPDATE SKIP LOCKED tells Postgres to skip rows another transaction has locked and take the next free one, so workers never wait on each other and never claim the same row.

use sqlx::{PgPool, Row};

pub struct Job {
    pub id: i64,
    pub kind: String,
    pub payload: serde_json::Value,
    pub attempts: i32,
    pub max_attempts: i32,
}

pub async fn claim(pool: &PgPool) -> Result<Option<Job>, sqlx::Error> {
    let row = sqlx::query(
        "UPDATE jobs
            SET status = 'running', attempts = attempts + 1, locked_at = now()
          WHERE id = (
                SELECT id FROM jobs
                 WHERE status = 'queued' AND run_at <= now()
                 ORDER BY run_at
                 FOR UPDATE SKIP LOCKED
                 LIMIT 1)
      RETURNING id, kind, payload, attempts, max_attempts",
    )
    .fetch_optional(pool)
    .await?;

    Ok(row.map(|r| Job {
        id: r.get("id"),
        kind: r.get("kind"),
        payload: r.get("payload"),
        attempts: r.get("attempts"),
        max_attempts: r.get("max_attempts"),
    }))
}

The subquery picks and locks one row, the outer UPDATE flips it to running and increments attempts, and RETURNING hands the row back in a single round trip. Because the claim is one statement, there is no window where a job is selected but not yet marked.

Finish, retry or bury

The completion step mirrors the in-memory match. Success marks the row done. A failure under the limit puts it back in queued with a later run_at. A failure at the limit buries it.

use sqlx::PgPool;

pub async fn finish(pool: &PgPool, job: &Job, result: Result<(), String>) -> Result<(), sqlx::Error> {
    match result {
        Ok(()) => {
            sqlx::query("UPDATE jobs SET status = 'done', locked_at = NULL WHERE id = $1")
                .bind(job.id)
                .execute(pool)
                .await?;
        }
        Err(e) if job.attempts >= job.max_attempts => {
            sqlx::query("UPDATE jobs SET status = 'dead', last_error = $2, locked_at = NULL WHERE id = $1")
                .bind(job.id)
                .bind(e)
                .execute(pool)
                .await?;
        }
        Err(e) => {
            // 5s, 10s, 20s, 40s: the delay doubles with each attempt.
            let delay_secs = 5.0 * 2f64.powi(job.attempts - 1);
            sqlx::query(
                "UPDATE jobs
                    SET status = 'queued', last_error = $2, locked_at = NULL,
                        run_at = now() + make_interval(secs => $3)
                  WHERE id = $1",
            )
            .bind(job.id)
            .bind(e)
            .bind(delay_secs)
            .execute(pool)
            .await?;
        }
    }
    Ok(())
}

Production retry delays should carry jitter. If a provider is down for ten minutes and a thousand jobs all failed at the same second, a pure doubling schedule brings them all back at the same second too, and the recovering provider is hit by a synchronized wave. Add a random fraction of the delay, commonly up to 20 percent.

The worker loop

With claim and finish in place, the worker is a loop that claims, runs under a timeout and finishes. Handlers return Result<(), String> here to keep the example small; a real service would use its own error enum and decide per variant whether a retry can help. A malformed payload will fail identically on every attempt, so retrying it five times only delays the dead letter.

use std::time::Duration;
use tokio::time::{sleep, timeout};

pub async fn run_worker(pool: sqlx::PgPool) {
    loop {
        match claim(&pool).await {
            Ok(Some(job)) => {
                let result = match timeout(Duration::from_secs(60), handle(&job)).await {
                    Ok(r) => r,
                    Err(_) => Err("handler timed out after 60s".to_string()),
                };
                if let Err(e) = finish(&pool, &job, result).await {
                    tracing::error!(job_id = job.id, "could not record result: {e}");
                }
            }
            Ok(None) => sleep(Duration::from_secs(1)).await,
            Err(e) => {
                tracing::error!("claim failed: {e}");
                sleep(Duration::from_secs(5)).await;
            }
        }
    }
}

async fn handle(job: &Job) -> Result<(), String> {
    match job.kind.as_str() {
        "welcome_email" => {
            let user_id = job.payload["user_id"].as_i64().ok_or("payload has no user_id")?;
            // load the user, render the template, call the mail provider
            tracing::info!(user_id, "sending welcome email");
            Ok(())
        }
        other => Err(format!("unknown job kind {other}")),
    }
}

The timeout is not decoration. A handler that waits forever on a socket holds its worker hostage, and with four workers, four stuck jobs stop the whole queue silently. tokio::time::timeout cancels the future at its next await point, which is good enough for I/O-bound work and useless for a handler that spins on the CPU without awaiting, so CPU-heavy jobs belong on spawn_blocking or a separate process.

Polling once a second is crude but adequate for most services: an idle worker costs one cheap indexed query per second. If you need sub-second pickup, LISTEN and NOTIFY can wake workers on insert, with the poll kept as a safety net because notifications are lost if nobody is listening when they fire.

At least once means twice sometimes

No queue gives exactly-once execution. Consider a worker that sends the email, then loses its database connection before it can mark the row done. The row stays running, a reaper eventually resets it, and the next worker sends the email again. The provider call succeeded and the bookkeeping failed, and no protocol can distinguish that from the provider call failing.

So jobs must be idempotent: running one twice must be safe. There are three common ways to get there.

• Make the operation naturally repeatable. "Set the user's plan to pro" is idempotent. "Add 10 credits" is not.
• Record the effect with a unique key. Insert into a sent_emails table with a unique constraint on (user_id, template) before sending, and skip when the insert conflicts.
• Pass the provider an idempotency key derived from the job ID. Payment and email APIs that accept one will deduplicate for you.

The second approach is the most portable. The failing version, a handler that sends first and records afterward, is the one that double-sends:

// Wrong order: a crash after send() and before the insert sends again.
mailer.send(&message).await?;
sqlx::query("INSERT INTO sent_emails (user_id, template) VALUES ($1, 'welcome')")
    .bind(user_id).execute(pool).await?;

// Safer: claim the right to send first, then send. A unique violation means
// another run already did, so the job can finish without sending.
let claimed = sqlx::query(
    "INSERT INTO sent_emails (user_id, template) VALUES ($1, 'welcome')
     ON CONFLICT DO NOTHING",
)
.bind(user_id)
.execute(pool)
.await?
.rows_affected();

if claimed == 1 {
    mailer.send(&message).await?;
}

This version trades one failure for another: a crash after the insert and before the send means the email is never sent. Which risk you prefer depends on the job. For a welcome email, a rare missed send is usually cheaper than a duplicate. For charging a card, pass the provider an idempotency key and let it make the call safe to repeat.

Rescuing jobs from crashed workers

A worker that is killed mid-job leaves its row in running forever. The locked_at column exists so a reaper can detect that. Run this on a timer, once a minute is plenty:

UPDATE jobs
   SET status = CASE WHEN attempts >= max_attempts THEN 'dead' ELSE 'queued' END,
       last_error = 'worker lost: lock expired',
       locked_at = NULL
 WHERE status = 'running'
   AND locked_at < now() - interval '10 minutes';

The interval must be longer than your handler timeout, otherwise the reaper hands a slow but alive job to a second worker and you have manufactured a duplicate. With a 60 second handler timeout, ten minutes leaves a wide margin, and a crash costs ten minutes of delay for that one job.

Shutting down without losing work

On deploy, Kubernetes or systemd sends SIGTERM and waits. A worker that exits immediately abandons its running job to the reaper, so the job waits ten minutes for no good reason. The fix is to stop claiming new jobs on the signal and let the current one finish.

use tokio_util::sync::CancellationToken;

async fn run_worker(pool: sqlx::PgPool, stop: CancellationToken) {
    while !stop.is_cancelled() {
        match claim(&pool).await {
            Ok(Some(job)) => { /* run and finish as before */ }
            _ => {
                tokio::select! {
                    _ = stop.cancelled() => break,
                    _ = tokio::time::sleep(std::time::Duration::from_secs(1)) => {}
                }
            }
        }
    }
}

Wire a tokio::signal::unix::signal(SignalKind::terminate()) future to call stop.cancel(), then join the worker tasks. Give the orchestrator a grace period at least as long as your handler timeout, or it will SIGKILL the process mid-job and you are back to the reaper.

Ready-made crates

Writing the queue yourself is about 200 lines, which is why many teams do. The crates are worth considering when you want features beyond that: apalis offers workers, middleware and several backends; sqlxmq is a Postgres queue built on the same SKIP LOCKED idea; graphile_worker is a Rust port of a well-known Node worker. Judge them on the same checklist as your own code: do they retry with backoff, bury after a limit, expire dead locks, and let you enqueue in your own transaction? The last one rules out several otherwise capable options.

Choosing attempts and delays with a worked example

Retry settings are a budget, and you can compute it. Say the mail provider has an outage that typically lasts eight minutes, and you want a welcome email to survive one. With a base delay of 5 seconds doubling each attempt, the waits are 5, 10, 20, 40, 80 and 160 seconds. The cumulative time before attempt seven is 315 seconds, about five minutes, so six retries do not cover the outage. Raise the base to 15 seconds and the waits are 15, 30, 60, 120, 240 and 480 seconds, a cumulative 945 seconds, almost sixteen minutes. Same shape, longer reach, and now a typical outage is absorbed by the retries rather than ending up in the dead state.

The opposite mistake is retrying a job that cannot succeed. A welcome_email job whose payload names a user who was deleted will fail on every attempt. Classify errors at the handler: a transient error (timeout, 503, connection reset) returns a retryable result, while a permanent one (missing record, validation failure) goes straight to dead on the first attempt. In code that is one enum with two variants and a match in finish, and it removes the most common source of dead-letter noise.

pub enum JobError {
    /// Worth trying again: the world may be different in a minute.
    Transient(String),
    /// Will fail identically every time: bury it now.
    Permanent(String),
}

// In finish(): a Permanent error skips the retry branch entirely.
// Err(JobError::Permanent(e)) => mark_dead(pool, job, e).await?,

Testing the queue without a clock

Retry logic is the part nobody wants to test because it involves time. Two moves make it testable. First, keep the delay calculation a pure function, like backoff in the in-memory program, so a unit test can assert the schedule directly. Second, in database tests, set run_at in the past instead of sleeping. A test that inserts a job, claims it, finishes it with an error and then asserts status = 'queued' and run_at > now() exercises the whole retry path in milliseconds.

fn backoff_secs(attempts: u32, base: u64) -> u64 {
    base * 2u64.pow(attempts - 1)
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn delay_doubles_each_attempt() {
        let schedule: Vec<u64> = (1..=6).map(|a| backoff_secs(a, 15)).collect();
        assert_eq!(schedule, [15, 30, 60, 120, 240, 480]);
        assert_eq!(schedule.iter().sum::<u64>(), 945);
    }
}

A second test worth writing is the concurrency one. Insert 50 jobs, start eight workers, and assert that every job ran exactly once. If SKIP LOCKED is missing from the claim query this test fails with duplicates or deadlocks within a few runs, which is a far cheaper way to find out than a production incident.

Priorities, fairness and poison jobs

A single FIFO queue lets one noisy tenant starve everyone else: a customer who imports 100,000 rows enqueues 100,000 jobs, and every other customer's password reset waits behind them. Two cheap fixes exist. Add a priority column and order the claim by priority DESC, run_at, so interactive work like password resets jumps the line. Or run separate queues per class of work, with their own workers, so a flood of imports can only fill the import workers. Separate queues are the sturdier answer because priority alone still lets the bulk jobs occupy every worker at the moment an urgent one arrives.

Then there is the poison job: one that crashes the worker process itself, for example by exhausting memory or hitting a bug that panics outside the handler's catch. The row is claimed, the worker dies, the reaper requeues it, the next worker dies. The attempts counter is what saves you, because claim increments it before the handler runs. After max_attempts crashes the reaper sees attempts >= max_attempts and buries the job. This is the reason to increment on claim, not on failure.

What to measure

A queue you cannot see is a queue that fills up quietly. Export four numbers and alert on the second:

• Jobs queued, per kind
• Age of the oldest queued job, which is the real health signal because a growing backlog and a stuck worker look identical in a count
• Jobs dead, which should be near zero and always reviewed by a person
• Handler duration at the 95th percentile, to size the timeout

The age query is one line, SELECT now() - min(run_at) FROM jobs WHERE status = 'queued' AND run_at <= now(), and alerting when it passes a few minutes catches stuck workers, a failed deploy and an overloaded provider with the same rule.

When a database queue is the wrong tool

Postgres handles a few thousand jobs a minute without strain, and many services never leave that range. Past that, the constant churn of updates produces dead tuples that autovacuum has to keep up with, and the queue starts competing with your primary workload for the same connections and I/O. If you are enqueuing tens of thousands of jobs a second, or you need fan-out to many independent consumers, or you want delayed jobs and rate limits built in, move to Redis streams or a broker. Keep the same four rules; only the storage changes. And if the work is genuinely disposable, such as warming a cache, a bare tokio::spawn is simpler and the honest choice.