Implementing Async Batch Queues for Sales Data

Synchronous sales-data pipelines fracture under the weight of high-volume POS transactions, late EDI 810/850 submissions, and multi-tier promotional accruals: when the reconciliation engine waits on a real-time API handshake before it can match a deduction, latency compounds, accrual forecasts drift, and vendor disputes escalate. This page documents the exact procedure for decoupling ingestion from reconciliation compute with idempotent, partitioned async batch queues — deterministic dedup keys, version-pinned envelopes, decimal-exact settlement, and explicit dead-letter routing — so that a 500,000-line quarterly claim and a continuous trickle of daily store feeds share the same infrastructure without double-posting a single accrual. It is the implementation-level companion to the async batch processing cluster, which frames the queue topology and the financial guarantees the broker must preserve inside the broader data ingestion normalization pipelines discipline.

Sequence of an async batch queue settling one sales-data message exactly once A sequence diagram across six lifelines — producer, broker, idempotent worker, dedup store, ledger, and router. The producer enqueues a write-once envelope carrying a pinned agreement_version and correlation_id. The broker, partitioned by reconciliation_key so no single feed blocks others, delivers the message to a worker under at-least-once semantics. The worker claims the reconciliation_key in a dedup store with an atomic SET NX EX before any compute; if the key is already held the worker acknowledges the broker and drops the message with no accrual. On a first claim the worker settles the accrual in decimal arithmetic with ROUND_HALF_UP, then inside one all-or-nothing transaction posts the accrual to the ledger and marks the key processed in the dedup store. The router then sends the result to one of three terminal destinations: auto-post to POS and ERP sync when the claim matches the model, open a dispute when claim and model disagree, or route to the dead-letter queue once the retry budget is exhausted. EXACTLY-ONCE SETTLEMENT OF ONE MESSAGE · CLAIM BEFORE COMPUTE · ATOMIC COMMIT Producer ingestion edge Broker partitioned Worker idempotent Dedup store SET NX EX Ledger accrual post Router outcome enqueue envelope agreement_version pinned key: reconciliation_key · no head-of-line block deliver · at-least-once SET key NX EX (claim) first claim? → yes duplicate → ack & drop, no accrual settle() units × tier_rate · Decimal · ROUND_HALF_UP ONE TRANSACTION · ALL-OR-NOTHING post(accrual, version) mark_processed(key) route result POS & ERP sync auto-post · claim matches model Dispute queue claim ≠ model Dead-letter queue retry budget exhausted call return / ack

Prerequisites

Before wiring a queue between ingestion and the reconciliation engine, confirm the following are in place:

  • Normalized, canonical payloads. Raw vendor files must already be flattened and schema-validated by the CSV & EDI parsing workflows before anything is enqueued, with SKU hierarchies and deduction reason codes resolved against master data by field mapping strategies. Workers consume canonical records, never raw vendor terminology.
  • A version-pinned agreement contract. Each message carries a monotonic agreement_version sourced from agreement schema design — pinned per message, never resolved as “latest” at compute time.
  • A message broker and a dedup state store. Any of RabbitMQ, Apache Kafka, or AWS SQS for transport, plus a low-latency store (Redis or a Postgres unique index) to hold the deduplication index.
  • Python packages: pydantic>=2.6, redis>=5.0 (or the broker SDK), and the standard-library decimal module. Every monetary field uses decimal.Decimal; never float.
  • Access role: publish/consume rights on the broker and write access to the ledger and dedup store (typically the reconciliation_etl service role).

Step-by-step implementation

Step 1 — Model the immutable message envelope

Encode the write-once envelope as a Pydantic v2 model so that a float accrual, a missing version pin, or an unmapped vendor is rejected at enqueue rather than discovered mid-batch. The envelope captures what was submitted; mutable processing state is appended separately.

python
from decimal import Decimal
from enum import Enum
from pydantic import BaseModel, field_validator


class QueueTier(str, Enum):
    fast_path = "fast_path"
    compute_heavy = "compute_heavy"
    dead_letter = "dead_letter"


