diff --git a/open-sse/config/constants.ts b/open-sse/config/constants.ts index 462dcd481c..6faf2df056 100644 --- a/open-sse/config/constants.ts +++ b/open-sse/config/constants.ts @@ -17,6 +17,11 @@ export const FETCH_TIMEOUT_MS = upstreamTimeouts.fetchTimeoutMs; // idle for this duration. Override with STREAM_IDLE_TIMEOUT_MS env var. export const STREAM_IDLE_TIMEOUT_MS = upstreamTimeouts.streamIdleTimeoutMs; +// Timeout for the first useful SSE event. Keep this much shorter than the +// post-start idle timeout so slow-thinking models can keep streaming after the +// first token, while dead 200 OK streams fail fast enough for combo fallback. +export const STREAM_READINESS_TIMEOUT_MS = upstreamTimeouts.streamReadinessTimeoutMs; + // Timeout for reading the full response body after headers arrive (ms). // Prevents indefinite hangs when the upstream sends headers but stalls on the body. // Defaults to FETCH_TIMEOUT_MS. Override with FETCH_BODY_TIMEOUT_MS env var. diff --git a/open-sse/handlers/chatCore.ts b/open-sse/handlers/chatCore.ts index bc5310a4ea..0a2c7b9dad 100644 --- a/open-sse/handlers/chatCore.ts +++ b/open-sse/handlers/chatCore.ts @@ -10,6 +10,7 @@ import { } from "../utils/stream.ts"; import { ensureStreamReadiness } from "../utils/streamReadiness.ts"; import { createStreamController, pipeWithDisconnect } from "../utils/streamHandler.ts"; +import { createSseHeartbeatTransform } from "../utils/sseHeartbeat.ts"; import { addBufferToUsage, filterUsageForFormat, estimateUsage } from "../utils/usageTracking.ts"; import { refreshWithRetry } from "../services/tokenRefresh.ts"; import { createRequestLogger } from "../utils/requestLogger.ts"; @@ -32,6 +33,7 @@ import { HTTP_STATUS, PROVIDER_MAX_TOKENS, STREAM_IDLE_TIMEOUT_MS, + STREAM_READINESS_TIMEOUT_MS, } from "../config/constants.ts"; import { classifyProviderError, @@ -3941,7 +3943,7 @@ export async function handleChatCore({ // Streaming response const streamReadiness = await ensureStreamReadiness(providerResponse, { - timeoutMs: STREAM_IDLE_TIMEOUT_MS, + timeoutMs: STREAM_READINESS_TIMEOUT_MS, provider, model, log, @@ -3990,8 +3992,9 @@ export async function handleChatCore({ const responseHeaders = { "Content-Type": "text/event-stream", - "Cache-Control": "no-cache", + "Cache-Control": "no-cache, no-transform", Connection: "keep-alive", + "X-Accel-Buffering": "no", [OMNIROUTE_RESPONSE_HEADERS.cache]: "MISS", ...buildOmniRouteResponseMetaHeaders({ provider, @@ -4236,6 +4239,9 @@ export async function handleChatCore({ } else { finalStream = pipeWithDisconnect(providerResponse, transformStream, streamController); } + finalStream = finalStream.pipeThrough( + createSseHeartbeatTransform({ signal: streamController.signal }) + ); return { success: true, diff --git a/open-sse/utils/sseHeartbeat.ts b/open-sse/utils/sseHeartbeat.ts index 1ceb6820b3..db8d0ff624 100644 --- a/open-sse/utils/sseHeartbeat.ts +++ b/open-sse/utils/sseHeartbeat.ts @@ -1,4 +1,4 @@ -const DEFAULT_INTERVAL_MS = 15_000; +export const DEFAULT_SSE_HEARTBEAT_INTERVAL_MS = 15_000; type SseHeartbeatTransformOptions = { intervalMs?: number; @@ -6,7 +6,7 @@ type SseHeartbeatTransformOptions = { }; export function createSseHeartbeatTransform({ - intervalMs = DEFAULT_INTERVAL_MS, + intervalMs = DEFAULT_SSE_HEARTBEAT_INTERVAL_MS, signal, }: SseHeartbeatTransformOptions = {}) { let intervalId: ReturnType | undefined; diff --git a/open-sse/utils/streamHandler.ts b/open-sse/utils/streamHandler.ts index 89d695c6fb..2bfbcef68e 100644 --- a/open-sse/utils/streamHandler.ts +++ b/open-sse/utils/streamHandler.ts @@ -3,6 +3,7 @@ import { trackPendingRequest } from "@/lib/usageDb"; // Stream handler with disconnect detection - shared for all providers const PENDING_REQUEST_CLEARED_MARKER = "__omniroutePendingRequestCleared"; +const DISCONNECT_ABORT_DELAY_MS = 2_000; type StreamDisconnectEvent = { reason: string; @@ -97,7 +98,7 @@ export function createStreamController({ // Delay abort to allow cleanup abortTimeout = setTimeout(() => { abortController.abort(); - }, 500); + }, DISCONNECT_ABORT_DELAY_MS); onDisconnect?.({ reason, duration: Date.now() - startTime }); }, @@ -201,7 +202,9 @@ export function createDisconnectAwareStream(transformStream, streamController) { cancel(reason) { streamController.handleDisconnect(reason || "cancelled"); reader.cancel(); - writer.abort(); + setTimeout(() => { + writer.abort(); + }, DISCONNECT_ABORT_DELAY_MS).unref?.(); }, }); } diff --git a/src/shared/utils/runtimeTimeouts.ts b/src/shared/utils/runtimeTimeouts.ts index b2794a7745..672466a1c2 100644 --- a/src/shared/utils/runtimeTimeouts.ts +++ b/src/shared/utils/runtimeTimeouts.ts @@ -8,6 +8,7 @@ type ReadTimeoutOptions = { export const DEFAULT_FETCH_TIMEOUT_MS = 600_000; export const DEFAULT_STREAM_IDLE_TIMEOUT_MS = 600_000; +export const DEFAULT_STREAM_READINESS_TIMEOUT_MS = 30_000; export const DEFAULT_FETCH_CONNECT_TIMEOUT_MS = 30_000; export const DEFAULT_FETCH_KEEPALIVE_TIMEOUT_MS = 4_000; export const DEFAULT_API_BRIDGE_PROXY_TIMEOUT_MS = 30_000; @@ -24,6 +25,7 @@ function hasEnvValue(env: EnvSource, name: string): boolean { export type UpstreamTimeoutConfig = { fetchTimeoutMs: number; streamIdleTimeoutMs: number; + streamReadinessTimeoutMs: number; fetchHeadersTimeoutMs: number; fetchBodyTimeoutMs: number; fetchConnectTimeoutMs: number; @@ -89,10 +91,20 @@ export function getUpstreamTimeoutConfig( logger, } ); + const streamReadinessTimeoutMs = readTimeoutMs( + env, + "STREAM_READINESS_TIMEOUT_MS", + DEFAULT_STREAM_READINESS_TIMEOUT_MS, + { + allowZero: true, + logger, + } + ); return { fetchTimeoutMs, streamIdleTimeoutMs, + streamReadinessTimeoutMs, fetchHeadersTimeoutMs: readTimeoutMs(env, "FETCH_HEADERS_TIMEOUT_MS", fetchTimeoutMs, { allowZero: true, logger,