Optionally disable disconnects in read_response (#2695)

* Add regression tests and fixes for issue #1128

* Fix tests for resumable read_response to use "disconnect_on_error"

* undo prevision fix attempts in async client and cluster

* re-enable cluster test

* Suggestions from code review

* Add CHANGES

1 files changed, 18 insertions, 6 deletions

diff --git a/redis/connection.py b/redis/connection.py
index 8b2389c..5af8928 100644
--- a/redis/connection.py
+++ b/redis/connection.py
@@ -834,7 +834,11 @@ class AbstractConnection:
                 errno = e.args[0]
                 errmsg = e.args[1]
             raise ConnectionError(f"Error {errno} while writing to socket. {errmsg}.")
-        except Exception:
+        except BaseException:
+            # BaseExceptions can be raised when a socket send operation is not
+            # finished, e.g. due to a timeout.  Ideally, a caller could then re-try
+            # to send un-sent data. However, the send_packed_command() API
+            # does not support it so there is no point in keeping the connection open.
             self.disconnect()
             raise
 
@@ -859,7 +863,9 @@ class AbstractConnection:
             self.disconnect()
             raise ConnectionError(f"Error while reading from {host_error}: {e.args}")
 
-    def read_response(self, disable_decoding=False):
+    def read_response(
+        self, disable_decoding=False, *, disconnect_on_error: bool = True
+    ):
         """Read the response from a previously sent command"""
 
         host_error = self._host_error()
@@ -867,15 +873,21 @@ class AbstractConnection:
         try:
             response = self._parser.read_response(disable_decoding=disable_decoding)
         except socket.timeout:
-            self.disconnect()
+            if disconnect_on_error:
+                self.disconnect()
             raise TimeoutError(f"Timeout reading from {host_error}")
         except OSError as e:
-            self.disconnect()
+            if disconnect_on_error:
+                self.disconnect()
             raise ConnectionError(
                 f"Error while reading from {host_error}" f" : {e.args}"
             )
-        except Exception:
-            self.disconnect()
+        except BaseException:
+            # Also by default close in case of BaseException.  A lot of code
+            # relies on this behaviour when doing Command/Response pairs.
+            # See #1128.
+            if disconnect_on_error:
+                self.disconnect()
             raise
 
         if self.health_check_interval: