fix(decorators): async lock double-check L2 hits record get telemetry (LAB-3769)

pyproject.toml

@@ -66,6 +66,22 @@ dependencies = [
     "xxhash>=3.5.0",
     # HTTP client for SaaS backend (cachekit.io)
     "httpx[http2]>=0.28.1",
+    # anyio is transitive via the mandatory httpx dependency above, so it ships
+    # to EVERY install, not just dev. Declared here rather than as a
+    # [tool.uv] constraint because that table is uv-local: it never reaches
+    # requires-dist, so `pip install cachekit` would ignore it. 4.14.2 fixes
+    # GHSA-82r6-8w77-94w6 / CVE-2026-63374 (IDNA-2003 hostname encoding lets a
+    # hijacked connection to an internationalised domain pass TLS certificate
+    # validation; CVSS 9.3), GHSA-5p39-cfhj-2xmp / CVE-2026-64847 (undrained
+    # process-pool stderr pipe deadlocks the worker) and GHSA-3w57-8xmc-8v26 /
+    # CVE-2026-63349 (extra_groups ignored, parent supplementary groups kept).
+    "anyio>=4.14.2",
+    # h2 arrives via the http2 extra on that same mandatory httpx dependency, so
+    # it ships to every install too, and was declared as a [tool.uv] constraint
+    # with the same no-op effect. 4.4.1 fixes GHSA-6hr6-w5qg-qmwg (duplicate Host
+    # headers forwarded across an HTTP/2 -> HTTP/1.1 downgrade — a request
+    # smuggling primitive).
+    "h2>=4.4.1",
 ]
 
 [project.optional-dependencies]
@@ -248,7 +264,11 @@ fuzz = [
     "atheris>=2.3.0",
 ]
 
