diff --git a/backend/crates/providers/openai/src/config.rs b/backend/crates/providers/openai/src/config.rs index a4f7a4eeb..957ffb33e 100644 --- a/backend/crates/providers/openai/src/config.rs +++ b/backend/crates/providers/openai/src/config.rs @@ -213,6 +213,7 @@ impl CodexWebSocketPoolSettings { #[derive(Debug, Clone, Deserialize, PartialEq, Eq)] #[serde(deny_unknown_fields)] pub struct CodexQuotaSettings { + /// 保留模型目录的刷新周期;额度独立每 30 秒检查周期复核和 reset 到期条件。 pub refresh_interval_minutes: u64, } diff --git a/backend/crates/providers/openai/src/credential/quota/mod.rs b/backend/crates/providers/openai/src/credential/quota/mod.rs index 512e45d2a..a31c19485 100644 --- a/backend/crates/providers/openai/src/credential/quota/mod.rs +++ b/backend/crates/providers/openai/src/credential/quota/mod.rs @@ -57,8 +57,8 @@ use snapshot::{ 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 EXHAUSTED_QUOTA_FALLBACK_RECHECK_INTERVAL: Duration = Duration::from_secs(30 * 60); const EXHAUSTED_QUOTA_REFRESH_RETRY_INTERVAL: Duration = Duration::from_secs(30 * 60); +const EXHAUSTED_QUOTA_RESET_GRACE: Duration = Duration::from_secs(2 * 60); /// 首次 OAuth 异步观察失败时,由既有 quota worker 兜底重试的单轮上限。 const INITIAL_QUOTA_SYNC_BATCH: usize = 100; // 5xx 上游拒绝的短退避重试预算;指数退避 1s/2s,吞掉瞬时抖动。 @@ -68,8 +68,8 @@ const QUOTA_FETCH_5XX_BASE_DELAY: Duration = Duration::from_secs(1); /// OpenAI Provider 主动额度刷新的调度策略。 /// /// 正常账号依赖请求响应的被动额度同步;周期 worker 仅复核已耗尽账号。 -/// 该策略只控制 Provider 自己的 quota worker 频率,避免把 OpenAI -/// 额度语义泄漏到公共调度层。 +/// 该策略保留模型目录的周期刷新频率;额度到期检查使用独立的短周期, +/// 避免到达 reset 后还要等待完整的目录刷新周期。 #[derive(Debug, Clone, Copy)] pub struct CodexQuotaRefreshPolicy { interval: Duration, @@ -204,7 +204,12 @@ struct CodexQuotaSchedulingProjection { struct CodexQuotaProjectionState { next_version: u64, entries: BTreeMap, - last_periodic_refresh_at: BTreeMap, + last_periodic_refresh_at: BTreeMap, +} + +struct CodexQuotaRefreshAttempt { + monotonic_at: Instant, + wall_at: SystemTime, } #[derive(Debug, Clone, Copy)] @@ -398,16 +403,27 @@ impl CodexQuotaSchedulingProjection { .last_periodic_refresh_at .retain(|account_id, _| candidate_ids.contains(account_id)); - // 正常账号只由真实请求的响应头和 `codex.rate_limits` 被动同步。定时器只复核 - // 已耗尽且已到 reset/fallback 时刻的账号;重复复核至少间隔 30 分钟。 + // 正常账号只由真实请求的响应头和 `codex.rate_limits` 被动同步。 + // 已耗尽账号每 30 分钟复核,不能等待旧 reset:官方活动可能提前重置额度。 + // reset + 2 分钟可提前触发一次复核,给上游重置留出传播时间。 let mut reserved = Vec::new(); for account in candidates { - if !periodic_quota_refresh_due(&state, account.id(), refreshed_at) { + if !periodic_quota_refresh_due( + &state, + account.id(), + account.quota().reset_at(), + now, + refreshed_at, + ) { continue; } - state - .last_periodic_refresh_at - .insert(account.id().clone(), refreshed_at); + state.last_periodic_refresh_at.insert( + account.id().clone(), + CodexQuotaRefreshAttempt { + monotonic_at: refreshed_at, + wall_at: now, + }, + ); reserved.push(account); } reserved @@ -415,24 +431,27 @@ impl CodexQuotaSchedulingProjection { } fn quota_refresh_candidate(account: ProviderAccount, now: SystemTime) -> Option { - // reset 到期是“应刷新”的信号,不是“已恢复”的证据。 - (eligible_periodic_quota_refresh(&account, now) - && account - .quota() - .exhaustion_refresh_due(now, EXHAUSTED_QUOTA_FALLBACK_RECHECK_INTERVAL)) - .then_some(account) + (eligible_periodic_quota_refresh(&account, now) && account.quota().is_exhausted()) + .then_some(account) } fn periodic_quota_refresh_due( state: &CodexQuotaProjectionState, account_id: &ProviderAccountId, - now: Instant, + reset_at: Option, + now: SystemTime, + monotonic_now: Instant, ) -> bool { state .last_periodic_refresh_at .get(account_id) .is_none_or(|last| { - now.saturating_duration_since(*last) >= EXHAUSTED_QUOTA_REFRESH_RETRY_INTERVAL + monotonic_now.saturating_duration_since(last.monotonic_at) + >= EXHAUSTED_QUOTA_REFRESH_RETRY_INTERVAL + || reset_at + .and_then(|reset| reset.checked_add(EXHAUSTED_QUOTA_RESET_GRACE)) + // 已在该边界之后复核过时回到周期重试,避免过期 reset 每轮触发。 + .is_some_and(|due_at| last.wall_at < due_at && due_at <= now) }) } diff --git a/backend/crates/providers/openai/src/provider/workers.rs b/backend/crates/providers/openai/src/provider/workers.rs index 84a6a03fd..6bab4e449 100644 --- a/backend/crates/providers/openai/src/provider/workers.rs +++ b/backend/crates/providers/openai/src/provider/workers.rs @@ -7,8 +7,10 @@ pub(super) const WORKER_MAXIMUM_BACKOFF: Duration = Duration::from_secs(60); pub(super) const WORKER_LEASE_TTL: Duration = Duration::from_secs(15 * 60); pub(super) const WORKER_LEASE_RENEWAL: Duration = Duration::from_secs(5 * 60); pub(super) const OAUTH_REFRESH_INTERVAL: Duration = Duration::from_secs(30); +pub(super) const QUOTA_CHECK_INTERVAL: Duration = Duration::from_secs(30); pub(super) const DESKTOP_RELEASE_WORKER_OWNER: &str = "openai-desktop-release"; pub(super) const MODEL_ETAG_WORKER_OWNER: &str = "openai-model-etag"; +pub(super) const MODEL_CATALOG_WORKER_OWNER: &str = "openai-model-catalog"; pub(crate) fn worker_contributions( refresh: Arc, @@ -20,6 +22,7 @@ pub(crate) fn worker_contributions( ) -> Result, WorkerDefinitionError> { let refresh_id = WorkerId::try_new(WorkerKind::OAuthRefresh, PROVIDER_NAME)?; let quota_id = WorkerId::try_new(WorkerKind::QuotaCatalogHealth, PROVIDER_NAME)?; + let catalog_id = WorkerId::try_new(WorkerKind::QuotaCatalogHealth, MODEL_CATALOG_WORKER_OWNER)?; let etag_id = WorkerId::try_new(WorkerKind::QuotaCatalogHealth, MODEL_ETAG_WORKER_OWNER)?; let desktop_release_id = WorkerId::try_new(WorkerKind::QuotaCatalogHealth, DESKTOP_RELEASE_WORKER_OWNER)?; @@ -34,9 +37,13 @@ pub(crate) fn worker_contributions( contributions.extend([ WorkerContribution::Registration(scheduled_registration( quota_id, + QUOTA_CHECK_INTERVAL, + Box::new(OpenAiQuotaTask { quota }), + )?), + WorkerContribution::Registration(scheduled_registration( + catalog_id, quota_refresh_policy.interval(), - Box::new(OpenAiQuotaTask { - quota, + Box::new(OpenAiCatalogTask { catalog: Arc::clone(&catalog), }), )?), @@ -159,6 +166,9 @@ impl ScheduledTask for OpenAiOAuthRefreshTask { pub(super) struct OpenAiQuotaTask { quota: Arc, +} + +pub(super) struct OpenAiCatalogTask { catalog: Arc, } @@ -195,7 +205,6 @@ impl ScheduledTask for OpenAiQuotaTask { if context.cancellation().is_cancelled() { return Ok(()); } - let mut failures = false; match self.quota.synchronize().await { Ok(summary) if summary.has_operational_failures() => { tracing::warn!( @@ -209,30 +218,33 @@ impl ScheduledTask for OpenAiQuotaTask { } Ok(_) => {} Err(error) => { - failures = true; tracing::warn!( error = %error, "OpenAI quota synchronization failed" ); + return Err(WorkerTaskError::safe("OpenAI quota synchronization failed")); } } + Ok(()) + }) + } +} + +impl ScheduledTask for OpenAiCatalogTask { + fn run_cycle(&self, context: WorkerCycleContext) -> BoxFuture<'_, Result<(), WorkerTaskError>> { + Box::pin(async move { if context.cancellation().is_cancelled() { return Ok(()); } match self.catalog.refresh_catalogs().await { - Ok(_) | Err(CodexCredentialCatalogError::NoEligibleCredential) => {} + Ok(_) | Err(CodexCredentialCatalogError::NoEligibleCredential) => Ok(()), Err(error) => { - failures = true; tracing::warn!(error = %error, "OpenAI model catalog refresh failed"); + Err(WorkerTaskError::safe( + "OpenAI model catalog synchronization failed", + )) } } - if failures { - Err(WorkerTaskError::safe( - "OpenAI quota or catalog synchronization failed", - )) - } else { - Ok(()) - } }) } } diff --git a/backend/crates/providers/openai/tests/admin.rs b/backend/crates/providers/openai/tests/admin.rs index 2af4e64eb..f55e08ed8 100644 --- a/backend/crates/providers/openai/tests/admin.rs +++ b/backend/crates/providers/openai/tests/admin.rs @@ -74,7 +74,7 @@ async fn openai_bundle_exposes_one_core_provider_and_drains_worker_contributions assert_eq!(bundle.core_provider().name(), "openai"); assert_eq!(bundle.admin_provider().provider_kind().as_str(), "openai"); let contributions = bundle.take_worker_contributions(); - assert_eq!(contributions.len(), 4); + assert_eq!(contributions.len(), 5); assert!( contributions .iter() @@ -101,6 +101,30 @@ async fn openai_bundle_exposes_one_core_provider_and_drains_worker_contributions panic!("Desktop release worker must be scheduled"); }; assert_eq!(schedule.interval(), APPCAST_POLL_INTERVAL); + for (owner, interval) in [ + ("openai", Duration::from_secs(30)), + ( + "openai-model-catalog", + config.config.quota_refresh_policy().interval(), + ), + ] { + let schedule = contributions + .iter() + .find_map(|contribution| match contribution { + WorkerContribution::Registration(registration) + if registration.id.kind() == WorkerKind::QuotaCatalogHealth + && registration.id.owner() == owner => + { + match ®istration.runnable { + WorkerRunnable::Scheduled { schedule, .. } => Some(schedule), + WorkerRunnable::Daemon { .. } => None, + } + } + _ => None, + }) + .expect("quota/catalog scheduled worker"); + assert_eq!(schedule.interval(), interval, "{owner}"); + } assert!(contributions.iter().any(|contribution| { matches!( contribution, diff --git a/backend/crates/providers/openai/tests/credential/quota/mod.rs b/backend/crates/providers/openai/tests/credential/quota/mod.rs index f05f4070d..7474c9367 100644 --- a/backend/crates/providers/openai/tests/credential/quota/mod.rs +++ b/backend/crates/providers/openai/tests/credential/quota/mod.rs @@ -1,6 +1,7 @@ //! OpenAI 额度事实边界与展示快照回归。 mod recovery; +mod refresh_timing; 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 new file mode 100644 index 000000000..2a2129972 --- /dev/null +++ b/backend/crates/providers/openai/tests/credential/quota/refresh_timing.rs @@ -0,0 +1,130 @@ +//! 通过 quota 服务验证周期复核与 reset 宽限期,不依赖内部调度状态。 + +use std::time::Duration; + +use super::*; + +async fn mount_usage(server: &MockServer, value: serde_json::Value) { + server.reset().await; + Mock::given(method("GET")) + .and(path("/api/codex/usage")) + .respond_with(ResponseTemplate::new(200).set_body_json(value)) + .mount(server) + .await; +} + +async fn wait_for_reset_grace(reset: i64) { + let due = SystemTime::UNIX_EPOCH + Duration::from_secs((reset + 120) as u64); + if let Ok(remaining) = due.duration_since(SystemTime::now()) { + tokio::time::sleep(remaining + Duration::from_millis(50)).await; + } +} + +#[tokio::test] +async fn reset_grace_bypasses_periodic_throttle_once_then_allows_the_next_window() { + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, "acct_quota_timing").await; + let account = store.account("acct_quota_timing").expect("account"); + let server = MockServer::start().await; + let service = quota_service_with_base_url(&store, reqwest::Client::new(), server.uri()); + // 使用接近宽限期终点的真实窗口,在数秒内走完两个窗口的调度与落库路径。 + let short_reset = Utc::now().timestamp() - 115; + let week_reset = short_reset + 3; + let usage = |short_used, short_reset, week_used, week_reset| { + json!({"rate_limit": { + "allowed": false, + "primary_window": { + "used_percent": short_used, "reset_at": short_reset, "limit_window_seconds": 18_000, + }, + "secondary_window": { + "used_percent": week_used, "reset_at": week_reset, "limit_window_seconds": 604_800, + }, + }}) + }; + mount_usage(&server, usage(100, short_reset, 100, week_reset)).await; + service + .refresh_account(account.id()) + .await + .expect("seed exhaustion"); + service.synchronize().await.expect("initial periodic check"); + let requests = server.received_requests().await.expect("requests").len(); + + assert!( + Utc::now().timestamp() < short_reset + 120, + "fixture must precede grace deadline" + ); + service.synchronize().await.expect("before grace deadline"); + assert_eq!( + server.received_requests().await.expect("requests").len(), + requests + ); + + mount_usage(&server, usage(0, short_reset + 18_000, 100, week_reset)).await; + wait_for_reset_grace(short_reset).await; + let partial = service.synchronize().await.expect("short reset check"); + assert_eq!(partial.exhausted, 1); + assert_eq!( + store + .account("acct_quota_timing") + .expect("account") + .quota() + .reset_at(), + Some(SystemTime::UNIX_EPOCH + Duration::from_secs(week_reset as u64)) + ); + service + .synchronize() + .await + .expect("next scan before weekly grace deadline"); + assert_eq!(server.received_requests().await.expect("requests").len(), 1); + + // 周窗口的到期复核仍未恢复时,也不能每轮扫描重复请求。 + wait_for_reset_grace(week_reset).await; + assert_eq!( + service + .synchronize() + .await + .expect("weekly reset check") + .exhausted, + 1 + ); + service + .synchronize() + .await + .expect("repeat scan after weekly check"); + assert_eq!(server.received_requests().await.expect("requests").len(), 2); +} + +#[tokio::test] +async fn periodic_checks_continue_when_reset_is_unknown_or_far_in_the_future() { + for reset in [None, Some(Utc::now().timestamp() + 604_800)] { + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, "acct_quota_periodic").await; + let account = store.account("acct_quota_periodic").expect("account"); + let server = MockServer::start().await; + let service = quota_service_with_base_url(&store, reqwest::Client::new(), server.uri()); + let mut usage = json!({"rate_limit": { + "allowed": false, + "primary_window": {"used_percent": 100, "limit_window_seconds": 604_800}, + }}); + if let Some(reset) = reset { + usage["rate_limit"]["primary_window"]["reset_at"] = json!(reset); + } + mount_usage(&server, usage.clone()).await; + service + .refresh_account(account.id()) + .await + .expect("seed exhaustion"); + mount_usage(&server, usage).await; + + assert_eq!( + service + .synchronize() + .await + .expect("periodic check") + .exhausted, + 1 + ); + service.synchronize().await.expect("throttled repeat check"); + assert_eq!(server.received_requests().await.expect("requests").len(), 1); + } +} diff --git a/backend/crates/providers/openai/tests/credential/quota/slots.rs b/backend/crates/providers/openai/tests/credential/quota/slots.rs index d9d46e594..e38a9eb07 100644 --- a/backend/crates/providers/openai/tests/credential/quota/slots.rs +++ b/backend/crates/providers/openai/tests/credential/quota/slots.rs @@ -172,6 +172,99 @@ async fn recovery_checks_each_exhausted_slot_in_worker_and_manual_refresh() { } } +#[tokio::test] +async fn worker_detects_early_resets_without_unlocking_other_exhausted_windows() { + let short_reset = Utc::now().timestamp() + 18_000; + let week_reset = Utc::now().timestamp() + 604_800; + for (name, old_short, old_week, new_short, new_week, recovered) in [ + ("short_only", Some(100), None, Some(0), None, true), + ("weekly_only", None, Some(100), None, Some(0), true), + ( + "short_with_weekly_available", + Some(100), + Some(45), + Some(0), + Some(45), + true, + ), + ( + "weekly_still_exhausted", + Some(100), + Some(100), + Some(0), + Some(100), + false, + ), + ] { + let store = Arc::new(MemoryAccountStore::default()); + let account_id = format!("acct_early_{name}"); + create_account(&store, &account_id).await; + let account = store.account(&account_id).expect("account"); + let server = MockServer::start().await; + let service = quota_service_with_base_url(&store, reqwest::Client::new(), server.uri()); + let window = |used: Option, reset, seconds| { + used.map(|used| { + json!({ + "used_percent": used, + "reset_at": reset, + "limit_window_seconds": seconds, + }) + }) + }; + mount_usage( + &server, + json!({"rate_limit": { + "allowed": false, + "primary_window": window(old_short, short_reset, 18_000), + "secondary_window": window(old_week, week_reset, 604_800), + }}), + ) + .await; + let baseline = service + .refresh_account(account.id()) + .await + .expect("seed exhaustion"); + assert!(baseline.quota().reset_at().expect("future reset") > SystemTime::now()); + + mount_usage( + &server, + json!({"rate_limit": { + "allowed": recovered, + "primary_window": window(new_short, short_reset + 18_000, 18_000), + "secondary_window": window(new_week, if new_week == Some(0) { + week_reset + 604_800 + } else { + week_reset + }, 604_800), + }}), + ) + .await; + let summary = service + .synchronize() + .await + .expect("periodic early reset check"); + assert_eq!(summary.updated, u64::from(recovered), "{name}"); + assert_eq!(summary.exhausted, u64::from(!recovered), "{name}"); + assert_eq!( + store + .account(&account_id) + .expect("updated account") + .quota() + .is_exhausted(), + !recovered, + "{name}" + ); + + // 未恢复的周窗口仍在未来,也必须保留复核节流;恢复后则退出定时复核。 + service.synchronize().await.expect("next worker cycle"); + assert_eq!( + server.received_requests().await.expect("requests").len(), + 1, + "{name}" + ); + } +} + #[tokio::test] async fn partial_recovery_survives_restart_missing_windows_and_changed_roles() { let store = Arc::new(MemoryAccountStore::default()); diff --git a/docs/api.md b/docs/api.md index 2c06d7934..4b29c3050 100644 --- a/docs/api.md +++ b/docs/api.md @@ -306,6 +306,9 @@ OAuth start 使用: 但上游 authorization code 本身通常只能交换一次;已完成过 token exchange 时应重新创建 OAuth flow。 - `GET /accounts/quota` 只读取最后一次落库快照;`POST /accounts/quota/refresh` 才访问上游。access token 已过期时,额度刷新要求先走 credential 刷新或重新授权,不会拿过期 token 探测额度。 +- OpenAI 已耗尽账号每 30 分钟主动复核一次,也会在最早未恢复窗口的 `resetAt + 2 分钟` 到期后 + 提前复核。后台每 30 秒检查触发条件;同一重置边界复核后仍未恢复时回到 30 分钟重试, + 避免旧 reset 持续触发请求。各窗口独立确认恢复,时间到期本身不会直接解除账号耗尽。 - `POST /accounts/recover` 是管理员对本地事实的强制恢复:它清除 Redis cooldown 和已保存的额度/错误, 把账号重新启用并恢复为可调度 credential;它不验证上游账号是否已经恢复,下一次真实请求仍可重新写入 失败事实。