Backpressure and Rate Limiting in Asyncio Ingestion Pipelines

An asyncio ingestion pipeline that reads logbook pages faster than its OCR and validation workers can drain them will happily buffer the whole backlog into memory and then fall over. This page shows how bounded queues, semaphores, and a token bucket keep a high-volume maintenance-log pipeline stable under load, applying the concurrency model set out in Async Batch Processing for High-Volume Logs. The goal is a pipeline whose memory footprint and downstream call rate stay flat whether you feed it ten pages or ten thousand.

Prerequisites & inputs

The pattern assumes a producer that yields work items — say, scanned page references for a fleet of A319s including D-AIMK — and a pool of coroutine workers that each perform an expensive step (an OCR call, then schema validation). Backpressure only matters when the producer can outrun the consumers, which in practice is always: a directory walk or a Kafka topic emits references at memory speed, while an OCR request for an ATA 21 air-conditioning work card takes hundreds of milliseconds.

You need three control primitives, each solving a distinct problem. A bounded asyncio.Queue caps how many unprocessed items exist at once, so await queue.put() blocks the producer once the buffer is full — that blocking is the backpressure. A semaphore caps how many workers hit a shared downstream at the same instant, protecting an OCR service that only licenses N concurrent sessions. A token bucket caps the sustained request rate, independent of concurrency, so you honor a provider’s requests-per-second ceiling even during a burst. Getting the interaction right between these three is the whole job; the scaling context for why it matters at fleet volume is laid out in Handling Async Log Ingestion at Scale.

Bounded-queue backpressure

Bounded-queue backpressureProducerBoundedqueueSemaphoregateWorkers

Step-by-step implementation

  1. Size the queue deliberately. A maxsize of a few multiples of the worker count is usually right — large enough to smooth jitter, small enough that a stalled downstream stops the producer within seconds rather than after gigabytes of buffering.
  2. Gate concurrency with a semaphore. Wrap the downstream call in async with semaphore so no more than the licensed number of OCR sessions run at once.
  3. Throttle the sustained rate with a token bucket. Refill tokens on a wall-clock schedule and await a token before each call; this decouples rate from concurrency.
  4. Signal completion with sentinels. Push one None sentinel per worker after the producer finishes so each worker exits cleanly instead of hanging on an empty queue.
from __future__ import annotations

import asyncio
import logging
import time
from dataclasses import dataclass, field

logging.basicConfig(level=logging.INFO, format="%(asctime)s | %(levelname)s | %(message)s")
logger = logging.getLogger("mro.backpressure")


@dataclass
class TokenBucket:
    """Rate limiter honoring a sustained requests-per-second ceiling.

    Keeps downstream OCR call rate within the electronic-recordkeeping
    processing envelope discussed in FAA AC 120-78B (electronic signatures,
    electronic recordkeeping systems).
    """

    rate_per_sec: float
    capacity: float
    _tokens: float = field(default=0.0)
    _updated: float = field(default_factory=time.monotonic)
    _lock: asyncio.Lock = field(default_factory=asyncio.Lock)

    async def acquire(self) -> None:
        async with self._lock:
            while True:
                now = time.monotonic()
                self._tokens = min(self.capacity, self._tokens + (now - self._updated) * self.rate_per_sec)
                self._updated = now
                if self._tokens >= 1.0:
                    self._tokens -= 1.0
                    return
                wait = (1.0 - self._tokens) / self.rate_per_sec
                await asyncio.sleep(wait)


async def ocr_and_validate(page_ref: str, semaphore: asyncio.Semaphore, bucket: TokenBucket) -> None:
    await bucket.acquire()
    async with semaphore:
        await asyncio.sleep(0.2)  # stand-in for the real OCR + validation round trip
        logger.info("processed page_ref=%s", page_ref)


async def worker(name: str, queue: "asyncio.Queue[str | None]",
                 semaphore: asyncio.Semaphore, bucket: TokenBucket) -> None:
    while True:
        item = await queue.get()
        try:
            if item is None:
                return
            await ocr_and_validate(item, semaphore, bucket)
        finally:
            queue.task_done()


async def run_pipeline(page_refs: list[str], worker_count: int = 4) -> None:
    queue: "asyncio.Queue[str | None]" = asyncio.Queue(maxsize=worker_count * 2)
    semaphore = asyncio.Semaphore(worker_count)
    bucket = TokenBucket(rate_per_sec=8.0, capacity=8.0)

    workers = [asyncio.create_task(worker(f"w{i}", queue, semaphore, bucket))
               for i in range(worker_count)]

    for ref in page_refs:
        await queue.put(ref)  # blocks here once the buffer is full: this is backpressure
    for _ in workers:
        await queue.put(None)

    await queue.join()
    await asyncio.gather(*workers)
    logger.info("pipeline drained cleanly, pages=%d", len(page_refs))


if __name__ == "__main__":
    refs = [f"D-AIMK/ATA21/page-{n:04d}" for n in range(40)]
    asyncio.run(run_pipeline(refs))

Worked example

Run the pipeline over 40 air-conditioning page references for D-AIMK with four workers, a queue of maxsize=8, a semaphore of 4, and a bucket refilling at eight tokens per second. When the producer races ahead, the queue fills to eight items and await queue.put() simply parks the producer coroutine until a worker calls task_done() — memory never grows past those eight in-flight items plus the four each worker holds. The semaphore guarantees no more than four OCR calls execute simultaneously, and the token bucket caps the sustained throughput at eight pages per second even though four workers could momentarily go faster.

Watch the log timestamps: the first burst of four completes quickly, then subsequent pages pace out to roughly 125 ms apart as the bucket meters them. The final pipeline drained cleanly, pages=40 prints only after queue.join() confirms every item was acknowledged and each worker consumed its sentinel. Nothing is dropped, and a downstream stall would surface as a stalled producer — a visible, debuggable condition rather than an out-of-memory kill.

Edge cases & compliance gotchas

  • Sentinel accounting. Push exactly one sentinel per worker. One too few leaves a worker blocked forever on queue.get(); one too many is harmless only because the extra None is consumed after the loop already exited on the first.
  • Exception leaks in task_done. If a worker raises before calling task_done(), queue.join() hangs. The try/finally guarantees the counter is decremented even when OCR fails — the failed page still routes onward to schema handling, described in Schema Validation & Error Handling.
  • Rate vs. concurrency confusion. A semaphore alone does not bound sustained rate: four fast workers can still exceed a provider’s per-second quota between bursts. You need the token bucket for the rate ceiling and the semaphore for the concurrency ceiling — they are not substitutes.
  • Clock choice. The bucket uses time.monotonic(), not wall-clock time, so an NTP step or timezone change cannot make the refill math jump backward mid-run.

Backpressure is what turns a fragile fan-out into a pipeline you can leave running unattended overnight. Bound the queue so the producer feels downstream pressure, gate concurrency with a semaphore, meter sustained rate with a token bucket, and drain deterministically with sentinels — and the same code that processes forty pages processes forty thousand without a memory-shape change.

Related pages