AsyncIO Race Condition Fix (#2639)v4.3.64.3

6 files changed, 64 insertions, 8 deletions

diff --git a/.github/workflows/integration.yaml b/.github/workflows/integration.yaml
index 59f6c7d..87ebc27 100644
--- a/.github/workflows/integration.yaml
+++ b/.github/workflows/integration.yaml
@@ -32,10 +32,11 @@ jobs:
            invoke linters
 
    run-tests:
-     runs-on: ubuntu-latest
+     runs-on: ubuntu-20.04
      timeout-minutes: 30
      strategy:
        max-parallel: 15
+       fail-fast: false
        matrix:
          python-version: ['3.6', '3.7', '3.8', '3.9', '3.10', 'pypy-3.7']
          test-type: ['standalone', 'cluster']
@@ -79,8 +80,9 @@ jobs:
 
    install_package_from_commit:
     name: Install package from commit hash
-    runs-on: ubuntu-latest
+    runs-on: ubuntu-20.04
     strategy:
+      fail-fast: false
       matrix:
         python-version: ['3.6', '3.7', '3.8', '3.9', '3.10', 'pypy-3.7']
     steps:
diff --git a/redis/asyncio/client.py b/redis/asyncio/client.py
index 3d59016..5c0b546 100644
--- a/redis/asyncio/client.py
+++ b/redis/asyncio/client.py
@@ -1349,10 +1349,16 @@ class Pipeline(Redis):  # lgtm [py/init-calls-subclass]
         conn = cast(Connection, conn)
 
         try:
-            return await conn.retry.call_with_retry(
-                lambda: execute(conn, stack, raise_on_error),
-                lambda error: self._disconnect_raise_reset(conn, error),
+            return await asyncio.shield(
+                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()
 
diff --git a/redis/asyncio/cluster.py b/redis/asyncio/cluster.py
index 3fe3ebc..8dfb1cb 100644
--- a/redis/asyncio/cluster.py
+++ b/redis/asyncio/cluster.py
@@ -879,10 +879,18 @@ 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[0], **kwargs)
+            return await self.parse_response(connection, *args, **kwargs)
+        except asyncio.CancelledError:
+            # should not be possible
+            await connection.disconnect(nowait=True)
+            raise
         finally:
-            # Release connection
             self._free.append(connection)
 
     async def execute_pipeline(self, commands: List["PipelineCommand"]) -> bool:
diff --git a/setup.py b/setup.py
index 35c59f5..3c66aea 100644
--- a/setup.py
+++ b/setup.py
@@ -8,7 +8,7 @@ setup(
     long_description_content_type="text/markdown",
     keywords=["Redis", "key-value store", "database"],
     license="MIT",
-    version="4.3.5",
+    version="4.3.6",
     packages=find_packages(
         include=[
             "redis",
diff --git a/tests/test_asyncio/test_cluster.py b/tests/test_asyncio/test_cluster.py
index d6e01f7..2e44cdd 100644
--- a/tests/test_asyncio/test_cluster.py
+++ b/tests/test_asyncio/test_cluster.py
@@ -333,6 +333,23 @@ class TestRedisClusterObj:
                 called_count += 1
         assert called_count == 1
 
+    async def test_asynckills(self, r) -> None:
+
+        await r.set("foo", "foo")
+        await r.set("bar", "bar")
+
+        t = asyncio.create_task(r.get("foo"))
+        await asyncio.sleep(1)
+        t.cancel()
+        try:
+            await t
+        except asyncio.CancelledError:
+            pytest.fail("connection is left open with unread response")
+
+        assert await r.get("bar") == b"bar"
+        assert await r.ping()
+        assert await r.get("foo") == b"foo"
+
     async def test_execute_command_default_node(self, r: RedisCluster) -> None:
         """
         Test command execution without node flag is being executed on the
diff --git a/tests/test_asyncio/test_connection.py b/tests/test_asyncio/test_connection.py
index f6259ad..c414ee0 100644
--- a/tests/test_asyncio/test_connection.py
+++ b/tests/test_asyncio/test_connection.py
@@ -28,6 +28,29 @@ async def test_invalid_response(create_redis):
     assert str(cm.value) == f"Protocol Error: {raw!r}"
 
 
+@pytest.mark.onlynoncluster
+async def test_asynckills():
+    from redis.asyncio.client import Redis
+
+    for b in [True, False]:
+        r = Redis(single_connection_client=b)
+
+        await r.set("foo", "foo")
+        await r.set("bar", "bar")
+
+        t = asyncio.create_task(r.get("foo"))
+        await asyncio.sleep(1)
+        t.cancel()
+        try:
+            await t
+        except asyncio.CancelledError:
+            pytest.fail("connection left open with unread response")
+
+        assert await r.get("bar") == b"bar"
+        assert await r.ping()
+        assert await r.get("foo") == b"foo"
+
+
 @skip_if_server_version_lt("4.0.0")
 @pytest.mark.redismod
 @pytest.mark.onlynoncluster