Graceful Degradation Patterns

Graceful degradation is the practice of designing systems so that when individual components fail or become unavailable, the overall system continues to function, albeit with reduced capability or performance. This pattern matters when building distributed services where failure is inevitable and user experience must be preserved during partial outages.

Handle Missing Dependencies

When a service depends on external systems, implement fallback behaviors for missing dependencies.

HTTP Service Fallbacks

Use conditional logic in API responses to indicate degraded states:

GET /api/user/123 HTTP/1.1
Host: example.com

HTTP/1.1 200 OK
Content-Type: application/json

{
  "user": {
    "id": 123,
    "name": "John Doe",
    "email": "[email protected]"
  },
  "status": "partial_degradation",
  "warnings": [
    "profile_image_unavailable",
    "recent_activity_missing"
  ]
}

When a dependency like an image service fails, return user data with a placeholder:

GET /api/user/123 HTTP/1.1
Host: example.com

HTTP/1.1 200 OK
Content-Type: application/json

{
  "user": {
    "id": 123,
    "name": "John Doe",
    "email": "[email protected]",
    "profile_image": "/static/images/default-avatar.png"
  },
  "status": "degraded",
  "warnings": [
    "image_service_unreachable"
  ]
}

Database Fallbacks

When a database becomes unavailable, implement circuit breaker patterns:

import time
from functools import wraps

class CircuitBreaker:
    def __init__(self, failure_threshold=5, timeout=60):
        self.failure_threshold = failure_threshold
        self.timeout = timeout
        self.failure_count = 0
        self.last_failure_time = None
        self.state = "CLOSED"

    def call(self, func, *args, **kwargs):
        if self.state == "OPEN":
            if time.time() - self.last_failure_time > self.timeout:
                self.state = "HALF_OPEN"
            else:
                raise Exception("Circuit breaker is OPEN")

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

    def success(self):
        self.failure_count = 0
        self.state = "CLOSED"

    def failure(self):
        self.failure_count += 1
        self.last_failure_time = time.time()
        if self.failure_count >= self.failure_threshold:
            self.state = "OPEN"

Implement Feature Toggles

Control functionality based on system health or configuration state:

feature_toggles:
  enable_profile_images: true
  enable_recent_activity: false
  enable_user_ratings: true
  enable_recommendations: false

In code, check feature flags before enabling functionality:

def get_user_profile(user_id):
    profile = fetch_user_from_db(user_id)
    
    if config.get('feature_toggles.enable_profile_images'):
        profile['avatar'] = fetch_avatar_image(user_id)
    else:
        profile['avatar'] = '/static/images/default.png'
        
    if config.get('feature_toggles.enable_recent_activity'):
        profile['recent_activity'] = fetch_recent_activity(user_id)
    else:
        profile['recent_activity'] = []
        
    return profile

Manage Resource Constraints

When memory or CPU becomes limited, prioritize essential operations:

Memory Management

Implement resource-aware data structures:

class CacheWithEviction:
    def __init__(self, max_size=1000):
        self.max_size = max_size
        self.cache = {}
        self.access_order = []

    def get(self, key):
        if key in self.cache:
            self.access_order.remove(key)
            self.access_order.append(key)
            return self.cache[key]
        return None

    def put(self, key, value):
        if key in self.cache:
            self.access_order.remove(key)
        elif len(self.cache) >= self.max_size:
            # Evict least recently used
            lru_key = self.access_order.pop(0)
            del self.cache[lru_key]
            
        self.cache[key] = value
        self.access_order.append(key)

CPU Budgeting

Limit concurrent operations in high-load scenarios:

import asyncio
from asyncio import Semaphore

# Limit to 10 concurrent operations
semaphore = Semaphore(10)

async def process_request(request_data):
    async with semaphore:
        # Process request here
        return await heavy_computation(request_data)

Handle Network Partitioning

Design for intermittent connectivity:

Retry Logic with Exponential Backoff

import time
import random

def exponential_backoff_retry(func, max_retries=3, base_delay=1):
    for attempt in range(max_retries):
        try:
            return func()
        except Exception as e:
            if attempt == max_retries - 1:
                raise e
            delay = base_delay * (2 ** attempt) + random.uniform(0, 1)
            time.sleep(delay)

Request Timeout Management

import requests

def fetch_with_timeout(url, timeout=5):
    try:
        response = requests.get(url, timeout=timeout)
        return response.json()
    except requests.exceptions.Timeout:
        # Return cached data or default response
        return {"error": "timeout", "fallback": True}
    except requests.exceptions.RequestException as e:
        # Handle other HTTP errors gracefully
        return {"error": str(e), "fallback": True}

Maintain Data Consistency

When partial failure occurs, ensure data integrity:

Write-Through Cache Pattern

class WriteThroughCache:
    def __init__(self):
        self.cache = {}
        self.storage = Database()

    def set(self, key, value):
        # Write to both cache and storage
        self.cache[key] = value
        try:
            self.storage.save(key, value)
        except Exception:
            # Cache is still updated even if storage fails
            pass

    def get(self, key):
        if key in self.cache:
            return self.cache[key]
        # If not in cache, try to fetch from storage
        try:
            value = self.storage.load(key)
            self.cache[key] = value
            return value
        except Exception:
            return None

Implement Health Checks

Monitor system components and respond appropriately:

Service Health Endpoints

GET /health HTTP/1.1
Host: example.com

HTTP/1.1 200 OK
Content-Type: application/json

{
  "status": "degraded",
  "components": {
    "database": "healthy",
    "image_service": "unhealthy",
    "notification_service": "degraded"
  },
  "timestamp": "2023-01-01T00:00:00Z"
}

Graceful Shutdown Procedures

import signal
import sys

shutdown_event = asyncio.Event()

def signal_handler(signum, frame):
    print("Shutdown signal received")
    shutdown_event.set()
    sys.exit(0)

signal.signal(signal.SIGTERM, signal_handler)
signal.signal(signal.SIGINT, signal_handler)

Handle Timeouts and Latency

Design for predictable behavior under varying performance conditions:

Timeout Configuration

class ServiceClient:
    def __init__(self, timeout_config):
        self.timeout_config = timeout_config  # {read: 5, connect: 2}
        
    def call_service(self, endpoint):
        try:
            response = requests.get(
                f"{self.base_url}/{endpoint}",
                timeout=(
                    self.timeout_config['connect'],
                    self.timeout_config['read']
                )
            )
            return response.json()
        except requests.exceptions.Timeout:
            # Return partial data or default
            return {"status": "timeout", "partial_data": True}

Manage External Service Failures

When third-party services become unavailable, maintain functionality:

Fallback Response Patterns

def get_user_data(user_id):
    try:
        # Primary service call
        user_data = primary_service.fetch_user(user_id)
        return user_data
    except ServiceUnavailableError:
        # Fallback to cached data or defaults
        cached = cache.get(f"user:{user_id}")
        if cached:
            return cached
        return {
            "id": user_id,
            "name": "Unknown User",
            "email": f"user-{user_id}@example.com"
        }

Circuit Breaker Implementation

class ServiceCircuitBreaker:
    def __init__(self, failure_threshold=5, timeout=30):
        self.failure_threshold = failure_threshold
        self.timeout = timeout
        self.failure_count = 0
        self.last_failure_time = None
        self.state = "CLOSED"  # CLOSED, OPEN, HALF_OPEN
        
    def call(self, service_func, *args, **kwargs):
        if self.state == "OPEN":
            if time.time() - self.last_failure_time > self.timeout:
                self.state = "HALF_OPEN"
                result = service_func(*args, **kwargs)
                if result is not None:
                    self.success()
                else:
                    self.failure()
                return result
            else:
                raise ServiceUnavailableError("Service is currently unavailable")
        else:
            try:
                result = service_func(*args, **kwargs)
                self.success()
                return result
            except Exception:
                self.failure()
                raise

    def success(self):
        self.failure_count = 0
        self.state = "CLOSED"

    def failure(self):
        self.failure_count += 1
        self.last_failure_time = time.time()
        if self.failure_count >= self.failure_threshold:
            self.state = "OPEN"

Monitor and Report Degradation

Track system behavior during failures:

Error Logging with Context

import logging

logger = logging.getLogger(__name__)

def degraded_operation():
    try:
        # Primary operation
        result = primary_operation()
        return result
    except Exception as e:
        logger.warning(f"Degraded operation completed with error: {e}")
        # Return fallback result
        return fallback_result()

Sharp Edges and Common Pitfalls

Silent Failures

Dependency failures that don't propagate errors properly:

# ❌ Bad: Silent failure in service call
def process_user_data(user_id):
    try:
        image_url = fetch_avatar_image(user_id)
        # No error handling for image fetch
        return {"user_id": user_id, "avatar": image_url}
    except Exception:
        return {"user_id": user_id, "avatar": "/default.png"}

# ✅ Good: Explicit error reporting
def process_user_data(user_id):
    try:
        image_url = fetch_avatar_image(user_id)
        return {"user_id": user_id, "avatar": image_url}
    except ServiceUnavailableError:
        logger.warning(f"Avatar service unavailable for user {user_id}")
        return {
            "user_id": user_id, 
            "avatar": "/default.png",
            "warnings": ["avatar_unavailable"]
        }

Configuration Confusion

Misunderstanding which features to disable:

# ❌ Bad: Overly complex configuration
feature_flags:
  image_service_enabled: true
  image_service_fallback_enabled: false
  image_service_cache_enabled: true

# ✅ Good: Simple, clear configuration
feature_toggles:
  enable_user_avatars: true
  enable_recent_activity: false

Memory Leaks in Degraded States

Failure to clean up resources during partial failures:

# ❌ Bad: Resource leak during fallback
def handle_request(request):
    resource = acquire_resource()
    try:
        return process_request(request)
    except Exception:
        # Fallback but don't release resource
        return fallback_response()

# ✅ Good: Proper cleanup
def handle_request(request):
    resource = acquire_resource()
    try:
        return process_request(request)
    except Exception:
        # Cleanup before fallback
        release_resource(resource)
        return fallback_response()

The key to successful graceful degradation is implementing clear failure modes that maintain user experience while providing appropriate feedback about system status. The patterns above should be implemented consistently across services, with clear error boundaries and predictable behavior when components fail.

This page was rewritten on 15 October 2026. It replaced a templated version that was not specific to its own title. How these pages are written and checked. If anything here is wrong, tell us at [email protected] and we will correct it.