voting_service: add a refresh queue for standstill messages (anza-xyz#13293)

Author: AshwinSekar

core/src/tvu.rs

@@ -35,7 +35,9 @@ use {
         event::{LatestSwitchRequest, LeaderWindowInfo, VotorEventReceiver, VotorEventSender},
         vote_history::VoteHistory,
         vote_history_storage::VoteHistoryStorage,
-        voting_service::{VotingService as BLSVotingService, VotingServiceOverride},
+        voting_service::{
+            VOTOR_RATE_LIMIT_PPS, VotingService as BLSVotingService, VotingServiceOverride,
+        },
         votor::{Votor, VotorConfig},
     },
     agave_votor_messages::{
@@ -311,7 +313,7 @@ impl Tvu {
                     ..Default::default()
                 };
                 let qos_config = SimpleQosConfig {
-                    max_streams_per_second: 30,
+                    max_streams_per_second: VOTOR_RATE_LIMIT_PPS,
                     // Cap by # of active validators (some overhead for epoch boundaries)
                     max_staked_connections: MAX_ALPENGLOW_VOTE_ACCOUNTS * 2,
                     // Two staked connection per validator to account for hotspares
@@ -522,7 +524,7 @@ impl Tvu {
             cluster_info: cluster_info.clone(),
             leader_schedule_cache: leader_schedule_cache.clone(),
             consensus_metrics_sender,
-            highest_finalized,
+            highest_finalized: highest_finalized.clone(),
             bank_forks_controller,
             bls_sender: bls_sender.clone(),
             commitment_sender: votor_commitment_sender,
@@ -615,6 +617,7 @@ impl Tvu {
             vote_history_storage,
             bls_connection_cache,
             bank_forks.clone(),
+            highest_finalized,
             voting_service_test_override,
         );
 

runtime/src/validated_block_finalization.rs

@@ -239,6 +239,20 @@ impl ValidatedBlockFinalizationCert {
         )
     }
 
+    /// Returns cloned copies of the certificates that prove this finalization.
+    ///
+    /// For slow finalization this returns the finalize certificate followed by the notarize
+    /// certificate. For fast finalization this returns the single fast-finalize certificate.
+    pub fn clone_certificates(&self) -> Vec<Certificate> {
+        match &self.kind {
+            ValidatedBlockFinalizationCertKind::Finalize {
+                finalize_cert,
+                notarize_cert,
+            } => vec![finalize_cert.clone(), notarize_cert.clone()],
+            ValidatedBlockFinalizationCertKind::FastFinalize(cert) => vec![cert.clone()],
+        }
+    }
+
     /// Returns the data needed to calculating and paying vote rewards.
     pub fn vote_rewards_input(&self) -> (&HashSet<Pubkey>, Slot) {
         (&self.signers, self.slot())
@@ -438,11 +452,10 @@ mod tests {
                 notar_aggregate: None,
             };
 
-            let result = ValidatedBlockFinalizationCert::try_from_footer(block_final_cert, &bank);
-            assert!(
-                result.is_ok(),
-                "Valid fast finalize certificate should pass verification: {result:?}"
-            );
+            let validated =
+                ValidatedBlockFinalizationCert::try_from_footer(block_final_cert, &bank)
+                    .expect("Valid fast finalize certificate should pass verification");
+            assert_eq!(validated.clone_certificates(), vec![fast_finalize_cert]);
         }
 
         // Test 2: Slow finalize (requires 60% stake = 3300 for both certs)
@@ -475,10 +488,12 @@ mod tests {
                 notar_aggregate: Some(VotesAggregate::from_certificate(&notarize_cert)),
             };
 
-            let result = ValidatedBlockFinalizationCert::try_from_footer(block_final_cert, &bank);
-            assert!(
-                result.is_ok(),
-                "Valid slow finalize certificate should pass verification: {result:?}"
+            let validated =
+                ValidatedBlockFinalizationCert::try_from_footer(block_final_cert, &bank)
+                    .expect("Valid slow finalize certificate should pass verification");
+            assert_eq!(
+                validated.clone_certificates(),
+                vec![finalize_cert, notarize_cert]
             );
         }
     }

votor-messages/src/certificate.rs

@@ -26,7 +26,7 @@ pod_wrapper! {
     derive(AbiExample),
     frozen_abi(digest = "5WqvPnvSnVXQFrAs9o29szFGDiCk45Pgk8K1evTZSrwo")
 )]
-#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, SchemaWrite, SchemaRead)]
+#[derive(Clone, Debug, PartialEq, Eq, Hash, Serialize, Deserialize, SchemaWrite, SchemaRead)]
 pub struct Certificate {
     /// The certificate type.
     pub cert_type: CertificateType,

votor-messages/src/consensus_message.rs

@@ -54,7 +54,7 @@ pub struct Block {
     derive(AbiExample),
     frozen_abi(digest = "CTiXEk2aQbpf6TS6PNKcaTsGkLruDvAYsTLFhHKW2vsm")
 )]
-#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, SchemaWrite, SchemaRead)]
+#[derive(Clone, Debug, PartialEq, Eq, Hash, Serialize, Deserialize, SchemaWrite, SchemaRead)]
 pub struct VoteMessage {
     /// The type of the vote.
     pub vote: Vote,
@@ -71,7 +71,7 @@ pub struct VoteMessage {
     derive(AbiExample, AbiEnumVisitor),
     frozen_abi(digest = "CbPatwRWz8NyUAj3HeAxAAWAWTxJnHGAfekLspUQpMHN")
 )]
-#[derive(Clone, Debug, PartialEq, Serialize, Deserialize, SchemaWrite, SchemaRead)]
+#[derive(Clone, Debug, PartialEq, Eq, Hash, Serialize, Deserialize, SchemaWrite, SchemaRead)]
 #[allow(clippy::large_enum_variant)]
 pub enum ConsensusMessage {
     /// A vote from a single party.
@@ -102,6 +102,14 @@ impl ConsensusMessage {
             bitmap,
         })
     }
