Background jobs and queues with retries

Some work should not happen inside a web request: sending email, resizing an image, generating a report, calling a slow third-party API. If you do it inline, the user waits, a timeout loses the work, and one failure breaks the page. The standard answer is a background job queue: the request records a job and returns quickly, and a separate worker does the work later. The code and outputs below were run with Python 3 before publishing.

The moving parts

Producer: your web app, which adds a job (enqueue).
Queue: durable storage of jobs. This can be a database table, Redis, or a broker such as RabbitMQ or a cloud queue.
Worker: a process that takes a job, runs it, and records the result.
Retries: if a job fails, try again later, but not instantly and not forever.
Dead jobs: after the final attempt, park the job for a human to look at.

A working example

A queue does not have to be exotic. This one is a SQLite table. Save as jobs.py:

import json
import sqlite3

SCHEMA = """
CREATE TABLE IF NOT EXISTS jobs (
    id INTEGER PRIMARY KEY,
    kind TEXT NOT NULL,
    payload TEXT NOT NULL,
    status TEXT NOT NULL DEFAULT 'queued',   -- queued | running | done | dead
    attempts INTEGER NOT NULL DEFAULT 0,
    max_attempts INTEGER NOT NULL DEFAULT 3,
    run_at REAL NOT NULL,
    last_error TEXT
);
CREATE INDEX IF NOT EXISTS jobs_ready ON jobs (status, run_at);
"""


def connect(path=":memory:"):
    db = sqlite3.connect(path, isolation_level=None)  # we manage transactions
    db.row_factory = sqlite3.Row
    db.executescript(SCHEMA)
    return db


def enqueue(db, kind, payload, now, max_attempts=3):
    cur = db.execute(
        "INSERT INTO jobs (kind, payload, run_at, max_attempts) VALUES (?, ?, ?, ?)",
        (kind, json.dumps(payload), now, max_attempts),
    )
    return cur.lastrowid


def claim(db, now):
    """Atomically take one ready job. Two workers can never get the same row."""
    db.execute("BEGIN IMMEDIATE")
    row = db.execute(
        "SELECT * FROM jobs WHERE status = 'queued' AND run_at <= ? ORDER BY run_at, id LIMIT 1",
        (now,),
    ).fetchone()
    if row:
        db.execute(
            "UPDATE jobs SET status = 'running', attempts = attempts + 1 WHERE id = ?",
            (row["id"],),
        )
    db.execute("COMMIT")
    return row


def run_one(db, handlers, now):
    job = claim(db, now)
    if job is None:
        return None
    attempt = job["attempts"] + 1
    try:
        handlers[job["kind"]](json.loads(job["payload"]))
    except Exception as exc:  # a job must never crash the worker
        if attempt >= job["max_attempts"]:
            status, delay = "dead", 0
        else:
            status, delay = "queued", 2 ** attempt  # 2s, 4s, 8s ... backoff
        db.execute(
            "UPDATE jobs SET status = ?, run_at = ?, last_error = ? WHERE id = ?",
            (status, now + delay, repr(exc), job["id"]),
        )
        return job["id"], status
    db.execute("UPDATE jobs SET status = 'done' WHERE id = ?", (job["id"],))
    return job["id"], "done"


if __name__ == "__main__":
    db = connect()
    calls = {"flaky": 0}

    def send_email(p):
        print(f"    sent email to {p['to']}")

    def flaky(p):
        calls["flaky"] += 1
        if calls["flaky"] < 3:
            raise ConnectionError("provider timeout")
        print("    flaky job finally worked")

    def broken(p):
        raise ValueError("bad data")

    handlers = {"email": send_email, "flaky": flaky, "broken": broken}
    enqueue(db, "email", {"to": "a@example.com"}, now=0)
    enqueue(db, "flaky", {}, now=0)
    enqueue(db, "broken", {}, now=0)

    for t in range(0, 12):
        while (r := run_one(db, handlers, now=t)) is not None:
            print(f"t={t:>2}s job {r[0]} -> {r[1]}")

    print()
    for r in db.execute("SELECT id, kind, status, attempts, last_error FROM jobs"):
        print(dict(r))

Output of python3 jobs.py (time is simulated in seconds so it runs instantly):

    sent email to a@example.com
t= 0s job 1 -> done
t= 0s job 2 -> queued
t= 0s job 3 -> queued
t= 2s job 2 -> queued
t= 2s job 3 -> queued
    flaky job finally worked
t= 6s job 2 -> done
t= 6s job 3 -> dead

{'id': 1, 'kind': 'email', 'status': 'done', 'attempts': 1, 'last_error': None}
{'id': 2, 'kind': 'flaky', 'status': 'done', 'attempts': 3, 'last_error': "ConnectionError('provider timeout')"}
{'id': 3, 'kind': 'broken', 'status': 'dead', 'attempts': 3, 'last_error': "ValueError('bad data')"}

Reading the result

Job 1 (email) works first time.
Job 2 (flaky) fails twice with a timeout, then succeeds on the third attempt. A temporary problem healed itself.
Job 3 (broken) fails every time. After 3 attempts it becomes dead and stays in the table with its last_error. Retrying a bug forever helps nobody.
Between attempts the job waits: 2 seconds, then 4. That is exponential backoff: each wait is longer, so a struggling service gets room to recover instead of being hammered.

The details that matter

Claim atomically. claim selects a job and marks it running inside one BEGIN IMMEDIATE transaction, so two workers cannot pick the same row. The test test_claim_is_exclusive checks this. With PostgreSQL the usual tool is SELECT ... FOR UPDATE SKIP LOCKED, which lets many workers take different rows without blocking each other.

Retry only what can succeed. A timeout may pass; a ValueError caused by bad data will not. Real systems often separate temporary errors (retry) from permanent ones (go straight to dead).

Add jitter. If 1,000 jobs fail together and all retry after exactly 4 seconds, they fail together again. Adding a small random amount to each delay spreads them out. The example leaves it out so the output is predictable.

Make jobs idempotent. Delivery in queue systems is usually at least once: a worker may finish the work and crash before saving "done", so the job runs again. Write handlers so running twice is safe, for example with an idempotency key or a check like "email already sent for this order".

Keep payloads small. Store an id (order_id), not the whole order. The worker loads fresh data, and you avoid stale or sensitive copies in the queue.

Tests

Save as test_jobs.py:

import unittest

from jobs import claim, connect, enqueue, run_one


class JobTests(unittest.TestCase):
    def setUp(self):
        self.db = connect()

    def status(self, job_id):
        return self.db.execute("SELECT status, attempts FROM jobs WHERE id=?", (job_id,)).fetchone()

    def test_success(self):
        j = enqueue(self.db, "ok", {}, now=0)
        run_one(self.db, {"ok": lambda p: None}, now=0)
        self.assertEqual(tuple(self.status(j)), ("done", 1))

    def test_retry_then_dead(self):
        def fail(p):
            raise RuntimeError("x")
        j = enqueue(self.db, "f", {}, now=0, max_attempts=2)
        run_one(self.db, {"f": fail}, now=0)
        self.assertEqual(self.status(j)["status"], "queued")
        self.assertIsNone(run_one(self.db, {"f": fail}, now=1))  # backoff: not ready yet
        run_one(self.db, {"f": fail}, now=2)
        self.assertEqual(tuple(self.status(j)), ("dead", 2))

    def test_claim_is_exclusive(self):
        enqueue(self.db, "ok", {}, now=0)
        self.assertIsNotNone(claim(self.db, now=0))
        self.assertIsNone(claim(self.db, now=0))  # already running

    def test_future_job_waits(self):
        enqueue(self.db, "ok", {}, now=100)
        self.assertIsNone(claim(self.db, now=50))
        self.assertIsNotNone(claim(self.db, now=100))


if __name__ == "__main__":
    unittest.main()

$ python3 -m unittest -v
Ran 4 tests in 0.009s

OK

What this example does not handle

Be honest about the gaps before copying it:

Crashed workers. If a worker dies while a job is running, that row stays running forever. Real queues use a visibility timeout or heartbeat: if a running job is not finished in time, it becomes available again.
Concurrency. SQLite allows one writer at a time. Fine for learning and small apps, not for heavy parallel load.
Scheduling, priorities, rate limits, monitoring. A mature library or service gives you these.

Build or use a library?

For production, usually use a proven tool instead of writing your own: Celery or RQ (Python), BullMQ (Node.js), Sidekiq (Ruby), or a managed cloud queue. Writing a small one, as above, is excellent for understanding them. Choose based on what you already run: if you have Redis, a Redis-based queue is natural; if you only have PostgreSQL, a Postgres-backed queue is often enough at modest scale.

Common mistakes

Doing slow work inside the request.
Retrying instantly, with no backoff.
Retrying forever, with no dead state or alert.
Non-idempotent handlers under at-least-once delivery.
No visibility into the queue: nobody notices when 5,000 jobs are waiting or 50 are dead. Track queue length, age of the oldest job and failure counts.
Putting large or secret data in the payload.

Main idea: accept the request, record a durable job, return fast. Let a worker run it with limited retries, growing delays and a dead state, and write every job so that running it twice is harmless.

Backend series, Part 3 (reliable APIs): 3 of 5. Previous: Webhooks done right: verify HMAC signatures and block replays.