Async Batch Processing for Government Records and FOIA Automation #

Within Document Retrieval & Parsing, async batch processing is the throughput layer that lets a single pipeline pull, validate, and hand off thousands of responsive documents concurrently without blocking the compliance-review queue or breaching a statutory deadline. For public sector engineering teams and the compliance officers who certify their output, the problem is not raw speed — it is bounded, observable speed: concurrency that respects upstream rate limits, fails safe under partial outage, and emits a tamper-evident audit trail for every document it touches. This guide walks through a production-ready implementation: modelling the batch, constraining concurrency with semaphores and backpressure, degrading gracefully through a circuit breaker, and proving correctness with idempotency and log assertions.

Problem Framing and Statutory Requirement #

A FOIA fulfillment run is a deadline-bound bulk operation. The federal Freedom of Information Act sets a 20-business-day clock on the agency’s substantive response (5 U.S.C. § 552(a)(6)(A)(i)), and state open-records analogues impose their own, often shorter, windows. When a single request resolves to thousands of documents — an email export, a permit archive, a body-camera index — sequential retrieval simply cannot clear the volume inside the window. Async batch processing exists to compress that wall-clock time while keeping every other guarantee intact.

The failure modes here are compliance failures, not merely performance ones. Unbounded concurrency exhausts connection pools and trips the rate limits of the upstream document store, which under load looks identical to a denial-of-service condition the pipeline inflicted on itself — exactly what NIST SP 800-53 SC-5 controls are meant to prevent. A worker that crashes mid-batch and silently drops records produces an incomplete production, which is a defective FOIA response. And a batch that processes documents without a per-record audit line leaves the agency unable to prove what it disclosed, when, and on what input during litigation. The controls a compliant batch engine must enforce are therefore: idempotent ingestion so retries never duplicate or skip a record, bounded concurrency with backpressure so the pipeline never overruns its dependencies, deterministic fallback so a degraded extraction service never blocks the event loop, and an append-only audit record keyed by a trace ID that survives across service boundaries.

Async batch processing does not stand alone. It consumes work classified upstream by Async Queue Management and ordered by Priority Scoring Algorithms, validates provenance through Repository Sync Protocols, and hands successful payloads to OCR Processing Pipelines for text extraction.

Prerequisites and Environment Setup #

This implementation targets Python 3.11 or later and keeps the dependency surface deliberately small so the retrieval path is easy to vet:

• Python 3.11+ for asyncio (including asyncio.TaskGroup and timeout), dataclasses, hashlib, and the logging module used for structured JSON audit output.
• aiohttp (3.9+) for non-blocking HTTP retrieval against the document store, or an async SDK wrapper that exposes the same connector semantics.
• tenacity (8.x) for declarative retry with exponential backoff and jitter, scoped to transient exceptions only.
• A durable message broker (RabbitMQ, AWS SQS, or Redis Streams) as the source of the manifest queue, so ingestion is decoupled from execution and survives a worker restart.
• Append-only audit storage — WORM object storage or a SIEM-forwarding log handler — so the per-record audit lines satisfy NIST SP 800-53 AU-9 (protection of audit information) and cannot be rewritten after the fact.
• Least-privilege execution. The worker identity must be scoped by Security Boundary Configuration so a batch job can read only the document store it is authorized for and can never widen its own access during a run.

A useful invariant before you write any worker code: every unit of work entering the batch must carry a stable, content-derived identifier. That single decision is what makes the entire pipeline safe to retry.

Architecture Overview #

The engine pulls manifests from a durable broker, gates each retrieval behind a shared semaphore, wraps the network call in retry-plus-circuit-breaker logic, and emits exactly one terminal audit line per record. Successful payloads flow downstream to extraction; transient failures retry with backoff; hard failures and open-circuit records divert to a quarantine queue rather than failing the whole batch. When queue depth exceeds a high-water mark, the worker signals backpressure to pause ingestion at the broker.

Step-by-Step Implementation #

1. Model the work unit with idempotent identity #

Every document in the batch is represented by an immutable manifest carrying a content-derived idempotency key. Freezing the dataclass guarantees the key and source URL cannot drift between enqueue and processing, which is the precondition for safe retries and for a meaningful audit hash.

import hashlib
from dataclasses import dataclass
from enum import Enum


class RecordStatus(str, Enum):
    COMPLETED = "completed"
    QUARANTINED = "quarantined"
    DEFERRED = "deferred"


@dataclass(frozen=True)
class DocumentManifest:
    doc_id: str          # agency record identifier
    url: str             # location in the authoritative document store
    request_id: str      # parent FOIA request, for deadline accounting
    content_sha256: str  # idempotency key: identical content -> identical key


def manifest_from_source(doc_id: str, url: str, request_id: str, raw: bytes) -> DocumentManifest:
    """Derive the idempotency key from content so a re-enqueued record
    is recognised as a duplicate rather than processed twice."""
    digest = hashlib.sha256(raw).hexdigest()
    return DocumentManifest(doc_id=doc_id, url=url, request_id=request_id,
                            content_sha256=digest)

