• 14 min read
• 5 code samples

Error Handling & Retry Logic

In production customs brokerage pipelines, document ingestion is rarely deterministic. Trade documents arrive with structural inconsistencies, OCR artifacts, and jurisdiction-specific formatting variations that directly impact downstream HS code classification. For trade compliance officers and licensed customs brokers, a single misclassified line item can trigger regulatory penalties; for logistics developers and Python ETL teams, an unhandled parsing failure cascades into a data-pipeline deadlock. This page addresses one engineering gap in the Document Ingestion & Parsing Workflows domain: how do you keep a high-throughput clearance pipeline running when individual documents fail — without retrying failures that can never succeed, without duplicating declarations, and without losing the audit trail? The answer is a strict failure taxonomy, idempotent retries bounded by full-jitter backoff, a circuit breaker that pauses cleanly, and a dead-letter queue that preserves a forensic chain of custody.

# Problem framing: what actually fails, and how it hurts compliance

Customs documentation pipelines transform payloads across several stages, each introducing distinct failure surfaces. During Commercial Invoice PDF Extraction, malformed PDFs, image-only pages, or non-standard font encodings cause parser timeouts and schema-validation failures. When those documents proceed to Packing List Data Normalization, unit-of-measure mismatches, missing net/gross weight declarations, or inconsistent SKU mappings generate classification errors downstream. A resilient pipeline must capture these failures at the record level, preserve the original payload for forensic review, and route exceptions to a dead-letter queue rather than terminating the batch.

The single decision that governs everything below is whether a failure is transient or permanent — because that decides whether the record is retried or escalated:

• Transient errors — database connection drops, external classification-API throttling, momentary OCR-engine unavailability, or network timeouts. These are re-tryable with jittered exponential backoff; the same document will likely succeed on a later attempt.
• Permanent errors — fundamentally unparseable document structures, missing mandatory customs fields (country of origin, declared value, currency), or invalid HTS/HS syntax. These will fail identically on every attempt, so they bypass the retry loop entirely and escalate to human-in-the-loop validation.

Conflating the two is the most common and most expensive mistake in a customs ETL pipeline: retrying a permanent error wastes worker capacity during a clearance window, while treating a transient throttle as permanent dumps recoverable documents into manual review and inflates broker labor.

# Schema / data contract

Before any retry logic runs, the pipeline needs a formal contract for what a payload is and what an error means. Two dataclasses and an enum pin this down: the immutable payload snapshot that makes retries idempotent, and the failure classification that steers routing. Every attempt carries a deterministic document_id and a versioned attempt_id, so a re-run can be reconciled against prior history rather than creating a second customs entry.

from __future__ import annotations

import uuid
from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Optional


class FailureType(Enum):
    TRANSIENT = "transient"   # retry with backoff
    PERMANENT = "permanent"   # route to DLQ, escalate to human review


@dataclass(frozen=True)
class CustomsPayload:
    """Immutable ingestion envelope. Freezing the record guarantees a retry
    replays the *same* bytes, which is what makes the pipeline idempotent."""
    document_id: str                       # deterministic, stable across retries
    raw_bytes: bytes                        # original document, archived for audit
    source_system: str                     # forwarder / portal that submitted it
    metadata: dict[str, Any] = field(default_factory=dict)
    attempt_id: str = field(default_factory=lambda: str(uuid.uuid4()))


class CustomsETLError(Exception):
    """Base error carrying the classification and the HS/HTS code in scope."""
    def __init__(self, message: str, failure_type: FailureType,
                 hts_code: Optional[str] = None) -> None:
        super().__init__(message)
        self.failure_type = failure_type
        self.hts_code = hts_code


class TransientParsingError(CustomsETLError):
    def __init__(self, message: str, hts_code: Optional[str] = None) -> None:
        super().__init__(message, FailureType.TRANSIENT, hts_code)


