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

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_bulks.rs

@@ -0,0 +1,175 @@
+// SPDX-License-Identifier: BUSL-1.1
+
+//! The predicate writes of one dependent submission.
+//!
+//! A submission can hold several `BulkUpdate`/`BulkDelete` tasks, for example
+//! the statements of one request. Each task has its own collection, filters
+//! and edge lifecycle, so each gets its own reconnaissance scan and carries
+//! only the rows and edges that scan found.
+
+use std::collections::BTreeMap;
+
+use nodedb_physical::physical_plan::OllpPredictedEdge;
+use nodedb_physical::physical_task::PhysicalTask;
+
+use crate::control::planner::calvin::is_dependent_predicate;
+use crate::control::planner::calvin::preexec::PreexecScan;
+use crate::control::state::SharedState;
+use crate::types::{DatabaseId, TenantId};
+
+use super::dependent_recon_node_edges::reconnoitre;
+use super::dependent_recon_plan::{inject_ollp_predicted_edges, inject_ollp_surrogates};
+use super::dependent_recon_predicate::{
+    EdgeLifecycle, classify_edge_lifecycle, extract_bulk_predicate_info,
+};
+
+/// What one predicate write carries into the submission: the rows it matched
+/// and the implicit edges of those rows.
+pub(super) type BulkPrediction = (Vec<u32>, Vec<OllpPredictedEdge>);
+
+/// One predicate write of a submission and the inputs of its reconnaissance.
+/// They are constant across retries.
+pub(super) struct BulkRecon {
+    /// Position of the task in the submission's task list.
+    pub(super) index: usize,
+    pub(super) collection: String,
+    pub(super) filters: Vec<u8>,
+    pub(super) edge_mode: EdgeLifecycle,
+}
+
+impl BulkRecon {
+    /// Every predicate write of `tasks`, in task order.
+    pub(super) fn of(tasks: &[PhysicalTask]) -> crate::Result<Vec<Self>> {
+        tasks
+            .iter()
+            .enumerate()
+            .filter(|(_, task)| is_dependent_predicate(&task.plan))
+            .map(|(index, task)| {
+                let (collection, filters) = extract_bulk_predicate_info(&task.plan);
+                Ok(Self {
+                    index,
+                    collection,
+                    filters,
+                    edge_mode: classify_edge_lifecycle(&task.plan)?,
+                })
+            })
+            .collect()
+    }
+}
+
+/// The reconnaissance scan of every predicate write, in the order of `bulks`.
+pub(super) async fn reconnoitre_all(
+    state: &SharedState,
+    tenant_id: TenantId,
+    database_id: DatabaseId,
+    bulks: &[BulkRecon],
+) -> crate::Result<Vec<PreexecScan>> {
+    let mut scans = Vec::with_capacity(bulks.len());
+    for bulk in bulks {
+        scans.push(
+            reconnoitre(
+                state,
+                tenant_id,
+                database_id,
+                &bulk.collection,
+                bulk.filters.clone(),
+                &bulk.edge_mode,
+            )
+            .await?,
+        );
+    }
+    Ok(scans)
+}
+
+/// A copy of `tasks` in which the predicate write at each index of
+/// `predictions` carries its own rows and edges.
+pub(super) fn with_bulk_predictions(
+    tasks: &[PhysicalTask],
+    predictions: &BTreeMap<usize, BulkPrediction>,
+) -> Vec<PhysicalTask> {
+    let mut out = tasks.to_vec();
+    for (index, (surrogates, edges)) in predictions {
+        if let Some(task) = out.get_mut(*index) {
+            inject_ollp_surrogates(&mut task.plan, surrogates.clone());
+            inject_ollp_predicted_edges(&mut task.plan, edges.clone());
+        }
+    }
+    out
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use nodedb_physical::physical_plan::{DocumentOp, PhysicalPlan};
+
+    fn bulk_delete(collection: &str) -> PhysicalTask {
+        PhysicalTask {
+            tenant_id: TenantId::new(1),
+            vshard_id: crate::types::VShardId::new(0),
+            database_id: DatabaseId::DEFAULT,
+            plan: PhysicalPlan::Document(DocumentOp::BulkDelete {
+                collection: nodedb_types::QualifiedCollection::new(DatabaseId::DEFAULT, collection),
+                filters: Vec::new(),
+                returning: None,
+                ollp_predicted_surrogates: None,
+                ollp_predicted_edges: None,
+                rls_filters: Vec::new(),
+                rls_write_check: nodedb_types::RlsWriteCheck::pending_injection(),
+                resolved_sum_targets: Vec::new(),
+                declared_primary_key: None,
+            }),
+            post_set_op: nodedb_physical::physical_task::PostSetOp::None,
+            txn_id: None,
+        }
+    }
+
+    fn carried(task: &PhysicalTask) -> (Option<Vec<u32>>, Option<Vec<OllpPredictedEdge>>) {
+        match &task.plan {
+            PhysicalPlan::Document(DocumentOp::BulkDelete {
+                ollp_predicted_surrogates,
+                ollp_predicted_edges,
+                ..
+            }) => (
+                ollp_predicted_surrogates.clone(),
+                ollp_predicted_edges.clone(),
+            ),
+            _ => (None, None),
+        }
+    }
+
+    #[test]
+    fn each_predicate_write_receives_its_own_prediction() {
+        let edge = OllpPredictedEdge {
+            surrogate: 3,
+            from: "a".into(),
+            to: "z".into(),
+            label: None,
+        };
+        let tasks = vec![bulk_delete("orders"), bulk_delete("users")];
+        let predictions = BTreeMap::from([
+            (0, (vec![3, 5], vec![edge.clone()])),
+            (1, (vec![9], Vec::new())),
+        ]);
+
+        let out = with_bulk_predictions(&tasks, &predictions);
+
+        assert_eq!(carried(&out[0]), (Some(vec![3, 5]), Some(vec![edge])));
+        assert_eq!(carried(&out[1]), (Some(vec![9]), Some(Vec::new())));
+    }
+
+    #[test]
+    fn a_predicate_write_is_found_at_its_task_position() {
+        let tasks = vec![bulk_delete("orders"), bulk_delete("users")];
+
+        let bulks = BulkRecon::of(&tasks).expect("classify the predicate writes");
+
+        let found: Vec<_> = bulks
+            .iter()
+            .map(|bulk| (bulk.index, bulk.collection.as_str()))
+            .collect();
+        assert_eq!(found.len(), 2);
+        assert_eq!(found[0].0, 0);
+        assert_eq!(found[1].0, 1);
+        assert!(found[0].1.ends_with("orders") && found[1].1.ends_with("users"));
+    }
+}

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

