7. Graceful Degradation

Overview

In this capsule you'll design your application to degrade functionality instead of crashing when a cloud service doesn't respond. S3 being down doesn't mean a dead app — it means an app with reduced functionality. A slow SageMaker endpoint doesn't mean a timeout for the user — it means the app uses a fallback. You'll implement circuit breakers, fallback patterns, retry with backoff, and a health check with degradation levels that communicates exactly what works and what doesn't.

Context: In the previous capsule, feature flags let you enable/disable features per environment. But there's a scenario that feature flags don't cover: a feature is enabled, the service exists, but it fails at runtime. S3 can have a 30-second blip. SageMaker can be scaling. Lambda can be in a cold start. If your app crashes in those moments, you lose requests, users, and trust. Graceful degradation is the pattern that turns "app dead for 30 seconds" into "app with reduced functionality for 30 seconds."

The Problem: Cloud Services Fail

Why your app can't assume everything always works

Reality of cloud services:

S3: 99.99% SLA = ~52 minutes of downtime/year
    → Your app receives ~52 minutes of S3 errors per year

Lambda: 99.95% SLA = ~4.4 hours of downtime/year
    → Includes cold starts, throttling, timeouts

SageMaker: Endpoints scale on demand
    → Latency goes from 100ms to 5000ms during scaling

Network: Blips of 1-30 seconds
    → Intermittent timeouts that aren't an "outage" but transient

LocalStack: Doesn't guarantee an SLA
    → In development, it can restart, run out of memory

The anti-pattern: assuming 100% availability

# ❌ Anti-pattern — if S3 fails, the whole app crashes
def process_request(document: dict) -> dict:
    template = s3.get_object(Bucket=bucket, Key="prompts/v1/system.txt")
    # If S3 doesn't respond → ConnectionError → 500 Internal Server Error
    # The user sees a cryptic error
    # No retry, no fallback, no degradation

The pattern: the app decides what to do when something fails

# ✅ Pattern — if S3 fails, the app degrades
def process_request(document: dict) -> dict:
    try:
        template = get_prompt_with_fallback("summarizer", "v1")
    except ServiceUnavailableError:
        return {
            "status": "degraded",
            "message": "Service processing with reduced capacity",
            "result": process_with_default_template(document),
        }

Pattern 1: Retry with Exponential Backoff

The first line of defense

Many failures are transient: a 500ms network blip, an S3 throttle from a rate limit, a Lambda cold start. A simple retry resolves most of these cases.

"""services/retry.py — Retry with exponential backoff."""

import time
import logging
from typing import TypeVar, Callable
from functools import wraps

logger = logging.getLogger(__name__)
T = TypeVar("T")


class RetryExhaustedError(Exception):
    """All retries failed."""
    def __init__(self, operation: str, attempts: int, last_error: Exception):
        self.operation = operation
        self.attempts = attempts
        self.last_error = last_error
        super().__init__(
            f"{operation} failed after {attempts} attempts: {last_error}"
        )


def retry_with_backoff(
    max_retries: int = 3,
    base_delay: float = 1.0,
    max_delay: float = 30.0,
    retryable_exceptions: tuple = (Exception,),
):
    """Decorator that adds retry with exponential backoff."""

    def decorator(func: Callable[..., T]) -> Callable[..., T]:
        @wraps(func)
        def wrapper(*args, **kwargs) -> T:
            last_exception = None
            for attempt in range(1, max_retries + 1):
                try:
                    result = func(*args, **kwargs)
                    if attempt > 1:
                        logger.info(
                            f"{func.__name__} succeeded on attempt {attempt}"
                        )
                    return result
                except retryable_exceptions as e:
                    last_exception = e
                    if attempt < max_retries:
                        delay = min(base_delay * (2 ** (attempt - 1)), max_delay)
                        logger.warning(
                            f"{func.__name__} failed (attempt {attempt}/{max_retries}): "
                            f"{e}. Retrying in {delay:.1f}s"
                        )
                        time.sleep(delay)

            raise RetryExhaustedError(
                operation=func.__name__,
                attempts=max_retries,
                last_error=last_exception,
            )

        return wrapper
    return decorator

Using the retry

from botocore.exceptions import ClientError, EndpointConnectionError

@retry_with_backoff(
    max_retries=3,
    base_delay=1.0,
    retryable_exceptions=(ClientError, EndpointConnectionError, ConnectionError),
)
def get_prompt_template(s3_client, bucket: str, name: str, version: str) -> str:
    key = f"prompts/{name}/{version}/system.txt"
    response = s3_client.get_object(Bucket=bucket, Key=key)
    return response["Body"].read().decode("utf-8")

