feat(control): native SQL in-txn read-your-own-writes (U3) + cross-cluster txn_id groundwork

nodedb-cluster-tests/tests/cluster_execute_request.rs

@@ -53,6 +53,7 @@ fn make_kv_put_request(
             collection: collection.into(),
             version: descriptor_version,
         }],
+        txn_id: None,
     }
 }
 
@@ -209,6 +210,7 @@ async fn execute_request_cross_node_dispatch() {
             collection: "cross_node_kv".into(),
             version: 0, // Accept any version (pre-B.1 sentinel bypass)
         }],
+        txn_id: None,
     };
 
     let resp = send_execute_request(sender_transport, target_addr, req).await;

nodedb-cluster-tests/tests/cross_node_ryow_probe.rs

@@ -0,0 +1,145 @@
+// SPDX-License-Identifier: BUSL-1.1
+
+//! DEFERRED PROBE (design unit U4/U5) — CROSS-NODE in-transaction
+//! read-your-own-writes. `#[ignore]`d: cross-node in-txn RYOW is out of scope
+//! for U3, whose validation surface is single-node only (LOCKED design §8:
+//! "single-node-verifiable: U1, U2, U3 (RYOW), U7"; cross-node is "cluster-bound
+//! (stage-2): U4, U5, U6, U8"). Kept in-tree, ignored, as the executable
+//! specification for the U4/U5 cross-shard-transaction work.
+//!
+//! SINGLE-NODE in-txn RYOW works today (U3): an in-block read takes the
+//! NON-streaming dispatch path (`maybe_stream_select` bails on `InBlock`), and
+//! `dispatch_local` threads the session `txn_id` to the local leaseholder so the
+//! Data Plane merges this transaction's staging overlay. The native twin
+//! (`native_sql_ryow_probe.rs`) and the graph twin
+//! (`single_node_calvin_graph_txn.rs`) both cover that and stay GREEN.
+//!
+//! CONFIRMED ROOT CAUSE of the cross-node gap (diagnosed 2026-07, live trace):
+//! the whole per-transaction machinery — staging gate, COMMIT buffer, and the
+//! COMMIT/ROLLBACK overlay flush — is LOCAL-ONLY. On a NON-owner node the pgwire
+//! dispatcher takes the remote-leader forward short-circuit
+//! (`routing/execute.rs`: `should_forward_via_gateway` → `dispatch_tasks_via_gateway`)
+//! and RETURNS before `dispatch_task_loop`, so `route_in_tx_write` never runs.
+//! The in-txn write is therefore shipped to the owner as a PLAIN replicable
+//! write (Raft-proposed = committed at statement time, NOT staged into
+//! `txn_overlays[txn_id]`), and `dispatch_tasks_via_gateway` hardcodes
+//! `QueryContext.txn_id = None` (`routing/gateway_dispatch.rs`) so the in-txn
+//! read reaches the owner with no txn id and cannot merge the overlay. Net: a
+//! cross-node in-txn write bypasses the transaction entirely — it neither
+//! reads-its-own-write NOR honours ROLLBACK atomicity.
+//!
+//! The `ExecuteRequest.txn_id` wire field (+ `WIRE_VERSION` 2→3) and the
+//! globally-unique `TxnId::from_origin` packing landed on this branch are the
+//! necessary GROUNDWORK for the fix, but are NOT sufficient on their own: making
+//! this probe pass requires routing the txn staging / read / COMMIT / ROLLBACK /
+//! overlay-drop through the gateway to the owning node — i.e. the U4/U5
+//! cross-shard-transaction unit.
+//!
+//! This probe drives BEGIN / INSERT / SELECT(own write) / ROLLBACK on EVERY node
+//! of a 3-node cluster against a single-vShard-homed collection. Exactly one node
+//! leads that vShard (local RYOW — the GREEN control); the other two route the
+//! whole transaction to it over QUIC and so exercise the cross-node path, which
+//! MISSES today. Remove `#[ignore]` when U4/U5 lands to turn it back into a gate.
+
+mod common;
+
+use std::time::Duration;
+
+use common::cluster_harness::{TestCluster, wait_for};
+
+/// `n` for `id` read on `client` in its current session. `None` when absent, the
+/// column did not parse as an integer, or the query itself errored (a cross-node
+/// failure is recorded by the caller as a RYOW miss, not a panic, so every node
+/// is exercised).
+async fn n_for(client: &tokio_postgres::Client, id: &str) -> Option<i64> {
+    let msgs = client
+        .simple_query(&format!("SELECT n FROM t WHERE id = '{id}'"))
+        .await
+        .ok()?;
+    msgs.iter().find_map(|m| match m {
+        tokio_postgres::SimpleQueryMessage::Row(r) => {
+            r.get("n").and_then(|s| s.parse::<i64>().ok())
+        }
+        _ => None,
+    })
+}
+
+/// A transaction reads its OWN staged write even when the collection's vShard is
+/// led by a DIFFERENT node (cross-node read-your-own-writes).
+///
+/// `#[ignore]`d: DEFERRED to U4/U5 (cross-shard transactions). U3's validation
+/// surface is single-node only (LOCKED design §8). See the module docstring for
+/// the confirmed root cause. Run explicitly with `--ignored` to reproduce the
+/// cross-node gap while iterating on U4/U5.
+#[ignore = "cross-node in-txn RYOW is U4/U5 (cross-shard txn); U3 is single-node only — see module docstring"]
+#[tokio::test(flavor = "multi_thread", worker_threads = 6)]
+async fn cross_node_in_txn_select_reads_own_write() {
+    let cluster = TestCluster::spawn_three().await.expect("3-node cluster");
+
+    cluster
+        .exec_ddl_on_any_leader(
+            "CREATE COLLECTION t \
+             (id TEXT PRIMARY KEY, n INT) \
+             WITH (engine='document_strict')",
+        )
+        .await
+        .expect("CREATE COLLECTION t");
+
+    wait_for(
+        "all 3 nodes see collection t",
+        Duration::from_secs(10),
+        Duration::from_millis(50),
+        || {
+            cluster
+                .nodes
+                .iter()
+                .all(|n| n.cached_collection_count() >= 1)
+        },
+    )
+    .await;
+
+    // Drive the whole transaction on each node in turn. The collection is
+    // single-vShard-homed, so every key lives on the ONE owning vShard: the node
+    // leading it does local RYOW (the control that passes today), the other two
+    // route the transaction to it over QUIC and exercise the cross-node path.
+    let mut failures: Vec<String> = Vec::new();
+    for idx in 0..cluster.nodes.len() {
+        let client = &cluster.nodes[idx].client;
+        let id = format!("k{idx}");
+
+        client.simple_query("BEGIN").await.expect("BEGIN");
+
+        // Single-shard write; on a non-owning node it stages to the remote
+        // leaseholder's overlay. An error here is itself a cross-node gap
+        // symptom — record it and move on rather than aborting the whole probe.
+        if let Err(e) = client
+            .simple_query(&format!("INSERT INTO t (id, n) VALUES ('{id}', 42)"))
+            .await
+        {
+            let _ = client.simple_query("ROLLBACK").await;
+            failures.push(format!("node {idx}: in-txn INSERT errored: {e}"));
+            continue;
+        }
+
+        // THE PROBE: the transaction's OWN read must observe its staged write.
+        let seen = n_for(client, &id).await;
+
+        // ROLLBACK before asserting so a miss never leaves the txn open and every
+        // node's staging overlay is torn down regardless of outcome.
+        client.simple_query("ROLLBACK").await.expect("ROLLBACK");
+
+        if seen != Some(42) {
+            failures.push(format!(
+                "node {idx}: RYOW MISS — in-txn SELECT saw {seen:?}, expected Some(42)"
+            ));
+        }
+    }
+
+    assert!(
+        failures.is_empty(),
+        "CROSS-NODE in-txn read-your-own-writes is broken (ExecuteRequest carries \
+         no txn_id, so the owning node stages/reads with txn_id=None): {failures:#?}"
+    );
+
+    cluster.shutdown().await;
+}

