Error Handling and Retry Logic for Ingestion Pipelines

18 min read 1 sub-topic

This guide is the resilience layer of the broader automated permit ingestion and parsing workflows: the code that decides what happens when a fetch times out, a payload is malformed, or a downstream database refuses a write. Municipal permit and inspection workflows run under rigid statutory deadlines, public-records transparency mandates, and jurisdiction-specific compliance rules. When automated systems pull applications from fragmented sources — legacy spreadsheets, public-facing portals, scanned documents — failure handling stops being a technical nicety and becomes a compliance obligation. Deterministic recovery is what keeps a transient network blip or a single bad row from cascading into a missed inspection window or an audit finding.

Problem Statement and Scope

Without a deliberate error strategy, an ingestion pipeline fails in the worst possible way: silently. A permit submission disappears between a vendor portal and the records database, no one notices until the applicant calls, and there is no log entry to reconstruct what happened. That outcome is unacceptable when a public records request can compel you to account for every byte you received.

Three groups feel the pain directly. Python automation builders inherit pagers that fire at 2 a.m. because a retry loop hammered a recovering API. Municipal clerks lose trust in the system when duplicate permit records appear after a replay, or when a legitimate application is quarantined with no human ever told. Compliance officers cannot certify retention or chain-of-custody when failures leave no audit trail.

The scope of this component is narrow and well-defined. Its input is any unit of work the pipeline is mid-processing — an HTTP response, a parsed row, a document, a database transaction — together with the exception or status that interrupted it. Its output is a routing decision: retry now, retry later with backoff, send to a quarantine (dead-letter) store for human review, or fail the whole batch. Everything in this guide builds toward making that decision deterministic, observable, and idempotent.

Routing a single failed work item through the error handler A work item plus its raised exception enters a classify() decision node that routes the fault three ways. A transient fault takes the RETRY path through a circuit breaker (showing closed, open and half-open states; open means fail fast and requeue) into a retry-with-backoff stage (exponential plus jitter, capped at five attempts); on success it is stored once via an idempotent upsert, and when retries are exhausted it drops into the dead-letter store. A permanent data error takes the QUARANTINE path straight to the dead-letter store, which preserves the raw payload and error code for a clerk to correct and replay. A compliance violation takes the ESCALATE path to a quarantine-plus-audit-log record that also pages the on-call engineer. One failed work item — classify, then route Deterministic, idempotent recovery: retry, quarantine, or escalate — never silently drop RETRY · transient QUARANTINE · permanent ESCALATE · compliance success exhausted Work item + raised exception classify() route the fault Circuit breaker closed open half-open open ⇒ fail fast, requeue Retry w/ backoff exp + jitter · cap 5× Stored once idempotent upsert Dead-letter store raw payload + error code Clerk review correct & replay Quarantine + audit verbatim, timestamped Alert on-call PagerDuty / Slack

Prerequisites

This component targets Python 3.10+ (the examples use structural match statements and X | Y type unions). Install the runtime dependencies:

pip install "tenacity>=8.2" "httpx>=0.27" "pybreaker>=1.2" "structlog>=24.1"

You will also need:

  • A message broker or table for dead-lettering. Anything durable works — a Postgres quarantine table, a Redis stream, or an SQS/RabbitMQ queue. The examples assume a thin QuarantineStore you can back with any of these.
  • Idempotency keys on inbound records. Each permit submission must carry (or let you derive) a stable identifier — a county-issued application number, or a SHA-256 of the canonical payload — so a replay can be deduplicated. This pairs directly with the schema work in designing JSON schemas for building permits.
  • Write access to a structured log sink (stdout JSON shipped to your aggregator is fine) so every retry and quarantine event is queryable for audits.

Classify Every Failure Before You React