Pattern 2: Circuit Breaker

Avoiding retries when the service is clearly down

If S3 has failed 10 times in the last 30 seconds, continuing to retry only adds latency. A circuit breaker "opens the circuit" after N failures and returns the fallback immediately until the service recovers.

"""services/circuit_breaker.py — Circuit breaker for cloud services."""

import time
import logging
from enum import Enum
from typing import Any, Callable

logger = logging.getLogger(__name__)


class CircuitState(str, Enum):
    CLOSED = "closed"       # Normal: requests pass through
    OPEN = "open"           # Open: requests go straight to fallback
    HALF_OPEN = "half_open" # Testing: lets 1 request through to check


class CircuitBreakerError(Exception):
    """Circuit breaker open — service unavailable."""
    def __init__(self, service: str, failures: int, reset_in: float):
        self.service = service
        self.failures = failures
        self.reset_in = reset_in
        super().__init__(
            f"Circuit open for '{service}': {failures} failures, "
            f"reset in {reset_in:.0f}s"
        )


class CircuitBreaker:
    """Circuit breaker for a cloud service.

    States:
    - CLOSED: normal operation, requests pass through to the service
    - OPEN: service down, requests go to fallback immediately
    - HALF_OPEN: lets 1 request through to test if the service recovered
    """

    def __init__(
        self,
        service_name: str,
        failure_threshold: int = 5,
        reset_timeout: float = 60.0,
        half_open_max_calls: int = 1,
    ):
        self.service_name = service_name
        self.failure_threshold = failure_threshold
        self.reset_timeout = reset_timeout
        self.half_open_max_calls = half_open_max_calls

        self._state = CircuitState.CLOSED
        self._failure_count = 0
        self._success_count = 0
        self._last_failure_time: float = 0
        self._half_open_calls = 0

    @property
    def state(self) -> CircuitState:
        if self._state == CircuitState.OPEN:
            elapsed = time.time() - self._last_failure_time
            if elapsed >= self.reset_timeout:
                self._state = CircuitState.HALF_OPEN
                self._half_open_calls = 0
                logger.info(
                    f"[{self.service_name}] Circuit → HALF_OPEN "
                    f"(testing recovery)"
                )
        return self._state

    def execute(self, func: Callable, *args, **kwargs) -> Any:
        """Runs the function through the circuit breaker."""
        current_state = self.state

        if current_state == CircuitState.OPEN:
            raise CircuitBreakerError(
                service=self.service_name,
                failures=self._failure_count,
                reset_in=self.reset_timeout - (time.time() - self._last_failure_time),
            )

        if current_state == CircuitState.HALF_OPEN:
            if self._half_open_calls >= self.half_open_max_calls:
                raise CircuitBreakerError(
                    service=self.service_name,
                    failures=self._failure_count,
                    reset_in=self.reset_timeout,
                )
            self._half_open_calls += 1

        try:
            result = func(*args, **kwargs)
            self._on_success()
            return result
        except Exception as e:
            self._on_failure(e)
            raise

    def _on_success(self):
        if self._state == CircuitState.HALF_OPEN:
            logger.info(
                f"[{self.service_name}] Circuit → CLOSED (service recovered)"
            )
        self._failure_count = 0
        self._state = CircuitState.CLOSED
        self._success_count += 1

    def _on_failure(self, error: Exception):
        self._failure_count += 1
        self._last_failure_time = time.time()

        if self._failure_count >= self.failure_threshold:
            self._state = CircuitState.OPEN
            logger.error(
                f"[{self.service_name}] Circuit → OPEN "
                f"({self._failure_count} failures): {error}"
            )

    def status(self) -> dict:
        return {
            "service": self.service_name,
            "state": self.state.value,
            "failure_count": self._failure_count,
            "success_count": self._success_count,
            "threshold": self.failure_threshold,
            "reset_timeout": self.reset_timeout,
        }

Pattern 3: Fallback Strategies

Fallback strategies for different services

"""services/fallbacks.py — Fallback strategies for cloud services."""

import json
import logging
from typing import Any

logger = logging.getLogger(__name__)


