diff --git a/core/archipelago/src/server.rs b/core/archipelago/src/server.rs index fbd0047c..ecc04456 100644 --- a/core/archipelago/src/server.rs +++ b/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| { 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() + ); + } } }); }