From 19ef4849694cb07671240d4890751761abe48199 Mon Sep 17 00:00:00 2001 From: Zyy Date: Wed, 4 Nov 2026 18:32:42 +0800 Subject: [PATCH] feat(peer-link)!: share upstream key concurrency and RPM limits Use a fresh upstream key concurrency limit as the account default for scheduling, leases, and capacity reporting. Show disclosed limits alongside the upstream pool summary. BREAKING CHANGE: Both peers must support CPR Link schema v3. Refs: #587 Co-authored-by: Codex --- backend/crates/gateway-admin/src/bootstrap.rs | 1 + .../gateway-admin/src/model/account_groups.rs | 1 + .../gateway-admin/src/model/accounts.rs | 2 +- .../gateway-admin/src/model/observability.rs | 4 +- .../gateway-admin/src/model/peer_link.rs | 1 + .../gateway-admin/src/model/upstream_info.rs | 2 + .../crates/gateway-admin/src/ports/peer.rs | 7 + .../src/use_case/account_groups.rs | 15 +- .../src/use_case/accounts/mod.rs | 12 +- .../src/use_case/observability/mod.rs | 14 +- .../src/use_case/peer_link/mod.rs | 1 + .../src/use_case/peer_link/upstream/mod.rs | 19 +++ .../use_case/peer_link/upstream/projection.rs | 23 +++ .../use_case/peer_link/upstream/scheduling.rs | 52 ++++--- .../use_case/peer_link/upstream/service.rs | 2 +- .../use_case/peer_link/upstream/snapshot.rs | 13 +- .../tests/use_case/account_groups.rs | 1 + .../tests/use_case/accounts/mod.rs | 5 + .../tests/use_case/credentials/peer.rs | 10 +- .../tests/use_case/observability/mod.rs | 4 +- .../tests/use_case/peer_link/upstream/mod.rs | 20 +-- .../use_case/peer_link/upstream/refresh.rs | 2 +- .../src/admin/accounts/upstream_info.rs | 4 + .../gateway-api/src/openai/peer_link/wire.rs | 12 +- backend/crates/gateway-api/tests/admin/mod.rs | 2 + .../tests/admin/observability/response.rs | 2 + .../gateway-api/tests/admin/upstream_info.rs | 10 +- .../gateway-api/tests/openai/peer_link/mod.rs | 15 +- .../crates/gateway-core/src/account/model.rs | 6 + .../gateway-core/src/account/selection.rs | 18 ++- .../gateway-core/src/account/upstream.rs | 5 +- .../gateway-core/src/diagnostics/selection.rs | 2 +- .../crates/gateway-core/src/engine/policy.rs | 2 +- .../crates/gateway-core/tests/account/mod.rs | 10 +- .../tests/engine/execution/mod.rs | 1 + .../gateway-core/tests/engine/policy.rs | 1 + .../runtime/tests/generation/policy/mod.rs | 1 + backend/crates/gateway-protocol/src/peer.rs | 13 +- .../src/postgres/account_groups.rs | 4 +- .../postgres/observability/admin_adapter.rs | 4 + .../src/postgres/provider_accounts/quota.rs | 26 +--- .../src/postgres/provider_accounts/rows.rs | 15 ++ .../src/postgres/provider_accounts/runtime.rs | 8 +- .../openai/src/credential/selector.rs | 60 ++++---- .../providers/openai/tests/admin/peer_info.rs | 134 +++++++++++++++++- .../tests/credential/quota/scheduling.rs | 1 + .../providers/xai/src/credential/selector.rs | 12 +- deploy/README.md | 6 +- docs/api.md | 28 ++-- docs/architecture.md | 1 + frontend/src/api/modules/accounts/types.ts | 2 + frontend/src/i18n/locales/en/accounts.json | 4 +- frontend/src/i18n/locales/en/keys.json | 2 +- .../src/i18n/locales/zh-Hans/accounts.json | 4 +- frontend/src/i18n/locales/zh-Hans/keys.json | 2 +- .../src/i18n/locales/zh-Hant/accounts.json | 4 +- frontend/src/i18n/locales/zh-Hant/keys.json | 2 +- .../account-quota-panel/UpstreamPool.vue | 30 ++-- .../components/account-quota-panel/index.vue | 5 +- 59 files changed, 511 insertions(+), 158 deletions(-) diff --git a/backend/crates/gateway-admin/src/bootstrap.rs b/backend/crates/gateway-admin/src/bootstrap.rs index ea1bcd2ce..da1c0aacc 100644 --- a/backend/crates/gateway-admin/src/bootstrap.rs +++ b/backend/crates/gateway-admin/src/bootstrap.rs @@ -197,6 +197,7 @@ async fn initialize_inner( auth, accounts: accounts.clone(), account_groups: Arc::new(DefaultAccountGroupService::new( + upstream_info.clone(), store.account_groups(), store.account_runtime(), snapshot.clone(), diff --git a/backend/crates/gateway-admin/src/model/account_groups.rs b/backend/crates/gateway-admin/src/model/account_groups.rs index b4ac7a82c..9a4bc4bbb 100644 --- a/backend/crates/gateway-admin/src/model/account_groups.rs +++ b/backend/crates/gateway-admin/src/model/account_groups.rs @@ -73,6 +73,7 @@ pub struct AccountGroupUsage { /// 当前页账号组 membership 对应的持久账号事实；运行态在 Admin query service 合并 #[derive(Debug, Clone, PartialEq, Eq)] pub struct AccountGroupMemberFact { + pub inherited_version: Option, pub group_id: AccountGroupId, pub account_id: String, pub status: AccountStatusFacts, diff --git a/backend/crates/gateway-admin/src/model/accounts.rs b/backend/crates/gateway-admin/src/model/accounts.rs index 14ca52e5d..0bd038f51 100644 --- a/backend/crates/gateway-admin/src/model/accounts.rs +++ b/backend/crates/gateway-admin/src/model/accounts.rs @@ -219,7 +219,7 @@ pub struct AccountPageItem { pub struct AccountCapacity { /// 实时租约读取失败时为 `None`，不能当作空闲 pub used_slots: Option, - /// 应用账号覆盖或全局默认值后的上限；`None` 表示不限 + /// 应用账号覆盖或继承默认值后的普通并发上限；`None` 表示不限 pub total_slots: Option, } diff --git a/backend/crates/gateway-admin/src/model/observability.rs b/backend/crates/gateway-admin/src/model/observability.rs index 0ab92f477..4c4e349c4 100644 --- a/backend/crates/gateway-admin/src/model/observability.rs +++ b/backend/crates/gateway-admin/src/model/observability.rs @@ -684,8 +684,10 @@ pub struct DashboardAccountUsage { } /// Dashboard 当前账号池的可重建运行时槽位事实 -#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[derive(Debug, Clone, PartialEq, Eq)] pub struct DashboardRuntimeSlots { + /// 仅未设置本地覆盖的账号可能继承有效互联观测 + pub inherited_versions: Vec, /// 继承运行参数默认并发上限的可调度账号数 pub inherited_accounts: u64, /// 已配置账号级并发上限的可调度账号槽位总数 diff --git a/backend/crates/gateway-admin/src/model/peer_link.rs b/backend/crates/gateway-admin/src/model/peer_link.rs index ba15a8e67..fe89a8acc 100644 --- a/backend/crates/gateway-admin/src/model/peer_link.rs +++ b/backend/crates/gateway-admin/src/model/peer_link.rs @@ -17,6 +17,7 @@ pub struct PeerLinkLimits { pub struct PeerLinkExport { pub observed_at: DateTime, pub expires_at: DateTime, + pub key_limits: gateway_core::policy::RateLimits, pub budget: ClientBudgetStatus, pub budget_access: ClientBudgetAccess, pub budget_observed_at: DateTime, diff --git a/backend/crates/gateway-admin/src/model/upstream_info.rs b/backend/crates/gateway-admin/src/model/upstream_info.rs index f50198fa1..bd2e60d14 100644 --- a/backend/crates/gateway-admin/src/model/upstream_info.rs +++ b/backend/crates/gateway-admin/src/model/upstream_info.rs @@ -10,6 +10,8 @@ pub struct ProviderUpstreamInfo { pub stale: bool, pub connection_revision: String, pub last_error: Option, + pub max_concurrency: Option, + pub requests_per_minute: Option, pub pool: Option, } diff --git a/backend/crates/gateway-admin/src/ports/peer.rs b/backend/crates/gateway-admin/src/ports/peer.rs index d5b460bed..f67c33588 100644 --- a/backend/crates/gateway-admin/src/ports/peer.rs +++ b/backend/crates/gateway-admin/src/ports/peer.rs @@ -38,6 +38,13 @@ pub trait PeerTransport: Send + Sync { /// 账号用例读取公共互联观测，显式刷新等待额度查询完成 #[async_trait] pub trait PeerUpstreamService: gateway_core::account::UpstreamInfoSource { + /// 仅在指定持久观测与当前调度缓存一致时估算容量 + fn observed_concurrency_limit( + &self, + version: &gateway_core::account::QuotaObservationVersion, + now: std::time::SystemTime, + ) -> Option; + async fn account_changed( &self, id: &gateway_core::account::ProviderAccountId, diff --git a/backend/crates/gateway-admin/src/use_case/account_groups.rs b/backend/crates/gateway-admin/src/use_case/account_groups.rs index 4bfe9354c..a3ba763c2 100644 --- a/backend/crates/gateway-admin/src/use_case/account_groups.rs +++ b/backend/crates/gateway-admin/src/use_case/account_groups.rs @@ -60,6 +60,7 @@ pub trait AccountGroupService: Send + Sync { } pub(crate) struct DefaultAccountGroupService { + upstream_info: Option>, store: Arc, runtime: Arc, snapshot: Arc, @@ -68,11 +69,13 @@ pub(crate) struct DefaultAccountGroupService { impl DefaultAccountGroupService { #[must_use] pub(crate) fn new( + upstream_info: Option>, store: Arc, runtime: Arc, snapshot: Arc, ) -> Self { Self { + upstream_info, store, runtime, snapshot, @@ -105,7 +108,7 @@ impl DefaultAccountGroupService { Default::default() } }; - project_group_runtime(records, &members, &runtime); + project_group_runtime(records, &members, &runtime, self.upstream_info.as_deref()); Ok(()) } @@ -215,6 +218,7 @@ fn project_group_runtime( records: &mut [AccountGroupRecord], members: &[AccountGroupMemberFact], runtime: &crate::model::accounts::AccountRuntimeSnapshot, + upstream: Option<&dyn crate::ports::peer::PeerUpstreamService>, ) { let now = SystemTime::now(); let mut status_by_account = BTreeMap::new(); @@ -228,7 +232,14 @@ fn project_group_runtime( .or_insert_with(|| resolve_account_status(&status, now).status); slots_by_account .entry(member.account_id.as_str()) - .or_insert(member.total_slots); + .or_insert_with(|| { + member + .inherited_version + .as_ref() + .and_then(|version| upstream?.observed_concurrency_limit(version, now)) + .map(|limit| u64::from(limit.get())) + .or(member.total_slots) + }); accounts_by_group .entry(member.group_id.as_str()) .or_default() diff --git a/backend/crates/gateway-admin/src/use_case/accounts/mod.rs b/backend/crates/gateway-admin/src/use_case/accounts/mod.rs index 177b3ea1e..cdc68cf2c 100644 --- a/backend/crates/gateway-admin/src/use_case/accounts/mod.rs +++ b/backend/crates/gateway-admin/src/use_case/accounts/mod.rs @@ -504,7 +504,11 @@ impl DefaultAccountsService { }); Ok(AccountDirectoryItem { protection_rules: stored.protection_rules, - capacity: stored.capacity, + capacity: super::peer_link::upstream::account_capacity( + &stored.account, + stored.capacity, + "a, + ), capabilities: provider .account_capabilities(account_id, &stored.account.authentication_kind), plan_type_display: self.providers.resolve_account_plan( @@ -618,7 +622,11 @@ impl AccountsService for DefaultAccountsService { used_slots: in_flight .as_ref() .map(|counts| counts.get(&item.account.id).copied().unwrap_or(0)), - ..item.capacity + ..super::peer_link::upstream::account_capacity( + &item.account, + item.capacity, + "a, + ) }, capabilities, plan_type_display: providers.resolve_account_plan( diff --git a/backend/crates/gateway-admin/src/use_case/observability/mod.rs b/backend/crates/gateway-admin/src/use_case/observability/mod.rs index 389dd9da2..d593c7b10 100644 --- a/backend/crates/gateway-admin/src/use_case/observability/mod.rs +++ b/backend/crates/gateway-admin/src/use_case/observability/mod.rs @@ -133,7 +133,7 @@ impl DefaultObservabilityService { self.settings.load_runtime_settings(), ) .map_err(|error| map_store_error(error, "dashboard"))?; - let runtime_slots = observation.runtime_slots; + let runtime_slots = observation.runtime_slots.as_ref(); self.enrich_list_records(&mut observation.recent_requests) .await; self.enrich_dashboard_quotas(&mut observation.account_usage) @@ -180,11 +180,19 @@ impl DefaultObservabilityService { .providers .dashboard_wire_profiles(&settings.request_profiles); let max_concurrent_per_account = u64::from(settings.values.max_concurrent_per_account); - let (inherited_accounts, overridden_slots) = runtime_slots - .as_ref() + let (mut inherited_accounts, mut overridden_slots) = runtime_slots .map_or((observation.provider_accounts.normal, 0), |slots| { (slots.inherited_accounts, slots.overridden_slots) }); + if let (Some(peer), Some(slots)) = (&self.upstream_info, runtime_slots) { + let now = std::time::SystemTime::now(); + for version in &slots.inherited_versions { + if let Some(limit) = peer.observed_concurrency_limit(version, now) { + inherited_accounts -= 1; + overridden_slots = overridden_slots.saturating_add(u64::from(limit.get())); + } + } + } let total_slots = (max_concurrent_per_account > 0 || inherited_accounts == 0).then(|| { inherited_accounts .saturating_mul(max_concurrent_per_account) diff --git a/backend/crates/gateway-admin/src/use_case/peer_link/mod.rs b/backend/crates/gateway-admin/src/use_case/peer_link/mod.rs index 6d66f8621..69f6521f4 100644 --- a/backend/crates/gateway-admin/src/use_case/peer_link/mod.rs +++ b/backend/crates/gateway-admin/src/use_case/peer_link/mod.rs @@ -117,6 +117,7 @@ impl PeerLinkService for DefaultPeerLinkService { .budget .limits .access(key.budget.daily_used_usd, key.budget.weekly_used_usd), + key_limits: key.limits, budget: key.budget, budget_observed_at, pool: None, diff --git a/backend/crates/gateway-admin/src/use_case/peer_link/upstream/mod.rs b/backend/crates/gateway-admin/src/use_case/peer_link/upstream/mod.rs index f22dce6a9..ac93b2cb1 100644 --- a/backend/crates/gateway-admin/src/use_case/peer_link/upstream/mod.rs +++ b/backend/crates/gateway-admin/src/use_case/peer_link/upstream/mod.rs @@ -25,6 +25,7 @@ use gateway_core::{ error::StoreError, task::{WorkerContribution, WorkerDefinitionError}, }; +pub(crate) use projection::account_capacity; use service::PeerInfoService; /// 组合根独立注册公共互联 Worker，不属于任何具体 Provider 的任务集合 @@ -62,6 +63,24 @@ pub fn initialize( #[async_trait::async_trait] impl PeerUpstreamService for PeerInfoService { + fn observed_concurrency_limit( + &self, + version: &gateway_core::account::QuotaObservationVersion, + now: std::time::SystemTime, + ) -> Option { + let observations = self + .observations + .read() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let cached = observations.get(&version.account_id)?; + if cached.observed_at != Some(version.observed_at) + || cached.connection.credential_revision != version.credential_revision + { + return None; + } + cached.scheduling.as_ref()?.concurrency_limit(now) + } + async fn account_changed(&self, id: &ProviderAccountId) -> Result<(), StoreError> { PeerInfoService::account_changed(self, id).await } diff --git a/backend/crates/gateway-admin/src/use_case/peer_link/upstream/projection.rs b/backend/crates/gateway-admin/src/use_case/peer_link/upstream/projection.rs index 5ec6076ef..f3ab5ebd4 100644 --- a/backend/crates/gateway-admin/src/use_case/peer_link/upstream/projection.rs +++ b/backend/crates/gateway-admin/src/use_case/peer_link/upstream/projection.rs @@ -16,6 +16,26 @@ use gateway_protocol::peer::{PeerComponentStatus, PeerLink}; use super::{PeerInfoService, quota, snapshot}; +pub(crate) fn account_capacity( + account: &crate::model::accounts::AccountRecord, + mut capacity: crate::model::accounts::AccountCapacity, + quota: &ProviderQuota, +) -> crate::model::accounts::AccountCapacity { + if account.concurrency_limit.is_none() + && let Some(info) = "a.upstream + && !info.stale + && info + .observed_at + .is_some_and(|at| snapshot::fresh_at(at.into(), SystemTime::now())) + && let Some(limit) = info + .max_concurrency + .and_then(gateway_core::account::AccountConcurrencyLimit::from_key_limit) + { + capacity.total_slots = Some(u64::from(limit.get())); + } + capacity +} + impl PeerInfoService { pub(super) fn project_peer( connection: &UpstreamConnection, @@ -30,6 +50,7 @@ impl PeerInfoService { let stale = peer.as_ref().is_none_or(|peer| { !snapshot::component_fresh(Some(&peer.observed_at), Some(&peer.expires_at), now) }); + let limits = peer.and_then(|peer| peer.key_limits); let info = ProviderUpstreamInfo { discovered: peer.is_some() || unavailable.is_some(), observed_at, @@ -38,6 +59,8 @@ impl PeerInfoService { last_error: last_error .or(unavailable) .map(|error| error.as_str().to_owned()), + max_concurrency: limits.and_then(|limits| limits.max_concurrency), + requests_per_minute: limits.and_then(|limits| limits.requests_per_minute), pool: peer .as_ref() .and_then(|peer| peer.pool.as_ref()) diff --git a/backend/crates/gateway-admin/src/use_case/peer_link/upstream/scheduling.rs b/backend/crates/gateway-admin/src/use_case/peer_link/upstream/scheduling.rs index 11b769146..1fb04fced 100644 --- a/backend/crates/gateway-admin/src/use_case/peer_link/upstream/scheduling.rs +++ b/backend/crates/gateway-admin/src/use_case/peer_link/upstream/scheduling.rs @@ -63,6 +63,7 @@ struct ModelObservation { pub(super) struct SchedulingObservation { validity: Validity, + concurrency_limit: Option, exhausted_budgets: Vec, models: HashMap>, } @@ -166,43 +167,52 @@ impl SchedulingObservation { } Some(Self { validity, + concurrency_limit: peer + .key_limits + .and_then(|limits| limits.max_concurrency) + .and_then(gateway_core::account::AccountConcurrencyLimit::from_key_limit), exhausted_budgets, models, }) } pub(super) fn fresh(&self, now: SystemTime) -> bool { - self.validity.fresh(now) - && now - .duration_since(self.validity.observed_at) - .unwrap_or_default() - <= snapshot::FRESH_FOR + self.validity.fresh(now) && snapshot::fresh_at(self.validity.observed_at, now) } - pub(super) fn signal(&self, model: &str, now: SystemTime) -> Option { + pub(super) fn concurrency_limit( + &self, + now: SystemTime, + ) -> Option { + self.concurrency_limit.filter(|_| self.fresh(now)) + } + + pub(super) fn signal( + &self, + model: Option<&str>, + now: SystemTime, + ) -> Option { if !self.fresh(now) { return None; } - if self + let budget_exhausted = self .exhausted_budgets .iter() - .any(|validity| validity.fresh(now)) - { - return Some(UpstreamSchedulingSignal { - exhausted: true, - remaining_percent: None, - }); - } - let model = self.models.get(model)?.as_ref()?; - if !model.validity.fresh(now) { + .any(|validity| validity.fresh(now)); + let model = model + .and_then(|id| self.models.get(id)?.as_ref()) + .filter(|model| model.validity.fresh(now)); + if !budget_exhausted && model.is_none() && self.concurrency_limit.is_none() { return None; } Some(UpstreamSchedulingSignal { - exhausted: model.exhausted, - remaining_percent: model - .account - .as_ref() - .and_then(|account| account.remaining_percent(now)), + concurrency_limit: self.concurrency_limit, + exhausted: budget_exhausted || model.is_some_and(|model| model.exhausted), + remaining_percent: if budget_exhausted { + None + } else { + model.and_then(|model| model.account.as_ref()?.remaining_percent(now)) + }, }) } } diff --git a/backend/crates/gateway-admin/src/use_case/peer_link/upstream/service.rs b/backend/crates/gateway-admin/src/use_case/peer_link/upstream/service.rs index 91722c281..0db9344ab 100644 --- a/backend/crates/gateway-admin/src/use_case/peer_link/upstream/service.rs +++ b/backend/crates/gateway-admin/src/use_case/peer_link/upstream/service.rs @@ -290,7 +290,7 @@ impl UpstreamInfoSource for PeerInfoService { fn signal( &self, account: &ProviderAccount, - model: &str, + model: Option<&str>, now: SystemTime, ) -> Option { if !account.enabled() { diff --git a/backend/crates/gateway-admin/src/use_case/peer_link/upstream/snapshot.rs b/backend/crates/gateway-admin/src/use_case/peer_link/upstream/snapshot.rs index 0f77155ae..1224cc6e5 100644 --- a/backend/crates/gateway-admin/src/use_case/peer_link/upstream/snapshot.rs +++ b/backend/crates/gateway-admin/src/use_case/peer_link/upstream/snapshot.rs @@ -89,10 +89,12 @@ pub(super) fn datetime(value: &str) -> Option> { } pub(super) fn fresh(observed: &str, now: SystemTime) -> bool { - timestamp(observed).is_some_and(|observed| { - observed <= now + Duration::from_secs(60) - && now.duration_since(observed).unwrap_or_default() <= FRESH_FOR - }) + timestamp(observed).is_some_and(|observed| fresh_at(observed, now)) +} + +pub(super) fn fresh_at(observed: SystemTime, now: SystemTime) -> bool { + observed <= now + Duration::from_secs(60) + && now.duration_since(observed).unwrap_or_default() <= FRESH_FOR } pub(super) fn component_fresh( @@ -162,6 +164,9 @@ pub(super) fn parse( } // 未授权能力不能借同名数据块越过逐层公开边界 let disclosed = |capability| peer.capabilities.iter().any(|value| value == capability); + if !disclosed("key_limits") { + peer.key_limits = None; + } if !disclosed("key_usage") { peer.usage = None; } diff --git a/backend/crates/gateway-admin/tests/use_case/account_groups.rs b/backend/crates/gateway-admin/tests/use_case/account_groups.rs index 5cff343ca..dd2d3ef65 100644 --- a/backend/crates/gateway-admin/tests/use_case/account_groups.rs +++ b/backend/crates/gateway-admin/tests/use_case/account_groups.rs @@ -373,6 +373,7 @@ fn group_record() -> AccountGroupRecord { fn member(account_id: &str, total_slots: u64) -> AccountGroupMemberFact { AccountGroupMemberFact { + inherited_version: None, group_id: group_id(), account_id: account_id.to_owned(), status: AccountStatusFacts { diff --git a/backend/crates/gateway-admin/tests/use_case/accounts/mod.rs b/backend/crates/gateway-admin/tests/use_case/accounts/mod.rs index 9fdbdc1b3..dfdc3dc6e 100644 --- a/backend/crates/gateway-admin/tests/use_case/accounts/mod.rs +++ b/backend/crates/gateway-admin/tests/use_case/accounts/mod.rs @@ -2686,12 +2686,15 @@ async fn peer_account_usage_should_follow_budget_windows_without_borrowing_pool_ for longest in [86_400, 604_800] { let provider = FakeProviderAdmin::new("openai", events()); let mut account = account_record("openai"); + account.concurrency_limit = None; account.authentication_kind = "api_key".to_owned(); account.created_at = Utc::now() - TimeDelta::days(60); let store = FakeAccountStore::with_account(account, events()); let reset_at = Utc::now() + TimeDelta::hours(1); let mut quota = empty_quota(); quota.upstream = Some(ProviderUpstreamInfo { + max_concurrency: Some(7), + requests_per_minute: Some(60), discovered: true, observed_at: Some(Utc::now()), stale: false, @@ -2762,6 +2765,8 @@ async fn peer_account_usage_should_follow_budget_windows_without_borrowing_pool_ page.items[0].usage.as_ref().unwrap().total_tokens, Some(1_234) ); + assert_eq!(page.items[0].capacity.total_slots, Some(7)); + assert_eq!(page.items[0].account.concurrency_limit, None); let queries = store.quota_window_queries(); assert_eq!(queries.len(), window_usage.len()); for (query, expected) in queries.iter().zip(&window_usage) { diff --git a/backend/crates/gateway-admin/tests/use_case/credentials/peer.rs b/backend/crates/gateway-admin/tests/use_case/credentials/peer.rs index a80736b8c..dc8d63efc 100644 --- a/backend/crates/gateway-admin/tests/use_case/credentials/peer.rs +++ b/backend/crates/gateway-admin/tests/use_case/credentials/peer.rs @@ -143,7 +143,7 @@ impl UpstreamInfoSource for PendingPeer { fn signal( &self, _: &ProviderAccount, - _: &str, + _: Option<&str>, _: std::time::SystemTime, ) -> Option { None @@ -152,6 +152,14 @@ impl UpstreamInfoSource for PendingPeer { #[async_trait] impl PeerUpstreamService for PendingPeer { + fn observed_concurrency_limit( + &self, + _: &gateway_core::account::QuotaObservationVersion, + _: std::time::SystemTime, + ) -> Option { + None + } + async fn account_changed(&self, id: &ProviderAccountId) -> Result<(), StoreError> { self.changed.lock().unwrap().push(id.as_str().into()); Ok(()) diff --git a/backend/crates/gateway-admin/tests/use_case/observability/mod.rs b/backend/crates/gateway-admin/tests/use_case/observability/mod.rs index bc2d9bbda..f15fcce9e 100644 --- a/backend/crates/gateway-admin/tests/use_case/observability/mod.rs +++ b/backend/crates/gateway-admin/tests/use_case/observability/mod.rs @@ -448,6 +448,7 @@ async fn dashboard_summary_should_project_rebuildable_runtime_slots() { let now = Utc::now(); let store = Arc::new(FixtureObservabilityStore::new(observation_range(now))); store.replace_runtime_slots(Some(DashboardRuntimeSlots { + inherited_versions: Vec::new(), inherited_accounts: 2, overridden_slots: 5, used_slots: Some(2), @@ -476,6 +477,7 @@ async fn dashboard_capacity_distinguishes_unlimited_inheritance_finite_overrides let now = Utc::now(); let store = Arc::new(FixtureObservabilityStore::new(observation_range(now))); store.replace_runtime_slots(Some(DashboardRuntimeSlots { + inherited_versions: Vec::new(), inherited_accounts, overridden_slots, used_slots: Some(2), @@ -1212,7 +1214,7 @@ impl ObservabilityStore for FixtureObservabilityStore { range: query.range, totals: Default::default(), provider_accounts: AccountPoolMetrics::default(), - runtime_slots: *self.runtime_slots.lock().expect("runtime slots"), + runtime_slots: self.runtime_slots.lock().expect("runtime slots").clone(), trend: self.trend.lock().expect("trend").clone(), account_usage: self.account_usage.lock().expect("account usage").clone(), recent_requests: Vec::new(), diff --git a/backend/crates/gateway-admin/tests/use_case/peer_link/upstream/mod.rs b/backend/crates/gateway-admin/tests/use_case/peer_link/upstream/mod.rs index 2e92dd52e..c70bb6117 100644 --- a/backend/crates/gateway-admin/tests/use_case/peer_link/upstream/mod.rs +++ b/backend/crates/gateway-admin/tests/use_case/peer_link/upstream/mod.rs @@ -241,7 +241,7 @@ async fn discovery_uses_provider_capabilities_and_publishes_neutral_signals() { let reader = UpstreamInfoReader::default(); assert!( reader - .signal(&initial_accounts[0], "model", SystemTime::now()) + .signal(&initial_accounts[0], Some("model"), SystemTime::now()) .is_none() ); let source = bundle.scheduling_source(); @@ -273,7 +273,11 @@ async fn discovery_uses_provider_capabilities_and_publishes_neutral_signals() { ); for account in &initial_accounts[..2] { if Some(account.provider().as_str()) == failed_provider { - assert!(reader.signal(account, "model", SystemTime::now()).is_none()); + assert!( + reader + .signal(account, Some("model"), SystemTime::now()) + .is_none() + ); continue; } let observed = bundle @@ -288,21 +292,21 @@ async fn discovery_uses_provider_capabilities_and_publishes_neutral_signals() { ); assert!( reader - .signal(account, "model", SystemTime::now()) + .signal(account, Some("model"), SystemTime::now()) .unwrap() .exhausted ); } assert!( reader - .signal(&initial_accounts[2], "model", SystemTime::now()) + .signal(&initial_accounts[2], Some("model"), SystemTime::now()) .is_none() ); drop(worker); drop(bundle); assert!( reader - .signal(&initial_accounts[0], "model", SystemTime::now()) + .signal(&initial_accounts[0], Some("model"), SystemTime::now()) .is_none() ); } @@ -332,7 +336,7 @@ async fn account_edits_preserve_observation_and_refresh_budget_until_connection_ assert!( bundle .scheduling_source() - .signal(&edited, "model", SystemTime::now()) + .signal(&edited, Some("model"), SystemTime::now()) .is_some() ); let after = service.refresh_quota(edited.id()).await.unwrap(); @@ -347,7 +351,7 @@ async fn account_edits_preserve_observation_and_refresh_budget_until_connection_ assert!( bundle .scheduling_source() - .signal(&reconnected, "model", SystemTime::now()) + .signal(&reconnected, Some("model"), SystemTime::now()) .is_none() ); service.refresh_quota(reconnected.id()).await.unwrap(); @@ -392,7 +396,7 @@ impl PeerTransport for TestTransport { .push(account.id().as_str().to_owned()); let now = Utc::now(); Ok(serde_json::to_vec(&serde_json::json!({ - "object":"cpr.peer", "schema_version":2, + "object":"cpr.peer", "schema_version":3, "observed_at":now.to_rfc3339(), "expires_at":(now + chrono::Duration::minutes(5)).to_rfc3339(), "capabilities":["key_usage"], diff --git a/backend/crates/gateway-admin/tests/use_case/peer_link/upstream/refresh.rs b/backend/crates/gateway-admin/tests/use_case/peer_link/upstream/refresh.rs index 99c1ef5f9..2d84a377d 100644 --- a/backend/crates/gateway-admin/tests/use_case/peer_link/upstream/refresh.rs +++ b/backend/crates/gateway-admin/tests/use_case/peer_link/upstream/refresh.rs @@ -216,7 +216,7 @@ async fn known_peer_failures_keep_identity_and_only_revocation_discards_shared_d assert_eq!( bundle .scheduling_source() - .signal(&account, "model", SystemTime::now()) + .signal(&account, Some("model"), SystemTime::now()) .is_none(), clears_data ); diff --git a/backend/crates/gateway-api/src/admin/accounts/upstream_info.rs b/backend/crates/gateway-api/src/admin/accounts/upstream_info.rs index 469a8a6e4..b6b3747c3 100644 --- a/backend/crates/gateway-api/src/admin/accounts/upstream_info.rs +++ b/backend/crates/gateway-api/src/admin/accounts/upstream_info.rs @@ -14,6 +14,8 @@ pub struct AccountUpstreamInfoView { pub stale: bool, pub connection_revision: String, pub last_error: Option, + pub max_concurrency: Option, + pub requests_per_minute: Option, pub pool: Option, } @@ -78,6 +80,8 @@ pub(crate) fn view( stale: info.stale, connection_revision: info.connection_revision, last_error: info.last_error, + max_concurrency: info.max_concurrency, + requests_per_minute: info.requests_per_minute, pool: info.pool.map(|pool| UpstreamPoolView { account_count: pool.account_count, states: UpstreamPoolStatesView { diff --git a/backend/crates/gateway-api/src/openai/peer_link/wire.rs b/backend/crates/gateway-api/src/openai/peer_link/wire.rs index e953c68ad..80c2f5462 100644 --- a/backend/crates/gateway-api/src/openai/peer_link/wire.rs +++ b/backend/crates/gateway-api/src/openai/peer_link/wire.rs @@ -4,14 +4,14 @@ use chrono::{DateTime, Utc}; use gateway_admin::model::peer_link::{PeerLinkExport, PeerModelState as ModelState}; use gateway_core::metering::Decimal; use gateway_protocol::peer::{ - OBJECT, PeerBudgetWindow, PeerComponentStatus, PeerKeyUsage, PeerLink, PeerModelAvailability, - PeerModelCoverage, PeerModelState, PeerPlanCount, PeerPool, PeerPoolStates, PeerQuota, - PeerQuotaAccount, PeerQuotaWindow, PeerUsage, SCHEMA_VERSION, + OBJECT, PeerBudgetWindow, PeerComponentStatus, PeerKeyLimits, PeerKeyUsage, PeerLink, + PeerModelAvailability, PeerModelCoverage, PeerModelState, PeerPlanCount, PeerPool, + PeerPoolStates, PeerQuota, PeerQuotaAccount, PeerQuotaWindow, PeerUsage, SCHEMA_VERSION, }; use std::time::SystemTime; pub(super) fn response(export: PeerLinkExport) -> PeerLink { - let mut capabilities = vec!["key_usage".to_owned()]; + let mut capabilities = vec!["key_usage".to_owned(), "key_limits".to_owned()]; let pool = export.pool.map(|pool| { capabilities.push("pool_summary".to_owned()); PeerPool { @@ -74,6 +74,10 @@ pub(super) fn response(export: PeerLinkExport) -> PeerLink { capabilities, observed_at: export.observed_at.to_rfc3339(), expires_at: export.expires_at.to_rfc3339(), + key_limits: Some(PeerKeyLimits { + max_concurrency: Some(export.key_limits.max_concurrency), + requests_per_minute: Some(export.key_limits.requests_per_minute), + }), usage: Some(PeerUsage { status: PeerComponentStatus::Available, observed_at: Some(export.budget_observed_at.to_rfc3339()), diff --git a/backend/crates/gateway-api/tests/admin/mod.rs b/backend/crates/gateway-api/tests/admin/mod.rs index 7110753d1..7a777ac6a 100644 --- a/backend/crates/gateway-api/tests/admin/mod.rs +++ b/backend/crates/gateway-api/tests/admin/mod.rs @@ -884,6 +884,7 @@ impl AccountGroupStore for MemoryAccountGroupStore { } Ok(vec![ AccountGroupMemberFact { + inherited_version: None, group_id: group_id(PRIMARY_GROUP_ID), account_id: "acct_group_ready".to_owned(), status: AccountStatusFacts { @@ -898,6 +899,7 @@ impl AccountGroupStore for MemoryAccountGroupStore { total_slots: Some(1), }, AccountGroupMemberFact { + inherited_version: None, group_id: group_id(PRIMARY_GROUP_ID), account_id: "acct_group_disabled".to_owned(), status: AccountStatusFacts { diff --git a/backend/crates/gateway-api/tests/admin/observability/response.rs b/backend/crates/gateway-api/tests/admin/observability/response.rs index 4dc801996..081672cae 100644 --- a/backend/crates/gateway-api/tests/admin/observability/response.rs +++ b/backend/crates/gateway-api/tests/admin/observability/response.rs @@ -702,6 +702,8 @@ fn peer_summary( ) -> gateway_admin::model::provider_credentials::ProviderQuota { gateway_admin::model::provider_credentials::ProviderQuota { upstream: Some(gateway_admin::model::upstream_info::ProviderUpstreamInfo { + max_concurrency: None, + requests_per_minute: None, discovered, observed_at: observed.then(chrono::Utc::now), stale: true, diff --git a/backend/crates/gateway-api/tests/admin/upstream_info.rs b/backend/crates/gateway-api/tests/admin/upstream_info.rs index 429c544f2..ee749a582 100644 --- a/backend/crates/gateway-api/tests/admin/upstream_info.rs +++ b/backend/crates/gateway-api/tests/admin/upstream_info.rs @@ -13,7 +13,7 @@ impl UpstreamInfoSource for TestPeerInfo { fn signal( &self, _: &ProviderAccount, - _: &str, + _: Option<&str>, _: SystemTime, ) -> Option { None @@ -22,6 +22,14 @@ impl UpstreamInfoSource for TestPeerInfo { #[async_trait] impl PeerUpstreamService for TestPeerInfo { + fn observed_concurrency_limit( + &self, + _: &gateway_core::account::QuotaObservationVersion, + _: std::time::SystemTime, + ) -> Option { + None + } + async fn account_changed(&self, _: &ProviderAccountId) -> Result<(), StoreError> { Ok(()) } diff --git a/backend/crates/gateway-api/tests/openai/peer_link/mod.rs b/backend/crates/gateway-api/tests/openai/peer_link/mod.rs index 45e0a0b14..b41e768d6 100644 --- a/backend/crates/gateway-api/tests/openai/peer_link/mod.rs +++ b/backend/crates/gateway-api/tests/openai/peer_link/mod.rs @@ -69,6 +69,10 @@ async fn fixture(sharing: ClientPeerSharing, plaintext: &str) -> (AdminTestFixtu .await .unwrap() .record; + key.limits = RateLimits { + max_concurrency: 5, + requests_per_minute: 60, + }; key.peer_sharing = sharing; key.groups = vec![group.clone()]; key.name = "private-key-sentinel".into(); @@ -243,7 +247,11 @@ async fn exports_scoped_observations_even_when_the_key_budget_is_exhausted_witho assert_eq!(response.status(), StatusCode::OK); let body = response_json(response).await; assert_eq!(body["object"], "cpr.peer"); - assert_eq!(body["schema_version"], 2); + assert_eq!(body["schema_version"], 3); + assert_eq!( + body["key_limits"], + json!({"max_concurrency":5,"requests_per_minute":60}) + ); assert_eq!(body["usage"]["data"]["windows"][0]["remaining"], "0"); assert_eq!(body["usage"]["data"]["allowed"], false); assert_eq!(body["usage"]["data"]["windows"][0]["limit_reached"], true); @@ -358,7 +366,10 @@ async fn plan_only_summary_limits_exported_data_and_preserves_unknown_on_quota_r body["pool"]["plans"], json!([{ "name": "Free", "count": 1 }]) ); - assert_eq!(body["capabilities"], json!(["key_usage", "pool_summary"])); + assert_eq!( + body["capabilities"], + json!(["key_usage", "key_limits", "pool_summary"]) + ); assert!(body.get("quota").is_none()); *fixture.provider_quota.lock().unwrap() = None; diff --git a/backend/crates/gateway-core/src/account/model.rs b/backend/crates/gateway-core/src/account/model.rs index f55f9fd44..a61366aaa 100644 --- a/backend/crates/gateway-core/src/account/model.rs +++ b/backend/crates/gateway-core/src/account/model.rs @@ -39,6 +39,12 @@ impl ProviderAccountId { pub struct AccountConcurrencyLimit(NonZeroU32); impl AccountConcurrencyLimit { + /// Key 的正数上限映射到账号可表达的范围，零沿用本机默认值 + #[must_use] + pub fn from_key_limit(limit: u64) -> Option { + Self::new(u32::try_from(limit).unwrap_or(u32::MAX)) + } + #[must_use] pub const fn new(value: u32) -> Option { match NonZeroU32::new(value) { diff --git a/backend/crates/gateway-core/src/account/selection.rs b/backend/crates/gateway-core/src/account/selection.rs index 552c110a8..93ad8f4c9 100644 --- a/backend/crates/gateway-core/src/account/selection.rs +++ b/backend/crates/gateway-core/src/account/selection.rs @@ -470,6 +470,8 @@ impl AccountQuotaSignals { /// 账号持久事实与可重建运行信号的请求级组合 #[derive(Debug, Clone, PartialEq, Eq)] pub struct AccountCandidate { + /// 本次选号已确认的上游默认上限，与本地可编辑覆盖独立 + pub inherited_concurrency: Option, pub account: ProviderAccount, pub signals: AccountRuntimeSignals, } @@ -525,11 +527,16 @@ pub struct AccountSelectionContext { impl AccountSelectionContext { /// 本请求在该账号上可使用的并发上限；资格判断、租约与策略投影必须共用这一口径 #[must_use] - pub fn concurrency_limit(&self, account: &ProviderAccount) -> AccountConcurrency { + pub fn concurrency_limit(&self, candidate: &AccountCandidate) -> AccountConcurrency { if self.reserved_concurrency > 0 { AccountConcurrency::new(self.reserved_concurrency) } else { - account.effective_concurrency(self.policy.max_concurrent_per_account()) + let default = candidate + .inherited_concurrency + .map_or(self.policy.max_concurrent_per_account(), |limit| { + limit.into_non_zero().into() + }); + candidate.account.effective_concurrency(default) } } } @@ -671,8 +678,7 @@ impl AccountSelector { ) }) .try_fold((0_u64, 0_u64), |(used, total), candidate| { - let capacity = - u64::from(context.concurrency_limit(&candidate.account).limit()?.get()); + let capacity = u64::from(context.concurrency_limit(candidate).limit()?.get()); Some(( used.saturating_add(u64::from(candidate.signals.in_flight)), total.saturating_add(capacity), @@ -868,7 +874,7 @@ impl AccountSelector { return Some(AccountSchedulingBlocker::ConcurrencyLimit); } if context - .concurrency_limit(&candidate.account) + .concurrency_limit(candidate) .limit() .is_some_and(|limit| candidate.signals.in_flight >= limit.get()) { @@ -885,7 +891,7 @@ const SMART_RESET_HALF_SCORE_SECONDS: f64 = 3_600.0; fn capacity_utilization(candidate: &AccountCandidate, context: &AccountSelectionContext) -> f64 { context - .concurrency_limit(&candidate.account) + .concurrency_limit(candidate) .limit() .map_or(0.0, |limit| { f64::from(candidate.signals.in_flight) / f64::from(limit.get()) diff --git a/backend/crates/gateway-core/src/account/upstream.rs b/backend/crates/gateway-core/src/account/upstream.rs index eead5631e..df9468c4d 100644 --- a/backend/crates/gateway-core/src/account/upstream.rs +++ b/backend/crates/gateway-core/src/account/upstream.rs @@ -76,6 +76,7 @@ impl UpstreamInfoFailure { /// 调度侧只消费已发布的中立观测，不读取协议、凭据或持久存储 #[derive(Debug, Clone, Copy, PartialEq)] pub struct UpstreamSchedulingSignal { + pub concurrency_limit: Option, pub exhausted: bool, pub remaining_percent: Option, } @@ -84,7 +85,7 @@ pub trait UpstreamInfoSource: Send + Sync { fn signal( &self, account: &ProviderAccount, - model: &str, + model: Option<&str>, now: SystemTime, ) -> Option; } @@ -110,7 +111,7 @@ impl UpstreamInfoSource for UpstreamInfoReader { fn signal( &self, account: &ProviderAccount, - model: &str, + model: Option<&str>, now: SystemTime, ) -> Option { self.source.get()?.upgrade()?.signal(account, model, now) diff --git a/backend/crates/gateway-core/src/diagnostics/selection.rs b/backend/crates/gateway-core/src/diagnostics/selection.rs index 1609bce12..fe3573bfe 100644 --- a/backend/crates/gateway-core/src/diagnostics/selection.rs +++ b/backend/crates/gateway-core/src/diagnostics/selection.rs @@ -41,7 +41,7 @@ impl TraceContext { "blocker": AccountSelector.scheduling_blocker(candidate, context) .map(|blocker| format!("{blocker:?}")), "inFlight": signals.in_flight, - "concurrencyLimit": context.concurrency_limit(&candidate.account).get(), + "concurrencyLimit": context.concurrency_limit(candidate).get(), "quotaRemainingBasisPoints": signals.quota_remaining_rank, "quotaResetAtUnixMs": signals.quota_reset_at.and_then(|reset| reset.duration_since(std::time::UNIX_EPOCH).ok() diff --git a/backend/crates/gateway-core/src/engine/policy.rs b/backend/crates/gateway-core/src/engine/policy.rs index 1d3b70167..a97370751 100644 --- a/backend/crates/gateway-core/src/engine/policy.rs +++ b/backend/crates/gateway-core/src/engine/policy.rs @@ -483,7 +483,7 @@ impl RequestPolicyContext { .is_none(), weight: candidate.account.weight().get(), in_flight: candidate.signals.in_flight, - maximum_concurrency: context.concurrency_limit(&candidate.account).get(), + maximum_concurrency: context.concurrency_limit(candidate).get(), last_started_at: candidate.signals.last_started_at, quota_reset_at: candidate.signals.quota_reset_at, quota_remaining_rank: candidate.signals.quota_remaining_rank, diff --git a/backend/crates/gateway-core/tests/account/mod.rs b/backend/crates/gateway-core/tests/account/mod.rs index c86858264..93e1e1634 100644 --- a/backend/crates/gateway-core/tests/account/mod.rs +++ b/backend/crates/gateway-core/tests/account/mod.rs @@ -50,6 +50,7 @@ fn account(id: &str) -> ProviderAccount { fn candidate(id: &str, in_flight: u32, remaining: Option) -> AccountCandidate { AccountCandidate { + inherited_concurrency: None, account: account(id), signals: AccountRuntimeSignals { in_flight, @@ -215,6 +216,7 @@ fn inconclusive_quota_observation_cannot_erase_confirmed_access() { fn diagnostic_selection_bypasses_all_local_account_eligibility() { let observed_at = SystemTime::now(); let exhausted = AccountCandidate { + inherited_concurrency: None, account: account("acct_exhausted").with_account_facts( true, CredentialState::Ready, @@ -233,6 +235,7 @@ fn diagnostic_selection_bypasses_all_local_account_eligibility() { }, }; let disabled = AccountCandidate { + inherited_concurrency: None, account: account("acct_disabled").with_account_facts( false, CredentialState::Ready, @@ -1113,11 +1116,8 @@ fn reserved_concurrency_uses_its_own_limit_without_reducing_normal_capacity() { // 独立额度可以大于账号普通上限，极值不能回绕成无限并发 reserved.reserved_concurrency = u32::MAX; - assert_eq!( - reserved.concurrency_limit(&candidates[0].account).get(), - u32::MAX - ); - assert_eq!(normal.concurrency_limit(&candidates[0].account).get(), 5); + assert_eq!(reserved.concurrency_limit(&candidates[0]).get(), u32::MAX); + assert_eq!(normal.concurrency_limit(&candidates[0]).get(), 5); } #[test] diff --git a/backend/crates/gateway-core/tests/engine/execution/mod.rs b/backend/crates/gateway-core/tests/engine/execution/mod.rs index 9b0c594fc..d899df6ee 100644 --- a/backend/crates/gateway-core/tests/engine/execution/mod.rs +++ b/backend/crates/gateway-core/tests/engine/execution/mod.rs @@ -2114,6 +2114,7 @@ impl Provider for PolicySelectingProvider { ) .with_scheduling(None, AccountWeight::new(100).unwrap()); let candidates = [AccountCandidate { + inherited_concurrency: None, account, signals: AccountRuntimeSignals { in_flight: 0, diff --git a/backend/crates/gateway-core/tests/engine/policy.rs b/backend/crates/gateway-core/tests/engine/policy.rs index 91f8f5c33..10e47a59a 100644 --- a/backend/crates/gateway-core/tests/engine/policy.rs +++ b/backend/crates/gateway-core/tests/engine/policy.rs @@ -55,6 +55,7 @@ fn account_id(id: &str) -> ProviderAccountId { fn candidate(id: &str) -> AccountCandidate { AccountCandidate { + inherited_concurrency: None, account: ProviderAccount::new( account_id(id), ProviderKind::new("openai").unwrap(), diff --git a/backend/crates/gateway-plugin/runtime/tests/generation/policy/mod.rs b/backend/crates/gateway-plugin/runtime/tests/generation/policy/mod.rs index 14acd116e..583cb485d 100644 --- a/backend/crates/gateway-plugin/runtime/tests/generation/policy/mod.rs +++ b/backend/crates/gateway-plugin/runtime/tests/generation/policy/mod.rs @@ -608,6 +608,7 @@ async fn model_router_can_delegate_after_a_managed_http_side_effect() { fn candidate(id: &str, weight: u16) -> AccountCandidate { AccountCandidate { + inherited_concurrency: None, account: ProviderAccount::new( ProviderAccountId::new(id).unwrap(), ProviderKind::new("example").unwrap(), diff --git a/backend/crates/gateway-protocol/src/peer.rs b/backend/crates/gateway-protocol/src/peer.rs index 7fe9648fe..3b4226335 100644 --- a/backend/crates/gateway-protocol/src/peer.rs +++ b/backend/crates/gateway-protocol/src/peer.rs @@ -3,7 +3,7 @@ use serde::{Deserialize, Serialize}; pub const OBJECT: &str = "cpr.peer"; -pub const SCHEMA_VERSION: u32 = 2; +pub const SCHEMA_VERSION: u32 = 3; pub const MAX_ACCOUNTS: usize = 3_000; pub const MAX_QUOTA_WINDOWS: usize = 6_000; pub const MAX_MODELS: usize = 2_000; @@ -20,6 +20,8 @@ pub struct PeerLink { #[serde(default, skip_serializing_if = "Option::is_none")] pub usage: Option, #[serde(default, skip_serializing_if = "Option::is_none")] + pub key_limits: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] pub pool: Option, #[serde(default, skip_serializing_if = "Option::is_none")] pub quota: Option, @@ -27,6 +29,15 @@ pub struct PeerLink { pub models: Vec, } +/// 当前认证密钥公开的请求上限，缺失表示未知，零表示不限 +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +pub struct PeerKeyLimits { + #[serde(default, skip_serializing_if = "Option::is_none")] + pub max_concurrency: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub requests_per_minute: Option, +} + #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] pub enum PeerComponentStatus { diff --git a/backend/crates/gateway-store/src/postgres/account_groups.rs b/backend/crates/gateway-store/src/postgres/account_groups.rs index 9a121f4be..5e7d0b5ec 100644 --- a/backend/crates/gateway-store/src/postgres/account_groups.rs +++ b/backend/crates/gateway-store/src/postgres/account_groups.rs @@ -243,7 +243,7 @@ impl AccountGroupStore for PgAccountGroupRepository { account.authentication_kind, account.credential_revision, account.outbound_proxy_id, account.outbound_proxy_url, account.has_refresh_token, account.access_token_expires_at, account.next_refresh_at, account.enabled, account.concurrency_limit, - account.weight, account.model_access_json, account.credential_state, account.quota_access_state, + account.quota_observed_at, account.weight, account.model_access_json, account.credential_state, account.quota_access_state, account.quota_evidence, account.quota_access_observed_at, account.quota_reset_at, account.last_error_reason, account.last_error_message, account.credential_observed_at, @@ -277,8 +277,10 @@ impl AccountGroupStore for PgAccountGroupRepository { .map_err(|source| { invalid("invalid default account concurrency").with_source(source) })?; + let version = super::provider_accounts::quota_version_from_row(&row)?; let account = account_summary_from_row(row)?; Ok(AccountGroupMemberFact { + inherited_version: version.filter(|_| account.concurrency_limit.is_none()), group_id, account_id: account.id, status: AccountStatusFacts { diff --git a/backend/crates/gateway-store/src/postgres/observability/admin_adapter.rs b/backend/crates/gateway-store/src/postgres/observability/admin_adapter.rs index 71d8db3bb..42343a6c2 100644 --- a/backend/crates/gateway-store/src/postgres/observability/admin_adapter.rs +++ b/backend/crates/gateway-store/src/postgres/observability/admin_adapter.rs @@ -123,6 +123,10 @@ impl PgObservabilityRepository { None }; Ok(admin_observability::DashboardRuntimeSlots { + inherited_versions: normal_accounts + .iter() + .filter_map(|account| account.quota_version.clone()) + .collect(), inherited_accounts, overridden_slots, used_slots, diff --git a/backend/crates/gateway-store/src/postgres/provider_accounts/quota.rs b/backend/crates/gateway-store/src/postgres/provider_accounts/quota.rs index 16f88d5da..b0ddbf9ba 100644 --- a/backend/crates/gateway-store/src/postgres/provider_accounts/quota.rs +++ b/backend/crates/gateway-store/src/postgres/provider_accounts/quota.rs @@ -13,28 +13,10 @@ pub(super) async fn versions( .collect::>(); let rows = sqlx::query("select id, credential_revision, quota_observed_at from provider_accounts where id = any($1) and quota_observed_at is not null") .bind(ids).fetch_all(&repository.pool).await.map_err(database_error)?; - rows.into_iter() - .map(|row| { - Ok(QuotaObservationVersion { - account_id: CoreProviderAccountId::new( - row.try_get::("id").map_err(database_error)?, - ) - .map_err(|_| CoreStoreError::new(CoreStoreErrorKind::InvalidData))?, - credential_revision: CoreCredentialRevision::new( - u64::try_from( - row.try_get::("credential_revision") - .map_err(database_error)?, - ) - .map_err(|_| CoreStoreError::new(CoreStoreErrorKind::InvalidData))?, - ) - .map_err(|_| CoreStoreError::new(CoreStoreErrorKind::InvalidData))?, - observed_at: row - .try_get::, _>("quota_observed_at") - .map_err(database_error)? - .into(), - }) - }) - .collect() + rows.iter() + .filter_map(|row| super::quota_version_from_row(row).transpose()) + .collect::>>() + .map_err(crate::core_store_error) } pub(super) async fn write( diff --git a/backend/crates/gateway-store/src/postgres/provider_accounts/rows.rs b/backend/crates/gateway-store/src/postgres/provider_accounts/rows.rs index 51a86b9ee..f36a1cc88 100644 --- a/backend/crates/gateway-store/src/postgres/provider_accounts/rows.rs +++ b/backend/crates/gateway-store/src/postgres/provider_accounts/rows.rs @@ -435,6 +435,21 @@ pub(crate) fn require_core_update(updated: bool) -> Result<(), CoreStoreError> { } } +pub(crate) fn quota_version_from_row( + row: &sqlx::postgres::PgRow, +) -> StoreResult> { + let Some(observed_at) = get::>>(row, "quota_observed_at")? else { + return Ok(None); + }; + Ok(Some(gateway_core::account::QuotaObservationVersion { + account_id: CoreProviderAccountId::new(get::(row, "id")?) + .map_err(|_| invalid("invalid account ID"))?, + credential_revision: CoreCredentialRevision::new(to_u64(get(row, "credential_revision")?)?) + .map_err(|_| invalid("invalid credential revision"))?, + observed_at: observed_at.into(), + })) +} + pub(crate) fn account_summary_from_row( row: sqlx::postgres::PgRow, ) -> StoreResult { diff --git a/backend/crates/gateway-store/src/postgres/provider_accounts/runtime.rs b/backend/crates/gateway-store/src/postgres/provider_accounts/runtime.rs index 26eaedaff..16fc4de03 100644 --- a/backend/crates/gateway-store/src/postgres/provider_accounts/runtime.rs +++ b/backend/crates/gateway-store/src/postgres/provider_accounts/runtime.rs @@ -64,6 +64,7 @@ pub(crate) async fn load_cooldown( pub(crate) struct ProviderAccountStatus { pub id: String, pub concurrency_limit: Option, + pub quota_version: Option, pub facts: AccountStatusFacts, } @@ -72,7 +73,7 @@ pub(crate) async fn load_account_statuses<'e, E: sqlx::Executor<'e, Database = s ) -> crate::StoreResult> { use super::rows::{account_status_facts_from_row, get}; let rows = sqlx::query( - "select id, enabled, concurrency_limit, credential_state, access_token_expires_at, + "select id, enabled, concurrency_limit, credential_revision, quota_observed_at, credential_state, access_token_expires_at, quota_access_state, quota_evidence, quota_access_observed_at, quota_reset_at, last_error_reason, last_error_message from provider_accounts", @@ -91,6 +92,11 @@ pub(crate) async fn load_account_statuses<'e, E: sqlx::Executor<'e, Database = s }) .transpose()?; Ok(ProviderAccountStatus { + quota_version: if concurrency_limit.is_none() { + super::quota_version_from_row(row)? + } else { + None + }, id: get(row, "id")?, concurrency_limit, facts: account_status_facts_from_row(row)?, diff --git a/backend/crates/providers/openai/src/credential/selector.rs b/backend/crates/providers/openai/src/credential/selector.rs index 0020351c6..4763a7cf7 100644 --- a/backend/crates/providers/openai/src/credential/selector.rs +++ b/backend/crates/providers/openai/src/credential/selector.rs @@ -332,31 +332,36 @@ impl CodexCredentialSelector { self } - fn quota_signals( + fn scheduling_signals( &self, account: &ProviderAccount, model: CodexSelectionModel<'_>, strategy: RotationStrategy, - ) -> Option { + ) -> ( + Option, + Option, + ) { if account.authentication_kind() == CODEX_AUTHENTICATION_KIND_OAUTH { - return self.quota.scheduling_signals(account); + return (self.quota.scheduling_signals(account), None); } - if strategy != RotationStrategy::Smart { - return None; - } - let signal = - self.upstream_info - .as_ref()? - .signal(account, model.model()?, SystemTime::now())?; - let rank = if signal.exhausted { - Some(0) - } else { - signal - .remaining_percent - .map(|percent| (percent.clamp(0.0, 100.0) * 100.0).round() as u64) + let Some(signal) = self + .upstream_info + .as_ref() + .and_then(|source| source.signal(account, model.model(), SystemTime::now())) + else { + return (None, None); }; - // 选号只读取 Smart 额度分，额度恢复由互联同步事务独立提交 - Some(AccountQuotaSignals::new(None, rank)) + let quota = (strategy == RotationStrategy::Smart).then(|| { + let rank = if signal.exhausted { + Some(0) + } else { + signal + .remaining_percent + .map(|percent| (percent.clamp(0.0, 100.0) * 100.0).round() as u64) + }; + AccountQuotaSignals::new(None, rank) + }); + (quota, signal.concurrency_limit) } pub(crate) async fn select_with_cyber_policy( @@ -777,6 +782,11 @@ impl CodexCredentialSelector { let health = self .account_feedback .scheduling_signals(&self.provider_kind, account.id()); + let (quota, inherited_concurrency) = self.scheduling_signals( + &account, + model, + request.attempt.account_selection_policy().strategy(), + ); let signals = scheduling .signals() .get(account.id()) @@ -790,14 +800,14 @@ impl CodexCredentialSelector { failure_rate_basis_points: None, first_output_latency_ms: None, }) - .with_provider_quota(self.quota_signals( - &account, - model, - request.attempt.account_selection_policy().strategy(), - )) + .with_provider_quota(quota) .with_rate_limit(rate_limits.get(account.id()).copied().flatten()) .with_runtime_health(health.0, health.1); - AccountCandidate { account, signals } + AccountCandidate { + account, + signals, + inherited_concurrency, + } }) .collect::>(); let mut affinity = binding @@ -1060,7 +1070,7 @@ impl CodexCredentialSelector { self.provider_kind.clone(), account.id().clone(), account.revision(), - context.concurrency_limit(&account), + context.concurrency_limit(selected), request.attempt.deadline(), ) .with_metadata(request.attempt.lease_metadata()) diff --git a/backend/crates/providers/openai/tests/admin/peer_info.rs b/backend/crates/providers/openai/tests/admin/peer_info.rs index 7a0c08a09..d95bc42f5 100644 --- a/backend/crates/providers/openai/tests/admin/peer_info.rs +++ b/backend/crates/providers/openai/tests/admin/peer_info.rs @@ -11,7 +11,7 @@ use provider_openai::credential::ResponsesTransport; fn peer_document() -> Value { let now = Utc::now(); json!({ - "object":"cpr.peer", "schema_version":2, + "object":"cpr.peer", "schema_version":3, "observed_at":now.to_rfc3339(), "expires_at":(now + chrono::Duration::minutes(5)).to_rfc3339(), "capabilities":["key_usage", "pool_summary"], "usage":{"status":"available","observed_at":now.to_rfc3339(),"data":{"unit":"USD","allowed":true,"windows":[ @@ -480,7 +480,7 @@ async fn model_only_snapshots_publish_signals_without_budget_or_account_disclosu let server = MockServer::start().await; let now = Utc::now(); let document = json!({ - "object":"cpr.peer", "schema_version":2, + "object":"cpr.peer", "schema_version":3, "observed_at":now.to_rfc3339(), "expires_at":(now + chrono::Duration::minutes(5)).to_rfc3339(), "capabilities":["model_availability"], "models":[{"model":"gpt-5.4", "coverage":coverage, "state":state, @@ -502,7 +502,7 @@ async fn model_only_snapshots_publish_signals_without_budget_or_account_disclosu reader .signal( &accounts.account(id.as_str()).unwrap(), - "gpt-5.4", + Some("gpt-5.4"), SystemTime::now() ) .map(|signal| signal.exhausted), @@ -778,7 +778,7 @@ async fn invalid_or_redirected_responses_never_replace_the_last_valid_snapshot() cycle(&worker).await; let observed_at = quota(&bundle, &id).await.observed_at; let mut wrong_schema = peer_document(); - wrong_schema["schema_version"] = json!(99); + wrong_schema["schema_version"] = json!(2); let mut wrong_object = peer_document(); wrong_object["object"] = json!("other.peer"); let mut negative = peer_document(); @@ -899,6 +899,127 @@ async fn peer_sync_excludes_budget_exhausted_accounts_without_pool_disclosure() server.verify().await; } +#[tokio::test] +async fn peer_key_limits_set_default_capacity_and_lease_without_model_or_pool_disclosure() { + use gateway_core::account::{AccountConcurrencyLimit, AccountWeight}; + for (limits, observed, stale, local_limit, expected) in [ + ( + Some(json!({"max_concurrency":5})), + (Some(5), None), + false, + None, + 5, + ), + ( + Some(json!({"max_concurrency":5,"requests_per_minute":60})), + (Some(5), Some(60)), + false, + Some(1), + 1, + ), + ( + Some(json!({"max_concurrency":0})), + (Some(0), None), + false, + None, + 2, + ), + ( + Some(json!({"requests_per_minute":60})), + (None, Some(60)), + false, + None, + 2, + ), + (None, (None, None), false, None, 2), + ( + Some(json!({"max_concurrency":5})), + (Some(5), None), + true, + None, + 2, + ), + ] { + let server = MockServer::start().await; + let now = Utc::now(); + let observed_at = if stale { + now - chrono::Duration::minutes(6) + } else { + now + }; + let mut document = json!({ + "object":"cpr.peer", "schema_version":3, + "capabilities": ["key_limits"], + "observed_at":observed_at.to_rfc3339(), "expires_at":(now + chrono::Duration::minutes(2)).to_rfc3339() + }); + if let Some(limits) = limits { + document["key_limits"] = limits; + } + Mock::given(method("GET")) + .and(path("/custom/v1/cpr/link")) + .respond_with(ResponseTemplate::new(200).set_body_json(document)) + .expect(1) + .mount(&server) + .await; + Mock::given(method("POST")) + .and(path("/custom/v1/responses")) + .respond_with( + ResponseTemplate::new(200) + .insert_header("content-type", "text/event-stream") + .set_body_string(COMPLETED_SESSION_SSE), + ) + .expect(1) + .mount(&server) + .await; + let (mut bundle, accounts, _, id, _config) = fixture(&server).await; + accounts.set_scheduling( + id.as_str(), + local_limit.and_then(AccountConcurrencyLimit::new), + AccountWeight::DEFAULT, + ); + cycle(&worker(&mut bundle)).await; + let info = quota(&bundle, &id).await.upstream.unwrap(); + assert_eq!((info.max_concurrency, info.requests_per_minute), observed); + let version = accounts + .get_quota_versions(std::slice::from_ref(&id)) + .await + .unwrap() + .pop() + .unwrap(); + let inherited = bundle + .admin_provider() + .observed_concurrency_limit(&version, SystemTime::now()); + assert_eq!( + inherited.map(|limit| u64::from(limit.get())), + observed.0.filter(|limit| !stale && *limit > 0) + ); + let mut stream = execute_peer_request(&bundle, "round_robin").await.unwrap(); + let capacity = stream + .metadata() + .selection_observation() + .unwrap() + .capacity() + .unwrap(); + assert_eq!(capacity.total_slots(), u64::from(expected)); + assert_eq!( + bundle + .leases + .requests + .lock() + .unwrap() + .last() + .unwrap() + .max_concurrent() + .get(), + expected + ); + while let Some(event) = stream.next().await { + event.unwrap(); + } + server.verify().await; + } +} + #[tokio::test] async fn lists_reuse_versioned_summaries_and_details_keep_node_windows() { let server = MockServer::start().await; @@ -1238,6 +1359,7 @@ async fn execute_peer_request( } struct PeerFixture { + leases: Arc, provider: provider_openai::ProviderBundle, peer: gateway_admin::PeerUpstreamBundle, cooldowns: Arc, @@ -1263,11 +1385,12 @@ async fn peer_fixture( ) -> PeerFixture { let reader = Arc::new(gateway_core::account::UpstreamInfoReader::default()); let cooldowns = Arc::new(MemoryCooldownPort::new()); + let leases = Arc::new(TestLeaseCoordinator::default()); let provider = provider_openai::initialize( config, ProviderStorePorts::new( accounts.clone(), - Arc::new(TestLeaseCoordinator::default()), + leases.clone(), Arc::new(MemorySessionAffinity::default()), Arc::new(MemorySessionExclusions::default()), Arc::new(TestCatalogCache::default()), @@ -1295,6 +1418,7 @@ async fn peer_fixture( .unwrap(); reader.bind(&peer.scheduling_source()).unwrap(); PeerFixture { + leases, provider, peer, cooldowns, diff --git a/backend/crates/providers/openai/tests/credential/quota/scheduling.rs b/backend/crates/providers/openai/tests/credential/quota/scheduling.rs index 194f28f7c..1025c9b74 100644 --- a/backend/crates/providers/openai/tests/credential/quota/scheduling.rs +++ b/backend/crates/providers/openai/tests/credential/quota/scheduling.rs @@ -80,6 +80,7 @@ async fn old_weekly_quota_should_outweigh_one_in_flight_title_request() { let candidates = accounts .into_iter() .map(|account| AccountCandidate { + inherited_concurrency: None, signals: AccountRuntimeSignals { in_flight: u32::from(account.id().as_str() == "acct_74"), last_started_at: None, diff --git a/backend/crates/providers/xai/src/credential/selector.rs b/backend/crates/providers/xai/src/credential/selector.rs index 1a043709d..a9b116670 100644 --- a/backend/crates/providers/xai/src/credential/selector.rs +++ b/backend/crates/providers/xai/src/credential/selector.rs @@ -241,7 +241,11 @@ impl GrokAccountSessionSelector { .ok_or(GrokSessionSelectorError::Unavailable(None))? .with_provider_quota(self.quota.scheduling_signals(&account)) .with_runtime_health(health.0, health.1); - Ok(AccountCandidate { account, signals }) + Ok(AccountCandidate { + account, + signals, + inherited_concurrency: None, + }) }) .collect::, GrokSessionSelectorError>>()?; if let Some(required) = request.required_account() { @@ -319,11 +323,7 @@ impl GrokAccountSessionSelector { self.provider_kind.clone(), selected_id.clone(), selected_revision, - selected.account.effective_concurrency( - request - .account_selection_policy() - .max_concurrent_per_account(), - ), + context.concurrency_limit(selected), request.deadline(), ) .with_metadata(request.lease_metadata.clone()) diff --git a/deploy/README.md b/deploy/README.md index 4b10eea44..f4c26e7a7 100644 --- a/deploy/README.md +++ b/deploy/README.md @@ -399,17 +399,17 @@ PostgreSQL/Redis 启动密码。日常校验使用 `config --quiet` ### CPR 互联 -在上游的「API 密钥」中为专用 Key 选择账号分组并开启「CPR 互联」，按需公开账号池、额度和套餐信息。 +在上游的「API 密钥」中为专用密钥选择账号分组并开启「CPR 互联」，即可共享密钥预算、并发上限和 RPM，并按需公开账号池、账号额度和套餐信息。 共享 Key 的实际推理只使用这些分组中的直连 OAuth 账号；分组内的 API Key 转供账号不会参与 在下游导入或编辑 OpenAI API Key 账号，填写上游 API 地址和该 Key，后台自动发现 CPR 互联能力。 地址包含自定义前缀时保留完整前缀，后台在该地址下请求 `cpr/link`。 -列表与详情显示已保存的观测，手动刷新等待上游查询完成；旧观测和查询失败分别提示，不要求上游实时可用 +列表与详情显示已保存的观测。展开账号可查看上游密钥限制，普通账号并发留空时跟随上游有效的正数上限。手动刷新等待上游查询完成，旧观测和查询失败分别提示 第三方网关通过插件实现相同协议也可被识别，不按软件名称判定兼容。 上游明确返回 Key 预算耗尽时，下游暂停该账号；后续同步确认允许使用后自动恢复，普通限流冷却独立保留。 金额只供展示，缺少访问状态时不根据余额推断恢复。账号池额度观测作为 Smart 选择的参考，不能保证下一次请求成功 -互联两端需支持 CPR Link v2；协议、第三方接入与公开边界见 [CPR 互联 API](../docs/api.md#cpr-互联只读协议) +互联两端均须使用 CPR Link v3；协议、第三方接入与公开边界见 [CPR 互联 API](../docs/api.md#cpr-互联只读协议) ### 批量维护代理 diff --git a/docs/api.md b/docs/api.md index 33003f8e8..e8d6662eb 100644 --- a/docs/api.md +++ b/docs/api.md @@ -755,29 +755,31 @@ OpenAI 选号阶段确认本次可选账号全部额度耗尽时，HTTP 返回 ` | 顶层必填字段 | 合同 | | --- | --- | | `object` | 固定为 `"cpr.peer"` | -| `schema_version` | 当前为 `2`，不支持的版本按不兼容处理 | -| `capabilities` | 当前 Key 获准公开的能力名称列表，最多 32 项；未知能力可以忽略 | +| `schema_version` | 固定为 `3`，接入端只接受此版本 | +| `capabilities` | 当前密钥 获准公开的能力名称列表，最多 32 项；未知能力可以忽略 | | `observed_at` | 本次快照组装时间，RFC3339 UTC 时间 | | `expires_at` | 快照到期时间，必须晚于 `observed_at`，且不超过其后 30 分钟 | -`usage`、`pool`、`quota`、`models` 按能力独立提供，未提供的数据块直接省略。 +`usage`、`key_limits`、`pool`、`quota`、`models` 按能力独立提供，未提供的数据块直接省略。 接入端只消费已声明能力的数据，缺失值保持未知，不解释为零余额、空池或额度恢复。 未知字段可以忽略；没有独立节点 ID，连接由本地账号及其凭据、出口版本识别 | 能力 | 对应数据 | | --- | --- | | `key_usage` | `usage`：当前认证 Key 的预算访问结论与可选金额窗口 | +| `key_limits` | `key_limits`：当前认证密钥公开的并发上限和 RPM | | `pool_summary` | `pool`：账号数量、互斥状态计数及可选套餐分布 | | `quota_windows` | `quota.accounts`：匿名账号及各自额度窗口 | | `model_availability` | `models`：按当前 Key 公开的模型额度汇总，可独立于账号明细提供 | -只公开预算访问结论的完整响应： +提供请求限制和预算访问结论的响应示例： ```json { "object": "cpr.peer", - "schema_version": 2, - "capabilities": ["key_usage"], + "schema_version": 3, + "key_limits": { "max_concurrency": 5, "requests_per_minute": 60 }, + "capabilities": ["key_usage", "key_limits"], "observed_at": "2026-11-01T12:00:00Z", "expires_at": "2026-11-01T12:05:00Z", "usage": { @@ -788,6 +790,11 @@ OpenAI 选号阶段确认本次可选账号全部额度耗尽时，HTTP 返回 ` } ``` +#### 密钥请求限制 + +`key_limits` 形如 `{"max_concurrency":5,"requests_per_minute":60}`。上游可省略整个对象或其中任一项。已提供的值为非负整数，零表示该项不限；缺失表示未知。 +限制适用于整个密钥，有效期随顶层快照 + #### Key 预算 `usage.status` 为 `available`、`unavailable` 或 `unknown`，未知状态按 `unknown` 处理。 @@ -883,7 +890,7 @@ CPR 默认路径为 `/v1/cpr/link`，使用同一数据面入口认证，不接 | 共享配置 | 导出内容 | | --- | --- | -| `enabled` | `key_usage`，包含明确访问结论及本地日、周美元账本；金额与周期口径见 [API Key 额度查询](#api-key-额度查询) | +| `enabled` | `key_limits` 与 `key_usage`，包含密钥请求限制、明确访问结论及本地日、周美元账本；金额与周期口径见 [API Key 额度查询](#api-key-额度查询) | | `sharePool` | `pool_summary` | | `sharePlans` | 已公开账号池中的 `pool.plans` | | `shareQuota` | `quota_windows` 与 `model_availability` | @@ -1085,8 +1092,7 @@ config 返回 `{ name, plaintextKey }`，仅读取服务端会话绑定的当前 账号列表的 `search` 仅按字面量包含匹配名称和邮箱，不区分大小写；`%`、`_` 和 `\` 按普通字符处理 账号视图的 `capacity` 返回查询时的普通并发容量：`usedSlots` 不含独立审批池和排队请求， -读取运行态失败时为 `null`；`totalSlots` 是应用账号独立配置或全局默认值后的上限，`null` 表示不限。 -该上限不代表上游实际允许的并发数 +读取运行态失败时为 `null`。`totalSlots` 是按[并发与排队](#并发与排队)取值的本机上限，`null` 表示不限。上游独立执行其密钥限制 并发明细返回 `accountId`、`entries`、`total`、`page`、`pageSize` 和 `observedAt`，只统计普通并发池。每条包含 `leaseId`、`requestId`、`clientKeyId`、`clientKeyName`、`model`（入口请求模型）、`clientIp`、`startedAt`、`startedAtDisplay` 和 `occupiedMs`。占用时长从取得账号租约开始，不包含排队；缺少入口事实时相应字段为 `null`，租约仍计入总数 @@ -1159,6 +1165,7 @@ OpenAI 主动额度刷新和正常响应携带的明确套餐会同步到账号 | `stale` | 快照是否已过期；过期值仅供查看 | | `connectionRevision` | 当前连接身份的不可解释标识，用于隔离地址、密钥或出口变化前的展示结果 | | `lastError` | 当前实例最近同步失败分类：`unauthorized`、`unsupported`、`rate_limited`、`invalid_response`、`unavailable`；已保存的不支持结论及已确认互联的失权记录重启后仍返回对应分类，同步成功时为 `null` | +| `maxConcurrency`、`requestsPerMinute` | 上游密钥的并发上限和 RPM，零表示不限，未观测到的项为 `null` | | `pool` | 上游授权公开的账号数量、状态数量及套餐数量，未公开时为 `null` | 互联额度复用 `quota.windows`。上游 Key 账本按实际提供的周期展示，不补造缺失窗口；不限额也保留已用金额。 @@ -2042,7 +2049,8 @@ UA、平台参数及自动 Git 上下文版本由 [OpenAI 上游客户端身份] ### 并发与排队 `maxConcurrentPerAccount` 是默认账号并发上限，初始值为 3，取值 0～4294967295；`0` 表示不限制。 -账号的 `concurrencyLimit: null` 继承该默认值，单独设置的正数上限仍优先生效。 +账号单独设置的正数 `concurrencyLimit` 优先。留空时，互联账号使用有效快照中密钥的正数并发上限，最大取 4294967295。 +无有效的上游正数上限时使用本机全局默认值。上游 RPM 供展示，由上游执行。 无限并发仍统计在途请求，并遵守账号可用性与 Client Key 限制 `maxWaitingPerKey` 与 `maxWaitingPerAccount` 是全局统一的普通排队容量，取值 0～1,000，默认 10，`0` 表示关闭； diff --git a/docs/architecture.md b/docs/architecture.md index f91700d86..44a0d3405 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -1063,6 +1063,7 @@ OAuth 额度解析属于 Provider，CPR 发现、解析与投影属于 Admin； Admin 按连接身份及额度观察时间缓存管理摘要与调度投影，解析和投影在锁外完成。 Core 定义中立调度信号与只读端口，组合根绑定观测来源；Provider 选号不持有控制面服务，弱引用避免互相保活。 +互联密钥的并发上限随有效观测参与普通账号选号和租约准入，账号手动配置优先。分组与概览按额度观测版本核对容量，账号配置与观测各自保存。 账号和凭据提交后的共用通知更新原生 Provider 与互联派生状态，插件账号写入复用同一流程； 互联按当前连接身份失效观测，普通信息编辑保留已有摘要与刷新退避，停用或删除时释放运行态。 账号列表、概览与代理关联账号核对观测版本后复用预算与池摘要，详情读取逐节点窗口；手动刷新通过共用账号额度接口等待查询完成并返回结果。 diff --git a/frontend/src/api/modules/accounts/types.ts b/frontend/src/api/modules/accounts/types.ts index 486e83d1d..429b6cd35 100644 --- a/frontend/src/api/modules/accounts/types.ts +++ b/frontend/src/api/modules/accounts/types.ts @@ -189,6 +189,8 @@ export interface AccountUpstream { stale: boolean connectionRevision: string lastError: 'unauthorized' | 'unsupported' | 'rate_limited' | 'invalid_response' | 'unavailable' | null + maxConcurrency: number | null + requestsPerMinute: number | null pool: { accountCount: number states: { ready: number, cooldown: number, exhausted: number, disabled: number, error: number, unknown: number } diff --git a/frontend/src/i18n/locales/en/accounts.json b/frontend/src/i18n/locales/en/accounts.json index 4da5e6e68..f84fbc0cd 100644 --- a/frontend/src/i18n/locales/en/accounts.json +++ b/frontend/src/i18n/locales/en/accounts.json @@ -631,7 +631,9 @@ "invalidResponse": "Invalid upstream data", "unavailable": "Upstream unavailable", "pool": "Shared pool", - "ready": "available", + "readyAccounts": "Ready / total accounts", + "keyConcurrency": "Upstream key concurrency limit", + "keyRpm": "Upstream key requests per minute", "otherPlans": "Other plans", "morePlans": "View {count} more plans", "nodeQuota": "Node quota", diff --git a/frontend/src/i18n/locales/en/keys.json b/frontend/src/i18n/locales/en/keys.json index 8c88e3b5b..d144107e6 100644 --- a/frontend/src/i18n/locales/en/keys.json +++ b/frontend/src/i18n/locales/en/keys.json @@ -79,7 +79,7 @@ "invalidKey": "Use visible HTTP header characters, without spaces or line breaks", "requestsPerMinute": "Requests per minute", "peerSharing": "CPR Link", - "peerSharingHelp": "Allow other nodes to read this key’s budget; requests use only direct accounts in the selected groups", + "peerSharingHelp": "Share this key’s budget, concurrency limit and RPM with other nodes. Model requests use only direct accounts in the selected groups", "sharePool": "Share account pool summary", "shareQuota": "Share account quotas", "sharePlans": "Share plan breakdown", diff --git a/frontend/src/i18n/locales/zh-Hans/accounts.json b/frontend/src/i18n/locales/zh-Hans/accounts.json index 3e1b54019..ac0b2156e 100644 --- a/frontend/src/i18n/locales/zh-Hans/accounts.json +++ b/frontend/src/i18n/locales/zh-Hans/accounts.json @@ -631,7 +631,9 @@ "invalidResponse": "上游数据异常", "unavailable": "上游暂不可用", "pool": "共享池", - "ready": "可用", + "readyAccounts": "可用账号 / 总账号", + "keyConcurrency": "上游密钥并发上限", + "keyRpm": "上游密钥每分钟请求数", "otherPlans": "其他套餐", "morePlans": "查看其余 {count} 种套餐", "nodeQuota": "节点额度", diff --git a/frontend/src/i18n/locales/zh-Hans/keys.json b/frontend/src/i18n/locales/zh-Hans/keys.json index 7a467fc6b..2c6b40cd0 100644 --- a/frontend/src/i18n/locales/zh-Hans/keys.json +++ b/frontend/src/i18n/locales/zh-Hans/keys.json @@ -79,7 +79,7 @@ "invalidKey": "自定义 API密钥只能包含 HTTP 可传输的可见字符，不能包含空格或换行", "requestsPerMinute": "每分钟请求数", "peerSharing": "CPR 互联", - "peerSharingHelp": "允许其他节点查询此 Key 的额度，模型请求仅使用所选分组中的直连账号", + "peerSharingHelp": "允许其他节点查询此密钥的预算、并发上限和 RPM，模型请求仅使用所选分组中的直连账号", "sharePool": "共享账号池概览", "shareQuota": "共享账号额度", "sharePlans": "共享套餐构成", diff --git a/frontend/src/i18n/locales/zh-Hant/accounts.json b/frontend/src/i18n/locales/zh-Hant/accounts.json index 33b0b3658..bd680cce9 100644 --- a/frontend/src/i18n/locales/zh-Hant/accounts.json +++ b/frontend/src/i18n/locales/zh-Hant/accounts.json @@ -631,7 +631,9 @@ "invalidResponse": "上游資料異常", "unavailable": "上游暫不可用", "pool": "共享池", - "ready": "可用", + "readyAccounts": "可用帳號 / 總帳號", + "keyConcurrency": "上游密鑰並行上限", + "keyRpm": "上游密鑰每分鐘請求數", "otherPlans": "其他套餐", "morePlans": "查看其餘 {count} 種套餐", "nodeQuota": "節點額度", diff --git a/frontend/src/i18n/locales/zh-Hant/keys.json b/frontend/src/i18n/locales/zh-Hant/keys.json index 16d92b569..50c062fa6 100644 --- a/frontend/src/i18n/locales/zh-Hant/keys.json +++ b/frontend/src/i18n/locales/zh-Hant/keys.json @@ -79,7 +79,7 @@ "invalidKey": "自定義 API密鑰只能包含 HTTP 可傳輸的可見字符，不能包含空格或換行", "requestsPerMinute": "每分鐘請求數", "peerSharing": "CPR 互聯", - "peerSharingHelp": "允許其他節點查詢此 Key 的額度，模型請求僅使用所選分組中的直連帳號", + "peerSharingHelp": "允許其他節點查詢此密鑰的預算、並行上限和 RPM，模型請求僅使用所選分組中的直連帳號", "sharePool": "共享帳號池概覽", "shareQuota": "共享帳號額度", "sharePlans": "共享方案組成", diff --git a/frontend/src/views/accounts/components/account-quota-panel/UpstreamPool.vue b/frontend/src/views/accounts/components/account-quota-panel/UpstreamPool.vue index 341417177..070506ef3 100644 --- a/frontend/src/views/accounts/components/account-quota-panel/UpstreamPool.vue +++ b/frontend/src/views/accounts/components/account-quota-panel/UpstreamPool.vue @@ -1,17 +1,19 @@