A real account of building an Async Job Queue with Backpressure & Priority Scheduling using Python, FastAPI, and Redis Streams — covering the Reaper service, 47 tests, 85% coverage, and every bug along the way.

No Celery. No shortcuts. No pretending the first approach worked.

I decided to build a production-grade project from scratch to bypass the abstraction layer of tools like Celery and truly master backend internals—the kind of work you can defend in any technical interview.

Instead of reaching for a framework that abstracts the hard parts, I built an async job queue from scratch — backpressure, priority scheduling, crash recovery, zombie detection, the full picture. This is the honest account of what I built, in what order, what broke, what I fixed, and what the final numbers looked like.

Stack: Python 3.12, FastAPI, Redis Streams, PostgreSQL, SQLAlchemy, Alembic, Prometheus, Grafana, Locust, Docker
Final state: 47 tests passing, 85% coverage, services layer at ~92%

Why a Job Queue? Why Not Just Use Celery?

Celery is the right tool when you need to ship business features fast and the queue is infrastructure, not the product. Building from scratch is the right choice when you need to understand exactly why a system fails — and how to keep it alive when it does.

When an interviewer asks "how does a visibility timeout work?" or "how do you prevent a zombie job from blocking your metrics?", knowing how to configure Celery doesn't answer that. Understanding the mechanism does.

A job queue is the backbone of anything that works outside the HTTP request-response cycle: sending emails, generating reports, processing images, running inference. The interesting engineering isn't the queue — it's everything around it:

• How do you stop a flood of jobs from overwhelming your workers?
• How do you ensure critical jobs run before low-priority ones without starving everything else?
• What happens when a worker crashes mid-job?
• How do you detect a job that's permanently broken vs. one that just needs a retry?

The Part Most Tutorials Skip: The Reaper

Before walking through the build, I want to call out the component that separates a job queue from a production-grade job queue: the Reaper.

Most tutorials build a producer and a consumer and call it done. But the hardest failure mode isn't a job that errors — it's a worker that disappears. A worker that picks up a job, starts executing, and then crashes. The message is claimed. No ACK is coming. The job is stuck in limbo.

The Reaper is the background service that fixes this. It runs every 10 seconds and does two things:

1. Visibility timeout recovery — queries XPENDING per Redis Stream for messages claimed but not acknowledged beyond the timeout threshold. Re-enqueues them as retries.
2. Zombie job detection — queries PostgreSQL for jobs in RUNNING status whose heartbeat_at timestamp is older than 60 seconds. Marks them FAILED and re-enqueues.

Without the Reaper, a crashed worker creates jobs that show RUNNING forever — poisoning your metrics, silently losing work, and giving you no signal that anything went wrong. With it, the system self-heals.

Every other component in this build is interesting. The Reaper is what makes the system trustworthy.

The Build Order (And Why It Matters)

I enforced a strict eight-stage sequence. The rule: if Stage N is broken, do not start Stage N+1. A broken Redis client makes every queue test confusing. A broken DB session makes every status transition unreliable.

Stage 1 — Environment & tooling
Stage 2 — Docker Compose infrastructure
Stage 3 — Configuration layer
Stage 4 — Database models & migrations
Stage 5 — Core clients (Redis, DB session)
Stage 6 — Business logic (services)
Stage 7 — FastAPI layer (routers, schemas)
Stage 8 — Background workers

Every stage had a verification step before moving on. Here's what actually happened at each one.

Stage 1 — Environment & Tooling

Poetry over pip. Poetry gives you a lockfile (exact versions of every dependency), separate dev and production groups, and a consistent virtual environment tied to the project. First thing I added was the dev dependencies — pytest, ruff, mypy — before any application code existed.

One non-obvious setting: asyncio_mode = "auto" and addopts = "--no-cov" in pyproject.toml. The --no-cov suppresses noisy coverage warnings before there's any application code to cover. Easy win.

.env discipline from the start. .env.example committed. .env in .gitignore immediately. If you accidentally commit a .env with real credentials, you must rotate everything in it. Don't find out the hard way.

