FortyDN Dev
v3.0.0: fully autonomous engine — remove MillionVerifier, add DoH DNS worker, AI risk scorer (HF PRO, circuit breaker), i18n, normalized verdicts, remote SMTP worker
de745bf | import time | |
| import asyncio | |
| from typing import Optional | |
| from ..config import settings | |
| # Redis is optional at import-time so the API runs without it. | |
| try: | |
| import redis.asyncio as aioredis | |
| _HAS_REDIS = True | |
| except Exception: # pragma: no cover - redis not installed | |
| aioredis = None | |
| _HAS_REDIS = False | |
| _redis_client: Optional[object] = None | |
| # In-memory TTL fallback cache: {key: (value, expires_at)} | |
| _memory_cache: dict = {} | |
| class CircuitBreaker: | |
| """Fail fast to the in-memory cache when Redis is slow or down. | |
| Without this, `validate_batch` pays Redis' socket timeout (~2s) on every one | |
| of the ~8 cache calls it makes per row — measured at 10.0s per row on a | |
| host where the Upstash link is slow, while the engine itself needs ~70ms. | |
| After `failure_threshold` consecutive slow/failed ops the circuit opens and | |
| calls skip Redis entirely until `cooldown_seconds` elapse, then probe once. | |
| """ | |
| def __init__(self, failure_threshold: int = 3, | |
| cooldown_seconds: float = 30.0, slow_threshold: float = 1.0): | |
| self.failure_threshold = failure_threshold | |
| self.cooldown_seconds = cooldown_seconds | |
| self.slow_threshold = slow_threshold | |
| self._failures = 0 | |
| self._open_until = 0.0 | |
| def allow(self) -> bool: | |
| """True when Redis may be contacted (closed, or cooldown elapsed).""" | |
| return time.time() >= self._open_until | |
| def record(self, success: bool, duration: float) -> None: | |
| """Record one op: slow ops count as failures even when they succeed.""" | |
| if success and duration < self.slow_threshold: | |
| self._failures = 0 | |
| return | |
| self._failures += 1 | |
| if self._failures >= self.failure_threshold: | |
| self._open_until = time.time() + self.cooldown_seconds | |
| self._failures = 0 | |
| def reset(self) -> None: | |
| self._failures = 0 | |
| self._open_until = 0.0 | |
| # Trips when Redis is consistently slow/failing; see CircuitBreaker. | |
| _circuit = CircuitBreaker() | |
| async def get_redis(): | |
| """Lazily create a shared Redis client (returns None if unavailable). | |
| Redis connections must be configured explicitly. Missing or unreachable | |
| Redis leaves the API on its bounded in-memory cache instead of silently | |
| provisioning a temporary external database. | |
| """ | |
| global _redis_client | |
| if not _HAS_REDIS: | |
| return None | |
| if _redis_client is None: | |
| url = (settings.redis_url or "").strip() | |
| if not url: | |
| return None | |
| try: | |
| _redis_client = aioredis.from_url( | |
| url, | |
| encoding="utf-8", | |
| decode_responses=True, | |
| # The TLS handshake to Upstash measures ~750-850 ms on some | |
| # networks; a 2 s cap killed the very first PING at exactly | |
| # the timeout (observed live), which made every cache op pay | |
| # the full budget and trip the breaker. 5 s connects reliably | |
| # while the breaker still fails fast on real outages. | |
| socket_connect_timeout=5, | |
| socket_timeout=5, | |
| ) | |
| await _redis_client.ping() | |
| except Exception: | |
| _redis_client = None | |
| return _redis_client | |
| def _memory_get(key: str) -> Optional[str]: | |
| item = _memory_cache.get(key) | |
| if item is None: | |
| return None | |
| value, expires = item | |
| if time.time() > expires: | |
| _memory_cache.pop(key, None) | |
| return None | |
| return value | |
| def _memory_set(key: str, value: str, ttl: int): | |
| _memory_cache[key] = (value, time.time() + ttl) | |
| # keep cache bounded | |
| if len(_memory_cache) > 20000: | |
| now = time.time() | |
| for k in [k for k, (_, exp) in _memory_cache.items() if exp < now]: | |
| _memory_cache.pop(k, None) | |
| async def get_cache(key: str) -> Optional[str]: | |
| if _circuit.allow(): | |
| t0 = time.perf_counter() | |
| try: | |
| client = await get_redis() | |
| if client is not None: | |
| value = await client.get(key) | |
| _circuit.record(success=True, duration=time.perf_counter() - t0) | |
| return value | |
| # Redis is configured but get_redis() could not establish a client | |
| # (connect/ping failed). Without this the 2s timeout is paid on | |
| # EVERY call and the breaker never trips. | |
| if (settings.redis_url or "").strip(): | |
| _circuit.record(success=False, | |
| duration=time.perf_counter() - t0) | |
| except Exception: | |
| _circuit.record(success=False, duration=time.perf_counter() - t0) | |
| return _memory_get(key) | |
| async def set_cache(key: str, value: str, ttl: int = 86400): | |
| if _circuit.allow(): | |
| t0 = time.perf_counter() | |
| try: | |
| client = await get_redis() | |
| if client is not None: | |
| await client.setex(key, ttl, value) | |
| _circuit.record(success=True, duration=time.perf_counter() - t0) | |
| return | |
| if (settings.redis_url or "").strip(): | |
| _circuit.record(success=False, | |
| duration=time.perf_counter() - t0) | |
| except Exception: | |
| _circuit.record(success=False, duration=time.perf_counter() - t0) | |
| _memory_set(key, value, ttl) | |
| async def redis_available(timeout: float = 4.0) -> bool: | |
| """ | |
| Quick non-blocking check: can we talk to Redis? | |
| Used to gate Celery .delay() calls so a missing Redis broker fails | |
| fast instead of entering Celery's infinite reconnect/retry loop. | |
| """ | |
| if not _HAS_REDIS: | |
| return False | |
| try: | |
| client = await asyncio.wait_for( | |
| get_redis(), timeout=timeout | |
| ) | |
| if client is None: | |
| return False | |
| await asyncio.wait_for(client.ping(), timeout=timeout) | |
| return True | |
| except asyncio.TimeoutError: | |
| print("[REDIS] Connection timeout - Redis may not be running") | |
| return False | |
| except Exception as e: | |
| print(f"[REDIS] Connection failed: {str(e)}") | |
| return False | |