fix(native): dispatch txn commit batch to the owning vshard

A row committed inside an explicit native-protocol transaction (Begin op,
SQL DML, Commit op) was durably applied on the wrong core: the native
COMMIT seam routed the commit's ResolveTxn/TransactionBatch meta-ops
through the gateway, discarding the task's pre-classified vshard_id. The
gateway router cannot derive a route for collection-less Meta plans and
fell back to vShard 0, so the commit batch (heap put + index maintenance)
landed on vShard 0's core while point lookups and index-assisted reads
consulted the collection's owning vShard — the row was visible to full
scans but permanently invisible to PK point lookups and filtered
aggregates, across connections (#193).

NativeTxnDp::dispatch_no_wal now always dispatches through the direct
write path with task.vshard_id, matching pgwire's dispatch_task_no_wal
and native's own pre-gateway branch. The gateway router additionally
rejects ResolveTxn/TransactionBatch plans outright instead of silently
misrouting them to vShard 0.

Fixes #193

Author: laksamanakeris

CHANGELOG.md

@@ -7,6 +7,14 @@ NodeDB uses [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ---
 
+## [Unreleased]
+
+### Fixed
+
+- **Native protocol transactions** — a row committed inside an explicit transaction over the native binary protocol (`Begin` op → SQL DML → `Commit` op) is now visible to PK point lookups and filtered aggregates, not just full scans (#193). The native COMMIT seam routed the commit's `ResolveTxn`/`TransactionBatch` meta-ops through the gateway, which cannot derive a route for collection-less meta plans and fell back to vShard 0 — durably applying the commit batch on the wrong core. Native COMMIT now dispatches with the task's pre-classified vShard, matching pgwire, and the gateway router rejects commit meta-ops outright instead of silently misrouting them. **Note:** rows committed through native-protocol transactions on affected multi-core builds were applied to vShard 0's core and remain there after upgrading — they stay visible to full scans but invisible to point lookups; re-inserting them (or dump/reload) re-homes them.
+
+---
+
 ## [0.4.0] - 2026-07-20
 
 NodeDB 0.4.0 is a substantial distributed-correctness and durability release. It adds cross-shard transactional execution and distributed query/graph processing, extends temporal and sparse-vector SQL, and hardens replication, recovery, indexing, authentication, and sync across the storage engines.

nodedb/src/control/gateway/router.rs

@@ -63,6 +63,26 @@ pub fn route_plan(
     strategy_fn: impl Fn(&str) -> PartitionStrategy,
     extractor: &dyn KeyExtractor,
 ) -> Result<Vec<TaskRoute>> {
+    // Commit-time meta-ops (ResolveTxn / TransactionBatch) carry no collection
+    // name, so their vShard cannot be derived here — the primary_vshard
+    // fallback would silently send them to vShard 0 and durably apply the
+    // commit batch on the wrong core (#193). They are dispatched with the
+    // task's pre-classified `vshard_id` (see `dispatch_single_shard`), never
+    // through the gateway.
+    {
+        use nodedb_physical::physical_plan::MetaOp;
+        if matches!(
+            &plan,
+            PhysicalPlan::Meta(MetaOp::ResolveTxn { .. } | MetaOp::TransactionBatch { .. })
+        ) {
+            return Err(crate::Error::Internal {
+                detail: "commit meta-op cannot be routed by the gateway; \
+                         dispatch it with the task's explicit vshard_id"
+                    .to_owned(),
+            });
+        }
+    }
+
     // In single-node mode every plan runs locally.
     let Some(routing) = routing else {
         let vshard_id = primary_vshard(&plan, database_id);
@@ -440,4 +460,39 @@ mod tests {
         }
         unreachable!()
     }
+
+    /// Commit-time meta-ops carry no collection name, so the router cannot
+    /// derive their vShard — silently falling back to vShard 0 durably applies
+    /// the commit batch on the wrong core (#193). They must be rejected here;
+    /// callers dispatch them with the task's pre-classified `vshard_id`.
+    #[test]
+    fn commit_meta_ops_are_rejected() {
+        use nodedb_physical::physical_plan::MetaOp;
+
+        for plan in [
+            PhysicalPlan::Meta(MetaOp::TransactionBatch {
+                plans: vec![],
+                txn_id: None,
+            }),
+            PhysicalPlan::Meta(MetaOp::ResolveTxn {
+                txn_id: nodedb_types::id::TxnId::new(7),
+                plans: vec![],
+            }),
+        ] {
+            for table in [None, Some(single_node_table())] {
+                let result = route_plan(
+                    plan.clone(),
+                    1,
+                    table.as_ref(),
+                    DatabaseId::DEFAULT,
+                    |_| PartitionStrategy::CollectionHomed,
+                    &crate::control::gateway::UnwiredKeyExtractor,
+                );
+                assert!(
+                    result.is_err(),
+                    "commit meta-op must not be routable via the gateway: {plan:?}"
+                );
+            }
+        }
+    }
 }

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

@@ -15,26 +15,26 @@ use std::pin::Pin;
 use nodedb_types::TraceId;
 use nodedb_types::protocol::NativeResponse;
 
-use crate::bridge::envelope::{ErrorCode, Payload, Response, Status};
-use crate::control::gateway::core::QueryContext as GatewayQueryContext;
+use crate::bridge::envelope::{ErrorCode, Response};
 use crate::control::server::shared::ddl::sqlstate::error_code_to_sqlstate;
 use crate::control::server::shared::session::{
     AbortReason, CommitOutcome, TxnDataPlane, commit, lifecycle,
 };
 use crate::control::state::SharedState;
-use crate::types::{Lsn, RequestId};
+use crate::types::Lsn;
 use nodedb_physical::physical_task::PhysicalTask;
 
 use super::super::super::dispatch_utils;
 use super::DispatchCtx;
 
 /// Native Data-Plane dispatch seam for the neutral transaction orchestrator.
 ///
-/// Routes a task through the cluster gateway when one is configured, otherwise
-/// through the direct SPSC dispatch path — the exact branch native COMMIT used
-/// before extraction. The gateway path synthesizes an `Ok` [`Response`] on
-/// success (carrying the first vShard payload so overlay-marker meta-ops still
-/// decode), and surfaces gateway errors as a Rust `Err`.
+/// Always dispatches through the direct SPSC write path using the task's
+/// pre-classified `vshard_id`, mirroring pgwire's `dispatch_task_no_wal`.
+/// The gateway must NOT be used here: commit-time tasks carry `MetaOp` plans
+/// (`ResolveTxn`, `TransactionBatch`) with no named collection, so the
+/// gateway's router cannot derive a route for them and falls back to
+/// vShard 0 — durably applying the commit batch on the wrong core (#193).
 pub(crate) struct NativeTxnDp<'a> {
     pub(crate) state: &'a SharedState,
 }
@@ -47,48 +47,23 @@ impl TxnDataPlane for NativeTxnDp<'_> {
     ) -> Pin<Box<dyn Future<Output = crate::Result<Response>> + Send + 'a>> {
         let state = self.state;
         Box::pin(async move {
-            match state.gateway.get() {
-                Some(gw) => {
-                    let gw_ctx = GatewayQueryContext {
-                        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 {
-                        request_id: RequestId::new(0),
-                        status: Status::Ok,
-                        attempt: 0,
-                        partial: false,
-                        payload: Payload::from_vec(payloads.into_iter().next().unwrap_or_default()),
-                        watermark_lsn: Lsn::new(0),
-                        error_code: None,
-                        read_set_valid: None,
-                        read_version_lsn: crate::types::Lsn::ZERO,
-                        write_set: Vec::new(),
-                    })
-                }
-                None => {
-                    dispatch_utils::dispatch_write_to_data_plane(
-                        state,
-                        dispatch_utils::WriteDispatch {
-                            tenant_id: task.tenant_id,
-                            database_id: task.database_id,
-                            vshard_id: task.vshard_id,
-                            plan: task.plan,
-                            trace_id: TraceId::ZERO,
-                            event_source: crate::event::EventSource::User,
-                            txn_id: None,
-                            wal_lsn,
-                            // Batch COMMIT record, not per-task WAL append — see
-                            // `dispatch_task_no_wal`'s equivalent limitation.
-                            resolved_now_ms: None,
-                        },
-                    )
-                    .await
-                }
-            }
+            dispatch_utils::dispatch_write_to_data_plane(
+                state,
+                dispatch_utils::WriteDispatch {
+                    tenant_id: task.tenant_id,
+                    database_id: task.database_id,
+                    vshard_id: task.vshard_id,
+                    plan: task.plan,
+                    trace_id: TraceId::ZERO,
+                    event_source: crate::event::EventSource::User,
+                    txn_id: None,
+                    wal_lsn,
+                    // Batch COMMIT record, not per-task WAL append — see
+                    // `dispatch_task_no_wal`'s equivalent limitation.
+                    resolved_now_ms: None,
+                },
+            )
+            .await
         })
     }
 }

