fix(shard): bounded cross-shard fan-out + reply awaits (c10k E1/E3/E4), lazy rehydrated buffers (D3) - #441
Merged
Merged
Conversation
…), lazy rehydrated buffers (D3)
Cross-shard PUBLISH fan-out (EXEC-queued via publish_post_txn, and the
batched flush in both connection handlers) and SCRIPT LOAD propagation
each pushed with a single bare try_push: a transiently-full SPSC ring
lost the message with no log and no metric. Subscribers on that shard
silently missed the publish (count under-reported), or its script cache
diverged — EVALSHA answering NOSCRIPT for a sha this server had just
returned. All five sites now retry through push_with_backpressure (the
same bounded, shutdown-aware helper the slotted dispatch path already
uses; ring borrow taken per attempt, never held across the backoff
await), and a final give-up is loud: tracing::warn plus the new
moon_xshard_fanout_drop_total counter (E1, E3). The monoio SCRIPT
intercept becomes async to reach the shared bounded fan-out helper.
Reply awaits get the coordinator's 30s bound (E4), via two new helpers
in shard::dispatch sharing coordinator.rs's XSHARD_REPLY_TIMEOUT:
- await_pubsub_slot_bounded: publish slots are per-call Arcs, so expiry
simply degrades the reply to whatever responded in time (under-count,
moon_xshard_reply_timeout_total).
- await_response_slot_bounded: the slotted-dispatch ResponseSlot is
REUSED across batches — a late fill after an abandoned await would be
read by the NEXT batch as its own reply — so expiry is fatal: the
batch's entries error ("ERR cross-shard reply timeout"), the replies
flush, and the connection closes. Both handlers carry the identical
logic; the fsync barrier is skipped for a timed-out target (no reply,
nothing durable to confirm).
D3 (lazy rehydrated-buffer sizing): a handler spawned for a park wake or
migration now starts with 512 B read/write/tmp buffers instead of
3×8 KiB — fleet-synchronized wakes (NAT keepalives) rehydrate N handlers
at once, and the fixed 24 KiB per wake dominated the burst working set.
Buffers grow back on first real traffic: BytesMut on demand, and the
owned tmp read buffer via a saturation check sited with the C2
query-ceiling check (a 512 B read that fills the buffer restores the
full 8 KiB, so bulk transfers are never capped at 512 B per syscall).
Fresh connections are unchanged.
Red/green: the four E4 helper tests (wedged-shard timeout + happy path
for both slot kinds) failed to resolve pre-implementation and pass now
under runtime-tokio. Verified locally: clippy clean on both runtimes,
release build green, and the affected integration suites (pubsub x4,
sharded multi/exec x3, parked-idle parity) all pass with a fresh binary.
Also fixes a pre-existing dir-lock flake unmasked while validating:
pubsub_kv_ordering and pubsub_multi_channel_acl keyed their --dir on
process id alone, so their tests raced the per-dir instance lock
(moon.lock) under default cargo-test threading and the second server
exited at boot. Dir names now include the shards tag (same convention
as acl_privileged_intercepts). Reproduced on unmodified main.
refs: tmp/C10K-HARDENING-REVIEW.md clusters D/E, #438
author: Tin Dang
📝 WalkthroughWalkthroughCross-shard publish and script fan-out now use shutdown-aware bounded retries. Reply waits have a shared 30-second timeout. Timeout failures update responses, metrics, batch state, and connection lifecycle. Rehydrated connections use smaller initial buffers. ChangesCross-shard reliability
Estimated code review effort: 4 (Complex) | ~45 minutes Possibly related issues
Possibly related PRs
Suggested reviewers: 🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
PR Summary by QodoFix cross-shard fan-out drops, bound reply awaits, and lazy buffer rehydration
AI Description
Diagram
High-Level Assessment
Files changed (10)
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
🧹 Nitpick comments (2)
src/server/conn/shared.rs (1)
696-718: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueConsider extracting the retain-and-retry push closure.
The same
pending/try_push/Err(back)closure now appears inpublish_post_txn,script_fanout_bounded, and both handler dispatch sites. A helper such aspush_shard_message_bounded(ctx, target, msg, shutdown) -> PushOutcomewould hold the closure once. Each call site keeps its own give-up arm. This is optional and can be deferred.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/server/conn/shared.rs` around lines 696 - 718, Optionally extract the repeated pending-message retry closure from publish_post_txn, script_fanout_bounded, and both handler dispatch sites into a shared push_shard_message_bounded helper that accepts the dispatch context, target, message, and shutdown signal and returns PushOutcome. Preserve each caller’s existing give-up handling and retry behavior.src/server/conn/handler_monoio/mod.rs (1)
995-1002: 🚀 Performance & Scalability | 🔵 Trivial | 💤 Low valuePrefer
resizeover a fresh allocation for the buffer upgrade.
vec![0u8; 8192]discards the existing allocation.tmp_buf.resize(8192, 0)reuses the current capacity when it is already large enough. The idle-park path at Lines 791-798 can shrinktmp_bufback toIDLE_PROBE_BUFon a later downshift, which re-arms this guard, so the upgrade can run more than once per connection. This is optional; the cost is one 8 KiB allocation per idle-wake cycle.♻️ Proposed change
if tmp_buf.len() < 8192 && read_buf.len() >= tmp_buf.len() { - tmp_buf = vec![0u8; 8192]; + tmp_buf.resize(8192, 0); }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/server/conn/handler_monoio/mod.rs` around lines 995 - 1002, Update the buffer upgrade logic around tmp_buf so it calls resize(8192, 0) instead of replacing the vector with vec![0u8; 8192], preserving the existing allocation when capacity permits while retaining the current guard and behavior.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@src/server/conn/handler_sharded/mod.rs`:
- Around line 2379-2384: Replace the direct return in the xshard_reply_fatal
branch of handle_connection_sharded_inner with a loop break carrying the same
HandlerResult::Done and None values, so control reaches the existing post-loop
cleanup including abort_cross_store_txn_routed, unpropagate_subscription, and
untrack_all. Align the behavior with the monoio handler while preserving the
fatal timeout result.
In `@src/server/conn/shared.rs`:
- Around line 691-718: Update script_fanout_bounded’s sequential cross-shard
dispatch to enforce an overall latency budget for SCRIPT LOAD fan-out across all
non-local shards, rather than allowing each push_with_backpressure retry budget
to multiply by num_shards - 1. Add a small supported-shard-count or per-leg
budget guard so the total admin-reply wait remains within the accepted latency
limit, while preserving successful delivery and shutdown handling.
---
Nitpick comments:
In `@src/server/conn/handler_monoio/mod.rs`:
- Around line 995-1002: Update the buffer upgrade logic around tmp_buf so it
calls resize(8192, 0) instead of replacing the vector with vec![0u8; 8192],
preserving the existing allocation when capacity permits while retaining the
current guard and behavior.
In `@src/server/conn/shared.rs`:
- Around line 696-718: Optionally extract the repeated pending-message retry
closure from publish_post_txn, script_fanout_bounded, and both handler dispatch
sites into a shared push_shard_message_bounded helper that accepts the dispatch
context, target, message, and shutdown signal and returns PushOutcome. Preserve
each caller’s existing give-up handling and retry behavior.
🪄 Autofix
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: 2c9d46db-c0e9-47d5-88ad-e30333f16e3c
📒 Files selected for processing (10)
CHANGELOG.mdsrc/admin/metrics_setup.rssrc/server/conn/handler_monoio/dispatch.rssrc/server/conn/handler_monoio/mod.rssrc/server/conn/handler_sharded/mod.rssrc/server/conn/shared.rssrc/shard/coordinator.rssrc/shard/dispatch.rstests/pubsub_kv_ordering.rstests/pubsub_multi_channel_acl.rs
Comment on lines
+2379
to
+2384
| // 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); | ||
| } | ||
|
|
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Use break so the disconnect cleanup runs after a cross-shard reply timeout.
This arm returns directly from handle_connection_sharded_inner. The cleanup block after the outer loop is therefore skipped:
abort_cross_store_txn_routed(Line 2472) does not run. Recordedkv_write_intentsstay in place, which keeps the affected keys invisible to every later reader on that shard.unpropagate_subscription(Lines 2489-2517) does not run. Other shards keep stale remote-subscriber entries for this connection's channels and patterns.untrack_all(Line 2525) does not run. The CLIENT TRACKING registration stays live.
A cross-shard read is still dispatched while a cross-store TXN is open (only cross-shard writes are rejected at Line 2024), so the TXN-leak path is reachable. The monoio twin uses break at src/server/conn/handler_monoio/mod.rs Line 2905 and reaches its equivalent cleanup. Align this handler with it.
🐛 Proposed fix
// 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);
+ break;
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| // 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); | |
| } | |
| // 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; | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/server/conn/handler_sharded/mod.rs` around lines 2379 - 2384, Replace the
direct return in the xshard_reply_fatal branch of
handle_connection_sharded_inner with a loop break carrying the same
HandlerResult::Done and None values, so control reaches the existing post-loop
cleanup including abort_cross_store_txn_routed, unpropagate_subscription, and
untrack_all. Align the behavior with the monoio handler while preserving the
fatal timeout result.
Comment on lines
+691
to
+718
| 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; |
There was a problem hiding this comment.
🚀 Performance & Scalability | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Description: Read the cross-shard push retry budget constants and their uses.
set -euo pipefail
rg -n -C 4 'CROSS_SHARD_PUSH_MAX_RETRIES|CROSS_SHARD_PUSH_BACKOFF' --type=rustRepository: pilotspace/moon
Length of output: 153
🏁 Script executed:
#!/bin/bash
# Locate the referenced file and inspect nearby code plus cross-shard retry symbols without failing on no matches.
set -euo pipefail
echo "== git files matching shared.rs =="
git ls-files | rg '(^|/)shared\.rs$|src/server/conn/shared.rs'
if [ -f src/server/conn/shared.rs ]; then
echo "== outline around script_fanout_bounded =="
ast-grep outline src/server/conn/shared.rs --view expanded | rg -n "script_fanout_bounded|script" || true
echo "== lines 650-760 =="
sed -n '650,760p' src/server/conn/shared.rs | nl -ba -v650
fi
echo "== symbol searches =="
rg -n -C 3 'push_with_backpressure|CROSS_SHARD|TARGET_INDEX|target_index' --type=rust . || trueRepository: pilotspace/moon
Length of output: 370
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== lines 650-760 of shared.rs =="
sed -n '650,760p' src/server/conn/shared.rs
echo "== symbol searches =="
rg -n -C 4 'push_with_backpressure|CROSS_SHARD|target_index|ScriptLoad|script_fanout_bounded' --type=rust . || true
echo "== relevant file sizes =="
wc -l src/shard/dispatch.rs src/server/conn/shared.rs src/main.rsRepository: pilotspace/moon
Length of output: 50371
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== dispatch config constants =="
rg -n -C 3 'XSHARD_REPLY_TIMEOUT|XSHARD|num_shards|NUM_SHARDS|max.*shard|shard_count' --type=rust . | rg -n -C 3 'XSHARD_REPLY_TIMEOUT|XSHARD|num_shards|NUM_SHARDS|max.*shard|shard_count' || true
echo "== max shard references =="
rg -n -C 3 'num_shards|num_shards|num_shards' --type=rust .
echo "== push helper implementation =="
sed -n '1020,1115pH ' src/shard/dispatch.rs
echo "== publish serial fan-out implementation =="
sed -n '580,650p' src/server/conn/shared.rs
echo "== blocking comment for retry budget =="
sed -n '168,190p' src/server/conn/blocking.rsRepository: pilotspace/moon
Length of output: 50372
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== targeted config refs =="
rg -n 'pub\s+const\s+num_shards|num_shards\s*=|NUM_SHARDS|XSHARD_REPLY_TIMEOUT|MAX_SHARDS|max.*shard' --type=rust .
echo "== dispatch helper implementation =="
sed -n '1020,1115p' src/shard/dispatch.rs
echo "== blocking retry budget comment =="
sed -n '168,190p' src/server/conn/blocking.rs
echo "== publish fanout loop =="
sed -n '590,635p' src/server/conn/shared.rsRepository: pilotspace/moon
Length of output: 27788
Confirm SCRIPT LOAD’s serialized fan-out budget for the supported shard count range.
script_fanout_bounded awaits push_with_backpressure for each non-local shard in order, and CROSS_SHARD_PUSH_MAX_RETRIES * CROSS_SHARD_PUSH_BACKOFF is documented as roughly 0.5s up to a few seconds. With many wedged target rings, the reply can wait up to (num_shards - 1) * retry_budget; add a small max-shard or per-leg budget guard if that total exceeds an acceptable admin-reply latency.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/server/conn/shared.rs` around lines 691 - 718, Update
script_fanout_bounded’s sequential cross-shard dispatch to enforce an overall
latency budget for SCRIPT LOAD fan-out across all non-local shards, rather than
allowing each push_with_backpressure retry budget to multiply by num_shards - 1.
Add a small supported-shard-count or per-leg budget guard so the total
admin-reply wait remains within the accepted latency limit, while preserving
successful delivery and shutdown handling.
Code Review by Qodo
1. Boxed timer per await
|
![qodo-code-review[bot]](https://avatars.githubusercontent.com/in/484649?s=60&v=4){kind=small}
Comment on lines
+1119
to
+1122
| 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. |
There was a problem hiding this comment.
1. Boxed timer per await 🐞 Bug ➹ Performance
The new bounded-await helpers allocate a boxed timer future on every call because they use TimerImpl::sleep(timeout) (which is Pin<Box<dyn Future>>), adding allocator pressure to cross-shard reply awaits. This is avoidable since the coordinator already implements bounded waits using non-boxed runtime sleep futures.
Agent Prompt
### Issue description
`await_pubsub_slot_bounded` / `await_response_slot_bounded` unconditionally call `TimerImpl::sleep(timeout)`, and `TimerImpl::sleep` is implemented as `Box::pin(...)` for both tokio and monoio. This introduces a heap allocation per bounded await, which can add allocator pressure in cross-shard-heavy workloads.
### Issue Context
The coordinator’s `recv_reply_bounded` already implements the same race pattern using runtime-specific `tokio::time::sleep` / `monoio::time::sleep` without boxing.
### Fix Focus Areas
- src/shard/dispatch.rs[1114-1149]
- src/shard/coordinator.rs[201-215]
- src/runtime/traits.rs[12-22]
- src/runtime/tokio_impl.rs[21-30]
- src/runtime/monoio_impl.rs[22-31]
### Suggested fix
- Mirror `coordinator::recv_reply_bounded`:
- In `dispatch.rs`, create the sleep future via `#[cfg(feature = "runtime-tokio")] tokio::time::sleep(timeout)` and `#[cfg(feature = "runtime-monoio")] monoio::time::sleep(timeout)` (or add a small internal helper that returns an unboxed sleep future behind cfg).
- Keep `race2` polling order unchanged.
- Alternatively (larger change): evolve the `RuntimeTimer::sleep` trait to return an unboxed concrete future (or a generic associated type) to avoid boxing across the codebase.
ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools
Comment on lines
+1306
to
+1308
| if cmd_len == 6 | ||
| && dispatch::try_handle_script(cmd, cmd_args, ctx, &shutdown, &mut responses).await | ||
| { |
There was a problem hiding this comment.
2. Async script check on length 🐞 Bug ➹ Performance
The monoio handler now awaits try_handle_script for every 6-byte command, even when the command is not SCRIPT (the async function immediately returns false). This adds avoidable async-future construction/poll overhead to common 6-letter commands (e.g., CONFIG/SELECT/GETSET).
Agent Prompt
### Issue description
`handler_monoio/mod.rs` calls/awaits an async `try_handle_script` based only on `cmd_len == 6`. For all other 6-byte commands, `try_handle_script` immediately returns `false` after its internal `eq_ignore_ascii_case(b"SCRIPT")` check, so the await is unnecessary work on a hot path.
### Issue Context
`try_handle_script` became `async` to support bounded fan-out retries for `SCRIPT LOAD` (cold path), but the current call-site pattern makes the async wrapper run for unrelated 6-byte commands.
### Fix Focus Areas
- src/server/conn/handler_monoio/mod.rs[1294-1330]
- src/server/conn/handler_monoio/dispatch.rs[216-239]
### Suggested fix
- Change the call site to avoid invoking the async helper unless the command is actually SCRIPT, e.g.:
- `if cmd_len == 6 && cmd.eq_ignore_ascii_case(b"SCRIPT") { if dispatch::try_handle_script(...).await { continue; } }`
- Or split into:
- a sync predicate (`is_script(cmd) -> bool`), and
- an async handler only executed on true.
ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools
This was referenced Aug 13, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Continuation of the c10k hardening campaign (
tmp/C10K-HARDENING-REVIEW.mdclusters D/E, tracked in #438). Closes the remaining cross-shard drop-on-full and unbounded-await findings, plus D3's lazy-buffer lever.E1 — PUBLISH fan-out silently dropped on full ring (3 sites).
publish_post_txn(EXEC-queued) and the batched flush in both connection handlers each used one baretry_push; a transiently-full SPSC ring lost the message with no log or metric. All sites now retry viapush_with_backpressure(bounded, shutdown-aware, borrow-per-attempt); final give-up warns + increments the newmoon_xshard_fanout_drop_total.E3 — SCRIPT LOAD fan-out dropped (2 sites). Same fix via a shared
script_fanout_boundedhelper; the monoio SCRIPT intercept becomes async to reach it. Divergent per-shard script caches (NOSCRIPT for a just-returned sha) now require a genuinely wedged shard, and are loud.E4 — unbounded reply awaits (5 sites). Two new bounded-await helpers in
shard::dispatchshare the coordinator's 30 sXSHARD_REPLY_TIMEOUT(single const, no drift). Publish slots are per-call Arcs, so expiry degrades to an under-reported count (moon_xshard_reply_timeout_total). The slotted-dispatchResponseSlotis reused across batches — a late fill after an abandoned await would be read by the next batch as its own reply — so expiry there errors the batch (ERR cross-shard reply timeout), flushes, and closes the connection. Fsync barrier skipped for the timed-out target.D3 — lazy rehydrated-buffer sizing. Handlers spawned for a park wake or migration start at 512 B read/write/tmp buffers instead of 3×8 KiB (fleet-synchronized wakes rehydrate N handlers at once). A saturation check restores the full 8 KiB tmp buffer on first real traffic so bulk transfers are never capped at 512 B/syscall. Fresh connections unchanged. The kill-burst herd and park-thrash halves of D3 were already structurally mitigated by the merged campaign (kill-driven wakes close in the watcher; re-parks are sweep-driven at 1/threshold per conn) — documented as such in #438.
Drive-by: fixes a pre-existing dir-lock flake (reproduced on unmodified main):
pubsub_kv_orderingandpubsub_multi_channel_aclkeyed--diron pid alone, so their tests raced themoon.lockinstance lock under parallel cargo-test.Test plan
cargo test --release(monoio, CI-invisible path): 4476 lib + all integration suites green locallycargo clippyclean on both runtimes;cargo fmtappliedSummary by CodeRabbit
Bug Fixes
Performance