7. Project: Reliability Layer

Description

In the previous capsules you built each reliability pattern in isolation. In this project you'll integrate them into a complete Reliability Layer that connects with the Module 6 app. The DI you implemented in M6 makes this work almost transparent to the domain: you only change dependencies.py to add the reliability layer. Your domain service (analyze_sentiment) won't know it has retry, circuit breaker, or fallback — and that's the sign that the architecture is well designed.

Project structure

src/
├── infrastructure/
│   ├── llm_provider.py            # LLMProvider Protocol (from M6)
│   ├── openai_provider.py         # OpenAI implementation (from M6)
│   ├── mock_provider.py           # For tests (from M6)
│   │
│   │   # ─── Reliability Layer (new in M7) ───
│   ├── error_classifier.py        # Classifies errors: TRANSIENT, AUTH, etc.
│   ├── retry_provider.py          # Wrapper with tenacity
│   ├── circuit_breaker.py         # CircuitBreaker state machine
│   ├── circuit_breaker_provider.py # CircuitBreaker wrapper
│   ├── rate_limiter.py            # TokenBucket
│   ├── rate_limited_provider.py   # TokenBucket wrapper
│   ├── fallback_provider.py       # Fallback chain
│   └── cache_provider.py         # Cache as a fallback
│
├── health/
│   ├── __init__.py
│   └── checks.py                  # /health/live, /ready, /deps
│
├── app/
│   ├── dependencies.py            # Everything is assembled here (the only file that changes)
│   ├── main.py
│   └── routers/
│       └── sentiment.py
│
tests/
├── conftest.py
└── unit/
    ├── test_retry_provider.py
    ├── test_circuit_breaker.py
    ├── test_rate_limiter.py
    ├── test_fallback_provider.py
    └── test_reliability_integration.py  # Tests of the complete composition

Step 1: Assemble the reliability layer in dependencies.py

# src/app/dependencies.py
from functools import lru_cache
from src.config import get_settings
from src.infrastructure.llm_provider import LLMProvider
from src.infrastructure.openai_provider import OpenAIProvider
from src.infrastructure.mock_provider import MockProvider
from src.infrastructure.retry_provider import RetryProvider
from src.infrastructure.circuit_breaker import CircuitBreaker
from src.infrastructure.circuit_breaker_provider import CircuitBreakerProvider
from src.infrastructure.rate_limited_provider import RateLimitedProvider
from src.infrastructure.fallback_provider import FallbackProvider
import structlog

log = structlog.get_logger()

# Global instances of components that need state
# (the circuit breaker and rate limiter must be singletons to work correctly)
_primary_circuit_breaker: CircuitBreaker = None
_fallback_circuit_breaker: CircuitBreaker = None
_rate_limiter_metrics: dict = {}

def _get_primary_circuit_breaker() -> CircuitBreaker:
    global _primary_circuit_breaker
    if _primary_circuit_breaker is None:
        settings = get_settings()
        _primary_circuit_breaker = CircuitBreaker(
            name="openai_primary",
            failure_threshold=settings.circuit_breaker_threshold,
            recovery_timeout=settings.circuit_breaker_timeout,
        )
    return _primary_circuit_breaker

def _get_fallback_circuit_breaker() -> CircuitBreaker:
    global _fallback_circuit_breaker
    if _fallback_circuit_breaker is None:
        _fallback_circuit_breaker = CircuitBreaker(
            name="openai_fallback",
            failure_threshold=10,  # Fallback has a higher threshold (more tolerant)
            recovery_timeout=30,
        )
    return _fallback_circuit_breaker

