Routing failed payloads to quarantine and a dead-letter queue #

When a maritime payload fails validation, the expensive mistake is treating every failure the same. Routing failed payloads to quarantine and a dead-letter queue is the decision layer that inspects each rejected interchange and sends it to exactly one destination: a dead-letter queue for unrecoverable structural corruption that only engineering can fix, or a quarantine topic for a parsed-but-non-compliant record that an ops team can reconcile against a refreshed registry and replay. Collapse those two paths into one and you either drop legally valid Bills of Lading or drown your on-call engineers in recoverable business exceptions. This page specifies the classifier, the immutable audit record, and the replay loop that keep the two failure modes strictly separated.

Architecture Alignment #

This routing layer is the failure half of the gate defined by Schema Validation Frameworks, the reference that owns synchronous structural, semantic, and regulatory checks inside the wider Document Ingestion & EDI Parsing Workflows domain. The validation tiers decide whether a payload passes; this component decides where a failure goes once it does not. It runs immediately after a tier raises, and it is deliberately narrow: it classifies an exception into a routing verdict, writes one immutable audit record, and publishes to a broker destination — nothing else. Durable delivery, retry backoff, and worker back-pressure for those destinations belong to Async Batch Processing Pipelines, and the concrete Redis dead-letter exchange and quarantine topic are provisioned by building Celery queues for maritime doc ingestion. This page owns only the classification and audit contract that sits between them.

Prerequisites & Environment Setup #

The routing layer targets Python 3.11+ for enum.StrEnum, exception groups, and datetime.UTC. Install the broker client, structured logging, and typed-model stack:

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

Every routing decision must be reconstructable during a customs or port-state-control audit, so bare print() is unacceptable — each verdict is an event queried later by correlation_id. Configure structlog once at worker boot to emit JSON:

import structlog

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

The router reads its destinations and its poison cap from the environment so the same code runs against RabbitMQ or Redis without edits:

Step-by-step Implementation #

Step 1 — Fix the error taxonomy #

The routing decision is only as sound as the exception taxonomy behind it. Two families exist, and they map to two destinations. Structural failures mean the bytes never became a record: a broken UN/EDIFACT envelope with no UNB/UNZ pairing, an undecodable payload, or a pydantic.ValidationError on a mandatory segment such as a missing BGM. No human can reconcile these without engineering, so they go to the dead-letter queue. Semantic and regulatory failures mean the record parsed cleanly but breaks a business or compliance rule: an unknown UN/LOCODE, an ISO 6346 container check-digit that drifted under OCR, a stale HS/HTS tariff, a SOLAS Verified Gross Mass below the plausibility floor, or a non-standard NAD qualifier like NAD+CZ where a handler expected NAD+CN. These are recoverable against a refreshed registry and route to quarantine.

from __future__ import annotations

import enum


class FailureClass(enum.StrEnum):
    STRUCTURAL = "STRUCTURAL"     # envelope broken / mandatory segment absent
    SEMANTIC = "SEMANTIC"         # unknown LOCODE, ISO 6346 drift, bad NAD qualifier
    REGULATORY = "REGULATORY"     # stale tariff, VGM below SOLAS floor
    UNKNOWN = "UNKNOWN"           # unclassified — fail safe, do not drop


# Structural exceptions are the only ones that reach the DLQ. Everything a human
# can reconcile against a registry stays out of engineering's queue.
STRUCTURAL_EXC: tuple[type[Exception], ...] = (
    ValueError,            # pydantic.ValidationError subclasses ValueError
    UnicodeDecodeError,    # payload never decoded
    EOFError,              # truncated interchange / missing UNZ trailer
)

Step 2 — Build the immutable audit record #

Before any payload moves, the router writes one append-only audit record so the decision is replayable even if the broker later loses the message. The record is keyed on the correlation_id plus the SHA-256 of the raw bytes: the correlation id ties the failure to its intake event, and the payload hash makes the record content-addressed so a replay of identical bytes collapses onto the same audit row instead of forging a second history. The model is frozen — an audit record is never mutated, only superseded by a later record sharing the same correlation id.