nodedb-cluster-tests/tests/native_sql_ryow_probe.rs

@@ -0,0 +1,161 @@
+// SPDX-License-Identifier: BUSL-1.1
+
+//! REGRESSION GUARDS for two native (MessagePack) explicit-transaction
+//! behaviours that were broken on `upstream/main` and fixed on this branch.
+//! Upstream had NO native explicit-txn wire coverage
+//! (`single_node_calvin_graph_txn.rs` drives BEGIN over pgwire), so these paths
+//! were unguarded; both tests assert the CORRECT post-fix behaviour.
+//!
+//! GAP A — first-frame BEGIN must buffer. Previously `handle_begin` ->
+//! `run_begin` -> `sessions.begin(addr)` used
+//! `write_session(addr,..).unwrap_or(Ok(()))`; with NO session yet for the addr
+//! it silently returned `Ok(())` WITHOUT setting `InBlock`, so a BEGIN issued
+//! before any SQL no-oped and the next INSERT autocommitted. `handle_begin` now
+//! calls `ensure_session` before `run_begin`, so the session transitions to
+//! `InBlock` even when BEGIN is the first post-auth frame.
+//!
+//! GAP B (U1) — in-txn SQL read-your-own-writes. Native SQL reads route through
+//! `sql_gateway.rs::dispatch_task_via_gateway`, which previously built a
+//! `GatewayQueryContext { tenant_id, trace_id, database_id }` with NO `txn_id`
+//! and called `gw.execute(&gw_ctx, plan)`, dropping the connection's `txn_id`
+//! before the Data Plane so the per-txn staging overlay-merge never fired. It
+//! now propagates `txn_id` into `GatewayQueryContext` so the Data Plane resolves
+//! this transaction's staging overlay. The pgwire twin
+//! (`nodedb/tests/sql_transactions_staged_point_writes.rs`) asserts `["42"]`.
+//!
+//! The first test guards GAP A; the second WARMS UP the session (a SQL read
+//! triggers `ensure_session`) so a subsequent BEGIN buffers, then guards the
+//! GAP B RYOW path. A failure here signals a REGRESSION of either fix.
+
+mod common;
+
+use std::time::Duration;
+
+use common::cluster_harness::TestClusterNode;
+use nodedb_client::native::connection::NativeConnection;
+use nodedb_types::protocol::AuthMethod;
+use tokio_postgres::SimpleQueryMessage;
+
+/// Single-node harness with collection `t` created and settled.
+async fn setup() -> TestClusterNode {
+    let node = TestClusterNode::spawn(1, vec![])
+        .await
+        .expect("spawn single-node cluster");
+    tokio::time::sleep(Duration::from_millis(300)).await;
+
+    node.exec(
+        "CREATE COLLECTION t \
+         (id STRING NOT NULL PRIMARY KEY, n INT) WITH (engine='document_strict')",
+    )
+    .await
+    .expect("CREATE COLLECTION t");
+    tokio::time::sleep(Duration::from_millis(150)).await;
+
+    node
+}
+
+/// One pinned native connection (explicit txns are peer-address-scoped, so the
+/// whole transaction must ride a single socket).
+async fn native_conn(node: &TestClusterNode) -> NativeConnection {
+    let mut conn = NativeConnection::connect(&format!("127.0.0.1:{}", node.native_port))
+        .await
+        .expect("connect native socket");
+    conn.authenticate(
+        AuthMethod::Trust {
+            username: "admin".into(),
+        },
+        None,
+    )
+    .await
+    .expect("native trust auth");
+    conn
+}
+
+/// `n` for `id` read over the INDEPENDENT pgwire session (durable state, not the
+/// txn connection's overlay). `None` when absent.
+async fn durable_n(node: &TestClusterNode, id: &str) -> Option<String> {
+    let msgs = node
+        .client
+        .simple_query(&format!("SELECT n FROM t WHERE id = '{id}'"))
+        .await
+        .expect("pgwire durable read");
+    msgs.iter().find_map(|m| match m {
+        SimpleQueryMessage::Row(r) => r.get("n").map(str::to_string),
+        _ => None,
+    })
+}
+
+/// GAP A: a BEGIN issued as the first post-auth frame does NOT buffer; the
+/// following INSERT autocommits (visible to an independent session pre-COMMIT).
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn native_first_frame_begin_does_not_buffer_probe() {
+    let node = setup().await;
+    let mut conn = native_conn(&node).await;
+
+    // BEGIN is the very first post-auth frame -> no session exists yet.
+    conn.begin().await.expect("BEGIN");
+    conn.execute_sql("INSERT INTO t (id, n) VALUES ('ff', 7)")
+        .await
+        .expect("INSERT after first-frame BEGIN");
+
+    let leaked = durable_n(&node, "ff").await;
+    assert_eq!(
+        leaked, None,
+        "GAP A GUARD: first-frame BEGIN must buffer the write (invisible \
+         pre-COMMIT). leaked={leaked:?}. Some(\"7\") -> the INSERT \
+         autocommitted (first-frame BEGIN no-oped) -> GAP A REGRESSED."
+    );
+
+    node.shutdown().await;
+}
+
+/// GAP B (U1): with the session warmed up (so BEGIN buffers), the transaction's
+/// OWN native SQL read must observe the staged write (read-your-own-writes).
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn native_in_txn_sql_select_reads_own_write_probe() {
+    let node = setup().await;
+    let mut conn = native_conn(&node).await;
+
+    // WARM UP: a SQL statement triggers `ensure_session` (sql.rs), creating the
+    // session so the following BEGIN can transition it to InBlock.
+    conn.execute_sql("SELECT n FROM t WHERE id = 'warmup'")
+        .await
+        .expect("warm-up read (creates native session)");
+
+    conn.begin().await.expect("BEGIN");
+    conn.execute_sql("INSERT INTO t (id, n) VALUES ('a', 42)")
+        .await
+        .expect("in-tx INSERT");
+
+    // Precondition for a meaningful RYOW probe: the write must actually be
+    // buffered (invisible to an independent session). If this trips, staging is
+    // broken even with a warm session (a broader gap than U1).
+    assert_eq!(
+        durable_n(&node, "a").await,
+        None,
+        "warmed in-block write must be invisible to other sessions mid-transaction"
+    );
+
+    // THE U1 PROBE: the transaction's OWN native SQL read.
+    let res = conn
+        .execute_sql("SELECT n FROM t WHERE id = 'a'")
+        .await
+        .expect("in-tx read");
+    let seen = res
+        .rows
+        .first()
+        .and_then(|r| r.first())
+        .and_then(|v| v.as_i64());
+
+    assert_eq!(
+        seen,
+        Some(42),
+        "GAP B (U1) GUARD: native in-txn SQL SELECT must read-its-own-write \
+         (Some(42)); seen={seen:?}, rows={:?}. None -> native RYOW \
+         REGRESSED (gateway dropped txn_id).",
+        res.rows
+    );
+
+    conn.commit().await.expect("COMMIT");
+    node.shutdown().await;
+}

