codex: recover refresh failures

Co-authored-by: Codex <noreply@openai.com>
2026-11-06 16:06:08 +00:00 · 2026-06-15 11:26:46 +08:00
parent aa060bfec4
commit cce1233aa7
3 changed files with 110 additions and 2 deletions
@@ -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"] }
@@ -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

@@ -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<u64> {
        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<AtomicUsize>,
+    }
+
+    #[async_trait]
+    impl TokenRefresher for AlwaysTransportFailingRefresher {
+        async fn refresh(&self, _refresh_token: &str) -> Result<TokenPair, RefreshFailure> {
+            self.calls.fetch_add(1, Ordering::SeqCst);
+            Err(RefreshFailure::Transport)
+        }
+    }
+
    fn test_config(database_url: String) -> AppConfig {
        AppConfig {
            server: ServerConfig {