fix: treat malformed numeric frames as connection-invalidating

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Qnxf2u2kSBxRM1AT7kD8BD
Signed-off-by: Kush Zingade <kush.zingade@gmail.com>

Author: UgaTheDev

redis/_parsers/base.py

@@ -55,7 +55,9 @@
 # - RecursionError: deeply nested aggregate replies exhaust the stack in the
 #   pure-Python parsers. #4144 turns this into InvalidResponse at a bounded
 #   depth; until then, catching it here is what stops the loop.
-UNRECOVERABLE_PARSE_ERRORS = (InvalidResponse, UnicodeDecodeError, RecursionError)
+# - ValueError: malformed numeric frames (e.g., `:abc\r\n`) where int(response)
+#   fails when parsing integer, bulk length, or aggregate length fields.
+UNRECOVERABLE_PARSE_ERRORS = (InvalidResponse, UnicodeDecodeError, RecursionError, ValueError)
 
 MODULE_LOAD_ERROR = "Error loading the extension. Please check the server logs."
 NO_SUCH_MODULE_ERROR = "Error unloading module: no such module with that name"

tests/test_asyncio/test_connection.py

@@ -1205,3 +1205,78 @@ async def test_binary_pubsub_payload_invalidates_connection(parser_class):
     )
 
     await conn.disconnect()
+
+
+@pytest.mark.parametrize(
+    "parser_class",
+    [_AsyncRESP2Parser],
+    ids=["AsyncRESP2Parser"],
+)
+class TestAsyncMalformedNumericFrameInvalidatesConnection:
+    """Async version: malformed numeric frames raise ValueError and must drop
+    the connection even with disconnect_on_error=False. The async parser
+    re-parses via self._pos = 0, so undecodable bytes stay queued on retry
+    unless the connection is invalidated.
+    """
+
+    @pytest.mark.asyncio
+    async def test_malformed_integer_frame_disconnects(self, parser_class):
+        """Malformed integer frame `:abc\r\n` raises ValueError."""
+        conn = Connection(protocol=2, parser_class=parser_class)
+        _attach_stream(conn, b":abc\r\n+SECOND\r\n")
+
+        with pytest.raises(ValueError):
+            await conn.read_response(disconnect_on_error=False, push_request=True)
+
+        assert conn.is_connected is False
+        await conn.disconnect()
+
+    @pytest.mark.asyncio
+    async def test_malformed_bulk_length_disconnects(self, parser_class):
+        """Malformed bulk string length `$xyz\r\n` raises ValueError."""
+        conn = Connection(protocol=2, parser_class=parser_class)
+        _attach_stream(conn, b"$xyz\r\n+SECOND\r\n")
+
+        with pytest.raises(ValueError):
+            await conn.read_response(disconnect_on_error=False, push_request=True)
+
+        assert conn.is_connected is False
+        await conn.disconnect()
+
+    @pytest.mark.asyncio
+    async def test_malformed_array_length_disconnects(self, parser_class):
+        """Malformed array length `*abc\r\n` raises ValueError."""
+        conn = Connection(protocol=2, parser_class=parser_class)
+        _attach_stream(conn, b"*abc\r\n+SECOND\r\n")
+
+        with pytest.raises(ValueError):
+            await conn.read_response(disconnect_on_error=False, push_request=True)
+
+        assert conn.is_connected is False
+        await conn.disconnect()
+
+    @pytest.mark.asyncio
+    async def test_next_read_is_clean_after_malformed_integer(self, parser_class):
+        """Reconnect after malformed integer frame serves new stream cleanly."""
+        conn = Connection(protocol=2, parser_class=parser_class)
+        _attach_stream(conn, b":abc\r\n")
+
+        with pytest.raises(ValueError):
+            await conn.read_response(disconnect_on_error=False, push_request=True)
+
+        # Reconnect with fresh stream
+        _attach_stream(conn, b"+RECOVERED\r\n")
+        result = await conn.read_response(disconnect_on_error=False, push_request=True)
+        assert result == b"RECOVERED"
+        await conn.disconnect()
+
+    @pytest.mark.asyncio
+    async def test_valid_integer_frame_leaves_connection_up(self, parser_class):
+        """Valid integer frames must not trigger the new predicate."""
+        conn = Connection(protocol=2, parser_class=parser_class)
+        _attach_stream(conn, b":42\r\n")
+
+        result = await conn.read_response(disconnect_on_error=False, push_request=True)
+        assert result == 42
+        assert conn.is_connected is True
+        await conn.disconnect()

tests/test_connection.py

@@ -2275,3 +2275,63 @@ def test_shallow_nesting_still_parses(self):
 
         assert conn.read_response(**self.PUBSUB_KWARGS) == [[[[]]]]
         assert conn.is_connected is True
+
+@pytest.mark.parametrize(
+    "parser_class",
+    [_RESP2Parser],
+    ids=["RESP2Parser"],
+)
+class TestMalformedNumericFrameInvalidatesConnection:
+    """Malformed numeric frames (non-numeric integers, bulk lengths, or array
+    lengths such as `:abc\r\n`, `$xyz\r\n`, `*abc\r\n`) raise ValueError when
+    int(response) is called. Before the fix, these stayed queued with
+    disconnect_on_error=False, causing infinite retries on the same frame.
+    See #4291.
+    """
+
+    PUBSUB_KWARGS = dict(disconnect_on_error=False, push_request=True)
+
+    def test_malformed_integer_frame_disconnects(self, parser_class):
+        """Malformed integer frame `:abc\r\n` raises ValueError."""
+        conn = _connection_with_stream(b":abc\r\n+SECOND\r\n", parser_class)
+
+        with pytest.raises(ValueError):
+            conn.read_response(**self.PUBSUB_KWARGS)
+
+        assert conn.is_connected is False
+
+    def test_malformed_bulk_length_disconnects(self, parser_class):
+        """Malformed bulk string length `$xyz\r\n` raises ValueError."""
+        conn = _connection_with_stream(b"$xyz\r\n+SECOND\r\n", parser_class)
+
+        with pytest.raises(ValueError):
+            conn.read_response(**self.PUBSUB_KWARGS)
+
+        assert conn.is_connected is False
+
+    def test_malformed_array_length_disconnects(self, parser_class):
+        """Malformed array length `*abc\r\n` raises ValueError."""
+        conn = _connection_with_stream(b"*abc\r\n+SECOND\r\n", parser_class)
+
+        with pytest.raises(ValueError):
+            conn.read_response(**self.PUBSUB_KWARGS)
+
+        assert conn.is_connected is False
+
+    def test_next_read_is_clean_after_malformed_integer(self, parser_class, monkeypatch):
+        """Reconnect after malformed integer frame serves new stream cleanly."""
+        conn = _connection_with_stream(b":abc\r\n+SECOND\r\n", parser_class)
+        _reconnect_with(conn, monkeypatch, b"+RECOVERED\r\n")
+
+        with pytest.raises(ValueError):
+            conn.read_response(**self.PUBSUB_KWARGS)
+
+        conn.connect()
+        assert conn.read_response(**self.PUBSUB_KWARGS) == b"RECOVERED"
+
+    def test_valid_integer_frame_leaves_connection_up(self, parser_class):
+        """Valid integer frames must not trigger the new predicate."""
+        conn = _connection_with_stream(b":42\r\n", parser_class)
+
+        assert conn.read_response(**self.PUBSUB_KWARGS) == 42
+        assert conn.is_connected is True