From 6d2ef91e912bf7cb42257af80bd1862b6e6d1a6f Mon Sep 17 00:00:00 2001
From: lllakshit <llakshitmathur239@gmail.com>
Date: Tue, 1 Sep 2026 01:24:21 +0530
Subject: [PATCH 1/4] fix(async): avoid asyncio.timeout on PubSub poll path
 (#3748)

Mirror the sync PubSub read path: check can_read(timeout) before read_response so non-blocking get_message() polling does not schedule asyncio.timeout(), which can corrupt the event loop callback heap under tight PubSub loops.
---
 redis/asyncio/client.py           | 12 ++++++++-
 redis/asyncio/connection.py       | 41 ++++++++++++++++++++++---------
 tests/test_asyncio/test_pubsub.py | 22 +++++++++++++++++
 3 files changed, 63 insertions(+), 12 deletions(-)

diff --git a/redis/asyncio/client.py b/redis/asyncio/client.py
index 5110a85fd9..575bfae69f 100644
--- a/redis/asyncio/client.py
+++ b/redis/asyncio/client.py
@@ -1445,7 +1445,17 @@ async def parse_response(self, block: bool = True, timeout: float = 0):
         # removed. That swap is a breaking change to the
         # Connection.read_response signature so it must wait for a
         # major release.
-        read_timeout = math.inf if block else timeout
+        #
+        # Non-blocking reads mirror the sync PubSub path: wait for
+        # readability (or buffered data) via can_read() and then read
+        # without scheduling asyncio.timeout(), which can corrupt the
+        # event loop callback heap under tight PubSub polling (#3748).
+        if not block:
+            if not await conn.can_read(timeout=timeout):
+                return None
+            read_timeout = math.inf
+        else:
+            read_timeout = math.inf
         response = await self._execute(
             conn,
             conn.read_response,
diff --git a/redis/asyncio/connection.py b/redis/asyncio/connection.py
index 61fbc20106..7328289085 100644
--- a/redis/asyncio/connection.py
+++ b/redis/asyncio/connection.py
@@ -1234,21 +1234,33 @@ async def send_command(self, *args: Any, **kwargs: Any) -> None:
     @deprecated_function(
         version="8.0.0", reason="Use can_read() instead", name="can_read_destructive"
     )
-    async def can_read_destructive(self) -> bool:
+    async def can_read_destructive(self, timeout: float = 0) -> bool:
         """Check the socket to see if there's data loaded in the buffer."""
-        try:
-            return await self._parser.can_read()
-        except OSError as e:
-            await self.disconnect(nowait=True)
-            host_error = self._host_error()
-            raise ConnectionError(f"Error while reading from {host_error}: {e.args}")
+        return await self.can_read(timeout=timeout)
 
-    async def can_read(self) -> bool:
-        """Check the socket to see if there's data loaded in the buffer."""
+    async def can_read(self, timeout: float = 0) -> bool:
+        """Check the socket to see if there's data loaded in the buffer.
+
+        When ``timeout`` is ``0``, return immediately if no data is buffered.
+        When ``timeout`` is positive, wait up to that many seconds for data
+        to become readable without entering ``asyncio.timeout(0)``, which can
+        corrupt the event loop's callback heap under concurrent PubSub polling
+        (#3748).
+        """
         # TODO: Rename this API; it detects pending data or dirty/closed
         # connection state, not only whether application data can be read.
         try:
-            return await self._parser.can_read()
+            if await self._parser.can_read():
+                return True
+            if timeout == 0:
+                return False
+            stream = self._parser._stream
+            if stream is None:
+                return False
+            await asyncio.wait_for(stream._wait_for_data("read"), timeout=timeout)
+            return True
+        except asyncio.TimeoutError:
+            return False
         except OSError as e:
             await self.disconnect(nowait=True)
             host_error = self._host_error()
@@ -1293,7 +1305,14 @@ async def read_response(
             read_timeout = timeout if timeout is not None else self.socket_timeout
         host_error = self._host_error()
         try:
-            if read_timeout is not None:
+            if read_timeout == 0:
+                if not await self.can_read(timeout=0):
+                    return None
+                response = await self._read_response_from_parser(
+                    disable_decoding=disable_decoding,
+                    push_request=push_request,
+                )
+            elif read_timeout is not None:
                 timeout_context = async_timeout(read_timeout)
                 if timeout is None:
                     async with timeout_context as active_timeout:
diff --git a/tests/test_asyncio/test_pubsub.py b/tests/test_asyncio/test_pubsub.py
index 7ba75830d9..463fbb57a8 100644
--- a/tests/test_asyncio/test_pubsub.py
+++ b/tests/test_asyncio/test_pubsub.py
@@ -1515,6 +1515,28 @@ async def test_get_message_timeout_zero_returns_immediately(self, r):
         assert elapsed < 0.1
         await p.aclose()
 
+    @pytest.mark.asyncio
+    async def test_pubsub_polling_avoids_async_timeout_scheduler(self, r, mocker):
+        """
+        Regression for #3748: tight PubSub polling must not schedule
+        asyncio.timeout(), which can corrupt the event loop callback heap.
+        """
+        from redis.asyncio import connection as async_connection
+
+        mock_timeout = mocker.patch.object(async_connection, "async_timeout")
+
+        p = r.pubsub()
+        await p.subscribe("foo")
+        msg = await wait_for_message(p, timeout=1.0)
+        assert msg is not None
+
+        for _ in range(100):
+            await p.get_message(timeout=0)
+            await p.get_message(timeout=0.01)
+
+        mock_timeout.assert_not_called()
+        await p.aclose()
+
     @pytest.mark.asyncio
     async def test_get_message_timeout_none_blocks(self, r):
         """

From 491224227a07ad8326518556903301c771568979 Mon Sep 17 00:00:00 2001
From: lllakshit <llakshitmathur239@gmail.com>
Date: Tue, 1 Sep 2026 13:23:09 +0530
Subject: [PATCH 2/4] fix(async): address PubSub poll review feedback (#3748)

Move can_read inside the PubSub retry path, preserve finite read timeouts after readiness checks, and tolerate legacy no-arg can_read overrides on custom connection classes. Tighten the regression test to assert asyncio.timeout(0) is never scheduled during timeout=0 polling.
---
 redis/asyncio/client.py           | 41 +++++++++++++++++--------------
 tests/test_asyncio/test_pubsub.py | 25 ++++++++++++-------
 2 files changed, 39 insertions(+), 27 deletions(-)

diff --git a/redis/asyncio/client.py b/redis/asyncio/client.py
index 575bfae69f..67a6158c21 100644
--- a/redis/asyncio/client.py
+++ b/redis/asyncio/client.py
@@ -1138,6 +1138,15 @@ async def listen(self) -> AsyncIterator[MonitorCommandInfo]:
             yield await self.next_command()
 
 
+async def _await_connection_can_read(conn: "Connection", timeout: float = 0) -> bool:
+    """Call ``can_read`` on a connection, tolerating legacy no-arg overrides."""
+    can_read = conn.can_read
+    try:
+        return await can_read(timeout=timeout)
+    except TypeError:
+        return await can_read()
+
+
 class PubSub:
     """
     PubSub provides publish, subscribe and listen support to Redis channels.
@@ -1445,24 +1454,20 @@ async def parse_response(self, block: bool = True, timeout: float = 0):
         # removed. That swap is a breaking change to the
         # Connection.read_response signature so it must wait for a
         # major release.
-        #
-        # Non-blocking reads mirror the sync PubSub path: wait for
-        # readability (or buffered data) via can_read() and then read
-        # without scheduling asyncio.timeout(), which can corrupt the
-        # event loop callback heap under tight PubSub polling (#3748).
-        if not block:
-            if not await conn.can_read(timeout=timeout):
-                return None
-            read_timeout = math.inf
-        else:
-            read_timeout = math.inf
-        response = await self._execute(
-            conn,
-            conn.read_response,
-            timeout=read_timeout,
-            disconnect_on_error=False,
-            push_request=True,
-        )
+        async def try_read():
+            if not block:
+                if not await _await_connection_can_read(conn, timeout):
+                    return None
+                read_timeout = timeout
+            else:
+                read_timeout = math.inf
+            return await conn.read_response(
+                timeout=read_timeout,
+                disconnect_on_error=False,
+                push_request=True,
+            )
+
+        response = await self._execute(conn, try_read)
 
         if conn.health_check_interval and response in self.health_check_response:
             # ignore the health check message as user might not expect it
diff --git a/tests/test_asyncio/test_pubsub.py b/tests/test_asyncio/test_pubsub.py
index 463fbb57a8..3ec028345b 100644
--- a/tests/test_asyncio/test_pubsub.py
+++ b/tests/test_asyncio/test_pubsub.py
@@ -1516,25 +1516,32 @@ async def test_get_message_timeout_zero_returns_immediately(self, r):
         await p.aclose()
 
     @pytest.mark.asyncio
-    async def test_pubsub_polling_avoids_async_timeout_scheduler(self, r, mocker):
+    async def test_pubsub_timeout_zero_avoids_async_timeout_zero(self, r):
         """
-        Regression for #3748: tight PubSub polling must not schedule
-        asyncio.timeout(), which can corrupt the event loop callback heap.
+        Regression for #3748: timeout=0 polling must not schedule
+        asyncio.timeout(0), which can corrupt the event loop callback heap.
         """
         from redis.asyncio import connection as async_connection
 
-        mock_timeout = mocker.patch.object(async_connection, "async_timeout")
-
         p = r.pubsub()
         await p.subscribe("foo")
         msg = await wait_for_message(p, timeout=1.0)
         assert msg is not None
 
-        for _ in range(100):
-            await p.get_message(timeout=0)
-            await p.get_message(timeout=0.01)
+        async_timeout_calls = []
+        real_async_timeout = async_connection.async_timeout
+
+        def tracking_async_timeout(timeout):
+            async_timeout_calls.append(timeout)
+            return real_async_timeout(timeout)
+
+        with patch.object(
+            async_connection, "async_timeout", side_effect=tracking_async_timeout
+        ):
+            for _ in range(50):
+                await p.get_message(timeout=0)
 
-        mock_timeout.assert_not_called()
+        assert 0 not in async_timeout_calls
         await p.aclose()
 
     @pytest.mark.asyncio

From ed058f6be2d54fceaa3c6a4470aecdfdb5e04147 Mon Sep 17 00:00:00 2001
From: lllakshit <llakshitmathur239@gmail.com>
Date: Tue, 1 Sep 2026 17:04:12 +0530
Subject: [PATCH 3/4] fix(async): address zero-timeout PubSub poll edge cases
 (#3748)

Yield to the event loop when polling with timeout=0 and no buffered data,
and parse timeout=0 reads from buffered bytes only so partial RESP frames
cannot block indefinitely.
---
 redis/_parsers/base.py                | 12 ++++-
 redis/_parsers/hiredis.py             | 12 ++++-
 redis/_parsers/resp2.py               | 30 +++++++++---
 redis/_parsers/resp3.py               | 67 +++++++++++++++++++++------
 redis/asyncio/client.py               |  4 ++
 redis/asyncio/connection.py           | 16 +++++--
 tests/test_asyncio/test_connection.py | 27 +++++++++++
 tests/test_asyncio/test_pubsub.py     | 23 +++++++++
 8 files changed, 162 insertions(+), 29 deletions(-)

diff --git a/redis/_parsers/base.py b/redis/_parsers/base.py
index f69db0ee09..1c9cb80696 100644
--- a/redis/_parsers/base.py
+++ b/redis/_parsers/base.py
@@ -72,6 +72,10 @@
 logger = logging.getLogger(__name__)
 
 
+class BufferedResponseIncomplete(Exception):
+    """Raised when a buffered-only read needs more socket data."""
+
+
 class BaseParser(ABC):
     EXCEPTION_CLASSES = {
         "ERR": {
@@ -562,7 +566,7 @@ async def can_read(self) -> bool:
         # parser and fail loudly if the private buffer API changes.
         return bool(self._stream._buffer)
 
-    async def _read(self, length: int) -> bytes:
+    async def _read(self, length: int, *, buffered_only: bool = False) -> bytes:
         """
         Read `length` bytes of data.  These are assumed to be followed
         by a '\r\n' terminator which is subsequently discarded.
@@ -572,6 +576,8 @@ async def _read(self, length: int) -> bytes:
         if len(self._buffer) >= end:
             result = self._buffer[self._pos : end - 2]
         else:
+            if buffered_only:
+                raise BufferedResponseIncomplete()
             tail = self._buffer[self._pos :]
             try:
                 data = await self._stream.readexactly(want - len(tail))
@@ -582,7 +588,7 @@ async def _read(self, length: int) -> bytes:
         self._pos += want
         return result
 
-    async def _readline(self) -> bytes:
+    async def _readline(self, *, buffered_only: bool = False) -> bytes:
         """
         read an unknown number of bytes up to the next '\r\n'
         line separator, which is discarded.
@@ -591,6 +597,8 @@ async def _readline(self) -> bytes:
         if found >= 0:
             result = self._buffer[self._pos : found]
         else:
+            if buffered_only:
+                raise BufferedResponseIncomplete()
             tail = self._buffer[self._pos :]
             data = await self._stream.readline()
             if not data.endswith(b"\r\n"):
diff --git a/redis/_parsers/hiredis.py b/redis/_parsers/hiredis.py
index 0c977b93c9..105bdc3465 100644
--- a/redis/_parsers/hiredis.py
+++ b/redis/_parsers/hiredis.py
@@ -348,7 +348,11 @@ async def read_from_socket(self):
         return True
 
     async def read_response(
-        self, disable_decoding: bool = False, push_request: bool = False
+        self,
+        disable_decoding: bool = False,
+        push_request: bool = False,
+        *,
+        buffered_only: bool = False,
     ) -> Union[EncodableT, List[EncodableT]]:
         # If `on_disconnect()` has been called, prohibit any more reads
         # even if they could happen because data might be present.
@@ -362,6 +366,8 @@ async def read_response(
             response = self._reader.gets()
 
         while response is NOT_ENOUGH_DATA:
+            if buffered_only:
+                return None
             await self.read_from_socket()
             if disable_decoding:
                 response = self._reader.gets(False)
@@ -379,7 +385,9 @@ async def read_response(
             response = await self.handle_push_response(response)
             if not push_request:
                 return await self.read_response(
-                    disable_decoding=disable_decoding, push_request=push_request
+                    disable_decoding=disable_decoding,
+                    push_request=push_request,
+                    buffered_only=buffered_only,
                 )
             else:
                 return response
diff --git a/redis/_parsers/resp2.py b/redis/_parsers/resp2.py
index 26701157f0..b0d9b563c4 100644
--- a/redis/_parsers/resp2.py
+++ b/redis/_parsers/resp2.py
@@ -3,7 +3,7 @@
 from ..exceptions import ConnectionError, InvalidResponse, ResponseError
 from ..typing import EncodableT
 from ..utils import SENTINEL
-from .base import _AsyncRESPBase, _RESPBase
+from .base import BufferedResponseIncomplete, _AsyncRESPBase, _RESPBase
 from .socket import SERVER_CLOSED_CONNECTION_ERROR
 
 
@@ -78,7 +78,9 @@ def _read_response(
 class _AsyncRESP2Parser(_AsyncRESPBase):
     """Async class for the RESP2 protocol"""
 
-    async def read_response(self, disable_decoding: bool = False):
+    async def read_response(
+        self, disable_decoding: bool = False, *, buffered_only: bool = False
+    ):
         if not self._connected:
             raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR)
         if self._chunks:
@@ -86,15 +88,25 @@ async def read_response(self, disable_decoding: bool = False):
             self._buffer += b"".join(self._chunks)
             self._chunks.clear()
         self._pos = 0
-        response = await self._read_response(disable_decoding=disable_decoding)
+        try:
+            response = await self._read_response(
+                disable_decoding=disable_decoding, buffered_only=buffered_only
+            )
+        except BufferedResponseIncomplete:
+            if buffered_only:
+                return None
+            raise
         # Successfully parsing a response allows us to clear our parsing buffer
         self._clear()
         return response
 
     async def _read_response(
-        self, disable_decoding: bool = False
+        self,
+        disable_decoding: bool = False,
+        *,
+        buffered_only: bool = False,
     ) -> Union[EncodableT, ResponseError, None]:
-        raw = await self._readline()
+        raw = await self._readline(buffered_only=buffered_only)
         response: Any
         byte, response = raw[:1], raw[1:]
 
@@ -122,13 +134,17 @@ async def _read_response(
         elif byte == b"$" and response == b"-1":
             return None
         elif byte == b"$":
-            response = await self._read(int(response))
+            response = await self._read(int(response), buffered_only=buffered_only)
         # multi-bulk response
         elif byte == b"*" and response == b"-1":
             return None
         elif byte == b"*":
             response = [
-                (await self._read_response(disable_decoding))
+                (
+                    await self._read_response(
+                        disable_decoding, buffered_only=buffered_only
+                    )
+                )
                 for _ in range(int(response))  # noqa
             ]
         else:
diff --git a/redis/_parsers/resp3.py b/redis/_parsers/resp3.py
index c0429d4b33..1b1321d3cc 100644
--- a/redis/_parsers/resp3.py
+++ b/redis/_parsers/resp3.py
@@ -6,6 +6,7 @@
 from ..utils import SENTINEL
 from .base import (
     AsyncPushNotificationsParser,
+    BufferedResponseIncomplete,
     PushNotificationsParser,
     _AsyncRESPBase,
     _RESPBase,
@@ -175,7 +176,11 @@ async def handle_pubsub_push_response(self, response):
         return response
 
     async def read_response(
-        self, disable_decoding: bool = False, push_request: bool = False
+        self,
+        disable_decoding: bool = False,
+        push_request: bool = False,
+        *,
+        buffered_only: bool = False,
     ):
         if not self._connected:
             raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR)
@@ -184,19 +189,30 @@ async def read_response(
             self._buffer += b"".join(self._chunks)
             self._chunks.clear()
         self._pos = 0
-        response = await self._read_response(
-            disable_decoding=disable_decoding, push_request=push_request
-        )
+        try:
+            response = await self._read_response(
+                disable_decoding=disable_decoding,
+                push_request=push_request,
+                buffered_only=buffered_only,
+            )
+        except BufferedResponseIncomplete:
+            if buffered_only:
+                return None
+            raise
         # Successfully parsing a response allows us to clear our parsing buffer
         self._clear()
         return response
 
     async def _read_response(
-        self, disable_decoding: bool = False, push_request: bool = False
+        self,
+        disable_decoding: bool = False,
+        push_request: bool = False,
+        *,
+        buffered_only: bool = False,
     ) -> Union[EncodableT, ResponseError, None]:
         if not self._stream or not self.encoder:
             raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR)
-        raw = await self._readline()
+        raw = await self._readline(buffered_only=buffered_only)
         response: Any
         byte, response = raw[:1], raw[1:]
 
@@ -206,7 +222,7 @@ async def _read_response(
         # server returned an error
         if byte in (b"-", b"!"):
             if byte == b"!":
-                response = await self._read(int(response))
+                response = await self._read(int(response), buffered_only=buffered_only)
             response = response.decode("utf-8", errors="replace")
             error = self.parse_error(response)
             # if the error is a ConnectionError, raise immediately so the user
@@ -236,14 +252,21 @@ async def _read_response(
             return response == b"t"
         # bulk response
         elif byte == b"$":
-            response = await self._read(int(response))
+            response = await self._read(int(response), buffered_only=buffered_only)
         # verbatim string response
         elif byte == b"=":
-            response = (await self._read(int(response)))[4:]
+            response = (await self._read(int(response), buffered_only=buffered_only))[
+                4:
+            ]
         # array response
         elif byte == b"*":
             response = [
-                (await self._read_response(disable_decoding=disable_decoding))
+                (
+                    await self._read_response(
+                        disable_decoding=disable_decoding,
+                        buffered_only=buffered_only,
+                    )
+                )
                 for _ in range(int(response))
             ]
         # set response
@@ -251,7 +274,12 @@ async def _read_response(
             # redis can return unhashable types (like dict) in a set,
             # so we always convert to a list, to have predictable return types
             response = [
-                (await self._read_response(disable_decoding=disable_decoding))
+                (
+                    await self._read_response(
+                        disable_decoding=disable_decoding,
+                        buffered_only=buffered_only,
+                    )
+                )
                 for _ in range(int(response))
             ]
         # map response
@@ -261,9 +289,14 @@ async def _read_response(
             # became defined to be left-right in version 3.8
             resp_dict = {}
             for _ in range(int(response)):
-                key = await self._read_response(disable_decoding=disable_decoding)
+                key = await self._read_response(
+                    disable_decoding=disable_decoding,
+                    buffered_only=buffered_only,
+                )
                 resp_dict[key] = await self._read_response(
-                    disable_decoding=disable_decoding, push_request=push_request
+                    disable_decoding=disable_decoding,
+                    push_request=push_request,
+                    buffered_only=buffered_only,
                 )
             response = resp_dict
         # push response
@@ -271,7 +304,9 @@ async def _read_response(
             response = [
                 (
                     await self._read_response(
-                        disable_decoding=disable_decoding, push_request=push_request
+                        disable_decoding=disable_decoding,
+                        push_request=push_request,
+                        buffered_only=buffered_only,
                     )
                 )
                 for _ in range(int(response))
@@ -279,7 +314,9 @@ async def _read_response(
             response = await self.handle_push_response(response)
             if not push_request:
                 return await self._read_response(
-                    disable_decoding=disable_decoding, push_request=push_request
+                    disable_decoding=disable_decoding,
+                    push_request=push_request,
+                    buffered_only=buffered_only,
                 )
             else:
                 return response
diff --git a/redis/asyncio/client.py b/redis/asyncio/client.py
index 67a6158c21..23a3c4f7a5 100644
--- a/redis/asyncio/client.py
+++ b/redis/asyncio/client.py
@@ -1457,6 +1457,10 @@ async def parse_response(self, block: bool = True, timeout: float = 0):
         async def try_read():
             if not block:
                 if not await _await_connection_can_read(conn, timeout):
+                    if timeout == 0:
+                        # Match PubSub.run(): yield so other tasks make progress
+                        # when polling with timeout=0 and no buffered data.
+                        await asyncio.sleep(0)
                     return None
                 read_timeout = timeout
             else:
diff --git a/redis/asyncio/connection.py b/redis/asyncio/connection.py
index 7328289085..eeae52e9da 100644
--- a/redis/asyncio/connection.py
+++ b/redis/asyncio/connection.py
@@ -1311,6 +1311,7 @@ async def read_response(
                 response = await self._read_response_from_parser(
                     disable_decoding=disable_decoding,
                     push_request=push_request,
+                    buffered_only=True,
                 )
             elif read_timeout is not None:
                 timeout_context = async_timeout(read_timeout)
@@ -1364,13 +1365,22 @@ async def read_response(
         return response
 
     async def _read_response_from_parser(
-        self, disable_decoding: bool = False, push_request: bool | None = False
+        self,
+        disable_decoding: bool = False,
+        push_request: bool | None = False,
+        *,
+        buffered_only: bool = False,
     ):
         if check_protocol_version(self.protocol, 3):
             return await self._parser.read_response(
-                disable_decoding=disable_decoding, push_request=push_request
+                disable_decoding=disable_decoding,
+                push_request=push_request,
+                buffered_only=buffered_only,
             )
-        return await self._parser.read_response(disable_decoding=disable_decoding)
+        return await self._parser.read_response(
+            disable_decoding=disable_decoding,
+            buffered_only=buffered_only,
+        )
 
     def pack_command(self, *args: EncodableT) -> List[bytes]:
         """Pack a series of arguments into the Redis protocol"""
diff --git a/tests/test_asyncio/test_connection.py b/tests/test_asyncio/test_connection.py
index 721600f145..aab3cbce3f 100644
--- a/tests/test_asyncio/test_connection.py
+++ b/tests/test_asyncio/test_connection.py
@@ -182,6 +182,16 @@ async def test_async_hiredis_can_read_leaves_decoding_to_read_response():
     assert await parser.read_response() == raw.decode()
 
 
+async def test_async_hiredis_read_response_buffered_only_returns_none_incomplete():
+    stream = DummyAsyncStream()
+    parser = make_async_hiredis_parser(
+        stream, response=NOT_ENOUGH_DATA, has_data=True
+    )
+
+    assert await parser.read_response(buffered_only=True) is None
+    assert stream.read_called is False
+
+
 @pytest.mark.parametrize("parser_class", [_AsyncRESP2Parser, _AsyncRESP3Parser])
 async def test_async_resp_can_read_detects_stream_buffer(parser_class):
     stream = DummyAsyncStream(buffer=b"+OK\r\n")
@@ -222,6 +232,23 @@ async def test_async_resp_can_read_prefers_buffered_data_over_eof(parser_class):
     assert stream.read_called is False
 
 
+@pytest.mark.parametrize("parser_class", [_AsyncRESP2Parser, _AsyncRESP3Parser])
+async def test_async_resp_read_response_buffered_only_returns_none_incomplete(
+    parser_class,
+):
+    stream = DummyAsyncStream()
+    parser = parser_class(socket_read_size=65536)
+    parser._connected = True
+    parser._stream = stream
+    parser._buffer = b"$10\r\n"
+    parser._pos = 0
+    parser.encoder = mock.Mock()
+    parser.encoder.decode.side_effect = lambda value: value
+
+    assert await parser.read_response(buffered_only=True) is None
+    assert stream.read_called is False
+
+
 @pytest.mark.parametrize("parser_class", [_AsyncRESP2Parser, _AsyncRESP3Parser])
 async def test_async_resp_read_response_raises_after_disconnect(parser_class):
     # A late reply read after disconnect could be assigned to the next
diff --git a/tests/test_asyncio/test_pubsub.py b/tests/test_asyncio/test_pubsub.py
index 3ec028345b..708406d6d0 100644
--- a/tests/test_asyncio/test_pubsub.py
+++ b/tests/test_asyncio/test_pubsub.py
@@ -1544,6 +1544,29 @@ def tracking_async_timeout(timeout):
         assert 0 not in async_timeout_calls
         await p.aclose()
 
+    @pytest.mark.asyncio
+    async def test_get_message_timeout_zero_yields_event_loop(self, r):
+        """
+        timeout=0 polling must yield to the event loop when no data is
+        buffered, matching PubSub.run() behavior.
+        """
+        p = r.pubsub()
+        await p.subscribe("foo")
+        msg = await wait_for_message(p, timeout=1.0)
+        assert msg is not None
+
+        other_ran = asyncio.Event()
+
+        async def other_task():
+            other_ran.set()
+
+        task = asyncio.create_task(other_task())
+        for _ in range(10):
+            await p.get_message(timeout=0)
+        await task
+        assert other_ran.is_set()
+        await p.aclose()
+
     @pytest.mark.asyncio
     async def test_get_message_timeout_none_blocks(self, r):
         """

From 29c6aeea05b33b1be46e3b0827d0e17a1cc623f9 Mon Sep 17 00:00:00 2001
From: lllakshit <llakshitmathur239@gmail.com>
Date: Tue, 1 Sep 2026 17:29:21 +0530
Subject: [PATCH 4/4] fix(async): recheck parser after can_read wait (#3748)

After waiting on the stream, re-run parser.can_read() so a peer close
that wakes _wait_for_data without payload is not reported as readable.
---
 redis/asyncio/connection.py           |  2 +-
 tests/test_asyncio/test_connection.py | 17 +++++++++++++++++
 2 files changed, 18 insertions(+), 1 deletion(-)

diff --git a/redis/asyncio/connection.py b/redis/asyncio/connection.py
index eeae52e9da..44f398e29a 100644
--- a/redis/asyncio/connection.py
+++ b/redis/asyncio/connection.py
@@ -1258,7 +1258,7 @@ async def can_read(self, timeout: float = 0) -> bool:
             if stream is None:
                 return False
             await asyncio.wait_for(stream._wait_for_data("read"), timeout=timeout)
-            return True
+            return await self._parser.can_read()
         except asyncio.TimeoutError:
             return False
         except OSError as e:
diff --git a/tests/test_asyncio/test_connection.py b/tests/test_asyncio/test_connection.py
index aab3cbce3f..b4678a7789 100644
--- a/tests/test_asyncio/test_connection.py
+++ b/tests/test_asyncio/test_connection.py
@@ -68,6 +68,9 @@ async def readline(self):
         self._buffer.clear()
         return data
 
+    async def _wait_for_data(self, _):
+        return None
+
 
 def make_async_hiredis_parser(
     stream, response=NOT_ENOUGH_DATA, decoded_response=None, has_data=False
@@ -249,6 +252,20 @@ async def test_async_resp_read_response_buffered_only_returns_none_incomplete(
     assert stream.read_called is False
 
 
+@pytest.mark.parametrize("parser_class", [_AsyncRESP2Parser, _AsyncRESP3Parser])
+async def test_async_connection_can_read_eof_after_wait_raises(parser_class):
+    stream = DummyAsyncStream(eof=True)
+    parser = parser_class(socket_read_size=65536)
+    parser._connected = True
+    parser._stream = stream
+    conn = Connection()
+    conn._parser = parser
+
+    with pytest.raises(ConnectionError):
+        await conn.can_read(timeout=0.1)
+    assert stream.read_called is False
+
+
 @pytest.mark.parametrize("parser_class", [_AsyncRESP2Parser, _AsyncRESP3Parser])
 async def test_async_resp_read_response_raises_after_disconnect(parser_class):
     # A late reply read after disconnect could be assigned to the next