From 7e2fdac9ba57f002ae1cc14518285fed2f168b2a Mon Sep 17 00:00:00 2001 From: Makcim Ivanov Date: Tue, 23 Jun 2026 13:57:38 +0000 Subject: [PATCH] fix(proxy): fan out direct dispatcher streams --- open-sse/utils/proxyDispatcher.ts | 82 +++++++++++++++++-- .../direct-dispatcher-pipelining-4580.test.ts | 33 +++++++- 2 files changed, 107 insertions(+), 8 deletions(-) diff --git a/open-sse/utils/proxyDispatcher.ts b/open-sse/utils/proxyDispatcher.ts index a557e058ea04..e7871722530f 100644 --- a/open-sse/utils/proxyDispatcher.ts +++ b/open-sse/utils/proxyDispatcher.ts @@ -43,6 +43,58 @@ type ProxyConfigObject = { family?: string; }; +/** + * Direct upstream fan-out dispatcher. + * + * A single Undici Agent configured with `connections > 1` should be enough in + * theory, but real Codex `/backend-api/codex/responses` streams on Node 24 have + * still been observed queuing every subsequent same-origin request until the + * previous stream emits trailers. Using several one-connection Agents gives + * each long SSE stream an independent pool/client and prevents one stream from + * monopolizing the effective queue while keeping pipelining disabled. + */ +class RoundRobinDispatcher { + private readonly dispatchers: Dispatcher[]; + private nextIndex = 0; + + constructor(dispatchers: Dispatcher[]) { + this.dispatchers = dispatchers; + } + + dispatch(options: Dispatcher.DispatchOptions, handler: Dispatcher.DispatchHandler): boolean { + const dispatcher = this.dispatchers[this.nextIndex % this.dispatchers.length]; + this.nextIndex = (this.nextIndex + 1) % this.dispatchers.length; + return dispatcher.dispatch(options, handler); + } + + close(callback?: () => void): Promise | void { + const done = Promise.all(this.dispatchers.map((dispatcher) => dispatcher.close())).then( + () => undefined + ); + if (callback) { + done.then(callback); + return; + } + return done; + } + + destroy( + errorOrCallback?: Error | null | (() => void), + callback?: () => void + ): Promise | void { + const callbackFn = typeof errorOrCallback === "function" ? errorOrCallback : callback; + const error = typeof errorOrCallback === "function" ? null : (errorOrCallback ?? null); + const done = Promise.all(this.dispatchers.map((dispatcher) => dispatcher.destroy(error))).then( + () => undefined + ); + if (callbackFn) { + done.then(callbackFn); + return; + } + return done; + } +} + function getDispatcherCache(): DispatcherCache { const globalWithCache = globalThis as GlobalWithDispatcherCache; if (!globalWithCache[DISPATCHER_CACHE_KEY]) { @@ -136,12 +188,11 @@ export function getDefaultDispatcherConnectionLimit( function getDefaultDispatcherOptions(env: Record = process.env) { const options = getDispatcherOptions(); - // #4580 — On the direct egress path, undici's default pipelining (1) lets a long - // SSE stream monopolize the single pooled socket per origin, serializing every - // other concurrent request to that same provider. Mirror the proxy fix (#4288): - // disable pipelining and keep several connections available. Unlike the proxy - // path we KEEP keep-alive — the 1ms TTL there is a cheap-proxy-socket workaround, - // not needed (and harmful to perf) for direct connections. + // #4580 — On the direct egress path, undici's default pipelining (1) let a long + // SSE stream monopolize the single pooled socket per origin. Keep the public + // connection-limit option here, but getDefaultDispatcher() fans it out across + // independent one-connection Agents; in production traces, one multi-connection + // Agent could still queue same-origin Codex streams behind prior trailers. return { ...options, connections: getDefaultDispatcherConnectionLimit(env), @@ -149,10 +200,23 @@ function getDefaultDispatcherOptions(env: Record = p }; } +function createRoundRobinDirectDispatcher(connectionLimit: number): Dispatcher { + const baseOptions = getDispatcherOptions(); + const perAgentOptions = { + ...baseOptions, + connections: 1, + pipelining: 0, + }; + const dispatchers = Array.from({ length: connectionLimit }, () => new Agent(perAgentOptions)); + return new RoundRobinDispatcher(dispatchers) as unknown as Dispatcher; +} + export function getDefaultDispatcher(): Dispatcher { const globalWithCache = globalThis as GlobalWithDispatcherCache; if (!globalWithCache[DEFAULT_DISPATCHER_KEY]) { - globalWithCache[DEFAULT_DISPATCHER_KEY] = new Agent(getDefaultDispatcherOptions()); + globalWithCache[DEFAULT_DISPATCHER_KEY] = createRoundRobinDirectDispatcher( + getDefaultDispatcherConnectionLimit() + ); } return globalWithCache[DEFAULT_DISPATCHER_KEY]; } @@ -415,6 +479,10 @@ export function __getDefaultDispatcherOptionsForTest( return getDefaultDispatcherOptions(env); } +export function __createRoundRobinDispatcherForTest(dispatchers: Dispatcher[]): Dispatcher { + return new RoundRobinDispatcher(dispatchers) as unknown as Dispatcher; +} + export function createProxyDispatcher(proxyUrl: string): Dispatcher { const normalizedUrl = normalizeProxyUrl(proxyUrl, "proxy dispatcher"); const dispatcherCache = getDispatcherCache(); diff --git a/tests/unit/direct-dispatcher-pipelining-4580.test.ts b/tests/unit/direct-dispatcher-pipelining-4580.test.ts index a56ed9083148..3f6ae944df57 100644 --- a/tests/unit/direct-dispatcher-pipelining-4580.test.ts +++ b/tests/unit/direct-dispatcher-pipelining-4580.test.ts @@ -1,6 +1,8 @@ import { describe, it, afterEach } from "node:test"; import assert from "node:assert/strict"; +import type { Dispatcher } from "undici"; import { + __createRoundRobinDispatcherForTest, __getDefaultDispatcherOptionsForTest, __getProxyDispatcherOptionsForTest, getDefaultDispatcherConnectionLimit, @@ -39,7 +41,10 @@ describe("#4580 direct dispatcher options", () => { }); it("connection limit honors OMNIROUTE_DIRECT_DISPATCHER_CONNECTIONS", () => { - assert.equal(getDefaultDispatcherConnectionLimit({ OMNIROUTE_DIRECT_DISPATCHER_CONNECTIONS: "8" }), 8); + assert.equal( + getDefaultDispatcherConnectionLimit({ OMNIROUTE_DIRECT_DISPATCHER_CONNECTIONS: "8" }), + 8 + ); }); it("connection limit clamps invalid values to the default", () => { @@ -49,4 +54,30 @@ describe("#4580 direct dispatcher options", () => { ); assert.equal(getDefaultDispatcherConnectionLimit({}), 32); }); + + it("fans out direct requests across independent dispatcher pools", () => { + const calls: number[] = []; + const dispatchers = [0, 1, 2].map( + (index) => + ({ + dispatch() { + calls.push(index); + return true; + }, + close() {}, + destroy() {}, + }) as unknown as Dispatcher + ); + const dispatcher = __createRoundRobinDispatcherForTest(dispatchers); + const dispatchOptions = { + origin: "https://chatgpt.com", + path: "/backend-api/codex/responses", + method: "POST", + } as unknown as Parameters[0]; + const handler = {} as Parameters[1]; + + for (let i = 0; i < 7; i++) dispatcher.dispatch(dispatchOptions, handler); + + assert.deepEqual(calls, [0, 1, 2, 0, 1, 2, 0]); + }); });