def build_llm_provider(settings=None) -> LLMProvider:
    """
    Build the LLM provider with the complete reliability layer.
    
    The resulting chain (from outer to inner):
    FallbackProvider
      → RateLimitedProvider (primary)
        → CircuitBreakerProvider (primary)
          → RetryProvider (primary)
            → OpenAIProvider("gpt-4o")    ← the one that makes the real call
    
    If the primary fails on all retries:
      → FallbackProvider tries the secondary
        → CircuitBreakerProvider (fallback)
          → RetryProvider (fallback)
            → OpenAIProvider("gpt-4o-mini")  ← cheaper, fallback
    
    If the secondary also fails:
      → FallbackProvider returns the static_fallback
    """
    if settings is None:
        settings = get_settings()
    
    # ─── Mock mode (for tests and development without an API key) ───
    if settings.use_mock_provider:
        log.info("using_mock_provider")
        return MockProvider(
            response='{"sentiment": "positive", "score": 0.8, "confidence": 0.9}'
        )
    
    # ─── Build the primary provider ───
    primary_base = OpenAIProvider(
        client=settings.create_openai_client(),
        model=settings.openai_model,
        temperature=settings.temperature,
        max_tokens=settings.max_tokens,
    )
    
    primary_with_retry = RetryProvider(
        inner=primary_base,
        max_attempts=settings.max_retry_attempts,
        min_wait_seconds=settings.retry_min_wait,
        max_wait_seconds=settings.retry_max_wait,
    )
    
    primary_with_cb = CircuitBreakerProvider(
        inner=primary_with_retry,
        circuit_breaker=_get_primary_circuit_breaker()
    )
    
    primary_rate_limited = RateLimitedProvider(
        inner=primary_with_cb,
        requests_per_minute=int(settings.max_requests_per_minute * 0.8),
        max_wait_seconds=30.0,
        model=settings.openai_model
    )
    
    # ─── Build the fallback provider (smaller model) ───
    fallback_base = OpenAIProvider(
        client=settings.create_openai_client(),
        model="gpt-4o-mini",  # Always the cheapest as fallback
        temperature=settings.temperature,
        max_tokens=settings.max_tokens,
    )
    
    fallback_with_retry = RetryProvider(
        inner=fallback_base,
        max_attempts=2,  # Fewer retries for the fallback
        min_wait_seconds=1.0,
        max_wait_seconds=10.0,
    )
    
    fallback_with_cb = CircuitBreakerProvider(
        inner=fallback_with_retry,
        circuit_breaker=_get_fallback_circuit_breaker()
    )
    
    # ─── Complete fallback chain ───
    combined = FallbackProvider(
        providers=[primary_rate_limited, fallback_with_cb],
        names=["primary_gpt4o", "fallback_gpt4o_mini"],
        static_fallback='{"sentiment": "unknown", "score": 0.0, "confidence": 0.0}',
    )
    
    log.info(
        "reliability_layer_initialized",
        primary_model=settings.openai_model,
        fallback_model="gpt-4o-mini",
        max_retries=settings.max_retry_attempts,
        circuit_threshold=settings.circuit_breaker_threshold
    )
    
    return combined

def get_llm_provider() -> LLMProvider:
    """FastAPI dependency to inject the LLM provider into the endpoints."""
    return build_llm_provider()

def get_circuit_breaker_metrics() -> dict:
    """Returns metrics of the circuit breakers (for health checks)."""
    metrics = {}
    if _primary_circuit_breaker:
        metrics["primary"] = _primary_circuit_breaker.get_metrics()
    if _fallback_circuit_breaker:
        metrics["fallback"] = _fallback_circuit_breaker.get_metrics()
    return metrics

def get_rate_limiter_metrics() -> dict:
    """Returns metrics of the rate limiter (for health checks)."""
    return _rate_limiter_metrics

Step 2: Configuration in Settings

# src/config.py (add to the M6 Settings class)
from pydantic_settings import BaseSettings
from pydantic import Field

class Settings(BaseSettings):
    # ... all the M6 fields ...
    
    # ─── Reliability: Retry ───
    max_retry_attempts: int = Field(
        default=4,
        description="Maximum number of attempts (1 original + 3 retries)"
    )
    retry_min_wait: float = Field(
        default=1.0,
        description="Minimum wait time between retries (seconds)"
    )
    retry_max_wait: float = Field(
        default=30.0,
        description="Maximum wait time between retries (seconds)"
    )
    
    # ─── Reliability: Circuit Breaker ───
    circuit_breaker_threshold: int = Field(
        default=5,
        description="Consecutive failures to open the circuit"
    )
    circuit_breaker_timeout: int = Field(
        default=60,
        description="Seconds in OPEN before transitioning to HALF_OPEN"
    )
    
    # ─── Reliability: Rate Limiting ───
    max_requests_per_minute: int = Field(
        default=60,
        description="Requests per minute allowed (client-side limit)"
    )
    rate_limit_max_wait: float = Field(
        default=30.0,
        description="Maximum seconds to wait in the rate limiter queue"
    )

