Pipeline Fallback & Retry Logic

A deterministic error-handling layer for nonprofit grant automation: bounded exponential retries for transient faults, immutable state snapshots, and audited fallback routing to a dead-letter queue for permanent failures.

This module is part of the Core Architecture & Compliance Mapping reference, where it owns one discrete responsibility: deciding, deterministically, whether a failed pipeline operation is retried, abandoned, or escalated. It sits beside every other stage rather than inside any of them. A transient infrastructure fault triggers bounded, auditable reprocessing; a permanent schema or authorization fault bypasses the retry queue entirely and routes to deterministic exception handling. Every state transition is anchored to compliance metadata so that financial and regulatory traceability survives across the full grant lifecycle.

Explicitly out of scope. This module does not validate the contents of a grant filing, map fields to a canonical contract, or render a regulatory artifact. Schema normalization belongs to the IRS 990 Data Schema Mapping layer; jurisdictional filing rules belong to State Charity Registration Compliance; funder eligibility logic belongs to Grantor-Specific Rule Taxonomies; encryption and role boundaries belong to Data Security & Access Boundaries. This module only classifies failures and moves payloads between the retry queue, the next stage, and the dead-letter queue. It never mutates a payload — only validated payloads and immutable compliance metadata traverse its boundaries.

The four stakeholders served are nonprofit operations and grant managers (no silently dropped submissions before a filing deadline), Python automation developers (a single, testable error-routing contract), and compliance officers (a cryptographically chained record of every retry and escalation).

Fallback and retry control plane as a deterministic state machine A failed operation result enters a single classify_failure decision node, which routes to exactly one of three lanes. A 2xx success forwards the validated payload to the next stage. Transient faults (429, 503, timeout) enter a bounded exponential-backoff loop capped at 3 attempts within a 15-minute budget, which feeds back into classify_failure; once attempts are exhausted the payload escalates to the write-once dead-letter queue. Permanent faults (400, 403, 404, schema) bypass retries entirely and route straight to the dead-letter queue for manual review. An append-only audit hook stamps idempotency_key, attempt_count and payload_hash on every transition. Fallback & retry control plane One deterministic decision routes every failed operation to next-stage, the retry loop, or the dead-letter queue. 429 · 503 · timeout retry · backoff 2xx success 400 · 403 · 404 · schema attempts exhausted Operation result classify_ failure Exponential backoff + jitter max 3 attempts · 15-min budget Next-stage handoff validated payload forwarded Dead-letter queue write-once · manual review Audit hook on every transition Append-only record stamped before and after each edge idempotency_key attempt_count payload_hash

Prerequisites

This module pins its runtime so that retry timing and idempotency keys are reproducible across CI workers and production.

  • Python: 3.11 or newer (relies on datetime.UTC, time.monotonic_ns, and structured exception groups).
  • Pinned packages:
bash
# requirements.txt (fallback/retry module only)
tenacity==8.4.1          # bounded exponential backoff with jitter
pydantic==2.7.1          # payload contract shared with the mapping layer
orjson==3.10.3           # deterministic canonical serialization for hashing
pytest==8.2.0            # retry-policy and routing tests
freezegun==1.5.1         # deterministic time control in backoff tests
  • Environment variables: RETRY_MAX_ELAPSED_SECONDS (total retry budget, default 900), RETRY_MAX_ATTEMPTS (hard attempt ceiling, default 3), DLQ_SINK_URL (write-once dead-letter target), and AUDIT_SINK_URL (append-only audit log target shared with Compliance Metadata Standards).
  • Upstream dependency: this module wraps operations that talk to grantor APIs. The HTTP transport, polling cadence, and primary error surfacing are owned by API Polling & Rate Limiting and Error Categorization & Logging in the ingestion reference. This module assumes the call already failed; it decides what happens next.

Core implementation: failure classification and retry execution

The control plane is a pure decision function plus a tenacity-driven executor. Classification is deterministic: an exception or HTTP status maps to exactly one of three terminal dispositions — PERMANENT, TRANSIENT, or SUCCESS — with no heuristic fallthrough. Logging uses the standard library logger, and every branch returns a structured result rather than raising past the boundary.

python
from __future__ import annotations

import hashlib
import logging
import time
from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Callable, Mapping

import orjson
import tenacity
from tenacity import (
    retry_if_exception_type,
    stop_after_attempt,
    stop_after_delay,
    wait_exponential_jitter,
)

logger = logging.getLogger("core_architecture.fallback_retry")

