Asynchronous Batch Adjudication Workflows

Synchronous, blocking adjudication collapses the moment a Pharmacy Benefit Manager (PBM) has to price millions of daily claims against rate-limited eligibility, formulary, and MAC pricing services. This page covers the specific sub-problem of decoupling raw claim ingestion from the pricing and drug utilization review (DUR) engines using event-driven workers, so that point-of-sale latency, downstream HTTP 429 throttling, and formulary refresh windows never stall the whole pipeline. It is the execution layer of the broader Claims Ingestion & NCPDP Parsing domain: once a payload has been parsed and structurally validated upstream, this workflow governs how it is queued, dispatched to a bounded worker pool, adjudicated with idempotent state transitions, and reconciled against 835 remittance. The goal is deterministic throughput with an immutable audit trail, not raw speed at the expense of correctness.

Prerequisites #

Before this workflow runs, the following upstream dependencies and environment assumptions must hold:

• Parsed, structurally valid payloads. Segment-level parsing from NCPDP D.0 Message Parsing Strategies and severity classification from Schema Validation & Error Categorization must run before enqueue. Malformed B1/B2 transactions never reach a worker; they are diverted to a dead-letter queue (DLQ) at validation time.
• Resolved drug taxonomy. A canonical 11-digit NDC (407-D7 Product/Service ID) mapped to a 14-digit GPI via the NDC to GPI Crosswalk Automation subsystem, so pricing lookups key on a stable therapeutic identifier.
• A broker and result backend. Redis 7.x (or RabbitMQ 3.13.x) as the transport, plus a durable result store. Examples below use Celery 5.4 with redis-py 5.x and Pydantic 2.x.
• Bounded downstream clients. Eligibility, formulary, and pricing endpoints are governed by the token-bucket / sliding-window limiter described in PBM API Sync & Rate Limiting, and wrapped by the circuit breakers from Fallback Routing Logic Design.
• A versioned formulary snapshot. Every batch pins a formulary snapshot ID at dispatch time so that tier and copay decisions are reproducible under payer audit, even after the live formulary changes mid-batch.
• PHI handling contract. The security controls from Security & Compliance Boundaries for Claims Data apply to every task: strip 302-C2 (Cardholder ID) and 310-CA (Patient First Name) from the payload immediately after routing keys are derived, and never log raw claim bytes.

Queue Topology & Dispatch Rules #

High-volume adjudication demands deliberate partitioning so that workload classes do not contend for the same broker memory or database connection pool. Claims are routed to dedicated queues keyed on plan sponsor, pharmacy NPI, priority tier, and reprocessing flags. Isolated queues give operations teams granular backpressure control — a retroactive MAC repricing run can be throttled independently of live point-of-sale traffic during a CMS reporting window. The Configuring async queues for high-volume claims ingestion guide covers the concrete broker sizing; the routing contract that governs which queue a claim lands in is summarized below.

Adjudication Disposition & Reject-Code Mapping #

The core algorithm each worker runs is a deterministic classification: given a validated claim and the downstream responses, decide whether the claim is PAID, terminally REJECTED, RETRYABLE (transient dependency failure), or PENDED (awaiting prior authorization). The disposition drives both the outbound NCPDP 511-FB Reject Code returned to the switch and the queue the message moves to next. Mapping failure classes to concrete reject codes keeps handler logic auditable and prevents silent misrouting.

Quantity and refill checks (76, 79) reuse the same logic documented under Quantity Limit & Days Supply Validation; step-therapy and PA pends (608, 75) are produced by Step Therapy & Prior Auth Trigger Rules. Transient failures (RETRYABLE) are distinguished from terminal rejects because retrying a 70 or 79 wastes downstream budget and can produce duplicate rejects at the pharmacy — the distinction is exactly the one drawn in Handling PBM 404 and 503 errors in adjudication scripts.

Reference Python Implementation #

The Celery task below validates the payload with Pydantic, derives an idempotency key, strips PHI immediately after routing keys are extracted, enforces a Redis sliding-window rate limit on the formulary call, and computes patient responsibility with decimal.Decimal so no penny is lost to binary float rounding. Formulary tier is resolved against a pinned snapshot ID, and every log line references NCPDP field codes rather than raw claim bytes.

import logging
import time
import uuid
from decimal import Decimal, ROUND_HALF_UP
from typing import Any

import redis
import requests
from celery import Celery
from pydantic import BaseModel, Field, ValidationError

app = Celery("pbm_adjudication", broker="redis://localhost:6379/0",
             backend="redis://localhost:6379/1")
redis_client = redis.Redis(host="localhost", port=6379, db=2)
logger = logging.getLogger(__name__)

