Skip to content

Keep a partly received pubsub message instead of reconnecting #4613

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
yamamoto7 wants to merge 2 commits into
base: master
Choose a base branch
from
+55 −3
Open

Keep a partly received pubsub message instead of reconnecting #4613

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
0917815
Keep a partly received pubsub message instead of reconnecting
yamamoto7 Nov 11, 2026
7b923e8
Leave a push handler's timeout to the retry layer and use a dedicated…
yamamoto7 Nov 11, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 13 additions & 3 deletions redis/client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -66,6 +66,7 @@
PubSubError,
RedisError,
ResponseError,
TimeoutError,
WatchError,
)
from redis.himport import HImportRegistry, parse_himport_set_args
Expand Down Expand Up @@ -1514,9 +1515,18 @@ def try_read():
# (and restore the original socket_timeout afterwards), so the
# configured socket_timeout does not abort the read.
read_timeout = None
return conn.read_response(
disconnect_on_error=False, push_request=True, timeout=read_timeout
)
try:
return conn.read_response(
disconnect_on_error=False, push_request=True, timeout=read_timeout
)
except TimeoutError:

@chatgpt-codex-connector chatgpt-codex-connector Bot Nov 11, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Propagate timeouts raised by RESP3 push handlers

When RESP3 is used and push_handler_func raises redis.exceptions.TimeoutError (for example, its own Redis I/O times out), Connection.read_response() routes that callback exception through this catch too, so nonblocking get_message() returns None rather than surfacing the handler failure. The Python RESP3 parser rewinds the frame on that exception, causing each poll to invoke the same failing handler again, while hiredis has already consumed the frame and silently drops the failure/message; the async PubSub path still propagates Redis TimeoutError. Narrow the suppression to a timeout produced by the bounded socket read.

AGENTS.md reference: AGENTS.md:L147-L148

Useful? React with 👍 / 👎.

@yamamoto7 yamamoto7 Nov 11, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good catch, thanks. A TimeoutError from a push_handler_func reaches try_read() the same way as a read timeout, so the two cannot be told apart there without relying on the implicit exception chain. With a push handler installed, the timeout now goes to the retry layer as on master; only reads without one return None. test_get_message_timeout_with_push_handler_reaches_retry_layer covers this.

One detail: on master, with the default retry, the handler failure is not surfaced either. The retry layer reconnects and the next read returns the resubscribe reply, so the failing message and the one after it are lost without an error. The regression was elsewhere: the pure-Python parser ran a handler that keeps failing again on every poll, so the subscriber spun without delivering anything. With this change the push handler case behaves as on master.

if block:
raise
# Only part of the message arrived within the timeout. The
# parser keeps what it has read, so the next call resumes the
# same message; letting the timeout reach the retry layer
# would reconnect and drop it.
return None

response = self._execute(conn, try_read)

Expand Down
22 changes: 22 additions & 0 deletions tests/test_pubsub.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2851,6 +2851,28 @@ def test_multiple_messages_with_timeout(self, r):
assert messages[1]["data"] == b"msg2"
assert messages[2]["data"] == b"msg3"

def test_get_message_resumes_a_partly_received_message(self, r):
"""
A message that is only partly readable when get_message() polls is
returned by a later call, not dropped by a reconnect.
"""
p = r.pubsub()
p.subscribe("foo")

@chatgpt-codex-connector chatgpt-codex-connector Bot Nov 11, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Use a test-specific pub/sub channel

In the cluster test profile, this unmarked test shares foo with many other cases. The existing wait_for_handled_message helper (lines 72-75) and adjacent test_timeout_with_pattern_subscribe explicitly account for an earlier PUBLISH foo arriving late over the cluster bus; if that happens after this subscription, can_read() observes the stale frame and the final b"hello" assertion fails even though resume behavior is correct. Use a channel unique to this test for both calls.

Useful? React with 👍 / 👎.

@yamamoto7 yamamoto7 Nov 11, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fixed: the test now uses its own channel, partly-received.

assert wait_for_message(p, timeout=1.0) is not None
r.publish("foo", "hello")
assert p.connection.can_read(timeout=1.0)

# What the parser raises when the rest of the message does not arrive
# within the read timeout; the unread data stays on the connection.
with (
mock.patch.object(p.connection, "read_response", side_effect=TimeoutError),
mock.patch.object(p, "_reconnect") as reconnect,
):
assert p.get_message(timeout=0) is None
reconnect.assert_not_called()

assert wait_for_message(p, timeout=1.0)["data"] == b"hello"

def test_timeout_with_pattern_subscribe(self, r):
"""
Test that timeout works correctly with pattern subscriptions.
Expand Down
Loading