wizard-proxy: survive fd exhaustion and close leaked tunnels

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"

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

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:#}");
             }

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<()> {

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

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<Bytes, hyper::Error>;
 type Driver = Pin<Box<dyn Future<Output = Result<(), hyper::Error>> + Send>>;
@@ -50,7 +53,10 @@ pub(crate) async fn serve_client(stream: TcpStream, upstream: Arc<Upstream>) {
         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<Upstream>) {
     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<Incoming>,
     upstream: Arc<Upstream>,
@@ -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(

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(())
+}

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/<version> (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<io::Result<()>>,
+    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<Output = ()> + 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<Proxy> {
         .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,29 +110,56 @@ pub async fn start(address: &str, token: &str) -> io::Result<Proxy> {
 
 async fn serve(
     listener: TcpListener,
+    address: SocketAddr,
     upstream: Arc<upstream::Upstream>,
     mut shutdown: oneshot::Receiver<()>,
-) -> io::Result<()> {
+) {
     let mut connections = JoinSet::new();
-    let result = loop {
+    let mut warned_at: Option<Instant> = 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 {
                     Ok((stream, _)) => {
                         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<TcpStream> {
+    let stream = TcpStream::connect(address).await?;
+    SockRef::from(&stream).set_tcp_keepalive(&KEEPALIVE)?;
+    Ok(stream)
 }