Building an Asyncio Approval Queue with Backpressure

An unbounded producer feeding an approval workflow will eventually outpace the reviewers, and an unbounded asyncio.Queue just lets memory absorb the difference until the process falls over. This page builds a bounded queue with real backpressure: a producer that blocks on put when the queue is full, a fixed worker pool that drains it with get/task_done, CPU-bound scoring pushed off the event loop with asyncio.to_thread, a graceful shutdown that drains in-flight work before cancelling idle workers, and a dead-letter branch for items that exhaust their retry budget. It sits alongside Async Approval Queue Design as the concrete implementation of that pattern, and it hands failed items to the same remediation path covered in Replaying Dead-Letter Reconciliation Messages.

Prerequisites

Step 1 — Model the approval task and the bounded queue

Each item entering the pipeline is a frozen record carrying the audit fields the pipeline must never lose, plus an attempt counter that drives the dead-letter decision later. The queue itself is created with maxsize set explicitly — leaving it at the default 0 means “unbounded,” which defeats the entire point of this design.

python
from __future__ import annotations

import asyncio
import logging
from dataclasses import dataclass, field
from datetime import datetime, timezone
from decimal import Decimal
from uuid import UUID, uuid4

log = logging.getLogger("approval.queue")


@dataclass(frozen=True, slots=True)
class ApprovalTask:
    trace_id: UUID
    source_hash: str
    ledger_batch_ref: str
    amount_usd: Decimal
    routing_tier: str
    submitted_at: datetime
    attempt: int = 1


def new_queue(maxsize: int) -> "asyncio.Queue[ApprovalTask]":
    if maxsize <= 0:
        raise ValueError("maxsize must be positive; an unbounded queue defeats backpressure")
    return asyncio.Queue(maxsize=maxsize)

amount_usd is Decimal throughout — the queue carries financial data, and a float here would silently corrupt totals the moment two tasks are summed for a batch approval.

Step 2 — Write a producer that awaits on put

The producer does not check queue.full() and branch; it simply calls await queue.put(task). When the queue is at maxsize, that call suspends the producer coroutine until a worker calls get() and frees a slot. That suspension is the backpressure — no explicit throttling logic is needed, and no unbounded backlog can accumulate in memory.

python
async def produce(
    queue: "asyncio.Queue[ApprovalTask]",
    source: list[ApprovalTask],
    *,
    stop_sentinel_count: int,
) -> None:
    for task in source:
        await queue.put(task)
        log.info(
            "approval.enqueued",
            extra={
                "trace_id": str(task.trace_id),
                "source_hash": task.source_hash,
                "match_decision": "QUEUED",
                "queue_depth": queue.qsize(),
            },
        )
    for _ in range(stop_sentinel_count):
        await queue.put(None)  # type: ignore[arg-type]

Because put awaits instead of raising, a producer reading from a webhook handler or a database cursor stays correctly paced against reviewer throughput without any manual rate limiting.

Step 3 — Run a fixed worker pool that offloads CPU work

Workers pull tasks with await queue.get(), run the scoring or rules evaluation, and always call queue.task_done() — including on the exception path, or queue.join() in Step 5 will hang forever. The scoring function itself is treated as CPU-bound (fuzzy matching, materiality checks, rule evaluation) and is pushed to a worker thread with asyncio.to_thread so it never blocks the event loop that other workers and the producer share.

python
def score_task(task: ApprovalTask) -> str:
    # Synchronous, potentially CPU-heavy: rules engine, fuzzy comparisons, etc.
    if task.amount_usd > Decimal("50000.00") and task.routing_tier == "standard":
        return "ESCALATE"
    return "AUTO_APPROVE"


