Implement gossipsub IDONTWANT (#5422)

* move gossipsub into a separate crate

* Merge branch 'unstable' of github.com:sigp/lighthouse into separate-gossipsub

* update rpc.proto and generate rust bindings

* gossipsub: implement IDONTWANT messages

* address review

* move GossipPromises out of PeerScore

* impl PeerKind::is_gossipsub

that returns true if peer speaks any version of gossipsub

* address review 2

* Merge branch 'separate-gossipsub' of github.com:sigp/lighthouse into impl-gossipsub-idontwant

* Merge branch 'unstable' of github.com:sigp/lighthouse into impl-gossipsub-idontwant

* add metrics

* add tests

* make 1.2 beta before spec is merged

* Merge branch 'unstable' of github.com:sigp/lighthouse into impl-gossipsub-idontwant

* cargo clippy

* Collect decoded IDONTWANT messages

* Use the beta tag in most places to simplify the transition

* Fix failed test by using fresh message-ids

* Gossipsub v1.2-beta

* Merge latest unstable

* Cargo update

* Merge pull request #5 from ackintosh/impl-gossipsub-idontwant-ackintosh-fix-test

Fix `test_ignore_too_many_messages_in_ihave` test

* Merge branch 'unstable' of github.com:sigp/lighthouse into impl-gossipsub-idontwant

* update CHANGELOG.md

* remove beta for 1.2 IDONTWANT spec has been merged

* Merge branch 'unstable' of github.com:sigp/lighthouse into impl-gossipsub-idontwant

* Merge branch 'impl-gossipsub-idontwant' of github.com:jxs/lighthouse into impl-gossipsub-idontwant

* Merge branch 'unstable' of github.com:sigp/lighthouse into impl-gossipsub-idontwant

* improve comments wording

* Merge branch 'impl-gossipsub-idontwant' of github.com:jxs/lighthouse into impl-gossipsub-idontwant

beacon_node/lighthouse_network/gossipsub/src/behaviour.rs

@@ -31,6 +31,7 @@ use std::{
 
 use futures::StreamExt;
 use futures_ticker::Ticker;
+use hashlink::LinkedHashMap;
 use prometheus_client::registry::Registry;
 use rand::{seq::SliceRandom, thread_rng};
 
@@ -45,6 +46,8 @@ use libp2p::swarm::{
 };
 use web_time::{Instant, SystemTime};
 
+use crate::types::IDontWant;
+
 use super::gossip_promises::GossipPromises;
 use super::handler::{Handler, HandlerEvent, HandlerIn};
 use super::mcache::MessageCache;
@@ -73,6 +76,12 @@ use std::{cmp::Ordering::Equal, fmt::Debug};
 #[cfg(test)]
 mod tests;
 
+/// IDONTWANT cache capacity.
+const IDONTWANT_CAP: usize = 10_000;
+
+/// IDONTWANT timeout before removal.
+const IDONTWANT_TIMEOUT: Duration = Duration::new(3, 0);
+
 /// Determines if published messages should be signed or not.
 ///
 /// Without signing, a number of privacy preserving modes can be selected.
@@ -304,9 +313,8 @@ pub struct Behaviour<D = IdentityTransform, F = AllowAllSubscriptionFilter> {
     /// discovery and not by PX).
     outbound_peers: HashSet<PeerId>,
 
-    /// Stores optional peer score data together with thresholds, decay interval and gossip
-    /// promises.
-    peer_score: Option<(PeerScore, PeerScoreThresholds, Ticker, GossipPromises)>,
+    /// Stores optional peer score data together with thresholds and decay interval.
+    peer_score: Option<(PeerScore, PeerScoreThresholds, Ticker)>,
 
     /// Counts the number of `IHAVE` received from each peer since the last heartbeat.
     count_received_ihave: HashMap<PeerId, usize>,
@@ -331,6 +339,9 @@ pub struct Behaviour<D = IdentityTransform, F = AllowAllSubscriptionFilter> {
 
     /// Tracks the numbers of failed messages per peer-id.
     failed_messages: HashMap<PeerId, FailedMessages>,
+
+    /// Tracks recently sent `IWANT` messages and checks if peers respond to them.
+    gossip_promises: GossipPromises,
 }
 
 impl<D, F> Behaviour<D, F>
@@ -467,6 +478,7 @@ where
             subscription_filter,
             data_transform,
             failed_messages: Default::default(),
+            gossip_promises: Default::default(),
         })
     }
 }
@@ -919,7 +931,7 @@ where
 
         let interval = Ticker::new(params.decay_interval);
         let peer_score = PeerScore::new_with_message_delivery_time_callback(params, callback);
-        self.peer_score = Some((peer_score, threshold, interval, GossipPromises::default()));
+        self.peer_score = Some((peer_score, threshold, interval));
         Ok(())
     }
 
@@ -1187,7 +1199,7 @@ where
     }
 
     fn score_below_threshold_from_scores(
-        peer_score: &Option<(PeerScore, PeerScoreThresholds, Ticker, GossipPromises)>,
+        peer_score: &Option<(PeerScore, PeerScoreThresholds, Ticker)>,
         peer_id: &PeerId,
         threshold: impl Fn(&PeerScoreThresholds) -> f64,
     ) -> (bool, f64) {
@@ -1248,10 +1260,7 @@ where
                 return false;
             }
 
-            self.peer_score
-                .as_ref()
-                .map(|(_, _, _, promises)| !promises.contains(id))
-                .unwrap_or(true)
+            !self.gossip_promises.contains(id)
         };
 
         for (topic, ids) in ihave_msgs {
@@ -1298,13 +1307,11 @@ where
             iwant_ids_vec.truncate(iask);
             *iasked += iask;
 
-            if let Some((_, _, _, gossip_promises)) = &mut self.peer_score {
-                gossip_promises.add_promise(
-                    *peer_id,
-                    &iwant_ids_vec,
-                    Instant::now() + self.config.iwant_followup_time(),
-                );
-            }
+            self.gossip_promises.add_promise(
+                *peer_id,
+                &iwant_ids_vec,
+                Instant::now() + self.config.iwant_followup_time(),
+            );
 
             if let Some(peer) = &mut self.connected_peers.get_mut(peer_id) {
                 tracing::trace!(
@@ -1369,6 +1376,11 @@ where
                         "IWANT: Peer has asked for message too many times; ignoring request"
                     );
                 } else if let Some(peer) = &mut self.connected_peers.get_mut(peer_id) {
+                    if peer.dont_send.get(&id).is_some() {
+                        tracing::debug!(%peer_id, message=%id, "Peer already sent IDONTWANT for this message");
+                        continue;
+                    }
+
                     tracing::debug!(peer=%peer_id, "IWANT: Sending cached messages to peer");
                     if peer
                         .sender
@@ -1706,14 +1718,15 @@ where
                 peer=%propagation_source,
                 "Rejecting message from blacklisted peer"
             );
-            if let Some((peer_score, .., gossip_promises)) = &mut self.peer_score {
+            self.gossip_promises
+                .reject_message(msg_id, &RejectReason::BlackListedPeer);
+            if let Some((peer_score, ..)) = &mut self.peer_score {
                 peer_score.reject_message(
                     propagation_source,
                     msg_id,
                     &raw_message.topic,
                     RejectReason::BlackListedPeer,
                 );
-                gossip_promises.reject_message(msg_id, &RejectReason::BlackListedPeer);
             }
             return false;
         }
@@ -1795,6 +1808,9 @@ where
         // Calculate the message id on the transformed data.
         let msg_id = self.config.message_id(&message);
 
+        // Broadcast IDONTWANT messages.
+        self.send_idontwant(&raw_message, &msg_id, propagation_source);
+
         // Check the validity of the message
         // Peers get penalized if this message is invalid. We don't add it to the duplicate cache
         // and instead continually penalize peers that repeatedly send this message.
@@ -1820,11 +1836,12 @@ where
             metrics.msg_recvd(&message.topic);
         }
 
-        // Tells score that message arrived (but is maybe not fully validated yet).
         // Consider the message as delivered for gossip promises.
