Merge pull request #2031 from mintlayer/p2p_fix_connection_stalling

Fix for potential indefinite connection stalling

Author: ImplOfAnImpl

CHANGELOG.md

@@ -73,9 +73,13 @@ The format is loosely based on [Keep a Changelog](https://keepachangelog.com/en/
       constantly attempting to re-establish a connection with a peer that has banned it.
 
 ### Fixed
-  - P2p: when a peer sends a message that can't be decoded, it will now be discouraged (which is what
-    is normally done for misbehaving peers) and the node won't try connecting to it again.\
-    Also, the peer will be sent an appropriate `WillDisconnect` message prior to disconnection.
+  - P2p:
+    - When a peer sends a message that can't be decoded, it will now be discouraged (which is what
+      is normally done for misbehaving peers) and the node won't try connecting to it again.\
+      Also, the peer will be sent an appropriate `WillDisconnect` message prior to disconnection.
+
+    - Fixed a potential indefinite stalling of a particular connection when both nodes start sending
+      large amounts of data to each other.
 
   - Wallet CLI and RPC: the commands `account-utxos` and `standalone-multisig-utxos` and their RPC
     counterparts now return correct decimal amounts for tokens with non-default number of decimals.

Cargo.lock

@@ -262,9 +262,10 @@ tracing-subscriber = { version = "0.3", features = [
   "env-filter",
   "json",
 ] }
-# Note: we use an unreleased version of Tokio, because we need a certain bugfix, see the `[patch.crates-io]`
-# section for additional details.
-tokio = { git = "https://github.com/tokio-rs/tokio", rev = "0d6c7af3e43457350bdc03a6dbcafa276fab7352", default-features = false }
+# Note: we need at least tokio 1.50, because it includes the PR https://github.com/tokio-rs/tokio/pull/7879, which fixes
+# the potential indefinite stalling of the chainstate subsystem (reproducible on a slow machine during initial sync
+# with 30+ connected peers).
+tokio = { version = "1.50", default-features = false }
 tokio-socks = "0.5"
 tokio-stream = "0.1"
 tokio-util = { version = "0.7", default-features = false }
@@ -349,11 +350,6 @@ default = ["trezor", "ledger"]
 # TODO: investigate this further.
 fontconfig-parser = { git = "https://github.com/Riey/fontconfig-parser", rev = "f7d13a779e6ee282ce75acbc00a1270c0350e0c2" }
 
-# We need this PR - https://github.com/tokio-rs/tokio/pull/7879 - to fix potential indefinite stalling of the chainstate
-# subsystem (reproducible on a slow machine during initial sync with 30+ connected peers).
-# The PR was merged after 1.49, so it should probably be part of 1.50 when it comes out.
-tokio = { git = "https://github.com/tokio-rs/tokio", rev = "0d6c7af3e43457350bdc03a6dbcafa276fab7352" }
-
 # This patch is needed because there is no release of the library and because ledger-lib depends on ledger-proto, so this is the only way to make the former find the latter.
 # Note that the revision specified here must be the same as the one used in the workspace.dependencies section
 ledger-proto = { git = "https://github.com/ImplOfAnImpl/rust-ledger.git", rev = "035789ec436d47b938e8a3d2085ffb2fbf6f0559" }

Cargo.toml

@@ -2,7 +2,6 @@
 github = [
     "mintlayer",  # allow any code from mintlayer's github
     "paritytech", # we have to use an unreleased version of parity-scale-codec at this moment
-    "tokio-rs",   # we have to use an unreleased version of tokio at this moment
 ]
 
 [licenses]

deny.toml

@@ -76,17 +76,16 @@ async fn run(options: DnsServerRunOptions) -> anyhow::Result<Never> {
         // Note: this ban config (as well as any other settings related to the peer or sync manager)
         // won't have any effect on the dns server.
         ban_config: Default::default(),
-        outbound_connection_timeout: Default::default(),
         ping_check_period: Default::default(),
         ping_timeout: Default::default(),
-        peer_handshake_timeout: Default::default(),
         max_clock_diff: Default::default(),
         node_type: NodeType::DnsServer.into(),
         allow_discover_private_ips: Default::default(),
         user_agent,
         sync_stalling_timeout: Default::default(),
         peer_manager_config: Default::default(),
         protocol_config: Default::default(),
+        backend_timeouts: Default::default(),
         custom_disconnection_reason_for_banning: Default::default(),
     });
 

dns-server/src/main.rs

@@ -20,15 +20,21 @@ pub enum NetworkingError {
     // Note: std::io::Error is neither clonable nor comparable, so we only store its "kind" here.
     #[error("IO error: {0}")]
     IoError(std::io::ErrorKind),
+
     #[error("Message codec error: {0}")]
     MessageCodecError(#[from] MessageCodecError),
+
     #[error("Noise protocol handshake error")]
     NoiseHandshakeError(String),
+
     #[error("Proxy error: {0}")]
     ProxyError(String),
 
     #[error("Channel transport error: {0}")]
     ChannelTransportError(#[from] MpscChannelTransportError),
+
+    #[error("Socket write timed out")]
+    SocketWriteTimedOut,
 }
 
 #[derive(thiserror::Error, Debug, PartialEq, Eq, Clone)]

networking/src/error.rs

@@ -79,6 +79,12 @@ impl TestTransportMaker for TestTransportChannel {
     }
 }
 
+impl TestTransportChannel {
+    pub fn make_transport_with_max_buf_size(max_buf_size: usize) -> MpscChannelTransport {
+        MpscChannelTransport::new().with_max_buf_size(max_buf_size)
+    }
+}
+
 pub struct TestTransportNoise {}
 
 impl TestTransportMaker for TestTransportNoise {

networking/src/test_helpers.rs

@@ -41,8 +41,8 @@ use crate::{
     Result,
 };
 
-// How much bytes is allowed for write (without reading on the other side).
-const MAX_BUF_SIZE: usize = 10 * 1024 * 1024;
+/// The default value for how much bytes is allowed for write (without reading on the other side).
+const DEFAULT_MAX_BUF_SIZE: usize = 10 * 1024 * 1024;
 
 struct IncomingConnection {
     from: SocketAddr,
@@ -68,6 +68,7 @@ static NEXT_IP_ADDRESS: StdAtomicU32 = StdAtomicU32::new(1);
 pub struct MpscChannelTransport {
     local_address: IpAddr,
     last_port: AtomicU16,
+    max_buf_size: usize,
 }
 
 impl MpscChannelTransport {
@@ -80,9 +81,16 @@ impl MpscChannelTransport {
         Self {
             local_address,
             last_port: 1024.into(),
+            max_buf_size: DEFAULT_MAX_BUF_SIZE,
         }
     }
 
+    /// Specified how much bytes is allowed for write (without reading on the other side).
+    pub fn with_max_buf_size(mut self, max_buf_size: usize) -> Self {
+        self.max_buf_size = max_buf_size;
+        self
+    }
+
     /// Return the next u32 value that can be used to construct a unique local address for this kind of transport.
     pub fn next_local_address_as_u32() -> u32 {
         NEXT_IP_ADDRESS.fetch_add(1, Ordering::Relaxed)
@@ -105,6 +113,7 @@ impl TransportSocket for MpscChannelTransport {
             return Ok(Self::Listener {
                 addresses,
                 receiver: None,
+                max_buf_size: self.max_buf_size,
             });
         }
 
@@ -142,6 +151,7 @@ impl TransportSocket for MpscChannelTransport {
         Ok(Self::Listener {
             addresses,
             receiver: Some(receiver),
+            max_buf_size: self.max_buf_size,
         })
     }
 
@@ -186,6 +196,7 @@ impl TransportSocket for MpscChannelTransport {
 pub struct ChannelListener {
     addresses: Vec<SocketAddr>,
     receiver: Option<UnboundedReceiver<IncomingConnection>>,
+    max_buf_size: usize,
 }
 
 #[async_trait]
@@ -209,7 +220,7 @@ impl TransportListener for ChannelListener {
 
         assert!(self.addresses.contains(&local_address));
 
-        let (server_stream, client_stream) = tokio::io::duplex(MAX_BUF_SIZE);
+        let (server_stream, client_stream) = tokio::io::duplex(self.max_buf_size);
 
         client_stream_sender.send(client_stream).map_err(|_| {
             MpscChannelTransportError::ConnectorDroppedUnexpectedly {
@@ -320,7 +331,7 @@ mod tests {
     use randomness::Rng;
     use test_utils::random::{gen_random_bytes, Seed};
 
-    use crate::transport::BufferedTranscoder;
+    use crate::transport::new_message_stream;
 
     use super::*;
 
@@ -346,11 +357,12 @@ mod tests {
         let message_size = rng.gen_range(128..1024);
 
         let message = gen_random_bytes(&mut rng, 1, message_size);
-        let mut peer_stream = BufferedTranscoder::new(peer_stream, Some(message.encoded_size()));
-        peer_stream.send(message.clone()).await.unwrap();
+        let (_, mut peer_stream_writer) =
+            new_message_stream(peer_stream, Some(message.encoded_size()));
+        peer_stream_writer.send(message.clone()).await.unwrap();
 
-        let mut server_stream =
-            BufferedTranscoder::<_, Vec<u8>>::new(server_stream, Some(message.encoded_size()));
-        assert_eq!(server_stream.recv().await.unwrap(), message);
+        let (mut server_stream_reader, _) =
+            new_message_stream::<_, Vec<u8>>(server_stream, Some(message.encoded_size()));
+        assert_eq!(server_stream_reader.recv().await.unwrap(), message);
     }
 }

networking/src/transport/impls/channel.rs

@@ -36,7 +36,7 @@ use crate::{
     test_helpers::{TestTransportChannel, TestTransportMaker, TestTransportTcp},
     transport::{
         impls::stream_adapter::wrapped_transport::wrapped_listener::MAX_CONCURRENT_HANDSHAKES,
-        BufferedTranscoder, ChannelListener, IdentityStreamAdapter, MpscChannelTransport,
+        new_message_stream, ChannelListener, IdentityStreamAdapter, MpscChannelTransport,
         NoiseEncryptionAdapter, NoiseEncryptionAdapterMaker, PeerStream, TcpTransportSocket,
         TransportListener, TransportSocket,
     },
@@ -252,13 +252,13 @@ async fn send_2_reqs(#[case] seed: Seed) {
     let message_1 = gen_random_bytes(&mut rng, 0, 1000);
     let message_2 = gen_random_bytes(&mut rng, 0, 1000);
 
-    let mut peer_stream = BufferedTranscoder::<_, Vec<u8>>::new(peer_stream, None);
-    peer_stream.send(message_1.clone()).await.unwrap();
-    peer_stream.send(message_2.clone()).await.unwrap();
+    let (_, mut peer_stream_writer) = new_message_stream::<_, Vec<u8>>(peer_stream, None);
+    peer_stream_writer.send(message_1.clone()).await.unwrap();
+    peer_stream_writer.send(message_2.clone()).await.unwrap();
 
-    let mut server_stream = BufferedTranscoder::<_, Vec<u8>>::new(server_stream, None);
-    assert_eq!(server_stream.recv().await.unwrap(), message_1);
-    assert_eq!(server_stream.recv().await.unwrap(), message_2);
+    let (mut server_stream_reader, _) = new_message_stream::<_, Vec<u8>>(server_stream, None);
+    assert_eq!(server_stream_reader.recv().await.unwrap(), message_1);
+    assert_eq!(server_stream_reader.recv().await.unwrap(), message_2);
 }
 
 #[tracing::instrument]

networking/src/transport/impls/stream_adapter/wrapped_transport/tests.rs

@@ -156,7 +156,7 @@ mod tests {
 
     use crate::{
         test_helpers::{TestTransportMaker, TestTransportTcp},
-        transport::BufferedTranscoder,
+        transport::new_message_stream,
     };
 
     use super::*;
@@ -178,12 +178,13 @@ mod tests {
         let peer_stream = peer_res.unwrap();
 
         let message = gen_random_bytes(&mut rng, 0, 1000);
-        let mut peer_stream = BufferedTranscoder::new(peer_stream, Some(message.encoded_size()));
-        peer_stream.send(message.clone()).await.unwrap();
+        let (_, mut peer_stream_writer) =
+            new_message_stream(peer_stream, Some(message.encoded_size()));
+        peer_stream_writer.send(message.clone()).await.unwrap();
 
-        let mut server_stream =
-            BufferedTranscoder::<_, Vec<u8>>::new(server_stream, Some(message.encoded_size()));
-        assert_eq!(server_stream.recv().await.unwrap(), message);
+        let (mut server_stream_reader, _) =
+            new_message_stream::<_, Vec<u8>>(server_stream, Some(message.encoded_size()));
+        assert_eq!(server_stream_reader.recv().await.unwrap(), message);
     }
 
     #[tracing::instrument(skip(seed))]
@@ -205,12 +206,12 @@ mod tests {
         let message_1 = gen_random_bytes(&mut rng, 0, 1000);
         let message_2 = gen_random_bytes(&mut rng, 0, 1000);
 
-        let mut peer_stream = BufferedTranscoder::new(peer_stream, None);
-        peer_stream.send(message_1.clone()).await.unwrap();
-        peer_stream.send(message_2.clone()).await.unwrap();
+        let (_, mut peer_stream_writer) = new_message_stream(peer_stream, None);
+        peer_stream_writer.send(message_1.clone()).await.unwrap();
+        peer_stream_writer.send(message_2.clone()).await.unwrap();
 
-        let mut server_stream = BufferedTranscoder::<_, Vec<u8>>::new(server_stream, None);
-        assert_eq!(server_stream.recv().await.unwrap(), message_1);
-        assert_eq!(server_stream.recv().await.unwrap(), message_2);
+        let (mut server_stream_reader, _) = new_message_stream::<_, Vec<u8>>(server_stream, None);
+        assert_eq!(server_stream_reader.recv().await.unwrap(), message_1);
+        assert_eq!(server_stream_reader.recv().await.unwrap(), message_2);
     }
 }