From cf6bdf8560378ddd576867c94d379e48ff7b0a7d Mon Sep 17 00:00:00 2001 From: Sergei Razmetov Date: Thu, 29 Oct 2026 00:47:52 +0300 Subject: [PATCH] wizard-proxy: survive fd exhaustion and close leaked tunnels - Retry failed accepts instead of stopping the listener (EMFILE killed it for good). - Raise the Unix soft open-file limit to min(10240, hard); macOS GUI apps start at 256. - Close a tunnel 30 s after one side's clean close (a FIN_WAIT_2 leak seen in the field). - TCP keepalive on outgoing sockets; hyper's 30 s header read timeout. --- Cargo.lock | 3 ++ Cargo.toml | 1 + crates/wizard-gui/Cargo.toml | 3 ++ crates/wizard-gui/src/lib.rs | 2 + crates/wizard-gui/src/proxy.rs | 38 ++++++++++++++++ crates/wizard-proxy/Cargo.toml | 2 + crates/wizard-proxy/src/http.rs | 50 ++++++++++++++++----- crates/wizard-proxy/src/http_test.rs | 37 ++++++++++++++- crates/wizard-proxy/src/lib.rs | 67 +++++++++++++++++++++++----- crates/wizard-proxy/src/upstream.rs | 2 +- 10 files changed, 180 insertions(+), 25 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 3404455..f918c45 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5429,6 +5429,7 @@ name = "wizard-gui" version = "0.2.8" dependencies = [ "anyhow", + "libc", "log", "serde", "serde_json", @@ -5452,7 +5453,9 @@ dependencies = [ "http-body-util", "hyper", "hyper-util", + "log", "rustls-native-certs", + "socket2", "tokio", "tokio-rustls", "url", diff --git a/Cargo.toml b/Cargo.toml index b026c98..f364eb7 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -62,6 +62,7 @@ semver = "1.0.28" sha2 = "0.10.9" serde = { version = "1", features = ["derive"] } serde_json = "1.0.151" +socket2 = "0.6.5" tauri = "2.11.5" tauri-build = "2.6.3" tauri-plugin-autostart = "2.5.1" diff --git a/crates/wizard-gui/Cargo.toml b/crates/wizard-gui/Cargo.toml index 9adc961..b082788 100644 --- a/crates/wizard-gui/Cargo.toml +++ b/crates/wizard-gui/Cargo.toml @@ -36,6 +36,9 @@ wizard-core.workspace = true wizard-proxy.workspace = true wizard-settings.workspace = true +[target.'cfg(unix)'.dependencies] +libc.workspace = true + [target.'cfg(any(target_os = "macos", target_os = "windows"))'.dependencies] tauri = { workspace = true, features = ["tray-icon"] } tauri-plugin-autostart.workspace = true diff --git a/crates/wizard-gui/src/lib.rs b/crates/wizard-gui/src/lib.rs index ecd1965..332c945 100644 --- a/crates/wizard-gui/src/lib.rs +++ b/crates/wizard-gui/src/lib.rs @@ -98,6 +98,8 @@ pub fn run_application() -> Result<()> { #[cfg(target_os = "linux")] let tray_enabled = false; + #[cfg(unix)] + proxy::raise_open_file_limit(); if let Err(error) = tauri::async_runtime::block_on(proxy::sync(app.handle())) { log::error!("proxy startup failed: {error:#}"); } diff --git a/crates/wizard-gui/src/proxy.rs b/crates/wizard-gui/src/proxy.rs index b48e20a..0184e37 100644 --- a/crates/wizard-gui/src/proxy.rs +++ b/crates/wizard-gui/src/proxy.rs @@ -94,6 +94,44 @@ fn publish(app: &tauri::AppHandle, proxy_app: App, running: bool) { }); } +/// macOS starts GUI apps with a soft limit of 256 descriptors, and each tunnel holds two; +/// as the system proxy, the listeners would run out. Never lowers the limit. +#[cfg(unix)] +#[expect( + unsafe_code, + reason = "std has no safe bindings for getrlimit and setrlimit" +)] +pub(crate) fn raise_open_file_limit() { + // macOS OPEN_MAX: setrlimit(2) there recommends min(OPEN_MAX, rlim_max) for RLIMIT_NOFILE. + const TARGET: libc::rlim_t = 10240; + + let mut limit = libc::rlimit { + rlim_cur: 0, + rlim_max: 0, + }; + // SAFETY: `limit` is a valid, writable rlimit. + if unsafe { libc::getrlimit(libc::RLIMIT_NOFILE, &raw mut limit) } != 0 { + let error = std::io::Error::last_os_error(); + log::warn!("failed to read open file limit: {error}"); + return; + } + let previous = limit.rlim_cur; + limit.rlim_cur = TARGET.min(limit.rlim_max); + if limit.rlim_cur <= previous { + return; + } + // SAFETY: `limit` is a valid rlimit whose soft limit does not exceed the hard one. + if unsafe { libc::setrlimit(libc::RLIMIT_NOFILE, &raw const limit) } != 0 { + let error = std::io::Error::last_os_error(); + log::warn!("failed to raise open file limit from {previous}: {error}"); + return; + } + log::info!( + "open file limit raised from {previous} to {}", + limit.rlim_cur + ); +} + /// Called at startup and after a change of either app's proxy setting. Each listener runs while /// its app has a token and is restarted when that token changes; one failure does not skip the other. pub(crate) async fn sync(app: &tauri::AppHandle) -> Result<()> { diff --git a/crates/wizard-proxy/Cargo.toml b/crates/wizard-proxy/Cargo.toml index 2d4b516..4dae19e 100644 --- a/crates/wizard-proxy/Cargo.toml +++ b/crates/wizard-proxy/Cargo.toml @@ -15,7 +15,9 @@ base64.workspace = true http-body-util.workspace = true hyper = { workspace = true, features = ["client", "server", "http1"] } hyper-util = { workspace = true, features = ["tokio"] } +log.workspace = true rustls-native-certs.workspace = true +socket2.workspace = true tokio = { workspace = true, features = ["rt", "net", "io-util", "macros"] } tokio-rustls.workspace = true url.workspace = true diff --git a/crates/wizard-proxy/src/http.rs b/crates/wizard-proxy/src/http.rs index b84167b..40fe237 100644 --- a/crates/wizard-proxy/src/http.rs +++ b/crates/wizard-proxy/src/http.rs @@ -14,9 +14,9 @@ use hyper::http::uri::Authority; use hyper::service::service_fn; use hyper::upgrade::{OnUpgrade, Upgraded}; use hyper::{HeaderMap, Method, Request, Response, StatusCode, Uri}; -use hyper_util::rt::TokioIo; +use hyper_util::rt::{TokioIo, TokioTimer}; use std::io; -use tokio::io::{AsyncRead, AsyncWrite, copy_bidirectional}; +use tokio::io::{AsyncRead, AsyncWrite, AsyncWriteExt as _}; use tokio::net::TcpStream; use tokio::sync::mpsc; use tokio::time::timeout; @@ -26,6 +26,9 @@ use tokio::time::timeout; mod tests; const CONNECT_TIMEOUT: Duration = Duration::from_secs(30); +/// After one side of a tunnel closes cleanly, the other may finish for this long. A client +/// that half-closes and then vanishes would otherwise leave a silent remote open forever. +const HALF_CLOSE_TIMEOUT: Duration = Duration::from_secs(if cfg!(test) { 1 } else { 30 }); type Body = UnsyncBoxBody; type Driver = Pin> + Send>>; @@ -50,7 +53,10 @@ pub(crate) async fn serve_client(stream: TcpStream, upstream: Arc) { async move { Ok::<_, Infallible>(handle(request, upstream, driver_sender, tunnel_sender).await) } }); let mut builder = hyper::server::conn::http1::Builder::new(); - builder.keep_alive(false); + builder + .keep_alive(false) + // Enables hyper's default 30 s timeout for clients that never send a request head. + .timer(TokioTimer::new()); let connection = builder .serve_connection(TokioIo::new(stream), service) .with_upgrades(); @@ -74,19 +80,39 @@ pub(crate) async fn serve_client(stream: TcpStream, upstream: Arc) { if let Ok(tunnel) = tunnels.try_recv() && let Ok(client) = tunnel.client.await { - // TLS and application protocols inside CONNECT are opaque bytes. - let mut client = TokioIo::new(client); + let client = TokioIo::new(client); match tunnel.remote { - Remote::Direct(mut remote) => { - let _ = copy_bidirectional(&mut client, &mut remote).await; - } - Remote::Upstream(remote) => { - let _ = copy_bidirectional(&mut client, &mut TokioIo::new(remote)).await; - } + Remote::Direct(remote) => relay(client, remote).await, + Remote::Upstream(remote) => relay(client, TokioIo::new(remote)).await, } } } +/// Copy both directions until both close; an error, or [`HALF_CLOSE_TIMEOUT`] after +/// the first clean close, drops both sides. TLS and application protocols inside +/// CONNECT are opaque bytes. +async fn relay(client: impl AsyncRead + AsyncWrite, remote: impl AsyncRead + AsyncWrite) { + let (mut client_read, mut client_write) = tokio::io::split(client); + let (mut remote_read, mut remote_write) = tokio::io::split(remote); + let upload = async { + tokio::io::copy(&mut client_read, &mut remote_write).await?; + remote_write.shutdown().await + }; + let download = async { + tokio::io::copy(&mut remote_read, &mut client_write).await?; + client_write.shutdown().await + }; + tokio::pin!(upload, download); + tokio::select! { + result = &mut upload => if result.is_ok() { + let _ = timeout(HALF_CLOSE_TIMEOUT, download).await; + }, + result = &mut download => if result.is_ok() { + let _ = timeout(HALF_CLOSE_TIMEOUT, upload).await; + }, + } +} + async fn handle( mut request: Request, upstream: Arc, @@ -223,7 +249,7 @@ async fn connect_direct(authority: &Authority, default_port: u16) -> io::Result< .strip_prefix('[') .and_then(|host| host.strip_suffix(']')) .unwrap_or(host); - TcpStream::connect((host, authority.port_u16().unwrap_or(default_port))).await + crate::connect((host, authority.port_u16().unwrap_or(default_port))).await } async fn connect_remote( diff --git a/crates/wizard-proxy/src/http_test.rs b/crates/wizard-proxy/src/http_test.rs index b7cc338..69b5c36 100644 --- a/crates/wizard-proxy/src/http_test.rs +++ b/crates/wizard-proxy/src/http_test.rs @@ -1,6 +1,8 @@ use crate::{Proxy, serve, upstream}; use alloc::sync::Arc; use core::{net::SocketAddr, time::Duration}; +use hyper::http::uri::Authority; +use socket2::SockRef; use std::io; use tokio::io::{AsyncRead, AsyncReadExt as _, AsyncWriteExt as _}; use tokio::net::{TcpListener, TcpStream}; @@ -28,7 +30,7 @@ async fn start_proxy() -> io::Result<(SocketAddr, Proxy)> { let (finished, stopped) = watch::channel(()); let task = tokio::spawn(async move { let _finished = finished; - serve(listener, upstream, receiver).await + serve(listener, address, upstream, receiver).await; }); Ok(( address, @@ -184,6 +186,29 @@ async fn direct_connect_preserves_buffered_bytes_and_half_close() -> io::Result< }).await.expect("direct CONNECT timed out") } +#[tokio::test] +async fn half_closed_tunnel_to_silent_remote_closes_after_timeout() -> io::Result<()> { + timeout(Duration::from_secs(5), async { + let origin = TcpListener::bind("127.0.0.1:0").await?; + let target = origin.local_addr()?; + let (address, proxy) = start_proxy().await?; + let mut client = TcpStream::connect(address).await?; + client + .write_all(format!("CONNECT {target} HTTP/1.1\r\nHost: {target}\r\n\r\n").as_bytes()) + .await?; + // The remote stays open and silent after the client's FIN. + let (mut remote, _) = origin.accept().await?; + assert!(read_head(&mut client).await?.starts_with("HTTP/1.1 200 ")); + client.shutdown().await?; + let mut byte = [0]; + assert_eq!(remote.read(&mut byte).await?, 0); + assert_eq!(client.read(&mut byte).await?, 0); + proxy.stop().await + }) + .await + .expect("half-closed tunnel stayed open") +} + #[tokio::test] async fn stopping_proxy_closes_both_direct_tunnel_sockets() -> io::Result<()> { timeout(Duration::from_secs(5), async { @@ -249,3 +274,13 @@ async fn malformed_destination_is_rejected_before_dialing() -> io::Result<()> { proxy.stop().await }).await.expect("invalid destination timed out") } + +#[tokio::test] +async fn direct_connections_use_tcp_keepalive() -> io::Result<()> { + let origin = TcpListener::bind("127.0.0.1:0").await?; + let authority = + Authority::try_from(origin.local_addr()?.to_string()).map_err(io::Error::other)?; + let stream = super::connect_direct(&authority, 80).await?; + assert!(SockRef::from(&stream).keepalive()?); + Ok(()) +} diff --git a/crates/wizard-proxy/src/lib.rs b/crates/wizard-proxy/src/lib.rs index d629cdd..b606ec0 100644 --- a/crates/wizard-proxy/src/lib.rs +++ b/crates/wizard-proxy/src/lib.rs @@ -3,14 +3,20 @@ //! return 405 with Allow: CONNECT); unlisted ports return 403 for every method. //! Unmatched domains and explicit IP addresses connect directly, without proxy credentials. //! Malformed destinations are rejected before opening a connection. +//! Failed accepts, e.g. while file descriptors are exhausted, are retried: the listener +//! stops only on request. Outgoing sockets use TCP keepalive, so a peer that vanishes +//! without closing does not hold a tunnel open forever; neither does a half-closed tunnel. //! Dropping the handle requests shutdown; use [`Proxy::stop`] to wait for all sockets to close. extern crate alloc; use alloc::sync::Arc; +use core::{net::SocketAddr, time::Duration}; +use socket2::{SockRef, TcpKeepalive}; use std::env::consts::{ARCH, OS}; use std::io; -use tokio::net::TcpListener; +use std::time::Instant; +use tokio::net::{TcpListener, TcpStream, ToSocketAddrs}; use tokio::sync::{oneshot, watch}; use tokio::task::{JoinHandle, JoinSet}; @@ -25,6 +31,17 @@ pub const CLAUDE_DESKTOP_ADDRESS: &str = "127.0.0.1:33787"; /// Listener of `ChatGPT`, with its own token. pub const CHATGPT_ADDRESS: &str = "127.0.0.1:33788"; +/// While descriptors are exhausted, an immediate retry would fail again at once. +const ACCEPT_RETRY_DELAY: Duration = Duration::from_millis(100); +/// Accept failures are logged at most this often, so they cannot rotate older entries out of the log. +const ACCEPT_WARNING_INTERVAL: Duration = Duration::from_secs(60); + +/// Detects a dead peer within about three minutes. Tunnels have no idle timeout, +/// because idle WebSocket and streaming connections are legitimate. +const KEEPALIVE: TcpKeepalive = TcpKeepalive::new() + .with_time(Duration::from_secs(60)) + .with_interval(Duration::from_secs(15)); + /// `User-Agent` of requests Wizard itself sends, e.g. `CoreInfraAI-Wizard/ (macos; aarch64)`. /// OS and architecture are those of the running binary. #[must_use] @@ -38,12 +55,12 @@ pub fn user_agent() -> String { /// Owns a running listener. No settings are read or watched by this crate. pub struct Proxy { shutdown: oneshot::Sender<()>, - task: JoinHandle>, + task: JoinHandle<()>, stopped: watch::Receiver<()>, } impl Proxy { - /// Wait for the listener task to finish, including failure, panic or cancellation. + /// Wait for the listener task to finish, including panic or cancellation. /// The returned future does not borrow the handle and can be observed independently. pub fn stopped(&self) -> impl Future + Send + 'static + use<> { let mut stopped = self.stopped.clone(); @@ -55,7 +72,7 @@ impl Proxy { /// Stop accepting requests, cancel active connections, and wait for their cleanup. pub async fn stop(self) -> io::Result<()> { drop(self.shutdown); - self.task.await.map_err(io::Error::other)? + self.task.await.map_err(io::Error::other) } } @@ -76,12 +93,13 @@ pub async fn start(address: &str, token: &str) -> io::Result { .map_err(io::Error::other)??; let upstream = Arc::new(upstream::Upstream::new(tls, authorization)); let listener = TcpListener::bind(address).await?; + let address = listener.local_addr()?; let (shutdown, receiver) = oneshot::channel(); let (finished, stopped) = watch::channel(()); let task = tokio::spawn(async move { // Dropping the only sender wakes observers on every task-exit path. let _finished = finished; - serve(listener, upstream, receiver).await + serve(listener, address, upstream, receiver).await; }); Ok(Proxy { shutdown, @@ -92,14 +110,16 @@ pub async fn start(address: &str, token: &str) -> io::Result { async fn serve( listener: TcpListener, + address: SocketAddr, upstream: Arc, mut shutdown: oneshot::Receiver<()>, -) -> io::Result<()> { +) { let mut connections = JoinSet::new(); - let result = loop { + let mut warned_at: Option = None; + loop { tokio::select! { biased; - _ = &mut shutdown => break Ok(()), + _ = &mut shutdown => break, Some(_) = connections.join_next(), if !connections.is_empty() => {}, accepted = listener.accept() => { match accepted { @@ -107,14 +127,39 @@ async fn serve( let upstream = Arc::clone(&upstream); connections.spawn(http::serve_client(stream, upstream)); } - Err(error) => break Err(error), + // Only this client is gone; the listener is unaffected. + Err(error) + if matches!( + error.kind(), + io::ErrorKind::ConnectionAborted | io::ErrorKind::ConnectionReset + ) => {} + // Typically EMFILE: closing tunnels will free descriptors. + Err(error) => { + if warned_at.is_none_or(|at| at.elapsed() >= ACCEPT_WARNING_INTERVAL) { + warned_at = Some(Instant::now()); + log::warn!( + "local proxy on {address} cannot accept connections, retrying: {error}" + ); + } + tokio::select! { + biased; + _ = &mut shutdown => break, + () = tokio::time::sleep(ACCEPT_RETRY_DELAY) => {}, + } + } } } } - }; + } drop(listener); connections.abort_all(); while connections.join_next().await.is_some() {} - result +} + +/// Connect to a remote host, enabling [`KEEPALIVE`] on the socket. +async fn connect(address: impl ToSocketAddrs) -> io::Result { + let stream = TcpStream::connect(address).await?; + SockRef::from(&stream).set_tcp_keepalive(&KEEPALIVE)?; + Ok(stream) } diff --git a/crates/wizard-proxy/src/upstream.rs b/crates/wizard-proxy/src/upstream.rs index b56d462..bfb1bd9 100644 --- a/crates/wizard-proxy/src/upstream.rs +++ b/crates/wizard-proxy/src/upstream.rs @@ -27,7 +27,7 @@ impl Upstream { pub(crate) async fn connect(&self) -> io::Result> { // Deliberately bypass HTTP_PROXY/HTTPS_PROXY: selected destinations must // use this upstream, never an environment-provided proxy or direct fallback. - let stream = TcpStream::connect((HOST, PORT)).await?; + let stream = crate::connect((HOST, PORT)).await?; let server_name = ServerName::try_from(HOST).map_err(io::Error::other)?; self.connector.connect(server_name, stream).await }