From 900f62f5a37a2d6db6e8c0a478876469747abdc2 Mon Sep 17 00:00:00 2001 From: Hamza Date: Mon, 19 Oct 2026 20:27:55 +0500 Subject: [PATCH 1/2] Control server: a rejected line's reason survives a pipelined peer A client that sent another line while its previous one was being rejected never saw the reason. `serve` writes the error and drops the socket; the client's second write lands on a socket the server has already closed, the kernel answers it with RST, and an RST discards what was queued for the client -- including the reply. The client reads a connection reset instead of being told why. The comment there, "a peer that can't read the reason loses nothing", holds for a peer that stopped reading. A peer that is reading loses the reason, and it is the one that gets reset. `close_politely` shuts the write side down so the FIN follows the reply, then reads and discards for up to the write timeout and at most 64 KiB, so the peer's pipelined line is absorbed instead of answered with RST. Both paths that reject and close use it: a line that is not a request, and a line over the length limit. A peer that keeps talking is still dropped at the deadline. This is also why `a_line_that_is_not_a_request_closes_the_connection` was red about four times in five: it writes a second line after the junk one. Run alone twelve times, it failed 10 on `main`, 0 with the second line removed, and 0 with this change and the second line kept. --- apps/deckcraft/src/control_server.rs | 53 ++++++++++++++++++++++++++-- 1 file changed, 50 insertions(+), 3 deletions(-) diff --git a/apps/deckcraft/src/control_server.rs b/apps/deckcraft/src/control_server.rs index b4b80c1d..49101a4c 100644 --- a/apps/deckcraft/src/control_server.rs +++ b/apps/deckcraft/src/control_server.rs @@ -8,11 +8,11 @@ //! socket may sit idle or blocked on a write. use std::io::{BufRead, BufReader, ErrorKind, Read, Write}; -use std::net::{TcpListener, TcpStream}; +use std::net::{Shutdown, TcpListener, TcpStream}; use std::sync::Arc; use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::mpsc::{Receiver, SyncSender, TrySendError, sync_channel}; -use std::time::Duration; +use std::time::{Duration, Instant}; use deckcraft_ui_egui::ControlRequest; use serde_json::{Value, json}; @@ -170,6 +170,28 @@ fn is_http_request_line(text: &str) -> bool { || text.rsplit_once(' ').is_some_and(|(_, version)| version.starts_with("HTTP/")) } +/// Close after the last reply so the peer can still read it. A peer that had sent another line +/// before the reply arrived -- two lines in one write, which any script does -- leaves input +/// unread in the socket, and closing a socket with unread input resets the connection: the reply +/// already queued for the peer is discarded and it sees only the reset. Sending the FIN first and +/// then draining, briefly and boundedly, makes the close orderly and the reason survive. +fn close_politely(out: &TcpStream, reader: &mut BufReader, grace: Duration) { + if out.shutdown(Shutdown::Write).is_err() || out.set_read_timeout(Some(grace)).is_err() { + return; + } + // Enough for a pipelined line or two; a peer that keeps talking is dropped at the deadline. + const MAX_DRAIN: usize = 64 << 10; + let until = Instant::now() + grace; + let mut sink = [0u8; 4 << 10]; + let mut drained = 0; + while drained < MAX_DRAIN && Instant::now() < until { + match reader.read(&mut sink) { + Ok(0) | Err(_) => break, + Ok(n) => drained += n, + } + } +} + fn serve(stream: TcpStream, tx: &SyncSender, ctx: &egui::Context, limits: Limits) { if stream.set_read_timeout(Some(limits.idle_timeout)).is_err() || stream.set_write_timeout(Some(limits.write_timeout)).is_err() { return; @@ -182,6 +204,7 @@ fn serve(stream: TcpStream, tx: &SyncSender, ctx: &egui::Context Line::Text(t) => t, Line::TooLong => { let _ = writeln!(out, "{}", json!({"ok": false, "error": format!("request line longer than {} bytes", limits.max_line)})); + close_politely(&out, &mut reader, limits.write_timeout); break; } Line::Closed => break, @@ -189,8 +212,11 @@ fn serve(stream: TcpStream, tx: &SyncSender, ctx: &egui::Context let reply = match parse(&line) { Parsed::Blank => continue, Parsed::Reject(e) => { - // The connection closes either way; a peer that can't read the reason loses nothing. + // The connection closes either way, but the peer is told why first: see + // `close_politely`, without which a peer that had pipelined another line had this + // reply reset away and saw nothing. let _ = writeln!(out, "{}", json!({"ok": false, "error": e})); + close_politely(&out, &mut reader, limits.write_timeout); break; } Parsed::Request { id, method, params } => { @@ -327,6 +353,27 @@ mod tests { assert!(!is_http_request_line("{\"method\":\"GET /\"}")); } + /// A peer that pipelines -- two lines in one write, as a script does -- still reads why its + /// line was rejected. Closing with its second line unread resets the connection and discards + /// the reply already on its way, so the peer saw nothing at all; repeated here because it + /// depended on the timing of the two. + #[test] + fn a_pipelined_peer_still_reads_why_its_line_was_rejected() { + let (addr, rx) = server(small()); + for _ in 0..12 { + let (mut r, mut w) = client(addr); + w.write_all(b"hello\n").unwrap(); + // A second line, written after the first as a script writes it: it races the close. + let _ = w.write_all(b"{\"id\":2,\"method\":\"m\"}\n"); + let replies = replies_until_closed(&mut r); + assert_eq!(replies.len(), 1, "{replies:?}"); + assert_eq!(replies[0]["ok"], false); + assert!(replies[0]["error"].as_str().unwrap_or_default().contains("closing the connection"), "{replies:?}"); + } + // The line after the rejected one is never dispatched, pipelined or not. + assert!(rx.try_recv().is_err(), "a line after the rejection was dispatched"); + } + #[test] fn oversized_line_closes_connection() { let (addr, rx) = server(small()); From 7040bb06195a784fdb74cf42a4e83df45ec09d56 Mon Sep 17 00:00:00 2001 From: Hamza Date: Mon, 19 Oct 2026 23:44:53 +0500 Subject: [PATCH 2/2] Control server: cap the polite drain at the grace deadline set_read_timeout applies to each read, so a peer that sent a byte just before every timeout expired could hold the connection thread, and its slot, for roughly twice the grace. Set the remaining time before each read instead. --- apps/deckcraft/src/control_server.rs | 11 +++++++++-- 1 file changed, 9 insertions(+), 2 deletions(-) diff --git a/apps/deckcraft/src/control_server.rs b/apps/deckcraft/src/control_server.rs index 49101a4c..24ea6562 100644 --- a/apps/deckcraft/src/control_server.rs +++ b/apps/deckcraft/src/control_server.rs @@ -176,15 +176,22 @@ fn is_http_request_line(text: &str) -> bool { /// already queued for the peer is discarded and it sees only the reset. Sending the FIN first and /// then draining, briefly and boundedly, makes the close orderly and the reason survive. fn close_politely(out: &TcpStream, reader: &mut BufReader, grace: Duration) { - if out.shutdown(Shutdown::Write).is_err() || out.set_read_timeout(Some(grace)).is_err() { + if out.shutdown(Shutdown::Write).is_err() { return; } // Enough for a pipelined line or two; a peer that keeps talking is dropped at the deadline. + // The timeout is what is left of the grace before each read, not the whole of it: a peer that + // sends a byte just before each one expires would otherwise hold this thread, and its slot, + // for close to twice as long as the deadline says. const MAX_DRAIN: usize = 64 << 10; let until = Instant::now() + grace; let mut sink = [0u8; 4 << 10]; let mut drained = 0; - while drained < MAX_DRAIN && Instant::now() < until { + while drained < MAX_DRAIN { + let left = until.saturating_duration_since(Instant::now()); + if left.is_zero() || out.set_read_timeout(Some(left)).is_err() { + break; + } match reader.read(&mut sink) { Ok(0) | Err(_) => break, Ok(n) => drained += n,