feat(cluster): record write-versions for committed Calvin applies

farhan-syah · farhan-syah · commit 444811a79806 · 2026-07-09T07:15:56.000+08:00
A Calvin apply's committed WAL LSN is only allocated after the apply
succeeds (the CalvinApplied WAL record is appended once the executor
response returns), so the apply itself could not advance the per-core
write-version index in place — leaving cross-shard-committed writes
with no recorded version at all, unlike fast-path writes and reads.

Once the scheduler obtains that LSN, it now dispatches a one-way,
record-only MetaOp::RecordCalvinWriteVersions op back to the same
core, funneling the transaction's locally-applied write plans through
the shared write-version recorder at that LSN — landing in the same
shard-local WAL-LSN space fast-path writes and read watermarks use.
Adds a node-global counter so the recording can be observed without
reaching into the !Send Data-Plane index, and a cluster test asserting
a cross-shard Calvin write advances it.
diff --git a/nodedb-cluster-tests/tests/single_node_calvin_write_versions.rs b/nodedb-cluster-tests/tests/single_node_calvin_write_versions.rs
@@ -0,0 +1,126 @@
+// SPDX-License-Identifier: BUSL-1.1
+
+//! A cross-shard, Calvin-committed write RECORDS its write-version in the
+//! per-core write-version index — closing the gap where Calvin applies advanced
+//! no version at all (fast-path writes and reads already do).
+//!
+//! The version index itself lives on the `!Send` Data-Plane core and its
+//! readers are test-only, so this asserts the recording FIRED via the
+//! node-global `calvin_write_versions_recorded` counter, which the per-vShard
+//! scheduler increments once per committed Calvin apply for which it dispatched
+//! a write-version record op (at the CalvinApplied WAL LSN). The counter is the
+//! standalone-observable proof that a cross-shard-committed write now advances
+//! the version index; the end-to-end serializability regression (via read-set
+//! validation) is covered separately once validation is enforcing.
+
+mod common;
+
+use std::sync::atomic::Ordering;
+use std::time::Duration;
+
+use nodedb_cluster::calvin::SEQUENCER_GROUP_ID;
+use nodedb_types::id::VShardId;
+
+use common::cluster_harness::{TestClusterNode, wait_for};
+
+/// Observed sequencer-group leader id from a node's local Raft status, or `0`
+/// if no leader is known yet.
+fn sequencer_leader(node: &TestClusterNode) -> u64 {
+    let Some(status_fn) = node.shared.raft_status_fn.get() else {
+        return 0;
+    };
+    status_fn()
+        .into_iter()
+        .find(|g| g.group_id == SEQUENCER_GROUP_ID)
+        .map(|g| g.leader_id)
+        .unwrap_or(0)
+}
+
+/// Count of committed Calvin applies whose write versions were recorded into
+/// the per-core write-version index.
+fn write_versions_recorded(node: &TestClusterNode) -> u64 {
+    node.shared
+        .calvin_write_versions_recorded
+        .load(Ordering::Relaxed)
+}
+
+/// A `(src, dst)` pair of graph node keys whose home vShards differ, so an edge
+/// between them is genuinely cross-shard. Deterministic: `VShardId::from_key` is
+/// a pure function of the key bytes.
+fn distinct_vshard_node_keys() -> (String, String) {
+    let dst = "sncalvin_wv_hub".to_string();
+    let vdst = VShardId::from_key(dst.as_bytes()).as_u32();
+    for i in 0u32..4096 {
+        let src = format!("sncalvin_wv_src_{i}");
+        if VShardId::from_key(src.as_bytes()).as_u32() != vdst {
+            return (src, dst);
+        }
+    }
+    panic!("could not find a node key on a distinct vShard from the hub in 4096 tries");
+}
+
+/// A cross-shard (dual-home edge) Calvin-committed write increments the
+/// node-global write-version-recorded counter — i.e. its keys' versions are
+/// recorded at the CalvinApplied WAL LSN instead of being silently dropped.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn cross_shard_calvin_write_records_write_version() {
+    // 4 Data-Plane cores so distinct vShards land on distinct cores — a genuine
+    // cross-core transaction routed through the Calvin scheduler.
+    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),
+        || sequencer_leader(&node) == node.node_id,
+    )
+    .await;
+
+    node.client
+        .simple_query("CREATE COLLECTION sncalvin_wv_graph")
+        .await
+        .expect("CREATE COLLECTION sncalvin_wv_graph");
+    wait_for(
+        "collection visible on the node",
+        Duration::from_secs(10),
+        Duration::from_millis(50),
+        || node.cached_collection_count() >= 1,
+    )
+    .await;
+
+    let (src, dst) = distinct_vshard_node_keys();
+    let recorded_before = write_versions_recorded(&node);
+
+    // AUTOCOMMIT cross-shard edge insert: because the endpoints home to distinct
+    // vShards, the edge dual-homes atomically through the Calvin scheduler
+    // (`submit_calvin_routed`), not the single-home fast path. On commit each
+    // participating vShard's scheduler records its slice's write versions.
+    node.client
+        .simple_query(&format!(
+            "GRAPH INSERT EDGE IN 'sncalvin_wv_graph' FROM '{src}' TO '{dst}' TYPE 'l'"
+        ))
+        .await
+        .expect("cross-shard edge insert must commit via the single-node Calvin path");
+
+    // Recording is dispatched fire-and-forget after the CalvinApplied WAL LSN is
+    // obtained, so wait for the counter to advance.
+    wait_for(
+        "calvin write-version recording fired for the cross-shard commit",
+        Duration::from_secs(10),
+        Duration::from_millis(25),
+        || write_versions_recorded(&node) > recorded_before,
+    )
+    .await;
+
+    let recorded_after = write_versions_recorded(&node);
+    assert!(
+        recorded_after > recorded_before,
+        "a cross-shard Calvin-committed write must record its write version \
+         (before={recorded_before}, after={recorded_after})"
+    );
+
+    node.shutdown().await;
+}
diff --git a/nodedb-physical/src/physical_plan/meta.rs b/nodedb-physical/src/physical_plan/meta.rs
@@ -479,4 +479,22 @@ pub enum MetaOp {
         value_marker: u64,
         graph_marker: u64,
     },
