Close cluster connection on failover

Author: vladvildanov

redis/asyncio/connection.py

@@ -212,6 +212,7 @@ def __init__(
         self._connect_callbacks: List[weakref.WeakMethod[ConnectCallbackT]] = []
         self._buffer_cutoff = 6000
         self._re_auth_token: Optional[TokenInterface] = None
+        self._should_reconnect = False
 
         try:
             p = int(protocol)
@@ -342,6 +343,12 @@ async def connect_check_health(
             if task and inspect.isawaitable(task):
                 await task
 
+    def mark_for_reconnect(self):
+        self._should_reconnect = True
+
+    def should_reconnect(self):
+        return self._should_reconnect
+
     @abstractmethod
     async def _connect(self):
         pass
@@ -1198,6 +1205,9 @@ async def release(self, connection: AbstractConnection):
         # Connections should always be returned to the correct pool,
         # not doing so is an error that will cause an exception here.
         self._in_use_connections.remove(connection)
+        if connection.should_reconnect():
+            await connection.disconnect()
+
         self._available_connections.append(connection)
         await self._event_dispatcher.dispatch_async(
             AsyncAfterConnectionReleasedEvent(connection)
@@ -1225,6 +1235,14 @@ async def disconnect(self, inuse_connections: bool = True):
         if exc:
             raise exc
 
+    async def update_active_connections_for_reconnect(self):
+        """
+        Mark all active connections for reconnect.
+        """
+        async with self._lock:
+            for conn in self._in_use_connections:
+                    conn.mark_for_reconnect()
+
     async def aclose(self) -> None:
         """Close the pool, disconnecting all connections"""
         await self.disconnect()

redis/asyncio/multidb/event.py

@@ -62,6 +62,7 @@ async def listen(self, event: AsyncActiveDatabaseChanged):
         await event.old_database.client.aclose()
 
         if isinstance(event.old_database.client, Redis):
+            await event.old_database.client.connection_pool.update_active_connections_for_reconnect()
             await event.old_database.client.connection_pool.disconnect()
 
 class RegisterCommandFailure(AsyncEventListenerInterface):

redis/multidb/event.py

@@ -65,6 +65,10 @@ def listen(self, event: ActiveDatabaseChanged):
         if isinstance(event.old_database.client, Redis):
             event.old_database.client.connection_pool.update_active_connections_for_reconnect()
             event.old_database.client.connection_pool.disconnect()
+        else:
+            for node in event.old_database.client.nodes_manager.nodes_cache.values():
+                node.redis_connection.connection_pool.update_active_connections_for_reconnect()
+                node.redis_connection.connection_pool.disconnect()
 
 class RegisterCommandFailure(EventListenerInterface):
     """

tests/test_asyncio/test_scenario/conftest.py

@@ -45,7 +45,7 @@ async def r_multi_db(request) -> AsyncGenerator[tuple[MultiDBClient, CheckActive
 
      # Retry configuration different for health checks as initial health check require more time in case
      # if infrastructure wasn't restored from the previous test.
-     health_check_interval = request.param.get('health_check_interval', DEFAULT_HEALTH_CHECK_INTERVAL)
+     health_check_interval = request.param.get('health_check_interval', 10)
      health_checks = request.param.get('health_checks', [])
      event_dispatcher = EventDispatcher()
      listener = CheckActiveDatabaseChangedListener()