feat(cluster): validate Calvin read-sets against write versions at apply

Each participant now checks its own vShard's slice of the transaction's
LSN-versioned read-set against the local write-version index immediately
before applying, and reports whether it was still current on the executor
response. The check is observation-only: apply always commits regardless
of outcome. A node-global counter tracks how often a not-current read-set
is observed, giving tests and metrics visibility into staleness without
reaching into the per-core write-version index.

Author: farhan-syah

nodedb-physical/src/physical_plan/meta.rs

@@ -4,7 +4,7 @@
 
 use std::collections::BTreeMap;
 
-use nodedb_types::calvin::PassiveReadKey;
+use nodedb_types::calvin::{PassiveReadKey, VersionedReadEntry};
 use nodedb_types::timeseries::continuous_agg::ContinuousAggregateDef;
 use nodedb_types::{TenantId, Value};
 
@@ -303,19 +303,19 @@ pub enum MetaOp {
         /// the wall clock independently. Wire-additive: defaults to 0 on decode
         /// of older entries.
         epoch_system_ms: i64,
-        /// Whether THIS node is the leader of the data-group owning this
-        /// vshard, stamped by the dispatching scheduler at dispatch time.
-        ///
-        /// OLLP determinism: the optimistic-lock verification (`actual !=
-        /// predicted`) and the resulting `OllpRetryRequired` are LEADER-ONLY.
-        /// Followers trust the leader's verified decision and apply the carried
-        /// `ollp_predicted_surrogates` set verbatim, so every replica mutates
-        /// the identical surrogate set (Calvin determinism). This flag is a
-        /// per-node, non-replicated dispatch property — the scheduler on each
-        /// node stamps its OWN role; it is never part of the replicated log.
-        /// Wire-additive: defaults to `false` on decode of older entries (safe:
-        /// a follower-style apply against the carried predicted set).
+        /// Whether THIS node is the leader of the data-group owning this vshard,
+        /// stamped by the dispatching scheduler. OLLP verification (`actual !=
+        /// predicted` → `OllpRetryRequired`) is leader-only; followers apply the
+        /// carried `ollp_predicted_surrogates` verbatim so every replica mutates
+        /// the identical set (Calvin determinism). Per-node, never replicated.
+        /// Wire-additive: defaults to `false` on decode of older entries.
         is_group_leader: bool,
+        /// The transaction's LSN-versioned read-set from the replicated `TxClass`.
+        /// Each participant checks its own vShard's reads at apply and records
+        /// whether they were still current, without gating apply. Wire-additive:
+        /// defaults to empty on decode of older entries.
+        #[serde(default)]
+        versioned_reads: Vec<VersionedReadEntry>,
     },
 
     /// Calvin dependent-read executor: passive participant reads keys and

nodedb/src/bridge/dispatch.rs

@@ -579,6 +579,7 @@ mod tests {
                     payload: Payload::from_vec(b"result".to_vec()),
                     watermark_lsn: Lsn::new(42),
                     error_code: None,
+                    read_set_valid: None,
                 },
             })
             .unwrap();
@@ -631,6 +632,7 @@ mod tests {
                     payload: Payload::empty(),
                     watermark_lsn: Lsn::ZERO,
                     error_code: None,
+                    read_set_valid: None,
                 },
             })
             .unwrap();

nodedb/src/bridge/envelope.rs

@@ -245,6 +245,16 @@ pub struct Response {
 
     /// Error code if status is not Ok.
     pub error_code: Option<ErrorCode>,
+
+    /// Whether this response's originating transaction found its slice of the
+    /// versioned read-set still current against the local write versions at
+    /// apply time. `Some(true)` = still current (or no reads observed for this
+    /// slice); `Some(false)` = at least one read was superseded; `None` = the
+    /// response did not carry a read-set check (reads, control ops, and every
+    /// non-transaction response). Populated only by the transaction-apply path.
+    /// Reporting only — the apply commits regardless; a `Some(false)` here does
+    /// not abort or alter the write.
+    pub read_set_valid: Option<bool>,
 }
 
 pub use nodedb_physical::physical_plan::PhysicalPlan;
@@ -480,6 +490,7 @@ mod tests {
             payload: Payload::from_vec(b"result".to_vec()),
             watermark_lsn: Lsn::new(42),
             error_code: None,
+            read_set_valid: None,
         };
         assert_eq!(resp.status, Status::Ok);
         assert_eq!(resp.watermark_lsn, Lsn::new(42));