+
+    /// Record the per-key / per-collection write versions of a committed
+    /// Calvin transaction's locally-applied write plans.
+    ///
+    /// A Calvin apply's committed WAL LSN is known only after the apply
+    /// succeeds, so the apply itself cannot advance the version index. The
+    /// scheduler stamps that LSN onto this op's `wal_lsn` and dispatches it back
+    /// to the same core, which funnels `plans` through the shared write-version
+    /// recorder at that LSN — landing in the same shard-local WAL-LSN space the
+    /// single-shard fast path and read watermarks use. Records only: no base
+    /// mutation, no WAL append, no event emission. Wire-additive (appended last)
+    /// so older log entries decode unchanged.
+    RecordCalvinWriteVersions {
+        /// Tenant scope for all plans.
+        tenant_id: TenantId,
+        /// The locally-applied write plans whose keys' versions are recorded.
+        plans: Vec<super::PhysicalPlan>,
+    },
 }
diff --git a/nodedb/src/control/cluster/calvin/scheduler/driver/core/dispatch.rs b/nodedb/src/control/cluster/calvin/scheduler/driver/core/dispatch.rs
@@ -128,7 +128,7 @@ impl Scheduler {
         mr.vshard_role_is_leader(self.vshard_id)
     }
 
-    fn local_calvin_plans(
+    pub(in crate::control::cluster::calvin::scheduler::driver::core) fn local_calvin_plans(
         &self,
         plans: Vec<PhysicalPlan>,
         epoch: u64,
diff --git a/nodedb/src/control/cluster/calvin/scheduler/driver/core/mod.rs b/nodedb/src/control/cluster/calvin/scheduler/driver/core/mod.rs
@@ -15,6 +15,8 @@
 //! - [`dispatch`] — static / active dispatch to the Data Plane executor.
 //! - [`read_result`] — `CalvinReadResult` handling and barrier timeouts.
 //! - [`propose`] — propose `CalvinReadResult` Raft entries.
+//! - [`write_version_record`] — post-apply write-version recording for
+//!   committed Calvin transactions (at the CalvinApplied WAL LSN).
 //!
 //! # Determinism
 //!
@@ -34,6 +36,7 @@ pub mod process;
 pub mod propose;
 pub mod read_result;
 pub mod scheduler;
+pub mod write_version_record;
 
 #[cfg(test)]
 mod tests;
diff --git a/nodedb/src/control/cluster/calvin/scheduler/driver/core/scheduler.rs b/nodedb/src/control/cluster/calvin/scheduler/driver/core/scheduler.rs
@@ -375,18 +375,25 @@ impl Scheduler {
                     }
                 }
             }
