7. Graceful Degradation

Descripción

En esta cápsula vas a diseñar tu aplicación para que degrade funcionalidad en lugar de crashear cuando un servicio cloud no responde. S3 caído no significa app muerta — significa app con funcionalidad reducida. Un endpoint de SageMaker lento no significa timeout para el usuario — significa que la app usa un fallback. Vas a implementar circuit breakers, fallback patterns, retry con backoff, y un health check con niveles de degradación que comunica exactamente qué funciona y qué no.

Contexto: En la cápsula anterior, feature flags te permiten habilitar/deshabilitar features por entorno. Pero hay un escenario que feature flags no cubren: una feature está habilitada, el servicio existe, pero falla en runtime. S3 puede tener un blip de 30 segundos. SageMaker puede estar escalando. Lambda puede estar en cold start. Si tu app crashea en esos momentos, pierdes requests, usuarios, y confianza. Graceful degradation es el patrón que convierte "app muerta por 30 segundos" en "app con funcionalidad reducida por 30 segundos."

El Problema: Servicios Cloud Fallan

Por qué tu app no puede asumir que todo funciona siempre

Realidad de servicios cloud:

S3: 99.99% SLA = ~52 minutos de downtime/año
    → Tu app recibe ~52 minutos de errores S3 por año

Lambda: 99.95% SLA = ~4.4 horas de downtime/año
    → Incluye cold starts, throttling, timeouts

SageMaker: Endpoints escalan bajo demanda
    → Latencia pasa de 100ms a 5000ms durante escalado

Network: Blips de 1-30 segundos
    → Timeouts intermitentes que no son "caída" sino transitorios

LocalStack: No garantiza SLA
    → En desarrollo, puede reiniciarse, quedarse sin memoria

El anti-patrón: asumir disponibilidad al 100%

# ❌ Anti-patrón — si S3 falla, toda la app crashea
def process_request(document: dict) -> dict:
    template = s3.get_object(Bucket=bucket, Key="prompts/v1/system.txt")
    # Si S3 no responde → ConnectionError → 500 Internal Server Error
    # El usuario ve un error críptico
    # No hay retry, no hay fallback, no hay degradación

El patrón: la app decide qué hacer cuando algo falla

# ✅ Patrón — si S3 falla, la app degrada
def process_request(document: dict) -> dict:
    try:
        template = get_prompt_with_fallback("summarizer", "v1")
    except ServiceUnavailableError:
        return {
            "status": "degraded",
            "message": "Servicio procesando con capacidad reducida",
            "result": process_with_default_template(document),
        }

Patrón 1: Retry con Exponential Backoff

La primera línea de defensa

Muchas fallas son transitorias: un blip de red de 500ms, un throttle de S3 por rate limit, un cold start de Lambda. Un retry simple resuelve la mayoría de estos casos.

"""services/retry.py — Retry con 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):
    """Todos los reintentos fallaron."""
    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} falló tras {attempts} intentos: {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 que agrega retry con 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__} exitoso en intento {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__} falló (intento {attempt}/{max_retries}): "
                            f"{e}. Reintentando en {delay:.1f}s"
                        )
                        time.sleep(delay)

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

        return wrapper
    return decorator

Uso del 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")

Patrón 2: Circuit Breaker

Evitar reintentos cuando el servicio está claramente caído

Si S3 ha fallado 10 veces en los últimos 30 segundos, seguir reintentando solo agrega latencia. Un circuit breaker "abre el circuito" después de N fallos y retorna el fallback inmediatamente hasta que el servicio se recupere.

"""services/circuit_breaker.py — Circuit breaker para servicios cloud."""

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 pasan
    OPEN = "open"           # Abierto: requests van a fallback directo
    HALF_OPEN = "half_open" # Probando: deja pasar 1 request para verificar


class CircuitBreakerError(Exception):
    """Circuit breaker abierto — servicio no disponible."""
    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 para '{service}': {failures} fallos, "
            f"reset en {reset_in:.0f}s"
        )


class CircuitBreaker:
    """Circuit breaker para un servicio cloud.

    Estados:
    - CLOSED: operación normal, requests pasan al servicio
    - OPEN: servicio caído, requests van a fallback inmediatamente
    - HALF_OPEN: deja pasar 1 request para probar si el servicio se recuperó
    """

    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"(probando recuperación)"
                )
        return self._state

    def execute(self, func: Callable, *args, **kwargs) -> Any:
        """Ejecuta la función a través del 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 (servicio recuperado)"
            )
        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} fallos): {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,
        }

