Background Job Queue with Redis Streams

Build a production job queue with asyncio workers, crash recovery, and retries.

Introduction

30 Second Summary

Every time you upload a photo, place an order, or request a report, the app does not make you wait while it crunches the work. A background job queue handles those tasks behind the scenes, picking them up and completing them without you ever noticing.

In this project, you will build a background job queue from scratch using Python, Redis Streams, and FastAPI. You will crash a worker on purpose, watch jobs get stuck, then build the crash recovery and reliability features that every production queue needs.

What You'll Build

You will fire off jobs from your browser, watch workers race to process them, deliberately crash a worker mid-flight, and see the system heal itself as another worker reclaims and completes the orphaned job.

By the end of this project, you'll have:

  • A job submission API where you POST a task and watch it flow into a worker pool that processes it in real time.
  • Proven crash recovery: kill a worker mid-job, see the job stuck in the pending entry list, then watch another worker automatically reclaim and finish it.
  • Production-grade reliability features that block duplicate submissions, retry failing jobs with exponential backoff, and quarantine permanently broken jobs in a dead letter queue you can replay on demand.
  • An optional Secret Mission to push your skills further.

Are there any prerequisites?

This project assumes you have experience with Python and are familiar with async/await syntax. You will need Docker installed on your machine. No Redis experience is required.

Before We Start

Before we dive in, let's lock in what you're about to build and why it matters.

Set Up Redis and Project Scaffold

Your job queue needs Redis as its backbone. Before you can write any queue logic, Redis must be running and reachable from your Python code.

Docker lets you spin up Redis in a single command without installing it directly on your machine. In this step, you'll launch Redis in a container, scaffold your Python project, and wire up a health check that proves the connection works.

In this step, get ready to:
  • Start a Redis container in Docker and verify it responds.
  • Create your Python project with a virtual environment and install dependencies.
  • Build a FastAPI health check endpoint that pings Redis and confirm the connection.
Start Redis in Docker

Redis will store every job, track which worker claimed it, and remember which jobs are still pending. Instead of installing Redis on your machine, you'll run it inside a Docker container.

  • Launch a Redis container by running this command in your terminal:
docker run -d --name jobqueue-redis -p 6379:6379 redis:8

What does this command do?

  • docker run -d starts a container in detached mode, so it runs in the background and gives you your terminal back.
  • --name jobqueue-redis gives the container a name you can reference later.
  • -p 6379:6379 maps Redis's default port to the same port on your machine, so your Python code can reach it at localhost:6379.
  • redis:8 pulls the Redis 8 image from Docker Hub.

Before you run the next command, what do you think Redis will reply if the connection is working?

  • Verify Redis is responding by running:
docker exec jobqueue-redis redis-cli PING

You should see PONG printed in your terminal. That single word confirms Redis is running and accepting connections.

Don't see PONG?

  • Make sure Docker Desktop is running. If it is not open, start it and try the docker run command again.
  • If you see an error about the container name already in use, a previous attempt left a container behind. Remove it with docker rm -f jobqueue-redis and then re-run the docker run command.
  • Check if another process is using port 6379 by running docker ps. If another container is on that port, stop it first.

Still stuck? Help me troubleshoot why my Redis Docker container isn't responding to PING.

Create the project and install dependencies

With Redis running, you need a Python project to talk to it. You'll create a project folder, isolate your dependencies in a virtual environment, and install the three packages your queue needs.

  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir jobqueue
cd jobqueue
  • Confirm the folder exists by running:
pwd

You should see a path ending in jobqueue.

A virtual environment keeps this project's packages separate from your system Python. Create and activate one now.

  • Create and activate a virtual environment by running these commands:

macOS

python3 -m venv .venv
source .venv/bin/activate

Linux users can use the same commands.

Windows

python -m venv .venv
.venv\Scripts\activate

Your terminal prompt should now show (.venv) at the beginning. That confirms the virtual environment is active.

  • Install the three packages your project needs by running:
pip install fastapi redis uvicorn

You should see output ending with Successfully installed followed by a list of packages.

What are these packages?

  • fastapi is the web framework you'll use to build the job submission API.
  • redis is the Python client for Redis. It includes built-in async support so your code can talk to Redis without blocking.
  • uvicorn is the ASGI server that runs your FastAPI app.

Seeing install errors?

  • If you see a permissions error, make sure your virtual environment is active. Look for (.venv) in your terminal prompt.
  • If pip is not found, try pip3 instead.

Help me fix my pip install for fastapi, redis, and uvicorn.

Write the FastAPI app and test the Redis connection

Now you'll create the FastAPI app that will become your job queue's API server. The first version connects to Redis on startup and exposes a health check endpoint to prove the connection works.

  • Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type the name of your code editor (e.g., Visual Studio Code or Cursor) and press Enter to open it.
  • Click File in the top menu bar.
  • Click Open Folder.
  • Select the jobqueue folder on your Desktop and click Open.
  • Create a new file called app.py in the jobqueue folder.
  • Set up the imports, constants, and app instance by copying this code into app.py:
import asyncio
import json
import time

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
import redis.asyncio as aioredis

# Stream and consumer group constants
STREAM_KEY = "jobs:stream"
DLQ_KEY = "jobs:dlq"
GROUP_NAME = "workers"
IDEMPOTENCY_PREFIX = "idempotency:"
IDEMPOTENCY_TTL = 86400

# Create the FastAPI app and a placeholder for the Redis connection
app = FastAPI(title="Job Queue")
redis_client: aioredis.Redis | None = None

What does this code do?

  • The imports pull in FastAPI for the web framework and redis.asyncio for an async Redis client. Some imports like json and BaseModel are for later steps.
  • The constants define stream names and keys that every part of the system will reference. Defining them once here avoids typos later.
  • redis_client starts as None and gets set when the server starts up. This pattern lets you share a single connection across all requests.
  • Add the lifecycle hooks and health endpoint below the existing code in app.py:

@app.on_event("startup")
async def startup():
    # Connect to Redis when the server starts
    global redis_client
    redis_client = aioredis.from_url("redis://localhost:6379", decode_responses=True)


@app.on_event("shutdown")
async def shutdown():
    # Close the Redis connection when the server stops
    if redis_client:
        await redis_client.aclose()


@app.get("/health")
async def health():
    # Ping Redis to confirm the connection is alive
    pong = await redis_client.ping()
    return {"status": "ok", "redis": pong}

What does this code do?

  • The startup hook runs automatically when uvicorn starts your app. It creates the async Redis connection with decode_responses=True so Redis returns strings instead of raw bytes.
  • The shutdown hook cleanly closes the Redis connection when the server stops, preventing leaked connections.
  • The /health endpoint calls redis_client.ping() and returns the result. If Redis is down, this endpoint will fail, making it a reliable liveness check.
  • Save app.py.
  • Switch back to your terminal (make sure you see (.venv) in the prompt) and start the server by running:
uvicorn app:app --reload

What does this command do?

uvicorn app:app tells uvicorn to look in the file app.py for the variable named app. The --reload flag automatically restarts the server whenever you change a file, so you do not have to stop and restart manually during development.

The server will take over this terminal window. You should see output indicating it is running on http://127.0.0.1:8000.

Before you open the health endpoint, what do you expect to see in the response if Redis is connected successfully?

  • Open your browser and go to http://localhost:8000/health.

You should see a JSON response:

{"status": "ok", "redis": true}

That confirms your FastAPI app is running and talking to Redis. The entire pipeline from browser to server to Redis and back is working.

Seeing a connection error?

  • If the page will not load, make sure uvicorn is still running in your terminal. You should see log output, not your command prompt.
  • If you see a Redis connection error in the terminal, confirm the Docker container is still running with docker ps. You should see jobqueue-redis in the list.
  • If you see a module import error, make sure you activated your virtual environment before running uvicorn. Check for (.venv) in your prompt.

Help me figure out why my /health endpoint is failing.

✔️ Awesome, I've got everything!

Great. Make sure you have saved app.py and keep uvicorn running in your terminal.

ⓧ I'd like to double check the full code

Scroll up through your app.py and confirm these sections are present in order:

  • Imports at the top: asyncio, json, time, then FastAPI imports, then redis.asyncio.
  • Five constants: STREAM_KEY, DLQ_KEY, GROUP_NAME, IDEMPOTENCY_PREFIX, IDEMPOTENCY_TTL.
  • app and redis_client declarations.
  • startup() and shutdown() lifecycle hooks.
  • The /health GET endpoint at the bottom.

Redis is running, your app is connected, and the health check proves the whole pipeline works. Next up, you'll build the API endpoint that actually submits jobs into a Redis Stream.

Build the Job API and Enqueue to a Redis Stream

Your health check endpoint proved the Redis connection works. Now you need a way to get jobs into the system.

In this step, you'll build the API layer that accepts job submissions and writes them into a Redis Stream using XADD. You'll also add a read endpoint so you can inspect what's queued up. Jobs will pile up with no worker to process them. That gap is intentional.

In this step, get ready to:
  • Define a job submission model and endpoint that writes jobs to a Redis Stream.
  • Add a listing endpoint to inspect queued jobs.
  • Submit test jobs and confirm they accumulate in the stream.
Define the job submission endpoint

Every job submission needs a consistent shape. A Pydantic model validates incoming requests and sets sensible defaults before anything hits Redis.

  • In app.py, find the shutdown function. Add the following model directly below it, above the @app.get("/health") line:
class JobRequest(BaseModel):
    job_type: str
    payload: dict = Field(default_factory=dict)
    idempotency_key: str | None = None
    max_retries: int = 3

What does this model do?

  • job_type is the only required field. It tells the worker what kind of work to perform.
  • payload defaults to an empty dict, so callers can attach whatever data the job needs.
  • idempotency_key and max_retries are optional. You'll use them in a later step to prevent duplicate submissions and control retry behavior.
  • Save app.py. The POST endpoint you build next validates every incoming job request against this model.

Now you need the endpoint that actually writes jobs into the stream. Each call takes a validated JobRequest, serializes its fields, and pushes them into the jobs:stream Redis Stream.

  • In app.py, add the following endpoint below the health function:
@app.post("/jobs", status_code=201)
async def submit_job(job: JobRequest):
    # Write the job into the Redis Stream
    message_id = await redis_client.xadd(
        STREAM_KEY,
        {
            "job_type": job.job_type,
            "payload": json.dumps(job.payload),
            "max_retries": str(job.max_retries),
            "attempts": "0",
            "created_at": str(time.time()),
        },
    )
    return {"job_id": message_id}

What does this endpoint do?

  • xadd appends a new entry to the jobs:stream Redis Stream. Each entry is a dictionary of string field-value pairs.
  • payload is JSON-serialized because Redis Stream fields only store strings.
  • attempts starts at "0". The worker will increment this on each retry so the system knows when to give up.
  • Redis returns a unique message ID (like 1692000000000-0) that the endpoint sends back to the caller.
  • Save app.py. If your uvicorn server was started with --reload, it picks up the changes automatically. If not, stop it with Ctrl+C and restart it.
  • Open a new terminal window so you can send requests while uvicorn keeps running.
  • Submit a test job by running this command:
curl -X POST http://localhost:8000/jobs \
  -H "Content-Type: application/json" \
  -d '{"job_type": "email", "payload": {"to": "test@example.com"}}'

You should see a JSON response with a job_id containing a Redis Stream message ID.

Getting a connection error or 404?

  • Make sure uvicorn is still running in your other terminal. If it crashed after your edit, check app.py for a syntax error (missing colon, mismatched parentheses).
  • Confirm the Redis container is running by running docker ps. You should see jobqueue-redis listed.
  • If you see a 422 error, double-check that your curl command includes the -H "Content-Type: application/json" header. Without it, FastAPI cannot parse the JSON body.

Help me debug my POST /jobs endpoint

Add a job listing endpoint and test the queue

Jobs are flowing into the stream, but you have no way to see them from the API. A quick listing endpoint lets you inspect what's queued without dropping into the Redis CLI.

  • In app.py, add the following endpoint below the submit_job function:
@app.get("/jobs")
async def list_jobs():
    # Read up to 50 entries from the stream
    messages = await redis_client.xrange(STREAM_KEY, count=50)
    return {"jobs": messages}

What does this endpoint do?

xrange reads entries from the stream in order, oldest to newest. The count=50 parameter caps the result so a massive backlog doesn't overwhelm the response.

  • Save app.py.
  • Submit a couple more test jobs to build up the queue by running these commands:
curl -X POST http://localhost:8000/jobs \
  -H "Content-Type: application/json" \
  -d '{"job_type": "resize", "payload": {"width": 800}}'
curl -X POST http://localhost:8000/jobs \
  -H "Content-Type: application/json" \
  -d '{"job_type": "notify", "payload": {"channel": "slack"}}'
  • Open http://localhost:8000/jobs in your browser.

You should see a JSON response listing all three jobs you submitted, each with its message ID and the fields you sent. Three jobs sitting in the stream, zero workers processing them.

Seeing an empty jobs list?

  • Make sure you saved app.py after adding the list_jobs endpoint. An unsaved file means uvicorn is still serving the old version.
  • Confirm your earlier curl POST commands each returned a job_id. If they returned errors, the jobs never made it into the stream.

Help me figure out why my GET /jobs endpoint returns no data

Before you move on, confirm the jobs are also visible at the Redis level. What count do you expect to see after three POST requests?

  • Verify the stream length by running this command:
