diff --git a/crates/rttp-client/src/config.rs b/crates/rttp-client/src/config.rs index fd971e60..c86d3571 100644 --- a/crates/rttp-client/src/config.rs +++ b/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(), diff --git a/crates/rttp-client/src/connection/connection.rs b/crates/rttp-client/src/connection/connection.rs index 19241e59..ddd7c087 100644 --- a/crates/rttp-client/src/connection/connection.rs +++ b/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( + addr: A, + config: &Config, + timeout_read: time::Duration, + timeout_write: time::Duration, +) -> error::Result +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,7 +663,7 @@ 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; } @@ -658,11 +671,21 @@ where 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 { + 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 { 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(); diff --git a/crates/rttp-client/src/connection/mod.rs b/crates/rttp-client/src/connection/mod.rs index dfccb12b..a836ae59 100644 --- a/crates/rttp-client/src/connection/mod.rs +++ b/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")] diff --git a/crates/rttp-client/src/error.rs b/crates/rttp-client/src/error.rs index 15b1556c..e2102c78 100644 --- a/crates/rttp-client/src/error.rs +++ b/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::()).unwrap_or(false) } @@ -192,6 +193,17 @@ pub(crate) fn request>(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: 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()); + } +} diff --git a/crates/rttp-client/src/http2.rs b/crates/rttp-client/src/http2.rs index b4c24919..80d0fd03 100644 --- a/crates/rttp-client/src/http2.rs +++ b/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 { Ok(format!("{}:{}", host, port)) } -fn connect_tcp_stream(addr: A, config: &Config) -> error::Result -where - A: ToSocketAddrs, -{ +fn connect_tcp_stream( + addr: impl std::net::ToSocketAddrs, + config: &Config, +) -> error::Result { 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(