diff --git a/Cargo.toml b/Cargo.toml index ff1bcf6cb..0bcba6a7a 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -35,7 +35,7 @@ serde_yml = "0.0.13" sha2 = "0.11.0" sqlx = { version = "0.9.0", features = ["runtime-tokio", "tls-rustls", "sqlite", "chrono", "uuid", "migrate"] } thiserror = "2.0.18" -tokio = { version = "1.52.3", features = ["macros", "rt-multi-thread", "signal", "fs", "time", "sync", "net"] } +tokio = { version = "1.52.3", features = ["macros", "rt-multi-thread", "signal", "fs", "time", "sync", "net", "test-util"] } tokio-rustls = { version = "0.26.4", default-features = false, features = ["ring", "tls12"] } tokio-tungstenite = { version = "0.29.0", default-features = false, features = ["connect", "rustls-tls-native-roots"] } tower = { version = "0.5.3", features = ["util"] } diff --git a/docs/OPENAI_CHAIN_PARITY_REAUDIT.md b/docs/OPENAI_CHAIN_PARITY_REAUDIT.md index 0932be2ce..21acbdc1b 100644 --- a/docs/OPENAI_CHAIN_PARITY_REAUDIT.md +++ b/docs/OPENAI_CHAIN_PARITY_REAUDIT.md @@ -148,6 +148,7 @@ Rust 证据: - 本轮修正后,`src/codex/accounts/service/refresh.rs` 提供 `probe_account_refresh()`,用于执行 OAuth refresh 探测但失败不直接落库状态;admin/manual refresh 仍通过 `refresh_account()` 应用结果。 - 本轮修正后,`src/codex/tasks/token_refresh.rs` 后台调度器调用 `probe_account_refresh()`,只有 `AccountProbeOutcome::Alive` 视为成功,`Dead` 会进入重试与 permanent threshold 判断;第二次 permanent failure 后才写入 expired/banned。 - `src/codex/tasks/token_refresh.rs` 单元测试 `do_refresh_inner_should_mark_expired_only_after_second_permanent_failure` 覆盖连续两次 `InvalidGrant` 前数据库状态仍为 `active`,第二次失败后才写入 `expired`。 +- 本轮修正后,`src/codex/tasks/token_refresh.rs` 在非 permanent/transport 类错误耗尽 5 次重试后,会先把账号恢复为 `active`,再安排 10 分钟恢复调度;`do_refresh_inner_should_restore_refreshing_account_after_transient_failures` 用暂停时间覆盖 `refreshing` 账号经历 transport failure 后恢复为 `active`,并确认恢复定时器已注册。 - `src/codex/gateway/oauth/client.rs:402-423` 把 `quota`、`banned`、`token_revoked` 等映射为结构化 `RefreshFailure`。 - 本轮修正后,`src/codex/accounts/service/health.rs` 健康检查通过 `refresh_account()` 调用 OAuth refresh token 链路,不再访问 Codex usage endpoint;`tests/admin/accounts/cookies_quota.rs` 断言 `/api/codex/usage` 调用次数为 0。 @@ -162,6 +163,7 @@ Rust 证据: - 每账号 refresh 防重已经存在,Rust 用 SQLite lease 替代文件锁,语义上更适合当前数据库架构。 - refresh 成功时保留未轮换的旧 refresh_token,这一点符合原版保护逻辑。 - 后台 refresh 调度器已不再把 `Dead` probe 误判为成功;scheduler 路径 permanent failure 的状态写入已延后到连续两次失败。 +- 后台 refresh 调度器的临时错误耗尽语义已与原版一致:不把账号标为 expired,恢复为 active,并安排 10 分钟 recovery。 - 账号健康检查已回到原版 OAuth refresh-only 安全边界。 未完全对齐: @@ -169,7 +171,7 @@ Rust 证据: - 原版 refresh 对一次性 RT 的网络错误非常谨慎,只有请求确认未到服务器时才 fallback/retry;Rust 当前 `reqwest` refresh client 不能区分 pre-flight 与 mid-flight 失败,且 runtime 调度器错误处理路径与原版不等价。 缺口/后续动作: -- 增加 transport error 的后台恢复调度测试,覆盖非 permanent 错误耗尽重试后进入 10 分钟恢复调度。 +- 若要继续把 refresh client 做到原版级细粒度,需要区分 refresh token 请求的 pre-flight 网络失败与 mid-flight 失败,避免一次性 RT 被服务器接收后仍重试。 ## 4. `/v1/responses` 请求构造与 prompt cache identity diff --git a/src/codex/tasks/token_refresh.rs b/src/codex/tasks/token_refresh.rs index 943e029d0..408586bbd 100644 --- a/src/codex/tasks/token_refresh.rs +++ b/src/codex/tasks/token_refresh.rs @@ -394,6 +394,7 @@ impl RefreshScheduler { error = %error, "token 刷新重试次数已耗尽" ); + self.restore_account_for_recovery(account_id).await; self.schedule_recovery( account_id.to_string(), Duration::from_millis(RECOVERY_DELAY_MS), @@ -409,6 +410,34 @@ impl RefreshScheduler { Ok(()) } + async fn restore_account_for_recovery(&self, account_id: &str) { + match self + .account_service + .update_status(account_id, "active") + .await + { + Ok(Some(_)) => { + info!( + account_id = %account_id, + "临时 refresh 失败后已恢复账号状态" + ); + } + Ok(None) => { + warn!( + account_id = %account_id, + "临时 refresh 失败后恢复账号状态失败:账号不存在" + ); + } + Err(error) => { + warn!( + account_id = %account_id, + error = ?error, + "临时 refresh 失败后恢复账号状态失败" + ); + } + } + } + /// 解析 JWT 获取过期时间 fn parse_jwt_exp(&self, token: &str) -> Option { let parts: Vec<&str> = token.split('.').collect(); @@ -589,6 +618,70 @@ mod tests { assert_eq!(stored.status, AccountStatus::Expired); } + #[tokio::test] + async fn do_refresh_inner_should_restore_refreshing_account_after_transient_failures() { + let dir = tempfile::tempdir().unwrap(); + let db = dir.path().join("refresh-scheduler-recovery.sqlite"); + let url = format!("sqlite://{}", db.display()); + let pool = connect_sqlite(&url).await.unwrap(); + let repo = AccountRepository::new(pool.clone(), SecretBox::new([42u8; 32])); + repo.insert(NewAccount { + id: "acct_recovery".to_string(), + email: None, + account_id: None, + user_id: None, + label: None, + plan_type: None, + access_token: SecretString::new("access-old".to_string().into()), + refresh_token: Some(SecretString::new("refresh-old".to_string().into())), + access_token_expires_at: None, + status: AccountStatus::Refreshing, + }) + .await + .unwrap(); + let refresher = AlwaysTransportFailingRefresher::default(); + let config = test_config(url); + let account_service = Arc::new(AccountService::new( + Arc::new(config.clone()), + AccountServiceDependencies { + repository: Some(repo.clone()), + usage_repository: Some(AccountUsageRepository::new(pool)), + cookie_repository: None, + token_refresher: Some(Arc::new(refresher.clone())), + account_pool: Arc::new(Mutex::new(AccountPool::default())), + websocket_pool: Arc::new(CodexWebSocketPool::with_default_max_age()), + fingerprint: Fingerprint::default_for_tests(), + }, + )); + let scheduler = RefreshScheduler::new(account_service, config); + tokio::time::pause(); + let running_scheduler = scheduler.clone(); + let refresh = + tokio::spawn(async move { running_scheduler.do_refresh_inner("acct_recovery").await }); + + for delay in [ + Duration::from_secs(5), + Duration::from_secs(15), + Duration::from_secs(45), + Duration::from_secs(135), + ] { + tokio::task::yield_now().await; + tokio::time::advance(delay).await; + } + + let result = refresh.await.unwrap(); + + assert!(result.is_err()); + assert_eq!( + refresher.calls.load(Ordering::SeqCst), + MAX_ATTEMPTS as usize + ); + let stored = repo.get("acct_recovery").await.unwrap().unwrap(); + assert_eq!(stored.status, AccountStatus::Active); + assert!(scheduler.timers.read().await.contains_key("acct_recovery")); + scheduler.destroy().await; + } + #[derive(Clone)] struct AlwaysInvalidGrantRefresher { repo: AccountRepository, @@ -613,6 +706,19 @@ mod tests { } } + #[derive(Clone, Default)] + struct AlwaysTransportFailingRefresher { + calls: Arc, + } + + #[async_trait] + impl TokenRefresher for AlwaysTransportFailingRefresher { + async fn refresh(&self, _refresh_token: &str) -> Result { + self.calls.fetch_add(1, Ordering::SeqCst); + Err(RefreshFailure::Transport) + } + } + fn test_config(database_url: String) -> AppConfig { AppConfig { server: ServerConfig {