Reliability & Fault Tolerance

# Production fallback chain implementation

class ResilientPredictor:
    def __init__(self):
        self.primary_model = GPUModelClient("triton:8001", timeout_ms=100)
        self.fallback_model = CPUModelClient("lightgbm-server:8080", timeout_ms=30)
        self.rule_engine = RuleBasedRecommender()
        self.cache = Redis(host="redis:6379")

    async def predict(self, user_id: str, context: dict) -> PredictionResult:
        # Tier 1: Try primary model
        try:
            result = await self.primary_model.predict(user_id, context)
            # Cache successful result for Tier 4 fallback
            self.cache.setex(f"pred:{user_id}", 86400, json.dumps(result))
            return PredictionResult(items=result, tier="primary", quality="full")
        except (TimeoutError, ModelUnavailableError) as e:
            metrics.increment("fallback.tier1_failed", tags={"error": type(e).__name__})

        # Tier 2: Try simpler model
        try:
            result = await self.fallback_model.predict(user_id, context)
            return PredictionResult(items=result, tier="fallback", quality="reduced")
        except Exception as e:
            metrics.increment("fallback.tier2_failed")

        # Tier 3: Rule-based
        try:
            result = self.rule_engine.recommend(context.get("category"), context.get("country"))
            return PredictionResult(items=result, tier="rules", quality="generic")
        except Exception:
            metrics.increment("fallback.tier3_failed")

        # Tier 4: Cached response
        cached = self.cache.get(f"pred:{user_id}")
        if cached:
            return PredictionResult(items=json.loads(cached), tier="cache", quality="stale")

        # Tier 5: Static default (absolute last resort)
        return PredictionResult(items=GLOBAL_TOP_20, tier="static", quality="minimal")

# Circuit breaker for model inference

class CircuitBreaker:
    """
    States:
    - CLOSED: Normal operation, requests go to the model
    - OPEN: Model is failing, requests go to fallback immediately
    - HALF_OPEN: Testing if model has recovered
    """
    def __init__(self, failure_threshold=5, recovery_timeout=30, half_open_max=3):
        self.state = "CLOSED"
        self.failure_count = 0
        self.failure_threshold = failure_threshold   # Open after N failures
        self.recovery_timeout = recovery_timeout     # Seconds before trying again
        self.half_open_max = half_open_max           # Test requests in half-open
        self.last_failure_time = None
        self.half_open_successes = 0

    async def call(self, func, *args, **kwargs):
        if self.state == "OPEN":
            if self._should_attempt_reset():
                self.state = "HALF_OPEN"
                self.half_open_successes = 0
            else:
                raise CircuitOpenError(
                    f"Circuit open, retry in {self._time_until_retry()}s"
                )

        try:
            result = await asyncio.wait_for(func(*args, **kwargs), timeout=0.1)
            self._on_success()
            return result
        except (TimeoutError, ConnectionError, ModelError) as e:
            self._on_failure()
            raise

    def _on_failure(self):
        self.failure_count += 1
        self.last_failure_time = time.time()
        if self.failure_count >= self.failure_threshold:
            self.state = "OPEN"
            metrics.increment("circuit_breaker.opened")
            alert("Circuit breaker OPENED for model service")

    def _on_success(self):
        if self.state == "HALF_OPEN":
            self.half_open_successes += 1
            if self.half_open_successes >= self.half_open_max:
                self.state = "CLOSED"
                self.failure_count = 0
                metrics.increment("circuit_breaker.closed")
        else:
            self.failure_count = 0

    def _should_attempt_reset(self):
        return time.time() - self.last_failure_time > self.recovery_timeout

# Usage:
breaker = CircuitBreaker(failure_threshold=5, recovery_timeout=30)

async def get_prediction(user_id):
    try:
        return await breaker.call(primary_model.predict, user_id)
    except CircuitOpenError:
        return await fallback_model.predict(user_id)  # Skip primary entirely

# Three-level health check for AI services

@app.get("/health/live")
async def liveness():
    """Is the process alive? (Kubernetes liveness probe)"""
    return {"status": "alive"}

@app.get("/health/ready")
async def readiness():
    """Is the model loaded and ready to serve? (Kubernetes readiness probe)"""
    checks = {
        "model_loaded": model_server.is_model_loaded(),
        "gpu_available": torch.cuda.is_available(),
        "gpu_memory_ok": torch.cuda.memory_allocated() < 0.95 * torch.cuda.get_device_properties(0).total_mem,
        "feature_store_connected": feature_store.ping(),
    }
    all_ok = all(checks.values())
    return JSONResponse(
        status_code=200 if all_ok else 503,
        content={"status": "ready" if all_ok else "not_ready", "checks": checks}
    )

@app.get("/health/inference")
async def inference_health():
    """Can the model produce valid predictions? (Deep health check, run every 60s)"""
    try:
        # Run inference on a known test input
        test_input = load_canary_input()  # Pre-defined input with known expected output
        start = time.time()
        result = await model.predict(test_input)
        latency_ms = (time.time() - start) * 1000

        checks = {
            "inference_works": result is not None,
            "output_valid": validate_output_schema(result),
            "latency_ok": latency_ms < 200,  # Under 200ms
            "output_reasonable": is_output_in_expected_range(result, test_input),
            "latency_ms": round(latency_ms, 1)
        }
        all_ok = all(v for k, v in checks.items() if k != "latency_ms")
        return JSONResponse(
            status_code=200 if all_ok else 503,
            content={"status": "healthy" if all_ok else "degraded", "checks": checks}
        )
    except Exception as e:
        return JSONResponse(
            status_code=503,
            content={"status": "unhealthy", "error": str(e)}
        )