diff --git a/.gitignore b/.gitignore
index 3a3efb0a..7e0f9f9a 100644
--- a/.gitignore
+++ b/.gitignore
@@ -101,6 +101,7 @@ tmp/
/target-linux/
/target-linux-tokio/
/target-tokio/
+/target-tokio2/
/target-fast/
/target-check-tokio/
/target-check-monoio/
diff --git a/CHANGELOG.md b/CHANGELOG.md
index f4a6b115..7b6397da 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -6,6 +6,37 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
## [Unreleased]
+### Fixed
+- **A timed-out cross-shard `BLPOP` no longer eats the next push to that
+ key (c10k hardening A2/A3).** Two defects compounded into silent data
+ loss at `--shards > 1`. First, the tokio single-key cleanup condition
+ `is_remote && !matches!(result, Frame::Null) || matches!(result, Frame::Error(_))`
+ parses as `(is_remote && !Null) || Error`, so on a **timeout** — where
+ the result *is* `Frame::Null` — the `BlockCancel` was never sent and the
+ owning shard kept the `WaitEntry` forever (remote entries carry
+ `deadline: None`, so the deadline sweep never reaps them). Second, when
+ a later push found that ghost waiter, every wake path popped the element
+ and then did `let _ = reply_tx.send(...)`: the receiver was long gone, so
+ the value was neither delivered, nor requeued, nor offered to the next
+ FIFO waiter — it simply vanished. Wakes now skip waiters whose client has
+ disconnected *before* touching the datastore, and restore anything popped
+ if the receiver drops in the remaining race window (including BLMOVE's
+ destination push and BLMPOP's multi-element pops, in order). The same
+ boolean bug also sent a local waiter's shutdown cleanup through
+ `target_index(shard, shard)`, which underflows to a panic on shard 0.
+- **Blocking registrations are no longer silently dropped or retried
+ forever (c10k hardening A4/A5/A6).** Cross-shard `BlockRegister` /
+ `BlockCancel` messages were pushed with a bare `try_push` (tokio) or an
+ uncapped `loop { try_push; sleep(10µs) }` with no shutdown arm (monoio).
+ A dropped `BlockRegister` drops the reply sender with it, so `BLPOP key 0`
+ returned nil at t=0 — a protocol violation; a dropped `BlockCancel` leaked
+ the owner-side waiter permanently; and the uncapped retry turned a
+ saturated owner shard into a zombie connection that also blocked graceful
+ shutdown. All four call sites now use the plane-wide bounded,
+ shutdown-aware push, notify the owning shard on success, and fail the
+ command loudly (`MOONERR blocking registration failed`) rather than
+ blocking on a key nobody is watching.
+
## [0.8.4] — 2026-07-29
### Added
diff --git a/src/blocking/wakeup.rs b/src/blocking/wakeup.rs
index 6997e4fc..205a9769 100644
--- a/src/blocking/wakeup.rs
+++ b/src/blocking/wakeup.rs
@@ -5,6 +5,85 @@ use crate::command::sorted_set::format_score_bytes;
use crate::framevec;
use crate::protocol::Frame;
use crate::storage::Database;
+
+/// What a wake attempt consumed from the datastore, so it can be put back if
+/// the woken client turns out to be gone.
+///
+/// c10k hardening A2: every wake path used to `pop` first and only then
+/// `let _ = reply_tx.send(...)`. A failed send — the overwhelmingly common
+/// case for a client that RST'd or timed out while its registration was still
+/// live on this shard — silently destroyed the popped element: not delivered,
+/// not requeued, not offered to the next FIFO waiter. Redis guarantees an
+/// element is either delivered or stays in the key.
+///
+/// Two defences, because neither alone is sufficient:
+/// 1. `is_disconnected()` is checked BEFORE touching the datastore, so a
+/// known-dead waiter never causes a mutation at all (the common case);
+/// 2. this undo covers the residual race — the receiver can drop between
+/// that check and the send.
+enum WakeUndo {
+ /// Values popped from the FRONT of the key, in pop order.
+ ListFront(smallvec::SmallVec<[bytes::Bytes; 4]>),
+ /// Values popped from the BACK of the key, in pop order.
+ ListBack(smallvec::SmallVec<[bytes::Bytes; 4]>),
+ /// BLMOVE: one value left the key via `wherefrom` and was pushed onto
+ /// `destination` via `whereto`. Undoing means reversing BOTH halves.
+ Moved {
+ destination: Bytes,
+ wherefrom: Direction,
+ whereto: Direction,
+ },
+ /// (member, score) pairs popped from the sorted set at the key.
+ Zset(smallvec::SmallVec<[(bytes::Bytes, f64); 4]>),
+}
+
+impl WakeUndo {
+ /// Put everything back exactly where it came from.
+ fn restore(self, db: &mut Database, key: &Bytes) {
+ match self {
+ // Pops came off the front in order [v0, v1, ..]; pushing them
+ // back front-first in REVERSE order restores the original
+ // sequence (push v_n first, v0 last => v0 ends up at the front).
+ WakeUndo::ListFront(vals) => {
+ for v in vals.into_iter().rev() {
+ db.list_push_front(key, v);
+ }
+ }
+ WakeUndo::ListBack(vals) => {
+ for v in vals.into_iter().rev() {
+ db.list_push_back(key, v);
+ }
+ }
+ WakeUndo::Moved {
+ destination,
+ wherefrom,
+ whereto,
+ } => {
+ // Take back the element we just pushed onto the destination.
+ // It is still at the end we pushed it to: this runs on the
+ // shard thread with no await in between, so nothing else can
+ // have touched the list.
+ let moved = match whereto {
+ Direction::Left => db.list_pop_front(&destination),
+ Direction::Right => db.list_pop_back(&destination),
+ };
+ if let Some(v) = moved {
+ match wherefrom {
+ Direction::Left => db.list_push_front(key, v),
+ Direction::Right => db.list_push_back(key, v),
+ }
+ }
+ }
+ // Sorted sets are order-free: reinsertion order is irrelevant.
+ WakeUndo::Zset(pairs) => {
+ for (member, score) in pairs {
+ db.zset_restore(key, member, score);
+ }
+ }
+ }
+ }
+}
+
/// Called after LPUSH/RPUSH successfully adds elements to a list key.
/// Pops the first waiter (FIFO) and executes the appropriate pop operation.
/// Returns true if a blocked client was woken (element was consumed by the waiter).
@@ -22,29 +101,48 @@ pub fn try_wake_list_waiter(
Some(w) => w,
None => return false,
};
- let wait_id = waiter.wait_id;
+ let crate::blocking::WaitEntry {
+ wait_id,
+ cmd,
+ reply_tx,
+ ..
+ } = waiter;
+
+ // A2: never mutate the datastore on behalf of a waiter whose client
+ // is already gone. Reap the registration and move to the next waiter
+ // with the key untouched.
+ if reply_tx.is_disconnected() {
+ registry.remove_wait(wait_id);
+ continue;
+ }
// Execute the pop based on command type
- let result = match &waiter.cmd {
+ let (result, undo) = match &cmd {
BlockedCommand::BLPop => {
// Pop from left, return [key, value]
- let val = db.list_pop_front(key);
- val.map(|v| {
- Frame::Array(framevec![
- Frame::BulkString(key.clone()),
- Frame::BulkString(v),
- ])
- })
+ match db.list_pop_front(key) {
+ Some(v) => (
+ Some(Frame::Array(framevec![
+ Frame::BulkString(key.clone()),
+ Frame::BulkString(v.clone()),
+ ])),
+ Some(WakeUndo::ListFront(smallvec::smallvec![v])),
+ ),
+ None => (None, None),
+ }
}
BlockedCommand::BRPop => {
// Pop from right, return [key, value]
- let val = db.list_pop_back(key);
- val.map(|v| {
- Frame::Array(framevec![
- Frame::BulkString(key.clone()),
- Frame::BulkString(v),
- ])
- })
+ match db.list_pop_back(key) {
+ Some(v) => (
+ Some(Frame::Array(framevec![
+ Frame::BulkString(key.clone()),
+ Frame::BulkString(v.clone()),
+ ])),
+ Some(WakeUndo::ListBack(smallvec::smallvec![v])),
+ ),
+ None => (None, None),
+ }
}
BlockedCommand::BLMove {
destination,
@@ -55,17 +153,27 @@ pub fn try_wake_list_waiter(
Direction::Left => db.list_pop_front(key),
Direction::Right => db.list_pop_back(key),
};
- val.map(|v| {
- // Push to destination
- match whereto {
- Direction::Left => db.list_push_front(destination, v.clone()),
- Direction::Right => db.list_push_back(destination, v.clone()),
+ match val {
+ Some(v) => {
+ // Push to destination
+ match whereto {
+ Direction::Left => db.list_push_front(destination, v.clone()),
+ Direction::Right => db.list_push_back(destination, v.clone()),
+ }
+ (
+ Some(Frame::BulkString(v)),
+ Some(WakeUndo::Moved {
+ destination: destination.clone(),
+ wherefrom: *wherefrom,
+ whereto: *whereto,
+ }),
+ )
}
- Frame::BulkString(v)
- })
+ None => (None, None),
+ }
}
BlockedCommand::BLMPop { dir, count } => {
- let mut elems = smallvec::SmallVec::<[Frame; 16]>::new();
+ let mut popped = smallvec::SmallVec::<[Bytes; 4]>::new();
let n = *count as usize;
for _ in 0..n {
let val = match dir {
@@ -73,33 +181,49 @@ pub fn try_wake_list_waiter(
Direction::Right => db.list_pop_back(key),
};
match val {
- Some(v) => elems.push(Frame::BulkString(v)),
+ Some(v) => popped.push(v),
None => break,
}
}
- if elems.is_empty() {
- None
+ if popped.is_empty() {
+ (None, None)
} else {
- let elem_vec: Vec = elems.into_vec();
- Some(Frame::Array(framevec![
- Frame::BulkString(key.clone()),
- Frame::Array(elem_vec.into()),
- ]))
+ let elem_vec: Vec =
+ popped.iter().cloned().map(Frame::BulkString).collect();
+ let undo = match dir {
+ Direction::Left => WakeUndo::ListFront(popped),
+ Direction::Right => WakeUndo::ListBack(popped),
+ };
+ (
+ Some(Frame::Array(framevec![
+ Frame::BulkString(key.clone()),
+ Frame::Array(elem_vec.into()),
+ ])),
+ Some(undo),
+ )
}
}
- _ => None, // BZPopMin/BZPopMax don't watch list keys
+ _ => (None, None), // BZPopMin/BZPopMax don't watch list keys
};
// Clean up all other key registrations for this wait_id
registry.remove_wait(wait_id);
if let Some(frame) = result {
- let _ = waiter.reply_tx.send(Some(frame));
- return true;
+ if reply_tx.send(Some(frame)).is_ok() {
+ return true;
+ }
+ // A2 residual race: the receiver dropped between the liveness
+ // check above and this send. Put the data back and offer it to
+ // the next waiter instead of destroying it.
+ if let Some(undo) = undo {
+ undo.restore(db, key);
+ }
+ continue;
}
// If pop returned None (list became empty -- shouldn't happen in single-threaded
// model but handle gracefully), try next waiter
- let _ = waiter.reply_tx.send(None);
+ let _ = reply_tx.send(None);
}
false
}
@@ -118,62 +242,93 @@ pub fn try_wake_zset_waiter(
Some(w) => w,
None => return false,
};
- let wait_id = waiter.wait_id;
-
- let result = match &waiter.cmd {
- BlockedCommand::BZPopMin => db.zset_pop_min(key).map(|(member, score)| {
- Frame::Array(framevec![
- Frame::BulkString(key.clone()),
- Frame::BulkString(member),
- Frame::BulkString(format_score_bytes(score)),
- ])
- }),
- BlockedCommand::BZPopMax => db.zset_pop_max(key).map(|(member, score)| {
- Frame::Array(framevec![
- Frame::BulkString(key.clone()),
- Frame::BulkString(member),
- Frame::BulkString(format_score_bytes(score)),
- ])
- }),
+ let crate::blocking::WaitEntry {
+ wait_id,
+ cmd,
+ reply_tx,
+ ..
+ } = waiter;
+
+ // A2: see try_wake_list_waiter — never pop for a dead client.
+ if reply_tx.is_disconnected() {
+ registry.remove_wait(wait_id);
+ continue;
+ }
+
+ let (result, undo) = match &cmd {
+ BlockedCommand::BZPopMin => match db.zset_pop_min(key) {
+ Some((member, score)) => (
+ Some(Frame::Array(framevec![
+ Frame::BulkString(key.clone()),
+ Frame::BulkString(member.clone()),
+ Frame::BulkString(format_score_bytes(score)),
+ ])),
+ Some(WakeUndo::Zset(smallvec::smallvec![(member, score)])),
+ ),
+ None => (None, None),
+ },
+ BlockedCommand::BZPopMax => match db.zset_pop_max(key) {
+ Some((member, score)) => (
+ Some(Frame::Array(framevec![
+ Frame::BulkString(key.clone()),
+ Frame::BulkString(member.clone()),
+ Frame::BulkString(format_score_bytes(score)),
+ ])),
+ Some(WakeUndo::Zset(smallvec::smallvec![(member, score)])),
+ ),
+ None => (None, None),
+ },
BlockedCommand::BZMPop { min, count } => {
let n = *count as usize;
- let mut elems = smallvec::SmallVec::<[Frame; 16]>::new();
+ let mut popped = smallvec::SmallVec::<[(Bytes, f64); 4]>::new();
for _ in 0..n {
- let popped = if *min {
+ let entry = if *min {
db.zset_pop_min(key)
} else {
db.zset_pop_max(key)
};
- match popped {
- Some((member, score)) => {
- elems.push(Frame::Array(framevec![
- Frame::BulkString(member),
- Frame::BulkString(format_score_bytes(score)),
- ]));
- }
+ match entry {
+ Some(pair) => popped.push(pair),
None => break,
}
}
- if elems.is_empty() {
- None
+ if popped.is_empty() {
+ (None, None)
} else {
- let elem_vec: Vec = elems.into_vec();
- Some(Frame::Array(framevec![
- Frame::BulkString(key.clone()),
- Frame::Array(elem_vec.into()),
- ]))
+ let elem_vec: Vec = popped
+ .iter()
+ .map(|(member, score)| {
+ Frame::Array(framevec![
+ Frame::BulkString(member.clone()),
+ Frame::BulkString(format_score_bytes(*score)),
+ ])
+ })
+ .collect();
+ (
+ Some(Frame::Array(framevec![
+ Frame::BulkString(key.clone()),
+ Frame::Array(elem_vec.into()),
+ ])),
+ Some(WakeUndo::Zset(popped)),
+ )
}
}
- _ => None, // List commands don't watch zset keys
+ _ => (None, None), // List commands don't watch zset keys
};
registry.remove_wait(wait_id);
if let Some(frame) = result {
- let _ = waiter.reply_tx.send(Some(frame));
- return true;
+ if reply_tx.send(Some(frame)).is_ok() {
+ return true;
+ }
+ // A2 residual race — restore and offer to the next waiter.
+ if let Some(undo) = undo {
+ undo.restore(db, key);
+ }
+ continue;
}
- let _ = waiter.reply_tx.send(None);
+ let _ = reply_tx.send(None);
}
false
}
@@ -197,9 +352,26 @@ pub fn try_wake_stream_waiter(
Some(w) => w,
None => return false,
};
- let wait_id = waiter.wait_id;
+ let crate::blocking::WaitEntry {
+ wait_id,
+ cmd,
+ reply_tx,
+ ..
+ } = waiter;
- let result = match &waiter.cmd {
+ // A2: skip waiters whose client is already gone. Unlike the list and
+ // zset paths there is no undo here, and none is needed: XRead is a
+ // non-destructive range read, and XReadGroup's delivery leaves the
+ // entries in the stream — only the PEL records them as delivered to a
+ // consumer that vanished, which is exactly Redis's dead-consumer
+ // semantics (recoverable via XAUTOCLAIM). The pre-check still matters
+ // so a dead waiter does not consume the wake that a live one needs.
+ if reply_tx.is_disconnected() {
+ registry.remove_wait(wait_id);
+ continue;
+ }
+
+ let result = match &cmd {
BlockedCommand::XRead { streams, count } => {
// Find this key's last_seen_id in the streams list
let last_seen = streams.iter().find(|(k, _)| k == key).map(|(_, id)| *id);
@@ -272,10 +444,172 @@ pub fn try_wake_stream_waiter(
registry.remove_wait(wait_id);
if let Some(frame) = result {
- let _ = waiter.reply_tx.send(Some(frame));
+ let _ = reply_tx.send(Some(frame));
return true;
}
- let _ = waiter.reply_tx.send(None);
+ let _ = reply_tx.send(None);
}
false
}
+
+#[cfg(test)]
+mod dead_waiter_tests {
+ use super::*;
+ use crate::blocking::WaitEntry;
+ use crate::storage::Database;
+
+ fn register(
+ reg: &mut BlockingRegistry,
+ key: &Bytes,
+ cmd: BlockedCommand,
+ ) -> crate::runtime::channel::OneshotReceiver> {
+ let wait_id = reg.next_wait_id();
+ let (tx, rx) = crate::runtime::channel::oneshot();
+ reg.register(
+ 0,
+ key.clone(),
+ WaitEntry {
+ wait_id,
+ cmd,
+ reply_tx: tx,
+ deadline: None,
+ },
+ );
+ rx
+ }
+
+ fn list_len(db: &mut Database, key: &Bytes) -> usize {
+ db.get_list(key).ok().flatten().map_or(0, |l| l.len())
+ }
+
+ /// A2: the woken client is already gone. The element must stay in the
+ /// list — the old code popped first, then dropped the value on the floor
+ /// when `reply_tx.send` failed.
+ #[test]
+ fn dead_list_waiter_does_not_consume_element() {
+ let mut reg = BlockingRegistry::new(0);
+ let mut db = Database::new();
+ let key = Bytes::from_static(b"mylist");
+
+ let rx = register(&mut reg, &key, BlockedCommand::BLPop);
+ drop(rx); // client vanished (RST / CLIENT KILL / timeout cleanup)
+
+ db.list_push_back(&key, Bytes::from_static(b"v1"));
+ let woke = try_wake_list_waiter(&mut reg, &mut db, 0, &key);
+
+ assert!(!woke, "a dead waiter is not a wakeup");
+ assert_eq!(list_len(&mut db, &key), 1, "element must survive");
+ }
+
+ /// A2 + FIFO: a dead head-of-queue waiter must yield the element to the
+ /// next live waiter, not swallow it.
+ #[test]
+ fn dead_waiter_yields_element_to_next_live_waiter() {
+ let mut reg = BlockingRegistry::new(0);
+ let mut db = Database::new();
+ let key = Bytes::from_static(b"mylist");
+
+ let dead = register(&mut reg, &key, BlockedCommand::BLPop);
+ drop(dead);
+ let live = register(&mut reg, &key, BlockedCommand::BLPop);
+
+ db.list_push_back(&key, Bytes::from_static(b"v1"));
+ let woke = try_wake_list_waiter(&mut reg, &mut db, 0, &key);
+
+ assert!(woke, "the live waiter must be served");
+ assert_eq!(list_len(&mut db, &key), 0, "element was delivered");
+ assert!(
+ matches!(live.try_recv(), Ok(Some(Frame::Array(_)))),
+ "live waiter received the element"
+ );
+ }
+
+ /// A2 for BLMOVE: a dead waiter must not leave the element stranded in
+ /// the destination list (the pop AND the push both have to be undone).
+ #[test]
+ fn dead_blmove_waiter_does_not_move_element() {
+ let mut reg = BlockingRegistry::new(0);
+ let mut db = Database::new();
+ let src = Bytes::from_static(b"src");
+ let dst = Bytes::from_static(b"dst");
+
+ let rx = register(
+ &mut reg,
+ &src,
+ BlockedCommand::BLMove {
+ destination: dst.clone(),
+ wherefrom: Direction::Left,
+ whereto: Direction::Right,
+ },
+ );
+ drop(rx);
+
+ db.list_push_back(&src, Bytes::from_static(b"v1"));
+ let woke = try_wake_list_waiter(&mut reg, &mut db, 0, &src);
+
+ assert!(!woke);
+ assert_eq!(list_len(&mut db, &src), 1, "source keeps the element");
+ assert_eq!(list_len(&mut db, &dst), 0, "destination untouched");
+ }
+
+ /// A2 for BLMPOP: every popped element must be restored, in order.
+ #[test]
+ fn dead_blmpop_waiter_restores_all_elements_in_order() {
+ let mut reg = BlockingRegistry::new(0);
+ let mut db = Database::new();
+ let key = Bytes::from_static(b"mylist");
+
+ let rx = register(
+ &mut reg,
+ &key,
+ BlockedCommand::BLMPop {
+ dir: Direction::Left,
+ count: 3,
+ },
+ );
+ drop(rx);
+
+ for v in [&b"a"[..], b"b", b"c"] {
+ db.list_push_back(&key, Bytes::copy_from_slice(v));
+ }
+ let woke = try_wake_list_waiter(&mut reg, &mut db, 0, &key);
+
+ assert!(!woke);
+ let list: Vec = db
+ .get_list(&key)
+ .ok()
+ .flatten()
+ .map(|l| l.iter().cloned().collect())
+ .unwrap_or_default();
+ assert_eq!(
+ list,
+ vec![
+ Bytes::from_static(b"a"),
+ Bytes::from_static(b"b"),
+ Bytes::from_static(b"c")
+ ],
+ "order must be preserved, not reversed"
+ );
+ }
+
+ /// A2 for the zset family.
+ #[test]
+ fn dead_zset_waiter_does_not_consume_member() {
+ let mut reg = BlockingRegistry::new(0);
+ let mut db = Database::new();
+ let key = Bytes::from_static(b"myzset");
+
+ let rx = register(&mut reg, &key, BlockedCommand::BZPopMin);
+ drop(rx);
+
+ db.zset_restore(&key, Bytes::from_static(b"m1"), 1.5);
+ let woke = try_wake_zset_waiter(&mut reg, &mut db, 0, &key);
+
+ assert!(!woke);
+ assert_eq!(
+ db.zset_pop_min(&key),
+ Some((Bytes::from_static(b"m1"), 1.5)),
+ "member must survive a dead waiter"
+ );
+ }
+}
diff --git a/src/runtime/channel.rs b/src/runtime/channel.rs
index 32f519b8..37541712 100644
--- a/src/runtime/channel.rs
+++ b/src/runtime/channel.rs
@@ -43,6 +43,17 @@ impl OneshotSender {
pub fn send(self, value: T) -> Result<(), T> {
self.tx.send(value).map_err(|e| e.into_inner())
}
+
+ /// True once the receiving half is gone (client disconnected, timed out,
+ /// or was cancelled), so `send` is guaranteed to fail.
+ ///
+ /// Used by the blocking-wakeup path to skip abandoned waiters BEFORE
+ /// mutating the datastore on their behalf: popping first and only then
+ /// discovering the dead receiver is what silently destroyed the popped
+ /// element (c10k hardening A2).
+ pub fn is_disconnected(&self) -> bool {
+ self.tx.is_disconnected()
+ }
}
pub struct OneshotReceiver {
diff --git a/src/server/conn/blocking.rs b/src/server/conn/blocking.rs
index c5f2db64..3195eb2d 100644
--- a/src/server/conn/blocking.rs
+++ b/src/server/conn/blocking.rs
@@ -21,6 +21,140 @@ use crate::storage::Database;
use super::util::extract_bytes;
+/// Returned to the client when a blocking command cannot be registered on the
+/// shard that owns its key. Fails the command instead of the two silent
+/// alternatives the old code had: replying nil at t=0 (protocol violation) or
+/// blocking forever on a key nobody is watching.
+const BLOCK_REGISTER_FAILED: &[u8] =
+ b"MOONERR blocking registration failed: owning shard not draining";
+
+/// Push a blocking-registry control message (`BlockRegister` / `BlockCancel`)
+/// to the shard that owns the key.
+///
+/// c10k hardening A4/A5/A6. Every call site was previously either a bare
+/// `try_push` — silently dropping the message when the ring was full — or an
+/// unbounded `loop { try_push; sleep(10µs) }` with no shutdown arm. Each
+/// failure mode is a real defect:
+/// * a dropped `BlockRegister` drops `reply_tx` with it, so the waiter's
+/// receiver disconnects and `BLPOP key 0` returns nil immediately;
+/// * a dropped `BlockCancel` leaks the owner-side `WaitEntry` permanently
+/// (remote entries carry `deadline: None`, so the W6 sweep never reaps
+/// them) — the leak that then feeds the A2 data-loss path;
+/// * the unbounded retry turned a saturated owner into a zombie connection
+/// that also prevented graceful shutdown from completing.
+///
+/// Uses the same budget as every other cross-shard push
+/// ([`CROSS_SHARD_PUSH_MAX_RETRIES`] × [`CROSS_SHARD_PUSH_BACKOFF`] ≈ 0.5 s):
+/// wedged-shard detection scale, generous enough not to false-reject a merely
+/// saturated shard. Deliberately NOT a tighter blocking-specific budget — a
+/// registration that gives up early is a client-visible error, and the old
+/// monoio path retried forever, so the only safe direction to move is toward
+/// the plane-wide bound.
+///
+/// Returns `false` if the message could not be delivered.
+async fn push_block_msg(
+ shutdown: &CancellationToken,
+ dispatch_tx: &Rc>>>,
+ my_shard: usize,
+ target_shard: usize,
+ msg: ShardMessage,
+ notify: Option<&channel::Notify>,
+) -> bool {
+ debug_assert_ne!(
+ my_shard, target_shard,
+ "blocking control message must target a remote shard"
+ );
+ let target_idx = ChannelMesh::target_index(my_shard, target_shard);
+ let mut slot = Some(msg);
+ let outcome = crate::shard::dispatch::push_with_backpressure(
+ shutdown,
+ crate::shard::dispatch::CROSS_SHARD_PUSH_MAX_RETRIES,
+ crate::shard::dispatch::CROSS_SHARD_PUSH_BACKOFF,
+ || {
+ let Some(m) = slot.take() else { return true };
+ // Borrow is scoped to this closure body, never held across the
+ // caller's `.await` between retries.
+ let mut producers = dispatch_tx.borrow_mut();
+ match producers[target_idx].try_push(m) {
+ Ok(()) => true,
+ Err(back) => {
+ slot = Some(back);
+ false
+ }
+ }
+ },
+ )
+ .await;
+ if outcome != crate::shard::dispatch::PushOutcome::Pushed {
+ return false;
+ }
+ // A5: the owner's event loop may be parked; without this kick the
+ // registration sits in the ring until the next periodic tick.
+ //
+ // `notify` is `Some` only on the monoio runtime, whose shard loops park in
+ // the driver and need the explicit eventfd kick. The tokio shard loop has
+ // no equivalent task-local `Notify` receive path — it discovers ring items
+ // through its own periodic drain — so tokio call sites pass `None` and
+ // accept up to one tick of latency rather than a lost registration. Wiring
+ // a `Notify` into the tokio loop is a separate change; this comment states
+ // what the code actually does today.
+ if let Some(n) = notify {
+ n.notify_one();
+ }
+ true
+}
+
+/// Tear down every registration a multi-key waiter holds: the local ones
+/// directly, the remote ones with a `BlockCancel` per owning shard.
+///
+/// A6: the cancels were bare `try_push`es, so a full ring leaked the owner's
+/// `WaitEntry` — which is exactly the ghost waiter the A2 wake path then had
+/// to defend against. `notifiers` is `Some` only on the monoio runtime, whose
+/// shard loops park and need an explicit kick.
+async fn cancel_multikey_registrations(
+ shutdown: &CancellationToken,
+ blocking_registry: &Rc>,
+ dispatch_tx: &Rc>>>,
+ shard_id: usize,
+ wait_id: u64,
+ registered_remote_shards: &[usize],
+ notifiers: Option<&[std::sync::Arc]>,
+) {
+ blocking_registry.borrow_mut().remove_wait(wait_id);
+ for &remote_shard in registered_remote_shards {
+ let delivered = push_block_msg(
+ shutdown,
+ dispatch_tx,
+ shard_id,
+ remote_shard,
+ ShardMessage::BlockCancel { wait_id },
+ notifiers.map(|n| &*n[remote_shard]),
+ )
+ .await;
+ if !delivered {
+ // The push budget is bounded by design (A4 — the old unbounded
+ // retry was itself the bug), so a shard wedged for the whole
+ // ~0.5 s budget, or a shutdown mid-cancel, can still drop this.
+ // The owner then keeps a `WaitEntry` that its deadline sweep will
+ // not reap (remote entries carry `deadline: None`).
+ //
+ // 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. Log it so the leak is
+ // observable instead of silent — a shard wedged this long is
+ // already an incident.
+ tracing::warn!(
+ wait_id,
+ from_shard = shard_id,
+ owner_shard = remote_shard,
+ "BlockCancel undelivered: owner shard not draining; its waiter \
+ entry leaks until a later push prunes it"
+ );
+ }
+ }
+}
+
/// Convert a blocking command to its non-blocking equivalent for MULTI/EXEC.
/// BLPOP key [key ...] timeout -> LPOP key [key ...]
/// BRPOP key [key ...] timeout -> RPOP key [key ...]
@@ -145,38 +279,37 @@ pub(crate) async fn handle_blocking_command(
// --- Single-key fast path: one registration, direct await (zero overhead) ---
if keys.len() == 1 {
let target = key_to_shard(&keys[0], num_shards);
+ let is_remote = target != shard_id;
let (reply_tx, reply_rx) = channel::oneshot::>();
- let wait_id = {
- let mut reg = blocking_registry.borrow_mut();
- let wait_id = reg.next_wait_id();
- if target == shard_id {
- // Local registration
- let entry = crate::blocking::WaitEntry {
+ let wait_id = blocking_registry.borrow_mut().next_wait_id();
+ if is_remote {
+ // Remote registration via SPSC — bounded and shutdown-aware
+ // (A4/A5). The borrow above is released before this await.
+ let msg = ShardMessage::BlockRegister(Box::new(
+ crate::shard::dispatch::BlockRegisterPayload {
+ db_index: selected_db,
+ key: keys[0].clone(),
wait_id,
cmd: blocked_cmd_factory(),
reply_tx,
- deadline,
- };
- reg.register(selected_db, keys[0].clone(), entry);
- } else {
- // Remote registration via SPSC
- let mut producers = dispatch_tx.borrow_mut();
- let target_idx = ChannelMesh::target_index(shard_id, target);
- let msg = ShardMessage::BlockRegister(Box::new(
- crate::shard::dispatch::BlockRegisterPayload {
- db_index: selected_db,
- key: keys[0].clone(),
- wait_id,
- cmd: blocked_cmd_factory(),
- reply_tx,
- },
- ));
- let _ = producers[target_idx].try_push(msg);
+ },
+ ));
+ if !push_block_msg(shutdown, dispatch_tx, shard_id, target, msg, None).await {
+ return Frame::Error(Bytes::from_static(BLOCK_REGISTER_FAILED));
}
- wait_id
- }; // reg borrow dropped
+ } else {
+ // Local registration
+ let entry = crate::blocking::WaitEntry {
+ wait_id,
+ cmd: blocked_cmd_factory(),
+ reply_tx,
+ deadline,
+ };
+ blocking_registry
+ .borrow_mut()
+ .register(selected_db, keys[0].clone(), entry);
+ }
- let is_remote = target != shard_id;
let result = if let Some(dl) = deadline {
tokio::select! {
res = reply_rx => {
@@ -208,11 +341,34 @@ pub(crate) async fn handle_blocking_command(
}
}
};
- // Cleanup remote registration on timeout/shutdown
- if is_remote && !matches!(result, Frame::Null) || matches!(result, Frame::Error(_)) {
- let mut producers = dispatch_tx.borrow_mut();
- let target_idx = ChannelMesh::target_index(shard_id, target);
- let _ = producers[target_idx].try_push(ShardMessage::BlockCancel { wait_id });
+ // Cleanup remote registration on timeout/shutdown.
+ //
+ // A3: this used to read
+ // `is_remote && !matches!(result, Frame::Null) || matches!(result, Frame::Error(_))`
+ // which Rust parses as `(is_remote && !Null) || Error`. Two bugs fell
+ // out of that precedence:
+ // * on TIMEOUT (`result == Frame::Null`) the cancel was never sent,
+ // so the owner shard kept a `WaitEntry` that nothing can ever
+ // reap (remote entries carry `deadline: None`, so the W6 deadline
+ // sweep skips them) — a permanent leak that also hands the next
+ // pusher a ghost waiter;
+ // * on a LOCAL wait ending in shutdown (`Frame::Error`) it took the
+ // remote branch anyway and computed `target_index(shard, shard)`,
+ // which underflows to a panic on shard 0 and misroutes the cancel
+ // to an unrelated shard otherwise.
+ // The correct condition is simply "did we register remotely" — the
+ // monoio twin below always had it right. A cancel that races a
+ // successful wake is a harmless no-op (`remove_wait` on an unknown id).
+ if is_remote {
+ let _ = push_block_msg(
+ shutdown,
+ dispatch_tx,
+ shard_id,
+ target,
+ ShardMessage::BlockCancel { wait_id },
+ None,
+ )
+ .await;
}
return result;
}
@@ -224,9 +380,13 @@ pub(crate) async fn handle_blocking_command(
FuturesUnordered::new();
let mut registered_remote_shards: Vec = Vec::new();
+ // A4/A5: build every registration under one borrow, but STAGE the remote
+ // ones and push them only after the borrow is released — the old code
+ // pushed while holding both borrows, which forced the bare `try_push`
+ // (no await possible, so no backpressure retry and no shutdown arm).
+ let mut pending_remote: Vec<(usize, ShardMessage)> = Vec::new();
{
let mut reg = blocking_registry.borrow_mut();
- let mut producers = dispatch_tx.borrow_mut();
wait_id = reg.next_wait_id();
for key in &keys {
@@ -245,7 +405,6 @@ pub(crate) async fn handle_blocking_command(
reg.register(selected_db, key.clone(), entry);
} else {
// Remote registration via SPSC
- let target_idx = ChannelMesh::target_index(shard_id, target);
let msg = ShardMessage::BlockRegister(Box::new(
crate::shard::dispatch::BlockRegisterPayload {
db_index: selected_db,
@@ -255,7 +414,7 @@ pub(crate) async fn handle_blocking_command(
reply_tx: tx,
},
));
- let _ = producers[target_idx].try_push(msg);
+ pending_remote.push((target, msg));
if !registered_remote_shards.contains(&target) {
registered_remote_shards.push(target);
}
@@ -263,6 +422,31 @@ pub(crate) async fn handle_blocking_command(
}
} // borrows dropped -- CRITICAL before await
+ // A5: a silently dropped registration leaves this waiter blocked on a key
+ // nobody is watching — it would sleep through data that IS available.
+ // Fail the whole command instead, after unwinding what did register.
+ 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;
+ }
+ }
+ if registration_failed {
+ cancel_multikey_registrations(
+ shutdown,
+ blocking_registry,
+ dispatch_tx,
+ shard_id,
+ wait_id,
+ ®istered_remote_shards,
+ None,
+ )
+ .await;
+ drop(receivers);
+ return Frame::Error(Bytes::from_static(BLOCK_REGISTER_FAILED));
+ }
+
// Await first successful result from any key/shard.
// FuturesUnordered may return Err (sender dropped by remove_wait cleanup) before
// returning the successful Ok. We must skip Err/None results and keep polling.
@@ -308,14 +492,16 @@ pub(crate) async fn handle_blocking_command(
};
// Cleanup: cancel all remaining registrations (local + remote)
- blocking_registry.borrow_mut().remove_wait(wait_id);
- if !registered_remote_shards.is_empty() {
- let mut producers = dispatch_tx.borrow_mut();
- for &remote_shard in ®istered_remote_shards {
- let target_idx = ChannelMesh::target_index(shard_id, remote_shard);
- let _ = producers[target_idx].try_push(ShardMessage::BlockCancel { wait_id });
- }
- }
+ cancel_multikey_registrations(
+ shutdown,
+ blocking_registry,
+ dispatch_tx,
+ shard_id,
+ wait_id,
+ ®istered_remote_shards,
+ None,
+ )
+ .await;
// Drop remaining receivers; remote senders get Err on send -- harmless
drop(receivers);
@@ -384,51 +570,46 @@ pub(crate) async fn handle_blocking_command_monoio(
// --- Single-key fast path: one registration, direct await (zero overhead) ---
if keys.len() == 1 {
let target = key_to_shard(&keys[0], num_shards);
+ let is_remote = target != shard_id;
let (reply_tx, reply_rx) = channel::oneshot::>();
- let wait_id = {
- let mut reg = blocking_registry.borrow_mut();
- let wait_id = reg.next_wait_id();
- if target == shard_id {
- let entry = crate::blocking::WaitEntry {
+ let wait_id = blocking_registry.borrow_mut().next_wait_id();
+ if is_remote {
+ // A4: the old `loop { try_push; sleep(10µs) }` had no retry cap
+ // and no shutdown arm — a saturated owner turned this into a
+ // zombie connection that also blocked graceful shutdown.
+ let msg = ShardMessage::BlockRegister(Box::new(
+ crate::shard::dispatch::BlockRegisterPayload {
+ db_index: selected_db,
+ key: keys[0].clone(),
wait_id,
cmd: blocked_cmd_factory(),
reply_tx,
- deadline,
- };
- reg.register(selected_db, keys[0].clone(), entry);
- } else {
- drop(reg);
- let mut producers = dispatch_tx.borrow_mut();
- let target_idx = ChannelMesh::target_index(shard_id, target);
- let msg = ShardMessage::BlockRegister(Box::new(
- crate::shard::dispatch::BlockRegisterPayload {
- db_index: selected_db,
- key: keys[0].clone(),
- wait_id,
- cmd: blocked_cmd_factory(),
- reply_tx,
- },
- ));
- let mut msg_pending = msg;
- loop {
- match producers[target_idx].try_push(msg_pending) {
- Ok(()) => {
- spsc_notifiers[target].notify_one();
- break;
- }
- Err(val) => {
- msg_pending = val;
- drop(producers);
- monoio::time::sleep(std::time::Duration::from_micros(10)).await;
- producers = dispatch_tx.borrow_mut();
- }
- }
- }
+ },
+ ));
+ if !push_block_msg(
+ shutdown,
+ dispatch_tx,
+ shard_id,
+ target,
+ msg,
+ Some(&spsc_notifiers[target]),
+ )
+ .await
+ {
+ return Frame::Error(Bytes::from_static(BLOCK_REGISTER_FAILED));
}
- wait_id
- }; // reg borrow dropped
+ } else {
+ let entry = crate::blocking::WaitEntry {
+ wait_id,
+ cmd: blocked_cmd_factory(),
+ reply_tx,
+ deadline,
+ };
+ blocking_registry
+ .borrow_mut()
+ .register(selected_db, keys[0].clone(), entry);
+ }
- let is_remote = target != shard_id;
let result = if let Some(dl) = deadline {
monoio::select! {
res = reply_rx => {
@@ -461,9 +642,17 @@ pub(crate) async fn handle_blocking_command_monoio(
}
};
if is_remote {
- let mut producers = dispatch_tx.borrow_mut();
- let target_idx = ChannelMesh::target_index(shard_id, target);
- let _ = producers[target_idx].try_push(ShardMessage::BlockCancel { wait_id });
+ // A6: bounded push + notify instead of a bare `try_push` that
+ // leaked the owner's WaitEntry whenever the ring was full.
+ let _ = push_block_msg(
+ shutdown,
+ dispatch_tx,
+ shard_id,
+ target,
+ ShardMessage::BlockCancel { wait_id },
+ Some(&spsc_notifiers[target]),
+ )
+ .await;
}
return result;
}
@@ -475,9 +664,11 @@ pub(crate) async fn handle_blocking_command_monoio(
FuturesUnordered::new();
let mut registered_remote_shards: Vec = Vec::new();
+ // A4/A5: stage remote registrations, push them after the borrow is
+ // released. See the tokio twin for the full rationale.
+ let mut pending_remote: Vec<(usize, ShardMessage)> = Vec::new();
{
let mut reg = blocking_registry.borrow_mut();
- let mut producers = dispatch_tx.borrow_mut();
wait_id = reg.next_wait_id();
for key in &keys {
@@ -496,7 +687,6 @@ pub(crate) async fn handle_blocking_command_monoio(
reg.register(selected_db, key.clone(), entry);
} else {
// Remote registration via SPSC
- let target_idx = ChannelMesh::target_index(shard_id, target);
let msg = ShardMessage::BlockRegister(Box::new(
crate::shard::dispatch::BlockRegisterPayload {
db_index: selected_db,
@@ -506,25 +696,7 @@ pub(crate) async fn handle_blocking_command_monoio(
reply_tx: tx,
},
));
- // SPSC push with backpressure retry (monoio pattern)
- let mut msg_pending = msg;
- loop {
- match producers[target_idx].try_push(msg_pending) {
- Ok(()) => {
- spsc_notifiers[target].notify_one();
- break;
- }
- Err(val) => {
- msg_pending = val;
- // Drop borrows before await
- drop(producers);
- drop(reg);
- monoio::time::sleep(std::time::Duration::from_micros(10)).await;
- reg = blocking_registry.borrow_mut();
- producers = dispatch_tx.borrow_mut();
- }
- }
- }
+ pending_remote.push((target, msg));
if !registered_remote_shards.contains(&target) {
registered_remote_shards.push(target);
}
@@ -532,6 +704,37 @@ pub(crate) async fn handle_blocking_command_monoio(
}
} // borrows dropped -- CRITICAL before await
+ let mut registration_failed = false;
+ for (target, msg) in pending_remote {
+ if !push_block_msg(
+ shutdown,
+ dispatch_tx,
+ shard_id,
+ target,
+ msg,
+ Some(&spsc_notifiers[target]),
+ )
+ .await
+ {
+ registration_failed = true;
+ break;
+ }
+ }
+ if registration_failed {
+ cancel_multikey_registrations(
+ shutdown,
+ blocking_registry,
+ dispatch_tx,
+ shard_id,
+ wait_id,
+ ®istered_remote_shards,
+ Some(spsc_notifiers),
+ )
+ .await;
+ drop(receivers);
+ return Frame::Error(Bytes::from_static(BLOCK_REGISTER_FAILED));
+ }
+
// Await first successful result from any key/shard.
// FuturesUnordered may return Err (sender dropped by remove_wait cleanup) before
// returning the successful Ok. We must skip Err/None results and keep polling.
@@ -578,15 +781,16 @@ pub(crate) async fn handle_blocking_command_monoio(
};
// Cleanup: cancel all remaining registrations (local + remote)
- blocking_registry.borrow_mut().remove_wait(wait_id);
- if !registered_remote_shards.is_empty() {
- let mut producers = dispatch_tx.borrow_mut();
- for &remote_shard in ®istered_remote_shards {
- let target_idx = ChannelMesh::target_index(shard_id, remote_shard);
- let _ = producers[target_idx].try_push(ShardMessage::BlockCancel { wait_id });
- spsc_notifiers[remote_shard].notify_one();
- }
- }
+ cancel_multikey_registrations(
+ shutdown,
+ blocking_registry,
+ dispatch_tx,
+ shard_id,
+ wait_id,
+ ®istered_remote_shards,
+ Some(spsc_notifiers),
+ )
+ .await;
// Drop remaining receivers; remote senders get Err on send -- harmless
drop(receivers);
diff --git a/src/storage/db/accessors.rs b/src/storage/db/accessors.rs
index bf416208..7ba0ccf8 100644
--- a/src/storage/db/accessors.rs
+++ b/src/storage/db/accessors.rs
@@ -640,6 +640,24 @@ impl Database {
Some(result)
}
+ /// Put a (member, score) pair back into a sorted set.
+ ///
+ /// The exact inverse of `zset_pop_min` / `zset_pop_max`, used by the
+ /// blocking-wakeup undo path when the woken client turns out to be gone
+ /// (c10k hardening A2). Memory accounting deliberately mirrors the pops:
+ /// they do not `credit_memory` for the removed member, so this does not
+ /// `charge_memory` for putting it back — the pair is a no-op on
+ /// `used_memory`, which is what keeps the estimate consistent across a
+ /// pop/restore cycle.
+ pub fn zset_restore(&mut self, key: &[u8], member: Bytes, score: f64) {
+ if let Ok((members, tree)) = self.get_or_create_sorted_set(key) {
+ if let Some(old) = members.insert(member.clone(), score) {
+ tree.remove(ordered_float::OrderedFloat(old), &member);
+ }
+ tree.insert(ordered_float::OrderedFloat(score), member);
+ }
+ }
+
/// Get or create a stream at the given key. Returns WRONGTYPE if key holds another type.
pub fn get_or_create_stream(&mut self, key: &[u8]) -> Result<&mut StreamData, Frame> {
self.get_or_create::(key)
diff --git a/tests/blocking_ghost_waiter.rs b/tests/blocking_ghost_waiter.rs
new file mode 100644
index 00000000..b85542f9
--- /dev/null
+++ b/tests/blocking_ghost_waiter.rs
@@ -0,0 +1,279 @@
+//! c10k hardening cluster A — the blocking registry must not leak waiters
+//! across shards, and a wake must never destroy the element it popped.
+//!
+//! Two defects made a timed-out cross-shard `BLPOP` eat the NEXT push to that
+//! key, permanently and silently:
+//!
+//! * **A3** — the tokio single-key cleanup read
+//! `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 entries carry
+//! `deadline: None`, so the W6 deadline sweep never reaps them either.
+//! * **A2** — when a later push found that ghost waiter, the wake path popped
+//! the element and then did `let _ = reply_tx.send(...)`. The receiver was
+//! long gone, so the value was neither delivered, nor requeued, nor offered
+//! to the next waiter. It simply vanished.
+//!
+//! This suite drives the pair end-to-end through a real 4-shard server.
+//! Sixteen distinct keys are used because a connection is pinned to one shard
+//! by SO_REUSEPORT: with 4 shards most keys hash to a *different* shard than
+//! the client's, which is the only configuration that reproduces A3.
+//!
+//! Note the runtime asymmetry: the monoio twin of the cleanup condition was
+//! always correct, so this test only turns red under `runtime-tokio`. CI runs
+//! both feature sets, and the invariant is worth pinning on both.
+//!
+//! Run with:
+//! cargo test --release --test blocking_ghost_waiter
+
+mod common;
+
+use std::io::{Read, Write};
+use std::net::TcpStream;
+use std::process::{Command, Stdio};
+use std::time::Duration;
+
+const SHARDS: &str = "4";
+/// Enough keys that several provably hash to a shard other than the one the
+/// client connection landed on.
+const KEYS: usize = 16;
+
+/// Serialize `parts` as a RESP array and write it.
+fn send(stream: &mut TcpStream, parts: &[&str]) {
+ let mut out = format!("*{}\r\n", parts.len());
+ for p in parts {
+ out.push_str(&format!("${}\r\n{}\r\n", p.len(), p));
+ }
+ stream.write_all(out.as_bytes()).expect("write command");
+}
+
+/// Read one reply, tolerating TCP segmentation.
+///
+/// A single `read()` is NOT enough even for small replies: TCP gives no
+/// framing guarantee, so a multi-bulk `BLPOP` answer (`*2\r\n$5\r\nghost...`)
+/// can arrive split across segments and a one-shot read would see a prefix,
+/// failing the assertions spuriously. Accumulate until the buffer parses as a
+/// complete RESP reply. The socket's read timeout still bounds a wedged
+/// server, so this fails the test rather than hanging it.
+fn read_reply(stream: &mut TcpStream) -> String {
+ let mut acc = Vec::new();
+ let mut buf = [0u8; 4096];
+ loop {
+ let n = stream.read(&mut buf).expect("read reply");
+ if n == 0 {
+ break; // EOF — return whatever we have; the caller asserts on it.
+ }
+ acc.extend_from_slice(&buf[..n]);
+ if resp_reply_complete(&acc) {
+ break;
+ }
+ }
+ String::from_utf8_lossy(&acc).into_owned()
+}
+
+/// Is `buf` exactly one complete RESP reply? Covers the reply shapes this
+/// suite issues: simple string / error / integer (one CRLF-terminated line),
+/// bulk string, null bulk, and one level of array.
+fn resp_reply_complete(buf: &[u8]) -> bool {
+ complete_len(buf).is_some()
+}
+
+/// Length of the complete RESP value at the front of `buf`, or `None` if more
+/// bytes are needed.
+fn complete_len(buf: &[u8]) -> Option {
+ let line_end = find_crlf(buf)?;
+ let header = &buf[..line_end];
+ let after = line_end + 2;
+ match header.first()? {
+ b'+' | b'-' | b':' => Some(after),
+ b'$' => {
+ let len: i64 = std::str::from_utf8(&header[1..]).ok()?.parse().ok()?;
+ if len < 0 {
+ return Some(after); // null bulk
+ }
+ let end = after + len as usize + 2;
+ (buf.len() >= end).then_some(end)
+ }
+ b'*' => {
+ let n: i64 = std::str::from_utf8(&header[1..]).ok()?.parse().ok()?;
+ if n < 0 {
+ return Some(after); // null array
+ }
+ let mut cursor = after;
+ for _ in 0..n {
+ cursor += complete_len(buf.get(cursor..)?)?;
+ }
+ Some(cursor)
+ }
+ _ => None,
+ }
+}
+
+fn find_crlf(buf: &[u8]) -> Option {
+ buf.windows(2).position(|w| w == b"\r\n")
+}
+
+fn connect(port: u16) -> TcpStream {
+ let stream = TcpStream::connect(("127.0.0.1", port)).expect("connect to moon");
+ stream
+ .set_read_timeout(Some(Duration::from_secs(10)))
+ .expect("set read timeout");
+ stream.set_nodelay(true).expect("set nodelay");
+ stream
+}
+
+#[test]
+fn timed_out_cross_shard_blpop_does_not_swallow_the_next_push() {
+ let bin = common::find_moon_binary();
+ if !bin.exists() {
+ eprintln!("skipping: no moon binary; build with `cargo build --release`");
+ return;
+ }
+
+ let (mut child, port) = common::spawn_listening(|port| {
+ let tmp_dir = std::env::temp_dir().join(format!("moon-ghost-waiter-{port}"));
+ let _ = std::fs::create_dir_all(&tmp_dir);
+ Command::new(&bin)
+ .args([
+ "--port",
+ &port.to_string(),
+ "--shards",
+ SHARDS,
+ "--admin-port",
+ "0",
+ "--appendonly",
+ "no",
+ // The repo's data volume hovers near the diskfull guard's 5%
+ // threshold; without this the server refuses writes and the
+ // assertions below fail for an unrelated reason.
+ "--disk-free-min-pct",
+ "0",
+ "--dir",
+ tmp_dir.to_str().unwrap(),
+ ])
+ .stdout(Stdio::null())
+ .stderr(Stdio::null())
+ .spawn()
+ .expect("spawn moon")
+ });
+ let tmp_dir = std::env::temp_dir().join(format!("moon-ghost-waiter-{port}"));
+
+ let result = std::panic::catch_unwind(|| {
+ let mut blocker = connect(port);
+ let mut pusher = connect(port);
+
+ for i in 0..KEYS {
+ let key = format!("ghost:{i}");
+
+ // 1. Block on an empty key and let the timeout expire. This is
+ // what leaves the ghost `WaitEntry` on the owning shard.
+ send(&mut blocker, &["BLPOP", &key, "0.2"]);
+ let reply = read_reply(&mut blocker);
+ assert!(
+ reply.starts_with("*-1") || reply.starts_with("$-1"),
+ "{key}: BLPOP should time out with nil, got {reply:?}"
+ );
+
+ // 2. Push AFTER the timeout. No waiter exists any more, so this
+ // element belongs to the keyspace, not to a wakeup.
+ send(&mut pusher, &["RPUSH", &key, "payload"]);
+ let reply = read_reply(&mut pusher);
+ assert!(
+ reply.starts_with(":1"),
+ "{key}: RPUSH should report length 1, got {reply:?}"
+ );
+
+ // 3. The element must still be there. Before the fix a ghost
+ // waiter on a remote shard consumed it into a dead oneshot.
+ send(&mut pusher, &["LLEN", &key]);
+ let reply = read_reply(&mut pusher);
+ assert!(
+ reply.starts_with(":1"),
+ "{key}: element was swallowed by a ghost waiter — LLEN {reply:?}"
+ );
+ }
+ });
+
+ common::sigkill(&mut child);
+ let _ = std::fs::remove_dir_all(&tmp_dir);
+ if let Err(panic) = result {
+ std::panic::resume_unwind(panic);
+ }
+}
+
+/// The other half of the same invariant: a waiter that is served normally
+/// must still consume its element exactly once, so the fix above cannot have
+/// been implemented by simply never delivering.
+#[test]
+fn woken_cross_shard_blpop_consumes_exactly_one_element() {
+ let bin = common::find_moon_binary();
+ if !bin.exists() {
+ eprintln!("skipping: no moon binary; build with `cargo build --release`");
+ return;
+ }
+
+ let (mut child, port) = common::spawn_listening(|port| {
+ let tmp_dir = std::env::temp_dir().join(format!("moon-wake-once-{port}"));
+ let _ = std::fs::create_dir_all(&tmp_dir);
+ Command::new(&bin)
+ .args([
+ "--port",
+ &port.to_string(),
+ "--shards",
+ SHARDS,
+ "--admin-port",
+ "0",
+ "--appendonly",
+ "no",
+ "--disk-free-min-pct",
+ "0",
+ "--dir",
+ tmp_dir.to_str().unwrap(),
+ ])
+ .stdout(Stdio::null())
+ .stderr(Stdio::null())
+ .spawn()
+ .expect("spawn moon")
+ });
+ let tmp_dir = std::env::temp_dir().join(format!("moon-wake-once-{port}"));
+
+ let result = std::panic::catch_unwind(|| {
+ for i in 0..KEYS {
+ let key = format!("wake:{i}");
+ let mut blocker = connect(port);
+ let mut pusher = connect(port);
+
+ // Park a real waiter, then push two elements from another
+ // connection: the waiter takes one, one stays.
+ send(&mut blocker, &["BLPOP", &key, "5"]);
+ std::thread::sleep(Duration::from_millis(150));
+
+ send(&mut pusher, &["RPUSH", &key, "first", "second"]);
+ let reply = read_reply(&mut pusher);
+ assert!(
+ reply.starts_with(":"),
+ "{key}: RPUSH should report a length, got {reply:?}"
+ );
+
+ let woken = read_reply(&mut blocker);
+ assert!(
+ woken.contains("first"),
+ "{key}: waiter should receive the head element, got {woken:?}"
+ );
+
+ send(&mut pusher, &["LLEN", &key]);
+ let reply = read_reply(&mut pusher);
+ assert!(
+ reply.starts_with(":1"),
+ "{key}: exactly one element should remain, got {reply:?}"
+ );
+ }
+ });
+
+ common::sigkill(&mut child);
+ let _ = std::fs::remove_dir_all(&tmp_dir);
+ if let Err(panic) = result {
+ std::panic::resume_unwind(panic);
+ }
+}