Use Gossipsub 1.1 (#1516)

## Issue Addressed

#1172

## Proposed Changes

* updates the libp2p dependency
* small adaptions based on changes in libp2p
* report not just valid messages but also invalid and distinguish between `IGNORE`d messages and `REJECT`ed messages


Co-authored-by: Age Manning <Age@AgeManning.com>

beacon_node/eth2_libp2p/Cargo.toml

@@ -41,7 +41,7 @@ rand = "0.7.3"
 [dependencies.libp2p]
 #version = "0.23.0"
 git = "https://github.com/sigp/rust-libp2p"
-rev = "bbf0cfbaff2f733b3ae7bfed3caba8b7ee542803"
+rev = "d0f9d6b9b3fef9616026f3ddf11d75fe9f7a41df"
 default-features = false
 features = ["websocket", "identify", "mplex", "noise", "gossipsub", "dns", "tcp-tokio"]
 

beacon_node/eth2_libp2p/src/behaviour/handler/delegate.rs

@@ -131,8 +131,9 @@ impl<TSpec: EthSpec> ProtocolsHandler for DelegatingHandler<TSpec> {
     type InboundProtocol = DelegateInProto<TSpec>;
     type OutboundProtocol = DelegateOutProto<TSpec>;
     type OutboundOpenInfo = DelegateOutInfo<TSpec>;
+    type InboundOpenInfo = ();
 
-    fn listen_protocol(&self) -> SubstreamProtocol<Self::InboundProtocol> {
+    fn listen_protocol(&self) -> SubstreamProtocol<Self::InboundProtocol, ()> {
         let gossip_proto = self.gossip_handler.listen_protocol();
         let rpc_proto = self.rpc_handler.listen_protocol();
         let identify_proto = self.identify_handler.listen_protocol();
@@ -147,24 +148,27 @@ impl<TSpec: EthSpec> ProtocolsHandler for DelegatingHandler<TSpec> {
             SelectUpgrade::new(rpc_proto.into_upgrade().1, identify_proto.into_upgrade().1),
         );
 
-        SubstreamProtocol::new(select).with_timeout(timeout)
+        SubstreamProtocol::new(select, ()).with_timeout(timeout)
     }
 
     fn inject_fully_negotiated_inbound(
         &mut self,
         out: <Self::InboundProtocol as InboundUpgrade<NegotiatedSubstream>>::Output,
+        _info: Self::InboundOpenInfo,
     ) {
         match out {
             // Gossipsub
-            EitherOutput::First(out) => self.gossip_handler.inject_fully_negotiated_inbound(out),
+            EitherOutput::First(out) => {
+                self.gossip_handler.inject_fully_negotiated_inbound(out, ())
+            }
             // RPC
             EitherOutput::Second(EitherOutput::First(out)) => {
-                self.rpc_handler.inject_fully_negotiated_inbound(out)
+                self.rpc_handler.inject_fully_negotiated_inbound(out, ())
             }
             // Identify
-            EitherOutput::Second(EitherOutput::Second(out)) => {
-                self.identify_handler.inject_fully_negotiated_inbound(out)
-            }
+            EitherOutput::Second(EitherOutput::Second(out)) => self
+                .identify_handler
+                .inject_fully_negotiated_inbound(out, ()),
         }
     }
 
@@ -317,10 +321,11 @@ impl<TSpec: EthSpec> ProtocolsHandler for DelegatingHandler<TSpec> {
                     event,
                 )));
             }