class FallbackStrategy:
    """Fallback strategies for when a service isn't available."""

    def __init__(self, s3_client: Any = None, bucket: str = ""):
        self.s3 = s3_client
        self.bucket = bucket
        self._cache: dict[str, Any] = {}

    def get_prompt_with_fallback(
        self, s3_client, bucket: str, name: str, version: str
    ) -> str:
        """Gets a prompt template with a chain of fallbacks.

        1. S3 (primary source)
        2. In-memory cache (if read before)
        3. Hardcoded default (last resort)
        """
        cache_key = f"prompts/{name}/{version}"

        # Attempt 1: S3
        try:
            key = f"prompts/{name}/{version}/system.txt"
            response = s3_client.get_object(Bucket=bucket, Key=key)
            content = response["Body"].read().decode("utf-8")
            self._cache[cache_key] = content
            return content
        except Exception as e:
            logger.warning(f"S3 fallback activated for {cache_key}: {e}")

        # Attempt 2: in-memory cache
        if cache_key in self._cache:
            logger.info(f"Using cached prompt for {cache_key}")
            return self._cache[cache_key]

        # Attempt 3: default
        logger.warning(f"Using default prompt for {name}")
        return self._get_default_prompt(name)

    def _get_default_prompt(self, name: str) -> str:
        """Hardcoded default prompts as a last resort."""
        defaults = {
            "summarizer": (
                "Generate a concise summary of the following text. "
                "Maximum 3 paragraphs."
            ),
            "classifier": (
                "Classify the following text into one of these categories: "
                "technical, business, general."
            ),
        }
        return defaults.get(name, "Process the following text.")

    def invoke_with_fallback(
        self,
        primary_func,
        fallback_func,
        *args,
        **kwargs,
    ) -> dict:
        """Runs primary_func; if it fails, runs fallback_func."""
        try:
            result = primary_func(*args, **kwargs)
            result["_degraded"] = False
            return result
        except Exception as e:
            logger.warning(f"Primary function failed: {e}. Using fallback.")
            result = fallback_func(*args, **kwargs)
            result["_degraded"] = True
            result["_degradation_reason"] = str(e)
            return result

Pattern 4: Health Check with Degradation Levels

A health check that communicates the exact state

"""services/health.py — Health check with degradation levels."""

import time
import logging
from enum import Enum
from typing import Any
from dataclasses import dataclass, field

logger = logging.getLogger(__name__)


class HealthLevel(str, Enum):
    HEALTHY = "healthy"         # Everything works
    DEGRADED = "degraded"       # Reduced functionality
    CRITICAL = "critical"       # Minimal functionality
    UNAVAILABLE = "unavailable" # App can't operate


@dataclass
class ServiceHealth:
    name: str
    status: str
    latency_ms: float = 0
    error: str | None = None
    critical: bool = False


@dataclass
class AppHealth:
    level: HealthLevel
    services: list[ServiceHealth] = field(default_factory=list)
    message: str = ""
    degraded_features: list[str] = field(default_factory=list)
    available_features: list[str] = field(default_factory=list)


