Async Approval Queue Design for High-Throughput Reconciliation Exception Handling

The async approval queue is the non-blocking transport layer that carries unmatched exceptions from the matching engine to the reviewers who resolve them. Where a state machine governs what may legally happen to an exception, this queue governs how fast and how safely exceptions move — it is a throughput problem, not a lifecycle problem. It sits at the exact seam in the reconciliation pipeline where the deterministic matching cascade has produced a backlog of UNMATCHED and PARTIAL decisions faster than humans can act on them, and its job is to absorb that burst without dropping work, without double-delivering an exception to two reviewers, and without letting an unbounded in-memory backlog exhaust the process. A production-grade approval queue is a bounded, back-pressured, at-least-once delivery pipeline whose consumers are idempotent by construction, and whose shutdown is graceful enough that a deploy in the middle of a financial close never abandons an in-flight sign-off. This page extends Exception Routing & Human-in-the-Loop Workflows with the concurrency mechanics that let the review tier scale horizontally under peak reconciliation load.

Problem statement: throughput, not lifecycle

It is tempting to conflate this queue with the state-managed control plane described in Manual Review Queue Design, but the two solve orthogonal problems and are designed to compose. The manual review queue owns the finite state machine — the legal transitions from PENDING through APPROVED or REJECTED to RESOLVED, the optimistic-concurrency guards, and the segregation-of-duties rules. The async approval queue owns the delivery substrate underneath that state machine: the producer/consumer topology, the bounded buffer that applies backpressure, the durable dequeue that guarantees single delivery, and the drain sequence that makes shutdown safe. In practice one wraps the other. The delivery layer pulls a durable record, hands it to a handler, and the handler invokes the state machine’s transition; the queue never decides whether a transition is legal, and the state machine never decides how many exceptions are dispatched per second.

Keeping these concerns separate matters because they fail differently and scale differently. A state-machine bug produces an illegal transition — a correctness fault caught by a set-membership test. A delivery-layer bug produces lost work, duplicated work, or a saturated process — a throughput and liveness fault caught by queue-depth and consumer-lag metrics. Conflating them produces a system where you cannot reason about either. The design goal here is a queue that treats every message as at-least-once and therefore demands idempotent handlers, so that a redelivery after a crash re-enters a terminal state as a no-op rather than a second sign-off.

Prerequisites: upstream pipeline state

The approval queue never ingests raw feed data or unclassified records. By the time an exception is enqueued, the upstream stages must already have produced:

• A durable, matched record with a stable decision. Every message references a persisted row carrying a MATCHED, PARTIAL, or UNMATCHED decision from the cascade in Transaction Matching Algorithms & Logic. The queue transports exceptions; it does not re-match them.
• A globally unique idempotency_key and a trace_id, both assigned at ingestion, so the same exception published twice — by a producer retry or an upstream replay — collapses to one logical unit of work at the consumer.
• A partition_key (typically ledger_entity or counterparty) so that multiple breaks against the same entity can be routed to the same consumer and resolved in order, rather than racing across the pool.
• Precomputed variance metrics in Decimal, so the handler consumes exact monetary inputs and never recomputes materiality under floating-point drift.

If any prerequisite is missing the message is rejected to a dead-letter branch with a POISON_MESSAGE code rather than dispatched, because a malformed message that cannot be parsed will fail identically on every redelivery and must not block the consumer group.

Producer/consumer topology and backpressure

The core topology is a classic asyncio producer/consumer graph with one non-negotiable property: the buffer between producers and consumers is bounded. Producers read durable exceptions from the store and await queue.put(...); consumers await queue.get(...), invoke the handler, and acknowledge. The bound is what turns an unbounded firehose into a self-regulating system. When reviewers or their downstream handlers slow down, the bounded asyncio.Queue fills to maxsize, every subsequent put suspends the producer coroutine, and the producer stops pulling new rows from the durable store. Ingestion throttles itself to exactly the rate the review tier can sustain — no unbounded memory growth, no dropped work, no explicit rate limiter to tune.

The formal behaviour is that of a bounded blocking channel: throughput settles at min(producer_rate, consumer_rate), and the queue depth oscillates around the point where the two rates equalise. The buffer’s purpose is not to store a large backlog — the durable store does that — but to keep exactly enough work in memory that consumers never idle waiting for a put, while never buffering so much that a crash loses a large window of in-flight messages. A maxsize on the order of a few multiples of the worker count is usually correct; a maxsize of thousands is a smell that indicates the durable store is being drained into memory rather than streamed.

