Don't wait for disconnect() when handling errors.

This can result in other errors such as timeouts.

Author: kristjanvalur

redis/asyncio/connection.py

@@ -839,7 +839,7 @@ async def on_connect(self) -> None:
             if str_if_bytes(await self.read_response()) != "OK":
                 raise ConnectionError("Invalid Database")
 
-    async def disconnect(self) -> None:
+    async def disconnect(self, nowait: bool = False) -> None:
         """Disconnects from the Redis server"""
         try:
             async with async_timeout.timeout(self.socket_connect_timeout):
@@ -849,8 +849,9 @@ async def disconnect(self) -> None:
                 try:
                     if os.getpid() == self.pid:
                         self._writer.close()  # type: ignore[union-attr]
-                        # py3.6 doesn't have this method
-                        if hasattr(self._writer, "wait_closed"):
+                        # wait for close to finish, except when handling errors and
+                        # forcecully disconnecting.
+                        if not nowait:
                             await self._writer.wait_closed()  # type: ignore[union-attr]
                 except OSError:
                     pass
@@ -905,10 +906,10 @@ async def send_packed_command(
                 self._writer.writelines(command)
                 await self._writer.drain()
         except asyncio.TimeoutError:
-            await self.disconnect()
+            await self.disconnect(nowait=True)
             raise TimeoutError("Timeout writing to socket") from None
         except OSError as e:
-            await self.disconnect()
+            await self.disconnect(nowait=True)
             if len(e.args) == 1:
                 err_no, errmsg = "UNKNOWN", e.args[0]
             else:
@@ -918,7 +919,7 @@ async def send_packed_command(
                 f"Error {err_no} while writing to socket. {errmsg}."
             ) from e
         except BaseException:
-            await self.disconnect()
+            await self.disconnect(nowait=True)
             raise
 
     async def send_command(self, *args: Any, **kwargs: Any) -> None:
@@ -934,7 +935,7 @@ async def can_read(self, timeout: float = 0):
         try:
             return await self._parser.can_read(timeout)
         except OSError as e:
-            await self.disconnect()
+            await self.disconnect(nowait=True)
             raise ConnectionError(
                 f"Error while reading from {self.host}:{self.port}: {e.args}"
             )
@@ -985,15 +986,15 @@ async def read_response_without_lock(self, disable_decoding: bool = False):
                     disable_decoding=disable_decoding
                 )
         except asyncio.TimeoutError:
-            await self.disconnect()
+            await self.disconnect(nowait=True)
             raise TimeoutError(f"Timeout reading from {self.host}:{self.port}")
         except OSError as e:
-            await self.disconnect()
+            await self.disconnect(nowait=True)
             raise ConnectionError(
                 f"Error while reading from {self.host}:{self.port} : {e.args}"
             )
         except BaseException:
-            await self.disconnect()
+            await self.disconnect(nowait=True)
             raise
 
         if self.health_check_interval: