diff --git a/docs/openai-compatible-providers.md b/docs/openai-compatible-providers.md index a874d227..2456d98a 100644 --- a/docs/openai-compatible-providers.md +++ b/docs/openai-compatible-providers.md @@ -327,6 +327,14 @@ That flag is respected by both: - CCS now preserves upstream rate-limit errors and retry headers - Empty or malformed provider JSON is returned as Anthropic-style `api_error` +### Slow upstreams: `socket connection was closed unexpectedly` + +- Long-running upstreams (self-hosted LLMs with queue/prefill phases) can stay + silent for minutes before the first or next token. The proxy already allows up + to 10 minutes per request; set `CCS_OPENAI_PROXY_REQUEST_TIMEOUT_MS` in the + profile settings to raise or lower that ceiling. +- Restart the proxy after changing the setting. + ### Requests route to the wrong model/profile - Use an explicit selector such as `profile:model` diff --git a/src/proxy/server/messages-route.ts b/src/proxy/server/messages-route.ts index ab99b5e0..4a5f2f93 100644 --- a/src/proxy/server/messages-route.ts +++ b/src/proxy/server/messages-route.ts @@ -10,9 +10,16 @@ import { import { ProxySseStreamTransformer } from '../transformers/sse-stream-transformer'; import { isAnthropicPassthroughProfile, resolveOpenAIChatCompletionsUrl } from '../upstream-url'; import { createLogger } from '../../services/logging'; +import { + createGlobalFetchProxyDispatcher, + type UpstreamAgentTimeoutOptions, +} from '../../utils/fetch-proxy-setup'; import { pipeWebResponseToNode, readRawBody, writeJson } from './http-helpers'; const REQUEST_TIMEOUT_MS = 600_000; +// Keep undici's per-phase timeouts above the explicit request timeout so the +// AbortController is the single authority on when an upstream request dies. +const UPSTREAM_TIMEOUT_GRACE_MS = 30_000; const DIRECT_OPENAI_REASONING_CHAT_MODEL = /^(?:gpt-5|o[134])(?:[-.]|$)/; const logger = createLogger('proxy:openai-compat:messages'); @@ -375,7 +382,7 @@ function buildFetchInit( signal: AbortSignal, incomingHeaders: http.IncomingHttpHeaders, preserveUserAgent: boolean, - insecureDispatcher?: Dispatcher + dispatcher?: Dispatcher ): RequestInit { const init: RequestInit = { method: 'POST', @@ -384,8 +391,8 @@ function buildFetchInit( signal, }; - if (insecureDispatcher) { - (init as Record).dispatcher = insecureDispatcher; + if (dispatcher) { + (init as Record).dispatcher = dispatcher; } return init; @@ -401,6 +408,30 @@ function getRequestTimeoutMs(): number { return Number.isFinite(parsed) && parsed > 0 ? parsed : REQUEST_TIMEOUT_MS; } +/** + * undici defaults `headersTimeout`/`bodyTimeout` to 300s, which silently + * undercuts {@link REQUEST_TIMEOUT_MS} (600s) and the + * `CCS_OPENAI_PROXY_REQUEST_TIMEOUT_MS` override: slow upstreams (self-hosted + * LLMs with long queue + prefill phases) get their socket closed at 300s with + * a generic connection error instead of the proxy's timeout response. Every + * dispatcher used for upstream fetches must carry these options. + */ +export function buildUpstreamAgentTimeouts(): UpstreamAgentTimeoutOptions { + const ceiling = getRequestTimeoutMs() + UPSTREAM_TIMEOUT_GRACE_MS; + return { headersTimeout: ceiling, bodyTimeout: ceiling }; +} + +let defaultUpstreamDispatcher: Dispatcher | null = null; + +function getDefaultUpstreamDispatcher(): Dispatcher { + if (!defaultUpstreamDispatcher) { + const timeouts = buildUpstreamAgentTimeouts(); + // Honor HTTP(S)_PROXY routing when configured; otherwise a plain Agent. + defaultUpstreamDispatcher = createGlobalFetchProxyDispatcher(timeouts) ?? new Agent(timeouts); + } + return defaultUpstreamDispatcher; +} + function formatTimeoutDuration(timeoutMs: number): string { return timeoutMs % 1000 === 0 ? `${timeoutMs / 1000} seconds` : `${timeoutMs}ms`; } @@ -538,19 +569,20 @@ export async function handleProxyMessagesRequest( const useProfileInsecureTls = upstream.route.profile.insecure === true; const ephemeralInsecureDispatcher = useProfileInsecureTls && !useSharedInsecureDispatcher - ? new Agent({ connect: { rejectUnauthorized: false } }) + ? new Agent({ connect: { rejectUnauthorized: false }, ...buildUpstreamAgentTimeouts() }) : undefined; + const insecureTls = useSharedInsecureDispatcher || useProfileInsecureTls; const dispatcher = useSharedInsecureDispatcher ? insecureDispatcher : useProfileInsecureTls ? ephemeralInsecureDispatcher - : undefined; + : getDefaultUpstreamDispatcher(); try { logger.stage('dispatch', 'upstream.dispatch', 'Dispatching upstream fetch', { profileName: profile.profileName, routedProfileName: upstream.route.profile.profileName, - insecureTls: dispatcher !== undefined, + insecureTls, passthrough, }); const upstreamUrl = resolveOpenAIChatCompletionsUrl(upstream.route.profile.baseUrl, { diff --git a/src/proxy/server/proxy-server.ts b/src/proxy/server/proxy-server.ts index 134583d4..f9af6885 100644 --- a/src/proxy/server/proxy-server.ts +++ b/src/proxy/server/proxy-server.ts @@ -5,6 +5,7 @@ import type { OpenAICompatProfileConfig } from '../profile-router'; import { OPENAI_COMPAT_PROXY_SERVICE_NAME } from '../proxy-daemon-paths'; import { createLogger, withRequestContext } from '../../services/logging'; import { + buildUpstreamAgentTimeouts, handleProxyMessagesRequest, handleProxyModelsRequest, validateIncomingProxyAuth, @@ -39,7 +40,7 @@ export function startOpenAICompatProxyServer(options: OpenAICompatProxyServerOpt port: options.port, }); const insecureDispatcher = options.insecure - ? new Agent({ connect: { rejectUnauthorized: false } }) + ? new Agent({ connect: { rejectUnauthorized: false }, ...buildUpstreamAgentTimeouts() }) : undefined; const server = http.createServer((req, res) => { const requestId = resolveInboundRequestId(req.headers); diff --git a/src/utils/fetch-proxy-setup.ts b/src/utils/fetch-proxy-setup.ts index b52c6d74..d71b4039 100644 --- a/src/utils/fetch-proxy-setup.ts +++ b/src/utils/fetch-proxy-setup.ts @@ -12,6 +12,8 @@ const FETCH_PROXY_PROTOCOLS = ['http:', 'https:']; type RoutingDispatchOptions = Parameters[0]; type RoutingDispatchHandler = Parameters[1]; +export type UpstreamAgentTimeoutOptions = Pick; + type GlobalFetchProxyConfig = { httpProxyUrl?: string; httpsProxyUrl?: string; @@ -19,14 +21,23 @@ type GlobalFetchProxyConfig = { }; class RoutingProxyDispatcher extends Dispatcher { - private readonly directDispatcher = new Agent(); + private readonly directDispatcher: Agent; private readonly httpProxyDispatcher: ProxyAgent | null; private readonly httpsProxyDispatcher: ProxyAgent | null; - constructor(httpProxyUrl: string | undefined, httpsProxyUrl: string | undefined) { + constructor( + httpProxyUrl: string | undefined, + httpsProxyUrl: string | undefined, + agentOptions: UpstreamAgentTimeoutOptions = {} + ) { super(); - this.httpProxyDispatcher = httpProxyUrl ? new ProxyAgent(httpProxyUrl) : null; - this.httpsProxyDispatcher = httpsProxyUrl ? new ProxyAgent(httpsProxyUrl) : null; + this.directDispatcher = new Agent(agentOptions); + this.httpProxyDispatcher = httpProxyUrl + ? new ProxyAgent({ uri: httpProxyUrl, ...agentOptions }) + : null; + this.httpsProxyDispatcher = httpsProxyUrl + ? new ProxyAgent({ uri: httpsProxyUrl, ...agentOptions }) + : null; } dispatch(options: RoutingDispatchOptions, handler: RoutingDispatchHandler): boolean { @@ -114,14 +125,16 @@ class RoutingProxyDispatcher extends Dispatcher { } } -export function createGlobalFetchProxyDispatcher(): Dispatcher | null { +export function createGlobalFetchProxyDispatcher( + agentOptions: UpstreamAgentTimeoutOptions = {} +): Dispatcher | null { const { httpProxyUrl, httpsProxyUrl } = resolveGlobalFetchProxyConfig(); if (!httpProxyUrl && !httpsProxyUrl) { return null; } - return new RoutingProxyDispatcher(httpProxyUrl, httpsProxyUrl); + return new RoutingProxyDispatcher(httpProxyUrl, httpsProxyUrl, agentOptions); } export function applyGlobalFetchProxy(): { enabled: boolean; error?: string } { diff --git a/tests/unit/proxy/messages-route.test.ts b/tests/unit/proxy/messages-route.test.ts index e398ea33..eb66ee5a 100644 --- a/tests/unit/proxy/messages-route.test.ts +++ b/tests/unit/proxy/messages-route.test.ts @@ -8,6 +8,7 @@ import { Agent } from 'undici'; import { resolveOpenAICompatProfileConfig } from '../../../src/proxy/profile-router'; import { attachDisconnectAbortHandlers, + buildUpstreamAgentTimeouts, handleProxyMessagesRequest, } from '../../../src/proxy/server/messages-route'; import { loadSettings } from '../../../src/utils/config-manager'; @@ -171,7 +172,55 @@ describe('attachDisconnectAbortHandlers', () => { }); }); +describe('buildUpstreamAgentTimeouts', () => { + afterEach(() => { + delete process.env.CCS_OPENAI_PROXY_REQUEST_TIMEOUT_MS; + }); + + it('keeps undici per-phase timeouts above the default request timeout', () => { + const timeouts = buildUpstreamAgentTimeouts(); + expect(timeouts.headersTimeout).toBeGreaterThan(600_000); + expect(timeouts.bodyTimeout).toBeGreaterThan(600_000); + }); + + it('tracks CCS_OPENAI_PROXY_REQUEST_TIMEOUT_MS overrides', () => { + process.env.CCS_OPENAI_PROXY_REQUEST_TIMEOUT_MS = '120000'; + const timeouts = buildUpstreamAgentTimeouts(); + expect(timeouts.headersTimeout).toBeGreaterThan(120_000); + expect(timeouts.bodyTimeout).toBeGreaterThan(120_000); + }); +}); + describe('handleProxyMessagesRequest', () => { + it('dispatches secure upstream fetches with an explicit dispatcher (not undici defaults)', async () => { + const activeProfile = buildProfile('hf'); + let capturedDispatcher: unknown; + + globalThis.fetch = (async (_input: RequestInfo | URL, init?: RequestInit) => { + capturedDispatcher = (init as RequestInit & { dispatcher?: unknown })?.dispatcher; + throw new Error('upstream exploded'); + }) as typeof globalThis.fetch; + + const req = new FakeRequest({ + 'x-api-key': 'local-token', + }); + const res = new FakeResponse(); + const pending = handleProxyMessagesRequest(req as never, res as never, activeProfile, 'local-token'); + req.end( + JSON.stringify({ + model: 'hf-default', + stream: true, + messages: [{ role: 'user', content: 'stay secure' }], + }) + ); + await pending; + + // A bare global-dispatcher fallback would reapply undici's 300s + // headersTimeout/bodyTimeout and undercut the proxy's request timeout. + expect(capturedDispatcher).toBeDefined(); + expect(res.statusCode).toBe(502); + }); + it('auto-passes through Kimi requests and preserves the coding-agent User-Agent', async () => { const activeProfile = buildProfile('kimic'); let capturedInput: RequestInfo | URL | undefined;