diff --git a/docs/configuration.md b/docs/configuration.md index f5b9578..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 | @@ -237,6 +237,40 @@ 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 — 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. + +```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, ±10% jitter), 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..aecb6df 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. @@ -489,7 +495,28 @@ 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() + # 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 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 = _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}" @@ -510,6 +537,81 @@ 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() + + # 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, 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 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, 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, version) + _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) + + 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. + + 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: + try: + result = await func(*call_args, **call_kwargs) + except BaseException: + _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: + _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). + + 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: + try: + result = func(*call_args, **call_kwargs) + except Exception as exc: + _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) + 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 if _l1_only_mode: @@ -583,10 +685,39 @@ 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: - 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) + snapshot = _l1_swr_acquire(cache_key, version, 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 this exact refresh so a later call retries + _l1_swr_slots.release() + _object_cache.cancel_refresh(cache_key, version) features.clear_correlation_id() reset_current_function_stats(token) return cached_value @@ -908,10 +1039,29 @@ 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: - 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 + snapshot = _l1_swr_acquire(cache_key, version, 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 bafe9ee..bcfb5ae 100644 --- a/src/cachekit/object_cache.py +++ b/src/cachekit/object_cache.py @@ -1,25 +1,90 @@ -"""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 + generation: int # anti-resurrection token: allocated per stored entry, never reused 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). 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. @@ -38,24 +103,55 @@ 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: 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 # ------------------------------------------------------------------ # Public API @@ -79,24 +175,145 @@ 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 = 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 + if (now - entry.cached_at) > ttl * self._swr_threshold_ratio * jitter and key not in self._refreshing: + self._refreshing[key] = entry.generation + 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 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: + # 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: + # 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, 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: + 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. - 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 +325,38 @@ 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 + # 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._store[key] = (value, expires_at) - self._store.move_to_end(key) - return + self._remove(key) - # 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._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. + 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. @@ -134,7 +365,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 +373,13 @@ def clear(self) -> None: """Remove all entries from the cache. Hit/miss counters are NOT reset; they represent lifetime statistics. + In-flight SWR refreshes cannot resurrect cleared entries — their + target entries no longer exist. """ with self._lock: self._store.clear() + self._current_size_bytes = 0 + self._refreshing.clear() # ------------------------------------------------------------------ # Properties @@ -169,31 +404,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 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._refreshing.pop(key, None) + + 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..906730f --- /dev/null +++ b/tests/unit/test_l1_only_swr.py @@ -0,0 +1,367 @@ +"""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 copy +import threading +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 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.""" + + 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..344d2ef 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,312 @@ 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_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) + 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.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.""" + 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