From e99fbea7f96404eb956088f2ed3e41d3613c80f6 Mon Sep 17 00:00:00 2001 From: insideTheFlux Date: Fri, 23 Oct 2026 11:59:47 -0700 Subject: [PATCH] Control port: bound the close drain in total; the reset test waits for the rejection close_gracefully read with a 500 ms timeout per read but no overall limit, so a local client sending one byte every few hundred milliseconds kept the drain (and its connection slot) going until it had sent MAX_LINE. Each read now waits at most the time left of DRAIN_TIME (1 s). New test a_closing_connection_is_not_held_open_by_a_trickling_peer: without the limit it drains for 10 s. tcp_connection_is_closed_after_junk no longer sleeps before its second write: it reads the reply and the rejection first, so the server has rejected the junk before the next line is sent. It still pauses before the last read so that a reset can arrive. Without close_gracefully it fails with os error 10053 on Windows. From CodeRabbit's review of #2723. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01PtjHGt5gnCHXgLt4NbGorf --- apps/lightcraft/src/control_server.rs | 62 +++++++++++++++++++++------ 1 file changed, 50 insertions(+), 12 deletions(-) diff --git a/apps/lightcraft/src/control_server.rs b/apps/lightcraft/src/control_server.rs index 7ba1072a6..275341183 100644 --- a/apps/lightcraft/src/control_server.rs +++ b/apps/lightcraft/src/control_server.rs @@ -96,6 +96,9 @@ fn serve(stream: TcpStream, tx: &Sender, ctx: &egui::Context, id }); } +/// The longest a closing connection is drained, however slowly its peer keeps sending. +const DRAIN_TIME: Duration = Duration::from_secs(1); + /// [`serve_lines`] on a socket, then close it so the peer reads every reply. fn serve_stream(stream: TcpStream, handle: impl FnMut(String, Value) -> Option) { serve_lines(&stream, &stream, handle); @@ -105,15 +108,21 @@ fn serve_stream(stream: TcpStream, handle: impl FnMut(String, Value) -> Option 0 { + // a peer that keeps trickling bytes must not hold the connection slot past DRAIN_TIME + let wait = DRAIN_TIME.saturating_sub(start.elapsed()).min(Duration::from_millis(500)); + if wait.is_zero() || stream.set_read_timeout(Some(wait)).is_err() { + break; + } match reader.read(&mut sink) { Ok(0) | Err(_) => break, Ok(n) => left = left.saturating_sub(n), @@ -525,18 +534,47 @@ mod tests { let mut c = TcpStream::connect(addr).unwrap(); c.set_read_timeout(Some(Duration::from_secs(10))).unwrap(); c.write_all(b"{\"id\":7,\"method\":\"x\"}\nhello\n").unwrap(); + let mut replies = BufReader::new(c.try_clone().unwrap()); + let (mut first, mut second) = (String::new(), String::new()); + replies.read_line(&mut first).unwrap(); + replies.read_line(&mut second).unwrap(); + assert!(first.contains("\"id\":7"), "{first}"); + assert!(second.contains("closing the connection"), "{second}"); // A client that keeps sending after the junk: its next line reaches the server after the - // rejection. A socket closed by then answers that line with a reset, and on Windows the reset - // discarded the replies the client had not read yet (os error 10054). - std::thread::sleep(Duration::from_millis(100)); + // rejection. A socket closed by then answers that line with a reset, which the client reads + // as "connection reset" (os error 10053 / 10054 on Windows) instead of the end of the + // replies. The pause lets a reset arrive before the read. c.write_all(b"{\"method\":\"y\"}\n").unwrap(); std::thread::sleep(Duration::from_millis(100)); - let mut text = String::new(); - c.read_to_string(&mut text).unwrap(); // returns at EOF: the server closed the connection + let mut rest = String::new(); + let read = replies.read_to_string(&mut rest); server.join().unwrap(); - let lines: Vec<&str> = text.lines().collect(); - assert_eq!(lines.len(), 2, "{text}"); - assert!(lines[0].contains("\"id\":7")); - assert!(lines[1].contains("closing the connection")); + assert!(read.is_ok(), "the close must be clean, not a reset: {read:?}"); + assert_eq!(rest, "", "nothing follows the rejection"); + } + + #[test] + fn a_closing_connection_is_not_held_open_by_a_trickling_peer() { + let listener = TcpListener::bind(("127.0.0.1", 0)).unwrap(); + let addr = listener.local_addr().unwrap(); + let done = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let server_done = Arc::clone(&done); + let server = std::thread::spawn(move || { + let (s, _) = listener.accept().unwrap(); + let t0 = std::time::Instant::now(); + close_gracefully(&s); + server_done.store(true, Ordering::SeqCst); + t0.elapsed() + }); + let mut c = TcpStream::connect(addr).unwrap(); + // one byte every 100 ms: each read returns well within its 500 ms timeout + for _ in 0..100 { + if done.load(Ordering::SeqCst) || c.write_all(b"x").is_err() { + break; + } + std::thread::sleep(Duration::from_millis(100)); + } + let drained = server.join().unwrap(); + assert!(drained < DRAIN_TIME + Duration::from_secs(1), "drained for {drained:?}"); } }