feat(control): thread transaction id through cross-node dispatch (F1)

nodedb-cluster-tests/tests/cluster_txn_cross_node_ryow.rs

@@ -0,0 +1,150 @@
+// SPDX-License-Identifier: BUSL-1.1
+
+//! Cross-node in-transaction read-your-own-writes.
+//!
+//! An explicit transaction's staged writes live in per-core overlays keyed by
+//! the session's transaction id. When the coordinator is NOT the leader for
+//! the collection's data group, both the staged write (`MetaOp::StageWrite`)
+//! and every in-block read cross the gateway to a remote node — so the
+//! transaction id must survive the `ExecuteRequest` wire hop, or the remote
+//! executor cannot key the overlay (staged writes are rejected with
+//! "StageWrite dispatched without a txn_id") and in-block reads cannot see
+//! the transaction's own staged rows.
+//!
+//! Steps, from EVERY node of a 3-node cluster (at most one of the three is
+//! the collection's data-group leader, so at least two exercise the
+//! cross-node hop):
+//! 1. BEGIN on a dedicated connection (session state is per-connection).
+//! 2. INSERT one row — a stageable point write, staged to the target shard's
+//!    overlay at statement time.
+//! 3. Point SELECT of that row — must see the staged row (RYOW via the
+//!    overlay merge on the shard that staged it).
+//! 4. Full-collection SELECT — must also see it (the gather path threads the
+//!    same transaction id).
+//! 5. ROLLBACK — drops the overlay on every staged shard.
+//! 6. Both SELECTs again — the row must be gone (nothing was committed).
+//!
+//! Committed rows are never touched: the collection stays empty throughout,
+//! so any row visible outside the transaction block is a bug on its own.
+//!
+//! File name contains "cluster" via the cluster-tests crate so nextest
+//! applies the cluster test-group serialization.
+
+use std::time::Duration;
+
+mod common;
+
+use crate::common::cluster_harness::{TestCluster, wait_for};
+
+fn count_rows(msgs: &[tokio_postgres::SimpleQueryMessage]) -> usize {
+    msgs.iter()
+        .filter(|m| matches!(m, tokio_postgres::SimpleQueryMessage::Row(_)))
+        .count()
+}
+
+/// Render every returned row's columns — failure diagnostics only.
+fn dump_rows(msgs: &[tokio_postgres::SimpleQueryMessage]) -> Vec<String> {
+    msgs.iter()
+        .filter_map(|m| match m {
+            tokio_postgres::SimpleQueryMessage::Row(r) => Some(
+                (0..r.len())
+                    .map(|i| r.get(i).unwrap_or("<null>").to_string())
+                    .collect::<Vec<_>>()
+                    .join("|"),
+            ),
+            _ => None,
+        })
+        .collect()
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 6)]
+async fn in_txn_staged_row_is_visible_and_rolls_back_from_every_node() {
+    let cluster = TestCluster::spawn_three().await.expect("3-node cluster");
+
+    let coll = "txn_xnode_ryow";
+
+    cluster
+        .exec_ddl_on_any_leader(&format!(
+            "CREATE COLLECTION {coll} (id TEXT PRIMARY KEY, v TEXT)"
+        ))
+        .await
+        .expect("CREATE COLLECTION");
+
+    wait_for(
+        "all 3 nodes see the collection",
+        Duration::from_secs(15),
+        Duration::from_millis(50),
+        || {
+            cluster
+                .nodes
+                .iter()
+                .all(|n| n.cached_collection_count() >= 1)
+        },
+    )
+    .await;
+
+    for (i, node) in cluster.nodes.iter().enumerate() {
+        let id = format!("row-from-node-{i}");
+
+        node.client
+            .simple_query("BEGIN")
+            .await
+            .unwrap_or_else(|e| panic!("node {i}: BEGIN failed: {e}"));
+
+        // Stageable point write. On a non-leader coordinator this ships a
+        // StageWrite to the remote data-group leader — the transaction id
+        // must ride the wire with it.
+        node.client
+            .simple_query(&format!(
+                "INSERT INTO {coll} (id, v) VALUES ('{id}', 'staged')"
+            ))
+            .await
+            .unwrap_or_else(|e| {
+                panic!("node {i}: in-transaction INSERT (staged write) failed: {e}")
+            });
+
+        // RYOW via point lookup (overlay merge on the owning shard).
+        let point = node
+            .client
+            .simple_query(&format!("SELECT * FROM {coll} WHERE id = '{id}'"))
+            .await
+            .unwrap_or_else(|e| panic!("node {i}: in-transaction point SELECT failed: {e}"));
+        assert_eq!(
+            count_rows(&point),
+            1,
+            "node {i}: point read inside the transaction must see its own staged row"
+        );
+
+        // RYOW via the scan/gather path.
+        let scan = node
+            .client
+            .simple_query(&format!("SELECT * FROM {coll}"))
+            .await
+            .unwrap_or_else(|e| panic!("node {i}: in-transaction scan SELECT failed: {e}"));
+        assert_eq!(
+            count_rows(&scan),
+            1,
+            "node {i}: scan inside the transaction must see exactly its own staged row; got {:?}",
+            dump_rows(&scan)
+        );
+
+        node.client
+            .simple_query("ROLLBACK")
+            .await
+            .unwrap_or_else(|e| panic!("node {i}: ROLLBACK failed: {e}"));
+
+        // The overlay is dropped on every staged shard; nothing was committed.
+        let after = node
+            .client
+            .simple_query(&format!("SELECT * FROM {coll} WHERE id = '{id}'"))
+            .await
+            .unwrap_or_else(|e| panic!("node {i}: post-ROLLBACK SELECT failed: {e}"));
+        assert_eq!(
+            count_rows(&after),
+            0,
+            "node {i}: the staged row must vanish on ROLLBACK"
+        );
+    }
+
+    cluster.shutdown().await;
+}

