fix(server): accept loop can never starve — shed load instead of parking

The HTTP accept loop parked on acquire_owned() when the connection
budget drained, freezing accept() for every client (the .228
session-flapping / CLOSE-WAIT signature). Permits drained because
half-open clients and hung upstreams held them indefinitely.

- try_acquire_owned + immediate 503-and-close when the budget is
  exhausted; the accept loop itself never blocks
- 30s http1_header_read_timeout drops slowloris/half-open clients
- 900s watchdog bounds non-upgraded connections; websocket upgrades
  are exempt (legitimately long-lived)

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

core/archipelago/src/server.rs

@@ -1322,11 +1322,46 @@ async fn accept_loop(
                     }
                 };
                 let handler = handler.clone();
-                let permit = active_connections.clone().acquire_owned().await;
+                // NEVER park the accept loop on the connection budget.
+                // `acquire_owned().await` here froze accept() entirely when
+                // permits drained — and permits drained because half-open
+                // clients and hung upstreams held them forever (the .228
+                // session-flapping / CLOSE-WAIT `inode: 0` signature). Shed
+                // load instead: accept, answer 503, close.
+                let permit = match active_connections.clone().try_acquire_owned() {
+                    Ok(p) => p,
+                    Err(_) => {
+                        warn!(
+                            "{} connection budget exhausted — shedding {}",
+                            local_addr, peer_addr
+                        );
+                        tokio::spawn(async move {
+                            use tokio::io::AsyncWriteExt;
+                            let mut stream = stream;
+                            let _ = tokio::time::timeout(
+                                std::time::Duration::from_secs(5),
+                                stream.write_all(
+                                    b"HTTP/1.1 503 Service Unavailable\r\nConnection: close\r\nContent-Length: 0\r\n\r\n",
+                                ),
+                            )
+                            .await;
+                            let _ = stream.shutdown().await;
+                        });
+                        continue;
+                    }
+                };
                 tokio::spawn(async move {
                     let _permit = permit;
+                    // Set when a request carries an Upgrade header (websocket):
+                    // upgraded connections are legitimately long-lived and are
+                    // exempt from the non-upgraded connection deadline below.
+                    let upgraded = Arc::new(std::sync::atomic::AtomicBool::new(false));
+                    let upgraded_flag = upgraded.clone();
                     let service = service_fn(move |mut req: hyper::Request<hyper::Body>| {
                         let handler = handler.clone();
+                        if req.headers().contains_key(hyper::header::UPGRADE) {
+                            upgraded_flag.store(true, std::sync::atomic::Ordering::Relaxed);
+                        }
                         async move {
                             // Record the TCP peer so rate limiting only trusts
                             // forwarded headers on loopback (nginx) connections.
@@ -1345,13 +1380,48 @@ async fn accept_loop(
                                 .map_err(|e| std::io::Error::other(format!("{}", e)))
                         }
                     });
-                    if let Err(e) = Http::new()
+                    // header_read_timeout: a client that connects and never
+                    // sends a request (slowloris / half-open) is dropped
+                    // instead of holding a permit until the heat death of the
+                    // node. Long RPCs are safe — the clock only covers header
+                    // read.
+                    let conn = Http::new()
                         .http1_keep_alive(false)
+                        .http1_header_read_timeout(std::time::Duration::from_secs(30))
                         .serve_connection(stream, service)
-                        .with_upgrades()
-                        .await
-                    {
-                        error!("Error serving connection from {}: {}", peer_addr, e);
+                        .with_upgrades();
+                    tokio::pin!(conn);
+                    // Deadline watchdog for NON-upgraded connections. With
+                    // keep-alive off a plain connection serves one exchange;
+                    // 15 min bounds even the slowest legitimate RPC/stream
+                    // while guaranteeing a hung upstream can't hold a permit
+                    // forever. Upgraded (websocket) connections are exempt.
+                    const NON_UPGRADED_DEADLINE: std::time::Duration =
+                        std::time::Duration::from_secs(900);
+                    let started = std::time::Instant::now();
+                    let watchdog = async {
+                        loop {
+                            tokio::time::sleep(std::time::Duration::from_secs(30)).await;
+                            if !upgraded.load(std::sync::atomic::Ordering::Relaxed)
+                                && started.elapsed() >= NON_UPGRADED_DEADLINE
+                            {
+                                return;
+                            }
+                        }
+                    };
+                    tokio::select! {
+                        r = &mut conn => {
+                            if let Err(e) = r {
+                                error!("Error serving connection from {}: {}", peer_addr, e);
+                            }
+                        }
+                        _ = watchdog => {
+                            warn!(
+                                "connection from {} exceeded {}s without completing or upgrading — dropping",
+                                peer_addr,
+                                NON_UPGRADED_DEADLINE.as_secs()
+                            );
+                        }
                     }
                 });
             }