From a8cfc37119071f77c83957799e5971e0c2e16b99 Mon Sep 17 00:00:00 2001 From: Ray Walker Date: Thu, 29 Oct 2026 16:42:13 +1100 Subject: [PATCH 01/12] perf(redis): send the key-registry SADD+EXPIRE in the same pipeline as the SET A tracked decorated write on PerRequestRedisBackend now sends SET, SADD and EXPIRE in one non-transactional pipeline (a private _set_tracked), and on the async miss paths in one thread hop: one round trip per write instead of two. The write's success is the SET reply alone; a registry error stays advisory and goes to the existing throttled 'Key tracking failed' warning. A backend without _set_tracked, or a custom handler strategy without set_tracked, keeps the separate track_key call. KeyTrackableBackend, the registry id and the set's TTL are unchanged; track_key and the fused write share one helper for the registry commands. --- .secrets.baseline | 4 +- docs/backends/redis.md | 2 +- docs/features/l1-invalidation.md | 4 +- src/cachekit/backends/redis/provider.py | 69 ++++- src/cachekit/cache_handler.py | 95 ++++++- src/cachekit/decorators/wrapper.py | 84 +++--- tests/integration/test_tracked_write_redis.py | 244 ++++++++++++++++++ tests/unit/backends/test_redis_backend.py | 30 +++ 8 files changed, 488 insertions(+), 44 deletions(-) create mode 100644 tests/integration/test_tracked_write_redis.py diff --git a/.secrets.baseline b/.secrets.baseline index 03f09022..3918c916 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": 653 } ], "src/cachekit/serializers/interop_serializer.py": [ @@ -2020,5 +2020,5 @@ } ] }, - "generated_at": "2026-10-19T08:46:47Z" + "generated_at": "2026-10-29T05:42:01Z" } diff --git a/docs/backends/redis.md b/docs/backends/redis.md index 44dc4571..2c82efb0 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, a decorated write included (its [key-registry](../features/l1-invalidation.md#whole-function-invalidation) entry rides in the same pipeline as its `SET`), which depends on your network and region ([not measured here](../performance.md#not-measured-here)) - 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..3aabcdf7 100644 --- a/src/cachekit/backends/redis/provider.py +++ b/src/cachekit/backends/redis/provider.py @@ -50,6 +50,18 @@ # 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. ``pipeline`` is a parameter here only + for the two queuing calls, which raise only on an argument type no caller passes. + """ + 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 +718,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 +728,56 @@ 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: ``CacheHandler.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) + try: + raise error from error.original_exception + finally: + del error + 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..9290f965 100644 --- a/src/cachekit/cache_handler.py +++ b/src/cachekit/cache_handler.py @@ -367,6 +367,12 @@ 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``.""" + return callable(getattr(type(backend), "_set_tracked", None)) and supports_key_tracking(backend) + + 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 +1571,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 +2002,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] | None = None, ) -> StoreOutcome: """Store result in backend cache with optional tenant context for encryption. @@ -2006,6 +2017,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 +2063,25 @@ 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) + # getattr: custom CacheHandlerStrategy impls predate set_tracked, as set_streaming above. + set_tracked = None if registry_id is None else getattr(self._cache_handler, "set_tracked", None) + stored = ( + None + if set_tracked is None + else set_tracked(cache_key, serialized_data, ttl, registry_id, on_track_error, stale_ttl=stale_ttl) + ) + tracked = stored is not None # Only thread the SWR kwarg when set: strategy implementations without # **metadata (tests, custom handlers) must keep working unchanged. - if stale_ttl is not None: + if not tracked and stale_ttl is not None: stored = self._cache_handler.set(cache_key, serialized_data, ttl, stale_ttl=stale_ttl) - else: + elif not tracked: stored = self._cache_handler.set(cache_key, serialized_data, ttl) 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 +2454,42 @@ 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: Optional[Callable[[str, Exception], None]], + stale_ttl: Optional[int] = 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 self._fuses_tracking(stale_ttl): + 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 and on_track_error is not None: + on_track_error(key, track_error) + return True + + def _fuses_tracking(self, stale_ttl: Optional[int]) -> bool: + """Whether set_tracked can fuse on this backend. An SWR window is a set argument the fused write + has no slot for, so an SWR-capable backend given one keeps set.""" + return _fuses_tracked_write(self.backend) and (stale_ttl is None or not supports_swr(self.backend)) + 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 +2613,36 @@ 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: Optional[Callable[[str, Exception], None]], + stale_ttl: Optional[int] = None, + ) -> Optional[bool]: + """Async variant of :meth:`set_tracked`: the write and its registry entry in one thread-pool hop.""" + if not self._fuses_tracking(stale_ttl): + 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 and on_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..28bddf20 100644 --- a/src/cachekit/decorators/wrapper.py +++ b/src/cachekit/decorators/wrapper.py @@ -1302,24 +1302,43 @@ 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, stale_ttl=_stale_ttl + ) + 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 +1841,24 @@ 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: + # As in _set_l2_async: trackable first, and custom strategies predate set_tracked. + set_tracked = getattr(handler, "set_tracked", None) if _is_trackable() else None + stored = ( + None + if set_tracked is None + else set_tracked(cache_key, envelope, ttl, _registry_id, _warn_track_failure, stale_ttl=_stale_ttl) + ) + tracked = stored is not None + if not tracked and _stale_ttl is not None: stored = handler.set(cache_key, envelope, ttl, stale_ttl=_stale_ttl) - else: + elif not tracked: stored = handler.set(cache_key, envelope, ttl) 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 +2554,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 +2564,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 +3120,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 +3239,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..c4bb3429 --- /dev/null +++ b/tests/integration/test_tracked_write_redis.py @@ -0,0 +1,244 @@ +"""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 +from typing import Any + +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 +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 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) -> 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_backend_raises_on_set_error_reply(self, set_denied: redis.Redis) -> None: + with pytest.raises(BackendError): + PerRequestRedisBackend(set_denied, "default")._set_tracked("k", b"v", 60, REG) + + def test_registry_error_reply_is_stored_and_warns(self, client: redis.Redis, caplog: pytest.LogCaptureFixture) -> 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") + with caplog.at_level(logging.WARNING, logger="cachekit.decorators.wrapper"): + assert fn(1) == 1 + 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")) + assert handler.set_tracked("k", b"v", 60, REG, None) is False + + @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.""" From 68d35c4944259f39f8892aa1d5871f759274ab80 Mon Sep 17 00:00:00 2001 From: Ray Walker Date: Thu, 29 Oct 2026 16:59:54 +1100 Subject: [PATCH 02/12] refactor(redis): one fuse-or-fall-back helper; tracking errors always reported on_track_error is required, so a registry error can never be dropped silently. The SWR guard moves into the one capability predicate, and the stale_ttl parameter it was carried by goes. store_result and the write-behind writer share _set_l2 instead of two copies of the fallback. --- .secrets.baseline | 4 +- docs/backends/redis.md | 2 +- src/cachekit/backends/redis/provider.py | 5 +- src/cachekit/cache_handler.py | 71 +++++++++++-------- src/cachekit/decorators/wrapper.py | 20 ++---- tests/integration/test_tracked_write_redis.py | 14 ++-- 6 files changed, 60 insertions(+), 56 deletions(-) diff --git a/.secrets.baseline b/.secrets.baseline index 3918c916..845de13b 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": 653 + "line_number": 684 } ], "src/cachekit/serializers/interop_serializer.py": [ @@ -2020,5 +2020,5 @@ } ] }, - "generated_at": "2026-10-29T05:42:01Z" + "generated_at": "2026-10-29T05:59:48Z" } diff --git a/docs/backends/redis.md b/docs/backends/redis.md index 2c82efb0..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, a decorated write included (its [key-registry](../features/l1-invalidation.md#whole-function-invalidation) entry rides in the same pipeline as its `SET`), 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/src/cachekit/backends/redis/provider.py b/src/cachekit/backends/redis/provider.py index 3aabcdf7..c2e01d45 100644 --- a/src/cachekit/backends/redis/provider.py +++ b/src/cachekit/backends/redis/provider.py @@ -56,8 +56,7 @@ 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. ``pipeline`` is a parameter here only - for the two queuing calls, which raise only on an argument type no caller passes. + Queuing sends nothing; the caller chains ``execute()`` on the result. """ return pipeline.sadd(scoped_reg, key).expire(scoped_reg, _TRACKING_SET_TTL_SECONDS) @@ -732,7 +731,7 @@ def _set_tracked(self, key: str, value: bytes, ttl: Optional[int], registry_id: """``set`` and ``track_key`` in one round trip: ``SET``, ``SADD`` and ``EXPIRE`` on one non-transactional pipeline. - Private: ``CacheHandler.set_tracked`` finds it on the backend class, and a backend without it + 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 diff --git a/src/cachekit/cache_handler.py b/src/cachekit/cache_handler.py index 9290f965..72b50a32 100644 --- a/src/cachekit/cache_handler.py +++ b/src/cachekit/cache_handler.py @@ -369,8 +369,39 @@ def supports_key_tracking(backend: object) -> TypeGuard[KeyTrackableBackend]: 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``.""" - return callable(getattr(type(backend), "_set_tracked", None)) and supports_key_tracking(backend) + found on the CLASS as ``supports_key_tracking`` finds ``track_key``. An SWR-capable backend keeps ``set``: + the fused write has no slot for a stale-grace window.""" + return ( + callable(getattr(type(backend), "_set_tracked", None)) and supports_key_tracking(backend) and not supports_swr(backend) # pyright: ignore[reportArgumentType] + ) + + +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): @@ -2003,7 +2034,7 @@ def store_result( stale_ttl: int | None = None, defer: Callable[[bytes], bool] | None = None, registry_id: str | None = None, - on_track_error: Callable[[str, Exception], None] | 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. @@ -2063,20 +2094,9 @@ 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) - # getattr: custom CacheHandlerStrategy impls predate set_tracked, as set_streaming above. - set_tracked = None if registry_id is None else getattr(self._cache_handler, "set_tracked", None) - stored = ( - None - if set_tracked is None - else set_tracked(cache_key, serialized_data, ttl, registry_id, on_track_error, stale_ttl=stale_ttl) + stored, tracked = _set_l2( + self._cache_handler, cache_key, serialized_data, ttl, stale_ttl, registry_id, on_track_error ) - tracked = stored is not None - # Only thread the SWR kwarg when set: strategy implementations without - # **metadata (tests, custom handlers) must keep working unchanged. - if not tracked and stale_ttl is not None: - stored = self._cache_handler.set(cache_key, serialized_data, ttl, stale_ttl=stale_ttl) - elif not tracked: - stored = self._cache_handler.set(cache_key, serialized_data, ttl) if stored: get_logger().cache_stored(cache_key, ttl) @@ -2460,8 +2480,7 @@ def set_tracked( value: bytes, ttl: Optional[int], registry_id: str, - on_track_error: Optional[Callable[[str, Exception], None]], - stale_ttl: Optional[int] = None, + on_track_error: Callable[[str, Exception], None], ) -> Optional[bool]: """``set`` plus the key-registry write (``KeyTrackableBackend.track_key``), in one backend round trip. @@ -2471,7 +2490,7 @@ def set_tracked( 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 self._fuses_tracking(stale_ttl): + 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] @@ -2481,15 +2500,10 @@ def set_tracked( 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 and on_track_error is not None: + if track_error is not None: on_track_error(key, track_error) return True - def _fuses_tracking(self, stale_ttl: Optional[int]) -> bool: - """Whether set_tracked can fuse on this backend. An SWR window is a set argument the fused write - has no slot for, so an SWR-capable backend given one keeps set.""" - return _fuses_tracked_write(self.backend) and (stale_ttl is None or not supports_swr(self.backend)) - 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. @@ -2619,11 +2633,10 @@ async def set_tracked_async( value: bytes, ttl: Optional[int], registry_id: str, - on_track_error: Optional[Callable[[str, Exception], None]], - stale_ttl: Optional[int] = None, + 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 self._fuses_tracking(stale_ttl): + if not _fuses_tracked_write(self.backend): return None try: track_error = await self._with_backpressure_and_timeout_async( @@ -2639,7 +2652,7 @@ async def set_tracked_async( 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 and on_track_error is not None: + if track_error is not None: on_track_error(key, track_error) return True diff --git a/src/cachekit/decorators/wrapper.py b/src/cachekit/decorators/wrapper.py index 28bddf20..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, @@ -1332,9 +1333,7 @@ async def _set_l2_async(cache_key: str, serialized_data: bytes) -> tuple[bool, b # 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, stale_ttl=_stale_ttl - ) + 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] @@ -1841,18 +1840,9 @@ 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() - # As in _set_l2_async: trackable first, and custom strategies predate set_tracked. - set_tracked = getattr(handler, "set_tracked", None) if _is_trackable() else None - stored = ( - None - if set_tracked is None - else set_tracked(cache_key, envelope, ttl, _registry_id, _warn_track_failure, stale_ttl=_stale_ttl) - ) - tracked = stored is not None - if not tracked and _stale_ttl is not None: - stored = handler.set(cache_key, envelope, ttl, stale_ttl=_stale_ttl) - elif not tracked: - 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) diff --git a/tests/integration/test_tracked_write_redis.py b/tests/integration/test_tracked_write_redis.py index c4bb3429..6f18e288 100644 --- a/tests/integration/test_tracked_write_redis.py +++ b/tests/integration/test_tracked_write_redis.py @@ -198,18 +198,18 @@ def test_set_error_reply_through_the_decorator(self, client: redis.Redis, set_de assert fn(3) == 6 # the write failed; the caller still gets its value assert _cache_keys(client) == set() - def test_backend_raises_on_set_error_reply(self, set_denied: redis.Redis) -> None: - with pytest.raises(BackendError): - PerRequestRedisBackend(set_denied, "default")._set_tracked("k", b"v", 60, REG) - - def test_registry_error_reply_is_stored_and_warns(self, client: redis.Redis, caplog: pytest.LogCaptureFixture) -> None: + 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) @@ -225,7 +225,9 @@ def dropped(self: Any, raise_on_error: bool = True) -> list[Any]: monkeypatch.setattr(redis.client.Pipeline, "execute", dropped) handler = StandardCacheHandler(PerRequestRedisBackend(client, "default")) - assert handler.set_tracked("k", b"v", 60, REG, None) is False + 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: From 63489ff67580da20aab631c745f739929a02affc Mon Sep 17 00:00:00 2001 From: Ray Walker Date: Thu, 29 Oct 2026 17:01:11 +1100 Subject: [PATCH 03/12] fix(redis): drop the raw pipeline replies before raising a failed SET --- src/cachekit/backends/redis/provider.py | 1 + tests/integration/test_tracked_write_redis.py | 4 ++-- 2 files changed, 3 insertions(+), 2 deletions(-) diff --git a/src/cachekit/backends/redis/provider.py b/src/cachekit/backends/redis/provider.py index c2e01d45..44486310 100644 --- a/src/cachekit/backends/redis/provider.py +++ b/src/cachekit/backends/redis/provider.py @@ -772,6 +772,7 @@ def _set_tracked(self, key: str, value: bytes, ttl: Optional[int], registry_id: 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: diff --git a/tests/integration/test_tracked_write_redis.py b/tests/integration/test_tracked_write_redis.py index 6f18e288..396d1b8b 100644 --- a/tests/integration/test_tracked_write_redis.py +++ b/tests/integration/test_tracked_write_redis.py @@ -10,7 +10,7 @@ import asyncio import logging import time -from collections.abc import Callable +from collections.abc import Callable, Iterator from typing import Any import pytest @@ -169,7 +169,7 @@ async def hops_for(namespace: str) -> int: @pytest.fixture -def set_denied(client: redis.Redis) -> redis.Redis: +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"]) From 9e52a4988c595c8a9790723dcc69fed579f68db5 Mon Sep 17 00:00:00 2001 From: Ray Walker Date: Thu, 29 Oct 2026 17:15:03 +1100 Subject: [PATCH 04/12] fix(redis): keep the separate calls when a subclass overrides set or track_key The fused write was inherited by any PerRequestRedisBackend subclass, so an override of set (a TTL cap, extra writes) or track_key stopped running. Fuse only when both come from the class that defines _set_tracked. --- .secrets.baseline | 4 +- src/cachekit/cache_handler.py | 17 +++++-- tests/integration/test_tracked_write_redis.py | 48 +++++++++++++++++++ 3 files changed, 62 insertions(+), 7 deletions(-) diff --git a/.secrets.baseline b/.secrets.baseline index 845de13b..602e8cdc 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": 684 + "line_number": 691 } ], "src/cachekit/serializers/interop_serializer.py": [ @@ -2020,5 +2020,5 @@ } ] }, - "generated_at": "2026-10-29T05:59:48Z" + "generated_at": "2026-10-29T06:14:57Z" } diff --git a/src/cachekit/cache_handler.py b/src/cachekit/cache_handler.py index 72b50a32..713af016 100644 --- a/src/cachekit/cache_handler.py +++ b/src/cachekit/cache_handler.py @@ -369,11 +369,18 @@ def supports_key_tracking(backend: object) -> TypeGuard[KeyTrackableBackend]: 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``. An SWR-capable backend keeps ``set``: - the fused write has no slot for a stale-grace window.""" - return ( - callable(getattr(type(backend), "_set_tracked", None)) and supports_key_tracking(backend) and not supports_swr(backend) # pyright: ignore[reportArgumentType] - ) + found on the CLASS as ``supports_key_tracking`` finds ``track_key``. + + Only when ``set`` and ``track_key`` come from the class that defines ``_set_tracked``: a subclass that + overrides either one (a TTL cap, extra writes) keeps both calls, so its override still runs. An + SWR-capable backend keeps ``set`` too: the fused write has no slot for a stale-grace window. + """ + cls = type(backend) + owner = next((klass for klass in cls.__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 + defined = vars(owner) + return getattr(cls, "set", None) is defined.get("set") and getattr(cls, "track_key", None) is defined.get("track_key") def _set_l2( diff --git a/tests/integration/test_tracked_write_redis.py b/tests/integration/test_tracked_write_redis.py index 396d1b8b..ac9de665 100644 --- a/tests/integration/test_tracked_write_redis.py +++ b/tests/integration/test_tracked_write_redis.py @@ -133,6 +133,54 @@ async def lookup(x: int) -> int: self._check(client, flushes, fused) +class _CappedTTL(PerRequestRedisBackend): + """A subclass whose ``set`` caps every TTL: the fused write must not bypass it.""" + + calls = 0 + + def set(self, key: str, value: bytes, ttl: int | None = None) -> None: + type(self).calls += 1 + super().set(key, value, 2) + + +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: + type(self).calls += 1 + super().track_key(registry_id, key) + + +class TestSubclassOverridesKeepTheSeparateCalls: + @pytest.mark.parametrize("backend_class", [_CappedTTL, _OwnTracking]) + def test_sync_miss(self, client: redis.Redis, monkeypatch: pytest.MonkeyPatch, backend_class: Any) -> None: + monkeypatch.setattr(backend_class, "calls", 0) + fn = cache(backend=backend_class(client, "default"), ttl=300, l1_enabled=False)(lambda x: x) + flushes = _write_flushes(monkeypatch) + for x in range(N): + fn(x) + assert backend_class.calls == N + assert len(flushes) == 2 * N + assert _tracked(client) == _cache_keys(client) + if backend_class is _CappedTTL: + assert all(0 < client.ttl(f"t:default:{key.decode()}") <= 2 for key in _cache_keys(client)) + + @pytest.mark.asyncio + async def test_async_miss(self, client: redis.Redis, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(_CappedTTL, "calls", 0) + + async def lookup(x: int) -> int: + return x + + fn = cache(backend=_CappedTTL(client, "default"), ttl=300, l1_enabled=False)(lookup) + for x in range(N): + await fn(x) + assert _CappedTTL.calls == N + assert all(0 < client.ttl(f"t:default:{key.decode()}") <= 2 for key in _cache_keys(client)) + + class TestOneThreadHopPerAsyncTrackedWrite: @pytest.mark.asyncio @pytest.mark.parametrize("backend_class", [PerRequestRedisBackend, _Unlockable], ids=["locked", "unlocked"]) From 3df9c0606b22bcb05642f1152bd571296dd0a4ba Mon Sep 17 00:00:00 2001 From: Ray Walker Date: Thu, 29 Oct 2026 17:16:45 +1100 Subject: [PATCH 05/12] test(redis): override tests cap at 60 s and count per instance --- tests/integration/test_tracked_write_redis.py | 36 ++++++++++--------- 1 file changed, 19 insertions(+), 17 deletions(-) diff --git a/tests/integration/test_tracked_write_redis.py b/tests/integration/test_tracked_write_redis.py index ac9de665..60e475dd 100644 --- a/tests/integration/test_tracked_write_redis.py +++ b/tests/integration/test_tracked_write_redis.py @@ -134,13 +134,13 @@ async def lookup(x: int) -> int: class _CappedTTL(PerRequestRedisBackend): - """A subclass whose ``set`` caps every TTL: the fused write must not bypass it.""" + """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: - type(self).calls += 1 - super().set(key, value, 2) + self.calls += 1 + super().set(key, value, 60) class _OwnTracking(PerRequestRedisBackend): @@ -149,36 +149,38 @@ class _OwnTracking(PerRequestRedisBackend): calls = 0 def track_key(self, registry_id: str, key: str) -> None: - type(self).calls += 1 + self.calls += 1 super().track_key(registry_id, key) class TestSubclassOverridesKeepTheSeparateCalls: - @pytest.mark.parametrize("backend_class", [_CappedTTL, _OwnTracking]) - def test_sync_miss(self, client: redis.Redis, monkeypatch: pytest.MonkeyPatch, backend_class: Any) -> None: - monkeypatch.setattr(backend_class, "calls", 0) - fn = cache(backend=backend_class(client, "default"), ttl=300, l1_enabled=False)(lambda x: x) + """A subclass's own set or track_key runs on every write: N override calls, 2N round trips, its TTL kept.""" + + @pytest.mark.parametrize( + ("backend_class", "max_ttl"), [(_CappedTTL, 60), (_OwnTracking, 300)], ids=["own-set", "own-track-key"] + ) + def test_sync_miss(self, client: redis.Redis, monkeypatch: pytest.MonkeyPatch, backend_class: Any, max_ttl: int) -> None: + backend = backend_class(client, "default") + 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_class.calls == N + assert backend.calls == N assert len(flushes) == 2 * N assert _tracked(client) == _cache_keys(client) - if backend_class is _CappedTTL: - assert all(0 < client.ttl(f"t:default:{key.decode()}") <= 2 for key in _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 - async def test_async_miss(self, client: redis.Redis, monkeypatch: pytest.MonkeyPatch) -> None: - monkeypatch.setattr(_CappedTTL, "calls", 0) - + async def test_async_miss(self, client: redis.Redis) -> None: async def lookup(x: int) -> int: return x - fn = cache(backend=_CappedTTL(client, "default"), ttl=300, l1_enabled=False)(lookup) + backend = _CappedTTL(client, "default") + fn = cache(backend=backend, ttl=300, l1_enabled=False)(lookup) for x in range(N): await fn(x) - assert _CappedTTL.calls == N - assert all(0 < client.ttl(f"t:default:{key.decode()}") <= 2 for key in _cache_keys(client)) + assert backend.calls == N + assert all(50 < client.ttl(f"t:default:{key.decode()}") <= 60 for key in _cache_keys(client)) class TestOneThreadHopPerAsyncTrackedWrite: From a231bf87902656ba342be437704791f30fe2fc39 Mon Sep 17 00:00:00 2001 From: Ray Walker Date: Thu, 29 Oct 2026 17:23:43 +1100 Subject: [PATCH 06/12] fix(redis): decide the fused write from what the instance resolves An instance-bound set or track_key, or a subclass that intercepts lookup with __getattribute__, still passed the class-level check and was bypassed. Fuse only when the instance's own set and track_key are the plain methods of the class that defines _set_tracked. --- src/cachekit/cache_handler.py | 12 ++-- tests/integration/test_tracked_write_redis.py | 63 ++++++++++++++++--- 2 files changed, 59 insertions(+), 16 deletions(-) diff --git a/src/cachekit/cache_handler.py b/src/cachekit/cache_handler.py index 713af016..337094b3 100644 --- a/src/cachekit/cache_handler.py +++ b/src/cachekit/cache_handler.py @@ -371,16 +371,16 @@ 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 when ``set`` and ``track_key`` come from the class that defines ``_set_tracked``: a subclass that - overrides either one (a TTL cap, extra writes) keeps both calls, so its override still runs. An - SWR-capable backend keeps ``set`` too: the fused write has no slot for a stale-grace window. + Only when the backend's own ``set`` and ``track_key``, as the instance resolves them, are the plain methods + of the class that defines ``_set_tracked``. An override anywhere (a subclass, an instance attribute, + ``__getattribute__``) 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. """ - cls = type(backend) - owner = next((klass for klass in cls.__mro__ if "_set_tracked" in vars(klass)), None) + 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 defined = vars(owner) - return getattr(cls, "set", None) is defined.get("set") and getattr(cls, "track_key", None) is defined.get("track_key") + return all(getattr(getattr(backend, name, None), "__func__", None) is defined.get(name) for name in ("set", "track_key")) def _set_l2( diff --git a/tests/integration/test_tracked_write_redis.py b/tests/integration/test_tracked_write_redis.py index 60e475dd..8c02e64b 100644 --- a/tests/integration/test_tracked_write_redis.py +++ b/tests/integration/test_tracked_write_redis.py @@ -153,14 +153,55 @@ def track_key(self, registry_id: str, key: str) -> None: super().track_key(registry_id, key) -class TestSubclassOverridesKeepTheSeparateCalls: - """A subclass's own set or track_key runs on every write: N override calls, 2N round trips, its TTL kept.""" +class _InterceptedSet(PerRequestRedisBackend): + """A subclass that caps ``set``'s TTL at 60 s through ``__getattribute__``, not by defining ``set``.""" - @pytest.mark.parametrize( - ("backend_class", "max_ttl"), [(_CappedTTL, 60), (_OwnTracking, 300)], ids=["own-set", "own-track-key"] - ) - def test_sync_miss(self, client: redis.Redis, monkeypatch: pytest.MonkeyPatch, backend_class: Any, max_ttl: int) -> None: - backend = backend_class(client, "default") + 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 + + +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(key: str, value: Any, ttl: int | None = None) -> None: + backend.calls += 1 # type: ignore[attr-defined] + original(key, value, 60) if name == "set" else original(key, value) + + setattr(backend, name, hooked) + 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), +} + + +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): @@ -171,16 +212,18 @@ def test_sync_miss(self, client: redis.Redis, monkeypatch: pytest.MonkeyPatch, b assert all(max_ttl - 10 < client.ttl(f"t:default:{key.decode()}") <= max_ttl for key in _cache_keys(client)) @pytest.mark.asyncio - async def test_async_miss(self, client: redis.Redis) -> None: + @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 - backend = _CappedTTL(client, "default") + 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(50 < client.ttl(f"t:default:{key.decode()}") <= 60 for key in _cache_keys(client)) + assert all(max_ttl - 10 < client.ttl(f"t:default:{key.decode()}") <= max_ttl for key in _cache_keys(client)) class TestOneThreadHopPerAsyncTrackedWrite: From f2f8303e889176d58ed79def7261ebead5c2edf6 Mon Sep 17 00:00:00 2001 From: Ray Walker Date: Thu, 29 Oct 2026 17:28:35 +1100 Subject: [PATCH 07/12] fix(redis): a method the fuse owner inherits never matches When the class defining _set_tracked inherits set or track_key, its namespace holds neither, and a plain function bound on the instance has no __func__, so None matched None and the instance override was bypassed. Require the owner's own function. --- .secrets.baseline | 4 +- src/cachekit/cache_handler.py | 6 ++- tests/integration/test_tracked_write_redis.py | 40 ++++++++++++++----- 3 files changed, 38 insertions(+), 12 deletions(-) diff --git a/.secrets.baseline b/.secrets.baseline index 602e8cdc..40547782 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": 691 + "line_number": 695 } ], "src/cachekit/serializers/interop_serializer.py": [ @@ -2020,5 +2020,5 @@ } ] }, - "generated_at": "2026-10-29T06:14:57Z" + "generated_at": "2026-10-29T06:28:35Z" } diff --git a/src/cachekit/cache_handler.py b/src/cachekit/cache_handler.py index 337094b3..4bcebdb3 100644 --- a/src/cachekit/cache_handler.py +++ b/src/cachekit/cache_handler.py @@ -380,7 +380,11 @@ def _fuses_tracked_write(backend: object) -> bool: if owner is None or not supports_key_tracking(backend) or supports_swr(backend): # pyright: ignore[reportArgumentType] return False defined = vars(owner) - return all(getattr(getattr(backend, name, None), "__func__", None) is defined.get(name) for name in ("set", "track_key")) + for name in ("set", "track_key"): + func = defined.get(name) # None when the owner inherits it: no plain method of its own to match + if func is None or getattr(getattr(backend, name, None), "__func__", None) is not func: + return False + return True def _set_l2( diff --git a/tests/integration/test_tracked_write_redis.py b/tests/integration/test_tracked_write_redis.py index 8c02e64b..bc24e7a0 100644 --- a/tests/integration/test_tracked_write_redis.py +++ b/tests/integration/test_tracked_write_redis.py @@ -19,7 +19,7 @@ from cachekit import cache from cachekit.backends.errors import BackendError from cachekit.backends.redis.provider import PerRequestRedisBackend -from cachekit.cache_handler import StandardCacheHandler +from cachekit.cache_handler import StandardCacheHandler, _fuses_tracked_write from cachekit.decorators import write_behind as wb pytestmark = pytest.mark.integration @@ -170,17 +170,26 @@ def capped(key: str, value: bytes, ttl: int | None = None) -> None: return capped -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") +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(backend: PerRequestRedisBackend, name: str) -> PerRequestRedisBackend: + """``backend`` with its ``set`` (capped at 60 s) or ``track_key`` replaced on the instance only.""" backend.calls = 0 # type: ignore[attr-defined] original = getattr(backend, name) - def hooked(key: str, value: Any, ttl: int | None = None) -> None: + 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(key, value, 60) if name == "set" else original(key, value) + original(registry_id, key) - setattr(backend, name, hooked) + setattr(backend, name, hooked_set if name == "set" else hooked_track_key) return backend @@ -189,11 +198,24 @@ def hooked(key: str, value: Any, ttl: int | None = None) -> None: "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), + "instance-set": (lambda client: _instance_bound(PerRequestRedisBackend(client, "default"), "set"), 60), + "instance-track-key": (lambda client: _instance_bound(PerRequestRedisBackend(client, "default"), "track_key"), 300), } +class TestFuseDecision: + def test_plain_backend_fuses(self, client: redis.Redis) -> None: + assert _fuses_tracked_write(PerRequestRedisBackend(client, "default")) + + def test_owner_that_inherits_set_and_track_key_never_fuses(self, client: redis.Redis) -> None: + """Its own methods are None in its namespace, and plain functions bound on the instance have no __func__: + None must not match None.""" + backend = _OwnFused(client, "default") + backend.set = lambda *args: None # type: ignore[method-assign] + backend.track_key = lambda *args: None # type: ignore[method-assign] + assert not _fuses_tracked_write(backend) + + class TestOverridesKeepTheSeparateCalls: """Whatever way set or track_key is overridden, the override runs on every write: N calls, 2N round trips, its TTL kept.""" From 78ca00464b92226f9e522c25918b002ecae7769c Mon Sep 17 00:00:00 2001 From: Ray Walker Date: Thu, 29 Oct 2026 17:33:29 +1100 Subject: [PATCH 08/12] fix(redis): fuse only while set and track_key are the ones the class recorded A patch of PerRequestRedisBackend.set or track_key on the class itself (a test double, autospec, an app hook) replaced the very functions the guard compared against, so the fused write still ran and skipped the patch. The class now records the functions _set_tracked stands in for at creation. --- src/cachekit/backends/redis/provider.py | 5 ++++ src/cachekit/cache_handler.py | 12 +++++----- tests/integration/test_tracked_write_redis.py | 23 +++++++++++++++---- 3 files changed, 30 insertions(+), 10 deletions(-) diff --git a/src/cachekit/backends/redis/provider.py b/src/cachekit/backends/redis/provider.py index 44486310..037419d6 100644 --- a/src/cachekit/backends/redis/provider.py +++ b/src/cachekit/backends/redis/provider.py @@ -778,6 +778,11 @@ def _set_tracked(self, key: str, value: bytes, ttl: Optional[int], registry_id: finally: del error + # What _set_tracked stands in for, captured at class creation. The fused write runs only while the instance + # still resolves set and track_key to these (cache_handler._fuses_tracked_write), so a later override or patch + # of either one, on a subclass, the class or the instance, keeps the separate calls and runs. + _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 4bcebdb3..2f98c50e 100644 --- a/src/cachekit/cache_handler.py +++ b/src/cachekit/cache_handler.py @@ -371,17 +371,17 @@ 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 when the backend's own ``set`` and ``track_key``, as the instance resolves them, are the plain methods - of the class that defines ``_set_tracked``. An override anywhere (a subclass, an instance attribute, - ``__getattribute__``) 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. + Only while the instance resolves ``set`` and ``track_key`` to the functions the class that defines + ``_set_tracked`` recorded as standing in for (``_set_tracked_replaces``). An override or patch anywhere (a + subclass, the class itself, the instance, ``__getattribute__``) 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 - defined = vars(owner) + replaces = vars(owner).get("_set_tracked_replaces", {}) for name in ("set", "track_key"): - func = defined.get(name) # None when the owner inherits it: no plain method of its own to match + func = replaces.get(name) # None when the owner records nothing for it: nothing to match if func is None or getattr(getattr(backend, name, None), "__func__", None) is not func: return False return True diff --git a/tests/integration/test_tracked_write_redis.py b/tests/integration/test_tracked_write_redis.py index bc24e7a0..7be82474 100644 --- a/tests/integration/test_tracked_write_redis.py +++ b/tests/integration/test_tracked_write_redis.py @@ -12,6 +12,7 @@ import time from collections.abc import Callable, Iterator from typing import Any +from unittest.mock import patch import pytest import redis @@ -176,8 +177,9 @@ class _OwnFused(PerRequestRedisBackend): _set_tracked = PerRequestRedisBackend._set_tracked -def _instance_bound(backend: PerRequestRedisBackend, name: str) -> PerRequestRedisBackend: - """``backend`` with its ``set`` (capped at 60 s) or ``track_key`` replaced on the instance only.""" +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) @@ -198,8 +200,8 @@ def hooked_track_key(registry_id: str, key: str) -> None: "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(PerRequestRedisBackend(client, "default"), "set"), 60), - "instance-track-key": (lambda client: _instance_bound(PerRequestRedisBackend(client, "default"), "track_key"), 300), + "instance-set": (lambda client: _instance_bound(client, "set"), 60), + "instance-track-key": (lambda client: _instance_bound(client, "track_key"), 300), } @@ -215,6 +217,19 @@ def test_owner_that_inherits_set_and_track_key_never_fuses(self, client: redis.R 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_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, From cbc9a905f2de750e9c25ae7a3ba52666fbd9c147 Mon Sep 17 00:00:00 2001 From: Ray Walker Date: Thu, 29 Oct 2026 17:35:28 +1100 Subject: [PATCH 09/12] test(redis): name the owner-without-a-record case for what it checks --- .secrets.baseline | 4 ++-- src/cachekit/backends/redis/provider.py | 5 ++--- src/cachekit/cache_handler.py | 7 +++---- tests/integration/test_tracked_write_redis.py | 10 +++------- 4 files changed, 10 insertions(+), 16 deletions(-) diff --git a/.secrets.baseline b/.secrets.baseline index 40547782..791a39dc 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": 695 + "line_number": 694 } ], "src/cachekit/serializers/interop_serializer.py": [ @@ -2020,5 +2020,5 @@ } ] }, - "generated_at": "2026-10-29T06:28:35Z" + "generated_at": "2026-10-29T06:35:28Z" } diff --git a/src/cachekit/backends/redis/provider.py b/src/cachekit/backends/redis/provider.py index 037419d6..bfd69e74 100644 --- a/src/cachekit/backends/redis/provider.py +++ b/src/cachekit/backends/redis/provider.py @@ -778,9 +778,8 @@ def _set_tracked(self, key: str, value: bytes, ttl: Optional[int], registry_id: finally: del error - # What _set_tracked stands in for, captured at class creation. The fused write runs only while the instance - # still resolves set and track_key to these (cache_handler._fuses_tracked_write), so a later override or patch - # of either one, on a subclass, the class or the instance, keeps the separate calls and runs. + # 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]: diff --git a/src/cachekit/cache_handler.py b/src/cachekit/cache_handler.py index 2f98c50e..85deeb9c 100644 --- a/src/cachekit/cache_handler.py +++ b/src/cachekit/cache_handler.py @@ -371,10 +371,9 @@ 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 the class that defines - ``_set_tracked`` recorded as standing in for (``_set_tracked_replaces``). An override or patch anywhere (a - subclass, the class itself, the instance, ``__getattribute__``) 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. + Only while the instance resolves ``set`` and ``track_key`` to the functions in the owner's + ``_set_tracked_replaces``. An override or patch anywhere (a subclass, the class itself, the instance, + ``__getattribute__``) keeps both calls, so the override 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] diff --git a/tests/integration/test_tracked_write_redis.py b/tests/integration/test_tracked_write_redis.py index 7be82474..52d2f451 100644 --- a/tests/integration/test_tracked_write_redis.py +++ b/tests/integration/test_tracked_write_redis.py @@ -209,13 +209,9 @@ class TestFuseDecision: def test_plain_backend_fuses(self, client: redis.Redis) -> None: assert _fuses_tracked_write(PerRequestRedisBackend(client, "default")) - def test_owner_that_inherits_set_and_track_key_never_fuses(self, client: redis.Redis) -> None: - """Its own methods are None in its namespace, and plain functions bound on the instance have no __func__: - None must not match None.""" - backend = _OwnFused(client, "default") - backend.set = lambda *args: None # type: ignore[method-assign] - backend.track_key = lambda *args: None # type: ignore[method-assign] - assert not _fuses_tracked_write(backend) + 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.""" + assert not _fuses_tracked_write(_OwnFused(client, "default")) @pytest.mark.parametrize("name", ["set", "track_key"]) def test_class_level_patch_never_fuses(self, client: redis.Redis, monkeypatch: pytest.MonkeyPatch, name: str) -> None: From 62c6bdec838c395db02222a8623a1554aec8aabc Mon Sep 17 00:00:00 2001 From: Ray Walker Date: Thu, 29 Oct 2026 17:39:25 +1100 Subject: [PATCH 10/12] test(redis): the owner-without-a-record case exercises the missing-record clause --- .secrets.baseline | 4 ++-- src/cachekit/cache_handler.py | 3 ++- tests/integration/test_tracked_write_redis.py | 7 +++++-- 3 files changed, 9 insertions(+), 5 deletions(-) diff --git a/.secrets.baseline b/.secrets.baseline index 791a39dc..e998fc80 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": 694 + "line_number": 695 } ], "src/cachekit/serializers/interop_serializer.py": [ @@ -2020,5 +2020,5 @@ } ] }, - "generated_at": "2026-10-29T06:35:28Z" + "generated_at": "2026-10-29T06:39:25Z" } diff --git a/src/cachekit/cache_handler.py b/src/cachekit/cache_handler.py index 85deeb9c..bd52f990 100644 --- a/src/cachekit/cache_handler.py +++ b/src/cachekit/cache_handler.py @@ -373,7 +373,8 @@ def _fuses_tracked_write(backend: object) -> bool: Only while the instance resolves ``set`` and ``track_key`` to the functions in the owner's ``_set_tracked_replaces``. An override or patch anywhere (a subclass, the class itself, the instance, - ``__getattribute__``) keeps both calls, so the override still runs on every write. An SWR-capable backend keeps ``set`` too: the fused write has no slot for a stale-grace window. + ``__getattribute__``) keeps both calls, so the override 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] diff --git a/tests/integration/test_tracked_write_redis.py b/tests/integration/test_tracked_write_redis.py index 52d2f451..0bd8b69a 100644 --- a/tests/integration/test_tracked_write_redis.py +++ b/tests/integration/test_tracked_write_redis.py @@ -210,8 +210,11 @@ 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.""" - assert not _fuses_tracked_write(_OwnFused(client, "default")) + """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: From b0b40bd4a284e8b52eaeb02c73c2bf099cbf1982 Mon Sep 17 00:00:00 2001 From: Ray Walker Date: Thu, 29 Oct 2026 17:46:47 +1100 Subject: [PATCH 11/12] fix(redis): fuse only when set and track_key are bound to the backend itself The recorded functions bound to another backend (delegation by instance assignment or __getattribute__) passed the guard, so the fused write landed in this backend's namespace while reads went to the delegate's. --- .secrets.baseline | 4 +- src/cachekit/cache_handler.py | 11 +-- tests/integration/test_tracked_write_redis.py | 68 +++++++++++++++++++ 3 files changed, 77 insertions(+), 6 deletions(-) diff --git a/.secrets.baseline b/.secrets.baseline index e998fc80..60fc15b2 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": 695 + "line_number": 698 } ], "src/cachekit/serializers/interop_serializer.py": [ @@ -2020,5 +2020,5 @@ } ] }, - "generated_at": "2026-10-29T06:39:25Z" + "generated_at": "2026-10-29T06:46:47Z" } diff --git a/src/cachekit/cache_handler.py b/src/cachekit/cache_handler.py index bd52f990..6722e1c4 100644 --- a/src/cachekit/cache_handler.py +++ b/src/cachekit/cache_handler.py @@ -372,9 +372,10 @@ def _fuses_tracked_write(backend: object) -> bool: 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``. An override or patch anywhere (a subclass, the class itself, the instance, - ``__getattribute__``) keeps both calls, so the override still runs on every write. An SWR-capable backend - keeps ``set`` too: the fused write has no slot for a stale-grace window. + ``_set_tracked_replaces``, bound to the backend itself. An override, patch or delegation anywhere (a + subclass, the class itself, the instance, ``__getattribute__``) 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] @@ -382,7 +383,9 @@ def _fuses_tracked_write(backend: object) -> bool: 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 - if func is None or getattr(getattr(backend, name, None), "__func__", None) is not func: + bound = getattr(backend, name, None) + # The receiver too: the recorded function bound to ANOTHER backend (delegation) writes elsewhere. + if func is None or getattr(bound, "__func__", None) is not func or getattr(bound, "__self__", None) is not backend: return False return True diff --git a/tests/integration/test_tracked_write_redis.py b/tests/integration/test_tracked_write_redis.py index 0bd8b69a..c2ec91a6 100644 --- a/tests/integration/test_tracked_write_redis.py +++ b/tests/integration/test_tracked_write_redis.py @@ -205,6 +205,74 @@ def hooked_track_key(registry_id: str, key: str) -> None: } +_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:*") == [] + assert any(b":ck:reg:" not in key for key in client.keys("t:target:*")) + + @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")) From 9f37928a5195d0d3cf9ad3f9ee6194312b1e838e Mon Sep 17 00:00:00 2001 From: Ray Walker Date: Thu, 29 Oct 2026 17:52:18 +1100 Subject: [PATCH 12/12] fix(redis): fuse only for plain bound methods, never a forwarding proxy A wrapt-style instrumentation proxy forwards __func__ and __self__, so it passed the guard and its wrapper never ran. Require types.MethodType. --- .secrets.baseline | 4 +-- src/cachekit/cache_handler.py | 13 ++++---- tests/integration/test_tracked_write_redis.py | 32 ++++++++++++++++++- 3 files changed, 40 insertions(+), 9 deletions(-) diff --git a/.secrets.baseline b/.secrets.baseline index 60fc15b2..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": 698 + "line_number": 699 } ], "src/cachekit/serializers/interop_serializer.py": [ @@ -2020,5 +2020,5 @@ } ] }, - "generated_at": "2026-10-29T06:46:47Z" + "generated_at": "2026-10-29T06:52:18Z" } diff --git a/src/cachekit/cache_handler.py b/src/cachekit/cache_handler.py index 6722e1c4..d9aa0965 100644 --- a/src/cachekit/cache_handler.py +++ b/src/cachekit/cache_handler.py @@ -372,10 +372,10 @@ def _fuses_tracked_write(backend: object) -> bool: 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``, bound to the backend itself. An override, patch or delegation anywhere (a - subclass, the class itself, the instance, ``__getattribute__``) 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. + ``_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] @@ -384,8 +384,9 @@ def _fuses_tracked_write(backend: object) -> bool: 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) - # The receiver too: the recorded function bound to ANOTHER backend (delegation) writes elsewhere. - if func is None or getattr(bound, "__func__", None) is not func or getattr(bound, "__self__", None) is not backend: + 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 diff --git a/tests/integration/test_tracked_write_redis.py b/tests/integration/test_tracked_write_redis.py index c2ec91a6..70a1b1b2 100644 --- a/tests/integration/test_tracked_write_redis.py +++ b/tests/integration/test_tracked_write_redis.py @@ -254,7 +254,6 @@ def f(x: int) -> int: fn(1) assert calls == 1 assert client.keys("t:source:*") == [] - assert any(b":ck:reg:" not in key for key in client.keys("t:target:*")) @pytest.mark.asyncio @pytest.mark.parametrize("delegation", list(_DELEGATIONS)) @@ -291,6 +290,37 @@ def test_class_level_patch_never_fuses(self, client: redis.Redis, monkeypatch: p 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: