diff --git a/.secrets.baseline b/.secrets.baseline index 7deb5010..cee764bc 100644 --- a/.secrets.baseline +++ b/.secrets.baseline @@ -215,7 +215,7 @@ "filename": "src/cachekit/cache_handler.py", "hashed_secret": "5baa61e4c9b93f3f0682250b6cf8331b7ee68fd8", "is_verified": false, - "line_number": 699 + "line_number": 700 } ], "src/cachekit/serializers/interop_serializer.py": [ diff --git a/docs/data-flow-architecture.md b/docs/data-flow-architecture.md index 052137b2..1bacb8cd 100644 --- a/docs/data-flow-architecture.md +++ b/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 diff --git a/docs/features/prometheus-metrics.md b/docs/features/prometheus-metrics.md index bbbe1143..ff995119 100644 --- a/docs/features/prometheus-metrics.md +++ b/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. diff --git a/docs/performance.md b/docs/performance.md index a82bb091..1a4146b8 100644 --- a/docs/performance.md +++ b/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: diff --git a/src/cachekit/cache_handler.py b/src/cachekit/cache_handler.py index d9aa0965..0018a123 100644 --- a/src/cachekit/cache_handler.py +++ b/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 diff --git a/src/cachekit/decorators/wrapper.py b/src/cachekit/decorators/wrapper.py index 52894c2f..927f3298 100644 --- a/src/cachekit/decorators/wrapper.py +++ b/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) diff --git a/src/cachekit/l1_cache.py b/src/cachekit/l1_cache.py index 5ed1f139..311823a4 100644 --- a/src/cachekit/l1_cache.py +++ b/src/cachekit/l1_cache.py @@ -14,11 +14,14 @@ from collections.abc import Callable, Iterable from dataclasses import dataclass from itertools import islice -from typing import Any, Concatenate, Optional, ParamSpec +from typing import TYPE_CHECKING, Any, Concatenate, Optional, ParamSpec from cachekit import _fork from cachekit.hash_utils import redact_error_for_log, redact_key_for_log +if TYPE_CHECKING: + from cachekit.serializers.standard_serializer import VerifiedMsgpack + # Default L1 entry lifetime when the caller supplies no TTL. Shared with the # decorator's freshness backfill bound: the server's Fresh-For may only ever # SHORTEN the L1 lifetime relative to this default, never extend it. @@ -49,7 +52,7 @@ class CacheEntry: read time in CacheHandler, not storage time. """ - value: bytes + value: "bytes | VerifiedMsgpack" # VerifiedMsgpack once a hit swapped it in (shrink_if_held) expires_at: float size_bytes: int @@ -150,7 +153,7 @@ def __init__( ttl_buffer_seconds, ) - def _estimate_size(self, value: bytes) -> int: + def _estimate_size(self, value: "bytes | VerifiedMsgpack") -> int: """Estimate memory size of bytes value. Simplified from recursive sys.getsizeof() - faster and more accurate @@ -164,7 +167,7 @@ def _estimate_size(self, value: bytes) -> int: """ return len(value) - def get(self, key: str) -> tuple[bool, Optional[bytes]]: + def get(self, key: str) -> "tuple[bool, Optional[bytes | VerifiedMsgpack]]": """Get value from L1 cache if present and not expired. Args: @@ -172,7 +175,8 @@ def get(self, key: str) -> tuple[bool, Optional[bytes]]: Returns: Tuple of (found, value) where found is True if hit. - Value is bytes (encrypted or plaintext msgpack), not deserialized object. + Value is bytes (encrypted or plaintext msgpack), or the VerifiedMsgpack a hit swapped in + (shrink_if_held), not deserialized object. Caller (CacheHandler) is responsible for decryption/deserialization. """ s = self._state @@ -197,7 +201,7 @@ def get(self, key: str) -> tuple[bool, Optional[bytes]]: return True, entry.value - def peek(self, key: str) -> Optional[bytes]: + def peek(self, key: str) -> "Optional[bytes | VerifiedMsgpack]": """The bytes ``get`` would return, without counting a hit or miss or moving the entry in LRU order. For a second lookup within one call, such as the single-flight re-check after a miss, which @@ -214,6 +218,24 @@ def peek(self, key: str) -> Optional[bytes]: entry = s.cache.get(key) return None if entry is None or entry.is_expired() else entry.value + def shrink_if_held(self, key: str, held: "bytes | VerifiedMsgpack", smaller: "VerifiedMsgpack") -> bool: + """Hold ``smaller`` in place of ``held``, the very object a read of ``key`` returned, if the entry still + holds it and ``smaller`` is no longer. The entry keeps its expiry and LRU place. Returns whether it did. + + For a hit that re-encodes the entry it read. The identity check leaves alone an entry that was + removed, or stored again, since that read, so an invalidation is never undone. Never growing an + entry means L1 never evicts for it. + """ + size = self._estimate_size(smaller) + s = self._state + with s.lock: + entry = s.cache.get(key) + if entry is None or entry.value is not held or size > entry.size_bytes: + return False + s.memory_bytes -= entry.size_bytes - size + entry.value, entry.size_bytes = smaller, size + return True + def put( self, key: str, diff --git a/src/cachekit/serializers/standard_serializer.py b/src/cachekit/serializers/standard_serializer.py index be354370..f42a992a 100644 --- a/src/cachekit/serializers/standard_serializer.py +++ b/src/cachekit/serializers/standard_serializer.py @@ -21,7 +21,7 @@ from __future__ import annotations from datetime import date, datetime, time -from typing import Any, ClassVar +from typing import Any, ClassVar, final import msgpack @@ -47,6 +47,26 @@ _ENVELOPE_PROBE_MAX_COMPRESSED = 20 + _ENVELOPE_PROBE_MAX_SIZE * 110 // 100 _CROSS_CONFIG_ERROR = "Cache entry was written with integrity checking on but this reader has integrity checking disabled" + +@final # the read paths test `type(data) is VerifiedMsgpack` +class VerifiedMsgpack: + """MessagePack a successful ``ByteStorage.retrieve()`` returned, which an L1 entry holds in place of its + envelope (``CacheSerializationHandler.deserialize_l1``). It decodes without another retrieve(). + + The type is the mark, not the bytes: no backend returns this type, so bytes from L2 never take + the unverified decode, whatever they hold. It holds retrieve()'s own bytes object rather than + copying it into a bytes subclass, and its length is theirs, which L1 accounts. + """ + + __slots__ = ("data",) + + def __init__(self, data: bytes) -> None: + self.data = data + + def __len__(self) -> int: + return len(self.data) + + # Error message constants for unsupported types (Task 2) NUMPY_ERROR_MESSAGE = ( "StandardSerializer does not support NumPy arrays (Python-specific type). " @@ -392,6 +412,36 @@ def deserialize(self, data: bytes | memoryview, metadata: SerializationMetadata except PAYLOAD_DECODE_ERRORS as e: raise SerializationError(f"Failed to deserialize MessagePack data: {e}") from e + def _deserialize_keeping_msgpack(self, data: bytes | memoryview, limit: int) -> tuple[Any, VerifiedMsgpack | None]: + """What :meth:`deserialize` returns for a ByteStorage envelope when integrity checking is on, plus the + MessagePack ``retrieve()`` verified when it is at most ``limit`` bytes, for the caller to keep and decode + later without retrieve() (:meth:`_deserialize_verified`). It always unwraps the envelope, so call it + only for an integrity-on serializer. Raises SerializationError as deserialize does then. + + Examples: + >>> serializer = StandardSerializer() + >>> envelope, _ = serializer.serialize({"test": 123}) + >>> value, kept = serializer._deserialize_keeping_msgpack(envelope, len(envelope)) + >>> value, serializer._deserialize_verified(kept) + ({'test': 123}, {'test': 123}) + >>> serializer._deserialize_keeping_msgpack(envelope, 4) # the MessagePack is 7 bytes + ({'test': 123}, None) + """ + try: + msgpack_data, _ = self._byte_storage.retrieve(data) + value = unpackb_bounded(msgpack_data, **self._msgpack_unpack_opts) + except PAYLOAD_DECODE_ERRORS as e: + raise SerializationError(f"Failed to deserialize MessagePack data: {e}") from e + return value, VerifiedMsgpack(msgpack_data) if len(msgpack_data) <= limit else None + + def _deserialize_verified(self, kept: VerifiedMsgpack) -> Any: + """Decode MessagePack that :meth:`_deserialize_keeping_msgpack` kept, without another ``retrieve()``. + Raises SerializationError as deserialize does.""" + try: + return unpackb_bounded(kept.data, **self._msgpack_unpack_opts) + except PAYLOAD_DECODE_ERRORS as e: + raise SerializationError(f"Failed to deserialize MessagePack data: {e}") from e + def _is_verified_envelope(self, data: bytes | memoryview, value: Any, markers: frozenset[int]) -> bool: """True only when ``data``, already decoded to ``value``, is a ByteStorage envelope whose payload marker is in ``markers`` and that passes ``retrieve()``.""" diff --git a/tests/unit/test_error_path_key_redaction.py b/tests/unit/test_error_path_key_redaction.py index 497eee79..ca431b5a 100644 --- a/tests/unit/test_error_path_key_redaction.py +++ b/tests/unit/test_error_path_key_redaction.py @@ -554,9 +554,9 @@ class TestDecoratorWrapperRedaction: @staticmethod def _poison_deserialize(monkeypatch: pytest.MonkeyPatch, error: Exception) -> None: - # Class-level patch: the wrapper reaches deserialize_data through the handler instance - # it built at decoration time, so an instance patch has nothing to attach to. - monkeypatch.setattr(CacheSerializationHandler, "deserialize_data", MagicMock(side_effect=error)) + # Class-level patch: the wrapper reaches deserialize_l1 (the L1 hit sites' decode) through the + # handler instance it built at decoration time, so an instance patch has nothing to attach to. + monkeypatch.setattr(CacheSerializationHandler, "deserialize_l1", MagicMock(side_effect=error)) @staticmethod def _assert_l1_sink(caplog: pytest.LogCaptureFixture, cache_key: str, error: Exception) -> None: diff --git a/tests/unit/test_l1_verified_msgpack.py b/tests/unit/test_l1_verified_msgpack.py new file mode 100644 index 00000000..f835b762 --- /dev/null +++ b/tests/unit/test_l1_verified_msgpack.py @@ -0,0 +1,547 @@ +"""A non-encrypted backed cache's L1 hit swaps its entry for the verified MessagePack, where that is no longer. + +The first L1 hit on an entry decodes the envelope as before, and its retrieve() yields the MessagePack; +L1 then holds that in place of the envelope (``CacheSerializationHandler.deserialize_l1``, +``L1Cache.shrink_if_held``), and later hits decode it without retrieve(). Each rule that keeps this from costing +L1 hits or security is pinned here: + +- L1 admits and stores envelopes exactly as before; a hit only ever shrinks an entry in place. So L1 never + holds fewer entries than it would hold envelopes (the replay property below). +- MessagePack enters L1 only from a retrieve() that verified it; a corrupt envelope is never swapped. +- Encrypted handlers keep their ciphertext envelope in L1, and every hit on it decrypts (protocol PRE-10, + PRE-11: the secure preset holds only ciphertext in L1). +""" + +from __future__ import annotations + +import os +import uuid +from collections.abc import Callable, Iterator +from typing import Any + +import msgpack +import pytest +from hypothesis import given, settings +from hypothesis import strategies as st + +from cachekit import cache +from cachekit.cache_handler import CacheSerializationHandler +from cachekit.config import reset_settings +from cachekit.config.nested import L1CacheConfig +from cachekit.l1_cache import L1Cache, get_l1_cache +from cachekit.serializers.base import SerializationError, SuspiciousCacheEntryError +from cachekit.serializers.encryption_wrapper import EncryptionWrapper +from cachekit.serializers.standard_serializer import StandardSerializer, VerifiedMsgpack +from cachekit.serializers.wrapper import SerializationWrapper +from tests.fixtures.zero_config import clear_cachekit_env + +pytestmark = pytest.mark.unit + +MASTER_KEY = "a" * 64 + + +@pytest.fixture(autouse=True) +def _no_cachekit_env(monkeypatch: pytest.MonkeyPatch) -> Iterator[None]: + """Handlers here read key settings when built, so no CACHEKIT_* variable in the shell may reach them.""" + clear_cachekit_env(monkeypatch) + reset_settings() + yield + reset_settings() + + +class _Backend: + """In-memory byte store that counts its 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] = bytes(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, {} + + +def _namespace() -> str: + return f"l1vm-{uuid.uuid4().hex}" # backed-mode L1 is shared by namespace across decorations + + +def _stored(backend: _Backend, namespace: str) -> tuple[bytes, bytes | None]: + """The one L2 envelope and the L1 entry under its key.""" + ((key, envelope),) = backend.store.items() + return envelope, get_l1_cache(namespace).peek(key) + + +def _corrupt(envelope: bytes) -> bytes: + """``envelope`` with one byte of its ByteStorage payload flipped: the frame still parses, retrieve() fails.""" + corrupt = bytearray(envelope) + corrupt[-20] ^= 0xFF + return bytes(corrupt) + + +# ── what L1 holds ───────────────────────────────────────────────────────────────────────────── + + +@pytest.mark.parametrize( + "value", ["hello", 42, {"id": 7, "tags": ["a"]}, os.urandom(5000)], ids=["str", "int", "dict", "random"] +) +def test_the_first_hit_swaps_an_envelope_for_its_msgpack_when_no_longer(value: Any) -> None: + backend, namespace, runs = _Backend(), _namespace(), [] + + @cache(backend=backend, ttl=60, namespace=namespace) + def compute(x: int) -> Any: + runs.append(x) + return value + + assert compute(1) == value # miss: L1 gets the envelope L2 got, as before + envelope, entry = _stored(backend, namespace) + assert type(entry) is bytes and entry == envelope + + before = compute.cache_info().l1_hits # type: ignore[attr-defined] # stats are per function name + assert compute(1) == value # first hit: decodes the envelope, keeps its MessagePack + _, entry = _stored(backend, namespace) + assert type(entry) is VerifiedMsgpack + assert entry.data == msgpack.packb(value, use_bin_type=True) + assert len(entry) <= len(envelope) + + assert compute(1) == value # decodes the MessagePack + assert (runs, backend.gets) == ([1], 1) + assert compute.cache_info().l1_hits == before + 2 # type: ignore[attr-defined] + + +def test_a_value_longer_as_msgpack_keeps_its_envelope() -> None: + backend, namespace = _Backend(), _namespace() + + @cache(backend=backend, ttl=60, namespace=namespace) + def compute(x: int) -> str: + return "ab" * 600 + + for _ in range(3): + assert compute(1) == "ab" * 600 + envelope, entry = _stored(backend, namespace) + assert len(msgpack.packb("ab" * 600)) > len(envelope) + assert type(entry) is bytes and entry == envelope + assert backend.gets == 1 + + +def test_an_envelope_over_the_admission_cap_stays_out_though_its_msgpack_would_fit() -> None: + """Admission is judged on the envelope, as before. Judged on the MessagePack, this entry would get in and + evict others.""" + backend, namespace = _Backend(), _namespace() + value = os.urandom(131_067) # MessagePack 131,072 B, the cap at 1 MiB; the envelope is longer + + @cache(backend=backend, ttl=60, namespace=namespace, l1=L1CacheConfig(max_size_mb=1)) + def compute(x: int) -> bytes: + return value + + for _ in range(3): + assert compute(1) == value + envelope, entry = _stored(backend, namespace) + assert len(envelope) > 131_072 >= len(msgpack.packb(value)) + assert entry is None + assert backend.gets == 3 # every call reads L2 + + +def test_an_l2_hit_backfills_the_envelope_and_the_next_hit_swaps_it() -> None: + backend, namespace = _Backend(), _namespace() + + @cache(backend=backend, ttl=60, namespace=namespace) + def compute(x: int) -> dict[str, int]: + return {"x": x} + + assert compute(1) == {"x": 1} + get_l1_cache(namespace).clear() + assert compute(1) == {"x": 1} # L2 hit + envelope, entry = _stored(backend, namespace) + assert type(entry) is bytes and entry == envelope + assert compute(1) == {"x": 1} + _, entry = _stored(backend, namespace) + assert type(entry) is VerifiedMsgpack + assert backend.gets == 2 + + +def test_a_corrupt_envelope_backfilled_into_l1_is_never_swapped(monkeypatch: pytest.MonkeyPatch) -> None: + """MessagePack enters L1 only from a retrieve() that verified it. The L2 decode is stubbed to accept the + corrupt envelope, so the backfill puts it in L1; the next hit's retrieve() refuses it, and L1 drops it.""" + backend, namespace = _Backend(), _namespace() + + @cache(backend=backend, ttl=60, namespace=namespace) + def compute(x: int) -> dict[str, int]: + return {"x": x} + + compute(1) + ((key, envelope),) = backend.store.items() + backend.store[key] = corrupt = _corrupt(envelope) + l1 = get_l1_cache(namespace) + l1.clear() + real = CacheSerializationHandler.deserialize_data + monkeypatch.setattr( + CacheSerializationHandler, + "deserialize_data", + lambda self, data, *a, **kw: {"x": 1} if bytes(data) == corrupt else real(self, data, *a, **kw), + ) + + assert compute(1) == {"x": 1} # L2 hit on the corrupt envelope: backfilled as it is + assert l1.peek(key) == corrupt + assert compute(1) == {"x": 1} # the L1 hit refuses it; L2 serves and backfills it again + entry = l1.peek(key) + assert type(entry) is bytes and entry == corrupt + + +_VALUES = st.recursive( + st.none() | st.booleans() | st.integers(-(2**63), 2**64 - 1) | st.floats(allow_nan=False) | st.text() | st.binary(), + lambda inner: st.lists(inner, max_size=8) | st.dictionaries(st.text(max_size=8), inner, max_size=8), + max_leaves=40, +) + + +@settings(max_examples=300, deadline=None) +@given(value=_VALUES | st.binary(min_size=1000, max_size=20_000) | st.text(min_size=1000, max_size=20_000)) +def test_deserialize_l1_keeps_msgpack_only_when_no_longer_and_it_decodes_the_same(value: Any) -> None: + handler = CacheSerializationHandler() + envelope = handler.serialize_data(value, cache_key="k") + decoded, entry = handler.deserialize_l1(envelope, cache_key="k") + expected = handler.deserialize_data(envelope, cache_key="k") + assert decoded == expected + if len(msgpack.packb(value, use_bin_type=True)) > len(envelope): + assert entry is None + else: + assert type(entry) is VerifiedMsgpack and len(entry) <= len(envelope) + assert handler.deserialize_l1(entry, cache_key="k") == (expected, None) + + +def test_msgpack_exactly_as_long_as_the_limit_is_kept() -> None: + serializer = StandardSerializer() + envelope, _ = serializer.serialize({"x": 1}) + size = len(msgpack.packb({"x": 1})) + assert serializer._deserialize_keeping_msgpack(envelope, size)[1] is not None + assert serializer._deserialize_keeping_msgpack(envelope, size - 1)[1] is None + + +_OTHER_HANDLERS: dict[str, Callable[[], CacheSerializationHandler]] = { + "auto": lambda: CacheSerializationHandler(serializer_name="auto"), + "integrity-off": lambda: CacheSerializationHandler(enable_integrity_checking=False), + "encrypted": lambda: CacheSerializationHandler(encryption=True, master_key=MASTER_KEY, single_tenant_mode=True), +} + + +@pytest.mark.parametrize("make", list(_OTHER_HANDLERS.values()), ids=list(_OTHER_HANDLERS)) +def test_other_handlers_keep_their_envelope_and_refuse_verified_msgpack(make: Callable[[], CacheSerializationHandler]) -> None: + """The L1 hit sites send these handlers' entries to deserialize_data, which refuses a swapped entry another + handler left under the same key: under encryption as the plaintext envelope it replaced would be.""" + handler = make() + envelope = handler.serialize_data({"x": 1}, cache_key="k") + assert not handler.keeps_msgpack_in_l1 + assert handler.deserialize_l1(envelope, cache_key="k") == ({"x": 1}, None) + planted = VerifiedMsgpack(msgpack.packb({"x": 2})) + refusal = SuspiciousCacheEntryError if handler.encryption else SerializationError + for decode in (handler.deserialize_l1, handler.deserialize_data): + with pytest.raises(refusal) as raised: + decode(planted, cache_key="k") # type: ignore[arg-type] + assert handler.encryption or "Corrupt cache envelope" in str(raised.value) + + +def _frames() -> dict[str, bytes]: + """Entries a plaintext default-serializer handler must decode exactly as deserialize_data does. Only the + first is its own plaintext envelope, the one entry L1 may swap.""" + k = "k" + own = CacheSerializationHandler(encryption=False).serialize_data({"x": 1}, cache_key=k) + payload, metadata, name = SerializationWrapper.unwrap(own) + return { + "own-envelope": own, + "header-not-compressed": SerializationWrapper.wrap(bytes(payload), metadata | {"compressed": False}, name), + "other-serializer": CacheSerializationHandler(serializer_name="auto").serialize_data({"x": 1}, cache_key=k), + "encrypted-drift": CacheSerializationHandler( + encryption=True, master_key=MASTER_KEY, single_tenant_mode=True + ).serialize_data({"x": 1}, cache_key=k), + "integrity-off": CacheSerializationHandler(enable_integrity_checking=False).serialize_data({"x": 1}, cache_key=k), + "corrupt-header": b"\x00" + own[1:], + "corrupt-payload": _corrupt(own), + "truncated": own[:-30], + "not-a-frame": msgpack.packb({"x": 1}), + } + + +_FRAME_NAMES = ( + "own-envelope", + "header-not-compressed", + "other-serializer", + "encrypted-drift", + "integrity-off", + "corrupt-header", + "corrupt-payload", + "truncated", + "not-a-frame", +) + + +@pytest.mark.parametrize("name", _FRAME_NAMES) +def test_deserialize_l1_decodes_or_refuses_every_entry_as_deserialize_data_does( + name: str, monkeypatch: pytest.MonkeyPatch +) -> None: + """deserialize_l1's fast path repeats deserialize_data's frame checks, so the two must agree on every entry, + and only the handler's own plaintext envelope may come back with MessagePack to keep.""" + data = _frames()[name] + monkeypatch.setenv("CACHEKIT_MASTER_KEY", MASTER_KEY) # lets the encrypted-drift entry decrypt, as it would + reset_settings() + handler = CacheSerializationHandler(encryption=False) + + def outcome(decode: Callable[[], Any]) -> Any: + try: + return decode() + except Exception as e: # noqa: BLE001 - the outcome under comparison + return type(e) + + expected = outcome(lambda: handler.deserialize_data(data, cache_key="k")) + got = outcome(lambda: handler.deserialize_l1(data, cache_key="k")) + assert (got[0] if isinstance(got, tuple) else got) == expected + assert (isinstance(got, tuple) and got[1] is not None) == (name == "own-envelope") + + +# ── L1Cache.shrink_if_held ─────────────────────────────────────────────────────────────────── + + +@pytest.mark.parametrize("size", [40, 100], ids=["shorter", "as-long"]) +def test_shrink_if_held_swaps_the_entry_in_place(size: int) -> None: + l1 = L1Cache(max_memory_mb=1, namespace=_namespace()) + old = b"e" * 100 + l1.put("a", old, redis_ttl=60) + l1.put("b", b"b" * 10, redis_ttl=60) + expires_at = l1._state.cache["a"].expires_at + + new = VerifiedMsgpack(b"m" * size) + assert l1.shrink_if_held("a", old, new) + + assert l1.peek("a") is new + assert l1._state.memory_bytes == size + 10 + assert l1._state.cache["a"].expires_at == expires_at + assert list(l1._state.cache) == ["a", "b"] # LRU place kept + + +@pytest.mark.parametrize( + "change", ["removed", "stored-again", "longer"], ids=["removed-since-read", "stored-again-since-read", "longer"] +) +def test_shrink_if_held_leaves_an_entry_it_did_not_read_or_would_grow(change: str) -> None: + l1 = L1Cache(max_memory_mb=1, namespace=_namespace()) + old = b"e" * 100 + l1.put("a", old, redis_ttl=60) + new = VerifiedMsgpack(b"m" * 40) + if change == "removed": + l1.invalidate("a") + elif change == "stored-again": + l1.put("a", bytes(bytearray(old)), redis_ttl=60) # equal bytes, another object: a newer write + else: + new = VerifiedMsgpack(b"m" * 101) + held = l1.peek("a") + + assert not l1.shrink_if_held("a", old, new) + + assert l1.peek("a") is held + assert l1._state.memory_bytes == (0 if held is None else 100) + + +# ── encrypted handlers: ciphertext in L1, every hit decrypts ────────────────────────────────── + + +_SECURE = { + "secure": lambda ns, b: cache.secure(master_key=MASTER_KEY, backend=b, ttl=60, namespace=ns), + "secure-default-serializer": lambda ns, b: cache.secure( + master_key=MASTER_KEY, backend=b, ttl=60, namespace=ns, serializer="default" + ), + "encryption-true": lambda ns, b: cache( + backend=b, ttl=60, namespace=ns, encryption=True, single_tenant_mode=True, master_key=MASTER_KEY + ), +} + + +@pytest.mark.parametrize("decorate", list(_SECURE.values()), ids=list(_SECURE)) +def test_encrypted_l1_holds_the_l2_ciphertext_and_every_hit_decrypts( + decorate: Callable[[str, _Backend], Any], monkeypatch: pytest.MonkeyPatch +) -> None: + decrypts: list[int] = [] + real = EncryptionWrapper.deserialize + monkeypatch.setattr(EncryptionWrapper, "deserialize", lambda self, *a, **kw: decrypts.append(1) or real(self, *a, **kw)) + backend, namespace = _Backend(), _namespace() + + @decorate(namespace, backend) + def compute(x: int) -> dict[str, int]: + return {"x": x} + + assert compute(1) == {"x": 1} # miss: L1 gets the very envelope L2 got + for hits in (1, 2): + assert compute(1) == {"x": 1} # L1 hit + envelope, entry = _stored(backend, namespace) + assert type(entry) is bytes and entry == envelope + assert (len(decrypts), backend.gets) == (hits, 1) + + get_l1_cache(namespace).clear() + assert compute(1) == {"x": 1} # L2 hit: the backfill is the ciphertext too + assert compute(1) == {"x": 1} + envelope, entry = _stored(backend, namespace) + assert type(entry) is bytes and entry == envelope + assert (len(decrypts), backend.gets) == (4, 2) + + +def test_an_encrypted_reader_refuses_verified_msgpack_in_its_l1() -> None: + """A plaintext wrapper sharing the namespace and key cannot make an encrypted one serve plaintext.""" + backend, namespace = _Backend(), _namespace() + + @_SECURE["secure"](namespace, backend) + def compute(x: int) -> dict[str, int]: + return {"x": x} + + compute(1) + ((key, _),) = backend.store.items() + l1 = get_l1_cache(namespace) + assert l1.shrink_if_held(key, l1.peek(key), VerifiedMsgpack(msgpack.packb({"x": "planted"}))) # type: ignore[arg-type] + assert type(l1.peek(key)) is VerifiedMsgpack + assert compute(1) == {"x": 1} # read from L2 instead + assert type(l1.peek(key)) is bytes + + +# ── L1 hits: never fewer than with envelopes ────────────────────────────────────────────────── +# +# A replay over the real L1Cache. A read that misses puts the entry a put rule picks from the entry's +# envelope and MessagePack sizes; with promote, a read that hits an envelope swaps it for its MessagePack +# when that is no longer, as the L1 hit sites do (deserialize_l1, then L1Cache.shrink_if_held, which alone +# refuses a longer value here). The shipped design +# is the envelope rule with promote. The two rejected put rules are counterexamples against storing +# MessagePack at write time; they show the pinned traces bite. + +Rule = Callable[[L1Cache, int, int], int] + + +def _envelope_rule(l1: L1Cache, envelope: int, raw: int) -> int: + return envelope + + +def _growth_allowance_rule(l1: L1Cache, envelope: int, raw: int) -> int: + return raw if raw <= envelope * 1.1 and envelope <= l1.max_memory_bytes / 8 else envelope + + +def _admit_on_raw_rule(l1: L1Cache, envelope: int, raw: int) -> int: + return raw if raw <= envelope else envelope + + +class _Sized(bytes): + """A replayed L1 value: L1Cache accounts len(), and the replay never reads the bytes, so none are allocated.""" + + def __new__(cls, size: int) -> _Sized: + value = super().__new__(cls) + value.size = size # type: ignore[attr-defined] + return value + + def __len__(self) -> int: + return self.size # type: ignore[attr-defined] + + +class _SizedMsgpack(_Sized): + """A replayed entry a hit swapped in.""" + + +def _replay(trace: list[tuple[str, str, int, int]], budget_mb: int, rule: Rule, *, promote: bool = False) -> list[int]: + """The indices of the trace's reads that hit L1. + + Ops: ("read", key, ...) reads and, on a miss, puts; ("write", key, ...) puts whether or not L1 holds the + key (a revalidation, another writer); ("set", key, envelope, raw) is a new value at L2: L1 drops the key, and + later puts use the new sizes. A key's first op fixes its sizes until a "set". + """ + l1 = L1Cache(max_memory_mb=budget_mb, namespace=_namespace()) + sizes: dict[str, tuple[int, int]] = {} + hits = [] + for i, (op, key, envelope, raw) in enumerate(trace): + if op == "set": + l1.invalidate(key) + sizes[key] = (envelope, raw) + continue + envelope, raw = sizes.setdefault(key, (envelope, raw)) + if op == "read": + found, held = l1.get(key) + if found: + hits.append(i) + if promote and not isinstance(held, _SizedMsgpack): + l1.shrink_if_held(key, held, _SizedMsgpack(raw)) # type: ignore[arg-type] + assert l1._state.cache[key].size_bytes <= envelope # a swap never grows an entry + continue + l1.put(key, _Sized(rule(l1, envelope, raw))) + return hits + + +def _msgpack_sizes(e: int, cap: int) -> st.SearchStrategy[tuple[int, int]]: + """An envelope of ``e`` bytes and its MessagePack: the same, a little shorter or longer, anywhere up to + double, or either side of the admission cap.""" + return st.tuples( + st.just(e), + st.one_of(st.just(e), st.integers(max(1, e - 512), e + 512), st.integers(1, 2 * e), st.integers(cap - 256, cap + 256)), + ) + + +@st.composite +def _traces(draw: st.DrawFn, budget_mb: int) -> list[tuple[str, str, int, int]]: + """50 to 200 ops over n keys that together about fill L1, two keys of any size, and two that straddle the + admission cap: one whose envelope is refused but whose MessagePack would fit, and the reverse. The trace + opens with a "set" fixing each key's sizes; later "set"s draw new ones.""" + budget = budget_mb * 1024 * 1024 + cap = budget // 8 + n = draw(st.integers(2, 12)) + fill = min(cap, budget // n) + any_size = st.one_of(st.integers(cap - 256, cap + 256), st.integers(cap // 2, cap), st.integers(1, cap // 32)).flatmap( + lambda e: _msgpack_sizes(e, cap) + ) + sizes = {f"fill{i}": st.integers(max(1, fill - 512), fill).flatmap(lambda e: _msgpack_sizes(e, cap)) for i in range(n)} + sizes |= {"any0": any_size, "any1": any_size} + sizes |= { + "refused": st.tuples(st.integers(cap + 1, cap + 512), st.integers(cap - 512, cap)), + "admitted": st.tuples(st.integers(cap - 512, cap), st.integers(cap + 1, cap + 512)), + } + trace = [("set", key, *draw(strategy)) for key, strategy in sizes.items()] + ops = st.sampled_from(["read", "read", "read", "write", "set"]) + for op, key in draw(st.lists(st.tuples(ops, st.sampled_from(sorted(sizes))), min_size=50, max_size=200)): + trace.append((op, key, *draw(sizes[key])) if op == "set" else (op, key, 0, 0)) + return trace + + +@pytest.mark.parametrize("budget_mb", [1, 100], ids=["1MiB-minimum", "100MiB-default"]) +@settings(max_examples=100, deadline=None) +@given(data=st.data()) +def test_no_access_trace_gets_fewer_l1_hits_than_with_envelopes(budget_mb: int, data: st.DataObject) -> None: + trace = data.draw(_traces(budget_mb)) + with_envelopes = _replay(trace, budget_mb, _envelope_rule) + promoted = _replay(trace, budget_mb, _envelope_rule, promote=True) + assert set(with_envelopes) <= set(promoted) # every read that hit before still hits + + +def test_entries_growing_past_their_envelope_keep_it_so_a_near_full_lru_keeps_its_hits() -> None: + """104 keys cycling through 100 MiB, each 1,000,000 B as an envelope and 1,040,000 B as MessagePack. The + envelopes fit; with a 1.1x growth allowance the MessagePack would not, and LRU would miss every read.""" + keys = [f"k{i}" for i in range(104)] + trace = [("read", k, 1_000_000, 1_040_000) for _ in range(3) for k in keys] + hot = [i for i in range(len(trace)) if i >= len(keys)] + + assert _replay(trace, 100, _envelope_rule) == hot # 208/208 + assert _replay(trace, 100, _envelope_rule, promote=True) == hot + assert _replay(trace, 100, _growth_allowance_rule) == [] # 0/208 + + +def test_an_entry_shrunk_under_the_admission_cap_stays_out_so_the_hot_set_keeps_its_hits() -> None: + """Under 1 MiB (cap 131,072 B), an entry whose envelope is 131,080 B is refused. Admitted on its 131,000 B + MessagePack, it would evict the least recent hot entry, and each hot read would then evict the next.""" + hot_keys = [f"h{i}" for i in range(10)] + warm = [("read", k, 100_000, 100_000) for k in hot_keys] + rounds = [op for _ in range(2) for op in [("read", "big", 131_080, 131_000), *warm]] + trace = warm + rounds + + assert len(_replay(trace, 1, _envelope_rule)) == 20 + assert len(_replay(trace, 1, _envelope_rule, promote=True)) == 20 + assert _replay(trace, 1, _admit_on_raw_rule) == [] # 20 hits fell to 0 diff --git a/tests/unit/test_single_flight.py b/tests/unit/test_single_flight.py index 53b0b999..f766f1fe 100644 --- a/tests/unit/test_single_flight.py +++ b/tests/unit/test_single_flight.py @@ -37,12 +37,14 @@ from cachekit import cache from cachekit import object_cache as object_cache_module from cachekit.backends.errors import BackendError +from cachekit.cache_handler import CacheSerializationHandler from cachekit.config.nested import L1CacheConfig from cachekit.decorators import single_flight from cachekit.decorators.single_flight import AsyncFlights, ThreadFlights from cachekit.decorators.tenant_context import ContextVarExtractor from cachekit.l1_cache import L1Cache, get_l1_cache_manager from cachekit.reliability.circuit_breaker import CircuitBreaker, CircuitState +from cachekit.serializers.standard_serializer import VerifiedMsgpack pytestmark = pytest.mark.unit @@ -945,6 +947,69 @@ def compute(x: int) -> dict[str, int]: assert counts == (1, 0, 1) +@pytest.fixture +def decoded(monkeypatch: pytest.MonkeyPatch) -> list[type]: + """The type of every entry an L1 hit or a joined caller decodes (deserialize_l1): VerifiedMsgpack, or bytes for + an envelope.""" + seen: list[type] = [] + real = CacheSerializationHandler.deserialize_l1 + + def recording(self: CacheSerializationHandler, data: Any, *args: Any, **kwargs: Any) -> Any: + seen.append(type(data)) + return real(self, data, *args, **kwargs) + + monkeypatch.setattr(CacheSerializationHandler, "deserialize_l1", recording) + return seen + + +def test_sync_backed_recheck_decodes_a_verified_msgpack_l1_entry_without_a_trip( + monkeypatch: pytest.MonkeyPatch, decoded: list[type] +) -> None: + """Another caller's trip and then its L1 hit settle before this caller's flight lookup: the hit left the + value's verified MessagePack in L1 (deserialize_l1). The re-check hands it to _decode_shared, which decodes + it without retrieve() and without a trip of its own.""" + runs = 0 + + def compute(x: int) -> dict[str, int]: + nonlocal runs + runs += 1 + return {"x": x} + + backend = _Backend() + fn = _decorator("backed", backend)(compute) + real_run = ThreadFlights.run + lookups = 0 + + def run_after_another_trip_and_hit(self: ThreadFlights, *args: Any) -> Any: + nonlocal lookups + lookups += 1 + if lookups == 1: + assert fn(1) == fn(1) == {"x": 1} # another caller's trip, then its L1 hit + return real_run(self, *args) + + monkeypatch.setattr(ThreadFlights, "run", run_after_another_trip_and_hit) + assert fn(1) == {"x": 1} + + assert (runs, backend.requests) == (1, ["get", "set"]) # one function call and one backend read in all + assert decoded == [bytes, VerifiedMsgpack] # the other caller's hit, then this caller's re-check + + +def test_sync_backed_joiners_each_decode_their_own_copy_of_the_trips_envelope(waiting: list[int], decoded: list[type]) -> None: + """Each joined caller decodes the trip's envelope once, for its own copy, as before.""" + + def compute(x: int) -> dict[str, int]: + _until(lambda: waiting[0] == HERD - 1) + return {"x": x} + + backend = _Backend() + results = _thread_herd(_decorator("backed", backend)(compute)) + + assert results == [{"x": 1}] * HERD + assert len({id(result) for result in results}) == HERD + assert backend.requests == ["get", "set"] + assert decoded == [bytes] * (HERD - 1) + + def test_sync_backed_recheck_hit_hands_its_half_open_probe_slot_back( live_breakers: list[CircuitBreaker], monkeypatch: pytest.MonkeyPatch ) -> None: