Building Async Batch Processors for Daily Policy Ingestion

This guide extends the Field Mapping Strategies cluster with the concurrency layer that wraps the mapping pipeline — the event-driven execution model that lets a single worker fleet absorb thousands of carrier policy PDFs every night without exhausting memory or losing audit lineage.

A synchronous extraction script that works fine against ten sample policies collapses at production volume in three specific ways, and each one is a distinct engineering failure rather than a vague “performance” issue.

The first is unbounded memory growth. Loading a multi-megabyte declarations PDF directly into a buffer, rendering every page eagerly, and holding the rendered objects until the whole document is mapped means peak resident memory scales with document size times concurrency. A 40 MB endorsement package fanned out across 32 workers will OOM-kill a container that comfortably handles single-document runs. Worse, pdfplumber page objects retain references to their parent document, so a coroutine that keeps a page in scope across an await boundary holds the entire PDF alive — a reference cycle the cyclic garbage collector cannot reclaim promptly.

The second is head-of-line blocking from extraction failures. One carrier ships a corrupt xref table or a scanned-only page with zero text density. In a naive for doc in batch: loop that single document raises, the batch aborts, and the night’s run dies at 03:00 with 6,000 documents unprocessed.

The third is state drift in the mapping layer. Retries, at-least-once message delivery, and overlapping nightly windows mean the same document can be processed twice. Without an idempotency key, the second pass commits a duplicate canonical record, and the audit trail can no longer answer “which run produced this premium value?”

The pattern below resolves all three: bounded concurrency with explicit backpressure, per-document failure isolation with deterministic routing, and an idempotency key derived from document content so reprocessing is a no-op.

This pattern sits downstream of the extraction engines and upstream of the canonical mapping contract, so the same pins that the Field Mapping Strategies layer requires apply here, plus the async I/O stack:

• Python 3.10+ — required for asyncio.TaskGroup semantics this guide mirrors, structural match, and X | Y union types.
• aiofiles>=23.2 — non-blocking file reads so disk I/O on large PDFs does not stall the event loop. See the asyncio task documentation.
• pdfplumber>=0.11 — the extraction primitive; CPU-bound page rendering must run in a thread executor, never directly on the loop.
• A durable message broker (Redis Streams, SQS, or RabbitMQ) that supports visibility timeouts and a dead-letter target. Documents arrive as work items, not as a static list.
• tracemalloc enabled in staging — production memory tuning depends on real allocation snapshots, documented in the tracemalloc module reference.

Confirm before running: extraction libraries are pinned in a lockfile, the broker has a configured dead-letter queue, and the mapping ruleset manifest version is available as an environment variable so the audit log can record exactly which version processed each batch.

pdfplumber rendering is CPU-bound and synchronous; running it directly blocks the event loop and serializes the whole pool. Wrap it in run_in_executor, and gate the number of in-flight documents with an asyncio.Semaphore so peak memory is concurrency × max_document_footprint, a number you can size against the container limit.

Copy
import asyncio
import logging
from concurrent.futures import ProcessPoolExecutor
from dataclasses import dataclass, field
from datetime import datetime, timezone
from hashlib import sha256

import aiofiles
import pdfplumber

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


@dataclass(frozen=True, slots=True)
class WorkItem:
    document_id: str
    storage_path: str
    carrier_code: str
    batch_id: str


@dataclass(slots=True)
class ExtractionResult:
    document_id: str
    idempotency_key: str
    pages_text: list[str] = field(default_factory=list)
    extracted_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))


def _render_pdf(raw: bytes) -> list[str]:
    """CPU-bound. Runs in a worker process, not on the event loop.
    Each page is read and immediately released so no page object
    outlives the loop iteration that produced it."""
    import io

    pages: list[str] = []
    with pdfplumber.open(io.BytesIO(raw)) as pdf:
        for page in pdf.pages:
            pages.append(page.extract_text() or "")
            page.flush_cache()  # drop cached rendering buffers eagerly
    return pages

