Add VoteTracker for tracking cluster's votes in gossip (#8327)

Track votes by slot in cluster_vote_listener

core/src/replay_stage.rs

@@ -2,6 +2,7 @@
 
 use crate::{
     cluster_info::ClusterInfo,
+    cluster_info_vote_listener::VoteTracker,
     commitment::{AggregateCommitmentService, BlockCommitmentCache, CommitmentAggregationData},
     consensus::{StakeLockout, Tower},
     poh_recorder::{PohRecorder, GRACE_TICKS_FACTOR, MAX_GRACE_SLOTS},
@@ -176,6 +177,7 @@ impl ReplayStage {
         cluster_info: Arc<RwLock<ClusterInfo>>,
         ledger_signal_receiver: Receiver<bool>,
         poh_recorder: Arc<Mutex<PohRecorder>>,
+        _vote_tracker: Arc<VoteTracker>,
     ) -> (Self, Receiver<Vec<Arc<Bank>>>) {
         let ReplayStageConfig {
             my_pubkey,

core/src/tpu.rs

@@ -5,14 +5,16 @@ use crate::{
     banking_stage::BankingStage,
     broadcast_stage::{BroadcastStage, BroadcastStageType},
     cluster_info::ClusterInfo,
-    cluster_info_vote_listener::ClusterInfoVoteListener,
+    cluster_info_vote_listener::{ClusterInfoVoteListener, VoteTracker},
     fetch_stage::FetchStage,
     poh_recorder::{PohRecorder, WorkingBankEntry},
     sigverify::TransactionSigVerifier,
     sigverify_stage::{DisabledSigVerifier, SigVerifyStage},
 };
 use crossbeam_channel::unbounded;
-use solana_ledger::{blockstore::Blockstore, blockstore_processor::TransactionStatusSender};
+use solana_ledger::{
+    bank_forks::BankForks, blockstore::Blockstore, blockstore_processor::TransactionStatusSender,
+};
 use std::{
     net::UdpSocket,
     sync::{
@@ -46,6 +48,8 @@ impl Tpu {
         broadcast_type: &BroadcastStageType,
         exit: &Arc<AtomicBool>,
         shred_version: u16,
+        vote_tracker: Arc<VoteTracker>,
+        bank_forks: Arc<RwLock<BankForks>>,
     ) -> Self {
         let (packet_sender, packet_receiver) = channel();
         let fetch_stage = FetchStage::new_with_sender(
@@ -72,6 +76,8 @@ impl Tpu {
             sigverify_disabled,
             verified_vote_sender,
             &poh_recorder,
+            vote_tracker,
+            bank_forks,
         );
 
         let banking_stage = BankingStage::new(

core/src/tvu.rs

@@ -4,6 +4,7 @@
 use crate::{
     blockstream_service::BlockstreamService,
     cluster_info::ClusterInfo,
+    cluster_info_vote_listener::VoteTracker,
     commitment::BlockCommitmentCache,
     ledger_cleanup_service::LedgerCleanupService,
     poh_recorder::PohRecorder,
@@ -88,6 +89,7 @@ impl Tvu {
         transaction_status_sender: Option<TransactionStatusSender>,
         rewards_recorder_sender: Option<RewardsRecorderSender>,
         snapshot_package_sender: Option<SnapshotPackageSender>,
+        vote_tracker: Arc<VoteTracker>,
     ) -> Self {
         let keypair: Arc<Keypair> = cluster_info
             .read()
@@ -171,6 +173,7 @@ impl Tvu {
             cluster_info.clone(),
             ledger_signal_receiver,
             poh_recorder.clone(),
+            vote_tracker,
         );
 
         let blockstream_service = if let Some(blockstream_unix_socket) = blockstream_unix_socket {
@@ -302,6 +305,7 @@ pub mod tests {
             None,
             None,
             None,
+            Arc::new(VoteTracker::new(&bank)),
         );
         exit.store(true, Ordering::Relaxed);
         tvu.join().unwrap();

core/src/validator.rs

@@ -3,6 +3,7 @@
 use crate::{
     broadcast_stage::BroadcastStageType,
     cluster_info::{ClusterInfo, Node},
+    cluster_info_vote_listener::VoteTracker,
     commitment::BlockCommitmentCache,
     contact_info::ContactInfo,
     gossip_service::{discover_cluster, GossipService},
@@ -378,6 +379,8 @@ impl Validator {
             "New shred signal for the TVU should be the same as the clear bank signal."
         );
 
+        let vote_tracker = Arc::new({ VoteTracker::new(bank_forks.read().unwrap().root_bank()) });
+
         let tvu = Tvu::new(
             vote_account,
             voting_keypair,
@@ -426,6 +429,7 @@ impl Validator {
             transaction_status_sender.clone(),
             rewards_recorder_sender,
             snapshot_package_sender,
+            vote_tracker.clone(),
         );
 
         if config.dev_sigverify_disabled {
@@ -445,6 +449,8 @@ impl Validator {
             &config.broadcast_stage_type,
             &exit,
             node.info.shred_version,
+            vote_tracker,
+            bank_forks,
         );
 
         datapoint_info!("validator-new", ("id", id.to_string(), String));

ledger/src/bank_forks.rs

@@ -122,6 +122,10 @@ impl BankForks {
         self.banks.get(&bank_slot)
     }
 
+    pub fn root_bank(&self) -> &Arc<Bank> {
+        self.banks.get(&self.root()).expect("Root bank must exist")
+    }
+
     pub fn new_from_banks(initial_forks: &[Arc<Bank>], rooted_path: Vec<Slot>) -> Self {
         let mut banks = HashMap::new();
         let working_bank = initial_forks[0].clone();

runtime/src/message_processor.rs

@@ -178,7 +178,7 @@ impl MessageProcessor {
             .accounts
             .iter()
             .map(|&index| {
-                let is_signer = index < message.header.num_required_signatures;
+                let is_signer = message.is_signer(index as usize);
                 let index = index as usize;
                 let key = &message.account_keys[index];
                 let account = &accounts[index];

sdk/src/message.rs

@@ -260,6 +260,10 @@ impl Message {
                     - self.header.num_readonly_unsigned_accounts as usize)
     }
 
+    pub fn is_signer(&self, i: usize) -> bool {
+        i < self.header.num_required_signatures as usize
+    }
+
     pub fn get_account_keys_by_lock_type(&self) -> (Vec<&Pubkey>, Vec<&Pubkey>) {
         let mut writable_keys = vec![];
         let mut readonly_keys = vec![];