Streaming Large PDF Batches with Asyncio and Backpressure #

Within Async Batch Processing, the workload that quietly breaks the most pipelines is the multi-gigabyte PDF production: a single request that resolves to tens of thousands of scanned pages, contract packets, and email-export PDFs that all have to be pulled, rasterized or parsed, and handed downstream inside a statutory window. The instinct is to fan out — spawn a task per document and let the event loop sort it out. That instinct is exactly what fills memory, trips the upstream store’s rate limits, and produces a run that either dies with an out-of-memory kill or stalls without anyone noticing. This page shows how to stream that batch instead: a bounded producer, a semaphore-capped worker pool, and true backpressure so the pipeline runs at the speed the slowest dependency can sustain, holds resident memory flat, and proves its own progress line by line.

Scenario and Statutory Stakes #

A state agency receives a request for “every inspection report PDF filed against a facility since 2009.” The responsive set is 41,000 PDFs living in an object store, ranging from 40 KB single-page forms to 900 MB scanned binders. The response clock under 5 U.S.C. § 552(a)(6)(A)(i) allows 20 business days, and it is already running while the batch executes. The engineering problem looks like throughput; the compliance problem is that throughput bought the wrong way is indistinguishable from failure.

Loading every PDF into memory to parallelize it is the first failure. Even a modest fraction of 41,000 documents materialized at once exhausts the worker’s heap, the kernel OOM-kills the process, and whatever was in flight is abandoned. The agency now cannot prove it searched the whole set, which is a defective production, not a slow one. The second failure is unbounded network fan-out: firing 41,000 concurrent reads at the store saturates its connection quota, and under load the store’s own throttling looks identical to an outage the pipeline inflicted on itself — the condition NIST SP 800-53 SC-5 controls exist to prevent. The third and most insidious failure is silence: a stalled or partially crashed batch that produces no evidence of where it stopped. A defensible run has to bound memory, bound concurrency, and emit a per-page audit line so that “we processed every responsive document” is a claim backed by a ledger rather than a hope. The extracted text this pass produces then flows on through the rest of the Document Retrieval & Parsing pipeline.

Prerequisites #

• Python 3.11+ for asyncio.Queue, asyncio.Semaphore, asyncio.gather, and the logging module used for structured JSON audit output.
• An async I/O client — aiohttp (3.9+) or an async object-store SDK — so retrieval never blocks the event loop, plus a page-level PDF reader (PyMuPDF/fitz 1.24+) invoked in a worker thread for the CPU-bound parse.
• A bounded work source. The producer streams page units lazily; it must never build the full 41,000-document list in memory before starting.
• Append-only audit storage — a WORM bucket or a SIEM-forwarding log handler — so per-page lines satisfy NIST SP 800-53 AU-9 and cannot be rewritten after the fact. The shape of those lines should match what Audit Logging Architecture expects across the platform.
• A dead-letter queue for pages that fail after retries, so a bad scan is isolated rather than allowed to abort the surrounding run.

Why Backpressure, Not Just Concurrency #

Concurrency limits how many things run at once. Backpressure limits how fast work is admitted when the things already running cannot keep up. A semaphore alone gives you the first; a bounded asyncio.Queue gives you the second, and streaming a large batch safely needs both. The producer enumerates pages and calls await queue.put(unit). When the queue reaches its maxsize, that put suspends — the producer simply stops pulling more pages into memory until a worker drains a slot. Resident memory therefore plateaus at roughly maxsize units regardless of whether the batch holds 400 documents or 400,000. The workers pull from the queue under a shared semaphore, so the number of in-flight extractions never exceeds what the store’s quota and the host’s I/O bandwidth can sustain.

The dashed loop at the top is the whole point: it is not a control message the code sends, it is the natural consequence of a coroutine awaiting a full queue. You get flow control for free from the runtime, and the batch self-regulates to the throughput of the slowest stage.

Implementation #

Model each page as a small immutable unit — never the page bytes themselves, only the coordinates needed to fetch and account for it. The producer streams those units into a bounded queue; workers drain them under a semaphore and emit one audit line per page.

import asyncio
import datetime
import json
import logging
import uuid
from dataclasses import dataclass

