From c32f9771d801889d28498a508d5f837636bdc7d9 Mon Sep 17 00:00:00 2001
From: Mark S <ray@insighttimer.com>
Date: Thu, 17 Sep 2026 07:12:28 +1000
Subject: [PATCH 1/2] perf(decorators): backfill L1 on sync L2 hits; size stats
 by envelope length (LAB-348)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The sync wrapper served every L2 hit without re-warming L1, so after an L1
eviction or restart each sync call re-paid the L2 read + deserialize until
the next miss-store; the async wrapper has backfilled L1 on L2 hits all
along. The sync handler getters now return the raw envelope alongside the
value — the same (True, value, raw_bytes) shape as the async variants, None
on the mmap fast path whose view must never reach L1 — and the sync wrapper
feeds it through the shared _l1_backfill_from_l2 helper, so the
stale-exclusion and remaining-freshness bound apply to sync exactly as they
do to async.

The sync L2-hit stats sized the payload as len(str(value)): a repr, not the
payload, rendered on every hit. Both paths now record the envelope's byte
length, and the async hit records gain the serializer/hit labels the sync
path already emitted, so the get metric reads the same for both.

Refs cachekit-io/cachekit-py#164
---
 src/cachekit/cache_handler.py          |  26 +++--
 src/cachekit/decorators/wrapper.py     |  44 +++++---
 tests/unit/test_decrypt_fail_policy.py |   2 +-
 tests/unit/test_l2_hit_l1_backfill.py  | 143 +++++++++++++++++++++++++
 tests/unit/test_mmap_read_path.py      |   7 +-
 tests/unit/test_swr_decorator.py       |  38 ++++++-
 6 files changed, 233 insertions(+), 27 deletions(-)
 create mode 100644 tests/unit/test_l2_hit_l1_backfill.py

diff --git a/src/cachekit/cache_handler.py b/src/cachekit/cache_handler.py
index 0f47dfcd..630111c8 100644
--- a/src/cachekit/cache_handler.py
+++ b/src/cachekit/cache_handler.py
@@ -1362,7 +1362,11 @@ def get_cached_value(self, cache_key: str, refresh_ttl: Optional[int] = None) ->
             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) 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).
 
         Note:
             Requires cache_handler to be set via set_cache_handler() before calling.
@@ -1381,7 +1385,7 @@ 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))
+                        return (True, self.serialization_handler.deserialize_data(handle.view, cache_key), None)
                     finally:
                         handle.close()
 
@@ -1390,8 +1394,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)
             return None
         except KeyringConfigurationError:
             # LOCAL keyring config fault (bad tenant_id, bad keyring entry index) —
