feat(control): replicate per-participant Calvin commit votes via sequencer

Add a SequencerEntry::Vote entry so a data-group leader durably proposes
its read-set validation outcome for a staged cross-shard txn through the
Calvin sequencer Raft log, keyed by vshard so retries overwrite rather
than double-count. CalvinCompletionRegistry tallies the votes per txn;
the tally is observed-only for now and does not yet drive the commit
verdict, which the local leader decision continues to determine.

Author: farhan-syah

nodedb-cluster/src/calvin/completion.rs

@@ -49,6 +49,12 @@ struct PendingCompletion {
     /// `mismatched` and completion: a routing failure is never retried and never
     /// falsely reported as success.
     routing_failed: Option<String>,
+    /// Durable per-participant commit votes tallied from `SequencerEntry::Vote`.
+    /// Keyed by vshard so a re-proposed vote (retry) overwrites deterministically
+    /// rather than double-counting. Currently observed-only: nothing reads this
+    /// tally to change flush/drop behavior yet (that is a follow-up); the
+    /// leader's local decision still drives.
+    votes: BTreeMap<u32, bool>,
 }
 
 impl PendingCompletion {
@@ -59,6 +65,7 @@ impl PendingCompletion {
             completion_tx: None,
             mismatched: false,
             routing_failed: None,
+            votes: BTreeMap::new(),
         }
     }
 
@@ -258,6 +265,34 @@ impl CalvinCompletionRegistry {
         }
     }
 
+    /// Record one participant vshard's durable commit vote for `txn`, tallied
+    /// from a replicated `SequencerEntry::Vote`.
+    ///
+    /// This is observed-only: it accumulates the vote per `vshard` (last write
+    /// wins, deterministic across re-proposals from retries) but does NOT
+    /// compute a verdict or wake any waiter — the local flush/drop decision
+    /// still drives. A follow-up aggregates the tally into the commit
+    /// verdict.
+    pub fn note_vote(&self, txn_id: TxnId, vshard: u32, commit: bool) {
+        let mut inner = self.inner.lock().unwrap_or_else(|p| p.into_inner());
+        let entry = inner
+            .completions
+            .entry(txn_id)
+            .or_insert_with(|| PendingCompletion::new(0));
+        entry.votes.insert(vshard, commit);
+    }
+
+    /// Test/inspection accessor: returns the current per-vshard vote tally for
+    /// `txn_id`, or `None` if no entry (and therefore no votes) exist yet.
+    pub fn vote_tally(&self, txn_id: TxnId) -> Option<BTreeMap<u32, bool>> {
+        self.inner
+            .lock()
+            .unwrap_or_else(|p| p.into_inner())
+            .completions
+            .get(&txn_id)
+            .map(|entry| entry.votes.clone())
+    }
+
     /// Test-only: returns the number of pending completion entries.
     /// Used to verify entries are removed once all acks arrive (no leak).
     #[cfg(test)]
@@ -468,6 +503,18 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn note_vote_tallies_one_vote_per_vshard() {
+        let reg = CalvinCompletionRegistry::new();
+        let txn = TxnId::new(30, 0);
+        reg.note_vote(txn, 1, true);
+        reg.note_vote(txn, 2, false);
+        let tally = reg.vote_tally(txn).expect("entry created by note_vote");
+        assert_eq!(tally.len(), 2);
+        assert_eq!(tally.get(&1), Some(&true));
+        assert_eq!(tally.get(&2), Some(&false));
+    }
+
     #[tokio::test]
     async fn mismatch_takes_precedence_over_pending_acks() {
         let reg = CalvinCompletionRegistry::new();

nodedb-cluster/src/calvin/sequencer/entry.rs

@@ -57,6 +57,18 @@ pub enum SequencerEntry {
         position: u32,
         detail: String,
     },
+    /// One participant vShard's durable commit vote for a staged cross-shard txn.
+    /// Generalizes `OllpMismatch` (an implicit vote=abort). Unlike
+    /// `OllpMismatch`/`CompletionAck` which key only on `(epoch, position)`, `Vote`
+    /// carries `vshard` because the verdict aggregator must attribute exactly one
+    /// vote per participant to know when the tally is complete. `commit` is the
+    /// participant's `read_set_valid` outcome (true = valid/commit, false = abort).
+    Vote {
+        epoch: u64,
+        position: u32,
+        vshard: u32,
+        commit: bool,
+    },
 }
 
 #[cfg(test)]
