cerc-io/lighthouse
16
0
Fork 0
Code Issues 1 Pull Requests Packages Projects Releases Wiki Activity

Subnet discovery fixes (#2095)

Browse Source 
## 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
This commit is contained in:
Pawan Dhananjay 2020-12-17 00:39:15 +00:00
parent ca08fc7831
commit f998eff7ce
10 changed files with 110 additions and 37 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

34
Cargo.lock generated
View File

@ -3432,7 +3432,7 @@ dependencies = [
[[package]]
name = "libp2p"
version = "0.30.1"
source = "git+https://github.com/sigp/rust-libp2p?rev=e3caf9e0e5e78c9d51c6dccf0d6277cef553bb25#e3caf9e0e5e78c9d51c6dccf0d6277cef553bb25"
source = "git+https://github.com/sigp/rust-libp2p?rev=830e6fabb7ee51281a98f5e092f056668adbef25#830e6fabb7ee51281a98f5e092f056668adbef25"
dependencies = [
"atomic",
"bytes 0.5.6",
@ -3449,7 +3449,7 @@ dependencies = [
"libp2p-tcp",
"libp2p-websocket",
"libp2p-yamux",
"parity-multiaddr 0.9.6 (git+https://github.com/sigp/rust-libp2p?rev=e3caf9e0e5e78c9d51c6dccf0d6277cef553bb25)",
"parity-multiaddr 0.9.6 (git+https://github.com/sigp/rust-libp2p?rev=830e6fabb7ee51281a98f5e092f056668adbef25)",
"parking_lot",
"pin-project 1.0.2",
"smallvec",
@ -3493,7 +3493,7 @@ dependencies = [
[[package]]
name = "libp2p-core"
version = "0.25.0"
source = "git+https://github.com/sigp/rust-libp2p?rev=e3caf9e0e5e78c9d51c6dccf0d6277cef553bb25#e3caf9e0e5e78c9d51c6dccf0d6277cef553bb25"
source = "git+https://github.com/sigp/rust-libp2p?rev=830e6fabb7ee51281a98f5e092f056668adbef25#830e6fabb7ee51281a98f5e092f056668adbef25"
dependencies = [
"asn1_der",
"bs58 0.4.0",
@ -3507,8 +3507,8 @@ dependencies = [
"libsecp256k1",
"log",
"multihash 0.13.2",
"multistream-select 0.8.5 (git+https://github.com/sigp/rust-libp2p?rev=e3caf9e0e5e78c9d51c6dccf0d6277cef553bb25)",
"parity-multiaddr 0.9.6 (git+https://github.com/sigp/rust-libp2p?rev=e3caf9e0e5e78c9d51c6dccf0d6277cef553bb25)",
"multistream-select 0.8.5 (git+https://github.com/sigp/rust-libp2p?rev=830e6fabb7ee51281a98f5e092f056668adbef25)",
"parity-multiaddr 0.9.6 (git+https://github.com/sigp/rust-libp2p?rev=830e6fabb7ee51281a98f5e092f056668adbef25)",
"parking_lot",
"pin-project 1.0.2",
"prost",
@ -3527,7 +3527,7 @@ dependencies = [
[[package]]
name = "libp2p-core-derive"
version = "0.20.2"
source = "git+https://github.com/sigp/rust-libp2p?rev=e3caf9e0e5e78c9d51c6dccf0d6277cef553bb25#e3caf9e0e5e78c9d51c6dccf0d6277cef553bb25"
source = "git+https://github.com/sigp/rust-libp2p?rev=830e6fabb7ee51281a98f5e092f056668adbef25#830e6fabb7ee51281a98f5e092f056668adbef25"
dependencies = [
"quote",
"syn",
@ -3536,7 +3536,7 @@ dependencies = [
[[package]]
name = "libp2p-dns"
version = "0.25.0"
source = "git+https://github.com/sigp/rust-libp2p?rev=e3caf9e0e5e78c9d51c6dccf0d6277cef553bb25#e3caf9e0e5e78c9d51c6dccf0d6277cef553bb25"
source = "git+https://github.com/sigp/rust-libp2p?rev=830e6fabb7ee51281a98f5e092f056668adbef25#830e6fabb7ee51281a98f5e092f056668adbef25"
dependencies = [
"futures 0.3.8",
"libp2p-core 0.25.0",
@ -3546,7 +3546,7 @@ dependencies = [
[[package]]
name = "libp2p-gossipsub"
version = "0.25.0"
source = "git+https://github.com/sigp/rust-libp2p?rev=e3caf9e0e5e78c9d51c6dccf0d6277cef553bb25#e3caf9e0e5e78c9d51c6dccf0d6277cef553bb25"
source = "git+https://github.com/sigp/rust-libp2p?rev=830e6fabb7ee51281a98f5e092f056668adbef25#830e6fabb7ee51281a98f5e092f056668adbef25"
dependencies = [
"base64 0.13.0",
"byteorder",
@ -3570,7 +3570,7 @@ dependencies = [
[[package]]
name = "libp2p-identify"
version = "0.25.0"
source = "git+https://github.com/sigp/rust-libp2p?rev=e3caf9e0e5e78c9d51c6dccf0d6277cef553bb25#e3caf9e0e5e78c9d51c6dccf0d6277cef553bb25"
source = "git+https://github.com/sigp/rust-libp2p?rev=830e6fabb7ee51281a98f5e092f056668adbef25#830e6fabb7ee51281a98f5e092f056668adbef25"
dependencies = [
"futures 0.3.8",
"libp2p-core 0.25.0",
@ -3585,7 +3585,7 @@ dependencies = [
[[package]]
name = "libp2p-mplex"
version = "0.25.0"
source = "git+https://github.com/sigp/rust-libp2p?rev=e3caf9e0e5e78c9d51c6dccf0d6277cef553bb25#e3caf9e0e5e78c9d51c6dccf0d6277cef553bb25"
source = "git+https://github.com/sigp/rust-libp2p?rev=830e6fabb7ee51281a98f5e092f056668adbef25#830e6fabb7ee51281a98f5e092f056668adbef25"
dependencies = [
"bytes 0.5.6",
"futures 0.3.8",
@ -3602,7 +3602,7 @@ dependencies = [
[[package]]
name = "libp2p-noise"
version = "0.27.0"
source = "git+https://github.com/sigp/rust-libp2p?rev=e3caf9e0e5e78c9d51c6dccf0d6277cef553bb25#e3caf9e0e5e78c9d51c6dccf0d6277cef553bb25"
source = "git+https://github.com/sigp/rust-libp2p?rev=830e6fabb7ee51281a98f5e092f056668adbef25#830e6fabb7ee51281a98f5e092f056668adbef25"
dependencies = [
"bytes 0.5.6",
"curve25519-dalek",
@ -3623,7 +3623,7 @@ dependencies = [
[[package]]
name = "libp2p-swarm"
version = "0.25.0"
source = "git+https://github.com/sigp/rust-libp2p?rev=e3caf9e0e5e78c9d51c6dccf0d6277cef553bb25#e3caf9e0e5e78c9d51c6dccf0d6277cef553bb25"
source = "git+https://github.com/sigp/rust-libp2p?rev=830e6fabb7ee51281a98f5e092f056668adbef25#830e6fabb7ee51281a98f5e092f056668adbef25"
dependencies = [
"either",
"futures 0.3.8",
@ -3638,7 +3638,7 @@ dependencies = [
[[package]]
name = "libp2p-tcp"
version = "0.25.0"
source = "git+https://github.com/sigp/rust-libp2p?rev=e3caf9e0e5e78c9d51c6dccf0d6277cef553bb25#e3caf9e0e5e78c9d51c6dccf0d6277cef553bb25"
source = "git+https://github.com/sigp/rust-libp2p?rev=830e6fabb7ee51281a98f5e092f056668adbef25#830e6fabb7ee51281a98f5e092f056668adbef25"
dependencies = [
"futures 0.3.8",
"futures-timer",
@ -3653,7 +3653,7 @@ dependencies = [
[[package]]
name = "libp2p-websocket"
version = "0.26.0"
source = "git+https://github.com/sigp/rust-libp2p?rev=e3caf9e0e5e78c9d51c6dccf0d6277cef553bb25#e3caf9e0e5e78c9d51c6dccf0d6277cef553bb25"
source = "git+https://github.com/sigp/rust-libp2p?rev=830e6fabb7ee51281a98f5e092f056668adbef25#830e6fabb7ee51281a98f5e092f056668adbef25"
dependencies = [
"async-tls",
"either",
@ -3672,7 +3672,7 @@ dependencies = [
[[package]]
name = "libp2p-yamux"
version = "0.28.0"
source = "git+https://github.com/sigp/rust-libp2p?rev=e3caf9e0e5e78c9d51c6dccf0d6277cef553bb25#e3caf9e0e5e78c9d51c6dccf0d6277cef553bb25"
source = "git+https://github.com/sigp/rust-libp2p?rev=830e6fabb7ee51281a98f5e092f056668adbef25#830e6fabb7ee51281a98f5e092f056668adbef25"
dependencies = [
"futures 0.3.8",
"libp2p-core 0.25.0",
@ -4137,7 +4137,7 @@ dependencies = [
[[package]]
name = "multistream-select"
version = "0.8.5"
source = "git+https://github.com/sigp/rust-libp2p?rev=e3caf9e0e5e78c9d51c6dccf0d6277cef553bb25#e3caf9e0e5e78c9d51c6dccf0d6277cef553bb25"
source = "git+https://github.com/sigp/rust-libp2p?rev=830e6fabb7ee51281a98f5e092f056668adbef25#830e6fabb7ee51281a98f5e092f056668adbef25"
dependencies = [
"bytes 0.5.6",
"futures 0.3.8",
@ -4482,7 +4482,7 @@ dependencies = [
[[package]]
name = "parity-multiaddr"
version = "0.9.6"
source = "git+https://github.com/sigp/rust-libp2p?rev=e3caf9e0e5e78c9d51c6dccf0d6277cef553bb25#e3caf9e0e5e78c9d51c6dccf0d6277cef553bb25"
source = "git+https://github.com/sigp/rust-libp2p?rev=830e6fabb7ee51281a98f5e092f056668adbef25#830e6fabb7ee51281a98f5e092f056668adbef25"
dependencies = [
"arrayref",
"bs58 0.4.0",

2
beacon_node/eth2_libp2p/Cargo.toml
View File

@ -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"]

20
beacon_node/eth2_libp2p/src/behaviour/mod.rs
View File

@ -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

7
beacon_node/eth2_libp2p/src/discovery/mod.rs
View File

@ -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);

24
beacon_node/eth2_libp2p/src/peer_manager/mod.rs
View File

@ -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

12
beacon_node/eth2_libp2p/src/peer_manager/peer_info.rs
View File

@ -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

8
beacon_node/eth2_libp2p/src/peer_manager/peerdb.rs
View File

@ -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 {

2
beacon_node/eth2_libp2p/src/types/mod.rs
View File

@ -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};

23
beacon_node/eth2_libp2p/src/types/topics.rs
View File

@ -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))
);
}
}

15
beacon_node/network/src/service.rs
View File

@ -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);