class HealthChecker:
    """Checks the app's health with degradation levels."""

    def __init__(self, factory, settings, feature_flags):
        self.factory = factory
        self.settings = settings
        self.flags = feature_flags

    def check(self) -> AppHealth:
        """Runs a complete health check."""
        services = []

        # S3 — critical (no prompts or documents without S3)
        s3_health = self._check_s3()
        services.append(s3_health)

        # Lambda — critical (no inference without Lambda)
        lambda_health = self._check_lambda()
        services.append(lambda_health)

        # SageMaker — optional (if the feature flag is active)
        if self.flags.is_enabled("sagemaker_enrichment"):
            sm_health = self._check_sagemaker()
            services.append(sm_health)

        # CloudWatch — optional
        if self.flags.is_enabled("cloudwatch_metrics"):
            cw_health = self._check_cloudwatch()
            services.append(cw_health)

        return self._compute_overall(services)

    def _check_s3(self) -> ServiceHealth:
        start = time.time()
        try:
            s3 = self.factory.s3
            s3.head_bucket(Bucket=self.settings.s3_bucket)
            latency = (time.time() - start) * 1000
            return ServiceHealth(
                name="s3", status="healthy", latency_ms=latency, critical=True
            )
        except Exception as e:
            latency = (time.time() - start) * 1000
            return ServiceHealth(
                name="s3", status="unhealthy", latency_ms=latency,
                error=str(e), critical=True
            )

    def _check_lambda(self) -> ServiceHealth:
        start = time.time()
        try:
            lam = self.factory.lambda_client
            lam.list_functions(MaxItems=1)
            latency = (time.time() - start) * 1000
            return ServiceHealth(
                name="lambda", status="healthy", latency_ms=latency, critical=True
            )
        except Exception as e:
            latency = (time.time() - start) * 1000
            return ServiceHealth(
                name="lambda", status="unhealthy", latency_ms=latency,
                error=str(e), critical=True
            )

    def _check_sagemaker(self) -> ServiceHealth:
        start = time.time()
        try:
            sm = self.factory.get_client("sagemaker")
            sm.list_endpoints(MaxResults=1)
            latency = (time.time() - start) * 1000
            return ServiceHealth(
                name="sagemaker", status="healthy", latency_ms=latency
            )
        except Exception as e:
            latency = (time.time() - start) * 1000
            return ServiceHealth(
                name="sagemaker", status="unhealthy", latency_ms=latency,
                error=str(e)
            )

    def _check_cloudwatch(self) -> ServiceHealth:
        start = time.time()
        try:
            cw = self.factory.get_client("cloudwatch")
            cw.list_metrics(Namespace="AIService", Limit=1)
            latency = (time.time() - start) * 1000
            return ServiceHealth(
                name="cloudwatch", status="healthy", latency_ms=latency
            )
        except Exception as e:
            latency = (time.time() - start) * 1000
            return ServiceHealth(
                name="cloudwatch", status="unhealthy", latency_ms=latency,
                error=str(e)
            )

    def _compute_overall(self, services: list[ServiceHealth]) -> AppHealth:
        """Computes the overall health level based on the services."""
        critical_down = [s for s in services if s.critical and s.status == "unhealthy"]
        optional_down = [s for s in services if not s.critical and s.status == "unhealthy"]
        all_healthy = all(s.status == "healthy" for s in services)

        degraded_features = []
        available_features = []

        if all_healthy:
            level = HealthLevel.HEALTHY
            message = "All services operating normally"
            available_features = [s.name for s in services]
        elif critical_down:
            if len(critical_down) == len([s for s in services if s.critical]):
                level = HealthLevel.UNAVAILABLE
                message = f"Critical services unavailable: {[s.name for s in critical_down]}"
            else:
                level = HealthLevel.CRITICAL
                message = f"Some critical services degraded: {[s.name for s in critical_down]}"
            degraded_features = [s.name for s in critical_down + optional_down]
            available_features = [s.name for s in services if s.status == "healthy"]
        elif optional_down:
            level = HealthLevel.DEGRADED
            message = f"Optional services unavailable: {[s.name for s in optional_down]}"
            degraded_features = [s.name for s in optional_down]
            available_features = [s.name for s in services if s.status == "healthy"]
        else:
            level = HealthLevel.HEALTHY
            message = "OK"
            available_features = [s.name for s in services]

        return AppHealth(
            level=level,
            services=services,
            message=message,
            degraded_features=degraded_features,
            available_features=available_features,
        )

Health endpoint with degradation levels

"""handler.py — Health endpoint that reports degradation levels."""

import json
from services.health import HealthChecker, HealthLevel


def health_endpoint(checker: HealthChecker) -> dict:
    """/health endpoint with degradation information."""
    health = checker.check()

    status_codes = {
        HealthLevel.HEALTHY: 200,
        HealthLevel.DEGRADED: 200,
        HealthLevel.CRITICAL: 503,
        HealthLevel.UNAVAILABLE: 503,
    }

    body = {
        "status": health.level.value,
        "message": health.message,
        "available_features": health.available_features,
        "degraded_features": health.degraded_features,
        "services": [
            {
                "name": s.name,
                "status": s.status,
                "latency_ms": round(s.latency_ms, 1),
                "critical": s.critical,
                "error": s.error,
            }
            for s in health.services
        ],
    }

    return {
        "statusCode": status_codes.get(health.level, 500),
        "body": json.dumps(body),
    }

Example degraded response:

{
  "status": "degraded",
  "message": "Optional services unavailable: ['sagemaker']",
  "available_features": ["s3", "lambda", "cloudwatch"],
  "degraded_features": ["sagemaker"],
  "services": [
    {"name": "s3", "status": "healthy", "latency_ms": 12.3, "critical": true},
    {"name": "lambda", "status": "healthy", "latency_ms": 45.1, "critical": true},
    {"name": "sagemaker", "status": "unhealthy", "latency_ms": 5001.2, "critical": false,
     "error": "Endpoint scaling in progress"},
    {"name": "cloudwatch", "status": "healthy", "latency_ms": 23.4, "critical": false}
  ]
}

Putting It All Together: Processor with Graceful Degradation

"""services/resilient_processor.py — Processor with complete degradation."""