+
+    /// Returns the slot this message is for.
+    pub fn slot(&self) -> Slot {
+        match self {
+            Self::Vote(vote) => vote.vote.slot(),
+            Self::Certificate(certificate) => certificate.cert_type.slot(),
+        }
+    }
 }
 
 impl From<Certificate> for ConsensusMessage {

votor/src/consensus_pool.rs

@@ -28,7 +28,7 @@ use {
     solana_hash::Hash,
     solana_pubkey::Pubkey,
     solana_runtime::{bank::Bank, validated_block_finalization::ValidatedBlockFinalizationCert},
-    std::{cmp::Ordering, collections::BTreeMap, num::NonZero, sync::Arc},
+    std::{collections::BTreeMap, num::NonZero, sync::Arc},
     thiserror::Error,
 };
 
@@ -629,33 +629,15 @@ impl ConsensusPool {
     }
 
     pub(crate) fn get_certs_for_standstill(&self) -> Vec<Arc<Certificate>> {
-        let (highest_slot, has_fast_finalize) = self
+        let highest_slot = self
             .highest_finalized_slot_cert
             .as_ref()
-            .map(|certs| (certs.slot(), certs.is_fast()))
-            .unwrap_or((0, false));
+            .map(ValidatedBlockFinalizationCert::slot)
+            .unwrap_or(0);
         self.completed_certificates
             .iter()
             .filter_map(|(cert_type, cert)| {
-                let cert_to_send = match (
-                    cert_type.slot().cmp(&highest_slot),
-                    cert_type,
-                    has_fast_finalize,
-                ) {
-                    (Ordering::Greater, _, _)
-                    | (
-                        Ordering::Equal,
-                        CertificateType::Finalize(_) | CertificateType::Notarize(_),
-                        false,
-                    )
-                    | (Ordering::Equal, CertificateType::FinalizeFast(_), true) => {
-                        Some(cert.clone())
-                    }
-                    (Ordering::Equal, CertificateType::FinalizeFast(_), false) => {
-                        panic!("Should not happen while certificate pool is single threaded")
-                    }
-                    _ => None,
-                };
+                let cert_to_send = (cert_type.slot() > highest_slot).then(|| cert.clone());
                 if cert_to_send.is_some() {
                     trace!(
                         "{}: Refreshing certificate {:?}",
@@ -1917,13 +1899,10 @@ mod tests {
             bitmap: dummy_bitmap(),
         };
         ctx.add_message(ConsensusMessage::Certificate(cert_5));
-        // Should return only FinalizeFast cert on 5
+        // Slot 5 is now the highest finalized slot, so standstill cert refresh only returns
+        // certificates for later slots.
         let certs = ctx.pool.get_certs_for_standstill();
-        assert_eq!(certs.len(), 1);
-        assert!(
-            certs[0].cert_type.slot() == 5
-                && matches!(certs[0].cert_type, CertificateType::FinalizeFast(_))
-        );
+        assert!(certs.is_empty());
 
         // Now add Notarize cert on 6
         let cert_6 = Certificate {
@@ -1935,11 +1914,9 @@ mod tests {
             bitmap: dummy_bitmap(),
         };
         ctx.add_message(ConsensusMessage::Certificate(cert_6));
-        // Should return certs on 5 and 6
+        // Should return certs after highest finalized slot 5.
         let certs = ctx.pool.get_certs_for_standstill();
-        assert_eq!(certs.len(), 2);
-        assert!(certs.iter().any(|cert| cert.cert_type.slot() == 5
-            && matches!(cert.cert_type, CertificateType::FinalizeFast(_))));
+        assert_eq!(certs.len(), 1);
         assert!(certs.iter().any(|cert| cert.cert_type.slot() == 6
             && matches!(cert.cert_type, CertificateType::Notarize(_))));
 
@@ -1960,14 +1937,10 @@ mod tests {
             bitmap: dummy_bitmap(),
         };
         ctx.add_message(ConsensusMessage::Certificate(cert_6_notarize_fallback));
-        // This should not be returned because 6 is the current highest finalized slot
-        // only Notarize/Finalze/FinalizeFast should be returned
+        // Slot 6 is now the current highest finalized slot, so no slot-6 certs should
+        // be returned by the queued refresh path.
         let certs = ctx.pool.get_certs_for_standstill();
-        assert_eq!(certs.len(), 2);
-        assert!(certs.iter().any(|cert| cert.cert_type.slot() == 6
-            && matches!(cert.cert_type, CertificateType::Finalize(_))));
-        assert!(certs.iter().any(|cert| cert.cert_type.slot() == 6
-            && matches!(cert.cert_type, CertificateType::Notarize(_))));
+        assert!(certs.is_empty());
 
         // Add another skip on 7
         let cert_7 = Certificate {
@@ -1976,13 +1949,9 @@ mod tests {
             bitmap: dummy_bitmap(),
         };
         ctx.add_message(ConsensusMessage::Certificate(cert_7));
-        // Should return certs on 6 and 7
+        // Should return certs after highest finalized slot 6.
         let certs = ctx.pool.get_certs_for_standstill();
-        assert_eq!(certs.len(), 3);
-        assert!(certs.iter().any(|cert| cert.cert_type.slot() == 6
-            && matches!(cert.cert_type, CertificateType::Finalize(_))));
-        assert!(certs.iter().any(|cert| cert.cert_type.slot() == 6
-            && matches!(cert.cert_type, CertificateType::Notarize(_))));
+        assert_eq!(certs.len(), 1);
         assert!(
             certs.iter().any(|cert| cert.cert_type.slot() == 7
                 && matches!(cert.cert_type, CertificateType::Skip(_)))
@@ -2005,13 +1974,10 @@ mod tests {
         };
         ctx.add_message(ConsensusMessage::Certificate(cert_8_notarize));
 
-        // Should only return certs on 8 now
+        // Slot 8 is now the current highest finalized slot, so latest-finalization
+        // certs are refreshed from highest_finalized instead of this queue.
         let certs = ctx.pool.get_certs_for_standstill();
-        assert_eq!(certs.len(), 2);
-        assert!(certs.iter().any(|cert| cert.cert_type.slot() == 8
-            && matches!(cert.cert_type, CertificateType::Finalize(_))));
-        assert!(certs.iter().any(|cert| cert.cert_type.slot() == 8
-            && matches!(cert.cert_type, CertificateType::Notarize(_))));
+        assert!(certs.is_empty());
     }
 
     #[test]