Graceful Degradation

GPU failures, traffic spikes, and model crashes are not if but when. Graceful degradation maintains service by progressively reducing quality rather than failing completely. A degraded response is better than a timeout.

---

1. The Degradation Ladder

When capacity drops or demand exceeds it, escalate through progressive steps:

Normal operation
  │ capacity < demand
  ▼
Level 1: Disable speculative decoding (free GPU compute)
  │ still not enough
  ▼
Level 2: Reduce max_tokens (shorter responses)
  │ still not enough
  ▼
Level 3: Shed low-priority requests (free tier, batch jobs)
  │ still not enough
  ▼
Level 4: Fallback to smaller model (70B → 8B)
  │ still not enough
  ▼
Level 5: Return cached/static responses for common queries
  │ still not enough
  ▼
Level 6: Return "service degraded" error (429 with retry-after)

class DegradationController:
    LEVELS = [
        {"name": "normal", "actions": []},
        {"name": "reduce_speculation", "actions": ["disable_speculative_decoding"]},
        {"name": "shorter_responses", "actions": ["max_tokens=1024"]},
        {"name": "shed_low_priority", "actions": ["reject_free_tier", "reject_batch"]},
        {"name": "fallback_model", "actions": ["route_to_8b"]},
        {"name": "static_responses", "actions": ["serve_cached_only"]},
        {"name": "service_unavailable", "actions": ["return_429"]},
    ]
    
    def __init__(self):
        self.current_level = 0
    
    async def evaluate(self, metrics: SystemMetrics):
        """Adjust degradation level based on system health."""
        if metrics.queue_depth > 100 or metrics.error_rate > 0.05:
            self.current_level = min(self.current_level + 1, len(self.LEVELS) - 1)
        elif metrics.queue_depth < 10 and metrics.error_rate < 0.01:
            self.current_level = max(self.current_level - 1, 0)
        
        level = self.LEVELS[self.current_level]
        await self.apply_actions(level["actions"])
        
        if self.current_level > 0:
            logger.warning(f"Degradation level: {level['name']}")
    
    async def apply_actions(self, actions: list[str]):
        for action in actions:
            if action == "disable_speculative_decoding":
                await self.engine.set_config(speculative_model=None)
            elif action == "max_tokens=1024":
                self.gateway.default_max_tokens = 1024
            elif action == "reject_free_tier":
                self.gateway.reject_tiers.add("free")
            elif action == "route_to_8b":
                self.router.force_model = "llama-8b"

---

2. Priority-Based Load Shedding

Not all requests are equal. Shed low-value requests first:

class PriorityLoadShedder:
    PRIORITIES = {
        "enterprise": 100,   # Never shed (unless Level 6)
        "pro":        80,    # Shed only at Level 5+
        "starter":    50,    # Shed at Level 3+
        "free":       20,    # Shed at Level 2+
        "batch":      10,    # Shed first
    }
    
    def should_admit(self, request: Request, current_level: int) -> bool:
        tier = request.tenant.tier
        priority = self.PRIORITIES.get(tier, 0)
        
        # Priority threshold increases with degradation level
        threshold = current_level * 20  # Level 1: >20, Level 2: >40, etc.
        
        if priority < threshold:
            metrics.increment("requests_shed", labels={"tier": tier})
            return False
        
        return True
    
    def shed_response(self, request: Request) -> Response:
        """Return appropriate error for shed requests."""
        return JSONResponse(
            status_code=429,
            content={
                "error": {
                    "type": "rate_limit_exceeded",
                    "message": "Service is experiencing high load. Please retry.",
                },
            },
            headers={
                "Retry-After": "30",
                "X-Degradation-Level": str(self.current_level),
            },
        )

---

3. Model Fallback

When the primary model is unavailable or overloaded, fall back to a cheaper alternative:

