diff --git a/CHANGELOG.md b/CHANGELOG.md index 6e7e5bcb5..7cbd2f284 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,100 @@ 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 +`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 +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/handler_monoio/write.rs b/src/server/conn/handler_monoio/write.rs index 1f1356ac5..091882afc 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 b1fba363c..5bb87f18d 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 4b9d8a50e..d936ab32a 100644 --- a/src/server/conn/shared.rs +++ b/src/server/conn/shared.rs @@ -203,10 +203,21 @@ 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, @@ -214,7 +225,7 @@ pub(crate) fn execute_transaction_sharded( 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()); @@ -225,6 +236,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, 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_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) { @@ -245,6 +271,51 @@ 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 (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)); + 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_records.extend(records.into_iter().map(|r| (entry_db, r))); + } + 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,7 +412,7 @@ pub(crate) fn execute_transaction_sharded( results.push(response); } - (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. @@ -755,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/src/shard/spsc_handler.rs b/src/shard/spsc_handler.rs index 9d2a5d6ab..af7cef03e 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/crash_matrix_cross_plane.rs b/tests/crash_matrix_cross_plane.rs index 175c32070..b6786fede 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 14bf8e6a6..f5ae5f23e 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); } diff --git a/tests/replication_graph.rs b/tests/replication_graph.rs index 7a28fe738..fc0c1ac62 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}" + ); +} diff --git a/tests/sharded_multi_exec_locality.rs b/tests/sharded_multi_exec_locality.rs index 94b123118..461869978 100644 --- a/tests/sharded_multi_exec_locality.rs +++ b/tests/sharded_multi_exec_locality.rs @@ -25,6 +25,10 @@ use std::net::TcpStream; use std::process::{Child, Command, Stdio}; use std::time::{Duration, Instant}; +#[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") { return Some(std::path::PathBuf::from(p)); @@ -324,3 +328,160 @@ 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. +#[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) { + 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. +#[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}"); + 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`. +#[cfg(feature = "graph")] +#[cfg(feature = "graph")] +#[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. +#[cfg(feature = "graph")] +#[cfg(feature = "graph")] +#[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)" + ); + } +}