diff --git a/CHANGELOG.md b/CHANGELOG.md
index da9c1d70a..de170025d 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -6,6 +6,26 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
## [Unreleased]
+### Fixed
+- **Cross-shard fan-out no longer drops messages silently on a full SPSC ring
+ (c10k E1/E3), and cross-shard reply awaits are bounded (E4).** PUBLISH
+ fan-out (immediate, batched, and EXEC-queued) and SCRIPT LOAD propagation
+ used a single `try_push` — a transiently-full ring lost the message with no
+ log or metric: subscribers on that shard silently missed the publish, or its
+ script cache diverged (NOSCRIPT for a sha the server had just returned).
+ All five sites now retry with the same bounded, shutdown-aware backpressure
+ as the command dispatch path, and a final give-up is loud
+ (`moon_xshard_fanout_drop_total`). Reply awaits on the slotted dispatch and
+ publish paths — previously unbounded, so one wedged shard could hang a
+ client task forever — now share the coordinator's 30 s bound: publish counts
+ degrade (under-report, counted by `moon_xshard_reply_timeout_total`), while
+ a slotted-dispatch timeout errors the batch and closes the connection,
+ because the per-connection reply slot cannot be safely reused after an
+ abandoned await. Rehydrated connection handlers (park wake / migration)
+ also start with 512 B I/O buffers instead of 3×8 KiB, shrinking the memory
+ spike of fleet-synchronized wakes; buffers grow back on first real traffic
+ (c10k D3).
+
### Added
- **`INFO memory` can now explain a `used_memory`-vs-RSS gap.** Build with
`--features jemalloc-stats` and `INFO memory` reports Redis's `allocator_*`
diff --git a/src/admin/metrics_setup.rs b/src/admin/metrics_setup.rs
index a18a2122a..c82a4b54c 100644
--- a/src/admin/metrics_setup.rs
+++ b/src/admin/metrics_setup.rs
@@ -923,6 +923,31 @@ pub fn record_dispatch_cross_spsc() {
counter!("moon_dispatch_path_total", "path" => "cross_spsc").increment(1);
}
+/// Cross-shard fan-out message dropped after bounded retry (c10k E1/E3):
+/// the target shard's SPSC ring stayed full through every backoff. `kind` is
+/// `"publish"` (that shard's subscribers miss the message; PUBLISH count
+/// under-reports) or `"script_load"` (that shard's script cache diverges —
+/// EVALSHA there answers NOSCRIPT until the next SCRIPT LOAD).
+#[inline]
+pub fn record_xshard_fanout_drop(kind: &'static str) {
+ if !METRICS_INITIALIZED.load(Ordering::Relaxed) {
+ return;
+ }
+ counter!("moon_xshard_fanout_drop_total", "kind" => kind).increment(1);
+}
+
+/// Cross-shard reply await expired (c10k E4): the owner shard did not fill
+/// the reply slot within `XSHARD_REPLY_TIMEOUT`. `kind` `"dispatch"` is
+/// fatal for the connection (the reusable slot may be filled late);
+/// `"publish"` degrades to an under-reported subscriber count.
+#[inline]
+pub fn record_xshard_reply_timeout(kind: &'static str) {
+ if !METRICS_INITIALIZED.load(Ordering::Relaxed) {
+ return;
+ }
+ counter!("moon_xshard_reply_timeout_total", "kind" => kind).increment(1);
+}
+
/// Batched variant of `record_dispatch_cross_spsc`.
#[inline]
pub fn record_dispatch_cross_spsc_batch(count: u64) {
diff --git a/src/server/conn/handler_monoio/dispatch.rs b/src/server/conn/handler_monoio/dispatch.rs
index 7c3c2e097..ba43c1c81 100644
--- a/src/server/conn/handler_monoio/dispatch.rs
+++ b/src/server/conn/handler_monoio/dispatch.rs
@@ -8,7 +8,6 @@
//! Each helper returns `true` if the command was consumed (caller should `continue`).
use bytes::Bytes;
-use ringbuf::traits::Producer;
use std::cell::RefCell;
use std::rc::Rc;
use std::sync::Arc;
@@ -20,8 +19,6 @@ use crate::runtime::cancel::CancellationToken;
use crate::runtime::channel;
use crate::server::conn::core::{ConnectionContext, ConnectionState};
use crate::server::conn::util::extract_bytes;
-use crate::shard::dispatch::ShardMessage;
-use crate::shard::mesh::ChannelMesh;
use crate::tracking::TrackingState;
use crate::workspace::strip_workspace_prefix_from_response;
@@ -218,12 +215,14 @@ pub(super) fn try_handle_eval(
/// Handle SCRIPT subcommands (LOAD, EXISTS, FLUSH). Returns `true` if consumed.
///
-/// `#[inline]`: see `try_handle_cluster` rationale.
-#[inline]
-pub(super) fn try_handle_script(
+/// Async since E3: the SCRIPT LOAD fan-out retries a full ring with bounded
+/// backpressure instead of silently dropping the load (divergent per-shard
+/// script caches). Cold path — SCRIPT is never hot.
+pub(super) async fn try_handle_script(
cmd: &[u8],
cmd_args: &[Frame],
ctx: &ConnectionContext,
+ shutdown: &crate::runtime::cancel::CancellationToken,
responses: &mut Vec,
) -> bool {
if !cmd.eq_ignore_ascii_case(b"SCRIPT") {
@@ -232,21 +231,8 @@ pub(super) fn try_handle_script(
let (response, fanout) =
crate::scripting::handle_script_subcommand(&ctx.script_cache, cmd_args);
if let Some((sha1, script_bytes)) = fanout {
- let mut producers = ctx.dispatch_tx.borrow_mut();
- for target in 0..ctx.num_shards {
- if target == ctx.shard_id {
- continue;
- }
- let idx = ChannelMesh::target_index(ctx.shard_id, target);
- let msg = ShardMessage::ScriptLoad {
- sha1: sha1.clone(),
- script: script_bytes.clone(),
- };
- if producers[idx].try_push(msg).is_ok() {
- ctx.spsc_notifiers[target].notify_one();
- }
- }
- drop(producers);
+ crate::server::conn::shared::script_fanout_bounded(ctx, shutdown, &sha1, &script_bytes)
+ .await;
}
responses.push(response);
true
diff --git a/src/server/conn/handler_monoio/mod.rs b/src/server/conn/handler_monoio/mod.rs
index f2992d528..c5b99e376 100644
--- a/src/server/conn/handler_monoio/mod.rs
+++ b/src/server/conn/handler_monoio/mod.rs
@@ -333,14 +333,23 @@ pub(crate) async fn handle_connection_sharded_monoio<
// NOTE: do NOT call record_connection_opened() here — the caller
// (conn_accept.rs) already increments via try_accept_connection().
+ // c10k D3 (lazy resumed-buffer sizing): a fresh connection warms up with
+ // the full 8 KiB per buffer (no growth stall on its first pipeline). A
+ // REHYDRATED handler (park wake / migration) starts at 512 B: wakes
+ // arrive in fleet-sized bursts (synchronized keepalives), and 2×8 KiB
+ // per wake dominated the burst working set before the first byte was
+ // even parsed. BytesMut grows on demand — a busy resumed conn pays one
+ // amortized regrow; the idle majority never pays the 16 KiB at all.
+ let rehydrated = migrated_state.is_some();
+ let init_cap = if rehydrated { 512 } else { 8192 };
let mut read_buf = if initial_read_buf.is_empty() {
- BytesMut::with_capacity(8192)
+ BytesMut::with_capacity(init_cap)
} else {
let mut buf = initial_read_buf;
- buf.reserve(8192);
+ buf.reserve(init_cap);
buf
};
- let mut write_buf = BytesMut::with_capacity(8192);
+ let mut write_buf = BytesMut::with_capacity(init_cap);
// c10k A1: set when a blocking command leaves unparsed input in
// `read_buf` (see the read-skip guard in the main loop).
let mut carried_input = false;
@@ -401,7 +410,10 @@ pub(crate) async fn handle_connection_sharded_monoio<
// Pre-allocate read buffer outside the loop to avoid per-read heap allocation.
// Monoio's ownership I/O takes ownership and returns the buffer, so we reassign.
- let mut tmp_buf = vec![0u8; 8192];
+ // D3: same lazy sizing as read_buf/write_buf — the first read of a
+ // rehydrated conn is usually a probe-sized frame (keepalive PING); the
+ // shrink logic at the loop tail governs the steady state either way.
+ let mut tmp_buf = vec![0u8; init_cap];
// c10k W11: two-stage idle park (see idle_park.rs). Cancel-capable
// streams register for the shard chore's ≥1s sweep; `downshifted` tracks
@@ -980,6 +992,15 @@ pub(crate) async fn handle_connection_sharded_monoio<
}
}
+ // D3: a rehydrated conn starts with a small (512 B) owned read buffer;
+ // the moment a read saturates it (real traffic, not a keepalive-sized
+ // probe), restore the full 8 KiB so bulk transfers aren't capped at
+ // 512 B per syscall. Sited with the C2 check below: after every read
+ // arm, once per iteration.
+ if tmp_buf.len() < 8192 && read_buf.len() >= tmp_buf.len() {
+ tmp_buf = vec![0u8; 8192];
+ }
+
// c10k C2: query-buffer ceiling. One check per read iteration, sited
// after every read arm and ahead of both parse paths — an incomplete
// frame is exactly what makes `read_buf` grow, and an incomplete
@@ -1282,7 +1303,9 @@ pub(crate) async fn handle_connection_sharded_monoio<
{
continue;
}
- if cmd_len == 6 && dispatch::try_handle_script(cmd, cmd_args, ctx, &mut responses) {
+ if cmd_len == 6
+ && dispatch::try_handle_script(cmd, cmd_args, ctx, &shutdown, &mut responses).await
+ {
continue;
}
if dispatch::try_handle_cluster_routing(cmd, cmd_args, &mut conn, ctx, &mut responses) {
@@ -1548,7 +1571,10 @@ pub(crate) async fn handle_connection_sharded_monoio<
) {
Some(err) => err,
None => Frame::Integer(
- crate::server::conn::shared::publish_post_txn(ctx, &ch, &msg).await,
+ crate::server::conn::shared::publish_post_txn(
+ ctx, &shutdown, &ch, &msg,
+ )
+ .await,
),
};
if let Frame::Array(items) = &mut responses[exec_idx] {
@@ -2544,33 +2570,75 @@ pub(crate) async fn handle_connection_sharded_monoio<
std::sync::Arc,
Vec,
)> = Vec::new();
- {
- let mut producers = ctx.dispatch_tx.borrow_mut();
- for (target, entries) in publish_batches.drain() {
- let n = entries.len();
- let slot = std::sync::Arc::new(
- crate::shard::dispatch::PubSubResponseSlot::with_counts(1, n),
- );
- let resp_indices: Vec = entries.iter().map(|(idx, _, _)| *idx).collect();
- let pairs: Vec<(Bytes, Bytes)> =
- entries.into_iter().map(|(_, ch, msg)| (ch, msg)).collect();
-
- let idx = ChannelMesh::target_index(ctx.shard_id, target);
- let batch_msg = ShardMessage::PubSubPublishBatch {
- pairs,
- slot: slot.clone(),
- };
- if producers[idx].try_push(batch_msg).is_ok() {
+ for (target, entries) in publish_batches.drain() {
+ let n = entries.len();
+ let slot = std::sync::Arc::new(
+ crate::shard::dispatch::PubSubResponseSlot::with_counts(1, n),
+ );
+ let resp_indices: Vec = entries.iter().map(|(idx, _, _)| *idx).collect();
+ let pairs: Vec<(Bytes, Bytes)> =
+ entries.into_iter().map(|(_, ch, msg)| (ch, msg)).collect();
+
+ let idx = ChannelMesh::target_index(ctx.shard_id, target);
+ // E1: bounded backpressure retry instead of one bare try_push
+ // — a transiently-full ring no longer loses the batch. Borrow
+ // taken+released per attempt, never held across the backoff
+ // await (tokio parity — handler_sharded).
+ let mut pending = Some(ShardMessage::PubSubPublishBatch {
+ pairs,
+ slot: slot.clone(),
+ });
+ let outcome = crate::shard::dispatch::push_with_backpressure(
+ &shutdown,
+ crate::shard::dispatch::CROSS_SHARD_PUSH_MAX_RETRIES,
+ crate::shard::dispatch::CROSS_SHARD_PUSH_BACKOFF,
+ || match pending.take() {
+ None => true,
+ Some(m) => {
+ let mut producers = ctx.dispatch_tx.borrow_mut();
+ match producers[idx].try_push(m) {
+ Ok(()) => true,
+ Err(back) => {
+ pending = Some(back);
+ false
+ }
+ }
+ }
+ },
+ )
+ .await;
+ match outcome {
+ crate::shard::dispatch::PushOutcome::Pushed => {
ctx.spsc_notifiers[target].notify_one();
- } else {
- slot.add(0); // push failed, mark as done
}
- batch_slots.push((slot, resp_indices));
+ outcome => {
+ // Give-up: deliver-to-zero so the reply can't hang —
+ // but loudly (was a silent drop pre-E1).
+ tracing::warn!(
+ "Shard {}: PUBLISH batch fan-out to shard {target} dropped ({outcome:?})",
+ ctx.shard_id
+ );
+ crate::admin::metrics_setup::record_xshard_fanout_drop("publish");
+ slot.add(0);
+ }
}
+ batch_slots.push((slot, resp_indices));
}
- // Resolve all batch slots
+ // Resolve all batch slots (E4: bounded — a wedged shard degrades
+ // to an under-reported count, never a hung client).
for (slot, resp_indices) in &batch_slots {
- crate::shard::dispatch::PubSubResponseFuture::new(slot.clone()).await;
+ if !crate::shard::dispatch::await_pubsub_slot_bounded(
+ slot,
+ crate::shard::dispatch::XSHARD_REPLY_TIMEOUT,
+ )
+ .await
+ {
+ tracing::warn!(
+ "Shard {}: PUBLISH batch reply timed out awaiting remote shard",
+ ctx.shard_id
+ );
+ crate::admin::metrics_setup::record_xshard_reply_timeout("publish");
+ }
for (i, resp_idx) in resp_indices.iter().enumerate() {
let remote_count = slot.counts[i].load(std::sync::atomic::Ordering::Relaxed);
if remote_count > 0 {
@@ -2582,6 +2650,12 @@ pub(crate) async fn handle_connection_sharded_monoio<
}
}
+ // E4: set when a cross-shard reply await times out. The per-connection
+ // ResponseSlot is REUSED across batches — a late fill after a timeout
+ // would be read by the NEXT batch as its own reply — so a timeout is
+ // fatal: error the affected entries, flush, then close the connection.
+ let mut xshard_reply_fatal = false;
+
// Phase 2b: Dispatch all deferred remote commands as batched
// PipelineBatchSlotted messages (one per target shard), await all in parallel.
if !remote_groups.is_empty() {
@@ -2674,10 +2748,11 @@ pub(crate) async fn handle_connection_sharded_monoio<
// keeps the slot alive until the last handle drops. (This replaced the
// old raw-pointer-into-stack-pool design, whose contract required the
// await to run to completion to avoid a panic-unwind UAF; see the
- // `ResponseSlotPtr` doc + PR review.) The await is still unbounded here
- // for simplicity — the target shard always fills every message it
- // drains — but a shutdown-aware bound is now a safe, tracked follow-up.
- // The tokio handler carries the identical await.
+ // `ResponseSlotPtr` doc + PR review.) Since E4 the await is BOUNDED
+ // (XSHARD_REPLY_TIMEOUT) — expiry errors the batch and closes the
+ // connection, because the reused slot could otherwise hand a late
+ // fill to the next batch. The tokio handler carries the identical
+ // bounded await.
//
// C2 pipeline guard (see XSHARD_SPIN_MAX_BATCH_REMOTE): total cross-shard
// commands in THIS batch. The reply-side spin may engage only for a singleton
@@ -2705,9 +2780,33 @@ pub(crate) async fn handle_connection_sharded_monoio<
}
}
match spun {
- Some(r) => r,
- None => response_pool.future_for(target).await,
+ Some(r) => Some(r),
+ // E4: bounded await — a wedged owner shard can no
+ // longer hang this client task forever.
+ None => {
+ crate::shard::dispatch::await_response_slot_bounded(
+ response_pool.future_for(target),
+ crate::shard::dispatch::XSHARD_REPLY_TIMEOUT,
+ )
+ .await
+ }
+ }
+ };
+ let Some(shard_responses) = shard_responses else {
+ tracing::error!(
+ "Shard {}: cross-shard reply from shard {target} timed out; \
+ failing the batch and closing the connection (slot unsafe to reuse)",
+ ctx.shard_id
+ );
+ crate::admin::metrics_setup::record_xshard_reply_timeout("dispatch");
+ for (resp_idx, _, _, _) in &meta {
+ responses[*resp_idx] =
+ Frame::Error(Bytes::from_static(b"ERR cross-shard reply timeout"));
}
+ xshard_reply_fatal = true;
+ // Skip the fsync barrier too: with no reply there is
+ // nothing durable to confirm for these entries.
+ continue;
};
// H1-BARRIER: collect write resp_idxs before consuming meta
// so we can overwrite them if the fsync barrier fails.
@@ -2800,6 +2899,12 @@ pub(crate) async fn handle_connection_sharded_monoio<
}
}
+ // E4: a timed-out cross-shard reply slot must never be reused — the
+ // error replies are flushed above, now close.
+ if xshard_reply_fatal {
+ break;
+ }
+
// Update live state after each batch — lock-free (QW8, 2026-06
// review: this was a global registry write lock per batch), and
// clock-free (shard-cached ms, not Instant::now()).
diff --git a/src/server/conn/handler_sharded/mod.rs b/src/server/conn/handler_sharded/mod.rs
index 1877bb4ab..62e779719 100644
--- a/src/server/conn/handler_sharded/mod.rs
+++ b/src/server/conn/handler_sharded/mod.rs
@@ -785,16 +785,12 @@ pub(crate) async fn handle_connection_sharded_inner<
if cmd.eq_ignore_ascii_case(b"SCRIPT") {
let (response, fanout) = crate::scripting::handle_script_subcommand(&ctx.script_cache, cmd_args);
if let Some((sha1, script_bytes)) = fanout {
- let mut producers = ctx.dispatch_tx.borrow_mut();
- for target in 0..ctx.num_shards {
- if target == ctx.shard_id { continue; }
- let idx = ChannelMesh::target_index(ctx.shard_id, target);
- let msg = ShardMessage::ScriptLoad { sha1: sha1.clone(), script: script_bytes.clone() };
- if producers[idx].try_push(msg).is_ok() {
- ctx.spsc_notifiers[target].notify_one();
- }
- }
- drop(producers);
+ // E3: bounded fan-out — a full ring no longer
+ // silently diverges that shard's script cache.
+ crate::server::conn::shared::script_fanout_bounded(
+ ctx, &shutdown, &sha1, &script_bytes,
+ )
+ .await;
}
responses.push(response);
continue;
@@ -1074,7 +1070,7 @@ pub(crate) async fn handle_connection_sharded_inner<
) {
Some(err) => err,
None => Frame::Integer(
- crate::server::conn::shared::publish_post_txn(ctx, &ch, &msg).await,
+ crate::server::conn::shared::publish_post_txn(ctx, &shutdown, &ch, &msg).await,
),
};
if let Frame::Array(items) = &mut responses[exec_idx] {
@@ -2088,6 +2084,13 @@ pub(crate) async fn handle_connection_sharded_inner<
crate::admin::metrics_setup::record_dispatch_local_batch(local_dispatches as u64);
crate::admin::metrics_setup::record_dispatch_cross_spsc_batch(cross_spsc_dispatches as u64);
+ // E4: set when a cross-shard reply await times out. The
+ // per-connection ResponseSlot is REUSED across batches — a
+ // late fill after a timeout would be read by the NEXT batch
+ // as its own reply — so a timeout is fatal: error the
+ // affected entries, flush, then close the connection.
+ let mut xshard_reply_fatal = false;
+
// Phase 2: Dispatch deferred remote commands (zero-allocation via ResponseSlotPool)
if !remote_groups.is_empty() {
type RemoteMeta = (usize, Option, Bytes, Option);
@@ -2174,9 +2177,32 @@ pub(crate) async fn handle_connection_sharded_inner<
}
}
match spun {
- Some(r) => r,
- None => response_pool.future_for(target).await,
+ Some(r) => Some(r),
+ // E4: bounded await — a wedged owner shard
+ // can no longer hang this client task forever.
+ None => crate::shard::dispatch::await_response_slot_bounded(
+ response_pool.future_for(target),
+ crate::shard::dispatch::XSHARD_REPLY_TIMEOUT,
+ )
+ .await,
+ }
+ };
+ let Some(shard_responses) = shard_responses else {
+ tracing::error!(
+ "Shard {}: cross-shard reply from shard {target} timed out; \
+ failing the batch and closing the connection (slot unsafe to reuse)",
+ ctx.shard_id
+ );
+ crate::admin::metrics_setup::record_xshard_reply_timeout("dispatch");
+ for (resp_idx, _, _, _) in &meta {
+ responses[*resp_idx] = Frame::Error(Bytes::from_static(
+ b"ERR cross-shard reply timeout",
+ ));
}
+ xshard_reply_fatal = true;
+ // Skip the fsync barrier too: with no reply there
+ // is nothing durable to confirm for these entries.
+ continue;
};
// H1-BARRIER: collect (resp_idx, had_aof_bytes) pairs so
@@ -2248,30 +2274,72 @@ pub(crate) async fn handle_connection_sharded_inner<
// Phase 3: Flush accumulated PUBLISH batches as PubSubPublishBatch messages
if !publish_batches.is_empty() {
let mut batch_slots: Vec<(std::sync::Arc, Vec)> = Vec::new();
- {
- let mut producers = ctx.dispatch_tx.borrow_mut();
- for (target, entries) in publish_batches.drain() {
- let n = entries.len();
- let slot = std::sync::Arc::new(crate::shard::dispatch::PubSubResponseSlot::with_counts(1, n));
- let resp_indices: Vec = entries.iter().map(|(idx, _, _)| *idx).collect();
- let pairs: Vec<(Bytes, Bytes)> = entries.into_iter().map(|(_, ch, msg)| (ch, msg)).collect();
-
- let idx = ChannelMesh::target_index(ctx.shard_id, target);
- let batch_msg = ShardMessage::PubSubPublishBatch {
- pairs,
- slot: slot.clone(),
- };
- if producers[idx].try_push(batch_msg).is_ok() {
+ for (target, entries) in publish_batches.drain() {
+ let n = entries.len();
+ let slot = std::sync::Arc::new(crate::shard::dispatch::PubSubResponseSlot::with_counts(1, n));
+ let resp_indices: Vec = entries.iter().map(|(idx, _, _)| *idx).collect();
+ let pairs: Vec<(Bytes, Bytes)> = entries.into_iter().map(|(_, ch, msg)| (ch, msg)).collect();
+
+ let idx = ChannelMesh::target_index(ctx.shard_id, target);
+ // E1: bounded backpressure retry instead of one bare
+ // try_push — a transiently-full ring no longer loses
+ // the batch. Borrow taken+released per attempt, never
+ // held across the backoff await.
+ let mut pending = Some(ShardMessage::PubSubPublishBatch {
+ pairs,
+ slot: slot.clone(),
+ });
+ let outcome = crate::shard::dispatch::push_with_backpressure(
+ &shutdown,
+ crate::shard::dispatch::CROSS_SHARD_PUSH_MAX_RETRIES,
+ crate::shard::dispatch::CROSS_SHARD_PUSH_BACKOFF,
+ || match pending.take() {
+ None => true,
+ Some(m) => {
+ let mut producers = ctx.dispatch_tx.borrow_mut();
+ match producers[idx].try_push(m) {
+ Ok(()) => true,
+ Err(back) => {
+ pending = Some(back);
+ false
+ }
+ }
+ }
+ },
+ )
+ .await;
+ match outcome {
+ crate::shard::dispatch::PushOutcome::Pushed => {
ctx.spsc_notifiers[target].notify_one();
- } else {
- slot.add(0); // push failed, mark as done
}
- batch_slots.push((slot, resp_indices));
+ outcome => {
+ // Give-up: deliver-to-zero so the reply can't
+ // hang — but loudly (was a silent drop pre-E1).
+ tracing::warn!(
+ "Shard {}: PUBLISH batch fan-out to shard {target} dropped ({outcome:?})",
+ ctx.shard_id
+ );
+ crate::admin::metrics_setup::record_xshard_fanout_drop("publish");
+ slot.add(0);
+ }
}
+ batch_slots.push((slot, resp_indices));
}
- // Resolve all batch slots
+ // Resolve all batch slots (E4: bounded — a wedged shard
+ // degrades to an under-reported count, never a hung client).
for (slot, resp_indices) in &batch_slots {
- crate::shard::dispatch::PubSubResponseFuture::new(slot.clone()).await;
+ if !crate::shard::dispatch::await_pubsub_slot_bounded(
+ slot,
+ crate::shard::dispatch::XSHARD_REPLY_TIMEOUT,
+ )
+ .await
+ {
+ tracing::warn!(
+ "Shard {}: PUBLISH batch reply timed out awaiting remote shard",
+ ctx.shard_id
+ );
+ crate::admin::metrics_setup::record_xshard_reply_timeout("publish");
+ }
for (i, resp_idx) in resp_indices.iter().enumerate() {
let remote_count = slot.counts[i].load(std::sync::atomic::Ordering::Relaxed);
if remote_count > 0 {
@@ -2308,6 +2376,12 @@ pub(crate) async fn handle_connection_sharded_inner<
return (HandlerResult::Done, None);
}
+ // E4: a timed-out cross-shard reply slot must never be reused
+ // — the error replies are flushed above, now close.
+ if xshard_reply_fatal {
+ return (HandlerResult::Done, None);
+ }
+
// Update live state after each batch — lock-free (QW8, 2026-06
// review: this was a global registry write lock per batch), and
// clock-free (shard-cached ms, not Instant::now()).
diff --git a/src/server/conn/shared.rs b/src/server/conn/shared.rs
index 85e00a588..5d5547c6c 100644
--- a/src/server/conn/shared.rs
+++ b/src/server/conn/shared.rs
@@ -584,6 +584,7 @@ pub(crate) fn pubsub_command_acl_deny(
/// before the PUBLISH has been applied.
pub(crate) async fn publish_post_txn(
ctx: &super::core::ConnectionContext,
+ shutdown: &crate::runtime::cancel::CancellationToken,
channel: &Bytes,
message: &Bytes,
) -> i64 {
@@ -605,30 +606,132 @@ pub(crate) async fn publish_post_txn(
let slot = Arc::new(crate::shard::dispatch::PubSubResponseSlot::new(
remote_targets.len() as u32,
));
- {
- let mut producers = ctx.dispatch_tx.borrow_mut();
- for target in &remote_targets {
- let msg = crate::shard::dispatch::ShardMessage::PubSubPublish(Box::new(
- crate::shard::dispatch::PubSubPublishPayload {
- channel: channel.clone(),
- message: message.clone(),
- slot: slot.clone(),
- },
- ));
- let idx = ChannelMesh::target_index(ctx.shard_id, *target);
- if producers[idx].try_push(msg).is_ok() {
+ for target in &remote_targets {
+ // E1: bounded backpressure retry instead of one bare `try_push` — a
+ // transiently-full ring no longer loses the message. The borrow of
+ // `dispatch_tx` is taken+released inside each attempt, never held
+ // across the backoff await.
+ let mut pending = Some(crate::shard::dispatch::ShardMessage::PubSubPublish(
+ Box::new(crate::shard::dispatch::PubSubPublishPayload {
+ channel: channel.clone(),
+ message: message.clone(),
+ slot: slot.clone(),
+ }),
+ ));
+ let idx = ChannelMesh::target_index(ctx.shard_id, *target);
+ let outcome = crate::shard::dispatch::push_with_backpressure(
+ shutdown,
+ crate::shard::dispatch::CROSS_SHARD_PUSH_MAX_RETRIES,
+ crate::shard::dispatch::CROSS_SHARD_PUSH_BACKOFF,
+ || match pending.take() {
+ None => true,
+ Some(m) => {
+ let mut producers = ctx.dispatch_tx.borrow_mut();
+ match producers[idx].try_push(m) {
+ Ok(()) => true,
+ Err(back) => {
+ pending = Some(back);
+ false
+ }
+ }
+ }
+ },
+ )
+ .await;
+ match outcome {
+ crate::shard::dispatch::PushOutcome::Pushed => {
ctx.spsc_notifiers[*target].notify_one();
- } else {
- // Ring full: count this target as delivered-to-zero rather
- // than hanging the EXEC reply (mirrors the batch-flush path).
+ }
+ outcome => {
+ // Give-up: count the target as delivered-to-zero so the EXEC
+ // reply can't hang — but LOUDLY: this is real message loss to
+ // that shard's subscribers (was a silent drop pre-E1).
+ tracing::warn!(
+ "shard {}: EXEC PUBLISH fan-out to shard {target} dropped ({outcome:?})",
+ ctx.shard_id
+ );
+ crate::admin::metrics_setup::record_xshard_fanout_drop("publish");
slot.add(0);
}
}
}
- crate::shard::dispatch::PubSubResponseFuture::new(slot.clone()).await;
+ // E4: bounded await — a wedged target shard can't hang the EXEC reply
+ // forever. The slot is per-call, so abandoning it on expiry is safe; the
+ // count degrades to whatever responded in time (under-report, loud).
+ if !crate::shard::dispatch::await_pubsub_slot_bounded(
+ &slot,
+ crate::shard::dispatch::XSHARD_REPLY_TIMEOUT,
+ )
+ .await
+ {
+ tracing::warn!(
+ "shard {}: EXEC PUBLISH reply timed out awaiting remote shards",
+ ctx.shard_id
+ );
+ crate::admin::metrics_setup::record_xshard_reply_timeout("publish");
+ }
local_count + slot.get()
}
+/// Fan one `SCRIPT LOAD` out to every other shard with bounded backpressure
+/// (E3). A full ring used to drop the load SILENTLY, leaving that shard's
+/// script cache divergent: EVALSHA there answered NOSCRIPT for a sha this
+/// server had just returned. On give-up the drop is loud (warn + counter);
+/// the client still gets the sha — the script IS loaded locally, and client
+/// libraries' NOSCRIPT→EVAL fallback covers the divergent-shard window.
+pub(crate) async fn script_fanout_bounded(
+ ctx: &super::core::ConnectionContext,
+ shutdown: &crate::runtime::cancel::CancellationToken,
+ sha1: &str,
+ script: &Bytes,
+) {
+ use crate::shard::mesh::ChannelMesh;
+ use ringbuf::traits::Producer;
+
+ for target in 0..ctx.num_shards {
+ if target == ctx.shard_id {
+ continue;
+ }
+ let idx = ChannelMesh::target_index(ctx.shard_id, target);
+ let mut pending = Some(crate::shard::dispatch::ShardMessage::ScriptLoad {
+ sha1: sha1.to_owned(),
+ script: script.clone(),
+ });
+ let outcome = crate::shard::dispatch::push_with_backpressure(
+ shutdown,
+ crate::shard::dispatch::CROSS_SHARD_PUSH_MAX_RETRIES,
+ crate::shard::dispatch::CROSS_SHARD_PUSH_BACKOFF,
+ || match pending.take() {
+ None => true,
+ Some(m) => {
+ let mut producers = ctx.dispatch_tx.borrow_mut();
+ match producers[idx].try_push(m) {
+ Ok(()) => true,
+ Err(back) => {
+ pending = Some(back);
+ false
+ }
+ }
+ }
+ },
+ )
+ .await;
+ match outcome {
+ crate::shard::dispatch::PushOutcome::Pushed => {
+ ctx.spsc_notifiers[target].notify_one();
+ }
+ outcome => {
+ tracing::warn!(
+ "shard {}: SCRIPT LOAD fan-out to shard {target} dropped ({outcome:?}); \
+ that shard's script cache is divergent until the next load",
+ ctx.shard_id
+ );
+ crate::admin::metrics_setup::record_xshard_fanout_drop("script_load");
+ }
+ }
+ }
+}
+
/// Extract the primary key from a parsed command for shard routing.
///
/// Returns `None` for keyless commands (PING, DBSIZE, SELECT, etc.)
diff --git a/src/shard/coordinator.rs b/src/shard/coordinator.rs
index ea4b3abae..c48e37f5e 100644
--- a/src/shard/coordinator.rs
+++ b/src/shard/coordinator.rs
@@ -188,7 +188,9 @@ fn run_local(
/// any legitimate cross-shard command latency (including group-commit fsync
/// under load) — it exists only so a genuinely wedged shard surfaces an error
/// instead of an unbounded hang.
-const XSHARD_REPLY_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30);
+// One shared bound with the connection handlers' slot awaits (E4) — a
+// single constant so the two reply paths can't drift apart.
+use crate::shard::dispatch::XSHARD_REPLY_TIMEOUT;
/// Await a cross-shard `reply_rx` with a bounded timeout (#11).
///
diff --git a/src/shard/dispatch.rs b/src/shard/dispatch.rs
index 037143f67..f89f0f6a5 100644
--- a/src/shard/dispatch.rs
+++ b/src/shard/dispatch.rs
@@ -1097,6 +1097,57 @@ pub(crate) async fn push_with_backpressure(
PushOutcome::Backpressure
}
+/// One shared bound for every cross-shard reply await on the connection
+/// handlers (E4 — design-for-failure), mirroring the coordinator's
+/// [`crate::shard::coordinator::recv_reply_bounded`]: a wedged owner shard
+/// must never hang a client task forever. 30s is far beyond any legitimate
+/// drain latency, so expiry means "the owner is not coming back".
+pub(crate) const XSHARD_REPLY_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30);
+
+/// Await a [`PubSubResponseSlot`] with a bounded timeout (E4).
+///
+/// Returns `true` when every pending shard responded within `timeout`,
+/// `false` on expiry. The slot is per-call (freshly allocated by every
+/// publish fan-out), so abandoning it on timeout is SAFE: a late `add()`
+/// lands on the Arc'd slot and is simply never read — unlike the reusable
+/// per-connection `ResponseSlot` (see [`await_response_slot_bounded`]).
+pub(crate) async fn await_pubsub_slot_bounded(
+ slot: &std::sync::Arc,
+ timeout: std::time::Duration,
+) -> bool {
+ use crate::runtime::race::{Arm, race2};
+ use crate::runtime::{TimerImpl, traits::RuntimeTimer};
+ let fut = std::pin::pin!(PubSubResponseFuture::new(slot.clone()));
+ let sleep = std::pin::pin!(TimerImpl::sleep(timeout));
+ // race2 polls the first arm first: a ready slot always wins the tie.
+ matches!(race2(fut, sleep).await, Arm::First(_))
+}
+
+/// Await a [`crate::server::response_slot::ResponseSlotFuture`] with a
+/// bounded timeout (E4). `None` means the owner shard never filled the slot
+/// within `timeout`.
+///
+/// # Caller contract — `None` is FATAL for the connection
+///
+/// Unlike the per-call pubsub slot, a connection's `ResponseSlot` is REUSED
+/// across batches. After a timeout the wedged owner may still fill the slot
+/// late, and the next batch to that shard would read the stale reply as its
+/// own. Callers MUST error the affected entries and close the connection —
+/// never continue dispatching to the timed-out slot.
+pub(crate) async fn await_response_slot_bounded(
+ fut: crate::server::response_slot::ResponseSlotFuture,
+ timeout: std::time::Duration,
+) -> Option> {
+ use crate::runtime::race::{Arm, race2};
+ use crate::runtime::{TimerImpl, traits::RuntimeTimer};
+ let fut = std::pin::pin!(fut);
+ let sleep = std::pin::pin!(TimerImpl::sleep(timeout));
+ match race2(fut, sleep).await {
+ Arm::First(frames) => Some(frames),
+ Arm::Second(()) => None,
+ }
+}
+
#[cfg(test)]
mod tests {
use super::*;
@@ -1506,4 +1557,65 @@ mod tests {
"FtHybridPayload size sanity: {payload_sz}",
);
}
+
+ // ── E4: bounded cross-shard reply awaits ──
+ // Tokio-gated like the F3 suite: these drive the runtime timer.
+
+ #[cfg(feature = "runtime-tokio")]
+ #[tokio::test]
+ async fn pubsub_slot_bounded_completes_when_all_shards_respond() {
+ let slot = Arc::new(PubSubResponseSlot::new(1));
+ slot.add(3);
+ assert!(
+ await_pubsub_slot_bounded(&slot, std::time::Duration::from_secs(5)).await,
+ "a fully-responded slot must complete, not time out"
+ );
+ assert_eq!(slot.get(), 3);
+ }
+
+ #[cfg(feature = "runtime-tokio")]
+ #[tokio::test]
+ async fn pubsub_slot_bounded_times_out_on_a_wedged_shard() {
+ // 2 pending, only 1 ever responds — pre-E4 this await hung forever.
+ let slot = Arc::new(PubSubResponseSlot::new(2));
+ slot.add(1);
+ let start = std::time::Instant::now();
+ assert!(
+ !await_pubsub_slot_bounded(&slot, std::time::Duration::from_millis(50)).await,
+ "a slot with a missing shard response must time out"
+ );
+ assert!(
+ start.elapsed() < std::time::Duration::from_secs(5),
+ "timeout must be bounded by the requested duration, took {:?}",
+ start.elapsed()
+ );
+ }
+
+ #[cfg(feature = "runtime-tokio")]
+ #[tokio::test]
+ async fn response_slot_bounded_returns_frames_when_filled() {
+ let pool = crate::server::response_slot::ResponseSlotPool::new(2, 0);
+ pool.slot_arc(1).fill(vec![Frame::Integer(7)]);
+ let frames =
+ await_response_slot_bounded(pool.future_for(1), std::time::Duration::from_secs(5))
+ .await
+ .expect("filled slot must resolve");
+ assert_eq!(frames.len(), 1);
+ }
+
+ #[cfg(feature = "runtime-tokio")]
+ #[tokio::test]
+ async fn response_slot_bounded_times_out_on_a_wedged_owner() {
+ let pool = crate::server::response_slot::ResponseSlotPool::new(2, 0);
+ let start = std::time::Instant::now();
+ let got =
+ await_response_slot_bounded(pool.future_for(1), std::time::Duration::from_millis(50))
+ .await;
+ assert!(got.is_none(), "an unfilled slot must time out with None");
+ assert!(
+ start.elapsed() < std::time::Duration::from_secs(5),
+ "timeout must be bounded, took {:?}",
+ start.elapsed()
+ );
+ }
}
diff --git a/tests/pubsub_kv_ordering.rs b/tests/pubsub_kv_ordering.rs
index 5e0263125..d048c26e5 100644
--- a/tests/pubsub_kv_ordering.rs
+++ b/tests/pubsub_kv_ordering.rs
@@ -61,7 +61,12 @@ impl Drop for Moon {
fn spawn_moon(shards: &str) -> Option {
let bin = moon_binary()?;
- let tmp_dir = std::env::temp_dir().join(format!("moon-pubsub-ord-{}", std::process::id()));
+ // `{shards}` in the dir name: the two tests in this file run CONCURRENTLY
+ // under the default cargo-test threading, and the per-dir instance lock
+ // (moon.lock) makes whichever server starts second exit at boot when they
+ // share a --dir (same convention as tests/acl_privileged_intercepts.rs).
+ let tmp_dir =
+ std::env::temp_dir().join(format!("moon-pubsub-ord-{}-{shards}", std::process::id()));
let _ = std::fs::create_dir_all(&tmp_dir);
let (child, port) = common::spawn_listening(|port| {
Command::new(&bin)
diff --git a/tests/pubsub_multi_channel_acl.rs b/tests/pubsub_multi_channel_acl.rs
index 830053823..4e9a8c608 100644
--- a/tests/pubsub_multi_channel_acl.rs
+++ b/tests/pubsub_multi_channel_acl.rs
@@ -58,7 +58,12 @@ impl Drop for Moon {
fn spawn_moon(shards: &str) -> Option {
let bin = moon_binary()?;
- let tmp_dir = std::env::temp_dir().join(format!("moon-pubsub-acl-{}", std::process::id()));
+ // `{shards}` in the dir name: the tests in this file run CONCURRENTLY
+ // under the default cargo-test threading, and the per-dir instance lock
+ // (moon.lock) makes whichever server starts second exit at boot when they
+ // share a --dir (same convention as tests/acl_privileged_intercepts.rs).
+ let tmp_dir =
+ std::env::temp_dir().join(format!("moon-pubsub-acl-{}-{shards}", std::process::id()));
let _ = std::fs::create_dir_all(&tmp_dir);
let (child, port) = common::spawn_listening(|port| {
Command::new(&bin)