From 133432b523df28ddf5a099bf8d2bfb2f6f995530 Mon Sep 17 00:00:00 2001 From: Wilson Date: Fri, 19 Jun 2026 17:39:57 -0300 Subject: [PATCH] fix(proxy): allow concurrent proxy dispatcher streams (#4288) --- .env.example | 6 ++++ docs/reference/ENVIRONMENT.md | 1 + open-sse/utils/proxyDispatcher.ts | 33 +++++++++++++++++++++- tests/unit/proxy-dispatcher-family.test.ts | 25 ++++++++++++++++ 4 files changed, 64 insertions(+), 1 deletion(-) diff --git a/.env.example b/.env.example index 3941efae2e..4d8a72bc4c 100644 --- a/.env.example +++ b/.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 diff --git a/docs/reference/ENVIRONMENT.md b/docs/reference/ENVIRONMENT.md index b536aa6b5d..632e940a33 100644 --- a/docs/reference/ENVIRONMENT.md +++ b/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. | diff --git a/open-sse/utils/proxyDispatcher.ts b/open-sse/utils/proxyDispatcher.ts index 34623d4ff8..94d88958c7 100644 --- a/open-sse/utils/proxyDispatcher.ts +++ b/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; type GlobalWithDispatcherCache = typeof globalThis & { @@ -67,14 +69,36 @@ function getDispatcherOptions() { }; } -function getProxyDispatcherOptions() { +export function getProxyDispatcherConnectionLimit( + env: Record = 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 = 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 = process.env +) { + return getProxyDispatcherOptions(env); +} + export function createProxyDispatcher(proxyUrl: string): Dispatcher { const normalizedUrl = normalizeProxyUrl(proxyUrl, "proxy dispatcher"); const dispatcherCache = getDispatcherCache(); diff --git a/tests/unit/proxy-dispatcher-family.test.ts b/tests/unit/proxy-dispatcher-family.test.ts index 9546ac2ccd..db0cb8e099 100644 --- a/tests/unit/proxy-dispatcher-family.test.ts +++ b/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); + }); +});