Async Batch Processing Workflows

When a culinary director edits a yield factor or procurement lands a new vendor rate, the cost of every affected plate across every location has to be recomputed — and if that recomputation runs inline with the request that triggered it, the whole estate feels it. Synchronous rollups block HTTP handlers, stall menu-engineering dashboards, and open race windows on the inventory ledger while thousands of SKUs are recosted under a held lock. This guide, part of the Data Ingestion & Recipe Parsing Workflows framework, isolates one sub-problem: how to move food-cost rollups off the request path and onto a durable queue so they execute as discrete, replayable batch tasks that never block the systems taking orders. It defines the task data contract, argues for a message broker over an inline call, and walks the three-phase build — enqueue, validate, publish — that turns a pricing change into an auditable set of cost deltas. The concrete Celery mechanics are covered in the companion walkthrough on implementing Celery for async menu syncs; here we define the framework those tasks run inside.

Concept Definition and Data Contract

An async batch workflow, in this domain, is a system where a state change to costing inputs is captured as a message rather than executed as a call. The producer — a bulk import job, a vendor-rate webhook, a chef editing a spec sheet — does not run the rollup. It serializes a description of the work, hands it to a broker, and returns immediately. One or more stateless workers later consume that message and perform the recalculation on their own schedule. The two sides are coupled only by the message schema, and that schema is the contract that keeps the pipeline honest.

The input contract is a single recalculation request:

• recipe_id — the canonical recipe whose cost must be re-derived; it maps directly to the root of a recipe BOM roll-up so the worker knows which ingredient tree to walk.
• location_ids — the units affected by the change, because a regional vendor contract touches some kitchens and not others.
• effective_at — a UTC timestamp marking when the pricing change takes effect; it is part of the identity of the work, not just metadata.
• pricing_snapshot — the as-purchased costs in play, each carried as an exact Decimal, never a float, keyed by SKU.
• trigger — an enum (vendor_rate, yield_edit, spec_change, bulk_import) used for routing and audit.
• signature — an HMAC over the canonical payload so a worker can reject spoofed or corrupted messages before acting on them.

The output contract is equally narrow: for each (recipe_id, location_id) the worker emits a theoretical_cost as a quantized Decimal, a cost_delta against the prior committed value, and an idempotency_key. A message that cannot satisfy the schema is never partially processed — it is routed to a dead-letter queue and excluded from publication, so a malformed request never poisons a downstream POS price.

Architecture Decision Rationale

Three decisions define this framework, and each was chosen over a tempting shortcut.

A broker-backed queue over a synchronous call. The direct approach recomputes cost inside the request that triggered the edit. That ties dashboard latency to estate size: a change touching two hundred locations makes the culinary manager wait for two hundred rollups, and any failure mid-loop leaves the ledger half-updated. Enqueuing the work instead gives the producer an immediate acknowledgement and predictable API response times regardless of fan-out, while the broker’s durability means a worker crash re-delivers the message rather than dropping the recalculation. Backpressure becomes a queue-depth number you can scale against, not a cascade of timeouts.

Micro-batches over one-message-per-plate. Emitting a task per affected plate maximizes parallelism but drowns the broker in millions of tiny messages and multiplies per-message overhead. The framework groups a recalculation into micro-batches keyed by recipe and location, so a worker amortizes database round-trips and ingredient lookups across a bounded slice of work. Throughput stays high without the coordination cost of extreme fan-out.

Idempotent workers over at-most-once delivery. Durable brokers guarantee at-least-once delivery: network retries and visibility-timeout expiries mean the same task can arrive twice. Rather than fight that with fragile exactly-once machinery, the worker is made idempotent — a composite key derived from the payload lets a duplicate be recognized and short-circuited to the cached result. Re-delivery becomes a non-event instead of a double-counted cost delta.

Phase 1 — Enqueue and Payload Modeling

The first phase captures a pricing change as a validated, signable message. A Pydantic v2 model enforces the contract at the producer boundary so a malformed request fails at enqueue time rather than corrupting a rollup three hops downstream.

from __future__ import annotations

import hashlib
import hmac
from datetime import datetime
from decimal import Decimal
from enum import Enum

from pydantic import BaseModel, ConfigDict, Field, field_validator


class Trigger(str, Enum):
    vendor_rate = "vendor_rate"
    yield_edit = "yield_edit"
    spec_change = "spec_change"
    bulk_import = "bulk_import"


class CostRecalcTask(BaseModel):
    """One recalculation request destined for the broker."""

    model_config = ConfigDict(frozen=True, extra="forbid")

    recipe_id: str = Field(min_length=1)
    location_ids: tuple[str, ...] = Field(min_length=1)
    effective_at: datetime
    pricing_snapshot: dict[str, Decimal]
    trigger: Trigger

    @field_validator("pricing_snapshot", mode="before")
    @classmethod
    def _as_decimal(cls, v: dict[str, object]) -> dict[str, Decimal]:
        # Route every price through str() so a float literal never taints the Decimal.
        return {sku: Decimal(str(cost)) for sku, cost in v.items()}

    def canonical_bytes(self) -> bytes:
        """Stable byte encoding for signing and idempotency (sorted, no whitespace)."""
        locs = ",".join(sorted(self.location_ids))
        prices = ",".join(f"{k}={v}" for k, v in sorted(self.pricing_snapshot.items()))
        return f"{self.recipe_id}|{locs}|{self.effective_at.isoformat()}|{prices}".encode()

    def sign(self, secret: bytes) -> str:
        return hmac.new(secret, self.canonical_bytes(), hashlib.sha256).hexdigest()

The frozen=True / extra="forbid" config is deliberate: the task is an immutable value object, and any unexpected field is a schema drift that should fail loudly at construction. Casting each price through Decimal(str(cost)) — not Decimal(cost) — keeps a stray float literal from injecting binary-rounding error before the value is ever serialized. The canonical_bytes encoding sorts locations and prices so that two logically identical requests produce byte-identical payloads, which is what makes both the HMAC signature and the idempotency key stable across producers. The change events that feed this step arrive from the same ingestion surfaces the rest of the framework owns — structured spreadsheet loads via CSV bulk import automation and chef spec sheets via PDF recipe extraction pipelines — each of which normalizes its output to this one task shape before enqueueing.

Phase 2 — Validation and Dead-Letter Routing

Passing the type contract at the producer proves the message was well-formed when it was created; it does not prove it is still trustworthy when a worker picks it up. Phase 2 re-validates at the consumer boundary: it re-parses the payload, verifies the signature, and confirms the referenced recipe and locations still exist. Anything that fails is acknowledged off the main queue and written to a dead-letter queue with a machine-readable reason — never retried blindly and never allowed to stall the working thread.

import logging

from pydantic import ValidationError

logger = logging.getLogger("cost_pipeline")


class TaskRejected(Exception):
    """Raised when a message must be dead-lettered rather than processed."""

    def __init__(self, reason: str) -> None:
        self.reason = reason
        super().__init__(reason)


def validate_task(raw: dict, secret: bytes, signature: str) -> CostRecalcTask:
    """Re-validate at the worker boundary; dead-letter on any failure."""
    try:
        task = CostRecalcTask.model_validate(raw)
    except ValidationError as exc:
        raise TaskRejected(f"schema_invalid:{exc.error_count()}") from exc

    expected = task.sign(secret)
    if not hmac.compare_digest(expected, signature):
        raise TaskRejected("signature_mismatch")

    return task


def consume(raw: dict, secret: bytes, signature: str, dead_letter) -> CostRecalcTask | None:
    try:
        return validate_task(raw, secret, signature)
    except TaskRejected as exc:
        # Structured log line per rejected message; ack + route, do not raise.
        logger.warning(
            "task_dead_lettered",
            extra={"reason": exc.reason, "recipe_id": raw.get("recipe_id")},
        )
        dead_letter.put({"payload": raw, "reason": exc.reason})
        return None

The distinction between a transient failure and a permanent one is what keeps the pipeline stable. A malformed schema or a signature mismatch is permanent — retrying it a thousand times just burns worker cycles, so it goes straight to the dead-letter queue for human triage. A database lock or a broker hiccup is transient and belongs to the retry path in Phase 3. Conflating the two is the classic mistake that turns one poison message into an infinite redelivery loop. Every rejection carries a structured reason so an operator can distinguish a broken producer (schema_invalid) from a key-rotation problem (signature_mismatch) without reading raw payloads.

Phase 3 — Idempotent Rollup and Delta Publication

Validated tasks reach the worker, which recomputes theoretical cost in micro-batches and publishes the deltas downstream. Idempotency is enforced first: a composite key derived from the canonical payload is checked against a result store, and a duplicate returns the cached result without recomputing. Monetary math runs in Decimal and is quantized once at the boundary.

from decimal import ROUND_HALF_UP, Decimal

CENTS = Decimal("0.0001")


def idempotency_key(task: CostRecalcTask) -> str:
    return hashlib.sha256(task.canonical_bytes()).hexdigest()


def process_task(task: CostRecalcTask, ep_costs: dict[str, Decimal], results, bus) -> dict:
    """Recompute EP-based plate cost per location; publish deltas; dedupe on retry."""
    key = idempotency_key(task)
    cached = results.get(key)          # Redis SETNX-style guard, omitted for brevity
    if cached is not None:
        return cached                  # duplicate delivery => no double count

    deltas: dict[str, dict[str, Decimal]] = {}
    for loc in task.location_ids:
        # Walk the recipe BOM, costing each leaf at its edible-portion cost.
        theoretical = Decimal("0")
        for sku, qty in ep_costs.items():
            theoretical += (qty * task.pricing_snapshot.get(sku, Decimal("0")))
        theoretical = theoretical.quantize(CENTS, rounding=ROUND_HALF_UP)

        prior = results.get_prior_cost(task.recipe_id, loc) or Decimal("0")
        deltas[loc] = {"theoretical_cost": theoretical, "cost_delta": theoretical - prior}

    payload = {"recipe_id": task.recipe_id, "key": key, "deltas": deltas}
    results.set(key, payload)          # store before publishing => at-least-once safe
    bus.publish("cost.delta", payload) # downstream POS + inventory consumers
    return payload

