What a Queue Adds Beyond a Worker Loop

Lesson 6 built a worker loop that polled an in-memory JobStore for pending jobs. That was enough to demonstrate the core mechanism, but it has real limitations worth naming precisely: if the process holding the JobStore restarts, every job in it is lost; polling a shared dictionary does not scale cleanly to multiple worker processes running on different machines; and there is no way to retry a job that failed due to a transient error, or to stop retrying one that fails consistently.

A queue is a durable, ordered (or at least fairly ordered) store of pending work items that lives independently of any single producer or consumer process. Producers (your web server, handling a user's request) add jobs to the queue and return immediately. Consumers (worker processes) take jobs off the queue, process them, and report success or failure back to the queue system. Because the queue itself is durable — typically backed by Redis, a managed message broker, or a database table designed for this purpose — a worker process can crash and restart without losing track of what work remains to be done.

The Producer/Consumer Job Architecture

The general shape of this architecture, regardless of which specific queue technology backs it, looks like this:

1. A client sends a request that requires long-running work (for example, "summarize this document").
2. The web server validates the request, generates a job id, enqueues the job, and immediately responds with that job id — this response takes milliseconds, not minutes.
3. One or more worker processes, running independently, pull jobs off the queue and execute them.
4. The client polls a status endpoint (GET /jobs/{job_id}) with the job id, or the system notifies the client via a webhook once the job completes.

import uuid
import queue
from dataclasses import dataclass
from enum import Enum


class JobStatus(str, Enum):
    QUEUED = "queued"
    RUNNING = "running"
    DONE = "done"
    FAILED = "failed"


@dataclass
class Job:
    id: str
    prompt: str
    status: JobStatus = JobStatus.QUEUED
    result: str | None = None
    attempts: int = 0


class JobQueue:
    """A minimal in-process stand-in for a real message queue (Redis, SQS,
    RabbitMQ). Real queues add durability across restarts and support for
    multiple consumer processes on different machines; the enqueue/dequeue
    contract shown here is the same shape either way."""

    def __init__(self) -> None:
        self._queue: "queue.Queue[str]" = queue.Queue()
        self._jobs: dict[str, Job] = {}

    def enqueue(self, prompt: str) -> Job:
        job = Job(id=str(uuid.uuid4()), prompt=prompt)
        self._jobs[job.id] = job
        self._queue.put(job.id)
        return job

    def dequeue(self, timeout: float | None = None) -> Job | None:
        try:
            job_id = self._queue.get(timeout=timeout)
        except queue.Empty:
            return None
        return self._jobs[job_id]

    def get(self, job_id: str) -> Job | None:
        return self._jobs.get(job_id)

The important contract here is the separation between enqueue (called by the producer, the web request handler) and dequeue (called by the consumer, the worker). Neither side needs to know anything about the other's implementation — the web server does not know how many workers exist or when they will pick up the job, and a worker does not know which request handler produced any given job. This decoupling is precisely what allows the two sides to be scaled, deployed, and restarted independently, which Lesson 8 relies on directly for horizontal scaling.

Retries and Idempotency

Real queue systems distinguish between a job failing because of a transient problem (a momentary network blip, a rate-limited API call) and a job failing because it is fundamentally broken (malformed input that will never succeed no matter how many times you retry it). The standard mechanism is a bounded retry count with backoff, combined with a dead-letter queue — a separate holding area for jobs that have exhausted their retries, so they can be inspected manually rather than disappearing silently or retrying forever.

import time


MAX_ATTEMPTS = 3


def process_with_retries(client, job: Job, model: str) -> None:
    while job.attempts < MAX_ATTEMPTS:
        job.attempts += 1
        job.status = JobStatus.RUNNING
        try:
            response = client.responses.create(model=model, input=job.prompt)
            job.result = response.output_text
            job.status = JobStatus.DONE
            return
        except Exception:
            if job.attempts >= MAX_ATTEMPTS:
                job.status = JobStatus.FAILED
                return
            time.sleep(2 ** job.attempts)  # exponential backoff: 2s, 4s, ...

The exponential backoff (2 ** job.attempts) is deliberate: retrying immediately after a failure caused by, say, a rate limit only makes the rate-limit situation worse, since the retry itself counts against the same limit almost immediately. Waiting longer between each successive attempt gives a transient problem more time to clear before trying again, which is far more likely to succeed than an immediate retry, and it also reduces the total load your retries place on an already-struggling dependency.

Retrying safely, however, depends on idempotency — the property that running the same operation twice has the same effect as running it once. If a job's side effect is "call the OpenAI API and store the result," retrying that job after a partial failure is usually safe: worst case, you make a duplicate API call and overwrite a stored result with an equivalent one. But if a job's side effect is "call the OpenAI API and then charge the user's account" or "call the OpenAI API and then send them an email," a naive retry after the API call succeeded but the subsequent step failed would double-charge or double-send. The standard fix is an idempotency key: a unique identifier for the logical operation (often the job id itself) that downstream systems use to recognize and discard a duplicate execution rather than repeating its side effects.

def send_result_email_once(email_client, job: Job, sent_job_ids: set[str]) -> None:
    if job.id in sent_job_ids:
        return  # already sent; retrying this job must not send a duplicate email
    email_client.send(to="user@example.com", body=job.result)
    sent_job_ids.add(job.id)

sent_job_ids here plays the role of an idempotency record: before performing a side effect that is not safe to repeat, the code checks whether this specific job id has already triggered it. In a real system, this set would be a persistent store (a database table or a Redis set), not an in-memory Python set, so the check survives a process restart — but the principle is the same regardless of storage: check first, act once, record that you acted.

Job Status Endpoints

Completing the producer/consumer picture requires a way for the original client to find out what happened to their job:

def get_job_status(job_queue: JobQueue, job_id: str) -> tuple[dict, int]:
    job = job_queue.get(job_id)
    if job is None:
        return {"error": "job not found"}, 404

    body = {"job_id": job.id, "status": job.status.value}
    if job.status == JobStatus.DONE:
        body["result"] = job.result
    elif job.status == JobStatus.FAILED:
        body["error"] = f"failed after {job.attempts} attempts"

    return body, 200

This function is intentionally plain — it takes a JobQueue and a job_id and returns a response body and status code, with no dependency on any web framework. That keeps it trivially testable and reusable regardless of whether it is wired up behind FastAPI, Flask, or any other framework the rest of the application uses.

class FakeResponse:
    def __init__(self, text: str) -> None:
        self.output_text = text


class FakeClient:
    def __init__(self, fail_times: int = 0) -> None:
        self._fail_times = fail_times
        self._calls = 0

    class _Responses:
        def __init__(self, outer: "FakeClient") -> None:
            self._outer = outer

        def create(self, model: str, input: str) -> FakeResponse:
            self._outer._calls += 1
            if self._outer._calls <= self._outer._fail_times:
                raise RuntimeError("simulated transient failure")
            return FakeResponse("done: " + input[:10])

    @property
    def responses(self) -> "FakeClient._Responses":
        return FakeClient._Responses(self)


def test_job_succeeds_after_transient_failures() -> None:
    client = FakeClient(fail_times=2)
    job = Job(id="job-1", prompt="Summarize this report")

    process_with_retries(client, job, model="gpt-5.6-terra")

    assert job.status == JobStatus.DONE
    assert job.attempts == 3
    print("PASS: job recovers after transient failures within the retry limit")


def test_job_fails_after_exhausting_retries() -> None:
    client = FakeClient(fail_times=10)
    job = Job(id="job-2", prompt="Summarize this report")

    process_with_retries(client, job, model="gpt-5.6-terra")

    assert job.status == JobStatus.FAILED
    assert job.attempts == MAX_ATTEMPTS
    print("PASS: job is marked failed once retries are exhausted")


test_job_succeeds_after_transient_failures()
test_job_fails_after_exhausting_retries()

Note the tests call time.sleep indirectly through process_with_retries with real (small) backoff delays — acceptable for a lesson, but in a real test suite you would typically inject the sleep function itself (another application of dependency injection) so tests run instantly instead of waiting on real backoff timers.

Common Mistakes

Retrying non-idempotent side effects without a deduplication mechanism. A job that sends a notification, writes a billing record, or has any other effect that should happen exactly once will double-execute that effect on retry unless it explicitly checks whether it already ran, as shown with sent_job_ids above.

Retrying immediately, with no backoff. Immediate retries against a rate-limited or overloaded dependency tend to make the underlying problem worse rather than better, and can turn a brief blip into a sustained outage as retries pile up faster than the dependency can recover.

No dead-letter path for permanently failed jobs. Without one, jobs that can never succeed either retry forever (wasting resources and API quota) or disappear silently once retries are exhausted, leaving no trace for anyone to investigate.

Best Practices

Design every job handler to be safely retryable, either because its side effects are naturally idempotent or because it explicitly tracks what has already been done using an idempotency key.

Use exponential backoff between retry attempts, not immediate or fixed-interval retries, so transient failures get a genuine chance to clear before the next attempt.

Route permanently failed jobs somewhere visible — a dead-letter queue, a failed-jobs table, an alert — rather than letting them vanish after their final failed attempt.