Refactor RPC and implement hello protocol

blockchain/chain_sync/src/network_context.rs

@@ -4,7 +4,8 @@
 use async_std::sync::Sender;
 use blocks::TipSetKeys;
 use forest_libp2p::{
-    rpc::{BlockSyncRequest, RPCEvent, RPCRequest, RequestId},
+    blocksync::BlockSyncRequest,
+    rpc::{RPCEvent, RPCRequest, RequestId},
     NetworkMessage,
 };
 use libp2p::core::PeerId;
@@ -51,8 +52,8 @@ impl SyncNetworkContext {
         peer_id: PeerId,
         request: BlockSyncRequest,
     ) -> RequestId {
-        trace!("Sending Blocksync Request {:?}", request);
-        self.send_rpc_request(peer_id, RPCRequest::Blocksync(request))
+        trace!("Sending BlockSync Request {:?}", request);
+        self.send_rpc_request(peer_id, RPCRequest::BlockSync(request))
             .await
     }
 

blockchain/chain_sync/src/sync.rs

@@ -85,7 +85,7 @@ impl<'db, DB, ST> Future for ChainSyncer<'db, DB, ST> {
 
     fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
         match self.project().network_rx.poll_next(cx) {
-            Poll::Ready(Some(event)) => info!("chain syncer received event: {:?}", event),
+            Poll::Ready(Some(_event)) => (),
             Poll::Pending | Poll::Ready(None) => (),
         };
         Poll::Pending
@@ -147,7 +147,7 @@ where
         loop {
             select! {
                 network_msg = nw.next().fuse() => match network_msg {
-                    Some(event) => info!("received some other event: {:?}", event),
+                    Some(event) =>(),
                     None => break,
                 }
             }

node/forest_libp2p/Cargo.toml

@@ -20,6 +20,8 @@ forest_cid = { path = "../../ipld/cid" }
 bytes = "0.5.2"
 fnv = "1.0.6"
 smallvec = "1.1.0"
+clock = { path = "../clock" }
+num-bigint = { path = "../../math/bigint", package = "forest_bigint" }
 
 [dev-dependencies]
 forest_address = { path = "../../vm/address" }

node/forest_libp2p/src/blocksync/mod.rs

@@ -0,0 +1,8 @@
+// Copyright 2020 ChainSafe Systems
+// SPDX-License-Identifier: Apache-2.0, MIT
+
+mod message;
+
+pub use self::message::*;
+
+pub const BLOCKSYNC_PROTOCOL_ID: &[u8] = b"/fil/sync/blk/0.0.1";

node/forest_libp2p/src/hello/message.rs

@@ -0,0 +1,94 @@
+// Copyright 2020 ChainSafe Systems
+// SPDX-License-Identifier: Apache-2.0, MIT
+
+use clock::ChainEpoch;
+use forest_cid::Cid;
+use forest_encoding::{de::Deserializer, ser::Serializer};
+use num_bigint::{bigint_ser, BigInt};
+use serde::{Deserialize, Serialize};
+
+/// Hello message https://filecoin-project.github.io/specs/#hello-spec
+#[derive(Clone, Debug, PartialEq, Default)]
+pub struct HelloMessage {
+    pub heaviest_tip_set: Vec<Cid>,
+    pub heaviest_tipset_height: ChainEpoch,
+    pub heaviest_tipset_weight: BigInt,
+    pub genesis_hash: Cid,
+}
+
+impl Serialize for HelloMessage {
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: Serializer,
+    {
+        #[derive(Serialize)]
+        struct TupleHelloMessage<'a>(
+            &'a [Cid],
+            &'a ChainEpoch,
+            #[serde(with = "bigint_ser")] &'a BigInt,
+            &'a Cid,
+        );
+        TupleHelloMessage(
+            &self.heaviest_tip_set,
+            &self.heaviest_tipset_height,
+            &self.heaviest_tipset_weight,
+            &self.genesis_hash,
+        )
+        .serialize(serializer)
+    }
+}
+
+impl<'de> Deserialize<'de> for HelloMessage {
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: Deserializer<'de>,
+    {
+        #[derive(Deserialize)]
+        struct TupleHelloMessage(
+            Vec<Cid>,
+            ChainEpoch,
+            #[serde(with = "bigint_ser")] BigInt,
+            Cid,
+        );
+        let TupleHelloMessage(
+            heaviest_tip_set,
+            heaviest_tipset_height,
+            heaviest_tipset_weight,
+            genesis_hash,
+        ) = Deserialize::deserialize(deserializer)?;
+        Ok(HelloMessage {
+            heaviest_tip_set,
+            heaviest_tipset_height,
+            heaviest_tipset_weight,
+            genesis_hash,
+        })
+    }
+}
+
+/// Response to a Hello
+#[derive(Clone, Debug, PartialEq)]
+pub struct HelloResponse {
+    /// Time of arrival in unix nanoseconds
+    pub arrival: i64,
+    /// Time sent in unix nanoseconds
+    pub sent: i64,
+}
+
+impl Serialize for HelloResponse {
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: Serializer,
+    {
+        (&self.arrival, &self.sent).serialize(serializer)
+    }
+}
+
+impl<'de> Deserialize<'de> for HelloResponse {
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: Deserializer<'de>,
+    {
+        let (arrival, sent) = Deserialize::deserialize(deserializer)?;
+        Ok(HelloResponse { arrival, sent })
+    }
+}

node/forest_libp2p/src/hello/mod.rs

@@ -0,0 +1,8 @@
+// Copyright 2020 ChainSafe Systems
+// SPDX-License-Identifier: Apache-2.0, MIT
+
+mod message;
+
+pub use self::message::*;
+
+pub const HELLO_PROTOCOL_ID: &[u8] = b"/fil/hello/1.0.0";

node/forest_libp2p/src/lib.rs

@@ -4,7 +4,9 @@
 #![recursion_limit = "1024"]
 
 mod behaviour;
+pub mod blocksync;
 mod config;
+pub mod hello;
 pub mod rpc;
 mod service;
 

node/forest_libp2p/src/rpc/codec.rs

@@ -1,46 +1,21 @@
 // Copyright 2020 ChainSafe Systems
 // SPDX-License-Identifier: Apache-2.0, MIT
 
-use super::{RPCRequest, RPCResponse};
+use super::{RPCError, RPCRequest, RPCResponse};
+use crate::blocksync::BLOCKSYNC_PROTOCOL_ID;
+use crate::hello::HELLO_PROTOCOL_ID;
 use bytes::BytesMut;
-use forest_encoding::{error::Error as EncodingError, from_slice, to_vec};
+use forest_encoding::{from_slice, to_vec};
 use futures_codec::{Decoder, Encoder};
-use std::fmt;
 
 /// Codec used for inbound connections. Decodes the inbound message into a RPCRequest, and encodes the RPCResponse to send.
-pub struct InboundCodec;
-/// Codec used for outbound connections. Encodes the outbound message into a RPCRequest to send, and decodes the RPCResponse when received.
-pub struct OutboundCodec;
-
-#[derive(Debug, Clone, PartialEq)]
-pub enum RPCError {
-    Codec(String),
-    Custom(String),
-}
-impl From<std::io::Error> for RPCError {
-    fn from(err: std::io::Error) -> Self {
-        Self::Custom(err.to_string())
-    }
+pub struct InboundCodec {
+    protocol: &'static [u8],
 }
 
-impl From<EncodingError> for RPCError {
-    fn from(err: EncodingError) -> Self {
-        Self::Codec(err.to_string())
-    }
-}
-
-impl fmt::Display for RPCError {
-    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        match self {
-            RPCError::Codec(err) => write!(f, "Codec Error: {}", err),
-            RPCError::Custom(err) => write!(f, "{}", err),
-        }
-    }
-}
-
-impl std::error::Error for RPCError {
-    fn description(&self) -> &str {
-        "Libp2p RPC Error"
+impl InboundCodec {
+    pub fn new(protocol: &'static [u8]) -> Self {
+        Self { protocol }
     }
 }
 
@@ -50,7 +25,13 @@ impl Encoder for InboundCodec {
 
     fn encode(&mut self, item: Self::Item, dst: &mut BytesMut) -> Result<(), Self::Error> {
         match item {
-            RPCResponse::Blocksync(response) => {
+            RPCResponse::BlockSync(response) => {
+                let resp = to_vec(&response)?;
+                dst.clear();
+                dst.extend_from_slice(&resp);
+                Ok(())
+            }
+            RPCResponse::Hello(response) => {
                 let resp = to_vec(&response)?;
                 dst.clear();
                 dst.extend_from_slice(&resp);
@@ -69,19 +50,41 @@ impl Decoder for InboundCodec {
             return Ok(None);
         }
 
-        Ok(Some(RPCRequest::Blocksync(
-            from_slice(bz).map_err(|err| RPCError::Codec(err.to_string()))?,
-        )))
+        match self.protocol {
+            HELLO_PROTOCOL_ID => Ok(Some(RPCRequest::Hello(
+                from_slice(bz).map_err(|err| RPCError::Codec(err.to_string()))?,
+            ))),
+            BLOCKSYNC_PROTOCOL_ID => Ok(Some(RPCRequest::BlockSync(
+                from_slice(bz).map_err(|err| RPCError::Codec(err.to_string()))?,
+            ))),
+            _ => Err(RPCError::Codec("Unsupported codec".to_string())),
+        }
+    }
+}
+
+/// Codec used for outbound connections. Encodes the outbound message into a RPCRequest to send, and decodes the RPCResponse when received.
+pub struct OutboundCodec {
+    protocol: &'static [u8],
+}
+
+impl OutboundCodec {
+    pub fn new(protocol: &'static [u8]) -> Self {
+        Self { protocol }
     }
 }
 
 impl Encoder for OutboundCodec {
     type Error = RPCError;
     type Item = RPCRequest;
-
     fn encode(&mut self, item: Self::Item, dst: &mut BytesMut) -> Result<(), Self::Error> {
         match item {
-            RPCRequest::Blocksync(request) => {
+            RPCRequest::BlockSync(request) => {
+                let resp = to_vec(&request)?;
+                dst.clear();
+                dst.extend_from_slice(&resp);
+                Ok(())
+            }
+            RPCRequest::Hello(request) => {
                 let resp = to_vec(&request)?;
                 dst.clear();
                 dst.extend_from_slice(&resp);
@@ -90,18 +93,22 @@ impl Encoder for OutboundCodec {
         }
     }
 }
+
 impl Decoder for OutboundCodec {
     type Error = RPCError;
     type Item = RPCResponse;
-
     fn decode(&mut self, bz: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
         if bz.is_empty() {
             return Ok(None);
         }
-
-        Ok(Some(RPCResponse::Blocksync(
-            // Replace map
-            from_slice(bz).map_err(|err| RPCError::Codec(err.to_string()))?,
-        )))
+        match self.protocol {
+            HELLO_PROTOCOL_ID => Ok(Some(RPCResponse::Hello(
+                from_slice(bz).map_err(|err| RPCError::Codec(err.to_string()))?,
+            ))),
+            BLOCKSYNC_PROTOCOL_ID => Ok(Some(RPCResponse::BlockSync(
+                from_slice(bz).map_err(|err| RPCError::Codec(err.to_string()))?,
+            ))),
+            _ => Err(RPCError::Codec("Unsupported codec".to_string())),
+        }
     }
 }

node/forest_libp2p/src/rpc/error.rs

@@ -0,0 +1,38 @@
+// Copyright 2020 ChainSafe Systems
+// SPDX-License-Identifier: Apache-2.0, MIT
+
+use forest_encoding::error::Error as EncodingError;
+use std::fmt;
+
+#[derive(Debug, Clone, PartialEq)]
+pub enum RPCError {
+    Codec(String),
+    Custom(String),
+}
+
+impl From<std::io::Error> for RPCError {
+    fn from(err: std::io::Error) -> Self {
+        Self::Custom(err.to_string())
+    }
+}
+
+impl From<EncodingError> for RPCError {
+    fn from(err: EncodingError) -> Self {
+        Self::Codec(err.to_string())
+    }
+}
+
+impl fmt::Display for RPCError {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            RPCError::Codec(err) => write!(f, "Codec Error: {}", err),
+            RPCError::Custom(err) => write!(f, "{}", err),
+        }
+    }
+}
+
+impl std::error::Error for RPCError {
+    fn description(&self) -> &str {
+        "Libp2p RPC Error"
+    }
+}

node/forest_libp2p/src/rpc/handler.rs

@@ -1,10 +1,8 @@
 // Copyright 2020 ChainSafe Systems
 // SPDX-License-Identifier: Apache-2.0, MIT
 
-use super::{
-    InboundCodec, OutboundFramed, RPCError, RPCEvent, RPCInbound, RPCOutbound, RPCResponse,
-    RequestId,
-};
+use super::protocol::{OutboundFramed, RPCInbound};
+use super::{InboundCodec, RPCError, RPCEvent, RPCRequest, RPCResponse, RequestId};
 use fnv::FnvHashMap;
 use futures::prelude::*;
 use futures::{AsyncRead, AsyncWrite};
@@ -140,7 +138,7 @@ where
     type Error = RPCError;
     type Substream = TSubstream;
     type InboundProtocol = RPCInbound;
-    type OutboundProtocol = RPCOutbound;
+    type OutboundProtocol = RPCRequest;
     type OutboundOpenInfo = RPCEvent;
 
     fn listen_protocol(&self) -> SubstreamProtocol<Self::InboundProtocol> {
@@ -347,9 +345,10 @@ where
             if self.dial_negotiated < self.max_dial_negotiated {
                 self.dial_negotiated += 1;
                 let event = self.dial_queue.remove(0);
+                self.dial_queue.shrink_to_fit();
                 if let RPCEvent::Request(id, req) = event {
                     return Poll::Ready(ProtocolsHandlerEvent::OutboundSubstreamRequest {
-                        protocol: SubstreamProtocol::new(RPCOutbound { req: req.clone() }),
+                        protocol: SubstreamProtocol::new(req.clone()),
                         info: RPCEvent::Request(id, req),
                     });
                 }