fix(csc): avoid draining checked-out cache connections

redis/_parsers/base.py

@@ -373,6 +373,9 @@ class PushNotificationsParser(Protocol):
     node_moving_push_handler_func: Optional[Callable] = None
     maintenance_push_handler_func: Optional[Callable] = None
     oss_cluster_maint_push_handler_func: Optional[Callable] = None
+    # Socket owners can postpone topology work without replacing the registered
+    # handler, which a concurrent pool configuration update may change.
+    _deferred_oss_cluster_handler: Optional[Callable] = None
 
     def handle_pubsub_push_response(self, response):
         """Handle pubsub push responses"""
@@ -432,8 +435,12 @@ def handle_push_response(self, response, **kwargs):
                 if notification is not None:
                     if self.maintenance_push_handler_func:
                         self.maintenance_push_handler_func(notification)
-                    if self.oss_cluster_maint_push_handler_func:
-                        self.oss_cluster_maint_push_handler_func(notification)
+                    handler = self.oss_cluster_maint_push_handler_func
+                    if handler:
+                        if self._deferred_oss_cluster_handler is not None:
+                            self._deferred_oss_cluster_handler(handler, notification)
+                        else:
+                            handler(notification)
         except Exception as e:
             logger.error(
                 "Error handling {} message ({}): {}".format(msg_type, response, e)

redis/asyncio/connection.py

@@ -3417,6 +3417,7 @@ async def re_auth_callback(self, token: TokenInterface):
                             error, AuthenticationError
                         ),
                     )
+                    conn.set_re_auth_token(None)
                 except AuthenticationError:
                     raise
                 except Exception as e:

redis/cache.py

@@ -369,40 +369,43 @@ def delete_by_cache_keys(self, cache_keys: list[CacheKey]) -> list[bool]:
         return response
 
     def delete_by_redis_keys(self, redis_keys: list[bytes] | list[str]) -> list[bool]:
-        response = []
-        keys_to_delete = []
-
-        for redis_key in redis_keys:
-            # Prepare both versions for lookup
-            candidates = [redis_key]
-            if isinstance(redis_key, str):
-                candidates.append(redis_key.encode("utf-8"))
-            elif isinstance(redis_key, bytes):
-                try:
-                    candidates.append(redis_key.decode("utf-8"))
-                except UnicodeDecodeError:
-                    pass  # Non-UTF-8 bytes, skip str version
-
-            # The reverse index answers this without walking the map. Both spellings are
-            # looked up because an entry is indexed under its keys exactly as the invocation
-            # supplied them, while the server names them in its own encoding. The two results
-            # are unioned, so an entry indexed under both spellings of this one key is
-            # collected once.
-            holders: set[CacheKey] = set()
-            for candidate in candidates:
-                holders |= self._cache.holders_of(candidate)
-
-            # An invalidation message never carries more than one key, so an entry holding
-            # several keys (MGET) cannot be collected twice by one call. A duplicate pop for
-            # a multi-key batch is not a reachable case - do not "fix" it.
-            for cache_key in holders:
-                keys_to_delete.append(cache_key)
-                response.append(True)
-
-        for key in keys_to_delete:
-            self._cache.pop(key)
+        # Independent owners can receive invalidations for the same Redis key.
+        # Keep the holder snapshots and their removal in one atomic operation.
+        with self._cache._lock:
+            response = []
+            keys_to_delete = []
+
+            for redis_key in redis_keys:
+                # Prepare both versions for lookup
+                candidates = [redis_key]
+                if isinstance(redis_key, str):
+                    candidates.append(redis_key.encode("utf-8"))
+                elif isinstance(redis_key, bytes):
+                    try:
+                        candidates.append(redis_key.decode("utf-8"))
+                    except UnicodeDecodeError:
+                        pass  # Non-UTF-8 bytes, skip str version
+
+                # The reverse index answers this without walking the map. Both spellings are
+                # looked up because an entry is indexed under its keys exactly as the invocation
+                # supplied them, while the server names them in its own encoding. The two results
+                # are unioned, so an entry indexed under both spellings of this one key is
+                # collected once.
+                holders: set[CacheKey] = set()
+                for candidate in candidates:
+                    holders |= self._cache.holders_of(candidate)
+
+                # An invalidation message never carries more than one key, so an entry holding
+                # several keys (MGET) cannot be collected twice by one call. A duplicate pop for
+                # a multi-key batch is not a reachable case - do not "fix" it.
+                for cache_key in holders:
+                    keys_to_delete.append(cache_key)
+                    response.append(True)
+
+            for key in keys_to_delete:
+                self._cache.pop(key)
 
-        return response
+            return response
 
     def flush(self) -> int:
         elem_count = len(self._cache)

redis/cluster.py

@@ -96,6 +96,7 @@
 from redis.maint_notifications import (
     MaintNotificationsConfig,
     OSSMaintNotificationsHandler,
+    _defer_maintenance_events,
 )
 from redis.observability.recorder import (
     record_error_count,
@@ -3044,7 +3045,7 @@ def _free_pool_connections(pool) -> None:
         closed once their current command completes, and disconnect its free
         connections.
         """
-        with pool._lock:
+        with _defer_maintenance_events(), pool._lock:
             pool.update_active_connections_for_reconnect()
             pool.disconnect_free_connections()
 
@@ -3425,6 +3426,7 @@ def initialize(
             additional_startup_nodes_info = []
 
         with (
+            _defer_maintenance_events(),
             self._initialization_lock,
             self._initializing_on_this_thread(),
             self._counting_failed_refresh(),