Implement Happy Eyeballs (RFC 8305) for dual-stack IPv6/IPv4 connections

When DNS returns both IPv6 and IPv4 addresses, connect_direct now tries
IPv6 first and starts an IPv4 attempt after 250ms if IPv6 hasn't
connected yet, using whichever succeeds first. This prevents broken
IPv6 connectivity from blocking connections entirely, matching the
behavior of web browsers.

This fixes an error reported to ngit which uses rust-nostr which uses
this library.

Author: DanConwayDev

Cargo.toml

@@ -23,7 +23,7 @@ url = { version = "2.5", default-features = false }
 
 
 [target.'cfg(not(target_arch = "wasm32"))'.dependencies]
-tokio = { version = "1", features = ["net", "time"] }
+tokio = { version = "1", features = ["net", "time", "macros", "rt"] }
 tokio-rustls = { version = "0.26", default-features = false, features = ["ring", "tls12"] } # Required to enable the necessary features for tokio-tungstenite
 tokio-socks = { version = "0.5", optional = true }
 tokio-tungstenite = { version = "0.28", features = ["rustls-tls-webpki-roots"] }

src/native/error.rs

@@ -13,6 +13,8 @@ use super::tor;
 pub enum Error {
     /// Ws error
     Ws(WsError),
+    /// I/O error
+    Io(std::io::Error),
     /// Socks error
     #[cfg(feature = "socks")]
     Socks(tokio_socks::Error),
@@ -31,6 +33,7 @@ impl fmt::Display for Error {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         match self {
             Self::Ws(e) => write!(f, "{e}"),
+            Self::Io(e) => write!(f, "{e}"),
             #[cfg(feature = "socks")]
             Self::Socks(e) => write!(f, "{e}"),
             #[cfg(feature = "tor")]
@@ -47,6 +50,12 @@ impl From<WsError> for Error {
     }
 }
 
+impl From<std::io::Error> for Error {
+    fn from(e: std::io::Error) -> Self {
+        Self::Io(e)
+    }
+}
+
 #[cfg(feature = "socks")]
 impl From<tokio_socks::Error> for Error {
     fn from(e: tokio_socks::Error) -> Self {
@@ -63,13 +72,11 @@ impl From<tor::Error> for Error {
 
 impl Error {
     #[inline]
-    #[cfg(any(feature = "socks", feature = "tor"))]
     pub(super) fn empty_host() -> Self {
         Self::Url(ParseError::EmptyHost)
     }
 
     #[inline]
-    #[cfg(any(feature = "socks", feature = "tor"))]
     pub(super) fn invalid_port() -> Self {
         Self::Url(ParseError::InvalidPort)
     }

src/native/mod.rs

@@ -3,15 +3,14 @@
 
 //! Native
 
-#[cfg(feature = "socks")]
 use std::net::SocketAddr;
 #[cfg(feature = "tor")]
 use std::path::PathBuf;
+use std::time::Duration;
 
 #[cfg(feature = "tor")]
 use arti_client::DataStream;
 use tokio::io::{AsyncRead, AsyncWrite};
-#[cfg(feature = "socks")]
 use tokio::net::TcpStream;
 use tokio_tungstenite::tungstenite::protocol::Role;
 pub use tokio_tungstenite::tungstenite::Message;
@@ -40,13 +39,156 @@ pub async fn connect(url: &Url, mode: &ConnectionMode) -> Result<WebSocket, Erro
     }
 }
 
+/// Happy Eyeballs connection delay (RFC 8305).
+const HAPPY_EYEBALLS_DELAY: Duration = Duration::from_millis(250);
+
 async fn connect_direct(url: &Url) -> Result<WebSocket, Error> {
+    let host: &str = url.host_str().ok_or_else(Error::empty_host)?;
+    let port: u16 = url
+        .port_or_known_default()
+        .ok_or_else(Error::invalid_port)?;
+
+    let tcp_stream = happy_eyeballs_connect(host, port).await?;
+
     // NOT REMOVE `Box::pin`!
     // Use `Box::pin` to fix stack overflow on windows targets due to large `Future`
-    let (stream, _) = Box::pin(tokio_tungstenite::connect_async(url.as_str())).await?;
+    let (stream, _) = Box::pin(tokio_tungstenite::client_async_tls(
+        url.as_str(),
+        tcp_stream,
+    ))
+    .await?;
     Ok(WebSocket::tokio(Box::new(stream)))
 }
 
+/// Connect to a host using the Happy Eyeballs algorithm (RFC 8305).
+///
+/// When DNS returns both IPv6 and IPv4 addresses, tries the preferred family
+/// first and starts the other family after a 250ms delay if the first hasn't
+/// connected yet. Uses whichever connection succeeds first.
+async fn happy_eyeballs_connect(host: &str, port: u16) -> Result<TcpStream, Error> {
+    let addrs: Vec<SocketAddr> = tokio::net::lookup_host(format!("{host}:{port}"))
+        .await?
+        .collect();
+
+    if addrs.is_empty() {
+        return Err(std::io::Error::new(
+            std::io::ErrorKind::AddrNotAvailable,
+            "DNS resolution returned no addresses",
+        )
+        .into());
+    }
+
+    // Separate into IPv6 and IPv4, preserving order within each group
+    let mut ipv6: Vec<SocketAddr> = Vec::new();
+    let mut ipv4: Vec<SocketAddr> = Vec::new();
+    for addr in addrs {
+        if addr.is_ipv6() {
+            ipv6.push(addr);
+        } else {
+            ipv4.push(addr);
+        }
+    }
+
+    // If only one family, try addresses sequentially
+    if ipv4.is_empty() {
+        return try_addrs_sequential(&ipv6).await;
+    }
+    if ipv6.is_empty() {
+        return try_addrs_sequential(&ipv4).await;
+    }
+
+    // Both families available: Happy Eyeballs
+    // Try first IPv6 address, after delay start first IPv4 in parallel
+    let ipv6_first = ipv6[0];
+    let ipv4_first = ipv4[0];
+
+    // Pin the IPv6 future so it survives across select boundaries
+    let ipv6_fut = TcpStream::connect(ipv6_first);
+    tokio::pin!(ipv6_fut);
+
+    // Phase 1: Give IPv6 a 250ms head start
+    tokio::select! {
+        result = &mut ipv6_fut => {
+            match result {
+                Ok(stream) => return Ok(stream),
+                // IPv6 failed fast, try IPv4 directly
+                Err(_) => return try_addrs_sequential(&ipv4).await,
+            }
+        }
+        _ = tokio::time::sleep(HAPPY_EYEBALLS_DELAY) => {
+            // Timer fired, IPv6 still pending. Start IPv4 and race both.
+        }
+    }
+
+    // Phase 2: Race the still-pending IPv6 against a new IPv4 attempt.
+    // Use a loop so that if one fails, we keep waiting for the other.
+    let ipv4_fut = TcpStream::connect(ipv4_first);
+    tokio::pin!(ipv4_fut);
+
+    let mut ipv6_done = false;
+    let mut ipv4_done = false;
+
+    loop {
+        tokio::select! {
+            result = &mut ipv6_fut, if !ipv6_done => {
+                match result {
+                    Ok(stream) => return Ok(stream),
+                    Err(_) => { ipv6_done = true; }
+                }
+            }
+            result = &mut ipv4_fut, if !ipv4_done => {
+                match result {
+                    Ok(stream) => return Ok(stream),
+                    Err(_) => { ipv4_done = true; }
+                }
+            }
+        }
+        if ipv6_done && ipv4_done {
+            break;
+        }
+    }
+
+    // Both initial attempts failed, try remaining addresses sequentially
+    // Interleave remaining IPv4 and IPv6 per RFC 8305
+    let mut remaining = Vec::new();
+    let ipv4_remaining = ipv4.iter().skip(1);
+    let ipv6_remaining = ipv6.iter().skip(1);
+
+    let mut ipv4_iter = ipv4_remaining.peekable();
+    let mut ipv6_iter = ipv6_remaining.peekable();
+
+    // Interleave: take one from ipv4, then one from ipv6, alternating
+    while ipv4_iter.peek().is_some() || ipv6_iter.peek().is_some() {
+        if let Some(addr) = ipv4_iter.next() {
+            remaining.push(*addr);
+        }
+        if let Some(addr) = ipv6_iter.next() {
+            remaining.push(*addr);
+        }
+    }
+
+    try_addrs_sequential(&remaining).await
+}
+
+/// Try connecting to addresses sequentially, returning the first success.
+async fn try_addrs_sequential(addrs: &[SocketAddr]) -> Result<TcpStream, Error> {
+    let mut last_err = None;
+    for addr in addrs {
+        match TcpStream::connect(addr).await {
+            Ok(stream) => return Ok(stream),
+            Err(e) => last_err = Some(e),
+        }
+    }
+    Err(last_err
+        .unwrap_or_else(|| {
+            std::io::Error::new(
+                std::io::ErrorKind::AddrNotAvailable,
+                "no addresses to connect to",
+            )
+        })
+        .into())
+}
+
 #[cfg(feature = "socks")]
 async fn connect_proxy(url: &Url, proxy: SocketAddr) -> Result<WebSocket, Error> {
     let host: &str = url.host_str().ok_or_else(Error::empty_host)?;