Configuring Async Celery Workers for High-Volume CTE Ingestion

The default Celery configuration is tuned for short, uniform, idempotency-agnostic jobs — exactly the opposite of FSMA 204 Critical Tracking Event ingestion, where each task commits a compliance record that must never be lost and must never be written twice. Left on its defaults, a worker prefetches a large batch of tasks into memory, acknowledges each task the moment it is received rather than after it commits, and happily reprocesses anything a crashed worker had already half-finished. Under peak-harvest load those defaults produce two visible symptoms: memory climbs until the worker is killed, and the traceability ledger accumulates duplicate rows for the same lot. This guide shows the exact settings that fix both, as an incremental change to the High-Volume CTE Ingestion design, with a reproducible failure and a verification you can run.

Root Cause Analysis

Two Celery defaults collide badly with compliance ingestion. The first is worker_prefetch_multiplier, which defaults to 4. With a concurrency of 12, a single worker prefetches up to 48 tasks and holds them in memory before it has committed any of them. That is efficient for tiny tasks, but a CTE payload plus its validation state is not tiny, and at peak the prefetch buffer becomes an unbounded-looking memory sink — the worker’s resident set grows with the queue, and the OOM killer eventually reaps it, dropping every prefetched-but-uncommitted event with it.

The second default is early acknowledgment. By default Celery acks a task when the worker receives it, not when it finishes. If the worker dies after acking but before committing the CTE to the ledger, the broker considers the task done and never redelivers it — the event is silently lost. Flip the reasoning and the opposite failure appears: if you enable late acknowledgment but set a Redis visibility_timeout shorter than a task can run, the broker redelivers a task that is still in flight, and two workers commit the same CTE. Without an idempotency key, that is a duplicate ledger row; with one, it is wasted capacity. Both symptoms trace to the same root: acknowledgment and prefetch are not aligned with the durability the record demands. The corrective settings are drawn from the same discipline the Async Batch Processing engine applies to bounded concurrency.

Minimal Reproducible Example

The app below reproduces both failures. It leaves worker_prefetch_multiplier at its default, acks early, and its task is not idempotent — it inserts a row unconditionally. Run a burst through it and you will see memory grow with the backlog and duplicate rows appear whenever a worker is redelivered a task.

# mre_worker.py  -- demonstrates the failure; do not ship this.
from celery import Celery

app = Celery("mre", broker="redis://localhost:6379/0")

# No overrides: prefetch_multiplier defaults to 4, acks are early.

LEDGER: list[dict] = []  # stand-in for the traceability ledger


@app.task(name="mre.ingest")
def ingest(event: dict) -> None:
    # Not idempotent: a redelivered task inserts a second row for the same CTE.
    LEDGER.append(event)
    # Heavy per-task state that the prefetch buffer multiplies under load.
    _ = [event.copy() for _ in range(2000)]

# flood.py -- enqueue a burst that exposes both defects.
from mre_worker import ingest

for i in range(50_000):
    ingest.apply_async(
        args=[{
            "traceability_lot_code": "ROMA-2026-0714",
            "event_type": "receiving",
            "source_event_id": f"evt-{i}",
        }],
        queue="celery",
    )

Start one worker with celery -A mre_worker worker --concurrency=12 and run flood.py. Watch the resident memory of the worker climb as it prefetches thousands of tasks, and — if you kill and restart the worker mid-run — watch len(LEDGER) exceed the number of distinct source_event_id values, because redelivered tasks re-inserted rows that were already committed.

Fix Implementation

The fix is a short, deliberate configuration block plus an idempotent task. Four settings do the work: worker_prefetch_multiplier=1 caps each process at one unacknowledged task, so memory tracks concurrency instead of queue depth; task_acks_late=True acknowledges only after the task returns cleanly, so a crash redelivers rather than drops; task_reject_on_worker_lost=True requeues a task whose worker died; and a visibility_timeout set above the worst-case task duration prevents in-flight redelivery. The task itself upserts on an idempotency key so any redelivery collapses to a single row.

