paulpas System Reliability Architecture
Designs and implements production-grade reliability patterns for distributed systems. When loaded, this skill makes the model build fault-tolerant architectures with circuit breakers, retry strategies with exponential backoff and jitter, bulkhead isolation, comprehensive health checks, graceful degradation, chaos engineering practices, observability foundations, distributed tracing with OpenTelemetry, saga-based distributed transactions, and idempotency guarantees.
TL;DR Checklist
- Implement circuit breaker per downstream dependency with configurable failure threshold (default: 5 consecutive failures)
- Layer retry policies using exponential backoff + jitter — never use fixed-delay retries in production
- Isolate resource pools via bulkhead pattern — each critical path gets its own thread pool or executor
- Deploy liveness probes (process health → restart) and readiness probes (traffic health → drain/load balance out) separately
- Define graceful degradation strategies with fallback responses before implementing any new service dependency
- Instrument all services with metrics, structured logging, and distributed tracing from day one — never retrofit
- Apply idempotency keys to every write operation that may be retried or replayed
- Run at least one chaos experiment per quarter to validate failure assumptions
When to Use
Use this skill when:
- Designing a new distributed system and you need to define its reliability architecture from scratch
- A production incident occurred due to cascading failures — you need to implement circuit breakers, timeouts, or bulkheads to prevent recurrence
- Building inter-service communication where downstream service failures could take down your own service
- Implementing an API gateway or service mesh that needs health-based routing and graceful degradation for end users
- Auditing an existing system for reliability gaps — identifying missing observability, retry storms, or lack of idempotency
When NOT to Use
Avoid this skill for:
- Building monolithic single-process applications with no external dependencies — reliability patterns add overhead that isn't justified
- One-off scripts or throwaway prototypes — the cost of implementing circuit breakers and distributed tracing outweighs benefits
- Performance-critical hot paths where even microsecond latency from retry jitter or tracing spans is unacceptable (use inline timeout-only without full pattern infrastructure)
Core Workflow
Map External Dependencies — Catalog every downstream service, database, cache, and external API your system calls. For each dependency, classify its failure impact: critical path (system halts), important path (degraded experience), or optional (nice to have). Checkpoint: Every critical-path dependency MUST have a circuit breaker and bulkhead pool before any production release.
Implement Circuit Breakers Per Dependency — Deploy three-state circuit breakers on each downstream call with configurable failure threshold, recovery timeout, and half-open success threshold. Use the decorator pattern for clean integration. Checkpoint: Verify the circuit opens after N consecutive failures, transitions to half-open after the recovery timeout, and closes only after M consecutive successes in half-open state.
Layer Retry Policies with Exponential Backoff and Jitter — On transient errors (503, timeouts, connection refused), implement retries using
delay = min(base_delay * (2^attempt) + random_jitter, max_delay). Always use jitter (random.uniform) to prevent thundering herd when the downstream service recovers. Never retry idempotent reads more than 3 times without a fallback. Checkpoint: Confirm retry logic skips non-retriable errors (4xx Client Errors except 429, 500/502/503/504 Server Errors) and includes jitter in every delay calculation.Establish Bulkhead Isolation — Create separate thread pools or executor instances per critical downstream dependency. When one pool exhausts its threads due to a slow service, other services continue functioning independently. Implement explicit rejection policies (raise
BulkheadFullExceptionvs. block-and-queue). Checkpoint: Verify each bulkhead has independent configuration for max_concurrent_calls and queue_size, and that the calling service detects rejection promptly rather than queuing indefinitely.Deploy Comprehensive Health Check Endpoints — Implement three types of probes: startup probe (is initialization complete?), liveness probe (is the process in a consistent state? → Kubernetes restarts), and readiness probe (can this service handle traffic? → Kubernetes removes from load balancer). Liveness must be fast and fail-open; readiness must check actual downstream dependencies. Checkpoint: Readiness probe MUST verify its own critical downstream dependencies — a service reporting ready while its database is unreachable routes broken requests to itself.
Define Graceful Degradation Strategies — For each critical dependency, define what "degraded mode" means for the end user: serve stale cached data, return default values, show cached search results, or display maintenance messaging. Prioritize fallbacks by data freshness requirements. Checkpoint: Every degraded path must be tested independently — verify that cached responses have proper TTL headers and that stale data is clearly labeled when served.
Instrument Observability Foundations — Add Prometheus-style metrics (request count, error rate, latency histograms at p50/p95/p99), structured JSON logging with correlation IDs propagated across async boundaries, and OpenTelemetry distributed tracing with W3C Trace Context headers (
traceparent,tracestate). Checkpoint: Every trace must have a correlation ID that flows through the entire request chain; verify this end-to-end with a test request that spans all services.
Reliability Patterns
Pattern 1: Circuit Breaker Implementation
Three-state circuit breaker (Closed → Open → Half-Open) with configurable thresholds. Closes on consecutive successes in half-open state to prevent premature traffic recovery.
from enum import Enum
import time
import threading
from functools import wraps
from typing import Callable, Any
class CircuitState(Enum):
CLOSED = "closed"
OPEN = "open"
HALF_OPEN = "half_open"
class CircuitBreakerOpenError(Exception):
"""Raised when circuit breaker is open and request is rejected."""
pass
class CircuitBreaker:
"""Three-state circuit breaker with configurable thresholds.
States:
- CLOSED: Normal operation. Requests pass through. Failure counter increments on error.
Opens to OPEN after `failure_threshold` consecutive failures.
- OPEN: All requests rejected immediately. After `recovery_timeout` seconds, transitions to HALF_OPEN.
- HALF_OPEN: Allows `success_threshold` test requests through. If all succeed → CLOSED.
If any fails → back to OPEN for another recovery_timeout period.
"""
def __init__(
self,
failure_threshold: int = 5,
recovery_timeout: float = 30.0,
success_threshold: int = 3,
name: str = "default",
):
self.failure_threshold = failure_threshold
self.recovery_timeout = recovery_timeout
self.success_threshold = success_threshold
self.name = name
self._state = CircuitState.CLOSED
self._failure_count = 0
self._success_count = 0
self._last_failure_time: float | None = None
self._lock = threading.RLock()
@property
def state(self) -> CircuitState:
with self._lock:
if self._state == CircuitState.OPEN and self._last_failure_time is not None:
elapsed = time.monotonic() - self._last_failure_time
if elapsed >= self.recovery_timeout:
self._state = CircuitState.HALF_OPEN
self._success_count = 0
return self._state
def call(self, func: Callable, *args: Any, **kwargs: Any) -> Any:
"""Execute func through the circuit breaker."""
current_state = self.state
if current_state == CircuitState.OPEN:
raise CircuitBreakerOpenError(
f"Circuit breaker '{self.name}' is OPEN. "
f"Recovery in {self.recovery_timeout - (time.monotonic() - self._last_failure_time):.1f}s"
)
try:
result = func(*args, **kwargs)
self._record_success()
return result
except Exception as e:
self._record_failure()
raise
def _record_success(self) -> None:
with self._lock:
if self.state == CircuitState.HALF_OPEN:
self._success_count += 1
if self._success_count >= self.success_threshold:
self._state = CircuitState.CLOSED
self._failure_count = 0
self._success_count = 0
else:
self._failure_count = 0
def _record_failure(self) -> None:
with self._lock:
self._failure_count += 1
self._last_failure_time = time.monotonic()
if self.state == CircuitState.HALF_OPEN:
self._state = CircuitState.OPEN
self._success_count = 0
elif self._failure_count >= self.failure_threshold:
self._state = CircuitState.OPEN
def reset(self) -> None:
with self._lock:
self._state = CircuitState.CLOSED
self._failure_count = 0
self._success_count = 0
def __call__(self, func: Callable) -> Callable:
"""Decorator usage: @circuit_breaker on any function."""
@wraps(func)
def wrapper(*args: Any, **kwargs: Any) -> Any:
return self.call(func, *args, **kwargs)
return wrapper
# Usage example with decorator and manual call
breaker = CircuitBreaker(failure_threshold=3, recovery_timeout=15.0, name="payment-service")
@breaker
def process_payment(user_id: int, amount: float) -> dict:
"""Payment processing protected by circuit breaker."""
response = payment_api.charge(user_id, amount)
return {"status": "success", "transaction_id": response.id}
# Manual call with explicit exception handling
try:
result = breaker.call(payment_api.check_status, txn_id="abc123")
except CircuitBreakerOpenError as e:
# Fallback: return cached or queued payment status
result = get_cached_payment_status(txn_id="abc123")
Pattern 2: Retry with Exponential Backoff and Jitter
Prevents thundering herd by adding random jitter to exponential backoff delays. Distinguishes retriable vs non-retriable errors.
import time
import random
from typing import Type, Tuple, Callable, Any
from functools import wraps
class RetriableError(Exception):
"""Mark an exception as eligible for retry."""
pass
class NonRetriableError(Exception):
"""Mark an exception as permanently failed — do not retry."""
pass
def with_retry(
max_retries: int = 3,
base_delay: float = 1.0,
max_delay: float = 60.0,
retriable_exceptions: Tuple[Type[Exception], ...] = (RetriableError,),
jitter: bool = True,
):
"""Retry decorator with exponential backoff and optional jitter.
Delay formula: min(base_delay * (2 ^ attempt) + jitter_random, max_delay)
Jitter uses random.uniform(0, delay / 2) to prevent synchronized retries
from all clients simultaneously when the downstream service recovers.
"""
def decorator(func: Callable) -> Callable:
@wraps(func)
def wrapper(*args: Any, **kwargs: Any) -> Any:
last_exception = None
for attempt in range(max_retries + 1):
try:
return func(*args, **kwargs)
except NonRetriableError:
raise # Never retry non-retriable errors
except retriable_exceptions as e:
last_exception = e
if attempt == max_retries:
raise # Exhausted all retries
# Calculate delay with exponential backoff
delay = min(base_delay * (2 ** attempt), max_delay)
# Add jitter to prevent thundering herd
if jitter:
delay += random.uniform(0, delay / 2)
time.sleep(delay)
raise RuntimeError(f"Failed after {max_retries} retries") from last_exception
return wrapper
return decorator
# Classification helper for HTTP responses
def classify_http_error(status_code: int, response_body: str) -> Exception:
"""Classify HTTP errors as retriable or non-retriable.
Retriable: 429 (Too Many Requests), 500/502/503/504 (Server errors)
Non-retriable: 4xx client errors except 429, connection resets
"""
retriable_codes = {429, 500, 502, 503, 504}
if status_code == 401 or status_code == 403:
return NonRetriableError(f"Authentication failed — do not retry: HTTP {status_code}")
if status_code in retriable_codes:
return RetriableError(f"Transient server error: HTTP {status_code} — {response_body[:200]}")
if 400 <= status_code < 500:
return NonRetriableError(f"Client error — do not retry: HTTP {status_code}")
return RetriableError(f"Unexpected status code: {status_code}")
# ❌ BAD: Fixed-delay retry without jitter causes thundering herd when all clients
# retry simultaneously upon service recovery, creating a secondary outage.
def bad_fixed_retry(url: str, payload: dict) -> dict:
"""Fixed 5-second delay between retries — never use in production."""
import httpx, time
client = httpx.Client(timeout=10.0)
for attempt in range(4): # 3 retries + 1 attempt
try:
response = client.post(url, json=payload)
if response.status_code < 500:
return response.json()
except Exception:
pass
time.sleep(5.0) # ❌ FIXED delay — every client sends retry at exactly T+5, T+10, T+15
finally:
client.close()
raise RuntimeError("All retries exhausted")
# ✅ GOOD: Exponential backoff with random jitter prevents thundering herd.
# Delay = base_delay * 2^attempt + uniform(0, delay/2), capped at max_delay.
@with_retry(
max_retries=3,
base_delay=0.5,
max_delay=30.0,
jitter=True,
)
def call_downstream_service(url: str, payload: dict) -> dict:
"""Production-ready retry with exponential backoff and random jitter."""
import httpx
client = httpx.Client(timeout=10.0)
try:
response = client.post(url, json=payload)
error = classify_http_error(response.status_code, response.text)
if isinstance(error, NonRetriableError):
raise NonRetriableError(str(error))
if response.status_code >= 500 or response.status_code == 429:
raise RetriableError(str(error))
response.raise_for_status()
return response.json()
finally:
client.close()
Pattern 3: Bulkhead Isolation
Thread-pool-based bulkhead isolation prevents one slow downstream service from exhausting all worker threads and cascading failure.
import threading
from concurrent.futures import ThreadPoolExecutor, Future, TimeoutError as FuturesTimeout
from typing import Callable, Any, Optional
from dataclasses import dataclass, field
@dataclass
class BulkheadMetrics:
"""Track bulkhead utilization for observability."""
active_calls: int = 0
rejected_calls: int = 0
completed_calls: int = 0
@property
def utilization(self) -> float:
return self.active_calls / self.max_capacity if self.max_capacity > 0 else 0.0
class BulkheadFullError(Exception):
"""Raised when the bulkhead pool is at capacity and rejecting new calls."""
pass
class Bulkhead:
"""Thread-pool-based bulkhead isolation per downstream dependency.
Each critical service gets its own Bulkhead instance with independent
max_concurrent_calls and queue_size limits. When the pool is full,
callers get an immediate rejection (REJECT policy) rather than queuing
indefinitely and creating a cascading timeout chain.
Example: PaymentService has 20 threads, SearchService has 50 threads.
A slow SearchService exhausting its pool does NOT block payment processing.
"""
def __init__(
self,
max_concurrent_calls: int = 10,
queue_size: int = 0,
rejection_policy: str = "raise",
name: str = "default",
):
self.max_concurrent_calls = max_concurrent_calls
self.queue_size = queue_size
self.rejection_policy = rejection_policy
self.name = name
self._executor = ThreadPoolExecutor(
max_workers=max_concurrent_calls,
thread_name_prefix=f"bulkhead-{name}",
)
self._semaphore = threading.Semaphore(max_concurrent_calls)
self.metrics = BulkheadMetrics()
self.metrics.max_capacity = max_concurrent_calls
def execute(self, func: Callable, *args: Any, timeout: Optional[float] = None, **kwargs: Any) -> Any:
"""Execute func through the bulkhead with concurrency limiting.
Raises:
BulkheadFullError: When pool is at capacity and rejection policy is 'raise'.
FuturesTimeout: When the callable does not complete within the timeout.
"""
if not self._semaphore.acquire(blocking=False):
self.metrics.rejected_calls += 1
if self.rejection_policy == "raise":
raise BulkheadFullError(
f"Bulkhead '{self.name}' is full ({self.max_concurrent_calls} concurrent calls). "
f"Rejected. Active: {self.metrics.active_calls}"
)
elif self.rejection_policy == "queue":
queued = self._semaphore.acquire(blocking=True, timeout=timeout or 5.0)
if not queued:
self.metrics.rejected_calls += 1
raise BulkheadFullError(f"Bulkhead '{self.name}' queue timed out after {timeout}s")
future: Future = self._executor.submit(func, *args, **kwargs)
try:
self.metrics.active_calls += 1
if timeout is not None:
return future.result(timeout=timeout)
return future.result()
finally:
self.metrics.active_calls -= 1
self.metrics.completed_calls += 1
self._semaphore.release()
def shutdown(self, wait: bool = True) -> None:
"""Clean up the executor. Call during graceful shutdown."""
self._executor.shutdown(wait=wait)
@property
def available_capacity(self) -> int:
return self.max_concurrent_calls - self.metrics.active_calls
# Practical bulkhead setup for a service with multiple downstream dependencies
class ResilientOrderService:
"""Orders service with bulkhead isolation per downstream dependency."""
def __init__(self):
# Each dependency gets its own resource pool — a slow inventory check
# does NOT exhaust payment processing threads
self._payment_bulkhead = Bulkhead(
max_concurrent_calls=30,
queue_size=10,
rejection_policy="raise",
name="payments",
)
self._inventory_bulkhead = Bulkhead(
max_concurrent_calls=50,
queue_size=20,
rejection_policy="raise",
name="inventory",
)
self._notification_bulkhead = Bulkhead(
max_concurrent_calls=10,
queue_size=0,
rejection_policy="raise",
name="notifications",
)
def create_order(self, user_id: int, items: list[dict]) -> dict:
"""Process an order with bulkhead-protected downstream calls."""
# Check inventory in its own thread pool
inventory_result = self._inventory_bulkhead.execute(
self._check_inventory, items, timeout=3.0
)
if not inventory_result.available:
raise ValueError(f"Items unavailable: {inventory_result.unavailable_items}")
# Process payment in a separate thread pool
payment_result = self._payment_bulkhead.execute(
self._process_payment, user_id, items, timeout=10.0
)
# Fire-and-forget notification in yet another pool
try:
self._notification_bulkhead.execute(
self._send_confirmation, user_id, payment_result.order_id
)
except BulkheadFullError:
# Notification is optional — log and continue, do not fail the order
pass
return {
"order_id": payment_result.order_id,
"status": "confirmed",
"items": items,
}
def _check_inventory(self, items: list[dict]) -> dict: ...
def _process_payment(self, user_id: int, items: list[dict]) -> dict: ...
def _send_confirmation(self, user_id: int, order_id: str) -> None: ...
Pattern 4: Health Check Patterns (Liveness vs Readiness)
Kubernetes liveness probes restart unhealthy processes; readiness probes control traffic routing. Implement both separately — they answer different questions.
import time
from enum import Enum
from dataclasses import dataclass, field
from typing import Callable, Any
class ProbeType(Enum):
STARTUP = "startup"
LIVENESS = "liveness"
READINESS = "readiness"
@dataclass
class HealthStatus:
"""Unified health check result for all probe types."""
status: str = "healthy" # "healthy", "unhealthy", "degraded"
probe_type: ProbeType = ProbeType.LIVENESS
details: dict[str, Any] = field(default_factory=dict)
timestamp: float = field(default_factory=time.monotonic)
dependencies_ok: bool = True
def unhealthy(self, reason: str, **extra) -> "HealthStatus":
self.status = "unhealthy"
self.dependencies_ok = False
self.details["reason"] = reason
self.details.update(extra)
return self
def degraded(self, reason: str, **extra) -> "HealthStatus":
self.status = "degraded"
self.details["reason"] = reason
self.details.update(extra)
return self
class HealthCheckRegistry:
"""Central health check registry supporting startup, liveness, and readiness probes.
Kubernetes configuration:
- startupProbe: initialDelaySeconds=30, periodSeconds=5
Prevents premature restart during slow initialization
- livenessProbe: initialDelaySeconds=15, periodSeconds=10
Returns 503 if process is stuck — K8s restarts the pod
- readinessProbe: periodSeconds=5, successThreshold=1, failureThreshold=3
Returns 503 to remove from load balancer pool
"""
def __init__(self):
self._checkers: dict[str, Callable[[], HealthStatus]] = {}
def register(self, name: str, checker: Callable[[], HealthStatus], probe_type: ProbeType = ProbeType.LIVENESS) -> None:
self._checkers[name] = (checker, probe_type)
def check_liveness(self) -> HealthStatus:
"""Liveness check: is the process alive and not in a bad state?
Must be fast (< 500ms). Do NOT check slow downstream dependencies here —
liveness failures cause pod restarts which amplify load on those same dependencies.
If this service's database connection pool is exhausted, that's a readiness issue,
not a liveness issue. The process can still function once the pool recovers.
"""
status = HealthStatus(probe_type=ProbeType.LIVENESS)
# Check internal state: thread deadlocks, memory pressure
import threading
active_threads = threading.active_count()
if active_threads > 200:
return status.unhealthy(
"Thread count critically high",
active_threads=active_threads,
)
for name, (checker, probe_type) in self._checkers.items():
if probe_type == ProbeType.LIVENESS:
try:
result = checker()
if not status.dependencies_ok:
return result # Liveness failed — fail fast
except Exception as e:
return status.unhealthy(f"Liveness check '{name}' threw", error=str(e))
return status
def check_readiness(self) -> HealthStatus:
"""Readiness check: can this service handle traffic?
MUST verify all critical downstream dependencies. A service that reports
healthy but cannot reach its database will receive broken requests from
the load balancer and contribute to user-visible failures.
"""
status = HealthStatus(probe_type=ProbeType.READINESS)
for name, (checker, probe_type) in self._checkers.items():
if probe_type == ProbeType.READINESS:
try:
result = checker()
if not result.dependencies_ok:
return result # Dependency failed — not ready for traffic
except Exception as e:
return status.unhealthy(f"Readiness check '{name}' threw", error=str(e))
return status
# Example: Flask/Sanic-style HTTP handlers for Kubernetes probes
from flask import Flask, jsonify
app = Flask(__name__)
health_registry = HealthCheckRegistry()
# Register a database health checker as readiness probe
def db_health_check() -> HealthStatus:
"""Verify the database connection pool has working connections."""
try:
# Simple query that verifies connectivity and permission
with db_pool.connection() as conn:
conn.execute("SELECT 1")
return HealthStatus(probe_type=ProbeType.READINESS, status="healthy", dependencies_ok=True)
except Exception:
return HealthStatus(
probe_type=ProbeType.READINESS,
status="unhealthy",
dependencies_ok=False,
reason="database_unreachable",
)
health_registry.register("database", db_health_check, ProbeType.READINESS)
# Register application-specific liveness check
def app_liveness_check() -> HealthStatus:
"""Verify the application process is not in a zombie state."""
return HealthStatus(probe_type=ProbeType.LIVENESS, status="healthy")
health_registry.register("app", app_liveness_check, ProbeType.LIVENESS)
@app.get("/startup")
def startup_probe():
"""Kubernetes startup probe — returns 200 when initialization is complete."""
if not app.initialized:
return jsonify({"status": "initializing"}), 503
return jsonify({"status": "ready"}), 200
@app.get("/healthz")
def liveness_probe():
"""Kubernetes liveness probe — returns 503 when process is broken."""
result = health_registry.check_liveness()
code = 200 if result.status == "healthy" else 503
return jsonify(result.details), code
@app.get("/ready")
def readiness_probe():
"""Kubernetes readiness probe — returns 503 when not ready to receive traffic."""
result = health_registry.check_readiness()
code = 200 if result.dependencies_ok else 503
return jsonify(result.details), code
# Kubernetes probe configuration example:
"""
livenessProbe:
httpGet:
path: /healthz
port: 8080
initialDelaySeconds: 15
periodSeconds: 10
failureThreshold: 3 # 3 consecutive failures → restart pod
readinessProbe:
httpGet:
path: /ready
port: 8080
initialDelaySeconds: 5
periodSeconds: 5
failureThreshold: 3 # 3 consecutive failures → remove from service endpoints
startupProbe:
httpGet:
path: /startup
port: 8080
failureThreshold: 30 # Allow up to 150 seconds for startup (30 * 5s)
periodSeconds: 5
"""
Pattern 5: Graceful Degradation Strategies
Feature flag-based degradation system that serves fallback responses when downstream dependencies fail, prioritized by data freshness requirements.
import time
from enum import IntEnum
from typing import Any, Optional
from dataclasses import dataclass, field
class FallbackPriority(IntEnum):
"""Priority ordering for fallback responses. Lower number = higher priority."""
STALE_CACHE = 1 # Cached data with TTL — serve if available and not expired
DEFAULT_VALUE = 2 # Sensible defaults (e.g., empty list, zero count)
CACHED_LEGACY = 3 # Older cached version from long-term storage
MAINTENANCE_MESSAGE = 4 # User-facing message explaining the limitation
@dataclass
class FallbackResponse:
"""Wrapped response with metadata about its provenance."""
data: Any
source: str # "fresh", "stale_cache", "default", etc.
priority: FallbackPriority = FallbackPriority.STALE_CACHE
staleness_seconds: float = 0.0
is_degraded: bool = False
@property
def fresh(self) -> bool:
return self.source == "fresh"
class DegradationManager:
"""Manages fallback strategies for downstream dependency failures.
Each protected service has a priority-ordered list of fallback responses.
When the primary call fails, the manager tries each fallback in order until
one succeeds. This ensures users always see SOMETHING — even if degraded —
rather than a hard error.
"""
def __init__(self):
self._strategies: dict[str, list[Callable]] = {}
def register(self, service_name: str, fallback_fn: Callable) -> None:
"""Register a fallback function for a specific downstream service."""
if service_name not in self._strategies:
self._strategies[service_name] = []
self._strategies[service_name].append(fallback_fn)
def serve_with_fallback(self, service_name: str, primary_fn: Callable, *args, **kwargs) -> FallbackResponse:
"""Execute primary function; fall back through registered handlers on failure.
Returns the highest-priority successful response from any source.
Raises only if ALL fallbacks also fail.
"""
# Try the primary path first (fresh data)
try:
result = primary_fn(*args, **kwargs)
return FallbackResponse(data=result, source="fresh", is_degraded=False)
except Exception as primary_error:
pass
# Try fallbacks in registration order (highest priority first)
fallbacks = self._strategies.get(service_name, [])
last_error = primary_error
for fallback_fn in fallbacks:
try:
result = fallback_fn(*args, **kwargs)
return FallbackResponse(
data=result,
source=f"{service_name}_fallback",
priority=FallbackPriority.STALE_CACHE,
is_degraded=True,
)
except Exception as fallback_error:
last_error = fallback_error
continue
raise RuntimeError(
f"All paths failed for '{service_name}'. Primary error: {primary_error}. "
f"Last fallback error: {last_error}"
)
# Practical example: Product catalog with 3-level degradation
class ProductService:
"""Product service with graceful degradation across cache tiers."""
def __init__(self, degradation_mgr: DegradationManager):
self.degradation = degradation_mgr
self.degradation.register("product_catalog", self._get_stale_cache)
self.degradation.register("product_catalog", self._get_default_category)
def get_product(self, product_id: str) -> FallbackResponse:
"""Get product details — always returns something if possible."""
return self.degradation.serve_with_fallback(
service_name="product_catalog",
primary_fn=self._fetch_from_database,
product_id=product_id,
)
def get_category_list(self) -> FallbackResponse:
"""Get category list — serve stale cache if database is down."""
return self.degradation.serve_with_fallback(
service_name="category_list",
primary_fn=self._fetch_categories_db,
)
def _fetch_from_database(self, product_id: str) -> dict:
"""Primary: query the production database."""
...
def _get_stale_cache(self, product_id: str) -> dict:
"""Fallback 1: serve from Redis cache if TTL hasn't expired."""
cached = redis_client.get(f"product:{product_id}")
if cached and not self._is_expired(cached):
return cached
raise Exception("Stale cache miss or expired")
def _get_default_category(self) -> list[str]:
"""Fallback 2: serve a hardcoded default category list."""
return ["Electronics", "Books", "Home & Garden"]
@staticmethod
def _is_expired(cached_data: bytes) -> bool:
data = cached_data.decode()
# Parse timestamp embedded in cache value
ttl_seconds = 300 # 5-minute TTL
return (time.monotonic() - float(data.split("|")[1])) > ttl_seconds
Pattern 6: Distributed Idempotency
Idempotency keys ensure that duplicate or retried requests produce the same result without side effects. Uses a store with TTL to auto-expire old keys.
import time
import hashlib
from typing import Any, Optional
from dataclasses import dataclass
from enum import Enum
class IdempotencyStatus(Enum):
PENDING = "pending"
COMPLETED = "completed"
FAILED = "failed"
@dataclass
class IdempotencyRecord:
"""Stored idempotency record for a request key."""
request_key: str
response_data: Any
status: IdempotencyStatus
created_at: float = field(default_factory=time.monotonic)
expires_at: float = 0.0
@property
def is_expired(self) -> bool:
return self.expires_at > 0 and time.monotonic() > self.expires_at
class IdempotencyStore:
"""Abstract interface for idempotency key storage.
Production implementations use Redis (with TTL auto-expiry) or
a dedicated database table with periodic cleanup jobs.
"""
def get(self, key: str) -> Optional[IdempotencyRecord]:
raise NotImplementedError
def put(self, record: IdempotencyRecord) -> None:
raise NotImplementedError
def delete(self, key: str) -> None:
raise NotImplementedError
class RedisIdempotencyStore(IdempotencyStore):
"""Redis-backed idempotency store with automatic TTL expiry.
Key format: `idemp:{hash_of_key}` — prevents key collision attacks.
Value: JSON-serialized IdempotencyRecord.
TTL: configured per-record (default 24 hours), allowing retries within that window
while preventing indefinite storage growth.
"""
def __init__(self, redis_client, default_ttl: int = 86400):
self._redis = redis_client
self._default_ttl = default_ttl
def get(self, key: str) -> Optional[IdempotencyRecord]:
raw = self._redis.get(f"idemp:{self._hash(key)}")
if raw is None:
return None
import json
data = json.loads(raw.decode())
return IdempotencyRecord(**data)
def put(self, record: IdempotencyRecord) -> None:
import json
key = f"idemp:{self._hash(record.request_key)}"
data = {
"request_key": record.request_key,
"response_data": record.response_data,
"status": record.status.value,
"created_at": record.created_at,
"expires_at": record.expires_at,
}
self._redis.setex(key, int(record.expires_at - time.monotonic()), json.dumps(data))
def delete(self, key: str) -> None:
self._redis.delete(f"idemp:{self._hash(key)}")
@staticmethod
def _hash(raw_key: str) -> str:
return hashlib.sha256(raw_key.encode()).hexdigest()[:32]
class IdempotencyEngine:
"""Enforces idempotency on API endpoints using request-specific keys.
Workflow:
1. Client sends `Idempotency-Key` header with a UUID
2. Engine checks store for existing record with this key
3. If found and completed → returns the original response immediately
4. If not found → executes the handler, stores result, returns it
5. Store auto-expires records after TTL to prevent unbounded growth
This protects against:
- Network retries (client resends because it didn't get a response)
- Webhook delivery duplicates (provider sends same event twice)
- Client-side duplicate submission (double-click on "Pay" button)
"""
def __init__(self, store: IdempotencyStore, default_ttl: int = 86400):
self._store = store
self._default_ttl = default_ttl
def execute(
self,
handler: Any,
idempotency_key: str,
ttl: Optional[int] = None,
*args: Any,
**kwargs: Any,
) -> dict[str, Any]:
"""Execute a handler with idempotency protection.
Args:
handler: Callable that performs the actual work (side effects).
idempotency_key: Unique key from client (UUID recommended).
ttl: Time-to-live in seconds for this key (default: 24h).
Returns:
Dict with 'status' ('replayed' or 'executed'), 'result', and 'idempotency_key'.
Raises:
IdempotencyKeyRequired: If no idempotency key was provided.
"""
if not idempotency_key:
raise ValueError("Idempotency-Key header is required for write operations")
ttl = ttl or self._default_ttl
store_key = idempotency_key
# Check for existing record
existing = self._store.get(store_key)
if existing and not existing.is_expired:
if existing.status == IdempotencyStatus.COMPLETED:
return {"status": "replayed", "result": existing.response_data, "idempotency_key": store_key}
elif existing.status == IdempotencyStatus.FAILED:
# Retry failed requests — the original handler might succeed this time
pass
# Execute handler and store result
try:
result = handler(*args, **kwargs)
record = IdempotencyRecord(
request_key=store_key,
response_data=result,
status=IdempotencyStatus.COMPLETED,
created_at=time.monotonic(),
expires_at=time.monotonic() + ttl,
)
self._store.put(record)
return {"status": "executed", "result": result, "idempotency_key": store_key}
except Exception as e:
# Store failure so retries don't re-attempt (prevents retry storms on known failures)
record = IdempotencyRecord(
request_key=store_key,
response_data={"error": str(e)},
status=IdempotencyStatus.FAILED,
created_at=time.monotonic(),
expires_at=time.monotonic() + ttl,
)
self._store.put(record)
raise
def cleanup_expired(self) -> int:
"""Remove expired records from the store. Call periodically or rely on Redis TTL."""
# Implementation depends on storage backend
return 0
# Usage with Flask — extract key from header and protect write endpoints
@app.post("/orders")
def create_order():
idempotency_key = request.headers.get("Idempotency-Key")
if not idempotency_key:
return jsonify({"error": "Idempotency-Key header required"}), 400
engine = IdempotencyEngine(RedisIdempotencyStore(redis_client))
try:
result = engine.execute(
handler=order_service.c
…(truncated)