How to Build a Durable Job Queue on SQLite in Python With Leases, Retries, and Dead Letters

What you will build, and why a job queue needs leases

Imagine a sign-up form. When someone submits it, you want to send a welcome email, but you do not want the visitor to stare at a spinner while your server talks to a mail provider. So the web request writes a note that says “send a welcome email to [email protected]” and returns at once. A separate program, called a worker, reads those notes and does the slow work in the background. The notes are called jobs, and the place where they wait is a job queue.

Writing the happy path takes ten minutes. The interesting part is everything else. What if two workers pick up the same note and send two emails? What if a worker is killed halfway through, so the note says “in progress” forever? What if a worker is not dead but only very slow, and by the time it reports back another worker has already finished the job?

The standard answer is a lease. A worker that takes a job does not own it. It borrows the job until a deadline, like a library book. If the worker finishes, it returns the job marked done. If the worker disappears, the deadline passes and the job becomes available to someone else. Amazon’s queue service calls this a visibility timeout and describes it like this in its documentation: “The visibility timeout in standard queues prevents multiple consumers from processing the same message at the same time.” The same page then adds the catch that we will spend half of this tutorial on: “However, because of the at-least-once delivery model, Amazon SQS doesn’t guarantee that a message won’t be delivered more than once within the visibility timeout period.”

In this tutorial you will build that machinery yourself, on a single SQLite file, in a 190-line Python module. At every step you will first break something on purpose and measure the damage, then fix it and measure again. Here is what the measurements will show:

• Four processes racing to claim 20,000 jobs made 5,331 duplicate claims when the code looked first and claimed second. One SQL statement made that number zero.
• A worker killed in the middle of a job lost nothing: another worker took the job about two seconds later, when the lease ran out.
• A slow worker that came back late put a finished job back in the queue and caused a third run, until a one-line check stopped it.
• Claiming a job took 111 ms with no index and 278 ms with the obvious index once a million finished jobs sat in the table. With a partial index it took 0.13 ms.
• The finished queue handled about 3,400 trivial jobs per second, and about 420 per second when every commit was forced to disk.
• A 300-job stress run with eight workers, 30 hard crashes, jobs that stall past their lease and jobs that always fail ended with every job done exactly once or parked in a dead-letter state.

When is SQLite the right place for a queue? When all your workers run on one machine and you need a few thousand jobs per second at most, a SQLite file removes a whole service from your stack: there is no broker to install, and the queue is backed up together with your data. If you already run PostgreSQL, the usual tool is SELECT ... FOR UPDATE SKIP LOCKED. The PostgreSQL manual says of it: “Skipping locked rows provides an inconsistent view of the data, so this is not suitable for general purpose work, but can be used to avoid lock contention with multiple consumers accessing a queue-like table.” If your workers live on several machines, SQLite is the wrong tool (the limits section explains why). Everything you learn here, the lease, the token, the attempt counter and the dead letter, carries over to those systems.

Nine words to know before we start

Job. One unit of work, stored as a row: what kind of work it is and the data it needs (its payload).

Worker. A process that loops forever: claim a job, run it, report the result.

Claim. A worker’s request for the next job that nobody else is working on.

Lease. A claim with a deadline. When the deadline passes without a report, the job can be claimed again.

Attempt. One try at a job. We count an attempt every time a worker claims the job.

Lease token. A random string created at each claim. Only the holder of the current token may finish the job. Amazon’s queue calls the same thing a receipt handle and notes: “This handle is associated with the action of receiving the message, not with the message itself.”

Dead letter. A job that failed too many times. We park it in a dead state, with its last error, for a human to look at.

At-least-once delivery. Every job runs at least once and maybe more than once. A queue cannot promise “exactly once” because it cannot know whether a vanished worker finished its work before it vanished. Handlers therefore need to be idempotent: running one twice must do no more harm than running it once.

Transaction. A group of database changes that either all happen or none of them do.

Prerequisites

• Operating system: everything here was run on Windows 11. Linux and macOS should work too; the only operating-system-specific code is the kill_tree helper in Step 1, which has a branch for each. I did not run the Linux branch.
• Access: an ordinary user account. You need no admin rights and no network access.
• Software: Python 3.10 or newer (tested on 3.13.14), SQLite 3.35.0 or newer (it ships inside Python), and pytest for the test suite. Nothing else is installed.
• Prior knowledge: you can read Python functions and classes, and you know what INSERT, UPDATE and WHERE do in SQL.

Create an empty folder with a virtual environment. Every file in this tutorial goes into that one folder, and the code blocks are meant to be pasted in page order. Where a file grows over several steps, the later block starts with a comment that says “continued”: append it to the same file.

mkdir jobqueue-lab
cd jobqueue-lab
python -m venv .venv
.venv\Scripts\activate
python -m pip install pytest==9.1.1

On Linux or macOS, activate the environment with source .venv/bin/activate instead.

Step 0: Check your versions

Two version facts matter. The queue uses RETURNING, which the SQLite documentation says “has been supported by SQLite since version 3.35.0 (2021-03-12).” And SQLite had a rare bug in WAL mode, the journal mode we will use, that is fixed only in newer releases. Save this as step00_versions.py and run it.

"""step00_versions.py: which Python and SQLite are we running, and is the WAL-reset bug fixed in this build?"""
import sqlite3
import sys

v = sqlite3.sqlite_version_info
fixed = v >= (3, 51, 3) or (3, 50, 7) <= v < (3, 51, 0) or (3, 44, 6) <= v < (3, 45, 0)
print("Python", sys.version.split()[0], "| SQLite", sqlite3.sqlite_version)
print("RETURNING needs SQLite 3.35.0 or newer:", v >= (3, 35, 0))
print("WAL-reset fix present (3.51.3 or later, or the 3.50.7 or 3.44.6 backport):", fixed)

python step00_versions.py

Python 3.13.14 | SQLite 3.50.4
RETURNING needs SQLite 3.35.0 or newer: True
WAL-reset fix present (3.51.3 or later, or the 3.50.7 or 3.44.6 backport): False

The last line prints False on my machine, which has SQLite 3.50.4. The SQLite WAL page describes the problem: “The bug only affects databases in WAL mode when there are two or more database connections open on the same file, in separate threads or processes, and when those two connections attempt to write or checkpoint at the same instant.” It adds that the bug “is unlikely to occur in common use” and that it “is fixed in version 3.51.3 (2026-03-13) and later,” with backports in 3.44.6 and 3.50.7. A queue is exactly the pattern the page describes, several processes writing to one file, so run a fixed SQLite build before you rely on this design in production. My tutorial on SQLite WAL mode and “database is locked” errors shows how to check your build in more detail. The labs below use short-lived scratch files, so they are fine as long as the RETURNING line prints True.

Check: the RETURNING line must say True before you go on, because the rest of the code needs it.

Step 1: See why “look first, then claim” fails

The first version of any queue selects the oldest waiting job and then updates it to say “running”. Between those two statements there is a gap, and another worker can walk through it. Two workers look at the same row, both see it waiting, and both decide it is theirs.

We will measure the damage with three ways to claim a job. First, the helpers that every script uses. Save this as labutil.py. db_path gives each demo a fresh database file, FakeClock is a clock you move by hand so that lease timing is exact in later steps, and kill_tree ends a process the hard way, with no chance to clean up (Step 3 uses it).

"""labutil.py: tiny helpers shared by the lab scripts."""
import os
import subprocess
from pathlib import Path

WORK = Path(__file__).resolve().parent / "work"


class FakeClock:
    """A clock you move by hand, so lease and backoff behaviour is deterministic."""

    def __init__(self, start=1000.0):
        self.now = start

    def __call__(self):
        return self.now


def db_path(name):
    """Return work/<name>.db with any old copy (and its -wal and -shm files) removed."""
    WORK.mkdir(exist_ok=True)
    path = WORK / f"{name}.db"
    for suffix in ("", "-wal", "-shm"):
        try:
            os.remove(str(path) + suffix)
        except FileNotFoundError:
            pass
    return path


def kill_tree(proc):
    """A hard kill: no exception handlers and no finally blocks run in the worker."""
    if os.name == "nt":  # taskkill /T also ends any child process the worker started
        subprocess.run(["taskkill", "/PID", str(proc.pid), "/T", "/F"], capture_output=True)
    else:
        proc.kill()
    proc.wait()

Now the race. Save the next two blocks as one file, step01_claim_race.py. The table in this step is deliberately minimal: an id, a state and the name of the worker that claimed the row.

"""step01_claim_race.py: three ways to claim a job, raced by four worker processes."""
import multiprocessing as mp
import sqlite3
import time

from labutil import db_path

N_JOBS = 20000
WORKERS = 4


def connect(path):
    db = sqlite3.connect(path, isolation_level=None, timeout=30.0)
    db.execute("PRAGMA synchronous = NORMAL")
    return db


def claim_naive(db, worker):
    """Look first, then write. Another worker can look at the same row in between."""
    row = db.execute("SELECT id FROM jobs WHERE state = 'queued' ORDER BY id LIMIT 1").fetchone()
    if row is None:
        return None, 0
    db.execute("UPDATE jobs SET state = 'running', claimed_by = ? WHERE id = ?", (worker, row[0]))
    return row[0], 0


def claim_cas(db, worker):
    """Compare and swap: the UPDATE only succeeds if the job is still queued."""
    wasted = 0
    while True:
        row = db.execute("SELECT id FROM jobs WHERE state = 'queued' ORDER BY id LIMIT 1").fetchone()
        if row is None:
            return None, wasted
        cur = db.execute(
            "UPDATE jobs SET state = 'running', claimed_by = ? WHERE id = ? AND state = 'queued'",
            (worker, row[0]))
        if cur.rowcount == 1:
            return row[0], wasted
        wasted += 1


def claim_atomic(db, worker):
    """One statement picks the row and claims it."""
    rows = db.execute(
        "UPDATE jobs SET state = 'running', claimed_by = ? "
        "WHERE id = (SELECT id FROM jobs WHERE state = 'queued' ORDER BY id LIMIT 1) RETURNING id",
        (worker,)).fetchall()
    return (rows[0][0] if rows else None), 0

Each claim function returns the id it claimed (or None when nothing is left) and the number of retries it wasted.

• claim_naive looks first and writes second, with nothing in between. This is the bug.
• claim_cas still looks first, but its UPDATE adds AND state = 'queued' and then checks cur.rowcount. If another worker got there first, no row matches, rowcount is 0, and the worker looks again. This pattern is called compare-and-swap: change a row only if it still looks the way you last saw it. It is correct, but every lost race costs a wasted round trip.
• claim_atomic folds the choice into the update. One statement picks the oldest queued row, marks it, and hands back its id with RETURNING.

The second half of the file starts four fresh Python processes and lines them up on one start time, so that they really overlap.

# step01_claim_race.py (continued)
STRATEGIES = {"naive": claim_naive, "cas": claim_cas, "atomic": claim_atomic}


def worker_main(path, strategy, worker, start_at, results):
    db = connect(path)
    claim = STRATEGIES[strategy]
    mine, wasted = [], 0
    while time.time() < start_at:  # line the processes up so they really overlap
        time.sleep(0.001)
    while True:
        job_id, retries = claim(db, worker)
        wasted += retries
        if job_id is None:
            break
        mine.append(job_id)
    results.put((worker, mine, wasted, time.time()))


def race(strategy):
    path = db_path(f"race_{strategy}")
    db = sqlite3.connect(path, isolation_level=None)
    db.execute("PRAGMA journal_mode = WAL")
    db.execute("CREATE TABLE jobs (id INTEGER PRIMARY KEY, state TEXT NOT NULL DEFAULT 'queued', claimed_by TEXT)")
    db.execute("BEGIN")
    db.executemany("INSERT INTO jobs (state) VALUES ('queued')", [()] * N_JOBS)
    db.execute("COMMIT")
    db.close()
    ctx = mp.get_context("spawn")
    results = ctx.Queue()
    start_at = time.time() + 3.0
    procs = [ctx.Process(target=worker_main, args=(str(path), strategy, f"w{i}", start_at, results))
             for i in range(WORKERS)]
    for p in procs:
        p.start()
    rows = [results.get() for _ in procs]
    for p in procs:
        p.join()
    claimed = [job_id for _, mine, _, _ in rows for job_id in mine]
    return {
        "claims": len(claimed),
        "distinct": len(set(claimed)),
        "extra": len(claimed) - len(set(claimed)),
        "wasted": sum(r[2] for r in rows),
        "seconds": max(r[3] for r in rows) - start_at,
    }


if __name__ == "__main__":
    print(f"{N_JOBS} jobs, {WORKERS} worker processes\n")
    print(f"{'strategy':<9} {'claims':>7} {'distinct':>9} {'duplicate claims':>17} {'wasted retries':>15} {'seconds':>8}")
    for name in STRATEGIES:
        r = race(name)
        print(f"{name:<9} {r['claims']:>7} {r['distinct']:>9} {r['extra']:>17} {r['wasted']:>15} {r['seconds']:>8.2f}")

python step01_claim_race.py

20000 jobs, 4 worker processes

strategy   claims  distinct  duplicate claims  wasted retries  seconds
naive       25331     20000              5331               0     7.52
cas         20000     20000                 0            4654     6.06
atomic      20000     20000                 0               0     5.76

Read the table like this. The naive version made 25,331 claims for 20,000 jobs, so 5,331 of its claims were duplicates. In a real queue that is 5,331 duplicate emails. The compare-and-swap version made exactly 20,000 claims, but it wasted 4,654 retries along the way. The single-statement version made 20,000 claims, no duplicates, no retries, and it finished first (5.76 seconds against 6.06 and 7.52). Your exact numbers will differ from run to run. The pattern should not: only the naive row goes above 20,000 claims.

Why is one statement safe? SQLite lets only one connection write at a time. The WAL page puts it this way: “since there is only one WAL file, there can only be one writer at a time.” The UPDATE and its sub-select are a single statement, so no other writer can slip in between choosing the row and marking it. The measurement agrees: not one duplicate in 20,000 jobs. PostgreSQL needs SKIP LOCKED because many sessions can write at once; SQLite does not.

One warning about RETURNING: the manual says “The rows emitted by the RETURNING clause appear in an arbitrary order.” When we claim several jobs at once later, we sort them ourselves.

Check: the duplicate claims column is above zero for naive and exactly zero for the other two. If naive shows zero on a very fast machine, raise N_JOBS.

Step 2: Design the table and add leases

The table: one row per job

Now the real queue. One table holds every job.

The column doing the most work is visible_at. For a queued job it is the time the job becomes ready: now for a normal job, later for a delayed job or a retry. For a running job it is the moment its lease ends. In both cases the question “may a worker take this job?” has one answer: the state is queued or running, and visible_at has passed. An expired lease therefore needs no special code, no reaper process and no extra column. A job that is leased is simply invisible to the next claim until its deadline.

Save this as jobqueue.py. The two CREATE INDEX statements make claiming fast; Step 8 explains each one with measurements, so for now read them as speed-ups and move on.

"""jobqueue.py: a durable job queue on SQLite with leases, fencing tokens, retries and dead letters."""
import json
import random
import sqlite3
import time
import uuid
from contextlib import contextmanager
from dataclasses import dataclass

TABLE = """
CREATE TABLE IF NOT EXISTS jobs (
    id           INTEGER PRIMARY KEY,
    kind         TEXT    NOT NULL,
    payload      TEXT    NOT NULL,
    state        TEXT    NOT NULL DEFAULT 'queued'
                         CHECK (state IN ('queued', 'running', 'done', 'dead')),
    attempts     INTEGER NOT NULL DEFAULT 0,
    max_attempts INTEGER NOT NULL DEFAULT 5,
    visible_at   REAL    NOT NULL,
    lease_token  TEXT,
    worker       TEXT,
    last_error   TEXT,
    dedupe_key   TEXT UNIQUE,
    created_at   REAL    NOT NULL,
    finished_at  REAL
);
"""

INDEXES = """
CREATE INDEX IF NOT EXISTS jobs_ready ON jobs (visible_at, id)
    WHERE state IN ('queued', 'running');
CREATE INDEX IF NOT EXISTS jobs_exhausted ON jobs (visible_at)
    WHERE state = 'running' AND attempts >= max_attempts;
CREATE INDEX IF NOT EXISTS jobs_dead ON jobs (id) WHERE state = 'dead';
"""

SCHEMA = TABLE + INDEXES


@dataclass(frozen=True)
class Job:
    id: int
    kind: str
    payload: dict
    attempts: int
    max_attempts: int
    lease_token: str
    lease_expires_at: float

The class: connection, transactions and enqueue

Next, the class. Four details in __init__ and _tx deserve a sentence each.

• isolation_level=None stops Python from opening transactions behind your back, so the code decides where each transaction starts and ends.
• timeout=30.0 sets how long a connection waits for the write lock. The Python documentation describes the parameter as “How many seconds the connection should wait before raising an OperationalError when a table is locked” and gives the default as five seconds. A busy queue should wait longer.
• PRAGMA journal_mode = WAL switches on write-ahead logging, in which readers do not block the writer. PRAGMA synchronous = NORMAL trades some durability for speed; Step 9 measures the trade.
• _tx wraps a block in BEGIN IMMEDIATE and COMMIT, rolling back on any exception. The SQLite transaction documentation explains the keyword: “IMMEDIATE causes the database connection to start a new write immediately, without waiting for a write statement.” We want the write lock before our first read, so nothing can slip in between our reads and our writes.

# jobqueue.py (continued)
class JobQueue:
    def __init__(self, path, *, lease_seconds=30.0, max_attempts=5, backoff_base=1.0,
                 backoff_cap=60.0, synchronous="NORMAL", clock=time.time, rng=random.random,
                 schema=SCHEMA):
        self.lease_seconds = lease_seconds
        self.max_attempts = max_attempts
        self.backoff_base = backoff_base
        self.backoff_cap = backoff_cap
        self.clock = clock
        self.rng = rng
        self.db = sqlite3.connect(str(path), isolation_level=None, timeout=30.0)
        self.db.row_factory = sqlite3.Row
        self.db.execute("PRAGMA journal_mode = WAL")
        self.db.execute(f"PRAGMA synchronous = {synchronous}")
        if schema:
            self.db.executescript(schema)

    @contextmanager
    def _tx(self):
        """One write transaction. BEGIN IMMEDIATE takes the write lock first, so nothing
        can slip in between our reads and our writes."""
        self.db.execute("BEGIN IMMEDIATE")
        try:
            yield self.db
        except BaseException:
            self.db.execute("ROLLBACK")
            raise
        self.db.execute("COMMIT")

    def enqueue(self, kind, payload, *, delay=0.0, max_attempts=None, dedupe_key=None):
        """Add a job. Returns its id, or None when dedupe_key already exists."""
        now = self.clock()
        cur = self.db.execute(
            "INSERT OR IGNORE INTO jobs (kind, payload, max_attempts, visible_at, dedupe_key, created_at) "
            "VALUES (?, ?, ?, ?, ?, ?)",
            (kind, json.dumps(payload), max_attempts or self.max_attempts, now + delay, dedupe_key, now))
        return cur.lastrowid if cur.rowcount == 1 else None

    def enqueue_many(self, kind, payloads):
        now = self.clock()
        with self._tx() as db:
            db.executemany(
                "INSERT INTO jobs (kind, payload, max_attempts, visible_at, created_at) VALUES (?, ?, ?, ?, ?)",
                [(kind, json.dumps(p), self.max_attempts, now, now) for p in payloads])
        return len(payloads)

enqueue uses INSERT OR IGNORE together with the unique dedupe_key: a second enqueue with the same key changes nothing and returns None. enqueue_many loads many jobs in one transaction, which we use for the benchmarks.

Claim and complete

Now the heart of the queue, claim, and its other half, complete.

# jobqueue.py (continued)
    def claim(self, worker, *, limit=1):
        """Lease up to `limit` jobs. A job is claimable when its visible_at has passed:
        for a queued job that is its run time, for a running job it is the end of its lease."""
        now = self.clock()
        token = uuid.uuid4().hex
        with self._tx() as db:
            # A lease that expired on the last allowed attempt means the worker died every
            # time. Retire the job instead of running it again.
            db.execute(
                "UPDATE jobs SET state = 'dead', lease_token = NULL, finished_at = ?, "
                "last_error = COALESCE(last_error, 'lease expired on the last attempt') "
                "WHERE state = 'running' AND attempts >= max_attempts AND visible_at <= ?",
                (now, now))
            rows = db.execute(
                """
                UPDATE jobs
                   SET state = 'running', attempts = attempts + 1,
                       visible_at = ?, lease_token = ?, worker = ?
                 WHERE id IN (SELECT id FROM jobs
                               WHERE state IN ('queued', 'running') AND visible_at <= ?
                               ORDER BY visible_at, id LIMIT ?)
                RETURNING id, kind, payload, attempts, max_attempts, lease_token, visible_at
                """,
                (now + self.lease_seconds, token, worker, now, limit)).fetchall()
        jobs = [Job(id=r["id"], kind=r["kind"], payload=json.loads(r["payload"]),
                    attempts=r["attempts"], max_attempts=r["max_attempts"],
                    lease_token=r["lease_token"], lease_expires_at=r["visible_at"]) for r in rows]
        return sorted(jobs, key=lambda j: j.id)

    def complete(self, job, effect=None):
        """Mark the job done, but only for the worker that holds the current lease.
        `effect(db)` runs in the same transaction, so it commits with the completion or not at all."""
        with self._tx() as db:
            cur = db.execute(
                "UPDATE jobs SET state = 'done', finished_at = ?, lease_token = NULL "
                "WHERE id = ? AND lease_token = ? AND state = 'running'",
                (self.clock(), job.id, job.lease_token))
            if cur.rowcount != 1:
                return False
            if effect is not None:
                effect(db)
        return True

claim runs two statements inside one write transaction. The first one retires jobs that ran out of attempts; Step 7 explains it, so skip it for now. The second is the atomic claim from Step 1, made lease-aware:

• The sub-select picks the oldest claimable jobs: state queued or running, and visible_at already in the past. A running job qualifies only once its lease has ended.
• The update marks them running, adds one to attempts, pushes visible_at to now plus the lease length, and stores a fresh random lease_token and the worker’s name.
• RETURNING hands the updated rows back, and the code sorts them by id because the order is arbitrary.

complete only succeeds if the WHERE clause still matches: same job, same lease token, still running. Step 4 shows what that check prevents. Ignore the effect argument until Step 5.

Watch a lease expire

To see the lease in action without waiting for real time, this script uses the fake clock. Save it as step02_leases.py.

"""step02_leases.py: walk one job through claim, lease expiry and re-claim with a fake clock."""
from jobqueue import JobQueue
from labutil import FakeClock, db_path


def show(q, label):
    r = q.db.execute("SELECT state, attempts, visible_at, lease_token, worker FROM jobs").fetchone()
    token = "set" if r["lease_token"] else "none"
    print(f"{label:<38} state={r['state']:<8} attempts={r['attempts']} "
          f"visible_at={r['visible_at']:.0f} token={token:<4} worker={r['worker']}")