Not every interruption deserves the same response, and the most common production bug in municipal pipelines is retrying something that will never succeed. Triage first. Sort failures into three tiers:

  • Transient infrastructure faults — HTTP 503s, connection-pool exhaustion, DNS hiccups, temporary file locks, vendor rate limiting (429). These are recoverable by waiting and retrying.
  • Permanent data errors — structurally invalid JSON, a missing statutory field, an OCR confidence score below the jurisdiction’s threshold. No amount of retrying fixes a malformed payload; these must skip the retry path entirely and go to quarantine.
  • Compliance-level exceptions — a record that parsed cleanly but violates a policy rule (e.g., a permit type your jurisdiction does not issue, or a deadline already breached). These need both a quarantine route and an immediate alert.

Encode this as a single classifier so every stage of the pipeline routes consistently:

from enum import Enum
import httpx


class Disposition(str, Enum):
    RETRY = "retry"          # transient — safe to retry with backoff
    QUARANTINE = "quarantine"  # permanent — needs clerk review, never retry
    ESCALATE = "escalate"    # compliance — quarantine AND alert on-call


# HTTP statuses that represent a recoverable server-side or throttling fault.
_TRANSIENT_STATUS: frozenset[int] = frozenset({408, 425, 429, 500, 502, 503, 504})


def classify(exc: Exception) -> Disposition:
    """Map a raised exception to a routing decision. Order matters:
    check the specific, permanent cases before the broad transient ones."""
    match exc:
        case ValidationError():            # malformed/missing statutory fields
            return Disposition.QUARANTINE
        case ComplianceError():            # parsed fine, violates policy
            return Disposition.ESCALATE
        case httpx.HTTPStatusError() as e:
            status = e.response.status_code
            if status in _TRANSIENT_STATUS:
                return Disposition.RETRY
            return Disposition.QUARANTINE  # 4xx client errors are permanent
        case httpx.TransportError():       # connect/read timeouts, resets
            return Disposition.RETRY
        case _:
            return Disposition.QUARANTINE  # unknown == do not blindly retry

ValidationError and ComplianceError are your own exception types raised by the validation stage; defining the classifier in one place means the acquisition, parsing, and storage stages all make the same call. The default branch is deliberately conservative: an unrecognized exception is quarantined for a human, never retried into a storm.

Retry Transient Faults With Backoff and Jitter

Once a fault is classified RETRY, the retry policy controls how. Three rules keep a recovering dependency from being knocked over again:

  1. Exponential backoff — multiply the wait after each attempt (1s, 2s, 4s, 8s…) so a struggling service gets progressively more breathing room.
  2. Jitter — add randomness to each wait so that twenty worker processes restarting after a network partition do not all retry on the same tick and create a synchronized “retry storm.”
  3. A hard stop — cap total attempts (or elapsed time) so a genuinely dead endpoint surfaces as a quarantine rather than an infinite loop.

Rather than hand-roll this, lean on tenacity, which encodes exactly these primitives:

import httpx
from tenacity import (
    retry,
    retry_if_exception,
    stop_after_attempt,
    wait_exponential_jitter,
    before_sleep_log,
)
import structlog

log = structlog.get_logger()


def _is_retryable(exc: BaseException) -> bool:
    return isinstance(exc, Exception) and classify(exc) is Disposition.RETRY


@retry(
    retry=retry_if_exception(_is_retryable),
    # 1s, 2s, 4s, 8s ... capped at 30s, with random jitter folded in
    wait=wait_exponential_jitter(initial=1, max=30),
    stop=stop_after_attempt(5),          # five tries, then give up to quarantine
    before_sleep=before_sleep_log(log, log_level="warning"),
    reraise=True,                        # re-raise the last error so the caller can route it
)
async def fetch_permit(client: httpx.AsyncClient, url: str) -> httpx.Response:
    """Fetch one permit record, retrying only on transient faults."""
    resp = await client.get(url, timeout=10.0)
    resp.raise_for_status()              # turns 5xx/429 into HTTPStatusError
    return resp

Because the policy reuses the same classify() function, a 404 (permanent) is not retried — raise_for_status() raises, _is_retryable returns False, and tenacity re-raises immediately so the outer handler can quarantine it. Only the genuinely transient set burns retry attempts. This is the same discipline applied when web scraping municipal permit portals with Python, where expired session tokens and aggressive anti-automation throttling are the dominant transient failure modes.

