address review

Author: jxs

Cargo.lock

@@ -124,6 +124,7 @@ fnv = "1"
 fs2 = "0.4"
 futures = "0.3"
 hex = "0.4"
+hashlink = "0.9.0"
 hyper = "1"
 itertools = "0.10"
 lazy_static = "1"

Cargo.toml

@@ -38,7 +38,7 @@ tracing = "0.1.37"
 void = "1.0.2"
 
 prometheus-client = "0.22.0"
-lru.workspace = true
+hashlink.workspace = true
 
 [dev-dependencies]
 quickcheck = { workspace = true }

beacon_node/gossipsub/Cargo.toml

@@ -25,18 +25,16 @@ use std::{
     collections::{BTreeSet, HashMap},
     fmt,
     net::IpAddr,
-    num::NonZeroUsize,
     task::{Context, Poll},
-    time::Duration,
+    time::{Duration, Instant},
 };
 
 use futures::StreamExt;
 use futures_ticker::Ticker;
-use lru::LruCache;
+use hashlink::LinkedHashMap;
 use prometheus_client::registry::Registry;
 use rand::{seq::SliceRandom, thread_rng};
 
-use instant::Instant;
 use libp2p::core::{multiaddr::Protocol::Ip4, multiaddr::Protocol::Ip6, Endpoint, Multiaddr};
 use libp2p::identity::Keypair;
 use libp2p::identity::PeerId;
@@ -78,8 +76,11 @@ use std::{cmp::Ordering::Equal, fmt::Debug};
 #[cfg(test)]
 mod tests;
 
-/// IDONTWANT Cache capacity.
-const IDONTWANT_CAP: usize = 100;
+/// 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.
 ///
@@ -1377,6 +1378,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
@@ -2485,6 +2491,17 @@ where
         }
         self.failed_messages.shrink_to_fit();
 
+        // Clear 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);
@@ -2677,9 +2694,18 @@ where
             return;
         };
 
-        let recipient_peers = mesh_peers.iter().filter(|peer_id| {
-            *peer_id != propagation_source && Some(*peer_id) != message.source.as_ref()
-        });
+        let iwant_peers = self
+            .peer_score
+            .as_ref()
+            .map(|(_peer_score, .., gossip_promises)| gossip_promises.peers_for_message(msg_id))
+            .unwrap_or(vec![]);
+
+        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 {
@@ -3121,7 +3147,7 @@ where
                 connections: vec![],
                 sender: RpcSender::new(self.config.connection_handler_queue_len()),
                 topics: Default::default(),
-                dont_send: LruCache::new(NonZeroUsize::new(IDONTWANT_CAP).unwrap()),
+                dont_send: LinkedHashMap::new(),
             });
         // Add the new connection
         connected_peer.connections.push(connection_id);
@@ -3152,7 +3178,7 @@ where
                 connections: vec![],
                 sender: RpcSender::new(self.config.connection_handler_queue_len()),
                 topics: Default::default(),
-                dont_send: LruCache::new(NonZeroUsize::new(IDONTWANT_CAP).unwrap()),
+                dont_send: LinkedHashMap::new(),
             });
         // Add the new connection
         connected_peer.connections.push(connection_id);
@@ -3319,7 +3345,11 @@ where
                                 continue;
                             };
                             for message_id in message_ids {
-                                peer.dont_send.push(message_id, ());
+                                peer.dont_send.insert(message_id, Instant::now());
+                                // Don't exceed capacity.
+                                if peer.dont_send.len() > IDONTWANT_CAP {
+                                    peer.dont_send.pop_front();
+                                }
                             }
                         }
                     }

beacon_node/gossipsub/src/behaviour.rs

@@ -41,6 +41,14 @@ impl GossipPromises {
         self.promises.contains_key(message)
     }
 
+    ///Get the peers we sent IWANT the input message id.
+    pub(crate) fn peers_for_message(&self, message_id: &MessageId) -> Vec<PeerId> {
+        self.promises
+            .get(message_id)
+            .map(|peers| peers.keys().copied().collect())
+            .unwrap_or(vec![])
+    }
+
     /// Track a promise to deliver a message from a list of [`MessageId`]s we are requesting.
     pub(crate) fn add_promise(&mut self, peer: PeerId, messages: &[MessageId], expires: Instant) {
         for message_id in messages {

beacon_node/gossipsub/src/gossip_promises.rs

@@ -43,7 +43,7 @@ pub(crate) const SIGNING_PREFIX: &[u8] = b"libp2p-pubsub:";
 
 pub(crate) const GOSSIPSUB_1_2_0_PROTOCOL: ProtocolId = ProtocolId {
     protocol: StreamProtocol::new("/meshsub/1.2.0"),
-    kind: PeerKind::Gossipsubv1_1,
+    kind: PeerKind::Gossipsubv1_2,
 };
 pub(crate) const GOSSIPSUB_1_1_0_PROTOCOL: ProtocolId = ProtocolId {
     protocol: StreamProtocol::new("/meshsub/1.1.0"),

beacon_node/gossipsub/src/protocol.rs

@@ -25,17 +25,18 @@ use async_channel::{Receiver, Sender};
 use futures::stream::Peekable;
 use futures::{Future, Stream, StreamExt};
 use futures_timer::Delay;
+use hashlink::LinkedHashMap;
 use instant::Duration;
 use libp2p::identity::PeerId;
 use libp2p::swarm::ConnectionId;
-use lru::LruCache;
 use prometheus_client::encoding::EncodeLabelValue;
 use quick_protobuf::MessageWrite;
 use std::collections::BTreeSet;
 use std::fmt::Debug;
 use std::sync::atomic::{AtomicUsize, Ordering};
 use std::sync::Arc;
 use std::task::{Context, Poll};
+use std::time::Instant;
 use std::{fmt, pin::Pin};
 
 use crate::rpc_proto::proto;
@@ -123,7 +124,7 @@ pub(crate) struct PeerConnections {
     /// Subscribed topics.
     pub(crate) topics: BTreeSet<TopicHash>,
     /// Don't send messages.
-    pub(crate) dont_send: LruCache<MessageId, ()>,
+    pub(crate) dont_send: LinkedHashMap<MessageId, Instant>,
 }
 
 /// Describes the types of peers that can exist in the gossipsub context.
@@ -300,10 +301,10 @@ pub struct Prune {
     pub(crate) backoff: Option<u64>,
 }
 
-/// The node requests us to not forward message ids (peer_id + sequence _number) - IDontWant control message.
+/// The node requests us to not forward message ids - IDontWant control message.
 #[derive(Debug, Clone, PartialEq, Eq, Hash)]
 pub struct IDontWant {
-    /// A list of known message ids (peer_id + sequence _number) as a string.
+    /// A list of known message ids.
     pub(crate) message_ids: Vec<MessageId>,
 }
 
@@ -568,10 +569,10 @@ impl From<Rpc> for proto::RPC {
                     control.prune.push(rpc_prune);
                 }
                 ControlAction::IDontWant(IDontWant { message_ids }) => {
-                    let rpc_iwant = proto::ControlIDontWant {
+                    let rpc_idontwant = proto::ControlIDontWant {
                         message_ids: message_ids.into_iter().map(|msg_id| msg_id.0).collect(),
                     };
-                    control.idontwant.push(rpc_iwant);
+                    control.idontwant.push(rpc_idontwant);
                 }
             }
         }