# The FOIA response clock (5 U.S.C. 552(a)(6)(A)(i)) gives the agency 20
# business days. A batch that OOMs or stalls silently burns that window, so
# this pipeline bounds memory and proves progress rather than maximizing raw
# speed: correctness under load is the requirement, not peak pages-per-second.

AUDIT = logging.getLogger("foia_pdf_stream_audit")
AUDIT.setLevel(logging.INFO)


@dataclass(frozen=True)
class PageUnit:
    doc_id: str          # agency document identifier
    page: int            # 1-based page index within the document
    request_id: str      # parent FOIA request, for deadline accounting
    url: str             # location of the document in the object store


def _audit(event: str, unit: "PageUnit", trace_id: str, **fields) -> None:
    """Emit exactly one structured JSON audit line per terminal page outcome."""
    AUDIT.info(json.dumps({
        "ts": datetime.datetime.now(datetime.timezone.utc).isoformat(),
        "event": event,
        "trace_id": trace_id,
        "doc_id": unit.doc_id,
        "page": unit.page,
        "request_id": unit.request_id,
        **fields,
    }, sort_keys=True))


async def produce(queue: asyncio.Queue, units, worker_count: int) -> None:
    """Stream one work unit per page into a bounded queue.

    await queue.put(...) suspends when the queue is full: that is the
    backpressure that pauses enumeration until a worker frees a slot, so
    resident memory plateaus at roughly maxsize units no matter how large
    the responsive set is.
    """
    for unit in units:
        await queue.put(unit)          # blocks at maxsize -> producer awaits
    for _ in range(worker_count):
        await queue.put(None)          # one sentinel per worker for clean shutdown

The worker owns the failure policy. A single try/except around each page means one corrupt scan is dead-lettered, never allowed to escape and abort the surrounding tasks. The semaphore inside the worker caps how many extractions touch the store and the CPU at once.

async def extract_page(unit: "PageUnit") -> str:
    """Fetch and parse a single page off the event loop. The blocking PDF
    parse runs in a thread so it never stalls other coroutines."""
    def _parse() -> str:
        # Placeholder for a real page-scoped read, e.g. PyMuPDF opening the
        # document and returning get_text() for unit.page only.
        return f"text of {unit.doc_id} p{unit.page}"
    return await asyncio.to_thread(_parse)


async def worker(queue: asyncio.Queue, sem: asyncio.Semaphore,
                 dead_letter: asyncio.Queue, summary: dict) -> None:
    while True:
        unit = await queue.get()
        try:
            if unit is None:                       # sentinel: no more work
                return
            trace_id = str(uuid.uuid4())           # travels with the downstream handoff
            async with sem:                        # bounded concurrency: never overrun the store
                try:
                    text = await extract_page(unit)
                    _audit("EXTRACTED", unit, trace_id, chars=len(text))
                    summary["completed"] += 1
                    # ... hand text + trace_id to the OCR / metadata stage here ...
                except Exception as exc:           # per-item isolation, never abort the run
                    _audit("DEAD_LETTER", unit, trace_id, error=repr(exc))
                    await dead_letter.put((unit, repr(exc)))
                    summary["dead_letter"] += 1
        finally:
            queue.task_done()

The runner wires a bounded queue to a fixed pool of workers, starts the producer, and waits for the queue to drain before gathering the workers. maxsize and concurrency are the two knobs that decide memory footprint and load on the store respectively — tune them independently.

async def stream_batch(units, max_queue: int = 256, concurrency: int = 16) -> dict:
    queue: asyncio.Queue = asyncio.Queue(maxsize=max_queue)   # bounded -> backpressure + flat memory
    dead_letter: asyncio.Queue = asyncio.Queue()
    sem = asyncio.Semaphore(concurrency)
    summary = {"completed": 0, "dead_letter": 0}

    workers = [asyncio.create_task(worker(queue, sem, dead_letter, summary))
               for _ in range(concurrency)]
    await produce(queue, units, worker_count=concurrency)
    await queue.join()                    # block until every enqueued unit is task_done
    await asyncio.gather(*workers)        # collect worker results / exceptions
    return {"summary": summary, "dead_lettered": dead_letter.qsize()}