Guarantee Idempotency So Retries Never Double-Write

Retries and replays are only safe if processing the same permit twice produces the same result as processing it once. Without this guarantee, a retry after a write that actually succeeded but whose acknowledgement was lost creates a duplicate record — and in a fee system, a duplicate fee assessment. Idempotency is the contract that makes every other pattern on this page safe.

The reliable mechanism is a natural or derived idempotency key plus a conditional upsert:

import hashlib
import json
from typing import Any
import asyncpg


def idempotency_key(record: dict[str, Any]) -> str:
    """Stable SHA-256 over the canonical payload. Sorting keys makes the
    hash independent of dict ordering across runs and Python versions."""
    canonical = json.dumps(record, sort_keys=True, separators=(",", ":"))
    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()


async def upsert_permit(pool: asyncpg.Pool, record: dict[str, Any]) -> bool:
    """Insert a permit exactly once. Returns True if newly written,
    False if this exact payload was already stored (a safe replay)."""
    key = idempotency_key(record)
    row = await pool.fetchrow(
        """
        INSERT INTO permits (idem_key, application_no, payload)
        VALUES ($1, $2, $3)
        ON CONFLICT (idem_key) DO NOTHING
        RETURNING idem_key
        """,
        key,
        record["application_no"],
        json.dumps(record),
    )
    return row is not None  # None => conflict => already ingested, no-op

The unique constraint on idem_key does the heavy lifting: a replayed batch collides on the constraint, DO NOTHING skips it, and the function reports the write as a harmless no-op instead of a duplicate. Deriving the key from the canonical payload (not the wall clock or a UUID) is what makes a retry after a lost acknowledgement converge to the same row.

Trip a Circuit Breaker on Degraded Dependencies

Backoff protects a dependency from one worker. A circuit breaker protects it from all of them. When an upstream county API or a vendor portal starts failing en masse, continuing to send requests — even politely backed off — wastes worker capacity and slows recovery. A breaker watches the recent failure rate and, once it crosses a threshold, “opens”: it fails fast for a cool-down window without touching the dependency at all, then probes with a single trial request (“half-open”) before fully closing again.

import pybreaker
import httpx

# Open after 5 consecutive failures; stay open for 60s before a trial probe.
permit_api_breaker = pybreaker.CircuitBreaker(
    fail_max=5,
    reset_timeout=60,
    exclude=[lambda e: classify(e) is Disposition.QUARANTINE],  # don't trip on bad data
)


@permit_api_breaker
async def call_county_api(client: httpx.AsyncClient, url: str) -> httpx.Response:
    """Wrapped call. While the breaker is open this raises CircuitBreakerError
    immediately instead of hitting a downstream that is already on the floor."""
    return await fetch_permit(client, url)

The exclude predicate is the subtle, important part: a stream of bad documents (quarantine-class) should never trip the breaker, because the dependency is healthy — only the data is wrong. Tripping on validation errors would take a working API offline for everyone the moment one county exports a malformed file. When the breaker is open, the caller catches pybreaker.CircuitBreakerError and parks the work item back on the queue with a delay, rather than quarantining it as a data problem.

Route Permanent Failures to Quarantine

Everything classified QUARANTINE or ESCALATE, plus anything that exhausts its retries, lands in a dead-letter store. Quarantine is not a trash can — it is an auditable holding area where a clerk can review, correct, and replay a record, and where every entry carries enough context to reconstruct what happened.

from dataclasses import dataclass, field
from datetime import datetime, timezone
import traceback


@dataclass(slots=True)
class QuarantineRecord:
    idem_key: str
    source_system: str
    disposition: Disposition
    error_code: str
    error_detail: str
    raw_payload: str                      # preserved verbatim for chain-of-custody
    quarantined_at: datetime = field(
        default_factory=lambda: datetime.now(timezone.utc)
    )