Step 3: Integrate health checks in main.py

# src/app/main.py
from fastapi import FastAPI
from src.health.checks import router as health_router
from src.app.routers.sentiment import router as sentiment_router
from src.logging_config import configure_logging
from src.startup import run_startup_checks

def create_app() -> FastAPI:
    settings = get_settings()
    
    configure_logging(env=settings.environment)
    
    app = FastAPI(
        title="Sentiment Analysis API",
        description="API with complete reliability layer",
        version="2.0.0"
    )
    
    # Register middleware (from M6)
    from src.middleware import RequestTracingMiddleware
    app.add_middleware(RequestTracingMiddleware)
    
    # Health checks (without authentication)
    app.include_router(health_router)
    
    # API routes
    app.include_router(sentiment_router, prefix="/api/v1")
    
    # Startup checks
    @app.on_event("startup")
    async def startup():
        run_startup_checks()
    
    return app

app = create_app()

Step 4: Reliability layer tests

# tests/unit/test_reliability_integration.py
"""
Integration tests of the complete reliability layer.
Verify that the patterns compose correctly.
"""
import pytest
import time
from src.infrastructure.mock_provider import MockProvider
from src.infrastructure.retry_provider import RetryProvider
from src.infrastructure.circuit_breaker import CircuitBreaker, CircuitState
from src.infrastructure.circuit_breaker_provider import CircuitBreakerProvider
from src.infrastructure.rate_limited_provider import RateLimitedProvider
from src.infrastructure.fallback_provider import FallbackProvider
from src.infrastructure.llm_provider import LLMProviderError
from src.infrastructure.error_classifier import ErrorCategory

# ─── Helper to create transient errors ───
def transient_error():
    return LLMProviderError(
        "Rate limit", category=ErrorCategory.TRANSIENT, should_retry=True
    )

def permanent_error():
    return LLMProviderError(
        "Auth failed", category=ErrorCategory.AUTH_ERROR, should_retry=False
    )

class TestRetryPlusFallback:
    """Verifies that retry and fallback work together correctly."""
    
    def test_primary_succeeds_no_fallback_needed(self):
        """If primary works, fallback never triggers."""
        primary = MockProvider('{"sentiment": "positive", "score": 0.8, "confidence": 0.9}')
        secondary = MockProvider('{"sentiment": "negative", "score": 0.2, "confidence": 0.7}')
        
        provider = FallbackProvider(
            providers=[primary, secondary],
            names=["primary", "secondary"]
        )
        
        result = provider.complete([{"role": "user", "content": "test"}])
        
        assert "positive" in result
        assert primary.call_count == 1
        assert secondary.call_count == 0
    
    def test_primary_fails_fallback_activates(self):
        """If primary fails (after retries), fallback must activate."""
        fail_count = [0]
        
        def primary_complete(messages, **kwargs):
            fail_count[0] += 1
            raise LLMProviderError(
                "Outage", category=ErrorCategory.OUTAGE, should_retry=True
            )
        
        primary_mock = MockProvider()
        primary_mock.complete = primary_complete
        
        primary_with_retry = RetryProvider(
            primary_mock,
            max_attempts=2,
            min_wait_seconds=0.01,
            max_wait_seconds=0.05
        )
        
        secondary = MockProvider('{"sentiment": "neutral", "score": 0.5, "confidence": 0.6}')
        
        provider = FallbackProvider(
            providers=[primary_with_retry, secondary],
            names=["primary", "secondary"],
            static_fallback='{"sentiment": "unknown"}'
        )
        
        result = provider.complete([{"role": "user", "content": "test"}])
        
        # Primary was tried 2 times (max_attempts), then it went to the secondary
        assert fail_count[0] == 2
        assert "neutral" in result
        assert secondary.call_count == 1

