Enable CI and fix clippy errors

.github/workflows/ci.yml

@@ -0,0 +1,47 @@
+name: Tests
+
+on:
+  push:
+    branches:
+      - master
+      - develop
+  pull_request:
+    types:
+      - opened
+      - synchronize
+
+jobs:
+  test:
+    name: Tests on ubuntu-latest
+    runs-on: ubuntu-latest
+
+    steps:
+      - name: Install Protoc
+        uses: arduino/setup-protoc@v1
+      - uses: actions/checkout@v2
+        with:
+          fetch-depth: 0
+      - uses: actions-rs/toolchain@v1
+        with:
+          toolchain: stable
+          override: true
+      - name: build and run tests
+        run: cargo test --features integration_tests -- --test-threads=1
+      - name: Run `peer-punish-too-slow` integration test
+        run: cargo test integration_test_peer_punish_too_slow --features test_peer_punish_too_slow,integration_tests 
+  clippy:
+    name: Clippy (linter)
+    runs-on: ubuntu-latest
+    steps:
+      - name: Install Protoc
+        uses: arduino/setup-protoc@v1
+      - uses: actions/checkout@v2
+      - uses: actions-rs/toolchain@v1
+        with:
+          components: clippy
+          override: true
+      - name: Check with Clippy
+        uses: actions-rs/clippy-check@v1
+        with:
+          token: ${{ secrets.GITHUB_TOKEN }}
+          args: --all-features -- -D warnings

rust-toolchain

@@ -0,0 +1 @@
+1.68.2

spectrum-network/src/network_controller.rs

@@ -452,11 +452,10 @@ where
                 }) = self.enabled_peers.get_mut(&peer_id)
                 {
                     let protocol_id = protocol_tag.protocol_id();
-                    match enabled_protocols.get(&protocol_id) {
-                        Some((_, prot_handler)) => {
-                            prot_handler.incoming_msg(peer_id, protocol_tag.protocol_ver(), content);
-                        }
-                        None => {} // todo: probably possible?
+                    if let Some((_, prot_handler)) = enabled_protocols.get(&protocol_id) {
+                        prot_handler.incoming_msg(peer_id, protocol_tag.protocol_ver(), content);
+                    } else {
+                        // todo: probably possible?
                     };
                 }
             }

spectrum-network/src/peer_conn_handler.rs

@@ -60,31 +60,24 @@ pub enum ProtocolState {
     Opened {
         substream_in: ProtocolSubstreamIn<NegotiatedSubstream>,
         substream_out: ProtocolSubstreamOut<NegotiatedSubstream>,
-        pending_messages_recv: stream::Peekable<
-            stream::Select<
-                stream::Fuse<mpsc::Receiver<StreamNotification>>,
-                stream::Fuse<mpsc::Receiver<StreamNotification>>,
-            >,
-        >,
+        pending_messages_recv: stream::Peekable<stream::Select<AsyncReceiver, SyncReceiver>>,
     },
     /// Inbound substream is closed by peer.
     InboundClosedByPeer {
         /// None in the case when the peer closed inbound substream while outbound one
         /// hasn't been negotiated yet.
         substream_out: ProtocolSubstreamOut<NegotiatedSubstream>,
-        pending_messages_recv: stream::Peekable<
-            stream::Select<
-                stream::Fuse<mpsc::Receiver<StreamNotification>>,
-                stream::Fuse<mpsc::Receiver<StreamNotification>>,
-            >,
-        >,
+        pending_messages_recv: stream::Peekable<stream::Select<AsyncReceiver, SyncReceiver>>,
     },
     /// Outbound substream is closed by peer.
     OutboundClosedByPeer {
         substream_in: ProtocolSubstreamIn<NegotiatedSubstream>,
     },
 }
 
+type AsyncReceiver = stream::Fuse<mpsc::Receiver<StreamNotification>>;
+type SyncReceiver = stream::Fuse<mpsc::Receiver<StreamNotification>>;
+
 impl Debug for ProtocolState {
     fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
         match self {
@@ -246,6 +239,7 @@ impl IntoConnectionHandler for PartialPeerConnHandler {
     }
 }
 
+#[allow(dead_code)]
 pub struct PeerConnHandler {
     conf: PeerConnHandlerConf,
     protocols: HashMap<ProtocolId, Protocol>,
@@ -719,5 +713,6 @@ enum ThrottleStage {
     Start,
     InProgress,
     Finish,
+    #[cfg(not(feature = "test_peer_punish_too_slow"))]
     Disable,
 }

spectrum-network/src/peer_conn_handler/message_sink.rs

@@ -1,6 +1,6 @@
 use crate::types::RawMessage;
 use futures::{
-    channel::mpsc,
+    channel::mpsc::{self, SendError},
     lock::{Mutex as AsyncMutex, MutexGuard},
     prelude::*,
 };
@@ -46,6 +46,11 @@ struct MessageSinkIn {
     sync_channel: Mutex<Option<mpsc::Sender<StreamNotification>>>,
 }
 
+pub enum MessageSinkError {
+    BufferFull,
+    SenderDestroyed,
+}
+
 impl MessageSink {
     /// Returns the [`PeerId`] the sink is connected to.
     pub fn peer_id(&self) -> &PeerId {
@@ -56,7 +61,7 @@ impl MessageSink {
     ///
     /// If the buffer is exhausted, the channel will be closed
     /// via `SyncNotification::ForceClose` directive.
-    pub fn send_message(&self, msg: RawMessage) -> Result<(), ()> {
+    pub fn send_message(&self, msg: RawMessage) -> Result<(), MessageSinkError> {
         let lock = self.inner.sync_channel.lock();
         if let Ok(mut permit) = lock {
             if let Some(snd) = permit.as_mut() {
@@ -71,13 +76,13 @@ impl MessageSink {
 
                     // Destroy the sender in order to not send more `ForceClose` messages.
                     *permit = None;
-                    return Err(());
+                    return Err(MessageSinkError::BufferFull);
                 }
             } else {
-                return Err(());
+                return Err(MessageSinkError::SenderDestroyed);
             }
         }
-        return Ok(());
+        Ok(())
     }
 
     /// Wait until the remote is ready to accept a message.
@@ -106,9 +111,7 @@ impl<'a> Ready<'a> {
     /// Consumes this slots reservation and actually queues the notification.
     ///
     /// Returns an error if the substream has been closed.
-    pub fn send(mut self, msg: RawMessage) -> Result<(), ()> {
-        self.lock
-            .start_send(StreamNotification::Message(msg))
-            .map_err(|_| ())
+    pub fn send(mut self, msg: RawMessage) -> Result<(), SendError> {
+        self.lock.start_send(StreamNotification::Message(msg))
     }
 }

spectrum-network/src/peer_manager.rs

@@ -484,7 +484,7 @@ impl<S: PeersState> PeerManagerNotificationsBehavior for PeerManager<S> {
         }
     }
 
-    fn on_connection_established(&mut self, peer_id: PeerId, conn_id: ConnectionId) {
+    fn on_connection_established(&mut self, peer_id: PeerId, _conn_id: ConnectionId) {
         if let Some(PeerInState::Connected(mut cp)) = self.state.peer(&peer_id) {
             cp.confirm_connection();
         } else {

spectrum-network/src/peer_manager/data.rs

@@ -168,7 +168,7 @@ impl<'de> Deserialize<'de> for PeerDestination {
             }
         }
 
-        const VARIANTS: &'static [&'static str] = &["PeerId", "PeerIdWithAddr"];
+        const VARIANTS: &[&str] = &["PeerId", "PeerIdWithAddr"];
         deserializer.deserialize_enum("PeerDestination", VARIANTS, PeerDestinationVisitor)
     }
 }

spectrum-network/src/peer_manager/peers_state.rs

@@ -286,6 +286,7 @@ pub enum PeerStateFilter {
 
 const MAX_BOOT_PEERS: usize = 8;
 
+#[allow(clippy::large_enum_variant)]
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub enum NetworkingState {
     /// The node has few known peers.

spectrum-network/src/protocol_handler.rs

@@ -99,6 +99,7 @@ pub trait ProtocolBehaviour {
     /// Inject an event of protocol being disabled with a peer.
     fn inject_protocol_disabled(&mut self, peer_id: PeerId);
 
+    #[allow(clippy::type_complexity)]
     /// Poll for output actions.
     fn poll(
         &mut self,
@@ -155,7 +156,10 @@ where
                         trace!("Sending message {:?} to peer {}", message, peer_id);
                         if let Some(sink) = self.peers.get(&peer_id) {
                             trace!("Sink is available");
-                            if let Err(_) = sink.send_message(codec::BinCodec::encode(message.clone())) {
+                            if sink
+                                .send_message(codec::BinCodec::encode(message.clone()))
+                                .is_err()
+                            {
                                 trace!("Failed to submit a message to {:?}. Channel is closed.", peer_id)
                             }
                             trace!("Sent");

spectrum-network/src/types.rs

@@ -150,6 +150,6 @@ impl From<RawMessage> for Vec<u8> {
 
 impl AsRef<[u8]> for RawMessage {
     fn as_ref(&self) -> &[u8] {
-        &*self.0
+        &self.0
     }
 }

spectrum-network/tests/integration_tests/fake_sync_behaviour.rs

@@ -58,6 +58,7 @@ impl spectrum_network::protocol_handler::ProtocolSpec for FakeSyncSpec {
 
 type SyncBehaviourOut = ProtocolBehaviourOut<SyncHandshake, FakeSyncMessage>;
 
+#[allow(dead_code)]
 #[derive(Debug, Display)]
 pub enum SyncBehaviorError {
     EmptyPeers,
@@ -146,11 +147,11 @@ where
             }))
     }
 
-    fn inject_message(&mut self, peer_id: PeerId, msg: FakeSyncMessage) {
+    fn inject_message(&mut self, peer_id: PeerId, _msg: FakeSyncMessage) {
         self.send_fake_msg(peer_id);
     }
 
-    fn inject_malformed_mesage(&mut self, peer_id: PeerId, details: MalformedMessage) {}
+    fn inject_malformed_mesage(&mut self, _peer_id: PeerId, _detailss: MalformedMessage) {}
 
     fn inject_protocol_requested(&mut self, peer_id: PeerId, handshake: Option<SyncHandshake>) {
         if let Some(SyncHandshake::HandshakeV1(hs)) = handshake {