feat(connect): add TCP_USER_TIMEOUT option for Linux sockets (#213)

Author: 0x676e67

src/connect.rs

@@ -39,23 +39,13 @@ pub enum TargetAddr {
 /// with an IPv6 CIDR and a fallback IP address.
 #[derive(Clone)]
 pub struct Connector {
-    /// Optional IPv6 CIDR (Classless Inter-Domain Routing), used to optionally
-    /// configure an IPv6 address.
     cidr: Option<IpCidr>,
-
-    /// Optional CIDR range for IP addresses.
     cidr_range: Option<u8>,
-
-    /// Optional IP address as a fallback option in case of connection failure.
     fallback: Option<Fallback>,
-
-    /// Connect timeout in milliseconds.
     connect_timeout: Duration,
-
-    /// Enable SO_REUSEADDR for outbond connection socket
+    #[cfg(any(target_os = "android", target_os = "fuchsia", target_os = "linux"))]
+    tcp_user_timeout: Option<Duration>,
     reuseaddr: Option<bool>,
-
-    /// Default http connector
     http: connect::HttpConnector,
 }
 
@@ -134,6 +124,8 @@ impl Connector {
         cidr_range: Option<u8>,
         fallback: Option<Fallback>,
         connect_timeout: u64,
+        #[cfg(any(target_os = "android", target_os = "fuchsia", target_os = "linux"))]
+        tcp_user_timeout: Option<u64>,
         reuseaddr: Option<bool>,
     ) -> Self {
         let connect_timeout = Duration::from_secs(connect_timeout);
@@ -147,8 +139,10 @@ impl Connector {
             cidr_range,
             fallback,
             connect_timeout,
-            http: http_connector,
+            #[cfg(any(target_os = "android", target_os = "fuchsia", target_os = "linux"))]
+            tcp_user_timeout: tcp_user_timeout.map(Duration::from_secs),
             reuseaddr,
+            http: http_connector,
         }
     }
 
@@ -204,20 +198,28 @@ impl TcpConnector<'_> {
 
     /// Creates a [`TcpSocket`] and binds it to an IP address within the provided CIDR range.
     async fn create_socket_with_cidr(&self, cidr: IpCidr) -> std::io::Result<TcpSocket> {
-        match cidr {
+        let socket = match cidr {
             IpCidr::V4(cidr) => {
                 let socket = TcpSocket::new_v4()?;
                 let addr = assign_ipv4_from_extension(cidr, self.inner.cidr_range, self.extension);
                 socket.bind(SocketAddr::new(IpAddr::V4(addr), 0))?;
-                Ok(socket)
+                socket
             }
             IpCidr::V6(cidr) => {
                 let socket = TcpSocket::new_v6()?;
                 let addr = assign_ipv6_from_extension(cidr, self.inner.cidr_range, self.extension);
                 socket.bind(SocketAddr::new(IpAddr::V6(addr), 0))?;
-                Ok(socket)
+                socket
             }
+        };
+
+        #[cfg(any(target_os = "android", target_os = "fuchsia", target_os = "linux"))]
+        if let Some(tcp_user_timeout) = self.inner.tcp_user_timeout {
+            let socket_ref = socket2::SockRef::from(&socket);
+            socket_ref.set_tcp_user_timeout(Some(tcp_user_timeout))?;
         }
+
+        Ok(socket)
     }
 
     /// Creates a [`TcpSocket`] and binds it to the fallback address.
@@ -289,10 +291,17 @@ impl TcpConnector<'_> {
         };
 
         socket.set_nodelay(true)?;
+
         if let Some(reuseaddr) = self.inner.reuseaddr {
             socket.set_reuseaddr(reuseaddr)?;
         }
 
+        #[cfg(any(target_os = "android", target_os = "fuchsia", target_os = "linux"))]
+        if let Some(tcp_user_timeout) = self.inner.tcp_user_timeout {
+            let socket_ref = socket2::SockRef::from(&socket);
+            socket_ref.set_tcp_user_timeout(Some(tcp_user_timeout))?;
+        }
+
         Ok(socket)
     }
 
@@ -672,10 +681,16 @@ impl HttpConnector<'_> {
         }
 
         connector.set_nodelay(true);
