fix: treat InvalidResponse as connection-invalidating even when disconnect_on_error=False

[UgaTheDev]

Fixes #4291

Problem

Connection.read_response protects callers from a malformed reply by disconnecting, which drops the parser buffer so the poisoned stream is discarded and the pool reconnects. Three callers opt out with disconnect_on_error=False:

• PubSub.parse_response (redis/client.py:1420)
• the pending-push drain in 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 (resp2.py:13-26, same in resp3.py), so the offending reply is not consumed — the read position returns to where it started and every byte stays queued. Combined with no disconnect, the next read re-parses the same bytes and raises the same error, indefinitely. A subscriber spins on the error with no way to make progress, and any reply queued behind the bad one is permanently unreachable.

Fix

Catch InvalidResponse ahead of the general except BaseException handler and disconnect regardless of disconnect_on_error, in both redis/connection.py and redis/asyncio/connection.py.

The reasoning is a semantic distinction between two error classes that happen to share a code path:

• An in-band ResponseError legitimately belongs to a reply. The stream stays framed, the error may need re-reading, and disconnect_on_error=False plus the rewind exist to serve exactly this — pubsub surviving an in-band error without dropping its subscriptions. Unchanged by this PR.
• An InvalidResponse is a framing violation: the parser can no longer locate reply boundaries, so the stream position is untrustworthy. Preserving it only guarantees the next read fails identically. Handling this in the parser (consuming and discarding the bad reply instead of rewinding) is not viable — once framing is lost there is no reliable way to find where the next reply begins.

InvalidResponse and ResponseError are siblings under RedisError with no subclasses, so the new handler cannot catch an in-band error.

The failure changes from permanent to terminal-but-recoverable: one clean error, connection dropped, subscriber reconnects and resubscribes via the on_connect callback.

Behavior change

InvalidResponse still propagates to the caller on all three paths — only the connection state changes. Per-path effect:

• PubSub: previously an infinite loop on the same error. Now the error surfaces once and the connection is dropped; the next get_message/listen reconnects (conn.connect() on the blocking path, conn.can_read() on the non-blocking one, both of which reconnect a closed connection) and resubscribes. InvalidResponse is not in the retry-supported errors, so it is not silently retried.
• The two while can_read() drain loops (send_command pending-push drain, _process_pending_invalidations): these previously left the connection open with unread bytes on a framing error. They now lose the connection. Both already let a non-TimeoutError propagate, so the loops exited via the exception either way; the difference is that the connection is no longer handed back in an unusable state. can_read() reconnects a closed connection, so neither loop can spin.

Callers that relied on keeping a connection alive across a framing violation will see it dropped — that is the intent, since such a connection could not produce a correct reply again.

Tests

tests/test_connection.py::TestInvalidResponseInvalidatesConnection — fake-socket repro from the issue, no server needed:

• framing error on the pubsub call signature (disconnect_on_error=False, push_request=True) disconnects and releases the buffer — fails on master
• pubsub path recovers on the next read after reconnecting instead of re-raising forever — fails on master
• in-band ResponseError on the same path still does not disconnect, and the next reply is readable — guards the rewind's purpose
• framing error still disconnects on the default path

tests/test_asyncio/test_connection.py::test_invalid_response_always_disconnects — parametrized over disconnect_on_error, asserts disconnect(nowait=True) in both cases; the False case fails on master.

Full runs of tests/test_connection.py, tests/test_asyncio/test_connection.py, tests/test_pubsub.py, tests/test_asyncio/test_pubsub.py: 477 passed, with the same 8 pre-existing cluster-dependent failures as on master (471 passed baseline; the delta is the 6 new tests).

Relationship to #4144

Independent. #4144 makes deep nesting raise InvalidResponse instead of RecursionError; the loop described here is identical before and after it, because the cause is the missing invalidation rather than the exception type. This PR does make #4144's error recoverable on the pubsub path.

---

Note

Medium Risk
Changes connection teardown on PubSub and CLIENT TRACKING cache paths; behavior is intentional but alters recovery when the stream is corrupted mid-reply.

Overview
Fixes #4291 by treating certain mid-reply parse failures as connection-invalidating even when callers pass disconnect_on_error=False (PubSub, push drains, CLIENT TRACKING invalidation drains).

Introduces UNRECOVERABLE_PARSE_ERRORS (InvalidResponse, UnicodeDecodeError, RecursionError) and handles it in sync/async Connection.read_response by always disconnecting and re-raising. Plain ValueError from push handlers is explicitly excluded.