The complexity of the hot path is dominated by two operations, not by the dispatch logic: the locked dequeue against the durable store (one indexed row lock) and the acknowledge on completion. Both are O(1) per message, so the queue’s ceiling is set by store round-trip latency and by how much CPU each handler burns, which is why CPU-heavy work must be pushed off the event loop rather than run inline.

Durable dequeue: SKIP LOCKED versus Redis Streams

The in-memory asyncio.Queue is a buffer, never the source of truth. If the process dies, everything still in that buffer must be recoverable, which means the authoritative backlog lives in a durable store and the producers stream from it. Two implementations dominate.

PostgreSQL FOR UPDATE SKIP LOCKED. A producer runs SELECT ... FROM approval_queue WHERE state = 'PENDING' ORDER BY priority, created_at FOR UPDATE SKIP LOCKED LIMIT :prefetch. The SKIP LOCKED clause is what makes horizontal scaling safe: two producers running the identical query never return the same row, because rows already locked by a competing transaction are silently skipped rather than blocked on. The row stays locked for the life of the transaction — that lock is the lease — and committing after a successful handler marks the row done. If the worker crashes, the transaction rolls back, the lock releases, and the row is immediately eligible for another producer. This gives strong single-delivery semantics with no extra infrastructure, at the cost of holding a database transaction open for the duration of processing. The detailed transaction shape is covered in SKIP LOCKED queue consumption in Postgres.

Redis Streams consumer groups. XREADGROUP delivers each entry to exactly one consumer within a group and records it in that consumer’s Pending Entries List (PEL) until an explicit XACK. Unacknowledged entries can be reclaimed by another consumer with XAUTOCLAIM after a minimum idle time, which is how a crashed consumer’s in-flight work is recovered. Streams scale dequeue throughput past what a single Postgres instance sustains and decouple the buffer from the transactional ledger, at the cost of a second system to operate and an at-least-once contract that pushes deduplication entirely onto the handler.

Both are at-least-once, never exactly-once. SKIP LOCKED can redeliver if a worker commits its side effects but crashes before committing the queue transaction; Streams redeliver on any un-acked reclaim. True exactly-once delivery across a process boundary is unattainable without a distributed transaction spanning the queue and every side effect, which is impractical at reconciliation volumes. The engineering answer is not to chase exactly-once delivery but to guarantee exactly-once effect through idempotent handlers keyed on idempotency_key, so a redelivery is provably harmless.

Production-grade Python implementation

The worker pool below drains a bounded asyncio.Queue, pushes the CPU-bound decision off the event loop with asyncio.to_thread, emits a structured audit record on every decision, and routes unprocessable messages to a dead-letter branch instead of crashing the consumer. Monetary values are Decimal and timestamps are timezone-aware UTC throughout.

import asyncio
import hashlib
import logging
import uuid
from dataclasses import dataclass, field
from decimal import Decimal, ROUND_HALF_UP
from datetime import datetime, timezone
from typing import AsyncIterator

logger = logging.getLogger("reconciliation.approval_queue")


@dataclass(frozen=True)
class ApprovalMessage:
    idempotency_key: str
    ledger_entity: str
    variance_amount: Decimal
    match_decision: str            # MATCHED | PARTIAL | UNMATCHED
    trace_id: str = field(default_factory=lambda: str(uuid.uuid4()))

    @property
    def source_hash(self) -> str:
        material = (
            f"{self.idempotency_key}:{self.ledger_entity}:"
            f"{self.variance_amount}:{self.match_decision}"
        )
        return hashlib.sha256(material.encode("utf-8")).hexdigest()


class PoisonMessage(Exception):
    """Unprocessable payload: will fail identically on every redelivery."""


async def _decide(msg: ApprovalMessage) -> str:
    # CPU-bound work (hashing, rule evaluation) runs OFF the event loop so a
    # slow decision never stalls dequeue for every other worker.
    def _cpu() -> str:
        amount = Decimal(str(msg.variance_amount)).quantize(
            Decimal("0.01"), rounding=ROUND_HALF_UP
        )
        if msg.match_decision not in {"MATCHED", "PARTIAL", "UNMATCHED"}:
            raise PoisonMessage(f"unknown decision {msg.match_decision!r}")
        return "AUTO_RESOLVE" if amount == Decimal("0.00") else "ROUTE_REVIEW"

    return await asyncio.to_thread(_cpu)


async def worker(
    name: str,
    queue: "asyncio.Queue[ApprovalMessage]",
    dlq: "asyncio.Queue[ApprovalMessage]",
    ack,                                   # async callable: durable acknowledge
    seen: set[str],                        # idempotency ledger (bounded / TTL in prod)
) -> None:
    while True:
        msg = await queue.get()
        try:
            if msg.idempotency_key in seen:
                # At-least-once delivery: a redelivery is a harmless no-op.
                logger.info(
                    "approval_duplicate_skipped",
                    extra={"trace_id": msg.trace_id,
                           "source_hash": msg.source_hash,
                           "match_decision": msg.match_decision},
                )
                continue
            decision = await _decide(msg)
            seen.add(msg.idempotency_key)
            await ack(msg)                 # commit the durable lease AFTER effect
            logger.info(
                "approval_decision",
                extra={
                    "trace_id": msg.trace_id,
                    "source_hash": msg.source_hash,
                    "match_decision": msg.match_decision,
                    "worker": name,
                    "route": decision,
                    "variance_amount": str(msg.variance_amount),
                    "ts": datetime.now(timezone.utc).isoformat(),
                },
            )
        except PoisonMessage as exc:
            await dlq.put(msg)
            logger.warning(
                "approval_poison_message",
                extra={"trace_id": msg.trace_id,
                       "source_hash": msg.source_hash,
                       "match_decision": msg.match_decision,
                       "error": str(exc)},
            )
        except asyncio.CancelledError:
            queue.put_nowait(msg)          # requeue on drain; nothing is dropped
            raise
        finally:
            queue.task_done()


async def run_pool(
    source: AsyncIterator[ApprovalMessage],
    ack,
    concurrency: int = 8,
    queue_maxsize: int = 64,
    drain_timeout_s: float = 30.0,
) -> None:
    queue: "asyncio.Queue[ApprovalMessage]" = asyncio.Queue(maxsize=queue_maxsize)
    dlq: "asyncio.Queue[ApprovalMessage]" = asyncio.Queue()
    seen: set[str] = set()
    workers = [
        asyncio.create_task(worker(f"w-{i}", queue, dlq, ack, seen))
        for i in range(concurrency)
    ]
    try:
        async for msg in source:
            await queue.put(msg)           # producer blocks when the pool saturates
        # Graceful drain: let in-flight work finish before tearing down.
        await asyncio.wait_for(queue.join(), timeout=drain_timeout_s)
    finally:
        for w in workers:
            w.cancel()
        await asyncio.gather(*workers, return_exceptions=True)

The seen set is a schematic stand-in for a durable idempotency ledger — in production this is a Redis set with a TTL or a unique-constrained table, so that the guarantee survives a restart. The critical ordering is that the durable ack happens after the effect is recorded but the idempotency key is registered before the ack, so a crash between the two produces a redelivery that the seen check absorbs rather than a lost message. Note that await queue.put(msg) in the producer loop is the entire backpressure mechanism: when the pool is saturated the coroutine suspends there, and the durable source is not polled again until a worker frees a slot.

Configuration reference

Every parameter below is version-controlled and timestamped, and a change to any of them is itself an auditable event. Treat the defaults as conservative starting points and widen them only with evidence from replayed load.

The interlock to watch is prefetch against queue_maxsize: a producer must never lease more rows than the bounded queue can hold, or the surplus sits in producer memory outside the backpressure contract. Keep prefetch ≤ queue_maxsize and size lease_timeout_ms comfortably above the 99th-percentile handler duration so healthy-but-slow workers are not reclaimed out from under themselves.

Multi-dimensional validation

A single check is never sufficient to certify that the queue is healthy; validation composes several independent dimensions, and a message advances only when all of them hold — the logical intersection of gates, never an average that lets one mask another.

