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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01K93HTyCiwYvbRGaAkjPvXn

apps/lightcraft/src/control_server.rs

@@ -87,7 +87,7 @@ fn serve(stream: TcpStream, tx: &Sender<ControlRequest>, 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<ControlRequest>, 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<Value>) {
+    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();