docker exec jobqueue-redis redis-cli XLEN jobs:stream

You should see 3 printed. That confirms all three jobs landed in the stream and are waiting for a worker.

Why are the jobs just sitting there?

A Redis Stream is an append-only log. Without a consumer reading from it, entries accumulate indefinitely. You've built the producer side of the queue. In the next step, you'll build the consumer side: an asyncio worker pool that reads these jobs through a consumer group and processes them.

✔️ Awesome, I've got everything!

Great. Double check you've saved app.py and all three jobs show in both the API and the XLEN output.

ⓧ I'd like to double check the full code

Scroll through your app.py and confirm you have each of these sections, in order:

  • Imports (asyncio, json, time, FastAPI, HTTPException, BaseModel, Field, aioredis).
  • Constants (STREAM_KEY, DLQ_KEY, GROUP_NAME, IDEMPOTENCY_PREFIX, IDEMPOTENCY_TTL).
  • App creation and redis_client declaration.
  • startup and shutdown lifecycle functions.
  • JobRequest Pydantic model with job_type, payload, idempotency_key, and max_retries.
  • GET /health endpoint.
  • POST /jobs endpoint using xadd.
  • GET /jobs endpoint using xrange.

You now have a working producer feeding jobs into a Redis Stream. In the next step, you'll build the asyncio worker pool that consumes and processes these jobs.

Build the Asyncio Worker Pool with Consumer Groups

Your API is live and jobs are flowing into the Redis Stream. But nothing is on the other end to pick them up. Every job you submit just sits in the stream, waiting.

This step builds the worker side: an asyncio pool that reads jobs through a consumer group and processes them concurrently. By the end, you will have workers pulling jobs off the stream in real time.

In this step, get ready to:
  • Create a worker module with a Redis consumer group.
  • Build concurrent asyncio tasks that read and process stream messages.
  • Discover what happens when a worker crashes mid-job.
Create the worker module and initialize the consumer group

The worker is a separate Python process from your API server. It connects to the same Redis instance and reads from the same stream, but through a consumer group that coordinates which messages go to which worker.

  • Create a new file called worker.py in the same folder as app.py.
  • Add the imports, constants, and logging setup by pasting this code:
import asyncio
import json
import logging
import os
import time

import redis.asyncio as aioredis

# Stream and consumer group configuration
STREAM_KEY = "jobs:stream"
DLQ_KEY = "jobs:dlq"
GROUP_NAME = "workers"
CONSUMER_PREFIX = f"worker-{os.getpid()}"
CONCURRENCY = 3

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(message)s")
logger = logging.getLogger("worker")

What does this code do?

The constants mirror the same stream key and group name your API already uses. CONSUMER_PREFIX embeds the process ID so each worker process gets unique consumer names, even if you run multiple workers later.

CONCURRENCY = 3 means this single worker process will run 3 concurrent tasks, each independently pulling messages from the stream.

  • Save worker.py.
  • Confirm the file was created by running:
ls

You should see worker.py alongside app.py in the output.

  • Now add the main() function and entry point at the bottom of worker.py:
async def main() -> None:
    # Connect to Redis
    redis_client = aioredis.from_url("redis://localhost:6379", decode_responses=True)

    # Create the consumer group (skip if it already exists)
    try:
        await redis_client.xgroup_create(
            STREAM_KEY, GROUP_NAME, id="0", mkstream=True
        )
        logger.info(f"Created consumer group '{GROUP_NAME}'")
    except Exception:
        logger.info(f"Consumer group '{GROUP_NAME}' already exists")

    # Spawn concurrent consumer tasks
    tasks = []
    for i in range(CONCURRENCY):
        consumer_name = f"{CONSUMER_PREFIX}-{i}"
        tasks.append(
            asyncio.create_task(
                consumer_loop(redis_client, consumer_name)
            )
        )

    logger.info(f"Worker started with {CONCURRENCY} consumers (PID {os.getpid()})")
    await asyncio.gather(*tasks)
    await redis_client.aclose()

if __name__ == "__main__":
    asyncio.run(main())

What does this code do?

xgroup_create registers a consumer group called workers on the jobs:stream stream. The id="0" argument tells Redis to deliver all existing messages (not just new ones) to this group. The mkstream=True flag creates the stream automatically if it does not exist yet.

The for loop spawns 3 asyncio tasks, each running a consumer_loop coroutine with a unique consumer name. asyncio.gather keeps main alive until all tasks finish.

  • Save worker.py and try running it:
python worker.py

You will see a log line confirming the consumer group was created, followed by a NameError for consumer_loop. That is expected. You have built the skeleton and proven that xgroup_create works. The processing functions come next.

Seeing a connection error instead?

Make sure your Redis container is still running. Run docker ps and look for jobqueue-redis. If it is not listed, restart it with:

docker start jobqueue-redis

Still stuck? Help me debug my worker.py connection error.

Add the job processing loop

The NameError told you exactly what is missing. You need three functions: one to simulate job work, one to handle acknowledgment, and the consumer loop itself that reads from the stream.

  • In worker.py, find the line async def main() -> None: and add the following two functions above it:
async def process_job(job_data: dict) -> None:
    # Parse job data and simulate work
    job_type = job_data.get("job_type", "unknown")
    payload = json.loads(job_data.get("payload", "{}"))
    sleep_time = payload.get("duration", 1)
    logger.info(f"Processing job type={job_type}, sleeping {sleep_time}s")
    await asyncio.sleep(sleep_time)
    logger.info(f"Completed job type={job_type}")

async def handle_message(
    redis_client: aioredis.Redis,
    msg_id: str,
    msg_data: dict,
) -> None:
    # Try to process the job, then acknowledge or log failure
    try:
        await process_job(msg_data)
        await redis_client.xack(STREAM_KEY, GROUP_NAME, msg_id)
        logger.info(f"ACKed {msg_id}")
    except Exception as exc:
        logger.warning(f"Job {msg_id} failed: {exc}")

What does this code do?

process_job extracts the job type and payload, then simulates work by sleeping for the number of seconds specified in the payload's duration field. The default is 1 second.

handle_message wraps process_job in a try/except. If processing succeeds, it calls xack to tell Redis the message has been handled. Acknowledged messages are removed from the pending entry list. If processing fails, it logs the error but does not acknowledge, leaving the message pending.

  • Now add the consumer_loop function between handle_message and main():
async def consumer_loop(
    redis_client: aioredis.Redis,
    consumer_name: str,
) -> None:
    # Continuously read new messages from the stream
    logger.info(f"{consumer_name} started")
    while True:
        try:
            messages = await redis_client.xreadgroup(
                GROUP_NAME,
                consumer_name,
                streams={STREAM_KEY: ">"},
                count=1,
                block=2000,
            )
        except Exception as exc:
            logger.error(f"{consumer_name} read error: {exc}")
            await asyncio.sleep(1)
            continue

        if not messages:
            continue

        for stream_name, stream_messages in messages:
            for msg_id, msg_data in stream_messages:
                await handle_message(redis_client, msg_id, msg_data)

