Merge remote-tracking branch 'origin/main' into fix/issues-188-193

# Conflicts:
#	nodedb/src/control/server/native/dispatch/transaction.rs

Author: farhan-syah

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/cluster/metadata_applier/audit.rs

@@ -109,22 +109,50 @@ pub(super) fn emit_ddl_audit(
     // `descriptor_kind` + `descriptor_name`.
     let _ = std::any::type_name::<StoredCollection>();
 
-    let mut log = match shared.audit.lock() {
-        Ok(g) => g,
-        Err(p) => p.into_inner(),
+    // Hand the record off rather than taking the audit mutex here.
+    //
+    // This runs on the Raft metadata apply loop — the single thread that
+    // applies committed entries for group 0. Blocking it on a process-wide
+    // mutex lets audit-log contention stall EVERY metadata apply, and with it
+    // collection materialization, DDL, and any proposer waiting on the applied
+    // index. That is a liveness bug rather than a slow path: one contended
+    // acquisition here was measured parking the loop for a full 5s propose
+    // timeout, so peer `CollectionSchema` announces never materialized and the
+    // engine writes that followed them were rejected as unknown collections.
+    //
+    // The record's payload is fully built above and needs nothing further from
+    // the apply loop, so deferring the emit keeps the compliance row (no silent
+    // drop) while letting the loop advance. Hash-chain integrity is unaffected:
+    // `record_with_auth` allocates the sequence number under the same mutex, so
+    // chain order follows lock-acquisition order exactly as it does for every
+    // other audit writer.
+    let audit = std::sync::Arc::clone(&shared.audit);
+    let emit = move || {
+        let mut log = match audit.lock() {
+            Ok(g) => g,
+            Err(p) => p.into_inner(),
+        };
+        log.record_with_auth(
+            AuditEvent::DdlChange,
+            None,
+            None,
+            "metadata_group",
+            &detail_json,
+            &AuditAuth {
+                user_id,
+                user_name,
+                session_id: String::new(),
+            },
+        );
     };
-    log.record_with_auth(
-        AuditEvent::DdlChange,
-        None,
-        None,
-        "metadata_group",
-        &detail_json,
-        &AuditAuth {
-            user_id,
-            user_name,
-            session_id: String::new(),
-        },
-    );
+    match tokio::runtime::Handle::try_current() {
+        Ok(handle) => {
+            handle.spawn_blocking(emit);
+        }
+        // No reactor (unit tests, non-Tokio callers): emit inline. There is no
+        // apply loop to protect in that context.
+        Err(_) => emit(),
+    }
 }
 
 /// Return `(descriptor_name, version_after, hlc_string)` for a

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

@@ -16,10 +16,6 @@ use nodedb_types::TraceId;
 use nodedb_types::protocol::NativeResponse;
 
 use crate::bridge::envelope::{ErrorCode, Response};
-use crate::control::gateway::RouteDecision;
-use crate::control::server::dispatch_utils::{
-    ChangeFeedOwner, SubmitWrite, WalDurability, WriteOrdering, submit_write,
-};
 use crate::control::server::shared::ddl::sqlstate::error_code_to_sqlstate;
 use crate::control::server::shared::session::{
     AbortReason, CommitOutcome, TxnDataPlane, commit, lifecycle,
@@ -28,15 +24,17 @@ use crate::control::state::SharedState;
 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,
 }
@@ -49,42 +47,21 @@ impl TxnDataPlane for NativeTxnDp<'_> {
     ) -> Pin<Box<dyn Future<Output = crate::Result<Response>> + Send + 'a>> {
         let state = self.state;
         Box::pin(async move {
-            let decision = crate::control::server::shared::session::resolve_leader(&task, state);
-            if matches!(decision, RouteDecision::Local) {
-                return submit_write(
-                    state,
-                    SubmitWrite {
-                        tenant_id: task.tenant_id,
-                        database_id: task.database_id,
-                        vshard_id: task.vshard_id,
-                        plan: task.plan,
-                        trace_id: TraceId::generate(),
-                        event_source: crate::event::EventSource::User,
-                        txn_id: task.txn_id,
-                        user_id: None,
-                        durability: WalDurability::CallerSupplied {
-                            wal_lsn,
-                            resolved_now_ms: None,
-                        },
-                        ordering: WriteOrdering::Gate,
-                        change_feed: ChangeFeedOwner::Funnel,
-                    },
-                )
-                .await
-                .map(|outcome| outcome.response);
-            }
-
-            if wal_lsn.is_some() {
-                return Err(crate::Error::Internal {
-                    detail: "non-local transaction commit must route through Calvin".into(),
-                });
-            }
-            let version_plan = task.plan.clone();
-            crate::control::server::shared::session::forward_to_leader(
+            dispatch_utils::dispatch_write_to_data_plane(
                 state,
-                decision,
-                task,
-                &version_plan,
+                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/src/control/server/shared/session/mod.rs

@@ -35,7 +35,6 @@ pub mod plan_cache;
 pub mod prepared_cache;
 
 pub use self::cross_shard_mode::{CrossShardTxnMode, parse_value as parse_cross_shard_value};
-pub(crate) use self::leader_forward::{forward_to_leader, resolve_leader};
 pub use self::outcome::{AbortReason, CommitOutcome, TxnDataPlane};
 pub use self::params::{
     is_known_pg_runtime_parameter, is_known_settable_runtime_parameter, parse_set_command,