Fix blocking commands timeout when blocking_timeout exceeds socket_timeout

redis/asyncio/client.py

@@ -874,12 +874,24 @@ async def parse_response(
         self, connection: Connection, command_name: Union[str, bytes], **options
     ):
         """Parses a response from the Redis server"""
+        # Extract blocking timeout for blocking commands (BLPOP, BRPOP, etc.)
+        # This ensures the socket timeout is long enough to wait for the
+        # blocking command's timeout
+        # Default None means "no blocking override" so read_response falls
+        # back to connection.socket_timeout. This differs from the sync client,
+        # where the default is SENTINEL and timeout=None means block forever.
+        blocking_timeout = options.pop("_blocking_timeout", None)
+        # Redis timeout=0 means block indefinitely on the server. Async
+        # Connection.read_response treats timeout=None as "use socket_timeout"
+        # and timeout=math.inf as "block with no timeout", so map 0 -> inf.
+        if blocking_timeout == 0:
+            blocking_timeout = math.inf
         try:
             if NEVER_DECODE in options:
                 response = await connection.read_response(disable_decoding=True)
                 options.pop(NEVER_DECODE)
             else:
-                response = await connection.read_response()
+                response = await connection.read_response(timeout=blocking_timeout)
         except ResponseError:
             if EMPTY_RESPONSE in options:
                 return options[EMPTY_RESPONSE]

redis/asyncio/cluster.py

@@ -1,6 +1,7 @@
 import asyncio
 import collections
 import logging
+import math
 import random
 import socket
 import threading
@@ -1657,12 +1658,24 @@ async def disconnect_free_connections(self) -> None:
     async def parse_response(
         self, connection: Connection, command: str, **kwargs: Any
     ) -> Any:
+        # Extract blocking timeout for blocking commands (BLPOP, BRPOP, etc.)
+        # This ensures the socket timeout is long enough to wait for the
+        # blocking command's timeout
+        # Default None means "no blocking override" so read_response falls
+        # back to connection.socket_timeout. This differs from the sync client,
+        # where the default is SENTINEL and timeout=None means block forever.
+        blocking_timeout = kwargs.pop("_blocking_timeout", None)
+        # Redis timeout=0 means block indefinitely on the server. Async
+        # Connection.read_response treats timeout=None as "use socket_timeout"
+        # and timeout=math.inf as "block with no timeout", so map 0 -> inf.
+        if blocking_timeout == 0:
+            blocking_timeout = math.inf
         try:
             if NEVER_DECODE in kwargs:
                 response = await connection.read_response(disable_decoding=True)
                 kwargs.pop(NEVER_DECODE)
             else:
-                response = await connection.read_response()
+                response = await connection.read_response(timeout=blocking_timeout)
         except ResponseError:
             if EMPTY_RESPONSE in kwargs:
                 return kwargs[EMPTY_RESPONSE]

redis/client.py

@@ -856,12 +856,24 @@ def failure_callback(error, failure_count):
 
     def parse_response(self, connection, command_name, **options):
         """Parses a response from the Redis server"""
+        # Extract blocking timeout for blocking commands (BLPOP, BRPOP, etc.)
+        # This ensures the socket timeout is long enough to wait for the
+        # blocking command's timeout
+        # Default SENTINEL means "no blocking override" so read_response uses
+        # the configured socket_timeout. Unlike async, timeout=None here means
+        # block forever (settimeout(None)), so Redis timeout=0 maps to None.
+        blocking_timeout = options.pop("_blocking_timeout", SENTINEL)
+        # When the Redis server is told to block indefinitely (timeout=0),
+        # the client must also wait indefinitely rather than issuing a
+        # zero-second socket read.
+        if blocking_timeout == 0:
+            blocking_timeout = None
         try:
             if NEVER_DECODE in options:
                 response = connection.read_response(disable_decoding=True)
                 options.pop(NEVER_DECODE)
             else:
