"""A resilient rate-limit store: a `limits` sliding-window-counter limiter over Redis (primary). Falls back to an in-memory limiter, guarded by the hardening CircuitBreaker. NEVER-500: any exception on the Redis path is caught and served by the in-memory fallback, so Redis being slow/down/buggy never 500s a request. Only Redis/OSError count toward the breaker. memory:// is unchanged: build_rate_limiter returns the plain strategy behind a trivial factory. """ import logging import threading import time from typing import Callable, Protocol, TypeVar from limits import RateLimitItem from limits.storage import MemoryStorage, storage_from_string from limits.strategies import STRATEGIES, RateLimiter from limits.util import WindowStats from core.hardening import observability from core.hardening.breaker import CircuitBreaker, CircuitBreakerError from core.hardening.rate_policy import RATE_STRATEGY logger = logging.getLogger(__name__) _T = TypeVar('_T') # Bound the redis-py connection pool so a recovery reconnect storm (15 threads x M pods) can't # spike connections; > the 15-thread pool with headroom. limits forwards **options to the client. _REDIS_MAX_CONNECTIONS = 32 try: import redis _REDIS_ERRORS: tuple[type[BaseException], ...] = ( redis.exceptions.RedisError, OSError, ) except ImportError: # pragma: no cover -- redis client absent (memory://-only deploy) _REDIS_ERRORS = (OSError,) # Config-minimal: not env-tunable. _REDIS_SOCKET_TIMEOUT_S = 0.05 # in-VPC RTT is sub-ms; 50ms caps brownout latency _REDIS_CONNECT_TIMEOUT_S = 0.05 _BREAKER_FAIL_MAX = 5 # 5 consecutive FAILING REQUESTS (the per-request latch means # each request contributes at most one breaker failure). _BREAKER_RESET_TIMEOUT_S = 10.0 _BREAKER_SUCCESS_THRESHOLD = 2 _OTHER_ERROR_LOG_INTERVAL_S = 60.0 # throttle the non-Redis-error ERROR log _other_error_lock = threading.Lock() _other_error_last_log = 0.0 def _is_redis_failure(exc: BaseException) -> bool: return isinstance(exc, _REDIS_ERRORS) def _log_other_error(exc: BaseException) -> None: """Log a non-Redis primary error at most once per interval (the counter carries the rate).""" global _other_error_last_log now = time.monotonic() with _other_error_lock: if now - _other_error_last_log < _OTHER_ERROR_LOG_INTERVAL_S: return _other_error_last_log = now logger.error( 'rate-limit Redis primary raised %s; serving from in-memory fallback', type(exc).__name__, exc_info=True, ) class RequestLimiter(Protocol): """The per-request limiter surface shared by the primary and fallback strategies.""" def hit(self, item: RateLimitItem, *identifiers: str) -> bool: """Consume one unit against item for identifiers, returning whether it was admitted.""" ... def get_window_stats(self, item: RateLimitItem, *identifiers: str) -> WindowStats: """Return the current window stats for item and identifiers without consuming a unit.""" ... class RateLimiterFactory(Protocol): """Builds a request-scoped RequestLimiter.""" def for_request(self) -> RequestLimiter: """Return a RequestLimiter scoped to the current request.""" ... class _PlainFactory: """memory:// factory: for_request() returns the shared, stateless strategy (unchanged).""" def __init__(self, strategy: RateLimiter) -> None: self._strategy = strategy def for_request(self) -> RateLimiter: return self._strategy class _ResilientRequest: """One request's view. Latches to the fallback on the first primary failure (instance state, request-scoped by construction -- no flask.g, works with or without an app context). """ def __init__( self, primary: RateLimiter, fallback: RateLimiter, breaker: CircuitBreaker ) -> None: """Store the shared primary/fallback strategies and breaker for this request.""" self._primary = primary self._fallback = fallback self._breaker = breaker self._degraded = False def _via( self, primary_call: Callable[[], _T], fallback_call: Callable[[], _T], fail_open: Callable[[], _T], ) -> _T: if not self._degraded: try: return self._breaker.call(primary_call) except CircuitBreakerError: self._degraded = True observability.ratelimit_fallback('breaker_open') except ( Exception ) as exc: # NEVER-500: any primary error -> fallback # noqa: BLE001 self._degraded = True if _is_redis_failure(exc): observability.ratelimit_fallback('redis_error') else: observability.ratelimit_fallback('other_error') _log_other_error(exc) # Degraded: serve from the in-memory fallback. Guard it too -- if even the fallback raises # (a genuine logic bug), fail OPEN rather than 500 (a rate limiter's failure bias is # availability). This makes the NEVER-500 contract absolute. try: return fallback_call() except Exception: # noqa: BLE001 -- last resort: never 500 on the limiter observability.ratelimit_fallback('fallback_error') logger.error( 'rate-limit in-memory fallback raised; failing open', exc_info=True ) return fail_open() def hit(self, item: RateLimitItem, *identifiers: str) -> bool: return self._via( lambda: self._primary.hit(item, *identifiers), lambda: self._fallback.hit(item, *identifiers), lambda: True, # fail open: admit the request ) def get_window_stats(self, item: RateLimitItem, *identifiers: str) -> WindowStats: return self._via( lambda: self._primary.get_window_stats(item, *identifiers), lambda: self._fallback.get_window_stats(item, *identifiers), lambda: WindowStats( int(time.time()) + item.get_expiry(), item.amount ), # permissive ) class ResilientRateLimiter: """Shared factory holding the shared primary/fallback strategies and the shared breaker.""" def __init__( self, primary: RateLimiter, fallback: RateLimiter, breaker: CircuitBreaker ) -> None: """Store the shared primary/fallback strategies and breaker.""" self._primary = primary self._fallback = fallback self._breaker = breaker def for_request(self) -> _ResilientRequest: """Return a new _ResilientRequest sharing this factory's primary/fallback/breaker.""" return _ResilientRequest(self._primary, self._fallback, self._breaker) def build_rate_limiter(uri: str) -> RateLimiterFactory: """Return a rate-limiter factory for the storage URI. memory:// (and any non-redis/valkey scheme) -> plain strategy, unchanged. redis*/valkey* -> ResilientRateLimiter; on any construction error, log CRITICAL, count a boot_error fallback, and degrade to the plain in-memory strategy (never abort startup). """ if not uri.startswith(('redis', 'valkey')): return _PlainFactory(STRATEGIES[RATE_STRATEGY](storage_from_string(uri))) try: redis_storage = storage_from_string( uri, socket_timeout=_REDIS_SOCKET_TIMEOUT_S, socket_connect_timeout=_REDIS_CONNECT_TIMEOUT_S, max_connections=_REDIS_MAX_CONNECTIONS, ) primary = STRATEGIES[RATE_STRATEGY](redis_storage) fallback = STRATEGIES[RATE_STRATEGY](MemoryStorage()) breaker = CircuitBreaker( name='ratelimit-redis', fail_max=_BREAKER_FAIL_MAX, reset_timeout=_BREAKER_RESET_TIMEOUT_S, success_threshold=_BREAKER_SUCCESS_THRESHOLD, count_failure=_is_redis_failure, on_state_change=observability.on_breaker_state, ) return ResilientRateLimiter(primary, fallback, breaker) except Exception: # noqa: BLE001 -- one-shot boot path; never abort startup logger.critical( 'rate-limit Redis store construction failed for %r; falling back to in-memory ' '(per-process) limiting until redeploy', uri, exc_info=True, ) observability.ratelimit_fallback('boot_error') return _PlainFactory(STRATEGIES[RATE_STRATEGY](MemoryStorage()))