nodedb-cluster/src/rpc_codec/execute.rs

@@ -38,6 +38,13 @@ 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>,
+    /// The coordinator session's active transaction id, when this plan runs
+    /// inside an explicit transaction block. The receiver stamps it onto the
+    /// local task so the Data Plane keys the staging overlay correctly:
+    /// `MetaOp::StageWrite` stages under it, and in-transaction reads merge
+    /// the transaction's own staged rows (read-your-own-writes). `None` for
+    /// autocommit plans.
+    pub txn_id: Option<u64>,
 }
 
 /// Response to an `ExecuteRequest`.
@@ -238,6 +245,7 @@ mod tests {
                     version: 1,
                 },
             ],
+            txn_id: Some(0xFEED_BEEF),
         };
         let decoded = roundtrip_req(req.clone());
         assert_eq!(decoded.plan_bytes, req.plan_bytes);
@@ -250,6 +258,11 @@ mod tests {
         assert_eq!(decoded.descriptor_versions.len(), 2);
         assert_eq!(decoded.descriptor_versions[0].collection, "orders");
         assert_eq!(decoded.descriptor_versions[0].version, 42);
+        assert_eq!(
+            decoded.txn_id,
+            Some(0xFEED_BEEF),
+            "an in-transaction plan's txn_id must survive the wire"
+        );
     }
 
     #[test]
@@ -261,9 +274,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 plans carry no txn_id");
     }
 
     #[test]
@@ -382,6 +397,7 @@ mod tests {
                 collection: "wide".into(),
                 version: 3,
             }],
+            txn_id: Some(77),
         };
         let rpc = RaftRpc::ExecuteStreamRequest(req.clone());
         let encoded = super::super::encode(&rpc).unwrap();

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

@@ -230,6 +230,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();

nodedb/src/config/server/cluster.rs

@@ -160,6 +160,21 @@ impl ClusterSettings {
                 detail: "cluster.node_id must be non-zero".into(),
             });
         }