class SalesEnvelope(BaseModel):
    message_id: str            # ULID, time-ordered, write-once
    reconciliation_key: str    # sha256 of business identity
    correlation_id: str        # propagated ingestion -> ERP sync
    vendor_id: str             # FK to vendor master
    claim_period: str          # ISO fiscal window
    agreement_version: int     # pinned, never "latest"
    payload: dict              # canonical, UTC-anchored, normalized
    queue_tier: QueueTier = QueueTier.compute_heavy

    @field_validator("agreement_version")
    @classmethod
    def version_pinned(cls, v: int) -> int:
        if v < 1:
            raise ValueError("agreement_version must be pinned (>= 1)")
        return v

Validation check: load a fixture with agreement_version=0 and assert SalesEnvelope(**fixture) raises ValidationError. A message that cannot name the contract version that priced it must never reach a worker.

Step 2 — Derive a deterministic reconciliation key

At-least-once delivery is the norm for distributed brokers, but a reconciliation engine needs exactly-once semantics at the business level. Build the dedup key from a composite of immutable business fields so a replayed message hashes to the identical value.

python
import hashlib


def reconciliation_key(retailer_invoice: str, line_item_hash: str,
                       promo_tier: str, claim_period: str) -> str:
    blob = "|".join([retailer_invoice, line_item_hash,
                     promo_tier, claim_period])
    return hashlib.sha256(blob.encode()).hexdigest()

Validation check: compute the key twice from the same inputs and assert equality, then mutate promo_tier and assert the key changes. Determinism here is what lets the consumer silently discard a redelivery instead of double-accruing.

Step 3 — Partition the topology and apply backpressure

Partition queues by immutable keys (vendor_id, store_id, promotion_code, transaction_date) so a single stalled retailer POS feed cannot head-of-line-block every other vendor. Route by queue_tier so a slow tiered-accrual calculation never starves cheap validation work.

python
TIER_ROUTING = {
    QueueTier.fast_path: "recon.validate",       # stateless, highly parallel
    QueueTier.compute_heavy: "recon.accrue",     # rate-limited vs ERP
    QueueTier.dead_letter: "recon.dlq",
}

def publish(broker, env: SalesEnvelope) -> None:
    routing_key = f"{TIER_ROUTING[env.queue_tier]}.{env.vendor_id}"
    broker.publish(routing_key=routing_key,
                   body=env.model_dump_json(),
                   headers={"correlation_id": env.correlation_id})

Validation check: publish two envelopes for different vendors on a tier whose consumer is paused, then resume and assert both drain independently — neither blocks the other. Per-vendor partitioning must isolate a stalled feed.

Step 4 — Claim work idempotently before compute

Before any settlement runs, claim the reconciliation key with an atomic SET NX EX. A redelivered or replayed message finds the key already held and is acknowledged without re-posting.

python
def claim(redis, key: str, ttl: int = 86_400) -> bool:
    # returns True only on the first claim of this business identity
    return bool(redis.set(f"recon:claim:{key}", "1", nx=True, ex=ttl))


def consume(redis, env: SalesEnvelope) -> str:
    if not claim(redis, env.reconciliation_key):
        return "duplicate_skipped"   # ack and drop, no accrual
    return settle(env)

Validation check: call consume twice on the same envelope and assert the second returns "duplicate_skipped" with no second ledger write. Replay safety is the property the whole queue exists to guarantee.

Step 5 — Settle with decimal arithmetic, then commit atomically

Invoke the deterministic payout function with the pinned agreement_version, compute accruals under a fixed decimal context, and commit the ledger post and the dedup-key release marker in one transaction so a worker crash can never leave a half-posted entry.

python
from decimal import Decimal, ROUND_HALF_UP, getcontext

getcontext().prec = 28

