From 4e7d8319f1c04428672e7c84e80d63b9eb11f966 Mon Sep 17 00:00:00 2001 From: "Dylan M. Taylor" Date: Tue, 3 Nov 2026 04:22:15 +0000 Subject: [PATCH] fix(collab-net): hold a health check's header read to the handshake deadline The request line was read within the hello timeout, but the header loop after it only timed out per read, so a peer dripping header bytes kept the worker thread and its connection slot for as long as it liked. Each header read now gets what is left of the one deadline. --- crates/collab-net/src/hub/conn.rs | 34 +++++++++++++++++++++++++++++++ 1 file changed, 34 insertions(+) diff --git a/crates/collab-net/src/hub/conn.rs b/crates/collab-net/src/hub/conn.rs index 1f365cbab..345f94117 100644 --- a/crates/collab-net/src/hub/conn.rs +++ b/crates/collab-net/src/hub/conn.rs @@ -51,6 +51,12 @@ fn health_check(stream: &mut TcpStream, until: Instant) -> bool { let mut req = Vec::new(); let mut chunk = [0u8; 1024]; while !req.windows(4).any(|w| w == b"\r\n\r\n") && req.len() < 16 << 10 { + // The same deadline as the request line: a peer dripping header bytes would otherwise + // hold this thread and its connection slot for as long as it liked. + let left = until.saturating_duration_since(Instant::now()); + if left.is_zero() || stream.set_read_timeout(Some(left)).is_err() { + break; + } match stream.read(&mut chunk) { Ok(0) | Err(_) => break, Ok(n) => req.extend_from_slice(chunk.get(..n).unwrap_or_default()), @@ -407,3 +413,31 @@ fn linger(ws: &mut WebSocket) { } } } + +#[cfg(test)] +mod health_check_tests { + use super::*; + + #[test] + fn a_health_check_dripping_headers_ends_at_the_deadline() { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let addr = listener.local_addr().unwrap(); + let client = std::thread::spawn(move || { + let mut c = TcpStream::connect(addr).unwrap(); + c.write_all(b"GET /healthz HTTP/1.1\r\n").unwrap(); + // One header byte every 20 ms, far past the deadline, until the server hangs up. + for _ in 0..200 { + if c.write_all(b"X").is_err() { + break; + } + std::thread::sleep(Duration::from_millis(20)); + } + }); + let (mut stream, _) = listener.accept().unwrap(); + let start = Instant::now(); + assert!(health_check(&mut stream, start + Duration::from_millis(300))); + assert!(start.elapsed() < Duration::from_secs(2), "{:?}", start.elapsed()); + drop(stream); + client.join().unwrap(); + } +}