Put the disconnection and socket write timeouts to BackendTimeoutsConfig; use a smaller socket write timeout in a test

Author: ImplOfAnImpl

node-lib/src/config_files/p2p.rs

@@ -191,6 +191,8 @@ impl From<P2pConfigFile> for P2pConfig {
                     .map(|t| Duration::from_secs(t.into()))
                     .into(),
                 peer_handshake_timeout: Default::default(),
+                disconnection_timeout: Default::default(),
+                socket_write_timeout: Default::default(),
             },
             protocol_config: Default::default(),
             custom_disconnection_reason_for_banning,

p2p/src/config.rs

@@ -35,6 +35,8 @@ make_config_setting!(PingTimeout, Duration, Duration::from_secs(150));
 make_config_setting!(MaxClockDiff, Duration, Duration::from_secs(10));
 make_config_setting!(SyncStallingTimeout, Duration, Duration::from_secs(25));
 make_config_setting!(PeerHandshakeTimeout, Duration, Duration::from_secs(10));
+make_config_setting!(DisconnectionTimeout, Duration, Duration::from_secs(10));
+make_config_setting!(SoketWriteTimeout, Duration, Duration::from_secs(60));
 
 /// A node type.
 #[derive(Debug, Copy, Clone)]
@@ -144,4 +146,10 @@ pub struct BackendTimeoutsConfig {
 
     /// Timeout for initial peer handshake
     pub peer_handshake_timeout: PeerHandshakeTimeout,
+
+    /// Timeout for disconnection
+    pub disconnection_timeout: DisconnectionTimeout,
+
+    /// Timeout for the socket write call
+    pub socket_write_timeout: SoketWriteTimeout,
 }

p2p/src/net/default_backend/peer/mod.rs

@@ -17,7 +17,7 @@ mod handshake_handler;
 #[cfg(test)]
 mod tests;
 
-use std::{sync::Arc, time::Duration};
+use std::sync::Arc;
 
 use tokio::{
     sync::mpsc::{self, unbounded_channel},
@@ -232,6 +232,7 @@ where
         let (writer_cmd_sender, writer_cmd_receiver) = unbounded_channel();
         let (writer_event_sender, mut writer_event_receiver) = unbounded_channel();
         let writer_join_handle = spawn_writer(
+            Arc::clone(&p2p_config),
             peer_id,
             common_protocol_version.0,
             socket_writer,
@@ -332,16 +333,19 @@ where
             });
             match send_result {
                 Ok(()) => {
-                    let disconnect_result = tokio::time::timeout(DISCONNECTION_TIMEOUT, async {
-                        match writer_event_receiver.recv().await {
-                            Some(WriterEvent::WriterClosed(result)) => {
-                                log::debug!("Socket writer closing confirmed with result: {result:?}");
-                            },
-                            None => {
-                                log::debug!("Socket writer task already closed when waiting for disconnection");
-                            },
+                    let disconnect_result = tokio::time::timeout(
+                        *p2p_config.backend_timeouts.disconnection_timeout,
+                        async {
+                            match writer_event_receiver.recv().await {
+                                Some(WriterEvent::WriterClosed(result)) => {
+                                    log::debug!("Socket writer closing confirmed with result: {result:?}");
+                                },
+                                None => {
+                                    log::debug!("Socket writer task already closed when waiting for disconnection");
+                                },
+                            }
                         }
-                    }).await;
+                    ).await;
 
                     match disconnect_result {
                         Ok(()) => {}
@@ -377,9 +381,6 @@ where
     }
 }
 
-const DISCONNECTION_TIMEOUT: Duration = Duration::from_secs(10);
-const SOCKET_WRITE_TIMEOUT: Duration = Duration::from_secs(60);
-
 async fn maybe_send_will_disconnect<S: PeerStream>(
     reason: Option<DisconnectionReason>,
     peer_protocol_version: ProtocolVersion,
@@ -409,6 +410,7 @@ enum WriterEvent {
 }
 
 fn spawn_writer<S: PeerStream + 'static>(
+    p2p_config: Arc<P2pConfig>,
     peer_id: PeerId,
     common_protocol_version: SupportedProtocolVersion,
     socket_writer: MessageWriter<S, Message>,
@@ -417,8 +419,13 @@ fn spawn_writer<S: PeerStream + 'static>(
 ) -> JoinHandle<()> {
     tokio_spawn_in_current_tracing_span(
         async move {
-            let writer_result =
-                writer_loop(common_protocol_version, socket_writer, cmd_receiver).await;
+            let writer_result = writer_loop(
+                &p2p_config,
+                common_protocol_version,
+                socket_writer,
+                cmd_receiver,
+            )
+            .await;
 
             if let Err(_) = event_sender.send(WriterEvent::WriterClosed(writer_result)) {
                 log::debug!("Peer task already closed");
@@ -429,6 +436,7 @@ fn spawn_writer<S: PeerStream + 'static>(
 }
 
 async fn writer_loop<S: PeerStream>(
+    p2p_config: &P2pConfig,
     common_protocol_version: SupportedProtocolVersion,
     mut socket_writer: MessageWriter<S, Message>,
     mut cmd_receiver: mpsc::UnboundedReceiver<WriterCommand>,
@@ -442,11 +450,12 @@ async fn writer_loop<S: PeerStream>(
                     message.encoded_size()
                 );
 
-                tokio::time::timeout(SOCKET_WRITE_TIMEOUT, socket_writer.send(*message))
-                    .await
-                    .map_err(|_| {
-                        P2pError::NetworkingError(NetworkingError::SocketWriteTimedOut)
-                    })??;
+                tokio::time::timeout(
+                    *p2p_config.backend_timeouts.socket_write_timeout,
+                    socket_writer.send(*message),
+                )
+                .await
+                .map_err(|_| P2pError::NetworkingError(NetworkingError::SocketWriteTimedOut))??;
             }
             WriterCommand::Disconnect { reason } => {
                 log::debug!("Disconnection requested, the reason is {:?}", reason);

p2p/src/peer_manager/tests/connections.rs

@@ -1004,6 +1004,8 @@ async fn connection_timeout_rpc_notified<T>(
         backend_timeouts: BackendTimeoutsConfig {
             outbound_connection_timeout: Duration::from_secs(1).into(),
             peer_handshake_timeout: Default::default(),
+            disconnection_timeout: Default::default(),
+            socket_write_timeout: Default::default(),
         },
 
         bind_addresses: Default::default(),

p2p/src/sync/tests/network_sync.rs

@@ -426,6 +426,8 @@ async fn send_block_from_the_future_again(#[case] seed: Seed) {
             backend_timeouts: BackendTimeoutsConfig {
                 peer_handshake_timeout: Duration::from_secs(1).into(),
                 outbound_connection_timeout: Default::default(),
+                disconnection_timeout: Default::default(),
+                socket_write_timeout: Default::default(),
             },
 
             bind_addresses: Default::default(),

p2p/src/tests/connection_lockup_when_socket_not_read.rs

@@ -13,15 +13,15 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::sync::Arc;
+use std::{sync::Arc, time::Duration};
 
 use itertools::Itertools as _;
 use rstest::rstest;
 
 use chainstate::{BlockSource, ChainstateConfig, Locator};
 use common::{
     chain,
-    primitives::{Id, Idable as _},
+    primitives::{user_agent::mintlayer_core_user_agent, Id, Idable as _},
 };
 use logging::log;
 use networking::{
@@ -38,6 +38,7 @@ use test_utils::{
 };
 
 use crate::{
+    config::{BackendTimeoutsConfig, P2pConfig},
     message::{HeaderList, HeaderListRequest},
     net::{
         default_backend::types::{HandshakeMessage, Message, P2pTimestamp},
@@ -258,9 +259,38 @@ async fn timeout_when_socket_not_read(#[case] seed: Seed) {
         let channel_max_buf_size = 1000;
         let node_headers_count = 100;
 
+        let socket_write_timeout = Duration::from_secs(3);
+        let no_disconnection_after = Duration::from_secs(1);
+
         let time_getter = BasicTestTimeGetter::new();
         let chain_config = Arc::new(chain::config::create_unit_test_config());
-        let p2p_config = Arc::new(test_p2p_config());
+        let p2p_config = Arc::new(P2pConfig {
+            backend_timeouts: BackendTimeoutsConfig {
+                socket_write_timeout: socket_write_timeout.into(),
+
+                outbound_connection_timeout: Default::default(),
+                peer_handshake_timeout: Default::default(),
+                disconnection_timeout: Default::default(),
+            },
+
+            bind_addresses: Default::default(),
+            socks5_proxy: Default::default(),
+            disable_noise: Default::default(),
+            boot_nodes: Default::default(),
+            reserved_nodes: Default::default(),
+            whitelisted_addresses: Default::default(),
+            ban_config: Default::default(),
+            ping_check_period: Default::default(),
+            ping_timeout: Default::default(),
+            max_clock_diff: Default::default(),
+            node_type: Default::default(),
+            allow_discover_private_ips: Default::default(),
+            user_agent: mintlayer_core_user_agent(),
+            sync_stalling_timeout: Default::default(),
+            peer_manager_config: Default::default(),
+            protocol_config: Default::default(),
+            custom_disconnection_reason_for_banning: Default::default(),
+        });
 
         log::debug!("Generating blocks");
 
@@ -367,6 +397,15 @@ async fn timeout_when_socket_not_read(#[case] seed: Seed) {
             }
         );
 
+        // Wait for a short period of time, there should be no disconnection.
+        tokio::time::timeout(
+            no_disconnection_after,
+            test_node.peer_mgr_notification_receiver().recv(),
+        )
+        .await
+        .unwrap_err();
+
+        // But eventually the disconnection should happen when the socket write timeout expires.
         log::debug!("Expecting PeerManagerNotification::ConnectionClosed");
         let peer_mgr_notif = test_node.peer_mgr_notification_receiver().recv().await.unwrap();
         assert_matches!(