From 7e400658688b3514a887c3e142d7e76da5e253da Mon Sep 17 00:00:00 2001 From: Ray Walker Date: Sun, 19 Jul 2026 01:58:27 +1000 Subject: [PATCH 1/3] fix(decorators): honor L1CacheConfig SWR + max_size_mb in L1-only mode (LAB-106) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit With backend=None the decorator routed through ObjectCache(max_entries=256), which ignored every L1CacheConfig field: swr_enabled/swr_threshold_ratio never scheduled a background refresh (L1Cache.get_with_swr had zero callers anywhere in src/) and max_size_mb was dead — the store was entry-count-bounded. The docs presented all of it as working config for exactly this mode (#207). ObjectCache learns the missing semantics instead of rerouting through L1Cache, because L1Cache stores bytes only and would destroy L1-only mode's documented type preservation (raw tuples/sets/frozensets, no serialization): - max_size_bytes bound with LRU eviction by estimated bytes (iterative object-graph estimator with cycle detection; oversized values are returned but never cached, and a smaller stale entry under the same key is dropped). - get_with_swr/complete_refresh/cancel_refresh with per-key version tokens so a refresh completing after delete()/clear() cannot resurrect stale data (same anti-resurrection design as L1Cache). With no L2 source of truth, a successful refresh restarts both the freshness clock and the TTL. The decorator serves the stale value immediately and refreshes in the background: asyncio.create_task for async functions (strong task refs held — asyncio only keeps weak ones), a daemon thread for sync functions. Raising for sync was not viable because swr_enabled defaults to True and would have broken every existing sync @cache(backend=None) user at decoration time. SWR requires a ttl (staleness is undefined without one). Refresh failures log at debug and the next qualifying hit retries. L1CacheConfig.validate() now rejects swr_threshold_ratio outside (0.0, 1.0]. @cache.local() is unchanged (entry bound only, no sizing cost, no SWR). Fixes #207 Co-authored-by: multica-agent --- docs/configuration.md | 32 +++ src/cachekit/config/nested.py | 15 +- src/cachekit/decorators/wrapper.py | 78 ++++++- src/cachekit/object_cache.py | 339 +++++++++++++++++++++++++---- tests/unit/test_l1_only_swr.py | 242 ++++++++++++++++++++ tests/unit/test_object_cache.py | 210 +++++++++++++++++- 6 files changed, 865 insertions(+), 51 deletions(-) create mode 100644 tests/unit/test_l1_only_swr.py diff --git a/docs/configuration.md b/docs/configuration.md index f5b9578..069483f 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -237,6 +237,38 @@ def my_function(): - **Expiry**: Hard deadline when entry is deleted from cache - **Namespace**: Logical grouping for bulk invalidation (see [L1 Invalidation Guide](features/l1-invalidation.md)) +### L1-Only Mode (`backend=None`) + +With `backend=None` the decorator caches raw Python objects in process memory (no +serialization — tuples, sets, and frozensets keep their types). `L1CacheConfig` is +honored as follows: + +- **`max_size_mb`** bounds the cache by *estimated bytes*, not entry count. Sizes of + raw objects are estimated best-effort (builtin containers are walked recursively; + other objects are counted via `sys.getsizeof`). A single value larger than the whole + budget is returned to the caller but never cached. +- **SWR requires a `ttl`.** With `swr_enabled=True` and a `ttl` set, a cache hit past + `ttl * swr_threshold_ratio` (±10% jitter) serves the cached value immediately and + refreshes it in the background — via `asyncio.create_task` for `async def` functions, + or a daemon thread for sync functions. A successful refresh restarts both the + freshness clock and the TTL. With `ttl=None` entries never go stale, so no refresh + is ever scheduled. +- **Refresh failures are non-fatal**: the stale value keeps being served until hard + expiry, and the next qualifying hit retries the refresh. + +```python notest +import asyncio +from cachekit import cache +from cachekit.config import L1CacheConfig + +@cache(ttl=60, backend=None, l1=L1CacheConfig(swr_enabled=True, swr_threshold_ratio=0.5)) +async def load_dashboard(): + return await fetch_expensive_data() # illustrative - not defined + +# After 30s (50% of TTL), the next call returns the cached value instantly +# and schedules a background refresh — callers never block on revalidation. +``` + ### Intent Presets Use intent presets to configure L1 and other features for different use cases: diff --git a/src/cachekit/config/nested.py b/src/cachekit/config/nested.py index ac8f892..636da38 100644 --- a/src/cachekit/config/nested.py +++ b/src/cachekit/config/nested.py @@ -22,7 +22,15 @@ class L1CacheConfig: Attributes: enabled: Enable L1 in-memory cache (default: True) - max_size_mb: Maximum L1 cache size in megabytes (default: 100) + max_size_mb: Maximum L1 cache size in megabytes (default: 100). In L1-only + mode (backend=None) this is a best-effort byte bound on raw object sizes. + swr_enabled: Enable stale-while-revalidate background refresh (default: True). + Requires a ttl; in L1-only mode async functions refresh via an asyncio + task and sync functions via a daemon thread. + swr_threshold_ratio: Fraction of TTL after which a hit triggers a background + refresh, in (0.0, 1.0] (default: 0.5) + invalidation_enabled: Enable invalidation event broadcasts (default: True) + namespace_index: Enable fast namespace-based invalidation (default: True) Examples: Create with defaults: @@ -57,10 +65,13 @@ def validate(self) -> None: """Validate L1 cache configuration. Raises: - ConfigurationError: If max_size_mb < 1 + ConfigurationError: If max_size_mb < 1 or swr_threshold_ratio is + outside (0.0, 1.0] """ if self.max_size_mb < 1: raise ConfigurationError(f"L1 max_size_mb must be >= 1, got {self.max_size_mb}") + if not (0.0 < self.swr_threshold_ratio <= 1.0): + raise ConfigurationError(f"L1 swr_threshold_ratio must be in (0.0, 1.0], got {self.swr_threshold_ratio}") @dataclass(frozen=True) diff --git a/src/cachekit/decorators/wrapper.py b/src/cachekit/decorators/wrapper.py index 9365b28..8b0f53a 100644 --- a/src/cachekit/decorators/wrapper.py +++ b/src/cachekit/decorators/wrapper.py @@ -489,7 +489,25 @@ def create_cache_wrapper( # L1-only mode: use ObjectCache for raw Python object storage (no serialization). # This preserves types (tuples, sets, frozensets) that MessagePack would degrade. - _object_cache: ObjectCache | None = ObjectCache(max_entries=256) if _l1_only_mode else None + # L1CacheConfig is honored here (#207): max_size_mb bounds bytes (best-effort + # object-graph estimate, not entry count) and swr_enabled/swr_threshold_ratio + # drive background refresh via get_with_swr. + from ..config.nested import L1CacheConfig + + _l1_config: L1CacheConfig = config.l1 if config is not None else L1CacheConfig() + _object_cache: ObjectCache | None = ( + ObjectCache( + max_entries=None, + max_size_bytes=_l1_config.max_size_mb * 1024 * 1024, + swr_threshold_ratio=_l1_config.swr_threshold_ratio, + ) + if _l1_only_mode + else None + ) + + # SWR needs a TTL: freshness is measured against ttl * swr_threshold_ratio. + # With ttl=None entries never go stale, so there is nothing to revalidate. + _l1_swr_active = _l1_only_mode and _l1_config.swr_enabled and ttl is not None and ttl > 0 # Create per-function statistics tracker with lazy session ID generation # Session ID format: "{process_uuid}:{module}.{function_name}" @@ -510,6 +528,37 @@ def create_cache_wrapper( # Pass l1_enabled for rate limit classification header _stats = _FunctionStats(function_identifier=function_identifier, l1_enabled=l1_enabled) + # L1-only SWR: strong refs to in-flight refresh tasks. asyncio only keeps weak + # refs to tasks, so a fire-and-forget refresh could be GC'd mid-flight otherwise. + _l1_swr_tasks: set[asyncio.Task[None]] = set() + + def _l1_swr_task_done(task: asyncio.Task[None], cache_key: str) -> None: + _l1_swr_tasks.discard(task) + _ttl_refresh_done_callback(task, cache_key) + + async def _l1_swr_refresh_async( + cache_key: str, version: int, call_args: tuple[Any, ...], call_kwargs: dict[str, Any] + ) -> None: + """Background SWR refresh for async functions in L1-only mode.""" + assert _object_cache is not None and ttl is not None # noqa: S101 - _l1_swr_active guarantees both + try: + result = await func(*call_args, **call_kwargs) + except BaseException: + _object_cache.cancel_refresh(cache_key) # let a later call retry + raise # logged (at debug) by _l1_swr_task_done + _object_cache.complete_refresh(cache_key, version, result, ttl=ttl) + + def _l1_swr_refresh_sync(cache_key: str, version: int, call_args: tuple[Any, ...], call_kwargs: dict[str, Any]) -> None: + """Background SWR refresh for sync functions in L1-only mode (runs on a daemon thread).""" + assert _object_cache is not None and ttl is not None # noqa: S101 - _l1_swr_active guarantees both + try: + result = func(*call_args, **call_kwargs) + except Exception as exc: + _object_cache.cancel_refresh(cache_key) # let a later call retry + _logger.debug("L1-only SWR background refresh failed for %s: %s", cache_key, exc) + return + _object_cache.complete_refresh(cache_key, version, result, ttl=ttl) + # L1-only mode: debug log if backend would have been available # Helps developers understand that Redis config is being intentionally ignored if _l1_only_mode: @@ -584,9 +633,22 @@ def sync_wrapper(*args: Any, **kwargs: Any) -> Any: # noqa: PLR0912 # L1-ONLY MODE: Store raw Python objects (no serialization). # Preserves types (tuples, sets, frozensets) that MessagePack would degrade. if _l1_only_mode and _object_cache: - found, cached_value = _object_cache.get(cache_key) + if _l1_swr_active and ttl is not None: + found, cached_value, needs_refresh, version = _object_cache.get_with_swr(cache_key, ttl) + else: + found, cached_value = _object_cache.get(cache_key) + needs_refresh, version = False, 0 if found: _stats.record_l1_hit() + if needs_refresh: + # SWR: serve the stale value now, refresh on a daemon thread + # (sync functions have no event loop to schedule a task on) + threading.Thread( + target=_l1_swr_refresh_sync, + args=(cache_key, version, args, kwargs), + name=f"cachekit-swr-{func.__name__}", + daemon=True, + ).start() features.clear_correlation_id() reset_current_function_stats(token) return cached_value @@ -909,9 +971,19 @@ async def async_wrapper(*args: Any, **kwargs: Any) -> Any: # L1-ONLY MODE: Store raw Python objects (no serialization). # Preserves types (tuples, sets, frozensets) that MessagePack would degrade. if _l1_only_mode and _object_cache: - found, cached_value = _object_cache.get(cache_key) + if _l1_swr_active and ttl is not None: + found, cached_value, needs_refresh, version = _object_cache.get_with_swr(cache_key, ttl) + else: + found, cached_value = _object_cache.get(cache_key) + needs_refresh, version = False, 0 if found: _stats.record_l1_hit() + if needs_refresh: + # SWR: serve the stale value now, refresh in the background + # without blocking the caller + refresh_task = asyncio.create_task(_l1_swr_refresh_async(cache_key, version, args, kwargs)) + _l1_swr_tasks.add(refresh_task) + refresh_task.add_done_callback(functools.partial(_l1_swr_task_done, cache_key=cache_key)) features.clear_correlation_id() return cached_value diff --git a/src/cachekit/object_cache.py b/src/cachekit/object_cache.py index bafe9ee..a43467e 100644 --- a/src/cachekit/object_cache.py +++ b/src/cachekit/object_cache.py @@ -1,25 +1,86 @@ -"""Thread-safe in-memory object cache with TTL and entry-count LRU eviction. +"""Thread-safe in-memory object cache with TTL, LRU eviction, byte bounds, and SWR. Stores Python object references directly — no serialization. Used by @cache.local() -to provide ultra-low-latency (~50ns) caching for objects that do not need to cross -process boundaries or survive restarts. +and by @cache(backend=None) (L1-only mode) to provide ultra-low-latency (~50ns) +caching for objects that do not need to cross process boundaries or survive restarts. """ from __future__ import annotations import math +import random +import sys import threading import time from collections import OrderedDict -from typing import Any +from collections.abc import Iterable +from dataclasses import dataclass +from typing import Any, cast + + +def _estimate_object_size(obj: Any) -> int: + """Best-effort byte-size estimate of a Python object graph. + + Iteratively walks builtin containers (dict/list/tuple/set/frozenset) with + cycle detection, so a list of large strings is counted at its real weight + instead of pointer-size. Non-container objects are counted shallow via + ``sys.getsizeof`` (pandas/numpy implement ``__sizeof__`` and report real + memory), so arbitrary instances holding large attribute graphs are + under-estimated. This is a memory *bound* heuristic, not exact accounting. + + Args: + obj: Object to estimate. + + Returns: + Estimated size in bytes. + """ + seen: set[int] = set() + stack: list[Any] = [obj] + total = 0 + while stack: + item = stack.pop() + if id(item) in seen: + continue + seen.add(id(item)) + total += sys.getsizeof(item, 64) # 64: fallback for objects without __sizeof__ + if isinstance(item, dict): + mapping = cast("dict[Any, Any]", item) + stack.extend(mapping.keys()) + stack.extend(mapping.values()) + elif isinstance(item, (list, tuple, set, frozenset)): + stack.extend(cast("Iterable[Any]", item)) + return total + + +@dataclass(slots=True) +class _Entry: + """Cache entry: value reference plus timing and size bookkeeping.""" + + value: Any + expires_at: float # time.monotonic() hard-expiry deadline + cached_at: float # time.monotonic() write timestamp (SWR freshness clock) + size_bytes: int # 0 when the cache is not byte-bounded class ObjectCache: """Thread-safe in-memory cache storing Python object references directly. - Uses an OrderedDict for O(1) LRU ordering. On a put() when the cache is - full, expired entries are swept first; if still full, the oldest fresh - entry is evicted (LRU). + Uses an OrderedDict for O(1) LRU ordering. On a put() when a bound is + exceeded, expired entries are swept first; if still over, the oldest fresh + entries are evicted (LRU). + + Bounds (at least one required): + - max_entries: entry-count bound (default 256, pass None to disable) + - max_size_bytes: byte bound using a best-effort recursive size estimate + (default None = disabled). Values larger than the whole budget are never + cached — the function result is still returned, just not stored. + + Stale-while-revalidate (SWR): ``get_with_swr`` serves a fresh-enough entry + while flagging it for background refresh once past + ``ttl * swr_threshold_ratio`` (±10% jitter). The caller runs the refresh and + finishes the cycle with ``complete_refresh`` (or ``cancel_refresh`` on + failure). A per-key version token prevents a refresh that completes after + an invalidation from resurrecting stale data (mirrors L1Cache semantics). Thread safety: RLock on every public method so callers need no external synchronisation. @@ -38,24 +99,50 @@ class ObjectCache: 1 """ - def __init__(self, max_entries: int = 256) -> None: + def __init__( + self, + max_entries: int | None = 256, + max_size_bytes: int | None = None, + swr_threshold_ratio: float = 0.5, + ) -> None: """Initialise the object cache. Args: - max_entries: Maximum number of entries to hold. Must be >= 1. + max_entries: Maximum number of entries to hold (>= 1), or None to + disable the entry-count bound. + max_size_bytes: Maximum estimated bytes to hold (>= 1), or None to + disable the byte bound. At least one bound must be set. + swr_threshold_ratio: Fraction of TTL after which ``get_with_swr`` + flags an entry for background refresh. Must be in (0.0, 1.0]. Raises: - ValueError: If max_entries is less than 1. + ValueError: If both bounds are None, a bound is < 1, or + swr_threshold_ratio is outside (0.0, 1.0]. """ - if max_entries < 1: + if max_entries is None and max_size_bytes is None: + raise ValueError("ObjectCache requires at least one bound: max_entries or max_size_bytes") + if max_entries is not None and max_entries < 1: raise ValueError(f"max_entries must be >= 1, got {max_entries}") + if max_size_bytes is not None and max_size_bytes < 1: + raise ValueError(f"max_size_bytes must be >= 1, got {max_size_bytes}") + if not (0.0 < swr_threshold_ratio <= 1.0): + raise ValueError(f"swr_threshold_ratio must be in (0.0, 1.0], got {swr_threshold_ratio}") self._max_entries = max_entries - # value: (stored_value, expires_at) - self._store: OrderedDict[str, tuple[Any, float]] = OrderedDict() + self._max_size_bytes = max_size_bytes + self._swr_threshold_ratio = swr_threshold_ratio + self._store: OrderedDict[str, _Entry] = OrderedDict() self._lock = threading.RLock() self._hits = 0 self._misses = 0 + self._current_size_bytes = 0 + + # SWR state: keys with an in-flight background refresh, and a per-key + # version bumped on every removal so a refresh completing after an + # invalidation cannot resurrect stale data. Like L1Cache, _versions + # keeps one int per removed key for the cache's lifetime. + self._refreshing: set[str] = set() + self._versions: dict[str, int] = {} # ------------------------------------------------------------------ # Public API @@ -79,24 +166,136 @@ def get(self, key: str) -> tuple[bool, Any]: self._misses += 1 return False, None - value, expires_at = entry - if time.monotonic() >= expires_at: + if time.monotonic() >= entry.expires_at: # Lazy expiry — remove and report miss - del self._store[key] + self._remove(key) self._misses += 1 return False, None # Move to end (most-recently-used) self._store.move_to_end(key) self._hits += 1 - return True, value + return True, entry.value + + def get_with_swr(self, key: str, ttl: float) -> tuple[bool, Any, bool, int]: + """Get value with stale-while-revalidate support. + + Once an entry is older than ``ttl * swr_threshold_ratio`` (±10% jitter + to stagger refreshes), the first caller is told to refresh it in the + background while the cached value keeps being served. When + ``needs_refresh`` is True, the key is marked as refreshing — the caller + MUST finish the cycle with ``complete_refresh`` or ``cancel_refresh``, + otherwise no further refresh is ever flagged for that key. + + Args: + key: Cache key. + ttl: TTL in seconds used for the refresh-threshold calculation. + + Returns: + Tuple of (hit, value, needs_refresh, version): + - hit: Whether key was found and not hard-expired + - value: Cached object or None + - needs_refresh: Whether the caller should trigger a background refresh + - version: Entry version at read time (pass to complete_refresh) + """ + with self._lock: + entry = self._store.get(key) + if entry is None: + self._misses += 1 + return False, None, False, 0 + + now = time.monotonic() + if now >= entry.expires_at: + self._remove(key) + self._misses += 1 + return False, None, False, 0 + + self._store.move_to_end(key) + self._hits += 1 + + version = self._versions.get(key, 0) + needs_refresh = False + # ±10% jitter staggers refreshes when many keys cross the threshold together + jitter = random.uniform(0.9, 1.1) # noqa: S311 - not cryptographic + if (now - entry.cached_at) > ttl * self._swr_threshold_ratio * jitter and key not in self._refreshing: + self._refreshing.add(key) + needs_refresh = True + + return True, entry.value, needs_refresh, version + + def complete_refresh(self, key: str, version: int, value: Any, ttl: float) -> bool: + """Complete a background refresh started by ``get_with_swr``. + + Unlike L1Cache (where Redis owns expiry), there is no L2 source of + truth here — the refreshed value restarts both the freshness clock and + the hard-expiry deadline. + + Args: + key: Cache key. + version: Version token returned by ``get_with_swr``. + value: Freshly computed value. + ttl: TTL in seconds for the refreshed entry. + + Returns: + True if the write succeeded; False if the entry was invalidated or + evicted while the refresh ran (stale data is never resurrected). + + Raises: + ValueError: If ttl is not a finite number >= 1. + """ + if not math.isfinite(ttl) or ttl < 1: + raise ValueError(f"ttl must be a finite number >= 1, got {ttl!r}") + + size = _estimate_object_size(value) if self._max_size_bytes is not None else 0 + with self._lock: + # Clear in-flight marker regardless of outcome so future refreshes can run + self._refreshing.discard(key) + + if self._versions.get(key, 0) != version: + # Entry was invalidated during refresh — don't resurrect stale data + return False + + entry = self._store.get(key) + if entry is None: + # Entry was evicted (LRU/byte pressure) during refresh + return False + + if self._max_size_bytes is not None and size > self._max_size_bytes: + # Refreshed value can no longer fit — drop the entry rather than + # keep serving the stale one forever + self._remove(key) + return False + + now = time.monotonic() + self._current_size_bytes += size - entry.size_bytes + entry.value = value + entry.cached_at = now + entry.expires_at = now + ttl + entry.size_bytes = size + self._store.move_to_end(key) + # New value may be larger — restore the byte bound by evicting LRU others + self._evict(extra_bytes=0, need_slot=False) + return True + + def cancel_refresh(self, key: str) -> None: + """Cancel a background refresh so a later call can retry it. + + Args: + key: Cache key whose refresh failed or was abandoned. + """ + with self._lock: + self._refreshing.discard(key) def put(self, key: str, value: Any, ttl: int) -> None: """Store a value in the cache. - When the cache is at capacity: + When a bound is exceeded: 1. Expired entries are removed first. - 2. If still at capacity, the oldest (LRU) fresh entry is evicted. + 2. If still over, the oldest (LRU) fresh entries are evicted. + + A value whose estimated size exceeds the entire byte budget is never + cached (any smaller stale entry under the same key is dropped so it + stops being served). Args: key: Cache key. @@ -108,24 +307,33 @@ def put(self, key: str, value: Any, ttl: int) -> None: """ if not math.isfinite(ttl) or ttl < 1: raise ValueError(f"ttl must be a finite number >= 1, got {ttl!r}") - expires_at = time.monotonic() + ttl + + size = _estimate_object_size(value) if self._max_size_bytes is not None else 0 + if self._max_size_bytes is not None and size > self._max_size_bytes: + with self._lock: + if key in self._store: + self._remove(key) + return + with self._lock: - # If key already present, update in-place and move to end - if key in self._store: - self._store[key] = (value, expires_at) - self._store.move_to_end(key) - return + # Replacing? Release the old entry's bytes first (pop re-appends at MRU below) + existing = self._store.pop(key, None) + if existing is not None: + self._current_size_bytes -= existing.size_bytes - # Need a slot — evict if at capacity - if len(self._store) >= self._max_entries: - self._evict_to_make_room() + self._evict(extra_bytes=size, need_slot=True) - self._store[key] = (value, expires_at) + now = time.monotonic() + self._store[key] = _Entry(value=value, expires_at=now + ttl, cached_at=now, size_bytes=size) + self._current_size_bytes += size # No move_to_end needed — OrderedDict.__setitem__ appends new keys to end def delete(self, key: str) -> bool: """Remove a single entry from the cache. + Bumps the entry's version so an in-flight SWR refresh cannot + resurrect it. + Args: key: Cache key to remove. @@ -134,7 +342,7 @@ def delete(self, key: str) -> bool: """ with self._lock: if key in self._store: - del self._store[key] + self._remove(key) return True return False @@ -142,9 +350,15 @@ def clear(self) -> None: """Remove all entries from the cache. Hit/miss counters are NOT reset; they represent lifetime statistics. + In-flight SWR refreshes are invalidated (version bump) so they cannot + resurrect cleared entries. """ with self._lock: + for key in self._store: + self._versions[key] = self._versions.get(key, 0) + 1 self._store.clear() + self._current_size_bytes = 0 + self._refreshing.clear() # ------------------------------------------------------------------ # Properties @@ -169,31 +383,68 @@ def size(self) -> int: return len(self._store) @property - def max_entries(self) -> int: - """Maximum number of entries this cache will hold.""" + def size_bytes(self) -> int: + """Current estimated bytes held (always 0 when not byte-bounded).""" + with self._lock: + return self._current_size_bytes + + @property + def max_entries(self) -> int | None: + """Maximum number of entries this cache will hold (None = unbounded count).""" return self._max_entries + @property + def max_size_bytes(self) -> int | None: + """Maximum estimated bytes this cache will hold (None = no byte bound).""" + return self._max_size_bytes + # ------------------------------------------------------------------ # Private helpers # ------------------------------------------------------------------ - def _evict_to_make_room(self) -> None: - """Evict entries to make room for one new entry. + def _remove(self, key: str) -> None: + """Remove an entry and update all bookkeeping. + + Must be called with self._lock held. Every removal path funnels here so + byte accounting, version bumps (anti-resurrection), and in-flight + refresh cancellation stay consistent. + """ + entry = self._store.pop(key, None) + if entry is None: + return + self._current_size_bytes -= entry.size_bytes + self._versions[key] = self._versions.get(key, 0) + 1 + self._refreshing.discard(key) + + def _evict(self, extra_bytes: int, need_slot: bool) -> None: + """Evict entries until both bounds accommodate the pending write. Must be called with self._lock held. Strategy: - 1. Remove all expired entries. - 2. If still at capacity, evict the oldest (LRU) fresh entry. + 1. If any bound is exceeded, remove all expired entries first. + 2. While still over a bound, evict the oldest (LRU) fresh entry. + + Args: + extra_bytes: Estimated size of the value about to be stored. + need_slot: Whether the pending write adds a new entry (entry-count + bound only applies then). """ - now = time.monotonic() - expired_keys = [k for k, (_, exp) in self._store.items() if now >= exp] - for k in expired_keys: - del self._store[k] - # If removing expired entries freed a slot, we are done - if len(self._store) < self._max_entries: + def over_bounds() -> bool: + over_entries = need_slot and self._max_entries is not None and len(self._store) >= self._max_entries + over_bytes = self._max_size_bytes is not None and self._current_size_bytes + extra_bytes > self._max_size_bytes + return over_entries or over_bytes + + if not over_bounds(): return - # Still full — evict the least-recently-used fresh entry - self._store.popitem(last=False) + # Sweep expired entries first + now = time.monotonic() + expired_keys = [k for k, e in self._store.items() if now >= e.expires_at] + for k in expired_keys: + self._remove(k) + + # Still over a bound — evict the least-recently-used fresh entries + while self._store and over_bounds(): + self._remove(next(iter(self._store))) diff --git a/tests/unit/test_l1_only_swr.py b/tests/unit/test_l1_only_swr.py new file mode 100644 index 0000000..bd6d775 --- /dev/null +++ b/tests/unit/test_l1_only_swr.py @@ -0,0 +1,242 @@ +"""L1-only mode (backend=None) honors L1CacheConfig SWR + size config. + +Regression tests for cachekit-py#207: with backend=None the decorator used to +route through an ObjectCache that ignored every L1CacheConfig field — SWR never +scheduled a background refresh and max_size_mb was dead (the store was +entry-count-bounded). These tests pin the fixed behavior: + +- swr_enabled=True + ttl schedules a non-blocking background refresh past + ttl * swr_threshold_ratio (asyncio task for async functions, daemon thread + for sync functions) +- max_size_mb bounds bytes, not entry count + +No Redis or external services required. +""" + +from __future__ import annotations + +import asyncio +import time + +import pytest + +from cachekit import cache +from cachekit.config import L1CacheConfig + + +async def _wait_for_calls(get_calls, expected: int, timeout: float = 2.0) -> None: + """Poll until the call counter reaches ``expected`` (refresh is fire-and-forget).""" + deadline = time.monotonic() + timeout + while get_calls() < expected and time.monotonic() < deadline: + await asyncio.sleep(0.02) + + +@pytest.mark.unit +class TestL1OnlySWRAsync: + """SWR background refresh for async functions in L1-only mode.""" + + async def test_issue_207_repro_background_refresh_happens(self): + """Exact repro from #207: call count reaches 2 after the SWR window passes.""" + calls = 0 + + @cache(ttl=2, backend=None, l1=L1CacheConfig(swr_enabled=True, swr_threshold_ratio=0.2)) + async def fn(): + nonlocal calls + calls += 1 + return calls + + assert await fn() == 1 # miss -> executes + await asyncio.sleep(1.0) # past 20% (±10% jitter) of ttl=2 + assert await fn() == 1 # serves cached value, schedules background refresh + await _wait_for_calls(lambda: calls, 2) + assert calls == 2, f"no background refresh happened (calls={calls})" + + async def test_stale_serve_does_not_block_caller(self): + """The hit that triggers a refresh returns the stale value without awaiting it.""" + calls = 0 + + @cache(ttl=2, backend=None, l1=L1CacheConfig(swr_enabled=True, swr_threshold_ratio=0.2)) + async def fn(): + nonlocal calls + calls += 1 + if calls > 1: + await asyncio.sleep(0.5) # slow refresh must not delay the caller + return calls + + assert await fn() == 1 + await asyncio.sleep(0.6) + + start = time.perf_counter() + result = await fn() + elapsed = time.perf_counter() - start + + assert result == 1 # stale value served + assert elapsed < 0.25, f"caller blocked on refresh ({elapsed:.3f}s)" + await _wait_for_calls(lambda: calls, 2) + assert calls == 2 + + async def test_refreshed_value_served_after_refresh_completes(self): + """Once the background refresh lands, subsequent hits serve the new value.""" + calls = 0 + + @cache(ttl=2, backend=None, l1=L1CacheConfig(swr_enabled=True, swr_threshold_ratio=0.2)) + async def fn(): + nonlocal calls + calls += 1 + return calls + + assert await fn() == 1 + await asyncio.sleep(0.6) + assert await fn() == 1 # stale served, refresh scheduled + await _wait_for_calls(lambda: calls, 2) + assert await fn() == 2 # refreshed value now served from cache + assert calls == 2 # ... without another execution + + async def test_swr_disabled_no_background_refresh(self): + """swr_enabled=False must never schedule a refresh.""" + calls = 0 + + @cache(ttl=2, backend=None, l1=L1CacheConfig(swr_enabled=False)) + async def fn(): + nonlocal calls + calls += 1 + return calls + + assert await fn() == 1 + await asyncio.sleep(1.2) # well past any threshold, before hard expiry + assert await fn() == 1 + await asyncio.sleep(0.2) + assert calls == 1 + + async def test_swr_without_ttl_serves_cached_without_refresh(self): + """SWR needs a ttl — with ttl=None entries never go stale, so no refresh.""" + calls = 0 + + @cache(backend=None, l1=L1CacheConfig(swr_enabled=True, swr_threshold_ratio=0.2)) + async def fn(): + nonlocal calls + calls += 1 + return calls + + assert await fn() == 1 + await asyncio.sleep(0.3) + assert await fn() == 1 + await asyncio.sleep(0.2) + assert calls == 1 + + async def test_failing_refresh_keeps_serving_stale_value(self): + """A refresh that raises is swallowed (logged) and the stale value survives.""" + calls = 0 + + @cache(ttl=5, backend=None, l1=L1CacheConfig(swr_enabled=True, swr_threshold_ratio=0.1)) + async def fn(): + nonlocal calls + calls += 1 + if calls > 1: + raise RuntimeError("refresh boom") + return calls + + assert await fn() == 1 + await asyncio.sleep(0.7) + assert await fn() == 1 # triggers a refresh that will fail + await _wait_for_calls(lambda: calls, 2) + assert calls == 2 + await asyncio.sleep(0.05) # let the failed task's done-callback run + assert await fn() == 1 # stale value still served, caller unaffected + + +@pytest.mark.unit +class TestL1OnlySWRSync: + """SWR background refresh for sync functions in L1-only mode (daemon thread).""" + + def test_sync_function_background_refresh_via_thread(self): + """Sync functions get SWR too — refreshed on a daemon thread, not an error.""" + calls = 0 + + @cache(ttl=2, backend=None, l1=L1CacheConfig(swr_enabled=True, swr_threshold_ratio=0.2)) + def fn(): + nonlocal calls + calls += 1 + return calls + + assert fn() == 1 + time.sleep(1.0) + assert fn() == 1 # stale served, refresh scheduled on a thread + + deadline = time.monotonic() + 2.0 + while calls < 2 and time.monotonic() < deadline: + time.sleep(0.02) + assert calls == 2, f"no background refresh happened (calls={calls})" + + def test_sync_failing_refresh_is_swallowed(self): + """A failing sync refresh must not propagate into any caller.""" + calls = 0 + + @cache(ttl=5, backend=None, l1=L1CacheConfig(swr_enabled=True, swr_threshold_ratio=0.1)) + def fn(): + nonlocal calls + calls += 1 + if calls > 1: + raise RuntimeError("refresh boom") + return calls + + assert fn() == 1 + time.sleep(0.7) + assert fn() == 1 # triggers failing refresh + + deadline = time.monotonic() + 2.0 + while calls < 2 and time.monotonic() < deadline: + time.sleep(0.02) + assert calls == 2 + assert fn() == 1 # stale value still served + + +@pytest.mark.unit +class TestL1OnlySizeBound: + """max_size_mb is a byte bound in L1-only mode, not an entry count.""" + + def test_max_size_mb_bounds_bytes_not_entry_count(self): + """Two ~700KB values under max_size_mb=1 evict by byte pressure at 2 entries.""" + calls = 0 + + @cache(ttl=60, backend=None, l1=L1CacheConfig(max_size_mb=1, swr_enabled=False)) + def fn(i: int) -> str: + nonlocal calls + calls += 1 + return "x" * (700 * 1024) + + fn(1) # cached (~700KB) + fn(2) # ~1.4MB total > 1MB -> LRU-evicts the i=1 entry + assert calls == 2 + fn(2) # MRU entry survived the eviction + assert calls == 2 + fn(1) # evicted at only 2 entries (far below any entry-count bound) -> re-executes + assert calls == 3 + + def test_oversized_value_returned_but_never_cached(self): + """A single value larger than max_size_mb is returned but not stored.""" + calls = 0 + + @cache(ttl=60, backend=None, l1=L1CacheConfig(max_size_mb=1, swr_enabled=False)) + def fn() -> str: + nonlocal calls + calls += 1 + return "x" * (2 * 1024 * 1024) + + assert len(fn()) == 2 * 1024 * 1024 + assert len(fn()) == 2 * 1024 * 1024 + assert calls == 2 # never cached — every call executes + + def test_small_values_cached_normally_under_byte_bound(self): + """Values comfortably within the budget still hit as before.""" + calls = 0 + + @cache(ttl=60, backend=None, l1=L1CacheConfig(max_size_mb=1, swr_enabled=False)) + def fn(i: int) -> str: + nonlocal calls + calls += 1 + return f"value-{i}" + + assert fn(1) == "value-1" + assert fn(1) == "value-1" + assert calls == 1 diff --git a/tests/unit/test_object_cache.py b/tests/unit/test_object_cache.py index 5ef3763..c415a64 100644 --- a/tests/unit/test_object_cache.py +++ b/tests/unit/test_object_cache.py @@ -1,4 +1,4 @@ -"""Unit tests for ObjectCache — TTL, LRU eviction, stats, and thread safety. +"""Unit tests for ObjectCache — TTL, LRU eviction, byte bounds, SWR, stats, and thread safety. All tests are isolated; no Redis or external services required. """ @@ -10,7 +10,7 @@ import pytest -from cachekit.object_cache import ObjectCache +from cachekit.object_cache import ObjectCache, _estimate_object_size @pytest.mark.unit @@ -241,3 +241,209 @@ def worker(thread_id: int) -> None: # Every get either hit (entry still present) or missed (evicted under LRU) # but hits + misses must equal the total number of get() calls assert oc.hits + oc.misses == total_gets + + +@pytest.mark.unit +class TestObjectCacheByteBound: + """max_size_bytes bounds estimated bytes, independent of entry count (#207).""" + + def test_requires_at_least_one_bound(self) -> None: + with pytest.raises(ValueError, match="at least one bound"): + ObjectCache(max_entries=None, max_size_bytes=None) + + def test_invalid_bounds_raise(self) -> None: + with pytest.raises(ValueError, match="max_size_bytes"): + ObjectCache(max_size_bytes=0) + with pytest.raises(ValueError, match="swr_threshold_ratio"): + ObjectCache(swr_threshold_ratio=0.0) + with pytest.raises(ValueError, match="swr_threshold_ratio"): + ObjectCache(swr_threshold_ratio=1.5) + + def test_byte_pressure_evicts_lru(self) -> None: + """Third same-sized value over a ~2.5x budget evicts the LRU entry.""" + value = "x" * 1000 + budget = int(_estimate_object_size(value) * 2.5) + oc = ObjectCache(max_entries=None, max_size_bytes=budget) + + oc.put("a", value, ttl=60) + oc.put("b", value, ttl=60) + assert oc.size == 2 + + oc.put("c", value, ttl=60) # over budget -> "a" (LRU) evicted + + assert oc.get("a")[0] is False + assert oc.get("b")[0] is True + assert oc.get("c")[0] is True + assert oc.size_bytes <= budget + + def test_oversized_value_declined_and_stale_entry_dropped(self) -> None: + """A value bigger than the whole budget is never stored; a smaller stale + entry under the same key is dropped so it stops being served.""" + small = "x" * 100 + budget = _estimate_object_size(small) * 3 + oc = ObjectCache(max_entries=None, max_size_bytes=budget) + + oc.put("k", small, ttl=60) + assert oc.get("k")[0] is True + + oc.put("k", "x" * 100_000, ttl=60) # far over budget -> declined + assert oc.get("k")[0] is False # stale small value no longer served + assert oc.size == 0 + assert oc.size_bytes == 0 + + def test_replacing_entry_updates_byte_accounting(self) -> None: + value = "x" * 1000 + oc = ObjectCache(max_entries=None, max_size_bytes=_estimate_object_size(value) * 10) + + oc.put("k", value, ttl=60) + first_bytes = oc.size_bytes + oc.put("k", value, ttl=60) # replace with same-sized value + assert oc.size_bytes == first_bytes + assert oc.size == 1 + + def test_estimator_counts_container_contents(self) -> None: + """A list of large strings must weigh (roughly) its contents, not pointer size.""" + big_list = ["x" * 10_000 for _ in range(10)] + assert _estimate_object_size(big_list) > 10 * 10_000 + + def test_estimator_handles_cycles(self) -> None: + cyclic: list[object] = [] + cyclic.append(cyclic) + assert _estimate_object_size(cyclic) > 0 # terminates + + +@pytest.mark.unit +class TestObjectCacheSWR: + """Stale-while-revalidate: threshold flagging, refresh completion, anti-resurrection. + + Time is monkeypatched. Elapsed times are chosen with margin around the ±10% + jitter window (threshold in [0.9, 1.1] * ttl * ratio) so tests stay deterministic. + """ + + @staticmethod + def _fake_clock(monkeypatch: pytest.MonkeyPatch, start: float = 1000.0) -> types.SimpleNamespace: + fake_time = types.SimpleNamespace(monotonic=lambda: start) + monkeypatch.setattr("cachekit.object_cache.time", fake_time) + return fake_time + + def test_fresh_entry_no_refresh_flag(self, monkeypatch: pytest.MonkeyPatch) -> None: + fake = self._fake_clock(monkeypatch) + oc = ObjectCache(swr_threshold_ratio=0.5) + oc.put("k", "v1", ttl=10) + + fake.monotonic = lambda: 1004.0 # elapsed 4.0 < 4.5 (min jittered threshold) + hit, value, needs_refresh, _ = oc.get_with_swr("k", ttl=10) + + assert hit is True + assert value == "v1" + assert needs_refresh is False + + def test_stale_entry_flags_refresh_exactly_once(self, monkeypatch: pytest.MonkeyPatch) -> None: + fake = self._fake_clock(monkeypatch) + oc = ObjectCache(swr_threshold_ratio=0.5) + oc.put("k", "v1", ttl=10) + + fake.monotonic = lambda: 1006.0 # elapsed 6.0 > 5.5 (max jittered threshold) + hit, value, needs_refresh, _ = oc.get_with_swr("k", ttl=10) + assert hit is True + assert value == "v1" + assert needs_refresh is True + + # Concurrent readers must not be told to refresh again while one is in flight + hit2, _, needs_refresh2, _ = oc.get_with_swr("k", ttl=10) + assert hit2 is True + assert needs_refresh2 is False + + def test_hard_expired_entry_is_miss_not_stale(self, monkeypatch: pytest.MonkeyPatch) -> None: + fake = self._fake_clock(monkeypatch) + oc = ObjectCache() + oc.put("k", "v1", ttl=10) + + fake.monotonic = lambda: 1011.0 # past hard expiry + hit, value, needs_refresh, _ = oc.get_with_swr("k", ttl=10) + + assert hit is False + assert value is None + assert needs_refresh is False + assert oc.size == 0 + + def test_complete_refresh_updates_value_and_extends_expiry(self, monkeypatch: pytest.MonkeyPatch) -> None: + """L1-only has no L2 source of truth — a refresh restarts the TTL clock.""" + fake = self._fake_clock(monkeypatch) + oc = ObjectCache(swr_threshold_ratio=0.5) + oc.put("k", "v1", ttl=10) # expires at 1010 + + fake.monotonic = lambda: 1006.0 + hit, _, needs_refresh, version = oc.get_with_swr("k", ttl=10) + assert hit and needs_refresh + + assert oc.complete_refresh("k", version, "v2", ttl=10) is True # now expires at 1016 + + fake.monotonic = lambda: 1012.0 # past the ORIGINAL expiry, inside the extended one + hit, value = oc.get("k") + assert hit is True + assert value == "v2" + + def test_complete_refresh_after_delete_does_not_resurrect(self, monkeypatch: pytest.MonkeyPatch) -> None: + """A refresh landing after invalidation must not bring stale data back (#207).""" + fake = self._fake_clock(monkeypatch) + oc = ObjectCache(swr_threshold_ratio=0.5) + oc.put("k", "v1", ttl=10) + + fake.monotonic = lambda: 1006.0 + _, _, needs_refresh, version = oc.get_with_swr("k", ttl=10) + assert needs_refresh + + oc.delete("k") # invalidated while the refresh is "in flight" + + assert oc.complete_refresh("k", version, "v2", ttl=10) is False + assert oc.get("k")[0] is False + + def test_complete_refresh_after_clear_does_not_resurrect(self, monkeypatch: pytest.MonkeyPatch) -> None: + fake = self._fake_clock(monkeypatch) + oc = ObjectCache(swr_threshold_ratio=0.5) + oc.put("k", "v1", ttl=10) + + fake.monotonic = lambda: 1006.0 + _, _, needs_refresh, version = oc.get_with_swr("k", ttl=10) + assert needs_refresh + + oc.clear() + + assert oc.complete_refresh("k", version, "v2", ttl=10) is False + assert oc.get("k")[0] is False + + def test_cancel_refresh_allows_retry(self, monkeypatch: pytest.MonkeyPatch) -> None: + """After a failed refresh is cancelled, the next stale hit flags again.""" + fake = self._fake_clock(monkeypatch) + oc = ObjectCache(swr_threshold_ratio=0.5) + oc.put("k", "v1", ttl=10) + + fake.monotonic = lambda: 1006.0 + _, _, needs_refresh, _ = oc.get_with_swr("k", ttl=10) + assert needs_refresh + + oc.cancel_refresh("k") + + _, _, needs_refresh_retry, _ = oc.get_with_swr("k", ttl=10) + assert needs_refresh_retry is True + + def test_oversized_refresh_result_drops_entry(self, monkeypatch: pytest.MonkeyPatch) -> None: + """If the refreshed value no longer fits the byte budget, the stale entry + is dropped rather than served forever.""" + fake = self._fake_clock(monkeypatch) + small = "x" * 100 + oc = ObjectCache( + max_entries=None, + max_size_bytes=_estimate_object_size(small) * 3, + swr_threshold_ratio=0.5, + ) + oc.put("k", small, ttl=10) + + fake.monotonic = lambda: 1006.0 + _, _, needs_refresh, version = oc.get_with_swr("k", ttl=10) + assert needs_refresh + + assert oc.complete_refresh("k", version, "x" * 100_000, ttl=10) is False + assert oc.get("k")[0] is False + assert oc.size_bytes == 0 From 8481e9dc91efe7e5531b0e587378548807c675de Mon Sep 17 00:00:00 2001 From: Ray Walker Date: Sun, 19 Jul 2026 06:28:30 +1000 Subject: [PATCH 2/3] =?UTF-8?q?fix:=20address=20coderabbit=20review=20?= =?UTF-8?q?=E2=80=94=20L1-only=20SWR=20correctness=20and=20docs?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Honor L1CacheConfig.enabled in L1-only mode: with enabled=False no ObjectCache is created and both wrappers call straight through (falling into the L2 path would contradict backend=None semantics). - Bound background refresh concurrency (32 per wrapped function) with a non-blocking semaphore; at capacity the refresh is skipped and the per-key marker released so a later hit retries. - Snapshot args/kwargs (deepcopy) before scheduling a deferred refresh so caller mutation after a stale return can't store new state under the old key; non-copyable arguments skip the refresh instead. - Replace ObjectCache per-key version tombstones with a per-entry generation counter — anti-resurrection with zero retained state for removed keys (fixes unbounded memory growth under key churn). - put() replacing an entry now funnels through _remove(), and the fresh generation makes an older in-flight refresh unable to overwrite the newer value. - Docs: ttl=None stores with a one-year sentinel (not indefinitely), SWR refresh timing is ~30s under ±10% jitter, and swr_threshold_ratio range is (0.0, 1.0] matching the validator. CodeRabbit-Resolved: docs/configuration.md:255:Correct the documented TTL timin CodeRabbit-Resolved: src/cachekit/config/nested.py:74:Align the documented SWR range w CodeRabbit-Resolved: src/cachekit/decorators/wrapper.py:510:Honour L1CacheConfig.enabled in CodeRabbit-Resolved: src/cachekit/decorators/wrapper.py:533:Bound background refresh concurr CodeRabbit-Resolved: src/cachekit/decorators/wrapper.py:648:Snapshot arguments before deferr CodeRabbit-Resolved: src/cachekit/object_cache.py:145:Replace the unbounded per-key ve CodeRabbit-Resolved: src/cachekit/object_cache.py:327:Invalidate active refreshes when Co-authored-by: multica-agent --- docs/configuration.md | 10 ++- src/cachekit/decorators/wrapper.py | 124 ++++++++++++++++++++++------ src/cachekit/object_cache.py | 61 +++++++------- tests/unit/test_l1_only_swr.py | 125 +++++++++++++++++++++++++++++ tests/unit/test_object_cache.py | 53 ++++++++++++ 5 files changed, 318 insertions(+), 55 deletions(-) diff --git a/docs/configuration.md b/docs/configuration.md index 069483f..fb5b6c2 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -228,7 +228,7 @@ def my_function(): | `enabled` | bool | `True` | Enable L1 in-memory cache | | `max_size_mb` | int | `100` | Maximum L1 cache size in MB | | `swr_enabled` | bool | `True` | Enable stale-while-revalidate (SWR) | -| `swr_threshold_ratio` | float | `0.5` | Refresh at X% of TTL (0.1-1.0) | +| `swr_threshold_ratio` | float | `0.5` | Refresh at X% of TTL, in `(0.0, 1.0]` | | `invalidation_enabled` | bool | `True` | Enable invalidation event broadcasts | | `namespace_index` | bool | `True` | Enable fast namespace-based invalidation | @@ -252,7 +252,9 @@ honored as follows: refreshes it in the background — via `asyncio.create_task` for `async def` functions, or a daemon thread for sync functions. A successful refresh restarts both the freshness clock and the TTL. With `ttl=None` entries never go stale, so no refresh - is ever scheduled. + is ever scheduled — they are stored with a one-year (31,536,000 s) sentinel + expiry rather than truly indefinitely, and can still be evicted earlier under + byte pressure. - **Refresh failures are non-fatal**: the stale value keeps being served until hard expiry, and the next qualifying hit retries the refresh. @@ -265,8 +267,8 @@ from cachekit.config import L1CacheConfig async def load_dashboard(): return await fetch_expensive_data() # illustrative - not defined -# After 30s (50% of TTL), the next call returns the cached value instantly -# and schedules a background refresh — callers never block on revalidation. +# After ~30s (50% of TTL, ±10% jitter), the next call returns the cached value +# instantly and schedules a background refresh — callers never block on revalidation. ``` ### Intent Presets diff --git a/src/cachekit/decorators/wrapper.py b/src/cachekit/decorators/wrapper.py index 8b0f53a..ae413ad 100644 --- a/src/cachekit/decorators/wrapper.py +++ b/src/cachekit/decorators/wrapper.py @@ -1,6 +1,7 @@ from __future__ import annotations import asyncio +import copy import functools import inspect import logging @@ -36,6 +37,11 @@ _logger = logging.getLogger(__name__) +# Cap on concurrent L1-only SWR background refreshes per wrapped function. +# Bounds resource usage when many distinct keys go stale together; at capacity +# the refresh is skipped (stale keeps being served) and a later hit retries. +_L1_SWR_MAX_CONCURRENT_REFRESHES = 32 + def _ttl_refresh_done_callback(task: asyncio.Task, cache_key: str) -> None: """Callback for background TTL refresh tasks to handle errors. @@ -495,19 +501,22 @@ def create_cache_wrapper( from ..config.nested import L1CacheConfig _l1_config: L1CacheConfig = config.l1 if config is not None else L1CacheConfig() + # l1_enabled already merges the decorator param with config.l1.enabled (see + # config handling above) — with it False in L1-only mode there is no cache + # at all and the wrappers call the function directly. _object_cache: ObjectCache | None = ( ObjectCache( max_entries=None, max_size_bytes=_l1_config.max_size_mb * 1024 * 1024, swr_threshold_ratio=_l1_config.swr_threshold_ratio, ) - if _l1_only_mode + if _l1_only_mode and l1_enabled else None ) # SWR needs a TTL: freshness is measured against ttl * swr_threshold_ratio. # With ttl=None entries never go stale, so there is nothing to revalidate. - _l1_swr_active = _l1_only_mode and _l1_config.swr_enabled and ttl is not None and ttl > 0 + _l1_swr_active = _object_cache is not None and _l1_config.swr_enabled and ttl is not None and ttl > 0 # Create per-function statistics tracker with lazy session ID generation # Session ID format: "{process_uuid}:{module}.{function_name}" @@ -532,6 +541,36 @@ def create_cache_wrapper( # refs to tasks, so a fire-and-forget refresh could be GC'd mid-flight otherwise. _l1_swr_tasks: set[asyncio.Task[None]] = set() + # Per-key suppression alone doesn't bound refresh concurrency: a workload + # crossing the SWR threshold on many distinct keys at once would spawn one + # task/thread per key. This semaphore caps in-flight refreshes per wrapped + # function; at capacity the refresh is skipped (stale keeps being served) + # and a later qualifying hit retries. + _l1_swr_slots = threading.BoundedSemaphore(_L1_SWR_MAX_CONCURRENT_REFRESHES) + + def _l1_swr_acquire(cache_key: str, call_args: tuple[Any, ...], call_kwargs: dict[str, Any]) -> tuple[Any, Any] | None: + """Reserve a refresh slot and snapshot the live arguments. + + The cache key was computed from the arguments as they were at call + time; the refresh runs later, so it must not see mutations the caller + makes after receiving the stale value (it would store the new state + under the old key). Returns deep-copied (args, kwargs), or None when at + capacity or the arguments can't be copied — in both cases the refresh + marker is released so a later call retries, and the caller must not + schedule a refresh. + """ + assert _object_cache is not None # noqa: S101 - only called when scheduling a refresh + if not _l1_swr_slots.acquire(blocking=False): + _object_cache.cancel_refresh(cache_key) + return None + try: + return copy.deepcopy((call_args, call_kwargs)) + except Exception as exc: + _l1_swr_slots.release() + _object_cache.cancel_refresh(cache_key) + _logger.debug("L1-only SWR refresh skipped for %s: arguments not deep-copyable: %s", cache_key, exc) + return None + def _l1_swr_task_done(task: asyncio.Task[None], cache_key: str) -> None: _l1_swr_tasks.discard(task) _ttl_refresh_done_callback(task, cache_key) @@ -539,25 +578,37 @@ def _l1_swr_task_done(task: asyncio.Task[None], cache_key: str) -> None: async def _l1_swr_refresh_async( cache_key: str, version: int, call_args: tuple[Any, ...], call_kwargs: dict[str, Any] ) -> None: - """Background SWR refresh for async functions in L1-only mode.""" + """Background SWR refresh for async functions in L1-only mode. + + Only ever scheduled with a slot held via _l1_swr_acquire; releases it. + """ assert _object_cache is not None and ttl is not None # noqa: S101 - _l1_swr_active guarantees both try: - result = await func(*call_args, **call_kwargs) - except BaseException: - _object_cache.cancel_refresh(cache_key) # let a later call retry - raise # logged (at debug) by _l1_swr_task_done - _object_cache.complete_refresh(cache_key, version, result, ttl=ttl) + try: + result = await func(*call_args, **call_kwargs) + except BaseException: + _object_cache.cancel_refresh(cache_key) # let a later call retry + raise # logged (at debug) by _l1_swr_task_done + _object_cache.complete_refresh(cache_key, version, result, ttl=ttl) + finally: + _l1_swr_slots.release() def _l1_swr_refresh_sync(cache_key: str, version: int, call_args: tuple[Any, ...], call_kwargs: dict[str, Any]) -> None: - """Background SWR refresh for sync functions in L1-only mode (runs on a daemon thread).""" + """Background SWR refresh for sync functions in L1-only mode (runs on a daemon thread). + + Only ever scheduled with a slot held via _l1_swr_acquire; releases it. + """ assert _object_cache is not None and ttl is not None # noqa: S101 - _l1_swr_active guarantees both try: - result = func(*call_args, **call_kwargs) - except Exception as exc: - _object_cache.cancel_refresh(cache_key) # let a later call retry - _logger.debug("L1-only SWR background refresh failed for %s: %s", cache_key, exc) - return - _object_cache.complete_refresh(cache_key, version, result, ttl=ttl) + try: + result = func(*call_args, **call_kwargs) + except Exception as exc: + _object_cache.cancel_refresh(cache_key) # let a later call retry + _logger.debug("L1-only SWR background refresh failed for %s: %s", cache_key, exc) + return + _object_cache.complete_refresh(cache_key, version, result, ttl=ttl) + finally: + _l1_swr_slots.release() # L1-only mode: debug log if backend would have been available # Helps developers understand that Redis config is being intentionally ignored @@ -632,6 +683,13 @@ def sync_wrapper(*args: Any, **kwargs: Any) -> Any: # noqa: PLR0912 # L1-ONLY MODE: Store raw Python objects (no serialization). # Preserves types (tuples, sets, frozensets) that MessagePack would degrade. + if _l1_only_mode and _object_cache is None: + # L1 disabled in L1-only mode -> no cache anywhere; call through + try: + return func(*args, **kwargs) + finally: + features.clear_correlation_id() + reset_current_function_stats(token) if _l1_only_mode and _object_cache: if _l1_swr_active and ttl is not None: found, cached_value, needs_refresh, version = _object_cache.get_with_swr(cache_key, ttl) @@ -643,12 +701,21 @@ def sync_wrapper(*args: Any, **kwargs: Any) -> Any: # noqa: PLR0912 if needs_refresh: # SWR: serve the stale value now, refresh on a daemon thread # (sync functions have no event loop to schedule a task on) - threading.Thread( - target=_l1_swr_refresh_sync, - args=(cache_key, version, args, kwargs), - name=f"cachekit-swr-{func.__name__}", - daemon=True, - ).start() + snapshot = _l1_swr_acquire(cache_key, args, kwargs) + if snapshot is not None: + refresh_args, refresh_kwargs = snapshot + try: + threading.Thread( + target=_l1_swr_refresh_sync, + args=(cache_key, version, refresh_args, refresh_kwargs), + name=f"cachekit-swr-{func.__name__}", + daemon=True, + ).start() + except RuntimeError: + # Thread couldn't start (resource pressure) — release + # the slot and marker so a later call retries + _l1_swr_slots.release() + _object_cache.cancel_refresh(cache_key) features.clear_correlation_id() reset_current_function_stats(token) return cached_value @@ -970,6 +1037,10 @@ async def async_wrapper(*args: Any, **kwargs: Any) -> Any: # L1-ONLY MODE: Store raw Python objects (no serialization). # Preserves types (tuples, sets, frozensets) that MessagePack would degrade. + if _l1_only_mode and _object_cache is None: + # L1 disabled in L1-only mode -> no cache anywhere; call through + # (outer finally clears correlation ID and resets stats context) + return await func(*args, **kwargs) if _l1_only_mode and _object_cache: if _l1_swr_active and ttl is not None: found, cached_value, needs_refresh, version = _object_cache.get_with_swr(cache_key, ttl) @@ -981,9 +1052,14 @@ async def async_wrapper(*args: Any, **kwargs: Any) -> Any: if needs_refresh: # SWR: serve the stale value now, refresh in the background # without blocking the caller - refresh_task = asyncio.create_task(_l1_swr_refresh_async(cache_key, version, args, kwargs)) - _l1_swr_tasks.add(refresh_task) - refresh_task.add_done_callback(functools.partial(_l1_swr_task_done, cache_key=cache_key)) + snapshot = _l1_swr_acquire(cache_key, args, kwargs) + if snapshot is not None: + refresh_args, refresh_kwargs = snapshot + refresh_task = asyncio.create_task( + _l1_swr_refresh_async(cache_key, version, refresh_args, refresh_kwargs) + ) + _l1_swr_tasks.add(refresh_task) + refresh_task.add_done_callback(functools.partial(_l1_swr_task_done, cache_key=cache_key)) features.clear_correlation_id() return cached_value diff --git a/src/cachekit/object_cache.py b/src/cachekit/object_cache.py index a43467e..5689bb2 100644 --- a/src/cachekit/object_cache.py +++ b/src/cachekit/object_cache.py @@ -60,6 +60,7 @@ class _Entry: expires_at: float # time.monotonic() hard-expiry deadline cached_at: float # time.monotonic() write timestamp (SWR freshness clock) size_bytes: int # 0 when the cache is not byte-bounded + generation: int # anti-resurrection token: allocated per stored entry, never reused class ObjectCache: @@ -79,8 +80,11 @@ class ObjectCache: while flagging it for background refresh once past ``ttl * swr_threshold_ratio`` (±10% jitter). The caller runs the refresh and finishes the cycle with ``complete_refresh`` (or ``cancel_refresh`` on - failure). A per-key version token prevents a refresh that completes after - an invalidation from resurrecting stale data (mirrors L1Cache semantics). + failure). Each stored entry carries a generation token from a monotonic + counter; a refresh only lands if the same entry (same generation) is still + live, so a refresh that completes after an invalidation, eviction, or + replacement can never resurrect stale data — without retaining any per-key + state for removed entries. Thread safety: RLock on every public method so callers need no external synchronisation. @@ -137,12 +141,12 @@ def __init__( self._misses = 0 self._current_size_bytes = 0 - # SWR state: keys with an in-flight background refresh, and a per-key - # version bumped on every removal so a refresh completing after an - # invalidation cannot resurrect stale data. Like L1Cache, _versions - # keeps one int per removed key for the cache's lifetime. + # SWR state: keys with an in-flight background refresh, plus a monotonic + # generation counter stamped onto every stored entry. A refresh captures + # the entry's generation at read time and only lands if that exact entry + # is still live — removal leaves no per-key residue (no tombstones). self._refreshing: set[str] = set() - self._versions: dict[str, int] = {} + self._generation = 0 # ------------------------------------------------------------------ # Public API @@ -213,7 +217,7 @@ def get_with_swr(self, key: str, ttl: float) -> tuple[bool, Any, bool, int]: self._store.move_to_end(key) self._hits += 1 - version = self._versions.get(key, 0) + version = entry.generation needs_refresh = False # ±10% jitter staggers refreshes when many keys cross the threshold together jitter = random.uniform(0.9, 1.1) # noqa: S311 - not cryptographic @@ -251,13 +255,13 @@ def complete_refresh(self, key: str, version: int, value: Any, ttl: float) -> bo # Clear in-flight marker regardless of outcome so future refreshes can run self._refreshing.discard(key) - if self._versions.get(key, 0) != version: - # Entry was invalidated during refresh — don't resurrect stale data - return False - entry = self._store.get(key) if entry is None: - # Entry was evicted (LRU/byte pressure) during refresh + # Entry was invalidated or evicted during refresh — don't resurrect it + return False + if entry.generation != version: + # Entry was replaced (e.g. by put()) during refresh — the newer + # value wins; the stale refresh result is discarded return False if self._max_size_bytes is not None and size > self._max_size_bytes: @@ -316,23 +320,28 @@ def put(self, key: str, value: Any, ttl: int) -> None: return with self._lock: - # Replacing? Release the old entry's bytes first (pop re-appends at MRU below) - existing = self._store.pop(key, None) - if existing is not None: - self._current_size_bytes -= existing.size_bytes + # Replacing? Remove through _remove so byte accounting and any + # in-flight refresh marker stay consistent (the new entry re-appends + # at MRU below). The fresh generation below makes an older in-flight + # refresh unable to overwrite this newer value. + if key in self._store: + self._remove(key) self._evict(extra_bytes=size, need_slot=True) now = time.monotonic() - self._store[key] = _Entry(value=value, expires_at=now + ttl, cached_at=now, size_bytes=size) + self._generation += 1 + self._store[key] = _Entry( + value=value, expires_at=now + ttl, cached_at=now, size_bytes=size, generation=self._generation + ) self._current_size_bytes += size # No move_to_end needed — OrderedDict.__setitem__ appends new keys to end def delete(self, key: str) -> bool: """Remove a single entry from the cache. - Bumps the entry's version so an in-flight SWR refresh cannot - resurrect it. + An in-flight SWR refresh cannot resurrect it: the refresh only lands + on the exact entry (generation) it was started against. Args: key: Cache key to remove. @@ -350,12 +359,10 @@ def clear(self) -> None: """Remove all entries from the cache. Hit/miss counters are NOT reset; they represent lifetime statistics. - In-flight SWR refreshes are invalidated (version bump) so they cannot - resurrect cleared entries. + In-flight SWR refreshes cannot resurrect cleared entries — their + target entries no longer exist. """ with self._lock: - for key in self._store: - self._versions[key] = self._versions.get(key, 0) + 1 self._store.clear() self._current_size_bytes = 0 self._refreshing.clear() @@ -406,14 +413,14 @@ def _remove(self, key: str) -> None: """Remove an entry and update all bookkeeping. Must be called with self._lock held. Every removal path funnels here so - byte accounting, version bumps (anti-resurrection), and in-flight - refresh cancellation stay consistent. + byte accounting and in-flight refresh cancellation stay consistent. + Anti-resurrection needs no per-key residue: a refresh can only land on + the exact entry (generation) it was started against. """ entry = self._store.pop(key, None) if entry is None: return self._current_size_bytes -= entry.size_bytes - self._versions[key] = self._versions.get(key, 0) + 1 self._refreshing.discard(key) def _evict(self, extra_bytes: int, need_slot: bool) -> None: diff --git a/tests/unit/test_l1_only_swr.py b/tests/unit/test_l1_only_swr.py index bd6d775..906730f 100644 --- a/tests/unit/test_l1_only_swr.py +++ b/tests/unit/test_l1_only_swr.py @@ -16,6 +16,8 @@ from __future__ import annotations import asyncio +import copy +import threading import time import pytest @@ -191,6 +193,129 @@ def fn(): assert fn() == 1 # stale value still served +@pytest.mark.unit +class TestL1OnlyDisabled: + """backend=None + L1CacheConfig(enabled=False) must not cache at all.""" + + def test_sync_no_caching_when_l1_disabled(self): + calls = 0 + + @cache(ttl=60, backend=None, l1=L1CacheConfig(enabled=False)) + def fn(): + nonlocal calls + calls += 1 + return calls + + assert fn() == 1 + assert fn() == 2 # every call executes — nothing was cached + assert calls == 2 + + async def test_async_no_caching_when_l1_disabled(self): + calls = 0 + + @cache(ttl=60, backend=None, l1=L1CacheConfig(enabled=False)) + async def fn(): + nonlocal calls + calls += 1 + return calls + + assert await fn() == 1 + assert await fn() == 2 + assert calls == 2 + + +@pytest.mark.unit +class TestL1OnlySWRArgumentSnapshot: + """The background refresh must see the arguments as they were at call time. + + The cache key is computed before the refresh is scheduled; if the caller + mutates an argument after receiving the stale value, an un-snapshotted + refresh would compute from the new state and store it under the old key. + """ + + async def test_async_refresh_uses_snapshot_not_live_args(self): + seen: list[dict] = [] + + @cache(ttl=2, backend=None, l1=L1CacheConfig(swr_enabled=True, swr_threshold_ratio=0.2)) + async def fn(payload: dict): + seen.append(copy.deepcopy(payload)) + return dict(payload) + + payload = {"v": 1} + assert await fn(payload) == {"v": 1} # miss -> executes + await asyncio.sleep(0.6) + assert await fn(payload) == {"v": 1} # stale hit -> refresh scheduled (snapshot taken) + payload["v"] = 999 # caller mutates BEFORE the refresh task first runs + + deadline = time.monotonic() + 2.0 + while len(seen) < 2 and time.monotonic() < deadline: + await asyncio.sleep(0.02) + assert len(seen) == 2, "no background refresh happened" + assert seen[1] == {"v": 1}, f"refresh saw the caller's mutation: {seen[1]}" + + def test_sync_refresh_uses_snapshot_not_live_args(self): + seen: list[dict] = [] + release = threading.Event() + + @cache(ttl=2, backend=None, l1=L1CacheConfig(swr_enabled=True, swr_threshold_ratio=0.2)) + def fn(payload: dict): + if seen: # only the refresh call waits, so the mutation happens first + release.wait(timeout=2.0) + seen.append(copy.deepcopy(payload)) + return dict(payload) + + payload = {"v": 1} + assert fn(payload) == {"v": 1} + time.sleep(1.0) + assert fn(payload) == {"v": 1} # snapshot taken synchronously before this returns + payload["v"] = 999 + release.set() # now let the refresh thread read its (copied) argument + + deadline = time.monotonic() + 2.0 + while len(seen) < 2 and time.monotonic() < deadline: + time.sleep(0.02) + assert len(seen) == 2, "no background refresh happened" + assert seen[1] == {"v": 1}, f"refresh saw the caller's mutation: {seen[1]}" + + +@pytest.mark.unit +class TestL1OnlySWRBoundedConcurrency: + """Background refresh concurrency is capped (32 per wrapped function). + + Per-key suppression alone would spawn one task per distinct stale key. At + capacity the refresh is skipped (stale keeps being served) and the per-key + marker is released so a later hit retries. + """ + + async def test_async_refreshes_capped_at_32_distinct_stale_keys(self): + n_keys = 40 + refresh_calls = 0 + + @cache(ttl=2, backend=None, l1=L1CacheConfig(swr_enabled=True, swr_threshold_ratio=0.2)) + async def fn(i: int): + nonlocal refresh_calls + refresh_calls += 1 + return i + + for i in range(n_keys): # seed + assert await fn(i) == i + assert refresh_calls == n_keys + + await asyncio.sleep(1.0) # everything stale, nothing hard-expired + + # The wrapper's hit path has no await points, so all 40 stale hits + # reserve slots before any refresh task gets to run: exactly 32 slots + # grant, 8 are rejected (marker released for a later retry). + for i in range(n_keys): + assert await fn(i) == i # stale value served either way + + deadline = time.monotonic() + 3.0 + while refresh_calls < n_keys + 32 and time.monotonic() < deadline: + await asyncio.sleep(0.02) + await asyncio.sleep(0.1) # settle: catch any over-cap stragglers + assert refresh_calls == n_keys + 32, f"expected exactly 32 refreshes, got {refresh_calls - n_keys}" + + @pytest.mark.unit class TestL1OnlySizeBound: """max_size_mb is a byte bound in L1-only mode, not an entry count.""" diff --git a/tests/unit/test_object_cache.py b/tests/unit/test_object_cache.py index c415a64..500a9c0 100644 --- a/tests/unit/test_object_cache.py +++ b/tests/unit/test_object_cache.py @@ -413,6 +413,59 @@ def test_complete_refresh_after_clear_does_not_resurrect(self, monkeypatch: pyte assert oc.complete_refresh("k", version, "v2", ttl=10) is False assert oc.get("k")[0] is False + def test_complete_refresh_after_put_replacement_does_not_overwrite(self, monkeypatch: pytest.MonkeyPatch) -> None: + """put() replacing an entry mid-refresh invalidates the in-flight refresh. + + Regression: put() used a bare pop() that kept the old refresh valid, so + an older in-flight refresh could overwrite the newer value. + """ + fake = self._fake_clock(monkeypatch) + oc = ObjectCache(swr_threshold_ratio=0.5) + oc.put("k", "v1", ttl=10) + + fake.monotonic = lambda: 1006.0 + _, _, needs_refresh, version = oc.get_with_swr("k", ttl=10) + assert needs_refresh + + oc.put("k", "v2-newer", ttl=10) # replaced while the refresh is "in flight" + + assert oc.complete_refresh("k", version, "v1-stale-refresh", ttl=10) is False + hit, value = oc.get("k") + assert hit is True + assert value == "v2-newer" # the newer value survived + + def test_put_replacement_clears_refreshing_marker(self, monkeypatch: pytest.MonkeyPatch) -> None: + """After put() replaces mid-refresh, a later stale hit can flag a new refresh.""" + fake = self._fake_clock(monkeypatch) + oc = ObjectCache(swr_threshold_ratio=0.5) + oc.put("k", "v1", ttl=10) + + fake.monotonic = lambda: 1006.0 + _, _, needs_refresh, _ = oc.get_with_swr("k", ttl=10) + assert needs_refresh # marker now set + + oc.put("k", "v2", ttl=10) # replacement clears the in-flight marker + + fake.monotonic = lambda: 1012.0 # new entry (cached at 1006) is stale again + _, _, needs_refresh_again, _ = oc.get_with_swr("k", ttl=10) + assert needs_refresh_again is True + + def test_complete_refresh_after_delete_and_reput_does_not_overwrite(self, monkeypatch: pytest.MonkeyPatch) -> None: + """delete() + a fresh put() of the same key must still reject the old refresh.""" + fake = self._fake_clock(monkeypatch) + oc = ObjectCache(swr_threshold_ratio=0.5) + oc.put("k", "v1", ttl=10) + + fake.monotonic = lambda: 1006.0 + _, _, needs_refresh, version = oc.get_with_swr("k", ttl=10) + assert needs_refresh + + oc.delete("k") + oc.put("k", "v2-new-entry", ttl=10) + + assert oc.complete_refresh("k", version, "v1-stale-refresh", ttl=10) is False + assert oc.get("k")[1] == "v2-new-entry" + def test_cancel_refresh_allows_retry(self, monkeypatch: pytest.MonkeyPatch) -> None: """After a failed refresh is cancelled, the next stale hit flags again.""" fake = self._fake_clock(monkeypatch) From d978d2d1cc76ad53f0d44fdde7169edfbb0c2549 Mon Sep 17 00:00:00 2001 From: Ray Walker Date: Sun, 19 Jul 2026 06:39:30 +1000 Subject: [PATCH 3/3] =?UTF-8?q?fix:=20address=20coderabbit=20review=20?= =?UTF-8?q?=E2=80=94=20generation-aware=20refresh=20ownership?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The in-flight refresh marker was key-scoped while refresh identity is generation-scoped: a stale refresh from a replaced entry could clear the marker owned by the current entry's refresh, letting a third reader schedule a duplicate concurrent refresh racing last-write-wins. _refreshing is now key -> owning generation; complete_refresh and cancel_refresh only release the marker they own (cancel_refresh now requires the version token), and the wrapper propagates the version through the capacity, copy-failure, execution-failure, and thread-startup-failure paths. Invariant: a present marker always equals the live entry's generation, since every entry change funnels through _remove(), which pops it. CodeRabbit-Resolved: src/cachekit/object_cache.py:149:Track refresh ownership by key a Co-authored-by: multica-agent --- src/cachekit/decorators/wrapper.py | 26 +++++++------- src/cachekit/object_cache.py | 36 ++++++++++++++------ tests/unit/test_object_cache.py | 54 ++++++++++++++++++++++++++++-- 3 files changed, 91 insertions(+), 25 deletions(-) diff --git a/src/cachekit/decorators/wrapper.py b/src/cachekit/decorators/wrapper.py index ae413ad..aecb6df 100644 --- a/src/cachekit/decorators/wrapper.py +++ b/src/cachekit/decorators/wrapper.py @@ -548,26 +548,28 @@ def create_cache_wrapper( # and a later qualifying hit retries. _l1_swr_slots = threading.BoundedSemaphore(_L1_SWR_MAX_CONCURRENT_REFRESHES) - def _l1_swr_acquire(cache_key: str, call_args: tuple[Any, ...], call_kwargs: dict[str, Any]) -> tuple[Any, Any] | None: + def _l1_swr_acquire( + cache_key: str, version: int, call_args: tuple[Any, ...], call_kwargs: dict[str, Any] + ) -> tuple[Any, Any] | None: """Reserve a refresh slot and snapshot the live arguments. The cache key was computed from the arguments as they were at call time; the refresh runs later, so it must not see mutations the caller makes after receiving the stale value (it would store the new state under the old key). Returns deep-copied (args, kwargs), or None when at - capacity or the arguments can't be copied — in both cases the refresh - marker is released so a later call retries, and the caller must not - schedule a refresh. + capacity or the arguments can't be copied — in both cases this exact + refresh (version) is cancelled so a later call retries, and the caller + must not schedule a refresh. """ assert _object_cache is not None # noqa: S101 - only called when scheduling a refresh if not _l1_swr_slots.acquire(blocking=False): - _object_cache.cancel_refresh(cache_key) + _object_cache.cancel_refresh(cache_key, version) return None try: return copy.deepcopy((call_args, call_kwargs)) except Exception as exc: _l1_swr_slots.release() - _object_cache.cancel_refresh(cache_key) + _object_cache.cancel_refresh(cache_key, version) _logger.debug("L1-only SWR refresh skipped for %s: arguments not deep-copyable: %s", cache_key, exc) return None @@ -587,7 +589,7 @@ async def _l1_swr_refresh_async( try: result = await func(*call_args, **call_kwargs) except BaseException: - _object_cache.cancel_refresh(cache_key) # let a later call retry + _object_cache.cancel_refresh(cache_key, version) # let a later call retry raise # logged (at debug) by _l1_swr_task_done _object_cache.complete_refresh(cache_key, version, result, ttl=ttl) finally: @@ -603,7 +605,7 @@ def _l1_swr_refresh_sync(cache_key: str, version: int, call_args: tuple[Any, ... try: result = func(*call_args, **call_kwargs) except Exception as exc: - _object_cache.cancel_refresh(cache_key) # let a later call retry + _object_cache.cancel_refresh(cache_key, version) # let a later call retry _logger.debug("L1-only SWR background refresh failed for %s: %s", cache_key, exc) return _object_cache.complete_refresh(cache_key, version, result, ttl=ttl) @@ -701,7 +703,7 @@ def sync_wrapper(*args: Any, **kwargs: Any) -> Any: # noqa: PLR0912 if needs_refresh: # SWR: serve the stale value now, refresh on a daemon thread # (sync functions have no event loop to schedule a task on) - snapshot = _l1_swr_acquire(cache_key, args, kwargs) + snapshot = _l1_swr_acquire(cache_key, version, args, kwargs) if snapshot is not None: refresh_args, refresh_kwargs = snapshot try: @@ -713,9 +715,9 @@ def sync_wrapper(*args: Any, **kwargs: Any) -> Any: # noqa: PLR0912 ).start() except RuntimeError: # Thread couldn't start (resource pressure) — release - # the slot and marker so a later call retries + # the slot and this exact refresh so a later call retries _l1_swr_slots.release() - _object_cache.cancel_refresh(cache_key) + _object_cache.cancel_refresh(cache_key, version) features.clear_correlation_id() reset_current_function_stats(token) return cached_value @@ -1052,7 +1054,7 @@ async def async_wrapper(*args: Any, **kwargs: Any) -> Any: if needs_refresh: # SWR: serve the stale value now, refresh in the background # without blocking the caller - snapshot = _l1_swr_acquire(cache_key, args, kwargs) + snapshot = _l1_swr_acquire(cache_key, version, args, kwargs) if snapshot is not None: refresh_args, refresh_kwargs = snapshot refresh_task = asyncio.create_task( diff --git a/src/cachekit/object_cache.py b/src/cachekit/object_cache.py index 5689bb2..bcfb5ae 100644 --- a/src/cachekit/object_cache.py +++ b/src/cachekit/object_cache.py @@ -141,11 +141,16 @@ def __init__( self._misses = 0 self._current_size_bytes = 0 - # SWR state: keys with an in-flight background refresh, plus a monotonic - # generation counter stamped onto every stored entry. A refresh captures - # the entry's generation at read time and only lands if that exact entry - # is still live — removal leaves no per-key residue (no tombstones). - self._refreshing: set[str] = set() + # SWR state: in-flight refresh ownership (key -> owning generation) plus + # a monotonic generation counter stamped onto every stored entry. A + # refresh captures the entry's generation at read time and only lands — + # or clears/cancels its marker — for that exact generation, so a stale + # refresh from a replaced entry can neither resurrect data nor release + # a newer refresh's marker (which would allow duplicate concurrent + # refreshes racing last-write-wins). Invariant: a present marker always + # equals the live entry's generation, because every entry change funnels + # through _remove(), which pops it. Removal leaves no per-key residue. + self._refreshing: dict[str, int] = {} self._generation = 0 # ------------------------------------------------------------------ @@ -222,7 +227,7 @@ def get_with_swr(self, key: str, ttl: float) -> tuple[bool, Any, bool, int]: # ±10% jitter staggers refreshes when many keys cross the threshold together jitter = random.uniform(0.9, 1.1) # noqa: S311 - not cryptographic if (now - entry.cached_at) > ttl * self._swr_threshold_ratio * jitter and key not in self._refreshing: - self._refreshing.add(key) + self._refreshing[key] = entry.generation needs_refresh = True return True, entry.value, needs_refresh, version @@ -252,8 +257,11 @@ def complete_refresh(self, key: str, version: int, value: Any, ttl: float) -> bo size = _estimate_object_size(value) if self._max_size_bytes is not None else 0 with self._lock: - # Clear in-flight marker regardless of outcome so future refreshes can run - self._refreshing.discard(key) + # Clear the in-flight marker only if this refresh still owns it — a + # stale refresh must not release a newer refresh's marker (that + # would let a third reader schedule a duplicate concurrent refresh) + if self._refreshing.get(key) == version: + del self._refreshing[key] entry = self._store.get(key) if entry is None: @@ -281,14 +289,20 @@ def complete_refresh(self, key: str, version: int, value: Any, ttl: float) -> bo self._evict(extra_bytes=0, need_slot=False) return True - def cancel_refresh(self, key: str) -> None: + def cancel_refresh(self, key: str, version: int) -> None: """Cancel a background refresh so a later call can retry it. + Only the refresh that owns the in-flight marker (same generation) may + release it — a stale refresh cancelling after the entry was replaced + must not release a newer refresh's marker. + Args: key: Cache key whose refresh failed or was abandoned. + version: Version token returned by ``get_with_swr``. """ with self._lock: - self._refreshing.discard(key) + if self._refreshing.get(key) == version: + del self._refreshing[key] def put(self, key: str, value: Any, ttl: int) -> None: """Store a value in the cache. @@ -421,7 +435,7 @@ def _remove(self, key: str) -> None: if entry is None: return self._current_size_bytes -= entry.size_bytes - self._refreshing.discard(key) + self._refreshing.pop(key, None) def _evict(self, extra_bytes: int, need_slot: bool) -> None: """Evict entries until both bounds accommodate the pending write. diff --git a/tests/unit/test_object_cache.py b/tests/unit/test_object_cache.py index 500a9c0..344d2ef 100644 --- a/tests/unit/test_object_cache.py +++ b/tests/unit/test_object_cache.py @@ -473,14 +473,64 @@ def test_cancel_refresh_allows_retry(self, monkeypatch: pytest.MonkeyPatch) -> N oc.put("k", "v1", ttl=10) fake.monotonic = lambda: 1006.0 - _, _, needs_refresh, _ = oc.get_with_swr("k", ttl=10) + _, _, needs_refresh, version = oc.get_with_swr("k", ttl=10) assert needs_refresh - oc.cancel_refresh("k") + oc.cancel_refresh("k", version) _, _, needs_refresh_retry, _ = oc.get_with_swr("k", ttl=10) assert needs_refresh_retry is True + def test_stale_refresh_cannot_clear_newer_refresh_marker(self, monkeypatch: pytest.MonkeyPatch) -> None: + """An old refresh finishing after a replacement's refresh started must not + release the newer refresh's marker (that would allow duplicate concurrent + refreshes racing last-write-wins) nor overwrite its result. + """ + fake = self._fake_clock(monkeypatch) + oc = ObjectCache(swr_threshold_ratio=0.5) + oc.put("k", "v1", ttl=10) + + fake.monotonic = lambda: 1006.0 + _, _, needs_refresh_a, version_a = oc.get_with_swr("k", ttl=10) + assert needs_refresh_a # refresh A in flight + + oc.put("k", "v2", ttl=10) # replacement clears A's marker, new generation + + fake.monotonic = lambda: 1012.0 # replacement entry (cached at 1006) stale again + _, _, needs_refresh_b, version_b = oc.get_with_swr("k", ttl=10) + assert needs_refresh_b # refresh B in flight + assert version_b != version_a + + # A finishes late: must neither land nor release B's marker + assert oc.complete_refresh("k", version_a, "vA-stale", ttl=10) is False + _, _, needs_refresh_dup, _ = oc.get_with_swr("k", ttl=10) + assert needs_refresh_dup is False, "stale refresh released the in-flight marker" + + # B still owns the cycle and lands normally + assert oc.complete_refresh("k", version_b, "vB-new", ttl=10) is True + assert oc.get("k") == (True, "vB-new") + + def test_stale_cancel_cannot_clear_newer_refresh_marker(self, monkeypatch: pytest.MonkeyPatch) -> None: + """A failed old refresh cancelling late must not release a newer refresh's marker.""" + fake = self._fake_clock(monkeypatch) + oc = ObjectCache(swr_threshold_ratio=0.5) + oc.put("k", "v1", ttl=10) + + fake.monotonic = lambda: 1006.0 + _, _, needs_refresh_a, version_a = oc.get_with_swr("k", ttl=10) + assert needs_refresh_a + + oc.put("k", "v2", ttl=10) + + fake.monotonic = lambda: 1012.0 + _, _, needs_refresh_b, _ = oc.get_with_swr("k", ttl=10) + assert needs_refresh_b + + oc.cancel_refresh("k", version_a) # A failed and cancels late + + _, _, needs_refresh_dup, _ = oc.get_with_swr("k", ttl=10) + assert needs_refresh_dup is False, "stale cancel released the in-flight marker" + def test_oversized_refresh_result_drops_entry(self, monkeypatch: pytest.MonkeyPatch) -> None: """If the refreshed value no longer fits the byte budget, the stale entry is dropped rather than served forever."""