class TestCircuitBreakerPlusRetry:
    """Verifies that circuit breaker and retry interact correctly."""
    
    def test_circuit_opens_after_persistent_failures(self):
        """
        When primary fails repeatedly (with retries),
        the circuit breaker must open.
        """
        circuit = CircuitBreaker("test", failure_threshold=3, recovery_timeout=60)
        
        call_count = [0]
        def always_fail(messages, **kwargs):
            call_count[0] += 1
            raise LLMProviderError("Timeout", category=ErrorCategory.TRANSIENT, should_retry=True)
        
        mock = MockProvider()
        mock.complete = always_fail
        
        with_retry = RetryProvider(mock, max_attempts=2, min_wait_seconds=0.01, max_wait_seconds=0.05)
        with_cb = CircuitBreakerProvider(with_retry, circuit_breaker=circuit)
        
        # 3 complete requests (each one makes 2 internal attempts)
        for _ in range(3):
            with pytest.raises(LLMProviderError):
                with_cb.complete([{"role": "user", "content": "test"}])
        
        # The circuit must be open
        assert circuit.state == CircuitState.OPEN
        
        # The next call must fail immediately without calling the inner provider
        call_count_before = call_count[0]
        with pytest.raises(LLMProviderError) as exc_info:
            with_cb.complete([{"role": "user", "content": "test"}])
        
        # No new calls were made to the provider
        assert call_count[0] == call_count_before

class TestRateLimiterIntegration:
    """Verifies that the rate limiter works within the chain."""
    
    def test_rate_limited_provider_throttles(self):
        """The rate limiter must reject when the bucket is empty."""
        mock = MockProvider('{"sentiment": "positive", "score": 0.8, "confidence": 0.9}')
        
        # Very low rate to test easily
        rate_limited = RateLimitedProvider(
            inner=mock,
            requests_per_minute=6,   # 0.1 requests/second
            burst_size=2,            # Burst of only 2
            max_wait_seconds=0.1     # Don't wait long in tests
        )
        
        # The first 2 must pass (the burst)
        for _ in range(2):
            result = rate_limited.complete([{"role": "user", "content": "test"}])
            assert result is not None
        
        # The third one may fail or wait (depends on timing)
        # In tests with 0.1s of max_wait, it's eventually rejected
        # Verify that the bucket can replenish

class TestFullStack:
    """Test of the complete stack: Rate → Circuit → Retry → Fallback."""
    
    def test_all_components_compose_correctly(self):
        """
        End-to-end test of the complete composition.
        Primary fails, secondary works, a result is returned.
        """
        secondary = MockProvider('{"sentiment": "neutral", "score": 0.5, "confidence": 0.6}')
        
        def primary_fail(messages, **kwargs):
            raise LLMProviderError("Primary down", category=ErrorCategory.OUTAGE, should_retry=False)
        
        primary_mock = MockProvider()
        primary_mock.complete = primary_fail
        
        # Build the complete chain
        primary_circuit = CircuitBreaker("primary", failure_threshold=2, recovery_timeout=60)
        primary_with_cb = CircuitBreakerProvider(primary_mock, circuit_breaker=primary_circuit)
        primary_with_retry = RetryProvider(primary_with_cb, max_attempts=2, min_wait_seconds=0.01, max_wait_seconds=0.05)
        
        full_provider = FallbackProvider(
            providers=[primary_with_retry, secondary],
            names=["primary", "secondary"],
            static_fallback='{"sentiment": "unknown"}'
        )
        
        result = full_provider.complete([{"role": "user", "content": "test text"}])
        assert "neutral" in result
        
        # Verify degradation metrics
        metrics = full_provider.get_metrics()
        assert metrics["degraded_calls"] == 1
        assert metrics["fallback_counts"]["secondary"] == 1

Step 5: Test the health checks

# tests/unit/test_health.py
import pytest
from fastapi.testclient import TestClient
from unittest.mock import patch, MagicMock

def test_liveness_always_returns_200(client: TestClient):
    """The liveness endpoint must always return 200."""
    response = client.get("/health/live")
    assert response.status_code == 200
    assert response.json()["status"] == "alive"

def test_readiness_with_closed_circuit(client: TestClient):
    """Readiness must return 200 when the circuit is closed."""
    with patch("src.app.dependencies.get_circuit_breaker_metrics") as mock_metrics:
        mock_metrics.return_value = {
            "primary": {"state": "closed"},
            "fallback": {"state": "closed"}
        }
        response = client.get("/health/ready")
        assert response.status_code == 200

