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

retry capacity-style upstream errors

Browse Source
This commit is contained in:
shujakuin
2026-06-29 15:54:36 +08:00
parent 778068af42
commit 283c14b6b4
8 changed files with 384 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gateway.mjs
+233 -4
View File
@@ -39,6 +39,10 @@ const DEFAULT_CONFIG = {
model_remap: "",
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.",
],
non_stream_status_code: 502,
stream_action: "strict_502",
log_match: true,
@@ -223,6 +227,21 @@ function normalizeStringList(values, fallback = []) {
return [...new Set(normalized)];
}
function normalizePhraseList(values, fallback = []) {
const source = values === undefined || values === null ? fallback : values;
const normalized = flattenValues(source)
.flatMap((value) => {
if (typeof value === "string") {
return value.split(/\r?\n/);
}
return [value];
})
.map((value) => `${value ?? ""}`.trim())
.filter(Boolean);
return [...new Set(normalized)];
}
function parseModelRemapMap(value) {
const map = {};
for (const rawEntry of `${value || ""}`.split(/\r?\n|[;,]/)) {
@@ -315,6 +334,19 @@ function buildBlockedBody(pathname, reasoning, statusCode) {
});
}
function buildRetryableUpstreamErrorBody(pathname, upstreamStatusCode, upstreamMessage, statusCode) {
return JSON.stringify({
error: {
message: `codex retry gateway converted retryable upstream error on ${pathname}`,
type: "codex_retry_gateway",
code: "upstream_error_retry_triggered",
upstream_status_code: upstreamStatusCode,
upstream_error_message: upstreamMessage,
status_code: statusCode,
},
});
}
function buildGatewayErrorBody(message) {
return JSON.stringify({
error: {
@@ -724,6 +756,11 @@ function buildProfileFormModel(env) {
request_history_limit: env.CODEX_RETRY_GATEWAY_REQUEST_HISTORY_LIMIT || `${DEFAULT_CONFIG.request_history_limit}`,
model_remap: env.CODEX_RETRY_GATEWAY_MODEL_REMAP || "",
reasoning_equals: env.CODEX_RETRY_GATEWAY_REASONING_EQUALS || "",
retryable_status_codes: env.CODEX_RETRY_GATEWAY_RETRYABLE_STATUS_CODES || "",
retryable_error_messages: normalizePhraseList(
env.CODEX_RETRY_GATEWAY_RETRYABLE_ERROR_MESSAGES || DEFAULT_CONFIG.retryable_error_messages,
DEFAULT_CONFIG.retryable_error_messages,
),
endpoints: normalizeStringList(env.CODEX_RETRY_GATEWAY_ENDPOINTS || DEFAULT_CONFIG.endpoints, DEFAULT_CONFIG.endpoints),
};
}
@@ -751,6 +788,14 @@ function buildConfigFromProfileEnv(profileName, env) {
model_remap: env.CODEX_RETRY_GATEWAY_MODEL_REMAP || "",
endpoints: normalizeStringList(env.CODEX_RETRY_GATEWAY_ENDPOINTS || DEFAULT_CONFIG.endpoints, DEFAULT_CONFIG.endpoints).map(normalizePath),
reasoning_equals: normalizeIntegerList(env.CODEX_RETRY_GATEWAY_REASONING_EQUALS || DEFAULT_CONFIG.reasoning_equals, DEFAULT_CONFIG.reasoning_equals),
retryable_status_codes: normalizeIntegerList(
env.CODEX_RETRY_GATEWAY_RETRYABLE_STATUS_CODES || DEFAULT_CONFIG.retryable_status_codes,
DEFAULT_CONFIG.retryable_status_codes,
),
retryable_error_messages: normalizePhraseList(
env.CODEX_RETRY_GATEWAY_RETRYABLE_ERROR_MESSAGES || DEFAULT_CONFIG.retryable_error_messages,
DEFAULT_CONFIG.retryable_error_messages,
),
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,
@@ -847,12 +892,29 @@ async function buildProfileEnvText(payload) {
throw new Error("History Limit 必须是 0 或正整数；0 表示不裁剪");
}
const retryableStatusCodes = normalizeIntegerList(
payload.retryable_status_codes,
DEFAULT_CONFIG.retryable_status_codes,
);
const retryableErrorMessages = normalizePhraseList(
payload.retryable_error_messages,
DEFAULT_CONFIG.retryable_error_messages,
);
if (retryableStatusCodes.length === 0) {
throw new Error("retryable_status_codes 不能为空");
}
if (retryableErrorMessages.length === 0) {
throw new Error("retryable_error_messages 不能为空");
}
const envPairs = [
["CODEX_RETRY_GATEWAY_LISTEN_HOST", `${payload.listen_host || DEFAULT_CONFIG.listen_host}`.trim()],
["CODEX_RETRY_GATEWAY_LISTEN_PORT", `${listenPort}`],
["CODEX_RETRY_GATEWAY_UPSTREAM_BASE_URL", upstreamBaseUrl],
["CODEX_RETRY_GATEWAY_UPSTREAM_AUTH_MODE", authMode],
["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_REQUEST_HISTORY_LIMIT", `${requestHistoryLimit}`],
["CODEX_RETRY_GATEWAY_ENDPOINTS", endpoints.join(",")],
];
@@ -1075,6 +1137,14 @@ async function loadConfig(configPath) {
config.reasoning_equals,
DEFAULT_CONFIG.reasoning_equals,
);
config.retryable_status_codes = normalizeIntegerList(
config.retryable_status_codes,
DEFAULT_CONFIG.retryable_status_codes,
);
config.retryable_error_messages = normalizePhraseList(
config.retryable_error_messages,
DEFAULT_CONFIG.retryable_error_messages,
);
if (!config.upstream_base_url) {
throw new Error("配置缺少 upstream_base_url");
}
@@ -1602,6 +1672,14 @@ async function serveManagementUi(res, requestPathname) {
function buildEditableConfig(currentConfig, payload) {
const nextReasoning = normalizeIntegerList(payload.reasoning_equals, currentConfig.reasoning_equals);
const nextRetryableStatusCodes = normalizeIntegerList(
payload.retryable_status_codes,
currentConfig.retryable_status_codes,
);
const nextRetryableErrorMessages = normalizePhraseList(
payload.retryable_error_messages,
currentConfig.retryable_error_messages,
);
const nextEndpoints = normalizeStringList(payload.endpoints, currentConfig.endpoints).map(normalizePath);
const nextStatusCode =
payload.non_stream_status_code === undefined
@@ -1611,6 +1689,12 @@ function buildEditableConfig(currentConfig, payload) {
if (nextReasoning.length === 0) {
throw new Error("reasoning_equals 不能为空");
}
if (nextRetryableStatusCodes.length === 0) {
throw new Error("retryable_status_codes 不能为空");
}
if (nextRetryableErrorMessages.length === 0) {
throw new Error("retryable_error_messages 不能为空");
}
if (nextEndpoints.length === 0) {
throw new Error("endpoints 不能为空");
}
@@ -1621,6 +1705,8 @@ function buildEditableConfig(currentConfig, payload) {
return {
...currentConfig,
reasoning_equals: nextReasoning,
retryable_status_codes: nextRetryableStatusCodes,
retryable_error_messages: nextRetryableErrorMessages,
endpoints: nextEndpoints,
non_stream_status_code: nextStatusCode,
log_match: payload.log_match === undefined ? currentConfig.log_match : Boolean(payload.log_match),
@@ -1822,7 +1908,7 @@ async function handleManagementRequest(runtime, req, res, requestUrl) {
await writeConfig(runtime.configPath, nextConfig);
runtime.config = nextConfig;
runtime.logger(
`[config] updated reasoning_equals=${nextConfig.reasoning_equals.join(",")} endpoints=${nextConfig.endpoints.join(",")}`,
`[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, {
@@ -1996,6 +2082,109 @@ function reasoningMatched(config, reasoning) {
return reasoning !== null && config.reasoning_equals.includes(reasoning);
}
function collectRetryableMessageCandidates(value, state = { seen: new Set(), results: [] }, depth = 0) {
if (value === null || value === undefined || depth > 5 || state.results.length >= 64) {
return state.results;
}
if (typeof value === "string") {
const trimmed = value.trim();
if (trimmed) {
state.results.push(trimmed);
}
return state.results;
}
if (typeof value !== "object") {
return state.results;
}
if (state.seen.has(value)) {
return state.results;
}
state.seen.add(value);
if (Array.isArray(value)) {
for (const item of value) {
collectRetryableMessageCandidates(item, state, depth + 1);
}
return state.results;
}
const preferredKeys = ["message", "error", "detail", "details", "description", "title"];
for (const key of preferredKeys) {
if (Object.hasOwn(value, key)) {
collectRetryableMessageCandidates(value[key], state, depth + 1);
}
}
for (const [key, nested] of Object.entries(value)) {
if (preferredKeys.includes(key)) {
continue;
}
collectRetryableMessageCandidates(nested, state, depth + 1);
}
return state.results;
}
function truncateRetryableMessage(value, maxLength = 280) {
const text = `${value || ""}`.trim();
if (!text) {
return "";
}
return text.length > maxLength ? `${text.slice(0, maxLength - 3)}...` : text;
}
function findRetryableUpstreamErrorMatch(config, upstreamStatusCode, parsedBody, bodyText) {
if (!isUpstreamErrorStatus(upstreamStatusCode)) {
return null;
}
const retryableMessages = normalizePhraseList(
config.retryable_error_messages,
DEFAULT_CONFIG.retryable_error_messages,
);
if (retryableMessages.length === 0) {
return null;
}
const retryableStatusCodes = normalizeIntegerList(
config.retryable_status_codes,
DEFAULT_CONFIG.retryable_status_codes,
);
if (retryableStatusCodes.length > 0 && !retryableStatusCodes.includes(upstreamStatusCode)) {
return null;
}
const normalizedPatterns = retryableMessages.map((message) => ({
original: message,
normalized: message.toLowerCase(),
}));
const candidateMessages = [];
if (parsedBody) {
candidateMessages.push(...collectRetryableMessageCandidates(parsedBody));
}
const trimmedBodyText = `${bodyText || ""}`.trim();
if (trimmedBodyText) {
candidateMessages.push(trimmedBodyText);
}
for (const candidate of candidateMessages) {
const normalizedCandidate = candidate.toLowerCase();
for (const pattern of normalizedPatterns) {
if (normalizedCandidate.includes(pattern.normalized)) {
return {
matched_pattern: pattern.original,
matched_message: truncateRetryableMessage(candidate),
};
}
}
}
return null;
}
function isExpectedStreamTermination(error) {
if (!error) {
return false;
@@ -2090,14 +2279,21 @@ async function handleNonStreaming({
}) {
markAndPersistFirstResponse(runtime, requestEntry);
const bodyBuffer = Buffer.from(await upstreamResponse.arrayBuffer());
const bodyText = bodyBuffer.toString("utf8");
const parsed = isJsonContentType(upstreamResponse.headers.get("content-type"))
? parseJsonSafely(bodyBuffer)
: null;
const reasoning = parsed ? extractReasoningTokens(parsed) : null;
const usage = parsed ? normalizeUsageSnapshot(parsed) : null;
const matched = reasoningMatched(config, reasoning);
const retryableUpstreamError = findRetryableUpstreamErrorMatch(
config,
upstreamResponse.status,
parsed,
bodyText,
);
recordInspectedResponse(monitor, reasoning, matched);
recordInspectedResponse(monitor, reasoning, matched || Boolean(retryableUpstreamError));
if (matched) {
if (config.log_match) {
@@ -2121,6 +2317,35 @@ async function handleNonStreaming({
};
}
if (retryableUpstreamError) {
if (config.log_match) {
logger(
`[match] non-stream path=${pathname} upstream_status=${upstreamResponse.status} retryable_error=${JSON.stringify(retryableUpstreamError.matched_pattern)} action=status_${config.non_stream_status_code}`,
);
}
const blockedBody = buildRetryableUpstreamErrorBody(
pathname,
upstreamResponse.status,
retryableUpstreamError.matched_message || retryableUpstreamError.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);
return {
inspected: true,
matched: true,
status_code: config.non_stream_status_code,
upstream_status_code: upstreamResponse.status,
reasoning_tokens: reasoning,
usage,
error: `retryable upstream error: ${retryableUpstreamError.matched_pattern}`,
match_reason: "retryable_upstream_error",
};
}
copyHeadersToClient(upstreamResponse.headers, res);
res.writeHead(upstreamResponse.status);
res.end(bodyBuffer);
@@ -2347,8 +2572,12 @@ async function proxyRequest(runtime, req, res) {
}, logger);
const shouldInspect = matchPath(config, pathname);
const responseIsStream =
requestIsStream || isSseContentType(upstreamResponse.headers.get("content-type"));
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;