-                response = connection.read_response()
+                response = connection.read_response(timeout=blocking_timeout)
         except ResponseError:
             if EMPTY_RESPONSE in options:
                 return options[EMPTY_RESPONSE]

redis/commands/core.py

@@ -3603,7 +3603,9 @@ def blmove(
         For more information, see https://redis.io/commands/blmove
         """
         params = [first_list, second_list, src, dest, timeout]
-        return self.execute_command("BLMOVE", *params)
+        return self.execute_command(
+            "BLMOVE", *params, _blocking_timeout=timeout
+        )
 
     @overload
     def mget(
@@ -4717,7 +4719,9 @@ def blpop(
             timeout = 0
         keys = list_or_args(keys, None)
         keys.append(timeout)
-        return self.execute_command("BLPOP", *keys)
+        return self.execute_command(
+            "BLPOP", *keys, _blocking_timeout=timeout
+        )
 
     @overload
     def brpop(
@@ -4748,7 +4752,9 @@ def brpop(
             timeout = 0
         keys = list_or_args(keys, None)
         keys.append(timeout)
-        return self.execute_command("BRPOP", *keys)
+        return self.execute_command(
+            "BRPOP", *keys, _blocking_timeout=timeout
+        )
 
     @overload
     def brpoplpush(
@@ -4775,7 +4781,9 @@ def brpoplpush(self, src: KeyT, dst: KeyT, timeout: Number | None = 0) -> (
         """
         if timeout is None:
             timeout = 0
-        return self.execute_command("BRPOPLPUSH", src, dst, timeout)
+        return self.execute_command(
+            "BRPOPLPUSH", src, dst, timeout, _blocking_timeout=timeout
+        )
 
     @overload
     def blmpop(
@@ -4816,7 +4824,9 @@ def blmpop(
         """
         cmd_args = [timeout, numkeys, *args, direction, "COUNT", count]
 
-        return self.execute_command("BLMPOP", *cmd_args)
+        return self.execute_command(
+            "BLMPOP", *cmd_args, _blocking_timeout=timeout
+        )
 
     @overload
     def lmpop(

tests/test_asyncio/test_blocking_timeout.py

@@ -0,0 +1,55 @@
+"""Unit tests for async blocking-command timeout mapping.
+
+These do not require a live Redis server.
+"""
+
+import math
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+import redis.asyncio as redis
+
+
+@pytest.mark.asyncio
+async def test_async_parse_response_maps_zero_blocking_timeout_to_math_inf():
+    """Redis timeout=0 must block indefinitely on async reads.
+
+    Async Connection.read_response treats:
+      - None as \"use socket_timeout\"
+      - math.inf as \"no timeout\"
+    so parse_response must map _blocking_timeout=0 to math.inf, not None.
+    """
+    conn = MagicMock()
+    conn.read_response = AsyncMock(return_value=None)
+
+    client = redis.Redis(connection_pool=MagicMock())
+    result = await client.parse_response(conn, "BLPOP", _blocking_timeout=0)
+
+    assert result is None
+    conn.read_response.assert_awaited_once_with(timeout=math.inf)
+
+
+@pytest.mark.asyncio
+async def test_async_parse_response_forwards_positive_blocking_timeout():
+    conn = MagicMock()
+    conn.read_response = AsyncMock(return_value=None)
+
+    client = redis.Redis(connection_pool=MagicMock())
+    await client.parse_response(conn, "BLPOP", _blocking_timeout=5)
+
+    conn.read_response.assert_awaited_once_with(timeout=5)
+
+
+@pytest.mark.asyncio
+async def test_async_parse_response_without_blocking_timeout_uses_default():
+    conn = MagicMock()
+    conn.read_response = AsyncMock(return_value=b"OK")
+
+    client = redis.Redis(connection_pool=MagicMock())
+    # Use a command without a response callback so the raw value is returned.
+    result = await client.parse_response(conn, "CUSTOMCMD")
+
+    assert result == b"OK"
+    # No override: default None -> read_response falls back to socket_timeout.
+    conn.read_response.assert_awaited_once_with(timeout=None)