diff --git a/redis/asyncio/client.py b/redis/asyncio/client.py index a0ef8172a3..6910080213 100644 --- a/redis/asyncio/client.py +++ b/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] diff --git a/redis/asyncio/cluster.py b/redis/asyncio/cluster.py index 7c86b85bff..ddabb7c21d 100644 --- a/redis/asyncio/cluster.py +++ b/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] diff --git a/redis/client.py b/redis/client.py index 9f1496b7e9..58773303bc 100755 --- a/redis/client.py +++ b/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] diff --git a/redis/commands/core.py b/redis/commands/core.py index 5967dd3f35..b76f6c3340 100644 --- a/redis/commands/core.py +++ b/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( diff --git a/tests/test_asyncio/test_blocking_timeout.py b/tests/test_asyncio/test_blocking_timeout.py new file mode 100644 index 0000000000..85fbaff68c --- /dev/null +++ b/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)