more aggressive deprecated CRDS filter (solana-labs#6048)

* more aggressive deprecated CRDS filter

Also split off into separate file as we are not backporting this.
---------

Co-authored-by: Greg Cusack <[email protected]>

Author: alexpyattaev

gossip/src/cluster_info.rs

@@ -19,6 +19,7 @@ use {
         contact_info::{self, ContactInfo, ContactInfoQuery, Error as ContactInfoError},
         crds::{Crds, Cursor, GossipRoute},
         crds_data::{self, CrdsData, EpochSlotsIndex, LowestSlot, SnapshotHashes, Vote, MAX_VOTES},
+        crds_filter::{should_retain_crds_value, GossipFilterDirection, MIN_STAKE_FOR_GOSSIP},
         crds_gossip::CrdsGossip,
         crds_gossip_error::CrdsGossipError,
         crds_gossip_pull::{
@@ -133,11 +134,6 @@ pub const DEFAULT_CONTACT_DEBUG_INTERVAL_MILLIS: u64 = 10_000;
 pub const DEFAULT_CONTACT_SAVE_INTERVAL_MILLIS: u64 = 60_000;
 // Limit number of unique pubkeys in the crds table.
 pub(crate) const CRDS_UNIQUE_PUBKEY_CAPACITY: usize = 8192;
-/// Minimum stake that a node should have so that its CRDS values are
-/// propagated through gossip (few types are exempted).
-const MIN_STAKE_FOR_GOSSIP: u64 = solana_native_token::LAMPORTS_PER_SOL;
-/// Minimum number of staked nodes for enforcing stakes in gossip.
-const MIN_NUM_STAKED_NODES: usize = 500;
 
 // Must have at least one socket to monitor the TVU port
 pub const MINIMUM_NUM_TVU_RECEIVE_SOCKETS: NonZeroUsize = NonZeroUsize::new(1).unwrap();
@@ -177,42 +173,6 @@ pub struct ClusterInfo {
     socket_addr_space: SocketAddrSpace,
 }
 
-// Returns false if the CRDS value should be discarded.
-#[inline]
-#[must_use]
-fn should_retain_crds_value(
-    value: &CrdsValue,
-    stakes: &HashMap<Pubkey, u64>,
-    drop_unstaked_node_instance: bool,
-) -> bool {
-    match value.data() {
-        CrdsData::ContactInfo(_) => true,
-        CrdsData::LegacyContactInfo(_) => true,
-        // May Impact new validators starting up without any stake yet.
-        CrdsData::Vote(_, _) => true,
-        // Unstaked nodes can still help repair.
-        CrdsData::EpochSlots(_, _) => true,
-        // Unstaked nodes can still serve snapshots.
-        CrdsData::LegacySnapshotHashes(_) | CrdsData::SnapshotHashes(_) => true,
-        // Otherwise unstaked voting nodes will show up with no version in
-        // the various dashboards.
-        CrdsData::Version(_) => true,
-        CrdsData::AccountsHashes(_) => true,
-        CrdsData::NodeInstance(_) if !drop_unstaked_node_instance => true,
-        CrdsData::LowestSlot(_, _)
-        | CrdsData::LegacyVersion(_)
-        | CrdsData::DuplicateShred(_, _)
-        | CrdsData::RestartHeaviestFork(_)
-        | CrdsData::RestartLastVotedForkSlots(_)
-        | CrdsData::NodeInstance(_) => {
-            stakes.len() < MIN_NUM_STAKED_NODES || {
-                let stake = stakes.get(&value.pubkey()).copied();
-                stake.unwrap_or_default() >= MIN_STAKE_FOR_GOSSIP
-            }
-        }
-    }
-}
-
 impl ClusterInfo {
     pub fn new(
         contact_info: ContactInfo,
@@ -1282,9 +1242,7 @@ impl ClusterInfo {
             self.flush_push_queue();
             self.gossip
                 .new_push_messages(&self_id, timestamp(), stakes, |value| {
-                    should_retain_crds_value(
-                        value, stakes, /*drop_unstaked_node_instance:*/ false,
-                    )
+                    should_retain_crds_value(value, stakes, GossipFilterDirection::EgressPush)
                 })
         };
         self.stats
@@ -1692,7 +1650,9 @@ impl ClusterInfo {
                 now,
                 |value| {
                     should_retain_crds_value(
-                        value, stakes, /*drop_unstaked_node_instance:*/ true,
+                        value,
+                        stakes,
+                        GossipFilterDirection::EgressPullResponse,
                     )
                 },
                 self.my_shred_version(),
@@ -2132,9 +2092,7 @@ impl ClusterInfo {
                 &mut protocol
             {
                 values.retain(|value| {
-                    should_retain_crds_value(
-                        value, stakes, /*drop_unstaked_node_instance:*/ false,
-                    )
+                    should_retain_crds_value(value, stakes, GossipFilterDirection::Ingress)
                 });
                 if values.is_empty() {
                     return None;
@@ -3003,7 +2961,7 @@ fn verify_gossip_addr<R: Rng + CryptoRng>(
         _ => return true, // If not a contact-info, nothing to verify.
     };
     // For (sufficiently) staked nodes, don't bother with ping/pong.
-    if stakes.get(pubkey) >= Some(&MIN_STAKE_FOR_GOSSIP) {
+    if stakes.get(pubkey).copied() >= Some(MIN_STAKE_FOR_GOSSIP) {
         return true;
     }
     // Invalid addresses are not verifiable.

gossip/src/crds_filter.rs

@@ -0,0 +1,71 @@
+use {
+    crate::{crds_data::CrdsData, crds_value::CrdsValue},
+    solana_pubkey::Pubkey,
+    std::collections::HashMap,
+};
+
+pub(crate) enum GossipFilterDirection {
+    Ingress,
+    EgressPush,
+    EgressPullResponse,
+}
+
+/// Minimum number of staked nodes for enforcing stakes in gossip.
+const MIN_NUM_STAKED_NODES: usize = 500;
+
+/// Minimum stake that a node should have so that all its CRDS values are
+/// propagated through gossip (below this only subset of CRDS is propagated).
+pub(crate) const MIN_STAKE_FOR_GOSSIP: u64 = solana_native_token::LAMPORTS_PER_SOL;
+
+/// Returns false if the CRDS value should be discarded.
+/// `direction` controls whether we are looking at
+/// incoming packet (via Push or PullResponse) or
+/// we are about to make a packet
+#[inline]
+#[must_use]
+pub(crate) fn should_retain_crds_value(
+    value: &CrdsValue,
+    stakes: &HashMap<Pubkey, u64>,
+    direction: GossipFilterDirection,
+) -> bool {
+    let retain_if_staked = || {
+        stakes.len() < MIN_NUM_STAKED_NODES || {
+            let stake = stakes.get(&value.pubkey()).copied();
+            stake.unwrap_or_default() >= MIN_STAKE_FOR_GOSSIP
+        }
+    };
+
+    use GossipFilterDirection::*;
+    match value.data() {
+        // All nodes can send ContactInfo
+        CrdsData::ContactInfo(_) => true,
+        // Unstaked nodes can still serve snapshots.
+        CrdsData::SnapshotHashes(_) => true,
+        // Consensus related messages only allowed for staked nodes
+        CrdsData::DuplicateShred(_, _)
+        | CrdsData::LowestSlot(0, _)
+        | CrdsData::RestartHeaviestFork(_)
+        | CrdsData::RestartLastVotedForkSlots(_) => retain_if_staked(),
+        // Unstaked nodes can technically send EpochSlots, but we do not want them
+        // eating gossip bandwidth
+        CrdsData::EpochSlots(_, _) => match direction {
+            // always store if we have received them
+            // to avoid getting them again in PullResponses
+            Ingress => true,
+            // only forward if the origin is staked
+            EgressPush | EgressPullResponse => retain_if_staked(),
+        },
+        CrdsData::Vote(_, _) => match direction {
+            Ingress | EgressPush => true,
+            EgressPullResponse => retain_if_staked(),
+        },
+        // Fully deprecated messages
+        CrdsData::AccountsHashes(_) => false,
+        CrdsData::LegacyContactInfo(_) => false,
+        CrdsData::LegacySnapshotHashes(_) => false,
+        CrdsData::LegacyVersion(_) => false,
+        CrdsData::LowestSlot(1.., _) => false,
+        CrdsData::NodeInstance(_) => false,
+        CrdsData::Version(_) => false,
+    }
+}

gossip/src/lib.rs

@@ -7,6 +7,7 @@ pub mod contact_info;
 pub mod crds;
 pub mod crds_data;
 pub mod crds_entry;
+mod crds_filter;
 pub mod crds_gossip;
 pub mod crds_gossip_error;
 pub mod crds_gossip_pull;