Async Queue Management for Public Records & FOIA Intake

Within Intake & Routing Workflows, async queue management is the durability layer that decouples request ingestion from the slower work of validation, redaction, and fulfillment so that an unpredictable submission spike never costs an agency a statutory deadline. For government engineering teams and the compliance officers who certify their output, the requirement is not throughput for its own sake — it is deterministic throughput: a broker that survives a restart without losing a request, a consumer that never starts two deadline clocks for one submission, and an audit trail that can reconstruct exactly what happened to every record if the matter reaches an inspector general or a court. This guide walks a production-ready implementation: serializing requests into traceable tasks, configuring a persistent broker, routing by priority into isolated worker pools, and proving correctness with idempotency and log assertions.

Problem Framing & Statutory Requirement

A public records request is a deadline-bound obligation the moment it is received by the proper agency component. The federal Freedom of Information Act sets a 20-business-day clock on the agency’s substantive determination (5 U.S.C. § 552(a)(6)(A)(i)), and state open-records analogues impose their own — often shorter — windows. Synchronous request handling, where the HTTP worker that accepts a submission also validates, classifies, and stores it inline, fractures under load: a burst of submissions blocks the accepting thread, requests time out at the edge, and the agency cannot even prove what it received. Every one of those is a compliance failure, not merely a performance one.

Async queue management exists to absorb that burst safely. The accepting endpoint does the minimum needed to admit a request durably — stamp it, hash it, write it to a persistent broker — and returns. The heavy work happens on consumers that the agency can scale, rate-limit, and observe independently. The controls a compliant queue must enforce follow directly from the statute: durable persistence so a broker or worker crash re-delivers rather than loses a request; idempotent admission so a duplicate submission or a re-delivered message never starts a second deadline clock; priority-aware routing so an expedited-processing request is not stuck behind a bulk export; deadline-respecting retries so transient-error recovery never silently pushes a task past its statutory window; and an append-only audit record keyed by a correlation ID that survives across the broker, the worker, and downstream storage. Those guarantees are what let a records manager certify that nothing was dropped, duplicated, or quietly delayed.

Prerequisites & Environment Setup

This implementation targets Python 3.11+ and a broker that supports durable, acknowledged delivery. The reference stack is:

  • Celery 5.3+ as the task framework, backed by RabbitMQ 3.12+ (or Redis 7+ with persistence enabled) as the message broker. RabbitMQ is preferred where strict per-message acknowledgment and publisher confirms are required.
  • Pydantic 2.x for task-payload schema enforcement at the admission boundary.
  • The standard-library hashlib, json, logging, and datetime modules for hashing, structured logging, and deadline math.
  • A write-once audit sink — an append-only Postgres table, or object storage with retention lock (S3 Object Lock / equivalent) — to satisfy chain-of-custody and NIST SP 800-53 AU-9 (protection of audit information).

Access requirements: the admission service runs under a least-privilege identity that can publish to the broker but cannot consume or delete from the dead-letter exchange; consumers run under separate identities scoped to their worker pool, consistent with the Security Boundary Configuration for the deployment. Validated, deadline-stamped requests arrive here from Email & Form Parsing Pipelines; this page assumes those payloads are already normalized and integrity-stamped before they reach the queue.

Architecture Overview

The queue sits between a thin admission endpoint and the custodial worker pools. Requests are validated and hashed at admission, persisted to the broker, scored by Priority Scoring Algorithms into an execution tier, and dispatched by Department Routing Logic to an isolated pool. Rejected payloads and exhausted retries divert to a dead-letter queue (DLQ) rather than vanishing.

Async queue topology with priority preemption and isolated worker pools A thin intake endpoint feeds schema validation. Valid payloads publish to a persistent message broker; rejected payloads divert to a dead-letter queue. The broker hands each persisted task to priority scoring, which routes high-tier requests scored 80 or above into a high-priority lane and standard requests into a FIFO default lane. Both lanes converge on a department router that dispatches into three isolated worker pools: PII extraction, bulk generation, and cross-jurisdiction. Each pool and the broker emit one line per state transition to an append-only audit log keyed by the x-correlation-id. Intake endpoint stamp · hash · return Schema validation idempotency key · Pydantic Message broker durable · publisher confirms valid Dead-letter queue never silently dropped rejected persisted task Priority scoring tier select High-priority lane score ≥ 80 · preempts Default lane FIFO standard Department router jurisdiction → pool PII extraction pool dedicated · least-privilege Bulk generation pool heavy I/O · isolated Cross-jurisdiction pool chain-of-custody Append-only audit log · one JSON line per transition · x-correlation-id

