From cef9f0b21e714c45230364e8cd6d41db81fc0cb4 Mon Sep 17 00:00:00 2001 From: Zyy <0x@ainz.cc> Date: Sun, 18 Oct 2026 22:45:15 +0800 Subject: [PATCH] fix(openai): honor usable credits and recover quota after reset (#469) Refs #465 Co-authored-by: Zyy Co-authored-by: Codex --- .../openai/src/credential/quota/mod.rs | 4 + .../openai/src/credential/quota/recovery.rs | 27 ++- .../openai/src/credential/quota/scheduling.rs | 48 ++-- .../openai/src/credential/quota/service.rs | 18 +- .../openai/src/credential/quota/snapshot.rs | 38 +++- .../openai/tests/credential/quota/credits.rs | 215 ++++++++++++++++++ .../openai/tests/credential/quota/mod.rs | 2 + .../tests/credential/quota/refresh_timing.rs | 167 ++++++++++++-- .../tests/credential/quota/reset_recovery.rs | 160 +++++++++++++ .../openai/tests/provider/contract/mod.rs | 76 +++++++ .../crates/providers/openai/tests/support.rs | 15 +- docs/api.md | 18 +- docs/architecture.md | 6 + 13 files changed, 735 insertions(+), 59 deletions(-) create mode 100644 backend/crates/providers/openai/tests/credential/quota/credits.rs create mode 100644 backend/crates/providers/openai/tests/credential/quota/reset_recovery.rs diff --git a/backend/crates/providers/openai/src/credential/quota/mod.rs b/backend/crates/providers/openai/src/credential/quota/mod.rs index 387c7119f..45980b1e1 100644 --- a/backend/crates/providers/openai/src/credential/quota/mod.rs +++ b/backend/crates/providers/openai/src/credential/quota/mod.rs @@ -73,6 +73,10 @@ const DEFAULT_RATE_LIMIT_COOLDOWN: Duration = Duration::from_secs(60); pub(crate) const QUOTA_SCHEDULING_TTL: Duration = Duration::from_secs(10 * 60); const QUOTA_HYDRATION_FAILURE_TTL: Duration = Duration::from_secs(5); const PERIODIC_QUOTA_REFRESH_RETRY_INTERVAL: Duration = Duration::from_secs(30 * 60); +const QUOTA_RESET_FAST_REFRESH_WINDOW: Duration = Duration::from_secs(5 * 60); +const QUOTA_RESET_FAST_REFRESH_INTERVAL: Duration = Duration::from_secs(30); +const QUOTA_RESET_REFRESH_RETRY_INTERVAL: Duration = Duration::from_secs(5 * 60); +const EXHAUSTED_QUOTA_RESET_GRACE: Duration = Duration::from_secs(10); const QUOTA_RESET_GRACE: Duration = Duration::from_secs(2 * 60); /// 首次 OAuth 异步观察失败时，由既有 quota worker 兜底重试的单轮上限 /// 随机启动延迟已把候选摊到多轮，该上限仅作单轮安全边界 diff --git a/backend/crates/providers/openai/src/credential/quota/recovery.rs b/backend/crates/providers/openai/src/credential/quota/recovery.rs index 703ec1ac3..23490e2e8 100644 --- a/backend/crates/providers/openai/src/credential/quota/recovery.rs +++ b/backend/crates/providers/openai/src/credential/quota/recovery.rs @@ -4,7 +4,7 @@ use std::collections::BTreeMap; use std::time::SystemTime; use chrono::{DateTime, Utc}; -use gateway_core::account::{QuotaEvidence, QuotaState}; +use gateway_core::account::{QuotaAccessState, QuotaEvidence, QuotaState}; use serde::{Deserialize, Serialize}; use serde_json::{Map, Value}; @@ -127,11 +127,8 @@ fn window_reset_recovered(window: &CodexQuotaWindow, previous: Option, ) -> Result { document.remove(RECOVERY_FIELD); + if refreshed.credits_usable() { + // 点数独立于套餐窗口，恢复后不能继续携带已失效的窗口恢复进度 + refreshed.recovery = None; + return Ok(QuotaState::allowed(refreshed.observed_at())); + } let state = if current.is_exhausted() { current } else { @@ -155,7 +157,18 @@ pub(super) fn reconcile_refresh( if recovery.pending.is_empty() { recovery = QuotaRecovery::new(state, refreshed); } - if current.is_exhausted() && recovery.observe(refreshed) { + // 只信任本次耗尽之后的边界，防止再次拒绝后复用已经过期的 reset 反复解封 + // 多个耗尽窗口必须全部到期，未知 reset 仍走原有恢复规则 + let allowed_after_reset = refreshed.quota().access() == QuotaAccessState::Allowed + && !recovery.pending.is_empty() + && recovery.pending.values().all(|reset| { + reset.is_some_and(|reset| { + reset.timestamp_micros() > recovery.exhausted_at_micros + && SystemTime::from(reset) <= refreshed.observed_at() + }) + }); + if current.is_exhausted() && (allowed_after_reset || recovery.observe(refreshed)) { + refreshed.recovery = None; return Ok(QuotaState::allowed(refreshed.observed_at())); } let reset_at = recovery diff --git a/backend/crates/providers/openai/src/credential/quota/scheduling.rs b/backend/crates/providers/openai/src/credential/quota/scheduling.rs index 4b3226ea8..2651242f3 100644 --- a/backend/crates/providers/openai/src/credential/quota/scheduling.rs +++ b/backend/crates/providers/openai/src/credential/quota/scheduling.rs @@ -1,6 +1,7 @@ //! 请求级额度投影、首查与周期刷新排程 use super::*; +use tokio::time::Instant as TokioInstant; impl CodexCredentialQuotaService { /// 批量预热请求级额度投影；持久层或 Provider JSON 异常只退化为未知额度 @@ -67,7 +68,7 @@ struct CodexQuotaProjectionState { } struct CodexQuotaRefreshAttempt { - monotonic_at: Instant, + monotonic_at: TokioInstant, wall_at: SystemTime, } @@ -230,7 +231,7 @@ impl CodexQuotaSchedulingProjection { .iter() .map(|(account, _)| account.id().clone()) .collect::>(); - let refreshed_at = Instant::now(); + let refreshed_at = TokioInstant::now(); let mut state = self .state .write() @@ -239,12 +240,11 @@ impl CodexQuotaSchedulingProjection { .last_periodic_refresh_at .retain(|account_id, _| candidate_ids.contains(account_id)); - // 已耗尽账号首次立即复核，之后每 30 分钟复核，以发现官方提前重置 - // reset + 2 分钟额外触发一次复核，给上游重置留出传播时间 + // 已耗尽账号到达 reset 后短周期复核，其余时间保留 30 分钟兜底 // 正常账号仅在非零用量窗口经过宽限期后参与，未更新时复用周期节流 let mut reserved = Vec::new(); for (account, target_reset) in candidates { - if !periodic_quota_refresh_due(&state, account.id(), target_reset, now, refreshed_at) { + if !periodic_quota_refresh_due(&state, &account, target_reset, now, refreshed_at) { continue; } state.last_periodic_refresh_at.insert( @@ -364,22 +364,34 @@ fn quota_refresh_candidate( fn periodic_quota_refresh_due( state: &CodexQuotaProjectionState, - account_id: &ProviderAccountId, + account: &ProviderAccount, reset_at: Option, now: SystemTime, - monotonic_now: Instant, + monotonic_now: TokioInstant, ) -> bool { - state - .last_periodic_refresh_at - .get(account_id) - .is_none_or(|last| { - monotonic_now.saturating_duration_since(last.monotonic_at) - >= PERIODIC_QUOTA_REFRESH_RETRY_INTERVAL - || reset_at - .and_then(|reset| reset.checked_add(QUOTA_RESET_GRACE)) - // 已在该边界之后复核过时回到周期重试，避免过期 reset 每轮触发 - .is_some_and(|due_at| last.wall_at < due_at && due_at <= now) - }) + let mut interval = PERIODIC_QUOTA_REFRESH_RETRY_INTERVAL; + let due_at = if account.quota().is_exhausted() { + if let Some(since_reset) = reset_at.and_then(|reset| now.duration_since(reset).ok()) { + // 上游边界附近可能仍返回旧额度，首次复核也保留传播余量 + if since_reset < EXHAUSTED_QUOTA_RESET_GRACE { + return false; + } + interval = if since_reset < QUOTA_RESET_FAST_REFRESH_WINDOW { + QUOTA_RESET_FAST_REFRESH_INTERVAL + } else { + QUOTA_RESET_REFRESH_RETRY_INTERVAL + }; + } + reset_at.and_then(|reset| reset.checked_add(EXHAUSTED_QUOTA_RESET_GRACE)) + } else { + reset_at.and_then(|reset| reset.checked_add(QUOTA_RESET_GRACE)) + }; + let Some(last) = state.last_periodic_refresh_at.get(account.id()) else { + return true; + }; + monotonic_now.saturating_duration_since(last.monotonic_at) >= interval + // 跨越 reset 宽限期的首轮复核不受旧间隔阻挡，之后按对应间隔节流 + || due_at.is_some_and(|due_at| last.wall_at < due_at && due_at <= now) } fn insert_projection_entry( diff --git a/backend/crates/providers/openai/src/credential/quota/service.rs b/backend/crates/providers/openai/src/credential/quota/service.rs index 53e48d5ad..b714196cc 100644 --- a/backend/crates/providers/openai/src/credential/quota/service.rs +++ b/backend/crates/providers/openai/src/credential/quota/service.rs @@ -515,7 +515,7 @@ impl CodexCredentialQuotaService { .await } - /// 真实限额失败后的异步刷新只补齐展示快照，不允许 usage 快照撤销已确认的失败状态 + /// 真实限额失败后的异步回读仅允许明确可用的点数恢复访问，套餐状态仍保留失败保护 pub(crate) async fn refresh_account_after_failure( &self, account_id: &ProviderAccountId, @@ -589,9 +589,10 @@ impl CodexCredentialQuotaService { } else { None }; - if account.quota().is_exhausted() - && matches!(authority, QuotaRefreshAuthority::PreserveAccess) - { + // 失败后回读只允许明确可用的点数恢复访问，套餐 allowed 仍可能滞后 + let observe_access = + matches!(authority, QuotaRefreshAuthority::ObserveAccess) || snapshot.credits_usable(); + if account.quota().is_exhausted() && !observe_access { if self .store .touch_quota_observation(QuotaObservationTouch { @@ -615,14 +616,15 @@ impl CodexCredentialQuotaService { .with_observed_at(observed_at), }); } - let state = match authority { - QuotaRefreshAuthority::ObserveAccess => reconcile_refresh( + let state = if observe_access { + reconcile_refresh( account.quota(), &mut snapshot, previous.as_ref(), &mut object, - )?, - QuotaRefreshAuthority::PreserveAccess => account.quota(), + )? + } else { + account.quota() }; let snapshot = snapshot.with_quota_state(state); if self diff --git a/backend/crates/providers/openai/src/credential/quota/snapshot.rs b/backend/crates/providers/openai/src/credential/quota/snapshot.rs index 499d7e7bd..175381012 100644 --- a/backend/crates/providers/openai/src/credential/quota/snapshot.rs +++ b/backend/crates/providers/openai/src/credential/quota/snapshot.rs @@ -116,6 +116,7 @@ pub struct CodexAccountQuotaSnapshot { observed_at: SystemTime, plan_type: Option, credits: Option, + credits_usable: bool, fact: CodexQuotaFact, quota: QuotaState, windows: Vec, @@ -144,12 +145,16 @@ impl CodexAccountQuotaSnapshot { self.plan_type.as_deref() } - /// 点数只用于展示，不改变套餐限额或账号可用性 + /// 上游明确返回的点数事实；点数可用性独立于套餐窗口 #[must_use] pub fn credits(&self) -> Option<&CreditsSnapshot> { self.credits.as_ref() } + pub(super) const fn credits_usable(&self) -> bool { + self.credits_usable + } + #[must_use] pub const fn fact(&self) -> CodexQuotaFact { self.fact @@ -209,7 +214,7 @@ fn parse_codex_quota_object( for limit in canonical_rate_limits(object)? { aggregate.observe_rate_limit(limit.rate_limit)?; } - // credits / spend_control 只证明文档结构已识别，不参与账号级访问结论 + // 点数与支出控制不换算成套餐剩余百分比，访问结论由账号快照统一判断 if let Some(spend_control) = object.get("spend_control") { aggregate.observe_metadata_object(spend_control, "reached")?; } @@ -345,6 +350,12 @@ pub(crate) fn parse_account_quota_snapshot( .then_with(|| left.key.cmp(&right.key)) }); let quota = authoritative_quota_state(&object, &windows, observed_at)?; + let credits_usable = has_usable_credits(&object); + let quota = if credits_usable { + QuotaState::allowed(observed_at) + } else { + quota + }; let recovery = object .get(RECOVERY_FIELD) .map(|value| serde_json::from_value(value.clone())) @@ -359,6 +370,7 @@ pub(crate) fn parse_account_quota_snapshot( .and_then(Value::as_str) .map(str::to_owned), credits: object.get("credits").and_then(parse_credits_from_object), + credits_usable, fact, quota, windows, @@ -374,7 +386,27 @@ fn quota_source_order(source: &str) -> u8 { } } -/// 默认 `codex` 桶是唯一可改变账号额度访问结论的 quota 响应字段 +fn has_usable_credits(usage: &Map) -> bool { + let Some(credits) = usage.get("credits") else { + return false; + }; + let available = credits.get("has_credits").and_then(Value::as_bool) == Some(true) + || credits.get("unlimited").and_then(Value::as_bool) == Some(true); + // 套餐耗尽不阻止点数支出，但不能绕过显式支出上限或工作区限制 + let spend_reached = usage + .get("spend_control") + .and_then(|spend| spend.get("reached")) + .and_then(Value::as_bool) + == Some(true); + let other_limit_reached = usage + .get("rate_limit_reached_type") + .and_then(|reason| reason.get("type")) + .and_then(Value::as_str) + .is_some_and(|reason| reason != "rate_limit_reached"); + available && !spend_reached && !other_limit_reached +} + +/// 只有默认 `codex` 桶能提供账号级套餐访问结论，具名桶只保留展示事实 fn authoritative_quota_state( usage: &Map, windows: &[CodexQuotaWindow], diff --git a/backend/crates/providers/openai/tests/credential/quota/credits.rs b/backend/crates/providers/openai/tests/credential/quota/credits.rs new file mode 100644 index 000000000..5bf24afff --- /dev/null +++ b/backend/crates/providers/openai/tests/credential/quota/credits.rs @@ -0,0 +1,215 @@ +//! 点数独立可用性、耗尽恢复和过期查询隔离 + +use super::*; + +fn exhausted_usage() -> serde_json::Value { + json!({"rate_limit": { + "allowed": false, + "limit_reached": true, + "primary_window": {"used_percent": 100, "reset_at": 1_900_000_000}, + "secondary_window": {"used_percent": 100, "reset_at": 1_900_604_800} + }}) +} + +async fn mount_usage(server: &MockServer, usage: serde_json::Value) { + server.reset().await; + Mock::given(method("GET")) + .and(path("/api/codex/usage")) + .respond_with(ResponseTemplate::new(200).set_body_json(usage)) + .mount(server) + .await; +} + +#[tokio::test] +async fn credits_allow_exhausted_subscription_without_bypassing_spend_restrictions() { + for (metadata, allowed) in [ + ( + json!({"credits": {"has_credits": true, "unlimited": false}}), + true, + ), + ( + json!({"credits": {"has_credits": false, "unlimited": true}}), + true, + ), + (json!({"credits": {"has_credits": true}}), true), + (json!({"credits": {"unlimited": true}}), true), + ( + json!({"credits": {"has_credits": false, "unlimited": false, "balance": "500"}}), + false, + ), + (json!({"credits": null}), false), + ( + json!({"credits": {"has_credits": true}, "spend_control": {"reached": true}}), + false, + ), + ( + json!({"credits": {"has_credits": true}, "rate_limit_reached_type": {"type": "spend_limit_reached"}}), + false, + ), + ( + json!({"credits": {"has_credits": true}, "rate_limit_reached_type": {"type": "rate_limit_reached"}}), + true, + ), + ] { + let store = Arc::new(MemoryAccountStore::default()); + create_account_with_enabled(&store, "acct_credits", false).await; + let account = store.account("acct_credits").expect("account"); + persist_credential_state(&store, &account, CredentialState::Expired).await; + let server = MockServer::start().await; + let service = quota_service_with_base_url(&store, reqwest::Client::new(), server.uri()); + let mut usage = exhausted_usage(); + usage + .as_object_mut() + .expect("usage") + .extend(metadata.as_object().expect("metadata").clone()); + mount_usage(&server, usage).await; + + let snapshot = service + .refresh_account(account.id()) + .await + .expect("refresh credits"); + assert_eq!( + snapshot.quota().access() == QuotaAccessState::Allowed, + allowed, + "{metadata}" + ); + assert_eq!(snapshot.fact().remaining_percent(), Some(0)); + let current = store.account("acct_credits").expect("current account"); + assert_eq!(current.quota().is_exhausted(), !allowed, "{metadata}"); + assert!(!current.enabled()); + assert_eq!(current.credential_state(), CredentialState::Expired); + } +} + +#[tokio::test] +async fn credits_refresh_clears_old_recovery_and_locks_again_only_when_spent() { + for periodic in [false, true] { + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, "acct_credits_recovery").await; + let account = store.account("acct_credits_recovery").expect("account"); + let server = MockServer::start().await; + let service = quota_service_with_base_url(&store, reqwest::Client::new(), server.uri()); + mount_usage(&server, exhausted_usage()).await; + service + .refresh_account(account.id()) + .await + .expect("seed exhaustion"); + assert!( + store + .quota_json("acct_credits_recovery") + .expect("quota") + .get("_quota_recovery") + .is_some() + ); + + let mut usage = exhausted_usage(); + usage["credits"] = + json!({"has_credits": !periodic, "unlimited": periodic, "balance": "20"}); + mount_usage(&server, usage.clone()).await; + if periodic { + assert_eq!( + service + .synchronize() + .await + .expect("periodic recovery") + .updated, + 1 + ); + } else { + service + .refresh_account(account.id()) + .await + .expect("manual recovery"); + } + let snapshot = service + .read_account(account.id()) + .await + .expect("persisted snapshot") + .expect("snapshot"); + assert_eq!(snapshot.quota().access(), QuotaAccessState::Allowed); + assert_eq!(snapshot.fact().remaining_percent(), Some(0)); + assert_eq!( + snapshot.credits().expect("credits").balance.as_deref(), + Some("20") + ); + assert!( + store + .quota_json("acct_credits_recovery") + .expect("quota") + .get("_quota_recovery") + .is_none() + ); + + usage["credits"] = json!({"has_credits": false, "unlimited": false, "balance": "0"}); + mount_usage(&server, usage).await; + let snapshot = service + .refresh_account(account.id()) + .await + .expect("spent credits"); + assert!(snapshot.quota().is_exhausted()); + assert!( + store + .account("acct_credits_recovery") + .expect("account") + .quota() + .is_exhausted() + ); + } +} + +#[tokio::test] +async fn credits_query_started_before_new_exhaustion_cannot_restore_access() { + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, "acct_stale_credits").await; + let account = store.account("acct_stale_credits").expect("account"); + let server = MockServer::start().await; + let service = quota_service_with_base_url(&store, reqwest::Client::new(), server.uri()); + let mut usage = exhausted_usage(); + usage["credits"] = json!({"has_credits": true, "unlimited": false}); + Mock::given(method("GET")) + .and(path("/api/codex/usage")) + .respond_with( + ResponseTemplate::new(200) + .set_delay(Duration::from_millis(200)) + .set_body_json(usage), + ) + .mount(&server) + .await; + let (result, ()) = tokio::join!(service.refresh_account(account.id()), async { + tokio::time::timeout(Duration::from_secs(2), async { + while server + .received_requests() + .await + .expect("requests") + .is_empty() + { + tokio::time::sleep(Duration::from_millis(1)).await; + } + }) + .await + .expect("query started"); + persist_quota_state( + &store, + &account, + QuotaState::exhausted(QuotaEvidence::UsageLimitReached, SystemTime::now(), None), + ) + .await; + }); + result.expect("old display observation can persist independently of newer access state"); + assert!( + store + .account("acct_stale_credits") + .expect("account") + .quota() + .is_exhausted() + ); + assert!( + service + .read_account(account.id()) + .await + .expect("persisted quota") + .expect("snapshot") + .quota() + .is_exhausted() + ); +} diff --git a/backend/crates/providers/openai/tests/credential/quota/mod.rs b/backend/crates/providers/openai/tests/credential/quota/mod.rs index 15972114a..3d740cb31 100644 --- a/backend/crates/providers/openai/tests/credential/quota/mod.rs +++ b/backend/crates/providers/openai/tests/credential/quota/mod.rs @@ -1,10 +1,12 @@ //! OpenAI 额度事实边界与展示快照回归 mod capacity_freeze; +mod credits; mod initial_sync; mod recovery; mod refresh_timing; mod reset_credits; +mod reset_recovery; mod scheduling; mod slots; mod snapshot; diff --git a/backend/crates/providers/openai/tests/credential/quota/refresh_timing.rs b/backend/crates/providers/openai/tests/credential/quota/refresh_timing.rs index c2f1049cc..44058a4d2 100644 --- a/backend/crates/providers/openai/tests/credential/quota/refresh_timing.rs +++ b/backend/crates/providers/openai/tests/credential/quota/refresh_timing.rs @@ -1,4 +1,4 @@ -//! 通过 quota 服务验证周期复核与 reset 宽限期，不依赖内部调度状态 +//! 通过 quota 服务验证耗尽复核节奏、恢复与正常窗口宽限期 use std::time::Duration; @@ -17,8 +17,17 @@ fn reset_grace(reset: i64) -> SystemTime { SystemTime::UNIX_EPOCH + Duration::from_secs((reset + 120) as u64) } +async fn advance_clock(now: &mut SystemTime, seconds: u64) { + let duration = Duration::from_secs(seconds); + // 只在推进时暂停 Tokio 时钟，HTTP 请求期间保持真实计时 + tokio::time::pause(); + tokio::time::advance(duration).await; + tokio::time::resume(); + *now += duration; +} + #[tokio::test] -async fn reset_grace_bypasses_periodic_throttle_once_then_allows_the_next_window() { +async fn exhausted_reset_bypasses_periodic_throttle_and_rechecks_each_pending_window() { let mut now = SystemTime::now(); let store = Arc::new(MemoryAccountStore::default()); create_account(&store, "acct_quota_timing").await; @@ -30,8 +39,8 @@ async fn reset_grace_bypasses_periodic_throttle_once_then_allows_the_next_window .duration_since(SystemTime::UNIX_EPOCH) .unwrap() .as_secs() as i64 - - 115; - let week_reset = short_reset + 3; + + 5; + let week_reset = short_reset + 30; let usage = |short_used, short_reset, week_used, week_reset| { json!({"rate_limit": { "allowed": false, @@ -55,20 +64,28 @@ async fn reset_grace_bypasses_periodic_throttle_once_then_allows_the_next_window let requests = server.received_requests().await.expect("requests").len(); assert!( - now < reset_grace(short_reset), - "fixture must precede grace deadline" + now < SystemTime::UNIX_EPOCH + Duration::from_secs(short_reset as u64), + "fixture must precede reset" ); - service - .synchronize_at(now) - .await - .expect("before grace deadline"); + service.synchronize_at(now).await.expect("before reset"); assert_eq!( server.received_requests().await.expect("requests").len(), requests ); mount_usage(&server, usage(0, short_reset + 18_000, 100, week_reset)).await; - now = reset_grace(short_reset); + for delay in [0, 9] { + now = SystemTime::UNIX_EPOCH + Duration::from_secs(short_reset as u64 + delay); + service.synchronize_at(now).await.expect("reset grace"); + assert!( + server + .received_requests() + .await + .expect("requests") + .is_empty() + ); + } + now = SystemTime::UNIX_EPOCH + Duration::from_secs(short_reset as u64 + 10); let partial = service .synchronize_at(now) .await @@ -85,11 +102,11 @@ async fn reset_grace_bypasses_periodic_throttle_once_then_allows_the_next_window service .synchronize_at(now) .await - .expect("next scan before weekly grace deadline"); + .expect("next scan before weekly reset"); assert_eq!(server.received_requests().await.expect("requests").len(), 1); // 周窗口的到期复核仍未恢复时，也不能每轮扫描重复请求 - now = reset_grace(week_reset); + now = SystemTime::UNIX_EPOCH + Duration::from_secs(week_reset as u64 + 10); assert_eq!( service .synchronize_at(now) @@ -127,19 +144,139 @@ async fn periodic_checks_continue_when_reset_is_unknown_or_far_in_the_future() { .expect("seed exhaustion"); mount_usage(&server, usage).await; + let mut now = SystemTime::now(); assert_eq!( service - .synchronize() + .synchronize_at(now) .await .expect("periodic check") .exhausted, 1 ); - service.synchronize().await.expect("throttled repeat check"); + advance_clock(&mut now, 29 * 60).await; + service + .synchronize_at(now) + .await + .expect("throttled repeat check"); assert_eq!(server.received_requests().await.expect("requests").len(), 1); + advance_clock(&mut now, 60).await; + service.synchronize_at(now).await.expect("periodic retry"); + assert_eq!(server.received_requests().await.expect("requests").len(), 2); } } +#[tokio::test] +async fn exhausted_reset_refreshes_quickly_then_backs_off_and_recovers_from_usage() { + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, "acct_reset_retry").await; + let account = store.account("acct_reset_retry").expect("account"); + let server = MockServer::start().await; + let service = quota_service_with_base_url(&store, reqwest::Client::new(), server.uri()); + let reset = Utc::now().timestamp(); + let mut now = SystemTime::UNIX_EPOCH + Duration::from_secs(reset as u64); + let usage = |allowed, used, reset| { + json!({"rate_limit": { + "allowed": allowed, + "primary_window": {"used_percent": used, "reset_at": reset} + }}) + }; + mount_usage(&server, usage(false, 100, reset)).await; + service + .refresh_account(account.id()) + .await + .expect("seed exhaustion"); + service + .synchronize_at(now) + .await + .expect("reset grace without periodic history"); + advance_clock(&mut now, 9).await; + service + .synchronize_at(now) + .await + .expect("still in reset grace"); + assert_eq!(server.received_requests().await.expect("requests").len(), 1); + advance_clock(&mut now, 1).await; + service + .synchronize_at(now) + .await + .expect("initial reset check"); + assert_eq!(server.received_requests().await.expect("requests").len(), 2); + + // 本次拒绝发生在 reset 之后，滞后的 allowed 不能复用旧边界解封 + mount_usage(&server, usage(true, 100, reset)).await; + for (elapsed, requests) in [ + (29, 0), + (1, 1), + (29, 1), + (1, 2), + (229, 3), + (1, 3), + (298, 3), + (1, 4), + ] { + advance_clock(&mut now, elapsed).await; + service.synchronize_at(now).await.expect("scheduled check"); + assert_eq!( + server.received_requests().await.expect("requests").len(), + requests + ); + assert!( + store + .account("acct_reset_retry") + .expect("account") + .quota() + .is_exhausted() + ); + } + + // 查询失败也占用一次尝试，避免网络故障时取消节流 + server.reset().await; + Mock::given(method("GET")) + .and(path("/api/codex/usage")) + .respond_with(ResponseTemplate::new(429)) + .mount(&server) + .await; + advance_clock(&mut now, 300).await; + assert_eq!( + service + .synchronize_at(now) + .await + .expect("failed check") + .transient, + 1 + ); + advance_clock(&mut now, 299).await; + service + .synchronize_at(now) + .await + .expect("throttled failed check"); + assert_eq!(server.received_requests().await.expect("requests").len(), 1); + + mount_usage(&server, usage(true, 0, reset + 18_000)).await; + advance_clock(&mut now, 1).await; + assert_eq!( + service + .synchronize_at(now) + .await + .expect("recovery check") + .updated, + 1 + ); + assert!( + !store + .account("acct_reset_retry") + .expect("account") + .quota() + .is_exhausted() + ); + advance_clock(&mut now, 300).await; + service + .synchronize_at(now) + .await + .expect("recovered account leaves retry schedule"); + assert_eq!(server.received_requests().await.expect("requests").len(), 1); +} + #[tokio::test] async fn allowed_account_with_expired_window_synchronizes_at_reset_grace() { let mut now = SystemTime::now(); diff --git a/backend/crates/providers/openai/tests/credential/quota/reset_recovery.rs b/backend/crates/providers/openai/tests/credential/quota/reset_recovery.rs new file mode 100644 index 000000000..bba2fd1b7 --- /dev/null +++ b/backend/crates/providers/openai/tests/credential/quota/reset_recovery.rs @@ -0,0 +1,160 @@ +//! 到期后的明确放行恢复，保留到期前、多窗口与再次拒绝的保护 + +use gateway_core::account::{OpaqueProviderData, QuotaObservation}; + +use super::*; + +fn usage(now: i64, resets: &[Option], allowed: Option) -> serde_json::Value { + let mut value = json!({"rate_limit": {}}); + if let Some(allowed) = allowed { + value["rate_limit"]["allowed"] = json!(allowed); + } + for ((role, seconds), reset) in [("primary_window", 18_000), ("secondary_window", 604_800)] + .into_iter() + .zip(resets) + { + let mut window = json!({"used_percent": 100, "limit_window_seconds": seconds}); + if let Some(offset) = reset { + window["reset_at"] = json!(now + offset); + } + value["rate_limit"][role] = window; + } + value +} + +#[tokio::test] +async fn allowed_refresh_recovers_full_windows_only_after_the_confirmed_resets() { + for (name, resets, failure_offset, allowed, recovered) in [ + ("before_reset", vec![Some(3600)], -120, Some(true), false), + ("after_reset", vec![Some(-60)], -120, Some(true), true), + ("still_denied", vec![Some(-60)], -120, Some(false), false), + ("unknown_access", vec![Some(-60)], -120, None, false), + ("unknown_reset", vec![None], -120, Some(true), false), + ("new_failure", vec![Some(-60)], -30, Some(true), false), + ( + "weekly_not_due", + vec![Some(-60), Some(3600)], + -120, + Some(true), + false, + ), + ( + "both_due", + vec![Some(-90), Some(-60)], + -120, + Some(true), + true, + ), + ( + "weekly_unknown", + vec![Some(-60), None], + -120, + Some(true), + false, + ), + ] { + for worker in [false, true] { + let store = Arc::new(MemoryAccountStore::default()); + let account_id = format!("acct_reset_{name}_{worker}"); + create_account(&store, &account_id).await; + let account = store.account(&account_id).expect("account"); + let now = Utc::now().timestamp(); + let observed_at = + SystemTime::UNIX_EPOCH + Duration::from_secs((now + failure_offset) as u64); + let reset_at = resets + .iter() + .filter_map(|reset| *reset) + .min() + .map(|offset| SystemTime::UNIX_EPOCH + Duration::from_secs((now + offset) as u64)); + let old_usage = usage(now, &resets, Some(false)); + store + .compare_and_swap_quota(QuotaObservation { + plan_type: None, + account_id: account.id().clone(), + expected_revision: account.revision(), + quota: OpaqueProviderData::new(old_usage.as_object().expect("usage").clone()), + observed_at, + state: QuotaState::exhausted( + QuotaEvidence::UsageLimitReached, + observed_at, + reset_at, + ), + }) + .await + .expect("persist confirmed exhaustion before refresh"); + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/api/codex/usage")) + .respond_with( + ResponseTemplate::new(200).set_body_json(usage(now, &resets, allowed)), + ) + .mount(&server) + .await; + let service = quota_service_with_base_url(&store, reqwest::Client::new(), server.uri()); + if worker { + let summary = service.synchronize().await.expect("worker refresh"); + assert_eq!(summary.updated, u64::from(recovered), "{name}"); + } else { + service + .refresh_account(account.id()) + .await + .expect("manual refresh"); + } + let refreshed = service + .read_account(account.id()) + .await + .expect("read") + .expect("snapshot"); + assert_eq!( + refreshed.quota().is_exhausted(), + !recovered, + "{name}, worker={worker}" + ); + assert!( + refreshed + .windows() + .iter() + .all(|window| window.used_percent() == Some(100.0)) + ); + assert_eq!( + store + .account(&account_id) + .expect("account") + .quota() + .is_exhausted(), + !recovered + ); + assert_eq!( + store + .quota_json(&account_id) + .expect("quota") + .get("_quota_recovery") + .is_none(), + recovered + ); + + if recovered { + service + .refresh_account(account.id()) + .await + .expect("allowed remains allowed"); + assert!( + !store + .account(&account_id) + .expect("account") + .quota() + .is_exhausted() + ); + // 新拒绝不能沿用刚刚解除时的旧 reset，即使查询仍然 allowed=true + persist_quota_state(&store, &account, exhausted_quota(reset_at)).await; + for _ in 0..2 { + let snapshot = service + .refresh_account(account.id()) + .await + .expect("refresh after new refusal"); + assert!(snapshot.quota().is_exhausted(), "{name}"); + } + } + } + } +} diff --git a/backend/crates/providers/openai/tests/provider/contract/mod.rs b/backend/crates/providers/openai/tests/provider/contract/mod.rs index da2b42ab7..83ad59dbf 100644 --- a/backend/crates/providers/openai/tests/provider/contract/mod.rs +++ b/backend/crates/providers/openai/tests/provider/contract/mod.rs @@ -8421,6 +8421,82 @@ async fn official_usage_limit_failure_persists_fact_without_fabricating_usage() ); } +#[tokio::test] +async fn usage_limit_failure_recovers_with_credits_without_replaying_the_failed_request() { + let store = Arc::new(MemoryAccountStore::default()); + let account_id = "acct_provider_contract"; + create_account(&store, account_id).await; + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/api/codex/usage")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "rate_limit": { + "allowed": false, + "limit_reached": true, + "primary_window": {"used_percent": 100, "reset_at": 1_900_000_000} + }, + "credits": {"has_credits": true, "unlimited": false, "balance": "20"} + }))) + .expect(1) + .mount(&server) + .await; + Mock::given(method("POST")) + .and(path("/codex/responses")) + .respond_with(ResponseTemplate::new(429).set_body_json(json!({ + "error": {"type": "usage_limit_reached", "message": "usage limit reached", "resets_at": 1_900_000_000} + }))) + .expect(1) + .mount(&server) + .await; + let mut stream = provider_with_base_url(&store, server.uri()) + .execute( + planned_request("openai", http_generate_operation()), + context("req_usage_limit_credits", CancellationToken::new()), + ) + .await + .expect("provider stream"); + let failure = loop { + match stream.next().await { + Some(Ok(_)) => {} + Some(Err(error)) => break error, + None => panic!("quota failure must reach the caller"), + } + }; + assert_eq!(failure.kind(), ProviderErrorKind::QuotaExhausted); + assert!( + store + .account(account_id) + .expect("account") + .quota() + .is_exhausted() + ); + + timeout(Duration::from_secs(5), async { + while store + .account(account_id) + .expect("account") + .quota() + .is_exhausted() + { + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("credits restore access after background refresh"); + assert!(stream.next().await.is_none()); + let account = store.account(account_id).expect("recovered account"); + assert_eq!(account.quota().access(), QuotaAccessState::Allowed); + assert_eq!(account.credential_state(), CredentialState::Ready); + let quota = store.quota_json(account_id).expect("persisted credits"); + assert_eq!(quota.pointer("/credits/has_credits"), Some(&json!(true))); + assert_eq!( + quota.pointer("/rate_limits_by_limit_id/codex/primary_window/used_percent"), + Some(&json!(100)) + ); + assert!(quota.get("_quota_recovery").is_none()); + server.verify().await; +} + #[tokio::test] async fn capacity_http_and_websocket_opening_rejections_use_bounded_business_retry() { for use_websocket in [false, true] { diff --git a/backend/crates/providers/openai/tests/support.rs b/backend/crates/providers/openai/tests/support.rs index fdc89fc26..050f8cff9 100644 --- a/backend/crates/providers/openai/tests/support.rs +++ b/backend/crates/providers/openai/tests/support.rs @@ -430,7 +430,7 @@ impl ProviderAccountStore for MemoryAccountStore { async fn compare_and_swap_quota( &self, - observation: QuotaObservation, + mut observation: QuotaObservation, ) -> Result { let mut accounts = self.accounts.lock().expect("account store lock"); let stored = accounts @@ -446,7 +446,18 @@ impl ProviderAccountStore for MemoryAccountStore { { return Ok(QuotaWriteOutcome::Conflict); } - let quota = observation.state; + // 与 PostgreSQL 一致，展示快照与访问事实分别按各自观察时间隔离 + let current = stored.account.quota(); + let quota = if observation + .state + .observed_at() + .is_some_and(|observed| current.observed_at().is_none_or(|time| time <= observed)) + { + observation.state + } else { + current + }; + observation.state = quota; let mut replacement = AccountRebuild::preserving(&stored.account).with_quota(quota); if let Some(plan_type) = &observation.plan_type { replacement.profile = Some(( diff --git a/docs/api.md b/docs/api.md index dd7804a28..8576f95a2 100644 --- a/docs/api.md +++ b/docs/api.md @@ -754,7 +754,10 @@ Codex 的 `model_catalog_json` 配置。账号设置保存不等待上游模型 `quota.credits` 返回上游点数余额，未提供点数信息时为 `null`。对象包含 `hasCredits`、`unlimited` 和 `balance`，余额为保留上游精度的十进制字符串，未提供余额时为 `null`，明确的零余额为 `"0"`。 OpenAI 点数随现有额度刷新和正常请求的额度信息同步，不与主动重置卡库存或百分比限额合并， -不参与账号可用性判断 +不根据 `balance` 单独推导可用性。主动查询明确 `hasCredits=true` 或 `unlimited=true`，且无阻止 +点数支出的上限或工作区限制时，即使套餐窗口耗尽也可使用点数，并解除已有额度耗尽状态。 +真实额度拒绝后的后台回读同样支持此恢复路径，但不会自动重发本次失败请求；点数不能清除 +凭据错误、停用或冷却状态。上游点数信息也可能延迟，后续真实拒绝仍会记录额度耗尽 账号列表和详情返回 `notes`（无备注时为 `null`）。编辑时省略或 `null` 保留原备注；字符串最多 500 个 Unicode 字符，允许换行和制表符，保存时去除首尾空白，空字符串清空备注。备注独立于上游身份，导入时未显式提供备注、 @@ -1118,11 +1121,14 @@ OAuth start 使用： 但上游 authorization code 本身通常只能交换一次；已完成过 token exchange 时应重新创建 OAuth flow - `GET /accounts/quota` 只读取最后一次落库快照；`POST /accounts/quota/refresh` 才访问上游。access token 已过期时，额度刷新要求先走 credential 刷新或重新授权，不会拿过期 token 探测额度 -- OpenAI 已耗尽账号每 30 分钟主动复核一次，也会在最早未恢复窗口的 `resetAt + 2 分钟` 到期后 - 提前复核；周期复核不等待旧重置时间，因此也能发现官方提前重置。正常账号有非零用量或触顶窗口时， - 在该窗口的 `resetAt + 2 分钟` 后主动复核。后台每 30 秒检查触发条件；同一重置边界复核后仍未更新时 - 回到 30 分钟重试，避免旧 reset 持续触发请求。各窗口独立确认恢复，时间到期本身不会直接解除账号 - 耗尽或将展示用量归零 +- OpenAI 已耗尽账号首次可复核，在 `resetAt` 尚未到达或未知时按 30 分钟间隔兜底，也可发现 + 官方提前重置。到达最早未恢复窗口的 `resetAt` 后，先等待 10 秒传播余量，再在下一轮后台检查 + 复核；reset 后前 5 分钟最小查询间隔为 30 秒，之后仍耗尽则改为 5 分钟。正常账号有非零用量或触顶窗口时，仍在 + `resetAt + 2 分钟` 后复核，未更新则按 30 分钟间隔重试。后台每 30 秒检查条件，实际恢复时间还受 + 排队、查询耗时和上游传播影响；查询失败也遵守间隔。无可用点数时，到期前仍按原窗口证据恢复； + 本次耗尽之后的已记录窗口 reset 全部到期后，新查询明确 `allowed=true` 可解除限制，即使用量仍为 + 100%。未知 reset 或再次拒绝后残留的旧 reset 仍走原规则，不复用旧边界解封；时间到期本身不会 + 解除限制，也不会将展示用量归零 - 账号 `enabled` 控制是否参与请求调度，不控制 OAuth 凭据续期。OpenAI 与 xAI 的停用账号仍按各自刷新 策略维护 Token；刷新结果更新凭据状态，但不会启用调度。无 refresh token 或凭据已进入失效、无效、 封禁状态的账号不参加后台自动续期 diff --git a/docs/architecture.md b/docs/architecture.md index bd4fe8ba7..5fc8f326f 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -890,6 +890,12 @@ credential 与 quota 是两组独立事实：credential refresh 不等于 quota OpenAI Provider 将其中明确的套餐变更与额度原子提交，共用凭据版本和观察时间保护； 空值及 `unknown` 不覆盖套餐，同族泛化值保留具体子类型。Token 刷新保留提交时的账号资料 +OpenAI Provider 的额度快照统一解释点数可用性，主动查询、耗尽恢复与失败后回读复用同一规则。 +可用点数独立于套餐窗口；通过点数恢复时原子保存新快照、访问事实并清理旧窗口恢复进度。 +无可用点数的失败回读保留已确认的拒绝；常规刷新在到期前仍按窗口证据恢复，本次耗尽之后的 +已记录窗口 reset 全部到期后可依据新的明确放行观察恢复。查询沿用 credential revision 与 +观察时间隔离，不允许旧查询撤销较新的失败；恢复条件和后台复核节奏见 [账号 API](api.md#账号能力导入与-oauth) + OpenAI 订阅周期属于按需个人信息，不是额度事实。Admin 账号用例通过现有 Provider 管理端口并发读取 个人资料统计与订阅，汇聚为一次只读响应；Provider 继续拥有各自的认证、出站代理与上游协议处理。 订阅不由 quota、导入或后台任务触发，不持久化。汇聚结果返回前核对账号身份与 credential revision，