fix(blocking): stop cross-shard BLPOP from destroying the next push (c10k A2/A3/A4/A5/A6) - #426
Open
TinDang97 wants to merge 2 commits into
Open
fix(blocking): stop cross-shard BLPOP from destroying the next push (c10k A2/A3/A4/A5/A6)#426TinDang97 wants to merge 2 commits into
TinDang97 wants to merge 2 commits into
Conversation
…c10k A2/A3/A4/A5/A6)
Cluster A of the c10k hardening review (tmp/C10K-HARDENING-REVIEW.md). Two
compounding defects made a timed-out cross-shard blocking command silently
eat the next push to that key at --shards > 1.
A3 — inverted cleanup condition. The tokio single-key path read
if is_remote && !matches!(result, Frame::Null) || matches!(result, Frame::Error(_))
which Rust parses as `(is_remote && !Null) || Error`. On a TIMEOUT the result
IS Frame::Null, so the BlockCancel was never sent and the owning shard kept
the WaitEntry forever — remote registrations carry `deadline: None`, so the
W6 deadline sweep never reaps them either. The same precedence bug drove a
LOCAL waiter's shutdown cleanup into the remote branch, computing
`target_index(shard, shard)`, which underflows to a panic on shard 0 and
misroutes the cancel to an unrelated shard otherwise. The monoio twin always
had the correct `if is_remote`.
A2 — the wake path then destroyed the element. Every wake popped first and
did `let _ = reply_tx.send(...)` second; for a ghost waiter the receiver is
gone, so the value was neither delivered, nor requeued, nor offered to the
next FIFO waiter. Redis guarantees delivered-or-stays. Fixed with two
defences: waiters whose client has disconnected are skipped BEFORE the
datastore is touched, and anything popped in the remaining race window is
restored — including BLMOVE's destination push and BLMPOP's multi-element
pops, in original order. XRead needs no undo (non-destructive) and
XReadGroup's PEL entry matches Redis's dead-consumer semantics.
A4/A5/A6 — dispatch integrity. BlockRegister/BlockCancel were pushed with a
bare try_push (tokio) or an uncapped `loop { try_push; sleep(10us) }` with no
shutdown arm (monoio). A dropped BlockRegister drops the reply sender with
it, so `BLPOP key 0` returned nil at t=0 (protocol violation); a dropped
BlockCancel leaked the owner-side waiter — the very ghost A2 then had to
defend against; and the uncapped retry turned a saturated owner shard into a
zombie connection that also blocked graceful shutdown. All four call sites
now share one bounded, shutdown-aware helper on the plane-wide
CROSS_SHARD_PUSH budget, notify the owning shard on success, and fail the
command loudly rather than blocking on a key nobody is watching.
Tests (red/green, both verified failing first):
- 5 unit tests in src/blocking/wakeup.rs covering BLPOP/BLMOVE/BLMPOP/
BZPOPMIN and dead-head-yields-to-live-waiter. All 5 failed pre-fix.
- tests/blocking_ghost_waiter.rs drives a real 4-shard server. With the
three defences temporarily reverted it fails with
"ghost:1: element was swallowed by a ghost waiter — LLEN :0";
its positive control (a woken waiter consumes exactly one element)
stays green throughout, so the suite is not failing incidentally.
Gates: cargo fmt --check; clippy --all-targets on monoio and on
runtime-tokio,jemalloc (zero warnings); lib tests 4449 monoio / 3614 tokio;
the new e2e green on both runtimes.
A1 (immortal blocked clients — the unauthenticated permanent-maxclients DoS
in the same cluster) is deliberately NOT in this commit: the only safe fix
threads a read-readiness arm and a byte-carry through both handlers, which
is its own change.
author: Tin Dang
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: defaults Review profile: CHILL Plan: Pro Plus Run ID: 📒 Files selected for processing (2)
🚧 Files skipped from review as they are similar to previous changes (2)
📝 WalkthroughWalkthroughThe PR hardens blocked-client wakeups by preserving popped data for disconnected receivers and adds shutdown-aware cross-shard blocking registration and cancellation for tokio and monoio paths. Unit and integration tests cover dead waiters and cross-shard BLPOP behavior. ChangesBlocking waiter hardening
Estimated code review effort: 4 (Complex) | ~45 minutes Sequence Diagram(s)sequenceDiagram
participant Client
participant BlockingCommand
participant RemoteShard
participant Wakeup
participant Database
Client->>BlockingCommand: issue cross-shard blocking command
BlockingCommand->>RemoteShard: deliver BlockRegister
RemoteShard-->>BlockingCommand: register waiter
RemoteShard->>Wakeup: notify matching waiter
Wakeup->>Database: remove list or sorted-set data
Wakeup-->>Client: send wake response
Wakeup->>Database: restore data if receiver is disconnected
BlockingCommand->>RemoteShard: deliver BlockCancel on timeout or shutdown
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
PR Summary by QodoFix cross-shard blocking waiter leaks and wakeup data loss (c10k A2–A6)
AI Description
Diagram
High-Level Assessment
Files changed (7)
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
🧹 Nitpick comments (4)
src/server/conn/blocking.rs (2)
106-127: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick winCleanup fans out cancels serially, each with the full ~0.5 s retry budget.
cancel_multikey_registrationsruns on the client's response path (both the end-of-command path and the registration-failure path). With N distinct remote owners wedged, the reply is delayed by up to N ×CROSS_SHARD_PUSH_MAX_RETRIES × CROSS_SHARD_PUSH_BACKOFF. Consider issuing the cancels concurrently (e.g.futures::future::join_all) so the worst case stays one budget rather than N.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/server/conn/blocking.rs` around lines 106 - 127, Update cancel_multikey_registrations to dispatch BlockCancel messages to all registered_remote_shards concurrently rather than awaiting each push_msg sequentially. Preserve the existing arguments, notifier selection, and retry behavior for each push, while ensuring the overall cleanup waits for the concurrent operations as one batch.
400-405: 🚀 Performance & Scalability | 🔵 Trivial | 💤 Low valueFailure path cancels shards that were never registered.
registered_remote_shardsis populated during staging, so afterbreakit also names owners whoseBlockRegisterwas never delivered. Those cancels are no-ops on the owner but each still consumes a full push budget. Track successful pushes instead.Note the same pattern exists in the monoio twin at Lines 679-693.♻️ Track delivered targets
- let mut registration_failed = false; - for (target, msg) in pending_remote { - if !push_block_msg(shutdown, dispatch_tx, shard_id, target, msg, None).await { - registration_failed = true; - break; - } - } + let mut registration_failed = false; + let mut delivered_remote_shards: Vec<usize> = Vec::new(); + for (target, msg) in pending_remote { + if !push_block_msg(shutdown, dispatch_tx, shard_id, target, msg, None).await { + registration_failed = true; + break; + } + if !delivered_remote_shards.contains(&target) { + delivered_remote_shards.push(target); + } + }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/server/conn/blocking.rs` around lines 400 - 405, Update the failure cleanup around the pending_remote push loop in the blocking connection flow to track only targets for which push_block_msg returned true, rather than using the staging-derived registered_remote_shards set. Use this successfully delivered-target collection when issuing cancellation pushes, and apply the same correction in the monoio twin.tests/blocking_ghost_waiter.rs (1)
132-137: 🎯 Functional Correctness | 🔵 Trivial | 💤 Low valueTighten integer-reply assertions.
starts_with(":1")also matches:10/:12, so an over-count would pass. Assert":1\r\n".Also applies to: 208-213
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/blocking_ghost_waiter.rs` around lines 132 - 137, Update the integer-reply assertions in the LLEN checks near the shown assertion and the corresponding occurrence to require an exact ":1\r\n" reply instead of using starts_with(":1"), preventing over-counts such as :10 or :12 from passing.src/blocking/wakeup.rs (1)
596-614: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueTest seeds the zset with the very API it validates.
dead_zset_waiter_does_not_consume_memberuseszset_restoreas the fixture, so a brokenzset_restorecould make both the seed and the assertion agree. Prefer a normal ZADD-equivalent insertion path (e.g.get_or_create_sorted_set+ insert) for setup.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/blocking/wakeup.rs` around lines 596 - 614, The test fixture in dead_zset_waiter_does_not_consume_member should avoid using zset_restore, since that is the behavior being indirectly validated. Seed the sorted set through the normal ZADD-equivalent path using get_or_create_sorted_set and its insertion API, then keep the existing wake and pop assertions unchanged.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@src/server/conn/blocking.rs`:
- Line 268: Align the tokio `push_block_msg` handling in both the normal path
and the cancel path around lines 334–341 with its actual wake behavior: either
add a task-local `Notify`/`ParkIdle` receive path to the tokio shard loop and
pass it through, or revise the A5 note to state that tokio uses the non-eventfd
drain/timer wake instead of `notify_one()`. Ensure both call sites document and
implement the same chosen behavior.
In `@tests/blocking_ghost_waiter.rs`:
- Around line 51-58: Update read_reply to accumulate data across multiple
TcpStream reads instead of assuming one read contains the complete response. Add
a small resp_is_complete helper that recognizes completion of the expected RESP
reply, including the required bulk elements, and stop reading only once complete
while preserving the socket timeout behavior.
---
Nitpick comments:
In `@src/blocking/wakeup.rs`:
- Around line 596-614: The test fixture in
dead_zset_waiter_does_not_consume_member should avoid using zset_restore, since
that is the behavior being indirectly validated. Seed the sorted set through the
normal ZADD-equivalent path using get_or_create_sorted_set and its insertion
API, then keep the existing wake and pop assertions unchanged.
In `@src/server/conn/blocking.rs`:
- Around line 106-127: Update cancel_multikey_registrations to dispatch
BlockCancel messages to all registered_remote_shards concurrently rather than
awaiting each push_msg sequentially. Preserve the existing arguments, notifier
selection, and retry behavior for each push, while ensuring the overall cleanup
waits for the concurrent operations as one batch.
- Around line 400-405: Update the failure cleanup around the pending_remote push
loop in the blocking connection flow to track only targets for which
push_block_msg returned true, rather than using the staging-derived
registered_remote_shards set. Use this successfully delivered-target collection
when issuing cancellation pushes, and apply the same correction in the monoio
twin.
In `@tests/blocking_ghost_waiter.rs`:
- Around line 132-137: Update the integer-reply assertions in the LLEN checks
near the shown assertion and the corresponding occurrence to require an exact
":1\r\n" reply instead of using starts_with(":1"), preventing over-counts such
as :10 or :12 from passing.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: 7dd83cd9-48e4-4a56-b414-ee16f39be20c
📒 Files selected for processing (7)
.gitignoreCHANGELOG.mdsrc/blocking/wakeup.rssrc/runtime/channel.rssrc/server/conn/blocking.rssrc/storage/db/accessors.rstests/blocking_ghost_waiter.rs
Code Review by Qodo
Context used✅ Compliance rules (platform):
55 rules 1. Cancel drop leaks waiters
|
![qodo-code-review[bot]](https://avatars.githubusercontent.com/in/484649?s=60&v=4){kind=small}
…cel loud (#426 review) Three review findings, none of them behaviour bugs in the shipped fix but all worth correcting. 1. `read_reply` in the ghost-waiter suite took ONE `read()` per reply. TCP gives no framing guarantee, so a multi-bulk BLPOP answer can arrive split and the `contains("first")` / `starts_with(":1")` assertions would fail spuriously. It now accumulates until the buffer parses as a complete RESP value (simple/error/integer, bulk incl. null, and nested arrays). The socket read timeout still bounds a wedged server, so a real hang fails the test rather than blocking it. 2. `cancel_multikey_registrations` discarded `push_block_msg`'s result. The push budget is bounded by design — A4 fixed the unbounded retry, so the only safe direction is to keep the bound — which means a shard wedged for the whole ~0.5 s budget can still drop a BlockCancel and leave the owner with a `deadline: None` entry its sweep will not reap. This is a memory leak, not data loss: since A2 every wake path checks `is_disconnected()` before touching the datastore and restores anything it popped, so a stale entry is skipped and pruned rather than swallowing an element. Now logged at warn with wait_id and both shard ids, so the leak is observable instead of silent. 3. The A5 note claimed a `notify_one()` kick that tokio never performs — tokio has no task-local Notify receive path and passes `notify: None`, discovering ring items through its own periodic drain. The comment now says what the code does rather than what monoio does. Gates (Linux VM): fmt; clippy -D warnings on default and runtime-tokio,jemalloc; lib 4471; blocking_ghost_waiter 2/2. author: Tin Dang
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What
Cluster A (wave 1) of the c10k hardening review. A timed-out cross-shard
blocking command was silently eating the next push to that key at
--shards > 1, and the cross-shard registration plane could drop controlmessages or retry them forever.
The bugs
A3 — inverted cleanup condition (data loss root cause)
src/server/conn/blocking.rs, tokio single-key path:Rust parses this as
(is_remote && !Null) || Error. Two consequences:Frame::Null, so theBlockCancelwasnever sent — the owning shard kept the
WaitEntryforever. Remoteregistrations carry
deadline: None, so the W6 deadline sweep never reapsthem either. Permanent ghost waiter.
Frame::Error) it took the remotebranch anyway and computed
target_index(shard, shard)— which underflowsto a panic on shard 0, and misroutes the cancel to an unrelated shard
otherwise.
The monoio twin always had the correct
if is_remote, so this reproducesonly under
runtime-tokio.A2 — the wake path then destroyed the element
Every wake path popped first and did
let _ = reply_tx.send(...)second. Fora ghost waiter the receiver is long gone, so the popped value was neither
delivered, nor requeued, nor offered to the next FIFO waiter. Redis
guarantees delivered-or-stays.
Fixed with two defences, because neither alone is sufficient:
is touched (
OneshotSender::is_disconnected), so the common case causes nomutation at all;
BLMOVE's destination push (both halves reversed) and BLMPOP's
multi-element pops, in original order.
XReadneeds no undo (non-destructive range read).XReadGroupleaves theentries in the stream and only records them in the PEL for a consumer that
vanished — which is exactly Redis's dead-consumer semantics, recoverable via
XAUTOCLAIM— so it takes the liveness pre-check only.A4 / A5 / A6 — dispatch integrity
BlockRegister/BlockCancelwere pushed with a baretry_push(tokio) oran uncapped
loop { try_push; sleep(10us) }with no shutdown arm (monoio):BlockRegisterdrops the reply sender with it, soBLPOP key 0returned nil at t=0 — a protocol violation;
BlockCancelleaked the owner-side waiter — the very ghost A2then had to defend against;
that also blocked graceful shutdown.
All four call sites now share one bounded, shutdown-aware helper using the
plane-wide
CROSS_SHARD_PUSH_MAX_RETRIES/CROSS_SHARD_PUSH_BACKOFFbudget (not a tighter blocking-specific one — the old monoio path retried
forever, so the only safe direction is toward the existing bound), notify the
owning shard on success, and fail loudly (
MOONERR blocking registration failed) rather than blocking on a key nobody is watching.Red/green evidence
Both test layers were verified failing first, not just passing after.
5 unit tests in
src/blocking/wakeup.rs(BLPOP / BLMOVE / BLMPOP /BZPOPMIN + dead-head-yields-to-live-waiter). All 5 failed pre-fix, e.g.
dead_list_waiter_does_not_consume_element ... a dead waiter is not a wakeup.tests/blocking_ghost_waiter.rsdrives a real 4-shard server over rawRESP. With the three defences temporarily reverted:
Its positive control (
woken_cross_shard_blpop_consumes_exactly_one_element)stayed green throughout that run, so the suite is not failing
incidentally — and it proves the fix wasn't implemented by simply never
delivering.
16 distinct keys are used because a connection is pinned to one shard by
SO_REUSEPORT; with 4 shards most keys hash elsewhere, which is the only
configuration that reproduces A3.
Gates
cargo fmt --checkcargo clippy --all-targets— monoio andruntime-tokio,jemalloc, zero warnings--test integration: 108 tokioNot in this PR
A1 — immortal blocked clients, the unauthenticated permanent-maxclients
DoS in the same cluster (
BLPOP k 0+ RST leaks the handler task, theWaitEntry, the registry entry and the slot, unreapable by disconnect,CLIENT KILLortimeout). The agreed fix adds a read-readiness select armthat performs a real read and carries any pipelined bytes back to the
handler's
read_buf— that threads through both handlers and is its ownchange. Tracked separately.
Summary by CodeRabbit