Building async batch queues for high-volume receipt uploads

A month-end surge of tens of thousands of receipts overwhelms a synchronous uploader because ingestion, OCR, and policy validation share one blocking call path, so a single slow scan stalls the whole worker and re-submitted uploads get processed twice. This guide builds the transport-level queue topology and idempotent payload contract that the parent Async Batch Processing engine consumes — the broker, the message schema, backpressure, and exactly-once delivery that sit one layer below the pipeline state machine. It lives inside the broader Receipt Ingestion & OCR Data Extraction framework and hands deduplicated, ordered batches to the OCR and parsing stages that follow.

Why Standard Approaches Fail

Synchronous, in-process ingestion breaks at high volume for three named reasons, each of which corrupts the audit trail rather than surfacing a clean error:

1. Thread-pool exhaustion under head-of-line blocking. A synchronous handler holds a worker thread for the entire ingest → OCR → validate span. When OCR takes 800 ms on a faded scan, every queued upload waits behind it; the pool saturates during peak submission windows and new uploads time out at the load balancer. OCR is CPU-bound and must be isolated from the network-bound ingest path, not serialized behind it.
2. Unbounded buffer materialization. Loading an entire upload batch into memory before processing makes peak resident memory scale with total volume, so a 40,000-receipt surge triggers OOM kills on fixed-size workers. Memory must scale with batch_size × concurrency, never with the size of the run.
3. Non-idempotent redelivery. Any at-least-once broker (SQS, RabbitMQ, Redis Streams) will redeliver a message whose ack was lost to a worker crash or a visibility-timeout expiry. Without a deduplication key checked at the consumer entry point, the same receipt is OCR’d and policy-evaluated twice, producing duplicate ledger entries that the Duplicate Receipt Detection pipeline then has to unwind after the fact.

Architecture & Algorithm

Decouple the stages with a message broker configured for explicit visibility timeouts, a consumer group, and a dead-letter queue (DLQ). Group messages by a logical boundary — expense report UUID or a submission-timestamp window — so a report is never split across batches, which keeps downstream reconciliation deterministic. Each message carries a strict, versioned schema whose idempotency_key is a SHA-256 hash of receipt_id concatenated with submission_ts:

{
  "schema_version": 1,
  "receipt_id": "550e8400-e29b-41d4-a716-446655440000",
  "report_id": "9b1deb4d-3b7d-4bad-9bdd-2b0d7b3dcb6d",
  "image_uri": "s3://expense-bucket/raw/2024-05-12/receipt_001.jpg",
  "submission_ts": "2024-05-12T14:32:01Z",
  "idempotency_key": "a1b2c3d4e5f6...",
  "retry_count": 0,
  "batch_window": "2024-05-12T14:30:00Z"
}

Workers pull batches of 50–200 messages, check the idempotency key before any work, offload CPU-bound OCR to a ProcessPoolExecutor (never inline on the event loop, where it would stall every other coroutine), and ack only after the extracted record is routed downstream. If a worker crashes mid-batch, the broker redelivers only the unacknowledged messages; the entry-point dedup check absorbs the redelivery, giving exactly-once effects on top of at-least-once delivery. Memory notes are inline: the process pool caps parallel OCR, and gc.collect() between batches reclaims image buffers and OCR model weights that would otherwise accumulate across iterations.

import asyncio
import gc
import logging
from concurrent.futures import ProcessPoolExecutor
from typing import Any, Dict, List

from pydantic import BaseModel, Field

logger = logging.getLogger("expense.ingest.queue")


class ReceiptPayload(BaseModel):
    schema_version: int = 1
    receipt_id: str
    report_id: str
    image_uri: str
    submission_ts: str
    idempotency_key: str
    retry_count: int = 0
    batch_window: str = ""


class IdempotencyStore:
    """Entry-point dedup. Back this with Redis SET key NX in production so the
    guarantee survives worker restarts; the in-memory set is for tests."""

    def __init__(self) -> None:
        self._seen: set[str] = set()

    def claim(self, key: str) -> bool:
        """Return True if this key is new (claimed), False if already processed."""
        if key in self._seen:
            return False
        self._seen.add(key)
        return True