@@ -3,6 +3,7 @@
 pub mod abort_error;
 pub mod cross_shard_mode;
 pub mod dependent_recon;
+mod dependent_recon_bulks;
 pub mod dependent_recon_crdt;
 mod dependent_recon_finish;
 pub mod dependent_recon_node_edges;
@@ -20,6 +21,7 @@ pub mod read_dependent;
 pub mod reservation;
 pub mod retry_budget;
 pub mod retry_loop;
+pub mod settling_writes;
 pub mod submit;
 pub mod tx_class;
 pub mod txn_predict;
@@ -44,6 +46,7 @@ pub use edge_sequencing::{
 pub use explain::calvin_explain_preamble;
 pub use predicate::predicate_class_for_filters;
 pub use retry_loop::{DependentOutcome, DependentRetryArgs, run_dependent_with_retry};
+pub use settling_writes::refuse_several_settling_writes;
 pub use submit::{
     RoutedAssignment, submit_and_await_calvin, submit_and_await_calvin_until, submit_calvin_routed,
     submit_calvin_routed_assign, submit_calvin_routed_write,

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

@@ -90,7 +90,7 @@ pub enum RowIdentityRead<'a> {
 /// row's identity, in surrogate order, when the scan read it. `node_edges`
 /// holds the live edges incident on each matched row's node, when the caller
 /// read them. Edge order is irrelevant; surrogates remain sorted.
-#[derive(Default)]
+#[derive(Clone, Default)]
 pub struct PreexecScan {
     pub surrogates: Vec<u32>,
     pub edges: Vec<ScannedEdge>,

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

@@ -0,0 +1,78 @@
+// SPDX-License-Identifier: BUSL-1.1
+
+//! The request guard for writes that move one materialized-sum balance.
+
+use std::collections::BTreeSet;
+
+use nodedb_physical::physical_task::PhysicalTask;
+
+use super::is_dependent_predicate;
+use crate::Error;
+
+/// Refuse a request in which a predicate write and another write move the
+/// same materialized-sum balance row. Each write settles its delta from the
+/// committed rows, so a later write cannot see the rows an earlier one
+/// changes and the balance would drift from the sum of the source.
+pub fn refuse_several_settling_writes(tasks: &[PhysicalTask]) -> crate::Result<()> {
+    use nodedb_physical::physical_plan::{DocumentOp, PhysicalPlan};
+    if !tasks.iter().any(|task| is_dependent_predicate(&task.plan)) {
+        return Ok(());
+    }
+    let mut targets = BTreeSet::new();
+    for task in tasks {
+        if let PhysicalPlan::Document(DocumentOp::ApplyBalanceDelta {
+            collection,
+            document_id,
+            ..
+        }) = &task.plan
+            && !targets.insert((collection.to_string(), document_id.clone()))
+        {
+            return Err(Error::BadRequest {
+                detail: "several writes of one request move the same materialized-sum balance \
+                         and one is a predicate write; run them as separate statements or in a \
+                         transaction block"
+                    .to_owned(),
+            });
+        }
+    }
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use nodedb_physical::physical_plan::{DocumentOp, PhysicalPlan};
+
+    use crate::types::{DatabaseId, TenantId, VShardId};
+
+    fn task(plan: PhysicalPlan) -> PhysicalTask {
+        PhysicalTask {
+            tenant_id: TenantId::new(1),
+            vshard_id: VShardId::new(0),
+            database_id: DatabaseId::DEFAULT,
+            plan,
+            post_set_op: nodedb_physical::physical_task::PostSetOp::None,
+            txn_id: None,
+        }
+    }
+
+    fn bulk_delete(collection: &str) -> PhysicalTask {
+        task(PhysicalPlan::Document(DocumentOp::BulkDelete {
+            collection: nodedb_types::QualifiedCollection::new(DatabaseId::DEFAULT, collection),
+            filters: Vec::new(),
+            returning: None,
+            ollp_predicted_surrogates: None,
+            ollp_predicted_edges: None,
+            rls_filters: Vec::new(),
+            rls_write_check: nodedb_types::RlsWriteCheck::pending_injection(),
+            resolved_sum_targets: Vec::new(),
+            declared_primary_key: None,
+        }))
+    }
+
+    #[test]
+    fn plain_predicate_writes_are_not_refused() {
+        let tasks = vec![bulk_delete("orders"), bulk_delete("users")];
+        assert!(refuse_several_settling_writes(&tasks).is_ok());
+    }
+}