Harden the thread-pool runtime against connection-exhaustion DoS

Build on the EMFILE backoff with the structural fixes that keep a public
listener solid under load and abuse:

- Bound the connection queue: replace the unbounded mpsc channel with a
  sync_channel sized to workers*64 (clamped 256..4096). Each queued stream
  holds an fd, so an unbounded queue let a burst exhaust the descriptor
  limit; the cap is a generous safety ceiling that stays clear of normal
  and ACME-issuance concurrency, with the kernel backlog shedding overflow.

- Add read/write socket timeouts (30s) to every accepted stream. A peer
  that connects and stalls (slowloris, dead keep-alive, a client that stops
  reading) now releases its worker instead of pinning it forever. TLS rides
  the same raw stream, so this covers the handshake too.

- Cap the plain-HTTP redirect listener at 256 concurrent connections so a
  flood can't spawn unbounded threads; excess is shed by closing.

- Wrap per-connection handling in catch_unwind so one panicking request
  can't kill a worker and slowly drain the pool.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: MagicalTux

src/rt/common.rs

@@ -1,6 +1,7 @@
 //! Pieces shared across the blocking runtimes.
 
 use std::io::{self, Read, Write};
+use std::net::TcpStream;
 use std::sync::Mutex;
 use std::time::{Duration, Instant};
 
@@ -10,6 +11,20 @@ use crate::session::Session;
 /// Read buffer size used by the blocking drive loop.
 pub(crate) const READ_BUF: usize = 16 * 1024;
 
+/// Per-operation inactivity timeout for blocking connections. A read or write
+/// that makes no progress within this window fails, so a peer that connects and
+/// then stalls (slowloris, dead keep-alive, a client that stops reading our
+/// response) releases its worker thread instead of pinning it forever.
+pub(crate) const IO_TIMEOUT: Duration = Duration::from_secs(30);
+
+/// Apply [`IO_TIMEOUT`] to a freshly accepted stream. Errors are ignored: a
+/// socket that won't take a timeout will still be bounded by the read/write
+/// loop, and failing the connection here would be worse than serving it.
+pub(crate) fn apply_timeouts(stream: &TcpStream) {
+    let _ = stream.set_read_timeout(Some(IO_TIMEOUT));
+    let _ = stream.set_write_timeout(Some(IO_TIMEOUT));
+}
+
 /// Backoff applied after a descriptor-exhaustion `accept()` error.
 pub(crate) const ACCEPT_BACKOFF: Duration = Duration::from_millis(50);
 

src/rt/threadpool.rs

@@ -6,13 +6,14 @@
 //! served concurrently. There is no async runtime involved.
 
 use std::net::{TcpListener, TcpStream};
-use std::sync::mpsc::{Receiver, Sender};
+use std::sync::atomic::{AtomicUsize, Ordering};
+use std::sync::mpsc::{Receiver, SyncSender};
 use std::sync::{Arc, Mutex};
 use std::thread;
 
 use crate::error::Result;
 use crate::rt::TlsMode;
-use crate::rt::common::serve_blocking;
+use crate::rt::common::{self, serve_blocking};
 use crate::rt::redirect::{self, HttpCtx};
 use crate::session::{Session, SessionConfig};
 
@@ -34,7 +35,17 @@ pub(crate) fn run(
 ) -> Result<()> {
     let shared = Arc::new(Shared { cfg, tls });
     let workers = workers.max(1);
-    let (tx, rx): (Sender<TcpStream>, Receiver<TcpStream>) = std::sync::mpsc::channel();
+    // Bound the queue of accepted-but-unserved connections. Each queued stream
+    // holds an open descriptor, so an unbounded queue lets a connection burst
+    // exhaust the fd limit (which is exactly what melted the server once). The
+    // cap is generous — a safety ceiling, not tight backpressure — so it stays
+    // well clear of normal concurrency, including the ACME path where one worker
+    // blocks on issuance while another answers the validation connection. When
+    // the queue is full `send` blocks the accept loop; further connections wait
+    // in the kernel backlog and are shed there if it too fills.
+    let backlog = workers.saturating_mul(64).clamp(256, 4096);
+    let (tx, rx): (SyncSender<TcpStream>, Receiver<TcpStream>) =
+        std::sync::mpsc::sync_channel(backlog);
     let rx = Arc::new(Mutex::new(rx));
 
     for _ in 0..workers {
@@ -51,8 +62,8 @@ pub(crate) fn run(
                 }
             }
             Err(e) => {
-                if crate::rt::common::note_accept_error("accept error", &e) {
-                    thread::sleep(crate::rt::common::ACCEPT_BACKOFF);
+                if common::note_accept_error("accept error", &e) {
+                    thread::sleep(common::ACCEPT_BACKOFF);
                 }
             }
         }
@@ -72,16 +83,25 @@ fn worker_loop(rx: Arc<Mutex<Receiver<TcpStream>>>, shared: Arc<Shared>) {
         let Ok(stream) = stream else {
             return;
         };
-        if let Err(e) = handle(stream, &shared)
-            && cfg!(debug_assertions)
-        {
-            eprintln!("httpsd: connection ended: {e}");
+        // Isolate each connection: a panic while serving one client must not
+        // kill the worker thread, which would permanently shrink the pool and,
+        // once every worker died, wedge the whole server.
+        let result =
+            std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| handle(stream, &shared)));
+        match result {
+            Ok(Ok(())) => {}
+            Ok(Err(e)) if cfg!(debug_assertions) => {
+                eprintln!("httpsd: connection ended: {e}");
+            }
+            Ok(Err(_)) => {}
+            Err(_) => eprintln!("httpsd: worker recovered from a panic while serving a connection"),
         }
     }
 }
 
 fn handle(mut stream: TcpStream, shared: &Shared) -> Result<()> {
     stream.set_nodelay(true).ok();
+    common::apply_timeouts(&stream);
 
     match &shared.tls {
         TlsMode::Plain => {
@@ -123,26 +143,39 @@ fn handle_acme(
     crate::rt::common::serve_blocking_prefed(&mut stream, &mut session, &initial)
 }
 
-/// Accept loop for the plain-HTTP listener (redirects + ACME HTTP-01). One
-/// thread per connection — this listener carries little traffic.
+/// Accept loop for the plain-HTTP listener (redirects + ACME HTTP-01). It spawns
+/// a thread per connection — these are short-lived (read a request, reply, close)
+/// — but caps how many run at once so a flood can't spawn unbounded threads.
 pub(crate) fn run_http_redirect(listener: TcpListener, ctx: HttpCtx) {
+    /// Maximum redirect connections served concurrently; excess is shed.
+    const MAX_INFLIGHT: usize = 256;
+
     let ctx = Arc::new(ctx);
+    let inflight = Arc::new(AtomicUsize::new(0));
     for incoming in listener.incoming() {
         match incoming {
             Ok(mut stream) => {
+                common::apply_timeouts(&stream);
+                stream.set_nodelay(true).ok();
+                // Shed load past the cap by closing the connection (drop).
+                if inflight.fetch_add(1, Ordering::Relaxed) >= MAX_INFLIGHT {
+                    inflight.fetch_sub(1, Ordering::Relaxed);
+                    continue;
+                }
                 let ctx = Arc::clone(&ctx);
+                let inflight = Arc::clone(&inflight);
                 thread::spawn(move || {
-                    stream.set_nodelay(true).ok();
                     if let Err(e) = redirect::serve(&mut stream, &ctx)
                         && cfg!(debug_assertions)
                     {
                         eprintln!("httpsd: http connection ended: {e}");
                     }
+                    inflight.fetch_sub(1, Ordering::Relaxed);
                 });
             }
             Err(e) => {
-                if crate::rt::common::note_accept_error("http accept error", &e) {
-                    thread::sleep(crate::rt::common::ACCEPT_BACKOFF);
+                if common::note_accept_error("http accept error", &e) {
+                    thread::sleep(common::ACCEPT_BACKOFF);
                 }
             }
         }