clock = FakeClock(1000.0)
q = JobQueue(db_path("leases"), lease_seconds=30, clock=clock)
q.enqueue("email", {"to": "[email protected]"})
show(q, "t=1000 enqueued")

(a,) = q.claim("worker-a")
show(q, "t=1000 worker-a claims")

clock.now = 1010
print("t=1010 worker-b claims:", q.claim("worker-b"))

clock.now = 1031
(b,) = q.claim("worker-b")
show(q, "t=1031 lease ran out, worker-b claims")
print("same token for both claims:", a.lease_token == b.lease_token)

print("worker-a complete():", q.complete(a))
print("worker-b complete():", q.complete(b))
show(q, "t=1031 worker-b completed")

python step02_leases.py

t=1000 enqueued                        state=queued   attempts=0 visible_at=1000 token=none worker=None
t=1000 worker-a claims                 state=running  attempts=1 visible_at=1030 token=set  worker=worker-a
t=1010 worker-b claims: []
t=1031 lease ran out, worker-b claims  state=running  attempts=2 visible_at=1061 token=set  worker=worker-b
same token for both claims: False
worker-a complete(): False
worker-b complete(): True
t=1031 worker-b completed              state=done     attempts=2 visible_at=1061 token=none worker=worker-b

Follow the job from line to line. After the enqueue it is queued and ready at time 1000. When worker-a claims it, the state becomes running, attempts is 1 and visible_at moves to 1030, the end of the 30-second lease. At time 1010 worker-b asks for work and gets an empty list: the job is leased, so it is invisible. At time 1031 the lease has run out, and worker-b claims the same job. attempts is now 2, the lease runs to 1061, and the two claims have different tokens. Worker-a’s late complete() returns False; worker-b’s returns True. Nobody had to clean up after worker-a: the claim query noticed the expired lease by itself.

Check: the third line shows an empty list and the last line shows state=done. If the empty list is missing, your visible_at <= ? comparison or your lease length is wrong.

Step 3: Kill a worker and watch the lease hand its job over

Now with real processes. A worker is just a loop, and it needs jobs to run. Save the first part of the handlers as handlers.py. report pretends to build a report: on its first attempt it hangs (so we can kill the process in the middle of it), and later attempts finish quickly. poison kills its own process with os._exit(1), which stops Python instantly: no except block runs and no finally block runs.

"""handlers.py: the jobs the demos run. A handler may return a function; the worker runs it inside the transaction that completes the job."""
import os
import time


def report(job, q):
    # Attempt 1 hangs, so the demo can kill this process mid-job. Later attempts finish quickly.
    if job.payload.get("stall_first_attempt") and job.attempts == 1:
        time.sleep(60)
    time.sleep(0.2)


def poison(job, q):
    os._exit(1)  # a hard crash: no except block runs and nothing calls fail()


HANDLERS = {"report": report, "poison": poison}

Then the worker loop, worker.py.

"""worker.py: the loop a worker process runs. The crash, poison-job and end-to-end demos start it."""
import sys
import time

from handlers import HANDLERS
from jobqueue import JobQueue


def run(path, worker_id, lease_seconds, idle_exit=2.0):
    q = JobQueue(path, lease_seconds=lease_seconds)
    idle_since = None
    while True:
        jobs = q.claim(worker_id)
        if not jobs:  # nothing ready: wait a moment, and leave after idle_exit seconds
            idle_since = idle_since or time.monotonic()
            if time.monotonic() - idle_since >= idle_exit:
                return
            time.sleep(0.05)
            continue
        idle_since = None
        job = jobs[0]
        try:
            effect = HANDLERS[job.kind](job, q)
        except Exception as exc:
            q.fail(job, f"{type(exc).__name__}: {exc}")
        else:
            q.complete(job, effect=effect)


if __name__ == "__main__":
    idle = float(sys.argv[4]) if len(sys.argv) > 4 else 2.0
    run(sys.argv[1], sys.argv[2], float(sys.argv[3]), idle)

If nothing is ready, the worker sleeps 50 milliseconds and asks again; after idle_exit seconds of silence it leaves, so the demos end by themselves. It looks up the handler for the job’s kind and runs it. A handler may return a function, which the worker hands to complete; Step 5 uses that. If the handler raises, the worker reports the failure with fail. Notice what is missing: nothing releases the job when a worker dies. A crash simply stops the loop in the middle of a job, and the lease does the cleanup.

The demo enqueues one job, starts worker w1, waits until w1 has claimed it, kills w1 hard, and starts w2. Save it as step03_crash_recovery.py.

"""step03_crash_recovery.py: kill a worker mid-job and watch the lease hand the job to another worker."""
import subprocess
import sys
import time
from pathlib import Path

from jobqueue import JobQueue
from labutil import db_path, kill_tree

HERE = Path(__file__).resolve().parent
LEASE = 2.0


def start_worker(path, name):
    return subprocess.Popen([sys.executable, str(HERE / "worker.py"), str(path), name, str(LEASE)],
                            stdout=subprocess.DEVNULL)


def watch(q, t0, seen, until):
    """Print every change of (state, attempts, worker) until `until` says stop."""
    while True:
        r = q.db.execute("SELECT state, attempts, worker FROM jobs WHERE id = 1").fetchone()
        snap = (r["state"], r["attempts"], r["worker"])
        if snap != seen[0]:
            seen[0] = snap
            print(f"t={time.monotonic() - t0:4.1f}s  state={snap[0]:<8} attempts={snap[1]} worker={snap[2]}")
        if until(snap):
            return
        time.sleep(0.02)


path = db_path("crash")
q = JobQueue(path, lease_seconds=LEASE)
q.enqueue("report", {"stall_first_attempt": True})
t0, seen = time.monotonic(), [None]
print(f"lease is {LEASE:.0f} seconds; attempt 1 will hang inside the handler")

w1 = start_worker(path, "w1")
watch(q, t0, seen, lambda s: s[0] == "running")
print(f"t={time.monotonic() - t0:4.1f}s  killing w1 while it is inside the handler")
kill_tree(w1)

w2 = start_worker(path, "w2")
watch(q, t0, seen, lambda s: s[0] == "done")
w2.wait()

python step03_crash_recovery.py

lease is 2 seconds; attempt 1 will hang inside the handler
t= 0.0s  state=queued   attempts=0 worker=None
t= 0.1s  state=running  attempts=1 worker=w1
t= 0.1s  killing w1 while it is inside the handler
t= 2.1s  state=running  attempts=2 worker=w2
t= 2.3s  state=done     attempts=2 worker=w2

The timeline tells the whole story. w1 claims the job at the start (attempts=1) and is killed while it is inside the handler. For about two seconds nothing happens, because the job is still leased to a dead process. That gap is the lease length: the price of crash recovery is that a job waits up to one lease before someone else may take it. Then w2 claims it (attempts=2) and finishes it quickly. Your times will differ by a few tenths of a second; the two-second gap should not.

This is at-least-once delivery in action. The first attempt never reported back, and nobody can know whether it did half of the work, all of it, or none. The queue chose to run the job again. Every handler you write for it must survive that.

Check: you see exactly one running line with attempts=1 and one with attempts=2, then done. If the script hangs after the kill line, the lease never expired: check that claim compares visible_at with the current time and that the lease length reaches the worker. On Windows, kill_tree uses taskkill /T, which also ends any process the worker started.

Step 4: Stop the slow worker from undoing the new worker’s work

A dead worker is the easy case. The hard case is a worker that is merely slow: it stalls for longer than its lease, another worker takes over and finishes the job, and then the slow worker wakes up and reports its own result. What should the queue do with that late report?

The bug: a late report from a stale worker

Here is what many tutorials do: they finish or fail a job by its id alone. The script below simulates exactly that with a fake clock. Worker-a stalls past its 30-second lease. Worker-b takes over at time 31 and succeeds. At time 40 worker-a finally gets a timeout error from the mail server and reports a failure the naive way. Save this as step04_zombie_workers.py.

"""step04_zombie_workers.py: a worker whose lease expired comes back late, first without the fence and then with it."""
from jobqueue import JobQueue
from labutil import FakeClock, db_path


def make(name):
    clock = FakeClock(0.0)
    q = JobQueue(db_path(name), lease_seconds=30, clock=clock)
    q.enqueue("email", {"to": "[email protected]"})
    return q, clock


def unfenced_fail(q, job, error):
    """What many tutorials write: fail a job by id, without asking who holds the lease."""
    q.db.execute("UPDATE jobs SET state = 'queued', visible_at = ?, last_error = ? WHERE id = ?",
                 (q.clock(), error, job.id))


def row(q):
    r = q.db.execute("SELECT state, attempts FROM jobs").fetchone()
    return f"state={r['state']} attempts={r['attempts']}"


print("--- without the fence")
q, clock = make("zombie_unfenced")
runs = []
(a,) = q.claim("worker-a")
runs.append(f"worker-a (attempt {a.attempts})")
clock.now = 31  # worker-a is still busy, but its 30 second lease ran out
(b,) = q.claim("worker-b")
runs.append(f"worker-b (attempt {b.attempts})")
print("t=31  worker-b completes:", q.complete(b), "->", row(q))
clock.now = 40  # worker-a finally gets a timeout error from the mail server
unfenced_fail(q, a, "TimeoutError")
print("t=40  worker-a fails late  ->", row(q))
(c,) = q.claim("worker-c")
runs.append(f"worker-c (attempt {c.attempts})")
print("t=40  worker-c claims      ->", row(q))
print(f"the handler ran {len(runs)} times: " + ", ".join(runs))

print()

python step04_zombie_workers.py

--- without the fence
t=31  worker-b completes: True -> state=done attempts=2
t=40  worker-a fails late  -> state=queued attempts=2
t=40  worker-c claims      -> state=running attempts=3
the handler ran 3 times: worker-a (attempt 1), worker-b (attempt 2), worker-c (attempt 3)

The job had succeeded, and worker-a’s late failure report put it back in the queue. A third worker claimed it, so the handler ran three times for one job. In a real system that is three emails for one sign-up, from a queue that said “done” in the middle.

The fix: check the lease token on every report

The fix is to make every report prove that it comes from the current claim. That is the whole purpose of the lease token. The WHERE clause of complete already checks it, and fail and heartbeat must check it too: WHERE id = ? AND lease_token = ? AND state = 'running'. If the statement changes zero rows, the report came from a stale claim, and the queue ignores it. Martin Kleppmann describes the same idea for distributed locks in How to do distributed locking: “you need to include a fencing token with every write request to the storage service. In this context, a fencing token is simply a number that increases (e.g. incremented by the lock service) every time a client acquires the lock.” Amazon’s queue does it with receipt handles: “If you receive a message more than once, each time you receive it, you get a different receipt handle. You must provide the most recently received receipt handle when you request to delete the message (otherwise, the message might not be deleted).”

One difference is worth knowing. Kleppmann’s tokens increase, because a storage service that did not issue them must be able to tell which holder is newer. In our queue the same table issues the token and checks it, so the only question is “is this still the current claim?”, and a random value answers that.

Here is fail, the other fenced method. Append it to the JobQueue class in jobqueue.py. Its CASE expressions pick the next state; Step 7 explains them.

