fix(persistence): MULTI/EXEC graph leg — route GRAPH.* through the txn executor + replicate (kernel M3 stage 3, task #52)

CHANGELOG.md

@@ -6,6 +6,59 @@ 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
+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`,

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.

src/server/conn/handler_sharded/write.rs

@@ -666,14 +666,29 @@ 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,
                 conn.selected_db,
                 &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.

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<crate::shard::shared_databases::ShardDatabases>,
     _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<u8>)>) {
     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<u8>)> = Vec::new();
+    #[cfg(not(feature = "graph"))]
+    let graph_records: Vec<(usize, Vec<u8>)> = 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.

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