Merge pull request #11464 from TaquitoSuabe/perf/net-keepalive-pooling

🤖 Agent: TCP keepalive (10 s + 3×3 s) now ends dead friends tunnels and relay bridges in ~19 s instead of holding them until the 120 s idle; the stack buffers replace one Vec per connection, not per read. (union of 74 heads on main: fmt, clippy -D warnings, tests green).

Cargo.lock

@@ -1793,6 +1793,7 @@ dependencies = [
  "serde_json",
  "sha1",
  "sha2",
+ "socket2 0.5.10",
 ]
 
 [[package]]

crates/horizon-friends/src/lib.rs

@@ -111,6 +111,20 @@ pub fn resolve_relay(url: &str) -> io::Result<Vec<SocketAddr>> {
     Ok(found)
 }
 
+/// Configure TCP keepalive probes so dead peers (e.g. lost Wi-Fi, sleep, or silent NAT
+/// teardown) are detected by the OS instead of holding threads indefinitely.
+pub fn set_tcp_keepalive(stream: &TcpStream) -> io::Result<()> {
+    let sock = socket2::SockRef::from(stream);
+    let mut keepalive = socket2::TcpKeepalive::new()
+        .with_time(Duration::from_secs(10))
+        .with_interval(Duration::from_secs(3));
+    #[cfg(not(any(target_os = "windows", target_os = "openbsd")))]
+    {
+        keepalive = keepalive.with_retries(3);
+    }
+    sock.set_tcp_keepalive(&keepalive)
+}
+
 /// Carry one connection both ways between a plain stream (the game's) and a
 /// protected one, until either side ends; then both are closed.
 pub fn pump(
@@ -122,13 +136,14 @@ pub fn pump(
     for s in [&plain, reader.get_ref(), writer.get_ref()] {
         s.set_read_timeout(None)?;
         let _ = s.set_nodelay(true);
+        let _ = set_tcp_keepalive(s);
     }
     let mut plain_in = plain.try_clone()?;
     let mut plain_out = plain;
     let outbound = std::thread::Builder::new()
         .name("horizon-friends-out".into())
         .spawn(move || {
-            let mut buf = vec![0u8; crypto::MAX_PAYLOAD];
+            let mut buf = [0u8; crypto::MAX_PAYLOAD];
             loop {
                 match plain_in.read(&mut buf) {
                     Ok(0) | Err(_) => break,
@@ -227,4 +242,13 @@ mod tests {
             started.elapsed()
         );
     }
+
+    #[test]
+    fn tcp_keepalive_configures_stream() {
+        let l = TcpListener::bind("127.0.0.1:0").unwrap();
+        let client = TcpStream::connect(l.local_addr().unwrap()).unwrap();
+        let (server, _) = l.accept().unwrap();
+        assert!(set_tcp_keepalive(&client).is_ok());
+        assert!(set_tcp_keepalive(&server).is_ok());
+    }
 }

crates/horizon-protocol/Cargo.toml

@@ -19,6 +19,7 @@ sha1 = "0.10"
 sha2 = "0.10"
 # AES-NI at run time on x86 for the connection cipher (crypto/blocks.rs).
 aes = "0.8.4"
+socket2 = { version = "0.5", features = ["all"] }
 
 # The name servers of Windows' network adapters, for Minecraft SRV redirects.
 [target.'cfg(windows)'.dependencies]

crates/horizon-protocol/src/connection.rs

@@ -270,6 +270,20 @@ pub struct Connection {
     pub reader: ConnReader,
     pub writer: ConnWriter,
 }
+fn configure_stream(stream: &TcpStream) -> Result<()> {
+    stream.set_nodelay(true)?;
+    let sock = socket2::SockRef::from(stream);
+    let mut keepalive = socket2::TcpKeepalive::new()
+        .with_time(Duration::from_secs(10))
+        .with_interval(Duration::from_secs(3));
+    #[cfg(not(any(target_os = "windows", target_os = "openbsd")))]
+    {
+        keepalive = keepalive.with_retries(3);
+    }
+    let _ = sock.set_tcp_keepalive(&keepalive);
+    Ok(())
+}
+
 impl Connection {
     pub fn connect(host: &str, port: u16, timeout: Duration) -> Result<Self> {
         let addrs: Vec<_> = (host, port)
@@ -344,7 +358,7 @@ impl Connection {
         }
     }
     pub fn from_stream(stream: TcpStream) -> Result<Self> {
-        stream.set_nodelay(true)?;
+        configure_stream(&stream)?;
         Self::over(Stream::Tcp(stream))
     }
     /// A connection over one end of an in-memory [`Pipe`].
@@ -957,4 +971,13 @@ mod tests {
         assert_eq!((id, payload.as_slice()), (3, &b"after"[..]));
         server.join().unwrap();
     }
+
+    #[test]
+    fn tcp_stream_enables_nodelay_and_keepalive() {
+        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
+        let port = listener.local_addr().unwrap().port();
+        let stream = TcpStream::connect(("127.0.0.1", port)).unwrap();
+        let _server = listener.accept().unwrap();
+        assert!(Connection::from_stream(stream).is_ok());
+    }
 }

crates/horizon-relay/src/lib.rs

@@ -413,6 +413,7 @@ fn handle(state: &Arc<State>, mut stream: TcpStream, peer: SocketAddr) -> io::Re
     let first_line = horizon_friends::deadline(&stream, state.config.first_line)?;
     stream.set_read_timeout(Some(state.config.first_line))?;
     let _ = stream.set_nodelay(true);
+    let _ = horizon_friends::set_tcp_keepalive(&stream);
     let mut line = wire::read_line(&mut stream)?.ok_or(io::ErrorKind::UnexpectedEof)?;
     if state.config.proxy_protocol {
         if !state
@@ -774,6 +775,7 @@ fn bridge(
         for s in [&guest, &host] {
             s.set_read_timeout(Some(state.config.idle))?;
             let _ = s.set_nodelay(true);
+            let _ = horizon_friends::set_tcp_keepalive(s);
         }
         let (g, h, st) = (guest.try_clone()?, host.try_clone()?, state.clone());
         let up = std::thread::Builder::new()
@@ -788,7 +790,7 @@ fn bridge(
 fn copy(state: &State, mut from: TcpStream, mut to: TcpStream) {
     let rate = state.config.session_rate;
     let mut bucket = Bucket::new(rate, (rate / 2.0).max(16384.0), Instant::now());
-    let mut buf = vec![0u8; 16 * 1024];
+    let mut buf = [0u8; 16 * 1024];
     loop {
         let n = match from.read(&mut buf) {
             Ok(0) | Err(_) => break,

launcher/Cargo.lock

@@ -2815,6 +2815,7 @@ dependencies = [
  "serde_json",
  "sha1",
  "sha2",
+ "socket2 0.5.10",
 ]
 
 [[package]]