# jobqueue.py (continued)
    def fail(self, job, error):
        """Record a failure. Returns 'queued' (retry scheduled), 'dead' (out of attempts)
        or None (this worker no longer holds the job)."""
        now = self.clock()
        ceiling = min(self.backoff_cap, self.backoff_base * 2 ** (job.attempts - 1))
        retry_at = now + self.rng() * ceiling
        rows = self.db.execute(
            """
            UPDATE jobs
               SET state       = CASE WHEN attempts >= max_attempts THEN 'dead' ELSE 'queued' END,
                   visible_at  = CASE WHEN attempts >= max_attempts THEN visible_at ELSE ? END,
                   finished_at = CASE WHEN attempts >= max_attempts THEN ? END,
                   lease_token = NULL,
                   last_error  = ?
             WHERE id = ? AND lease_token = ? AND state = 'running'
            RETURNING state
            """,
            (retry_at, now, error[:500], job.id, job.lease_token)).fetchall()
        return rows[0]["state"] if rows else None

Now the second half of the demo, which runs the same timeline with the fenced methods, plus one more case. Append it to step04_zombie_workers.py.

# step04_zombie_workers.py (continued)
print("--- with the fence")
q, clock = make("zombie_fenced")
(a,) = q.claim("worker-a")
clock.now = 31
(b,) = q.claim("worker-b")
print("t=31  worker-b completes:", q.complete(b), "->", row(q))
clock.now = 40
print("t=40  worker-a q.fail() returned:", q.fail(a, "TimeoutError"), "->", row(q))
print("t=40  worker-c claims:", q.claim("worker-c"))

print()
print("--- a slow worker that nobody replaced")
q, clock = make("zombie_unreplaced")
(a,) = q.claim("worker-a")
clock.now = 45  # the lease ended at t=30, but no other worker has claimed the job
print("t=45  worker-a complete():", q.complete(a), "->", row(q))

--- with the fence
t=31  worker-b completes: True -> state=done attempts=2
t=40  worker-a q.fail() returned: None -> state=done attempts=2
t=40  worker-c claims: []

--- a slow worker that nobody replaced
t=45  worker-a complete(): True -> state=done attempts=1

With the fence, worker-a’s late fail() returns None and the job stays done. No third worker gets it. The last case shows the other side: when a slow worker’s lease has expired but nobody has claimed the job, the token still matches, complete() returns True, and the job finishes. That is the right behavior. The work is done, nobody replaced the worker, and throwing the result away would only waste it. The fence rejects a worker only once somebody else has taken over.

Check: the unfenced run shows handler ran 3 times and the fenced run shows q.fail() returned: None with the job still done. If the fenced run also shows three runs, the lease_token condition is missing from one of the UPDATE statements.

Step 5: Make the side effect part of the same transaction

The fence protects the queue’s own state. It does not stop a stale worker from doing the job’s work: worker-a still sent its email before it found out that it had lost the lease. For work that happens inside the same SQLite file, we can close that gap completely.

Take a job that adds 10 credits to a user’s account. The script below runs the stale-worker timeline twice. In the first run each worker applies the credits and then calls complete(), as two separate commits. In the second run the credits are applied by a function that complete() runs inside its own transaction, after the lease check has passed. Save it as step05_effect_in_transaction.py.

"""step05_effect_in_transaction.py: put the job's database effect and its completion in one transaction."""
from jobqueue import JobQueue
from labutil import FakeClock, db_path


def setup(name):
    clock = FakeClock(0.0)
    q = JobQueue(db_path(name), lease_seconds=30, clock=clock)
    q.db.execute("CREATE TABLE accounts (user TEXT PRIMARY KEY, credits INTEGER NOT NULL)")
    q.enqueue("grant", {"user": "ada", "credits": 10})
    return q, clock


def add_credits(job):
    def effect(db):
        db.execute(
            "INSERT INTO accounts (user, credits) VALUES (?, ?) "
            "ON CONFLICT (user) DO UPDATE SET credits = credits + excluded.credits",
            (job.payload["user"], job.payload["credits"]))
    return effect


def credits(q):
    return q.db.execute("SELECT credits FROM accounts WHERE user = 'ada'").fetchone()[0]


print("--- effect first, then complete (two separate commits)")
q, clock = setup("effect_separate")
(a,) = q.claim("worker-a")
clock.now = 31  # worker-a stalls past its lease
(b,) = q.claim("worker-b")
add_credits(b)(q.db)
print("worker-b complete():", q.complete(b))
add_credits(a)(q.db)  # worker-a wakes up and does its work anyway
print("worker-a complete():", q.complete(a))
print("ada has", credits(q), "credits")

print()
print("--- effect passed to complete() (one transaction)")
q, clock = setup("effect_together")
(a,) = q.claim("worker-a")
clock.now = 31
(b,) = q.claim("worker-b")
print("worker-b complete():", q.complete(b, effect=add_credits(b)))
print("worker-a complete():", q.complete(a, effect=add_credits(a)))
print("ada has", credits(q), "credits")

python step05_effect_in_transaction.py

--- effect first, then complete (two separate commits)
worker-b complete(): True
worker-a complete(): False
ada has 20 credits

--- effect passed to complete() (one transaction)
worker-b complete(): True
worker-a complete(): False
ada has 10 credits

In the first run both workers added credits, so Ada has 20 even though only one completion was accepted. In the second run worker-a’s complete() returned False before the effect ran, so Ada has 10. Look back at complete in Step 2 to see how: it opens BEGIN IMMEDIATE, runs the fenced UPDATE, returns early if no row matched, and otherwise runs effect(db) before it commits. The effect and the completion become visible together or not at all. If the effect raises an exception, the whole transaction rolls back and the job stays running, so the worker can report the error with fail.

Know the limits of this trick:

• It works only for effects stored in the same SQLite file. An email, a payment API call or an HTTP request cannot join the transaction. For those, send the job’s id as an idempotency key and use a provider that honors it. My webhook tutorial shows an atomic claim-and-apply pattern for deduplicating replayed events, and the transactional outbox tutorial covers the related problem of writing to two systems at once.
• The handler’s computation can still run twice. Only its database effect is protected. If the work is expensive, avoid the duplicate run with the heartbeat in the next step.

Check: the first run ends with ada has 20 credits and the second with ada has 10 credits.

Step 6: Keep a long job alive with a heartbeat

Lease length is a trade-off. A short lease recovers from crashes quickly but expires under honest slow work. A long lease is patient but leaves a crashed worker’s job stuck for a long time. A heartbeat removes the dilemma: a worker that is alive keeps pushing its deadline forward, so the lease can stay short. Amazon’s queue offers the same tool: “Use the ChangeMessageVisibility action to programmatically extend or shorten the timeout as needed.”

First, the heartbeat method. Append it to the JobQueue class. It carries the same fence as complete and fail, so a worker that has lost its lease cannot extend somebody else’s.

# jobqueue.py (continued)
    def heartbeat(self, job):
        """Push the lease out again. False means this worker no longer holds the job."""
        cur = self.db.execute(
            "UPDATE jobs SET visible_at = ? WHERE id = ? AND lease_token = ? AND state = 'running'",
            (self.clock() + self.lease_seconds, job.id, job.lease_token))
        return cur.rowcount == 1

Then a helper that calls it from a background thread while the handler works. Save it as keepalive.py. The thread opens its own connection, because a SQLite connection belongs to the thread that created it. If a heartbeat finds that the lease is gone, it sets lost, and the handler can check that flag between steps of its work and stop.

"""keepalive.py: extend a job's lease from a background thread while a handler works."""
import threading

from jobqueue import JobQueue


class KeepAlive:
    """Use as `with KeepAlive(path, job, lease_seconds=30) as hb:` around the slow part of a handler.
    The thread opens its own connection, because a sqlite3 connection belongs to one thread."""

    def __init__(self, path, job, *, lease_seconds, interval=None):
        self.path = path
        self.job = job
        self.lease_seconds = lease_seconds
        self.interval = interval or lease_seconds / 3
        self.lost = threading.Event()  # set when a heartbeat finds the lease gone
        self.beats = 0
        self._stop = threading.Event()
        self._thread = threading.Thread(target=self._run, daemon=True)

    def _run(self):
        q = JobQueue(self.path, lease_seconds=self.lease_seconds)
        while not self._stop.wait(self.interval):
            if q.heartbeat(self.job):
                self.beats += 1
            else:
                self.lost.set()
                break
        q.db.close()

    def __enter__(self):
        self._thread.start()
        return self

    def __exit__(self, *exc):
        self._stop.set()
        self._thread.join()

The demo runs a three-second job against a one-second lease, three ways. Save it as step06_heartbeat.py.

"""step06_heartbeat.py: a job that takes longer than its lease, with and without a heartbeat."""
import threading
import time

from jobqueue import JobQueue
from keepalive import KeepAlive
from labutil import db_path

LEASE = 1.0
WORK = 3.0


def scenario(name, heartbeat):
    path = db_path(name)
    JobQueue(path, lease_seconds=LEASE).enqueue("report", {})
    runs, results, lock = [], {}, threading.Lock()

    def worker(worker_id, start_delay):
        time.sleep(start_delay)
        q = JobQueue(path, lease_seconds=LEASE)  # one connection per thread
        deadline = time.monotonic() + 5.5
        while time.monotonic() < deadline:
            jobs = q.claim(worker_id)
            if not jobs:
                time.sleep(0.05)
                continue
            job = jobs[0]
            with lock:
                runs.append(f"{worker_id} attempt {job.attempts}")
            if heartbeat:
                with KeepAlive(path, job, lease_seconds=LEASE) as hb:
                    time.sleep(WORK)  # the "work"
                results[worker_id] = f"complete={q.complete(job)} heartbeats={hb.beats}"
            else:
                time.sleep(WORK)
                results[worker_id] = f"complete={q.complete(job)}"
            return

    threads = [threading.Thread(target=worker, args=("w1", 0.0)), threading.Thread(target=worker, args=("w2", 0.3))]
    for t in threads:
        t.start()
    for t in threads:
        t.join()
    print(f"handler ran {len(runs)} time(s): {', '.join(runs)}")
    for worker_id in sorted(results):
        print(f"  {worker_id}: {results[worker_id]}")


def lost_lease():
    path = db_path("hb_lost")
    q = JobQueue(path, lease_seconds=LEASE)
    q.enqueue("report", {})
    (job,) = q.claim("w1")
    with KeepAlive(path, job, lease_seconds=LEASE, interval=0.2) as hb:
        t0 = time.monotonic()
        time.sleep(0.5)
        # To keep the demo short we expire the lease by hand, then let w2 take the job.
        other = JobQueue(path, lease_seconds=LEASE)
        other.db.execute("UPDATE jobs SET visible_at = 0")
        (taken,) = other.claim("w2")
        while not hb.lost.is_set() and time.monotonic() - t0 < 3:
            time.sleep(0.05)  # w1 would check hb.lost between steps of its work
    print(f"w2 holds the job (attempt {taken.attempts}); w1 saw lost={hb.lost.is_set()} "
          f"after {time.monotonic() - t0:.1f}s and stops working")


print(f"lease {LEASE:.0f}s, work {WORK:.0f}s")
print("--- no heartbeat")
scenario("hb_none", heartbeat=False)
print("--- with KeepAlive")
scenario("hb_keepalive", heartbeat=True)
print("--- the lease is lost while working")
lost_lease()

python step06_heartbeat.py

lease 1s, work 3s
--- no heartbeat
handler ran 2 time(s): w1 attempt 1, w2 attempt 2
  w1: complete=False
  w2: complete=True
--- with KeepAlive
handler ran 1 time(s): w1 attempt 1
  w1: complete=True heartbeats=8
--- the lease is lost while working
w2 holds the job (attempt 2); w1 saw lost=True after 0.7s and stops working