The frozen=True, slots=True dataclasses keep work items immutable and compact — slots removes per-instance __dict__, which matters when millions of items pass through the pool nightly. page.flush_cache() is the single most effective memory fix: it releases the rendered character and rect buffers as soon as text is captured.

The idempotency key is a SHA-256 over the raw bytes plus the batch identifier. Identical content reprocessed in a later run produces the same key, so the downstream commit can be a conditional upsert that no-ops on a duplicate. Each document runs inside its own try/except so one failure never propagates to its siblings.

Copy
class TransientExtractionError(Exception):
    """Recoverable: broker hiccup, rate-limited OCR, transient I/O."""


class StructuralExtractionError(Exception):
    """Fatal for this document: corrupt xref, encrypted, zero-text scan."""


async def extract_one(
    item: WorkItem,
    pool: ProcessPoolExecutor,
    semaphore: asyncio.Semaphore,
) -> ExtractionResult:
    async with semaphore:  # backpressure: never exceed N in-flight documents
        try:
            async with aiofiles.open(item.storage_path, "rb") as fh:
                raw = await fh.read()
        except OSError as exc:
            raise TransientExtractionError(f"read failed: {item.document_id}") from exc

        key = sha256(raw + item.batch_id.encode()).hexdigest()
        loop = asyncio.get_running_loop()
        try:
            pages = await loop.run_in_executor(pool, _render_pdf, raw)
        except Exception as exc:  # pdfplumber raises broad PDF parse errors
            raise StructuralExtractionError(
                f"render failed: {item.document_id}"
            ) from exc

        if not any(p.strip() for p in pages):
            # zero-text density: a scanned-only document the native path cannot map
            raise StructuralExtractionError(f"zero-text-density: {item.document_id}")

        logger.info("extracted %s | key=%s | pages=%d", item.document_id, key, len(pages))
        return ExtractionResult(item.document_id, key, pages)

A zero-text-density document is raised as StructuralExtractionError rather than silently committed, because the correct destination for a scanned-only policy is the OCR Integration & Sync fallback queue, not the native-text path. Distinguishing transient from structural faults here is what makes the retry budget in the next step meaningful.

Drive the whole batch with one coroutine per document, collect outcomes with return_exceptions=True so a single failure cannot cancel the gather, then route each outcome: success commits, transient faults retry with exponential backoff and jitter, structural faults go straight to the dead-letter queue with a full diagnostic envelope.

Copy
import random


@dataclass(slots=True)
class DeadLetter:
    document_id: str
    reason: str
    failure_class: str
    batch_id: str
    occurred_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))


async def process_batch(
    items: list[WorkItem],
    *,
    max_concurrency: int = 16,
    max_retries: int = 3,
) -> tuple[list[ExtractionResult], list[DeadLetter]]:
    semaphore = asyncio.Semaphore(max_concurrency)
    results: list[ExtractionResult] = []
    dead: list[DeadLetter] = []

    with ProcessPoolExecutor(max_workers=max_concurrency) as pool:

        async def run(item: WorkItem) -> None:
            for attempt in range(1, max_retries + 1):
                try:
                    results.append(await extract_one(item, pool, semaphore))
                    return
                except TransientExtractionError as exc:
                    backoff = min(2 ** attempt, 30) + random.uniform(0, 1)
                    logger.warning("transient %s attempt=%d: %s", item.document_id, attempt, exc)
                    await asyncio.sleep(backoff)
                except StructuralExtractionError as exc:
                    dead.append(DeadLetter(item.document_id, str(exc), "structural", item.batch_id))
                    return
            dead.append(DeadLetter(item.document_id, "retry budget exhausted", "transient", item.batch_id))

        await asyncio.gather(*(run(item) for item in items))

    logger.info("batch done | ok=%d | dead=%d", len(results), len(dead))
    return results, dead

Each successful ExtractionResult carries its idempotency key into the canonical mapping pass, where the Field Mapping Strategies router normalizes values against the typed contract from Policy Schema Design before commit. The DeadLetter envelope is intentionally rich — analysts triage the structural failures, and the failure_class lets you alert separately on a spike of transient broker faults versus a carrier that suddenly started shipping unparseable PDFs.

