fix(control): native gateway session txn_id for in-txn SQL reads

nodedb-cluster-tests/tests/gateway_execute.rs

@@ -32,6 +32,7 @@ fn test_ctx() -> QueryContext {
         tenant_id: TenantId::new(0),
         trace_id: nodedb_types::TraceId::ZERO,
         database_id: nodedb_types::id::DatabaseId::DEFAULT,
+        txn_id: None,
     }
 }
 

nodedb-cluster-tests/tests/http_gateway_migration.rs

@@ -30,6 +30,7 @@ fn test_ctx() -> QueryContext {
         tenant_id: TenantId::new(0),
         trace_id: nodedb_types::TraceId::ZERO,
         database_id: nodedb_types::id::DatabaseId::DEFAULT,
+        txn_id: None,
     }
 }
 

nodedb-cluster-tests/tests/ilp_gateway_migration.rs

@@ -28,6 +28,7 @@ fn test_ctx() -> QueryContext {
         tenant_id: TenantId::new(1),
         trace_id: nodedb_types::TraceId::ZERO,
         database_id: nodedb_types::id::DatabaseId::DEFAULT,
+        txn_id: None,
     }
 }
 

nodedb-cluster-tests/tests/listeners_gateway_smoke.rs

@@ -36,6 +36,7 @@ fn test_ctx(_trace_id: u64) -> QueryContext {
         tenant_id: TenantId::new(0),
         trace_id: nodedb_types::TraceId::ZERO,
         database_id: nodedb_types::id::DatabaseId::DEFAULT,
+        txn_id: None,
     }
 }
 

nodedb-cluster-tests/tests/listeners_typed_not_leader.rs

@@ -73,6 +73,7 @@ fn test_ctx() -> QueryContext {
         tenant_id: TenantId::new(0),
         trace_id: nodedb_types::TraceId::ZERO,
         database_id: nodedb_types::id::DatabaseId::DEFAULT,
+        txn_id: None,
     }
 }
 

nodedb-cluster-tests/tests/native_gateway_migration.rs

@@ -29,6 +29,7 @@ fn test_ctx() -> QueryContext {
         tenant_id: TenantId::new(0),
         trace_id: nodedb_types::TraceId::ZERO,
         database_id: nodedb_types::id::DatabaseId::DEFAULT,
+        txn_id: None,
     }
 }
 

nodedb-cluster-tests/tests/pgwire_gateway_migration.rs

@@ -33,6 +33,7 @@ fn test_ctx() -> QueryContext {
         tenant_id: TenantId::new(0),
         trace_id: nodedb_types::TraceId::ZERO,
         database_id: nodedb_types::id::DatabaseId::DEFAULT,
+        txn_id: None,
     }
 }
 

nodedb-cluster-tests/tests/resp_gateway_migration.rs

@@ -26,6 +26,7 @@ fn test_ctx() -> QueryContext {
         tenant_id: TenantId::new(0),
         trace_id: nodedb_types::TraceId::ZERO,
         database_id: nodedb_types::id::DatabaseId::DEFAULT,
+        txn_id: None,
     }
 }
 

nodedb/src/control/gateway/core.rs

@@ -27,7 +27,7 @@ use tracing::{Instrument, debug, info_span};
 use crate::Error;
 use crate::control::state::SharedState;
 use crate::control::trace_export::EmitSpanParams;
-use crate::types::{DatabaseId, Lsn, TenantId, TraceId, VShardId};
+use crate::types::{DatabaseId, Lsn, TenantId, TraceId, TxnId, VShardId};
 use nodedb_physical::physical_plan::PhysicalPlan;
 
 use super::dispatcher::{DispatchRouteParams, default_deadline_ms, dispatch_route};
@@ -48,6 +48,11 @@ pub struct QueryContext {
     /// catalog lookups. Single-database deployments pass
     /// [`DatabaseId::DEFAULT`].
     pub database_id: DatabaseId,
+    /// Session-transaction id for resolving the per-transaction staging
+    /// overlay (read-your-own-writes) on local SPSC dispatch and for
+    /// forwarding on remote `ExecuteRequest`. `None` for autocommit and
+    /// non-interactive callers.
+    pub txn_id: Option<TxnId>,
 }
 
 /// The gateway: routes, dispatches, retries, and caches physical plans.
@@ -309,6 +314,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 = {
@@ -351,10 +357,7 @@ impl Gateway {
                         trace_id,
                         deadline_ms,
                         version_set: &version_set,
-                        // The gateway does not yet carry session-transaction
-                        // context across this boundary (known cross-node
-                        // in-transaction read gap), so `None`.
-                        txn_id: None,
+                        txn_id,
                     })
                     .await
                 }

