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 <[email protected]>
sigp · Aug 30, 2020 · c18d37c · c18d37c
1 parent b6340ec
commit c18d37c
Show file tree

Hide file tree

Showing 15 changed files with 332 additions and 142 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/beacon_node/eth2_libp2p/Cargo.toml b/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"]
 

diff --git a/beacon_node/eth2_libp2p/src/behaviour/handler/delegate.rs b/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 => (),

diff --git a/beacon_node/eth2_libp2p/src/behaviour/handler/mod.rs b/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 => (),
         }