PERMANENT_STATUS = frozenset({400, 401, 403, 404, 409, 422})
TRANSIENT_STATUS = frozenset({425, 429, 500, 502, 503, 504})


class Disposition(str, Enum):
    SUCCESS = "success"
    TRANSIENT = "transient_retry_eligible"
    PERMANENT = "permanent_fallback_required"


class TransientError(Exception):
    """Raised to re-enter the bounded retry loop."""


class PermanentError(Exception):
    """Raised to route directly to the dead-letter queue — never retried."""


@dataclass(frozen=True)
class RetryOutcome:
    disposition: Disposition
    payload_hash: str
    attempt_count: int
    last_status: int | None = None
    detail: str = ""
    metadata: Mapping[str, Any] = field(default_factory=dict)


def payload_hash(payload: Mapping[str, Any]) -> str:
    """SHA-256 over the canonical serialization — stable idempotency key."""
    canonical = orjson.dumps(payload, option=orjson.OPT_SORT_KEYS)
    return hashlib.sha256(canonical).hexdigest()


def classify(status: int | None, exc: Exception | None) -> Disposition:
    """Deterministically map a failure to a single disposition."""
    if status is not None and 200 <= status < 300:
        return Disposition.SUCCESS
    if status in PERMANENT_STATUS or isinstance(exc, (ValueError, PermanentError)):
        return Disposition.PERMANENT
    if status in TRANSIENT_STATUS or isinstance(
        exc, (TimeoutError, ConnectionError, TransientError)
    ):
        return Disposition.TRANSIENT
    # Unknown faults are treated as permanent: silent retries are forbidden.
    return Disposition.PERMANENT


def execute_with_retry(
    operation: Callable[[Mapping[str, Any]], int],
    payload: Mapping[str, Any],
    *,
    max_attempts: int = 3,
    max_elapsed_seconds: int = 900,
) -> RetryOutcome:
    """Run `operation` under a bounded, audited retry policy.

    `operation` returns an HTTP-style status code or raises. Transient faults
    are retried with capped exponential backoff + jitter; permanent faults
    short-circuit to a fallback outcome with no further attempts.
    """
    key = payload_hash(payload)
    attempts = {"n": 0}

    @tenacity.retry(
        retry=retry_if_exception_type(TransientError),
        wait=wait_exponential_jitter(initial=2, max=30),
        stop=(stop_after_attempt(max_attempts) | stop_after_delay(max_elapsed_seconds)),
        reraise=True,
    )
    def _run() -> int:
        attempts["n"] += 1
        status = operation(payload)
        disposition = classify(status, None)
        if disposition is Disposition.TRANSIENT:
            logger.warning(
                "retry_triggered",
                extra={"payload_hash": key, "attempt": attempts["n"], "status": status},
            )
            raise TransientError(f"transient status {status}")
        if disposition is Disposition.PERMANENT:
            raise PermanentError(f"permanent status {status}")
        return status

    try:
        status = _run()
        logger.info(
            "operation_succeeded",
            extra={"payload_hash": key, "attempt": attempts["n"], "status": status},
        )
        return RetryOutcome(Disposition.SUCCESS, key, attempts["n"], status)
    except PermanentError as exc:
        logger.error(
            "permanent_failure",
            extra={"payload_hash": key, "attempt": attempts["n"], "detail": str(exc)},
        )
        return RetryOutcome(Disposition.PERMANENT, key, attempts["n"], detail=str(exc))
    except TransientError as exc:
        # Retry budget exhausted — degrade transient into a fallback.
        logger.error(
            "retry_exhausted",
            extra={"payload_hash": key, "attempt": attempts["n"], "detail": str(exc)},
        )
        return RetryOutcome(
            Disposition.PERMANENT, key, attempts["n"],
            detail=f"retry_exhausted: {exc}",
            metadata={"escalation_code": "RETRY_EXHAUSTED"},
        )

A PERMANENT outcome — whether by classification or by exhausting the retry budget — is wrapped in an immutable exception envelope and routed to the dead-letter queue. The envelope carries the full attempt history, the payload hash, and an escalation code; no downstream automation may modify it until a compliance officer adjudicates it. The queue topology and operator workflow for that escalation are detailed in Building a fallback routing system for grant APIs.