Expected behaviour: because the manifest is frozen, any attempt to reassign a field raises dataclasses.FrozenInstanceError, and two manifests built from identical bytes produce identical content_sha256 values — the property a deduplicating broker relies on.

2. Bound concurrency, retry transient faults, and break circuits #

Concurrency is constrained by a single asyncio.Semaphore sized to the smallest of available I/O bandwidth, connection-pool size, and the upstream service quota — never to an arbitrary thread count. Transient faults retry with exponential backoff and jitter; a CircuitBreaker short-circuits repeated failures so the worker stops hammering a degraded dependency. Each terminal outcome is written as one structured JSON audit line carrying the cross-service trace_id.

import asyncio
import datetime
import json
import logging
import time
import uuid
from aiohttp import ClientSession, ClientTimeout, ClientResponseError
from tenacity import (retry, stop_after_attempt, wait_exponential_jitter,
                      retry_if_exception_type)

# Append-only structured audit log. In production this handler forwards to
# WORM storage / a SIEM (NIST SP 800-53 AU-9) so lines cannot be altered.
AUDIT = logging.getLogger("foia_batch_audit")
AUDIT.setLevel(logging.INFO)


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


class CircuitBreaker:
    """Trips OPEN after repeated failures; recovers via a HALF_OPEN probe."""
    def __init__(self, failure_threshold: int = 5, recovery_timeout: float = 30.0):
        self.failure_count = 0
        self.threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.last_failure = 0.0
        self.state = "CLOSED"

    def allow(self) -> bool:
        if self.state == "OPEN":
            if time.monotonic() - self.last_failure > self.recovery_timeout:
                self.state = "HALF_OPEN"   # allow a single probe through
                return True
            return False
        return True

    def record_success(self) -> None:
        self.failure_count = 0
        self.state = "CLOSED"

    def record_failure(self) -> None:
        self.failure_count += 1
        self.last_failure = time.monotonic()
        if self.failure_count >= self.threshold:
            self.state = "OPEN"


@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential_jitter(initial=2, max=10),   # backoff + jitter
    retry=retry_if_exception_type((ConnectionError, asyncio.TimeoutError,
                                   ClientResponseError)),
    reraise=True,
)
async def fetch_document(session: ClientSession, manifest: "DocumentManifest") -> bytes:
    """Non-blocking retrieval. raise_for_status surfaces 5xx as a retryable
    ClientResponseError; a per-request timeout caps tail latency."""
    async with session.get(manifest.url, timeout=ClientTimeout(total=15)) as resp:
        resp.raise_for_status()
        return await resp.read()

Expected behaviour: a 503 from the document store raises ClientResponseError, which tenacity retries up to three times with jittered backoff; if the breaker has already tripped OPEN, allow() returns False and the worker defers the record instead of issuing a doomed request.

3. Drive the bounded worker pool with backpressure #

The pool runs under an asyncio.TaskGroup so a failure in one worker is collected without aborting the batch. The semaphore caps in-flight retrievals; deferred records (open circuit) are re-queued for a later cycle, hard failures are quarantined, and the run produces a per-status summary the operator can reconcile against the request.

async def process_one(manifest: "DocumentManifest", session: ClientSession,
                      sem: asyncio.Semaphore, breaker: CircuitBreaker,
                      retry_queue: asyncio.Queue, summary: dict) -> None:
    trace_id = str(uuid.uuid4())   # propagates across the OCR handoff boundary
    async with sem:                # bounded concurrency: never overrun the store
        if not breaker.allow():
            _audit("CIRCUIT_OPEN_DEFERRED", manifest, trace_id)
            await retry_queue.put(manifest)         # revisit on the next cycle
            summary[RecordStatus.DEFERRED] += 1
            return
        try:
            payload = await fetch_document(session, manifest)
            breaker.record_success()
            _audit("RETRIEVED", manifest, trace_id, bytes=len(payload))
            summary[RecordStatus.COMPLETED] += 1
            # ... hand payload + trace_id to the OCR pipeline here ...
        except Exception as exc:
            breaker.record_failure()
            # Fail safe: quarantine for human review, never drop silently —
            # a dropped record is an incomplete FOIA production.
            _audit("QUARANTINED", manifest, trace_id, error=repr(exc))
            summary[RecordStatus.QUARANTINED] += 1


async def run_batch(manifests: list["DocumentManifest"], max_concurrency: int = 50,
                    high_water: int = 5000) -> dict:
    sem = asyncio.Semaphore(max_concurrency)
    breaker = CircuitBreaker(failure_threshold=5, recovery_timeout=20.0)
    retry_queue: asyncio.Queue = asyncio.Queue()
    summary = {s: 0 for s in RecordStatus}

    # Backpressure: if the pending set exceeds the high-water mark, the caller
    # should pause broker ingestion until this batch drains (NIST SP 800-53 SC-5).
    if len(manifests) > high_water:
        _audit_marker = "BACKPRESSURE_ENGAGED"

    async with ClientSession() as session:
        async with asyncio.TaskGroup() as tg:
            for m in manifests:
                tg.create_task(process_one(m, session, sem, breaker,
                                           retry_queue, summary))
    return {"summary": {k.value: v for k, v in summary.items()},
            "deferred_for_retry": retry_queue.qsize()}

