fix(server): connection-plane robustness hardening (Track B)

[pilotspacex-byte]

Connection-plane robustness hardening (Track B)

A self-review of Moon's connection/event-loop plane (part of a broader 5-dimension deep review — see tmp/CONN-DEEP-REVIEW.md) surfaced four robustness gaps that bite at high parallel connection counts (1K–10K clients). All four are fixed here, each with red/green tests. Both runtimes build; full lib suites pass (tokio 3115, monoio 3770); clippy/fmt clean.

R-2 — inline-command length cap

src/protocol/inline.rs, src/protocol/frame.rs

The RESP-less inline (telnet) path had no maximum line length, so a client that never sends CRLF (raw non-RESP bytes) grew the connection read buffer without bound — a per-connection memory-exhaustion vector. Added ParseConfig::max_inline_size (default 64 KB, mirrors Redis PROTO_INLINE_MAX_SIZE); parse_inline now rejects an over-cap line (complete or incomplete) with Protocol error: too big inline request.

R-1 — bounded cross-shard spsc_send

src/shard/coordinator.rs

The single dispatch primitive behind every scatter-gather command (MGET/MSET/DEL/EXISTS/SCAN/KEYS/DBSIZE, vector scatter, FT.INFO, graph traverse) retried a full SPSC ring in an unbounded loop { try_push; yield/sleep } — on tokio yield_now() reschedules with no backoff, busy-spinning a full core, and a wedged target could block graceful shutdown forever. Replaced with a bounded retry (CROSS_SHARD_PUSH_MAX_RETRIES × CROSS_SHARD_PUSH_BACKOFF, same budget as push_with_backpressure) returning PushOutcome. On give-up the message is dropped; reply-carrying callers observe the closed reply channel and synthesize a per-shard error via the path they already handle — no call-site signature change.

R-4 — accept-loop backoff on fd exhaustion

src/server/accept_backoff.rs (new), src/server/listener.rs, src/shard/event_loop.rs

All six socket-accept loops (tokio and monoio; sharded, non-sharded, TLS) retried accept() errors with zero backoff, so EMFILE/ENFILE under a connection storm pinned a shard core at 100% and flooded the log. Added AcceptBackoff: capped exponential backoff (1ms → 1s) on resource-exhaustion errors only, plus rate-limited logging; benign per-connection errors (ECONNABORTED) log without sleeping. The io_uring multishot-accept resubmit (opt-in MOON_URING=1 bridge) is documented as a scoped follow-up (synchronous CQE handler, no async context to sleep in).

R-3 — CLIENT KILL force-closes idle connections

src/client_registry.rs, connection handlers, src/shard/conn_accept.rs

CLIENT KILL only set a cooperative kill_flag checked once per batch, so a connection parked in read().await (idle, default timeout 0) was never torn down until it next sent bytes. The registry now stores each connection's socket fd and kill_clients also shutdown(2)s it, so the parked read returns Ok(0)/Err immediately (existing disconnect path). The raw-fd close is race-free: kill_clients holds the registry read lock, and a connection's RegistryGuard deregisters (needs the write lock) strictly before its stream drops — so a visible entry always has a live fd. No-op on non-unix.

---

Testing: each fix ships red/green unit tests (inline cap boundaries; bounded spsc_send give-up + success; AcceptBackoff classification/backoff curve; CLIENT KILL peer-EOF via UnixStream::pair). Full CI matrix run locally via OrbStack (fmt, clippy ×2, tokio + monoio lib suites).

Deep-review context (Tier-0 multi-tenant isolation and Tier-2 head-of-line-blocking tracks, not in this PR) is in tmp/CONN-DEEP-REVIEW.md.

Summary by CodeRabbit

• New Features

  • Added a configurable TCP listen backlog for server startup.
  • Added cross-shard flush behavior so FLUSHDB/FLUSHALL clears data consistently across shards.

• Bug Fixes

  • Improved CLIENT KILL so targeted connections close more reliably.
  • Made transaction commits replay to the correct database.
  • Hardened connection migration to avoid dropping clients when handoff fails.

• Performance & Reliability

  • Added backpressure handling and rate-limited accept-loop retry behavior to reduce hangs under load.

[qodo-code-review]

Qodo reviews are paused for this user.

Troubleshooting steps vary by plan Learn more →

On a Teams plan?
Reviews resume once this user has a paid seat and their Git account is linked in Qodo.
Link Git account →

Using GitHub Enterprise Server, GitLab Self-Managed, or Bitbucket Data Center?
These require an Enterprise plan - Contact us
Contact us →

[coderabbitai]

Warning

Review limit reached

@TinDang97, you've reached your PR review limit, so we couldn't start this review.

Next review available in: 45 minutes

Enable usage-based reviews in Billing to review now. Otherwise, wait until the next included review is available.
You're only billed for reviews past your plan's rate limits ($0.25/file).

How can I continue?

After more reviews become available, a review can be triggered using the @coderabbitai review command as a PR comment. Alternatively, push new commits to this PR.

To avoid repeated limits, reduce automatic review volume by pausing incremental auto-reviews earlier, using label-based review opt-in, excluding WIP or generated PR titles, or requesting reviews manually when the PR is ready. If your team needs uninterrupted high-volume reviews, an organization admin can enable usage-based reviews.

How do review limits work?

CodeRabbit enforces per-developer PR review limits for each organization. Most developers receive the normal plan review availability.

For paid Pro and Pro+ PR reviews, CodeRabbit uses adaptive limits for sustained high-volume activity. When a developer's recent PR review activity reaches the 95th percentile or higher among CodeRabbit users, additional reviews become available more gradually as earlier reviews age out of the rolling window.

Please refer docs for additional details.

Review details

⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Pro

Run ID: 81fa5a90-985b-4bbf-a3b4-f1fa84ed74fd

📥 Commits

Reviewing files that changed from the base of the PR and between f1a01a5 and 842f17c.

📒 Files selected for processing (6)

• src/client_registry.rs
• src/server/conn/handler_sharded/mod.rs
• src/shard/conn_accept.rs
• tests/mq_integration.rs
• tests/txn_kv_wiring.rs
• tests/workspace_integration.rs

📝 Walkthrough

Walkthrough

This PR bundles connection-plane durability and robustness fixes: shard-wide FLUSHDB/FLUSHALL broadcast, WAL v3 XactCommitV2 db-aware crash replay, migration fail-open handling, connected-client/backpressure metrics, lazy FunctionRegistry init, configurable TCP backlog, CLIENT KILL force-close, a max inline command size, bounded cross-shard spsc_send retries, and accept-loop backoff.

Changes

Connection-plane durability and lifecycle wave

Layer / File(s)	Summary
FLUSHDB/FLUSHALL cross-shard broadcast src/server/conn/handler_monoio/mod.rs, src/server/conn/handler_sharded/mod.rs, src/shard/coordinator.rs, tests/flush_cross_shard_scatter.rs	FLUSHDB/FLUSHALL now broadcasts to other shards via coordinate_flush_broadcast, returning an error frame on partial broadcast failure; a new integration test suite validates multi-shard flush isolation and clearing.
WAL v3 XactCommitV2 db-aware replay src/persistence/wal_v3/record.rs, src/persistence/wal_v3/replay.rs, src/transaction/mod.rs, src/server/conn/handler_monoio/txn.rs, src/server/conn/handler_sharded/txn.rs, src/server/conn/shared.rs	Adds WalRecordType::XactCommitV2 and encode_xact_commit_payload_v2 carrying db_index; CrossStoreTxn tracks db_index from TXN.BEGIN; replay selects the correct database (falling back to db 0) for crash recovery.
Connection migration fail-open src/server/conn/handler_sharded/mod.rs, src/shard/conn_accept.rs	Migration handoff failures now retry and fall back to serving the connection locally instead of dropping it, with corrected fd ownership.
Connected-client and backpressure metrics src/admin/metrics_setup.rs, src/client_registry.rs	New gauge/counter helpers track per-shard connected clients and cross-shard backpressure drops.
Lazy FunctionRegistry initialization src/server/conn/core.rs, src/server/conn/handler_monoio/dispatch.rs, src/server/conn/handler_monoio/mod.rs, src/server/conn/handler_sharded/mod.rs	Per-connection FunctionRegistry is now lazily constructed on first FUNCTION/FCALL use via ensure_function_registry.
Configurable TCP backlog src/config.rs, src/main.rs, src/shard/conn_accept.rs	Adds --tcp-backlog CLI flag applied to listener sockets at startup.

