fix(proxy): resolve active tunnel growth under load by capping concurrent connections

Author: hightemp

.ai-factory/patches/2026-04-07-18.46.md

@@ -0,0 +1,43 @@
+# Active tunnel growth fix
+
+**Date:** 2026-04-07 18:46
+**Files:** `ProxyServer.kt`, `TASKS.md`
+**Severity:** high
+
+## Problem
+
+On a real device, the Statistics screen showed active/open tunnels growing after YouTube was closed. Device logcat confirmed the proxy hit `IO thread pool full (256 threads)` around 135 active connections, after which pipe tasks were rejected.
+
+## Root Cause
+
+`ProxyServer.tunnel()` scheduled both tunnel directions as independent IO executor tasks after the client handler had already consumed one executor thread. Under load this required roughly three worker tasks per tunnel. When the executor rejected one of the pipe tasks, the active tunnel slot could remain counted until timeout or cleanup.
+
+## Solution
+
+- Run the client-to-upstream pipe in the current client handler worker.
+- Schedule only the upstream-to-client pipe as an extra worker task.
+- Add an explicit active tunnel limit with reserve capacity for accept and handshake work.
+- Reserve the tunnel slot before upstream connection and before returning `200 Connection Established`.
+- Return `503 Service Unavailable` when the active tunnel limit is reached.
+- Release tunnel slots via a guarded decrement so counters cannot go negative.
+
+## Verification
+
+- `GRADLE_USER_HOME=/tmp/gradle ./gradlew testDebugUnitTest lintDebug`
+- `GRADLE_USER_HOME=/tmp/gradle ./gradlew installDebug`
+- Started the app on `SM-A165F - 15` through the UI via ADB.
+- Sent 120 short CONNECT requests through an ADB-forwarded local port: all completed and active tunnels dropped back down.
+- Sent 130 concurrent CONNECT requests: 96 were accepted, 34 were rejected by the explicit tunnel limit, and logcat did not show `IO thread pool full`.
+- Confirmed the Statistics screen showed `Peak = 96` and active tunnels did not keep growing after the test sockets were closed.
+- Re-installed the final APK, selected a saved HTTPS upstream proxy, and verified 5 CONNECT requests completed through the upstream path.
+- Stopped the proxy and verified system proxy was restored to `:0` and `127.0.0.1:8080` no longer listened.
+
+## Prevention
+
+- Do not schedule both directions of a long-lived tunnel without reserving executor capacity.
+- Apply backpressure before sending success responses to clients.
+- Treat rejected executor tasks as a cleanup path, not just a log event.
+
+## Tags
+
+`#android` `#proxy` `#sockets` `#threading` `#statistics` `#adb`

TASKS.md

@@ -47,3 +47,9 @@ Fix the review findings from the April 7, 2026 project audit.
   - [x] Track uptime, active tunnels, peak active tunnels, total tunnels, and failures
   - [x] Add a Statistics screen and Home toolbar entry
   - [x] Verify statistics update on device through ADB
+- [x] Fix active tunnel growth under load
+  - [x] Confirm thread-pool saturation from device logcat
+  - [x] Run one pipe direction in the client worker instead of scheduling both directions separately
+  - [x] Cap active tunnels before returning `200 Connection Established`
+  - [x] Close sockets and record failures when the active tunnel limit is reached
+  - [x] Verify short-lived and overloaded CONNECT traffic through ADB

app/src/main/java/com/hightemp/proxy_switcher/proxy/ProxyServer.kt

