fix(server): sharded MULTI/EXEC safety, durability & owner-routing (Phase A+B)#247
Merged
Merged
Conversation
…cing writes (Phase A)
`execute_transaction_sharded` runs the whole queued transaction body on the
connection's OWN shard via `with_shard_db`, with no per-key routing. At
`--shards >= 2` a queued key owned by a different shard was therefore silently
written to (or read from) the wrong shard's table: EXEC reported success while
the data diverged, and other connections — which route each key to its true
owner — saw nothing. Silent lost updates, the worst failure class.
Hash tags did NOT mitigate this: `{tag}` co-locates keys with each other but
not with the connection's execution shard (random via SO_REUSEPORT accept), and
connection migration is explicitly disabled during MULTI.
Phase A closes the silent-corruption hole:
- New `shared::analyze_txn_locality` classifies a queued body (Keyless /
SingleShard(s) / CrossShard) using the command-metadata key specs
(`command_keys`) and `key_to_shard`, so multi-key commands contribute every
key and `{hash tags}` are honored identically to the normal routing path.
- Both sharded EXEC handlers (handler_sharded, handler_monoio) reject with
`CROSSSLOT` any transaction whose keys aren't all owned by the executing
shard, instead of misplacing the writes. Guarded on `num_shards > 1`, so
`--shards 1` is entirely unaffected.
Invariant restored: EXEC-success => every write is visible to other
connections; CROSSSLOT => nothing was written.
Phase B (route a single-owner-shard body to its owner so hash-tagged
transactions run correctly from any connection) is the follow-up; genuinely
multi-shard transactions stay rejected (a shared-nothing engine can't span
shards atomically without distributed commit).
Red/green: tests/sharded_multi_exec_locality.rs pins the no-silent-divergence
invariant (fails on the pre-fix binary: "EXEC OK but GET nil"), multi-shard
span rejection, and single-shard non-regression; plus analyze_txn_locality unit
tests. fmt + clippy (default and tokio,jemalloc) clean; 3948 lib tests green.
Found in passing (unrelated, out of scope): a solo GET sent mid-MULTI on the
monoio single-shard handler executes immediately instead of queueing.
author: Tin Dang
… restart) `execute_transaction_sharded` — the MULTI/EXEC executor used by the monoio handler at EVERY shard count (including `--shards 1`) and by the tokio sharded handler at `--shards >= 2` — appended nothing to the AOF. Every transactional write was therefore silently lost on restart under `appendonly=yes`, while an identical write issued OUTSIDE MULTI survived. EXEC acked success and the data was gone after a restart — the worst durability failure class. Root cause: unlike the single-shard tokio `execute_transaction` (which returns `(Frame, Vec<Bytes>)` and whose caller persists the AOF entries), the sharded executor returned a bare `Frame` and never serialized or logged its writes. Fix: - `execute_transaction_sharded` now serializes each write command BEFORE dispatch (matching `execute_transaction`) and returns `(Frame, Vec<Bytes>)` with the AOF bytes of every write that actually succeeded (errors excluded). - New shared `persist_txn_aof(ctx, aof_entries)` appends the entries to the owning shard's writer via the normal group-commit path (`issue_append_lsn` + `send_append_group`) and issues ONE `fsync_barrier` under `appendfsync=always` before the EXEC ack. All keys in the body are owned by `ctx.shard_id` (Phase A rejects foreign-owned bodies), so that is the correct AOF target. - On a barrier/append failure EXEC returns `AOF_FSYNC_ERR` and suppresses any queued PUBLISH fan-out — parity with the normal write path; it must not ack a durability it can't guarantee, nor emit pub/sub side effects for a txn the client sees fail. - `try_handle_multi_exec` is now `async` in both sharded handlers; both call sites updated with `.await`. Fully-qualified `crate::command::metadata::is_write` / `bytes::BytesMut` are used because those imports are `runtime-tokio`-gated but this executor compiles under both runtimes. Red/green: tests/sharded_multi_exec_durability.rs pins "an EXEC-committed write survives kill-9 + restart, exactly like a non-MULTI write" (deterministic at `--shards 1` + monoio + `appendfsync=always`). Verified RED against a neutered persist path: the txn key returns nil after restart while the plain control key survives; GREEN with the fix. fmt + clippy (default and tokio,jemalloc) clean; existing locality + transaction suites green (4 integration, 5 locality-unit, 40 transaction-lib). Follow-up (same effort): Phase B routes a single-owner-shard body to its owner so hash-tagged transactions run correctly from any connection. author: Tin Dang
Sharded MULTI/EXEC executes on the connection's accept shard (random via
SO_REUSEPORT), but keys route per-key to their owning shard. Phase A closed the
resulting silent-corruption hole by rejecting any transaction whose keys weren't
owned by the accept shard with CROSSSLOT — which meant a hash-tagged `{tag}`
multi-key transaction only worked from the ~1/N connections that happened to
land on the owning shard.
Phase B routes such a body to its owner instead of rejecting it:
- New boxed `ShardMessage::TxnExecute(Box<TxnExecutePayload>)` (kept within the
64-byte cache-line cap, mirroring `MqCommand`) carrying the queued body +
db_index + a `TxnExecReply` oneshot.
- `coordinator::execute_txn_on_owner` performs the SPSC hop and awaits the reply
(returns None if the owner shard's reply channel closed → caller surfaces an
error, never a false success).
- The owner's SPSC handler runs the whole body via the existing
`execute_transaction_sharded` on ITS slice (correct because that fn manages
its own with_shard/with_shard_db visits — called at the top of the arm, never
nested in a slice borrow), then persists each write to ITS AOF/WAL via the
same `wal_append_and_fanout` path normal cross-shard writes use, with ONE
shared backpressure budget for the whole body.
- PUBLISH fan-out is deferred back to the originating connection (returned in
`TxnExecReply.exec_publishes`) so it keeps the normal originator-side scatter
path and placeholder patching.
- Durability: under appendfsync=always the originator issues one
`fsync_barrier(owner)` after the reply (H1-BARRIER parity — the owner enqueued
its appends before replying, so the barrier's ordered AppendSync covers them);
owner-side append backpressure surfaces `AOF_APPEND_LOST_ERR`.
Both sharded handlers (handler_sharded, handler_monoio) replace the Phase-A
`SingleShard(other)` CROSSSLOT rejection with this route. Genuinely multi-shard
bodies (`CrossShard`) stay rejected — a shared-nothing engine can't commit
across shards atomically without distributed commit.
Red/green: tests/sharded_multi_exec_routing.rs —
1. a hash-tagged multi-key MULTI/EXEC succeeds (never CROSSSLOT) from EVERY
connection and its writes are visible from a fresh reader (40 tags × all 4
owner shards under random accept placement),
2. routed transactional writes survive kill-9 + restart (persisted on the
OWNER shard's AOF),
3. a 20-key untagged span is still rejected with CROSSSLOT.
The Phase-A locality suite still passes (single-key bodies now route+succeed;
the success⇒visible / crossslot⇒unwritten invariant is unchanged). fmt + clippy
(default and tokio,jemalloc) clean; 40 transaction + 37 dispatch lib tests green;
the 64-byte ShardMessage size assertion still holds.
Follow-up: WATCH in sharded mode remains unimplemented (returns ERR unknown
command — loud/safe, no CAS).
author: Tin Dang
Qodo reviews are paused for this user.Troubleshooting steps vary by plan Learn more → On a Teams plan? Using GitHub Enterprise Server, GitLab Self-Managed, or Bitbucket Data Center? |
|
Warning Review limit reached
Next review available in: 38 minutes Enable usage-based reviews in Billing to review now. Otherwise, wait until the next included review is available. How can I continue?After more reviews become available, a review can be triggered using the 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 configurationConfiguration used: defaults Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (1)
📝 WalkthroughWalkthroughThis PR adds sharded MULTI/EXEC locality analysis, routes single-owner-shard transaction bodies to their owning shard, persists transactional writes to AOF with fsync handling, rejects cross-shard bodies with CROSSSLOT, and updates the transaction execution path, supporting message flow, and regression coverage. ChangesSharded MULTI/EXEC routing and AOF durability
Estimated code review effort: 4 (Complex) | ~60 minutes Possibly related issues
Possibly related PRs
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches🧪 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 |
![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 (6)
tests/sharded_multi_exec_routing.rs (2)
47-57: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winSame
Moonleak-on-panic concern astests/sharded_multi_exec_durability.rs.No
Dropimpl; a panic beforekill9()(e.g. anywhere in the loops at lines 225-259 or 281-298) leaks the child process anddir. See the proposed fix intests/sharded_multi_exec_durability.rs(lines 54-65).🤖 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 `@tests/sharded_multi_exec_routing.rs` around lines 47 - 57, The Moon helper still relies on an explicit kill9() call, so a panic in the sharded routing tests can leave the child process and temp dir behind. Add a Drop implementation for Moon, following the same pattern used in sharded_multi_exec_durability, and have it cleanly kill and wait on the Child so cleanup happens automatically even if the test panics. Keep kill9() only as an explicit teardown helper if still needed.
1-355: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winTriplicated RESP-client/Moon-spawn boilerplate across the three new test files.
free_port,moon_binary,Moon,Reply, andClient(send/fill/read_line/read_exact_bytes/read_reply/cmd) are duplicated almost verbatim acrosstests/sharded_multi_exec_durability.rs,tests/sharded_multi_exec_locality.rs, and this file. A sharedtests/common/moon_client.rs(or similar) module would cut ~150-200 duplicated lines per file and avoid the three copies drifting (as already happened with theDrop/leak fix above, present in one file but not the other two).🤖 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 `@tests/sharded_multi_exec_routing.rs` around lines 1 - 355, The RESP client and Moon process helper setup is duplicated across the sharded MULTI/EXEC test files, so factor the shared test utilities into a common module and reuse them here. Move the repeated helpers like free_port, moon_binary, Moon, Reply, and Client (including send/fill/read_line/read_exact_bytes/read_reply/cmd) into a shared tests/common/moon_client.rs or equivalent, then update this test file to import and use those symbols so the three test files stay in sync.src/server/conn/handler_sharded/write.rs (2)
565-566: 🚀 Performance & Scalability | 🔵 Trivial | 💤 Low valueSame clone-before-move as the monoio counterpart.
to_vec()+clear()can be replaced withstd::mem::take(&mut conn.command_queue)to avoid cloning the queued frames.As per coding guidelines,
**src/**/*.rs**: Avoid hot-path allocations in command dispatch... no ... clone() ... in those paths.🤖 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 565 - 566, The command dispatch path in write handling is cloning queued frames by calling to_vec() and then clearing command_queue, which adds unnecessary hot-path allocation. Update the logic in the write handler around conn.command_queue to move the queue out with std::mem::take instead of cloning, keeping the same ownership flow for commands while avoiding the extra copy.Source: Coding guidelines
520-676: 📐 Maintainability & Code Quality | 🔵 Trivial | 🏗️ Heavy liftConsider extracting the shared
try_handle_multi_execlogic.This function is now duplicated almost verbatim between
handler_sharded/write.rsandhandler_monoio/write.rs(locality classification, owner routing, AOF persistence, tracking invalidation — all identical). The missing-invalidation bug flagged above already had to be raised in both copies; a shared helper inserver::conn::shared(which already hostsanalyze_txn_locality/persist_txn_aof) parameterized over the differing connection/context types would remove this drift risk.🤖 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 520 - 676, The `try_handle_multi_exec` flow is duplicated in both sharded and monoio write handlers, which makes the locality routing, AOF persistence, and tracking invalidation paths drift-prone. Move the shared EXEC/MULTI handling into a helper under `server::conn::shared` alongside `analyze_txn_locality` and `persist_txn_aof`, and parameterize it over the connection/context types so both `handler_sharded::write::try_handle_multi_exec` and the monoio equivalent call the same implementation.src/server/conn/handler_monoio/write.rs (1)
615-616: 🚀 Performance & Scalability | 🔵 Trivial | 💤 Low valueAvoid cloning the whole queue before the owner hop.
to_vec()clones every queuedFrameright beforecommandsis moved intoexecute_txn_on_owner. Since it's immediately followed byconn.command_queue.clear(),std::mem::take(&mut conn.command_queue)gets the sameVec<Frame>with no clone and no separateclear()call.As per coding guidelines,
**src/**/*.rs**: Avoid hot-path allocations in command dispatch... no ... clone() ... in those paths; prefer preallocated buffers, SmallVec, itoa, write!, or borrowing.♻️ Proposed fix
- let commands: Vec<Frame> = conn.command_queue.to_vec(); - conn.command_queue.clear(); + let commands: Vec<Frame> = std::mem::take(&mut conn.command_queue);🤖 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 615 - 616, The command dispatch path is cloning the entire queue with command_queue.to_vec() and then clearing it, which adds an avoidable hot-path allocation before execute_txn_on_owner receives commands. In write.rs, update the code around conn.command_queue handling to move the Vec<Frame> out of conn.command_queue instead of cloning it, and remove the separate clear call; use the existing command_queue and execute_txn_on_owner flow to preserve behavior while avoiding the extra copy.Source: Coding guidelines
tests/sharded_multi_exec_durability.rs (1)
54-65: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick win
Moonleaks the child process/dir on assertion failure.
kill9requires explicit, successful completion of the test body to run. Anyassert_eq!panic between spawn andm1.kill9()(e.g. lines 232-258) skips cleanup, leaving an orphanedmoonprocess bound toportand the temp dir on disk — every test failure compounds this.tests/sharded_multi_exec_locality.rs'sMoonalready implementsDropfor this reason; mirroring it here (and intests/sharded_multi_exec_routing.rs) makes cleanup panic-safe.🧹 Proposed fix
struct Moon { child: Child, port: u16, } +impl Drop for Moon { + fn drop(&mut self) { + let _ = self.child.kill(); + let _ = self.child.wait(); + } +} + impl Moon { fn kill9(mut self) { // Hard kill: exercise crash recovery, not graceful shutdown flush. let _ = self.child.kill(); let _ = self.child.wait(); } }🤖 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 `@tests/sharded_multi_exec_durability.rs` around lines 54 - 65, `Moon` cleanup is not panic-safe, so a failed assertion can leave the child process and temp dir behind. Add a `Drop` implementation for `Moon` that performs the same hard-kill cleanup as `kill9`, and keep `kill9` as the explicit early-exit path if needed. Make sure the new `Drop` covers the `child` field so spawned `moon` processes are always reaped even when the test aborts; mirror the existing `Moon` pattern used elsewhere in the sharded multi-exec tests.
🤖 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_monoio/write.rs`:
- Around line 626-654: Routed owner-shard EXEC currently returns early from the
Some(r) success path in write.rs, so the existing CLIENT TRACKING invalidation
block after the fallthrough path is skipped for remote executions. Update the
EXEC flow around execute_txn_on_owner and the Some(r) arm to preserve tracking
invalidation for routed writes by capturing the command/key data before moving
commands into execute_txn_on_owner and invoking invalidate_after_write when
crate::tracking::tracking_active() is true, even on the owner-shard success
path.
In `@src/server/conn/handler_sharded/write.rs`:
- Around line 577-604: The Some(r) path in write handling bypasses the existing
CLIENT TRACKING invalidation logic, so routed owner-shard transactions can
return before tracked keys are invalidated. Update the `handler_sharded::write`
flow around `Some(r)` so it preserves or invokes the same tracking-invalidation
handling already used in `handler_monoio::write`, ensuring the post-write
invalidation block still runs for routed writes before the response is
finalized.
---
Nitpick comments:
In `@src/server/conn/handler_monoio/write.rs`:
- Around line 615-616: The command dispatch path is cloning the entire queue
with command_queue.to_vec() and then clearing it, which adds an avoidable
hot-path allocation before execute_txn_on_owner receives commands. In write.rs,
update the code around conn.command_queue handling to move the Vec<Frame> out of
conn.command_queue instead of cloning it, and remove the separate clear call;
use the existing command_queue and execute_txn_on_owner flow to preserve
behavior while avoiding the extra copy.
In `@src/server/conn/handler_sharded/write.rs`:
- Around line 565-566: The command dispatch path in write handling is cloning
queued frames by calling to_vec() and then clearing command_queue, which adds
unnecessary hot-path allocation. Update the logic in the write handler around
conn.command_queue to move the queue out with std::mem::take instead of cloning,
keeping the same ownership flow for commands while avoiding the extra copy.
- Around line 520-676: The `try_handle_multi_exec` flow is duplicated in both
sharded and monoio write handlers, which makes the locality routing, AOF
persistence, and tracking invalidation paths drift-prone. Move the shared
EXEC/MULTI handling into a helper under `server::conn::shared` alongside
`analyze_txn_locality` and `persist_txn_aof`, and parameterize it over the
connection/context types so both `handler_sharded::write::try_handle_multi_exec`
and the monoio equivalent call the same implementation.
In `@tests/sharded_multi_exec_durability.rs`:
- Around line 54-65: `Moon` cleanup is not panic-safe, so a failed assertion can
leave the child process and temp dir behind. Add a `Drop` implementation for
`Moon` that performs the same hard-kill cleanup as `kill9`, and keep `kill9` as
the explicit early-exit path if needed. Make sure the new `Drop` covers the
`child` field so spawned `moon` processes are always reaped even when the test
aborts; mirror the existing `Moon` pattern used elsewhere in the sharded
multi-exec tests.
In `@tests/sharded_multi_exec_routing.rs`:
- Around line 47-57: The Moon helper still relies on an explicit kill9() call,
so a panic in the sharded routing tests can leave the child process and temp dir
behind. Add a Drop implementation for Moon, following the same pattern used in
sharded_multi_exec_durability, and have it cleanly kill and wait on the Child so
cleanup happens automatically even if the test panics. Keep kill9() only as an
explicit teardown helper if still needed.
- Around line 1-355: The RESP client and Moon process helper setup is duplicated
across the sharded MULTI/EXEC test files, so factor the shared test utilities
into a common module and reuse them here. Move the repeated helpers like
free_port, moon_binary, Moon, Reply, and Client (including
send/fill/read_line/read_exact_bytes/read_reply/cmd) into a shared
tests/common/moon_client.rs or equivalent, then update this test file to import
and use those symbols so the three test files stay in sync.
🪄 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: 9e6041de-dc98-42a1-adc4-9fc8edb54afc
📒 Files selected for processing (12)
CHANGELOG.mdsrc/server/conn/handler_monoio/mod.rssrc/server/conn/handler_monoio/write.rssrc/server/conn/handler_sharded/mod.rssrc/server/conn/handler_sharded/write.rssrc/server/conn/shared.rssrc/shard/coordinator.rssrc/shard/dispatch.rssrc/shard/spsc_handler.rstests/sharded_multi_exec_durability.rstests/sharded_multi_exec_locality.rstests/sharded_multi_exec_routing.rs
… follow-up) The Phase B routed-EXEC path returned as soon as it pushed the owner's result, BEFORE reaching the CLIENT TRACKING invalidation block that the local EXEC path runs — so a transaction routed to a remote owner shard never invalidated cached readers of the keys it wrote (CodeRabbit Major finding on PR). The tracking table is process-global and invalidation is issued originating-side (the owner does not invalidate), so the fix mirrors the local EXEC block on the originating shard: after the routed reply is confirmed durable, iterate the body and call `invalidate_after_write` for each successful write. A copy of the body is captured only when `tracking_active()` (the routed reply carries results, not the command frames), so the hot path pays nothing when no client is tracking. Applied to both the sharded and monoio handlers. Red/green: tests/client_tracking_invalidation.rs gains `routed_multi_exec_invalidates_tracking_client`. A single warmed writer connection (fixed accept shard) commits 12 distinct single-key bodies via MULTI/EXEC, most of which route to a remote owner; a fresh tracking reader must receive the RESP3 invalidate push. Delivery to an idle cross-shard reader is ~75% per attempt in this suite (a documented 4-shard async-push timing property that hits plain SET and MULTI/EXEC equally — verified 9/12 vs 9/12), so the test asserts a MAJORITY deliver: the fix yields ~9/12, the pre-fix routed-skip yields 0–2/12. Verified RED against a neutered routed path (0/12), GREEN with the fix (3× green). fmt + clippy (default and tokio,jemalloc) clean. author: Tin Dang
Contributor
Author
|
Addressed in dc50077. Finding (Major): routed EXEC bypassed CLIENT TRACKING invalidation — confirmed and fixed. The Phase B routed path returned right after pushing the owner's result, before the local EXEC path's tracking-invalidation block. Since the tracking table is process-global and invalidation is issued originating-side (the owner doesn't invalidate), the fix mirrors the local block on the originating shard: after the routed reply is confirmed durable, it iterates the body and calls Regression test |
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 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 `@tests/client_tracking_invalidation.rs`:
- Around line 339-346: The test in client_tracking_invalidation is only checking
for INVALIDATE delivery after txw.cmd("EXEC"), so failures in routed execution
can be misreported as an invalidation regression. Add an assertion around the
EXEC path using txw and its buffer state after txw.clear() to verify EXEC
succeeded before waiting on INVALIDATE, so this test clearly distinguishes EXEC
failure from invalidation delivery issues.
🪄 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: 4615c48a-a727-4edb-b89b-c12977756e4c
📒 Files selected for processing (3)
src/server/conn/handler_monoio/write.rssrc/server/conn/handler_sharded/write.rstests/client_tracking_invalidation.rs
🚧 Files skipped from review as they are similar to previous changes (2)
- src/server/conn/handler_sharded/write.rs
- src/server/conn/handler_monoio/write.rs
`routed_multi_exec_invalidates_tracking_client` only waited on an INVALIDATE push after `EXEC`, so a routed-execution failure (e.g. the owner shard being unavailable → CROSSSLOT) would silently under-count `delivered` and misreport a routing regression as an invalidation-delivery regression. A routed single-key EXEC that committed returns `*1\r\n+OK\r\n`; assert that reply is present before the invalidation wait so the two failure modes are cleanly distinguished in the test output. Test-only change; the assertion passes with the fix in place (EXEC commits and the majority of invalidations still deliver). Addresses CodeRabbit inline review comment on PR #247. author: Tin Dang
Contributor
Author
|
Addressed in 771c715: the routed |
7 tasks
pilotspacex-byte
added a commit
that referenced
this pull request
Jul 13, 2026
…n executor + replicate (kernel M3 stage 3, task #52) (#300) * fix(persistence): route GRAPH.* through the graph store inside MULTI/EXEC (kernel M3 stage 3, task #52) A committed cross-store transaction (SET + GRAPH.ADDNODE in one MULTI/EXEC) survived kill-9 on the KV leg (PR #247's persist_txn_aof) but silently lost the graph leg. Root cause: execute_transaction_sharded (src/server/conn/shared.rs), the executor shared by the sharded and monoio connection handlers for EXEC, had no GRAPH.* branch at all -- a queued GRAPH.ADDNODE fell through to the generic KV dispatch() table (which only knows keyspace commands) and errored "unknown command", an error MULTI/EXEC's per-command tolerance swallowed silently. The mutation never even applied in memory, let alone reached durable storage. Fixed by giving the txn executor a GRAPH.* branch that mirrors the single-command path (try_handle_graph_command): dispatches writes/reads against ShardSlice::graph_store, and flushes every command's drained wal-v3 records via wal_append after the whole transaction body runs (same per-shard append-order contract as the live path). Both the KV AOF leg and the graph wal-v3 leg are independently proven durable by the crash-matrix scenario's two separate sync-waits; ordering between the two log families is otherwise unconstrained by design. Verified 3x consecutive GREEN at shards=1 and shards=4. The sibling atomicity claim (a transaction queued but killed before EXEC applies nothing) was already GREEN and is unaffected -- nothing in this path executes before EXEC runs. Crash-matrix cells cross_plane_prod_s1_txn_isolated_committed and cross_plane_prod_s4_txn_isolated_committed (tests/crash_matrix_cross_plane/) flip from red_guard-gated RED to default-GREEN regression tripwires. Full 42-cell suite green-only run passes (33->35 GREEN by default, 9->7 RED); fmt/clippy clean on both default and runtime-tokio+jemalloc feature sets; relevant lib test suites (persistence::, graph::, command::graph, txn, shared::) all green. Files: - src/server/conn/shared.rs: GRAPH.* branch in execute_transaction_sharded + post-loop wal_append flush - tests/crash_matrix_cross_plane/tests_prod.rs: remove red_guard from the two committed-txn cells, document the fix - tests/crash_matrix_cross_plane.rs: update RED/GREEN cell inventory doc - CHANGELOG.md: Unreleased entry Refs: task #52, kernel M3 stage 3 author: Tin Dang * fix(replication): close master/replica divergence in MULTI/EXEC graph leg (task #52 review round 2, P1) The durability fix in dde3240 gave execute_transaction_sharded a GRAPH.* branch that wal_append'd the drained graph records itself, from inside the function. That function has no ConnectionContext/replication access, so the graph leg of a committed cross-store transaction applied and persisted on the master but never reached a replica -- a NEW master/replica divergence the durability fix introduced (pre-fix the in-txn GRAPH.* command applied nowhere at all, so there was no divergence to have). execute_transaction_sharded now RETURNS the collected graph records (db, bytes) instead of appending them internally, bound to the db each command executed in (same per-entry-db rule PR #282 established for the KV AOF entries -- a queued SELECT redirects only the commands after it). Callers now take over replication + durability, matching the live single-command graph path's contract exactly: - handler_monoio/write.rs (EXEC block): replicates each graph record via record_local_write_db, gated `ctx.num_shards == 1 && replication_fanout_active(ctx)` -- the same scope the live try_handle_graph_command path uses (graph replication is single-shard only; multi-shard graph replication rides the R2 broadcast redesign) -- in the same synchronous stretch as the mutation, THEN wal_append, same replicate-then-append order as the live path. - handler_sharded/write.rs (tokio caller): wal_append only. Replication is monoio-only by design; this caller has no replication plane to fan out to. - shard/spsc_handler.rs (TxnExecute cross-shard hop arm): wal_append only. This arm only runs when the accepting connection's shard differs from the txn's owner shard, which by construction requires num_shards > 1 -- exactly the regime where graph replication is already out of scope, so no replication call is needed here either. New regression test tests/replication_graph.rs::replica_syncs_multi_exec_graph_leg: shards=1 master+replica, MULTI/SET+GRAPH.ADDNODE/EXEC on the master, asserts both the KV and graph legs land on the replica over one live stream (stream-health check: zero reconnects, same convention as this file's other tests). Confirmed the test is not vacuous by disabling just the new record_local_write_db call and re-running: RED (graph leg missing on replica, KV leg present) -- then restored the fix and confirmed GREEN 3x consecutive. Re-ran full gate set after this change: - crash_matrix_cross_plane's 4 txn cells (committed x2 shard counts, atomicity x2 shard counts): GREEN 3x consecutive, unaffected. - Full 42-cell crash-matrix default (green-only) run: 41/42, one unrelated pre-existing flake (cross_plane_prod_s4_mq_isolated, MQ DLQ kill-9 timing) confirmed 3/3 green in isolation -- not touched by this change. - fmt --check: clean. - clippy --release -- -D warnings: clean. - clippy --no-default-features --features runtime-tokio,jemalloc -- -D warnings: clean. - cargo build --release --no-default-features --features runtime-tokio,jemalloc: clean. - lib tests (persistence::, graph::, command::graph, txn, shared::): all green. Files: - src/server/conn/shared.rs: execute_transaction_sharded returns graph records instead of appending them; per-entry db binding - src/server/conn/handler_monoio/write.rs: EXEC block replicates + wal_appends the graph records - src/server/conn/handler_sharded/write.rs: EXEC block wal_appends the graph records (no replication leg) - src/shard/spsc_handler.rs: TxnExecute arm wal_appends the graph records - tests/replication_graph.rs: new replica_syncs_multi_exec_graph_leg test + send_txn helper - CHANGELOG.md: Unreleased entry Refs: task #52, kernel M3 stage 3, follow-up to dde3240 author: Tin Dang * fix(routing): fold GRAPH.* name into MULTI/EXEC shard-locality classification (task #52 review round 3, CodeRabbit P1) CodeRabbit review on PR #300: analyze_txn_locality (the pre-EXEC classifier deciding whether a queued body runs locally, hops to a remote owner shard, or is rejected CROSSSLOT) only scans KV keys via command_keys. GRAPH.* commands declare no keys in command metadata (first_key==last_key==0), so a graph-bearing MULTI body was misclassified -- Keyless for a graph-only body, or by its KV keys alone for a mixed body -- ignoring the graph name's REAL owner shard entirely. The standalone (non-MULTI) GRAPH.* path routes by graph_to_shard(name, num_shards), which is IDENTICAL hashing to key_to_shard (same hash-tag-aware xxh64). At num_shards > 1 a transaction whose true graph owner differed from the classified owner would durably apply the GRAPH.* write to the WRONG shard's graph_store -- invisible to subsequent normally-routed single-command reads. Verified at source before coding: the s4 crash cells (dde3240/9f44816e) never caught this because scenarios::txn_isolated_committed's KV key and graph name deliberately share one {txniso} hash tag (co-location by construction, see tests/crash_matrix_cross_plane/planes.rs's graph_name helper), so they always agree on owner regardless of the bug -- not scatter reads, not luck, just a test fixture that happened to sidestep the exact case the finding describes. Fix: fold the graph name into analyze_txn_locality's existing `visit` closure -- the SAME one KV keys and the SORT/GEORADIUS STORE-dest already use. A body whose KV keys and graph name disagree on owner becomes CrossShard (rejected CROSSSLOT, same posture as any other genuine cross-shard body -- the REJECT posture, chosen over a hop per review guidance). A graph-only body becomes SingleShard(graph_owner), which routes through the SAME whole-body-atomic owner hop (execute_txn_on_owner / ShardMessage::TxnExecute) a KV-only remote-owner body already uses -- no NEW hop mechanism is introduced, so the "body runs on ONE local shard slice" invariant holds and atomicity is preserved. GRAPH.LIST is excluded (no name argument; scatter-gather read, not owned by one shard). New tests in tests/sharded_multi_exec_locality.rs: - graph_leg_cross_shard_rejected: a KV key and graph name deterministically computed (not guessed, not random) to hash to different shards at shards=4 -- asserts CROSSSLOT and that neither leg applied. - graph_only_txn_routes_to_true_owner: a graph-only MULTI body commits and is visible via a FRESH, normally-routed connection across 12 distinct graph names -- proves owner-shard placement, not "whatever shard the writing connection happened to land on" (SO_REUSEPORT placement is not client-controllable, so this doesn't need to force which shard the writer lands on to prove correctness). Confirmed RED without the analyze_txn_locality fix (git stash the fix, rebuild, rerun): graph_leg_cross_shard_rejected instead applied the KV leg and errored "ERR graph not found" on the misrouted GRAPH.ADDNODE (proving it silently landed on the wrong shard's graph_store); graph_only_txn_routes _to_true_owner read back 0 rows on 1 of 12 graph names. Restored the fix, confirmed GREEN. Re-ran full gate set: - New locality tests: GREEN. Pre-existing KV-only locality tests (no_silent_divergence_across_shards, multi_shard_span_rejected, single_shard_unaffected) and sharded_multi_exec_routing.rs: unaffected, GREEN. - crash_matrix_cross_plane's 4 txn cells (committed x2 shard counts, atomicity x2 shard counts): GREEN 3x consecutive. - Full 42-cell crash-matrix default (green-only) run: 42/42 GREEN. - replication_graph.rs::replica_syncs_multi_exec_graph_leg: GREEN 3x consecutive (unaffected -- same-hashtag scenario, no locality change in its path). - lib tests (server::conn::shared::, incl. pre-existing txn_locality_tests module): 15/15 GREEN. - fmt --check: clean. - clippy --release -- -D warnings: clean. - clippy --no-default-features --features runtime-tokio,jemalloc -- -D warnings: clean. - cargo build --release --no-default-features --features runtime-tokio,jemalloc: clean. Files: - src/server/conn/shared.rs: analyze_txn_locality folds GRAPH.* name into the existing key-locality visit closure - tests/sharded_multi_exec_locality.rs: graph_leg_cross_shard_rejected + graph_only_txn_routes_to_true_owner + graph_row_count/ pick_kv_and_graph_on_different_shards helpers - CHANGELOG.md: Unreleased entry Refs: task #52, kernel M3 stage 3, PR #300 review round 3 (CodeRabbit) author: Tin Dang * fix(test): gate graph locality tests on the graph feature tests/sharded_multi_exec_locality.rs imports graph_to_shard, which is cfg(feature = "graph") — the tokio CI matrix builds with --no-default-features --features runtime-tokio,jemalloc (no graph) and failed with E0432. Gate the import, the two graph helpers, and the two graph tests on the same feature. author: Tin Dang --------- Co-authored-by: Tin Dang <tindang.ht97@gmail.com>
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
2 participants
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
Hardens sharded
MULTI/EXECin three commits (safety → durability → routing).execute_transaction_shardedruns the queued body on the connection's accept shard (random via SO_REUSEPORT), but keys route per-key to their owning shard — a mismatch that caused silent corruption, silent data loss, and forced hash-tagged transactions to fail from most connections.What changed
Phase A — reject instead of misplace (
b69fa001)analyze_txn_localityclassifies a queued body (Keyless / SingleShard / CrossShard) via command-metadata key specs +key_to_shard. Foreign-owned bodies are rejected withCROSSSLOTinstead of being silently written to the wrong shard's table. Restores: EXEC-success ⇒ visible to other connections; CROSSSLOT ⇒ nothing written.Durability — persist the writes (
62330dc3)The sharded executor logged nothing to the AOF, so every transactional write was lost on restart under
appendonly=yesat every shard count (incl.--shards 1under monoio). It now returns serialized AOF bytes per successful write (mirroring the single-shard tokioexecute_transaction); a sharedpersist_txn_aofappends them via group-commit + onefsync_barrierunderappendfsync=always. Barrier failure →AOF_FSYNC_ERR+ suppressed PUBLISH fan-out (no false success).try_handle_multi_execis nowasync.Phase B — route to the owner (
82afb94b)New boxed
ShardMessage::TxnExecute(within the 64-byte cache-line cap) +coordinator::execute_txn_on_ownerhop. A single-owner-shard body is executed atomically on the owner's slice and persisted to the owner's AOF viawal_append_and_fanout; PUBLISH fan-out is deferred back to the originator; always-mode durability via an originator-issuedfsync_barrier(owner)(H1-BARRIER parity). Genuinely cross-shard bodies stayCROSSSLOT— a shared-nothing engine can't commit across shards atomically.Testing (red/green)
tests/sharded_multi_exec_durability.rs(1) andtests/sharded_multi_exec_routing.rs(3: routing-from-any-connection + visible, routed writes survive kill-9 + restart, span still CROSSSLOT); Phase-Asharded_multi_exec_locality.rs(3) still green.3952default lib tests + tokio txn/dispatch subsets green;fmt+clippy(default andtokio,jemalloc) clean; 64-byteShardMessagesize assertion holds.Notes / follow-ups
ERR unknown command, no CAS).execute_transaction_sharded, not introduced here.Summary by CodeRabbit
MULTI/EXECcan now be routed to the owning shard, allowing the transaction to succeed from any connection when all queued keys map to one shard.MULTI/EXECdata survives restarts.CROSSSLOT(no partial/incorrect application).