nodedb/src/control/gateway/dispatcher.rs

@@ -21,7 +21,7 @@ use std::time::Duration;
 use nodedb_cluster::rpc_codec::TypedClusterError;
 
 use crate::Error;
-use crate::control::server::dispatch_utils::dispatch_to_data_plane;
+use crate::control::server::dispatch_utils::dispatch_to_data_plane_with_txn;
 use crate::control::server::result_stream::ResultStream;
 use crate::control::state::SharedState;
 use crate::types::{DatabaseId, Lsn, TenantId, TraceId, TxnId, VShardId};
@@ -57,8 +57,9 @@ pub struct DispatchOutcome {
 /// `trace_id` — distributed trace ID propagated from the client request.
 /// `deadline_ms` — remaining deadline in milliseconds.
 /// `version_set` — descriptor versions for the collections touched by the plan.
-/// `txn_id` — session-transaction context to forward to the remote executor, or
-///   `None` for non-transactional dispatch (the common case).
+/// `txn_id` — session-transaction context for local overlay resolution and
+///   remote `ExecuteRequest` forwarding, or `None` for non-transactional
+///   dispatch (the common case).
 pub struct DispatchRouteParams<'a> {
     pub route: TaskRoute,
     pub shared: &'a Arc<SharedState>,
@@ -84,7 +85,7 @@ pub async fn dispatch_route(params: DispatchRouteParams<'_>) -> Result<DispatchO
     } = params;
     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 {
@@ -197,21 +198,26 @@ pub async fn dispatch_route_stream(
 }
 
 /// Local dispatch via SPSC bridge.
+///
+/// Carries `txn_id` so the Data Plane can resolve this session transaction's
+/// staging overlay (read-your-own-writes) for in-block SQL and direct ops.
 async fn dispatch_local(
     route: TaskRoute,
     shared: &Arc<SharedState>,
     tenant_id: TenantId,
     database_id: DatabaseId,
     trace_id: TraceId,
+    txn_id: Option<TxnId>,
 ) -> Result<DispatchOutcome, Error> {
     let vshard_id = VShardId::new(route.vshard_id);
-    let resp = dispatch_to_data_plane(
+    let resp = dispatch_to_data_plane_with_txn(
         shared,
         tenant_id,
         database_id,
         vshard_id,
         route.plan,
         trace_id,
+        txn_id,
     )
     .await?;
     Ok(DispatchOutcome {

nodedb/src/control/planner/calvin/preexec.rs

@@ -126,6 +126,7 @@ pub async fn run_preexec_scan(
             tenant_id,
             trace_id: TraceId::ZERO,
             database_id,
+            txn_id: None,
         };
         let payloads =
             gateway

nodedb/src/control/scatter_gather.rs

@@ -345,6 +345,7 @@ pub async fn coordinate_cross_shard_hop(
                     tenant_id: crate::types::TenantId::new(tenant_id_u64),
                     trace_id: TraceId::generate(),
                     database_id,
+                    txn_id: None,
                 };
 
                 // Build a fresh QueryContext per traversal using cloned inputs

nodedb/src/control/server/exchange/gather.rs

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

nodedb/src/control/server/exchange/resolve/exchange.rs

@@ -169,6 +169,7 @@ async fn resolve_exchange(
                         tenant_id,
                         trace_id,
                         database_id,
+                        txn_id: None,
                     };
                     // NOTE: cluster mode does not yet thread `txn_id` through
                     // `gateway.execute_stream` — cross-node in-transaction

nodedb/src/control/server/http/routes/promql/remote.rs

@@ -90,6 +90,7 @@ pub async fn remote_write(
                     tenant_id,
                     trace_id: TraceId::generate(),
                     database_id: nodedb_types::id::DatabaseId::DEFAULT,
+                    txn_id: None,
                 };
                 gw.execute(&gw_ctx, plan).await
             }

nodedb/src/control/server/http/routes/query/materialized.rs

@@ -293,6 +293,7 @@ pub async fn query(
                         tenant_id: task.tenant_id,
                         trace_id,
                         database_id,
+                        txn_id: None,
                     };
                     gw.execute(&gw_ctx, task.plan)
                         .await

nodedb/src/control/server/http/routes/query/ndjson.rs

@@ -161,6 +161,7 @@ pub async fn query_ndjson(
                     tenant_id: task.tenant_id,
                     trace_id,
                     database_id,
+                    txn_id: None,
                 };
                 gw.execute(&gw_ctx, task.plan).await
             }

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: None,
         };
         gw.execute_stream(&ctx, child_plan).await
     } else {

nodedb/src/control/server/http/routes/ws_rpc/execute_sql.rs

@@ -228,6 +228,7 @@ pub async fn execute_sql(
                     tenant_id: task.tenant_id,
                     trace_id,
                     database_id: task.database_id,
+                    txn_id: None,
                 };
                 gw.execute(&gw_ctx, task.plan).await
             }

nodedb/src/control/server/ilp_batch.rs

@@ -151,6 +151,7 @@ async fn flush_ilp_batch_inner(
                     tenant_id,
                     trace_id: TraceId::generate(),
                     database_id: nodedb_types::id::DatabaseId::DEFAULT,
+                    txn_id: None,
                 };
                 gw.execute(&gw_ctx, plan)
                     .await

nodedb/src/control/server/native/dispatch/direct_ops.rs

@@ -525,6 +525,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)),

