fix(proxy): allow concurrent proxy dispatcher streams (#4288)

.env.example

@@ -402,6 +402,12 @@ NEXT_PUBLIC_ENABLE_SOCKS5_PROXY=true
 # ALL_PROXY=socks5://127.0.0.1:7890
 # NO_PROXY=localhost,127.0.0.1
 
+# Max concurrent sockets per cached HTTP/SOCKS proxy dispatcher.
+# Long-lived SSE streams such as Codex /v1/responses need more than one
+# connection when multiple requests share the same account-level proxy.
+# Set to 1 only for legacy diagnostics. Values above 256 are capped.
+# OMNIROUTE_PROXY_DISPATCHER_CONNECTIONS=32
+
 # Proxy fail-open mode (default: false = fail-closed).
 # When false, a request whose assigned proxy fails to resolve is REFUSED rather than
 # falling back to a direct connection — prevents real-IP leaks in egress-controlled

docs/reference/ENVIRONMENT.md

@@ -284,6 +284,7 @@ Route upstream LLM provider calls through an HTTP or SOCKS5 proxy for egress con
 | `HTTPS_PROXY`                           | _(unset)_ | Node.js standard                             | HTTPS proxy for upstream calls.                                                                                                                                                                                                  |
 | `ALL_PROXY`                             | _(unset)_ | Node.js standard                             | Universal proxy (supports `socks5://`).                                                                                                                                                                                          |
 | `NO_PROXY`                              | _(unset)_ | Node.js standard                             | Comma-separated hostnames/IPs to bypass the proxy.                                                                                                                                                                               |
+| `OMNIROUTE_PROXY_DISPATCHER_CONNECTIONS` | `32`      | `open-sse/utils/proxyDispatcher.ts`          | Max concurrent sockets per cached HTTP/SOCKS proxy dispatcher. Long-lived SSE streams such as Codex `/v1/responses` need more than one connection when several requests share the same account-level proxy. Values above `256` are capped. |
 | `PROXY_FAIL_OPEN`                       | `false`   | `src/sse/handlers/chatHelpers.ts`            | When `false` (default), a request whose assigned proxy fails to resolve is **refused (fail-closed)** rather than falling back to a direct connection — prevents real-IP leaks. Set `true` to restore the legacy DIRECT fallback. |
 | `ENABLE_TLS_FINGERPRINT`                | `false`   | `open-sse/executors`                         | Spoof TLS fingerprint using wreq-js (mimics Chrome 124). Counters JA3/JA4 blocking.                                                                                                                                              |
 | `OMNIROUTE_TURNSTILE_IGNORE_TLS_ERRORS` | `false`   | `open-sse/services/claudeTurnstileSolver.ts` | Allow the Claude Turnstile Playwright browser context to ignore HTTPS certificate errors.                                                                                                                                        |

open-sse/utils/proxyDispatcher.ts

@@ -8,6 +8,8 @@ import { createSocksDispatcherWithFamily } from "./socksConnectorWithFamily.ts";
 const DISPATCHER_CACHE_KEY = Symbol.for("omniroute.proxyDispatcher.cache");
 const DEFAULT_DISPATCHER_KEY = Symbol.for("omniroute.proxyDispatcher.default");
 const SUPPORTED_PROTOCOLS = new Set(["http:", "https:", "socks5:"]);
+const DEFAULT_PROXY_DISPATCHER_CONNECTIONS = 32;
+const MAX_PROXY_DISPATCHER_CONNECTIONS = 256;
 
 type DispatcherCache = Map<string, Dispatcher>;
 type GlobalWithDispatcherCache = typeof globalThis & {
@@ -67,14 +69,36 @@ function getDispatcherOptions() {
   };
 }
 
-function getProxyDispatcherOptions() {
+export function getProxyDispatcherConnectionLimit(
+  env: Record<string, string | undefined> = process.env
+): number {
+  const raw = env.OMNIROUTE_PROXY_DISPATCHER_CONNECTIONS;
+  if (raw == null || raw.trim() === "") return DEFAULT_PROXY_DISPATCHER_CONNECTIONS;
+
+  const parsed = Number(raw);
+  if (!Number.isFinite(parsed) || parsed < 1) {
+    console.warn(
+      `[ProxyDispatcher] Invalid OMNIROUTE_PROXY_DISPATCHER_CONNECTIONS="${raw}". Using default ${DEFAULT_PROXY_DISPATCHER_CONNECTIONS}.`
+    );
+    return DEFAULT_PROXY_DISPATCHER_CONNECTIONS;
+  }
+
+  return Math.min(Math.floor(parsed), MAX_PROXY_DISPATCHER_CONNECTIONS);
+}
+
+function getProxyDispatcherOptions(env: Record<string, string | undefined> = process.env) {
   const options = getDispatcherOptions();
   // Disable keep-alive and pipelining for proxy connections.
   // Cheap proxy servers aggressively drop idle sockets without sending TCP RST,
   // causing "socket hang up" or "Client network socket disconnected" errors
   // on subsequent requests that try to reuse the pooled connection.
+  //
+  // Keep multiple connections available anyway: with pipelining disabled, long
+  // SSE streams such as Codex /v1/responses otherwise bottleneck through the
+  // cached proxy dispatcher under concurrency (#4163).
   return {
     ...options,
+    connections: getProxyDispatcherConnectionLimit(env),
     keepAliveTimeout: 1,
     keepAliveMaxTimeout: 1,
     pipelining: 0,
@@ -302,6 +326,13 @@ export function __resolveDispatcherFamilyForTest(proxyUrl: string): 4 | 6 | null
   return resolveDispatcherFamily(new URL(proxyUrl));
 }
 
+/** Test-only accessor for proxy dispatcher pool options. */
+export function __getProxyDispatcherOptionsForTest(
+  env: Record<string, string | undefined> = process.env
+) {
+  return getProxyDispatcherOptions(env);
+}
+
 export function createProxyDispatcher(proxyUrl: string): Dispatcher {
   const normalizedUrl = normalizeProxyUrl(proxyUrl, "proxy dispatcher");
   const dispatcherCache = getDispatcherCache();

tests/unit/proxy-dispatcher-family.test.ts

@@ -1,6 +1,7 @@
 import { describe, it, afterEach } from "node:test";
 import assert from "node:assert/strict";
 import {
+  __getProxyDispatcherOptionsForTest,
   __getSocksOptionsForTest,
   __resolveDispatcherFamilyForTest,
   proxyConfigToUrl,
@@ -53,3 +54,27 @@ describe("proxyDispatcher family marker does not corrupt port", () => {
     assert.ok(out.endsWith("?family=ipv6"), out);
   });
 });
+
+describe("proxyDispatcher connection pool", () => {
+  it("keeps enough proxy connections for concurrent SSE streams by default", () => {
+    const options = __getProxyDispatcherOptionsForTest({});
+    assert.equal(options.connections, 32);
+    assert.equal(options.pipelining, 0);
+    assert.equal(options.keepAliveTimeout, 1);
+    assert.equal(options.keepAliveMaxTimeout, 1);
+  });
+
+  it("allows operators to force a single proxy connection for diagnostics", () => {
+    const options = __getProxyDispatcherOptionsForTest({
+      OMNIROUTE_PROXY_DISPATCHER_CONNECTIONS: "1",
+    });
+    assert.equal(options.connections, 1);
+  });
+
+  it("caps excessive proxy connection overrides", () => {
+    const options = __getProxyDispatcherOptionsForTest({
+      OMNIROUTE_PROXY_DISPATCHER_CONNECTIONS: "9999",
+    });
+    assert.equal(options.connections, 256);
+  });
+});