@@ -170,4 +182,17 @@ mod tests {
         assert_eq!(position, 3);
         assert_eq!(detail, "unroutable plan: Vector");
     }
+
+    #[test]
+    fn vote_msgpack_roundtrip() {
+        let entry = SequencerEntry::Vote {
+            epoch: 5,
+            position: 2,
+            vshard: 9,
+            commit: true,
+        };
+        let bytes = zerompk::to_msgpack_vec(&entry).expect("encode");
+        let decoded: SequencerEntry = zerompk::from_msgpack(&bytes).expect("decode");
+        assert_eq!(entry, decoded);
+    }
 }

nodedb-cluster/src/calvin/sequencer/service.rs

@@ -255,6 +255,7 @@ fn entry_txn_count(entry: &SequencerEntry) -> usize {
         SequencerEntry::CompletionAck { .. } => 0,
         SequencerEntry::OllpMismatch { .. } => 0,
         SequencerEntry::TxnRoutingFailed { .. } => 0,
+        SequencerEntry::Vote { .. } => 0,
     }
 }
 

nodedb-cluster/src/calvin/sequencer/state_machine.rs

@@ -197,6 +197,10 @@ impl SequencerStateMachine {
                 // Routing-failure signals are coordinator-side notifications
                 // only, like `OllpMismatch`; they carry no txn data to replay.
                 SequencerEntry::TxnRoutingFailed { .. } => {}
+                // Vote signals are coordinator-side notifications only, like
+                // `OllpMismatch`/`TxnRoutingFailed`; they carry no txn data to
+                // replay. Replay re-derives tallies via live `apply` instead.
+                SequencerEntry::Vote { .. } => {}
             }
         }
 
@@ -341,6 +345,22 @@ impl SequencerStateMachine {
                 self.completion_registry
                     .note_routing_failed(crate::calvin::TxnId::new(epoch, position), detail);
             }
+            // Durable per-participant commit vote for a staged cross-shard txn.
+            // Currently observed-only: the registry tallies votes but nothing
+            // yet reads the tally to change flush/drop behavior (that is a
+            // follow-up); the leader's local decision still drives.
+            SequencerEntry::Vote {
+                epoch,
+                position,
+                vshard,
+                commit,
+            } => {
+                self.completion_registry.note_vote(
+                    crate::calvin::TxnId::new(epoch, position),
+                    vshard,
+                    commit,
+                );
+            }
         }
     }
 }

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

@@ -44,6 +44,27 @@ impl Scheduler {
         staged_response: &Response,
     ) {
         let committed = staged_response.read_set_valid != Some(false);
+
+        // Durably propose this participant's commit vote via the sequencer
+        // Raft group, leader-guarded like `OllpMismatch`: only the data-group
+        // leader ran read-set validation, so only a leader's vote is
+        // authoritative. Currently observed-only — the registry tallies the
+        // vote (`CalvinCompletionRegistry::note_vote`) but nothing yet reads
+        // the tally to change flush/drop behavior; the local `committed`
+        // decision below still drives the resolve path unchanged.
+        if self.is_group_leader() {
+            self.propose_sequencer_entry(
+                SequencerEntry::Vote {
+                    epoch: txn_id.epoch,
+                    position: txn_id.position,
+                    vshard: self.vshard_id,
+                    commit: committed,
+                },
+                txn_id,
+                "commit vote",
+            );
+        }
+
         if !committed {
             // The staged slice's read-set was no longer current: observe it, the
             // same node-global signal the direct-apply path records.