fix(cluster): prevent lock-inversion deadlock in MultiNodePipelineBase (#4557) - #4645
anshullakra007 wants to merge 8 commits into
Conversation
![cursor[bot]](https://avatars.githubusercontent.com/in/1210556?s=60&v=4){kind=small}
![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: c191b7e246
ℹ️ 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".
| } catch (JedisConnectionException jce) { | ||
| log.error("Error with connection to " + nodeKey, jce); |
There was a problem hiding this comment.
When a JedisConnectionException occurs, both queues are retained: a failure partway through sendCommand leaves the whole command queue to be replayed by the next sync() or close(), potentially duplicating already-executed writes, while a failure in getMany occurs after cmdQueue.clear() and leaves stale responses so the next sync waits for more replies than it sent. Remove both queues for this node after a failed batch rather than allowing an unsafe retry.
Useful? React with 👍 / 👎.
| for (Map.Entry<HostAndPort, Queue<Response<?>>> entry : pipelinedResponses.entrySet()) { | ||
| HostAndPort nodeKey = entry.getKey(); | ||
| Queue<Response<?>> queue = entry.getValue(); | ||
| Connection connection = connections.get(nodeKey); | ||
| Queue<CommandObject<?>> cmdQueue = pipelinedCommands.get(nodeKey); |
There was a problem hiding this comment.
After a successful sync, the response and command queues are empty but their map entries remain, so every later no-op sync()—including the implicit sync in close()—still creates an executor and borrows a connection from every node previously used. If another client has exhausted one of those pools, closing an already-synchronized pipeline can block or time out despite having no commands; remove drained entries or skip empty command queues.
Useful? React with 👍 / 👎.
| // Simple routing based on key | ||
| for (Object arg : args) { | ||
| if (arg instanceof byte[]) { | ||
| String s = new String((byte[])arg); |
There was a problem hiding this comment.
When the JVM default charset differs from Jedis' configured charset, new String(byte[]) may not decode the bytes produced by CommandArguments.key, causing both A and B to fall through to shardA. The regression test then stops exercising the intended two-shard inverse acquisition sequence; use SafeEncoder.encode((byte[]) arg) as required by the repository's encoding convention.
AGENTS.md reference: AGENTS.md:L78-L80
Useful? React with 👍 / 👎.
| executor.execute(() -> { | ||
| Connection connection = null; | ||
| try { | ||
| connection = getConnection(nodeKey); |
There was a problem hiding this comment.
When pool exhaustion or a connection-factory error makes getConnection fail, Pool.getResource() commonly throws a generic JedisException, which is not caught here. For a multi-node pipeline the executor silently captures that exception because its Future is discarded, so sync() returns normally with unset responses; for a single-node pipeline the inline executor propagates it before syncing is reset, causing every subsequent sync() to return without doing work. Capture worker failures, restore synchronization state in a finally block, and propagate the failure to the caller.
Useful? React with 👍 / 👎.
| assertDoesNotThrow(() -> { | ||
| f1.get(5, TimeUnit.SECONDS); | ||
| f2.get(5, TimeUnit.SECONDS); | ||
| }, "Pipelines deadlocked while acquiring connections!"); |
There was a problem hiding this comment.
If the regression returns, either Future.get times out and assertDoesNotThrow throws before shutdownNow() is reached, while the executor's non-daemon workers remain blocked in Semaphore.acquire() or at the barrier. The test process can therefore hang indefinitely instead of reporting the intended five-second failure; cancel the futures and shut down the executor from a finally block.
AGENTS.md reference: AGENTS.md:L8-L8
Useful? React with 👍 / 👎.
| } | ||
|
|
||
| connection.sendCommand(commandObject.getArguments()); | ||
| commandQueue.add(commandObject); |
There was a problem hiding this comment.
The public sendCommand(CommandArguments) and executeCommand(CommandObject) paths accept mutable command objects, but this now stores the object and reads its arguments only during a later sync(), after routing was already calculated. If a caller reuses the CommandArguments builder or mutates a supplied byte[] before syncing, the transmitted command can differ from the one used to select nodeKey, producing malformed commands or MOVED replies; the previous implementation serialized the arguments before returning from the pipeline call. Queue an immutable snapshot of the arguments instead.
Useful? React with 👍 / 👎.
…ndant pool borrows
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: b1736ca8f8
ℹ️ 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".
| } catch (RuntimeException jce) { | ||
| log.error("Error with connection to " + nodeKey, jce); | ||
| } finally { | ||
| IOUtils.closeQuietly(connection); |
There was a problem hiding this comment.
When callers split stateful commands across sync boundaries through the public sendCommand API—for example, MULTI followed by sync() and a later EXEC—this returns the first connection to the shared pool while it is still in transaction mode. Another borrower can then receive that connection and have its commands unexpectedly return QUEUED, while the pipeline's later EXEC may run on a different connection and fail with EXEC without MULTI; the previous implementation retained each node connection until the pipeline closed. Keep ownership of connections carrying session state, reset them before pooling, or reject stateful sequences that span syncs.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: eacaa5b31b
ℹ️ 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".
| connection = heldConnections.get(nodeKey); | ||
| if (connection == null) { | ||
| connection = getConnection(nodeKey); | ||
| heldConnections.put(nodeKey, connection); |
There was a problem hiding this comment.
When two pipelines concurrently sync against pools with only one available connection per shard, one pipeline's worker can retain shard A here while waiting for shard B, while the other retains B and waits for A. Because these connections remain in heldConnections until the pipeline closes, neither sync() can return to perform that close, recreating the lock-inversion deadlock this change is intended to eliminate. Release ordinary batch connections after reading their replies, while separately handling or rejecting stateful command sequences that require affinity.
Useful? React with 👍 / 👎.
| private final Map<HostAndPort, Queue<Response<?>>> pipelinedResponses; | ||
| private final Map<HostAndPort, Connection> connections; | ||
| private final Map<HostAndPort, Queue<CommandObject<?>>> pipelinedCommands; | ||
| private final Map<HostAndPort, Connection> heldConnections; |
There was a problem hiding this comment.
For every multi-node sync, separate executor workers concurrently call get and put on this LinkedHashMap. Concurrent first inserts or linked-order updates can lose/corrupt an entry, so close() may never return one shard's connection to its pool; subsequent pipelines can then exhaust or block on that pool. Store connection ownership through thread-safe coordination rather than mutating a LinkedHashMap from shard workers.
Useful? React with 👍 / 👎.
…and pipeline edge cases" This reverts commit eacaa5b.
There was a problem hiding this comment.
Cursor Bugbot has reviewed your changes using default effort and found 1 potential issue.
Reviewed by Cursor Bugbot for commit e722177. Configure here.
| } | ||
| } | ||
| return shardA; | ||
| } |
There was a problem hiding this comment.
Test routing logic never matches, always returns same shard
High Severity
The getNodeKey override iterates over CommandArguments (which implements Iterable<Rawable>) and checks arg instanceof byte[]. Since the iterator yields Rawable wrapper objects (not raw byte[] arrays), the instanceof byte[] check is always false. The method always falls through to return shardA, routing all commands to a single shard. This means the test never exercises multi-node behavior and cannot actually verify the deadlock fix it claims to test.
Reviewed by Cursor Bugbot for commit e722177. Configure here.
|
@anshullakra007 Looking at the suggested approach is a breaking change and changes the behavior of |
@ggivo Thank you for the architectural guidance. I completely agree with preserving the legacy socket-writing behavior to avoid any breaking changes for existing ClusterPipeline users. To address this, I have: Completely reverted the edits to MultiNodePipelineBase and ClusterPipeline. Introduced a parallel, opt-in architecture (DeferredMultiNodePipelineBase and DeferredClusterPipeline) that implements the deferred connection acquisition to safely eliminate the lock-inversion deadlock. Exposed this safely via deferredPipelined() on JedisCluster and RedisClusterClient. Addressed all AI bot reviews regarding charset encoding (SafeEncoder) and corrected the test routing logic to deterministically verify the deadlock fix. I understand the team is heavily focused on the 8.10 server release right now, so please take your time. Let me know if the deferredPipelined naming convention aligns with the team's vision! |
|
This PR looks great! |
3 participants
What does this PR do?
Fixes a confirmed lock-inversion deadlock in
ClusterPipeline(and any pipelines extendingMultiNodePipelineBase) when accessing multiple shards concurrently under heavy load.Root Cause Analysis
Resolves Issue #4557.
When
MultiNodePipelineBase#appendCommand()processes commands targeting different shards, it used a Hold-and-Wait locking pattern. Specifically, it borrowed the connection from the target node's connection pool immediately and held it in an internal map untilpipeline.sync()orpipeline.close()was executed.If two pipelines executed concurrently with different target sequences (e.g., Pipeline 1 appends Command for Shard A then B; Pipeline 2 appends Command for Shard B then A), they permanently deadlocked waiting for the other pipeline to release its connection pool lock.
Architectural Fix
To break this circular wait condition, this PR eliminates eager connection acquisition during the command-building phase:
MultiNodePipelineBase#appendCommand()now only buffers theCommandObjectin memory grouped byHostAndPort.sync()phase.sync(), each worker thread securely borrows one connection, streams the entire buffered batch to the socket, reads the responses, and immediately releases the connection.A worker thread never holds a connection to Shard A while blocking on Shard B's pool. Deadlock is architecturally eliminated.
Testing & Verification
Added a deterministic JUnit integration test (
MultiNodePipelineDeadlockTest.java).The test configures a strict 1-connection pool across two mock shards and uses a
CyclicBarrierto enforce an exact lock-inversion trace scenario across two threads.TimeoutException.~250ms).Note
Medium Risk
Changes core cluster pipeline connection lifecycle and error propagation for all
MultiNodePipelineBasesubclasses; behavior should be equivalent for success paths but failure handling and timing under load differ from eager connection holding.Overview
Fixes a lock-inversion deadlock in
MultiNodePipelineBase(includingClusterPipeline) when multiple pipelines append to different shards in opposite order under a small connection pool.appendCommandno longer borrows or keeps per-shard connections; it only queuesCommandObjects andResponses byHostAndPort. Network I/O moves tosync(): each worker gets a connection, sends the buffered batch, reads replies, then closes the connection.sync()also aggregates the first worker failure via an atomic reference, clears pipeline state infinally, andclose()relies on that path instead of holding a connection map across the build phase.Adds
MultiNodePipelineDeadlockTest, which uses single-slot pools and aCyclicBarrierto reproduce the A-then-B vs B-then-A scenario and assert both threads finish within a timeout.Reviewed by Cursor Bugbot for commit e722177. Bugbot is set up for automated code reviews on this repo. Configure here.