Fix blocking commands timeout when blocking_timeout exceeds socket_timeout

redis/asyncio/client.py

@@ -874,12 +874,21 @@ 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
+        blocking_timeout = options.pop("_blocking_timeout", None)
+        # 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 = 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

@@ -1657,12 +1657,21 @@ 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
+        blocking_timeout = kwargs.pop("_blocking_timeout", None)
+        # 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 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,21 @@ 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
+        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(