1. Structural validity. The payload deserialises, carries a non-empty idempotency_key, and its match_decision is one of the known enum values. A structural failure is deterministic across redeliveries and routes straight to the dead-letter branch as POISON_MESSAGE.
2. Monetary precision. variance_amount parses as Decimal and quantises cleanly to two places. A float that arrived over the wire is rejected before it can misclassify materiality by a sub-cent margin.
3. Ordering integrity. Messages sharing a partition_key are dispatched to the same consumer so per-entity ordering is preserved; a message whose partition is being drained is held rather than reordered.
4. Idempotency. The idempotency_key is checked against the durable ledger before any effect, so a redelivery is provably a no-op rather than a second decision.

Because the gates are independent, a message can be structurally valid and monetarily exact yet fail the idempotency check — and that combination is exactly the redelivery case the queue must swallow silently. The routing decision that feeds these gates comes from Threshold-Based Routing Logic, which the queue consumes as an immutable input rather than recomputing.

Async / high-throughput patterns

The patterns that keep this queue fast are the ones that keep the event loop unblocked and the buffer honest. Four are load-bearing.

Keep CPU off the loop. Any synchronous, CPU-bound step — hashing a payload, evaluating a rule tree, quantising a batch of Decimal values — must run under asyncio.to_thread (or a process pool for genuinely heavy work). A single inline CPU burst stalls every coroutine on that loop, collapsing concurrency to one. The implementation above isolates all such work inside _decide.

Let the bound do the throttling. The bounded queue is the rate limiter. There is no separate token bucket to tune, because await queue.put() suspends the producer the instant the pool falls behind. This makes the system self-correcting: a slow downstream naturally slows ingestion, and recovery naturally speeds it back up, with no oscillation to damp.

Partition for ordering. When several breaks reference the same ledger_entity, they must resolve in a consistent order or a later adjustment can overwrite an earlier one. Routing by partition_key — a per-partition sub-queue or a consistent hash onto workers — preserves per-key ordering while still parallelising across keys. Global ordering is neither needed nor affordable; per-partition ordering is both.

Drain, don’t drop. Graceful shutdown is a first-class requirement in a system that deploys during a close. On SIGTERM the producer stops pulling new work, queue.join() waits up to drain_timeout_s for in-flight messages to acknowledge, and only then are workers cancelled — and a cancelled worker requeues its current message rather than abandoning it. The child page Building an asyncio approval queue with backpressure walks through this drain sequence and the signal handling end to end. Messages that survive the drain window and cannot be resolved fall through to Dead-Letter Queue Remediation, where they are inspected and replayed rather than lost.

Failure modes and remediation

Each named code is emitted with the trace_id and source_hash so it can be triaged from the audit stream rather than reproduced by hand.

The unifying principle mirrors the state machine’s: no failure silently resolves a message. QUEUE_SATURATED throttles, LEASE_EXPIRED reclaims, POISON_MESSAGE diverts, and CONSUMER_LAG scales — none of them drops work or double-commits an effect. Ordering-sensitive throughput at very high volume, and the way approvals are batched once they reach a reviewer, is treated separately in Batch Approval UI Patterns.

Compliance and audit-trail requirements

The delivery layer is inside the audit boundary even though it makes no lifecycle decisions, because when and to whom an exception was dispatched is itself evidence. Three requirements apply. First, every decision emits a structured record carrying trace_id, source_hash, and match_decision, so the movement of each exception through the queue is replayable from the log stream independently of the ledger — this satisfies SOX Section 404 and PCI-DSS Requirement 10 traceability. Second, at-least-once delivery is reconciled to exactly-once effect through the durable idempotency ledger, so an auditor can prove that a redelivery storm or a mid-close deploy produced no duplicate sign-off and no duplicate posting. Third, the dead-letter branch is durable and reviewable, not a silent discard: a POISON_MESSAGE is retained with its original payload and diagnostic context so a regulator can see that an unprocessable exception was quarantined and remediated rather than dropped.

By treating the async approval queue as a bounded, back-pressured, idempotent delivery substrate — and delegating every lifecycle decision to the state machine it feeds — reconciliation teams get a review tier that scales horizontally under peak load, survives deploys mid-close without abandoning work, and preserves an audit trail a regulator can replay message by message.

Related

• Manual Review Queue Design
• Dead-Letter Queue Remediation
• Batch Approval UI Patterns
• Building an asyncio Approval Queue with Backpressure

Part of Exception Routing & Human-in-the-Loop Workflows.