Without a heartbeat, the lease ran out after one second and w2 took the job, so the handler ran twice. The fence did its work: w1’s late complete() returned False and only w2’s counted, but the job’s computation was done twice. With KeepAlive, the lease was renewed about every third of a second (the thread sent 8 heartbeats during the three seconds), w2 never saw the job, and the handler ran once. In the last case a second worker took the job while w1 was still working, and w1’s heartbeat noticed within a few tenths of a second that it had lost the lease. To keep the demo short, the code expires the lease by hand; in real life the cause would be a long stall.

Check: the middle scenario says handler ran 1 time(s) and reports a heartbeat count, and the last line shows lost=True. Heartbeats vary a little from run to run, so any count of about eight is fine.

Step 7: Retries, backoff, and dead letters

Jobs fail. A mail server times out and a retry succeeds; a payload is missing a field and no retry ever will. The queue needs a retry schedule and a place for jobs that will never succeed. Look back at fail. When a handler raises, the worker calls q.fail(job, error), and one fenced UPDATE decides what happens:

• If attempts is still below max_attempts, the state goes back to queued and visible_at moves into the future: this is the retry.
• If the attempts are used up, the state becomes dead. The job stays in the table with its last_error: this is the dead letter. Amazon’s queue works the same way: “The maxReceiveCount is the number of times a consumer can receive a message from a source queue before it is moved to a dead-letter queue.”

Backoff with jitter

The retry delay is random. After attempt n, the ceiling is min(backoff_cap, backoff_base * 2 ** (n - 1)) seconds, and the actual delay is a random number between zero and that ceiling. This is called exponential backoff with full jitter: the ceiling doubles with every attempt, and the randomness stops a crowd of failed jobs from retrying in the same instant. My tutorial on retrying failed API calls explains why jitter matters.

The script below runs two jobs through the queue with a fake clock and a seeded random generator, so the output repeats exactly. flaky fails twice and then works. always_fails never works. Save it as step07_retries.py.

"""step07_retries.py: retries with full-jitter backoff, then the dead-letter state."""
import random

from jobqueue import JobQueue
from labutil import FakeClock, db_path


def handle(job):
    if job.kind == "flaky" and job.attempts < 3:
        raise TimeoutError("upstream timed out")
    if job.kind == "always_fails":
        raise ValueError("payload has no recipient")


clock = FakeClock(0.0)
q = JobQueue(db_path("retries"), lease_seconds=30, max_attempts=4, backoff_base=2.0,
             backoff_cap=60.0, clock=clock, rng=random.Random(7).random)
q.enqueue("flaky", {})
q.enqueue("always_fails", {})

while True:
    nxt = q.db.execute("SELECT MIN(visible_at) FROM jobs WHERE state IN ('queued', 'running')").fetchone()[0]
    if nxt is None:
        break
    clock.now = max(clock.now, nxt)
    for job in q.claim("sim", limit=10):
        label = f"t={clock.now:6.1f}  job {job.id} {job.kind:<12} attempt {job.attempts}/{job.max_attempts}"
        try:
            handle(job)
        except Exception as exc:
            state = q.fail(job, f"{type(exc).__name__}: {exc}")
            if state == "queued":
                due = q.db.execute("SELECT visible_at FROM jobs WHERE id = ?", (job.id,)).fetchone()[0]
                print(f"{label}  failed, retry in {due - clock.now:4.1f}s")
            else:
                print(f"{label}  failed, state={state}")
        else:
            print(f"{label}  complete={q.complete(job)}")

print()
for r in q.db.execute("SELECT id, kind, state, attempts, last_error FROM jobs ORDER BY id"):
    print(f"job {r['id']} {r['kind']:<12} state={r['state']:<5} attempts={r['attempts']} last_error={r['last_error']}")

python step07_retries.py

t=   0.0  job 1 flaky        attempt 1/4  failed, retry in  0.6s
t=   0.0  job 2 always_fails attempt 1/4  failed, retry in  0.3s
t=   0.3  job 2 always_fails attempt 2/4  failed, retry in  2.6s
t=   0.6  job 1 flaky        attempt 2/4  failed, retry in  0.3s
t=   0.9  job 1 flaky        attempt 3/4  complete=True
t=   2.9  job 2 always_fails attempt 3/4  failed, retry in  4.3s
t=   7.2  job 2 always_fails attempt 4/4  failed, state=dead

job 1 flaky        state=done  attempts=3 last_error=TimeoutError: upstream timed out
job 2 always_fails state=dead  attempts=4 last_error=ValueError: payload has no recipient

The flaky job failed on attempts 1 and 2, waited 0.6 and then 0.3 seconds, and succeeded on attempt 3. The hopeless job waited 0.3, 2.6 and 4.3 seconds; each wait is somewhere between zero and the ceiling for that attempt (2, 4 and 8 seconds with these settings). After its fourth failure it became dead, with the error message kept for a human.

The poison job: a job that kills its worker

There is a nastier failure. A job that crashes the whole process, for example by running out of memory, never reaches fail. The only trace is a lease that expires. That is why the code counts an attempt when a job is claimed, not when it fails. The first statement in claim finishes the job off: any running job whose lease has ended and whose attempts are used up becomes dead. The script starts a fresh worker for each round, lets the poison job kill it, and waits for the lease to expire. Save it as step07b_poison.py.

"""step07b_poison.py: a job that kills its worker is still counted, because attempts are counted at claim time."""
import subprocess
import sys
import time
from pathlib import Path

from jobqueue import JobQueue
from labutil import FakeClock, db_path

HERE = Path(__file__).resolve().parent

path = db_path("poison")
q = JobQueue(path, lease_seconds=1.0, max_attempts=3)
q.enqueue("poison", {})
for rnd in range(1, 5):
    proc = subprocess.run([sys.executable, str(HERE / "worker.py"), str(path), f"w{rnd}", "1.0", "0.5"], timeout=60)
    r = q.db.execute("SELECT state, attempts, last_error FROM jobs").fetchone()
    print(f"round {rnd}: worker exit code {proc.returncode}, job state={r['state']} "
          f"attempts={r['attempts']} last_error={r['last_error']}")
    time.sleep(1.2)  # let the 1 second lease run out before the next worker starts

print()
print("--- the same crash loop if attempts were only counted when fail() runs")
clock = FakeClock(0.0)
q2 = JobQueue(db_path("poison_uncounted"), lease_seconds=1.0, max_attempts=3, clock=clock)
q2.enqueue("poison", {})
handed_out = 0
for rnd in range(1, 7):
    cur = q2.db.execute(
        "UPDATE jobs SET state = 'running', visible_at = ?, lease_token = 'x' "
        "WHERE state IN ('queued', 'running') AND visible_at <= ?", (clock.now + 1.0, clock.now))
    handed_out += cur.rowcount
    clock.now += 1.5  # the worker crashed, so nothing called fail(); its lease runs out
r = q2.db.execute("SELECT state, attempts FROM jobs").fetchone()
print(f"handed out {handed_out} times in 6 rounds, state={r['state']} attempts={r['attempts']}: it would never stop")

python step07b_poison.py

round 1: worker exit code 1, job state=running attempts=1 last_error=None
round 2: worker exit code 1, job state=running attempts=2 last_error=None
round 3: worker exit code 1, job state=running attempts=3 last_error=None
round 4: worker exit code 0, job state=dead attempts=3 last_error=lease expired on the last attempt

--- the same crash loop if attempts were only counted when fail() runs
handed out 6 times in 6 rounds, state=running attempts=0: it would never stop

Each of the first three workers claimed the job (attempts 1, 2 and 3) and died with exit code 1. The fourth worker’s claim found a lease that had ended on the last allowed attempt and moved the job to dead with the reason lease expired on the last attempt. The worker found nothing to run and left with exit code 0. The second half of the output shows the alternative design, counting attempts only inside fail(): the job was handed out six times in six rounds, attempts stayed at 0, and it would never stop. Amazon’s redrive policy counts receives in the same way.

Check: the poison job ends with state=dead attempts=3, and the uncounted variant shows attempts=0 after six hand-outs.

Step 8: Keep the claim fast with partial indexes

A queue table grows: every finished job stays in it as a done row unless you delete it. The claim has to stay fast anyway. This script builds a table with one million finished jobs and 1,000 queued ones, and times the claim under three index setups. The finished jobs have older visible_at values than the queued ones, which is what a long-running queue looks like. It also prints the plan SQLite picks for the claim and for the retire statement, using EXPLAIN QUERY PLAN; my tutorial on reading SQLite query plans explains the notation. Save it as step08_claim_index.py.

"""step08_claim_index.py: what a claim costs when a million finished jobs sit in the table."""
import sqlite3
import statistics
import time

from jobqueue import INDEXES, TABLE, JobQueue
from labutil import FakeClock, db_path

N_DONE = 1_000_000
N_QUEUED = 1_000
BASE = 1_000_000.0
NOW = BASE + N_DONE * 0.5 + N_QUEUED + 10  # later than every visible_at in the table

CLAIM_PLAN = ("EXPLAIN QUERY PLAN SELECT id FROM jobs WHERE state IN ('queued', 'running') "
              "AND visible_at <= ? ORDER BY visible_at, id LIMIT 1")
RETIRE_PLAN = ("EXPLAIN QUERY PLAN SELECT id FROM jobs "
               "WHERE state = 'running' AND attempts >= max_attempts AND visible_at <= ?")

VARIANTS = [  # (label, SQL that creates the indexes, claims to time)
    ("no index", None, 10),
    ("plain index on (visible_at, id)", "CREATE INDEX jobs_plain ON jobs (visible_at, id)", 10),
    ("the partial indexes in jobqueue.py", INDEXES, 200),
]


def seed(path):
    db = sqlite3.connect(path, isolation_level=None)
    db.execute("PRAGMA journal_mode = WAL")
    db.execute("PRAGMA synchronous = OFF")  # bulk load only
    db.executescript(TABLE)
    db.execute("BEGIN")
    db.executemany(
        "INSERT INTO jobs (kind, payload, state, attempts, visible_at, created_at, finished_at) "
        "VALUES ('email', '{}', 'done', 1, ?, ?, ?)",
        ((BASE + i * 0.5,) * 3 for i in range(N_DONE)))
    db.executemany(
        "INSERT INTO jobs (kind, payload, visible_at, created_at) VALUES ('email', '{}', ?, ?)",
        ((BASE + N_DONE * 0.5 + i,) * 2 for i in range(N_QUEUED)))
    db.execute("COMMIT")
    db.close()


path = db_path("claim_index")
t0 = time.perf_counter()
seed(path)
print(f"seeded {N_DONE:,} done and {N_QUEUED:,} queued jobs in {time.perf_counter() - t0:.1f}s")

db = sqlite3.connect(path, isolation_level=None)
for label, create, runs in VARIANTS:
    if create:
        db.executescript(create)
    claim_plan = [row[3] for row in db.execute(CLAIM_PLAN, (NOW,))]
    retire_plan = [row[3] for row in db.execute(RETIRE_PLAN, (NOW,))]
    q = JobQueue(path, schema=None, clock=FakeClock(NOW), lease_seconds=30)
    ms = []
    for _ in range(runs):
        t = time.perf_counter()
        (job,) = q.claim("bench")
        ms.append((time.perf_counter() - t) * 1000)
        q.complete(job)
    q.db.close()
    print(f"\n{label}")
    print("   claim plan: ", " | ".join(claim_plan))
    print("   retire plan:", " | ".join(retire_plan))
    print(f"   claim: median {statistics.median(ms):9.2f} ms over {runs} claims")
    if label.startswith("plain"):
        db.execute("DROP INDEX jobs_plain")

