feat(control): fence uncontended writes on per-vShard deterministic locks

The write-admission gate previously stamped every write Admitted as a
no-op seam with no actual locking. It now acquires the exact
deterministic lock keys for uncontended POINT writes (Document, KV,
Vector, single-home graph edges) against the same per-vShard
Arc<Mutex<LockManager>> the Calvin scheduler validates against, and
holds the resulting guard across enqueue and response, releasing on
drop.

A point write whose keys are already held by a pending cross-shard
commit, along with any predicate, bulk, or multi-home write, now routes
through the deterministic Calvin scheduler instead of the fast path:
point writes submit a statically-built TxClass, predicate writes go
through dependent reconnaissance to discover their affected surrogates.
Both share the same lock table as the fast path, so there is no
time-of-check/time-of-use gap between the two paths.

TxClass construction is split out of the planner's dispatch module into
its own tx_class module to be shared by the new routing path.

Author: farhan-syah

nodedb-cluster-tests/tests/single_node_calvin_affected_count.rs

@@ -0,0 +1,177 @@
+// SPDX-License-Identifier: BUSL-1.1
+
+//! A plain (non-RETURNING) write routed through Calvin surfaces its ACTUAL
+//! affected-row count, not a synthesized zero.
+//!
+//! An edge-bearing schemaless collection routes a predicate `DELETE FROM coll
+//! WHERE <non-pk>` through the OLLP/Calvin dependent path (rewritten as a
+//! `BulkDelete` so mirrored edges are cleaned atomically). The Calvin completion
+//! ack carries no payload, so before this fix the applied Data-Plane response —
+//! including the affected-row count — was deposited ONLY for a RETURNING write.
+//! A plain delete therefore came back as `DELETE 0` even though it removed rows.
+//!
+//! This test proves the fix: the primary-write participant now deposits its full
+//! applied response (count included) for a plain write too, so a multi-row
+//! delete routed through Calvin reports the correct `DELETE N`.
+//!
+//! File name contains "calvin" within the cluster-tests crate so nextest applies
+//! the cluster test-group serialization.
+
+mod common;
+
+use std::sync::atomic::Ordering;
+use std::time::Duration;
+
+use common::cluster_harness::{TestClusterNode, wait_for};
+use tokio_postgres::SimpleQueryMessage;
+
+/// Documents seeded with `mark = 'del'` (all deleted by the predicate) and with
+/// `mark = 'keep'` (all retained).
+const TO_DELETE: usize = 3;
+const TO_KEEP: usize = 3;
+
+/// Count of transactions the single-node sequencer has admitted to an epoch, or
+/// `0` if the sequencer metrics handle is not installed yet.
+fn sequencer_admitted(node: &TestClusterNode) -> u64 {
+    node.shared
+        .sequencer_metrics
+        .get()
+        .map(|m| m.admitted_total.load(Ordering::Relaxed))
+        .unwrap_or(0)
+}
+
+/// Whether `coll` is flagged edge-bearing in this node's local catalog. The
+/// implicit-edge mark is committed via the replicated metadata path, so the
+/// DELETE must wait for it before planning — otherwise the predicate delete
+/// lowers to a fast-path bulk delete instead of the Calvin/OLLP `BulkDelete`.
+fn collection_edge_bearing(node: &TestClusterNode, coll: &str) -> bool {
+    node.shared
+        .credentials
+        .catalog()
+        .load_collections_for_tenant(nodedb_types::DatabaseId::DEFAULT, 1)
+        .map(|v| v.iter().any(|c| c.name == coll && c.has_implicit_edges))
+        .unwrap_or(false)
+}
+
+/// Affected-row count carried by the first `CommandComplete` in a simple-query
+/// response (PostgreSQL's `DELETE N` count).
+fn command_count(msgs: &[SimpleQueryMessage]) -> Option<u64> {
+    msgs.iter().find_map(|m| match m {
+        SimpleQueryMessage::CommandComplete(n) => Some(*n),
+        _ => None,
+    })
+}
+
+/// A plain predicate `DELETE` on an edge-bearing collection routes through the
+/// single-node Calvin OLLP path and reports the correct affected-row count
+/// through the completion path (previously reported `DELETE 0`).
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn calvin_plain_delete_surfaces_affected_count() {
+    // 4 Data-Plane cores so the edge endpoints land on distinct vShards: the
+    // doc-delete participant carries the primary write (and its count) while the
+    // edge-delete participants carry none — exercising the primary-write deposit
+    // gate on a genuinely multi-participant transaction.
+    let node = TestClusterNode::spawn_single_node_calvin(4)
+        .await
+        .expect("spawn standalone single-node-calvin server");
+
+    // The lone sequencer voter self-elects; wait for it before submitting.
+    wait_for(
+        "single-node sequencer leader elected",
+        Duration::from_secs(10),
+        Duration::from_millis(50),
+        || node.sequencer_leader() == node.node_id,
+    )
+    .await;
+
+    let coll = "sncalvin_affected_count";
+    node.client
+        .simple_query(&format!(
+            "CREATE COLLECTION {coll} WITH (engine='document_schemaless')"
+        ))
+        .await
+        .expect("CREATE COLLECTION");
+    wait_for(
+        "collection visible on the node",
+        Duration::from_secs(10),
+        Duration::from_millis(50),
+        || node.cached_collection_count() >= 1,
+    )
+    .await;
+
+    // Seed implicit-edge documents (plain docs carrying _from/_to/_type, which
+    // marks the collection edge-bearing). `mark = 'del'` docs will be matched by
+    // the predicate delete; `mark = 'keep'` docs must survive.
+    for i in 0..TO_DELETE {
+        node.client
+            .simple_query(&format!(
+                "INSERT INTO {coll} {{ id: 'del_{i}', _from: 'src_del_{i}', _to: 'hub', _type: 'l', mark: 'del' }}"
+            ))
+            .await
+            .unwrap_or_else(|e| panic!("insert deletable edge del_{i}: {e}"));
+    }
+    for i in 0..TO_KEEP {
+        node.client
+            .simple_query(&format!(
+                "INSERT INTO {coll} {{ id: 'keep_{i}', _from: 'src_keep_{i}', _to: 'hub', _type: 'l', mark: 'keep' }}"
+            ))
+            .await
+            .unwrap_or_else(|e| panic!("insert retained edge keep_{i}: {e}"));
+    }
+
+    // Wait for the implicit-edge mark to land so the predicate delete plans as a
+    // `BulkDelete` and routes through Calvin (not the fast bulk path).
+    wait_for(
+        "collection marked edge-bearing",
+        Duration::from_secs(10),
+        Duration::from_millis(50),
+        || collection_edge_bearing(&node, coll),
+    )
+    .await;
+
+    let admitted_before = sequencer_admitted(&node);
+
+    // A non-PK predicate DELETE on the edge-bearing collection is a `BulkDelete`
+    // routed through the OLLP/Calvin dependent path. Plain (no RETURNING): it
+    // must report the actual number of matched rows.
+    let msgs = node
+        .client
+        .simple_query(&format!("DELETE FROM {coll} WHERE mark = 'del'"))
+        .await
+        .expect("plain predicate delete routed through Calvin must complete");
+
+    // Proof it traversed the sequencer→scheduler path (not a fast path that
+    // never touches Calvin): the delete was admitted to a Calvin epoch.
+    let admitted_after = sequencer_admitted(&node);
+    assert!(
+        admitted_after > admitted_before,
+        "the predicate delete must be admitted to a Calvin epoch \
+         (before={admitted_before}, after={admitted_after})"
+    );
+
+    // The core assertion: the reported affected count is the number of matched
+    // rows, carried through the completion sidecar. Before the fix it was 0.
+    let count = command_count(&msgs).expect("DELETE returns a CommandComplete count");
+    assert_eq!(
+        count, TO_DELETE as u64,
+        "a plain DELETE routed through Calvin must report the actual affected-row \
+         count ({TO_DELETE}), not a synthesized 0; got {count}"
+    );
+
+    // Sanity: exactly the matched documents are gone and the rest remain.
+    let remaining = node
+        .client
+        .simple_query(&format!("SELECT * FROM {coll}"))
+        .await
+        .expect("SELECT all rows");
+    let remaining_count = remaining
+        .iter()
+        .filter(|m| matches!(m, SimpleQueryMessage::Row(_)))
+        .count();
+    assert_eq!(
+        remaining_count, TO_KEEP,
+        "exactly the mark='del' documents must be removed"
+    );
+
+    node.shutdown().await;
+}

