Managing Celery Task Queues for Overnight Batch Imports

12 min read

This guide is a focused build within Implementing Async Batch Processing for High-Volume Submissions, itself part of the broader automated permit ingestion and parsing workflows that municipal technology teams run end to end. Here we drill into one operational problem: how to route, isolate, and harden Celery task queues so that hundreds of thousands of permit records can be imported during a fixed overnight maintenance window without leaking into daytime service hours.

Municipal permit systems run inside rigid maintenance windows. Bulk ingestion of nightly CSV exports from legacy county systems, scraped portal submissions, and OCR-processed PDF attachments must complete between (typically) 01:00 and 05:00, then hand a clean database back to clerks before the public portal reopens. A naive delay() call on a single default queue cannot meet that contract. When an overnight batch shares workers with interactive lookups, a slow county GIS call or a malformed export stalls the whole broker, daytime SLAs break, and reconciliation officers inherit a half-written database with no audit trail. The task is therefore to design queue topology, idempotency, and worker lifecycle as compliance controls — not as performance afterthoughts.

Celery queue topology for overnight permit batch imports A Celery beat dispatcher fires at 01:00 and chunks three sources — nightly CSV exports, scraped portal submissions, and OCR-processed PDFs — into idempotent batches routed by task_routes to three priority queues on a TLS RabbitMQ broker: permit.ingest.critical (TTL 1 hour, priority 10), permit.ingest.standard (TTL 3 hours), and permit.ingest.maintenance (TTL 4 hours). A dedicated overnight worker pool (prefork, concurrency 8, prefetch 1) consumes only these ingest queues and commits to PostgreSQL inside atomic transactions. When a message exceeds its TTL or exhausts retries it dead-letters through the DLX into permit.ingest.quarantine for compliance review. A dashed isolation boundary encloses the overnight ingest subsystem; a separate daytime pool serving permit.lookup sits outside it and shares no workers. OVERNIGHT INGEST · 01:00–05:00 WINDOW DAYTIME SERVICE ISOLATION BOUNDARY — no shared workers CSV export legacy county Portal scrape submissions OCR PDF attachments Beat dispatch crontab(1, 0) chunk · idempotent scope by FIPS RabbitMQ broker durable · TLS 5671 · DLX permit.ingest.critical TTL 1h · priority 10 permit.ingest.standard TTL 3h · routine permits permit.ingest.maintenance TTL 4h · backfill task_routes Overnight pool --queues *.ingest.* prefork · concurrency 8 prefetch 1 · acks_late recycle per 200 tasks consume PostgreSQL atomic txn unique(idem_key) commit permit.ingest.quarantine held for compliance review · not /dev/null TTL expiry / retries exhausted → DLX Daytime pool --queues permit.lookup interactive clerk lookups never touches ingest queues
Overnight beat dispatch fans CSV, portal, and OCR sources into three TTL-bounded, dead-lettering ingest queues consumed by a dedicated overnight pool — isolated from the daytime permit.lookup pool so a stalled batch can never break daytime SLAs.

Step 1: Declare Isolated Queues With Dead-Letter Routing

Overnight batch workloads must be physically segregated from real-time inspection scheduling, public-facing API endpoints, and interactive clerk dashboards. The isolation starts in the Celery configuration: declare explicit queues with deterministic routing keys, attach a dead-letter exchange (DLX) and a message TTL to every ingest queue, and bind a dedicated quarantine queue so failed messages are captured for forensic review rather than silently dropped.

# celeryconfig.py — Python 3.10+, celery>=5.3, kombu broker declarations
from kombu import Exchange, Queue

broker_url = "amqp://permit_svc:***@broker.internal:5671//"  # TLS (5671) for sensitive applications
result_backend = "redis://cache.internal:6379/2"

# Survive a broker restart mid-window without crashing the beat scheduler.
broker_connection_retry_on_startup = True
broker_pool_limit = 32  # must track total overnight worker concurrency (see Step 3)

ingest_exchange = Exchange("permit.ingest", type="direct", durable=True)
dlx_exchange = Exchange("permit.ingest.dlx", type="direct", durable=True)

# Quarantine: failed messages land here for compliance-officer review, not /dev/null.
quarantine = Queue(
    "permit.ingest.quarantine",
    exchange=dlx_exchange,
    routing_key="quarantine",
    durable=True,
)

def _ingest_queue(name: str, ttl_ms: int) -> Queue:
    """An ingest queue that dead-letters to quarantine after `ttl_ms`."""
    return Queue(
        name,
        exchange=ingest_exchange,
        routing_key=name,
        durable=True,
        queue_arguments={
            "x-message-ttl": ttl_ms,                 # stale payloads expire, never block the window
            "x-dead-letter-exchange": "permit.ingest.dlx",
            "x-dead-letter-routing-key": "quarantine",
            "x-max-priority": 10,                    # zoning/fee-sensitive work jumps the line
        },
    )

task_queues = (
    _ingest_queue("permit.ingest.critical", ttl_ms=3_600_000),    # zoning, environmental, fee-bearing
    _ingest_queue("permit.ingest.standard", ttl_ms=10_800_000),   # routine building permits
    _ingest_queue("permit.ingest.maintenance", ttl_ms=14_400_000),# data-hygiene / backfill
    quarantine,
)

# Deterministic routing — no task ever falls back to a shared "celery" queue.
task_routes = {
    "ingest.tasks.import_zoning_chunk":   {"queue": "permit.ingest.critical"},
    "ingest.tasks.import_permit_chunk":   {"queue": "permit.ingest.standard"},
    "ingest.tasks.backfill_chunk":        {"queue": "permit.ingest.maintenance"},
}

task_acks_late = True          # redeliver on worker crash — safe ONLY because tasks are idempotent
task_reject_on_worker_lost = True
task_default_priority = 5

RabbitMQ is the recommended broker for these deployments because of native DLX support, durable persistence, and priority-queue extensions. Sensitive zoning, environmental, and fee-bearing submissions traverse the TLS broker on the critical queue while lower-priority backfill jobs stay isolated on maintenance, so a flood of data-hygiene work can never starve compliance-critical imports.

Step 2: Chunk Imports Into Idempotent Units

Never load an entire municipal export into application memory. A generator-based dispatcher streams the source file and emits fixed-size batches to the broker, keeping the memory footprint flat across worker nodes. Critically, chunk boundaries align with jurisdiction (jurisdiction_fips) so that one county’s records never co-mingle with another’s — a hard requirement under state data-residency statutes and a prerequisite for the row-level security applied downstream.

# ingest/dispatch.py
import csv
from collections.abc import Iterator
from itertools import islice
from ingest.tasks import import_permit_chunk

CHUNK_SIZE = 500  # bounded so a single chunk fits comfortably on a constrained municipal VM

def _stream_rows(path: str) -> Iterator[dict[str, str]]:
    # newline="" + utf-8-sig defends against BOM-prefixed legacy county exports.
    with open(path, newline="", encoding="utf-8-sig") as fh:
        yield from csv.DictReader(fh)

def dispatch_overnight_import(path: str, jurisdiction_fips: str, batch_id: str) -> int:
    """Fan a nightly export out into idempotent, jurisdiction-scoped chunks."""
    rows = _stream_rows(path)
    dispatched = 0
    while chunk := list(islice(rows, CHUNK_SIZE)):
        import_permit_chunk.apply_async(
            kwargs={
                "records": chunk,
                "jurisdiction_fips": jurisdiction_fips,
                "batch_id": batch_id,          # ties every chunk back to one run for reconciliation
            },
            queue="permit.ingest.standard",
            priority=5,
        )
        dispatched += 1
    return dispatched

The import task itself must be idempotent because acks_late and broker retries will redeliver messages after a crash. Build a deterministic deduplication key by hashing the natural identity of each record, and reserve that key (here in Redis with a TTL matching the reconciliation window) before committing. Wrap the database writes in one transaction so a partially applied chunk can never leave orphaned rows.

# ingest/tasks.py
import hashlib
import json
import redis
from celery import shared_task
from django.db import transaction
from ingest.models import Permit

_dedup = redis.Redis(host="cache.internal", port=6379, db=3)
DEDUP_TTL = 60 * 60 * 30  # 30h — outlives the overnight window + morning reconciliation

def _idempotency_key(record: dict[str, str], jurisdiction_fips: str) -> str:
    basis = f"{record['submission_id']}|{record['permit_type_code']}|{jurisdiction_fips}|{record['filing_timestamp']}"
    return "permit:imp:" + hashlib.sha256(basis.encode()).hexdigest()

@shared_task(
    bind=True,
    name="ingest.tasks.import_permit_chunk",
    max_retries=3,
    retry_backoff=True,        # exponential backoff...
    retry_jitter=True,         # ...with jitter to avoid a thundering herd on broker recovery
    acks_late=True,
)
def import_permit_chunk(self, records: list[dict], jurisdiction_fips: str, batch_id: str) -> dict:
    written, skipped = 0, 0
    try:
        with transaction.atomic():  # all-or-nothing: a crash rolls the whole chunk back
            for record in records:
                key = _idempotency_key(record, jurisdiction_fips)
                # SETNX: only the first delivery of this record wins.
                if not _dedup.set(key, self.request.id, nx=True, ex=DEDUP_TTL):
                    skipped += 1
                    continue
                Permit.objects.create(
                    audit_task_id=self.request.id,   # immutable link: row -> dispatch event
                    jurisdiction_fips=jurisdiction_fips,
                    batch_id=batch_id,
                    **_normalize(record),
                )
                written += 1
    except Exception as exc:
        # Roll forward through transient broker/db faults; exhausted retries dead-letter to quarantine.
        raise self.retry(exc=exc)
    return {"batch_id": batch_id, "written": written, "skipped": skipped}

Step 3: Run a Dedicated Overnight Worker Pool

Daytime lookups and overnight imports should never share a worker process. Start a worker that consumes only the ingest queues, sized so total concurrency stays at or below broker_pool_limit to prevent connection exhaustion. The prefork pool is the safe default for CPU-bound parsing and validation; reserve gevent/eventlet for the I/O-bound scrapers that feed the pipeline.

# Overnight worker — bound to ingest queues only, isolated from the daytime lookup pool.
celery -A permits worker \
  --queues permit.ingest.critical,permit.ingest.standard,permit.ingest.maintenance \
  --pool prefork \
  --concurrency 8 \
  --max-tasks-per-child 200 \   # recycle workers to cap memory creep on long batches
  --prefetch-multiplier 1 \     # one chunk in flight per worker — fair priority handling
  --hostname overnight@%h

Schedule the run with Celery beat so dispatch only fires inside the maintenance window, and register a shutdown signal so in-flight chunks finish (or roll back) cleanly when the window closes.

# permits/celery.py
from celery.schedules import crontab
from celery.signals import worker_shutting_down

app.conf.beat_schedule = {
    "overnight-permit-import": {
        "task": "ingest.tasks.kickoff_nightly_imports",
        "schedule": crontab(hour=1, minute=0),   # 01:00 local maintenance window
        "options": {"queue": "permit.ingest.maintenance"},
    },
}

@worker_shutting_down.connect
def drain_in_flight(sig, how, exitcode, **kwargs):
    # `how == "Warm"` lets the current chunk's atomic transaction commit or roll back
    # before the process exits — no half-written batches at window close.
    log.info("overnight worker draining", extra={"signal": sig, "mode": how})

Parameter and Flag Reference

Setting Recommended value Rationale for overnight permit imports
x-message-ttl 1–4 h per queue Stale or malformed payloads expire into quarantine instead of blocking the window.
x-dead-letter-exchange permit.ingest.dlx Routes failures to a reviewable queue; nothing is silently dropped.
x-max-priority 10 Lets fee-bearing and zoning work preempt routine backfill.
task_acks_late True Redeliver after a crash — safe only because each chunk is idempotent.
task_reject_on_worker_lost True A SIGKILLed worker’s message is requeued, not lost.
--concurrency ≤ broker_pool_limit Prevents broker connection starvation during bulk dispatch.
--prefetch-multiplier 1 One chunk per worker keeps priority ordering honest.
--max-tasks-per-child 100–500 Recycles workers to bound memory growth across a long batch.
retry_backoff + retry_jitter True / True Exponential backoff with jitter avoids a thundering herd on broker recovery.
CHUNK_SIZE 500 Caps per-task memory on constrained municipal VMs.

Common Failure Patterns and Fixes

Duplicate rows from broker redelivery

With acks_late=True, any worker crash redelivers the in-flight chunk. Without the deduplication reservation in Step 2, the redelivery re-inserts every record. Fix: reserve the idempotency key with an atomic Redis SET ... NX before the insert, or enforce a UNIQUE constraint on the hashed key at the database level so the second attempt fails closed rather than duplicating.

Connection exhaustion during fan-out

A dispatcher that publishes tens of thousands of chunks can exhaust the broker connection pool when --concurrency exceeds broker_pool_limit. Fix: keep concurrency at or below broker_pool_limit, and batch-publish with a shared producer connection rather than opening one per apply_async.

Cross-jurisdiction contamination

Chunking by row count alone can split one county’s records across batches that interleave with another’s, breaking downstream row-level security keyed on jurisdiction_fips. Fix: scope every chunk to a single jurisdiction (pass jurisdiction_fips explicitly, as in Step 2) and assert it again before the transaction commits.

A poison message stalls the window

A malformed legacy CSV chunk that raises on every retry can consume all three retry attempts and, without a TTL, sit redelivering until 05:00. Fix: the x-message-ttl plus DLX binding guarantees the message expires into permit.ingest.quarantine; alert on quarantine depth so a single bad export never silently consumes the maintenance window. Wire this into the broader error handling and retry logic for ingestion pipelines so retries and quarantines share one policy.

Memory creep on multi-hour batches

Long-lived prefork children accumulate fragmented memory parsing large exports and get OOM-killed mid-chunk. Fix: set --max-tasks-per-child to recycle workers periodically and stream rows with a generator (Step 2) instead of materializing the whole file.

Audit and Logging Guidance

Municipal pipelines must satisfy state records-retention laws and federal data-handling standards, so every task execution emits a structured JSON log that a compliance officer can reconstruct months later. Cross-reference the Celery task_id with the database audit_task_id written in Step 2 to rebuild complete data lineage for any imported permit.

# Emit one structured record per chunk; ship to WORM object storage / Splunk / Elasticsearch.
log.info("chunk.committed", extra={
    "task_id": self.request.id,        # joins to Permit.audit_task_id for lineage
    "batch_id": batch_id,              # joins every chunk back to one nightly run
    "jurisdiction_fips": jurisdiction_fips,
    "records_written": written,
    "records_skipped_dedup": skipped,
    "duration_ms": elapsed_ms,
    "disposition": "committed",        # committed | retried | quarantined
})

Log, at minimum: the task_id and batch_id, jurisdiction code, record counts (written vs. deduplicated), processing duration, and final disposition. Route these to an immutable, append-only store (object storage under a WORM policy, or Splunk/Elasticsearch) so the trail cannot be edited after the fact. Track quarantine-queue depth and retry rates as first-class metrics — a rising quarantine count is the earliest signal that a county changed its export schema. For the alerting layer that turns these signals into pages, see logging and alerting strategies for failed CSV parsing jobs.

Related in this area

Related Choosing Celery vs asyncio for Permit Batch Jobs