diff --git a/CHANGELOG.md b/CHANGELOG.md index c107166fc..4b8cb8b1e 100644 --- a/CHANGELOG.md +++ b/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. diff --git a/nodedb/src/control/gateway/router.rs b/nodedb/src/control/gateway/router.rs index 762ed120b..c6e15f518 100644 --- a/nodedb/src/control/gateway/router.rs +++ b/nodedb/src/control/gateway/router.rs @@ -63,6 +63,26 @@ pub fn route_plan( strategy_fn: impl Fn(&str) -> PartitionStrategy, extractor: &dyn KeyExtractor, ) -> Result> { + // 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:?}" + ); + } + } + } } diff --git a/nodedb/src/control/server/native/dispatch/transaction.rs b/nodedb/src/control/server/native/dispatch/transaction.rs index a7c0b17cd..90cf5a69c 100644 --- a/nodedb/src/control/server/native/dispatch/transaction.rs +++ b/nodedb/src/control/server/native/dispatch/transaction.rs @@ -15,14 +15,13 @@ 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; @@ -30,11 +29,12 @@ 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> + 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 }) } } diff --git a/nodedb/tests/native_txn_commit_visibility.rs b/nodedb/tests/native_txn_commit_visibility.rs new file mode 100644 index 000000000..173c0db11 --- /dev/null +++ b/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::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 { + 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:?}" + ); +}