-            Poll::Ready(ProtocolsHandlerEvent::OutboundSubstreamRequest { protocol, info }) => {
+            Poll::Ready(ProtocolsHandlerEvent::OutboundSubstreamRequest { protocol }) => {
                 return Poll::Ready(ProtocolsHandlerEvent::OutboundSubstreamRequest {
-                    protocol: protocol.map_upgrade(EitherUpgrade::A),
-                    info: EitherOutput::First(info),
+                    protocol: protocol
+                        .map_upgrade(EitherUpgrade::A)
+                        .map_info(EitherOutput::First),
                 });
             }
             Poll::Pending => (),
@@ -333,10 +338,11 @@ impl<TSpec: EthSpec> ProtocolsHandler for DelegatingHandler<TSpec> {
             Poll::Ready(ProtocolsHandlerEvent::Close(event)) => {
                 return Poll::Ready(ProtocolsHandlerEvent::Close(DelegateError::RPC(event)));
             }
-            Poll::Ready(ProtocolsHandlerEvent::OutboundSubstreamRequest { protocol, info }) => {
+            Poll::Ready(ProtocolsHandlerEvent::OutboundSubstreamRequest { protocol }) => {
                 return Poll::Ready(ProtocolsHandlerEvent::OutboundSubstreamRequest {
-                    protocol: protocol.map_upgrade(|u| EitherUpgrade::B(EitherUpgrade::A(u))),
-                    info: EitherOutput::Second(EitherOutput::First(info)),
+                    protocol: protocol
+                        .map_upgrade(|u| EitherUpgrade::B(EitherUpgrade::A(u)))
+                        .map_info(|info| EitherOutput::Second(EitherOutput::First(info))),
                 });
             }
             Poll::Pending => (),
@@ -351,10 +357,11 @@ impl<TSpec: EthSpec> ProtocolsHandler for DelegatingHandler<TSpec> {
             Poll::Ready(ProtocolsHandlerEvent::Close(event)) => {
                 return Poll::Ready(ProtocolsHandlerEvent::Close(DelegateError::Identify(event)));
             }
-            Poll::Ready(ProtocolsHandlerEvent::OutboundSubstreamRequest { protocol, info: () }) => {
+            Poll::Ready(ProtocolsHandlerEvent::OutboundSubstreamRequest { protocol }) => {
                 return Poll::Ready(ProtocolsHandlerEvent::OutboundSubstreamRequest {
-                    protocol: protocol.map_upgrade(|u| EitherUpgrade::B(EitherUpgrade::B(u))),
-                    info: EitherOutput::Second(EitherOutput::Second(())),
+                    protocol: protocol
+                        .map_upgrade(|u| EitherUpgrade::B(EitherUpgrade::B(u)))
+                        .map_info(|_| EitherOutput::Second(EitherOutput::Second(()))),
                 });
             }
             Poll::Pending => (),

beacon_node/eth2_libp2p/src/behaviour/handler/mod.rs

@@ -54,16 +54,18 @@ impl<TSpec: EthSpec> ProtocolsHandler for BehaviourHandler<TSpec> {
     type InboundProtocol = DelegateInProto<TSpec>;
     type OutboundProtocol = DelegateOutProto<TSpec>;
     type OutboundOpenInfo = DelegateOutInfo<TSpec>;
+    type InboundOpenInfo = ();
 
-    fn listen_protocol(&self) -> SubstreamProtocol<Self::InboundProtocol> {
+    fn listen_protocol(&self) -> SubstreamProtocol<Self::InboundProtocol, ()> {
         self.delegate.listen_protocol()
     }
 
     fn inject_fully_negotiated_inbound(
         &mut self,
         out: <Self::InboundProtocol as InboundUpgrade<NegotiatedSubstream>>::Output,
+        _info: Self::InboundOpenInfo,
     ) {
-        self.delegate.inject_fully_negotiated_inbound(out)
+        self.delegate.inject_fully_negotiated_inbound(out, ())
     }
 
     fn inject_fully_negotiated_outbound(
@@ -127,11 +129,8 @@ impl<TSpec: EthSpec> ProtocolsHandler for BehaviourHandler<TSpec> {
             Poll::Ready(ProtocolsHandlerEvent::Close(err)) => {
                 return Poll::Ready(ProtocolsHandlerEvent::Close(err))
             }
-            Poll::Ready(ProtocolsHandlerEvent::OutboundSubstreamRequest { protocol, info }) => {
-                return Poll::Ready(ProtocolsHandlerEvent::OutboundSubstreamRequest {
-                    protocol,
-                    info,
-                });
+            Poll::Ready(ProtocolsHandlerEvent::OutboundSubstreamRequest { protocol }) => {
+                return Poll::Ready(ProtocolsHandlerEvent::OutboundSubstreamRequest { protocol });
             }
             Poll::Pending => (),
         }

beacon_node/eth2_libp2p/src/behaviour/mod.rs

@@ -11,7 +11,10 @@ use libp2p::{
         identity::Keypair,
         Multiaddr,
     },
-    gossipsub::{Gossipsub, GossipsubEvent, MessageAuthenticity, MessageId},
+    gossipsub::{
+        Gossipsub, GossipsubEvent, IdentTopic as Topic, MessageAcceptance, MessageAuthenticity,
+        MessageId,
+    },
     identify::{Identify, IdentifyEvent},
     swarm::{
         NetworkBehaviour, NetworkBehaviourAction as NBAction, NotifyHandler, PollParameters,
@@ -94,15 +97,19 @@ impl<TSpec: EthSpec> Behaviour<TSpec> {
 
         let meta_data = load_or_build_metadata(&net_conf.network_dir, &log);
 
-        // TODO: Until other clients support no author, we will use a 0 peer_id as our author.
-        let message_author = PeerId::from_bytes(vec![0, 1, 0]).expect("Valid peer id");
+        let gossipsub = Gossipsub::new(MessageAuthenticity::Anonymous, net_conf.gs_config.clone())
+            .map_err(|e| format!("Could not construct gossipsub: {:?}", e))?;
+
+        // Temporarily disable scoring until parameters are tested.
+        /*
+        gossipsub
+            .with_peer_score(PeerScoreParams::default(), PeerScoreThresholds::default())
+            .expect("Valid score params and thresholds");
+        */
 
         Ok(Behaviour {
             eth2_rpc: RPC::new(log.clone()),
-            gossipsub: Gossipsub::new(
-                MessageAuthenticity::Author(message_author),
-                net_conf.gs_config.clone(),
-            ),
+            gossipsub,
             identify,
             peer_manager: PeerManager::new(local_key, net_conf, network_globals.clone(), log)
                 .await?,
@@ -147,6 +154,10 @@ impl<TSpec: EthSpec> Behaviour<TSpec> {
             GossipEncoding::default(),
             self.enr_fork_id.fork_digest,
         );
+
+        // TODO: Implement scoring
+        // let topic: Topic = gossip_topic.into();
+        // self.gossipsub.set_topic_params(t.hash(), TopicScoreParams::default());
         self.subscribe(gossip_topic)
     }
 
@@ -168,6 +179,12 @@ impl<TSpec: EthSpec> Behaviour<TSpec> {
             GossipEncoding::default(),
             self.enr_fork_id.fork_digest,
         );
+        // TODO: Implement scoring
+        /*
+        let t: Topic = topic.clone().into();
+        self.gossipsub
+            .set_topic_params(t.hash(), TopicScoreParams::default());
+        */
         self.subscribe(topic)
     }
 
@@ -189,9 +206,18 @@ impl<TSpec: EthSpec> Behaviour<TSpec> {
             .write()
             .insert(topic.clone());
 
-        let topic_str: String = topic.clone().into();
-        debug!(self.log, "Subscribed to topic"; "topic" => topic_str);
-        self.gossipsub.subscribe(topic.into())
+        let topic: Topic = topic.into();
+
+        match self.gossipsub.subscribe(&topic) {
+            Err(_) => {
+                warn!(self.log, "Failed to subscribe to topic"; "topic" => topic.to_string());
+                false
+            }
+            Ok(v) => {
+                debug!(self.log, "Subscribed to topic"; "topic" => topic.to_string());
+                v
+            }
+        }
     }
 
     /// Unsubscribe from a gossipsub topic.
@@ -201,8 +227,20 @@ impl<TSpec: EthSpec> Behaviour<TSpec> {
             .gossipsub_subscriptions
             .write()
             .remove(&topic);
+
         // unsubscribe from the topic
-        self.gossipsub.unsubscribe(topic.into())
+        let topic: Topic = topic.into();
+
+        match self.gossipsub.unsubscribe(&topic) {
+            Err(_) => {
+                warn!(self.log, "Failed to unsubscribe from topic"; "topic" => topic.to_string());
+                false
+            }
+            Ok(v) => {
+                debug!(self.log, "Unsubscribed to topic"; "topic" => topic.to_string());
+                v
+            }
+        }
     }
 
     /// Publishes a list of messages on the pubsub (gossipsub) behaviour, choosing the encoding.
@@ -211,7 +249,7 @@ impl<TSpec: EthSpec> Behaviour<TSpec> {
             for topic in message.topics(GossipEncoding::default(), self.enr_fork_id.fork_digest) {
                 match message.encode(GossipEncoding::default()) {
                     Ok(message_data) => {
-                        if let Err(e) = self.gossipsub.publish(&topic.into(), message_data) {
+                        if let Err(e) = self.gossipsub.publish(topic.into(), message_data) {
                             slog::warn!(self.log, "Could not publish message"; "error" => format!("{:?}", e));
                         }
                     }
@@ -221,11 +259,21 @@ impl<TSpec: EthSpec> Behaviour<TSpec> {
         }
     }
 
-    /// Forwards a message that is waiting in gossipsub's mcache. Messages are only propagated
-    /// once validated by the beacon chain.
-    pub fn validate_message(&mut self, propagation_source: &PeerId, message_id: MessageId) {
-        self.gossipsub
-            .validate_message(&message_id, propagation_source);
+    /// Informs the gossipsub about the result of a message validation.
+    /// If the message is valid it will get propagated by gossipsub.
+    pub fn report_message_validation_result(
+        &mut self,
+        propagation_source: &PeerId,
+        message_id: MessageId,
+        validation_result: MessageAcceptance,
+    ) {
+        if let Err(e) = self.gossipsub.report_message_validation_result(
+            &message_id,
+            propagation_source,
+            validation_result,
+        ) {
+            warn!(self.log, "Failed to report message validation"; "message_id" => message_id.to_string(), "peer_id" => propagation_source.to_string(), "error" => format!("{:?}", e));
+        }
     }
 
     /* Eth2 RPC behaviour functions */
@@ -392,11 +440,25 @@ impl<TSpec: EthSpec> Behaviour<TSpec> {
 
     fn on_gossip_event(&mut self, event: GossipsubEvent) {
         match event {
-            GossipsubEvent::Message(propagation_source, id, gs_msg) => {
+            GossipsubEvent::Message {
+                propagation_source,
+                message_id: id,
+                message: gs_msg,
+            } => {
                 // Note: We are keeping track here of the peer that sent us the message, not the
                 // peer that originally published the message.
                 match PubsubMessage::decode(&gs_msg.topics, &gs_msg.data) {
-                    Err(e) => debug!(self.log, "Could not decode gossipsub message"; "error" => e),
+                    Err(e) => {
+                        debug!(self.log, "Could not decode gossipsub message"; "error" => e);
+                        //reject the message
+                        if let Err(e) = self.gossipsub.report_message_validation_result(
+                            &id,
+                            &propagation_source,
+                            MessageAcceptance::Reject,
+                        ) {
+                            warn!(self.log, "Failed to report message validation"; "message_id" => id.to_string(), "peer_id" => propagation_source.to_string(), "error" => format!("{:?}", e));
+                        }
+                    }
                     Ok(msg) => {
                         // Notify the network
                         self.add_event(BehaviourEvent::PubsubMessage {

beacon_node/eth2_libp2p/src/config.rs

@@ -99,8 +99,8 @@ impl Default for Config {
         let gs_config = GossipsubConfigBuilder::new()
             .max_transmit_size(GOSSIP_MAX_SIZE)
             .heartbeat_interval(Duration::from_millis(700))
-            .mesh_n(6)
-            .mesh_n_low(5)
+            .mesh_n(8)
+            .mesh_n_low(6)
             .mesh_n_high(12)
             .gossip_lazy(6)
             .fanout_ttl(Duration::from_secs(60))
@@ -111,7 +111,8 @@ impl Default for Config {
             // prevent duplicates for 550 heartbeats(700millis * 550) = 385 secs
             .duplicate_cache_time(Duration::from_secs(385))
             .message_id_fn(gossip_message_id)
-            .build();
+            .build()
+            .expect("valid gossipsub configuration");
 
         // discv5 configuration
         let discv5_config = Discv5ConfigBuilder::new()

beacon_node/eth2_libp2p/src/lib.rs

@@ -19,7 +19,7 @@ pub use behaviour::{BehaviourEvent, PeerRequestId, Request, Response};
 pub use config::Config as NetworkConfig;
 pub use discovery::{CombinedKeyExt, EnrExt, Eth2Enr};
 pub use discv5;
-pub use libp2p::gossipsub::{MessageId, Topic, TopicHash};
+pub use libp2p::gossipsub::{MessageAcceptance, MessageId, Topic, TopicHash};
 pub use libp2p::{core::ConnectedPoint, PeerId, Swarm};
 pub use libp2p::{multiaddr, Multiaddr};
 pub use metrics::scrape_discovery_metrics;

beacon_node/eth2_libp2p/src/rpc/handler.rs

@@ -81,7 +81,7 @@ where
     TSpec: EthSpec,
 {
     /// The upgrade for inbound substreams.
-    listen_protocol: SubstreamProtocol<RPCProtocol<TSpec>>,
+    listen_protocol: SubstreamProtocol<RPCProtocol<TSpec>, ()>,
 
     /// Errors occurring on outbound and inbound connections queued for reporting back.
     pending_errors: Vec<HandlerErr>,
@@ -225,7 +225,10 @@ impl<TSpec> RPCHandler<TSpec>
 where
     TSpec: EthSpec,
 {
-    pub fn new(listen_protocol: SubstreamProtocol<RPCProtocol<TSpec>>, log: &slog::Logger) -> Self {
+    pub fn new(
+        listen_protocol: SubstreamProtocol<RPCProtocol<TSpec>, ()>,
+        log: &slog::Logger,
+    ) -> Self {
         RPCHandler {
             listen_protocol,
             pending_errors: Vec::new(),
@@ -249,7 +252,7 @@ where
     ///
     /// > **Note**: If you modify the protocol, modifications will only applies to future inbound
     /// >           substreams, not the ones already being negotiated.
-    pub fn listen_protocol_ref(&self) -> &SubstreamProtocol<RPCProtocol<TSpec>> {
+    pub fn listen_protocol_ref(&self) -> &SubstreamProtocol<RPCProtocol<TSpec>, ()> {
         &self.listen_protocol
     }
 
@@ -257,7 +260,7 @@ where
     ///
     /// > **Note**: If you modify the protocol, modifications will only apply to future inbound
     /// >           substreams, not the ones already being negotiated.
-    pub fn listen_protocol_mut(&mut self) -> &mut SubstreamProtocol<RPCProtocol<TSpec>> {
+    pub fn listen_protocol_mut(&mut self) -> &mut SubstreamProtocol<RPCProtocol<TSpec>, ()> {
         &mut self.listen_protocol
     }
 
@@ -344,14 +347,16 @@ where
     type InboundProtocol = RPCProtocol<TSpec>;
     type OutboundProtocol = RPCRequest<TSpec>;
     type OutboundOpenInfo = (RequestId, RPCRequest<TSpec>); // Keep track of the id and the request
+    type InboundOpenInfo = ();
 
-    fn listen_protocol(&self) -> SubstreamProtocol<Self::InboundProtocol> {
+    fn listen_protocol(&self) -> SubstreamProtocol<Self::InboundProtocol, ()> {
         self.listen_protocol.clone()
     }
 
     fn inject_fully_negotiated_inbound(
         &mut self,
         substream: <Self::InboundProtocol as InboundUpgrade<NegotiatedSubstream>>::Output,
+        _info: Self::InboundOpenInfo,
     ) {
         // only accept new peer requests when active
         if !matches!(self.state, HandlerState::Active) {
@@ -863,8 +868,7 @@ where
             let (id, req) = self.dial_queue.remove(0);
             self.dial_queue.shrink_to_fit();
             return Poll::Ready(ProtocolsHandlerEvent::OutboundSubstreamRequest {
-                protocol: SubstreamProtocol::new(req.clone()),
-                info: (id, req),
+                protocol: SubstreamProtocol::new(req.clone(), ()).map_info(|()| (id, req)),
             });
         }
         Poll::Pending

beacon_node/eth2_libp2p/src/rpc/mod.rs

@@ -169,9 +169,12 @@ where
 
     fn new_handler(&mut self) -> Self::ProtocolsHandler {
         RPCHandler::new(
-            SubstreamProtocol::new(RPCProtocol {
-                phantom: PhantomData,
-            }),
+            SubstreamProtocol::new(
+                RPCProtocol {
+                    phantom: PhantomData,
+                },
+                (),
+            ),
             &self.log,
         )
     }

beacon_node/eth2_libp2p/src/types/topics.rs

@@ -1,4 +1,4 @@
-use libp2p::gossipsub::Topic;
+use libp2p::gossipsub::IdentTopic as Topic;
 use serde_derive::{Deserialize, Serialize};
 use types::SubnetId;
 
@@ -139,7 +139,7 @@ impl GossipTopic {
 
 impl Into<Topic> for GossipTopic {
     fn into(self) -> Topic {
-        Topic::new(self.into())
+        Topic::new(self)
     }
 }
 

beacon_node/eth2_libp2p/tests/common/mod.rs

@@ -10,6 +10,7 @@ use std::time::Duration;
 use types::{EnrForkId, MinimalEthSpec};
 
 type E = MinimalEthSpec;
+use libp2p::gossipsub::GossipsubConfigBuilder;
 use tempdir::TempDir;
 
 pub struct Libp2pInstance(LibP2PService<E>, exit_future::Signal);
@@ -83,8 +84,11 @@ pub fn build_config(port: u16, mut boot_nodes: Vec<Enr>) -> NetworkConfig {
     config.boot_nodes_enr.append(&mut boot_nodes);
     config.network_dir = path.into_path();
     // Reduce gossipsub heartbeat parameters
-    config.gs_config.heartbeat_initial_delay = Duration::from_millis(500);
-    config.gs_config.heartbeat_interval = Duration::from_millis(500);
+    config.gs_config = GossipsubConfigBuilder::from(config.gs_config)
+        .heartbeat_initial_delay(Duration::from_millis(500))
+        .heartbeat_interval(Duration::from_millis(500))
+        .build()
+        .unwrap();
     config
 }