BadgeprintingEvent registration · badge automation

Using Celery for Async Registration Batch Processing: Fixing Payment Sync Gaps & Worker OOM Kills

Symptom Statement

During high-velocity ticket drops, corporate bulk uploads, or multi-track check-ins, the registration pipeline exhibits a 30–120 second payment sync gap: attendees are marked REGISTERED in the CRM while badge printers queue PAYMENT_PENDING jobs against the same records. Webhook payloads arrive out of sequence, synchronous form polling exhausts connection pools, and application servers take OOM kills mid-batch. The fallout is duplicate badge prints, orphaned payment rows, and a manual reconciliation queue that grows linearly with attendee volume. This page is the distributed-worker implementation documented under async batch processing, the decoupling stage of the Registration Ingestion & Payment Reconciliation pipeline — it covers the exact Celery configuration that turns a stalling synchronous ingest into idempotent, chunked, memory-bounded task processing.

Root Cause Analysis

Four concrete faults produce this class of failure. Each sits at a different boundary — the request thread, the broker, the worker process, and the payment ledger — and each fails in a way that a naive “just add Celery” migration leaves untouched.

  1. Synchronous coupling inside one request thread. Form ingestion, payment verification, and template rendering execute in the same blocking handler. The request holds database row locks during schema validation while waiting on external gateway latency, so every downstream hiccup applies backpressure straight back to the registration form.
  2. Monolithic, unchunked task payloads. Dispatching an entire bulk upload as a single task monopolizes one worker thread, blows past the broker’s message size ceiling, and triggers ACK timeouts that cause the whole batch to be redelivered and reprocessed from zero.
  3. Unbounded worker memory growth. Long-lived worker children accumulate ORM sessions, large JSON payloads, and unreleased file descriptors. Resident set size climbs until the kernel OOM-kills the worker, dropping every in-flight task with it.
  4. Out-of-order webhook reconciliation. Payment webhooks arrive out of sequence because of network jitter and gateway retries. A last-writer-wins update lets a late pending event overwrite a COMPLETED one, un-gating a badge that should already have printed. Signature verification and the real-time event contract are owned upstream by payment webhook handling; this page owns only the idempotent settlement into the ledger.
Celery chunked registration processing path An API request enqueues chunked RegistrationChunk tasks into a Redis broker; a memory-bounded prefork worker pool with prefetch=1 drains them through schema revalidate, idempotency guard, payment reconciliation upsert and badge print dispatch. A dashed exponential-backoff retry loop returns transient failures to the broker while FATAL and MAX_RETRIES chunks branch into a dead-letter queue. API Request chunk_payload() .delay(chunk) Broker Redis FIFO queue Prefork Worker Pool — prefetch=1 · max_tasks_per_child=100 · max_memory_per_child=512 MB SchemaRevalidate IdempotencyGuard Payment UpsertON CONFLICT Badge PrintDispatch printer retry · exponential backoff (2^n) FATAL / MAX_RETRIES Dead-Letter Queue quarantine · manual review

Symptom-to-Resolution Matrix

Synchronous Coupling & Unchunked Payloads

Symptoms

  • Registration form latency spikes to seconds during bulk uploads; connection pools saturate.
  • A single large upload pins one worker at 100% and starves every other task.
  • Broker rejects the message or the task is redelivered repeatedly with ACK timeouts.

Root cause. The batch is one blocking unit of work with no failure isolation, so partial success is impossible and one bad record fails the whole payload.

Fix steps

  1. Segment the payload into fixed-size chunks before dispatch so each carries a deterministic batch_id and idempotency_key derived from the source submission hash.
  2. Dispatch each chunk with .delay(chunk.model_dump()) — pass the JSON-serializable dict, not the Pydantic object, because Celery serializes arguments as JSON.
  3. Set task_acks_late=True and worker_prefetch_multiplier=1 so a crashed worker releases its in-flight chunk for redelivery instead of losing it.
PYTHON
import uuid, hashlib
from typing import List, Dict, Any
from pydantic import BaseModel

class RegistrationChunk(BaseModel):
    batch_id: str
    idempotency_key: str
    records: List[Dict[str, Any]]

def generate_idempotency_key(payload: Dict[str, Any]) -> str:
    raw = f"{payload.get('email','')}-{payload.get('ticket_type','')}-{payload.get('timestamp','')}"
    return hashlib.sha256(raw.encode()).hexdigest()

def chunk_payload(records: List[Dict[str, Any]], chunk_size: int = 50) -> List[RegistrationChunk]:
    batch_id = str(uuid.uuid4())
    return [
        RegistrationChunk(
            batch_id=batch_id,
            idempotency_key=generate_idempotency_key(records[i]),
            records=records[i : i + chunk_size],
        )
        for i in range(0, len(records), chunk_size)
    ]

Worker OOM Kills

Symptoms

  • Workers restart with SIGKILL; dmesg shows the OOM killer reaping the prefork child.
  • Resident memory climbs monotonically across a batch and never returns to baseline.
  • Result backend fills with stale entries the worker never reclaims.

Root cause. Worker children live indefinitely, so leaked ORM sessions and large payloads accumulate until RSS exceeds the host ceiling.

