implement explicit peers

examples/gossipsub-chat.rs

@@ -50,7 +50,7 @@ use async_std::{io, task};
 use env_logger::{Builder, Env};
 use futures::prelude::*;
 use libp2p::gossipsub::protocol::MessageId;
-use libp2p::gossipsub::{GossipsubEvent, GossipsubMessage, Topic};
+use libp2p::gossipsub::{GossipsubEvent, GossipsubMessage, IdentTopic};
 use libp2p::{gossipsub, identity, PeerId};
 use std::collections::hash_map::DefaultHasher;
 use std::hash::{Hash, Hasher};
@@ -59,6 +59,7 @@ use std::{
     error::Error,
     task::{Context, Poll},
 };
+use libp2p_core::Multiaddr;
 
 fn main() -> Result<(), Box<dyn Error>> {
     Builder::from_env(Env::default().default_filter_or("info")).init();
@@ -72,7 +73,7 @@ fn main() -> Result<(), Box<dyn Error>> {
     let transport = libp2p::build_development_transport(local_key.clone())?;
 
     // Create a Gossipsub topic
-    let topic = Topic::new("test-net".into());
+    let topic = IdentTopic::new("test-net");
 
     // Create a Swarm to manage peers and events
     let mut swarm = {
@@ -95,6 +96,13 @@ fn main() -> Result<(), Box<dyn Error>> {
         // build a gossipsub network behaviour
         let mut gossipsub = gossipsub::Gossipsub::new(local_key, gossipsub_config);
         gossipsub.subscribe(topic.clone());
+        if let Some(explicit) = std::env::args().nth(2) {
+            let explicit = explicit.clone();
+            match explicit.parse() {
+                Ok(id) => gossipsub.add_explicit_peer(&id),
+                Err(err) => println!("Failed to parse explicit peer id: {:?}", err),
+            }
+        }
         libp2p::Swarm::new(transport, gossipsub, local_peer_id)
     };
 
@@ -121,7 +129,7 @@ fn main() -> Result<(), Box<dyn Error>> {
     task::block_on(future::poll_fn(move |cx: &mut Context| {
         loop {
             match stdin.try_poll_next_unpin(cx)? {
-                Poll::Ready(Some(line)) => swarm.publish(&topic, line.as_bytes()),
+                Poll::Ready(Some(line)) => swarm.publish(topic.clone(), line.as_bytes()),
                 Poll::Ready(None) => panic!("Stdin closed"),
                 Poll::Pending => break,
             };

examples/ipfs-private.rs

@@ -168,7 +168,7 @@ fn main() -> Result<(), Box<dyn Error>> {
     let transport = build_transport(local_key.clone(), psk);
 
     // Create a Gosspipsub topic
-    let gossipsub_topic = gossipsub::Topic::new("chat".into());
+    let gossipsub_topic = gossipsub::IdentTopic::new("chat");
 
     // We create a custom network behaviour that combines gossipsub, ping and identify.
     #[derive(NetworkBehaviour)]
@@ -276,7 +276,7 @@ fn main() -> Result<(), Box<dyn Error>> {
         loop {
             match stdin.try_poll_next_unpin(cx)? {
                 Poll::Ready(Some(line)) => {
-                    swarm.gossipsub.publish(&gossipsub_topic, line.as_bytes());
+                    swarm.gossipsub.publish(gossipsub_topic.clone(), line.as_bytes());
                 }
                 Poll::Ready(None) => panic!("Stdin closed"),
                 Poll::Pending => break,

protocols/gossipsub/Cargo.toml

@@ -31,6 +31,7 @@ lru_time_cache = "0.10.0"
 [dev-dependencies]
 async-std = "1.6.2"
 env_logger = "0.7.1"
+futures-test = "0.3.5"
 libp2p-plaintext = { path = "../plaintext" }
 libp2p-yamux = { path = "../../muxers/yamux" }
 quickcheck = "0.9.2"

protocols/gossipsub/src/behaviour.rs

@@ -18,22 +18,6 @@
 // FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
 // DEALINGS IN THE SOFTWARE.
 
-use crate::config::GossipsubConfig;
-use crate::handler::GossipsubHandler;
-use crate::mcache::MessageCache;
-use crate::protocol::{
-    GossipsubControlAction, GossipsubMessage, GossipsubSubscription, GossipsubSubscriptionAction,
-    MessageId,
-};
-use crate::topic::{Hasher, Topic, TopicHash};
-use futures::prelude::*;
-use libp2p_core::{connection::ConnectionId, identity::Keypair, Multiaddr, PeerId};
-use libp2p_swarm::{
-    NetworkBehaviour, NetworkBehaviourAction, NotifyHandler, PollParameters, ProtocolsHandler,
-};
-use log::{debug, error, info, trace, warn};
-use rand;
-use rand::{seq::SliceRandom, thread_rng};
 use std::{
     collections::hash_map::HashMap,
     collections::HashSet,
@@ -42,8 +26,26 @@ use std::{
     sync::Arc,
     task::{Context, Poll},
 };
+use std::iter::FromIterator;
+
+use futures::prelude::*;
+use log::{debug, error, info, trace, warn};
+use rand;
+use rand::{seq::SliceRandom, thread_rng};
 use wasm_timer::{Instant, Interval};
 
+use libp2p_core::{connection::ConnectionId, identity::Keypair, Multiaddr, PeerId};
+use libp2p_swarm::{DialPeerCondition, NetworkBehaviour, NetworkBehaviourAction, NotifyHandler, PollParameters, ProtocolsHandler};
+
+use crate::config::GossipsubConfig;
+use crate::handler::GossipsubHandler;
+use crate::mcache::MessageCache;
+use crate::protocol::{
+    GossipsubControlAction, GossipsubMessage, GossipsubSubscription, GossipsubSubscriptionAction,
+    MessageId,
+};
+use crate::topic::{Hasher, Topic, TopicHash};
+
 mod tests;
 
 /// Network behaviour that handles the gossipsub protocol.
@@ -70,6 +72,9 @@ pub struct Gossipsub {
     /// A map of all connected peers to their subscribed topics.
     peer_topics: HashMap<PeerId, Vec<TopicHash>>,
 
+    ///A set of all explicit peers
+    explicit_peers: HashSet<PeerId>,
+
     /// Overlay network of connected peers - Maps topics to connected gossipsub peers.
     mesh: HashMap<TopicHash, Vec<PeerId>>,
 
@@ -84,6 +89,9 @@ pub struct Gossipsub {
 
     /// Heartbeat interval stream.
     heartbeat: Interval,
+
+    /// Interval stream for checking explicit peers.
+    check_explicit_peers: Interval,
 }
 
 impl Gossipsub {
@@ -109,6 +117,7 @@ impl Gossipsub {
             message_source_id,
             topic_peers: HashMap::new(),
             peer_topics: HashMap::new(),
+            explicit_peers: HashSet::new(),
             mesh: HashMap::new(),
             fanout: HashMap::new(),
             fanout_last_pub: HashMap::new(),
@@ -121,6 +130,10 @@ impl Gossipsub {
                 Instant::now() + gs_config.heartbeat_initial_delay,
                 gs_config.heartbeat_interval,
             ),
+            check_explicit_peers: Interval::new_at(
+                Instant::now() + gs_config.heartbeat_initial_delay,
+                gs_config.check_explicit_peers_interval,
+            ),
         }
     }
 
@@ -336,6 +349,23 @@ impl Gossipsub {
         true
     }
 
+    /// Adds a new peer to the list of explicitly connected peers.
+    pub fn add_explicit_peer(&mut self, peer_id: &PeerId) {
+        debug!("Adding explicit peer {:?}", peer_id);
+
+        self.explicit_peers.insert(peer_id.clone());
+
+        self.check_explicit_peer_connection(peer_id);
+    }
+
+    /// This removes the peer from explicitly connected peers, note that this does not disconnect
+    /// the peer.
+    pub fn remove_explicit_peer(&mut self, peer_id: &PeerId) {
+        debug!("Removing explicit peer {:?}", peer_id);
+
+        self.explicit_peers.remove(peer_id);
+    }
+
     /// Gossipsub JOIN(topic) - adds topic peers to mesh and sends them GRAFT messages.
     fn join(&mut self, topic_hash: &TopicHash) {
         debug!("Running JOIN for topic: {:?}", topic_hash);
@@ -350,11 +380,14 @@ impl Gossipsub {
 
         // check if we have mesh_n peers in fanout[topic] and add them to the mesh if we do,
         // removing the fanout entry.
-        if let Some((_, peers)) = self.fanout.remove_entry(topic_hash) {
+        if let Some((_, mut peers)) = self.fanout.remove_entry(topic_hash) {
             debug!(
                 "JOIN: Removing peers from the fanout for topic: {:?}",
                 topic_hash
             );
+            //remove explicit peers
+            peers = peers.into_iter().filter(|p| !self.explicit_peers.contains(p)).collect();
+
             // add up to mesh_n of them them to the mesh
             // Note: These aren't randomly added, currently FIFO
             let add_peers = std::cmp::min(peers.len(), self.config.mesh_n);
@@ -376,7 +409,7 @@ impl Gossipsub {
                 &self.topic_peers,
                 topic_hash,
                 self.config.mesh_n - added_peers.len(),
-                |_| true,
+                |p| !self.explicit_peers.contains(p),
             );
             added_peers.extend_from_slice(&new_peers);
             // add them to the mesh
@@ -426,6 +459,18 @@ impl Gossipsub {
         debug!("Completed LEAVE for topic: {:?}", topic_hash);
     }
 
+    ///Checks if the given peer is still connected and if not dials the peer again
+    fn check_explicit_peer_connection(&mut self, peer_id: &PeerId) {
+        if !self.peer_topics.contains_key(peer_id) {
+            //connect to peer
+            debug!("Connecting to explicit peer {:?}", peer_id);
+            self.events.push_back(NetworkBehaviourAction::DialPeer {
+                peer_id: peer_id.clone(),
+                condition: DialPeerCondition::Disconnected,
+            });
+        }
+    }
+
     /// Handles an IHAVE control message. Checks our cache of messages. If the message is unknown,
     /// requests it with an IWANT control message.
     fn handle_ihave(&mut self, peer_id: &PeerId, ihave_msgs: Vec<(TopicHash, Vec<MessageId>)>) {
@@ -503,19 +548,26 @@ impl Gossipsub {
         debug!("Handling GRAFT message for peer: {:?}", peer_id);
 
         let mut to_prune_topics = HashSet::new();
-        for topic_hash in topics {
-            if let Some(peers) = self.mesh.get_mut(&topic_hash) {
-                // if we are subscribed, add peer to the mesh, if not already added
-                info!(
-                    "GRAFT: Mesh link added for peer: {:?} in topic: {:?}",
-                    peer_id, topic_hash
-                );
-                // ensure peer is not already added
-                if !peers.contains(peer_id) {
-                    peers.push(peer_id.clone());
+        // we don't GRAFT to/from explicit peers; complain loudly if this happens
+        if self.explicit_peers.contains(peer_id) {
+            warn!("GRAFT: ignoring request from direct peer {:?}", peer_id);
+            // this is possibly a bug from non-reciprocal configuration; send a PRUNE for all topics
+            to_prune_topics = HashSet::from_iter(topics.into_iter());
+        } else {
+            for topic_hash in topics {
+                if let Some(peers) = self.mesh.get_mut(&topic_hash) {
+                    // if we are subscribed, add peer to the mesh, if not already added
+                    info!(
+                        "GRAFT: Mesh link added for peer: {:?} in topic: {:?}",
+                        peer_id, topic_hash
+                    );
+                    // ensure peer is not already added
+                    if !peers.contains(peer_id) {
+                        peers.push(peer_id.clone());
+                    }
+                } else {
+                    to_prune_topics.insert(topic_hash.clone());
                 }
-            } else {
-                to_prune_topics.insert(topic_hash.clone());
             }
         }
 
@@ -636,14 +688,16 @@ impl Gossipsub {
                     }
 
                     // if the mesh needs peers add the peer to the mesh
-                    if let Some(peers) = self.mesh.get_mut(&subscription.topic_hash) {
-                        if peers.len() < self.config.mesh_n_low {
-                            debug!(
-                                "SUBSCRIPTION: Adding peer {:?} to the mesh",
-                                propagation_source,
-                            );
+                    if !self.explicit_peers.contains(propagation_source) {
+                        if let Some(peers) = self.mesh.get_mut(&subscription.topic_hash) {
+                            if peers.len() < self.config.mesh_n_low {
+                                debug!(
+                                    "SUBSCRIPTION: Adding peer {:?} to the mesh",
+                                    propagation_source,
+                                );
+                            }
+                            peers.push(propagation_source.clone());
                         }
-                        peers.push(propagation_source.clone());
                     }
                     // generates a subscription event to be polled
                     self.events.push_back(NetworkBehaviourAction::GenerateEvent(
@@ -697,6 +751,7 @@ impl Gossipsub {
         let mut to_prune = HashMap::new();
 
         // maintain the mesh for each topic
+        let explicit_peers = &self.explicit_peers;
         for (topic_hash, peers) in self.mesh.iter_mut() {
             // too little peers - add some
             if peers.len() < self.config.mesh_n_low {
@@ -710,7 +765,7 @@ impl Gossipsub {
                 let desired_peers = self.config.mesh_n - peers.len();
                 let peer_list =
                     Self::get_random_peers(&self.topic_peers, topic_hash, desired_peers, {
-                        |peer| !peers.contains(peer)
+                        |peer| !peers.contains(peer) && !explicit_peers.contains(peer)
                     });
                 for peer in &peer_list {
                     let current_topic = to_graft.entry(peer.clone()).or_insert_with(|| vec![]);
@@ -813,6 +868,7 @@ impl Gossipsub {
 
         // shift the memcache
         self.mcache.shift();
+
         debug!("Completed Heartbeat");
     }
 
@@ -831,7 +887,7 @@ impl Gossipsub {
                 &self.topic_peers,
                 &topic_hash,
                 self.config.gossip_lazy,
-                |peer| !peers.contains(peer),
+                |peer| !peers.contains(peer) && !self.explicit_peers.contains(peer),
             );
             for peer in to_msg_peers {
                 // send an IHAVE message
@@ -925,6 +981,15 @@ impl Gossipsub {
             }
         }
 
+        //add explicit peers
+        for p in &self.explicit_peers {
+            if let Some(topics) = self.peer_topics.get(p) {
+                if message.topics.iter().any(|t| topics.contains(t)) {
+                    recipient_peers.insert(p.clone());
+                }
+            }
+        }
+
         // forward the message to peers
         if !recipient_peers.is_empty() {
             let event = Arc::new(GossipsubRpc {
@@ -1193,6 +1258,13 @@ impl NetworkBehaviour for Gossipsub {
             self.heartbeat();
         }
 
+        //check connections to explicit peers
+        while let Poll::Ready(Some(())) = self.check_explicit_peers.poll_next_unpin(cx) {
+            for p in self.explicit_peers.clone() {
+                self.check_explicit_peer_connection(&p);
+            }
+        }
+
         Poll::Pending
     }
 }