nodedb/src/control/cluster/calvin/scheduler/driver/core/dispatch.rs

@@ -83,36 +83,25 @@ pub(crate) fn plan_vshard(plan: &PhysicalPlan) -> Vec<VShardId> {
     )]
 }
 
-/// Whether any plan in `plans` carries a RETURNING clause.
+/// Whether this vShard's slice carries a PRIMARY user data write — the write
+/// whose applied `Response` (affected-count + any RETURNING rows) the
+/// coordinator surfaces.
 ///
-/// Only the `Document` DML variants can carry `returning`
-/// (`PointUpdate` / `BulkUpdate` / `PointDelete` / `BulkDelete` /
-/// `UpdateFromJoin`); every other plan is row-less. Used to gate the applied
-/// `Response` deposit so only the RETURNING-bearing participant populates the
-/// coordinator's sidecar.
-pub(crate) fn plans_have_returning(plans: &[PhysicalPlan]) -> bool {
+/// A primary write is a Document / KV / Vector / Timeseries / Columnar / Array
+/// write — NOT the implicit graph-edge cleanup (`EdgePut` / `EdgeDelete`) that
+/// dual-homes alongside a document delete/update. For a single-collection user
+/// DML (plus its implicit edges) exactly ONE participant carries the primary
+/// write, so only it deposits the applied `Response` into the coordinator's
+/// sidecar and the edge participants never clobber the entry.
+///
+/// This gate subsumes the RETURNING case (a RETURNING write IS a primary write,
+/// so its rows are still deposited) while ALSO carrying the affected-count of a
+/// plain (non-RETURNING) write — which a RETURNING-only gate dropped, making a
+/// routed plain write report zero rows affected.
+pub(crate) fn plans_have_primary_write(plans: &[PhysicalPlan]) -> bool {
     plans.iter().any(|plan| {
-        matches!(
-            plan,
-            PhysicalPlan::Document(
-                DocumentOp::PointUpdate {
-                    returning: Some(_),
-                    ..
-                } | DocumentOp::BulkUpdate {
-                    returning: Some(_),
-                    ..
-                } | DocumentOp::PointDelete {
-                    returning: Some(_),
-                    ..
-                } | DocumentOp::BulkDelete {
-                    returning: Some(_),
-                    ..
-                } | DocumentOp::UpdateFromJoin {
-                    returning: Some(_),
-                    ..
-                }
-            )
-        )
+        crate::control::planner::calvin::is_write_plan(plan)
+            && !matches!(plan, PhysicalPlan::Graph(_))
     })
 }
 
@@ -215,7 +204,7 @@ impl Scheduler {
                 return;
             }
         };
