Async Batch Processing for High-Volume Receipt Auditing

Async batch processing decouples receipt ingestion from OCR, parsing, and policy validation so that finance teams can audit tens of thousands of receipts per hour under strict memory, latency, and compliance bounds.

Within the broader Receipt Ingestion & OCR Data Extraction framework, this component owns the orchestration layer: how receipts are chunked, queued, and advanced through deterministic pipeline stages without blocking user workflows or exhausting worker memory. It delegates the transport-level queue topology to Building async batch queues for high-volume receipt uploads, the character-recognition step to Tesseract OCR configuration, and tabular extraction to pdfplumber line-item parsing. What follows is the batch engine that ties those stages together and hands validated records to the downstream policy engines.

Problem Framing & Root Causes

Synchronous ingestion collapses at scale for three named reasons. Buffer materialization loads an entire upload batch into RAM before processing, so a month-end surge of 40,000 receipts triggers out-of-memory (OOM) kills on fixed-size workers. Head-of-line blocking serializes CPU-bound OCR behind network-bound uploads, wasting cores and missing ERP sync windows. Non-deterministic error handling lets a single malformed receipt abort a whole batch, fracturing the audit trail and leaving partially-validated financial data in an ambiguous state.

The fix is to treat the pipeline as a directed acyclic graph (DAG) of stages joined by a bounded queue: producers stream fixed-size chunks, consumers apply backpressure, and each stage emits structured telemetry before it triggers the next. This guarantees the strict data lineage that SOX Section 404 internal-control reviews require, and it isolates failures to individual batches instead of the whole run.

Design Constraints & Prerequisites

Before wiring this engine into an accounts-payable stack, four contracts must hold:

Upstream data contract. Each receipt arrives as a JSON record with a stable receipt_id, an image or PDF URI, and a submission_ts. The idempotency and grouping guarantees are established one layer up, in Building async batch queues for high-volume receipt uploads; this engine assumes messages are already deduplicated by an idempotency key.
Memory budget. Peak resident memory must scale with chunk_size × concurrency, never with total batch size. That rules out list materialization and forces async generators plus a maxsize-bounded queue.
CPU isolation. OCR is CPU-bound and must run in a thread or process pool via asyncio.to_thread() — never inline on the event loop, which would stall every other coroutine.
Compliance preconditions. Every stage transition emits an append-only log line carrying batch_id, correlation_id, stage, and status. Policy violations must halt downstream routing deterministically so unvalidated records never reach the general ledger. The rule set applied downstream comes from the Core Policy Architecture & Taxonomy Design layer.

Production Python Implementation

The engine below is self-contained and runnable on Python 3.11+. It streams receipts as fixed-size chunks, advances each batch through an explicit state machine, offloads CPU-bound OCR to a thread pool, and emits audit-ready JSON telemetry at every transition. Policy evaluation delegates to whatever rule engine your organization runs downstream — here a minimal daily-limit check stands in for the real Automated Policy Validation & Anomaly Flagging service.

import asyncio
from enum import Enum
from typing import AsyncIterator, Callable, Dict, List, Any, Awaitable

from pydantic import BaseModel, Field
import structlog

# --- Structured, append-only audit logging -------------------------------
structlog.configure(
    processors=[
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.add_log_level,
        structlog.processors.JSONRenderer(),
    ],
    logger_factory=structlog.PrintLoggerFactory(),
)
logger = structlog.get_logger()


class PipelineStage(str, Enum):
    OCR = "ocr"
    LINE_ITEM_PARSE = "line_item_parse"
    POLICY_VALIDATE = "policy_validate"
    ERP_ROUTING = "erp_routing"          # terminal success state
    REQUIRES_REVIEW = "requires_review"  # terminal quarantine state
    FAILED = "failed"                    # terminal error state


class ExpenseReceipt(BaseModel):
    receipt_id: str
    image_uri: str
    raw_text: str = ""
    line_items: List[Dict[str, Any]] = Field(default_factory=list)
    policy_violations: List[str] = Field(default_factory=list)
    status: str = "INGESTED"


class BatchContext(BaseModel):
    batch_id: str
    correlation_id: str
    stage: PipelineStage = PipelineStage.OCR
    receipts: List[ExpenseReceipt]
    processed_count: int = 0
    error_count: int = 0