Step-by-Step Implementation

1. Serialize and admit each request idempotently

Compliant queues begin at admission: every incoming request becomes a discrete, schema-validated task payload carrying submission metadata, jurisdiction tags, and SHA-256 hashes of its attachments. A deterministic idempotency key derived from the canonical payload is checked against a durable store before the task is published, so a re-submitted form or a re-delivered message never starts a second deadline clock. Validation failures route immediately to the DLQ with structured error context — no statutory request is ever silently dropped.

python 
import hashlib
import json
import logging
from datetime import datetime, timezone

from pydantic import BaseModel, ValidationError, field_validator
from celery import Celery

# Structured JSON audit logger — forwarded to an append-only sink (NIST SP 800-53 AU-9)
logger = logging.getLogger("foia.queue.audit")


class FOIATaskPayload(BaseModel):
    request_id: str            # server-generated, assigned at ingestion
    submission_ts: str         # ISO-8601 UTC receipt time — anchors the statutory clock
    jurisdiction: str          # resolves the applicable deadline window
    attachment_hashes: list[str]
    priority_score: int

    @field_validator("attachment_hashes")
    @classmethod
    def hashes_are_sha256(cls, value: list[str]) -> list[str]:
        for h in value:
            if len(h) != 64 or any(c not in "0123456789abcdef" for c in h):
                raise ValueError("attachment hash is not a valid SHA-256 digest")
        return value


def idempotency_key(task_data: dict) -> str:
    # Canonical (sorted-key) form so logically identical payloads hash identically
    canonical = json.dumps(task_data, sort_keys=True, separators=(",", ":"))
    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()


def admit_request(task_data: dict, broker: Celery, seen: set[str]) -> None:
    key = idempotency_key(task_data)
    if key in seen:  # durable store in production; set shown for illustration
        # Duplicate submission must NOT start a second 20-business-day clock
        logger.info(json.dumps({"event": "IDEMPOTENCY_BLOCK", "idempotency_key": key}))
        return
    try:
        payload = FOIATaskPayload(**task_data)
        broker.send_task(
            "foia.process_request",
            args=[payload.model_dump()],
            headers={"x-correlation-id": payload.request_id},
            # delivery_mode=2 => persistent: survives a broker restart without payload loss
            delivery_mode=2,
        )
        seen.add(key)
        logger.info(json.dumps({
            "event": "TASK_ADMITTED",
            "request_id": payload.request_id,
            "idempotency_key": key,
            "ts": datetime.now(timezone.utc).isoformat(),
        }))
    except ValidationError as exc:
        # Schema failure is a hard reject — route to DLQ with compliance context, never drop
        logger.critical(json.dumps({"event": "SCHEMA_REJECT", "error": exc.errors()}))
        broker.send_task("foia.dlq_handler", args=[task_data, str(exc)])

Expected behavior: a first submission emits a TASK_ADMITTED audit line and publishes a persistent task; an identical re-submission emits IDEMPOTENCY_BLOCK and publishes nothing; a malformed payload emits SCHEMA_REJECT and lands in the DLQ.

2. Configure the broker for durability and confirm persistence

Broker persistence is non-negotiable for government deployments. Tasks must be published persistently, acknowledgments must be late (after the work succeeds, not on receipt), and the broker must reject rather than discard unroutable messages. The settings below make every admitted request survive a restart and make every redelivery safe because consumers are idempotent.

python 
from celery import Celery

app = Celery("foia")
app.conf.update(
    broker_url="amqp://intake:***@rabbit:5672/foia",
    result_backend="db+postgresql://audit:***@pg:5432/foia",
    task_serializer="json",
    accept_content=["json"],            # reject pickle — untrusted payloads must not deserialize code
    task_acks_late=True,                # ack only after success so a crash re-delivers the request
    task_reject_on_worker_lost=True,    # requeue if the worker dies mid-task
    task_default_delivery_mode=2,       # persistent messages survive broker restart
    broker_transport_options={
        "confirm_publish": True,        # publisher confirms — admission fails loudly if the broker did not store it
        "visibility_timeout": 3600,     # redelivery window; safe because consumers are idempotent
    },
    worker_prefetch_multiplier=1,       # one in-flight task per worker — no starvation, fair priority
)