-            if let Err(e) = self.shared.wal.append_calvin_applied(
+            match self.shared.wal.append_calvin_applied(
                 crate::types::VShardId::new(self.vshard_id),
                 txn_id.epoch,
                 txn_id.position,
             ) {
-                tracing::error!(
-                    vshard_id = self.vshard_id,
-                    epoch = txn_id.epoch,
-                    position = txn_id.position,
-                    error = %e,
-                    "calvin: failed to write CalvinApplied WAL record"
-                );
+                // The CalvinApplied WAL LSN is the committed write-LSN for this
+                // apply — the SAME shard-local WAL-LSN space fast-path writes and
+                // read watermarks use. Record the apply's per-key write versions
+                // at it now that it exists (it did not at dispatch time).
+                Ok(applied_lsn) => self.record_calvin_write_versions(txn_id, applied_lsn),
+                Err(e) => {
+                    tracing::error!(
+                        vshard_id = self.vshard_id,
+                        epoch = txn_id.epoch,
+                        position = txn_id.position,
+                        error = %e,
+                        "calvin: failed to write CalvinApplied WAL record"
+                    );
+                }
             }
             self.propose_sequencer_entry(
                 SequencerEntry::CompletionAck {
diff --git a/nodedb/src/control/cluster/calvin/scheduler/driver/core/write_version_record.rs b/nodedb/src/control/cluster/calvin/scheduler/driver/core/write_version_record.rs
@@ -0,0 +1,147 @@
+// SPDX-License-Identifier: BUSL-1.1
+
+//! Post-apply write-version recording for committed Calvin transactions.
+//!
+//! A Calvin apply's committed WAL LSN is allocated only AFTER the apply
+//! succeeds — the `CalvinApplied` WAL record is appended on the Control Plane
+//! once the executor response returns — so the apply itself carries no
+//! committed LSN and the per-core write-version index cannot be advanced in
+//! place (the dispatch stamps `wal_lsn: None`). Once the scheduler has that
+//! LSN it dispatches a one-way, record-only op back to the same core, which
+//! funnels the transaction's locally-applied write plans through the shared
+//! write-version recorder at that LSN — the same shard-local WAL-LSN space the
+//! single-shard fast path and read watermarks use.
+
+use std::sync::atomic::Ordering;
+use std::time::Instant;
+
+use super::scheduler::Scheduler;
+use crate::bridge::envelope::{Admission, ExemptReason, Priority, Request};
+use crate::control::cluster::calvin::scheduler::lock_manager::TxnId;
+use crate::types::{DatabaseId, Lsn, ReadConsistency, VShardId};
+use nodedb_physical::physical_plan::PhysicalPlan;
+use nodedb_physical::physical_plan::meta::MetaOp;
+
+impl Scheduler {
+    /// Record the per-key write versions of a just-committed Calvin
+    /// transaction's locally-applied write plans at its CalvinApplied WAL
+    /// `applied_lsn`.
+    ///
+    /// Dispatches a record-only [`MetaOp::RecordCalvinWriteVersions`] op back to
+    /// this vShard's core with `applied_lsn` stamped on the request envelope's
+    /// `wal_lsn`; the core funnels the plans through the shared write-version
+    /// recorder at that LSN. The recorded version therefore lands in the same
+    /// WAL-LSN space as fast-path writes, so a later read-set validation against
+    /// these keys is not a false-Valid serializability hole.
+    ///
+    /// Fire-and-forget: the recorded version is not needed to complete the
+    /// transaction, so the response is drained and discarded. A brief index-lag
+    /// window before the record op lands is harmless — nothing enforces read-set
+    /// validation against these versions yet. A dropped record (decode failure,
+    /// no local write plan, or dispatch backpressure) simply leaves the version
+    /// unrecorded and never blocks the commit.
+    pub(in crate::control::cluster::calvin::scheduler::driver::core) fn record_calvin_write_versions(
+        &self,
+        txn_id: TxnId,
+        applied_lsn: Lsn,
+    ) {
+        let epoch = txn_id.epoch;
+        let position = txn_id.position;
+
+        let Some(pending) = self.pending.get(&txn_id) else {
+            return;
+        };
+        let tenant_id = pending.txn.tx_class.tenant_id;
+        let plans = match super::super::helpers::decode_plans(&pending.txn.tx_class.plans) {
+            Ok(p) => p,
+            Err(e) => {
+                tracing::warn!(
+                    vshard_id = self.vshard_id,
+                    epoch,
+                    position,
+                    error = %e,
+                    "calvin: write-version recording skipped — plan decode failed"
+                );
+                return;
+            }
+        };
+        // The locally-applied slice this vShard committed. `local_calvin_plans`
+        // guarantees a non-empty set (this completion path only fires for a
+        // write participant); the recorder no-ops any plan without a per-key or
+        // collection version, so no gate on plan kind is applied here — gating
+        // on a narrower write predicate would silently skip recordable writes
+        // (e.g. a CRDT apply) and reopen the version gap.
+        let local = match self.local_calvin_plans(plans, epoch, position) {
+            Ok(p) => p,
+            Err(e) => {
+                tracing::warn!(
+                    vshard_id = self.vshard_id,
+                    epoch,
+                    position,
+                    error = %e,
+                    "calvin: write-version recording skipped — local plan routing failed"
+                );
+                return;
+            }
+        };
+
+        let request_id = self.next_request_id();
+        let deadline = Instant::now()
+            + std::time::Duration::from_millis(
+                self.config.epoch_duration_ms * u64::from(self.config.txn_deadline_multiplier),
+            );
+        let request = Request {
+            request_id,
+            tenant_id,
+            database_id: DatabaseId::DEFAULT,
+            vshard_id: VShardId::new(self.vshard_id),
+            plan: PhysicalPlan::Meta(MetaOp::RecordCalvinWriteVersions {
+                tenant_id,
+                plans: local,
+            }),
+            deadline,
+            priority: Priority::Normal,
+            trace_id: nodedb_types::TraceId([0u8; 16]),
+            consistency: ReadConsistency::Strong,
+            idempotency_key: None,
+            event_source: crate::event::EventSource::User,
+            user_roles: Vec::new(),
+            user_id: None,
+            statement_digest: None,
+            txn_id: None,
+            // The committed write-LSN for this Calvin apply — recorded against
+            // every key the plans wrote, in the same WAL-LSN space as fast-path.
+            wal_lsn: Some(applied_lsn),
+            // Recording only — the scheduler already held (and is releasing) this
+            // transaction's locks; the write-admission gate must not re-acquire.
+            admission: Admission::Exempt(ExemptReason::AlreadyOrdered),
+        };
+
+        // Register so the response routes to a real receiver (not the
+        // unknown-request warning path), then discard it — the recording is
+        // one-way.
+        let resp_rx = self.shared.tracker.register(request_id);
+        let dispatch_result = match self.shared.dispatcher.lock() {
+            Ok(mut d) => d.dispatch(request),
+            Err(poisoned) => poisoned.into_inner().dispatch(request),
+        };
+        if let Err(e) = dispatch_result {
+            self.shared.tracker.cancel(&request_id);
+            tracing::warn!(
+                vshard_id = self.vshard_id,
+                epoch,
+                position,
+                error = %e,
+                "calvin: write-version record dispatch failed"
+            );
+            return;
+        }
+        tokio::spawn(async move {
+            let mut rx = resp_rx;
+            let _ = rx.recv().await;
+        });
+        self.shared
+            .calvin_write_versions_recorded
+            .fetch_add(1, Ordering::Relaxed);
+    }
+}
diff --git a/nodedb/src/control/security/identity/plan_permission.rs b/nodedb/src/control/security/identity/plan_permission.rs
@@ -313,7 +313,8 @@ pub fn required_permission(plan: &crate::bridge::envelope::PhysicalPlan) -> Perm
         PhysicalPlan::Meta(
             MetaOp::CalvinExecuteStatic { .. }
             | MetaOp::CalvinExecutePassive { .. }
-            | MetaOp::CalvinExecuteActive { .. },
+            | MetaOp::CalvinExecuteActive { .. }
+            | MetaOp::RecordCalvinWriteVersions { .. },
         ) => Permission::Write,
 
         // Synonym group DDL: Alter permission (same tier as CREATE/DROP other DDL objects).
diff --git a/nodedb/src/control/server/exchange/all_cores/dispatch.rs b/nodedb/src/control/server/exchange/all_cores/dispatch.rs
@@ -180,7 +180,8 @@ pub async fn execute_plan_all_local_cores(
             | MetaOp::StageWrite { .. }
             | MetaOp::DropTxnOverlay { .. }
             | MetaOp::MarkSavepoint { .. }
-            | MetaOp::RollbackToSavepoint { .. } => {
+            | MetaOp::RollbackToSavepoint { .. }
+            | MetaOp::RecordCalvinWriteVersions { .. } => {
                 generic_gather(state, tenant_id, database_id, plan, trace_id).await
             }
         },
diff --git a/nodedb/src/control/state/fields.rs b/nodedb/src/control/state/fields.rs
@@ -375,6 +375,15 @@ pub struct SharedState {
     /// the schedulers (which hold `Arc<SharedState>`) advance the same counter
     /// the session reads. 0 in single-node / no-Calvin deployments.
     pub last_applied_calvin_epoch: Arc<AtomicU64>,
+    /// Count of committed Calvin applies whose write versions were recorded into
+    /// the per-core write-version index — incremented once per apply for which a
+    /// per-vShard scheduler dispatched a write-version record op after obtaining
+    /// the CalvinApplied WAL LSN. Node-global observability so tests and metrics
+    /// can confirm cross-shard-committed writes advance the version index without
+    /// reaching into the `!Send` Data-Plane index. `Arc` so the schedulers (which
+    /// hold `Arc<SharedState>`) increment the same counter a reader observes. 0 in
+    /// single-node / no-Calvin deployments and until the first Calvin write apply.
+    pub calvin_write_versions_recorded: Arc<AtomicU64>,
     /// Local, in-process sidecar carrying the applied Data-Plane [`Response`]
     /// (including RETURNING rows) of a completed Calvin transaction, keyed by
     /// its sequencer-assigned `TxnId`.
diff --git a/nodedb/src/control/state/init.rs b/nodedb/src/control/state/init.rs
diff --git a/nodedb/src/control/state/init_prod/mod.rs b/nodedb/src/control/state/init_prod/mod.rs
diff --git a/nodedb/src/data/executor/dispatch/meta.rs b/nodedb/src/data/executor/dispatch/meta.rs

Original file line number	Diff line number	Diff line change
`@@ -128,7 +128,7 @@ impl Scheduler {`
`128`	`128`	`mr.vshard_role_is_leader(self.vshard_id)`
`129`	`129`	`}`
`130`	`130`
`131`		`- fn local_calvin_plans(`
	`131`	`+ pub(in crate::control::cluster::calvin::scheduler::driver::core) fn local_calvin_plans(`
`132`	`132`	`&self,`
`133`	`133`	`plans: Vec<PhysicalPlan>,`
`134`	`134`	`epoch: u64,`