diff --git a/docs-site/src/content/docs/reference/configuration/server.md b/docs-site/src/content/docs/reference/configuration/server.md index 0d2c008a47..6b6f280120 100644 --- a/docs-site/src/content/docs/reference/configuration/server.md +++ b/docs-site/src/content/docs/reference/configuration/server.md @@ -143,6 +143,28 @@ Compare the diagnostic on the same machine and account under the two network modes. A successful TUN test alone does not identify why the service's HTTP proxy path failed, and does not establish a general fix. +## Connection reuse for specific upstream hosts + +Some upstreams keep a connection open after they have stopped serving it. The next request reuses +that pooled socket and fails without reaching the provider. `OCX_FRESH_CONNECTION_HOSTS` names the +hosts that should never reuse a pooled connection, as an environment variable rather than a config +field, so it can be applied to one machine without editing shared configuration: + +```bash +OCX_FRESH_CONNECTION_HOSTS="api.example.com, relay.example.net" ocx start +``` + +The value is a comma-separated list of hostnames. Matching is case-insensitive, covers each named +host and its subdomains, and ignores leading dots, so `.example.com` and `example.com` both match +`api.example.com`. Do not include a scheme, port or path. An unset or empty variable leaves the +default connection behavior unchanged. + +A matching send carries `Connection: close` and is dispatched with keep-alive disabled. The +decision is made against the address actually used on the wire, so it still applies when a provider +transport rewrites the destination after credential selection. Per-request latency rises slightly +for those hosts, since each request pays a fresh TCP and TLS handshake; name only the hosts that +need it. + ## Remote access The default `127.0.0.1` bind is loopback-only. A non-loopback address such as `0.0.0.0` or a tailnet diff --git a/src/server/chat-native.ts b/src/server/chat-native.ts index 837b112e2f..69361aa9f9 100644 --- a/src/server/chat-native.ts +++ b/src/server/chat-native.ts @@ -50,7 +50,7 @@ import { enrichOpenCodeZenFreeTierMessage } from "../providers/opencode-zen-rate import type { OcxProviderTransport } from "../providers/xai-transport"; import type { RouteResult } from "../router"; import type { OcxConfig, OcxProviderConfig } from "../types"; -import { fetchWithHeaderTimeout, providerFetch, safeHostLabel } from "./responses/fetch-helpers"; +import { fetchWithHeaderTimeout, providerFetch, safeHostLabel, sendWithConnectionPolicy } from "./responses/fetch-helpers"; import { linkAbortSignal } from "./responses"; import { addFinalRequestLog, @@ -351,9 +351,15 @@ export async function handleNativeChatCompletions(options: HandleNativeChatOptio if (!headers.has("accept-encoding") && encoding) headers.set("accept-encoding", encoding); if (init.signal?.aborted) throw init.signal.reason; noteProviderAttemptSend(logCtx, route.providerName, activeProvider, logCtx.usageLogInputTokens, transportRecovery ?? recovery); - const dispatched = await ((activeProvider as OcxProviderTransport).fetch ?? execute)(request.url, applyUpstreamRecoveryInit({ - ...init, method: request.method, headers, body: request.body, - }, transportRecovery)); + // A reselected provider transport is still a physical send: the connection policy + // and manual-redirect ownership wrap the selected implementation (#4992). + const dispatched = await sendWithConnectionPolicy( + (activeProvider as OcxProviderTransport).fetch ?? execute, + request.url, + applyUpstreamRecoveryInit({ + ...init, method: request.method, headers, body: request.body, + }, transportRecovery), + ); if (!dispatched.ok) await recordKeyAttemptFailure(logCtx, dispatched, init.signal ?? upstream.signal); return dispatched; }, diff --git a/src/server/responses/fetch-helpers.ts b/src/server/responses/fetch-helpers.ts index d2ce734da4..b2e4496974 100644 --- a/src/server/responses/fetch-helpers.ts +++ b/src/server/responses/fetch-helpers.ts @@ -79,6 +79,38 @@ export interface PaceAwareFetch { export type ProviderFetch = typeof globalThis.fetch & PaceAwareFetch; +/** + * Apply the physical-send connection policy to whichever fetch actually performs the send. + * + * The executor `providerFetch` builds is not the only physical boundary. A `dispatchOverride` + * that revalidates credentials re-reads `route.provider.fetch` at send time -- reselection can + * install a different provider transport after this wrapper was constructed -- and then calls + * that fetch directly instead of the supplied executor. Keeping the policy inside the executor + * alone therefore left every provider-scoped transport reusing a pooled socket for a host the + * operator had named in `OCX_FRESH_CONNECTION_HOSTS` (#4992). The policy belongs around the + * selected fetch so it follows the selection rather than the construction. + * + * Idempotent on purpose: an override that hands the send back to the supplied executor passes + * through here twice, and both passes derive the same headers from the same wire URL. + */ +export function sendWithConnectionPolicy( + physicalFetch: typeof globalThis.fetch, + input: Parameters[0], + init?: RequestInit, +): Promise { + const headers = new Headers(init?.headers ?? (input instanceof Request ? input.headers : undefined)); + const fresh = wantsFreshConnection(input); + if (fresh) { + headers.set("Connection", "close"); + } + return physicalFetch(input, { + ...init, + headers, + redirect: "manual", + ...(fresh ? { keepalive: false } : {}), + }); +} + export interface ProviderFetchOptions { nativeControl?: NativeResponseControl; providerName?: string; @@ -109,19 +141,8 @@ export function providerFetch( // Rebuilt dispatches must use the same physical-send boundary as ordinary HTTP sends. // Return the original 3xx so the response owner retains its retry/health/relay contract. const dispatch = Object.assign( - (input: Parameters[0], init?: RequestInit) => { - const headers = new Headers(init?.headers ?? (input instanceof Request ? input.headers : undefined)); - const fresh = wantsFreshConnection(input); - if (fresh) { - headers.set("Connection", "close"); - } - return base(input, { - ...init, - headers, - redirect: "manual", - ...(fresh ? { keepalive: false } : {}), - }); - }, + (input: Parameters[0], init?: RequestInit) => + sendWithConnectionPolicy(base, input, init), { preconnect }, ) as typeof globalThis.fetch; const httpFetch = Object.assign( diff --git a/src/server/responses/request-transport.ts b/src/server/responses/request-transport.ts index 3e878959e2..2cc5baf9a7 100644 --- a/src/server/responses/request-transport.ts +++ b/src/server/responses/request-transport.ts @@ -41,7 +41,7 @@ import { resolveCurrentProviderApiKeyTransport, } from "../../providers/api-key-selection"; import { resolveAdapter, resolveWireProtocolOverride } from "../adapter-resolve"; -import { providerFetch } from "./fetch-helpers"; +import { providerFetch, sendWithConnectionPolicy } from "./fetch-helpers"; import type { ProviderFetchOptions } from "./fetch-helpers"; import { captureConfigGeneration } from "../../lib/state-store-sweeper"; import { recordAnthropicAccountQuotaFromHeaders, hasPassiveAccountQuota } from "../../providers/quota"; @@ -405,8 +405,10 @@ export async function prepareResponsesTransport( && sentHeaders?.get("authorization") === `Bearer ${snapshot.accessToken}` && !sentHeaders?.has("x-api-key"); // Reselection can choose a provider override instead of the supplied executor. + // Either way the send crosses the physical boundary, so the connection policy is + // applied around whichever implementation was just selected (#4992). commitKeyAttemptSend(); - const response = await fetchImpl(destination, { ...dispatchInit, redirect: "manual" }); + const response = await sendWithConnectionPolicy(fetchImpl, destination, { ...dispatchInit, redirect: "manual" }); if (!response.ok) await recordKeyAttemptFailure(logCtx, response, dispatchInit.signal ?? options.abortSignal); // Observe each physical response before retries replace it. The binding belongs to // this dispatch, so a manual switch cannot file A's headers against B. Header diff --git a/structure/transports/responses.md b/structure/transports/responses.md index c508528628..e30a6424fb 100644 --- a/structure/transports/responses.md +++ b/structure/transports/responses.md @@ -45,9 +45,20 @@ modules merely because those imports existed in the pre-split `responses.ts` mon `OCX_FRESH_CONNECTION_HOSTS` accepts comma-separated hostnames whose outbound HTTP sends bypass keep-alive reuse with `Connection: close` and `keepalive: false`; exact hosts and their subdomains -match case-insensitively. The helper applies this policy at the final executor boundary, after a -dispatch override has selected or rebuilt the destination, so matching follows the URL sent on the -wire rather than the URL supplied before credential revalidation. +match case-insensitively. `sendWithConnectionPolicy` applies the policy around the fetch that +performs the physical send, after a dispatch override has selected or rebuilt the destination, so +matching follows the URL sent on the wire rather than the URL supplied before credential +revalidation. + +The wrapped executor alone is not that boundary. An override that revalidates credentials re-reads +`route.provider.fetch` at send time, because reselection can install a different provider transport +after the wrapper was built, and then calls that implementation instead of the executor. Both +production overrides do this -- `oauthDispatch` in `request-transport.ts` and the native Chat +key-revalidation override in `chat-native.ts` -- so both wrap the selected implementation rather +than choosing between policy and provider transport. Reporting the executor as the boundary while +the code let a provider-scoped transport past it is what #4992 recorded, and it is why a +regression for this policy has to enter through `handleResponses` rather than through a +hand-written override that cooperates by calling the executor it was handed. ### Semantic progress ownership diff --git a/tests/responses/fresh-connection-optout.test.ts b/tests/responses/fresh-connection-optout.test.ts index 97f9bd7607..944afbd222 100644 --- a/tests/responses/fresh-connection-optout.test.ts +++ b/tests/responses/fresh-connection-optout.test.ts @@ -1,6 +1,14 @@ import { describe, expect, test } from "bun:test"; +import { mkdtempSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; import { wantsFreshConnection, providerFetch } from "../../src/server/responses/fetch-helpers"; -import type { OcxProviderConfig } from "../../src/types"; +import { saveCredential } from "../../src/oauth/store"; +import { XAI_GROK_CLI_BASE_URL } from "../../src/providers/xai-transport"; +import { handleResponses } from "../../src/server/responses"; +import type { RequestLogContext } from "../../src/server/request-log"; +import type { OcxConfig, OcxProviderConfig } from "../../src/types"; +import { removeTreeWithRetry } from "../helpers/remove-tree"; describe("wantsFreshConnection", () => { test("returns false when env is unset or empty", () => { @@ -246,3 +254,90 @@ describe("providerFetch fresh connection dispatch", () => { } }); }); + +/** + * The cases above drive `providerFetch` with an override that cooperates by calling the executor + * it was handed. Production's OAuth override does not: it re-reads `route.provider.fetch` at the + * send boundary, because credential reselection can install a different provider transport after + * the wrapper was built, and calls that implementation directly. Every test above still passed + * while the operator's configured host reused a pooled socket on that path (#4992), so the + * regression has to enter through `handleResponses` rather than through a hand-written override. + * + * xAI OAuth is the live consumer: `resolveProviderTransport` installs a provider-scoped fetch and + * rewrites the destination to the Grok CLI host, which is what makes it observable here. + */ +describe("the OAuth dispatch boundary", () => { + test("a provider-scoped transport selected at dispatch receives the fresh-connection policy", async () => { + const freshHost = new URL(XAI_GROK_CLI_BASE_URL).hostname; + const home = mkdtempSync(join(tmpdir(), "ocx-fresh-connection-oauth-")); + const nativeFetch = globalThis.fetch; + const previousHosts = process.env.OCX_FRESH_CONNECTION_HOSTS; + const previousOpencodexHome = process.env.OPENCODEX_HOME; + const previousCodexHome = process.env.CODEX_HOME; + process.env.OPENCODEX_HOME = home; + process.env.CODEX_HOME = home; + process.env.OCX_FRESH_CONNECTION_HOSTS = freshHost; + const sends: Array<{ url: string; init?: RequestInit }> = []; + + try { + await saveCredential("xai", { + access: "xai-access-token", + refresh: "xai-refresh-token", + expires: Date.now() + 3_600_000, + accountId: "xai-acct-1", + source: "local-cli", + }); + globalThis.fetch = (async (input: Parameters[0], init?: RequestInit) => { + const url = typeof input === "string" ? input : input instanceof URL ? input.href : input.url; + sends.push({ url, init }); + return Response.json({ + id: "resp_fresh_connection", + status: "completed", + output: [], + usage: { input_tokens: 10, output_tokens: 5, total_tokens: 15 }, + }); + }) as typeof globalThis.fetch; + + const config = { + defaultProvider: "xai", + providers: { + xai: { adapter: "openai-chat", baseUrl: "https://api.x.ai/v1", authMode: "oauth" }, + }, + } as OcxConfig; + const logCtx: RequestLogContext = { model: "", provider: "" }; + const response = await handleResponses( + new Request("http://localhost/v1/responses", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ model: "grok-4.6", input: "hello", stream: false }), + }), + config, + logCtx, + {}, + ); + + expect(response.status).toBe(200); + // Asserted on the host rather than a path, and on the mapped list rather than a filtered + // one, so a destination change reports the addresses it observed instead of an empty length. + expect(sends.map(send => new URL(send.url).hostname)).toContain(freshHost); + const policed = sends.filter(send => new URL(send.url).hostname === freshHost); + for (const send of policed) { + const headers = new Headers(send.init?.headers); + expect(headers.get("Connection")).toBe("close"); + expect((send.init as { keepalive?: boolean } | undefined)?.keepalive).toBe(false); + // And the provider's own implementation still ran: only the xAI wrapper pins this header, + // so wrapping the selected fetch did not replace it with the generic executor. + expect(headers.get("x-grok-req-id")).toBeTruthy(); + } + } finally { + globalThis.fetch = nativeFetch; + if (previousHosts === undefined) delete process.env.OCX_FRESH_CONNECTION_HOSTS; + else process.env.OCX_FRESH_CONNECTION_HOSTS = previousHosts; + if (previousOpencodexHome === undefined) delete process.env.OPENCODEX_HOME; + else process.env.OPENCODEX_HOME = previousOpencodexHome; + if (previousCodexHome === undefined) delete process.env.CODEX_HOME; + else process.env.CODEX_HOME = previousCodexHome; + removeTreeWithRetry(home); + } + }); +});