import json
import logging
from typing import Any
from services.circuit_breaker import CircuitBreaker, CircuitBreakerError
from services.retry import retry_with_backoff, RetryExhaustedError
from services.fallbacks import FallbackStrategy

logger = logging.getLogger(__name__)


class ResilientDocumentProcessor:
    """Processor that degrades gracefully when services fail."""

    def __init__(
        self,
        s3_client: Any,
        bucket: str,
        feature_flags,
        sagemaker_client: Any = None,
    ):
        self.s3 = s3_client
        self.bucket = bucket
        self.flags = feature_flags
        self.sagemaker = sagemaker_client

        self.s3_breaker = CircuitBreaker("s3", failure_threshold=5, reset_timeout=30)
        self.sm_breaker = CircuitBreaker("sagemaker", failure_threshold=3, reset_timeout=60)

        self.fallbacks = FallbackStrategy()

    def process(self, prompt_name: str, prompt_version: str, document: dict) -> dict:
        """Processes a document with degradation at each step."""
        result = {"degraded": False, "degradation_details": []}

        # Step 1: get the prompt (with fallback)
        try:
            template = self.s3_breaker.execute(
                self._get_prompt, prompt_name, prompt_version
            )
        except (CircuitBreakerError, RetryExhaustedError) as e:
            template = self.fallbacks._get_default_prompt(prompt_name)
            result["degraded"] = True
            result["degradation_details"].append(f"Prompt: using default ({e})")

        # Step 2: store the document (best effort)
        doc_key = None
        try:
            doc_key = self.s3_breaker.execute(
                self._store_document, document
            )
        except (CircuitBreakerError, RetryExhaustedError) as e:
            result["degraded"] = True
            result["degradation_details"].append(f"Storage: document not persisted ({e})")

        # Step 3: SageMaker enrichment (optional)
        enrichment = None
        if self.flags.is_enabled("sagemaker_enrichment") and self.sagemaker:
            try:
                enrichment = self.sm_breaker.execute(
                    self._enrich_sagemaker, document
                )
            except (CircuitBreakerError, RetryExhaustedError) as e:
                result["degradation_details"].append(
                    f"SageMaker: enrichment not available ({e})"
                )

        result.update({
            "prompt_used": f"{prompt_name}/{prompt_version}",
            "template_preview": template[:80],
            "document_stored": doc_key,
            "sagemaker_enrichment": enrichment,
            "circuit_breakers": {
                "s3": self.s3_breaker.status(),
                "sagemaker": self.sm_breaker.status(),
            },
        })

        return result

    @retry_with_backoff(max_retries=2, base_delay=0.5)
    def _get_prompt(self, name: str, version: str) -> str:
        key = f"prompts/{name}/{version}/system.txt"
        response = self.s3.get_object(Bucket=self.bucket, Key=key)
        return response["Body"].read().decode("utf-8")

    @retry_with_backoff(max_retries=2, base_delay=0.5)
    def _store_document(self, document: dict) -> str:
        doc_id = document.get("id", "unknown")
        key = f"documents/inbox/{doc_id}.json"
        self.s3.put_object(
            Bucket=self.bucket,
            Key=key,
            Body=json.dumps(document).encode("utf-8"),
        )
        return key

    @retry_with_backoff(max_retries=1, base_delay=1.0)
    def _enrich_sagemaker(self, document: dict) -> dict:
        response = self.sagemaker.invoke_endpoint(
            EndpointName="document-enrichment",
            ContentType="application/json",
            Body=json.dumps(document).encode("utf-8"),
        )
        return json.loads(response["Body"].read().decode("utf-8"))

Troubleshooting

Problem 1: The circuit breaker opens too fast

5 errors in a burst open the circuit, but the errors were transient.

# Solution: increase the threshold or add a time window
# Instead of "5 total errors" → "5 errors in the last 30 seconds"

class WindowedCircuitBreaker(CircuitBreaker):
    def __init__(self, *args, window_seconds: float = 30.0, **kwargs):
        super().__init__(*args, **kwargs)
        self._failure_times: list[float] = []
        self._window = window_seconds

    def _on_failure(self, error):
        now = time.time()
        self._failure_times.append(now)
        # Only count failures within the window
        self._failure_times = [
            t for t in self._failure_times
            if now - t < self._window
        ]
        self._failure_count = len(self._failure_times)
        # ... rest of the logic

Problem 2: The fallback returns stale data from the cache

The in-memory cache isn't invalidated and the template changed in S3.

# Solution: cache with TTL
import time

