[On ice] Gossipsub: an extensible baseline pubsub protocol, based on randomized topic meshes and gossip

Cargo.toml

@@ -23,6 +23,7 @@ libp2p-mplex = { version = "0.2.0", path = "./muxers/mplex" }
 libp2p-identify = { version = "0.2.0", path = "./protocols/identify" }
 libp2p-kad = { version = "0.2.0", path = "./protocols/kad" }
 libp2p-floodsub = { version = "0.2.0", path = "./protocols/floodsub" }
+libp2p-gossipsub = { version = "0.2.0", path = "./protocols/gossipsub" }
 libp2p-ping = { version = "0.2.0", path = "./protocols/ping" }
 libp2p-plaintext = { version = "0.2.0", path = "./protocols/plaintext" }
 libp2p-ratelimit = { version = "0.2.0", path = "./transports/ratelimit" }
@@ -64,6 +65,7 @@ members = [
     "muxers/mplex",
     "muxers/yamux",
     "protocols/floodsub",
+    "protocols/gossipsub",
     "protocols/identify",
     "protocols/kad",
     "protocols/observed",

core/src/peer_id.rs

@@ -162,6 +162,11 @@ impl Into<multihash::Multihash> for PeerId {
     }
 }
 
+impl AsRef<PeerId> for PeerId {
+    fn as_ref(&self) -> &PeerId {
+        &self
+    }
+}
 quick_error! {
     #[derive(Debug)]
     pub enum ParseError {

protocols/floodsub/src/handler.rs

@@ -182,36 +182,43 @@ where
             let mut substream = self.substreams.swap_remove(n);
             loop {
                 substream = match substream {
-                    SubstreamState::WaitingInput(mut substream) => match substream.poll() {
+                    SubstreamState::WaitingInput(mut substream)
+                    => match substream.poll() {
                         Ok(Async::Ready(Some(message))) => {
                             self.substreams
                                 .push(SubstreamState::WaitingInput(substream));
-                            return Ok(Async::Ready(ProtocolsHandlerEvent::Custom(message)));
-                        }
-                        Ok(Async::Ready(None)) => SubstreamState::Closing(substream),
+                            return Ok(Async::Ready(
+                                ProtocolsHandlerEvent::Custom(message)));
+                        },
+                        Ok(Async::Ready(None)) => SubstreamState::Closing(
+                            substream),
                         Ok(Async::NotReady) => {
                             self.substreams
                                 .push(SubstreamState::WaitingInput(substream));
                             return Ok(Async::NotReady);
-                        }
+                        },
                         Err(_) => SubstreamState::Closing(substream),
                     },
                     SubstreamState::PendingSend(mut substream, message) => {
                         match substream.start_send(message)? {
-                            AsyncSink::Ready => SubstreamState::PendingFlush(substream),
+                            AsyncSink::Ready => SubstreamState::PendingFlush(
+                                substream),
                             AsyncSink::NotReady(message) => {
                                 self.substreams
-                                    .push(SubstreamState::PendingSend(substream, message));
+                                    .push(SubstreamState::PendingSend(
+                                        substream, message));
                                 return Ok(Async::NotReady);
                             }
                         }
                     }
                     SubstreamState::PendingFlush(mut substream) => {
                         match substream.poll_complete()? {
-                            Async::Ready(()) => SubstreamState::Closing(substream),
+                            Async::Ready(()) => SubstreamState::Closing(
+                                substream),
                             Async::NotReady => {
                                 self.substreams
-                                    .push(SubstreamState::PendingFlush(substream));
+                                    .push(SubstreamState::PendingFlush(
+                                        substream));
                                 return Ok(Async::NotReady);
                             }
                         }

protocols/floodsub/src/layer.rs

@@ -130,22 +130,26 @@ impl<TSubstream> Floodsub<TSubstream> {
     /// Publishes a message to the network.
     ///
     /// > **Note**: Doesn't do anything if we're not subscribed to the topic.
-    pub fn publish(&mut self, topic: impl Into<TopicHash>, data: impl Into<Vec<u8>>) {
+    pub fn publish(&mut self, topic: impl Into<TopicHash>,
+        data: impl Into<Vec<u8>>) {
         self.publish_many(iter::once(topic), data)
     }
 
     /// Publishes a message with multiple topics to the network.
     ///
-    /// > **Note**: Doesn't do anything if we're not subscribed to any of the topics.
-    pub fn publish_many(&mut self, topic: impl IntoIterator<Item = impl Into<TopicHash>>, data: impl Into<Vec<u8>>) {
+    /// > **Note**: Doesn't do anything if we're not subscribed to any of the
+    /// > topics.
+    pub fn publish_many(&mut self, topics: impl IntoIterator<Item = impl
+        Into<TopicHash>>, data: impl Into<Vec<u8>>) {
         let message = FloodsubMessage {
             source: self.local_peer_id.clone(),
             data: data.into(),
-            // If the sequence numbers are predictable, then an attacker could flood the network
-            // with packets with the predetermined sequence numbers and absorb our legitimate
-            // messages. We therefore use a random number.
+            // If the sequence numbers are predictable, then an attacker could
+            // flood the network with packets with the predetermined sequence
+            // numbers and absorb our legitimate messages. We therefore use a
+            // random number.
             sequence_number: rand::random::<[u8; 20]>().to_vec(),
-            topics: topic.into_iter().map(|t| t.into().clone()).collect(),
+            topics: topics.into_iter().map(|t| t.into().clone()).collect(),
         };
 
         // Don't publish the message if we're not subscribed ourselves to any of the topics.

protocols/floodsub/src/protocol.rs

@@ -61,7 +61,8 @@ where
 
     #[inline]
     fn upgrade_inbound(self, socket: TSocket, _: Self::Info) -> Self::Future {
-        future::ok(Framed::new(socket, FloodsubCodec { length_prefix: Default::default() }))
+        future::ok(Framed::new(socket, FloodsubCodec {
+            length_prefix: Default::default() }))
     }
 }
 
@@ -122,8 +123,9 @@ impl Encoder for FloodsubCodec {
         proto
             .write_length_delimited_to_writer(&mut dst.by_ref().writer())
             .expect(
-                "there is no situation in which the protobuf message can be invalid, and \
-                 writing to a BytesMut never fails as we reserved enough space beforehand",
+                "there is no situation in which the protobuf message can be \
+                invalid, and writing to a BytesMut never fails as we reserved \
+                enough space beforehand",
             );
         Ok(())
     }

protocols/gossipsub/Cargo.toml

@@ -0,0 +1,26 @@
+[package]
+name = "libp2p-gossipsub"
+description = "Gossipsub for libp2p."
+version = "0.2.0"
+authors = ["James Ray <[email protected]>"]
+repository = "https://github.com/libp2p/rust-libp2p"
+keywords = ["peer-to-peer", "libp2p", "networking", "pubsub", "gossipsub"]
+categories = ["network-programming", "asynchronous"]
+
+[dependencies]
+bs58 = "0.2.0"
+bytes = "0.4"
+chrono = "0.4"
+cuckoofilter = "0.3.2"
+custom_error = "1.3.0"
+fnv = "1.0"
+futures = "0.1"
+libp2p-core = { path = "../../core" }
+libp2p-floodsub = { path = "../floodsub" }
+lru_time_cache = "0.8.1"
+protobuf = "2.0.2"
+rand = "0.6"
+smallvec = "0.6.5"
+tokio-codec = "0.1"
+tokio-io = "0.1"
+unsigned-varint = { version = "0.2.1", features = ["codec"] }

protocols/gossipsub/gen_rpc_proto.sh

@@ -0,0 +1,10 @@
+#!/bin/sh
+
+# Similar to regen_structs_proto.sh but leaves installation of any
+# dependencies up to the user.
+# TODO: write a script that will install dependencies across platforms.
+# Basic tip: run `chmod +x gen_rpc_proto.sh && ./gen_rpc_proto.sh`
+
+protoc --rust_out . rpc.proto
+sudo chown $USER:$USER *.rs
+mv -f rpc.rs ./src/rpc_proto.rs

protocols/gossipsub/regen_structs_proto.sh

@@ -0,0 +1,13 @@
+#!/bin/sh
+
+# This script regenerates the `src/rpc_proto.rs` file from `rpc.proto`.
+
+docker run --rm -v `pwd`:/usr/code:z -w /usr/code rust /bin/bash -c " \
+    apt-get update; \
+    apt-get install -y protobuf-compiler; \
+    cargo install --version 2.0.2 protobuf-codegen; \
+    protoc --rust_out . rpc.proto"
+
+sudo chown $USER:$USER *.rs
+
+mv -f rpc.rs ./src/rpc_proto.rs

protocols/gossipsub/rpc.proto

@@ -0,0 +1,97 @@
+// From https://github.com/libp2p/go-libp2p-pubsub/blob/master/pb/rpc.proto,
+// should be the same unless that has been changed.
+
+syntax = "proto2";
+
+// import "google/protobuf/descriptor.proto";
+
+// extend google.protobuf.EnumValueOptions {
+// 	optional string type= 51234;
+// 	optional string code= 51235;
+//   }
+
+package pubsub.pb;
+
+message RPC {
+	repeated SubOpts subscriptions = 1;
+	repeated Message publish = 2;
+
+	message SubOpts {
+		optional bool subscribe = 1; // subscribe or unsubcribe
+		optional string topic_hash = 2;
+	}
+
+	optional ControlMessage control = 3;
+}
+
+message Message {
+	optional bytes from = 1;
+	optional bytes data = 2;
+	optional bytes seqno = 3;
+	repeated string topic_hashes = 4;
+	// TODO: use
+	// optional bytes signature = 5;
+	// optional bytes key = 6;
+}
+
+message ControlMessage {
+	repeated ControlIHave ihave = 1;
+	repeated ControlIWant iwant = 2;
+	repeated ControlGraft graft = 3;
+	repeated ControlPrune prune = 4;
+}
+
+// Using an ID doesn't seem to be worth it compared to using a hash.
+// While this may not be strictly necessary as a messageID and a messageHash
+// are both strings, it would be useful to differentiate between them in the
+// protobuf file, e.g. due to the accessors and for clarity.
+// enum MessageRep {
+// 	ID = 0 [(type) = "ID", (code) = 'I'];
+// 	HASH = 1 [(type) = "HSH", (code) = 'H'];
+// }
+
+message ControlIHave {
+	optional string topic_hash = 1;
+	repeated string message_hashes = 2;
+}
+
+message ControlIWant {
+	repeated string message_hashes = 1;
+}
+
+message ControlGraft {
+	optional string topic_hash = 1;
+}
+
+message ControlPrune {
+	optional string topic_hash = 1;
+}
+
+message TopicDescriptor {
+	optional string name = 1;
+	// TODO: use
+	// optional AuthOpts auth = 2;
+	// optional EncOpts enc = 3;
+
+	// message AuthOpts {
+	// 	optional AuthMode mode = 1;
+	// 	repeated bytes keys = 2; // root keys to trust
+
+	// 	enum AuthMode {
+	// 		NONE = 0; // no authentication, anyone can publish
+	// 		KEY = 1; // only messages signed by keys in the topic descriptor are accepted
+	// 		WOT = 2; // web of trust, certificates can allow publisher set to grow
+	// 	}
+	// }
+
+	// message EncOpts {
+	// 	optional EncMode mode = 1;
+	// 	repeated bytes keyHashes = 2; // the hashes of the shared keys used (salted)
+
+	// 	enum EncMode {
+	// 		NONE = 0; // no encryption, anyone can read
+	// 		SHAREDKEY = 1; // messages are encrypted with shared key
+	// 		WOT = 2; // web of trust, certificates can allow publisher set to grow
+	// 	}
+	// }
+}

protocols/gossipsub/src/constants.rs

@@ -0,0 +1,31 @@
+// Overlay parameters
+
+/// The target number of peers in the mesh to gossip to and from.
+pub const TARGET_MESH_DEGREE: u32 = 6;
+/// Low water mark for the mesh degree, any lower and it could take longer to
+/// find messages.
+pub const LOW_WM_MESH_DEGREE: u32 = 4;
+/// High water mark for the mesh degree, any higher and it could be too
+/// much for bandwidth (particularly for low-end devices).
+pub const HIGH_WM_MESH_DEGREE: u32 = 12;
+
+// Gossip parameters
+/// Length of gossip history
+pub const GOSSIP_HIST_LEN: u32 = 5;
+
+/// We get message IDs from up to (but not including) this index in the
+/// `MCache's` history window.
+pub const HISTORY_GOSSIP: u32 = 3;
+
+/// The length of the total message history.
+pub const MSG_HIST_LEN: u32 = 120;
+/// The duration of messages in the seen cache, in seconds.
+pub const SEEN_MSGS_CACHE: u32 = 120;
+
+/// In milliseconds
+pub const HEARTBEAT_INITIAL_DELAY: u32 = 100;
+/// The interval at which to run the heartbeat procedure, in seconds
+pub const HEARTBEAT_INTERVAL: u32 = 1;
+
+/// The time to live for fanout in seconds.
+pub const FANOUT_TTL: u32 = 60;