db.close()
print(f"\ndatabase file after the run: {path.stat().st_size / 1e6:.0f} MB")

python step08_claim_index.py

seeded 1,000,000 done and 1,000 queued jobs in 2.5s

no index
   claim plan:  SCAN jobs | USE TEMP B-TREE FOR ORDER BY
   retire plan: SCAN jobs
   claim: median    111.24 ms over 10 claims

plain index on (visible_at, id)
   claim plan:  SEARCH jobs USING INDEX jobs_plain (visible_at<?)
   retire plan: SEARCH jobs USING INDEX jobs_plain (visible_at<?)
   claim: median    278.02 ms over 10 claims

the partial indexes in jobqueue.py
   claim plan:  SEARCH jobs USING INDEX jobs_ready (visible_at<?)
   retire plan: SEARCH jobs USING INDEX jobs_exhausted (visible_at<?)
   claim: median      0.13 ms over 200 claims

database file after the run: 78 MB

Read the three results:

• No index: SQLite scans all million rows and sorts the matching ones, 111 ms per claim.
• A plain index on (visible_at, id) is slower, not faster, at 278 ms. The plan says it uses the index, but the index lists rows starting from the smallest visible_at, and the oldest rows are all finished jobs. SQLite walks through about a million index entries, looks up each row in the table to check its state, and rejects it, before it reaches the first queued job.
• The partial indexes in jobqueue.py take 0.13 ms, more than 800 times faster than no index and more than 2,000 times faster than the plain index. The SQLite documentation defines the idea in one sentence: “A partial index is an index over a subset of the rows of a table.” Ours covers only queued and running jobs, so a million finished rows add nothing to it.

The catch with partial indexes

A partial index has a catch. SQLite uses it only when it can prove that your query’s WHERE clause implies the index’s WHERE clause, and the manual admits that its prover is simple: “SQLite does not have a sophisticated theorem prover.” It applies a few simple rules instead, among them: “If W is AND-connected terms and X is OR-connected terms and if any term of W appears as a term of X, then the partial index is usable.” My first version of the retire statement said state = 'running', which is not the same term as the index’s state IN ('queued', 'running'), and the plan showed a full table scan on every claim. The script below prints the plan of every statement the queue runs. Save it as check_claims.py.

"""check_claims.py: facts the article states about the sqlite3 module and the partial indexes."""
import sqlite3

from jobqueue import TABLE, JobQueue
from labutil import db_path

path = db_path("claims")
q = JobQueue(path)
q.enqueue_many("t", [{}] * 5)
other = sqlite3.connect(path, isolation_level=None, timeout=0.3)


def try_write(label):
    try:
        other.execute("UPDATE jobs SET attempts = attempts + 1 WHERE id = 5")
        print(f"{label}: a second connection could write")
    except sqlite3.OperationalError as exc:
        print(f"{label}: a second connection was blocked ({exc})")


cur = q.db.execute("UPDATE jobs SET attempts = attempts + 1 WHERE id <= 3 RETURNING id")
cur.fetchone()
try_write("after fetchone() on a 3-row RETURNING")
cur.fetchall()
try_write("after fetchall() on the same cursor")

cur = q.db.execute("UPDATE jobs SET attempts = attempts + 1 WHERE id = 1 RETURNING id")
cur.fetchone()
try_write("after fetchone() on a 1-row RETURNING")

print()
CLAIM = ("SELECT id FROM jobs WHERE state IN ('queued', 'running') AND visible_at <= ? "
         "ORDER BY visible_at, id LIMIT 1")
RETIRE = "SELECT id FROM jobs WHERE state = 'running' AND attempts >= max_attempts AND visible_at <= ?"
queries = {
    "claim sub-select": CLAIM,
    "retire exhausted": RETIRE,
    "count dead": "SELECT COUNT(*) FROM jobs WHERE state = 'dead'",
    "count done": "SELECT COUNT(*) FROM jobs WHERE state = 'done'",
}
for name, sql in queries.items():
    params = (1.0,) if "?" in sql else ()
    plan = [row[3] for row in q.db.execute("EXPLAIN QUERY PLAN " + sql, params)]
    print(f"{name:<18} {' | '.join(plan)}")

print()
ready_only = TABLE + "CREATE INDEX jobs_ready ON jobs (visible_at, id) WHERE state IN ('queued', 'running');"
q2 = JobQueue(db_path("claims_ready_only"), schema=ready_only)
plan = [row[3] for row in q2.db.execute("EXPLAIN QUERY PLAN " + RETIRE, (1.0,))]
print(f"{'retire, jobs_ready only':<24} {' | '.join(plan)}")

import time

empty = JobQueue(db_path("claims_empty"))
t = time.perf_counter()
for _ in range(2000):
    empty.claim("w")
print()
print(f"an empty claim() takes {(time.perf_counter() - t) / 2000 * 1e6:.0f} microseconds")

python check_claims.py

after fetchone() on a 3-row RETURNING: a second connection was blocked (database is locked)
after fetchall() on the same cursor: a second connection could write
after fetchone() on a 1-row RETURNING: a second connection could write

claim sub-select   SEARCH jobs USING INDEX jobs_ready (visible_at<?)
retire exhausted   SEARCH jobs USING INDEX jobs_exhausted (visible_at<?)
count dead         SCAN jobs USING COVERING INDEX jobs_dead
count done         SCAN jobs

retire, jobs_ready only  SCAN jobs

an empty claim() takes 30 microseconds

This output answers four questions.

• The last plan line shows the retire statement with only jobs_ready in place: SCAN jobs. That is why the schema also has jobs_exhausted, an index whose WHERE clause is copied from the retire statement. It holds almost no rows, because the set of running jobs with no attempts left is nearly always empty, and with it the retire statement shows a SEARCH in the plan lines above.
• Counting done rows is SCAN jobs, a full scan. This is why stats() in Step 10 never counts finished jobs. Counting dead rows uses the small jobs_dead index, so it stays cheap.
• The first three lines are a trap in Python’s sqlite3 module. An UPDATE ... RETURNING that has handed back only one of its three rows has not finished, and in autocommit mode the implicit transaction ends only when the statement finishes. A second connection was blocked with “database is locked” until the cursor was drained. A one-row result finished immediately. Always call fetchall() on a RETURNING cursor, as claim does.
• The last line says an empty claim() takes about 30 microseconds, so an idle worker that polls every 50 milliseconds costs almost nothing.

Finished jobs cost the claim path nothing now, but they still take space: the database file was 78 MB after this run. Delete or archive finished jobs in small batches. This tutorial does not include a purge function.

Check: your no index and plain index results are tens or hundreds of milliseconds and the partial-index result is a fraction of a millisecond, and the plan for the retire statement says jobs_exhausted.

Step 9: Measure throughput and choose durability

How many jobs per second can one SQLite file hand out and finish? This script loads 10,000 no-op jobs, so it measures the queue’s overhead alone, and drains them with different numbers of worker processes, different claim sizes and the two durability settings. Save it as step09_throughput.py.

"""step09_throughput.py: how many jobs per second can one SQLite file hand out and finish?"""
import multiprocessing as mp
import os
import time

from jobqueue import JobQueue
from labutil import db_path

N_JOBS = 10_000
CONFIGS = [  # (workers, jobs per claim, synchronous)
    (1, 1, "NORMAL"), (2, 1, "NORMAL"), (4, 1, "NORMAL"), (8, 1, "NORMAL"),
    (4, 10, "NORMAL"), (1, 1, "FULL"), (4, 1, "FULL"),
]


def drain(path, worker, batch, synchronous, start_at, results):
    q = JobQueue(path, synchronous=synchronous, lease_seconds=60)
    while time.time() < start_at:
        time.sleep(0.001)
    done = 0
    while True:
        jobs = q.claim(worker, limit=batch)
        if not jobs:
            break
        for job in jobs:
            q.complete(job)
            done += 1
    results.put((done, time.time()))


def measure(workers, batch, synchronous):
    path = db_path(f"tp_{workers}_{batch}_{synchronous.lower()}")
    q = JobQueue(path)
    q.enqueue_many("noop", [{}] * N_JOBS)
    q.db.close()
    ctx = mp.get_context("spawn")
    results = ctx.Queue()
    start_at = time.time() + 3.0
    procs = [ctx.Process(target=drain, args=(str(path), f"w{i}", batch, synchronous, start_at, results))
             for i in range(workers)]
    for p in procs:
        p.start()
    rows = [results.get() for _ in procs]
    for p in procs:
        p.join()
    return sum(r[0] for r in rows), max(r[1] for r in rows) - start_at


if __name__ == "__main__":
    print(f"{os.cpu_count()} logical CPUs, {N_JOBS:,} no-op jobs per run\n")
    print("workers  per claim  synchronous  jobs/s  seconds")
    for workers, batch, sync in CONFIGS:
        done, seconds = measure(workers, batch, sync)
        assert done == N_JOBS, done
        print(f"{workers:>7}  {batch:>9}  {sync:<11} {done / seconds:>7.0f}  {seconds:7.2f}")

python step09_throughput.py

14 logical CPUs, 10,000 no-op jobs per run

workers  per claim  synchronous  jobs/s  seconds
      1          1  NORMAL         3392     2.95
      2          1  NORMAL         3277     3.05
      4          1  NORMAL         3125     3.20
      8          1  NORMAL         2771     3.61
      4         10  NORMAL         8670     1.15
      1          1  FULL            422    23.68
      4          1  FULL            421    23.76

Three things stand out, from one run each on a 14-thread Windows machine:

• More workers do not add throughput. One worker handled 3,392 jobs per second and eight handled 2,771. Every job needs two write transactions, one to claim and one to complete, and SQLite allows one writer at a time, so extra workers only queue up for the lock. Real handlers spend their time doing real work, so workers still help when jobs are slow. For tiny jobs, the file is the limit.
• Claiming ten jobs at a time raised throughput to 8,670 jobs per second. Ten jobs now cost eleven transactions instead of twenty. The price is that a worker holds ten leases from the same moment, so the tenth job waits behind nine others while its lease runs.
• Forcing every commit to disk cost about eight times the speed. With synchronous = FULL the queue handled 422 jobs per second, with one worker or four. The SQLite documentation describes what you give up with NORMAL: “A transaction committed in WAL mode with synchronous=NORMAL might roll back following a power loss or system crash. Transactions are durable across application crashes regardless of the synchronous setting or journal mode.” Everything we crashed in Step 3 was an application crash. Choose FULL if losing the last few jobs in a power cut is unacceptable, for example when each job is a payment. The same page says: “If durability is not a concern, then synchronous=NORMAL is normally all one needs in WAL mode.”

Treat these numbers as orders of magnitude, not as a benchmark: one run each, one machine, empty handlers. Run the script on your hardware before you decide.

Check: your FULL rows are several times slower than your NORMAL rows, and the throughput barely changes between 1 and 8 workers.

Step 10: Watch the queue and put dead jobs back

Two more methods finish the module. Append them to the JobQueue class. requeue_dead gives dead jobs a fresh set of attempts, after you have fixed whatever killed them. stats reads the live part of the table, using only the small indexes from Step 8.