class TTLCache:
    def __init__(self, ttl_seconds: int = 300):
        self._cache: dict[str, tuple[Any, float]] = {}
        self._ttl = ttl_seconds

    def get(self, key: str) -> Any | None:
        if key in self._cache:
            value, timestamp = self._cache[key]
            if time.time() - timestamp < self._ttl:
                return value
            del self._cache[key]
        return None

    def set(self, key: str, value: Any):
        self._cache[key] = (value, time.time())

Problem 3: Health check timeout when a service is slow

A SageMaker endpoint taking 10s makes the health check slow.

# Solution: per-service timeout in the health check
import signal

class TimeoutError(Exception):
    pass

def with_timeout(func, timeout_seconds=5):
    def handler(signum, frame):
        raise TimeoutError(f"Timeout after {timeout_seconds}s")
    signal.signal(signal.SIGALRM, handler)
    signal.alarm(timeout_seconds)
    try:
        return func()
    finally:
        signal.alarm(0)

Problem 4: Retry and circuit breaker interfere

Retry retries 3 times, each attempt counts as a failure in the circuit breaker.

# Solution: only count as a failure after retries are exhausted
# The circuit breaker wraps the function with retry:
# circuit_breaker.execute(retry_function)
# NOT: retry(circuit_breaker.execute(function))

# Correct:
result = self.s3_breaker.execute(self._get_prompt_with_retry, name, version)

# Where _get_prompt_with_retry already has @retry_with_backoff

Practical Exercises

Exercise 1: Cache with multi-level fallback

Implement a cache with three levels: memory (L1), local file (L2), and S3 (L3). If L1 fails, try L2. If L2 fails, try L3. Each level caches what it gets from the lower level.

import json
import os
from typing import Any, Optional


class MultiLevelCache:
    """Cache with 3 levels: memory → file → S3."""

    def __init__(self, s3_client=None, bucket: str = "", cache_dir: str = ".cache"):
        self._memory: dict[str, Any] = {}
        self._cache_dir = cache_dir
        self._s3 = s3_client
        self._bucket = bucket
        os.makedirs(cache_dir, exist_ok=True)

    def get(self, key: str) -> Optional[Any]:
        # L1: memory
        if key in self._memory:
            return self._memory[key]

        # L2: local file
        file_path = os.path.join(self._cache_dir, key.replace("/", "_"))
        if os.path.exists(file_path):
            with open(file_path) as f:
                value = json.load(f)
            self._memory[key] = value  # Promote to L1
            return value

        # L3: S3
        if self._s3 and self._bucket:
            try:
                response = self._s3.get_object(
                    Bucket=self._bucket, Key=f"cache/{key}"
                )
                value = json.loads(response["Body"].read().decode("utf-8"))
                self._memory[key] = value  # Promote to L1
                self._save_to_file(key, value)  # Promote to L2
                return value
            except Exception:
                pass

        return None

    def set(self, key: str, value: Any):
        # Write to all levels
        self._memory[key] = value
        self._save_to_file(key, value)
        if self._s3 and self._bucket:
            try:
                self._s3.put_object(
                    Bucket=self._bucket,
                    Key=f"cache/{key}",
                    Body=json.dumps(value).encode("utf-8"),
                )
            except Exception:
                pass

    def _save_to_file(self, key: str, value: Any):
        file_path = os.path.join(self._cache_dir, key.replace("/", "_"))
        with open(file_path, "w") as f:
            json.dump(value, f)

    def stats(self) -> dict:
        file_count = len(os.listdir(self._cache_dir))
        return {
            "l1_memory": len(self._memory),
            "l2_files": file_count,
            "l3_s3": "connected" if self._s3 else "disabled",
        }


cache = MultiLevelCache()
cache.set("prompts/summarizer/v1", {"template": "Summarize in 3 points"})

result = cache.get("prompts/summarizer/v1")
print(f"Result: {result}")
print(f"Stats: {cache.stats()}")

Exercise 2: Circuit breaker with metrics

Extend the CircuitBreaker so it records metrics: total number of calls, success rate, average response times, and the number of times it opened the circuit.

import time
from collections import deque


