From ea4e3059db5d2e744518fec5dd8df10b5905b867 Mon Sep 17 00:00:00 2001 From: Zyy Date: Sun, 27 Sep 2026 22:36:51 +0800 Subject: [PATCH] feat(accounts): show current concurrency capacity --- .../gateway-admin/src/model/accounts.rs | 10 ++ .../src/model/provider_credentials.rs | 1 + .../gateway-admin/src/use_case/accounts.rs | 19 ++++ .../gateway-admin/tests/use_case/accounts.rs | 8 ++ .../tests/use_case/accounts/capacity.rs | 107 ++++++++++++++++++ .../src/admin/accounts/presenter.rs | 5 + .../gateway-api/src/admin/accounts/wire.rs | 8 ++ .../crates/gateway-api/tests/admin/errors.rs | 36 ++++++ .../provider_accounts/admin_adapter.rs | 33 ++++++ .../provider_accounts/admin_queries.rs | 7 +- .../provider_accounts/admin_adapter.rs | 69 +++++++++++ docs/api.md | 4 + frontend/src/api/modules/accounts.ts | 4 + .../components/AccountCapacityIndicator.vue | 31 +++++ .../AccountQuotaSummaryCell/Entry.vue | 5 +- .../AccountQuotaSummaryCell/WindowGroup.vue | 6 +- .../AccountQuotaSummaryCell/index.vue | 16 ++- .../components/AccountUsageWindow/index.vue | 1 - 18 files changed, 359 insertions(+), 11 deletions(-) create mode 100644 backend/crates/gateway-admin/tests/use_case/accounts/capacity.rs create mode 100644 frontend/src/views/accounts/components/AccountCapacityIndicator.vue diff --git a/backend/crates/gateway-admin/src/model/accounts.rs b/backend/crates/gateway-admin/src/model/accounts.rs index 61579e8d9..86ea9810d 100644 --- a/backend/crates/gateway-admin/src/model/accounts.rs +++ b/backend/crates/gateway-admin/src/model/accounts.rs @@ -223,6 +223,16 @@ pub struct AccountPage { pub struct AccountPageItem { pub account: AccountRecord, pub projection: AccountStatusProjection, + pub capacity: AccountCapacity, +} + +/// 网关配置的账号并发上限与查询时的占用，不包含排队请求或上游隐藏限制。 +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct AccountCapacity { + /// 实时租约读取失败时为 `None`，不能当作空闲。 + pub used_slots: Option, + /// 应用账号覆盖或全局默认值后的上限；`None` 表示不限。 + pub total_slots: Option, } /// 统一账号目录的全局状态计数，不受当前筛选和分页影响。 diff --git a/backend/crates/gateway-admin/src/model/provider_credentials.rs b/backend/crates/gateway-admin/src/model/provider_credentials.rs index 9a6dc2d8b..0c25c44ed 100644 --- a/backend/crates/gateway-admin/src/model/provider_credentials.rs +++ b/backend/crates/gateway-admin/src/model/provider_credentials.rs @@ -1227,6 +1227,7 @@ impl fmt::Debug for ProviderExport { #[derive(Debug, Clone, PartialEq)] pub struct AccountDirectoryItem { pub account: AccountRecord, + pub capacity: super::accounts::AccountCapacity, pub capabilities: super::accounts::ProviderAccountCapabilities, /// Provider 提供的套餐展示名称；未识别到套餐时为空。 pub plan_type_display: Option, diff --git a/backend/crates/gateway-admin/src/use_case/accounts.rs b/backend/crates/gateway-admin/src/use_case/accounts.rs index 6e7b38bb4..5830222ee 100644 --- a/backend/crates/gateway-admin/src/use_case/accounts.rs +++ b/backend/crates/gateway-admin/src/use_case/accounts.rs @@ -363,6 +363,7 @@ impl DefaultAccountsService { .and_then(|(window, _)| window.local_usage.clone()) }); Ok(AccountDirectoryItem { + capacity: stored.capacity, capabilities: provider .account_capabilities(account_id, &stored.account.authentication_kind), plan_type_display: self.providers.resolve_account_plan( @@ -402,6 +403,18 @@ impl AccountsService for DefaultAccountsService { .iter() .map(|item| item.account.id.clone()) .collect::>(); + // 只读取当前页的租约占用；观测失败不能把账号误报为空闲或拖垮目录。 + let in_flight = if ids.is_empty() { + None + } else { + match self.account_runtime.account_runtime(&ids).await { + Ok(runtime) => runtime.in_flight, + Err(error) => { + tracing::warn!(error = %error, "account capacity projection is unavailable"); + None + } + } + }; let rolling_usage = self .accounts .load_account_usage(rolling_range, &ids) @@ -472,6 +485,12 @@ impl AccountsService for DefaultAccountsService { .and_then(|(window, _)| window.local_usage.clone()) }); Ok(AccountDirectoryItem { + capacity: crate::model::accounts::AccountCapacity { + used_slots: in_flight + .as_ref() + .map(|counts| counts.get(&item.account.id).copied().unwrap_or(0)), + ..item.capacity + }, capabilities, plan_type_display: providers.resolve_account_plan( item.account.provider_kind.as_str(), diff --git a/backend/crates/gateway-admin/tests/use_case/accounts.rs b/backend/crates/gateway-admin/tests/use_case/accounts.rs index 8cacec0ce..d34a19ed8 100644 --- a/backend/crates/gateway-admin/tests/use_case/accounts.rs +++ b/backend/crates/gateway-admin/tests/use_case/accounts.rs @@ -1,5 +1,7 @@ use std::sync::{Arc, Mutex}; +mod capacity; + use async_trait::async_trait; use chrono::{TimeDelta, Utc}; use futures::{StreamExt as _, future::BoxFuture}; @@ -668,6 +670,12 @@ impl FakeAccountStore { last_error_message: account.last_error_message.clone(), }; AccountPageItem { + capacity: gateway_admin::model::accounts::AccountCapacity { + used_slots: None, + total_slots: account + .concurrency_limit + .map(|limit| u64::from(limit.get())), + }, account, projection: resolve_account_status(&facts, std::time::SystemTime::now()), } diff --git a/backend/crates/gateway-admin/tests/use_case/accounts/capacity.rs b/backend/crates/gateway-admin/tests/use_case/accounts/capacity.rs new file mode 100644 index 000000000..bafb43568 --- /dev/null +++ b/backend/crates/gateway-admin/tests/use_case/accounts/capacity.rs @@ -0,0 +1,107 @@ +use std::collections::BTreeMap; + +use gateway_admin::ports::store::AccountRuntimeStore; + +use super::*; + +struct CapacityRuntime { + counts: Option>, + fail: bool, + reads: Mutex>>, +} + +#[async_trait] +impl AccountRuntimeStore for CapacityRuntime { + async fn active_rate_limits(&self) -> AdminStoreResult { + Ok(AccountRuntimeSnapshot::default()) + } + + async fn account_runtime(&self, ids: &[String]) -> AdminStoreResult { + self.reads.lock().unwrap().push(ids.to_vec()); + if self.fail { + return Err(store_unavailable()); + } + Ok(AccountRuntimeSnapshot { + in_flight: self.counts.clone(), + ..Default::default() + }) + } + + async fn active_freezes( + &self, + ) -> AdminStoreResult> { + Ok(BTreeMap::new()) + } + + async fn capacity_peaks(&self, _: &[String]) -> AdminStoreResult> { + Ok(BTreeMap::new()) + } + + async fn finish_freeze( + &self, + _: &str, + _: &gateway_admin::model::accounts::AccountFreeze, + _: Option>, + ) -> AdminStoreResult { + Ok(false) + } +} + +#[tokio::test] +async fn account_capacity_should_batch_page_ids_and_distinguish_idle_from_unavailable() { + for (counts, fail, expected) in [ + ( + Some(BTreeMap::from([("acct_test".to_owned(), 3)])), + false, + Some(3), + ), + (Some(BTreeMap::new()), false, Some(0)), + (None, false, None), + (None, true, None), + ] { + for empty in [false, true] { + let runtime = Arc::new(CapacityRuntime { + counts: counts.clone(), + fail, + reads: Mutex::default(), + }); + let mut account = account_record("openai"); + account.concurrency_limit = gateway_core::account::AccountConcurrencyLimit::new(5); + let store = FakeAccountStore::with_account(account, events()); + if empty { + store.set_accounts(Vec::new()); + } + let services = super::super::AdminHarness::new() + .accounts(store) + .account_runtime(runtime.clone()) + .settings(Arc::new(StaticSettingsStore)) + .provider(FakeProviderAdmin::new("openai", events())) + .build() + .await; + let page = services + .accounts() + .list(AccountListQuery { + page: 1, + page_size: gateway_admin::model::PageSize::new(20).unwrap(), + provider_kind: None, + group_filter: None, + search: None, + status: None, + sort: None, + }) + .await + .unwrap(); + if empty { + assert!(page.items.is_empty()); + assert!(runtime.reads.lock().unwrap().is_empty()); + } else { + assert_eq!( + *runtime.reads.lock().unwrap(), + [vec!["acct_test".to_owned()]] + ); + assert_eq!(page.items[0].capacity.used_slots, expected); + assert_eq!(page.items[0].capacity.total_slots, Some(5)); + } + } + } +} diff --git a/backend/crates/gateway-api/src/admin/accounts/presenter.rs b/backend/crates/gateway-api/src/admin/accounts/presenter.rs index c5171194f..9fd269ac0 100644 --- a/backend/crates/gateway-api/src/admin/accounts/presenter.rs +++ b/backend/crates/gateway-api/src/admin/accounts/presenter.rs @@ -60,6 +60,7 @@ pub(super) fn account_models_data(result: ProviderModels) -> AccountModelsData { pub(super) fn account_view(item: AccountDirectoryItem, now: DateTime) -> AccountView { let AccountDirectoryItem { account, + capacity, capabilities, plan_type_display, projection, @@ -109,6 +110,10 @@ pub(super) fn account_view(item: AccountDirectoryItem, now: DateTime) -> Ac error_message: projection.error_message, enabled: account.enabled, concurrency_limit: account.concurrency_limit.map(|limit| limit.get()), + capacity: AccountCapacityView { + used_slots: capacity.used_slots, + total_slots: capacity.total_slots, + }, weight: account.weight.get(), model_access: account.model_access, outbound_proxy_endpoint: account diff --git a/backend/crates/gateway-api/src/admin/accounts/wire.rs b/backend/crates/gateway-api/src/admin/accounts/wire.rs index 10a632e54..21f590dde 100644 --- a/backend/crates/gateway-api/src/admin/accounts/wire.rs +++ b/backend/crates/gateway-api/src/admin/accounts/wire.rs @@ -290,6 +290,7 @@ pub struct AccountView { pub error_message: Option, pub enabled: bool, pub concurrency_limit: Option, + pub capacity: AccountCapacityView, pub weight: u16, pub model_access: gateway_core::account::AccountModelAccess, pub access_token_expires_at: Option, @@ -304,6 +305,13 @@ pub struct AccountView { pub usage: AccountUsageView, } +#[derive(Debug, Clone, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct AccountCapacityView { + pub used_slots: Option, + pub total_slots: Option, +} + #[derive(Debug, Clone, Serialize)] #[serde(rename_all = "camelCase")] pub struct AccountCapabilitiesView { diff --git a/backend/crates/gateway-api/tests/admin/errors.rs b/backend/crates/gateway-api/tests/admin/errors.rs index 7bd913474..9582ac2a8 100644 --- a/backend/crates/gateway-api/tests/admin/errors.rs +++ b/backend/crates/gateway-api/tests/admin/errors.rs @@ -336,6 +336,38 @@ mod provider { use super::*; + #[tokio::test] + async fn unsupported_quota_should_preserve_account_capacity_wire() { + let fixture = AdminTestFixture::new().await; + fixture.auth.insert_session("valid-session"); + for (used_slots, total_slots) in [(Some(3), Some(5)), (Some(0), None), (None, Some(10))] { + let mut stored = account("openai"); + stored.capacity = gateway_admin::model::accounts::AccountCapacity { + used_slots, + total_slots, + }; + *fixture.account.lock().unwrap() = Some(stored); + let mut request = request( + Method::GET, + "/api/admin/accounts/quota?accountId=acct_error_test", + Body::empty(), + ); + request + .headers_mut() + .insert(header::COOKIE, SESSION_COOKIE.parse().unwrap()); + let response = app(fixture.state()).oneshot(request).await.unwrap(); + let (status, _, body) = response_json(response).await; + assert_eq!(status, StatusCode::OK); + assert_eq!( + body["data"]["account"]["capacity"], + json!({ + "usedSlots": used_slots, + "totalSlots": total_slots, + }) + ); + } + } + #[tokio::test] async fn provider_public_errors_survive_import_credential_and_quota_refresh_handlers() { let fixture = AdminTestFixture::new().await; @@ -472,6 +504,10 @@ mod provider { last_error_message: None, }; AccountPageItem { + capacity: gateway_admin::model::accounts::AccountCapacity { + used_slots: None, + total_slots: None, + }, account: AccountRecord { notes: None, model_access: Default::default(), diff --git a/backend/crates/gateway-store/src/postgres/provider_accounts/admin_adapter.rs b/backend/crates/gateway-store/src/postgres/provider_accounts/admin_adapter.rs index 8537dde38..e0fa013f1 100644 --- a/backend/crates/gateway-store/src/postgres/provider_accounts/admin_adapter.rs +++ b/backend/crates/gateway-store/src/postgres/provider_accounts/admin_adapter.rs @@ -8,6 +8,20 @@ use gateway_core::provider_ports::ProviderCooldownPort; use super::*; use crate::postgres::ObservabilityQueryBudget; +fn account_capacity( + account: &AccountRecord, + default_concurrency: u64, + in_flight: Option<&BTreeMap>, +) -> gateway_admin::model::accounts::AccountCapacity { + gateway_admin::model::accounts::AccountCapacity { + used_slots: in_flight.map(|counts| counts.get(&account.id).copied().unwrap_or(0)), + total_slots: account.concurrency_limit.map_or_else( + || (default_concurrency > 0).then_some(default_concurrency), + |limit| Some(u64::from(limit.get())), + ), + } +} + /// Admin 账号用例所需的公共账号、留存观测与 revision 事务能力。 /// /// 三个 PostgreSQL adapter 都保持私有，调用方只能取得 [`AccountStore`] 暴露的领域能力。 @@ -376,6 +390,11 @@ impl AccountStore for PgAdminAccountStore { let mut account = admin_account_record(summary)?; account.groups = groups_by_account.remove(&account_id).unwrap_or_default(); Ok(AccountPageItem { + capacity: account_capacity( + &account, + page.default_concurrency, + runtime.in_flight.as_ref(), + ), account, projection, }) @@ -411,7 +430,21 @@ impl AccountStore for PgAdminAccountStore { .await?; let mut account = admin_account_record(record.summary)?; account.groups = groups.remove(&account_id).unwrap_or_default(); + let default_concurrency: i64 = sqlx::query_scalar( + "select max_concurrent_per_account from runtime_settings where id = 1", + ) + .fetch_one(&self.pool) + .await + .map_err(|_| admin_store_error(ENTITY, postgres_unavailable("load account concurrency")))?; + let default_concurrency = u64::try_from(default_concurrency).map_err(|_| { + AdminStoreError::new( + AdminStoreErrorKind::Invalid, + ENTITY, + "invalid default account concurrency", + ) + })?; Ok(Some(AccountPageItem { + capacity: account_capacity(&account, default_concurrency, runtime.in_flight.as_ref()), account, projection, })) diff --git a/backend/crates/gateway-store/src/postgres/provider_accounts/admin_queries.rs b/backend/crates/gateway-store/src/postgres/provider_accounts/admin_queries.rs index c314b775d..17cd58913 100644 --- a/backend/crates/gateway-store/src/postgres/provider_accounts/admin_queries.rs +++ b/backend/crates/gateway-store/src/postgres/provider_accounts/admin_queries.rs @@ -4,6 +4,7 @@ use super::*; pub(crate) struct AdminAccountPageRows { pub(crate) config_revision: AdminRevision, + pub(crate) default_concurrency: u64, pub(crate) accounts: Vec, pub(crate) total: u64, pub(crate) summary: AccountSummary, @@ -112,7 +113,7 @@ pub(crate) async fn load_admin_account_page( select count(*)::bigint as filtered_total from filtered ), settings as ( - select config_revision, usage_retention_days + select config_revision, usage_retention_days, max_concurrent_per_account from runtime_settings where id = 1 ) @@ -127,7 +128,7 @@ pub(crate) async fn load_admin_account_page( global_summary.summary_total, global_summary.summary_normal, global_summary.summary_quota_exhausted, global_summary.summary_rate_limited, global_summary.summary_disabled, global_summary.summary_error, - settings.config_revision + settings.config_revision, settings.max_concurrent_per_account from filtered_total cross join global_summary cross join settings @@ -168,6 +169,7 @@ pub(crate) async fn load_admin_account_page( ) })?; let config_revision = revision_from_row(metadata)?; + let default_concurrency = unsigned_metadata(metadata, "max_concurrent_per_account")?; let total = unsigned_metadata(metadata, "filtered_total")?; let summary = AccountSummary { total: unsigned_metadata(metadata, "summary_total")?, @@ -190,6 +192,7 @@ pub(crate) async fn load_admin_account_page( .map_err(|error| admin_store_error(ENTITY, error))?; Ok(AdminAccountPageRows { config_revision, + default_concurrency, accounts, total, summary, diff --git a/backend/crates/gateway-store/tests/postgres/provider_accounts/admin_adapter.rs b/backend/crates/gateway-store/tests/postgres/provider_accounts/admin_adapter.rs index 6c1c4e8bb..578e49230 100644 --- a/backend/crates/gateway-store/tests/postgres/provider_accounts/admin_adapter.rs +++ b/backend/crates/gateway-store/tests/postgres/provider_accounts/admin_adapter.rs @@ -1,5 +1,74 @@ use super::*; +#[tokio::test] +async fn account_capacity_should_resolve_limits_for_list_and_detail() { + let Some(database) = TestDatabase::create("account_capacity").await else { + return; + }; + let repository = PgProviderAccountRepository::new(database.pool.clone()); + let store = admin_account_store(&database.pool); + let mut limited = account("acct_override", "user-override"); + limited.concurrency_limit = gateway_core::account::AccountConcurrencyLimit::new(3); + for record in [account("acct_inherited", "user-inherited"), limited] { + repository.insert_provider_account(record).await.unwrap(); + } + for default_limit in [10_i64, 0] { + sqlx::query("update runtime_settings set max_concurrent_per_account = $1 where id = 1") + .bind(default_limit) + .execute(&database.pool) + .await + .unwrap(); + for counts in [ + None, + Some(BTreeMap::from([("acct_override".to_owned(), 2)])), + ] { + let runtime = AccountRuntimeSnapshot { + in_flight: counts, + ..Default::default() + }; + let page = store + .list_accounts( + AccountListQuery { + page: 1, + page_size: PageSize::new(20).unwrap(), + provider_kind: None, + group_filter: None, + search: None, + status: None, + sort: None, + }, + runtime.clone(), + ) + .await + .unwrap(); + assert_eq!(page.items.len(), 2); + for item in page.items { + let inherited = item.account.id == "acct_inherited"; + let expected_limit = if inherited { + (default_limit > 0).then_some(10) + } else { + Some(3) + }; + assert_eq!(item.capacity.total_slots, expected_limit); + assert_eq!( + item.capacity.used_slots, + runtime + .in_flight + .as_ref() + .map(|_| if inherited { 0 } else { 2 }) + ); + let detail = store + .load_account(&item.account.id, runtime.clone()) + .await + .unwrap() + .unwrap(); + assert_eq!(detail.capacity, item.capacity); + } + } + } + database.close().await; +} + #[tokio::test] async fn provider_filter_applies_before_pagination_even_on_an_empty_page() { let Some(database) = TestDatabase::create("account_provider_filters").await else { diff --git a/docs/api.md b/docs/api.md index 8c6c49731..bfe966458 100644 --- a/docs/api.md +++ b/docs/api.md @@ -527,6 +527,10 @@ config 返回 `{ name, plaintextKey }`，仅读取服务端会话绑定的当前 账号与用量页面使用固定的 OpenAI/xAI 平台选项，省略 `provider` 表示不过滤。 +账号视图的 `capacity` 返回查询时的网关并发容量：`usedSlots` 是正在执行的请求占用数，不含排队请求， +读取运行态失败时为 `null`；`totalSlots` 是应用账号独立配置或全局默认值后的上限，`null` 表示不限。 +该上限不代表上游实际允许的并发数。管理端标记随账号列表查询刷新。 + 模型目录导出保留上游原生模型对象和能力字段，不包含账号凭据；不支持 Codex 原生目录的账号不能导出。 管理端下载文件名为 `cpr-model-catalog-<套餐>-<账号名称>.json`，文件正文为 `catalog`，可用于 Codex 的 `model_catalog_json` 配置。账号设置保存不等待上游模型目录刷新完成。 diff --git a/frontend/src/api/modules/accounts.ts b/frontend/src/api/modules/accounts.ts index ce64d66b8..138a32c3b 100644 --- a/frontend/src/api/modules/accounts.ts +++ b/frontend/src/api/modules/accounts.ts @@ -139,6 +139,10 @@ export interface Account { errorMessage: string | null enabled: boolean concurrencyLimit: number | null + capacity: { + usedSlots: number | null + totalSlots: number | null + } weight: number modelAccess: AccountModelAccess accessTokenExpiresAt: string | null diff --git a/frontend/src/views/accounts/components/AccountCapacityIndicator.vue b/frontend/src/views/accounts/components/AccountCapacityIndicator.vue new file mode 100644 index 000000000..c107dbf22 --- /dev/null +++ b/frontend/src/views/accounts/components/AccountCapacityIndicator.vue @@ -0,0 +1,31 @@ + + + diff --git a/frontend/src/views/accounts/components/AccountQuotaSummaryCell/Entry.vue b/frontend/src/views/accounts/components/AccountQuotaSummaryCell/Entry.vue index df81d0890..420e3deba 100644 --- a/frontend/src/views/accounts/components/AccountQuotaSummaryCell/Entry.vue +++ b/frontend/src/views/accounts/components/AccountQuotaSummaryCell/Entry.vue @@ -77,11 +77,12 @@ function quotaWindowCode(windowSeconds: number | null, role: AccountQuotaWindow[