+        // Transaction ids pack the node id into the high 16 bits
+        // (`session::transaction`), so a node id that does not fit in 16 bits
+        // would silently lose its high bits and mint colliding ids across
+        // nodes. Reject it loudly rather than mask (see CLAUDE.md on silent
+        // truncation at hard caps).
+        if self.node_id >= 1 << 16 {
+            return Err(crate::Error::Config {
+                detail: format!(
+                    "cluster.node_id must be < 65536 (got {}): it is packed into the \
+                     high 16 bits of a transaction id, and a larger value would silently \
+                     collide with another node's ids",
+                    self.node_id
+                ),
+            });
+        }
         if self.seed_nodes.is_empty() {
             return Err(crate::Error::Config {
                 detail: "cluster.seed_nodes must contain at least one address".into(),
@@ -185,3 +200,43 @@ impl ClusterSettings {
         Ok(())
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::ClusterSettings;
+
+    fn settings_with_node_id(node_id: u64) -> ClusterSettings {
+        let toml = format!(
+            r#"
+            node_id = {node_id}
+            listen = "127.0.0.1:7000"
+            seed_nodes = ["127.0.0.1:7000"]
+            cert = "/tmp/cert.pem"
+            key = "/tmp/key.pem"
+            ca = "/tmp/ca.pem"
+            "#
+        );
+        toml::from_str(&toml).expect("valid cluster TOML")
+    }
+
+    #[test]
+    fn node_id_fitting_in_16_bits_is_accepted() {
+        assert!(settings_with_node_id(1).validate().is_ok());
+        assert!(settings_with_node_id(65_535).validate().is_ok());
+    }
+
+    #[test]
+    fn node_id_at_or_above_16_bits_is_rejected_not_truncated() {
+        // 65536 would pack to 0 in the high 16 bits of a TxnId and collide
+        // with node 1's ids — must be rejected, never silently masked.
+        for bad in [65_536u64, 65_537, 1 << 20] {
+            let err = settings_with_node_id(bad)
+                .validate()
+                .expect_err("node_id >= 65536 must be rejected");
+            assert!(
+                matches!(err, crate::Error::Config { .. }),
+                "expected Config error for node_id {bad}, got {err:?}"
+            );
+        }
+    }
+}

nodedb/src/control/backup/orchestrator.rs

@@ -322,6 +322,8 @@ async fn snapshot_remote(
         deadline_remaining_ms: NODE_SNAPSHOT_TIMEOUT.as_millis() as u64,
         trace_id: TraceId::generate().0,
         descriptor_versions: Vec::new(),
+        // Snapshot capture is not transactional.
+        txn_id: None,
     });
 
     let resp = transport

nodedb/src/control/backup/restore/remote.rs

@@ -38,6 +38,8 @@ pub(super) async fn dispatch_remote(
         deadline_remaining_ms: NODE_RESTORE_TIMEOUT.as_millis() as u64,
         trace_id: TraceId::generate().0,
         descriptor_versions: Vec::new(),
+        // Snapshot restore is not transactional.
+        txn_id: None,
     });
     let resp = transport
         .send_rpc(node_id, req)

nodedb/src/control/exec_receiver/executor.rs

@@ -279,9 +279,17 @@ impl LocalPlanExecutor {
             };
         }
 
+        let txn_id = req.txn_id.map(crate::types::TxnId::new);
         match tokio::time::timeout(
             deadline,
-            execute_plan_all_local_cores(&self.state, tenant_id, database_id, plan, trace_id),
+            execute_plan_all_local_cores(
+                &self.state,
+                tenant_id,
+                database_id,
+                plan,
+                trace_id,
+                txn_id,
+            ),
         )
         .await
         {
@@ -317,15 +325,14 @@ impl LocalPlanExecutor {
         let tenant_id = crate::types::TenantId::new(req.tenant_id);
         let trace_id = nodedb_types::TraceId(req.trace_id);
 
-        // Cluster RPC receiver (remote-node local execution): `ExecuteRequest`
-        // carries no session-transaction context yet, so `None`.
+        let txn_id = req.txn_id.map(crate::types::TxnId::new);
         let mut stream = match crate::control::server::exchange::gather::gather_all_cores_stream(
             &self.state,
             tenant_id,
             database_id,
             plan,
             trace_id,
-            None,
+            txn_id,
         ) {
             Ok(s) => s,
             Err(e) => {

nodedb/src/control/gateway/core.rs

@@ -48,6 +48,13 @@ pub struct QueryContext {
     /// catalog lookups. Single-database deployments pass
     /// [`DatabaseId::DEFAULT`].
     pub database_id: DatabaseId,
+    /// The session's active transaction id when this plan executes inside an
+    /// explicit transaction block, `None` for autocommit. Threaded through
+    /// remote dispatch (`ExecuteRequest.txn_id`) so every shard a plan
+    /// touches — local or on another node — keys the staging overlay
+    /// correctly: staged writes land under it and in-transaction reads merge
+    /// the transaction's own staged rows (read-your-own-writes).
+    pub txn_id: Option<crate::types::TxnId>,
 }
 
 /// The gateway: routes, dispatches, retries, and caches physical plans.
@@ -243,6 +250,7 @@ impl Gateway {
                 let tenant_id = ctx.tenant_id;
                 let database_id = ctx.database_id;
                 let trace_id = ctx.trace_id;
+                let txn_id = ctx.txn_id;
                 let version_set = version_set_for_route.clone();
                 async move {
                     let decision = {
@@ -285,6 +293,7 @@ impl Gateway {
                         trace_id,
                         deadline_ms,
                         &version_set,
+                        txn_id,
                     )
                     .await
                 }