Expected behavior: stopping and restarting the broker between admission and consumption leaves the task intact and re-delivered; an admission call returns only after a publisher confirm, so a broker that failed to persist surfaces as an error at the endpoint rather than a silent loss.

3. Route by priority into isolated worker pools

Not every request carries equal legal weight. Expedited-processing requests — imminent litigation, statutory media deadlines, or vulnerable-population matters — need preemption without violating baseline FIFO fairness for standard submissions. Topology enforces this: distinct queues per tier, and high-risk operations (PII extraction, bulk generation, cross-jurisdiction transfers) isolated into dedicated pools so resource contention cannot couple a routine bulk export to a deadline-critical task.

python 
from kombu import Queue

# Priority is expressed as queue topology, not a single shared queue with a number field
app.conf.task_queues = (
    Queue("foia.high", routing_key="foia.high"),
    Queue("foia.default", routing_key="foia.default"),
)

# Isolated execution lanes enforce least-privilege per operation class
app.conf.task_routes = {
    "foia.extract_pii":       {"queue": "foia.high"},     # sensitive — dedicated, audited pool
    "foia.generate_bulk":     {"queue": "foia.default"},  # heavy I/O — must not block the high lane
    "foia.cross_jurisdiction": {"queue": "foia.high"},    # chain-of-custody critical
}


def select_queue(priority_score: int) -> str:
    # Threshold set by Priority Scoring Algorithms; >=80 escalates ahead of the FIFO default lane
    return "foia.high" if priority_score >= 80 else "foia.default"

Expected behavior: a request scored 80 or above is dispatched to foia.high and picked up ahead of the default backlog, while standard requests retain stable FIFO ordering in foia.default; the sensitive PII pool drains independently of the bulk-generation pool.

4. Retry transient faults without overrunning a statutory deadline

Retries must distinguish transient faults (network timeouts, brief database locks) from permanent ones (malformed payload, missing jurisdiction mapping). Transient faults retry with exponential backoff and jitter; permanent faults divert to the DLQ immediately. Crucially, a retry that would push the task past its statutory window must escalate for manual handling rather than retry into a deadline breach.

python 
from datetime import datetime, timezone
from celery import shared_task

PERMANENT_ERRORS = (KeyError, ValueError)  # bad payload / unmapped jurisdiction — no point retrying


@shared_task(
    bind=True,
    max_retries=5,
    acks_late=True,
    autoretry_for=(ConnectionError, TimeoutError),
    retry_backoff=True,       # exponential backoff
    retry_jitter=True,        # jitter avoids synchronized retry storms against a degraded dependency
    rate_limit="10/m",        # protects downstream stores from self-inflicted overload (NIST SP 800-53 SC-5)
)
def process_request(self, payload: dict) -> None:
    corr = payload["request_id"]
    try:
        deadline = datetime.fromisoformat(payload["statutory_deadline"])
        # 5 U.S.C. § 552(a)(6)(A)(i): never let backoff silently consume the response window
        if datetime.now(timezone.utc) >= deadline:
            logger.critical(json.dumps({"event": "DEADLINE_ESCALATION", "request_id": corr}))
            escalate_to_human.delay(payload)   # manual handling, not another retry
            return
        fulfill(payload)
        logger.info(json.dumps({"event": "PROCESSED", "request_id": corr}))
    except PERMANENT_ERRORS as exc:
        logger.error(json.dumps({"event": "DLQ_ROUTE", "request_id": corr, "error": str(exc)}))
        app.send_task("foia.dlq_handler", args=[payload, str(exc)])

Expected behavior: a connection blip triggers jittered retries and an eventual PROCESSED line; an unmapped jurisdiction emits DLQ_ROUTE on the first attempt; a task whose deadline has already elapsed emits DEADLINE_ESCALATION and is handed to a human instead of being retried into a breach.

Validation & Verification

Treat the queue’s compliance guarantees as testable invariants, not assumptions. Three checks catch most regressions:

  • Idempotency under redelivery. Admit the same canonical payload twice and assert exactly one TASK_ADMITTED and one IDEMPOTENCY_BLOCK audit line. In an integration test, kill a worker mid-task with task_acks_late=True enabled and assert the request is processed exactly once after redelivery.
  • Persistence across restart. Admit a task, restart the broker container, then start a consumer and assert the task still executes. With confirm_publish on, assert that an admission call against a stopped broker raises rather than returning success.
  • Trace continuity. Assert that the x-correlation-id minted at admission appears on every audit line the request produces — admission, dispatch, processing, and any DLQ or escalation event — so a single grep on the correlation ID reconstructs the full lifecycle.
