fix(net): handle outgoing connection error correctly (#798)

Cargo.lock

@@ -1278,9 +1278,9 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "c9720bba047d567ffc8a3cba48bf19126600e249ab7f128e9233e6376976a116"
 dependencies = [
  "heck",
- "proc-macro2",
- "quote",
- "syn",
+ "proc-macro2 1.0.49",
+ "quote 1.0.23",
+ "syn 1.0.107",
 ]
 
 [[package]]

crates/net/network/src/error.rs

@@ -5,7 +5,7 @@ use reth_eth_wire::{
     errors::{EthHandshakeError, EthStreamError, P2PHandshakeError, P2PStreamError},
     DisconnectReason,
 };
-use std::{fmt, io::ErrorKind};
+use std::{fmt, io, io::ErrorKind};
 
 /// All error variants for the network
 #[derive(Debug, thiserror::Error)]
@@ -114,15 +114,7 @@ impl SessionError for EthStreamError {
 
     fn should_backoff(&self) -> Option<BackoffKind> {
         if let Some(err) = self.as_io() {
-            return match err.kind() {
-                // these usually happen when the remote instantly drops the connection, for example
-                // if the previous connection isn't properly cleaned up yet and the peer is temp.
-                // banned.
-                ErrorKind::ConnectionRefused |
-                ErrorKind::ConnectionReset |
-                ErrorKind::BrokenPipe => Some(BackoffKind::Low),
-                _ => Some(BackoffKind::Medium),
-            }
+            return err.should_backoff()
         }
 
         if let Some(err) = self.as_disconnected() {
@@ -183,6 +175,28 @@ impl SessionError for PendingSessionHandshakeError {
     }
 }
 
+impl SessionError for io::Error {
+    fn merits_discovery_ban(&self) -> bool {
+        false
+    }
+
+    fn is_fatal_protocol_error(&self) -> bool {
+        false
+    }
+
+    fn should_backoff(&self) -> Option<BackoffKind> {
+        match self.kind() {
+            // these usually happen when the remote instantly drops the connection, for example
+            // if the previous connection isn't properly cleaned up yet and the peer is temp.
+            // banned.
+            ErrorKind::ConnectionRefused | ErrorKind::ConnectionReset | ErrorKind::BrokenPipe => {
+                Some(BackoffKind::Low)
+            }
+            _ => Some(BackoffKind::Medium),
+        }
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

crates/net/network/src/manager.rs

@@ -705,10 +705,16 @@ where
                         ?error,
                         "Outgoing connection error"
                     );
-                    this.swarm
-                        .state_mut()
-                        .peers_mut()
-                        .apply_reputation_change(&peer_id, ReputationChangeKind::FailedToConnect);
+
+                    this.swarm.state_mut().peers_mut().on_outgoing_connection_failure(
+                        &remote_addr,
+                        &peer_id,
+                        &error,
+                    );
+
+                    this.metrics
+                        .outgoing_connections
+                        .set(this.swarm.state().peers().num_outbound_connections() as f64);
                 }
                 SwarmEvent::BadMessage { peer_id } => {
                     this.swarm

crates/net/network/src/peers/manager.rs

@@ -13,6 +13,7 @@ use reth_primitives::{ForkId, NodeRecord, PeerId};
 use std::{
     collections::{hash_map::Entry, HashMap, HashSet, VecDeque},
     fmt::Display,
+    io,
     net::{IpAddr, SocketAddr},
     task::{Context, Poll},
     time::Duration,
@@ -338,6 +339,16 @@ impl PeersManager {
         self.on_connection_failure(remote_addr, peer_id, err, ReputationChangeKind::Dropped)
     }
 
+    /// Called when an attempt to create a pending session failed while setting up a tcp connection.
+    pub(crate) fn on_outgoing_connection_failure(
+        &mut self,
+        remote_addr: &SocketAddr,
+        peer_id: &PeerId,
+        err: &io::Error,
+    ) {
+        self.on_connection_failure(remote_addr, peer_id, err, ReputationChangeKind::FailedToConnect)
+    }
+
     fn on_connection_failure(
         &mut self,
         remote_addr: &SocketAddr,
@@ -1036,6 +1047,7 @@ mod test {
     use std::{
         collections::HashSet,
         future::{poll_fn, Future},
+        io,
         net::{IpAddr, Ipv4Addr, SocketAddr},
         pin::Pin,
         task::{Context, Poll},
@@ -1494,6 +1506,41 @@ mod test {
         assert!(peers.peers.get(&peer).is_none());
     }
 
+    #[tokio::test]
+    async fn test_outgoing_connection_error() {
+        let peer = PeerId::random();
+        let socket_addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 1, 2)), 8008);
+        let mut peers = PeersManager::default();
+        peers.add_peer(peer, socket_addr);
+
+        match event!(peers) {
+            PeerAction::PeerAdded(peer_id) => {
+                assert_eq!(peer_id, peer);
+            }
+            _ => unreachable!(),
+        }
+        match event!(peers) {
+            PeerAction::Connect { peer_id, remote_addr } => {
+                assert_eq!(peer_id, peer);
+                assert_eq!(remote_addr, socket_addr);
+            }
+            _ => unreachable!(),
+        }
+
+        let p = peers.peers.get(&peer).unwrap();
+        assert_eq!(p.state, PeerConnectionState::Out);
+
+        assert_eq!(peers.num_outbound_connections(), 1);
+
+        peers.on_outgoing_connection_failure(
+            &socket_addr,
+            &peer,
+            &io::Error::new(io::ErrorKind::ConnectionRefused, ""),
+        );
+
+        assert_eq!(peers.num_outbound_connections(), 0);
+    }
+
     #[tokio::test]
     async fn test_discovery_ban_list() {
         let ip = IpAddr::V4(Ipv4Addr::new(127, 0, 1, 2));