From 5e234d503d2579a5466a137d1fee7e0575e1fe45 Mon Sep 17 00:00:00 2001 From: Diego Rodrigues de Sa e Souza <8016841+diegosouzapw@users.noreply.github.com> Date: Tue, 21 Jul 2026 21:41:58 -0300 Subject: [PATCH] fix(sse): bound Codex SSE peek read with per-read timeout (#8020) (#8043) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit peekCodexSseTransientError() ran before chatCore's normal readiness/idle-timeout pipeline and read the first SSE chunk with a bare reader.read() — no timeout wrapper. A 200 text/event-stream body that never emitted a byte hung for ~15min (901399ms observed) before the platform killed the connection and surfaced a generic 502. Wrap the peek loop's read and the re-assembled passthrough body's pull() in readStreamChunkWithTimeout, bounded PER READ (not a total deadline) so a long-but-alive reasoning stream keeps resetting the window on every chunk it emits. On timeout the reader is cancelled and the request now fails fast with a 504 instead of hanging. New small module open-sse/executors/codex/bodyTimeout.ts holds the wrapping helpers to keep codex.ts within its frozen size baseline. --- .../fixes/8020-codex-peek-read-timeout.md | 1 + open-sse/executors/codex.ts | 55 ++++++------ open-sse/executors/codex/bodyTimeout.ts | 88 +++++++++++++++++++ .../bug-8020-codex-peek-body-timeout.test.ts | 71 +++++++++++++++ 4 files changed, 190 insertions(+), 25 deletions(-) create mode 100644 changelog.d/fixes/8020-codex-peek-read-timeout.md create mode 100644 open-sse/executors/codex/bodyTimeout.ts create mode 100644 tests/unit/bug-8020-codex-peek-body-timeout.test.ts diff --git a/changelog.d/fixes/8020-codex-peek-read-timeout.md b/changelog.d/fixes/8020-codex-peek-read-timeout.md new file mode 100644 index 0000000000..a9b28e327c --- /dev/null +++ b/changelog.d/fixes/8020-codex-peek-read-timeout.md @@ -0,0 +1 @@ +- fix(sse): bound the Codex SSE peek/passthrough body reads with a per-read timeout so a silently stalled upstream body settles in FETCH_BODY_TIMEOUT_MS instead of hanging ~15min and returning a generic 502 (#8020) diff --git a/open-sse/executors/codex.ts b/open-sse/executors/codex.ts index 74daed8064..463a815ca0 100644 --- a/open-sse/executors/codex.ts +++ b/open-sse/executors/codex.ts @@ -17,7 +17,8 @@ import { CODEX_CHAT_DEFAULT_INSTRUCTIONS, CODEX_DEFAULT_INSTRUCTIONS, } from "../config/codexInstructions.ts"; -import { HTTP_STATUS, PROVIDERS } from "../config/constants.ts"; +import { FETCH_BODY_TIMEOUT_MS, HTTP_STATUS, PROVIDERS } from "../config/constants.ts"; +import { readCodexPeekChunk, buildCodexTimeoutSafePassthroughBody } from "./codex/bodyTimeout.ts"; import { getCodexClientVersion, getCodexUserAgent, @@ -675,15 +676,23 @@ function extractCodexSseErrorMessage(text: string, fallback: string): string { } type CodexSseTransientErrorPeek = - | { matched: string; message: string; replacementBody: null } - | { matched: null; message: null; replacementBody: ReadableStream | null }; + | { matched: string; message: string; replacementBody: null; timedOut?: false } + | { + matched: null; + message: null; + replacementBody: ReadableStream | null; + timedOut?: boolean; + }; /** * Peek the first bytes of a Codex SSE response body looking for a transient * error embedded in an otherwise 200-OK stream. Exported for unit testing. + * `timeoutMs` bounds EACH individual read (#8020) — defaults to + * FETCH_BODY_TIMEOUT_MS; overridable so tests can settle fast/deterministically. */ export async function peekCodexSseTransientError( - response: Response + response: Response, + timeoutMs: number = FETCH_BODY_TIMEOUT_MS ): Promise { const contentType = response.headers.get("content-type") || ""; // #7536: check content-type BEFORE touching `response.body`. On the wreq-js @@ -706,8 +715,12 @@ export async function peekCodexSseTransientError( try { while (text.length < CODEX_SSE_PEEK_MAX_BYTES) { - const { done, value } = await reader.read(); + const { done, value, timedOut } = await readCodexPeekChunk(reader, timeoutMs); + if (timedOut) { + return { matched: null, message: null, replacementBody: null, timedOut: true }; + } if (done) break; + if (!value) continue; chunks.push(value); text += decoder.decode(value, { stream: true }); const lower = text.toLowerCase(); @@ -749,26 +762,7 @@ export async function peekCodexSseTransientError( // undici (every non-stream Codex request 502'd, then got mis-classified as a // 60s rate limit). Keep the original reader; never touch response.body again. const upstreamReader = reader; - const replacementBody = new ReadableStream({ - start(controller) { - for (const chunk of chunks) controller.enqueue(chunk); - }, - async pull(controller) { - const { done, value } = await upstreamReader.read(); - if (done) { - controller.close(); - return; - } - controller.enqueue(value); - }, - cancel(reason) { - try { - upstreamReader.cancel(reason).catch(() => {}); - } catch { - // noop — upstream socket may already be closing. - } - }, - }); + const replacementBody = buildCodexTimeoutSafePassthroughBody(chunks, upstreamReader, timeoutMs); return { matched: null, message: null, replacementBody }; } @@ -905,6 +899,17 @@ export class CodexExecutor extends BaseExecutor { HTTP_STATUS.SERVICE_UNAVAILABLE, peek.message ); + } else if (peek.timedOut) { + // #8020: the peek's first-chunk read never returned (upstream body went + // silent). Convert to a bounded 504 instead of letting the caller hang. + input.log?.warn?.( + "TIMEOUT", + "CODEX | 200-OK SSE peek read timed out — upstream body stalled, returning 504" + ); + (httpResult as { response: Response }).response = errorResponse( + HTTP_STATUS.GATEWAY_TIMEOUT, + "Upstream Codex SSE body read timed out" + ); } else if (peek.replacementBody) { (httpResult as { response: Response }).response = new Response(peek.replacementBody, { status: resp.status, diff --git a/open-sse/executors/codex/bodyTimeout.ts b/open-sse/executors/codex/bodyTimeout.ts new file mode 100644 index 0000000000..a07c4b150c --- /dev/null +++ b/open-sse/executors/codex/bodyTimeout.ts @@ -0,0 +1,88 @@ +/** + * Per-read timeout helpers for the Codex SSE peek/passthrough body reads (#8020). + * + * `peekCodexSseTransientError()` in ../codex.ts reads the first bytes of a Codex + * SSE response body BEFORE the response reaches chatCore's normal readiness/idle + * pipeline, so a 200 text/event-stream whose body never emits a byte bypassed + * FETCH_BODY_TIMEOUT_MS / STREAM_IDLE_TIMEOUT_MS entirely and hung on a bare + * `reader.read()` for ~15 minutes before the platform killed the connection as a + * generic 502. These helpers wrap every read (the peek loop AND the re-assembled + * passthrough body's pull()) in `readStreamChunkWithTimeout`, PER READ rather than + * against a single total-request deadline, so a long-but-alive reasoning stream + * that keeps emitting chunks never trips the timeout — only a stream that goes + * silent for `timeoutMs` does. + */ +import { readStreamChunkWithTimeout } from "../../handlers/chatCore/upstreamTimeouts.ts"; + +function isBodyTimeoutError(err: unknown): boolean { + return err instanceof Error && err.name === "BodyTimeoutError"; +} + +async function cancelReaderSafely(reader: ReadableStreamDefaultReader): Promise { + try { + await reader.cancel(); + } catch { + // Upstream socket may already be closing; nothing to clean up. + } +} + +/** + * Reads the next peek-loop chunk under `timeoutMs`. On a `BodyTimeoutError` the + * upstream reader is cancelled (releasing the socket) and `timedOut: true` is + * returned instead of throwing, so the caller can short-circuit straight to a + * bounded error response rather than falling through to an unbounded passthrough. + */ +export async function readCodexPeekChunk( + reader: ReadableStreamDefaultReader, + timeoutMs: number +): Promise<{ done: boolean; value?: Uint8Array; timedOut: boolean }> { + try { + const { done, value } = await readStreamChunkWithTimeout(reader, timeoutMs); + return { done, value, timedOut: false }; + } catch (err) { + if (isBodyTimeoutError(err)) { + await cancelReaderSafely(reader); + return { done: true, timedOut: true }; + } + throw err; + } +} + +/** + * Builds the re-assembled Codex SSE body (peeked prefix chunks + continued drain + * of the same reader), with every subsequent read bounded by `timeoutMs`. A + * timeout on the passthrough cancels the upstream reader and errors the stream + * controller instead of hanging the client connection forever. + */ +export function buildCodexTimeoutSafePassthroughBody( + chunks: Uint8Array[], + upstreamReader: ReadableStreamDefaultReader, + timeoutMs: number +): ReadableStream { + return new ReadableStream({ + start(controller) { + for (const chunk of chunks) controller.enqueue(chunk); + }, + async pull(controller) { + try { + const { done, value } = await readStreamChunkWithTimeout(upstreamReader, timeoutMs); + if (done) { + controller.close(); + return; + } + if (!value) return; + controller.enqueue(value); + } catch (err) { + await cancelReaderSafely(upstreamReader); + controller.error(err); + } + }, + cancel(reason) { + try { + upstreamReader.cancel(reason).catch(() => {}); + } catch { + // noop — upstream socket may already be closing. + } + }, + }); +} diff --git a/tests/unit/bug-8020-codex-peek-body-timeout.test.ts b/tests/unit/bug-8020-codex-peek-body-timeout.test.ts new file mode 100644 index 0000000000..b74398cd2f --- /dev/null +++ b/tests/unit/bug-8020-codex-peek-body-timeout.test.ts @@ -0,0 +1,71 @@ +// #8020: `peekCodexSseTransientError()`'s first-chunk read (open-sse/executors/codex.ts) had +// NO timeout wrapper — a 200 text/event-stream response whose body never emits a byte hung on a +// bare `reader.read()` for ~15 minutes (901399ms observed) before the platform killed the +// connection and surfaced a generic 502. This runs BEFORE chatCore's normal readiness/idle-timeout +// pipeline takes over, so FETCH_BODY_TIMEOUT_MS / STREAM_IDLE_TIMEOUT_MS never applied to it. +// +// Fix: every read in the peek loop and the re-assembled passthrough body is now bounded by a +// PER-READ timeout (open-sse/executors/codex/bodyTimeout.ts), so a stalled body settles fast +// instead of hanging, while a long-but-alive stream that keeps emitting chunks never trips it. +import test from "node:test"; +import assert from "node:assert/strict"; + +import { peekCodexSseTransientError } from "../../open-sse/executors/codex.ts"; + +// Small, explicit override — never depends on the ~120s/600s production default, so this test +// settles fast and deterministically regardless of env configuration. +const TEST_TIMEOUT_MS = 200; + +function stuckSseResponse(): Response { + const stuck = new ReadableStream({ + pull() { + // Never enqueue and never close — simulates an upstream body that goes silent. + }, + }); + return new Response(stuck, { + status: 200, + headers: { "content-type": "text/event-stream" }, + }); +} + +test( + "peekCodexSseTransientError does not hang forever on a silently stuck SSE body (#8020)", + { timeout: 5000 }, + async () => { + const start = Date.now(); + const peek = await peekCodexSseTransientError(stuckSseResponse(), TEST_TIMEOUT_MS); + const elapsed = Date.now() - start; + + assert.ok( + elapsed < 2000, + `expected peek to settle within 2000ms via a body timeout, took ${elapsed}ms` + ); + assert.equal(peek.timedOut, true, "expected the peek to report timedOut on a stalled body"); + assert.equal(peek.matched, null); + assert.equal(peek.replacementBody, null); + } +); + +test( + "peekCodexSseTransientError still detects a transient error when the body responds promptly", + async () => { + const encoder = new TextEncoder(); + const response = new Response( + new ReadableStream({ + pull(controller) { + controller.enqueue( + encoder.encode( + 'event: error\ndata: {"error":{"message":"Selected model is at capacity."}}\n\n' + ) + ); + controller.close(); + }, + }), + { status: 200, headers: { "content-type": "text/event-stream" } } + ); + + const peek = await peekCodexSseTransientError(response, TEST_TIMEOUT_MS); + assert.equal(peek.timedOut ?? false, false); + assert.match(peek.matched ?? "", /capacity/); + } +);