-        let has_returning = plans_have_returning(&plans);
+        let has_primary_write = plans_have_primary_write(&plans);
         let plan = PhysicalPlan::Meta(MetaOp::CalvinExecuteStatic {
             epoch,
             position,
@@ -294,7 +283,7 @@ impl Scheduler {
                 // no-determinism: dispatch_time is scheduler observability, not Calvin WAL data
                 dispatch_time: dispatch_instant,
                 lock_acquired_time,
-                has_returning,
+                has_primary_write,
             },
         );
     }
@@ -345,7 +334,7 @@ impl Scheduler {
                 return;
             }
         };
-        let has_returning = plans_have_returning(&plans);
+        let has_primary_write = plans_have_primary_write(&plans);
         let plan = PhysicalPlan::Meta(MetaOp::CalvinExecuteActive {
             epoch,
             position,
@@ -425,7 +414,7 @@ impl Scheduler {
                 // no-determinism: dispatch_time is scheduler observability, not Calvin WAL data
                 dispatch_time: dispatch_instant,
                 lock_acquired_time,
-                has_returning,
+                has_primary_write,
             },
         );
     }

nodedb/src/control/cluster/calvin/scheduler/driver/core/process.rs

@@ -4,6 +4,7 @@
 //! bookkeeping for the Calvin scheduler.
 
 use std::collections::BTreeMap;
