Async bulk B/L ingestion with Celery and a Redis DLQ #

Carriers and freight forwarders rarely hand you one Bill of Lading at a time — they drop a consolidated file: a UN/EDIFACT IFCSUM manifest carrying hundreds of consignments, or a day’s B/L export as a single multi-thousand-line payload. Async bulk B/L ingestion with Celery and a Redis dead-letter queue is the pattern that turns that monolithic drop into a fan-out of independent per-B/L tasks, aggregates the survivors in one idempotent callback, and quarantines the handful of unrecoverable consignments to a Redis list instead of losing them or blocking the whole batch. This page specifies the fan-out/fan-in mechanics and the manual DLQ that Celery itself does not provide, and it deliberately assumes the queue, routing, and worker-pool wiring from Building Celery queues for maritime doc ingestion is already in place.

Architecture Alignment #

This task lives in the Async Batch Processing Pipelines area of the Document Ingestion & EDI Parsing Workflows domain, and it occupies a narrow slice of it: the moment a single enqueued batch has to explode into many units of work and then collapse back into one result. Where the sibling queue page defines how a payload is validated and routed onto edi.parse, this page owns what happens once thousands of B/Ls need to be processed under one batch identity — Celery’s chord and group primitives for fan-out, acks_late with a tuned broker visibility timeout so a redelivery never double-processes a consignment, and a hand-built Redis dead-letter queue because Celery has no native DLQ construct. The business-versus-corruption distinction that decides what gets dead-lettered is defined once in routing failed payloads to quarantine and DLQ; here we only implement the transport for the unrecoverable tail.

Prerequisites & Environment Setup #

The pipeline targets Python 3.11+ for tomllib, exception groups, and the faster asyncio scheduler. Install Celery with its Redis transport, the raw redis client (the DLQ is built directly on redis-py list commands, not through Celery), structured logging, and typed validation:

pip install "celery[redis]==5.4.*" "redis>=5.0" structlog pydantic

Structured JSON logging is non-negotiable in this niche: every fan-out, every dead-letter, and every replay must be queryable by batch_id and B/L number for a customs or port-state-control audit, so bare print() and unstructured logging strings are unacceptable. Configure structlog once, at worker boot:

import structlog

structlog.configure(
    processors=[
        structlog.contextvars.merge_contextvars,
        structlog.processors.add_log_level,
        structlog.processors.TimeStamper(fmt="iso", utc=True),
        structlog.processors.JSONRenderer(),
    ],
)
log = structlog.get_logger("maritime.bulk")

The workers and the replay job read the following environment:

Because the chord result backend must survive a worker restart to fire its callback, point CELERY_RESULT_BACKEND at a persistent Redis database, never at RPC. Run the batch dispatcher and the per-B/L parsers on separate queues (-Q bulk.dispatch,bl.parse) so a 4,000-line manifest split never starves the parse pool.

Step-by-step Implementation #

Step 1 — Split the bulk manifest into per-B/L units of work #

A consolidated IFCSUM manifest repeats a CNI consignment-information group once per Bill of Lading, and the B/L reference itself rides in an RFF+BM segment inside that group. The split stage streams the file into one immutable BLUnit per consignment, computing a content-addressed idempotency key so a replayed or duplicated batch never re-books the same B/L. It decodes nothing downstream depends on and never mutates a segment in place — the raw text travels with the unit for full audit reproducibility.

from __future__ import annotations

import hashlib
import re
from collections.abc import Iterator
from dataclasses import dataclass

import structlog

log = structlog.get_logger("maritime.bulk.split")
_BL_REF = re.compile(r"RFF\+BM:([A-Za-z0-9]+)")


@dataclass(frozen=True, slots=True)
class BLUnit:
    """One Bill of Lading carved out of a consolidated manifest."""

    bl_number: str
    raw_segment: str
    idempotency_key: str


def split_bulk_manifest(raw: bytes, *, batch_id: str) -> Iterator[BLUnit]:
    text: str = raw.decode("utf-8-sig", errors="replace")
    consignments: list[str] = text.split("CNI+")[1:]  # drop the interchange header
    for index, chunk in enumerate(consignments, start=1):
        segment: str = f"CNI+{chunk}"
        match = _BL_REF.search(segment)
        bl_number: str = match.group(1) if match else f"{batch_id}-{index:05d}"
        key: str = hashlib.sha256(
            f"{batch_id}:{bl_number}:{segment}".encode()
        ).hexdigest()
        log.info("bl_unit_extracted", batch_id=batch_id, bl_number=bl_number, position=index)
        yield BLUnit(bl_number=bl_number, raw_segment=segment, idempotency_key=key)

Step 2 — Fan out with a Celery chord over a group of parse tasks #

Fan-out is a chord: a group header of one parse_bl signature per BLUnit, plus a single callback that runs only after every header task settles. The batch-level settings matter as much as the topology. task_acks_late=True holds each B/L on the broker until its task succeeds, so a worker crash mid-batch redelivers just that consignment rather than silently dropping it; the paired visibility_timeout must exceed the slowest realistic parse_bl runtime, or Redis will redeliver a still-running task and you will process a B/L twice. This is the batch-scale counterpart to the queue configuration in the async batch processing pipelines reference.

