feat: RTTP: Add an explicit TCP connect timeout

crates/rttp-client/src/config.rs

@@ -38,6 +38,7 @@ impl H2cClientPolicy {
 
 #[derive(Clone, Debug)]
 pub struct Config {
+  connect_timeout: u64,
   read_timeout: u64,
   write_timeout: u64,
   auto_redirect: bool,
@@ -51,6 +52,7 @@ pub struct Config {
 impl Default for Config {
   fn default() -> Self {
     Config::builder()
+      .connect_timeout(10000)
       .read_timeout(10000)
       .write_timeout(10000)
       .auto_redirect(false)
@@ -66,6 +68,13 @@ impl Config {
 }
 
 impl Config {
+  /// Return the TCP connect timeout in milliseconds.
+  ///
+  /// Each resolved address receives this timeout independently. The default
+  /// is 10,000 milliseconds.
+  pub fn connect_timeout(&self) -> u64 {
+    self.connect_timeout
+  }
   pub fn read_timeout(&self) -> u64 {
     self.read_timeout
   }
@@ -119,6 +128,7 @@ impl ConfigBuilder {
   pub fn new() -> Self {
     Self {
       config: Config {
+        connect_timeout: 10000,
         read_timeout: 10000,
         write_timeout: 10000,
         auto_redirect: false,
@@ -135,6 +145,16 @@ impl ConfigBuilder {
     self.config.clone()
   }
 
+  /// Set the TCP connect timeout in milliseconds for each resolved address.
+  ///
+  /// The value must be greater than zero. Addresses are attempted in resolver
+  /// order, so the total connect time is bounded by the number of resolved
+  /// addresses multiplied by this timeout.
+  pub fn connect_timeout(&mut self, connect_timeout: u64) -> &mut Self {
+    self.config.connect_timeout = connect_timeout;
+    self
+  }
+
   pub fn read_timeout(&mut self, read_timeout: u64) -> &mut Self {
     self.config.read_timeout = read_timeout;
     self
@@ -210,6 +230,7 @@ mod tests {
   #[test]
   fn builder_updates_values() {
     let config = Config::builder()
+      .connect_timeout(2468)
       .read_timeout(1234)
       .write_timeout(4321)
       .auto_redirect(true)
@@ -220,6 +241,7 @@ mod tests {
       .http2_header_table_size(64)
       .build();
 
+    assert_eq!(config.connect_timeout(), 2468);
     assert_eq!(config.read_timeout(), 1234);
     assert_eq!(config.write_timeout(), 4321);
     assert!(config.auto_redirect());
@@ -235,6 +257,11 @@ mod tests {
     let default_config = Config::default();
     let builder_config = Config::builder().build();
 
+    assert_eq!(
+      default_config.connect_timeout(),
+      builder_config.connect_timeout()
+    );
+    assert_eq!(default_config.connect_timeout(), 10_000);
     assert_eq!(default_config.read_timeout(), builder_config.read_timeout());
     assert_eq!(
       default_config.write_timeout(),

crates/rttp-client/src/connection/connection.rs

@@ -628,6 +628,19 @@ where
 {
   let timeout_read = tcp_timeout_duration("read", config.read_timeout())?;
   let timeout_write = tcp_timeout_duration("write", config.write_timeout())?;
+  connect_tcp_stream_with_io_timeouts(addr, config, timeout_read, timeout_write)
+}
+
+pub(crate) fn connect_tcp_stream_with_io_timeouts<A>(
+  addr: A,
+  config: &Config,
+  timeout_read: time::Duration,
+  timeout_write: time::Duration,
+) -> error::Result<std::net::TcpStream>
+where
+  A: ToSocketAddrs,
+{
+  let timeout_connect = tcp_connect_timeout_duration(config.connect_timeout())?;
   let mut last_err = None;
 
   let addrs = addr.to_socket_addrs().map_err(error::request)?;
@@ -650,19 +663,29 @@ where
       continue;
     }
 
-    if let Err(err) = socket.connect(&addr.into()) {
+    if let Err(err) = socket.connect_timeout(&addr.into(), timeout_connect) {
       last_err = Some(err);
       continue;
     }
 
     return Ok(std::net::TcpStream::from(socket));
   }
 
-  Err(error::request(
+  Err(error::connect(
     last_err.unwrap_or_else(|| io::Error::other("failed to connect")),
   ))
 }
 
+fn tcp_connect_timeout_duration(millis: u64) -> error::Result<time::Duration> {
+  if millis == 0 {
+    return Err(error::request(io::Error::new(
+      io::ErrorKind::InvalidInput,
+      "connect timeout must be greater than 0",
+    )));
+  }
+  tcp_timeout_duration("connect", millis)
+}
+
 fn tcp_timeout_duration(name: &str, millis: u64) -> error::Result<time::Duration> {
   if millis > i64::MAX as u64 {
     return Err(error::request(io::Error::new(
@@ -1206,10 +1229,12 @@ mod tests {
   use std::io::{self, Cursor, Read, Write};
   use std::net::{Ipv4Addr, Ipv6Addr, SocketAddr, TcpListener};
   use std::thread;
+  use std::time::{Duration, Instant};
 
   use crate::request::RequestBody;
   use crate::types::Proxy;
   use crate::Config;
+  use socket2::{Domain, Protocol, Socket, Type};
   use url::Url;
 
   use super::{
@@ -1358,6 +1383,63 @@ mod tests {
     assert!(err.to_string().contains("too large"));
   }
 
+  #[test]
+  fn test_connect_tcp_stream_rejects_zero_connect_timeout() {
+    let listener = TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).unwrap();
+    let addr = listener.local_addr().unwrap();
+    let config = Config::builder().connect_timeout(0).build();
+
+    let err = connect_tcp_stream(&[addr][..], &config).unwrap_err();
+
+    assert!(err
+      .to_string()
+      .contains("connect timeout must be greater than 0"));
+  }
+
+  #[test]
+  fn test_connect_tcp_stream_rejects_too_large_connect_timeout() {
+    let listener = TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).unwrap();
+    let addr = listener.local_addr().unwrap();
+    let config = Config::builder().connect_timeout(u64::MAX).build();
+
+    let err = connect_tcp_stream(&[addr][..], &config).unwrap_err();
+
+    assert!(err.to_string().contains("connect timeout is too large"));
+  }
+
+  #[test]
+  fn test_connect_tcp_stream_times_out() {
+    let listener = Socket::new(Domain::IPV4, Type::STREAM, Some(Protocol::TCP)).unwrap();
+    listener
+      .bind(&SocketAddr::new(Ipv4Addr::LOCALHOST.into(), 0).into())
+      .unwrap();
+    listener.listen(1).unwrap();
+    let addr = listener.local_addr().unwrap().as_socket().unwrap();
+    let mut queued_connections = Vec::new();
+    loop {
+      let socket = Socket::new(Domain::IPV4, Type::STREAM, Some(Protocol::TCP)).unwrap();
+      match socket.connect_timeout(&addr.into(), Duration::from_millis(50)) {
+        Ok(()) => queued_connections.push(socket),
+        Err(err)
+          if matches!(
+            err.kind(),
+            io::ErrorKind::TimedOut | io::ErrorKind::WouldBlock
+          ) =>
+        {
+          break;
+        }
+        Err(err) => panic!("failed to saturate local listen queue: {err}"),
+      }
+    }
+
+    let config = Config::builder().connect_timeout(25).build();
+    let started = Instant::now();
+    let err = connect_tcp_stream(&[addr][..], &config).unwrap_err();
+
+    assert!(err.is_timeout());
+    assert!(started.elapsed() < Duration::from_secs(1));
+  }
+
   #[test]
   fn test_connect_tcp_stream_reports_write_timeout_configuration_errors() {
     let listener = TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).unwrap();

crates/rttp-client/src/connection/mod.rs

@@ -6,7 +6,7 @@ pub(crate) use self::async_connection::AsyncStreamingRequestBody;
 pub use self::async_connection::*;
 pub use self::block_connection::*;
 pub use self::connection::HandoffConnection;
-pub(crate) use self::connection::StreamingRequestBody;
+pub(crate) use self::connection::{connect_tcp_stream_with_io_timeouts, StreamingRequestBody};
 pub use self::connection_reader::{ConnectionReader, ResponseBodyReader, StreamingResponse};
 
 #[cfg(feature = "async")]

crates/rttp-client/src/error.rs

@@ -57,7 +57,8 @@ impl Error {
     matches!(self.inner.kind, Kind::Status(_))
   }
 
-  /// Returns true if the error is related to a timeout.
+  /// Returns true if the error is related to a timeout, including a TCP
+  /// connect timeout.
   pub fn is_timeout(&self) -> bool {
     self.source().map(|e| e.is::<TimedOut>()).unwrap_or(false)
   }
@@ -192,6 +193,17 @@ pub(crate) fn request<E: Into<BoxError>>(e: E) -> Error {
   Error::new(Kind::Request, Some(e))
 }
 
+pub(crate) fn connect(e: io::Error) -> Error {
+  if matches!(
+    e.kind(),
+    io::ErrorKind::TimedOut | io::ErrorKind::WouldBlock
+  ) {
+    Error::new(Kind::Request, Some(TimedOut(e)))
+  } else {
+    request(e)
+  }
+}
+
 pub(crate) fn response<E: Into<BoxError>>(e: E) -> Error {
   Error::new(Kind::Response, Some(e))
 }
@@ -277,12 +289,28 @@ pub(crate) fn decode_io(e: io::Error) -> Error {
 // internal Error "sources"
 
 #[derive(Debug)]
-pub(crate) struct TimedOut;
+pub(crate) struct TimedOut(io::Error);
 
 impl fmt::Display for TimedOut {
   fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
-    f.write_str("operation timed out")
+    f.write_str("TCP connect timed out")
+  }
+}
+
+impl StdError for TimedOut {
+  fn source(&self) -> Option<&(dyn StdError + 'static)> {
+    Some(&self.0)
   }
 }
 
-impl StdError for TimedOut {}
+#[cfg(test)]
+mod tests {
+  use super::*;
+
+  #[test]
+  fn connect_timeout_is_exposed_as_timeout() {
+    let err = connect(io::Error::new(io::ErrorKind::TimedOut, "connect timed out"));
+
+    assert!(err.is_timeout());
+  }
+}

crates/rttp-client/src/http2.rs

@@ -1,11 +1,11 @@
 use std::io::{self, Read, Write};
-use std::net::{TcpStream, ToSocketAddrs};
+use std::net::TcpStream;
 use std::time::Duration;
 
 use base64::Engine;
-use socket2::{Domain, Protocol, Socket, Type};
 use url::Url;
 
+use crate::connection::connect_tcp_stream_with_io_timeouts;
 use crate::request::RawRequest;
 use crate::response::Response;
 use crate::types::{Header, RoUrl, ToUrl};
@@ -323,40 +323,13 @@ fn addr(url: &Url) -> error::Result<String> {
   Ok(format!("{}:{}", host, port))
 }
 
-fn connect_tcp_stream<A>(addr: A, config: &Config) -> error::Result<TcpStream>
-where
-  A: ToSocketAddrs,
-{
+fn connect_tcp_stream(
+  addr: impl std::net::ToSocketAddrs,
+  config: &Config,
+) -> error::Result<TcpStream> {
   let timeout_read = timeout_duration("read", config.read_timeout())?;
   let timeout_write = timeout_duration("write", config.write_timeout())?;
-  let mut last_err = None;
-
-  for addr in addr.to_socket_addrs().map_err(error::request)? {
-    let socket = match Socket::new(Domain::for_address(addr), Type::STREAM, Some(Protocol::TCP)) {
-      Ok(socket) => socket,
-      Err(err) => {
-        last_err = Some(err);
-        continue;
-      }
-    };
-    if let Err(err) = socket.set_read_timeout(Some(timeout_read)) {
-      last_err = Some(err);
-      continue;
-    }
-    if let Err(err) = socket.set_write_timeout(Some(timeout_write)) {
-      last_err = Some(err);
-      continue;
-    }
-    if let Err(err) = socket.connect(&addr.into()) {
-      last_err = Some(err);
-      continue;
-    }
-    return Ok(socket.into());
-  }
-
-  Err(error::request(last_err.unwrap_or_else(|| {
-    io::Error::new(io::ErrorKind::NotFound, "no socket address resolved")
-  })))
+  connect_tcp_stream_with_io_timeouts(addr, config, timeout_read, timeout_write)
 }
 
 fn reject_goaway_before_opening_request_stream(