CENTS = Decimal("0.01")


class NCPDPD0Claim(BaseModel):
    bin: str = Field(min_length=6, max_length=6)   # 101-A1 BIN Number (routing)
    pcn: str                                        # 104-A4 Processor Control Number
    group_id: str                                   # 301-C1 Group ID
    cardholder_id: str                              # 302-C2 Cardholder ID (PHI)
    ndc: str = Field(min_length=11, max_length=11)  # 407-D7 Product/Service ID (NDC-11)
    gpi: str = Field(min_length=14, max_length=14)  # resolved via NDC->GPI crosswalk
    quantity: Decimal                               # 442-E7 Quantity Dispensed
    days_supply: int                                # 405-D5 Days Supply
    service_date: str                               # 401-D1 Date of Service (CCYYMMDD)
    pricing_tier: str


def routing_key(claim: NCPDPD0Claim) -> str:
    # Idempotency key derived from routing fields BEFORE PHI is discarded.
    # 302-C2 is hashed, never stored or logged in the clear.
    ch = str(hash(claim.cardholder_id) & 0xFFFFFFFF)
    return f"adj:{claim.bin}:{claim.pcn}:{ch}:{claim.ndc}:{claim.service_date}"


class SlidingWindowLimiter:
    """Redis sorted-set sliding window for a rate-limited downstream endpoint."""

    def __init__(self, conn: redis.Redis, key: str, max_calls: int, window: int):
        self.redis, self.key, self.max_calls, self.window = conn, key, max_calls, window

    def acquire(self) -> bool:
        now = time.time()
        pipe = self.redis.pipeline()
        pipe.zremrangebyscore(self.key, 0, now - self.window)  # evict expired
        pipe.zcard(self.key)                                   # count live calls
        pipe.expire(self.key, self.window)
        _, count, _ = pipe.execute()
        if count >= self.max_calls:
            return False
        self.redis.zadd(self.key, {f"{now}:{uuid.uuid4().hex}": now})
        return True


@app.task(bind=True, max_retries=3, acks_late=True)
def adjudicate_claim_batch(self, payloads: list[dict[str, Any]],
                           formulary_snapshot_id: str) -> dict[str, Any]:
    results = {"paid": 0, "rejected": 0, "pended": 0, "failed": 0}
    limiter = SlidingWindowLimiter(redis_client, "rl:formulary_api", max_calls=50, window=1)

    for payload in payloads:
        try:
            claim = NCPDPD0Claim(**payload)
        except ValidationError as ve:
            # Never echo the payload; log field paths only (no PHI, no raw bytes).
            logger.warning("schema_reject fields=%s", [e["loc"] for e in ve.errors()])
            results["failed"] += 1
            continue

        idem = routing_key(claim)
        del payload  # drop the raw dict (carried 302-C2 / 310-CA) once keyed

        # Idempotency guard: a completed adjudication is never reprocessed.
        if redis_client.get(f"{idem}:state") == b"ADJUDICATED":
            logger.info("duplicate_skip key=%s", idem)
            continue

        while not limiter.acquire():          # respect downstream 429 budget
            time.sleep(0.05)

        try:
            resp = requests.get(
                "https://api.pbm.internal/v1/formulary",
                params={"gpi": claim.gpi, "tier": claim.pricing_tier,
                        "snapshot": formulary_snapshot_id},   # pinned, audit-reproducible
                timeout=5,
            )
            resp.raise_for_status()
        except requests.exceptions.RequestException as err:
            logger.warning("transient_downstream ndc_suffix=%s", claim.ndc[-4:])
            raise self.retry(exc=err, countdown=2 ** self.request.retries)

        tier = resp.json()
        disposition, reject_code, responsibility = adjudicate(claim, tier)

        redis_client.setex(f"{idem}:state", 86400, "ADJUDICATED")
        redis_client.setex(f"{idem}:result", 86400,
                           f"{disposition}:{reject_code or ''}:{responsibility}")
        results[disposition.lower()] = results.get(disposition.lower(), 0) + 1

    return results


def adjudicate(claim: NCPDPD0Claim, tier: dict) -> tuple[str, str | None, Decimal]:
    if not tier.get("covered", False):
        return "REJECTED", "70", Decimal("0.00")           # Product/Service Not Covered
    if tier.get("pa_required", False):
        return "PENDED", "75", Decimal("0.00")             # Prior Authorization Required
    # Decimal end-to-end: MAC price minus plan copay, floored at zero, penny-exact.
    mac = Decimal(str(tier["mac_price"]))
    copay = Decimal(str(tier["copay_amount"]))
    responsibility = max(Decimal("0.00"), (mac - copay)).quantize(CENTS, ROUND_HALF_UP)
    return "PAID", None, responsibility