-# Override vulnerable transitive dependencies
+# Override vulnerable DEV-ONLY transitive dependencies.
+# This table is uv-local: it constrains resolution of this repo's lockfile and
+# never reaches requires-dist, so a floor placed here does NOT protect anyone who
+# runs `pip install cachekit`. A transitive that reaches users belongs in
+# [project] dependencies instead — see the anyio/h2 entries above.
 [tool.uv]
 constraint-dependencies = [
     "urllib3>=2.7.0",
@@ -259,8 +279,4 @@ constraint-dependencies = [
     # confusion) and GHSA-jp4c-xjxw-mgf9 (self-update import ordering); 26.2 fixes
     # PYSEC-2026-3721 (doubly-encoded index URLs install to arbitrary paths).
     "pip>=26.2",
-    # h2 is a transitive dep (httpx[http2] -> h2). 4.4.1 fixes
-    # GHSA-6hr6-w5qg-qmwg (duplicate Host headers forwarded on HTTP/2 ->
-    # HTTP/1.1 downgrade — request smuggling primitive).
-    "h2>=4.4.1",
 ]

src/cachekit/decorators/wrapper.py

@@ -802,6 +802,64 @@ def _l1_backfill_from_l2(cache_key: str, cached_data: Any, is_stale: bool, fresh
             return
         _cached_keys.add(cache_key)
 
+    def _record_l2_hit_async(cached_data: Any, get_duration_ms: float) -> None:
+        """Record the telemetry for an async L2 hit — the uncontended read and
+        both post-lock double-check hits (LAB-3769) share this so a
+        thundering-herd hit is never invisible to cache_operations_total /
+        cache_info() just because it arrived via the lock's double-check.
+
+        size_bytes is computed here rather than read from the handler's
+        size_bytes tuple slot (LAB-348), so this label never depends on the
+        handler's tuple shape; the two agree on every in-contract async hit.
+
+        Best-effort, for the same reason _l1_backfill_from_l2 is (LAB-348):
+        every call site sits inside an `except Exception` that falls through to
+        a recompute, so a throwing metrics collector would silently turn a hit
+        already in hand into a full recompute — under exactly the stampede the
+        lock exists to absorb. Telemetry never costs a served hit.
+
+        The two clauses are deliberately split rather than narrowed to the
+        collector's own error types. Narrowing does NOT fail fast here: an
+        unexpected raise would land in the caller's `except Exception`, which
+        logs at DEBUG under "Double-check cache failed after lock acquisition"
+        and recomputes — quieter than this, misattributed, and a recompute per
+        contended hit. So the unexpected case is caught too, and made loud
+        instead: ERROR with the exception type named, which is the signal a
+        narrow clause was meant to produce.
+        """
+        try:
+            # Local stat first, external collector second: this is pure arithmetic
+            # under a lock and cannot realistically refuse, whereas the collector can
+            # — and once the hit is served anyway, a collector refusal must not leave
+            # cache_info() omitting a hit the caller was handed. Losing the counter to
+            # someone else's registry error is the same invisibility this helper exists
+            # to remove.
+            _stats.record_l2_hit(get_duration_ms)
+            features.set_operation_context("get", duration_ms=get_duration_ms)
+            features.record_success()
+            if features.collect_stats:
+                # Defensive encode for an out-of-contract backend: both async readers
+                # annotate this slot `bytes`, so the ternary is a guard, not a live path.
+                envelope = cached_data.encode("utf-8") if isinstance(cached_data, str) else cached_data
+                features.record_cache_operation(
+                    operation="get",
+                    namespace=namespace or "default",
+                    serializer="rust",
+                    success=True,
+                    duration_ms=get_duration_ms,
+                    size_bytes=len(envelope),
+                    hit=True,
+                )
+        except (ValueError, TypeError) as exc:
+            # The collector's documented refusals: duplicated timeseries, a label set
+            # that disagrees with the registered metric, a non-numeric observation.
+            logger().warning(f"L2 hit telemetry skipped: {redact_error_for_log(exc)}")
+        except Exception as exc:
+            # Not a collector refusal — a bug in the telemetry stack. Still must not
+            # cost the served hit, so surface it at ERROR with its type rather than
+            # letting the caller demote this hit into a recompute.
+            logger().error(f"L2 hit telemetry failed unexpectedly ({type(exc).__name__}): {redact_error_for_log(exc)}")
+
     def _l2_swr_try_begin(cache_key: str) -> bool:
         """Claim a revalidation slot for this key; False = already in flight or at capacity.
 
@@ -1740,23 +1798,7 @@ async def async_wrapper(*args: Any, **kwargs: Any) -> Any:
 
                     # Record cache hit (always compute for L2 latency stats)
                     get_duration_ms = (time.perf_counter() - start_time) * 1000
-                    features.set_operation_context("get", duration_ms=get_duration_ms)
-                    features.record_success()
-
-                    if features.collect_stats:
-                        # size_bytes: the encoded envelope, matching the bytes _l1_backfill_from_l2
-                        # stores and the L1 site's len(l1_bytes). A str envelope is UTF-8 encoded
-                        # first, so non-ASCII payloads report byte length, not character count.
-                        _l2_envelope = cached_data.encode("utf-8") if isinstance(cached_data, str) else cached_data
-                        features.record_cache_operation(
-                            operation="get",
-                            namespace=namespace or "default",
-                            serializer="rust",
-                            success=True,
-                            duration_ms=get_duration_ms,
-                            size_bytes=len(_l2_envelope),
-                            hit=True,
-                        )
+                    _record_l2_hit_async(cached_data, get_duration_ms)
 
                     # Update L1 cache with the L2 value (serialized bytes) for subsequent
                     # fast access — stale-exclusion + remaining-freshness bound (LAB-557).
@@ -1778,9 +1820,6 @@ async def async_wrapper(*args: Any, **kwargs: Any) -> Any:
                         # the opted-in flag (LAB-446). Still degrades gracefully.
                         warn_ttl_refresh_unsupported(_backend)
 
-                    # Record L2 hit with latency for cache_info()
-                    _stats.record_l2_hit(get_duration_ms)
-
                     # SWR: stale hit — value already in hand; revalidate in the
                     # background so no request pays the recompute at a TTL boundary.
                     # Gated on _l2_swr_active (not just the read gate above): a
@@ -1834,10 +1873,13 @@ async def async_wrapper(*args: Any, **kwargs: Any) -> Any:
                             # Routed through the operation handler: corrupt entries evict (#159),
                             # stale hits skip L1, fresh backfill bounded by fresh_for (LAB-557).
                             try:
+                                _dc_start = time.perf_counter()
                                 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, _size_bytes = cached_result
+                                    _dc_duration_ms = (time.perf_counter() - _dc_start) * 1000
+                                    _record_l2_hit_async(cached_data, _dc_duration_ms)
                                     _l1_backfill_from_l2(cache_key, cached_data, _dc_stale, _dc_fresh_for)
                                     return result
                             except DecryptionAuthenticationError:
@@ -1861,10 +1903,13 @@ async def async_wrapper(*args: Any, **kwargs: Any) -> Any:
                             try:
                                 # Routed through the operation handler: corrupt entries evict (#159),
                                 # stale hits skip L1, fresh backfill bounded by fresh_for (LAB-557).
+                                _dc_start = time.perf_counter()
                                 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, _size_bytes = cached_result
+                                    _dc_duration_ms = (time.perf_counter() - _dc_start) * 1000
+                                    _record_l2_hit_async(cached_data, _dc_duration_ms)
                                     _l1_backfill_from_l2(cache_key, cached_data, _dc_stale, _dc_fresh_for)
                                     return result
                             except DecryptionAuthenticationError:

tests/unit/test_async_l2_double_check_hit_labels.py

@@ -0,0 +1,175 @@
+"""Async lock double-check L2 hit-record parity (LAB-3769).
+
+The uncontended async L2 hit site records get/serializer="rust"/hit=True telemetry
+(LAB-3765); the two post-lock ``_l2_double_check`` hit returns did not — so a
+thundering-herd hit, filled by another worker while this one waited on the
+distributed lock, was invisible to ``cache_operations_total`` and ``cache_info()``
+L2 stats. That is exactly the traffic the lock exists to absorb.
+
+Reproduces contention by patching the backend's ``get`` to miss once (forcing the
+wrapper past the pre-lock check into the lock path) then hit on the next call —
+the double-check read standing in for "another request filled the cache while we
+waited". Parametrised over the lock outcome so both hit returns are covered: the
+lock-acquired branch and the lock-timeout branch, which record via the same
+``_record_l2_hit_async`` helper but are reached by different control flow.
+
+``_LockableByteStore`` is defined locally rather than imported from
+tests/unit/test_async_set_record_labels.py (cachekit-py#295): that file does not
+exist on `main` yet. Hoist to a shared fixture once #295 merges.
+"""
+
+from __future__ import annotations
+
+from collections.abc import AsyncIterator
+from contextlib import asynccontextmanager
+from typing import Any
+
+import pytest
+
+from cachekit import cache
+from cachekit.decorators.orchestrator import FeatureOrchestrator
+
+
+class _LockableByteStore:
+    """In-memory byte store implementing LockableBackend.
+
+    ``lock_acquired`` selects which double-check hit return is exercised: True
+    takes the lock-acquired branch, False the lock-timeout branch.
+    """
+
+    def __init__(self, *, lock_acquired: bool = True) -> None:
+        self.store: dict[str, bytes] = {}
+        self._lock_acquired = lock_acquired
+
+    def get(self, key: str) -> bytes | None:
+        return self.store.get(key)
+
+    def set(self, key: str, value: bytes, ttl: int | None = None) -> None:
+        self.store[key] = value
+
+    def delete(self, key: str) -> bool:
+        return self.store.pop(key, None) is not None
+
+    def exists(self, key: str) -> bool:
+        return key in self.store
+
+    def health_check(self) -> tuple[bool, dict[str, Any]]:
+        return True, {}
+
+    @asynccontextmanager
+    async def acquire_lock(self, key: str, timeout: float, blocking_timeout: float | None = None) -> AsyncIterator[bool]:
+        yield self._lock_acquired
+
+
+@pytest.fixture
+def recorded(monkeypatch: pytest.MonkeyPatch) -> list[dict[str, Any]]:
+    """Capture the wrapper's explicit features.record_cache_operation(...) calls.
+
+    Patched at the orchestrator, not the collector: record_success() also forwards
+    an operation-context record to the collector, which would shadow the labels
+    under test (same rationale as test_async_get_record_labels.py).
+    """
+    calls: list[dict[str, Any]] = []
+    monkeypatch.setattr(FeatureOrchestrator, "record_cache_operation", lambda self, **kw: calls.append(kw))
+    return calls
+
+
+@pytest.mark.unit
+@pytest.mark.parametrize("lock_acquired", [True, False], ids=["lock-acquired", "lock-timeout"])
+async def test_async_l2_double_check_hit_records_get(recorded: list[dict[str, Any]], lock_acquired: bool) -> None:
+    backend = _LockableByteStore(lock_acquired=lock_acquired)
+
+    @cache(backend=backend, ttl=60, namespace="async-dc-labels", l1_enabled=False)
+    async def compute() -> dict[str, int]:
+        return {"answer": 42}
+
+    assert await compute() == {"answer": 42}  # miss: primes L2 with the real envelope
+    assert backend.store
+    expected_size = len(next(iter(backend.store.values())))
+
+    # Make the pre-lock L2 check miss exactly once so the wrapper falls into the
+    # lock/double-check path; the primed value is still in the real store, so the
+    # double-check read inside the lock finds it — the contended-hit scenario.
+    real_get = backend.get
+    call_count = 0
+
+    def patched_get(key: str) -> bytes | None:
+        nonlocal call_count
+        call_count += 1
+        return None if call_count == 1 else real_get(key)
+
+    backend.get = patched_get  # type: ignore[method-assign]
+    recorded.clear()
+
+    assert await compute() == {"answer": 42}
+    assert call_count >= 2  # pre-lock miss, then the double-check hit
+
+    gets = [c for c in recorded if c["operation"] == "get"]
+    assert len(gets) == 1
+    assert (gets[0].get("serializer"), gets[0].get("hit")) == ("rust", True)
+    assert gets[0].get("size_bytes") == expected_size
+    # The double-check read is timed on its own window, so a duration is always
+    # recorded — a regression that drops it would still pass the label asserts.
+    assert isinstance(gets[0].get("duration_ms"), float)
+    assert gets[0]["duration_ms"] >= 0.0
+
+
+@pytest.mark.unit
+@pytest.mark.parametrize(
+    "raised",
+    [ValueError("duplicated timeseries"), AttributeError("collector refactored away")],
+    ids=["collector-refusal", "unexpected-bug"],
+)
+async def test_throwing_collector_does_not_cost_the_served_hit(monkeypatch: pytest.MonkeyPatch, raised: Exception) -> None:
+    """A throwing metrics collector must never demote a contended hit into a recompute.
+
+    Both double-check returns sit inside an `except Exception` that falls through to
+    executing the function again, so an unguarded telemetry call would turn the hit
+    this lock exists to protect into exactly the recompute it exists to prevent —
+    once per contender. Parametrised over both handler clauses in
+    `_record_l2_hit_async`: the collector's own refusal types, and an unexpected
+    error that is caught too (narrowing there would not fail fast, it would just
+    hand the raise to the caller's DEBUG-level handler and recompute anyway).
+    """
+    backend = _LockableByteStore()
+    calls = 0
+
+    @cache(backend=backend, ttl=60, namespace="async-dc-throw", l1_enabled=False)
+    async def compute() -> dict[str, int]:
+        nonlocal calls
+        calls += 1
+        return {"answer": 42}
+
+    assert await compute() == {"answer": 42}
+    assert calls == 1
+
+    real_get = backend.get
+    gets = 0
+
+    def patched_get(key: str) -> bytes | None:
+        nonlocal gets
+        gets += 1
+        return None if gets == 1 else real_get(key)
+
+    backend.get = patched_get  # type: ignore[method-assign]
+
+    def boom(self: Any, **kw: Any) -> None:
+        raise raised
+
+    monkeypatch.setattr(FeatureOrchestrator, "record_cache_operation", boom)
+
+    # Counters are keyed by module.qualname and shared across decorator
+    # applications (see cache_info's docstring), so both parametrised runs share
+    # one set — assert the delta, not an absolute.
+    l2_hits_before = compute.cache_info().l2_hits
+
+    # The hit is still served from the double-check read, and the function body
+    # never runs a second time.
+    assert await compute() == {"answer": 42}
+    assert calls == 1
+
+    # ...and cache_info() still counts it. A refusal from the external collector
+    # must not cost the local L2 stat, or a served hit goes missing from
+    # cache_info().l2_hits and the average L2 latency — the same invisibility
+    # this helper exists to remove, just moved to a different sink.
+    assert compute.cache_info().l2_hits == l2_hits_before + 1