Applying backpressure to OCR worker pools under load

When receipts arrive faster than OCR workers can drain them, an unbounded queue silently absorbs the difference until the process is OOM-killed mid-batch, leaving half-validated financial data in an ambiguous state. This guide composes three admission limits — a bounded queue, a concurrency semaphore, and a rate limiter — with a hysteresis controller that pushes a stop signal back to upstream producers, so a lagging worker pool degrades into a bounded wait instead of a memory blowup. It is the flow-control detail delegated by the parent Async Batch Processing guide, inside the broader Receipt Ingestion & OCR Data Extraction framework.

The design principle is that memory is a resource to be reserved, not consumed on demand: every unit of work must acquire a permit before it can occupy queue space, and the slowest stage — CPU-bound OCR — sets the pace for everything upstream of it.

Why Standard Approaches Fail

Under sustained load, the naive worker-pool patterns each fail by converting a throughput mismatch into a correctness or availability incident:

• The unbounded queue. asyncio.Queue() with no maxsize, or a plain list fed by producers, absorbs every arriving receipt regardless of how far behind the pool is. During a month-end surge the queue grows to hundreds of thousands of pending image references, resident memory climbs monotonically, and the OS OOM-killer terminates the worker mid-batch — the worst possible moment, because in-flight batches are lost and the audit trail fractures.
• Fire-and-forget task spawning. Calling asyncio.create_task (or loop.run_in_executor) once per arriving receipt with no ceiling spawns unbounded concurrency. Thousands of coroutines each pin an image buffer and an OCR model handle, so memory scales with arrival rate, not with pool width, and the CPU thrashes between far more tasks than there are cores.
• Dropping or timing out silently. Shedding load by discarding receipts when “busy” keeps memory bounded but destroys receipts that were legitimately submitted, so an expense never reaches Automated Policy Validation & Anomaly Flagging and surfaces months later as a reimbursement dispute. Backpressure must slow producers, not silently drop their work.

The fix is to make admission explicit: bound the queue, cap concurrency with a semaphore, cap the rate with a token bucket, and propagate a pause signal upstream when the pool falls behind so producers wait rather than pile on.

Architecture & Algorithm

Three limits compose into a single admission decision. The semaphore caps how many units are in flight at once (queued plus processing), so it bounds concurrency independently of queue depth. The token bucket caps the sustained admission rate, smoothing bursts that would otherwise fill the queue instantly. The bounded queue is the backstop: once it is full, queue.put() blocks the producer, which is the primitive form of backpressure. A worker releases its semaphore permit only after OCR completes, so the slowest stage sets the pace for admission.

from __future__ import annotations

import asyncio
import logging
import time
from dataclasses import dataclass
from typing import Awaitable, Callable, Iterable

logger = logging.getLogger("expense.ocr.backpressure")


@dataclass(frozen=True)
class BackpressureConfig:
    max_queue: int = 64        # hard memory backstop
    permits: int = 32          # max in-flight units (queued + processing)
    rate_per_sec: float = 50.0 # sustained admission rate
    burst: int = 50            # token-bucket capacity


class TokenBucket:
    """Async rate limiter. A producer awaits a token before enqueueing, so the
    sustained admission rate cannot exceed rate_per_sec even during a burst."""

    def __init__(self, rate_per_sec: float, burst: int) -> None:
        self._rate = rate_per_sec
        self._capacity = float(burst)
        self._tokens = float(burst)
        self._updated = time.monotonic()
        self._lock = asyncio.Lock()

    async def acquire(self) -> None:
        async with self._lock:
            while self._tokens < 1.0:
                now = time.monotonic()
                self._tokens = min(
                    self._capacity,
                    self._tokens + (now - self._updated) * self._rate,
                )
                self._updated = now
                if self._tokens < 1.0:
                    await asyncio.sleep((1.0 - self._tokens) / self._rate)
            self._tokens -= 1.0


async def produce(
    queue: asyncio.Queue,
    receipts: Iterable[dict],
    bucket: TokenBucket,
    sem: asyncio.Semaphore,
    concurrency: int,
) -> None:
    """Admit a receipt only when a token, a permit, and queue space are free.

    Each limit blocks independently: the token bucket paces the rate, the
    semaphore caps in-flight units, and queue.put() blocks when the bounded
    queue is full. A worker releases the permit after OCR, so admission tracks
    the pool's real drain rate rather than the producer's arrival rate."""
    for receipt in receipts:
        await bucket.acquire()      # rate limit
        await sem.acquire()         # bounded concurrency (released by a worker)
        await queue.put(receipt)    # blocks when full -> backpressure to producer
    for _ in range(concurrency):    # one poison pill per worker
        await queue.put(None)


async def worker(
    name: str,
    queue: asyncio.Queue,
    sem: asyncio.Semaphore,
    process: Callable[[dict], Awaitable[None]],
) -> None:
    """Drain the queue; release exactly one permit per real unit processed."""
    while True:
        item = await queue.get()
        try:
            if item is None:        # poison pill: no permit was taken for it
                return
            await process(item)
        except Exception:
            logger.exception("ocr_failed worker=%s", name)
        finally:
            if item is not None:
                sem.release()       # free a slot so the producer can admit more
            queue.task_done()


