diff --git a/backend/crates/gateway-core/src/account/mod.rs b/backend/crates/gateway-core/src/account/mod.rs index d8d9652a3..348306157 100644 --- a/backend/crates/gateway-core/src/account/mod.rs +++ b/backend/crates/gateway-core/src/account/mod.rs @@ -9,4 +9,5 @@ mod store; pub use error::CredentialError; pub use model::*; pub use selection::*; +pub(crate) use selection::{SMART_SCORE_TOLERANCE, smart_score}; pub use store::ProviderAccountStore; diff --git a/backend/crates/gateway-core/src/account/selection.rs b/backend/crates/gateway-core/src/account/selection.rs index 578de09b5..a3a9708b6 100644 --- a/backend/crates/gateway-core/src/account/selection.rs +++ b/backend/crates/gateway-core/src/account/selection.rs @@ -590,7 +590,7 @@ impl AccountSelector { }) } - fn scheduling_blocker( + pub(crate) fn scheduling_blocker( &self, candidate: &AccountCandidate, context: &AccountSelectionContext, @@ -643,7 +643,7 @@ const SMART_QUOTA_WEIGHT: f64 = 0.8; const SMART_FAILURE_WEIGHT: f64 = 1.0; const SMART_LATENCY_WEIGHT: f64 = 0.5; // 容忍 5 个百分点的单项负载/失败率差异,避免微小信号波动独占新会话。 -const SMART_SCORE_TOLERANCE: f64 = 0.05; +pub(crate) const SMART_SCORE_TOLERANCE: f64 = 0.05; // 首输出 10 秒时延迟得分减半;固定尺度不随其他候选账号变化。 const SMART_LATENCY_HALF_SCORE_MS: f64 = 10_000.0; @@ -677,7 +677,7 @@ fn select_smart_candidate<'a>( Some(ranked[index].0) } -fn smart_score(candidate: &AccountCandidate, default_concurrency: NonZeroU32) -> f64 { +pub(crate) fn smart_score(candidate: &AccountCandidate, default_concurrency: NonZeroU32) -> f64 { let load = 1.0 - capacity_utilization(candidate, default_concurrency).clamp(0.0, 1.0); let quota = candidate .signals diff --git a/backend/crates/gateway-core/src/diagnostics/mod.rs b/backend/crates/gateway-core/src/diagnostics/mod.rs index b5de9a0ef..ad3f346f8 100644 --- a/backend/crates/gateway-core/src/diagnostics/mod.rs +++ b/backend/crates/gateway-core/src/diagnostics/mod.rs @@ -1,6 +1,7 @@ //! 有界、请求局部的诊断证据;不参与路由、重试或客户端协议决策。 mod capture; +mod selection; mod stream; mod trace; diff --git a/backend/crates/gateway-core/src/diagnostics/selection.rs b/backend/crates/gateway-core/src/diagnostics/selection.rs new file mode 100644 index 000000000..7d146771e --- /dev/null +++ b/backend/crates/gateway-core/src/diagnostics/selection.rs @@ -0,0 +1,73 @@ +//! 复用选号时的输入,保留有界的候选评分与阻断原因。 + +use serde_json::json; + +use super::TraceContext; +use crate::account::{ + AccountCandidate, AccountSelection, AccountSelectionContext, AccountSelector, RotationStrategy, + SMART_SCORE_TOLERANCE, smart_score, +}; + +// 单个 trace 事件最多 4 KiB;优先保留实际选中的账号。 +const MAX_CANDIDATES: usize = 6; + +impl TraceContext { + /// 记录实际选号输入与结果;只复用评分规则,不参与调度决策。 + pub fn account_selection( + &self, + candidates: &[AccountCandidate], + context: &AccountSelectionContext, + selection: Option<&AccountSelection<'_>>, + ) { + if !self.is_enabled() { + return; + } + let selected = selection.as_ref().map(|s| s.candidate()); + let selected_id = selected.map(|c| c.account.id()); + let smart = context.policy.strategy() == RotationStrategy::Smart; + let observations = selected + .into_iter() + .chain( + candidates + .iter() + .filter(|c| Some(c.account.id()) != selected_id), + ) + .take(MAX_CANDIDATES) + .map(|candidate| { + let signals = &candidate.signals; + json!({ + "accountId": candidate.account.id().as_str(), + "weight": candidate.account.weight().get(), + "blocker": AccountSelector.scheduling_blocker(candidate, context) + .map(|blocker| format!("{blocker:?}")), + "inFlight": signals.in_flight, + "concurrencyLimit": candidate.account + .effective_concurrency(context.policy.max_concurrent_per_account()).get(), + "quotaRemainingBasisPoints": signals.quota_remaining_rank, + "quotaResetAtUnixMs": signals.quota_reset_at.and_then(|reset| + reset.duration_since(std::time::UNIX_EPOCH).ok() + ).map(|duration| duration.as_millis()), + "failureRateBasisPoints": signals.failure_rate_basis_points, + "firstOutputLatencyMs": signals.first_output_latency_ms, + "smartScore": smart.then(|| smart_score( + candidate, context.policy.max_concurrent_per_account() + )), + }) + }) + .collect::>(); + self.record( + "account.selection", + json!({ + "strategy": context.policy.strategy().as_str(), + "selectedAccountId": selected_id.map(|id| id.as_str()), + "preferredAccountId": context.preferred_account.as_ref().map(|id| id.as_str()), + "preferredResult": selection.as_ref().map(|s| format!("{:?}", s.preferred())), + "roundRobinCursor": context.round_robin_cursor, + "smartScoreTolerance": smart.then_some(SMART_SCORE_TOLERANCE), + "candidateCount": candidates.len(), + "omittedCandidates": candidates.len().saturating_sub(observations.len()), + "candidates": observations, + }), + ); + } +} diff --git a/backend/crates/gateway-core/src/diagnostics/trace.rs b/backend/crates/gateway-core/src/diagnostics/trace.rs index 01a056b96..624dbb122 100644 --- a/backend/crates/gateway-core/src/diagnostics/trace.rs +++ b/backend/crates/gateway-core/src/diagnostics/trace.rs @@ -65,6 +65,12 @@ struct TraceEvent { } impl TraceContext { + /// 是否已启用请求诊断,供调用方避免构造不会保存的事实。 + #[must_use] + pub const fn is_enabled(&self) -> bool { + self.state.is_some() + } + /// 一个模型请求只创建一次,所有 attempt 共享同一有界时间线。 #[must_use] pub fn new(request_id: &str) -> Self { @@ -306,6 +312,7 @@ impl TraceContext { !matches!( state.events[*index].stage, "attempt.started" + | "account.selection" | "account.selected" | "attempt.failed" | "retry.decided" diff --git a/backend/crates/gateway-core/tests/account/mod.rs b/backend/crates/gateway-core/tests/account/mod.rs index 7c15bbe43..c18cfd4e0 100644 --- a/backend/crates/gateway-core/tests/account/mod.rs +++ b/backend/crates/gateway-core/tests/account/mod.rs @@ -1,4 +1,5 @@ mod selection; +mod selection_diagnostics; use std::collections::{BTreeMap, BTreeSet}; use std::num::NonZeroU32; diff --git a/backend/crates/gateway-core/tests/account/selection_diagnostics.rs b/backend/crates/gateway-core/tests/account/selection_diagnostics.rs new file mode 100644 index 000000000..b28016e18 --- /dev/null +++ b/backend/crates/gateway-core/tests/account/selection_diagnostics.rs @@ -0,0 +1,92 @@ +use gateway_core::account::{AccountSelector, RotationStrategy}; +use gateway_core::diagnostics::TraceContext; +use serde_json::json; + +use super::{candidate, candidate_with_concurrency, context}; + +#[test] +fn selection_trace_should_distinguish_load_quota_and_affinity() { + let mut candidates = [ + candidate_with_concurrency("acct_74", 1, 10), + candidate_with_concurrency("acct_96", 0, 10), + ]; + candidates[0].signals.quota_remaining_rank = Some(2600); + candidates[1].signals.quota_remaining_rank = Some(400); + let mut context = context(RotationStrategy::Smart); + let trace = TraceContext::new("req_smart_trace"); + let selection = AccountSelector.select(&candidates, &context); + trace.account_selection(&candidates, &context, selection.as_ref()); + context.preferred_account = Some(candidates[1].account.id().clone()); + let selection = AccountSelector.select(&candidates, &context); + trace.account_selection(&candidates, &context, selection.as_ref()); + + let snapshot = trace.snapshot().expect("trace"); + let first = &snapshot["events"][0]["data"]; + let second = &snapshot["events"][1]["data"]; + assert_eq!(first["selectedAccountId"], "acct_74"); + assert_eq!(first["preferredResult"], "NotRequested"); + let busy = &first["candidates"][0]; + assert_eq!(busy["inFlight"], 1); + assert_eq!(busy["concurrencyLimit"], 10); + assert_eq!(busy["quotaRemainingBasisPoints"], 2600); + assert!(busy["blocker"].is_null()); + assert!( + busy["smartScore"].as_f64().expect("score") + > first["candidates"][1]["smartScore"] + .as_f64() + .expect("score") + ); + assert_eq!(second["selectedAccountId"], "acct_96"); + assert_eq!(second["preferredResult"], "Hit"); +} + +#[test] +fn selection_trace_should_bound_candidates_and_preserve_the_selected_account() { + let mut candidates = (0..30) + .map(|index| candidate(&format!("acct_{index}"), 0, None)) + .collect::>(); + candidates.push(candidate("acct_best", 0, Some(10_000))); + let trace = TraceContext::new("req_many_candidates"); + // 选号位于首部保留区之外,也应在长流的常规事件淘汰后保留。 + for _ in 0..10 { + trace.record("test.event", json!({})); + } + let context = context(RotationStrategy::Smart); + let selection = AccountSelector.select(&candidates, &context); + trace.account_selection(&candidates, &context, selection.as_ref()); + for index in 0..400 { + trace.record("test.event", json!({"index": index})); + } + + let snapshot = trace.snapshot().expect("trace"); + let event = snapshot["events"] + .as_array() + .expect("events") + .iter() + .find(|event| event["stage"] == "account.selection") + .expect("selection survives event eviction"); + let data = &event["data"]; + let recorded = data["candidates"].as_array().expect("bounded candidates"); + assert!(recorded.len() < candidates.len()); + assert_eq!(recorded[0]["accountId"], "acct_best"); + assert!(recorded[1]["quotaRemainingBasisPoints"].is_null()); + assert_eq!( + recorded.len() as u64 + data["omittedCandidates"].as_u64().expect("omitted"), + candidates.len() as u64 + ); + assert!(data.to_string().len() < 4096); +} + +#[test] +fn selection_trace_should_explain_when_all_accounts_are_busy() { + let candidates = [candidate_with_concurrency("acct_busy", 10, 10)]; + let trace = TraceContext::new("req_no_candidate"); + let context = context(RotationStrategy::Smart); + let selection = AccountSelector.select(&candidates, &context); + assert!(selection.is_none()); + trace.account_selection(&candidates, &context, selection.as_ref()); + let snapshot = trace.snapshot().expect("trace"); + let data = &snapshot["events"][0]["data"]; + assert!(data["selectedAccountId"].is_null()); + assert_eq!(data["candidates"][0]["blocker"], "ConcurrencyLimit"); +} diff --git a/backend/crates/providers/openai/src/credential/quota/snapshot.rs b/backend/crates/providers/openai/src/credential/quota/snapshot.rs index dfccd3353..caa6a34bd 100644 --- a/backend/crates/providers/openai/src/credential/quota/snapshot.rs +++ b/backend/crates/providers/openai/src/credential/quota/snapshot.rs @@ -236,25 +236,36 @@ pub(crate) fn quota_snapshot_from_observation( Some(snapshot) } -/// 调度投影缓存 TTL:`min(常规 TTL, 最近 active reset - now)`。 -/// reset 到期时缓存自然失效,不固定多等一个常规 TTL 周期。 +/// 缓存定期重读持久化事实,额度窗口独立按 reset 到期。 +/// 正常账号依赖被动同步,空闲超过缓存 TTL 不能把尚未重置的额度抹成未知。 pub(crate) fn quota_projection_ttl(snapshot: &CodexAccountQuotaSnapshot) -> Option { - let age = SystemTime::now() - .duration_since(snapshot.observed_at()) - .unwrap_or(Duration::ZERO); - let normal_ttl = QUOTA_SCHEDULING_TTL.checked_sub(age)?; let now = SystemTime::now(); - let reset_ttl = snapshot + snapshot .windows() .iter() - .filter_map(|window| window.reset_at()) - .map(SystemTime::from) - .filter(|reset| *reset > now) - .map(|reset| reset.duration_since(now).unwrap_or(Duration::ZERO)) - .min(); - let ttl = match reset_ttl { - Some(reset_ttl) if reset_ttl < normal_ttl => reset_ttl, - _ => normal_ttl, + .filter_map(|window| quota_window_ttl(window, snapshot.observed_at(), now)) + .min() + // 新观测即使没有有效窗口,也要用“未知”替换此前缓存的排序信号。 + .or_else(|| { + QUOTA_SCHEDULING_TTL.checked_sub( + now.duration_since(snapshot.observed_at()) + .unwrap_or(Duration::ZERO), + ) + }) + .filter(|ttl| !ttl.is_zero()) + .map(|ttl| ttl.min(QUOTA_SCHEDULING_TTL)) +} + +fn quota_window_ttl( + window: &CodexQuotaWindow, + observed_at: SystemTime, + now: SystemTime, +) -> Option { + let ttl = match window.reset_at() { + Some(reset_at) => SystemTime::from(reset_at).duration_since(now).ok()?, + // 没有明确窗口边界时,不能无限延长旧观测的有效期。 + None => QUOTA_SCHEDULING_TTL + .checked_sub(now.duration_since(observed_at).unwrap_or(Duration::ZERO))?, }; (!ttl.is_zero()).then_some(ttl) } @@ -273,12 +284,13 @@ pub(crate) fn quota_scheduling_signals( let mut remaining_rank: Option = None; let mut reset_at: Option = None; for window in &snapshot.windows { + if quota_window_ttl(window, snapshot.observed_at(), now).is_none() { + continue; + } let window_reset_at = window.reset_at().map(SystemTime::from); - let observation_expired = window_reset_at.is_some_and(|reset_at| reset_at <= now); - if !observation_expired - && let Some(used) = window - .used_percent() - .filter(|used| used.is_finite() && (0.0..=100.0).contains(used)) + if let Some(used) = window + .used_percent() + .filter(|used| used.is_finite() && (0.0..=100.0).contains(used)) { let rank = ((100.0 - used) * 100.0).round() as u64; remaining_rank = Some(remaining_rank.map_or(rank, |current| current.min(rank))); diff --git a/backend/crates/providers/openai/src/credential/selector.rs b/backend/crates/providers/openai/src/credential/selector.rs index c6040607c..0ba7228d1 100644 --- a/backend/crates/providers/openai/src/credential/selector.rs +++ b/backend/crates/providers/openai/src/credential/selector.rs @@ -512,7 +512,12 @@ impl CodexCredentialSelector { account_scope: request.attempt.account_scope().cloned(), }; let capacity = AccountSelector.capacity_snapshot(&candidates, &context); - let Some(selection) = AccountSelector.select(&candidates, &context) else { + let selection = AccountSelector.select(&candidates, &context); + request + .attempt + .trace() + .account_selection(&candidates, &context, selection.as_ref()); + let Some(selection) = selection else { return match shortest_retry { Some(retry_after) => Err(CredentialSelectionError::CapacityUnavailable { retry_after: Some(retry_after), diff --git a/backend/crates/providers/openai/tests/credential/quota/mod.rs b/backend/crates/providers/openai/tests/credential/quota/mod.rs index 48dcc25f2..f05f4070d 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 scheduling; mod slots; mod snapshot; diff --git a/backend/crates/providers/openai/tests/credential/quota/scheduling.rs b/backend/crates/providers/openai/tests/credential/quota/scheduling.rs new file mode 100644 index 000000000..98764671b --- /dev/null +++ b/backend/crates/providers/openai/tests/credential/quota/scheduling.rs @@ -0,0 +1,277 @@ +//! 持久化额度的有效期与短暂并发占用共同参与选号。 + +use std::collections::BTreeSet; +use std::num::NonZeroU32; +use std::sync::Arc; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +use gateway_core::account::{ + AccountCandidate, AccountConcurrencyLimit, AccountEligibilityPolicy, AccountRuntimeSignals, + AccountSelectionContext, AccountSelectionPolicy, AccountSelector, AccountWeight, + OpaqueProviderData, ProviderAccount, ProviderAccountStore as _, QuotaAccessState, + QuotaEvidence, QuotaObservation, QuotaState, RotationStrategy, +}; +use serde_json::{Value, json}; + +use super::{MemoryAccountStore, create_account, quota_service}; + +async fn persist_snapshot( + store: &Arc, + id: &str, + observed_at: SystemTime, + rate_limit: Value, +) -> ProviderAccount { + create_account(store, id).await; + let account = store.account(id).expect("created account"); + store + .compare_and_swap_quota(QuotaObservation { + account_id: account.id().clone(), + expected_revision: account.revision(), + quota: OpaqueProviderData::new( + json!({"rate_limit": rate_limit}) + .as_object() + .expect("quota object") + .clone(), + ), + observed_at, + state: QuotaState::allowed(observed_at), + }) + .await + .expect("persist quota"); + store.account(id).expect("persisted account") +} + +fn timestamp(time: SystemTime) -> u64 { + time.duration_since(UNIX_EPOCH).expect("epoch").as_secs() +} + +#[tokio::test] +async fn old_weekly_quota_should_outweigh_one_in_flight_title_request() { + let store = Arc::new(MemoryAccountStore::default()); + let now = SystemTime::now(); + let reset_at = timestamp(now + Duration::from_secs(6 * 3600)); + let mut accounts = Vec::new(); + for (id, used_percent, age_minutes) in [("acct_74", 74, 41), ("acct_96", 96, 16)] { + accounts.push( + persist_snapshot( + &store, + id, + now - Duration::from_secs(age_minutes * 60), + json!({ + "allowed": true, + "primary_window": { + "used_percent": used_percent, + "reset_at": reset_at, + "limit_window_seconds": 604_800 + } + }), + ) + .await + .with_scheduling( + Some(AccountConcurrencyLimit::new(10).expect("concurrency")), + AccountWeight::DEFAULT, + ), + ); + } + let service = quota_service(&store); + service.prepare_scheduling(&accounts).await; + let candidates = accounts + .into_iter() + .map(|account| AccountCandidate { + signals: AccountRuntimeSignals { + in_flight: u32::from(account.id().as_str() == "acct_74"), + last_started_at: None, + quota_reset_at: None, + quota_remaining_rank: None, + rate_limited_until: None, + failure_rate_basis_points: None, + first_output_latency_ms: None, + } + .with_provider_quota(service.scheduling_signals(&account)), + account, + }) + .collect::>(); + for cursor in 0..20 { + let context = AccountSelectionContext { + policy: AccountSelectionPolicy::new( + RotationStrategy::Smart, + NonZeroU32::new(3).expect("default concurrency"), + Duration::ZERO, + ), + now, + excluded_accounts: BTreeSet::new(), + preferred_account: None, + preferred_account_overrides_weight: true, + round_robin_cursor: cursor, + eligibility: AccountEligibilityPolicy::Enforce, + account_scope: None, + }; + let selected = AccountSelector + .select(&candidates, &context) + .expect("available account"); + assert_eq!(selected.candidate().account.id().as_str(), "acct_74"); + } +} + +#[tokio::test] +async fn old_quota_should_drop_reset_short_window_and_keep_weekly_window() { + let store = Arc::new(MemoryAccountStore::default()); + let now = SystemTime::now(); + let weekly_reset = now + Duration::from_secs(6 * 3600); + let account = persist_snapshot( + &store, + "acct_mixed_resets", + now - Duration::from_secs(41 * 60), + json!({ + "primary_window": { + "used_percent": 99, + "reset_at": timestamp(now - Duration::from_secs(60)), + "limit_window_seconds": 18_000 + }, + "secondary_window": { + "used_percent": 74, + "reset_at": timestamp(weekly_reset), + "limit_window_seconds": 604_800 + } + }), + ) + .await; + let service = quota_service(&store); + service + .read_account(account.id()) + .await + .expect("read quota"); + let signals = service.scheduling_signals(&account).expect("weekly quota"); + assert_eq!( + (signals.remaining_rank(), signals.reset_at().map(timestamp)), + (Some(2600), Some(timestamp(weekly_reset))) + ); +} + +#[tokio::test] +async fn undated_quota_should_expire_independently_of_dated_weekly_quota() { + for (age_minutes, expected_remaining) in [(1, 100), (41, 2600)] { + let store = Arc::new(MemoryAccountStore::default()); + let now = SystemTime::now(); + let account = persist_snapshot( + &store, + "acct_undated_window", + now - Duration::from_secs(age_minutes * 60), + json!({ + "primary_window": {"used_percent": 99}, + "secondary_window": { + "used_percent": 74, + "reset_at": timestamp(now + Duration::from_secs(6 * 3600)), + "limit_window_seconds": 604_800 + } + }), + ) + .await; + let service = quota_service(&store); + service + .prepare_scheduling(std::slice::from_ref(&account)) + .await; + assert_eq!( + service + .scheduling_signals(&account) + .and_then(|s| s.remaining_rank()), + Some(expected_remaining), + "observation age: {age_minutes} minutes" + ); + } +} + +#[tokio::test] +async fn old_undated_quota_should_remain_unknown_after_cache_hydration() { + let store = Arc::new(MemoryAccountStore::default()); + let account = persist_snapshot( + &store, + "acct_undated_only", + SystemTime::now() - Duration::from_secs(41 * 60), + json!({"primary_window": {"used_percent": 74}}), + ) + .await; + let service = quota_service(&store); + for _ in 0..2 { + service + .prepare_scheduling(std::slice::from_ref(&account)) + .await; + assert_eq!(service.scheduling_signals(&account), None); + } + assert_eq!(store.quota_reads(), 1); +} + +#[tokio::test] +async fn fresh_reset_observation_should_clear_previous_scheduling_signals() { + let store = Arc::new(MemoryAccountStore::default()); + let now = SystemTime::now(); + let account = persist_snapshot( + &store, + "acct_fresh_reset", + now, + json!({"primary_window": { + "used_percent": 74, + "reset_at": timestamp(now + Duration::from_secs(3600)) + }}), + ) + .await; + let service = quota_service(&store); + service + .prepare_scheduling(std::slice::from_ref(&account)) + .await; + assert_eq!( + service + .scheduling_signals(&account) + .and_then(|s| s.remaining_rank()), + Some(2600) + ); + + let headers = vec![ + ("x-codex-active-limit".to_owned(), "codex".to_owned()), + ("x-codex-primary-used-percent".to_owned(), "74".to_owned()), + ( + "x-codex-primary-reset-at".to_owned(), + timestamp(now - Duration::from_secs(60)).to_string(), + ), + ]; + assert!( + service + .synchronize_passive_headers(&account, &headers) + .await + .expect("reset observation") + ); + assert_eq!(service.scheduling_signals(&account), None); +} + +#[tokio::test] +async fn reset_window_should_not_clear_confirmed_exhaustion() { + let store = Arc::new(MemoryAccountStore::default()); + let now = SystemTime::now(); + let reset_at = now - Duration::from_secs(60); + let account = persist_snapshot( + &store, + "acct_exhausted_reset", + now - Duration::from_secs(41 * 60), + json!({"primary_window": {"used_percent": 100, "reset_at": timestamp(reset_at)}}), + ) + .await; + super::persist_quota_state( + &store, + &account, + QuotaState::exhausted(QuotaEvidence::ProviderDenied, now, Some(reset_at)), + ) + .await; + let service = quota_service(&store); + service + .prepare_scheduling(std::slice::from_ref(&account)) + .await; + assert_eq!(service.scheduling_signals(&account), None); + assert_eq!( + store + .account("acct_exhausted_reset") + .expect("account") + .quota() + .access(), + QuotaAccessState::Exhausted + ); +}