Subnet discovery fixes (#2095)

## Issue Addressed

N/A

## Proposed Changes

Fixes multiple issues related to discovering of subnet peers.
1. Subnet discovery retries after yielding no results
2. Metadata updates if peer send older metadata
3. peerdb stores the peer subscriptions from gossipsub

beacon_node/eth2_libp2p/Cargo.toml

@@ -42,7 +42,7 @@ regex = "1.3.9"
 [dependencies.libp2p]
 #version = "0.23.0"
 git = "https://github.com/sigp/rust-libp2p"
-rev = "e3caf9e0e5e78c9d51c6dccf0d6277cef553bb25"
+rev = "830e6fabb7ee51281a98f5e092f056668adbef25"
 default-features = false
 features = ["websocket", "identify", "mplex", "yamux", "noise", "gossipsub", "dns", "tcp-tokio"]
 

beacon_node/eth2_libp2p/src/behaviour/mod.rs

@@ -5,7 +5,10 @@ use crate::peer_manager::{
 };
 use crate::rpc::*;
 use crate::service::METADATA_FILENAME;
-use crate::types::{GossipEncoding, GossipKind, GossipTopic, MessageData, SubnetDiscovery};
+use crate::types::{
+    subnet_id_from_topic_hash, GossipEncoding, GossipKind, GossipTopic, MessageData,
+    SubnetDiscovery,
+};
 use crate::Eth2Enr;
 use crate::{error, metrics, Enr, NetworkConfig, NetworkGlobals, PubsubMessage, TopicHash};
 use futures::prelude::*;
@@ -100,8 +103,6 @@ pub enum BehaviourEvent<TSpec: EthSpec> {
         /// The message itself.
         message: PubsubMessage<TSpec>,
     },
-    /// Subscribed to peer for given topic
-    PeerSubscribed(PeerId, TopicHash),
     /// Inform the network to send a Status to this peer.
     StatusPeer(PeerId),
 }
@@ -665,9 +666,15 @@ impl<TSpec: EthSpec> Behaviour<TSpec> {
                 }
             }
             GossipsubEvent::Subscribed { peer_id, topic } => {
-                self.add_event(BehaviourEvent::PeerSubscribed(peer_id, topic));
+                if let Some(subnet_id) = subnet_id_from_topic_hash(&topic) {
+                    self.peer_manager.add_subscription(&peer_id, subnet_id);
+                }
+            }
+            GossipsubEvent::Unsubscribed { peer_id, topic } => {
+                if let Some(subnet_id) = subnet_id_from_topic_hash(&topic) {
+                    self.peer_manager.remove_subscription(&peer_id, subnet_id);
+                }
             }
-            GossipsubEvent::Unsubscribed { .. } => {}
         }
     }
 