async def stream_receipt_chunks(
    receipts: List[Dict[str, Any]], chunk_size: int = 25
) -> AsyncIterator[List[ExpenseReceipt]]:
    """Memory-efficient generator yielding fixed-size, validated chunks.

    Resident memory scales with chunk_size, not len(receipts): the source list
    is sliced lazily and each slice is validated into models on demand.
    """
    for i in range(0, len(receipts), chunk_size):
        yield [ExpenseReceipt(**r) for r in receipts[i : i + chunk_size]]
        await asyncio.sleep(0)  # yield control so the loop stays responsive


def _run_tesseract(image_uri: str) -> str:
    """CPU-bound OCR. Bind to your Tesseract wrapper; kept pure for testing.

    See /receipt-ingestion-ocr-data-extraction/tesseract-ocr-configuration/
    for DPI, PSM, and whitelist tuning that raises accuracy on faded receipts.
    """
    return f"OCR_EXTRACTED::{image_uri}"


async def run_ocr(batch: BatchContext) -> BatchContext:
    """Offload CPU-bound OCR to a worker thread so the event loop never blocks."""
    logger.info("ocr_started", batch_id=batch.batch_id,
                correlation_id=batch.correlation_id, count=len(batch.receipts))
    for receipt in batch.receipts:
        receipt.raw_text = await asyncio.to_thread(_run_tesseract, receipt.image_uri)
    batch.stage = PipelineStage.LINE_ITEM_PARSE
    logger.info("ocr_completed", batch_id=batch.batch_id, stage=batch.stage.value)
    return batch


async def parse_line_items(batch: BatchContext) -> BatchContext:
    """Deterministic tabular extraction with schema validation.

    Production implementations use pdfplumber for ruled tables; see
    /receipt-ingestion-ocr-data-extraction/pdfplumber-line-item-parsing/.
    """
    logger.info("line_item_parse_started", batch_id=batch.batch_id)
    for receipt in batch.receipts:
        # Placeholder deterministic extraction — replace with real parser output.
        receipt.line_items = [
            {"description": "Travel", "amount": 142.50, "currency": "USD"}
        ]
    batch.stage = PipelineStage.POLICY_VALIDATE
    logger.info("line_item_parse_completed", batch_id=batch.batch_id)
    return batch


async def validate_policies(batch: BatchContext) -> BatchContext:
    """Rule-based evaluation with deterministic violation routing.

    A violation halts downstream routing for the whole batch, guaranteeing that
    unvalidated financial data never reaches the ledger.
    """
    logger.info("policy_validate_started", batch_id=batch.batch_id)
    flagged: List[str] = []
    for receipt in batch.receipts:
        for item in receipt.line_items:
            if item["amount"] > 100.00:
                receipt.policy_violations.append("EXCEEDS_DAILY_LIMIT")
                flagged.append(receipt.receipt_id)

    if flagged:
        batch.stage = PipelineStage.REQUIRES_REVIEW
        logger.warning("policy_violations_detected", batch_id=batch.batch_id,
                       violation_count=len(flagged))
    else:
        batch.stage = PipelineStage.ERP_ROUTING
    logger.info("policy_validate_completed", batch_id=batch.batch_id,
                stage=batch.stage.value)
    return batch


# Explicit DAG: each non-terminal stage maps to exactly one handler. A batch
# can only advance along declared edges, so stages can never be skipped.
STAGE_HANDLERS: Dict[PipelineStage, Callable[[BatchContext], Awaitable[BatchContext]]] = {
    PipelineStage.OCR: run_ocr,
    PipelineStage.LINE_ITEM_PARSE: parse_line_items,
    PipelineStage.POLICY_VALIDATE: validate_policies,
}
TERMINAL_STAGES = {
    PipelineStage.ERP_ROUTING,
    PipelineStage.REQUIRES_REVIEW,
    PipelineStage.FAILED,
}


async def process_batch(batch: BatchContext) -> BatchContext:
    """Drive a batch through the DAG until it reaches a terminal stage."""
    while batch.stage not in TERMINAL_STAGES:
        handler = STAGE_HANDLERS.get(batch.stage)
        if handler is None:
            raise RuntimeError(f"Unhandled pipeline stage: {batch.stage}")
        batch = await handler(batch)
    batch.processed_count = len(batch.receipts)
    logger.info("batch_terminal", batch_id=batch.batch_id,
                correlation_id=batch.correlation_id, stage=batch.stage.value)
    return batch