Expected behaviour: with max_queue=256, only ~256 page units are ever resident, so a 41,000-document batch runs in bounded memory. concurrency extractions are in flight at most, so the store never sees more than 16 simultaneous reads. Each page produces exactly one terminal audit line — EXTRACTED or DEAD_LETTER — and the two counters plus the dead-letter depth must reconcile against the batch size.

Expected Output and Verification #

Each page emits one append-only JSON line. A successful page looks like this:

{"chars": 1843, "doc_id": "INSP-2011-0442", "event": "EXTRACTED", "page": 7, "request_id": "REQ-77", "trace_id": "8f2c...d1", "ts": "2026-07-12T14:03:11.442110+00:00"}

Verify three invariants before trusting the run. First, memory stays flat: sample the worker RSS during the batch and confirm it plateaus rather than climbing with batch size — a rising trend means an unbounded structure is defeating the queue’s maxsize. Second, page accounting reconciles: completed + dead_lettered must equal the number of page units enumerated, with no third silent state. Third, backpressure actually engages: instrument queue.qsize() under load and confirm it sits at maxsize while the producer is faster than the workers — a queue that never fills means either the batch is trivially small or the bound is set too high to protect memory. A short assertion enforces the reconciliation in a test harness:

def assert_reconciled(result: dict, enumerated: int) -> None:
    s = result["summary"]
    accounted = s["completed"] + s["dead_letter"]
    if accounted != enumerated:            # never close a FOIA item with a gap
        raise AssertionError(f"unaccounted_pages:{enumerated - accounted}")

Common Pitfalls #

• Unbounded queue defeats the whole design. Creating asyncio.Queue() with no maxsize lets the producer race ahead and buffer the entire batch in memory — you have concurrency but no backpressure, and you OOM exactly as if you had never used a queue. Always set maxsize, and size it to the memory budget divided by the average unit footprint.
• Blocking parse stalls the event loop. Calling a synchronous PDF library directly inside a coroutine freezes every other worker until it returns, collapsing your concurrency to one. Push CPU-bound parsing into asyncio.to_thread (or a process pool for heavier work) as shown, so the loop stays responsive.
• gather without per-item isolation kills the batch. If a worker lets an exception propagate out of the page loop, asyncio.gather cancels its siblings and the run dies with thousands of pages unprocessed. Keep the try/except tight around each unit and route failures to the dead-letter queue rather than raising. When the failure is a whole-framework durability question rather than a single bad page, that decision belongs upstream — see Celery vs RQ vs Dramatiq for FOIA Batch Processing.
• Sentinels lost on early cancellation. If the producer is cancelled before it enqueues one sentinel per worker, some workers block forever on queue.get() and gather never returns. Enqueue the sentinels in a finally or guarantee the producer runs to completion, and prefer queue.join() to detect drain rather than guessing.

Compliance Verification Checklist #

• The work queue is created with an explicit maxsize sized to the memory budget, so await queue.put(...) provides backpressure and resident memory stays flat across the batch.
• In-flight extractions are capped by a shared asyncio.Semaphore sized to the smallest of upstream quota, connection-pool size, and host I/O bandwidth (NIST SP 800-53 SC-5).
• CPU-bound PDF parsing runs off the event loop via asyncio.to_thread or a process pool, so one document never stalls the pool.
• Each page is wrapped in its own try/except; a failed page is dead-lettered with full context, never allowed to abort the surrounding tasks.
• Every page produces exactly one terminal JSON audit line carrying the cross-service trace_id, forwarded to append-only storage (AU-9).
• completed + dead_lettered reconciles against the enumerated page count before the production is certified complete — there is no silent third state.
• Worker RSS is sampled during the run and confirmed to plateau, proving the bound holds rather than assuming it.
• One sentinel per worker guarantees clean shutdown, and queue.join() confirms the queue fully drained before workers are gathered.

FAQ #

Related #

• Async Batch Processing — the parent system this streaming pass plugs into
• Celery vs RQ vs Dramatiq for FOIA Batch Processing
• Optimizing Batch OCR Processing for Large Municipal Archives
• OCR Processing Pipelines
• Audit Logging Architecture

← Back to Document Retrieval & Parsing