fix(blocking): spanning multi-key pops serve exactly once in key order; a serve racing the end of a wait is never lost (moon#1019, moon#1023)

[TinDang97]

Closes #1019. Closes #1023.

Decision implemented (from the owner): spanning BLPOP/BRPOP/BZPOPMIN/BZPOPMAX keep working across shards and are not refused with CROSSSLOT. An untagged BLPOP q1 q2 q3 0 worker loop at --shards > 1 answers the way standalone redis does. Spanning BLMPOP/BZMPOP are still refused with CROSSSLOT (from #1021, matching LMPOP/ZMPOP).

The claim protocol

One token per waiter that is registered on more than one thread: src/blocking/claim.rs, Arc<AtomicU8>.

                try_claim (a shard, with the element already popped)
    WAITING  ─────────────────────────────────────────────────▶  CLAIMED   (final)
       │
       │  settle (the waiter: timeout / shutdown / vanished peer / failed registration)
       ▼
     DEAD  (final)

Who does a CAS, and when

• Owner shards (wakers and register_group) first skip a settled waiter without touching the datastore (WaitEntry::is_settled). Otherwise they pop first and claim second, with the element in hand. If the owner wins, it sends the reply in the same synchronous stretch. If it loses, WakeUndo::restore puts the element back in that same stretch, where no other client can observe the round trip, and the waker moves on to the next waiter. Errors (-WRONGTYPE, and XREADGROUP's errors at registration) are answered only on a won claim too. So the invariant is: nobody sends Some(..) without winning, and at most one answer of any kind exists.
• The waiter settles the token only when its wait ended without a reply.

"A winner that finds the key empty" cannot happen here. The claim is attempted only after a successful pop, so there is no claimed-but-empty state to release and no retry protocol. That is what keeps the machine at three states with one CAS per side. The alternative, claim-then-pop, needs a CLAIMED → WAITING release, and a waiter settling during that window would have to spin. I rejected it.

The waker's one-waiter-per-push policy, and a lost claim. A lost claim is treated like a dead waiter: continue to the next waiter, not return. The pushed element is never spent on a waiter another shard already served. In addition, both destructive wakers now keep serving the key's waiters while it still has data. This also closes the review P3 in register_group, where each key was offered to the waker once. Separately from the claim, one RPUSH k a b used to answer one of two parked waiters and leave the other next to b until its timeout. Redis 8.6.1 answers both; this was verified with two redis-cli waiters on a live redis-server.

Argument order across shards (#1019, immediate path)

immediate_scan now stops at the first key this shard does not own, instead of skipping it. The remaining keys are cut into runs of consecutive same-owner keys and registered one run at a time, in argument order. Every run except the last is acknowledged (a new ack in BlockRegisterGroupPayload) before the next is sent, and registration stops early once the token is claimed. So a run is decided only after every earlier key's owner found that key empty and of the right type, and the owner can apply redis's full ladder, -WRONGTYPE included. The whole_command flag is gone because every run is now decisive. A later local run goes through the same register_group on the client's own thread.

The ack is also what keeps later arrivals linearizable. If an earlier key receives data after its owner looked, that owner serves the waiter itself, since the waiter is registered there. Because a push and its wake run in one synchronous stretch, it claims at the instant of the push, so a later owner's claim fails and that owner puts its element back.

Cost: one extra round trip per remote run that is not the last. Co-located keys (one run) and single-key waits send exactly the messages they sent before. A waiter whose keys are all local carries no token and keeps its fast path.

Alternative considered: a coordinator shard that probes the owners sequentially. It needs the same N round trips plus a second hop for every wake, and it still needs something like the token to arbitrate concurrent wakes. The shared claim needs neither.

Settling a wait that ends without a reply (#1023)

blocking_multikey::settle, used by both runtimes, the single-key and multi-key paths:

• DEAD: no shard served and none ever can. Every receiver is dropped as is. This is why the drain the issue asked for is unnecessary with a token: after DEAD, nothing of value can be buffered or arrive, because every send of Some follows a won claim.
• CLAIMED: the winner's reply is in flight on exactly one receiver. It is taken (without a timer if it is already buffered, otherwise bounded by XSHARD_REPLY_TIMEOUT), and then:
  • timeout: delivered. The client was served before the timeout was observed, which is what redis answers.
  • shutdown: delivered too. The connection is still alive and the reply is the only place the element still exists. I chose this over restoring, because a restore needs the owner to drain one more message during shutdown. This deviates from the issue's wording on purpose.
  • peer gone: the serve stands and is logged like a delivered reply (review F3; see "Review follow-up" below). The connection gets BlockingOutcome::ServedPeerGone, runs the tracking invalidation and the blocking_effect AOF/replication record, and closes. This is redis's behaviour: redis pops and propagates when it serves, and a client that disconnects with the reply in its output buffer loses it. The first cut restored the element instead (WakeUndo::from_reply + ShardMessage::BlockRestore); both are removed.
• No token (all local): remove_wait runs in the same synchronous stretch that ended the wait, then an exact now_or_never drain. This also covers the local case the issue did not name: select! can pick the timer over a receiver that already holds a reply.

Second-round follow-up (head 80885575)

The fixes are in c2b78584, then origin/main was merged in (a merge commit, no force-push). Each code fix was RED on the unfixed code and GREEN after (unit tests, release-fast):

item	RED	GREEN
P2-1: a later local run's registration was orphaned by a timeout	expiry_removes_undeadlined_sibling_registrations: "the undeadlined sibling registration outlived its timed-out waiter"	expire_timed_out now calls remove_wait for every timed-out id, so every registration of that waiter goes, including those with no deadline
P3-1: a waiter of the wrong family was answered nil	a_waiter_of_another_family_is_left_parked_on_a_wrong_typed_key: "zset waiter on a list: the waiter was answered from a key of another type"; every_waiter_of_another_family_stays_parked_in_order also RED	A list/zset waker whose pop yields nothing puts the waiter back at the front of its queue (BlockingRegistry::requeue_front) unanswered, then stops. FIFO is intact. deliver now takes the Frame by value, so the path that answered None is gone. This covers a wrong-typed key, an absent key and a cold miss alike, with no type probe on the push path. The stream waker already decides before it pops, so it needed no change.
P3-2: review finding codes in comments/tests	~30 occurrences	rewritten as the invariants they stood for; issue numbers kept
P3-3: an overclaim about the record of a gone client's serve	CHANGELOG, finish_unserved and BlockingOutcome::ServedPeerGone said it "keeps one history"	Reworded. The record takes the delivered path and inherits its gaps: for a key another shard owns, replay drops it (moon#1056), and on runtime-tokio it is appended to the AOF only, never to replication. Verified in handler_sharded/mod.rs (only aof_pool, no record_local_write).

Out of scope, as instructed: the unbounded ack wait for timeout-0, the env-var stall hook, loom being run by hand, and the BLMOVE destination wake.

Gates at 80885575 (after the merge):

• cargo fmt --check passes.
• cargo clippy --all-targets -D warnings passes on monoio and on tokio.
• --lib blocking unit tests pass: 133 on monoio (filter blocking bsc claim std_), 93 on tokio (filter blocking).
• Every blocking_* suite, blocking_spanning_claim (8/8) and loom_blocking_claim pass on both runtimes, against pinned release-fast binaries (MOON_BIN).

Review follow-up (head 37a4def5)

Fixes for the adversarial review of ffacf2ef. Measured on macOS, --shards 4, with pinned binaries. tests/blocking_spanning_claim.rs used to hard-code CARGO_BIN_EXE_moon and ignore MOON_BIN; it now goes through common::find_moon_binary.

finding	RED (before)	GREEN (after)
F1 a wake on an absent key answered a parked BLPOP k 0 with nil	unit a_wake_on_an_absent_key_answers_nobody: "list: a parked waiter was answered from an absent key"	passes (list + zset); wakers return early when !db.exists(key)
F2 run ack bounded only by shutdown + 30 s	bsc7 (BLPOP k1 k2 k3 0.5, owner stalled 3 s by MOON_TEST_BLOCK_ACK_STALL_MS): answered after 3.009 s / 6.025 s (monoio), 3.003 s / 6.022 s (tokio)	nil in < 1.5 s, both runtimes. await_run_ack races the client deadline (→ WaitEnd::Timeout) and shutdown (→ the normal shutdown reply). register_runs returns Result<(), WaitEnd> and the caller settles it like any other end
F3 peer-gone restore over third-party logged writes	bsc8 against the ffacf2ef binary: 6/6 served-then-disconnected waiters had a put back over later writes. With RPUSH b; LPOP the master held [a] (the AOF replays [b]); a DEL was undone	0/6 on both runtimes: the serve stands and is logged (below)
F4 lost-claim put-back after an emptying pop dropped the TTL	unit a_lost_claim_put_back_keeps_the_ttl: "the put-back dropped the TTL" (left None)	passes (list + zset), through the real deliver. Also covered: a won claim whose send fails puts back with the TTL; a surviving key's TTL is left alone. Encoding is not preserved: the key is recreated in its natural encoding
F6 loom modelled a hand-copied token	the model could not catch a bug in claim.rs	the real ClaimToken via #[path], plus the won-claim-send-fails model; mutations above fail it
P3 BlockRegister for a settled claim	registered a ghost / answered an error nobody reads	early return if !claim.is_open()
P3 no-token drain capped at 1024	a reply behind ≥1024 resolved receivers was missed	drain stops only after two consecutive idle polls; unit test at 0/1/1023/1024/5000 closed receivers

F3 — which semantics, and why. I chose redis semantics: the serve stands and is logged like a delivered one. I did not make the restore a logged write, because that cannot give one history. The owner's pop is never logged when it happens; its record is written later by the connection, from the reply. A restore lands after whatever other clients wrote and logged meanwhile, on top of a pop the log never saw. Logging the restore only adds a record to a history that is already missing the pop.

Not fixed here — pre-existing on origin/main, filed as moon#1056. While measuring F3 against the AOF I found that the blocking_effect record (moon#827) of any cross-shard blocking serve is appended to the connection's shard AOF, not the key owner's, and per-shard replay drops it.

Repro on origin/main 75375155, pinned binary:

1. --shards 4 --appendonly yes --appendfsync always.
2. For each of 4 keys: BLPOP k 0, delivered by RPUSH k a; then RPUSH k b; LPOP k.
3. kill -9 and restart.

The master held []; the replay gave [b] for the 3 keys on other shards. So every delivered cross-shard BLPOP comes back after a restart. There is a related ordering gap even on one shard: the record is written after the connection receives the reply, so a third party's write in between can be logged first. For example, after LPUSH c the master holds [c,b] and the replay [a,b]. For the same reason, bsc8 asserts the master, not the AOF. Fixing this means logging the pop on the owner, in the stretch that pops. That touches every wake site and the moon#827 record path, which is out of scope for this PR.

Other limits:

• A vanished peer is not watched while a run's ack is pending. The wait ends at the deadline or shutdown, and the peer is noticed afterwards.
• Shutdown during the ack wait is covered by code review and the Result plumbing, not by an integration test.
• With no deadline (BLPOP … 0), the ack wait is no longer capped at 30 s. It ends on the ack, on a dropped ack sender (→ RegisterFailed), or on shutdown.

Gates at 37a4def5:

• cargo fmt --check and cargo clippy --all-targets -D warnings pass on both runtimes.
• Every blocking_* suite and loom_blocking_claim pass on both runtimes; the spanning suite passes 8/8.
• Loom under cfg(loom): 3/3.
• scripts/test-consistency.sh --shards 4 (monoio): 1185/1191. The same 6 non-blocking rows fail on origin/main (5 tracking: controls and ROLE on a master). Main also fails the 4 moon#962 … span rows that this PR fixes.

Evidence

All on native macOS (aarch64), release-fast. Nothing here was measured on Linux. io_uring and SO_REUSEPORT placement change which connection lands where, but not the protocol.

RED → GREEN, tests/blocking_spanning_claim.rs, --shards 4. RED was measured on #1021's tip plus the test hook only; its blocking files are byte-identical to main's.

test	monoio before	tokio before	after (both runtimes)
bsc1: immediate spanning pop differs from redis (3 seeds × 4 cmds × 16 placements)	91/192 (67 wrong key, 24 extra pop)	95/192 (84 wrong key, 10 extra pop, 1 both)	0/192
bsc2: spanning -WRONGTYPE ladder skipped	14/16	10/16	0/16
bsc3: 3 concurrent pushes to 3 owners, pushes = pop + LLEN	39/80 lost (63/80 in an earlier run)	64/80 (57/80 earlier)	0/80
bsc4: push racing a timeout (settle window held open)	13/16 lost	13/16	0/16, with ≥6 served inside the window (asserted, so it cannot pass vacuously)
bsc5 (first cut): push racing a disconnect	14/16 lost	14/16	0/16. Superseded by the F3 semantics; bsc5 now checks that a push after an observed disconnect stays in the key (regression guard, passes before and after)
bsc6: one RPUSH k a b, two waiters	(see mutation)		0/4 owners

GREEN held on 3 consecutive runs per runtime (parallel test threads). The window in bsc4/bsc5 comes from MOON_TEST_BLOCK_SETTLE_DELAY_MS, read once through a OnceLock, following the MOON_TEST_AOF_FSYNC_STALL_MS pattern.

Every new guard fails when removed:

• Ignoring the claim in deliver: bsc1 83/192, bsc3 48/80.
• No run acknowledgement: bsc1 87/192, bsc2 RED, and bsc3 stays green. This shows order and exactly-once are guarded independently.
• restore_undelivered as a no-op (first cut, since removed): bsc5 10/16.
• One waiter per push, which is main's behaviour: bsc6 fails 4/4 owners, and the second waiter times out after 2.0 s beside an element.

Loom: tests/loom_blocking_claim.rs compiles the REAL src/blocking/claim.rs through #[path]; under cfg(loom) that file takes loom's Arc/AtomicU8, and build.rs declares the cfg. Three models: two owners (pop, claim, send-or-undo) racing a waiter that settles; two owners serving a live waiter; and a won claim whose send fails because the receiver closed without a settle, which must put its element back. Checked: at most one send; DEAD ⇒ no send ever, including afterwards; CLAIMED ⇒ exactly one reply taken; conservation. Exhaustive under cfg(loom) (3/3 pass). If try_claim in claim.rs is changed from a CAS to load-then-store, all 3 models fail. If the failed-send put-back is removed, the send-fails model fails. Run it with cargo rustc --profile release-fast --test loom_blocking_claim -- --cfg loom, then run the built binary. RUSTFLAGS=--cfg loom leaks into hyper-util and does not build.

Gates, both runtimes:

• cargo clippy --all-targets -D warnings and cargo fmt --check are clean.
• cargo clippy --all-targets -D warnings passes after rebasing onto 75375155.
• Blocking suites are green: blocking_{exec_wakeup,ghost_waiter,multikey_cross_shard,peer_eof,pop_propagation_827,spanning_claim,stream_read,waiter_cannibalisation,wrongtype_immediate} and loom_blocking_claim.
• --lib blocking unit tests (filter blocking): 108 on monoio, 122 on tokio.
• Full --lib: 1 failure on each runtime, gate_is_skipped_with_spill_sender_when_no_limit_is_configured (lib test gate_is_skipped_with_spill_sender_when_no_limit_is_configured fails on main under parallel execution #856). It fails the same way on fix(blocking): multi-key blocking pops serve exactly once; refuse spanning BLMPOP/BZMPOP (moon#989) #1021 and passes in isolation.
• blocking_list_timeout -- --ignored (it needs a server on :16479): 5/5 at --shards 1. At --shards 4, brpoplpush_legacy_alias gets moon#570's CROSSSLOT (untagged src and dst on different shards; I confirmed the reply with redis-cli). That refusal runs before anything this PR touches.
• scripts/test-consistency.sh --shards 4 against redis 8.6.1: 0 blocking failures on either binary. The spanning blpop/brpop/bzpopmin/bzpopmax rows moved from colo-only to span+colo, and a CROSSSLOT for them now fails the row (mk_must_answer). The failures that remain are the same 6 fix(blocking): multi-key blocking pops serve exactly once; refuse spanning BLMPOP/BZMPOP (moon#989) #1021 recorded (5 tracking: controls and ROLE on a master). The tokio binary also fails 6 graph/text rows, which is expected because that leg builds without graph/text-index.

Dispatch paths

• command::dispatch / dispatch_read: not reached, because blocking commands are intercepted by the connection handlers.
• MULTI: blocking commands are queued as their non-blocking forms and run by blocking_txn, which does not use immediate_scan. It is unchanged.
• try_inline_dispatch: GET/SET only.
• Both runtimes' handle_blocking_command* share the staging, registration (register_runs) and settle code in blocking_multikey.rs.

Hygiene

• No new unsafe.
• No unwrap/expect in library code.
• No change to src/command, src/protocol, src/shard/event_loop.rs or src/io.
• flume oneshots for every cross-thread message, including the new ack; no Arc<Mutex>.
• src/server/conn/blocking.rs (already over the cap on main) grows by 26 lines net. The new helpers live in blocking_multikey.rs (676 lines). The waker-side claim tests live in src/blocking/claim_wake_tests.rs, so claim.rs stays self-contained for loom.
• ci-local.sh was not run: this sandboxed session has no moon-dev VM. The hosted matrix is dispatched.

[coderabbitai]

Warning

Review limit reached

Next included review available in 14 minutes.

Check out review usage here.

View limit details

Limit details: You’ve used the included review currently available.

You've used all free OSS reviews for now. Wait for the free limit to reset to keep reviewing this public repository.

Learn how review limits work.

Review configuration:

⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Advanced

Run ID: c65c3177-aef6-4af5-91f7-a621185256e6

📥 Commits

Reviewing files that changed from the base of the PR and between a1e1366 and 8088557.

📒 Files selected for processing (18)

• CHANGELOG.md
• build.rs
• docs/commands.md
• scripts/test-consistency.sh
• src/blocking/claim.rs
• src/blocking/claim_wake_tests.rs
• src/blocking/group.rs
• src/blocking/mod.rs
• src/blocking/wakeup.rs
• src/server/conn/blocking.rs
• src/server/conn/blocking_multikey.rs
• src/server/conn/blocking_tests.rs
• src/server/conn/handler_monoio/dispatch.rs
• src/server/conn/handler_sharded/mod.rs
• src/shard/dispatch.rs
• src/shard/spsc_handler.rs
• tests/blocking_spanning_claim.rs
• tests/loom_blocking_claim.rs

---

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.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands.}

[TinDang97]

Marked ready. Both review rounds are addressed: the first round's findings were fixed earlier, and the second round's P2 (timed-out siblings left registered), wrong-family nil, finding-code scrub and CHANGELOG wording were fixed at 8088557, with red→green unit tests. Full dispatch run 35421740138 is green. Follow-ups tracked elsewhere: #1056 (cross-shard pop logged on the waiter's shard) and #1059 (BLMOVE wake doesn't reach destination waiters).

[qodo-code-review]

ⓘ Qodo reviews are paused because the subscription is no longer active. Ask your workspace admin to reactivate the subscription to resume reviews. Manage billing