Add async client-side caching support - #4220
Conversation
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: e77bfb8c07
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| if is_cacheable and kwargs.get("keys") is None: | ||
| raise ValueError("Cannot create cache key.") |
There was a problem hiding this comment.
With the normal async client path, redis/asyncio/client.py::_send_command_parse_response still calls conn.send_command(*args) and drops **options, so high-level cacheable commands such as await r.get("k") never pass the keys=[...] metadata added by the command mixins. Because GET is cacheable by default, this new check sees no keys and raises ValueError("Cannot create cache key.") before any request is sent, making Redis(cache_config=...) fail for ordinary reads; the sync client forwards **options to send_command, so the async path needs the same behavior.
AGENTS.md reference: AGENTS.md:L145-L148
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Addressed in 00fd0ac. Async _send_command_parse_response now forwards command options, including cache keys, to the connection. Added a direct forwarding regression test; the asyncio connection/cache suites pass (106 passed, 41 skipped).
| if self._cache.get(self._current_command_cache_key) is not None: | ||
| return |
There was a problem hiding this comment.
With concurrent cache misses for the same key on different pooled connections, the first request leaves an IN_PROGRESS entry. This branch treats any surviving entry as usable and returns without sending the second command; if the first connection already has its reply, the drain above can even consume that response as a pending invalidation. The second caller then awaits read_response() on a connection that never sent a command, and the first caller may have lost its reply, so ordinary concurrent cached reads can time out or leave the entry stuck in progress.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Addressed in 00fd0ac. IN_PROGRESS cache entries now expose a completion event. Concurrent async cache misses wait for the original fill, then reuse its value; failed fills delete and signal the entry so waiters retry instead of hanging. Added success and failure concurrency tests.
| if data[1] is None: | ||
| self._cache.flush() | ||
| else: | ||
| self._cache.delete_by_redis_keys(data[1]) |
There was a problem hiding this comment.
When an async invalidation removes cached entries, this branch discards the delete_by_redis_keys() result and never records a client-side-cache eviction, while the sync proxy records invalidation evictions after deleting keys. Async clients will report hits, misses, network savings, and cache size, but invalidation-driven evictions are missing from metrics, making cache observability inaccurate for production async users.
AGENTS.md reference: AGENTS.md:L145-L148
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Addressed in 00fd0ac. Async invalidation now records CSCReason.INVALIDATION with the number of removed entries, matching the sync proxy behavior. Added a metric regression test.
|
Please make sure resolve Codex comments first |
00fd0ac to
468e0ca
Compare
![cursor[bot]](https://avatars.githubusercontent.com/in/1210556?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 468e0caf96
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| async with self._pool_lock: | ||
| while await cached_entry.connection_ref.can_read(): | ||
| try: | ||
| await cached_entry.connection_ref.read_response( | ||
| push_request=True, |
There was a problem hiding this comment.
In the current code this drain still runs for a VALID cache entry by reading from cached_entry.connection_ref, which can be a different raw connection that the pool has since handed to another coroutine; the pool lock is not held while borrowed connections wait for their own replies. If that coroutine's reply is buffered when a second connection serves this cache hit, this read_response(push_request=True, ...) consumes and discards the normal command reply as if it were an invalidation, leaving the real owner to time out or see a corrupted stream. Only drain the entry's connection when you can prove it is idle/currently owned, or process invalidations on the borrowed connection instead.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Addressed in db91b4b. Cache entries now retain enough pool context to verify that the owner connection is either the current connection or idle in the pool before draining it. A busy foreign connection is not read from; its entry is discarded and the command is refetched. Added a regression test proving the borrowed connection is not drained.
| ) | ||
| ) | ||
|
|
||
| await self._conn.send_command(*args, **kwargs) |
There was a problem hiding this comment.
After setting the shared cache entry to IN_PROGRESS, this await can exit via asyncio.CancelledError while the underlying connection disconnects in its BaseException handler; because the proxy only deletes/signals the entry in read_response(), no completion event is set for this key. A later request for the same cache key then waits forever at the in-flight fill path instead of retrying, which is easy to hit when callers wrap Redis operations in asyncio.wait_for or cancel tasks.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Addressed in db91b4b. If the underlying async send exits with cancellation or another BaseException after an IN_PROGRESS entry is installed, the entry is deleted and its completion event is signaled. Added coverage that a cancelled fill can be retried immediately.
| self._current_command_cache_key = CacheKey( | ||
| command=args[0], | ||
| redis_keys=tuple(kwargs["keys"]), | ||
| redis_args=args, |
There was a problem hiding this comment.
When a cacheable command passes a single key as a scalar, this conversion splits it into characters/integers instead of storing the Redis key; for example zrevrange() sets options["keys"] = name, so name="myzset" becomes ('m', 'y', ...) (or byte integers). Redis invalidations arrive for b"myzset", which will not match that tuple, so async caching can keep returning stale ZREVRANGE results after the sorted set is modified. Normalize scalar str/bytes keys to a one-item tuple before building CacheKey.
AGENTS.md reference: AGENTS.md:L158-L162
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Addressed in db91b4b. Scalar str/bytes cache keys are normalized to a one-item tuple before CacheKey creation, so commands such as ZREVRANGE do not split a key into characters or byte integers. Added a regression test.
| if (cache_config or cache) and check_protocol_version(protocol, 3): | ||
| kwargs.update( | ||
| { | ||
| "cache": cache, | ||
| "cache_config": cache_config, | ||
| } | ||
| ) |
There was a problem hiding this comment.
Because this cache wiring is nested only under the TCP branch, Redis(unix_socket_path=..., protocol=3, cache_config=...) silently constructs a pool with cache is None even though the user explicitly requested client-side caching and the RESP3 validation below still succeeds. Direct async ConnectionPool(path=..., connection_class=UnixDomainSocketConnection, cache_config=...) does create the proxy, so the high-level async constructor should pass the cache/cache_config kwargs for Unix sockets as well or reject the combination instead of disabling caching.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Addressed in db91b4b. Cache/cache_config wiring now happens after the Unix-socket/TCP branch, so Redis(unix_socket_path=..., protocol=3, cache_config=...) creates a cache-enabled pool. Added a constructor regression test.
| await self._conn.read_response( | ||
| push_request=True, | ||
| timeout=0, | ||
| disconnect_on_error=False, |
There was a problem hiding this comment.
When a pending invalidation frame is only partially buffered, async read_response(timeout=0) returns None on the user-requested timeout rather than raising TimeoutError, but this loop only breaks on the exception. Because can_read() remains true while the partial bytes stay buffered, a cached async client can spin here and starve the event loop before the rest of the frame arrives; treat a None result like no complete push was available.
AGENTS.md reference: AGENTS.md:L145-L148
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Addressed in db91b4b. Shared invalidation draining now stops when timeout=0 returns None as well as when it raises TimeoutError, preventing a busy loop on partially buffered frames. Added a regression test.
|
Follow-up fixes are in db91b4b. This revision addresses the post-rebase cache review feedback: packed-command key reset, safe idle-owner invalidation draining, cancelled fill cleanup, scalar cache-key normalization, Unix-socket cache configuration, and zero-timeout invalidation reads. Focused async connection/cache tests pass (112 passed, 41 skipped); ruff check and format validation pass. |
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: db91b4b165
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| async with self._cache_lock: | ||
| self._cache.set(cache_entry) |
There was a problem hiding this comment.
When max_size is smaller than the number of concurrent cache misses, this set() can evict an existing IN_PROGRESS entry via the LRU policy without setting its completion_event. A coroutine that already saw that entry and is awaiting completion_event.wait() will never resume; the original filler also won't signal later because read_response() finds the entry missing. This can hang ordinary concurrent cached reads with small caches, e.g. two tasks waiting on GET k1 while another miss for k2 evicts the k1 placeholder.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Addressed in 6d81ecf. CacheProxy.set now captures the LRU entry selected for eviction and signals its completion_event after eviction, so async waiters cannot remain blocked on an evicted fill. Added test_async_cache_proxy_signals_evicted_in_progress_fill. The focused async connection and cache suites pass: 115 passed, 41 skipped.
| async with self._pool_lock: | ||
| is_current_connection = cached_entry.connection_ref is self._conn | ||
| is_available_connection = ( | ||
| self._pool is None | ||
| or is_current_connection | ||
| or self._pool._is_connection_available(cached_entry.connection_ref) | ||
| ) |
There was a problem hiding this comment.
This lock only protects the normal ConnectionPool; BlockingConnectionPool.get_connection() skips _lock outside maintenance mode, so a cached hit can decide the cached entry's connection is idle and then await _drain_pending_invalidations() while another task borrows that same connection from the blocking pool. In that context a concurrent command reply can be consumed as a push invalidation, recreating the stream desynchronization this availability check is trying to prevent.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Addressed in 6d81ecf. BlockingConnectionPool._maybe_pool_lock now always serializes pool mutations with the cache owner availability check, including the normal get path. Added test_blocking_pool_serializes_cache_owner_checks to cover the previously unprotected interleaving.
| async with self._cache_lock: | ||
| if self._cache.get(self._current_command_cache_key) is not None: | ||
| return |
There was a problem hiding this comment.
After a failed or invalidated fill wakes waiters, another coroutine can install a new IN_PROGRESS entry for the same cache key before this waiter reacquires the lock. This branch treats any non-missing entry as reusable and returns without sending the command; the following read_response() then records a miss and waits on a connection that never sent a request. Check that the entry is valid, or loop back to wait for the replacement fill instead of returning on a fresh in-progress placeholder.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Addressed in 6d81ecf. After an in-progress fill completes, send_command now re-reads the cache and returns only when the current entry is VALID. If the entry was replaced by another in-progress fill, it loops and waits for that replacement instead of returning with no response pending. Added test_async_cache_proxy_waits_for_replacement_in_progress_fill.
There was a problem hiding this comment.
Cursor Bugbot has reviewed your changes using default effort and found 1 potential issue.
Reviewed by Cursor Bugbot for commit b944df4. Configure here.
| self._cache = cache | ||
| self._pool_lock = pool_lock | ||
| self._pool = pool | ||
| self._cache_lock = asyncio.Lock() |
There was a problem hiding this comment.
Per-proxy cache lock doesn't protect shared cache
Medium Severity
Each AsyncCacheProxyConnection creates its own _cache_lock = asyncio.Lock(), but all proxies from the same pool share the same _cache instance. This means two proxies can independently acquire their own locks and concurrently read/modify the shared cache at interleaving await points, leading to race conditions in cache state management (e.g., TOCTOU issues between checking and writing entries).
Additional Locations (1)
Reviewed by Cursor Bugbot for commit b944df4. Configure here.
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: b944df45f1
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| await self._conn.send_packed_command(command, check_health=check_health) | ||
|
|
||
| async def send_command(self, *args, **kwargs) -> None: | ||
| await self._process_pending_invalidations() |
There was a problem hiding this comment.
When a cache-enabled Redis instance creates a PubSub, that connection is wrapped in AsyncCacheProxyConnection too. PubSub.execute_command intentionally avoids parsing pending responses because a subscribed connection can already have messages queued (redis/asyncio/client.py:1289-1291), but this unconditional drain runs before every SUBSCRIBE/UNSUBSCRIBE/PING and read_response(push_request=True) returns ordinary Pub/Sub pushes as well as invalidations. With a pending message, the drain consumes and discards it before PubSub.parse_response() can deliver it, so cache_config can silently lose Pub/Sub messages; restrict the drain to non-PubSub connections or preserve non-invalidation pushes.
Useful? React with 👍 / 👎.
| if is_cacheable and kwargs.get("keys") is None: | ||
| raise ValueError("Cannot create cache key.") |
There was a problem hiding this comment.
When caching is enabled, xpending_range() sends the cache-allowlisted XPENDING command with parse_detail=True but no keys option (redis/commands/core.py:7799). This new guard raises ValueError before sending, so await r.xpending_range(...) stops working on a cache-enabled async client even though it is a read-only Redis command; add key metadata for that command path and keep the sync mirror aligned, or skip caching when metadata is unavailable.
AGENTS.md reference: AGENTS.md:L145-L148
Useful? React with 👍 / 👎.
| keys_deleted = 0 | ||
| else: | ||
| entries = tuple(self._cache.collection.values()) | ||
| keys_deleted = len(self._cache.delete_by_redis_keys(data[1])) |
There was a problem hiding this comment.
When Redis sends one invalidation push containing both keys from a multi-key cache entry, for example an in-progress MGET k1 k2 invalidated by MSET k1 k2, DefaultCache.delete_by_redis_keys() queues the same CacheKey once per matched Redis key and raises KeyError on the second pop. Because this new async callback calls it before signaling completion events, that exception prevents waiters on the in-progress entry from being awakened, and the filler later finds the entry gone and also will not signal; dedupe the matched cache keys or signal waiters in a finally path.
Useful? React with 👍 / 👎.
| cache_entry = self._cache.get(cache_key) | ||
| if cache_entry is not None: | ||
| cache_entry.status = CacheEntryStatus.VALID | ||
| cache_entry.cache_value = response |
There was a problem hiding this comment.
For cacheable commands whose response callback returns the raw mutable container, such as zrange(..., withscores=False), this stores the exact object that is returned to the first caller. If that caller mutates the list or dict it received, the cached value is mutated too, so later cache hits return caller-added data that was never in Redis; store a defensive copy here (and mirror the same fix in the sync cache path) before exposing the response.
AGENTS.md reference: AGENTS.md:L145-L148
Useful? React with 👍 / 👎.
| if self.cache is not None: | ||
| return AsyncCacheProxyConnection( | ||
| connection, self.cache, self._lock, pool=self | ||
| ) |
There was a problem hiding this comment.
When maintenance notifications are disabled (for example Unix-socket clients force them off), these wrapped cache connections can legitimately have pending RESP3 invalidation pushes while idle in the pool. ensure_connection() below still treats any readable data as a dirty connection because the async pool lacks the sync pool's self.cache is None guard, so the next borrow disconnects and flushes the whole cache instead of letting send_command() drain the invalidation; skip that dirty-read reconnect for cache-enabled pools as the sync implementation does.
AGENTS.md reference: AGENTS.md:L145-L148
Useful? React with 👍 / 👎.
| if response is None: | ||
| return False |
There was a problem hiding this comment.
read_response(push_request=True) returns None not only for a zero-timeout partial read, but also after a handled invalidation because _on_invalidation_callback() has no return value. This means a connection with queued invalidations for k1 and then k2 stops after processing k1; a following cached GET k2 can still hit the stale entry because the second push remains buffered. Continue the loop after a handled invalidation and only stop on the timeout/no-complete-frame case, matching the sync drain behavior.
AGENTS.md reference: AGENTS.md:L145-L148
Useful? React with 👍 / 👎.
| async with self._cache_lock: | ||
| self._cache.set(cache_entry) |
There was a problem hiding this comment.
The cache lookup above and this placeholder insertion are separated by awaits and not protected by a shared cache-level critical section, so two concurrent misses for the same key on different connections can both send the command and the later placeholder can replace the earlier one. If the first response then marks the later placeholder valid, the second caller returns that cached value without reading its own already-sent reply, leaving the socket dirty for the next borrower; re-read the cache immediately before setting the IN_PROGRESS entry and wait/retry if another filler won the race.
Useful? React with 👍 / 👎.
2 participants
Fixes #3916
What changed
The synchronous client already supports RESP3 client-side caching, but the async standalone client did not expose cache or cache_config and its async connection pool could not wrap connections for tracking and invalidation handling.
This change:
Verification
Note
Medium Risk
Touches the hot async command path and connection pooling with cache invalidation and concurrency; mistakes could cause stale reads or pool deadlocks, though behavior is heavily tested and gated on RESP3 and Redis 7.4+.
Overview
Adds client-side caching (CSC) to the async standalone
redis.asyncioclient, matching behavior already available on the sync client.Redisnow acceptscache/cache_config(RESP3-only, Redis 7.4+), exposesget_cache(), and forwards commandkeys(and other options) through to connections so cache keys can be built. The async connection pool constructs the shared cache, wraps connections in a newAsyncCacheProxyConnection(CLIENT TRACKING, invalidation push handling, hit/miss/eviction metrics), and uses the pool lock when CSC is enabled so in-flight fills and connection availability stay consistent.Shared cache types gain
completion_eventonCacheEntryso concurrent async callers can wait on in-progress fills;CacheProxysignals that event on LRU eviction. Sync pool wiring avoids double-wrapping when a factory already returns aCacheProxy.Docs in
resp3_features.rstdocument async CSC usage; tests cover proxy behavior, RESP2 rejection, and integration round-trips.Reviewed by Cursor Bugbot for commit b944df4. Bugbot is set up for automated code reviews on this repo. Configure here.