class PermanentClassificationError(CustomsETLError):
    def __init__(self, message: str, hts_code: Optional[str] = None) -> None:
        super().__init__(message, FailureType.PERMANENT, hts_code)

The contract is deliberately narrow: an exception is either a TransientParsingError or a PermanentClassificationError, and there is no third state. Any exception the pipeline does not explicitly classify is treated as permanent and sent to the dead-letter queue, so an unknown failure never enters an unbounded retry loop by default.

# Step-by-step implementation

The retry controller is built in three stages. Each stage has a single responsibility, explicit inputs and outputs, and a defined error condition.

# Stage 1 — Validate at the extraction boundary

Structural validation runs before any retry decision, so a malformed HS code is caught as permanent immediately rather than burning retry attempts. HTS/HS codes are legal only at 6, 8, or 10 digits per WCO nomenclature — never 4, 7, or 9, and never with dot separators left in.

• Input: the raw HTS/HS string extracted from the document.
• Output: a boolean gate; a false result raises PermanentClassificationError.
• Error condition: non-digit content, a dot-separated code that was not stripped, or an illegal digit length.

def validate_hts_structure(hts_code: Optional[str]) -> bool:
    """Strict HTS/HS length validation per WCO HS 2022 nomenclature.

    Accepts only the legal lengths: 6-digit international HS plus 8- or
    10-digit national extensions. A 4-digit heading, a 7/9-digit code, or a
    dot-separated code that has not been normalized is rejected outright."""
    if not hts_code or not hts_code.isdigit():
        return False
    return len(hts_code) in (6, 8, 10)

# Stage 2 — Process one document idempotently

The processor is wrapped in a tenacity retry that fires only on TransientParsingError. A PermanentClassificationError propagates out immediately without a single retry, because retry_if_exception_type does not match it.

• Input: a frozen CustomsPayload.
• Output: a classified record dict on success.
• Error condition: transient failures are retried up to the cap; permanent failures raise immediately; exhausted retries raise RetryError.

import logging

from tenacity import (
    retry,
    stop_after_attempt,
    wait_exponential_jitter,
    retry_if_exception_type,
    before_sleep_log,
    RetryError,
)

logger = logging.getLogger("customs.etl.retry_controller")


@retry(
    retry=retry_if_exception_type(TransientParsingError),
    stop=stop_after_attempt(4),                             # 1 try + 3 retries
    wait=wait_exponential_jitter(initial=1, max=30, jitter=1),  # full jitter
    before_sleep=before_sleep_log(logger, logging.WARNING),
    reraise=True,
)
def process_customs_document(payload: CustomsPayload) -> dict[str, Any]:
    """Idempotent document processor with an explicit HTS validation gate.

    Because the deterministic document_id keys the downstream commit, replaying
    the same payload never files a second customs declaration."""
    logger.info(
        "processing document",
        extra={"doc_id": payload.document_id, "attempt_id": payload.attempt_id},
    )

    extracted = extract_invoice_fields(payload.raw_bytes)

    # HTS/HS compliance gate — a structural failure here is PERMANENT.
    hts = extracted.get("hs_code")
    if not validate_hts_structure(hts):
        raise PermanentClassificationError(
            f"invalid HTS structure: {hts!r}", hts_code=hts
        )

    # Low OCR confidence is a TRANSIENT surface: a re-extraction pass may recover.
    if extracted.get("ocr_confidence", 0.0) < 0.85:
        raise TransientParsingError("OCR confidence below threshold", hts_code=hts)

    return {
        "status": "classified",
        "hs_code": hts,
        "doc_id": payload.document_id,
        "hts_validation_state": "valid",
    }


def extract_invoice_fields(raw_bytes: bytes) -> dict[str, Any]:
    """Placeholder for the real PDF/OCR extraction. hs_code must be digit-only
    (dot separators stripped) so validate_hts_structure passes correctly."""
    return {"hs_code": "8517620000", "ocr_confidence": 0.92}

# Stage 3 — Route exhausted or permanent failures to the DLQ