nodedb/tests/native_txn_commit_visibility.rs

@@ -0,0 +1,207 @@
+// SPDX-License-Identifier: BUSL-1.1
+
+//! Issue #193: a row committed inside an explicit native-protocol transaction
+//! (OpCode::Begin → SQL INSERT → OpCode::Commit) must be visible to every
+//! read path — PK point lookups and filtered aggregates, not just full scans —
+//! on the writing connection and on fresh connections.
+//!
+//! Pre-fix, `NativeTxnDp::dispatch_no_wal` routed the commit's `MetaOp`
+//! tasks through the gateway without `task.vshard_id`; the gateway's
+//! `primary_vshard` fallback sent them to vShard 0, so the commit batch was
+//! durably applied on the wrong core. The bug needs (a) the gateway wired,
+//! as production boot does, and (b) more than one Data Plane core, so that
+//! vShard 0 and the collection's owning vShard live on different cores.
+
+mod common;
+
+use std::time::Duration;
+
+use common::native_harness::{do_handshake, read_frame, write_frame};
+use common::pgwire_harness::TestServer;
+
+use nodedb_types::id::VShardId;
+use nodedb_types::protocol::opcodes::ResponseStatus;
+use nodedb_types::protocol::text_fields::TextFields;
+use nodedb_types::protocol::{HelloFrame, NativeRequest, NativeResponse, OpCode, RequestFields};
+use nodedb_types::value::Value;
+use tokio::net::TcpStream;
+
+/// Send one request and read frames until the response for `seq` completes
+/// (non-`Partial`), accumulating streamed rows. Unlike the harness
+/// `send_request`, this cannot desync on multi-frame responses.
+async fn request_drain(
+    stream: &mut TcpStream,
+    seq: u64,
+    op: OpCode,
+    fields: TextFields,
+) -> NativeResponse {
+    let req = NativeRequest {
+        op,
+        seq,
+        fields: RequestFields::Text(fields),
+    };
+    let json = sonic_rs::to_vec(&req).expect("json encode");
+    write_frame(stream, &json).await;
+
+    let mut acc_rows: Vec<Vec<Value>> = Vec::new();
+    loop {
+        let payload = tokio::time::timeout(Duration::from_secs(5), read_frame(stream))
+            .await
+            .expect("timeout waiting for response")
+            .expect("response frame");
+        let mut resp: NativeResponse =
+            sonic_rs::from_slice(&payload).expect("json decode NativeResponse");
+        assert_eq!(resp.seq, seq, "response seq mismatch: {resp:?}");
+        if let Some(rows) = resp.rows.take() {
+            acc_rows.extend(rows);
+        }
+        if resp.status != ResponseStatus::Partial {
+            resp.rows = Some(acc_rows);
+            return resp;
+        }
+    }
+}
+
+async fn sql_drain(stream: &mut TcpStream, seq: u64, sql: &str) -> NativeResponse {
+    request_drain(
+        stream,
+        seq,
+        OpCode::Sql,
+        TextFields {
+            sql: Some(sql.into()),
+            ..Default::default()
+        },
+    )
+    .await
+}
+
+const NUM_CORES: usize = 4;
+
+/// Pick a collection name whose owning vShard maps to a core other than
+/// core 0 (round-robin: core = vshard % NUM_CORES), so a commit misrouted
+/// to vShard 0 lands on a different core than the one reads consult.
+fn collection_on_nonzero_core() -> String {
+    for i in 0..64u32 {
+        let name = format!("native_txn_vis_{i}");
+        let vshard =
+            VShardId::from_collection_in_database(nodedb::types::DatabaseId::DEFAULT, &name)
+                .as_u32();
+        if vshard as usize % NUM_CORES != 0 {
+            return name;
+        }
+    }
+    unreachable!("no candidate collection hashed off core 0");
+}
+
+async fn native_session(srv: &TestServer) -> TcpStream {
+    let addr = format!("127.0.0.1:{}", srv.native_port)
+        .parse()
+        .expect("native addr");
+    let (stream, _ack) = do_handshake(addr, &HelloFrame::current())
+        .await
+        .expect("native handshake");
+    stream
+}
+
+fn first_cell(resp: &nodedb_types::protocol::NativeResponse) -> Option<Value> {
+    resp.rows
+        .as_ref()
+        .and_then(|rows| rows.first())
+        .and_then(|r| r.first())
+        .cloned()
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn native_committed_txn_row_visible_to_point_and_filtered_reads() {
+    let server = TestServer::start_multicores(NUM_CORES).await;
+
+    // Wire the gateway exactly as production boot does (state_wiring.rs).
+    // The bug only manifests on the gateway branch of
+    // `NativeTxnDp::dispatch_no_wal`; without this, tests exercise the
+    // non-gateway fallback, which routes commits correctly.
+    {
+        use std::sync::Arc;
+        let gateway = Arc::new(nodedb::control::gateway::Gateway::new(Arc::clone(
+            &server.shared,
+        )));
+        let invalidator = Arc::new(nodedb::control::gateway::PlanCacheInvalidator::new(
+            &gateway.plan_cache,
+        ));
+        let _ = server.shared.gateway.set(gateway);
+        let _ = server.shared.gateway_invalidator.set(invalidator);
+    }
+
+    let coll = collection_on_nonzero_core();
+
+    server
+        .exec(&format!(
+            "CREATE COLLECTION {coll} (id STRING PRIMARY KEY, name STRING) \
+             WITH (engine='document_strict')"
+        ))
+        .await
+        .unwrap();
+
+    let mut stream = native_session(&server).await;
+
+    // Explicit transaction via protocol opcodes, as in the issue repro.
+    let begin = request_drain(&mut stream, 1, OpCode::Begin, TextFields::default()).await;
+    assert_eq!(begin.status, ResponseStatus::Ok, "BEGIN op: {begin:?}");
+
+    let insert = sql_drain(
+        &mut stream,
+        2,
+        &format!("INSERT INTO {coll} (id, name) VALUES ('a1', 'alpha')"),
+    )
+    .await;
+    assert_eq!(insert.status, ResponseStatus::Ok, "in-tx INSERT: {insert:?}");
+
+    let commit = request_drain(&mut stream, 3, OpCode::Commit, TextFields::default()).await;
+    assert_eq!(commit.status, ResponseStatus::Ok, "COMMIT op: {commit:?}");
+
+    // PK point lookup on the writing connection.
+    let point = sql_drain(&mut stream, 4, &format!("SELECT id FROM {coll} WHERE id = 'a1'")).await;
+    assert_ne!(point.status, ResponseStatus::Error, "point lookup: {point:?}");
+    assert_eq!(
+        first_cell(&point),
+        Some(Value::String("a1".into())),
+        "PK point lookup must see the committed row: {point:?}"
+    );
+
+    // Filtered aggregate.
+    let count = sql_drain(
+        &mut stream,
+        5,
+        &format!("SELECT count(*) FROM {coll} WHERE name = 'alpha'"),
+    )
+    .await;
+    assert_ne!(count.status, ResponseStatus::Error, "filtered count: {count:?}");
+    assert_eq!(
+        first_cell(&count),
+        Some(Value::Integer(1)),
+        "filtered count(*) must see the committed row: {count:?}"
+    );
+
+    // Full scan — control: the row is durably stored. Streams (Partial
+    // frames), so it runs last on this connection.
+    let scan = sql_drain(&mut stream, 6, &format!("SELECT id FROM {coll}")).await;
+    assert_ne!(scan.status, ResponseStatus::Error, "scan: {scan:?}");
+    assert_eq!(
+        first_cell(&scan),
+        Some(Value::String("a1".into())),
+        "full scan must see the committed row: {scan:?}"
+    );
+
+    // Fresh connection — the miss must not persist across sessions.
+    let mut fresh = native_session(&server).await;
+    let point2 = sql_drain(&mut fresh, 1, &format!("SELECT id FROM {coll} WHERE id = 'a1'")).await;
+    assert_ne!(
+        point2.status,
+        ResponseStatus::Error,
+        "fresh point lookup: {point2:?}"
+    );
+    assert_eq!(
+        first_cell(&point2),
+        Some(Value::String("a1".into())),
+        "PK point lookup on a fresh connection must see the committed row: {point2:?}"
+    );
+}