def settle(env: SalesEnvelope) -> str:
    units = Decimal(str(env.payload["qualifying_units"]))
    rate = Decimal(str(env.payload["tier_rate"]))   # from pinned version
    accrual = (units * rate).quantize(Decimal("0.01"), ROUND_HALF_UP)
    with ledger.transaction() as txn:
        txn.post(reconciliation_key=env.reconciliation_key,
                 correlation_id=env.correlation_id,
                 agreement_version=env.agreement_version,
                 amount=accrual)
        txn.mark_processed(env.reconciliation_key)
    return "posted"

Validation check: assert settle returns a Decimal accrual (not float) and that interrupting the transaction before commit leaves no ledger row. The post and the processed marker must be all-or-nothing.

Step 6 — Categorize failures and route to the dead-letter queue

Queue-driven pipelines fail gracefully only when errors are classified and routed, not blindly retried. Segment failures into three classes and route each deterministically.

python
class TransientError(Exception): ...     # broker throttle, DB lock, timeout
class StructuralError(Exception): ...     # schema mismatch, bad date format
class BusinessError(Exception): ...       # invalid promo code, negative sales

MAX_ATTEMPTS = 5

def route_failure(broker, env, exc, attempt: int) -> str:
    if isinstance(exc, TransientError) and attempt < MAX_ATTEMPTS:
        backoff = min(2 ** attempt, 60) + random.uniform(0, attempt)
        broker.requeue(env, delay=backoff)          # exp backoff + jitter
        return "retried"
    env.queue_tier = QueueTier.dead_letter
    broker.publish_dlq(env, reason=type(exc).__name__)
    return "dead_lettered"

Validation check: raise a StructuralError and assert the message lands in the DLQ on the first attempt (no retry), while a TransientError is requeued with a growing delay. Only retryable classes should consume retry budget.

Common failure modes and fixes

  1. Floating-point drift in accruals. Computing rebate amounts with float lets rounding error compound across millions of line items and breaks ledger ties. Keep every monetary value in decimal.Decimal under a fixed context and quantize explicitly with ROUND_HALF_UP — never let a float enter the chain.
  2. Double-posting on redelivery. At-least-once brokers redeliver after a consumer timeout; without the Step 4 claim, the second delivery posts a duplicate accrual. Claim the deterministic key with SET NX EX before compute and commit the processed marker in the same transaction as the ledger post.
  3. Head-of-line blocking on a stalled feed. A single queue lets one retailer’s slow POS feed freeze every other vendor. Partition by vendor_id / store_id and split fast-path validation from compute-heavy accrual so latency stays local to the affected partition.
  4. Version drift between enqueue and compute. Resolving the agreement as “latest” at settlement time means a mid-batch contract edit reprices half the run. Pin agreement_version on the envelope at enqueue (Step 1) and invoke the payout function with that frozen version.
  5. Poison messages retried forever. A StructuralError retried with backoff never succeeds and saturates the worker pool. Classify the failure (Step 6) and dead-letter structural and business errors immediately, reserving retries for transient faults — then surface DLQ depth on a dashboard so unmatched records are adjudicated through fallback routing logic rather than silently lost.

Operational checklist

Frequently asked questions

What happens when a retroactive tier is crossed mid-cycle? The envelope pins the agreement_version at enqueue, so messages already in flight settle against the contract version they were submitted under. A retroactive recalculation is a new run with a new pinned version, and because every accrual carries its correlation_id and version, the recomputation is reconstructable rather than a silent overwrite. The tier math itself stays with payout structure modeling.

Does the queue recompute promotional eligibility? No. Qualification is owned upstream by the claim validation rule engine, and workers consume already-qualified records. Keeping business rules out of the consumer avoids version drift and double-evaluation.

How do I guarantee exactly-once accrual on an at-least-once broker? You do not chase exactly-once delivery; you enforce exactly-once effect. Derive a deterministic reconciliation key, claim it atomically before compute, and commit the ledger post and the processed marker in one transaction — replays then find the key claimed and drop harmlessly.

Where do validated journal entries go after settlement? Posted accruals synchronize to the general ledger through POS & ERP sync patterns on the retailer’s fiscal cutoff, carrying the same correlation_id so the entry is traceable back to the originating sales line.

Back to