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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01PtjHGt5gnCHXgLt4NbGorf
This commit is contained in:
insideTheFlux
2026-10-23 11:59:47 -07:00
co-authored by Claude Opus 5.5
parent 2b341bbdee
commit e99fbea7f9
+50 -12
View File
@@ -96,6 +96,9 @@ fn serve(stream: TcpStream, tx: &Sender<ControlRequest>, 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<Value>) {
serve_lines(&stream, &stream, handle);
@@ -105,15 +108,21 @@ fn serve_stream(stream: TcpStream, handle: impl FnMut(String, Value) -> Option<V
/// 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).
/// "connection reset" instead of why. Send FIN, then read and drop what is left (at most
/// [`MAX_LINE`] bytes, 500 ms per read and [`DRAIN_TIME`] in all) 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 start = std::time::Instant::now();
let mut sink = [0u8; 8192];
let mut left = MAX_LINE;
let mut reader = stream;
while left > 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:?}");
}
}