The diagram shows how these settings shape one task’s lifecycle: the broker hands exactly one task per process into the worker, the worker validates and commits, and only then does the acknowledgment travel back to the broker.

# ingest_worker.py -- production configuration for high-volume CTE ingestion.
import hashlib
import logging
from datetime import datetime, timezone
from decimal import Decimal
from typing import Any

from celery import Celery
from celery.utils.log import get_task_logger
from pydantic import BaseModel, ConfigDict, Field, ValidationError, field_validator

logging.basicConfig(level=logging.INFO)
logger = get_task_logger("fsma204.celery_cfg")

app = Celery(
    "fsma204_cte",
    broker="redis://localhost:6379/0",     # RabbitMQ (amqp://) is equally valid
    backend="redis://localhost:6379/1",
)

app.conf.update(
    # --- Throughput and memory -----------------------------------------------
    worker_concurrency=12,               # processes per node; the throughput lever
    worker_prefetch_multiplier=1,        # one unacked task per process -> bounded memory
    worker_max_tasks_per_child=1000,     # recycle processes to cap memory fragmentation
    # --- Durability ----------------------------------------------------------
    task_acks_late=True,                 # ack only after the task returns cleanly
    task_reject_on_worker_lost=True,     # requeue a task whose worker died mid-flight
    # --- Redis redelivery safety --------------------------------------------
    broker_transport_options={"visibility_timeout": 3600},  # > longest task time
    # --- Routing: one queue per lot-hash bucket keeps a lot's events ordered --
    task_default_queue="cte.ingest",
    task_routes={"fsma204.*.ingest_cte": {"queue": "cte.ingest"}},
)


class CTEEvent(BaseModel):
    """FSMA 204 Critical Tracking Event with an idempotency key."""

    model_config = ConfigDict(str_strip_whitespace=True, extra="ignore")

    traceability_lot_code: str = Field(..., min_length=1, max_length=64)
    event_type: str = Field(..., min_length=1)
    location_gln: str = Field(..., pattern=r"^\d{13}$")
    event_timestamp: datetime
    quantity: Decimal = Field(..., gt=0)
    source_event_id: str = Field(..., min_length=1)

    @field_validator("event_timestamp")
    @classmethod
    def require_aware(cls, v: datetime) -> datetime:
        if v.tzinfo is None or v.tzinfo.utcoffset(v) is None:
            raise ValueError("event_timestamp must be timezone-aware")
        return v

    def idempotency_key(self) -> str:
        composite = f"{self.traceability_lot_code}|{self.event_type}|{self.source_event_id}"
        return hashlib.sha256(composite.encode("utf-8")).hexdigest()


class TransientLedgerError(Exception):
    """Retryable downstream fault."""


@app.task(
    bind=True,
    name="fsma204.cte.ingest_cte",
    max_retries=5,
    autoretry_for=(TransientLedgerError,),
    retry_backoff=True,
    retry_jitter=True,
)
def ingest_cte(self: Any, raw_event: dict[str, Any]) -> dict[str, str]:
    """Validate and idempotently append one CTE. Safe to run more than once."""
    try:
        event = CTEEvent(**raw_event)
    except ValidationError as exc:
        logger.error("KDE validation failed, quarantining | %s", exc)
        return {"status": "quarantined"}

    key = event.idempotency_key()
    # upsert_cte returns True on a new insert, False when the key already exists.
    was_new = upsert_cte(event.model_dump(mode="json"), key)
    logger.info(
        "CTE %s | lot=%s | key=%s",
        "committed" if was_new else "deduplicated",
        event.traceability_lot_code, key[:12],
    )
    return {"status": "committed" if was_new else "deduplicated"}