nodedb-cluster/src/rpc_codec/execute.rs

@@ -38,6 +38,19 @@ pub struct ExecuteRequest {
     pub trace_id: [u8; 16],
     /// Caller's view of descriptor versions for every collection touched by the plan.
     pub descriptor_versions: Vec<DescriptorVersionEntry>,
+    /// Owning session transaction, when this plan executes inside an explicit
+    /// `BEGIN` block. `Some(txn_id)` directs the receiving leaseholder to
+    /// resolve this transaction's staging overlay so a cross-node in-transaction
+    /// read observes its own staged writes (read-your-own-writes); `None` is an
+    /// autocommit statement with no session transaction.
+    ///
+    /// Wire compatibility: this field was added in `WIRE_VERSION` 3. Because
+    /// `ExecuteRequest` is an rkyv fixed-layout message, the added field changes
+    /// its archived layout. The bumped envelope version means a v2 node rejects
+    /// a v3-stamped frame outright (`unwrap_bytes_versioned`) rather than
+    /// silently misdecoding it, so nodes must be upgraded together (see
+    /// `crate::wire`).
+    pub txn_id: Option<u64>,
 }
 
 /// Response to an `ExecuteRequest`.
@@ -238,11 +251,17 @@ mod tests {
                     version: 1,
                 },
             ],
