From 2b341bbdee6fa128320d8e7918360b9348dd12f9 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 21 Oct 2026 23:56:44 -0700 Subject: [PATCH 1/2] Control port: close gracefully so the peer reads the last reply (Windows reset) After rejecting a line the server closed its socket at once. A line the client sent after the junk then arrived at a closed socket, which answers with a reset, and on Windows the reset discards the replies the client has not read yet: the client saw os error 10053 / 10054 instead of the error explaining why. This is the Windows x64 failure of tcp_connection_is_closed_after_junk in #2377's CI. serve_stream now shuts down the write side (FIN), reads and drops what is left (at most 4 MiB, 500 ms), then closes. close_gracefully is taken from #1970 by LochLogic, split out so it can land ahead of the per-run token work; #1991 did not include it. The test sends its last line 100 ms after the junk and reads after another 100 ms, so it fails on every run without the fix on Windows (10053) instead of depending on timing. Co-Authored-By: LochLogic <252105752+LochLogic@users.noreply.github.com> Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01K93HTyCiwYvbRGaAkjPvXn --- apps/lightcraft/src/control_server.rs | 38 ++++++++++++++++++++++++--- 1 file changed, 34 insertions(+), 4 deletions(-) diff --git a/apps/lightcraft/src/control_server.rs b/apps/lightcraft/src/control_server.rs index c7244c703..7ba1072a6 100644 --- a/apps/lightcraft/src/control_server.rs +++ b/apps/lightcraft/src/control_server.rs @@ -87,7 +87,7 @@ fn serve(stream: TcpStream, tx: &Sender, ctx: &egui::Context, id log::warn!("control connection closed: could not set idle read timeout: {e}"); return; } - serve_lines(&stream, &stream, |method, params| { + serve_stream(stream, |method, params| { let (req, rrx) = ControlRequest::new(method, params); let state = Arc::clone(&req.state); tx.send(req).ok()?; @@ -96,6 +96,31 @@ fn serve(stream: TcpStream, tx: &Sender, ctx: &egui::Context, id }); } +/// [`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); + close_gracefully(&stream); +} + +/// Close a connection the server is done with without destroying its last reply. A socket closed +/// while the peer's later lines are unread, or arrive after it, answers them with a reset, and a +/// reset can discard the replies the peer has not read yet (on Windows it does): the client sees +/// "connection reset" instead of why. Send FIN, then read and drop what is left (bounded in bytes +/// and time) before closing. From #1970 (LochLogic). +fn close_gracefully(stream: &TcpStream) { + let _ = stream.shutdown(std::net::Shutdown::Write); + let _ = stream.set_read_timeout(Some(Duration::from_millis(500))); + let mut sink = [0u8; 8192]; + let mut left = MAX_LINE; + let mut reader = stream; + while left > 0 { + match reader.read(&mut sink) { + Ok(0) | Err(_) => break, + Ok(n) => left = left.saturating_sub(n), + } + } +} + /// Wait for a request's reply. While the app has not picked the request up and `frames` (its /// frame counter) stands still for `no_frames`, the request is abandoned (it will not run late) /// and the reply says why; a request the app is running gets the whole `timeout`. @@ -495,12 +520,17 @@ mod tests { let addr = listener.local_addr().unwrap(); let server = std::thread::spawn(move || { let (s, _) = listener.accept().unwrap(); - let read = s.try_clone().unwrap(); - serve_lines(read, s, |_, _| Some(json!({"ok": true}))); + serve_stream(s, |_, _| Some(json!({"ok": true}))); }); 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{\"method\":\"y\"}\n").unwrap(); + c.write_all(b"{\"id\":7,\"method\":\"x\"}\nhello\n").unwrap(); + // 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)); + 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 server.join().unwrap(); From e99fbea7f96404eb956088f2ed3e41d3613c80f6 Mon Sep 17 00:00:00 2001 From: insideTheFlux Date: Fri, 23 Oct 2026 11:59:47 -0700 Subject: [PATCH 2/2] 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:?}"); } }