Stage 2 — Docker Compose Infrastructure

No Kafka. Redis Streams provides the same PENDING/ACK semantics Kafka gives you, without the operational overhead of running Zookeeper alongside it.

The most important thing I did here: used condition: service_healthy in depends_on instead of just listing service names. depends_on without a health condition only waits for the container to start, not for the service inside it to be ready. Redis can take a second to accept connections. Without this, your app crashes on startup trying to connect to Redis before it's listening.

depends_on:
  redis:
    condition: service_healthy
  postgres:
    condition: service_healthy

Verification gate: make check-infra running redis-cli ping and pg_isready before touching any application code.

Stage 3 — Configuration Layer

One file: app/config.py. Its entire job is to read environment variables and expose them as a typed Python object. Nothing else in the codebase calls os.environ directly.

from functools import lru_cache
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    redis_url: str = "redis://localhost:6379/0"
    database_url: str = "postgresql+asyncpg://..."
    high_watermark: int = 10000
    low_watermark: int = 2000
    weight_critical: int = 60
    weight_high: int = 30
    weight_normal: int = 10
    max_retries: int = 5
    job_timeout_seconds: int = 30

@lru_cache
def get_settings() -> Settings:
    return Settings()

@lru_cache ensures settings are read exactly once per process. Without it, environment variables get read on every function call — wasteful and occasionally surprising.

First tests I wrote: verify defaults load, weights sum to 100, high watermark is greater than low watermark. Three tests. All pass. Move on.

Stage 4 — Database Models & Migrations

Use Alembic from day one. I learned this the hard way. SQLAlchemy's Base.metadata.create_all() is convenient for getting started, but it creates untracked schema you can't evolve safely in production. Alembic tracks every change as a versioned migration file.

The bug I hit: I ran alembic revision --autogenerate multiple times across sessions. Each run created a new migration file with no dependency on the previous one, breaking Alembic's chain silently. The fix was to delete everything in migrations/versions/ and generate one single clean migration.

The jobs table columns that matter most:

class Job(Base):
    __tablename__ = "jobs"

    id: Mapped[UUID] = mapped_column(primary_key=True, default=uuid4)
    status: Mapped[str]          # PENDING → RUNNING → COMPLETED/FAILED
    priority: Mapped[str]        # critical, high, normal
    payload: Mapped[dict]        # JSON
    result: Mapped[dict | None]  # JSON, nullable
    retry_count: Mapped[int] = mapped_column(default=0)
    max_retries: Mapped[int] = mapped_column(default=5)
    worker_id: Mapped[str | None]
    heartbeat_at: Mapped[datetime | None]  # ← key for zombie detection
    error: Mapped[str | None]
    created_at: Mapped[datetime] = mapped_column(default=func.now())
    updated_at: Mapped[datetime] = mapped_column(default=func.now(), onupdate=func.now())

heartbeat_at is the column that makes the entire zombie detection system work. Without it, you have no way to distinguish a job that's genuinely running from one whose worker died.

Stage 5 — Core Clients

redis[hiredis], not just redis. The hiredis extra is a C extension that makes Redis response parsing ~10x faster. One extra word in the install command, significant throughput difference at load.

The bug that cost me time: I wrote the health endpoint as:

async def health(request):  # WRONG

FastAPI saw request without a type annotation and treated it as a query parameter, not the HTTP request object. The fix:

async def health(request: Request):  # RIGHT

First-timer mistake. It'll happen to you too. Now you know.

The Redis client lifecycle: Create it inside the FastAPI lifespan function, not at module import time. Store it on app.state.redis. Never create it globally.

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.redis = await create_redis_client(settings.redis_url)
    yield
    await app.state.redis.aclose()

Creating it at import time causes "Event loop is closed" errors that are confusing to debug.

Stage 6 — Business Logic

This is the interesting part. All services are plain Python async functions — no FastAPI imports, no HTTP, no request/response objects. Pure logic that's easy to unit test.

Backpressure: Two Watermarks, Not One

