report peers below score threshold

Author: jxs

protocols/gossipsub/src/behaviour.rs

@@ -164,6 +164,8 @@ pub enum Event {
         /// The types and amounts of failed messages that are occurring for this peer.
         failed_messages: FailedMessages,
     },
+    /// A Peer is below the score threshold.
+    LowScorePeers { peer_ids: Vec<PeerId> },
 }
 
 /// A data structure for storing configuration for publishing messages. See [`MessageAuthenticity`]
@@ -2139,7 +2141,9 @@ where
             let mesh_outbound_min = self.config.mesh_outbound_min_for_topic(topic_hash);
 
             // drop all peers with negative score, without PX
+            // report peers below the score report threshold.
             let mut removed_peers = 0;
+            let mut peers_to_report = vec![];
             peers.retain(|peer_id| {
                 let peer_score = scores.get(peer_id).map(|r| r.score).unwrap_or_default();
                 // Record the score per mesh
@@ -2148,6 +2152,12 @@ where
                     metrics.observe_mesh_peers_score(topic_hash, peer_score);
                 }
 
+                if let Some(threshold) = self.config.score_report_threshold() {
+                    if peer_score < threshold {
+                        peers_to_report.push(*peer_id);
+                    }
+                }
+
                 if peer_score < 0.0 {
                     tracing::debug!(
                         peer=%peer_id,
@@ -2166,6 +2176,13 @@ where
                 true
             });
 
+            if !peers_to_report.is_empty() {
+                self.events
+                    .push_back(ToSwarm::GenerateEvent(Event::LowScorePeers {
+                        peer_ids: peers_to_report,
+                    }));
+            }
+
             #[cfg(feature = "metrics")]
             if let Some(m) = self.metrics.as_mut() {
                 m.peers_removed(topic_hash, Churn::BadScore, removed_peers)

protocols/gossipsub/src/behaviour/tests.rs

@@ -6787,3 +6787,55 @@ fn test_validation_message_size_within_topic_specific() {
         _ => panic!("Unexpected event"),
     }
 }
+
+#[test]
+fn test_low_score_peer_is_reported() {
+    let score_report_threshold = -5.0;
+
+    let config = ConfigBuilder::default()
+        .score_report_threshold(score_report_threshold)
+        .build()
+        .expect("valid config");
+
+    let (mut gs, peers, _receivers, _topics) = inject_nodes1()
+        .peer_no(3)
+        .topics(vec!["test".into()])
+        .to_subscribe(true)
+        .gs_config(config)
+        .scoring(Some((
+            PeerScoreParams::default(),
+            PeerScoreThresholds::default(),
+        )))
+        .create_network();
+
+    // Reduce the score of the first peer below the threshold
+    gs.as_peer_score_mut().add_penalty(&peers[0], 10);
+
+    // Reduce the score of the second peer below the threshold
+    gs.as_peer_score_mut().add_penalty(&peers[1], 8);
+
+    // Verify initial scores are below threshold
+    assert!(gs.as_peer_score_mut().score_report(&peers[0]).score < score_report_threshold);
+    assert!(gs.as_peer_score_mut().score_report(&peers[1]).score < score_report_threshold);
+    assert!(gs.as_peer_score_mut().score_report(&peers[2]).score >= score_report_threshold);
+
+    // Trigger a heartbeat which should generate the LowScorePeers event
+    gs.heartbeat();
+
+    // Check for the LowScorePeers event
+    let low_score_event = gs
+        .events
+        .iter()
+        .find(|event| matches!(event, ToSwarm::GenerateEvent(Event::LowScorePeers { .. })))
+        .unwrap();
+
+    match low_score_event {
+        ToSwarm::GenerateEvent(Event::LowScorePeers { peer_ids }) => {
+            assert_eq!(peer_ids.len(), 2);
+            assert!(peer_ids.contains(&peers[0]));
+            assert!(peer_ids.contains(&peers[1]));
+            assert!(!peer_ids.contains(&peers[2]));
+        }
+        _ => unreachable!(),
+    }
+}

protocols/gossipsub/src/config.rs

@@ -132,6 +132,7 @@ pub struct Config {
     connection_handler_forward_duration: Duration,
     idontwant_message_size_threshold: usize,
     idontwant_on_publish: bool,
+    score_report_threshold: Option<f64>,
     topic_configuration: TopicConfigs,
 }
 
@@ -476,6 +477,12 @@ impl Config {
     pub fn idontwant_on_publish(&self) -> bool {
         self.idontwant_on_publish
     }
+
+    /// Score threshold below which peers should be reported. If set to None, no peer reporting
+    /// based on score will occur. Default is None.
+    pub fn score_report_threshold(&self) -> Option<f64> {
+        self.score_report_threshold
+    }
 }
 
 impl Default for Config {
@@ -545,6 +552,7 @@ impl Default for ConfigBuilder {
                 connection_handler_forward_duration: Duration::from_secs(1),
                 idontwant_message_size_threshold: 1000,
                 idontwant_on_publish: false,
+                score_report_threshold: None,
                 topic_configuration: TopicConfigs::default(),
             },
             invalid_protocol: false,
@@ -1040,6 +1048,14 @@ impl ConfigBuilder {
         self
     }
 
+    /// Sets the score threshold below which peers should be reported.
+    /// When set, the behaviour will report peers whose score falls below this threshold.
+    /// Default is None (no reporting).
+    pub fn score_report_threshold(&mut self, threshold: f64) -> &mut Self {
+        self.config.score_report_threshold = Some(threshold);
+        self
+    }
+
     /// The topic configuration sets mesh parameter sizes for a given topic. Notes on default
     /// below.
     ///