class MetricCircuitBreaker:
    """Circuit breaker with detailed metrics."""

    def __init__(self, service_name: str, failure_threshold: int = 5,
                 reset_timeout: float = 60.0):
        self.service_name = service_name
        self.failure_threshold = failure_threshold
        self.reset_timeout = reset_timeout
        self._failure_count = 0
        self._state = "closed"
        self._last_failure_time = 0.0

        self._total_calls = 0
        self._total_successes = 0
        self._total_failures = 0
        self._total_circuit_opens = 0
        self._response_times: deque = deque(maxlen=100)

    def execute(self, func, *args, **kwargs):
        if self._state == "open":
            elapsed = time.time() - self._last_failure_time
            if elapsed < self.reset_timeout:
                self._total_calls += 1
                raise RuntimeError(f"Circuit open for {self.service_name}")
            self._state = "half_open"

        self._total_calls += 1
        start = time.time()

        try:
            result = func(*args, **kwargs)
            elapsed_ms = (time.time() - start) * 1000
            self._response_times.append(elapsed_ms)
            self._total_successes += 1
            self._failure_count = 0
            self._state = "closed"
            return result
        except Exception as e:
            elapsed_ms = (time.time() - start) * 1000
            self._response_times.append(elapsed_ms)
            self._total_failures += 1
            self._failure_count += 1
            self._last_failure_time = time.time()

            if self._failure_count >= self.failure_threshold:
                self._state = "open"
                self._total_circuit_opens += 1
            raise

    def metrics(self) -> dict:
        avg_response = (
            sum(self._response_times) / len(self._response_times)
            if self._response_times else 0
        )
        success_rate = (
            self._total_successes / self._total_calls * 100
            if self._total_calls > 0 else 0
        )
        return {
            "service": self.service_name,
            "state": self._state,
            "total_calls": self._total_calls,
            "successes": self._total_successes,
            "failures": self._total_failures,
            "success_rate_pct": round(success_rate, 1),
            "avg_response_ms": round(avg_response, 1),
            "circuit_opens": self._total_circuit_opens,
            "current_failures": self._failure_count,
        }


# Demo
cb = MetricCircuitBreaker("s3", failure_threshold=3)

for i in range(10):
    try:
        cb.execute(lambda: "ok" if i % 4 != 0 else (_ for _ in ()).throw(Exception("fail")))
    except Exception:
        pass

print(json.dumps(cb.metrics(), indent=2))

Exercise 3: Degradation response builder

Create a builder that constructs appropriate HTTP responses based on the degradation level: 200 with a complete result, 200 with a partial result and a degradation header, or 503 with a service-unavailable message.

import json
from dataclasses import dataclass, field


@dataclass
class DegradationInfo:
    level: str  # "none", "partial", "severe", "unavailable"
    affected_features: list[str] = field(default_factory=list)
    fallbacks_used: list[str] = field(default_factory=list)


class DegradedResponseBuilder:
    """Builds appropriate HTTP responses based on the degradation level."""

    def build(self, result: dict, degradation: DegradationInfo) -> dict:
        if degradation.level == "none":
            return self._healthy_response(result)
        elif degradation.level == "partial":
            return self._partial_response(result, degradation)
        elif degradation.level == "severe":
            return self._severe_response(result, degradation)
        else:
            return self._unavailable_response(degradation)

    def _healthy_response(self, result: dict) -> dict:
        return {
            "statusCode": 200,
            "headers": {"X-Degradation": "none"},
            "body": json.dumps({"status": "ok", "data": result}),
        }

    def _partial_response(self, result: dict, deg: DegradationInfo) -> dict:
        return {
            "statusCode": 200,
            "headers": {
                "X-Degradation": "partial",
                "X-Degraded-Features": ",".join(deg.affected_features),
            },
            "body": json.dumps({
                "status": "partial",
                "data": result,
                "degradation": {
                    "level": "partial",
                    "message": "Some features unavailable",
                    "affected": deg.affected_features,
                    "fallbacks": deg.fallbacks_used,
                },
            }),
        }

    def _severe_response(self, result: dict, deg: DegradationInfo) -> dict:
        return {
            "statusCode": 200,
            "headers": {"X-Degradation": "severe"},
            "body": json.dumps({
                "status": "degraded",
                "data": result,
                "degradation": {
                    "level": "severe",
                    "message": "Significantly reduced functionality",
                    "affected": deg.affected_features,
                    "fallbacks": deg.fallbacks_used,
                },
            }),
        }

    def _unavailable_response(self, deg: DegradationInfo) -> dict:
        return {
            "statusCode": 503,
            "headers": {
                "X-Degradation": "unavailable",
                "Retry-After": "30",
            },
            "body": json.dumps({
                "status": "unavailable",
                "message": "Service temporarily unavailable",
                "affected": deg.affected_features,
            }),
        }


# Demo
builder = DegradedResponseBuilder()