python
def route_to_dlq(outcome: RetryOutcome, payload: Mapping[str, Any]) -> dict[str, Any]:
    """Wrap a permanent outcome in a write-once envelope for manual review."""
    envelope = {
        "original_payload": dict(payload),
        "compliance_metadata": {
            "payload_hash": outcome.payload_hash,
            "attempt_count": outcome.attempt_count,
            "boundary": "fallback",
            "compliance_state": "manual_review_required",
            "escalation_code": outcome.metadata.get(
                "escalation_code", "PERMANENT_FAILURE"
            ),
            "audit_trail": outcome.detail,
            "routed_at": time.time(),
        },
        "routing_target": "compliance_exception_dlq",
        "immutable": True,
    }
    logger.critical(
        "fallback_routing_executed",
        extra={"payload_hash": outcome.payload_hash, "target": envelope["routing_target"]},
    )
    return envelope

Field mapping / schema contract

The retry module does not own the grant payload schema — that contract is set by the IRS 990 Data Schema Mapping layer. What it does own is the compliance-metadata envelope it stamps onto every operation. These are the canonical fields, the aliases this module accepts from upstream stages, and the coercion rules applied before any value is hashed or logged.

Canonical field Accepted aliases (upstream) Type / coercion Notes
payload_hash idempotency_key, dedupe_key str (64-char hex) SHA-256 of the sorted-key canonical JSON; never regenerated mid-retry
attempt_count retries, try_no int, coerce str→int, floor at 0 Monotonic; appended, never overwritten
disposition state, routing_state Disposition enum One of success / transient_retry_eligible / permanent_fallback_required
last_status http_status, code int | None Null only when failure was an exception, not a response
boundary stage, pipeline_stage str, lowercased ingestion / reconciliation / fallback — stamps which stage classified the fault
escalation_code error_code, reason str, upper-snake Present only on dead-letter envelopes
routed_at ts, timestamp float epoch seconds (UTC) Set once at envelope creation

Coercion is intentionally narrow: an alias that cannot be coerced to its canonical type is a PERMANENT fault, not a best-effort guess. This keeps the audit chain unambiguous — a record either has a valid attempt_count or it never entered the retry loop.

python
ALIASES: dict[str, str] = {
    "idempotency_key": "payload_hash", "dedupe_key": "payload_hash",
    "retries": "attempt_count", "try_no": "attempt_count",
    "http_status": "last_status", "code": "last_status",
    "stage": "boundary", "pipeline_stage": "boundary",
    "error_code": "escalation_code", "reason": "escalation_code",
}


def normalize_metadata(raw: Mapping[str, Any]) -> dict[str, Any]:
    """Resolve aliases to canonical keys; raise PermanentError on bad coercion."""
    out: dict[str, Any] = {}
    for key, value in raw.items():
        canonical = ALIASES.get(key, key)
        if canonical == "attempt_count":
            try:
                value = max(0, int(value))
            except (TypeError, ValueError) as exc:
                raise PermanentError(f"uncoercible attempt_count: {value!r}") from exc
        if canonical == "boundary" and isinstance(value, str):
            value = value.lower()
        out[canonical] = value
    return out

Validation & testing

Retry policy is the easiest thing in a pipeline to get subtly wrong, so it is tested with deterministic time control. The suite asserts three invariants: permanent faults never retry, transient faults stop at the attempt ceiling, and the payload hash is stable across every attempt. Audit-log assertions confirm a retry_exhausted record is emitted exactly once.

python
import logging
import pytest
from freezegun import freeze_time

from fallback_retry import (
    Disposition, PermanentError, execute_with_retry, payload_hash,
)

PAYLOAD = {"grantor_id": "G-114", "award_amount": 50000.0, "fiscal_year": 2026}


def test_payload_hash_is_stable_across_key_order() -> None:
    reordered = {"fiscal_year": 2026, "award_amount": 50000.0, "grantor_id": "G-114"}
    assert payload_hash(PAYLOAD) == payload_hash(reordered)


def test_permanent_fault_is_not_retried() -> None:
    calls = {"n": 0}

    def op(_: dict) -> int:
        calls["n"] += 1
        return 403  # permanent

    outcome = execute_with_retry(op, PAYLOAD)
    assert outcome.disposition is Disposition.PERMANENT
    assert calls["n"] == 1  # no retry on a 403


def test_transient_fault_stops_at_attempt_ceiling(caplog) -> None:
    calls = {"n": 0}

    def op(_: dict) -> int:
        calls["n"] += 1
        return 503  # always transient

    with caplog.at_level(logging.ERROR):
        outcome = execute_with_retry(op, PAYLOAD, max_attempts=3)

    assert outcome.disposition is Disposition.PERMANENT
    assert outcome.metadata["escalation_code"] == "RETRY_EXHAUSTED"
    assert calls["n"] == 3
    assert sum(r.message == "retry_exhausted" for r in caplog.records) == 1