@@ -1102,6 +1109,9 @@ impl<TSpec: EthSpec> NetworkBehaviour for Behaviour<TSpec> {
     fn inject_disconnected(&mut self, peer_id: &PeerId) {
         // If the application/behaviour layers thinks this peer has connected inform it of the disconnect.
 
+        // Remove all subnet subscriptions from peerdb for the disconnected peer.
+        self.peer_manager().remove_all_subscriptions(&peer_id);
+
         if self
             .network_globals
             .peers

beacon_node/eth2_libp2p/src/discovery/mod.rs

@@ -767,6 +767,13 @@ impl<TSpec: EthSpec> Discovery<TSpec> {
                 match query_result.1 {
                     Ok(r) if r.is_empty() => {
                         debug!(self.log, "Grouped subnet discovery query yielded no results."; "subnets_searched_for" => ?subnets_searched_for);
+                        queries.iter().for_each(|query| {
+                            self.add_subnet_query(
+                                query.subnet_id,
+                                query.min_ttl,
+                                query.retries + 1,
+                            );
+                        })
                     }
                     Ok(r) => {
                         debug!(self.log, "Peer grouped subnet discovery request completed"; "peers_found" => r.len(), "subnets_searched_for" => ?subnets_searched_for);

beacon_node/eth2_libp2p/src/peer_manager/mod.rs

@@ -285,6 +285,27 @@ impl<TSpec: EthSpec> PeerManager<TSpec> {
         self.status_peers.insert(peer_id.clone());
     }
 
+    /// Adds a gossipsub subscription to a peer in the peerdb.
+    pub fn add_subscription(&self, peer_id: &PeerId, subnet_id: SubnetId) {
+        if let Some(info) = self.network_globals.peers.write().peer_info_mut(peer_id) {
+            info.subnets.insert(subnet_id);
+        }
+    }
+
+    /// Removes a gossipsub subscription to a peer in the peerdb.
+    pub fn remove_subscription(&self, peer_id: &PeerId, subnet_id: SubnetId) {
+        if let Some(info) = self.network_globals.peers.write().peer_info_mut(peer_id) {
+            info.subnets.remove(&subnet_id);
+        }
+    }
+
+    /// Removes all gossipsub subscriptions to a peer in the peerdb.
+    pub fn remove_all_subscriptions(&self, peer_id: &PeerId) {
+        if let Some(info) = self.network_globals.peers.write().peer_info_mut(peer_id) {
+            info.subnets = Default::default();
+        }
+    }
+
     /* Notifications from the Swarm */
 
     /// Updates the state of the peer as disconnected.
@@ -541,6 +562,9 @@ impl<TSpec: EthSpec> PeerManager<TSpec> {
                 } else {
                     debug!(self.log, "Received old metadata";
                         "peer_id" => %peer_id, "known_seq_no" => known_meta_data.seq_number, "new_seq_no" => meta_data.seq_number);
+                    // Updating metadata even in this case to prevent storing
+                    // incorrect  `metadata.attnets` for a peer
+                    peer_info.meta_data = Some(meta_data);
                 }
             } else {
                 // we have no meta-data for this peer, update

beacon_node/eth2_libp2p/src/peer_manager/peer_info.rs

@@ -38,6 +38,8 @@ pub struct PeerInfo<T: EthSpec> {
     /// The ENR subnet bitfield of the peer. This may be determined after it's initial
     /// connection.
     pub meta_data: Option<MetaData<T>>,
+    /// Subnets the peer is connected to.
+    pub subnets: HashSet<SubnetId>,
     /// The time we would like to retain this peer. After this time, the peer is no longer
     /// necessary.
     #[serde(skip)]
@@ -60,6 +62,7 @@ impl<TSpec: EthSpec> Default for PeerInfo<TSpec> {
             connection_status: Default::default(),
             listening_addresses: Vec::new(),
             seen_addresses: HashSet::new(),
+            subnets: HashSet::new(),
             sync_status: PeerSyncStatus::Unknown,
             meta_data: None,
             min_ttl: None,
@@ -80,14 +83,19 @@ impl<T: EthSpec> PeerInfo<T> {
         }
     }
 
-    /// Returns if the peer is subscribed to a given `SubnetId`
-    pub fn on_subnet(&self, subnet_id: SubnetId) -> bool {
+    /// Returns if the peer is subscribed to a given `SubnetId` from the metadata attnets field.
+    pub fn on_subnet_metadata(&self, subnet_id: SubnetId) -> bool {
         if let Some(meta_data) = &self.meta_data {
             return meta_data.attnets.get(*subnet_id as usize).unwrap_or(false);
         }
         false
     }
 
+    /// Returns if the peer is subscribed to a given `SubnetId` from the gossipsub subscriptions.
+    pub fn on_subnet_gossipsub(&self, subnet_id: SubnetId) -> bool {
+        self.subnets.contains(&subnet_id)
+    }
+
     /// Returns the seen IP addresses of the peer.
     pub fn seen_addresses<'a>(&'a self) -> impl Iterator<Item = IpAddr> + 'a {
         self.seen_addresses

beacon_node/eth2_libp2p/src/peer_manager/peerdb.rs

@@ -246,7 +246,11 @@ impl<TSpec: EthSpec> PeerDB<TSpec> {
         self.peers
             .iter()
             .filter(move |(_, info)| {
-                info.is_connected() && info.on_subnet(subnet_id) && info.is_good_gossipsub_peer()
+                // We check both the metadata and gossipsub data as we only want to count long-lived subscribed peers
+                info.is_connected()
+                    && info.on_subnet_metadata(subnet_id)
+                    && info.on_subnet_gossipsub(subnet_id)
+                    && info.is_good_gossipsub_peer()
             })
             .map(|(peer_id, _)| peer_id)
     }
@@ -357,7 +361,7 @@ impl<TSpec: EthSpec> PeerDB<TSpec> {
         let log = &self.log;
         self.peers.iter_mut()
             .filter(move |(_, info)| {
-                info.is_connected() && info.on_subnet(subnet_id)
+                info.is_connected() && info.on_subnet_metadata(subnet_id) && info.on_subnet_gossipsub(subnet_id)
             })
             .for_each(|(peer_id,info)| {
                 if info.min_ttl.is_none() || Some(min_ttl) > info.min_ttl {

beacon_node/eth2_libp2p/src/types/mod.rs

@@ -16,4 +16,4 @@ pub use globals::NetworkGlobals;
 pub use pubsub::{MessageData, PubsubMessage};
 pub use subnet::SubnetDiscovery;
 pub use sync_state::SyncState;
-pub use topics::{GossipEncoding, GossipKind, GossipTopic, CORE_TOPICS};
+pub use topics::{subnet_id_from_topic_hash, GossipEncoding, GossipKind, GossipTopic, CORE_TOPICS};

beacon_node/eth2_libp2p/src/types/topics.rs

@@ -1,4 +1,4 @@
-use libp2p::gossipsub::IdentTopic as Topic;
+use libp2p::gossipsub::{IdentTopic as Topic, TopicHash};
 use serde_derive::{Deserialize, Serialize};
 use types::SubnetId;
 
@@ -193,6 +193,15 @@ impl From<SubnetId> for GossipKind {
 
 // helper functions
 
+/// Get subnet id from an attestation subnet topic hash.
+pub fn subnet_id_from_topic_hash(topic_hash: &TopicHash) -> Option<SubnetId> {
+    let gossip_topic = GossipTopic::decode(topic_hash.as_str()).ok()?;
+    if let GossipKind::Attestation(subnet_id) = gossip_topic.kind() {
+        return Some(*subnet_id);
+    }
+    None
+}
+
 // Determines if a string is a committee topic.
 fn committee_topic_index(topic: &str) -> Option<SubnetId> {
     if topic.starts_with(BEACON_ATTESTATION_PREFIX) {
@@ -289,4 +298,16 @@ mod tests {
         // Empty parts
         assert!(GossipTopic::decode("////").is_err());
     }
+
+    #[test]
+    fn test_subnet_id_from_topic_hash() {
+        let topic_hash = TopicHash::from_raw("/eth2/e1925f3b/beacon_block/ssz_snappy");
+        assert!(subnet_id_from_topic_hash(&topic_hash).is_none());
+
+        let topic_hash = TopicHash::from_raw("/eth2/e1925f3b/beacon_attestation_42/ssz_snappy");
+        assert_eq!(
+            subnet_id_from_topic_hash(&topic_hash),
+            Some(SubnetId::new(42))
+        );
+    }
 }

beacon_node/network/src/service.rs

@@ -465,13 +465,13 @@ fn spawn_service<T: BeaconChainTypes>(
                                 // We currently do not perform any action here.
                             },
                             BehaviourEvent::PeerDisconnected(peer_id) => {
-                            let _ = service
-                                .router_send
-                                .send(RouterMessage::PeerDisconnected(peer_id))
-                                .map_err(|_| {
-                                    debug!(service.log, "Failed to send peer disconnect to router");
-                                });
-                            },
+                                let _ = service
+                                    .router_send
+                                    .send(RouterMessage::PeerDisconnected(peer_id))
+                                    .map_err(|_| {
+                                        debug!(service.log, "Failed to send peer disconnect to router");
+                                    });
+                                },
                             BehaviourEvent::RequestReceived{peer_id, id, request} => {
                                 let _ = service
                                     .router_send
@@ -543,7 +543,6 @@ fn spawn_service<T: BeaconChainTypes>(
                                     }
                                 }
                             }
-                            BehaviourEvent::PeerSubscribed(_, _) => {},
                         }
                         Libp2pEvent::NewListenAddr(multiaddr) => {
                             service.network_globals.listen_multiaddrs.write().push(multiaddr);