Fix steps

  1. Recycle each child after a bounded number of tasks with worker_max_tasks_per_child so leaked memory is reclaimed by a fresh process.
  2. Enforce a hard RSS ceiling with worker_max_memory_per_child (in KB) — the child is retired the moment it crosses it, mid-batch-safe because tasks are idempotent.
  3. Expire results with result_expires so the backend does not grow unbounded across a multi-day event.
PYTHON
app.conf.update(
    worker_max_tasks_per_child=100,      # restart child to clear leaks
    worker_max_memory_per_child=512000,  # retire child if RSS > 512 MB (KB)
    worker_concurrency=4,                # match CPU cores; avoid oversubscription
    broker_connection_retry_on_startup=True,
    result_expires=3600,                 # reclaim stale results after 1 hour
)

Out-of-Order Payment Reconciliation

Symptoms

  • A record flips from COMPLETED back to PENDING; the badge re-queues after it already printed.
  • Gateway dashboard shows succeeded while the ledger shows an earlier state.
  • Duplicate print jobs appear for a single transaction.

Root cause. Reconciliation blindly overwrites ledger state, so a delayed lower-priority event clobbers a terminal one.

Fix steps

  1. Upsert with ON CONFLICT and a state-machine guard so a row already COMPLETED is never downgraded.
  2. Discard malformed webhook payloads with raise Ignore() so they never enter the retry loop.
  3. Retry only on genuine transient faults (DB connection loss) with capped exponential backoff.
PYTHON
import os, logging
from typing import Dict, Any
import psycopg2
from celery.exceptions import Ignore

logger = logging.getLogger(__name__)

@app.task(bind=True, name="tasks.reconcile_payment_webhook", max_retries=8)
def reconcile_payment_webhook(self, webhook_payload: Dict[str, Any]) -> str:
    txn_id = webhook_payload.get("transaction_id")
    status = webhook_payload.get("status")
    if not txn_id or not status:
        raise Ignore()  # malformed; discard without retry
    try:
        with psycopg2.connect(os.getenv("DB_URL")) as conn, conn.cursor() as cur:
            cur.execute(
                """
                INSERT INTO payment_ledger (txn_id, status, updated_at, idempotency_hash)
                VALUES (%s, %s, NOW(), %s)
                ON CONFLICT (txn_id) DO UPDATE SET
                    status = EXCLUDED.status, updated_at = NOW()
                WHERE payment_ledger.status != 'COMPLETED'
                """,
                (txn_id, status, webhook_payload.get("idempotency_key")),
            )
            conn.commit()
        if status == "COMPLETED":
            trigger_badge_print.delay(txn_id)
        return f"Reconciled {txn_id} -> {status}"
    except psycopg2.OperationalError as e:
        logger.warning("DB connection lost, retrying: %s", e)
        raise self.retry(exc=e, countdown=2 ** self.request.retries)

Minimal Working Implementation

A single self-contained path: configure the Celery app with late-ACK and idempotency-safe defaults, validate each chunk at the task boundary, categorize failures, and dispatch. The structural contract for the records inside a chunk is owned by schema validation pipelines; this task revalidates only enough to route safely. Malformed records are quarantined via the dead-letter queue rather than crashing the batch.

PYTHON
import os, logging
from enum import Enum
from typing import Annotated, Dict, Any
from celery import Celery
from pydantic import BaseModel, EmailStr, Field, ValidationError

logger = logging.getLogger(__name__)

app = Celery(
    "event_reg",
    broker=os.getenv("CELERY_BROKER_URL", "redis://localhost:6379/0"),
    backend=os.getenv("CELERY_RESULT_BACKEND", "redis://localhost:6379/1"),
)
app.conf.update(
    task_serializer="json", accept_content=["json"], result_serializer="json",
    timezone="UTC", enable_utc=True,
    task_acks_late=True,                 # release chunk if worker dies mid-task
    task_reject_on_worker_lost=True,
    worker_prefetch_multiplier=1,        # no over-buffering; fair redelivery
    task_default_retry_delay=15, task_max_retries=5,
    task_time_limit=300, task_soft_time_limit=240,
)

class ErrorCategory(str, Enum):
    RETRYABLE = "RETRYABLE"        # transient network / 5xx — retry
    FATAL = "FATAL"               # schema violation — dead-letter, no retry
    MANUAL_REVIEW = "MANUAL_REVIEW"  # business-rule conflict — human queue

class RegistrationPayload(BaseModel):
    email: EmailStr
    ticket_type: str = Field(min_length=3, max_length=50)
    quantity: Annotated[int, Field(ge=1, le=10)]  # conint is deprecated in v2
    payment_method: str

def validate_and_dispatch(record: Dict[str, Any]) -> None:
    try:
        RegistrationPayload.model_validate(record)  # proceed to payment gateway
    except ValidationError as e:
        logger.error("registration_error category=%s error=%s", ErrorCategory.FATAL.value, e)
    except Exception as e:
        msg = str(e).lower()
        if any(t in msg for t in ("timeout", "503", "502")):
            logger.error("registration_error category=%s error=%s", ErrorCategory.RETRYABLE.value, e)
            raise
        logger.error("registration_error category=%s error=%s", ErrorCategory.MANUAL_REVIEW.value, e)