from __future__ import annotations

import hashlib
from datetime import UTC, datetime

from pydantic import BaseModel, ConfigDict, Field


class AuditRecord(BaseModel):
    """Append-only receipt for one failed-payload routing decision."""

    model_config = ConfigDict(frozen=True)

    correlation_id: str
    payload_sha256: str = Field(min_length=64, max_length=64)
    failure_class: str
    destination: str
    rule_field: str | None = None
    detail: str = ""
    attempt: int = Field(ge=1)
    recorded_at: datetime = Field(default_factory=lambda: datetime.now(UTC))

    @classmethod
    def from_payload(cls, raw: bytes, **kw: object) -> "AuditRecord":
        digest = hashlib.sha256(raw).hexdigest()
        return cls(payload_sha256=digest, **kw)  # type: ignore[arg-type]

Step 3 — Classify the failure and route the verdict #

route_failure() is the whole decision. It maps a caught exception to a FailureClass, enforces the poison-message cap first so a record that has looped too many times is parked rather than requeued forever, writes the audit record, and returns the broker destination. Note the fail-safe: an exception that matches nothing known is treated as UNKNOWN and sent to quarantine for a human, never silently dropped — a rule the validation framework shares, because an unclassified failure is a gap in the taxonomy, not a licence to discard cargo data.

from __future__ import annotations

import os

import structlog

log = structlog.get_logger("maritime.routing")

DLX = os.environ.get("INGEST_DLX", "edi.dlx")
QUARANTINE = os.environ.get("INGEST_QUARANTINE_TOPIC", "edi.quarantine")
AUDIT_HOLD = os.environ.get("INGEST_AUDIT_HOLD", "edi.audit.hold")
POISON_MAX = int(os.environ.get("POISON_MAX_ATTEMPTS", "5"))


def _classify(exc: Exception) -> FailureClass:
    if isinstance(exc, STRUCTURAL_EXC):
        return FailureClass.STRUCTURAL
    if isinstance(exc, (KeyError, LookupError)):
        return FailureClass.SEMANTIC     # unknown LOCODE / NAD qualifier lookup miss
    if isinstance(exc, PermissionError):
        return FailureClass.REGULATORY   # tariff / VGM rule breach
    return FailureClass.UNKNOWN


def route_failure(
    raw: bytes,
    exc: Exception,
    *,
    correlation_id: str,
    attempt: int,
    field: str | None = None,
) -> AuditRecord:
    log.bind(correlation_id=correlation_id, attempt=attempt)
    failure = _classify(exc)

    # Poison-message cap wins over class: a record that keeps failing is parked.
    if attempt >= POISON_MAX:
        destination = AUDIT_HOLD
        log.error("poison_cap_reached", failure_class=failure, attempt=attempt)
    elif failure is FailureClass.STRUCTURAL:
        destination = DLX
        log.error("dead_letter_routed", failure_class=failure, detail=str(exc))
    else:
        destination = QUARANTINE
        log.warning("quarantine_routed", failure_class=failure, field=field)

    record = AuditRecord.from_payload(
        raw,
        correlation_id=correlation_id,
        failure_class=str(failure),
        destination=destination,
        rule_field=field,
        detail=str(exc)[:500],
        attempt=attempt,
    )
    return record

Step 4 — Publish with the audit record attached, then reconcile and replay #

The verdict is worthless until it is durably published, and the audit receipt must ride with the message so ops can act on it from the quarantine topic alone. The publisher below is idempotent on the payload hash, so an at-least-once broker redelivery does not create a duplicate quarantine entry. Reconciliation reverses the flow: an ops tool pulls a quarantine record, resolves the offending field — refreshes the UN/LOCODE snapshot, corrects the ISO 6346 check digit, remaps the NAD qualifier — and replays the corrected payload back to intake with its original correlation_id and an incremented attempt. That single loop is what lets a stale-registry failure clear itself once the async batch pipeline refreshes its cache, without engineering ever touching the message.

from __future__ import annotations

import json

import structlog

log = structlog.get_logger("maritime.routing")


def publish_failure(broker, record: AuditRecord, raw: bytes) -> None:
    audit_store.upsert(record)  # keyed on (correlation_id, payload_sha256): replays no-op
    broker.publish(
        record.destination,
        raw,
        headers={
            "correlation_id": record.correlation_id,
            "payload_sha256": record.payload_sha256,
            "failure_class": record.failure_class,
            "attempt": record.attempt,
            "audit": json.loads(record.model_dump_json()),
        },
    )
    log.info(
        "failure_published",
        correlation_id=record.correlation_id,
        destination=record.destination,
        attempt=record.attempt,
    )


def replay_from_quarantine(broker, record: AuditRecord, raw: bytes) -> None:
    # Ops has reconciled the offending field; send corrected bytes back to intake.
    broker.publish(
        "edi.intake",
        raw,
        headers={"correlation_id": record.correlation_id, "attempt": record.attempt + 1},
    )
    log.info("quarantine_replayed", correlation_id=record.correlation_id)

Edge Cases & Carrier Deviations #

The subtlest deviation is the non-standard NAD qualifier. A carrier that emits NAD+CZ (consignee’s agent) where your handler expects NAD+CN (consignee) produces a lookup miss, not corrupt bytes. The document is legally valid EDIFACT; classifying that miss as STRUCTURAL and dead-lettering it would strand a shippable consignment behind an engineering ticket. It belongs in quarantine, where an ops analyst maps the qualifier and replays — the exact behaviour the _classify LookupError branch encodes.

Verification & Testing #

Assert that structural faults dead-letter, recoverable faults quarantine, and the poison cap parks a looping record. Use byte fixtures so the SHA-256 keying is exercised, and structlog’s capture helper to assert the audit event fires with the right destination.

import pytest
import structlog

from maritime_routing.router import FailureClass, route_failure


@pytest.fixture
def raw() -> bytes:
    return b"UNB+UNOA:2'UNH+1+IFCSUM:D:96A:UN'"


def test_structural_goes_to_dlq(raw: bytes) -> None:
    rec = route_failure(raw, ValueError("missing BGM"), correlation_id="c1", attempt=1)
    assert rec.failure_class == FailureClass.STRUCTURAL
    assert rec.destination.endswith("dlx")


def test_unknown_locode_quarantines(raw: bytes) -> None:
    rec = route_failure(raw, KeyError("USNYC"), correlation_id="c2", attempt=1, field="LOC")
    assert rec.destination.endswith("quarantine")


def test_poison_cap_parks_record(raw: bytes) -> None:
    rec = route_failure(raw, KeyError("USNYC"), correlation_id="c3", attempt=5)
    assert rec.destination.endswith("audit.hold")


def test_audit_event_shape(raw: bytes) -> None:
    cap = structlog.testing.LogCapture()
    structlog.configure(processors=[cap])
    route_failure(raw, ValueError("broken envelope"), correlation_id="c4", attempt=1)
    assert cap.entries[-1]["event"] == "dead_letter_routed"

A passing run emits one JSON line per verdict; a dead-letter decision looks like {"event": "dead_letter_routed", "correlation_id": "c4", "failure_class": "STRUCTURAL", "attempt": 1, "log_level": "error", "timestamp": "..."}. Assert on the structured keys, never on a formatted message string.

Frequently Asked Questions #

Related #

• Schema Validation Frameworks — the structural, semantic, and regulatory tiers whose failures this layer routes
• Async Batch Processing Pipelines — durable delivery, retry backoff, and back-pressure for the destinations here
• Building Celery queues for maritime doc ingestion — the concrete DLX and quarantine topic this router publishes to
• Document Ingestion & EDI Parsing Workflows — the domain governing ingestion, schema, and resilience contracts

Up: Schema Validation Frameworks — the validation gate this failure-routing layer completes.