Control channel: refused connections linger outside the auth places

A connection refused before auth used to keep its place among
max_unauthenticated while it lingered (up to 1 s), so refusals could
keep real clients out. Now `refuse` gives that place back and takes one
of a separate, smaller set of lingering places (Limits::max_lingering,
8) before the reply is written. When none is free the connection is
closed right after the FIN instead of lingering, so threads stay
bounded as #1806 intended.

Tests: a_lingering_connection_holds_no_place_to_authenticate and
only_so_many_refused_connections_linger.

crates/ui-egui/src/control.rs

@@ -796,6 +796,10 @@ struct Limits {
     /// How long a connection refused after a reply is kept half-open, discarding input, so the
     /// reply is not lost to a reset (see [`close_after_reply`]).
     linger: std::time::Duration,
+    /// How many refused connections may linger at once. A lingering connection has given back
+    /// its place among [`Self::max_unauthenticated`], so this is what bounds their threads;
+    /// beyond it a refused connection is closed at once.
+    max_lingering: usize,
 }
 
 #[cfg(not(target_arch = "wasm32"))]
@@ -804,10 +808,12 @@ const LIMITS: Limits = Limits {
     max_unauthenticated: 8,
     write_timeout: std::time::Duration::from_secs(30),
     linger: std::time::Duration::from_secs(1),
+    max_lingering: 8,
 };
 
-/// One of the [`Limits::max_unauthenticated`] places, given back when dropped: when the
-/// connection authenticates, and on every way its thread can end.
+/// One of a limited number of places, given back when dropped: a place among
+/// [`Limits::max_unauthenticated`] (given back when the connection authenticates or is refused,
+/// and on every way its thread can end), or among [`Limits::max_lingering`].
 #[cfg(not(target_arch = "wasm32"))]
 struct Slot(std::sync::Arc<std::sync::atomic::AtomicUsize>);
 