Estimated code review effort: 4 (Complex) | ~75 minutes

Connection-plane robustness hardening (Track B)

Layer / File(s)	Summary
CLIENT KILL force-close src/client_registry.rs, src/server/conn/handler_monoio/*, src/server/conn/handler_sharded/*, src/shard/conn_accept.rs	ClientLiveState gains kill_fd; kill_clients forcibly shuts down the target socket unless killing self; the raw fd is threaded through registration and migration paths.
Inline command max-size enforcement src/protocol/frame.rs, src/protocol/inline.rs, src/protocol/parse.rs, fuzz/fuzz_targets/*	DEFAULT_MAX_INLINE_SIZE (64KB) is added and parse_inline now rejects oversized inline lines instead of growing the buffer unbounded.
Bounded cross-shard spsc_send retry src/shard/coordinator.rs, src/shard/dispatch.rs, src/server/conn/handler_monoio/*, src/server/conn/handler_sharded/*, src/shard/scatter_*.rs, src/transaction/abort.rs	spsc_send now returns a #[must_use] PushOutcome after bounded retries, dropping messages on exhaustion; call sites now branch on or explicitly discard the outcome, surfacing partial-commit/rollback errors.
Accept-loop backoff on fd exhaustion src/server/accept_backoff.rs, src/server/listener.rs, src/server/mod.rs, src/shard/event_loop.rs, src/shard/uring_handler.rs	New AcceptBackoff applies capped exponential backoff and rate-limited logging on EMFILE/ENFILE-class accept errors across Tokio and Monoio accept loops.

Estimated code review effort: 4 (Complex) | ~80 minutes

Possibly related PRs

• pilotspace/moon#17: Extends the same connection migration/handler plumbing (MigratedConnectionState, MigrateConnection) touched by this PR's kill_fd and fail-open changes.
• pilotspace/moon#69: Overlaps src/client_registry.rs/kill_clients CLIENT KILL logic, extended here with force-close and self-scoping.
• pilotspace/moon#136: Related work on bounded cross-shard push/backpressure (PushOutcome) in the dispatch layer.

Suggested labels: enhancement

🚥 Pre-merge checks | ✅ 4 | ❌ 1

❌ Failed checks (1 warning)

Check name	Status	Explanation	Resolution
Description check	⚠️ Warning	The PR description covers the changes, but it misses the required Summary/Checklist/Performance Impact/Notes template structure.	Add the required ## Summary, ## Checklist, ## Performance Impact, and ## Notes sections, and fill the checklist and performance details explicitly.

✅ Passed checks (4 passed)

Check name	Status	Explanation
Title check	✅ Passed	The title is concise and accurately reflects the main connection-plane hardening work in this PR.
Docstring Coverage	✅ Passed	No functions found in the changed files to evaluate docstring coverage. Skipping docstring coverage check.
Linked Issues check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Out of Scope Changes check	✅ Passed	Check skipped because no linked issues were found for this pull request.

✨ Finishing Touches

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch fix/conn-plane-robustness-track-b

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

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

[pilotspacex-byte]

Two follow-up commits from a self-review + architecture pass:

b1a9f10 — review fixes on the original 4 hardenings:

• CLIENT KILL self-kill stays cooperative (fd shutdown skipped for the executing connection, so the +count reply is delivered — Redis parity)
• R-1 give-up now fails loud at the two side-effect senders: dropped MqTxnMaterialize ⇒ explicit MOONERR TXN.COMMIT partial error (was silent +OK with lost MQ intents); dropped GraphRollback ⇒ error!
• PushOutcome is #[must_use]; ≤64-iter spin before the first 100µs backoff (keeps transient-full latency in the old 10µs class); accept backoff capped 1s→100ms (sleep runs inside select! arms — bounds shutdown latency)

8f3f44d — durability & lifecycle wave (from the arch review):

• D-1: cross-store TXN crash replay restored KV ops into db 0 regardless of SELECT — new XactCommitV2 WAL record carries the db index; replay targets it (red/green tests)
• R-6: connection migration fails open on both runtimes — source socket kept until SPSC hand-off confirms; on give-up the client is re-served locally instead of dropped (also fixes a connected_clients leak on the tokio loss path)
• New metrics: moon_xshard_backpressure_drops_total{target_shard}, moon_shard_connected_clients{shard}
• Lazy per-connection FunctionRegistry (P-1 footprint); --tcp-backlog flag (S-5)

Full matrix locally green: fmt, clippy ×2, monoio 3773 / tokio 3118 lib tests, Linux (OrbStack) check.

[coderabbitai]

Actionable comments posted: 1

Caution

Some comments are outside the diff and can’t be posted inline due to platform limitations.

⚠️ Outside diff range comments (5)

src/server/conn/handler_sharded/write.rs (1)

140-162: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win

Don't silently drop the workspace-cleanup hop.

If this send backpressures, the workspace record is already removed but the key sweep never runs, and this branch still returns OK. Please branch on PushOutcome here and surface/retry the failure before acknowledging WS.DROP.

Suggested fix

-                                    let _ = crate::shard::coordinator::spsc_send(
-                                        &ctx.dispatch_tx,
-                                        ctx.shard_id,
-                                        cleanup_owner,
-                                        msg,
-                                        &ctx.spsc_notifiers,
-                                    )
-                                    .await;
-                                    match reply_rx.recv().await {
-                                        Ok(count) => {
-                                            tracing::debug!(
-                                                "WS.DROP cleanup: deleted {} keys on shard {}",
-                                                count,
-                                                cleanup_owner
-                                            );
-                                        }
-                                        Err(_) => {
-                                            tracing::warn!(
-                                                "WS.DROP cleanup: reply channel closed for shard {}",
-                                                cleanup_owner
-                                            );
-                                        }
-                                    }
+                                    if !matches!(
+                                        crate::shard::coordinator::spsc_send(
+                                            &ctx.dispatch_tx,
+                                            ctx.shard_id,
+                                            cleanup_owner,
+                                            msg,
+                                            &ctx.spsc_notifiers,
+                                        )
+                                        .await,
+                                        crate::shard::dispatch::PushOutcome::Pushed
+                                    ) {
+                                        responses.push(Frame::Error(Bytes::from_static(
+                                            b"ERR workspace cleanup backpressure",
+                                        )));
+                                        return true;
+                                    }
+                                    match reply_rx.recv().await {
+                                        Ok(count) => {
+                                            tracing::debug!(
+                                                "WS.DROP cleanup: deleted {} keys on shard {}",
+                                                count,
+                                                cleanup_owner
+                                            );
+                                        }
+                                        Err(_) => {
+                                            tracing::warn!(
+                                                "WS.DROP cleanup: reply channel closed for shard {}",
+                                                cleanup_owner
+                                            );
+                                        }
+                                    }

🤖 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/write.rs` around lines 140 - 162, In the
workspace-drop cleanup path inside the write handler, do not ignore the result
of shard::coordinator::spsc_send when sending the cleanup hop. Update the
WS.DROP branch in the sharded write flow to inspect PushOutcome before awaiting
reply_rx.recv(), and if the send is backpressured or otherwise fails, surface or
retry the failure instead of continuing to return OK. Use the existing
cleanup_owner, reply_rx, and spsc_send call site in the write handler to locate
the branch and make sure acknowledgement only happens after the cleanup hop is
successfully queued.

src/server/conn/handler_monoio/write.rs (1)

145-153: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win

Don't silently drop the workspace-cleanup hop.

If this send backpressures, the workspace record is already removed but the key sweep never runs, and this branch still returns OK. Please branch on PushOutcome here and surface/retry the failure before acknowledging WS.DROP.

Suggested fix

-                                    let _ = crate::shard::coordinator::spsc_send(
-                                        &ctx.dispatch_tx,
-                                        ctx.shard_id,
-                                        owner,
-                                        msg,
-                                        &ctx.spsc_notifiers,
-                                    )
-                                    .await;
-                                    let _ = reply_rx.recv().await;
+                                    if !matches!(
+                                        crate::shard::coordinator::spsc_send(
+                                            &ctx.dispatch_tx,
+                                            ctx.shard_id,
+                                            owner,
+                                            msg,
+                                            &ctx.spsc_notifiers,
+                                        )
+                                        .await,
+                                        crate::shard::dispatch::PushOutcome::Pushed
+                                    ) {
+                                        responses.push(Frame::Error(Bytes::from_static(
+                                            b"ERR workspace cleanup backpressure",
+                                        )));
+                                        return true;
+                                    }
+                                    let _ = reply_rx.recv().await;

🤖 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/write.rs` around lines 145 - 153, The
workspace-cleanup path in the monoio write handler is swallowing the result of
the coordinator hop, so a backpressured send can still end in an OK reply
without running the key sweep. Update the branch in the write handler around
spsc_send and reply_rx.recv to inspect the returned PushOutcome, and only
acknowledge WS.DROP after the cleanup hop succeeds; on failure, surface or retry
the error instead of discarding it.

src/shard/event_loop.rs (2)

322-347: 🚀 Performance & Scalability | 🟠 Major | ⚡ Quick win

Hoist per-error format!() out of the accept-error hot path.

ctx is recomputed with format!() on every accept error inside the loop, even though shard_id_copy never changes. As per coding guidelines, format!() must not be used on hot paths in shard event loops (src/shard/event_loop.rs is explicitly listed). Compute it once before the loop.

As per coding guidelines, "Do not use Box::new(), Vec::new(), String::new(), Arc::new(), clone(), format!(), or to_string() on hot paths in command dispatch, protocol parsing, shard event loops, or I/O drivers."

♻️ Proposed fix

                 monoio::spawn(async move {
                     let mut accept_backoff = crate::server::accept_backoff::AcceptBackoff::new();
+                    let ctx = format!("Shard {shard_id_copy}: per-shard accept error");
                     loop {
                         match listener.accept().await {
                             Ok((stream, _addr)) => {
                                 accept_backoff.reset();
                                 ...
                             }
                             Err(e) => {
                                 // R-4: capped backoff + rate-limited log so an
                                 // fd-exhaustion storm can't hot-spin this shard.
-                                let ctx = format!("Shard {shard_id_copy}: per-shard accept error");
                                 accept_backoff.record_error(&ctx, &e).await;
                             }
                         }
                     }
                 });

🤖 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/shard/event_loop.rs` around lines 322 - 347, The accept-error path in the
shard event loop is recomputing a formatted context string on every loop
iteration, which violates the hot-path guidelines. In the monoio::spawn accept
loop, hoist the per-shard context creation out of the loop in the surrounding
closure before calling listener.accept(), and then pass that precomputed value
into AcceptBackoff::record_error. Use the existing shard_id_copy and the
accept_backoff/record_error flow to locate and update the code without changing
the loop’s behavior.

Source: Coding guidelines

---

1065-1136: 🚀 Performance & Scalability | 🟠 Major | ⚡ Quick win

Same per-error format!() allocation on the tokio per-shard accept branch.

ctx is rebuilt via format!() inside the Err arm of the tokio::select! accept branch on every accept failure. Hoist it out, computed once alongside per_shard_accept_backoff, since shard_id is constant for the life of the loop.

As per coding guidelines, "Do not use Box::new(), Vec::new(), String::new(), Arc::new(), clone(), format!(), or to_string() on hot paths in command dispatch, protocol parsing, shard event loops, or I/O drivers."

♻️ Proposed fix

         #[cfg(feature = "runtime-tokio")]
         let mut per_shard_accept_backoff = crate::server::accept_backoff::AcceptBackoff::new();
+        #[cfg(feature = "runtime-tokio")]
+        let per_shard_accept_ctx = format!("Shard {shard_id}: per-shard accept error");
...
                         Err(e) => {
-                            let ctx = format!("Shard {shard_id}: per-shard accept error");
-                            per_shard_accept_backoff.record_error(&ctx, &e).await;
+                            per_shard_accept_backoff.record_error(&per_shard_accept_ctx, &e).await;
                         }

🤖 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/shard/event_loop.rs` around lines 1065 - 1136, The tokio per-shard accept
error path in the shard event loop rebuilds the same context string on every
failure, which violates the hot-path allocation guideline. Hoist the accept
error context out of the `tokio::select!` branch in `event_loop.rs`, near
`per_shard_accept_backoff`, so it is computed once per shard/loop instead of
inside the `Err` arm. Keep the existing `record_error` call in the
`per_shard_accept_backoff` handling, but pass the precomputed context value
there. Use the `per_shard_accept_backoff` setup and the per-shard accept branch
as the main anchors when updating `event_loop`.

Source: Coding guidelines

src/server/conn/handler_sharded/txn.rs (1)

151-151: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win

MQ commit should use txn.db_index

conn.selected_db can change after TXN.BEGIN, but txn.db_index is the transaction’s BEGIN-time db used for KV WAL/replay. Using the live connection db here can materialize self-intents and MqTxnMaterialize into a different database than the transaction’s KV state. Update both this local apply and the foreign-shard payload to use txn.db_index.

🤖 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/txn.rs` at line 151, The MQ commit path is
using the live connection database instead of the transaction’s BEGIN-time
database, which can desync materialization from KV WAL/replay. Update the local
apply in txn::handle_commit and the foreign-shard payload in the sharded
transaction flow to use txn.db_index rather than conn.selected_db, so
self-intents and MqTxnMaterialize are applied against the same database as the
transaction state.

🤖 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/shard/conn_accept.rs`:
- Around line 782-845: The fail-open migration fallback is dropping
already-buffered bytes because both resume paths call
handle_connection_sharded_monoio with BytesMut::new(). Update the local-recovery
logic in conn_accept.rs to pass the saved remainder from
state.read_buf_remainder into the initial_read_buf argument when retrying
locally, including both the branch after the initial monoio handoff and the
branch after a full SPSC push failure. Keep the fix centered around
handle_connection_sharded_monoio and the MigrateConnectionPayload/state recovery
path.

---

Outside diff comments:
In `@src/server/conn/handler_monoio/write.rs`:
- Around line 145-153: The workspace-cleanup path in the monoio write handler is
swallowing the result of the coordinator hop, so a backpressured send can still
end in an OK reply without running the key sweep. Update the branch in the write
handler around spsc_send and reply_rx.recv to inspect the returned PushOutcome,
and only acknowledge WS.DROP after the cleanup hop succeeds; on failure, surface
or retry the error instead of discarding it.

In `@src/server/conn/handler_sharded/txn.rs`:
- Line 151: The MQ commit path is using the live connection database instead of
the transaction’s BEGIN-time database, which can desync materialization from KV
WAL/replay. Update the local apply in txn::handle_commit and the foreign-shard
payload in the sharded transaction flow to use txn.db_index rather than
conn.selected_db, so self-intents and MqTxnMaterialize are applied against the
same database as the transaction state.

In `@src/server/conn/handler_sharded/write.rs`:
- Around line 140-162: In the workspace-drop cleanup path inside the write
handler, do not ignore the result of shard::coordinator::spsc_send when sending
the cleanup hop. Update the WS.DROP branch in the sharded write flow to inspect
PushOutcome before awaiting reply_rx.recv(), and if the send is backpressured or
otherwise fails, surface or retry the failure instead of continuing to return
OK. Use the existing cleanup_owner, reply_rx, and spsc_send call site in the
write handler to locate the branch and make sure acknowledgement only happens
after the cleanup hop is successfully queued.

In `@src/shard/event_loop.rs`:
- Around line 322-347: The accept-error path in the shard event loop is
recomputing a formatted context string on every loop iteration, which violates
the hot-path guidelines. In the monoio::spawn accept loop, hoist the per-shard
context creation out of the loop in the surrounding closure before calling
listener.accept(), and then pass that precomputed value into
AcceptBackoff::record_error. Use the existing shard_id_copy and the
accept_backoff/record_error flow to locate and update the code without changing
the loop’s behavior.
- Around line 1065-1136: The tokio per-shard accept error path in the shard
event loop rebuilds the same context string on every failure, which violates the
hot-path allocation guideline. Hoist the accept error context out of the
`tokio::select!` branch in `event_loop.rs`, near `per_shard_accept_backoff`, so
it is computed once per shard/loop instead of inside the `Err` arm. Keep the
existing `record_error` call in the `per_shard_accept_backoff` handling, but
pass the precomputed context value there. Use the `per_shard_accept_backoff`
setup and the per-shard accept branch as the main anchors when updating
`event_loop`.

🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Pro

Run ID: 5f3d749b-4bec-4a6e-977d-1c30e35da9a6

📥 Commits

Reviewing files that changed from the base of the PR and between 73adcc6 and f1a01a5.

📒 Files selected for processing (37)

• .gitignore
• CHANGELOG.md
• fuzz/fuzz_targets/inline_parse.rs
• fuzz/fuzz_targets/resp_parse.rs
• fuzz/fuzz_targets/resp_parse_differential.rs
• src/admin/metrics_setup.rs
• src/client_registry.rs
• src/config.rs
• src/main.rs
• src/persistence/wal_v3/record.rs
• src/persistence/wal_v3/replay.rs
• src/protocol/frame.rs
• src/protocol/inline.rs
• src/protocol/parse.rs
• src/server/accept_backoff.rs
• src/server/conn/core.rs
• src/server/conn/handler_monoio/dispatch.rs
• src/server/conn/handler_monoio/mod.rs
• src/server/conn/handler_monoio/txn.rs
• src/server/conn/handler_monoio/write.rs
• src/server/conn/handler_sharded/dispatch.rs
• src/server/conn/handler_sharded/mod.rs
• src/server/conn/handler_sharded/txn.rs
• src/server/conn/handler_sharded/write.rs
• src/server/conn/shared.rs
• src/server/listener.rs
• src/server/mod.rs
• src/shard/conn_accept.rs
• src/shard/coordinator.rs
• src/shard/dispatch.rs
• src/shard/event_loop.rs
• src/shard/scatter_aggregate.rs
• src/shard/scatter_hybrid.rs
• src/shard/uring_handler.rs
• src/transaction/abort.rs
• src/transaction/mod.rs
• tests/flush_cross_shard_scatter.rs

[coderabbitai]

🎯 Functional Correctness | 🟠 Major | ⚡ Quick win

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
set -euo pipefail

# Locate the relevant handler and state definitions.
rg -n --no-heading \
  "read_buf_remainder|take_migration_read_buf|initial_read_buf|MigrateConnection|handle_connection_sharded_monoio|ConnectionState::new" \
  src/shard/conn_accept.rs src -g '*.rs'

# Show the relevant slice around the reviewed lines.
sed -n '740,880p' src/shard/conn_accept.rs

# Inspect the connection handler/state definitions that consume initial_read_buf.
fd -a 'handler_monoio.rs' src
fd -a 'connection_state.rs' src

Repository: pilotspace/moon

Length of output: 16492

---

🏁 Script executed:

#!/bin/bash
set -euo pipefail

# Show the handler setup that consumes initial_read_buf in monoio.
sed -n '106,170p' src/server/conn/handler_monoio/mod.rs

# Show the migration result construction where read_buf_remainder is captured.
sed -n '1938,1968p' src/server/conn/handler_monoio/mod.rs

# Show the spawn path that reconstructs migrated connections.
sed -n '400,445p' src/shard/conn_accept.rs

Repository: pilotspace/moon

Length of output: 6009

---

Preserve the buffered remainder on fail-open migration
Both fail-open paths restart the handler with BytesMut::new(), so any bytes already read into state.read_buf_remainder are dropped and the connection can desync. Thread that remainder into initial_read_buf when resuming locally in both branches.

🤖 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/shard/conn_accept.rs` around lines 782 - 845, The fail-open migration
fallback is dropping already-buffered bytes because both resume paths call
handle_connection_sharded_monoio with BytesMut::new(). Update the local-recovery
logic in conn_accept.rs to pass the saved remainder from
state.read_buf_remainder into the initial_read_buf argument when retrying
locally, including both the branch after the initial monoio handoff and the
branch after a full SPSC push failure. Keep the fix centered around
handle_connection_sharded_monoio and the MigrateConnectionPayload/state recovery
path.