async def run_pool(
    receipts: Iterable[dict],
    process: Callable[[dict], Awaitable[None]],
    cfg: BackpressureConfig = BackpressureConfig(),
) -> None:
    """Wire producer, bounded queue, and worker pool with composed limits."""
    queue: asyncio.Queue = asyncio.Queue(maxsize=cfg.max_queue)
    sem = asyncio.Semaphore(cfg.permits)
    bucket = TokenBucket(cfg.rate_per_sec, cfg.burst)
    workers = [
        asyncio.create_task(worker(f"w{i}", queue, sem, process))
        for i in range(cfg.permits // 8 or 1)
    ]
    await produce(queue, receipts, bucket, sem, concurrency=len(workers))
    await asyncio.gather(*workers)

Peak resident memory is now (max_queue + pool_width) × avg_receipt_bytes and cannot exceed it: the queue is capped by maxsize, and the number of in-flight units is capped by the semaphore. No arrival rate can breach that ceiling.

Blocking put() is enough within one process, but a producer on the far side of an HTTP boundary needs an explicit signal so it stops sending rather than piling requests against a stalled socket. A hysteresis gate reads queue depth and flips a pause flag at a high watermark, clearing it only once depth falls back to a low watermark — the gap prevents the flapping that a single threshold would cause.

import asyncio
import logging

logger = logging.getLogger("expense.ocr.backpressure")


class WatermarkGate:
    """Hysteresis gate over queue depth. Signals PAUSE at the high watermark
    and RESUME only after depth drops to the low watermark, so upstream HTTP
    producers can return 429 while paused, pushing backpressure past the
    process boundary to the client instead of buffering in memory."""

    def __init__(self, queue: asyncio.Queue, hi: int, lo: int) -> None:
        if lo >= hi:
            raise ValueError("lo watermark must be below hi")
        self._q = queue
        self._hi = hi
        self._lo = lo
        self._paused = False

    def should_admit(self) -> bool:
        depth = self._q.qsize()
        if self._paused and depth <= self._lo:
            self._paused = False
            logger.info("backpressure_resume depth=%d", depth)
        elif not self._paused and depth >= self._hi:
            self._paused = True
            logger.warning("backpressure_pause depth=%d", depth)
        return not self._paused

An HTTP ingest handler calls should_admit() and returns 503/429 with a Retry-After header while paused, so a well-behaved client backs off exponentially. That converts an internal memory limit into a polite, standards-compliant flow-control signal that reaches all the way to the upload client.

Step-by-Step Integration

1. Set the queue bound from the memory budget. Choose max_queue so (max_queue + pool_width) × avg_receipt_bytes stays comfortably under the container limit. This bound, not the arrival rate, is what guarantees the process cannot be OOM-killed.

2. Size the semaphore to real in-flight work. Set permits to the number of receipts that may occupy the queue plus the pool at once. Too high and the queue does the bounding alone; too low and workers starve. Start at roughly max_queue / 2.

3. Pace bursts with the token bucket. Set rate_per_sec to the pool’s measured sustained OCR throughput and burst to a few seconds of that rate, so brief spikes are absorbed but a sustained flood is throttled to the drain rate.

4. Match watermarks with hysteresis. Put hi near 75% of max_queue and lo near 25%; a wide gap avoids pause/resume flapping. Verify the gate transitions correctly:

   import asyncio
   q: asyncio.Queue = asyncio.Queue(maxsize=64)
   gate = WatermarkGate(q, hi=48, lo=16)
   for _ in range(48):
       q.put_nowait(object())
   assert gate.should_admit() is False       # at hi watermark -> paused
   while q.qsize() > 16:
       q.get_nowait()
   assert gate.should_admit() is True        # dropped to lo -> resumed

5. Return a real HTTP backoff signal. In the ingest handler, gate admission on should_admit(); when paused, respond 429 with Retry-After instead of enqueueing, so clients slow down rather than the server buffering their overflow.

6. Alarm on the leading indicators. Emit queue depth, semaphore saturation (permits held), and per-batch OCR latency as metrics. A queue pinned at max_queue and permits fully held means the pool is the bottleneck — scale workers or shard, do not raise the bound.

Edge Cases & Gotchas

FAQ

What exactly is backpressure in an OCR pipeline?

Backpressure is the mechanism by which a slow consumer forces a fast producer to slow down instead of overflowing memory. In this pipeline it takes three forms that compose: a bounded queue.put() that blocks when full, a semaphore that refuses new admissions past a concurrency ceiling, and a watermark signal that tells an HTTP producer to pause. The common thread is that excess load is turned into waiting, never into unbounded buffering.

Why use a semaphore when the queue is already bounded?

They bound different things. The bounded queue caps how many items are waiting, while the semaphore caps how many are in flight including those currently being OCR’d in the pool. Without the semaphore, a producer could refill the queue the instant a worker dequeues an item, so total in-flight memory would be max_queue + pool_width with no coordination; the semaphore makes admission track the pool’s real drain rate.

Should backpressure drop receipts when the pool is overwhelmed?

No — dropping loses legitimately submitted expenses and creates reconciliation gaps that surface as disputes. Backpressure should slow admission, not shed it: block the producer, or return a 429 with Retry-After so the client retries later. Load shedding is only acceptable for truly idempotent, replayable sources, and even then a dead-letter path is safer than a silent drop.

How do I tune the high and low watermarks?

Set the high watermark where you want producers to pause — around 75% of the queue bound leaves headroom for in-flight admissions — and the low watermark low enough that resuming does not immediately re-trip the pause, typically around 25%. The gap between them is deliberate hysteresis; a narrow gap makes the controller flap, pausing and resuming on nearly every message.

Where does chunking fit relative to backpressure?

They are complementary. Chunking decides how a large batch is sliced into units and reassembled in order; backpressure decides how fast those units are admitted so the pool is never overwhelmed. Chunk first to bound the size of each unit, then apply backpressure so producers cannot outrun the Tesseract OCR workers draining them.

Related

• Async Batch Processing — the parent guide whose pipeline this flow-control layer protects.
• Chunking large receipt batches for OCR throughput — the sibling guide that sizes the units backpressure admits.
• Building async batch queues for high-volume receipt uploads — the broker-level queue and dead-letter design this pool consumes from.
• Tesseract OCR configuration — the CPU-bound step whose drain rate sets the pace for admission.