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

CHANGELOG.md

@@ -52,6 +52,7 @@ NodeDB 0.4.0 is a substantial distributed-correctness and durability release. It
 - **WAL replay correctness** — timeseries samples are no longer rejected during replay and a mid-record flush no longer duplicates rows on recovery; per-row surrogates are persisted and restored through columnar WAL replay; complete KV WAL replay (`incr`/`expire`/`persist`/`cas`/`field_set`/`register_index`/`drop_index`) with resolved expiry instants; graph node-label and columnar predicate UPDATE/DELETE records persisted and replayed.
 - **Query and SQL correctness** — scan and join execution no longer silently truncates rows at implicit caps; streamed scans drain every chunk; bitemporal range scans, indexed residual predicates, computed and distributed `GROUP BY`, and scalar aggregates over empty input return complete results; strict `AS OF SYSTEM TIME NULL` queries resolve the correct historical schema.
 - **Restore / backup** — WAL tombstones replicated via Raft on restore; plain-columnar rows re-issued durably rather than snapshot-installed; columnar/flushed-timeseries data and catalog propagated cluster-wide; replica multiplication and CRDT loss under RF>1 prevented.
+- **Native protocol transactions** — a row committed inside an explicit `Begin`/`Commit` over the native protocol is now visible to PK point lookups and filtered aggregates, not just full scans; the commit batch was routed to vShard 0 instead of the collection's owning vShard, and the gateway router now rejects unroutable commit meta-ops instead of silently misrouting them (#193).
 - **Session plan cache** — a repeated literal PK point lookup no longer replays a stale empty result after the same session inserted that key. Document point reads and PK-targeted document mutations are excluded from the schema-only plan cache, so a byte-identical `SELECT` reflects the session's own committed writes and the simple and extended protocols agree.
 - **Object ownership** — `DROP USER` reassigns every object the user owned to a validated administrative principal — the tenant's recorded admin, else an active `tenant_admin`, else an active superuser in that tenant — and is refused when no such principal exists, instead of repointing objects at a synthetic name that was never created and leaving the data directory unbootable. Collections inside their drop-retention window are included. Ownership records are keyed by `(object_type, database_id, tenant_id, object_name)`, so ownership of a collection no longer extends to a same-named collection in another database. The startup catalog check now repairs dangling owner references and revokes grants to removed users instead of refusing to start, so a data directory already affected by this recovers on its next boot.
 - **Tenant management** — tenant IDs allocated via a durable high-water-mark; ghost rows in `SHOW TENANTS` after `DROP TENANT` eliminated; an existence gate enforced for unknown numeric tenant IDs; `DROP`/`ALTER`/`PURGE TENANT` accept a tenant name; duplicate `CREATE TENANT` names rejected with `42710` rather than allocating a second tenant id under the same name.

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:?}"
+    );
+}