perf(l1): keep verified MessagePack in L1 after an entry's first hit (LAB-7116)

docs/data-flow-architecture.md

@@ -31,7 +31,7 @@ cachekit uses a hybrid Python-Rust architecture to provide production caching wi
 > **Architecture Principles:**
 > - **L1 + L2 Caching**: In-memory bytes cache (L1) + pluggable backend storage (L2) for optimal performance
 > - **Backend Abstraction**: Protocol-based L2 backend layer (Redis, CachekitIO, File, Memcached, custom)
-> - **Bytes-Only Storage**: with a backend configured, L1 stores serialized bytes (encrypted or plaintext), not Python objects; L1-only mode (`backend=None`) stores the object itself
+> - **Bytes-Only Storage**: with a backend configured, L1 stores serialized bytes (encrypted or plaintext), not Python objects; L1-only mode (`backend=None`) stores the object itself. After its first hit, a plaintext entry may be held as its verified MessagePack instead (see [Storage Architecture](#storage-architecture)); encrypted caches keep the ciphertext
 > - **Security First**: Decompression bomb protection, checksum validation, size limits, encrypted-at-rest
 > - **Performance Optimized**: Thread affinity, connection pooling, pre-cached serializers
 > - **Flexible Configuration**: L1+L2, L1-only, L2-only, or observability-only modes
@@ -307,6 +307,10 @@ cachekit uses a hybrid Python-Rust architecture to provide production caching wi
 │                                                                             │
 │  Note: with a backend, L1 stores bytes (encrypted or plaintext), not Python │
 │  objects. L1-only mode (backend=None) stores the object itself.             │
+│                                                                             │
+│  L1 admits and evicts envelopes. A later L1 hit on a plaintext entry of the │
+│  default serializer (integrity on) may shrink it in place to its verified   │
+│  MessagePack: see Storage Architecture. Encrypted caches keep ciphertext.   │
 └──────────────────────────────────┬──────────────────────────────────────────┘
                                    │
                                    ▼
@@ -480,9 +484,10 @@ L1-only mode (`backend=None`) skips this layer: it stores the Python object itse
 - **After refactor**: Stores bytes (encrypted or plaintext) → `len(bytes)` for size estimation
 - **Performance improvement**: O(1) size calculation vs recursive traversal
 - **Security improvement**: Encrypted bytes at rest (decrypt-at-read-time only)
+- **Verified MessagePack**: L1 stores, admits and evicts envelopes. The first L1 hit on a plaintext entry of the default serializer with integrity checking on decodes the envelope as before, and when the MessagePack its `ByteStorage.retrieve()` verified is no longer than the envelope, L1 holds that MessagePack in its place from then on (same expiry and LRU place, fewer bytes). Later hits decode it without decompressing or checksumming it again. Compressible values are usually longer as MessagePack, so they keep the envelope; small and incompressible ones switch. The swap only shrinks an entry, so L1 never holds fewer entries than it would hold envelopes, and it never touches an entry stored or removed since the hit read it. Bytes from L2 never take this path: L1 marks the entry by its type (`VerifiedMsgpack`), which no backend returns
 
 **Hierarchical Read Path:**
-1. Check L1 cache (bytes) → if hit, deserialize and return
+1. Check L1 cache (bytes) → if hit, deserialize and return (a plaintext entry's first hit may swap it for its verified MessagePack, which later hits decode without the decompress and checksum)
 2. If L1 miss, check L2 backend (bytes) → if hit, deserialize, populate L1 with bytes, return
 3. If L2 miss, execute function → serialize to bytes → store in L2 and L1 → return
 

docs/features/prometheus-metrics.md

@@ -149,6 +149,11 @@ What each `cache_operation_duration_ms` series times, measured with a monotonic
 | `operation="get"`, `serializer="rust"` | The L2 read of a hit |
 | `operation="set"` | Serialization plus the L2 write: not the miss read, the function, a lock wait, the L1 put or key tracking |
 
+`cache_operation_size_bytes` records the length of the bytes the operation read or wrote. For an L1 `get`, that is the L1 entry:
+on a plaintext cache with the default serializer and integrity checking on, an entry's first L1 hit swaps its envelope for its
+MessagePack when that is no longer, so later L1 `get`s of it record a smaller size than its L2 `get`. Earlier releases recorded
+the envelope's length on every hit.
+
 > **Changed durations.** Earlier releases recorded every L1 hit as a constant 0.001 ms, and timed
 > `set` from before the L2 read, so each `set` sample also held the miss round trip and the
 > decorated function's own run time, plus the lock wait and a second read on the async locked path.

docs/performance.md

@@ -12,7 +12,7 @@
 > **Key numbers** (mean of five run medians; GitHub-hosted `ubuntu-latest` runner, AMD EPYC 7763, CPython 3.14.7, 2026-10-23; see [GitHub-Hosted Runner](#github-hosted-runner)):
 > - **Decorator + L1 hit, `@cache(backend=None)`, 100-user nested dict (23.5KB as plain MessagePack)**: 0.906 ± 0.007μs (about 10,900 instructions per call on CPython 3.14, see [Instruction Budgets](#instruction-budgets))
 > - **Same hit, 10K-row DataFrame**: 0.91 ± 0.01μs. An L1-only hit never serializes, so payload size barely matters
-> - **Decorator + L1 hit with an L2 backend configured, same dict**: 310 ± 20μs. With a backend, L1 holds bytes and every hit deserializes them
+> - **Decorator + L1 hit with an L2 backend configured, same dict**: 310 ± 20μs. With a backend, L1 holds bytes and every hit deserializes them. This dict compresses, so L1 holds its envelope; see [Why an L1-Only Hit Costs the Same for Any Payload](#why-an-l1-only-hit-costs-the-same-for-any-payload) for the values L1 holds as MessagePack
 > - **Redis L2 hit**, L1 disabled, same dict, Redis in a container on the same runner: 551 ± 18μs
 > - **Raw L1 byte-cache lookup** (`L1Cache.get`, no decorator): 354–362ns on a shared x86_64 Linux host (see [Measured Figures](#measured-figures))
 
@@ -73,6 +73,8 @@ With `backend=None`, cachekit keeps the returned object itself in memory and han
 
 With an L2 backend configured, L1 holds the serialized bytes instead, and a hit deserializes them, so that hit costs more as the payload grows. For the 100-user dict it takes 310 ± 20μs on the same runner, about 340 times an L1-only hit of the same dict. Nearly all of the difference is deserializing 23.5KB of MessagePack back into about 400 nested dicts. It is still about half of a Redis L2 hit there (551 ± 18μs), which deserializes the same bytes after the round trip.
 
+For a plaintext cache with the default serializer and integrity checking on, an entry's first L1 hit swaps its envelope for its verified MessagePack when the MessagePack is no longer, which is true of small values and incompressible ones. Later hits decode the MessagePack without the envelope's LZ4 decompress and xxHash3-64 checksum. The 100-user dict compresses to a 3,171-byte envelope from 23.5KB of MessagePack, so L1 keeps its envelope and the figures above are unchanged. Encrypted caches always keep the ciphertext envelope.
+
 ### Decorator + L1 Hit (Hot Path)
 
 An L1-only hit takes 0.906μs for the 100-user dict and 0.91μs for the DataFrame on the hosted EPYC 7763 ([GitHub-Hosted Runner](#github-hosted-runner)). What it runs:

src/cachekit/cache_handler.py

@@ -64,6 +64,7 @@
     KeyringConfigurationError,
     require_cross_sdk_serializer,
 )
+from cachekit.serializers.standard_serializer import StandardSerializer, VerifiedMsgpack
 from cachekit.serializers.wrapper import SerializationWrapper
 
 if TYPE_CHECKING:
@@ -947,6 +948,15 @@ def __init__(
         else:
             self._base_serializer = _get_cached_serializer_instance(serializer_name, enable_integrity_checking)
         self._custom_serializer = not isinstance(serializer_name, str)
+        # Whether an L1 hit may swap this handler's entry for its VerifiedMsgpack (deserialize_l1): plaintext
+        # CK-frame entries of the built-in StandardSerializer with integrity on (interop mode's InteropSerializer
+        # is not one). Encrypted handlers keep ciphertext in L1 (protocol PRE-10, PRE-11).
+        self.keeps_msgpack_in_l1 = (
+            not encryption
+            and not self._custom_serializer
+            and type(self._base_serializer) is StandardSerializer
+            and self._base_serializer.enable_integrity_checking
+        )
 
         # Cache EncryptionWrapper instances per tenant to prevent
         # 360K key copies/hour at 100 req/sec. Uses thread-safe LRU cache (maxsize=256)
@@ -1333,6 +1343,52 @@ def write_serialized_to(self, sink: BinaryIO, data: Any) -> None:
         budget = get_settings().max_value_size - len(prefix)
         serializer.serialize_to_sink(data, sink, max_bytes=budget)
 
+    def deserialize_l1(
+        self,
+        data: bytes | VerifiedMsgpack,
+        cache_key: str = "",
+        args: tuple[Any, ...] = (),
+        kwargs: dict[str, Any] | None = None,
+    ) -> tuple[Any, VerifiedMsgpack | None]:
+        """:meth:`deserialize_data` for an entry read from L1, plus what L1 should hold in its place, if anything.
+
+        That is the entry's verified MessagePack, when this handler keeps it (``keeps_msgpack_in_l1``), the
+        entry is its own plaintext envelope, and the MessagePack is no longer. It comes from the retrieve()
+        this decode ran anyway, so the caller's swap (``L1Cache.shrink_if_held``) costs no second retrieve(),
+        and later hits skip it. Everything else gets None: a VerifiedMsgpack this handler keeps decodes
+        without retrieve(), an envelope whose MessagePack is longer decodes as deserialize does, and any
+        other entry goes to deserialize_data, which refuses another handler's VerifiedMsgpack. Raises as
+        deserialize_data.
+
+        Examples:
+            >>> handler = CacheSerializationHandler()
+            >>> envelope = handler.serialize_data("hello", cache_key="greeting")
+            >>> value, entry = handler.deserialize_l1(envelope, cache_key="greeting")
+            >>> value, type(entry).__name__, len(entry) <= len(envelope)
+            ('hello', 'VerifiedMsgpack', True)
+            >>> handler.deserialize_l1(entry, cache_key="greeting")  # already MessagePack
+            ('hello', None)
+            >>> compressible = handler.serialize_data("ab" * 600, cache_key="k")
+            >>> handler.deserialize_l1(compressible, cache_key="k")[1] is None  # its MessagePack is longer
+            True
+        """
+        # Inline, not helpers: this is every L1 hit's decode, and an envelope it keeps must cost what
+        # deserialize_data costs.
+        if not self.keeps_msgpack_in_l1:
+            return self.deserialize_data(data, cache_key, args, kwargs), None  # pyright: ignore[reportArgumentType]
+        if type(data) is VerifiedMsgpack:  # only L1 holds this type: only such a handler swapped it in
+            return self._base_serializer._deserialize_verified(data), None  # type: ignore[union-attr]  # noqa: SLF001
+        try:
+            payload, metadata, serializer_name = SerializationWrapper.unwrap_metadata(data)
+        except (AttributeError, KeyError, TypeError, ValueError):
+            return self.deserialize_data(data, cache_key, args, kwargs), None  # a corrupt frame: it raises
+        # Only this handler's own plaintext envelope, whose header says what this handler writes
+        # (compressed: integrity on). Anything else, such as a config-drift entry whose ciphertext must
+        # stay in L1, or a header that misdescribes its payload, is decoded by deserialize_data, not kept.
+        if serializer_name != self._serializer_string_name or metadata.encrypted or not metadata.compressed:
+            return self.deserialize_data(data, cache_key, args, kwargs), None
+        return self._base_serializer._deserialize_keeping_msgpack(payload, len(data))  # type: ignore[union-attr]  # noqa: SLF001
+
     def deserialize_data(
         self,
         data: str | bytes | memoryview,
@@ -1426,6 +1482,13 @@ def deserialize_data(
                     data, shared=not self._custom_serializer
                 )
             except (AttributeError, KeyError, TypeError, ValueError) as e:
+                if self.encryption and type(data) is VerifiedMsgpack:  # pyright: ignore[reportUnnecessaryComparison]
+                    # A plaintext handler's swapped L1 entry under this key: refused as the plaintext
+                    # envelope it replaced would be, and counted the same way (suspicious_envelope).
+                    raise SuspiciousCacheEntryError(
+                        "Encryption is enabled but the L1 entry is another handler's plaintext MessagePack. "
+                        "Refusing it (fail closed)."
+                    ) from e
                 raise SerializationError(f"Corrupt cache envelope: {bounded_error(e)}") from e
 
             # Get base serializer

src/cachekit/decorators/wrapper.py

@@ -77,6 +77,7 @@
 
     from ..backends.base import BaseBackend, LockableBackend
     from ..serializers.base import SerializerProtocol
+    from ..serializers.standard_serializer import VerifiedMsgpack
 
 
 def _resolve_lazy_backend() -> BaseBackend:
@@ -1145,6 +1146,9 @@ def _on_l2_deserialize_error(error: Exception, key: str) -> None:
     # applies only when this namespace's cache is first created — namespaces share one
     # L1Cache, so give functions with distinct budgets distinct namespaces (issue #163).
     _l1_cache = get_l1_cache(namespace or "default", max_size_mb=l1_max_size_mb) if l1_enabled else None
+    # Whether an L1 hit may swap its entry for the verified MessagePack (deserialize_l1). Every other handler's
+    # hit decodes exactly as before, at no extra cost.
+    _l1_keeps_msgpack = serialization_handler.keeps_msgpack_in_l1
 
     # L1-only mode: use ObjectCache for raw Python object storage (no serialization).
     # This preserves types (tuples, sets, frozensets) that MessagePack would degrade.
@@ -1912,17 +1916,26 @@ def _flight_tenant(args: tuple[Any, ...], kwargs: dict[str, Any], cache_key: str
             return None
 
     def _decode_shared(
-        envelope: bytes | None, cache_key: str, args: tuple[Any, ...], kwargs: dict[str, Any]
+        entry: bytes | VerifiedMsgpack | None, cache_key: str, args: tuple[Any, ...], kwargs: dict[str, Any]
     ) -> tuple[bool, Any]:
-        """A joined backed-mode caller's own copy of the trip's envelope, decoded as an L1 hit would be:
-        every caller gets its own object, and the decrypt checks bind to this caller's tenant.
+        """A joined backed-mode caller's own copy of the trip's envelope, or of the L1 entry the re-check
+        found (which may be a VerifiedMsgpack), decoded as an L1 hit would be: every caller gets its own
+        object, and the decrypt checks bind to this caller's tenant. It never swaps the L1 entry.
 
-        (False, None) when there is no envelope or it does not decode for this caller, who then takes
+        (False, None) when there is no entry or it does not decode for this caller, who then takes
         its own trip, unshared: its probe slot went back on joining, so it is admitted afresh.
         """
-        if envelope is not None:
+        if entry is not None:
             try:
-                value = operation_handler.serialization_handler.deserialize_data(envelope, cache_key, args, kwargs)
+                if _l1_keeps_msgpack:
+                    value = operation_handler.serialization_handler.deserialize_l1(entry, cache_key, args, kwargs)[0]
+                else:
+                    value = operation_handler.serialization_handler.deserialize_data(
+                        entry,  # pyright: ignore[reportArgumentType]
+                        cache_key,
+                        args,
+                        kwargs,
+                    )
             except KeyringConfigurationError:
                 raise  # a LOCAL keyring config fault: see the L1 guard in _sync_call
             except Exception as e:
@@ -2230,7 +2243,19 @@ def _sync_call(  # noqa: PLR0912
                 if l1_found and l1_bytes:
                     # L1 cache hit - deserialize bytes
                     try:
-                        l1_value = operation_handler.serialization_handler.deserialize_data(l1_bytes, cache_key, args, kwargs)
+                        if _l1_keeps_msgpack:
+                            l1_value, l1_kept = operation_handler.serialization_handler.deserialize_l1(
+                                l1_bytes, cache_key, args, kwargs
+                            )
+                            if l1_kept is not None:  # its verified MessagePack: later hits skip retrieve()
+                                _l1_cache.shrink_if_held(cache_key, l1_bytes, l1_kept)
+                        else:  # decoded as before; an entry another handler swapped in fails as a corrupt frame
+                            l1_value = operation_handler.serialization_handler.deserialize_data(
+                                l1_bytes,  # pyright: ignore[reportArgumentType]
+                                cache_key,
+                                args,
+                                kwargs,
+                            )
                         l1_duration_ms = (time.perf_counter() - l1_start) * 1000  # L1 get plus deserialize
 
                         # Record L1 cache hit metrics
@@ -2384,13 +2409,13 @@ def _start_trip() -> tuple[Any, bytes | None]:
         decoded, value = _decode_shared(envelope, cache_key, args, kwargs)
         return value if decoded else _own_trip_sync(cache_key, args, kwargs, twin_key)[0]
 
-    def _peek_l1(cache_key: str) -> tuple[bool, tuple[None, bytes | None]]:
+    def _peek_l1(cache_key: str) -> tuple[bool, tuple[None, bytes | VerifiedMsgpack | None]]:
         """The sync backed flight's re-check (ThreadFlights.run): the L1 entry a trip that settled since
-        the caller's own L1 lookup left, as the envelope a joined caller decodes. Never an L2 read: it
-        runs under the flight map's lock."""
+        the caller's own L1 lookup left (its envelope, or the VerifiedMsgpack a hit swapped in), for a joined
+        caller to decode. Never an L2 read: it runs under the flight map's lock."""
         assert _l1_cache is not None  # noqa: S101 - passed only with L1 on
-        envelope = _l1_cache.peek(cache_key)
-        return bool(envelope), (None, envelope)
+        entry = _l1_cache.peek(cache_key)
+        return bool(entry), (None, entry)
 
     def _found_nothing() -> tuple[bool, tuple[None, None]]:
         """The re-check with L1 off: no entry to peek."""
@@ -2728,7 +2753,19 @@ async def _async_call(
                 if l1_found and l1_bytes:
                     # L1 cache hit - deserialize bytes
                     try:
-                        l1_value = operation_handler.serialization_handler.deserialize_data(l1_bytes, cache_key, args, kwargs)
+                        if _l1_keeps_msgpack:  # as in sync_wrapper
+                            l1_value, l1_kept = operation_handler.serialization_handler.deserialize_l1(
+                                l1_bytes, cache_key, args, kwargs
+                            )
+                            if l1_kept is not None:
+                                _l1_cache.shrink_if_held(cache_key, l1_bytes, l1_kept)
+                        else:
+                            l1_value = operation_handler.serialization_handler.deserialize_data(
+                                l1_bytes,  # pyright: ignore[reportArgumentType]
+                                cache_key,
+                                args,
+                                kwargs,
+                            )
                         l1_duration_ms = (time.perf_counter() - l1_start) * 1000  # L1 get plus deserialize
 
                         # Record L1 cache hit metrics (same labels as the sync L1 hit)