net/connector: Retry transport mixing routes

src/net/connector.rs

@@ -17,6 +17,8 @@
  */
 
 use std::{
+    future::Future,
+    io,
     sync::{atomic::Ordering, Arc},
     time::Duration,
 };
@@ -37,6 +39,70 @@ use super::{
 };
 use crate::{net::hosts::HostContainer, system::CondVar, util::logger::verbose, Error, Result};
 
+type DialRoute = (Url, bool, Duration);
+type DialFailures = Vec<(Url, io::Error)>;
+
+#[derive(Debug)]
+enum DialRoutesError {
+    Stopped(Url),
+    Failed(DialFailures),
+}
+
+fn build_dial_routes(endpoints: Vec<(Url, bool)>, settings: &Settings) -> Vec<DialRoute> {
+    endpoints
+        .into_iter()
+        .map(|(endpoint, mixed_transport)| {
+            let timeout = Duration::from_secs(settings.outbound_connect_timeout(endpoint.scheme()));
+            (endpoint, mixed_transport, timeout)
+        })
+        .collect()
+}
+
+async fn try_dial_routes<T, F, Fut>(
+    routes: Vec<DialRoute>,
+    stop_signal: &CondVar,
+    mut dial: F,
+) -> std::result::Result<(Url, bool, T), DialRoutesError>
+where
+    F: FnMut(Url, Duration) -> Fut,
+    Fut: Future<Output = io::Result<T>>,
+{
+    let mut failures = vec![];
+
+    for (endpoint, mixed_transport, timeout) in routes {
+        let stop_fut = stop_signal.wait();
+        let dial_fut = dial(endpoint.clone(), timeout);
+
+        pin_mut!(stop_fut);
+        pin_mut!(dial_fut);
+
+        match select(dial_fut, stop_fut).await {
+            Either::Left((Ok(stream), _)) => return Ok((endpoint, mixed_transport, stream)),
+            Either::Left((Err(err), _)) => failures.push((endpoint, err)),
+            Either::Right((_, _)) => return Err(DialRoutesError::Stopped(endpoint)),
+        }
+    }
+
+    Err(DialRoutesError::Failed(failures))
+}
+
+fn sanitized_url(url: &Url) -> String {
+    let mut sanitized = url.clone();
+    let _ = sanitized.set_password(None);
+    let _ = sanitized.set_username("");
+    sanitized.set_query(None);
+    sanitized.set_fragment(None);
+    sanitized.to_string()
+}
+
+fn summarize_failures(failures: &DialFailures) -> String {
+    failures
+        .iter()
+        .map(|(endpoint, err)| format!("{} ({:?})", sanitized_url(endpoint), err.kind()))
+        .collect::<Vec<_>>()
+        .join(", ")
+}
+
 /// Create outbound socket connections
 pub struct Connector {
     /// P2P settings
@@ -57,6 +123,7 @@ impl Connector {
     pub async fn connect(&self, url: &Url) -> Result<(Url, ChannelPtr)> {
         let hosts = self.session.upgrade().unwrap().p2p().hosts();
         if hosts.container.contains(HostColor::Black, url) || hosts.block_all_ports(url) {
+            let url = sanitized_url(url);
             verbose!(target: "net::connector::connect", "Peer {url} is blacklisted");
             return Err(Error::ConnectFailed(format!("[{url}]: Peer is blacklisted")));
         }
@@ -65,38 +132,32 @@ impl Connector {
         let datastore = settings.p2p_datastore.clone();
         let i2p_socks5_proxy = settings.i2p_socks5_proxy.clone();
 
-        let Some((endpoint, mixed_transport)) = HostContainer::resolve_dial_endpoints(
+        let endpoints = HostContainer::resolve_dial_endpoints(
             url,
             &settings.active_profiles,
             &settings.mixed_profiles,
             &settings.tor_socks5_proxy,
             &settings.nym_socks5_proxy,
-        )
-        .first()
-        .cloned() else {
+        );
+        if endpoints.is_empty() {
             return Err(Error::UnsupportedTransport(url.scheme().to_string()))
-        };
+        }
 
-        let outbound_connect_timeout = settings.outbound_connect_timeout(endpoint.scheme());
+        let routes = build_dial_routes(endpoints, &settings);
         drop(settings);
 
-        let dialer =
-            match Dialer::new(endpoint.clone(), datastore, Some(i2p_socks5_proxy), true).await {
-                Ok(dialer) => dialer,
-                Err(err) => return Err(Error::ConnectFailed(format!("[{endpoint}]: {err}"))),
-            };
-        let timeout = Duration::from_secs(outbound_connect_timeout);
-
-        let stop_fut = async {
-            self.stop_signal.wait().await;
-        };
-        let dial_fut = async { dialer.dial(Some(timeout)).await };
-
-        pin_mut!(stop_fut);
-        pin_mut!(dial_fut);
+        let result = try_dial_routes(routes, &self.stop_signal, |endpoint, timeout| {
+            let datastore = datastore.clone();
+            let i2p_socks5_proxy = i2p_socks5_proxy.clone();
+            async move {
+                let dialer = Dialer::new(endpoint, datastore, Some(i2p_socks5_proxy), true).await?;
+                dialer.dial(Some(timeout)).await
+            }
+        })
+        .await;
 
-        match select(dial_fut, stop_fut).await {
-            Either::Left((Ok(ptstream), _)) => {
+        match result {
+            Ok((endpoint, mixed_transport, ptstream)) => {
                 let channel = Channel::new(
                     ptstream,
                     Some(endpoint.clone()),
@@ -108,21 +169,20 @@ impl Connector {
                 Ok((endpoint, channel))
             }
 
-            Either::Left((Err(e), _)) => {
+            Err(DialRoutesError::Failed(failures)) => {
                 // If we get ENETUNREACH, we don't have IPv6 connectivity so note it down.
-                if e.raw_os_error() == Some(libc::ENETUNREACH) {
-                    self.session
-                        .upgrade()
-                        .unwrap()
-                        .p2p()
-                        .hosts()
-                        .ipv6_available
-                        .store(false, Ordering::SeqCst);
+                if failures.iter().any(|(_, err)| err.raw_os_error() == Some(libc::ENETUNREACH)) {
+                    hosts.ipv6_available.store(false, Ordering::SeqCst);
                 }
-                Err(Error::ConnectFailed(format!("[{endpoint}]: {e}")))
+                Err(Error::ConnectFailed(format!(
+                    "All connection routes failed: {}",
+                    summarize_failures(&failures)
+                )))
             }
 
-            Either::Right((_, _)) => Err(Error::ConnectorStopped(format!("[{endpoint}]"))),
+            Err(DialRoutesError::Stopped(endpoint)) => {
+                Err(Error::ConnectorStopped(format!("[{}]", sanitized_url(&endpoint))))
+            }
         }
     }
 
@@ -130,3 +190,131 @@ impl Connector {
         self.stop_signal.notify()
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use std::sync::{Arc, Mutex};
+
+    use crate::net::settings::NetworkProfile;
+
+    use super::*;
+
+    fn route(url: &str, timeout: u64) -> DialRoute {
+        (Url::parse(url).unwrap(), true, Duration::from_secs(timeout))
+    }
+
+    #[test]
+    fn test_dial_routes_use_endpoint_profile_timeouts() {
+        let mut settings = Settings::default();
+        settings.profiles.insert(
+            "tor".to_string(),
+            NetworkProfile { outbound_connect_timeout: 3, ..Default::default() },
+        );
+        settings.profiles.insert(
+            "nym".to_string(),
+            NetworkProfile { outbound_connect_timeout: 7, ..Default::default() },
+        );
+        let endpoints = vec![
+            (Url::parse("tor://peer.example:28880").unwrap(), true),
+            (Url::parse("nym://peer.example:28880").unwrap(), true),
+        ];
+
+        let routes = build_dial_routes(endpoints, &settings);
+
+        assert_eq!(routes[0].2, Duration::from_secs(3));
+        assert_eq!(routes[1].2, Duration::from_secs(7));
+    }
+
+    #[test]
+    fn test_dial_routes_falls_back_after_failure() {
+        smol::block_on(async {
+            let attempts = Arc::new(Mutex::new(vec![]));
+            let recorded = attempts.clone();
+            let routes = vec![
+                route("socks5://proxy-one.example:9050/peer.example:28880", 3),
+                route("socks5://proxy-two.example:9050/peer.example:28880", 7),
+            ];
+
+            let result = try_dial_routes(routes, &CondVar::new(), move |endpoint, timeout| {
+                let recorded = recorded.clone();
+                async move {
+                    recorded.lock().unwrap().push((endpoint.clone(), timeout));
+                    if endpoint.host_str() == Some("proxy-one.example") {
+                        return Err(io::Error::from(io::ErrorKind::ConnectionRefused))
+                    }
+                    Ok(42)
+                }
+            })
+            .await
+            .unwrap();
+
+            assert_eq!(result.0.host_str(), Some("proxy-two.example"));
+            assert_eq!(result.2, 42);
+            assert_eq!(
+                attempts
+                    .lock()
+                    .unwrap()
+                    .iter()
+                    .map(|(endpoint, timeout)| (endpoint.host_str().unwrap().to_string(), *timeout))
+                    .collect::<Vec<_>>(),
+                [
+                    ("proxy-one.example".to_string(), Duration::from_secs(3)),
+                    ("proxy-two.example".to_string(), Duration::from_secs(7)),
+                ]
+            );
+        });
+    }
+
+    #[test]
+    fn test_dial_routes_reports_all_failures_without_credentials() {
+        smol::block_on(async {
+            let routes = vec![
+                route("socks5://alice:secret@proxy-one.example:9050/peer.example:28880", 3),
+                route("socks5://bob:hidden@proxy-two.example:9050/peer.example:28880", 7),
+            ];
+
+            let Err(DialRoutesError::Failed(failures)) =
+                try_dial_routes(routes, &CondVar::new(), |_, _| async {
+                    Err::<(), _>(io::Error::from(io::ErrorKind::ConnectionRefused))
+                })
+                .await
+            else {
+                panic!("all routes should fail")
+            };
+            let summary = summarize_failures(&failures);
+
+            assert!(summary.contains("proxy-one.example"));
+            assert!(summary.contains("proxy-two.example"));
+            assert!(!summary.contains("alice"));
+            assert!(!summary.contains("secret"));
+            assert!(!summary.contains("bob"));
+            assert!(!summary.contains("hidden"));
+        });
+    }
+
+    #[test]
+    fn test_dial_routes_stops_without_trying_fallback() {
+        smol::block_on(async {
+            let stop_signal = CondVar::new();
+            stop_signal.notify();
+            let attempts = Arc::new(Mutex::new(vec![]));
+            let recorded = attempts.clone();
+            let routes = vec![
+                route("socks5://proxy-one.example:9050/peer.example:28880", 3),
+                route("socks5://proxy-two.example:9050/peer.example:28880", 7),
+            ];
+
+            let result = try_dial_routes(routes, &stop_signal, move |endpoint, _| {
+                let recorded = recorded.clone();
+                async move {
+                    recorded.lock().unwrap().push(endpoint);
+                    futures::future::pending::<io::Result<()>>().await
+                }
+            })
+            .await;
+
+            assert!(matches!(result, Err(DialRoutesError::Stopped(_))));
+            assert_eq!(attempts.lock().unwrap().len(), 1);
+        });
+    }
+}