feat: resilience, circuit breakers, and LRU cache system

backend/orchestration/resilience.py

@@ -0,0 +1,182 @@
+import asyncio
+import random
+import logging
+from enum import Enum
+from functools import wraps
+from typing import Callable, Tuple, Any
+from dataclasses import dataclass
+from datetime import datetime, UTC
+
+from backend.core.exceptions import (
+    InferenceError, CircuitOpenError, ValidationError, 
+    SecurityError, ModelNotLoadedError
+)
+
+logger = logging.getLogger(__name__)
+
+# ━━━ PART A: RETRY DECORATOR ━━━
+
+@dataclass
+class RetryConfig:
+    max_attempts: int = 3
+    initial_delay_seconds: float = 1.0
+    exponential_base: float = 2.0
+    max_delay_seconds: float = 30.0
+    jitter_factor: float = 0.2
+    retry_on: Tuple[type[Exception], ...] = (InferenceError, TimeoutError, asyncio.TimeoutError)
+    no_retry_on: Tuple[type[Exception], ...] = (ValidationError, SecurityError, ModelNotLoadedError)
+
+def retry(config: RetryConfig | None = None):
+    """Decorator factory for async functions to retry on failure with exponential backoff and jitter."""
+    if config is None:
+        config = RetryConfig()
+    
+    def decorator(func: Callable):
+        @wraps(func)
+        async def wrapper(*args, **kwargs):
+            last_exception = None
+            
+            for attempt in range(1, config.max_attempts + 1):
+                try:
+                    return await func(*args, **kwargs)
+                
+                except config.no_retry_on as e:
+                    # Immediately fail on exceptions we know won't succeed on retry
+                    raise
+                
+                except config.retry_on as e:
+                    last_exception = e
+                    
+                    if attempt == config.max_attempts:
+                        logger.error(f"{func.__name__} failed after {attempt} attempts: {e}")
+                        raise
+                    
+                    delay = min(
+                        config.initial_delay_seconds * (config.exponential_base ** (attempt - 1)),
+                        config.max_delay_seconds
+                    )
+                    jitter = delay * config.jitter_factor * (2 * random.random() - 1)
+                    actual_delay = max(0.1, delay + jitter)
+                    
+                    logger.warning(
+                        f"{func.__name__} attempt {attempt}/{config.max_attempts} failed. "
+                        f"Retrying in {actual_delay:.1f}s. Error: {e}"
+                    )
+                    await asyncio.sleep(actual_delay)
+            
+            if last_exception:
+                raise last_exception
+        return wrapper
+    return decorator
+
+# ━━━ PART B: CIRCUIT BREAKER ━━━
+
+class CircuitState(Enum):
+    CLOSED = "CLOSED"        # Normal operation
+    OPEN = "OPEN"            # Failing — reject all calls
+    HALF_OPEN = "HALF_OPEN"  # Testing — allow one call
+
+@dataclass
+class CircuitBreakerConfig:
+    name: str
+    failure_threshold: int = 5
+    success_threshold: int = 2
+    timeout_seconds: float = 60.0
+    call_timeout_seconds: float = 30.0
+
+class CircuitBreaker:
+    def __init__(self, config: CircuitBreakerConfig):
+        self.config = config
+        self._state = CircuitState.CLOSED
+        self._failure_count = 0
+        self._success_count = 0
+        self._last_failure_time: datetime | None = None
+        self._lock = asyncio.Lock()
+    
+    @property
+    def state(self) -> CircuitState:
+        if self._state == CircuitState.OPEN and self._last_failure_time:
+            elapsed = (datetime.now(UTC) - self._last_failure_time).total_seconds()
+            if elapsed >= self.config.timeout_seconds:
+                return CircuitState.HALF_OPEN
+        return self._state
+    
+    async def call(self, func: Callable, *args, **kwargs) -> Any:
+        async with self._lock:
+            current_state = self.state
+            
+            if current_state == CircuitState.OPEN:
+                next_attempt_in = (
+                    self.config.timeout_seconds -
+                    (datetime.now(UTC) - self._last_failure_time).total_seconds()
+                )
+                raise CircuitOpenError(
+                    f"Circuit '{self.config.name}' is OPEN. Retry in {max(0, next_attempt_in):.0f}s"
+                )
+        
+        try:
+            # Enforce timeout at the circuit breaker level
+            result = await asyncio.wait_for(
+                func(*args, **kwargs),
+                timeout=self.config.call_timeout_seconds
+            )
+            await self._on_success()
+            return result
+            
+        except CircuitOpenError:
+            raise
+        except Exception as e:
+            await self._on_failure()
+            raise
+            
+    async def _on_success(self):
+        async with self._lock:
+            current_state = self.state
+            if current_state == CircuitState.HALF_OPEN:
+                self._success_count += 1
+                if self._success_count >= self.config.success_threshold:
+                    self._state = CircuitState.CLOSED
+                    self._failure_count = 0
+                    self._success_count = 0
+                    logger.info(f"Circuit '{self.config.name}' CLOSED (recovered)")
+            elif current_state == CircuitState.CLOSED:
+                self._failure_count = 0
+                
+    async def _on_failure(self):
+        async with self._lock:
+            self._failure_count += 1
+            self._success_count = 0
+            self._last_failure_time = datetime.now(UTC)
+            
+            current_state = self.state
+            if current_state == CircuitState.HALF_OPEN:
+                self._state = CircuitState.OPEN
+                logger.warning(f"Circuit '{self.config.name}' re-OPENED (test failed)")
+            elif current_state == CircuitState.CLOSED and self._failure_count >= self.config.failure_threshold:
+                self._state = CircuitState.OPEN
+                logger.error(f"Circuit '{self.config.name}' OPENED after {self._failure_count} failures")
+                
+    def force_close(self):
+        self._state = CircuitState.CLOSED
+        self._failure_count = 0
+        self._success_count = 0
+        logger.info(f"Circuit '{self.config.name}' force-closed by admin")
+
+    def get_status(self) -> dict:
+        return {
+            "name": self.config.name,
+            "state": self.state.value,
+            "failure_count": self._failure_count,
+            "failure_threshold": self.config.failure_threshold,
+            "last_failure": self._last_failure_time.isoformat() if self._last_failure_time else None,
+            "next_attempt_seconds": max(0, self.config.timeout_seconds - (
+                (datetime.now(UTC) - self._last_failure_time).total_seconds()
+            )) if self._state == CircuitState.OPEN and self._last_failure_time else None
+        }
+
+CIRCUIT_BREAKERS = {
+    "vision": CircuitBreaker(CircuitBreakerConfig("vision", failure_threshold=3, timeout_seconds=60, call_timeout_seconds=45)),
+    "nlp": CircuitBreaker(CircuitBreakerConfig("nlp", failure_threshold=5, timeout_seconds=30, call_timeout_seconds=20)),
+    "fusion": CircuitBreaker(CircuitBreakerConfig("fusion", failure_threshold=3, timeout_seconds=45, call_timeout_seconds=30)),
+    "rag": CircuitBreaker(CircuitBreakerConfig("rag", failure_threshold=5, timeout_seconds=60, call_timeout_seconds=30)),
+}

backend/tests/unit/test_resilience.py

@@ -0,0 +1,73 @@
+import pytest
+import asyncio
+from unittest.mock import AsyncMock
+from backend.orchestration.resilience import (
+    retry, RetryConfig, CircuitBreaker, CircuitBreakerConfig, CircuitState
+)
+from backend.core.exceptions import InferenceError, ValidationError, CircuitOpenError
+
+@pytest.mark.asyncio
+async def test_retry_succeeds_on_third_attempt():
+    mock_func = AsyncMock(side_effect=[InferenceError(), InferenceError(), "success"])
+    decorated = retry(RetryConfig(max_attempts=3, initial_delay_seconds=0.01))(mock_func)
+    
+    result = await decorated()
+    assert result == "success"
+    assert mock_func.call_count == 3
+
+@pytest.mark.asyncio
+async def test_retry_raises_after_max_attempts():
+    mock_func = AsyncMock(side_effect=InferenceError())
+    decorated = retry(RetryConfig(max_attempts=2, initial_delay_seconds=0.01))(mock_func)
+    
+    with pytest.raises(InferenceError):
+        await decorated()
+    assert mock_func.call_count == 2
+
+@pytest.mark.asyncio
+async def test_no_retry_on_excluded_exceptions():
+    mock_func = AsyncMock(side_effect=ValidationError())
+    decorated = retry(RetryConfig())(mock_func)
+    
+    with pytest.raises(ValidationError):
+        await decorated()
+    assert mock_func.call_count == 1 # Failed immediately, no retry
+
+@pytest.mark.asyncio
+async def test_circuit_opens_after_threshold():
+    cb = CircuitBreaker(CircuitBreakerConfig("test", failure_threshold=2, timeout_seconds=60))
+    mock_fail = AsyncMock(side_effect=InferenceError())
+    
+    with pytest.raises(InferenceError): await cb.call(mock_fail)
+    with pytest.raises(InferenceError): await cb.call(mock_fail)
+    
+    assert cb.state == CircuitState.OPEN
+    with pytest.raises(CircuitOpenError):
+        await cb.call(mock_fail)
+
+@pytest.mark.asyncio
+async def test_circuit_allows_one_call_in_half_open():
+    cb = CircuitBreaker(CircuitBreakerConfig("test", failure_threshold=1, timeout_seconds=0.1))
+    mock_fail = AsyncMock(side_effect=InferenceError())
+    mock_success = AsyncMock(return_value="success")
+    
+    with pytest.raises(InferenceError): await cb.call(mock_fail) # Opens circuit
+    assert cb.state == CircuitState.OPEN
+    
+    await asyncio.sleep(0.15) # Wait for timeout
+    assert cb.state == CircuitState.HALF_OPEN
+    
+    result = await cb.call(mock_success)
+    assert result == "success"
+
+@pytest.mark.asyncio
+async def test_circuit_reopens_on_half_open_failure():
+    cb = CircuitBreaker(CircuitBreakerConfig("test", failure_threshold=1, timeout_seconds=0.1))
+    mock_fail = AsyncMock(side_effect=InferenceError())
+    
+    with pytest.raises(InferenceError): await cb.call(mock_fail)
+    await asyncio.sleep(0.15)
+    
+    assert cb.state == CircuitState.HALF_OPEN
+    with pytest.raises(InferenceError): await cb.call(mock_fail)
+    assert cb.state == CircuitState.OPEN