async def worker(name: str, queue: "asyncio.Queue[BatchContext]") -> None:
    """Consumer coroutine with strict per-batch error isolation."""
    while True:
        batch = await queue.get()
        try:
            await process_batch(batch)
        except Exception as exc:  # isolate failure to this batch only
            batch.stage = PipelineStage.FAILED
            batch.error_count = len(batch.receipts)
            logger.error("batch_failed", worker=name, batch_id=batch.batch_id,
                         error=str(exc))
        finally:
            queue.task_done()


async def run_pipeline(
    receipts: List[Dict[str, Any]],
    *,
    chunk_size: int = 25,
    max_queue: int = 50,
    concurrency: int = 4,
) -> None:
    """Fan receipts into a bounded queue and drain it with N workers.

    Backpressure: producers block on queue.put() once max_queue batches are
    in flight, so memory stays bounded by chunk_size * (max_queue + concurrency).
    """
    queue: "asyncio.Queue[BatchContext]" = asyncio.Queue(maxsize=max_queue)
    workers = [asyncio.create_task(worker(f"w{i}", queue)) for i in range(concurrency)]

    seq = 0
    async for chunk in stream_receipt_chunks(receipts, chunk_size=chunk_size):
        seq += 1
        await queue.put(BatchContext(
            batch_id=f"batch-{seq:06d}",
            correlation_id=f"run-{seq:06d}",
            receipts=chunk,
        ))

    await queue.join()               # wait for all batches to reach a terminal stage
    for w in workers:
        w.cancel()
    await asyncio.gather(*workers, return_exceptions=True)


if __name__ == "__main__":
    sample = [
        {"receipt_id": f"R-{i}", "image_uri": f"s3://receipts/{i}.jpg"}
        for i in range(1000)
    ]
    asyncio.run(run_pipeline(sample, chunk_size=25, max_queue=50, concurrency=4))

The key invariants: memory is a function of chunk_size × (max_queue + concurrency), never of total volume; the state machine forbids stage-skipping; and worker catches exceptions per batch so one bad receipt cannot abort the run. Batches that reach REQUIRES_REVIEW are held for the anomaly-flagging workflow rather than routed to the ledger.

Configuration Reference

Tune the engine through the run_pipeline keyword arguments and the environment. Pin the dependencies (pydantic>=2.6, structlog>=24.1) so schema-validation semantics stay stable across deploys.

Flag / key	Type	Default	Rationale
`chunk_size`	int	`25`	Receipts per batch. Larger chunks amortize per-batch overhead but raise peak memory linearly.
`max_queue`	int	`50`	Bounded in-flight batches. This is the backpressure valve — lower it if producers outrun OCR.
`concurrency`	int	`4`	Number of worker coroutines. Match to OCR thread-pool width, not to CPU count directly.
`OCR_THREAD_LIMIT`	int (env)	CPU count	Caps `asyncio.to_thread` executor threads so OCR cannot starve the loop.
`POLICY_DAILY_LIMIT`	float (env)	`100.00`	Threshold that trips `EXCEEDS_DAILY_LIMIT`; source real values from the policy taxonomy.
`LOG_LEVEL`	str (env)	`INFO`	`structlog` verbosity. Keep at `INFO` in production so the audit trail stays complete.

Peak resident memory is approximately chunk_size × (max_queue + concurrency) × avg_receipt_bytes. Size max_queue and chunk_size against your worker’s memory limit before raising concurrency.

Validation & Testing

Because every handler is a pure async function over a typed BatchContext, the DAG is unit-testable without a broker or real OCR. Assert on terminal stage, violation routing, and error isolation.

import asyncio
import pytest
from decimal import Decimal  # use for money in real code; float here matches the demo

from pipeline import (  # module shown above
    BatchContext, ExpenseReceipt, PipelineStage,
    process_batch, run_pipeline,
)