@@ -863,16 +869,32 @@ impl std::io::Read for Reading {
 /// the reset can discard the reply before the client reads it (Linux does): a client still
 /// sending when it is refused would never learn why. So send FIN first, then read and discard
 /// whatever arrives until the client closes its side or `linger` passes, and only then close.
+///
+/// Only [`Limits::max_lingering`] connections linger at once: `place` is one of those places,
+/// taken by [`refuse`] before the reply was written, and held for the drain only. Without one
+/// the connection is closed at once after the FIN, accepting the risk of a reset rather than
+/// holding another thread.
 #[cfg(not(target_arch = "wasm32"))]
-fn close_after_reply(reading: &mut Reading, linger: std::time::Duration) {
+fn close_after_reply(reading: &mut Reading, place: Option<Slot>, linger: std::time::Duration) {
     use std::io::Read;
     let _ = reading.stream.shutdown(std::net::Shutdown::Write);
+    let Some(_place) = place else { return };
     reading.until = Some(std::time::Instant::now() + linger);
     let mut sink = [0u8; 1024];
     // Each read is bounded by the time left, and none is made once `linger` has passed.
     while matches!(reading.read(&mut sink), Ok(n) if n > 0) {}
 }
 
+/// A connection is being refused before `auth`: give back its place among those waiting to
+/// authenticate, so it does not keep a real client out while it lingers, and take a place to
+/// linger in if one is free. Done before the reply is written, so a client that has read the
+/// reply sees the places as they will be.
+#[cfg(not(target_arch = "wasm32"))]
+fn refuse(slot: &mut Option<Slot>, lingering: &std::sync::Arc<std::sync::atomic::AtomicUsize>, limits: Limits) -> Option<Slot> {
+    drop(slot.take());
+    Slot::take(lingering, limits.max_lingering)
+}
+
 /// Listen on `127.0.0.1` (random port) and forward authenticated requests to `client`.
 ///
 /// Protocol: newline-delimited JSON-RPC 2.0. The first request on a connection must be
@@ -891,6 +913,7 @@ fn serve_with(client: ControlClient, limits: Limits) -> std::io::Result<Endpoint
     let token = random_token()?;
     let expected = token.clone();
     let waiting = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
+    let lingering = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
     std::thread::Builder::new().name("pdfcraft-control".into()).spawn(move || {
         for stream in listener.incoming().flatten() {
             let _ = stream.set_write_timeout(Some(limits.write_timeout));
@@ -904,6 +927,7 @@ fn serve_with(client: ControlClient, limits: Limits) -> std::io::Result<Endpoint
             };
             let client = client.clone();
             let expected = expected.clone();
+            let lingering = lingering.clone();
             let _ = std::thread::Builder::new().name("pdfcraft-control-conn".into()).spawn(move || {
                 // Held until `auth` succeeds; given back however this thread ends.
                 let mut slot = Some(slot);
@@ -922,8 +946,9 @@ fn serve_with(client: ControlClient, limits: Limits) -> std::io::Result<Endpoint
                         Err(e) if !authed && matches!(e.kind(), std::io::ErrorKind::TimedOut | std::io::ErrorKind::WouldBlock) => {
                             let secs = limits.auth_deadline.as_secs_f64();
                             let e = rpc_error(Value::Null, -32004, &format!("not authenticated within {secs} s"));
+                            let place = refuse(&mut slot, &lingering, limits);
                             if writeln!(write, "{e}").and_then(|_| write.flush()).is_ok() {
-                                close_after_reply(reader.get_mut(), limits.linger);
+                                close_after_reply(reader.get_mut(), place, limits.linger);
                             }
                             break;
                         }
@@ -931,8 +956,9 @@ fn serve_with(client: ControlClient, limits: Limits) -> std::io::Result<Endpoint
                         Ok(LineRead::End) | Err(_) => break,
                         Ok(LineRead::TooLong) => {
                             let e = rpc_error(Value::Null, -32600, &format!("request line longer than {max} bytes before auth"));
+                            let place = refuse(&mut slot, &lingering, limits);
                             if writeln!(write, "{e}").and_then(|_| write.flush()).is_ok() {
-                                close_after_reply(reader.get_mut(), limits.linger);
+                                close_after_reply(reader.get_mut(), place, limits.linger);
                             }
                             break;
                         }
@@ -972,11 +998,13 @@ fn serve_with(client: ControlClient, limits: Limits) -> std::io::Result<Endpoint
                             }
                         }
                     };
+                    // `close` is only ever set before `auth`.
+                    let place = if close { refuse(&mut slot, &lingering, limits) } else { None };
                     if writeln!(write, "{reply}").and_then(|_| write.flush()).is_err() {
                         break;
                     }
                     if close {
-                        close_after_reply(reader.get_mut(), limits.linger);
+                        close_after_reply(reader.get_mut(), place, limits.linger);
                         break;
                     }
                 }
@@ -1121,6 +1149,7 @@ mod transport_tests {
         max_unauthenticated: 2,
         write_timeout: std::time::Duration::from_secs(5),
         linger: std::time::Duration::from_secs(1),
+        max_lingering: 2,
     };
 
     /// The next line the server sends, as JSON, or `None` when it closed the connection instead.
@@ -1198,6 +1227,70 @@ mod transport_tests {
         assert_eq!(reply["error"]["code"], -32600, "{reply}");
     }
 
+    /// A refused connection gives back its place among those waiting to authenticate before it
+    /// lingers, so lingering does not keep real clients out.
+    #[cfg(not(target_arch = "wasm32"))]
+    #[test]
+    fn a_lingering_connection_holds_no_place_to_authenticate() {
+        use std::io::{BufReader, Write};
+        let (_control, client) = super::attach(&egui::Context::default());
+        let ep = super::serve_with(client, TEST_LIMITS).unwrap();
+        let auth = serde_json::json!({ "jsonrpc": "2.0", "id": 1, "method": "auth", "params": { "token": ep.token } });
+        // Refused with most of its line unread, and left open: it lingers for `linger`.
+        let refused = connect(ep.port);
+        refused.try_clone().unwrap().write_all(&vec![b'x'; 16 * super::MAX_AUTH_LINE]).unwrap();
+        let mut r = BufReader::new(refused.try_clone().unwrap());
+        assert_eq!(next_line(&mut r).expect("refused with a reason")["error"]["code"], -32600);
+        // Well within the linger, both places are free for new connections.
+        let started = std::time::Instant::now();
+        let others: Vec<_> = (0..TEST_LIMITS.max_unauthenticated).map(|_| connect(ep.port)).collect();
+        for s in &others {
+            // A refused one has already been closed, so its write may fail; its reply says why.
+            let _ = writeln!(s.try_clone().unwrap(), "{auth}");
+            let reply = next_line(&mut BufReader::new(s.try_clone().unwrap())).expect("a reply");
+            assert_eq!(reply["result"]["ok"], true, "admitted while the refused one lingers: {reply}");
+        }
+        assert!(started.elapsed() < TEST_LIMITS.linger, "checked while it lingered: {:?}", started.elapsed());
+        drop(refused);
+    }
+
+    /// Only [`super::Limits::max_lingering`] refused connections linger at once: one more still
+    /// gets its reply, and is then closed at once instead of lingering.
+    #[cfg(not(target_arch = "wasm32"))]
+    #[test]
+    fn only_so_many_refused_connections_linger() {
+        use std::io::{BufReader, Write};
+        let limits = super::Limits { max_lingering: 1, ..TEST_LIMITS };
+        let (_control, client) = super::attach(&egui::Context::default());
+        let ep = super::serve_with(client, limits).unwrap();
+        let bad = r#"{"jsonrpc": "2.0", "id": 1, "method": "page_count"}"#;
+        // The first refused connection takes the only lingering place and keeps it.
+        let first = connect(ep.port);
+        writeln!(first.try_clone().unwrap(), "{bad}").unwrap();
+        let mut r1 = BufReader::new(first.try_clone().unwrap());
+        assert_eq!(next_line(&mut r1).expect("refused with a reason")["error"]["code"], -32001);
+        // The second still gets its reason (it has nothing unread, so no reset)...
+        let second = connect(ep.port);
+        let mut w2 = second.try_clone().unwrap();
+        writeln!(w2, "{bad}").unwrap();
+        let mut r2 = BufReader::new(second.try_clone().unwrap());
+        assert_eq!(next_line(&mut r2).expect("refused with a reason")["error"]["code"], -32001);
+        // ...and is closed outright, not lingering: writing to it soon fails, where a lingering
+        // connection would read and discard for the whole of `linger`.
+        let started = std::time::Instant::now();
+        let closed = loop {
+            if w2.write_all(b"x").is_err() {
+                break true;
+            }
+            if started.elapsed() > limits.linger {
+                break false;
+            }
+            std::thread::sleep(std::time::Duration::from_millis(20));
+        };
+        assert!(closed && started.elapsed() < limits.linger / 2, "it lingered: {:?}", started.elapsed());
+        drop(first);
+    }
+
     /// #1806: only so many connections may wait to authenticate at once. One more is refused
     /// on the spot; once a waiting one authenticates or times out, its place is free again.
     #[cfg(not(target_arch = "wasm32"))]