From dde324042871c5fdca3ad18cbe4928fab9cad433 Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Mon, 13 Jul 2026 10:54:27 +0700 Subject: [PATCH 1/4] 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 --- CHANGELOG.md | 24 +++++++ src/server/conn/shared.rs | 75 +++++++++++++++++++- tests/crash_matrix_cross_plane.rs | 58 +++++++++------ tests/crash_matrix_cross_plane/tests_prod.rs | 33 ++++----- 4 files changed, 146 insertions(+), 44 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 6e7e5bcb..51e20cd3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,30 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Fixed — cross-store MULTI/EXEC graph leg durability (kernel M3 stage 3 / task #52) + +A committed cross-store transaction (`MULTI`; `SET k v`; `GRAPH.ADDNODE g +Label id 1`; `EXEC`) survived kill-9 on the KV leg (PR #247's +`persist_txn_aof`) but silently lost the graph leg — worse than a missed +durability leg, the `GRAPH.ADDNODE` never even applied in memory. Root +cause: `execute_transaction_sharded` (`src/server/conn/shared.rs`), the +executor both the sharded and monoio connection handlers use 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 `ERR unknown command`, an error MULTI/EXEC's per-command tolerance +swallowed without surfacing it to the client. 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). Verified 3× consecutive +GREEN at shards=1 and shards=4; the sibling atomicity claim (a transaction +queued but killed before `EXEC` applies nothing) was unaffected and stays +GREEN. Crash-matrix cells `cross_plane_prod_s1_txn_isolated_committed` / +`cross_plane_prod_s4_txn_isolated_committed` (`tests/crash_matrix_cross_plane/`) +flip from `red_guard`-gated RED to default-GREEN tripwires (42 cells: 33→35 +GREEN by default, 9→7 RED). + ### Added — unified per-shard floor register + min-across-planes WAL recycle (kernel M3 stage 2 / K2) `ShardControlFile` (`src/persistence/control.rs`) gains `graph_floor_lsn`, diff --git a/src/server/conn/shared.rs b/src/server/conn/shared.rs index 4b9d8a50..0c294749 100644 --- a/src/server/conn/shared.rs +++ b/src/server/conn/shared.rs @@ -209,7 +209,7 @@ pub(crate) fn execute_transaction( /// persistence, so MULTI/EXEC writes were silently lost on restart). pub(crate) fn execute_transaction_sharded( shard_databases: &std::sync::Arc, - _shard_id: usize, + shard_id: usize, command_queue: &[Frame], selected_db: usize, cached_clock: &CachedClock, @@ -225,6 +225,21 @@ pub(crate) fn execute_transaction_sharded( // switches dbs. let mut aof_entries: Vec<(usize, Bytes)> = Vec::new(); let mut selected = selected_db; + // task #52: graph-leg WAL records produced by GRAPH.* commands queued + // inside this MULTI/EXEC body. The single-command path + // (`try_handle_graph_command`) drains+appends these to wal-v3 per + // command; the txn executor previously had NO graph branch at all, so a + // queued GRAPH.* command fell through to the generic KV `dispatch()` + // table and errored as "unknown command" -- never applied to the graph + // store, never durable. Collected here and flushed to wal-v3 AFTER the + // whole body runs (below), same append-order-per-shard contract as the + // live path, mirrored 1:1 into a Vec instead of an immediate per-command + // append only because this function has no `.await` point to interleave + // with the KV AOF group-commit -- ordering between the two log families + // is otherwise unconstrained (each is its own durability leg, proven + // independently by the crash-matrix scenario's two separate sync-waits). + #[cfg(feature = "graph")] + let mut graph_wal_records: Vec> = Vec::new(); for cmd_frame in command_queue { let (cmd, cmd_args) = match extract_command(cmd_frame) { @@ -245,6 +260,48 @@ pub(crate) fn execute_transaction_sharded( continue; } + // task #52: GRAPH.* commands live in a separate per-shard store + // (`ShardSlice::graph_store`), not the KV `Database` this loop + // otherwise dispatches against -- route them to the graph engine + // exactly like the single-command path (`try_handle_graph_command`) + // does, and stash the drained WAL records for the post-loop flush. + #[cfg(feature = "graph")] + if cmd.len() > 6 && cmd[..6].eq_ignore_ascii_case(b"GRAPH.") { + let is_write = crate::command::graph::is_graph_write_cmd(cmd) + || (cmd.eq_ignore_ascii_case(b"GRAPH.QUERY") + && crate::command::graph::is_cypher_write_query(cmd_args)); + let (response, records) = crate::shard::slice::with_shard(|s| { + if is_write { + let resp = if cmd.eq_ignore_ascii_case(b"GRAPH.QUERY") { + crate::command::graph::graph_query_or_write(&mut s.graph_store, cmd_args).0 + } else { + crate::command::graph::dispatch_graph_write( + &mut s.graph_store, + cmd, + cmd_args, + ) + }; + let records = s.graph_store.drain_wal(); + (resp, records) + } else { + let resp = crate::command::graph::dispatch_graph_read( + &s.graph_store, + cmd, + cmd_args, + None, + ); + (resp, Vec::new()) + } + }); + // Parity with the KV branch below: an errored write must not + // reach the WAL. + if !records.is_empty() && !matches!(&response, Frame::Error(_)) { + graph_wal_records.extend(records); + } + results.push(response); + continue; + } + // Serialize write commands for AOF *before* dispatch (matches the // single-shard `execute_transaction` path). Without this the sharded // MULTI/EXEC path logged nothing, so every transactional write was @@ -341,6 +398,22 @@ pub(crate) fn execute_transaction_sharded( results.push(response); } + // task #52: flush every graph-leg WAL record collected above through the + // same K1 typed channel (`wal_append`) the live single-command graph + // path uses -- this is what makes the graph leg of a committed + // cross-store MULTI/EXEC durable across kill-9. Order preserved + // (per-shard wal-v3 append order = per-command execution order). + #[cfg(feature = "graph")] + for record in graph_wal_records { + shard_databases.wal_append( + shard_id, + crate::persistence::wal_v3::record::WalRecordType::Command, + bytes::Bytes::from(record), + ); + } + #[cfg(not(feature = "graph"))] + let _ = shard_id; + (Frame::Array(results.into()), aof_entries) } diff --git a/tests/crash_matrix_cross_plane.rs b/tests/crash_matrix_cross_plane.rs index 175c3207..b6786fed 100644 --- a/tests/crash_matrix_cross_plane.rs +++ b/tests/crash_matrix_cross_plane.rs @@ -77,7 +77,9 @@ //! `cross_plane_prod_s1_graph_drop_survives_repeated_checkpoints` / //! `cross_plane_prod_s4_graph_drop_survives_repeated_checkpoints`, both //! GREEN by default (RED-first verified against the pre-fix code before -//! the fix landed). **42 cells total: 33 GREEN by default, 9 RED.** +//! the fix landed). Kernel M3 stage 3 (task #52) then root-caused and fixed +//! former RED cell 3 (cross-store TXN graph leg not durable), flipping its +//! 2 cells GREEN. **42 cells total: 35 GREEN by default, 7 RED.** //! //! RED cells are gated by `harness::red_guard` — NOT by //! `#[ignore = "RED: ..."]` alone, because this suite's own execution @@ -118,24 +120,34 @@ //! WAL records DO replay in order); it stays an ordinary `#[ignore]` (needs //! a release binary), not RED. //! -//! **RED cell 3 — cross-store TXN graph leg not durable, task #52 (NEW, 2 -//! cells):** `cross_plane_prod_s1_txn_isolated_committed`, +//! **FIXED (was RED cell 3) — cross-store TXN graph leg not durable, task +//! #52 (2 cells, now GREEN):** `cross_plane_prod_s1_txn_isolated_committed`, //! `cross_plane_prod_s4_txn_isolated_committed`. A committed cross-store -//! transaction (`SET` + `GRAPH.ADDNODE` in one `MULTI`/`EXEC`) survives -//! kill-9 on the KV leg (PR #247's `persist_txn_aof`) but the graph leg is -//! lost. Reviewed and confirmed NOT a harness artifact; confirmed -//! deterministic 3× consecutive at both shard counts after a task #52 -//! hardening pass (own crash cycle per round, `--checkpoint-timeout 3600`, -//! kill immediately after the sync-wait — see -//! `scenarios::txn_isolated_committed`'s doc for the full analysis and -//! minimal repro). New P1 finding for kernel M3. The SIBLING atomicity -//! claim (`cross_plane_prod_s1_txn_isolated_atomicity` / -//! `_s4_txn_isolated_atomicity` — a transaction queued but killed BEFORE -//! `EXEC` must apply NOTHING) is a genuinely different, unrelated claim and -//! is GREEN on these two configs (review round 3, P1: it used to share this -//! same `red_guard`, incorrectly hiding a working regression tripwire by -//! default — split into its own ungated `#[test]` per config). It stays RED -//! on `legacy_yes_s1` only, folded into RED cell 1 above (same +//! transaction (`SET` + `GRAPH.ADDNODE` in one `MULTI`/`EXEC`) survived +//! kill-9 on the KV leg (PR #247's `persist_txn_aof`) but lost the graph +//! leg. Root cause: `execute_transaction_sharded` +//! (`src/server/conn/shared.rs`) had no `GRAPH.*` branch — the txn body +//! dispatcher only knew the generic KV command table, so a queued +//! `GRAPH.ADDNODE` fell through to it and errored "unknown command", +//! never reaching the graph store (let alone its WAL) at all. This is a +//! stronger bug than a missed durability leg: the mutation itself never +//! applied in memory before the kill, MULTI/EXEC's per-command error +//! tolerance just swallowed the error silently. Fixed by giving the txn +//! executor a `GRAPH.*` branch that mirrors the single-command path +//! (`try_handle_graph_command`) — dispatches to `ShardSlice::graph_store`, +//! collects the drained wal-v3 records per command, and flushes them via +//! `wal_append` after the whole transaction body runs (same per-shard +//! append-order contract as the live path). Verified 3× consecutive GREEN +//! at both shard counts (kernel M3 stage 3 hardening pass); both cells now +//! run ungated (green-only default suite). The SIBLING atomicity claim +//! (`cross_plane_prod_s1_txn_isolated_atomicity` / `_s4_txn_isolated_atomicity` +//! — a transaction queued but killed BEFORE `EXEC` must apply NOTHING) was +//! always a genuinely different, unrelated claim and was already GREEN on +//! these two configs (review round 3, P1: it used to share this same +//! `red_guard`, incorrectly hiding a working regression tripwire by default +//! — split into its own ungated `#[test]` per config, unaffected by this +//! fix since nothing in that path executes before `EXEC`). It stays RED on +//! `legacy_yes_s1` only, folded into RED cell 1 above (same //! legacy-graph-reconstruction root cause — confirmed on the VM, not //! assumed: `GRAPH.CREATE` itself never survives a restart there, so the //! atomicity check's own graph-leg assertion errors "graph not found" @@ -174,7 +186,7 @@ //! (pre-fix baseline: prod_s1 hit at iteration 11/20, prod_s4 at iteration //! 7/20) — these two tests now run ungated (green-only default suite). //! -//! **Every other cell (33/42) is GREEN** — including, notably, +//! **Every other cell (35/42) is GREEN** — including, notably, //! `kv_isolated`/`kv_spilled_isolated`/`vector_isolated` on ALL configs (KV //! and vector-HSET durability via the AOF; `kv_spilled_isolated`'s filler //! sizing was hardened in review round 3 — see @@ -183,9 +195,11 @@ //! both `prod_s1`/`prod_s4` (structural writes + GraphTemporal //! `TEMPORAL.INVALIDATE`, incl. a VALID_AT positive control — checkpoint-backed //! mode does NOT share legacy mode's gap), `mq_isolated` everywhere (PR -//! #291's effect-record fix holds), `txn_isolated_atomicity` on -//! `prod_s1`/`prod_s4` (queuing without `EXEC` then killing applies nothing -//! — verified 3× consecutive), `mixed_all_planes_synced`/`mid_pass_c` on +//! #291's effect-record fix holds), `txn_isolated_committed`/`txn_isolated_atomicity` +//! on `prod_s1`/`prod_s4` (committed KV+graph legs both survive kill-9 — +//! kernel M3 stage 3 / task #52 fix above; queuing without `EXEC` then +//! killing applies nothing — verified 3× consecutive each), +//! `mixed_all_planes_synced`/`mid_pass_c` on //! both production shard counts, `mixed_all_planes_mid_checkpoint` on both //! production shard counts (kernel M3 stage 2 / K2 fix above — now //! unconditionally GREEN, not just at the default single-iteration diff --git a/tests/crash_matrix_cross_plane/tests_prod.rs b/tests/crash_matrix_cross_plane/tests_prod.rs index 14bf8e6a..f5ae5f23 100644 --- a/tests/crash_matrix_cross_plane/tests_prod.rs +++ b/tests/crash_matrix_cross_plane/tests_prod.rs @@ -4,23 +4,9 @@ //! (flat function names, per the brief's explicit suggestion — avoids the //! `g1_` prefix collision `crash_recovery_graph_durability.rs` already has). -use crate::harness::{self, Config}; +use crate::harness::Config; use crate::scenarios; -const TXN_GRAPH_LEG_RED_REASON: &str = "cross-store TXN graph leg not durable, task #52. A \ - committed cross-store transaction (SET+GRAPH.ADDNODE) survives kill-9 \ - on the KV leg (PR #247's persist_txn_aof) but the graph leg is lost — \ - reviewed and confirmed not a harness artifact (property-value query, \ - no VALID_AT, both legs' WAL flush confirmed synced before the kill), \ - and confirmed deterministic 3x consecutive at both shards=1 and \ - shards=4 after the task #52 hardening pass (own crash cycle per \ - round, --checkpoint-timeout 3600, kill immediately post-sync-wait). \ - Minimal repro in scenarios::txn_isolated_committed's doc comment. \ - P1 finding for kernel M3 — tracked separately, not this stage's job to \ - fix. Gates ONLY the committed-transaction half (review round 3, P1): \ - the atomicity half (`cross_plane_*_txn_isolated_atomicity`) is a \ - genuinely different, unrelated, GREEN claim and runs ungated."; - /// FIXED (kernel M3 stage 2 / K2, task #53). Root cause: `save_graph_store` /// (`src/graph/recovery.rs`) wrote the reference/floor /// (`graph_metadata.json` via `store.save_metadata`) BEFORE the payload it @@ -79,12 +65,18 @@ fn cross_plane_prod_s1_mq_isolated() { scenarios::mq_isolated(&Config::PROD_S1); } +/// FIXED (kernel M3 stage 3 / task #52). Root cause: `execute_transaction_sharded` +/// (`src/server/conn/shared.rs`) had no GRAPH.* branch at all — a queued +/// GRAPH.ADDNODE fell through to the generic KV `dispatch()` table (which +/// doesn't know graph commands), erroring as "unknown command" and never +/// reaching the graph store, let alone its WAL. Fixed by routing GRAPH.* +/// commands queued inside MULTI/EXEC to the graph engine (mirroring the +/// single-command path in `try_handle_graph_command`) and flushing the +/// drained wal-v3 records after the transaction body runs. Verified 3x +/// consecutive GREEN at both shards=1 and shards=4. #[test] #[ignore] // Requires built release binary; run explicitly. fn cross_plane_prod_s1_txn_isolated_committed() { - if !harness::red_guard(TXN_GRAPH_LEG_RED_REASON) { - return; - } scenarios::txn_isolated_committed(&Config::PROD_S1); } @@ -160,12 +152,11 @@ fn cross_plane_prod_s4_mq_isolated() { scenarios::mq_isolated(&Config::PROD_S4); } +/// FIXED (kernel M3 stage 3 / task #52). See +/// `cross_plane_prod_s1_txn_isolated_committed`'s doc for the root cause. #[test] #[ignore] // Requires built release binary; run explicitly. fn cross_plane_prod_s4_txn_isolated_committed() { - if !harness::red_guard(TXN_GRAPH_LEG_RED_REASON) { - return; - } scenarios::txn_isolated_committed(&Config::PROD_S4); } From 9f44816ef746f9ddab146205f0132e61d0c2392c Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Mon, 13 Jul 2026 11:29:51 +0700 Subject: [PATCH 2/4] fix(replication): close master/replica divergence in MULTI/EXEC graph leg (task #52 review round 2, P1) The durability fix in dde32404 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 dde32404 author: Tin Dang --- CHANGELOG.md | 29 +++++ src/server/conn/handler_monoio/write.rs | 33 +++++- src/server/conn/handler_sharded/write.rs | 17 ++- src/server/conn/shared.rs | 74 ++++++------- src/shard/spsc_handler.rs | 36 ++++-- tests/replication_graph.rs | 133 +++++++++++++++++++++++ 6 files changed, 274 insertions(+), 48 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 51e20cd3..10e210c6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,35 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Fixed — cross-store MULTI/EXEC graph leg replication (kernel M3 stage 3 / task #52, review round 2, P1) + +Follow-up to the durability fix below: `execute_transaction_sharded`'s new +`GRAPH.*` branch originally `wal_append`ed the graph-leg records itself, +*inside* the function — but that function has no `ConnectionContext` / +replication access, so the graph leg 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.*` applied nowhere, so there was +no divergence to have). `execute_transaction_sharded` now returns the +collected graph records (bound to the db each command executed in, same +per-entry-db rule as the AOF entries — a queued `SELECT` redirects later +commands) instead of appending them itself. The monoio EXEC caller +(`handler_monoio/write.rs`) now replicates each record via +`record_local_write_db` — gated `ctx.num_shards == 1 && +replication_fanout_active(ctx)`, matching the live single-command graph +path's scope exactly (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`s, same +replicate-then-append order as the live path. The tokio/sharded caller and +the cross-shard `TxnExecute` shard-message arm (`src/shard/spsc_handler.rs`) +only `wal_append` (no replication plane in scope there; the `TxnExecute` hop +only fires at `num_shards > 1`, where graph replication is out of scope by +design regardless). New regression test +`replica_syncs_multi_exec_graph_leg` (`tests/replication_graph.rs`): +shards=1 master+replica, `MULTI`/`SET`+`GRAPH.ADDNODE`/`EXEC` on the master, +asserts both legs land on the replica over one live stream (no +reconnect/full-resync masking) — confirmed RED (graph leg missing) with +just the new replication call disabled, GREEN 3× consecutive with the fix. + ### Fixed — cross-store MULTI/EXEC graph leg durability (kernel M3 stage 3 / task #52) A committed cross-store transaction (`MULTI`; `SET k v`; `GRAPH.ADDNODE g diff --git a/src/server/conn/handler_monoio/write.rs b/src/server/conn/handler_monoio/write.rs index 1f1356ac..091882af 100644 --- a/src/server/conn/handler_monoio/write.rs +++ b/src/server/conn/handler_monoio/write.rs @@ -841,7 +841,7 @@ pub(super) async fn try_handle_multi_exec( _ => {} } } - let (result, aof_entries) = execute_transaction_sharded( + let (result, aof_entries, graph_records) = execute_transaction_sharded( &ctx.shard_databases, ctx.shard_id, &conn.command_queue, @@ -875,6 +875,37 @@ pub(super) async fn try_handle_multi_exec( super::ft::record_local_write_db(ctx, *entry_db, bytes.clone()); } } + // task #52 review round 2 (P1): the graph-leg records need the + // SAME replicate-then-append treatment as the live single-command + // graph path (`try_handle_graph_command`, ~line 1123 below) — + // `execute_transaction_sharded` only collects them (it has no + // `ConnectionContext`/replication access), so the fan-out and the + // wal_append both happen HERE, in the same synchronous stretch as + // the txn body that just ran. Graph replication is single-shard + // scope only (multi-shard graph replication rides the R2 + // broadcast redesign), matching the live path's `num_shards == 1` + // gate exactly — this is NOT relaxed to the KV leg's + // `repl_active` alone. + #[cfg(feature = "graph")] + { + let graph_repl_active = ctx.num_shards == 1 && repl_active; + for (entry_db, record) in &graph_records { + if graph_repl_active { + super::ft::record_local_write_db( + ctx, + *entry_db, + bytes::Bytes::from(record.clone()), + ); + } + ctx.shard_databases.wal_append( + ctx.shard_id, + crate::persistence::wal_v3::record::WalRecordType::Command, + bytes::Bytes::from(record.clone()), + ); + } + } + #[cfg(not(feature = "graph"))] + let _ = graph_records; // DURABILITY: append every successful write in the body to THIS // shard's AOF via the same group-commit path as normal writes, then // issue ONE fsync barrier under appendfsync=always before acking. diff --git a/src/server/conn/handler_sharded/write.rs b/src/server/conn/handler_sharded/write.rs index b1fba363..5bb87f18 100644 --- a/src/server/conn/handler_sharded/write.rs +++ b/src/server/conn/handler_sharded/write.rs @@ -666,7 +666,7 @@ pub(super) async fn try_handle_multi_exec( _ => {} } } - let (result, aof_entries) = execute_transaction_sharded( + let (result, aof_entries, graph_records) = execute_transaction_sharded( &ctx.shard_databases, ctx.shard_id, &conn.command_queue, @@ -674,6 +674,21 @@ pub(super) async fn try_handle_multi_exec( &ctx.cached_clock, exec_publishes, ); + // task #52: flush the graph-leg wal-v3 records collected by the + // txn executor. Replication is monoio-only by design (see + // `handler_monoio::write`'s EXEC handling) — this tokio/sharded + // caller has no replication plane to fan out to, so it only + // needs the local durability leg. + #[cfg(feature = "graph")] + for (_entry_db, record) in graph_records { + ctx.shard_databases.wal_append( + ctx.shard_id, + crate::persistence::wal_v3::record::WalRecordType::Command, + bytes::Bytes::from(record), + ); + } + #[cfg(not(feature = "graph"))] + let _ = graph_records; // DURABILITY: append every successful write in the body to THIS // shard's AOF via the same group-commit path as normal writes, then // issue ONE fsync barrier under appendfsync=always before acking. diff --git a/src/server/conn/shared.rs b/src/server/conn/shared.rs index 0c294749..be13c73b 100644 --- a/src/server/conn/shared.rs +++ b/src/server/conn/shared.rs @@ -203,18 +203,29 @@ pub(crate) fn execute_transaction( /// (see `analyze_txn_locality`) — the body runs on the local slice with no /// per-key routing. /// -/// Returns the result Frame (an array of per-command responses) **and** the -/// serialized AOF bytes for each successful write, in order. The caller MUST -/// append those entries to the shard's AOF (previously this path did no -/// persistence, so MULTI/EXEC writes were silently lost on restart). +/// Returns the result Frame (an array of per-command responses), the +/// serialized AOF bytes for each successful write in order, and the +/// graph-leg wal-v3 records (db, bytes) produced by any `GRAPH.*` command in +/// the body. The caller MUST append the AOF entries to the shard's AOF +/// (previously this path did no persistence, so MULTI/EXEC writes were +/// silently lost on restart) and MUST both fan the graph records out to +/// replicas (single-shard, monoio only — see the live single-command +/// contract at `handler_monoio::write::try_handle_graph_command`) and +/// `wal_append` them (task #52 review round 2: appending them INSIDE this +/// function skipped the replication leg the live path always takes, +/// silently diverging a replica from a committed cross-store transaction — +/// this function has no `ConnectionContext`/replication access, so the +/// caller must do both, in the same synchronous stretch as the +/// already-synchronous call to this function, replicate-then-append, same +/// order as the live path). pub(crate) fn execute_transaction_sharded( shard_databases: &std::sync::Arc, - shard_id: usize, + _shard_id: usize, command_queue: &[Frame], selected_db: usize, cached_clock: &CachedClock, exec_publishes: &mut Vec<(usize, Bytes, Bytes)>, -) -> (Frame, Vec<(usize, Bytes)>) { +) -> (Frame, Vec<(usize, Bytes)>, Vec<(usize, Vec)>) { let db_count = shard_databases.db_count(); let mut results = Vec::with_capacity(command_queue.len()); @@ -226,20 +237,20 @@ pub(crate) fn execute_transaction_sharded( let mut aof_entries: Vec<(usize, Bytes)> = Vec::new(); let mut selected = selected_db; // task #52: graph-leg WAL records produced by GRAPH.* commands queued - // inside this MULTI/EXEC body. The single-command path - // (`try_handle_graph_command`) drains+appends these to wal-v3 per - // command; the txn executor previously had NO graph branch at all, so a - // queued GRAPH.* command fell through to the generic KV `dispatch()` - // table and errored as "unknown command" -- never applied to the graph - // store, never durable. Collected here and flushed to wal-v3 AFTER the - // whole body runs (below), same append-order-per-shard contract as the - // live path, mirrored 1:1 into a Vec instead of an immediate per-command - // append only because this function has no `.await` point to interleave - // with the KV AOF group-commit -- ordering between the two log families - // is otherwise unconstrained (each is its own durability leg, proven - // independently by the crash-matrix scenario's two separate sync-waits). + // inside this MULTI/EXEC body, bound to the db THIS command executed in + // (same per-entry-db discipline as `aof_entries` — a queued SELECT + // redirects the commands after it, never itself). The single-command + // path (`try_handle_graph_command`) drains+appends these to wal-v3 (and + // replicates them) per command; the txn executor previously had NO graph + // branch at all, so a queued GRAPH.* command fell through to the generic + // KV `dispatch()` table and errored as "unknown command" -- never + // applied to the graph store, never durable, never replicated. Collected + // here and returned for the CALLER to replicate + flush to wal-v3 (this + // function has no replication access — see the doc comment above). #[cfg(feature = "graph")] - let mut graph_wal_records: Vec> = Vec::new(); + let mut graph_records: Vec<(usize, Vec)> = Vec::new(); + #[cfg(not(feature = "graph"))] + let graph_records: Vec<(usize, Vec)> = Vec::new(); for cmd_frame in command_queue { let (cmd, cmd_args) = match extract_command(cmd_frame) { @@ -264,9 +275,12 @@ pub(crate) fn execute_transaction_sharded( // (`ShardSlice::graph_store`), not the KV `Database` this loop // otherwise dispatches against -- route them to the graph engine // exactly like the single-command path (`try_handle_graph_command`) - // does, and stash the drained WAL records for the post-loop flush. + // does, and stash the drained WAL records (bound to the db THIS + // command executed in, captured before dispatch, same rule as + // `entry_db` below) for the caller to replicate + flush. #[cfg(feature = "graph")] if cmd.len() > 6 && cmd[..6].eq_ignore_ascii_case(b"GRAPH.") { + let entry_db = selected; let is_write = crate::command::graph::is_graph_write_cmd(cmd) || (cmd.eq_ignore_ascii_case(b"GRAPH.QUERY") && crate::command::graph::is_cypher_write_query(cmd_args)); @@ -296,7 +310,7 @@ pub(crate) fn execute_transaction_sharded( // Parity with the KV branch below: an errored write must not // reach the WAL. if !records.is_empty() && !matches!(&response, Frame::Error(_)) { - graph_wal_records.extend(records); + graph_records.extend(records.into_iter().map(|r| (entry_db, r))); } results.push(response); continue; @@ -398,23 +412,7 @@ pub(crate) fn execute_transaction_sharded( results.push(response); } - // task #52: flush every graph-leg WAL record collected above through the - // same K1 typed channel (`wal_append`) the live single-command graph - // path uses -- this is what makes the graph leg of a committed - // cross-store MULTI/EXEC durable across kill-9. Order preserved - // (per-shard wal-v3 append order = per-command execution order). - #[cfg(feature = "graph")] - for record in graph_wal_records { - shard_databases.wal_append( - shard_id, - crate::persistence::wal_v3::record::WalRecordType::Command, - bytes::Bytes::from(record), - ); - } - #[cfg(not(feature = "graph"))] - let _ = shard_id; - - (Frame::Array(results.into()), aof_entries) + (Frame::Array(results.into()), aof_entries, graph_records) } /// Persist the AOF entries of a just-executed sharded MULTI/EXEC body. diff --git a/src/shard/spsc_handler.rs b/src/shard/spsc_handler.rs index 9d2a5d6a..af7cef03 100644 --- a/src/shard/spsc_handler.rs +++ b/src/shard/spsc_handler.rs @@ -2703,14 +2703,34 @@ pub(crate) fn handle_shard_message_shared( reply_tx, } = *payload; let mut exec_publishes: Vec<(usize, bytes::Bytes, bytes::Bytes)> = Vec::new(); - let (result, aof_entries) = crate::server::conn::shared::execute_transaction_sharded( - shard_databases, - shard_id, - &commands, - db_index, - cached_clock, - &mut exec_publishes, - ); + let (result, aof_entries, graph_records) = + crate::server::conn::shared::execute_transaction_sharded( + shard_databases, + shard_id, + &commands, + db_index, + cached_clock, + &mut exec_publishes, + ); + // task #52: this arm is the CROSS-SHARD EXEC hop (the accepting + // connection's shard differs from the owner shard, which by + // construction only happens at num_shards > 1) -- graph + // replication is single-shard scope only (see + // `execute_transaction_sharded`'s doc / the monoio EXEC handler's + // `graph_repl_active = ctx.num_shards == 1 && ...` gate), so it + // never applies here. Only the local wal-v3 durability leg is + // needed, via the same K1 typed channel every other graph write + // path uses. + #[cfg(feature = "graph")] + for (_entry_db, record) in graph_records { + shard_databases.wal_append( + shard_id, + crate::persistence::wal_v3::record::WalRecordType::Command, + bytes::Bytes::from(record), + ); + } + #[cfg(not(feature = "graph"))] + let _ = graph_records; // Persist via the SAME sync path as normal cross-shard writes (the // MultiExecute arm): fire-and-forget append + WAL/replica fan-out, // ONE shared backpressure budget for the whole body so a stall costs diff --git a/tests/replication_graph.rs b/tests/replication_graph.rs index 7a28fe73..fc0c1ac6 100644 --- a/tests/replication_graph.rs +++ b/tests/replication_graph.rs @@ -94,6 +94,40 @@ fn send_resp(addr: &str, parts: &[&str]) -> String { read_quiet(&mut stream) } +/// RESP-encode and send MULTI + every queued command in `cmds` + the commit +/// command as ONE pipelined write on a single connection (transaction state +/// is per-connection, so this can't reuse `send_resp`'s connect-per-call +/// helper). Returns the concatenated raw reply text (QUEUED markers + the +/// commit's array reply), same "eyeball the buffer" convention as +/// `send_resp`/`send_cmd`. +fn send_txn(addr: &str, cmds: &[&[&str]]) -> String { + let Ok(mut stream) = TcpStream::connect(addr) else { + return String::new(); + }; + stream + .set_read_timeout(Some(Duration::from_millis(500))) + .ok(); + let mut out = Vec::new(); + let encode = |parts: &[&str], out: &mut Vec| { + out.extend_from_slice(format!("*{}\r\n", parts.len()).as_bytes()); + for p in parts { + out.extend_from_slice(format!("${}\r\n", p.len()).as_bytes()); + out.extend_from_slice(p.as_bytes()); + out.extend_from_slice(b"\r\n"); + } + }; + encode(&["MULTI"], &mut out); + for cmd in cmds { + encode(cmd, &mut out); + } + let commit = ["E", "X", "E", "C"].concat(); + encode(&[commit.as_str()], &mut out); + if stream.write_all(&out).is_err() { + return String::new(); + } + read_quiet(&mut stream) +} + fn read_quiet(stream: &mut TcpStream) -> String { let mut buf = Vec::new(); let mut chunk = [0u8; 4096]; @@ -376,3 +410,102 @@ fn replica_applies_temporal_invalidate() { send_resp(replica_addr, &visible_query) ); } + +/// REPL-GRAPH-04 (task #52 review round 2, P1): the graph leg of a +/// cross-store `MULTI`/`EXEC` transaction must reach a replica exactly like +/// any other successful local write — `execute_transaction_sharded`'s +/// `GRAPH.*` branch (added for task #52's durability fix) must not skip the +/// replication fan-out that the KV leg (`persist_txn_aof`'s +/// `record_local_write_db` call) already takes, or the master and replica +/// silently diverge on a COMMITTED transaction. Single-shard master scope, +/// matching every other graph-replication test in this file. +#[test] +#[ignore] +fn replica_syncs_multi_exec_graph_leg() { + let master_dir = tempfile::tempdir().unwrap(); + let replica_dir = tempfile::tempdir().unwrap(); + let master_addr = "127.0.0.1:16740"; + let replica_addr = "127.0.0.1:16741"; + + let replica_log = replica_dir.path().join("replica-stderr.log"); + + let _master = Killer(start_moon(16740, master_dir.path().to_str().unwrap())); + assert!( + wait_until(Duration::from_secs(5), || send_cmd(master_addr, "PING") + .starts_with("+PONG")), + "master never became ready" + ); + let _replica = Killer(start_moon_logged( + 16741, + replica_dir.path().to_str().unwrap(), + Some(&replica_log), + )); + assert!( + wait_until(Duration::from_secs(5), || send_cmd(replica_addr, "PING") + .starts_with("+PONG")), + "replica never became ready" + ); + assert!( + send_cmd(replica_addr, "REPLICAOF 127.0.0.1 16740").starts_with("+OK"), + "REPLICAOF failed" + ); + // Let the handshake settle before the first write. + std::thread::sleep(Duration::from_millis(500)); + + assert!(send_cmd(master_addr, "GRAPH.CREATE gtxn").contains("OK")); + + // A single MULTI/EXEC body that writes BOTH a KV key and a graph node — + // the minimal repro from the task #52 crash-matrix scenario + // (`scenarios::txn_isolated_committed`), reused here for the + // replication claim instead of the durability claim. + let reply = send_txn( + master_addr, + &[ + &["SET", "txnkey", "txnval"], + &["GRAPH.ADDNODE", "gtxn", "Person", "name", "carol"], + ], + ); + assert!( + !reply.contains("-ERR") && !reply.contains("unknown command"), + "MULTI/EXEC body errored: {reply}" + ); + + // 1. KV leg replicates (the PR #247 baseline this bug shared a code path + // with — asserted here as a control, not the new claim). + let kv_synced = wait_until(Duration::from_secs(10), || { + send_cmd(replica_addr, "GET txnkey") == "$6\r\ntxnval\r\n" + }); + assert!( + kv_synced, + "replica never saw the txn's KV leg: {}", + send_cmd(replica_addr, "GET txnkey") + ); + + // 2. THE CLAIM: the graph leg — committed in the SAME transaction — + // replicates too. + let graph_synced = wait_until(Duration::from_secs(10), || { + send_resp( + replica_addr, + &["GRAPH.QUERY", "gtxn", "MATCH (n:Person) RETURN n.name"], + ) + .contains("carol") + }); + assert!( + graph_synced, + "replica never saw the txn's graph leg (node 'carol'): {}", + send_resp( + replica_addr, + &["GRAPH.QUERY", "gtxn", "MATCH (n:Person) RETURN n.name"] + ) + ); + + // 3. STREAM-HEALTH: both legs must have arrived on ONE live stream, not + // via a reconnect/full-resync loop masking a dead live producer (same + // check as `replica_syncs_graph_live_stream`). + let log = std::fs::read_to_string(&replica_log).unwrap_or_default(); + let reconnects = log.matches("reconnecting").count(); + assert_eq!( + reconnects, 0, + "replica reconnected {reconnects}x — live stream did not hold:\n{log}" + ); +} From 91b28d662dd7625e0f195c7b0b8d112d6f69e58d Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Mon, 13 Jul 2026 12:29:37 +0700 Subject: [PATCH 3/4] 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 (dde32404/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 --- CHANGELOG.md | 41 +++++++ src/server/conn/shared.rs | 33 ++++++ tests/sharded_multi_exec_locality.rs | 153 +++++++++++++++++++++++++++ 3 files changed, 227 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 10e210c6..7cbd2f28 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,47 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Fixed — cross-shard MULTI/EXEC graph-leg misrouting (kernel M3 stage 3 / task #52, review round 3, P1) + +CodeRabbit finding on PR #300: `analyze_txn_locality` (the pre-EXEC classifier +that decides 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 body was misclassified — +`Keyless` for a graph-only body, or by its KV keys alone for a mixed body — +completely ignoring the graph name's REAL owner shard (the standalone +`GRAPH.*` path routes by `graph_to_shard(name, num_shards)`, identical +hash-tag-aware xxh64 to `key_to_shard`). 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. The task #52 crash cells +never caught this because their KV key and graph name deliberately share one +`{txniso}` hash tag (co-location by construction), so they always agree on +owner regardless of the bug. + +Fixed by folding 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); 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, so the "body runs on ONE local +shard slice" invariant is preserved. + +New tests in `tests/sharded_multi_exec_locality.rs`: +`graph_leg_cross_shard_rejected` (a KV key and graph name deterministically +computed, not guessed, to hash to different shards at shards=4 → `CROSSSLOT`, +neither leg applied) and `graph_only_txn_routes_to_true_owner` (a graph-only +body commits and is visible via a fresh, normally-routed connection — +proving owner placement, not "whatever shard the connection landed on"). +Confirmed RED without the `analyze_txn_locality` fix (both new tests fail: +the rejected-body test instead applies the KV leg and errors "graph not +found" on the misrouted `GRAPH.ADDNODE`; the owner-routing test reads back 0 +rows), GREEN with it, 3× consecutive. `no_silent_divergence_across_shards` / +`multi_shard_span_rejected` / `single_shard_unaffected` (pre-existing KV-only +locality tests) and the task #52 crash-matrix txn cells are unaffected. + ### Fixed — cross-store MULTI/EXEC graph leg replication (kernel M3 stage 3 / task #52, review round 2, P1) Follow-up to the durability fix below: `execute_transaction_sharded`'s new diff --git a/src/server/conn/shared.rs b/src/server/conn/shared.rs index be13c73b..d936ab32 100644 --- a/src/server/conn/shared.rs +++ b/src/server/conn/shared.rs @@ -826,6 +826,39 @@ pub(crate) fn analyze_txn_locality(command_queue: &[Frame], num_shards: usize) - return TxnLocality::CrossShard; } } + // task #52 review round 3 (CodeRabbit, P1): GRAPH.* commands declare + // NO keys in command metadata (first_key==last_key==0), so + // `command_keys` above contributes nothing for them — a graph-only + // body was misclassified `Keyless` (executed on whatever shard the + // connection happened to be pinned to) and a mixed KV+graph body was + // classified by its KV keys alone, ignoring the graph name entirely. + // The STANDALONE `GRAPH.*` path (`try_handle_graph_command`) routes + // by `graph_to_shard(name, num_shards)` — which is IDENTICALLY + // `key_to_shard` (same hash-tag-aware xxh64, see + // `shard::dispatch::graph_to_shard`'s doc) — so treating the graph + // name as just another "key" via the SAME `visit` closure used above + // reuses that exact routing rule: a body whose graph name and KV + // keys disagree on owner becomes `CrossShard` (rejected CROSSSLOT, + // same as the SORT/GEORADIUS STORE-dest case above) instead of + // silently applying the graph write to the wrong shard's store, + // invisible to later normally-routed single-command reads. A + // graph-only body becomes `SingleShard(graph_owner)`, which the + // caller already hops to the owner shard for (`execute_txn_on_owner` + // / `ShardMessage::TxnExecute`, the same whole-body-atomic mechanism + // a KV-only body routes through when queued from a non-owner shard) + // — no NEW hop is introduced, so the "body runs on ONE local slice" + // invariant holds. `GRAPH.LIST` is excluded: it has no name argument + // and is a scatter-gather read across every shard, not owned by one. + if cmd.len() > 6 + && cmd[..6].eq_ignore_ascii_case(b"GRAPH.") + && !cmd.eq_ignore_ascii_case(b"GRAPH.LIST") + { + if let Some(name) = args.first().and_then(super::util::extract_bytes) { + if !visit(&name, &mut owner) { + return TxnLocality::CrossShard; + } + } + } } match owner { None => TxnLocality::Keyless, diff --git a/tests/sharded_multi_exec_locality.rs b/tests/sharded_multi_exec_locality.rs index 94b12311..7dc32a62 100644 --- a/tests/sharded_multi_exec_locality.rs +++ b/tests/sharded_multi_exec_locality.rs @@ -25,6 +25,8 @@ use std::net::TcpStream; use std::process::{Child, Command, Stdio}; use std::time::{Duration, Instant}; +use moon::shard::dispatch::{graph_to_shard, key_to_shard}; + fn moon_binary() -> Option { if let Ok(p) = std::env::var("MOON_BIN") { return Some(std::path::PathBuf::from(p)); @@ -324,3 +326,154 @@ fn single_shard_unaffected() { assert_eq!(r.cmd(&["GET", "s:a"]), Reply::Bulk(Some("1".into()))); assert_eq!(r.cmd(&["GET", "s:b"]), Reply::Bulk(Some("2".into()))); } + +// --------------------------------------------------------------------------- +// GRAPH.* leg locality (task #52 review round 3, CodeRabbit finding, P1) +// --------------------------------------------------------------------------- +// +// `GRAPH.*` commands declare NO keys in command metadata (`first_key == +// last_key == 0`) — `analyze_txn_locality`'s KV-key scan (`command_keys`) +// therefore contributed nothing for them, so a queued body was classified +// purely by its KV keys (or `Keyless` if it had none), ignoring the graph +// name's REAL owner shard entirely. The standalone (non-MULTI) `GRAPH.*` +// path routes by `graph_to_shard(name, num_shards)` — identical hashing to +// `key_to_shard` (same hash-tag-aware xxh64) — so a graph write committed +// inside a misclassified MULTI/EXEC body could silently land on the WRONG +// shard's `graph_store`, invisible to that normally-routed standalone read +// path. Fixed by folding the graph name into `analyze_txn_locality`'s same +// `visit` closure used for KV keys: a body whose KV keys and graph name +// disagree on owner becomes `CrossShard` (rejected `CROSSSLOT`, same +// posture as the SORT/GEORADIUS `STORE`-dest case above); 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, so the "body runs on ONE local shard slice" +// invariant is preserved. + +/// Rows array from a GRAPH.QUERY reply (`[headers, rows, stats]`), returning +/// just the row count — enough to prove presence/absence without depending +/// on cell encoding. +fn graph_row_count(c: &mut Client, graph: &str, cypher: &str) -> usize { + match c.cmd(&["GRAPH.QUERY", graph, cypher]) { + Reply::Array(outer) => match outer.get(1) { + Some(Reply::Array(rows)) => rows.len(), + other => panic!("malformed GRAPH.QUERY reply, rows slot = {other:?}"), + }, + other => panic!("malformed GRAPH.QUERY reply: {other:?}"), + } +} + +/// Deterministically (not randomly) search fixed candidate suffixes for a KV +/// key and a graph name whose owners differ at `num_shards` — reproducible +/// every run since the hash function is pure and the candidate list is +/// fixed, unlike relying on which shard a client connection happens to land +/// on. +fn pick_kv_and_graph_on_different_shards(num_shards: usize) -> (String, String) { + for i in 0..256u32 { + let key = format!("txnglocxs:{i}"); + let key_owner = key_to_shard(key.as_bytes(), num_shards); + for j in 0..256u32 { + let graph = format!("txnglocxsgraph:{j}"); + let graph_owner = graph_to_shard(graph.as_bytes(), num_shards); + if key_owner != graph_owner { + return (key, graph); + } + } + } + panic!("could not find a KV key / graph name pair on different shards at {num_shards} shards"); +} + +/// THE CODERABBIT FINDING, made RED-then-GREEN: a `MULTI` body whose KV key +/// and `GRAPH.*` name are computed (not guessed) to hash to DIFFERENT +/// shards at shards=4 must be rejected `CROSSSLOT`, and neither leg may have +/// applied — same golden invariant as `multi_shard_span_rejected`, extended +/// across the KV/graph boundary. Pre-fix this body was silently +/// misclassified `SingleShard(kv_owner)` (or `Keyless`, ignoring the KV key +/// entirely for a graph-only body) and the `GRAPH.ADDNODE` durably applied +/// to the WRONG shard's `graph_store`. +#[test] +fn graph_leg_cross_shard_rejected() { + let Some(m) = spawn_moon("4") else { return }; + let (key, graph) = pick_kv_and_graph_on_different_shards(4); + + assert_eq!( + Client::connect(m.port).cmd(&["GRAPH.CREATE", &graph]), + Reply::Simple("OK".into()), + "GRAPH.CREATE {graph}" + ); + + let mut c = Client::connect(m.port); + assert_eq!(c.cmd(&["MULTI"]), Reply::Simple("OK".into())); + assert_eq!(c.cmd(&["SET", &key, "v"]), Reply::Simple("QUEUED".into())); + assert_eq!( + c.cmd(&["GRAPH.ADDNODE", &graph, "Label", "id", "1"]), + Reply::Simple("QUEUED".into()) + ); + match c.cmd(&["EXEC"]) { + Reply::Error(e) => assert!( + e.starts_with("CROSSSLOT"), + "KV key and graph name on different shards must be CROSSSLOT, got: {e}" + ), + other => panic!("cross-shard KV+graph body must be rejected, got: {other:?}"), + } + + // Neither leg applied — same "EXEC says OK but the data isn't there" + // golden invariant, checked from a fresh connection routed normally. + let mut r = Client::connect(m.port); + assert_eq!( + r.cmd(&["GET", &key]), + Reply::Bulk(None), + "rejected cross-shard txn must not have written the KV leg" + ); + assert_eq!( + graph_row_count(&mut r, &graph, "MATCH (n:Label {id: 1}) RETURN n.id"), + 0, + "rejected cross-shard txn must not have applied the graph leg" + ); +} + +/// A graph-only `MULTI` body (no KV key at all — the `Keyless` misclassification +/// half of the finding) must still commit to the graph's TRUE owner shard, +/// regardless of which shard the writing connection happened to land on +/// (`SO_REUSEPORT` placement is not client-controllable). Verified by +/// reading back through a FRESH connection, which routes normally by +/// `graph_to_shard` — if the write had landed on the wrong shard (the +/// pre-fix `Keyless` bug: always the connection's OWN current shard), this +/// read would see nothing. +#[test] +fn graph_only_txn_routes_to_true_owner() { + let Some(m) = spawn_moon("4") else { return }; + + for i in 0..12 { + let graph = format!("txnglocowner:{i}"); + assert_eq!( + Client::connect(m.port).cmd(&["GRAPH.CREATE", &graph]), + Reply::Simple("OK".into()), + "GRAPH.CREATE {graph}" + ); + + let mut c = Client::connect(m.port); + assert_eq!(c.cmd(&["MULTI"]), Reply::Simple("OK".into())); + assert_eq!( + c.cmd(&["GRAPH.ADDNODE", &graph, "Label", "id", &i.to_string()]), + Reply::Simple("QUEUED".into()) + ); + match c.cmd(&["EXEC"]) { + Reply::Array(items) => assert_eq!( + items.len(), + 1, + "EXEC array has one reply per queued write for {graph}" + ), + other => panic!("graph-only txn must succeed (owner hop), got: {other:?} for {graph}"), + } + + let mut r = Client::connect(m.port); + let cypher = format!("MATCH (n:Label {{id: {i}}}) RETURN n.id"); + assert_eq!( + graph_row_count(&mut r, &graph, &cypher), + 1, + "committed graph-only txn for {graph} must be visible via the normally-routed read \ + (proves owner-shard placement, not the connection's current shard)" + ); + } +} From 5c9320aeaf9a963551b19c81ade71baf41eab01a Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Mon, 13 Jul 2026 12:52:20 +0700 Subject: [PATCH 4/4] fix(test): gate graph locality tests on the graph feature MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- tests/sharded_multi_exec_locality.rs | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) diff --git a/tests/sharded_multi_exec_locality.rs b/tests/sharded_multi_exec_locality.rs index 7dc32a62..46186997 100644 --- a/tests/sharded_multi_exec_locality.rs +++ b/tests/sharded_multi_exec_locality.rs @@ -25,7 +25,9 @@ use std::net::TcpStream; use std::process::{Child, Command, Stdio}; use std::time::{Duration, Instant}; -use moon::shard::dispatch::{graph_to_shard, key_to_shard}; +#[cfg(feature = "graph")] +use moon::shard::dispatch::graph_to_shard; +use moon::shard::dispatch::key_to_shard; fn moon_binary() -> Option { if let Ok(p) = std::env::var("MOON_BIN") { @@ -353,6 +355,7 @@ fn single_shard_unaffected() { /// Rows array from a GRAPH.QUERY reply (`[headers, rows, stats]`), returning /// just the row count — enough to prove presence/absence without depending /// on cell encoding. +#[cfg(feature = "graph")] fn graph_row_count(c: &mut Client, graph: &str, cypher: &str) -> usize { match c.cmd(&["GRAPH.QUERY", graph, cypher]) { Reply::Array(outer) => match outer.get(1) { @@ -368,6 +371,7 @@ fn graph_row_count(c: &mut Client, graph: &str, cypher: &str) -> usize { /// every run since the hash function is pure and the candidate list is /// fixed, unlike relying on which shard a client connection happens to land /// on. +#[cfg(feature = "graph")] fn pick_kv_and_graph_on_different_shards(num_shards: usize) -> (String, String) { for i in 0..256u32 { let key = format!("txnglocxs:{i}"); @@ -391,6 +395,8 @@ fn pick_kv_and_graph_on_different_shards(num_shards: usize) -> (String, String) /// misclassified `SingleShard(kv_owner)` (or `Keyless`, ignoring the KV key /// entirely for a graph-only body) and the `GRAPH.ADDNODE` durably applied /// to the WRONG shard's `graph_store`. +#[cfg(feature = "graph")] +#[cfg(feature = "graph")] #[test] fn graph_leg_cross_shard_rejected() { let Some(m) = spawn_moon("4") else { return }; @@ -440,6 +446,8 @@ fn graph_leg_cross_shard_rejected() { /// `graph_to_shard` — if the write had landed on the wrong shard (the /// pre-fix `Keyless` bug: always the connection's OWN current shard), this /// read would see nothing. +#[cfg(feature = "graph")] +#[cfg(feature = "graph")] #[test] fn graph_only_txn_routes_to_true_owner() { let Some(m) = spawn_moon("4") else { return };