def _batch(*amounts: float) -> BatchContext:
    receipts = [
        ExpenseReceipt(
            receipt_id=f"R-{i}",
            image_uri=f"s3://receipts/{i}.jpg",
            line_items=[{"description": "x", "amount": a, "currency": "USD"}],
        )
        for i, a in enumerate(amounts)
    ]
    # Skip OCR/parse for the assertion by pre-seeding line items:
    b = BatchContext(batch_id="b1", correlation_id="c1", receipts=receipts)
    b.stage = PipelineStage.POLICY_VALIDATE
    return b


@pytest.mark.asyncio
async def test_clean_batch_routes_to_erp():
    result = await process_batch(_batch(42.0, 88.0))
    assert result.stage is PipelineStage.ERP_ROUTING
    assert result.processed_count == 2


@pytest.mark.asyncio
async def test_violation_quarantines_batch():
    result = await process_batch(_batch(42.0, 250.0))
    assert result.stage is PipelineStage.REQUIRES_REVIEW
    assert "EXCEEDS_DAILY_LIMIT" in result.receipts[1].policy_violations


@pytest.mark.asyncio
async def test_worker_isolates_failure(monkeypatch):
    # A batch missing line_items still terminates FAILED, never crashes the run.
    bad = BatchContext(batch_id="b2", correlation_id="c2",
                       receipts=[ExpenseReceipt(receipt_id="R", image_uri="s3://x")])
    bad.stage = PipelineStage.POLICY_VALIDATE
    bad.receipts[0].line_items = [{"description": "x"}]  # missing "amount"
    with pytest.raises(KeyError):
        await process_batch(bad)  # process_batch re-raises; worker() would catch it


@pytest.mark.asyncio
async def test_pipeline_bounded_memory_smoke():
    # 1k receipts through 4 workers must complete without unbounded queue growth.
    await asyncio.wait_for(
        run_pipeline([{"receipt_id": f"R-{i}", "image_uri": f"s3://{i}"}
                      for i in range(1000)], chunk_size=25, max_queue=10),
        timeout=10,
    )

Edge-case fixtures worth keeping in the suite: faded receipts that yield empty raw_text (must not silently pass to routing), timezone-drifted submission_ts that belong to a prior batch window, and split transactions where one receipt produces multiple line items across categories. Duplicate detection across those windows is owned by the Duplicate Receipt Detection pipeline and should be asserted there, not here.

Operational Runbook

Deploy, monitor, and roll back the batch engine with the following checklist.

Pre-deploy. Pin dependency versions, run the test suite, and confirm OCR_THREAD_LIMIT matches the container’s CPU quota. Verify the downstream policy service and ERP export endpoint are reachable.
Deploy behind the queue. Start workers first, then enable producers. Because the queue is bounded, workers draining an empty queue is harmless; producers filling a queue with no workers would block — so ordering matters.

Monitor these signals.

Signal	Healthy	Alert threshold
Queue depth	oscillates below `max_queue`	pinned at `max_queue` for > 5 min (consumers behind)
`batch_failed` rate	< 0.5% of batches	> 2% sustained over 10 min
`REQUIRES_REVIEW` rate	matches historical baseline	2× baseline (policy or OCR regression)
Worker RSS	flat, bounded	trending upward (leak / chunk too large)

Roll back. Stop producers, let queue.join() drain in-flight batches to a terminal stage, then redeploy the prior image. No batch is left half-processed because terminal stages are the only exit.
Reprocess quarantined batches. After compliance review, resubmit REQUIRES_REVIEW batches at the POLICY_VALIDATE stage with the resolved policy version — never earlier, so OCR and parsing results are not recomputed and lineage stays intact. Misclassified OCR failures are triaged via Receipt error categorization.

Together, bounded-queue backpressure, a stage-skip-proof state machine, and per-batch error isolation turn high-volume receipt auditing from a fragile synchronous job into a continuous, audit-ready control that scales horizontally.

Receipt Ingestion & OCR Data Extraction — parent overview of the ingestion and extraction stages.
Building async batch queues for high-volume receipt uploads — the broker-level queue topology and idempotency design this engine sits on.
Tesseract OCR configuration — tuning the CPU-bound OCR step invoked per batch.
pdfplumber line-item parsing — deterministic tabular extraction for the parse stage.
Receipt error categorization — routing failed and low-confidence receipts to manual review.
Automated Policy Validation & Anomaly Flagging — the downstream engine that consumes validated batches.