Patrón 3: Fallback Strategies

Estrategias de fallback para diferentes servicios

"""services/fallbacks.py — Estrategias de fallback para servicios cloud."""

import json
import logging
from typing import Any

logger = logging.getLogger(__name__)


class FallbackStrategy:
    """Estrategias de fallback para cuando un servicio no está disponible."""

    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:
        """Obtiene prompt template con cadena de fallbacks.

        1. S3 (fuente primaria)
        2. Cache en memoria (si se leyó antes)
        3. Default hardcodeado (último recurso)
        """
        cache_key = f"prompts/{name}/{version}"

        # Intento 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 activado para {cache_key}: {e}")

        # Intento 2: cache en memoria
        if cache_key in self._cache:
            logger.info(f"Usando prompt cacheado para {cache_key}")
            return self._cache[cache_key]

        # Intento 3: default
        logger.warning(f"Usando prompt default para {name}")
        return self._get_default_prompt(name)

    def _get_default_prompt(self, name: str) -> str:
        """Prompts default hardcodeados como último recurso."""
        defaults = {
            "summarizer": (
                "Genera un resumen conciso del siguiente texto. "
                "Máximo 3 párrafos."
            ),
            "classifier": (
                "Clasifica el siguiente texto en una de estas categorías: "
                "técnico, negocio, general."
            ),
        }
        return defaults.get(name, "Procesa el siguiente texto.")

    def invoke_with_fallback(
        self,
        primary_func,
        fallback_func,
        *args,
        **kwargs,
    ) -> dict:
        """Ejecuta primary_func; si falla, ejecuta 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

Patrón 4: Health Check con Niveles de Degradación

Health check que comunica el estado exacto

"""services/health.py — Health check con niveles de degradación."""

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"         # Todo funciona
    DEGRADED = "degraded"       # Funcionalidad reducida
    CRITICAL = "critical"       # Funcionalidad mínima
    UNAVAILABLE = "unavailable" # App no puede operar


@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:
    """Verifica la salud de la app con niveles de degradación."""

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

    def check(self) -> AppHealth:
        """Ejecuta health check completo."""
        services = []

        # S3 — crítico (sin S3 no hay prompts ni documentos)
        s3_health = self._check_s3()
        services.append(s3_health)

        # Lambda — crítico (sin Lambda no hay inferencia)
        lambda_health = self._check_lambda()
        services.append(lambda_health)

        # SageMaker — opcional (si feature flag activo)
        if self.flags.is_enabled("sagemaker_enrichment"):
            sm_health = self._check_sagemaker()
            services.append(sm_health)

        # CloudWatch — opcional
        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:
        """Computa el nivel de salud general basado en los servicios."""
        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 = "Todos los servicios operando normalmente"
            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"Servicios críticos no disponibles: {[s.name for s in critical_down]}"
            else:
                level = HealthLevel.CRITICAL
                message = f"Algunos servicios críticos degradados: {[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"Servicios opcionales no disponibles: {[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 con degradation levels

"""handler.py — Health endpoint que reporta niveles de degradación."""

import json
from services.health import HealthChecker, HealthLevel


def health_endpoint(checker: HealthChecker) -> dict:
    """Endpoint /health con información de degradación."""
    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),
    }

Ejemplo de response degradado:

{
  "status": "degraded",
  "message": "Servicios opcionales no disponibles: ['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}
  ]
}

Integrando Todo: Processor con Graceful Degradation

"""services/resilient_processor.py — Processor con degradación completa."""

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 que degrada gracefully cuando servicios fallan."""

    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:
        """Procesa documento con degradación en cada paso."""
        result = {"degraded": False, "degradation_details": []}

        # Paso 1: obtener prompt (con 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: usando default ({e})")

        # Paso 2: almacenar documento (mejor esfuerzo)
        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: documento no persistido ({e})")

        # Paso 3: enriquecimiento SageMaker (opcional)
        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: enriquecimiento no disponible ({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

Problema 1: Circuit breaker abre demasiado rápido

5 errores en ráfaga abren el circuit, pero los errores eran transitorios.

# Solución: aumentar threshold o agregar ventana de tiempo
# En lugar de "5 errores totales" → "5 errores en los últimos 30 segundos"

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)
        # Solo contar fallos dentro de la ventana
        self._failure_times = [
            t for t in self._failure_times
            if now - t < self._window
        ]
        self._failure_count = len(self._failure_times)
        # ... resto de la lógica

Problema 2: Fallback retorna datos obsoletos del cache

El cache en memoria no se invalida y el template cambió en S3.

# Solución: cache con 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())

Problema 3: Health check timeout cuando un servicio es lento

SageMaker endpoint tardando 10s hace que el health check sea lento.

# Solución: timeout por servicio en 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)

Problema 4: Retry y circuit breaker interfieren

Retry reintenta 3 veces, cada intento cuenta como fallo en el circuit breaker.

# Solución: solo contar como fallo después de agotar retries
# El circuit breaker envuelve la función con retry:
# circuit_breaker.execute(retry_function)
# NO: retry(circuit_breaker.execute(function))

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

# Donde _get_prompt_with_retry ya tiene @retry_with_backoff

Ejercicios Prácticos

Ejercicio 1: Cache con fallback multi-nivel

Implementa un cache con tres niveles: memoria (L1), archivo local (L2), y S3 (L3). Si L1 falla, prueba L2. Si L2 falla, prueba L3. Cada nivel cachea lo que obtiene del nivel inferior.

import json
import os
from typing import Any, Optional


class MultiLevelCache:
    """Cache con 3 niveles: memoria → archivo → 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: memoria
        if key in self._memory:
            return self._memory[key]

        # L2: archivo local
        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": "Resume en 3 puntos"})

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

Ejercicio 2: Circuit breaker con métricas

Extiende el CircuitBreaker para que registre métricas: número total de calls, success rate, tiempos de respuesta promedio, y número de veces que abrió el circuito.

import time
from collections import deque


class MetricCircuitBreaker:
    """Circuit breaker con métricas detalladas."""

    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))

Ejercicio 3: Degradation response builder

Crea un builder que construya respuestas HTTP apropiadas según el nivel de degradación: 200 con resultado completo, 200 con resultado parcial y header de degradación, o 503 con mensaje de servicio no disponible.

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:
    """Construye responses HTTP apropiados según el nivel de degradación."""

    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": "Algunos features no disponibles",
                    "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": "Funcionalidad significativamente reducida",
                    "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": "Servicio temporalmente no disponible",
                "affected": deg.affected_features,
            }),
        }


# Demo
builder = DegradedResponseBuilder()

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

# Respuesta parcial
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']}")

Ejercicio 4: Degradation simulation test

Escribe un test que simule la caída de S3 y verifique que el ResilientDocumentProcessor retorna una respuesta degradada en lugar de crashear.

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():
    """Verifica que el processor degrada cuando S3 no responde."""
    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: usando default" in result["degradation_details"][0]
    assert result["template_preview"] is not None
    print(f"✅ Processor degradó correctamente: {result['degradation_details']}")


def test_processor_recovers_after_circuit_reset():
    """Verifica que el processor se recupera cuando S3 vuelve."""
    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,
    )

    # Forzar reset inmediato del circuit breaker para test
    processor.s3_breaker.reset_timeout = 0.1

    import time
    time.sleep(0.2)  # Esperar reset

    call_count = fail_until  # Ya pasaron los fallos
    result = processor.process("summarizer", "v1", {"id": "test"})

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


test_processor_degrades_when_s3_fails()

Resumen

• Graceful degradation convierte "app muerta" en "app con funcionalidad reducida." S3 caído no significa error 500 — significa que la app usa prompts default del cache.
• Retry con backoff es la primera línea de defensa. La mayoría de fallas cloud son transitorias (500ms-5s). Un retry las resuelve sin que el usuario note nada.
• Circuit breaker previene cascadas de fallo. Si S3 está caído, seguir reintentando solo agrega latencia. El circuit breaker retorna fallback inmediatamente.
• El health check comunica el estado exacto. HEALTHY → DEGRADED → CRITICAL → UNAVAILABLE. Cada nivel dice qué funciona y qué no.
• Fallback strategies definen qué hacer cuando algo falla. S3 → cache en memoria → default hardcodeado. Cada nivel es peor, pero ninguno es un crash.
• En la siguiente cápsula, integras todo en una Migration-Ready AI App — el proyecto del módulo.

Recursos Adicionales

1. Microsoft — Circuit Breaker Pattern — Referencia del patrón
2. AWS Well-Architected — Reliability — Principios de reliability
3. Martin Fowler — Circuit Breaker — Artículo original
4. Retry Pattern — Azure — Patrón retry con backoff
5. Hystrix Wiki — Netflix Hystrix (circuit breaker reference)
6. Graceful Degradation vs Progressive Enhancement — Concepto general