-        if let Some((peer_score, .., gossip_promises)) = &mut self.peer_score {
+        self.gossip_promises.message_delivered(&msg_id);
+
+        // Tells score that message arrived (but is maybe not fully validated yet).
+        if let Some((peer_score, ..)) = &mut self.peer_score {
             peer_score.validate_message(propagation_source, &msg_id, &message.topic);
-            gossip_promises.message_delivered(&msg_id);
         }
 
         // Add the message to our memcache
@@ -1871,7 +1888,7 @@ where
         raw_message: &RawMessage,
         reject_reason: RejectReason,
     ) {
-        if let Some((peer_score, .., gossip_promises)) = &mut self.peer_score {
+        if let Some((peer_score, ..)) = &mut self.peer_score {
             if let Some(metrics) = self.metrics.as_mut() {
                 metrics.register_invalid_message(&raw_message.topic);
             }
@@ -1886,7 +1903,8 @@ where
                     reject_reason,
                 );
 
-                gossip_promises.reject_message(&message_id, &reject_reason);
+                self.gossip_promises
+                    .reject_message(&message_id, &reject_reason);
             } else {
                 // The message is invalid, we reject it ignoring any gossip promises. If a peer is
                 // advertising this message via an IHAVE and it's invalid it will be double
@@ -1959,7 +1977,7 @@ where
                     }
                     // if the mesh needs peers add the peer to the mesh
                     if !self.explicit_peers.contains(propagation_source)
-                        && matches!(peer.kind, PeerKind::Gossipsubv1_1 | PeerKind::Gossipsub)
+                        && peer.kind.is_gossipsub()
                         && !Self::score_below_threshold_from_scores(
                             &self.peer_score,
                             propagation_source,
@@ -2066,8 +2084,8 @@ where
 
     /// Applies penalties to peers that did not respond to our IWANT requests.
     fn apply_iwant_penalties(&mut self) {
-        if let Some((peer_score, .., gossip_promises)) = &mut self.peer_score {
-            for (peer, count) in gossip_promises.get_broken_promises() {
+        if let Some((peer_score, ..)) = &mut self.peer_score {
+            for (peer, count) in self.gossip_promises.get_broken_promises() {
                 peer_score.add_penalty(&peer, count);
                 if let Some(metrics) = self.metrics.as_mut() {
                     metrics.register_score_penalty(Penalty::BrokenPromise);
@@ -2288,7 +2306,7 @@ where
                 && peers.len() > 1
                 && self.peer_score.is_some()
             {
-                if let Some((_, thresholds, _, _)) = &self.peer_score {
+                if let Some((_, thresholds, _)) = &self.peer_score {
                     // Opportunistic grafting works as follows: we check the median score of peers
                     // in the mesh; if this score is below the opportunisticGraftThreshold, we
                     // select a few peers at random with score over the median.
@@ -2381,7 +2399,7 @@ where
         for (topic_hash, peers) in self.fanout.iter_mut() {
             let mut to_remove_peers = Vec::new();
             let publish_threshold = match &self.peer_score {
-                Some((_, thresholds, _, _)) => thresholds.publish_threshold,
+                Some((_, thresholds, _)) => thresholds.publish_threshold,
                 _ => 0.0,
             };
             for peer_id in peers.iter() {
@@ -2474,6 +2492,17 @@ where
         }
         self.failed_messages.shrink_to_fit();
 
+        // Flush stale IDONTWANTs.
+        for peer in self.connected_peers.values_mut() {
+            while let Some((_front, instant)) = peer.dont_send.front() {
+                if (*instant + IDONTWANT_TIMEOUT) >= Instant::now() {
+                    break;
+                } else {
+                    peer.dont_send.pop_front();
+                }
+            }
+        }
+
         tracing::debug!("Completed Heartbeat");
         if let Some(metrics) = self.metrics.as_mut() {
             let duration = u64::try_from(start.elapsed().as_millis()).unwrap_or(u64::MAX);
@@ -2655,6 +2684,59 @@ where
         }
     }
 
+    /// Helper function which sends an IDONTWANT message to mesh\[topic\] peers.
+    fn send_idontwant(
+        &mut self,
+        message: &RawMessage,
+        msg_id: &MessageId,
+        propagation_source: &PeerId,
+    ) {
+        let Some(mesh_peers) = self.mesh.get(&message.topic) else {
+            return;
+        };
+
+        let iwant_peers = self.gossip_promises.peers_for_message(msg_id);
+
+        let recipient_peers = mesh_peers
+            .iter()
+            .chain(iwant_peers.iter())
+            .filter(|peer_id| {
+                *peer_id != propagation_source && Some(*peer_id) != message.source.as_ref()
+            });
+
+        for peer_id in recipient_peers {
+            let Some(peer) = self.connected_peers.get_mut(peer_id) else {
+                tracing::error!(peer = %peer_id,
+                    "Could not IDONTWANT, peer doesn't exist in connected peer list");
+                continue;
+            };
+
+            // Only gossipsub 1.2 peers support IDONTWANT.
+            if peer.kind != PeerKind::Gossipsubv1_2_beta {
+                continue;
+            }
+
+            if peer
+                .sender
+                .idontwant(IDontWant {
+                    message_ids: vec![msg_id.clone()],
+                })
+                .is_err()
+            {
+                tracing::warn!(peer=%peer_id, "Send Queue full. Could not send IDONTWANT");
+
+                if let Some((peer_score, ..)) = &mut self.peer_score {
+                    peer_score.failed_message_slow_peer(peer_id);
+                }
+                // Increment failed message count
+                self.failed_messages
+                    .entry(*peer_id)
+                    .or_default()
+                    .non_priority += 1;
+            }
+        }
+    }
+
     /// Helper function which forwards a message to mesh\[topic\] peers.
     ///
     /// Returns true if at least one peer was messaged.
@@ -2708,6 +2790,11 @@ where
         if !recipient_peers.is_empty() {
             for peer_id in recipient_peers.iter() {
                 if let Some(peer) = self.connected_peers.get_mut(peer_id) {
+                    if peer.dont_send.get(msg_id).is_some() {
+                        tracing::debug!(%peer_id, message=%msg_id, "Peer doesn't want message");
+                        continue;
+                    }
+
                     tracing::debug!(%peer_id, message=%msg_id, "Sending message to peer");
                     if peer
                         .sender
@@ -3057,6 +3144,7 @@ where
                 connections: vec![],
                 sender: RpcSender::new(self.config.connection_handler_queue_len()),
                 topics: Default::default(),
+                dont_send: LinkedHashMap::new(),
             });
         // Add the new connection
         connected_peer.connections.push(connection_id);
@@ -3087,6 +3175,7 @@ where
                 connections: vec![],
                 sender: RpcSender::new(self.config.connection_handler_queue_len()),
                 topics: Default::default(),
+                dont_send: LinkedHashMap::new(),
             });
         // Add the new connection
         connected_peer.connections.push(connection_id);
@@ -3136,7 +3225,7 @@ where
             }
             HandlerEvent::MessageDropped(rpc) => {
                 // Account for this in the scoring logic
-                if let Some((peer_score, _, _, _)) = &mut self.peer_score {
+                if let Some((peer_score, _, _)) = &mut self.peer_score {
                     peer_score.failed_message_slow_peer(&propagation_source);
                 }
 
@@ -3245,6 +3334,24 @@ where
                             peers,
                             backoff,
                         }) => prune_msgs.push((topic_hash, peers, backoff)),
+                        ControlAction::IDontWant(IDontWant { message_ids }) => {
+                            let Some(peer) = self.connected_peers.get_mut(&propagation_source)
+                            else {
+                                tracing::error!(peer = %propagation_source,
+                                    "Could not handle IDONTWANT, peer doesn't exist in connected peer list");
+                                continue;
+                            };
+                            if let Some(metrics) = self.metrics.as_mut() {
+                                metrics.register_idontwant(message_ids.len());
+                            }
+                            for message_id in message_ids {
+                                peer.dont_send.insert(message_id, Instant::now());
+                                // Don't exceed capacity.
+                                if peer.dont_send.len() > IDONTWANT_CAP {
+                                    peer.dont_send.pop_front();
+                                }
+                            }
+                        }
                     }
                 }
                 if !ihave_msgs.is_empty() {
@@ -3270,7 +3377,7 @@ where
         }
 
         // update scores
-        if let Some((peer_score, _, interval, _)) = &mut self.peer_score {
+        if let Some((peer_score, _, interval)) = &mut self.peer_score {
             while let Poll::Ready(Some(_)) = interval.poll_next_unpin(cx) {
                 peer_score.refresh_scores();
             }
@@ -3395,7 +3502,7 @@ fn get_random_peers_dynamic(
         .iter()
         .filter(|(_, p)| p.topics.contains(topic_hash))
         .filter(|(peer_id, _)| f(peer_id))
-        .filter(|(_, p)| p.kind == PeerKind::Gossipsub || p.kind == PeerKind::Gossipsubv1_1)
+        .filter(|(_, p)| p.kind.is_gossipsub())
         .map(|(peer_id, _)| *peer_id)
         .collect::<Vec<PeerId>>();