fix(shard): FLUSHALL inside MULTI/EXEC clears every shard, not just one (c10k E2)

[TinDang97]

Stacked on #434 (fix/c10k-d2-partial-frame-park). Review the top commit only; the base merges first.

The bug (c10k hardening finding E2)

A keyless FLUSHDB/FLUSHALL queued inside a transaction cleared only the slice the transaction body ran on, while EXEC still answered +OK. At --shards 4 that leaves ~¾ of the keyspace alive after the client was told the database was emptied — a silent wrong answer to a destructive command, typically noticed much later via a non-zero DBSIZE. Measured before the fix: 45 of 64 keys survived a transaction that reported success.

Why

The live (non-MULTI) path has broadcast since D-2 via coordinate_flush_broadcast, which also turns any failed leg into an explicit partial-flush error rather than a silent success. The transactional executor never got the same treatment — execute_transaction_sharded runs the queued body against the local slice with no per-key routing and no fan-out.

Fix

Follows the exec_publishes contract already established for deferred transactional work: the executor records each flush as (result_index, command, db) and the originator performs the fan-out (broadcast_txn_flushes). The executor can't do it itself — it's synchronous while the broadcast awaits, and for a routed transaction it runs on the owner shard where fanning out from inside that shard's own message loop risks a shard-to-shard wait cycle.

broadcast_txn_flushes skips the leg that already ran and patches result[result_index] with the partial-flush error if any remaining leg fails, so a +OK for a flush inside a transaction can be trusted exactly as on the live path.

Both handlers covered (monoio + sharded/tokio) and both transaction shapes (local body, routed-to-owner body). The per-entry selected db is recorded, so a queued SELECT before the flush is honoured.

Cross-shard atomicity is unchanged and unchangeable: a concurrent reader can still observe shard A flushed before shard B. MULTI bounds the report, not the visibility, in a shared-nothing engine.

Test (red/green)

flushall_inside_multi_clears_every_shard — SETs 64 keys at --shards 4, asserts DBSIZE == 64, runs MULTI; FLUSHALL; EXEC, then asserts either DBSIZE == 0 (if EXEC claimed success) or an explicit error. Asserts through DBSIZE rather than per-key GETs so it fails on ANY surviving key. Verified 4/4 under both runtimes: DBSIZE went Int(45) → Int(0).

The suite's Moon harness also gained a Drop guard: kill9 only runs on the happy path, so a failed assertion unwound past it and stranded the server (one such leak ran four hours holding its port + data dir).

Refs: c10k hardening review, finding E2

[coderabbitai]

Warning

Review limit reached

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

Next review available in: 20 minutes

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

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 Plus

Run ID: f31e5780-f4fa-479f-9ebd-fc8c977dc0a0

📥 Commits

Reviewing files that changed from the base of the PR and between 4c25c9d and c4d39f8.

📒 Files selected for processing (8)

• CHANGELOG.md
• src/server/conn/handler_monoio/write.rs
• src/server/conn/handler_sharded/write.rs
• src/server/conn/shared.rs
• src/shard/coordinator.rs
• src/shard/dispatch.rs
• src/shard/spsc_handler.rs
• tests/sharded_multi_exec_routing.rs

---

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

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

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

[qodo-code-review]

PR Summary by Qodo

Fix sharded MULTI/EXEC FLUSHALL/FLUSHDB to broadcast and report partial failures

🐞 Bug fix 🧪 Tests 📝 Documentation 🕐 40+ Minutes

AI Description

• Record keyless FLUSHALL/FLUSHDB inside MULTI/EXEC and broadcast to all shards after local
 execution.
• Patch EXEC result entries to explicit partial-flush errors when any broadcast leg fails.
• Add regression test ensuring a reported successful FLUSHALL leaves DBSIZE==0 across shards.

Diagram

graph TD
  A[Client] --> B["Conn handler (monoio/tokio)"] --> C["execute_transaction_sharded"] --> D["Local shard flush"]
  C --> E["exec_flushes (record)"] --> F["broadcast_txn_flushes"] --> G["coordinate_flush_broadcast"] --> H["Other shards flush"]

Loading

High-Level Assessment

The following are alternative approaches to this PR:

1. Broadcast inside transaction executor

• ➕ Keeps all flush semantics encapsulated within execute_transaction_sharded
• ➕ Avoids passing exec_flushes through dispatch/reply plumbing
• ➖ Executor is synchronous; broadcast requires awaiting
• ➖ For routed txns, executor runs on owner shard; broadcasting from inside its message loop risks shard-to-shard wait cycles/deadlock patterns

2. Reject FLUSHALL/FLUSHDB inside MULTI when sharded

• ➕ Simple and avoids non-atomic cross-shard visibility concerns
• ➖ Behavioral breaking change vs Redis expectations and current live-path behavior
• ➖ Does not address existing need for transactional flush semantics; forces clients to restructure

3. Pre-route queued FLUSH* at queue time (convert to special distributed op)

• ➕ Could avoid recording (result_index, cmd, db) and later patching results
• ➕ Potentially simplifies EXEC-time logic
• ➖ More invasive changes to the command queueing/transaction model
• ➖ Harder to preserve per-entry SELECT semantics and result ordering guarantees

Recommendation: Keep the PR’s approach: record flush intents during synchronous txn execution and have the originator perform the async fan-out via the existing broadcast machinery, patching the EXEC result entry on failures. This matches the established exec_publishes deferral contract, preserves queued SELECT semantics (db captured per flush), and avoids deadlock risk from broadcasting within the owner shard’s message loop.

Files changed (8) +255 / -4

Bug fix (6) +142 / -4

write.rsBroadcast recorded txn flushes and patch EXEC results (monoio handler) +33/-2

Broadcast recorded txn flushes and patch EXEC results (monoio handler)

• Collects exec_flushes during execute_transaction_sharded and, after local execution/AOF/replication, broadcasts those flushes to remaining shards. For routed transactions, applies the same broadcast-and-patch logic to the routed reply before returning it to the client.

src/server/conn/handler_monoio/write.rs

write.rsBroadcast recorded txn flushes and patch EXEC results (tokio/sharded handler) +31/-2

Broadcast recorded txn flushes and patch EXEC results (tokio/sharded handler)

• Mirrors the monoio handler fix: collects exec_flushes from the txn executor, broadcasts them after local execution, and patches the EXEC result entry if any broadcast leg fails. Also applies the logic to routed transaction replies.

src/server/conn/handler_sharded/write.rs

shared.rsRecord FLUSHALL/FLUSHDB occurrences during sharded transaction execution +14/-0

Record FLUSHALL/FLUSHDB occurrences during sharded transaction execution

• Extends execute_transaction_sharded to accept an exec_flushes out-parameter. When a queued FLUSHDB/FLUSHALL succeeds locally, records (result_index, command_frame, selected_db) so the originator can later broadcast to other shards while preserving queued SELECT semantics.

src/server/conn/shared.rs

coordinator.rsAdd broadcast_txn_flushes helper to fan out txn-local flushes +48/-0

Add broadcast_txn_flushes helper to fan out txn-local flushes

• Introduces broadcast_txn_flushes, which reuses coordinate_flush_broadcast for each recorded flush and replaces the corresponding EXEC array element with the partial-flush error if any leg fails. Skips the shard that already executed the txn body.

src/shard/coordinator.rs

dispatch.rsExtend TxnExecReply to return recorded flushes to the originator +9/-0

Extend TxnExecReply to return recorded flushes to the originator

• Adds exec_flushes to the cross-shard transaction reply payload, carrying (result_index, command_frame, db). This enables originator-side broadcast and result patching without awaiting inside the owner shard’s message loop.

src/shard/dispatch.rs

spsc_handler.rsCollect exec_flushes during cross-shard EXEC handling +7/-0

Collect exec_flushes during cross-shard EXEC handling

• Creates and passes an exec_flushes vector into execute_transaction_sharded and returns it in TxnExecReply. This ensures routed transactions provide the originator enough information to broadcast flushes and correct the reported result.

src/shard/spsc_handler.rs

Tests (1) +93 / -0

sharded_multi_exec_routing.rsAdd regression test for FLUSHALL inside MULTI across shards; ensure Moon cleanup on failure +93/-0

Add regression test for FLUSHALL inside MULTI across shards; ensure Moon cleanup on failure

• Adds a Drop guard to Moon to prevent leaked server processes when assertions fail. Introduces flushall_inside_multi_clears_every_shard, which writes 64 keys across 4 shards, runs MULTI; FLUSHALL; EXEC, and asserts either DBSIZE==0 on claimed success or an explicit error response.

tests/sharded_multi_exec_routing.rs

Documentation (1) +20 / -0

CHANGELOG.mdDocument fix for transactional FLUSHALL/FLUSHDB sharding correctness (c10k E2) +20/-0

Document fix for transactional FLUSHALL/FLUSHDB sharding correctness (c10k E2)

• Adds a detailed changelog entry describing the prior silent partial-flush behavior for FLUSHALL/FLUSHDB inside MULTI/EXEC and the new broadcast-and-error-patching behavior. Clarifies the non-atomic visibility caveat across shards.

CHANGELOG.md

[qodo-code-review]

Code Review by Qodo

🐞 Bugs (0) 📘 Rule violations (0) 📜 Skill insights (0)

1. Misrouted txn flush broadcast ⊘ Outdated 🐞 Bug ≡ Correctness

Description

broadcast_txn_flushes passes exec_shard as the sender shard into
coordinate_flush_broadcast/spsc_send, but spsc_send indexes producers via
ChannelMesh::target_index(my_shard,target_shard). When a transaction is routed to an owner shard,
the originator calls broadcast_txn_flushes with its own dispatch_tx but a different exec_shard, so
messages are pushed into the wrong producer slots and some shards may not flush (or the wrong shards
flush).

Code

src/shard/coordinator.rs[R1074-1078]

