Reliability & Fault Tolerance

import asyncio
import random
import time
from functools import wraps
from typing import Callable, TypeVar, Type, Tuple, Optional, Any
from dataclasses import dataclass
from enum import Enum
import logging

T = TypeVar('T')
logger = logging.getLogger(__name__)

class RetryStrategy(Enum):
    EXPONENTIAL = "exponential"
    LINEAR = "linear"
    FIBONACCI = "fibonacci"
    DECORRELATED_JITTER = "decorrelated_jitter"

@dataclass
class RetryConfig:
    max_retries: int = 5
    base_delay: float = 1.0
    max_delay: float = 60.0
    strategy: RetryStrategy = RetryStrategy.EXPONENTIAL
    jitter: bool = True
    retryable_exceptions: Tuple[Type[Exception], ...] = (Exception,)
    non_retryable_exceptions: Tuple[Type[Exception], ...] = ()
    on_retry: Optional[Callable[[Exception, int, float], None]] = None

class RetryExhausted(Exception):
    def __init__(self, last_exception: Exception, attempts: int):
        self.last_exception = last_exception
        self.attempts = attempts
        super().__init__(f"All {attempts} retry attempts failed")

class RetryHandler:
    """Advanced retry handler with multiple strategies"""
    
    def __init__(self, config: RetryConfig = None):
        self.config = config or RetryConfig()
        self._fib_cache = {0: 0, 1: 1}
        self._last_delay = self.config.base_delay
    
    def _fibonacci(self, n: int) -> int:
        if n not in self._fib_cache:
            self._fib_cache[n] = self._fibonacci(n - 1) + self._fibonacci(n - 2)
        return self._fib_cache[n]
    
    def _calculate_delay(self, attempt: int) -> float:
        cfg = self.config
        
        if cfg.strategy == RetryStrategy.EXPONENTIAL:
            delay = cfg.base_delay * (2 ** attempt)
        elif cfg.strategy == RetryStrategy.LINEAR:
            delay = cfg.base_delay * (attempt + 1)
        elif cfg.strategy == RetryStrategy.FIBONACCI:
            delay = cfg.base_delay * self._fibonacci(attempt + 2)
        elif cfg.strategy == RetryStrategy.DECORRELATED_JITTER:
            # AWS recommended: sleep = min(cap, random(base, sleep * 3))
            delay = random.uniform(cfg.base_delay, self._last_delay * 3)
            self._last_delay = delay
        else:
            delay = cfg.base_delay
        
        delay = min(delay, cfg.max_delay)
        
        # Add standard jitter (not for decorrelated which has built-in)
        if cfg.jitter and cfg.strategy != RetryStrategy.DECORRELATED_JITTER:
            delay = delay * (0.5 + random.random())
        
        return delay
    
    def _should_retry(self, exception: Exception) -> bool:
        # Check non-retryable first
        if isinstance(exception, self.config.non_retryable_exceptions):
            return False
        return isinstance(exception, self.config.retryable_exceptions)
    
    async def execute(self, func: Callable[[], T]) -> T:
        """Execute function with retry logic"""
        last_exception = None
        
        for attempt in range(self.config.max_retries):
            try:
                if asyncio.iscoroutinefunction(func):
                    return await func()
                return func()
                
            except Exception as e:
                last_exception = e
                
                if not self._should_retry(e):
                    raise
                
                if attempt == self.config.max_retries - 1:
                    break
                
                delay = self._calculate_delay(attempt)
                
                logger.warning(
                    f"Attempt {attempt + 1}/{self.config.max_retries} failed: {e}. "
                    f"Retrying in {delay:.2f}s"
                )
                
                if self.config.on_retry:
                    self.config.on_retry(e, attempt + 1, delay)
                
                await asyncio.sleep(delay)
        
        raise RetryExhausted(last_exception, self.config.max_retries)
    
    def __call__(self, func: Callable) -> Callable:
        """Use as decorator"""
        @wraps(func)
        async def async_wrapper(*args, **kwargs):
            return await self.execute(lambda: func(*args, **kwargs))
        
        @wraps(func)
        def sync_wrapper(*args, **kwargs):
            return asyncio.run(self.execute(lambda: func(*args, **kwargs)))
        
        return async_wrapper if asyncio.iscoroutinefunction(func) else sync_wrapper


# Retry with circuit breaker integration
class ResilientCaller:
    """Combines retry, circuit breaker, and timeout"""
    
    def __init__(
        self,
        circuit_breaker: 'CircuitBreaker',
        retry_config: RetryConfig = None,
        timeout_seconds: float = 30.0
    ):
        self.circuit = circuit_breaker
        self.retry = RetryHandler(retry_config or RetryConfig(max_retries=3))
        self.timeout = timeout_seconds
    
    async def call(
        self, 
        func: Callable, 
        fallback: Callable = None,
        *args, 
        **kwargs
    ) -> Any:
        """Execute with full resilience pattern"""
        
        async def wrapped():
            async with asyncio.timeout(self.timeout):
                return await self.circuit.call_async(
                    lambda: func(*args, **kwargs)
                )
        
        try:
            return await self.retry.execute(wrapped)
        except (RetryExhausted, CircuitBreakerOpenException) as e:
            if fallback:
                logger.warning(f"Using fallback due to: {e}")
                return await fallback(*args, **kwargs) if asyncio.iscoroutinefunction(fallback) else fallback(*args, **kwargs)
            raise


# Usage examples
@RetryHandler(RetryConfig(
    max_retries=3,
    strategy=RetryStrategy.EXPONENTIAL,
    retryable_exceptions=(ConnectionError, TimeoutError)
))
async def fetch_data(url: str) -> dict:
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as resp:
            return await resp.json()


# With metrics callback
def log_retry(exc: Exception, attempt: int, delay: float):
    metrics.increment("retries", tags={"attempt": attempt})
    
retry_handler = RetryHandler(RetryConfig(
    max_retries=5,
    strategy=RetryStrategy.DECORRELATED_JITTER,
    on_retry=log_retry
))

async def call_payment_api(amount: float):
    return await retry_handler.execute(
        lambda: payment_client.charge(amount)
    )

import asyncio
from contextlib import asynccontextmanager
from dataclasses import dataclass, field
from typing import Dict, Any, Optional, Callable
from enum import Enum
import time
import logging

logger = logging.getLogger(__name__)

class BulkheadFullError(Exception):
    def __init__(self, bulkhead_name: str, current: int, max_size: int):
        self.bulkhead_name = bulkhead_name
        self.current = current
        self.max_size = max_size
        super().__init__(
            f"Bulkhead '{bulkhead_name}' is full ({current}/{max_size})"
        )

@dataclass
class BulkheadMetrics:
    accepted: int = 0
    rejected: int = 0
    active: int = 0
    peak_active: int = 0
    total_wait_time: float = 0.0

class Bulkhead:
    """
    Bulkhead pattern with semaphore and queue.
    Isolates failures to prevent cascade effects.
    """
    
    def __init__(
        self,
        name: str,
        max_concurrent: int,
        max_queue: int = 0,
        queue_timeout: float = 10.0
    ):
        self.name = name
        self.max_concurrent = max_concurrent
        self.max_queue = max_queue
        self.queue_timeout = queue_timeout
        
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.metrics = BulkheadMetrics()
        self._queue_size = 0
    
    @asynccontextmanager
    async def acquire(self, timeout: Optional[float] = None):
        """Acquire a slot in the bulkhead"""
        timeout = timeout or self.queue_timeout
        start = time.time()
        
        # Check if we can queue
        if self.semaphore.locked():
            if self._queue_size >= self.max_queue:
                self.metrics.rejected += 1
                raise BulkheadFullError(
                    self.name, 
                    self.metrics.active, 
                    self.max_concurrent
                )
            self._queue_size += 1
        
        try:
            acquired = await asyncio.wait_for(
                self.semaphore.acquire(),
                timeout=timeout
            )
        except asyncio.TimeoutError:
            self._queue_size = max(0, self._queue_size - 1)
            self.metrics.rejected += 1
            raise BulkheadFullError(
                self.name,
                self.metrics.active,
                self.max_concurrent
            )
        finally:
            if self._queue_size > 0:
                self._queue_size -= 1
        
        wait_time = time.time() - start
        self.metrics.total_wait_time += wait_time
        self.metrics.active += 1
        self.metrics.accepted += 1
        self.metrics.peak_active = max(self.metrics.peak_active, self.metrics.active)
        
        try:
            yield
        finally:
            self.metrics.active -= 1
            self.semaphore.release()
    
    def __call__(self, func: Callable):
        """Use as decorator"""
        async def wrapper(*args, **kwargs):
            async with self.acquire():
                return await func(*args, **kwargs)
        return wrapper
    
    def get_metrics(self) -> Dict[str, Any]:
        return {
            "name": self.name,
            "active": self.metrics.active,
            "max_concurrent": self.max_concurrent,
            "queue_size": self._queue_size,
            "max_queue": self.max_queue,
            "accepted": self.metrics.accepted,
            "rejected": self.metrics.rejected,
            "rejection_rate": self.metrics.rejected / max(1, self.metrics.accepted + self.metrics.rejected),
            "peak_active": self.metrics.peak_active,
            "avg_wait_ms": (self.metrics.total_wait_time / max(1, self.metrics.accepted)) * 1000
        }


class ThreadPoolBulkhead:
    """
    Thread pool based bulkhead for CPU-bound or blocking operations.
    Uses a dedicated thread pool to isolate work.
    """
    
    def __init__(self, name: str, max_workers: int):
        from concurrent.futures import ThreadPoolExecutor
        self.name = name
        self.executor = ThreadPoolExecutor(
            max_workers=max_workers,
            thread_name_prefix=f"bulkhead-{name}"
        )
        self.max_workers = max_workers
    
    async def run(self, func: Callable, *args, **kwargs) -> Any:
        """Run blocking function in isolated thread pool"""
        loop = asyncio.get_event_loop()
        return await loop.run_in_executor(
            self.executor,
            lambda: func(*args, **kwargs)
        )
    
    def shutdown(self):
        self.executor.shutdown(wait=True)


class BulkheadManager:
    """Manage multiple bulkheads for different services"""
    
    _instance = None
    
    def __new__(cls):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
            cls._instance.bulkheads = {}
        return cls._instance
    
    def create(
        self,
        name: str,
        max_concurrent: int,
        max_queue: int = 0
    ) -> Bulkhead:
        bulkhead = Bulkhead(name, max_concurrent, max_queue)
        self.bulkheads[name] = bulkhead
        return bulkhead
    
    def get(self, name: str) -> Optional[Bulkhead]:
        return self.bulkheads.get(name)
    
    def get_all_metrics(self) -> Dict[str, Dict]:
        return {name: bh.get_metrics() for name, bh in self.bulkheads.items()}


# Usage
bulkhead_manager = BulkheadManager()

# Different pools for different services
payment_bulkhead = bulkhead_manager.create("payment", max_concurrent=10, max_queue=20)
inventory_bulkhead = bulkhead_manager.create("inventory", max_concurrent=30)
notification_bulkhead = bulkhead_manager.create("notification", max_concurrent=50)

@payment_bulkhead
async def charge_payment(order_id: str, amount: float):
    """Payment calls isolated in their own pool"""
    return await payment_client.charge(order_id, amount)

@inventory_bulkhead
async def reserve_inventory(items: list):
    """Inventory calls isolated - won't affect payments"""
    return await inventory_client.reserve(items)

async def process_order(order):
    """Even if inventory is slow, payments still work"""
    try:
        # These run in isolated pools
        payment = await charge_payment(order.id, order.total)
        inventory = await reserve_inventory(order.items)
        return {"status": "success", "payment": payment}
    except BulkheadFullError as e:
        # One service full doesn't crash everything
        logger.warning(f"Bulkhead full: {e}")
        return {"status": "retry_later", "reason": str(e)}

# FastAPI integration
from fastapi import FastAPI, Request, HTTPException

app = FastAPI()

@app.middleware("http")
async def bulkhead_metrics_middleware(request: Request, call_next):
    response = await call_next(request)
    # Add bulkhead metrics to response headers
    metrics = bulkhead_manager.get_all_metrics()
    response.headers["X-Bulkhead-Active"] = str(
        sum(m["active"] for m in metrics.values())
    )
    return response

@app.get("/metrics/bulkheads")
async def get_bulkhead_metrics():
    return bulkhead_manager.get_all_metrics()