# Healthy response
resp = builder.build({"answer": "42"}, DegradationInfo(level="none"))
print(f"Healthy: {resp['statusCode']}")

# Partial response
resp = builder.build(
    {"answer": "42"},
    DegradationInfo(
        level="partial",
        affected_features=["sagemaker"],
        fallbacks_used=["default_enrichment"],
    ),
)
print(f"Partial: {resp['statusCode']}, headers: {resp['headers']}")

# Unavailable
resp = builder.build(
    {},
    DegradationInfo(level="unavailable", affected_features=["s3", "lambda"]),
)
print(f"Unavailable: {resp['statusCode']}")

Exercise 4: Degradation simulation test

Write a test that simulates S3 going down and verifies that the ResilientDocumentProcessor returns a degraded response instead of crashing.

from unittest.mock import MagicMock, patch
from botocore.exceptions import ClientError
from services.feature_flags import FeatureFlags
from config.settings import Settings


def test_processor_degrades_when_s3_fails():
    """Verifies that the processor degrades when S3 doesn't respond."""
    mock_s3 = MagicMock()
    mock_s3.get_object.side_effect = ClientError(
        {"Error": {"Code": "500", "Message": "Internal Server Error"}},
        "GetObject",
    )
    mock_s3.put_object.side_effect = ClientError(
        {"Error": {"Code": "500", "Message": "Internal Server Error"}},
        "PutObject",
    )

    settings = Settings(environment="local")
    flags = FeatureFlags(settings)

    from services.resilient_processor import ResilientDocumentProcessor
    processor = ResilientDocumentProcessor(
        s3_client=mock_s3,
        bucket="test-bucket",
        feature_flags=flags,
    )

    result = processor.process(
        prompt_name="summarizer",
        prompt_version="v1",
        document={"id": "test-doc", "content": "Test content"},
    )

    assert result["degraded"] is True
    assert len(result["degradation_details"]) > 0
    assert "Prompt: using default" in result["degradation_details"][0]
    assert result["template_preview"] is not None
    print(f"✅ Processor degraded correctly: {result['degradation_details']}")


def test_processor_recovers_after_circuit_reset():
    """Verifies that the processor recovers when S3 comes back."""
    call_count = 0
    fail_until = 3

    def conditional_get(*args, **kwargs):
        nonlocal call_count
        call_count += 1
        if call_count <= fail_until:
            raise ClientError(
                {"Error": {"Code": "500", "Message": "Temporary failure"}},
                "GetObject",
            )
        body = MagicMock()
        body.read.return_value = b"Recovered prompt template"
        return {"Body": body}

    mock_s3 = MagicMock()
    mock_s3.get_object.side_effect = conditional_get
    mock_s3.put_object.return_value = {}

    settings = Settings(environment="local")
    flags = FeatureFlags(settings)

    from services.resilient_processor import ResilientDocumentProcessor
    processor = ResilientDocumentProcessor(
        s3_client=mock_s3,
        bucket="test-bucket",
        feature_flags=flags,
    )

    # Force an immediate circuit breaker reset for the test
    processor.s3_breaker.reset_timeout = 0.1

    import time
    time.sleep(0.2)  # Wait for reset

    call_count = fail_until  # The failures have already passed
    result = processor.process("summarizer", "v1", {"id": "test"})

    print(f"✅ Recovery result: degraded={result['degraded']}")


test_processor_degrades_when_s3_fails()

Summary

• Graceful degradation turns "dead app" into "app with reduced functionality." S3 being down doesn't mean a 500 error — it means the app uses default prompts from the cache.
• Retry with backoff is the first line of defense. Most cloud failures are transient (500ms-5s). A retry resolves them without the user noticing anything.
• The circuit breaker prevents failure cascades. If S3 is down, continuing to retry only adds latency. The circuit breaker returns the fallback immediately.
• The health check communicates the exact state. HEALTHY → DEGRADED → CRITICAL → UNAVAILABLE. Each level says what works and what doesn't.
• Fallback strategies define what to do when something fails. S3 → in-memory cache → hardcoded default. Each level is worse, but none is a crash.
• In the next capsule, you integrate everything into a Migration-Ready AI App — the module project.

Additional Resources

1. Microsoft — Circuit Breaker Pattern — Pattern reference
2. AWS Well-Architected — Reliability — Reliability principles
3. Martin Fowler — Circuit Breaker — Original article
4. Retry Pattern — Azure — Retry pattern with backoff
5. Hystrix Wiki — Netflix Hystrix (circuit breaker reference)
6. Graceful Degradation vs Progressive Enhancement — General concept