+        if let Err(err) = coordinate_flush_broadcast(
+            command,
+            exec_shard,
+            num_shards,
+            *db_index,

Relevance

●●● Strong

High-severity cross-shard correctness bug; team historically accepts coordinator correctness
hardening.

PR-#17

ⓘ Recommendations generated based on similar findings in past PRs

Evidence

spsc_send selects the outbound producer by computing an index from (my_shard,target_shard);
therefore my_shard must match the shard that owns dispatch_tx. broadcast_txn_flushes instead passes
exec_shard, and the routed EXEC path calls it with the owner shard id (s) while still using the
originator shard’s ctx.dispatch_tx, so the producer index is computed against the wrong my_shard and
pushes to the wrong channel slot.

src/shard/coordinator.rs[867-907]
src/shard/mesh.rs[157-173]
src/shard/coordinator.rs[1062-1088]
src/server/conn/handler_monoio/write.rs[820-838]

Agent prompt

The issue below was found during a code review. Follow the provided context and guidance below and implement a solution

### Issue description
`broadcast_txn_flushes()` currently forwards `exec_shard` into `coordinate_flush_broadcast()` as the `my_shard`/sender identity. This is incorrect when the MULTI/EXEC body was executed on an owner shard different from the originator shard that is performing the broadcast.

`coordinate_flush_broadcast()` ultimately uses `spsc_send()`, which selects a producer slot via `ChannelMesh::target_index(my_shard, target_shard)`. That mapping assumes `my_shard` is the *actual* sender shard owning `dispatch_tx`. Passing the owner/execution shard while using the originator’s `dispatch_tx` misroutes messages.

### Issue Context
- Routed transaction path calls `broadcast_txn_flushes(..., exec_shard=s, dispatch_tx=&ctx.dispatch_tx, ...)`, where `s` is the owner shard that ran the txn body.
- `ctx.dispatch_tx` belongs to the connection’s originator shard.
- SPSC producer indexing depends on the originator shard id.

### Fix Focus Areas
- src/shard/coordinator.rs[1062-1088]
- src/server/conn/handler_monoio/write.rs[824-837]
- src/server/conn/handler_sharded/write.rs[649-662]

### Implementation direction
1. Change the API so broadcast has both:
  - `sender_shard` (the shard that owns `dispatch_tx` / current handler shard), and
  - `already_flushed_shard` (the shard that ran the txn body and already executed the local flush).
2. Ensure `spsc_send()` is always called with `sender_shard`.
3. Decide how to flush `sender_shard` when `sender_shard != already_flushed_shard`:
  - Either execute the flush locally on `sender_shard` through the normal local command path (so it gets the same durability/index-clearing behavior), then broadcast to the remaining shards; or
  - Broadcast to *all* other shards (including `already_flushed_shard`) and accept the idempotent re-flush, but still handle the sender shard’s local flush explicitly since SPSC cannot send to self.
4. Preserve the result patching (`result[result_index] = partial-flush error`) on any failed leg.

ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools

---

2. Double kill/wait on drop ⊘ Outdated 🐞 Bug ☼ Reliability

Description

Moon::kill9 consumes Moon, but Drop also unconditionally calls kill()+wait(), so the happy path
executes a second kill/wait after kill9 completes. This is redundant and makes the harness fragile;
other tests avoid double cleanup by storing Child in an Option and taking it in either kill9 or
Drop.

Code

tests/sharded_multi_exec_routing.rs[R62-70]

+impl Drop for Moon {
+    /// `kill9` only runs on the happy path — a failed assertion unwinds past
+    /// it and strands the server. One such leak ran for four hours before it
+    /// was noticed, holding its port and its data dir the whole time. Killing
+    /// again after `kill9` has already reaped is a harmless no-op error.
+    fn drop(&mut self) {
+        let _ = self.child.kill();
+        let _ = self.child.wait();
+    }

Relevance

●●● Strong

Deterministic reliability improvement to test harness cleanup; aligns with prior accepted test
hardening changes.

PR-#65

ⓘ Recommendations generated based on similar findings in past PRs

Evidence

The file defines both a consuming kill9() that kills+waits and a Drop impl that also kills+waits. A
nearby test shows the preferred Option<Child> take()-on-drop pattern used elsewhere in the repo to
avoid repeated cleanup.

tests/sharded_multi_exec_routing.rs[55-70]
tests/allocator_mimalloc_smoke.rs[65-77]

Agent prompt

The issue below was found during a code review. Follow the provided context and guidance below and implement a solution

### Issue description
`Moon::kill9(self)` kills and waits for the child process, but because it consumes `self`, Rust will drop the `Moon` immediately afterward, triggering the new `Drop` impl which kills and waits again.

While errors are ignored, this double-cleanup is unnecessary and makes the harness pattern inconsistent with other tests that use a kill-on-drop guard.

### Issue Context
Other tests in this repo use a `MoonGuard(Option<Child>)` pattern and `take()` the child in `Drop` to guarantee the kill/wait happens at most once.

### Fix Focus Areas
- tests/sharded_multi_exec_routing.rs[50-71]

### Implementation direction
- Change `Moon` to store `child: Option<Child>`.
- Implement `kill9(&mut self)` (or `kill9(mut self)` but `take()` internally) that does:
 - `if let Some(mut child) = self.child.take() { child.kill(); child.wait(); }`
- Update `Drop` to also `take()` and only kill/wait when still present.
- Update call sites to use `moon.kill9();` with a mutable binding if needed.

ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools

---

3. Non-deterministic shard coverage ⊘ Outdated 🐞 Bug ☼ Reliability

Description

flushall_inside_multi_clears_every_shard assumes 64 distinct untagged keys will cover all shards
“with high probability”, but it never verifies coverage. If the chosen keys land on only one shard
(now or after future routing/hash changes), the test can pass even when only one shard is flushed.

Code

tests/sharded_multi_exec_routing.rs[R408-411]

+    // Spread keys across all 4 shards. Untagged keys hash-route by full key,
+    // so a spread of distinct names covers every shard with high probability;
+    // 64 makes an all-on-one-shard fluke effectively impossible.
+    for i in 0..64 {

Relevance

●● Moderate

Test determinism improvement is plausible, but similar “reduce flakiness/determinism” suggestions
have been rejected before.

PR-#426

ⓘ Recommendations generated based on similar findings in past PRs

Evidence

The test explicitly states it relies on “high probability” shard spread and does not assert
per-shard coverage. Routing is deterministic and supports hash tags, so the test can
deterministically generate keys for each shard instead of relying on probability.

tests/sharded_multi_exec_routing.rs[408-418]
src/shard/dispatch.rs[162-173]

Agent prompt

The issue below was found during a code review. Follow the provided context and guidance below and implement a solution

### Issue description
The new regression test relies on probabilistic distribution of untagged keys across shards. For a correctness regression test, it should deterministically ensure that at least one key exists on every shard before issuing FLUSHALL inside MULTI/EXEC.

### Issue Context
Shard routing is deterministic (`key_to_shard` hashes the key or `{tag}` content). The test can select keys that intentionally map to each shard.

### Fix Focus Areas
- tests/sharded_multi_exec_routing.rs[408-418]

### Implementation direction
Option A (preferred):
- In the test, import and use the server’s routing function (e.g. `moon::shard::dispatch::key_to_shard`) to generate keys until you have N keys for each shard (0..3).
- Assert that the selected key set covers all shards before proceeding.

Option B:
- Use hash tags `{tag}` and brute-force small tag strings until you find tags that map to each shard (using the same `key_to_shard` function in-test).

Either way, make shard coverage explicit so the test cannot silently pass without exercising multi-shard state.

ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools

---

4. Test doesn't set MOON_BIN ⊘ Outdated 📘 Rule violation ▣ Testability

Description

The new integration test starts a moon server without explicitly setting MOON_BIN, relying on
spawn_moon_first's fallback behavior. This violates the requirement to pin the server binary
explicitly to avoid accidentally running a stale or wrong build artifact.

Code

tests/sharded_multi_exec_routing.rs[R401-404]

+    let dir = tempfile::tempdir().expect("tempdir");
+    let Some(moon) = spawn_moon_first(dir.path(), false) else {
+        return; // binary missing — skip
+    };

Relevance

● Weak

Close precedent: team rejected enforcing explicit MOON_BIN pinning for tests; likely keep current
spawn fallback/skip behavior.

PR-#216
PR-#421

ⓘ Recommendations generated based on similar findings in past PRs

Evidence

PR Compliance ID 992389 requires integration tests that spawn the moon server to set MOON_BIN
explicitly. The new test calls spawn_moon_first(...) directly and even allows skipping when the
binary is missing, without setting MOON_BIN first.

Rule 992389: Integration tests must set MOON_BIN explicitly for server binaries
tests/sharded_multi_exec_routing.rs[401-404]

Agent prompt

The issue below was found during a code review. Follow the provided context and guidance below and implement a solution

## Issue description
The new integration test `flushall_inside_multi_clears_every_shard` spawns the server without explicitly pinning the server binary via `MOON_BIN`, which violates the integration-test requirement and can lead to running an unintended binary.

## Issue Context
Rule PR Compliance ID 992389 requires integration tests that spawn the moon server to set `MOON_BIN` explicitly and not rely on fallback discovery logic.

## Fix Focus Areas
- tests/sharded_multi_exec_routing.rs[401-404]

ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools

---

Context used

✅ Compliance rules (platform): 55 rules

_{To customize comments, go to the Qodo configuration screen, or learn more in the docs.}