Correctness for an async batch processor means three invariants hold under concurrency: memory stays bounded, no document is lost, and reprocessing is idempotent. Assert each one directly.

Copy
import pytest


@pytest.mark.asyncio
async def test_no_document_is_lost():
    items = [WorkItem(f"doc-{i}", f"/fixtures/policy-{i}.pdf", "ACME", "batch-2026-06-27") for i in range(50)]
    results, dead = await process_batch(items, max_concurrency=8)
    # every document reaches a terminal state — committed or dead-lettered, never silently dropped
    assert len(results) + len(dead) == len(items)


@pytest.mark.asyncio
async def test_idempotency_key_is_content_stable():
    item = WorkItem("doc-1", "/fixtures/policy-1.pdf", "ACME", "batch-A")
    pool = ProcessPoolExecutor(max_workers=1)
    sem = asyncio.Semaphore(1)
    first = await extract_one(item, pool, sem)
    second = await extract_one(item, pool, sem)
    pool.shutdown()
    assert first.idempotency_key == second.idempotency_key  # same bytes + batch ⇒ same key

For the memory invariant, wrap a representative batch in a tracemalloc snapshot and assert peak traced memory stays under max_concurrency × expected_document_footprint; a regression here usually means a page object escaped its loop iteration. Run the suite under pytest-asyncio with --forked so a worker-process leak in one test cannot mask a failure in the next.

Every terminal outcome — a committed ExtractionResult or a DeadLetter — must emit an append-only audit event recording the idempotency key, the batch identifier, the mapping manifest version in force, and a UTC timestamp, chained to the prior entry’s hash for the same document. This satisfies the lineage requirement that lets a compliance officer reconstruct, months later, exactly which nightly run and which ruleset version produced a given premium or coverage limit. Because the idempotency key is derived from document content, a reprocessed document produces an identical key, so the audit log proves the second pass was a no-op rather than a silent overwrite — the property regulators and internal SOX controls actually test for. Fields classified as PII or PHI in the dead-letter envelope inherit the field-level controls defined by Data Boundary Enforcement, so regulated values never leak into a plaintext triage dashboard.

Event loop stalls under load. Symptom: throughput collapses to single-document speed despite a 16-worker pool. Cause: pdfplumber rendering is running on the loop instead of in the executor. Fix: confirm every CPU-bound call goes through run_in_executor and that no synchronous file read bypasses aiofiles.

OOM kill mid-batch. Symptom: the container dies around peak concurrency with large documents. Cause: page objects held across await, or the semaphore set higher than container_limit ÷ max_document_footprint. Fix: keep page.flush_cache() in the render path and lower max_concurrency until peak RSS sits under seventy percent of the limit.

Duplicate canonical records. Symptom: the same policy appears twice after a retried batch. Cause: the downstream commit is an unconditional insert. Fix: make the commit a conditional upsert keyed on the idempotency key so a reprocessed document no-ops.

Dead-letter queue floods at a fixed hour. Symptom: a wave of structural dead-letters from one carrier every night. Cause: that carrier ships scanned-only PDFs the native path cannot read. Fix: route zero-text-density documents to the OCR Integration & Sync fallback before dead-lettering, and reserve the dead-letter path for genuinely unrecoverable input.

Retry storm on a transient outage. Symptom: synchronized retries hammer a recovering broker. Cause: fixed backoff with no jitter. Fix: keep the random.uniform(0, 1) jitter term and cap the retry budget so a sustained outage drains to dead-letter rather than looping forever.

• Field Mapping Strategies — the parent layer this concurrency model wraps and feeds
• Optimizing Camelot for Complex Insurance Tables — table reconstruction the batch hands off for schedule pages
• Extracting Coverage Limits from Scanned Policy PDFs — the native-text extraction this pipeline parallelizes
• Handling Rotated Pages in Policy Documents — a structural fault the dead-letter router escalates to OCR
• Implementing Priority Queues for Catastrophic Claims — backpressure and queue design for the downstream triage stage