starred/codex-proxy-rs
1
0
Fork 0
mirror of https://github.com/zyycn/codex-proxy-rs.git synced 2026-11-15 00:06:24 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: account recovery edge cases

Browse Source 
Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
zyycn
2026-06-23 20:44:52 +08:00
co-authored by Codex
parent b53cb82008
commit f8cbbf19e2
6 changed files with 260 additions and 76 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/openai-real-chain-request-test.md
+77
View File
@@ -0,0 +1,77 @@
# OpenAI/Codex 真实链路测试记录
## 本轮上下文
- 开始时间：2026-06-23 19:32:16 Asia/Shanghai
- 目标：继续真实数据全链路测试，重点覆盖边界情况、封禁链路剥离、防止上下文污染、重试边界。
- 运行目录：`.runtime/real-chain-openai-20260623-193216`
- 规则：确认问题后先修复并复测，再继续后续场景。
## 场景矩阵
| 场景 | 状态 | 证据 |
| --- | --- | --- |
| 本地编译检查 | 通过 | `cargo check` |
| 封禁/禁用亲和剥离集成覆盖 | 通过 | `cargo test --test integration proxy::dispatch::chat_upstream::responses_http:: -- --nocapture`；`responses_websocket::` |
| 5xx/429/403/402/Cloudflare 重试集成覆盖 | 通过 | `responses_recovery::` 37 个用例；`chat_routes::` 15 个用例；`responses_websocket::` 30 个用例 |
| 真实链路基础健康检查 | 通过 | `/debug/diagnostics`：5 accounts，1 active，3 disabled，1 banned |
| 真实 Responses 基线请求 | 通过 | `req_e29e505e-4623-4502-9676-922a27c1d247`，`resp_09b95050e9f8ff4b016a3a703bc2f8819a8e248aa2fc31ff4d` |
| 真实 previous_response_id 正常续接 | 通过 | `req_d8d28aed-6711-42fa-9ef5-8e1152d20bf3`，返回体保留 previous response |
| 真实 banned 账号亲和剥离 | 通过 | `req_e7d5625a-543b-4a64-9038-98e4afc84b57`，返回体 `previous_response_id=null` |
| 真实 disabled 账号亲和剥离 | 通过 | `req_9b52e5e8-eb57-471d-8825-0510a7e4017b`，返回体 `previous_response_id=null` |
| 真实 all accounts unavailable 边界 | 通过 | `req_63812d11-c9f9-4eac-afde-3d2837ba7754`，SSE `response.failed` |
| 真实 quota verify 边界 | 通过 | `req_faf18f0b-1e35-4dc5-8f47-632eb03d2c9e`，`quota_verify_required` 从 1 清回 0 |
| 误置 active 的 banned/disabled 账号状态回写 | 通过 | 发现并修复无有效 token exp 时只内存过期、不落库的问题 |
| 真实 5xx/429/Cloudflare 重试边界 | 受限 | 真实上游无法稳定强制触发，已由集成夹具覆盖 |
## 本地验证记录
- `cargo check`：通过。
- `cargo test --test integration proxy::dispatch::chat_upstream::responses_recovery:: -- --nocapture`：37 passed，覆盖 Responses HTTP/SSE 的 401/402/403/429、Cloudflare challenge/path-block、5xx same-account retry、模型不支持 fallback、previous_response_id/函数调用/加密 replay 历史剥离。
- `cargo test --test integration proxy::dispatch::chat_upstream::responses_websocket:: -- --nocapture`：30 passed，覆盖 WebSocket 连接复用、复用连接断开后 fresh retry、previous_response_id fallback、隐式 banned 亲和剥离、quota/rate limit/path-block 边界。
- `cargo test --test integration proxy::dispatch::chat_upstream::chat_routes:: -- --nocapture`：15 passed，覆盖 Chat Completions 的 401/402/403/429、Cloudflare、模型不支持和 SSE 翻译。
- `cargo test --test integration proxy::dispatch::chat_upstream::responses_http:: -- --nocapture`：23 passed，覆盖 HTTP Responses 参数转换、cookie path 作用域、banned 亲和剥离、quota 亲和保留和下游断开。
- `cargo test --test integration upstream::accounts::token_refresh:: -- --nocapture`：13 passed，覆盖 transient retry、ambiguous transport 不重试、invalid_grant 二次确认、retry exhaustion 延后恢复、stale refresh token 防复用。
- `cargo test --test integration upstream::accounts::account_pool::quota:: -- --nocapture`：9 passed，覆盖 quota/cooldown 选择边界。
- `cargo test --test integration upstream::accounts::cloudflare:: -- --nocapture`：2 passed，覆盖 Cloudflare cooldown 升级和过期重置。
- `cargo test --test integration mark_banned -- --nocapture`：5 passed，覆盖 401 `account_deactivated` 和 403 banned 信号将账号标记为 banned。
- `cargo test --test integration mark_expired_after_401 -- --nocapture`：3 passed，覆盖普通 401 将账号标记为 expired。
- `cargo test --test integration runtime_account_pool_should_persist_expired_status_when_jwt_expiry_is_discovered -- --nocapture`：通过，覆盖运行时调度发现 active 账号 token 已过期时持久化为 expired。
- `cargo test --test integration upstream::accounts::account_pool:: -- --nocapture`：27 passed，覆盖账号池调度、quota/cooldown、运行时状态回写。
## 真实链路记录
### 环境
- 服务：`CODEX_PROXY_WS_AUDIT_DIR=.runtime/real-chain-openai-20260623-193216/ws-audit target/debug/codex-proxy-rs`
- 模型：`gpt-5.4-mini`
- `/v1/models`：HTTP 200，返回 `codex-auto-review`、`gpt-5.4-mini`、`gpt-5.5`。
- 初始账号池：5 total，1 active，3 disabled，1 banned。
- 结束账号池：5 total，1 active，3 disabled，1 banned；active 账号 `quota_verify_required=0`，`quota_cooldown_until=null`，`cloudflare_cooldown_until=null`。
### 结果
- 基线 Responses JSON：`req_e29e505e-4623-4502-9676-922a27c1d247`，HTTP 200，`response_id=resp_09b95050e9f8ff4b016a3a703bc2f8819a8e248aa2fc31ff4d`，输出 `EDGE-BASELINE-OK`，event log 记录 `transport=websocket`、`account_id=a18bcfa9ae932857`。
- 正常 previous_response_id 续接：`req_d8d28aed-6711-42fa-9ef5-8e1152d20bf3`，HTTP 200，输出 `EDGE-CONTINUE-OK`，返回体 `previous_response_id=resp_09b95050e9f8ff4b016a3a703bc2f8819a8e248aa2fc31ff4d`，prompt cache key 与基线一致。
- banned 亲和剥离：预置 `resp_real_edge_seed_banned_20260623 -> acct_be7f5c37f60b44ff8058b1b9b164fd42(banned)`，重启加载后请求 `req_e7d5625a-543b-4a64-9038-98e4afc84b57`，HTTP 200，输出 `EDGE-BANNED-STRIPPED-OK`，返回体 `previous_response_id=null`，event log 落到 active 账号 `a18bcfa9ae932857`。
- disabled 亲和剥离：预置 `resp_real_edge_seed_disabled_20260623 -> 10e584b410a5f1ab(disabled)`，重启加载后请求 `req_9b52e5e8-eb57-471d-8825-0510a7e4017b`，HTTP 200，输出 `EDGE-DISABLED-STRIPPED-OK`，返回体 `previous_response_id=null`，event log 落到 active 账号 `a18bcfa9ae932857`。
- 防止历史污染：两条 seed affinity 在剥离后已不存在；WS audit 中正常续接请求包含 `previous_response_id`，banned/disabled 剥离请求的上游 payload body 不包含 `previous_response_id` 或 `turn_state`。
- all accounts unavailable：临时通过 admin API 将唯一 active 账号置为 disabled，请求 `req_63812d11-c9f9-4eac-afde-3d2837ba7754`，HTTP 200 SSE，返回 `event: response.failed`，message 为 `No active upstream account is available`；随后通过 admin API 恢复 active。
- quota verify：停止服务后将 active 账号 `quota_verify_required=1`，重启后请求 `req_faf18f0b-1e35-4dc5-8f47-632eb03d2c9e`，HTTP 200，输出 `EDGE-QUOTA-VERIFY-OK`；请求后 DB 中 `quota_verify_required=0` 且 `quota_fetched_at=2026-06-23T11:44:04.888488741+00:00`。
### 误置 active 的禁用/封禁账号状态回写
- 测试方法：临时将原 active 账号 `a18bcfa9ae932857` 置为 disabled，逐个将目标 disabled/banned 账号置为 active，确保请求只会命中目标账号；请求后先读取 DB 状态，再恢复原状态。
- banned 账号 `acct_be7f5c37f60b44ff8058b1b9b164fd42`：请求 `req_116f4b30-c8fa-4718-a69a-ccd22b66d562`，真实上游返回 401 `token_invalidated`，请求后状态从 active 变为 expired。该响应不是 banned 信号，所以不会标记为 banned。
- disabled 账号 `64e49d697f15dd02`：请求 `req_c2514f79-e429-49c8-9635-2383a2cfe470`，真实上游返回 401 `token_invalidated`，请求后状态从 active 变为 expired。
- disabled 账号 `10e584b410a5f1ab` 和 `38907867f4c6c36e`：首次测试时响应为 503 `no_available_accounts`，状态仍为 active。原因是两者 `access_token_expires_at=null` 且 token 无有效 JWT exp，账号池在调度阶段内存判定 expired，没有持久化到 DB。
- 修复：`RuntimeAccountPoolService::acquire_with` 持久化账号池调度阶段发现的 runtime-expired 账号状态。
- 修复后复测：
- `10e584b410a5f1ab`：请求 `req_25d53384-1165-4da5-91f9-918b1709f8bc`，HTTP 503 `no_available_accounts`，请求返回前 DB 状态已从 active 变为 expired。
- `38907867f4c6c36e`：请求 `req_4b0181b9-a248-4402-ad7f-b533c8669907`，HTTP 503 `no_available_accounts`，请求返回前 DB 状态已从 active 变为 expired。
- 结论：真实打到上游并收到 401/403 时会按信号及时回写；不打上游、但运行时已经能判定 token 过期/无效时，修复后也会及时回写 expired。
### 未真实强制触发的边界
- 真实上游 5xx、429、Cloudflare challenge/path-block 不能稳定人工触发，本轮没有对真实 OpenAI 强行制造这些状态。
- 对应重试和状态传播已通过集成测试覆盖：HTTP/SSE 5xx same-account retry、429 fallback/exhausted、402 quota fallback/exhausted、403 banned fallback、Cloudflare challenge/path-block cooldown/清 cookie/三次 path-block disable，以及 WebSocket previous_response_id fallback、quota/rate limit/path-block stream error。
src/proxy/dispatch/cloudflare.rs
+26 -10
View File
@@ -1,5 +1,7 @@
//! Cloudflare challenge and path-block recovery shared by dispatch routes.
use std::time::Duration as StdDuration;
use chrono::Utc;
use crate::{
@@ -57,8 +59,6 @@ impl CloudflareRecovery {
account_pool: &RuntimeAccountPoolService,
account_id: &str,
) {
self.delete_account_cookies(account_id, "Cloudflare challenge")
.await;
let now = Utc::now();
let cooldown = self
.challenge_tracker
@@ -67,6 +67,7 @@ impl CloudflareRecovery {
account_pool
.set_cloudflare_cooldown_until(account_id, cooldown.cooldown_until)
.await;
self.clear_challenge_cookies_after_cooldown(account_id, cooldown.delay_seconds);
}
pub async fn apply_path_block(
@@ -97,14 +98,29 @@ impl CloudflareRecovery {
}
async fn delete_account_cookies(&self, account_id: &str, reason: &str) {
if let Err(error) = self.cookie_store.delete_account_cookies(account_id).await {
tracing::warn!(
account_id,
reason,
error = %error,
"failed to delete account cookies after Cloudflare recovery signal"
);
}
delete_account_cookies(&self.cookie_store, account_id, reason).await;
}
fn clear_challenge_cookies_after_cooldown(&self, account_id: &str, delay_seconds: i64) {
let cookie_store = self.cookie_store.clone();
let account_id = account_id.to_string();
let delay = StdDuration::from_secs(delay_seconds.max(0) as u64);
tokio::spawn(async move {
tokio::time::sleep(delay).await;
delete_account_cookies(&cookie_store, &account_id, "Cloudflare challenge").await;
});
}
}
async fn delete_account_cookies(cookie_store: &SqliteCookieStore, account_id: &str, reason: &str) {
if let Err(error) = cookie_store.delete_account_cookies(account_id).await {
tracing::warn!(
account_id,
reason,
error = %error,
"failed to delete account cookies after Cloudflare recovery signal"
);
}
}
src/upstream/accounts/pool.rs
+103 -59
View File
@@ -123,6 +123,12 @@ pub struct AcquiredAccount {
pub previous_slot_at: Option<DateTime<Utc>>,
}
#[derive(Debug)]
struct AccountAcquireWithStatusRefresh {
acquired: Option<AcquiredAccount>,
expired_account_ids: Vec<String>,
}
/// 账号池容量摘要。
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct AccountCapacitySummary {
@@ -422,48 +428,90 @@ impl AccountPool {
/// 使用完整调度请求获取账号。
pub fn acquire_with(&mut self, request: AccountAcquireRequest) -> Option<AcquiredAccount> {
self.acquire_with_status_refresh(request).acquired
}
fn acquire_with_status_refresh(
&mut self,
request: AccountAcquireRequest,
) -> AccountAcquireWithStatusRefresh {
self.cleanup_stale_slots(request.now);
self.refresh_account_statuses(request.now);
let expired_account_ids = self.refresh_account_statuses(request.now);
let candidates = self.candidates(&request);
let (selected, previous_slot_at) =
if let Some(preferred_account_id) = &request.preferred_account_id {
let preferred = candidates
.iter()
.find(|account| account.id == *preferred_account_id)
.cloned();
preferred.map(|account| (account, self.previous_slot_at(preferred_account_id)))
} else {
None
}
.or_else(|| {
Some(match self.options.rotation_strategy {
RotationStrategy::LeastUsed => {
let account = self.select_least_used(&candidates)?;
let previous = self.previous_slot_at(&account.id);
(account, previous)
}
RotationStrategy::RoundRobin => {
let account = self.select_round_robin(&candidates)?;
let previous = self.previous_slot_at(&account.id);
(account, previous)
}
RotationStrategy::Sticky => {
let account = self.select_sticky(&candidates)?;
let previous = self.previous_slot_at(&account.id);
(account, previous)
}
})
})?;
let selected = if let Some(preferred_account_id) = &request.preferred_account_id {
let preferred = candidates
.iter()
.find(|account| account.id == *preferred_account_id)
.cloned();
preferred.map(|account| (account, self.previous_slot_at(preferred_account_id)))
} else {
None
}
.or_else(|| {
Some(match self.options.rotation_strategy {
RotationStrategy::LeastUsed => {
let account = self.select_least_used(&candidates)?;
let previous = self.previous_slot_at(&account.id);
(account, previous)
}
RotationStrategy::RoundRobin => {
let account = self.select_round_robin(&candidates)?;
let previous = self.previous_slot_at(&account.id);
(account, previous)
}
RotationStrategy::Sticky => {
let account = self.select_sticky(&candidates)?;
let previous = self.previous_slot_at(&account.id);
(account, previous)
}
})
});
let Some((selected, previous_slot_at)) = selected else {
return AccountAcquireWithStatusRefresh {
acquired: None,
expired_account_ids,
};
};
let selected_id = selected.id.clone();
self.push_slot(&selected_id, request.now);
let selected = self
let acquired = self
.mark_usage(&selected_id, request.now)
.unwrap_or(selected);
Some(AcquiredAccount {
account: selected,
previous_slot_at,
})
AccountAcquireWithStatusRefresh {
acquired: Some(AcquiredAccount {
account: acquired,
previous_slot_at,
}),
expired_account_ids,
}
}
fn refresh_account_statuses(&mut self, now: DateTime<Utc>) -> Vec<String> {
let account_ids = self.accounts.keys().cloned().collect::<Vec<_>>();
account_ids
.into_iter()
.filter(|account_id| self.refresh_account_status(account_id, now))
.collect()
}
fn refresh_account_status(&mut self, account_id: &str, now: DateTime<Utc>) -> bool {
let mut should_clear_slots = false;
let mut expired = false;
if let Some(account) = self.accounts.get_mut(account_id) {
if account.status == AccountStatus::Active && access_token_expired(account, now) {
account.status = AccountStatus::Expired;
should_clear_slots = true;
expired = true;
} else {
refresh_quota_window(account, now);
refresh_cloudflare_cooldown(account, now);
}
}
if should_clear_slots {
self.slots.remove(account_id);
}
expired
}
/// 释放指定账号的一个在途槽位。
@@ -675,29 +723,6 @@ impl AccountPool {
});
}
fn refresh_account_statuses(&mut self, now: DateTime<Utc>) {
let account_ids = self.accounts.keys().cloned().collect::<Vec<_>>();
for account_id in account_ids {
self.refresh_account_status(&account_id, now);
}
}
fn refresh_account_status(&mut self, account_id: &str, now: DateTime<Utc>) {
let mut should_clear_slots = false;
if let Some(account) = self.accounts.get_mut(account_id) {
if account.status == AccountStatus::Active && access_token_expired(account, now) {
account.status = AccountStatus::Expired;
should_clear_slots = true;
} else {
refresh_quota_window(account, now);
refresh_cloudflare_cooldown(account, now);
}
}
if should_clear_slots {
self.slots.remove(account_id);
}
}
fn is_model_allowed(&self, account: &Account, model: &str) -> bool {
let Some(allowed_plans) = self.options.model_plan_allowlist.get(model) else {
return true;
@@ -830,7 +855,10 @@ impl RuntimeAccountPoolService {
/// 按请求上下文获取可用账号。
pub async fn acquire_with(&self, request: AccountAcquireRequest) -> Option<AcquiredAccount> {
let acquired = self.pool.lock().await.acquire_with(request)?;
let refresh = self.pool.lock().await.acquire_with_status_refresh(request);
self.persist_expired_statuses(refresh.expired_account_ids)
.await;
let acquired = refresh.acquired?;
if let Err(error) = self.store.record_request(&acquired.account.id).await {
tracing::warn!(
account_id = acquired.account.id,
@@ -841,6 +869,22 @@ impl RuntimeAccountPoolService {
Some(acquired)
}
async fn persist_expired_statuses(&self, account_ids: Vec<String>) {
for account_id in account_ids {
if let Err(error) = self
.store
.set_status(&account_id, AccountStatus::Expired)
.await
{
tracing::warn!(
account_id,
error = %error,
"failed to persist runtime-expired account status"
);
}
}
}
/// 按账号上一个在途槽位控制请求间隔。
pub async fn wait_for_request_interval(&self, acquired: &AcquiredAccount) {
if self.request_interval.is_zero() {
tests/integration/proxy/dispatch/chat_upstream/chat_routes.rs
+3 -2
View File
@@ -790,8 +790,9 @@ async fn chat_completions_should_cool_down_cloudflare_403_and_fallback() {
cookie_store
.cookie_header("acct_primary", "chatgpt.com")
.await
.unwrap(),
None
.unwrap()
.as_deref(),
Some("cf_clearance=old")
);
assert_eq!(
cookie_store
tests/integration/proxy/dispatch/chat_upstream/responses_recovery.rs
+7 -5
View File
@@ -485,7 +485,7 @@ async fn responses_should_return_cloudflare_challenge_error_when_403_fallback_is
.cookie_header("acct_chat", "chatgpt.com")
.await
.unwrap(),
None
Some("cf_clearance=old".to_string())
);
}
@@ -564,8 +564,9 @@ async fn responses_should_cool_down_cloudflare_403_and_fallback() {
cookie_store
.cookie_header("acct_primary", "chatgpt.com")
.await
.unwrap(),
None
.unwrap()
.as_deref(),
Some("cf_clearance=old")
);
assert_eq!(
cookie_store
@@ -1607,8 +1608,9 @@ async fn responses_stream_should_cool_down_cloudflare_403_and_fallback() {
cookie_store
.cookie_header("acct_primary", "chatgpt.com")
.await
.unwrap(),
None
.unwrap()
.as_deref(),
Some("cf_clearance=old")
);
assert_eq!(
cookie_store
tests/integration/upstream/accounts/account_pool/selection.rs
+44
View File
@@ -64,6 +64,50 @@ fn acquire_should_skip_accounts_with_expired_jwt_when_metadata_is_missing() {
assert!(acquired.is_none());
}
#[tokio::test]
async fn runtime_account_pool_should_persist_expired_status_when_jwt_expiry_is_discovered() {
let (pool, _dir) =
crate::support::sqlite::init_test_db("runtime-pool-expired-status.sqlite").await;
let store = codex_proxy_rs::upstream::accounts::store::SqliteAccountStore::new(
pool.clone(),
codex_proxy_rs::infra::crypto::SecretBox::new([91u8; 32]),
);
store
.insert(codex_proxy_rs::upstream::accounts::store::NewAccount {
id: "acct_expired".to_string(),
email: None,
account_id: Some("chatgpt-expired".to_string()),
user_id: None,
label: None,
plan_type: Some("free".to_string()),
access_token: secrecy::SecretString::new(test_jwt(-60).into()),
refresh_token: None,
access_token_expires_at: None,
status: AccountStatus::Active,
added_at: None,
})
.await
.unwrap();
let runtime_pool = codex_proxy_rs::upstream::accounts::pool::RuntimeAccountPoolService::new(
std::sync::Arc::new(store),
AccountPoolOptions::default(),
0,
);
runtime_pool.restore_from_repository().await.unwrap();
let acquired = runtime_pool
.acquire_with(AccountAcquireRequest::new("gpt-5.5", Utc::now()))
.await;
let status: (String,) = sqlx::query_as("select status from accounts where id = ?")
.bind("acct_expired")
.fetch_one(&pool)
.await
.unwrap();
assert!(acquired.is_none());
assert_eq!(status.0, "expired");
}
#[test]
fn account_pool_should_prefer_configured_tier_priority() {
let mut pool = AccountPool::with_options(AccountPoolOptions {