What does this code do?

xreadgroup reads from the stream as a member of the consumer group. The > ID tells Redis to deliver only messages that have never been delivered to this consumer before. count=1 fetches one message at a time. block=2000 makes the call wait up to 2 seconds for a new message before returning empty.

Each consumer task runs this loop independently. Because they all belong to the same group, Redis guarantees that each message is delivered to exactly one consumer. No two tasks ever process the same job.

  • Save worker.py and run it again:
python worker.py

This time you should see log lines showing the consumer group is ready and all 3 consumers have started. If you submitted test jobs in the previous step, the worker processes those immediately as it clears the backlog.

Seeing a NameError or import error?

Make sure process_job, handle_message, and consumer_loop are all defined above the main() function. Python reads from top to bottom, so functions must be defined before main() references them.

Still stuck? Help me fix my worker.py so it starts without errors.

✔️ Awesome, I've got everything!

Great. Make sure you have saved worker.py and the worker is running with all 3 consumers started.

ⓧ I'd like to double check the full code

Your worker.py should contain these sections in order. Scroll through your file and confirm each one is present:

  • Imports (asyncio, json, logging, os, time, redis.asyncio).
  • Constants: STREAM_KEY, DLQ_KEY, GROUP_NAME, CONSUMER_PREFIX, CONCURRENCY.
  • Logging setup with basicConfig and getLogger.
  • process_job function (parses payload, sleeps, logs).
  • handle_message function (calls process_job, then xack).
  • consumer_loop function (calls xreadgroup in a while True loop).
  • main function (creates group, spawns tasks, gathers).
  • if __name__ == "__main__": entry point.
Run the worker and discover the crash recovery gap

Your worker is running and processing jobs. Time to put it through a real test and see how it handles concurrent work. Then you will deliberately break it.

  • Make sure your API server is still running in its original terminal (uvicorn app:app --reload).
  • If your worker is not running from the previous verify, open a new terminal, navigate to the folder containing app.py, activate your virtual environment, and start it with python worker.py.
  • Open another terminal window for running commands.
  • Submit a test job by running:
curl -X POST http://localhost:8000/jobs \
  -H "Content-Type: application/json" \
  -d '{"job_type": "email", "payload": {"to": "user@example.com"}}'
  • Run this command 3 times to submit 3 jobs.

Before you check the worker terminal, what do you expect to see? Think about how many consumers are running and what the default job duration is.

Switch to the terminal running worker.py. You should see all 3 jobs picked up and completed within seconds. Each log line shows which consumer handled which message, followed by an ACK confirmation.

Why so fast?

The default duration is 1 second per job. With 3 concurrent consumers, all 3 jobs can process in parallel. The consumer group ensures each job goes to exactly one consumer, with no duplicate work.

Everything looks great. But what happens if a worker dies in the middle of a job? Time to find out.

  • Submit a long-running job by running:
curl -X POST http://localhost:8000/jobs \
  -H "Content-Type: application/json" \
  -d '{"job_type": "slow_task", "payload": {"duration": 30}}'

Switch to the worker terminal. You should see a log line that says the worker is processing slow_task and sleeping for 30 seconds. The worker is now mid-flight on that job.

What do you think will happen to that job if you kill the worker right now? Will Redis know the job was not finished?

  • Press Ctrl+C in the worker terminal to kill the process.

The worker exits. The 30-second job was never acknowledged because xack never ran. That message is now stuck in the pending entry list (PEL).

  • Verify the stuck job by running:
docker exec jobqueue-redis redis-cli XPENDING jobs:stream workers

You should see output showing at least 1 pending entry, the message ID range, and the consumer name from the crashed worker. It looks something like this:

1) (integer) 1
2) "1724000000000-0"
3) "1724000000000-0"
4) 1) 1) "worker-12345-0"
      2) "1"

What is the pending entry list?

Every message delivered to a consumer via xreadgroup goes into the pending entry list (PEL) until the consumer calls xack. This is how Redis tracks in-flight work. If a consumer crashes before acknowledging, the message stays in the PEL indefinitely.

The problem is clear: no other worker knows this message exists, and no mechanism exists to reclaim it. That job will sit there forever unless someone intervenes.

Your worker pool processes jobs concurrently and everything looks great on a sunny day. But you have just proven it has a critical flaw. When a worker crashes mid-job, the job is silently lost. Next up, you will fix this with graceful shutdown and automatic crash recovery.

Add Graceful Shutdown and Crash Recovery

In the previous step you proved the problem firsthand. When a worker crashes mid-job, the job sits in the pending entry list forever with no one to finish it.

This step fixes that gap with two mechanisms. Graceful shutdown catches termination signals so in-flight jobs finish cleanly before the process exits. A reaper loop using XAUTOCLAIM periodically scans for jobs that have been idle too long and reclaims them for a healthy worker.

In this step, get ready to:
  • Add signal handlers for SIGTERM and SIGINT so the worker shuts down gracefully.
  • Build a reaper coroutine that reclaims orphaned jobs using XAUTOCLAIM.
  • Test the full crash recovery cycle by killing a worker mid-job and watching a new worker reclaim it.
Add graceful shutdown with signal handling

Right now, pressing Ctrl+C or sending a termination signal kills the worker instantly. Any job mid-flight gets abandoned in the pending entry list. Signal handling lets you intercept these signals, stop accepting new work, and wait for in-flight jobs to finish before exiting.

Why not just let the process die?

Without graceful shutdown, every normal stop (a deploy, a container restart, Ctrl+C during development) creates orphaned jobs. Graceful shutdown means only unexpected crashes (power loss, kill -9) leave work behind. The reaper handles those.

  • In worker.py, find the import block at the top of the file. Add functools and signal to the imports so the block looks like this:
import asyncio
import functools
import json
import logging
import os
import signal
import time
  • Below the CONCURRENCY = 3 line, add three new constants that the reaper will use later:
REAPER_INTERVAL = 5
REAPER_MIN_IDLE_MS = 10000
MAX_BACKOFF = 30

What do these constants control?

REAPER_INTERVAL is how often (in seconds) the reaper scans for stuck jobs. REAPER_MIN_IDLE_MS is the idle threshold (10 seconds) before a pending job is considered abandoned. MAX_BACKOFF caps the retry delay for failed jobs (used in the next step).

  • Replace your entire consumer_loop function with the shutdown-aware version below:
