shujakuin/codex-retry-gateway
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: retry Codex overload error codes

Browse Source
This commit is contained in:
yunyaozhou
2026-07-11 13:44:12 +08:00
parent 29eb91e0c3
commit bf01fcd727
3 changed files with 132 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+1 -1
View File
@@ -21,7 +21,7 @@
- 保持 Codex 继续使用现有 `auth.json`
- 只把 `config.toml` 的当前 provider `base_url` 改成本地网关
- 非流式命中 `reasoning_tokens = 516` 时返回 `502`
- 上游若返回明确的容量错误（默认匹配错误文案 `Selected model is at capacity. Please try a different model.`，以及 `stream disconnected before completion: Concurrency limit exceeded for account, please retry later`），也会自动重试；重试耗尽后转成本地 `502`
- 上游若返回明确的容量错误（默认匹配错误文案 `Selected model is at capacity. Please try a different model.`、`stream disconnected before completion: Concurrency limit exceeded for account, please retry later`，以及 Responses 错误码 `server_is_overloaded` / `slow_down`），也会自动重试；重试耗尽后转成本地 `502`
- 除了 `429/503` JSON 错误响应，也会识别 `200` 但返回体本质是错误、以及流式失败事件里携带同样文案的情况
- 流式命中时默认先缓存并判断；一旦命中 `516`，统一返回 `502`
- 流式成功响应在严格检查模式下仍会先缓存完成；成功后会按真实且规范化的 Responses 生命周期与输出顺序逐块回放给 Codex，每个 SSE 块之间至少间隔 5ms，并关闭 TCP 小包聚合，避免大量 delta 在同一事件循环突发到达；不会伪造缺少 response ID 的生命周期事件
gateway.mjs
+52 -1
View File
@@ -4403,6 +4403,47 @@ function truncateRetryableMessage(value, maxLength = 280) {
return text.length > maxLength ? `${text.slice(0, maxLength - 3)}...` : text;
}
const RETRYABLE_OVERLOAD_ERROR_CODES = new Set([
"server_is_overloaded",
"slow_down",
]);
function findRetryableOverloadErrorCode(value, state = { seen: new Set() }, depth = 0) {
if (!value || typeof value !== "object" || depth > 6 || state.seen.has(value)) {
return null;
}
state.seen.add(value);
if (Array.isArray(value)) {
for (const item of value) {
const match = findRetryableOverloadErrorCode(item, state, depth + 1);
if (match) {
return match;
}
}
return null;
}
const code = typeof value.code === "string" ? value.code.trim().toLowerCase() : "";
if (RETRYABLE_OVERLOAD_ERROR_CODES.has(code)) {
return {
matched_code: code,
matched_pattern: `error_code:${code}`,
matched_message: truncateRetryableMessage(
firstNonEmptyString(value.message, value.detail, value.description, code),
),
};
}
for (const nested of Object.values(value)) {
const match = findRetryableOverloadErrorCode(nested, state, depth + 1);
if (match) {
return match;
}
}
return null;
}
function findRetryableUpstreamErrorMatch(config, upstreamStatusCode, parsedBody, bodyText) {
const retryableMessages = normalizePhraseList(
config.retryable_error_messages,
@@ -4420,10 +4461,16 @@ function findRetryableUpstreamErrorMatch(config, upstreamStatusCode, parsedBody,
retryableStatusCodes.length === 0 ||
retryableStatusCodes.includes(upstreamStatusCode)
);
if (!statusEligible && !isRetryableErrorPayloadShape(parsedBody)) {
const failurePayload = isRetryableErrorPayloadShape(parsedBody);
if (!statusEligible && !failurePayload) {
return null;
}
const overloadCodeMatch = findRetryableOverloadErrorCode(parsedBody);
if (overloadCodeMatch) {
return overloadCodeMatch;
}
return matchRetryableMessage(config, parsedBody, bodyText);
}
@@ -4525,6 +4572,10 @@ function findRetryableStreamErrorMatch(config, parsedBody, bodyText, eventName =
if (!isRetryableErrorPayloadShape(parsedBody, eventName)) {
return null;
}
const overloadCodeMatch = findRetryableOverloadErrorCode(parsedBody);
if (overloadCodeMatch) {
return overloadCodeMatch;
}
return matchRetryableMessage(config, parsedBody, bodyText);
}
scripts/test-gateway-e2e.mjs
+79 -13
View File
@@ -88,25 +88,22 @@ function createTerminatedSseResponse(res, chunks, destroyDelayMs = 20, options =
}, destroyDelayMs);
}
function buildCapacityStreamPayload(message, shape = "default") {
function buildCapacityStreamPayload(message, shape = "default", errorCode = "") {
const error = {
message,
type: "server_error",
...(errorCode ? { code: errorCode } : {}),
};
if (shape === "response_failed") {
return {
type: "response.failed",
response: {
status: "failed",
error: {
message,
type: "server_error",
},
error,
},
};
}
return {
error: {
message,
type: "server_error",
},
};
return { error };
}
function createCapacityErrorSseResponse(
@@ -116,7 +113,11 @@ function createCapacityErrorSseResponse(
options = {},
) {
const eventName = options.eventName || "error";
const payload = options.payload || buildCapacityStreamPayload(message, options.payloadShape || "default");
const payload = options.payload || buildCapacityStreamPayload(
message,
options.payloadShape || "default",
options.errorCode || "",
);
createSseResponse(
res,
[
@@ -231,7 +232,7 @@ function startFakeUpstream(port, options = {}) {
]);
return;
}
if (parsed.test_capacity_error) {
if (parsed.test_capacity_error && !parsed.stream) {
const capacityMessage = parsed.test_capacity_message || "Selected model is at capacity. Please try a different model.";
createJsonResponse(
res,
@@ -240,6 +241,7 @@ function startFakeUpstream(port, options = {}) {
error: {
message: capacityMessage,
type: "server_error",
...(parsed.test_capacity_code ? { code: parsed.test_capacity_code } : {}),
},
},
{ "x-upstream-test": `capacity-error-${parsed.test_capacity_status ?? 503}` },
@@ -255,6 +257,8 @@ function startFakeUpstream(port, options = {}) {
parsed.test_capacity_status ?? "default",
parsed.test_capacity_stream_event_name || "",
parsed.test_capacity_stream_payload_shape || "",
parsed.test_capacity_code || "",
parsed.test_capacity_message || "",
].join(":");
const capacityCount = (capacityBeforeSuccessCounts.get(capacityKey) || 0) + 1;
capacityBeforeSuccessCounts.set(capacityKey, capacityCount);
@@ -268,6 +272,7 @@ function startFakeUpstream(port, options = {}) {
{
eventName: parsed.test_capacity_stream_event_name || "error",
payloadShape: parsed.test_capacity_stream_payload_shape || "default",
errorCode: parsed.test_capacity_code || "",
},
);
return;
@@ -279,6 +284,7 @@ function startFakeUpstream(port, options = {}) {
error: {
message: capacityMessage,
type: "server_error",
...(parsed.test_capacity_code ? { code: parsed.test_capacity_code } : {}),
},
},
{ "x-upstream-test": `capacity-error-${parsed.test_capacity_status ?? 503}` },
@@ -294,6 +300,7 @@ function startFakeUpstream(port, options = {}) {
{
eventName: parsed.test_capacity_stream_event_name || "error",
payloadShape: parsed.test_capacity_stream_payload_shape || "default",
errorCode: parsed.test_capacity_code || "",
},
);
return;
@@ -1482,6 +1489,28 @@ async function run() {
"stream+response.failed capacity error 返回体未标记 retry trigger",
);
const streamSlowDownCodeResponse = 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_message: "Raw provider asked the client to wait.",
test_capacity_code: "slow_down",
test_capacity_stream_event_name: "response.failed",
test_capacity_stream_payload_shape: "response_failed",
}),
});
const streamSlowDownCodeBody = await streamSlowDownCodeResponse.json();
assert(
streamSlowDownCodeResponse.status === 502,
`stream+slow_down 持续过载未返回 502: ${streamSlowDownCodeResponse.status}`,
);
assert(
streamSlowDownCodeBody?.error?.code === "upstream_error_retry_triggered",
"stream+slow_down 持续过载未标记 retry trigger",
);
const streamCapacityRecoveredResponse = await fetch(`http://127.0.0.1:${gatewayPort}/responses`, {
method: "POST",
headers: { "content-type": "application/json" },
@@ -1520,6 +1549,43 @@ async function run() {
);
assert(streamCapacityResponseFailedRecoveredEntry, "stream response.failed capacity 恢复后的请求记录未保留重试次数");
const overloadCodeThreadId = "thread-server-overloaded-code-retry";
const streamOverloadCodeRecoveredResponse = await readSseUntilClose(
`http://127.0.0.1:${gatewayPort}/responses`,
{
stream: true,
thread_id: overloadCodeThreadId,
test_capacity_before_success_times: 2,
test_capacity_message: "Raw provider reported a temporary overload.",
test_capacity_code: "server_is_overloaded",
test_capacity_stream_event_name: "response.failed",
test_capacity_stream_payload_shape: "response_failed",
test_reasoning_tokens: 128,
},
);
assert(
streamOverloadCodeRecoveredResponse.status === 200,
`stream+server_is_overloaded 未自动恢复: ${streamOverloadCodeRecoveredResponse.status}`,
);
assert(
streamOverloadCodeRecoveredResponse.text.includes("hello"),
"stream+server_is_overloaded 恢复后未拿到正常 SSE 内容",
);
assert(
!streamOverloadCodeRecoveredResponse.text.includes("Raw provider reported a temporary overload."),
"stream+server_is_overloaded 自动恢复前不应向客户端泄漏失败轮次",
);
const overloadCodeRequestsResponse = await fetch(
`http://127.0.0.1:${gatewayPort}/__codex_retry_gateway/api/requests?query=${encodeURIComponent(overloadCodeThreadId)}`,
{ headers: adminHeaders },
);
const overloadCodeRequestsPayload = await overloadCodeRequestsResponse.json();
const overloadCodeEntry = (overloadCodeRequestsPayload?.entries || []).find(
(entry) => entry.thread_id === overloadCodeThreadId,
);
assert(overloadCodeEntry?.status_code === 200, "server_is_overloaded 恢复请求未记录最终 200");
assert(overloadCodeEntry?.upstream_attempt_count === 3, "server_is_overloaded 恢复请求未记录 3 次上游尝试");
const streamThreadResponse = await readSseUntilClose(
`http://127.0.0.1:${gatewayPort}/responses`,
{ stream: true, test_reasoning_tokens: 128, thread_id: "thread_stream_ok" },