diff --git a/nodedb/src/control/fail_gate.rs b/nodedb/src/control/fail_gate.rs index 62d87f273..bd7c9ebbf 100644 --- a/nodedb/src/control/fail_gate.rs +++ b/nodedb/src/control/fail_gate.rs @@ -93,6 +93,15 @@ pub(crate) async fn after_calvin_stamp( } } +/// The recon submit gate of node `node_id`, `recon::before_submit`: a +/// dependent predicate write has settled its materialized sums and +/// reconnoitred its rows, and parks before it builds and submits its Calvin +/// transaction. A test changes a row the settlement read, then releases the +/// gate. +pub(crate) async fn before_recon_submit(node_id: u64) { + wait(FailScope::Node(node_id), "recon::before_submit").await; +} + /// The Calvin scheduler's redo gate on node `node_id`: after a committed /// slice's redo is resolved and before the data-group leader proposes it. /// Named per collection, `calvin::before_redo_propose::`. diff --git a/nodedb/src/control/planner/calvin/dependent_recon.rs b/nodedb/src/control/planner/calvin/dependent_recon.rs index ade4f953c..6487347ab 100644 --- a/nodedb/src/control/planner/calvin/dependent_recon.rs +++ b/nodedb/src/control/planner/calvin/dependent_recon.rs @@ -19,6 +19,8 @@ //! [`DependentReconOutcome`]; each protocol synthesises its own command tags //! from the original task list AFTER this returns `Ok`. +use std::collections::{BTreeMap, BTreeSet}; + use crate::Error; use crate::control::cluster::calvin::executor::ollp::error::OllpError; use crate::control::planner::calvin::preexec::PreexecScan; @@ -31,18 +33,17 @@ use crate::control::planner::implicit_edges::{ EdgeUpdateCtx, append_implicit_edge_delete_tasks, append_implicit_edge_update_tasks, }; use crate::control::server::shared::session::StatementDeadline; +use crate::control::server::shared::session::read_set::ReadSetEntry; use crate::control::state::SharedState; use crate::types::{DatabaseId, TenantId, TraceId}; use nodedb_physical::physical_plan::OllpPredictedEdge; -use super::dependent_recon_finish::finish_committed; -use super::dependent_recon_node_edges::{ - NodeDeletePlan, node_delete_tasks, planned_edge_deletes, reconnoitre, -}; -use super::dependent_recon_plan::{inject_ollp_predicted_edges, inject_ollp_surrogates}; -use super::dependent_recon_predicate::{ - EdgeLifecycle, classify_edge_lifecycle, extract_bulk_predicate_info, +use super::dependent_recon_bulks::{ + BulkPrediction, BulkRecon, reconnoitre_all, with_bulk_predictions, }; +use super::dependent_recon_finish::finish_committed; +use super::dependent_recon_node_edges::{NodeDeletePlan, node_delete_tasks, planned_edge_deletes}; +use super::dependent_recon_predicate::EdgeLifecycle; use nodedb_physical::physical_task::PhysicalTask; /// Protocol-neutral result of [`dispatch_dependent_edge_recon`]. @@ -151,12 +152,19 @@ pub fn is_edge_recon_plan(plan: &nodedb_physical::physical_plan::PhysicalPlan) - /// guard and edges all home on one vShard is a legitimate one-vShard /// transaction, and so is a contended single-collection predicate write /// routed here by the write-admission gate. +/// +/// `reads` are the images materialized-sum settlement folded the tasks' +/// cross-shard balance deltas from. Every attempt carries them unchanged in +/// its `TxClass`, so a source row that moved after settlement aborts the +/// transaction. A fresh reconnaissance cannot refresh them: the shipped +/// deltas stay the ones computed from the settled images. pub async fn dispatch_authorized_dependent_edge_recon( state: &SharedState, authorized: crate::control::server::shared::authorization::AuthorizedTaskSet, identity: &crate::control::security::identity::AuthenticatedIdentity, tenant_id: TenantId, database_id: DatabaseId, + reads: &[ReadSetEntry], ) -> crate::Result { let tasks = authorized .into_tasks() @@ -171,6 +179,7 @@ pub async fn dispatch_authorized_dependent_edge_recon( tenant_id, database_id, user_roles, + reads, ) .await } @@ -185,7 +194,8 @@ pub(crate) async fn dispatch_dependent_edge_recon( database_id: DatabaseId, user_roles: Vec, ) -> crate::Result { - dispatch_dependent_edge_recon_inner(state, tasks, None, tenant_id, database_id, user_roles) + // The caller settled no materialized sum, so it has no images to carry. + dispatch_dependent_edge_recon_inner(state, tasks, None, tenant_id, database_id, user_roles, &[]) .await } @@ -196,11 +206,12 @@ async fn dispatch_dependent_edge_recon_inner( tenant_id: TenantId, database_id: DatabaseId, user_roles: Vec, + reads: &[ReadSetEntry], ) -> crate::Result { // A TRUNCATE empties the collection and tombstones its edges on every // vShard, in one transaction. if let Some(truncated) = - super::edge_truncate::dispatch_truncate(state, &tasks, identity, tenant_id).await? + super::edge_truncate::dispatch_truncate(state, &tasks, identity, tenant_id, reads).await? { return Ok(DependentReconOutcome { tasks_dispatched: tasks.len() as u64, @@ -231,31 +242,29 @@ async fn dispatch_dependent_edge_recon_inner( // submits + awaits the assignment/completion via the local registry and, on // a `PredictionDrift` abort verdict, runs a FRESH pre-execution scan // (`rescan`) before resubmitting with the fresh prediction. - let dep_task = tasks - .iter() - .find(|t| is_dependent_predicate(&t.plan)) - .ok_or_else(|| Error::Internal { - detail: "dependent-edge recon dispatch invoked without a dependent-predicate task" + let bulks = BulkRecon::of(&tasks)?; + if bulks.len() > 1 { + // A request of several predicate writes runs as an implicit + // transaction, so each reads the rows the one before it wrote. + return Err(Error::Internal { + detail: "dependent-edge recon dispatch invoked with several predicate writes" .to_owned(), - })?; + }); + } + let first = bulks.first().ok_or_else(|| Error::Internal { + detail: "dependent-edge recon dispatch invoked without a dependent-predicate task" + .to_owned(), + })?; let orc = orchestrator.ok_or(Error::SequencerUnavailable)?; - // Hoisted across the retry loop so both `submit` and `rescan` can borrow them. - let (dep_collection, dep_filter_bytes) = extract_bulk_predicate_info(&dep_task.plan); - let pred_class = predicate_class_for_filters(&dep_filter_bytes, &dep_collection); + // The first predicate write keys the backoff class and the collection whose + // matched rows the transaction locks by row. + let primary = first.collection.clone(); + let pred_class = predicate_class_for_filters(&first.filters, &primary); - let edge_mode = classify_edge_lifecycle(&dep_task.plan)?; - - // Initial reconnaissance — the first prediction the loop submits. - let initial_predicted = reconnoitre( - state, - tenant_id, - database_id, - &dep_collection, - dep_filter_bytes.clone(), - &edge_mode, - ) - .await?; + // Initial reconnaissance, one scan per predicate write: the first + // prediction the loop submits. + let initial_predicted = reconnoitre_all(state, tenant_id, database_id, &bulks).await?; // One deadline for every attempt: the statement's. let deadline = StatementDeadline::of_statement(state).tokio_instant(); @@ -267,13 +276,11 @@ async fn dispatch_dependent_edge_recon_inner( // via `submit_calvin_routed_assign` (returning the leader-assigned // `RoutedAssignment`). This lets a non-leader coordinator drive the // dependent (OLLP) cross-shard write to completion. - let submit = |predicted: &PreexecScan| { - let surrogates = predicted.surrogates.clone(); - let edges = predicted.edges.clone(); - let node_edges = predicted.node_edges.clone(); + let submit = |predicted: &Vec| { + let scans = predicted.clone(); let tasks = &tasks; - let dep_collection = &dep_collection; - let edge_mode = &edge_mode; + let bulks = &bulks; + let primary = &primary; let user_roles = &user_roles; async move { // Implicit-edge reconciliation: a matched edge document @@ -295,67 +302,82 @@ async fn dispatch_dependent_edge_recon_inner( // `OllpRetryRequired` on any divergence BEFORE writing. The // existing retry loop then re-scans and re-derives fresh edges. // - // `predicted_edges` mirrors the recon `edges` (which carry the + // Each predicate write carries the rows and edges of its own scan. + // `predicted_edges` mirrors a scan's `edges` (which carry the // surrogate of each edge doc) into the plan-carried wire type. - let predicted_edges: Vec = - edges.iter().map(OllpPredictedEdge::from).collect(); - let mut edge_tasks: Vec = Vec::new(); // Node-delete guards run before every other task of the // transaction, so each compares the edges the transaction found. let mut guard_tasks: Vec = Vec::new(); - match edge_mode { - EdgeLifecycle::Delete { .. } => { - append_implicit_edge_delete_tasks( - state, - &mut edge_tasks, - tenant_id, - database_id, - TraceId::ZERO, - dep_collection, - &edges, - ) - .await - .map_err(|e| OllpError::Terminal(Box::new(e)))?; - // Each deleted row is a graph node: its incident edges in - // this collection are tombstoned in this transaction. - let (guards, deletes) = node_delete_tasks( - state, - tenant_id, - database_id, - NodeDeletePlan { - collection: dep_collection, - guarded: &node_edges, - deleted: &node_edges, - already_deleted: &planned_edge_deletes(&edge_tasks), - }, - ) - .await - .map_err(|e| OllpError::Terminal(Box::new(e)))?; - guard_tasks = guards; - edge_tasks.extend(deletes); - } - EdgeLifecycle::Update(overrides) => { - append_implicit_edge_update_tasks( - EdgeUpdateCtx { + // One guard per node, even when several writes reach the same node. + let mut guarded: BTreeSet<(String, String)> = BTreeSet::new(); + let mut predictions: BTreeMap = BTreeMap::new(); + for (bulk, scan) in bulks.iter().zip(&scans) { + let collection = &bulk.collection; + let predicted_edges: Vec = + scan.edges.iter().map(OllpPredictedEdge::from).collect(); + predictions.insert(bulk.index, (scan.surrogates.clone(), predicted_edges)); + match &bulk.edge_mode { + EdgeLifecycle::Delete { .. } => { + append_implicit_edge_delete_tasks( state, + &mut edge_tasks, tenant_id, database_id, - trace_id: TraceId::ZERO, - collection: dep_collection, - }, - &mut edge_tasks, - &edges, - &surrogates, - overrides, - ) - .await - .map_err(|e| OllpError::Terminal(Box::new(e)))?; + TraceId::ZERO, + collection, + &scan.edges, + ) + .await + .map_err(|e| OllpError::Terminal(Box::new(e)))?; + // Each deleted row is a graph node: its incident edges in + // this collection are tombstoned in this transaction. + let fresh: Vec<_> = scan + .node_edges + .iter() + .filter(|n| guarded.insert((collection.clone(), n.node.clone()))) + .cloned() + .collect(); + let (guards, deletes) = node_delete_tasks( + state, + tenant_id, + database_id, + NodeDeletePlan { + collection, + guarded: &fresh, + deleted: &scan.node_edges, + already_deleted: &planned_edge_deletes(&edge_tasks), + }, + ) + .await + .map_err(|e| OllpError::Terminal(Box::new(e)))?; + guard_tasks.extend(guards); + edge_tasks.extend(deletes); + } + EdgeLifecycle::Update(overrides) => { + append_implicit_edge_update_tasks( + EdgeUpdateCtx { + state, + tenant_id, + database_id, + trace_id: TraceId::ZERO, + collection, + }, + &mut edge_tasks, + &scan.edges, + &scan.surrogates, + overrides, + ) + .await + .map_err(|e| OllpError::Terminal(Box::new(e)))?; + } } } + // The edge tasks are appended after the predicate writes, so the + // predictions, keyed by position in `tasks`, reach only those. let mut submission_tasks: Vec = guard_tasks; - submission_tasks.extend(tasks.iter().cloned()); + submission_tasks.extend(with_bulk_predictions(tasks, &predictions)); submission_tasks.extend(edge_tasks); if let Some(identity) = identity { let emitter = crate::control::security::audit::ArcAuditEmitter( @@ -376,31 +398,29 @@ async fn dispatch_dependent_edge_recon_inner( .collect(); } + #[cfg(feature = "failpoints")] + crate::control::fail_gate::before_recon_submit(state.node_id).await; orc.submit_with_retry_via( pred_class, tenant_id, || { - let modified_tasks: Vec = submission_tasks + // The rows the transaction locks by row are those matched + // in the first write's collection. The other collections + // keep their whole-collection lock. + let rows: Vec = bulks .iter() - .map(|t| { - let mut t = t.clone(); - // `inject_ollp_surrogates` / `_predicted_edges` - // only touch the original BulkUpdate/BulkDelete - // doc tasks (no-ops on any other plan); the - // edge-delete tasks are appended AFTER, so they - // are untouched. The tx_builder can run more than - // once, so clone the predicted sets per task. - inject_ollp_surrogates(&mut t.plan, surrogates.clone()); - inject_ollp_predicted_edges(&mut t.plan, predicted_edges.clone()); - t - }) + .zip(&scans) + .filter(|(bulk, _)| bulk.collection == *primary) + .flat_map(|(_, scan)| scan.surrogates.iter().copied()) + .collect::>() + .into_iter() .collect(); let mut tx_class = build_single_vshard_predicted_tx_class( - &modified_tasks, + &submission_tasks, tenant_id, - dep_collection, - &surrogates, - &[], + primary, + &rows, + reads, ) .map_err(|e| OllpError::Terminal(Box::new(e)))?; if let Some(class) = tx_class.as_mut() { @@ -421,16 +441,7 @@ async fn dispatch_dependent_edge_recon_inner( }; // `rescan`: FRESH reconnaissance on each drift abort. - let rescan = || { - reconnoitre( - state, - tenant_id, - database_id, - &dep_collection, - dep_filter_bytes.clone(), - &edge_mode, - ) - }; + let rescan = || reconnoitre_all(state, tenant_id, database_id, &bulks); let (completed_txn, ack_results) = match run_dependent_with_retry(DependentRetryArgs { registry, diff --git a/nodedb/src/control/planner/calvin/dependent_recon_bulks.rs b/nodedb/src/control/planner/calvin/dependent_recon_bulks.rs new file mode 100644 index 000000000..558f4a792 --- /dev/null +++ b/nodedb/src/control/planner/calvin/dependent_recon_bulks.rs @@ -0,0 +1,175 @@ +// SPDX-License-Identifier: BUSL-1.1 + +//! The predicate writes of one dependent submission. +//! +//! A submission can hold several `BulkUpdate`/`BulkDelete` tasks, for example +//! the statements of one request. Each task has its own collection, filters +//! and edge lifecycle, so each gets its own reconnaissance scan and carries +//! only the rows and edges that scan found. + +use std::collections::BTreeMap; + +use nodedb_physical::physical_plan::OllpPredictedEdge; +use nodedb_physical::physical_task::PhysicalTask; + +use crate::control::planner::calvin::is_dependent_predicate; +use crate::control::planner::calvin::preexec::PreexecScan; +use crate::control::state::SharedState; +use crate::types::{DatabaseId, TenantId}; + +use super::dependent_recon_node_edges::reconnoitre; +use super::dependent_recon_plan::{inject_ollp_predicted_edges, inject_ollp_surrogates}; +use super::dependent_recon_predicate::{ + EdgeLifecycle, classify_edge_lifecycle, extract_bulk_predicate_info, +}; + +/// What one predicate write carries into the submission: the rows it matched +/// and the implicit edges of those rows. +pub(super) type BulkPrediction = (Vec, Vec); + +/// One predicate write of a submission and the inputs of its reconnaissance. +/// They are constant across retries. +pub(super) struct BulkRecon { + /// Position of the task in the submission's task list. + pub(super) index: usize, + pub(super) collection: String, + pub(super) filters: Vec, + pub(super) edge_mode: EdgeLifecycle, +} + +impl BulkRecon { + /// Every predicate write of `tasks`, in task order. + pub(super) fn of(tasks: &[PhysicalTask]) -> crate::Result> { + tasks + .iter() + .enumerate() + .filter(|(_, task)| is_dependent_predicate(&task.plan)) + .map(|(index, task)| { + let (collection, filters) = extract_bulk_predicate_info(&task.plan); + Ok(Self { + index, + collection, + filters, + edge_mode: classify_edge_lifecycle(&task.plan)?, + }) + }) + .collect() + } +} + +/// The reconnaissance scan of every predicate write, in the order of `bulks`. +pub(super) async fn reconnoitre_all( + state: &SharedState, + tenant_id: TenantId, + database_id: DatabaseId, + bulks: &[BulkRecon], +) -> crate::Result> { + let mut scans = Vec::with_capacity(bulks.len()); + for bulk in bulks { + scans.push( + reconnoitre( + state, + tenant_id, + database_id, + &bulk.collection, + bulk.filters.clone(), + &bulk.edge_mode, + ) + .await?, + ); + } + Ok(scans) +} + +/// A copy of `tasks` in which the predicate write at each index of +/// `predictions` carries its own rows and edges. +pub(super) fn with_bulk_predictions( + tasks: &[PhysicalTask], + predictions: &BTreeMap, +) -> Vec { + let mut out = tasks.to_vec(); + for (index, (surrogates, edges)) in predictions { + if let Some(task) = out.get_mut(*index) { + inject_ollp_surrogates(&mut task.plan, surrogates.clone()); + inject_ollp_predicted_edges(&mut task.plan, edges.clone()); + } + } + out +} + +#[cfg(test)] +mod tests { + use super::*; + use nodedb_physical::physical_plan::{DocumentOp, PhysicalPlan}; + + fn bulk_delete(collection: &str) -> PhysicalTask { + PhysicalTask { + tenant_id: TenantId::new(1), + vshard_id: crate::types::VShardId::new(0), + database_id: DatabaseId::DEFAULT, + plan: PhysicalPlan::Document(DocumentOp::BulkDelete { + collection: nodedb_types::QualifiedCollection::new(DatabaseId::DEFAULT, collection), + filters: Vec::new(), + returning: None, + ollp_predicted_surrogates: None, + ollp_predicted_edges: None, + rls_filters: Vec::new(), + rls_write_check: nodedb_types::RlsWriteCheck::pending_injection(), + resolved_sum_targets: Vec::new(), + declared_primary_key: None, + }), + post_set_op: nodedb_physical::physical_task::PostSetOp::None, + txn_id: None, + } + } + + fn carried(task: &PhysicalTask) -> (Option>, Option>) { + match &task.plan { + PhysicalPlan::Document(DocumentOp::BulkDelete { + ollp_predicted_surrogates, + ollp_predicted_edges, + .. + }) => ( + ollp_predicted_surrogates.clone(), + ollp_predicted_edges.clone(), + ), + _ => (None, None), + } + } + + #[test] + fn each_predicate_write_receives_its_own_prediction() { + let edge = OllpPredictedEdge { + surrogate: 3, + from: "a".into(), + to: "z".into(), + label: None, + }; + let tasks = vec![bulk_delete("orders"), bulk_delete("users")]; + let predictions = BTreeMap::from([ + (0, (vec![3, 5], vec![edge.clone()])), + (1, (vec![9], Vec::new())), + ]); + + let out = with_bulk_predictions(&tasks, &predictions); + + assert_eq!(carried(&out[0]), (Some(vec![3, 5]), Some(vec![edge]))); + assert_eq!(carried(&out[1]), (Some(vec![9]), Some(Vec::new()))); + } + + #[test] + fn a_predicate_write_is_found_at_its_task_position() { + let tasks = vec![bulk_delete("orders"), bulk_delete("users")]; + + let bulks = BulkRecon::of(&tasks).expect("classify the predicate writes"); + + let found: Vec<_> = bulks + .iter() + .map(|bulk| (bulk.index, bulk.collection.as_str())) + .collect(); + assert_eq!(found.len(), 2); + assert_eq!(found[0].0, 0); + assert_eq!(found[1].0, 1); + assert!(found[0].1.ends_with("orders") && found[1].1.ends_with("users")); + } +} diff --git a/nodedb/src/control/planner/calvin/edge_truncate.rs b/nodedb/src/control/planner/calvin/edge_truncate.rs index 232f6bbad..21d103289 100644 --- a/nodedb/src/control/planner/calvin/edge_truncate.rs +++ b/nodedb/src/control/planner/calvin/edge_truncate.rs @@ -34,6 +34,7 @@ use crate::control::server::dispatch_utils::{ use crate::control::server::shared::clone_write::{ CloneCheckedOutcome, InterceptAndAuthorizeParams, intercept_and_authorize, }; +use crate::control::server::shared::session::read_set::ReadSetEntry; use crate::control::state::SharedState; use crate::types::{DatabaseId, TenantId, TraceId, VShardId}; @@ -56,6 +57,10 @@ pub(super) struct TruncateOutcome { /// Calvin transaction when an edge was ever written into it. `None` when /// `tasks` holds no TRUNCATE. /// +/// Every other task of the request, such as the settled balance deltas of a +/// cross-shard materialized sum, joins the transaction, and `reads` guard +/// them. The transaction then runs even when the collection has no edge. +/// /// With `identity`, the TRUNCATE passes the clone-write gate and /// authorization. Without it, the caller already authorized it. pub(super) async fn dispatch_truncate( @@ -63,6 +68,7 @@ pub(super) async fn dispatch_truncate( tasks: &[PhysicalTask], identity: Option<&AuthenticatedIdentity>, tenant_id: TenantId, + reads: &[ReadSetEntry], ) -> crate::Result> { let Some(task) = tasks.iter().find(|task| is_truncate(&task.plan)) else { return Ok(None); @@ -76,6 +82,13 @@ pub(super) async fn dispatch_truncate( // stays true or false until the TRUNCATE's outcome. let edge_bearing = collection_is_edge_bearing(state, task.tenant_id, task.database_id, collection.as_str())?; + let others: Vec = tasks + .iter() + .filter(|other| !is_truncate(&other.plan)) + .cloned() + .collect(); + // A request that ships balance deltas needs the transaction to carry them. + let in_transaction = edge_bearing || !others.is_empty(); // The clone-write gate and authorization of the rows' truncate. The // lease holds until the transaction's outcome. @@ -95,7 +108,7 @@ pub(super) async fn dispatch_truncate( }) .await? { - CloneCheckedOutcome::Handled(response) if !edge_bearing => { + CloneCheckedOutcome::Handled(response) if !in_transaction => { return Ok(Some(TruncateOutcome { apply_result: Some(response), })); @@ -108,7 +121,7 @@ pub(super) async fn dispatch_truncate( ), }); } - CloneCheckedOutcome::Proceed(checked) if !edge_bearing => { + CloneCheckedOutcome::Proceed(checked) if !in_transaction => { let response = dispatch_authorized_durable_write(state, checked, TraceId::ZERO).await?; crate::control::local_dispatch::reject_data_plane_error(&response)?; @@ -122,7 +135,7 @@ pub(super) async fn dispatch_truncate( } } } - None if !edge_bearing => { + None if !in_transaction => { let response = dispatch_rows_plan(state, task).await?; crate::control::local_dispatch::reject_data_plane_error(&response)?; return Ok(Some(TruncateOutcome { @@ -132,15 +145,20 @@ pub(super) async fn dispatch_truncate( None => (task.clone(), None), }; - let edge_shares = edge_share_tasks(&rows_task, &collection); + let edge_shares = if edge_bearing { + edge_share_tasks(&rows_task, &collection) + } else { + Vec::new() + }; let edge_shares = match identity { Some(identity) => authorize(state, identity, edge_shares)?, None => edge_shares, }; - let mut submission = Vec::with_capacity(edge_shares.len() + 1); + let mut submission = Vec::with_capacity(edge_shares.len() + others.len() + 1); submission.push(rows_task); submission.extend(edge_shares); - let tx_class = build_static_tx_class(&submission, tenant_id, &[])?; + submission.extend(others); + let tx_class = build_static_tx_class(&submission, tenant_id, reads)?; let apply_result = submit_calvin_routed(state, tx_class).await?; Ok(Some(TruncateOutcome { apply_result })) } diff --git a/nodedb/src/control/planner/calvin/mod.rs b/nodedb/src/control/planner/calvin/mod.rs index efd331f29..f5fea84bb 100644 --- a/nodedb/src/control/planner/calvin/mod.rs +++ b/nodedb/src/control/planner/calvin/mod.rs @@ -3,6 +3,7 @@ pub mod abort_error; pub mod cross_shard_mode; pub mod dependent_recon; +mod dependent_recon_bulks; pub mod dependent_recon_crdt; mod dependent_recon_finish; pub mod dependent_recon_node_edges; @@ -20,6 +21,7 @@ pub mod read_dependent; pub mod reservation; pub mod retry_budget; pub mod retry_loop; +pub mod settling_writes; pub mod submit; pub mod tx_class; pub mod txn_predict; @@ -44,6 +46,7 @@ pub use edge_sequencing::{ pub use explain::calvin_explain_preamble; pub use predicate::predicate_class_for_filters; pub use retry_loop::{DependentOutcome, DependentRetryArgs, run_dependent_with_retry}; +pub use settling_writes::refuse_several_settling_writes; pub use submit::{ RoutedAssignment, submit_and_await_calvin, submit_and_await_calvin_until, submit_calvin_routed, submit_calvin_routed_assign, submit_calvin_routed_write, diff --git a/nodedb/src/control/planner/calvin/preexec.rs b/nodedb/src/control/planner/calvin/preexec.rs index 24a54c6b4..0479f1561 100644 --- a/nodedb/src/control/planner/calvin/preexec.rs +++ b/nodedb/src/control/planner/calvin/preexec.rs @@ -90,7 +90,7 @@ pub enum RowIdentityRead<'a> { /// row's identity, in surrogate order, when the scan read it. `node_edges` /// holds the live edges incident on each matched row's node, when the caller /// read them. Edge order is irrelevant; surrogates remain sorted. -#[derive(Default)] +#[derive(Clone, Default)] pub struct PreexecScan { pub surrogates: Vec, pub edges: Vec, diff --git a/nodedb/src/control/planner/calvin/settling_writes.rs b/nodedb/src/control/planner/calvin/settling_writes.rs new file mode 100644 index 000000000..9141846e6 --- /dev/null +++ b/nodedb/src/control/planner/calvin/settling_writes.rs @@ -0,0 +1,78 @@ +// SPDX-License-Identifier: BUSL-1.1 + +//! The request guard for writes that move one materialized-sum balance. + +use std::collections::BTreeSet; + +use nodedb_physical::physical_task::PhysicalTask; + +use super::is_dependent_predicate; +use crate::Error; + +/// Refuse a request in which a predicate write and another write move the +/// same materialized-sum balance row. Each write settles its delta from the +/// committed rows, so a later write cannot see the rows an earlier one +/// changes and the balance would drift from the sum of the source. +pub fn refuse_several_settling_writes(tasks: &[PhysicalTask]) -> crate::Result<()> { + use nodedb_physical::physical_plan::{DocumentOp, PhysicalPlan}; + if !tasks.iter().any(|task| is_dependent_predicate(&task.plan)) { + return Ok(()); + } + let mut targets = BTreeSet::new(); + for task in tasks { + if let PhysicalPlan::Document(DocumentOp::ApplyBalanceDelta { + collection, + document_id, + .. + }) = &task.plan + && !targets.insert((collection.to_string(), document_id.clone())) + { + return Err(Error::BadRequest { + detail: "several writes of one request move the same materialized-sum balance \ + and one is a predicate write; run them as separate statements or in a \ + transaction block" + .to_owned(), + }); + } + } + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + use nodedb_physical::physical_plan::{DocumentOp, PhysicalPlan}; + + use crate::types::{DatabaseId, TenantId, VShardId}; + + fn task(plan: PhysicalPlan) -> PhysicalTask { + PhysicalTask { + tenant_id: TenantId::new(1), + vshard_id: VShardId::new(0), + database_id: DatabaseId::DEFAULT, + plan, + post_set_op: nodedb_physical::physical_task::PostSetOp::None, + txn_id: None, + } + } + + fn bulk_delete(collection: &str) -> PhysicalTask { + task(PhysicalPlan::Document(DocumentOp::BulkDelete { + collection: nodedb_types::QualifiedCollection::new(DatabaseId::DEFAULT, collection), + filters: Vec::new(), + returning: None, + ollp_predicted_surrogates: None, + ollp_predicted_edges: None, + rls_filters: Vec::new(), + rls_write_check: nodedb_types::RlsWriteCheck::pending_injection(), + resolved_sum_targets: Vec::new(), + declared_primary_key: None, + })) + } + + #[test] + fn plain_predicate_writes_are_not_refused() { + let tasks = vec![bulk_delete("orders"), bulk_delete("users")]; + assert!(refuse_several_settling_writes(&tasks).is_ok()); + } +} diff --git a/nodedb/src/control/planner/sql_plan_convert/scan/search.rs b/nodedb/src/control/planner/sql_plan_convert/scan/search.rs index 3e3b86a71..3d67163e9 100644 --- a/nodedb/src/control/planner/sql_plan_convert/scan/search.rs +++ b/nodedb/src/control/planner/sql_plan_convert/scan/search.rs @@ -182,6 +182,18 @@ fn build_array_prefilter_plan( )) } +/// Refuse a score alias that would overwrite the field a hybrid hit names its row by. +/// The hit carries the row under `doc_id`, so an alias of that name hides the row's id. +fn check_score_alias(alias: Option<&str>) -> crate::Result<()> { + match alias { + Some("doc_id") => Err(crate::Error::BadRequest { + detail: "score alias 'doc_id' names the field a hybrid hit identifies its row by" + .to_owned(), + }), + _ => Ok(()), + } +} + pub(in crate::control::planner::sql_plan_convert) fn convert_hybrid_search( p: HybridSearchParams<'_>, ) -> crate::Result> { @@ -201,6 +213,7 @@ pub(in crate::control::planner::sql_plan_convert) fn convert_hybrid_search( tenant_id, database_id, } = p; + check_score_alias(score_alias)?; let collection_key = nodedb_types::CollectionKey::from_bare(database_id, collection); let qualified_collection = nodedb_types::QualifiedCollection::new(database_id, collection); let vshard = collection_key.vshard(); @@ -251,6 +264,7 @@ pub(in crate::control::planner::sql_plan_convert) fn convert_hybrid_search_tripl tenant_id, database_id, } = p; + check_score_alias(score_alias)?; let collection_key = nodedb_types::CollectionKey::from_bare(database_id, collection); let qualified_collection = nodedb_types::QualifiedCollection::new(database_id, collection); let vshard = collection_key.vshard(); diff --git a/nodedb/src/control/server/native/dispatch/direct_ops.rs b/nodedb/src/control/server/native/dispatch/direct_ops.rs index c78caa762..62c52f77a 100644 --- a/nodedb/src/control/server/native/dispatch/direct_ops.rs +++ b/nodedb/src/control/server/native/dispatch/direct_ops.rs @@ -285,6 +285,10 @@ pub(crate) async fn handle_direct_op( // source of a cross-shard materialized sum, runs in one transaction, as // the SQL path's does. The loop there clone-checks, authorizes and // meters each task itself. + if let Err(error) = crate::control::planner::calvin::refuse_several_settling_writes(&tasks) + { + return error_to_native(seq, &error); + } if statement_needs_implicit_txn(ctx.state, &tasks) { let _request = ctx.state.tenant_request_guard(tenant_id); needs_top_level_metering = false; @@ -304,7 +308,9 @@ pub(crate) async fn handle_direct_op( // cleanup in the same Calvin transaction, as the SQL path's does. use super::edge_recon_gate::{EdgeReconResult, try_edge_recon_dispatch}; let (tasks, authorized_tasks) = - match try_edge_recon_dispatch(ctx, seq, tasks, authorized_tasks).await { + match try_edge_recon_dispatch(ctx, seq, tasks, authorized_tasks, &sum_target_reads) + .await + { EdgeReconResult::Outcome(outcome) => return outcome.into_response(), EdgeReconResult::NotFired(tasks, authorized) => (tasks, authorized), }; @@ -350,6 +356,21 @@ pub(crate) async fn handle_direct_op( DispatchClass::MultiShard { .. } ); if route_to_calvin { + // A predicate write names the rows it matched only after the + // coordinator reads them, so it takes the same route as on pgwire. + use super::edge_recon_gate::try_multi_shard_predicate_recon_dispatch; + let (tasks, authorized_tasks) = match try_multi_shard_predicate_recon_dispatch( + ctx, + seq, + tasks, + authorized_tasks, + &sum_target_reads, + ) + .await + { + EdgeReconResult::Outcome(outcome) => return outcome.into_response(), + EdgeReconResult::NotFired(tasks, authorized) => (tasks, authorized), + }; let plans: Vec = tasks.iter().map(|task| task.plan.clone()).collect(); match dispatch_authorized_tasks_to_calvin( ctx.state, diff --git a/nodedb/src/control/server/native/dispatch/edge_recon_gate.rs b/nodedb/src/control/server/native/dispatch/edge_recon_gate.rs index a042734b6..cab34b29a 100644 --- a/nodedb/src/control/server/native/dispatch/edge_recon_gate.rs +++ b/nodedb/src/control/server/native/dispatch/edge_recon_gate.rs @@ -10,7 +10,10 @@ use nodedb_types::protocol::NativeResponse; use crate::control::server::shared::authorization::AuthorizedTaskSet; use crate::control::server::shared::session::TransactionState; -use crate::control::server::shared::statement_exec::{EdgeRecon, try_edge_recon}; +use crate::control::server::shared::session::read_set::ReadSetEntry; +use crate::control::server::shared::statement_exec::{ + EdgeRecon, try_edge_recon, try_multi_shard_predicate_recon, +}; use nodedb_physical::physical_task::PhysicalTask; use super::{DispatchCtx, SqlOutcome, error_to_native}; @@ -23,11 +26,40 @@ pub(super) async fn try_edge_recon_dispatch( seq: u64, tasks: Vec, authorized: AuthorizedTaskSet, + reads: &[ReadSetEntry], ) -> EdgeReconResult { // The native `handle_begin` / `handle_commit` / `handle_rollback` drive // the same `SessionStore` state machine pgwire uses. let in_txn_block = ctx.sessions.transaction_state(ctx.peer_addr) == TransactionState::InBlock; - match try_edge_recon(ctx.state, ctx.identity, in_txn_block, tasks, authorized).await { + let routed = try_edge_recon( + ctx.state, + ctx.identity, + in_txn_block, + tasks, + authorized, + reads, + ) + .await; + outcome(ctx, seq, routed) +} + +/// Route `tasks` through the OLLP/Calvin path when the shared multi-shard +/// gate fires for them. The caller has classified the statement as +/// multi-shard and outside a transaction block. +pub(super) async fn try_multi_shard_predicate_recon_dispatch( + ctx: &DispatchCtx<'_>, + seq: u64, + tasks: Vec, + authorized: AuthorizedTaskSet, + reads: &[ReadSetEntry], +) -> EdgeReconResult { + let routed = + try_multi_shard_predicate_recon(ctx.state, ctx.identity, tasks, authorized, reads).await; + outcome(ctx, seq, routed) +} + +fn outcome(ctx: &DispatchCtx<'_>, seq: u64, routed: crate::Result) -> EdgeReconResult { + match routed { Ok(EdgeRecon::NotFired(tasks, authorized)) => EdgeReconResult::NotFired(tasks, authorized), // A RETURNING dependent write surfaces its deleted or updated rows. A // plain write reports the count its own mutation returned. diff --git a/nodedb/src/control/server/pgwire/handler/routing/calvin_dispatch.rs b/nodedb/src/control/server/pgwire/handler/routing/calvin_dispatch.rs index 0abe95fc2..da4b67fee 100644 --- a/nodedb/src/control/server/pgwire/handler/routing/calvin_dispatch.rs +++ b/nodedb/src/control/server/pgwire/handler/routing/calvin_dispatch.rs @@ -217,6 +217,7 @@ impl NodeDbPgHandler { identity, tenant_id, database_id, + reads, ) .await .map_err(|e| { diff --git a/nodedb/src/control/server/pgwire/handler/routing/execute.rs b/nodedb/src/control/server/pgwire/handler/routing/execute.rs index 3c7c9a272..438c42cf3 100644 --- a/nodedb/src/control/server/pgwire/handler/routing/execute.rs +++ b/nodedb/src/control/server/pgwire/handler/routing/execute.rs @@ -229,6 +229,7 @@ impl NodeDbPgHandler { formats: shaping.formats, }, &auth_ctx, + &sum_target_reads, ) .await? { diff --git a/nodedb/src/control/server/pgwire/handler/routing/pre_dispatch.rs b/nodedb/src/control/server/pgwire/handler/routing/pre_dispatch.rs index 0e28184be..1a5e64045 100644 --- a/nodedb/src/control/server/pgwire/handler/routing/pre_dispatch.rs +++ b/nodedb/src/control/server/pgwire/handler/routing/pre_dispatch.rs @@ -74,6 +74,7 @@ impl NodeDbPgHandler { session_id: SessionId, shaping: ResultShaping<'_>, auth: &crate::control::security::auth_context::AuthContext, + sum_target_reads: &[crate::control::server::shared::session::read_set::ReadSetEntry], ) -> PgWireResult>> { let ResultShaping { projection, @@ -107,8 +108,9 @@ impl NodeDbPgHandler { auth, projection, }, - // No settled image read to carry — this fires before materialized-sum settlement. - &[], + // The tasks are already settled: the reads stamped on the source images + // their cross-shard balances were folded from travel with them. + sum_target_reads, ) .await .map(Some) diff --git a/nodedb/src/control/server/shared/statement_exec/atomic.rs b/nodedb/src/control/server/shared/statement_exec/atomic.rs index 1528899d7..3686a598a 100644 --- a/nodedb/src/control/server/shared/statement_exec/atomic.rs +++ b/nodedb/src/control/server/shared/statement_exec/atomic.rs @@ -24,7 +24,7 @@ use nodedb_physical::physical_task::PhysicalTask; use crate::control::lease::QueryLeaseScope; use crate::control::planner::calvin::{ CrossShardTxnMode, DispatchClass, TxnDispatchPosition, classify_dispatch, - dispatch_authorized_tasks_to_calvin, read_vshards_of, + dispatch_authorized_tasks_to_calvin, read_vshards_of, refuse_several_settling_writes, }; use crate::control::security::audit::ArcAuditEmitter; use crate::control::server::response_shape::calvin_fold::{ @@ -41,7 +41,9 @@ use crate::control::server::shared::write_admission::all_writes_bufferable; use crate::control::trigger::statement_txn::in_block; use super::answer::StatementAnswer; -use super::edge_recon::{CalvinApplied, EdgeRecon, try_edge_recon}; +use super::edge_recon::{ + CalvinApplied, EdgeRecon, try_edge_recon, try_multi_shard_predicate_recon, +}; use super::error::StatementError; use super::task_loop::{PlannedStatement, StatementExec, run_implicit_statement}; @@ -95,6 +97,9 @@ pub(crate) async fn route_atomic_statement( .map_err(|error| StatementError::Error(crate::Error::from(error)))?; let in_txn_block = in_block(session); + if !in_txn_block { + refuse_several_settling_writes(&tasks).map_err(StatementError::Error)?; + } if !in_txn_block && statement_needs_implicit_txn(state, &tasks) { if !all_writes_bufferable(&tasks) { return Err(StatementError::Error(unbufferable_joined_statement())); @@ -112,17 +117,23 @@ pub(crate) async fn route_atomic_statement( return Ok(AtomicRoute::Implicit(answer)); } - let (tasks, authorized) = - match try_edge_recon(state, exec.identity, in_txn_block, tasks, authorized) - .await - .map_err(StatementError::Error)? - { - EdgeRecon::Applied(applied) => { - meter_applied(exec, &applied); - return Ok(AtomicRoute::Calvin(applied)); - } - EdgeRecon::NotFired(tasks, authorized) => (tasks, authorized), - }; + let (tasks, authorized) = match try_edge_recon( + state, + exec.identity, + in_txn_block, + tasks, + authorized, + &sum_target_reads, + ) + .await + .map_err(StatementError::Error)? + { + EdgeRecon::Applied(applied) => { + meter_applied(exec, &applied); + return Ok(AtomicRoute::Calvin(applied)); + } + EdgeRecon::NotFired(tasks, authorized) => (tasks, authorized), + }; // An autocommit statement has no session read set to widen with. The // reads to widen with are those materialized-sum settlement stamped. @@ -141,6 +152,25 @@ pub(crate) async fn route_atomic_statement( } let cross_shard_mode = session.sessions.cross_shard_txn_mode(session.session_id); if !in_txn_block && cross_shard_mode == CrossShardTxnMode::Strict { + // A predicate write names the rows it matched only after the + // coordinator reads them, so it takes the same route as on + // pgwire. + let (tasks, authorized) = match try_multi_shard_predicate_recon( + state, + exec.identity, + tasks, + authorized, + &sum_target_reads, + ) + .await + .map_err(StatementError::Error)? + { + EdgeRecon::Applied(applied) => { + meter_applied(exec, &applied); + return Ok(AtomicRoute::Calvin(applied)); + } + EdgeRecon::NotFired(tasks, authorized) => (tasks, authorized), + }; let plans = tasks.iter().map(|task| task.plan.clone()).collect(); let apply_result = dispatch_authorized_tasks_to_calvin( state, diff --git a/nodedb/src/control/server/shared/statement_exec/edge_recon.rs b/nodedb/src/control/server/shared/statement_exec/edge_recon.rs index 21a0ffb11..f6d8c0b7e 100644 --- a/nodedb/src/control/server/shared/statement_exec/edge_recon.rs +++ b/nodedb/src/control/server/shared/statement_exec/edge_recon.rs @@ -15,10 +15,11 @@ use nodedb_physical::physical_task::PhysicalTask; use crate::bridge::envelope::{PhysicalPlan, Response}; use crate::control::planner::calvin::{ - dispatch_authorized_dependent_edge_recon, plan_needs_implicit_edge_recon, + dispatch_authorized_dependent_edge_recon, is_edge_recon_plan, plan_needs_implicit_edge_recon, }; use crate::control::security::identity::AuthenticatedIdentity; use crate::control::server::shared::authorization::AuthorizedTaskSet; +use crate::control::server::shared::session::read_set::ReadSetEntry; use crate::control::state::SharedState; use crate::types::{DatabaseId, TenantId}; @@ -49,12 +50,16 @@ pub(crate) enum EdgeRecon { /// client's transaction needs a two-phase commit across its boundary. A /// catalog I/O error propagates. Misrouting on a real fault skips the edge /// cleanup. +/// +/// `reads` are the images materialized-sum settlement stamped on the tasks. +/// The coordinator carries them into the transaction it submits. pub(crate) async fn try_edge_recon( state: &Arc, identity: &AuthenticatedIdentity, in_txn_block: bool, tasks: Vec, authorized: AuthorizedTaskSet, + reads: &[ReadSetEntry], ) -> crate::Result { if in_txn_block { return Ok(EdgeRecon::NotFired(tasks, authorized)); @@ -65,6 +70,47 @@ pub(crate) async fn try_edge_recon( else { return Ok(EdgeRecon::NotFired(tasks, authorized)); }; + run_recon(state, identity, tasks, authorized, database_id, reads).await +} + +/// Route `tasks` through the OLLP/Calvin coordinator when one of them is a +/// predicate write and the statement spans several vShards, outside a +/// transaction block. +/// +/// The caller has classified the statement as multi-shard. A predicate write +/// that reaches Calvin must carry the rows it matched, which only the +/// coordinator reads. A shipped balance of a materialized sum puts a +/// single-collection write on two vShards, so the edge-bearing gate does not +/// see it. +/// +/// `reads` are the images settlement folded that balance from. The cross-shard +/// delta is fixed at planning, so the transaction must abort when one of those +/// images moved before it applies. +pub(crate) async fn try_multi_shard_predicate_recon( + state: &Arc, + identity: &AuthenticatedIdentity, + tasks: Vec, + authorized: AuthorizedTaskSet, + reads: &[ReadSetEntry], +) -> crate::Result { + let Some(database_id) = tasks + .iter() + .find(|task| is_edge_recon_plan(&task.plan)) + .map(|task| task.database_id) + else { + return Ok(EdgeRecon::NotFired(tasks, authorized)); + }; + run_recon(state, identity, tasks, authorized, database_id, reads).await +} + +async fn run_recon( + state: &Arc, + identity: &AuthenticatedIdentity, + tasks: Vec, + authorized: AuthorizedTaskSet, + database_id: DatabaseId, + reads: &[ReadSetEntry], +) -> crate::Result { // Captured before the coordinator consumes the tasks, so the answer can // shape a RETURNING dependent write's rows and read a plain write's // count from the applied response. @@ -73,8 +119,9 @@ pub(crate) async fn try_edge_recon( state, authorized, identity, - tenant_id, + identity.tenant_id, database_id, + reads, ) .await?; Ok(EdgeRecon::Applied(CalvinApplied { diff --git a/nodedb/src/control/server/shared/statement_exec/mod.rs b/nodedb/src/control/server/shared/statement_exec/mod.rs index 52b0928a6..c0ebc7c16 100644 --- a/nodedb/src/control/server/shared/statement_exec/mod.rs +++ b/nodedb/src/control/server/shared/statement_exec/mod.rs @@ -14,7 +14,7 @@ pub mod task_loop; pub(crate) use answer::StatementAnswer; pub(crate) use atomic::{AtomicRoute, AtomicStatement, calvin_answer, route_atomic_statement}; pub(crate) use dispatch::authorize_one_task; -pub(crate) use edge_recon::{EdgeRecon, try_edge_recon}; +pub(crate) use edge_recon::{EdgeRecon, try_edge_recon, try_multi_shard_predicate_recon}; pub(crate) use error::StatementError; pub(crate) use task_loop::{ PlannedStatement, StatementExec, run_implicit_statement, run_statement_loop, diff --git a/nodedb/src/control/server/shared/txn_route/joined.rs b/nodedb/src/control/server/shared/txn_route/joined.rs index f0d11d6e5..c7408ab9a 100644 --- a/nodedb/src/control/server/shared/txn_route/joined.rs +++ b/nodedb/src/control/server/shared/txn_route/joined.rs @@ -6,6 +6,7 @@ use nodedb_physical::physical_plan::DocumentOp; use nodedb_physical::physical_task::PhysicalTask; use crate::bridge::envelope::PhysicalPlan; +use crate::control::planner::calvin::{is_dependent_predicate, is_write_plan}; use crate::control::state::SharedState; use crate::control::trigger::dml_hook::classify_dml_write; use crate::control::trigger::statement_txn::fires_joined_body; @@ -44,8 +45,12 @@ pub fn statement_fires_joined_body(state: &SharedState, tasks: &[PhysicalTask]) } /// Whether a statement outside a transaction block runs in an implicit -/// transaction. It does in three cases: +/// transaction. It does in four cases: /// +/// - It holds a predicate write that follows another write. A request of +/// several statements plans as one task set, and each later statement must +/// read the rows the earlier ones wrote, which only the block's staging scan +/// does. /// - It fires a joined body. /// - It is a `MERGE` into a collection an edge was ever written into. The /// MERGE then stages each removed row with its node's edge tasks, and @@ -55,12 +60,38 @@ pub fn statement_fires_joined_body(state: &SharedState, tasks: &[PhysicalTask]) /// The expansion ships that balance on an `ApplyBalanceDelta` task homed /// on the target, and COMMIT applies it with the source rows. pub fn statement_needs_implicit_txn(state: &SharedState, tasks: &[PhysicalTask]) -> bool { - statement_fires_joined_body(state, tasks) + holds_ordered_predicate_write(tasks) + || statement_fires_joined_body(state, tasks) || tasks .iter() .any(|task| merges_into_edge_bearing(state, task) || moves_cross_shard_sum(state, task)) } +/// Whether a predicate write of `tasks` must read the writes that come before +/// it in the request: it follows another predicate write or any other write. A +/// later statement matches against the rows the earlier ones wrote, which only +/// the block's staging scan sees. One SQL statement plans at most one predicate +/// task and no write ahead of it, so the position is the statement order. The +/// balance deltas of a settled sum follow the write that moved them and do not +/// count. +fn holds_ordered_predicate_write(tasks: &[PhysicalTask]) -> bool { + let mut earlier_write = false; + for task in tasks { + if is_dependent_predicate(&task.plan) && earlier_write { + return true; + } + earlier_write |= is_write_plan(&task.plan) && !is_balance_delta(&task.plan); + } + false +} + +fn is_balance_delta(plan: &PhysicalPlan) -> bool { + matches!( + plan, + PhysicalPlan::Document(DocumentOp::ApplyBalanceDelta { .. }) + ) +} + /// Whether `task` is an unexpanded `INSERT ... SELECT`, `UPDATE ... FROM` or /// `MERGE` whose written collection drives a materialized sum with a /// cross-shard target. Its orchestrator applies on the written collection's diff --git a/nodedb/tests/wire/cases/materialized_sum_cross_shard_expanders.rs b/nodedb/tests/wire/cases/materialized_sum_cross_shard_expanders.rs index 0757cee4b..e0f8d38e4 100644 --- a/nodedb/tests/wire/cases/materialized_sum_cross_shard_expanders.rs +++ b/nodedb/tests/wire/cases/materialized_sum_cross_shard_expanders.rs @@ -76,7 +76,15 @@ async fn exec(server: &TestServer, sql: &str) { /// `acc2` start empty. Posting `e0` puts 10 on `acc`, posting `e1` puts 5 on /// `acc`. pub(super) async fn sum_fixture() -> TestServer { - let server = TestServer::start_multicores(CORES).await; + sum_fixture_on(TestServer::start_multicores(CORES).await).await +} + +/// The data cores a `server` handed to [`sum_fixture_on`] must run. +#[cfg(feature = "failpoints")] +pub(super) const FIXTURE_CORES: usize = CORES; + +/// [`sum_fixture`] on a `server` the caller started with [`CORES`] cores. +pub(super) async fn sum_fixture_on(server: TestServer) -> TestServer { exec( &server, &format!( diff --git a/nodedb/tests/wire/cases/materialized_sum_cross_shard_http.rs b/nodedb/tests/wire/cases/materialized_sum_cross_shard_http.rs index 4a48c9746..30a2246c5 100644 --- a/nodedb/tests/wire/cases/materialized_sum_cross_shard_http.rs +++ b/nodedb/tests/wire/cases/materialized_sum_cross_shard_http.rs @@ -11,6 +11,10 @@ //! answers one `{"affected": n}` row: the balance task adds no count, as on //! pgwire and native. //! +//! A predicate `UPDATE` or `DELETE` of the source plans the source write and a +//! balance task on the target's vShard, so it commits through Calvin with the +//! rows it matched. Each route settles the balance the way pgwire does. +//! //! The fixture is the one the pgwire cases use: two Data Plane cores, and a //! target and source that land on different cores. @@ -175,3 +179,85 @@ async fn ws_rpc_query_moves_a_cross_shard_sum_atomically() { ); assert_balances(&server).await; } + +/// A predicate write of the two postings of `acc`: both set to 1, then both deleted. +const PREDICATE_UPDATE: &str = "UPDATE ptv_post SET amount = '1' WHERE account_id = 'acc'"; +const PREDICATE_DELETE: &str = "DELETE FROM ptv_post WHERE account_id = 'acc'"; + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn pgwire_predicate_write_settles_a_cross_shard_sum() { + let server = sum_fixture().await; + + server + .exec(PREDICATE_UPDATE) + .await + .expect("predicate update"); + assert_eq!(balance(&server, "acc").await, "2", "two postings at 1"); + + server + .exec(PREDICATE_DELETE) + .await + .expect("predicate delete"); + assert_eq!(balance(&server, "acc").await, "0", "both postings gone"); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn http_query_predicate_write_settles_a_cross_shard_sum() { + let server = sum_fixture().await; + + assert_eq!( + http_rows(server.http_port, PREDICATE_UPDATE).await, + vec![affected(2)], + "two matched postings: the balance task adds no count" + ); + assert_eq!(balance(&server, "acc").await, "2", "two postings at 1"); + + assert_eq!( + http_rows(server.http_port, PREDICATE_DELETE).await, + vec![affected(2)], + "two matched postings: the balance task adds no count" + ); + assert_eq!(balance(&server, "acc").await, "0", "both postings gone"); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn ndjson_query_predicate_write_settles_a_cross_shard_sum() { + let server = sum_fixture().await; + + assert_eq!( + ndjson_lines(server.http_port, PREDICATE_UPDATE).await, + vec![affected(2)], + "two matched postings: the balance task adds no count" + ); + assert_eq!(balance(&server, "acc").await, "2", "two postings at 1"); + + assert_eq!( + ndjson_lines(server.http_port, PREDICATE_DELETE).await, + vec![affected(2)], + "two matched postings: the balance task adds no count" + ); + assert_eq!(balance(&server, "acc").await, "0", "both postings gone"); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn ws_rpc_predicate_write_settles_a_cross_shard_sum() { + let server = sum_fixture().await; + let (mut ws, _) = + tokio_tungstenite::connect_async(format!("ws://127.0.0.1:{}/v1/ws", server.http_port)) + .await + .expect("connect the WebSocket RPC endpoint"); + + assert_eq!( + ws_result(&mut ws, 1, PREDICATE_UPDATE).await, + affected(2), + "two matched postings: the balance task adds no count" + ); + assert_eq!(balance(&server, "acc").await, "2", "two postings at 1"); + + assert_eq!( + ws_result(&mut ws, 2, PREDICATE_DELETE).await, + affected(2), + "two matched postings: the balance task adds no count" + ); + assert_eq!(balance(&server, "acc").await, "0", "both postings gone"); +} diff --git a/nodedb/tests/wire/cases/materialized_sum_recon_settlement_reads.rs b/nodedb/tests/wire/cases/materialized_sum_recon_settlement_reads.rs new file mode 100644 index 000000000..e6fdea348 --- /dev/null +++ b/nodedb/tests/wire/cases/materialized_sum_recon_settlement_reads.rs @@ -0,0 +1,196 @@ +// SPDX-License-Identifier: BUSL-1.1 + +//! A predicate write of the source of a cross-shard materialized sum, with +//! the source changed between settlement and the recon transaction. +//! +//! Settlement folds the balance delta of the write from the source images it +//! reads, and ships the delta to the target's vShard. A source row that moves +//! after that leaves the delta stale: the recon scan sees the same row +//! identity, so its prediction does not drift. Only the OCC read versions +//! settlement stamped on those images guard the delta, so the recon +//! transaction must carry them. +//! +//! The `recon::before_submit` gate parks the write after settlement and +//! reconnaissance. The test commits the source change while the write is +//! parked, then releases it. The balance must equal the source sum whether the +//! write aborts or recomputes. + +#![cfg(feature = "failpoints")] + +use std::path::{Path, PathBuf}; +use std::time::Duration; + +use super::materialized_sum_cross_shard_expanders::{FIXTURE_CORES, balance, sum_fixture_on}; +use crate::harness::TestServer; + +/// Sets both postings of `acc` (10 and 5) to 1. Settlement ships -13. +const PREDICATE_UPDATE: &str = "UPDATE ptv_post SET amount = '1' WHERE account_id = 'acc'"; +/// The concurrent change: `e0` goes from 10 to 20, and `acc` from 15 to 25. +const MOVE_SOURCE: &str = "UPDATE ptv_post SET amount = '20' WHERE id = 'e0'"; + +/// A server whose recon writes park at `recon::before_submit` until +/// `release` exists, with the sum fixture created. +async fn parked_fixture(release: &Path) -> TestServer { + let spec = format!("recon::before_submit=wait_file({})", release.display()); + let server = TestServer::start_multicores_with_failpoints(FIXTURE_CORES, &spec).await; + sum_fixture_on(server).await +} + +/// Wait until the gate reports a write parked. +async fn wait_parked(release: &Path) { + let mut marker = release.as_os_str().to_owned(); + marker.push(".parked"); + let marker = PathBuf::from(marker); + for _ in 0..600 { + if marker.exists() { + return; + } + tokio::time::sleep(Duration::from_millis(50)).await; + } + panic!("the predicate write never reached recon::before_submit"); +} + +/// The sum of the `acc` postings, read from the source rows. +async fn source_sum(server: &TestServer) -> i64 { + server + .query_text("SELECT amount FROM ptv_post WHERE account_id = 'acc'") + .await + .expect("read the source postings") + .iter() + .map(|amount| amount.parse::().expect("a numeric amount")) + .sum() +} + +/// Move the source while `parked` waits at the gate, release it, and check +/// that the balance equals the source sum. Returns how the write ended. +async fn move_source_then_release( + server: &TestServer, + release: &Path, + parked: tokio::task::JoinHandle, +) -> F { + wait_parked(release).await; + server + .exec(MOVE_SOURCE) + .await + .expect("the concurrent source change"); + assert_eq!( + balance(server, "acc").await, + "25", + "the concurrent change settles its own delta" + ); + std::fs::write(release, b"release").expect("release the parked write"); + let ended = parked.await.expect("the parked write task"); + + let sum = source_sum(server).await; + assert_eq!( + balance(server, "acc").await, + sum.to_string(), + "the balance must equal the source sum: the parked write either aborts \ + or settles again, and never applies the delta it computed from the old images" + ); + ended +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn http_recon_write_keeps_the_sum_when_a_settled_source_row_moved() { + let gate_dir = tempfile::tempdir().expect("gate tempdir"); + let release = gate_dir.path().join("release-recon"); + let server = parked_fixture(&release).await; + + let http_port = server.http_port; + let parked = tokio::spawn(async move { + reqwest::Client::new() + .post(format!("http://127.0.0.1:{http_port}/v1/query")) + .json(&serde_json::json!({ "sql": PREDICATE_UPDATE })) + .send() + .await + .map(|response| response.status()) + .map_err(|e| e.to_string()) + }); + move_source_then_release(&server, &release, parked) + .await + .expect("the HTTP answer"); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn pgwire_recon_write_keeps_the_sum_when_a_settled_source_row_moved() { + let gate_dir = tempfile::tempdir().expect("gate tempdir"); + let release = gate_dir.path().join("release-recon"); + let server = parked_fixture(&release).await; + + let (writer, _writer_handle) = server + .connect_as("nodedb", "nodedb") + .await + .expect("connect the parked writer"); + let parked = tokio::spawn(async move { + writer + .simple_query(PREDICATE_UPDATE) + .await + .map(|_| ()) + .map_err(|e| e.to_string()) + }); + // Either end is fine: the check is the balance, not the answer. + let _ = move_source_then_release(&server, &release, parked).await; +} + +/// The same predicate write as `PREDICATE_UPDATE`, over the native protocol's +/// SQL opcode. It reaches the shared `route_atomic_statement` path. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn native_predicate_write_keeps_the_sum_after_a_concurrent_change() { + let gate_dir = tempfile::tempdir().expect("gate tempdir"); + let release = gate_dir.path().join("release-recon"); + let server = parked_fixture(&release).await; + + let mut stream = + nodedb_test_support::native_harness::open_trust_session(server.native_port, "nodedb").await; + let parked = tokio::spawn(async move { + nodedb_test_support::native_harness::send_sql(&mut stream, 2, PREDICATE_UPDATE) + .await + .status + }); + // Either end is fine: the check is the balance, not the answer. + let _ = move_source_then_release(&server, &release, parked).await; +} + +/// The same predicate write through `OpCode::DocumentBulkUpdate`, which runs in +/// `handle_direct_op` and passes its own `sum_target_reads` to the recon gates. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn native_bulk_update_keeps_the_sum_after_a_concurrent_change() { + use nodedb::bridge::scan_filter::{FilterOp, ScanFilter}; + use nodedb_test_support::native_harness::{open_trust_session, send_request}; + use nodedb_types::Value; + use nodedb_types::protocol::opcodes::OpCode; + use nodedb_types::protocol::text_fields::TextFields; + + let gate_dir = tempfile::tempdir().expect("gate tempdir"); + let release = gate_dir.path().join("release-recon"); + let server = parked_fixture(&release).await; + + let filters = zerompk::to_msgpack_vec(&vec![ScanFilter { + field: "account_id".to_string(), + op: FilterOp::Eq, + value: Value::String("acc".to_string()), + clauses: Vec::new(), + expr: None, + }]) + .expect("encode the filter"); + let amount = nodedb_types::value_to_msgpack(&Value::String("1".to_string())) + .expect("encode the new amount"); + let mut stream = open_trust_session(server.native_port, "nodedb").await; + let parked = tokio::spawn(async move { + send_request( + &mut stream, + 2, + OpCode::DocumentBulkUpdate, + TextFields { + collection: Some("ptv_post".to_string()), + filters: Some(filters), + updates: Some(vec![("amount".to_string(), amount)]), + ..Default::default() + }, + ) + .await + .status + }); + let _ = move_source_then_release(&server, &release, parked).await; +} diff --git a/nodedb/tests/wire/cases/materialized_sum_same_request.rs b/nodedb/tests/wire/cases/materialized_sum_same_request.rs new file mode 100644 index 000000000..b0d78ac84 --- /dev/null +++ b/nodedb/tests/wire/cases/materialized_sum_same_request.rs @@ -0,0 +1,65 @@ +// SPDX-License-Identifier: BUSL-1.1 + +//! Two predicate writes of one materialized-sum source in one HTTP request. +//! +//! The request plans as one task set, and each write settles its balance +//! delta from the committed rows, so the second cannot see the first. The +//! request is refused whole. + +use super::materialized_sum_cross_shard_expanders::{balance, sum_fixture}; +use crate::harness::TestServer; + +/// The sum of the `amount` column of the postings of `acc`. +async fn posted_sum(server: &TestServer) -> i64 { + let rows = server + .query_text("SELECT amount FROM ptv_post WHERE account_id = 'acc'") + .await + .unwrap_or_else(|e| panic!("read the postings: {e}")); + rows.iter() + .map(|amount| { + amount + .parse::() + .unwrap_or_else(|e| panic!("amount {amount:?}: {e}")) + }) + .sum() +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn two_settling_updates_in_one_request_are_refused_and_change_nothing() { + let server = sum_fixture().await; + refused_whole( + &server, + "UPDATE ptv_post SET amount = '1' WHERE account_id = 'acc'; \ + UPDATE ptv_post SET amount = '50' WHERE id = 'e0'", + ) + .await; +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn two_predicate_updates_of_one_balance_are_refused_and_change_nothing() { + let server = sum_fixture().await; + refused_whole( + &server, + "UPDATE ptv_post SET amount = '1' WHERE account_id = 'acc'; \ + UPDATE ptv_post SET amount = '50' WHERE account_id = 'acc'", + ) + .await; +} + +async fn refused_whole(server: &TestServer, sql: &str) { + let response = reqwest::Client::new() + .post(format!("http://127.0.0.1:{}/v1/query", server.http_port)) + .json(&serde_json::json!({ "sql": sql })) + .send() + .await + .unwrap_or_else(|e| panic!("POST /v1/query: {e}")); + let status = response.status(); + let body = response.text().await.unwrap_or_default(); + + // The request is refused whole, so no write lands and the balance still + // equals the sum of the postings. + assert!(!status.is_success(), "the request must be refused: {body}"); + assert!(body.contains("separate statements"), "refusal text: {body}"); + assert_eq!(posted_sum(server).await, 15); + assert_eq!(balance(server, "acc").await, "15"); +} diff --git a/nodedb/tests/wire/cases/materialized_sum_truncate_edge_bearing.rs b/nodedb/tests/wire/cases/materialized_sum_truncate_edge_bearing.rs new file mode 100644 index 000000000..3d80c8d76 --- /dev/null +++ b/nodedb/tests/wire/cases/materialized_sum_truncate_edge_bearing.rs @@ -0,0 +1,75 @@ +// SPDX-License-Identifier: BUSL-1.1 + +//! TRUNCATE of an edge-bearing source of a materialized sum whose target +//! lives on another vShard and core. +//! +//! An edge-bearing TRUNCATE runs through the recon route as one Calvin +//! transaction. The balance delta the sum settles for the removed rows must +//! reach the target, so the balance ends equal to the remaining source sum. + +use super::materialized_sum_cross_shard_expanders::{balance, sum_fixture}; +use crate::harness::TestServer; + +const POSTINGS: &str = "ptv_post"; + +/// The sum of the `acc` postings, read from the source rows. +async fn source_sum(server: &TestServer) -> i64 { + server + .query_text(&format!( + "SELECT amount FROM {POSTINGS} WHERE account_id = 'acc'" + )) + .await + .expect("read the source postings") + .iter() + .map(|amount| amount.parse::().expect("a numeric amount")) + .sum() +} + +async fn post(server: &TestServer, sql: &str) { + let response = reqwest::Client::new() + .post(format!("http://127.0.0.1:{}/v1/query", server.http_port)) + .json(&serde_json::json!({ "sql": sql })) + .send() + .await + .unwrap_or_else(|e| panic!("POST /v1/query: {e}")); + let status = response.status(); + let body = response.text().await.unwrap_or_default(); + assert!(status.is_success(), "the request answered {status}: {body}"); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn truncate_of_a_cross_shard_sum_source_zeroes_the_balance() { + let server = sum_fixture().await; + // The source holds one edge, so its TRUNCATE takes the edge route. + server + .exec(&format!( + "GRAPH INSERT EDGE IN '{POSTINGS}' FROM 'e0' TO 'e1' TYPE 'l'" + )) + .await + .expect("write one edge into the source"); + assert_eq!(balance(&server, "acc").await, "15", "seed balance"); + + post(&server, &format!("TRUNCATE {POSTINGS}")).await; + + assert_eq!(source_sum(&server).await, 0, "the source is empty"); + assert_eq!( + balance(&server, "acc").await, + "0", + "the balance must equal the remaining source sum" + ); +} + +/// Control: the same TRUNCATE without an edge zeroes the balance. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn truncate_of_an_edge_free_cross_shard_sum_source_zeroes_the_balance() { + let server = sum_fixture().await; + + post(&server, &format!("TRUNCATE {POSTINGS}")).await; + + assert_eq!(source_sum(&server).await, 0, "the source is empty"); + assert_eq!( + balance(&server, "acc").await, + "0", + "the balance must equal the remaining source sum" + ); +} diff --git a/nodedb/tests/wire/cases/mod.rs b/nodedb/tests/wire/cases/mod.rs index 2787346bc..12b5d6d7d 100644 --- a/nodedb/tests/wire/cases/mod.rs +++ b/nodedb/tests/wire/cases/mod.rs @@ -154,6 +154,9 @@ mod kv_write_row_level_security; mod kv_write_row_level_security_atomics; mod materialized_sum_cross_shard_expanders; mod materialized_sum_cross_shard_http; +mod materialized_sum_recon_settlement_reads; +mod materialized_sum_same_request; +mod materialized_sum_truncate_edge_bearing; mod materialized_sum_txn_command_tag; mod materialized_sum_txn_predicate_writes; mod materialized_sum_txn_preimage; @@ -195,6 +198,8 @@ mod pgwire_tenant_scoping; mod pgwire_txn_control_spellings; mod pgwire_txn_overlay_teardown_reclaim; mod point_get_after_miss_then_insert; +mod predicate_write_sequence_http; +mod predicate_write_two_statements_http; mod procedural_txn_visibility; mod procedure_e2e; mod query_function_authorization; diff --git a/nodedb/tests/wire/cases/predicate_write_sequence_http.rs b/nodedb/tests/wire/cases/predicate_write_sequence_http.rs new file mode 100644 index 000000000..601057dc6 --- /dev/null +++ b/nodedb/tests/wire/cases/predicate_write_sequence_http.rs @@ -0,0 +1,220 @@ +// SPDX-License-Identifier: BUSL-1.1 + +//! One HTTP request that holds several predicate writes. +//! +//! The statements of a request plan as one task set. A later statement reads +//! the rows the earlier ones wrote, as in a transaction block, so the whole +//! request applies as one transaction. + +use nodedb_types::id::DatabaseId; + +use crate::harness::TestServer; + +/// Data Plane cores the server runs. +const CORES: usize = 2; + +fn core_of(name: &str) -> usize { + nodedb_types::CollectionKey::from_bare(DatabaseId::DEFAULT, name) + .vshard() + .as_u32() as usize + % CORES +} + +/// Two collection names that hash to different cores. +fn collections_on_two_cores() -> (String, String) { + let first = "ps_one".to_owned(); + let second = (0..) + .map(|n| format!("ps_two_{n}")) + .find(|name| core_of(name) != core_of(&first)) + .expect("a name on the other core"); + (first, second) +} + +async fn exec(server: &TestServer, sql: &str) { + server + .exec(sql) + .await + .unwrap_or_else(|e| panic!("run {sql}: {e}")); +} + +async fn post(server: &TestServer, sql: &str) { + let response = reqwest::Client::new() + .post(format!("http://127.0.0.1:{}/v1/query", server.http_port)) + .json(&serde_json::json!({ "sql": sql })) + .send() + .await + .unwrap_or_else(|e| panic!("POST /v1/query: {e}")); + let status = response.status(); + let body = response.text().await.unwrap_or_default(); + assert!(status.is_success(), "the request answered {status}: {body}"); +} + +async fn column(server: &TestServer, sql: &str) -> Vec { + server + .query_text(sql) + .await + .unwrap_or_else(|e| panic!("read {sql}: {e}")) +} + +/// A schemaless collection with one edge written into it, so its predicate +/// writes carry edge tasks. +async fn edge_bearing(server: &TestServer, name: &str) { + exec(server, &format!("CREATE COLLECTION {name}")).await; + for id in ["a", "b"] { + exec( + server, + &format!("UPSERT INTO {name} {{ id: '{id}', n: 1, tag: 'x' }}"), + ) + .await; + } + exec( + server, + &format!("GRAPH INSERT EDGE IN '{name}' FROM 'a' TO 'b' TYPE 'l'"), + ) + .await; +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn two_increments_of_one_row_in_one_request_both_apply() { + let server = TestServer::start().await; + edge_bearing(&server, "ps_inc").await; + + post( + &server, + "UPDATE ps_inc SET n = n + 1 WHERE id = 'a'; \ + UPDATE ps_inc SET n = n + 1 WHERE id = 'a'", + ) + .await; + + assert_eq!( + column(&server, "SELECT n FROM ps_inc WHERE id = 'a'").await, + ["3"] + ); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn a_second_update_matches_the_value_the_first_wrote() { + let server = TestServer::start().await; + exec(&server, "CREATE COLLECTION ps_dep").await; + exec(&server, "UPSERT INTO ps_dep { id: 'a', n: 1 }").await; + exec(&server, "UPSERT INTO ps_dep { id: 'b', n: 5 }").await; + exec( + &server, + "GRAPH INSERT EDGE IN 'ps_dep' FROM 'a' TO 'b' TYPE 'l'", + ) + .await; + + post( + &server, + "UPDATE ps_dep SET n = 2 WHERE n = 1; UPDATE ps_dep SET n = 3 WHERE n = 2", + ) + .await; + + assert_eq!( + column(&server, "SELECT n FROM ps_dep WHERE id = 'a'").await, + ["3"] + ); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn a_later_update_matches_the_tag_an_earlier_one_wrote() { + let server = TestServer::start().await; + edge_bearing(&server, "ps_chain").await; + + post( + &server, + "UPDATE ps_chain SET tag = 'moved' WHERE id = 'a'; \ + UPDATE ps_chain SET n = 7 WHERE tag = 'moved'", + ) + .await; + + assert_eq!( + column(&server, "SELECT id FROM ps_chain WHERE n = 7").await, + ["a"] + ); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn a_third_statement_reads_the_first_across_two_vshards() { + let server = TestServer::start_multicores(CORES).await; + let (one, two) = collections_on_two_cores(); + for collection in [&one, &two] { + exec( + &server, + &format!( + "CREATE COLLECTION {collection} (id TEXT PRIMARY KEY, v TEXT) \ + WITH (engine='document_strict')" + ), + ) + .await; + exec( + &server, + &format!("INSERT INTO {collection} (id, v) VALUES ('r1', 'v1'), ('r2', 'keep')"), + ) + .await; + } + + post( + &server, + &format!( + "UPDATE {one} SET v = 'v2' WHERE v = 'v1'; \ + UPDATE {two} SET v = 'w2' WHERE v = 'v1'; \ + UPDATE {one} SET v = 'v3' WHERE v = 'v2'" + ), + ) + .await; + + assert_eq!( + column(&server, &format!("SELECT v FROM {one} ORDER BY id")).await, + ["v3", "keep"] + ); + assert_eq!( + column(&server, &format!("SELECT v FROM {two} ORDER BY id")).await, + ["w2", "keep"] + ); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn a_predicate_reads_the_point_write_before_it_across_two_vshards() { + let server = TestServer::start_multicores(CORES).await; + let (one, two) = collections_on_two_cores(); + for collection in [&one, &two] { + exec( + &server, + &format!( + "CREATE COLLECTION {collection} (id TEXT PRIMARY KEY, n INT) \ + WITH (engine='document_strict')" + ), + ) + .await; + } + exec( + &server, + &format!("INSERT INTO {one} (id, n) VALUES ('x', 0)"), + ) + .await; + exec( + &server, + &format!("INSERT INTO {two} (id, n) VALUES ('y', 0)"), + ) + .await; + + post( + &server, + &format!( + "UPDATE {one} SET n = 1 WHERE id = 'x'; \ + UPDATE {one} SET n = 2 WHERE n = 1; \ + UPDATE {two} SET n = 9 WHERE id = 'y'" + ), + ) + .await; + + assert_eq!( + column(&server, &format!("SELECT n FROM {one} WHERE id = 'x'")).await, + ["2"] + ); + assert_eq!( + column(&server, &format!("SELECT n FROM {two} WHERE id = 'y'")).await, + ["9"] + ); +} diff --git a/nodedb/tests/wire/cases/predicate_write_two_statements_http.rs b/nodedb/tests/wire/cases/predicate_write_two_statements_http.rs new file mode 100644 index 000000000..15a46404c --- /dev/null +++ b/nodedb/tests/wire/cases/predicate_write_two_statements_http.rs @@ -0,0 +1,86 @@ +// SPDX-License-Identifier: BUSL-1.1 + +//! One HTTP request that holds a predicate `UPDATE` of each of two +//! collections on different Data Plane cores. +//! +//! The request plans as one task set that spans two vShards, so it commits +//! through Calvin. Each predicate write must apply to the rows its own +//! predicate matched, never to the rows the other one matched. + +use nodedb_types::id::DatabaseId; + +use crate::harness::TestServer; + +/// Data Plane cores the server runs. +const CORES: usize = 2; + +fn core_of(name: &str) -> usize { + nodedb_types::CollectionKey::from_bare(DatabaseId::DEFAULT, name) + .vshard() + .as_u32() as usize + % CORES +} + +/// Two collection names that hash to different cores. +fn collections_on_two_cores() -> (String, String) { + let first = "pw_one".to_owned(); + let second = (0..) + .map(|n| format!("pw_two_{n}")) + .find(|name| core_of(name) != core_of(&first)) + .expect("a name on the other core"); + (first, second) +} + +async fn exec(server: &TestServer, sql: &str) { + server + .exec(sql) + .await + .unwrap_or_else(|e| panic!("run {sql}: {e}")); +} + +async fn values(server: &TestServer, collection: &str) -> Vec { + server + .query_text(&format!("SELECT v FROM {collection} ORDER BY id")) + .await + .unwrap_or_else(|e| panic!("read {collection}: {e}")) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn two_predicate_updates_in_one_http_request_each_touch_their_own_rows() { + let server = TestServer::start_multicores(CORES).await; + let (one, two) = collections_on_two_cores(); + for collection in [&one, &two] { + exec( + &server, + &format!( + "CREATE COLLECTION {collection} (id TEXT PRIMARY KEY, v TEXT) \ + WITH (engine='document_strict')" + ), + ) + .await; + } + for (collection, tag) in [(&one, "one"), (&two, "two")] { + exec( + &server, + &format!("INSERT INTO {collection} (id, v) VALUES ('r1', '{tag}-old'), ('r2', 'keep')"), + ) + .await; + } + + let sql = format!( + "UPDATE {one} SET v = 'one-new' WHERE v = 'one-old'; \ + UPDATE {two} SET v = 'two-new' WHERE v = 'two-old'" + ); + let response = reqwest::Client::new() + .post(format!("http://127.0.0.1:{}/v1/query", server.http_port)) + .json(&serde_json::json!({ "sql": sql })) + .send() + .await + .unwrap_or_else(|e| panic!("POST /v1/query: {e}")); + let status = response.status(); + let body = response.text().await.unwrap_or_default(); + assert!(status.is_success(), "the request answered {status}: {body}"); + + assert_eq!(values(&server, &one).await, ["one-new", "keep"]); + assert_eq!(values(&server, &two).await, ["two-new", "keep"]); +} diff --git a/nodedb/tests/wire/cases/sql_hybrid_search.rs b/nodedb/tests/wire/cases/sql_hybrid_search.rs index a0d5b68a0..3d568fade 100644 --- a/nodedb/tests/wire/cases/sql_hybrid_search.rs +++ b/nodedb/tests/wire/cases/sql_hybrid_search.rs @@ -457,3 +457,26 @@ async fn hybrid_vector_leg_reads_the_named_column() { ); assert_eq!(rows[0][0], "x", "emb_b ranks x first: {rows:?}"); } + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn a_score_alias_named_doc_id_is_refused() { + let server = TestServer::start().await; + create_hybrid_collection(&server, "hs_docid").await; + + let result = server + .query_rows( + "SELECT id, \ + rrf_score(\ + vector_distance(embedding, ARRAY[0.1, 0.2, 0.3, 0.4]), \ + bm25_score(content, 'consensus')\ + ) AS doc_id \ + FROM hs_docid ORDER BY doc_id DESC LIMIT 5", + ) + .await; + + let err = result.expect_err("an alias named doc_id must be refused"); + assert!( + err.contains("doc_id"), + "the refusal names the field; got: {err}" + ); +} diff --git a/nodedb/tests/wire/harness/lifecycle.rs b/nodedb/tests/wire/harness/lifecycle.rs index 799837cf4..7ec022818 100644 --- a/nodedb/tests/wire/harness/lifecycle.rs +++ b/nodedb/tests/wire/harness/lifecycle.rs @@ -32,6 +32,21 @@ impl TestServer { Self::connect_and_build(spawned, dir, AuthMode::Trust).await } + /// [`Self::start_with_failpoints`] with `cores` Data Plane cores, for a + /// test whose collections must land on different cores. + #[cfg(feature = "failpoints")] + pub async fn start_multicores_with_failpoints(cores: usize, spec: &str) -> Self { + let dir = tempfile::tempdir().expect("tempdir"); + let spawned = process::spawn_with_failpoints( + dir.path(), + AuthMode::Trust, + TuningOverrides::none(), + cores, + Some(spec), + ); + Self::connect_and_build(spawned, dir, AuthMode::Trust).await + } + /// Spawn a single-core server in pgwire password mode (SCRAM-SHA-256) /// with the credential lockout policy enabled (`5` failures -> `300s`). /// The harness user `nodedb` keeps password `nodedb`.