def test_readiness_with_open_circuit(client: TestClient):
    """Readiness must return 503 when the circuit is open."""
    with patch("src.app.dependencies.get_circuit_breaker_metrics") as mock_metrics:
        mock_metrics.return_value = {
            "primary": {"state": "open"},
            "fallback": {"state": "closed"}
        }
        response = client.get("/health/ready")
        assert response.status_code == 503

def test_dependency_check_openai_ok(client: TestClient):
    """deps check must report OK when OpenAI responds."""
    with patch("src.config.get_settings") as mock_settings:
        mock_client = MagicMock()
        mock_client.models.list.return_value = MagicMock(data=[1, 2, 3])
        mock_settings.return_value.create_openai_client.return_value = mock_client
        
        response = client.get("/health/deps")
        assert response.status_code == 200
        assert response.json()["checks"]["openai"]["status"] == "ok"

def test_dependency_check_openai_down(client: TestClient):
    """deps check must report error when OpenAI doesn't respond."""
    with patch("src.config.get_settings") as mock_settings:
        mock_client = MagicMock()
        mock_client.models.list.side_effect = ConnectionError("timeout")
        mock_settings.return_value.create_openai_client.return_value = mock_client
        
        response = client.get("/health/deps")
        assert response.status_code == 503
        assert response.json()["checks"]["openai"]["status"] == "error"

Comparison: before vs after

Troubleshooting the composition

Problem: The order of the wrappers seems confusing. Solution: Remember the flow from inside out. The one "closest" to the real provider handles first. A call travels like this:

FallbackProvider → RateLimitedProvider → CircuitBreakerProvider → RetryProvider → OpenAIProvider
(outermost)      (speed control)         (detects outage)         (retries)       (the real call)

Problem: The circuit breakers and rate limiters must be singletons, but I'm creating the provider on every FastAPI request. Solution: Extract the stateful components to global variables or a dependency container. build_llm_provider() already does this — the circuit breakers are singletons outside the function.

Problem: Tests fail because of the retry wait times. Solution: In tests, use min_wait_seconds=0.01, max_wait_seconds=0.05 so the retries are almost instantaneous. In production the real Settings values are used.

Exercises

Exercise 1: Add Anthropic as secondary

The current project uses gpt-4o-mini as fallback. Modify dependencies.py to use a hypothetical AnthropicProvider as the secondary provider in the fallback chain.

from src.infrastructure.anthropic_provider import AnthropicProvider  # hypothetical

# In build_llm_provider():
secondary_base = AnthropicProvider(
    model="claude-haiku-3",
    api_key=settings.anthropic_api_key
)
secondary_with_retry = RetryProvider(secondary_base, max_attempts=2, ...)
secondary_with_cb = CircuitBreakerProvider(secondary_with_retry, ...)

combined = FallbackProvider(
    providers=[primary_rate_limited, secondary_with_cb],
    names=["openai_primary", "anthropic_secondary"],
    static_fallback=...
)

Exercise 2: Metrics in /health/deps

Add the reliability layer metrics to the /health/deps endpoint: circuit breaker state and rate limiter utilization.

# In checks.py:
from src.app.dependencies import get_circuit_breaker_metrics, get_rate_limiter_metrics

@router.get("/deps")
async def dependency_check():
    checks = {}
    
    # ... OpenAI verification ...
    
    checks["circuit_breakers"] = get_circuit_breaker_metrics()
    checks["rate_limits"] = get_rate_limiter_metrics()
    
    return {"checks": checks}

Exercise 3: Simulate a complete outage

Write an integration test that simulates a complete outage: the primary and the secondary fail, and verify that the system returns the static fallback with the degraded=True flag:

def test_full_outage_returns_static_fallback():
    error = LLMProviderError(
        "Service unavailable",
        category=ErrorCategory.OUTAGE,
        should_retry=False
    )
    
    primary = MockProvider()
    primary.complete = lambda m, **k: (_ for _ in ()).throw(error)
    
    secondary = MockProvider()
    secondary.complete = lambda m, **k: (_ for _ in ()).throw(error)
    
    fallback = FallbackProvider(
        providers=[primary, secondary],
        names=["primary", "secondary"],
        static_fallback='{"sentiment": "unknown", "score": 0.0, "confidence": 0.0}'
    )
    
    result = fallback.complete([{"role": "user", "content": "test"}])
    
    data = json.loads(result)
    assert data["sentiment"] == "unknown"
    assert fallback.is_degraded is True

