feat(accounts): show current concurrency capacity

This commit is contained in:
Zyy
2026-09-27 22:36:51 +08:00
parent 3bf7030fba
commit ea4e3059db
18 changed files with 359 additions and 11 deletions
@@ -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<u64>,
/// 应用账号覆盖或全局默认值后的上限;`None` 表示不限。
pub total_slots: Option<u64>,
}
/// 统一账号目录的全局状态计数,不受当前筛选和分页影响。
@@ -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<String>,
@@ -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::<Vec<_>>();
// 只读取当前页的租约占用;观测失败不能把账号误报为空闲或拖垮目录。
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(),
@@ -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()),
}
@@ -0,0 +1,107 @@
use std::collections::BTreeMap;
use gateway_admin::ports::store::AccountRuntimeStore;
use super::*;
struct CapacityRuntime {
counts: Option<BTreeMap<String, u64>>,
fail: bool,
reads: Mutex<Vec<Vec<String>>>,
}
#[async_trait]
impl AccountRuntimeStore for CapacityRuntime {
async fn active_rate_limits(&self) -> AdminStoreResult<AccountRuntimeSnapshot> {
Ok(AccountRuntimeSnapshot::default())
}
async fn account_runtime(&self, ids: &[String]) -> AdminStoreResult<AccountRuntimeSnapshot> {
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<BTreeMap<String, gateway_admin::model::accounts::AccountFreeze>> {
Ok(BTreeMap::new())
}
async fn capacity_peaks(&self, _: &[String]) -> AdminStoreResult<BTreeMap<String, u32>> {
Ok(BTreeMap::new())
}
async fn finish_freeze(
&self,
_: &str,
_: &gateway_admin::model::accounts::AccountFreeze,
_: Option<chrono::DateTime<Utc>>,
) -> AdminStoreResult<bool> {
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));
}
}
}
}
@@ -60,6 +60,7 @@ pub(super) fn account_models_data(result: ProviderModels) -> AccountModelsData {
pub(super) fn account_view(item: AccountDirectoryItem, now: DateTime<Utc>) -> AccountView {
let AccountDirectoryItem {
account,
capacity,
capabilities,
plan_type_display,
projection,
@@ -109,6 +110,10 @@ pub(super) fn account_view(item: AccountDirectoryItem, now: DateTime<Utc>) -> 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
@@ -290,6 +290,7 @@ pub struct AccountView {
pub error_message: Option<String>,
pub enabled: bool,
pub concurrency_limit: Option<u32>,
pub capacity: AccountCapacityView,
pub weight: u16,
pub model_access: gateway_core::account::AccountModelAccess,
pub access_token_expires_at: Option<String>,
@@ -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<u64>,
pub total_slots: Option<u64>,
}
#[derive(Debug, Clone, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct AccountCapabilitiesView {
@@ -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(),
@@ -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<String, u64>>,
) -> 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,
}))
@@ -4,6 +4,7 @@ use super::*;
pub(crate) struct AdminAccountPageRows {
pub(crate) config_revision: AdminRevision,
pub(crate) default_concurrency: u64,
pub(crate) accounts: Vec<ProviderAccountSummary>,
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,
@@ -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 {
+4
View File
@@ -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` 配置。账号设置保存不等待上游模型目录刷新完成。
+4
View File
@@ -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
@@ -0,0 +1,31 @@
<script setup lang="ts">
import type { Account } from '@/api'
import { Grid2X2, Slash } from '@lucide/vue'
import { computed } from 'vue'
const props = defineProps<{
capacity: Account['capacity']
}>()
const full = computed(() => props.capacity.usedSlots !== null
&& props.capacity.totalSlots !== null
&& props.capacity.usedSlots >= props.capacity.totalSlots)
const label = computed(() => `账号并发占用 ${props.capacity.usedSlots ?? '未知'},上限 ${props.capacity.totalSlots ?? '不限'}`)
</script>
<template>
<span
role="img"
:aria-label="label"
class="inline-flex h-3 shrink-0 items-center gap-1 whitespace-nowrap px-1 font-mono text-[9px] leading-none tabular-nums relative top-px"
:class="full ? 'text-cp-warning' : 'text-cp-text-tertiary'"
>
<Grid2X2 class="size-2.5 shrink-0" :stroke-width="1.8" aria-hidden="true" />
<!-- 数字按字面高度居中,分隔线使用图标,避免不同字形的基线影响对齐。 -->
<span aria-hidden="true" class="inline-flex items-center gap-0.5">
<span class="[text-box:trim-both_cap_alphabetic]">{{ capacity.usedSlots ?? '—' }}</span>
<Slash class="size-1.75 shrink-0 -rotate-20" aria-hidden="true" />
<span class="[text-box:trim-both_cap_alphabetic]">{{ capacity.totalSlots ?? '∞' }}</span>
</span>
</span>
</template>
@@ -77,11 +77,12 @@ function quotaWindowCode(windowSeconds: number | null, role: AccountQuotaWindow[
</script>
<template>
<BasePopover class="w-full" trigger="hover-click" placement="right" :hover-delay="240">
<BasePopover class="grid! w-full grid-cols-subgrid" trigger="hover-click" placement="right" :hover-delay="240">
<template #trigger="{ open }">
<button
type="button"
class="block w-full cursor-pointer rounded-sm border-0 bg-transparent p-0 text-left outline-none focus-visible:ring-2 focus-visible:ring-cp-control-outline"
class="w-full min-w-0 cursor-pointer rounded-sm border-0 bg-transparent p-0 text-left outline-none focus-visible:ring-2 focus-visible:ring-cp-control-outline"
:class="hasQuotaWindow ? 'col-span-full grid grid-cols-subgrid' : 'col-start-1 block'"
:aria-label="`查看${detailTitle}详情`"
:aria-expanded="open"
aria-haspopup="dialog"
@@ -37,8 +37,8 @@ const trackGridStyle = computed(() => ({
</script>
<template>
<div class="grid min-w-0 gap-1.5">
<div class="flex min-w-0 items-baseline justify-between gap-1 text-[10px] leading-3 font-bold">
<div class="col-span-full grid min-w-0 grid-cols-subgrid gap-y-1.5">
<div class="col-span-full grid min-w-0 grid-cols-subgrid items-baseline text-[10px] leading-3 font-bold">
<span class="min-w-0 truncate text-cp-text-quaternary" :title="label">
{{ label }}
</span>
@@ -51,7 +51,7 @@ const trackGridStyle = computed(() => ({
</strong>
</div>
<div class="grid min-w-0 gap-1" :style="trackGridStyle">
<div class="col-span-full grid min-w-0 gap-1" :style="trackGridStyle">
<div
v-for="item in windowItems"
:key="item.key"
@@ -3,6 +3,7 @@ import type { AccountRow } from '../../constants'
import { computed } from 'vue'
import { groupedAccountQuotaWindows, visibleSummaryQuotaWindows } from '../../constants'
import AccountCapacityIndicator from '../AccountCapacityIndicator.vue'
import AccountUsageWindow from '../AccountUsageWindow/index.vue'
import { quotaWindowPresentation } from '../AccountUsageWindow/presenter'
import AccountQuotaSummaryEntry from './Entry.vue'
@@ -32,14 +33,17 @@ const additionalEntryCount = computed(() => Math.max(summaryEntries.value.length
<div class="box-border grid min-h-16.5 w-full min-w-0 content-center gap-1.5 py-1.5">
<template v-if="account.authenticationKind === 'api_key'">
<span
class="flex min-w-0 items-baseline gap-1 font-mono tabular-nums"
class="flex min-w-0 items-baseline gap-1 font-mono leading-none tabular-nums"
title="本地累计总 Token"
>
<strong class="truncate text-cp-xs font-heavy text-cp-text">{{ account.usage.totalTokensDisplay }}</strong>
<span class="shrink-0 text-[9px] font-emphasis tracking-[0.02em] text-cp-text-quaternary">Tokens</span>
</span>
<div class="grid min-w-0 gap-1.5">
<span class="text-[10px] leading-3 font-bold text-cp-text-quaternary">{{ account.usage.windowLabelDisplay }}</span>
<div class="flex min-w-0 items-center justify-between gap-2 leading-3">
<span class="truncate text-[10px] font-bold text-cp-text-quaternary">{{ account.usage.windowLabelDisplay }}</span>
<AccountCapacityIndicator :capacity="account.capacity" />
</div>
<div class="h-1 w-full rounded-full bg-cp-success" title="上游额度未提供,绿色条不表示剩余额度" aria-hidden="true" />
</div>
</template>
@@ -71,12 +75,18 @@ const additionalEntryCount = computed(() => Math.max(summaryEntries.value.length
</div>
<div v-if="recentUsageEntry" class="flex min-w-0 items-end gap-2">
<div class="flex min-w-0 flex-1">
<!-- 共用列轨道为容量留位,进度条仍铺满整行,容量标记独立于额度弹层 -->
<div class="grid min-w-0 flex-1 grid-cols-[minmax(0,1fr)_auto] gap-x-2">
<AccountQuotaSummaryEntry
class="col-span-full col-start-1 row-start-1"
:label="recentUsageEntry.label"
:windows="recentUsageEntry.windows"
:show-percentage="false"
/>
<AccountCapacityIndicator
:capacity="account.capacity"
class="z-1 col-start-2 row-start-1 self-start"
/>
</div>
<span
v-if="additionalEntryCount > 0"
@@ -127,7 +127,6 @@ const view = computed(() => resolveAccountUsageWindowPresentation({
<div v-else :class="view.classes.header">
<span class="min-w-0 text-cp-text-secondary">额度待观测</span>
<span class="shrink-0 font-mono text-cp-text-quaternary" :class="view.classes.value">—</span>
</div>
</div>
</template>