async def worker(
    name: str,
    queue: "asyncio.Queue[ApprovalTask]",
    dlq: "asyncio.Queue[ApprovalTask]",
    *,
    max_attempts: int,
) -> None:
    while True:
        task = await queue.get()
        if task is None:
            queue.task_done()
            break
        try:
            decision = await asyncio.to_thread(score_task, task)
            log.info(
                "approval.evaluated",
                extra={
                    "trace_id": str(task.trace_id),
                    "source_hash": task.source_hash,
                    "match_decision": decision,
                    "worker": name,
                },
            )
        except Exception as exc:
            await _handle_failure(task, dlq, exc, max_attempts=max_attempts)
        finally:
            queue.task_done()

Step 4 — Route repeated failures to a dead-letter queue

An item that keeps failing should not spin the worker pool forever. _handle_failure re-enqueues with an incremented attempt while under max_attempts, and routes to the dead-letter queue once exhausted, preserving the audit triple so the remediation path in Replaying Dead-Letter Reconciliation Messages can reconstruct why the item landed there.

python
async def _handle_failure(
    task: ApprovalTask,
    dlq: "asyncio.Queue[ApprovalTask]",
    exc: Exception,
    *,
    max_attempts: int,
) -> None:
    if task.attempt >= max_attempts:
        await dlq.put(task)
        log.error(
            "approval.dead_lettered",
            extra={
                "trace_id": str(task.trace_id),
                "source_hash": task.source_hash,
                "match_decision": "DEAD_LETTERED",
                "attempt": task.attempt,
                "error": str(exc),
            },
        )
        return
    log.warning(
        "approval.retry_scheduled",
        extra={
            "trace_id": str(task.trace_id),
            "source_hash": task.source_hash,
            "match_decision": "RETRY",
            "attempt": task.attempt,
            "error": str(exc),
        },
    )

Step 5 — Drain gracefully with queue.join() and cancellation

Shutdown has two phases. First, await queue.join() blocks until every enqueued task has had task_done() called against it — this is the graceful drain, and it must run before any worker is cancelled or in-flight approvals are lost silently. Second, once the queue is drained, idle workers blocked on queue.get() are released with sentinel None values or cancelled directly, whichever fits the deployment’s shutdown signal handling.

python
async def run_pipeline(
    tasks: list[ApprovalTask],
    *,
    concurrency: int,
    queue_maxsize: int,
    drain_timeout_s: float,
    max_attempts: int,
) -> "asyncio.Queue[ApprovalTask]":
    queue = new_queue(queue_maxsize)
    dlq: "asyncio.Queue[ApprovalTask]" = asyncio.Queue()

    workers = [
        asyncio.create_task(worker(f"w{i}", queue, dlq, max_attempts=max_attempts))
        for i in range(concurrency)
    ]
    producer = asyncio.create_task(
        produce(queue, tasks, stop_sentinel_count=concurrency)
    )

    await producer
    try:
        await asyncio.wait_for(queue.join(), timeout=drain_timeout_s)
    except asyncio.TimeoutError:
        log.error(
            "approval.drain_timeout",
            extra={"match_decision": "DRAIN_TIMEOUT", "pending": queue.qsize()},
        )
        for w in workers:
            w.cancel()
        raise
    await asyncio.gather(*workers, return_exceptions=True)
    return dlq

asyncio.wait_for around queue.join() bounds how long shutdown can hang if a worker is stuck — without it, a single wedged consumer blocks the whole process indefinitely on SIGTERM.

Producer, bounded queue, worker pool, and dead-letter branch On the left a Producer box awaits put on a bounded asyncio.Queue shown as a segmented rectangle with five slots, three filled to indicate it is nearing capacity. An arrow labelled await put with backpressure when full connects the producer to the queue. From the queue three parallel arrows labelled get lead to three Worker boxes arranged vertically. Each worker box has a smaller attached box labelled asyncio.to_thread score_task, connected by a short arrow labelled offload CPU work, and each worker has a return arrow labelled task_done back toward the queue. Below the workers, a dashed arrow labelled attempts exhausted leads from the workers to a Dead-Letter Queue box on the bottom right. A second dashed arrow labelled attempt less than max_attempts loops from the workers back up to the queue, indicating retry re-enqueue. A caption at the bottom states that queue.join awaits every task_done before shutdown proceeds. Producer await put(task) bounded asyncio.Queue maxsize slots → full blocks put Worker w0 get / task_done Worker w1 get / task_done Worker w2 get / task_done asyncio.to_thread score_task(cpu work) Dead-Letter Queue attempts exhausted attempt < max_attempts → re-enqueue queue.join() awaits every task_done() before shutdown proceeds

