fix(txn,search): route predicate writes outside a block, refuse two writes of one sum balance, refuse the doc_id score alias

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.

nodedb/src/control/fail_gate.rs

@@ -93,6 +93,15 @@ pub(crate) async fn after_calvin_stamp(
     }
 }
 
+/// The recon submit gate of node `node_id`, `recon::before_submit`: a
+/// dependent predicate write has settled its materialized sums and
+/// reconnoitred its rows, and parks before it builds and submits its Calvin
+/// transaction. A test changes a row the settlement read, then releases the
+/// gate.
+pub(crate) async fn before_recon_submit(node_id: u64) {
+    wait(FailScope::Node(node_id), "recon::before_submit").await;
+}
+
 /// The Calvin scheduler's redo gate on node `node_id`: after a committed
 /// slice's redo is resolved and before the data-group leader proposes it.
 /// Named per collection, `calvin::before_redo_propose::<collection>`.

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<DependentReconOutcome> {
     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<String>,
 ) -> crate::Result<DependentReconOutcome> {
-    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<String>,
+    reads: &[ReadSetEntry],
 ) -> crate::Result<DependentReconOutcome> {
     // 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,
@@ -239,6 +250,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 +401,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 +427,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() {

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,13 +57,18 @@ 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(
     state: &SharedState,
     tasks: &[PhysicalTask],
     identity: Option<&AuthenticatedIdentity>,
     tenant_id: TenantId,
+    reads: &[ReadSetEntry],
 ) -> crate::Result<Option<TruncateOutcome>> {
     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<PhysicalTask> = tasks
+        .iter()
+        .filter(|other| !is_truncate(&other.plan))
+        .cloned()
+        .collect();
+    // A request that ships balance deltas needs the transaction to carry them.
+    let in_transaction = edge_bearing || !others.is_empty();
 
     // The clone-write gate and authorization of the rows' truncate. The
     // lease holds until the transaction's outcome.
@@ -95,7 +108,7 @@ pub(super) async fn dispatch_truncate(
             })
             .await?
             {
-                CloneCheckedOutcome::Handled(response) if !edge_bearing => {
+                CloneCheckedOutcome::Handled(response) if !in_transaction => {
                     return Ok(Some(TruncateOutcome {
                         apply_result: Some(response),
                     }));
@@ -108,7 +121,7 @@ pub(super) async fn dispatch_truncate(
                         ),
                     });
                 }
-                CloneCheckedOutcome::Proceed(checked) if !edge_bearing => {
+                CloneCheckedOutcome::Proceed(checked) if !in_transaction => {
                     let response =
                         dispatch_authorized_durable_write(state, checked, TraceId::ZERO).await?;
                     crate::control::local_dispatch::reject_data_plane_error(&response)?;
@@ -122,7 +135,7 @@ pub(super) async fn dispatch_truncate(
                 }
             }
         }
-        None if !edge_bearing => {
+        None if !in_transaction => {
             let response = dispatch_rows_plan(state, task).await?;
             crate::control::local_dispatch::reject_data_plane_error(&response)?;
             return Ok(Some(TruncateOutcome {
@@ -132,15 +145,20 @@ pub(super) async fn dispatch_truncate(
         None => (task.clone(), None),
     };
 
-    let edge_shares = edge_share_tasks(&rows_task, &collection);
+    let edge_shares = if edge_bearing {
+        edge_share_tasks(&rows_task, &collection)
+    } else {
+        Vec::new()
+    };
     let edge_shares = match identity {
         Some(identity) => authorize(state, identity, edge_shares)?,
         None => edge_shares,
     };
-    let mut submission = Vec::with_capacity(edge_shares.len() + 1);
+    let mut submission = Vec::with_capacity(edge_shares.len() + others.len() + 1);
     submission.push(rows_task);
     submission.extend(edge_shares);
-    let tx_class = build_static_tx_class(&submission, tenant_id, &[])?;
+    submission.extend(others);
+    let tx_class = build_static_tx_class(&submission, tenant_id, reads)?;
     let apply_result = submit_calvin_routed(state, tx_class).await?;
     Ok(Some(TruncateOutcome { apply_result }))
 }

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;
 pub use submit::{
     RoutedAssignment, submit_and_await_calvin, submit_and_await_calvin_until, submit_calvin_routed,
     submit_calvin_routed_assign, submit_calvin_routed_write,

nodedb/src/control/planner/calvin/settling_writes.rs

@@ -0,0 +1,118 @@
+// 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(())
+}
+
+#[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());
+    }
+}

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<Vec<PhysicalTask>> {
@@ -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();