perf(decorators): backfill L1 on sync L2 hits; size stats by envelope length (LAB-348)

src/cachekit/cache_handler.py

@@ -1354,15 +1354,23 @@ async def _handle_l2_read_error_async(self, e: SerializationError, cache_key: st
             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]:
+    def get_cached_value(
+        self, cache_key: str, refresh_ttl: Optional[int] = None
+    ) -> Optional[tuple[bool, Any, Optional[bytes], int]]:
         """Get value from cache if it exists.
 
         Args:
             cache_key: Cache key to retrieve (also used for AAD verification if encrypted)
             refresh_ttl: Optional TTL to refresh on hit
 
         Returns:
-            Tuple (True, value) if cache hit, None if cache miss or error
+            Tuple (True, value, raw_bytes, size_bytes) if cache hit, None if cache
+            miss or error. raw_bytes is the serialized envelope so the decorator can
+            backfill L1 without re-serializing (re-encrypting) — same shape as the
+            async variant (LAB-348). It is None on the mmap fast path: the mapped view
+            is confined to this frame and must never reach L1 (#171 blocker C).
+            size_bytes is the envelope's byte length on every hit, mmap included, so
+            payload-size stats never depend on holding the bytes.
 
         Note:
             Requires cache_handler to be set via set_cache_handler() before calling.
@@ -1381,7 +1389,8 @@ def get_cached_value(self, cache_key: str, refresh_ttl: Optional[int] = None) ->
                 if handle is not None:
                     try:
                         get_logger().cache_hit(cache_key, "Backend(mmap)")
-                        return (True, self.serialization_handler.deserialize_data(handle.view, cache_key))
+                        size_bytes = handle.view.nbytes  # payload length; the view is released in `finally`
+                        return (True, self.serialization_handler.deserialize_data(handle.view, cache_key), None, size_bytes)
                     finally:
                         handle.close()
 
@@ -1390,8 +1399,8 @@ def get_cached_value(self, cache_key: str, refresh_ttl: Optional[int] = None) ->
                 get_logger().cache_hit(cache_key, "Backend")
                 # Pass cache_key for AAD verification (required for encrypted data)
                 deserialized = self.serialization_handler.deserialize_data(cached_data, cache_key)
-                # Return a tuple (True, value) to distinguish from "no cache entry"
-                return (True, deserialized)
+                # Tuple distinguishes a hit from "no cache entry"; raw bytes ride along for L1
+                return (True, deserialized, cached_data, len(cached_data))
             return None
         except KeyringConfigurationError:
             # LOCAL keyring config fault (bad tenant_id, bad keyring entry index) —
@@ -1410,12 +1419,16 @@ 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, Optional[int]]]:
+    def get_cached_value_with_freshness(
+        self, cache_key: str
+    ) -> Optional[tuple[tuple[bool, Any, bytes, int], 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, fresh_for)`` on a hit, None on
-        miss/error. fresh_for is None when no signal exists (pre-signal server,
+        Returns ``((True, value, raw_bytes, size_bytes), is_stale, fresh_for)`` on
+        a hit — the inner tuple matches the async variant so the sync decorator
+        backfills L1 without re-serializing (LAB-348) — 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
@@ -1436,7 +1449,7 @@ def get_cached_value_with_freshness(self, cache_key: str) -> Optional[tuple[tupl
             cached_data, is_stale, fresh_for = hit
             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, fresh_for)
+            return ((True, deserialized, cached_data, len(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
@@ -1456,12 +1469,12 @@ def get_cached_value_with_freshness(self, cache_key: str) -> Optional[tuple[tupl
 
     async def get_cached_value_with_freshness_async(
         self, cache_key: str
-    ) -> Optional[tuple[tuple[bool, Any, bytes], bool, Optional[int]]]:
+    ) -> Optional[tuple[tuple[bool, Any, bytes, int], 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
+        Returns ``((True, value, raw_bytes, size_bytes), is_stale, fresh_for)`` on
+        a hit — the inner 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
@@ -1477,7 +1490,7 @@ async def get_cached_value_with_freshness_async(
             cached_data, is_stale, fresh_for = hit  # same 2-tuple contract as the sync variant above
             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, fresh_for)
+            return ((True, deserialized, cached_data, len(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
@@ -1495,17 +1508,20 @@ async def get_cached_value_with_freshness_async(
             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]:
+    async def get_cached_value_async(
+        self, cache_key: str, refresh_ttl: Optional[int] = None
+    ) -> Optional[tuple[bool, Any, bytes, int]]:
         """Get value from cache if it exists (async version).
 
         Args:
             cache_key: Cache key to retrieve (also used for AAD verification if encrypted)
             refresh_ttl: Optional TTL to refresh on hit
 
         Returns:
