From 111764f1de67db3c97a15c840fa638bae18b3063 Mon Sep 17 00:00:00 2001 From: Ray Walker Date: Wed, 22 Jul 2026 11:40:44 +1000 Subject: [PATCH 1/2] =?UTF-8?q?fix(swr):=20LAB-381=20panel=20fast-follow?= =?UTF-8?q?=20=E2=80=94=20contextvar=20propagation,=20key=20redaction,=20o?= =?UTF-8?q?rphan=20cut,=20dedup?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Panel FIX-FIRST verdict on merged #228 head (Ray-approved); gates v0.14.0. Must-fix: - [MAJ] Sync background revalidation now runs under a copy_context() snapshot captured in-request, so contextvar-based tenant extraction (ContextVarExtractor + encryption) works off-thread exactly as the async path does via create_task. Without this the daemon thread hit an unset var -> fail-closed ValueError -> silently swallowed -> every stale window degraded to a sync miss at evict_at. AC test proves the revalidation actually executes AND stores through the real encryption stack. - [MIN/CWE-532] The three SWR debug logs emit redact_cache_key() digests; the revalidation thread name is static ('cachekit-swr-revalidate', no key slice). Regression test forces the debug line and asserts no raw key. Should-fix (Ray agreed the recommendations): - Cut orphaned CachekitIOBackend.get_with_freshness_async + its tests (the handler runs sync backend methods in the thread pool; the async variant had no callers — LAB-388 trust-bug pattern). Dropped the dead refresh_ttl params from both operation-handler freshness getters. - Backed-mode SWR gets its own _L2_SWR_MAX_CONCURRENT_REFRESHES constant; _swr_* -> _l2_swr_* rename disentangles it from the _l1_swr_* machinery. - Fail-closed propagation tests through BOTH freshness getters (DecryptionAuthenticationError raises; poisoned entry retained) so an except-reorder can't demote to fail-open with green tests. - Dedup, net deletion in source: 4x decrypt-failure tail -> _handle_l2_read_error[_async]; 4x L1-put idiom -> _put_l1 closure. Docs pass: the contextvar fix CHANGES documented behavior — cache() docstring + configuration.md now state contextvars are snapshotted and visible to the recompute (other request-scoped resources still are not). Gates: 2197 passed (7 new tests), ruff clean, basedpyright 0 errors. Co-authored-by: multica-agent --- docs/configuration.md | 2 +- src/cachekit/backends/cachekitio/backend.py | 11 -- src/cachekit/cache_handler.py | 89 +++++-------- src/cachekit/decorators/intent.py | 4 +- src/cachekit/decorators/wrapper.py | 123 +++++++++-------- .../backends/test_cachekitio_swr_transport.py | 27 ---- tests/unit/test_swr_decorator.py | 126 +++++++++++++++++- 7 files changed, 224 insertions(+), 158 deletions(-) diff --git a/docs/configuration.md b/docs/configuration.md index bbf9d2c..bbbeb2b 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -184,7 +184,7 @@ Rules and behavior: - **CachekitIO only** — other backends have no read-side freshness signal and raise `ConfigurationError` if `stale_ttl` is set. - Concurrent stale hits trigger at most one revalidation: per-process dedup plus (async functions) a non-blocking distributed lease on the backend's lock. Contested = serve stale, don't wait. - 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 **outside the request context** — don't rely on request-scoped state (contextvars, open sessions) inside functions that enable SWR. +- 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. ### File Backend Environment Variables diff --git a/src/cachekit/backends/cachekitio/backend.py b/src/cachekit/backends/cachekitio/backend.py index a8c7bd0..ef0402d 100644 --- a/src/cachekit/backends/cachekitio/backend.py +++ b/src/cachekit/backends/cachekitio/backend.py @@ -511,17 +511,6 @@ async def get_async(self, key: str) -> bytes | None: return None raise - async def get_with_freshness_async(self, key: str) -> tuple[bytes, bool] | None: - """Retrieve value plus its SWR freshness (async). See :meth:`get_with_freshness`.""" - try: - response = await self._request_async("GET", key) - return response.content, self._is_stale(response) - except BackendError as exc: - if exc.original_exception and isinstance(exc.original_exception, httpx.HTTPStatusError): - if exc.original_exception.response.status_code == 404: - return None - raise - async def set_async(self, key: str, value: bytes, ttl: int | None = None, stale_ttl: int | None = None) -> None: """Store value in cache (async). diff --git a/src/cachekit/cache_handler.py b/src/cachekit/cache_handler.py index 043752e..29c1b9d 100644 --- a/src/cachekit/cache_handler.py +++ b/src/cachekit/cache_handler.py @@ -1232,6 +1232,31 @@ def get_cache_key( filtered_kwargs = {k: v for k, v in kwargs.items() if k != "_bypass_cache"} return self.key_generator.generate_key(func, args, filtered_kwargs, namespace, integrity_checking) + def _handle_l2_read_error(self, e: SerializationError, cache_key: str) -> None: + """Shared decrypt/integrity failure tail for sync L2 reads (LAB-108/#159). + + Routes through the single policy point (raises DecryptionAuthenticationError + when fail-closed — poisoned entry retained as evidence), else best-effort + evicts the poisoned entry and notifies, so the caller treats it as a miss. + """ + handle_decrypt_failure(e, tier="l2", cache_key=cache_key, fail_closed=self.serialization_handler.encryption_fail_closed) + try: + if self._cache_handler is not None: + self._cache_handler.delete(cache_key) + except Exception as del_err: # best-effort eviction; never mask the miss/recompute + get_logger().warning(f"Failed to evict poisoned L2 entry {redact_cache_key(cache_key)}: {del_err}") + self._notify_deserialize_error(e, cache_key) + + async def _handle_l2_read_error_async(self, e: SerializationError, cache_key: str) -> None: + """Async twin of :meth:`_handle_l2_read_error` (delete_async eviction).""" + handle_decrypt_failure(e, tier="l2", cache_key=cache_key, fail_closed=self.serialization_handler.encryption_fail_closed) + try: + if self._cache_handler is not None: + await self._cache_handler.delete_async(cache_key) + except Exception as del_err: # best-effort eviction; never mask the miss/recompute + get_logger().warning(f"Failed to evict poisoned L2 entry {redact_cache_key(cache_key)}: {del_err}") + self._notify_deserialize_error(e, cache_key) + def get_cached_value(self, cache_key: str, refresh_ttl: Optional[int] = None) -> Optional[Any]: """Get value from cache if it exists. @@ -1272,29 +1297,13 @@ def get_cached_value(self, cache_key: str, refresh_ttl: Optional[int] = None) -> return (True, deserialized) return None except SerializationError as e: - # Single policy point (cachekit-py#170): records the metric and raises when - # fail-closed — in that case the poisoned entry is deliberately RETAINED as - # evidence for the operator (eviction only happens on the fail-open return). - handle_decrypt_failure( - e, tier="l2", cache_key=cache_key, fail_closed=self.serialization_handler.encryption_fail_closed - ) - # Fail open: best-effort evict the poisoned entry so subsequent reads don't - # re-pay full decompress+verify only to fail again; the caller recomputes - # and re-stores the value (#159). - try: - if self._cache_handler is not None: - self._cache_handler.delete(cache_key) - except Exception as del_err: # best-effort eviction; never mask the miss/recompute - get_logger().warning(f"Failed to evict poisoned L2 entry {redact_cache_key(cache_key)}: {del_err}") - self._notify_deserialize_error(e, cache_key) + self._handle_l2_read_error(e, cache_key) # raises when fail-closed (LAB-108) return None except Exception as e: 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, refresh_ttl: Optional[int] = None - ) -> Optional[tuple[tuple[bool, Any], bool]]: + 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. Returns ``((True, value), is_stale)`` on a hit, None on miss/error. The mmap @@ -1315,26 +1324,13 @@ def get_cached_value_with_freshness( deserialized = self.serialization_handler.deserialize_data(cached_data, cache_key) return ((True, deserialized), is_stale) except SerializationError as e: - # Single policy point (cachekit-py#170): records the metric and raises when - # fail-closed — in that case the poisoned entry is deliberately RETAINED as - # evidence for the operator (eviction only happens on the fail-open return). - handle_decrypt_failure( - e, tier="l2", cache_key=cache_key, fail_closed=self.serialization_handler.encryption_fail_closed - ) - try: - if self._cache_handler is not None: - self._cache_handler.delete(cache_key) - except Exception as del_err: # best-effort eviction; never mask the miss/recompute - get_logger().warning(f"Failed to evict poisoned L2 entry {redact_cache_key(cache_key)}: {del_err}") - self._notify_deserialize_error(e, cache_key) + self._handle_l2_read_error(e, cache_key) # raises when fail-closed (LAB-108) return None except Exception as e: get_logger().warning(f"Backend operation failed for get on {cache_key}: {e}") return None - async def get_cached_value_with_freshness_async( - self, cache_key: str, refresh_ttl: Optional[int] = None - ) -> Optional[tuple[tuple[bool, Any, bytes], bool]]: + 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 @@ -1354,16 +1350,7 @@ async def get_cached_value_with_freshness_async( deserialized = self.serialization_handler.deserialize_data(cached_data, cache_key) return ((True, deserialized, cached_data), is_stale) except SerializationError as e: - # Single policy point (cachekit-py#170) — see get_cached_value_async. - handle_decrypt_failure( - e, tier="l2", cache_key=cache_key, fail_closed=self.serialization_handler.encryption_fail_closed - ) - try: - if self._cache_handler is not None: - await self._cache_handler.delete_async(cache_key) - except Exception as del_err: # best-effort eviction; never mask the miss/recompute - get_logger().warning(f"Failed to evict poisoned L2 entry {redact_cache_key(cache_key)}: {del_err}") - self._notify_deserialize_error(e, cache_key) + await self._handle_l2_read_error_async(e, cache_key) # raises when fail-closed (LAB-108) return None except Exception as e: get_logger().warning(f"Backend operation failed for get on {cache_key}: {e}") @@ -1400,21 +1387,7 @@ async def get_cached_value_async(self, cache_key: str, refresh_ttl: Optional[int return (True, deserialized, cached_data) return None except SerializationError as e: - # Single policy point (cachekit-py#170): records the metric and raises when - # fail-closed — in that case the poisoned entry is deliberately RETAINED as - # evidence for the operator (eviction only happens on the fail-open return). - handle_decrypt_failure( - e, tier="l2", cache_key=cache_key, fail_closed=self.serialization_handler.encryption_fail_closed - ) - # Fail open: best-effort evict the poisoned entry so subsequent reads don't - # re-pay full decompress+verify only to fail again; the caller recomputes - # and re-stores the value (#159). - try: - if self._cache_handler is not None: - await self._cache_handler.delete_async(cache_key) - except Exception as del_err: # best-effort eviction; never mask the miss/recompute - get_logger().warning(f"Failed to evict poisoned L2 entry {redact_cache_key(cache_key)}: {del_err}") - self._notify_deserialize_error(e, cache_key) + await self._handle_l2_read_error_async(e, cache_key) # raises when fail-closed (LAB-108) return None except Exception as e: get_logger().warning(f"Backend operation failed for get on {cache_key}: {e}") diff --git a/src/cachekit/decorators/intent.py b/src/cachekit/decorators/intent.py index 776d3b3..69d47d1 100644 --- a/src/cachekit/decorators/intent.py +++ b/src/cachekit/decorators/intent.py @@ -106,7 +106,9 @@ def cache( ``ttl`` and an SWR-capable backend (CachekitIO); ``ttl + stale_ttl`` is capped at 2,592,000 s (30 days). ``@cache.io`` defaults it to ``ttl`` — pass ``stale_ttl=0`` to opt out. The background recompute - runs outside the request context (no request-scoped state). + sees a snapshot of the caller's ``contextvars`` (so contextvar-based + tenant extraction works), but no other request-scoped resources — + open sessions/connections from the request must not be relied on. Returns: Decorated function with intelligent caching diff --git a/src/cachekit/decorators/wrapper.py b/src/cachekit/decorators/wrapper.py index e2c35f4..7182d7f 100644 --- a/src/cachekit/decorators/wrapper.py +++ b/src/cachekit/decorators/wrapper.py @@ -1,6 +1,7 @@ from __future__ import annotations import asyncio +import contextvars import copy import functools import inspect @@ -53,6 +54,10 @@ # the refresh is skipped (stale keeps being served) and a later hit retries. _L1_SWR_MAX_CONCURRENT_REFRESHES = 32 +# Backed-mode (L2) SWR revalidation pool — deliberately separate from the L1 +# constant above so the two features can be tuned independently (LAB-381 panel). +_L2_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. @@ -602,7 +607,7 @@ 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) - _swr_backend_capable = _backend is not None and hasattr(_backend, "get_with_freshness") + _l2_swr_backend_capable = _backend is not None and hasattr(_backend, "get_with_freshness") _stale_ttl: int | None = None if stale_ttl is not None and stale_ttl != 0: from ..config.validation import ConfigurationError @@ -613,7 +618,7 @@ def _on_l2_deserialize_error(error: Exception, key: str) -> None: raise ConfigurationError("stale_ttl requires a positive ttl (the stale window starts where freshness ends)") if ttl + stale_ttl > _max_total_ttl: raise ConfigurationError(f"ttl + stale_ttl must not exceed {_max_total_ttl} seconds (30-day storage cap)") - if not _swr_backend_capable: + if not _l2_swr_backend_capable: raise ConfigurationError( "stale_ttl requires an SWR-capable backend (CachekitIO). " "Other backends have no read-side freshness signal — remove stale_ttl or switch to @cache.io." @@ -625,7 +630,7 @@ def _on_l2_deserialize_error(error: Exception, key: str) -> None: and getattr(config, "swr_by_default", False) and ttl is not None and ttl > 0 - and _swr_backend_capable + and _l2_swr_backend_capable ): # Preset default (io()): stale window = ttl, capped so the total stays # within the 30-day bound. stale_ttl=0 opts out explicitly. A ttl at or @@ -633,7 +638,7 @@ def _on_l2_deserialize_error(error: Exception, key: str) -> None: _default_window = min(ttl, _max_total_ttl - ttl) _stale_ttl = _default_window if _default_window > 0 else None - _swr_active = _stale_ttl is not None + _l2_swr_active = _stale_ttl is not None # Background revalidation machinery — mirrors the L1-only SWR shapes below: # per-key in-flight dedup + a bounded slot pool so a burst of distinct stale @@ -641,30 +646,36 @@ def _on_l2_deserialize_error(error: Exception, key: str) -> None: # backend's async lock as a non-blocking lease (contested = serve stale, no # wait, no retry — _try_acquire_lock already treats 409 AND 200+null as # contested, LAB-240). The lease is best-effort per spec. - _swr_inflight: set[str] = set() - _swr_tasks: set[asyncio.Task[None]] = set() - _swr_slots = threading.BoundedSemaphore(_L1_SWR_MAX_CONCURRENT_REFRESHES) - _swr_lease_seconds = 30.0 # same server-side lease bound as the miss-path lock + _l2_swr_inflight: set[str] = set() + _l2_swr_tasks: set[asyncio.Task[None]] = set() + _l2_swr_slots = threading.BoundedSemaphore(_L2_SWR_MAX_CONCURRENT_REFRESHES) + _l2_swr_lease_seconds = 30.0 # same server-side lease bound as the miss-path lock + + def _put_l1(cache_key: str, serialized_data: Any) -> None: + """Backfill L1 with serialized bytes (str payloads encoded) under the fresh TTL.""" + if _l1_cache and cache_key: + _b = serialized_data.encode("utf-8") if isinstance(serialized_data, str) else serialized_data + _l1_cache.put(cache_key, _b, redis_ttl=ttl) - def _swr_try_begin(cache_key: str) -> bool: + def _l2_swr_try_begin(cache_key: str) -> bool: """Claim a revalidation slot for this key; False = already in flight or at capacity. - The check-then-add on _swr_inflight is not atomic across OS threads; a rare + The check-then-add on _l2_swr_inflight is not atomic across OS threads; a rare duplicate schedule is benign (the backend lease or last-write-wins between two freshly computed values absorbs it — spec explicitly allows duplicates). """ - if cache_key in _swr_inflight: + if cache_key in _l2_swr_inflight: return False - if not _swr_slots.acquire(blocking=False): + if not _l2_swr_slots.acquire(blocking=False): return False - _swr_inflight.add(cache_key) + _l2_swr_inflight.add(cache_key) return True - def _swr_end(cache_key: str) -> None: - _swr_inflight.discard(cache_key) - _swr_slots.release() + def _l2_swr_end(cache_key: str) -> None: + _l2_swr_inflight.discard(cache_key) + _l2_swr_slots.release() - async def _swr_recompute_store_async(cache_key: str, call_args: tuple[Any, ...], call_kwargs: dict[str, Any]) -> None: + async def _l2_swr_recompute_store_async(cache_key: str, call_args: tuple[Any, ...], call_kwargs: dict[str, Any]) -> None: result = await func(*call_args, **call_kwargs) serialized_data = operation_handler.serialization_handler.serialize_data( result, call_args, call_kwargs, cache_key=cache_key @@ -673,29 +684,27 @@ async def _swr_recompute_store_async(cache_key: str, call_args: tuple[Any, ...], cache_key, serialized_data, ttl=ttl, stale_ttl=_stale_ttl ) # Refresh L1 with the new fresh bytes (mirrors the miss-path store). - if _l1_cache and cache_key: - _b = serialized_data.encode("utf-8") if isinstance(serialized_data, str) else serialized_data - _l1_cache.put(cache_key, _b, redis_ttl=ttl) + _put_l1(cache_key, serialized_data) - async def _swr_revalidate_async(cache_key: str, call_args: tuple[Any, ...], call_kwargs: dict[str, Any]) -> None: + async def _l2_swr_revalidate_async(cache_key: str, call_args: tuple[Any, ...], call_kwargs: dict[str, Any]) -> None: """Background revalidation for async functions. Failures are silent by design: the caller already got the stale value; the entry hard-expires at evict_at and the next request takes the ordinary synchronous miss path (spec degradation).""" try: _acquire_lock = getattr(_backend, "acquire_lock", None) if _acquire_lock is not None: - async with _acquire_lock(cache_key, timeout=_swr_lease_seconds, blocking_timeout=None) as got_lease: + async with _acquire_lock(cache_key, timeout=_l2_swr_lease_seconds, blocking_timeout=None) as got_lease: if not got_lease: return # another client is revalidating — stale already served - await _swr_recompute_store_async(cache_key, call_args, call_kwargs) + await _l2_swr_recompute_store_async(cache_key, call_args, call_kwargs) else: - await _swr_recompute_store_async(cache_key, call_args, call_kwargs) + await _l2_swr_recompute_store_async(cache_key, call_args, call_kwargs) except Exception as exc: # noqa: BLE001 — spec: revalidation failure must never surface to callers - _logger.debug("SWR revalidation failed for %s: %s", cache_key, exc) + _logger.debug("SWR revalidation failed for %s: %s", redact_cache_key(cache_key), exc) finally: - _swr_end(cache_key) + _l2_swr_end(cache_key) - def _swr_revalidate_sync(cache_key: str, call_args: tuple[Any, ...], call_kwargs: dict[str, Any]) -> None: + def _l2_swr_revalidate_sync(cache_key: str, call_args: tuple[Any, ...], call_kwargs: dict[str, Any]) -> None: """Background revalidation for sync functions (daemon thread). ponytail: per-process single-flight only — the distributed lease API is @@ -711,15 +720,13 @@ def _swr_revalidate_sync(cache_key: str, call_args: tuple[Any, ...], call_kwargs operation_handler.cache_handler.set( # type: ignore[attr-defined] cache_key, serialized_data, ttl=ttl, stale_ttl=_stale_ttl ) - if _l1_cache and cache_key: - _b = serialized_data.encode("utf-8") if isinstance(serialized_data, str) else serialized_data - _l1_cache.put(cache_key, _b, redis_ttl=ttl) + _put_l1(cache_key, serialized_data) except Exception as exc: # noqa: BLE001 — spec: revalidation failure must never surface to callers - _logger.debug("SWR revalidation failed for %s: %s", cache_key, exc) + _logger.debug("SWR revalidation failed for %s: %s", redact_cache_key(cache_key), exc) finally: - _swr_end(cache_key) + _l2_swr_end(cache_key) - def _swr_schedule(cache_key: str, call_args: tuple[Any, ...], call_kwargs: dict[str, Any], *, is_async: bool) -> None: + def _l2_swr_schedule(cache_key: str, call_args: tuple[Any, ...], call_kwargs: dict[str, Any], *, is_async: bool) -> None: """Kick off background revalidation for a stale hit (at most one per key). Arguments are deep-copied before scheduling (same contract as the L1-only @@ -730,29 +737,35 @@ def _swr_schedule(cache_key: str, call_args: tuple[Any, ...], call_kwargs: dict[ later hit retries). Any scheduling failure releases the slot so the key never becomes permanently unrevalidatable. """ - if not _swr_try_begin(cache_key): + if not _l2_swr_try_begin(cache_key): return try: call_args, call_kwargs = copy.deepcopy((call_args, call_kwargs)) except Exception as exc: - _swr_end(cache_key) - _logger.debug("SWR revalidation skipped for %s: arguments not deep-copyable: %s", cache_key, exc) + _l2_swr_end(cache_key) + _logger.debug("SWR revalidation skipped for %s: arguments not deep-copyable: %s", redact_cache_key(cache_key), exc) return try: if is_async: - task = asyncio.create_task(_swr_revalidate_async(cache_key, call_args, call_kwargs)) - _swr_tasks.add(task) # strong ref until done (same pattern as _l1_swr_tasks) - task.add_done_callback(_swr_tasks.discard) + task = asyncio.create_task(_l2_swr_revalidate_async(cache_key, call_args, call_kwargs)) + _l2_swr_tasks.add(task) # strong ref until done (same pattern as _l1_swr_tasks) + task.add_done_callback(_l2_swr_tasks.discard) else: + # Snapshot the caller's context (captured in-request, where e.g. a + # ContextVarExtractor's tenant var is set) so the daemon thread sees + # the same contextvars the async path inherits via create_task — + # without this, encryption tenant extraction fails off-request and + # the refresh silently no-ops (LAB-381 panel, MAJ). + ctx = contextvars.copy_context() threading.Thread( - target=_swr_revalidate_sync, - args=(cache_key, call_args, call_kwargs), + target=ctx.run, + args=(_l2_swr_revalidate_sync, cache_key, call_args, call_kwargs), daemon=True, - name=f"cachekit-swr-{cache_key[:40]}", + name="cachekit-swr-revalidate", # no key material (CWE-532) ).start() except Exception as exc: # e.g. Thread.start() RuntimeError under resource pressure - _swr_end(cache_key) - _logger.debug("SWR revalidation could not be scheduled for %s: %s", cache_key, exc) + _l2_swr_end(cache_key) + _logger.debug("SWR revalidation could not be scheduled for %s: %s", redact_cache_key(cache_key), exc) # Create per-function statistics tracker with lazy session ID generation # Session ID format: "{process_uuid}:{module}.{function_name}" @@ -1128,8 +1141,8 @@ def sync_wrapper(*args: Any, **kwargs: Any) -> Any: # noqa: PLR0912 # (LAB-381): a stale hit is served immediately and revalidated on a # background daemon thread below. _sync_l2_stale = False - if _swr_active: - _fresh_hit = operation_handler.get_cached_value_with_freshness(cache_key, refresh_ttl) + if _l2_swr_active: + _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 else: @@ -1185,7 +1198,7 @@ def sync_wrapper(*args: Any, **kwargs: Any) -> Any: # noqa: PLR0912 # SWR: stale hit — serve now, revalidate on a daemon thread. if _sync_l2_stale: - _swr_schedule(cache_key, args, kwargs, is_async=False) + _l2_swr_schedule(cache_key, args, kwargs, is_async=False) # WHY: L2 cache hit returns from try block that lacks finally cleanup # (only inner try at line ~567, not the outer try-finally at ~645-720) @@ -1512,7 +1525,7 @@ async def async_wrapper(*args: Any, **kwargs: Any) -> Any: # read also carries the server's freshness signal (LAB-381): a stale hit # is served immediately and revalidated in the background below. _l2_is_stale = False - if _swr_active: + if _l2_swr_active: _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 @@ -1567,7 +1580,7 @@ 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: - _swr_schedule(cache_key, args, kwargs, is_async=True) + _l2_swr_schedule(cache_key, args, kwargs, is_async=True) return result @@ -1689,11 +1702,7 @@ async def async_wrapper(*args: Any, **kwargs: Any) -> Any: ) # Also store in L1 cache for fast subsequent access (using serialized bytes) - if _l1_cache and cache_key: - serialized_bytes = ( - serialized_data.encode("utf-8") if isinstance(serialized_data, str) else serialized_data - ) - _l1_cache.put(cache_key, serialized_bytes, redis_ttl=ttl) + _put_l1(cache_key, serialized_data) _cached_keys.add(cache_key) # Record successful cache set @@ -1780,11 +1789,7 @@ async def async_wrapper(*args: Any, **kwargs: Any) -> Any: ) # Also store in L1 cache for fast subsequent access (using serialized bytes) - if _l1_cache and cache_key: - serialized_bytes = ( - serialized_data.encode("utf-8") if isinstance(serialized_data, str) else serialized_data - ) - _l1_cache.put(cache_key, serialized_bytes, redis_ttl=ttl) + _put_l1(cache_key, serialized_data) _cached_keys.add(cache_key) # Record successful cache set diff --git a/tests/unit/backends/test_cachekitio_swr_transport.py b/tests/unit/backends/test_cachekitio_swr_transport.py index bf49631..a668d82 100644 --- a/tests/unit/backends/test_cachekitio_swr_transport.py +++ b/tests/unit/backends/test_cachekitio_swr_transport.py @@ -185,33 +185,6 @@ async def test_async_variants(self) -> None: assert backend.set_calls == [("k", b"v", 300, 600)] -class TestAsyncBackendVariants: - """Async mirrors of the freshness read/write (codecov: the async bodies count).""" - - async def test_get_with_freshness_async_maps_header(self, backend: CachekitIOBackend) -> None: - resp = _response(200, b"payload", {FRESHNESS_HEADER: "stale"}) - with patch.object(backend, "_request_async", return_value=resp): - assert await backend.get_with_freshness_async("k") == (b"payload", True) - - async def test_get_with_freshness_async_miss_and_error(self, backend: CachekitIOBackend) -> None: - miss = BackendError( - "not found", - error_type=BackendErrorType.PERMANENT, - original_exception=httpx.HTTPStatusError("404", request=_DUMMY_REQUEST, response=_response(404)), - ) - with patch.object(backend, "_request_async", side_effect=miss): - assert await backend.get_with_freshness_async("k") is None - - boom = BackendError( - "boom", - error_type=BackendErrorType.TRANSIENT, - original_exception=httpx.HTTPStatusError("500", request=_DUMMY_REQUEST, response=_response(500)), - ) - with patch.object(backend, "_request_async", side_effect=boom): - with pytest.raises(BackendError): - await backend.get_with_freshness_async("k") - - class _ExplodingBackend(_SWRBackend): """SWR backend whose reads raise (handler degradation paths).""" diff --git a/tests/unit/test_swr_decorator.py b/tests/unit/test_swr_decorator.py index 7daf283..986191d 100644 --- a/tests/unit/test_swr_decorator.py +++ b/tests/unit/test_swr_decorator.py @@ -371,7 +371,7 @@ def test_slot_exhaustion_skips_revalidation(self, monkeypatch: pytest.MonkeyPatc revalidation (stale keeps being served; a later hit retries).""" import cachekit.decorators.wrapper as wrapper_mod - monkeypatch.setattr(wrapper_mod, "_L1_SWR_MAX_CONCURRENT_REFRESHES", 1) + monkeypatch.setattr(wrapper_mod, "_L2_SWR_MAX_CONCURRENT_REFRESHES", 1) backend = FakeSWRBackend() calls = {"n": 0} gate = threading.Event() @@ -541,3 +541,127 @@ def start(self) -> None: assert compute() == 1 # slot NOT leaked: retry schedules successfully assert _wait_for(lambda: calls["n"] == 2) assert _wait_for(lambda: len(backend.set_calls) == 2) + + +class TestPanelFollowUps: + """LAB-381 panel fast-follow regressions (fixes on top of the merged #228).""" + + def test_sync_revalidation_preserves_contextvars_for_encryption(self) -> None: + """Panel MAJ: the daemon thread must see the caller's contextvars. With + encryption + ContextVarExtractor (fail-closed on unset var), background + revalidation must ACTUALLY execute and store — not silently no-op.""" + from cachekit.decorators.tenant_context import ContextVarExtractor + + backend = FakeSWRBackend() + calls = {"n": 0} + + @cache( + backend=backend, + ttl=60, + stale_ttl=120, + l1_enabled=False, + encryption=True, + master_key="a" * 64, + tenant_extractor=ContextVarExtractor(), + ) + def compute() -> dict[str, int]: + calls["n"] += 1 + return {"call": calls["n"]} + + ContextVarExtractor.set_tenant_id("550e8400-e29b-41d4-a716-446655440000") + assert compute()["call"] == 1 + backend.stale = True + assert compute()["call"] == 1 # stale served + + # Without the copy_context() snapshot the daemon thread hits an unset + # tenant var -> fail-closed ValueError -> swallowed -> no recompute, no store. + assert _wait_for(lambda: calls["n"] == 2), "background revalidation must run off-request" + assert _wait_for(lambda: len(backend.set_calls) == 2), "revalidated value must be stored (encrypt succeeded)" + + def test_no_raw_cache_key_in_thread_name_or_debug_logs(self, caplog: pytest.LogCaptureFixture) -> None: + """Panel MIN (CWE-532): thread names and SWR debug logs carry no raw key. + + The namespace segment of a cache key can carry tenant/user IDs; the + revalidation thread name must be static and the SWR debug lines must + emit only the blake2b-redacted form. + """ + import logging + + backend = FakeSWRBackend() + seen_thread_names: list[str] = [] + + @cache(backend=backend, ttl=60, stale_ttl=120, l1_enabled=False, namespace="tenant-secret-ns") + def compute() -> int: + seen_thread_names.append(threading.current_thread().name) + return 1 + + # Happy path: static thread name, no key slice. + assert compute() == 1 + backend.stale = True + assert compute() == 1 + assert _wait_for(lambda: len(backend.set_calls) == 2) + revalidation_threads = [n for n in seen_thread_names if n.startswith("cachekit-swr")] + assert revalidation_threads == ["cachekit-swr-revalidate"] # static, no key slice + + # Failure path: force the 'skipped' debug line (uncopyable arg) and prove + # the raw key never reaches the log — only the digest does. + lock = threading.Lock() + backend3 = FakeSWRBackend() + + @cache(backend=backend3, ttl=60, stale_ttl=120, l1_enabled=False, namespace="tenant-secret-ns", key=lambda lock: "k3") + def compute3(lock) -> int: + return 1 + + assert compute3(lock) == 1 + backend3.stale = True + with caplog.at_level(logging.DEBUG): + assert compute3(lock) == 1 # stale served; deepcopy fails -> skipped debug line + time.sleep(0.1) + + swr_lines = [r.getMessage() for r in caplog.records if "SWR revalidation" in r.getMessage()] + assert swr_lines, "expected the 'skipped' SWR debug line to fire" + for line in swr_lines: + assert "tenant-secret-ns" not in line # raw key redacted (CWE-532) + assert " None: + from cachekit.serializers.encryption_wrapper import DecryptionAuthenticationError + + op, cache_handler = self._make_op_fail_closed() + with pytest.raises(DecryptionAuthenticationError): + op.get_cached_value_with_freshness("k") + cache_handler.delete.assert_not_called() # evidence retained when fail-closed + + async def test_async_freshness_getter_propagates_fail_closed(self) -> None: + from cachekit.serializers.encryption_wrapper import DecryptionAuthenticationError + + op, cache_handler = self._make_op_fail_closed() + 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 From a2bdb3001758b7dff9829d66a768675934036412 Mon Sep 17 00:00:00 2001 From: Ray Walker Date: Wed, 22 Jul 2026 11:51:55 +1000 Subject: [PATCH 2/2] =?UTF-8?q?fix(swr):=20re-review=20round=20=E2=80=94?= =?UTF-8?q?=20redact=20remaining=20SWR-surface=20key=20logs=20(CWE-532)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The crypto re-review of the fast-follow verified LAB-108 preservation, contextvar isolation, and the rename/dedup, and found two remaining redaction gaps on the SWR surface: - StandardCacheHandler.get_with_freshness[_async] + the operation-handler freshness getters logged raw cache_key at ERROR/WARNING on backend errors — fires on routine outages, tenant-bearing keys in prod logs. - The L1-only SWR debug lines (+ shared TTL-refresh done-callback) in wrapper.py still emitted raw keys. All now use redact_cache_key(). NOTE: the pre-existing raw-key logging convention in the non-SWR getters (plain get/get_buffer, sets) is deliberately out of scope for this release-gating PR — repo-wide sweep belongs in its own issue. 2197 tests green; ruff clean. Co-authored-by: multica-agent --- src/cachekit/cache_handler.py | 12 ++++++------ src/cachekit/decorators/wrapper.py | 8 +++++--- 2 files changed, 11 insertions(+), 9 deletions(-) diff --git a/src/cachekit/cache_handler.py b/src/cachekit/cache_handler.py index 29c1b9d..5f7c7f4 100644 --- a/src/cachekit/cache_handler.py +++ b/src/cachekit/cache_handler.py @@ -1327,7 +1327,7 @@ def get_cached_value_with_freshness(self, cache_key: str) -> Optional[tuple[tupl self._handle_l2_read_error(e, cache_key) # raises when fail-closed (LAB-108) return None except Exception as e: - get_logger().warning(f"Backend operation failed for get on {cache_key}: {e}") + 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]]: @@ -1353,7 +1353,7 @@ async def get_cached_value_with_freshness_async(self, cache_key: str) -> Optiona await self._handle_l2_read_error_async(e, cache_key) # raises when fail-closed (LAB-108) return None except Exception as e: - get_logger().warning(f"Backend operation failed for get on {cache_key}: {e}") + get_logger().warning(f"Backend operation failed for get on {redact_cache_key(cache_key)}: {e}") return None async def get_cached_value_async(self, cache_key: str, refresh_ttl: Optional[int] = None) -> Optional[Any]: @@ -1805,10 +1805,10 @@ def get_with_freshness(self, key: str) -> Optional[tuple[bytes, bool]]: try: return self._with_backpressure_and_timeout(self.backend.get_with_freshness, key) except BackendError as e: - get_logger().error(f"Backend error getting key {key}: {e}") + get_logger().error(f"Backend error getting key {redact_cache_key(key)}: {e}") return None except Exception as e: - get_logger().error(f"Unexpected error getting key {key}: {e}") + 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]]: @@ -1819,10 +1819,10 @@ async def get_with_freshness_async(self, key: str) -> Optional[tuple[bytes, bool try: return await self._with_backpressure_and_timeout_async(self.backend.get_with_freshness, key) except BackendError as e: - get_logger().error(f"Backend error getting key {key}: {e}") + get_logger().error(f"Backend error getting key {redact_cache_key(key)}: {e}") return None except Exception as e: - get_logger().error(f"Unexpected error getting key {key}: {e}") + get_logger().error(f"Unexpected error getting key {redact_cache_key(key)}: {e}") return None def set( diff --git a/src/cachekit/decorators/wrapper.py b/src/cachekit/decorators/wrapper.py index 7182d7f..6562c7f 100644 --- a/src/cachekit/decorators/wrapper.py +++ b/src/cachekit/decorators/wrapper.py @@ -72,7 +72,7 @@ def _ttl_refresh_done_callback(task: asyncio.Task, cache_key: str) -> None: try: exc = task.exception() if exc is not None: - _logger.debug("Background TTL refresh failed for %s: %s", cache_key, exc) + _logger.debug("Background TTL refresh failed for %s: %s", redact_cache_key(cache_key), exc) except asyncio.CancelledError: # Task was cancelled (e.g., during shutdown) - this is expected, don't log pass @@ -831,7 +831,9 @@ def _l1_swr_acquire( 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) + _logger.debug( + "L1-only SWR refresh skipped for %s: arguments not deep-copyable: %s", redact_cache_key(cache_key), exc + ) return None def _l1_swr_task_done(task: asyncio.Task[None], cache_key: str) -> None: @@ -867,7 +869,7 @@ def _l1_swr_refresh_sync(cache_key: str, version: int, call_args: tuple[Any, ... 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) + _logger.debug("L1-only SWR background refresh failed for %s: %s", redact_cache_key(cache_key), exc) return _object_cache.complete_refresh(cache_key, version, result, ttl=ttl) finally: