connection: try a server's addresses side by side, 250 ms apart

crates/horizon-protocol/src/connection.rs

@@ -187,6 +187,9 @@ fn ipv4_first(mut addrs: Vec<SocketAddr>) -> Vec<SocketAddr> {
     addrs.sort_by_key(SocketAddr::is_ipv6);
     addrs
 }
+/// RFC 8305's connection attempt delay: how long an address has alone before the next
+/// one is tried beside it.
+const ATTEMPT_DELAY: Duration = Duration::from_millis(250);
 /// Both halves before the split.
 pub struct Connection {
     pub reader: ConnReader,
@@ -206,19 +209,54 @@ impl Connection {
         let addrs = crate::resolver::resolve(host, port)?;
         Self::connect_addresses(host, addrs, timeout)
     }
+    /// Each address starts once the earlier ones fail or [`ATTEMPT_DELAY`] after the last
+    /// start, alongside those still trying, and the first to connect wins: a dead address
+    /// in a round-robin set costs 250 ms instead of the whole timeout.
     fn connect_addresses(host: &str, addrs: Vec<SocketAddr>, timeout: Duration) -> Result<Self> {
-        // Report the first address's failure: it is the one vanilla tries, and the IPv6
-        // fallbacks after it often fail only because the client has no IPv6 route.
-        let mut first = None;
-        for addr in ipv4_first(addrs) {
-            match TcpStream::connect_timeout(&addr, timeout) {
-                Ok(stream) => return Self::from_stream(stream),
-                Err(e) => {
-                    first.get_or_insert_with(|| Error::new(format!("Connection refused: {e}")));
+        let addrs = ipv4_first(addrs);
+        let (send, recv) = std::sync::mpsc::channel();
+        let mut errors: Vec<Option<std::io::Error>> = addrs.iter().map(|_| None).collect();
+        let (mut started, mut pending) = (0, 0);
+        loop {
+            if let Some(&addr) = addrs.get(started) {
+                let (send, index) = (send.clone(), started);
+                let spawned = std::thread::Builder::new()
+                    .name("horizon-connect".into())
+                    .spawn(move || {
+                        let _ = send.send((index, TcpStream::connect_timeout(&addr, timeout)));
+                    });
+                match spawned {
+                    Ok(_) => pending += 1,
+                    Err(e) => errors[index] = Some(e),
                 }
+                started += 1;
+            }
+            if pending == 0 {
+                if started < addrs.len() {
+                    continue;
+                }
+                break;
+            }
+            let reply = if started < addrs.len() {
+                match recv.recv_timeout(ATTEMPT_DELAY) {
+                    Ok(reply) => reply,
+                    Err(_) => continue,
+                }
+            } else {
+                recv.recv().expect("a sender is held here")
+            };
+            pending -= 1;
+            match reply {
+                (_, Ok(stream)) => return Self::from_stream(stream),
+                (index, Err(e)) => errors[index] = Some(e),
             }
         }
-        Err(first.unwrap_or_else(|| Error::new(format!("Unknown host {host}"))))
+        // Report the first address's failure: it is the one vanilla tries, and the IPv6
+        // fallbacks after it often fail only because the client has no IPv6 route.
+        match errors.into_iter().flatten().next() {
+            Some(e) => Err(Error::new(format!("Connection refused: {e}"))),
+            None => Err(Error::new(format!("Unknown host {host}"))),
+        }
     }
     pub fn from_stream(stream: TcpStream) -> Result<Self> {
         stream.set_nodelay(true)?;
@@ -400,6 +438,42 @@ mod tests {
         assert_eq!(peer(&c), v6);
     }
 
+    #[test]
+    fn a_dead_address_does_not_hold_up_the_next() {
+        // TEST-NET-1 never answers (or is unreachable at once).
+        let dead: SocketAddr = "192.0.2.1:25565".parse().unwrap();
+        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
+        let live = listener.local_addr().unwrap();
+        let start = Instant::now();
+        let c = Connection::connect_addresses("test", vec![dead, live], Duration::from_secs(10))
+            .unwrap();
+        assert_eq!(peer(&c), live);
+        assert!(
+            start.elapsed() < Duration::from_secs(5),
+            "{:?}",
+            start.elapsed()
+        );
+    }
+
+    #[test]
+    fn every_failure_reports_the_first_address() {
+        let closed = |_| {
+            TcpListener::bind("127.0.0.1:0")
+                .unwrap()
+                .local_addr()
+                .unwrap()
+        };
+        let error = |host, addrs| {
+            Connection::connect_addresses(host, addrs, Duration::from_secs(5))
+                .err()
+                .unwrap()
+                .to_string()
+        };
+        let refused = error("test", (0..3).map(closed).collect());
+        assert!(refused.starts_with("Connection refused"), "{refused}");
+        assert_eq!(error("nowhere", Vec::new()), "Unknown host nowhere");
+    }
+
     #[test]
     fn loopback_with_compression_and_encryption() {
         let listener = TcpListener::bind("127.0.0.1:0").unwrap();