diff --git a/backend/crates/gateway-core/src/account/model.rs b/backend/crates/gateway-core/src/account/model.rs index 269c929f5..cc8f08ec1 100644 --- a/backend/crates/gateway-core/src/account/model.rs +++ b/backend/crates/gateway-core/src/account/model.rs @@ -90,18 +90,6 @@ impl AccountConcurrency { Self::Limited(limit) => Some(limit), } } - - /// 扣除当前请求不可占用的预留名额；不限并发不受影响，有限上限至少保留 1 个名额 - #[must_use] - pub const fn excluding_reserved(self, reserved: u32) -> Self { - match self { - Self::Unlimited => Self::Unlimited, - Self::Limited(limit) => match NonZeroU32::new(limit.get().saturating_sub(reserved)) { - Some(remaining) => Self::Limited(remaining), - None => Self::Limited(NonZeroU32::MIN), - }, - } - } } impl From for AccountConcurrency { diff --git a/backend/crates/gateway-core/src/account/selection.rs b/backend/crates/gateway-core/src/account/selection.rs index 79c03c3c6..098b622c7 100644 --- a/backend/crates/gateway-core/src/account/selection.rs +++ b/backend/crates/gateway-core/src/account/selection.rs @@ -206,6 +206,7 @@ impl AccountSelectionPolicy { /// Store 提供并发事实，Provider 叠加自己解释的额度事实；全部信号均可重建 #[derive(Debug, Clone, PartialEq, Eq)] pub struct AccountRuntimeSignals { + /// 本次调度容量池的在途数，不合并其他独立池 pub in_flight: u32, pub last_started_at: Option, pub quota_reset_at: Option, @@ -523,7 +524,7 @@ pub struct AccountSelectionContext { pub round_robin_cursor: u64, pub eligibility: AccountEligibilityPolicy, pub account_scope: Option>, - /// 本请求不可占用的每账号预留并发名额，由 Provider 按请求类别决定；0 表示不预留 + /// 本请求使用的独立预留池上限，由 Provider 按请求类别决定；0 表示使用普通账号池 pub reserved_concurrency: u32, } @@ -531,9 +532,11 @@ impl AccountSelectionContext { /// 本请求在该账号上可使用的并发上限；资格判断、租约与策略投影必须共用这一口径 #[must_use] pub fn concurrency_limit(&self, account: &ProviderAccount) -> AccountConcurrency { - account - .effective_concurrency(self.policy.max_concurrent_per_account()) - .excluding_reserved(self.reserved_concurrency) + if self.reserved_concurrency > 0 { + AccountConcurrency::new(self.reserved_concurrency) + } else { + account.effective_concurrency(self.policy.max_concurrent_per_account()) + } } } @@ -723,7 +726,6 @@ impl AccountSelector { let candidate = match context.policy.strategy() { RotationStrategy::QuotaResetPriority => { - let default_concurrency = context.policy.max_concurrent_per_account(); eligible.sort_by(|left, right| { ( left.signals.quota_reset_at.is_none(), @@ -734,8 +736,8 @@ impl AccountSelector { right.signals.quota_reset_at, )) .then_with(|| { - capacity_utilization(left, default_concurrency) - .total_cmp(&capacity_utilization(right, default_concurrency)) + capacity_utilization(left, context) + .total_cmp(&capacity_utilization(right, context)) }) .then_with(|| { left.signals @@ -751,13 +753,7 @@ impl AccountSelector { let index = context.round_robin_cursor as usize % eligible.len(); eligible.get(index).copied()? } - RotationStrategy::Smart => select_smart_candidate( - &eligible, - context.policy.max_concurrent_per_account(), - context.round_robin_cursor, - context.policy.smart_scheduling(), - context.now, - )?, + RotationStrategy::Smart => select_smart_candidate(&eligible, context)?, RotationStrategy::Sticky => { eligible.sort_by_key(|candidate| { ( @@ -843,17 +839,7 @@ impl AccountSelector { // 账号信号在队列锁外评分，锁内只读取实时队长并查表，避免扫描候选阻塞其他等待者 let scores = candidates .iter() - .map(|candidate| { - ( - candidate.account.id(), - smart_score( - candidate, - context.policy.max_concurrent_per_account(), - config, - context.now, - ), - ) - }) + .map(|candidate| (candidate.account.id(), smart_score(candidate, context))) .collect::>(); waiting .wait_with_priority(keys, |key, count| { @@ -914,13 +900,9 @@ const SMART_LATENCY_HALF_SCORE_MS: f64 = 10_000.0; // 距重置一小时时得分减半；未知和已过期时间不提供重置奖励 const SMART_RESET_HALF_SCORE_SECONDS: f64 = 3_600.0; -fn capacity_utilization( - candidate: &AccountCandidate, - default_concurrency: AccountConcurrency, -) -> f64 { - candidate - .account - .effective_concurrency(default_concurrency) +fn capacity_utilization(candidate: &AccountCandidate, context: &AccountSelectionContext) -> f64 { + context + .concurrency_limit(&candidate.account) .limit() .map_or(0.0, |limit| { f64::from(candidate.signals.in_flight) / f64::from(limit.get()) @@ -929,38 +911,29 @@ fn capacity_utilization( fn select_smart_candidate<'a>( candidates: &[&'a AccountCandidate], - default_concurrency: AccountConcurrency, - cursor: u64, - config: SmartSchedulingConfig, - now: SystemTime, + context: &AccountSelectionContext, ) -> Option<&'a AccountCandidate> { let mut ranked = candidates .iter() - .map(|candidate| { - ( - *candidate, - smart_score(candidate, default_concurrency, config, now), - ) - }) + .map(|candidate| (*candidate, smart_score(candidate, context))) .collect::>(); let best_score = ranked .iter() .map(|(_, score)| *score) .max_by(f64::total_cmp)?; - ranked.retain(|(_, score)| best_score - score <= config.score_tolerance()); + ranked.retain(|(_, score)| { + best_score - score <= context.policy.smart_scheduling().score_tolerance() + }); // 轮换顺序保持稳定，避免分数轻微交错与 cursor 同步后仍反复命中同一账号 ranked.sort_unstable_by(|(left, _), (right, _)| left.account.id().cmp(right.account.id())); - let index = (cursor % ranked.len() as u64) as usize; + let index = (context.round_robin_cursor % ranked.len() as u64) as usize; Some(ranked[index].0) } -pub(crate) fn smart_score( - candidate: &AccountCandidate, - default_concurrency: AccountConcurrency, - config: SmartSchedulingConfig, - now: SystemTime, -) -> f64 { - let load = 1.0 - capacity_utilization(candidate, default_concurrency).clamp(0.0, 1.0); +pub(crate) fn smart_score(candidate: &AccountCandidate, context: &AccountSelectionContext) -> f64 { + let config = context.policy.smart_scheduling(); + let now = context.now; + let load = 1.0 - capacity_utilization(candidate, context).clamp(0.0, 1.0); let quota = candidate .signals .quota_remaining_rank diff --git a/backend/crates/gateway-core/src/diagnostics/selection.rs b/backend/crates/gateway-core/src/diagnostics/selection.rs index dd5b4ed2c..1609bce12 100644 --- a/backend/crates/gateway-core/src/diagnostics/selection.rs +++ b/backend/crates/gateway-core/src/diagnostics/selection.rs @@ -41,17 +41,14 @@ impl TraceContext { "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(), + "concurrencyLimit": context.concurrency_limit(&candidate.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(), context.policy.smart_scheduling(), context.now - )), + "smartScore": smart.then(|| smart_score(candidate, context)), }) }) .collect::>(); diff --git a/backend/crates/gateway-core/src/provider_ports/lease.rs b/backend/crates/gateway-core/src/provider_ports/lease.rs index 6b3202085..7d7230f9e 100644 --- a/backend/crates/gateway-core/src/provider_ports/lease.rs +++ b/backend/crates/gateway-core/src/provider_ports/lease.rs @@ -9,6 +9,14 @@ use crate::{ use futures::future::BoxFuture; use std::{collections::BTreeMap, fmt, num::NonZeroU32, time::Duration}; +/// 账号内独立计数的调度容量池；Provider 决定请求归属，Store 只隔离租约 +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub enum ProviderConcurrencyPool { + #[default] + Shared, + Reserved, +} + /// 一个 Provider 的完整可重建调度状态 #[derive(Debug, Clone, PartialEq, Eq)] pub struct ProviderSchedulingState { @@ -46,6 +54,7 @@ pub struct ProviderSchedulingLeaseRequest { account_id: ProviderAccountId, credential_revision: CredentialRevision, max_concurrent: AccountConcurrency, + concurrency_pool: ProviderConcurrencyPool, request_interval: Duration, deadline: crate::lifecycle::Deadline, cancellation: crate::lifecycle::CancellationToken, @@ -66,6 +75,7 @@ impl ProviderSchedulingLeaseRequest { account_id, credential_revision, max_concurrent: max_concurrent.into(), + concurrency_pool: ProviderConcurrencyPool::Shared, request_interval, deadline: deadline.into(), cancellation: crate::lifecycle::CancellationToken::new(), @@ -78,6 +88,17 @@ impl ProviderSchedulingLeaseRequest { self } + #[must_use] + pub const fn with_concurrency_pool(mut self, pool: ProviderConcurrencyPool) -> Self { + self.concurrency_pool = pool; + self + } + + #[must_use] + pub const fn concurrency_pool(&self) -> ProviderConcurrencyPool { + self.concurrency_pool + } + #[must_use] pub fn cancellation(&self) -> crate::lifecycle::CancellationToken { self.cancellation.clone() @@ -145,11 +166,13 @@ pub enum ProviderLeaseRequest { } pub trait ProviderLeasePort: Send + Sync { + /// 按目标容量池读取在途数，账号最小请求间隔仍跨池共享 fn load_state<'a>( &'a self, client_api_key_id: &'a ClientApiKeyId, provider_kind: &'a ProviderKind, accounts: &'a [ProviderAccountId], + pool: ProviderConcurrencyPool, ) -> BoxFuture<'a, Result>; fn try_acquire( @@ -157,7 +180,7 @@ pub trait ProviderLeasePort: Send + Sync { request: ProviderLeaseRequest, ) -> BoxFuture<'_, Result>; - /// 读取指定账号当前的在途请求数；只用于容量熔断的峰值证据， + /// 读取指定账号普通池的在途请求数；只用于容量熔断的峰值证据， /// 支持租约信号的存储实现覆盖，否则视为不可观测（空映射） fn account_in_flight<'a>( &'a self, diff --git a/backend/crates/gateway-core/src/provider_ports/mod.rs b/backend/crates/gateway-core/src/provider_ports/mod.rs index 858c3e1b2..b0d80ba50 100644 --- a/backend/crates/gateway-core/src/provider_ports/mod.rs +++ b/backend/crates/gateway-core/src/provider_ports/mod.rs @@ -4,9 +4,9 @@ mod error; pub use error::{ProviderStoreError, ProviderStoreErrorKind}; mod lease; pub use lease::{ - ProviderLeaseAcquisition, ProviderLeaseGuard, ProviderLeasePort, ProviderLeaseRequest, - ProviderRefreshCapacityRequest, ProviderRefreshLeaseRequest, ProviderSchedulingLeaseRequest, - ProviderSchedulingState, + ProviderConcurrencyPool, ProviderLeaseAcquisition, ProviderLeaseGuard, ProviderLeasePort, + ProviderLeaseRequest, ProviderRefreshCapacityRequest, ProviderRefreshLeaseRequest, + ProviderSchedulingLeaseRequest, ProviderSchedulingState, }; mod affinity; pub use affinity::{ diff --git a/backend/crates/gateway-core/tests/account/mod.rs b/backend/crates/gateway-core/tests/account/mod.rs index 4c878a378..a40f0a340 100644 --- a/backend/crates/gateway-core/tests/account/mod.rs +++ b/backend/crates/gateway-core/tests/account/mod.rs @@ -15,13 +15,12 @@ use std::time::{Duration, SystemTime}; use serde_json::{Map, Value}; use gateway_core::account::{ - AccountAttemptFeedback, AccountCandidate, AccountConcurrency, AccountConcurrencyLimit, - AccountEligibilityPolicy, AccountFeedbackStats, AccountQuotaSignals, AccountRuntimeSignals, - AccountSchedulingBlocker, AccountSelectionContext, AccountSelectionPolicy, AccountSelector, - AccountStatus, AccountWeight, CredentialCasUpdate, CredentialRevision, CredentialState, - OpaqueProviderData, PlaintextCredential, PreferredAccountSelection, ProviderAccount, - ProviderAccountId, ProviderAccountIdentity, ProviderAccountUpdate, QuotaEvidence, QuotaState, - RotationStrategy, + AccountAttemptFeedback, AccountCandidate, AccountConcurrencyLimit, AccountEligibilityPolicy, + AccountFeedbackStats, AccountQuotaSignals, AccountRuntimeSignals, AccountSchedulingBlocker, + AccountSelectionContext, AccountSelectionPolicy, AccountSelector, AccountStatus, AccountWeight, + CredentialCasUpdate, CredentialRevision, CredentialState, OpaqueProviderData, + PlaintextCredential, PreferredAccountSelection, ProviderAccount, ProviderAccountId, + ProviderAccountIdentity, ProviderAccountUpdate, QuotaEvidence, QuotaState, RotationStrategy, }; use gateway_core::routing::{ ClientRoutingScope, FrozenAccountScope, ProviderKind, RuntimeAccount, RuntimeAccountDirectory, @@ -1099,40 +1098,57 @@ fn due_probe_freeze_blocks_scheduling_and_remains_visible_as_rate_limited() { } #[test] -fn reserved_concurrency_only_blocks_requests_that_cannot_use_the_reserved_slots() { +fn reserved_concurrency_uses_its_own_limit_without_reducing_normal_capacity() { + let normal = context(RotationStrategy::Smart); let candidates = [candidate_with_concurrency("acct_reserved", 4, 5)]; - let mut normal = context(RotationStrategy::Smart); - normal.reserved_concurrency = 1; - assert_eq!(normal.concurrency_limit(&candidates[0].account).get(), 4); - assert!(AccountSelector.select(&candidates, &normal).is_none()); + assert!(AccountSelector.select(&candidates, &normal).is_some()); let capacity = AccountSelector .capacity_snapshot(&candidates, &normal) .unwrap(); - assert_eq!((capacity.used_slots(), capacity.total_slots()), (4, 4)); - // 预留造成的暂满仍是可等待的容量约束，不能被当成不可用账号 - assert_eq!( - AccountSelector.wait_candidates(&candidates, &normal).len(), - 1 - ); - - let prioritized = context(RotationStrategy::Smart); - let capacity = AccountSelector - .capacity_snapshot(&candidates, &prioritized) - .unwrap(); assert_eq!((capacity.used_slots(), capacity.total_slots()), (4, 5)); - let selected = AccountSelector - .select(&candidates, &prioritized) - .expect("reserved slot available"); - assert_eq!(selected.candidate().account.id().as_str(), "acct_reserved"); - let full = [candidate_with_concurrency("acct_reserved", 5, 5)]; - assert!(AccountSelector.select(&full, &prioritized).is_none()); - // 预留不能让普通请求失去全部名额，也不改变不限并发 - let mut oversized = context(RotationStrategy::Smart); - oversized.reserved_concurrency = 10; - assert_eq!(oversized.concurrency_limit(&candidates[0].account).get(), 1); + let mut reserved = normal.clone(); + reserved.reserved_concurrency = 3; + let candidates = [candidate_with_concurrency("acct_reserved", 2, 5)]; + assert!(AccountSelector.select(&candidates, &reserved).is_some()); + let capacity = AccountSelector + .capacity_snapshot(&candidates, &reserved) + .unwrap(); + assert_eq!((capacity.used_slots(), capacity.total_slots()), (2, 3)); + let full = [candidate_with_concurrency("acct_reserved", 3, 5)]; + assert!(AccountSelector.select(&full, &reserved).is_none()); + assert_eq!(AccountSelector.wait_candidates(&full, &reserved).len(), 1); + assert!(AccountSelector.select(&full, &normal).is_some()); + + // 独立额度可以大于账号普通上限，极值不能回绕成无限并发 + reserved.reserved_concurrency = u32::MAX; assert_eq!( - AccountConcurrency::Unlimited.excluding_reserved(2), - AccountConcurrency::Unlimited + reserved.concurrency_limit(&candidates[0].account).get(), + u32::MAX ); + assert_eq!(normal.concurrency_limit(&candidates[0].account).get(), 5); +} + +#[test] +fn reserved_capacity_stays_bounded_when_normal_concurrency_is_unlimited() { + let candidates = [candidate("acct_unlimited", 3, None)]; + let mut normal = context(RotationStrategy::Smart); + normal.policy = AccountSelectionPolicy::new( + RotationStrategy::Smart, + gateway_core::account::AccountConcurrency::Unlimited, + Duration::ZERO, + ); + assert!(AccountSelector.select(&candidates, &normal).is_some()); + assert!( + AccountSelector + .capacity_snapshot(&candidates, &normal) + .is_none() + ); + let mut reserved = normal; + reserved.reserved_concurrency = 3; + assert!(AccountSelector.select(&candidates, &reserved).is_none()); + let capacity = AccountSelector + .capacity_snapshot(&candidates, &reserved) + .unwrap(); + assert_eq!((capacity.used_slots(), capacity.total_slots()), (3, 3)); } diff --git a/backend/crates/gateway-core/tests/account/smart_scheduling.rs b/backend/crates/gateway-core/tests/account/smart_scheduling.rs index bb8db1c24..3ab03f92d 100644 --- a/backend/crates/gateway-core/tests/account/smart_scheduling.rs +++ b/backend/crates/gateway-core/tests/account/smart_scheduling.rs @@ -13,6 +13,31 @@ use gateway_core::concurrency::{ use super::{candidate_with_concurrency, context, weighted_candidate}; +#[test] +fn reserved_pool_scoring_and_trace_use_its_capacity_instead_of_the_normal_limit() { + let mut context = context(RotationStrategy::Smart); + context.reserved_concurrency = 5; + context.policy = context.policy.with_smart_scheduling( + SmartSchedulingConfig::new([1.0, 0.0, 0.0, 0.0, 0.0, 0.0], false).unwrap(), + ); + let candidates = [ + candidate_with_concurrency("acct_large_normal", 2, 100), + candidate_with_concurrency("acct_small_normal", 1, 1), + ]; + let selection = AccountSelector.select(&candidates, &context).unwrap(); + assert_eq!( + selection.candidate().account.id().as_str(), + "acct_small_normal" + ); + let trace = gateway_core::diagnostics::TraceContext::new("req_reserved_scores"); + trace.account_selection(&candidates, &context, Some(&selection)); + let snapshot = trace.snapshot().unwrap(); + let selected = &snapshot["events"][0]["data"]["candidates"][0]; + assert_eq!(selected["concurrencyLimit"], 5); + assert_eq!(selected["inFlight"], 1); + assert_eq!(selected["smartScore"], 0.8); +} + #[test] fn smart_config_rejects_invalid_weights_and_incomplete_wire_values() { for value in [-0.1, 10.1, 0.01, f64::NAN, f64::INFINITY] { diff --git a/backend/crates/gateway-store/src/redis/credential_leases.rs b/backend/crates/gateway-store/src/redis/credential_leases.rs index 629059093..e674fcad4 100644 --- a/backend/crates/gateway-store/src/redis/credential_leases.rs +++ b/backend/crates/gateway-store/src/redis/credential_leases.rs @@ -12,7 +12,7 @@ use gateway_core::account::{AccountRuntimeSignals, ProviderAccountId}; use gateway_core::lifecycle::REQUEST_LEASE_TTL; use gateway_core::policy::ClientApiKeyId; use gateway_core::provider_ports::{ - ProviderLeaseAcquisition, ProviderLeasePort, ProviderLeaseRequest, + ProviderConcurrencyPool, ProviderLeaseAcquisition, ProviderLeasePort, ProviderLeaseRequest, ProviderRefreshCapacityRequest, ProviderSchedulingLeaseRequest, ProviderSchedulingState, ProviderStoreError, ProviderStoreErrorKind, }; @@ -128,6 +128,7 @@ return tostring(cursor - 1) pub enum CredentialLeaseScope { Provider, ProviderAccount, + ProviderAccountReserved, OAuthRefreshCapacity, OAuthRefresh, ProviderTask, @@ -138,11 +139,19 @@ impl CredentialLeaseScope { match self { Self::Provider => "provider", Self::ProviderAccount => "account", + Self::ProviderAccountReserved => "account-reserved", Self::OAuthRefreshCapacity => "refresh-capacity", Self::OAuthRefresh => "refresh", Self::ProviderTask => "task", } } + + const fn scheduling(pool: ProviderConcurrencyPool) -> Self { + match pool { + ProviderConcurrencyPool::Shared => Self::ProviderAccount, + ProviderConcurrencyPool::Reserved => Self::ProviderAccountReserved, + } + } } #[derive(Debug, Clone, PartialEq, Eq)] @@ -334,6 +343,7 @@ pub trait CredentialLeaseRepository: Send + Sync { request: &CredentialLeaseRequest, grant: &CredentialLeaseGrant, ) -> StoreResult; + /// 管理端容量与自适应并发观测只统计普通池，不把审批占用投影为普通并发压力 async fn credential_runtime_signals( &self, resource_ids: &[String], @@ -395,10 +405,16 @@ impl RedisCredentialLeaseRepository { let fingerprint = resource_fingerprint("credential lease", &request.resource_id)?; let tag = format!("{{{fingerprint}}}"); let prefix = format!("{}:lease:{}:{tag}", self.namespace, request.scope.as_str()); + // 两个容量池独立计数，但账号最小请求间隔仍由同一时间戳原子裁决 + let interval_prefix = if request.scope == CredentialLeaseScope::ProviderAccountReserved { + format!("{}:lease:account:{tag}", self.namespace) + } else { + prefix.clone() + }; Ok([ format!("{prefix}:active"), format!("{prefix}:fence"), - format!("{prefix}:last-started"), + format!("{interval_prefix}:last-started"), ]) } @@ -472,9 +488,29 @@ impl RedisCredentialLeaseRepository { }) } - async fn load_signal(&self, resource_id: String) -> StoreResult { + async fn runtime_signals( + &self, + resource_ids: &[String], + pool: ProviderConcurrencyPool, + ) -> StoreResult> { + for resource_id in resource_ids { + require_nonempty("credential runtime signal", "resource_id", resource_id)?; + } + let mut signals = Vec::with_capacity(resource_ids.len()); + for ids in resource_ids.chunks(super::ACCOUNT_STATE_READ_CONCURRENCY) { + let batch = join_all(ids.iter().cloned().map(|id| self.load_signal(id, pool))).await; + signals.extend(batch.into_iter().collect::>>()?); + } + Ok(signals) + } + + async fn load_signal( + &self, + resource_id: String, + pool: ProviderConcurrencyPool, + ) -> StoreResult { let request = CredentialLeaseRequest { - scope: CredentialLeaseScope::ProviderAccount, + scope: CredentialLeaseScope::scheduling(pool), resource_id: resource_id.clone(), owner_id: "signal-reader".to_owned(), ttl: Duration::from_secs(1), @@ -539,6 +575,7 @@ impl RedisProviderLeaseCoordinator { async fn load_signals( &self, accounts: &[ProviderAccountId], + pool: ProviderConcurrencyPool, ) -> Result, ProviderStoreError> { let ids = accounts .iter() @@ -546,7 +583,7 @@ impl RedisProviderLeaseCoordinator { .collect::>(); let signals = self .repository - .credential_runtime_signals(&ids) + .runtime_signals(&ids, pool) .await .map_err(|source| crate::provider_unavailable("load scheduling signals", source))?; signals @@ -588,7 +625,7 @@ impl RedisProviderLeaseCoordinator { let acquisition = self .repository .try_acquire_bounded_lease(&CredentialBoundedLeaseRequest { - scope: CredentialLeaseScope::ProviderAccount, + scope: CredentialLeaseScope::scheduling(request.concurrency_pool()), resource_id: request.account_id().as_str().to_owned(), owner_id: self.owner_id("request"), max_concurrent: request.max_concurrent().get(), @@ -646,9 +683,10 @@ impl ProviderLeasePort for RedisProviderLeaseCoordinator { client_api_key_id: &'a ClientApiKeyId, provider_kind: &'a ProviderKind, accounts: &'a [ProviderAccountId], + pool: ProviderConcurrencyPool, ) -> futures::future::BoxFuture<'a, Result> { Box::pin(async move { - let signals = self.load_signals(accounts).await?; + let signals = self.load_signals(accounts, pool).await?; let round_robin_cursor = self .next_scheduling_cursor(client_api_key_id, provider_kind) .await?; @@ -782,15 +820,8 @@ impl CredentialLeaseRepository for RedisCredentialLeaseRepository { &self, resource_ids: &[String], ) -> StoreResult> { - for resource_id in resource_ids { - require_nonempty("credential runtime signal", "resource_id", resource_id)?; - } - let mut signals = Vec::with_capacity(resource_ids.len()); - for ids in resource_ids.chunks(super::ACCOUNT_STATE_READ_CONCURRENCY) { - let batch = join_all(ids.iter().cloned().map(|id| self.load_signal(id))).await; - signals.extend(batch.into_iter().collect::>>()?); - } - Ok(signals) + self.runtime_signals(resource_ids, ProviderConcurrencyPool::Shared) + .await } async fn try_acquire_bounded_lease( diff --git a/backend/crates/gateway-store/tests/redis/credential_leases.rs b/backend/crates/gateway-store/tests/redis/credential_leases.rs index b5f2767c1..e00629e8e 100644 --- a/backend/crates/gateway-store/tests/redis/credential_leases.rs +++ b/backend/crates/gateway-store/tests/redis/credential_leases.rs @@ -34,95 +34,223 @@ fn only_account_scheduling_leases_allow_unlimited_concurrency() { } #[tokio::test] -async fn maintained_scheduling_lease_survives_initial_ttl_and_releases_on_drop() { +async fn reserved_and_normal_leases_enforce_independent_atomic_limits_and_release() { let Some((repository, _, _)) = repository().await else { return; }; - let mut request = scheduling_request("acct_long_request", "worker-long", 1, Duration::ZERO); - request.ttl = Duration::from_secs(1); - let lease = acquired( + let normal = scheduling_request("acct_independent", "normal", 2, Duration::ZERO); + let reserved = CredentialBoundedLeaseRequest { + scope: CredentialLeaseScope::ProviderAccountReserved, + max_concurrent: 3, + ..normal.clone() + }; + let attempts = futures::future::join_all((0..20).map(|index| { + let request = if index % 2 == 0 { + normal.clone() + } else { + reserved.clone() + }; + let repository = &repository; + async move { + ( + request.scope, + repository + .try_acquire_bounded_lease(&request) + .await + .unwrap(), + ) + } + })) + .await; + let mut normal_leases = Vec::new(); + let mut reserved_leases = Vec::new(); + for (scope, acquisition) in attempts { + if let CredentialBoundedLeaseAcquisition::Acquired(lease) = acquisition { + if scope == CredentialLeaseScope::ProviderAccount { + normal_leases.push(lease); + } else { + reserved_leases.push(lease); + } + } + } + assert_eq!((normal_leases.len(), reserved_leases.len()), (2, 3)); + assert_eq!( repository - .try_acquire_bounded_lease(&request) + .credential_runtime_signals(std::slice::from_ref(&normal.resource_id)) + .await + .unwrap()[0] + .in_flight, + 2 + ); + + reserved_leases.pop().unwrap().release().await.unwrap(); + assert!(matches!( + repository.try_acquire_bounded_lease(&normal).await.unwrap(), + CredentialBoundedLeaseAcquisition::Busy { .. } + )); + reserved_leases.push(acquired( + repository + .try_acquire_bounded_lease(&reserved) .await .unwrap(), - ); - let cancellation = gateway_core::lifecycle::CancellationToken::new(); - let lease = lease - .maintain(Default::default(), cancellation.clone()) - .unwrap(); - tokio::time::sleep(Duration::from_millis(1_100)).await; - assert!(!cancellation.is_cancelled()); + )); + normal_leases.pop().unwrap().release().await.unwrap(); assert!(matches!( repository - .try_acquire_bounded_lease(&request) + .try_acquire_bounded_lease(&reserved) .await .unwrap(), CredentialBoundedLeaseAcquisition::Busy { .. } )); - drop(lease); - tokio::time::timeout(Duration::from_secs(2), async { - loop { - if let CredentialBoundedLeaseAcquisition::Acquired(replacement) = repository + normal_leases.push(acquired( + repository.try_acquire_bounded_lease(&normal).await.unwrap(), + )); + for lease in normal_leases.into_iter().chain(reserved_leases) { + lease.release().await.unwrap(); + } + assert_eq!( + repository + .credential_runtime_signals(std::slice::from_ref(&normal.resource_id)) + .await + .unwrap()[0] + .in_flight, + 0 + ); +} + +#[tokio::test] +async fn independent_capacity_pools_preserve_the_shared_account_request_interval() { + let Some((repository, _, _)) = repository().await else { + return; + }; + for first_reserved in [false, true] { + let normal = scheduling_request( + &format!("acct_interval_{first_reserved}"), + "worker", + 2, + Duration::from_secs(30), + ); + let reserved = CredentialBoundedLeaseRequest { + scope: CredentialLeaseScope::ProviderAccountReserved, + ..normal.clone() + }; + let (first, second) = if first_reserved { + (&reserved, &normal) + } else { + (&normal, &reserved) + }; + acquired(repository.try_acquire_bounded_lease(first).await.unwrap()) + .release() + .await + .unwrap(); + assert!( + matches!(repository.try_acquire_bounded_lease(second).await.unwrap(), CredentialBoundedLeaseAcquisition::Busy { retry_after: Some(value) } if value > Duration::ZERO) + ); + } +} + +#[tokio::test] +async fn maintained_scheduling_lease_survives_initial_ttl_and_releases_on_drop() { + for scope in [ + CredentialLeaseScope::ProviderAccount, + CredentialLeaseScope::ProviderAccountReserved, + ] { + let Some((repository, _, _)) = repository().await else { + return; + }; + let mut request = scheduling_request("acct_long_request", "worker-long", 1, Duration::ZERO); + request.scope = scope; + request.ttl = Duration::from_secs(1); + let lease = acquired( + repository .try_acquire_bounded_lease(&request) .await - .unwrap() - { - replacement.release().await.unwrap(); - break; + .unwrap(), + ); + let cancellation = gateway_core::lifecycle::CancellationToken::new(); + let lease = lease + .maintain(Default::default(), cancellation.clone()) + .unwrap(); + tokio::time::sleep(Duration::from_millis(1_100)).await; + assert!(!cancellation.is_cancelled()); + assert!(matches!( + repository + .try_acquire_bounded_lease(&request) + .await + .unwrap(), + CredentialBoundedLeaseAcquisition::Busy { .. } + )); + drop(lease); + tokio::time::timeout(Duration::from_secs(2), async { + loop { + if let CredentialBoundedLeaseAcquisition::Acquired(replacement) = repository + .try_acquire_bounded_lease(&request) + .await + .unwrap() + { + replacement.release().await.unwrap(); + break; + } + tokio::time::sleep(Duration::from_millis(10)).await; } - tokio::time::sleep(Duration::from_millis(10)).await; - } - }) - .await - .expect("dropped request releases its slot"); + }) + .await + .expect("dropped request releases its slot"); + } } #[tokio::test] async fn maintained_scheduling_lease_does_not_resurrect_a_lost_slot() { - let Some((repository, _, _)) = repository().await else { - return; - }; - let request = scheduling_request("acct_lost_request", "worker-lost", 1, Duration::ZERO); - let lease = acquired( + for scope in [ + CredentialLeaseScope::ProviderAccount, + CredentialLeaseScope::ProviderAccountReserved, + ] { + let Some((repository, _, _)) = repository().await else { + return; + }; + let mut request = scheduling_request("acct_lost_request", "worker-lost", 1, Duration::ZERO); + request.scope = scope; + let lease = acquired( + repository + .try_acquire_bounded_lease(&request) + .await + .unwrap(), + ); + let grant = lease.grant().unwrap().clone(); + let raw = CredentialLeaseRequest { + scope: request.scope, + resource_id: request.resource_id.clone(), + owner_id: request.owner_id.clone(), + ttl: request.ttl, + }; repository - .try_acquire_bounded_lease(&request) + .release_credential_lease(&raw, &grant) .await - .unwrap(), - ); - let grant = lease.grant().unwrap().clone(); - let raw = CredentialLeaseRequest { - scope: request.scope, - resource_id: request.resource_id.clone(), - owner_id: request.owner_id.clone(), - ttl: request.ttl, - }; - repository - .release_credential_lease(&raw, &grant) - .await - .unwrap(); - let cancellation = gateway_core::lifecycle::CancellationToken::new(); - let lease = lease - .maintain(Default::default(), cancellation.clone()) - .unwrap(); - tokio::time::timeout(Duration::from_secs(2), cancellation.cancelled()) - .await - .expect("lost lease cancels the request"); - let replacement = acquired( - repository - .try_acquire_bounded_lease(&request) + .unwrap(); + let cancellation = gateway_core::lifecycle::CancellationToken::new(); + let lease = lease + .maintain(Default::default(), cancellation.clone()) + .unwrap(); + tokio::time::timeout(Duration::from_secs(2), cancellation.cancelled()) .await - .unwrap(), - ); - drop(lease); - tokio::time::sleep(Duration::from_millis(20)).await; - assert!(matches!( - repository - .try_acquire_bounded_lease(&request) - .await - .unwrap(), - CredentialBoundedLeaseAcquisition::Busy { .. } - )); - replacement.release().await.unwrap(); + .expect("lost lease cancels the request"); + let replacement = acquired( + repository + .try_acquire_bounded_lease(&request) + .await + .unwrap(), + ); + drop(lease); + tokio::time::sleep(Duration::from_millis(20)).await; + assert!(matches!( + repository + .try_acquire_bounded_lease(&request) + .await + .unwrap(), + CredentialBoundedLeaseAcquisition::Busy { .. } + )); + replacement.release().await.unwrap(); + } } #[tokio::test] diff --git a/backend/crates/providers/openai/src/credential/selector.rs b/backend/crates/providers/openai/src/credential/selector.rs index 7b2c5e0ef..7ff035086 100644 --- a/backend/crates/providers/openai/src/credential/selector.rs +++ b/backend/crates/providers/openai/src/credential/selector.rs @@ -16,10 +16,10 @@ use gateway_core::account::{ use gateway_core::concurrency::{CapacityWait, ConcurrencyWaitQueue, QueueRejection, WaitPriority}; use gateway_core::engine::{AttemptContext, ContinuationAttempt, policy::AccountPolicyError}; use gateway_core::provider_ports::{ - ProviderLeaseAcquisition, ProviderLeaseGuard, ProviderLeasePort, ProviderLeaseRequest, - ProviderSchedulingLeaseRequest, ProviderSessionAffinityKey, ProviderSessionAffinityPort, - ProviderSessionBinding, ProviderSessionExclusionPort, ProviderSessionExclusions, - ProviderStoreError, + ProviderConcurrencyPool, ProviderLeaseAcquisition, ProviderLeaseGuard, ProviderLeasePort, + ProviderLeaseRequest, ProviderSchedulingLeaseRequest, ProviderSessionAffinityKey, + ProviderSessionAffinityPort, ProviderSessionBinding, ProviderSessionExclusionPort, + ProviderSessionExclusions, ProviderStoreError, }; use gateway_core::routing::ProviderKind; use secrecy::ExposeSecret; @@ -114,7 +114,7 @@ struct CredentialSelectionInput<'a> { attempt: &'a AttemptContext, session_affinity_key: Option<&'a ProviderSessionAffinityKey>, session_affinity_observation: Option<&'a CodexSessionAffinity>, - /// Codex Guardian 自动审批请求；仅在配置了预留名额时获得预留与队列优先 + /// Codex Guardian 自动审批请求；仅在配置独立额度时使用审批容量池 guardian: bool, } @@ -126,6 +126,7 @@ pub(crate) struct CodexCyberPolicyScope { pub struct CodexCredentialSelector { waiting: ConcurrencyWaitQueue, + reserved_waiting: ConcurrencyWaitQueue, provider_kind: ProviderKind, repository: CodexCredentialRepository, leases: Arc, @@ -285,6 +286,7 @@ impl CodexCredentialSelector { cookie_policy, risk_recovery: Mutex::new(HashMap::new()), waiting: ConcurrencyWaitQueue::default(), + reserved_waiting: ConcurrencyWaitQueue::default(), account_feedback, } } @@ -465,15 +467,24 @@ impl CodexCredentialSelector { queue_policy.timeout = Duration::from_secs(30); } } - // 预留名额只对普通请求生效；Guardian 可用满全部名额，并在账号队列中排在普通请求之前 + // 审批使用独立容量与等待队列，任一池饱和都不能阻塞另一池取得空闲名额 let reserve = request .attempt .account_selection_policy() .openai_guardian_reserved_concurrency(); let prioritized = request.guardian && reserve > 0; - let reserved_concurrency = if prioritized { 0 } else { reserve }; + let reserved_concurrency = if prioritized { reserve } else { 0 }; + let concurrency_pool = if prioritized { + ProviderConcurrencyPool::Reserved + } else { + ProviderConcurrencyPool::Shared + }; let mut waiting = CapacityWait::new( - &self.waiting, + if prioritized { + &self.reserved_waiting + } else { + &self.waiting + }, queue_policy, request.attempt.deadline().at(), request.attempt.concurrency_wait_budget(), @@ -619,6 +630,7 @@ impl CodexCredentialSelector { request.attempt.client_api_key_ref(), &self.provider_kind, &account_ids, + concurrency_pool, ) .await?; let round_robin_cursor = scheduling.round_robin_cursor(); @@ -877,6 +889,7 @@ impl CodexCredentialSelector { policy.request_interval(), request.attempt.deadline(), ) + .with_concurrency_pool(concurrency_pool) .with_cancellation(request.attempt.cancellation().clone()), )) .await? diff --git a/backend/crates/providers/openai/tests/admin.rs b/backend/crates/providers/openai/tests/admin.rs index 9946579d9..c63a666f2 100644 --- a/backend/crates/providers/openai/tests/admin.rs +++ b/backend/crates/providers/openai/tests/admin.rs @@ -2545,6 +2545,7 @@ mod errors { _: &'a ClientApiKeyId, _: &'a ProviderKind, _: &'a [ProviderAccountId], + _pool: gateway_core::provider_ports::ProviderConcurrencyPool, ) -> BoxFuture<'a, Result> { panic!("manual refresh does not use scheduling leases") } diff --git a/backend/crates/providers/openai/tests/credential/refresh.rs b/backend/crates/providers/openai/tests/credential/refresh.rs index 1ae8824ff..cc07a3ed4 100644 --- a/backend/crates/providers/openai/tests/credential/refresh.rs +++ b/backend/crates/providers/openai/tests/credential/refresh.rs @@ -124,6 +124,7 @@ impl ProviderLeasePort for RefreshLeases { _: &'a ClientApiKeyId, _: &'a ProviderKind, _: &'a [ProviderAccountId], + _pool: gateway_core::provider_ports::ProviderConcurrencyPool, ) -> BoxFuture<'a, Result> { Box::pin(async { panic!("scheduled credential refresh does not load scheduling state") }) } diff --git a/backend/crates/providers/openai/tests/provider/contract/capacity.rs b/backend/crates/providers/openai/tests/provider/contract/capacity.rs index 62b8095d4..480c142f7 100644 --- a/backend/crates/providers/openai/tests/provider/contract/capacity.rs +++ b/backend/crates/providers/openai/tests/provider/contract/capacity.rs @@ -1,4 +1,4 @@ -//! 验证 OpenAI 请求等待账号容量与 Guardian 预留槽位 +//! 验证 OpenAI 请求等待账号容量与 Guardian 独立容量池 use futures::FutureExt; use gateway_core::account::AccountRuntimeSignals; @@ -123,7 +123,11 @@ async fn queued_root_and_new_child_both_observe_the_same_account_capacity() { server.verify().await; } -fn unqueued_context(request_id: &str, reserved: u32) -> AttemptContext { +fn capacity_context( + request_id: &str, + reserved: u32, + queue: ConcurrencyQueuePolicy, +) -> AttemptContext { AttemptContext::new( RequestAttemptContext::new( ModelRequestId::new(request_id).unwrap(), @@ -131,7 +135,9 @@ fn unqueued_context(request_id: &str, reserved: u32) -> AttemptContext { ), NonZeroU32::new(1).unwrap(), SystemTime::now() + Duration::from_secs(5), - account_policy().with_openai_guardian_reserved_concurrency(reserved), + account_policy() + .with_openai_guardian_reserved_concurrency(reserved) + .with_queue(queue), AccountAttemptContext::new(BTreeSet::new(), None, None) .with_account_scope(contract_account_scope()), None, @@ -158,7 +164,9 @@ fn subagent_operation(subagent: Option<&str>) -> Operation { } #[tokio::test] -async fn guardian_requests_can_use_the_reserved_slot_that_normal_requests_cannot() { +async fn guardian_and_normal_requests_use_independent_capacity_and_observations() { + use gateway_core::provider_ports::ProviderConcurrencyPool::{Reserved, Shared}; + let store = Arc::new(MemoryAccountStore::default()); create_account(&store, "acct_guardian").await; let account = ProviderAccountId::new("acct_guardian").unwrap(); @@ -171,90 +179,188 @@ async fn guardian_requests_can_use_the_reserved_slot_that_normal_requests_cannot .insert_header("content-type", "text/event-stream") .set_body_string(CAPTURE_COMPLETED_SSE), ) - .expect(3) + .expect(4) .mount(&server) .await; - // 默认账号上限为 2、预留 1：普通请求只能使用 1 个名额，Guardian 可以用满 2 个 let provider = provider_with_affinity_and_base_url_and_leases( &store, Arc::new(MemorySessionAffinity::default()), server.uri(), leases.clone(), ); - - let mut normal = provider - .clone() - .execute( - planned_request("openai", subagent_operation(Some("collab_spawn"))), - unqueued_context("req_guardian_normal_idle", 1), - ) - .await - .unwrap(); - while let Some(event) = normal.next().await { - event.unwrap(); - } - drop(normal); - let limits = |leases: &TestLeaseCoordinator| { + let signals = |in_flight| AccountRuntimeSignals { + in_flight, + last_started_at: None, + quota_reset_at: None, + quota_remaining_rank: None, + cooldown: None, + failure_rate_basis_points: None, + first_output_latency_ms: None, + }; + // 普通上限为 2，审批额度独立取配置值；关闭后新审批请求回到普通池 + for (id, normal_count, reserved_count, reserve, subagent, expected) in [ + ( + "req_large_reserve_normal", + 1, + 10, + 10, + None, + Some((Shared, 2)), + ), + ("req_normal_full", 2, 0, 1, None, None), + ( + "req_guardian_independent", + 2, + 0, + 1, + Some("guardian"), + Some((Reserved, 1)), + ), + ("req_guardian_full", 0, 1, 1, Some("guardian"), None), + ( + "req_normal_independent", + 1, + 1, + 1, + Some("collab_spawn"), + Some((Shared, 2)), + ), + ( + "req_guardian_disabled", + 1, + 1, + 0, + Some("guardian"), + Some((Shared, 2)), + ), + ] { leases - .requests + .signals .lock() .unwrap() - .iter() - .map(|request| request.max_concurrent().get()) - .collect::>() - }; - assert_eq!(limits(&leases), [1]); - - leases.signals.lock().unwrap().insert( - account.clone(), - AccountRuntimeSignals { - in_flight: 1, - last_started_at: None, - quota_reset_at: None, - quota_remaining_rank: None, - cooldown: None, - failure_rate_basis_points: None, - first_output_latency_ms: None, - }, - ); - assert!( - provider + .insert(account.clone(), signals(normal_count)); + leases + .reserved_signals + .lock() + .unwrap() + .insert(account.clone(), signals(reserved_count)); + leases.requests.lock().unwrap().clear(); + let result = provider .clone() .execute( - planned_request("openai", subagent_operation(None)), - unqueued_context("req_guardian_normal_busy", 1), + planned_request("openai", subagent_operation(subagent)), + capacity_context(id, reserve, ConcurrencyQueuePolicy::default()), ) + .await; + let Some((pool, limit)) = expected else { + assert!(result.is_err(), "{id} should wait for its own capacity"); + assert!(leases.requests.lock().unwrap().is_empty()); + continue; + }; + let mut stream = result.unwrap_or_else(|error| panic!("{id}: {error:?}")); + let capacity = stream + .metadata() + .selection_observation() + .unwrap() + .capacity() + .unwrap(); + let count = if pool == Shared { + normal_count + } else { + reserved_count + }; + assert_eq!( + (capacity.used_slots(), capacity.total_slots()), + (u64::from(count) + 1, limit) + ); + while let Some(event) = stream.next().await { + event.unwrap(); + } + drop(stream); + let requests = leases.requests.lock().unwrap(); + assert_eq!(requests.len(), 1); + assert_eq!(requests[0].concurrency_pool(), pool); + assert_eq!(u64::from(requests[0].max_concurrent().get()), limit); + } + server.verify().await; +} + +#[tokio::test] +async fn a_full_capacity_pool_does_not_block_the_other_pools_queue() { + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, "acct_guardian").await; + let account = ProviderAccountId::new("acct_guardian").unwrap(); + let leases = Arc::new(TestLeaseCoordinator::default()); + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/codex/responses")) + .respond_with( + ResponseTemplate::new(200) + .insert_header("content-type", "text/event-stream") + .set_body_string(CAPTURE_COMPLETED_SSE), + ) + .expect(4) + .mount(&server) + .await; + let provider = provider_with_affinity_and_base_url_and_leases( + &store, + Arc::new(MemorySessionAffinity::default()), + server.uri(), + leases.clone(), + ); + let queue = ConcurrencyQueuePolicy { + max_waiting: 1, + timeout: Duration::from_secs(2), + }; + for guardian_waits in [false, true] { + let blocked_signals = if guardian_waits { + &leases.reserved_signals + } else { + &leases.signals + }; + blocked_signals.lock().unwrap().insert( + account.clone(), + AccountRuntimeSignals { + in_flight: if guardian_waits { 1 } else { 2 }, + last_started_at: None, + quota_reset_at: None, + quota_remaining_rank: None, + cooldown: None, + failure_rate_basis_points: None, + first_output_latency_ms: None, + }, + ); + let mut waiting = Box::pin(provider.clone().execute( + planned_request( + "openai", + subagent_operation(guardian_waits.then_some("guardian")), + ), + capacity_context("req_pool_waiter", 1, queue), + )); + assert!(waiting.as_mut().now_or_never().is_none()); + + let mut available = provider + .clone() + .execute( + planned_request( + "openai", + subagent_operation((!guardian_waits).then_some("guardian")), + ), + capacity_context("req_other_pool", 1, queue), + ) .await - .is_err() - ); - assert_eq!(limits(&leases), [1]); + .expect("the other pool has independent capacity and queue position"); + while let Some(event) = available.next().await { + event.unwrap(); + } + drop(available); - let mut guardian = provider - .clone() - .execute( - planned_request("openai", subagent_operation(Some("guardian"))), - unqueued_context("req_guardian_reserved", 1), - ) - .await - .unwrap(); - assert_eq!(guardian.metadata().provider_account_id(), &account); - while let Some(event) = guardian.next().await { - event.unwrap(); + blocked_signals.lock().unwrap().clear(); + let mut resumed = waiting.await.expect("own pool capacity was released"); + while let Some(event) = resumed.next().await { + event.unwrap(); + } + drop(resumed); } - assert_eq!(limits(&leases), [1, 2]); - drop(guardian); - - // 同一个 Provider 的新请求读取关闭后的策略，不需要重新初始化选择器 - let mut unreserved = provider - .execute( - planned_request("openai", subagent_operation(None)), - unqueued_context("req_guardian_reservation_disabled", 0), - ) - .await - .unwrap(); - while let Some(event) = unreserved.next().await { - event.unwrap(); - } - assert_eq!(limits(&leases), [1, 2, 2]); server.verify().await; } diff --git a/backend/crates/providers/openai/tests/provider/contract/upstream_adapter.rs b/backend/crates/providers/openai/tests/provider/contract/upstream_adapter.rs index a0a150fcc..06ab16d2f 100644 --- a/backend/crates/providers/openai/tests/provider/contract/upstream_adapter.rs +++ b/backend/crates/providers/openai/tests/provider/contract/upstream_adapter.rs @@ -263,7 +263,7 @@ async fn selected_adapter_connection( #[tokio::test] async fn guardian_reservation_survives_upstream_adapters_and_metadata_precedence() { - // 上限 2、已有 1 个在途请求：预留启用时只有 Guardian 能继续取得租约 + // 普通池上限 2 且已满，审批独立池仍有空位；关闭独立池后审批也受普通容量约束 for (subagent, turn_kind, reserved, allowed) in [ (Some("guardian"), None, 1, true), (None, Some("guardian"), 1, true), @@ -271,7 +271,8 @@ async fn guardian_reservation_survives_upstream_adapters_and_metadata_precedence (Some("guardian"), Some("collab_spawn"), 1, false), (Some("collab_spawn"), None, 1, false), (None, None, 1, false), - (None, None, 0, true), + (None, None, 0, false), + (Some("guardian"), None, 0, false), ] { let store = Arc::new(MemoryAccountStore::default()); create_account(&store, "acct_provider_contract").await; @@ -279,7 +280,7 @@ async fn guardian_reservation_survives_upstream_adapters_and_metadata_precedence leases.signals.lock().unwrap().insert( ProviderAccountId::new("acct_provider_contract").unwrap(), AccountRuntimeSignals { - in_flight: 1, + in_flight: 2, last_started_at: None, quota_reset_at: None, quota_remaining_rank: None, @@ -355,7 +356,11 @@ async fn guardian_reservation_survives_upstream_adapters_and_metadata_precedence drop(receiver.await.unwrap()); let requests = leases.requests.lock().unwrap(); assert_eq!(requests.len(), 1); - assert_eq!(requests[0].max_concurrent().get(), 2); + assert_eq!(requests[0].max_concurrent().get(), 1); + assert_eq!( + requests[0].concurrency_pool(), + gateway_core::provider_ports::ProviderConcurrencyPool::Reserved + ); } else { assert!(leases.requests.lock().unwrap().is_empty()); } diff --git a/backend/crates/providers/openai/tests/support.rs b/backend/crates/providers/openai/tests/support.rs index 22b885be1..d05cd068b 100644 --- a/backend/crates/providers/openai/tests/support.rs +++ b/backend/crates/providers/openai/tests/support.rs @@ -673,6 +673,7 @@ pub(crate) struct TestLeaseCoordinator { pub(crate) busy: Mutex, pub(crate) busy_accounts: Mutex>, pub(crate) signals: Mutex>, + pub(crate) reserved_signals: Mutex>, round_robin_cursor: Mutex, } @@ -682,9 +683,17 @@ impl ProviderLeasePort for TestLeaseCoordinator { _client_api_key_id: &'a ClientApiKeyId, _provider_kind: &'a ProviderKind, accounts: &'a [ProviderAccountId], + pool: gateway_core::provider_ports::ProviderConcurrencyPool, ) -> BoxFuture<'a, Result> { Box::pin(async move { - let overrides = self.signals.lock().expect("scheduling signals lock"); + let overrides = match pool { + gateway_core::provider_ports::ProviderConcurrencyPool::Shared => &self.signals, + gateway_core::provider_ports::ProviderConcurrencyPool::Reserved => { + &self.reserved_signals + } + } + .lock() + .expect("scheduling signals lock"); let signals = accounts .iter() .map(|account| { diff --git a/backend/crates/providers/xai/src/credential/selector.rs b/backend/crates/providers/xai/src/credential/selector.rs index e99c156e6..88d1a7eb9 100644 --- a/backend/crates/providers/xai/src/credential/selector.rs +++ b/backend/crates/providers/xai/src/credential/selector.rs @@ -191,6 +191,7 @@ impl GrokAccountSessionSelector { request.client_api_key_id(), &self.provider_kind, &account_ids, + gateway_core::provider_ports::ProviderConcurrencyPool::Shared, ) .await .map_err(|source| { diff --git a/backend/crates/providers/xai/tests/admin.rs b/backend/crates/providers/xai/tests/admin.rs index 73031744b..16311fd12 100644 --- a/backend/crates/providers/xai/tests/admin.rs +++ b/backend/crates/providers/xai/tests/admin.rs @@ -715,6 +715,7 @@ impl ProviderLeasePort for TestLeases { _: &'a gateway_core::policy::ClientApiKeyId, _: &'a gateway_core::routing::ProviderKind, accounts: &'a [ProviderAccountId], + _pool: gateway_core::provider_ports::ProviderConcurrencyPool, ) -> BoxFuture< 'a, Result, @@ -1255,6 +1256,7 @@ mod errors { _: &'a ClientApiKeyId, _: &'a ProviderKind, _: &'a [ProviderAccountId], + _pool: gateway_core::provider_ports::ProviderConcurrencyPool, ) -> BoxFuture< 'a, Result, diff --git a/backend/crates/providers/xai/tests/credential/refresh.rs b/backend/crates/providers/xai/tests/credential/refresh.rs index 53f9407db..09cb6513f 100644 --- a/backend/crates/providers/xai/tests/credential/refresh.rs +++ b/backend/crates/providers/xai/tests/credential/refresh.rs @@ -265,6 +265,7 @@ impl ProviderLeasePort for TestRefreshLeases { _: &'a gateway_core::policy::ClientApiKeyId, _: &'a gateway_core::routing::ProviderKind, _: &'a [gateway_core::account::ProviderAccountId], + _pool: gateway_core::provider_ports::ProviderConcurrencyPool, ) -> futures::future::BoxFuture< 'a, Result, diff --git a/backend/crates/providers/xai/tests/credential/selector.rs b/backend/crates/providers/xai/tests/credential/selector.rs index 072bc6a22..3e040b8eb 100644 --- a/backend/crates/providers/xai/tests/credential/selector.rs +++ b/backend/crates/providers/xai/tests/credential/selector.rs @@ -59,6 +59,7 @@ impl ProviderLeasePort for SchedulingCoordinator { _: &'a ClientApiKeyId, _: &'a gateway_core::routing::ProviderKind, _: &'a [ProviderAccountId], + _pool: gateway_core::provider_ports::ProviderConcurrencyPool, ) -> futures::future::BoxFuture<'a, Result> { Box::pin(async move { Ok(ProviderSchedulingState::new( diff --git a/backend/crates/providers/xai/tests/provider/contract/mod.rs b/backend/crates/providers/xai/tests/provider/contract/mod.rs index 742cbb997..effeec37c 100644 --- a/backend/crates/providers/xai/tests/provider/contract/mod.rs +++ b/backend/crates/providers/xai/tests/provider/contract/mod.rs @@ -1839,6 +1839,7 @@ impl ProviderLeasePort for DiagnosticLeasePort { _: &'a ClientApiKeyId, _: &'a ProviderKind, account_ids: &'a [gateway_core::account::ProviderAccountId], + _pool: gateway_core::provider_ports::ProviderConcurrencyPool, ) -> futures::future::BoxFuture<'a, Result> { Box::pin(async move { Ok(ProviderSchedulingState::new( diff --git a/docs/api.md b/docs/api.md index 6de8899eb..2b644f963 100644 --- a/docs/api.md +++ b/docs/api.md @@ -644,7 +644,7 @@ config 返回 `{ name, plaintextKey }`，仅读取服务端会话绑定的当前 - `sortBy`: `email`、`status`、`planType`、`usage`、`lastUsedAt`、`expiresAt` - `sortDirection`: `asc`、`desc` -账号视图的 `capacity` 返回查询时的网关并发容量：`usedSlots` 是正在执行的请求占用数，不含排队请求， +账号视图的 `capacity` 返回查询时的普通并发容量：`usedSlots` 不含独立审批池和排队请求， 读取运行态失败时为 `null`；`totalSlots` 是应用账号独立配置或全局默认值后的上限，`null` 表示不限。 该上限不代表上游实际允许的并发数 @@ -1177,7 +1177,7 @@ HTTP 请求头及新建 WS 的握手提示按当时的最终出站档位构造 `providerCounts` 和 `clientKeyCount`。查询分组成员使用账号列表的 `groupId` 筛选， 不提供独立的分组成员路由；账号的 Provider 不代表整个分组的 Provider。 `capacity.totalSlots` 为 `number | null`：`null` 表示可用成员中存在继承无限并发的账号，`0` 表示没有可用槽位。 -`capacity.usedSlots` 继续返回实际在途数；Redis 不可用时为 `null` +分组容量只统计普通池，`capacity.usedSlots` 返回其实际在途数；Redis 不可用时为 `null` 分组费用按请求执行时实际服务账号的分组快照归属，不按 Client Key 绑定的分组分摊。 账号属于多个组时，各组均包含该请求费用；之后调整账号分组不重写历史归属 @@ -1361,12 +1361,13 @@ OpenAI 严格亲和下后代线程的会话账号等待不随普通账号排队 切换账号或内部重试不重新计时；若插件设置了请求总时限，等待也计入该时限。排队超时不用于中断已开始的上游生成。 设置更新请求须包含这三个字段，新请求使用更新后的快照 -`openaiGuardianReservedConcurrency`（默认 0，取值 0～4,294,967,295）为 Codex Guardian 自动审批保留账号并发，保存后对新请求生效。 +`openaiGuardianReservedConcurrency`（默认 0，取值 0～4,294,967,295）设置 Codex Guardian 自动审批的每账号独立并发，保存后对新请求生效。 Guardian 以 `subagent_kind` 或 `client_metadata.x-openai-subagent` 值 `guardian` 识别。取值 R 大于 0 时， -有限上限为 L 的账号对其他 OpenAI 请求只开放 `max(L − R, 1)` 个名额，Guardian 可用满 L； -开启账号排队后，Guardian 排在同账号已有 Guardian 之后、全部普通等待者之前，不受单账号排队上限约束， -仍受总等待容量与等待时限约束。不限并发的账号和关闭排队时的其余行为不变。 -设置更新请求须包含该字段 +普通请求保留账号原上限 L，Guardian 单独计数并最多运行 R 个，两类请求互不占用名额，也不借用对方空位。 +例如 L 为 10、R 为 3 时，可以同时运行 10 个普通请求和 3 个审批请求；普通并发不限时，审批仍受 R 限制。 +设为 0 时关闭独立额度，Guardian 使用普通并发。开启账号排队后，审批请求单独按 FIFO 等待， +不受普通队列位置或单账号排队上限约束，仍受自身队列总等待容量与等待时限约束。 +两类请求仍遵守账号最小请求间隔、可用性和 Client Key 限额。设置更新请求须包含该字段 `openaiAccountAffinity` 控制 OpenAI 账号亲和，默认 `strict`（严格），已有保存的模式保持不变： @@ -1710,7 +1711,7 @@ errorCode, errorMessage, startedAt, completedAt, expiresAt, createdAt, updatedAt ### Dashboard 容量与账号用量 -Dashboard 的 `capacityInfo.maxConcurrentPerAccount` 为默认账号并发上限，`0` 表示不限制。 +Dashboard 的 `capacityInfo` 统计普通容量，不含独立审批池；`maxConcurrentPerAccount` 为默认账号并发上限，`0` 表示不限制。 `capacityInfo.totalSlots` 为 `number | null`；可用账号池含无限并发账号时为 `null`，此时 `availableSlots` 也为 `null`。 `usedSlots` 仍表示实际在途数，Redis 不可用时为 `null`；没有可用账号时 `totalSlots` 为 `0` diff --git a/docs/architecture.md b/docs/architecture.md index 895edf0c7..18a97878b 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -657,21 +657,23 @@ Key 的 RPM 在成功准入时才计数，金额限制在入队前及成功准 切换等待层、账号或 Provider 不重置等待时限；没有发生排队时不启动该计时，也不据此中断已开始的上游生成。 队列满/超时属于本地容量拒绝，不作为上游限流反馈或 Provider 故障熔断证据 -#### Guardian 审批优先 +#### Guardian 独立并发 -Guardian 预留值由运行设置持久化并随调度快照冻结，每个请求沿用开始时的设置。 -Core 传递该值并拥有通用的并发上限与等待顺序，OpenAI Provider 负责识别请求类型并决定预留与优先级。 +Guardian 独立额度由运行设置持久化并随调度快照冻结，每个请求沿用开始时的设置。 +Core 拥有容量池、上限与等待合同，OpenAI Provider 负责识别请求类型并选择普通池或审批池。 - OpenAI Provider 在选号前通过统一协议解析识别 `guardian`；原生发送与插件上游适配器使用相同分类。 - 请求头形成的协议上下文与正文 metadata 沿用编码器的既有优先顺序；其他协议不得仅凭同名字段获得该优先级。 + 请求头形成的协议上下文与正文 metadata 沿用编码器的既有优先顺序；其他协议不得仅凭同名字段使用审批池。 适配器不因此继承原生会话亲和、续接或 WebSocket 约束，选号后的 attempt 中间件也不追溯改变已取得的租约。 -- `openaiGuardianReservedConcurrency = R > 0` 时，有限账号上限 L 对普通请求为 `max(L - R, 1)`， - Guardian 可用满 L，资格判断、策略投影、容量观测和原子租约共用同一请求级上限，数值边界见 [运行设置 API](api.md#8-运行设置) -- 开启账号排队时，Guardian 在普通等待者之前、已有 Guardian 之后；同级 FIFO，不抢占已运行请求。 - Guardian 可越过单账号等待数量上限，但仍受总等待容量、共享等待预算与截止时刻约束；取消及超时必须释放等待位置。 -- 配置为 0 时不启用预留或优先级；账号排队关闭时不引入隐式队列。Key 层准入仍按原有限额与排队规则执行。 +- `openaiGuardianReservedConcurrency = R > 0` 时，普通池使用账号原上限，审批池使用独立上限 R。 + 选号信号、负载评分、策略投影、请求容量观测和原子租约使用同一池的计数与上限，数值边界见 [运行设置 API](api.md#8-运行设置) +- Store 按账号和容量池隔离租约，复用续期、取消与释放生命周期，不复制总计数。 + 最小请求间隔仍按账号共享；管理端普通容量和自适应并发下调的观测只计普通池 +- 两个池分别持有等待队列，任一池的队首不会阻塞另一池。审批等待者保持 FIFO， + 可越过单账号等待数量上限，但仍受自身队列的总等待容量、共享等待预算与截止时刻约束；取消及超时必须释放等待位置 +- 配置为 0 时审批请求使用普通池；账号排队关闭时不引入隐式队列。Key 层准入仍按原有限额与排队规则执行 -等待者重新选号及取得租约前复核当前队首，异步读取期间被插队的普通请求继续等待，不凭已有 ticket 跳过优先顺序。 +等待者重新选号及取得租约前复核所属池的当前队首，不凭已有 ticket 跳过等待顺序 ### Client Key 限额与结算 diff --git a/frontend/src/views/settings/components/ConcurrencyPolicyCard.vue b/frontend/src/views/settings/components/ConcurrencyPolicyCard.vue index 7724eabe4..ce07a2e8e 100644 --- a/frontend/src/views/settings/components/ConcurrencyPolicyCard.vue +++ b/frontend/src/views/settings/components/ConcurrencyPolicyCard.vue @@ -58,21 +58,21 @@ const openaiGuardianReservedConcurrency = defineModel('openaiGuardianRes - + - + diff --git a/frontend/src/views/settings/composables/useSettingsForm.ts b/frontend/src/views/settings/composables/useSettingsForm.ts index 7ab66c9e6..70818c705 100644 --- a/frontend/src/views/settings/composables/useSettingsForm.ts +++ b/frontend/src/views/settings/composables/useSettingsForm.ts @@ -227,7 +227,7 @@ export function useSettingsForm() { return } if (!Number.isInteger(openaiGuardianReservedConcurrency) || openaiGuardianReservedConcurrency < 0 || openaiGuardianReservedConcurrency > 4294967295) { - toast.warning('自动审批预留并发应为 0～4294967295 的整数，0 表示关闭') + toast.warning('自动审批独立并发应为 0～4294967295 的整数，0 表示共用普通并发') return } if (responsesMaxDecompressedBodyMiB === null || !Number.isInteger(responsesMaxDecompressedBodyMiB) || responsesMaxDecompressedBodyMiB < 1