Repository navigation
Do not deliver pub/sub push messages as replies outside an active subscription - #4701
Conversation
Test Results 206 files ±0 206 suites ±0 7m 32s ⏱️ -6s Results for commit 344f2e2. ± Comparison against base commit cb03a88. This pull request skips 237 tests.♻️ This comment has been updated with latest results. |
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: a77fbd2597
ℹ️ About Codex in GitHub
Your team has set up Codex to 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 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
atakavci
left a comment
atakavci
left a comment
There was a problem hiding this comment.
left a note around atomics, which i feel needs change. other than that, looks good to me.
ggivo
left a comment
ggivo
left a comment
There was a problem hiding this comment.
@igor.malinovskiy
LGTM and resolve the issue.
One caveat is that pub/sub messages will no longer be propagated by getUnflushedObject().
An alternative solution: if we could actually flip the logic and drop all pub/sub push messages by default in Connection, and propagate them only in getUnflushedObject() (or a new dedicated method used by JedisPubSub)
A RESP3 pub/sub push arriving while no pub/sub read loop is running -- e.g. a message the server delivers after the unsubscribe confirmation, as observed behind the Redis Enterprise DMC proxy -- was propagated by the default push consumer and returned by Protocol.process as the next command's reply, desyncing the stream and typically failing with ClassCastException: java.util.ArrayList cannot be cast to [B. Track an active-subscription flag on Connection, set around the subscribe/process loops of JedisPubSubBase and JedisShardedPubSubBase, and gate the default pub/sub push consumer on it. During a subscription behavior is unchanged; outside one, a pub/sub push is logged and left to the rest of the consumer chain (consumed by default), so the reply stream stays aligned. RESP2 is unaffected: its pub/sub frames arrive as regular multi-bulk replies and never reach the push consumer chain. Custom push consumers keep their propagate/drop contract and now also observe pub/sub pushes that the default consumer previously short-circuited. Fixes the flaky advancedPubsubWithClientCache failures against Redis Enterprise. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
TokenBasedAuthenticationIntegrationTests.testJedisPubSubReauth wires the pooled connection through AuthXManager.addConnection, where the test substitutes a Mockito spy -- a shallow field-copy of the connection. The copied pushConsumers chain still held a consumer bound to the ORIGINAL object's boolean field, while the pub/sub loop set the flag on the copy, so the SUBSCRIBE confirmation push was dropped and the subscription never registered. Keep the flag in an AtomicBoolean holder and let the consumer capture the holder instead of `this`; shallow copies share the holder, so flag updates made through any copy stay visible to the chain. Adds a mock test reproducing the spy wiring (verified failing before this change).
Review feedback: an AtomicBoolean in core Connection is not a fair trade-off when the only consumer needing it was a test that wraps the pooled connection in a Mockito spy. Revert the flag to a plain volatile boolean and fix testJedisPubSubReauth instead: replace spy(connection) -- which works on a shallow field-copy and forks the connection's state from its push consumer chain -- with a delegating mock (mock(Connection.class, delegatesTo(original))). The mock still records invocations for the AUTH-count verification but forwards every call to the single real Connection, so the subscription flag set by the pub/sub loop stays visible to the push consumer chain. Also drop the ConnectionMockTest case that encoded the shallow-copy support, since that pattern is no longer meant to be supported.
The flag is written and read only by the thread currently driving the connection: inside the subscription the pub/sub loop thread both sets the flag and reads the pushes, and across threads the pool hand-off (returnObject/borrowObject) establishes the necessary happens-before. This matches the existing convention for connection state (e.g. the plain `broken` field). Also add the pub/sub push behavior change to the 8.1.0 release notes.
8bb0e1f to
344f2e2
Compare
|
run integration tests |
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 344f2e2d18
ℹ️ About Codex in GitHub
Your team has set up Codex to 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 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| subscribe(channels); | ||
| process(); | ||
| } finally { | ||
| authenticator.client.setActiveSubscription(false); |
There was a problem hiding this comment.
On RESP3, when the final UNSUBSCRIBE/PUNSUBSCRIBE confirmation sets subscribedChannels to zero, its callback still runs inside process() while this flag remains true. If that callback issues a normal command through the same Jedis connection and a delayed pub/sub push arrives before the command reply, the gated consumer propagates the push and the command receives the wrong response—the corruption this change is intended to prevent. Clear the gate as soon as the confirmation count reaches zero, before invoking the final unsubscribe callback; the sharded loop has the same ordering.
Useful? React with 👍 / 👎.
Jedis RE Integration Tests (RE 8.0.16)
Github Test Reporter by CTRF 💚 |
Jedis RE Integration Tests (RE 100.0.20)
Github Test Reporter by CTRF 💚 |
Jedis RE Integration Tests (RE 8.0.22)
Github Test Reporter by CTRF 💚 |
Problem
A RESP3 pub/sub push arriving while no pub/sub read loop is running — e.g. a
messageframe the server delivers after the unsubscribe confirmation, as observed behind the Redis Enterprise DMC proxy — was propagated by the default push consumer, andProtocol.processreturned it as the next command's reply. The reply stream desyncs and the command typically fails with:This is the root cause of the flaky
advancedPubsubWithClientCachefailures in the Redis Enterprise nightly CI (any client that reuses a pooled connection after subscribe/unsubscribe is exposed).Fix
Connectiontracks a volatileactiveSubscriptionflag, set/cleared around the subscribe/process loops ofJedisPubSubBase(proceed,proceedWithPatterns) andJedisShardedPubSubBase(proceed) — cleared infinally, so pooled connections are always returned with the flag off.PushConsumerChainImpl.pubSubConsumer(BooleanSupplier)and only propagates pub/sub-typed pushes while the flag is set. Outside a subscription, the push is debug-logged and left to the rest of the consumer chain (consumed by default), soProtocol.processcontinues to the real reply.Behavior notes
getUnflushedObject().PUBSUB_CONSUMERconstant is unchanged.Testing
ConnectionMockTestcases deterministically reproduce the bug withTcpMockServer: amessagepush injected before aPINGcorrupts the reply without the fix (verified failing) and is dropped with it; a second test verifies propagation still works while the flag is set.PushConsumerChainImplTestcases cover the gated consumer for all pub/sub push types and verify chain fall-through semantics when inactive.🤖 Generated with Claude Code
Note
Medium Risk
Changes core RESP3 read/push handling on every connection; low risk for normal pub/sub API usage but can affect hand-rolled subscribe +
getUnflushedObject()code paths.Overview
Fixes RESP3 reply-stream corruption when a pub/sub push (e.g. a
messageafterUNSUBSCRIBE, as seen behind Redis Enterprise) arrives on a pooled connection without an active subscribe loop. Those frames were propagated by the default push consumer and returned as the next command reply, often causingClassCastException.Connectionnow tracksactiveSubscription, set infinallyaroundJedisPubSubBase/JedisShardedPubSubBaseproceedloops. The default consumer ispubSubConsumer(this::isActiveSubscription)instead of always-onPUBSUB_CONSUMER: pub/sub pushes propagate only during a real read loop; otherwise they are debug-logged and not surfaced as command replies. Later **PushConsumer**s can still see those pushes when the default consumer does not propagate.PUBSUB_CONSUMERremains unchanged for unconditional propagation.Release notes document the behavior change. Tests add mock-server regression coverage and gated-consumer unit tests; auth integration tests use a delegating mock so connection state is not forked by
spy().Reviewed by Cursor Bugbot for commit 344f2e2. Bugbot is set up for automated code reviews on this repo. Configure here.