diff --git a/apps/deckcraft/src/control_server.rs b/apps/deckcraft/src/control_server.rs index 58701731..1773729d 100644 --- a/apps/deckcraft/src/control_server.rs +++ b/apps/deckcraft/src/control_server.rs @@ -66,6 +66,10 @@ pub fn start(port: u16, ctx: egui::Context) -> Receiver { /// core busy until it recovers. const ACCEPT_RETRY: Duration = Duration::from_millis(100); +/// The longest the accepting thread drains a connection it refused as busy before closing it. +/// It accepts no other connection meanwhile. +const REFUSED_GRACE: Duration = Duration::from_millis(250); + /// The connections `incoming` accepts, calling `wait` after each failed accept. fn accepted(incoming: impl Iterator>, mut wait: impl FnMut()) -> impl Iterator { incoming.filter_map(move |stream| stream.map_err(|_| wait()).ok()) @@ -81,6 +85,10 @@ fn serve_listener(listener: TcpListener, limits: Limits, ctx: egui::Context) -> if active.fetch_add(1, Ordering::SeqCst) >= limits.max_clients { active.fetch_sub(1, Ordering::SeqCst); let _ = writeln!(stream, "{}", json!({"ok": false, "error": "busy: too many control connections"})); + // A peer that wrote a line before it read this reply would see only a reset. + if let Ok(read) = stream.try_clone() { + close_politely(&stream, &mut BufReader::new(read), limits.write_timeout.min(REFUSED_GRACE)); + } continue; } let slot = Slot(Arc::clone(&active)); @@ -192,14 +200,16 @@ fn close_politely(out: &TcpStream, reader: &mut BufReader, grace: Dur // 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. + // for close to twice as long as the deadline says. The timeout is set on the handle that + // reads: on Windows, a read timeout set on one handle of a `try_clone` pair can fail to apply + // to reads through the other. 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 { let left = until.saturating_duration_since(Instant::now()); - if left.is_zero() || out.set_read_timeout(Some(left)).is_err() { + if left.is_zero() || reader.get_ref().set_read_timeout(Some(left)).is_err() { break; } match reader.read(&mut sink) { @@ -388,8 +398,11 @@ mod tests { /// 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 { + // A slot comes back only when the thread serving a peer ends, which can be after the next + // peer connects, so every peer has a slot of its own. + const PEERS: usize = 12; + let (addr, rx) = server(Limits { max_clients: PEERS, ..small() }); + for _ in 0..PEERS { 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. @@ -403,6 +416,49 @@ mod tests { assert!(rx.try_recv().is_err(), "a line after the rejection was dispatched"); } + /// A peer refused because every connection is taken still reads why when it had already + /// written a line: closing with that line unread resets the connection and discards the + /// reply. Repeated because it depends on the timing of the two. + #[test] + fn a_pipelined_peer_refused_as_busy_still_reads_why() { + let (addr, rx) = server(Limits { max_clients: 1, idle_timeout: Duration::from_secs(30), ..small() }); + ui(rx); + // Take the only slot, and make sure it is being served. + let (mut hr, mut hw) = client(addr); + writeln!(hw, r#"{{"id":1,"method":"m"}}"#).unwrap(); + assert!(reply(&mut hr).is_some()); + for _ in 0..50 { + let (mut r, mut w) = client(addr); + let _ = w.write_all(b"{\"id\":2,\"method\":\"m\"}\n"); + let replies = replies_until_closed(&mut r); + assert_eq!(replies.len(), 1, "{replies:?}"); + assert!(replies[0]["error"].as_str().unwrap_or_default().starts_with("busy"), "{replies:?}"); + } + } + + /// A peer that keeps its connection open after its line was rejected gives its slot back at + /// the drain deadline (the write timeout), not at the idle timeout. + #[test] + fn a_rejected_peer_that_stays_connected_frees_its_slot_at_the_deadline() { + let (addr, rx) = server(Limits { max_clients: 1, idle_timeout: Duration::from_secs(30), ..small() }); + ui(rx); + let (mut r, mut w) = client(addr); + w.write_all(b"hello\n").unwrap(); + assert_eq!(reply(&mut r).unwrap()["ok"], false); + // `w` stays open, sending nothing. + let deadline = Instant::now() + Duration::from_secs(4); + loop { + let (mut dr, mut dw) = client(addr); + writeln!(dw, r#"{{"id":2,"method":"m"}}"#).unwrap(); + if reply(&mut dr).is_some_and(|v| v["ok"] == true) { + break; + } + assert!(Instant::now() < deadline, "slot never freed"); + std::thread::sleep(Duration::from_millis(50)); + } + drop(w); + } + #[test] fn oversized_line_closes_connection() { let (addr, rx) = server(small()); @@ -454,17 +510,18 @@ mod tests { #[test] fn blocked_ui_times_out_and_full_queue_is_busy() { let (addr, rx) = server(small()); - // No UI: the first request fills the one-slot queue and times out; the second finds it full. + // No UI: the request that reaches the one-slot queue first fills it and times out; the + // other finds it full. Each connection has its own thread, so either request can be first. let (mut ar, mut aw) = client(addr); let (mut br, mut bw) = client(addr); writeln!(aw, r#"{{"id":1,"method":"m"}}"#).unwrap(); - std::thread::sleep(Duration::from_millis(50)); writeln!(bw, r#"{{"id":2,"method":"m"}}"#).unwrap(); - let b = reply(&mut br).unwrap(); - assert!(b["error"].as_str().unwrap().starts_with("busy")); - assert_eq!(b["id"], 2); let a = reply(&mut ar).unwrap(); - assert_eq!(a["error"], "timeout"); + let b = reply(&mut br).unwrap(); + assert_eq!((&a["id"], &b["id"]), (&json!(1), &json!(2))); + let mut errors = [a["error"].as_str().unwrap_or_default(), b["error"].as_str().unwrap_or_default()]; + errors.sort_unstable(); + assert_eq!(errors, ["busy: too many queued requests", "timeout"], "{a} {b}"); drop(rx); }