"""Shared base class for all Pattern-A JSON caches. Subclasses set three class variables and implement two abstract methods; ``load()`` and ``save()`` handle all disk I/O uniformly. Pattern A — mkstemp + replace ------------------------------ ``mkstemp`` gives each writer a process-unique temp file so two concurrent saves cannot interleave bytes. ``os.replace`` is atomic on POSIX — the final file is always a complete valid write, never a torn interleaving. Usage ----- :: class MyCache(MsgpackCache): _CACHE_FILENAME = "my.json" _CACHE_VERSION = 2 _TEMP_PREFIX = ".my_cache_" @classmethod def _deserialize_entries(cls, raw: _RawCacheMap) -> _RawCacheMap: # validate and return typed entries; skip invalid ones ... def _serialize_entries(self) -> _RawCacheMap: # return JSON-compatible dict of self._entries ... """ import json as _json import logging import os import pathlib import tempfile from abc import ABC, abstractmethod from muse.core.paths import muse_dir as _muse_dir logger = logging.getLogger(__name__) type _MsgVal = str | int | float | bool | None | list["_MsgVal"] | dict[str, "_MsgVal"] type _RawCacheMap = dict[str, _MsgVal] class MsgpackCache(ABC): """Abstract base for all Pattern-A JSON caches. Subclasses define ``_CACHE_FILENAME``, ``_CACHE_VERSION``, ``_TEMP_PREFIX`` and implement ``_deserialize_entries`` / ``_serialize_entries``. Everything else — ``load()``, ``save()``, ``get()``, ``put()``, ``prune()``, ``size``, ``empty()`` — is provided here. Attributes ---------- _cache_dir : pathlib.Path | None Absolute path to ``.muse/cache/``. ``None`` for in-memory-only instances (``empty()``). ``save()`` is a no-op when ``None``. _entries : dict The live in-memory entries dict. Subclass provides the typed form via ``_deserialize_entries`` on load and ``_serialize_entries`` on save. _dirty : bool Set to ``True`` by ``put()`` and ``prune()`` when entries change. Reset to ``False`` by a successful ``save()``. """ _CACHE_FILENAME: str _CACHE_VERSION: int = 1 _TEMP_PREFIX: str def __init__( self, cache_dir: pathlib.Path | None, entries: _RawCacheMap, ) -> None: self._cache_dir = cache_dir self._entries: _RawCacheMap = entries self._dirty = False # ------------------------------------------------------------------ # Abstract interface — subclasses implement these two methods only # ------------------------------------------------------------------ @classmethod @abstractmethod def _deserialize_entries(cls, raw: _RawCacheMap) -> _RawCacheMap: """Validate and convert raw JSON entries to typed entries. Called by ``load()`` with the ``"entries"`` value from the file. Invalid entries must be skipped — return only what is valid. A partially corrupt file must not poison valid entries. """ @abstractmethod def _serialize_entries(self) -> _RawCacheMap: """Convert ``self._entries`` to a JSON-serialisable dict. Called by ``save()`` to build the ``"entries"`` value written to disk. """ # ------------------------------------------------------------------ # Construction # ------------------------------------------------------------------ @classmethod def load(cls, muse_dir: pathlib.Path) -> "MsgpackCache": """Load from ``muse_dir / "cache" / cls._CACHE_FILENAME``. Returns an empty instance on any failure — missing file, corrupt bytes, version mismatch — and never raises. Invalid individual entries are skipped so a partially corrupt file does not poison the whole cache. Parameters ---------- muse_dir: Path to the ``.muse/`` directory of the repository root. """ cache_dir = muse_dir / "cache" cache_file = cache_dir / cls._CACHE_FILENAME if not cache_file.is_file(): return cls(cache_dir, {}) try: raw = cache_file.read_bytes() # Old binary msgpack files start with a byte > 0x7F. Treat as stale. if raw and raw[0] > 0x7F: logger.debug("⚠️ %s is old binary format — starting fresh", cls._CACHE_FILENAME) return cls(cache_dir, {}) doc = _json.loads(raw.decode("utf-8")) if not isinstance(doc, dict) or doc.get("version") != cls._CACHE_VERSION: logger.debug( "⚠️ %s version mismatch — starting fresh", cls._CACHE_FILENAME ) return cls(cache_dir, {}) raw_entries = doc.get("entries") if not isinstance(raw_entries, dict): return cls(cache_dir, {}) entries = cls._deserialize_entries(raw_entries) return cls(cache_dir, entries) except Exception as exc: # noqa: BLE001 logger.debug( "⚠️ %s unreadable (%s) — starting fresh", cls._CACHE_FILENAME, exc ) return cls(cache_dir, {}) @classmethod def empty(cls) -> "MsgpackCache": """Return a no-op instance for contexts without a ``.muse`` directory. ``save()`` is a no-op on instances returned by this method. """ return cls(None, {}) @classmethod def from_root(cls, root: pathlib.Path) -> "MsgpackCache": """Load for a repository root, returning ``empty()`` when ``.muse/`` is absent. Convenience alternative to calling ``load()`` directly. Handles the common caller pattern of "give me a cache for this repo root, or a no-op cache if it is not a Muse repository": .. code-block:: python cache = MyCache.from_root(root) # replaces: # _dir = muse_dir(root) # if _dir.is_dir(): # cache = MyCache.load(_dir) # else: # cache = MyCache.empty() Parameters ---------- root: Repository root (the directory that contains ``.muse/``). """ _dir = _muse_dir(root) if _dir.is_dir(): return cls.load(_dir) return cls.empty() # ------------------------------------------------------------------ # Data access # ------------------------------------------------------------------ def get(self, key: str) -> _MsgVal: """Return the cached value for *key*, or ``None`` on miss.""" return self._entries.get(key) def put(self, key: str, value: _MsgVal) -> None: """Store *value* under *key* and mark the cache dirty.""" self._entries[key] = value self._dirty = True def prune(self, live_ids: set[str]) -> None: """Remove entries whose keys are not in *live_ids*. Sets ``_dirty`` only when at least one entry is removed. """ stale = set(self._entries) - live_ids if stale: for k in stale: del self._entries[k] self._dirty = True logger.debug( "🗑️ %s pruned %d stale entries", self.__class__.__name__, len(stale), ) @property def size(self) -> int: """Number of cached entries.""" return len(self._entries) # ------------------------------------------------------------------ # Persistence # ------------------------------------------------------------------ def save(self) -> None: """Atomically persist the cache to disk if it has changed. Uses ``mkstemp`` (``_TEMP_PREFIX``) for a process-unique temp file so two concurrent writers cannot interleave bytes, then ``os.replace`` for an atomic rename. Silently skips when ``_dirty`` is ``False`` or ``_cache_dir`` is ``None``. """ if not self._dirty or self._cache_dir is None: return self._cache_dir.mkdir(parents=True, exist_ok=True) doc = {"version": self._CACHE_VERSION, "entries": self._serialize_entries()} payload = _json.dumps(doc, ensure_ascii=False, separators=(",", ":")).encode("utf-8") cache_file = self._cache_dir / self._CACHE_FILENAME fd, tmp_path = tempfile.mkstemp( dir=self._cache_dir, prefix=self._TEMP_PREFIX, suffix=".tmp" ) try: with os.fdopen(fd, "wb") as fh: fh.write(payload) fh.flush() os.fsync(fh.fileno()) except Exception: try: os.unlink(tmp_path) except OSError: pass raise try: os.replace(tmp_path, cache_file) self._dirty = False logger.debug( "✅ %s saved (%d entries)", self.__class__.__name__, len(self._entries), ) except OSError as exc: logger.warning("⚠️ %s save failed: %s", self.__class__.__name__, exc)