diff --git a/docs/architecture.md b/docs/architecture.md index 3dbcd2f99..d1c811e4b 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -160,6 +160,8 @@ SET cross_shard_txn = 'strict'; SET cross_shard_txn = 'best_effort_non_atomic'; ``` +In this mode a predicate write and the shipped balance of a materialized sum commit as separate groups. A failure of one group leaves the balance unequal to the source sum. + (Bare `best_effort` is deliberately rejected; invalid values return SQLSTATE `22023`.) **Single-node deployments always run Calvin.** A node with no `[cluster]` section synthesizes a one-node cluster with its own sequencer group, so transactions spanning multiple cores (vShards) commit atomically. Uncontended single-shard point writes bypass the sequencer entirely and go directly through the relevant data-group Raft; contended or predicate/bulk writes route through the deterministic scheduler. diff --git a/nodedb-cluster-tests/tests/common_suite/cases/materialized_sum_cross_shard.rs b/nodedb-cluster-tests/tests/common_suite/cases/materialized_sum_cross_shard.rs index d860d3442..a72395e83 100644 --- a/nodedb-cluster-tests/tests/common_suite/cases/materialized_sum_cross_shard.rs +++ b/nodedb-cluster-tests/tests/common_suite/cases/materialized_sum_cross_shard.rs @@ -23,10 +23,10 @@ use nodedb::types::DatabaseId; /// Source and target, chosen for readability rather than for their hashes — the /// homing assertion below is what makes the choice meaningful. -const SOURCE: &str = "xs_entries"; -const TARGET: &str = "xs_accounts"; +pub(super) const SOURCE: &str = "xs_entries"; +pub(super) const TARGET: &str = "xs_accounts"; -fn pg_detail(e: &tokio_postgres::Error) -> String { +pub(super) fn pg_detail(e: &tokio_postgres::Error) -> String { match e.as_db_error() { Some(db) => format!("{}: {}", db.code().code(), db.message()), None => format!("{e}"), @@ -34,7 +34,7 @@ fn pg_detail(e: &tokio_postgres::Error) -> String { } /// The balance column of one `TARGET` row, as read through `client`. -async fn balance_of(client: &tokio_postgres::Client, account: &str) -> String { +pub(super) async fn balance_of(client: &tokio_postgres::Client, account: &str) -> String { let rows = client .simple_query(&format!( "SELECT balance FROM {TARGET} WHERE id = '{account}'" @@ -55,7 +55,7 @@ async fn balance(client: &tokio_postgres::Client) -> String { } /// Create the two collections and declare the binding. -async fn declare_binding(cluster: &TestCluster) { +pub(super) async fn declare_binding(cluster: &TestCluster) { cluster .exec_ddl_on_any_leader(&format!( "CREATE COLLECTION {TARGET} (id TEXT PRIMARY KEY, owner TEXT) \ @@ -196,7 +196,12 @@ async fn repeated_cross_shard_inserts_accumulate() { /// The seeding entries go in through the INSERT path, which already credits the /// target across shards; every test below then exercises a NON-insert shape /// against that settled starting total. -async fn seed(cluster: &TestCluster, account: &str, start: &str, entries: &[(&str, &str)]) { +pub(super) async fn seed( + cluster: &TestCluster, + account: &str, + start: &str, + entries: &[(&str, &str)], +) { cluster.nodes[0] .client .simple_query(&format!( diff --git a/nodedb-cluster-tests/tests/common_suite/cases/materialized_sum_gateway_occ.rs b/nodedb-cluster-tests/tests/common_suite/cases/materialized_sum_gateway_occ.rs new file mode 100644 index 000000000..8fadea9ec --- /dev/null +++ b/nodedb-cluster-tests/tests/common_suite/cases/materialized_sum_gateway_occ.rs @@ -0,0 +1,184 @@ +// SPDX-License-Identifier: BUSL-1.1 + +//! A pgwire UPDATE received on a follower keeps its OCC read versions when +//! the leader of every group it touches is one remote node. +//! +//! Node `L` leads both the source group (`xs_entries`) and the target group +//! (`xs_accounts`, holding the materialized sum). Follower `F` receives +//! `UPDATE xs_entries SET amount = '40' WHERE id = 'e1'`. Every group the +//! statement touches has the same remote leader, so placement picks the +//! gateway forward. The forward must carry the read versions the sum +//! settlement stamped on the source row. Without them, a change to the +//! source row on `L` after the settlement read leaves a stale balance delta +//! to apply. +//! +//! The fail gate `stmt::before_route` parks the statement on `F` after +//! planning and before routing. While it is parked, `L` rewrites the same +//! source row. After the release and convergence the balance must equal +//! `100 + sum(source amounts)`. The statement on `F` can commit after a +//! retry or fail with an OCC error. Both outcomes keep the sum exact. +//! +//! Requires `--features failpoints`. + +#![cfg(feature = "failpoints")] + +use std::path::PathBuf; +use std::time::Duration; + +use nodedb_test_support::fail_point::{FailAction, FailGuard}; + +use super::materialized_sum_cross_shard::{ + SOURCE, TARGET, balance_of, declare_binding, pg_detail, seed, +}; +use crate::common::cluster_harness::shared_steps::group_of; +use crate::common::cluster_harness::{GroupLeader, TestCluster, wait_for}; + +/// Opening balance of the account, before any source row counts. +const START: i64 = 100; + +/// How long each wait in the case gets. +const STEP_DEADLINE: Duration = Duration::from_secs(30); + +/// Pause between two polls. +const STEP_POLL: Duration = Duration::from_millis(20); + +/// Bring up the cluster with `nodes[0]` leading the source and target +/// groups, then seed `acc-1` at `START` with one entry `e1` of 25. +async fn cluster_led_by_first_node() -> TestCluster { + let cluster = TestCluster::spawn_three() + .await + .expect("spawn 3-node cluster"); + declare_binding(&cluster).await; + cluster.wait_for_full_apply_convergence(STEP_DEADLINE).await; + + let leader = cluster.nodes[0].node_id; + for collection in [SOURCE, TARGET] { + let group = group_of(&cluster.nodes[0], collection); + cluster.transfer_leadership(group, leader).await; + let found = cluster + .wait_for_group_leader(group, GroupLeader::Node(leader), STEP_DEADLINE) + .await; + assert_eq!( + found, leader, + "node {leader} must lead the group of {collection}" + ); + assert_eq!( + cluster.group_leader(group), + Some(leader), + "group {group} of {collection} must have node {leader} as its one leader" + ); + } + + seed(&cluster, "acc-1", &START.to_string(), &[("e1", "25")]).await; + assert_eq!( + balance_of(&cluster.nodes[0].client, "acc-1").await, + "125", + "the seeded balance must be 100 + 25" + ); + cluster +} + +/// The sum of every `SOURCE` amount, read through node `idx`. +async fn source_sum(cluster: &TestCluster, idx: usize) -> f64 { + let rows = cluster.nodes[idx] + .client + .simple_query(&format!("SELECT amount FROM {SOURCE}")) + .await + .unwrap_or_else(|e| panic!("read source amounts: {}", pg_detail(&e))); + rows.into_iter() + .filter_map(|m| match m { + tokio_postgres::SimpleQueryMessage::Row(r) => r.get(0).map(str::to_string), + _ => None, + }) + .map(|amount| { + amount + .parse::() + .unwrap_or_else(|_| panic!("amount '{amount}' must parse as a number")) + }) + .sum() +} + +/// Assert, on every node, that the balance of `acc-1` equals +/// `START + sum(source amounts)`. +async fn assert_balance_equals_source_sum(cluster: &TestCluster) { + cluster.wait_for_full_apply_convergence(STEP_DEADLINE).await; + for idx in 0..cluster.nodes.len() { + let sum = source_sum(cluster, idx).await; + let text = balance_of(&cluster.nodes[idx].client, "acc-1").await; + let balance: f64 = text + .parse() + .unwrap_or_else(|_| panic!("balance '{text}' must parse as a number")); + assert!( + (balance - (START as f64 + sum)).abs() < 1e-9, + "node {}: the balance must equal the source sum: balance {balance}, \ + expected {START} + {sum}", + cluster.nodes[idx].node_id + ); + } +} + +/// Send `statement` from a follower that parks it before routing, rewrite +/// the source row `e1` on the leader meanwhile, release it, and check that +/// the balance equals the source sum on every node. +async fn forwarded_statement_then_move_source(statement: &'static str) { + let cluster = cluster_led_by_first_node().await; + let follower = &cluster.nodes[1]; + assert_ne!( + follower.node_id, cluster.nodes[0].node_id, + "the statement must arrive on a node other than the leader" + ); + + let mut conn = follower.raw_pgwire().await; + + let gate_dir = tempfile::tempdir().expect("gate directory"); + let release: PathBuf = gate_dir.path().join("release"); + let parked = gate_dir.path().join("release.parked"); + let _gate = FailGuard::for_node( + follower.node_id, + "stmt::before_route", + FailAction::WaitForFile(release.clone()), + ); + + // The raw session panics on an ErrorResponse, so a join error is the + // statement failing with an OCC error. That outcome is allowed. + let forwarded = tokio::spawn(async move { conn.simple_query(statement).await }); + + wait_for( + "the follower's statement parks before routing", + STEP_DEADLINE, + STEP_POLL, + || parked.exists(), + ) + .await; + + cluster.nodes[0] + .client + .simple_query(&format!( + "UPDATE {SOURCE} SET amount = '100' WHERE id = 'e1'" + )) + .await + .unwrap_or_else(|e| panic!("update on the leader: {}", pg_detail(&e))); + + std::fs::write(&release, b"go").expect("release the gate"); + match forwarded.await { + Ok(_) => {} + Err(join) => eprintln!("the forwarded statement ended with an error: {join}"), + } + + assert_balance_equals_source_sum(&cluster).await; +} + +/// A point UPDATE forwarded from a follower to the one leader of all its +/// groups stays exact when the source row changes on the leader mid-statement. +#[tokio::test(flavor = "multi_thread", worker_threads = 6)] +async fn a_forwarded_update_keeps_the_balance_equal_to_the_source_sum() { + forwarded_statement_then_move_source("UPDATE xs_entries SET amount = '40' WHERE id = 'e1'") + .await; +} + +/// The same for a predicate UPDATE, whose rows the coordinator reads first. +#[tokio::test(flavor = "multi_thread", worker_threads = 6)] +async fn a_forwarded_predicate_update_keeps_the_balance_equal_to_the_source_sum() { + forwarded_statement_then_move_source("UPDATE xs_entries SET amount = '40' WHERE amount = '25'") + .await; +} diff --git a/nodedb-cluster-tests/tests/common_suite/cases/mod.rs b/nodedb-cluster-tests/tests/common_suite/cases/mod.rs index d75f1fda7..cd80f3813 100644 --- a/nodedb-cluster-tests/tests/common_suite/cases/mod.rs +++ b/nodedb-cluster-tests/tests/common_suite/cases/mod.rs @@ -116,6 +116,7 @@ mod listeners_gateway_smoke; mod listeners_typed_not_leader; mod materialized_sum_cross_core; mod materialized_sum_cross_shard; +mod materialized_sum_gateway_occ; mod materialized_sum_replication; mod metadata_floor_restart; mod metadata_replay; diff --git a/nodedb/src/control/fail_gate.rs b/nodedb/src/control/fail_gate.rs index 62d87f273..a19f75af7 100644 --- a/nodedb/src/control/fail_gate.rs +++ b/nodedb/src/control/fail_gate.rs @@ -93,6 +93,22 @@ pub(crate) async fn after_calvin_stamp( } } +/// The recon submit gate of node `node_id`, `recon::before_submit`: a +/// dependent predicate write, a TRUNCATE or a CRDT document delete has +/// settled its materialized sums and parks before it submits its Calvin +/// transaction. A predicate write has also reconnoitred its rows. 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 statement routing gate of node `node_id`, `stmt::before_route`: a +/// statement has planned and parks before the gateway forward and the Calvin +/// classification. +pub(crate) async fn before_route(node_id: u64) { + wait(FailScope::Node(node_id), "stmt::before_route").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..cec795b19 100644 --- a/nodedb/src/control/planner/calvin/dependent_recon.rs +++ b/nodedb/src/control/planner/calvin/dependent_recon.rs @@ -31,6 +31,7 @@ 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; @@ -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, @@ -215,6 +226,7 @@ async fn dispatch_dependent_edge_recon_inner( identity, tenant_id, database_id, + reads, ) .await? { @@ -239,6 +251,20 @@ async fn dispatch_dependent_edge_recon_inner( .to_owned(), })?; + // A request of several predicate writes runs as an implicit transaction, + // so each reads the rows the one before it wrote. Recon scans one write. + if tasks + .iter() + .filter(|t| is_dependent_predicate(&t.plan)) + .count() + > 1 + { + return Err(Error::Internal { + detail: "dependent-edge recon dispatch invoked with several predicate writes" + .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); @@ -376,6 +402,8 @@ 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, @@ -400,7 +428,7 @@ async fn dispatch_dependent_edge_recon_inner( tenant_id, dep_collection, &surrogates, - &[], + reads, ) .map_err(|e| OllpError::Terminal(Box::new(e)))?; if let Some(class) = tx_class.as_mut() { diff --git a/nodedb/src/control/planner/calvin/dependent_recon_crdt.rs b/nodedb/src/control/planner/calvin/dependent_recon_crdt.rs index 439930ab2..653e18ca8 100644 --- a/nodedb/src/control/planner/calvin/dependent_recon_crdt.rs +++ b/nodedb/src/control/planner/calvin/dependent_recon_crdt.rs @@ -35,6 +35,7 @@ use crate::control::cluster::calvin::executor::ollp::error::OllpError; use crate::control::security::identity::AuthenticatedIdentity; use crate::control::server::graph_dispatch::{VShardRead, read_on_vshard}; use crate::control::server::shared::session::StatementDeadline; +use crate::control::server::shared::session::read_set::ReadSetEntry; use crate::control::state::SharedState; use crate::types::VShardId; use crate::types::{DatabaseId, TenantId}; @@ -46,12 +47,14 @@ pub fn is_crdt_doc_delete(plan: &PhysicalPlan) -> bool { /// Dispatch the CRDT document deletes among `tasks` with their nodes' edge /// tombstones. `None` when `tasks` holds no CRDT document delete. +/// `reads` guard the settled balance deltas that ride in `tasks`. pub(super) async fn dispatch_crdt_doc_deletes( state: &SharedState, tasks: &[PhysicalTask], identity: Option<&AuthenticatedIdentity>, tenant_id: TenantId, database_id: DatabaseId, + reads: &[ReadSetEntry], ) -> crate::Result> { // The node every deleted document id names, bound or not. An id unbound // at planning names no stored document, so the read finds it absent and @@ -141,11 +144,13 @@ pub(super) async fn dispatch_crdt_doc_deletes( .map(|task| task.into_physical_task()) .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 tx_class = build_single_vshard_tx_class(&submission, tenant_id, &[]) + let tx_class = build_single_vshard_tx_class(&submission, tenant_id, reads) .map_err(|e| OllpError::Terminal(Box::new(e)))?; Ok(Some(tx_class)) }, diff --git a/nodedb/src/control/planner/calvin/edge_truncate.rs b/nodedb/src/control/planner/calvin/edge_truncate.rs index 232f6bbad..b13183fd8 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,22 @@ 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)?; + #[cfg(feature = "failpoints")] + crate::control::fail_gate::before_recon_submit(state.node_id).await; 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..bdd9fb2e3 100644 --- a/nodedb/src/control/planner/calvin/mod.rs +++ b/nodedb/src/control/planner/calvin/mod.rs @@ -20,6 +20,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 +45,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, ships_balance_delta}; 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/settling_writes.rs b/nodedb/src/control/planner/calvin/settling_writes.rs new file mode 100644 index 000000000..c5bc397cc --- /dev/null +++ b/nodedb/src/control/planner/calvin/settling_writes.rs @@ -0,0 +1,136 @@ +// 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; + +/// The refusal text of [`refuse_several_settling_writes`]. The wire tests +/// match it, so a rewording cannot pass unseen. +pub const SEVERAL_WRITES_REFUSAL: &str = "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"; + +/// Refuse a request in which a predicate write and another write move the +/// same materialized-sum balance. 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. +/// +/// A balance is one column of a target row. Two sum columns of one row, fed +/// by one write, are two balances and pass. +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, + column, + .. + }) = &task.plan + && !targets.insert((collection.to_string(), document_id.clone(), column.clone())) + { + return Err(Error::BadRequest { + detail: SEVERAL_WRITES_REFUSAL.to_owned(), + }); + } + } + Ok(()) +} + +/// Whether a settled request ships a materialized-sum balance delta to another +/// vShard, or reads source images its deltas were folded from. Such a request +/// commits through the deterministic scheduler, which applies the deltas with +/// the writes that moved them and checks the images against the rows then. +pub fn ships_balance_delta( + tasks: &[PhysicalTask], + reads: &[crate::control::server::shared::session::read_set::ReadSetEntry], +) -> bool { + use nodedb_physical::physical_plan::{DocumentOp, PhysicalPlan}; + !reads.is_empty() + || tasks.iter().any(|task| { + matches!( + task.plan, + PhysicalPlan::Document(DocumentOp::ApplyBalanceDelta { .. }) + ) + }) +} + +#[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, + })) + } + + fn balance_delta(document_id: &str, column: &str) -> PhysicalTask { + task(PhysicalPlan::Document(DocumentOp::ApplyBalanceDelta { + collection: nodedb_types::QualifiedCollection::new(DatabaseId::DEFAULT, "accounts"), + document_id: document_id.to_owned(), + surrogate: nodedb_types::Surrogate::new(1), + column: column.to_owned(), + delta: "5".to_owned(), + join_column: "account_id".to_owned(), + join_value: "acc".to_owned(), + declared_primary_key: None, + })) + } + + #[test] + fn two_deltas_of_one_balance_beside_a_predicate_write_are_refused() { + let tasks = vec![ + bulk_delete("postings"), + balance_delta("0a", "balance"), + balance_delta("0a", "balance"), + ]; + assert!(refuse_several_settling_writes(&tasks).is_err()); + } + + #[test] + fn two_sum_columns_of_one_row_are_not_refused() { + let tasks = vec![ + bulk_delete("postings"), + balance_delta("0a", "balance"), + balance_delta("0a", "fees"), + ]; + assert!(refuse_several_settling_writes(&tasks).is_ok()); + } + + #[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..39fef0019 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,19 @@ 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; \ +choose another alias" + .to_owned(), + }), + _ => Ok(()), + } +} + pub(in crate::control::planner::sql_plan_convert) fn convert_hybrid_search( p: HybridSearchParams<'_>, ) -> crate::Result> { @@ -201,6 +214,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 +265,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..34089d4e7 100644 --- a/nodedb/src/control/server/native/dispatch/direct_ops.rs +++ b/nodedb/src/control/server/native/dispatch/direct_ops.rs @@ -285,6 +285,12 @@ 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. + // One direct op is one write, so inside a transaction block this refuses + // only a single write that ships two deltas to one balance column. + 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 +310,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 +358,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, @@ -391,6 +414,12 @@ pub(crate) async fn handle_direct_op( doc_response = Some(resp); } } + // The buffered tasks apply at COMMIT, which checks these images + // against the rows then. A refused write leaves nothing to guard. + if in_txn_block && error.is_none() { + ctx.sessions + .record_read_entries(ctx.peer_addr, sum_target_reads); + } error .or(doc_response) .unwrap_or_else(|| NativeResponse::ok(seq)) 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..129a834db 100644 --- a/nodedb/src/control/server/pgwire/handler/routing/execute.rs +++ b/nodedb/src/control/server/pgwire/handler/routing/execute.rs @@ -222,13 +222,14 @@ impl NodeDbPgHandler { .maybe_dispatch_implicit_edge_recon( &tasks, tenant_id, - identity, - session_id, - ResultShaping { + super::calvin_dispatch::CalvinDispatchSession { + identity, + session_id, + result_formats: shaping.formats, + auth: &auth_ctx, projection: effective_schema, - formats: shaping.formats, }, - &auth_ctx, + &sum_target_reads, ) .await? { @@ -236,8 +237,14 @@ impl NodeDbPgHandler { } // Read once, ahead of the gateway gate: an in-block write must never - // forward here, or it applies durably outside the transaction. + // forward here, or it applies durably outside the transaction. A + // request that ships a settled balance delta commits through Calvin + // below: a forward applies its tasks one by one, with no check of the + // images the delta was folded from. + #[cfg(feature = "failpoints")] + crate::control::fail_gate::before_route(self.state.node_id).await; if tx_state != TransactionState::InBlock + && !crate::control::planner::calvin::ships_balance_delta(&tasks, &sum_target_reads) && let Some(responses) = self .maybe_dispatch_tasks_via_gateway( &tasks, 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..3cffafbc2 100644 --- a/nodedb/src/control/server/pgwire/handler/routing/pre_dispatch.rs +++ b/nodedb/src/control/server/pgwire/handler/routing/pre_dispatch.rs @@ -70,16 +70,10 @@ impl NodeDbPgHandler { &self, tasks: &[PhysicalTask], tenant_id: TenantId, - identity: &AuthenticatedIdentity, - session_id: SessionId, - shaping: ResultShaping<'_>, - auth: &crate::control::security::auth_context::AuthContext, + session: super::calvin_dispatch::CalvinDispatchSession<'_>, + sum_target_reads: &[crate::control::server::shared::session::read_set::ReadSetEntry], ) -> PgWireResult>> { - let ResultShaping { - projection, - formats: result_formats, - } = shaping; - let tx_state = self.sessions.transaction_state(session_id); + let tx_state = self.sessions.transaction_state(session.session_id); if tx_state == TransactionState::InBlock { return Ok(None); } @@ -100,15 +94,10 @@ impl NodeDbPgHandler { self.dispatch_calvin_multishard( tasks.to_vec(), tenant_id, - super::calvin_dispatch::CalvinDispatchSession { - identity, - session_id, - result_formats, - auth, - projection, - }, - // No settled image read to carry — this fires before materialized-sum settlement. - &[], + session, + // 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/src/control/server/shared/write_admission/route.rs b/nodedb/src/control/server/shared/write_admission/route.rs index 3435f1826..c3d154a40 100644 --- a/nodedb/src/control/server/shared/write_admission/route.rs +++ b/nodedb/src/control/server/shared/write_admission/route.rs @@ -87,6 +87,12 @@ pub fn calvin_route_keeps(plan: &PhysicalPlan, event_source: EventSource, restor /// The transaction carries `user_roles`, the roles of the identity that wrote /// the write, so every replica judges a role-guarded state transition alike. /// +/// The transaction holds one task and no settlement reads. A write that ships +/// a materialized-sum balance to another vShard is a multi-shard task set. Under +/// `Strict` that set takes the group route, which carries the reads. Only +/// `BestEffortNonAtomic` splits it into single tasks, and that mode performs no +/// cross-shard OCC. +/// /// Returns a boxed future (rather than an `async fn`) to break the async /// recursion cycle described in the module docs. pub fn route_write_to_calvin<'a>( diff --git a/nodedb/tests/wire/cases/materialized_sum_crdt_delete_settlement_reads.rs b/nodedb/tests/wire/cases/materialized_sum_crdt_delete_settlement_reads.rs new file mode 100644 index 000000000..c9bc97745 --- /dev/null +++ b/nodedb/tests/wire/cases/materialized_sum_crdt_delete_settlement_reads.rs @@ -0,0 +1,109 @@ +// SPDX-License-Identifier: BUSL-1.1 + +//! A CRDT document delete in one request with a write to the source of a +//! cross-shard materialized sum, with the source changed between settlement +//! and the transaction. +//! +//! Settlement ships the balance delta of the source write from the images it +//! reads. The CRDT delete's transaction carries that delta, so it must carry +//! the read versions that guard it. The `recon::before_submit` gate parks the +//! request after settlement. The test commits a source change while it is +//! parked, then releases it. The balance must equal the source sum whether the +//! request aborts or settles again. + +#![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; + +const CRDT: &str = "ptv_crdt_nodes"; +/// The request's source write. Settlement ships -9 for `e0` going 10 to 1. +const REQUEST_WRITE: &str = "UPDATE ptv_post SET amount = '1' WHERE id = 'e0'"; +/// 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'"; + +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..200 { + if marker.exists() { + return; + } + tokio::time::sleep(Duration::from_millis(50)).await; + } + panic!("the request never reached recon::before_submit"); +} + +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() +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn crdt_delete_beside_a_sum_source_write_keeps_the_sum_when_the_source_moved() { + let gate_dir = tempfile::tempdir().expect("gate tempdir"); + let release = gate_dir.path().join("release-crdt"); + let spec = format!("recon::before_submit=wait_file({})", release.display()); + let server = TestServer::start_multicores_with_failpoints(FIXTURE_CORES, &spec).await; + let server = sum_fixture_on(server).await; + server + .exec(&format!( + "CREATE TABLE {CRDT} (id TEXT PRIMARY KEY, name TEXT) WITH (crdt='true')" + )) + .await + .expect("create the crdt collection"); + server + .exec(&format!("INSERT INTO {CRDT} (id, name) VALUES ('a', 'A')")) + .await + .expect("insert the document"); + server + .exec(&format!( + "GRAPH INSERT EDGE IN '{CRDT}' FROM 'a' TO 'b' TYPE 'knows'" + )) + .await + .expect("insert the edge"); + + let http_port = server.http_port; + let sql = format!("DELETE FROM {CRDT} WHERE id = 'a'; {REQUEST_WRITE}"); + 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": sql })) + .send() + .await + .map(|response| response.status()) + .map_err(|e| e.to_string()) + }); + wait_parked(&release).await; + tokio::time::timeout(Duration::from_secs(30), server.exec(MOVE_SOURCE)) + .await + .expect("the concurrent source change returned while the request was parked") + .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 request"); + parked + .await + .expect("the parked request task") + .expect("the HTTP answer"); + + let sum = source_sum(&server).await; + assert_eq!( + balance(&server, "acc").await, + sum.to_string(), + "the balance must equal the source sum: the request either aborts \ + or settles again, and never applies the delta it computed from the old images" + ); +} 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_native_block_reads.rs b/nodedb/tests/wire/cases/materialized_sum_native_block_reads.rs new file mode 100644 index 000000000..1430e304f --- /dev/null +++ b/nodedb/tests/wire/cases/materialized_sum_native_block_reads.rs @@ -0,0 +1,143 @@ +// SPDX-License-Identifier: BUSL-1.1 + +//! A native session updates the source of a cross-shard materialized sum +//! inside a `BEGIN` .. `COMMIT` block, and another session changes the same +//! source row before the `COMMIT`. +//! +//! The update settles a balance delta from the source image it read. The +//! block buffers that delta, and `COMMIT` applies it with the source rows. Only +//! the OCC read versions of the settled images guard the delta, so the block +//! must carry them. A `COMMIT` that applies a delta folded from the old image +//! leaves the balance different from the source sum. +//! +//! Each test checks the balance against the source sum, not the answer of the +//! `COMMIT`. An abort with a serialization error is a valid end. +//! +//! One test drives the update through the `DocumentUpdate` opcode. The other +//! drives it through the SQL opcode. + +use nodedb_test_support::native_harness::{open_trust_session, send_request, send_sql}; +use nodedb_types::Value; +use nodedb_types::protocol::opcodes::ResponseStatus; +use nodedb_types::protocol::{NativeResponse, OpCode, TextFields}; +use tokio::net::TcpStream; + +use super::materialized_sum_cross_shard_expanders::{balance, sum_fixture}; +use crate::harness::TestServer; + +/// Sets posting `e0` from 10 to 1. Settlement ships -9 to `acc`. +const SQL_UPDATE: &str = "UPDATE ptv_post SET amount = '1' WHERE id = 'e0'"; +/// 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 native session with a running sequence number. Seq 1 is the handshake. +struct Session { + stream: TcpStream, + seq: u64, +} + +impl Session { + async fn open(server: &TestServer) -> Self { + let stream = open_trust_session(server.native_port, "nodedb").await; + Self { stream, seq: 1 } + } + + async fn sql(&mut self, sql: &str) -> NativeResponse { + self.seq += 1; + send_sql(&mut self.stream, self.seq, sql).await + } + + /// Set `amount` of posting `e0` to 1 with the `DocumentUpdate` opcode. + async fn update_e0_by_opcode(&mut self) -> NativeResponse { + self.seq += 1; + let amount = nodedb_types::value_to_msgpack(&Value::String("1".into())).expect("encode"); + send_request( + &mut self.stream, + self.seq, + OpCode::DocumentUpdate, + TextFields { + collection: Some("ptv_post".to_string()), + document_id: Some("e0".to_string()), + updates: Some(vec![("amount".to_string(), amount)]), + ..Default::default() + }, + ) + .await + } +} + +/// 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() +} + +/// How a response ended, for an assert message. +fn describe(response: &NativeResponse) -> String { + match &response.error { + Some(error) => format!("{:?} {}: {}", response.status, error.code, error.message), + None => format!("{:?}", response.status), + } +} + +/// Open a block, run the update, change the source from a second session, and +/// commit. The balance must equal the source sum afterwards. +async fn update_in_block_then_move_source(by_opcode: bool) { + let server = sum_fixture().await; + let mut session = Session::open(&server).await; + + let begin = session.sql("BEGIN").await; + assert_ne!( + begin.status, + ResponseStatus::Error, + "BEGIN: {}", + describe(&begin) + ); + + let update = if by_opcode { + session.update_e0_by_opcode().await + } else { + session.sql(SQL_UPDATE).await + }; + assert_ne!( + update.status, + ResponseStatus::Error, + "the update inside the block: {}", + describe(&update) + ); + + server + .exec(MOVE_SOURCE) + .await + .expect("the concurrent source change"); + assert_eq!( + balance(&server, "acc").await, + "25", + "the concurrent change settles its own delta" + ); + + let commit = session.sql("COMMIT").await; + let sum = source_sum(&server).await; + assert_eq!( + balance(&server, "acc").await, + sum.to_string(), + "the balance must equal the source sum: COMMIT answered {}; the block either \ + aborts or settles again, and never applies the delta it computed from the old image", + describe(&commit) + ); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn native_direct_update_in_block_keeps_the_sum_when_the_source_moved() { + update_in_block_then_move_source(true).await; +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn native_sql_update_in_block_keeps_the_sum_when_the_source_moved() { + update_in_block_then_move_source(false).await; +} 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..4eac3062c --- /dev/null +++ b/nodedb/tests/wire/cases/materialized_sum_same_request.rs @@ -0,0 +1,66 @@ +// 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; +use nodedb::control::planner::calvin::settling_writes::SEVERAL_WRITES_REFUSAL; + +/// 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 (status, body) = server.http_query(sql).await; + + // The request is refused whole, so no write lands and the balance still + // equals the sum of the postings. + assert!( + !(200..300).contains(&status), + "the request must be refused: {body}" + ); + assert_eq!(status, 400, "a refused request is a 400: {body}"); + assert!( + body.contains(SEVERAL_WRITES_REFUSAL), + "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..4f54241c5 --- /dev/null +++ b/nodedb/tests/wire/cases/materialized_sum_truncate_edge_bearing.rs @@ -0,0 +1,63 @@ +// 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() +} + +#[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"); + + server.http_exec(&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; + + server.http_exec(&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/materialized_sum_truncate_protocols.rs b/nodedb/tests/wire/cases/materialized_sum_truncate_protocols.rs new file mode 100644 index 000000000..702d29c6e --- /dev/null +++ b/nodedb/tests/wire/cases/materialized_sum_truncate_protocols.rs @@ -0,0 +1,167 @@ +// SPDX-License-Identifier: BUSL-1.1 + +//! TRUNCATE of the source of a cross-shard materialized sum over the pgwire +//! and native protocols, with the source changed between settlement and the +//! TRUNCATE's transaction. +//! +//! Settlement ships the balance delta of the removed rows from the source +//! images it reads. Only the OCC read versions stamped on those images guard +//! the delta against a source row that moves afterwards. Every protocol must +//! carry them into the TRUNCATE's transaction. +//! +//! The `recon::before_submit` gate parks the TRUNCATE after settlement. The +//! test commits the source change while it is parked, then releases it. The +//! balance must equal the source sum whether the TRUNCATE aborts or settles +//! again. Each test runs on a collection without edges. + +#![cfg(feature = "failpoints")] + +use std::path::{Path, PathBuf}; +use std::time::Duration; + +use nodedb_test_support::native_harness::{open_trust_session, send_request_within}; +use nodedb_types::protocol::opcodes::OpCode; +use nodedb_types::protocol::text_fields::TextFields; + +use super::materialized_sum_cross_shard_expanders::{FIXTURE_CORES, balance, sum_fixture_on}; +use crate::harness::TestServer; + +/// The concurrent change: `e2` joins `acc`, which goes from 15 to 22. +const MOVE_SOURCE: &str = "INSERT INTO ptv_post (id, account_id, amount) VALUES ('e2', 'acc', '7')"; +/// How long a native request waits for its parked answer. +const PARKED_BOUND: Duration = Duration::from_secs(60); + +/// 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 TRUNCATE 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. +async fn move_source_then_release( + server: &TestServer, + release: &Path, + parked: tokio::task::JoinHandle, +) { + wait_parked(release).await; + tokio::time::timeout(Duration::from_secs(30), server.exec(MOVE_SOURCE)) + .await + .expect("the concurrent source change did not return while the TRUNCATE was parked") + .expect("the concurrent source change"); + assert_eq!( + balance(server, "acc").await, + "22", + "the concurrent change settles its own delta" + ); + std::fs::write(release, b"release").expect("release the parked TRUNCATE"); + // Either end is fine: the check is the balance, not the answer. + parked.await.expect("the parked TRUNCATE task"); + + let sum = source_sum(server).await; + assert_eq!( + balance(server, "acc").await, + sum.to_string(), + "the balance must equal the source sum: the TRUNCATE either aborts \ + or settles again, and never applies the delta it computed from the old images" + ); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn pgwire_truncate_keeps_the_sum_when_a_settled_source_row_moved() { + let gate_dir = tempfile::tempdir().expect("gate tempdir"); + let release = gate_dir.path().join("release-truncate"); + 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("TRUNCATE ptv_post") + .await + .map(|_| ()) + .map_err(|e| e.to_string()) + }); + move_source_then_release(&server, &release, parked).await; +} + +/// The same TRUNCATE over the native protocol's SQL opcode. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn native_sql_truncate_keeps_the_sum_when_a_settled_source_row_moved() { + let gate_dir = tempfile::tempdir().expect("gate tempdir"); + let release = gate_dir.path().join("release-truncate"); + let server = parked_fixture(&release).await; + + let mut stream = open_trust_session(server.native_port, "nodedb").await; + // `send_sql` gives up after 5 s. A parked request needs a longer bound. + let parked = tokio::spawn(async move { + send_request_within( + &mut stream, + 2, + OpCode::Sql, + TextFields { + sql: Some("TRUNCATE ptv_post".into()), + ..Default::default() + }, + Some(PARKED_BOUND), + ) + .await + .status + }); + move_source_then_release(&server, &release, parked).await; +} + +/// The same TRUNCATE through `OpCode::DocumentTruncate`, which runs in +/// `handle_direct_op`. `build_truncate` reads only the collection name. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn native_document_truncate_keeps_the_sum_when_a_settled_source_row_moved() { + let gate_dir = tempfile::tempdir().expect("gate tempdir"); + let release = gate_dir.path().join("release-truncate"); + let server = parked_fixture(&release).await; + + let mut stream = open_trust_session(server.native_port, "nodedb").await; + let parked = tokio::spawn(async move { + send_request_within( + &mut stream, + 2, + OpCode::DocumentTruncate, + TextFields { + collection: Some("ptv_post".to_string()), + ..Default::default() + }, + Some(PARKED_BOUND), + ) + .await + .status + }); + move_source_then_release(&server, &release, parked).await; +} diff --git a/nodedb/tests/wire/cases/materialized_sum_truncate_settlement_reads.rs b/nodedb/tests/wire/cases/materialized_sum_truncate_settlement_reads.rs new file mode 100644 index 000000000..afbead916 --- /dev/null +++ b/nodedb/tests/wire/cases/materialized_sum_truncate_settlement_reads.rs @@ -0,0 +1,111 @@ +// SPDX-License-Identifier: BUSL-1.1 + +//! TRUNCATE of the source of a cross-shard materialized sum, with the source +//! changed between settlement and the TRUNCATE's transaction. +//! +//! Settlement ships the balance delta of the removed rows from the source +//! images it reads. A source row that moves after that leaves the delta stale. +//! Only the OCC read versions settlement stamped on those images guard it, so +//! the TRUNCATE's transaction must carry them. +//! +//! The `recon::before_submit` gate parks the TRUNCATE after settlement. The +//! test commits the source change while it is parked, then releases it. The +//! balance must equal the source sum whether the TRUNCATE aborts or settles +//! again. + +#![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; + +/// The concurrent change: `e2` joins `acc`, which goes from 15 to 22. +const MOVE_SOURCE: &str = "INSERT INTO ptv_post (id, account_id, amount) VALUES ('e2', 'acc', '7')"; + +/// 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 TRUNCATE 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() +} + +/// Park a TRUNCATE of the source, move a source row, release it, and check +/// that the balance equals the source sum. `edges` adds one edge first, which +/// takes the edge-bearing route. +async fn truncate_with_moved_source(edges: bool) { + let gate_dir = tempfile::tempdir().expect("gate tempdir"); + let release = gate_dir.path().join("release-truncate"); + let spec = format!("recon::before_submit=wait_file({})", release.display()); + let server = TestServer::start_multicores_with_failpoints(FIXTURE_CORES, &spec).await; + let server = sum_fixture_on(server).await; + if edges { + server + .exec("GRAPH INSERT EDGE IN 'ptv_post' FROM 'e0' TO 'e1' TYPE 'l'") + .await + .expect("write one edge into the source"); + } + + 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": "TRUNCATE ptv_post" })) + .send() + .await + .map(|response| response.status()) + .map_err(|e| e.to_string()) + }); + wait_parked(&release).await; + tokio::time::timeout(Duration::from_secs(30), server.exec(MOVE_SOURCE)) + .await + .expect("the concurrent source change did not return while the TRUNCATE was parked") + .expect("the concurrent source change"); + assert_eq!( + balance(&server, "acc").await, + "22", + "the concurrent change settles its own delta" + ); + std::fs::write(&release, b"release").expect("release the parked TRUNCATE"); + parked + .await + .expect("the parked TRUNCATE task") + .expect("the HTTP answer"); + + let sum = source_sum(&server).await; + assert_eq!( + balance(&server, "acc").await, + sum.to_string(), + "the balance must equal the source sum: the TRUNCATE either aborts \ + or settles again, and never applies the delta it computed from the old images" + ); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn truncate_keeps_the_sum_when_a_settled_source_row_moved() { + truncate_with_moved_source(false).await; +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn edge_bearing_truncate_keeps_the_sum_when_a_settled_source_row_moved() { + truncate_with_moved_source(true).await; +} diff --git a/nodedb/tests/wire/cases/materialized_sum_two_columns.rs b/nodedb/tests/wire/cases/materialized_sum_two_columns.rs new file mode 100644 index 000000000..326371471 --- /dev/null +++ b/nodedb/tests/wire/cases/materialized_sum_two_columns.rs @@ -0,0 +1,122 @@ +// SPDX-License-Identifier: BUSL-1.1 + +//! One predicate write that moves two sum columns of one cross-shard target row. +//! +//! Each column is its own balance, so the write ships two deltas to the same +//! row. The several-writes guard counts balances by column, so the write runs +//! and both columns end equal to their source sums. + +use crate::harness::TestServer; + +/// The sum's target collection, as in `materialized_sum_cross_shard_expanders`. +/// The pair hashes to different vShards and cores there. +const ACCOUNTS: &str = "ptv_acct"; +/// The sum's source collection. +const POSTINGS: &str = "ptv_post"; +/// Data Plane cores the server runs. +const CORES: usize = 2; + +async fn exec(server: &TestServer, sql: &str) { + server + .exec(sql) + .await + .unwrap_or_else(|e| panic!("run {sql}: {e}")); +} + +/// Accounts `acc` holds `balance` and `fees`, each summed from a postings +/// column. Posting `e0` is 10 and 1, `e1` is 5 and 2. +async fn two_column_fixture() -> TestServer { + let server = TestServer::start_multicores(CORES).await; + exec( + &server, + &format!( + "CREATE COLLECTION {ACCOUNTS} (id TEXT PRIMARY KEY, owner TEXT) \ + WITH (engine='document_strict')" + ), + ) + .await; + exec( + &server, + &format!( + "CREATE COLLECTION {POSTINGS} \ + (id TEXT PRIMARY KEY, account_id TEXT, amount TEXT, fee TEXT) \ + WITH (engine='document_strict')" + ), + ) + .await; + for (column, value) in [("balance", "amount"), ("fees", "fee")] { + exec( + &server, + &format!( + "ALTER COLLECTION {ACCOUNTS} ADD COLUMN {column} TEXT \ + MATERIALIZED_SUM SOURCE {POSTINGS} \ + ON {POSTINGS}.account_id = {ACCOUNTS}.id VALUE {POSTINGS}.{value}" + ), + ) + .await; + } + exec( + &server, + &format!( + "INSERT INTO {ACCOUNTS} (id, owner, balance, fees) \ + VALUES ('acc', 'alice', '0', '0')" + ), + ) + .await; + for (id, amount, fee) in [("e0", "10", "1"), ("e1", "5", "2")] { + exec( + &server, + &format!( + "INSERT INTO {POSTINGS} (id, account_id, amount, fee) \ + VALUES ('{id}', 'acc', '{amount}', '{fee}')" + ), + ) + .await; + } + server +} + +async fn stored(server: &TestServer, column: &str) -> String { + let rows = server + .query_text(&format!("SELECT {column} FROM {ACCOUNTS} WHERE id = 'acc'")) + .await + .unwrap_or_else(|e| panic!("read {column}: {e}")); + match rows.as_slice() { + [value] => value.clone(), + other => panic!("one {column} row, got {other:?}"), + } +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn one_predicate_write_moves_two_sum_columns_of_one_row() { + let server = two_column_fixture().await; + assert_eq!(stored(&server, "balance").await, "15", "seed balance"); + assert_eq!(stored(&server, "fees").await, "3", "seed fees"); + + server + .http_exec(&format!( + "UPDATE {POSTINGS} SET amount = '1', fee = '7' WHERE account_id = 'acc'" + )) + .await; + + assert_eq!(stored(&server, "balance").await, "2", "1 + 1"); + assert_eq!(stored(&server, "fees").await, "14", "7 + 7"); +} + +/// Two predicate writes of one request, each moving a different column of the +/// same target row. Each column settles from its own source column, so the +/// pair runs and both columns still equal their source sums. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn two_predicate_writes_moving_different_columns_of_one_row_both_apply() { + let server = two_column_fixture().await; + + server + .http_exec(&format!( + "UPDATE {POSTINGS} SET amount = '1' WHERE account_id = 'acc'; \ + UPDATE {POSTINGS} SET fee = '7' WHERE account_id = 'acc'" + )) + .await; + + assert_eq!(stored(&server, "balance").await, "2", "1 + 1"); + assert_eq!(stored(&server, "fees").await, "14", "7 + 7"); +} diff --git a/nodedb/tests/wire/cases/mod.rs b/nodedb/tests/wire/cases/mod.rs index 2787346bc..e916a7a92 100644 --- a/nodedb/tests/wire/cases/mod.rs +++ b/nodedb/tests/wire/cases/mod.rs @@ -152,8 +152,16 @@ mod kv_sql_select; mod kv_transfer_decimal; mod kv_write_row_level_security; mod kv_write_row_level_security_atomics; +mod materialized_sum_crdt_delete_settlement_reads; mod materialized_sum_cross_shard_expanders; mod materialized_sum_cross_shard_http; +mod materialized_sum_native_block_reads; +mod materialized_sum_recon_settlement_reads; +mod materialized_sum_same_request; +mod materialized_sum_truncate_edge_bearing; +mod materialized_sum_truncate_protocols; +mod materialized_sum_truncate_settlement_reads; +mod materialized_sum_two_columns; mod materialized_sum_txn_command_tag; mod materialized_sum_txn_predicate_writes; mod materialized_sum_txn_preimage; @@ -195,6 +203,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..02a1b4922 --- /dev/null +++ b/nodedb/tests/wire/cases/predicate_write_sequence_http.rs @@ -0,0 +1,202 @@ +// 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 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; + + server + .http_exec( + "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; + + server + .http_exec("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; + + server + .http_exec( + "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; + } + + server + .http_exec(&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; + + server + .http_exec(&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..cc39d5753 --- /dev/null +++ b/nodedb/tests/wire/cases/predicate_write_two_statements_http.rs @@ -0,0 +1,78 @@ +// 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'" + ); + server.http_exec(&sql).await; + + 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`. diff --git a/nodedb/tests/wire/harness/query.rs b/nodedb/tests/wire/harness/query.rs index b752d6dde..77595f79e 100644 --- a/nodedb/tests/wire/harness/query.rs +++ b/nodedb/tests/wire/harness/query.rs @@ -116,6 +116,28 @@ impl TestServer { } } + /// POST `sql` to the HTTP query route. Returns the status code and the body. + pub async fn http_query(&self, sql: &str) -> (u16, String) { + let response = reqwest::Client::new() + .post(format!("http://127.0.0.1:{}/v1/query", self.http_port)) + .json(&serde_json::json!({ "sql": sql })) + .send() + .await + .unwrap_or_else(|e| panic!("POST /v1/query: {e}")); + let status = response.status().as_u16(); + let body = response.text().await.unwrap_or_default(); + (status, body) + } + + /// [`Self::http_query`] expecting a success status. + pub async fn http_exec(&self, sql: &str) { + let (status, body) = self.http_query(sql).await; + assert!( + (200..300).contains(&status), + "the request answered {status}: {body}" + ); + } + /// Execute a SQL statement expecting an error containing the given substring. pub async fn expect_error(&self, sql: &str, expected_substring: &str) { let client = self.client.as_ref();