import os

from celery import Celery, chord, group, shared_task

app = Celery(
    "maritime_bulk",
    broker=os.environ["CELERY_BROKER_URL"],
    backend=os.environ["CELERY_RESULT_BACKEND"],
)
app.conf.update(
    task_acks_late=True,
    task_reject_on_worker_lost=True,
    worker_prefetch_multiplier=1,
    broker_transport_options={
        "visibility_timeout": int(os.environ.get("BULK_VISIBILITY_TIMEOUT", "3600")),
    },
)


@shared_task(name="bulk.dispatch", queue="bulk.dispatch")
def dispatch_bulk_manifest(raw: bytes, batch_id: str) -> str:
    header = group(
        parse_bl.s(unit.raw_segment, unit.bl_number, unit.idempotency_key)
        for unit in split_bulk_manifest(raw, batch_id=batch_id)
    )
    chord(header)(aggregate_batch.s(batch_id=batch_id))
    log.info("batch_dispatched", batch_id=batch_id)
    return batch_id

Step 3 — Dead-letter unrecoverable B/Ls to a Redis list #

Celery ships no dead-letter primitive — task_reject_on_worker_lost requeues, and a raised exception in a chord header fails the entire chord, so one malformed consignment would sink an otherwise clean 3,000-B/L batch. The fix is a manual DLQ built on a Redis list: transient faults (a broker blip, a registry timeout) retry with exponential backoff, but a genuinely unrecoverable B/L is LPUSHed onto the DLQ list and the task returns a sentinel result so the chord still completes. The ISO 6346 container check-digit and RFF+BM B/L-number validation happen inside _parse_one_bl; only its unrecoverable failures reach the DLQ.

import json
import time
from typing import Any

import redis

_redis: redis.Redis = redis.Redis.from_url(os.environ["CELERY_BROKER_URL"])
DLQ_KEY: str = os.environ.get("BULK_DLQ_KEY", "bl.dlq")
MAX_BACKOFF: int = int(os.environ.get("BULK_MAX_BACKOFF", "600"))


def push_to_dlq(*, unit_key: str, bl_number: str, raw_segment: str, reason: str) -> None:
    record: str = json.dumps(
        {
            "idempotency_key": unit_key,
            "bl_number": bl_number,
            "raw_segment": raw_segment,
            "reason": reason,
            "dead_lettered_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
        }
    )
    _redis.lpush(DLQ_KEY, record)  # BRPOP on the replay side makes this FIFO
    log.error("bl_dead_lettered", bl_number=bl_number, reason=reason, dlq_key=DLQ_KEY)


@shared_task(bind=True, name="bulk.parse_bl", queue="bl.parse", max_retries=4, acks_late=True)
def parse_bl(self, raw_segment: str, bl_number: str, idempotency_key: str) -> dict[str, Any]:
    structlog.contextvars.bind_contextvars(bl_number=bl_number)
    try:
        record: dict[str, Any] = _parse_one_bl(raw_segment, bl_number)
        return {"status": "ok", "bl_number": bl_number, "record": record}
    except (ConnectionError, TimeoutError) as exc:  # transient — retry with backoff
        delay: int = min(2 ** self.request.retries * 10, MAX_BACKOFF)
        log.warning("bl_retry_transient", delay=delay, attempt=self.request.retries)
        raise self.retry(exc=exc, countdown=delay)
    except Exception as exc:  # noqa: BLE001 — unrecoverable for this consignment
        push_to_dlq(
            unit_key=idempotency_key, bl_number=bl_number,
            raw_segment=raw_segment, reason=str(exc),
        )
        return {"status": "dead_lettered", "bl_number": bl_number, "reason": str(exc)}

Step 4 — Aggregate results in an idempotent chord callback #

The callback receives the ordered list of every header result and must be safe to run more than once — a redelivered chord or a manually replayed batch can fire it twice. Guard it with a Redis SET ... NX: the first invocation claims the batch and forwards the normalised records to the terminal operating system; a duplicate short-circuits to a no-op. Aggregation simply partitions the ok records from the dead_lettered sentinels, so the DLQ count is visible in one structured log line rather than scattered across per-task events.

@shared_task(bind=True, name="bulk.aggregate", queue="bulk.dispatch")
def aggregate_batch(self, results: list[dict[str, Any]], *, batch_id: str) -> dict[str, int]:
    claim_key: str = f"bulk:aggregated:{batch_id}"
    if not _redis.set(claim_key, "1", nx=True, ex=86_400):
        log.info("aggregate_skipped_duplicate", batch_id=batch_id)
        return {"batch_id": batch_id, "skipped": 1}

    normalised: list[dict[str, Any]] = [r for r in results if r.get("status") == "ok"]
    dead: list[dict[str, Any]] = [r for r in results if r.get("status") == "dead_lettered"]
    for item in normalised:
        _forward_to_tos(item["record"])  # keyed on B/L number, itself idempotent

    log.info(
        "batch_aggregated",
        batch_id=batch_id,
        total=len(results),
        forwarded=len(normalised),
        dead_lettered=len(dead),
    )
    return {"batch_id": batch_id, "forwarded": len(normalised), "dead_lettered": len(dead)}