async def quarantine(
    store: "QuarantineStore",
    record: dict[str, Any],
    exc: Exception,
) -> None:
    """Persist a failed item with full context, then alert if it is a
    compliance-level escalation."""
    entry = QuarantineRecord(
        idem_key=idempotency_key(record),
        source_system=record.get("source_system", "unknown"),
        disposition=classify(exc),
        error_code=type(exc).__name__,
        error_detail="".join(traceback.format_exception_only(type(exc), exc)).strip(),
        raw_payload=json.dumps(record),
    )
    await store.put(entry)
    log.warning(
        "permit.quarantined",
        idem_key=entry.idem_key,
        source_system=entry.source_system,
        disposition=entry.disposition.value,
        error_code=entry.error_code,
    )
    if entry.disposition is Disposition.ESCALATE:
        await store.alert_oncall(entry)   # PagerDuty/Slack for compliance breaches

Every quarantine entry preserves the raw payload, the precise error code and message, the source system, and a UTC timestamp — exactly the fields a compliance officer needs during a public records request or a state-level system audit. The full telemetry side of this — structured logs, batched digests, and alert routing — is covered in depth in logging and alerting strategies for failed CSV parsing jobs.

Configuration Reference

Externalize every tunable into a per-source configuration object so a flaky vendor portal and a fast internal API can use different policies without code changes.

Parameter Type Default Municipal-context notes
max_attempts int 5 Total tries before quarantine. Lower (2-3) for interactive clerk submissions where fast failure beats a long wait; higher for overnight batch loads.
backoff_initial_s float 1.0 First retry wait. Raise to 5.0+ for vendor APIs with strict rate limits to avoid burning the 429 budget.
backoff_max_s float 30.0 Ceiling on any single wait so a batch never stalls past its inspection-deadline SLA.
breaker_fail_max int 5 Consecutive failures before the circuit opens. Tune to the dependency’s known flakiness during peak submission windows.
breaker_reset_s int 60 Cool-down before a half-open probe. Match to the vendor’s typical outage/maintenance duration.
ocr_conf_threshold float 0.80 Below this, quarantine rather than store. See OCR layout analysis for how the score is computed.
quarantine_ttl_days int 2555 Retention for dead-lettered records (~7 years) to satisfy public-records retention law; never auto-purge below the local mandate.

Error Handling and Edge Cases

Municipal data sources fail in specific, recurring ways that a generic retry policy mishandles:

  • Encoding drift in legacy exports. A county switches its CSV export from UTF-8 to Windows-1252 and previously valid rows raise UnicodeDecodeError. This is permanent for the current payload (retrying re-reads the same bytes) but recoverable with a re-decode, not a quarantine. Catch it at the boundary, attempt a fallback codec, and only quarantine if both fail. This dovetails with syncing legacy CSV exports to modern databases.
  • Scanned vs. digital PDFs. A digital PDF extracts cleanly; a scanned one falls below the OCR confidence threshold and must route to quarantine for manual keying — never retry, since the bytes will never improve. The threshold lives in config so each jurisdiction can set its own bar; see parsing PDF permit applications with OCR and layout analysis.
  • Legacy API timeouts that “succeed late.” An older portal may time out your client after committing the write. This is precisely why idempotency is non-negotiable: the retry collides on the idempotency key and no-ops instead of double-writing.
  • Poison messages. A single record that crashes the worker every time it is dequeued can block a whole queue. Track a per-item attempt counter and force-quarantine after the cap so one bad item never starves the batch.
async def process_one(item: dict[str, Any], store: "QuarantineStore",
                       pool: asyncpg.Pool, client: httpx.AsyncClient) -> None:
    """Top-level handler tying classification, retry, and routing together."""
    try:
        resp = await call_county_api(client, item["url"])
        record = validate(resp.json())          # raises ValidationError/ComplianceError
        await upsert_permit(pool, record)        # idempotent write
    except pybreaker.CircuitBreakerError:
        await store.requeue_later(item, delay_s=permit_api_breaker.reset_timeout)
    except Exception as exc:                      # retries already exhausted by here
        if classify(exc) is Disposition.RETRY:
            await store.requeue_later(item, delay_s=30)  # transient but breaker closed
        else:
            await quarantine(store, item, exc)

