RELIABILITY & FAILURE HANDLING

Most production failures:
- 60% Timing failures (slow dependencies)
- 25% Crash failures (process/server death)
- 10% Omission failures (network issues)
- 5% Response/Byzantine failures

1. Payment service becomes slow (2 second response)
2. Order service waits for payment (blocking threads)
3. Order service thread pool exhausted
4. Order service stops responding
5. API gateway times out
6. User sees error
7. User retries, making it worse

def get_user_profile(user_id):
    user = user_service.get(user_id)  # Hangs forever if service is slow
    return user

def get_user_profile(user_id):
    try:
        user = user_service.get(user_id, timeout=2)  # Max 2 seconds
        return user
    except TimeoutError:
        return get_cached_user(user_id)  # Fallback

Every external call must have a timeout.
No exceptions. Ever.

Timeout = P99 latency × 2 + buffer

Example:
- P99 latency: 200ms
- Buffer: 100ms
- Timeout: 200 × 2 + 100 = 500ms

# Bad: Inner timeout longer than outer
api_gateway_timeout = 5  # seconds
service_a_timeout = 10   # Service A will never complete!

# Good: Cascading timeouts
api_gateway_timeout = 10
service_a_timeout = 5
service_b_timeout = 2
database_timeout = 1

# Time to establish TCP connection
requests.get(url, timeout=(connect_timeout, read_timeout))
requests.get(url, timeout=(1, 10))  # 1s connect, 10s read

Time to receive response after connection established.

# Bad: Single timeout value
requests.get(url, timeout=10)  # Both connect and read = 10s

# Problem: If server is down, wait 10s just to discover it
# Good: Short connect, longer read
requests.get(url, timeout=(1, 10))  # Fail fast if server down

import requests
from requests.adapters import HTTPAdapter

session = requests.Session()
adapter = HTTPAdapter(
    pool_connections=100,
    pool_maxsize=100,
    max_retries=3
)
session.mount('https://', adapter)

# All requests use these timeouts
response = session.get(
    url,
    timeout=(3, 10)  # 3s connect, 10s read
)

RETRYABLE_EXCEPTIONS = (
    ConnectionError,
    TimeoutError,
    requests.exceptions.ConnectionError,
    requests.exceptions.Timeout,
)

def call_with_retry(func, max_retries=3):
    for attempt in range(max_retries):
        try:
            return func()
        except RetryableError:
            continue  # Retry immediately
    raise MaxRetriesExceeded()

def call_with_backoff(func, max_retries=5):
    for attempt in range(max_retries):
        try:
            return func()
        except RetryableError:
            if attempt == max_retries - 1:
                raise

            delay = (2 ** attempt)  # 1, 2, 4, 8, 16 seconds
            time.sleep(delay)

    raise MaxRetriesExceeded()

1000 requests fail at t=0
All retry at t=1
All fail again
All retry at t=2
Synchronized retry storm

def get_backoff_delay(attempt, base=1, cap=60):
    # Exponential backoff
    exp_delay = base * (2 ** attempt)

    # Cap the delay
    capped = min(exp_delay, cap)

    # Add jitter: random between 0 and capped
    jittered = random.uniform(0, capped)

    return jittered

Service receives 1000 requests/second
10% fail and retry 3 times
Extra load: 1000 × 0.10 × 3 = 300 requests/second
30% additional load on already struggling service

class RetryBudget:
    def __init__(self, budget_percent=20, window_seconds=10):
        self.budget_percent = budget_percent
        self.window = window_seconds
        self.requests = []
        self.retries = []

    def can_retry(self):
        now = time.time()

        # Count recent requests and retries
        recent_requests = len([r for r in self.requests if r > now - self.window])
        recent_retries = len([r for r in self.retries if r > now - self.window])

        # Check budget
        retry_percent = (recent_retries / max(recent_requests, 1)) * 100
        return retry_percent < self.budget_percent

    def record_request(self):
        self.requests.append(time.time())

    def record_retry(self):
        self.retries.append(time.time())

retry_budget = RetryBudget(budget_percent=20)

def call_with_budget(func, max_retries=3):
    retry_budget.record_request()

    for attempt in range(max_retries):
        try:
            return func()
        except RetryableError:
            if attempt == max_retries - 1:
                raise

            if not retry_budget.can_retry():
                raise RetryBudgetExceeded()

            retry_budget.record_retry()
            delay = get_backoff_delay(attempt)
            time.sleep(delay)

Request: "Transfer $100 from A to B"

1. Request sent
2. Server processes, transfers money
3. Response lost (network blip)
4. Client retries (no response = failure?)
5. Server processes again
6. $200 transferred instead of $100

def transfer_money(from_account, to_account, amount, idempotency_key):
    # Check if already processed
    existing = db.query(
        "SELECT * FROM transfers WHERE idempotency_key = ?",
        idempotency_key
    )

    if existing:
        return existing  # Already done

    # Process transfer
    result = execute_transfer(from_account, to_account, amount)

    # Store with idempotency key
    db.insert("transfers", {
        "idempotency_key": idempotency_key,
        "result": result
    })

    return result

def make_request_with_retry(endpoint, data, max_retries=3):
    # Generate idempotency key once
    idempotency_key = str(uuid.uuid4())

    for attempt in range(max_retries):
        try:
            response = requests.post(
                endpoint,
                json=data,
                headers={"Idempotency-Key": idempotency_key}
            )
            return response
        except RetryableError:
            delay = get_backoff_delay(attempt)
            time.sleep(delay)

    raise MaxRetriesExceeded()

1. Payment service goes down
2. Order service keeps calling payment service
3. All calls timeout (5 seconds each)
4. Order service threads blocked waiting
5. Order service thread pool exhausted
6. Order service becomes unresponsive
7. Cascade failure across system

- Requests pass through normally
- Failures are counted
- When failure threshold exceeded, transition to Open

- Requests immediately rejected (no actual call)
- Return error or fallback instantly
- After timeout period, transition to Half-Open

- Allow limited requests through
- If requests succeed, transition to Closed
- If requests fail, transition back to Open

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

    def call(self, func):
        if self.state == "OPEN":
            if time.time() - self.last_failure_time > self.reset_timeout:
                self.state = "HALF_OPEN"
            else:
                raise CircuitOpenError("Circuit breaker is open")

        try:
            result = func()
            self.on_success()
            return result
        except Exception as e:
            self.on_failure()
            raise

class CircuitBreaker:
    # ... __init__ and call from previous slide ...

    def on_success(self):
        self.failure_count = 0
        if self.state == "HALF_OPEN":
            self.state = "CLOSED"

    def on_failure(self):
        self.failure_count += 1
        self.last_failure_time = time.time()

        if self.failure_count >= self.failure_threshold:
            self.state = "OPEN"

        if self.state == "HALF_OPEN":
            self.state = "OPEN"

payment_circuit = CircuitBreaker(failure_threshold=5, reset_timeout=30)

def process_payment(order):
    try:
        return payment_circuit.call(lambda: payment_service.charge(order))
    except CircuitOpenError:
        return queue_for_later(order)  # Fallback

class SlidingWindowCircuitBreaker:
    def __init__(self, failure_threshold=5, window_size=60):
        self.failure_threshold = failure_threshold
        self.window_size = window_size
        self.failures = []  # List of failure timestamps

    def record_failure(self):
        now = time.time()
        self.failures.append(now)

        # Remove old failures outside window
        self.failures = [f for f in self.failures if f > now - self.window_size]

        return len(self.failures) >= self.failure_threshold

class CircuitBreakerRegistry:
    def __init__(self):
        self.breakers = {}

    def get(self, name, **config):
        if name not in self.breakers:
            self.breakers[name] = CircuitBreaker(**config)
        return self.breakers[name]

registry = CircuitBreakerRegistry()

# Different circuits for different operations
payment_circuit = registry.get("payment", failure_threshold=5, reset_timeout=30)
inventory_circuit = registry.get("inventory", failure_threshold=10, reset_timeout=60)
notification_circuit = registry.get("notification", failure_threshold=3, reset_timeout=10)

