diff --git a/.secrets.baseline b/.secrets.baseline index e6a4d1c0..44d94eac 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": 706 + "line_number": 719 } ], "src/cachekit/serializers/interop_serializer.py": [ @@ -2020,5 +2020,5 @@ } ] }, - "generated_at": "2026-10-31T23:25:29Z" + "generated_at": "2026-11-01T05:43:36Z" } diff --git a/docs/data-flow-architecture.md b/docs/data-flow-architecture.md index 302bf0b8..2d9467cd 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. After its first hit, a plaintext entry of the default serializer with integrity checking on is held as its verified MessagePack where that is no longer than the envelope, so later hits skip the decompress and checksum; every hit still runs the bounded decode (see [Storage Architecture](#storage-architecture)); encrypted caches keep the ciphertext +> - **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 of the default or auto serializer with integrity checking on is held as its verified MessagePack where that is no longer than the envelope, so later hits skip the decompress and checksum, and an orjson entry as its checksummed JSON, so later hits skip the checksum; every hit still runs the bounded decode (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 @@ -491,10 +491,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) -- **Kept entries**: 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 every read does: `ByteStorage.retrieve()` checks the envelope and returns the MessagePack its checksum covers, and the Rust decode (`unpackb_standard`) bounds and decodes it in one pass. When the verified MessagePack is no longer than the envelope, L1 then holds it in the envelope's place (`VerifiedMsgpack`, same expiry and LRU place), and later hits skip the decompress and checksum; otherwise L1 keeps the envelope. Every hit still runs the bounded decode. Compressible values are usually longer as MessagePack, so they keep the envelope; small and incompressible ones switch. The swap never grows 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, and bytes still held as bytes, never skip `retrieve()`: L1 marks a kept entry by its type, which no backend returns +- **Kept entries**: 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 every read does: `ByteStorage.retrieve()` checks the envelope and returns the MessagePack its checksum covers, and the Rust decode (`unpackb_standard`) bounds and decodes it in one pass. When the verified MessagePack is no longer than the envelope, L1 then holds it in the envelope's place (`VerifiedPayload`, same expiry and LRU place), and later hits skip the decompress and checksum; otherwise L1 keeps the envelope. The auto serializer does the same for its plain MessagePack envelopes, decoding with its own bounded decode and object hook; its NumPy, Arrow, DataFrame and Series entries keep the envelope and are verified on every hit. The orjson serializer keeps the JSON its xxHash3-64 checksum verified, which is always shorter than its envelope, so later hits skip the checksum. A kept entry records the serializer that verified it, and only that serializer decodes it. Every hit still runs the bounded decode. Compressible values are usually longer as MessagePack, so they keep the envelope; small and incompressible ones switch. The swap never grows 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, and bytes still held as bytes, never skip `retrieve()`: L1 marks a kept entry by its type, which no backend returns **Hierarchical Read Path:** -1. Check L1 cache (bytes) → if hit, deserialize and return (the first hit on a plaintext entry of the default serializer with integrity checking on marks it as verified, so later hits on held MessagePack skip the decompress and checksum) +1. Check L1 cache (bytes) → if hit, deserialize and return (on a plaintext cache with integrity checking on, the first hit keeps the payload it verified where the serializer allows: every orjson entry, and a default or auto MessagePack entry no longer than its envelope, but never an auto NumPy, Arrow, DataFrame or Series entry. Later hits on a kept payload skip the checksum, and for MessagePack the decompress; see Kept entries above) 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 9a981ba3..46aa970c 100644 --- a/docs/features/prometheus-metrics.md +++ b/docs/features/prometheus-metrics.md @@ -170,9 +170,11 @@ What each `cache_operation_duration_ms` series times, measured with a monotonic | `operation="set"`, a write `write_behind=True` deferred | The backend write alone, recorded when it lands. A write-behind miss that writes on the caller's thread instead is timed as the row above: an async miss under the distributed lock, a sync streaming write ([What stays on the caller's path](write-behind.md#what-stays-on-the-callers-path)), or a value the full writer queue refused ([The writer](write-behind.md#the-writer)) | `cache_operation_size_bytes` records the length of the bytes a `get` hit read; a `set` records no size. 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 the MessagePack is no longer than the envelope, so later L1 `get`s of it record a size no larger than its L2 `get`. Earlier releases recorded -the envelope's length on every hit. +on a plaintext cache with integrity checking on, an entry's first L1 hit can swap its envelope for the payload it verified, so later +L1 `get`s of it record a size no larger than its L2 `get`. With the default serializer, and with the auto serializer for a plain +MessagePack entry, the swap happens when the MessagePack is no longer than the envelope; the auto serializer's NumPy, Arrow, +DataFrame and Series entries keep the envelope and record its length on every hit. With the orjson serializer every entry swaps, +so later L1 `get`s record its JSON length. 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 diff --git a/docs/performance.md b/docs/performance.md index 8ec28060..a304ee20 100644 --- a/docs/performance.md +++ b/docs/performance.md @@ -75,7 +75,7 @@ With an L2 backend configured, L1 holds the serialized bytes instead, and a hit Every read with the default serializer decodes its MessagePack in the Rust extension, in one pass that also enforces the bounds on untrusted MessagePack: nesting at most 1,024 levels deep, and no header declaring more elements or bytes than the rest of the input can back. That covers L2 reads, L1 hits and the decrypted plaintext of encrypted reads. The decode builds the Python objects directly. It shares most repeated short strings, map keys up to 64 bytes and values up to 32, as one object within a value (a small table per decode; two strings that land in one slot replace each other). It runs the Python hook that restores datetimes only on maps holding a `__datetime__`, `__date__` or `__time__` key, because the hook hands back any other map unchanged. It returns the same values and refuses the same inputs with the same exception classes as the decode it replaced: cachekit-core's structure walk, then msgpack-python's C unpacker calling the hook once per map. On the 100-user dict (23,547 bytes of MessagePack, about 3,400 objects), the decode fell from 2,580,104 to 1,238,204 instructions on CPython 3.14 (−52.0%, callgrind main-thread instructions per call, the [Instruction Budgets](#instruction-budgets) method), level with `pickle.loads` of the same value (1,203,437). The 1,000-user dict (239 KB) fell 51.0% and a 1,090-byte one 45.3%. A list of 200 records each holding a datetime fell 33.2%, since every datetime still runs the hook. On the abi3t extension that CPython 3.15 and later load, the 100-user dict fell 50.5%. -For a plaintext cache with the default serializer and integrity checking on, an entry's first L1 hit decodes it as every read does, and then, when its verified MessagePack is no longer than the envelope, which is true of small values and incompressible ones, L1 holds that MessagePack in the envelope's place. Later hits on it skip the envelope's LZ4 decompress and xxHash3-64 checksum. Every hit still runs the bounded decode, which costs no more than the unbounded one did. The 100-user dict compresses to a 3,171-byte envelope from 23.5KB of MessagePack, so L1 keeps its envelope; no budget gates this large-value hit, and `l1_hit_backed` in [Instruction Budgets](#instruction-budgets) gates the small-value hit on held MessagePack. Encrypted caches always keep the ciphertext envelope, and every hit on one decrypts and decodes. +For a plaintext cache with the default or auto serializer and integrity checking on, an entry's first L1 hit decodes it as every read does, and then, when its verified MessagePack is no longer than the envelope, which is true of small values and incompressible ones, L1 holds that MessagePack in the envelope's place. Later hits on it skip the envelope's LZ4 decompress and xxHash3-64 checksum. The auto serializer keeps only plain MessagePack envelopes; its NumPy, Arrow, DataFrame and Series entries are verified on every hit. The orjson serializer keeps the JSON under its checksum, which is always shorter than the envelope, so its later hits skip the checksum. Every hit still runs the bounded decode, which costs no more than the unbounded one did. The 100-user dict compresses to a 3,171-byte envelope from 23.5KB of MessagePack, so L1 keeps its envelope; no budget gates this large-value hit, and `l1_hit_backed` and `l1_hit_backed_auto` in [Instruction Budgets](#instruction-budgets) gate the small-value hit on held MessagePack. Encrypted caches always keep the ciphertext envelope, and every hit on one decrypts and decodes. ### Decorator + L1 Hit (Hot Path) @@ -321,6 +321,7 @@ The regression gate is the instruction budget: `make perf-ir` on your machine, a |------|--------------------|-------------:|-------------:| | `l1_hit` | `@cache(backend=None)` L1 hit | 2,595 | 2,487 | | `l1_hit_backed` | `@cache` with a backend configured, L1 hit (never reaches the in-process dict backend; decodes the verified MessagePack L1 holds after the first hit) | 83,015 | 83,882 | +| `l1_hit_backed_auto` | `l1_hit_backed` with `serializer="auto"` (decodes the verified MessagePack the auto serializer keeps) | 91,986 | 92,950 | | `minimal_l1_hit` | `@cache.minimal(backend=None)` L1 hit | 2,595 | 2,487 | | `l2_hit` | `@cache`, L1 disabled, L2 hit, as one in-process single-flight trip | 172,539 | 167,686 | | `miss` | `@cache`, L1 disabled, L2 miss, compute, L2 write, all as one in-process single-flight trip | 213,364 | 204,897 | diff --git a/src/cachekit/cache_handler.py b/src/cachekit/cache_handler.py index fb3a791a..caae1f47 100644 --- a/src/cachekit/cache_handler.py +++ b/src/cachekit/cache_handler.py @@ -50,12 +50,14 @@ from cachekit.hash_utils import redact_cache_key, redact_error_for_log from cachekit.interop import InteropError from cachekit.key_generator import CacheKeyGenerator +from cachekit.serializers.auto_serializer import AutoSerializer from cachekit.serializers.base import ( EnvelopeShapeError, SerializationError, SerializationFormat, SerializationMetadata, SuspiciousCacheEntryError, + VerifiedPayload, bounded_error, ) from cachekit.serializers.encryption_wrapper import ( @@ -64,7 +66,7 @@ KeyringConfigurationError, require_cross_sdk_serializer, ) -from cachekit.serializers.standard_serializer import StandardSerializer, VerifiedMsgpack +from cachekit.serializers.standard_serializer import StandardSerializer from cachekit.serializers.wrapper import SerializationWrapper if TYPE_CHECKING: @@ -585,6 +587,17 @@ def _get_cached_serializer_class(serializer_name: str, import_path: str): raise +def _keeps_verified(serializer: Any) -> bool: + """Whether ``serializer`` is exactly a built-in that mints :class:`VerifiedPayload` for L1 to keep + (``_deserialize_keeping`` / ``_deserialize_verified``), with integrity checking on: integrity off, there is + no check for a kept entry to skip. Exact types: a subclass may override the decode. OrjsonSerializer's + module needs the [json] extra, so it is looked up only once loaded, as it is wherever an instance exists. + """ + orjson_module = sys.modules.get("cachekit.serializers.orjson_serializer") + keeping = (StandardSerializer, AutoSerializer, *((orjson_module.OrjsonSerializer,) if orjson_module else ())) + return type(serializer) in keeping and serializer.enable_integrity_checking + + def _get_cached_serializer_instance( serializer: Union[str, SerializerProtocol], enable_integrity_checking: bool = True ) -> SerializerProtocol: # type: ignore[name-defined] @@ -955,14 +968,9 @@ def __init__( 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 a kept one (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 - ) + # CK-frame entries of a built-in serializer that keeps one, with integrity on (_keeps_verified; interop + # mode's InteropSerializer is not one). Encrypted handlers keep ciphertext in L1 (protocol PRE-10, PRE-11). + self.keeps_verified_in_l1 = not encryption and not self._custom_serializer and _keeps_verified(self._base_serializer) # Cache EncryptionWrapper instances per tenant to prevent # 360K key copies/hour at 100 req/sec. Uses thread-safe LRU cache (maxsize=256) @@ -1351,27 +1359,28 @@ def write_serialized_to(self, sink: BinaryIO, data: Any) -> None: def deserialize_l1( self, - data: bytes | VerifiedMsgpack, + data: bytes | VerifiedPayload, cache_key: str = "", args: tuple[Any, ...] = (), kwargs: dict[str, Any] | None = None, - ) -> tuple[Any, VerifiedMsgpack | None]: + ) -> tuple[Any, VerifiedPayload | None]: """:meth:`deserialize_data` for an entry read from L1, plus what L1 should hold in its place, if anything. - When this handler keeps MessagePack (``keeps_msgpack_in_l1``), the entry is its own plaintext envelope, and - the MessagePack this decode verified is no longer than the entry, that MessagePack (:class:`VerifiedMsgpack`), - for the caller to swap in (``L1Cache.shrink_if_held``) at no second decode: later hits skip the envelope's - retrieve(), its decompress and checksum. Everything else gets None: a kept entry decodes that way, a longer - MessagePack would grow the entry so L1 keeps the envelope, and any other entry goes to deserialize_data, - which refuses another handler's kept entry. Every decode, kept or not, runs the bounded decode. Raises as - deserialize_data. + When this handler keeps verified payloads (``keeps_verified_in_l1``), the entry is its own plaintext envelope + of a kind its serializer keeps, and the payload this decode verified is no longer than the entry, that payload + (:class:`VerifiedPayload`: the MessagePack ``ByteStorage.retrieve()`` returned for the default and auto + serializers, the checksummed JSON for orjson), for the caller to swap in (``L1Cache.shrink_if_held``) at no + second decode: later hits skip the integrity check (retrieve()'s decompress and checksum, or orjson's + checksum). Everything else gets None: a kept entry decodes that way, a longer payload would grow the entry + so L1 keeps the envelope, and any other entry goes to deserialize_data, which refuses another handler's kept + entry. Every decode, kept or not, runs its serializer's bounded decode. 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) + ('hello', 'VerifiedPayload', True) >>> handler.deserialize_l1(entry, cache_key="greeting") # already verified MessagePack ('hello', None) >>> compressible = handler.serialize_data("ab" * 600, cache_key="k") @@ -1380,26 +1389,34 @@ def deserialize_l1( """ # 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: + if not self.keeps_verified_in_l1: return self.deserialize_data(data, cache_key, args, kwargs), None - # Only L1 holds this type, and only such a handler swapped it in, after a decode that accepted it. - if type(data) is VerifiedMsgpack: - return self._base_serializer._deserialize_verified(data.data), None # type: ignore[union-attr] + serializer = self._base_serializer + # Only L1 holds this type, and only a keeping handler swapped it in, after a decode that accepted it. Its + # owner decodes it; another serializer's (a key shared across serializers) goes to deserialize_data, which + # refuses it. + if type(data) is VerifiedPayload: + if data.owner is not type(serializer): + return self.deserialize_data(data, cache_key, args, kwargs), None + return serializer._deserialize_verified(data.data), None # type: ignore[union-attr] 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: + # Only this handler's own plaintext envelope. Anything else, such as a config-drift entry whose ciphertext + # must stay in L1, is decoded by deserialize_data, not kept; so is an entry the serializer does not keep + # (_deserialize_keeping returns None), such as one whose header misdescribes what this handler writes. + if serializer_name != self._serializer_string_name or metadata.encrypted: + return self.deserialize_data(data, cache_key, args, kwargs), None + keeping = serializer._deserialize_keeping(payload, metadata) # type: ignore[union-attr] + if keeping is None: return self.deserialize_data(data, cache_key, args, kwargs), None - value, kept = self._base_serializer._deserialize_keeping_msgpack(payload) # type: ignore[union-attr] + value, kept = keeping return value, kept if len(kept) <= len(data) else None def deserialize_data( self, - data: str | bytes | memoryview | VerifiedMsgpack, + data: str | bytes | memoryview | VerifiedPayload, cache_key: str = "", args: tuple[Any, ...] = (), kwargs: dict[str, Any] | None = None, @@ -1476,7 +1493,7 @@ def deserialize_data( True """ if self.interop_mode: - # A VerifiedMsgpack here is another handler's swapped L1 entry: the bytes-only decode refuses it. + # A VerifiedPayload here is another handler's swapped L1 entry: the bytes-only decode refuses it. return self._deserialize_interop(data, cache_key) # pyright: ignore[reportArgumentType] try: @@ -1488,11 +1505,11 @@ def deserialize_data( # A custom SerializerProtocol may write to the metadata it is handed, so it gets its own # copy rather than the header memo's shared, read-only instance. serialized_data, metadata, serializer_name = SerializationWrapper.unwrap_metadata( - data, # pyright: ignore[reportArgumentType] # a VerifiedMsgpack raises here and is refused below + data, # pyright: ignore[reportArgumentType] # a VerifiedPayload raises here and is refused below shared=not self._custom_serializer, ) except (AttributeError, KeyError, TypeError, ValueError) as e: - if self.encryption and type(data) is VerifiedMsgpack: + if self.encryption and type(data) is VerifiedPayload: # 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( diff --git a/src/cachekit/decorators/wrapper.py b/src/cachekit/decorators/wrapper.py index 773c8a2a..a65d51dd 100644 --- a/src/cachekit/decorators/wrapper.py +++ b/src/cachekit/decorators/wrapper.py @@ -78,8 +78,7 @@ from pydantic import SecretStr from ..backends.base import BaseBackend, LockableBackend - from ..serializers.base import SerializerProtocol - from ..serializers.standard_serializer import VerifiedMsgpack + from ..serializers.base import SerializerProtocol, VerifiedPayload def _resolve_lazy_backend() -> BaseBackend: @@ -1087,9 +1086,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 its verified MessagePack (deserialize_l1). Every other handler's + # Whether an L1 hit may swap its entry for its verified payload (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_keeps_verified = serialization_handler.keeps_verified_in_l1 # L1-only mode: use ObjectCache for raw Python object storage (no serialization). # This preserves types (tuples, sets, frozensets) that MessagePack would degrade. @@ -1838,13 +1837,13 @@ def _flight_tenant(args: tuple[Any, ...], kwargs: dict[str, Any], cache_key: str return None def _decode_shared( - entry: bytes | VerifiedMsgpack | None, cache_key: str, args: tuple[Any, ...], kwargs: dict[str, Any] + entry: bytes | VerifiedPayload | 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, 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 + found (which may be a VerifiedPayload), 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, so bytes go to deserialize_data, which decodes them as deserialize_l1 does without building a - VerifiedMsgpack for it to discard. + VerifiedPayload for it to discard. (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. @@ -2180,11 +2179,11 @@ def _sync_call(args: tuple[Any, ...], kwargs: dict[str, Any]) -> Any: # noqa: P if l1_found and l1_bytes: # L1 cache hit - deserialize bytes try: - if _l1_keeps_msgpack: + if _l1_keeps_verified: l1_value, l1_kept = operation_handler.serialization_handler.deserialize_l1( l1_bytes, cache_key, args, kwargs ) - if l1_kept is not None: # verified and kept: later hits skip retrieve() (deserialize_l1) + if l1_kept is not None: # verified and kept: later hits skip its integrity check (deserialize_l1) _l1_cache.shrink_if_held(cache_key, l1_bytes, l1_kept) else: # as before; a swapped entry fails: SuspiciousCacheEntryError if encrypted, else a corrupt frame l1_value = operation_handler.serialization_handler.deserialize_data( @@ -2352,9 +2351,9 @@ 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 | VerifiedMsgpack | None]]: + def _peek_l1(cache_key: str) -> tuple[bool, tuple[None, bytes | VerifiedPayload | 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 (its envelope, or the VerifiedMsgpack a hit swapped in), for a joined + the caller's own L1 lookup left (its envelope, or the VerifiedPayload 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 entry = _l1_cache.peek(cache_key) @@ -2675,7 +2674,7 @@ async def _async_call(args: tuple[Any, ...], kwargs: dict[str, Any]) -> Any: if l1_found and l1_bytes: # L1 cache hit - deserialize bytes try: - if _l1_keeps_msgpack: # as in sync_wrapper + if _l1_keeps_verified: # as in sync_wrapper l1_value, l1_kept = operation_handler.serialization_handler.deserialize_l1( l1_bytes, cache_key, args, kwargs ) diff --git a/src/cachekit/l1_cache.py b/src/cachekit/l1_cache.py index ea653d7d..885455cc 100644 --- a/src/cachekit/l1_cache.py +++ b/src/cachekit/l1_cache.py @@ -21,7 +21,7 @@ from cachekit.hash_utils import redact_error_for_log, redact_key_for_log if TYPE_CHECKING: - from cachekit.serializers.standard_serializer import VerifiedMsgpack + from cachekit.serializers.base import VerifiedPayload # 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 @@ -55,12 +55,12 @@ class CacheEntry: """L1 cache entry with value, TTL, and size tracking. Storage: Stores bytes (encrypted or plaintext msgpack) for unified - encrypted-at-rest architecture, or the VerifiedMsgpack a hit swapped in for a + encrypted-at-rest architecture, or the VerifiedPayload a hit swapped in for a plaintext entry. Decryption/deserialization happens at read time in CacheHandler, not storage time. """ - value: "bytes | VerifiedMsgpack" # a VerifiedMsgpack once a hit swapped it in (shrink_if_held) + value: "bytes | VerifiedPayload" # a VerifiedPayload once a hit swapped it in (shrink_if_held) expires_at: float size_bytes: int # This entry's tuple in its state's expiry heap. A tuple is live only while it is this very object, @@ -161,7 +161,7 @@ class L1Cache: - LRU eviction when memory limit reached - In-process lookups, with no network round trip - Background TTL synchronization - - Stores bytes (encrypted or plaintext msgpack), or the VerifiedMsgpack a hit swapped in, not Python objects + - Stores bytes (encrypted or plaintext msgpack), or the VerifiedPayload a hit swapped in, not Python objects Storage Architecture: - Stores serialized bytes for unified encrypted-at-rest @@ -211,21 +211,21 @@ def __init__( ttl_buffer_seconds, ) - def _estimate_size(self, value: "bytes | VerifiedMsgpack") -> int: + def _estimate_size(self, value: "bytes | VerifiedPayload") -> int: """Estimate memory size of an entry's value. Simplified from recursive sys.getsizeof() - faster and more accurate - for bytes storage. A VerifiedMsgpack's length is that of the bytes it holds. + for bytes storage. A VerifiedPayload's length is that of the bytes it holds. Args: - value: Bytes, or a VerifiedMsgpack, to estimate size for + value: Bytes, or a VerifiedPayload, to estimate size for Returns: Size in bytes """ return len(value) - def get(self, key: str) -> "tuple[bool, Optional[bytes | VerifiedMsgpack]]": + def get(self, key: str) -> "tuple[bool, Optional[bytes | VerifiedPayload]]": """Get value from L1 cache if present and not expired. Args: @@ -233,7 +233,7 @@ def get(self, key: str) -> "tuple[bool, Optional[bytes | VerifiedMsgpack]]": Returns: Tuple of (found, value) where found is True if hit. - Value is bytes (encrypted or plaintext msgpack), or the VerifiedMsgpack a hit swapped in + Value is bytes (encrypted or plaintext msgpack), or the VerifiedPayload a hit swapped in (shrink_if_held), not deserialized object. Caller (CacheHandler) is responsible for decryption/deserialization. """ @@ -259,8 +259,8 @@ def get(self, key: str) -> "tuple[bool, Optional[bytes | VerifiedMsgpack]]": return True, entry.value - def peek(self, key: str) -> "Optional[bytes | VerifiedMsgpack]": - """The value ``get`` would return (bytes, or the VerifiedMsgpack a hit swapped in), without counting a hit or miss or moving the entry in LRU order. + def peek(self, key: str) -> "Optional[bytes | VerifiedPayload]": + """The value ``get`` would return (bytes, or the VerifiedPayload a hit swapped in), 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 the call's first lookup already counted. An expired entry reads as absent and is left for @@ -276,7 +276,7 @@ def peek(self, key: str) -> "Optional[bytes | VerifiedMsgpack]": 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: + def shrink_if_held(self, key: str, held: "bytes | VerifiedPayload", smaller: "VerifiedPayload") -> 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. diff --git a/src/cachekit/serializers/auto_serializer.py b/src/cachekit/serializers/auto_serializer.py index 1c30d1f3..5a1e739c 100644 --- a/src/cachekit/serializers/auto_serializer.py +++ b/src/cachekit/serializers/auto_serializer.py @@ -52,7 +52,9 @@ SerializationError, SerializationFormat, SerializationMetadata, + VerifiedPayload, bounded_error, + immutable_buffer, unpackb_bounded, ) @@ -810,8 +812,10 @@ def deserialize(self, data: bytes | memoryview, metadata: Optional[Serialization whose header lost ``original_type`` still decodes on an integrity-off reader rather than failing closed. """ - # coerce unwrap's zero-copy memoryview; no-op when already bytes (enables .startswith below + Rust retrieve) - data = bytes(data) + # No bytes() copy of unwrap's zero-copy memoryview: every route below reads a buffer. A mutable one is + # snapshotted, so the routing tests and the decode judge the same bytes. bytes skips the call. + if type(data) is not bytes: + data = immutable_buffer(data) # The header's claim about the stored format. Read once: binding it twice under two names # is how a header-sourced value reached a variable holding the envelope's format before. # Typed once too, here rather than at each of the three sites that compare or echo it — @@ -829,7 +833,7 @@ def deserialize(self, data: bytes | memoryview, metadata: Optional[Serialization # digest, failed it, and raised on every read — a permanent miss on a healthy key. # Both structural routes hand the delegate the gate-verified body, so it is hashed once. numpy_body = None - if data.startswith(b"NUMPY_RAW"): + if data[:9] == b"NUMPY_RAW": numpy_body = memoryview(data) elif self._checksummed_prefix(data, b"NUMPY_RAW"): numpy_body = memoryview(data)[8:] @@ -910,27 +914,7 @@ def deserialize(self, data: bytes | memoryview, metadata: Optional[Serialization f"Rust envelope parsing failed, falling back to Python-only deserialization: {redact_error_for_log(e)}" ) else: - # The envelope verified (checksum matched), so its payload is exactly what was - # stored; a payload that then fails to decode is corruption or a forged entry - # (e.g. a decode bomb) and MUST fail closed. Falling through here used to - # re-decode the ENVELOPE bytes as plain MessagePack and return its positional - # fields as the cached value — wrong data, silently. - # Format by agreement, never by a winner — see Raises:. - if format_id not in _ENVELOPE_FORMATS: - raise self._envelope_failure(f"unknown envelope format {format_id!r:.40}") - if header_format is not None and header_format != format_id: - raise self._envelope_failure( - f"envelope format {format_id!r:.40} disagrees with header format {header_format!r:.40}" - ) - try: - if format_id in ("dataframe", "series"): - return self._decode_columnar(original_data, format_id) - return unpackb_bounded(original_data, **self._msgpack_unpack_opts) - except PAYLOAD_DECODE_ERRORS as e: - raise SerializationError( - f"Cache entry payload failed to decode inside a verified envelope (format={format_id!r:.40}): " - f"{bounded_error(e)}" - ) from e + return self._decode_verified_envelope(original_data, format_id, header_format) # Reached with integrity on when retrieve() raised the "not an envelope" ValueError # (envelope_error is set), or with integrity off (envelope_error is None; a @@ -974,6 +958,66 @@ def deserialize(self, data: bytes | memoryview, metadata: Optional[Serialization ) return value + def _decode_verified_envelope(self, original_data: bytes, format_id: str, header_format: str | None) -> Any: + """Decode the payload of an envelope ``retrieve()`` verified, whose format is ``format_id``. + + The envelope verified (checksum matched), so its payload is exactly what was stored; a payload that then + fails to decode is corruption or a forged entry (e.g. a decode bomb) and MUST fail closed. Falling through + used to re-decode the ENVELOPE bytes as plain MessagePack and return its positional fields as the cached + value — wrong data, silently. Format by agreement, never by a winner — see :meth:`deserialize`'s Raises:. + """ + if format_id not in _ENVELOPE_FORMATS: + raise self._envelope_failure(f"unknown envelope format {format_id!r:.40}") + if header_format is not None and header_format != format_id: + raise self._envelope_failure(f"envelope format {format_id!r:.40} disagrees with header format {header_format!r:.40}") + try: + if format_id in ("dataframe", "series"): + return self._decode_columnar(original_data, format_id) + return unpackb_bounded(original_data, **self._msgpack_unpack_opts) + except PAYLOAD_DECODE_ERRORS as e: + raise SerializationError( + f"Cache entry payload failed to decode inside a verified envelope (format={format_id!r:.40}): {bounded_error(e)}" + ) from e + + def _deserialize_keeping( + self, data: bytes | memoryview, metadata: SerializationMetadata + ) -> tuple[Any, VerifiedPayload] | None: + """What :meth:`deserialize` returns for a plain MessagePack envelope when integrity checking is on, plus + the MessagePack ``retrieve()`` verified and this decode accepted, for the caller to keep and decode later + without retrieve() (:meth:`_deserialize_verified`). Call it only for an integrity-on serializer. + + None, decoding nothing, for every other entry: a header that claims another format (NumPy, Arrow, + DataFrame, Series), and any bytes with the ``NUMPY_RAW`` magic at offset 0 or 8. That magic test is + deliberately broader than :meth:`deserialize`'s checksum-gated NumPy route, which refuses such bytes + under a ``"msgpack"`` header: a broader test only declines to keep. The caller decodes those as any other entry, so only the route + :meth:`deserialize` takes for this header to ``retrieve()`` runs here, and it fails as that route does. + + Examples: + >>> serializer = AutoSerializer() + >>> envelope, metadata = serializer.serialize({"when": (1, 2)}) + >>> value, kept = serializer._deserialize_keeping(envelope, metadata) + >>> value, serializer._deserialize_verified(kept.data) + ({'when': (1, 2)}, {'when': (1, 2)}) + """ + if metadata.original_type != "msgpack" or data[:9] == b"NUMPY_RAW" or data[8:17] == b"NUMPY_RAW": + return None + try: + msgpack_data, format_id = self._byte_storage.retrieve(data) + except Exception as e: # as deserialize does with a header present: fail closed, never re-parse + raise self._envelope_failure(e) from e + value = self._decode_verified_envelope(msgpack_data, format_id, "msgpack") + return value, VerifiedPayload(msgpack_data, AutoSerializer) + + def _deserialize_verified(self, msgpack_data: bytes) -> Any: + """Decode MessagePack a decode in this process already verified (:meth:`_deserialize_keeping`), without + retrieve(). Still the bounded decode. Raises SerializationError as deserialize does.""" + try: + return unpackb_bounded(msgpack_data, **self._msgpack_unpack_opts) + except PAYLOAD_DECODE_ERRORS as e: + raise SerializationError( + f"Cache entry payload failed to decode inside a verified envelope (format='msgpack'): {bounded_error(e)}" + ) from e + def _serialize_numpy(self, arr: np.ndarray) -> bytes: # type: ignore[name-defined] """Serialize a NumPy array into the ``NUMPY_RAW`` binary format. @@ -1015,7 +1059,7 @@ def _serialize_numpy(self, arr: np.ndarray) -> bytes: # type: ignore[name-defin return xxhash.xxh3_64_digest(raw) + raw return raw - def _deserialize_numpy(self, data: bytes) -> np.ndarray: + def _deserialize_numpy(self, data: bytes | memoryview) -> np.ndarray: """Deserialize NumPy array from NUMPY_RAW binary format. Raises: @@ -1026,7 +1070,7 @@ def _deserialize_numpy(self, data: bytes) -> np.ndarray: # serialization. Detect by structure (like ArrowSerializer): a checksummed entry is # [8-byte checksum][NUMPY_RAW...]; a raw entry (integrity-off / legacy) is [NUMPY_RAW...]. # A mismatch fails closed (#155) — never reconstructs the corrupted array. - if not data.startswith(b"NUMPY_RAW") and len(data) >= 17 and data[8:17] == b"NUMPY_RAW": + if data[:9] != b"NUMPY_RAW" and len(data) >= 17 and data[8:17] == b"NUMPY_RAW": body = memoryview(data)[8:] if xxhash.xxh3_64_digest(body) != data[:8]: raise SerializationError("NumPy integrity check failed: xxHash3-64 checksum mismatch (corrupted cache entry)") @@ -1199,7 +1243,7 @@ def _deserialize_series(self, document) -> pd.Series: return series @staticmethod - def _checksummed_prefix(data: bytes, magic: bytes) -> bool: + def _checksummed_prefix(data: bytes | memoryview, magic: bytes) -> bool: """True iff ``data`` is ``[8-byte xxHash3-64][magic...]`` and the digest matches the body. One gate for both self-checksummed formats (Arrow, NumPy). It disambiguates a real diff --git a/src/cachekit/serializers/base.py b/src/cachekit/serializers/base.py index 47b07d2a..b8fa6fbf 100644 --- a/src/cachekit/serializers/base.py +++ b/src/cachekit/serializers/base.py @@ -6,7 +6,7 @@ from __future__ import annotations from enum import Enum -from typing import Any, ClassVar, Protocol, runtime_checkable +from typing import Any, ClassVar, Protocol, final, runtime_checkable import msgpack @@ -429,6 +429,28 @@ def bounded_error(exc: BaseException | str) -> str: return text.translate(_LOG_UNSAFE_ESCAPES) +@final # the read paths test `type(data) is VerifiedPayload` +class VerifiedPayload: + """A payload its serializer's own integrity check returned and a decode in this process accepted, which an L1 + entry holds in place of its envelope (``CacheSerializationHandler.deserialize_l1``). ``owner`` (the serializer's + class) decodes it without that check again: the MessagePack ``ByteStorage.retrieve()`` returned for + StandardSerializer and AutoSerializer, the JSON under a matching xxHash3-64 for OrjsonSerializer. + + The type is the mark, not the bytes: no backend returns this type, so bytes from L2 never skip the check, + whatever they hold. ``owner`` keeps one serializer from decoding another's payload. Its length is that of + ``data``, which L1 accounts. + """ + + __slots__ = ("data", "owner") + + def __init__(self, data: bytes, owner: type) -> None: + self.data = data + self.owner = owner + + def __len__(self) -> int: + return len(self.data) + + def immutable_buffer(data: bytes | bytearray | memoryview) -> bytes | memoryview: """Return ``data`` as one flat, unsigned, immutable byte buffer, copying only when it must. diff --git a/src/cachekit/serializers/orjson_serializer.py b/src/cachekit/serializers/orjson_serializer.py index b6b051bb..70a9ede4 100644 --- a/src/cachekit/serializers/orjson_serializer.py +++ b/src/cachekit/serializers/orjson_serializer.py @@ -22,7 +22,7 @@ import xxhash -from .base import SerializationError, SerializationFormat, SerializationMetadata +from .base import SerializationError, SerializationFormat, SerializationMetadata, VerifiedPayload, immutable_buffer class OrjsonSerializer: @@ -182,32 +182,60 @@ def deserialize(self, data: bytes | memoryview, metadata: SerializationMetadata >>> result == {"test": 123} True """ - data = bytes(data) # coerce unwrap's zero-copy memoryview; no-op when already bytes + # No bytes() copy of unwrap's zero-copy memoryview, and no slice copy of the JSON in it: orjson and xxhash + # read buffers. A mutable one is snapshotted, so the checksum and the decode judge the same bytes. bytes + # skips the call, which costs more than the copy it saves on a small value. + if type(data) is not bytes: + data = immutable_buffer(data) try: if self.enable_integrity_checking: - # Guard clause: Minimum size check (8 bytes checksum + at least 2 bytes JSON: {}) - if len(data) < 10: - raise SerializationError( - f"Invalid data: Expected at least 10 bytes (8-byte checksum + 2-byte JSON), got {len(data)} bytes" - ) - - # Extract checksum and JSON data - expected_checksum = data[:8] - json_data = data[8:] - - # Validate checksum - computed_checksum = xxhash.xxh3_64_digest(json_data) - if computed_checksum != expected_checksum: - raise SerializationError("Checksum validation failed - data corruption detected") - - # Deserialize JSON - return orjson.loads(json_data) - else: - # No integrity checking - deserialize directly - # This handles both: data written with integrity=False AND backward compatible reads - # If data has checksum but we're not validating, just try to parse as JSON - return orjson.loads(data) - except orjson.JSONDecodeError as e: + return orjson.loads(self._verified_json(data)) + # No integrity checking - deserialize directly + # This handles both: data written with integrity=False AND backward compatible reads + # If data has checksum but we're not validating, just try to parse as JSON + return orjson.loads(data) + except ValueError as e: # orjson.JSONDecodeError is one raise SerializationError(f"Failed to deserialize JSON data: {e}") from e - except ValueError as e: + + @staticmethod + def _verified_json(data: bytes | memoryview) -> bytes | memoryview: + """The JSON of a ``[8-byte xxHash3-64][JSON]`` envelope, as a view, once its checksum matches. + + Raises: + SerializationError: The envelope is too short to hold a checksum and JSON, or the checksum does not + match (corruption). + """ + # Guard clause: Minimum size check (8 bytes checksum + at least 2 bytes JSON: {}) + if len(data) < 10: + raise SerializationError( + f"Invalid data: Expected at least 10 bytes (8-byte checksum + 2-byte JSON), got {len(data)} bytes" + ) + json_data = data[8:] # a view of a view; bytes, which only direct callers pass, copies as it always did + if xxhash.xxh3_64_digest(json_data) != data[:8]: + raise SerializationError("Checksum validation failed - data corruption detected") + return json_data + + def _deserialize_keeping(self, data: bytes | memoryview, metadata: SerializationMetadata) -> tuple[Any, VerifiedPayload]: + """What :meth:`deserialize` returns when integrity checking is on, plus the JSON its checksum verified, for + the caller to keep and decode later without the checksum (:meth:`_deserialize_verified`). Every entry this + serializer writes is that envelope, whatever its header (orjson reads none), so it never returns None. The + kept JSON is the one copy of the payload a hit makes. Call it only for an integrity-on serializer. Raises + SerializationError as deserialize does then. + + Examples: + >>> serializer = OrjsonSerializer() + >>> envelope, metadata = serializer.serialize({"test": 123}) + >>> value, kept = serializer._deserialize_keeping(envelope, metadata) + >>> value, serializer._deserialize_verified(kept.data), len(kept) == len(envelope) - 8 + ({'test': 123}, {'test': 123}, True) + """ + json_data = bytes(self._verified_json(immutable_buffer(data))) + return self._deserialize_verified(json_data), VerifiedPayload(json_data, OrjsonSerializer) + + def _deserialize_verified(self, json_data: bytes) -> Any: + """Decode JSON a decode in this process already verified (:meth:`_deserialize_keeping`), without the + checksum. Raises SerializationError as deserialize does.""" + try: + return orjson.loads(json_data) + except ValueError as e: # orjson.JSONDecodeError is one raise SerializationError(f"Failed to deserialize JSON data: {e}") from e diff --git a/src/cachekit/serializers/standard_serializer.py b/src/cachekit/serializers/standard_serializer.py index 8b3c0791..2792a56c 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, final +from typing import Any, ClassVar import msgpack @@ -33,6 +33,7 @@ SerializationError, SerializationFormat, SerializationMetadata, + VerifiedPayload, immutable_buffer, ) @@ -48,26 +49,6 @@ _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 and a decode in this process accepted, - 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 skip - retrieve(), 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). " @@ -408,28 +389,34 @@ 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) -> tuple[Any, VerifiedMsgpack]: + def _deserialize_keeping( + self, data: bytes | memoryview, metadata: SerializationMetadata + ) -> tuple[Any, VerifiedPayload] | None: """What :meth:`deserialize` returns for a ByteStorage envelope when integrity checking is on, plus the MessagePack ``retrieve()`` verified and this decode accepted, 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. + without retrieve() (:meth:`_deserialize_verified`). None, decoding nothing, when the header does not say + what an integrity-on writer wrote (``compressed``): the caller decodes such an entry as any other. 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) + >>> envelope, metadata = serializer.serialize({"test": 123}) + >>> value, kept = serializer._deserialize_keeping(envelope, metadata) >>> value, serializer._deserialize_verified(kept.data) ({'test': 123}, {'test': 123}) """ + if not metadata.compressed: + return None try: msgpack_data, _ = self._byte_storage.retrieve(data) value = unpackb_standard(msgpack_data, MSGPACK_MAX_NESTING, _standard_object_hook) except PAYLOAD_DECODE_ERRORS as e: raise SerializationError(f"Failed to deserialize MessagePack data: {e}") from e - return value, VerifiedMsgpack(msgpack_data) + return value, VerifiedPayload(msgpack_data, StandardSerializer) def _deserialize_verified(self, msgpack_data: bytes) -> Any: - """Decode MessagePack a decode in this process already verified (:meth:`_deserialize_keeping_msgpack`), + """Decode MessagePack a decode in this process already verified (:meth:`_deserialize_keeping`), without retrieve(). Raises SerializationError as deserialize does.""" try: return unpackb_standard(msgpack_data, MSGPACK_MAX_NESTING, _standard_object_hook) diff --git a/tests/performance/ir_baselines.json b/tests/performance/ir_baselines.json index 3a08d4b7..b9ddbf38 100644 --- a/tests/performance/ir_baselines.json +++ b/tests/performance/ir_baselines.json @@ -7,6 +7,7 @@ "file_set": 76035, "l1_hit": 2595, "l1_hit_backed": 83015, + "l1_hit_backed_auto": 91986, "l2_hit": 172539, "l2_hit_async_metrics": 171976, "minimal_l1_hit": 2595, @@ -31,6 +32,7 @@ "file_set": 74112, "l1_hit": 2487, "l1_hit_backed": 83882, + "l1_hit_backed_auto": 92950, "l2_hit": 167686, "l2_hit_async_metrics": 167728, "minimal_l1_hit": 2487, diff --git a/tests/performance/ir_budget.py b/tests/performance/ir_budget.py index 26540769..07c55fe8 100644 --- a/tests/performance/ir_budget.py +++ b/tests/performance/ir_budget.py @@ -105,6 +105,7 @@ PATHS = ( "l1_hit", "l1_hit_backed", + "l1_hit_backed_auto", "minimal_l1_hit", "l2_hit", "miss", diff --git a/tests/performance/ir_workload.py b/tests/performance/ir_workload.py index 0a660407..3697031d 100644 --- a/tests/performance/ir_workload.py +++ b/tests/performance/ir_workload.py @@ -114,10 +114,12 @@ def __init__(self, **kwargs: Any) -> None: # Every metrics path records through the same AsyncMetricsCollector.record_cache_operation, once # per call, so one batched-mode variant covers that mode; the L1-only hits record no metric. # With a backend, L1 holds serialized bytes and every hit decodes them, where an L1-only hit returns - # the stored object: after its first hit, this small value's entry is held as its verified MessagePack. + # the stored object: after its first hit, this small value's entry is held as its verified MessagePack, with + # the default serializer and with the auto one. decorated: dict[str, Callable[[], Callable[..., dict]]] = { "l1_hit": lambda: cache(backend=None, ttl=300)(body), "l1_hit_backed": lambda: cache(backend=DictBackend(), ttl=300)(body), + "l1_hit_backed_auto": lambda: cache(backend=DictBackend(), ttl=300, serializer="auto")(body), "minimal_l1_hit": lambda: cache.minimal(backend=None, ttl=300)(body), "l2_hit": l2, "miss": lambda: cache(backend=MissBackend(), l1_enabled=False, ttl=300)(body), diff --git a/tests/unit/test_ir_budget.py b/tests/unit/test_ir_budget.py index f857f68c..1e1749e1 100644 --- a/tests/unit/test_ir_budget.py +++ b/tests/unit/test_ir_budget.py @@ -446,13 +446,14 @@ def tracking_init(self, *args, **kwargs): assert direct == {} -def test_backed_l1_hit_path_is_a_plaintext_l1_hit_on_held_msgpack(monkeypatch) -> None: - """``l1_hit_backed`` budgets an L1 hit with a backend: after warmup, every call finds its key in L1, holding - the verified MessagePack, so it never reaches L2.""" +@pytest.mark.parametrize(("path", "owner"), [("l1_hit_backed", "StandardSerializer"), ("l1_hit_backed_auto", "AutoSerializer")]) +def test_backed_l1_hit_path_is_a_plaintext_l1_hit_on_held_msgpack(monkeypatch, path: str, owner: str) -> None: + """``l1_hit_backed`` and ``l1_hit_backed_auto`` budget an L1 hit with a backend: after warmup, every call finds + its key in L1, holding the verified MessagePack its serializer kept, so it never reaches L2.""" pytest.importorskip("pandas") # the workload builder makes the Arrow path's DataFrame too import cachekit from cachekit.l1_cache import L1Cache - from cachekit.serializers.standard_serializer import VerifiedMsgpack + from cachekit.serializers.base import VerifiedPayload entries: list[object] = [] real_get = L1Cache.get @@ -478,7 +479,7 @@ def counting_get(key): monkeypatch.setattr(L1Cache, "get", tracking_get) monkeypatch.setattr(cachekit, "cache", counting_cache) - op = ir_workload.build_workload("l1_hit_backed") + op = ir_workload.build_workload(path) for _ in range(3): # warmup: the miss stores the envelope, the first hit swaps in the MessagePack op() assert backend_reads, "the warmup miss reads the backend, so the count reaches it" @@ -487,7 +488,7 @@ def counting_get(key): for _ in range(20): assert op() == {"id": 42, "name": "Ada Lovelace", "roles": ["admin", "ops"], "score": 97.5, "tags": list(range(16))} assert len(entries) == 20 - assert all(type(entry) is VerifiedMsgpack for entry in entries) + assert all(type(entry) is VerifiedPayload and entry.owner.__name__ == owner for entry in entries) assert backend_reads == [] diff --git a/tests/unit/test_l1_verified_msgpack.py b/tests/unit/test_l1_verified_msgpack.py index 49c4842b..c4742b21 100644 --- a/tests/unit/test_l1_verified_msgpack.py +++ b/tests/unit/test_l1_verified_msgpack.py @@ -1,5 +1,7 @@ -"""A non-encrypted backed cache's L1 hit swaps its envelope for the verified MessagePack (``VerifiedMsgpack``) when -that is no longer; otherwise L1 keeps the envelope. +"""A non-encrypted backed cache's L1 hit swaps its envelope for the verified MessagePack (``VerifiedPayload``) when +that is no longer; otherwise L1 keeps the envelope. The auto serializer's MessagePack envelopes and the orjson +serializer's checksummed JSON are kept the same way (the last section); everything else here is the default +serializer's. The first L1 hit on an entry decodes the envelope as before, and its retrieve() yields the MessagePack; when that is no longer than the envelope, L1 then holds it in place of the envelope (``CacheSerializationHandler.deserialize_l1``, @@ -23,11 +25,12 @@ from __future__ import annotations import asyncio +import importlib.util import os import random import uuid from collections.abc import Callable, Iterator -from datetime import UTC, timedelta, timezone +from datetime import UTC, datetime, timedelta, timezone from typing import Any import msgpack @@ -42,9 +45,9 @@ from cachekit.config.nested import L1CacheConfig from cachekit.l1_cache import L1Cache, get_l1_cache from cachekit.serializers import base -from cachekit.serializers.base import SerializationError, SuspiciousCacheEntryError, unpackb_bounded +from cachekit.serializers.base import SerializationError, SuspiciousCacheEntryError, VerifiedPayload, unpackb_bounded from cachekit.serializers.encryption_wrapper import EncryptionWrapper -from cachekit.serializers.standard_serializer import VerifiedMsgpack, _standard_object_hook +from cachekit.serializers.standard_serializer import StandardSerializer, _standard_object_hook from cachekit.serializers.wrapper import SerializationWrapper from tests.fixtures.zero_config import clear_cachekit_env @@ -52,6 +55,13 @@ MASTER_KEY = "a" * 64 GUARD_ERROR_PREFIX = "Unpack failed: MessagePack document " # only the structure bounds raise this +_JSON_EXTRA = pytest.mark.skipif(importlib.util.find_spec("orjson") is None, reason="requires the [json] extra") + + +def _param(*values: Any, name: str) -> Any: + """``values`` as the parameter set ``name``; one that builds an orjson handler needs the [json] extra, which the + free-threaded CI lanes do not install, so it skips there.""" + return pytest.param(*values, id=name, marks=_JSON_EXTRA if "orjson" in name else ()) @pytest.fixture(autouse=True) @@ -123,7 +133,7 @@ def compute(x: int) -> Any: 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 type(entry) is VerifiedPayload assert entry.data == msgpack.packb(value, use_bin_type=True) assert len(entry) <= len(envelope) @@ -184,7 +194,7 @@ def compute(x: int) -> dict[str, int]: assert type(entry) is bytes and entry == envelope assert compute(1) == {"x": 1} _, entry = _stored(backend, namespace) - assert type(entry) is VerifiedMsgpack + assert type(entry) is VerifiedPayload assert backend.gets == 2 @@ -225,7 +235,7 @@ def compute(x: int) -> dict[str, int]: @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: +def test_deserialize_l1_keeps_verified_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") @@ -235,7 +245,7 @@ def test_deserialize_l1_keeps_msgpack_only_when_no_longer_and_it_decodes_the_sam assert entry is None assert handler.deserialize_l1(envelope, cache_key="k") == (expected, None) # every hit on it: the same again else: - assert type(entry) is VerifiedMsgpack and len(entry) <= len(envelope) + assert type(entry) is VerifiedPayload and len(entry) <= len(envelope) assert handler.deserialize_l1(entry, cache_key="k") == (expected, None) @@ -243,7 +253,7 @@ def test_msgpack_exactly_as_long_as_the_envelope_is_kept_and_one_byte_longer_is_ """Random bytes followed by k zero bytes: each zero lengthens the MessagePack by one byte, and LZ4 folds the run, so the MessagePack overtakes the envelope, passing through both lengths.""" handler, head = CacheSerializationHandler(), random.Random(0).randbytes(300) - for extra, kept in ((0, VerifiedMsgpack), (1, type(None))): + for extra, kept in ((0, VerifiedPayload), (1, type(None))): for k in range(2000): value = head + bytes(k) envelope = handler.serialize_data(value, cache_key="k") @@ -256,7 +266,14 @@ def test_msgpack_exactly_as_long_as_the_envelope_is_kept_and_one_byte_longer_is_ _CORRUPT = (SerializationError, "Corrupt cache envelope") _OTHER_HANDLERS: dict[str, tuple[Callable[[], CacheSerializationHandler], tuple[type[Exception], str]]] = { - "auto": (lambda: CacheSerializationHandler(serializer_name="auto"), _CORRUPT), + "auto-integrity-off": ( + lambda: CacheSerializationHandler(serializer_name="auto", enable_integrity_checking=False), + _CORRUPT, + ), + "orjson-integrity-off": ( + lambda: CacheSerializationHandler(serializer_name="orjson", enable_integrity_checking=False), + _CORRUPT, + ), "integrity-off": (lambda: CacheSerializationHandler(enable_integrity_checking=False), _CORRUPT), "encrypted": ( lambda: CacheSerializationHandler(encryption=True, master_key=MASTER_KEY, single_tenant_mode=True), @@ -265,12 +282,12 @@ def test_msgpack_exactly_as_long_as_the_envelope_is_kept_and_one_byte_longer_is_ } -def _planted() -> VerifiedMsgpack: +def _planted() -> VerifiedPayload: """The kept entry a plaintext default-serializer handler leaves in L1.""" - return VerifiedMsgpack(msgpack.packb({"x": 2})) + return VerifiedPayload(msgpack.packb({"x": 2}), StandardSerializer) -@pytest.mark.parametrize(("make", "refusal"), list(_OTHER_HANDLERS.values()), ids=list(_OTHER_HANDLERS)) +@pytest.mark.parametrize(("make", "refusal"), [_param(*v, name=k) for k, v in _OTHER_HANDLERS.items()]) def test_other_handlers_keep_their_envelope_and_refuse_a_kept_entry( make: Callable[[], CacheSerializationHandler], refusal: tuple[type[Exception], str] ) -> None: @@ -278,7 +295,7 @@ def test_other_handlers_keep_their_envelope_and_refuse_a_kept_entry( 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 not handler.keeps_verified_in_l1 assert handler.deserialize_l1(envelope, cache_key="k") == ({"x": 1}, None) error, message = refusal for decode in (handler.deserialize_l1, handler.deserialize_data): @@ -365,7 +382,7 @@ def recording(self: ByteStorage, data: Any) -> Any: return seen -_PLAIN = {"id": 7, "tags": ["a"]} # its MessagePack is no longer than its envelope: kept as VerifiedMsgpack +_PLAIN = {"id": 7, "tags": ["a"]} # its MessagePack is no longer than its envelope: kept as VerifiedPayload _COMPRESSIBLE = {"users": [{"name": f"user {i}", "theme": "dark"} for i in range(50)]} # L1 keeps its envelope @@ -392,7 +409,7 @@ def compute(x: int) -> Any: @pytest.mark.parametrize("mode", ["sync", "async"]) @pytest.mark.parametrize( ("value", "kept"), - [(_PLAIN, VerifiedMsgpack), (_COMPRESSIBLE, bytes)], + [(_PLAIN, VerifiedPayload), (_COMPRESSIBLE, bytes)], ids=["msgpack", "envelope"], ) def test_every_hit_runs_the_bounded_decode_and_only_kept_msgpack_skips_retrieve( @@ -412,7 +429,7 @@ def test_every_hit_runs_the_bounded_decode_and_only_kept_msgpack_skips_retrieve( retrieves.clear() for _ in range(2): assert _call(fn, mode) == value - assert (decodes, len(retrieves), backend.gets) == (["ok", "ok"], 0 if kept is VerifiedMsgpack else 2, 1) + assert (decodes, len(retrieves), backend.gets) == (["ok", "ok"], 0 if kept is VerifiedPayload else 2, 1) def test_a_nesting_bomb_put_in_l1_as_bytes_is_refused_by_the_bounded_decode(decodes: list[str]) -> None: @@ -518,9 +535,9 @@ def _outcome(decode: Callable[[], Any]) -> Any: return type(e) -def _verified(envelope: bytes) -> VerifiedMsgpack: +def _verified(envelope: bytes) -> VerifiedPayload: """``envelope``'s MessagePack as retrieve() verifies it, held as L1 holds a kept entry, whatever its length.""" - return VerifiedMsgpack(ByteStorage("msgpack").retrieve(SerializationWrapper.unwrap(envelope)[0])[0]) + return VerifiedPayload(ByteStorage("msgpack").retrieve(SerializationWrapper.unwrap(envelope)[0])[0], StandardSerializer) def _walked_hooked(envelope: bytes) -> Any: @@ -594,7 +611,7 @@ def test_shrink_if_held_swaps_the_entry_in_place(size: int) -> None: l1.put("b", b"b" * 10, redis_ttl=60) expires_at = l1._state.cache["a"].expires_at - new = VerifiedMsgpack(b"m" * size) + new = VerifiedPayload(b"m" * size, StandardSerializer) assert l1.shrink_if_held("a", old, new) assert l1.peek("a") is new @@ -610,13 +627,13 @@ def test_shrink_if_held_leaves_an_entry_it_did_not_read_or_would_grow(change: st l1 = L1Cache(max_memory_mb=1, namespace=_namespace()) old = b"e" * 100 l1.put("a", old, redis_ttl=60) - new = VerifiedMsgpack(b"m" * 40) + new = VerifiedPayload(b"m" * 40, StandardSerializer) 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) + new = VerifiedPayload(b"m" * 101, StandardSerializer) held = l1.peek("a") assert not l1.shrink_if_held("a", old, new) @@ -704,7 +721,7 @@ def compute(x: int) -> dict[str, int]: held = l1.peek(key) assert held is not None and len(planted) <= len(held) assert l1.shrink_if_held(key, held, planted) - assert type(l1.peek(key)) is VerifiedMsgpack + assert type(l1.peek(key)) is VerifiedPayload assert compute(1) == {"x": 1} # read from L2 instead assert type(l1.peek(key)) is bytes @@ -845,3 +862,209 @@ def test_an_entry_shrunk_under_the_admission_cap_stays_out_so_the_hot_set_keeps_ 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 + + +# ── auto and orjson: the same rules, their own payloads ─────────────────────────────────────── + + +def _auto_value() -> dict[str, Any]: + """Python types only the auto serializer's object hook restores, so a kept decode must run that hook.""" + return { + "pair": (1, "a"), + "set": {1, 2}, + "frozen": frozenset({"x"}), + "when": datetime(2026, 11, 1, 12, 30, tzinfo=UTC), + "id": uuid.UUID(int=7), + } + + +_KEEPING = {"auto": _auto_value, "orjson": lambda: {"id": 7, "tags": ["a"], "score": 97.5}} +_KEEPING_PARAMS = [_param(s, name=s) for s in _KEEPING] + + +def _verified_payload(serializer_name: str, envelope: bytes) -> bytes: + """What a verified hit on ``envelope`` keeps: the MessagePack retrieve() returns, or the JSON under the checksum.""" + payload = bytes(SerializationWrapper.unwrap(envelope)[0]) + return ByteStorage("msgpack").retrieve(payload)[0] if serializer_name == "auto" else payload[8:] + + +@pytest.fixture +def checks(monkeypatch: pytest.MonkeyPatch) -> list[str]: + """One entry per integrity check either serializer runs: ByteStorage.retrieve(), or orjson's xxHash3-64.""" + import xxhash + + seen: list[str] = [] + real_retrieve, real_digest = ByteStorage.retrieve, xxhash.xxh3_64_digest + + def retrieve(self: ByteStorage, data: Any) -> Any: + seen.append("retrieve") + return real_retrieve(self, data) + + def digest(data: Any) -> bytes: + seen.append("xxh3") + return real_digest(data) + + monkeypatch.setattr(ByteStorage, "retrieve", retrieve) + monkeypatch.setattr(xxhash, "xxh3_64_digest", digest) # the module both serializers call it through + return seen + + +@pytest.mark.parametrize("serializer_name", _KEEPING_PARAMS) +def test_auto_and_orjson_hits_keep_their_verified_payload_and_skip_the_check(serializer_name: str, checks: list[str]) -> None: + """The first hit verifies the envelope and keeps what it verified, owned by its serializer; later hits decode + the kept payload with no integrity check, through the same decoder, so they return equal fresh objects.""" + value, backend, namespace = _KEEPING[serializer_name](), _Backend(), _namespace() + + @cache(backend=backend, ttl=60, namespace=namespace, serializer=serializer_name) + def compute(x: int) -> Any: + return value + + assert compute(1) == value # miss: L1 holds the envelope + envelope, entry = _stored(backend, namespace) + assert type(entry) is bytes + checks.clear() + first = compute(1) # first hit: verifies, decodes, keeps + assert first == value and len(checks) == 1 + _, entry = _stored(backend, namespace) + assert type(entry) is VerifiedPayload + assert entry.owner is type(CacheSerializationHandler(serializer_name=serializer_name)._base_serializer) + assert entry.data == _verified_payload(serializer_name, envelope) + assert len(entry) <= len(envelope) + + checks.clear() + second = compute(1) + assert second == value and second is not first + assert (checks, backend.gets) == ([], 1) + + +@pytest.mark.parametrize("serializer_name", _KEEPING_PARAMS) +def test_a_corrupt_auto_or_orjson_entry_in_l1_is_never_kept(serializer_name: str) -> None: + """A payload enters L1 only from a check that verified it: a corrupt envelope in L1 is refused, never kept.""" + handler = CacheSerializationHandler(serializer_name=serializer_name) + envelope = handler.serialize_data(_KEEPING[serializer_name](), cache_key="k") + with pytest.raises(SerializationError): + handler.deserialize_l1(_corrupt(envelope), cache_key="k") + + +@pytest.mark.parametrize( + ("reader", "minter"), + [ + _param("auto", StandardSerializer, name="auto-StandardSerializer"), + _param("orjson", StandardSerializer, name="orjson-StandardSerializer"), + _param("default", "auto", name="default-auto"), + _param("auto", "orjson", name="auto-orjson"), + ], +) +def test_a_keeping_handler_refuses_another_serializers_kept_entry(reader: str, minter: Any) -> None: + """The owner is the mark of which decoder may read a kept payload: a kept entry under a key another serializer's + handler reads is refused as the corrupt frame it is to that handler, never decoded by the wrong decoder.""" + if isinstance(minter, str): + other = CacheSerializationHandler(serializer_name=minter) + minter = type(other._base_serializer) + planted = VerifiedPayload(msgpack.packb({"x": 2}), minter) + handler = CacheSerializationHandler(serializer_name=reader) + assert handler.keeps_verified_in_l1 + for decode in (handler.deserialize_l1, handler.deserialize_data): + with pytest.raises(_CORRUPT[0], match=_CORRUPT[1]): + decode(planted, cache_key="k") # type: ignore[arg-type] + + +def _keeping_frames(serializer_name: str) -> dict[str, bytes]: + """Entries an auto or orjson handler must decode exactly as deserialize_data does; only "own-envelope" is kept.""" + k = "k" + handler = CacheSerializationHandler(serializer_name=serializer_name) + own = handler.serialize_data(_KEEPING[serializer_name](), cache_key=k) + frames = { + "own-envelope": own, + "other-serializer": CacheSerializationHandler().serialize_data({"x": 1}, cache_key=k), + "integrity-off": CacheSerializationHandler( + serializer_name=serializer_name, enable_integrity_checking=False + ).serialize_data({"x": 1}, cache_key=k), + "corrupt-payload": _corrupt(own), + "truncated": own[:-30], + "not-a-frame": msgpack.packb({"x": 1}), + } + if serializer_name == "orjson": + frames["encrypted-drift"] = CacheSerializationHandler( + serializer_name="orjson", encryption=True, master_key=MASTER_KEY, single_tenant_mode=True + ).serialize_data({"x": 1}, cache_key=k) + return frames + np = pytest.importorskip("numpy") + pd = pytest.importorskip("pandas") + payload, metadata, name = SerializationWrapper.unwrap(own) + frames |= { + "numpy": handler.serialize_data(np.arange(12.0).reshape(3, 4), cache_key=k), + "dataframe-arrow": handler.serialize_data(pd.DataFrame({"a": [1, 2]}), cache_key=k), + "series": handler.serialize_data(pd.Series([1.5, 2.5], name="s"), cache_key=k), + "header-claims-series": SerializationWrapper.wrap(bytes(payload), metadata | {"original_type": "series"}, name), + "header-claims-none": SerializationWrapper.wrap( + bytes(payload), {key: v for key, v in metadata.items() if key != "original_type"}, name + ), + } + return frames + + +_KEEPING_FRAME_NAMES = { + "auto": ( + "own-envelope", + "other-serializer", + "integrity-off", + "corrupt-payload", + "truncated", + "not-a-frame", + "numpy", + "dataframe-arrow", + "series", + "header-claims-series", + "header-claims-none", + ), + "orjson": ( + "own-envelope", + "other-serializer", + "integrity-off", + "corrupt-payload", + "truncated", + "not-a-frame", + "encrypted-drift", + ), +} + + +@pytest.mark.parametrize("serializer_name", _KEEPING_PARAMS) +def test_keeping_frame_names_list_every_frame(serializer_name: str) -> None: + assert set(_keeping_frames(serializer_name)) == set(_KEEPING_FRAME_NAMES[serializer_name]) + + +def _same(a: Any, b: Any) -> bool: + """Equality that also compares NumPy arrays and pandas objects (whose == is elementwise).""" + if hasattr(a, "equals"): + return type(a) is type(b) and a.equals(b) + if hasattr(a, "shape") and hasattr(a, "dtype"): + return type(a) is type(b) and a.dtype == b.dtype and a.shape == b.shape and bool((a == b).all()) + return a == b + + +@pytest.mark.parametrize( + ("serializer_name", "name"), + [_param(s, n, name=f"{s}-{n}") for s, names in _KEEPING_FRAME_NAMES.items() for n in names], +) +def test_auto_and_orjson_deserialize_l1_decodes_or_refuses_every_entry_as_deserialize_data_does( + serializer_name: str, name: str, monkeypatch: pytest.MonkeyPatch +) -> None: + """The keeping path runs only the route deserialize takes for its own plain envelope: NumPy, Arrow, columnar + and every other entry go to deserialize_data, so the two agree on every entry and only that envelope is kept.""" + data = _keeping_frames(serializer_name)[name] + monkeypatch.setenv("CACHEKIT_MASTER_KEY", MASTER_KEY) # lets the encrypted-drift entry decrypt, as it would + reset_settings() + handler = CacheSerializationHandler(serializer_name=serializer_name, 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 _same(got[0] if isinstance(got, tuple) else got, expected) + assert (isinstance(got, tuple) and got[1] is not None) == (name == "own-envelope") diff --git a/tests/unit/test_single_flight.py b/tests/unit/test_single_flight.py index 07001d95..b1ebf815 100644 --- a/tests/unit/test_single_flight.py +++ b/tests/unit/test_single_flight.py @@ -42,7 +42,7 @@ 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 +from cachekit.serializers.base import VerifiedPayload from tests.utils.clock import shared_clock pytestmark = pytest.mark.unit @@ -1001,7 +1001,7 @@ def run_after_another_trip_and_hit(self: ThreadFlights, *args: Any) -> Any: 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 + assert decoded == [bytes, VerifiedPayload] # 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(