+
         if let Some(reuseaddr) = self.inner.reuseaddr {
             connector.set_reuse_address(reuseaddr);
         }
 
+        #[cfg(any(target_os = "android", target_os = "fuchsia", target_os = "linux"))]
+        if let Some(tcp_user_timeout) = self.inner.tcp_user_timeout {
+            connector.set_tcp_user_timeout(Some(tcp_user_timeout));
+        }
+
         Client::builder(TokioExecutor::new())
             .timer(TokioTimer::new())
             .http1_title_case_headers(true)

src/main.rs

@@ -160,11 +160,6 @@ pub struct BootArgs {
     )]
     log: Level,
 
-    /// Worker thread count. Default: number of logical CPU cores.
-    /// Too small limits concurrency; too large wastes context switches.
-    #[arg(long, short = 'w', verbatim_doc_comment)]
-    workers: Option<usize>,
-
     /// Bind address (listen endpoint).
     /// e.g. 0.0.0.0:1080, [::]:1080, 192.168.1.100:1080
     #[arg(
@@ -175,6 +170,17 @@ pub struct BootArgs {
     )]
     bind: SocketAddr,
 
+    /// Maximum concurrent active connections.
+    /// Protects resource exhaustion. Raise cautiously.
+    /// e.g. 128.
+    #[arg(long, short = 'c', default_value = "1024", verbatim_doc_comment)]
+    concurrent: u32,
+
+    /// Worker thread count. Default: number of logical CPU cores.
+    /// Too small limits concurrency; too large wastes context switches.
+    #[arg(long, short = 'w', verbatim_doc_comment)]
+    workers: Option<usize>,
+
     /// Base CIDR block for outbound source address selection.
     /// Used for session, TTL and range extensions.
     /// e.g. 2001:db8::/32 or 10.0.0.0/24
@@ -200,14 +206,17 @@ pub struct BootArgs {
     #[arg(long, short = 't', default_value = "10", verbatim_doc_comment)]
     connect_timeout: u64,
 
-    /// Maximum concurrent active connections.
-    /// Protects resource exhaustion. Raise cautiously.
-    /// e.g. 128.
-    #[arg(long, short = 'c', default_value = "1024", verbatim_doc_comment)]
-    concurrent: u32,
-
-    /// Enable SO_REUSEADDR for outbound sockets.
-    /// Helps mitigate TIME_WAIT port exhaustion.
+    /// Outbound TCP sockets user timeout (seconds).
+    /// Maximum time transmitted data may remain unacknowledged before aborting the connection.
+    /// Not a keepalive: idle connections without in-flight data are unaffected.
+    /// Linux only. Kernel expects milliseconds; this value is converted from seconds.
+    /// e.g. 15.
+    #[cfg(any(target_os = "android", target_os = "fuchsia", target_os = "linux"))]
+    #[arg(long, default_value = "30", verbatim_doc_comment)]
+    tcp_user_timeout: Option<u64>,
+
+    /// Outbound SO_REUSEADDR for TCP sockets.
+    /// Helps mitigate TIME_WAIT port exhaustion and enables fast rebinding after restarts.
     /// e.g. true.
     #[arg(long, default_value = "true", verbatim_doc_comment)]
     reuseaddr: Option<bool>,

src/server.rs

@@ -97,6 +97,8 @@ pub fn run(args: BootArgs) -> Result<()> {
                     args.cidr_range,
                     args.fallback,
                     args.connect_timeout,
+                    #[cfg(any(target_os = "android", target_os = "fuchsia", target_os = "linux"))]
+                    args.tcp_user_timeout,
                     args.reuseaddr,
                 ),
             };