class ModelFallback:
    FALLBACK_CHAIN = {
        "llama-405b": ["llama-70b", "llama-8b"],
        "llama-70b":  ["llama-8b"],
        "llama-8b":   [],  # No fallback — shed request
    }
    
    async def generate(self, request: ChatRequest) -> Response:
        chain = [request.model] + self.FALLBACK_CHAIN.get(request.model, [])
        
        for model in chain:
            try:
                backend = self.router.get_backend(model)
                if backend.queue_depth > backend.max_queue:
                    continue  # Skip overloaded backend
                
                response = await asyncio.wait_for(
                    backend.generate(request),
                    timeout=30,
                )
                
                if model != request.model:
                    response.headers["X-Fallback-Model"] = model
                    logger.warning(f"Fallback: {request.model} → {model}")
                
                return response
            except (asyncio.TimeoutError, BackendError):
                logger.error(f"Model {model} failed, trying next")
                continue
        
        # All models failed
        raise ServiceUnavailable("All model backends unavailable")

Communicating Fallback to Clients

{
  "id": "chatcmpl-abc123",
  "model": "llama-8b",
  "choices": [{"message": {"content": "..."}}],
  "_fallback": {
    "requested_model": "llama-70b",
    "served_model": "llama-8b",
    "reason": "primary_overloaded"
  }
}

---

4. Circuit Breaker Pattern

Prevent cascading failures by stopping requests to a failing backend:

class CircuitBreaker:
    """Stop sending to a backend that's repeatedly failing."""
    
    def __init__(self, failure_threshold=5, reset_timeout=30):
        self.failure_count = 0
        self.failure_threshold = failure_threshold
        self.reset_timeout = reset_timeout
        self.state = "closed"  # closed → open → half-open → closed
        self.last_failure_time = 0
    
    async def call(self, backend, request):
        if self.state == "open":
            if time.time() - self.last_failure_time > self.reset_timeout:
                self.state = "half-open"  # Try one request
            else:
                raise CircuitOpen(f"Backend {backend.id} circuit is open")
        
        try:
            result = await backend.generate(request)
            if self.state == "half-open":
                self.state = "closed"  # Backend recovered
                self.failure_count = 0
            return result
        except Exception as e:
            self.failure_count += 1
            self.last_failure_time = time.time()
            
            if self.failure_count >= self.failure_threshold:
                self.state = "open"
                logger.critical(f"Circuit OPEN for backend {backend.id}")
            
            raise

---

5. GPU Failure Handling

When a GPU fails mid-request or becomes unavailable:

class GPUFailureHandler:
    async def handle_gpu_failure(self, gpu_id: str, active_requests: list):
        """Handle a GPU going offline."""
        logger.critical(f"GPU {gpu_id} failure detected")
        
        # 1. Mark GPU as unavailable
        await self.scheduler.remove_gpu(gpu_id)
        
        # 2. Retry active requests on other GPUs
        for request in active_requests:
            if request.can_retry:
                await self.scheduler.requeue(request)
                metrics.increment("gpu_failure_retries")
            else:
                await request.fail("GPU failure during processing")
                metrics.increment("gpu_failure_drops")
        
        # 3. Rebalance model placement
        await self.model_manager.rebalance_without_gpu(gpu_id)
        
        # 4. Alert ops
        await self.alert(
            severity="critical",
            summary=f"GPU {gpu_id} offline, {len(active_requests)} requests affected",
        )

---

6. Recovery and Level-Down

Degradation must be reversible. Automate the recovery:

class RecoveryController:
    """Gradually restore service after degradation."""
    
    async def check_recovery(self):
        """Run every 30 seconds."""
        if self.degradation.current_level == 0:
            return  # Already normal
        
        metrics = await self.get_metrics()
        
        # Recovery conditions (must be stable for 5 minutes)
        if (metrics.queue_depth < 10 and 
            metrics.error_rate < 0.005 and
            metrics.gpu_utilization < 0.8):
            
            if self._stable_for(minutes=5):
                # Step down one level
                self.degradation.current_level -= 1
                logger.info(f"Recovery: level down to {self.degradation.current_level}")
                
                # Re-admit shed traffic gradually
                if "reject_free_tier" in self.degradation.active_actions:
                    await self._gradual_readmit("free", ramp_minutes=10)

---

Self-Check

• Design a 6-level degradation ladder appropriate for your service
• Implement priority-based load shedding by tenant tier
• Build a model fallback chain with proper error handling
• Implement a circuit breaker for failing inference backends
• Design an automated recovery controller that steps down gracefully

---

References

• Google SRE Book: Handling Overload
• Circuit Breaker Pattern (Martin Fowler)
• For alerting integration, see Alerting