My first instinct was a single threshold: accept below X, reject above X. This causes oscillation — rapid toggling between accepting and rejecting as queue depth bounces around the threshold under load.

The fix is a band:

BACKPRESSURE_STATE_KEY = "backpressure:active"

async def enqueue(redis, job_id, payload, priority):
    depth = await redis.xlen(f"queue:{priority}")
    is_active = await redis.exists(BACKPRESSURE_STATE_KEY)

    if depth >= settings.high_watermark:
        await redis.set(BACKPRESSURE_STATE_KEY, "1")
        raise BackpressureError()

    if is_active and depth >= settings.low_watermark:
        raise BackpressureError()

    if is_active and depth < settings.low_watermark:
        await redis.delete(BACKPRESSURE_STATE_KEY)

    await redis.xadd(f"queue:{priority}", {"job_id": job_id, "payload": payload})

High watermark fires the flag. The flag stays set until depth drops below the low watermark. The band gives the queue time to drain before opening again.

Weighted Fair Scheduling: No Starvation

Multiple priority queues are useless without a scheduler that prevents starvation. A naive scheduler draining critical first means low-priority jobs never run when critical is busy.

async def pick_queue(redis) -> str:
    weights = await get_weights(redis)
    # e.g. {"critical": 60, "high": 30, "normal": 10}

    roll = random.randint(1, 100)
    cumulative = 0
    for priority, weight in weights.items():
        cumulative += weight
        if roll <= cumulative:
            return f"queue:{priority}"

The weights are stored in Redis at runtime, not hardcoded. During an incident, you can set critical to 100% via an API call without redeploying.

I verified this with a test — 10,000 simulated picks, assert distribution matches configured weights within ±5%:

def test_scheduler_distribution():
    picks = {"critical": 0, "high": 0, "normal": 0}
    weights = {"critical": 60, "high": 30, "normal": 10}

    for _ in range(10_000):
        queue = pick_queue_sync(weights)
        picks[queue] += 1

    for priority, expected_pct in weights.items():
        actual_pct = picks[priority] / 100
        assert abs(actual_pct - expected_pct) < 5

If you can't prove the algorithm produces the right distribution, you haven't finished writing it.

Redis Streams: Not Just a List

My first instinct was LPUSH/RPOP — simple and familiar. Fatal flaw: if a worker pops a job and crashes before finishing, the job is gone. There's no recovery.

Redis Streams with consumer groups solve this via PENDING entries. A message delivered to a consumer but not yet acknowledged stays in the pending list. No ACK = not done.

# Dequeue
messages = await redis.xreadgroup(
    groupname="workers",
    consumername=worker_id,
    streams={stream_name: ">"},  # ">" = only undelivered messages
    count=1,
    block=2000
)

# After success
await redis.xack(stream_name, "workers", message_id)

The per-priority visibility timeouts I set:

Critical jobs get recovered faster. Normal jobs have more time before the reaper reclaims them.

Exponential Backoff with Full Jitter

def _backoff_seconds(retry_count: int, base: float = 1.0, max_delay: float = 60.0) -> float:
    return min(base * (2 ** retry_count), max_delay) * random.random()

The * random.random() is full jitter — prevents the thundering herd problem where all retrying workers wake up simultaneously and hammer the same resources. Without jitter, every worker that fails at the same time retries at exactly the same moment.

Bug I hit during implementation: The except Exception as e retry path referenced error_msg, a variable only defined inside the except asyncio.TimeoutError block above it. Python raised NameError at runtime. The fix was using str(e) directly in the general exception handler.

Stage 7 — FastAPI Layer

By the time I reached this stage, all business logic was written and tested. Route handlers are thin:

@router.post("/jobs", status_code=202)
async def submit_job(
    body: JobCreate,
    db: DBSession,
    redis: RedisDep,
    settings: SettingsDep,
) -> JobResponse:
    try:
        job = await create_job(db, body.payload, body.priority)
        await enqueue(redis, job.id, body.payload, body.priority, settings)
        return JobResponse.model_validate(job)
    except BackpressureError:
        backpressure_rejections_total.inc()
        raise HTTPException(
            status_code=503,
            headers={"Retry-After": "5"},
            detail="Queue at capacity. Retry after 5 seconds."
        )