@freeze_time("2026-06-27", auto_tick_seconds=400)
def test_elapsed_budget_halts_before_attempt_ceiling() -> None:
    """Two 400s ticks exceed the 900s budget before 3 attempts complete."""
    def op(_: dict) -> int:
        return 502

    outcome = execute_with_retry(op, PAYLOAD, max_attempts=10, max_elapsed_seconds=900)
    assert outcome.disposition is Disposition.PERMANENT
    assert outcome.attempt_count <= 3

Pass payloads return Disposition.SUCCESS with attempt_count == 1; fail payloads (a 403, an uncoercible attempt_count, or an unknown exception) return Disposition.PERMANENT and never increment past their first attempt.

Performance & scale considerations

Nonprofit-scale workloads are bursty but small — a few thousand grant submissions around a quarterly deadline, not millions per second. The retry policy is tuned for that profile rather than for high-throughput streaming.

  • Backoff envelope. With wait_exponential_jitter(initial=2, max=30) and a 3-attempt ceiling, a single payload occupies a worker for at most ~64 seconds of wall time including jitter, well under the 900-second elapsed budget. The budget exists to cap pathological slow-failure tails, not normal retries.
  • Concurrency. Run the executor inside a bounded worker pool (8–16 concurrent operations is ample for nonprofit volumes). The jitter term is what prevents a synchronized thundering herd against a single grantor portal after an outage — do not remove it. Per-grantor concurrency caps belong upstream in API Polling & Rate Limiting.
  • Memory. Each in-flight envelope holds the original payload plus metadata; cap individual payloads at a few hundred KB and the resident set stays trivial. Large bulk filings should be routed through Async Batch Processing Pipelines, which chunks before this module ever sees a payload.
  • Idempotency cost. The SHA-256 over canonical JSON is the only per-call CPU cost worth noting and is negligible (microseconds) at this payload size. Hashing once and threading the key through every attempt avoids recomputation.

Failure modes & troubleshooting

Symptom Root cause Remediation
Duplicate grant submissions land at grantor payload_hash regenerated per attempt instead of once Compute the hash before the retry loop; thread the same key through every attempt and the DLQ envelope
A 403 is retried 3 times Status placed in TRANSIENT_STATUS or caught as a generic exception Keep authorization/validation codes in PERMANENT_STATUS; never catch PermanentError as transient
Workers stall for minutes during an outage Elapsed budget unset or jitter removed, so all workers retry in lockstep Set RETRY_MAX_ELAPSED_SECONDS; keep wait_exponential_jitter; cap pool size
DLQ envelope is mutated after routing Downstream automation treats the queue as reprocessable Enforce write-once storage (object-lock / append-only) and RBAC per Data Security & Access Boundaries
Audit chain has gaps for failed attempts Logging swallowed inside operation instead of the executor Let the executor own all retry_triggered / retry_exhausted records; operations return status, they do not log dispositions
Unknown exception silently dropped Catch-all that returns success on uncertainty classify() maps unknown faults to PERMANENT; assert no bare except: pass exists in operations

Compliance alignment

This module is the control surface that makes the platform’s financial-management claims defensible. It satisfies specific, citable requirements rather than generic “compliance”:

  • 2 CFR §200.302 (financial management). Federal award recipients must maintain records that permit the tracing of funds to a level adequate to establish that funds were used appropriately. Threading an immutable payload_hash and monotonic attempt_count through every retry and escalation produces exactly that traceable chain for each grant operation.
  • 2 CFR §200.303 (internal controls). Recipients must establish internal controls that provide reasonable assurance of compliance. Deterministic classification — permanent faults can never silently retry, unknown faults default to escalation — is a documented, testable control rather than ad-hoc error handling.
  • 2 CFR §200.334 (record retention). Financial records must be retained for three years from the date of submission of the final financial report. Dead-letter envelopes and audit-log records inherit this retention floor; the routed_at timestamp anchors the retention clock. Where IRS substantiation periods are longer, retain to the longer horizon.
  • IRS Form 990 substantiation. Because the payloads handled here feed IRS 990 Data Schema Mapping, a clean, gap-free retry trail supports the books-and-records expectation behind Form 990 Part XII (Financial Statements and Reporting) by proving no disbursement record was dropped or double-counted during transient outages.

Audit artifacts emitted here conform to the field contract in Compliance Metadata Standards; access segmentation of the retry and dead-letter logs is enforced per Data Security & Access Boundaries.