This guide is part of the Pipeline Fallback & Retry Logic section within the broader Core Architecture & Compliance Mapping framework, and it solves one narrow problem: when a grantor’s primary API degrades, drifts, or throttles mid-cycle, how do you redirect each payload to a secondary endpoint, a local compliance cache, or a dead-letter queue without losing the audit trail that 2 CFR §200.302(b) requires?
A fallback router is the control surface that answers that question. It sits between your validated ingestion boundary and the network, makes a deterministic routing decision for every payload, and emits a structured audit record for each decision so a compliance officer can later reconstruct exactly where a submission went and why.
When to Use This Approach
Reach for a deterministic fallback router when all of the following hold:
- Multiple addressable endpoints exist. A grantor exposes a primary API plus a mirror, an aggregator, or you maintain a local compliance cache that can absorb writes. With a single endpoint there is nothing to fall back to — you want bounded retry instead.
- The payload is already validated. This router assumes inbound data has cleared the ingestion boundary and conforms to the canonical contract sealed by Compliance Metadata Standards. Routing is not the place to repair malformed JSON.
- Submissions are deadline-bound and auditable. Grant cycles close on fixed dates, and every dispatch must be traceable for IRS Form 990 Schedule O disclosure of operational controls and for state grant reviews. Silent loss is unacceptable.
- Failures are heterogeneous. You are absorbing a mix of HTTP
5xx, connection resets, schema drift on the response, and429throttling — not a single predictable error.
If your failures are purely rate limits with Retry-After headers, handle them upstream in API Polling & Rate Limiting before payloads ever reach this router; cadence control and fallback routing are separate concerns.
Input preconditions: a GrantPayload instance carrying an ein, grant_id, amount, compliance_status, and grant_period; a primary URL; one fallback URL; and a writable DLQ path. All network egress runs over TLS as enforced by Data Security & Access Boundaries.
Step-by-Step Implementation
The router is built in four runnable steps: validate, guard each endpoint with a circuit breaker, route deterministically, then preserve compliance metadata on the way to the DLQ.
Step 1: Validate the payload before any dispatch
Silent schema drift is the leading cause of downstream compliance corruption, so the router never trusts an unvalidated dict. Inbound payloads are checked against a rigid Pydantic contract aligned with IRS 990 Data Schema Mapping. A missing ein, grant_period, or compliance_status is rejected at ingestion, never blind-retried.
import hashlib
import json
import logging
import time
from enum import Enum
from typing import Any, Dict, Literal, Optional
from pydantic import BaseModel, Field, ValidationError
# Structured audit logger aligned with Compliance Metadata Standards.
logger = logging.getLogger("grant_pipeline.audit")
logger.setLevel(logging.INFO)
handler = logging.StreamHandler()
handler.setFormatter(logging.Formatter("%(asctime)s | %(levelname)s | %(message)s"))
logger.addHandler(handler)
class ComplianceDomain(str, Enum):
IRS_990 = "irs_990_schema"
STATE_REG = "state_charity_registration"
GRANTOR_RULES = "grantor_rule_taxonomy"
SECURITY_BOUNDARY = "data_security_access"
class GrantPayload(BaseModel):
ein: str = Field(..., pattern=r"^\d{2}-\d{7}$")
grant_id: str = Field(..., min_length=4)
amount: float = Field(..., gt=0)
compliance_status: Literal["active", "pending", "suspended"]
grant_period: str = Field(..., pattern=r"^\d{4}-\d{2}$")
metadata: Dict[str, Any] = Field(default_factory=dict)
def validate_ingestion(payload: Dict[str, Any]) -> GrantPayload:
"""Pre-flight validation with explicit drift isolation."""
try:
validated = GrantPayload.model_validate(payload)
digest = hashlib.sha256(
json.dumps(validated.model_dump(), sort_keys=True).encode()
).hexdigest()
logger.info("VALIDATION_SUCCESS | EIN=%s | HASH=%s", validated.ein, digest[:12])
return validated
except ValidationError as exc:
logger.error("VALIDATION_REJECTED | DOMAIN=%s | ERR=%s",
ComplianceDomain.IRS_990.value, exc)
raise RuntimeError("Invalid grant schema; route to Error Categorization & Logging") from exc
The ein pattern enforces the federal NN-NNNNNNN format, amount must be strictly positive, and compliance_status is constrained to a closed set so a typo cannot pass as a routable state. The SHA-256 digest computed here becomes the payload’s identity for the rest of the pipeline.
Step 2: Guard each endpoint with a circuit breaker
Blind exponential backoff against a failing endpoint wastes the submission window and exhausts buffers during high-volume cycles. Instead, a per-endpoint circuit breaker tracks consecutive failures; once it trips, traffic is diverted immediately rather than re-attempted.
class CircuitBreaker:
"""State-aware breaker for a single grantor endpoint."""
def __init__(self, failure_threshold: int = 3, reset_timeout: float = 60.0) -> None:
self.failure_threshold = failure_threshold
self.reset_timeout = reset_timeout
self.failure_count = 0
self.last_failure_time = 0.0
self.state: Literal["closed", "open", "half_open"] = "closed"
def record_failure(self) -> None:
self.failure_count += 1
self.last_failure_time = time.monotonic()
if self.failure_count >= self.failure_threshold:
self.state = "open"
logger.warning("CIRCUIT_OPEN | Threshold exceeded; routing diverted.")
def record_success(self) -> None:
self.failure_count = 0
self.state = "closed"
def allow_request(self) -> bool:
if self.state == "closed":
return True
if self.state == "open" and (time.monotonic() - self.last_failure_time) > self.reset_timeout:
self.state = "half_open"
return True
return False
failure_threshold=3 opens the circuit after three consecutive failures; reset_timeout=60.0 lets a single probe through after a minute (half_open) to test recovery. Crucially, each grantor endpoint owns its own breaker — shared state across unrelated grantors would let one funder’s outage cascade into another’s pipeline.
Step 3: Route through the deterministic cascade
The router tries the primary endpoint under a strict timeout, falls back to the secondary mirror on any failure, and lands in the DLQ if both fail. Every branch emits an AuditRecord, so the routing path is always reconstructable.
import httpx
class AuditRecord(BaseModel):
pipeline_stage: str
routing_decision: str
compliance_domain: ComplianceDomain
payload_hash: str
timestamp: float
error_trace: Optional[str] = None
class FallbackRouter:
"""Deterministic primary → secondary → DLQ router with strict stage isolation."""
def __init__(self, primary_url: str, fallback_url: str, dlq_path: str,
compliance_domain: ComplianceDomain = ComplianceDomain.GRANTOR_RULES) -> None:
self.primary_url = primary_url
self.fallback_url = fallback_url
self.dlq_path = dlq_path
self.domain = compliance_domain
self.circuit = CircuitBreaker()
# Enforce a <3s synchronous compliance-check SLA.
self.client = httpx.Client(timeout=httpx.Timeout(3.0, connect=1.0))
def _audit(self, stage: str, decision: str, payload_hash: str,
error: Optional[str] = None) -> None:
record = AuditRecord(
pipeline_stage=stage,
routing_decision=decision,
compliance_domain=self.domain,
payload_hash=payload_hash,
timestamp=time.time(),
error_trace=error,
)
logger.info(json.dumps(record.model_dump()))
def route_payload(self, payload: GrantPayload) -> Dict[str, Any]:
"""Primary dispatch with deterministic fallback cascade."""
payload_hash = hashlib.sha256(
json.dumps(payload.model_dump(), sort_keys=True).encode()
).hexdigest()
if not self.circuit.allow_request():
logger.info("CIRCUIT_OPEN | Bypassing primary endpoint.")
return self._execute_fallback(payload, payload_hash)
try:
response = self.client.post(self.primary_url, json=payload.model_dump())
response.raise_for_status()
self.circuit.record_success()
self._audit("primary_dispatch", "success", payload_hash)
return response.json()
except httpx.HTTPStatusError as exc:
self.circuit.record_failure()
self._audit("primary_dispatch", "http_error", payload_hash, str(exc))
return self._execute_fallback(payload, payload_hash)
except httpx.TimeoutException:
self.circuit.record_failure()
self._audit("primary_dispatch", "timeout", payload_hash, "SLA breach: >3s")
return self._execute_fallback(payload, payload_hash)
except httpx.HTTPError as exc:
self.circuit.record_failure()
self._audit("primary_dispatch", "transport_error", payload_hash, str(exc))
return self._execute_fallback(payload, payload_hash)
def _execute_fallback(self, payload: GrantPayload, payload_hash: str) -> Dict[str, Any]:
"""Secondary mirror, then DLQ terminal."""
try:
response = self.client.post(self.fallback_url, json=payload.model_dump())
response.raise_for_status()
self._audit("fallback_dispatch", "secondary_success", payload_hash)
return response.json()
except httpx.HTTPError as exc:
return self._write_dlq(payload, payload_hash, str(exc))
def _write_dlq(self, payload: GrantPayload, payload_hash: str, error: str) -> Dict[str, Any]:
dlq_entry = {
"payload": payload.model_dump(),
"hash": payload_hash,
"timestamp": time.time(),
"compliance_domain": self.domain.value,
"error": error,
"routing_path": "primary -> fallback -> dlq",
}
logger.critical("DLQ_WRITE | REF=%s | ENTRY=%s", payload_hash, json.dumps(dlq_entry))
self._audit("dlq_reconciliation", "queued", payload_hash, error)
return {"status": "queued_for_manual_reconciliation", "dlq_ref": payload_hash}
The catch blocks are ordered narrowest-first: HTTPStatusError (a 4xx/5xx response) and TimeoutException are subclasses of httpx.HTTPError, so the final HTTPError clause sweeps up connection resets and DNS failures without swallowing unrelated exceptions. Every failure path returns a structured dict — never a bare exception or None.
Step 4: Preserve compliance metadata into the DLQ
The DLQ entry is not just an error dump; it is an audit artifact. Before persistence, strip transient bearer tokens and apply field-level hashing to any PII so the queue itself never becomes a leak vector, per Data Security & Access Boundaries. The payload_hash doubles as the idempotency key: manual reconciliation consumes DLQ entries keyed by that hash, so re-processing a payload can never produce a duplicate grant submission. Wire alerting on dlq_reconciliation audit events to trigger a schema-drift review against the affected Grantor-Specific Rule Taxonomies.
Verification
Confirm the router is behaving deterministically with three checks:
- Happy path emits a success record. A reachable primary should yield one
primary_dispatch | successaudit line and return the endpoint’s JSON.
router = FallbackRouter(
primary_url="https://grants.example.org/v2/submit",
fallback_url="https://mirror.example.org/v2/submit",
dlq_path="/var/spool/grant_dlq",
)
result = router.route_payload(validate_ingestion({
"ein": "12-3456789",
"grant_id": "G-2026-0042",
"amount": 25000.0,
"compliance_status": "active",
"grant_period": "2026-06",
}))
-
Forced primary failure lands in the cascade. Point
primary_urlat a closed port and assert the return is{"status": "queued_for_manual_reconciliation", "dlq_ref": <hash>}when the fallback is also unreachable. The logs should showprimary_dispatch, thenfallback_dispatch, then aDLQ_WRITE | REF=<hash>critical line. -
Circuit opens after three failures. Drive three consecutive primary failures and assert
router.circuit.state == "open"; the fourthroute_payloadcall must logCIRCUIT_OPEN | Bypassing primary endpoint.and skip the primary entirely.
assert router.circuit.state == "open"
assert result["status"] == "queued_for_manual_reconciliation"
assert result["dlq_ref"] # SHA-256 hex digest, stable across identical payloads
Identical payloads must produce identical dlq_ref values — that stability is what makes reconciliation idempotent.
Common Errors & Fixes
| Error | Cause | Fix |
|---|---|---|
RuntimeError: Invalid grant schema on every payload |
Upstream sends ein as 123456789 (no hyphen) or amount as a string |
Normalize in Field Mapping & Normalization before routing; the router rejects, it does not repair |
| Circuit never reopens after an outage clears | reset_timeout longer than the grant window, or the half-open probe keeps failing on a stale connection |
Lower reset_timeout, and recreate the httpx.Client so the probe uses a fresh socket |
| Duplicate grant submissions after a DLQ replay | Reconciliation consumer ignores payload_hash and re-posts blindly |
Make the consumer upsert on dlq_ref; never re-route a hash already marked reconciled |
TimeoutException floods the DLQ during a slow-but-healthy window |
3s SLA too aggressive for large batched payloads | Raise the read timeout for batch endpoints only; keep the 1s connect timeout to fail fast on dead hosts |
| Audit records missing from the SIEM | Logger configured per-process, handler not attached in worker subprocesses | Configure grant_pipeline.audit once at worker bootstrap and forward to write-once storage for 2 CFR §200.302 retention |
Related
- Parent guide: Pipeline Fallback & Retry Logic
- Upstream cadence control: Handling Rate Limits in Grant Portal APIs
- Where rejected payloads go: Error Categorization & Logging
- Sealing the audit envelope: Compliance Metadata Standards