perf(redis): send the key-registry SADD+EXPIRE in the same pipeline as the SET (LAB-7115)

docs/backends/redis.md

@@ -82,7 +82,7 @@ Unset, a GIL build keeps hiredis, and so does a free-threaded build whose GIL is
 
 ## Characteristics
 
-- Network latency: one round trip per operation, which depends on your network and region ([not measured here](../performance.md#not-measured-here))
+- Network latency: one round trip per operation, which depends on your network and region ([not measured here](../performance.md#not-measured-here)). On a backend that keeps a [key registry](../features/l1-invalidation.md#whole-function-invalidation) (resolved from the URL or `RedisBackendProvider`), a decorated write's registry entry rides in the same pipeline as its `SET`, so the write is still one round trip
 - Automatic TTL support (Redis `SET … EX`)
 - Connection pooling built-in
 - Supports large values (up to Redis limits)

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.

src/cachekit/backends/redis/provider.py

@@ -50,6 +50,17 @@
 
 # Key registry (KeyTrackableBackend). A tracking set lives 7 days past its last write.
 _TRACKING_SET_TTL_SECONDS = 604_800
+
+
+def _registry_add(pipeline: Any, scoped_reg: str, key: str) -> Any:
+    """Queue a key's registry entry on ``pipeline``: ``SADD`` the raw key, then ``EXPIRE`` the set to 7 days.
+
+    The one place the registry's write commands live: ``track_key`` and ``_set_tracked`` both queue them here.
+    Queuing sends nothing; the caller chains ``execute()`` on the result.
+    """
+    return pipeline.sadd(scoped_reg, key).expire(scoped_reg, _TRACKING_SET_TTL_SECONDS)
+
+
 # Members popped (and keys unlinked) per script call. Bounds each atomic step well under
 # lua-time-limit: a script that has written cannot be SCRIPT KILLed, so unbounded work in
 # one call can leave every other client on this Redis answered with BUSY.
@@ -706,11 +717,8 @@ def track_key(self, registry_id: str, key: str) -> None:
         scoped_reg = self._scoped_key(registry_id)
         with _ClearedOnInterrupt():
             try:
-                # Chained, never bound to a local: a pipeline's repr lists the password. A pipelined
-                # command returns its pipeline; redis-py's annotations give the reply type instead.
-                self._client.pipeline(transaction=False).sadd(scoped_reg, key).expire(  # pyright: ignore[reportAttributeAccessIssue]
-                    scoped_reg, _TRACKING_SET_TTL_SECONDS
-                ).execute()
+                # Chained, never bound to a local: a pipeline's repr lists the password.
+                _registry_add(self._client.pipeline(transaction=False), scoped_reg, key).execute()
                 return
             except Exception as exc:
                 error = classify_redis_error(exc, operation="track_key", key=registry_id)
@@ -719,6 +727,61 @@ def track_key(self, registry_id: str, key: str) -> None:
         finally:
             del error
 
+    def _set_tracked(self, key: str, value: bytes, ttl: Optional[int], registry_id: str) -> Optional[BackendError]:
+        """``set`` and ``track_key`` in one round trip: ``SET``, ``SADD`` and ``EXPIRE`` on one
+        non-transactional pipeline.
+
+        Private: ``StandardCacheHandler.set_tracked`` finds it on the backend class, and a backend without it
+        keeps the separate ``track_key`` call. The write's success is the ``SET`` reply alone. The
+        registry commands run even when the ``SET`` fails (no transaction rolls a failed ``SET`` back),
+        which leaves a member with no key: the next drain unlinks the missing key, and the set's 7-day
+        expiry bounds it until then.
+
+        Args:
+            key: Cache key to store (will be tenant-scoped), as for ``set``
+            value: Bytes value to store
+            ttl: Time-to-live in seconds (None or <= 0 = no expiry), as for ``set``
+            registry_id: Unscoped registry id, as for ``track_key``
+
+        Returns:
+            None when the write and its tracking both succeeded; the classified tracking error when
+            the ``SET`` succeeded and ``SADD`` or ``EXPIRE`` did not. Tracking is advisory, so that
+            error is the caller's to report, never to raise.
+
+        Raises:
+            BackendError: If the ``SET`` failed, or the pipeline did (a dropped connection: whether
+                the ``SET`` landed is unknown, as for ``set``)
+        """
+        scoped_key = self._scoped_key(key)
+        scoped_reg = self._scoped_key(registry_id)
+        with _ClearedOnInterrupt():
+            try:
+                # Chained, never bound to a local: a pipeline's repr lists the password. Error replies come
+                # back in place, so each command's outcome is read on its own.
+                replies = _registry_add(
+                    self._client.pipeline(transaction=False).set(
+                        scoped_key, value, ex=ttl if ttl is not None and ttl > 0 else None
+                    ),
+                    scoped_reg,
+                    key,
+                ).execute(raise_on_error=False)
+            except Exception as exc:
+                error = classify_redis_error(exc, operation="set", key=key)
+            else:
+                if not isinstance(replies[0], Exception):
+                    failed = next((reply for reply in replies[1:] if isinstance(reply, Exception)), None)
+                    return None if failed is None else classify_redis_error(failed, operation="track_key", key=registry_id)
+                error = classify_redis_error(replies[0], operation="set", key=key)
+                replies = None  # its raw redis-py error replies must not stay in a frame on the raised error's traceback
+        try:
+            raise error from error.original_exception
+        finally:
+            del error
+
+    # The set and track_key that _set_tracked stands in for, captured at class creation; see
+    # cache_handler._fuses_tracked_write.
+    _set_tracked_replaces = {"set": set, "track_key": track_key}
+
     def drain_tracked(self, registry_id: str, local_keys: Iterable[str]) -> set[str]:
         """Delete every tracked key of a registry, then every local key the drain missed
         (KeyTrackableBackend protocol).