diff --git a/core/video/enrichment/cache.py b/core/video/enrichment/cache.py new file mode 100644 index 00000000..6a69f9b0 --- /dev/null +++ b/core/video/enrichment/cache.py @@ -0,0 +1,50 @@ +"""A small thread-safe TTL + LRU cache for the video enrichment engine. + +The engine is a process-wide singleton hit concurrently by Flask request threads +AND the worker threads, so the cache must be locked. Entries expire after a TTL; +when the cache is full the LEAST-RECENTLY-USED entry is evicted (not the whole +cache wholesale). Isolated: imports only the stdlib; no music, no DB. +""" + +from __future__ import annotations + +import threading +import time +from collections import OrderedDict + + +class TTLCache: + def __init__(self, maxsize: int = 256, ttl: float = 1800.0, clock=time.monotonic): + self._max = max(1, int(maxsize)) + self._ttl = float(ttl) + self._clock = clock # injectable for tests + self._lock = threading.Lock() + self._data: OrderedDict = OrderedDict() # key -> (expires_at, value) + + def get(self, key): + now = self._clock() + with self._lock: + hit = self._data.get(key) + if hit is None: + return None + if hit[0] <= now: # expired + del self._data[key] + return None + self._data.move_to_end(key) # mark recently used + return hit[1] + + def put(self, key, value, ttl: float | None = None) -> None: + expires_at = self._clock() + (self._ttl if ttl is None else float(ttl)) + with self._lock: + self._data[key] = (expires_at, value) + self._data.move_to_end(key) + while len(self._data) > self._max: # evict LRU (oldest), not everything + self._data.popitem(last=False) + + def clear(self) -> None: + with self._lock: + self._data.clear() + + def __len__(self) -> int: + with self._lock: + return len(self._data) diff --git a/core/video/enrichment/engine.py b/core/video/enrichment/engine.py index cf5409e7..849f7817 100644 --- a/core/video/enrichment/engine.py +++ b/core/video/enrichment/engine.py @@ -11,6 +11,7 @@ import threading from utils.logging_config import get_logger +from .cache import TTLCache from .worker import VideoEnrichmentWorker logger = get_logger("video_enrichment.engine") @@ -28,25 +29,20 @@ class VideoEnrichmentEngine: # OMDb ratings (IMDb/RT/Metacritic) — not a matcher, so not a worker; # backfilled on the lazy detail refresh. self.ratings_client = ratings_client - # Short-TTL cache for live TMDB detail extras / preview payloads, so - # re-opening a title is instant instead of re-hitting TMDB. - self._tmdb_cache = {} + # Thread-safe TTL+LRU cache for live TMDB detail extras / preview payloads / + # person pages / seasons / trending, so re-opening a title is instant + # instead of re-hitting TMDB. (Volatile by design — durable art/episodes/ + # ratings live in video.db; we don't persist this tier.) + self._cache = TTLCache(maxsize=256, ttl=1800) # Restore each worker's persisted pause state (survives restart). for w in self.workers.values(): w.restore_paused() def _cache_get(self, key): - import time - hit = self._tmdb_cache.get(key) - if hit and hit[0] > time.monotonic(): - return hit[1] - return None + return self._cache.get(key) def _cache_put(self, key, data, ttl=1800): - import time - if len(self._tmdb_cache) > 256: # cheap bound — clear wholesale - self._tmdb_cache.clear() - self._tmdb_cache[key] = (time.monotonic() + ttl, data) + self._cache.put(key, data, ttl=ttl) def _backfill_ratings(self, kind, item_id): # The OMDb worker owns the ratings client (fallback to an injected one @@ -263,11 +259,17 @@ class VideoEnrichmentEngine: w = self.workers.get("tmdb") if not w or not w.enabled: return [] - try: - results = w.client.search(query) or [] - except Exception: - logger.exception("video search failed for %r", query) - return [] + # Short TTL — identical queries within ~a minute reuse the result; ownership + # is re-stamped fresh below so 'In Library' badges stay current. + key = ("search", (query or "").strip().lower()) + results = self._cache_get(key) + if results is None: + try: + results = w.client.search(query) or [] + self._cache_put(key, results, ttl=60) + except Exception: + logger.exception("video search failed for %r", query) + return [] for r in results: if r.get("kind") in ("movie", "show") and r.get("tmdb_id"): r["library_id"] = self.db.library_id_for_tmdb(r["kind"], r["tmdb_id"]) diff --git a/tests/test_video_cache.py b/tests/test_video_cache.py new file mode 100644 index 00000000..6782dd4c --- /dev/null +++ b/tests/test_video_cache.py @@ -0,0 +1,61 @@ +"""Seam tests for the video enrichment TTL+LRU cache (importable core/ module).""" + +from __future__ import annotations + +import threading + +from core.video.enrichment.cache import TTLCache + + +def test_get_miss_and_hit(): + c = TTLCache(maxsize=4, ttl=100) + assert c.get("a") is None + c.put("a", 1) + assert c.get("a") == 1 + + +def test_ttl_expiry_with_injected_clock(): + t = {"now": 0.0} + c = TTLCache(maxsize=4, ttl=10, clock=lambda: t["now"]) + c.put("a", 1) + t["now"] = 9.9 + assert c.get("a") == 1 # still fresh + t["now"] = 10.1 + assert c.get("a") is None # expired → evicted + assert len(c) == 0 + + +def test_per_put_ttl_override(): + t = {"now": 0.0} + c = TTLCache(ttl=1000, clock=lambda: t["now"]) + c.put("a", 1, ttl=5) + t["now"] = 6 + assert c.get("a") is None + + +def test_lru_eviction_not_wholesale(): + c = TTLCache(maxsize=2, ttl=100) + c.put("a", 1) + c.put("b", 2) + assert c.get("a") == 1 # touch 'a' → most-recently-used + c.put("c", 3) # over capacity → evict LRU ('b'), NOT everything + assert c.get("a") == 1 # survived + assert c.get("c") == 3 # survived + assert c.get("b") is None # the only one evicted + assert len(c) == 2 + + +def test_thread_safe_under_concurrency(): + c = TTLCache(maxsize=64, ttl=100) + + def worker(n): + for i in range(500): + c.put(("k", (n * 500 + i) % 80), i) + c.get(("k", i % 80)) + + threads = [threading.Thread(target=worker, args=(n,)) for n in range(8)] + for t in threads: + t.start() + for t in threads: + t.join() + assert len(c) <= 64 # bound held, no crash/corruption