@@ -32,6 +32,10 @@ class ProxyServer {
         private const val BACKLOG = 128
         // Max concurrent IO threads — prevents OOM from unbounded thread/buffer creation
         private const val MAX_IO_THREADS = 256
+        // Each active tunnel needs one client handler plus one reverse pipe task.
+        // Keep a reserve for accept/handshake work so thread-pool saturation cannot
+        // leak tunnel counters or leave sockets half-managed under heavy app traffic.
+        private const val MAX_ACTIVE_TUNNELS = 96
         // Idle timeout for tunnel sockets (ms). Detects half-open connections
         // that would otherwise block a thread forever. 3 minutes — long enough
         // for paused video streams, short enough to reclaim leaked threads.
@@ -220,7 +224,14 @@ class ProxyServer {
         val targetPort = if (colon > 0) url.substring(colon + 1).toIntOrNull() ?: 443 else 443
 
         var upstream: ConnectedUpstream? = null
+        var tunnelSlotAcquired = false
         try {
+            if (!tryAcquireTunnelSlot()) {
+                rejectTunnel(clientSocket, sendHttpError = true)
+                return
+            }
+            tunnelSlotAcquired = true
+
             upstream = connectToUpstream(targetHost, targetPort, proxy)
 
             // Send 200 Connection Established to client
@@ -235,8 +246,10 @@ class ProxyServer {
             }
 
             tunnel(clientSocket, upstream)
+            tunnelSlotAcquired = false
 
         } catch (e: Exception) {
+            if (tunnelSlotAcquired) releaseTunnelSlot()
             ProxyStats.recordFailure()
             AppLogger.error("ProxyServer", "HTTPS Connect failed: $url. Error: ${e.message}", e)
             upstream?.socket?.let { closeTrackedSocket(it) }
@@ -257,7 +270,14 @@ class ProxyServer {
         proxy: ProxyEntity?
     ) {
         var upstream: ConnectedUpstream? = null
+        var tunnelSlotAcquired = false
         try {
+            if (!tryAcquireTunnelSlot()) {
+                rejectTunnel(clientSocket, sendHttpError = true)
+                return
+            }
+            tunnelSlotAcquired = true
+
             val url = if (urlStr.startsWith("http")) URL(urlStr) else URL("http://$urlStr")
             val host = url.host
             val port = if (url.port != -1) url.port else 80
@@ -307,8 +327,10 @@ class ProxyServer {
             ProxyStats.recordUpload(bytesToWrite)
 
             tunnel(clientSocket, upstream)
+            tunnelSlotAcquired = false
 
         } catch (e: Exception) {
+            if (tunnelSlotAcquired) releaseTunnelSlot()
             ProxyStats.recordFailure()
             AppLogger.error("ProxyServer", "HTTP Request failed: $urlStr", e)
             upstream?.socket?.let { closeTrackedSocket(it) }
@@ -484,39 +506,33 @@ class ProxyServer {
     /**
      * Bidirectional tunnel between [client] and [server].
      *
-     * Each direction runs on its own dedicated daemon thread rather than a coroutine.
+     * The client→server direction runs in the current client handler worker, while
+     * server→client runs on one extra worker.
      * Reason: Dispatchers.IO has a shared pool capped at 64 threads. In a tight
      * read→write loop that never suspends, coroutines still go through the scheduler
-     * on each iteration, adding latency jitter. A daemon thread blocks on read() and
-     * writes immediately — zero scheduler overhead, maximising throughput per connection.
+     * on each iteration, adding latency jitter. Blocking worker threads maximise
+     * throughput, but tunnel concurrency is capped so the pool keeps capacity for
+     * accept and handshake work.
      */
     private fun tunnel(client: Socket, upstream: ConnectedUpstream) {
         // Apply idle timeout to both sockets so read() throws SocketTimeoutException
         // instead of blocking forever on half-open connections.
         try { client.soTimeout = TUNNEL_SO_TIMEOUT_MS } catch (_: Exception) {}
         try { upstream.socket.soTimeout = TUNNEL_SO_TIMEOUT_MS } catch (_: Exception) {}
 
-        activeConnections.incrementAndGet()
         ProxyStats.recordConnectionOpened()
         val closed = AtomicBoolean(false)
         trackSocket(upstream.socket)
         fun closeBoth() {
             if (closed.compareAndSet(false, true)) {
-                activeConnections.decrementAndGet()
+                releaseTunnelSlot()
                 ProxyStats.recordConnectionClosed()
                 closeTrackedSocket(client)
                 closeTrackedSocket(upstream.socket)
             }
         }
 
-        daemonThread("c→s") {
-            try {
-                pipe(client.getInputStream(), upstream.output, ProxyStats::recordUpload)
-            } finally {
-                closeBoth()
-            }
-        }
-        daemonThread("s→c") {
+        val serverToClientStarted = daemonThread("s→c") {
             try {
                 // upstream.input may be a BufferedInputStream that already holds pre-read bytes
                 // from the SOCKS5/HTTP handshake — using it here ensures those bytes reach the client.
@@ -525,6 +541,48 @@ class ProxyServer {
                 closeBoth()
             }
         }
+
+        if (!serverToClientStarted) {
+            ProxyStats.recordFailure()
+            closeBoth()
+            return
+        }
+
+        try {
+            pipe(client.getInputStream(), upstream.output, ProxyStats::recordUpload)
+        } finally {
+            closeBoth()
+        }
+    }
+
+    private fun tryAcquireTunnelSlot(): Boolean {
+        while (true) {
+            val current = activeConnections.get()
+            if (current >= MAX_ACTIVE_TUNNELS) return false
+            if (activeConnections.compareAndSet(current, current + 1)) return true
+        }
+    }
+
+    private fun releaseTunnelSlot() {
+        while (true) {
+            val current = activeConnections.get()
+            if (current <= 0) return
+            if (activeConnections.compareAndSet(current, current - 1)) return
+        }
+    }
+
+    private fun rejectTunnel(client: Socket, sendHttpError: Boolean) {
+        ProxyStats.recordFailure()
+        AppLogger.log("ProxyServer", "Tunnel rejected: active tunnel limit reached ($MAX_ACTIVE_TUNNELS)")
+        if (sendHttpError) {
+            try {
+                client.getOutputStream().write("HTTP/1.1 503 Service Unavailable\r\n\r\n".toByteArray())
+                client.getOutputStream().flush()
+            } catch (_: Exception) {
+                // Client may already be gone.
+            }
+        }
+        closeTrackedSocket(client)
     }
 
     /** Saturates the pipe: reads up to [TUNNEL_BUF] bytes and writes immediately. */
@@ -557,8 +615,8 @@ class ProxyServer {
         }
     }
 
-    private inline fun daemonThread(name: String, crossinline block: () -> Unit) {
-        runOnIo(name) {
+    private inline fun daemonThread(name: String, crossinline block: () -> Unit): Boolean {
+        return runOnIo(name) {
             try {
                 block()
             } catch (e: Throwable) {