@app.task(bind=True, name="tasks.process_registration_chunk", max_retries=5)
def process_registration_chunk(self, chunk_data: Dict[str, Any]) -> Dict[str, Any]:
    try:
        chunk = RegistrationChunk.model_validate(chunk_data)  # revalidate at boundary
    except ValidationError as exc:
        logger.error("Invalid chunk schema: %s", exc)
        raise  # FATAL — routed to DLQ, not retried
    # Idempotency guard: if db.exists(chunk.idempotency_key): return {"status": "skipped"}
    try:
        for record in chunk.records:
            validate_and_dispatch(record)
        return {"batch_id": chunk.batch_id, "status": "completed", "processed": len(chunk.records)}
    except Exception as exc:
        raise self.retry(exc=exc, countdown=self.request.retries * 15)

if __name__ == "__main__":
    # Verification: chunking is deterministic and the dispatched arg is JSON-serializable.
    chunks = chunk_payload([{"email": "[email protected]", "ticket_type": "vip"}], chunk_size=50)
    assert len(chunks) == 1 and chunks[0].idempotency_key
    process_registration_chunk.delay(chunks[0].model_dump())  # plain dict, not the model
    print("Dispatched chunk", chunks[0].batch_id)

Start the worker with the prefork pool and the same memory guardrails so RSS never outruns the host:

BASH
celery -A event_reg worker \
  --loglevel=INFO --concurrency=4 --pool=prefork \
  --max-memory-per-child=512000 --max-tasks-per-child=100 \
  --without-gossip --without-mingle --without-heartbeat

Memory & Performance Constraints

Component Constraint Mitigation
Worker child RSS Leaked ORM sessions/payloads climb until OOM kill worker_max_tasks_per_child=100 + worker_max_memory_per_child=512000
Broker prefetch Over-prefetch buffers memory and blocks redelivery on crash worker_prefetch_multiplier=1 with task_acks_late=True
Chunk size Large chunks hold every record in memory and lengthen ACK window chunk_size≈50; tune against task_soft_time_limit=240
DB connection pool Concurrency above pool size exhausts connections Cap worker_concurrency at pool_size - headroom; use pgbouncer
Result backend Unbounded result growth over a multi-day event result_expires=3600; avoid storing large payloads as results
Retry amplification Backoff plus high concurrency floods a recovering gateway Cap max_retries; add jitter to countdown

Incident Triage & Rollback

Target under 15 minutes to a clean fallback. Contain first, diagnose second — never force-terminate a worker mid-reconciliation, since a half-applied ledger write is worse than a paused queue.

  1. Identify backpressure. celery -A event_reg inspect active --json and check consumer lag; confirm whether one source or all of them are affected.
  2. Inspect the broker. redis-cli LLEN registration.batch && redis-cli LLEN registration.batch.dlq; if fragmentation ratio > 1.5 from redis-cli INFO memory, plan a MEMORY PURGE only after in-flight webhooks drain.
  3. Recycle workers. sudo systemctl restart celery-worker to clear a leaking child; purge stale tasks with celery -A event_reg purge only when they are idempotent and safe to drop.
  4. Find orphans. Query for stuck records and re-drive them behind a manual gate:
    SQL
    SELECT id, email, created_at FROM registrations
WHERE payment_status = 'PAYMENT_PENDING'
  AND created_at < NOW() - INTERVAL '1 hour'
ORDER BY created_at ASC;

Rollback to synchronous fallback. If the broker fails or workers crash-loop, set export REGISTRATION_ASYNC_MODE=false to route /register at the legacy sync handler, then flush pending tasks with celery -A event_reg control shutdown. Verify pool limits (pool_size=10, max_overflow=5) before restoring load.

Rollback validation. Confirm payment_ledger row counts match CRM REGISTERED counts, the badge printer queue depth returns below 5, and celery -A event_reg inspect stats shows no task-rejection spikes before re-enabling async routing.

Frequently Asked Questions

Why pass chunk.model_dump() instead of the RegistrationChunk object to .delay()? Celery serializes task arguments as JSON, and a Pydantic model is not JSON-serializable by default. Passing the plain dict from model_dump() keeps the payload transport-safe; the task deserializes it back with RegistrationChunk.model_validate() at its own boundary, which also revalidates the contract before any work begins.

Doesn’t task_acks_late=True risk processing the same chunk twice? Yes — late ACK means a chunk redelivered after a worker crash can be picked up again. That is safe only because every side effect is keyed on idempotency_key: the idempotency guard turns a redelivered chunk into a no-op, and the payment upsert refuses to downgrade a COMPLETED row. Late ACK without those guards would double-print.

Why worker_prefetch_multiplier=1 for registration batches? Batch tasks are long-running and uneven in cost, so the default prefetch lets one worker hoard messages it cannot start, starving idle workers and delaying redelivery when it crashes. Prefetch of 1 gives fair distribution and immediate release of an in-flight chunk on worker loss.