diff --git a/.secrets.baseline b/.secrets.baseline index 03f09022..7deb5010 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": 647 + "line_number": 699 } ], "src/cachekit/serializers/interop_serializer.py": [ @@ -2020,5 +2020,5 @@ } ] }, - "generated_at": "2026-10-19T08:46:47Z" + "generated_at": "2026-10-29T06:52:18Z" } diff --git a/docs/backends/redis.md b/docs/backends/redis.md index 44dc4571..702a8c3a 100644 --- a/docs/backends/redis.md +++ b/docs/backends/redis.md @@ -82,7 +82,7 @@ Unset, a GIL build keeps hiredis, and so does a free-threaded build whose GIL is ## Characteristics -- Network latency: one round trip per operation, which depends on your network and region ([not measured here](../performance.md#not-measured-here)) +- Network latency: one round trip per operation, which depends on your network and region ([not measured here](../performance.md#not-measured-here)). On a backend that keeps a [key registry](../features/l1-invalidation.md#whole-function-invalidation) (resolved from the URL or `RedisBackendProvider`), a decorated write's registry entry rides in the same pipeline as its `SET`, so the write is still one round trip - Automatic TTL support (Redis `SET … EX`) - Connection pooling built-in - Supports large values (up to Redis limits) diff --git a/docs/features/l1-invalidation.md b/docs/features/l1-invalidation.md index 4c9a1651..0b4ac9fa 100644 --- a/docs/features/l1-invalidation.md +++ b/docs/features/l1-invalidation.md @@ -243,14 +243,14 @@ On a generated key with a non-default serializer, each entry this process wrote **Round trips on the "this process" backends.** A plain `RedisBackend` deletes the process's keys with one `UNLINK` per 10 000 keys. Memcached groups them by server and sends one pipelined, acknowledged `delete` batch per server per 1 000 keys. File deletes one key per call. CachekitIO has no bulk delete, so it sends one `DELETE` per key, up to 16 at a time: 100 keys take about 7 round trips instead of 100. Each `DELETE` still counts against your tier's rate limit, and 16 at a time spends it much faster than one at a time. So once a `DELETE` is rate limited (a 429 with `Retry-After`), cachekit starts no more concurrent deletes: it sends the remaining keys one at a time, waiting out each `Retry-After`, for at most about as long as one-at-a-time deletes would have taken. Keys it could not delete in that time, keys refused with a 429 that has no `Retry-After` (a quota or balance deny), and keys whose `DELETE` failed for any other reason stay tracked for the next `invalidate_cache()`. When the time runs out, cachekit logs a WARNING `CachekitIO rate limit: stopped pacing invalidation deletes` with the number of keys left. If an `UNLINK` fails, cachekit logs a WARNING `Multi-key L2 delete failed` and retries that batch one key at a time, so one failed batch does not stop the rest of the invalidation. If a Memcached send fails, or its server is in the client's retry window, its keys count as failed and stay tracked for the next `invalidate_cache()`. -**Key registry (tenant-scoped Redis).** Every L2 write also adds the key to a server-side set for the function, `t:{tenant}:ck:reg:{namespace}:{hash}`. No-args invalidation drains that set and deletes every key in it, plus any key this process remembers that the set missed. The drain runs in bounded steps of 10 000 keys, so it never blocks Redis for long, and a write that lands during the drain is either deleted or left for the next drain: in the set, or, if its tracking failed, in the writing process's own record of its keys. It adds one pipelined round-trip (`SADD` + `EXPIRE`) to each L2 write — cache hits pay nothing. If that round-trip fails, the write still succeeds and cachekit logs a WARNING `Key tracking failed`: that key is then deleted only by this process's next drain, and other processes' drains miss it until its TTL. The warning fires at most once a minute per function in each process and carries the count of failures since the last one, so a registry outage doesn't flood the logs. +**Key registry (tenant-scoped Redis).** Every L2 write also adds the key to a server-side set for the function, `t:{tenant}:ck:reg:{namespace}:{hash}`. No-args invalidation drains that set and deletes every key in it, plus any key this process remembers that the set missed. The drain runs in bounded steps of 10 000 keys, so it never blocks Redis for long, and a write that lands during the drain is either deleted or left for the next drain: in the set, or, if its tracking failed, in the writing process's own record of its keys. It adds no round trip: the `SADD` + `EXPIRE` go to Redis in the same pipeline as the write's `SET`, and cache hits pay nothing. If those two commands fail while the `SET` succeeds, the write still succeeds and cachekit logs a WARNING `Key tracking failed`: that key is then deleted only by this process's next drain, and other processes' drains miss it until its TTL. The warning fires at most once a minute per function in each process and carries the count of failures since the last one, so a registry outage doesn't flood the logs. If the `SET` fails, its key can still land in the set; the next drain deletes the missing key without error, and the set's expiry bounds it until then. Things to know: - **Server requirements.** A single-instance or primary/replica Redis **5.0 or newer**. Redis Cluster and sharding proxies (one endpoint in front of several shards) are not supported. A restricted ACL user needs the `@scripting` category. When the drain fails for any of these reasons, cachekit logs a WARNING and falls back to deleting only this process's keys. A drain removes a key from the set only after deleting it, so tracked keys a failed drain did not reach stay in the set for the next drain, unless the set expires first. - **Other processes' L1.** The registry cleans L2 only. Other processes keep serving their L1 copies until the L1 TTL, as with single-key invalidation, unless they run the [invalidation listener](#cross-process-l1-eviction). - **Invalidation announcements.** Once an invalidation's L2 change has succeeded, cachekit publishes one message on the Redis pub/sub channel `cachekit:py:invalidate:v1`, which processes running the [invalidation listener](#cross-process-l1-eviction) act on: after a no-args drain returns, and after an `invalidate_cache(args)` whose delete of the key returned. A failed delete, or a failed drain that falls back to this process's own keys, publishes nothing. The message carries the function's registry id and, for `invalidate_cache(args)`, the invalidated key. A custom `key=` function's key is never sent because it can embed caller identifiers; its message names only the function. Anyone allowed to `SUBSCRIBE` to the channel sees a live feed of which functions are invalidated, with their generated keys. A generated key spells out the function and carries an unkeyed hash of the arguments, so arguments that are easy to guess, such as small integer ids, can be recovered by trying candidates (see [Invalidation Channel](../../SECURITY.md#invalidation-channel-redis-pubsub)). Redis delivers pub/sub messages whatever a client's database number, so grant `subscribe` on the channel only to your application's own Redis users. A restricted ACL user needs the `publish` command and access to the channel (`&cachekit:py:invalidate:v1`). Without them the invalidation still completes, and cachekit logs a WARNING `Invalidation announcement failed` at most once a minute per process, with the count of failures since the last one; each failure in between logs at DEBUG. -- **Set lifetime.** A tracking set expires 7 days after the last tracked write to its function; each write whose tracking round-trip succeeds refreshes that expiry. While the set exists, a key leaves it only when a no-args drain deletes it: the key's own expiry and `invalidate_cache(args)` do not remove it. A zero-parameter function has no drain — its `invalidate_cache()` deletes its one key directly and leaves that key's member in the set until the set expires. A parameterised function that is written continuously and never invalidated with no args therefore grows its set by one member per distinct key. Keys that outlive it — `ttl=None` or a TTL above 7 days — are no longer reachable by a drain from a process that never saw them. Keys written before the upgrade to v0.20.0 sit in no set: a drain deletes them only in an upgraded process that wrote or read them since it started, and otherwise they stay until their TTL, or for good with `ttl=None`. Call `invalidate_cache()` before you decommission a function whose entries have no TTL or a TTL above 7 days. +- **Set lifetime.** A tracking set expires 7 days after the last tracked write to its function; each write whose tracking succeeds refreshes that expiry. While the set exists, a key leaves it only when a no-args drain deletes it: the key's own expiry and `invalidate_cache(args)` do not remove it. A zero-parameter function has no drain — its `invalidate_cache()` deletes its one key directly and leaves that key's member in the set until the set expires. A parameterised function that is written continuously and never invalidated with no args therefore grows its set by one member per distinct key. Keys that outlive it — `ttl=None` or a TTL above 7 days — are no longer reachable by a drain from a process that never saw them. Keys written before the upgrade to v0.20.0 sit in no set: a drain deletes them only in an upgraded process that wrote or read them since it started, and otherwise they stay until their TTL, or for good with `ttl=None`. Call `invalidate_cache()` before you decommission a function whose entries have no TTL or a TTL above 7 days. - **Tenants.** The set is tenant-scoped like every other key. Each write is tracked in the set of the tenant in `tenant_context` for that call, and a drain empties only the calling tenant's set and can only delete keys inside that tenant's prefix — `default` when no tenant is set (see **Tenant scope** below). The INFO line `Key registry drained N keys` shows how many keys a drain deleted. An operator shell or one-off script in a multi-tenant deployment must enter the tenant's `tenant_context` before it invalidates: without it the drain empties the `default` tenant's set, and `Key registry drained 0 keys` is the usual sign. - **Reserved namespace.** `namespace="ck"` and any namespace starting with `ck:` are rejected at decoration: a key written there could overwrite a tracking set. - **Same module path everywhere.** The set is named by the function's `module.qualname`, so every process must import the function from the same module path. diff --git a/src/cachekit/backends/redis/provider.py b/src/cachekit/backends/redis/provider.py index c2784772..bfd69e74 100644 --- a/src/cachekit/backends/redis/provider.py +++ b/src/cachekit/backends/redis/provider.py @@ -50,6 +50,17 @@ # Key registry (KeyTrackableBackend). A tracking set lives 7 days past its last write. _TRACKING_SET_TTL_SECONDS = 604_800 + + +def _registry_add(pipeline: Any, scoped_reg: str, key: str) -> Any: + """Queue a key's registry entry on ``pipeline``: ``SADD`` the raw key, then ``EXPIRE`` the set to 7 days. + + The one place the registry's write commands live: ``track_key`` and ``_set_tracked`` both queue them here. + Queuing sends nothing; the caller chains ``execute()`` on the result. + """ + return pipeline.sadd(scoped_reg, key).expire(scoped_reg, _TRACKING_SET_TTL_SECONDS) + + # Members popped (and keys unlinked) per script call. Bounds each atomic step well under # lua-time-limit: a script that has written cannot be SCRIPT KILLed, so unbounded work in # one call can leave every other client on this Redis answered with BUSY. @@ -706,11 +717,8 @@ def track_key(self, registry_id: str, key: str) -> None: scoped_reg = self._scoped_key(registry_id) with _ClearedOnInterrupt(): try: - # Chained, never bound to a local: a pipeline's repr lists the password. A pipelined - # command returns its pipeline; redis-py's annotations give the reply type instead. - self._client.pipeline(transaction=False).sadd(scoped_reg, key).expire( # pyright: ignore[reportAttributeAccessIssue] - scoped_reg, _TRACKING_SET_TTL_SECONDS - ).execute() + # Chained, never bound to a local: a pipeline's repr lists the password. + _registry_add(self._client.pipeline(transaction=False), scoped_reg, key).execute() return except Exception as exc: error = classify_redis_error(exc, operation="track_key", key=registry_id) @@ -719,6 +727,61 @@ def track_key(self, registry_id: str, key: str) -> None: finally: del error + def _set_tracked(self, key: str, value: bytes, ttl: Optional[int], registry_id: str) -> Optional[BackendError]: + """``set`` and ``track_key`` in one round trip: ``SET``, ``SADD`` and ``EXPIRE`` on one + non-transactional pipeline. + + Private: ``StandardCacheHandler.set_tracked`` finds it on the backend class, and a backend without it + keeps the separate ``track_key`` call. The write's success is the ``SET`` reply alone. The + registry commands run even when the ``SET`` fails (no transaction rolls a failed ``SET`` back), + which leaves a member with no key: the next drain unlinks the missing key, and the set's 7-day + expiry bounds it until then. + + Args: + key: Cache key to store (will be tenant-scoped), as for ``set`` + value: Bytes value to store + ttl: Time-to-live in seconds (None or <= 0 = no expiry), as for ``set`` + registry_id: Unscoped registry id, as for ``track_key`` + + Returns: + None when the write and its tracking both succeeded; the classified tracking error when + the ``SET`` succeeded and ``SADD`` or ``EXPIRE`` did not. Tracking is advisory, so that + error is the caller's to report, never to raise. + + Raises: + BackendError: If the ``SET`` failed, or the pipeline did (a dropped connection: whether + the ``SET`` landed is unknown, as for ``set``) + """ + scoped_key = self._scoped_key(key) + scoped_reg = self._scoped_key(registry_id) + with _ClearedOnInterrupt(): + try: + # Chained, never bound to a local: a pipeline's repr lists the password. Error replies come + # back in place, so each command's outcome is read on its own. + replies = _registry_add( + self._client.pipeline(transaction=False).set( + scoped_key, value, ex=ttl if ttl is not None and ttl > 0 else None + ), + scoped_reg, + key, + ).execute(raise_on_error=False) + except Exception as exc: + error = classify_redis_error(exc, operation="set", key=key) + else: + if not isinstance(replies[0], Exception): + failed = next((reply for reply in replies[1:] if isinstance(reply, Exception)), None) + return None if failed is None else classify_redis_error(failed, operation="track_key", key=registry_id) + error = classify_redis_error(replies[0], operation="set", key=key) + replies = None # its raw redis-py error replies must not stay in a frame on the raised error's traceback + try: + raise error from error.original_exception + finally: + del error + + # The set and track_key that _set_tracked stands in for, captured at class creation; see + # cache_handler._fuses_tracked_write. + _set_tracked_replaces = {"set": set, "track_key": track_key} + def drain_tracked(self, registry_id: str, local_keys: Iterable[str]) -> set[str]: """Delete every tracked key of a registry, then every local key the drain missed (KeyTrackableBackend protocol). diff --git a/src/cachekit/cache_handler.py b/src/cachekit/cache_handler.py index 561a3941..d9aa0965 100644 --- a/src/cachekit/cache_handler.py +++ b/src/cachekit/cache_handler.py @@ -367,6 +367,58 @@ def supports_key_tracking(backend: object) -> TypeGuard[KeyTrackableBackend]: return callable(getattr(cls, "track_key", None)) and callable(getattr(cls, "drain_tracked", None)) +def _fuses_tracked_write(backend: object) -> bool: + """Whether the backend writes a value and its registry entry in one call: a private ``_set_tracked``, + found on the CLASS as ``supports_key_tracking`` finds ``track_key``. + + Only while the instance resolves ``set`` and ``track_key`` to the functions in the owner's + ``_set_tracked_replaces``, as plain methods bound to the backend itself. An override, patch, wrapper or + delegation anywhere (a subclass, the class itself, the instance, ``__getattribute__``, a proxy) keeps + both calls, so it still runs on every write. An SWR-capable backend keeps ``set`` too: the fused write + has no slot for a stale-grace window. + """ + owner = next((klass for klass in type(backend).__mro__ if "_set_tracked" in vars(klass)), None) + if owner is None or not supports_key_tracking(backend) or supports_swr(backend): # pyright: ignore[reportArgumentType] + return False + replaces = vars(owner).get("_set_tracked_replaces", {}) + for name in ("set", "track_key"): + func = replaces.get(name) # None when the owner records nothing for it: nothing to match + bound = getattr(backend, name, None) + if func is None or type(bound) is not types.MethodType: # a wrapper or proxy (wrapt, APM) must run: not fused + return False + if bound.__func__ is not func or bound.__self__ is not backend: # another function, or another backend's + return False + return True + + +def _set_l2( + handler: Any, + key: str, + value: bytes, + ttl: int | None, + stale_ttl: int | None, + registry_id: str | None, + on_track_error: Callable[[str, Exception], None], +) -> tuple[bool, bool]: + """A buffered L2 write through ``handler``: (stored, tracked). + + With a ``registry_id``, the key-registry write rides in the same round trip where the handler can + (``set_tracked``), and tracked is True: its failure has gone to ``on_track_error``. Otherwise tracked is + False and the caller tracks a stored key itself. getattr: custom CacheHandlerStrategy impls predate + set_tracked, as they predate set_streaming. + """ + set_tracked = None if registry_id is None else getattr(handler, "set_tracked", None) + if set_tracked is not None: + stored = set_tracked(key, value, ttl, registry_id, on_track_error) + if stored is not None: + return stored, True + # Only thread the SWR kwarg when set: strategy implementations without + # **metadata (tests, custom handlers) must keep working unchanged. + if stale_ttl is not None: + return bool(handler.set(key, value, ttl, stale_ttl=stale_ttl)), False + return bool(handler.set(key, value, ttl)), False + + class _MultiDeleteBackend(Protocol): """Internal: a backend that deletes many keys in one multi-key call (the backend may split it into several sends; Memcached sends per server, 1,000 keys a send). @@ -1565,11 +1617,14 @@ class StoreOutcome(NamedTuple): envelope cannot say: a buffered write whose backend set failed still has an envelope, and a successful stream has none. Key tracking follows stored, L1 follows envelope. deferred says the backend write was handed to a write-behind writer (stored is then False: the writer tracks it). + tracked says the key-registry write rode with the backend write (``set_tracked``), its failure + already reported: the caller must not track the key again. """ envelope: Optional[bytes] stored: bool deferred: bool = False + tracked: bool = False class CacheOperationHandler: @@ -1993,6 +2048,8 @@ def store_result( kwargs: dict[str, Any] | None = None, stale_ttl: int | None = None, defer: Callable[[bytes], bool] | None = None, + registry_id: str | None = None, + on_track_error: Callable[[str, Exception], None] = lambda _key, _error: None, ) -> StoreOutcome: """Store result in backend cache with optional tenant context for encryption. @@ -2006,6 +2063,10 @@ def store_result( the backend write; True means it took the write (StoreOutcome.deferred), False that it refused and the write runs here. Serialization stays here either way, so its errors reach the caller as without it; the streaming path never defers. + registry_id: Key registry to record a stored key in, in the same round trip as the buffered + write when the handler can (``set_tracked``; StoreOutcome.tracked). None writes only. + on_track_error: Called with the key and the error when that registry write fails; the write + still counts as stored. Returns: StoreOutcome. envelope: serialized bytes when the buffered path produced them @@ -2048,17 +2109,14 @@ def store_result( serialized_data = self.serialization_handler.serialize_data(result, args, kwargs, cache_key) if defer is not None and defer(serialized_data): return StoreOutcome(envelope=serialized_data, stored=False, deferred=True) - # Only thread the SWR kwarg when set: strategy implementations without - # **metadata (tests, custom handlers) must keep working unchanged. - if stale_ttl is not None: - stored = self._cache_handler.set(cache_key, serialized_data, ttl, stale_ttl=stale_ttl) - else: - stored = self._cache_handler.set(cache_key, serialized_data, ttl) + stored, tracked = _set_l2( + self._cache_handler, cache_key, serialized_data, ttl, stale_ttl, registry_id, on_track_error + ) if stored: get_logger().cache_stored(cache_key, ttl) # The envelope goes to L1 even when the backend write failed (L1 serves it only within the per-entry share) - return StoreOutcome(envelope=serialized_data, stored=bool(stored)) + return StoreOutcome(envelope=serialized_data, stored=bool(stored), tracked=tracked) except (InteropError, KeyringConfigurationError): # Interop/v1 data-model rejection (spec-mandated; matches cachekit-ts), or # a LOCAL keyring config fault, which a cold key first hits here, on the @@ -2431,6 +2489,36 @@ def set( get_logger().error(f"Unexpected error setting key {redact_cache_key(key)}: {redact_error_for_log(e)}") return False + def set_tracked( + self, + key: str, + value: bytes, + ttl: Optional[int], + registry_id: str, + on_track_error: Callable[[str, Exception], None], + ) -> Optional[bool]: + """``set`` plus the key-registry write (``KeyTrackableBackend.track_key``), in one backend round trip. + + Returns: + None when the backend cannot fuse the two: the caller falls back to ``set``, then its own + ``track_key``, as for ``set_streaming``. Otherwise what ``set`` returns: whether the value + was stored. A registry write that fails after a stored value goes to ``on_track_error`` with + the key, and the value still counts as stored. + """ + if not _fuses_tracked_write(self.backend): + return None + try: + track_error = self._with_backpressure_and_timeout(self.backend._set_tracked, key, value, ttl, registry_id) # type: ignore[attr-defined] + except BackendError as e: + get_logger().error(f"Backend error setting key {redact_cache_key(key)}: {redact_error_for_log(e)}") + return False + except Exception as e: + get_logger().error(f"Unexpected error setting key {redact_cache_key(key)}: {redact_error_for_log(e)}") + return False + if track_error is not None: + on_track_error(key, track_error) + return True + def set_streaming(self, key: str, write_payload: Callable[[BinaryIO], None], ttl: Optional[int] = None) -> Optional[bool]: """Stream a value to a streaming-writable backend. @@ -2554,6 +2642,35 @@ async def set_async( get_logger().error(f"Unexpected error setting key {redact_cache_key(key)}: {redact_error_for_log(e)}") return False + async def set_tracked_async( + self, + key: str, + value: bytes, + ttl: Optional[int], + registry_id: str, + on_track_error: Callable[[str, Exception], None], + ) -> Optional[bool]: + """Async variant of :meth:`set_tracked`: the write and its registry entry in one thread-pool hop.""" + if not _fuses_tracked_write(self.backend): + return None + try: + track_error = await self._with_backpressure_and_timeout_async( + self.backend._set_tracked, # type: ignore[attr-defined] + key, + value, + ttl, + registry_id, + ) + except BackendError as e: + get_logger().error(f"Backend error setting key {redact_cache_key(key)}: {redact_error_for_log(e)}") + return False + except Exception as e: + get_logger().error(f"Unexpected error setting key {redact_cache_key(key)}: {redact_error_for_log(e)}") + return False + if track_error is not None: + on_track_error(key, track_error) + return True + async def delete_async(self, key: str) -> bool: """Delete key from cache asynchronously using backend. diff --git a/src/cachekit/decorators/wrapper.py b/src/cachekit/decorators/wrapper.py index d8056f64..52894c2f 100644 --- a/src/cachekit/decorators/wrapper.py +++ b/src/cachekit/decorators/wrapper.py @@ -28,6 +28,7 @@ L2MissProbe, StandardCacheHandler, TenantResolutionError, + _set_l2, _supports_multi_delete, get_backend_provider, get_logger, @@ -1302,24 +1303,41 @@ def _track_and_record(cache_key: str) -> None: try: _backend.track_key(_registry_id, cache_key) # type: ignore[union-attr] except Exception as e: - failures = _track_warn.claim() - if not failures: - _logger.debug("Key tracking failed for %s: %s", redact_cache_key(cache_key), redact_error_for_log(e)) - return - _logger.warning( - "Key tracking failed in registry %s (failures since the last warning: %d); other processes' " - "drains miss those keys until their TTL. Latest key %s: %s", - redact_cache_key(_registry_id), - failures, - redact_cache_key(cache_key), - redact_error_for_log(e), - ) + _warn_track_failure(cache_key, e) + + def _warn_track_failure(cache_key: str, e: Exception) -> None: + """Report a failed key-registry write: _track_and_record's, or one fused with the L2 write (set_tracked).""" + failures = _track_warn.claim() + if not failures: + _logger.debug("Key tracking failed for %s: %s", redact_cache_key(cache_key), redact_error_for_log(e)) + return + _logger.warning( + "Key tracking failed in registry %s (failures since the last warning: %d); other processes' " + "drains miss those keys until their TTL. Latest key %s: %s", + redact_cache_key(_registry_id), + failures, + redact_cache_key(cache_key), + redact_error_for_log(e), + ) async def _track_and_record_async(cache_key: str) -> None: """_track_and_record off the event loop, skipping the thread hop when nothing tracks.""" if _is_trackable(): await asyncio.to_thread(_track_and_record, cache_key) + async def _set_l2_async(cache_key: str, serialized_data: bytes) -> tuple[bool, bool]: + """An async miss's L2 write: (stored, tracked). Where the handler can (set_tracked_async), the + key-registry write rides in the same round trip and thread hop, and tracked is True; otherwise the + caller tracks a stored key itself (_track_and_record_async).""" + handler = operation_handler.cache_handler + # Asked first: a backend without a registry never pays for the probe. Custom strategies predate set_tracked_async. + set_tracked_async = getattr(handler, "set_tracked_async", None) if _is_trackable() else None + if set_tracked_async is not None: + stored = await set_tracked_async(cache_key, serialized_data, ttl, _registry_id, _warn_track_failure) + if stored is not None: + return stored, True + return await handler.set_async(cache_key, serialized_data, ttl=ttl, stale_ttl=_stale_ttl), False # type: ignore[union-attr] + def _l1_backfill_ttl(fresh_for: int | None) -> Any: """L1 TTL for a backfill from an L2 read, bounded by the server's remaining freshness (spec/saas-api.md#remaining-freshness). @@ -1822,15 +1840,15 @@ def _send_deferred(cache_key: str, envelope: bytes, twin: str | None, handler: C the backend write alone, the caller having serialized. """ start = time.perf_counter() - if _stale_ttl is not None: - stored = handler.set(cache_key, envelope, ttl, stale_ttl=_stale_ttl) - else: - stored = handler.set(cache_key, envelope, ttl) + # Trackable first: a backend without a registry never pays for set_tracked's probe. + registry_id = _registry_id if _is_trackable() else None + stored, tracked = _set_l2(handler, cache_key, envelope, ttl, _stale_ttl, registry_id, _warn_track_failure) duration_ms = (time.perf_counter() - start) * 1000 if stored: get_logger().cache_stored(cache_key, ttl) _record(cache_key, twin) - _track_and_record(cache_key) + if not tracked: + _track_and_record(cache_key) else: _warn_deferred_write(cache_key, "the backend refused or failed the write (logged above)") if features.collect_stats: @@ -2526,6 +2544,8 @@ def _backed_miss_sync( # With write-behind, store_result hands the serialized bytes to the writer instead of writing # them, and the L1 put below follows the hand-off as it follows a write: an invalidation # landing between the two leaves the same state either way. + # Asked once, before the write: a backend without a registry never pays for set_tracked's probe. + trackable = _is_trackable() outcome = operation_handler.store_result( cache_key, result, @@ -2534,13 +2554,15 @@ def _backed_miss_sync( kwargs, stale_ttl=_stale_ttl, defer=None if _write_behind is None else functools.partial(_defer_write, cache_key, twin_key), + registry_id=_registry_id if trackable else None, + on_track_error=_warn_track_failure, ) envelope = outcome.envelope set_duration_ms = (time.perf_counter() - set_start) * 1000 # Also store in L1 cache for fast subsequent access (using serialized bytes) _put_l1(cache_key, outcome.envelope, ttl, twin=twin_key) - if outcome.stored: + if outcome.stored and trackable and not outcome.tracked: _track_and_record(cache_key) # Record successful cache set @@ -3088,17 +3110,12 @@ async def _backed_miss_async( ) # Store in Redis with TTL - stored = await operation_handler.cache_handler.set_async( # type: ignore[attr-defined] - cache_key, - serialized_data, - ttl=ttl, - stale_ttl=_stale_ttl, - ) + stored, tracked = await _set_l2_async(cache_key, serialized_data) set_duration_ms = (time.perf_counter() - set_start) * 1000 # Also store in L1 cache for fast subsequent access (using serialized bytes) _put_l1(cache_key, serialized_data, ttl, twin=twin_key) - if stored: + if stored and not tracked: await _track_and_record_async(cache_key) # Record successful cache set @@ -3212,19 +3229,14 @@ async def _backed_miss_async( # Write-behind (unlocked path only: the locked path's write must land inside its lock, or # waiters re-read before L2 holds the value and recompute). A refusal writes here, as without it. deferred = _write_behind is not None and _defer_write(cache_key, twin_key, serialized_data) - stored = False # a deferred write is not stored yet: the writer tracks it once it lands + stored = tracked = False # a deferred write is not stored yet: the writer tracks it once it lands if not deferred: - stored = await operation_handler.cache_handler.set_async( # type: ignore[attr-defined] - cache_key, - serialized_data, - ttl=ttl, - stale_ttl=_stale_ttl, - ) + stored, tracked = await _set_l2_async(cache_key, serialized_data) set_duration_ms = (time.perf_counter() - set_start) * 1000 # Also store in L1 cache for fast subsequent access (using serialized bytes) _put_l1(cache_key, serialized_data, ttl, twin=twin_key) - if stored: + if stored and not tracked: await _track_and_record_async(cache_key) # Record successful cache set diff --git a/tests/integration/test_tracked_write_redis.py b/tests/integration/test_tracked_write_redis.py new file mode 100644 index 00000000..70a1b1b2 --- /dev/null +++ b/tests/integration/test_tracked_write_redis.py @@ -0,0 +1,473 @@ +"""The fused tracked write on real Redis: a decorated miss sends its SET and its key-registry SADD + EXPIRE +in one round trip (PerRequestRedisBackend._set_tracked), and on async in one thread hop. + +Every count is compared with the separate-call path, reached by removing ``_set_tracked`` from the class: the +path a backend without it keeps, and the path every Redis-backed write took before the fused write. +""" + +from __future__ import annotations + +import asyncio +import logging +import time +from collections.abc import Callable, Iterator +from typing import Any +from unittest.mock import patch + +import pytest +import redis + +from cachekit import cache +from cachekit.backends.errors import BackendError +from cachekit.backends.redis.provider import PerRequestRedisBackend +from cachekit.cache_handler import StandardCacheHandler, _fuses_tracked_write +from cachekit.decorators import write_behind as wb + +pytestmark = pytest.mark.integration + +REG = "ck:reg:ns:0123456789abcdef" +N = 5 + + +@pytest.fixture +def client(redis_test_client: redis.Redis) -> redis.Redis: + return redis_test_client + + +@pytest.fixture(params=["fused", "separate"]) +def fused(request: pytest.FixtureRequest, monkeypatch: pytest.MonkeyPatch) -> bool: + """Run a test on the fused path, then again on the separate-call path.""" + if request.param == "separate": + monkeypatch.delattr(PerRequestRedisBackend, "_set_tracked") + return request.param == "fused" + + +class _Unlockable(PerRequestRedisBackend): + """No distributed lock: an async miss takes the unlocked write (the only one write-behind defers).""" + + acquire_lock = None # type: ignore[assignment] + + +def _write_flushes(monkeypatch: pytest.MonkeyPatch) -> list[bytes]: + """Every send to Redis that carries a cache write (a SET that is not a lock's SET NX) or a registry SADD. + + One send is one round trip: a pipeline packs all its commands into one send_packed_command. + """ + flushes: list[bytes] = [] + send = redis.connection.AbstractConnection.send_packed_command + + def counting_send(self: Any, command: Any, check_health: bool = True) -> None: + packed = b"".join(command) if isinstance(command, list) else command + if b"\r\nSADD\r\n" in packed or (b"\r\nSET\r\n" in packed and b"\r\nNX\r\n" not in packed): + flushes.append(packed) + send(self, command, check_health) + + monkeypatch.setattr(redis.connection.AbstractConnection, "send_packed_command", counting_send) + return flushes + + +def _tracked(client: redis.Redis) -> set[bytes]: + """Members of every tracking set of tenant ``default``.""" + return {member for reg in client.keys("t:default:ck:reg:*") for member in client.smembers(reg)} + + +def _cache_keys(client: redis.Redis) -> set[bytes]: + return {k[len(b"t:default:") :] for k in client.keys("t:default:*") if b":ck:reg:" not in k} + + +def _drain_writers(created: list[wb.WriteBehind]) -> None: + for writer in created: + assert writer.close(time.monotonic() + 5.0) == [], "a deferred write never finished" + + +@pytest.fixture +def writers(monkeypatch: pytest.MonkeyPatch) -> list[wb.WriteBehind]: + """The write-behind writers the test creates, so it can wait for their writes.""" + created: list[wb.WriteBehind] = [] + init = wb.WriteBehind.__init__ + + def recording_init(self: wb.WriteBehind, *args: Any, **kwargs: Any) -> None: + init(self, *args, **kwargs) + created.append(self) + + monkeypatch.setattr(wb.WriteBehind, "__init__", recording_init) + return created + + +class TestOneRoundTripPerTrackedWrite: + """N tracked writes: N round trips fused, 2N separate. Every key is stored and tracked either way.""" + + def _check(self, client: redis.Redis, flushes: list[bytes], fused: bool) -> None: + assert len(flushes) == (N if fused else 2 * N) + assert len(_cache_keys(client)) == N + assert _tracked(client) == _cache_keys(client) + + def test_sync_miss(self, client: redis.Redis, monkeypatch: pytest.MonkeyPatch, fused: bool) -> None: + fn = cache(backend=PerRequestRedisBackend(client, "default"), ttl=300, l1_enabled=False)(lambda x: x) + flushes = _write_flushes(monkeypatch) + for x in range(N): + fn(x) + self._check(client, flushes, fused) + + def test_write_behind_writer( + self, client: redis.Redis, monkeypatch: pytest.MonkeyPatch, fused: bool, writers: list[wb.WriteBehind] + ) -> None: + fn = cache(backend=PerRequestRedisBackend(client, "default"), ttl=300, l1_enabled=False, write_behind=True)(lambda x: x) + flushes = _write_flushes(monkeypatch) + for x in range(N): + fn(x) + _drain_writers(writers) + self._check(client, flushes, fused) + + @pytest.mark.asyncio + @pytest.mark.parametrize("backend_class", [PerRequestRedisBackend, _Unlockable], ids=["locked", "unlocked"]) + async def test_async_miss( + self, client: redis.Redis, monkeypatch: pytest.MonkeyPatch, fused: bool, backend_class: type + ) -> None: + async def lookup(x: int) -> int: + return x + + fn = cache(backend=backend_class(client, "default"), ttl=300, l1_enabled=False)(lookup) + flushes = _write_flushes(monkeypatch) + for x in range(N): + await fn(x) + self._check(client, flushes, fused) + + +class _CappedTTL(PerRequestRedisBackend): + """A subclass whose ``set`` caps every TTL at 60 s: the fused write must not bypass it.""" + + calls = 0 + + def set(self, key: str, value: bytes, ttl: int | None = None) -> None: + self.calls += 1 + super().set(key, value, 60) + + +class _OwnTracking(PerRequestRedisBackend): + """A subclass whose ``track_key`` does more than the registry write: the fused write must not bypass it.""" + + calls = 0 + + def track_key(self, registry_id: str, key: str) -> None: + self.calls += 1 + super().track_key(registry_id, key) + + +class _InterceptedSet(PerRequestRedisBackend): + """A subclass that caps ``set``'s TTL at 60 s through ``__getattribute__``, not by defining ``set``.""" + + calls = 0 + + def __getattribute__(self, name: str) -> Any: + method = super().__getattribute__(name) + if name != "set": + return method + + def capped(key: str, value: bytes, ttl: int | None = None) -> None: + self.calls += 1 + method(key, value, 60) + + return capped + + +class _OwnFused(PerRequestRedisBackend): + """A subclass that defines its own ``_set_tracked`` and inherits ``set`` and ``track_key``.""" + + _set_tracked = PerRequestRedisBackend._set_tracked + + +def _instance_bound(client: redis.Redis, name: str) -> PerRequestRedisBackend: + """A plain backend whose ``set`` (capped at 60 s) or ``track_key`` is replaced on the instance only.""" + backend = PerRequestRedisBackend(client, "default") + backend.calls = 0 # type: ignore[attr-defined] + original = getattr(backend, name) + + def hooked_set(key: str, value: bytes, ttl: int | None = None) -> None: + backend.calls += 1 # type: ignore[attr-defined] + original(key, value, 60) + + def hooked_track_key(registry_id: str, key: str) -> None: + backend.calls += 1 # type: ignore[attr-defined] + original(registry_id, key) + + setattr(backend, name, hooked_set if name == "set" else hooked_track_key) + return backend + + +# (backend factory, the TTL a stored key must carry at most) +_OVERRIDES = { + "subclass-set": (lambda client: _CappedTTL(client, "default"), 60), + "subclass-track-key": (lambda client: _OwnTracking(client, "default"), 300), + "getattribute-set": (lambda client: _InterceptedSet(client, "default"), 60), + "instance-set": (lambda client: _instance_bound(client, "set"), 60), + "instance-track-key": (lambda client: _instance_bound(client, "track_key"), 300), +} + + +_DELEGATED = ("get", "set", "track_key") + + +def _delegating_by_assignment(client: redis.Redis) -> PerRequestRedisBackend: + """A tenant-``source`` facade whose get, set and track_key are tenant ``target``'s bound methods.""" + facade, target = PerRequestRedisBackend(client, "source"), PerRequestRedisBackend(client, "target") + for name in _DELEGATED: + setattr(facade, name, getattr(target, name)) + return facade + + +class _DelegatingFacade(PerRequestRedisBackend): + """A tenant-``source`` facade that hands get, set and track_key to tenant ``target`` through ``__getattribute__``.""" + + def __init__(self, client: redis.Redis) -> None: + super().__init__(client, "source") + self._target = PerRequestRedisBackend(client, "target") + + def __getattribute__(self, name: str) -> Any: + if name in _DELEGATED: + return getattr(object.__getattribute__(self, "_target"), name) + return super().__getattribute__(name) + + +_DELEGATIONS = {"assignment": _delegating_by_assignment, "getattribute": _DelegatingFacade} + + +class TestDelegationWritesWhereItReads: + """Methods bound to another backend carry the recorded functions but another receiver: the write must go + where the delegate's set sends it, so the next read, also delegated, hits.""" + + @pytest.mark.parametrize("delegation", list(_DELEGATIONS)) + def test_never_fuses(self, client: redis.Redis, delegation: str) -> None: + assert not _fuses_tracked_write(_DELEGATIONS[delegation](client)) + + @pytest.mark.parametrize("delegation", list(_DELEGATIONS)) + def test_sync_miss(self, client: redis.Redis, delegation: str) -> None: + calls = 0 + + def f(x: int) -> int: + nonlocal calls + calls += 1 + return x + + fn = cache(backend=_DELEGATIONS[delegation](client), ttl=300, l1_enabled=False)(f) + fn(1) + fn(1) + assert calls == 1 + assert client.keys("t:source:*") == [] + + @pytest.mark.asyncio + @pytest.mark.parametrize("delegation", list(_DELEGATIONS)) + async def test_async_miss(self, client: redis.Redis, delegation: str) -> None: + calls = 0 + + async def f(x: int) -> int: + nonlocal calls + calls += 1 + return x + + fn = cache(backend=_DELEGATIONS[delegation](client), ttl=300, l1_enabled=False)(f) + await fn(1) + await fn(1) + assert calls == 1 + assert client.keys("t:source:*") == [] + + +class TestFuseDecision: + def test_plain_backend_fuses(self, client: redis.Redis) -> None: + assert _fuses_tracked_write(PerRequestRedisBackend(client, "default")) + + def test_owner_without_a_record_never_fuses(self, client: redis.Redis) -> None: + """An owner that defines ``_set_tracked`` but records no ``_set_tracked_replaces`` never fuses, even when the + instance's set and track_key are plain functions: their missing ``__func__`` must not match the missing record.""" + backend = _OwnFused(client, "default") + backend.set = backend.track_key = lambda *args: None # type: ignore[method-assign] + assert not _fuses_tracked_write(backend) + + @pytest.mark.parametrize("name", ["set", "track_key"]) + def test_class_level_patch_never_fuses(self, client: redis.Redis, monkeypatch: pytest.MonkeyPatch, name: str) -> None: + """A patch of the class itself, as a test double or an app hook would apply it, keeps the separate calls.""" + original = getattr(PerRequestRedisBackend, name) + monkeypatch.setattr(PerRequestRedisBackend, name, lambda self, *args: original(self, *args)) + assert not _fuses_tracked_write(PerRequestRedisBackend(client, "default")) + + def test_forwarding_proxy_runs_its_wrapper(self, client: redis.Redis, monkeypatch: pytest.MonkeyPatch) -> None: + """A wrapt-style instrumentation proxy forwards __func__ and __self__ to the real bound method; its wrapper + must still run on every write.""" + calls = 0 + + class _Proxy: + def __init__(self, bound: Any) -> None: + self._bound = bound + + def __getattr__(self, name: str) -> Any: + return getattr(self._bound, name) + + def __call__(self, *args: Any) -> Any: + nonlocal calls + calls += 1 + return self._bound(*args) + + class _Wrapped: + def __init__(self, func: Any) -> None: + self._func = func + + def __get__(self, backend: Any, owner: Any = None) -> Any: + return _Proxy(self._func.__get__(backend, owner)) + + monkeypatch.setattr(PerRequestRedisBackend, "set", _Wrapped(PerRequestRedisBackend.set)) + backend = PerRequestRedisBackend(client, "default") + assert backend.set.__func__ is PerRequestRedisBackend._set_tracked_replaces["set"] # the proxy forwards + assert not _fuses_tracked_write(backend) + cache(backend=backend, ttl=300, l1_enabled=False)(lambda x: x)(1) + assert calls == 1 + + def test_autospec_patch_runs_the_patched_set(self, client: redis.Redis) -> None: + fn = cache(backend=PerRequestRedisBackend(client, "default"), ttl=300, l1_enabled=False)(lambda x: x) + with patch.object(PerRequestRedisBackend, "set", autospec=True) as patched: + fn(1) + assert patched.call_count == 1 + + +class TestOverridesKeepTheSeparateCalls: + """Whatever way set or track_key is overridden, the override runs on every write: N calls, 2N round trips, + its TTL kept.""" + + @pytest.mark.parametrize("override", list(_OVERRIDES)) + def test_sync_miss(self, client: redis.Redis, monkeypatch: pytest.MonkeyPatch, override: str) -> None: + make, max_ttl = _OVERRIDES[override] + backend = make(client) + fn = cache(backend=backend, ttl=300, l1_enabled=False)(lambda x: x) + flushes = _write_flushes(monkeypatch) + for x in range(N): + fn(x) + assert backend.calls == N + assert len(flushes) == 2 * N + assert _tracked(client) == _cache_keys(client) + assert all(max_ttl - 10 < client.ttl(f"t:default:{key.decode()}") <= max_ttl for key in _cache_keys(client)) + + @pytest.mark.asyncio + @pytest.mark.parametrize("override", list(_OVERRIDES)) + async def test_async_miss(self, client: redis.Redis, override: str) -> None: + async def lookup(x: int) -> int: + return x + + make, max_ttl = _OVERRIDES[override] + backend = make(client) + fn = cache(backend=backend, ttl=300, l1_enabled=False)(lookup) + for x in range(N): + await fn(x) + assert backend.calls == N + assert all(max_ttl - 10 < client.ttl(f"t:default:{key.decode()}") <= max_ttl for key in _cache_keys(client)) + + +class TestOneThreadHopPerAsyncTrackedWrite: + @pytest.mark.asyncio + @pytest.mark.parametrize("backend_class", [PerRequestRedisBackend, _Unlockable], ids=["locked", "unlocked"]) + async def test_fused_saves_one_hop_per_write( + self, client: redis.Redis, monkeypatch: pytest.MonkeyPatch, backend_class: type + ) -> None: + """The same N misses, fused then separate: the separate path takes exactly N more hops, the track's.""" + + async def lookup(x: int) -> int: + return x + + hops: list[Callable[..., Any]] = [] + to_thread = asyncio.to_thread + + async def counting_to_thread(func: Callable[..., Any], /, *args: Any, **kwargs: Any) -> Any: + hops.append(func) + return await to_thread(func, *args, **kwargs) + + monkeypatch.setattr(asyncio, "to_thread", counting_to_thread) + + async def hops_for(namespace: str) -> int: + fn = cache(backend=backend_class(client, "default"), ttl=300, l1_enabled=False, namespace=namespace)(lookup) + hops.clear() + for x in range(N): + await fn(x) + return len(hops) + + fused_hops = await hops_for("fused") + with monkeypatch.context() as m: + m.delattr(PerRequestRedisBackend, "_set_tracked") + separate_hops = await hops_for("separate") + assert separate_hops - fused_hops == N + assert len(_tracked(client)) == 2 * N + + +@pytest.fixture +def set_denied(client: redis.Redis) -> Iterator[redis.Redis]: + """A client whose ACL user may run everything but SET: the SET of a fused write gets an error reply while + its SADD and EXPIRE run.""" + client.acl_setuser("no-set", enabled=True, nopass=True, keys=["*"], commands=["+@all", "-set"]) + pool = client.connection_pool + denied = redis.Redis( + connection_pool=redis.ConnectionPool( + connection_class=pool.connection_class, **{**pool.connection_kwargs, "username": "no-set", "password": None} + ) + ) + yield denied + denied.close() + client.acl_deluser("no-set") + + +class TestFusedWriteErrors: + def test_set_error_reply_is_not_stored_and_never_raises(self, client: redis.Redis, set_denied: redis.Redis) -> None: + handler = StandardCacheHandler(PerRequestRedisBackend(set_denied, "default")) + track_errors: list[Exception] = [] + stored = handler.set_tracked("k", b"v", 60, REG, lambda _key, e: track_errors.append(e)) + assert stored is False + assert track_errors == [] + assert not client.exists("t:default:k") + + def test_set_error_reply_through_the_decorator(self, client: redis.Redis, set_denied: redis.Redis) -> None: + fn = cache(backend=PerRequestRedisBackend(set_denied, "default"), ttl=300, l1_enabled=False)(lambda x: x * 2) + assert fn(3) == 6 # the write failed; the caller still gets its value + assert _cache_keys(client) == set() + + def test_registry_error_reply_is_stored_and_warns( + self, client: redis.Redis, caplog: pytest.LogCaptureFixture, monkeypatch: pytest.MonkeyPatch + ) -> None: + fn = cache(backend=PerRequestRedisBackend(client, "default"), ttl=300, l1_enabled=False)(lambda x: x) + fn(0) # creates the registry set; turn it into a string so the next SADD gets WRONGTYPE + [reg] = client.keys("t:default:ck:reg:*") + client.delete(reg) + client.set(reg, b"not a set") + flushes = _write_flushes(monkeypatch) + with caplog.at_level(logging.WARNING, logger="cachekit.decorators.wrapper"): + assert fn(1) == 1 + assert len(flushes) == 1 # the SADD error came back from the fused pipeline, not a separate track_key + assert len(_cache_keys(client)) == 2 # stored despite the failed SADD + assert any("Key tracking failed" in r.getMessage() for r in caplog.records) + + def test_backend_returns_the_registry_error(self, client: redis.Redis) -> None: + client.set(f"t:default:{REG}", b"not a set") + error = PerRequestRedisBackend(client, "default")._set_tracked("k", b"v", 60, REG) + assert isinstance(error, BackendError) + assert client.get("t:default:k") == b"v" + + def test_connection_drop_is_not_stored(self, client: redis.Redis, monkeypatch: pytest.MonkeyPatch) -> None: + def dropped(self: Any, raise_on_error: bool = True) -> list[Any]: + raise redis.exceptions.ConnectionError("Connection reset by peer") + + monkeypatch.setattr(redis.client.Pipeline, "execute", dropped) + handler = StandardCacheHandler(PerRequestRedisBackend(client, "default")) + track_errors: list[Exception] = [] + assert handler.set_tracked("k", b"v", 60, REG, lambda _key, e: track_errors.append(e)) is False + assert track_errors == [] + + @pytest.mark.parametrize(("ttl", "expected"), [(None, -1), (0, -1), (-5, -1), (60, 60)]) + def test_ttl_none_or_not_positive_sends_no_expiry(self, client: redis.Redis, ttl: int | None, expected: int) -> None: + assert PerRequestRedisBackend(client, "default")._set_tracked("k", b"v", ttl, REG) is None + assert client.ttl("t:default:k") == expected + assert 604_000 < client.ttl(f"t:default:{REG}") <= 604_800 + + +class TestPhantomMember: + def test_member_left_by_a_failed_set_is_drained_without_error(self, client: redis.Redis, set_denied: redis.Redis) -> None: + """A fused write whose SET failed still ran its SADD: the next drain unlinks the missing key and empties the set.""" + with pytest.raises(BackendError): + PerRequestRedisBackend(set_denied, "default")._set_tracked("ghost", b"v", 60, REG) + assert client.smembers(f"t:default:{REG}") == {b"ghost"} + assert PerRequestRedisBackend(client, "default").drain_tracked(REG, set()) == {"ghost"} + assert not client.exists(f"t:default:{REG}") diff --git a/tests/unit/backends/test_redis_backend.py b/tests/unit/backends/test_redis_backend.py index a81fa37a..6147eb17 100644 --- a/tests/unit/backends/test_redis_backend.py +++ b/tests/unit/backends/test_redis_backend.py @@ -36,6 +36,7 @@ import uuid from concurrent.futures import ThreadPoolExecutor from types import SimpleNamespace +from typing import Any from unittest.mock import Mock, call, patch import pytest @@ -475,6 +476,35 @@ def evalsha(self, _sha: str, _numkeys: int, name: str, token: bytes) -> int: del self._store[name] return 1 + def pipeline(self, transaction: bool = True) -> _FakePipeline: + return _FakePipeline(self) + + +class _FakePipeline: + """The fused tracked write's pipeline over ``_FakeRedis``: SET runs, and the key registry's commands, + which the fake does not model, reply with an error, so tracking fails as it does without a pipeline.""" + + def __init__(self, fake: _FakeRedis) -> None: + self._fake = fake + self._queued: list[Any] = [] + + def set(self, name: str, value: bytes, ex: int | None = None) -> _FakePipeline: + self._queued.append(lambda: self._fake.set(name, value, ex=ex)) + return self + + def sadd(self, *_args: Any) -> _FakePipeline: + self._queued.append(lambda: redis.exceptions.ResponseError("unknown command")) + return self + + expire = sadd + + def execute(self, raise_on_error: bool = True) -> list[Any]: + replies = [command() for command in self._queued] + error = next((reply for reply in replies if isinstance(reply, Exception)), None) + if raise_on_error and error is not None: + raise error + return replies + async def _entered(event: threading.Event, what: str) -> None: """Poll a thread-side event from the loop; ``to_thread(event.wait)`` would take the very executor thread under test."""