+            txn_id: Some(0xABCD_1234),
         };
         let decoded = roundtrip_req(req.clone());
         assert_eq!(decoded.plan_bytes, req.plan_bytes);
         assert_eq!(decoded.tenant_id, 7);
         assert_eq!(decoded.deadline_remaining_ms, 5000);
+        assert_eq!(
+            decoded.txn_id,
+            Some(0xABCD_1234),
+            "txn_id roundtrips for an in-transaction plan"
+        );
         assert_eq!(
             decoded.trace_id, req.trace_id,
             "trace_id roundtrips correctly"
@@ -261,9 +280,11 @@ mod tests {
             deadline_remaining_ms: 1000,
             trace_id: [0u8; 16],
             descriptor_versions: vec![],
+            txn_id: None,
         };
         let decoded = roundtrip_req(req);
         assert!(decoded.descriptor_versions.is_empty());
+        assert_eq!(decoded.txn_id, None, "autocommit plan carries no txn_id");
     }
 
     #[test]
@@ -382,6 +403,7 @@ mod tests {
                 collection: "wide".into(),
                 version: 3,
             }],
+            txn_id: Some(99),
         };
         let rpc = RaftRpc::ExecuteStreamRequest(req.clone());
         let encoded = super::super::encode(&rpc).unwrap();
@@ -395,6 +417,7 @@ mod tests {
                 assert_eq!(r.descriptor_versions.len(), 1);
                 assert_eq!(r.descriptor_versions[0].collection, "wide");
                 assert_eq!(r.descriptor_versions[0].version, 3);
+                assert_eq!(r.txn_id, Some(99), "stream request carries txn_id");
             }
             other => panic!("expected ExecuteStreamRequest, got {other:?}"),
         }

nodedb-cluster/src/transport/client/tests.rs

@@ -233,6 +233,7 @@ async fn execute_stream_roundtrip() {
         deadline_remaining_ms: 5000,
         trace_id: [0u8; 16],
         descriptor_versions: vec![],
+        txn_id: None,
     });
 
     let stream = client.send_rpc_stream(1, req).await.unwrap();