# Bind these two hooks to your OCR engine and downstream policy service.
def run_ocr_extraction(payload: Dict[str, Any]) -> Dict[str, Any]:
    ...


async def route_to_policy_engine(ocr_result: Dict[str, Any]) -> None:
    ...


async def process_batch(
    broker: Any,
    raw_messages: List[Dict[str, Any]],
    dedup: IdempotencyStore,
    pool: ProcessPoolExecutor,
) -> Dict[str, int]:
    """Consume one batch: validate, deduplicate, OCR in a process pool, route.

    Emits an audit-ready count of accepted/duplicate/failed so batch outcomes
    are reconstructable from structured logs."""
    loop = asyncio.get_running_loop()
    accepted: List[ReceiptPayload] = []
    duplicates = 0

    for msg in raw_messages:
        receipt = ReceiptPayload(**msg)
        if not dedup.claim(receipt.idempotency_key):
            duplicates += 1
            logger.info("duplicate_skipped receipt_id=%s", receipt.receipt_id)
            await broker.ack(receipt.receipt_id)  # ack so it is not redelivered
            continue
        accepted.append(receipt)

    ocr_tasks = [
        loop.run_in_executor(pool, run_ocr_extraction, r.model_dump())
        for r in accepted
    ]
    results = await asyncio.gather(*ocr_tasks, return_exceptions=True)

    failed = 0
    for receipt, result in zip(accepted, results):
        if isinstance(result, Exception):
            failed += 1
            logger.error("ocr_failed receipt_id=%s error=%s", receipt.receipt_id, result)
            await broker.nack(receipt.receipt_id)  # redeliver / route to retry queue
        else:
            await route_to_policy_engine(result)
            await broker.ack(receipt.receipt_id)

    gc.collect()  # reclaim image buffers + model weights before the next batch
    counts = {"accepted": len(accepted), "duplicates": duplicates, "failed": failed}
    logger.info("batch_complete %s", counts)
    return counts

ReceiptPayload uses model_dump(), the Pydantic v2 API; on Pydantic v1 replace it with dict(). Isolating OCR in the process pool sidesteps GIL contention entirely, which is what eliminates the thread starvation and keeps per-receipt latency bounded even under sustained throughput.

Extractions whose OCR confidence is degraded should not reach the policy engine at all. Gate routing on a batch-level drift metric so crumpled paper, thermal fade, or a misread currency symbol (€ read as £, 12.50 shifted to 1250) diverts to a low-confidence queue instead of silently corrupting policy evaluation:

import re
from typing import Any, Dict

CURRENCY_PATTERN = re.compile(r"^(?:USD|EUR|GBP|CAD|JPY|AUD)\s?\d{1,3}(?:[.,]\d{3})*(?:[.,]\d{2})$")
BASELINE_CONFIDENCE = 0.92
DRIFT_THRESHOLD = 0.15


def evaluate_drift(ocr_result: Dict[str, Any]) -> Dict[str, Any]:
    """Route an extraction by confidence: policy engine, low-confidence review, or DLQ."""
    line_scores = [ln.get("confidence", 0.0) for ln in ocr_result.get("lines", [])]
    if not line_scores:
        return {"route": "dlq", "reason": "empty_extraction"}

    mean_conf = sum(line_scores) / len(line_scores)
    drift_score = round(1 - (mean_conf / BASELINE_CONFIDENCE), 4)
    currency_ok = bool(CURRENCY_PATTERN.match(ocr_result.get("extracted_amount", "")))

    if drift_score > DRIFT_THRESHOLD or not currency_ok:
        return {
            "route": "low_confidence_ocr",
            "drift_score": drift_score,
            "drift_reason": "currency_mismatch" if not currency_ok else "low_confidence",
            "payload": ocr_result,
        }
    return {"route": "policy_engine", "drift_score": drift_score, "payload": ocr_result}

Records that fail extraction outright are triaged through classifying OCR extraction errors for manual review, while confidence tuning on faded scans is the job of optimizing Tesseract for faded receipt text.

Step-by-Step Integration

1. Provision the broker with a DLQ and a realistic visibility timeout. Set the visibility timeout to at least 3× your p99 per-batch processing time so a slow batch is not redelivered while still in flight. Point the main queue’s redrive policy at a DLQ after a bounded maxReceiveCount (typically 5).

