task: metadata: install: pip install -e .[dev]
patch: diff --git a/CHANGES b/CHANGES
index 3865ed1..ea171f6 100644
--- a/CHANGES
+++ b/CHANGES
@@ -1,3 +1,4 @@
+    * Revert #2104, #2673, add `disconnect_on_error` option to `read_response()` (issues #2506, #2624)
     * Add `address_remap` parameter to `RedisCluster`
     * Fix incorrect usage of once flag in async Sentinel
     * asyncio: Fix memory leak caused by hiredis (#2693)
diff --git a/redis/asyncio/client.py b/redis/asyncio/client.py
index 3e6626a..7479b74 100644
--- a/redis/asyncio/client.py
+++ b/redis/asyncio/client.py
@@ -139,8 +139,12 @@ class Redis(
         arguments always win.
 
         """
+        single_connection_client = kwargs.pop("single_connection_client", False)
         connection_pool = ConnectionPool.from_url(url, **kwargs)
-        return cls(connection_pool=connection_pool)
+        return cls(
+            connection_pool=connection_pool,
+            single_connection_client=single_connection_client,
+        )
 
     def __init__(
         self,
@@ -455,7 +459,7 @@ class Redis(
     _DEL_MESSAGE = "Unclosed Redis client"
 
     def __del__(self, _warnings: Any = warnings) -> None:
-        if self.connection is not None:
+        if hasattr(self, "connection") and (self.connection is not None):
             _warnings.warn(
                 f"Unclosed client session {self!r}", ResourceWarning, source=self
             )
@@ -500,23 +504,6 @@ class Redis(
         ):
             raise error
 
-    async def _try_send_command_parse_response(self, conn, *args, **options):
-        try:
-            return await conn.retry.call_with_retry(
-                lambda: self._send_command_parse_response(
-                    conn, args[0], *args, **options
-                ),
-                lambda error: self._disconnect_raise(conn, error),
-            )
-        except asyncio.CancelledError:
-            await conn.disconnect(nowait=True)
-            raise
-        finally:
-            if self.single_connection_client:
-                self._single_conn_lock.release()
-            if not self.connection:
-                await self.connection_pool.release(conn)
-
     # COMMAND EXECUTION AND PROTOCOL PARSING
     async def execute_command(self, *args, **options):
         """Execute a command and return a parsed response"""
@@ -527,10 +514,18 @@ class Redis(
 
         if self.single_connection_client:
             await self._single_conn_lock.acquire()
-
-        return await asyncio.shield(
-            self._try_send_command_parse_response(conn, *args, **options)
-        )
+        try:
+            return await conn.retry.call_with_retry(
+                lambda: self._send_command_parse_response(
+                    conn, command_name, *args, **options
+                ),
+                lambda error: self._disconnect_raise(conn, error),
+            )
+        finally:
+            if self.single_connection_client:
+                self._single_conn_lock.release()
+            if not self.connection:
+                await pool.release(conn)
 
     async def parse_response(
         self, connection: Connection, command_name: Union[str, bytes], **options
@@ -774,18 +769,10 @@ class PubSub:
         is not a TimeoutError. Otherwise, try to reconnect
         """
         await conn.disconnect()
-
         if not (conn.retry_on_timeout and isinstance(error, TimeoutError)):
             raise error
         await conn.connect()
 
-    async def _try_execute(self, conn, command, *arg, **kwargs):
-        try:
-            return await command(*arg, **kwargs)
-        except asyncio.CancelledError:
-            await conn.disconnect()
-            raise
-
     async def _execute(self, conn, command, *args, **kwargs):
         """
         Connect manually upon disconnection. If the Redis server is down,
@@ -794,11 +781,9 @@ class PubSub:
         called by the # connection to resubscribe us to any channels and
         patterns we were previously listening to
         """
-        return await asyncio.shield(
-            conn.retry.call_with_retry(
-                lambda: self._try_execute(conn, command, *args, **kwargs),
-                lambda error: self._disconnect_raise_connect(conn, error),
-            )
+        return await conn.retry.call_with_retry(
+            lambda: command(*args, **kwargs),
+            lambda error: self._disconnect_raise_connect(conn, error),
         )
 
     async def parse_response(self, block: bool = True, timeout: float = 0):
@@ -816,7 +801,9 @@ class PubSub:
             await conn.connect()
 
         read_timeout = None if block else timeout
-        response = await self._execute(conn, conn.read_response, timeout=read_timeout)
+        response = await self._execute(
+            conn, conn.read_response, timeout=read_timeout, disconnect_on_error=False
+        )
 
         if conn.health_check_interval and response == self.health_check_response:
             # ignore the health check message as user might not expect it
@@ -1200,18 +1187,6 @@ class Pipeline(Redis):  # lgtm [py/init-calls-subclass]
             await self.reset()
             raise
 
-    async def _try_send_command_parse_response(self, conn, *args, **options):
-        try:
-            return await conn.retry.call_with_retry(
-                lambda: self._send_command_parse_response(
-                    conn, args[0], *args, **options
-                ),
-                lambda error: self._disconnect_reset_raise(conn, error),
-            )
-        except asyncio.CancelledError:
-            await conn.disconnect()
-            raise
-
     async def immediate_execute_command(self, *args, **options):
         """
         Execute a command immediately, but don't auto-retry on a
@@ -1227,8 +1202,12 @@ class Pipeline(Redis):  # lgtm [py/init-calls-subclass]
                 command_name, self.shard_hint
             )
             self.connection = conn
-        return await asyncio.shield(
-            self._try_send_command_parse_response(conn, *args, **options)
+
+        return await conn.retry.call_with_retry(
+            lambda: self._send_command_parse_response(
+                conn, command_name, *args, **options
+            ),
+            lambda error: self._disconnect_reset_raise(conn, error),
         )
 
     def pipeline_execute_command(self, *args, **options):
@@ -1396,19 +1375,6 @@ class Pipeline(Redis):  # lgtm [py/init-calls-subclass]
             await self.reset()
             raise
 
-    async def _try_execute(self, conn, execute, stack, raise_on_error):
-        try:
-            return await conn.retry.call_with_retry(
-                lambda: execute(conn, stack, raise_on_error),
-                lambda error: self._disconnect_raise_reset(conn, error),
-            )
-        except asyncio.CancelledError:
-            # not supposed to be possible, yet here we are
-            await conn.disconnect(nowait=True)
-            raise
-        finally:
-            await self.reset()
-
     async def execute(self, raise_on_error: bool = True):
         """Execute all the commands in the current pipeline"""
         stack = self.command_stack
@@ -1430,11 +1396,10 @@ class Pipeline(Redis):  # lgtm [py/init-calls-subclass]
         conn = cast(Connection, conn)
 
         try:
-            return await asyncio.shield(
-                self._try_execute(conn, execute, stack, raise_on_error)
+            return await conn.retry.call_with_retry(
+                lambda: execute(conn, stack, raise_on_error),
+                lambda error: self._disconnect_raise_reset(conn, error),
             )
-        except RuntimeError:
-            await self.reset()
         finally:
             await self.reset()
 
diff --git a/redis/asyncio/cluster.py b/redis/asyncio/cluster.py
index eb5f4db..929d3e4 100644
--- a/redis/asyncio/cluster.py
+++ b/redis/asyncio/cluster.py
@@ -1016,33 +1016,12 @@ class ClusterNode:
         await connection.send_packed_command(connection.pack_command(*args), False)
 
         # Read response
-        return await asyncio.shield(
-            self._parse_and_release(connection, args[0], **kwargs)
-        )
-
-    async def _parse_and_release(self, connection, *args, **kwargs):
         try:
-            return await self.parse_response(connection, *args, **kwargs)
-        except asyncio.CancelledError:
-            # should not be possible
-            await connection.disconnect(nowait=True)
-            raise
+            return await self.parse_response(connection, args[0], **kwargs)
         finally:
+            # Release connection
             self._free.append(connection)
 
-    async def _try_parse_response(self, cmd, connection, ret):
-        try:
-            cmd.result = await asyncio.shield(
-                self.parse_response(connection, cmd.args[0], **cmd.kwargs)
-            )
-        except asyncio.CancelledError:
-            await connection.disconnect(nowait=True)
-            raise
-        except Exception as e:
-            cmd.result = e
-            ret = True
-        return ret
-
     async def execute_pipeline(self, commands: List["PipelineCommand"]) -> bool:
         # Acquire connection
         connection = self.acquire_connection()
@@ -1055,7 +1034,13 @@ class ClusterNode:
         # Read responses
         ret = False
         for cmd in commands:
-            ret = await asyncio.shield(self._try_parse_response(cmd, connection, ret))
+            try:
+                cmd.result = await self.parse_response(
+                    connection, cmd.args[0], **cmd.kwargs
+                )
+            except Exception as e:
+                cmd.result = e
+                ret = True
 
         # Release connection
         self._free.append(connection)
diff --git a/redis/asyncio/connection.py b/redis/asyncio/connection.py
index 59f75aa..462673f 100644
--- a/redis/asyncio/connection.py
+++ b/redis/asyncio/connection.py
@@ -804,7 +804,11 @@ class Connection:
             raise ConnectionError(
                 f"Error {err_no} while writing to socket. {errmsg}."
             ) from e
-        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.
             await self.disconnect(nowait=True)
             raise
 
@@ -828,6 +832,8 @@ class Connection:
         self,
         disable_decoding: bool = False,
         timeout: Optional[float] = None,
+        *,
+        disconnect_on_error: bool = True,
     ):
         """Read the response from a previously sent command"""
         read_timeout = timeout if timeout is not None else self.socket_timeout
@@ -843,22 +849,24 @@ class Connection:
                 )
         except asyncio.TimeoutError:
             if timeout is not None:
-                # user requested timeout, return None
+                # user requested timeout, return None. Operation can be retried
                 return None
             # it was a self.socket_timeout error.
-            await self.disconnect(nowait=True)
+            if disconnect_on_error:
+                await self.disconnect(nowait=True)
             raise TimeoutError(f"Timeout reading from {self.host}:{self.port}")
         except OSError as e:
-            await self.disconnect(nowait=True)
+            if disconnect_on_error:
+                await self.disconnect(nowait=True)
             raise ConnectionError(
                 f"Error while reading from {self.host}:{self.port} : {e.args}"
             )
-        except asyncio.CancelledError:
-            # need this check for 3.7, where CancelledError
-            # is subclass of Exception, not BaseException
-            raise
-        except Exception:
-            await self.disconnect(nowait=True)
+        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:
+                await self.disconnect(nowait=True)
             raise
 
         if self.health_check_interval:
diff --git a/redis/client.py b/redis/client.py
index 79a7bff..9fd5b7c 100755
--- a/redis/client.py
+++ b/redis/client.py
@@ -420,9 +420,13 @@ def parse_slowlog_get(response, **options):
         # an O(N) complexity) instead of the command.
         if isinstance(item[3], list):
             result["command"] = space.join(item[3])
+            result["client_address"] = item[4]
+            result["client_name"] = item[5]
         else:
             result["complexity"] = item[3]
             result["command"] = space.join(item[4])
+            result["client_address"] = item[5]
+            result["client_name"] = item[6]
         return result
 
     return [parse_item(item) for item in response]
@@ -902,8 +906,12 @@ class Redis(AbstractRedis, RedisModuleCommands, CoreCommands, SentinelCommands):
         arguments always win.
 
         """
+        single_connection_client = kwargs.pop("single_connection_client", False)
         connection_pool = ConnectionPool.from_url(url, **kwargs)
-        return cls(connection_pool=connection_pool)
+        return cls(
+            connection_pool=connection_pool,
+            single_connection_client=single_connection_client,
+        )
 
     def __init__(
         self,
@@ -1529,7 +1537,7 @@ class PubSub:
                     return None
             else:
                 conn.connect()
-            return conn.read_response()
+            return conn.read_response(disconnect_on_error=False)
 
         response = self._execute(conn, try_read)
 
diff --git a/redis/commands/core.py b/redis/commands/core.py
index d67291b..1a4acb2 100644
--- a/redis/commands/core.py
+++ b/redis/commands/core.py
@@ -761,6 +761,17 @@ class ManagementCommands(CommandsProtocol):
         """
         return self.execute_command("CLIENT NO-EVICT", mode)
 
+    def client_no_touch(self, mode: str) -> Union[Awaitable[str], str]:
+        """
+        # The command controls whether commands sent by the client will alter
+        # the LRU/LFU of the keys they access.
+        # When turned on, the current client will not change LFU/LRU stats,
+        # unless it sends the TOUCH command.
+
+        For more information see https://redis.io/commands/client-no-touch
+        """
+        return self.execute_command("CLIENT NO-TOUCH", mode)
+
     def command(self, **kwargs):
         """
         Returns dict reply of details about all Redis commands.
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:

problem_statement: Redis.from_url() is a misnomer
**Version**: 4.5.2

**Platform**: Python 3.11.2

**Description**: Redis.from_url is a convenience function, but the kwargs are passed into the ConnectionPool object and not the Redis instance. This formerly wasn't a problem, but it appears as though auto_close_connection_pool now defaults to true, so you can't reliably use Redis.from_url() in a singleton. 

repo: redis/redis-py
test_cmd: pytest --no-header -rA --tb=line --color=no -p no:cacheprovider -W ignore::DeprecationWarning
test_patch: diff --git a/tests/test_asyncio/test_commands.py b/tests/test_asyncio/test_commands.py
index 409934c..955b9d4 100644
--- a/tests/test_asyncio/test_commands.py
+++ b/tests/test_asyncio/test_commands.py
@@ -1,9 +1,11 @@
 """
 Tests async overrides of commands from their mixins
 """
+import asyncio
 import binascii
 import datetime
 import re
+import sys
 from string import ascii_letters
 
 import pytest
@@ -18,6 +20,11 @@ from tests.conftest import (
     skip_unless_arch_bits,
 )
 
+if sys.version_info >= (3, 11, 3):
+    from asyncio import timeout as async_timeout
+else:
+    from async_timeout import timeout as async_timeout
+
 REDIS_6_VERSION = "5.9.0"
 
 
@@ -446,6 +453,14 @@ class TestRedisCommands:
         with pytest.raises(exceptions.RedisError):
             await r.client_pause(timeout="not an integer")
 
+    @skip_if_server_version_lt("7.2.0")
+    @pytest.mark.onlynoncluster
+    async def test_client_no_touch(self, r: redis.Redis):
+        assert await r.client_no_touch("ON") == b"OK"
+        assert await r.client_no_touch("OFF") == b"OK"
+        with pytest.raises(TypeError):
+            await r.client_no_touch()
+
     async def test_config_get(self, r: redis.Redis):
         data = await r.config_get()
         assert "maxmemory" in data
@@ -3008,6 +3023,37 @@ class TestRedisCommands:
         for x in await r.module_list():
             assert isinstance(x, dict)
 
+    @pytest.mark.onlynoncluster
+    async def test_interrupted_command(self, r: redis.Redis):
+        """
+        Regression test for issue #1128:  An Un-handled BaseException
+        will leave the socket with un-read response to a previous
+        command.
+        """
+        ready = asyncio.Event()
+
+        async def helper():
+            with pytest.raises(asyncio.CancelledError):
+                # blocking pop
+                ready.set()
+                await r.brpop(["nonexist"])
+            # If the following is not done, further Timout operations will fail,
+            # because the timeout won't catch its Cancelled Error if the task
+            # has a pending cancel.  Python documentation probably should reflect this.
+            if sys.version_info >= (3, 11):
+                asyncio.current_task().uncancel()
+            # if all is well, we can continue.  The following should not hang.
+            await r.set("status", "down")
+
+        task = asyncio.create_task(helper())
+        await ready.wait()
+        await asyncio.sleep(0.01)
+        # the task is now sleeping, lets send it an exception
+        task.cancel()
+        # If all is well, the task should finish right away, otherwise fail with Timeout
+        async with async_timeout(0.1):
+            await task
+
 
 @pytest.mark.onlynoncluster
 class TestBinarySave:
diff --git a/tests/test_asyncio/test_connection.py b/tests/test_asyncio/test_connection.py
index e2d77fc..158b854 100644
--- a/tests/test_asyncio/test_connection.py
+++ b/tests/test_asyncio/test_connection.py
@@ -184,7 +184,7 @@ async def test_connection_parse_response_resume(r: redis.Redis):
     conn._parser._stream = MockStream(message, interrupt_every=2)
     for i in range(100):
         try:
-            response = await conn.read_response()
+            response = await conn.read_response(disconnect_on_error=False)
             break
         except MockStream.TestError:
             pass
@@ -271,3 +271,9 @@ async def test_connection_disconect_race(parser_class):
 
     vals = await asyncio.gather(do_read(), do_close())
     assert vals == [b"Hello, World!", None]
+
+
+@pytest.mark.onlynoncluster
+def test_create_single_connection_client_from_url():
+    client = Redis.from_url("redis://localhost:6379/0?", single_connection_client=True)
+    assert client.single_connection_client is True
diff --git a/tests/test_asyncio/test_cwe_404.py b/tests/test_asyncio/test_cwe_404.py
index d3a0666..21f2ddd 100644
--- a/tests/test_asyncio/test_cwe_404.py
+++ b/tests/test_asyncio/test_cwe_404.py
@@ -128,7 +128,6 @@ async def test_standalone(delay, master_host):
                 assert await r.get("foo") == b"foo"
 
 
-@pytest.mark.xfail(reason="cancel does not cause disconnect")
 @pytest.mark.onlynoncluster
 @pytest.mark.parametrize("delay", argvalues=[0.05, 0.5, 1, 2])
 async def test_standalone_pipeline(delay, master_host):
diff --git a/tests/test_commands.py b/tests/test_commands.py
index 2b769be..c71e347 100644
--- a/tests/test_commands.py
+++ b/tests/test_commands.py
@@ -1,9 +1,12 @@
 import binascii
 import datetime
 import re
+import threading
 import time
+from asyncio import CancelledError
 from string import ascii_letters
 from unittest import mock
+from unittest.mock import patch
 
 import pytest
 
@@ -693,6 +696,14 @@ class TestRedisCommands:
         with pytest.raises(TypeError):
             r.client_no_evict()
 
+    @pytest.mark.onlynoncluster
+    @skip_if_server_version_lt("7.2.0")
+    def test_client_no_touch(self, r):
+        assert r.client_no_touch("ON") == b"OK"
+        assert r.client_no_touch("OFF") == b"OK"
+        with pytest.raises(TypeError):
+            r.client_no_touch()
+
     @pytest.mark.onlynoncluster
     @skip_if_server_version_lt("3.2.0")
     def test_client_reply(self, r, r_timeout):
@@ -861,6 +872,8 @@ class TestRedisCommands:
         # make sure other attributes are typed correctly
         assert isinstance(slowlog[0]["start_time"], int)
         assert isinstance(slowlog[0]["duration"], int)
+        assert isinstance(slowlog[0]["client_address"], bytes)
+        assert isinstance(slowlog[0]["client_name"], bytes)
 
         # Mock result if we didn't get slowlog complexity info.
         if "complexity" not in slowlog[0]:
@@ -4741,6 +4754,38 @@ class TestRedisCommands:
         res = r2.psync(r2.client_id(), 1)
         assert b"FULLRESYNC" in res
 
+    @pytest.mark.onlynoncluster
+    def test_interrupted_command(self, r: redis.Redis):
+        """
+        Regression test for issue #1128:  An Un-handled BaseException
+        will leave the socket with un-read response to a previous
+        command.
+        """
+
+        ok = False
+
+        def helper():
+            with pytest.raises(CancelledError):
+                # blocking pop
+                with patch.object(
+                    r.connection._parser, "read_response", side_effect=CancelledError
+                ):
+                    r.brpop(["nonexist"])
+            # if all is well, we can continue.
+            r.set("status", "down")  # should not hang
+            nonlocal ok
+            ok = True
+
+        thread = threading.Thread(target=helper)
+        thread.start()
+        thread.join(0.1)
+        try:
+            assert not thread.is_alive()
+            assert ok
+        finally:
+            # disconnect here so that fixture cleanup can proceed
+            r.connection.disconnect()
+
 
 @pytest.mark.onlynoncluster
 class TestBinarySave:
diff --git a/tests/test_connection.py b/tests/test_connection.py
index 25b4118..31268a9 100644
--- a/tests/test_connection.py
+++ b/tests/test_connection.py
@@ -160,7 +160,7 @@ def test_connection_parse_response_resume(r: redis.Redis, parser_class):
         conn._parser._sock = mock_socket
     for i in range(100):
         try:
-            response = conn.read_response()
+            response = conn.read_response(disconnect_on_error=False)
             break
         except MockSocket.TestError:
             pass
@@ -205,3 +205,11 @@ def test_pack_command(Class):
 
     actual = Class().pack_command(*cmd)[0]
     assert actual == expected, f"actual = {actual}, expected = {expected}"
+
+
+@pytest.mark.onlynoncluster
+def test_create_single_connection_client_from_url():
+    client = redis.Redis.from_url(
+        "redis://localhost:6379/0?", single_connection_client=True
+    )
+    assert client.connection is not None