-            Tuple (True, value, raw_bytes) if cache hit, None if cache miss or error.
-            Unlike the sync variant, the raw serialized envelope is included so the
-            async decorator can backfill L1 without re-serializing (re-encrypting).
+            Tuple (True, value, raw_bytes, size_bytes) if cache hit, None if cache miss
+            or error. The raw serialized envelope is included so the decorator can
+            backfill L1 without re-serializing (re-encrypting); same shape as the sync
+            variant, and size_bytes is its byte length.
 
         Note:
             Requires cache_handler to be set via set_cache_handler() before calling.
@@ -1523,7 +1539,7 @@ async def get_cached_value_async(self, cache_key: str, refresh_ttl: Optional[int
                 # Pass cache_key for AAD verification (required for encrypted data)
                 deserialized = self.serialization_handler.deserialize_data(cached_data, cache_key)
                 # Tuple distinguishes a hit from "no cache entry"; raw bytes ride along for L1
-                return (True, deserialized, cached_data)
+                return (True, deserialized, cached_data, len(cached_data))
             return None
         except KeyringConfigurationError:
             # LOCAL keyring config fault (bad tenant_id, bad keyring entry index) —

src/cachekit/decorators/wrapper.py

@@ -782,11 +782,23 @@ def _l1_backfill_from_l2(cache_key: str, cached_data: Any, is_stale: bool, fresh
         """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
+        fresh hit's local lifetime is bounded by _l1_backfill_ttl.
+
+        Best-effort: the hit is already decoded, so the one refusal L1Cache.put
+        documents — TypeError on a non-bytes envelope from an out-of-contract
+        backend — is logged and skipped; every caller sits inside an `except
+        Exception` that would otherwise demote the served hit into a recompute on
+        each call (LAB-348). Anything else is an L1 bug and propagates.
+        """
+        if not (_l1_cache and cache_key and cached_data and not is_stale):
+            return
+        cached_bytes = cached_data.encode("utf-8") if isinstance(cached_data, str) else cached_data
+        try:
             _l1_cache.put(cache_key, cached_bytes, redis_ttl=_l1_backfill_ttl(fresh_for))
-            _cached_keys.add(cache_key)
+        except TypeError as exc:
+            logger().warning(f"L1 backfill skipped for {redact_cache_key(cache_key)}: {bounded_error(exc)}")
+            return
+        _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.
@@ -1309,20 +1321,21 @@ def sync_wrapper(*args: Any, **kwargs: Any) -> Any:  # noqa: PLR0912
             # 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
+            _sync_l2_fresh_for: int | None = None
             if _l2_freshness_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
+                _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)
 
             duration = time.time() - start_time
 
             if cached_result is not None:
-                # Cached result is a tuple (True, actual_value)
+                # Cache hit: (True, value, envelope [None on the mmap fast path], size_bytes — set on every path)
+                _found, result, cached_data, size_bytes = cached_result
                 features.set_operation_context("get", duration_ms=duration * 1000)
                 features.record_success()
 
@@ -1337,7 +1350,6 @@ def sync_wrapper(*args: Any, **kwargs: Any) -> Any:  # noqa: PLR0912
                     )
 
                 # Record cache hit with structured logging
-                size_bytes = len(str(cached_result[1]).encode("utf-8")) if cached_result[1] is not None else 0
                 features.log_cache_operation(
                     operation="get",
                     key=cache_key,
@@ -1350,7 +1362,6 @@ def sync_wrapper(*args: Any, **kwargs: Any) -> Any:  # noqa: PLR0912
 
                 # Also record statistics if enabled
                 if features.collect_stats:
-                    size_bytes = len(str(cached_result[1]).encode("utf-8")) if cached_result[1] is not None else 0
                     features.record_cache_operation(
                         operation="get",
                         namespace=namespace or "default",
@@ -1361,6 +1372,10 @@ def sync_wrapper(*args: Any, **kwargs: Any) -> Any:  # noqa: PLR0912
                         hit=True,
                     )
 
+                # 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)
+
                 # Record L2 hit with latency for cache_info()
                 duration_ms = duration * 1000
                 _stats.record_l2_hit(duration_ms)
@@ -1374,7 +1389,7 @@ def sync_wrapper(*args: Any, **kwargs: Any) -> Any:  # noqa: PLR0912
                 # 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 cached_result[1]
+                return result
         except DecryptionAuthenticationError:
             # Fail-closed tamper failure propagated from get_cached_value — it only
             # raises when encryption.fail_closed=True (the metric and error log were
@@ -1610,13 +1625,16 @@ async def async_wrapper(*args: Any, **kwargs: Any) -> Any:
                         features.set_operation_context("l1_get", duration_ms=0.001)
                         features.record_success()
 
-                        # Record L1 cache hit metrics
+                        # Record L1 cache hit metrics (same labels as the sync L1 hit)
                         if features.collect_stats:
                             features.record_cache_operation(
                                 operation="get",
                                 namespace=namespace or "default",
+                                serializer="l1_memory",
                                 success=True,
                                 duration_ms=0.001,  # Sub-microsecond
+                                size_bytes=len(l1_bytes),
+                                hit=True,
                             )
 
                         # Record L1 hit for cache_info()
@@ -1701,8 +1719,8 @@ async def async_wrapper(*args: Any, **kwargs: Any) -> Any:
                     cached_result = await operation_handler.get_cached_value_async(cache_key)
 
                 if cached_result is not None:
-                    # Cache hit: (True, value, raw serialized envelope for L1 backfill)
-                    _found, result, cached_data = cached_result
+                    # Cache hit: (True, value, raw serialized envelope for L1 backfill, envelope size)
+                    _found, result, cached_data, size_bytes = cached_result
 
                     # Record cache hit (always compute for L2 latency stats)
                     get_duration_ms = (time.perf_counter() - start_time) * 1000
@@ -1713,8 +1731,11 @@ async def async_wrapper(*args: Any, **kwargs: Any) -> Any:
                         features.record_cache_operation(
                             operation="get",
                             namespace=namespace or "default",
+                            serializer="rust",
                             success=True,
                             duration_ms=get_duration_ms,
+                            size_bytes=size_bytes,
+                            hit=True,
                         )
 
                     # Update L1 cache with the L2 value (serialized bytes) for subsequent
@@ -1796,7 +1817,7 @@ async def async_wrapper(*args: Any, **kwargs: Any) -> Any:
                                 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
+                                    _found, result, cached_data, _size_bytes = cached_result
                                     _l1_backfill_from_l2(cache_key, cached_data, _dc_stale, _dc_fresh_for)
                                     return result
                             except DecryptionAuthenticationError:
@@ -1817,7 +1838,7 @@ async def async_wrapper(*args: Any, **kwargs: Any) -> Any:
                                 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
+                                    _found, result, cached_data, _size_bytes = cached_result
                                     _l1_backfill_from_l2(cache_key, cached_data, _dc_stale, _dc_fresh_for)
                                     return result
                             except DecryptionAuthenticationError:

tests/unit/test_decrypt_fail_policy.py

@@ -332,7 +332,7 @@ def test_fail_open_valid_entry_roundtrips(self):
         """Regression: the happy path is untouched by the policy plumbing."""
         handler, strategy, serialization = _make_operation_handler(fail_closed=True)
         strategy.store["key:a"] = serialization.serialize_data({"v": 7}, cache_key="key:a")
-        assert handler.get_cached_value("key:a") == (True, {"v": 7})
+        assert handler.get_cached_value("key:a") == (True, {"v": 7}, strategy.store["key:a"], len(strategy.store["key:a"]))
 
 
 class TestConfigDriftRead:

tests/unit/test_l2_decrypt_observability.py

@@ -139,7 +139,7 @@ async def test_async_eviction_failure_does_not_mask_miss(self, caplog: pytest.Lo
         assert any("Failed to evict poisoned" in r.message for r in caplog.records)
 
     async def test_async_hit_returns_value_and_raw_bytes(self) -> None:
-        """get_cached_value_async returns (True, value, raw_bytes) so the async
+        """get_cached_value_async returns (True, value, raw_bytes, size_bytes) so the async
         decorator can backfill L1 with the serialized envelope without re-serializing."""
         sentinel = object()
         mock_serialization = mock.MagicMock(spec=CacheSerializationHandler)
@@ -151,7 +151,7 @@ async def test_async_hit_returns_value_and_raw_bytes(self) -> None:
 
         result = await handler.get_cached_value_async("hit:key")
 
-        assert result == (True, sentinel, b"serialized-envelope")
+        assert result == (True, sentinel, b"serialized-envelope", len(b"serialized-envelope"))
 
 
 @pytest.mark.unit