def get_user_profile(user_id):
    try:
        return user_circuit.call(lambda: user_service.get(user_id))
    except CircuitOpenError:
        return cache.get(f"user:{user_id}")  # Return stale data

def get_recommendations(user_id):
    try:
        return recommendation_circuit.call(
            lambda: recommendation_service.get(user_id)
        )
    except CircuitOpenError:
        return DEFAULT_RECOMMENDATIONS  # Static fallback

def send_notification(user_id, message):
    try:
        return notification_circuit.call(
            lambda: notification_service.send(user_id, message)
        )
    except CircuitOpenError:
        queue.enqueue("send_notification", {"user_id": user_id, "message": message})

def get_product_page(product_id):
    product = product_service.get(product_id)  # Must succeed

    # Optional enrichments with fallbacks
    try:
        product.reviews = review_circuit.call(
            lambda: review_service.get(product_id)
        )
    except CircuitOpenError:
        product.reviews = []  # Show page without reviews

    try:
        product.recommendations = recommendation_circuit.call(
            lambda: recommendation_service.get(product_id)
        )
    except CircuitOpenError:
        product.recommendations = []  # Show page without recommendations

    return product

def get_product_page(product_id):
    product = product_service.get(product_id)  # Must succeed

    # Optional enrichments with fallbacks
    try:
        product.reviews = review_circuit.call(
            lambda: review_service.get(product_id)
        )
    except CircuitOpenError:
        product.reviews = []  # Show page without reviews

    try:
        product.recommendations = recommendation_circuit.call(
            lambda: recommendation_service.get(product_id)
        )
    except CircuitOpenError:
        product.recommendations = []  # Show page without recommendations

    return product

from concurrent.futures import ThreadPoolExecutor

class BulkheadExecutor:
    def __init__(self):
        self.executors = {
            "user": ThreadPoolExecutor(max_workers=30),
            "order": ThreadPoolExecutor(max_workers=40),
            "payment": ThreadPoolExecutor(max_workers=30),
        }

    def submit(self, bulkhead_name, func, *args):
        executor = self.executors[bulkhead_name]
        return executor.submit(func, *args)

bulkhead = BulkheadExecutor()

def get_user(user_id):
    future = bulkhead.submit("user", user_service.get, user_id)
    return future.result(timeout=5)

def process_payment(order):
    future = bulkhead.submit("payment", payment_service.charge, order)
    return future.result(timeout=10)

import threading

class SemaphoreBulkhead:
    def __init__(self, max_concurrent):
        self.semaphore = threading.Semaphore(max_concurrent)
        self.max_concurrent = max_concurrent

    def call(self, func, timeout=None):
        acquired = self.semaphore.acquire(timeout=timeout)

        if not acquired:
            raise BulkheadFullError("Bulkhead capacity exceeded")

        try:
            return func()
        finally:
            self.semaphore.release()

payment_bulkhead = SemaphoreBulkhead(max_concurrent=30)

def process_payment(order):
    try:
        return payment_bulkhead.call(
            lambda: payment_service.charge(order),
            timeout=1  # Wait max 1 second for bulkhead slot
        )
    except BulkheadFullError:
        return queue_for_later(order)

Required concurrent calls = (requests/second) × (average latency in seconds)

Example:
- Payment requests: 100/second
- Payment latency: 200ms (0.2 seconds)
- Required: 100 × 0.2 = 20 concurrent calls

Add buffer: 20 × 1.5 = 30 threads for payment bulkhead

def get_bulkhead_metrics(bulkhead):
    return {
        "active": bulkhead.max_concurrent - bulkhead.semaphore._value,
        "available": bulkhead.semaphore._value,
        "max": bulkhead.max_concurrent,
        "utilization": (bulkhead.max_concurrent - bulkhead.semaphore._value) / bulkhead.max_concurrent
    }

Product Page:
- Product details ✓
- Product images ✓
- Reviews ✓
- Recommendations ✓
- Inventory status ✓
- Shipping estimate ✓

Product Page:
- Product details ✓
- Product images ✓
- Reviews: "Reviews temporarily unavailable"
- Recommendations ✓
- Inventory status ✓
- Shipping estimate ✓

class FeaturePriority:
    CRITICAL = 1    # Must work or page fails
    IMPORTANT = 2   # Should work, degrade if needed
    OPTIONAL = 3    # Nice to have, drop under load

FEATURES = {
    "product_details": FeaturePriority.CRITICAL,
    "inventory_status": FeaturePriority.CRITICAL,
    "reviews": FeaturePriority.IMPORTANT,
    "recommendations": FeaturePriority.OPTIONAL,
    "shipping_estimate": FeaturePriority.IMPORTANT,
    "social_share_count": FeaturePriority.OPTIONAL,
}

def get_product_page(product_id, system_load):
    page = {}

    # Always fetch critical features
    page["product"] = product_service.get(product_id)
    page["inventory"] = inventory_service.check(product_id)

    # Skip optional features under high load
    if system_load < 0.8:
        page["recommendations"] = safe_call(
            lambda: recommendation_service.get(product_id),
            default=[]
        )

    # Degrade important features if circuit open
    page["reviews"] = safe_call(
        lambda: review_service.get(product_id),
        default={"message": "Reviews temporarily unavailable"}
    )

    return page

def safe_call(func, default=None, timeout=5):
    try:
        return func()
    except CircuitOpenError:
        return default
    except TimeoutError:
        return default
    except Exception as e:
        log.warning(f"Feature degraded:{e}")
        return default

def get_page_with_degradation(product_id):
    load = get_system_load()

    if load < 0.5:
        # Full features
        return get_full_product_page(product_id)
    elif load < 0.7:
        # Drop optional features
        return get_product_page_without_optional(product_id)
    elif load < 0.9:
        # Only critical features
        return get_minimal_product_page(product_id)
    else:
        # Emergency mode
        return get_cached_product_page(product_id)

System capacity: 10,000 requests/second
Incoming traffic: 50,000 requests/second

Without load shedding:
- All 50,000 requests accepted
- System overwhelmed
- All requests fail or timeout
- 0% success rate

With load shedding:
- 10,000 requests accepted
- 40,000 requests rejected immediately
- Accepted requests succeed
- 20% success rate (better than 0%)

import threading

class LoadShedder:
    def __init__(self, max_concurrent):
        self.current = 0
        self.max = max_concurrent
        self.lock = threading.Lock()

    def try_acquire(self):
        with self.lock:
            if self.current >= self.max:
                return False
            self.current += 1
            return True

    def release(self):
        with self.lock:
            self.current -= 1

shedder = LoadShedder(max_concurrent=1000)

def handle_request(request):
    if not shedder.try_acquire():
        return Response(status=503, body="Service overloaded, try later")

    try:
        return process_request(request)
    finally:
        shedder.release()

class PriorityLoadShedder:
    def __init__(self, capacities):
        # capacities = {"high": 500, "medium": 300, "low": 200}
        self.capacities = capacities
        self.current = {"high": 0, "medium": 0, "low": 0}
        self.lock = threading.Lock()

    def try_acquire(self, priority):
        with self.lock:
            total_current = sum(self.current.values())
            total_capacity = sum(self.capacities.values())

            # High priority always gets through if capacity available
            if priority == "high" and total_current < total_capacity:
                self.current["high"] += 1
                return True

            # Medium/low priority shed when approaching capacity
            if self.current[priority] < self.capacities[priority]:
                self.current[priority] += 1
                return True

            return False

def classify_request(request):
    if request.path.startswith("/checkout"):
        return "high"
    elif request.path.startswith("/api/analytics"):
        return "low"
    elif request.path.startswith("/api/recommendations"):
        return "low"
    else:
        return "medium"

def handle_request(request):
    priority = classify_request(request)

    if not shedder.try_acquire(priority):
        return Response(
            status=503,
            headers={"Retry-After": "30"},
            body="Service busy, please retry"
        )

    try:
        return process_request(request)
    finally:
        shedder.release(priority)