diff --git a/changelog.d/features/per-connection-upstream-timeout.md b/changelog.d/features/per-connection-upstream-timeout.md new file mode 100644 index 0000000000..a5987ed485 --- /dev/null +++ b/changelog.d/features/per-connection-upstream-timeout.md @@ -0,0 +1 @@ +- **feat(providers):** restore the operator-owned upstream timeout tier per connection via `providerSpecificData.timeoutMs` (preempts the maintainer-only model/provider registry tiers and the global `FETCH_TIMEOUT_MS`), and make the combo per-target timeout ceiling follow the selected connection \ No newline at end of file diff --git a/open-sse/handlers/chatCore.ts b/open-sse/handlers/chatCore.ts index beb842bf2d..9f530d4605 100644 --- a/open-sse/handlers/chatCore.ts +++ b/open-sse/handlers/chatCore.ts @@ -347,6 +347,7 @@ import { computeBillableTokens, normalizeExecutorResult, executeWithUpstreamStartTimeout, + resolveConnectionTimeoutMs, } from "./chatCore/upstreamTimeouts.ts"; import { getModelNormalizeToolCallId, getModelPreserveOpenAIDeveloperRole } from "@/lib/db/models"; import { getProviderCredentials, extractSessionAffinityKey } from "@/sse/services/auth"; @@ -3065,6 +3066,7 @@ export async function handleChatCore({ executor, provider, model: modelToCall, + connectionTimeoutMs: resolveConnectionTimeoutMs(execCreds?.providerSpecificData), signal: streamController.signal, log, execute: (signal) => @@ -3368,6 +3370,7 @@ export async function handleChatCore({ executor, provider, model: modelToCall, + connectionTimeoutMs: resolveConnectionTimeoutMs(execCreds?.providerSpecificData), signal: streamController.signal, log, execute: (signal) => diff --git a/open-sse/handlers/chatCore/upstreamTimeouts.ts b/open-sse/handlers/chatCore/upstreamTimeouts.ts index 9f0ace2b0a..b1a8da548d 100644 --- a/open-sse/handlers/chatCore/upstreamTimeouts.ts +++ b/open-sse/handlers/chatCore/upstreamTimeouts.ts @@ -9,6 +9,7 @@ import { getLoggedOutputTokens, getReasoningTokens, } from "@/lib/usage/tokenAccounting"; +import { MAX_PROVIDER_SPECIFIC_TIMEOUT_MS } from "@/shared/validation/providerSpecificData"; export function createBodyTimeoutError(timeoutMs: number): Error { const err = new Error(`Response body read timeout after ${timeoutMs}ms`); @@ -89,14 +90,44 @@ function resolveProviderTimeoutMs(executor: unknown): number { } } +/** Per-connection operator timeout tier: reads + * `providerSpecificData.timeoutMs`, bounded to 1..86_400_000 ms. + * Returns undefined when absent or invalid so the chain falls through. */ +export function resolveConnectionTimeoutMs(psd: unknown): number | undefined { + const timeoutMs = (psd as Record | null | undefined)?.timeoutMs; + if (typeof timeoutMs !== "number" || !Number.isFinite(timeoutMs)) return undefined; + const floored = Math.floor(timeoutMs); + if (floored < 1 || floored > MAX_PROVIDER_SPECIFIC_TIMEOUT_MS) return undefined; + return floored; +} + /** * Resolves the upstream header-response timeout in precedence order: + * connection-level override (`providerSpecificData.timeoutMs`) → * model-level override (registry `RegistryModel.timeoutMs`) → provider-level * override (`executor.getTimeoutMs()`) → global `FETCH_TIMEOUT_MS` default. * `provider`/`model` are optional so existing single-argument call sites * keep resolving to the provider/global chain unchanged (#6354). */ -export function getExecutorTimeoutMs(executor: unknown, provider?: string, model?: string): number { +export function getExecutorTimeoutMs( + executor: unknown, + provider?: string, + model?: string, + connectionTimeoutMs?: number +): number { + if ( + typeof connectionTimeoutMs === "number" && + Number.isFinite(connectionTimeoutMs) && + connectionTimeoutMs > 0 + ) { + // Defensive backstop for direct callers: resolveConnectionTimeoutMs is the + // gate (it rejects out-of-range values so the chain falls through); this + // clamp only caps values a future caller could pass unvetted. + return Math.min( + Math.max(0, Math.floor(connectionTimeoutMs)), + MAX_PROVIDER_SPECIFIC_TIMEOUT_MS + ); + } const modelOverride = resolveModelTimeoutOverride(provider, model); if (modelOverride !== undefined) return modelOverride; return resolveProviderTimeoutMs(executor); @@ -196,6 +227,7 @@ export async function executeWithUpstreamStartTimeout({ executor, provider, model, + connectionTimeoutMs, signal, log, execute, @@ -203,11 +235,12 @@ export async function executeWithUpstreamStartTimeout({ executor: unknown; provider: string; model: string; + connectionTimeoutMs?: number; signal: AbortSignal; log?: { warn?: (tag: string, message: string) => void } | null; execute: (signal: AbortSignal) => Promise; }): Promise { - const timeoutMs = getExecutorTimeoutMs(executor, provider, model); + const timeoutMs = getExecutorTimeoutMs(executor, provider, model, connectionTimeoutMs); if (timeoutMs <= 0) return execute(signal); if (signal.aborted) throw createAbortError(signal); diff --git a/open-sse/services/combo.ts b/open-sse/services/combo.ts index 6a1d7cbbf2..2690cdddb1 100644 --- a/open-sse/services/combo.ts +++ b/open-sse/services/combo.ts @@ -46,6 +46,7 @@ import { getDefaultComboConfig, resolveComboQueueDepth, isComboCooldownWaitEligible, + resolveComboTargetTimeoutMsForCombo, } from "./comboConfig.ts"; import { maybeGenerateHandoff, @@ -89,6 +90,7 @@ import { selectQuotaShareTarget } from "./combo/quotaShareStrategy.ts"; import { makeConnectionConcurrencyResolver, lookupPositiveCap } from "./combo/concurrencyCaps.ts"; import { acquireQuotaShareConcurrencySlot } from "./combo/quotaShareConcurrency.ts"; import { canAffordRequest } from "../../src/lib/quota/quotaScheduler.ts"; +import { resolveConnectionTimeoutMs } from "../handlers/chatCore/upstreamTimeouts.ts"; import { getCachedProviderConnectionById } from "../../src/lib/db/readCache.ts"; import { orderTargetsByEvalScores } from "./evalRouting.ts"; @@ -133,6 +135,7 @@ import { isProviderInCooldown, recordProviderCooldown } from "./providerCooldown import { resolveResilienceSettings, type ResilienceSettings, + type ComboCooldownWaitSettings, } from "../../src/lib/resilience/settings"; import { resolveReasoningBufferedMaxTokens, toPositiveInteger } from "./reasoningTokenBuffer.ts"; import { RESET_WINDOW_NAMES } from "./combo/types.ts"; @@ -141,6 +144,7 @@ import type { ComboRetryAfter, ComboErrorBody, SingleModelTarget, + ComboLogger, HandleComboChatOptions, HandleRoundRobinOptions, ResolvedComboTarget, @@ -621,6 +625,40 @@ export { pinIsDurablyUnhealthy }; /** @param {string} errorText */ /** @param {object} options */ +/** + * Resolves the per-target timeout ceiling for a combo target: when the target's + * connection carries `providerSpecificData.timeoutMs`, re-runs + * resolveComboTargetTimeoutMsForCombo with that timeout as the ceiling so the + * combo's per-target timer follows the selected connection. + * Returns undefined when the connection or its timeout is absent — the runner + * then falls back to the setup-time comboTargetTimeoutMs. + */ +export async function resolveTargetTimeoutMsForTarget( + config: Record | null | undefined, + strategy: string, + comboCooldownWait: Pick, + target?: SingleModelTarget, + log?: Pick | null +): Promise { + const connectionId = target && "connectionId" in target ? target.connectionId : null; + if (!connectionId) return undefined; + try { + const connection = await getCachedProviderConnectionById(connectionId); + if (!connection) return undefined; + const timeoutMs = resolveConnectionTimeoutMs(connection.providerSpecificData); + if (timeoutMs === undefined) return undefined; + return resolveComboTargetTimeoutMsForCombo(config, timeoutMs, strategy, comboCooldownWait); + } catch (err) { + log?.debug?.( + "COMBO", + `resolveTargetTimeoutMsForTarget connection lookup failed: ${ + err instanceof Error ? err.message : String(err) + }` + ); + return undefined; + } +} + /** * #10681 egress: every combo response carries the opaque trace id in an * `X-OmniRoute-Combo-Trace` header so a post-incident lookup of the ordered @@ -683,6 +721,14 @@ async function handleComboChatInner({ const handleSingleModelWithTimeout = buildTargetTimeoutRunner({ handleSingleModel, comboTargetTimeoutMs, + resolveTargetTimeoutMs: (target) => + resolveTargetTimeoutMsForTarget( + config, + strategy, + resilienceSettings.comboCooldownWait, + target, + log + ), log, }); diff --git a/open-sse/services/combo/targetTimeoutRunner.ts b/open-sse/services/combo/targetTimeoutRunner.ts index fb5c2262f0..402d093f35 100644 --- a/open-sse/services/combo/targetTimeoutRunner.ts +++ b/open-sse/services/combo/targetTimeoutRunner.ts @@ -93,25 +93,33 @@ export function buildTargetTimeoutRunner(deps: { handleSingleModel: HandleSingleModel; comboTargetTimeoutMs: number; log: ComboLogger; + resolveTargetTimeoutMs?: ( + target?: SingleModelTarget + ) => Promise | number | undefined; }): ( b: Record, modelStr: string, target?: SingleModelTarget ) => Promise { - const { handleSingleModel, comboTargetTimeoutMs, log } = deps; + const { handleSingleModel, comboTargetTimeoutMs, log, resolveTargetTimeoutMs } = deps; ensureDiagnosticListener(); return async ( b: Record, modelStr: string, target?: SingleModelTarget ): Promise => { - if (comboTargetTimeoutMs <= 0) { + const resolvedTimeoutMs = await resolveTargetTimeoutMs?.(target); + const effectiveTimeoutMs = + typeof resolvedTimeoutMs === "number" && Number.isFinite(resolvedTimeoutMs) + ? resolvedTimeoutMs + : comboTargetTimeoutMs; + if (effectiveTimeoutMs <= 0) { // G3 (silent-stop fix): a disabled per-model timeout means a hung upstream // stalls the target until the combo loop safety timer (COMBO_LOOP_SAFETY_TIMEOUT_MS) // force-terminates — surface that dependency instead of silently running bare. log.warn( "COMBO", - `Per-model combo timeout is DISABLED (comboTargetTimeoutMs=${comboTargetTimeoutMs}) for ${modelStr} — a hung upstream will hang this target until the combo loop safety timeout` + `Per-model combo timeout is DISABLED (effectiveTimeoutMs=${effectiveTimeoutMs}) for ${modelStr} — a hung upstream will hang this target until the combo loop safety timeout` ); return handleSingleModel(b, modelStr, target).catch((err) => errorResponse(502, err?.message ?? "Upstream model error") @@ -127,13 +135,13 @@ export function buildTargetTimeoutRunner(deps: { const abortErr = new Error(COMBO_PER_MODEL_TIMEOUT_REASON); recordTimeoutContext({ modelStr, - timeoutMs: comboTargetTimeoutMs, + timeoutMs: effectiveTimeoutMs, abortError: abortErr, timestamp: Date.now(), }); log.warn( "COMBO", - `Model ${modelStr} exceeded ${comboTargetTimeoutMs}ms timeout — falling back` + `Model ${modelStr} exceeded ${effectiveTimeoutMs}ms timeout — falling back` ); timeoutController.abort(abortErr); // HTTP 504 (not proprietary 524): this is OmniRoute's own per-target timer. @@ -154,7 +162,7 @@ export function buildTargetTimeoutRunner(deps: { } ) ); - }, comboTargetTimeoutMs); + }, effectiveTimeoutMs); }); const targetWithSignal = { ...(target ?? {}), diff --git a/src/lib/providers/requestDefaults.ts b/src/lib/providers/requestDefaults.ts index 05786ab7aa..b54bf3669d 100644 --- a/src/lib/providers/requestDefaults.ts +++ b/src/lib/providers/requestDefaults.ts @@ -218,6 +218,14 @@ export function normalizeProviderSpecificData( delete normalized.autoFetchModels; } + // Per-connection operator timeout — only persist a real integer. + if ( + "timeoutMs" in normalized && + (typeof normalized.timeoutMs !== "number" || !Number.isInteger(normalized.timeoutMs)) + ) { + delete normalized.timeoutMs; + } + if ("preset" in normalized) { const preset = provider === "openrouter" ? normalizeOpenRouterPreset(normalized.preset) : null; if (preset) { diff --git a/src/shared/validation/providerSpecificData.ts b/src/shared/validation/providerSpecificData.ts index 80168e51cf..c75eef108b 100644 --- a/src/shared/validation/providerSpecificData.ts +++ b/src/shared/validation/providerSpecificData.ts @@ -17,6 +17,7 @@ const CODEX_REASONING_EFFORT_VALUES = new Set(["none", "low", "medium", "high", const REQUEST_DEFAULT_SERVICE_TIER_VALUES = new Set(["default", "priority", "fast", "flex"]); const CODEX_FINGERPRINT_MODE_VALUES = new Set(["off", "device", "session", "full"]); const CACHE_PASSTHROUGH_VALUES = new Set(["strip", "openai-format", "claude-format"]); +export const MAX_PROVIDER_SPECIFIC_TIMEOUT_MS = 86_400_000; // 24h — operator cap, anti-DoS // #6880 — per-connection prompt-cache capability override, extracted so // validateProviderSpecificData() stays under the complexity gate. @@ -496,4 +497,23 @@ export function validateProviderSpecificData( }); } } + + // Per-connection operator timeout tier: a slow model must not monopolize an + // executor slot indefinitely. Bounded to 24h (anti-DoS); below 1ms is + // meaningless. + const timeoutMs = data.timeoutMs; + if (timeoutMs !== undefined && timeoutMs !== null) { + if ( + typeof timeoutMs !== "number" || + !Number.isInteger(timeoutMs) || + timeoutMs < 1 || + timeoutMs > MAX_PROVIDER_SPECIFIC_TIMEOUT_MS + ) { + ctx.addIssue({ + code: z.ZodIssueCode.custom, + message: `providerSpecificData.timeoutMs must be an integer between 1 and ${MAX_PROVIDER_SPECIFIC_TIMEOUT_MS}`, + path: ["timeoutMs"], + }); + } + } } diff --git a/tests/unit/combo-target-timeout-runner.test.ts b/tests/unit/combo-target-timeout-runner.test.ts index a058a94552..e56f54b023 100644 --- a/tests/unit/combo-target-timeout-runner.test.ts +++ b/tests/unit/combo-target-timeout-runner.test.ts @@ -204,3 +204,75 @@ test("drainLastTimeoutContexts returns and clears recorded contexts", async () = const second = drainLastTimeoutContexts(); assert.equal(second.length, 0, "second drain must return empty"); }); + +test("resolveTargetTimeoutMs provided: uses per-target timeout when present", async () => { + let aborted = false; + const runner = buildTargetTimeoutRunner({ + handleSingleModel: (_b, _m, target) => + new Promise((resolve) => { + const sig = target?.modelAbortSignal ?? undefined; + sig?.addEventListener("abort", () => { + aborted = true; + resolve(new Response(null, { status: 599 })); + }); + }), + comboTargetTimeoutMs: 20, + resolveTargetTimeoutMs: async (target) => + target?.connectionId === "conn-1" ? 50 : undefined, + log: noopLog, + }); + const res = await runner({}, "slow-model", { + connectionId: "conn-1", + modelAbortSignal: undefined as unknown as AbortSignal, + } as SingleModelTarget); + assert.equal(res.status, 504); + assert.equal(aborted, true); + const body = await res.json(); + assert.equal(body?.error?.code, "combo_target_timeout"); +}); + +test("resolveTargetTimeoutMs without connection: falls back to comboTargetTimeoutMs", async () => { + let aborted = false; + const runner = buildTargetTimeoutRunner({ + handleSingleModel: (_b, _m, target) => + new Promise((resolve) => { + target?.modelAbortSignal?.addEventListener("abort", () => { + aborted = true; + resolve(new Response(null, { status: 599 })); + }); + }), + comboTargetTimeoutMs: 20, + resolveTargetTimeoutMs: async () => undefined, + log: noopLog, + }); + const res = await runner({}, "slow-model", { + connectionId: "conn-unknown", + modelAbortSignal: undefined as unknown as AbortSignal, + } as SingleModelTarget); + assert.equal(res.status, 504); + assert.equal(aborted, true); +}); + +test("resolveTargetTimeoutMs extended: 50ms outlives the 20ms base (does not abort at 20ms)", async () => { + let resolvedWith = ""; + const runner = buildTargetTimeoutRunner({ + handleSingleModel: async (_b, _m, target) => { + await new Promise((resolve) => { + target?.modelAbortSignal?.addEventListener("abort", resolve); + setTimeout(resolve, 45); + }); + resolvedWith = target?.modelAbortSignal?.aborted ? "aborted" : "completed"; + return new Response(resolvedWith, { status: resolvedWith === "aborted" ? 599 : 200 }); + }, + comboTargetTimeoutMs: 20, + resolveTargetTimeoutMs: async (target) => + target?.connectionId === "conn-1" ? 50 : undefined, + log: noopLog, + }); + const res = await runner({}, "slow-model", { + connectionId: "conn-1", + modelAbortSignal: undefined as unknown as AbortSignal, + } as SingleModelTarget); + assert.equal(res.status, 200); + assert.equal(resolvedWith, "completed"); +}); diff --git a/tests/unit/provider-specific-data-schema.test.ts b/tests/unit/provider-specific-data-schema.test.ts index 32caa5141c..32568e0271 100644 --- a/tests/unit/provider-specific-data-schema.test.ts +++ b/tests/unit/provider-specific-data-schema.test.ts @@ -390,3 +390,26 @@ test("provider schemas reject invalid quotaPerUnit values", () => { assert.equal(negative.success, false); assert.equal(string.success, false); }); + +test("provider schemas accept integer timeoutMs in providerSpecificData", () => { + const created = createProviderSchema.safeParse({ + provider: "openai", + apiKey: "token", + name: "OpenAI", + providerSpecificData: { timeoutMs: 1_800_000 }, + }); + const updated = updateProviderConnectionSchema.safeParse({ + providerSpecificData: { timeoutMs: 900_000 }, + }); + assert.equal(created.success, true); + assert.equal(updated.success, true); +}); + +test("provider schemas reject invalid timeoutMs values", () => { + for (const bad of [-1, 0, 1.5, "60000", 86_400_001]) { + const updated = updateProviderConnectionSchema.safeParse({ + providerSpecificData: { timeoutMs: bad }, + }); + assert.equal(updated.success, false, `timeoutMs=${String(bad)} must be rejected`); + } +}); diff --git a/tests/unit/upstream-timeout-connection-tier.test.ts b/tests/unit/upstream-timeout-connection-tier.test.ts new file mode 100644 index 0000000000..30c4db7e53 --- /dev/null +++ b/tests/unit/upstream-timeout-connection-tier.test.ts @@ -0,0 +1,133 @@ +import { test } from "node:test"; +import assert from "node:assert/strict"; +import { + getExecutorTimeoutMs, + resolveConnectionTimeoutMs, + executeWithUpstreamStartTimeout, +} from "../../open-sse/handlers/chatCore/upstreamTimeouts.ts"; +import { resolveTargetTimeoutMsForTarget } from "../../open-sse/services/combo.ts"; + +const GLOBAL = 600_000; + +function fakeExecutor(timeoutMs?: number) { + return { getTimeoutMs: () => timeoutMs ?? GLOBAL }; +} + +test("resolveConnectionTimeoutMs: returns floored integer for valid values", () => { + assert.equal(resolveConnectionTimeoutMs({ timeoutMs: 1_800_000.9 }), 1_800_000); + assert.equal(resolveConnectionTimeoutMs({ timeoutMs: 1 }), 1); +}); + +test("resolveConnectionTimeoutMs: undefined for absent/invalid values", () => { + assert.equal(resolveConnectionTimeoutMs(undefined), undefined); + assert.equal(resolveConnectionTimeoutMs(null), undefined); + assert.equal(resolveConnectionTimeoutMs({}), undefined); + assert.equal(resolveConnectionTimeoutMs({ timeoutMs: 0 }), undefined); + assert.equal(resolveConnectionTimeoutMs({ timeoutMs: -5 }), undefined); + assert.equal(resolveConnectionTimeoutMs({ timeoutMs: "60000" }), undefined); + assert.equal(resolveConnectionTimeoutMs({ timeoutMs: 86_400_001 }), undefined); +}); + +test("getExecutorTimeoutMs: connection tier preempts model, provider and global", () => { + const executor = fakeExecutor(30_000); + // model override present (registry) + provider override + connection: + // connection wins + assert.equal(getExecutorTimeoutMs(executor, "openai", "gpt-5", 1_800_000), 1_800_000); +}); + +test("getExecutorTimeoutMs: invalid connection timeout falls through to model/provider/global", () => { + const executor = fakeExecutor(30_000); + assert.equal(getExecutorTimeoutMs(executor, "openai", "gpt-5", 0), 30_000); + assert.equal(getExecutorTimeoutMs(executor, "openai", "gpt-5", undefined), 30_000); + assert.equal(getExecutorTimeoutMs(fakeExecutor(), undefined, undefined, undefined), GLOBAL); +}); + +test("executeWithUpstreamStartTimeout: connection timeout aborts before the global", async () => { + let signalAbortedAt = Number.POSITIVE_INFINITY; + const started = Date.now(); + const result = await executeWithUpstreamStartTimeout({ + executor: fakeExecutor(600_000), + provider: "openai", + model: "gpt-5", + connectionTimeoutMs: 50, + signal: new AbortController().signal, + log: null, + execute: (signal) => + new Promise((resolve) => { + signal.addEventListener("abort", () => { + signalAbortedAt = Date.now() - started; + resolve("aborted"); + }); + }), + }); + assert.equal(result, "aborted"); + assert.ok(signalAbortedAt < 1_000, `aborted at ${signalAbortedAt}ms, expected < 1000ms`); +}); +test("connection timeout extraction matches execCreds.providerSpecificData shape", () => { + const execCreds = { providerSpecificData: { timeoutMs: 1_800_000, someOtherKey: 1 } }; + assert.equal(resolveConnectionTimeoutMs(execCreds?.providerSpecificData), 1_800_000); + assert.equal( + resolveConnectionTimeoutMs({ providerSpecificData: undefined }?.providerSpecificData), + undefined + ); +}); + +test("resolveTargetTimeoutMsForTarget: undefined without connectionId", async () => { + const result = await resolveTargetTimeoutMsForTarget( + null, + "fallback", + { enabled: false, budgetMs: 0 }, + { + kind: "model", + stepId: "s", + executionKey: "k", + modelStr: "m", + provider: "p", + providerId: null, + connectionId: null, + weight: 1, + label: null, + } + ); + assert.equal(result, undefined); +}); + +test("resolveTargetTimeoutMsForTarget: unknown connection -> undefined", async () => { + const result = await resolveTargetTimeoutMsForTarget( + null, + "fallback", + { enabled: false, budgetMs: 0 }, + { + kind: "model", + stepId: "s", + executionKey: "k", + modelStr: "m", + provider: "p", + providerId: null, + connectionId: "conn-does-not-exist", + weight: 1, + label: null, + } + ); + assert.equal(result, undefined); +}); + +test("resolveTargetTimeoutMsForTarget: connection without timeoutMs -> undefined", async () => { + const result = await resolveTargetTimeoutMsForTarget( + null, + "fallback", + { enabled: false, budgetMs: 0 }, + { + kind: "model", + stepId: "s", + executionKey: "k", + modelStr: "m", + provider: "p", + providerId: null, + connectionId: "conn-no-timeout", + weight: 1, + label: null, + } + ); + assert.equal(result, undefined); +});