计划执行与检查节点
executing-plans
按既定实施计划完成工作,保持过程检查和反馈。
microservice-resilience-patterns【效率与协作】Implements production-ready resilience patterns (circuit breaker, retry with exponential backoff, bulkhead isolation, timeout enforcement, graceful fallback) to prevent cascading failures in distributed microservice systems.
独立 GitHub 快照 · 自动分类 · 以原作者文档为准
name: microservice-resilience-patterns description: Implements production-ready resilience patterns (circuit breaker, retry with exponential backoff, bulkhead isolation, timeout enforcement, graceful fallback) to prevent cascading failures in distributed microservice systems. license: MIT compatibility: opencode metadata: version: "1.0.0" domain: coding triggers: circuit breaker, retry pattern, bulkhead pattern, resilience, fault tolerance, timeout handling, graceful degradation, fallback strategy timeout handling archetypes:
Implements production-ready resilience patterns to prevent cascading failures in distributed microservice systems. When loaded, this skill makes the model design, implement, and validate circuit breakers, retry strategies with jitter, bulkhead isolation, timeout enforcement, and graceful degradation mechanisms — all tailored to the specific failure modes of the target architecture.
random.uniform for avalanche preventionUse this skill when:
Avoid this skill for:
idempotent-distributed-operations insteadobservability-patterns for those concerns; resilience and observability are complementary but distinctMap Dependencies & Failure Modes — Audit every inter-service call in the system. Classify each dependency by its SLA tier (Tier-1: core revenue path, Tier-2: supporting functionality, Tier-3: best-effort). Identify single points of failure and latency-sensitive calls that could trigger cascade chains. Checkpoint: Produce a dependency matrix listing every service-to-service call, its timeout, retry policy, and circuit breaker configuration before proceeding.
Select Resilience Patterns Per Dependency — Apply the pattern selector based on each dependency's characteristics:
Implement Circuit Breaker Per Dependency — Deploy a three-state circuit breaker for every non-trivial outbound call. Configure failure_threshold (default 5), success_threshold (default 3 for half-open recovery), and timeout_seconds (default 30s). Integrate per-metric monitoring: success rate, failure count, state transitions, and latency percentiles. Checkpoint: Circuit breaker states are observable via metrics; state transitions log at INFO level with dependency name and reason.
Add Retry Logic with Exponential Backoff + Jitter — For idempotent operations, implement retry with a configurable max attempt count (default 3), base delay (default 1s), and exponential multiplier (default 2.0). Apply jitter using random.uniform(base_delay * factor / 2, base_delay * factor) to prevent thundering herd. Classify exceptions into retryable (timeout, connection error) vs non-retryable (4xx validation errors, serialization errors). Checkpoint: Retry logs capture attempt number, delay applied, and exception type. No retries on non-idempotent operations without explicit owner sign-off.
Apply Bulkhead Isolation to Contain Blast Radius — For high-cardinality dependencies or shared resources, create dedicated thread pools or executor groups per downstream service. Set explicit max_concurrent limits and overflow behavior (reject immediately vs queue with timeout). Use the bulkhead pattern on any endpoint where a single slow dependency could starve the entire thread pool. Checkpoint: Monitor bulkhead rejection rates; configure alerts when rejections exceed 1% of total calls to that dependency.
Enforce Timeouts & Implement Graceful Fallbacks — Wrap every outbound call in a hard timeout (connect + read). Define fallback behavior for each guarded call: cached data, stale values, default responses, or user-facing error messages. Never let a downstream timeout propagate as an unhandled exception to the client. Integrate fallback responses into metrics so degradation is visible. Checkpoint: Every protected endpoint returns a valid HTTP response (2xx/4xx) under any failure scenario — no 500s from cascaded timeouts.
A circuit breaker prevents repeated calls to a failing downstream service by transitioning through three states: Closed (normal, counting failures), Open (fast-fail after threshold exceeded), and Half-Open (testing recovery with limited probe traffic). This pattern follows the principle of fail-fast over slow degradation.
from enum import Enum
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from typing import Any, Callable, TypeVar
import threading
import time
T = TypeVar("T")
class CircuitState(Enum):
CLOSED = "closed" # Normal: requests pass through, failures counted
OPEN = "open" # Tripped: all requests fail fast immediately
HALF_OPEN = "half_open" # Testing recovery: allow limited probe requests
@dataclass
class CircuitBreakerConfig:
"""Configuration for a circuit breaker instance."""
failure_threshold: int = 5 # Failures before tripping to Open
success_threshold: int = 3 # Successes in Half-Open before closing
timeout_seconds: float = 30.0 # How long Open stays before moving to Half-Open
half_open_max_calls: int = 3 # Max probe calls allowed in Half-Open state
class CircuitBreakerError(Exception):
"""Raised when a call is rejected by an open circuit breaker."""
def __init__(self, dependency: str, elapsed_open_seconds: float) -> None:
self.dependency = dependency
self.elapsed_open_seconds = elapsed_open_seconds
super().__init__(
f"Circuit breaker OPEN for '{dependency}' — "
f"rejected after {elapsed_open_seconds:.1f}s in open state"
)
class CircuitBreaker:
"""Three-state circuit breaker with thread-safe state management.
Transitions:
Closed --(failure_threshold reached)--> Open
Open --(timeout expires)------------> Half-Open
Half-Open --(success_threshold met)--> Closed
Half-Open --(any failure)-------------> Open
"""
def __init__(self, name: str, config: CircuitBreakerConfig | None = None) -> None:
self.name = name
self._config = config or CircuitBreakerConfig()
self._state = CircuitState.CLOSED
self._failure_count: int = 0
self._success_count: int = 0
self._half_open_calls: int = 0
self._last_failure_time: datetime | None = None
self._opened_at: datetime | None = None
self._lock = threading.RLock()
@property
def state(self) -> CircuitState:
with self._lock:
# Auto-transition from Open to Half-Open when timeout expires
if self._state == CircuitState.OPEN and self._opened_at:
elapsed = (datetime.utcnow() - self._opened_at).total_seconds()
if elapsed >= self._config.timeout_seconds:
self._state = CircuitState.HALF_OPEN
self._success_count = 0
self._half_open_calls = 0
return self._state
@property
def failure_count(self) -> int:
with self._lock:
return self._failure_count
def record_success(self) -> None:
"""Record a successful call. Resets failure count in Closed state."""
with self._lock:
if self._state == CircuitState.CLOSED:
self._failure_count = 0
elif self._state == CircuitState.HALF_OPEN:
self._success_count += 1
if self._success_count >= self._config.success_threshold:
self._state = CircuitState.CLOSED
self._failure_count = 0
self._success_count = 0
def record_failure(self) -> None:
"""Record a failed call. May trip the circuit to Open or reopen from Half-Open."""
with self._lock:
now = datetime.utcnow()
self._last_failure_time = now
if self._state == CircuitState.CLOSED:
self._failure_count += 1
if self._failure_count >= self._config.failure_threshold:
self._state = CircuitState.OPEN
self._opened_at = now
elif self._state == CircuitState.HALF_OPEN:
# Any failure in Half-Open immediately re-trips the circuit
self._state = CircuitState.OPEN
self._opened_at = now
def allow_request(self) -> bool:
"""Check whether an outbound request is permitted.
Returns True if the call should proceed, False if it must be rejected.
Raises CircuitBreakerError when the circuit is tripped and caller should
invoke fallback logic immediately.
"""
current_state = self.state # Triggers Open→Half-Open transition check
if current_state == CircuitState.CLOSED:
return True
elif current_state == CircuitState.OPEN:
raise CircuitBreakerError(
dependency=self.name,
elapsed_open_seconds=(
(datetime.utcnow() - self._opened_at).total_seconds()
if self._opened_at else 0.0
),
)
else: # HALF_OPEN
with self._lock:
if self._half_open_calls < self._config.half_open_max_calls:
self._half_open_calls += 1
return True
return False
def execute(self, func: Callable[..., T], *args: Any, **kwargs: T) -> T:
"""Execute a function through the circuit breaker with automatic state management.
Usage:
cb = CircuitBreaker("payment-service")
result = cb.execute(call_payment_service, user_id=42)
"""
if not self.allow_request():
raise CircuitBreakerError(
dependency=self.name,
elapsed_open_seconds=(
(datetime.utcnow() - self._opened_at).total_seconds()
if self._opened_at else 0.0
),
)
try:
result = func(*args, **kwargs)
self.record_success()
return result
except Exception as exc:
self.record_failure()
raise
Below shows how to wire the circuit breaker into a real service call with fallback.
# breaker = CircuitBreaker("inventory-service", CircuitBreakerConfig(
# failure_threshold=3, success_threshold=2, timeout_seconds=15.0
# ))
# try:
# stock = breaker.execute(fetch_inventory, sku="PROD-123")
# except CircuitBreakerError:
# stock = get_cached_inventory(sku="PROD-123") # fallback
Retrying every failed call causes thundering herd problems. Exponential backoff spreads retries over time, while jitter adds randomness to prevent synchronized retry storms. This pattern is critical for maintaining stability under partial outages.
import asyncio
import random
import logging
from dataclasses import dataclass
from typing import Callable, TypeVar, Set, Type
T = TypeVar("T")
logger = logging.getLogger(__name__)
@dataclass
class RetryConfig:
"""Configuration for retry with exponential backoff and jitter."""
max_attempts: int = 3 # Total attempts including the first try
base_delay: float = 1.0 # Base delay in seconds
max_delay: float = 60.0 # Cap on delay to prevent excessive waits
backoff_multiplier: float = 2.0 # Multiplier applied each retry
jitter_factor: float = 0.5 # Randomness range: [0, jitter_factor * computed_delay]
retryable_exceptions: Set[Type[Exception]] | None = None
# Defaults to (TimeoutError, ConnectionError, OSError) if None
class RetryExhaustedError(Exception):
"""Raised when all retry attempts have been exhausted."""
def __init__(self, operation: str, final_exception: Exception,
attempts: int, total_elapsed_seconds: float) -> None:
self.operation = operation
self.final_exception = final_exception
self.attempts = attempts
self.total_elapsed_seconds = total_elapsed_seconds
super().__init__(
f"Operation '{operation}' exhausted {attempts} attempts "
f"over {total_elapsed_seconds:.1f}s — last error: {final_exception}"
)
def _compute_delay(attempt: int, config: RetryConfig) -> float:
"""Calculate delay for a given attempt number using exponential backoff with jitter.
Formula: min(base_delay * (multiplier ** attempt), max_delay) + random jitter.
The jitter is uniformly distributed in [0, jitter_factor * computed_base_delay].
"""
computed = config.base_delay * (config.backoff_multiplier ** attempt)
capped = min(computed, config.max_delay)
jitter = random.uniform(0, config.jitter_factor * capped)
return capped + jitter
async def retry_with_backoff(
func: Callable[..., T],
*args: Any,
config: RetryConfig | None = None,
operation_name: str = "unknown",
**kwargs: Any,
) -> T:
"""Execute an async function with retry logic using exponential backoff and jitter.
Args:
func: The async callable to retry.
*args: Positional arguments passed to the function.
config: Retry configuration. Uses defaults if None.
operation_name: Human-readable label for logging and error reporting.
**kwargs: Keyword arguments passed to the function.
Returns:
The result of the successful call.
Raises:
RetryExhaustedError: If all attempts fail with retryable exceptions.
Exception: Any non-retryable exception is raised immediately (no retry).
Usage:
config = RetryConfig(max_attempts=5, base_delay=0.5, max_delay=30.0)
result = await retry_with_backoff(
fetch_user_profile, user_id=42, config=config, operation_name="profile-api"
)
"""
cfg = config or RetryConfig()
# Set default retryable exceptions if none specified
if cfg.retryable_exceptions is None:
retryable = {TimeoutError, ConnectionError, OSError}
else:
retryable = cfg.retryable_exceptions
last_exception: Exception | None = None
total_elapsed = 0.0
for attempt in range(cfg.max_attempts):
start_time = time.monotonic()
try:
# Support both sync and async callables
import inspect
if asyncio.iscoroutinefunction(func):
result = await func(*args, **kwargs)
else:
result = func(*args, **kwargs)
return result
except Exception as exc:
last_exception = exc
elapsed = time.monotonic() - start_time
total_elapsed += elapsed
# Non-retryable exception — fail fast immediately
if not isinstance(exc, retryable):
logger.error(
"Non-retryable exception '%s' for '%s' — aborting without retry",
type(exc).__name__, operation_name, exc_info=exc,
)
raise
# Log the failure and remaining attempts
logger.warning(
"Attempt %d/%d failed for '%s': %s (%.1fs elapsed) — "
"retrying in %.2fs",
attempt + 1, cfg.max_attempts, operation_name,
type(exc).__name__, elapsed, _compute_delay(attempt, cfg),
)
# Don't sleep after the last attempt
if attempt < cfg.max_attempts - 1:
delay = _compute_delay(attempt, cfg)
await asyncio.sleep(delay)
raise RetryExhaustedError(
operation=operation_name,
final_exception=last_exception,
attempts=cfg.max_attempts,
total_elapsed_seconds=total_elapsed,
)
# --- Usage example ---
# config = RetryConfig(max_attempts=5, base_delay=0.5, max_delay=30.0)
# result = await retry_with_backoff(
# fetch_user_profile, user_id=42, config=config, operation_name="profile-api"
# )
For non-async contexts (e.g., synchronous HTTP clients, Celery tasks), use this variant:
# --- Synchronous variant for non-async contexts ---
def retry_with_backoff_sync(
func: Callable[..., T],
*args: Any,
config: RetryConfig | None = None,
operation_name: str = "unknown",
**kwargs: Any,
) -> T:
"""Synchronous version of retry_with_backoff for use outside async contexts."""
cfg = config or RetryConfig()
if cfg.retryable_exceptions is None:
retryable = {TimeoutError, ConnectionError, OSError}
else:
retryable = cfg.retryable_exceptions
last_exception: Exception | None = None
total_elapsed = 0.0
for attempt in range(cfg.max_attempts):
start_time = time.monotonic()
try:
result = func(*args, **kwargs)
return result
except Exception as exc:
last_exception = exc
elapsed = time.monotonic() - start_time
total_elapsed += elapsed
if not isinstance(exc, retryable):
logger.error(
"Non-retryable exception '%s' for '%s' — aborting",
type(exc).__name__, operation_name, exc_info=exc,
)
raise
logger.warning(
"Attempt %d/%d failed for '%s': %s — retrying in %.2fs",
attempt + 1, cfg.max_attempts, operation_name,
type(exc).__name__, _compute_delay(attempt, cfg),
)
if attempt < cfg.max_attempts - 1:
time.sleep(_compute_delay(attempt, cfg))
raise RetryExhaustedError(
operation=operation_name,
final_exception=last_exception,
attempts=cfg.max_attempts,
total_elapsed_seconds=total_elapsed,
)
Bulkhead isolation partitions thread pools or executor groups so that a failure in one dependency does not exhaust resources available to others. This is critical for the Single Responsibility Principle and preventing thread pool starvation.
import asyncio
import threading
from concurrent.futures import ThreadPoolExecutor, Future
from dataclasses import dataclass
from typing import Callable, Any, TypeVar
from enum import Enum
T = TypeVar("T")
class BulkheadPolicy(Enum):
REJECT_IMMEDIATELY = "reject_immediately" # Fast-fail when pool exhausted
QUEUE_WITH_TIMEOUT = "queue_with_timeout" # Queue with bounded wait
@dataclass
class BulkheadConfig:
"""Configuration for a bulkhead isolation boundary."""
max_concurrent: int = 10 # Max concurrent calls to this dependency
queue_size: int = 20 # Max queued requests (for QUEUE_WITH_TIMEOUT)
queue_timeout_seconds: float = 5.0 # How long a queued request waits
policy: BulkheadPolicy = BulkheadPolicy.REJECT_IMMEDIATELY
class BulkheadRejectedError(Exception):
"""Raised when the bulkhead cannot accept a new call."""
def __init__(self, dependency: str, concurrent_count: int, max_concurrent: int) -> None:
self.dependency = dependency
self.concurrent_count = concurrent_count
self.max_concurrent = max_concurrent
super().__init__(
f"Bulkhead for '{dependency}' full — "
f"{concurrent_count}/{max_concurrent} slots in use, policy rejects immediately"
)
class Bulkhead:
"""Thread-pool bulkhead that isolates a single downstream dependency.
Each Bulkhead owns its own ThreadPoolExecutor, ensuring that slow or
failed calls to one dependency never consume threads from other services.
Usage:
bulkhead = Bulkhead("payment-api", max_concurrent=8)
result = bulkhead.execute(process_payment, order_id=12345)
bulkhead.shutdown()
"""
def __init__(
self,
dependency_name: str,
config: BulkheadConfig | None = None,
) -> None:
self._config = config or BulkheadConfig()
self._dependency = dependency_name
self._executor: ThreadPoolExecutor = ThreadPoolExecutor(
max_workers=self._config.max_concurrent,
thread_name_prefix=f"bulkhead-{self._dependency}",
)
self._semaphore = threading.Semaphore(self._config.max_concurrent)
self._active_count: int = 0
self._rejected_count: int = 0
self._lock = threading.Lock()
@property
def active_count(self) -> int:
with self._lock:
return self._active_count
@property
def rejected_count(self) -> int:
with self._lock:
return self._rejected_count
def execute(self, func: Callable[..., T], *args: Any, **kwargs: Any) -> T:
"""Execute a callable within this bulkhead's isolation boundary.
If the maximum concurrent count is reached:
- REJECT_IMMEDIATELY: raises BulkheadRejectedError immediately
- QUEUE_WITH_TIMEOUT: queues the call and waits up to queue_timeout_seconds
"""
if self._config.policy == BulkheadPolicy.REJECT_IMMEDIATELY:
acquired = self._semaphore.acquire(timeout=0)
if not acquired:
with self._lock:
self._rejected_count += 1
raise BulkheadRejectedError(
dependency=self._dependency,
concurrent_count=self._active_count,
max_concurrent=self._config.max_concurrent,
)
else:
acquired = self._semaphore.acquire(
timeout=self._config.queue_timeout_seconds
)
if not acquired:
with self._lock:
self._rejected_count += 1
raise TimeoutError(
f"Bulkhead queue timeout for '{self._dependency}' "
f"after {self._config.queue_timeout_seconds}s"
)
try:
future = self._executor.submit(func, *args, **kwargs)
with self._lock:
self._active_count += 1
try:
return future.result()
finally:
with self._lock:
self._active_count -= 1
finally:
self._semaphore.release()
Without bulkheads, all services share a single thread pool. A slow downstream call starves everything else.
# ❌ BAD — No bulkhead isolation. All services share one thread pool.
# A slow inventory service can starve the entire application of threads.
shared_executor = ThreadPoolExecutor(max_workers=20)
def handle_order_request_bad(order_id: str) -> dict[str, Any]:
"""Processes an order with NO bulkhead isolation."""
# If inventory-service hangs, payment-service threads are blocked too
stock = shared_executor.submit(fetch_inventory, sku="PROD-1").result(timeout=60)
payment = shared_executor.submit(process_payment, order_id=order_id).result(timeout=30)
shipping = shared_executor.submit(calculate_shipping, items=stock).result(timeout=30)
return {"stock": stock, "payment": payment, "shipping": shipping}
Each dependency gets its own thread pool with bounded concurrency. Failures are contained to their own bulkhead.
# ✅ GOOD — Bulkhead isolation per dependency. Each service has its own pool.
inventory_bulkhead = Bulkhead("inventory-service", BulkheadConfig(max_concurrent=5))
payment_bulkhead = Bulkhead("payment-service", BulkheadConfig(max_concurrent=8))
shipping_bulkhead = Bulkhead("shipping-service", BulkheadConfig(max_concurrent=4))
def handle_order_request_good(order_id: str) -> dict[str, Any]:
"""Processes an order with bulkhead isolation per dependency."""
# Each service is isolated — inventory timeout does not affect payment threads
try:
stock = inventory_bulkhead.execute(fetch_inventory, sku="PROD-1")
except (BulkheadRejectedError, TimeoutError):
stock = get_cached_stock_fallback("PROD-1") # Graceful degradation
try:
payment = payment_bulkhead.execute(process_payment, order_id=order_id)
except (BulkheadRejectedError, TimeoutError):
raise PaymentUnavailableError("Payment service is at capacity")
try:
shipping = shipping_bulkhead.execute(calculate_shipping, items=[stock])
except (BulkheadRejectedError, TimeoutError):
shipping = {"rate": 0, "carrier": "unknown", "estimated_days": None} # Default fallback
return {"stock": stock, "payment": payment, "shipping": shipping}
Hard timeouts prevent indefinite waits on slow dependencies. Every outbound call must have both a connect timeout and a read deadline. When a timeout occurs, the system falls back to a pre-defined degradation strategy rather than propagating raw errors.
import asyncio
from dataclasses import dataclass
from typing import Any, Callable, TypeVar
T = TypeVar("T")
@dataclass
class TimeoutFallbackConfig:
"""Defines timeout boundaries and fallback behavior for an async call."""
connect_timeout: float = 2.0 # Time to establish connection
read_timeout: float = 10.0 # Time to receive first response byte
total_timeout: float = 15.0 # Absolute ceiling for the entire call
has_fallback: bool = True
fallback_source: str = "cache" # Where to get degraded data
class CallTimeoutError(Exception):
"""Raised when a call exceeds its timeout boundary."""
pass
async def enforce_timeout_with_fallback(
operation: Callable[..., Any],
*args: Any,
connect_timeout: float = 2.0,
read_timeout: float = 10.0,
total_timeout: float = 15.0,
fallback_fn: Callable[..., T] | None = None,
fallback_source_name: str = "cache",
operation_label: str = "unknown",
) -> T:
"""Execute an async operation with layered timeout enforcement and graceful fallback.
Applies three layers of timeout:
1. Connect timeout — time to establish the network connection
2. Read timeout — time to receive the first byte of response
3. Total timeout — absolute ceiling regardless of intermediate progress
If any timeout triggers and a fallback function is provided, the fallback
is invoked instead of propagating the exception to the caller.
Args:
operation: Async callable to execute with timeout protection.
*args: Arguments to pass to the operation.
connect_timeout: Maximum seconds to wait for connection establishment.
read_timeout: Maximum seconds to wait after connection for response data.
total_timeout: Hard ceiling — operation cannot exceed this duration.
fallback_fn: Optional callable returning degraded data when timeout occurs.
fallback_source_name: Label for metrics/logging (e.g., "stale cache", "default").
operation_label: Human-readable name of the guarded operation.
Returns:
The result from either the primary call or the fallback function.
Raises:
CallTimeoutError: Only if no fallback is available.
Usage:
async def fetch_user(user_id: int) -> dict: ...
user = await enforce_timeout_with_fallback(
fetch_user, 42,
connect_timeout=1.0, read_timeout=5.0, total_timeout=8.0,
fallback_fn=lambda uid: get_cached_user(uid),
fallback_source_name="stale-cache",
operation_label="user-profile-api",
)
"""
# Wrap the operation with nested timeouts using asyncio.wait_for
try:
# Layer 1 & 2: Connect + Read timeout via wrapper
async def _with_connect_and_read_timeout() -> Any:
loop = asyncio.get_running_loop()
async def _coro() -> Any:
return await operation(*args)
# Use asyncio.wait_for to enforce the read deadline
# The connect happens implicitly when the coroutine first runs
return await asyncio.wait_for(_coro(), timeout=read_timeout)
# Layer 3: Total timeout wraps everything
result = await asyncio.wait_for(
_with_connect_and_read_timeout(),
timeout=total_timeout,
)
return result # type: ignore[return-value]
except (asyncio.TimeoutError, TimeoutError):
if fallback_fn is not None:
logger.info(
"Timeout on '%s' (%.1fs total) — using fallback from '%s'",
operation_label, total_timeout, fallback_source_name,
)
return await fallback_fn(*args) if asyncio.iscoroutinefunction(fallback_fn) else fallback_fn(*args) # type: ignore[misc]
raise CallTimeoutError(
f"Operation '{operation_label}' timed out after {total_timeout}s "
f"(connect={connect_timeout}s, read={read_timeout}s)"
)
async def timeout_guarded_http_call(
url: str,
method: str = "GET",
connect_timeout: float = 1.0,
read_timeout: float = 5.0,
total_timeout: float = 8.0,
fallback_response: dict[str, Any] | None = None,
) -> dict[str, Any]:
"""Execute an HTTP call with timeout enforcement and structured fallback.
This is a concrete example showing how timeouts and fallbacks integrate
in a real-world microservice calling another via HTTP.
"""
async def _do_call() -> dict[str, Any]:
# Pseudo-code for illustrative purposes — use httpx or aiohttp in production
import aiohttp
timeout = aiohttp.ClientTimeout(
total=total_timeout,
connect=connect_timeout,
)
async with aiohttp.ClientSession(timeout=timeout) as session:
async with session.request(method, url) as resp:
return {"status": resp.status, "body": await resp.json()}
try:
result = await asyncio.wait_for(_do_call(), timeout=total_timeout)
return result
except (asyncio.TimeoutError, TimeoutError):
if fallback_response is not None:
logger.warning("HTTP call to '%s' timed out — returning stale response", url)
return {**fallback_response, "_degraded": True}
raise
# --- Usage example ---
# cached_user = {"id": 42, "name": "cached-value", "updated_at": "2026-01-01T00:00:00Z"}
# user_data = await enforce_timeout_with_fallback(
# fetch_user_profile, user_id=42,
# connect_timeout=1.0, read_timeout=5.0, total_timeout=8.0,
# fallback_fn=lambda uid: cached_user,
# fallback_source_name="stale-cache",
# operation_label="user-profile-api",
# )
timeout=None or omit the timeout parameter. Default connect timeout is 2s, default read timeout is 10s.random.uniform(base * factor / 2, base * factor) to prevent thundering herd when a dependency recovers and hundreds of clients retry simultaneously.max_concurrent as (total_threads / num_dependencies) minus a 20% safety margin._degraded: True) so downstream consumers know the response quality.When applying this skill, produce output containing:
failure_threshold, timeout_seconds, max_attempts, base_delay, max_concurrent based on the specific architecture context| Skill | Purpose |
|---|---|
microservices-architecture | Broader microservice design principles that inform where to apply resilience patterns in your architecture |
idempotent-distributed-operations | Ensures operations are safely retriable — required prerequisite before adding retry logic with exponential backoff |
observability-patterns | Metrics, tracing, and logging infrastructure needed to monitor circuit breaker states, retry rates, and bulkhead utilization |
event-driven-patterns | Alternative resilience approach using event sourcing and outbox patterns for eventual consistency across services |
📖 skill(local cache): microservice-resilience-patterns