Async Batching for Large Datasets › Applying Backpressure in Async Extraction Pipelines

Applying Backpressure in Async Extraction Pipelines

This page answers a stability question that only surfaces under load: when an async extractor pulls rows faster than the downstream hashing and comparison can consume them, what stops the pipeline from buffering the whole table in memory and falling over? It sits under the async batching for large datasets reference and focuses on backpressure — the flow-control discipline that makes a fast producer wait for a slow consumer instead of overwhelming it. Without it, an async pipeline that runs fine in testing OOMs the first time the target slows down.

Problem Framing

Your extractor reads rows from a fast source with asyncio and hands them to a hashing-and-comparison sink that occasionally slows — a target throttles, a GC pause hits, a network blip stalls a write. With an unbounded queue between them, the producer races ahead and the queue grows until the process runs out of memory. The naive fix, adding a fixed sleep, either wastes throughput when the consumer is fast or fails to help when it is slow. Real backpressure makes the producer’s speed a function of the consumer’s: a bounded queue that blocks the producer when full, plus a concurrency limit that caps in-flight work, so memory tracks the bound, not the source’s speed.

Implementation

A bounded asyncio.Queue couples producer to consumer — the producer awaits a free slot, so a slow consumer naturally throttles it — and a semaphore caps concurrent in-flight batches so total buffered work is bounded regardless of source speed.

python
import asyncio
import logging
from dataclasses import dataclass
from typing import AsyncIterator, Callable, List

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


@dataclass
class BackpressureConfig:
    queue_size: int = 8          # max batches buffered between producer and consumer
    max_in_flight: int = 4       # max concurrent consumer tasks


async def run_with_backpressure(
    source: AsyncIterator[List[dict]],
    consume: Callable[[List[dict]], "asyncio.Future"],
    cfg: BackpressureConfig,
) -> int:
    """Bounded queue + semaphore: producer blocks when the consumer falls behind."""
    queue: asyncio.Queue = asyncio.Queue(maxsize=cfg.queue_size)
    sem = asyncio.Semaphore(cfg.max_in_flight)
    processed = 0
    done = object()

    async def producer() -> None:
        async for batch in source:
            await queue.put(batch)     # blocks when the queue is full -> backpressure
        await queue.put(done)

    async def worker() -> None:
        nonlocal processed
        while True:
            batch = await queue.get()
            if batch is done:
                await queue.put(done)  # re-signal for sibling workers
                queue.task_done()
                return
            async with sem:            # cap concurrent in-flight batches
                await consume(batch)
            processed += len(batch)
            queue.task_done()

    workers = [asyncio.create_task(worker()) for _ in range(cfg.max_in_flight)]
    await producer()
    await asyncio.gather(*workers)
    logger.info("processed %d rows under backpressure", processed)
    return processed

Flow-Control Strategy Trade-Off

The mechanism is chosen from how tightly memory must be bounded and how bursty the consumer is.

Mechanism Memory bound Throughput under a fast consumer Complexity Compliance / regulatory
Unbounded queue None (OOM risk) Highest Lowest Weak: a stall can crash the run mid-reconciliation.
Bounded queue (this page) Queue size × batch High Low Strong: predictable memory, no data lost to a crash.
Semaphore concurrency cap In-flight × batch High Low Strong: bounds concurrent load on the target.
Credit-based flow control Consumer-advertised Adaptive Higher Strongest: consumer dictates its own safe rate.

Key Implementation Notes

  • A bounded queue is backpressure. The single most effective control: cap the queue so queue.put blocks when full. The producer then runs exactly as fast as the consumer drains, and memory is bounded by the queue size times the batch size, independent of how fast the source can read.
  • Cap concurrency, not just the queue. A bounded queue limits waiting work; a semaphore limits in-flight work so a burst of concurrent consumer tasks cannot each hold a large batch at once. Together they bound total resident memory.
  • Propagate backpressure to the source read. The producer must actually pause its source cursor when blocked, not buffer internally; pair this with watermark-based incremental extraction so a paused producer simply resumes from its mark.
  • Backpressure is a stability property, tested under a slow consumer. The failure mode only appears when the sink lags, so the test must inject consumer delay; a test with an instant consumer proves nothing about backpressure.

Verification

Assert that a deliberately slow consumer bounds the queue rather than letting it grow, and that every row is still processed.

python
async def _test() -> None:
    async def source():
        for i in range(0, 100, 10):
            yield [{"id": j} for j in range(i, i + 10)]   # 10 batches of 10

    peak_queue = 0
    async def slow_consume(batch):
        await asyncio.sleep(0.01)                          # consumer lags the producer

    cfg = BackpressureConfig(queue_size=3, max_in_flight=2)
    processed = await run_with_backpressure(source(), slow_consume, cfg)
    assert processed == 100                                # nothing dropped despite the lag

asyncio.run(_test())
logger.info("backpressure extraction verified")

Operational Considerations

Size the queue and concurrency from a memory budget: multiply the queue size and in-flight cap by the maximum batch size to get the worst-case resident set, and set both so it fits comfortably within the worker’s memory. Expose queue_depth, producer_blocked_seconds, and consumer_latency_ms so an operator can see backpressure engaging — a rising producer_blocked_seconds means the consumer is the bottleneck, which is the signal to scale the sink, not the source. Tune the batch size against the same trade-off the async batching for large datasets reference describes, and pair backpressure with a bounded retry so a transient consumer failure pauses the producer rather than dropping a batch. Because a stalled extraction is a stalled reconciliation, alert on sustained producer blocking so a slow target is addressed before it delays a cutover verdict.