def upsert_cte(row: dict[str, Any], idempotency_key: str) -> bool:
    """Insert row keyed on idempotency_key; return False if it already existed.

    Backed by a UNIQUE constraint on idempotency_key, e.g. Postgres:
        INSERT INTO cte_ledger (...) VALUES (...)
        ON CONFLICT (idempotency_key) DO NOTHING RETURNING id;
    """
    raise NotImplementedError

Start the pool with the settings baked into the app, not the command line, so they are versioned with the code: celery -A ingest_worker worker --loglevel=info. The visibility_timeout of 3600 seconds must exceed the longest a single ingest can take, including retries, or Redis will redeliver an in-flight task. The idempotent upsert_cte — backed by a UNIQUE constraint on idempotency_key — is what makes the whole configuration safe: even if a redelivery does happen, the second write is a no-op.

Verification Steps

Confirm the configuration along two axes: throughput with bounded memory, and zero duplicate ledger rows under redelivery.

• Memory stays bounded under a flood. Re-run the 50,000-event burst against the fixed worker while sampling resident memory (ps -o rss= -p <worker_pid> on a loop). With worker_prefetch_multiplier=1 the resident set should plateau — it tracks concurrency, not the 50,000-deep backlog — instead of climbing with the queue as the reproduction did.
• Prefetch is actually one per process. Inspect the broker mid-run: celery -A ingest_worker inspect active should show at most worker_concurrency active tasks per node, and the number of unacknowledged messages Redis reports should not exceed the process count. If it does, a stray --prefetch-multiplier flag is overriding the config.
• No duplicate ledger rows. Enqueue a known set of distinct source_event_id values, then kill -9 the worker mid-run and restart it to force redelivery. Afterward, the ledger row count must equal the number of distinct idempotency keys enqueued:

-- Every committed CTE is unique on its idempotency key; redeliveries collapse.
SELECT COUNT(*)                         AS total_rows,
       COUNT(DISTINCT idempotency_key)  AS distinct_keys
FROM cte_ledger;
-- total_rows MUST equal distinct_keys. Any gap means a non-idempotent write path.

• Late ack redelivers rather than drops. With task_acks_late=True, killing a worker mid-task should leave the task in the queue for another worker. Confirm by asserting the killed task’s source_event_id is present in the ledger after the run — it was redelivered and committed, not lost.

Related Edge Cases

Once the base configuration is correct, three adjacent settings deserve a check.

• acks_late without a bounded max_retries. Late acknowledgment plus an unbounded retry policy turns a poison task into an infinite redelivery loop that starves the pool. Keep max_retries finite and route the exhausted task to the dead-letter path owned by the Error Handling Workflows.
• Concurrency that outruns the ledger. Raising worker_concurrency past what the ledger can absorb just moves the queue from the broker into database lock contention. Pair concurrency tuning with the queue-depth control loop in Tuning Backpressure for Supplier Feeds so acceptance slows before the ledger saturates. Feeds that arrive by polling should also respect the cadence in API Polling Strategies.
• Result backend bloat. Storing every task result in Redis with no expiry slowly fills the backend. Set result_expires or disable results for fire-and-forget ingest tasks that only need their ledger side effect.

FSMA 204 requires that traceability records be maintained and producible within 24 hours of an FDA request, so a worker configuration that silently drops or duplicates CTEs is a compliance defect, not merely a tuning nuisance. The settings above make acknowledgment align with durability and make every write idempotent, which is what keeps the ledger both complete and truthful. For the authoritative requirement, see the FDA final rule on food traceability.

Frequently Asked Questions

Related

• High-Volume CTE Ingestion — the queue-fronted worker-pool design this configuration implements.
• Tuning Backpressure for Supplier Feeds — the queue-depth control loop that pairs with concurrency tuning.
• Async Batch Processing — the bounded-concurrency batch engine that shares these idempotency patterns.
• Error Handling Workflows — the dead-letter and retry path for tasks that exhaust their retry budget.
• API Polling Strategies — the fetch cadence that feeds events into these workers within supplier quotas.

Up: High-Volume CTE Ingestion — this guide configures the Celery pool at the heart of that intake tier.