Repository: mintlayer/mintlayer-core
Pull request #109 (2022-02-25 05:07:36)

Network event updates
This PR splits the `net::Event` enum into two separate enums, `net::ConnectivityEvent` and `net::FloodsubEvent`, which is required for the implementation of separated connectivity and floodsub handling in application-level code. It also temporarily disables libp2p as a network service provider because it's easier to demonstrate the changes I wish to make to the network interface if they are first introduced only to the mock interface which is significantly simpler.

The reason why we want to split these two is that code becomes easier to maintain, especially in the future when all of the peer and syncing-related features have been implemented. Without the split they would all be in `src/lib.rs` which would grow into a very long and unmaintainable file.


Patch:
diff --git a/p2p/src/lib.rs b/p2p/src/lib.rs
index e522c25ad..a686bdc16 100644
--- a/p2p/src/lib.rs
+++ b/p2p/src/lib.rs
@@ -320,12 +320,17 @@ where
         event: net::Event<NetworkingBackend>,
     ) -> error::Result<()> {
         match event {
-            net::Event::IncomingConnection(peer_id, socket) => {
-                self.on_connectivity_event(ConnectivityEvent::Accept(peer_id, socket)).await
+            net::Event::Connectivity(event) => match event {
+                net::ConnectivityEvent::IncomingConnection(peer_id, socket) => {
+                    self.on_connectivity_event(ConnectivityEvent::Accept(peer_id, socket)).await
+                }
+                net::ConnectivityEvent::PeerDiscovered(peers) => self.peer_discovered(&peers),
+                net::ConnectivityEvent::PeerExpired(peers) => self.peer_expired(&peers),
+            },
+            net::Event::Floodsub(event) => {
+                let net::FloodsubEvent::MessageReceived(topic, message) = event;
+                self.on_floodsub_event(topic, message)
             }
-            net::Event::PeerDiscovered(peers) => self.peer_discovered(&peers),
-            net::Event::PeerExpired(peers) => self.peer_expired(&peers),
-            net::Event::MessageReceived(topic, message) => self.on_floodsub_event(topic, message),
         }
     }
 
@@ -380,8 +385,8 @@ mod tests {
     use super::*;
     use crate::error::P2pError;
     use common::chain::config;
-    use libp2p::Multiaddr;
-    use net::{libp2p::Libp2pService, mock::MockService};
+    // use libp2p::Multiaddr;
+    use net::mock::MockService;
     use std::net::SocketAddr;
     use tokio::net::TcpListener;
 
@@ -400,6 +405,7 @@ mod tests {
         );
     }
 
+    /*
     // try to connect to an address that no one listening on and verify it fails
     #[tokio::test]
     async fn test_p2p_connect_libp2p() {
@@ -417,6 +423,7 @@ mod tests {
             Err(P2pError::SocketError(std::io::ErrorKind::ConnectionRefused))
         );
     }
+    */
 
     // verify that if handshake succeeds, peer state is set to `Active`
     #[tokio::test]
@@ -482,6 +489,7 @@ mod tests {
         assert_eq!(p2p.peers.len(), 0);
     }
 
+    /*
     #[tokio::test]
     async fn test_peer_discovered_libp2p() {
         let config = Arc::new(config::create_mainnet());
@@ -615,6 +623,7 @@ mod tests {
             vec![Arc::new("/ip6/::1/tcp/9097".parse().unwrap())],
         );
     }
+    */
 
     // verify that if the node is aware of any peers on the network,
     // call to `auto_connect()` will establish a connection with them
diff --git a/p2p/src/net/libp2p/backend.rs b/p2p/src/net/libp2p/backend.rs
index 8928da37f..0c1d386ed 100644
--- a/p2p/src/net/libp2p/backend.rs
+++ b/p2p/src/net/libp2p/backend.rs
@@ -15,6 +15,7 @@
 // limitations under the License.
 //
 // Author(s): A. Altonen
+/*
 use crate::{
     error::{self, Libp2pError, P2pError},
     message,
@@ -501,3 +502,4 @@ mod tests {
         assert_eq!(backend.run().await, Err(P2pError::ChannelClosed));
     }
 }
+*/
diff --git a/p2p/src/net/libp2p/mod.rs b/p2p/src/net/libp2p/mod.rs
index 4f4e8b121..adc56b049 100644
--- a/p2p/src/net/libp2p/mod.rs
+++ b/p2p/src/net/libp2p/mod.rs
@@ -16,6 +16,7 @@
 // limitations under the License.
 //
 // Author(s): A. Altonen
+#![allow(unused)]
 use crate::{
     error::{self, Libp2pError, P2pError},
     net::{self, Event, FloodsubTopic, NetworkService, SocketService},
@@ -161,6 +162,7 @@ where
         .collect::<Vec<net::AddrInfo<T>>>()
 }
 
+/*
 #[async_trait]
 impl NetworkService for Libp2pService {
     type Address = Multiaddr;
@@ -777,3 +779,4 @@ mod tests {
         );
     }
 }
+*/
diff --git a/p2p/src/net/mock/backend.rs b/p2p/src/net/mock/backend.rs
index 6ec406a59..5075eb5f8 100644
--- a/p2p/src/net/mock/backend.rs
+++ b/p2p/src/net/mock/backend.rs
@@ -44,19 +44,24 @@ pub struct Backend {
     cmd_rx: mpsc::Receiver<types::Command>,
 
     /// TX channel for sending events to the frontend
-    event_tx: mpsc::Sender<types::Event>,
+    conn_tx: mpsc::Sender<types::ConnectivityEvent>,
+
+    /// TX channel for sending events to the frontend
+    _flood_tx: mpsc::Sender<types::FloodsubEvent>,
 }
 
 impl Backend {
     pub fn new(
         socket: TcpListener,
         cmd_rx: mpsc::Receiver<types::Command>,
-        event_tx: mpsc::Sender<types::Event>,
+        conn_tx: mpsc::Sender<types::ConnectivityEvent>,
+        _flood_tx: mpsc::Sender<types::FloodsubEvent>,
     ) -> Self {
         Self {
             socket,
             cmd_rx,
-            event_tx,
+            conn_tx,
+            _flood_tx,
         }
     }
 
@@ -64,7 +69,7 @@ impl Backend {
         loop {
             tokio::select! {
                 event = self.socket.accept() => match event {
-                    Ok(socket) => self.event_tx.send(types::Event::IncomingConnection {
+                    Ok(socket) => self.conn_tx.send(types::ConnectivityEvent::IncomingConnection {
                         peer_id: socket.1,
                         socket: socket.0,
                     }).await?,
diff --git a/p2p/src/net/mock/mod.rs b/p2p/src/net/mock/mod.rs
index e7e8accd5..a2bbb08c0 100644
--- a/p2p/src/net/mock/mod.rs
+++ b/p2p/src/net/mock/mod.rs
@@ -17,10 +17,11 @@
 #![allow(dead_code, unused_variables, unused_imports)]
 use crate::{
     error::{self, P2pError},
-    net::{Event, FloodsubTopic, NetworkService, SocketService},
+    net::{ConnectivityEvent, Event, FloodsubEvent, FloodsubTopic, NetworkService, SocketService},
     peer::Peer,
 };
 use async_trait::async_trait;
+use futures::FutureExt;
 use logging::log;
 use parity_scale_codec::{Decode, Encode};
 use std::{
@@ -48,8 +49,11 @@ pub struct MockService {
     /// TX channel for sending commands to mock backend
     cmd_tx: mpsc::Sender<types::Command>,
 
-    /// RX channel for receiving events from mock backend
-    event_rx: mpsc::Receiver<types::Event>,
+    /// RX channel for receiving connectivity events from mock backend
+    conn_rx: mpsc::Receiver<types::ConnectivityEvent>,
+
+    /// RX channel for receiving floodsub events from mock backend
+    flood_rx: mpsc::Receiver<types::FloodsubEvent>,
 }
 
 #[derive(Debug)]
@@ -76,18 +80,20 @@ impl NetworkService for MockService {
         _topics: &[FloodsubTopic],
     ) -> error::Result<Self> {
         let (cmd_tx, cmd_rx) = mpsc::channel(16);
-        let (event_tx, event_rx) = mpsc::channel(16);
+        let (conn_tx, conn_rx) = mpsc::channel(16);
+        let (flood_tx, flood_rx) = mpsc::channel(16);
         let socket = TcpListener::bind(addr).await?;
 
         tokio::spawn(async move {
-            let mut mock = backend::Backend::new(socket, cmd_rx, event_tx);
+            let mut mock = backend::Backend::new(socket, cmd_rx, conn_tx, flood_tx);
             let _ = mock.run().await;
         });
 
         Ok(Self {
             addr,
             cmd_tx,
-            event_rx,
+            conn_rx,
+            flood_rx,
         })
     }
 
@@ -116,9 +122,16 @@ impl NetworkService for MockService {
     where
         T: NetworkService<Socket = MockSocket, PeerId = SocketAddr>,
     {
-        match self.event_rx.recv().await.ok_or(P2pError::ChannelClosed)? {
-            types::Event::IncomingConnection { peer_id, socket } => {
-                Ok(Event::IncomingConnection(peer_id, MockSocket { socket }))
+        tokio::select! {
+            event = self.conn_rx.recv().fuse() => match event.ok_or(P2pError::ChannelClosed)? {
+                types::ConnectivityEvent::IncomingConnection { peer_id, socket } => Ok(Event::Connectivity(
+                    ConnectivityEvent::IncomingConnection(peer_id, MockSocket { socket }),
+                )),
+            },
+            event = self.flood_rx.recv().fuse() => match event.ok_or(P2pError::ChannelClosed)? {
+                types::FloodsubEvent::MessageReceived { peer_id: _, topic, message } => {
+                    Ok(Event::Floodsub(FloodsubEvent::MessageReceived(topic, message)))
+                }
             }
         }
     }
@@ -172,7 +185,7 @@ impl SocketService for MockSocket {
 mod tests {
     use super::*;
     use crate::net::mock::MockService;
-    use crate::net::Event;
+    use crate::net::{ConnectivityEvent, Event};
     use crate::peer::{Peer, PeerRole};
     use common::chain::config;
     use parity_scale_codec::{Decode, Encode};
@@ -264,7 +277,7 @@ mod tests {
 
         let server_res: Event<MockService> = server_res.unwrap();
         let server_res = match server_res {
-            Event::IncomingConnection(_, socket) => socket,
+            Event::Connectivity(ConnectivityEvent::IncomingConnection(_, socket)) => socket,
             _ => panic!("invalid event received, expected incoming connection"),
         };
 
@@ -309,7 +322,7 @@ mod tests {
 
         let server_res: Event<MockService> = server_res.unwrap();
         let server_res = match server_res {
-            Event::IncomingConnection(_, socket) => socket,
+            Event::Connectivity(ConnectivityEvent::IncomingConnection(_, socket)) => socket,
             _ => panic!("invalid event received, expected incoming connection"),
         };
 
diff --git a/p2p/src/net/mock/types.rs b/p2p/src/net/mock/types.rs
index b87886d50..dbd046660 100644
--- a/p2p/src/net/mock/types.rs
+++ b/p2p/src/net/mock/types.rs
@@ -14,7 +14,7 @@
 // limitations under the License.
 //
 // Author(s): A. Altonen
-use crate::error;
+use crate::{error, message, net};
 use std::net::SocketAddr;
 use tokio::{net::TcpStream, sync::oneshot};
 
@@ -25,9 +25,19 @@ pub enum Command {
     },
 }
 
-pub enum Event {
+pub enum ConnectivityEvent {
     IncomingConnection {
         peer_id: SocketAddr,
         socket: TcpStream,
     },
 }
+
+// TODO: use two events, one for txs and one for blocks?
+pub enum FloodsubEvent {
+    /// Message received from one of the floodsub topics
+    MessageReceived {
+        peer_id: SocketAddr,
+        topic: net::FloodsubTopic,
+        message: message::Message,
+    },
+}
diff --git a/p2p/src/net/mod.rs b/p2p/src/net/mod.rs
index f7c2c2bd6..44882f33c 100644
--- a/p2p/src/net/mod.rs
+++ b/p2p/src/net/mod.rs
@@ -37,8 +37,7 @@ where
     pub ip6: Vec<Arc<T::Address>>,
 }
 
-#[derive(Debug)]
-pub enum Event<T>
+pub enum ConnectivityEvent<T>
 where
     T: NetworkService,
 {
@@ -50,11 +49,22 @@ where
 
     /// One one more peers have expired
     PeerExpired(Vec<AddrInfo<T>>),
+}
 
+// TODO: separate events for blocks and transactions?
+pub enum FloodsubEvent {
     /// Message received from a Floodsub topic
     MessageReceived(FloodsubTopic, message::Message),
 }
 
+pub enum Event<T>
+where
+    T: NetworkService,
+{
+    Connectivity(ConnectivityEvent<T>),
+    Floodsub(FloodsubEvent),
+}
+
 #[derive(Debug, PartialEq, Eq, Hash)]
 pub enum FloodsubTopic {
     Transactions,


Test patch:
diff --git a/p2p/test-utils/src/lib.rs b/p2p/test-utils/src/lib.rs
index 8551b4551..394064dee 100644
--- a/p2p/test-utils/src/lib.rs
+++ b/p2p/test-utils/src/lib.rs
@@ -17,12 +17,14 @@
 #![cfg(not(loom))]
 
 use common::chain::ChainConfig;
-use libp2p::Multiaddr;
+// use libp2p::Multiaddr;
 use p2p::{
     net::{
-        libp2p::Libp2pService,
+        // libp2p::Libp2pService,
         mock::{MockService, MockSocket},
-        Event, NetworkService,
+        ConnectivityEvent,
+        Event,
+        NetworkService,
     },
     peer::*,
 };
@@ -71,7 +73,7 @@ pub async fn create_two_mock_peers(
     let (remote_res, local_res) = tokio::join!(server.poll_next(), peer_fut);
     let remote_res: Event<MockService> = remote_res.unwrap();
     let remote_res = match remote_res {
-        Event::IncomingConnection(_, socket) => socket,
+        Event::Connectivity(ConnectivityEvent::IncomingConnection(_, socket)) => socket,
         _ => panic!("invalid event received, expected incoming connection"),
     };
     let local_res = local_res.unwrap();
@@ -109,6 +111,7 @@ pub async fn create_two_mock_peers(
     (local, remote)
 }
 
+/*
 // create two libp2p peers that are connected to each other
 pub async fn create_two_libp2p_peers(
     config: Arc<ChainConfig>,
@@ -161,3 +164,4 @@ pub async fn create_two_libp2p_peers(
 
     (local, remote)
 }
+*/
diff --git a/p2p/tests/libp2p.rs b/p2p/tests/libp2p.rs
index 2ca514837..7cff37761 100644
--- a/p2p/tests/libp2p.rs
+++ b/p2p/tests/libp2p.rs
@@ -17,6 +17,7 @@
 #![cfg(not(loom))]
 extern crate test_utils;
 
+/*
 use common::{chain::config, sync::Arc};
 use libp2p::{multiaddr::Protocol, Multiaddr};
 use p2p::{
@@ -149,3 +150,4 @@ async fn test_libp2p_floodsub() {
         panic!("invalid event received for server1, expected floodsub message");
     }
 }
+*/
diff --git a/p2p/tests/p2p.rs b/p2p/tests/p2p.rs
index 3a5eb5d8d..72d42c2c8 100644
--- a/p2p/tests/p2p.rs
+++ b/p2p/tests/p2p.rs
@@ -18,11 +18,8 @@
 extern crate test_utils;
 
 use common::{chain::config, sync::Arc};
-use libp2p::Multiaddr;
-use p2p::{
-    net::{libp2p::Libp2pService, mock::MockService},
-    P2P,
-};
+// use libp2p::Multiaddr;
+use p2p::{net::mock::MockService, P2P};
 use std::net::SocketAddr;
 
 // create new p2p object with mock service
@@ -43,6 +40,7 @@ async fn test_p2p_new_mock() {
     assert!(res.is_ok());
 }
 
+/*
 // create new p2p object with libp2p service
 #[ignore]
 #[tokio::test]
@@ -61,3 +59,4 @@ async fn test_p2p_new_libp2p() {
     let res = P2P::<Libp2pService>::new(256, 32, addr, Arc::clone(&config)).await;
     assert!(res.is_ok());
 }
+*/
diff --git a/p2p/tests/peer.rs b/p2p/tests/peer.rs
index 1a0d5f2aa..8185feb1a 100644
--- a/p2p/tests/peer.rs
+++ b/p2p/tests/peer.rs
@@ -18,8 +18,8 @@
 extern crate test_utils;
 
 use common::chain::config;
-use libp2p::Multiaddr;
-use p2p::net::{self, libp2p::Libp2pService, mock::MockService, NetworkService};
+// use libp2p::Multiaddr;
+use p2p::net::{self, /*libp2p::Libp2pService, */ mock::MockService, NetworkService};
 use p2p::peer::{Peer, PeerRole};
 use std::net::SocketAddr;
 use std::sync::Arc;
@@ -39,7 +39,7 @@ async fn test_peer_new_mock() {
 
     let server_res: net::Event<MockService> = server_res.unwrap();
     let server_res = match server_res {
-        net::Event::IncomingConnection(_, socket) => socket,
+        net::Event::Connectivity(net::ConnectivityEvent::IncomingConnection(_, socket)) => socket,
         _ => panic!("invalid event received, expected incoming connection"),
     };
 
@@ -55,6 +55,7 @@ async fn test_peer_new_mock() {
     );
 }
 
+/*
 // connect two libp2p service peers together
 #[tokio::test]
 async fn test_peer_new_libp2p() {
@@ -87,3 +88,4 @@ async fn test_peer_new_libp2p() {
         rx,
     );
 }
+*/