Storing the result before publishing, not after, is what makes at-least-once delivery safe: if the worker dies between the set and the publish, the redelivered task finds the cached result and re-publishes the identical delta, so the ledger converges rather than diverging. The leaf costs the worker multiplies against are not raw invoice prices — they are edible-portion costs produced upstream by the yield factor calculation frameworks, so a recalculation already accounts for trim and shrink before it ever reaches a plate total. Once deltas are published, the downstream handoff is deliberately decoupled from live ordering: terminals subscribe to the delta stream through the pull-based patterns described in POS API polling strategies, and the same numbers feed variance mapping methodologies where theoretical cost is diffed against actual usage.

Production Hardening

Turning a working consumer into a dependable estate-wide job comes down to a handful of disciplines:

• Idempotency keys. Derive the key from the canonical payload, not from a broker message id, so a message re-published by a producer retry collapses to the same key. Write the result under that key before acknowledging the message; a retried task then short-circuits instead of double-counting a cost delta.
• Retry with backoff and jitter. Wrap only transient failures — lock timeouts, broker disconnects, 429/503 from a downstream — in exponential backoff with randomized jitter, so a broker recovery does not trigger a thundering herd of synchronized retries. Permanent failures skip retry entirely and dead-letter.
• Memory bounds. A wide pricing_snapshot and a deep BOM can make a batch heavy. Consume tasks in bounded micro-batches, stream the ingredient tree rather than materializing it whole, and drop large dictionaries between batches so the resident set stays flat under sustained load.
• Location isolation. Key deltas and result storage on (recipe_id, location_id) so one unit’s recalculation never overwrites another’s — the same isolation the multi-location cost center architecture relies on to stop a regional vendor rate from corrupting an unaffected kitchen’s plate cost.
• Autoscaling on queue depth. Scale worker concurrency against broker queue depth and oldest-message age, not CPU. A rollup is I/O-bound on the database, so CPU stays low while the backlog grows; queue-depth targets keep the SLA window honest during a large bulk import.
• RBAC and audit boundaries. The signing secret lives in a secrets manager, not in code; producers hold the enqueue role, workers run as a service account with write access to the result store alone, and every published delta carries the idempotency_key and run_version so a reader can trace any price back to the task that set it.

Failure Modes and Troubleshooting

The through-line of every failure above is the same: a task must be validated, deduplicated, and costed in exact arithmetic before its delta is published — never trusted on face value, never processed twice, never computed in floating point where money is involved.

FAQ

Why use a broker instead of just calling the rollup function directly?

Because a direct call ties the caller’s latency and reliability to the size of the work. A change touching every location makes the dashboard wait for the whole fan-out, and a crash mid-loop leaves the ledger half-updated. Enqueuing returns immediately, survives worker restarts through broker durability, and turns backpressure into a queue-depth metric you can scale against instead of a wall of timeouts.

How do I stop a duplicated message from applying a cost change twice?

Make the worker idempotent. Derive a composite key from the canonical payload — recipe, sorted locations, effective timestamp, and sorted prices — and check it against a result store before doing any work. Store the computed result under that key before you acknowledge the message, so a redelivered task finds the cached result and re-publishes the identical delta rather than recomputing a second, double-counted change.

What is the difference between the dead-letter queue and the retry path?

They handle opposite failure classes. The retry path is for transient problems — a lock timeout or a broker disconnect — that a later attempt will likely clear, so those are retried with backoff and jitter. The dead-letter queue is for permanent problems — a malformed schema or a bad signature — that no amount of retrying will fix; routing them aside keeps one poison message from looping forever and gives an operator a triage surface.

Can I keep the pricing math in floats and only convert at the end?

No — money enters this pipeline as cost and stays monetary the whole way through, so it belongs in Decimal from ingestion onward. Cast every price through Decimal(str(v)) at the payload boundary, accumulate the plate total in Decimal, and quantize once before publishing. Storing the result as NUMERIC(12,4) mirrors that quantization so the value round-trips without float coercion reintroducing fractional-cent drift across millions of deltas.

Related

• Up one level: Data Ingestion & Recipe Parsing Workflows
• Implementing Celery for Async Menu Syncs
• CSV Bulk Import Automation
• PDF Recipe Extraction Pipelines
• POS API Polling Strategies
• Variance Mapping Methodologies

For library specifics, see the official Celery task documentation and Pydantic documentation.