Background Jobs for Long AI Tasks

Anything that takes longer than a browser will wait needs a job id, a state machine and a worker. Build it on a table you can query before you build it on a framework you cannot see into — the framework is a twenty-line change later, and the state machine is the part you will actually debug.

When a request cycle stops being enough

The threshold is not a duration, it is a set of properties. Move to a job queue when any of these is true:

• The work outlives a plausible HTTP timeout. Load balancers, CDNs and platform gateways commonly cut connections at 30, 60 or 100 seconds regardless of what your server does.
• The work must survive a deploy. A ten-minute batch in a request handler dies with the process on every release.
• The result is worth keeping. If a user would reasonably expect to close the tab and come back, the result needs an address of its own.
• The work must be rate-limited across users. A queue is where a global concurrency limit can actually live; a request handler has no idea what other requests are doing.

If the answer is under thirty seconds and the user is watching, stream it instead — a FastAPI endpoint that streams is a much better experience than a spinner over a job poll.

The state machine, written out

Five states. Writing them down before the code is what stops the “is it stuck or still going?” conversation later.

  queued ──claim──▶ running ──success──▶ succeeded   (terminal)
    ▲                 │
    │                 ├──retryable failure, attempts < max──┐
    └─────────────────┘                                     │
                      │                                     │
                      ├──permanent failure──▶ failed  ───────┤ (terminal)
                      │                                      │
                      └──worker died, lease expired──────────┘
                                                             │
                                        back to queued ◀──────┘

Invariants:
  * exactly one worker may hold a job in RUNNING
  * every non-terminal job has a lease_expires_at in the future
  * a job in RUNNING past its lease is available to be reclaimed
  * FAILED is never reached without a stored error message

The lease is the piece most hand-rolled queues omit, and its absence is why jobs sit at “running” for ever after a worker is OOM-killed. A worker does not own a job; it holds a time-limited claim it has to keep renewing.

The other consequence of the diagram is that a job can run twice — a worker that lost its lease may still be working when another claims the same row. So the work has to be safe to repeat, which is the same requirement idempotency imposes on any retried operation, and the reason a job that also sends an email needs the send keyed on the job id. Background jobs for AI features covers the same shape from the product side, and partial failure covers what to show a user while a job is half done.

The job table

SQLite is enough to run this in production for a single-writer workload, and the schema translates to Postgres unchanged apart from the types.

-- schema.sql
CREATE TABLE IF NOT EXISTS jobs (
  id                TEXT PRIMARY KEY,
  kind              TEXT NOT NULL,
  payload           TEXT NOT NULL,          -- JSON
  state             TEXT NOT NULL DEFAULT 'queued',
  attempts          INTEGER NOT NULL DEFAULT 0,
  max_attempts      INTEGER NOT NULL DEFAULT 3,
  progress_done     INTEGER NOT NULL DEFAULT 0,
  progress_total    INTEGER NOT NULL DEFAULT 0,
  result            TEXT,                   -- JSON, set on success
  error             TEXT,                   -- set on failure
  lease_expires_at  REAL,                   -- unix seconds
  created_at        REAL NOT NULL,
  updated_at        REAL NOT NULL
);

CREATE INDEX IF NOT EXISTS jobs_claimable
  ON jobs (state, lease_expires_at);

# jobs.py
import json
import sqlite3
import time
import uuid

DB = "jobs.db"
LEASE_SECONDS = 120.0


def connect() -> sqlite3.Connection:
    conn = sqlite3.connect(DB, isolation_level=None)  # autocommit; explicit BEGIN
    conn.row_factory = sqlite3.Row
    conn.execute("PRAGMA journal_mode=WAL")
    conn.execute("PRAGMA busy_timeout=5000")
    return conn


def enqueue(conn: sqlite3.Connection, kind: str, payload: dict,
            total: int = 0) -> str:
    job_id = uuid.uuid4().hex
    now = time.time()
    conn.execute(
        "INSERT INTO jobs (id, kind, payload, progress_total, created_at, updated_at)"
        " VALUES (?, ?, ?, ?, ?, ?)",
        (job_id, kind, json.dumps(payload), total, now, now),
    )
    return job_id


def claim(conn: sqlite3.Connection) -> sqlite3.Row | None:
    """Atomically take one claimable job. Returns None if there is nothing to do."""
    now = time.time()
    conn.execute("BEGIN IMMEDIATE")
    try:
        row = conn.execute(
            "SELECT * FROM jobs"
            " WHERE (state = 'queued')"
            "    OR (state = 'running' AND lease_expires_at < ?)"
            " ORDER BY created_at LIMIT 1",
            (now,),
        ).fetchone()
        if row is None:
            conn.execute("COMMIT")
            return None
        conn.execute(
            "UPDATE jobs SET state='running', attempts = attempts + 1,"
            " lease_expires_at = ?, updated_at = ? WHERE id = ?",
            (now + LEASE_SECONDS, now, row["id"]),
        )
        conn.execute("COMMIT")
    except Exception:
        conn.execute("ROLLBACK")
        raise
    return conn.execute("SELECT * FROM jobs WHERE id = ?", (row["id"],)).fetchone()

BEGIN IMMEDIATE is what makes the claim atomic: it takes the write lock before the SELECT, so two workers cannot both read the same queued row and both claim it. A plain BEGIN defers the lock and permits exactly that race. On Postgres the equivalent is SELECT ... FOR UPDATE SKIP LOCKED.

The worker loop

# worker.py
import json
import time
import traceback

from jobs import connect, claim, LEASE_SECONDS

POLL_SECONDS = 1.0


class PermanentError(Exception):
    """Do not retry: bad input, a 400, an unsupported model."""


def renew(conn, job_id: str, done: int) -> None:
    conn.execute(
        "UPDATE jobs SET lease_expires_at = ?, progress_done = ?, updated_at = ?"
        " WHERE id = ?",
        (time.time() + LEASE_SECONDS, done, time.time(), job_id),
    )


def run_job(conn, job) -> dict:
    payload = json.loads(job["payload"])
    items = payload["items"]
    results = []
    for index, item in enumerate(items, start=1):
        results.append(process_one(item))     # your model call
        if index % 5 == 0:
            renew(conn, job["id"], index)     # heartbeat AND progress
    return {"results": results}


def main() -> None:
    conn = connect()
    while True:
        job = claim(conn)
        if job is None:
            time.sleep(POLL_SECONDS)
            continue
        try:
            result = run_job(conn, job)
        except PermanentError as exc:
            conn.execute(
                "UPDATE jobs SET state='failed', error=?, updated_at=? WHERE id=?",
                (f"permanent: {exc}", time.time(), job["id"]),
            )
        except Exception:
            detail = traceback.format_exc(limit=5)
            if job["attempts"] >= job["max_attempts"]:
                conn.execute(
                    "UPDATE jobs SET state='failed', error=?, updated_at=? WHERE id=?",
                    (detail, time.time(), job["id"]),
                )
            else:
                conn.execute(
                    "UPDATE jobs SET state='queued', error=?, lease_expires_at=NULL,"
                    " updated_at=? WHERE id=?",
                    (detail, time.time(), job["id"]),
                )
        else:
            conn.execute(
                "UPDATE jobs SET state='succeeded', result=?, error=NULL,"
                " progress_done=progress_total, updated_at=? WHERE id=?",
                (json.dumps(result), time.time(), job["id"]),
            )


if __name__ == "__main__":
    main()

The heartbeat and the progress update are the same statement on purpose. Two separate mechanisms drift — you get a job that reports 90 per cent and has been dead for ten minutes — and one statement cannot.

For a job that processes many items, make process_one idempotent and record which items are done, so a reclaimed job resumes rather than restarting. That halves the cost of every crash and it is the same discipline as checkpointing a large classification run.

Progress the client can poll

@app.post("/jobs")
def create_job(body: JobRequest) -> dict:
    conn = connect()
    job_id = enqueue(conn, "classify", {"items": body.items}, total=len(body.items))
    return {"job_id": job_id, "state": "queued"}


@app.get("/jobs/{job_id}")
def get_job(job_id: str) -> dict:
    conn = connect()
    row = conn.execute("SELECT * FROM jobs WHERE id = ?", (job_id,)).fetchone()
    if row is None:
        raise HTTPException(status_code=404, detail="no such job")
    return {
        "job_id": row["id"],
        "state": row["state"],
        "done": row["progress_done"],
        "total": row["progress_total"],
        "result": json.loads(row["result"]) if row["result"] else None,
        "error": row["error"] if row["state"] == "failed" else None,
        "stale": (
            row["state"] == "running"
            and (row["lease_expires_at"] or 0) < time.time()
        ),
    }

The stale field costs one comparison and answers the question the user actually has. A progress bar that says 40 per cent tells you nothing about whether anything is still happening; stale: true means the worker holding this job stopped renewing its lease and the job is about to be reclaimed.

Poll with a backoff — every second for the first ten, then every five — rather than a fixed one-second interval, or a hundred idle browser tabs become a hundred requests per second against your database.

A dead-letter path that gets read

Every queue has a place failed jobs go. Most have no place anybody looks, which is the same as not having one. Four things make the difference:

1. Store the error, not a flag. The error column above holds a truncated traceback. A boolean failed column means the only way to find out what happened is to reproduce it.
2. Make it one query. SELECT kind, substr(error, 1, 80), count(*) FROM jobs WHERE state = 'failed' GROUP BY 1, 2 ORDER BY 3 DESC takes a second and tells you whether you have one bug or forty. Put it in a script in the repository, not in somebody’s shell history.
3. Alert on the rate, not the count. A total failed count only ever grows, so nobody watches it. The number worth an alert is failures in the last hour, or the ratio of failed to succeeded.
4. Make replay a command. Setting state back to queued and attempts back to zero for a set of ids is the entire replay mechanism. It has to exist before the incident, because during the incident nobody writes it.

SQLite, Postgres or Redis

The state machine above does not care where it lives. What changes with the store is how many workers you can run and what a crash costs, and the three answers are genuinely different rather than a ladder you climb.

Two properties matter more than throughput for this workload. AI jobs are minutes long, so a queue handling ten claims a second is overwhelming capacity — the bottleneck is never the queue. And the results are worth keeping, which is an argument for a store you can query six weeks later rather than one tuned for delivery.

Whichever you pick, do not put the model output in the job row if it is large. A result column holding 40 KB of text per job turns the jobs table into a document store and makes every status poll read it. Write the payload to object storage or a separate table and keep a reference.

The same thing in Celery or RQ

Once the state machine is clear, adopting a framework is a mapping exercise rather than a redesign. Both of these need a broker — Redis or RabbitMQ — which is the real cost of moving.