feat(control): stage cross-shard graph edges inside transactions

A GRAPH INSERT/DELETE EDGE whose endpoints home to distinct vShards was
previously rejected inside an explicit transaction block with
CrossShardInExplicitTransaction, since only single-home edges could
stage into the per-transaction overlay. An edge is reachable from
either endpoint (forward row on vsrc, reverse row on vdst), and each
Data-Plane core merges only its own overlay on read, so a dual-home
edge now stages the same GraphOp into both endpoint overlays via the
new stage_edge_dual_home helper, giving read-your-own-writes from
either direction; ROLLBACK fans DropTxnOverlay to both touched
vShards. Single-home edges keep staging once as before.

Adds cluster-test coverage on a single-node-calvin server proving the
cross-shard edge is accepted, observed from both endpoints in-txn, and
torn down from both on rollback.

Author: farhan-syah

nodedb-cluster-tests/tests/single_node_calvin_graph_txn.rs

@@ -0,0 +1,185 @@
+// SPDX-License-Identifier: BUSL-1.1
+
+//! Cross-shard (dual-home) graph edge STAGING inside an explicit transaction on
+//! a single-node-calvin server.
+//!
+//! Before this coverage, a `GRAPH INSERT EDGE` whose endpoints home to DISTINCT
+//! vShards, issued inside a `BEGIN..COMMIT` block, was REJECTED with
+//! `CrossShardInExplicitTransaction`: only single-home edges could stage. A
+//! cross-shard edge lives on BOTH endpoints (forward row on `from_key(src)`,
+//! reverse row on `from_key(dst)`), and a read scatters to the home vShard of
+//! the node it starts from — so each Data-Plane core merges only its OWN
+//! per-transaction overlay. To make read-your-own-writes work from either
+//! endpoint the edge must be staged into BOTH overlays.
+//!
+//! `dual_home_edge_stages_both_overlays_and_rollback_tears_down` proves the new
+//! behavior end to end on a `single_node_calvin` server (where `calvin_available`
+//! is true so a cross-shard edge is genuinely dual-home, not forced single-home):
+//!
+//! 1. `BEGIN`; `GRAPH INSERT EDGE` across two distinct vShards is ACCEPTED (no
+//!    `CrossShardInExplicitTransaction`) and staged.
+//! 2. In-txn RYOW from BOTH endpoints: `GRAPH NEIGHBORS ... DIRECTION out` from
+//!    the SRC endpoint sees the edge (forward overlay on `vsrc`), and
+//!    `GRAPH NEIGHBORS ... DIRECTION in` from the DST endpoint ALSO sees it
+//!    (reverse overlay on `vdst`) — proving both overlays hold the staged edge.
+//! 3. `ROLLBACK` fans `DropTxnOverlay` to both touched vShards, so a post-rollback
+//!    read from each endpoint sees the edge gone.
+
+mod common;
+
+use std::time::Duration;
+
+use nodedb_types::id::VShardId;
+
+use common::cluster_harness::{TestClusterNode, wait_for};
+
+/// The lone sequencer voter's observed leader id, or `0` if none known yet.
+/// Mirrors the wait used by the sibling `single_node_calvin` suite so the
+/// Calvin stack is up before the transaction runs.
+fn sequencer_leader(node: &TestClusterNode) -> u64 {
+    let Some(status_fn) = node.shared.raft_status_fn.get() else {
+        return 0;
+    };
+    status_fn()
+        .into_iter()
+        .find(|g| g.group_id == nodedb_cluster::calvin::SEQUENCER_GROUP_ID)
+        .map(|g| g.leader_id)
+        .unwrap_or(0)
+}
+
+/// A `(src, dst)` pair of graph node keys whose home vShards differ, so an edge
+/// between them is genuinely cross-shard. Deterministic: `VShardId::from_key` is
+/// a pure function of the key bytes, and it is how `insert_edge` homes each
+/// endpoint. Same key-picking approach as the sibling `single_node_calvin` suite.
+fn distinct_vshard_node_keys() -> (String, String) {
+    let dst = "sncgtx_hub".to_string();
+    let vdst = VShardId::from_key(dst.as_bytes()).as_u32();
+    for i in 0u32..4096 {
+        let src = format!("sncgtx_src_{i}");
+        if VShardId::from_key(src.as_bytes()).as_u32() != vdst {
+            return (src, dst);
+        }
+    }
+    panic!("could not find a node key on a distinct vShard from the hub in 4096 tries");
+}
+
+/// Run `GRAPH NEIGHBORS OF '<node>' LABEL '<label>' DIRECTION <dir>` and return
+/// the neighbor node ids from the single-row `[{"label":..,"node":..}, ...]`
+/// JSON payload in the first column.
+async fn neighbors(
+    client: &tokio_postgres::Client,
+    node: &str,
+    label: &str,
+    dir: &str,
+) -> Vec<String> {
+    let sql = format!("GRAPH NEIGHBORS OF '{node}' LABEL '{label}' DIRECTION {dir}");
+    let msgs = client.simple_query(&sql).await.expect("GRAPH NEIGHBORS");
+    let mut out = Vec::new();
+    for msg in msgs {
+        if let tokio_postgres::SimpleQueryMessage::Row(row) = msg {
+            let raw = row.get(0).unwrap_or("");
+            let parsed: Vec<serde_json::Value> = serde_json::from_str(raw).unwrap_or_default();
+            for entry in parsed {
+                if let Some(n) = entry.get("node").and_then(|v| v.as_str()) {
+                    out.push(n.to_string());
+                }
+            }
+        }
+    }
+    out
+}
+
+/// A cross-shard edge inserted inside a transaction stages into BOTH endpoint
+/// overlays (RYOW from either direction) and ROLLBACK tears both down.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn dual_home_edge_stages_both_overlays_and_rollback_tears_down() {
+    // 4 Data-Plane cores so distinct vShards land on distinct cores — a genuine
+    // cross-core (dual-home) edge.
+    let node = TestClusterNode::spawn_single_node_calvin(4)
+        .await
+        .expect("spawn standalone single-node-calvin server");
+
+    // The lone sequencer voter self-elects; wait for it so `calvin_available` is
+    // genuinely operational (a cross-shard edge is dual-home, not forced
+    // single-home).
+    wait_for(
+        "single-node sequencer leader elected",
+        Duration::from_secs(10),
+        Duration::from_millis(50),
+        || sequencer_leader(&node) == node.node_id,
+    )
+    .await;
+    assert!(
+        node.shared.cluster_transport.is_some() && node.shared.sequencer_inbox.get().is_some(),
+        "single-node calvin must wire calvin_available (cluster_transport + sequencer_inbox)"
+    );
+
+    node.client
+        .simple_query("CREATE COLLECTION sncgtx_graph")
+        .await
+        .expect("CREATE COLLECTION sncgtx_graph");
+    wait_for(
+        "collection visible on the node",
+        Duration::from_secs(10),
+        Duration::from_millis(50),
+        || node.cached_collection_count() >= 1,
+    )
+    .await;
+
+    let (src, dst) = distinct_vshard_node_keys();
+    let label = "l";
+
+    node.client.simple_query("BEGIN").await.expect("BEGIN");
+
+    // Cross-shard edge insert INSIDE the transaction. Pre-change this returned
+    // `CrossShardInExplicitTransaction`; now it stages into both endpoint
+    // overlays.
+    node.client
+        .simple_query(&format!(
+            "GRAPH INSERT EDGE IN 'sncgtx_graph' FROM '{src}' TO '{dst}' TYPE '{label}'"
+        ))
+        .await
+        .expect(
+            "cross-shard edge insert inside a transaction must STAGE (dual-home), \
+             not reject with CrossShardInExplicitTransaction",
+        );
+
+    // RYOW from the SRC endpoint: forward overlay on `vsrc` holds the edge.
+    let out_src = neighbors(&node.client, &src, label, "out").await;
+    assert!(
+        out_src.contains(&dst),
+        "in-tx forward NEIGHBORS from src '{src}' must observe the staged edge to \
+         '{dst}' (vsrc overlay), got: {out_src:?}"
+    );
+
+    // RYOW from the DST endpoint: reverse overlay on `vdst` ALSO holds the edge.
+    // This is the dual-home proof — a single-home stage would leave `vdst` empty.
+    let in_dst = neighbors(&node.client, &dst, label, "in").await;
+    assert!(
+        in_dst.contains(&src),
+        "in-tx reverse NEIGHBORS from dst '{dst}' must observe the staged edge from \
+         '{src}' (vdst overlay) — proves BOTH overlays hold it, got: {in_dst:?}"
+    );
+
+    node.client
+        .simple_query("ROLLBACK")
+        .await
+        .expect("ROLLBACK");
+
+    // Post-rollback: the edge is gone from BOTH endpoints (DropTxnOverlay fanned
+    // to both touched vShards).
+    let out_src_after = neighbors(&node.client, &src, label, "out").await;
+    assert!(
+        !out_src_after.contains(&dst),
+        "after ROLLBACK the staged edge must be gone from src '{src}' (vsrc \
+         overlay dropped), got: {out_src_after:?}"
+    );
+    let in_dst_after = neighbors(&node.client, &dst, label, "in").await;
+    assert!(
+        !in_dst_after.contains(&src),
+        "after ROLLBACK the staged edge must be gone from dst '{dst}' (vdst \
+         overlay dropped), got: {in_dst_after:?}"
+    );
+
+    node.shutdown().await;
+}

