Repository: mintlayer/mintlayer-core
Pull request #637 (2023-01-17 21:10:40)

Non-zero p2p timeouts and small cleanup
Allowing zero `outbound_connection_timeout` doesn't makes sense because a connection can never be established with such value.


Patch:
diff --git a/node/src/config_files/p2p.rs b/node/src/config_files/p2p.rs
index e23ceab01..0830b23f4 100644
--- a/node/src/config_files/p2p.rs
+++ b/node/src/config_files/p2p.rs
@@ -13,7 +13,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::{str::FromStr, time::Duration};
+use std::{num::NonZeroU64, str::FromStr, time::Duration};
 
 use serde::{Deserialize, Serialize};
 
@@ -60,7 +60,7 @@ pub struct P2pConfigFile {
     /// Duration of bans in seconds.
     pub ban_duration: Option<u64>,
     /// The outbound connection timeout value in seconds.
-    pub outbound_connection_timeout: Option<u64>,
+    pub outbound_connection_timeout: Option<NonZeroU64>,
     /// A node type.
     pub node_type: Option<NodeTypeConfigFile>,
 }
@@ -72,7 +72,10 @@ impl From<P2pConfigFile> for P2pConfig {
             added_nodes: c.added_nodes.clone().unwrap_or_default(),
             ban_threshold: c.ban_threshold.into(),
             ban_duration: c.ban_duration.map(Duration::from_secs).into(),
-            outbound_connection_timeout: c.outbound_connection_timeout.into(),
+            outbound_connection_timeout: c
+                .outbound_connection_timeout
+                .map(|t| Duration::from_secs(t.into()))
+                .into(),
             node_type: c.node_type.map(Into::into).into(),
         }
     }
diff --git a/node/src/options.rs b/node/src/options.rs
index e4b8c3f09..34fd35c8c 100644
--- a/node/src/options.rs
+++ b/node/src/options.rs
@@ -15,7 +15,7 @@
 
 //! The node command line options.
 
-use std::{ffi::OsString, fs, net::SocketAddr, path::PathBuf};
+use std::{ffi::OsString, fs, net::SocketAddr, num::NonZeroU64, path::PathBuf};
 
 use anyhow::{Context, Result};
 use clap::{Args, Parser, Subcommand};
@@ -93,7 +93,7 @@ pub struct RunOptions {
 
     /// The p2p timeout value in seconds.
     #[clap(long)]
-    pub p2p_outbound_connection_timeout: Option<u64>,
+    pub p2p_outbound_connection_timeout: Option<NonZeroU64>,
 
     /// A maximum tip age in seconds.
     ///
diff --git a/p2p/src/config.rs b/p2p/src/config.rs
index 28918b742..52246c987 100644
--- a/p2p/src/config.rs
+++ b/p2p/src/config.rs
@@ -23,7 +23,7 @@ pub const DEFAULT_BIND_PORT: u16 = 3031;
 
 make_config_setting!(BanThreshold, u32, 100);
 make_config_setting!(BanDuration, Duration, Duration::from_secs(60 * 60 * 24));
-make_config_setting!(OutboundConnectionTimeout, u64, 10);
+make_config_setting!(OutboundConnectionTimeout, Duration, Duration::from_secs(10));
 make_config_setting!(
     AnnouncementSubscriptions,
     BTreeSet<PubSubTopic>,
diff --git a/p2p/src/error.rs b/p2p/src/error.rs
index 09a786318..254974334 100644
--- a/p2p/src/error.rs
+++ b/p2p/src/error.rs
@@ -65,15 +65,6 @@ pub enum PublishError {
     TransformFailed,
 }
 
-/// PubSub errors for subscriptions
-#[derive(Error, Debug, PartialEq, Eq)]
-pub enum SubscriptionError {
-    #[error("Failed to publish subscription: {0}")]
-    FailedToPublish(PublishError),
-    #[error("Not allowed to subscribe to this topic")]
-    NotAllowed,
-}
-
 /// Errors related to establishing a connection with a remote peer
 #[derive(Error, Debug, PartialEq, Eq)]
 pub enum DialError {
@@ -116,8 +107,6 @@ pub enum P2pError {
     ProtocolError(ProtocolError),
     #[error("Failed to publish message: `{0}`")]
     PublishError(PublishError),
-    #[error("Failed to subscribe to pubsub topic: `{0}`")]
-    SubscriptionError(SubscriptionError),
     #[error("Failed to dial peer: `{0}`")]
     DialError(DialError),
     #[error("Connection to other task lost")]
@@ -179,7 +168,6 @@ impl BanScore for P2pError {
         match self {
             P2pError::ProtocolError(err) => err.ban_score(),
             P2pError::PublishError(err) => err.ban_score(),
-            P2pError::SubscriptionError(err) => err.ban_score(),
             P2pError::DialError(_) => 0,
             P2pError::ChannelClosed => 0,
             P2pError::PeerError(_) => 0,
@@ -216,15 +204,6 @@ impl BanScore for PublishError {
     }
 }
 
-impl BanScore for SubscriptionError {
-    fn ban_score(&self) -> u32 {
-        match self {
-            SubscriptionError::FailedToPublish(err) => err.ban_score(),
-            SubscriptionError::NotAllowed => 0,
-        }
-    }
-}
-
 impl BanScore for ConversionError {
     fn ban_score(&self) -> u32 {
         match self {
diff --git a/p2p/src/net/mock/backend.rs b/p2p/src/net/mock/backend.rs
index a60b7ca15..9320575ff 100644
--- a/p2p/src/net/mock/backend.rs
+++ b/p2p/src/net/mock/backend.rs
@@ -24,7 +24,6 @@
 use std::{
     collections::{BTreeSet, HashMap},
     sync::Arc,
-    time::Duration,
 };
 
 use futures::{
@@ -109,9 +108,6 @@ pub struct Backend<T: TransportSocket> {
     /// TX channel for sending syncing events
     sync_tx: mpsc::Sender<SyncingEvent>,
 
-    /// Timeout for outbound operations
-    timeout: Duration,
-
     /// Request manager for managing inbound/outbound requests and responses
     request_mgr: request_manager::RequestManager,
 
@@ -133,7 +129,6 @@ where
         cmd_rx: mpsc::Receiver<Command<T>>,
         conn_tx: mpsc::Sender<ConnectivityEvent<T>>,
         sync_tx: mpsc::Sender<SyncingEvent>,
-        timeout: Duration,
     ) -> Self {
         Self {
             transport,
@@ -143,7 +138,6 @@ where
             chain_config,
             p2p_config,
             sync_tx,
-            timeout,
             peers: HashMap::new(),
             pending: HashMap::new(),
             peer_chan: mpsc::channel(64),
@@ -597,7 +591,10 @@ where
                 // For now we always respond with success
                 response.send(Ok(())).map_err(|_| P2pError::ChannelClosed)?;
 
-                let connection_fut = timeout(self.timeout, self.transport.connect(address.clone()));
+                let connection_fut = timeout(
+                    *self.p2p_config.outbound_connection_timeout,
+                    self.transport.connect(address.clone()),
+                );
 
                 async move {
                     let connection_res = connection_fut.await.unwrap_or(Err(P2pError::DialError(
diff --git a/p2p/src/net/mock/mod.rs b/p2p/src/net/mock/mod.rs
index 47a343344..ea23b1d0e 100644
--- a/p2p/src/net/mock/mod.rs
+++ b/p2p/src/net/mock/mod.rs
@@ -128,7 +128,6 @@ impl<T: TransportSocket> NetworkingService for MockService<T> {
         let local_addresses = socket.local_addresses().expect("to have bind address available");
 
         tokio::spawn(async move {
-            let timeout = *p2p_config.outbound_connection_timeout;
             let mut backend = backend::Backend::<T>::new(
                 transport,
                 socket,
@@ -137,7 +136,6 @@ impl<T: TransportSocket> NetworkingService for MockService<T> {
                 cmd_rx,
                 conn_tx,
                 sync_tx,
-                std::time::Duration::from_secs(timeout),
             );
 
             if let Err(err) = backend.run().await {


Test patch:
diff --git a/node/tests/cli.rs b/node/tests/cli.rs
index ae85cb75d..8588a9073 100644
--- a/node/tests/cli.rs
+++ b/node/tests/cli.rs
@@ -13,7 +13,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::{net::SocketAddr, path::Path, str::FromStr};
+use std::{net::SocketAddr, num::NonZeroU64, path::Path, str::FromStr};
 
 use assert_cmd::Command;
 use directories::UserDirs;
@@ -89,7 +89,7 @@ fn read_config_override_values() {
     let p2p_addr = "address";
     let p2p_add_node = "add_node";
     let p2p_ban_threshold = 3;
-    let p2p_timeout = 10000;
+    let p2p_timeout = NonZeroU64::new(10000).unwrap();
     let http_rpc_addr = SocketAddr::from_str("127.0.0.1:5432").unwrap();
     let ws_rpc_addr = SocketAddr::from_str("127.0.0.1:5433").unwrap();
     let backend_type = StorageBackendConfigFile::InMemory;
diff --git a/p2p/src/peer_manager/tests/connections.rs b/p2p/src/peer_manager/tests/connections.rs
index 6b3faa45f..d344a691f 100644
--- a/p2p/src/peer_manager/tests/connections.rs
+++ b/p2p/src/peer_manager/tests/connections.rs
@@ -478,7 +478,7 @@ where
     pm1.peer_connectivity_handle.connect(addr2).await.expect("dial to succeed");
 
     match timeout(
-        Duration::from_secs(*pm1._p2p_config.outbound_connection_timeout),
+        *pm1._p2p_config.outbound_connection_timeout,
         pm1.peer_connectivity_handle.poll_next(),
     )
     .await
@@ -563,12 +563,7 @@ async fn connection_timeout_rpc_notified<T>(
     let (rtx, rrx) = oneshot::channel();
     tx.send(PeerManagerEvent::Connect(addr2, rtx)).unwrap();
 
-    match timeout(
-        Duration::from_secs(*p2p_config.outbound_connection_timeout),
-        rrx,
-    )
-    .await
-    {
+    match timeout(*p2p_config.outbound_connection_timeout, rrx).await {
         Ok(res) => assert!(std::matches!(
             res.unwrap(),
             Err(P2pError::DialError(DialError::ConnectionRefusedOrTimedOut))
diff --git a/p2p/src/sync/tests/mod.rs b/p2p/src/sync/tests/mod.rs
index d0d4458ff..2fc3ad9d9 100644
--- a/p2p/src/sync/tests/mod.rs
+++ b/p2p/src/sync/tests/mod.rs
@@ -74,7 +74,7 @@ where
         added_nodes: Vec::new(),
         ban_threshold: 100.into(),
         ban_duration: Default::default(),
-        outbound_connection_timeout: 10.into(),
+        outbound_connection_timeout: Default::default(),
         node_type: NodeType::Full.into(),
     });
     let (conn, sync) = T::start(