Step 5 — Drain the DLQ with a replay worker #

Dead-lettered B/Ls are not lost work — they are deferred work awaiting a fix (a deployed handler for a drifted CNI layout, a refreshed UN/LOCODE registry). A scheduled replay worker pops items FIFO with BRPOP (paired with the LPUSH producer) and re-enqueues each one through parse_bl, bounded per run so a poison record cannot spin forever. Because every unit carries its original idempotency_key, a replay that succeeds this time still cannot double-book a consignment downstream.

def replay_dlq(*, block_seconds: int = 5, max_items: int = 200) -> int:
    replayed: int = 0
    while replayed < max_items:
        popped: tuple[bytes, bytes] | None = _redis.brpop(DLQ_KEY, timeout=block_seconds)
        if popped is None:
            break  # list drained within the blocking window
        _, raw = popped
        record: dict[str, Any] = json.loads(raw)
        parse_bl.delay(record["raw_segment"], record["bl_number"], record["idempotency_key"])
        replayed += 1
        log.info("dlq_item_replayed", bl_number=record["bl_number"])
    log.info("dlq_replay_batch_complete", replayed=replayed, remaining=_redis.llen(DLQ_KEY))
    return replayed

Edge Cases & Carrier Deviations #

A recurring carrier deviation is RFF+BM drift: some lines emit the B/L number under RFF+BN (booking reference) or omit RFF+BM from a house-B/L consignment entirely. The split stage falls back to a synthetic batch_id-derived key rather than crashing, but a missing carrier B/L reference is a reconciliation flag — dead-letter the unit with a precise reason so an ops team can supply the number, never drop it, because a house Bill of Lading with no visible reference is still a legally binding contract of carriage.

Verification & Testing #

Assert the two behaviours that are easy to get wrong: the split emits one unit per consignment with stable keys, and an unrecoverable parse dead-letters instead of raising. Use fakeredis so the DLQ list is exercised without a live broker, and structlog’s capture helper to confirm the audit event fires.

import fakeredis
import pytest
import structlog

import maritime_bulk.tasks as tasks


@pytest.fixture(autouse=True)
def fake_redis(monkeypatch: pytest.MonkeyPatch) -> fakeredis.FakeRedis:
    fake = fakeredis.FakeRedis()
    monkeypatch.setattr(tasks, "_redis", fake)
    return fake


IFCSUM = (
    b"UNH+1+IFCSUM:D:96A:UN'"
    b"CNI+1'RFF+BM:MAEU12345'"
    b"CNI+2'RFF+BM:MAEU67890'"
)


def test_split_yields_one_unit_per_consignment() -> None:
    units = list(tasks.split_bulk_manifest(IFCSUM, batch_id="B-1"))
    assert [u.bl_number for u in units] == ["MAEU12345", "MAEU67890"]
    assert len({u.idempotency_key for u in units}) == 2  # keys are distinct


def test_unrecoverable_bl_is_dead_lettered(fake_redis: fakeredis.FakeRedis) -> None:
    tasks.push_to_dlq(
        unit_key="k1", bl_number="MAEU12345",
        raw_segment="CNI+1'", reason="unknown CNI layout",
    )
    assert fake_redis.llen(tasks.DLQ_KEY) == 1


def test_replay_drains_the_dlq(fake_redis: fakeredis.FakeRedis) -> None:
    tasks.push_to_dlq(unit_key="k1", bl_number="MAEU12345", raw_segment="CNI+1'", reason="x")
    assert tasks.replay_dlq(block_seconds=1, max_items=10) == 1
    assert fake_redis.llen(tasks.DLQ_KEY) == 0

A passing run emits one JSON line per dead-letter and per replay; the dead-letter looks like {"event": "bl_dead_lettered", "bl_number": "MAEU12345", "reason": "unknown CNI layout", "dlq_key": "bl.dlq", "log_level": "error", "timestamp": "..."}. Assert on the structured keys, never on a formatted message string.

Frequently Asked Questions #

Related #

• Building Celery queues for maritime doc ingestion — the queue, routing, and worker-pool wiring this fan-out runs on top of
• Routing failed payloads to quarantine and DLQ — the rules that decide what is unrecoverable enough to dead-letter
• Async Batch Processing Pipelines — the throughput, backpressure, and lifecycle contracts governing the batch
• IFCSUM EDI Message Parsing — the consignment-group segment structure the split stage carves apart
• Document Ingestion & EDI Parsing Workflows — the domain framework for maritime document ingestion

Up: Async Batch Processing Pipelines — the topic area governing asynchronous document throughput.