@@ -496,6 +507,7 @@ mod tests {
             payload: Payload::empty(),
             watermark_lsn: Lsn::ZERO,
             error_code: Some(ErrorCode::DeadlineExceeded),
+            read_set_valid: None,
         };
         assert_eq!(resp.error_code, Some(ErrorCode::DeadlineExceeded));
     }

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

@@ -212,6 +212,10 @@ impl Scheduler {
             plans,
             epoch_system_ms: txn.epoch_system_ms,
             is_group_leader: self.is_group_leader(),
+            // The replicated read-set travels to the apply core so each
+            // participant can check, at apply, whether its slice of the reads was
+            // still current. Empty for pure-write / autocommit transactions.
+            versioned_reads: txn.tx_class.versioned_reads.as_slice().to_vec(),
         });
 
         // no-determinism: request deadline is ephemeral, not written to WAL

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

@@ -318,6 +318,17 @@ impl Scheduler {
         }
 
         if response.status == crate::bridge::envelope::Status::Ok {
+            // Observe whether the applying participant reported its slice of the
+            // transaction's reads as no longer current against the local write
+            // versions. Observation only: the apply already committed and nothing
+            // is aborted here — the count is a node-global signal for tests and
+            // metrics. `None` means no read-set was checked (fast path / no reads).
+            if response.read_set_valid == Some(false) {
+                self.shared
+                    .read_set_validation_failures
+                    .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
+            }
+
             // Deposit the FULL applied Response (affected-count + watermark +
             // any RETURNING rows) into the local sidecar BEFORE proposing the
             // replicated CompletionAck. The ack fires the coordinator's

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

@@ -63,6 +63,7 @@ fn synthetic_returning_response(payload_bytes: Vec<u8>) -> Response {
         payload: Payload::from_vec(payload_bytes),
         watermark_lsn: Lsn::ZERO,
         error_code: None,
+        read_set_valid: None,
     }
 }
 

nodedb/src/control/request_tracker.rs

@@ -121,6 +121,7 @@ mod tests {
             payload: Payload::empty(),
             watermark_lsn: Lsn::ZERO,
             error_code: None,
+            read_set_valid: None,
         }
     }
 
@@ -133,6 +134,7 @@ mod tests {
             payload: Payload::from_vec(data.as_bytes().to_vec()),
             watermark_lsn: Lsn::ZERO,
             error_code: None,
+            read_set_valid: None,
         }
     }
 

nodedb/src/control/server/broadcast.rs

@@ -103,6 +103,7 @@ pub async fn broadcast_to_all_cores_txn(
         payload: crate::bridge::envelope::Payload::from_vec(outcome.merged_array),
         watermark_lsn: outcome.watermark_lsn,
         error_code: None,
+        read_set_valid: None,
     })
 }
 
@@ -209,6 +210,7 @@ pub async fn broadcast_count_to_all_cores(
         payload: crate::bridge::envelope::Payload::from_vec(payload),
         watermark_lsn: max_lsn,
         error_code: None,
+        read_set_valid: None,
     })
 }
 

nodedb/src/control/server/dispatch_utils.rs

@@ -618,6 +618,7 @@ mod collect_budget_tests {
             payload: Payload::from_vec(array_payload(n)),
             watermark_lsn: Lsn::ZERO,
             error_code: None,
+            read_set_valid: None,
         }
     }
 
@@ -630,6 +631,7 @@ mod collect_budget_tests {
             payload: Payload::from_vec(array_payload(n)),
             watermark_lsn: Lsn::ZERO,
             error_code: None,
+            read_set_valid: None,
         }
     }
 
@@ -643,6 +645,7 @@ mod collect_budget_tests {
             payload: Payload::from_vec(vec![0u8; bytes]),
             watermark_lsn: Lsn::ZERO,
             error_code: None,
+            read_set_valid: None,
         }
     }
 
@@ -655,6 +658,7 @@ mod collect_budget_tests {
             payload: Payload::from_vec(vec![0u8; bytes]),
             watermark_lsn: Lsn::ZERO,
             error_code: None,
+            read_set_valid: None,
         }
     }
 

nodedb/src/control/server/exchange/gather.rs

@@ -448,5 +448,6 @@ pub(super) fn outcome_to_response(merged_array: Vec<u8>, watermark_lsn: Lsn) ->
         payload: crate::bridge::envelope::Payload::from_vec(merged_array),
         watermark_lsn,
         error_code: None,
+        read_set_valid: None,
     }
 }