cache.py

from __future__ import annotations

import threading
import time
from collections import OrderedDict
from collections.abc import Callable
from dataclasses import dataclass
from typing import Any

MAX_KEY_LENGTH = 512
MAX_TTL_SECONDS = 31_536_000


class VersionConflict(RuntimeError):
 """Raised when an optimistic cache update targets a stale version."""


@dataclass(frozen=True)
class CacheEntry:
 value: Any
 expires_at: float | None
 version: int


class SkyCache:
 """Thread-safe bounded single-process TTL/LRU cache with optimistic writes."""

 def __init__(
 self,
 default_ttl_seconds: float | None = None,
 max_entries: int = 10_000,
 *,
 clock: Callable[[], float] = time.monotonic,
 ) -> None:
 self._validate_ttl(default_ttl_seconds)
 if not 1 <= max_entries <= 1_000_000:
 raise ValueError("max_entries must be between 1 and 1,000,000")
 self._default_ttl = default_ttl_seconds
 self._max_entries = max_entries
 self._clock = clock
 self._entries: OrderedDict[str, CacheEntry] = OrderedDict()
 self._lock = threading.RLock()
 self._version = 0
 self._stats = {
 "hits": 0,
 "misses": 0,
 "sets": 0,
 "deletes": 0,
 "evictions": 0,
 "expirations": 0,
 "cas_conflicts": 0,
 }

 @staticmethod
 def _validate_key(key: str) -> str:
 if not isinstance(key, str):
 raise TypeError("cache key must be a string")
 if not key or key.isspace():
 raise ValueError("cache key is required")
 if len(key) > MAX_KEY_LENGTH:
 raise ValueError(f"cache key cannot exceed {MAX_KEY_LENGTH} characters")
 return key

 @staticmethod
 def _validate_ttl(ttl_seconds: float | None) -> None:
 if ttl_seconds is None:
 return
 if not isinstance(ttl_seconds, (int, float)):
 raise TypeError("ttl_seconds must be numeric")
 if ttl_seconds < 0 or ttl_seconds > MAX_TTL_SECONDS:
 raise ValueError(
 f"ttl_seconds must be between 0 and {MAX_TTL_SECONDS} seconds"
 )

 def _next_version(self) -> int:
 self._version += 1
 return self._version

 def _expires_at(self, ttl_seconds: float | None) -> float | None:
 ttl = self._default_ttl if ttl_seconds is None else ttl_seconds
 self._validate_ttl(ttl)
 return None if ttl is None else self._clock() + ttl

 def _remove_if_expired(self, key: str, now: float) -> bool:
 entry = self._entries.get(key)
 if entry is None or entry.expires_at is None or now < entry.expires_at:
 return False
 del self._entries[key]
 self._stats["expirations"] += 1
 return True

 def _evict_if_needed(self) -> None:
 while len(self._entries) > self._max_entries:
 self._entries.popitem(last=False)
 self._stats["evictions"] += 1

 def set(self, key: str, value: Any, ttl_seconds: float | None = None) -> int:
 key = self._validate_key(key)
 expires_at = self._expires_at(ttl_seconds)
 with self._lock:
 version = self._next_version()
 self._entries[key] = CacheEntry(value=value, expires_at=expires_at, version=version)
 self._entries.move_to_end(key)
 self._stats["sets"] += 1
 self._evict_if_needed()
 return version

 def set_if_absent(
 self,
 key: str,
 value: Any,
 ttl_seconds: float | None = None,
 ) -> tuple[bool, int]:
 key = self._validate_key(key)
 with self._lock:
 self._remove_if_expired(key, self._clock())
 existing = self._entries.get(key)
 if existing is not None:
 return False, existing.version
 return True, self.set(key, value, ttl_seconds)

 def compare_and_set(
 self,
 key: str,
 value: Any,
 expected_version: int,
 ttl_seconds: float | None = None,
 ) -> int:
 key = self._validate_key(key)
 if expected_version < 1:
 raise ValueError("expected_version must be positive")
 with self._lock:
 self._remove_if_expired(key, self._clock())
 existing = self._entries.get(key)
 if existing is None or existing.version != expected_version:
 self._stats["cas_conflicts"] += 1
 raise VersionConflict("cache entry version does not match")
 return self.set(key, value, ttl_seconds)

 def get_entry(self, key: str) -> CacheEntry | None:
 key = self._validate_key(key)
 with self._lock:
 now = self._clock()
 if self._remove_if_expired(key, now):
 self._stats["misses"] += 1
 return None
 entry = self._entries.get(key)
 if entry is None:
 self._stats["misses"] += 1
 return None
 self._entries.move_to_end(key)
 self._stats["hits"] += 1
 return entry

 def get(self, key: str, default: Any = None) -> Any:
 entry = self.get_entry(key)
 return default if entry is None else entry.value

 def delete(self, key: str) -> bool:
 key = self._validate_key(key)
 with self._lock:
 removed = self._entries.pop(key, None) is not None
 if removed:
 self._stats["deletes"] += 1
 return removed

 def purge_expired(self) -> int:
 with self._lock:
 now = self._clock()
 expired = [
 key
 for key, entry in self._entries.items()
 if entry.expires_at is not None and now >= entry.expires_at
 ]
 for key in expired:
 del self._entries[key]
 self._stats["expirations"] += len(expired)
 return len(expired)

 def clear(self) -> int:
 with self._lock:
 count = len(self._entries)
 self._entries.clear()
 self._stats["deletes"] += count
 return count

 def stats(self) -> dict[str, int | float]:
 with self._lock:
 self.purge_expired()
 result: dict[str, int | float] = dict(self._stats)
 result["size"] = len(self._entries)
 result["max_entries"] = self._max_entries
 requests = self._stats["hits"] + self._stats["misses"]
 result["hit_rate"] = (
 self._stats["hits"] / requests if requests else 0.0
 )
 return result

 def __len__(self) -> int:
 with self._lock:
 self.purge_expired()
 return len(self._entries)


# Backward-compatible alias. The implementation is deliberately single-node;
# the historical repository name must not be interpreted as distributed consensus.
DistributedCache = SkyCache