BLS cert all to all. (#300)

Co-authored-by: Ashwin Sekar <[email protected]>

Author: wen-coding

core/src/tvu.rs

@@ -22,7 +22,7 @@ use {
         window_service::{WindowService, WindowServiceChannels},
     },
     bytes::Bytes,
-    crossbeam_channel::{unbounded, Receiver, Sender},
+    crossbeam_channel::{bounded, unbounded, Receiver, Sender},
     solana_client::connection_cache::ConnectionCache,
     solana_geyser_plugin_manager::block_metadata_notifier_interface::BlockMetadataNotifierArc,
     solana_gossip::{
@@ -298,7 +298,10 @@ impl Tvu {
         let (cost_update_sender, cost_update_receiver) = unbounded();
         let (drop_bank_sender, drop_bank_receiver) = unbounded();
         let (voting_sender, voting_receiver) = unbounded();
-        let (bls_sender, bls_receiver) = unbounded();
+        // The BLS sender channel should be mostly used during standstill handling,
+        // there could be 10s/400ms = 25 slots, <=5 votes and <=5 certificates per slot,
+        // we cap the channel at 512 to give some headroom.
+        let (bls_sender, bls_receiver) = bounded(512);
 
         let replay_senders = ReplaySenders {
             rpc_subscriptions: rpc_subscriptions.clone(),

core/src/voting_service.rs

@@ -260,6 +260,18 @@ impl VotingService {
                     staked_validators_cache,
                 );
             }
+            BLSOp::PushCertificate { certificate } => {
+                let vote_slot = certificate.certificate.slot;
+                let bls_message = BLSMessage::Certificate((*certificate).clone());
+                Self::broadcast_alpenglow_message(
+                    vote_slot,
+                    cluster_info,
+                    &bls_message,
+                    connection_cache,
+                    additional_listeners,
+                    staked_validators_cache,
+                );
+            }
         }
     }
 
@@ -308,9 +320,11 @@ mod tests {
         super::*,
         crate::consensus::tower_storage::NullTowerStorage,
         alpenglow_vote::{
-            bls_message::{BLSMessage, VoteMessage},
+            bls_message::{BLSMessage, CertificateMessage, VoteMessage},
+            certificate::{Certificate, CertificateType},
             vote::Vote,
         },
+        bitvec::prelude::*,
         solana_bls_signatures::Signature as BLSSignature,
         solana_gossip::{cluster_info::ClusterInfo, contact_info::ContactInfo},
         solana_ledger::{
@@ -326,10 +340,7 @@ mod tests {
                 ValidatorVoteKeypairs,
             },
         },
-        solana_sdk::{
-            hash::Hash,
-            signer::{keypair::Keypair, Signer},
-        },
+        solana_sdk::signer::{keypair::Keypair, Signer},
         solana_streamer::{
             packet::{Packet, PacketBatch},
             recvmmsg::recv_mmsg,
@@ -342,6 +353,7 @@ mod tests {
             net::SocketAddr,
             sync::{atomic::AtomicBool, Arc, RwLock},
         },
+        test_case::test_case,
     };
 
     fn create_voting_service(
@@ -397,9 +409,41 @@ mod tests {
         )
     }
 
-    #[allow(clippy::disallowed_methods)]
-    #[test]
-    fn test_send_bls_message() {
+    #[test_case(BLSOp::PushVote {
+        bls_message: Arc::new(BLSMessage::Vote(VoteMessage {
+            vote: Vote::new_skip_vote(5),
+            signature: BLSSignature::default(),
+            rank: 1,
+        })),
+        slot: 5,
+        saved_vote_history: SavedVoteHistoryVersions::Current(SavedVoteHistory::default()),
+    }, BLSMessage::Vote(VoteMessage {
+        vote: Vote::new_skip_vote(5),
+        signature: BLSSignature::default(),
+        rank: 1,
+    }))]
+    #[test_case(BLSOp::PushCertificate {
+        certificate: Arc::new(CertificateMessage {
+            certificate: Certificate {
+                certificate_type: CertificateType::Skip,
+                slot: 5,
+                block_id: None,
+                replayed_bank_hash: None,
+            },
+            signature: BLSSignature::default(),
+            bitmap: BitVec::new(),
+        }),
+    }, BLSMessage::Certificate(CertificateMessage {
+        certificate: Certificate {
+            certificate_type: CertificateType::Skip,
+            slot: 5,
+            block_id: None,
+            replayed_bank_hash: None,
+        },
+        signature: BLSSignature::default(),
+        bitmap: BitVec::new(),
+    }))]
+    fn test_send_bls_message(bls_op: BLSOp, expected_bls_message: BLSMessage) {
         solana_logger::setup();
         let (_vote_sender, vote_receiver) = crossbeam_channel::unbounded();
         let (bls_sender, bls_receiver) = crossbeam_channel::unbounded();
@@ -413,19 +457,7 @@ mod tests {
         let _ = create_voting_service(vote_receiver, bls_receiver, listener_addr);
 
         // Send a BLS message via the VotingService
-        let bls_message = BLSMessage::Vote(VoteMessage {
-            vote: Vote::new_notarization_vote(5, Hash::new_unique(), Hash::new_unique()),
-            signature: BLSSignature::default(),
-            rank: 1,
-        });
-        let saved_vote_history = SavedVoteHistoryVersions::Current(SavedVoteHistory::default());
-        assert!(bls_sender
-            .send(BLSOp::PushVote {
-                bls_message: bls_message.clone(),
-                slot: 5,
-                saved_vote_history,
-            })
-            .is_ok());
+        assert!(bls_sender.send(bls_op).is_ok());
 
         // Wait for the listener to receive the message
         let mut packet_batch = PacketBatch::new(vec![Packet::default()]);
@@ -443,6 +475,6 @@ mod tests {
                     err
                 )
             });
-        assert_eq!(received_bls_message, bls_message);
+        assert_eq!(received_bls_message, expected_bls_message);
     }
 }