@@ -1410,12 +1414,14 @@ 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], 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), is_stale, fresh_for)`` on a hit —
+        the inner 3-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 +1442,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), 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
@@ -1504,8 +1510,8 @@ async def get_cached_value_async(self, cache_key: str, refresh_ttl: Optional[int
 
         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).
+            The raw serialized envelope is included so the decorator can backfill L1
+            without re-serializing (re-encrypting); same shape as the sync variant.
 
         Note:
             Requires cache_handler to be set via set_cache_handler() before calling.
diff --git a/src/cachekit/decorators/wrapper.py b/src/cachekit/decorators/wrapper.py
index 08b2209b..b49df623 100644
--- a/src/cachekit/decorators/wrapper.py
+++ b/src/cachekit/decorators/wrapper.py
@@ -782,11 +782,22 @@ 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 a refused put (L1Cache.put
+        rejects a non-bytes envelope from an out-of-contract backend) is logged
+        and swallowed — every caller sits inside an `except Exception` that would
+        otherwise demote the served hit into a recompute on each call (LAB-348).
+        """
+        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 Exception as exc:  # noqa: BLE001 — a best-effort backfill must never surface to callers
+            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 +1320,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, raw envelope for L1 backfill — None on the mmap fast path)
+                _found, result, cached_data = cached_result
                 features.set_operation_context("get", duration_ms=duration * 1000)
                 features.record_success()
 
@@ -1337,7 +1349,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,17 +1361,20 @@ 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",
                         serializer="rust",
                         success=True,
                         duration_ms=duration * 1000,
-                        size_bytes=size_bytes,
+                        size_bytes=len(cached_data) if cached_data else 0,
                         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 +1388,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 +1624,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()
@@ -1713,8 +1730,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=len(cached_data) if cached_data else 0,
+                            hit=True,
                         )
 
                     # Update L1 cache with the L2 value (serialized bytes) for subsequent
diff --git a/tests/unit/test_decrypt_fail_policy.py b/tests/unit/test_decrypt_fail_policy.py
index 3961708d..8ba61e29 100644
--- a/tests/unit/test_decrypt_fail_policy.py
+++ b/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"])
 
 
 class TestConfigDriftRead:
diff --git a/tests/unit/test_l2_hit_l1_backfill.py b/tests/unit/test_l2_hit_l1_backfill.py
new file mode 100644
index 00000000..e7e89aa6
--- /dev/null
+++ b/tests/unit/test_l2_hit_l1_backfill.py
@@ -0,0 +1,143 @@
+"""Sync/async L2-hit parity (cachekit-py#164, LAB-348).
+
+An L2 hit must (a) backfill L1 so the next read of the same key never
+re-pays the L2 round-trip, and (b) record ``size_bytes`` as the serialized
+envelope's length, not ``len(str(value))`` — a repr is neither the payload
+size nor cheap to build on every hit. The async wrapper always did (a); the
+sync wrapper did neither. Both are pinned here through the real decorator
+stack against a plain (non-SWR, non-locking) byte store.
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+import pytest
+
+from cachekit import cache
+from cachekit.reliability.async_metrics import AsyncMetricsCollector
+
+VALUE = {"answer": 42}
+
+
+class _CountingBackend:
+    """Plain byte store that counts L2 reads."""
+
+    def __init__(self) -> None:
+        self.store: dict[str, bytes] = {}
+        self.gets = 0
+
+    def get(self, key: str) -> bytes | None:
+        self.gets += 1
+        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, {}
+
+
+class _BytearrayBackend(_CountingBackend):
+    """Out-of-contract backend: a bytearray envelope deserializes fine but L1Cache.put refuses it."""
+
+    def get(self, key: str) -> bytearray | None:  # type: ignore[override]
+        raw = super().get(key)
+        return None if raw is None else bytearray(raw)
+
+
+@pytest.fixture
+def recorded(monkeypatch: pytest.MonkeyPatch) -> list[dict[str, Any]]:
+    """Capture every features.record_cache_operation(...) call (all keyword args)."""
+    calls: list[dict[str, Any]] = []
+    monkeypatch.setattr(AsyncMetricsCollector, "record_cache_operation", lambda self, **kw: calls.append(kw))
+    return calls
+
+
+def _force_l2_only(backend: _CountingBackend, l2_snapshot: dict[str, bytes]) -> None:
+    """After invalidate_cache() wiped L1 + L2, restore L2 alone and zero the read counter."""
+    backend.store.update(l2_snapshot)
+    backend.gets = 0
+
+
+def _assert_parity(backend: _CountingBackend, recorded: list[dict[str, Any]], info: Any) -> None:
+    envelope = next(iter(backend.store.values()))
+    get_hits = [c for c in recorded if c["operation"] == "get" and c.get("hit")]
+    assert len(get_hits) == 2  # one L2 hit, one L1 hit
+    assert [c["serializer"] for c in get_hits] == ["rust", "l1_memory"]
+    assert [c["size_bytes"] for c in get_hits] == [len(envelope)] * 2
+    assert len(envelope) != len(str(VALUE).encode("utf-8"))  # the old str(value) accounting would differ
+    assert backend.gets == 1  # second read never reached L2
+    assert (info.l1_hits, info.l2_hits) == (1, 1)
+
+
+@pytest.mark.unit
+class TestL2HitParity:
+    def test_sync_l2_hit_backfills_l1_and_records_envelope_size(self, recorded: list[dict[str, Any]]) -> None:
+        backend = _CountingBackend()
+        calls = {"n": 0}
+
+        @cache(backend=backend, ttl=60, namespace="l2-parity-sync")
+        def compute() -> dict[str, int]:
+            calls["n"] += 1
+            return dict(VALUE)
+
+        assert compute() == VALUE  # miss -> L2 + L1 store
+        l2_snapshot = dict(backend.store)
+        compute.invalidate_cache()  # type: ignore[attr-defined]
+        _force_l2_only(backend, l2_snapshot)
+        recorded.clear()
+
+        assert compute() == VALUE  # L1 miss -> L2 hit -> backfill L1
+        assert compute() == VALUE  # served from L1
+        assert calls["n"] == 1
+        _assert_parity(backend, recorded, compute.cache_info())  # type: ignore[attr-defined]
+
+    async def test_async_l2_hit_backfills_l1_and_records_envelope_size(self, recorded: list[dict[str, Any]]) -> None:
+        backend = _CountingBackend()
+        calls = {"n": 0}
+
+        @cache(backend=backend, ttl=60, namespace="l2-parity-async")
+        async def compute() -> dict[str, int]:
+            calls["n"] += 1
+            return dict(VALUE)
+
+        assert await compute() == VALUE
+        l2_snapshot = dict(backend.store)
+        await compute.invalidate_cache()  # type: ignore[attr-defined]
+        _force_l2_only(backend, l2_snapshot)
+        recorded.clear()
+
+        assert await compute() == VALUE
+        assert await compute() == VALUE
+        assert calls["n"] == 1
+        _assert_parity(backend, recorded, compute.cache_info())  # type: ignore[attr-defined]
+
+    def test_sync_l2_hit_survives_refused_l1_backfill(self, caplog: pytest.LogCaptureFixture) -> None:
+        """A refused backfill is logged and skipped; the served hit is never demoted to a recompute."""
+        backend = _BytearrayBackend()
+        calls = {"n": 0}
+
+        @cache(backend=backend, ttl=60, namespace="l2-parity-refused")
+        def compute() -> dict[str, int]:
+            calls["n"] += 1
+            return dict(VALUE)
+
+        assert compute() == VALUE
+        l2_snapshot = dict(backend.store)
+        compute.invalidate_cache()  # type: ignore[attr-defined]
+        _force_l2_only(backend, l2_snapshot)
+
+        with caplog.at_level(logging.WARNING):
+            assert compute() == VALUE  # L2 hit served, backfill refused
+            assert compute() == VALUE  # L2 again: nothing landed in L1
+        assert calls["n"] == 1
+        assert backend.gets == 2
+        assert any("L1 backfill skipped" in r.message for r in caplog.records)
diff --git a/tests/unit/test_mmap_read_path.py b/tests/unit/test_mmap_read_path.py
index 8f08731d..52b954bf 100644
--- a/tests/unit/test_mmap_read_path.py
+++ b/tests/unit/test_mmap_read_path.py
@@ -104,7 +104,7 @@ def test_eligible_reads_via_mmap_and_confines_the_handle(self) -> None:
 
         result = self._handler(sh, ch).get_cached_value("k")
 
-        assert result == (True, sentinel)
+        assert result == (True, sentinel, None)  # no envelope: the mmap view never reaches L1 (blocker C)
         ch.get_buffer.assert_called_once_with("k")
         ch.get.assert_not_called()  # normal read path NOT used on the mmap hit
         sh.deserialize_data.assert_called_once_with(handle.view, "k")
@@ -133,7 +133,7 @@ def test_get_buffer_none_falls_through_to_normal_read(self) -> None:
 
         ch.get_buffer.assert_called_once()
         ch.get.assert_called_once()  # fell through
-        assert result == (True, "val")
+        assert result == (True, "val", b"frame")  # os.read fallback carries the envelope for L1
 
 
 @pytest.mark.unit
@@ -165,8 +165,9 @@ def test_arrow_dataframe_roundtrips_through_real_mmap(self, tmp_path) -> None:
             hit = oh.get_cached_value("k")
 
         assert hit is not None
-        found, value = hit
+        found, value, envelope = hit
         assert found is True
+        assert envelope is None  # mmap hit: nothing to backfill into L1
         pd.testing.assert_frame_equal(value, df)
         gb.assert_called_once()  # the real mmap path was taken
         g.assert_not_called()  # not the os.read fallback
diff --git a/tests/unit/test_swr_decorator.py b/tests/unit/test_swr_decorator.py
index ffe6dc36..31e11931 100644
--- a/tests/unit/test_swr_decorator.py
+++ b/tests/unit/test_swr_decorator.py
@@ -528,7 +528,7 @@ def test_legacy_two_tuple_handler_degrades_to_no_bound(self) -> None:
         cache_handler = mock.MagicMock()
         cache_handler.get_with_freshness.return_value = (b"bytes", False)  # 0.5.x 2-tuple
         op.set_cache_handler(cache_handler)
-        assert op.get_cached_value_with_freshness("k") == ((True, {"v": 1}), False, None)
+        assert op.get_cached_value_with_freshness("k") == ((True, {"v": 1}, b"bytes"), False, None)
 
     def test_backend_error_reads_as_miss(self) -> None:
         op, cache_handler = self._make_op()
@@ -573,6 +573,42 @@ def compute() -> int:
         assert compute() == 2  # revalidation refreshed L1 with fresh bytes
         assert calls["n"] == 2
 
+    def test_sync_freshness_hit_backfills_l1_unless_stale_or_expired(self) -> None:
+        """LAB-348 parity: the sync freshness read backfills L1 exactly as the
+        async read does — a fresh hit is recorded (next read is L1, no second
+        freshness read), a stale-labelled hit never is, and fresh_for=0 makes
+        the backfill a no-op (the remaining-freshness bound reaches the sync
+        path). No stale_ttl: a stale hit is served with no revalidation, so
+        nothing races the second read."""
+        backend = FakeSWRBackend()
+
+        @cache(backend=backend, ttl=60, namespace="swr-l1-sync-backfill")
+        def compute() -> int:
+            return 1
+
+        assert compute() == 1
+        l2_snapshot = dict(backend.store)
+
+        def force_l2() -> None:
+            compute.invalidate_cache()  # type: ignore[attr-defined]
+            backend.store.update(l2_snapshot)
+            backend.freshness_reads = 0
+
+        force_l2()
+        assert compute() == 1 and compute() == 1
+        assert backend.freshness_reads == 1  # fresh hit backfilled -> second read served by L1
+
+        force_l2()
+        backend.stale = True
+        assert compute() == 1 and compute() == 1
+        assert backend.freshness_reads == 2  # stale hit never recorded in L1
+
+        force_l2()
+        backend.stale = False
+        backend.fresh_for = 0
+        assert compute() == 1 and compute() == 1
+        assert backend.freshness_reads == 2  # nothing fresh remains -> L1Cache.put skips the entry
+
 
 class TestSWRSchedulingHardening:
     """CodeRabbit round-2 regressions: negative default window, arg snapshots,

From 3f25fd7f721006b547971c0f727586876b2e18be Mon Sep 17 00:00:00 2001
From: Mark S <ray@insighttimer.com>
Date: Thu, 17 Sep 2026 07:39:29 +1000
Subject: [PATCH 2/2] fix(decorators): carry payload size on mmap hits; narrow
 L1 backfill catch (LAB-348)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Review follow-ups on #294:

- CodeRabbit: an mmap fast-path hit returned no envelope, so size_bytes fell
  to 0 and the hit dropped out of cache_operation_size_bytes. The hit tuple
  now carries size_bytes as a fourth slot on every path — the handler owns
  the size (len(envelope), or view.nbytes read before the mmap handle closes)
  and the wrapper only records it. The mmap end-to-end test asserts it.
- Kody: the best-effort backfill catch was a broad `except Exception`.
  L1Cache.put documents exactly one refusal — TypeError on a non-bytes
  envelope — so the catch names it; anything else is an L1 bug and
  propagates to the existing cache_get error path.
---
 src/cachekit/cache_handler.py               | 52 ++++++++++++---------
 src/cachekit/decorators/wrapper.py          | 27 +++++------
 tests/unit/test_decrypt_fail_policy.py      |  2 +-
 tests/unit/test_l2_decrypt_observability.py |  4 +-
 tests/unit/test_mmap_read_path.py           | 12 +++--
 tests/unit/test_swr_decorator.py            |  2 +-
 6 files changed, 57 insertions(+), 42 deletions(-)

diff --git a/src/cachekit/cache_handler.py b/src/cachekit/cache_handler.py
index 630111c8..a85af0e1 100644
--- a/src/cachekit/cache_handler.py
+++ b/src/cachekit/cache_handler.py
@@ -1354,7 +1354,9 @@ 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:
@@ -1362,11 +1364,13 @@ def get_cached_value(self, cache_key: str, refresh_ttl: Optional[int] = None) ->
             refresh_ttl: Optional TTL to refresh on hit
 
         Returns:
-            Tuple (True, value, raw_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).
+            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.
@@ -1385,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), None)
+                        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()
 
@@ -1395,7 +1400,7 @@ def get_cached_value(self, cache_key: str, refresh_ttl: Optional[int] = None) ->
                 # 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) —
@@ -1414,12 +1419,14 @@ 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, bytes], 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, raw_bytes), is_stale, fresh_for)`` on a hit —
-        the inner 3-tuple matches the async variant so the sync decorator
+        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
@@ -1442,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, 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
@@ -1462,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
@@ -1483,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
@@ -1501,7 +1508,9 @@ 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:
@@ -1509,9 +1518,10 @@ async def get_cached_value_async(self, cache_key: str, refresh_ttl: Optional[int
             refresh_ttl: Optional TTL to refresh on hit
 
         Returns:
-            Tuple (True, value, raw_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.
+            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.
@@ -1529,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) —
diff --git a/src/cachekit/decorators/wrapper.py b/src/cachekit/decorators/wrapper.py
index b49df623..1a6bc62c 100644
--- a/src/cachekit/decorators/wrapper.py
+++ b/src/cachekit/decorators/wrapper.py
@@ -784,17 +784,18 @@ def _l1_backfill_from_l2(cache_key: str, cached_data: Any, is_stale: bool, fresh
         recorded (spec: local caches MUST NOT record stale as fresh), and a
         fresh hit's local lifetime is bounded by _l1_backfill_ttl.
 
-        Best-effort: the hit is already decoded, so a refused put (L1Cache.put
-        rejects a non-bytes envelope from an out-of-contract backend) is logged
-        and swallowed — every caller sits inside an `except Exception` that would
-        otherwise demote the served hit into a recompute on each call (LAB-348).
+        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))
-        except Exception as exc:  # noqa: BLE001 — a best-effort backfill must never surface to callers
+        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)
@@ -1333,8 +1334,8 @@ def sync_wrapper(*args: Any, **kwargs: Any) -> Any:  # noqa: PLR0912
             duration = time.time() - start_time
 
             if cached_result is not None:
-                # Cache hit: (True, value, raw envelope for L1 backfill — None on the mmap fast path)
-                _found, result, cached_data = cached_result
+                # 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()
 
@@ -1367,7 +1368,7 @@ def sync_wrapper(*args: Any, **kwargs: Any) -> Any:  # noqa: PLR0912
                         serializer="rust",
                         success=True,
                         duration_ms=duration * 1000,
-                        size_bytes=len(cached_data) if cached_data else 0,
+                        size_bytes=size_bytes,
                         hit=True,
                     )
 
@@ -1718,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
@@ -1733,7 +1734,7 @@ async def async_wrapper(*args: Any, **kwargs: Any) -> Any:
                             serializer="rust",
                             success=True,
                             duration_ms=get_duration_ms,
-                            size_bytes=len(cached_data) if cached_data else 0,
+                            size_bytes=size_bytes,
                             hit=True,
                         )
 
@@ -1816,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:
@@ -1837,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:
diff --git a/tests/unit/test_decrypt_fail_policy.py b/tests/unit/test_decrypt_fail_policy.py
index 8ba61e29..5c2fe202 100644
--- a/tests/unit/test_decrypt_fail_policy.py
+++ b/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}, strategy.store["key:a"])
+        assert handler.get_cached_value("key:a") == (True, {"v": 7}, strategy.store["key:a"], len(strategy.store["key:a"]))
 
 
 class TestConfigDriftRead:
diff --git a/tests/unit/test_l2_decrypt_observability.py b/tests/unit/test_l2_decrypt_observability.py
index 4334e58e..7a98d2bd 100644
--- a/tests/unit/test_l2_decrypt_observability.py
+++ b/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
diff --git a/tests/unit/test_mmap_read_path.py b/tests/unit/test_mmap_read_path.py
index 52b954bf..bfccb68c 100644
--- a/tests/unit/test_mmap_read_path.py
+++ b/tests/unit/test_mmap_read_path.py
@@ -99,12 +99,14 @@ def test_eligible_reads_via_mmap_and_confines_the_handle(self) -> None:
         sh.supports_mmap_read.return_value = True
         sh.deserialize_data.return_value = sentinel
         handle = MagicMock()
+        handle.view.nbytes = 4096
         ch = MagicMock()
         ch.get_buffer.return_value = handle
 
         result = self._handler(sh, ch).get_cached_value("k")
 
-        assert result == (True, sentinel, None)  # no envelope: the mmap view never reaches L1 (blocker C)
+        # No envelope (the mmap view never reaches L1, blocker C), but the payload size still rides along
+        assert result == (True, sentinel, None, 4096)
         ch.get_buffer.assert_called_once_with("k")
         ch.get.assert_not_called()  # normal read path NOT used on the mmap hit
         sh.deserialize_data.assert_called_once_with(handle.view, "k")
@@ -133,7 +135,7 @@ def test_get_buffer_none_falls_through_to_normal_read(self) -> None:
 
         ch.get_buffer.assert_called_once()
         ch.get.assert_called_once()  # fell through
-        assert result == (True, "val", b"frame")  # os.read fallback carries the envelope for L1
+        assert result == (True, "val", b"frame", 5)  # os.read fallback carries the envelope for L1
 
 
 @pytest.mark.unit
@@ -156,7 +158,8 @@ def test_arrow_dataframe_roundtrips_through_real_mmap(self, tmp_path) -> None:
         oh = CacheOperationHandler(sh, CacheKeyGenerator(), cache_handler=ch)
 
         df = pd.DataFrame({"a": range(2000), "b": [float(i) / 3 for i in range(2000)]})
-        ch.set("k", sh.serialize_data(df, cache_key="k"), 300)
+        payload = sh.serialize_data(df, cache_key="k")
+        ch.set("k", payload, 300)
 
         with (
             patch.object(backend, "get", wraps=backend.get) as g,
@@ -165,9 +168,10 @@ def test_arrow_dataframe_roundtrips_through_real_mmap(self, tmp_path) -> None:
             hit = oh.get_cached_value("k")
 
         assert hit is not None
-        found, value, envelope = hit
+        found, value, envelope, size_bytes = hit
         assert found is True
         assert envelope is None  # mmap hit: nothing to backfill into L1
+        assert size_bytes == len(payload)  # ...yet the payload size is accounted without an os.read copy
         pd.testing.assert_frame_equal(value, df)
         gb.assert_called_once()  # the real mmap path was taken
         g.assert_not_called()  # not the os.read fallback
diff --git a/tests/unit/test_swr_decorator.py b/tests/unit/test_swr_decorator.py
index 31e11931..42e42eee 100644
--- a/tests/unit/test_swr_decorator.py
+++ b/tests/unit/test_swr_decorator.py
@@ -528,7 +528,7 @@ def test_legacy_two_tuple_handler_degrades_to_no_bound(self) -> None:
         cache_handler = mock.MagicMock()
         cache_handler.get_with_freshness.return_value = (b"bytes", False)  # 0.5.x 2-tuple
         op.set_cache_handler(cache_handler)
-        assert op.get_cached_value_with_freshness("k") == ((True, {"v": 1}, b"bytes"), False, None)
+        assert op.get_cached_value_with_freshness("k") == ((True, {"v": 1}, b"bytes", 5), False, None)
 
     def test_backend_error_reads_as_miss(self) -> None:
         op, cache_handler = self._make_op()