PubSub and maintenance-push reads loop forever on `InvalidResponse` because the offending reply is never consumed

Summary

Connection.read_response normally protects callers from a malformed reply by disconnecting: the except BaseException handler calls self.disconnect(), which drops the parser buffer so the poisoned stream is discarded and the pool reconnects on next use.

Three callers opt out of that with disconnect_on_error=False (line numbers against master @ afe6523e):

• PubSub.parse_response — redis/client.py:1420
• the pending-push drain inside Connection.send_command — redis/connection.py:1811
• _process_pending_invalidations — redis/connection.py:2070

On those paths a protocol-level parse failure is unrecoverable. The parsers rewind the buffer on any exception (pos = self._buffer.get_pos() … except BaseException: self._buffer.rewind(pos) in redis/_parsers/resp2.py:13-26, and the same in resp3.py), so the offending reply is not partially consumed — the read position returns to where it started and every byte of the reply stays queued. Combined with no disconnect, the next read re-parses the exact same bytes and raises the exact same error, indefinitely. A subscriber that hits this spins on the error with no way to make progress, and any reply queued behind the bad one is permanently unreachable.

The send_command drain site fails differently but from the same cause: its while ... can_read() loop only catches TimeoutError, so an InvalidResponse propagates out of send_command while the unread bytes keep can_read() true, leaving the connection in the same unusable state on a path where callers expect a send, not a parse error.

The rewind is the right behaviour for an in-band ResponseError, which legitimately belongs to a pipeline reply and may need to be re-read. It is the wrong behaviour for a framing violation, where the stream position is no longer trustworthy and preserving it only guarantees the next read fails identically.

Reproduction

No server needed — a fake socket serves two queued replies, the first of which fails to parse. Run from the redis-py checkout root:

import os
import socket
import sys

sys.path.insert(0, os.getcwd())  # run this from the redis-py checkout root

from redis.connection import Connection


class FakeSocket:
    """Serves a canned byte stream, then behaves like an open-but-idle socket."""

    def __init__(self, data):
        self.data, self.timeout = data, None

    def recv(self, n):
        if not self.data:
            raise socket.timeout("idle")
        chunk, self.data = self.data[:n], self.data[n:]
        return chunk

    def settimeout(self, t):
        self.timeout = t

    def gettimeout(self):
        return self.timeout

    def close(self):
        pass

    def shutdown(self, how):
        pass


# Reply 1 is deeply nested; reply 2 is what the next read should return.
DEPTH = 3000  # on master nothing caps depth, so this must exceed the recursion limit
STREAM = b"*1\r\n" * DEPTH + b"+leaf\r\n" + b"+SECOND\r\n"


def run(label, **kw):
    conn = Connection(protocol=2)
    conn._sock = FakeSocket(STREAM)
    conn._parser.on_connect(conn)
    print(label)
    for i in (1, 2):
        try:
            print(f"  read {i} -> {conn.read_response(**kw)!r}")
        except BaseException as e:
            print(f"  read {i} -> {type(e).__name__}: {str(e)[:50]}")
        buf = conn._parser._buffer
        print(
            f"     is_connected={conn.is_connected} "
            f"unread={'released' if buf is None else buf.unread_bytes()}"
        )


run("default path (disconnect_on_error=True)")
run("pubsub path (disconnect_on_error=False)", disconnect_on_error=False)

On master (afe6523e):

default path (disconnect_on_error=True)
  read 1 -> RecursionError: maximum recursion depth exceeded
     is_connected=False unread=released
  read 2 -> AttributeError: 'NoneType' object has no attribute 'readline'
     is_connected=False unread=released
pubsub path (disconnect_on_error=False)
  read 1 -> RecursionError: maximum recursion depth exceeded
     is_connected=True unread=12016
  read 2 -> RecursionError: maximum recursion depth exceeded
     is_connected=True unread=12016

The default path disconnects and releases the buffer, so it recovers. The pubsub path keeps the connection open with all 12016 bytes still queued and fails identically on the second read; +SECOND is never reached.

(The AttributeError on the default path's second read is an artifact of this harness calling read_response directly on an already-closed connection. Real callers go through the pool, which reconnects. It is not part of the issue.)

Relationship to #4144

#4144 adds a MAX_NESTING_DEPTH guard so deep nesting raises InvalidResponse instead of RecursionError. Running the same script against that branch (DEPTH = 150, since the guard trips at 101):

default path (disconnect_on_error=True)
  read 1 -> InvalidResponse: Response nesting depth exceeded 100
     is_connected=False unread=released
  read 2 -> AttributeError: 'NoneType' object has no attribute 'readline'
     is_connected=False unread=released
pubsub path (disconnect_on_error=False)
  read 1 -> InvalidResponse: Response nesting depth exceeded 100
     is_connected=True unread=616
  read 2 -> InvalidResponse: Response nesting depth exceeded 100
     is_connected=True unread=616

This is not caused by #4144. The loop is identical before and after; only the exception type changes. #4144 is adjacent context only: it makes the failure cleaner and deterministic — a defined protocol error at a defined depth rather than a stack overflow at an unpredictable one — but it does not make the pubsub path recover, because the cause is the missing invalidation, not the exception type. Nesting depth is just one way to reach a parse failure; any framing violation on these paths behaves the same way.

Suggested direction

Treat InvalidResponse as connection-invalidating even when disconnect_on_error=False. A framing violation means the parser can no longer locate reply boundaries in the stream, so the connection is not reusable regardless of what the caller asked for. disconnect_on_error=False exists so pubsub can survive in-band errors without dropping its subscriptions; it was not intended to keep a connection whose byte stream is no longer parseable.

That turns the failure from permanent into terminal-but-recoverable: one clean error, connection dropped, subscriber reconnects and resubscribes. Handling in the parser instead (consuming and discarding the bad reply rather than rewinding) is not viable — once framing is lost there is no reliable way to find where the next reply begins.

Happy to put up a PR if this direction sounds right.