async def consumer_loop(
    redis_client: aioredis.Redis,
    consumer_name: str,
    shutdown_event: asyncio.Event,
) -> None:
    logger.info(f"{consumer_name} started")
    # Keep reading until the shutdown signal fires
    while not shutdown_event.is_set():
        try:
            messages = await redis_client.xreadgroup(
                GROUP_NAME,
                consumer_name,
                streams={STREAM_KEY: ">"},
                count=1,
                block=2000,
            )
        except Exception as exc:
            logger.error(f"{consumer_name} read error: {exc}")
            await asyncio.sleep(1)
            continue

        if not messages:
            continue

        for stream_name, stream_messages in messages:
            for msg_id, msg_data in stream_messages:
                await handle_message(redis_client, msg_id, msg_data)

    logger.info(f"{consumer_name} shutting down")

What changed?

The function now accepts a shutdown_event parameter. The loop condition switched from while True to while not shutdown_event.is_set(). When something sets the event, each consumer finishes its current job and exits cleanly.

The block=2000 timeout in xreadgroup is key here. It means the loop checks the shutdown flag every 2 seconds even when no messages arrive.

  • Now replace the main() function. Find everything after the except block that logs Consumer group 'workers' already exists, and replace it with this signal handler setup and updated task creation:
    # Shared flag that all tasks watch for shutdown
    shutdown_event = asyncio.Event()
    loop = asyncio.get_running_loop()

    # Define a handler that sets the shutdown flag
    def signal_handler(sig_name: str) -> None:
        logger.info(f"Received {sig_name}, initiating graceful shutdown...")
        shutdown_event.set()

    # Register the handler for both SIGINT (Ctrl+C) and SIGTERM
    for sig in (signal.SIGINT, signal.SIGTERM):
        loop.add_signal_handler(
            sig,
            functools.partial(signal_handler, sig.name),
        )

    tasks = []
    for i in range(CONCURRENCY):
        consumer_name = f"{CONSUMER_PREFIX}-{i}"
        tasks.append(
            asyncio.create_task(
                consumer_loop(redis_client, consumer_name, shutdown_event)
            )
        )

    logger.info(f"Worker started with {CONCURRENCY} consumers (PID {os.getpid()})")
    await asyncio.gather(*tasks)

    await redis_client.aclose()
    logger.info("Worker shut down cleanly")

What does this code do?

shutdown_event is a shared flag. When a signal fires, every consumer loop sees it and stops reading new messages.

loop.add_signal_handler registers a callback for SIGINT (Ctrl+C) and SIGTERM (the signal sent by kill and container orchestrators). functools.partial passes the signal name to the handler so the log shows which signal arrived.

await asyncio.gather(*tasks) blocks until every consumer finishes its current job and exits. Only then does the Redis connection close.

This only works on macOS and Linux

loop.add_signal_handler is a Unix-only API. If you are on Windows, you will see a NotImplementedError. Use WSL (Windows Subsystem for Linux) to run the worker on Windows.

  • Save worker.py.
  • Make sure your FastAPI server is still running in another terminal. If not, start it again by running:
uvicorn app:app --reload
  • Start the worker in a separate terminal by running:
python worker.py

You should see the three consumers start up, just like before.

  • Submit a job with a 10-second duration by running this in a third terminal:
curl -X POST http://localhost:8000/jobs \
  -H "Content-Type: application/json" \
  -d '{"job_type": "email", "payload": {"duration": 10}}'

Before you press Ctrl+C, think about what you expect to happen. Will the worker die instantly, or will it wait for the 10-second job to finish first?

  • While the job is processing, press Ctrl+C in the worker terminal.

The worker logs the signal, finishes the in-flight job, ACKs it, then exits cleanly. You should see lines like:

Received SIGINT, initiating graceful shutdown...
Completed job type=email
ACKed 1234567890-0
worker-12345-0 shutting down
worker-12345-1 shutting down
worker-12345-2 shutting down
Worker shut down cleanly

Worker exits immediately without finishing the job?

Check that the while loop condition is while not shutdown_event.is_set() and that signal_handler only calls shutdown_event.set() without raising an exception. The key is that xreadgroup blocks for up to 2 seconds, then the loop re-checks the flag.

help me debug my worker's graceful shutdown

Build the reaper coroutine

Graceful shutdown handles normal stops. But a hard crash (kill -9, power loss, an OOM kill) leaves jobs orphaned with no clean exit. The reaper is a background coroutine that scans for these stuck jobs and pulls them back into the processing pipeline.

  • In worker.py, add the reaper_loop function right below the consumer_loop function:
async def reaper_loop(
    redis_client: aioredis.Redis,
    consumer_name: str,
    shutdown_event: asyncio.Event,
) -> None:
    logger.info("Reaper started")
    while not shutdown_event.is_set():
        try:
            await asyncio.sleep(REAPER_INTERVAL)
            # Claim any messages idle for more than 10 seconds
            result = await redis_client.xautoclaim(
                STREAM_KEY,
                GROUP_NAME,
                consumer_name,
                min_idle_time=REAPER_MIN_IDLE_MS,
                start_id="0-0",
            )
            if result and len(result) >= 2:
                claimed_messages = result[1]
                for msg_id, msg_data in claimed_messages:
                    if msg_data:
                        logger.info(f"Reaper reclaimed {msg_id}")
                        await handle_message(redis_client, msg_id, msg_data)
        except Exception as exc:
            logger.error(f"Reaper error: {exc}")

    logger.info("Reaper shutting down")

What does this code do?

Every REAPER_INTERVAL seconds (5 by default), the reaper calls xautoclaim to find messages that have been pending for more than REAPER_MIN_IDLE_MS milliseconds (10 seconds). Any message sitting idle that long probably belongs to a dead consumer.

xautoclaim atomically transfers ownership of those messages to the reaper's consumer name. Then it processes each reclaimed message through the same handle_message function that normal consumers use.

Why XAUTOCLAIM over XPENDING plus XCLAIM?

