diff --git a/src/cachekit/decorators/wrapper.py b/src/cachekit/decorators/wrapper.py index ba154d5b..e5bbe993 100644 --- a/src/cachekit/decorators/wrapper.py +++ b/src/cachekit/decorators/wrapper.py @@ -1706,70 +1706,66 @@ def sync_wrapper(*args: Any, **kwargs: Any) -> Any: # noqa: PLR0912 token = set_current_function_stats(_stats) - cache_key = None # Initialize to avoid UnboundLocalError - - # Key generation - needed for both L1-only and L1+L2 modes try: - cache_key = _resolve_cache_key(args, kwargs) - except Exception as e: - if interop is not None: - # Interop/v1: out-of-model arguments MUST be rejected with an - # error — never silently degrade to uncached execution. - reset_current_function_stats(token) - raise - # Key generation failed - execute function without caching - features.handle_cache_error( - error=e, - operation="key_generation", - cache_key="", - namespace=namespace or "default", - duration_ms=0.0, - count_toward_breaker=False, # pre-admission, never reached the backend - ) - reset_current_function_stats(token) - return func(*args, **kwargs) + cache_key = None # Initialize to avoid UnboundLocalError - # 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 + # Key generation - needed for both L1-only and L1+L2 modes try: + cache_key = _resolve_cache_key(args, kwargs) + except Exception as e: + if interop is not None: + # Interop/v1: out-of-model arguments MUST be rejected with an + # error — never silently degrade to uncached execution. + raise + # Key generation failed - execute function without caching + features.handle_cache_error( + error=e, + operation="key_generation", + cache_key="", + namespace=namespace or "default", + duration_ms=0.0, + count_toward_breaker=False, # pre-admission, never reached the backend + ) return func(*args, **kwargs) - finally: - reset_current_function_stats(token) - if _l1_only_mode and _object_cache: - if _l1_swr_active and ttl is not None: - found, cached_value, needs_refresh, version = _object_cache.get_with_swr(cache_key, ttl) - 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: - slot, refresh_args, refresh_kwargs = snapshot - try: - threading.Thread( - target=_l1_swr_refresh_sync, - args=(slot, cache_key, version, refresh_args, refresh_kwargs), - name="cachekit-swr-refresh", # no function or key metadata (CWE-532) - daemon=True, - ).start() - except RuntimeError as exc: - # Thread couldn't start (resource pressure) — release - # the slot and this exact refresh so a later call retries - _l1_swr_pool.release(slot) - _object_cache.cancel_refresh(cache_key, version) - _warn_refresh(_refresh_unstarted_warn, "L1-only SWR refresh could not be started", cache_key, exc) - reset_current_function_stats(token) - return cached_value - - # Cache miss: run the function once for every concurrent miss on this key. A caller - # that joins gets the same object, as a later hit does, and counts as a hit. - try: + + # 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 + return func(*args, **kwargs) + if _l1_only_mode and _object_cache: + if _l1_swr_active and ttl is not None: + found, cached_value, needs_refresh, version = _object_cache.get_with_swr(cache_key, ttl) + 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: + slot, refresh_args, refresh_kwargs = snapshot + try: + threading.Thread( + target=_l1_swr_refresh_sync, + args=(slot, cache_key, version, refresh_args, refresh_kwargs), + name="cachekit-swr-refresh", # no function or key metadata (CWE-532) + daemon=True, + ).start() + except RuntimeError as exc: + # Thread couldn't start (resource pressure) — release + # the slot and this exact refresh so a later call retries + _l1_swr_pool.release(slot) + _object_cache.cancel_refresh(cache_key, version) + _warn_refresh( + _refresh_unstarted_warn, "L1-only SWR refresh could not be started", cache_key, exc + ) + return cached_value + + # Cache miss: run the function once for every concurrent miss on this key. A caller + # that joins gets the same object, as a later hit does, and counts as a hit. result, joined = _thread_flights.run( (cache_key,), functools.partial(_l1_only_fill, cache_key, args, kwargs), @@ -1778,287 +1774,251 @@ def sync_wrapper(*args: Any, **kwargs: Any) -> Any: # noqa: PLR0912 if joined: _stats.record_l1_hit() return result - finally: - reset_current_function_stats(token) - # L1+L2 MODE: Original behavior with backend initialization + # L1+L2 MODE: Original behavior with backend initialization - # Tenant scope, before the breaker check and outside every degrade try (LAB-5713): an - # unsupported tenant id type is a caller bug, so its UnsupportedTenantError reaches the - # caller before the function runs, whatever the breaker state, and never counts a failure - # on the breaker every tenant of this function shares. "" until the backend is resolved; - # the first call checks right after resolving it, below. Sits outside the main - # try/finally, so the raise path restores the context itself. - try: + # Tenant scope, before the breaker check and outside every degrade try (LAB-5713): an + # unsupported tenant id type is a caller bug, so its UnsupportedTenantError reaches the + # caller before the function runs, whatever the breaker state, and never counts a failure + # on the breaker every tenant of this function shares. "" until the backend is resolved; + # the first call checks right after resolving it, below. _l2_scope() - except Exception: - reset_current_function_stats(token) - raise - nonlocal _backend + nonlocal _backend - # Interop fail-closed guard (CWE-636): a key-prefixing backend would make - # this SDK read/write a key other SDKs cannot see. Re-checked on every call - # because the backend is lazily resolved and a prefix could appear - # dynamically. It runs before the L1 lookup, because an L1 hit returns early - # and must not skip it (LAB-5351). ensure_interop_backend_compatible(None) - # is a no-op, so until the backend is resolved an interop call skips L1 and - # is checked right after resolution, below. The raise paths sit outside the - # main try/finally, so they restore the stats context themselves (see - # test_context_leak_regression.py). - interop_checked = False - if interop is not None and _backend is not None: - try: + # Interop fail-closed guard (CWE-636): a key-prefixing backend would make + # this SDK read/write a key other SDKs cannot see. Re-checked on every call + # because the backend is lazily resolved and a prefix could appear + # dynamically. It runs before the L1 lookup, because an L1 hit returns early + # and must not skip it (LAB-5351). ensure_interop_backend_compatible(None) + # is a no-op, so until the backend is resolved an interop call skips L1 and + # is checked right after resolution, below. + interop_checked = False + if interop is not None and _backend is not None: ensure_interop_backend_compatible(_backend) - except Exception: - reset_current_function_stats(token) - raise - interop_checked = True - - # Guard clause: L1 cache check first - early return eliminates network latency. - # It runs before the breaker's admission check and records no breaker outcome: - # the breaker tracks backend health, and an L1 hit never reaches the backend, - # so it is served whatever the breaker state (LAB-5351). It waits for the backend: - # an encrypted entry's AAD binds the backend's key prefix, which is unknown until - # then, so an entry another wrapper wrote here would read as tampered. - if _l1_cache and cache_key and _backend is not None and (interop is None or interop_checked): - l1_start = time.perf_counter() - l1_found, l1_bytes = _l1_cache.get(cache_key) - if l1_found and l1_bytes: - # L1 cache hit (~50ns vs ~1000μs for Redis) - deserialize bytes - try: - l1_value = operation_handler.serialization_handler.deserialize_data(l1_bytes, cache_key, args, kwargs) - l1_duration_ms = (time.perf_counter() - l1_start) * 1000 # L1 get plus deserialize + interop_checked = True - # Record L1 cache hit metrics - if features.collect_stats: - features.record_cache_operation( - operation="get", + # Guard clause: L1 cache check first - early return eliminates network latency. + # It runs before the breaker's admission check and records no breaker outcome: + # the breaker tracks backend health, and an L1 hit never reaches the backend, + # so it is served whatever the breaker state (LAB-5351). It waits for the backend: + # an encrypted entry's AAD binds the backend's key prefix, which is unknown until + # then, so an entry another wrapper wrote here would read as tampered. + if _l1_cache and cache_key and _backend is not None and (interop is None or interop_checked): + l1_start = time.perf_counter() + l1_found, l1_bytes = _l1_cache.get(cache_key) + if l1_found and l1_bytes: + # L1 cache hit (~50ns vs ~1000μs for Redis) - deserialize bytes + try: + l1_value = operation_handler.serialization_handler.deserialize_data(l1_bytes, cache_key, args, kwargs) + l1_duration_ms = (time.perf_counter() - l1_start) * 1000 # L1 get plus deserialize + + # Record L1 cache hit metrics + if features.collect_stats: + features.record_cache_operation( + operation="get", + namespace=namespace or "default", + serializer="l1_memory", + success=True, + duration_ms=l1_duration_ms, + size_bytes=len(l1_bytes), + ) + + features.log_cache_operation( + operation="l1_get", + key=cache_key, namespace=namespace or "default", serializer="l1_memory", - success=True, duration_ms=l1_duration_ms, - size_bytes=len(l1_bytes), + hit=True, + ttl=ttl, ) - features.log_cache_operation( - operation="l1_get", - key=cache_key, - namespace=namespace or "default", - serializer="l1_memory", - duration_ms=l1_duration_ms, - hit=True, - ttl=ttl, - ) - - # Record L1 hit for cache_info() - _stats.record_l1_hit() - - # WHY: L1 cache hit returns BEFORE the try-finally block (line ~642-713) - # that handles context cleanup. Without this explicit reset, the contextvar - # leaks to subsequent calls, causing stats pollution between requests. - # ~34ns overhead, but required for correctness. See test_context_leak_regression.py - reset_current_function_stats(token) - return l1_value - except TenantResolutionError: - # No tenant in the caller's context: nothing to decrypt as, and nothing wrong - # with the entry, which stays. The L2 read below misses for the same reason. - logger().debug(f"L1 read skipped for {redact_cache_key(cache_key)}: caller's tenant unresolved") - except SerializationError as e: - # Poisoned L1 must not outlive remediation of the durable L2 copy — - # invalidate BEFORE the policy decision (a fail-closed raise would - # otherwise keep re-raising from stale process-local L1 after the - # operator fixes L2). L2 remains the retained evidence. - _l1_cache.invalidate(cache_key) - if not _foreign_l1_entry(e): - try: - # Single policy point (cachekit-py#170): metric + fail policy. - handle_decrypt_failure( - e, tier="l1", cache_key=cache_key, fail_closed=serialization_handler.encryption_fail_closed - ) - except DecryptionAuthenticationError: - reset_current_function_stats(token) - raise - # Fail open, or an L1 keying collision (_foreign_l1_entry): fall through to L2 - except KeyringConfigurationError: - # LOCAL keyring config fault — not a poisoned L1 entry, so - # neither the invalidate nor the "deserialization failed" - # message below is true, and swallowing it here degrades a - # misconfigured keyring into a silent L2 fall-through. Same - # re-raise as the L2 sites in cache_handler.py. - # - # The sync wrapper has no outer `finally`, so a raising exit - # must reset the stats token by hand — exactly as the - # DecryptionAuthenticationError sibling above does. (The - # async L1 guard needs no reset; its wrapper's outer - # `finally` covers every exit path.) - reset_current_function_stats(token) - raise - except Exception as e: - # L1 deserialization failed - invalidate and continue to L2 - logger().warning( - f"L1 cache deserialization failed for {redact_cache_key(cache_key)}: {redact_error_for_log(e)}" - ) - _l1_cache.invalidate(cache_key) - - # Guard clause: the circuit breaker rejected this call (OPEN, or HALF_OPEN - # with its probe budget spent) - run the function uncached. This sits - # outside the try below on purpose: that except records a failure, and a - # rejection is not one. Recorded, every rejected call would push the OPEN - # window forward and reopen HALF_OPEN, so the breaker never recovers. - probe_cycle = features.admit() # None when rejected; 0 is an admission too - if probe_cycle is None: - _log_breaker_rejection(cache_key) - reset_current_function_stats(token) - return _uncached_result(func(*args, **kwargs)) - - try: - if _backend is None: - _backend = _resolve_lazy_backend() - _l2_scope() # first call: the tenant check above ran before the backend existed + # Record L1 hit for cache_info() + _stats.record_l1_hit() + return l1_value + except TenantResolutionError: + # No tenant in the caller's context: nothing to decrypt as, and nothing wrong + # with the entry, which stays. The L2 read below misses for the same reason. + logger().debug(f"L1 read skipped for {redact_cache_key(cache_key)}: caller's tenant unresolved") + except SerializationError as e: + # Poisoned L1 must not outlive remediation of the durable L2 copy — + # invalidate BEFORE the policy decision (a fail-closed raise would + # otherwise keep re-raising from stale process-local L1 after the + # operator fixes L2). L2 remains the retained evidence. + _l1_cache.invalidate(cache_key) + if not _foreign_l1_entry(e): + # Explicit local re-raise, as in async_wrapper's L1 guard: it keeps a + # fail-closed tamper raise ahead of any `except Exception` later added + # to this inner try. It cannot stop a broad handler wrapped around this + # read path from outside; test_fail_closed_invalidates_poisoned_l1_before_raising + # pins that the raise reaches the caller. + try: + # Single policy point (cachekit-py#170): metric + fail policy. + handle_decrypt_failure( + e, tier="l1", cache_key=cache_key, fail_closed=serialization_handler.encryption_fail_closed + ) + except DecryptionAuthenticationError: + raise + # Fail open, or an L1 keying collision (_foreign_l1_entry): fall through to L2 + except KeyringConfigurationError: + # LOCAL keyring config fault — not a poisoned L1 entry, so + # neither the invalidate nor the "deserialization failed" + # message below is true, and swallowing it here degrades a + # misconfigured keyring into a silent L2 fall-through. Same + # re-raise as the L2 sites in cache_handler.py. + raise + except Exception as e: + # L1 deserialization failed - invalidate and continue to L2 + logger().warning( + f"L1 cache deserialization failed for {redact_cache_key(cache_key)}: {redact_error_for_log(e)}" + ) + _l1_cache.invalidate(cache_key) - _install_cache_handler(_backend) - except UnsupportedTenantError: - # From the first-call check above, or from a provider that checks the tenant while - # building the backend (RedisBackendProvider.get_backend): a caller bug, not a - # client failure, so never degraded or counted (LAB-5713). - reset_current_function_stats(token) - raise - except Exception as e: - # Guard clause: Client creation failed - early return with fallback - features.handle_cache_error( - error=e, - operation="client_creation", - cycle=probe_cycle, - cache_key=cache_key or "unknown", - namespace=namespace or "default", - duration_ms=0.0, - serializer="rust", - ) - # WHY: Early return on backend failure - outside main try-finally, needs explicit cleanup - reset_current_function_stats(token) - return _uncached_result(func(*args, **kwargs)) + # Guard clause: the circuit breaker rejected this call (OPEN, or HALF_OPEN + # with its probe budget spent) - run the function uncached. This sits + # outside the try below on purpose: that except records a failure, and a + # rejection is not one. Recorded, every rejected call would push the OPEN + # window forward and reopen HALF_OPEN, so the breaker never recovers. + probe_cycle = features.admit() # None when rejected; 0 is an admission too + if probe_cycle is None: + _log_breaker_rejection(cache_key) + return _uncached_result(func(*args, **kwargs)) - # First interop call: the check above had no backend to check (see there). - if interop is not None and not interop_checked: try: - ensure_interop_backend_compatible(_backend) - except Exception: - reset_current_function_stats(token) + if _backend is None: + _backend = _resolve_lazy_backend() + _l2_scope() # first call: the tenant check above ran before the backend existed + + _install_cache_handler(_backend) + except UnsupportedTenantError: + # From the first-call check above, or from a provider that checks the tenant while + # building the backend (RedisBackendProvider.get_backend): a caller bug, not a + # client failure, so never degraded or counted (LAB-5713). raise + except Exception as e: + # Guard clause: Client creation failed - early return with fallback + features.handle_cache_error( + error=e, + operation="client_creation", + cycle=probe_cycle, + cache_key=cache_key or "unknown", + namespace=namespace or "default", + duration_ms=0.0, + serializer="rust", + ) + return _uncached_result(func(*args, **kwargs)) - if _l1_cache and invalidation.listener_start_due(_backend): - invalidation.start_listener(_backend) - twin_key = _twin_key(args, kwargs) # after the L1 lookup: an L1 hit records nothing + # First interop call: the check above had no backend to check (see there). + if interop is not None and not interop_checked: + ensure_interop_backend_compatible(_backend) - # Continue with the rest of the sync wrapper logic... - # Try to get cached value with optional TTL refresh - start_time = time.perf_counter() - try: - refresh_ttl = ttl if refresh_ttl_on_get and ttl else None - - # Use operation handler for all cache access (uses backend internally). - # 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). - _sync_l2_stale = False - _sync_l2_fresh_for: int | None = None - if _l2_freshness_capable(): - _fresh_hit = operation_handler.get_cached_value_with_freshness(cache_key, args, kwargs) - 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 - _sync_l2_fresh_for = _fresh_hit[2] if _fresh_hit is not None else None - else: - cached_result = operation_handler.get_cached_value(cache_key, refresh_ttl, args, kwargs) + if _l1_cache and invalidation.listener_start_due(_backend): + invalidation.start_listener(_backend) + twin_key = _twin_key(args, kwargs) # after the L1 lookup: an L1 hit records nothing - duration = time.perf_counter() - start_time + # Continue with the rest of the sync wrapper logic... + # Try to get cached value with optional TTL refresh + start_time = time.perf_counter() + try: + refresh_ttl = ttl if refresh_ttl_on_get and ttl else None + + # Use operation handler for all cache access (uses backend internally). + # 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). + _sync_l2_stale = False + _sync_l2_fresh_for: int | None = None + if _l2_freshness_capable(): + _fresh_hit = operation_handler.get_cached_value_with_freshness(cache_key, args, kwargs) + 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 + _sync_l2_fresh_for = _fresh_hit[2] if _fresh_hit is not None else None + else: + cached_result = operation_handler.get_cached_value(cache_key, refresh_ttl, args, kwargs) - if cached_result is not None: - # Cache hit: envelope is None on the mmap fast path; size_bytes is set on every path - result, cached_data, size_bytes = cached_result.value, cached_result.envelope, cached_result.size_bytes - features.record_success(probe_cycle) + duration = time.perf_counter() - start_time - # Record cache hit with structured logging - features.log_cache_operation( - operation="get", - key=cache_key, - namespace=namespace or "default", - serializer="rust", - duration_ms=duration * 1000, - hit=True, - ttl=ttl, - ) + if cached_result is not None: + # Cache hit: envelope is None on the mmap fast path; size_bytes is set on every path + result, cached_data, size_bytes = cached_result.value, cached_result.envelope, cached_result.size_bytes + features.record_success(probe_cycle) - # Also record statistics if enabled - if features.collect_stats: - features.record_cache_operation( + # Record cache hit with structured logging + features.log_cache_operation( operation="get", + key=cache_key, namespace=namespace or "default", serializer="rust", - success=True, duration_ms=duration * 1000, - size_bytes=size_bytes, + hit=True, + ttl=ttl, ) - # Backfill L1 with the L2 envelope for subsequent fast access — stale-exclusion - # + remaining-freshness bound (LAB-557), as on the async path (LAB-348). - _l1_backfill_from_l2(cache_key, cached_data, _sync_l2_stale, _sync_l2_fresh_for, twin=twin_key) - - # Record L2 hit with latency for cache_info() - duration_ms = duration * 1000 - _stats.record_l2_hit(duration_ms) - - # SWR: stale hit — serve now, revalidate on a daemon thread. - # 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) + # Also record statistics if enabled + if features.collect_stats: + features.record_cache_operation( + operation="get", + namespace=namespace or "default", + serializer="rust", + success=True, + duration_ms=duration * 1000, + size_bytes=size_bytes, + ) - # 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) - reset_current_function_stats(token) - return result - except (DecryptionAuthenticationError, KeyringConfigurationError): - # Both propagate from get_cached_value* and must reach the caller: a - # fail-closed tamper failure (raised only when encryption.fail_closed=True; - # the metric and error log were recorded there), and a LOCAL keyring config - # fault (same contract as the L1 guard above). The generic clause below - # would log either as a cache error, count it on the breaker, and recompute - # uncached on every call (LAB-4841). - reset_current_function_stats(token) - raise - except Exception as e: - # Cache GET failed - execute function without caching - get_duration_ms = (time.perf_counter() - start_time) * 1000 - features.handle_cache_error( - error=e, - operation="cache_get", - cycle=probe_cycle, - cache_key=cache_key or "unknown", - namespace=namespace or "default", - duration_ms=get_duration_ms, - serializer="rust", - ) - # WHY: Early return on cache GET failure - same reason as L2 hit path - reset_current_function_stats(token) - return _uncached_result(func(*args, **kwargs)) + # Backfill L1 with the L2 envelope for subsequent fast access — stale-exclusion + # + remaining-freshness bound (LAB-557), as on the async path (LAB-348). + _l1_backfill_from_l2(cache_key, cached_data, _sync_l2_stale, _sync_l2_fresh_for, twin=twin_key) + + # Record L2 hit with latency for cache_info() + duration_ms = duration * 1000 + _stats.record_l2_hit(duration_ms) + + # SWR: stale hit — serve now, revalidate on a daemon thread. + # 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) + + return result + except (DecryptionAuthenticationError, KeyringConfigurationError): + # Both propagate from get_cached_value* and must reach the caller: a + # fail-closed tamper failure (raised only when encryption.fail_closed=True; + # the metric and error log were recorded there), and a LOCAL keyring config + # fault (same contract as the L1 guard above). The generic clause below + # would log either as a cache error, count it on the breaker, and recompute + # uncached on every call (LAB-4841). + raise + except Exception as e: + # Cache GET failed - execute function without caching + get_duration_ms = (time.perf_counter() - start_time) * 1000 + features.handle_cache_error( + error=e, + operation="cache_get", + cycle=probe_cycle, + cache_key=cache_key or "unknown", + namespace=namespace or "default", + duration_ms=get_duration_ms, + serializer="rust", + ) + return _uncached_result(func(*args, **kwargs)) - # CACHE MISS - Execute function and cache result - # Note: Sync wrappers don't support distributed locking (backend protocol is async-only) - # For thundering herd protection, use async decorators instead - # Record miss for cache_info() - _stats.record_miss() + # CACHE MISS - Execute function and cache result + # Note: Sync wrappers don't support distributed locking (backend protocol is async-only) + # For thundering herd protection, use async decorators instead + # Record miss for cache_info() + _stats.record_miss() - try: # Execute the original function. Its exception is not a backend failure, so it is - # never counted, but an admitted HALF_OPEN probe hands its slot to the next call - # instead of holding it until the cycle expires. probe_cycle confines that to a - # slot this call took: a call admitted while CLOSED, or by an ended cycle, gives - # the current cycle nothing back. + # never counted, and the function never reruns for it: rerunning it for a BackendError + # it raised would repeat its side effects. An admitted HALF_OPEN probe does hand its + # slot to the next call instead of holding it until the cycle expires. probe_cycle + # confines that to a slot this call took: a call admitted while CLOSED, or by an + # ended cycle, gives the current cycle nothing back. try: result = func(*args, **kwargs) except Exception: @@ -2113,12 +2073,10 @@ def sync_wrapper(*args: Any, **kwargs: Any) -> Any: # noqa: PLR0912 return result - # No handler here: the store's own failures are caught above, so what reaches this - # point is the function's exception (or a fail-loud write error). It is not a backend - # failure, so it never counts toward the breaker, and rerunning the function for a - # BackendError it raised would repeat its side effects. finally: - # ALWAYS reset stats context, even on exception + # The one reset, as in async_wrapper: every exit restores the caller's context here, + # an interrupt included, which no `except Exception` above sees. A token resets once: + # a second reset raises RuntimeError, so no path may also reset by hand. reset_current_function_stats(token) @functools.wraps(func) @@ -2249,9 +2207,10 @@ async def async_wrapper(*args: Any, **kwargs: Any) -> Any: # Explicit local re-raise mirrors the sync L1/L2 fail-closed # guards and the async lock-path guard: a fail-closed tamper raise # must reach the caller, never be demoted to a fail-open recompute - # if a future edit wraps this read path in a broad `except - # Exception` (defense-in-depth, LAB-108). No manual stats reset — - # the async wrapper's outer `finally` covers every exit path. + # (defense-in-depth, LAB-108). The clause keeps it ahead of any + # `except Exception` later added to this inner try; it cannot stop a + # broad handler wrapped around this read path from outside, which + # test_async_fail_closed_invalidates_poisoned_l1_before_raising catches. if not _foreign_l1_entry(e): try: handle_decrypt_failure( diff --git a/tests/unit/backends/test_redis_error_frames.py b/tests/unit/backends/test_redis_error_frames.py index 778d20ae..f4008a7e 100644 --- a/tests/unit/backends/test_redis_error_frames.py +++ b/tests/unit/backends/test_redis_error_frames.py @@ -22,7 +22,6 @@ import asyncio import concurrent.futures -import contextvars import errno import gc import logging @@ -53,6 +52,7 @@ from cachekit.backends.redis.client import reset_global_pool from cachekit.backends.redis.error_handler import RedisClientError, classify_redis_error from cachekit.backends.redis.provider import PerRequestRedisBackend, RedisBackendProvider +from cachekit.decorators.stats_context import get_current_function_stats from cachekit.hash_utils import redact_cache_key from tests.unit.config.test_redacting_settings import _CACHEKIT_SRC, _cachekit_locals_holding, _held_exception @@ -765,10 +765,10 @@ def double(x: int) -> int: fn = cache(backend=backend, ttl=60, l1_enabled=False, namespace=f"redis-interrupt-{uuid.uuid4().hex}")(double) with _interrupting_sends(interrupt): - # In a copy of the context: the interrupt cuts the call short, before the decorator restores its own context. - exc = _raised(lambda: contextvars.copy_context().run(fn, 1), interrupt) + exc = _raised(lambda: fn(1), interrupt) _assert_cleared(exc, interrupt, password, through=exc) + assert get_current_function_stats() is None # the decorator restored the caller's context on the way out breaker = fn.get_health_status()["circuit_breaker"] assert (breaker["state"], breaker["failure_count"]) == ("closed", 0) diff --git a/tests/unit/test_context_leak_regression.py b/tests/unit/test_context_leak_regression.py index 4f93bcf3..05e12aba 100644 --- a/tests/unit/test_context_leak_regression.py +++ b/tests/unit/test_context_leak_regression.py @@ -25,7 +25,24 @@ from cachekit import cache from cachekit.backends.errors import BackendError, BackendErrorType -from cachekit.decorators.stats_context import get_current_function_stats +from cachekit.decorators.stats_context import ( + get_current_function_stats, + reset_current_function_stats, + set_current_function_stats, +) + + +@pytest.fixture(autouse=True) +def _isolated_stats_context(): + """Start each test with no stats context and restore the outer one after it. + + A leak stays set in the pytest thread's context, so without this one leaking + test would also fail every later test here that checks for None, and the + failure would point at the wrong test. + """ + token = set_current_function_stats(None) + yield + reset_current_function_stats(token) @pytest.mark.unit @@ -38,7 +55,7 @@ def test_context_reset_on_backend_init_failure(self): Simulates backend provider.get_backend() raising exception. Function should execute uncached and context should be clean. """ - with patch("cachekit.cache_handler.get_backend_provider") as mock_provider: + with patch("cachekit.decorators.wrapper.get_backend_provider") as mock_provider: # Mock backend provider to fail mock_provider.return_value.get_backend.side_effect = RuntimeError("Backend init failed") @@ -49,6 +66,7 @@ def test_func(x): # Call function (should fallback to uncached execution) result = test_func(5) assert result == 10 + mock_provider.return_value.get_backend.assert_called_once() # Verify context is None (no leak) assert get_current_function_stats() is None @@ -356,6 +374,32 @@ def exit_func(x): # Verify context is None (no leak) assert get_current_function_stats() is None + @pytest.mark.parametrize("interrupt", [KeyboardInterrupt, SystemExit]) + @pytest.mark.parametrize("site", ["l2_read", "backend_resolution"]) + def test_context_reset_on_interrupt_from_backend_io(self, mock_backend, site, interrupt): + """An interrupt from the decorator's own backend I/O must reset context too. + + KeyboardInterrupt and SystemExit are BaseExceptions, so the decorator's + `except Exception` degrade handlers never see them; only a finally does. + The call runs in this test's own context: a copied one would hide a leak. + """ + with patch("cachekit.decorators.wrapper.get_backend_provider") as mock_provider: + if site == "l2_read": + mock_backend.get = Mock(side_effect=interrupt) + decorator = cache(backend=mock_backend) + else: + mock_provider.return_value.get_backend.side_effect = interrupt + decorator = cache() + + @decorator + def test_func(x): + return x * 2 + + with pytest.raises(interrupt): + test_func(5) + + assert get_current_function_stats() is None + def test_context_reset_after_recursive_calls(self, mock_backend): """Recursive decorated functions should maintain context isolation. @@ -462,6 +506,29 @@ async def error_func(): # Stats context should be reset assert get_current_function_stats() is None + @pytest.mark.asyncio + @pytest.mark.parametrize("interrupt", [KeyboardInterrupt, SystemExit]) + async def test_async_context_reset_on_interrupt_from_backend_resolution(self, interrupt): + """An interrupt from backend resolution must reset context, as on the sync path. + + Awaited directly, so the wrapper runs in this test's task and a leak would + show here; a separate task would run in a copied context. Raised from backend + resolution, not the L2 read: a miss reads L2 in its single-flight task, and + asyncio re-raises an interrupt from a task out of the event loop, not into + the awaiting caller. + """ + with patch("cachekit.decorators.wrapper.get_backend_provider") as mock_provider: + mock_provider.return_value.get_backend.side_effect = interrupt + + @cache() + async def async_func(x): + return x * 2 + + with pytest.raises(interrupt): + await async_func(5) + + assert get_current_function_stats() is None + @pytest.mark.unit class TestAsyncContextLeaksIntegration: