From 5b7b9b21ee29227429385aabc70f041805b2012b Mon Sep 17 00:00:00 2001 From: Zyy Date: Mon, 10 Aug 2026 15:54:02 +0800 Subject: [PATCH] fix(openai): require low usage for quota recovery Co-authored-by: Codex --- .../openai/src/credential/quota/mod.rs | 65 ++++++++----- .../openai/src/credential/quota/recovery.rs | 21 +++-- .../openai/src/credential/quota/snapshot.rs | 9 ++ .../openai/tests/credential/quota/mod.rs | 34 +++---- .../openai/tests/credential/quota/recovery.rs | 94 ++++++++++++++++++- 5 files changed, 172 insertions(+), 51 deletions(-) diff --git a/backend/crates/providers/openai/src/credential/quota/mod.rs b/backend/crates/providers/openai/src/credential/quota/mod.rs index 6c216c1e3..0a5129c13 100644 --- a/backend/crates/providers/openai/src/credential/quota/mod.rs +++ b/backend/crates/providers/openai/src/credential/quota/mod.rs @@ -397,7 +397,7 @@ impl CodexCredentialQuotaService { } } - /// 写入新快照前读取已确认耗尽窗口的 reset，用于判断窗口是否真的推进。 + /// 写入新快照前读取 core primary 已确认耗尽窗口的 reset，用于判断该窗口是否真的推进。 async fn previous_confirmed_reset(&self, account: &ProviderAccount) -> Option { let observations = self .store @@ -409,7 +409,10 @@ impl CodexCredentialQuotaService { && observation.expected_revision == account.revision() })?; let snapshot = quota_snapshot_from_observation(&observation)?; - snapshot.fact().resets_at().map(SystemTime::from) + snapshot + .core_primary_window() + .and_then(|window| window.reset_at()) + .map(SystemTime::from) } /// 将已确认的额度拒绝投影为 100%，使展示、调度和恢复共用同一窗口事实。 @@ -633,7 +636,6 @@ impl CodexCredentialQuotaService { observed_at, value, )?; - let refreshed_fact = refreshed_snapshot.fact(); let previous_reset_at = self.previous_confirmed_reset(account).await; let object = normalize_quota_window_placeholders( value @@ -642,12 +644,16 @@ impl CodexCredentialQuotaService { .ok_or(CodexCredentialQuotaError::InvalidCredentialData)?, ); let object = if account.availability() == AccountAvailability::QuotaExhausted - && quota_success_state(account.availability(), refreshed_fact, previous_reset_at) - .is_none() + && quota_success_state( + account.availability(), + &refreshed_snapshot, + previous_reset_at, + ) + .is_none() { confirmed_exhaustion_projection( object, - confirmed_exhaustion_reset_at(previous_reset_at, refreshed_fact), + confirmed_exhaustion_reset_at(previous_reset_at, &refreshed_snapshot), ) } else { object @@ -687,9 +693,11 @@ impl CodexCredentialQuotaService { } else { summary.updated += 1; } - if let Some(availability) = - quota_success_state(current.availability(), refreshed_fact, previous_reset_at) - { + if let Some(availability) = quota_success_state( + current.availability(), + &refreshed_snapshot, + previous_reset_at, + ) { let _ = self .repository .apply_state(¤t, availability, observed_at) @@ -755,13 +763,18 @@ impl CodexCredentialQuotaService { } let quota = merge_passive_quota(existing, &rate_limits); let observed_at = SystemTime::now(); - let refreshed_fact = parse_codex_quota_usage(&Value::Object(quota.clone()))?; + let refreshed_snapshot = parse_account_quota_snapshot( + account.id().clone(), + account.revision(), + observed_at, + &Value::Object(quota.clone()), + )?; let quota = if account.availability() == AccountAvailability::QuotaExhausted { // 被动成功响应不能解除确认额度耗尽；继续展示 100%，直到权威 usage 快照 // 证明旧 reset 已跨过且新窗口已经推进。 confirmed_exhaustion_projection( quota, - confirmed_exhaustion_reset_at(previous_reset_at, refreshed_fact), + confirmed_exhaustion_reset_at(previous_reset_at, &refreshed_snapshot), ) } else { quota @@ -957,7 +970,6 @@ impl CodexCredentialQuotaService { observed_at, &value, )?; - let refreshed_fact = refreshed_snapshot.fact(); let previous_reset_at = self.previous_confirmed_reset(&account).await; let object = normalize_quota_window_placeholders( value @@ -966,12 +978,16 @@ impl CodexCredentialQuotaService { .ok_or(CodexCredentialQuotaError::InvalidCredentialData)?, ); let object = if account.availability() == AccountAvailability::QuotaExhausted - && quota_success_state(account.availability(), refreshed_fact, previous_reset_at) - .is_none() + && quota_success_state( + account.availability(), + &refreshed_snapshot, + previous_reset_at, + ) + .is_none() { confirmed_exhaustion_projection( object, - confirmed_exhaustion_reset_at(previous_reset_at, refreshed_fact), + confirmed_exhaustion_reset_at(previous_reset_at, &refreshed_snapshot), ) } else { object @@ -1005,9 +1021,11 @@ impl CodexCredentialQuotaService { return Err(CodexCredentialQuotaError::RevisionConflict); } self.scheduling.observe(&snapshot); - if let Some(availability) = - quota_success_state(current.availability(), refreshed_fact, previous_reset_at) - { + if let Some(availability) = quota_success_state( + current.availability(), + &refreshed_snapshot, + previous_reset_at, + ) { self.repository .apply_state(¤t, availability, observed_at) .await?; @@ -1314,11 +1332,16 @@ fn confirmed_exhaustion_projection( /// 否则保留已确认失败的旧窗口，避免 98/99% 快照反向解锁。 fn confirmed_exhaustion_reset_at( previous_reset_at: Option, - refreshed_fact: CodexQuotaFact, + refreshed_snapshot: &CodexAccountQuotaSnapshot, ) -> Option { - let refreshed_reset_at = refreshed_fact.resets_at().map(SystemTime::from); + let refreshed_reset_at = refreshed_snapshot + .core_primary_window() + .and_then(|window| window.reset_at()) + .map(SystemTime::from); let reset_at = match (previous_reset_at, refreshed_reset_at) { - (Some(previous), Some(refreshed)) if refreshed_fact.exhausted() && refreshed > previous => { + (Some(previous), Some(refreshed)) + if refreshed_snapshot.fact().exhausted() && refreshed > previous => + { refreshed } (Some(previous), _) => previous, diff --git a/backend/crates/providers/openai/src/credential/quota/recovery.rs b/backend/crates/providers/openai/src/credential/quota/recovery.rs index 7c1a13333..023c81d95 100644 --- a/backend/crates/providers/openai/src/credential/quota/recovery.rs +++ b/backend/crates/providers/openai/src/credential/quota/recovery.rs @@ -1,7 +1,7 @@ //! 确认额度耗尽后的恢复证据与 availability 转换。 //! //! `QuotaExhausted -> Ready` 只能由同一 credential revision 的权威 usage 快照证明： -//! 新快照未耗尽，且 reset 已推进到下一窗口。 +//! 新快照未耗尽、core primary reset 已推进到下一窗口，且该窗口用量低于 10%。 use std::time::SystemTime; @@ -9,25 +9,30 @@ use gateway_core::engine::credential::AccountAvailability; use reqwest::StatusCode; use serde_json::Value; -use super::snapshot::CodexQuotaFact; +use super::snapshot::CodexAccountQuotaSnapshot; use crate::transport::CodexClientError; +const RESET_RECOVERY_MAX_USED_PERCENT: f64 = 10.0; + /// 只处理 `QuotaExhausted` 的恢复。`allowed=true`、成功响应和百分比回落都不能单独覆盖 /// 一次已确认的额度拒绝；它们在 reset 未推进时可能只是上游滞后的观测。 /// -/// `previous_reset_at` 是写入新快照前读取的已确认耗尽窗口。恢复只要求权威快照的 reset -/// 指向比旧窗口更晚的新周期，不等待旧 reset 的墙钟时间到达。 +/// `previous_reset_at` 是写入新快照前读取的 core primary 已确认耗尽窗口。上游可能只 +/// 调整 reset 时间而尚未重置额度，因此还要求同一新鲜 primary 窗口的用量低于 10%。 pub(crate) fn quota_success_state( current: AccountAvailability, - fact: CodexQuotaFact, + snapshot: &CodexAccountQuotaSnapshot, previous_reset_at: Option, ) -> Option { - if current != AccountAvailability::QuotaExhausted || fact.exhausted() { + if current != AccountAvailability::QuotaExhausted || snapshot.fact().exhausted() { return None; } let previous_reset_at = previous_reset_at?; - let refreshed_reset_at = fact.resets_at().map(SystemTime::from)?; - (refreshed_reset_at > previous_reset_at).then_some(AccountAvailability::Ready) + let primary_window = snapshot.core_primary_window()?; + let refreshed_reset_at = primary_window.reset_at().map(SystemTime::from)?; + let used_percent = primary_window.used_percent()?; + (refreshed_reset_at > previous_reset_at && used_percent < RESET_RECOVERY_MAX_USED_PERCENT) + .then_some(AccountAvailability::Ready) } pub(crate) fn quota_state_transition(error: &CodexClientError) -> Option { diff --git a/backend/crates/providers/openai/src/credential/quota/snapshot.rs b/backend/crates/providers/openai/src/credential/quota/snapshot.rs index 74073d67e..5dfb9af54 100644 --- a/backend/crates/providers/openai/src/credential/quota/snapshot.rs +++ b/backend/crates/providers/openai/src/credential/quota/snapshot.rs @@ -13,6 +13,8 @@ use serde_json::{Map, Value}; use super::{CodexCredentialQuotaError, QUOTA_SCHEDULING_TTL}; +const CORE_PRIMARY_WINDOW_KEY: &str = "core-primary"; + #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub struct CodexQuotaFact { remaining_percent: Option, @@ -132,6 +134,13 @@ impl CodexAccountQuotaSnapshot { &self.windows } + #[must_use] + pub(crate) fn core_primary_window(&self) -> Option<&CodexQuotaWindow> { + self.windows + .iter() + .find(|window| window.key() == CORE_PRIMARY_WINDOW_KEY) + } + /// 投影滚动已过期窗口：`reset_at` 已过的窗口 `used_percent=0`、 /// `limit_reached=false`、`reset_at` 滚到下一个周期（reset 到期自动归零）。 /// 只作用于展示/调度投影，不写回持久化 raw JSON。 diff --git a/backend/crates/providers/openai/tests/credential/quota/mod.rs b/backend/crates/providers/openai/tests/credential/quota/mod.rs index e9e9f5abf..389f7e073 100644 --- a/backend/crates/providers/openai/tests/credential/quota/mod.rs +++ b/backend/crates/providers/openai/tests/credential/quota/mod.rs @@ -1254,7 +1254,7 @@ async fn periodic_quota_synchronization_rechecks_exhausted_accounts_without_wait } #[tokio::test] -async fn periodic_quota_synchronization_does_not_recover_from_percent_drop_before_reset() { +async fn periodic_quota_synchronization_keeps_exhaustion_at_ten_percent_after_advanced_reset() { let store = Arc::new(MemoryAccountStore::default()); let account_id = "acct_periodic_recovery_evidence"; create_account(&store, account_id).await; @@ -1333,8 +1333,8 @@ async fn periodic_quota_synchronization_does_not_recover_from_percent_drop_befor .and(path("/api/codex/usage")) .respond_with(ResponseTemplate::new(200).set_body_json(json!({ "rate_limit": { - // 无 allowed/limit_reached 标记的模糊快照。 - "primary_window": {"used_percent": 99, "reset_at": reset_at} + // reset 已后移，但 10% 仍不能证明这是刚重置的新窗口。 + "primary_window": {"used_percent": 10, "reset_at": reset_at + 3600} } }))) .expect(1) @@ -1350,19 +1350,19 @@ async fn periodic_quota_synchronization_does_not_recover_from_percent_drop_befor server.uri(), ); - let lower_percent = next_cycle + let ten_percent = next_cycle .synchronize() .await - .expect("lower percent quota recheck"); + .expect("ten-percent quota recheck"); - assert_eq!(lower_percent.exhausted, 1); + assert_eq!(ten_percent.exhausted, 1); assert_eq!( store .account(account_id) - .expect("account after lower percent snapshot") + .expect("account after ten-percent snapshot") .availability(), AccountAvailability::QuotaExhausted, - "a provider percentage change is not sufficient to release a confirmed exhaustion" + "an advanced reset at exactly 10% is not sufficient to release confirmed exhaustion" ); let snapshot = next_cycle .read_account(account.id()) @@ -1373,12 +1373,12 @@ async fn periodic_quota_synchronization_does_not_recover_from_percent_drop_befor } #[tokio::test] -async fn periodic_quota_synchronization_recovers_after_recorded_reset() { +async fn periodic_quota_synchronization_recovers_with_advanced_reset_and_low_primary_usage() { let store = Arc::new(MemoryAccountStore::default()); let account_id = "acct_periodic_reset_elapsed"; create_account(&store, account_id).await; let account = store.account(account_id).expect("created account"); - let elapsed_reset = Utc::now().timestamp() - 60; + let old_reset = Utc::now().timestamp() + 3_600; store .compare_and_swap_quota(QuotaObservation { account_id: account.id().clone(), @@ -1388,7 +1388,7 @@ async fn periodic_quota_synchronization_recovers_after_recorded_reset() { "rate_limit": { "allowed": false, "limit_reached": true, - "primary_window": {"used_percent": 100, "reset_at": elapsed_reset} + "primary_window": {"used_percent": 100, "reset_at": old_reset} } }) .as_object() @@ -1399,7 +1399,7 @@ async fn periodic_quota_synchronization_recovers_after_recorded_reset() { limit_reached: Some(true), }) .await - .expect("record elapsed exhausted reset"); + .expect("record future exhausted reset"); store .apply_state_change(AccountStateChange { message: None, @@ -1409,9 +1409,9 @@ async fn periodic_quota_synchronization_recovers_after_recorded_reset() { observed_at: SystemTime::now(), }) .await - .expect("mark account exhausted with an elapsed reset"); + .expect("mark account exhausted with a future reset"); let server = MockServer::start().await; - let next_reset = Utc::now().timestamp() + 3600; + let next_reset = old_reset + 3_600; Mock::given(method("GET")) .and(path("/api/codex/usage")) .respond_with(ResponseTemplate::new(200).set_body_json(json!({ @@ -1433,7 +1433,7 @@ async fn periodic_quota_synchronization_recovers_after_recorded_reset() { server.uri(), ); - // 定时复核可提前执行，但只有已记录的 reset 到期后才释放额度耗尽状态。 + // 定时复核可在旧 reset 前执行；新主窗口已推进且用量低于 10% 才释放额度耗尽状态。 let summary = service.synchronize().await.expect("quota synchronization"); assert_eq!(summary.updated, 1); @@ -1585,7 +1585,7 @@ async fn invalid_quota_json_for_one_account_does_not_abort_the_batch() { .respond_with(ResponseTemplate::new(200).set_body_json(json!({ "rate_limit": { "allowed": true, - "primary_window": {"used_percent": 25.0, "reset_at": next_reset} + "primary_window": {"used_percent": 1.0, "reset_at": next_reset} } }))) .mount(&server) @@ -1614,5 +1614,5 @@ async fn invalid_quota_json_for_one_account_does_not_abort_the_batch() { .await .expect("read good quota") .expect("good quota persisted"); - assert_eq!(snapshot.fact().remaining_percent(), Some(75)); + assert_eq!(snapshot.fact().remaining_percent(), Some(99)); } diff --git a/backend/crates/providers/openai/tests/credential/quota/recovery.rs b/backend/crates/providers/openai/tests/credential/quota/recovery.rs index 333d88aeb..5e72bbe96 100644 --- a/backend/crates/providers/openai/tests/credential/quota/recovery.rs +++ b/backend/crates/providers/openai/tests/credential/quota/recovery.rs @@ -1,6 +1,6 @@ //! 402 恢复回归。 //! -//! 覆盖 `QuotaExhausted -> Ready` 的证据规则：新 reset 后移 + 新快照未耗尽。 +//! 覆盖 `QuotaExhausted -> Ready` 的证据规则：新 core primary reset 后移 + 新快照用量低于 10%。 use super::*; @@ -127,8 +127,8 @@ async fn manual_quota_refresh_keeps_confirmed_exhaustion_until_the_recorded_rese } #[tokio::test] -async fn advanced_reset_recovers_ready_before_the_recorded_reset_time() { - // 旧 reset 仍在未来；只要新窗口 reset 后移且额度正常，仍必须恢复 Ready。 +async fn advanced_reset_with_primary_usage_below_ten_percent_recovers_ready() { + // 旧 reset 仍在未来；同一主窗口 reset 后移且用量低于 10% 时才恢复 Ready。 let store = Arc::new(MemoryAccountStore::default()); let account_id = "acct_cross_window_recovery"; create_account(&store, account_id).await; @@ -181,7 +181,7 @@ async fn advanced_reset_recovers_ready_before_the_recorded_reset_time() { "rate_limit": { "allowed": true, "limit_reached": false, - "primary_window": {"used_percent": 5, "reset_at": future_reset} + "primary_window": {"used_percent": 9.9, "reset_at": future_reset} } }))) .mount(&server) @@ -207,7 +207,91 @@ async fn advanced_reset_recovers_ready_before_the_recorded_reset_time() { .expect("account after cross-window recovery") .availability(), AccountAvailability::Ready, - "an advanced reset must recover QuotaExhausted before the recorded reset time" + "an advanced reset with a nearly unused primary window must recover QuotaExhausted" + ); +} + +#[tokio::test] +async fn advanced_reset_with_primary_usage_at_ten_percent_keeps_confirmed_exhaustion() { + let store = Arc::new(MemoryAccountStore::default()); + let account_id = "acct_cross_window_ten_percent"; + create_account(&store, account_id).await; + let account = store.account(account_id).expect("created account"); + let old_reset = Utc::now().timestamp() + 3_600; + store + .compare_and_swap_quota(QuotaObservation { + account_id: account.id().clone(), + expected_revision: account.revision(), + quota: Some(OpaqueProviderData::new( + json!({ + "rate_limit": { + "allowed": false, + "limit_reached": true, + "primary_window": { + "used_percent": 100, + "reset_at": old_reset, + "limit_window_seconds": 3600 + } + } + }) + .as_object() + .expect("quota object") + .clone(), + )), + observed_at: Some(SystemTime::now()), + limit_reached: None, + }) + .await + .expect("seed old exhausted snapshot"); + store + .apply_state_change(AccountStateChange { + message: None, + account_id: account.id().clone(), + expected_revision: account.revision(), + availability: AccountAvailability::QuotaExhausted, + observed_at: SystemTime::now(), + }) + .await + .expect("mark account exhausted"); + let server = MockServer::start().await; + let future_reset = old_reset + 3_600; + Mock::given(method("GET")) + .and(path("/api/codex/usage")) + .and(header( + "authorization", + format!("Bearer token-{account_id}"), + )) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "rate_limit": { + "allowed": true, + "limit_reached": false, + "primary_window": {"used_percent": 10.0, "reset_at": future_reset} + } + }))) + .mount(&server) + .await; + let service = quota_service_with_base_url( + &store, + reqwest::Client::builder() + .no_proxy() + .build() + .expect("client"), + server.uri(), + ); + + let held = service + .refresh_account(account.id()) + .await + .expect("refresh non-empty new window quota"); + + assert!(held.fact().exhausted()); + assert_eq!( + store + .account(account_id) + .expect("account after ten-percent snapshot") + .availability(), + AccountAvailability::QuotaExhausted, + "a reset advance at exactly 10% must not release confirmed exhaustion" ); }