If your route handler is longer than 20 lines, business logic has leaked into it.

The pytest event loop nightmare: My first conftest used asyncio_default_fixture_loop_scope = "session" but asyncio_default_test_loop_scope defaulted to "function". This caused "Future attached to a different loop" errors — SQLAlchemy's connection pool was created on the session loop but tests ran on function loops. The fix was setting both to "session" and restructuring conftest with session-scoped engine and Redis client.

Stage 8 — Background Workers

Two workers, each as a separate service in Docker Compose:

job_worker.py — the main worker loop:

1. Pick a queue via weighted scheduler
2. XREADGROUP to claim a message
3. Mark job RUNNING in PostgreSQL
4. Start a heartbeat task (writes timestamp every 5 seconds)
5. Dispatch to handler with asyncio.wait_for(timeout=30)
6. On success: XACK + mark COMPLETED
7. On failure: increment retry, re-enqueue with backoff, or DLQ if exhausted
8. Cancel heartbeat task in finally

reaper.py — runs every 10 seconds:

1. XPENDING query per stream for messages older than visibility timeout
2. Re-enqueue stale pending messages as retries
3. Query PostgreSQL for RUNNING jobs with heartbeat_at < now() - 60 seconds
4. Mark zombies as FAILED

# docker-compose.yml
job_worker:
  build: .
  command: python -m app.workers.job_worker
  restart: unless-stopped
  depends_on:
    app:
      condition: service_healthy

reaper:
  build: .
  command: python -m app.workers.reaper
  restart: unless-stopped

restart: unless-stopped means a crashed worker comes back up automatically and resumes — same behavior as production.

The Dockerfile Problem (And the Right Fix)

My multi-stage Dockerfile put the venv at /app/.venv. Then I added a bind mount in docker-compose:

volumes:
  - .:/app

The bind mount overlaid the entire /app directory, hiding the venv the Dockerfile built. The container had no packages.

My first fix was a named Docker volume to protect the venv. Workable, but fragile — every time I added a new package, I had to run docker compose down -v to wipe and repopulate the volume.

The correct fix: move the venv outside /app:

# Stage 1: build
ENV VIRTUAL_ENV=/opt/venv
RUN python -m venv $VIRTUAL_ENV
ENV PATH="$VIRTUAL_ENV/bin:$PATH"
RUN pip install ...

# Stage 2: runtime
COPY --from=builder /opt/venv /opt/venv
ENV PATH="/opt/venv/bin:$PATH"

/opt/venv is never touched by the bind mount. No volume management needed. The bind mount only covers /app.

The Engineering Pillars I Hardened After the Initial Build

After completing all 8 stages and doing a thorough review of the system against production requirements, I found gaps across three areas: resilience, observability, and testing. Here's how I addressed each one.

Pillar 1 — Resilience

Two-watermark backpressure band (not a single threshold). My initial implementation used one cutoff point. The problem: as queue depth oscillates around that number under real traffic, the system rapidly toggles between accepting and rejecting requests. The fix is a band — the high watermark fires a Redis flag, and the flag stays set until depth drops below the separate low watermark. That gap prevents oscillation.

Exponential backoff with full jitter. I had retry logic but used a fixed delay. The issue with fixed delays is synchronization — when ten workers all fail at the same moment, they all retry at the same moment, and you hammer your dependencies in waves.

def _backoff_seconds(retry_count: int, base: float = 1.0, max_delay: float = 60.0) -> float:
    return min(base * (2 ** retry_count), max_delay) * random.random()

The * random.random() is full jitter. It spreads retries across a window, breaking the thundering herd.

Bug hit during implementation: The except Exception as e retry path referenced error_msg, a variable only defined inside the except asyncio.TimeoutError block above it. Python raised NameError at runtime. Fixed by using str(e) directly.

Per-priority visibility timeouts. The original reaper used a single global timeout for all streams. Critical jobs need faster recovery. Normal jobs can wait longer before the reaper reclaims them.