python 
def test_duplicate_submission_blocks_second_clock(broker, seen):
    payload = {"request_id": "REQ-1", "submission_ts": "2026-06-27T12:00:00Z",
               "jurisdiction": "US-FED", "attachment_hashes": [], "priority_score": 10}
    admit_request(payload, broker, seen)
    admit_request(payload, broker, seen)            # identical re-submission
    assert broker.send_task.call_count == 1         # second admission published nothing

For ongoing assurance, run an automated DLQ scanner that emits a daily reconciliation report comparing queue-acceptance audit lines against the statutory tracking system, so zero unacknowledged requests cross a deadline unnoticed.

Troubleshooting & Edge Cases

  • Lost tasks after a worker crash. With early acknowledgment, a worker that dies mid-task takes the request with it and no clock is ever satisfied. Diagnosis: requests with a TASK_ADMITTED line but no terminal PROCESSED/DLQ_ROUTE event. Fix: enable task_acks_late=True and task_reject_on_worker_lost=True so the broker re-delivers; this is safe only because admission is idempotent.
  • Duplicate deadline clocks from re-submission. A requester re-sends the same form, or a redelivery is treated as new, and the system tracks two obligations for one request. Diagnosis: two TASK_ADMITTED lines sharing one idempotency key. Fix: check the canonical idempotency key against a durable store before publishing, and log IDEMPOTENCY_BLOCK on a match.
  • Retry storm consuming the statutory window. Naive retries against a degraded dependency multiply load and burn the FOIA clock without ever succeeding. Diagnosis: retry counts climbing while success rate falls, and tasks approaching their deadline mid-retry. Fix: pair jittered exponential backoff with the deadline guard that escalates to a human instead of retrying past the window, and rate-limit consumers to protect the downstream store (NIST SP 800-53 SC-5).
  • Priority inversion under high prefetch. A worker that prefetches a long batch of default-lane tasks ignores a newly arrived expedited request. Diagnosis: a high-tier request sitting while default tasks drain. Fix: set worker_prefetch_multiplier=1 and route tiers to separate queues so the high lane is never blocked behind buffered default work.
  • Litigation-hold conflict surfacing in the queue. A request under an active hold must not be fulfilled even if it processes cleanly. Diagnosis: a PROCESSED event for a record flagged in the hold registry. Fix: re-check the hold registry from Records Retention Scheduling immediately before fulfillment and treat any hold as a hard stop that diverts the request out of the pipeline.

Compliance Verification Checklist

FAQ

Why decouple admission from processing instead of handling a request inline?

Because the statutory obligation attaches at receipt, and the only thing that must happen synchronously is durably recording that receipt. Doing validation, classification, and fulfillment inline ties the agency’s ability to accept requests to the speed of its slowest downstream step, so a submission spike causes edge timeouts and the agency cannot even prove what arrived. A thin admission endpoint that stamps, hashes, and persists the request — then returns — keeps intake available under any load while the scalable consumer fleet does the heavy work behind a durable broker.

How does the queue guarantee a duplicate submission does not start two deadline clocks?

Compute a deterministic idempotency key over the canonical (sorted-key) form of the payload and check it against a durable store before publishing. A matching key is logged with an IDEMPOTENCY_BLOCK flag and nothing is published. Pairing that with persistent, late-acknowledged delivery means a worker crash re-delivers the same task rather than creating a new one, so neither a re-submission nor a redelivery can ever track two obligations for one request.

What happens when a retry would push a task past its statutory deadline?

It must not retry. The consumer compares the current time against the request’s statutory_deadline before each attempt; once the window has elapsed it emits a DEADLINE_ESCALATION audit line and hands the request to a human queue instead of scheduling another backoff. Retrying into a breach turns a recoverable transient fault into a defensible-only-with-difficulty compliance failure, so the deadline guard always wins over the retry policy.

Does this async queue replace a worker-internal asyncio pipeline?

No — they operate at different layers. The durable broker and Celery decide which work runs, in what priority order, and survive process restarts. The async batch processing pattern governs how a single worker executes its slice of that work efficiently against an I/O-bound document store. In production you typically run the async batch worker as the execution body inside a queue-dispatched job.

← Back to all public records automation topics