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

harden upstream retry handling

Browse Source
This commit is contained in:
yunyaozhou
2026-06-29 18:04:53 +08:00
parent 94bece3cf2
commit 57c7264c7b
5 changed files with 465 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gateway.mjs
+293 -13
View File
@@ -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, {