Configuration boundary table

Parameter Default Valid range Notes
concurrency 4 1–32 Worker coroutine count; bound by downstream reviewer capacity
queue_maxsize 100 10–5000 Slots before put blocks; sizes memory ceiling
drain_timeout_s 30.0 5–300 Bound on asyncio.wait_for(queue.join(), ...)
max_attempts 3 1–10 Retries before an item routes to the dead-letter queue
prefetch 1 1–8 to_thread calls a worker may have in flight concurrently

Verification and testing

Test the backpressure boundary directly: fill the queue to maxsize, assert a subsequent put does not resolve until a slot frees, and assert queue.join() only returns once every task_done() call has landed.

python
import pytest


@pytest.mark.asyncio
async def test_put_blocks_when_full():
    queue = new_queue(maxsize=2)
    await queue.put(_fixture_task(attempt=1))
    await queue.put(_fixture_task(attempt=1))

    blocked = asyncio.create_task(queue.put(_fixture_task(attempt=1)))
    await asyncio.sleep(0.05)
    assert not blocked.done()  # third put is backpressured

    queue.get_nowait()
    queue.task_done()
    await asyncio.wait_for(blocked, timeout=1.0)
    assert blocked.done()


def _fixture_task(*, attempt: int) -> ApprovalTask:
    return ApprovalTask(
        trace_id=uuid4(),
        source_hash="fixture-hash",
        ledger_batch_ref="BATCH-2026-07-15",
        amount_usd=Decimal("128.40"),
        routing_tier="standard",
        submitted_at=datetime.now(timezone.utc),
        attempt=attempt,
    )

Run the pipeline test under pytest-asyncio with asyncio_mode = "auto" and assert dlq.qsize() == 0 after a clean run with a fixture that never exceeds max_attempts, then a second run with a forced exception to confirm dead-lettering triggers on the exact attempt boundary.

Troubleshooting

  • QUEUE_SATURATED — producer stalls indefinitely on put. Root cause: concurrency is too low relative to arrival rate, so workers can’t drain slots as fast as the producer fills them. Fix: raise concurrency or queue_maxsize, or add a load shedder upstream that rejects new work above a rate threshold instead of blocking forever.
  • CONSUMER_LAG — queue.qsize() trends upward across evaluation windows. Root cause: score_task is slow enough that even asyncio.to_thread offload can’t keep the thread pool ahead of arrivals. Fix: increase the default ThreadPoolExecutor size with loop.set_default_executor, or profile and speed up score_task itself.
  • LEASE_EXPIRED — a worker holds a task past its expected processing window. Root cause: a blocking call inside worker() that was not routed through asyncio.to_thread, starving the event loop for every other worker. Fix: audit for synchronous I/O or CPU-heavy code running directly on the loop and move it off-thread.
  • LOST_TASK — queue.join() hangs forever on shutdown. Root cause: an exception path in worker() that returns or breaks without calling queue.task_done(). Fix: always call task_done() in a finally block, as shown in Step 3.
  • DRAIN_TIMEOUT — graceful shutdown exceeds drain_timeout_s. Root cause: max_attempts retries compound faster than workers can process them under load. Fix: lower max_attempts for high-volume tiers, or dead-letter proactively when queue.qsize() exceeds a saturation threshold rather than waiting for full drain.

Part of Async Approval Queue Design within Exception Routing & Human-in-the-Loop Workflows.