codex: align fallback exhaustion errors

Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
zyycn
2026-06-15 09:30:17 +08:00
co-authored by Codex
parent 3d18ba2eb3
commit 1dac463af3
8 changed files with 245 additions and 40 deletions
+3 -3
View File
@@ -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 持久化
+35
View File
@@ -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<String, Account>,
@@ -308,6 +320,29 @@ impl AccountPool {
}
}
pub fn status_summary(&mut self, now: DateTime<Utc>) -> 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<String> {
self.accounts
+20 -1
View File
@@ -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
+31 -1
View File
@@ -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")
+50 -1
View File
@@ -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);
+47 -33
View File
@@ -7,42 +7,56 @@ use crate::{
};
pub(crate) fn codex_client_error_response(error: CodexClientError) -> (StatusCode, Json<Value>) {
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<Value>) {
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::<String>()
),
"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::<String>()
),
_ => "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",
}
}
+54 -1
View File
@@ -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);
+5
View File
@@ -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)