diff --git a/README.md b/README.md index 9f72edf..422344a 100644 --- a/README.md +++ b/README.md @@ -9,7 +9,8 @@ tg群:https://t.me/AI_INPUT_IM - 保持 Codex 继续使用现有 `auth.json` - 只把 `config.toml` 的当前 provider `base_url` 改成本地网关 - 非流式命中 `reasoning_tokens = 516` 时返回 `502` -- 上游若返回明确的容量错误（默认匹配 `429/503` 且错误文案包含 `Selected model is at capacity. Please try a different model.`），也会转成本地 `502` +- 上游若返回明确的容量错误（默认匹配错误文案 `Selected model is at capacity. Please try a different model.`），也会自动重试；重试耗尽后转成本地 `502` +- 除了 `429/503` JSON 错误响应，也会识别 `200` 但返回体本质是错误、以及流式失败事件里携带同样文案的情况 - 流式命中时默认先缓存并判断；一旦命中 `516`，统一返回 `502` - 默认同时拦截 root 路径和 `/v1` 路径： - `/responses` diff --git a/gateway.mjs b/gateway.mjs index e0fe262..cad1eb6 100644 --- a/gateway.mjs +++ b/gateway.mjs @@ -2419,10 +2419,6 @@ function truncateRetryableMessage(value, maxLength = 280) { } function findRetryableUpstreamErrorMatch(config, upstreamStatusCode, parsedBody, bodyText) { - if (!isUpstreamErrorStatus(upstreamStatusCode)) { - return null; - } - const retryableMessages = normalizePhraseList( config.retryable_error_messages, DEFAULT_CONFIG.retryable_error_messages, @@ -2435,7 +2431,11 @@ function findRetryableUpstreamErrorMatch(config, upstreamStatusCode, parsedBody, config.retryable_status_codes, DEFAULT_CONFIG.retryable_status_codes, ); - if (retryableStatusCodes.length > 0 && !retryableStatusCodes.includes(upstreamStatusCode)) { + const statusEligible = isUpstreamErrorStatus(upstreamStatusCode) && ( + retryableStatusCodes.length === 0 || + retryableStatusCodes.includes(upstreamStatusCode) + ); + if (!statusEligible && !isRetryableErrorPayloadShape(parsedBody)) { return null; } @@ -2480,27 +2480,64 @@ function matchRetryableMessage(config, parsedBody, bodyText) { return null; } -function isRetryableStreamErrorShape(parsedBody, eventName = "") { - if (`${eventName || ""}`.trim().toLowerCase() === "error") { +function hasRetryableFailureKeyword(value) { + const normalized = `${value || ""}`.trim().toLowerCase(); + if (!normalized) { + return false; + } + return ( + normalized.includes("error") || + normalized.includes("failed") || + normalized.includes("failure") || + normalized.includes("cancelled") || + normalized.includes("canceled") || + normalized.includes("rate_limit") || + normalized.includes("overloaded") || + normalized.includes("unavailable") + ); +} + +function hasRetryableFailureShape(value, depth = 0) { + if (!value || depth > 4) { + return false; + } + if (Array.isArray(value)) { + return value.some((item) => hasRetryableFailureShape(item, depth + 1)); + } + if (typeof value !== "object") { + return false; + } + if (Object.hasOwn(value, "error")) { + return true; + } + if (Number.isInteger(value.status) && value.status >= 400) { + return true; + } + for (const key of ["type", "event", "status", "state", "result", "code"]) { + if (typeof value[key] === "string" && hasRetryableFailureKeyword(value[key])) { + return true; + } + } + for (const key of ["response", "data", "meta", "details", "detail"]) { + if (hasRetryableFailureShape(value[key], depth + 1)) { + return true; + } + } + return false; +} + +function isRetryableErrorPayloadShape(parsedBody, eventName = "") { + if (hasRetryableFailureKeyword(eventName)) { 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; + return hasRetryableFailureShape(parsedBody); } function findRetryableStreamErrorMatch(config, parsedBody, bodyText, eventName = "") { - if (!isRetryableStreamErrorShape(parsedBody, eventName)) { + if (!isRetryableErrorPayloadShape(parsedBody, eventName)) { return null; } return matchRetryableMessage(config, parsedBody, bodyText); @@ -2564,14 +2601,6 @@ async function fetchUpstreamWithRetry(upstreamUrl, init, config, logger, request }; } - if (!isUpstreamErrorStatus(response.status)) { - return { - response, - attempt_count: attempt, - retryable_upstream_error: null, - }; - } - const contentType = response.headers.get("content-type"); if (!isJsonContentType(contentType)) { return { diff --git a/scripts/test-gateway-e2e.mjs b/scripts/test-gateway-e2e.mjs index 473d2b4..9a67a18 100644 --- a/scripts/test-gateway-e2e.mjs +++ b/scripts/test-gateway-e2e.mjs @@ -80,16 +80,40 @@ function createTerminatedSseResponse(res, chunks, destroyDelayMs = 20) { }, destroyDelayMs); } +function buildCapacityStreamPayload(message, shape = "default") { + if (shape === "response_failed") { + return { + type: "response.failed", + response: { + status: "failed", + error: { + message, + type: "server_error", + }, + }, + }; + } + return { + error: { + message, + type: "server_error", + }, + }; +} + function createCapacityErrorSseResponse( res, message = "Selected model is at capacity. Please try a different model.", intervalMs = 20, + options = {}, ) { + const eventName = options.eventName || "error"; + const payload = options.payload || buildCapacityStreamPayload(message, options.payloadShape || "default"); createSseResponse( res, [ - 'event: error\n', - `data: ${JSON.stringify({ error: { message, type: "server_error" } })}\n\n`, + `event: ${eventName}\n`, + `data: ${JSON.stringify(payload)}\n\n`, ], intervalMs, ); @@ -140,28 +164,43 @@ function startFakeUpstream(port) { return; } if (parsed.test_capacity_error) { + const capacityMessage = parsed.test_capacity_message || "Selected model is at capacity. Please try a different model."; createJsonResponse( res, parsed.test_capacity_status ?? 503, { error: { - message: parsed.test_capacity_message || "Selected model is at capacity. Please try a different model.", + message: capacityMessage, type: "server_error", }, }, - { "x-upstream-test": "capacity-error" }, + { "x-upstream-test": `capacity-error-${parsed.test_capacity_status ?? 503}` }, ); return; } if (parsed.test_capacity_before_success_times) { - const capacityKey = `${req.url}:capacity-before-success:${parsed.test_capacity_before_success_times}`; + const capacityKey = [ + req.url, + "capacity-before-success", + parsed.test_capacity_before_success_times, + parsed.stream ? "stream" : "non-stream", + parsed.test_capacity_status ?? "default", + parsed.test_capacity_stream_event_name || "", + parsed.test_capacity_stream_payload_shape || "", + ].join(":"); const capacityCount = (capacityBeforeSuccessCounts.get(capacityKey) || 0) + 1; capacityBeforeSuccessCounts.set(capacityKey, capacityCount); if (capacityCount <= parsed.test_capacity_before_success_times) { + const capacityMessage = parsed.test_capacity_message || "Selected model is at capacity. Please try a different model."; if (parsed.stream) { createCapacityErrorSseResponse( res, - parsed.test_capacity_message || "Selected model is at capacity. Please try a different model.", + capacityMessage, + 20, + { + eventName: parsed.test_capacity_stream_event_name || "error", + payloadShape: parsed.test_capacity_stream_payload_shape || "default", + }, ); return; } @@ -170,11 +209,11 @@ function startFakeUpstream(port) { parsed.test_capacity_status ?? 503, { error: { - message: parsed.test_capacity_message || "Selected model is at capacity. Please try a different model.", + message: capacityMessage, type: "server_error", }, }, - { "x-upstream-test": "capacity-error" }, + { "x-upstream-test": `capacity-error-${parsed.test_capacity_status ?? 503}` }, ); return; } @@ -183,6 +222,11 @@ function startFakeUpstream(port) { createCapacityErrorSseResponse( res, parsed.test_capacity_message || "Selected model is at capacity. Please try a different model.", + 20, + { + eventName: parsed.test_capacity_stream_event_name || "error", + payloadShape: parsed.test_capacity_stream_payload_shape || "default", + }, ); return; } @@ -432,6 +476,22 @@ async function run() { "capacity error 返回体未保留 upstream status", ); + const capacityStatus200Response = await fetch(`http://127.0.0.1:${gatewayPort}/responses`, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ test_capacity_error: true, test_capacity_status: 200 }), + }); + const capacityStatus200Body = await capacityStatus200Response.json(); + assert(capacityStatus200Response.status === 502, `200+capacity error 未返回 502: ${capacityStatus200Response.status}`); + assert( + capacityStatus200Body?.error?.code === "upstream_error_retry_triggered", + "200+capacity error 返回体未标记 retry trigger", + ); + assert( + capacityStatus200Body?.error?.upstream_status_code === 200, + "200+capacity error 返回体未保留 upstream status", + ); + const capacityRecoveredResponse = await fetch(`http://127.0.0.1:${gatewayPort}/responses`, { method: "POST", headers: { "content-type": "application/json" }, @@ -451,6 +511,25 @@ async function run() { ); assert(capacityRecoveredEntry, "capacity 抖动恢复后的请求记录未保留重试次数"); + const capacityStatus200RecoveredResponse = 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_capacity_status: 200, test_reasoning_tokens: 128 }), + }); + const capacityStatus200RecoveredBody = await capacityStatus200RecoveredResponse.json(); + assert(capacityStatus200RecoveredResponse.status === 200, `200+capacity 抖动后未自动恢复: ${capacityStatus200RecoveredResponse.status}`); + assert( + capacityStatus200RecoveredBody?.usage?.output_tokens_details?.reasoning_tokens === 128, + "200+capacity 抖动恢复后的返回体异常", + ); + + const requestsAfterCapacityStatus200RecoveryResponse = await fetch(`http://127.0.0.1:${gatewayPort}/__codex_retry_gateway/api/requests?limit=30`); + const requestsAfterCapacityStatus200Recovery = await requestsAfterCapacityStatus200RecoveryResponse.json(); + const capacityStatus200RecoveredEntry = requestsAfterCapacityStatus200Recovery?.entries?.find( + (entry) => entry.path === "/responses" && entry.status_code === 200 && entry.upstream_attempt_count >= 3, + ); + assert(capacityStatus200RecoveredEntry, "200+capacity 抖动恢复后的请求记录未保留重试次数"); + const streamCapacityResponse = await fetch(`http://127.0.0.1:${gatewayPort}/responses`, { method: "POST", headers: { "content-type": "application/json" }, @@ -463,6 +542,23 @@ async function run() { "stream+capacity error 返回体未标记 retry trigger", ); + const streamCapacityResponseFailed = await fetch(`http://127.0.0.1:${gatewayPort}/responses`, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + stream: true, + test_capacity_error: true, + test_capacity_stream_event_name: "response.failed", + test_capacity_stream_payload_shape: "response_failed", + }), + }); + const streamCapacityResponseFailedBody = await streamCapacityResponseFailed.json(); + assert(streamCapacityResponseFailed.status === 502, `stream+response.failed capacity error 未返回 502: ${streamCapacityResponseFailed.status}`); + assert( + streamCapacityResponseFailedBody?.error?.code === "upstream_error_retry_triggered", + "stream+response.failed capacity error 返回体未标记 retry trigger", + ); + const streamCapacityRecoveredResponse = await fetch(`http://127.0.0.1:${gatewayPort}/responses`, { method: "POST", headers: { "content-type": "application/json" }, @@ -479,6 +575,28 @@ async function run() { ); assert(streamCapacityRecoveredEntry, "stream capacity 抖动恢复后的请求记录未保留重试次数"); + const streamCapacityResponseFailedRecoveredResponse = 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, + test_capacity_stream_event_name: "response.failed", + test_capacity_stream_payload_shape: "response_failed", + }), + }); + const streamCapacityResponseFailedRecoveredText = await streamCapacityResponseFailedRecoveredResponse.text(); + assert(streamCapacityResponseFailedRecoveredResponse.status === 200, `stream response.failed capacity 抖动后未自动恢复: ${streamCapacityResponseFailedRecoveredResponse.status}`); + assert(streamCapacityResponseFailedRecoveredText.includes("hello"), "stream response.failed capacity 恢复后未拿到正常 SSE 内容"); + + const requestsAfterStreamCapacityResponseFailedRecoveryResponse = await fetch(`http://127.0.0.1:${gatewayPort}/__codex_retry_gateway/api/requests?limit=30`); + const requestsAfterStreamCapacityResponseFailedRecovery = await requestsAfterStreamCapacityResponseFailedRecoveryResponse.json(); + const streamCapacityResponseFailedRecoveredEntry = requestsAfterStreamCapacityResponseFailedRecovery?.entries?.find( + (entry) => entry.path === "/responses" && entry.status_code === 200 && entry.response_stream && entry.upstream_attempt_count >= 3, + ); + assert(streamCapacityResponseFailedRecoveredEntry, "stream response.failed capacity 恢复后的请求记录未保留重试次数"); + for (const streamPath of [ "/responses", "/v1/responses",