The resilience wrapper is where retry outcomes turn into routing decisions. An exhausted transient retry surfaces as RetryError; a permanent error propagates directly; anything unclassified is defensively treated as permanent.

• Input: a CustomsPayload.
• Output: the classified record, or None when the document is dead-lettered.
• Error condition: every failure path ends in a DLQ record — no document is ever silently dropped.

def route_to_dlq(payload: CustomsPayload, error: CustomsETLError) -> None:
    """Persist a failed payload to the dead-letter queue with forensic metadata.

    Integration point: SQS DLQ, a Kafka dead-letter topic, or write-once storage.
    payload.raw_bytes MUST be archived for the CBP/audit chain of custody."""
    logger.critical(
        "dlq routing",
        extra={
            "doc_id": payload.document_id,
            "failure_type": error.failure_type.value,
            "hts_code": error.hts_code or "N/A",
            "reason": str(error),
        },
    )


def execute_with_resilience(payload: CustomsPayload) -> Optional[dict[str, Any]]:
    try:
        return process_customs_document(payload)
    except RetryError as exc:                      # transient retries exhausted
        last = exc.last_attempt.exception()
        route_to_dlq(payload, last if isinstance(last, CustomsETLError)
                     else CustomsETLError(str(last), FailureType.TRANSIENT))
        return None
    except PermanentClassificationError as exc:    # never retried, straight to DLQ
        route_to_dlq(payload, exc)
        return None
    except Exception as exc:                        # unknown → treat as permanent
        logger.exception("unhandled failure", extra={"doc_id": payload.document_id})
        route_to_dlq(payload, CustomsETLError(str(exc), FailureType.PERMANENT))
        return None

# Validation & determinism

Determinism is what makes this pipeline audit-defensible: the same document, replayed, must produce the same outcome and the same records.

• Idempotency key. Every attempt reuses the deterministic document_id, and the frozen CustomsPayload guarantees the same bytes are replayed. The downstream commit is a no-op if that document already produced an entry, so a redelivery after a broker failover reconciles against the existing declaration instead of filing a second one.
• HS digit-length rule. validate_hts_structure enforces the 6/8/10-digit constraint at the boundary, matching the same rule applied during Commercial Invoice PDF Extraction so a code that passes there cannot be silently downgraded here.
• Bounded backoff. wait_exponential_jitter spreads each retry across the full interval between the 1-second base and the 30-second cap. The exact curve, cap selection, and thundering-herd analysis are worked through in Designing exponential backoff for failed parsing jobs.
• Ambiguous classification. When a classification API returns an ambiguous match or needs additional line-item descriptors, the pipeline emits a structured classification_pending state rather than defaulting to a code — preserving audit integrity and preventing downstream duty miscalculations.
• Quarantine routing. Every failure path terminates in a DLQ record. A document is never dropped and never left in an indeterminate state.

# Downstream integration

Error handling is not a standalone concern; it is wired into the specific stages that produce failures, so recovery is targeted rather than a blunt full-document retry.

• In Async Batch Processing for High Volume, workers consume from partitioned queues so a single malformed invoice cannot block parallel clearance streams; the retry controller here is the standard those consumers inherit.
• When OCR Drift Correction & Validation detects character-substitution patterns (0/O, 1/I inside HS codes), the pipeline triggers a targeted re-extraction pass — a transient retry against the corrected string rather than a full document replay.
• For Multi-language Invoice Parsing, an unsupported character encoding (GB18030, Shift-JIS) with no fallback translation layer is classified permanent, while a temporary translation-API throttle is retried with backoff. Encoding must be normalized before hashing so the idempotency key stays stable across attempts.

All retry states are tracked in a centralized job ledger so that a broker failover mid-batch cannot re-drive a document that already succeeded.

# Scaling & resilience

At volume, unbounded retries are more dangerous than the failures they chase: they exhaust connection pools and trigger cascading failures across the whole clearance fleet. Three controls keep the pipeline stable under load.

