From 1dac463af33ea429cd5d04c4dbe49ddb6d537e30 Mon Sep 17 00:00:00 2001 From: zyycn Date: Mon, 15 Jun 2026 09:30:17 +0800 Subject: [PATCH] codex: align fallback exhaustion errors Co-authored-by: Codex --- docs/OPENAI_CHAIN_PARITY_REAUDIT.md | 6 +- src/codex/accounts/pool.rs | 35 +++++++++++ src/codex/serving/chat.rs | 21 ++++++- src/codex/serving/dispatch/fallback.rs | 32 +++++++++- src/codex/serving/dispatch/mod.rs | 51 ++++++++++++++- src/codex/serving/http/errors.rs | 80 ++++++++++++++---------- src/codex/serving/responses.rs | 55 +++++++++++++++- tests/codex_serving/upstream_fallback.rs | 5 ++ 8 files changed, 245 insertions(+), 40 deletions(-) diff --git a/docs/OPENAI_CHAIN_PARITY_REAUDIT.md b/docs/OPENAI_CHAIN_PARITY_REAUDIT.md index cef529489..475b3af12 100644 --- a/docs/OPENAI_CHAIN_PARITY_REAUDIT.md +++ b/docs/OPENAI_CHAIN_PARITY_REAUDIT.md @@ -436,20 +436,20 @@ Rust 证据： - CF path-block 404 已接入主链路：空 body 404 会清理当前账号 cookies、记录 1 小时滑窗内连续次数、尝试 fallback account，达到 3 次后把账号标记为 `disabled`；无 fallback 时返回 502 `upstream_error`。 - model unsupported fallback 已对齐原版：账号 plan 不支持当前 model 时最多换一个备用账号重试一次，不改变账号状态；HTTP-time 错误与 non-streaming mid-SSE `response.failed` 均有测试覆盖。 - 401 token invalid fallback 已对齐原版：request path 不再 refresh 同账号；普通 401 会标记当前账号 `expired` 并尝试备用账号，包含 `deactivated` 的 401 会标记 `banned`；HTTP SSE streaming/non-streaming 与 WebSocket fallback exhausted 均有测试覆盖。 +- fallback exhausted response body 已对齐原版：当已经尝试 fallback 但没有可用账号时，Rust 会按账号池状态生成 `All accounts exhausted (...)` / `No accounts available...` 前缀，再保留原始上游错误消息；HTTP SSE stream、WebSocket stream、Responses non-stream、Compact 和 Chat fallback exhausted 路径复用同一消息构造，`v1_responses_should_mark_expired_and_return_401_when_401_has_no_fallback` 覆盖默认 streaming 响应体。 - Rust 的 Cloudflare challenge 处理比原版更持久：会把 cooldown 写入数据库，并清理对应账号 cookies。 - WS history 请求不跨账号重试这一点从安全角度更保守，避免 `previous_response_id` 的 server-side history 静默丢失。 未完全对齐： - recovery 优先级仍不是完整原版状态机。Rust 已实现 implicit resume restore/replay 与 strip-and-retry 优先于账号 fallback，但状态机仍分散在 responses/dispatch 中，没有抽成原版同等的 `RetryAction` 分层。 -- non-streaming mid-SSE failure 仍不是完整原版 catch 状态机。429/402/403、401、model unsupported、400 strip-and-retry 与隐式续链 restore 已补齐；剩余主要是最终错误外壳和 fallback exhausted body。 +- non-streaming mid-SSE failure 仍不是完整原版 catch 状态机。429/402/403、401、model unsupported、400 strip-and-retry 与隐式续链 restore 已补齐；剩余主要是最终错误外壳的结构细节。 - `previous_response_id` 限流/配额语义不一致。原版没有显式“history 请求保持原账户”的 retry hold，可能经 fallback 后再通过 `previous_response_not_found` strip-and-retry 恢复；Rust 遇到 429/402/403 会保持原账号并返回错误，保护历史但牺牲可恢复性。 -- 无 fallback 时的响应体不一致。原版会把 account pool exhaustion summary 拼进错误；Rust 多数路径直接返回 `Codex upstream error: ...`，缺少账号耗尽上下文。 - CF challenge exhausted status 不一致。原版 Cloudflare challenge fallback exhausted 使用 502 语义；Rust 若无 fallback，最终会走原 403 upstream error。 缺口/后续动作： - 引入原版级 `UpstreamRecoveryAction` 状态机：implicit resume replay、strip-and-retry、error handler、fallback transition 分层清晰，避免分散在 responses/dispatch 中。 - 继续补 HTTP SSE mid-SSE failure 的 `previous_response_not_found` 与 unanswered function call 覆盖；当前已有 WS 首帧错误覆盖 streaming/non-streaming。 -- 增加 fallback exhausted response body 的 parity tests。 +- 增加 fallback exhausted response body 的非 stream/Chat 直测，当前默认 streaming Responses 主链路已覆盖。 ## 9. rate-limit、usage、quota、cookie 持久化 diff --git a/src/codex/accounts/pool.rs b/src/codex/accounts/pool.rs index 5d1254b19..76bc3797f 100644 --- a/src/codex/accounts/pool.rs +++ b/src/codex/accounts/pool.rs @@ -86,6 +86,18 @@ pub struct AccountCapacitySummary { pub available_slots: usize, } +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub struct AccountPoolStatusSummary { + pub total: usize, + pub active: usize, + pub expired: usize, + pub quota_exhausted: usize, + pub rate_limited: usize, + pub refreshing: usize, + pub disabled: usize, + pub banned: usize, +} + #[derive(Debug)] pub struct AccountPool { accounts: BTreeMap, @@ -308,6 +320,29 @@ impl AccountPool { } } + pub fn status_summary(&mut self, now: DateTime) -> AccountPoolStatusSummary { + self.cleanup_stale_slots(now); + self.refresh_account_statuses(now); + let mut summary = AccountPoolStatusSummary { + total: self.accounts.len(), + ..AccountPoolStatusSummary::default() + }; + for account in self.accounts.values() { + match account.status { + AccountStatus::Active if !self.is_quota_available(account, now) => { + summary.rate_limited += 1; + } + AccountStatus::Active => summary.active += 1, + AccountStatus::Expired => summary.expired += 1, + AccountStatus::QuotaExhausted => summary.quota_exhausted += 1, + AccountStatus::Refreshing => summary.refreshing += 1, + AccountStatus::Disabled => summary.disabled += 1, + AccountStatus::Banned => summary.banned += 1, + } + } + summary + } + /// 获取所有配额锁定的账户ID列表（用于主动配额刷新） pub fn list_quota_locked_accounts(&self) -> Vec { self.accounts diff --git a/src/codex/serving/chat.rs b/src/codex/serving/chat.rs index abe4b5e77..d68b5588d 100644 --- a/src/codex/serving/chat.rs +++ b/src/codex/serving/chat.rs @@ -25,7 +25,10 @@ use crate::{ classify_upstream_account_retry, no_available_accounts_response, normalize_service_tier_for_upstream, CodexRequestLogContext, CodexUpstreamService, }, - codex::serving::http::errors::{codex_client_error_response, model_not_found_response}, + codex::serving::http::errors::{ + codex_client_error_message, codex_client_error_response, + codex_client_error_response_with_message, model_not_found_response, + }, config::ModelConfig, }; @@ -156,6 +159,22 @@ impl ChatService { acquired = fallback; continue; } + let message = self + .upstream + .fallback_exhausted_message(&codex_client_error_message(&error)) + .await; + let error_response = + codex_client_error_response_with_message(error, &message); + self.upstream + .log_response( + &log_context, + error_response.0, + EventLevel::Error, + "v1 chat completions fallback 已耗尽", + json!({"stream": client_stream}), + ) + .await; + return error_response.into_response(); } let error_response = codex_client_error_response(error); self.upstream diff --git a/src/codex/serving/dispatch/fallback.rs b/src/codex/serving/dispatch/fallback.rs index 24bcd4dbb..0d2f19983 100644 --- a/src/codex/serving/dispatch/fallback.rs +++ b/src/codex/serving/dispatch/fallback.rs @@ -5,7 +5,7 @@ use serde_json::{json, Value}; use crate::codex::{ accounts::{ model::{Account, AccountStatus}, - pool::{AccountAcquireRequest, AcquiredAccount}, + pool::{AccountAcquireRequest, AccountPoolStatusSummary, AcquiredAccount}, }, gateway::transport::{http_client::CodexClientError, rate_limits::cooldown_with_jitter}, }; @@ -323,6 +323,36 @@ pub(crate) fn websocket_history_retry_metadata(retry: UpstreamAccountRetry, stre } } +pub(super) fn build_account_exhaustion_detail( + summary: AccountPoolStatusSummary, + message: &str, +) -> String { + let mut parts = Vec::new(); + if summary.rate_limited > 0 { + parts.push(format!("{} rate-limited", summary.rate_limited)); + } + if summary.expired > 0 { + parts.push(format!("{} expired", summary.expired)); + } + if summary.banned > 0 { + parts.push(format!("{} banned", summary.banned)); + } + if summary.disabled > 0 { + parts.push(format!("{} disabled", summary.disabled)); + } + if summary.quota_exhausted > 0 { + parts.push(format!("{} quota-exhausted", summary.quota_exhausted)); + } + if summary.refreshing > 0 { + parts.push(format!("{} refreshing", summary.refreshing)); + } + if parts.is_empty() { + format!("No accounts available. {message}") + } else { + format!("All accounts exhausted ({}). {message}", parts.join(", ")) + } +} + fn is_cloudflare_challenge(body: &str) -> bool { let lower = body.to_ascii_lowercase(); lower.contains("cf-mitigated") diff --git a/src/codex/serving/dispatch/mod.rs b/src/codex/serving/dispatch/mod.rs index 87dd330d8..b5b2cb1cf 100644 --- a/src/codex/serving/dispatch/mod.rs +++ b/src/codex/serving/dispatch/mod.rs @@ -61,7 +61,10 @@ use crate::{ config::AppConfig, }; -use crate::codex::serving::http::errors::codex_client_error_response; +use crate::codex::serving::http::errors::{ + codex_client_error_message, codex_client_error_response, + codex_client_error_response_with_message, +}; pub(crate) use self::{ fallback::{ @@ -78,6 +81,7 @@ use self::affinity::{ }; use self::fallback::{ apply_upstream_account_retry_with_deps, apply_upstream_retry_and_acquire_fallback_with_deps, + build_account_exhaustion_detail, }; pub(crate) use self::implicit_resume::ImplicitResumeSnapshot; use self::implicit_resume::{continuation_input_start, implicit_resume_allowed}; @@ -267,6 +271,10 @@ impl CodexUpstreamService { .await } + pub(crate) async fn fallback_exhausted_message(&self, message: &str) -> String { + fallback_exhausted_message_with_deps(&self.deps, message).await + } + pub(crate) async fn apply_account_retry(&self, account: &Account, retry: UpstreamAccountRetry) { apply_upstream_account_retry_with_deps(&self.deps, account, retry).await; } @@ -363,6 +371,14 @@ impl CodexUpstreamService { const IMPLICIT_RESUME_MAX_AGE: StdDuration = StdDuration::from_secs(55 * 60); +async fn fallback_exhausted_message_with_deps( + deps: &CodexUpstreamDependencies, + message: &str, +) -> String { + let summary = deps.account_pool.lock().await.status_summary(Utc::now()); + build_account_exhaustion_detail(summary, message) +} + async fn apply_implicit_resume_with_deps( deps: &CodexUpstreamDependencies, request: &mut CodexResponsesRequest, @@ -581,6 +597,22 @@ async fn responses_http_sse_stream( acquired = fallback; continue; } + let message = fallback_exhausted_message_with_deps( + &deps, + &codex_client_error_message(&error), + ) + .await; + let error_response = codex_client_error_response_with_message(error, &message); + log_codex_upstream_response_with_deps( + &deps, + &log_context, + error_response.0, + EventLevel::Error, + "v1 responses stream fallback 已耗尽", + json!({"stream": true}), + ) + .await; + return error_response.into_response(); } let error_response = codex_client_error_response(error); log_codex_upstream_response_with_deps( @@ -1327,6 +1359,23 @@ async fn responses_websocket_stream( acquired = fallback; continue; } + let message = fallback_exhausted_message_with_deps( + &deps, + &codex_client_error_message(&error), + ) + .await; + let error_response = + codex_client_error_response_with_message(error, &message); + log_codex_upstream_response_with_deps( + &deps, + &log_context, + error_response.0, + EventLevel::Error, + "v1 responses WebSocket stream fallback 已耗尽", + json!({"stream": true, "transport": "websocket"}), + ) + .await; + return error_response.into_response(); } } let error_response = codex_client_error_response(error); diff --git a/src/codex/serving/http/errors.rs b/src/codex/serving/http/errors.rs index d1406ce88..bf795f6ec 100644 --- a/src/codex/serving/http/errors.rs +++ b/src/codex/serving/http/errors.rs @@ -7,42 +7,56 @@ use crate::{ }; pub(crate) fn codex_client_error_response(error: CodexClientError) -> (StatusCode, Json) { + let status = codex_client_error_status(&error); + let code = codex_client_error_code(&error); + let message = codex_client_error_message(&error); + (status, Json(openai_error(&message, code))) +} + +pub(crate) fn codex_client_error_response_with_message( + error: CodexClientError, + message: &str, +) -> (StatusCode, Json) { + let status = codex_client_error_status(&error); + let code = codex_client_error_code(&error); + (status, Json(openai_error(message, code))) +} + +pub(crate) fn codex_client_error_message(error: &CodexClientError) -> String { match error { - CodexClientError::UnsupportedTransport(_) => ( - StatusCode::BAD_REQUEST, - Json(openai_error( - "previous_response_id requires Codex WebSocket transport", - "websocket_required", - )), - ), - CodexClientError::Upstream { status, body, .. } - if status == StatusCode::NOT_FOUND && body.trim().is_empty() => - { - ( - StatusCode::BAD_GATEWAY, - Json(openai_error( - "Upstream blocked the request (Cloudflare path-block)", - "upstream_error", - )), - ) + CodexClientError::UnsupportedTransport(_) => { + "previous_response_id requires Codex WebSocket transport".to_string() } - CodexClientError::Upstream { status, body, .. } => ( - status, - Json(openai_error( - &format!( - "Codex upstream error: {}", - body.chars().take(300).collect::() - ), - "upstream_error", - )), - ), - _ => ( - StatusCode::BAD_GATEWAY, - Json(openai_error( - "Codex upstream request failed", - "upstream_error", - )), + CodexClientError::Upstream { status, body, .. } + if *status == StatusCode::NOT_FOUND && body.trim().is_empty() => + { + "Upstream blocked the request (Cloudflare path-block)".to_string() + } + CodexClientError::Upstream { body, .. } => format!( + "Codex upstream error: {}", + body.chars().take(300).collect::() ), + _ => "Codex upstream request failed".to_string(), + } +} + +fn codex_client_error_status(error: &CodexClientError) -> StatusCode { + match error { + CodexClientError::UnsupportedTransport(_) => StatusCode::BAD_REQUEST, + CodexClientError::Upstream { status, body, .. } + if *status == StatusCode::NOT_FOUND && body.trim().is_empty() => + { + StatusCode::BAD_GATEWAY + } + CodexClientError::Upstream { status, .. } => *status, + _ => StatusCode::BAD_GATEWAY, + } +} + +fn codex_client_error_code(error: &CodexClientError) -> &'static str { + match error { + CodexClientError::UnsupportedTransport(_) => "websocket_required", + _ => "upstream_error", } } diff --git a/src/codex/serving/responses.rs b/src/codex/serving/responses.rs index ff17085c0..579625d7a 100644 --- a/src/codex/serving/responses.rs +++ b/src/codex/serving/responses.rs @@ -22,7 +22,10 @@ use crate::{ websocket_history_retry_metadata, CodexRequestLogContext, CodexUpstreamService, CollectedResponse, ImplicitResumeSnapshot, UpstreamRequestRecovery, }, - codex::serving::http::errors::codex_client_error_response, + codex::serving::http::errors::{ + codex_client_error_message, codex_client_error_response, + codex_client_error_response_with_message, + }, config::ModelConfig, }; @@ -325,6 +328,24 @@ impl ResponsesService { acquired = fallback; continue; } + let message = self + .upstream + .fallback_exhausted_message(&codex_client_error_message( + &error, + )) + .await; + let error_response = + codex_client_error_response_with_message(error, &message); + self.upstream + .log_response( + &log_context, + error_response.0, + EventLevel::Error, + "v1 responses 上游 SSE fallback 已耗尽", + failure.metadata(false), + ) + .await; + return error_response.into_response(); } } let error_response = codex_client_error_response(error); @@ -403,6 +424,22 @@ impl ResponsesService { acquired = fallback; continue; } + let message = self + .upstream + .fallback_exhausted_message(&codex_client_error_message(&error)) + .await; + let error_response = + codex_client_error_response_with_message(error, &message); + self.upstream + .log_response( + &log_context, + error_response.0, + EventLevel::Error, + "v1 responses 上游请求 fallback 已耗尽", + json!({"stream": false}), + ) + .await; + return error_response.into_response(); } } let error_response = codex_client_error_response(error); @@ -718,6 +755,22 @@ impl ResponsesService { acquired = fallback; continue; } + let message = self + .upstream + .fallback_exhausted_message(&codex_client_error_message(&error)) + .await; + let error_response = + codex_client_error_response_with_message(error, &message); + self.upstream + .log_response( + &log_context, + error_response.0, + EventLevel::Error, + "v1 responses compact fallback 已耗尽", + json!({"stream": false, "compact": true}), + ) + .await; + return error_response.into_response(); } } let error_response = codex_client_error_response(error); diff --git a/tests/codex_serving/upstream_fallback.rs b/tests/codex_serving/upstream_fallback.rs index 8fa85f154..a2a871363 100644 --- a/tests/codex_serving/upstream_fallback.rs +++ b/tests/codex_serving/upstream_fallback.rs @@ -1328,6 +1328,11 @@ async fn v1_responses_should_mark_expired_and_return_401_when_401_has_no_fallbac .unwrap(); assert_eq!(response.status(), StatusCode::UNAUTHORIZED); + let body = response_json(response).await; + assert_eq!( + body["error"]["message"], + "All accounts exhausted (1 expired). Codex upstream error: {\"error\":{\"code\":\"token_revoked\",\"message\":\"token revoked\"}}" + ); let account_status: (String,) = sqlx::query_as("select status from accounts where id = ?") .bind("acct_401_single") .fetch_one(&imported.pool)