From 1cc57ce28bc5b5f579c9d370c0f3f0f5f0487b4e Mon Sep 17 00:00:00 2001 From: Securifi Dev Date: Wed, 8 Jul 2026 18:18:54 +0000 Subject: [PATCH 1/2] feat(control): thread transaction id through cross-node dispatch (F1) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit An explicit transaction's staged writes live in per-core overlays keyed by the session's transaction id. Before this change the id was threaded only on local/single-shard paths: the gateway's cross-node hop dropped it, so when a coordinator was not the leader for a collection's shard, in-transaction reads could not see the transaction's own staged rows and a forwarded StageWrite was stranded (rejected: "StageWrite dispatched without a txn_id"). Changes: - ExecuteRequest carries an Option txn_id; QueryContext, DispatchRouteStreamParams, and RemoteDispatchArgs carry Option. Autocommit call sites pass None; every transactional path threads the session id through the gateway, remote executor, and all-cores fan. - gather_all_vshards stops dropping txn_id at its QueryContext boundary — the cross-node in-transaction read now reaches the owning leader keyed. - dispatch_local routes in-transaction plans via a new dispatch_to_data_plane_in_txn so the id reaches the Data Plane overlay. - The pgwire whole-batch gateway-forward fast path is now skipped inside a transaction block: it is an autocommit optimization, and an in-transaction write forwarded through it reached the leader as a bare Raft-committed write, bypassing the staging overlay (broke read-your-own-writes and left ROLLBACK unable to undo it). In-block tasks fall through to the staging gate, which stages writes and forwards reads/stage-ops to the leader. - Transaction ids are made globally unique (node id in the high 16 bits) so two coordinators' overlays never collide on a shard that hosts both; the overlay-drop meta-op now carries the id so ROLLBACK's DropTxnOverlay routes to the owning leader, not a local replica. Cross-node RYOW is exercised by a new 3-node e2e that runs BEGIN / staged INSERT / point + scan RYOW / ROLLBACK / post-rollback-empty from every node, so at least two iterations drive the non-leader forward path. Single-node overlay regressions, cluster calvin e2e, and codec roundtrips stay green; clippy clean. Co-Authored-By: Claude Fable 5 --- Cargo.lock | 18 +++ .../tests/cluster_txn_cross_node_ryow.rs | 150 ++++++++++++++++++ nodedb-cluster/src/rpc_codec/execute.rs | 16 ++ nodedb-cluster/src/transport/client/tests.rs | 1 + nodedb/src/control/backup/orchestrator.rs | 2 + nodedb/src/control/backup/restore/remote.rs | 2 + nodedb/src/control/exec_receiver/executor.rs | 15 +- nodedb/src/control/gateway/core.rs | 9 ++ nodedb/src/control/gateway/dispatcher.rs | 39 +++-- nodedb/src/control/gateway/stream.rs | 2 + nodedb/src/control/planner/calvin/preexec.rs | 3 + nodedb/src/control/scatter_gather.rs | 3 + nodedb/src/control/server/dispatch_utils.rs | 32 ++++ .../server/exchange/all_cores/dispatch.rs | 13 +- nodedb/src/control/server/exchange/gather.rs | 1 + .../server/exchange/resolve/exchange.rs | 7 +- .../server/graph_dispatch/cluster_resolve.rs | 4 + .../src/control/server/graph_dispatch/hop.rs | 2 + .../match_scatter/round_loop.rs | 2 + .../match_scatter/round_zero.rs | 2 + .../server/http/routes/promql/remote.rs | 2 + .../src/control/server/http/routes/query.rs | 2 + .../server/http/routes/query_stream.rs | 1 + .../server/http/routes/ws_rpc/execute_sql.rs | 1 + nodedb/src/control/server/ilp_batch.rs | 2 + .../server/native/dispatch/direct_ops.rs | 1 + .../server/native/dispatch/sql_gateway.rs | 1 + .../server/native/dispatch/streaming.rs | 3 + .../server/native/dispatch/transaction.rs | 4 + .../control/server/pgwire/handler/dispatch.rs | 63 ++++++++ .../server/pgwire/handler/routing/execute.rs | 14 +- .../handler/routing/gateway_dispatch.rs | 5 +- .../pgwire/handler/routing/streaming.rs | 1 + .../control/server/resp/gateway_dispatch.rs | 4 + .../ddl/neutral/collection/index_fanout.rs | 2 + .../control/server/shared/session/commit.rs | 4 +- .../server/shared/session/lifecycle.rs | 2 +- .../control/server/shared/session/read_set.rs | 2 +- .../control/server/shared/session/tests.rs | 6 +- .../server/shared/session/transaction.rs | 10 +- .../control/server/shuffle/producer_hook.rs | 2 + nodedb/src/event/cdc/consume.rs | 2 + nodedb/src/event/topic/publish.rs | 2 + 43 files changed, 422 insertions(+), 37 deletions(-) create mode 100644 nodedb-cluster-tests/tests/cluster_txn_cross_node_ryow.rs diff --git a/Cargo.lock b/Cargo.lock index f223c0ee8..165b5b770 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2272,6 +2272,8 @@ checksum = "28a80e3145d8ad11ba0995949bbcf48b9df2be62772b3d351ef017dff6ecb853" [[package]] name = "fluxbench" version = "0.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "323b4f98b988b9779c6aa3f93781ef7421e36e73a5c610497857b2c01398acdc" dependencies = [ "fluxbench-cli", "fluxbench-core", @@ -2286,6 +2288,8 @@ dependencies = [ [[package]] name = "fluxbench-cli" version = "0.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "42072e01658451938143f620a08f13b308e30a7fe41f3de87a79f351c5753328" dependencies = [ "anyhow", "chrono", @@ -2313,6 +2317,8 @@ dependencies = [ [[package]] name = "fluxbench-core" version = "0.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "780bc18df5d1aa766ab5394389fb989c51d5d6ac3b00e51340f97c6510268edf" dependencies = [ "fluxbench-ipc", "inventory", @@ -2325,6 +2331,8 @@ dependencies = [ [[package]] name = "fluxbench-ipc" version = "0.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7c44dcc5fe5705f4933b7768a4fdb552ab97c098d4d3e49b93096152527b89fe" dependencies = [ "rkyv 0.8.16", "thiserror 2.0.18", @@ -2333,6 +2341,8 @@ dependencies = [ [[package]] name = "fluxbench-logic" version = "0.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ab4809eb2b1c33ee46c51cc28ee5d8c8ea8dbc4977151af00a03a1d1f5d487ab" dependencies = [ "evalexpr", "fluxbench-core", @@ -2347,6 +2357,8 @@ dependencies = [ [[package]] name = "fluxbench-macros" version = "0.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "71729e5cb94b94beb14f6a9319a129e6cf3e116112f56578182edadeaaddc79a" dependencies = [ "proc-macro2", "quote", @@ -2356,6 +2368,8 @@ dependencies = [ [[package]] name = "fluxbench-report" version = "0.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "02bf8934a0c315e390d691e3c9993e52b59ed88ecb8993d882a69721559342d0" dependencies = [ "chrono", "fluxbench-core", @@ -2369,6 +2383,8 @@ dependencies = [ [[package]] name = "fluxbench-stats" version = "0.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bec71a4355a6558d47beea9abda1fda67580e124fefdb9e93f0aaab4cc872063" dependencies = [ "rand 0.8.6", "rayon", @@ -4117,6 +4133,8 @@ dependencies = [ [[package]] name = "nexar" version = "0.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dac038912d8e0f36309f3368c0980377730edd4ad5d082627227277cdd870c3a" dependencies = [ "bytes", "crossbeam-queue", diff --git a/nodedb-cluster-tests/tests/cluster_txn_cross_node_ryow.rs b/nodedb-cluster-tests/tests/cluster_txn_cross_node_ryow.rs new file mode 100644 index 000000000..c62a7e4d6 --- /dev/null +++ b/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 { + msgs.iter() + .filter_map(|m| match m { + tokio_postgres::SimpleQueryMessage::Row(r) => Some( + (0..r.len()) + .map(|i| r.get(i).unwrap_or("").to_string()) + .collect::>() + .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; +} diff --git a/nodedb-cluster/src/rpc_codec/execute.rs b/nodedb-cluster/src/rpc_codec/execute.rs index 5bd1fa68c..7fdafb80c 100644 --- a/nodedb-cluster/src/rpc_codec/execute.rs +++ b/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, + /// 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, } /// 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(); diff --git a/nodedb-cluster/src/transport/client/tests.rs b/nodedb-cluster/src/transport/client/tests.rs index ea8c59798..2c3945926 100644 --- a/nodedb-cluster/src/transport/client/tests.rs +++ b/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(); diff --git a/nodedb/src/control/backup/orchestrator.rs b/nodedb/src/control/backup/orchestrator.rs index 169373b43..9c8a819cb 100644 --- a/nodedb/src/control/backup/orchestrator.rs +++ b/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 diff --git a/nodedb/src/control/backup/restore/remote.rs b/nodedb/src/control/backup/restore/remote.rs index 1394dcfba..8121d0263 100644 --- a/nodedb/src/control/backup/restore/remote.rs +++ b/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) diff --git a/nodedb/src/control/exec_receiver/executor.rs b/nodedb/src/control/exec_receiver/executor.rs index d5ff5cd3c..3b9dfeba7 100644 --- a/nodedb/src/control/exec_receiver/executor.rs +++ b/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) => { diff --git a/nodedb/src/control/gateway/core.rs b/nodedb/src/control/gateway/core.rs index dbf66f11d..3574ebc75 100644 --- a/nodedb/src/control/gateway/core.rs +++ b/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, } /// 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 } diff --git a/nodedb/src/control/gateway/dispatcher.rs b/nodedb/src/control/gateway/dispatcher.rs index 77df4f565..ca42ff990 100644 --- a/nodedb/src/control/gateway/dispatcher.rs +++ b/nodedb/src/control/gateway/dispatcher.rs @@ -23,10 +23,11 @@ use nodedb_cluster::rpc_codec::{ExecuteRequest, RaftRpc, TypedClusterError}; use tracing::debug; use crate::Error; -use crate::control::server::dispatch_utils::dispatch_to_data_plane; +use crate::control::server::dispatch_utils::dispatch_to_data_plane_in_txn; use crate::control::server::result_stream::{ResultStream, RowBatch}; use crate::control::state::SharedState; use crate::types::{DatabaseId, Lsn, TenantId, TraceId, VShardId}; +use crate::types::TxnId; use nodedb_physical::physical_plan::wire as plan_wire; use super::route::{RouteDecision, TaskRoute}; @@ -46,10 +47,11 @@ pub async fn dispatch_route( trace_id: TraceId, deadline_ms: u64, version_set: &GatewayVersionSet, + txn_id: Option, ) -> Result>, Error> { match route.decision { RouteDecision::Local => { - dispatch_local(route, shared, tenant_id, database_id, trace_id).await + dispatch_local(route, shared, tenant_id, database_id, trace_id, txn_id).await } RouteDecision::Remote { node_id, vshard_id } => { dispatch_remote(RemoteDispatchArgs { @@ -62,6 +64,7 @@ pub async fn dispatch_route( trace_id, deadline_ms, version_set, + txn_id, }) .await } @@ -96,6 +99,11 @@ pub struct DispatchRouteStreamParams<'a> { pub trace_id: TraceId, pub deadline_ms: u64, pub version_set: &'a GatewayVersionSet, + /// Active transaction id when streaming inside an explicit transaction + /// block; `None` for autocommit. See [`QueryContext::txn_id`]. + /// + /// [`QueryContext::txn_id`]: super::core::QueryContext::txn_id + pub txn_id: Option, } /// Streaming sibling of [`dispatch_route`]: dispatch a single route and return @@ -118,18 +126,16 @@ pub async fn dispatch_route_stream( trace_id, deadline_ms, version_set, + txn_id, } = args; match route.decision { - // Cluster gateway route dispatch: no session-transaction context - // crosses this boundary yet, so `None`. TRACKED: cross-node - // in-transaction reads are a known gap (see resolve/exchange.rs). RouteDecision::Local => crate::control::server::exchange::gather::gather_all_cores_stream( shared, tenant_id, database_id, route.plan, trace_id, - None, + txn_id, ), RouteDecision::Remote { node_id, vshard_id } => { dispatch_remote_stream(RemoteDispatchArgs { @@ -142,6 +148,7 @@ pub async fn dispatch_route_stream( trace_id, deadline_ms, version_set, + txn_id, }) .await } @@ -164,15 +171,17 @@ async fn dispatch_local( tenant_id: TenantId, database_id: DatabaseId, trace_id: TraceId, + txn_id: Option, ) -> Result>, Error> { let vshard_id = VShardId::new(route.vshard_id); - let resp = dispatch_to_data_plane( + let resp = dispatch_to_data_plane_in_txn( shared, tenant_id, database_id, vshard_id, route.plan, trace_id, + txn_id, ) .await?; Ok(vec![resp.payload.to_vec()]) @@ -190,6 +199,10 @@ struct RemoteDispatchArgs<'a> { trace_id: TraceId, deadline_ms: u64, version_set: &'a GatewayVersionSet, + /// Active transaction id when dispatching inside an explicit transaction + /// block; carried on [`ExecuteRequest::txn_id`] so the remote executor + /// keys the staging overlay for this transaction. `None` for autocommit. + txn_id: Option, } /// Remote dispatch via `ExecuteRequest` RPC. @@ -204,6 +217,7 @@ async fn dispatch_remote(args: RemoteDispatchArgs<'_>) -> Result>, E trace_id, deadline_ms, version_set, + txn_id, } = args; let transport = shared.cluster_transport.as_ref().ok_or(Error::Internal { detail: "gateway: cluster transport not available for remote dispatch".into(), @@ -218,16 +232,13 @@ async fn dispatch_remote(args: RemoteDispatchArgs<'_>) -> Result>, E // already done upstream on the pgwire/native paths that own the identity. // (`Box::pin` breaks the async-recursion cycle: resolving a Broadcast build // side calls `gather_all_vshards` → `gateway.execute` → routing → here.) - // Cluster remote-dispatch: no session-transaction context crosses this - // boundary yet, so `None`. TRACKED: cross-node in-transaction reads are a - // known gap (see resolve/exchange.rs). let plan = match Box::pin(crate::control::server::exchange::resolve_exchange_in_plan( shared, database_id, tenant_id, plan, trace_id, - None, + txn_id, )) .await? { @@ -270,6 +281,7 @@ async fn dispatch_remote(args: RemoteDispatchArgs<'_>) -> Result>, E deadline_remaining_ms: deadline_ms, trace_id: trace_id.0, descriptor_versions, + txn_id: txn_id.map(TxnId::as_u64), }); debug!( @@ -337,20 +349,20 @@ async fn dispatch_remote_stream(args: RemoteDispatchArgs<'_>) -> Result) -> Result, +) -> crate::Result { + dispatch_to_data_plane_inner( + shared, + DataPlaneDispatch { + tenant_id, + database_id, + vshard_id, + plan, + trace_id, + event_source: crate::event::EventSource::User, + txn_id, + wal_lsn: None, + }, + ) + .await +} + /// Dispatch a physical plan to the Data Plane with an explicit event source. /// /// Trigger-generated writes pass `EventSource::Trigger` so the Data Plane diff --git a/nodedb/src/control/server/exchange/all_cores/dispatch.rs b/nodedb/src/control/server/exchange/all_cores/dispatch.rs index 347933712..6b7f402c1 100644 --- a/nodedb/src/control/server/exchange/all_cores/dispatch.rs +++ b/nodedb/src/control/server/exchange/all_cores/dispatch.rs @@ -44,6 +44,7 @@ pub async fn execute_plan_all_local_cores( database_id: DatabaseId, plan: PhysicalPlan, trace_id: TraceId, + txn_id: Option, ) -> crate::Result { match &plan { PhysicalPlan::Graph(g) => match g { @@ -111,7 +112,7 @@ pub async fn execute_plan_all_local_cores( | GraphOp::TemporalNeighbors { .. } | GraphOp::TemporalAlgorithm { .. } | GraphOp::Stats { .. } => { - generic_gather(state, tenant_id, database_id, plan, trace_id).await + generic_gather(state, tenant_id, database_id, plan, trace_id, txn_id).await } }, @@ -181,7 +182,7 @@ pub async fn execute_plan_all_local_cores( | MetaOp::DropTxnOverlay { .. } | MetaOp::MarkSavepoint { .. } | MetaOp::RollbackToSavepoint { .. } => { - generic_gather(state, tenant_id, database_id, plan, trace_id).await + generic_gather(state, tenant_id, database_id, plan, trace_id, txn_id).await } }, @@ -197,7 +198,7 @@ pub async fn execute_plan_all_local_cores( | PhysicalPlan::Query(_) | PhysicalPlan::Array(_) | PhysicalPlan::ClusterArray(_) => { - generic_gather(state, tenant_id, database_id, plan, trace_id).await + generic_gather(state, tenant_id, database_id, plan, trace_id, txn_id).await } } } @@ -209,13 +210,11 @@ async fn generic_gather( database_id: DatabaseId, plan: PhysicalPlan, trace_id: TraceId, + txn_id: Option, ) -> crate::Result { use crate::control::server::exchange::gather::gather_all_cores; - // Cluster RPC receiver path (remote-node local execution): no session - // transaction context crosses the node boundary yet, so `txn_id` is - // `None` here. TRACKED: cross-node in-transaction reads are a known gap. - let outcome = gather_all_cores(state, tenant_id, database_id, plan, trace_id, None).await?; + let outcome = gather_all_cores(state, tenant_id, database_id, plan, trace_id, txn_id).await?; Ok(NodeLevelResult { payload: outcome.merged_array, watermark_lsn: outcome.watermark_lsn, diff --git a/nodedb/src/control/server/exchange/gather.rs b/nodedb/src/control/server/exchange/gather.rs index aaba53535..b549e74b1 100644 --- a/nodedb/src/control/server/exchange/gather.rs +++ b/nodedb/src/control/server/exchange/gather.rs @@ -366,6 +366,7 @@ pub async fn gather_all_vshards( tenant_id, trace_id, database_id, + txn_id, }; // `Box::pin` breaks an async-fn recursion cycle: the gateway dispatches the diff --git a/nodedb/src/control/server/exchange/resolve/exchange.rs b/nodedb/src/control/server/exchange/resolve/exchange.rs index be8a18ec2..134edd356 100644 --- a/nodedb/src/control/server/exchange/resolve/exchange.rs +++ b/nodedb/src/control/server/exchange/resolve/exchange.rs @@ -165,15 +165,14 @@ async fn resolve_exchange( // `shard_watermarks`. if !as_aggregate && txn_id.is_none() && child.is_streamable_unordered_scan() { let stream = if let Some(gw) = state.gateway.as_ref() { + // `txn_id` is None on this branch (guarded above); carried + // verbatim so the invariant lives in one place. let ctx = crate::control::gateway::core::QueryContext { tenant_id, trace_id, database_id, + txn_id, }; - // NOTE: cluster mode does not yet thread `txn_id` through - // `gateway.execute_stream` — cross-node in-transaction - // read-your-own-writes is a tracked gap; single-node - // (`gather_all_cores_stream` below) is fixed. gw.execute_stream(&ctx, child).await? } else { gather_all_cores_stream(state, tenant_id, database_id, child, trace_id, txn_id)? diff --git a/nodedb/src/control/server/graph_dispatch/cluster_resolve.rs b/nodedb/src/control/server/graph_dispatch/cluster_resolve.rs index 658b4000f..cfc97637d 100644 --- a/nodedb/src/control/server/graph_dispatch/cluster_resolve.rs +++ b/nodedb/src/control/server/graph_dispatch/cluster_resolve.rs @@ -93,6 +93,8 @@ pub(in crate::control::server::graph_dispatch) async fn dispatch_superstep_to_no database_id, plan, TraceId::ZERO, + // Graph scatter reads committed state only (no overlay merge). + None, ) .await?; Ok(Payload::from_vec(node_result.payload)) @@ -114,6 +116,8 @@ pub(in crate::control::server::graph_dispatch) async fn dispatch_superstep_to_no TraceId::ZERO, deadline_ms, version_set, + // Graph scatter reads committed state only (no overlay merge). + None, ) .await?; payloads diff --git a/nodedb/src/control/server/graph_dispatch/hop.rs b/nodedb/src/control/server/graph_dispatch/hop.rs index 74bfa6563..dccb57885 100644 --- a/nodedb/src/control/server/graph_dispatch/hop.rs +++ b/nodedb/src/control/server/graph_dispatch/hop.rs @@ -340,6 +340,8 @@ async fn expand_remote( TraceId::ZERO, deadline_ms, &version_set, + // Graph scatter reads committed state only (no overlay merge). + None, ) .await }) diff --git a/nodedb/src/control/server/graph_dispatch/match_scatter/round_loop.rs b/nodedb/src/control/server/graph_dispatch/match_scatter/round_loop.rs index b65c6df75..eba1e7e28 100644 --- a/nodedb/src/control/server/graph_dispatch/match_scatter/round_loop.rs +++ b/nodedb/src/control/server/graph_dispatch/match_scatter/round_loop.rs @@ -98,6 +98,8 @@ fn push_dispatch_fut<'f>( TraceId::ZERO, deadline_ms, &version_set, + // Graph scatter reads committed state only (no overlay merge). + None, ) .await?; collect_remote_envelopes(node_id, payloads) diff --git a/nodedb/src/control/server/graph_dispatch/match_scatter/round_zero.rs b/nodedb/src/control/server/graph_dispatch/match_scatter/round_zero.rs index 2c18942e8..738f56500 100644 --- a/nodedb/src/control/server/graph_dispatch/match_scatter/round_zero.rs +++ b/nodedb/src/control/server/graph_dispatch/match_scatter/round_zero.rs @@ -85,6 +85,8 @@ pub(super) async fn scatter_round_zero( TraceId::ZERO, deadline_ms, &version_set, + // Graph scatter reads committed state only (no overlay merge). + None, ) .await?; collect_remote_envelopes(node_id, payloads) diff --git a/nodedb/src/control/server/http/routes/promql/remote.rs b/nodedb/src/control/server/http/routes/promql/remote.rs index 0e4d500d9..3452882a8 100644 --- a/nodedb/src/control/server/http/routes/promql/remote.rs +++ b/nodedb/src/control/server/http/routes/promql/remote.rs @@ -90,6 +90,8 @@ pub async fn remote_write( tenant_id, trace_id: TraceId::generate(), database_id: nodedb_types::id::DatabaseId::DEFAULT, + // PromQL remote-read is autocommit. + txn_id: None, }; gw.execute(&gw_ctx, plan).await } diff --git a/nodedb/src/control/server/http/routes/query.rs b/nodedb/src/control/server/http/routes/query.rs index 677c50eb4..52c12be6f 100644 --- a/nodedb/src/control/server/http/routes/query.rs +++ b/nodedb/src/control/server/http/routes/query.rs @@ -201,6 +201,7 @@ pub async fn query( tenant_id: task.tenant_id, trace_id, database_id, + txn_id: task.txn_id, }; gw.execute(&gw_ctx, task.plan).await.map_err(|e| { let (status, msg) = GatewayErrorMap::to_http(&e); @@ -402,6 +403,7 @@ pub async fn query_ndjson( tenant_id: task.tenant_id, trace_id, database_id, + txn_id: task.txn_id, }; gw.execute(&gw_ctx, task.plan).await } diff --git a/nodedb/src/control/server/http/routes/query_stream.rs b/nodedb/src/control/server/http/routes/query_stream.rs index 775b82990..660bdd818 100644 --- a/nodedb/src/control/server/http/routes/query_stream.rs +++ b/nodedb/src/control/server/http/routes/query_stream.rs @@ -61,6 +61,7 @@ pub(super) async fn try_open_stream( tenant_id: task.tenant_id, trace_id, database_id, + txn_id: task.txn_id, }; gw.execute_stream(&ctx, child_plan).await } else { diff --git a/nodedb/src/control/server/http/routes/ws_rpc/execute_sql.rs b/nodedb/src/control/server/http/routes/ws_rpc/execute_sql.rs index 725bd07bd..9b4418758 100644 --- a/nodedb/src/control/server/http/routes/ws_rpc/execute_sql.rs +++ b/nodedb/src/control/server/http/routes/ws_rpc/execute_sql.rs @@ -35,6 +35,7 @@ pub async fn execute_sql( tenant_id: task.tenant_id, trace_id, database_id: nodedb_types::id::DatabaseId::DEFAULT, + txn_id: task.txn_id, }; gw.execute(&gw_ctx, task.plan).await } diff --git a/nodedb/src/control/server/ilp_batch.rs b/nodedb/src/control/server/ilp_batch.rs index ae6f6600f..2ec9cb357 100644 --- a/nodedb/src/control/server/ilp_batch.rs +++ b/nodedb/src/control/server/ilp_batch.rs @@ -151,6 +151,8 @@ async fn flush_ilp_batch_inner( tenant_id, trace_id: TraceId::generate(), database_id: nodedb_types::id::DatabaseId::DEFAULT, + // ILP ingest is autocommit — no transaction context. + txn_id: None, }; gw.execute(&gw_ctx, plan) .await diff --git a/nodedb/src/control/server/native/dispatch/direct_ops.rs b/nodedb/src/control/server/native/dispatch/direct_ops.rs index 77b2d700b..b44e0ed92 100644 --- a/nodedb/src/control/server/native/dispatch/direct_ops.rs +++ b/nodedb/src/control/server/native/dispatch/direct_ops.rs @@ -398,6 +398,7 @@ async fn dispatch_single_task_raw( tenant_id, trace_id: TraceId::generate(), database_id: ctx.database_id(), + txn_id, }; match gw.execute(&gw_ctx, plan).await { Ok(payloads) => Ok(gateway_payloads_to_response(payloads)), diff --git a/nodedb/src/control/server/native/dispatch/sql_gateway.rs b/nodedb/src/control/server/native/dispatch/sql_gateway.rs index 129e97e8a..9553a3684 100644 --- a/nodedb/src/control/server/native/dispatch/sql_gateway.rs +++ b/nodedb/src/control/server/native/dispatch/sql_gateway.rs @@ -39,6 +39,7 @@ pub(super) async fn dispatch_task_via_gateway( tenant_id, trace_id: TraceId::generate(), database_id, + txn_id, }; gw.execute(&gw_ctx, plan) .await diff --git a/nodedb/src/control/server/native/dispatch/streaming.rs b/nodedb/src/control/server/native/dispatch/streaming.rs index 7e12d2e0d..5299e1c86 100644 --- a/nodedb/src/control/server/native/dispatch/streaming.rs +++ b/nodedb/src/control/server/native/dispatch/streaming.rs @@ -106,6 +106,9 @@ pub(crate) async fn try_open_sql_stream( tenant_id: task.tenant_id, trace_id: crate::types::TraceId::ZERO, database_id, + // In-block statements never reach this stream path (guarded + // above); carried verbatim so the invariant lives in one place. + txn_id: task.txn_id, }; gw.execute_stream(&gw_ctx, child_plan).await? } else { diff --git a/nodedb/src/control/server/native/dispatch/transaction.rs b/nodedb/src/control/server/native/dispatch/transaction.rs index 1352bf675..e9919da9c 100644 --- a/nodedb/src/control/server/native/dispatch/transaction.rs +++ b/nodedb/src/control/server/native/dispatch/transaction.rs @@ -53,6 +53,10 @@ impl TxnDataPlane for NativeTxnDp<'_> { tenant_id: task.tenant_id, trace_id: TraceId::generate(), database_id: task.database_id, + // Carries the transaction's id so staging-overlay + // meta-ops (StageWrite / DropTxnOverlay) reach remote + // shards correctly keyed. + txn_id: task.txn_id, }; let payloads = gw.execute(&gw_ctx, task.plan).await?; Ok(Response { diff --git a/nodedb/src/control/server/pgwire/handler/dispatch.rs b/nodedb/src/control/server/pgwire/handler/dispatch.rs index 2107bf21c..37b80969c 100644 --- a/nodedb/src/control/server/pgwire/handler/dispatch.rs +++ b/nodedb/src/control/server/pgwire/handler/dispatch.rs @@ -237,9 +237,60 @@ impl NodeDbPgHandler { return self.dispatch_replicated_write(entry, async_proposer).await; } + // In-transaction tasks must execute on the vShard that OWNS the + // transaction's staging overlay — the target shard's leader — never a + // local replica. The staged writes live in the owning shard's per-core + // `txn_overlays[txn_id]`, so a local-replica read would miss the + // transaction's own staged rows, and a locally-dispatched + // `MetaOp::StageWrite` would strand the overlay on the coordinator + // where no leader-routed read can find it. `Strong` forces the + // leader-required check regardless of the session's read consistency. + if task.txn_id.is_some() + && self + .should_forward_via_gateway(std::slice::from_ref(&task), ReadConsistency::Strong) + { + return self.dispatch_in_txn_via_gateway(task).await; + } + self.dispatch_local(task, user_id).await } + /// Dispatch a single in-transaction task through the gateway to the vShard + /// leader that owns the transaction's staging overlay, returning the raw + /// envelope [`Response`] (no pgwire shaping — this sits on the neutral + /// dispatch path, whose callers do their own tagging). Mirrors the shape + /// of the native `TxnDataPlane` gateway leg: the first payload is carried + /// so staging-overlay meta-op responses still decode. + /// + /// The gateway carries no LSN metadata yet, so `watermark_lsn` is ZERO — + /// an over-conservative read version (never unsafe: a ZERO version can + /// only widen, not narrow, later conflict detection). + async fn dispatch_in_txn_via_gateway(&self, task: PhysicalTask) -> crate::Result { + let Some(gateway) = self.state.gateway.as_ref() else { + return Err(crate::Error::Internal { + detail: "gateway unavailable for in-transaction leader forward".into(), + }); + }; + let gw_ctx = crate::control::gateway::core::QueryContext { + tenant_id: task.tenant_id, + trace_id: TraceId::generate(), + database_id: task.database_id, + txn_id: task.txn_id, + }; + let payloads = gateway.execute(&gw_ctx, task.plan).await?; + Ok(Response { + request_id: crate::types::RequestId::new(0), + status: crate::bridge::envelope::Status::Ok, + attempt: 1, + partial: false, + payload: crate::bridge::envelope::Payload::from_vec( + payloads.into_iter().next().unwrap_or_default(), + ), + watermark_lsn: Lsn::new(0), + error_code: None, + }) + } + /// Dispatch a write through Raft: propose → register waiter → await apply. /// /// The `AsyncRaftProposer` handles propose + waiter registration in one @@ -325,6 +376,18 @@ impl NodeDbPgHandler { database_id: task.database_id, }); } + // Overlay meta-ops (no WAL record: StageWrite, DropTxnOverlay, + // savepoint marks) must reach the vShard leader that owns the + // transaction's staging overlay — the same routing rule as + // `dispatch_task_inner`. WAL-stamped COMMIT-replay tasks are NOT + // rerouted: their write version is bound to this node's WAL record. + if wal_lsn.is_none() + && task.txn_id.is_some() + && self + .should_forward_via_gateway(std::slice::from_ref(&task), ReadConsistency::Strong) + { + return self.dispatch_in_txn_via_gateway(task).await; + } let txn_id = task.txn_id; // The transaction's writes were durably recorded under a single // `RecordType::Transaction` WAL record at COMMIT; per-task WAL append is diff --git a/nodedb/src/control/server/pgwire/handler/routing/execute.rs b/nodedb/src/control/server/pgwire/handler/routing/execute.rs index 0ebd9f048..d09bb82e1 100644 --- a/nodedb/src/control/server/pgwire/handler/routing/execute.rs +++ b/nodedb/src/control/server/pgwire/handler/routing/execute.rs @@ -194,7 +194,19 @@ impl NodeDbPgHandler { let consistency = consistency_for_tasks(&tasks); // When all tasks target a remote leader, route through the gateway. - if self.should_forward_via_gateway(&tasks, consistency) { + // + // This whole-batch fast path is an AUTOCOMMIT optimization only. Inside + // an explicit transaction block it MUST be skipped: an in-transaction + // write forwarded here would reach the leader as a bare write and + // commit via Raft, bypassing the staging overlay entirely (no + // read-your-own-writes, and ROLLBACK could not undo it). In-block tasks + // instead fall through to the per-task staging gate (`route_task_in_txn`), + // which stages writes to the transaction's overlay and forwards reads + + // stage-ops to the owning leader carrying the transaction id. + if self.sessions.transaction_state(addr) + != crate::control::server::shared::session::TransactionState::InBlock + && self.should_forward_via_gateway(&tasks, consistency) + { let database_id = self .sessions .get_current_database(addr) diff --git a/nodedb/src/control/server/pgwire/handler/routing/gateway_dispatch.rs b/nodedb/src/control/server/pgwire/handler/routing/gateway_dispatch.rs index c3aef9864..92190721e 100644 --- a/nodedb/src/control/server/pgwire/handler/routing/gateway_dispatch.rs +++ b/nodedb/src/control/server/pgwire/handler/routing/gateway_dispatch.rs @@ -28,7 +28,7 @@ impl NodeDbPgHandler { /// Returns `true` when every task targets a single remote leader and the /// gateway is available to forward them. This replaces the old /// `remote_leader_for_tasks` helper which returned the leader node id. - pub(super) fn should_forward_via_gateway( + pub(in crate::control::server::pgwire::handler) fn should_forward_via_gateway( &self, tasks: &[PhysicalTask], consistency: ReadConsistency, @@ -103,6 +103,9 @@ impl NodeDbPgHandler { tenant_id, trace_id: TraceId::generate(), database_id, + // Every task in the batch belongs to this one session, so they + // all carry the same transaction id (stamped at the staging gate). + txn_id: tasks.first().and_then(|t| t.txn_id), }; let mut responses: Vec = Vec::with_capacity(tasks.len()); diff --git a/nodedb/src/control/server/pgwire/handler/routing/streaming.rs b/nodedb/src/control/server/pgwire/handler/routing/streaming.rs index f1592faee..405863c66 100644 --- a/nodedb/src/control/server/pgwire/handler/routing/streaming.rs +++ b/nodedb/src/control/server/pgwire/handler/routing/streaming.rs @@ -82,6 +82,7 @@ impl NodeDbPgHandler { tenant_id: task.tenant_id, trace_id: crate::types::TraceId::ZERO, database_id: task.database_id, + txn_id: task.txn_id, }; gw.execute_stream(&ctx, child_plan).await } else { diff --git a/nodedb/src/control/server/resp/gateway_dispatch.rs b/nodedb/src/control/server/resp/gateway_dispatch.rs index dbce5a96d..a1e967228 100644 --- a/nodedb/src/control/server/resp/gateway_dispatch.rs +++ b/nodedb/src/control/server/resp/gateway_dispatch.rs @@ -38,6 +38,8 @@ pub(super) async fn dispatch_kv( tenant_id: session.tenant_id, trace_id: TraceId::generate(), database_id: DatabaseId::DEFAULT, + // RESP has no transaction blocks — autocommit only. + txn_id: None, }; gw.execute(&gw_ctx, plan) .await @@ -86,6 +88,8 @@ pub(super) async fn dispatch_kv_write( tenant_id: session.tenant_id, trace_id: TraceId::generate(), database_id: DatabaseId::DEFAULT, + // RESP has no transaction blocks — autocommit only. + txn_id: None, }; gw.execute(&gw_ctx, plan) .await diff --git a/nodedb/src/control/server/shared/ddl/neutral/collection/index_fanout.rs b/nodedb/src/control/server/shared/ddl/neutral/collection/index_fanout.rs index 8fb845f47..bc1097764 100644 --- a/nodedb/src/control/server/shared/ddl/neutral/collection/index_fanout.rs +++ b/nodedb/src/control/server/shared/ddl/neutral/collection/index_fanout.rs @@ -130,6 +130,8 @@ pub(super) async fn backfill_on_peers( deadline_remaining_ms: deadline_ms, trace_id: trace_id.0, descriptor_versions: Vec::new(), + // DDL backfill fan-out is not transactional. + txn_id: None, }); joins.push(tokio::spawn(async move { let outcome = transport.send_rpc(node_id, req).await; diff --git a/nodedb/src/control/server/shared/session/commit.rs b/nodedb/src/control/server/shared/session/commit.rs index fbc1a38c3..84a9051f5 100644 --- a/nodedb/src/control/server/shared/session/commit.rs +++ b/nodedb/src/control/server/shared/session/commit.rs @@ -309,7 +309,9 @@ pub(super) async fn drop_txn_overlay( database_id: crate::types::DatabaseId::DEFAULT, plan: PhysicalPlan::Meta(MetaOp::DropTxnOverlay { txn_id }), post_set_op: PostSetOp::None, - txn_id: None, + // Envelope id doubles as the routing signal: the drop must reach the + // vShard leader hosting this transaction's overlay, not a local replica. + txn_id: Some(txn_id), }; // Overlay teardown is not a write — no WAL record, no write version. if let Err(e) = dp.dispatch_no_wal(task, None).await { diff --git a/nodedb/src/control/server/shared/session/lifecycle.rs b/nodedb/src/control/server/shared/session/lifecycle.rs index bbe6a1087..1fa46f6b4 100644 --- a/nodedb/src/control/server/shared/session/lifecycle.rs +++ b/nodedb/src/control/server/shared/session/lifecycle.rs @@ -36,7 +36,7 @@ pub fn run_begin( .load(std::sync::atomic::Ordering::Acquire); ddl_buffer::activate(); sessions - .begin(addr, snapshot_lsn, snapshot_epoch) + .begin(addr, snapshot_lsn, snapshot_epoch, state.node_id) .map_err(|msg| crate::Error::BadRequest { detail: msg.to_owned(), }) diff --git a/nodedb/src/control/server/shared/session/read_set.rs b/nodedb/src/control/server/shared/session/read_set.rs index 89ef2c38a..335eb76aa 100644 --- a/nodedb/src/control/server/shared/session/read_set.rs +++ b/nodedb/src/control/server/shared/session/read_set.rs @@ -155,7 +155,7 @@ mod tests { let sessions = SessionStore::new(); let a = addr(); sessions.ensure_session(a); - sessions.begin(&a, Lsn::new(5), 0).expect("begin"); + sessions.begin(&a, Lsn::new(5), 0, 0).expect("begin"); (sessions, a) } diff --git a/nodedb/src/control/server/shared/session/tests.rs b/nodedb/src/control/server/shared/session/tests.rs index 056f3643c..f6cfdd896 100644 --- a/nodedb/src/control/server/shared/session/tests.rs +++ b/nodedb/src/control/server/shared/session/tests.rs @@ -13,13 +13,13 @@ fn transaction_lifecycle() { assert_eq!(store.transaction_state(&addr), TransactionState::Idle); - store.begin(&addr, crate::types::Lsn::new(1), 0).unwrap(); + store.begin(&addr, crate::types::Lsn::new(1), 0, 0).unwrap(); assert_eq!(store.transaction_state(&addr), TransactionState::InBlock); store.commit(&addr).unwrap(); assert_eq!(store.transaction_state(&addr), TransactionState::Idle); - store.begin(&addr, crate::types::Lsn::new(1), 0).unwrap(); + store.begin(&addr, crate::types::Lsn::new(1), 0, 0).unwrap(); store.fail_transaction(&addr); assert_eq!(store.transaction_state(&addr), TransactionState::Failed); @@ -320,7 +320,7 @@ async fn multi_vshard_rollback_to_savepoint_rewinds_each_vshard() { let store = SessionStore::new(); let addr: std::net::SocketAddr = "127.0.0.1:5201".parse().unwrap(); store.ensure_session(addr); - store.begin(&addr, Lsn::new(1), 0).unwrap(); + store.begin(&addr, Lsn::new(1), 0, 0).unwrap(); let tenant = TenantId::new(1); let dp = RecordingDp::default(); diff --git a/nodedb/src/control/server/shared/session/transaction.rs b/nodedb/src/control/server/shared/session/transaction.rs index b9d5ee795..fa208ee8e 100644 --- a/nodedb/src/control/server/shared/session/transaction.rs +++ b/nodedb/src/control/server/shared/session/transaction.rs @@ -37,6 +37,7 @@ impl SessionStore { addr: &SocketAddr, current_lsn: Lsn, snapshot_epoch: u64, + node_id: u64, ) -> Result<(), &'static str> { self.write_session(addr, |session| match session.tx_state { TransactionState::Idle => { @@ -44,7 +45,14 @@ impl SessionStore { session.tx_snapshot_lsn = Some(current_lsn); session.tx_snapshot_epoch = Some(snapshot_epoch); session.tx_read_set.clear(); - session.tx_id = Some(TxnId::new(NEXT_TXN_ID.fetch_add(1, Ordering::Relaxed))); + // Transaction ids travel cross-node (staging overlays live on + // the OWNING shard), so they must be globally unique: node id in + // the high 16 bits, process-local counter in the low 48. Two + // coordinators can never mint the same id, so their overlays + // never collide on a shard that hosts both transactions. + session.tx_id = Some(TxnId::new( + (node_id << 48) | (NEXT_TXN_ID.fetch_add(1, Ordering::Relaxed) & 0xFFFF_FFFF_FFFF), + )); session.tx_vshards.clear(); Ok(()) } diff --git a/nodedb/src/control/server/shuffle/producer_hook.rs b/nodedb/src/control/server/shuffle/producer_hook.rs index 02ad97130..0eec8e90f 100644 --- a/nodedb/src/control/server/shuffle/producer_hook.rs +++ b/nodedb/src/control/server/shuffle/producer_hook.rs @@ -98,6 +98,8 @@ impl RegistryShuffleProducer { database_id: req.database_id, deadline_remaining_ms: req.deadline_remaining_ms, trace_id: req.trace_id, + // Shuffle produce is not transactional. + txn_id: None, descriptor_versions: req .descriptor_versions .iter() diff --git a/nodedb/src/event/cdc/consume.rs b/nodedb/src/event/cdc/consume.rs index 3980621c2..68b25c973 100644 --- a/nodedb/src/event/cdc/consume.rs +++ b/nodedb/src/event/cdc/consume.rs @@ -250,6 +250,8 @@ pub async fn consume_remote( tenant_id: crate::types::TenantId::new(tenant_id), trace_id: nodedb_types::TraceId::generate(), database_id: nodedb_types::id::DatabaseId::DEFAULT, + // Event-plane consume is autocommit. + txn_id: None, }; let query_ctx = crate::control::planner::context::QueryContext::for_state(state); diff --git a/nodedb/src/event/topic/publish.rs b/nodedb/src/event/topic/publish.rs index a2d582a1b..a10d04437 100644 --- a/nodedb/src/event/topic/publish.rs +++ b/nodedb/src/event/topic/publish.rs @@ -162,6 +162,8 @@ pub async fn publish_remote( tenant_id: crate::types::TenantId::new(tenant_id), trace_id: nodedb_types::TraceId::generate(), database_id: nodedb_types::id::DatabaseId::DEFAULT, + // Event-plane publish is autocommit. + txn_id: None, }; let query_ctx = crate::control::planner::context::QueryContext::for_state(state); From 30c79167530289b8e35d8e203a150815dd960089 Mon Sep 17 00:00:00 2001 From: Securifi Dev Date: Thu, 9 Jul 2026 05:07:50 +0000 Subject: [PATCH 2/2] fix(cluster,session): reject oversized node_id; drop unrelated Cargo.lock churn MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review follow-ups on the F1 branch: - Blocker 2: TxnId packs node_id into the high 16 bits, but node_id is a u64 and validate() only rejected zero, so node_id >= 65536 silently lost its high bits (65537 collided with 1) — exactly the collision the packing exists to prevent. ClusterSettings::validate now rejects node_id >= 1<<16 with a typed Error::Config; a debug-and-release shift never masks. Tests cover the boundary (65535 ok, 65536/65537/2^20 rejected). - Refresh the now-stale NEXT_TXN_ID doc comment to describe the low-48-bits / node-id-high-16 split and the cross-node global-uniqueness invariant. - Revert the fluxbench source/checksum churn in Cargo.lock (leaked local [patch.crates-io] resolution state) back to origin/main — orthogonal to F1. Blocker 1 (non-leader single-shard COMMIT applies locally + WALs outside Raft) is confirmed and NOT yet fixed here — its fix mechanism is an open question raised on the PR (no atomic transaction-batch Raft entry exists, so the choice between N replicated proposals vs a new batch entry type has single-shard-commit atomicity implications). Held pending that decision. Co-Authored-By: Claude Fable 5 --- Cargo.lock | 18 ------ nodedb/src/config/server/cluster.rs | 55 +++++++++++++++++++ .../server/shared/session/transaction.rs | 10 ++-- 3 files changed, 61 insertions(+), 22 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 165b5b770..f223c0ee8 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2272,8 +2272,6 @@ checksum = "28a80e3145d8ad11ba0995949bbcf48b9df2be62772b3d351ef017dff6ecb853" [[package]] name = "fluxbench" version = "0.1.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "323b4f98b988b9779c6aa3f93781ef7421e36e73a5c610497857b2c01398acdc" dependencies = [ "fluxbench-cli", "fluxbench-core", @@ -2288,8 +2286,6 @@ dependencies = [ [[package]] name = "fluxbench-cli" version = "0.1.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "42072e01658451938143f620a08f13b308e30a7fe41f3de87a79f351c5753328" dependencies = [ "anyhow", "chrono", @@ -2317,8 +2313,6 @@ dependencies = [ [[package]] name = "fluxbench-core" version = "0.1.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "780bc18df5d1aa766ab5394389fb989c51d5d6ac3b00e51340f97c6510268edf" dependencies = [ "fluxbench-ipc", "inventory", @@ -2331,8 +2325,6 @@ dependencies = [ [[package]] name = "fluxbench-ipc" version = "0.1.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7c44dcc5fe5705f4933b7768a4fdb552ab97c098d4d3e49b93096152527b89fe" dependencies = [ "rkyv 0.8.16", "thiserror 2.0.18", @@ -2341,8 +2333,6 @@ dependencies = [ [[package]] name = "fluxbench-logic" version = "0.1.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ab4809eb2b1c33ee46c51cc28ee5d8c8ea8dbc4977151af00a03a1d1f5d487ab" dependencies = [ "evalexpr", "fluxbench-core", @@ -2357,8 +2347,6 @@ dependencies = [ [[package]] name = "fluxbench-macros" version = "0.1.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "71729e5cb94b94beb14f6a9319a129e6cf3e116112f56578182edadeaaddc79a" dependencies = [ "proc-macro2", "quote", @@ -2368,8 +2356,6 @@ dependencies = [ [[package]] name = "fluxbench-report" version = "0.1.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "02bf8934a0c315e390d691e3c9993e52b59ed88ecb8993d882a69721559342d0" dependencies = [ "chrono", "fluxbench-core", @@ -2383,8 +2369,6 @@ dependencies = [ [[package]] name = "fluxbench-stats" version = "0.1.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bec71a4355a6558d47beea9abda1fda67580e124fefdb9e93f0aaab4cc872063" dependencies = [ "rand 0.8.6", "rayon", @@ -4133,8 +4117,6 @@ dependencies = [ [[package]] name = "nexar" version = "0.1.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "dac038912d8e0f36309f3368c0980377730edd4ad5d082627227277cdd870c3a" dependencies = [ "bytes", "crossbeam-queue", diff --git a/nodedb/src/config/server/cluster.rs b/nodedb/src/config/server/cluster.rs index 73fb0f0f4..814226e96 100644 --- a/nodedb/src/config/server/cluster.rs +++ b/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:?}" + ); + } + } +} diff --git a/nodedb/src/control/server/shared/session/transaction.rs b/nodedb/src/control/server/shared/session/transaction.rs index fa208ee8e..b5124e205 100644 --- a/nodedb/src/control/server/shared/session/transaction.rs +++ b/nodedb/src/control/server/shared/session/transaction.rs @@ -13,10 +13,12 @@ use super::read_set::ReadSetEntry; use super::state::{SavepointEntry, TransactionState}; use super::store::SessionStore; -/// Global monotonic counter minting `TxnId`s across all sessions on this -/// shard. Unique per shard for the lifetime of the process — sufficient -/// for keying the per-transaction staging overlay, which is scoped to a -/// single shard's in-memory state. +/// Process-local monotonic counter forming the low 48 bits of a `TxnId`. +/// The high 16 bits carry the node id (see [`SessionStore::begin`]), so the +/// full id is globally unique across the cluster — a transaction's staging +/// overlay lives on whichever shard leader owns the write, which may be a +/// different node than the coordinator, so two coordinators must never mint +/// the same id. static NEXT_TXN_ID: AtomicU64 = AtomicU64::new(1); impl SessionStore {