Circuit breaker. When the transient error rate exceeds 15% over a rolling 5-minute window, the breaker opens: it halts new ingestion jobs while allowing in-flight retries to drain, so no document is lost. Once external services stabilize, a controlled warm-up phase gradually restores throughput rather than releasing the full backlog at once.

import time


class CircuitBreaker:
    """Opens when the transient failure ratio exceeds a threshold over a
    rolling window; pauses new work while in-flight retries drain cleanly."""

    def __init__(self, threshold: float = 0.15, window_s: int = 300,
                 cooldown_s: int = 60) -> None:
        self.threshold = threshold
        self.window_s = window_s
        self.cooldown_s = cooldown_s
        self._events: list[tuple[float, bool]] = []   # (timestamp, is_failure)
        self._opened_at: Optional[float] = None

    def record(self, *, failed: bool) -> None:
        now = time.monotonic()
        self._events.append((now, failed))
        self._events = [(t, f) for t, f in self._events if now - t <= self.window_s]
        total = len(self._events)
        if total >= 20:                                # ignore small samples
            ratio = sum(1 for _, f in self._events if f) / total
            if ratio >= self.threshold and self._opened_at is None:
                self._opened_at = now
                logger.critical("circuit breaker opened",
                                 extra={"failure_ratio": round(ratio, 3)})

    def allow_ingestion(self) -> bool:
        """False while open; the probe re-closes after the cooldown elapses."""
        if self._opened_at is None:
            return True
        if time.monotonic() - self._opened_at >= self.cooldown_s:
            self._opened_at = None
            logger.warning("circuit breaker closed after cooldown")
            return True
        return False

Backpressure and semaphores. Consumers cap concurrent extraction with a semaphore so retry storms cannot spawn unbounded workers; the surplus stays durably buffered in the broker rather than in process memory.

Bounded retry budget. stop_after_attempt(4) caps every document at one try plus three retries, so a poison payload cannot loop forever — it dead-letters and frees the worker.

# Compliance obligations

Every failed payload routed to the dead-letter queue must retain immutable metadata and cryptographic integrity: the original binary document, each extraction attempt, the error classification, and the hts_validation_state together form a forensic chain of custody. Brokerage systems archive DLQ records in write-once storage with retention aligned to customs recordkeeping — typically 5–7 years under CBP requirements. Circuit-breaker state transitions must be logged with timestamps, failure counts, and affected document batches, because compliance officers need visibility into operational pauses to adjust clearance SLAs and communicate proactively with customs authorities.

Validation must be codified against the definitive nomenclature rather than developer assumptions: invalid chapter codes (00–97), malformed heading/subheading separators, or a missing General Rules of Interpretation (GRI) context halt processing and route to the DLQ. Periodic reconciliation jobs scan DLQ partitions, surface recurring failure patterns, and feed root-cause analysis back to the ingestion engineering team. Any failure that resolves to human-in-the-loop review must escalate through the same audited job ledger so no document exits the pipeline unrecorded.

By enforcing strict error classification, deterministic retry boundaries, and compliance-aligned validation gates, customs ETL pipelines reach the resilience required for high-volume, penalty-free clearance operations.

# Related

• Designing Exponential Backoff for Failed Parsing Jobs — the full-jitter backoff curve, cap selection, and thundering-herd analysis behind the retry helper above.
• Async Batch Processing for High Volume — partitioned queue consumers that inherit this retry and DLQ standard.
• OCR Drift Correction & Validation — the targeted re-extraction pass that a transient OCR failure triggers.
• Commercial Invoice PDF Extraction — the upstream extractor whose parser and schema failures this page classifies.
• Multi-language Invoice Parsing — encoding failures that split into permanent vs. retryable paths.

Up: Document Ingestion & Parsing Workflows

Authoritative references: WCO HS Nomenclature 2022 Edition, HTSUS (USITC), CBP ACE / ABI submission formats, WCO Data Model 3.x, ISO 4217 currency codes.