nodedb/src/control/server/native/dispatch/sql_gateway.rs

@@ -39,6 +39,9 @@ pub(super) async fn dispatch_task_via_gateway(
                 tenant_id,
                 trace_id: TraceId::generate(),
                 database_id,
+                // Propagate the in-block transaction id so gateway local
+                // dispatch resolves the per-txn staging overlay.
+                txn_id,
             };
             gw.execute(&gw_ctx, plan)
                 .await

nodedb/src/control/server/native/dispatch/streaming.rs

@@ -106,6 +106,7 @@ pub(crate) async fn try_open_sql_stream(
             tenant_id: task.tenant_id,
             trace_id: crate::types::TraceId::ZERO,
             database_id,
+            txn_id: task.txn_id,
         };
         gw.execute_stream(&gw_ctx, child_plan).await?
     } else {

nodedb/src/control/server/native/dispatch/transaction.rs

@@ -53,6 +53,7 @@ impl TxnDataPlane for NativeTxnDp<'_> {
                         tenant_id: task.tenant_id,
                         trace_id: TraceId::generate(),
                         database_id: task.database_id,
+                        txn_id: task.txn_id,
                     };
                     let payloads = gw.execute(&gw_ctx, task.plan).await?;
                     Ok(Response {
@@ -93,6 +94,11 @@ impl TxnDataPlane for NativeTxnDp<'_> {
 }
 
 pub(crate) fn handle_begin(ctx: &DispatchCtx<'_>, seq: u64) -> NativeResponse {
+    // OpCode::Begin can arrive before any SQL statement has created the
+    // session. `SessionStore::begin` no-ops when the peer has no entry, so
+    // ensure the session exists first (SQL "BEGIN" already does this in
+    // `sql.rs` before calling here).
+    ctx.sessions.ensure_session(*ctx.peer_addr);
     match lifecycle::run_begin(ctx.sessions, ctx.peer_addr, ctx.state) {
         Ok(()) => NativeResponse::status_row(seq, "BEGIN"),
         Err(e) => {

nodedb/src/control/server/pgwire/handler/routing/gateway_dispatch.rs

@@ -103,6 +103,7 @@ impl NodeDbPgHandler {
             tenant_id,
             trace_id: TraceId::generate(),
             database_id,
+            txn_id: None,
         };
 
         let mut responses: Vec<Response> = Vec::with_capacity(tasks.len());

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: None,
             };
             gw.execute_stream(&ctx, child_plan).await
         } else {

nodedb/src/control/server/resp/gateway_dispatch.rs

@@ -37,6 +37,7 @@ pub(super) async fn dispatch_kv(
                 tenant_id: session.tenant_id,
                 trace_id: TraceId::generate(),
                 database_id: DatabaseId::DEFAULT,
+                txn_id: None,
             };
             gw.execute(&gw_ctx, plan)
                 .await
@@ -81,6 +82,7 @@ pub(super) async fn dispatch_kv_write(
                 tenant_id: session.tenant_id,
                 trace_id: TraceId::generate(),
                 database_id: DatabaseId::DEFAULT,
+                txn_id: None,
             };
             gw.execute(&gw_ctx, plan)
                 .await

nodedb/src/event/cdc/consume.rs

@@ -250,6 +250,7 @@ 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,
+        txn_id: None,
     };
 
     let query_ctx = crate::control::planner::context::QueryContext::for_state(state);

nodedb/src/event/topic/publish.rs

@@ -162,6 +162,7 @@ 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,
+        txn_id: None,
     };
 
     let query_ctx = crate::control::planner::context::QueryContext::for_state(state);