Pr 13059 atomic commits

Cargo.toml

@@ -308,6 +308,7 @@ protobuf-src = "1.1.0"
 protosol = "=8.2.0"
 qualifier_attr = { version = "0.2.2", default-features = false }
 quinn = "0.11.9"
+quinn-proto = "0.11.14"
 rand = "0.9.4"
 rand_chacha = "0.9.0"
 rayon = "1.12.0"

core/src/admin_rpc_post_init.rs

@@ -30,10 +30,9 @@ pub enum KeyUpdaterType {
     Forward,
     /// For the RPC service
     RpcService,
-    /// BLS all-to-all streamer key updater
-    Bls,
-    /// BLS all-to-all connection cache key updater
-    BlsConnectionCache,
+    /// Votor QUIC datagram endpoint key updater (single endpoint multiplexes
+    /// inbound consensus messages and outbound votes/certs).
+    VotorDatagram,
 }
 
 /// Responsible for managing the updaters for identity key change

core/src/bls_sigverify/bls_cert_sigverify.rs

@@ -1,5 +1,9 @@
 use {
-    super::{bls_sigverifier::BAN_TIMEOUT, errors::SigVerifyCertError, stats::SigVerifyCertStats},
+    super::{
+        bls_sigverifier::{BAN_TIMEOUT, SigVerifierBanlist},
+        errors::SigVerifyCertError,
+        stats::SigVerifyCertStats,
+    },
     crate::bls_sigverify::{bls_sigverifier::NUM_SLOTS_FOR_VERIFY, utils::send_certs_to_pool},
     agave_bls_cert_verify::cert_verify::Error as BlsCertVerifyError,
     agave_votor_messages::{
@@ -16,7 +20,6 @@ use {
     solana_measure::measure::Measure,
     solana_pubkey::Pubkey,
     solana_runtime::bank::Bank,
-    solana_streamer::nonblocking::simple_qos::SimpleQosBanlist,
     std::{collections::HashSet, num::NonZeroU64},
     thiserror::Error,
 };
@@ -53,7 +56,7 @@ pub(super) fn verify_and_send_certificates(
     certs: Vec<CertPayload>,
     root_bank: &Bank,
     channel_to_pool: &Sender<SigVerifiedBatch>,
-    banlist: &SimpleQosBanlist,
+    banlist: &SigVerifierBanlist,
     thread_pool: &ThreadPool,
 ) -> Result<SigVerifyCertStats, SigVerifyCertError> {
     for cert in certs.iter().map(|cert_payload| &cert_payload.cert) {
@@ -95,7 +98,7 @@ fn verify_certs(
     root_bank: &Bank,
     verified_certs_set: &mut HashSet<CertificateType>,
     stats: &mut SigVerifyCertStats,
-    banlist: &SimpleQosBanlist,
+    banlist: &SigVerifierBanlist,
     thread_pool: &ThreadPool,
 ) -> SigVerifiedBatch {
     let verified = thread_pool.install(|| {

core/src/bls_sigverify/bls_sigverifier.rs

@@ -11,6 +11,7 @@ use {
         block_creation_loop::rewards::{certs_builder::wants_vote, msg_types::AddVoteMessage},
         cluster_info_vote_listener::VerifiedVoterSlotsSender,
     },
+    agave_votor::datagram_transport::endpoint::Datagram,
     agave_votor::{
         consensus_metrics::ConsensusMetricsEventSender, generated_cert_types::GeneratedCertTypes,
     },
@@ -26,9 +27,11 @@ use {
     solana_gossip::cluster_info::ClusterInfo,
     solana_ledger::leader_schedule_cache::LeaderScheduleCache,
     solana_measure::measure_us,
+    solana_net_utils::banlist::Banlist,
+    solana_perf::packet::{BytesPacket, Meta, PacketBatch},
     solana_pubkey::Pubkey,
     solana_runtime::{bank::Bank, bank_forks::SharableBanks},
-    solana_streamer::{nonblocking::simple_qos::SimpleQosBanlist, packet::PacketBatch},
+    solana_streamer::nonblocking::simple_qos::SimpleQosBanlist,
     std::{
         collections::HashSet,
         sync::{
@@ -50,9 +53,61 @@ pub(super) const NUM_SLOTS_FOR_VERIFY: Slot = 90_000;
 /// We ban the sender for 2 days which roughly corresponds to an epoch
 pub(super) const BAN_TIMEOUT: Duration = Duration::from_hours(48);
 
+#[derive(Clone)]
+pub(crate) enum SigVerifierBanlist {
+    Stream(Arc<SimpleQosBanlist>),
+    Datagram(Arc<Banlist<Pubkey>>),
+}
+
+impl SigVerifierBanlist {
+    pub(crate) fn ban(&self, pubkey: Pubkey, timeout: Duration) -> bool {
+        match self {
+            Self::Stream(banlist) => banlist.ban(pubkey, timeout),
+            Self::Datagram(banlist) => banlist.ban(pubkey, timeout),
+        }
+    }
+
+    #[cfg(test)]
+    fn is_banned(&self, pubkey: &Pubkey) -> bool {
+        match self {
+            Self::Stream(banlist) => banlist.is_banned(pubkey),
+            Self::Datagram(banlist) => banlist.is_banned(pubkey),
+        }
+    }
+}
+
+impl From<Arc<SimpleQosBanlist>> for SigVerifierBanlist {
+    fn from(banlist: Arc<SimpleQosBanlist>) -> Self {
+        Self::Stream(banlist)
+    }
+}
+
+impl From<Arc<Banlist<Pubkey>>> for SigVerifierBanlist {
+    fn from(banlist: Arc<Banlist<Pubkey>>) -> Self {
+        Self::Datagram(banlist)
+    }
+}
+
+pub(crate) enum SigVerifierPacketReceiver {
+    Stream(Receiver<PacketBatch>),
+    Datagram(Receiver<Datagram>),
+}
+
+impl From<Receiver<PacketBatch>> for SigVerifierPacketReceiver {
+    fn from(receiver: Receiver<PacketBatch>) -> Self {
+        Self::Stream(receiver)
+    }
+}
+
+impl From<Receiver<Datagram>> for SigVerifierPacketReceiver {
+    fn from(receiver: Receiver<Datagram>) -> Self {
+        Self::Datagram(receiver)
+    }
+}
+
 pub(crate) struct SigVerifierContext {
     pub(crate) migration_status: Arc<MigrationStatus>,
-    pub(crate) banlist: Arc<SimpleQosBanlist>,
+    pub(crate) banlist: SigVerifierBanlist,
     pub(crate) sharable_banks: SharableBanks,
     pub(crate) cluster_info: Arc<ClusterInfo>,
     pub(crate) leader_schedule: Arc<LeaderScheduleCache>,
@@ -61,7 +116,7 @@ pub(crate) struct SigVerifierContext {
 }
 
 pub(crate) struct SigVerifierChannels {
-    pub(crate) packet_receiver: Receiver<PacketBatch>,
+    pub(crate) packet_receiver: SigVerifierPacketReceiver,
     pub(crate) channel_to_repair: VerifiedVoterSlotsSender,
     pub(crate) channel_to_reward: Sender<AddVoteMessage>,
     pub(crate) channel_to_pool: Sender<SigVerifiedBatch>,
@@ -84,7 +139,7 @@ pub(crate) fn spawn_service(
 
 struct SigVerifier {
     migration_status: Arc<MigrationStatus>,
-    banlist: Arc<SimpleQosBanlist>,
+    banlist: SigVerifierBanlist,
     channels: SigVerifierChannels,
     /// Container to look up root banks from.
     sharable_banks: SharableBanks,
@@ -294,31 +349,78 @@ impl SigVerifier {
     }
 }
 
-/// Receives a `Vec<PacketBatch>` from the `receiver` while adhering to the `soft_receive_cap` limit.
+fn datagram_to_batch(datagram: Datagram) -> PacketBatch {
+    let Datagram {
+        peer_pubkey,
+        peer_address,
+        message,
+    } = datagram;
+    let mut meta = Meta::default();
+    meta.size = message.len();
+    meta.set_socket_addr(&peer_address);
+    meta.set_remote_pubkey(peer_pubkey);
+    PacketBatch::Single(BytesPacket::new(message, meta))
+}
+
+/// Receives packet batches from the configured BLS receiver while adhering to
+/// the `soft_receive_cap` limit.
 ///
 /// Returns `Err(())` if the channel disconnected.
 fn recv_batches(
+    receiver: &SigVerifierPacketReceiver,
+    soft_receive_cap: usize,
+) -> Result<Vec<PacketBatch>, ()> {
+    match receiver {
+        SigVerifierPacketReceiver::Stream(receiver) => {
+            recv_stream_batches(receiver, soft_receive_cap)
+        }
+        SigVerifierPacketReceiver::Datagram(receiver) => {
+            recv_datagram_batches(receiver, soft_receive_cap)
+        }
+    }
+}
+
+fn recv_stream_batches(
     receiver: &Receiver<PacketBatch>,
     soft_receive_cap: usize,
 ) -> Result<Vec<PacketBatch>, ()> {
     let batch = match receiver.recv_timeout(Duration::from_secs(1)) {
         Ok(b) => b,
         Err(e) => match e {
-            RecvTimeoutError::Timeout => {
-                return Ok(vec![]);
-            }
-            RecvTimeoutError::Disconnected => {
-                return Err(());
-            }
+            RecvTimeoutError::Timeout => return Ok(vec![]),
+            RecvTimeoutError::Disconnected => return Err(()),
         },
     };
     let mut batches = Vec::with_capacity(soft_receive_cap);
     batches.push(batch);
     while batches.len() < soft_receive_cap {
         match receiver.try_recv() {
-            Ok(b) => {
-                batches.push(b);
-            }
+            Ok(b) => batches.push(b),
+            Err(e) => match e {
+                TryRecvError::Empty => return Ok(batches),
+                TryRecvError::Disconnected => return Err(()),
+            },
+        }
+    }
+    Ok(batches)
+}
+
+fn recv_datagram_batches(
+    receiver: &Receiver<Datagram>,
+    soft_receive_cap: usize,
+) -> Result<Vec<PacketBatch>, ()> {
+    let batch = match receiver.recv_timeout(Duration::from_secs(1)) {
+        Ok(b) => datagram_to_batch(b),
+        Err(e) => match e {
+            RecvTimeoutError::Timeout => return Ok(vec![]),
+            RecvTimeoutError::Disconnected => return Err(()),
+        },
+    };
+    let mut batches = Vec::with_capacity(soft_receive_cap);
+    batches.push(batch);
+    while batches.len() < soft_receive_cap {
+        match receiver.try_recv() {
+            Ok(b) => batches.push(datagram_to_batch(b)),
             Err(e) => match e {
                 TryRecvError::Empty => return Ok(batches),
                 TryRecvError::Disconnected => return Err(()),
@@ -371,7 +473,7 @@ mod tests {
     struct TestContext {
         verifier: SigVerifier,
         validator_keypairs: Vec<ValidatorVoteKeypairs>,
-        banlist: Arc<SimpleQosBanlist>,
+        banlist: SigVerifierBanlist,
 
         _packet_sender: Sender<PacketBatch>,
         repair_receiver: VerifiedVoterSlotsReceiver,
@@ -427,15 +529,15 @@ mod tests {
             let verifier = SigVerifier::new(
                 SigVerifierContext {
                     migration_status: Arc::new(MigrationStatus::default()),
-                    banlist: banlist.clone(),
+                    banlist: banlist.clone().into(),
                     sharable_banks,
                     cluster_info,
                     leader_schedule,
                     num_threads: 4,
                     generated_cert_types: generated_cert_types.clone(),
                 },
                 SigVerifierChannels {
-                    packet_receiver,
+                    packet_receiver: SigVerifierPacketReceiver::Stream(packet_receiver),
                     channel_to_repair,
                     channel_to_reward,
                     channel_to_pool,
@@ -445,7 +547,7 @@ mod tests {
             Self {
                 validator_keypairs,
                 verifier,
-                banlist,
+                banlist: banlist.into(),
                 _packet_sender: packet_sender,
                 repair_receiver,
                 _reward_receiver: reward_receiver,
@@ -1385,15 +1487,15 @@ mod tests {
         let mut sig_verifier = SigVerifier::new(
             SigVerifierContext {
                 migration_status: Arc::new(MigrationStatus::default()),
-                banlist: new_test_banlist(),
+                banlist: new_test_banlist().into(),
                 sharable_banks,
                 cluster_info,
                 leader_schedule,
                 num_threads: 4,
                 generated_cert_types: Arc::new(GeneratedCertTypes::default()),
             },
             SigVerifierChannels {
-                packet_receiver,
+                packet_receiver: SigVerifierPacketReceiver::Stream(packet_receiver),
                 channel_to_repair: votes_for_repair_sender,
                 channel_to_reward: reward_votes_sender,
                 channel_to_pool: message_sender,

core/src/bls_sigverify/bls_vote_sigverify.rs

@@ -5,7 +5,9 @@ use {
     crate::{
         block_creation_loop::rewards::msg_types::AddVoteMessage,
         bls_sigverify::{
-            bls_sigverifier::{BAN_TIMEOUT, NUM_SLOTS_FOR_VERIFY, SigVerifierChannels},
+            bls_sigverifier::{
+                BAN_TIMEOUT, NUM_SLOTS_FOR_VERIFY, SigVerifierBanlist, SigVerifierChannels,
+            },
             utils::{
                 send_votes_to_metrics, send_votes_to_pool, send_votes_to_repair,
                 send_votes_to_rewards,
@@ -32,7 +34,6 @@ use {
     solana_measure::{measure::Measure, measure_us},
     solana_pubkey::Pubkey,
     solana_runtime::bank::Bank,
-    solana_streamer::nonblocking::simple_qos::SimpleQosBanlist,
     std::{collections::HashMap, sync::Arc},
 };
 
@@ -76,7 +77,7 @@ pub(super) fn verify_and_send_votes(
     root_bank: &Bank,
     cluster_info: &ClusterInfo,
     leader_schedule: &LeaderScheduleCache,
-    banlist: &SimpleQosBanlist,
+    banlist: &SigVerifierBanlist,
     thread_pool: &ThreadPool,
     channels: &SigVerifierChannels,
 ) -> Result<SigVerifyVoteStats, SigVerifyVoteError> {
@@ -180,7 +181,7 @@ fn verify_votes(
     root_bank: &Bank,
     votes_to_verify: Vec<VotePayload>,
     stats: &mut SigVerifyVoteStats,
-    banlist: &SimpleQosBanlist,
+    banlist: &SigVerifierBanlist,
     thread_pool: &ThreadPool,
 ) -> Vec<VotePayload> {
     // Filter votes too far in the future.