From f8759577df04992c5000037ed662b2a5f8860cfc Mon Sep 17 00:00:00 2001 From: Ray Walker Date: Mon, 31 Aug 2026 04:34:13 +1000 Subject: [PATCH] feat(backend): bound L1 backfill by the server's remaining freshness (LAB-557) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The read response now carries X-CacheKit-Fresh-For (protocol spec/saas-api.md#remaining-freshness). CachekitIO reads parse it (absent = None/legacy; unparseable/negative = 0, the conservative action) and thread (bytes, is_stale, fresh_for) through the freshness chain; L1 backfill uses min(ttl, fresh_for) so an entry read late in its server-side freshness window is never served fresh from L1 past the server's fresh_until. The freshness read path now gates on backend capability, not just configured SWR — the unbounded backfill predates SWR and applied to every CachekitIO read. Revalidation scheduling stays gated on an actually-configured stale window. Post-lock double-check reads share the same bound and stale-exclusion via _l2_double_check. --- .secrets.baseline | 4 +- docs/configuration.md | 1 + src/cachekit/backends/cachekitio/backend.py | 38 +++- src/cachekit/cache_handler.py | 92 +++++--- src/cachekit/decorators/wrapper.py | 132 ++++++++---- src/cachekit/l1_cache.py | 8 +- .../backends/test_cachekitio_swr_transport.py | 45 +++- tests/unit/test_swr_decorator.py | 198 +++++++++++++++++- 8 files changed, 422 insertions(+), 96 deletions(-) diff --git a/.secrets.baseline b/.secrets.baseline index 809c294..90635eb 100644 --- a/.secrets.baseline +++ b/.secrets.baseline @@ -222,7 +222,7 @@ "filename": "src/cachekit/cache_handler.py", "hashed_secret": "5baa61e4c9b93f3f0682250b6cf8331b7ee68fd8", "is_verified": false, - "line_number": 430 + "line_number": 440 } ], "src/cachekit/config/decorator.py": [ @@ -887,5 +887,5 @@ } ] }, - "generated_at": "2026-08-07T16:45:43Z" + "generated_at": "2026-08-30T19:02:25Z" } diff --git a/docs/configuration.md b/docs/configuration.md index c10c5ef..cb255ef 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -192,6 +192,7 @@ Rules and behavior: - A failed background recompute is silent: the entry keeps serving stale until its hard eviction bound, after which the next call takes the ordinary synchronous miss path. - The background recompute runs with a **snapshot of the caller's `contextvars`** (contextvar-based tenant extraction works), but outside the request otherwise — don't rely on other request-scoped resources (open sessions, connections) inside functions that enable SWR. - Stale values are never written to the L1 in-memory cache, and stale reads still count as cache **hits** for metered-misses billing. +- On the CachekitIO backend, every read (SWR-configured or not) also carries the server's remaining freshness (`X-CacheKit-Fresh-For`, [protocol spec](https://github.com/cachekit-io/protocol/blob/main/spec/saas-api.md#remaining-freshness)): an L2 hit backfilled into L1 lives at most `min(ttl, remaining)` locally, so a value read near the end of its server-side freshness window is never served fresh from L1 past the server's bound. Pre-signal servers omit the header and behavior is unchanged. ### File Backend Environment Variables diff --git a/src/cachekit/backends/cachekitio/backend.py b/src/cachekit/backends/cachekitio/backend.py index ef0402d..841ee67 100644 --- a/src/cachekit/backends/cachekitio/backend.py +++ b/src/cachekit/backends/cachekitio/backend.py @@ -44,9 +44,12 @@ # Stale-while-revalidate (LAB-381, spec/saas-api.md#stale-while-revalidate). # STALE_TTL_HEADER rides PUTs to open a stale-grace window past the fresh TTL; # FRESHNESS_HEADER labels every GET/HEAD 200 as fresh|stale. Pre-SWR servers -# ignore the former and never emit the latter. +# ignore the former and never emit the latter. FRESH_FOR_HEADER carries the +# remaining freshness in whole seconds on GET 200s (LAB-557, +# spec/saas-api.md#remaining-freshness); pre-signal servers omit it. STALE_TTL_HEADER = "X-CacheKit-Stale-TTL" FRESHNESS_HEADER = "X-CacheKit-Freshness" +FRESH_FOR_HEADER = "X-CacheKit-Fresh-For" def _inject_metrics_headers(stats: _FunctionStats | None) -> dict[str, str]: @@ -360,19 +363,42 @@ def _is_stale(response: httpx.Response) -> bool: value = response.headers.get(FRESHNESS_HEADER) return value is not None and value != "fresh" - def get_with_freshness(self, key: str) -> tuple[bytes, bool] | None: - """Retrieve value plus its SWR freshness (sync). + @staticmethod + def _fresh_for(response: httpx.Response) -> int | None: + """Parse X-CacheKit-Fresh-For (LAB-557, spec/saas-api.md#remaining-freshness). + + Absent = pre-signal server → None (legacy behavior: no bound). + Unparseable or negative = drift → 0 (do not extend local service — the + conservative action, mirroring the unrecognized-freshness → stale rule). + """ + value = response.headers.get(FRESH_FOR_HEADER) + if value is None: + return None + try: + parsed = int(value) + except ValueError: + # Drift signal, not a crash: a server/proxy emitting garbage here + # disables L1 backfill for affected reads — log so a fleet-wide + # latency regression is diagnosable (expert-panel finding). + _logger.debug(f"Unparseable {FRESH_FOR_HEADER} header {value!r}; treating as 0 (no L1 backfill)") + return 0 + return parsed if parsed >= 0 else 0 + + def get_with_freshness(self, key: str) -> tuple[bytes, bool, int | None] | None: + """Retrieve value plus its SWR freshness and remaining-freshness bound (sync). Returns: - ``(value, is_stale)`` on a hit — ``is_stale`` is True only for an - entry in its stale-grace window (LAB-381) — or None on a miss. + ``(value, is_stale, fresh_for)`` on a hit — ``is_stale`` is True only + for an entry in its stale-grace window (LAB-381); ``fresh_for`` is the + server's remaining freshness in seconds, or None from a pre-signal + server (LAB-557) — or None on a miss. Raises: BackendError: If operation fails (network, auth, etc.) """ try: response = self._request_sync("GET", key) - return response.content, self._is_stale(response) + return response.content, self._is_stale(response), self._fresh_for(response) except BackendError as exc: if exc.original_exception and isinstance(exc.original_exception, httpx.HTTPStatusError): if exc.original_exception.response.status_code == 404: diff --git a/src/cachekit/cache_handler.py b/src/cachekit/cache_handler.py index 1fce645..8cd84b2 100644 --- a/src/cachekit/cache_handler.py +++ b/src/cachekit/cache_handler.py @@ -243,19 +243,29 @@ def supports_streaming_write(backend: BaseBackend) -> TypeGuard[BufferWritableBa class SWRCapableBackend(Protocol): """Backend with server-signaled stale-while-revalidate reads (LAB-381). - Reads report whether the entry is in its stale-grace window; writes accept - the window length. Currently only CachekitIOBackend (the SaaS signals - freshness on read — see protocol spec/saas-api.md#stale-while-revalidate). + Reads report whether the entry is in its stale-grace window plus the + server's remaining freshness in seconds (LAB-557; None from a pre-signal + server); writes accept the window length. Currently only CachekitIOBackend + (the SaaS signals freshness on read — see protocol + spec/saas-api.md#stale-while-revalidate and #remaining-freshness). """ - def get_with_freshness(self, key: str) -> Optional[tuple[bytes, bool]]: ... + def get_with_freshness(self, key: str) -> Optional[tuple[bytes, bool, Optional[int]]]: ... def set(self, key: str, value: bytes, ttl: Optional[int] = None, stale_ttl: Optional[int] = None) -> None: ... def supports_swr(backend: BaseBackend) -> TypeGuard[SWRCapableBackend]: - """Type guard: backend supports server-signaled SWR stale-grace reads (LAB-381).""" - return hasattr(backend, "get_with_freshness") + """Type guard: backend supports server-signaled SWR stale-grace reads (LAB-381). + + Checked on the backend's CLASS, not the instance: protocol methods live on + classes, while instance-level hasattr reads dynamic-attribute objects + (unittest.mock.Mock, __getattr__ proxies) as SWR-capable and silently + reroutes their reads through the freshness path — reachable without any SWR + config since the LAB-557 gate widening (the freshness read now runs for + every capable backend, not only when stale_ttl is set). + """ + return callable(getattr(type(backend), "get_with_freshness", None)) # Import caching for serializer modules @@ -1385,10 +1395,13 @@ def get_cached_value(self, cache_key: str, refresh_ttl: Optional[int] = None) -> get_logger().warning(f"Backend operation failed for get on {cache_key}: {e}") return None - def get_cached_value_with_freshness(self, cache_key: str) -> Optional[tuple[tuple[bool, Any], bool]]: - """SWR variant of :meth:`get_cached_value` (LAB-381): also reports staleness. + def get_cached_value_with_freshness(self, cache_key: str) -> Optional[tuple[tuple[bool, Any], bool, Optional[int]]]: + """SWR variant of :meth:`get_cached_value` (LAB-381/LAB-557): also reports + staleness and the server's remaining freshness in seconds. - Returns ``((True, value), is_stale)`` on a hit, None on miss/error. The mmap + Returns ``((True, value), is_stale, fresh_for)`` on a hit, None on + miss/error. fresh_for is None when no signal exists (pre-signal server, + non-SWR backend) — the caller applies legacy L1 TTL behavior. The mmap fast path is skipped — SWR is CachekitIO-only, which is not buffer-readable. Error semantics mirror get_cached_value: the LAB-108 policy point raises DecryptionAuthenticationError when fail-closed (poisoned entry retained as @@ -1401,10 +1414,16 @@ def get_cached_value_with_freshness(self, cache_key: str) -> Optional[tuple[tupl hit = self._cache_handler.get_with_freshness(cache_key) if hit is None: return None - cached_data, is_stale = hit + # Length-tolerant unpack: a third-party SWR backend built against the + # released 2-tuple (bytes, is_stale) signature must degrade to + # fresh_for=None (legacy L1 behavior), not have a strict 3-unpack + # ValueError swallowed below into a permanent every-hit-is-a-miss + # cache bypass (expert-panel finding, LAB-557). + cached_data, is_stale, *_rest = hit + fresh_for = _rest[0] if _rest else None get_logger().cache_hit(cache_key, "Backend(stale)" if is_stale else "Backend") deserialized = self.serialization_handler.deserialize_data(cached_data, cache_key) - return ((True, deserialized), is_stale) + return ((True, deserialized), is_stale, fresh_for) except KeyringConfigurationError: # LOCAL keyring config fault (bad tenant_id, bad keyring entry index) — # never a legitimate miss, and not tamper. Re-raised past the broad @@ -1422,13 +1441,18 @@ def get_cached_value_with_freshness(self, cache_key: str) -> Optional[tuple[tupl get_logger().warning(f"Backend operation failed for get on {redact_cache_key(cache_key)}: {e}") return None - async def get_cached_value_with_freshness_async(self, cache_key: str) -> Optional[tuple[tuple[bool, Any, bytes], bool]]: - """Async SWR variant (LAB-381): staleness + the raw envelope for L1 backfill. - - Returns ``((True, value, raw_bytes), is_stale)`` on a hit — the 3-tuple - matches :meth:`get_cached_value_async` (LAB-111 routing) so the async - decorator backfills L1 without re-serializing. None on miss/error; the - LAB-108 fail-closed policy propagates DecryptionAuthenticationError. + async def get_cached_value_with_freshness_async( + self, cache_key: str + ) -> Optional[tuple[tuple[bool, Any, bytes], bool, Optional[int]]]: + """Async SWR variant (LAB-381/LAB-557): staleness + remaining freshness + + the raw envelope for L1 backfill. + + Returns ``((True, value, raw_bytes), is_stale, fresh_for)`` on a hit — + the inner 3-tuple matches :meth:`get_cached_value_async` (LAB-111 + routing) so the async decorator backfills L1 without re-serializing; + fresh_for (seconds, None = no signal) bounds that backfill to the + server's remaining freshness. None on miss/error; the LAB-108 + fail-closed policy propagates DecryptionAuthenticationError. """ try: if self._cache_handler is None: @@ -1437,10 +1461,13 @@ async def get_cached_value_with_freshness_async(self, cache_key: str) -> Optiona hit = await self._cache_handler.get_with_freshness_async(cache_key) if hit is None: return None - cached_data, is_stale = hit + # Length-tolerant unpack — same 2-tuple compatibility contract as the + # sync variant above. + cached_data, is_stale, *_rest = hit + fresh_for = _rest[0] if _rest else None get_logger().cache_hit(cache_key, "Backend(stale)" if is_stale else "Backend") deserialized = self.serialization_handler.deserialize_data(cached_data, cache_key) - return ((True, deserialized, cached_data), is_stale) + return ((True, deserialized, cached_data), is_stale, fresh_for) except KeyringConfigurationError: # LOCAL keyring config fault (bad tenant_id, bad keyring entry index) — # never a legitimate miss, and not tamper. Re-raised past the broad @@ -1793,11 +1820,13 @@ async def delete_async(self, key: str) -> bool: """Delete key from cache asynchronously.""" ... - def get_with_freshness(self, key: str) -> Optional[tuple[bytes, bool]]: - """Get value plus SWR staleness (LAB-381); (bytes, is_stale) or None.""" + def get_with_freshness(self, key: str) -> Optional[tuple[bytes, bool, Optional[int]]]: + """Get value plus SWR staleness and remaining freshness (LAB-381/LAB-557); + (bytes, is_stale, fresh_for) or None. fresh_for is None from a pre-signal + server or a non-SWR backend.""" ... - async def get_with_freshness_async(self, key: str) -> Optional[tuple[bytes, bool]]: + async def get_with_freshness_async(self, key: str) -> Optional[tuple[bytes, bool, Optional[int]]]: """Async variant of get_with_freshness.""" ... @@ -1964,16 +1993,17 @@ def get_buffer(self, key: str) -> Optional[BufferHandle]: get_logger().error(f"Unexpected error mmapping key {key}: {e}") return None - def get_with_freshness(self, key: str) -> Optional[tuple[bytes, bool]]: - """Get value plus SWR staleness from an SWR-capable backend (LAB-381). + def get_with_freshness(self, key: str) -> Optional[tuple[bytes, bool, Optional[int]]]: + """Get value plus SWR staleness and remaining freshness (LAB-381/LAB-557). - Returns ``(bytes, is_stale)`` on a hit, or None on miss/error (same - degradation contract as :meth:`get` — an error reads as a miss and the - caller takes the synchronous recompute path). + Returns ``(bytes, is_stale, fresh_for)`` on a hit, or None on miss/error + (same degradation contract as :meth:`get` — an error reads as a miss and + the caller takes the synchronous recompute path). Non-SWR backends read + as ``(bytes, False, None)`` — no freshness signal, legacy L1 behavior. """ if not supports_swr(self.backend): value = self.get(key) - return (value, False) if value is not None else None + return (value, False, None) if value is not None else None try: return self._with_backpressure_and_timeout(self.backend.get_with_freshness, key) except BackendError as e: @@ -1983,11 +2013,11 @@ def get_with_freshness(self, key: str) -> Optional[tuple[bytes, bool]]: get_logger().error(f"Unexpected error getting key {redact_cache_key(key)}: {e}") return None - async def get_with_freshness_async(self, key: str) -> Optional[tuple[bytes, bool]]: + async def get_with_freshness_async(self, key: str) -> Optional[tuple[bytes, bool, Optional[int]]]: """Async variant of :meth:`get_with_freshness` (sync backend call in the thread pool).""" if not supports_swr(self.backend): value = await self.get_async(key) - return (value, False) if value is not None else None + return (value, False, None) if value is not None else None try: return await self._with_backpressure_and_timeout_async(self.backend.get_with_freshness, key) except BackendError as e: diff --git a/src/cachekit/decorators/wrapper.py b/src/cachekit/decorators/wrapper.py index f84847a..9b02456 100644 --- a/src/cachekit/decorators/wrapper.py +++ b/src/cachekit/decorators/wrapper.py @@ -22,6 +22,7 @@ get_logger, handle_decrypt_failure, redact_cache_key, + supports_swr, warn_ttl_refresh_unsupported, ) from ..interop import ( @@ -32,7 +33,7 @@ validate_interop_config, ) from ..key_generator import CacheKeyGenerator -from ..l1_cache import get_l1_cache +from ..l1_cache import DEFAULT_L1_TTL_SECONDS, get_l1_cache from ..object_cache import ObjectCache from ..reliability import CircuitBreakerConfig from ..serializers.base import SerializationError @@ -671,7 +672,9 @@ def _on_l2_deserialize_error(error: Exception, key: str) -> None: # and re-run the wrapped function in the background. Requires an SWR-capable # backend (CachekitIO — the server signals freshness on read). _max_total_ttl = 2_592_000 # 30-day storage cap, shared with the stale window (spec) - _l2_swr_backend_capable = _backend is not None and hasattr(_backend, "get_with_freshness") + # Class-level capability check (shared with the read-path fallback): an + # instance-level hasattr would read Mock/proxy objects as SWR-capable. + _l2_swr_backend_capable = _backend is not None and supports_swr(_backend) _stale_ttl: int | None = None if stale_ttl is not None: from ..config.validation import ConfigurationError @@ -725,6 +728,55 @@ def _put_l1(cache_key: str, serialized_data: Any) -> None: _b = serialized_data.encode("utf-8") if isinstance(serialized_data, str) else serialized_data _l1_cache.put(cache_key, _b, redis_ttl=ttl) + def _l1_backfill_ttl(fresh_for: int | None) -> Any: + """L1 TTL for a backfill from an L2 read, bounded by the server's remaining + freshness (LAB-557, spec/saas-api.md#remaining-freshness). + + Unbounded, a read near the end of the server's freshness window restarts + the clock and serves from L1 as fresh past the server's fresh_until. + fresh_for=None (pre-signal server / no expiry / non-SWR backend) keeps + legacy behavior; fresh_for=0 makes L1Cache.put skip the entry entirely + (effective expiry <= now — nothing fresh remains to record). + + The signal may only ever SHORTEN the L1 lifetime: with ttl=None the + legacy baseline is L1Cache's own default (DEFAULT_L1_TTL_SECONDS), so a + long server remainder is clamped to it — returning raw fresh_for would + EXTEND local service up to the 30-day cap and turn the bound into the + very freshness-extension it exists to prevent (expert-panel finding, + CWE-613: server-side DELETE-as-revocation relies on the ≤300s ageout). + """ + if fresh_for is None: + return ttl + return min(DEFAULT_L1_TTL_SECONDS, fresh_for) if ttl is None else min(ttl, fresh_for) + + async def _l2_double_check(cache_key: str) -> tuple[Any, bool, int | None]: + """Post-lock L2 double-check read, freshness-aware on a capable backend + (LAB-557): a hit found after a lock wait gets the same stale-exclusion + and remaining-freshness bound on its L1 BACKFILL as the primary hit path + (the entry another client just wrote is usually full-window fresh, but + an L2 read error on the primary path can land here with the OLD entry + still live and late in its window). Deliberate asymmetry: a stale hit + here is served WITHOUT scheduling a revalidation — spec-permitted + (subsequent stale reads MAY re-trigger), and this path is already a + double-fault rarity. Returns (cached_result, is_stale, fresh_for); + miss/error = (None, False, None), same degradation contract as + get_cached_value_async. + """ + if _l2_swr_backend_capable: + hit = await operation_handler.get_cached_value_with_freshness_async(cache_key) + return hit if hit is not None else (None, False, None) + return await operation_handler.get_cached_value_async(cache_key), False, None + + def _l1_backfill_from_l2(cache_key: str, cached_data: Any, is_stale: bool, fresh_for: int | None) -> None: + """Backfill L1 from an L2 hit's raw envelope, holding both LAB-557 + invariants at every call site in lockstep: a stale-labelled hit is never + recorded (spec: local caches MUST NOT record stale as fresh), and a + fresh hit's local lifetime is bounded by _l1_backfill_ttl.""" + if _l1_cache and cache_key and cached_data and not is_stale: + cached_bytes = cached_data.encode("utf-8") if isinstance(cached_data, str) else cached_data + _l1_cache.put(cache_key, cached_bytes, redis_ttl=_l1_backfill_ttl(fresh_for)) + _cached_keys.add(cache_key) + def _l2_swr_try_begin(cache_key: str) -> bool: """Claim a revalidation slot for this key; False = already in flight or at capacity. @@ -1239,11 +1291,17 @@ def sync_wrapper(*args: Any, **kwargs: Any) -> Any: # noqa: PLR0912 refresh_ttl = ttl if refresh_ttl_on_get and ttl else None # Use operation handler for all cache access (uses backend internally). - # With SWR active the read carries the server's freshness signal - # (LAB-381): a stale hit is served immediately and revalidated on a - # background daemon thread below. + # A freshness-capable backend (CachekitIO) always takes the freshness + # read — not just when SWR is configured — so every hit carries the + # server's staleness label (LAB-381/LAB-557); a stale hit is served + # immediately and (with SWR active) revalidated on a background daemon + # thread below. The freshness path drops refresh_ttl, which is a + # documented no-op on the sync path anyway (StandardCacheHandler.get), + # and skips the mmap fast path (CachekitIO is not buffer-readable). + # The sync hit path performs no L1 backfill, so the fresh_for bound + # (tuple slot 2) has no consumer here. _sync_l2_stale = False - if _l2_swr_active: + if _l2_swr_backend_capable: _fresh_hit = operation_handler.get_cached_value_with_freshness(cache_key) cached_result = _fresh_hit[0] if _fresh_hit is not None else None _sync_l2_stale = _fresh_hit[1] if _fresh_hit is not None else False @@ -1297,7 +1355,9 @@ def sync_wrapper(*args: Any, **kwargs: Any) -> Any: # noqa: PLR0912 _stats.record_l2_hit(duration_ms) # SWR: stale hit — serve now, revalidate on a daemon thread. - if _sync_l2_stale: + # Gated on _l2_swr_active: without a configured stale window this + # decorator serves the mixed-reader hit but owns no revalidation. + if _sync_l2_stale and _l2_swr_active: _l2_swr_schedule(cache_key, args, kwargs, is_async=False) # WHY: L2 cache hit returns from try block that lacks finally cleanup @@ -1614,14 +1674,18 @@ async def async_wrapper(*args: Any, **kwargs: Any) -> Any: try: # Route through the operation handler so corrupt/tampered entries inherit # eviction + the cache_get_deserialize metric instead of persisting (#159), - # and fail-closed tamper errors propagate (LAB-108). With SWR active the - # read also carries the server's freshness signal (LAB-381): a stale hit - # is served immediately and revalidated in the background below. + # and fail-closed tamper errors propagate (LAB-108). A freshness-capable + # backend (CachekitIO) always takes the freshness read — not just when SWR + # is configured — so every hit carries the server's staleness label and + # remaining-freshness bound (LAB-381/LAB-557): a stale hit is never + # backfilled to L1, and a fresh hit's backfill can't outlive fresh_until. _l2_is_stale = False - if _l2_swr_active: + _l2_fresh_for: int | None = None + if _l2_swr_backend_capable: _fresh_hit = await operation_handler.get_cached_value_with_freshness_async(cache_key) cached_result = _fresh_hit[0] if _fresh_hit is not None else None _l2_is_stale = _fresh_hit[1] if _fresh_hit is not None else False + _l2_fresh_for = _fresh_hit[2] if _fresh_hit is not None else None else: cached_result = await operation_handler.get_cached_value_async(cache_key) @@ -1642,14 +1706,9 @@ async def async_wrapper(*args: Any, **kwargs: Any) -> Any: duration_ms=get_duration_ms, ) - # Update L1 cache with Redis value (serialized bytes) for subsequent fast access. - # Never record a stale-window value as fresh in L1 (spec: local caches - # MUST NOT extend service past the server's bounds). - if _l1_cache and cache_key and cached_data and not _l2_is_stale: - # cached_data is already serialized bytes from Redis - cached_bytes = cached_data.encode("utf-8") if isinstance(cached_data, str) else cached_data - _l1_cache.put(cache_key, cached_bytes, redis_ttl=ttl) - _cached_keys.add(cache_key) + # Update L1 cache with the L2 value (serialized bytes) for subsequent + # fast access — stale-exclusion + remaining-freshness bound (LAB-557). + _l1_backfill_from_l2(cache_key, cached_data, _l2_is_stale, _l2_fresh_for) # Handle TTL refresh if configured and threshold met if refresh_ttl_on_get and ttl and hasattr(_backend, "get_ttl") and hasattr(_backend, "refresh_ttl"): @@ -1672,7 +1731,10 @@ async def async_wrapper(*args: Any, **kwargs: Any) -> Any: # SWR: stale hit — value already in hand; revalidate in the # background so no request pays the recompute at a TTL boundary. - if _l2_is_stale: + # Gated on _l2_swr_active (not just the read gate above): a + # decorator without a configured stale window serves a + # stale-labelled mixed-reader hit but owns no revalidation. + if _l2_is_stale and _l2_swr_active: _l2_swr_schedule(cache_key, args, kwargs, is_async=True) return result @@ -1717,21 +1779,14 @@ async def async_wrapper(*args: Any, **kwargs: Any) -> Any: if lock_acquired: # Lock acquired - double-check cache # Another request may have populated it while we waited. - # Routed through get_cached_value_async: corrupt entries evict (#159). + # Routed through the operation handler: corrupt entries evict (#159), + # stale hits skip L1, fresh backfill bounded by fresh_for (LAB-557). try: - cached_result = await operation_handler.get_cached_value_async(cache_key) + cached_result, _dc_stale, _dc_fresh_for = await _l2_double_check(cache_key) if cached_result is not None: # Another request filled the cache while we waited _found, result, cached_data = cached_result - - # Update L1 cache with serialized bytes - if _l1_cache and cache_key and cached_data: - cached_bytes = ( - cached_data.encode("utf-8") if isinstance(cached_data, str) else cached_data - ) - _l1_cache.put(cache_key, cached_bytes, redis_ttl=ttl) - _cached_keys.add(cache_key) - + _l1_backfill_from_l2(cache_key, cached_data, _dc_stale, _dc_fresh_for) return result except DecryptionAuthenticationError: # Fail-closed tamper raise from get_cached_value_async @@ -1746,20 +1801,13 @@ async def async_wrapper(*args: Any, **kwargs: Any) -> Any: # Another request may have populated it while we waited logger().warning(f"Failed to acquire lock for {cache_key} after {blocking_timeout}s, checking cache") try: - # Routed through get_cached_value_async: corrupt entries evict (#159) - cached_result = await operation_handler.get_cached_value_async(cache_key) + # Routed through the operation handler: corrupt entries evict (#159), + # stale hits skip L1, fresh backfill bounded by fresh_for (LAB-557). + cached_result, _dc_stale, _dc_fresh_for = await _l2_double_check(cache_key) if cached_result is not None: # Cache was populated while waiting - use it _found, result, cached_data = cached_result - - # Update L1 cache with serialized bytes - if _l1_cache and cache_key and cached_data: - cached_bytes = ( - cached_data.encode("utf-8") if isinstance(cached_data, str) else cached_data - ) - _l1_cache.put(cache_key, cached_bytes, redis_ttl=ttl) - _cached_keys.add(cache_key) - + _l1_backfill_from_l2(cache_key, cached_data, _dc_stale, _dc_fresh_for) return result except DecryptionAuthenticationError: # Fail-closed tamper raise from get_cached_value_async diff --git a/src/cachekit/l1_cache.py b/src/cachekit/l1_cache.py index 845d4ec..5cb2ffb 100644 --- a/src/cachekit/l1_cache.py +++ b/src/cachekit/l1_cache.py @@ -12,6 +12,11 @@ from dataclasses import dataclass from typing import Any, Optional +# Default L1 entry lifetime when the caller supplies no TTL. Shared with the +# decorator's LAB-557 backfill bound: the server's Fresh-For may only ever +# SHORTEN the L1 lifetime relative to this default, never extend it. +DEFAULT_L1_TTL_SECONDS = 300 + logger = logging.getLogger(__name__) @@ -176,8 +181,7 @@ def put( elif redis_ttl is not None: expiry = current_time + redis_ttl - self.ttl_buffer_seconds else: - # Default 5 minute TTL if not specified - expiry = current_time + 300 - self.ttl_buffer_seconds + expiry = current_time + DEFAULT_L1_TTL_SECONDS - self.ttl_buffer_seconds # Skip caching if the effective TTL is non-finite (NaN/inf would create an # immortal entry that never expires) or too short (would expire immediately). diff --git a/tests/unit/backends/test_cachekitio_swr_transport.py b/tests/unit/backends/test_cachekitio_swr_transport.py index a668d82..c146da6 100644 --- a/tests/unit/backends/test_cachekitio_swr_transport.py +++ b/tests/unit/backends/test_cachekitio_swr_transport.py @@ -13,6 +13,7 @@ import pytest from cachekit.backends.cachekitio.backend import ( + FRESH_FOR_HEADER, FRESHNESS_HEADER, LEGACY_TTL_HEADER, STALE_TTL_HEADER, @@ -62,7 +63,7 @@ class TestFreshnessRead: def test_header_mapping(self, backend: CachekitIOBackend, headers: dict[str, str] | None, expected_stale: bool) -> None: with patch.object(backend, "_request_sync", return_value=_response(200, b"payload", headers)): result = backend.get_with_freshness("k") - assert result == (b"payload", expected_stale) + assert result == (b"payload", expected_stale, None) def test_miss_returns_none(self, backend: CachekitIOBackend) -> None: err = BackendError( @@ -84,6 +85,37 @@ def test_non_404_error_propagates(self, backend: CachekitIOBackend) -> None: backend.get_with_freshness("k") +class TestFreshForRead: + """X-CacheKit-Fresh-For mapping (LAB-557, spec/saas-api.md#remaining-freshness): + absent = None (pre-signal server, legacy); unparseable/negative = 0 (never + extend local service on drift — mirrors unrecognized-freshness → stale).""" + + @pytest.mark.parametrize( + ("headers", "expected_fresh_for"), + [ + (None, None), # pre-signal server: no header → no bound + ({FRESH_FOR_HEADER: "30"}, 30), + ({FRESH_FOR_HEADER: "0"}, 0), # freshness exhausted → do not backfill + ({FRESH_FOR_HEADER: "garbage"}, 0), # drift → conservative 0 + ({FRESH_FOR_HEADER: "-5"}, 0), # negative → conservative 0 + ({FRESH_FOR_HEADER: "2.5"}, 0), # non-integer → conservative 0 + ], + ) + def test_fresh_for_mapping( + self, backend: CachekitIOBackend, headers: dict[str, str] | None, expected_fresh_for: int | None + ) -> None: + with patch.object(backend, "_request_sync", return_value=_response(200, b"payload", headers)): + result = backend.get_with_freshness("k") + assert result is not None + assert result[2] == expected_fresh_for + + def test_fresh_for_rides_alongside_staleness(self, backend: CachekitIOBackend) -> None: + """A stale-window read carries 0 remaining freshness (server emits both headers).""" + headers = {FRESHNESS_HEADER: "stale", FRESH_FOR_HEADER: "0"} + with patch.object(backend, "_request_sync", return_value=_response(200, b"payload", headers)): + assert backend.get_with_freshness("k") == (b"payload", True, 0) + + class TestStaleGraceWrite: """PUT timing headers: canonical+legacy TTL dual-send, stale window rules.""" @@ -116,12 +148,13 @@ class _SWRBackend: def __init__(self) -> None: self.set_calls: list[tuple] = [] self.freshness: bool = False + self.fresh_for: int | None = None def get(self, key: str): return b"plain-get" def get_with_freshness(self, key: str): - return (b"swr-get", self.freshness) + return (b"swr-get", self.freshness, self.fresh_for) def set(self, key: str, value: bytes, ttl=None, stale_ttl=None) -> None: self.set_calls.append((key, value, ttl, stale_ttl)) @@ -153,14 +186,14 @@ def test_get_with_freshness_swr_backend(self) -> None: backend = _SWRBackend() backend.freshness = True handler = StandardCacheHandler(backend) # type: ignore[arg-type] - assert handler.get_with_freshness("k") == (b"swr-get", True) + assert handler.get_with_freshness("k") == (b"swr-get", True, None) def test_get_with_freshness_fallback_reads_as_fresh(self) -> None: """Non-SWR backends degrade to plain get(), always fresh.""" backend = _PlainBackend() backend.store["k"] = b"value" handler = StandardCacheHandler(backend) # type: ignore[arg-type] - assert handler.get_with_freshness("k") == (b"value", False) + assert handler.get_with_freshness("k") == (b"value", False, None) assert handler.get_with_freshness("missing") is None def test_set_threads_stale_ttl_to_swr_backend(self) -> None: @@ -180,7 +213,7 @@ async def test_async_variants(self) -> None: backend = _SWRBackend() backend.freshness = True handler = StandardCacheHandler(backend) # type: ignore[arg-type] - assert await handler.get_with_freshness_async("k") == (b"swr-get", True) + assert await handler.get_with_freshness_async("k") == (b"swr-get", True, None) assert await handler.set_async("k", b"v", ttl=300, stale_ttl=600) is True assert backend.set_calls == [("k", b"v", 300, 600)] @@ -216,5 +249,5 @@ async def test_get_with_freshness_async_fallback_for_plain_backend(self) -> None backend = _PlainBackend() backend.store["k"] = b"value" handler = StandardCacheHandler(backend) # type: ignore[arg-type] - assert await handler.get_with_freshness_async("k") == (b"value", False) + assert await handler.get_with_freshness_async("k") == (b"value", False, None) assert await handler.get_with_freshness_async("missing") is None diff --git a/tests/unit/test_swr_decorator.py b/tests/unit/test_swr_decorator.py index db6e013..9557c56 100644 --- a/tests/unit/test_swr_decorator.py +++ b/tests/unit/test_swr_decorator.py @@ -32,6 +32,8 @@ class FakeSWRBackend: def __init__(self, grant_lock: bool = True) -> None: self.store: dict[str, bytes] = {} self.stale = False + self.fresh_for: int | None = None + self.freshness_reads = 0 self.grant_lock = grant_lock self.set_calls: list[tuple[int | None, int | None]] = [] self.lock_attempts: list[str] = [] @@ -39,9 +41,10 @@ def __init__(self, grant_lock: bool = True) -> None: def get(self, key: str) -> bytes | None: return self.store.get(key) - def get_with_freshness(self, key: str) -> tuple[bytes, bool] | None: + def get_with_freshness(self, key: str) -> tuple[bytes, bool, int | None] | None: + self.freshness_reads += 1 value = self.store.get(key) - return None if value is None else (value, self.stale) + return None if value is None else (value, self.stale, self.fresh_for) def set(self, key: str, value: bytes, ttl: int | None = None, stale_ttl: int | None = None) -> None: self.store[key] = value @@ -377,9 +380,9 @@ def __init__(self) -> None: def get(self, key: str) -> bytes | None: return self.store.get(key) - def get_with_freshness(self, key: str) -> tuple[bytes, bool] | None: + def get_with_freshness(self, key: str) -> tuple[bytes, bool, int | None] | None: value = self.store.get(key) - return None if value is None else (value, self.stale) + return None if value is None else (value, self.stale, None) def set(self, key: str, value: bytes, ttl: int | None = None, stale_ttl: int | None = None) -> None: self.store[key] = value @@ -484,7 +487,7 @@ def compute(x: int) -> int: class TestOperationHandlerFreshnessDegradation: """get_cached_value_with_freshness error paths mirror get_cached_value (#159 contract).""" - def _make_op(self, deserialize_side_effect=None, get_result=(b"bytes", True)): + def _make_op(self, deserialize_side_effect=None, get_result=(b"bytes", True, None)): from unittest import mock from cachekit.cache_handler import CacheKeyGenerator, CacheOperationHandler, CacheSerializationHandler @@ -507,6 +510,24 @@ def test_no_handler_reads_as_miss(self) -> None: op = CacheOperationHandler(CacheSerializationHandler(), CacheKeyGenerator()) assert op.get_cached_value_with_freshness("k") is None # RuntimeError -> generic path -> miss + def test_legacy_two_tuple_backend_degrades_to_no_bound(self) -> None: + """LAB-557 compat: a third-party SWR backend still returning the released + 2-tuple (bytes, is_stale) must read as fresh_for=None (legacy L1 lifetime), + NOT raise a strict-unpack ValueError that the broad except swallows into a + permanent every-hit-is-a-miss cache bypass (expert-panel finding).""" + from unittest import mock + + from cachekit.cache_handler import CacheKeyGenerator, CacheOperationHandler, CacheSerializationHandler + + serialization = mock.MagicMock(spec=CacheSerializationHandler) + serialization.deserialize_data.return_value = {"v": 1} + serialization.encryption_fail_closed = False + op = CacheOperationHandler(serialization, CacheKeyGenerator()) + cache_handler = mock.MagicMock() + cache_handler.get_with_freshness.return_value = (b"bytes", False) # 0.5.x 2-tuple + op.set_cache_handler(cache_handler) + assert op.get_cached_value_with_freshness("k") == ((True, {"v": 1}), False, None) + def test_backend_error_reads_as_miss(self) -> None: op, cache_handler = self._make_op() cache_handler.get_with_freshness.side_effect = ValueError("backend exploded") @@ -727,10 +748,10 @@ def _make_op_fail_closed(self): serialization.encryption_fail_closed = True # fail closed: MUST propagate op = CacheOperationHandler(serialization, CacheKeyGenerator()) cache_handler = mock.MagicMock() - cache_handler.get_with_freshness.return_value = (b"tampered", True) + cache_handler.get_with_freshness.return_value = (b"tampered", True, 0) async def _gwfa(key: str): - return (b"tampered", True) + return (b"tampered", True, 0) cache_handler.get_with_freshness_async.side_effect = _gwfa op.set_cache_handler(cache_handler) @@ -751,3 +772,166 @@ async def test_async_freshness_getter_propagates_fail_closed(self) -> None: with pytest.raises(DecryptionAuthenticationError): await op.get_cached_value_with_freshness_async("k") cache_handler.delete_async.assert_not_called() # evidence retained when fail-closed + + +class TestFreshForBoundedL1Backfill: + """LAB-557 (spec/saas-api.md#remaining-freshness): L1 backfill from an L2 + hit is bounded by the server's remaining freshness — an entry read late in + its freshness window must not be served fresh from L1 past fresh_until.""" + + @staticmethod + def _l1_put_spy(): + """Patch context recording every redis_ttl passed to L1Cache.put.""" + from unittest import mock + + from cachekit.l1_cache import L1Cache + + seen: list[Any] = [] + original = L1Cache.put + + def spy(self: Any, key: str, value: bytes, redis_ttl: Any = None, expires_at: Any = None) -> None: + seen.append(redis_ttl) + return original(self, key, value, redis_ttl=redis_ttl, expires_at=expires_at) + + return seen, mock.patch.object(L1Cache, "put", spy) + + @staticmethod + async def _seed_then_clear_l1(compute: Any, backend: FakeSWRBackend) -> None: + """First call stores L2 + L1; clear L1 (restoring the L2 bytes) so the + next read is an L2 hit that backfills.""" + assert await compute() == 1 + l2_snapshot = dict(backend.store) + await compute.invalidate_cache() # type: ignore[attr-defined] + backend.store.update(l2_snapshot) + + async def test_backfill_bounded_to_remaining_freshness(self) -> None: + backend = FakeSWRBackend() + calls = {"n": 0} + + @cache(backend=backend, ttl=60, stale_ttl=120, namespace="ff-bound") + async def compute() -> int: + calls["n"] += 1 + return calls["n"] + + await self._seed_then_clear_l1(compute, backend) + backend.fresh_for = 2 # the read lands 2s before the server's fresh_until + + seen, patcher = self._l1_put_spy() + with patcher: + assert await compute() == 1 # L2 hit -> bounded L1 backfill + assert seen == [2] # min(ttl=60, fresh_for=2) — never the decorator-scale 60 + + async def test_absent_signal_keeps_legacy_backfill(self) -> None: + backend = FakeSWRBackend() + calls = {"n": 0} + + @cache(backend=backend, ttl=60, stale_ttl=120, namespace="ff-legacy") + async def compute() -> int: + calls["n"] += 1 + return calls["n"] + + await self._seed_then_clear_l1(compute, backend) + backend.fresh_for = None # pre-signal server + + seen, patcher = self._l1_put_spy() + with patcher: + assert await compute() == 1 + assert seen == [60] # legacy: the decorator ttl, unchanged behavior + + async def test_read_at_freshness_end_is_never_served_fresh_from_l1(self) -> None: + """The LAB-557 regression: fresh-labelled hit with 0s remaining must not + be recorded in L1 — the next read goes back to L2 instead of serving a + locally-resurrected 'fresh' value past the server's freshness end.""" + backend = FakeSWRBackend() + calls = {"n": 0} + + @cache(backend=backend, ttl=60, stale_ttl=120, namespace="ff-zero") + async def compute() -> int: + calls["n"] += 1 + return calls["n"] + + await self._seed_then_clear_l1(compute, backend) + backend.fresh_for = 0 + reads_before = backend.freshness_reads + + assert await compute() == 1 # fresh hit, 0s remaining -> no L1 record + assert await compute() == 1 # MUST reach L2 again (unbounded backfill would serve L1) + assert backend.freshness_reads == reads_before + 2 + assert calls["n"] == 1 # value itself still served from cache, no recompute + + async def test_bound_applies_without_configured_swr(self) -> None: + """The unbounded backfill predates SWR: a capable backend bounds the + backfill even when the decorator configures no stale window.""" + backend = FakeSWRBackend() + calls = {"n": 0} + + @cache(backend=backend, ttl=60, namespace="ff-noswr") # no stale_ttl + async def compute() -> int: + calls["n"] += 1 + return calls["n"] + + await self._seed_then_clear_l1(compute, backend) + backend.fresh_for = 3 + + seen, patcher = self._l1_put_spy() + with patcher: + assert await compute() == 1 + assert seen == [3] + + async def test_stale_hit_without_configured_swr_serves_but_never_revalidates(self) -> None: + """Gate-widening guard: a mixed-reader stale hit on a decorator without + a stale window is served (spec: never a blocking miss) but must not + schedule a background revalidation it doesn't own — and must not + backfill L1.""" + backend = FakeSWRBackend() + calls = {"n": 0} + + @cache(backend=backend, ttl=60, namespace="ff-mixed") # no stale_ttl + async def compute() -> int: + calls["n"] += 1 + return calls["n"] + + await self._seed_then_clear_l1(compute, backend) + backend.stale = True + backend.fresh_for = 0 + + seen, patcher = self._l1_put_spy() + with patcher: + assert await compute() == 1 # served, not an error, no recompute paid + assert seen == [] # stale is never recorded in L1 + await asyncio.sleep(0.2) # grace: a scheduled revalidation would recompute + assert calls["n"] == 1 + assert len(backend.set_calls) == 1 # only the seeding write — no revalidation PUT + + async def test_no_ttl_backfill_clamps_to_l1_default_never_extends(self) -> None: + """Panel finding (CWE-613): with ttl=None the legacy L1 lifetime is + DEFAULT_L1_TTL_SECONDS — a long server remainder must clamp to it, never + extend local service toward the 30-day cap (DELETE-as-revocation relies + on that ageout). A short remainder still shortens.""" + from cachekit.l1_cache import DEFAULT_L1_TTL_SECONDS + + backend = FakeSWRBackend() + calls = {"n": 0} + + @cache(backend=backend, namespace="ff-nottl") # ttl=None + async def compute() -> int: + calls["n"] += 1 + return calls["n"] + + await self._seed_then_clear_l1(compute, backend) + backend.fresh_for = 2_592_000 # 30-day remainder from the server + + seen, patcher = self._l1_put_spy() + with patcher: + assert await compute() == 1 + assert seen == [DEFAULT_L1_TTL_SECONDS] # clamped, not extended + + l2_snapshot = dict(backend.store) + await compute.invalidate_cache() # type: ignore[attr-defined] + backend.store.update(l2_snapshot) + backend.fresh_for = 2 # short remainder still shortens below the default + + seen2, patcher2 = self._l1_put_spy() + with patcher2: + assert await compute() == 1 + assert seen2 == [2]