# System Reliability Architecture > Implements production reliability patterns (circuit breakers, retry with exponential backoff, bulkhead isolation, health checks, graceful degradation, distributed tracing) to build fault-tolerant distributed systems. - Skill: `paulpas/system-reliability-architecture` (Agent Skill) - Install (CLI): `npx skillmds@latest add paulpas/system-reliability-architecture` - Raw SKILL.md: https://api.skillmd.com/api/skills/paulpas/system-reliability-architecture/raw - Safety review: pending - Works with: Claude Code, Claude.ai, OpenAI Codex - Category: Coding & Dev Tools - License: MIT - Author: paulpas (https://skillmd.com/u/paulpas) - Updated: 2026-09-17 - Page: https://skillmd.com/skills/paulpas/system-reliability-architecture --- # 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 1. **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. 2. **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. 3. **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. 4. **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 `BulkheadFullException` vs. 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. 5. **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. 6. **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. 7. **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. ```python 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. ```python 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. ```python 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. ```python 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. ```python 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. ```python 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)