+use std::sync::Arc;
 use std::time::Instant;
 
 use nodedb_cluster::calvin::types::SequencedTxn;
@@ -37,7 +38,10 @@ impl Scheduler {
             keys_count,
         )
         .entered();
-        let outcome = self.lock_manager.acquire(txn_id, keys.clone());
+        let outcome = {
+            let mut lm = self.lock_manager.lock().unwrap_or_else(|p| p.into_inner());
+            lm.acquire(txn_id, keys.clone())
+        };
 
         match outcome {
             AcquireOutcome::Ready => {
@@ -116,29 +120,46 @@ impl Scheduler {
     ) {
         let epoch = txn_id.epoch;
 
-        let newly_unblocked = self.lock_manager.release(txn_id);
-
-        for waiter_id in newly_unblocked {
-            if let Some(blocked) = self.blocked.get(&waiter_id)
-                && self.lock_manager.is_ready(waiter_id, &blocked.keys)
-            {
-                let keys = blocked.keys.clone();
-                let outcome = self.lock_manager.acquire(waiter_id, keys.clone());
-                debug_assert_eq!(
-                    outcome,
-                    AcquireOutcome::Ready,
-                    "is_ready returned true but acquire returned Blocked"
-                );
-
-                if let Some(blocked_txn) = self.blocked.remove(&waiter_id) {
-                    let wait_ms = blocked_txn.blocked_at.elapsed().as_millis() as u64;
-                    self.metrics.record_lock_wait_ms(wait_ms);
-                    // no-determinism: lock_acquired_time for unblocked txn is scheduler observability, not Calvin WAL data
-                    self.dispatch_or_barrier(blocked_txn.txn, waiter_id, keys, Instant::now());
+        // Release this txn's locks, promote any newly-unblocked waiters, and
+        // collect the ones ready to dispatch — all under ONE guard so the
+        // release and the promoted re-acquire are atomic against a concurrent
+        // gate probe. Dispatch happens AFTER the guard drops: `dispatch_or_barrier`
+        // does not touch the lock table, but holding the guard across it would
+        // deadlock if it ever did, so the collect-then-dispatch split keeps the
+        // critical section minimal and re-entrancy-safe.
+        let lm = Arc::clone(&self.lock_manager);
+        let mut to_dispatch: Vec<(SequencedTxn, TxnId, std::collections::BTreeSet<LockKey>)> =
+            Vec::new();
+        {
+            let mut guard = lm.lock().unwrap_or_else(|p| p.into_inner());
+            let newly_unblocked = guard.release(txn_id);
+
+            for waiter_id in newly_unblocked {
+                if let Some(blocked) = self.blocked.get(&waiter_id)
+                    && guard.is_ready(waiter_id, &blocked.keys)
+                {
+                    let keys = blocked.keys.clone();
+                    let outcome = guard.acquire(waiter_id, keys.clone());
+                    debug_assert_eq!(
+                        outcome,
+                        AcquireOutcome::Ready,
+                        "is_ready returned true but acquire returned Blocked"
+                    );
+
+                    if let Some(blocked_txn) = self.blocked.remove(&waiter_id) {
+                        let wait_ms = blocked_txn.blocked_at.elapsed().as_millis() as u64;
+                        self.metrics.record_lock_wait_ms(wait_ms);
+                        to_dispatch.push((blocked_txn.txn, waiter_id, keys));
+                    }
                 }
             }
         }
 
+        for (txn, waiter_id, keys) in to_dispatch {
+            // no-determinism: lock_acquired_time for unblocked txn is scheduler observability, not Calvin WAL data
+            self.dispatch_or_barrier(txn, waiter_id, keys, Instant::now());
+        }
+
         if self.last_applied_epoch
             == crate::control::cluster::calvin::scheduler::NOT_YET_APPLIED_EPOCH
             || epoch > self.last_applied_epoch