From dfabf0f9ddaa0dbf1f5789c669c03e776557ff3c Mon Sep 17 00:00:00 2001 From: 2han9wen71an <2han9wen71an@gmail.com> Date: Tue, 15 Sep 2026 15:38:46 +0800 Subject: [PATCH] fix(openai): recover quota after same-window usage drops (#95) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit OpenAI 周窗口的 used_percent 可以在同一窗口内回落（滚动计量），而 reset_at 固定到窗口边界才变化。恢复规则要求「reset 前进 + 用量低于 10%」，两个条件在滑动周窗口上都无法成立：耗尽账号被钉在旧 reset 日期（最长一周）无法参与调度，尽管 /usage 已显式报告 allowed=true / limit_reached=false 且 5h 窗口全新。 新增第二条解除路径：同一窗口（reset 未滚动）内连续两次观测都未触顶 即解除。候选记录持久化在 _quota_recovery，跨 worker 周期与重启累积； 连续两次的要求保持对「耗尽后立刻拉到的旧快照」竞态的防护（单次旧 快照撑不过两次观测）。原有「reset 前进 + 用量<10%」快路径不变。 --- .../openai/src/credential/quota/recovery.rs | 58 ++++++++--- .../openai/tests/credential/quota/slots.rs | 96 +++++++++++++++++++ 2 files changed, 142 insertions(+), 12 deletions(-) diff --git a/backend/crates/providers/openai/src/credential/quota/recovery.rs b/backend/crates/providers/openai/src/credential/quota/recovery.rs index 9b98a17f4..2ec068bd1 100644 --- a/backend/crates/providers/openai/src/credential/quota/recovery.rs +++ b/backend/crates/providers/openai/src/credential/quota/recovery.rs @@ -18,6 +18,10 @@ pub(super) struct QuotaRecovery { // 绑定本次耗尽事实，避免真实推理成功后再次耗尽时复用上次的恢复进度。 exhausted_at_micros: i64, pending: BTreeMap>>, + // 已连续观测到"未触顶"的待解除窗口；第二次确认后解除。旧文档没有该 + // 字段时反序列化为空集，从当前耗尽事实重新积累证据。 + #[serde(default, skip_serializing_if = "BTreeMap::is_empty")] + candidates: BTreeMap, } impl QuotaRecovery { @@ -59,23 +63,44 @@ impl QuotaRecovery { Self { exhausted_at_micros: exhaustion_time(state), pending, + candidates: BTreeMap::new(), } } fn observe(&mut self, refreshed: &CodexAccountQuotaSnapshot) -> bool { let had_baseline = !self.pending.is_empty(); + let mut candidates = BTreeMap::new(); for window in refreshed.windows().iter().filter(|w| w.is_account_wide()) { - if let Some(previous_reset) = self.pending.get_mut(window.key()) { - if window_reset_recovered(window, *previous_reset) { - self.pending.remove(window.key()); - } else if previous_reset.is_none() { - *previous_reset = window.reset_at(); + let key = window.key(); + if let Some(previous_reset) = self.pending.get(key).copied() { + if window_reset_recovered(window, previous_reset) { + self.pending.remove(key); + continue; + } + if previous_reset.is_none() { + // 首次观测到该窗口只锚定 reset 基线，不构成解除证据。 + self.pending.insert(key.to_owned(), window.reset_at()); + continue; + } + // 同一窗口内用量回落（reset 不滚动）时"reset 前进"永远不会 + // 成立，滑动窗口的正常用量也长期高于低用量阈值；以上游连续 + // 两次观测都报告未触顶为解除证据。单次观测无法排除耗尽后 + // 立刻拉到的旧快照（reset 未滚动、用量尚未触顶）。 + let rolled_forward = window + .reset_at() + .is_some_and(|current| Some(current) > previous_reset); + if !rolled_forward && !window.limit_reached() { + if self.candidates.contains_key(key) { + self.pending.remove(key); + } else { + candidates.insert(key.to_owned(), ()); + } } } else if window.limit_reached() { - self.pending - .insert(window.key().to_owned(), window.reset_at()); + self.pending.insert(key.to_owned(), window.reset_at()); } } + self.candidates = candidates; had_baseline && self.pending.is_empty() } } @@ -85,15 +110,24 @@ fn exhaustion_time(state: QuotaState) -> i64 { } fn window_reset_recovered(window: &CodexQuotaWindow, previous: Option>) -> bool { - previous + // reset 必须前进：防止耗尽后立刻拉到的旧快照（reset 未滚动）误解锁。 + if !previous .zip(window.reset_at()) .is_some_and(|(previous, current)| current > previous) - && window - .used_percent() - .is_some_and(|used| used < RESET_RECOVERY_MAX_USED_PERCENT) + { + return false; + } + // reset 前进说明已进入新窗口，新窗口用量应接近零；高用量说明 reset 基线 + // 不可信（例如滑动窗口），保持锁定，等待"同窗口用量回落"的连续观测解除。 + window + .used_percent() + .is_some_and(|used| used < RESET_RECOVERY_MAX_USED_PERCENT) } -/// 统一手动刷新和 worker 的恢复规则；上游 allowed 不参与已耗尽窗口的解除判断。 +/// 统一手动刷新和 worker 的恢复规则。已耗尽窗口的解除有两条证据路径： +/// reset 前进且用量低于低用量阈值（窗口已滚动），或同一窗口内连续两次观测 +/// 都未触顶（滑动窗口的用量回落，reset 不滚动）。后者防止账号被钉在旧 +/// reset 日期上；连续两次的要求排除耗尽后立刻拉到的旧快照。 pub(super) fn reconcile_refresh( current: QuotaState, refreshed: &mut CodexAccountQuotaSnapshot, diff --git a/backend/crates/providers/openai/tests/credential/quota/slots.rs b/backend/crates/providers/openai/tests/credential/quota/slots.rs index e38a9eb07..b12b65c32 100644 --- a/backend/crates/providers/openai/tests/credential/quota/slots.rs +++ b/backend/crates/providers/openai/tests/credential/quota/slots.rs @@ -561,6 +561,102 @@ async fn an_unknown_slot_establishes_a_baseline_before_it_can_recover() { } } +#[tokio::test] +async fn weekly_usage_falling_back_within_the_same_window_recovers_after_two_observations() { + // 滑动周窗口的用量可以在同一窗口内回落：reset 不滚动、用量从 100% 降到 + // 正常水平。此时"reset 前进 + 低用量"永远不会成立，解除依赖连续两次 + // 观测都未触顶；单次观测无法排除耗尽后立刻拉到的旧快照，保持锁定。 + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, "acct_weekly_fallout").await; + let account = store.account("acct_weekly_fallout").expect("account"); + let server = MockServer::start().await; + let service = quota_service_with_base_url(&store, reqwest::Client::new(), server.uri()); + let mut old = usage((100, SHORT_RESET), (100, WEEK_RESET)); + old["rate_limit"]["allowed"] = json!(false); + old["rate_limit"]["limit_reached"] = json!(true); + mount_usage(&server, old).await; + service + .refresh_account(account.id()) + .await + .expect("seed exhausted quota"); + + // 5h 窗口滚动前进，周窗口 reset 不变但用量回落到正常水平：只标记候选。 + let mut first_observation = usage((0, SHORT_RESET + 18_000), (18, WEEK_RESET)); + first_observation["rate_limit"]["allowed"] = json!(true); + mount_usage(&server, first_observation).await; + let first = service + .refresh_account(account.id()) + .await + .expect("first fallout observation"); + assert!(first.quota().is_exhausted()); + + // 第二次连续观测仍未触顶：解除周窗口。 + let mut second_observation = usage((9, SHORT_RESET + 36_000), (18, WEEK_RESET)); + second_observation["rate_limit"]["allowed"] = json!(true); + mount_usage(&server, second_observation).await; + let second = service + .refresh_account(account.id()) + .await + .expect("second fallout observation"); + assert_eq!(second.quota().access(), QuotaAccessState::Allowed); +} + +#[tokio::test] +async fn a_single_same_window_fallout_observation_cannot_break_the_reset_baseline() { + // 连续性中断（观测到重新触顶）后必须重新积累两次未触顶证据。 + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, "acct_weekly_fallout_gap").await; + let account = store.account("acct_weekly_fallout_gap").expect("account"); + let server = MockServer::start().await; + let service = quota_service_with_base_url(&store, reqwest::Client::new(), server.uri()); + let mut old = usage((100, SHORT_RESET), (100, WEEK_RESET)); + old["rate_limit"]["allowed"] = json!(false); + mount_usage(&server, old).await; + service + .refresh_account(account.id()) + .await + .expect("seed exhausted quota"); + + mount_usage(&server, usage((0, SHORT_RESET + 18_000), (18, WEEK_RESET))).await; + assert!( + service + .refresh_account(account.id()) + .await + .expect("first candidate") + .quota() + .is_exhausted() + ); + // 观测到重新触顶，候选证据清零。 + mount_usage(&server, usage((0, SHORT_RESET + 36_000), (100, WEEK_RESET))).await; + assert!( + service + .refresh_account(account.id()) + .await + .expect("reached again") + .quota() + .is_exhausted() + ); + mount_usage(&server, usage((0, SHORT_RESET + 54_000), (18, WEEK_RESET))).await; + assert!( + service + .refresh_account(account.id()) + .await + .expect("candidate restarts") + .quota() + .is_exhausted() + ); + mount_usage(&server, usage((0, SHORT_RESET + 72_000), (18, WEEK_RESET))).await; + assert_eq!( + service + .refresh_account(account.id()) + .await + .expect("second consecutive observation") + .quota() + .access(), + QuotaAccessState::Allowed + ); +} + #[tokio::test] async fn monthly_and_arbitrary_periods_use_the_same_recovery_rule() { for seconds in [18_000, 86_400, 604_800, 864_000, 2_592_000, 7_776_000] {