Testing and Verification

Resilience code that is not tested against failure is just hope. Simulate each failure class deterministically and assert the routing decision — never rely on a real flaky dependency in CI.

import httpx
import pytest


def _http_error(status: int) -> httpx.HTTPStatusError:
    request = httpx.Request("GET", "https://county.example.gov/permit/1")
    response = httpx.Response(status, request=request)
    return httpx.HTTPStatusError("err", request=request, response=response)


@pytest.mark.parametrize(
    "exc, expected",
    [
        (_http_error(503), Disposition.RETRY),       # transient server fault
        (_http_error(429), Disposition.RETRY),       # rate limited
        (_http_error(404), Disposition.QUARANTINE),  # permanent client error
        (ValidationError("missing parcel id"), Disposition.QUARANTINE),
        (ComplianceError("deadline passed"), Disposition.ESCALATE),
    ],
)
def test_classification_routes_correctly(exc: Exception, expected: Disposition) -> None:
    assert classify(exc) is expected


@pytest.mark.asyncio
async def test_idempotent_replay_is_a_noop(pool: asyncpg.Pool) -> None:
    record = {"application_no": "BLD-2026-0001", "source_system": "county-x"}
    assert await upsert_permit(pool, record) is True    # first write
    assert await upsert_permit(pool, record) is False   # replay: no duplicate

For verification beyond unit tests, run a fault-injection drill: point the pipeline at a mock server that returns a scripted sequence (three 503s then a 200) and confirm the record lands exactly once, the logs show three retries with growing backoff, and the quarantine table is empty. Then flip the mock to permanent 404s and confirm the item is quarantined after the first attempt with its raw payload intact. A passing drill is your evidence, during an audit, that no application is silently dropped.

Integration Notes

This component is the connective tissue between the other ingestion stages, not a standalone service. Acquisition wraps its fetches in the retry decorator; the parsing stage raises ValidationError so the classifier can quarantine low-confidence extractions; the storage stage relies on the idempotency key. It also reaches across to the platform layer: when an upstream system is fully down, the circuit-breaker’s open state is the signal that triggers building fallback routing for legacy system downtime, redirecting traffic to a degraded-mode path instead of quarantining a flood of records.

High-volume loads add their own wrinkle: the same retry and breaker policies must be wired into worker pools and bounded concurrency so a backoff does not silently grow an unbounded in-memory queue. That coordination is the subject of implementing async batch processing for high-volume submissions.

Frequently Asked Questions

When should a failure be retried versus quarantined immediately?

Retry only faults that a later attempt could plausibly fix — network timeouts, 5xx responses, 429 rate limiting, connection-pool exhaustion. Quarantine anything that is wrong about the data itself (malformed JSON, missing statutory fields, OCR below threshold) or any 4xx client error, because retrying the identical bytes will always fail the same way and only wastes worker capacity.

How many retries are appropriate for a municipal pipeline?

Five attempts with exponential backoff capped at 30 seconds is a sound default for batch loads. For interactive clerk submissions, drop to two or three so the user gets a fast, honest failure instead of a long stall. Always bound retries by a hard stop; an uncapped loop against a dead endpoint is how a single outage turns into a backlog that breaches inspection deadlines.

Why is idempotency required before enabling retries?

Because a retry can replay work that already succeeded — for example, a write that committed but whose acknowledgement was lost to a timeout. Without an idempotency key and a conditional upsert, that replay creates a duplicate permit record and, in a fee system, a duplicate charge. Deriving the key from a canonical hash of the payload makes every replay converge to the same row.

What must a quarantined record keep for compliance?

At minimum: the raw payload verbatim (chain-of-custody), the precise error code and message, the originating source system, the disposition, and a UTC timestamp. Retain these for the period your local public-records and records-retention law requires — commonly around seven years — and never auto-purge below that mandate.

Explore deeper

Sub-topic Logging and Alerting Strategies for Failed CSV Parsing Jobs