The older pattern requires two separate commands: XPENDING to find idle messages, then XCLAIM to take ownership. XAUTOCLAIM combines both into a single atomic operation, reducing round trips and race conditions.

  • Now wire the reaper into main(). Find the line that starts with logger.info(f"Worker started with and add the reaper task creation just above it:
    reaper_name = f"{CONSUMER_PREFIX}-reaper"
    tasks.append(
        asyncio.create_task(
            reaper_loop(redis_client, reaper_name, shutdown_event)
        )
    )
  • Update the log message on the same line to include the reaper. Change:
    logger.info(f"Worker started with {CONCURRENCY} consumers (PID {os.getpid()})")
  • To:
    logger.info(f"Worker started with {CONCURRENCY} consumers + reaper (PID {os.getpid()})")
  • Save worker.py and restart the worker by running:
python worker.py

You should now see four tasks starting: three consumers plus the reaper.

worker-12345-0 started
worker-12345-1 started
worker-12345-2 started
Reaper started
Worker started with 3 consumers + reaper (PID 12345)

Seeing a NameError for reaper_loop?

Make sure reaper_loop is defined before main() in the file. Python needs to see the function definition before it is referenced.

help me fix my reaper_loop setup

Test the full crash recovery cycle

The real test is a hard crash. Graceful shutdown will not help when a process is killed with kill -9. The reaper is the safety net for this scenario.

  • Start the worker if it is not already running:
python worker.py
  • Note the PID from the startup log. You will see it in a message like Worker started with 3 consumers + reaper (PID 12345).
  • Submit a job designed to take 30 seconds by running:
curl -X POST http://localhost:8000/jobs \
  -H "Content-Type: application/json" \
  -d '{"job_type": "slow", "payload": {"duration": 30}}'

You should see the worker pick it up and start processing.

  • While the job is mid-flight, simulate a hard crash by killing the worker with kill -9 in another terminal. Replace the PID with the one from your worker's startup log:
kill -9 [[WORKER_PID="the PID from your worker's startup log"]]

The worker dies instantly. The 30-second job is now stuck in the pending entry list.

  • Confirm the orphaned job by checking the pending entry list:
docker exec jobqueue-redis redis-cli XPENDING jobs:stream workers

You should see a non-zero pending count with at least one entry assigned to the dead worker's consumer name.

  • Now start a fresh worker to rescue the stuck job:
python worker.py

Before the reaper kicks in, think about what will happen. The new worker has a different PID, so its consumer names are different from the dead worker. Will the consumers pick up the orphaned job automatically, or does the reaper need to reclaim it first?

Wait about 10 seconds. You should see the reaper log that it reclaimed the stuck job and processed it.

Reaper reclaimed 1692345678901-0
Processing job type=slow, sleeping 30s
Completed job type=slow
ACKed 1692345678901-0
  • Verify zero pending entries by running:
docker exec jobqueue-redis redis-cli XPENDING jobs:stream workers

The first number in the output should be 0. Every orphaned job has been recovered.

Reaper not picking up the stuck job?

The reaper only claims messages idle for at least 10 seconds (REAPER_MIN_IDLE_MS = 10000). If you killed the old worker and started the new one very quickly, wait a few more seconds for the idle threshold to be reached.

Also check that the reaper scan interval (REAPER_INTERVAL = 5) means the reaper only runs every 5 seconds. Give it at least 15 seconds from the time you killed the old worker.

help me debug my reaper crash recovery

✔️ Awesome, I've got everything!

Great. Double check you have saved worker.py.

ⓧ I'd like to double check the full code

Your worker.py is longer than 30 lines, so instead of showing the full file here, confirm each section is present in order:

  • Imports: asyncio, functools, json, logging, os, signal, time, then redis.asyncio.
  • Constants: STREAM_KEY, DLQ_KEY, GROUP_NAME, CONSUMER_PREFIX, CONCURRENCY, REAPER_INTERVAL, REAPER_MIN_IDLE_MS, MAX_BACKOFF.
  • process_job function (unchanged from the previous step).
  • handle_message function (unchanged from the previous step).
  • consumer_loop with shutdown_event parameter and while not shutdown_event.is_set() loop.
  • reaper_loop with xautoclaim every 5 seconds.
  • main() with shutdown_event, signal handlers for SIGINT and SIGTERM, consumer tasks with shutdown_event, reaper task, and the Worker shut down cleanly log at the end.
  • if __name__ == "__main__" block calling asyncio.run(main()).

Your app.py has no changes in this step.

Your worker now handles both graceful stops and hard crashes. Next up, you will add the final reliability layer: idempotency keys to prevent duplicate submissions, retries with exponential backoff for transient failures, and a dead letter queue for permanently broken jobs.

Add Idempotency, Retries, and Dead Letter Queue

Your queue can now survive crashes. If a worker dies mid-job, the reaper reclaims the orphaned message and finishes it.

But three gaps remain. Nothing stops a client from submitting the same job twice. A single transient failure kills a job permanently. And when a job does fail for good, it vanishes with no record and no way to retry it later. This step closes all three gaps with idempotency keys, exponential backoff retries, and a dead letter queue with explicit replay.

In this step, get ready to:
  • Add idempotency key support to the job submission endpoint using SET NX to reject duplicate submissions.
  • Build inline retry logic with exponential backoff and route permanently failed jobs to a dead letter queue stream.
  • Add DLQ inspection and replay endpoints so failed jobs can be re-enqueued after the underlying problem is fixed.
Add idempotency key support

Imagine a network hiccup causes a client to retry a job submission. Without protection, the same job runs twice. Idempotency keys solve this by recording each key in Redis the first time it appears. If the same key shows up again, the server rejects it.

  • In app.py, find the submit_job function. It currently starts with the xadd call.
  • Replace the entire submit_job function with this updated version that checks the idempotency key before enqueuing:
@app.post("/jobs", status_code=201)
async def submit_job(job: JobRequest):
    # Reject duplicate submissions using SET NX
    if job.idempotency_key:
        key = f"{IDEMPOTENCY_PREFIX}{job.idempotency_key}"
        was_set = await redis_client.set(key, "1", nx=True, ex=IDEMPOTENCY_TTL)
        if not was_set:
            raise HTTPException(status_code=409, detail="Duplicate job submission")

    message_id = await redis_client.xadd(
        STREAM_KEY,
        {
            "job_type": job.job_type,
            "payload": json.dumps(job.payload),
            "max_retries": str(job.max_retries),
            "attempts": "0",
            "created_at": str(time.time()),
        },
    )
    return {"job_id": message_id}

What does this code do?

The idempotency check runs before the job is enqueued.

  • If the request includes an idempotency_key, the code builds a Redis key with the idempotency: prefix and calls set with nx=True. NX means "only set if Not eXists."
  • If the key already exists, was_set is None and the endpoint returns 409 Conflict.
  • The ex=IDEMPOTENCY_TTL sets a 24-hour expiry so keys don't accumulate forever.
  • Save app.py.
  • Stop the running FastAPI server with Ctrl+C, then restart it by running:
uvicorn app:app --reload
  • Submit a job with an idempotency key by running:
curl -X POST http://localhost:8000/jobs -H 'Content-Type: application/json' -d '{"job_type": "email", "payload": {}, "idempotency_key": "order-123"}'

You should see a 201 response with a job_id.

  • Now run the exact same command again:
curl -X POST http://localhost:8000/jobs -H 'Content-Type: application/json' -d '{"job_type": "email", "payload": {}, "idempotency_key": "order-123"}'

This time you should see {"detail":"Duplicate job submission"} with a 409 status. The same idempotency key can only enqueue a job once.

Still seeing 201 on the second request?

  • Make sure you saved app.py and restarted uvicorn.
  • Confirm the two curl commands use the exact same idempotency_key value.
  • Check that Redis is still running with docker exec jobqueue-redis redis-cli PING.

Why is my idempotency check not rejecting duplicates?

Add retry logic and dead letter queue routing to the worker

Right now, when a job throws an exception, the worker logs a warning and moves on. The job stays unacknowledged in the pending entry list, and it never retries or gets quarantined.

You need two things: a way to trigger failures on demand for testing, and a handle_message function that retries with exponential backoff or routes to a dead letter queue.

  • In worker.py, find the process_job function.
  • Find this line:
    payload = json.loads(job_data.get("payload", "{}"))
    sleep_time = payload.get("duration", 1)
  • Add a failure trigger between those two lines, so the function now looks like this:
async def process_job(job_data: dict) -> None:
    job_type = job_data.get("job_type", "unknown")
    payload = json.loads(job_data.get("payload", "{}"))

    # Deliberately fail jobs with type "fail" for testing
    if job_type == "fail":
        raise RuntimeError(f"Deliberate failure for job: {payload}")

    sleep_time = payload.get("duration", 1)
    logger.info(f"Processing job type={job_type}, sleeping {sleep_time}s")
    await asyncio.sleep(sleep_time)
    logger.info(f"Completed job type={job_type}")

What does this change do?

When a job arrives with job_type set to "fail", the function raises a RuntimeError immediately. This gives you a reliable way to test the retry and DLQ logic you are about to build.

Now for the core change. The current handle_message function only has a try/except that logs failures. You need it to track attempts, retry with backoff, and route exhausted jobs to the DLQ.

  • Find the existing handle_message function in worker.py. Replace the entire function with this first part, which covers the success path and the DLQ routing for exhausted retries:
async def handle_message(
    redis_client: aioredis.Redis,
    msg_id: str,
    msg_data: dict,
) -> None:
    attempts = int(msg_data.get("attempts", "0"))
    max_retries = int(msg_data.get("max_retries", "3"))

    try:
        await process_job(msg_data)
        await redis_client.xack(STREAM_KEY, GROUP_NAME, msg_id)
        logger.info(f"ACKed {msg_id}")
    except Exception as exc:
        attempts += 1
        logger.warning(f"Job {msg_id} failed (attempt {attempts}): {exc}")

        if attempts >= max_retries:
            # Move permanently failed job to dead letter queue
            await redis_client.xadd(
                DLQ_KEY,
                {
                    **msg_data,
                    "attempts": str(attempts),
                    "error": str(exc),
                    "failed_at": str(time.time()),
                },
            )
            await redis_client.xack(STREAM_KEY, GROUP_NAME, msg_id)
            logger.error(f"Job {msg_id} moved to DLQ after {attempts} attempts")

What does this code do?

  • The function reads attempts and max_retries from the message data so it knows how many times this job has already been tried.
  • On success, it acknowledges the message with xack as before.
  • On failure, it increments the attempt count. If retries are exhausted, it copies the job data into the jobs:dlq stream along with the error message and a timestamp, then acknowledges the original to remove it from the pending list.

The function is not complete yet. You still need the else branch that handles retries when the job has attempts remaining.

  • Add this else branch directly after the DLQ block you just pasted, at the same indentation level as the if attempts >= max_retries: line:
        else:
            # Retry with exponential backoff
            delay = min(2**attempts, MAX_BACKOFF)
            logger.info(f"Retrying {msg_id} in {delay}s (attempt {attempts})")
            await asyncio.sleep(delay)
            await redis_client.xadd(
                STREAM_KEY,
                {
                    **msg_data,
                    "attempts": str(attempts),
                },
            )
            await redis_client.xack(STREAM_KEY, GROUP_NAME, msg_id)

What does the retry branch do?

  • It computes the delay as min(2**attempts, MAX_BACKOFF). For attempt 1 the delay is 2 seconds, attempt 2 is 4 seconds, and it caps at 30 seconds.
  • After sleeping inline, it re-enqueues the job to the main stream with the updated attempt count.
  • It then acknowledges the original message. This is important because the re-enqueued copy is a new message with its own ID.
  • Save worker.py.
  • Stop any running worker with Ctrl+C, then start a fresh one by running:
python worker.py
  • In a separate terminal, submit a job designed to always fail by running:
curl -X POST http://localhost:8000/jobs -H 'Content-Type: application/json' -d '{"job_type": "fail", "payload": {"reason": "testing"}, "max_retries": 3}'

Watch the worker terminal. You should see the job fail on attempt 1, wait 2 seconds, fail on attempt 2, wait 4 seconds, fail on attempt 3, and then a log line saying the job moved to the DLQ.

  • Confirm the job landed in the dead letter queue by running:
docker exec jobqueue-redis redis-cli XLEN jobs:dlq

You should see (integer) 1.

Not seeing retry logs in the worker?

  • Make sure you replaced the entire handle_message function, not just added code inside the old one.
  • Verify the else branch is indented at the same level as if attempts >= max_retries:. Python's indentation determines which if the else belongs to.
  • Check that process_job includes the if job_type == "fail" block. Without it, the job succeeds silently.

Why is my worker not retrying failed jobs?

Add DLQ endpoints and test the full replay cycle

Jobs are landing in the dead letter queue, but you have no way to inspect them or move them back after fixing the problem. You need two new endpoints in app.py: one to list DLQ entries and one to replay a specific entry back to the main stream.

  • In app.py, add these two endpoints at the bottom of the file, after the existing list_jobs function:
@app.get("/dlq")
async def list_dlq():
    # Read entries from the dead letter queue stream
    messages = await redis_client.xrange(DLQ_KEY, count=50)
    return {"dead_letters": messages}


@app.post("/dlq/replay/{message_id}", status_code=201)
async def replay_dlq(message_id: str):
    # Find the specific DLQ entry by its message ID
    entries = await redis_client.xrange(
        DLQ_KEY, min=message_id, max=message_id, count=1
    )
    if not entries:
        raise HTTPException(status_code=404, detail="DLQ entry not found")

    entry_id, entry_data = entries[0]

What does this code do?

  • GET /dlq reads the jobs:dlq stream with xrange so you can see which jobs failed and why.
  • POST /dlq/replay/{message_id} looks up a single DLQ entry by passing the same ID as both min and max to xrange. This is the standard pattern for fetching one specific stream entry.
  • Still in the replay_dlq function, add the replay logic right after the entry_id, entry_data = entries[0] line:
    # Build a fresh job entry with attempts reset to 0
    replay_data = {
        "job_type": entry_data.get("job_type", "unknown"),
        "payload": entry_data.get("payload", "{}"),
        "max_retries": entry_data.get("max_retries", "3"),
        "attempts": "0",
        "created_at": str(time.time()),
    }
    new_id = await redis_client.xadd(STREAM_KEY, replay_data)
    # Remove the entry from the DLQ after successful re-enqueue
    await redis_client.xdel(DLQ_KEY, entry_id)
    return {"replayed_job_id": new_id, "original_dlq_id": entry_id}

What does the replay logic do?

  • It builds a fresh message from the DLQ entry's data with attempts reset to "0" so the job gets a full set of retries.
  • It uses xadd to push the job back into the main jobs:stream, then xdel to remove the DLQ entry so it cannot be replayed twice.
  • Save app.py.
  • Stop and restart uvicorn so it picks up the new endpoints.

Time to test the full end-to-end cycle. Before you run the next command, predict what will happen: will the idempotency check block this new job, or will it go through?

  • Submit the same idempotency key you used earlier by running:
curl -X POST http://localhost:8000/jobs -H 'Content-Type: application/json' -d '{"job_type": "email", "payload": {}, "idempotency_key": "order-123"}'

You should see 409 Conflict because the key is still active from your earlier test.

  • Now inspect the dead letter queue by running:
curl http://localhost:8000/dlq

You should see the failed job from the retry test, including its error and failed_at fields. Note the message ID from the response. It looks like 1234567890123-0.

  • Replay that DLQ entry back to the main stream by running this command with the message ID you noted:
curl -X POST http://localhost:8000/dlq/replay/[[DLQ_MESSAGE_ID="the message ID from the DLQ response above"]]

You should see a 201 response with a new replayed_job_id and the original DLQ ID.

  • Confirm the DLQ is now empty by running:
docker exec jobqueue-redis redis-cli XLEN jobs:dlq

You should see (integer) 0. The replayed job is back in the main stream for processing.

Seeing 404 on the replay endpoint?

  • Double-check the message ID you copied. It must match exactly, including the dash and sequence number (e.g. 1234567890123-0).
  • Make sure you are calling POST /dlq/replay/{message_id}, not GET.
  • If you already replayed it once, the entry is deleted from the DLQ. Submit another "fail" job to generate a new DLQ entry.

Why is my DLQ replay returning 404?

✔️ Awesome, I've got everything!

Great. Double check you have saved both app.py and worker.py.

ⓧ I'd like to double check the full code

Scroll through each section below and confirm your code matches. Both files are long, so compare section by section rather than all at once.

app.py should have these sections in order:

  • Imports (asyncio, json, time, FastAPI, HTTPException, BaseModel, Field, aioredis).
  • Constants: STREAM_KEY, DLQ_KEY, GROUP_NAME, IDEMPOTENCY_PREFIX, IDEMPOTENCY_TTL.
  • App setup, startup/shutdown events, JobRequest model.
  • GET /health endpoint.
  • POST /jobs with the idempotency check before xadd.
  • GET /jobs endpoint.
  • GET /dlq endpoint.
  • POST /dlq/replay/{message_id} endpoint.

worker.py should have these sections in order:

  • Imports and constants (STREAM_KEY, DLQ_KEY, GROUP_NAME, CONSUMER_PREFIX, CONCURRENCY, REAPER_INTERVAL, REAPER_MIN_IDLE_MS, MAX_BACKOFF).
  • process_job with the "fail" type trigger.
  • handle_message with attempts tracking, DLQ routing for exhausted retries, and else branch for exponential backoff retry.
  • consumer_loop (unchanged from the previous step).
  • reaper_loop (unchanged from the previous step).
  • main function with signal handlers, consumer tasks, and reaper (unchanged from the previous step).

Your job queue now has three layers of protection: idempotency keys block duplicate submissions, exponential backoff retries handle transient failures, and the dead letter queue quarantines permanently broken jobs with a replay path to recover them. Next up, you will take on an optional Secret Mission to add even more to your project.

Secret mission

Build a Queue Health Dashboard Endpoint

Right now, the only way to check if your queue is healthy is to manually run Redis CLI commands. Can you build a single API endpoint that tells you everything at a glance, including which workers might be dead?

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely locally, so there are no ongoing cloud costs. Decide whether to keep your resources running, pause them to come back later, or delete them entirely.

Resources you used:

  • Docker container jobqueue-redis running on port 6379 (data stored in Redis is ephemeral and lost when the container stops).
  • Project folder on your Desktop containing app.py, worker.py, and a Python virtual environment.
  • Running processes: FastAPI server on http://localhost:8000 and the asyncio worker pool.

Keep everything running

No action needed. Choose this if you are still actively building or experimenting with the job queue.

  • Your FastAPI server and worker process will keep running in their terminal windows.
  • The jobqueue-redis container will keep running until you stop it or restart Docker.
  • Your project folder and virtual environment stay on your Desktop.

Pause - I'll come back to this later

Shut down running processes to free up memory and ports, but keep your files so you can pick back up later.

  • In the terminal running the worker, press Ctrl+C to trigger the graceful shutdown.
  • In the terminal running the FastAPI server, press Ctrl+C to stop uvicorn.
  • Stop the Redis container by running:
docker stop jobqueue-redis

When you are ready to resume, start the container again by running:

docker start jobqueue-redis

Then restart the FastAPI server and worker from your project folder.

Delete - I don't want to use this again

Remove all project resources and start fresh. This cannot be undone.

  • In the terminal running the worker, press Ctrl+C to stop it.
  • In the terminal running the FastAPI server, press Ctrl+C to stop it.
  • Remove the Redis container by running:
docker rm -f jobqueue-redis
  • Deactivate the virtual environment (if it is still active) by running:
deactivate
  • Delete the project folder from your Desktop.

On macOS or Linux, navigate to your Desktop and remove the folder by running:

cd ~/Desktop
ls

Find your project folder name in the output, then run:

rm -rf [[PROJECT_FOLDER="your project folder name"]]

On Windows, open File Explorer, go to your Desktop, right-click the project folder, and select Delete.

Nice Work!

Nice Work!

Nice work! You built a production-grade background job queue from scratch using Redis Streams and asyncio, proving you can implement the exact worker pool and crash recovery patterns that power real distributed systems.

You've learned how to:

  • Built a FastAPI job submission API that enqueues work into a Redis Stream, with a concurrent asyncio worker pool using consumer groups to process jobs.
  • Killed a worker mid-flight, watched the job get stuck in the pending entry list, then fixed it with graceful shutdown signal handling and a reaper loop that uses XAUTOCLAIM to reclaim orphaned jobs automatically.
  • Added idempotency keys to prevent duplicate submissions, exponential backoff retries for transient failures, and a dead letter queue with explicit replay for permanently failed jobs.
  • Completed an optional Secret Mission to push your skills even further.

Ready to quiz yourself?