RESP2/RESP3 parsers now use _parse_int / _parse_float so malformed numeric frames (:abc, $xyz, *abc) raise InvalidResponse instead of ValueError, aligning them with the same disconnect path.

CacheProxyConnection refactors invalidation draining into _drain_invalidations and flushes the local cache when an unrecoverable parse error occurs during drain (because the raw disconnect bypasses the proxy’s normal disconnect() flush).

Tests cover framing errors, binary pub/sub with decode_responses=True, deep nesting, malformed numerics, in-band ResponseError still rewinding, and push-handler ValueError not disconnecting.

^{Reviewed by Cursor Bugbot for commit 739fe97. Bugbot is set up for automated code reviews on this repo. Configure here.}

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: e155765f47

ℹ️ About Codex in GitHub

Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".

[UgaTheDev]

Both good catches — fixed in the follow-up commit. Details below, including one place where the suggested fix would have hidden something real.

1. _CannedSocket only implemented recv() (tests/test_connection.py)

Confirmed and reproduced: with hiredis installed, Connection(protocol=2) selects _HiredisParser, whose read_from_socket calls sock.recv_into(...), so all four new tests died with AttributeError: '_CannedSocket' object has no attribute 'recv_into' before ever reaching the framing error. The hiredis CI jobs would have gone red.

I went with implementing recv_into on the fake rather than pinning parser_class=_RESP2Parser, because pinning would have left the more interesting question unanswered — does the fix actually fire under hiredis? It does, and it is worth spelling out why, since the answer is not obvious:

hiredis.Reader raises hiredis.ProtocolError on a framing violation by default, and that is a bare Exception subclass, not a RedisError. If that were what reached Connection.read_response, it would fall through to the except BaseException arm, honour disconnect_on_error=False, and #4291 would persist under hiredis exactly as before. But _HiredisParser constructs its reader with protocolError=InvalidResponse (redis/_parsers/hiredis.py:139 for sync, :293 for async), so hiredis raises InvalidResponse too and the new except InvalidResponse arm covers both parser families. Pinning one parser would have left that untested, so the fake now serves whichever parser is installed.

Worth noting the hiredis case makes the disconnect more necessary, not less: once its reader hits a protocol error it stays poisoned, returning the same error on every subsequent gets(). Dropping the connection is what clears it, since on_disconnect() sets _reader = None and reconnect builds a fresh one.

I also dropped the assert conn._parser._buffer is None line. It reaches into a pure-Python parser internal that has no hiredis equivalent (_buffer there is a preallocated bytearray that legitimately survives disconnect), and what it was standing in for is already covered by is_connected is False plus the recovery test.

2. Cache flush bypassed on the invalidation-drain path (redis/connection.py)

Also real, and the reasoning holds up. CacheProxyConnection is delegation, not inheritance: it holds self._conn and forwards. The pending-invalidation drain calls read_response directly on that raw connection, so self.disconnect() inside Connection.read_response is the raw connection's disconnect. CacheProxyConnection.disconnect() — the one that does self._cache.flush() before delegating — never runs. The next connect() then negotiates a fresh CLIENT TRACKING session that the server holds no invalidation state for, while the local cache keeps serving entries recorded under the old session. Stale reads, silently.

