From 1d7016bf65bbbe39e38a546922bb25994eaece20 Mon Sep 17 00:00:00 2001 From: yunyaozhou Date: Mon, 29 Jun 2026 20:54:25 +0800 Subject: [PATCH] retry stream capacity errors in gateway --- gateway.mjs | 420 ++++++++++++++++++++++++++--------- scripts/test-gateway-e2e.mjs | 45 ++++ ui-src/src/App.tsx | 23 +- 3 files changed, 371 insertions(+), 117 deletions(-) diff --git a/gateway.mjs b/gateway.mjs index 44d2db2..e0fe262 100644 --- a/gateway.mjs +++ b/gateway.mjs @@ -6,6 +6,7 @@ import { chmod, copyFile, mkdir, readFile, readdir, rm, writeFile } from "node:f import fs from "node:fs"; import path from "node:path"; import { DatabaseSync } from "node:sqlite"; +import zlib from "node:zlib"; import { TextDecoder } from "node:util"; import { fileURLToPath } from "node:url"; @@ -23,6 +24,7 @@ const PROFILE_PROBE_API_PATH = `${ADMIN_BASE_PATH}/api/profiles/probe`; const PROFILE_SWITCH_API_PATH = `${ADMIN_BASE_PATH}/api/profiles/switch`; const PROFILE_ITEM_API_PREFIX = `${ADMIN_BASE_PATH}/api/profiles/`; const RESTORE_API_PATH = `${ADMIN_BASE_PATH}/api/restore`; +const STATUS_REASONING_COUNT_LIMIT = 24; const DEFAULT_CONFIG = { profile_name: "default", @@ -1091,6 +1093,14 @@ async function writeProfile(runtime, payload) { function buildMetricsSnapshot(monitor) { const reasoning516Count = monitor.observed_reasoning_counts["516"] || 0; const inspectedResponseCount = monitor.inspected_response_count; + const reasoningEntries = Object.entries(monitor.observed_reasoning_counts).sort((left, right) => { + const countDelta = Number(right[1] || 0) - Number(left[1] || 0); + if (countDelta !== 0) { + return countDelta; + } + return Number(left[0] || 0) - Number(right[0] || 0); + }); + const visibleReasoningEntries = reasoningEntries.slice(0, STATUS_REASONING_COUNT_LIMIT); return { started_at: monitor.started_at, persistent_since: monitor.persistent_since, @@ -1101,7 +1111,12 @@ function buildMetricsSnapshot(monitor) { reasoning_516_ratio: inspectedResponseCount === 0 ? 0 : reasoning516Count / inspectedResponseCount, token_totals: { ...monitor.token_totals }, - observed_reasoning_counts: { ...monitor.observed_reasoning_counts }, + observed_reasoning_counts: Object.fromEntries(visibleReasoningEntries), + observed_reasoning_counts_total_keys: reasoningEntries.length, + observed_reasoning_counts_omitted: Math.max( + 0, + reasoningEntries.length - visibleReasoningEntries.length, + ), }; } @@ -1758,12 +1773,65 @@ async function restoreRuntimeState(runtime, state) { ]); } -function jsonResponse(res, statusCode, payload, headers = {}) { +function pickContentEncoding(acceptEncoding = "") { + const value = `${acceptEncoding}`.toLowerCase(); + if (value.includes("br")) { + return "br"; + } + if (value.includes("gzip")) { + return "gzip"; + } + return null; +} + +function shouldCompressContent(headers = {}) { + const contentType = `${headers["content-type"] || headers["Content-Type"] || ""}`.toLowerCase(); + return ( + contentType.startsWith("text/") || + contentType.includes("javascript") || + contentType.includes("json") || + contentType.includes("xml") || + contentType.includes("svg") + ); +} + +function maybeCompressBody(body, headers = {}, acceptEncoding = "") { + if (!body || body.length < 1024 || !shouldCompressContent(headers)) { + return { body, encoding: null }; + } + const encoding = pickContentEncoding(acceptEncoding); + if (encoding === "br") { + return { body: zlib.brotliCompressSync(body), encoding }; + } + if (encoding === "gzip") { + return { body: zlib.gzipSync(body), encoding }; + } + return { body, encoding: null }; +} + +function respondBuffer(res, statusCode, body, headers = {}, acceptEncoding = "") { + const { body: responseBody, encoding } = maybeCompressBody(body, headers, acceptEncoding); res.writeHead(statusCode, { - "content-type": "application/json; charset=utf-8", + "content-length": responseBody.length, + vary: "accept-encoding", + ...(encoding ? { "content-encoding": encoding } : {}), ...headers, }); - res.end(JSON.stringify(payload)); + res.end(responseBody); +} + +function jsonResponse(req, res, statusCode, payload, headers = {}) { + const body = Buffer.from(JSON.stringify(payload)); + respondBuffer( + res, + statusCode, + body, + { + "content-type": "application/json; charset=utf-8", + ...headers, + }, + req?.headers?.["accept-encoding"] || "", + ); } @@ -1794,26 +1862,31 @@ function safeJoinStatic(root, requestPath) { return fullPath; } -async function serveStaticFile(res, filePath) { +async function serveStaticFile(req, res, filePath) { try { const body = await readFile(filePath); - res.writeHead(200, { - "content-type": contentTypeForFile(filePath), - "cache-control": filePath.includes(`${path.sep}assets${path.sep}`) - ? "public, max-age=31536000, immutable" - : "no-cache", - }); - res.end(body); + respondBuffer( + res, + 200, + body, + { + "content-type": contentTypeForFile(filePath), + "cache-control": filePath.includes(`${path.sep}assets${path.sep}`) + ? "public, max-age=31536000, immutable" + : "no-cache", + }, + req?.headers?.["accept-encoding"] || "", + ); return true; } catch { return false; } } -async function serveManagementUi(res, requestPathname) { +async function serveManagementUi(req, res, requestPathname) { const uiPrefix = `${UI_PATH}/`; if (requestPathname === UI_PATH || requestPathname === `${UI_PATH}/`) { - return serveStaticFile(res, path.join(UI_STATIC_ROOT, "index.html")); + return serveStaticFile(req, res, path.join(UI_STATIC_ROOT, "index.html")); } if (!requestPathname.startsWith(uiPrefix)) { @@ -1822,7 +1895,7 @@ async function serveManagementUi(res, requestPathname) { const staticPath = safeJoinStatic(UI_STATIC_ROOT, requestPathname.slice(uiPrefix.length)); if (!staticPath) { - jsonResponse(res, 403, { + jsonResponse(req, res, 403, { error: { message: "invalid static path", code: "invalid_static_path", @@ -1831,11 +1904,11 @@ async function serveManagementUi(res, requestPathname) { return true; } - if (await serveStaticFile(res, staticPath)) { + if (await serveStaticFile(req, res, staticPath)) { return true; } - return serveStaticFile(res, path.join(UI_STATIC_ROOT, "index.html")); + return serveStaticFile(req, res, path.join(UI_STATIC_ROOT, "index.html")); } function buildEditableConfig(currentConfig, payload) { @@ -1913,8 +1986,8 @@ async function handleManagementRequest(runtime, req, res, requestUrl) { const pathname = normalizePath(requestUrl.pathname); if (pathname === UI_PATH || pathname.startsWith(`${UI_PATH}/`)) { - if (!(await serveManagementUi(res, requestUrl.pathname))) { - jsonResponse(res, 503, { + if (!(await serveManagementUi(req, res, requestUrl.pathname))) { + jsonResponse(req, res, 503, { error: { message: "UI assets were not built. Run: npm run build:ui", code: "ui_not_built", @@ -1926,7 +1999,7 @@ async function handleManagementRequest(runtime, req, res, requestUrl) { if (pathname === STATUS_API_PATH && req.method === "GET") { const state = await readRuntimeState(runtime); - jsonResponse(res, 200, { + jsonResponse(req, res, 200, { ok: true, listen: `${runtime.config.listen_host}:${runtime.config.listen_port}`, config: sanitizeConfigForStatus(runtime.config), @@ -1949,7 +2022,7 @@ async function handleManagementRequest(runtime, req, res, requestUrl) { const limitRaw = requestUrl.searchParams.get("limit"); const sinceSeq = sinceSeqRaw === null ? null : Number.parseInt(sinceSeqRaw, 10); const limit = limitRaw === null ? 500 : Number.parseInt(limitRaw, 10); - jsonResponse(res, 200, { + jsonResponse(req, res, 200, { ok: true, ...await buildPersistentLogsSnapshot( runtime, @@ -1967,7 +2040,7 @@ async function handleManagementRequest(runtime, req, res, requestUrl) { const filter = requestUrl.searchParams.get("filter") || "all"; const limit = limitRaw === null ? 50 : Number.parseInt(limitRaw, 10); const offset = offsetRaw === null ? 0 : Number.parseInt(offsetRaw, 10); - jsonResponse(res, 200, { + jsonResponse(req, res, 200, { ok: true, ...await buildPersistentRequestsSnapshot(runtime, { limit: Number.isInteger(limit) ? limit : 50, @@ -1980,7 +2053,7 @@ async function handleManagementRequest(runtime, req, res, requestUrl) { } if (pathname === PROFILES_API_PATH && req.method === "GET") { - jsonResponse(res, 200, { + jsonResponse(req, res, 200, { ok: true, profiles_dir: runtime.paths.profilesDir, active_profile: runtime.config.profile_name || "default", @@ -1993,7 +2066,7 @@ async function handleManagementRequest(runtime, req, res, requestUrl) { const body = await readRequestBody(req, runtime.config.request_body_limit_bytes); const payload = parseJsonSafely(body); if (!payload) { - jsonResponse(res, 400, { + jsonResponse(req, res, 400, { error: { message: "profile 保存请求必须是有效 JSON", code: "invalid_json", @@ -2008,7 +2081,7 @@ async function handleManagementRequest(runtime, req, res, requestUrl) { applied = await applyProfileConfig(runtime, result.name); } runtime.logger(`[profile] saved name=${result.name} path=${result.file_path}`); - jsonResponse(res, 200, { + jsonResponse(req, res, 200, { ok: true, message: applied ? "profile 已保存并已热应用" : "profile 已保存", saved_profile: result, @@ -2024,7 +2097,7 @@ async function handleManagementRequest(runtime, req, res, requestUrl) { const body = await readRequestBody(req, runtime.config.request_body_limit_bytes); const payload = parseJsonSafely(body); if (!payload) { - jsonResponse(res, 400, { + jsonResponse(req, res, 400, { error: { message: "profile probe 请求必须是有效 JSON", code: "invalid_json", @@ -2037,7 +2110,7 @@ async function handleManagementRequest(runtime, req, res, requestUrl) { runtime.logger( `[profile-probe] profile=${result.profile} auth=${result.auth_mode}/${result.auth_source} upstream=${result.upstream_base_url}`, ); - jsonResponse(res, 200, { + jsonResponse(req, res, 200, { ok: true, ...result, }); @@ -2049,7 +2122,7 @@ async function handleManagementRequest(runtime, req, res, requestUrl) { const payload = parseJsonSafely(body); const profileName = `${payload?.profile || ""}`.trim(); if (!profileName) { - jsonResponse(res, 400, { + jsonResponse(req, res, 400, { error: { message: "缺少 profile", code: "profile_required", @@ -2059,7 +2132,7 @@ async function handleManagementRequest(runtime, req, res, requestUrl) { } const result = await applyProfileConfig(runtime, profileName); - jsonResponse(res, 200, { + jsonResponse(req, res, 200, { ok: true, message: "profile 已热切换，无需重启 gateway", ...result, @@ -2070,7 +2143,7 @@ async function handleManagementRequest(runtime, req, res, requestUrl) { if (pathname.startsWith(PROFILE_ITEM_API_PREFIX) && req.method === "DELETE") { const rawName = pathname.slice(PROFILE_ITEM_API_PREFIX.length); if (!rawName || rawName.includes("/")) { - jsonResponse(res, 400, { + jsonResponse(req, res, 400, { error: { message: "无效的 profile 名称", code: "invalid_profile_name", @@ -2082,7 +2155,7 @@ async function handleManagementRequest(runtime, req, res, requestUrl) { const profileName = decodeURIComponent(rawName); const result = await deleteProfile(runtime, profileName); runtime.logger(`[profile] deleted name=${result.name} path=${result.file_path}`); - jsonResponse(res, 200, { + jsonResponse(req, res, 200, { ok: true, message: "profile 已删除", deleted_profile: result, @@ -2097,7 +2170,7 @@ async function handleManagementRequest(runtime, req, res, requestUrl) { const body = await readRequestBody(req, runtime.config.request_body_limit_bytes); const payload = parseJsonSafely(body); if (!payload) { - jsonResponse(res, 400, { + jsonResponse(req, res, 400, { error: { message: "配置保存请求必须是有效 JSON", code: "invalid_json", @@ -2113,7 +2186,7 @@ async function handleManagementRequest(runtime, req, res, requestUrl) { `[config] updated reasoning_equals=${nextConfig.reasoning_equals.join(",")} retryable_status_codes=${nextConfig.retryable_status_codes.join(",")} endpoints=${nextConfig.endpoints.join(",")}`, ); const state = await readRuntimeState(runtime); - jsonResponse(res, 200, { + jsonResponse(req, res, 200, { ok: true, message: "配置已保存并立即生效", config: sanitizeConfigForStatus(runtime.config), @@ -2132,7 +2205,7 @@ async function handleManagementRequest(runtime, req, res, requestUrl) { if (pathname === RESTORE_API_PATH && req.method === "POST") { const state = await readRuntimeState(runtime); if (!state) { - jsonResponse(res, 409, { + jsonResponse(req, res, 409, { error: { message: "当前未检测到安装状态，无法恢复 Codex 原设置", code: "state_not_found", @@ -2143,7 +2216,7 @@ async function handleManagementRequest(runtime, req, res, requestUrl) { await restoreRuntimeState(runtime, state); runtime.logger(`[restore] restored via UI state_root=${runtime.paths.stateRoot}`); - jsonResponse(res, 202, { + jsonResponse(req, res, 202, { ok: true, message: "原设置已恢复，gateway 即将关闭", }); @@ -2366,6 +2439,18 @@ function findRetryableUpstreamErrorMatch(config, upstreamStatusCode, parsedBody, return null; } + return matchRetryableMessage(config, parsedBody, bodyText); +} + +function matchRetryableMessage(config, parsedBody, bodyText) { + const retryableMessages = normalizePhraseList( + config.retryable_error_messages, + DEFAULT_CONFIG.retryable_error_messages, + ); + if (retryableMessages.length === 0) { + return null; + } + const normalizedPatterns = retryableMessages.map((message) => ({ original: message, normalized: message.toLowerCase(), @@ -2395,6 +2480,32 @@ function findRetryableUpstreamErrorMatch(config, upstreamStatusCode, parsedBody, return null; } +function isRetryableStreamErrorShape(parsedBody, eventName = "") { + if (`${eventName || ""}`.trim().toLowerCase() === "error") { + return true; + } + if (!parsedBody || typeof parsedBody !== "object" || Array.isArray(parsedBody)) { + return false; + } + if (Object.hasOwn(parsedBody, "error")) { + return true; + } + if (typeof parsedBody.type === "string" && parsedBody.type.toLowerCase().includes("error")) { + return true; + } + if (typeof parsedBody.event === "string" && parsedBody.event.toLowerCase().includes("error")) { + return true; + } + return Number.isInteger(parsedBody.status) && parsedBody.status >= 400; +} + +function findRetryableStreamErrorMatch(config, parsedBody, bodyText, eventName = "") { + if (!isRetryableStreamErrorShape(parsedBody, eventName)) { + return null; + } + return matchRetryableMessage(config, parsedBody, bodyText); +} + function isExpectedStreamTermination(error) { if (!error) { return false; @@ -2535,13 +2646,14 @@ async function fetchUpstreamWithRetry(upstreamUrl, init, config, logger, request }; } -function inspectSseChunk(state, chunk) { +function inspectSseChunk(state, chunk, config) { const decoded = state.decoder.decode(chunk, { stream: true }); state.buffer += decoded; const result = { reasoning: null, usage: null, + retryable_upstream_error: null, }; const blocks = state.buffer.split(/\r?\n\r?\n/); @@ -2552,6 +2664,10 @@ function inspectSseChunk(state, chunk) { .split(/\r?\n/) .map((line) => line.trimEnd()) .filter(Boolean); + const eventName = lines + .filter((line) => line.startsWith("event:")) + .map((line) => line.replace(/^event:\s?/, "").trim()) + .find(Boolean) || ""; const dataLines = lines .filter((line) => line.startsWith("data:")) .map((line) => line.replace(/^data:\s?/, "")); @@ -2563,8 +2679,9 @@ function inspectSseChunk(state, chunk) { if (payloadText === "[DONE]") { continue; } + let parsed = null; try { - const parsed = JSON.parse(payloadText); + parsed = JSON.parse(payloadText); const reasoning = extractReasoningTokens(parsed); if (reasoning !== null) { result.reasoning = reasoning; @@ -2573,6 +2690,15 @@ function inspectSseChunk(state, chunk) { } catch { // ignore malformed SSE payloads } + const retryableUpstreamError = findRetryableStreamErrorMatch( + config, + parsed, + payloadText, + eventName, + ); + if (retryableUpstreamError) { + result.retryable_upstream_error = retryableUpstreamError; + } } return result; } @@ -2764,8 +2890,27 @@ async function handleStreaming({ const chunkBuffer = Buffer.from(value); const now = new Date(); + const inspection = inspectSseChunk(sseState, value, config); + const retryableUpstreamError = inspection.retryable_upstream_error; + if (retryableUpstreamError) { + abortController.abort(); + reader.cancel().catch(() => {}); + return { + inspected: true, + matched: true, + retry_requested: strict502Mode || !wroteAnyChunk, + retryable_upstream_error: retryableUpstreamError, + upstream_status_code: upstreamResponse.status, + reasoning_tokens: observedReasoning, + usage: observedUsage, + error: `retryable upstream error: ${retryableUpstreamError.matched_pattern}`, + match_reason: "retryable_upstream_error", + response_bytes_received: requestEntry.response_bytes_received, + stream_chunk_count: requestEntry.stream_chunk_count, + }; + } + markAndPersistFirstResponse(runtime, requestEntry, now); - const inspection = inspectSseChunk(sseState, value); const reasoning = inspection.reasoning; const usageUpdated = Boolean(inspection.usage); observedUsage = mergeUsageSnapshots(observedUsage, inspection.usage); @@ -2830,6 +2975,10 @@ async function handleStreaming({ async function proxyRequest(runtime, req, res) { const { logger } = runtime; const config = runtime.config; + const maxUpstreamAttempts = normalizePositiveInteger( + config.upstream_fetch_retry_attempts, + DEFAULT_CONFIG.upstream_fetch_retry_attempts, + ); const requestStartedAt = new Date(); const requestStartedMs = Date.now(); const incomingUrl = new URL(req.url, `http://${req.headers.host || "127.0.0.1"}`); @@ -2880,6 +3029,7 @@ async function proxyRequest(runtime, req, res) { const { requestJson, remapped, forwardedModel } = remapRequestModel(config, parsedRequestJson); const requestBody = remapped ? Buffer.from(JSON.stringify(requestJson)) : rawRequestBody; const requestIsStream = Boolean(requestJson?.stream); + let totalUpstreamAttempts = 0; requestEntry.request_body_bytes = rawRequestBody.length; requestEntry.model = requestJson?.model || null; requestEntry.requested_model = parsedRequestJson?.model || null; @@ -2889,7 +3039,6 @@ async function proxyRequest(runtime, req, res) { upsertRequestEntry(runtime, requestEntry); const upstreamUrl = buildUpstreamUrl(config.upstream_base_url, incomingUrl); - const abortController = new AbortController(); const upstreamAuth = await resolveUpstreamAuth(config); requestEntry.upstream = buildUpstreamSnapshot({ upstreamUrl, upstreamAuth }); upsertRequestEntry(runtime, requestEntry); @@ -2899,57 +3048,129 @@ async function proxyRequest(runtime, req, res) { ); } - const { - response: upstreamResponse, - attempt_count: upstreamAttemptCount, - retryable_upstream_error: terminalRetryableUpstreamError, - } = await fetchUpstreamWithRetry(upstreamUrl, { - method: req.method, - headers: cloneHeadersForUpstream(req.headers, upstreamAuth), - body: requestBody.length > 0 ? requestBody : undefined, - signal: abortController.signal, - }, config, logger, { method: req.method, pathname }); + while (totalUpstreamAttempts < maxUpstreamAttempts) { + const abortController = new AbortController(); + const { + response: upstreamResponse, + attempt_count: upstreamAttemptCount, + retryable_upstream_error: terminalRetryableUpstreamError, + } = await fetchUpstreamWithRetry(upstreamUrl, { + method: req.method, + headers: cloneHeadersForUpstream(req.headers, upstreamAuth), + body: requestBody.length > 0 ? requestBody : undefined, + signal: abortController.signal, + }, { + ...config, + upstream_fetch_retry_attempts: Math.max(1, maxUpstreamAttempts - totalUpstreamAttempts), + }, logger, { method: req.method, pathname }); - const shouldInspect = matchPath(config, pathname); - const responseContentType = upstreamResponse.headers.get("content-type"); - const responseIsStream = isSseContentType(responseContentType) || ( - requestIsStream && - !isJsonContentType(responseContentType) && - !isUpstreamErrorStatus(upstreamResponse.status) - ); - requestEntry.response_stream = responseIsStream; - requestEntry.inspected = shouldInspect; - requestEntry.upstream_status_code = upstreamResponse.status; - requestEntry.upstream_attempt_count = upstreamAttemptCount; - requestEntry.upstream = buildUpstreamSnapshot({ upstreamUrl, upstreamAuth, upstreamResponse }); - upsertRequestEntry(runtime, requestEntry); - if (isUpstreamErrorStatus(upstreamResponse.status)) { - logger?.( - `[upstream] status=${upstreamResponse.status} profile=${config.profile_name || "default"} path=${requestEntry.upstream.path} auth=${requestEntry.upstream.auth_mode}/${requestEntry.upstream.auth_source} content_type=${requestEntry.upstream.content_type || "-"}`, + totalUpstreamAttempts += upstreamAttemptCount; + + const shouldInspect = matchPath(config, pathname); + const responseContentType = upstreamResponse.headers.get("content-type"); + const responseIsStream = isSseContentType(responseContentType) || ( + requestIsStream && + !isJsonContentType(responseContentType) && + !isUpstreamErrorStatus(upstreamResponse.status) ); - } - - if (!shouldInspect) { - markRequestFirstResponse(requestEntry); + requestEntry.response_stream = responseIsStream; + requestEntry.inspected = shouldInspect; + requestEntry.upstream_status_code = upstreamResponse.status; + requestEntry.upstream_attempt_count = totalUpstreamAttempts; + requestEntry.upstream = buildUpstreamSnapshot({ upstreamUrl, upstreamAuth, upstreamResponse }); upsertRequestEntry(runtime, requestEntry); - copyHeadersToClient(upstreamResponse.headers, res); - res.writeHead(upstreamResponse.status); - const body = Buffer.from(await upstreamResponse.arrayBuffer()); - res.end(body); - recordRequestEntry( - runtime, - finalizeRequestEntry(requestEntry, { - status_code: upstreamResponse.status, - upstream_status_code: upstreamResponse.status, - inspected: false, - }), - config.request_history_limit, - ); - return; - } + if (isUpstreamErrorStatus(upstreamResponse.status)) { + logger?.( + `[upstream] status=${upstreamResponse.status} profile=${config.profile_name || "default"} path=${requestEntry.upstream.path} auth=${requestEntry.upstream.auth_mode}/${requestEntry.upstream.auth_source} content_type=${requestEntry.upstream.content_type || "-"}`, + ); + } - if (responseIsStream) { - const result = await handleStreaming({ + if (!shouldInspect) { + markRequestFirstResponse(requestEntry); + upsertRequestEntry(runtime, requestEntry); + copyHeadersToClient(upstreamResponse.headers, res); + res.writeHead(upstreamResponse.status); + const body = Buffer.from(await upstreamResponse.arrayBuffer()); + res.end(body); + recordRequestEntry( + runtime, + finalizeRequestEntry(requestEntry, { + status_code: upstreamResponse.status, + upstream_status_code: upstreamResponse.status, + inspected: false, + }), + config.request_history_limit, + ); + return; + } + + if (responseIsStream) { + const result = await handleStreaming({ + runtime, + config, + logger, + monitor: runtime.monitor, + pathname, + upstreamResponse, + res, + abortController, + requestEntry, + }); + if (result.retry_requested) { + if (totalUpstreamAttempts < maxUpstreamAttempts) { + const backoffMs = computeRetryBackoffMs(config, totalUpstreamAttempts); + logger?.( + `[retry] upstream retryable stream error attempt=${totalUpstreamAttempts} next_attempt=${totalUpstreamAttempts + 1} status=${upstreamResponse.status} path=${pathname || "-"} reason=${JSON.stringify(result.retryable_upstream_error?.matched_pattern)} backoff_ms=${backoffMs}`, + ); + if (backoffMs > 0) { + await sleep(backoffMs); + } + continue; + } + + recordInspectedResponse(runtime.monitor, result.reasoning_tokens, true); + if (config.log_match) { + logger?.( + `[match] stream path=${pathname} upstream_status=${upstreamResponse.status} retryable_error=${JSON.stringify(result.retryable_upstream_error?.matched_pattern)} action=status_${config.non_stream_status_code}`, + ); + } + const blockedBody = buildRetryableUpstreamErrorBody( + pathname, + upstreamResponse.status, + result.retryable_upstream_error?.matched_message || result.retryable_upstream_error?.matched_pattern, + config.non_stream_status_code, + ); + res.writeHead(config.non_stream_status_code, { + "content-type": "application/json; charset=utf-8", + "x-codex-retry-gateway-reason": "upstream-error-retry-triggered", + }); + res.end(blockedBody); + recordRequestEntry( + runtime, + finalizeRequestEntry(requestEntry, { + response_stream: true, + ...result, + matched: true, + status_code: config.non_stream_status_code, + upstream_status_code: upstreamResponse.status, + }), + config.request_history_limit, + ); + return; + } + + recordRequestEntry( + runtime, + finalizeRequestEntry(requestEntry, { + response_stream: true, + ...result, + }), + config.request_history_limit, + ); + return; + } + + const result = await handleNonStreaming({ runtime, config, logger, @@ -2957,43 +3178,22 @@ async function proxyRequest(runtime, req, res) { pathname, upstreamResponse, res, - abortController, requestEntry, + terminalRetryableUpstreamError, }); recordRequestEntry( runtime, finalizeRequestEntry(requestEntry, { - response_stream: true, + response_stream: false, ...result, }), config.request_history_limit, ); return; } - - const result = await handleNonStreaming({ - runtime, - config, - logger, - monitor: runtime.monitor, - pathname, - upstreamResponse, - res, - requestEntry, - terminalRetryableUpstreamError, - }); - recordRequestEntry( - runtime, - finalizeRequestEntry(requestEntry, { - response_stream: false, - ...result, - }), - config.request_history_limit, - ); - return; } catch (error) { if (Number.isInteger(error?.gatewayAttemptCount)) { - requestEntry.upstream_attempt_count = error.gatewayAttemptCount; + requestEntry.upstream_attempt_count = (requestEntry.upstream_attempt_count || 0) + error.gatewayAttemptCount; } recordRequestEntry( runtime, diff --git a/scripts/test-gateway-e2e.mjs b/scripts/test-gateway-e2e.mjs index 2361d1d..473d2b4 100644 --- a/scripts/test-gateway-e2e.mjs +++ b/scripts/test-gateway-e2e.mjs @@ -80,6 +80,21 @@ function createTerminatedSseResponse(res, chunks, destroyDelayMs = 20) { }, destroyDelayMs); } +function createCapacityErrorSseResponse( + res, + message = "Selected model is at capacity. Please try a different model.", + intervalMs = 20, +) { + createSseResponse( + res, + [ + 'event: error\n', + `data: ${JSON.stringify({ error: { message, type: "server_error" } })}\n\n`, + ], + intervalMs, + ); +} + function startFakeUpstream(port) { const failBeforeResponseCounts = new Map(); const capacityBeforeSuccessCounts = new Map(); @@ -143,6 +158,13 @@ function startFakeUpstream(port) { const capacityCount = (capacityBeforeSuccessCounts.get(capacityKey) || 0) + 1; capacityBeforeSuccessCounts.set(capacityKey, capacityCount); if (capacityCount <= parsed.test_capacity_before_success_times) { + if (parsed.stream) { + createCapacityErrorSseResponse( + res, + parsed.test_capacity_message || "Selected model is at capacity. Please try a different model.", + ); + return; + } createJsonResponse( res, parsed.test_capacity_status ?? 503, @@ -157,6 +179,13 @@ function startFakeUpstream(port) { return; } } + if (parsed.stream && parsed.test_capacity_error) { + createCapacityErrorSseResponse( + res, + parsed.test_capacity_message || "Selected model is at capacity. Please try a different model.", + ); + return; + } if (parsed.stream) { createSseResponse(res, [ 'data: {"type":"response.output_text.delta","delta":"hello"}\n\n', @@ -434,6 +463,22 @@ async function run() { "stream+capacity error 返回体未标记 retry trigger", ); + const streamCapacityRecoveredResponse = await fetch(`http://127.0.0.1:${gatewayPort}/responses`, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ stream: true, test_capacity_before_success_times: 2, test_reasoning_tokens: 128 }), + }); + const streamCapacityRecoveredText = await streamCapacityRecoveredResponse.text(); + assert(streamCapacityRecoveredResponse.status === 200, `stream capacity 抖动后未自动恢复: ${streamCapacityRecoveredResponse.status}`); + assert(streamCapacityRecoveredText.includes("hello"), "stream capacity 恢复后未拿到正常 SSE 内容"); + + const requestsAfterStreamCapacityRecoveryResponse = await fetch(`http://127.0.0.1:${gatewayPort}/__codex_retry_gateway/api/requests?limit=20`); + const requestsAfterStreamCapacityRecovery = await requestsAfterStreamCapacityRecoveryResponse.json(); + const streamCapacityRecoveredEntry = requestsAfterStreamCapacityRecovery?.entries?.find( + (entry) => entry.path === "/responses" && entry.status_code === 200 && entry.response_stream && entry.upstream_attempt_count >= 3, + ); + assert(streamCapacityRecoveredEntry, "stream capacity 抖动恢复后的请求记录未保留重试次数"); + for (const streamPath of [ "/responses", "/v1/responses", diff --git a/ui-src/src/App.tsx b/ui-src/src/App.tsx index 5dab092..588b9f7 100644 --- a/ui-src/src/App.tsx +++ b/ui-src/src/App.tsx @@ -39,6 +39,8 @@ type Metrics = { cached_tokens?: number; }; observed_reasoning_counts?: Record; + observed_reasoning_counts_total_keys?: number; + observed_reasoning_counts_omitted?: number; }; type StatusPayload = { @@ -222,7 +224,7 @@ const api = { restore: "/__codex_retry_gateway/api/restore", }; -const REQUEST_PAGE_SIZE = 40; +const REQUEST_PAGE_SIZE = 20; const LOG_PAGE_SIZE = 200; const zhNumberFormatter = new Intl.NumberFormat("zh-CN"); @@ -666,7 +668,7 @@ export default function App() { }, [page, requestQuery, requestFilter, latestLogSeq, requestLimit]); useEffect(() => { - if (page === "overview" || page === "rules") { + if (page === "overview" || page === "rules" || page === "requests") { return; } loadPageData(page, { incrementalLogs: false }).catch((error) => { @@ -997,11 +999,18 @@ export default function App() { {reasoningChips.length === 0 ? ( 还没有 reasoning 观测 ) : ( - reasoningChips.map(([reasoning, count]) => ( - - reasoning {reasoning}: {count} - - )) + <> + {reasoningChips.map(([reasoning, count]) => ( + + reasoning {reasoning}: {count} + + ))} + {(metrics.observed_reasoning_counts_omitted || 0) > 0 ? ( + + 其余 {metrics.observed_reasoning_counts_omitted} 项未展开 + + ) : null} + )}