Distributed Systems
- 236 installs
- 213 repo stars
- Updated August 4, 2026
- yonatangross/orchestkit
Architect resilient microservices, queues, caching, and service boundaries when building scalable backends for SaaS products and multi-agent orchestration platforms.
About
Covers distributed systems architecture patterns for orchestkit backends including microservices decomposition, async messaging, caching layers, consistency tradeoffs, and resilience for production SaaS and agent platforms.
- Service boundary design
- Event-driven messaging patterns
- Caching and consistency models
- Fault tolerance strategies
- Cross-service observability
Distributed Systems by the numbers
- 236 all-time installs (skills.sh)
- +2 installs in the week ending Jul 28, 2026 (Skillselion tracking)
- Ranked #1,668 of 4,347 Backend & APIs skills by installs in the Skillselion catalog
- Data as of Aug 5, 2026 (Skillselion catalog sync)
npx skills add https://github.com/yonatangross/orchestkit --skill distributed-systemsAdd your badge
Show developers this skill is listed on Skillselion. Paste this into your README.
| Installs | 236 |
|---|---|
| repo stars | ★ 213 |
| Last updated | August 4, 2026 |
| Repository | yonatangross/orchestkit ↗ |
What it does
Architect resilient microservices, queues, caching, and service boundaries when building scalable backends for SaaS products and multi-agent orchestration platforms.
Files
Distributed Systems Patterns
Comprehensive patterns for building reliable distributed systems. Each category has individual rule files in rules/ loaded on-demand.
Quick Reference
| Category | Rules | Impact | When to Use |
|---|---|---|---|
| Distributed Locks | 3 | CRITICAL | Redis/Redlock locks, PostgreSQL advisory locks, fencing tokens |
| Resilience | 3 | CRITICAL | Circuit breakers, retry with backoff, bulkhead isolation |
| Idempotency | 3 | HIGH | Idempotency keys, request dedup, database-backed idempotency |
| Rate Limiting | 3 | HIGH | Token bucket, sliding window, distributed rate limits |
| Edge Computing | 2 | HIGH | Edge workers, V8 isolates, CDN caching, geo-routing |
| Event-Driven | 2 | HIGH | Event sourcing, CQRS, transactional outbox, sagas |
Total: 16 rules across 6 categories
Quick Start
# Redis distributed lock with Lua scripts
async with RedisLock(redis_client, "payment:order-123"):
await process_payment(order_id)
# Circuit breaker for external APIs
@circuit_breaker(failure_threshold=5, recovery_timeout=30)
@retry(max_attempts=3, base_delay=1.0)
async def call_external_api():
...
# Idempotent API endpoint
@router.post("/payments")
async def create_payment(
data: PaymentCreate,
idempotency_key: str = Header(..., alias="Idempotency-Key"),
):
return await idempotent_execute(db, idempotency_key, "/payments", process)
# Token bucket rate limiting
limiter = TokenBucketLimiter(redis_client, capacity=100, refill_rate=10)
if await limiter.is_allowed(f"user:{user_id}"):
await handle_request()Distributed Locks
Coordinate exclusive access to resources across multiple service instances.
| Rule | File | Key Pattern |
|---|---|---|
| Redis & Redlock | ${CLAUDE_SKILL_DIR}/rules/locks-redis-redlock.md | Lua scripts, SET NX, multi-node quorum |
| PostgreSQL Advisory | ${CLAUDE_SKILL_DIR}/rules/locks-postgres-advisory.md | Session/transaction locks, lock ID strategies |
| Fencing Tokens | ${CLAUDE_SKILL_DIR}/rules/locks-fencing-tokens.md | Owner validation, TTL, heartbeat extension |
Resilience
Production-grade fault tolerance for distributed systems.
| Rule | File | Key Pattern |
|---|---|---|
| Circuit Breaker | ${CLAUDE_SKILL_DIR}/rules/resilience-circuit-breaker.md | CLOSED/OPEN/HALF_OPEN states, sliding window |
| Retry & Backoff | ${CLAUDE_SKILL_DIR}/rules/resilience-retry-backoff.md | Exponential backoff, jitter, error classification |
| Bulkhead Isolation | ${CLAUDE_SKILL_DIR}/rules/resilience-bulkhead.md | Semaphore tiers, rejection policies, queue depth |
Idempotency
Ensure operations can be safely retried without unintended side effects.
| Rule | File | Key Pattern |
|---|---|---|
| Idempotency Keys | ${CLAUDE_SKILL_DIR}/rules/idempotency-keys.md | Deterministic hashing, Stripe-style headers |
| Request Dedup | ${CLAUDE_SKILL_DIR}/rules/idempotency-dedup.md | Event consumer dedup, Redis + DB dual layer |
| Database-Backed | ${CLAUDE_SKILL_DIR}/rules/idempotency-database.md | Unique constraints, upsert, TTL cleanup |
Rate Limiting
Protect APIs with distributed rate limiting using Redis.
| Rule | File | Key Pattern |
|---|---|---|
| Token Bucket | ${CLAUDE_SKILL_DIR}/rules/ratelimit-token-bucket.md | Redis Lua scripts, burst capacity, refill rate |
| Sliding Window | ${CLAUDE_SKILL_DIR}/rules/ratelimit-sliding-window.md | Sorted sets, precise counting, no boundary spikes |
| Distributed Limits | ${CLAUDE_SKILL_DIR}/rules/ratelimit-distributed.md | SlowAPI + Redis, tiered limits, response headers |
Edge Computing
Edge runtime patterns for Cloudflare Workers, Vercel Edge, and Deno Deploy.
| Rule | File | Key Pattern |
|---|---|---|
| Edge Workers | ${CLAUDE_SKILL_DIR}/rules/edge-workers.md | V8 isolate constraints, Web APIs, geo-routing, auth at edge |
| Edge Caching | ${CLAUDE_SKILL_DIR}/rules/edge-caching.md | Cache-aside at edge, CDN headers, KV storage, stale-while-revalidate |
Event-Driven
Event sourcing, CQRS, saga orchestration, and reliable messaging patterns.
| Rule | File | Key Pattern |
|---|---|---|
| Event Sourcing | ${CLAUDE_SKILL_DIR}/rules/event-sourcing.md | Event-sourced aggregates, CQRS read models, optimistic concurrency |
| Event Messaging | ${CLAUDE_SKILL_DIR}/rules/event-messaging.md | Transactional outbox, saga compensation, idempotent consumers |
Key Decisions
| Decision | Recommendation |
|---|---|
| Lock backend | Redis for speed, PostgreSQL if already using it, Redlock for HA |
| Lock TTL | 2-3x expected operation time |
| Circuit breaker recovery | Half-open probe with sliding window |
| Retry algorithm | Exponential backoff + full jitter |
| Bulkhead isolation | Semaphore-based tiers (Critical/Standard/Optional) |
| Idempotency storage | Redis (speed) + DB (durability), 24-72h TTL |
| Rate limit algorithm | Token bucket for most APIs, sliding window for strict quotas |
| Rate limit storage | Redis (distributed, atomic Lua scripts) |
When NOT to Use
No separate event-sourcing/saga/CQRS skills exist — they are rules within distributed-systems. But most projects never need them.
| Pattern | Interview | Hackathon | MVP | Growth | Enterprise | Simpler Alternative |
|---|---|---|---|---|---|---|
| Event sourcing | OVERKILL | OVERKILL | OVERKILL | OVERKILL | WHEN JUSTIFIED | Append-only table with status column |
| Saga orchestration | OVERKILL | OVERKILL | OVERKILL | SELECTIVE | APPROPRIATE | Sequential service calls with manual rollback |
| Circuit breaker | OVERKILL | OVERKILL | BORDERLINE | APPROPRIATE | REQUIRED | Try/except with timeout |
| Distributed locks | OVERKILL | OVERKILL | BORDERLINE | APPROPRIATE | REQUIRED | Database row-level lock (SELECT FOR UPDATE) |
| CQRS | OVERKILL | OVERKILL | OVERKILL | OVERKILL | WHEN JUSTIFIED | Single model for read/write |
| Transactional outbox | OVERKILL | OVERKILL | OVERKILL | SELECTIVE | APPROPRIATE | Direct publish after commit |
| Rate limiting | OVERKILL | OVERKILL | SIMPLE ONLY | APPROPRIATE | REQUIRED | Nginx rate limit or cloud WAF |
Rule of thumb: If you have a single server process, you do not need distributed systems patterns. Use in-process alternatives. Add distribution only when you actually have multiple instances.
Anti-Patterns (FORBIDDEN)
# LOCKS: Never forget TTL (causes deadlocks)
await redis.set(f"lock:{name}", "1") # WRONG - no expiry!
# LOCKS: Never release without owner check
await redis.delete(f"lock:{name}") # WRONG - might release others' lock
# RESILIENCE: Never retry non-retryable errors
@retry(max_attempts=5, retryable_exceptions={Exception}) # Retries 401!
# RESILIENCE: Never put retry outside circuit breaker
@retry # Would retry when circuit is open!
@circuit_breaker
async def call(): ...
# IDEMPOTENCY: Never use non-deterministic keys
key = str(uuid.uuid4()) # Different every time!
# IDEMPOTENCY: Never cache error responses
if response.status_code >= 400:
await cache_response(key, response) # Errors should retry!
# RATE LIMITING: Never use in-memory counters in distributed systems
request_counts = {} # Lost on restart, not shared across instancesDetailed Documentation
| Resource | Description |
|---|---|
${CLAUDE_SKILL_DIR}/scripts/ | Templates: lock implementations, circuit breaker, rate limiter |
${CLAUDE_SKILL_DIR}/checklists/ | Pre-flight checklists for each pattern category |
${CLAUDE_SKILL_DIR}/references/ | Deep dives: Redlock algorithm, bulkhead tiers, token bucket |
${CLAUDE_SKILL_DIR}/examples/ | Complete integration examples |
Related Skills
caching- Redis caching patterns, cache as fallbackbackground-jobs- Job deduplication, async processing with retryobservability-monitoring- Metrics and alerting for circuit breaker state changeserror-handling-rfc9457- Structured error responses for resilience failuresauth-patterns- API key management, authentication integration
Circuit Breaker Setup Guide
Step-by-step guide for adding circuit breakers to a service.
Step 1: Identify Services
List all external dependencies that need circuit breakers:
┌─────────────────────────────────────────────────────────────┐
│ Service Inventory │
├─────────────────────────────────────────────────────────────┤
│ │
│ External APIs: │
│ □ OpenAI API (LLM) │
│ □ Anthropic API (LLM) │
│ □ YouTube Data API │
│ □ GitHub API │
│ □ arXiv API │
│ │
│ Internal Services: │
│ □ Embedding service │
│ □ Database (PostgreSQL) │
│ □ Redis cache │
│ □ Semantic search │
│ │
│ For each, answer: │
│ 1. What's the expected failure rate? │
│ 2. How long does recovery typically take? │
│ 3. What's the fallback behavior? │
│ │
└─────────────────────────────────────────────────────────────┘Step 2: Configure Thresholds
Failure Threshold
| Service Type | Recommended | Reasoning |
|---|---|---|
| LLM API | 3 | APIs can be unstable, fail fast |
| External API | 5 | More tolerant of transient issues |
| Database | 2-3 | DB issues usually need immediate attention |
| Internal service | 3-5 | Depends on service criticality |
Recovery Timeout
| Service Type | Recommended | Reasoning |
|---|---|---|
| LLM API | 60s | Rate limits typically reset in minutes |
| External API | 30-120s | Depends on SLA |
| Database | 15-30s | Should recover quickly |
| Internal service | 15-60s | Depends on restart time |
Slow Call Threshold
| Service Type | Recommended | Reasoning |
|---|---|---|
| LLM API | 30s | LLM calls can be slow |
| External API | 10s | Most APIs should be fast |
| Database | 5s | DB queries should be optimized |
| Internal service | 5-10s | Depends on operation |
Step 3: Implement Circuit Breaker
Basic Implementation
from resilience import CircuitBreaker, CircuitBreakerFactory
# Option 1: Use factory for common patterns
openai_breaker = CircuitBreakerFactory.for_llm_api("openai")
db_breaker = CircuitBreakerFactory.for_database("postgres")
# Option 2: Custom configuration
custom_breaker = CircuitBreaker(
name="my-service",
failure_threshold=5,
success_threshold=2,
recovery_timeout=30.0,
slow_call_threshold=10.0,
)Wrap Service Calls
# Method 1: Decorator
@openai_breaker
async def call_openai(prompt: str) -> str:
return await openai_client.complete(prompt)
# Method 2: Explicit call
async def call_openai(prompt: str) -> str:
return await openai_breaker.call(
openai_client.complete,
prompt,
)Step 4: Add Fallback Handling
from resilience import CircuitOpenError
async def analyze_with_fallback(content: str) -> Analysis:
try:
return await circuit_breaker.call(primary_analysis, content)
except CircuitOpenError as e:
logger.warning(
f"Circuit open for {e.name}, using fallback",
time_until_recovery=e.time_until_recovery,
)
# Fallback 1: Try cache
cached = await cache.get(f"analysis:{hash(content)}")
if cached:
return Analysis.from_cache(cached, is_stale=True)
# Fallback 2: Degraded response
return Analysis(
status="degraded",
message="Full analysis temporarily unavailable",
basic_info=extract_basic_info(content),
)Step 5: Add Observability
Logging
def setup_circuit_logging(breaker: CircuitBreaker):
def on_state_change(old: str, new: str, name: str):
logger.warning(
"circuit_state_change",
circuit=name,
old_state=old,
new_state=new,
)
if new == "open":
# Send alert
alerting.send(
severity="warning",
message=f"Circuit {name} opened",
runbook="https://docs/runbooks/circuit-breaker",
)
breaker._on_state_change = on_state_changeMetrics
from prometheus_client import Gauge, Counter
circuit_state = Gauge(
"circuit_breaker_state",
"Circuit breaker state (0=closed, 1=open, 2=half_open)",
["service"],
)
circuit_rejections = Counter(
"circuit_breaker_rejections_total",
"Total requests rejected by circuit breaker",
["service"],
)
def update_metrics(breaker: CircuitBreaker):
state_map = {"closed": 0, "open": 1, "half_open": 2}
circuit_state.labels(service=breaker.name).set(
state_map[breaker.state.value]
)Health Endpoint
@app.get("/health/circuits")
async def circuit_health():
return {
name: {
"state": cb.state.value,
"failure_count": cb._failure_count,
"time_until_recovery": (
cb._time_until_recovery()
if cb.state == CircuitState.OPEN
else None
),
}
for name, cb in circuit_breakers.items()
}Step 6: Test Circuit Behavior
Unit Tests
@pytest.mark.asyncio
async def test_circuit_opens_on_failures():
breaker = CircuitBreaker(name="test", failure_threshold=3)
async def failing_call():
raise ConnectionError("Failed")
# Fail 3 times
for _ in range(3):
with pytest.raises(ConnectionError):
await breaker.call(failing_call)
# Should be open now
assert breaker.state == CircuitState.OPEN
# Next call rejected
with pytest.raises(CircuitOpenError):
await breaker.call(failing_call)
@pytest.mark.asyncio
async def test_circuit_recovers():
breaker = CircuitBreaker(
name="test",
failure_threshold=1,
recovery_timeout=0.1, # Fast for testing
)
async def failing_then_succeeding():
if breaker._failure_count > 0:
return "success"
raise ConnectionError("First call fails")
# Open circuit
with pytest.raises(ConnectionError):
await breaker.call(failing_then_succeeding)
# Wait for recovery
await asyncio.sleep(0.2)
# Should succeed now
result = await breaker.call(failing_then_succeeding)
assert result == "success"
assert breaker.state == CircuitState.CLOSEDIntegration Tests
@pytest.mark.asyncio
async def test_circuit_isolates_failures():
"""Verify circuit prevents cascade failures."""
openai_breaker = circuit_breakers["openai"]
# Simulate OpenAI outage
with patch("openai.complete", side_effect=ConnectionError):
# Multiple calls should fail then trip circuit
for _ in range(5):
try:
await analyze_content("test")
except (ConnectionError, CircuitOpenError):
pass
# Circuit should be open
assert openai_breaker.state == CircuitState.OPEN
# Anthropic should still work (different circuit)
anthropic_breaker = circuit_breakers["anthropic"]
assert anthropic_breaker.state == CircuitState.CLOSEDStep 7: Document and Monitor
Documentation
Add to your service's README:
## Circuit Breakers
| Service | Threshold | Recovery | Fallback |
|---------|-----------|----------|----------|
| openai | 3 failures | 60s | Use gpt-5-mini |
| anthropic | 3 failures | 60s | Use cache |
| youtube | 5 failures | 120s | Return partial data |
### Monitoring
- Dashboard: [Grafana Circuit Breakers](...)
- Alerts: PagerDuty channel #resilience
- Runbook: [Circuit Breaker Runbook](...)Runbook Template
## Circuit Breaker Open - {service}
### Symptoms
- Service returning 503 errors
- Alert: "Circuit {service} opened"
- Dashboard shows circuit in OPEN state
### Impact
- {describe impact on users}
### Resolution
1. Check {service} status page
2. Review logs for failure pattern
3. If transient: wait for auto-recovery
4. If persistent: {escalation steps}
### Verification
1. Circuit state returns to CLOSED
2. Service calls succeeding
3. Metrics returning to normalQuick Reference
┌─────────────────────────────────────────────────────────────┐
│ Circuit Breaker Quick Reference │
├─────────────────────────────────────────────────────────────┤
│ │
│ CREATE: │
│ breaker = CircuitBreaker("name", failure_threshold=5) │
│ │
│ USE: │
│ @breaker │
│ async def my_function(): ... │
│ │
│ result = await breaker.call(func, *args) │
│ │
│ HANDLE: │
│ try: │
│ await breaker.call(...) │
│ except CircuitOpenError: │
│ return fallback_response() │
│ │
│ MONITOR: │
│ breaker.get_status() │
│ breaker.state == CircuitState.OPEN │
│ │
│ RESET (manual): │
│ breaker.reset() │
│ │
└─────────────────────────────────────────────────────────────┘Distributed Locks Checklist
Lock Selection
- [ ] Chose appropriate lock backend
- Redis: Fast, TTL-based, requires Redis infrastructure
- PostgreSQL: No extra infra, integrates with transactions
- Redlock: Multi-node Redis for high availability
- [ ] Determined lock scope (session vs transaction)
- [ ] Set appropriate TTL (not too short, not too long)
Implementation
Acquire
- [ ] Non-blocking option available (
try_lock) - [ ] Timeout support for blocking acquire
- [ ] Retry logic with exponential backoff
- [ ] Jitter added to prevent thundering herd
- [ ] Unique owner ID generated (UUIDv7)
Release
- [ ] Owner validation (only owner can release)
- [ ] Atomic release (Lua script for Redis)
- [ ] Idempotent release (safe to call twice)
- [ ] Finally block ensures release on exception
Extension
- [ ] Heartbeat/extend for long operations
- [ ] Auto-extend background task option
- [ ] Extension validates ownership
Safety
Mutual Exclusion
- [ ] Atomic acquire (SET NX for Redis)
- [ ] Fencing token or owner ID validated
- [ ] No race conditions in acquire/release
Deadlock Prevention
- [ ] TTL prevents permanent deadlocks
- [ ] Lock ordering for multiple locks
- [ ] Timeout on acquire attempts
Split-Brain Protection
- [ ] Redlock for multi-node Redis
- [ ] Clock drift factored into validity
- [ ] Quorum required for lock acquisition
Error Handling
- [ ] Lock acquisition failures handled gracefully
- [ ] Release failures logged and handled
- [ ] Network partition scenarios considered
- [ ] Retry logic for transient failures
Testing
- [ ] Unit tests for lock logic
- [ ] Integration tests with real backend
- [ ] Concurrent access tests
- [ ] Failure scenario tests (network, timeout)
- [ ] Lock expiration tests
Monitoring
- [ ] Lock acquisition metrics
- [ ] Lock hold duration metrics
- [ ] Failed acquisition alerts
- [ ] Long-held lock alerts
- [ ] Deadlock detection
PostgreSQL Advisory Locks
- [ ] Correct lock function used (session vs xact)
- [ ] Lock ID strategy documented
- [ ] Namespace collisions prevented
- [ ]
pg_locksmonitoring query available
Redis Locks
- [ ] Lua scripts used for atomicity
- [ ] TTL always set (no deadlocks)
- [ ] Owner ID stored with lock
- [ ] Release validates owner
Redlock (Multi-Node)
- [ ] Minimum 3 Redis instances (recommend 5)
- [ ] Quorum calculated correctly (N/2 + 1)
- [ ] Clock drift factored in
- [ ] Failed nodes don't block acquire
- [ ] Release attempted on all nodes
Production Readiness
- [ ] Lock names are descriptive and namespaced
- [ ] TTL tuned for operation duration
- [ ] Metrics and alerting configured
- [ ] Runbook for lock-related incidents
- [ ] Graceful degradation strategy
Idempotency Implementation Checklist
Key Generation
- [ ] Keys are deterministic (same input = same key)
- [ ] Keys include all relevant parameters
- [ ] Keys are scoped appropriately (user, endpoint, etc.)
- [ ] Keys use consistent hash algorithm (SHA-256)
- [ ] Keys are reasonable length (32-64 chars)
API Endpoints
- [ ] POST/PUT/PATCH endpoints support Idempotency-Key header
- [ ] Idempotency key format is documented
- [ ] Key is validated (format, length)
- [ ] Duplicate requests return cached response
- [ ] Response includes header indicating replay
Storage
- [ ] Redis used for fast lookups
- [ ] Database used for durability
- [ ] TTL configured appropriately (24-72 hours)
- [ ] Cleanup job for expired records
- [ ] Storage sized for expected volume
Race Conditions
- [ ] Database constraint prevents duplicates
- [ ] Check-and-insert is atomic
- [ ] Lost updates are prevented
- [ ] Concurrent requests handled correctly
Response Handling
- [ ] Only successful responses cached (2xx)
- [ ] Error responses allow retry
- [ ] Response body stored completely
- [ ] Status code preserved
- [ ] Headers preserved if needed
Event Processing
- [ ] Events include idempotency key
- [ ] Consumer checks before processing
- [ ] Processed events tracked
- [ ] At-least-once delivery handled
- [ ] Dead letter queue for failures
Error Cases
- [ ] Missing key handled (process normally or reject)
- [ ] Invalid key format rejected
- [ ] Storage failures don't break processing
- [ ] Timeout during processing handled
Testing
- [ ] Duplicate request returns same response
- [ ] Different keys process independently
- [ ] Race condition tests pass
- [ ] TTL expiration verified
- [ ] Cache miss falls back to database
Documentation
- [ ] Idempotency behavior documented
- [ ] Key format documented
- [ ] TTL window documented
- [ ] Client retry guidance provided
Pre-Deployment Resilience Checklist
Use this checklist before deploying services with resilience patterns.
Circuit Breakers
- [ ] Threshold Configuration
- [ ] Failure threshold set appropriately (not too low, not too high)
- [ ] Recovery timeout allows service to actually recover
- [ ] Sliding window size captures representative sample
- [ ] Fallback Behavior
- [ ] Every circuit breaker has a defined fallback response
- [ ] Fallbacks return meaningful partial data when possible
- [ ] Fallbacks don't call other services with closed circuits
- [ ] Observability
- [ ] State changes logged with structured logging
- [ ] Metrics exported (Prometheus/Langfuse)
- [ ] Alerts configured for OPEN state
- [ ] Dashboard shows circuit status
- [ ] Testing
- [ ] Unit tests for state transitions
- [ ] Integration tests simulate failure scenarios
- [ ] Chaos testing validates circuit behavior under load
Bulkheads
- [ ] Tier Assignment
- [ ] Critical operations in Tier 1 (highest priority)
- [ ] Standard operations in Tier 2
- [ ] Optional/background operations in Tier 3
- [ ] No critical path through Tier 3
- [ ] Capacity Planning
- [ ] Max concurrent based on downstream capacity
- [ ] Queue sizes prevent memory exhaustion
- [ ] Timeouts shorter than caller's timeout
- [ ] Rejection Handling
- [ ] Rejection policy defined per tier
- [ ] HTTP 503 returned with Retry-After header
- [ ] Rejections logged and metriced
- [ ] Testing
- [ ] Load test validates bulkhead isolation
- [ ] Tier 3 failure doesn't affect Tier 1
- [ ] Queue depth monitored under load
Retry Logic
- [ ] Error Classification
- [ ] Retryable vs non-retryable errors defined
- [ ] HTTP status codes classified correctly
- [ ] LLM API errors handled specifically
- [ ] Backoff Strategy
- [ ] Exponential backoff configured
- [ ] Jitter enabled to prevent thundering herd
- [ ] Max delay caps retry storms
- [ ] Limits
- [ ] Max attempts bounded (typically 3-5)
- [ ] Total retry time < caller's timeout
- [ ] Retry budget prevents system overload
- [ ] Testing
- [ ] Transient failures recovered automatically
- [ ] Non-retryable errors fail immediately
- [ ] Retry budget depletes under sustained failures
LLM Resilience
- [ ] Fallback Chain
- [ ] Primary model defined
- [ ] At least one fallback model configured
- [ ] Semantic cache as final fallback
- [ ] Default response for complete outage
- [ ] Token Budget
- [ ] Budget allocation per category
- [ ] Truncation strategy defined
- [ ] Output reserve prevents overflow
- [ ] Rate Limiting
- [ ] Client-side rate limiter configured
- [ ] Respects API provider limits
- [ ] Graceful handling of 429 responses
- [ ] Cost Control
- [ ] Per-request cost tracking
- [ ] Hourly/daily budget alerts
- [ ] Cost circuit breaker configured
Integration
- [ ] Pattern Composition
- [ ] Retry INSIDE circuit breaker
- [ ] Bulkhead wraps retry+circuit
- [ ] Timeout inside all patterns
- [ ] Health Endpoints
- [ ] /health returns 200 (doesn't check circuit)
- [ ] /ready reflects degraded state
- [ ] /resilience shows all pattern status
- [ ] Configuration
- [ ] All thresholds configurable via env vars
- [ ] Defaults documented
- [ ] Per-environment overrides
Observability
- [ ] Logging
- [ ] Structured logging with trace IDs
- [ ] State changes logged at WARN level
- [ ] Rejections logged with context
- [ ] Metrics
- [ ] Circuit state gauge
- [ ] Bulkhead utilization gauge
- [ ] Retry counter
- [ ] Latency histograms
- [ ] Alerting
- [ ] Alert when circuit opens
- [ ] Alert when bulkhead consistently full
- [ ] Alert when retry budget exhausted
- [ ] Runbook links in alerts
Documentation
- [ ] Architecture
- [ ] Resilience patterns documented in ADR
- [ ] Diagram shows pattern composition
- [ ] Tier assignments documented
- [ ] Operations
- [ ] Runbook for circuit open scenarios
- [ ] Runbook for bulkhead exhaustion
- [ ] Manual override procedures documented
- [ ] API Documentation
- [ ] 503 responses documented
- [ ] Retry-After header usage documented
- [ ] Degraded response format documented
Final Verification
- [ ] Load Test
- [ ] System handles expected load
- [ ] Graceful degradation under 2x load
- [ ] Recovery after load spike
- [ ] Chaos Test
- [ ] Dependency failure isolated
- [ ] Recovery automatic when dependency restored
- [ ] No cascading failures
- [ ] Security Review
- [ ] Fallback responses don't leak sensitive data
- [ ] Error messages don't expose internals
- [ ] Rate limits prevent abuse
Rate Limiting Implementation Checklist
Planning
- [ ] Define rate limits for each endpoint category
- [ ] Read endpoints (GET) - higher limits
- [ ] Write endpoints (POST/PUT/DELETE) - lower limits
- [ ] Authentication endpoints - very strict limits
- [ ] Expensive operations (LLM calls, file processing) - strictest limits
- [ ] Choose limiting algorithm
- [ ] Token Bucket - for bursty traffic patterns
- [ ] Sliding Window - for strict quotas
- [ ] Fixed Window - for simple requirements
- [ ] Determine key strategy
- [ ] By IP address (anonymous users)
- [ ] By user ID (authenticated users)
- [ ] By API key (service accounts)
- [ ] By organization (enterprise customers)
Implementation
Backend Setup
- [ ] Install dependencies
pip install slowapi redis- [ ] Configure Redis connection
redis_client = Redis.from_url(settings.redis_url)- [ ] Set up SlowAPI or custom limiter
limiter = Limiter(key_func=get_user_identifier)
app.add_middleware(SlowAPIMiddleware)Route Protection
- [ ] Add
@limiter.limit()to all public endpoints - [ ] Set stricter limits for:
- [ ] Login/register endpoints (prevent brute force)
- [ ] Password reset (prevent enumeration)
- [ ] File upload (prevent abuse)
- [ ] LLM/AI operations (cost control)
Response Headers
- [ ] Include rate limit headers in all responses:
- [ ]
X-RateLimit-Limit- max requests in window - [ ]
X-RateLimit-Remaining- requests remaining - [ ]
X-RateLimit-Reset- Unix timestamp when limit resets
- [ ] Include
Retry-Afterheader in 429 responses
Error Handling
- [ ] Return proper 429 Too Many Requests status
- [ ] Include helpful error message
{
"type": "https://api.example.com/problems/rate-limit-exceeded",
"title": "Rate Limit Exceeded",
"status": 429,
"detail": "You have exceeded 100 requests per minute. Please wait 45 seconds.",
"retry_after": 45
}Tiered Limits
- [ ] Define limits per user tier:
| Tier | Requests/min | Burst |
|---|---|---|
| Anonymous | 10 | 5 |
| Free | 100 | 20 |
| Pro | 1000 | 100 |
| Enterprise | 10000 | 1000 |
- [ ] Implement dynamic limit function
def get_tier_limit(request: Request) -> str:
user = request.state.user
return TIER_LIMITS.get(user.tier, "10/minute")Distributed Systems
- [ ] Use Redis backend (not in-memory)
- [ ] Configure Redis connection pooling
- [ ] Set appropriate key TTLs
- [ ] Use Lua scripts for atomicity
- [ ] Handle Redis connection failures gracefully
Monitoring
- [ ] Log rate limit hits
logger.warning("Rate limit exceeded", extra={
"user_id": user.id,
"endpoint": request.url.path,
"limit": limit,
})- [ ] Track metrics:
- [ ] Rate limit hits per endpoint
- [ ] Rate limit hits per user
- [ ] Average remaining quota
- [ ] Set up alerts for:
- [ ] Unusual spike in 429 responses
- [ ] Single user hitting limits repeatedly
- [ ] Redis connection failures
Security Considerations
- [ ] Rate limit login endpoints strictly (prevent brute force)
- [ ] Rate limit password reset (prevent enumeration)
- [ ] Consider IP reputation for anonymous limits
- [ ] Don't expose internal rate limit keys
- [ ] Use secure Redis connection (TLS)
Documentation
- [ ] Document rate limits in OpenAPI/Swagger
- [ ] Add rate limit info to API documentation
- [ ] Include examples of handling 429 responses
- [ ] Explain tier limits for customers
Testing
- [ ] Unit test rate limit logic
- [ ] Integration test with Redis
- [ ] Load test to verify limits work
- [ ] Test retry logic in clients
- [ ] Test header values are correct
- [ ] Test limit reset behavior
Client SDK Recommendations
Document recommended client-side handling:
# Python client example
import time
import httpx
def make_request_with_retry(url: str, max_retries: int = 3):
for attempt in range(max_retries):
response = httpx.get(url)
if response.status_code == 429:
retry_after = int(response.headers.get("Retry-After", 60))
time.sleep(retry_after)
continue
return response
raise Exception("Rate limit exceeded after retries")Rollout Checklist
- [ ] Deploy with monitoring enabled
- [ ] Start with permissive limits
- [ ] Monitor for false positives
- [ ] Gradually tighten limits
- [ ] Communicate changes to users
- [ ] Provide upgrade path for users hitting limits
FastAPI Rate Limiting Examples
Complete examples for implementing rate limiting in FastAPI with Redis.
SlowAPI Setup (Recommended for Simple Cases)
Installation
pip install slowapi redisBasic Configuration
# app/core/rate_limit.py
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded
from slowapi.middleware import SlowAPIMiddleware
from redis import Redis
# Use Redis backend for distributed rate limiting
redis_client = Redis.from_url("redis://localhost:6379", decode_responses=True)
limiter = Limiter(
key_func=get_remote_address,
storage_uri="redis://localhost:6379",
default_limits=["100/minute"],
)
def setup_rate_limiting(app):
"""Configure rate limiting for the FastAPI app."""
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
app.add_middleware(SlowAPIMiddleware)Route-Level Limiting
# app/api/v1/routes/analyses.py
from fastapi import APIRouter, Request, Depends
from slowapi import Limiter
from slowapi.util import get_remote_address
router = APIRouter()
limiter = Limiter(key_func=get_remote_address)
@router.post("/analyses")
@limiter.limit("10/minute") # Override default
async def create_analysis(request: Request):
"""Create analysis - stricter limit due to resource cost."""
return {"message": "Analysis created"}
@router.get("/analyses")
@limiter.limit("100/minute")
async def list_analyses(request: Request):
"""List analyses - more permissive."""
return {"analyses": []}
@router.get("/analyses/{id}")
@limiter.limit("200/minute")
async def get_analysis(request: Request, id: str):
"""Get single analysis - most permissive."""
return {"id": id}User-Based Rate Limiting
# app/core/rate_limit.py
from fastapi import Request
from app.api.deps import get_current_user
def get_user_identifier(request: Request) -> str:
"""Get rate limit key from authenticated user or IP."""
# Try to get user from request state (set by auth middleware)
user = getattr(request.state, "user", None)
if user:
return f"user:{user.id}"
# Fallback to IP for unauthenticated requests
return f"ip:{get_remote_address(request)}"
limiter = Limiter(key_func=get_user_identifier)Tiered Rate Limits
# app/api/v1/routes/protected.py
from fastapi import APIRouter, Request, Depends
from slowapi import Limiter
router = APIRouter()
def get_tier_limit(request: Request) -> str:
"""Dynamic limit based on user tier."""
user = getattr(request.state, "user", None)
if not user:
return "10/minute" # Anonymous
tier_limits = {
"free": "100/minute",
"pro": "1000/minute",
"enterprise": "10000/minute",
}
return tier_limits.get(user.tier, "100/minute")
@router.post("/generate")
@limiter.limit(get_tier_limit)
async def generate_content(request: Request):
"""Rate limit based on user subscription tier."""
return {"content": "Generated"}Custom Redis Token Bucket
For more control, implement custom rate limiting:
# app/core/rate_limit.py
import time
from typing import NamedTuple
import redis.asyncio as redis
from fastapi import Request, HTTPException, status
class RateLimitResult(NamedTuple):
allowed: bool
remaining: int
reset_at: float
retry_after: int
class RedisRateLimiter:
"""Custom rate limiter with token bucket algorithm."""
SCRIPT = """
local key = KEYS[1]
local capacity = tonumber(ARGV[1])
local refill_rate = tonumber(ARGV[2])
local now = tonumber(ARGV[3])
local bucket = redis.call('HMGET', key, 'tokens', 'last_refill')
local tokens = tonumber(bucket[1]) or capacity
local last_refill = tonumber(bucket[2]) or now
local elapsed = (now - last_refill) / 1000
tokens = math.min(capacity, tokens + elapsed * refill_rate)
if tokens >= 1 then
tokens = tokens - 1
redis.call('HMSET', key, 'tokens', tokens, 'last_refill', now)
redis.call('EXPIRE', key, math.ceil(capacity / refill_rate) + 1)
return {1, math.floor(tokens), 0}
else
local retry_after = math.ceil((1 - tokens) / refill_rate)
redis.call('HMSET', key, 'tokens', tokens, 'last_refill', now)
return {0, 0, retry_after}
end
"""
def __init__(
self,
redis_url: str = "redis://localhost:6379",
capacity: int = 100,
refill_rate: float = 10,
):
self.redis = redis.from_url(redis_url)
self.capacity = capacity
self.refill_rate = refill_rate
self._script = None
async def _get_script(self):
if self._script is None:
self._script = self.redis.register_script(self.SCRIPT)
return self._script
async def check(self, key: str) -> RateLimitResult:
"""Check rate limit for a key."""
script = await self._get_script()
now_ms = int(time.time() * 1000)
result = await script(
keys=[f"ratelimit:{key}"],
args=[self.capacity, self.refill_rate, now_ms],
)
reset_at = time.time() + (self.capacity / self.refill_rate)
return RateLimitResult(
allowed=bool(result[0]),
remaining=int(result[1]),
reset_at=reset_at,
retry_after=int(result[2]),
)
# FastAPI Dependency
async def rate_limit_dependency(
request: Request,
limiter: RedisRateLimiter = Depends(get_rate_limiter),
):
"""Dependency that enforces rate limiting."""
# Get identifier
user = getattr(request.state, "user", None)
key = f"user:{user.id}" if user else f"ip:{request.client.host}"
result = await limiter.check(key)
# Set rate limit headers
request.state.rate_limit = result
if not result.allowed:
raise HTTPException(
status_code=status.HTTP_429_TOO_MANY_REQUESTS,
detail="Rate limit exceeded",
headers={
"Retry-After": str(result.retry_after),
"X-RateLimit-Limit": str(limiter.capacity),
"X-RateLimit-Remaining": "0",
"X-RateLimit-Reset": str(int(result.reset_at)),
},
)
# Middleware to add rate limit headers to all responses
@app.middleware("http")
async def add_rate_limit_headers(request: Request, call_next):
response = await call_next(request)
rate_limit = getattr(request.state, "rate_limit", None)
if rate_limit:
response.headers["X-RateLimit-Limit"] = str(100)
response.headers["X-RateLimit-Remaining"] = str(rate_limit.remaining)
response.headers["X-RateLimit-Reset"] = str(int(rate_limit.reset_at))
return responseUsage in Routes
@router.post("/expensive-operation")
async def expensive_operation(
request: Request,
_: None = Depends(rate_limit_dependency),
):
"""This endpoint is rate limited."""
return {"result": "success"}Rate Limit by Endpoint Cost
# app/core/rate_limit.py
from functools import wraps
from typing import Callable
class CostBasedLimiter:
"""Rate limiter where different operations cost different tokens."""
def __init__(self, redis_url: str, capacity: int = 1000):
self.limiter = RedisRateLimiter(redis_url, capacity=capacity)
def limit(self, cost: int = 1):
"""Decorator that consumes 'cost' tokens per request."""
def decorator(func: Callable):
@wraps(func)
async def wrapper(request: Request, *args, **kwargs):
key = get_user_identifier(request)
# Check if we have enough tokens
for _ in range(cost):
result = await self.limiter.check(key)
if not result.allowed:
raise HTTPException(
status_code=429,
detail=f"Rate limit exceeded (operation costs {cost} tokens)",
)
return await func(request, *args, **kwargs)
return wrapper
return decorator
cost_limiter = CostBasedLimiter("redis://localhost:6379")
@router.get("/simple")
@cost_limiter.limit(cost=1) # Cheap operation
async def simple_query(request: Request):
return {"data": "simple"}
@router.post("/generate")
@cost_limiter.limit(cost=10) # Expensive operation
async def generate_content(request: Request):
return {"data": "generated"}
@router.post("/bulk-process")
@cost_limiter.limit(cost=50) # Very expensive
async def bulk_process(request: Request):
return {"data": "processed"}Testing Rate Limits
# tests/test_rate_limiting.py
import pytest
from httpx import ASGITransport, AsyncClient
from app.main import app
@pytest.mark.asyncio
async def test_rate_limit_enforced():
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
# Make requests up to limit
for _ in range(10):
response = await client.post("/analyses")
assert response.status_code == 200
# Next request should be rate limited
response = await client.post("/analyses")
assert response.status_code == 429
assert "Retry-After" in response.headers
@pytest.mark.asyncio
async def test_rate_limit_headers():
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
response = await client.get("/analyses")
assert "X-RateLimit-Limit" in response.headers
assert "X-RateLimit-Remaining" in response.headers
assert "X-RateLimit-Reset" in response.headersRelated Files
- See
references/token-bucket-algorithm.mdfor algorithm details - See
checklists/rate-limiting-checklist.mdfor implementation checklist - See SKILL.md for sliding window and fixed window algorithms
Idempotency Implementation Examples
FastAPI Idempotency Middleware
from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse
from typing import Callable
import redis.asyncio as redis
import json
import hashlib
app = FastAPI()
redis_client = redis.from_url("redis://localhost:6379")
IDEMPOTENCY_TTL = 86400 # 24 hours
class IdempotencyMiddleware:
"""Stripe-style idempotency middleware."""
def __init__(self, app: FastAPI):
self.app = app
async def __call__(self, scope, receive, send):
if scope["type"] != "http":
await self.app(scope, receive, send)
return
request = Request(scope, receive)
# Only apply to mutating methods
if request.method not in ("POST", "PUT", "PATCH"):
await self.app(scope, receive, send)
return
# Get idempotency key
idempotency_key = request.headers.get("Idempotency-Key")
if not idempotency_key:
await self.app(scope, receive, send)
return
# Check for cached response
cache_key = f"idem:{request.url.path}:{idempotency_key}"
cached = await redis_client.get(cache_key)
if cached:
cached_response = json.loads(cached)
response = JSONResponse(
content=cached_response["body"],
status_code=cached_response["status"],
headers={"Idempotent-Replayed": "true"},
)
await response(scope, receive, send)
return
# Try to acquire lock
lock_key = f"idem_lock:{request.url.path}:{idempotency_key}"
acquired = await redis_client.set(lock_key, "1", nx=True, ex=60)
if not acquired:
response = JSONResponse(
content={"error": "Request with this idempotency key is being processed"},
status_code=409,
)
await response(scope, receive, send)
return
try:
# Process request and capture response
# (Simplified - real implementation needs response capture)
await self.app(scope, receive, send)
finally:
await redis_client.delete(lock_key)
app.add_middleware(IdempotencyMiddleware)Database-Backed Idempotency
from sqlalchemy import Column, String, DateTime, Text, Index
from sqlalchemy.ext.asyncio import AsyncSession
from datetime import UTC, datetime, timedelta
import json
class IdempotencyRecord(Base):
"""Track processed idempotency keys."""
__tablename__ = "idempotency_records"
idempotency_key = Column(String(64), primary_key=True)
endpoint = Column(String(256), nullable=False)
request_hash = Column(String(64), nullable=False)
response_body = Column(Text, nullable=True)
response_status = Column(Integer, default=200)
created_at = Column(DateTime, default=lambda: datetime.now(UTC))
expires_at = Column(DateTime, nullable=False)
__table_args__ = (
Index("ix_idempotency_expires", "expires_at"),
Index("ix_idempotency_endpoint_key", "endpoint", "idempotency_key"),
)
async def get_or_create_idempotency(
db: AsyncSession,
idempotency_key: str,
endpoint: str,
request_body: dict,
process_func: Callable,
) -> tuple[dict, int, bool]:
"""
Get cached response or process request idempotently.
Returns:
(response_body, status_code, was_replayed)
"""
# Hash the request to detect mismatched bodies
request_hash = hashlib.sha256(
json.dumps(request_body, sort_keys=True).encode()
).hexdigest()
# Check for existing record
result = await db.execute(
select(IdempotencyRecord).where(
IdempotencyRecord.idempotency_key == idempotency_key,
IdempotencyRecord.endpoint == endpoint,
)
)
existing = result.scalar_one_or_none()
if existing:
# Verify request body matches
if existing.request_hash != request_hash:
raise HTTPException(
status_code=422,
detail="Idempotency key reused with different request body",
)
# Return cached response
return (
json.loads(existing.response_body),
existing.response_status,
True,
)
# Process the request
try:
response_body, status_code = await process_func()
# Store the result
record = IdempotencyRecord(
idempotency_key=idempotency_key,
endpoint=endpoint,
request_hash=request_hash,
response_body=json.dumps(response_body),
response_status=status_code,
expires_at=datetime.now(UTC) + timedelta(hours=24),
)
db.add(record)
await db.commit()
return (response_body, status_code, False)
except Exception:
# Don't cache errors - allow retry
await db.rollback()
raise
# Usage in endpoint
@app.post("/api/orders")
async def create_order(
order: OrderCreate,
idempotency_key: str = Header(..., alias="Idempotency-Key"),
db: AsyncSession = Depends(get_db),
):
async def process():
# Actual order creation logic
new_order = Order(**order.model_dump())
db.add(new_order)
await db.commit()
return {"order_id": str(new_order.id)}, 201
response, status, replayed = await get_or_create_idempotency(
db=db,
idempotency_key=idempotency_key,
endpoint="/api/orders",
request_body=order.model_dump(),
process_func=process,
)
return JSONResponse(
content=response,
status_code=status,
headers={"Idempotent-Replayed": "true"} if replayed else {},
)Event Consumer Idempotency
from dataclasses import dataclass
from datetime import datetime
import asyncpg
@dataclass
class ProcessedEvent:
event_id: str
event_type: str
processed_at: datetime
result: str | None
class IdempotentEventProcessor:
"""Process events exactly once using database tracking."""
def __init__(self, pool: asyncpg.Pool):
self.pool = pool
async def setup(self):
"""Create tracking table if not exists."""
async with self.pool.acquire() as conn:
await conn.execute("""
CREATE TABLE IF NOT EXISTS processed_events (
event_id VARCHAR(64) PRIMARY KEY,
event_type VARCHAR(128) NOT NULL,
processed_at TIMESTAMPTZ DEFAULT NOW(),
result TEXT
)
""")
await conn.execute("""
CREATE INDEX IF NOT EXISTS ix_processed_events_type
ON processed_events (event_type, processed_at)
""")
async def is_processed(self, event_id: str) -> bool:
"""Check if event was already processed."""
async with self.pool.acquire() as conn:
result = await conn.fetchval(
"SELECT 1 FROM processed_events WHERE event_id = $1",
event_id,
)
return result is not None
async def process_event(
self,
event_id: str,
event_type: str,
handler: Callable,
*args,
**kwargs,
) -> tuple[any, bool]:
"""
Process event idempotently.
Returns:
(result, was_duplicate)
"""
async with self.pool.acquire() as conn:
async with conn.transaction():
# Try to insert tracking record (fails if exists)
try:
await conn.execute(
"""
INSERT INTO processed_events (event_id, event_type)
VALUES ($1, $2)
""",
event_id,
event_type,
)
except asyncpg.UniqueViolationError:
# Already processed
existing = await conn.fetchrow(
"SELECT result FROM processed_events WHERE event_id = $1",
event_id,
)
return existing["result"], True
# Process the event
result = await handler(*args, **kwargs)
# Update with result
await conn.execute(
"UPDATE processed_events SET result = $1 WHERE event_id = $2",
json.dumps(result) if result else None,
event_id,
)
return result, False
# Usage with Kafka consumer
async def consume_orders(processor: IdempotentEventProcessor):
consumer = AIOKafkaConsumer("orders", bootstrap_servers="localhost:9092")
await consumer.start()
try:
async for msg in consumer:
event = json.loads(msg.value)
event_id = event["event_id"]
result, was_duplicate = await processor.process_event(
event_id=event_id,
event_type="order.created",
handler=handle_order_created,
order_data=event["data"],
)
if was_duplicate:
logger.info(f"Skipped duplicate event: {event_id}")
else:
logger.info(f"Processed event: {event_id}")
finally:
await consumer.stop()Client-Side Retry with Idempotency
import httpx
import uuid
from tenacity import retry, stop_after_attempt, wait_exponential
class IdempotentClient:
"""HTTP client with automatic idempotency key handling."""
def __init__(self, base_url: str):
self.client = httpx.AsyncClient(base_url=base_url)
@retry(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=0.5, min=0.5, max=10),
reraise=True,
)
async def post_idempotent(
self,
path: str,
json: dict,
idempotency_key: str | None = None,
) -> httpx.Response:
"""
POST with idempotency key.
Args:
path: API endpoint path
json: Request body
idempotency_key: Optional key (auto-generated if not provided)
Returns:
Response from server (may be replayed)
"""
key = idempotency_key or str(uuid.uuid4())
response = await self.client.post(
path,
json=json,
headers={"Idempotency-Key": key},
)
# Don't retry client errors (4xx)
if 400 <= response.status_code < 500:
response.raise_for_status()
return response
async def create_payment(self, amount: int, currency: str) -> dict:
"""Create payment with idempotency protection."""
# Use deterministic key based on content
key = hashlib.sha256(
f"payment:{amount}:{currency}:{datetime.now().date()}".encode()
).hexdigest()
response = await self.post_idempotent(
"/api/payments",
json={"amount": amount, "currency": currency},
idempotency_key=key,
)
return response.json()
# Usage
async def main():
client = IdempotentClient("https://api.example.com")
# Safe to retry - same key prevents duplicate
payment = await client.create_payment(amount=1000, currency="USD")
# Check if it was a replay
if payment.get("_replayed"):
print("Payment was already processed")
else:
print(f"New payment created: {payment['id']}")Cleanup Job for Expired Records
import asyncio
from datetime import UTC, datetime, timedelta
async def cleanup_expired_idempotency_records(
db: AsyncSession,
retention_days: int = 7,
batch_size: int = 1000,
):
"""
Delete expired idempotency records in batches.
Run this as a scheduled job (e.g., daily).
"""
cutoff = datetime.now(UTC) - timedelta(days=retention_days)
total_deleted = 0
while True:
# Delete in batches to avoid long locks
result = await db.execute(
text("""
DELETE FROM idempotency_records
WHERE id IN (
SELECT id FROM idempotency_records
WHERE expires_at < :cutoff
LIMIT :batch_size
)
"""),
{"cutoff": cutoff, "batch_size": batch_size},
)
await db.commit()
deleted = result.rowcount
total_deleted += deleted
if deleted < batch_size:
break
# Small delay to reduce database load
await asyncio.sleep(0.1)
return total_deleted
# Redis cleanup (handled by TTL, but can force cleanup)
async def cleanup_redis_idempotency_keys(redis_client, pattern: str = "idem:*"):
"""Scan and delete expired keys (if TTL not working)."""
cursor = 0
deleted = 0
while True:
cursor, keys = await redis_client.scan(cursor, match=pattern, count=100)
for key in keys:
ttl = await redis_client.ttl(key)
if ttl == -1: # No TTL set
await redis_client.delete(key)
deleted += 1
if cursor == 0:
break
return deletedTesting Idempotency
import pytest
from httpx import ASGITransport, AsyncClient
@pytest.mark.asyncio
async def test_idempotent_request_returns_same_response():
"""Same idempotency key returns cached response."""
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
idempotency_key = str(uuid.uuid4())
# First request
response1 = await client.post(
"/api/orders",
json={"product": "widget", "quantity": 1},
headers={"Idempotency-Key": idempotency_key},
)
assert response1.status_code == 201
order_id = response1.json()["order_id"]
# Second request with same key
response2 = await client.post(
"/api/orders",
json={"product": "widget", "quantity": 1},
headers={"Idempotency-Key": idempotency_key},
)
assert response2.status_code == 201
assert response2.json()["order_id"] == order_id
assert response2.headers.get("Idempotent-Replayed") == "true"
@pytest.mark.asyncio
async def test_different_keys_process_independently():
"""Different idempotency keys process as separate requests."""
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
response1 = await client.post(
"/api/orders",
json={"product": "widget", "quantity": 1},
headers={"Idempotency-Key": str(uuid.uuid4())},
)
response2 = await client.post(
"/api/orders",
json={"product": "widget", "quantity": 1},
headers={"Idempotency-Key": str(uuid.uuid4())},
)
assert response1.json()["order_id"] != response2.json()["order_id"]
@pytest.mark.asyncio
async def test_mismatched_body_rejected():
"""Reusing key with different body is rejected."""
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
idempotency_key = str(uuid.uuid4())
await client.post(
"/api/orders",
json={"product": "widget", "quantity": 1},
headers={"Idempotency-Key": idempotency_key},
)
response = await client.post(
"/api/orders",
json={"product": "gadget", "quantity": 2}, # Different body!
headers={"Idempotency-Key": idempotency_key},
)
assert response.status_code == 422
assert "different request body" in response.json()["detail"]OrchestKit Workflow Resilience Integration
This example shows how to wire resilience patterns into the OrchestKit analysis pipeline.
Current Architecture
┌────────────────────────────────────────────────────────────────────┐
│ OrchestKit Analysis Pipeline │
├────────────────────────────────────────────────────────────────────┤
│ │
│ Content ─▶ [Supervisor] ─▶ [Agent Fan-Out] ─▶ [Aggregate] ─▶ ... │
│ │ │ │ │ │
│ ▼ ▼ ▼ ▼ │
│ Agent Selection A1 A2 A3 (Parallel Analysis) │
│ │ │ │ │
│ ▼ ▼ ▼ │
│ [findings, findings, findings] │
│ │ │
│ ▼ │
│ [Synthesize] ─▶ [Quality Gate] │
│ │
└────────────────────────────────────────────────────────────────────┘Resilience Layer Integration
┌────────────────────────────────────────────────────────────────────┐
│ With Resilience Patterns │
├────────────────────────────────────────────────────────────────────┤
│ │
│ Content ─▶ [Rate Limiter] ─▶ [Circuit Breaker: LLM] ─▶ ... │
│ │ │
│ ┌──────────────────────┼──────────────────────┐ │
│ │ ▼ │ │
│ │ ┌──────────────────────────────────────┐ │ │
│ │ │ TIER 1: CRITICAL │ │ │
│ │ │ [Supervisor] [Synthesis] [QualityGate]│ │ │
│ │ │ Bulkhead: 5 concurrent, 300s timeout │ │ │
│ │ └──────────────────────────────────────┘ │ │
│ │ │ │
│ │ ┌──────────────────────────────────────┐ │ │
│ │ │ TIER 2: STANDARD │ │ │
│ │ │ [Tech Comparator] [Impl Planner] │ │ │
│ │ │ [Security Auditor] [Learning Synth] │ │ │
│ │ │ Bulkhead: 3 concurrent, 120s timeout │ │ │
│ │ └──────────────────────────────────────┘ │ │
│ │ │ │
│ │ ┌──────────────────────────────────────┐ │ │
│ │ │ TIER 3: OPTIONAL │ │ │
│ │ │ [Enrichment] [Cache Warm] │ │ │
│ │ │ Bulkhead: 2 concurrent, 60s timeout │ │ │
│ │ └──────────────────────────────────────┘ │ │
│ │ │ │
│ └──────────────────────────────────────────────┘ │
│ │
│ Each agent call wrapped with: │
│ @circuit_breaker(name="agent-{agent_type}") │
│ @bulkhead(tier=agent.tier) │
│ @retry(max_attempts=2, base_delay=1.0) │
│ │
└────────────────────────────────────────────────────────────────────┘Implementation
1. Circuit Breaker Registry
# backend/app/shared/resilience/circuit_breakers.py
from app.core.circuit_breaker import CircuitBreaker
# Per-service circuit breakers
circuit_breakers = {
# LLM APIs
"openai": CircuitBreaker(
name="openai",
failure_threshold=3,
recovery_timeout=60.0,
),
"anthropic": CircuitBreaker(
name="anthropic",
failure_threshold=3,
recovery_timeout=60.0,
),
# External APIs
"youtube": CircuitBreaker(
name="youtube",
failure_threshold=5,
recovery_timeout=120.0,
),
"arxiv": CircuitBreaker(
name="arxiv",
failure_threshold=5,
recovery_timeout=60.0,
),
"github": CircuitBreaker(
name="github",
failure_threshold=5,
recovery_timeout=60.0,
),
# Internal services
"embedding": CircuitBreaker(
name="embedding",
failure_threshold=3,
recovery_timeout=30.0,
),
"database": CircuitBreaker(
name="database",
failure_threshold=2,
recovery_timeout=15.0,
),
}
def get_circuit_breaker(service: str) -> CircuitBreaker:
"""Get or create circuit breaker for service."""
if service not in circuit_breakers:
circuit_breakers[service] = CircuitBreaker(
name=service,
failure_threshold=5,
recovery_timeout=30.0,
)
return circuit_breakers[service]2. Bulkhead Registry
# backend/app/shared/resilience/bulkheads.py
from enum import Enum
from .bulkhead import Bulkhead, Tier, BulkheadRegistry
# Agent tier assignments
AGENT_TIERS = {
# Tier 1: Critical
"supervisor": Tier.CRITICAL,
"synthesis": Tier.CRITICAL,
"quality_gate": Tier.CRITICAL,
# Tier 2: Standard
"tech_comparator": Tier.STANDARD,
"implementation_planner": Tier.STANDARD,
"security_auditor": Tier.STANDARD,
"learning_synthesizer": Tier.STANDARD,
"codebase_analyzer": Tier.STANDARD,
"prerequisite_mapper": Tier.STANDARD,
"practical_applicator": Tier.STANDARD,
"complexity_assessor": Tier.STANDARD,
# Tier 3: Optional
"enrichment": Tier.OPTIONAL,
"cache_warm": Tier.OPTIONAL,
"metrics": Tier.OPTIONAL,
}
# Create registry
bulkhead_registry = BulkheadRegistry()
# Register bulkheads for each tier
for agent_name, tier in AGENT_TIERS.items():
bulkhead_registry.register(agent_name, tier)
def get_agent_bulkhead(agent_type: str) -> Bulkhead:
"""Get bulkhead for agent type."""
tier = AGENT_TIERS.get(agent_type, Tier.STANDARD)
return bulkhead_registry.get_or_create(agent_type, tier)3. Resilient Agent Wrapper
# backend/app/shared/resilience/agent_wrapper.py
from functools import wraps
from typing import TypeVar, Callable, Awaitable
import structlog
from .circuit_breakers import get_circuit_breaker
from .bulkheads import get_agent_bulkhead
from .retry_handler import retry, MaxRetriesExceededError
logger = structlog.get_logger()
T = TypeVar("T")
def resilient_agent(
agent_type: str,
llm_service: str = "anthropic",
max_retries: int = 2,
):
"""
Decorator to wrap agent execution with resilience patterns.
Applies (in order):
1. Circuit breaker for LLM service
2. Bulkhead for concurrency control
3. Retry for transient failures
Example:
@resilient_agent("tech_comparator", llm_service="anthropic")
async def run_tech_comparator(content: str) -> AgentOutput:
...
"""
def decorator(fn: Callable[..., Awaitable[T]]) -> Callable[..., Awaitable[T]]:
@wraps(fn)
async def wrapper(*args, **kwargs) -> T:
circuit = get_circuit_breaker(llm_service)
bulkhead = get_agent_bulkhead(agent_type)
# Track for observability
logger.info(
"agent_execution_start",
agent_type=agent_type,
circuit_state=circuit.state.value,
bulkhead_active=bulkhead.stats.current_active,
)
async def execute():
# Retry layer (innermost)
@retry(max_attempts=max_retries, base_delay=1.0)
async def with_retry():
return await fn(*args, **kwargs)
return await with_retry()
try:
# Bulkhead layer
async def with_bulkhead():
return await bulkhead.execute(execute)
# Circuit breaker layer (outermost)
result = await circuit.call(with_bulkhead)
logger.info(
"agent_execution_success",
agent_type=agent_type,
)
return result
except CircuitOpenError as e:
logger.warning(
"agent_circuit_open",
agent_type=agent_type,
time_until_recovery=e.time_until_recovery,
)
raise
except BulkheadFullError as e:
logger.warning(
"agent_bulkhead_full",
agent_type=agent_type,
tier=e.tier.name,
)
raise
except MaxRetriesExceededError as e:
logger.error(
"agent_max_retries_exceeded",
agent_type=agent_type,
attempts=e.attempts,
)
raise
return wrapper
return decorator4. Graph Builder Integration
# backend/app/domains/analysis/workflows/graph_builder.py
from app.shared.resilience.agent_wrapper import resilient_agent
from app.shared.resilience.circuit_breakers import circuit_breakers
async def build_analysis_graph() -> StateGraph:
"""Build the analysis workflow graph with resilience."""
# Wrap each agent node with resilience
@resilient_agent("supervisor", llm_service="anthropic")
async def supervisor_node(state: AnalysisState) -> AnalysisState:
# Existing supervisor logic
...
@resilient_agent("tech_comparator", llm_service="anthropic")
async def tech_comparator_node(state: AnalysisState) -> AnalysisState:
# Existing agent logic
...
# Build graph with wrapped nodes
graph = StateGraph(AnalysisState)
graph.add_node("supervisor", supervisor_node)
graph.add_node("tech_comparator", tech_comparator_node)
# ... add other nodes
# Add health check endpoint
@app.get("/health/resilience")
async def resilience_health():
return {
"circuit_breakers": {
name: cb.get_status()
for name, cb in circuit_breakers.items()
},
"bulkheads": bulkhead_registry.get_all_status(),
}
return graph.compile()5. LLM Fallback Chain Integration
# backend/app/shared/resilience/llm_chain.py
from app.shared.resilience.llm_fallback_chain import (
LLMFallbackChain,
LLMConfig,
)
from app.shared.services.cache.semantic_cache import SemanticCache
# Configure fallback chain for analysis
analysis_llm_chain = LLMFallbackChain(
primary=AnthropicProvider(
LLMConfig(
name="primary",
model="claude-sonnet-4-6",
timeout=60.0,
max_tokens=8192,
)
),
fallbacks=[
OpenAIProvider(
LLMConfig(
name="fallback",
model="gpt-5-mini",
timeout=30.0,
max_tokens=4096,
)
),
],
cache=SemanticCache(
redis_client=redis_client,
threshold=0.85,
),
default_response=lambda p: json.dumps({
"status": "degraded",
"message": "Analysis temporarily unavailable",
"partial_results": None,
}),
)6. Observability Integration
# backend/app/shared/resilience/observability.py
from langfuse import Langfuse
langfuse = Langfuse()
def on_circuit_state_change(old_state: str, new_state: str, name: str):
"""Record circuit state changes in Langfuse."""
langfuse.event(
name="circuit_breaker_state_change",
metadata={
"circuit_name": name,
"old_state": old_state,
"new_state": new_state,
},
level="WARNING" if new_state == "open" else "INFO",
)
def on_bulkhead_rejection(name: str, tier: str):
"""Record bulkhead rejections."""
langfuse.event(
name="bulkhead_rejection",
metadata={
"bulkhead_name": name,
"tier": tier,
},
level="WARNING",
)
# Wire up callbacks
for name, cb in circuit_breakers.items():
cb._on_state_change = on_circuit_state_change
for name, bh in bulkhead_registry._bulkheads.items():
bh._on_rejection = on_bulkhead_rejectionConfiguration
Environment Variables
# Circuit Breaker
CIRCUIT_FAILURE_THRESHOLD=5
CIRCUIT_RECOVERY_TIMEOUT=30
# Bulkhead Tier 1
BULKHEAD_TIER1_CONCURRENT=5
BULKHEAD_TIER1_QUEUE=10
BULKHEAD_TIER1_TIMEOUT=300
# Bulkhead Tier 2
BULKHEAD_TIER2_CONCURRENT=3
BULKHEAD_TIER2_QUEUE=5
BULKHEAD_TIER2_TIMEOUT=120
# Bulkhead Tier 3
BULKHEAD_TIER3_CONCURRENT=2
BULKHEAD_TIER3_QUEUE=3
BULKHEAD_TIER3_TIMEOUT=60
# Retry
RETRY_MAX_ATTEMPTS=3
RETRY_BASE_DELAY=1.0
RETRY_MAX_DELAY=30.0Monitoring Dashboard
┌────────────────────────────────────────────────────────────────────┐
│ OrchestKit Resilience Dashboard │
├────────────────────────────────────────────────────────────────────┤
│ │
│ CIRCUIT BREAKERS │
│ ┌─────────────┬──────────┬────────────┬───────────────────┐ │
│ │ Service │ State │ Failures │ Next Recovery │ │
│ ├─────────────┼──────────┼────────────┼───────────────────┤ │
│ │ anthropic │ ✅ CLOSED │ 0/5 │ - │ │
│ │ openai │ ✅ CLOSED │ 1/5 │ - │ │
│ │ youtube │ ⚠️ OPEN │ 5/5 │ 45s │ │
│ │ embedding │ ✅ CLOSED │ 0/3 │ - │ │
│ └─────────────┴──────────┴────────────┴───────────────────┘ │
│ │
│ BULKHEADS │
│ ┌─────────────────┬────────┬─────────┬─────────┬──────────┐ │
│ │ Tier │ Active │ Queued │ Max │ Rejected │ │
│ ├─────────────────┼────────┼─────────┼─────────┼──────────┤ │
│ │ 1: Critical │ 3/5 │ 1/10 │ 5 │ 0 │ │
│ │ 2: Standard │ 3/3 ⚠️ │ 4/5 │ 3 │ 12 │ │
│ │ 3: Optional │ 1/2 │ 0/3 │ 2 │ 45 │ │
│ └─────────────────┴────────┴─────────┴─────────┴──────────┘ │
│ │
│ RETRY STATS (Last Hour) │
│ • Total Attempts: 1,234 │
│ • Success Rate: 94.2% │
│ • Retries Used: 187 │
│ • Max Retries Exceeded: 23 │
│ │
└────────────────────────────────────────────────────────────────────┘Testing Resilience
# backend/tests/integration/test_resilience.py
import pytest
from unittest.mock import AsyncMock, patch
async def test_circuit_opens_after_failures():
"""Circuit should open after threshold failures."""
breaker = get_circuit_breaker("test-service")
# Simulate failures
for _ in range(5):
with pytest.raises(ConnectionError):
await breaker.call(failing_function)
# Next call should be rejected
with pytest.raises(CircuitOpenError):
await breaker.call(failing_function)
async def test_bulkhead_rejects_when_full():
"""Bulkhead should reject when queue is full."""
bulkhead = Bulkhead("test", Tier.STANDARD, max_concurrent=1, queue_size=1)
# Fill the bulkhead
task1 = asyncio.create_task(bulkhead.execute(slow_function))
task2 = asyncio.create_task(bulkhead.execute(slow_function))
# Third should be rejected
with pytest.raises(BulkheadFullError):
await bulkhead.execute(slow_function)
async def test_fallback_chain_uses_fallback():
"""Chain should use fallback when primary fails."""
chain = LLMFallbackChain(
primary=FailingProvider(),
fallbacks=[MockProvider()],
)
response = await chain.complete("test prompt")
assert response.is_fallback
assert response.source == ResponseSource.FALLBACK{
"version": "2.0.0",
"organization": "OrchestKit",
"date": "February 2026",
"abstract": "Distributed systems patterns covering distributed locking (Redis, Redlock, PostgreSQL advisory), resilience (circuit breaker, retry with backoff, bulkhead isolation), idempotency (key generation, request deduplication, database-backed), and rate limiting (token bucket, sliding window, distributed).",
"ruleCount": 12,
"categories": 4,
"consolidatedFrom": [
"distributed-locks",
"resilience-patterns",
"idempotency-patterns",
"rate-limiting"
]
}
Bulkhead Pattern
Overview
The bulkhead pattern isolates failures by partitioning system resources into independent pools. Named after ship bulkheads that prevent flooding from spreading, it ensures one failing component doesn't bring down the entire system.
Types of Bulkheads
1. Thread Pool Isolation
Dedicated thread pools per service/operation.
┌────────────────────────────────────────────────────────────┐
│ Thread Pool Bulkhead │
├────────────────────────────────────────────────────────────┤
│ │
│ Service A Pool (5 threads) Service B Pool (3 threads) │
│ ┌─┬─┬─┬─┬─┐ ┌─┬─┬─┐ │
│ │█│█│█│░│░│ │█│░│░│ │
│ └─┴─┴─┴─┴─┘ └─┴─┴─┘ │
│ │
│ If Service A hangs, only 5 threads blocked │
│ Service B continues with its own 3 threads │
│ │
└────────────────────────────────────────────────────────────┘2. Semaphore Isolation
Limits concurrent executions without dedicated threads.
┌────────────────────────────────────────────────────────────┐
│ Semaphore Bulkhead │
├────────────────────────────────────────────────────────────┤
│ │
│ Semaphore: permits=5, current=3 │
│ │
│ Request 1: acquire() → ✓ (permits=2) │
│ Request 2: acquire() → ✓ (permits=1) │
│ Request 3: acquire() → ✓ (permits=0) │
│ Request 4: acquire() → BLOCKED (queue) or REJECTED │
│ │
│ Request 1: release() → ✓ (permits=1) │
│ Request 4: acquire() → ✓ (permits=0) │
│ │
└────────────────────────────────────────────────────────────┘3. Tier-Based Bulkheads (Recommended for Multi-Agent)
Group operations by criticality.
┌────────────────────────────────────────────────────────────┐
│ Tier-Based Bulkheads │
├────────────────────────────────────────────────────────────┤
│ │
│ TIER 1: CRITICAL (50% resources) │
│ ├── Synthesis node │
│ ├── Quality gate │
│ └── User-facing responses │
│ │
│ TIER 2: STANDARD (35% resources) │
│ ├── Content analysis agents │
│ ├── Data processing │
│ └── API integrations │
│ │
│ TIER 3: OPTIONAL (15% resources) │
│ ├── Enrichment │
│ ├── Caching warmup │
│ └── Analytics │
│ │
│ When Tier 3 exhausted → operations queued/dropped │
│ Tier 1 & 2 continue unaffected │
│ │
└────────────────────────────────────────────────────────────┘Configuration for OrchestKit
Agent Tier Assignment
| Tier | Agents | Max Concurrent | Queue Size | Timeout |
|---|---|---|---|---|
| 1 (Critical) | synthesis, quality_gate, supervisor | 5 | 10 | 300s |
| 2 (Standard) | tech_comparator, implementation_planner, security_auditor, learning_synthesizer | 3 | 5 | 120s |
| 3 (Optional) | enrichment, cache_warming, metrics | 2 | 3 | 60s |
Rejection Policies
class RejectionPolicy(Enum):
ABORT = "abort" # Return error immediately
CALLER_RUNS = "caller" # Execute in caller's context (blocking)
DISCARD = "discard" # Silently drop (for optional ops)
QUEUE = "queue" # Wait in bounded queue
# Per-tier policies
TIER_POLICIES = {
1: RejectionPolicy.QUEUE, # Critical: wait for slot
2: RejectionPolicy.CALLER_RUNS, # Standard: degrade caller
3: RejectionPolicy.DISCARD, # Optional: skip if busy
}Implementation Pattern (Python asyncio)
from asyncio import Semaphore, wait_for, TimeoutError
from collections import defaultdict
from enum import Enum
from typing import TypeVar, Callable, Awaitable
T = TypeVar("T")
class Tier(Enum):
CRITICAL = 1
STANDARD = 2
OPTIONAL = 3
class Bulkhead:
def __init__(self, tier: Tier, max_concurrent: int, queue_size: int, timeout: float):
self.tier = tier
self.semaphore = Semaphore(max_concurrent)
self.queue_size = queue_size
self.timeout = timeout
self.waiting = 0
self.active = 0
async def execute(self, fn: Callable[[], Awaitable[T]]) -> T:
# Check queue
if self.waiting >= self.queue_size:
raise BulkheadFullError(f"Tier {self.tier.name} queue full")
self.waiting += 1
try:
# Acquire with timeout
await wait_for(self.semaphore.acquire(), timeout=self.timeout)
self.waiting -= 1
self.active += 1
try:
return await wait_for(fn(), timeout=self.timeout)
finally:
self.active -= 1
self.semaphore.release()
except TimeoutError:
self.waiting -= 1
raise BulkheadTimeoutError(f"Tier {self.tier.name} timeout")Best Practices (2026)
1. Size Based on Downstream Capacity
# BAD: Arbitrary numbers
bulkhead = Bulkhead(max_concurrent=100)
# GOOD: Based on downstream limits
# If OpenAI allows 60 RPM, don't have 100 concurrent
bulkhead = Bulkhead(max_concurrent=10) # 10 concurrent * 6s avg = 60 RPM2. Monitor Queue Depth
async def execute_with_metrics(self, fn):
# Metric: queue depth
metrics.gauge("bulkhead.queue_depth", self.waiting, tags={"tier": self.tier.name})
# Metric: active requests
metrics.gauge("bulkhead.active", self.active, tags={"tier": self.tier.name})
# Alert when queue consistently > 80% full
if self.waiting > self.queue_size * 0.8:
logger.warning(f"Bulkhead queue high", tier=self.tier.name, depth=self.waiting)
return await self.execute(fn)3. Graceful Degradation by Tier
async def run_analysis(content: str) -> Analysis:
results = {}
# Tier 1: Must succeed
results["core"] = await tier1_bulkhead.execute(
lambda: analyze_core(content)
)
# Tier 2: Best effort
try:
results["enriched"] = await tier2_bulkhead.execute(
lambda: enrich_analysis(content)
)
except BulkheadFullError:
results["enriched"] = None # Skip enrichment
# Tier 3: Optional
try:
await tier3_bulkhead.execute(
lambda: warm_cache(results)
)
except (BulkheadFullError, BulkheadTimeoutError):
pass # Don't even log
return Analysis(**results)4. Dynamic Tier Adjustment
class AdaptiveBulkhead:
"""Adjusts tier capacity based on system load."""
def adjust_for_load(self, cpu_percent: float, memory_percent: float):
if cpu_percent > 80 or memory_percent > 85:
# Reduce optional tier
self.tiers[Tier.OPTIONAL].max_concurrent = 1
self.tiers[Tier.STANDARD].max_concurrent = 2
elif cpu_percent < 50 and memory_percent < 60:
# Restore capacity
self.tiers[Tier.OPTIONAL].max_concurrent = 2
self.tiers[Tier.STANDARD].max_concurrent = 3Anti-Patterns
1. Too Many Bulkheads
# BAD: Bulkhead per endpoint
bulkheads = {
"/api/v1/users": Bulkhead(5),
"/api/v1/users/{id}": Bulkhead(5),
"/api/v1/users/{id}/profile": Bulkhead(5),
# ... 50 more
}
# Result: Complexity nightmare, no real isolation
# GOOD: Bulkhead per tier/dependency
bulkheads = {
"database_read": Bulkhead(10),
"database_write": Bulkhead(3),
"external_api": Bulkhead(5),
}2. Ignoring Rejection Handling
# BAD: Exception bubbles up as 500
@app.post("/analyze")
async def analyze(content: str):
return await bulkhead.execute(lambda: do_analysis(content))
# BulkheadFullError → 500 Internal Server Error
# GOOD: Proper error handling
@app.post("/analyze")
async def analyze(content: str):
try:
return await bulkhead.execute(lambda: do_analysis(content))
except BulkheadFullError:
raise HTTPException(
status_code=503,
detail="Service busy, please retry",
headers={"Retry-After": "30"}
)3. No Correlation with Circuit Breaker
# BAD: Bulkhead fills up, circuit never opens
# All slots blocked on slow service
# GOOD: Combine patterns
@circuit_breaker(failure_threshold=5)
@bulkhead(tier=Tier.STANDARD)
async def call_external_service():
...
# Slow calls → timeouts → circuit opens → bulkhead clearedMonitoring Dashboard
┌────────────────────────────────────────────────────────────┐
│ Bulkhead Status Dashboard │
├────────────────────────────────────────────────────────────┤
│ │
│ TIER 1: CRITICAL [████████░░] 8/10 active │
│ Queue: 2/10 [██░░░░░░░░] 2/10 queued │
│ Rejected (1h): 0 Timeouts (1h): 1 │
│ │
│ TIER 2: STANDARD [██████████] 3/3 active ⚠️ │
│ Queue: 5/5 FULL [██████████] 5/5 queued ⚠️ │
│ Rejected (1h): 23 Timeouts (1h): 5 │
│ │
│ TIER 3: OPTIONAL [█░░░░░░░░░] 1/2 active │
│ Queue: 0/3 [░░░░░░░░░░] 0/3 queued │
│ Rejected (1h): 156 Timeouts (1h): 0 │
│ │
└────────────────────────────────────────────────────────────┘Circuit Breaker Pattern
Overview
The circuit breaker pattern prevents cascade failures by "tripping" when a downstream service exceeds failure thresholds. Named after electrical circuit breakers, it protects your system from repeated failures.
States
CLOSED (Normal Operation)
- All requests pass through
- Failures are counted within a sliding window
- Success resets failure count (or decrements in sliding window)
- Transitions to OPEN when failures >= threshold
OPEN (Failing Fast)
- All requests immediately rejected
- Returns fallback response or error
- No calls made to downstream service
- After
recovery_timeout, transitions to HALF_OPEN
HALF_OPEN (Recovery Probe)
- Limited requests allowed (probe requests)
- If probe succeeds → CLOSED
- If probe fails → OPEN (reset recovery timer)
State Machine
failures >= threshold
CLOSED ──────────────────────────────▶ OPEN
▲ │
│ │
│ probe succeeds timeout │
│ expires │
│ ┌─────────────┐ │
└─────────│ HALF_OPEN │◀────────────┘
└─────────────┘
│
│ probe fails
▼
OPENConfiguration Parameters
| Parameter | Recommended | Description |
|---|---|---|
failure_threshold | 5 | Failures before opening |
success_threshold | 2 | Successes in half-open to close |
recovery_timeout | 30s | Time before half-open transition |
sliding_window_size | 10 | Requests to consider for failure rate |
sliding_window_type | count-based | count-based or time-based (60s) |
slow_call_threshold | 5s | Calls slower than this count as failures |
slow_call_rate | 50% | Percentage of slow calls to trip |
Best Practices (2026)
1. Use Sliding Windows, Not Fixed Counters
# BAD: Fixed counter resets on success
if failures >= 5:
open_circuit()
if success:
failures = 0 # One success resets everything!
# GOOD: Sliding window with time decay
window = deque(maxlen=10)
window.append(("fail", time.time()))
failure_rate = sum(1 for r, _ in window if r == "fail") / len(window)
if failure_rate >= 0.5:
open_circuit()2. Separate Health Checks from Circuit State
# Health endpoint should NOT check circuit state
@app.get("/health")
async def health():
return {"status": "healthy"} # Always returns 200
# Readiness endpoint CAN check circuit state
@app.get("/ready")
async def ready():
if circuit.is_open:
return {"status": "degraded", "reason": "circuit_open"}, 503
return {"status": "ready"}3. Include Observability
def on_state_change(from_state: str, to_state: str, service: str):
# Metric
metrics.increment(f"circuit_breaker.{service}.state_change",
tags={"from": from_state, "to": to_state})
# Log
logger.warning(f"Circuit breaker state change",
service=service, from_state=from_state, to_state=to_state)
# Alert (only on OPEN)
if to_state == "OPEN":
alert_service.send(
severity="warning",
message=f"Circuit breaker opened for {service}",
runbook="https://docs.internal/runbooks/circuit-breaker"
)4. Provide Meaningful Fallbacks
async def get_analysis_with_fallback(content: str) -> Analysis:
try:
return await circuit_breaker.call(analyze_content, content)
except CircuitOpenError:
# Fallback 1: Cached result
cached = await cache.get(f"analysis:{hash(content)}")
if cached:
return Analysis.from_cache(cached, is_stale=True)
# Fallback 2: Simplified analysis
return Analysis(
status="degraded",
message="Full analysis unavailable, showing basic info",
basic_info=extract_basic_info(content)
)5. Per-Service Breakers
# BAD: Single breaker for all services
global_breaker = CircuitBreaker()
# GOOD: Per-service breakers
breakers = {
"openai": CircuitBreaker(failure_threshold=3, recovery_timeout=60),
"anthropic": CircuitBreaker(failure_threshold=5, recovery_timeout=30),
"youtube_api": CircuitBreaker(failure_threshold=10, recovery_timeout=120),
}Anti-Patterns
1. Opening Too Quickly
# BAD: Opens on first failure
CircuitBreaker(failure_threshold=1) # One blip = outage
# GOOD: Tolerates transient failures
CircuitBreaker(failure_threshold=5, sliding_window_size=10)2. Recovery Timeout Too Short
# BAD: Hammers failing service
CircuitBreaker(recovery_timeout=5) # Tries every 5 seconds
# GOOD: Gives service time to recover
CircuitBreaker(recovery_timeout=30) # 30 seconds minimum3. No Fallback
# BAD: Just throws error
async def call():
if circuit.is_open:
raise CircuitOpenError() # User sees error page
# GOOD: Graceful degradation
async def call():
if circuit.is_open:
return await fallback_handler() # User sees partial dataIntegration with Other Patterns
Circuit Breaker + Retry
# Retry INSIDE circuit breaker
@circuit_breaker
@retry(max_attempts=3, backoff=exponential)
async def call_service():
...
# Circuit only sees final result after retries exhaustedCircuit Breaker + Bulkhead
# Bulkhead limits concurrency, circuit limits failures
@circuit_breaker(service="analysis")
@bulkhead(tier=2, max_concurrent=3)
async def analyze():
...Circuit Breaker + Timeout
# Timeout INSIDE circuit breaker
@circuit_breaker
@timeout(seconds=30)
async def call_service():
...
# Timeout counts as failure toward circuit thresholdMonitoring Queries
Prometheus
# Circuit state changes per minute
rate(circuit_breaker_state_changes_total[5m])
# Percentage of time in OPEN state
avg_over_time(circuit_breaker_state{state="open"}[1h])
# Requests rejected due to open circuit
rate(circuit_breaker_rejected_total[5m])Langfuse (LLM-specific)
# Tag traces with circuit state
trace.update(metadata={
"circuit_state": circuit.state,
"circuit_failure_count": circuit.failure_count,
})Error Classification
Overview
Proper error classification is the foundation of resilience. Different errors require different handling strategies: retry, fallback, fail fast, or alert.
Error Classification Matrix
┌────────────────────────────────────────────────────────────────────┐
│ Error Classification Matrix │
├────────────────────────────────────────────────────────────────────┤
│ │
│ TRANSIENT PERMANENT │
│ ┌─────────────────────┬─────────────────────┐ │
│ │ │ │ │
│ EXTERNAL │ • Rate limits │ • Invalid API key │ │
│ (API/Net) │ • Timeouts │ • 403 Forbidden │ │
│ │ • 502/503/504 │ • 404 Not Found │ │
│ │ • Connection reset │ • 400 Bad Request │ │
│ │ │ │ │
│ │ ACTION: Retry │ ACTION: Fail Fast │ │
│ │ with backoff │ Log & Alert │ │
│ │ │ │ │
│ ├─────────────────────┼─────────────────────┤ │
│ │ │ │ │
│ INTERNAL │ • DB connection │ • Schema error │ │
│ (System) │ • Memory pressure │ • Logic bug │ │
│ │ • Lock contention │ • Missing config │ │
│ │ • Resource exhaust │ • Invalid state │ │
│ │ │ │ │
│ │ ACTION: Retry │ ACTION: Fail Fast │ │
│ │ Circuit breaker │ Fix code, restart │ │
│ │ │ │ │
│ └─────────────────────┴─────────────────────┘ │
│ │
└────────────────────────────────────────────────────────────────────┘HTTP Status Code Classification
Retryable (Transient)
| Code | Name | Strategy |
|---|---|---|
| 408 | Request Timeout | Retry immediately |
| 429 | Too Many Requests | Retry with Retry-After header |
| 500 | Internal Server Error | Retry with backoff |
| 502 | Bad Gateway | Retry with backoff |
| 503 | Service Unavailable | Retry with Retry-After |
| 504 | Gateway Timeout | Retry with backoff |
Non-Retryable (Permanent)
| Code | Name | Strategy |
|---|---|---|
| 400 | Bad Request | Log, fix input, fail |
| 401 | Unauthorized | Refresh token or fail |
| 403 | Forbidden | Fail, alert |
| 404 | Not Found | Fail (resource doesn't exist) |
| 405 | Method Not Allowed | Fail, fix code |
| 409 | Conflict | May retry with merge logic |
| 422 | Unprocessable Entity | Fail, fix input |
LLM API Error Classification
OpenAI Errors
OPENAI_RETRYABLE = {
"rate_limit_exceeded", # Retry with backoff
"server_error", # Retry with backoff
"timeout", # Retry immediately
"overloaded", # Retry with longer backoff
}
OPENAI_NON_RETRYABLE = {
"invalid_api_key", # Fix config
"invalid_request_error", # Fix request
"context_length_exceeded", # Reduce input (special handling)
"content_policy_violation",# Change content
"insufficient_quota", # Add credits
"model_not_found", # Fix model name
}Anthropic Errors
ANTHROPIC_RETRYABLE = {
"overloaded_error", # Retry with backoff
"api_error", # Retry with backoff
"rate_limit_error", # Retry with Retry-After
}
ANTHROPIC_NON_RETRYABLE = {
"authentication_error", # Fix API key
"permission_error", # Check permissions
"invalid_request_error", # Fix request format
"not_found_error", # Fix resource reference
}Exception Classification Helper
from enum import Enum
from typing import Union, Type
import httpx
class ErrorCategory(Enum):
RETRYABLE = "retryable" # Transient, retry with backoff
NON_RETRYABLE = "non_retryable" # Permanent, fail fast
CIRCUIT_TRIP = "circuit_trip" # Count toward circuit breaker
ALERTABLE = "alertable" # Should trigger alert
DEGRADABLE = "degradable" # Can fall back to alternative
class ErrorClassifier:
"""Classify errors for resilience handling."""
RETRYABLE_STATUS_CODES = {408, 429, 500, 502, 503, 504}
NON_RETRYABLE_STATUS_CODES = {400, 401, 403, 404, 405, 422}
ALERTABLE_STATUS_CODES = {401, 403, 500}
RETRYABLE_EXCEPTIONS = {
ConnectionError,
TimeoutError,
ConnectionResetError,
ConnectionRefusedError,
BrokenPipeError,
httpx.ConnectError,
httpx.ConnectTimeout,
httpx.ReadTimeout,
}
def classify(self, error: Exception) -> set[ErrorCategory]:
"""Return set of applicable error categories."""
categories = set()
# HTTP errors
if hasattr(error, "status_code"):
code = error.status_code
if code in self.RETRYABLE_STATUS_CODES:
categories.add(ErrorCategory.RETRYABLE)
if code in self.NON_RETRYABLE_STATUS_CODES:
categories.add(ErrorCategory.NON_RETRYABLE)
if code in self.ALERTABLE_STATUS_CODES:
categories.add(ErrorCategory.ALERTABLE)
if code >= 500:
categories.add(ErrorCategory.CIRCUIT_TRIP)
# Exception types
if type(error) in self.RETRYABLE_EXCEPTIONS:
categories.add(ErrorCategory.RETRYABLE)
categories.add(ErrorCategory.CIRCUIT_TRIP)
# LLM-specific errors
if hasattr(error, "code"):
categories.update(self._classify_llm_error(error.code))
# Default: non-retryable if nothing matched
if not categories:
categories.add(ErrorCategory.NON_RETRYABLE)
categories.add(ErrorCategory.ALERTABLE)
return categories
def _classify_llm_error(self, error_code: str) -> set[ErrorCategory]:
"""Classify LLM API error codes."""
categories = set()
retryable_codes = {
"rate_limit_exceeded", "server_error", "timeout",
"overloaded", "overloaded_error", "api_error",
}
if error_code in retryable_codes:
categories.add(ErrorCategory.RETRYABLE)
categories.add(ErrorCategory.CIRCUIT_TRIP)
else:
categories.add(ErrorCategory.NON_RETRYABLE)
if error_code in {"context_length_exceeded"}:
categories.add(ErrorCategory.DEGRADABLE)
return categories
def should_retry(self, error: Exception) -> bool:
"""Quick check if error should be retried."""
return ErrorCategory.RETRYABLE in self.classify(error)
def should_trip_circuit(self, error: Exception) -> bool:
"""Check if error should count toward circuit breaker."""
return ErrorCategory.CIRCUIT_TRIP in self.classify(error)
def should_alert(self, error: Exception) -> bool:
"""Check if error should trigger an alert."""
return ErrorCategory.ALERTABLE in self.classify(error)Usage in Resilience Patterns
With Retry
classifier = ErrorClassifier()
async def call_with_smart_retry(fn, *args, **kwargs):
for attempt in range(max_attempts):
try:
return await fn(*args, **kwargs)
except Exception as e:
categories = classifier.classify(e)
if ErrorCategory.NON_RETRYABLE in categories:
logger.error(f"Non-retryable error: {e}")
raise
if ErrorCategory.ALERTABLE in categories:
await alert_service.send(
severity="warning",
message=f"Retryable error in {fn.__name__}",
error=str(e),
)
if attempt < max_attempts - 1:
await asyncio.sleep(backoff(attempt))
raise MaxRetriesExceeded()With Circuit Breaker
class SmartCircuitBreaker:
def __init__(self, classifier: ErrorClassifier):
self.classifier = classifier
self.failure_count = 0
async def call(self, fn, *args, **kwargs):
try:
result = await fn(*args, **kwargs)
self.failure_count = 0
return result
except Exception as e:
if self.classifier.should_trip_circuit(e):
self.failure_count += 1
if self.failure_count >= self.threshold:
self.open_circuit()
raiseWith Fallback
async def call_with_fallback(primary_fn, fallback_fn, *args):
try:
return await primary_fn(*args)
except Exception as e:
categories = classifier.classify(e)
if ErrorCategory.DEGRADABLE in categories:
logger.info(f"Degrading to fallback: {e}")
return await fallback_fn(*args)
raiseError Context Enrichment
class EnrichedError(Exception):
"""Exception with classification and context."""
def __init__(
self,
original: Exception,
classifier: ErrorClassifier,
context: dict = None,
):
self.original = original
self.categories = classifier.classify(original)
self.context = context or {}
self.timestamp = datetime.now(UTC)
self.trace_id = get_current_trace_id()
super().__init__(str(original))
@property
def is_retryable(self) -> bool:
return ErrorCategory.RETRYABLE in self.categories
@property
def should_alert(self) -> bool:
return ErrorCategory.ALERTABLE in self.categories
def to_dict(self) -> dict:
return {
"error": str(self.original),
"type": type(self.original).__name__,
"categories": [c.value for c in self.categories],
"context": self.context,
"timestamp": self.timestamp.isoformat(),
"trace_id": self.trace_id,
}Best Practices
1. Default to non-retryable: Unknown errors should fail fast 2. Log all classifications: Helps tune classification rules 3. Include context: Error classification without context is useless 4. Review regularly: New error types emerge, update rules 5. Test classification: Unit test your classification logic
LLM-Specific Resilience Patterns
Overview
LLM APIs have unique failure modes that require specialized resilience patterns. This guide covers fallback chains, token budget management, rate limiting, and cost optimization through resilience.
Unique LLM Failure Modes
┌────────────────────────────────────────────────────────────┐
│ LLM Failure Taxonomy │
├────────────────────────────────────────────────────────────┤
│ │
│ TRANSIENT (Retry) PERMANENT (Fail Fast) │
│ ───────────────── ────────────────────── │
│ • rate_limit_exceeded • invalid_api_key │
│ • model_overloaded • content_policy_violation │
│ • server_error • invalid_request_error │
│ • timeout • insufficient_quota │
│ • context_length_exceeded* • model_not_found │
│ │
│ * Can retry with truncation │
│ │
│ DEGRADABLE (Fallback) COSTLY (Budget Control) │
│ ───────────────────── ──────────────────────── │
│ • Primary model down • Large context = high cost │
│ • Quality below threshold • Streaming = token overhead │
│ • Latency too high • Retries multiply cost │
│ │
└────────────────────────────────────────────────────────────┘Pattern 1: Fallback Chain
┌────────────────────────────────────────────────────────────┐
│ LLM Fallback Chain │
├────────────────────────────────────────────────────────────┤
│ │
│ Request ─▶ [Primary Model] ──success──▶ Response │
│ │ │
│ fail (timeout, rate limit, error) │
│ ▼ │
│ [Fallback Model] ──success──▶ Response │
│ │ │
│ fail │
│ ▼ │
│ [Semantic Cache] ──hit──▶ Response │
│ │ │
│ miss │
│ ▼ │
│ [Default Response] ──▶ Graceful Degradation │
│ │
└────────────────────────────────────────────────────────────┘Implementation
from dataclasses import dataclass
from typing import Optional, List, Callable, Awaitable
@dataclass
class LLMConfig:
name: str
model: str
api_key: str
timeout: float = 30.0
max_tokens: int = 4096
temperature: float = 0.7
class FallbackChain:
def __init__(
self,
primary: LLMConfig,
fallbacks: List[LLMConfig],
cache: Optional[SemanticCache] = None,
default_response: Optional[Callable[[str], str]] = None,
):
self.primary = primary
self.fallbacks = fallbacks
self.cache = cache
self.default_response = default_response
async def complete(self, prompt: str, **kwargs) -> LLMResponse:
# Try primary
try:
return await self._call_model(self.primary, prompt, **kwargs)
except RetryableError as e:
logger.warning(f"Primary model failed: {e}")
# Try fallbacks
for fallback in self.fallbacks:
try:
response = await self._call_model(fallback, prompt, **kwargs)
response.is_fallback = True
return response
except RetryableError as e:
logger.warning(f"Fallback {fallback.name} failed: {e}")
# Try cache
if self.cache:
cached = await self.cache.get_similar(prompt, threshold=0.85)
if cached:
logger.info("Returning cached response")
return LLMResponse(
content=cached.content,
is_cached=True,
cache_similarity=cached.similarity,
)
# Default response
if self.default_response:
return LLMResponse(
content=self.default_response(prompt),
is_degraded=True,
)
raise AllModelsFailedError("All LLM options exhausted")Recommended Fallback Configurations
| Use Case | Primary | Fallback 1 | Fallback 2 | Notes |
|---|---|---|---|---|
| Analysis | Claude Sonnet | GPT-5.2-mini | Cache | Quality-first |
| Chat | GPT-5.2 | Claude Haiku | Default msg | Latency-first |
| Embedding | text-embedding-3-large | text-embedding-3-small | - | Dimension compat |
| Code Gen | Claude Sonnet | GPT-5.2 | - | Quality-first |
Pattern 2: Token Budget Management
┌────────────────────────────────────────────────────────────┐
│ Token Budget Guard │
├────────────────────────────────────────────────────────────┤
│ │
│ Context Window: 128K tokens │
│ ┌────────────────────────────────────────────────────┐ │
│ │████████████████████░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░│ │
│ └────────────────────────────────────────────────────┘ │
│ Used: 32K (25%) Available: 96K │
│ │
│ Budget Allocation: │
│ ├── System prompt: 2K (fixed) │
│ ├── Conversation: 20K (sliding window) │
│ ├── Retrieved docs: 8K (chunked) │
│ ├── Output reserve: 2K (for response) │
│ └── Safety margin: 5K (overflow buffer) │
│ │
│ When approaching limit: │
│ 1. Summarize conversation history (4:1 compression) │
│ 2. Reduce retrieved chunks │
│ 3. Truncate oldest messages │
│ 4. Fail with "context too large" error │
│ │
└────────────────────────────────────────────────────────────┘Implementation
import tiktoken
from dataclasses import dataclass
from typing import List
@dataclass
class BudgetAllocation:
system_prompt: int = 2000
conversation: int = 20000
retrieved_docs: int = 8000
output_reserve: int = 2000
safety_margin: int = 5000
@property
def total_budget(self) -> int:
return (
self.system_prompt +
self.conversation +
self.retrieved_docs +
self.output_reserve +
self.safety_margin
)
class TokenBudgetGuard:
def __init__(
self,
model: str,
context_limit: int,
allocation: BudgetAllocation = None,
):
self.encoding = tiktoken.encoding_for_model(model)
self.context_limit = context_limit
self.allocation = allocation or BudgetAllocation()
def count_tokens(self, text: str) -> int:
return len(self.encoding.encode(text))
def fit_to_budget(
self,
system_prompt: str,
messages: List[dict],
retrieved_docs: List[str],
) -> tuple[str, List[dict], List[str]]:
"""Fit content to token budget, compressing as needed."""
# Count fixed costs
system_tokens = self.count_tokens(system_prompt)
if system_tokens > self.allocation.system_prompt:
raise TokenBudgetError("System prompt exceeds budget")
# Fit messages with sliding window
fitted_messages = self._fit_messages(
messages,
self.allocation.conversation
)
# Fit retrieved docs
fitted_docs = self._fit_docs(
retrieved_docs,
self.allocation.retrieved_docs
)
return system_prompt, fitted_messages, fitted_docs
def _fit_messages(self, messages: List[dict], budget: int) -> List[dict]:
"""Keep most recent messages that fit in budget."""
fitted = []
used = 0
# Always keep system message if present
for msg in reversed(messages):
tokens = self.count_tokens(msg["content"])
if used + tokens <= budget:
fitted.insert(0, msg)
used += tokens
elif msg["role"] == "system":
# Summarize old messages
summary = self._summarize_old_messages(messages[:-len(fitted)])
fitted.insert(0, {"role": "system", "content": summary})
break
return fitted
def _fit_docs(self, docs: List[str], budget: int) -> List[str]:
"""Keep highest-scoring docs that fit in budget."""
fitted = []
used = 0
for doc in docs: # Assume already sorted by relevance
tokens = self.count_tokens(doc)
if used + tokens <= budget:
fitted.append(doc)
used += tokens
else:
break
return fittedPattern 3: Rate Limit Management
from asyncio import Semaphore, sleep
from collections import deque
from time import time
class RateLimiter:
"""Token bucket rate limiter for LLM APIs."""
def __init__(
self,
requests_per_minute: int = 60,
tokens_per_minute: int = 100000,
):
self.rpm_limit = requests_per_minute
self.tpm_limit = tokens_per_minute
self.request_times = deque(maxlen=rpm_limit)
self.token_counts = deque(maxlen=1000)
self.semaphore = Semaphore(rpm_limit)
async def acquire(self, estimated_tokens: int):
"""Wait until rate limit allows the request."""
async with self.semaphore:
now = time()
# Check RPM
while len(self.request_times) >= self.rpm_limit:
oldest = self.request_times[0]
wait_time = 60 - (now - oldest)
if wait_time > 0:
await sleep(wait_time)
now = time()
self.request_times.popleft()
# Check TPM
recent_tokens = sum(
t for t, ts in self.token_counts
if now - ts < 60
)
if recent_tokens + estimated_tokens > self.tpm_limit:
wait_time = 60 - (now - self.token_counts[0][1])
await sleep(wait_time)
# Record this request
self.request_times.append(now)
def record_usage(self, actual_tokens: int):
"""Record actual token usage after request completes."""
self.token_counts.append((actual_tokens, time()))Pattern 4: Cost Control Circuit Breaker
class CostCircuitBreaker:
"""Opens when LLM costs exceed budget."""
def __init__(
self,
hourly_budget: float = 10.0, # $10/hour
daily_budget: float = 100.0, # $100/day
alert_threshold: float = 0.8, # Alert at 80%
):
self.hourly_budget = hourly_budget
self.daily_budget = daily_budget
self.alert_threshold = alert_threshold
self.hourly_spend = 0.0
self.daily_spend = 0.0
self.last_hour_reset = time()
self.last_day_reset = time()
def record_cost(self, input_tokens: int, output_tokens: int, model: str):
"""Record cost and check budget."""
self._reset_if_needed()
cost = self._calculate_cost(input_tokens, output_tokens, model)
self.hourly_spend += cost
self.daily_spend += cost
# Alert at threshold
if self.hourly_spend > self.hourly_budget * self.alert_threshold:
logger.warning(
f"Hourly LLM budget at {self.hourly_spend/self.hourly_budget:.0%}"
)
# Trip circuit at limit
if self.hourly_spend >= self.hourly_budget:
raise CostBudgetExceeded("Hourly LLM budget exceeded")
if self.daily_spend >= self.daily_budget:
raise CostBudgetExceeded("Daily LLM budget exceeded")
def _calculate_cost(self, input_tokens: int, output_tokens: int, model: str) -> float:
"""Calculate cost based on model pricing (Dec 2025)."""
PRICING = {
"claude-sonnet-4-6": {"input": 3.0, "output": 15.0},
"gpt-5.5": {"input": 2.5, "output": 10.0},
"gpt-5-mini": {"input": 0.15, "output": 0.60},
"claude-haiku-4-5-20251001": {"input": 0.80, "output": 4.0},
}
prices = PRICING.get(model, {"input": 1.0, "output": 3.0})
return (
(input_tokens / 1_000_000) * prices["input"] +
(output_tokens / 1_000_000) * prices["output"]
)Pattern 5: Quality-Aware Fallback
class QualityAwareFallback:
"""Falls back when response quality is below threshold."""
def __init__(
self,
primary_chain: FallbackChain,
quality_evaluator: Callable[[str, str], float],
quality_threshold: float = 0.7,
max_retries: int = 2,
):
self.chain = primary_chain
self.evaluate = quality_evaluator
self.threshold = quality_threshold
self.max_retries = max_retries
async def complete(self, prompt: str, **kwargs) -> LLMResponse:
for attempt in range(self.max_retries + 1):
response = await self.chain.complete(prompt, **kwargs)
# Evaluate quality
quality_score = await self.evaluate(prompt, response.content)
if quality_score >= self.threshold:
response.quality_score = quality_score
return response
logger.warning(
f"Response quality {quality_score:.2f} below threshold",
attempt=attempt + 1,
)
# Try with different parameters on retry
if attempt < self.max_retries:
kwargs["temperature"] = max(0.3, kwargs.get("temperature", 0.7) - 0.2)
# Return best effort with warning
response.quality_warning = f"Below threshold: {quality_score:.2f}"
return responseBest Practices (2026)
1. Always have a fallback: Even a cached or default response is better than an error 2. Monitor costs per-request: Track token usage in traces (Langfuse) 3. Use streaming for long responses: Better UX and partial results on failure 4. Cache aggressively: Semantic cache with 0.85+ similarity saves 60-80% costs 5. Set appropriate timeouts: 30s for completion, 5s for embeddings 6. Log all fallback events: Critical for understanding system behavior
OrchestKit Integration
# Example integration for OrchestKit analysis pipeline
llm_chain = FallbackChain(
primary=LLMConfig(
name="primary",
model="claude-sonnet-4-6",
timeout=30.0,
),
fallbacks=[
LLMConfig(
name="fallback",
model="gpt-5-mini",
timeout=20.0,
),
],
cache=semantic_cache, # Redis-backed
default_response=lambda p: "Analysis temporarily unavailable",
)
budget_guard = TokenBudgetGuard(
model="claude-sonnet-4-6",
context_limit=200000,
allocation=BudgetAllocation(
system_prompt=3000,
conversation=10000,
retrieved_docs=15000, # RAG context
output_reserve=4000,
safety_margin=5000,
),
)
rate_limiter = RateLimiter(
requests_per_minute=50, # Leave headroom
tokens_per_minute=80000,
)Related skills
Lark Openapi ExplorerInstantly explore, test, and generate calls against the full Lark (Feishu) OpenAPI surface without leaving their agent workflow.493k
Lark EventConsume real-time events from Lark/Feishu as structured NDJSON streams inside AI agent workflows.387k16kLark Openapi ExplorerWhen an existing Lark/Feishu skill or CLI command cannot fulfill a specific requirement and they need to discover and invoke the exact native OpenAPI endpoint.386k16k
Just ScrapeQuickly search, crawl, extract structured JSON, or monitor web pages without writing custom scraping code.245k41Lark AppsQuery the current visibility and permission scope of a Lark (Feishu) app without writing HTTP client code.235k16k
Design An InterfaceGenerate multiple radically different interface or API designs in parallel before committing to one.188k198k