Both drain sites had the same copy of that loop (_process_pending_invalidations, and the re-check drain inside send_command, which reads off entry.connection_ref and so can be a different pool connection's raw socket). They now share one _drain_invalidations(conn) helper that flushes before re-raising, which matches what disconnect() already does — a full flush, since any invalidation messages still queued on the dropped session are lost and every entry tracked there is suspect.

Test evidence

Two unit tests added in tests/test_cache.py::TestUnitCacheProxyConnectionInvalidations, using the existing MagicMock + DefaultCache unit style already in that file, so no live server is needed:

• test_framing_error_while_draining_flushes_cache — cache is empty after the drain raises. Verified this fails on the parent commit and passes with the fix, so it is a real regression test rather than one written to the implementation.
• test_idle_connection_leaves_cache_intact — the loop's normal TimeoutError exit must not flush.

Runs (Python 3.13, local Redis on 6379):

• All 4 sync tests from fix: treat InvalidResponse as connection-invalidating even when disconnect_on_error=False #4293 pass with hiredis installed and without it.
• tests/test_cache.py -k Unit: 61 passed.
• tests/test_connection.py tests/test_cache.py tests/test_pubsub.py tests/test_connection_pool.py: 405 passed. The remaining failures are all environmental — TestClusterCache, TestSentinelCache, TestSSLCache, the cluster pubsub tests, and the busy-loading/auth tests in test_connection_pool.py, none of which have endpoints in my local setup. Notably plain TestCache — the CSC integration tests that actually exercise the drain code I changed, test_cache_clears_on_disconnect included — passed in full against the live server.

The async side needs no equivalent change: there is no CacheProxyConnection under redis/asyncio/, so this drain path is sync-only.

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: 98f2e2cf86

ℹ️ About Codex in GitHub

Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".

[chatgpt-codex-connector]

Flush proxy caches on every new forced disconnect

When client-side caching is enabled and a pooled connection that previously populated an entry is later used by PubSub, PubSub reads through CacheProxyConnection.read_response with disconnect_on_error=False, but that method delegates directly to the raw _conn. This new call therefore disconnects only the raw connection and bypasses CacheProxyConnection.disconnect(), leaving entries from the old CLIENT TRACKING session in the shared cache; after reconnecting, a missed invalidation can make those entries be served stale. Although the drain-loop fix addresses the earlier review comment, fresh evidence remains in the direct delegation at CacheProxyConnection.read_response, which is outside _drain_invalidations; this forced-disconnect path must notify the proxy or otherwise flush its cache too.

Useful? React with 👍 / 👎.

[petyaslavova]

Hey @UgaTheDev, thank you for your contribution!

The defect is real and Connection.read_response is the right place to fix it - it also covers SentinelManagedConnection.read_response, which defaults to disconnect_on_error=False and is a fourth opt-out site you did not list. Before merging, though, the predicate needs to be widened, because InvalidResponse is not what the reachable cases raise:

• The reproduction in #4291 raises RecursionError on current master, so this patch does not fix it unless #4144 lands first. That case is real, not synthetic: EVAL "local t={} local c=t for i=1,3000 do local n={} c[1]=n c=n end return t" 0 makes Redis emit exactly your fake stream (12004 bytes), and through a sentinel-managed connection it is on the ordinary command path.
• The realistic pubsub trigger is Encoder.decode at the tail of _read_response, which runs after the cursor has passed the payload. A decode_responses=True subscriber receiving a binary PUBLISH payload raises UnicodeDecodeError and loops exactly as described, on both parser backends. Please add that as a regression test.

Keep the allowlist shape rather than inverting to "disconnect unless resumable" - test_connection_parse_response_resume drives resume with a bare BaseException under disconnect_on_error=False and would break.

One correction to the rationale, since it is now in the code comments: the rewind and disconnect_on_error come from #2510 ("Make PythonParser resumable") and #2695, so they exist to re-parse an interrupted read, not to survive in-band errors. An in-band ResponseError is returned as a value and purged, never rewound.

Also needed: an async test that exercises the pubsub read path and asserts the next read is clean (the async parser re-parses via self._pos = 0, not the sync rewind), and a rebase for the end-of-file conflicts in the two test files.

[UgaTheDev]

Thanks for the detailed review, and in particular for the SentinelManagedConnection.read_response catch. That is a fourth disconnect_on_error=False opt-out site I had missed, and since it sits on the ordinary command path it is the one that makes the nesting case realistic rather than synthetic.

You were also right about the rationale. I had written the comments as though the rewind and disconnect_on_error existed to survive in-band errors. They come from #2510 ("Make PythonParser resumable") and #2695, and exist to re-parse an interrupted read; an in-band ResponseError is returned as a value and purged, never rewound. Both comments now say that.

What the reachable cases actually raise

Before widening anything I measured the two cases you named against this branch, on both parser backends, through read_response(disconnect_on_error=False, push_request=True):

Case	_RESP2Parser	_HiredisParser
Binary PUBLISH payload, decode_responses=True	UnicodeDecodeError, still connected	UnicodeDecodeError, still connected
3000-deep nested reply (12004 bytes)	RecursionError, still connected	InvalidResponse("Max nesting depth exceeded"), disconnected

So your read is confirmed on both counts. InvalidResponse alone was the wrong predicate: the pubsub trigger was not covered on either backend, and the nesting case was covered only by accident, on hiredis, because hiredis reports its own depth limit as a protocol error.

The widened predicate

Kept as an allowlist, defined once in redis/_parsers/base.py and applied at both sites:

UNRECOVERABLE_PARSE_ERRORS = (InvalidResponse, UnicodeDecodeError, RecursionError)

The shared name is because the sync and async handlers have to agree, and the third consumer is the CacheProxyConnection invalidation drain, which reads straight off the raw connection and so has to flush the cache on the same set of errors.

I did not invert to "disconnect unless resumable", for the reason you gave: test_connection_parse_response_resume drives resume with a bare BaseException under disconnect_on_error=False, and inverting would break it. All five variants still pass, which is the check that the allowlist shape survived.

RecursionError and #4144: what this covers and what it does not

Being precise, because these are two different problems:

• The loop (this PR) is now closed on both backends. On hiredis the nesting case already disconnected via InvalidResponse; on the pure-Python parsers it raised RecursionError and looped, and catching that stops it.
• The stack exhaustion (fix: guard RESP parsers against RecursionError on deeply nested replies (#4116) #4144) is not fixed here and should not be. Catching RecursionError is recovery after the stack has already been blown, and the depth at which it fires is not a stable boundary: it moves with sys.setrecursionlimit, frame size, and interpreter build. fix: guard RESP parsers against RecursionError on deeply nested replies (#4116) #4144 bounds the recursion so it never happens, and turns the case into a fixed-depth InvalidResponse that the first entry in the tuple already covers. After fix: guard RESP parsers against RecursionError on deeply nested replies (#4116) #4144 lands the RecursionError entry becomes a backstop rather than the mechanism.

So the accurate statement about #4291 is: the reported symptom (a disconnect_on_error=False caller looping forever on the same bytes) is fixed by this PR on both backends. The underlying unbounded recursion still needs #4144.

One consequence worth flagging: a user-supplied push-notification callback that recurses deeply enough to raise RecursionError inside read_response will now also invalidate the connection. That seems right to me, since the parser cursor position is not knowable at that point, but it is a widening beyond framing errors and I would rather name it than have it found later.

Tests

New regression coverage:

• TestBinaryPubSubPayloadInvalidatesConnection in tests/test_connection.py, parametrized over _RESP2Parser and _HiredisParser: binary payload under decode_responses=True raises UnicodeDecodeError, the connection is invalidated, and after the reconnect the next read is clean. Plus a decodable payload that must not trip the new predicate.
• TestDeeplyNestedReplyInvalidatesConnection: the 3000-deep reply, asserting RecursionError on the pure-Python parser and InvalidResponse on hiredis, both disconnecting, plus a shallow reply that must still parse.
• test_binary_pubsub_payload_invalidates_connection in tests/test_asyncio/test_connection.py, over _AsyncRESP2Parser and _AsyncHiredisParser, exercising the pubsub read path and asserting the next read is clean. The async parser re-parses from self._pos = 0 rather than rewinding a socket buffer, so this needed its own test rather than mirroring the sync one.

Fail-before

With the tuple temporarily narrowed back to (InvalidResponse,):

$ pytest tests/test_connection.py -k "BinaryPubSubPayload or DeeplyNestedReply" -q
5 failed, 4 passed

$ pytest tests/test_asyncio/test_connection.py -k binary_pubsub -q
2 failed

The four that pass pre-widening are the negative controls and the hiredis nesting case, which as noted was already covered. Restoring the widened tuple:

$ pytest tests/test_connection.py -k "BinaryPubSubPayload or DeeplyNestedReply" -q
9 passed

$ pytest tests/test_asyncio/test_connection.py -k binary_pubsub -q
2 passed

Full runs

$ pytest tests/test_connection.py -q
122 passed, 1 skipped

$ pytest tests/test_asyncio/test_connection.py -q
89 passed, 1 skipped

$ pytest tests/test_connection.py tests/test_asyncio/test_connection.py \
    -k test_connection_parse_response_resume -q
5 passed

All five resume variants: RESP2Parser, RESP3Parser, HiredisParser sync, and single / pool async.

Two suites have pre-existing failures in this environment, so I diffed the failing test names against the same run on master at 7ff0481 rather than just comparing counts:

• tests/test_cache.py: 10 failed, 85 passed, 12 errors here versus 10 failed, 83 passed, 12 errors on master. Identical failure set (TLS fixtures this environment does not provide); the +2 passed are the added TestUnitCacheProxyConnectionInvalidations cases.
• tests/test_pubsub.py tests/test_sentinel.py plus both async counterparts: 18 failed, 356 passed on both branch and master, identical failure set (cluster and sentinel servers not running here).

So no suite regressed; nothing that passes on master fails here.

Run with hiredis 3.4.1 installed, so the hiredis-parametrized cases were genuinely exercised rather than skipped.

Rebase

Rebased onto master at 7ff0481. The two conflicts in tests/test_connection.py and tests/test_asyncio/test_connection.py were both end-of-file collisions between the parse_url retry tests added upstream and the tests added here, with no overlapping content, so both sides are kept. ruff check and ruff format --check are clean on every file this PR touches.

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: 227ba2d3e7

ℹ️ About Codex in GitHub

Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".

[cursor]

Cursor Bugbot has reviewed your changes using high effort and found 1 potential issue.

^{Reviewed by Cursor Bugbot for commit 227ba2d. Configure here.}

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: a44aad3cef

ℹ️ About Codex in GitHub

Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: a62dc413ab

ℹ️ About Codex in GitHub

Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".

[chatgpt-codex-connector]

Separate push-handler exceptions from parser failures

With RESP3 PubSub using hiredis, the public push_handler_func runs after hiredis has consumed the complete push frame, so if that callback raises UnicodeDecodeError, RecursionError, or InvalidResponse, this tuple now causes Connection.read_response(disconnect_on_error=False) to disconnect an otherwise healthy subscription instead of leaving the next queued frame readable. The fresh evidence after excluding ValueError is that these other callback-raisable types remain classified solely by exception class; parser-originated failures need to be distinguished from exceptions escaping a user handler.

Useful? React with 👍 / 👎.

[petyaslavova]

Thanks for the follow-ups. The widened predicate is what I was after, and the measured per-backend table settles the reachability question. The async mirror and the cache flush on the invalidation drain both look right to me.

Two things before merge: please rebase onto master, and get the integration matrix green - the head now routes every RESP numeric field through _parse_int, so I want a full protocol x legacy_responses run rather than just the bot check. Also, the malformed-numeric test classes parametrize over a single parser; please extend them to the hiredis parsers, which report these frames as InvalidResponse, so backend parity is covered.

On the two open bot threads: the cache-flush one on CacheProxyConnection.read_response is a real gap, but it predates this PR - a raw disconnect already skips the proxy flush on every disconnect_on_error=True error path - so we will track it separately rather than grow this PR. The push-handler one is also real for hiredis, where the frame is consumed before the handler runs, though on the pure-Python parser your change improves that case rather than regressing it. If you would like to close it properly, converting the parsers' own decode failures to InvalidResponse at the decode sites and dropping UnicodeDecodeError from the tuple would do it; I am fine merging without that.

The _parse_int/_parse_float conversion is beyond what I asked for, but I am happy to keep it: without it a desynced stream yielding a non-numeric length still loops, and plain ValueError cannot go in the tuple because a push handler may raise it.

[UgaTheDev]

Thanks @petyaslavova. All three items are in, head is now 739fe971.

Rebased onto master. The conflicts were in redis/connection.py and redis/asyncio/connection.py, where master had meanwhile added add_debug_log_for_connection_failure and an as e binding to the BaseException handler. I resolved by keeping master's version and wiring the new handler into the same logging, so the sync UNRECOVERABLE_PARSE_ERRORS clause now reads:

except UNRECOVERABLE_PARSE_ERRORS as e:
    # ... (unchanged rationale)
    add_debug_log_for_connection_failure(self, e, "reading response")
    self.disconnect()
    raise

That keeps the new path consistent with the neighbouring handlers rather than being the one that logs nothing. Branch is 0 commits behind master.

Hiredis parity on the malformed-numeric classes. You were right that they parametrized over a single parser. Both classes now cover all three backends, and since hiredis reports these frames as InvalidResponse (the same exception the tests already asserted) no test body changed, only the parametrize list:

• TestMalformedNumericFrameInvalidatesConnection: _RESP2Parser, _RESP3Parser, _HiredisParser
• TestAsyncMalformedNumericFrameInvalidatesConnection: _AsyncRESP2Parser, _AsyncRESP3Parser, _AsyncHiredisParser

The hiredis params carry the same skipif(not HIREDIS_AVAILABLE) guard TestDeeplyNestedReplyInvalidatesConnection already uses. 30 tests pass locally against hiredis 3.4.1.

Integration matrix. Over to CI for the full protocol x legacy_responses run, since I do not have the sentinel/cluster/SSL setup locally. For what it is worth, tests/test_connection.py and tests/test_asyncio/test_connection.py are 255 passed / 2 skipped / 0 failed here; the only local failures are test_cache.py cases that need live sentinel and cluster servers (MasterNotFoundError). Happy to chase anything the matrix turns up.

On the two bot threads, agreed on both counts, and I am content to leave the CacheProxyConnection flush and the hiredis push-handler decode conversion out of this PR as you suggested.