Expected output for a 1,000-document batch where the store briefly degraded mid-run:

{"summary": {"completed": 982, "quarantined": 6, "deferred": 12}, "deferred_for_retry": 12}

The 12 deferred records are re-queued, not lost; the 6 quarantined records are flagged for human review before the production is certified complete. For the memory-tuning, I/O scheduling, and worker-lifecycle strategies that keep this pattern stable across hundreds of thousands of pages, see Optimizing batch OCR processing for large municipal archives.

Validation and Verification #

Treat the batch path as deadline-critical code and assert its invariants directly rather than eyeballing throughput:

import io
import logging


def test_idempotency_key_is_content_derived():
    a = manifest_from_source("D1", "http://store/1", "REQ-1", b"same bytes")
    b = manifest_from_source("D1", "http://store/1", "REQ-1", b"same bytes")
    assert a.content_sha256 == b.content_sha256   # safe to dedupe / retry


def test_breaker_opens_after_threshold():
    cb = CircuitBreaker(failure_threshold=2, recovery_timeout=60)
    cb.record_failure(); cb.record_failure()
    assert cb.allow() is False                    # stops hammering a dead dep


async def test_quarantine_emits_one_audit_line(caplog):
    caplog.set_level(logging.INFO, logger="foia_batch_audit")
    bad = DocumentManifest("D9", "http://unreachable", "REQ-2", "deadbeef")
    sem, cb, q = asyncio.Semaphore(1), CircuitBreaker(), asyncio.Queue()
    summary = {s: 0 for s in RecordStatus}
    async with ClientSession() as s:
        await process_one(bad, s, sem, cb, q, summary)
    lines = [r.message for r in caplog.records]
    assert sum('"event": "QUARANTINED"' in ln for ln in lines) == 1

Beyond unit tests, verify in production by reconciling the per-status summary count against the manifest count for each request_id (the totals must sum to the batch size, with no silent loss), filtering the audit stream on trace_id to confirm exactly one terminal event per document, and replaying a fixed manifest list to confirm deduplication holds.

Troubleshooting and Edge Cases #

• Connection-pool starvation under unbounded fan-out. Spawning a task per document without a semaphore exhausts aiohttp connectors and the document store’s quota, and the run stalls. Diagnosis: a spike in ClientConnectorError plus retrieval latency climbing with batch size. Fix: cap in-flight work with the shared asyncio.Semaphore shown above and size the TCPConnector limit to match the store’s documented quota.
• Retry amplification against a degraded service. Naive retries multiply load on a store that is already failing, deepening the outage and burning the FOIA clock. Diagnosis: retry counts rising while success rate falls. Fix: pair tenacity backoff-with-jitter with the circuit breaker so repeated failures trip OPEN and defer work instead of retrying into a wall.
• Duplicate submissions inflating a production. The same record re-enqueued after a worker restart is processed twice, double-counting the disclosure. Diagnosis: two RETRIEVED audit lines sharing one content_sha256. Fix: deduplicate on the content-derived idempotency key at the broker and assert uniqueness before handoff.
• Encoding and OCR artifacts breaking downstream handoff. A retrieved payload with a mislabelled charset or a corrupt scan crashes the extraction stage and, if uncaught, can abort the surrounding tasks. Diagnosis: extraction exceptions clustered on a particular agency’s exports. Fix: quarantine on any retrieval or decode error rather than raising, then route the quarantined batch through OCR Processing Pipelines with the appropriate preprocessing.
• Litigation-hold conflict surfacing mid-batch. A record under an active hold must not be released even if it processes cleanly. Diagnosis: a RETRIEVED/handoff event for a record flagged in the hold registry. Fix: re-check the hold registry from Records Retention Scheduling immediately before the OCR handoff, and treat any hold signal as a hard stop that diverts the record out of the production.

Compliance Verification Checklist #

• Every work unit carries a content-derived content_sha256 idempotency key, and the broker rejects duplicates on it.
• In-flight retrievals are capped by a shared asyncio.Semaphore sized to the smallest of bandwidth, pool size, and upstream quota.
• Transient faults retry with exponential backoff and jitter scoped to transient exception types only.
• A circuit breaker trips OPEN on repeated failure and defers work instead of retrying into a degraded dependency (NIST SP 800-53 SC-5).
• Backpressure pauses broker ingestion when pending work exceeds the high-water mark.
• Each record produces exactly one terminal JSON audit line carrying the cross-service trace_id, forwarded to append-only storage (AU-9).
• Hard failures are quarantined for human review — never silently dropped — and the run emits a per-status summary that sums to the batch size.
• Litigation holds are re-checked immediately before the OCR handoff and act as an absolute hard stop.
• The worker runs under a least-privilege identity scoped by the security boundary and cannot widen its own access mid-run.

FAQ #

Related #

• Document Retrieval & Parsing
• OCR Processing Pipelines
• Repository Sync Protocols
• Metadata Extraction Techniques
• Optimizing batch OCR processing for large municipal archives

← Back to all public records automation topics