Dead Letter Queue with replay. Permanently failed jobs (retry count exhausted) land in queue:dlq with their full failure metadata. A POST /queues/dlq/{id}/replay endpoint moves them back to their original priority stream for human-initiated retry. The DLQ depth is exposed in /queues/metrics so you know when it needs attention.

Runtime weight adjustment. Scheduling weights are stored in Redis, not config. PATCH /queues/weights validates that weights sum to 100 and stores them. DELETE /queues/weights reverts to config defaults. During an incident, you can redirect 100% of worker capacity to critical jobs via a single API call without redeploying.

Pillar 2 — Observability

Structured logging with structlog. Python's built-in logging produces unstructured text. structlog with JSON output in production produces log lines that are filterable and machine-parseable. Every log line includes request_id, job_id, priority, and the relevant domain context.

Bug hit: structlog.stdlib.add_logger_name is designed for logging.Logger objects. PrintLoggerFactory (structlog's default) produces PrintLogger objects with no .name attribute — the processor raises AttributeError at runtime. Fix: remove that processor from the chain.

Four custom Prometheus metrics — by name, not just by count:

# app/core/metrics.py
queue_depth = Gauge("queue_depth", "Current queue depth", ["priority"])
jobs_processed_total = Counter("jobs_processed_total", "Jobs processed", ["status"])
backpressure_rejections_total = Counter("backpressure_rejections_total", "Backpressure 503s fired")
zombie_jobs_reaped_total = Counter("zombie_jobs_reaped_total", "Zombie jobs recovered by reaper")

queue_depth updates on a 15-second background task. backpressure_rejections_total increments in the router's exception handler. jobs_processed_total increments in the worker's success and DLQ-failure paths. zombie_jobs_reaped_total increments in the reaper after each recovery.

Grafana dashboard provisioned automatically. Rather than requiring manual dashboard setup, I added grafana/provisioning/ with datasource and dashboard JSON files mounted into the Grafana container. make dev gives you a working dashboard at localhost:3000, no clicking required.

Pillar 3 — Testing

Locust load tests with three realistic user classes:

• LegitimateUser (weight 3): submits jobs, polls results, checks metrics
• BurstUser (weight 1): submits at 10-50ms intervals to trigger backpressure intentionally — both 202 and 503 are counted as success because 503 is the correct behavior under overload
• MixedWorkload (weight 2): submits across all three priorities in a 6:3:1 ratio

Results at 50 users, dev server: 56 req/s sustained, P50 220ms, P95 640ms.

Pre-commit hooks. ruff, ruff-format, mypy, and standard file checks (trailing-whitespace, check-yaml, debug-statements) run before every commit. Code that doesn't pass these never reaches the repo.

GitHub Actions CI pipeline. Every push triggers: ruff → mypy → alembic upgrade head → pytest with Redis 7 and Postgres 15 as sidecar containers. The migration step is included because schema drift between model and database is a silent failure mode that only shows up in integration tests.

The Tests That Actually Proved the System Works

I'm including three specific tests because they demonstrate how to test distributed system behavior without asyncio.sleep — which makes tests flaky and slow. More importantly, they test the failure modes that most implementations never cover: what happens when workers crash mid-execution.

The 85% overall coverage and ~92% services layer coverage isn't just a number — it specifically covers worker crashes during execution (test_visibility_timeout_reaper), stale-heartbeat zombie recovery (test_zombie_detection), and exhausted-retry DLQ routing. These are the edge cases that fail silently in production.

Visibility timeout reaper — directly patch the timeout to zero instead of waiting:

async def test_visibility_timeout_reaper(redis, mocker):
    await redis.xadd("queue:normal", {"job_id": "orphan-1", "payload": "{}"})
    # Claim it but never ACK (simulate crashed worker)
    await redis.xreadgroup("workers", "dead-worker", {"queue:normal": ">"}, count=1)

    # Patch timeout to 0ms — no sleeping required
    mocker.patch.dict("app.services.reaper_service.VISIBILITY_TIMEOUTS_MS",
                      {"critical": 0, "high": 0, "normal": 0})

    recovered = await recover_pending_messages(redis)
    assert recovered >= 1

Zombie detection — set stale heartbeat directly in the DB:

async def test_zombie_detection(db_session):
    job = await create_job(db_session, {"type": "send_email"}, "normal")
    await mark_running(db_session, job.id, worker_id="dead-worker")

    # Directly set heartbeat to 2 minutes ago
    stale_time = datetime.utcnow() - timedelta(seconds=120)
    await db_session.execute(
        update(Job).where(Job.id == job.id).values(heartbeat_at=stale_time)
    )
    await db_session.commit()

    reaped = await reap_zombie_jobs(db_session)
    assert reaped >= 1

    await db_session.refresh(job)
    assert job.status == "FAILED"
    assert "heartbeat timeout" in job.error

No sleeps. Direct state manipulation. Tests run in milliseconds.

Priority distribution — the test I showed interviewers:

async def test_priority_distribution(redis):
    picks = {"critical": 0, "high": 0, "normal": 0}
    for _ in range(10_000):
        queue = await pick_queue(redis)
        priority = queue.split(":")[1]
        picks[priority] += 1

    assert abs(picks["critical"] / 100 - 60) < 5
    assert abs(picks["high"] / 100 - 30) < 5
    assert abs(picks["normal"] / 100 - 10) < 5

Final Numbers

The 56 req/s and 640ms P95 are on a dev server (uvicorn single worker). The DESIGN.md section on 10x scale calls for Gunicorn with multiple workers as the first step — a fair and honest constraint to document.

Lessons for the Next Build

Knowing where your design breaks is more impressive than pretending it's perfect. These aren't regrets — they're the honest constraints of what a single-node dev server build looks like, and what comes next.

• Redis Cluster instead of single-node Redis — Redis is currently a single point of failure. At scale, the queue backend needs to survive node loss without losing PENDING state.
• Gunicorn with multiple Uvicorn workers — the dev server constraint (single-worker uvicorn) is what makes the 56 req/s the ceiling, not the floor. Multiple workers multiply that linearly.
• Sorted set for delayed re-enqueue — backoff delay currently uses asyncio.sleep inside the worker. That blocks the worker's event loop for the duration of the delay. The production pattern is a sorted set where the score is the execute_at timestamp, with a separate scheduler polling for ready entries. The worker enqueues and moves on immediately.
• Prometheus alerting rules — Grafana dashboards are reactive; you see a problem after it's already happening. Alerting rules on queue_depth, zombie_jobs_reaped_total, and DLQ depth make the system proactive.
• Per-job-type timeouts — there's one global job_timeout_seconds = 30. A send_email job and a generate_report job have very different expected runtimes. The next version reads timeout configuration per job type from Redis, same pattern as the scheduling weights.

The Honest Takeaways

Building this surfaced things that configuring a framework never would:

• depends_on doesn't mean ready — use condition: service_healthy or your app crashes on startup reaching a service that hasn't finished initializing
• Bind mounts hide Dockerfile artifacts — move your venv outside the mounted directory (/opt/venv instead of /app/.venv) or every docker compose up wipes it
• Pytest event loops don't mix — set both asyncio_default_fixture_loop_scope AND asyncio_default_test_loop_scope to "session" or you get "Future attached to a different loop" errors that are genuinely hard to diagnose
• Test your algorithms quantitatively — a 10,000-sample distribution test on the scheduler catches bias that type checks and unit assertions miss entirely
• The Reaper is what makes it production-grade — without visibility timeout recovery and zombie detection, crashed workers silently kill jobs and give you no signal that work was lost

The whole stack — FastAPI, Redis Streams, PostgreSQL, Prometheus, Grafana, Docker Compose — runs on a laptop. No cloud account required to build and demo it.

The GitHub repo, DESIGN.md, and CONTRIBUTING.md are https://github.com/macaulaypraise/async-job-queue-with-backpressure.git. If you have questions about the reaper logic, the backpressure band, or the worker integration tests — drop them in the comments.