Gossipsub update (#1400)

## Issue Addressed

N/A

## Proposed Changes

This provides a number of corrections and improvements to gossipsub. Specifically
- Enables options for greater privacy around the message author
- Provides greater flexibility on message validation
- Prevents unvalidated messages from being gossiped
- Shifts the duplicate cache to a time-based cache inside gossipsub
- Updates the message-id to handle bytes
- Bug fixes related to mesh maintenance and topic subscription. This should improve our attestation inclusion rate.

beacon_node/eth2_libp2p/Cargo.toml

@@ -41,7 +41,7 @@ rand = "0.7.3"
 [dependencies.libp2p]
 #version = "0.19.1"
 git = "https://github.com/sigp/rust-libp2p"
-rev = "147bb43fa56c1b84253606eabedb0794eeed8b94"
+rev = "8e9e35994e63716c6eb0a05b9702133d113b3822"
 default-features = false
 features = ["websocket", "identify", "mplex", "yamux", "noise", "gossipsub", "dns", "secio", "tcp-tokio"]
 

beacon_node/eth2_libp2p/src/behaviour/mod.rs

@@ -11,7 +11,7 @@ use libp2p::{
         identity::Keypair,
         Multiaddr,
     },
-    gossipsub::{Gossipsub, GossipsubEvent, MessageId},
+    gossipsub::{Gossipsub, GossipsubEvent, MessageAuthenticity, MessageId},
     identify::{Identify, IdentifyEvent},
     swarm::{
         NetworkBehaviour, NetworkBehaviourAction as NBAction, NotifyHandler, PollParameters,
@@ -19,7 +19,6 @@ use libp2p::{
     },
     PeerId,
 };
-use lru::LruCache;
 use slog::{crit, debug, o, trace};
 use std::{
     collections::VecDeque,
@@ -56,10 +55,6 @@ pub struct Behaviour<TSpec: EthSpec> {
     peers_to_dc: VecDeque<PeerId>,
     /// The current meta data of the node, so respond to pings and get metadata
     meta_data: MetaData<TSpec>,
-    /// A cache of recently seen gossip messages. This is used to filter out any possible
-    /// duplicates that may still be seen over gossipsub.
-    // TODO: Remove this
-    seen_gossip_messages: LruCache<MessageId, ()>,
     /// A collections of variables accessible outside the network service.
     network_globals: Arc<NetworkGlobals<TSpec>>,
     /// Keeps track of the current EnrForkId for upgrading gossipsub topics.
@@ -80,7 +75,6 @@ impl<TSpec: EthSpec> Behaviour<TSpec> {
         network_globals: Arc<NetworkGlobals<TSpec>>,
         log: &slog::Logger,
     ) -> error::Result<Self> {
-        let local_peer_id = local_key.public().into_peer_id();
         let behaviour_log = log.new(o!());
 
         let identify = Identify::new(
@@ -104,15 +98,20 @@ impl<TSpec: EthSpec> Behaviour<TSpec> {
             attnets,
         };
 
+        // 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");
+
         Ok(Behaviour {
             eth2_rpc: RPC::new(log.clone()),
-            gossipsub: Gossipsub::new(local_peer_id, net_conf.gs_config.clone()),
+            gossipsub: Gossipsub::new(
+                MessageAuthenticity::Author(message_author),
+                net_conf.gs_config.clone(),
+            ),
             identify,
             peer_manager: PeerManager::new(local_key, net_conf, network_globals.clone(), log)?,
             events: VecDeque::new(),
             handler_events: VecDeque::new(),
             peers_to_dc: VecDeque::new(),
-            seen_gossip_messages: LruCache::new(100_000),
             meta_data,
             network_globals,
             enr_fork_id,
@@ -215,7 +214,9 @@ 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) => {
-                        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));
+                        }
                     }
                     Err(e) => crit!(self.log, "Could not publish message"; "error" => e),
                 }
@@ -225,9 +226,9 @@ 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 propagate_message(&mut self, propagation_source: &PeerId, message_id: MessageId) {
+    pub fn validate_message(&mut self, propagation_source: &PeerId, message_id: MessageId) {
         self.gossipsub
-            .propagate_message(&message_id, propagation_source);
+            .validate_message(&message_id, propagation_source);
     }
 
     /* Eth2 RPC behaviour functions */
@@ -394,29 +395,16 @@ impl<TSpec: EthSpec> Behaviour<TSpec> {
             GossipsubEvent::Message(propagation_source, id, gs_msg) => {
                 // Note: We are keeping track here of the peer that sent us the message, not the
                 // peer that originally published the message.
-                if self.seen_gossip_messages.put(id.clone(), ()).is_none() {
-                    match PubsubMessage::decode(&gs_msg.topics, &gs_msg.data) {
-                        Err(e) => {
-                            debug!(self.log, "Could not decode gossipsub message"; "error" => e)
-                        }
-                        Ok(msg) => {
-                            // if this message isn't a duplicate, notify the network
-                            self.add_event(BehaviourEvent::PubsubMessage {
-                                id,
-                                source: propagation_source,
-                                topics: gs_msg.topics,
-                                message: msg,
-                            });
-                        }
-                    }
-                } else {
-                    match PubsubMessage::<TSpec>::decode(&gs_msg.topics, &gs_msg.data) {
-                        Err(e) => {
-                            debug!(self.log, "Could not decode gossipsub message"; "error" => e)
-                        }
-                        Ok(msg) => {
-                            debug!(self.log, "A duplicate gossipsub message was received"; "message_source" => format!("{}", gs_msg.source), "propagated_peer" => format!("{}",propagation_source), "message" => format!("{}", msg));
-                        }
+                match PubsubMessage::decode(&gs_msg.topics, &gs_msg.data) {
+                    Err(e) => debug!(self.log, "Could not decode gossipsub message"; "error" => e),
+                    Ok(msg) => {
+                        // Notify the network
+                        self.add_event(BehaviourEvent::PubsubMessage {
+                            id,
+                            source: propagation_source,
+                            topics: gs_msg.topics,
+                            message: msg,
+                        });
                     }
                 }
             }

beacon_node/eth2_libp2p/src/config.rs

@@ -1,7 +1,9 @@
 use crate::types::GossipKind;
 use crate::Enr;
 use discv5::{Discv5Config, Discv5ConfigBuilder};
-use libp2p::gossipsub::{GossipsubConfig, GossipsubConfigBuilder, GossipsubMessage, MessageId};
+use libp2p::gossipsub::{
+    GossipsubConfig, GossipsubConfigBuilder, GossipsubMessage, MessageId, ValidationMode,
+};
 use libp2p::Multiaddr;
 use serde_derive::{Deserialize, Serialize};
 use sha2::{Digest, Sha256};
@@ -82,7 +84,7 @@ impl Default for Config {
         // The function used to generate a gossipsub message id
         // We use base64(SHA256(data)) for content addressing
         let gossip_message_id = |message: &GossipsubMessage| {
-            MessageId(base64::encode_config(
+            MessageId::from(base64::encode_config(
                 &Sha256::digest(&message.data),
                 base64::URL_SAFE_NO_PAD,
             ))
@@ -94,8 +96,10 @@ impl Default for Config {
         let gs_config = GossipsubConfigBuilder::new()
             .max_transmit_size(GOSSIP_MAX_SIZE)
             .heartbeat_interval(Duration::from_secs(1))
-            .manual_propagation() // require validation before propagation
-            .no_source_id()
+            .validate_messages() // require validation before propagation
+            .validation_mode(ValidationMode::Permissive)
+            // Prevent duplicates by caching messages from an epoch + 1 slot amount of time (33*12)
+            .duplicate_cache_time(Duration::from_secs(396))
             .message_id_fn(gossip_message_id)
             .build();