2. Make the dedup store durable. Swap the in-memory set for SET key value NX against Redis with a TTL longer than your retry window, so the exactly-once guarantee survives a worker restart:

   claimed = await redis.set(f"idem:{key}", "1", nx=True, ex=86_400)
   assert claimed is None or claimed is True  # None => already processed

3. Size the process pool to cores, the batch to memory. Set ProcessPoolExecutor(max_workers=os.cpu_count()) and pick a batch size so batch_size × concurrency × avg_image_bytes stays under the container memory limit. Verify peak RSS stays flat across batches before raising either.

4. Verify redelivery is absorbed. Feed the same idempotency_key twice and assert the second pass is skipped:

   store = IdempotencyStore()
   assert store.claim("a1b2c3") is True    # first delivery processes
   assert store.claim("a1b2c3") is False   # redelivery is a no-op

5. Wire the drift gate before the policy hand-off. Route evaluate_drift(...)["route"] — never call route_to_policy_engine on a low_confidence_ocr or dlq result. This keeps the Automated Policy Validation & Anomaly Flagging engine fed only clean records.

6. Deploy consumers before producers. Start the worker group first; workers draining an empty queue is harmless, but producers filling a queue with no consumers wastes visibility-timeout budget. Then enable ingestion and watch queue depth settle below its bound.

7. Emit signed audit records for every routing decision. Before dispatching to any queue, publish a timestamped, HMAC-signed record to an append-only audit stream so SOX and internal-audit reviews can reconstruct why each receipt went where it did. Use hashlib.sha256 as the digestmod — passing the string "sha256" raises TypeError at runtime.

Edge Cases & Gotchas

Instrument queue depth, consumer lag, and the drift-score distribution as Prometheus metrics. Alert when drift_score > 0.15 for more than 10% of daily volume — that signals upstream capture degradation or OCR model decay, not a queue problem, and should page the ingestion owner rather than the platform team.

FAQ

How do I get exactly-once processing on an at-least-once broker?

You do not get exactly-once delivery — SQS, RabbitMQ, and Redis Streams all redeliver on a lost ack. You get exactly-once effects by making the consumer idempotent: hash receipt_id + submission_ts into a key, claim it with SET key NX in Redis before doing any work, and treat a failed claim as a signal to ack-and-skip. Redelivery then becomes a cheap no-op.

What batch size should workers pull?

Between 50 and 200 messages, sized so batch_size × concurrency × avg_image_bytes fits under the worker memory limit. Larger batches amortize per-batch broker overhead but raise peak RSS linearly and lengthen the visibility window a single failure holds. Start at 100, watch RSS and consumer lag, then adjust one variable at a time.

Why isolate OCR in a process pool instead of a thread pool?

OCR is CPU-bound, so threads contend on the GIL and give almost no parallelism while still blocking the event loop. A ProcessPoolExecutor runs extraction in separate interpreters with independent GILs, so N cores do N receipts at once, and the async ingest path stays responsive to new uploads.

When should a message go to the dead-letter queue versus the retry queue?

Transient failures — a timed-out OCR call, a downstream 503 — go back to a retry queue and are redelivered with an incremented retry_count. Only messages that exhaust maxReceiveCount, or that produce structurally unusable extractions (empty text, unparseable amount), belong in the DLQ, where they feed a human-review lane rather than looping forever.

How do I keep the queue’s routing decisions audit-ready?

Publish an HMAC-signed, timestamped record to an append-only audit stream before dispatching each message to its destination queue, capturing the route, the drift score, and a signature over the canonical payload. Because the write precedes the dispatch, a reviewer can always reconstruct the decision even if the downstream stage later fails.

Related

• Async Batch Processing — the parent guide whose pipeline state machine consumes the batches this queue emits.
• Extracting line items from scanned PDFs with pdfplumber — the sibling parse stage that runs on dequeued receipts.
• Optimizing Tesseract for faded receipt text — raises the OCR confidence that the drift gate keys on.
• Classifying OCR extraction errors for manual review — where DLQ and low-confidence messages divert.
• Duplicate Receipt Detection — the downstream pipeline that redelivery without idempotency would flood with duplicates.