Merge pull request #308 from mintlayer/fix/p2p_suggestions

Fix: p2p suggestions

Author: TheQuantumPhysicist

common/src/lib.rs

@@ -25,12 +25,3 @@ mod fixed_hash;
 
 pub use concurrency_impl::*;
 pub use uint::{Uint128, Uint256};
-
-#[cfg(test)]
-mod tests {
-    #[test]
-    #[allow(clippy::eq_op)]
-    fn it_works() {
-        assert_eq!(2 + 2, 4);
-    }
-}

p2p/src/error.rs

@@ -23,6 +23,8 @@ use libp2p::{
 };
 use thiserror::Error;
 
+/// Errors related to invalid data/peer information that results in connection getting closed
+/// and the peer getting banned.
 #[derive(Error, Debug, PartialEq, Eq)]
 pub enum ProtocolError {
     #[error("Peer is in different network. Our network {0:?}, their network {1:?}")]
@@ -31,16 +33,17 @@ pub enum ProtocolError {
     InvalidVersion(SemVer, SemVer),
     #[error("Peer sent an invalid message")]
     InvalidMessage,
-    #[error("Peer is incompatible")] // TODO: remove?
+    #[error("Peer is incompatible")]
     Incompatible,
     #[error("Peer is unresponsive")]
     Unresponsive,
-    #[error("Peer uses an invalid protocol")] // TODO: remove?
+    #[error("Peer uses an invalid protocol")]
     InvalidProtocol,
     #[error("Peer state is invalid for this operation. State is {0} but should be {1}")]
     InvalidState(&'static str, &'static str),
 }
 
+/// Peer state errors (Errors either for an individual peer or for the [`PeerManager`])
 #[derive(Error, Debug, PartialEq, Eq)]
 pub enum PeerError {
     #[error("Peer disconnected")]
@@ -61,6 +64,7 @@ pub enum PeerError {
     Pending(String),
 }
 
+/// PubSub errors for announcements
 #[derive(Error, Debug, PartialEq, Eq)]
 pub enum PublishError {
     #[error("Message has already been published")]
@@ -75,6 +79,7 @@ pub enum PublishError {
     TransformFailed,
 }
 
+/// PubSub errors for subscriptions
 #[derive(Error, Debug, PartialEq, Eq)]
 pub enum SubscriptionError {
     #[error("Failed to publish subscription: {0}")]
@@ -83,6 +88,7 @@ pub enum SubscriptionError {
     NotAllowed,
 }
 
+/// Errors related to establishing a connection with a remote peer
 #[derive(Error, Debug, PartialEq, Eq)]
 pub enum DialError {
     #[error("Peer is banned")]
@@ -107,6 +113,7 @@ pub enum DialError {
     Transport,
 }
 
+/// Low-level connection errors caused by libp2p
 #[derive(Error, Debug, PartialEq, Eq)]
 pub enum ConnectionError {
     #[error("Timeout")]
@@ -117,6 +124,7 @@ pub enum ConnectionError {
     Upgrade,
 }
 
+/// Conversion errors
 #[derive(Error, Debug, PartialEq, Eq)]
 pub enum ConversionError {
     #[error("Invalid peer ID: `{0}`")]

p2p/src/event.rs

@@ -48,10 +48,7 @@ pub enum SyncEvent {
 }
 
 #[derive(Debug)]
-pub enum SyncControlEvent<T>
-where
-    T: NetworkingService,
-{
+pub enum SyncControlEvent<T: NetworkingService> {
     /// Peer connected
     Connected(T::PeerId),
 

p2p/src/lib.rs

@@ -16,7 +16,7 @@
 use crate::{
     config::P2pConfig,
     error::{ConversionError, P2pError},
-    net::{ConnectivityService, NetworkingService, PubSubService, SyncingCodecService},
+    net::{ConnectivityService, NetworkingService, PubSubService, SyncingMessagingService},
 };
 use chainstate::chainstate_interface;
 use common::chain::ChainConfig;
@@ -30,9 +30,9 @@ pub mod error;
 pub mod event;
 pub mod message;
 pub mod net;
+pub mod peer_manager;
 pub mod pubsub;
 pub mod rpc;
-pub mod swarm;
 pub mod sync;
 
 /// Result type with P2P errors
@@ -45,7 +45,6 @@ pub struct P2pInterface<T: NetworkingService> {
     p2p: P2P<T>,
 }
 
-// TODO: reduce code duplication?
 impl<T> P2pInterface<T>
 where
     T: NetworkingService,
@@ -143,7 +142,7 @@ impl<T> P2P<T>
 where
     T: 'static + NetworkingService,
     T::ConnectivityHandle: ConnectivityService<T>,
-    T::SyncingCodecHandle: SyncingCodecService<T>,
+    T::SyncingMessagingHandle: SyncingMessagingService<T>,
     T::PubSubHandle: PubSubService<T>,
 {
     /// Start the P2P subsystem
@@ -170,63 +169,77 @@ where
         )
         .await?;
 
-        // TODO: think about these channel sizes
+        // P2P creates its components (such as PeerManager, sync, pubsub, etc) and makes
+        // communications with them in two possible ways:
+        //
+        // 1. Fire-and-forget
+        // 2. Request and wait for response
+        //
+        // The difference between these types is that enums that contain the events *can* have
+        // a `oneshot::channel` object that must be used to send the response.
         let (tx_swarm, rx_swarm) = mpsc::channel(CHANNEL_SIZE);
         let (tx_p2p_sync, rx_p2p_sync) = mpsc::channel(CHANNEL_SIZE);
         let (_tx_sync, _rx_sync) = mpsc::channel(CHANNEL_SIZE);
         let (tx_pubsub, rx_pubsub) = mpsc::channel(CHANNEL_SIZE);
 
-        let swarm_config = Arc::clone(&chain_config);
-        tokio::spawn(async move {
-            if let Err(e) = swarm::PeerManager::<T>::new(
-                swarm_config,
-                Arc::clone(&p2p_config),
-                conn,
-                rx_swarm,
-                tx_p2p_sync,
-            )
-            .run()
-            .await
-            {
-                log::error!("PeerManager failed: {:?}", e);
-            }
-        });
+        {
+            let chain_config = Arc::clone(&chain_config);
+            tokio::spawn(async move {
+                if let Err(err) = peer_manager::PeerManager::<T>::new(
+                    chain_config,
+                    Arc::clone(&p2p_config),
+                    conn,
+                    rx_swarm,
+                    tx_p2p_sync,
+                )
+                .run()
+                .await
+                {
+                    log::error!("PeerManager failed: {err}");
+                }
+            });
+        }
+        {
+            let consensus_handle = consensus_handle.clone();
+            let tx_swarm = tx_swarm.clone();
+            let chain_config = Arc::clone(&chain_config);
 
-        let sync_handle = consensus_handle.clone();
-        let tx_swarm_sync = tx_swarm.clone();
-        let sync_config = Arc::clone(&chain_config);
-        tokio::spawn(async move {
-            if let Err(e) = sync::SyncManager::<T>::new(
-                sync_config,
-                sync,
-                sync_handle,
-                rx_p2p_sync,
-                tx_swarm_sync,
-                tx_pubsub,
-            )
-            .run()
-            .await
-            {
-                log::error!("SyncManager failed: {:?}", e);
-            }
-        });
+            tokio::spawn(async move {
+                if let Err(err) = sync::BlockSyncManager::<T>::new(
+                    chain_config,
+                    sync,
+                    consensus_handle,
+                    rx_p2p_sync,
+                    tx_swarm,
+                    tx_pubsub,
+                )
+                .run()
+                .await
+                {
+                    log::error!("SyncManager failed: {err}");
+                }
+            });
+        }
 
-        let tx_swarm_pubsub = tx_swarm.clone();
-        tokio::spawn(async move {
-            if let Err(e) = pubsub::PubSubMessageHandler::<T>::new(
-                chain_config,
-                pubsub,
-                consensus_handle,
-                tx_swarm_pubsub,
-                rx_pubsub,
-                &[net::types::PubSubTopic::Blocks],
-            )
-            .run()
-            .await
-            {
-                log::error!("PubSubMessageHandler failed: {:?}", e);
-            }
-        });
+        {
+            let tx_swarm = tx_swarm.clone();
+
+            tokio::spawn(async move {
+                if let Err(err) = pubsub::PubSubMessageHandler::<T>::new(
+                    chain_config,
+                    pubsub,
+                    consensus_handle,
+                    tx_swarm,
+                    rx_pubsub,
+                    &[net::types::PubSubTopic::Blocks],
+                )
+                .run()
+                .await
+                {
+                    log::error!("PubSubMessageHandler failed: {err}");
+                }
+            });
+        }
 
         Ok(Self { tx_swarm, _tx_sync })
     }
@@ -244,7 +257,7 @@ pub async fn make_p2p<T>(
 where
     T: NetworkingService + 'static,
     T::ConnectivityHandle: ConnectivityService<T>,
-    T::SyncingCodecHandle: SyncingCodecService<T>,
+    T::SyncingMessagingHandle: SyncingMessagingService<T>,
     T::PubSubHandle: PubSubService<T>,
     <T as NetworkingService>::Address: FromStr,
     <<T as NetworkingService>::Address as FromStr>::Err: Debug,

p2p/src/message.rs

@@ -21,13 +21,13 @@ use common::{
 use serialization::{Decode, Encode};
 
 #[derive(Debug, Encode, Decode, Clone, PartialEq, Eq)]
-pub struct HeaderRequest {
+pub struct HeaderListRequest {
     locator: Locator,
 }
 
-impl HeaderRequest {
+impl HeaderListRequest {
     pub fn new(locator: Locator) -> Self {
-        HeaderRequest { locator }
+        HeaderListRequest { locator }
     }
 
     pub fn locator(&self) -> &Locator {
@@ -40,11 +40,11 @@ impl HeaderRequest {
 }
 
 #[derive(Debug, Encode, Decode, Clone, PartialEq, Eq)]
-pub struct BlockRequest {
+pub struct BlockListRequest {
     block_ids: Vec<Id<Block>>,
 }
 
-impl BlockRequest {
+impl BlockListRequest {
     pub fn new(block_ids: Vec<Id<Block>>) -> Self {
         Self { block_ids }
     }
@@ -61,17 +61,17 @@ impl BlockRequest {
 #[derive(Debug, Encode, Decode, Clone, PartialEq, Eq)]
 pub enum Request {
     #[codec(index = 0)]
-    HeaderRequest(HeaderRequest),
+    HeaderListRequest(HeaderListRequest),
     #[codec(index = 1)]
-    BlockRequest(BlockRequest),
+    BlockListRequest(BlockListRequest),
 }
 
 #[derive(Debug, Encode, Decode, Clone, PartialEq, Eq)]
-pub struct HeaderResponse {
+pub struct HeaderListResponse {
     headers: Vec<BlockHeader>,
 }
 
-impl HeaderResponse {
+impl HeaderListResponse {
     pub fn new(headers: Vec<BlockHeader>) -> Self {
         Self { headers }
     }
@@ -86,11 +86,11 @@ impl HeaderResponse {
 }
 
 #[derive(Debug, Encode, Decode, Clone, PartialEq, Eq)]
-pub struct BlockResponse {
+pub struct BlockListResponse {
     blocks: Vec<Block>,
 }
 
-impl BlockResponse {
+impl BlockListResponse {
     pub fn new(blocks: Vec<Block>) -> Self {
         Self { blocks }
     }
@@ -107,9 +107,9 @@ impl BlockResponse {
 #[derive(Debug, Encode, Decode, Clone, PartialEq, Eq)]
 pub enum Response {
     #[codec(index = 0)]
-    HeaderResponse(HeaderResponse),
+    HeaderListResponse(HeaderListResponse),
     #[codec(index = 1)]
-    BlockResponse(BlockResponse),
+    BlockListResponse(BlockListResponse),
 }
 
 #[derive(Debug, Encode, Decode, Clone, PartialEq, Eq)]