Filter gossipsub message duplication (#736)

* Add duplication prevention to gossipsub

* Clean up topic logs

* Add content addressed messages for gossip

beacon_node/network/src/lib.rs

@@ -1,9 +1,11 @@
 /// This crate provides the network server for Lighthouse.
 pub mod error;
 pub mod message_handler;
+pub mod message_processor;
 pub mod service;
 pub mod sync;
 
 pub use eth2_libp2p::NetworkConfig;
+pub use message_processor::MessageProcessor;
 pub use service::NetworkMessage;
 pub use service::Service;

beacon_node/network/src/message_handler.rs

@@ -1,12 +1,12 @@
 #![allow(clippy::unit_arg)]
 use crate::error;
 use crate::service::NetworkMessage;
-use crate::sync::MessageProcessor;
+use crate::MessageProcessor;
 use beacon_chain::{BeaconChain, BeaconChainTypes};
 use eth2_libp2p::{
     behaviour::PubsubMessage,
     rpc::{RPCError, RPCErrorResponse, RPCRequest, RPCResponse, RequestId, ResponseTermination},
-    PeerId, RPCEvent,
+    MessageId, PeerId, RPCEvent,
 };
 use futures::future::Future;
 use futures::stream::Stream;
@@ -41,7 +41,7 @@ pub enum HandlerMessage {
     RPC(PeerId, RPCEvent),
     /// A gossip message has been received. The fields are: message id, the peer that sent us this
     /// message and the message itself.
-    PubsubMessage(String, PeerId, PubsubMessage),
+    PubsubMessage(MessageId, PeerId, PubsubMessage),
 }
 
 impl<T: BeaconChainTypes> MessageHandler<T> {
@@ -220,7 +220,7 @@ impl<T: BeaconChainTypes> MessageHandler<T> {
     }
 
     /// Handle RPC messages
-    fn handle_gossip(&mut self, id: String, peer_id: PeerId, gossip_message: PubsubMessage) {
+    fn handle_gossip(&mut self, id: MessageId, peer_id: PeerId, gossip_message: PubsubMessage) {
         match gossip_message {
             PubsubMessage::Block(message) => match self.decode_gossip_block(message) {
                 Ok(block) => {
@@ -292,7 +292,7 @@ impl<T: BeaconChainTypes> MessageHandler<T> {
     }
 
     /// Informs the network service that the message should be forwarded to other peers.
-    fn propagate_message(&mut self, message_id: String, propagation_source: PeerId) {
+    fn propagate_message(&mut self, message_id: MessageId, propagation_source: PeerId) {
         self.network_send
             .try_send(NetworkMessage::Propagate {
                 propagation_source,

beacon_node/network/src/message_processor.rs

@@ -1,5 +1,5 @@
-use super::manager::SyncMessage;
 use crate::service::NetworkMessage;
+use crate::sync::SyncMessage;
 use beacon_chain::{
     AttestationProcessingOutcome, BeaconChain, BeaconChainTypes, BlockProcessingOutcome,
 };
@@ -77,7 +77,7 @@ impl<T: BeaconChainTypes> MessageProcessor<T> {
         let sync_logger = log.new(o!("service"=> "sync"));
 
         // spawn the sync thread
-        let (sync_send, _sync_exit) = super::manager::spawn(
+        let (sync_send, _sync_exit) = crate::sync::manager::spawn(
             executor,
             Arc::downgrade(&beacon_chain),
             network_send.clone(),
@@ -503,7 +503,9 @@ impl<T: BeaconChainTypes> MessageProcessor<T> {
                         self.log,
                         "Processed attestation";
                         "source" => "gossip",
-                        "outcome" => format!("{:?}", outcome)
+                        "outcome" => format!("{:?}", outcome),
+                        "peer" => format!("{:?}",peer_id),
+                        "data" => format!("{:?}", msg.data)
                     );
                 }
                 AttestationProcessingOutcome::UnknownHeadBlock { beacon_block_root } => {

beacon_node/network/src/service.rs

@@ -4,7 +4,7 @@ use crate::NetworkConfig;
 use beacon_chain::{BeaconChain, BeaconChainTypes};
 use core::marker::PhantomData;
 use eth2_libp2p::Service as LibP2PService;
-use eth2_libp2p::{rpc::RPCRequest, Enr, Libp2pEvent, Multiaddr, PeerId, Swarm, Topic};
+use eth2_libp2p::{rpc::RPCRequest, Enr, Libp2pEvent, MessageId, Multiaddr, PeerId, Swarm, Topic};
 use eth2_libp2p::{PubsubMessage, RPCEvent};
 use futures::prelude::*;
 use futures::Stream;
@@ -263,7 +263,7 @@ fn network_service(
                         id,
                         source,
                         message,
-                        ..
+                        topics: _,
                     } => {
                         message_handler_send
                             .try_send(HandlerMessage::PubsubMessage(id, source, message))
@@ -302,7 +302,7 @@ pub enum NetworkMessage {
     /// Propagate a received gossipsub message.
     Propagate {
         propagation_source: PeerId,
-        message_id: String,
+        message_id: MessageId,
     },
     /// Disconnect and bans a peer id.
     Disconnect { peer_id: PeerId },

beacon_node/network/src/sync/manager.rs

@@ -58,9 +58,9 @@
 //! if an attestation references an unknown block) this manager can search for the block and
 //! subsequently search for parents if needed.
 
-use super::message_processor::PeerSyncInfo;
 use super::network_context::SyncNetworkContext;
 use super::range_sync::RangeSync;
+use crate::message_processor::PeerSyncInfo;
 use crate::service::NetworkMessage;
 use beacon_chain::{BeaconChain, BeaconChainTypes, BlockProcessingOutcome};
 use eth2_libp2p::rpc::methods::*;

beacon_node/network/src/sync/mod.rs

@@ -1,14 +1,13 @@
 //! Syncing for lighthouse.
 //!
 //! Stores the various syncing methods for the beacon chain.
-mod manager;
-mod message_processor;
+pub mod manager;
 mod network_context;
 mod range_sync;
 
-pub use message_processor::MessageProcessor;
-
 /// Currently implemented sync methods.
 pub enum SyncMethod {
     SimpleSync,
 }
+
+pub use manager::SyncMessage;

beacon_node/network/src/sync/network_context.rs

@@ -1,7 +1,7 @@
 //! Provides network functionality for the Syncing thread. This fundamentally wraps a network
 //! channel and stores a global RPC ID to perform requests.
 
-use super::message_processor::status_message;
+use crate::message_processor::status_message;
 use crate::service::NetworkMessage;
 use beacon_chain::{BeaconChain, BeaconChainTypes};
 use eth2_libp2p::rpc::methods::*;

beacon_node/network/src/sync/range_sync/chain.rs

@@ -1,4 +1,4 @@
-use crate::sync::message_processor::FUTURE_SLOT_TOLERANCE;
+use crate::message_processor::FUTURE_SLOT_TOLERANCE;
 use crate::sync::network_context::SyncNetworkContext;
 use beacon_chain::{BeaconChain, BeaconChainTypes, BlockProcessingOutcome};
 use eth2_libp2p::rpc::methods::*;

beacon_node/network/src/sync/range_sync/chain_collection.rs

@@ -1,5 +1,5 @@
 use super::chain::{ChainSyncingState, ProcessingResult, SyncingChain};
-use crate::sync::message_processor::PeerSyncInfo;
+use crate::message_processor::PeerSyncInfo;
 use crate::sync::network_context::SyncNetworkContext;
 use beacon_chain::{BeaconChain, BeaconChainTypes};
 use eth2_libp2p::PeerId;

beacon_node/network/src/sync/range_sync/range.rs

@@ -1,6 +1,6 @@
 use super::chain::ProcessingResult;
 use super::chain_collection::{ChainCollection, SyncState};
-use crate::sync::message_processor::PeerSyncInfo;
+use crate::message_processor::PeerSyncInfo;
 use crate::sync::network_context::SyncNetworkContext;
 use beacon_chain::{BeaconChain, BeaconChainTypes};
 use eth2_libp2p::rpc::RequestId;