Handle HTTP connections in separate threads with cap of 8

A slow connection or DHT lookup (up to 30s) no longer blocks
the entire HTTP accept loop. Connections beyond the limit get
a 503 response.

1 files changed, 26 insertions, 2 deletions

diff --git a/src/daemon.rs b/src/daemon.rs
index c578e5c..3270c1b 100644
--- a/src/daemon.rs
+++ b/src/daemon.rs
@@ -8,6 +8,7 @@
 
 use std::io::{BufRead, BufReader, Read, Write};
 use std::path::Path;
+use std::sync::Arc;
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::mpsc;
 use std::time::{Duration, Instant};
@@ -280,7 +281,9 @@ fn handle_client(
         line.clear();
         // Limit read to MAX_LINE_SIZE to prevent a client from
         // exhausting memory with an unbounded request line.
-        let n = (&mut reader).take(MAX_LINE_SIZE as u64).read_line(&mut line)?;
+        let n = (&mut reader)
+            .take(MAX_LINE_SIZE as u64)
+            .read_line(&mut line)?;
         if n == 0 {
             break;
         }
@@ -326,6 +329,9 @@ fn handle_client(
 
 // ── HTTP server ─────────────────────────────────────
 
+/// Maximum concurrent HTTP handler threads.
+const MAX_HTTP_THREADS: usize = 8;
+
 /// Minimal HTTP server. Serves pastes at /<hash> or
 /// /<hash>/<enckey>. Queries the daemon via Unix socket
 /// so it can access DHT-replicated data too.
@@ -350,11 +356,29 @@ pub fn run_http(
 
     log::info!("http: listening on {addr}");
 
+    let active = Arc::new(std::sync::atomic::AtomicUsize::new(0));
+    let sock_owned = sock_path.to_path_buf();
+
     while !shutdown.load(Ordering::Relaxed) {
         match listener.accept() {
             Ok((stream, _)) => {
+                if active.load(Ordering::Relaxed) >= MAX_HTTP_THREADS {
+                    log::warn!("http: max connections reached, rejecting");
+                    let mut s = stream;
+                    let _ = s.write_all(
+                        b"HTTP/1.1 503 Service Unavailable\r\n\
+                          Connection: close\r\n\r\n",
+                    );
+                    continue;
+                }
                 let store = store.clone();
-                handle_http(stream, &store, sock_path);
+                let sock = sock_owned.clone();
+                let counter = Arc::clone(&active);
+                counter.fetch_add(1, Ordering::Relaxed);
+                std::thread::spawn(move || {
+                    handle_http(stream, &store, &sock);
+                    counter.fetch_sub(1, Ordering::Relaxed);
+                });
             }
             Err(ref e) if e.kind() == std::io::ErrorKind::WouldBlock => {
                 std::thread::sleep(Duration::from_millis(50));