# jobqueue.py (continued)
    def requeue_dead(self, ids=None):
        """Give dead jobs a fresh set of attempts. Returns the ids that were requeued."""
        sql = "UPDATE jobs SET state = 'queued', attempts = 0, visible_at = ?, finished_at = NULL WHERE state = 'dead'"
        params = [self.clock()]
        if ids:
            sql += " AND id IN (%s)" % ",".join("?" * len(ids))
            params += list(ids)
        return [r[0] for r in self.db.execute(sql + " RETURNING id", params).fetchall()]

    def stats(self):
        now = self.clock()
        live = {r["state"]: r["n"] for r in self.db.execute(
            "SELECT state, COUNT(*) AS n FROM jobs WHERE state IN ('queued', 'running') GROUP BY state")}
        ready = self.db.execute(
            "SELECT COUNT(*) AS n, MIN(visible_at) AS oldest FROM jobs "
            "WHERE state IN ('queued', 'running') AND visible_at <= ?", (now,)).fetchone()
        dead = self.db.execute("SELECT COUNT(*) AS n FROM jobs WHERE state = 'dead'").fetchone()["n"]
        return {
            "queued": live.get("queued", 0),
            "running": live.get("running", 0),
            "ready_now": ready["n"],
            "oldest_ready_age_s": round(now - ready["oldest"], 1) if ready["oldest"] is not None else 0.0,
            "dead": dead,
        }

The script builds a small queue with jobs in every state and prints the numbers. Save it as step10_ops.py.

"""step10_ops.py: what to watch, and how to put dead jobs back."""
from jobqueue import JobQueue
from labutil import FakeClock, db_path

clock = FakeClock(10_000.0)
q = JobQueue(db_path("ops"), lease_seconds=30, max_attempts=2, clock=clock, rng=lambda: 0.0)
for n in range(5):
    q.enqueue("email", {"n": n})  # jobs 1 to 5
q.enqueue("email", {"n": 5}, delay=600)  # job 6, scheduled for later

(j1,) = q.claim("w1")
q.complete(j1)  # job 1 is done
(j2,) = q.claim("w1")
q.fail(j2, "TimeoutError")  # job 2 fails once, and rng=0 retries it at once
(j2,) = q.claim("w1")
q.fail(j2, "TimeoutError")  # its second failure is its last attempt: dead
(j3,) = q.claim("w1")  # job 3: the worker will "crash" and never report back
clock.now += 31  # ... so its lease runs out
(j4,) = q.claim("w2")  # job 4 is next in line, with a live lease


def show(title):
    print(title)
    for key, value in q.stats().items():
        print(f"  {key:<20} {value}")


show("stats at t+31 s:")
print("requeue_dead():", q.requeue_dead())
show("after requeue_dead():")

python step10_ops.py

stats at t+31 s:
  queued               2
  running              2
  ready_now            2
  oldest_ready_age_s   31.0
  dead                 1
requeue_dead(): [2]
after requeue_dead():
  queued               3
  running              2
  ready_now            3
  oldest_ready_age_s   31.0
  dead                 0

Here is how to read it. Of six jobs, one is done and one is dead (job 2, which failed twice). queued is 2: job 5 is ready and job 6 is scheduled 600 seconds ahead. running is 2: job 4 holds a live lease and job 3 holds a lease that has run out, because its worker “crashed”. ready_now is 2, because it counts jobs a worker could claim this instant: job 5, and job 3 whose lease expired. oldest_ready_age_s is 31 seconds, the time job 5 has been waiting. After requeue_dead() the dead job is queued again.

Three of these numbers deserve an alert. A ready_now value or an oldest_ready_age_s that keeps growing means the workers are falling behind. A dead value above zero means a person needs to look at last_error. And ready_now that stays above the number of queued jobs you expect means leases are expiring, so workers are dying or hanging.

Check: after requeue_dead() the dead count is 0 and queued went from 2 to 3.

Step 11: Test it with a fake clock

A queue is mostly rules about time, and tests that sleep are slow and flaky. Because JobQueue takes a clock argument, the tests move a FakeClock by hand. The random generator is replaced by lambda: 1.0, so every retry delay equals its ceiling and can be asserted exactly. Two tests use the real world instead: four real processes claim 300 jobs, and a real heartbeat thread keeps a short lease alive. Save this as test_jobqueue.py.

"""test_jobqueue.py: behaviour of the queue, driven by a fake clock."""
import multiprocessing as mp
import time

import pytest

from jobqueue import JobQueue
from keepalive import KeepAlive
from labutil import FakeClock


@pytest.fixture
def clock():
    return FakeClock(1000.0)


@pytest.fixture
def q(tmp_path, clock):
    queue = JobQueue(tmp_path / "jobs.db", lease_seconds=30, max_attempts=3,
                     backoff_base=2.0, backoff_cap=60.0, clock=clock, rng=lambda: 1.0)
    yield queue
    queue.db.close()


def row(q, job_id=1):
    return q.db.execute("SELECT * FROM jobs WHERE id = ?", (job_id,)).fetchone()


def take_over(q, clock):
    """worker-a claims, its lease runs out, worker-b claims. Returns both jobs."""
    q.enqueue("t", {})
    (a,) = q.claim("worker-a")
    clock.now += 31
    (b,) = q.claim("worker-b")
    return a, b


def test_claims_in_fifo_order(q):
    ids = [q.enqueue("t", {"n": n}) for n in range(3)]
    assert [j.id for j in q.claim("w", limit=3)] == ids


def test_batch_claim_respects_the_limit(q):
    for n in range(5):
        q.enqueue("t", {"n": n})
    assert [j.id for j in q.claim("w", limit=2)] == [1, 2]
    assert [j.id for j in q.claim("w", limit=10)] == [3, 4, 5]


def test_dedupe_key_ignores_a_second_enqueue(q):
    assert q.enqueue("t", {}, dedupe_key="k") == 1
    assert q.enqueue("t", {}, dedupe_key="k") is None
    assert q.db.execute("SELECT COUNT(*) FROM jobs").fetchone()[0] == 1


def test_delayed_job_waits_for_its_time(q, clock):
    q.enqueue("t", {}, delay=60)
    assert q.claim("w") == []
    clock.now += 60
    assert len(q.claim("w")) == 1


def test_claim_sets_a_lease_and_counts_the_attempt(q):
    q.enqueue("t", {})
    (job,) = q.claim("w1")
    assert (job.attempts, job.lease_expires_at) == (1, 1030.0)
    r = row(q)
    assert (r["state"], r["worker"], r["lease_token"]) == ("running", "w1", job.lease_token)


def test_leased_job_stays_hidden_until_the_lease_ends(q, clock):
    q.enqueue("t", {})
    q.claim("w1")
    clock.now = 1029.9
    assert q.claim("w2") == []
    clock.now = 1030.0
    assert len(q.claim("w2")) == 1


def test_reclaim_counts_again_and_issues_a_new_token(q, clock):
    a, b = take_over(q, clock)
    assert b.attempts == 2
    assert b.lease_token != a.lease_token


def test_late_complete_wins_when_nobody_took_over(q, clock):
    q.enqueue("t", {})
    (a,) = q.claim("w1")
    clock.now += 45
    assert q.complete(a) is True
    assert row(q)["state"] == "done"


def test_stale_complete_is_rejected_after_takeover(q, clock):
    a, b = take_over(q, clock)
    assert q.complete(a) is False
    assert row(q)["state"] == "running"
    assert q.complete(b) is True


def test_stale_fail_cannot_resurrect_a_done_job(q, clock):
    a, b = take_over(q, clock)
    assert q.complete(b) is True
    assert q.fail(a, "boom") is None
    assert row(q)["state"] == "done"
    assert q.claim("w3") == []


def test_effect_commits_together_with_complete(q):
    q.db.execute("CREATE TABLE ledger (n INTEGER)")
    q.enqueue("t", {})
    (job,) = q.claim("w1")
    assert q.complete(job, effect=lambda db: db.execute("INSERT INTO ledger VALUES (1)")) is True
    assert q.db.execute("SELECT COUNT(*) FROM ledger").fetchone()[0] == 1


def test_effect_is_skipped_when_the_lease_was_lost(q, clock):
    a, b = take_over(q, clock)
    ran = []
    assert q.complete(a, effect=lambda db: ran.append(1)) is False
    assert ran == []


def test_failing_effect_rolls_back_and_keeps_the_lease(q):
    q.db.execute("CREATE TABLE ledger (n INTEGER)")
    q.enqueue("t", {})
    (job,) = q.claim("w1")

    def effect(db):
        db.execute("INSERT INTO ledger VALUES (1)")
        raise RuntimeError("downstream error")

    with pytest.raises(RuntimeError):
        q.complete(job, effect=effect)
    assert q.db.execute("SELECT COUNT(*) FROM ledger").fetchone()[0] == 0
    assert row(q)["state"] == "running"
    assert q.fail(job, "downstream error") == "queued"


def test_heartbeat_extends_the_lease(q, clock):
    q.enqueue("t", {})
    (job,) = q.claim("w1")
    clock.now += 20
    assert q.heartbeat(job) is True
    assert row(q)["visible_at"] == 1050.0
    clock.now = 1049.0
    assert q.claim("w2") == []


def test_heartbeat_after_takeover_is_false(q, clock):
    a, b = take_over(q, clock)
    assert q.heartbeat(a) is False
    assert q.heartbeat(b) is True


def test_fail_schedules_a_retry_at_the_backoff_ceiling(q, clock):
    q.enqueue("t", {})
    (job,) = q.claim("w1")
    assert q.fail(job, "boom") == "queued"
    assert row(q)["visible_at"] == 1000.0 + 2.0  # attempt 1: ceiling 2 s, rng pinned to 1.0
    clock.now += 2.0
    (job,) = q.claim("w1")
    assert q.fail(job, "boom") == "queued"
    assert row(q)["visible_at"] == clock.now + 4.0  # attempt 2: ceiling 4 s


def test_backoff_never_exceeds_the_cap(tmp_path, clock):
    q = JobQueue(tmp_path / "cap.db", lease_seconds=30, max_attempts=30,
                 backoff_base=2.0, backoff_cap=60.0, clock=clock, rng=lambda: 1.0)
    q.enqueue("t", {})
    delays = []
    for _ in range(9):
        (job,) = q.claim("w")
        q.fail(job, "boom")
        delays.append(row(q)["visible_at"] - clock.now)
        clock.now += 100
    assert delays == [2.0, 4.0, 8.0, 16.0, 32.0, 60.0, 60.0, 60.0, 60.0]


def test_fail_on_the_last_attempt_is_dead(q, clock):
    q.enqueue("t", {})
    for _ in range(3):
        (job,) = q.claim("w1")
        state = q.fail(job, "boom")
        clock.now += 100
    assert state == "dead"
    r = row(q)
    assert (r["state"], r["last_error"]) == ("dead", "boom")


def test_expired_lease_on_the_last_attempt_is_dead(q, clock):
    q.enqueue("t", {})
    for _ in range(3):
        q.claim("w1")  # the worker crashes every time
        clock.now += 31
    assert q.claim("w2") == []
    r = row(q)
    assert (r["state"], r["attempts"], r["last_error"]) == ("dead", 3, "lease expired on the last attempt")


def test_stats_count_ready_expired_and_dead(q, clock):
    for n in range(4):
        q.enqueue("t", {"n": n})
    q.enqueue("t", {}, delay=600)
    (j1,) = q.claim("w")
    q.complete(j1)
    q.claim("w")  # job 2: its lease will run out
    clock.now += 31
    s = q.stats()
    assert (s["queued"], s["running"], s["dead"]) == (3, 1, 0)
    assert s["ready_now"] == 3  # jobs 3 and 4, plus job 2 whose lease ended
    assert s["oldest_ready_age_s"] == 31.0


def test_requeue_dead_gives_fresh_attempts(q, clock):
    q.enqueue("t", {})
    for _ in range(3):
        (job,) = q.claim("w1")
        q.fail(job, "boom")
        clock.now += 100
    assert q.requeue_dead() == [1]
    (job,) = q.claim("w1")
    assert job.attempts == 1