Exercise 4: Verify correct composition

Write a test that creates the complete chain FallbackProvider(RateLimited(CircuitBreaker(Retry(Mock)))) and verifies that a successful call travels through all the layers without errors:

def test_full_chain_success():
    base = MockProvider('{"sentiment": "positive", "score": 0.9, "confidence": 0.95}')
    
    with_retry = RetryProvider(base, max_attempts=3,
                              min_wait_seconds=0.01, max_wait_seconds=0.01)
    with_cb = CircuitBreakerProvider(with_retry, failure_threshold=5)
    with_rate = RateLimitedProvider(with_cb, requests_per_minute=60)
    combined = FallbackProvider(
        providers=[with_rate],
        names=["primary"],
        static_fallback='{"sentiment": "unknown"}'
    )
    
    result = combined.complete([{"role": "user", "content": "test"}])
    
    assert "positive" in result
    assert base.call_count == 1
    assert not combined.is_degraded

Troubleshooting

"The circuit breaker is created anew on every request"

This nullifies its purpose — it needs to be a singleton to accumulate failures. Verify that in dependencies.py the circuit breakers are created outside the build_llm_provider() function or cached with a pattern like _get_primary_circuit_breaker() that always returns the same instance.

"The integration tests are slow because of the retries"

Always use min_wait_seconds=0.01, max_wait_seconds=0.05 in your test providers. If you don't parameterize these values, the retries use the production defaults (1-30s) and your test suite takes minutes.

"FallbackProvider doesn't detect that the primary failed"

The FallbackProvider needs the primary to raise an exception — not to return an error as a string. If your OpenAIProvider catches the exception and returns None or an error string, the fallback doesn't activate. Make sure errors propagate as exceptions all the way up to the FallbackProvider.

"The /health/deps metrics don't show the circuit breakers"

Verify that get_circuit_breaker_metrics() accesses the same instances that build_llm_provider() uses. If they're different instances, the metrics will be empty. Both functions must reference the same singletons.

"On deploy, the CircuitBreaker resets and I lose the history"

This is expected — the CircuitBreaker lives in memory. Each deploy creates new instances with counters at zero. For most AI apps, this is fine: you prefer to start "clean" rather than inherit an open circuit from the previous version. If you need to persist state across deploys, you can use Redis to store the circuit's state, but it adds complexity and rarely pays off.

"I don't know which provider is responding right now"

Add logging in the FallbackProvider to record which provider served each request. You can add a provider_used field to the log:

log.info("request_served",
    provider_used=provider_name,
    is_primary=(i == 0),
    request_id=request_id
)

This lets you filter in your logs how many requests use the primary vs fallback, which is a key system health metric.

Integration checklist

Before considering this project finished, verify each item:

- [ ] `dependencies.py` creates the complete chain of providers
- [ ] CircuitBreaker is a singleton (not recreated per request)
- [ ] Tests pass with `min_wait_seconds=0.01` (no real delays)
- [ ] Complete chain test: primary OK → normal response
- [ ] Fallback test: primary fails → secondary responds
- [ ] Total outage test: all fail → static fallback
- [ ] Health /live returns 200 without checking OpenAI
- [ ] Health /ready checks circuit breakers and config
- [ ] Health /deps shows CB and rate limiter metrics
- [ ] Logs show retry_attempt, circuit_opened, fallback_used
- [ ] Config has all the reliability parameters
- [ ] Settings validates production constraints

Summary

• M6's DI is the enabler: adding reliability only requires changing dependencies.py
• Composition: FallbackProvider(RateLimited(CircuitBreaker(Retry(OpenAI))))
• Stateful singletons: circuit breakers and rate limiters must live outside the request lifecycle
• Fast tests: use min_wait_seconds=0.01 in tests to avoid slow sleeps
• Health checks as a contract with Kubernetes: well-implemented liveness/readiness make the app cloud-native
• The complete project demonstrates that you can add resilience to an AI app without modifying the domain — that's the sign that your architecture is well designed

Additional resources

1. tenacity Documentation — Retry library
2. FastAPI Dependencies — DI system
3. Kubernetes Probes — Probes configuration
4. Release It! (Nygard) — The reference book