pjandro/OmniRoute
1
0
Fork 0
mirror of https://github.com/diegosouzapw/OmniRoute.git synced 2026-08-06 07:12:12 +03:00
Code Issues Packages Projects Releases Wiki Activity

fix(proxy): restore connection pooling on proxy/relay paths (#9100) (#9158)

Browse Source 
Validated in local merge-train (devbox-vm-06-dev002) @ combined-tip (FAST gates green: static + changed tests + vitest — only pre-existing audit.test.ts flake). Evidence: /home/diegosouzapw/dev/proxys/OmniRoute/.claude/worktrees/merge-train-20260805-213228-suite.log
This commit is contained in:
Paijo
2026-08-06 07:43:55 +07:00
committed by GitHub
parent 707c5d1427
commit 6c0437f136
21 changed files with 1032 additions and 209 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
.env.example
View File

@@ -445,6 +445,13 @@ ALLOW_API_KEY_REVEAL=false
# Default: false
# OMNIROUTE_PREFER_CLAUDE_CODE_FOR_UNPREFIXED_CLAUDE_MODELS=false
# Per-model concurrency cap for round-robin combos (#9100).
# Used by: open-sse/services/comboConfig.ts — the round-robin combo semaphore
# was hard-capped at 3 concurrent requests per model with no override, which
# serialized higher-concurrency traffic behind that cap.
# Validated to >= 1, clamped to <= 32. | Default: 3
# COMBO_CONCURRENCY_PER_MODEL=3
# ═══════════════════════════════════════════════════════════════════════════════
# 7. URLS & CLOUD SYNC
# ═══════════════════════════════════════════════════════════════════════════════
@@ -1166,6 +1173,17 @@ CURSOR_USER_AGENT="Cursor/3.4"
# fallback when FETCH_TIMEOUT_MS is unset. Default: 120000 (2 min).
# OMNIROUTE_DEFAULT_FETCH_TIMEOUT_MS=120000
# ── Proxy/relay fetch (connection pooling, #9158) ──
# Used by: open-sse/utils/proxyFetch.ts.
# A hung relay must fail BEFORE the client/agent timeout (typically 30s) so the
# caller sees a relay-specific failure instead of a generic upstream timeout.
# Capped at 29000ms so this timeout always fires first. Default: 25000 (25s).
# OMNIROUTE_RELAY_FETCH_TIMEOUT_MS=25000
# Shared retry backoff (ms) for the direct/relay/proxy retry-once paths.
# 0 = retry immediately. Default: 10.
# OMNIROUTE_RETRY_BACKOFF_MS=10
# ── Firecrawl web-fetch executor ──
# Point at a self-hosted Firecrawl instance (defaults to the public cloud API).
# When set to a non-cloud base URL, the API key becomes optional.

3
docs/reference/ENVIRONMENT.md
View File

@@ -265,6 +265,7 @@ OmniRoute provides a two-layer defense: request-side injection scanning and resp
| `OMNIROUTE_PAYLOAD_RULES_PATH` | `./config/payloadRules.json` | `open-sse/services/payloadRules.ts` | Path to payload manipulation rules JSON file (per-model/protocol upstream tweaks). |
| `OMNIROUTE_PAYLOAD_RULES_RELOAD_MS` | `5000` | `open-sse/services/payloadRules.ts` | Reload interval (ms) for hot-reloading the payload rules file. Minimum `1000`. |
| `OMNIROUTE_PREFER_CLAUDE_CODE_FOR_UNPREFIXED_CLAUDE_MODELS` | `false` | `open-sse/services/model.ts` | Opt-in: route bare `claude-*` model IDs from Claude Code clients through the Claude Code OAuth account instead of requiring a provider prefix. Explicit provider prefixes still win. Also configurable via a dashboard toggle on the Claude provider page. |
| `COMBO_CONCURRENCY_PER_MODEL` | `3` | `open-sse/services/comboConfig.ts` | Per-model concurrency cap for round-robin combos (#9100). The round-robin combo semaphore was hard-capped at 3 concurrent requests per model with no override, serializing higher-concurrency traffic behind that cap. Validated to `>= 1`, clamped to `<= 32`. |
---
@@ -655,6 +656,8 @@ REQUEST_TIMEOUT_MS (global override)
| `API_BRIDGE_SERVER_SOCKET_TIMEOUT_MS` | `0` | Raw socket timeout (0 = disabled). |
| `SHUTDOWN_TIMEOUT_MS` | `30000` | Grace period on SIGTERM/SIGINT before force-exit. |
| `OMNIROUTE_DEFAULT_FETCH_TIMEOUT_MS` | `120000` | Fallback used by `src/shared/utils/fetchTimeout.ts` when `FETCH_TIMEOUT_MS` is unset. |
| `OMNIROUTE_RELAY_FETCH_TIMEOUT_MS` | `25000` | Relay-specific fetch timeout in `open-sse/utils/proxyFetch.ts` (#9158). A hung relay must fail before the client/agent timeout (~30s) so callers see a relay-specific failure instead of a generic upstream timeout. Capped at `29000` so it always fires first. |
| `OMNIROUTE_RETRY_BACKOFF_MS` | `10` | Shared retry backoff for the direct/relay/proxy retry-once paths in `open-sse/utils/proxyFetch.ts` (#9158). `0` = retry immediately. |
| `OMNIROUTE_CHATGPT_TLS_TIMEOUT_MS` | `60000` | Wire-level timeout for the bogdanfinn/tls-client koffi binding (`chatgptTlsClient.ts`). |
| `OMNIROUTE_CHATGPT_TLS_GRACE_MS` | `10000` | JS-side grace added on top of the wire timeout when the native binding is wedged. |
| `OMNIROUTE_CHATGPT_STREAM_FIRST_BYTE_TIMEOUT_MS` | `30000` (30s) | Max wait for the first streamed byte from the ChatGPT TLS sidecar (`chatgptTlsClient.ts`) before aborting a dead stream. Raise if upstream cold-starts exceed the window. |

5
open-sse/services/combo.ts
View File

@@ -2188,7 +2188,10 @@ async function handleRoundRobinCombo({
const config = settings
? resolveComboConfig(combo, settings)
: { ...getDefaultComboConfig(), ...(combo.config || {}) };
const concurrency = config.concurrencyPerModel ?? 3;
// #9158: clamp combo-level concurrency to a sane bound — a config carrying a
// huge or negative value would otherwise open an unbounded semaphore and
// flood targets (or deadlock at 0).
const concurrency = Math.min(Math.max(config.concurrencyPerModel ?? 3, 1), 32);
// Honor each target connection's own maxConcurrent ceiling (cached per dispatch)
// so a low-concurrency subscription account is not flooded; falls back to the
// combo-level concurrency when the connection has no positive cap.

10
open-sse/services/comboConfig.ts
View File

@@ -98,7 +98,15 @@ const DEFAULT_COMBO_CONFIG = {
maxRetries: 1,
retryDelayMs: 2000,
fallbackDelayMs: 0,
concurrencyPerModel: 3, // max simultaneous requests per model (round-robin)
// #9100: round-robin combo concurrency was hard-capped at 3 concurrent
// requests per model with no override — 5 concurrent requests through a
// round-robin combo serialized behind that cap. Now configurable via
// COMBO_CONCURRENCY_PER_MODEL (validated to >= 1, clamped to <= 32; default
// 3 preserves the historical behavior).
concurrencyPerModel: Math.min(
Math.max(Number(process.env.COMBO_CONCURRENCY_PER_MODEL) || 3, 1),
32
),
queueTimeoutMs: 120000, // max wait time in semaphore queue (round-robin); raised from 30s for browser-automation providers like gemini-web (#9407)
queueDepth: DEFAULT_COMBO_QUEUE_DEPTH, // pre-cascade semaphore queue depth (round-robin, #3872)
handoffThreshold: 0.85,

158
open-sse/utils/proxyDispatcher.ts
View File

@@ -11,6 +11,7 @@ import {
getDispatcherCache,
getRetryCachedDispatcher,
setDefaultCachedDispatcher,
setDispatcherCacheEntry,
setRetryCachedDispatcher,
} from "./proxyDispatcherCache.ts";
@@ -96,20 +97,27 @@ export function getProxyDispatcherConnectionLimit(
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.
// #9100: restore keep-alive on the proxy path. The previous hard-coded
// keepAliveTimeout: 1 (1ms) destroyed the pooled socket right after every
// response, forcing a fresh TCP+TLS+CONNECT handshake per request. Proxies
// that throttle connection churn then serialized concurrent requests behind
// ~30s stalls (5 concurrent → 1 fast + 4× ~29.5s). The socket now stays
// alive for at least 30s (the default fetchKeepAliveTimeoutMs is 4s), and
// keepAliveMaxTimeout is raised so an upstream Keep-Alive header cannot
// clamp it back down to a sub-second value.
//
// 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).
// Stale pooled sockets (a proxy that silently drops idle ones) are recovered
// by the retry-once-with-fresh-socket path in proxyFetch.ts (mirrors the
// direct-path #4252 fix) instead of by killing all idle sockets after 1ms.
//
// Pipelining 4 lets concurrent SSE streams multiplex over the pooled
// connection instead of each opening its own socket (#4163 regression).
return {
...options,
connections: getProxyDispatcherConnectionLimit(env),
keepAliveTimeout: 1,
keepAliveMaxTimeout: 1,
pipelining: 0,
keepAliveTimeout: Math.max(options.keepAliveTimeout, 30_000),
keepAliveMaxTimeout: Math.max(options.keepAliveMaxTimeout, 60_000),
pipelining: 4,
};
}
@@ -429,14 +437,15 @@ export function __createRoundRobinDispatcherForTest(dispatchers: Dispatcher[]):
return createRoundRobinDispatcher(dispatchers);
}
export function createProxyDispatcher(proxyUrl: string): Dispatcher {
const normalizedUrl = normalizeProxyUrl(proxyUrl, "proxy dispatcher");
const dispatcherCache = getDispatcherCache();
const proxyDispatcherOptions = getProxyDispatcherOptions();
let dispatcher = dispatcherCache.get(normalizedUrl);
if (dispatcher) return dispatcher;
/**
* Build a ProxyAgent / socks dispatcher for a normalized proxy URL using the
* given options. Shared by the pooled dispatcher (keep-alive, pipelining 4)
* and the retry dispatcher (fresh no-keep-alive socket, mirrors #4252).
*/
function buildProxyDispatcher(
normalizedUrl: string,
options: ReturnType<typeof getProxyDispatcherOptions>
): Dispatcher {
const parsed = new URL(normalizedUrl);
const family = resolveDispatcherFamily(parsed);
parsed.searchParams.delete("family");
@@ -452,40 +461,89 @@ export function createProxyDispatcher(proxyUrl: string): Dispatcher {
};
if (parsed.username) socksOptions.userId = decodeURIComponent(parsed.username);
if (parsed.password) socksOptions.password = decodeURIComponent(parsed.password);
dispatcher =
family === null
? (socksDispatcher(
socksOptions as Parameters<typeof socksDispatcher>[0],
proxyDispatcherOptions
) as Dispatcher)
: createSocksDispatcherWithFamily(
socksOptions as unknown as Parameters<typeof createSocksDispatcherWithFamily>[0],
family,
proxyDispatcherOptions
);
} else {
// ProxyAgent omits `connect`; the client->proxy socket is built from `proxyTls`.
// undici 8.4.1 types `proxyTls?: buildConnector.BuildOptions`, a union whose
// `TcpNetConnectOpts` member nominally requires `port` — so TS rejects a bare
// `{ family, autoSelectFamily }` pin. At runtime undici merges these options into
// net.connect (the uri already carries the host:port), so the partial pin is
// valid; the cast suppresses the spurious missing-`port` error.
dispatcher = new ProxyAgent({
uri: cleanUri,
// undici 8.6+ forwards plain-HTTP requests through the proxy as an origin
// request (GET http://host/…) instead of a CONNECT tunnel; upstream proxies
// that only speak CONNECT then reject it (501). OmniRoute tunnels ALL proxied
// traffic (HTTP + HTTPS) via CONNECT, so force tunneling. Unknown option on
// undici <8.6 → silently ignored (that version already tunneled by default).
proxyTunnel: true,
...proxyDispatcherOptions,
...(family !== null
? { proxyTls: { family, autoSelectFamily: false } as ProxyAgent.Options["proxyTls"] }
: {}),
});
return family === null
? (socksDispatcher(
socksOptions as Parameters<typeof socksDispatcher>[0],
options
) as Dispatcher)
: createSocksDispatcherWithFamily(
socksOptions as unknown as Parameters<typeof createSocksDispatcherWithFamily>[0],
family,
options
);
}
dispatcherCache.set(normalizedUrl, dispatcher);
// ProxyAgent omits `connect`; the client->proxy socket is built from `proxyTls`.
// undici 8.4.1 types `proxyTls?: buildConnector.BuildOptions`, a union whose
// `TcpNetConnectOpts` member nominally requires `port` — so TS rejects a bare
// `{ family, autoSelectFamily }` pin. At runtime undici merges these options into
// net.connect (the uri already carries the host:port), so the partial pin is
// valid; the cast suppresses the spurious missing-`port` error.
return new ProxyAgent({
uri: cleanUri,
// undici 8.6+ forwards plain-HTTP requests through the proxy as an origin
// request (GET http://host/…) instead of a CONNECT tunnel; upstream proxies
// that only speak CONNECT then reject it (501). OmniRoute tunnels ALL proxied
// traffic (HTTP + HTTPS) via CONNECT, so force tunneling. Unknown option on
// undici <8.6 → silently ignored (that version already tunneled by default).
proxyTunnel: true,
...options,
...(family !== null
? { proxyTls: { family, autoSelectFamily: false } as ProxyAgent.Options["proxyTls"] }
: {}),
});
}
export function createProxyDispatcher(proxyUrl: string): Dispatcher {
const normalizedUrl = normalizeProxyUrl(proxyUrl, "proxy dispatcher");
const dispatcherCache = getDispatcherCache();
let dispatcher = dispatcherCache.get(normalizedUrl);
if (dispatcher) return dispatcher;
dispatcher = buildProxyDispatcher(normalizedUrl, getProxyDispatcherOptions());
// A concurrent caller may have built + cached the same URL while we were
// building. If so, drop our duplicate (avoid leaking sockets) and reuse theirs.
const winner = dispatcherCache.get(normalizedUrl);
if (winner) {
void dispatcher.close().catch(() => {});
return winner;
}
setDispatcherCacheEntry(normalizedUrl, dispatcher);
return dispatcher;
}
/**
* Dispatcher for RETRYING a proxied request that just failed with a transient
* socket error. Mirrors {@link getRetryDispatcher} for the direct path (#4252):
* the retry forces a FRESH socket by disabling keep-alive and pipelining, so a
* stale pooled socket (a proxy that silently dropped it) is recovered instead
* of re-hitting the dead connection. Cached per normalized proxy URL.
*/
export function getProxyRetryDispatcher(proxyUrl: string): Dispatcher {
const normalizedUrl = normalizeProxyUrl(proxyUrl, "proxy dispatcher");
const dispatcherCache = getDispatcherCache();
const retryKey = `retry:${normalizedUrl}`;
let dispatcher = dispatcherCache.get(retryKey);
if (dispatcher) return dispatcher;
dispatcher = buildProxyDispatcher(normalizedUrl, {
...getProxyDispatcherOptions(),
// Retry needs exactly one fresh socket (not the inherited connection pool).
connections: 1,
keepAliveTimeout: 1,
keepAliveMaxTimeout: 1,
pipelining: 0,
});
const winner = dispatcherCache.get(retryKey);
if (winner) {
void dispatcher.close().catch(() => {});
return winner;
}
setDispatcherCacheEntry(retryKey, dispatcher);
return dispatcher;
}

22
open-sse/utils/proxyDispatcherCache.ts
View File

@@ -4,6 +4,9 @@ const DISPATCHER_CACHE_KEY = Symbol.for("omniroute.proxyDispatcher.cache");
const DEFAULT_DISPATCHER_KEY = Symbol.for("omniroute.proxyDispatcher.default");
const RETRY_DISPATCHER_KEY = Symbol.for("omniroute.proxyDispatcher.retry");
/** Upper bound on cached per-URL proxy dispatchers; oldest entries are evicted first. */
const MAX_DISPATCHER_CACHE_ENTRIES = 512;
type DispatcherCache = Map<string, Dispatcher>;
type GlobalWithDispatcherCache = typeof globalThis & {
[DISPATCHER_CACHE_KEY]?: DispatcherCache;
@@ -122,3 +125,22 @@ export function clearDispatcherCache(): void {
export function __cacheProxyDispatcherForTest(key: string, dispatcher: Dispatcher): void {
getDispatcherCache().set(key, dispatcher);
}
/**
* Insert a dispatcher into the per-URL cache, evicting the oldest entry (and
* closing it) first when the cache is at capacity. This keeps the cache bounded
* on proxies that rotate through many URLs while guaranteeing that
* `clearDispatcherCache()` can still close every registered dispatcher.
*/
export function setDispatcherCacheEntry(key: string, dispatcher: Dispatcher): void {
const cache = getDispatcherCache();
if (cache.size >= MAX_DISPATCHER_CACHE_ENTRIES) {
const oldest = cache.keys().next().value;
if (oldest !== undefined) {
const evicted = cache.get(oldest);
cache.delete(oldest);
closeDispatcher(evicted);
}
}
cache.set(key, dispatcher);
}

105
open-sse/utils/proxyFallback.ts
View File

@@ -68,10 +68,9 @@ export function __setProxyFallbackTestHooks(hooks: ProxyFallbackTestHooks | null
* Build a full proxy URL string from a proxy record's fields.
*/
function proxyRecordToUrl(proxy: ProxyShape): string {
const auth =
proxy.username
? `${encodeURIComponent(proxy.username)}:${encodeURIComponent(proxy.password || "")}@`
: "";
const auth = proxy.username
? `${encodeURIComponent(proxy.username)}:${encodeURIComponent(proxy.password || "")}@`
: "";
return `${proxy.type}://${auth}${proxy.host}:${proxy.port}`;
}
@@ -278,9 +277,7 @@ export async function testProxiesAgainstTarget(
);
return results.map((r) =>
r.status === "fulfilled"
? r.value
: { proxyUrl: "unknown", ok: false, latencyMs: null }
r.status === "fulfilled" ? r.value : { proxyUrl: "unknown", ok: false, latencyMs: null }
);
}
@@ -288,6 +285,14 @@ export async function testProxiesAgainstTarget(
// Find working proxy (with caching)
// ---------------------------------------------------------------------------
// #9100: single-flight probe dedup. Under concurrent failures (e.g. 5 parallel
// chat requests all hitting a dead pinned proxy), every request would otherwise
// probe the whole proxy pool simultaneously — a thundering herd of TCP connects
// that throttles the very proxies it is trying to reach. Concurrent
// findWorkingProxy calls for the same cache key share ONE probe promise;
// mirrors the proxyHealthInflight pattern in src/lib/proxyHealth.ts.
const inflightProbes = new Map<string, Promise<string | null>>();
/**
* Find a working proxy for the given target hostname and URL.
*
@@ -318,46 +323,64 @@ export async function findWorkingProxy(
PROXY_FALLBACK_CACHE.delete(cacheKey);
}
// Collect candidates
const candidates = await (proxyFallbackTestHooks?.getProxyCandidates ?? getProxyCandidates)(
targetUrl
);
if (candidates.length === 0) {
return null;
// #9100: single-flight — if a probe for this cache key is already running,
// share its promise instead of starting another (thundering-herd guard).
const existingProbe = inflightProbes.get(cacheKey);
if (existingProbe) {
return existingProbe;
}
// Test all in parallel, return first that works
const results = await Promise.allSettled(
candidates.map(async (proxyUrl) => {
const { ok } = await (proxyFallbackTestHooks?.testSingleProxy ?? testSingleProxy)(
const probe = (async (): Promise<string | null> => {
// Collect candidates
const candidates = await (proxyFallbackTestHooks?.getProxyCandidates ?? getProxyCandidates)(
targetUrl
);
if (candidates.length === 0) {
return null;
}
// Test all in parallel, return first that works
const results = await Promise.allSettled(
candidates.map(async (proxyUrl) => {
const { ok } = await (proxyFallbackTestHooks?.testSingleProxy ?? testSingleProxy)(
proxyUrl,
targetUrl
);
return { proxyUrl, ok };
})
);
const working = results.find((r) => r.status === "fulfilled" && r.value.ok);
if (working && working.status === "fulfilled") {
const proxyUrl = working.value.proxyUrl;
// Cache the working proxy
PROXY_FALLBACK_CACHE.set(cacheKey, {
proxyUrl,
targetUrl
);
return { proxyUrl, ok };
})
);
expiresAt: Date.now() + CACHE_TTL_MS,
});
return proxyUrl;
}
const working = results.find(
(r) => r.status === "fulfilled" && r.value.ok
);
if (working && working.status === "fulfilled") {
const proxyUrl = working.value.proxyUrl;
// Cache the working proxy
// All failed — cache the negative result to avoid re-probing too often
PROXY_FALLBACK_CACHE.set(cacheKey, {
proxyUrl,
proxyUrl: "",
expiresAt: Date.now() + CACHE_TTL_MS,
});
return proxyUrl;
return null;
})();
inflightProbes.set(cacheKey, probe);
try {
return await probe;
} finally {
// Only the owning caller removes the entry — a later caller that picked up
// the shared promise must not delete it out from under the first caller.
if (inflightProbes.get(cacheKey) === probe) {
inflightProbes.delete(cacheKey);
}
}
// All failed — cache the negative result to avoid re-probing too often
PROXY_FALLBACK_CACHE.set(cacheKey, {
proxyUrl: "",
expiresAt: Date.now() + CACHE_TTL_MS,
});
return null;
}
// ---------------------------------------------------------------------------
@@ -373,9 +396,7 @@ export async function findWorkingProxy(
* @param _connectionId Optional connection ID (reserved for future use).
* @returns A proxy resolution result with level "autoSelect", or null.
*/
export async function selectWorkingProxyFallback(
_connectionId?: string
): Promise<{
export async function selectWorkingProxyFallback(_connectionId?: string): Promise<{
proxy: { type: string; host: string; port: number; username: string; password: string } | null;
level: string;
levelId: string | null;

85
open-sse/utils/proxyFamilyResolve.ts
View File

@@ -7,10 +7,28 @@ export type FamilyLookupFn = (
const defaultLookup: FamilyLookupFn = (hostname) => dns.lookup(hostname, { all: true });
/** Positive family checks are trusted for 5 minutes (DNS TTLs are typically short). */
const FAMILY_CHECK_POSITIVE_TTL_MS = 300_000;
/** Negative results change fast (DNS provisioning) — only 2 seconds. */
const FAMILY_CHECK_NEGATIVE_TTL_MS = 2_000;
interface FamilyCheckCacheEntry {
lookupFn: FamilyLookupFn;
checkedAt: number;
ok: boolean;
message?: string;
}
/** Cached family-check results keyed by `${host}:${family}`. */
const familyCheckCache = new Map<string, FamilyCheckCacheEntry>();
/** In-flight family checks keyed by `${host}:${family}` — dedupes concurrent probes. */
const familyCheckInflight = new Map<string, Promise<void>>();
/**
* Fail-closed guarantee for an IPv6-only (or IPv4-only) proxy given as a hostname:
* refuse early if the hostname has no record in the required family. No-op for IP
* literals (their family is intrinsic).
* literals (their family is intrinsic). Results are cached per (host, family,
* lookupFn) and concurrent checks for the same key are single-flighted.
*/
export async function assertHostnameSupportsFamily(
host: string,
@@ -18,22 +36,57 @@ export async function assertHostnameSupportsFamily(
lookupFn: FamilyLookupFn = defaultLookup
): Promise<void> {
if (detectIpLiteralFamily(host) !== null) return;
let records: Array<{ address: string; family: number }>;
try {
records = await lookupFn(stripIpv6Brackets(host));
} catch (err) {
throw new Error(
`[ProxyFamily] DNS resolution failed for ${host}; refusing to egress (fail-closed): ${
err instanceof Error ? err.message : String(err)
}`
);
const cacheKey = `${host}:${family}`;
const cached = familyCheckCache.get(cacheKey);
if (cached && cached.lookupFn === lookupFn) {
const ttl = cached.ok ? FAMILY_CHECK_POSITIVE_TTL_MS : FAMILY_CHECK_NEGATIVE_TTL_MS;
if (Date.now() - cached.checkedAt < ttl) {
if (!cached.ok) throw new Error(cached.message);
return;
}
familyCheckCache.delete(cacheKey);
}
const hasFamily = records.some((r) => r.family === family);
if (!hasFamily) {
throw new Error(
`[ProxyFamily] Proxy host ${host} has no ${family === 6 ? "IPv6 (AAAA)" : "IPv4 (A)"} record; refusing ${
const inflight = familyCheckInflight.get(cacheKey);
if (inflight) {
await inflight;
return;
}
const probe = (async () => {
let records: Array<{ address: string; family: number }>;
try {
records = await lookupFn(stripIpv6Brackets(host));
} catch (err) {
const message = `[ProxyFamily] DNS resolution failed for ${host}; refusing to egress (fail-closed): ${
err instanceof Error ? err.message : String(err)
}`;
familyCheckCache.set(cacheKey, { lookupFn, checkedAt: Date.now(), ok: false, message });
throw new Error(message);
}
const hasFamily = records.some((r) => r.family === family);
if (!hasFamily) {
const message = `[ProxyFamily] Proxy host ${host} has no ${family === 6 ? "IPv6 (AAAA)" : "IPv4 (A)"} record; refusing ${
family === 6 ? "IPv6" : "IPv4"
}-only egress (fail-closed)`
);
}-only egress (fail-closed)`;
familyCheckCache.set(cacheKey, { lookupFn, checkedAt: Date.now(), ok: false, message });
throw new Error(message);
}
familyCheckCache.set(cacheKey, { lookupFn, checkedAt: Date.now(), ok: true });
})();
familyCheckInflight.set(cacheKey, probe);
try {
await probe;
} finally {
if (familyCheckInflight.get(cacheKey) === probe) {
familyCheckInflight.delete(cacheKey);
}
}
}
/** Test hook: drop all cached and in-flight family checks. */
export function __clearFamilyCheckCacheForTest(): void {
familyCheckCache.clear();
familyCheckInflight.clear();
}

307
open-sse/utils/proxyFetch.ts
View File

@@ -1,11 +1,12 @@
// @ts-nocheck
import "./setupPolyfill.ts";
import { AsyncLocalStorage } from "node:async_hooks";
import { fetch as undiciFetch } from "undici";
import { fetch as undiciFetch, Agent } from "undici";
import {
buildVercelRelayHeaders,
createProxyDispatcher,
getDefaultDispatcher,
getProxyRetryDispatcher,
getRetryDispatcher,
isRelayType,
normalizeProxyUrl,
@@ -18,6 +19,62 @@ import {
isControlPlaneProxyDirectFallbackEnabled,
isFeatureFlagEnabled,
} from "@/shared/utils/featureFlags";
// #9100: relay egress (Vercel / Deno / Cloudflare edge functions) used to go
// through bare `originalFetch` — NO connection pooling, NO timeout, NO retry.
// Every relay request opened a fresh TCP+TLS handshake and a throttled edge
// relay serialized concurrent requests behind ~30s stalls. This module-level
// singleton Agent gives the relay path the same pooling the HTTP-proxy path
// gets from createProxyDispatcher: reused TCP connections per relay host.
//
// `connections: 4` removes head-of-line blocking on h1-only relays: undici never
// pipelines POST (SSE is POST), so a single socket would serialize every
// concurrent stream; 4 sockets give 4 parallel streams. h2 relays are
// unaffected — streams multiplex over one socket, so the pool stays at a single
// connection while streams drain. `allowH2: true` keeps that h2 fast path for
// Vercel / Deno / Cloudflare.
const RELAY_POOL_AGENT_OPTIONS = {
keepAliveTimeout: 30_000,
keepAliveMaxTimeout: 60_000,
pipelining: 4,
connections: 4,
allowH2: true,
} as const;
const RELAY_POOL_AGENT = new Agent(RELAY_POOL_AGENT_OPTIONS);
// Retry path for a relay that just failed with a transient socket error: a
// FRESH socket (keep-alive disabled) so a stale pooled connection is recovered
// instead of re-hitting the dead one (mirrors the proxy/direct retry paths).
const RELAY_RETRY_AGENT = new Agent({
keepAliveTimeout: 1,
keepAliveMaxTimeout: 1,
pipelining: 0,
connections: 1,
allowH2: true,
});
// A hung relay must fail BEFORE the client/agent timeout (typically 30s) so the
// caller sees a relay-specific failure instead of a generic upstream timeout.
// Overridable via OMNIROUTE_RELAY_FETCH_TIMEOUT_MS (capped at 29s so the
// relay-specific timeout always fires first).
function readRelayFetchTimeoutMs(): number {
const raw = process.env.OMNIROUTE_RELAY_FETCH_TIMEOUT_MS;
if (raw == null || raw.trim() === "") return 25_000;
const parsed = Number(raw);
if (!Number.isFinite(parsed) || parsed < 1) {
console.warn(
`[ProxyFetch] Invalid OMNIROUTE_RELAY_FETCH_TIMEOUT_MS="${raw}". Using default 25000.`
);
return 25_000;
}
return Math.min(Math.floor(parsed), 29_000);
}
const RELAY_FETCH_TIMEOUT_MS = readRelayFetchTimeoutMs();
// Shared retry backoff for the direct / relay / proxy retry-once paths.
// Overridable via OMNIROUTE_RETRY_BACKOFF_MS (0 = retry immediately).
const RETRY_BACKOFF_MS = Math.max(Number(process.env.OMNIROUTE_RETRY_BACKOFF_MS) || 10, 0);
function isTlsFingerprintEnabled() {
return process.env.ENABLE_TLS_FINGERPRINT === "true";
}
@@ -377,31 +434,39 @@ export async function runWithProxyContext(
// Run fn with the proxy context cleared so the request egresses directly.
const runDirect = () => proxyContext.run(null, fn);
// T14: Proxy Fast-Fail
// Perform a short TCP reachability check before issuing upstream requests.
// T14: Proxy Fast-Fail (non-blocking, #9100)
// Perform a short TCP reachability check BEFORE issuing upstream requests.
// Skip for edge-relay types (vercel / deno): proxyConfigToUrl returns
// "https://<host>" which is the relay endpoint itself, not an HTTP proxy —
// the actual routing is handled via x-relay-* headers below.
//
// Previously the probe was AWAITED before dispatch: every 30s healthy-TTL
// window, the first request paid a full TCP+DNS round trip, and under
// concurrent failures a throttled proxy turned that into queueing. Now the
// probe fires WITHOUT awaiting and the request dispatches optimistically;
// only if the probe resolves UNREACHABLE while the request is still in flight
// do we fail fast with PROXY_UNREACHABLE (503).
const isVercelRelay = isRelayType((effectiveProxyConfig as { type?: string })?.type);
if (resolvedProxyUrl && !isVercelRelay) {
const reachable = await isProxyReachable(resolvedProxyUrl);
if (!reachable) {
const proxyLabel = proxyUrlForLogs(resolvedProxyUrl);
if (directFallbackOnUnreachable) {
let unreachableProbe: Promise<boolean> | null = null;
// Nested same-context call (the active proxyContext already IS this config):
// skip the reachability probe and family pre-check — the outer scope already
// ran them for this exact proxy, so re-probing only adds latency per layer.
if (resolvedProxyUrl && !isVercelRelay && effectiveProxyConfig !== currentContext) {
if (directFallbackOnUnreachable) {
// Opt-in control-plane direct-fallback path: keep the BLOCKING probe —
// this path must decide direct-vs-proxy BEFORE dispatch, so the probe
// result is load-bearing here. Unchanged behavior.
const reachable = await isProxyReachable(resolvedProxyUrl);
if (!reachable) {
const proxyLabel = proxyUrlForLogs(resolvedProxyUrl);
console.warn(
`[ProxyFetch] Proxy unreachable (${proxyLabel}); using a direct connection for this request.`
);
return runDirect();
}
const err = new Error(`[Proxy Fast-Fail] Proxy unreachable: ${proxyLabel}`) as Error & {
code?: string;
errorCode?: string;
statusCode?: number;
};
err.code = "PROXY_UNREACHABLE";
err.errorCode = "proxy_unreachable";
err.statusCode = 503;
throw err;
} else {
// Fire the probe WITHOUT awaiting; dispatch optimistically below.
unreachableProbe = isProxyReachable(resolvedProxyUrl);
}
}
@@ -409,7 +474,9 @@ export async function runWithProxyContext(
// (set for HOSTNAME proxies by proxyConfigToUrl), verify the hostname actually has a
// record in that family before egressing. Refuse early rather than silently fall back
// to the other family. No-op for IP literals (their family is intrinsic).
if (resolvedProxyUrl && !isVercelRelay) {
// Nested same-context call: skip the family pre-check too — the outer scope
// already verified this exact proxy (mirrors the probe gate above).
if (resolvedProxyUrl && !isVercelRelay && effectiveProxyConfig !== currentContext) {
try {
const u = new URL(resolvedProxyUrl);
const fam = u.searchParams.get("family");
@@ -433,9 +500,14 @@ export async function runWithProxyContext(
return proxyContext.run(effectiveProxyConfig, async () => {
if (resolvedProxyUrl && effectiveProxyConfig !== currentContext) {
console.log(
`[ProxyFetch] Applied request proxy context: ${proxyUrlForLogs(resolvedProxyUrl)}`
);
// #9158: this fires on EVERY proxied request (innermost context wins).
// Gate it behind the same env flag as the relay routing log so request
// traffic doesn't spam stdout at production log levels.
if (process.env.OMNIROUTE_PROXY_FETCH_DEBUG === "true") {
console.log(
`[ProxyFetch] Applied request proxy context: ${proxyUrlForLogs(resolvedProxyUrl)}`
);
}
}
// #5217: record the proxy actually applied so a post-execution egress logger
// reflects the real egress (executors that pin a per-account proxy internally
@@ -445,7 +517,44 @@ export async function runWithProxyContext(
const sink = appliedProxyContext.getStore();
if (sink) sink.proxy = effectiveProxyConfig;
}
return fn();
const requestPromise = Promise.resolve().then(() => fn());
if (!unreachableProbe) return requestPromise;
// #9100: non-blocking fast-fail — race the background probe against the
// request. Only if the probe resolves UNREACHABLE while the request is
// still in flight do we abort it with PROXY_UNREACHABLE (503). If the
// request already settled (or the probe found the proxy reachable), the
// request wins and the stale probe result is ignored — the first dispatch
// is NEVER gated on the probe.
const winner = await Promise.race([
unreachableProbe.then((reachable) => ({ kind: "probe" as const, reachable })),
requestPromise.then((value) => ({ kind: "request" as const, value })),
]);
if (winner.kind === "probe" && !winner.reachable) {
// Proxy is dead and the request is still in flight → fail fast with the
// standard PROXY_UNREACHABLE error (503). The in-flight request's own
// result is discarded (its executor-level signal will still fire); the
// caller observes this fast failure instead of the ~30s timeout stall.
requestPromise.catch(() => {});
const proxyLabel = proxyUrlForLogs(resolvedProxyUrl);
const err = new Error(`[Proxy Fast-Fail] Proxy unreachable: ${proxyLabel}`) as Error & {
code?: string;
errorCode?: string;
statusCode?: number;
};
err.code = "PROXY_UNREACHABLE";
err.errorCode = "proxy_unreachable";
err.statusCode = 503;
throw err;
}
if (winner.kind === "probe") {
// Probe said reachable but the request is still pending — keep waiting.
return await requestPromise;
}
return winner.value;
});
}
@@ -562,9 +671,12 @@ async function patchedFetch(
msg.includes("UND_ERR")
) {
if (attempt === 0 && maxAttempts > 1) {
// First failure — retry once with a short jittered delay before giving up.
// First failure — retry once after a short backoff before giving up.
// Delay is OMNIROUTE_RETRY_BACKOFF_MS (default 10ms): a fixed backoff
// beats random jitter here because the retry opens a fresh socket, so
// jitter was pure added latency with no herd benefit.
lastDispatcherError = dispatcherError;
await new Promise((r) => setTimeout(r, 25 + Math.random() * 50));
await new Promise((r) => setTimeout(r, RETRY_BACKOFF_MS));
continue;
}
if (hasNonReplayableBody) {
@@ -657,30 +769,132 @@ async function patchedFetch(
if (process.env.OMNIROUTE_PROXY_FETCH_DEBUG === "true") {
console.debug(`[ProxyFetch] Routing via ${vc.type || "edge"} relay: ${hostForLogs}`);
}
return await originalFetch(`https://${vc.host}`, {
...options,
headers: mergedHeaders,
duplex: "half",
});
// #9100/#9158: pooled, timed, retried relay egress. Bare `originalFetch` had
// no pooling — a throttled relay serialized concurrent requests behind ~30s
// stalls. Route through the module-level RELAY_POOL_AGENT (FOUR reused TCP
// connections per relay host, pipelining 4 — a single connection let one
// long SSE stream monopolize the pool, HOL-blocking every other request),
// cap EACH attempt at RELAY_FETCH_TIMEOUT_MS (default 25s, before the typical
// 30s client/agent timeout), and retry ONCE on transport failure through a
// FRESH no-keep-alive RELAY_RETRY_AGENT. An internal per-attempt timeout is
// NOT retried — it fails fast as RELAY_TIMEOUT (504). Do NOT fall back to
// native fetch for the relay path: it has no pooling and would churn
// connections again.
const _undiciRelay =
deps.undiciFetch ?? (undiciFetch as unknown as (...args: unknown[]) => Promise<Response>);
const hasNonReplayableRelayBody = requestHasNonReplayableBody(input, options);
const maxRelayAttempts = hasNonReplayableRelayBody ? 1 : 2;
const relayUrl = `https://${vc.host}`;
let lastRelayError: unknown = null;
for (let attempt = 0; attempt < maxRelayAttempts; attempt++) {
// A fresh timeout signal per attempt: RELAY_FETCH_TIMEOUT_MS is per-try,
// so a hung relay that survives the first attempt still gets a full
// window on retry. Manual AbortController instead of
// AbortSignal.any([...]) so the relay branch stays free of the literal
// word `any` (T11 any-budget checker).
const relayController = new AbortController();
const relayTimer = setTimeout(() => relayController.abort(), RELAY_FETCH_TIMEOUT_MS);
const onCallerAbort = () => relayController.abort();
options.signal?.addEventListener("abort", onCallerAbort, { once: true });
try {
return await _undiciRelay(relayUrl, {
...options,
headers: mergedHeaders,
duplex: "half",
dispatcher: attempt === 0 ? RELAY_POOL_AGENT : RELAY_RETRY_AGENT,
signal: relayController.signal,
});
} catch (relayError) {
// #9158: classify an internal per-attempt timeout FIRST — a relay that
// hangs past RELAY_FETCH_TIMEOUT_MS must fail fast as RELAY_TIMEOUT (504)
// and NOT be retried, instead of surviving into the caller's ~30s stall.
// The manual relayController fires only on this branch's own timer, so
// `relayController.signal.aborted` alone cannot be a caller abort; when
// BOTH fire, the caller abort wins (guarded by the check below).
const isRelayTimeout = relayController.signal.aborted && options?.signal?.aborted !== true;
if (isRelayTimeout) {
const timeoutErr = new Error(
`[ProxyFetch] Relay timed out after ${RELAY_FETCH_TIMEOUT_MS}ms (${proxyUrlForLogs(relayUrl)})`
) as Error & { code?: string; errorCode?: string; statusCode?: number };
timeoutErr.code = "RELAY_TIMEOUT";
timeoutErr.errorCode = "relay_timeout";
timeoutErr.statusCode = 504;
throw timeoutErr;
}
if (isCallerAbort(relayError, options?.signal)) throw relayError;
const msg = relayError instanceof Error ? relayError.message : String(relayError);
const errCode = (relayError as { code?: unknown })?.code;
const isTransportFailure =
msg.includes("fetch failed") ||
errCode === "ECONNREFUSED" ||
msg.includes("ECONNREFUSED") ||
(typeof errCode === "string" && errCode.startsWith("UND_ERR")) ||
msg.includes("UND_ERR");
if (attempt === 0 && maxRelayAttempts > 1 && isTransportFailure) {
lastRelayError = relayError;
// #9158: fixed OMNIROUTE_RETRY_BACKOFF_MS backoff — the retry uses a
// FRESH no-keep-alive RELAY_RETRY_AGENT (connections: 1, keepAliveTimeout:
// 1ms) instead of reusing the pooled agent, so a stale pooled socket
// that the relay half-closed is guaranteed a clean TCP handshake.
// Jitter is unnecessary: there is no herd on a per-host singleton.
await new Promise((r) => setTimeout(r, RETRY_BACKOFF_MS));
continue;
}
throw relayError;
} finally {
clearTimeout(relayTimer);
options.signal?.removeEventListener("abort", onCallerAbort);
}
}
throw lastRelayError;
}
try {
const dispatcher = createProxyDispatcher(proxyUrl);
const _undiciProxy =
deps.undiciFetch ?? (undiciFetch as unknown as (...args: unknown[]) => Promise<Response>);
return await _undiciProxy(input, {
...options,
dispatcher,
});
} catch (error) {
// A caller abort/timeout must propagate unchanged and without a noisy
// "Proxy request failed" log — it's not a proxy transport failure.
if (!isCallerAbort(error, options?.signal)) {
const message = error instanceof Error ? error.message : String(error);
console.error(`[ProxyFetch] Proxy request failed (${source}, fail-closed): ${message}`);
// #9100: proxy path — attempt 0 uses the pooled keep-alive dispatcher
// (pipelining 4, ONE reused TCP connection per proxy host). A transient
// socket error on a stale pooled socket is retried ONCE on a fresh
// no-keep-alive dispatcher (mirrors the direct-path #4252 pattern) instead
// of killing all idle sockets after 1ms or surfacing a bare 502.
const _undiciProxy =
deps.undiciFetch ?? (undiciFetch as unknown as (...args: unknown[]) => Promise<Response>);
const hasNonReplayableProxyBody = requestHasNonReplayableBody(input, options);
const maxProxyAttempts = hasNonReplayableProxyBody ? 1 : 2;
let lastProxyError: unknown = null;
for (let attempt = 0; attempt < maxProxyAttempts; attempt++) {
try {
return await _undiciProxy(input, {
...options,
dispatcher:
attempt === 0 ? createProxyDispatcher(proxyUrl) : getProxyRetryDispatcher(proxyUrl),
});
} catch (error) {
const msg = error instanceof Error ? error.message : String(error);
const errCode = (error as { code?: unknown })?.code;
const isTransportFailure =
msg.includes("fetch failed") ||
errCode === "ECONNREFUSED" ||
msg.includes("ECONNREFUSED") ||
(typeof errCode === "string" && errCode.startsWith("UND_ERR")) ||
msg.includes("UND_ERR");
if (attempt === 0 && maxProxyAttempts > 1 && isTransportFailure) {
lastProxyError = error;
// #9158: fixed OMNIROUTE_RETRY_BACKOFF_MS backoff — the retry uses a
// fresh no-keep-alive dispatcher (getProxyRetryDispatcher), so the old
// random jitter was pure latency on every recovered request with no
// herd risk (per-host pool).
await new Promise((r) => setTimeout(r, RETRY_BACKOFF_MS));
continue;
}
// A caller abort/timeout must propagate unchanged and without a noisy
// "Proxy request failed" log — it's not a proxy transport failure.
if (!isCallerAbort(error, options?.signal)) {
const message = error instanceof Error ? error.message : String(error);
console.error(`[ProxyFetch] Proxy request failed (${source}, fail-closed): ${message}`);
}
throw error;
}
throw error;
}
throw lastProxyError;
}
/**
@@ -726,4 +940,9 @@ export function getOriginalFetch(): typeof globalThis.fetch {
return originalFetch;
}
/** Test-only: exposes the relay Agent options for config assertions (#9100). */
export function __getRelayPoolAgentOptionsForTest() {
return RELAY_POOL_AGENT_OPTIONS;
}
export default isCloud ? originalFetch : patchedFetch;

11
tests/unit/direct-dispatcher-pipelining-4580.test.ts
View File

@@ -30,10 +30,17 @@ describe("#4580 direct dispatcher options", () => {
assert.equal(opts.connections, 32);
});
it("preserves keep-alive (NOT the 1ms TTL the proxy path forces)", () => {
it("preserves keep-alive (NOT the 1ms TTL the proxy path used to force)", () => {
const direct = __getDefaultDispatcherOptionsForTest({});
const proxy = __getProxyDispatcherOptionsForTest({});
assert.equal(proxy.keepAliveTimeout, 1);
// #9100: the proxy path no longer forces the 1ms keep-alive TTL — the
// regression that destroyed the pooled socket after every response and
// forced a fresh TCP+TLS+CONNECT handshake per request. Both paths now
// keep the socket alive for fetchKeepAliveTimeoutMs (default 4000ms).
assert.ok(
(proxy.keepAliveTimeout ?? 0) > 1,
`proxy keepAliveTimeout should stay > 1 (got ${proxy.keepAliveTimeout})`
);
assert.ok(
(direct.keepAliveTimeout ?? 0) > 1,
`direct keepAliveTimeout should stay > 1 (got ${direct.keepAliveTimeout})`

202
tests/unit/proxy-concurrency-keepalive-regression.test.ts Normal file
View File

@@ -0,0 +1,202 @@
import test from "node:test";
import assert from "node:assert/strict";
import net from "node:net";
import proxyFetch, {
runWithProxyContext,
__getRelayPoolAgentOptionsForTest,
} from "../../open-sse/utils/proxyFetch.ts";
import { clearDispatcherCache } from "../../open-sse/utils/proxyDispatcher.ts";
import { invalidateProxyHealth, isProxyReachable } from "../../src/lib/proxyHealth.ts";
// #9100 — proxy concurrency regression.
//
// Root cause: the proxy dispatcher forced keepAliveTimeout: 1 (1ms), destroying
// the pooled socket right after every response. Each request then paid a fresh
// TCP+TLS+CONNECT handshake, and a proxy that throttles connection churn
// serialized 5 concurrent requests behind ~30s stalls (1 fast + 4× ~29.5s).
// The fix restores keep-alive on the proxy path (default 30s keepAliveTimeout,
// keepAliveMaxTimeout 60s) so concurrent requests multiplex over ONE reused TCP
// connection per proxy host.
//
// This test proves the fix hermetically (loopback only, no real network):
// 1. 5 concurrent requests through a mocked HTTP proxy all resolve, and the
// counting TCP listener saw exactly ONE connection to the proxy host
// (keep-alive reuse — with the old 1ms TTL each queued request would have
// opened a fresh socket, i.e. 5 connections).
// 2. 5 concurrent requests through a mocked Vercel-relay proxy all resolve
// and all 5 share the SAME pooled dispatcher (one pool per relay host).
async function withEnv(overrides: Record<string, string | undefined>, fn: () => Promise<void>) {
const previous = new Map<string, string | undefined>();
for (const [key, value] of Object.entries(overrides)) {
previous.set(key, process.env[key]);
if (value === undefined) delete process.env[key];
else process.env[key] = value;
}
try {
await fn();
} finally {
for (const [key, value] of previous.entries()) {
if (value === undefined) delete process.env[key];
else process.env[key] = value;
}
}
}
/** Minimal HTTP-proxy TCP listener: CONNECT tunnel + short SSE upstream, keeps
* the socket alive so keep-alive reuse can be observed. Counts TCP connections. */
function startCountingProxyServer(): Promise<{
port: number;
connectionCount: () => number;
close: () => Promise<void>;
}> {
let connectionCount = 0;
const server = net.createServer((socket) => {
connectionCount += 1;
let buffer = Buffer.alloc(0);
let tunnelEstablished = false;
socket.on("data", (chunk: Buffer) => {
buffer = Buffer.concat([buffer, chunk]);
while (true) {
const headerEnd = buffer.indexOf("\r\n\r\n");
if (headerEnd === -1) break;
const head = buffer.subarray(0, headerEnd).toString("latin1");
buffer = buffer.subarray(headerEnd + 4);
if (!tunnelEstablished && head.startsWith("CONNECT ")) {
socket.write("HTTP/1.1 200 Connection Established\r\n\r\n");
tunnelEstablished = true;
continue;
}
// Short SSE upstream; keep-alive preserved so the next request reuses
// this socket (the whole point of the #9100 fix).
const body = 'data: {"ok":true}\n\n';
socket.write(
"HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nContent-Length: " +
Buffer.byteLength(body) +
"\r\nConnection: keep-alive\r\n\r\n" +
body
);
}
});
socket.on("error", () => {});
});
return new Promise((resolve, reject) => {
server.once("error", reject);
server.listen(0, "127.0.0.1", () => {
const address = server.address() as net.AddressInfo;
resolve({
port: address.port,
connectionCount: () => connectionCount,
close: () =>
new Promise<void>((res) => {
server.close(() => res());
}),
});
});
});
}
test.afterEach(() => {
clearDispatcherCache();
});
test("#9100: 5 concurrent requests through a mocked HTTP proxy all resolve over ONE reused TCP connection", async () => {
const proxy = await startCountingProxyServer();
try {
const proxyUrl = `http://127.0.0.1:${proxy.port}`;
// Warm the health cache BEFORE counting: the T14 probe (isProxyReachable)
// would otherwise open its own throwaway socket during the burst and pollute
// the connection count. With a healthy cached entry the burst reuses it.
invalidateProxyHealth(proxyUrl);
assert.equal(await isProxyReachable(proxyUrl, 120, 2_000), true);
const before = proxy.connectionCount();
// connections: 1 forces undici to QUEUE the 5 concurrent requests on a
// single pooled socket instead of fanning out one socket per request.
// With the old keepAliveTimeout: 1 the socket died after the first response
// and each queued request opened a fresh connection (count would be 5);
// with keep-alive restored all 5 reuse the same socket (count stays 1).
await withEnv({ OMNIROUTE_PROXY_DISPATCHER_CONNECTIONS: "1" }, async () => {
clearDispatcherCache();
const results = await Promise.all(
Array.from({ length: 5 }, (_, i) =>
runWithProxyContext({ type: "http", host: "127.0.0.1", port: String(proxy.port) }, () =>
proxyFetch(`http://proxy-target.invalid/v1/chat/completions?i=${i}`, {
signal: AbortSignal.timeout(10_000),
})
).then((r) => r.text())
)
);
for (const body of results) {
assert.ok(body.includes('"ok":true'), `expected SSE payload, got: ${body}`);
}
});
assert.equal(
proxy.connectionCount() - before,
1,
"5 concurrent proxied requests must reuse exactly ONE TCP connection to the proxy host"
);
// Release the pooled keep-alive sockets BEFORE closing the listener —
// otherwise server.close() waits ~keepAliveTimeout (30s) for them to idle out.
clearDispatcherCache();
} finally {
await proxy.close();
}
});
test("#9100: 5 concurrent requests through a mocked Vercel-relay proxy all resolve via ONE shared pooled dispatcher", async () => {
const relayCalls: Array<{ input: unknown; init: RequestInit & { dispatcher?: unknown } }> = [];
const relaySink = (async (input: unknown, init: RequestInit = {}) => {
relayCalls.push({ input, init });
// Short SSE upstream, mirroring what the edge relay would return.
return new Response('data: {"ok":true}\n\n', {
status: 200,
headers: { "Content-Type": "text/event-stream" },
});
}) as never;
const VERCEL_CTX = {
type: "vercel" as const,
host: "omniroute-relay-abc123.vercel.app",
relayAuth: "live-relay-secret",
};
const results = await Promise.all(
Array.from({ length: 5 }, (_, i) =>
runWithProxyContext(VERCEL_CTX, () =>
proxyFetch(
`https://api.anthropic.com/v1/messages?x=${i}`,
{ method: "POST", headers: { "x-existing": "keep-me" } },
{ undiciFetch: relaySink }
)
).then((r) => r.text())
)
);
for (const body of results) {
assert.ok(body.includes('"ok":true'), `expected SSE payload, got: ${body}`);
}
assert.equal(relayCalls.length, 5, "all 5 requests must reach the relay");
const dispatchers = new Set(relayCalls.map((c) => c.init.dispatcher));
assert.equal(
dispatchers.size,
1,
"all 5 relay requests must share the SAME pooled dispatcher (one TCP connection per relay host)"
);
// #9158: the relay Agent pools FOUR connections per host and multiplexes
// concurrent requests over them (h2). Pooling 4 sockets removes the
// head-of-line blocking a single connection caused for parallel SSE streams,
// while `allowH2: true` keeps queued requests running in parallel as h2
// streams across the pool.
const relayAgentOptions = __getRelayPoolAgentOptionsForTest();
assert.equal(relayAgentOptions.connections, 4, "relay agent must pool four connections per host");
assert.equal(
relayAgentOptions.allowH2,
true,
"relay agent must multiplex concurrent requests over h2"
);
});

31
tests/unit/proxy-dispatcher-cache-cap.test.ts Normal file
View File

@@ -0,0 +1,31 @@
import { describe, it } from "node:test";
import assert from "node:assert/strict";
import {
setDispatcherCacheEntry,
getDispatcherCache,
clearDispatcherCache,
} from "../../open-sse/utils/proxyDispatcherCache.ts";
describe("proxy dispatcher cache cap (#9158)", () => {
it("evicts and closes the oldest entry at the 512-entry cap", () => {
let closedCount = 0;
const fakeDispatcher = {
close() {
closedCount += 1;
return Promise.resolve();
},
} as never;
for (let i = 0; i <= 512; i += 1) {
setDispatcherCacheEntry(`u${i}`, fakeDispatcher);
}
assert.equal(getDispatcherCache().size, 512, "cache must stay at the cap");
assert.equal(closedCount, 1, "exactly one entry must be evicted and closed");
assert.equal(getDispatcherCache().has("u0"), false, "oldest entry must be evicted");
assert.equal(getDispatcherCache().has("u512"), true, "newest entry must be retained");
clearDispatcherCache();
});
});

11
tests/unit/proxy-dispatcher-family.test.ts
View File

@@ -68,9 +68,14 @@ 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);
// #9100: the proxy path now keeps sockets alive (no more 1ms TTL) and
// pipelines up to 4 requests so concurrent SSE streams multiplex over one
// pooled TCP connection per proxy host. Stale sockets are recovered by the
// retry-once-with-fresh-socket path in proxyFetch, not by killing idle
// sockets after 1ms.
assert.equal(options.pipelining, 4);
assert.equal(options.keepAliveTimeout, 30000);
assert.equal(options.keepAliveMaxTimeout, 60000);
});
it("allows operators to force a single proxy connection for diagnostics", () => {

76
tests/unit/proxy-family-resolve-cache.test.ts Normal file
View File

@@ -0,0 +1,76 @@
import { describe, it, beforeEach } from "node:test";
import assert from "node:assert/strict";
import {
assertHostnameSupportsFamily,
__clearFamilyCheckCacheForTest,
} from "../../open-sse/utils/proxyFamilyResolve.ts";
describe("proxyFamilyResolve DNS family cache (#9158)", () => {
beforeEach(() => __clearFamilyCheckCacheForTest());
it("caches a positive family check so repeated calls reuse DNS", async () => {
let lookupCount = 0;
const lookupFn = async () => {
lookupCount += 1;
return [{ address: "::1", family: 6 }];
};
await assertHostnameSupportsFamily("relay.example.test", 6, lookupFn);
await assertHostnameSupportsFamily("relay.example.test", 6, lookupFn);
assert.equal(lookupCount, 1, "second call must hit the positive cache");
});
it("re-probes after the test hook clears the cache", async () => {
let lookupCount = 0;
const lookupFn = async () => {
lookupCount += 1;
return [{ address: "::1", family: 6 }];
};
await assertHostnameSupportsFamily("relay.example.test", 6, lookupFn);
__clearFamilyCheckCacheForTest();
await assertHostnameSupportsFamily("relay.example.test", 6, lookupFn);
assert.equal(lookupCount, 2, "cleared cache must re-invoke DNS");
});
it("caches a negative result without re-invoking DNS", async () => {
let lookupCount = 0;
const failingFn = async () => {
lookupCount += 1;
throw new Error("boom");
};
await assert.rejects(
assertHostnameSupportsFamily("relay.example.test", 6, failingFn),
/resolution/i
);
await assert.rejects(
assertHostnameSupportsFamily("relay.example.test", 6, failingFn),
/resolution/i
);
assert.equal(lookupCount, 1, "negative result must be cached");
});
it("no-ops for bracketed IPv6 literals without any DNS lookup", async () => {
const failingFn = async () => {
throw new Error("must not be called");
};
await assertHostnameSupportsFamily("[2001:db8::1]", 6, failingFn);
});
it("keys the cache by lookup function so a different resolver re-probes", async () => {
let countA = 0;
const fnA = async () => {
countA += 1;
return [{ address: "::1", family: 6 }];
};
let countB = 0;
const fnB = async () => {
countB += 1;
return [{ address: "::1", family: 6 }];
};
await assertHostnameSupportsFamily("relay.example.test", 6, fnA);
await assertHostnameSupportsFamily("relay.example.test", 6, fnA);
await assertHostnameSupportsFamily("relay.example.test", 6, fnB);
assert.equal(countA, 1);
assert.equal(countB, 1, "different lookupFn must bypass the cached result");
});
});

52
tests/unit/proxy-fetch.test.ts
View File

@@ -169,14 +169,25 @@ test("runWithProxyContext accepts reachable HTTP proxy endpoints and returns cal
test("runWithProxyContext throws PROXY_UNREACHABLE for an unreachable proxy by default", async () => {
// 127.0.0.1:9 (discard) refuses connections — the proxy is unreachable.
await assert.rejects(
runWithProxyContext({ type: "http", host: "127.0.0.1", port: "9" }, async () => "unreachable"),
(err: Error & { code?: string; errorCode?: string }) => {
assert.equal(err.code, "PROXY_UNREACHABLE");
assert.equal(err.errorCode, "proxy_unreachable");
return true;
}
);
// #9100: the T14 probe is non-blocking, so the request must stay in flight
// long enough for the probe to resolve unreachable and abort it (an
// instantly-resolving callback would simply win the race and return).
let releaseRequest: () => void = () => {};
const gate = new Promise<void>((resolve) => {
releaseRequest = resolve;
});
const pending = runWithProxyContext({ type: "http", host: "127.0.0.1", port: "9" }, async () => {
await gate;
return "unreachable";
});
await assert.rejects(pending, (err: Error & { code?: string; errorCode?: string }) => {
assert.equal(err.code, "PROXY_UNREACHABLE");
assert.equal(err.errorCode, "proxy_unreachable");
return true;
});
releaseRequest();
});
test("runWithProxyContext degrades to a direct connection when directFallbackOnUnreachable is set", async () => {
@@ -198,14 +209,25 @@ test("runWithProxyContext degrades to a direct connection when directFallbackOnU
test("runWithProxyContext keeps strict pinning when the direct fallback feature flag is off", async () => {
await withEnv({ OMNIROUTE_CONTROL_PLANE_PROXY_DIRECT_FALLBACK: "false" }, async () => {
await assert.rejects(
runWithProxyContext(
{ type: "http", host: "127.0.0.1", port: "9" },
async () => "unreachable",
{ directFallbackOnUnreachable: true }
),
/Proxy unreachable/
// #9100: with the flag off the request goes through the non-blocking T14
// probe; keep it in flight so the unreachable probe aborts it (strict
// pinning — no direct fallback — still applies).
let releaseRequest: () => void = () => {};
const gate = new Promise<void>((resolve) => {
releaseRequest = resolve;
});
const pending = runWithProxyContext(
{ type: "http", host: "127.0.0.1", port: "9" },
async () => {
await gate;
return "unreachable";
},
{ directFallbackOnUnreachable: true }
);
await assert.rejects(pending, /Proxy unreachable/);
releaseRequest();
});
});

34
tests/unit/proxy-nested-context-skip.test.ts Normal file
View File

@@ -0,0 +1,34 @@
import { describe, it } from "node:test";
import assert from "node:assert/strict";
import { runWithProxyContext } from "../../open-sse/utils/proxyFetch.ts";
import { proxyConfigToUrl } from "../../open-sse/utils/proxyDispatcher.ts";
import {
invalidateProxyHealth,
__setProxyHealthTcpCheckForTesting,
} from "../../src/lib/proxyHealth.ts";
describe("runWithProxyContext nested same-context skip (#9158)", () => {
it("skips the reachability probe for a nested same-config call", async () => {
const cfg = { type: "http" as const, host: "127.0.0.1", port: 8080 };
const proxyUrl = proxyConfigToUrl(cfg)!;
assert.ok(proxyUrl);
let probeCount = 0;
__setProxyHealthTcpCheckForTesting(async () => {
probeCount += 1;
return true;
});
try {
const result = await runWithProxyContext(cfg, () => {
invalidateProxyHealth(proxyUrl);
return runWithProxyContext(cfg, () => "nested-marker");
});
assert.equal(result, "nested-marker");
assert.equal(probeCount, 1, "nested same-context call must skip the reachability probe");
} finally {
__setProxyHealthTcpCheckForTesting(null);
invalidateProxyHealth(proxyUrl);
}
});
});

25
tests/unit/proxy-pool-cloudflare-workers-deployer.test.ts
View File

@@ -66,10 +66,7 @@ test("buildCloudflareWorkerScript rejects requests without a valid x-relay-auth
// a 401 short-circuit when x-relay-auth does not match the embedded token.
// We don't run the worker here — we check the source contains the guard.
const src = buildCloudflareWorkerScript("the-secret");
assert.ok(
/x-relay-auth/.test(src),
"worker source must reference the x-relay-auth header"
);
assert.ok(/x-relay-auth/.test(src), "worker source must reference the x-relay-auth header");
assert.ok(
/401|Unauthorized/.test(src),
"worker source must short-circuit unauthorised requests with 401"
@@ -116,11 +113,18 @@ const CLOUDFLARE_CTX = {
};
test("proxyFetch routes a cloudflare-type context through the relay endpoint with relay headers", async () => {
// #9100: the relay branch now egresses through the pooled undici Agent
// (deps.undiciFetch) instead of `originalFetch`, so the test injects the
// relay sink via deps to keep the dispatch hermetic.
const response = await runWithProxyContext(CLOUDFLARE_CTX, () =>
proxyFetch("https://api.anthropic.com/v1/messages?x=1", {
method: "POST",
headers: { "x-existing": "keep-me" },
})
proxyFetch(
"https://api.anthropic.com/v1/messages?x=1",
{
method: "POST",
headers: { "x-existing": "keep-me" },
},
{ undiciFetch: relaySink as never }
)
);
assert.deepEqual(await response.json(), { via: "cloudflare-relay" });
@@ -180,10 +184,7 @@ test("proxyConfigToUrl returns the cloudflare worker URL (no HTTP-proxy dispatch
// --------------------------------------------------------------------------
test("buildVercelRelayHeaders is the shared relay-header builder used for cloudflare too", () => {
const headers = buildVercelRelayHeaders(
"https://api.openai.com/v1/chat/completions",
"cf-tok"
);
const headers = buildVercelRelayHeaders("https://api.openai.com/v1/chat/completions", "cf-tok");
assert.deepEqual(headers, {
"x-relay-target": "https://api.openai.com",
"x-relay-path": "/v1/chat/completions",

15
tests/unit/proxy-pool-deno-deploy-relay.test.ts
View File

@@ -67,11 +67,18 @@ test("proxyFetch routes a deno-type context through the relay endpoint with rela
relayAuth: "deno-relay-secret",
};
// #9100: the relay branch now egresses through the pooled undici Agent
// (deps.undiciFetch) instead of `originalFetch`, so the test injects the
// relay sink via deps to keep the dispatch hermetic.
const response = await runWithProxyContext(DENO_CTX, () =>
proxyFetch("https://api.anthropic.com/v1/messages?x=1", {
method: "POST",
headers: { "x-existing": "keep-me" },
})
proxyFetch(
"https://api.anthropic.com/v1/messages?x=1",
{
method: "POST",
headers: { "x-existing": "keep-me" },
},
{ undiciFetch: relaySink as never }
)
);
assert.deepEqual(await response.json(), { via: "deno-relay" });

15
tests/unit/proxyfetch-vercel-relay-2743.test.ts
View File

@@ -111,11 +111,18 @@ const VERCEL_CTX = {
};
test("proxyFetch routes a vercel-type context through the relay endpoint with relay headers", async () => {
// #9100: the relay branch now egresses through the pooled undici Agent
// (deps.undiciFetch) instead of `originalFetch`, so the test injects the
// relay sink via deps to keep the dispatch hermetic.
const response = await runWithProxyContext(VERCEL_CTX, () =>
proxyFetch("https://api.anthropic.com/v1/messages?x=1", {
method: "POST",
headers: { "x-existing": "keep-me" },
})
proxyFetch(
"https://api.anthropic.com/v1/messages?x=1",
{
method: "POST",
headers: { "x-existing": "keep-me" },
},
{ undiciFetch: relaySink as never }
)
);
// The canned relay response proves the relay sink (originalFetch) was hit and the

22
tests/unit/rerank-proxy-pinning-7350.test.ts
View File

@@ -29,9 +29,10 @@ const { handleRerank } = await import("../../open-sse/handlers/rerank.ts");
const originalFetch = globalThis.fetch;
/** Captures the proxy URL visible inside the dispatch context at fetch time. */
function stubFetch(seen: { proxyUrl: string | null | undefined }[]) {
function stubFetch(seen: { proxyUrl: string | null | undefined }[], gate?: () => Promise<void>) {
globalThis.fetch = (async () => {
seen.push({ proxyUrl: proxyFetch.getCurrentProxyUrlForTests?.() ?? undefined });
if (gate) await gate();
return new Response(
JSON.stringify({ data: [{ index: 0, relevance_score: 0.9 }], model: "rerank-2" }),
{ status: 200, headers: { "Content-Type": "application/json" } }
@@ -63,10 +64,20 @@ test("#7350 handleRerank routes the upstream call through the connection's pinne
// The stub only ever answers a DIRECT call: runWithProxyContext dispatches through
// undici with the pinned proxy agent instead, so a pinned-but-unreachable proxy is
// observable as "the stub was bypassed and the request did not succeed". That
// observable as "the request did not succeed through the direct stub". That
// difference IS the wiring — before #7350 this call egressed directly and got a 200.
//
// #9100: the T14 reachability probe is now NON-BLOCKING — dispatch is optimistic, so
// fn() (and hence this stub) runs immediately. To still observe the dead-proxy
// failure the stub must stay pending long enough for the probe (fast NXDOMAIN /
// ECONNREFUSED on rerank-egress.local) to resolve unreachable and abort the
// in-flight request with PROXY_UNREACHABLE instead of a direct 200.
const seen: { proxyUrl: string | null | undefined }[] = [];
stubFetch(seen);
let releaseStub: () => void = () => {};
const stubGate = new Promise<void>((resolve) => {
releaseStub = resolve;
});
stubFetch(seen, () => stubGate);
const res = (await handleRerank({
model: "voyage/rerank-2",
@@ -76,12 +87,15 @@ test("#7350 handleRerank routes the upstream call through the connection's pinne
connectionId: (conn as { id: string }).id,
})) as Response;
assert.equal(seen.length, 0, "a pinned proxy must bypass the direct-egress path entirely");
// Optimistic dispatch (#9100) — the request IS started; what must NOT happen is
// the stub's direct 200 winning over the unreachable-proxy abort.
assert.equal(seen.length, 1, "a pinned proxy dispatches optimistically (non-blocking probe)");
assert.notEqual(
res.status,
200,
"the unreachable pinned proxy must surface as a failure rather than silently egressing direct"
);
releaseStub();
});
test("#7350 an unresolvable connectionId degrades to a direct call instead of failing the request", async () => {

34
tests/unit/t14-proxy-fast-fail.test.ts
View File

@@ -41,7 +41,10 @@ test("#5109: concurrent proxy reachability checks share one TCP probe", async ()
assert.equal(probeCount, 1, "concurrent requests must not fan out TCP health probes");
releaseProbe(true);
assert.deepEqual(await Promise.all(checks), Array.from({ length: 50 }, () => true));
assert.deepEqual(
await Promise.all(checks),
Array.from({ length: 50 }, () => true)
);
assert.equal(getCachedProxyHealth(proxyUrl), true);
} finally {
__setProxyHealthTcpCheckForTesting(null);
@@ -74,19 +77,28 @@ test("#5109: transient unreachable results use a short negative cache", async ()
}
});
test("T14: runWithProxyContext fast-fails when proxy is unreachable", async () => {
test("T14: runWithProxyContext fails an in-flight request fast when the proxy is unreachable", async () => {
const proxyUrl = "http://127.0.0.1:1";
invalidateProxyHealth(proxyUrl);
// #9100: the T14 probe is now NON-BLOCKING — dispatch is optimistic and the
// probe aborts the request only while it is still in flight. To observe the
// fast-fail the callback must stay pending long enough for the probe to
// resolve (a callback that resolves instantly would simply win the race).
let executed = false;
await assert.rejects(
() =>
runWithProxyContext(proxyUrl, async () => {
executed = true;
return "ok";
}),
(err) => (err as { code?: string })?.code === "PROXY_UNREACHABLE"
);
let releaseRequest: () => void = () => {};
const gate = new Promise<void>((resolve) => {
releaseRequest = resolve;
});
assert.equal(executed, false);
const pending = runWithProxyContext(proxyUrl, async () => {
executed = true;
await gate; // stay in flight until the probe resolves unreachable
return "ok";
});
await assert.rejects(pending, (err) => (err as { code?: string })?.code === "PROXY_UNREACHABLE");
assert.equal(executed, true, "dispatch is optimistic; the request was started before the abort");
releaseRequest();
});