Execution Flow #

Each claim moves through a bounded, observable state machine: enqueued → validated → rate-gated downstream call → adjudicated → persisted, with retry and DLQ edges for transient and terminal failures respectively. The diagram below traces a single claim from the producer through the concurrency-limited worker pool to the result store.

Figure: Asynchronous batch adjudication architecture — a producer enqueues claims to a Redis broker, a concurrency-limited Celery worker pool makes rate-limited downstream calls (absorbing HTTP 429 with backoff), and adjudicated results land in the result store.

Figure: Single-claim adjudication state machine — ENQUEUED → VALIDATED → RATE-GATED → ADJUDICATED, terminating as PAID, REJECTED (70/76/79), or PENDED (75/608), with a transient RETRYABLE backoff loop and a terminal adj.dlq edge once max_retries is exhausted.

Engineering Constraints & Known Failure Modes #

Batch adjudication fails in specific, recurring ways, and each has a deterministic mitigation:

• Reversal races (B2 vs. B1). A B2 reversal can arrive while the original B1 is still in flight. Serialize reversals per idempotency key (the adj.reversal queue) and gate them on the :state marker so a reversal cannot complete against a claim that has not yet reached ADJUDICATED.
• PA pend races. A claim PENDED with reject 75 must not be re-adjudicated as PAID if the PA approval and a retry land concurrently. The :state idempotency marker and a separate PA-hold key prevent double adjudication.
• NDC gaps and inactive taxonomy. An NDC that resolves to a discontinued or unmapped GPI must reject before the pricing call — otherwise it triggers an erroneous MAC lookup. This is enforced upstream in the crosswalk layer; the worker treats an unmapped GPI as a terminal 70.
• Snapshot drift. If the live formulary changes mid-batch, workers pinning different snapshot IDs would produce inconsistent tiering. Pin one formulary_snapshot_id at dispatch for the whole batch so results are reproducible under audit.
• Poison messages. A payload that repeatedly raises after max_retries must land in adj.dlq, not loop forever. acks_late=True plus a bounded retry count guarantees the message is acknowledged only after terminal disposition or DLQ routing.
• Downstream cascade. A failing pricing endpoint under retry storms amplifies load. Wrap the client in the circuit breaker from Fallback Routing Logic Design so an open breaker short-circuits to RETRYABLE instead of hammering a dead service.

Performance & Correctness Tuning #

Throughput and financial accuracy are tuned together, never traded off:

• Idempotency keys. Derive the key from routing fields (101-A1 BIN, 104-A4 PCN, hashed 302-C2, 407-D7 NDC, 401-D1 Date of Service) so retries, duplicate submissions, and worker restarts converge on one adjudication. The :state marker carries a 24-hour TTL matching the claim-day window.
• Decimal precision. Every monetary field — MAC price, copay, deductible accumulator, patient responsibility — uses decimal.Decimal with explicit quantize(Decimal("0.01"), ROUND_HALF_UP). Binary float on copay math drifts pennies that fail 835 reconciliation.
• Snapshot caching. Cache the pinned formulary snapshot per batch in worker memory keyed by formulary_snapshot_id; a batch of 50k claims should hit the pricing API for distinct GPIs only, not per claim.
• Backpressure and prefetch. Tune Celery prefetch_multiplier low for latency-bound retail queues and high for throughput-bound mail-order, so a slow worker does not hoard adj.pos.retail messages. Scale concurrency off live Redis queue depth.
• SLA envelopes. Retail point-of-sale claims carry a hard latency SLA; reprice batches do not. Isolating them by queue lets you drain the SLA-bound queue first while deferring adj.reprice to off-peak.
• Reconciliation. After each cycle, reconcile the persisted :result states against inbound 835 remittance advice to flag MAC or copay discrepancies before they become payer disputes.

In this section #

• Configuring async queues for high-volume claims ingestion — broker sizing, queue partitioning, prefetch and concurrency tuning for sustained high-volume ingestion.

Related #

• Claims Ingestion & NCPDP Parsing — parent domain: how raw claims are received and parsed before adjudication.
• Schema Validation & Error Categorization — the validation gate that classifies payloads before they are enqueued.
• PBM API Sync & Rate Limiting — token-bucket and sliding-window control for the downstream calls workers make.
• Fallback Routing Logic Design — circuit breakers and fallback tiers that keep the worker pool resilient.
• Tier Mapping & Copay Calculation Logic — the pricing and copay rules invoked during adjudication.

← Back to Claims Ingestion & NCPDP Parsing