From 57c7264c7b784ce4f375e30345eaf1ef67248ba7 Mon Sep 17 00:00:00 2001 From: yunyaozhou Date: Mon, 29 Jun 2026 18:04:53 +0800 Subject: [PATCH] harden upstream retry handling --- gateway.mjs | 306 +++++++++++++++++++++++++++++++++-- scripts/run-profile.mjs | 6 + scripts/test-gateway-e2e.mjs | 111 ++++++++++--- ui-src/src/App.tsx | 65 ++++++++ ui-src/src/styles.css | 10 ++ 5 files changed, 465 insertions(+), 33 deletions(-) diff --git a/gateway.mjs b/gateway.mjs index 80fda1c..96c7602 100644 --- a/gateway.mjs +++ b/gateway.mjs @@ -43,6 +43,8 @@ const DEFAULT_CONFIG = { retryable_error_messages: [ "Selected model is at capacity. Please try a different model.", ], + upstream_fetch_retry_attempts: 5, + upstream_fetch_retry_backoff_ms: 350, non_stream_status_code: 502, stream_action: "strict_502", log_match: true, @@ -242,6 +244,29 @@ function normalizePhraseList(values, fallback = []) { return [...new Set(normalized)]; } +function normalizePositiveInteger(value, fallback) { + const parsed = Number.parseInt(`${value ?? ""}`, 10); + if (Number.isInteger(parsed) && parsed > 0) { + return parsed; + } + return fallback; +} + +function normalizeNonNegativeInteger(value, fallback) { + const parsed = Number.parseInt(`${value ?? ""}`, 10); + if (Number.isInteger(parsed) && parsed >= 0) { + return parsed; + } + return fallback; +} + +function sleep(ms) { + if (!Number.isFinite(ms) || ms <= 0) { + return Promise.resolve(); + } + return new Promise((resolve) => setTimeout(resolve, ms)); +} + function parseModelRemapMap(value) { const map = {}; for (const rawEntry of `${value || ""}`.split(/\r?\n|[;,]/)) { @@ -817,6 +842,10 @@ function buildProfileFormModel(env) { env.CODEX_RETRY_GATEWAY_RETRYABLE_ERROR_MESSAGES || DEFAULT_CONFIG.retryable_error_messages, DEFAULT_CONFIG.retryable_error_messages, ), + upstream_fetch_retry_attempts: + env.CODEX_RETRY_GATEWAY_UPSTREAM_FETCH_RETRY_ATTEMPTS || `${DEFAULT_CONFIG.upstream_fetch_retry_attempts}`, + upstream_fetch_retry_backoff_ms: + env.CODEX_RETRY_GATEWAY_UPSTREAM_FETCH_RETRY_BACKOFF_MS || `${DEFAULT_CONFIG.upstream_fetch_retry_backoff_ms}`, endpoints: normalizeStringList(env.CODEX_RETRY_GATEWAY_ENDPOINTS || DEFAULT_CONFIG.endpoints, DEFAULT_CONFIG.endpoints), }; } @@ -852,6 +881,18 @@ function buildConfigFromProfileEnv(profileName, env) { env.CODEX_RETRY_GATEWAY_RETRYABLE_ERROR_MESSAGES || DEFAULT_CONFIG.retryable_error_messages, DEFAULT_CONFIG.retryable_error_messages, ), + upstream_fetch_retry_attempts: env.CODEX_RETRY_GATEWAY_UPSTREAM_FETCH_RETRY_ATTEMPTS + ? normalizePositiveInteger( + env.CODEX_RETRY_GATEWAY_UPSTREAM_FETCH_RETRY_ATTEMPTS, + DEFAULT_CONFIG.upstream_fetch_retry_attempts, + ) + : DEFAULT_CONFIG.upstream_fetch_retry_attempts, + upstream_fetch_retry_backoff_ms: env.CODEX_RETRY_GATEWAY_UPSTREAM_FETCH_RETRY_BACKOFF_MS + ? normalizeNonNegativeInteger( + env.CODEX_RETRY_GATEWAY_UPSTREAM_FETCH_RETRY_BACKOFF_MS, + DEFAULT_CONFIG.upstream_fetch_retry_backoff_ms, + ) + : DEFAULT_CONFIG.upstream_fetch_retry_backoff_ms, non_stream_status_code: env.CODEX_RETRY_GATEWAY_NON_STREAM_STATUS_CODE ? Number.parseInt(`${env.CODEX_RETRY_GATEWAY_NON_STREAM_STATUS_CODE}`, 10) : DEFAULT_CONFIG.non_stream_status_code, @@ -956,12 +997,26 @@ async function buildProfileEnvText(payload) { payload.retryable_error_messages, DEFAULT_CONFIG.retryable_error_messages, ); + const upstreamFetchRetryAttempts = normalizePositiveInteger( + payload.upstream_fetch_retry_attempts, + DEFAULT_CONFIG.upstream_fetch_retry_attempts, + ); + const upstreamFetchRetryBackoffMs = normalizeNonNegativeInteger( + payload.upstream_fetch_retry_backoff_ms, + DEFAULT_CONFIG.upstream_fetch_retry_backoff_ms, + ); if (retryableStatusCodes.length === 0) { throw new Error("retryable_status_codes 不能为空"); } if (retryableErrorMessages.length === 0) { throw new Error("retryable_error_messages 不能为空"); } + if (upstreamFetchRetryAttempts < 1) { + throw new Error("upstream_fetch_retry_attempts 必须是正整数"); + } + if (upstreamFetchRetryBackoffMs < 0) { + throw new Error("upstream_fetch_retry_backoff_ms 不能为负数"); + } const envPairs = [ ["CODEX_RETRY_GATEWAY_LISTEN_HOST", `${payload.listen_host || DEFAULT_CONFIG.listen_host}`.trim()], @@ -971,6 +1026,8 @@ async function buildProfileEnvText(payload) { ["CODEX_RETRY_GATEWAY_REASONING_EQUALS", reasoningEquals.join(",")], ["CODEX_RETRY_GATEWAY_RETRYABLE_STATUS_CODES", retryableStatusCodes.join(",")], ["CODEX_RETRY_GATEWAY_RETRYABLE_ERROR_MESSAGES", retryableErrorMessages.join("\n")], + ["CODEX_RETRY_GATEWAY_UPSTREAM_FETCH_RETRY_ATTEMPTS", `${upstreamFetchRetryAttempts}`], + ["CODEX_RETRY_GATEWAY_UPSTREAM_FETCH_RETRY_BACKOFF_MS", `${upstreamFetchRetryBackoffMs}`], ["CODEX_RETRY_GATEWAY_REQUEST_HISTORY_LIMIT", `${requestHistoryLimit}`], ["CODEX_RETRY_GATEWAY_ENDPOINTS", endpoints.join(",")], ]; @@ -1134,17 +1191,22 @@ function buildRequestEntry({ seq, startedAt, startedMs, req, pathname, requestJs started_at: startedAt.toISOString(), first_response_at: null, first_response_delay_ms: null, + last_activity_at: null, finished_at: null, duration_ms: null, profile_name: profileName || "default", method: req.method, path: pathname, request_body_bytes: null, + response_bytes_received: 0, model: requestJson?.model || null, requested_model: requestJson?.model || null, forwarded_model: requestJson?.model || null, request_stream: Boolean(requestJson?.stream), response_stream: false, + stream_chunk_count: 0, + usage_last_updated_at: null, + upstream_attempt_count: 0, inspected: false, matched: false, status_code: null, @@ -1167,14 +1229,54 @@ function markRequestFirstResponse(entry, at = new Date()) { entry.first_response_delay_ms = Number.isFinite(entry._started_ms) ? Math.max(0, firstMs - entry._started_ms) : null; + entry.last_activity_at = firstAt.toISOString(); entry.lifecycle_state = "receive_first"; return true; } +const STREAM_PROGRESS_PERSIST_INTERVAL_MS = 1000; + +function updateStreamingProgress(entry, { chunkBytes = 0, usage = null, reasoning = null, at = new Date() } = {}) { + const observedAt = at instanceof Date ? at : new Date(at); + const observedAtIso = observedAt.toISOString(); + + entry.last_activity_at = observedAtIso; + entry.lifecycle_state = "streaming"; + + if (chunkBytes > 0) { + entry.stream_chunk_count = (entry.stream_chunk_count || 0) + 1; + entry.response_bytes_received = (entry.response_bytes_received || 0) + chunkBytes; + } + + if (usage) { + entry.usage = mergeUsageSnapshots(entry.usage, usage); + entry.usage_last_updated_at = observedAtIso; + } + + if (Number.isInteger(reasoning)) { + entry.reasoning_tokens = reasoning; + } +} + +function persistStreamingProgress(runtime, entry, { force = false, usageUpdated = false } = {}, at = new Date()) { + const observedAt = at instanceof Date ? at : new Date(at); + const nowMs = observedAt.getTime(); + const lastPersistedMs = Number.isFinite(entry._last_stream_persisted_ms) + ? entry._last_stream_persisted_ms + : 0; + if (!force && !usageUpdated && nowMs - lastPersistedMs < STREAM_PROGRESS_PERSIST_INTERVAL_MS) { + return false; + } + entry._last_stream_persisted_ms = nowMs; + upsertRequestEntry(runtime, entry); + return true; +} + function finalizeRequestEntry(entry, result = {}) { const finishedAt = new Date(); const startedMs = entry._started_ms; delete entry._started_ms; + delete entry._last_stream_persisted_ms; return { ...entry, ...result, @@ -1202,6 +1304,14 @@ async function loadConfig(configPath) { config.retryable_error_messages, DEFAULT_CONFIG.retryable_error_messages, ); + config.upstream_fetch_retry_attempts = normalizePositiveInteger( + config.upstream_fetch_retry_attempts, + DEFAULT_CONFIG.upstream_fetch_retry_attempts, + ); + config.upstream_fetch_retry_backoff_ms = normalizeNonNegativeInteger( + config.upstream_fetch_retry_backoff_ms, + DEFAULT_CONFIG.upstream_fetch_retry_backoff_ms, + ); if (!config.upstream_base_url) { throw new Error("配置缺少 upstream_base_url"); } @@ -1551,10 +1661,10 @@ async function probeProfile(runtime, payload) { }; const modelsUrl = buildUpstreamUrl(config.upstream_base_url, new URL("http://local/v1/models")); - const modelsResponse = await fetchUpstreamWithRetry(modelsUrl, { + const { response: modelsResponse } = await fetchUpstreamWithRetry(modelsUrl, { method: "GET", headers: cloneHeadersForUpstream({}, upstreamAuth), - }, runtime.logger); + }, config, runtime.logger, { method: "GET", pathname: "/v1/models" }); result.probes.push({ kind: "models", target: "/v1/models", @@ -1571,11 +1681,11 @@ async function probeProfile(runtime, payload) { stream: false, }); const responsesUrl = buildUpstreamUrl(config.upstream_base_url, new URL("http://local/v1/responses")); - const responseProbe = await fetchUpstreamWithRetry(responsesUrl, { + const { response: responseProbe } = await fetchUpstreamWithRetry(responsesUrl, { method: "POST", headers: cloneHeadersForUpstream({ "content-type": "application/json" }, upstreamAuth), body: JSON.stringify(requestJson), - }, runtime.logger); + }, config, runtime.logger, { method: "POST", pathname: "/v1/responses" }); result.probes.push({ kind: "responses", target: "/v1/responses", @@ -1738,6 +1848,26 @@ function buildEditableConfig(currentConfig, payload) { currentConfig.retryable_error_messages, ); const nextEndpoints = normalizeStringList(payload.endpoints, currentConfig.endpoints).map(normalizePath); + const nextUpstreamFetchRetryAttempts = + payload.upstream_fetch_retry_attempts === undefined + ? normalizePositiveInteger( + currentConfig.upstream_fetch_retry_attempts, + DEFAULT_CONFIG.upstream_fetch_retry_attempts, + ) + : normalizePositiveInteger( + payload.upstream_fetch_retry_attempts, + DEFAULT_CONFIG.upstream_fetch_retry_attempts, + ); + const nextUpstreamFetchRetryBackoffMs = + payload.upstream_fetch_retry_backoff_ms === undefined + ? normalizeNonNegativeInteger( + currentConfig.upstream_fetch_retry_backoff_ms, + DEFAULT_CONFIG.upstream_fetch_retry_backoff_ms, + ) + : normalizeNonNegativeInteger( + payload.upstream_fetch_retry_backoff_ms, + DEFAULT_CONFIG.upstream_fetch_retry_backoff_ms, + ); const nextStatusCode = payload.non_stream_status_code === undefined ? currentConfig.non_stream_status_code @@ -1755,6 +1885,12 @@ function buildEditableConfig(currentConfig, payload) { if (nextEndpoints.length === 0) { throw new Error("endpoints 不能为空"); } + if (nextUpstreamFetchRetryAttempts < 1) { + throw new Error("upstream_fetch_retry_attempts 必须是正整数"); + } + if (nextUpstreamFetchRetryBackoffMs < 0) { + throw new Error("upstream_fetch_retry_backoff_ms 不能为负数"); + } if (!Number.isInteger(nextStatusCode) || nextStatusCode < 100 || nextStatusCode > 599) { throw new Error("non_stream_status_code 必须是 100-599 的整数"); } @@ -1765,6 +1901,8 @@ function buildEditableConfig(currentConfig, payload) { retryable_status_codes: nextRetryableStatusCodes, retryable_error_messages: nextRetryableErrorMessages, endpoints: nextEndpoints, + upstream_fetch_retry_attempts: nextUpstreamFetchRetryAttempts, + upstream_fetch_retry_backoff_ms: nextUpstreamFetchRetryBackoffMs, non_stream_status_code: nextStatusCode, log_match: payload.log_match === undefined ? currentConfig.log_match : Boolean(payload.log_match), }; @@ -2131,6 +2269,14 @@ function parseJsonSafely(buffer) { } } +function rebuildBufferedResponse(response, bodyBuffer) { + return new Response(bodyBuffer, { + status: response.status, + statusText: response.statusText, + headers: new Headers(response.headers), + }); +} + function matchPath(config, pathname) { return config.endpoints.includes(normalizePath(pathname)); } @@ -2263,23 +2409,123 @@ function isUpstreamErrorStatus(statusCode) { return Number.isInteger(statusCode) && statusCode >= 400; } -async function fetchUpstreamWithRetry(upstreamUrl, init, logger) { - const maxAttempts = 2; +function computeRetryBackoffMs(config, attempt) { + const baseDelay = normalizeNonNegativeInteger( + config?.upstream_fetch_retry_backoff_ms, + DEFAULT_CONFIG.upstream_fetch_retry_backoff_ms, + ); + if (baseDelay <= 0 || attempt <= 1) { + return baseDelay; + } + const multiplier = Math.min(attempt - 1, 4); + return baseDelay * multiplier; +} + +async function fetchUpstreamWithRetry(upstreamUrl, init, config, logger, requestContext = {}) { + const maxAttempts = normalizePositiveInteger( + config?.upstream_fetch_retry_attempts, + DEFAULT_CONFIG.upstream_fetch_retry_attempts, + ); + const method = `${requestContext.method || init?.method || "GET"}`.toUpperCase(); + const pathname = requestContext.pathname || ""; + let lastError = null; + let lastResponse = null; + let retryableUpstreamError = null; for (let attempt = 1; attempt <= maxAttempts; attempt += 1) { try { - return await fetch(upstreamUrl, init); + const response = await fetch(upstreamUrl, init); + lastResponse = response; + + if (method === "GET") { + return { + response, + attempt_count: attempt, + retryable_upstream_error: null, + }; + } + + if (!isUpstreamErrorStatus(response.status)) { + return { + response, + attempt_count: attempt, + retryable_upstream_error: null, + }; + } + + const contentType = response.headers.get("content-type"); + if (!isJsonContentType(contentType)) { + return { + response, + attempt_count: attempt, + retryable_upstream_error: null, + }; + } + + const bodyBuffer = Buffer.from(await response.arrayBuffer()); + const parsed = parseJsonSafely(bodyBuffer); + const bodyText = bodyBuffer.toString("utf8"); + retryableUpstreamError = findRetryableUpstreamErrorMatch( + config, + response.status, + parsed, + bodyText, + ); + + if (!retryableUpstreamError) { + return { + response: rebuildBufferedResponse(response, bodyBuffer), + attempt_count: attempt, + retryable_upstream_error: null, + }; + } + + if (attempt === maxAttempts) { + return { + response: rebuildBufferedResponse(response, bodyBuffer), + attempt_count: attempt, + retryable_upstream_error: { + ...retryableUpstreamError, + upstream_status_code: response.status, + }, + }; + } + + const backoffMs = computeRetryBackoffMs(config, attempt); + logger?.( + `[retry] upstream retryable error attempt=${attempt} next_attempt=${attempt + 1} status=${response.status} path=${pathname || "-"} reason=${JSON.stringify(retryableUpstreamError.matched_pattern)} backoff_ms=${backoffMs}`, + ); + if (backoffMs > 0) { + await sleep(backoffMs); + } } catch (error) { lastError = error; + if (error && typeof error === "object") { + error.gatewayAttemptCount = attempt; + } if (!isRetryableUpstreamFetchError(error) || attempt === maxAttempts) { break; } - logger?.(`[retry] upstream fetch failed attempt=${attempt} url=${upstreamUrl}`); + const backoffMs = computeRetryBackoffMs(config, attempt); + logger?.( + `[retry] upstream fetch failed attempt=${attempt} next_attempt=${attempt + 1} method=${method} path=${pathname || "-"} url=${upstreamUrl} backoff_ms=${backoffMs}`, + ); + if (backoffMs > 0) { + await sleep(backoffMs); + } } } - throw lastError; + if (lastError) { + throw lastError; + } + + return { + response: lastResponse, + attempt_count: maxAttempts, + retryable_upstream_error: retryableUpstreamError, + }; } function inspectSseChunk(state, chunk) { @@ -2333,6 +2579,7 @@ async function handleNonStreaming({ upstreamResponse, res, requestEntry, + terminalRetryableUpstreamError = null, }) { markAndPersistFirstResponse(runtime, requestEntry); const bodyBuffer = Buffer.from(await upstreamResponse.arrayBuffer()); @@ -2343,7 +2590,7 @@ async function handleNonStreaming({ const reasoning = parsed ? extractReasoningTokens(parsed) : null; const usage = parsed ? normalizeUsageSnapshot(parsed) : null; const matched = reasoningMatched(config, reasoning); - const retryableUpstreamError = findRetryableUpstreamErrorMatch( + const retryableUpstreamError = terminalRetryableUpstreamError || findRetryableUpstreamErrorMatch( config, upstreamResponse.status, parsed, @@ -2451,6 +2698,7 @@ async function handleStreaming({ } catch (error) { if (isExpectedStreamTermination(error)) { recordInspectedResponse(monitor, observedReasoning, false); + persistStreamingProgress(runtime, requestEntry, { force: true }, new Date()); if (strict502Mode) { logger?.(`[stream] upstream terminated before completion path=${pathname} action=status_502`); res.writeHead(502, { "content-type": "application/json; charset=utf-8" }); @@ -2463,6 +2711,8 @@ async function handleStreaming({ reasoning_tokens: observedReasoning, usage: observedUsage, error: "upstream stream terminated before completion", + response_bytes_received: requestEntry.response_bytes_received, + stream_chunk_count: requestEntry.stream_chunk_count, }; } else { res.end(); @@ -2474,6 +2724,8 @@ async function handleStreaming({ reasoning_tokens: observedReasoning, usage: observedUsage, error: "upstream stream terminated before completion", + response_bytes_received: requestEntry.response_bytes_received, + stream_chunk_count: requestEntry.stream_chunk_count, }; } } @@ -2483,6 +2735,7 @@ async function handleStreaming({ const { done, value } = readResult; if (done) { recordInspectedResponse(monitor, observedReasoning, false); + persistStreamingProgress(runtime, requestEntry, { force: true }, new Date()); if (strict502Mode) { copyHeadersToClient(upstreamResponse.headers, res); res.writeHead(upstreamResponse.status); @@ -2497,17 +2750,33 @@ async function handleStreaming({ upstream_status_code: upstreamResponse.status, reasoning_tokens: observedReasoning, usage: observedUsage, + response_bytes_received: requestEntry.response_bytes_received, + stream_chunk_count: requestEntry.stream_chunk_count, }; } const chunkBuffer = Buffer.from(value); - markAndPersistFirstResponse(runtime, requestEntry); + const now = new Date(); + markAndPersistFirstResponse(runtime, requestEntry, now); const inspection = inspectSseChunk(sseState, value); const reasoning = inspection.reasoning; + const usageUpdated = Boolean(inspection.usage); observedUsage = mergeUsageSnapshots(observedUsage, inspection.usage); if (Number.isInteger(reasoning)) { observedReasoning = reasoning; } + updateStreamingProgress(requestEntry, { + chunkBytes: chunkBuffer.length, + usage: inspection.usage, + reasoning, + at: now, + }); + persistStreamingProgress( + runtime, + requestEntry, + { usageUpdated, force: requestEntry.stream_chunk_count === 1 }, + now, + ); if (reasoningMatched(config, reasoning)) { recordInspectedResponse(monitor, reasoning, true); if (config.log_match) { @@ -2537,6 +2806,8 @@ async function handleStreaming({ upstream_status_code: upstreamResponse.status, reasoning_tokens: reasoning, usage: observedUsage, + response_bytes_received: requestEntry.response_bytes_received, + stream_chunk_count: requestEntry.stream_chunk_count, }; } @@ -2621,12 +2892,16 @@ async function proxyRequest(runtime, req, res) { ); } - const upstreamResponse = await fetchUpstreamWithRetry(upstreamUrl, { + 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, - }, logger); + }, config, logger, { method: req.method, pathname }); const shouldInspect = matchPath(config, pathname); const responseContentType = upstreamResponse.headers.get("content-type"); @@ -2638,6 +2913,7 @@ async function proxyRequest(runtime, req, res) { 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)) { @@ -2697,6 +2973,7 @@ async function proxyRequest(runtime, req, res) { upstreamResponse, res, requestEntry, + terminalRetryableUpstreamError, }); recordRequestEntry( runtime, @@ -2708,6 +2985,9 @@ async function proxyRequest(runtime, req, res) { ); return; } catch (error) { + if (Number.isInteger(error?.gatewayAttemptCount)) { + requestEntry.upstream_attempt_count = error.gatewayAttemptCount; + } recordRequestEntry( runtime, finalizeRequestEntry(requestEntry, { diff --git a/scripts/run-profile.mjs b/scripts/run-profile.mjs index fa98411..7eb8cdd 100644 --- a/scripts/run-profile.mjs +++ b/scripts/run-profile.mjs @@ -194,6 +194,12 @@ function buildProfileConfig({ profileName, profileEnv, existingGatewayConfig, pr profileEnv.CODEX_RETRY_GATEWAY_RETRYABLE_ERROR_MESSAGES || existingGatewayConfig?.retryable_error_messages, ["Selected model is at capacity. Please try a different model."], ), + upstream_fetch_retry_attempts: profileEnv.CODEX_RETRY_GATEWAY_UPSTREAM_FETCH_RETRY_ATTEMPTS + ? Number.parseInt(`${profileEnv.CODEX_RETRY_GATEWAY_UPSTREAM_FETCH_RETRY_ATTEMPTS}`, 10) + : Number.parseInt(`${existingGatewayConfig?.upstream_fetch_retry_attempts || 5}`, 10), + upstream_fetch_retry_backoff_ms: profileEnv.CODEX_RETRY_GATEWAY_UPSTREAM_FETCH_RETRY_BACKOFF_MS + ? Number.parseInt(`${profileEnv.CODEX_RETRY_GATEWAY_UPSTREAM_FETCH_RETRY_BACKOFF_MS}`, 10) + : Number.parseInt(`${existingGatewayConfig?.upstream_fetch_retry_backoff_ms || 350}`, 10), non_stream_status_code: profileEnv.CODEX_RETRY_GATEWAY_NON_STREAM_STATUS_CODE ? Number.parseInt(`${profileEnv.CODEX_RETRY_GATEWAY_NON_STREAM_STATUS_CODE}`, 10) : Number.parseInt(`${existingGatewayConfig?.non_stream_status_code || 502}`, 10), diff --git a/scripts/test-gateway-e2e.mjs b/scripts/test-gateway-e2e.mjs index e278ec2..2361d1d 100644 --- a/scripts/test-gateway-e2e.mjs +++ b/scripts/test-gateway-e2e.mjs @@ -39,24 +39,24 @@ function createJsonResponse(res, statusCode, body, extraHeaders = {}) { res.end(JSON.stringify(body)); } -function createSseResponse(res, chunks) { - res.writeHead(200, { - "content-type": "text/event-stream; charset=utf-8", - "cache-control": "no-cache", - connection: "keep-alive", +function createSseResponse(res, chunks, intervalMs = 20) { + res.writeHead(200, { + "content-type": "text/event-stream; charset=utf-8", + "cache-control": "no-cache", + connection: "keep-alive", "x-upstream-test": "sse", }); let index = 0; const timer = setInterval(() => { - if (index >= chunks.length) { - clearInterval(timer); - res.end(); - return; - } - res.write(chunks[index]); - index += 1; - }, 20); + if (index >= chunks.length) { + clearInterval(timer); + res.end(); + return; + } + res.write(chunks[index]); + index += 1; + }, intervalMs); res.on("close", () => { clearInterval(timer); @@ -82,6 +82,7 @@ function createTerminatedSseResponse(res, chunks, destroyDelayMs = 20) { function startFakeUpstream(port) { const failBeforeResponseCounts = new Map(); + const capacityBeforeSuccessCounts = new Map(); const server = http.createServer((req, res) => { const responsePaths = new Set(["/responses", "/v1/responses"]); const chatCompletionPaths = new Set(["/chat/completions", "/v1/chat/completions"]); @@ -137,12 +138,31 @@ function startFakeUpstream(port) { ); return; } + if (parsed.test_capacity_before_success_times) { + const capacityKey = `${req.url}:capacity-before-success:${parsed.test_capacity_before_success_times}`; + const capacityCount = (capacityBeforeSuccessCounts.get(capacityKey) || 0) + 1; + capacityBeforeSuccessCounts.set(capacityKey, capacityCount); + if (capacityCount <= parsed.test_capacity_before_success_times) { + createJsonResponse( + res, + parsed.test_capacity_status ?? 503, + { + error: { + message: parsed.test_capacity_message || "Selected model is at capacity. Please try a different model.", + type: "server_error", + }, + }, + { "x-upstream-test": "capacity-error" }, + ); + return; + } + } if (parsed.stream) { createSseResponse(res, [ 'data: {"type":"response.output_text.delta","delta":"hello"}\n\n', `data: {"response":{"usage":{"output_tokens_details":{"reasoning_tokens":${reasoning}}}}}\n\n`, "data: [DONE]\n\n", - ]); + ], parsed.test_stream_chunk_delay_ms ?? 20); return; } createJsonResponse( @@ -283,19 +303,21 @@ async function run() { const logPath = path.join(tempRoot, "gateway.log"); const config = { - listen_host: "127.0.0.1", - listen_port: gatewayPort, - upstream_base_url: `http://127.0.0.1:${upstreamPort}`, - request_body_limit_bytes: 10 * 1024 * 1024, + listen_host: "127.0.0.1", + listen_port: gatewayPort, + upstream_base_url: `http://127.0.0.1:${upstreamPort}`, + request_body_limit_bytes: 10 * 1024 * 1024, endpoints: ["/responses", "/chat/completions", "/v1/responses", "/v1/chat/completions"], reasoning_equals: [516], retryable_status_codes: [429, 503], retryable_error_messages: ["Selected model is at capacity. Please try a different model."], + upstream_fetch_retry_attempts: 5, + upstream_fetch_retry_backoff_ms: 25, non_stream_status_code: 502, stream_action: "strict_502", log_match: true, - health_path: "/__codex_retry_gateway/health", - }; + health_path: "/__codex_retry_gateway/health", + }; await writeFile(configPath, JSON.stringify(config, null, 2), "utf8"); @@ -381,6 +403,25 @@ async function run() { "capacity error 返回体未保留 upstream status", ); + const capacityRecoveredResponse = await fetch(`http://127.0.0.1:${gatewayPort}/responses`, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ test_capacity_before_success_times: 2, test_reasoning_tokens: 128 }), + }); + const capacityRecoveredBody = await capacityRecoveredResponse.json(); + assert(capacityRecoveredResponse.status === 200, `capacity 抖动后未自动恢复: ${capacityRecoveredResponse.status}`); + assert( + capacityRecoveredBody?.usage?.output_tokens_details?.reasoning_tokens === 128, + "capacity 抖动恢复后的返回体异常", + ); + + const requestsAfterCapacityRecoveryResponse = await fetch(`http://127.0.0.1:${gatewayPort}/__codex_retry_gateway/api/requests?limit=20`); + const requestsAfterCapacityRecovery = await requestsAfterCapacityRecoveryResponse.json(); + const capacityRecoveredEntry = requestsAfterCapacityRecovery?.entries?.find( + (entry) => entry.path === "/responses" && entry.status_code === 200 && entry.upstream_attempt_count >= 3, + ); + assert(capacityRecoveredEntry, "capacity 抖动恢复后的请求记录未保留重试次数"); + const streamCapacityResponse = await fetch(`http://127.0.0.1:${gatewayPort}/responses`, { method: "POST", headers: { "content-type": "application/json" }, @@ -421,6 +462,36 @@ async function run() { assert(!okStream.closedByError, `${streamPath} 流式 128 不应异常断开`); } + const streamProgressPromise = fetch(`http://127.0.0.1:${gatewayPort}/responses`, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ stream: true, test_reasoning_tokens: 128, test_stream_chunk_delay_ms: 180 }), + }); + await new Promise((resolve) => setTimeout(resolve, 260)); + const midRequestsResponse = await fetch(`http://127.0.0.1:${gatewayPort}/__codex_retry_gateway/api/requests?limit=20`); + const midRequestsPayload = await midRequestsResponse.json(); + const inFlightStreamEntry = midRequestsPayload?.entries?.find( + (entry) => + entry.path === "/responses" && + entry.response_stream === true && + (entry.lifecycle_state === "streaming" || entry.lifecycle_state === "receive_first") && + (entry.stream_chunk_count || 0) >= 1, + ); + assert(inFlightStreamEntry, "流式请求过程中未暴露进行中状态"); + assert( + (inFlightStreamEntry?.response_bytes_received || 0) > 0, + "流式请求过程中未累计接收字节数", + ); + const streamProgressResponse = await streamProgressPromise; + assert(streamProgressResponse.status === 200, `stream progress 响应状态异常: ${streamProgressResponse.status}`); + const streamReader = streamProgressResponse.body.getReader(); + while (true) { + const { done } = await streamReader.read(); + if (done) { + break; + } + } + const terminatedStream = await readSseUntilClose( `http://127.0.0.1:${gatewayPort}/responses`, { stream: true, test_force_terminate: true }, diff --git a/ui-src/src/App.tsx b/ui-src/src/App.tsx index d091a51..eda95ea 100644 --- a/ui-src/src/App.tsx +++ b/ui-src/src/App.tsx @@ -17,6 +17,8 @@ type GatewayConfig = { reasoning_equals?: number[]; retryable_status_codes?: number[]; retryable_error_messages?: string[]; + upstream_fetch_retry_attempts?: number; + upstream_fetch_retry_backoff_ms?: number; non_stream_status_code?: number; log_match?: boolean; }; @@ -71,15 +73,20 @@ type RequestEntry = { first_response_at?: string | null; first_response_delay_ms?: number | null; finished_at?: string | null; + last_activity_at?: string | null; duration_ms?: number | null; profile_name?: string; method?: string; path?: string; request_body_bytes?: number | null; + response_bytes_received?: number | null; model?: string | null; requested_model?: string | null; forwarded_model?: string | null; response_stream?: boolean; + stream_chunk_count?: number | null; + usage_last_updated_at?: string | null; + upstream_attempt_count?: number | null; matched?: boolean; status_code?: number | null; upstream_status_code?: number | null; @@ -119,6 +126,8 @@ type ProfileFormModel = { reasoning_equals?: string; retryable_status_codes?: string; retryable_error_messages?: string[]; + upstream_fetch_retry_attempts?: string; + upstream_fetch_retry_backoff_ms?: string; endpoints?: string[]; }; @@ -186,6 +195,8 @@ type ProfileFormState = { reasoning_equals: string; retryable_status_codes: string; retryable_error_messages: string; + upstream_fetch_retry_attempts: string; + upstream_fetch_retry_backoff_ms: string; endpoints: string; }; @@ -193,6 +204,8 @@ type RuleFormState = { reasoning_equals: string; retryable_status_codes: string; retryable_error_messages: string; + upstream_fetch_retry_attempts: string; + upstream_fetch_retry_backoff_ms: string; endpoints: string; non_stream_status_code: string; log_match: boolean; @@ -274,6 +287,8 @@ const defaultProfileForm: ProfileFormState = { reasoning_equals: "516,1034,1552", retryable_status_codes: "429,503", retryable_error_messages: "Selected model is at capacity. Please try a different model.", + upstream_fetch_retry_attempts: "5", + upstream_fetch_retry_backoff_ms: "350", endpoints: "/responses\n/chat/completions\n/v1/responses\n/v1/chat/completions", }; @@ -385,6 +400,12 @@ function profileFormFromStatus(status: StatusPayload | null): ProfileFormState { retryable_error_messages: Array.isArray(config.retryable_error_messages) ? config.retryable_error_messages.join("\n") : defaultProfileForm.retryable_error_messages, + upstream_fetch_retry_attempts: String( + config.upstream_fetch_retry_attempts ?? defaultProfileForm.upstream_fetch_retry_attempts, + ), + upstream_fetch_retry_backoff_ms: String( + config.upstream_fetch_retry_backoff_ms ?? defaultProfileForm.upstream_fetch_retry_backoff_ms, + ), endpoints: Array.isArray(config.endpoints) ? config.endpoints.join("\n") : defaultProfileForm.endpoints, }; } @@ -412,6 +433,10 @@ function profileFormFromProfile(profile: Profile): ProfileFormState { retryable_error_messages: Array.isArray(form.retryable_error_messages) ? form.retryable_error_messages.join("\n") : defaultProfileForm.retryable_error_messages, + upstream_fetch_retry_attempts: + form.upstream_fetch_retry_attempts || defaultProfileForm.upstream_fetch_retry_attempts, + upstream_fetch_retry_backoff_ms: + form.upstream_fetch_retry_backoff_ms || defaultProfileForm.upstream_fetch_retry_backoff_ms, endpoints: Array.isArray(form.endpoints) ? form.endpoints.join("\n") : "", }; } @@ -426,6 +451,12 @@ function ruleFormFromStatus(status: StatusPayload | null): RuleFormState { retryable_error_messages: Array.isArray(config.retryable_error_messages) ? config.retryable_error_messages.join("\n") : defaultProfileForm.retryable_error_messages, + upstream_fetch_retry_attempts: String( + config.upstream_fetch_retry_attempts ?? defaultProfileForm.upstream_fetch_retry_attempts, + ), + upstream_fetch_retry_backoff_ms: String( + config.upstream_fetch_retry_backoff_ms ?? defaultProfileForm.upstream_fetch_retry_backoff_ms, + ), endpoints: Array.isArray(config.endpoints) ? config.endpoints.join("\n") : "", non_stream_status_code: String(config.non_stream_status_code || 502), log_match: Boolean(config.log_match), @@ -591,6 +622,8 @@ export default function App() { .map((value) => Number.parseInt(value, 10)) .filter((value) => Number.isInteger(value)), retryable_error_messages: splitLines(ruleForm.retryable_error_messages), + upstream_fetch_retry_attempts: Number.parseInt(ruleForm.upstream_fetch_retry_attempts, 10), + upstream_fetch_retry_backoff_ms: Number.parseInt(ruleForm.upstream_fetch_retry_backoff_ms, 10), endpoints: splitLines(ruleForm.endpoints), non_stream_status_code: Number.parseInt(ruleForm.non_stream_status_code, 10), log_match: ruleForm.log_match, @@ -639,6 +672,8 @@ export default function App() { reasoning_equals: splitList(profileForm.reasoning_equals), retryable_status_codes: splitList(profileForm.retryable_status_codes), retryable_error_messages: splitLines(profileForm.retryable_error_messages), + upstream_fetch_retry_attempts: Number.parseInt(profileForm.upstream_fetch_retry_attempts, 10), + upstream_fetch_retry_backoff_ms: Number.parseInt(profileForm.upstream_fetch_retry_backoff_ms, 10), endpoints: splitLines(profileForm.endpoints), }), }); @@ -944,10 +979,20 @@ export default function App() { 总 {durationSeconds(entry.duration_ms)} + + 更 + {timestamp(entry.last_activity_at || entry.usage_last_updated_at)} + 体 {bytesFormat(entry.request_body_bytes)} + + 流 + {entry.response_stream + ? `${numberFormat(entry.stream_chunk_count)} chunk / ${bytesFormat(entry.response_bytes_received)}` + : "-"} + 收 {timestamp(entry.finished_at)} @@ -982,6 +1027,10 @@ export default function App() { {upstream.auth_mode || "-"} / {upstream.auth_source || "-"} {upstream.authorization_configured === false ? " / no auth" : ""} + + 重试 {numberFormat(entry.upstream_attempt_count)} 次 + {entry.upstream_status_code ? ` / upstream ${entry.upstream_status_code}` : ""} +
@@ -1144,6 +1193,14 @@ export default function App() {