Tighten async cache pool synchronization

Author: lh0156

redis/asyncio/connection.py

@@ -2979,7 +2979,10 @@ def __init__(
                     raise ValueError("Cache must implement CacheInterface")
                 self.cache = cache
             elif self._cache_factory is not None:
-                self.cache = CacheProxy(self._cache_factory.get_cache())
+                cache = self._cache_factory.get_cache()
+                self.cache = (
+                    cache if isinstance(cache, CacheProxy) else CacheProxy(cache)
+                )
             else:
                 self.cache = CacheFactory(
                     connection_kwargs.get("cache_config")
@@ -3442,7 +3445,10 @@ def set_in_maintenance(self, in_maintenance: bool) -> None:
     @contextlib.asynccontextmanager
     async def _maybe_pool_lock(self) -> AsyncIterator[None]:
         """Serialize pool mutations with cache-owner availability checks."""
-        async with self._lock:
+        if self.cache is not None or self._in_maintenance:
+            async with self._lock:
+                yield
+        else:
             yield
 
     @deprecated_args(

redis/connection.py

@@ -3021,7 +3021,10 @@ def __init__(
                 self.cache = cache
             else:
                 if self._cache_factory is not None:
-                    self.cache = CacheProxy(self._cache_factory.get_cache())
+                    cache = self._cache_factory.get_cache()
+                    self.cache = (
+                        cache if isinstance(cache, CacheProxy) else CacheProxy(cache)
+                    )
                 else:
                     self.cache = CacheFactory(
                         self._connection_kwargs.get("cache_config")

tests/test_asyncio/test_connection.py

@@ -31,6 +31,7 @@
     CacheEntryStatus,
     CacheFactory,
     CacheKey,
+    CacheProxy,
 )
 from redis.observability.attributes import CSCReason
 from redis.exceptions import ConnectionError, InvalidResponse, RedisError, TimeoutError
@@ -256,7 +257,9 @@ async def test_async_cache_proxy_signals_evicted_in_progress_fill():
 
 
 async def test_blocking_pool_serializes_cache_owner_checks():
-    pool = BlockingConnectionPool(max_connections=1)
+    pool = BlockingConnectionPool(
+        max_connections=1, protocol=3, cache_config=CacheConfig()
+    )
     entered = asyncio.Event()
 
     async def acquire_pool_lock():
@@ -272,6 +275,41 @@ async def acquire_pool_lock():
     await pool.aclose()
 
 
+async def test_blocking_pool_skips_lock_without_cache():
+    pool = BlockingConnectionPool(max_connections=1)
+    entered = asyncio.Event()
+
+    async def acquire_pool_lock():
+        async with pool._maybe_pool_lock():
+            entered.set()
+
+    async with pool._lock:
+        task = asyncio.create_task(acquire_pool_lock())
+        await asyncio.sleep(0)
+        assert entered.is_set() is True
+
+    await task
+    await pool.aclose()
+
+
+async def test_async_connection_pool_does_not_double_wrap_custom_cache_factory():
+    cache = CacheFactory(CacheConfig()).get_cache()
+    cache_factory = mock.Mock()
+    cache_factory.get_cache.return_value = cache
+    pool = ConnectionPool(
+        protocol=3,
+        cache_config=CacheConfig(),
+        cache_factory=cache_factory,
+    )
+
+    try:
+        assert pool.cache is cache
+        assert isinstance(pool.cache, CacheProxy)
+        assert not isinstance(pool.cache._cache, CacheProxy)
+    finally:
+        await pool.aclose()
+
+
 async def test_async_cache_proxy_waits_for_replacement_in_progress_fill():
     connection = mock.Mock()
     connection.can_read = mock.AsyncMock(return_value=False)

tests/test_connection.py

@@ -24,6 +24,7 @@
     CacheConfig,
     CacheEntry,
     CacheEntryStatus,
+    CacheFactory,
     CacheInterface,
     CacheKey,
     CacheProxy,
@@ -945,6 +946,22 @@ def test_creates_cache_with_custom_cache_factory(
         assert connection_pool.cache._cache == mock_cache
         connection_pool.disconnect()
 
+    def test_does_not_double_wrap_custom_cache_factory(self):
+        cache = CacheFactory(CacheConfig()).get_cache()
+        cache_factory = mock.Mock()
+        cache_factory.get_cache.return_value = cache
+
+        connection_pool = ConnectionPool(
+            protocol=3,
+            cache_config=CacheConfig(),
+            cache_factory=cache_factory,
+        )
+
+        assert connection_pool.cache is cache
+        assert isinstance(connection_pool.cache, CacheProxy)
+        assert not isinstance(connection_pool.cache._cache, CacheProxy)
+        connection_pool.disconnect()
+
     def test_creates_cache_with_given_configuration(self, mock_cache):
         connection_pool = ConnectionPool(
             protocol=3, cache_config=CacheConfig(max_size=100)