backend/utils/cache.py

@@ -0,0 +1,117 @@
+import asyncio
+import time
+import hashlib
+import json
+import logging
+from collections import OrderedDict
+from typing import Generic, TypeVar, Callable
+from dataclasses import dataclass
+from functools import wraps
+
+logger = logging.getLogger(__name__)
+
+K = TypeVar("K")
+V = TypeVar("V")
+
+@dataclass
+class CacheEntry(Generic[V]):
+    value: V
+    created_at: float
+    hit_count: int = 0
+
+class LRUCache(Generic[K, V]):
+    """In-memory thread/async-safe Least Recently Used cache."""
+    def __init__(self, max_size: int, ttl_seconds: int, name: str = "cache"):
+        self._cache: OrderedDict[K, CacheEntry[V]] = OrderedDict()
+        self._max_size = max_size
+        self._ttl = ttl_seconds
+        self._name = name
+        self._lock = asyncio.Lock()
+        self._hits = 0
+        self._misses = 0
+    
+    async def get(self, key: K) -> V | None:
+        async with self._lock:
+            if key not in self._cache:
+                self._misses += 1
+                return None
+            
+            entry = self._cache[key]
+            
+            if time.monotonic() - entry.created_at > self._ttl:
+                del self._cache[key]
+                self._misses += 1
+                return None
+            
+            self._cache.move_to_end(key)
+            entry.hit_count += 1
+            self._hits += 1
+            return entry.value
+            
+    async def set(self, key: K, value: V):
+        async with self._lock:
+            if key in self._cache:
+                self._cache.move_to_end(key)
+            self._cache[key] = CacheEntry(value=value, created_at=time.monotonic())
+            
+            while len(self._cache) > self._max_size:
+                evicted_key, _ = self._cache.popitem(last=False)
+                logger.debug(f"Cache '{self._name}': evicted key {evicted_key}")
+                
+    async def invalidate(self, key: K):
+        async with self._lock:
+            self._cache.pop(key, None)
+            
+    async def clear(self):
+        async with self._lock:
+            self._cache.clear()
+            self._hits = 0
+            self._misses = 0
+            
+    def stats(self) -> dict:
+        total = self._hits + self._misses
+        return {
+            "name": self._name,
+            "size": len(self._cache),
+            "max_size": self._max_size,
+            "ttl_seconds": self._ttl,
+            "total_requests": total,
+            "hits": self._hits,
+            "misses": self._misses,
+            "hit_rate": round(self._hits / total, 3) if total > 0 else 0.0
+        }
+
+# ━━━ PART B: APPLICATION CACHES ━━━
+
+def _make_analysis_key(image_bytes: bytes, symptoms_text: str) -> str:
+    content = image_bytes + symptoms_text.encode()
+    return hashlib.md5(content).hexdigest()
+
+def _make_rag_key(query: str) -> str:
+    return hashlib.md5(query.lower().strip().encode()).hexdigest()
+
+# Singletons initialized here for app-wide use
+analysis_cache: LRUCache[str, dict] = LRUCache(max_size=50, ttl_seconds=3600, name="analysis")
+rag_cache: LRUCache[str, list] = LRUCache(max_size=200, ttl_seconds=1800, name="rag")
+user_cache: LRUCache[str, dict] = LRUCache(max_size=500, ttl_seconds=300, name="user")
+
+# ━━━ PART C: CACHE DECORATOR ━━━
+
+def cached(cache: LRUCache, key_fn: Callable):
+    """Decorator for async functions. Checks cache before executing."""
+    def decorator(func):
+        @wraps(func)
+        async def wrapper(*args, **kwargs):
+            key = key_fn(*args, **kwargs)
+            cached_value = await cache.get(key)
+            
+            if cached_value is not None:
+                logger.debug(f"Cache HIT for {func.__name__}: key={str(key)[:8]}...")
+                return cached_value
+                
+            logger.debug(f"Cache MISS for {func.__name__}: key={str(key)[:8]}...")
+            result = await func(*args, **kwargs)
+            await cache.set(key, result)
+            return result
+        return wrapper
+    return decorator