nodedb/src/control/server/shared/ddl/neutral/graph_ops/edge.rs

@@ -150,32 +150,26 @@ pub async fn insert_edge(
         state.cluster_transport.is_some() && state.sequencer_inbox.get().is_some();
     let single_home = vsrc == vdst || !calvin_available;
 
-    // Inside an explicit transaction block a single-home edge insert stages into
-    // the per-transaction `GraphTxnOverlay` through the neutral gate instead of
+    // Inside an explicit transaction block an edge insert stages into the
+    // per-transaction `GraphTxnOverlay` through the neutral gate instead of
     // applying durably now: an in-transaction `MATCH` / `GRAPH NEIGHBORS` then
     // observes the edge as present (read-your-own-writes), COMMIT replays the
-    // buffered `EdgePut` via the single-shard WAL + `TransactionBatch` path, and
-    // ROLLBACK discards the overlay. This is the write-side complement to the
-    // `delete_edge` staging above; both reuse `stage_edge_write_in_txn`.
-    // Autocommit is untouched.
+    // buffered `EdgePut`, and ROLLBACK discards the overlay. A cross-shard
+    // (dual-home) edge stages into BOTH endpoint overlays via
+    // `stage_edge_dual_home` so RYOW works from either endpoint; a single-home
+    // edge stages once. This is the write-side complement to the `delete_edge`
+    // staging below. Autocommit is untouched.
     if txn_ctx.sessions.transaction_state(txn_ctx.addr) == TransactionState::InBlock {
-        if !single_home {
-            // A cross-shard (dual-home) edge insert inside an explicit
-            // transaction needs the cross-shard-commit machinery to stage both
-            // homes atomically across the outer transaction boundary — out of
-            // scope for single-home staging. Reject with the same signal a
-            // cross-shard predicate write inside a transaction produces.
-            return Err(ddl_err(
-                "XX000",
-                crate::Error::CrossShardInExplicitTransaction.to_string(),
-            ));
-        }
-        super::edge_stage::stage_edge_write_in_txn(
+        super::edge_stage::stage_edge_dual_home(
             state,
             tenant_id,
             database_id,
-            vsrc,
-            PhysicalPlan::Graph(edge_put),
+            EdgeHomes {
+                vsrc,
+                vdst,
+                single_home,
+            },
+            edge_put,
             txn_ctx,
         )
         .await?;
@@ -239,6 +233,19 @@ pub struct EdgeRef {
     pub label: String,
 }
 
+/// The home vShard(s) an edge resolves to. An edge is reachable from BOTH
+/// endpoints (forward from `src`, reverse from `dst`), so a cross-shard edge
+/// (`!single_home`) has two distinct homes: `vsrc` holds the forward row and
+/// `vdst` holds the reverse row. `single_home` is true when both endpoints share
+/// one vShard, or when Calvin is unavailable (single-node) so one write covers
+/// both. Bundled so [`stage_edge_dual_home`](super::edge_stage::stage_edge_dual_home)
+/// stays within the argument budget.
+pub struct EdgeHomes {
+    pub vsrc: VShardId,
+    pub vdst: VShardId,
+    pub single_home: bool,
+}
+
 /// `GRAPH DELETE EDGE IN '<collection>' FROM '<src>' TO '<dst>' TYPE '<label>'`
 pub async fn delete_edge(
     state: &SharedState,
@@ -316,30 +323,23 @@ pub async fn delete_edge(
         state.cluster_transport.is_some() && state.sequencer_inbox.get().is_some();
     let single_home = vsrc == vdst || !calvin_available;
 
-    // Inside an explicit transaction block a single-home edge delete stages into
-    // the per-transaction `GraphTxnOverlay` through the neutral gate instead of
+    // Inside an explicit transaction block an edge delete stages into the
+    // per-transaction `GraphTxnOverlay` through the neutral gate instead of
     // applying durably now: an in-transaction `MATCH` / `GRAPH NEIGHBORS` then
     // observes the edge as removed (read-your-own-writes), COMMIT replays the
-    // buffered `EdgeDelete` via the single-shard WAL + `TransactionBatch` path,
-    // and ROLLBACK discards the overlay. Autocommit is untouched.
+    // buffered `EdgeDelete`, and ROLLBACK discards the overlay. Autocommit is
+    // untouched.
     if txn_ctx.sessions.transaction_state(txn_ctx.addr) == TransactionState::InBlock {
-        if !single_home {
-            // A cross-shard (dual-home) edge delete inside an explicit
-            // transaction needs the cross-shard-commit machinery to stage both
-            // homes atomically across the outer transaction boundary — out of
-            // scope for single-home staging. Reject with the same signal a
-            // cross-shard predicate write inside a transaction produces.
-            return Err(ddl_err(
-                "XX000",
-                crate::Error::CrossShardInExplicitTransaction.to_string(),
-            ));
-        }
-        super::edge_stage::stage_edge_write_in_txn(
+        super::edge_stage::stage_edge_dual_home(
             state,
             tenant_id,
             database_id,
-            vsrc,
-            PhysicalPlan::Graph(edge_delete),
+            EdgeHomes {
+                vsrc,
+                vdst,
+                single_home,
+            },
+            edge_delete,
             txn_ctx,
         )
         .await?;