def claim_all(path, name, out):
    q = JobQueue(path, lease_seconds=60)
    mine = []
    while True:
        jobs = q.claim(name)
        if not jobs:
            break
        mine += [j.id for j in jobs]
    out.put(mine)


def test_four_processes_never_claim_the_same_job(tmp_path):
    path = str(tmp_path / "race.db")
    q = JobQueue(path)
    q.enqueue_many("t", [{}] * 300)
    q.db.close()
    ctx = mp.get_context("spawn")
    out = ctx.Queue()
    procs = [ctx.Process(target=claim_all, args=(path, f"w{i}", out)) for i in range(4)]
    for p in procs:
        p.start()
    claimed = [job_id for _ in procs for job_id in out.get()]
    for p in procs:
        p.join()
    assert sorted(claimed) == list(range(1, 301))


def test_keepalive_thread_extends_the_lease(tmp_path):
    path = tmp_path / "keep.db"
    a = JobQueue(path, lease_seconds=0.6)
    a.enqueue("t", {})
    (job,) = a.claim("w1")
    b = JobQueue(path, lease_seconds=0.6)
    with KeepAlive(path, job, lease_seconds=0.6, interval=0.1) as hb:
        time.sleep(1.5)  # two and a half lease lengths
        assert b.claim("w2") == []
        assert hb.beats >= 5 and not hb.lost.is_set()
    time.sleep(0.8)  # the heartbeats have stopped, so the lease runs out
    assert len(b.claim("w2")) == 1
    a.db.close()
    b.db.close()

python -m pytest -q test_jobqueue.py

....................... [100%]
23 passed in 3.23s

The tests pin down every rule from the steps above: first-in-first-out claiming, delayed jobs, lease timing, reclaiming with a new token, the late-but-unreplaced completion, rejection of stale completions, failures and heartbeats, the retry ceiling and its cap, the dead-letter paths (including the poison job), effects that commit with the completion or not at all, statistics, requeueing, and no double claims across four processes. If you change the module later, these tests tell you what you broke.

Check: 23 passed.

Step 12: Confirm everything works end to end

The last test is a stress run that combines everything. It enqueues 300 jobs of four kinds, shuffled together, and runs them with eight worker processes while a supervisor replaces any worker that dies. The remaining handlers go into handlers.py.

# handlers.py (continued)
# --- more handlers, used by the end-to-end run at the end of this tutorial


def ledger_effect(job):
    def effect(db):
        db.execute("INSERT INTO ledger (job_id, attempt) VALUES (?, ?)", (job.id, job.attempts))
    return effect


def credit(job, q):
    if job.attempts == 1 and job.id % 20 == 0:
        raise TimeoutError("upstream timed out")  # a transient failure that a retry absorbs
    time.sleep(0.02)
    return ledger_effect(job)


def slow(job, q):
    # The first attempt stalls for far longer than the 0.8 second lease. Later attempts are quick.
    time.sleep(2.0 if job.attempts == 1 else 0.1)
    return ledger_effect(job)


def bad(job, q):
    raise ValueError("payload has no recipient")


HANDLERS.update(credit=credit, slow=slow, bad=bad)

• credit (260 jobs) writes one row to a ledger table inside the completing transaction. About one credit job in twenty (13 of the 260 in this run) fails once with a timeout first, and the retry absorbs it.
• slow (6 jobs) stalls for two seconds on its first attempt, far beyond the 0.8-second lease, and is quick afterwards.
• poison (10 jobs) kills its worker.
• bad (24 jobs) always raises.

When the queue drains, the script checks the invariants: credit and slow jobs must be done with exactly one ledger row each, and poison and bad jobs must be dead with no ledger rows. Save it as step12_chaos.py.

"""step12_chaos.py: end-to-end check with hard crashes, a stalled first attempt and failing jobs."""
import random
import subprocess
import sys
import time
from collections import Counter
from pathlib import Path

from jobqueue import JobQueue
from labutil import db_path, kill_tree

HERE = Path(__file__).resolve().parent
LEASE = 0.8
WORKERS = 8
MIX = {"credit": 260, "slow": 6, "poison": 10, "bad": 24}  # 300 jobs
EXPECTED = {"credit": "done", "slow": "done", "poison": "dead", "bad": "dead"}

path = db_path("chaos")
q = JobQueue(path, lease_seconds=LEASE, max_attempts=3)
q.db.execute("CREATE TABLE ledger (job_id INTEGER NOT NULL, attempt INTEGER NOT NULL)")
order = [kind for kind, count in MIX.items() for _ in range(count)]
random.Random(42).shuffle(order)  # spread the awkward jobs through the whole run
for kind in order:
    q.enqueue(kind, {})


def live():
    return q.db.execute("SELECT COUNT(*) FROM jobs WHERE state IN ('queued', 'running')").fetchone()[0]


started = time.monotonic()
slots, spawned, crashed = [None] * WORKERS, 0, 0
while live() > 0:
    for i, proc in enumerate(slots):
        if proc is not None and proc.poll() is not None:
            crashed += proc.returncode != 0
            proc = None
        if proc is None:  # keep the pool at full size, replacing any that died
            slots[i] = subprocess.Popen(
                [sys.executable, str(HERE / "worker.py"), str(path), f"w{spawned}", str(LEASE), "1.0"])
            spawned += 1
    time.sleep(0.1)
elapsed = time.monotonic() - started
for proc in slots:
    kill_tree(proc)

jobs = q.db.execute("SELECT id, kind, state, attempts FROM jobs").fetchall()
ledger = Counter(r[0] for r in q.db.execute("SELECT job_id FROM ledger"))
winning = {r[0]: r[1] for r in q.db.execute("SELECT job_id, attempt FROM ledger")}
problems = []
for job in jobs:
    want = EXPECTED[job["kind"]]
    if job["state"] != want:
        problems.append(f"job {job['id']} ({job['kind']}) is {job['state']}, expected {want}")
    if ledger.get(job["id"], 0) != (1 if want == "done" else 0):
        problems.append(f"job {job['id']} ({job['kind']}) has {ledger.get(job['id'], 0)} ledger rows")

print(f"{len(jobs)} jobs ({', '.join(f'{n} {k}' for k, n in MIX.items())}), {WORKERS} workers, lease {LEASE} s")
print(f"finished in {elapsed:.1f} s; {spawned} worker processes started, {crashed} exited with a crash")
print("states:", dict(sorted(Counter(j["state"] for j in jobs).items())))
print(f"ledger rows: {sum(ledger.values())}; jobs with more than one ledger row: {sum(1 for n in ledger.values() if n > 1)}")
retried = sum(1 for j in jobs if j["kind"] == "credit" and j["attempts"] > 1)
print(f"credit jobs that needed a retry: {retried}")
slow_ids = [j["id"] for j in jobs if j["kind"] == "slow"]
by_attempt = Counter(winning[i] for i in slow_ids if i in winning)
print("slow jobs, by the attempt whose effect was recorded:", dict(sorted(by_attempt.items())))
print("attempts used by the dead jobs:", dict(Counter((j['kind'], j['attempts']) for j in jobs if j['state'] == 'dead')))
print("PASS: every job finished exactly once or was dead-lettered" if not problems else "FAIL: " + "; ".join(problems[:5]))

python step12_chaos.py

300 jobs (260 credit, 6 slow, 10 poison, 24 bad), 8 workers, lease 0.8 s
finished in 4.2 s; 39 worker processes started, 30 exited with a crash
states: {'dead': 34, 'done': 266}
ledger rows: 266; jobs with more than one ledger row: 0
credit jobs that needed a retry: 13
slow jobs, by the attempt whose effect was recorded: {2: 6}
attempts used by the dead jobs: {('bad', 3): 24, ('poison', 3): 10}
PASS: every job finished exactly once or was dead-lettered

The run took about four seconds. Thirty worker processes died mid-job: ten poison jobs times three attempts each. All 266 jobs that should succeed did (260 credit and 6 slow) and all 34 that should not were dead-lettered (24 bad and 10 poison, each after exactly three attempts). The ledger holds 266 rows, one per finished job, and no job has more than one row, even though the six slow jobs were each run twice. The slow-job line shows why: for every one of them the recorded effect came from attempt 2. A stalled first attempt could not have been accepted, or the ledger would show attempt 1, so each stalled worker’s late completion was refused after another worker had taken over. The final line says PASS.

One honest observation about this run. In an earlier version, with four workers and fifteen slow jobs, fourteen of the fifteen were finished late by their original worker, which fits every worker being busy stalling with none free to take over. A lease can hand a job over only when some other worker is free to claim it, so size your worker pool with that in mind.

Check: the last line says PASS, the ledger shows no job with more than one row, and python -m pytest -q test_jobqueue.py still prints 23 passed. If you edit the module and either check fails, the stress run prints the first few violations.

What each piece protects against

Limits and gotchas

• One machine only. The SQLite WAL page says: “All processes using a database must be on the same host computer; WAL does not work over a network filesystem.” For workers on several machines use PostgreSQL or a managed queue.
• Clocks. Every worker compares visible_at with its machine’s clock. On one host that is a single clock, but if someone sets the system time backwards or forwards, every lease shifts with it. I did not test that, so treat it as a risk to think about, not a measured result.
• SQLite version. Use a build that includes the WAL-reset fix from Step 0.
• Polling. SQLite has no built-in way to wake another process when a job arrives, so workers poll. A job can wait up to one poll interval. An empty claim takes microseconds, but it still takes the write lock briefly.
• Ordering is approximate. Claims go by visible_at and then id. A retried or reclaimed job gets a later visible_at, so it lines up behind jobs that were already waiting.
• Batch claims share a deadline. If a worker claims ten jobs, all ten leases end at the same moment, so size the lease for the whole batch or send heartbeats for each job.
• Match handler timeouts to the lease. A handler that can run longer than its lease needs a heartbeat, or its work will be duplicated.
• Exactly-once applies to database effects only, and only when they use the effect argument. Everything else is at-least-once.
• Housekeeping and features are missing. This module has no purge for finished jobs, no priorities and no fairness between job kinds.

Next steps

You now have a queue that survives crashed, slow and late workers, with tests and a stress run to prove it. To keep learning:

• When producers can outrun workers, decide what to refuse before the queue grows without bound. The backpressure and load shedding tutorial covers that side.
• When one business action spans several systems, compare the saga pattern and two-phase commit with the single-database trick in Step 5.
• The fence in Step 4 is the same idea as the leader term in leader election: a stale holder must not be able to act.
• If your queue sits next to a write-ahead log or other durability code, the write-ahead log tutorial shows what SQLite’s WAL does for you.
• Read the sources below, especially Kleppmann’s article on fencing and the Amazon SQS pages on visibility timeouts and dead-letter queues.

Sources and method

Documentation quoted above: the Amazon SQS pages on visibility timeouts, queue and message identifiers and dead-letter queues; the PostgreSQL SELECT reference; the SQLite pages on RETURNING, transactions, write-ahead logging, partial indexes and pragmas; the Python sqlite3 documentation; and Martin Kleppmann’s How to do distributed locking.

How the numbers were produced: Windows 11 with 14 logical CPUs, Python 3.13.14, SQLite 3.50.4 and pytest 9.1.1, one run of each script. The outputs of Steps 1, 3, 6, 8, 9 and 12 contain timings or counts that change from run to run; the others use the fake clock and repeat exactly. I rebuilt the lab from the code blocks on this page in an empty folder and a fresh virtual environment, ran every step and the tests, and compared every printed line. I ran nothing on Linux or macOS.