starred/codex-proxy-rs
1
0
Fork 0
mirror of https://github.com/zyycn/codex-proxy-rs.git synced 2026-11-06 16:06:08 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(observability): preserve query budgets and usage history

Browse Source 
Hold PostgreSQL observation slots through connection cleanup and optimize exact usage pagination. Keep usage panels visible when analytics queries fail and backfill missing historical identity snapshots in one migration.

Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
Zyy
2026-10-12 00:15:40 +08:00
co-authored by Codex
parent 676044c43f
commit 8a406ff594
21 changed files with 741 additions and 207 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/crates/gateway-store/src/postgres/observability/admin_adapter.rs
+140 -95
View File
@@ -49,8 +49,9 @@ impl PgObservabilityRepository {
let accounts = self
.query_budget
.run(
&self.pool,
"load dashboard provider accounts",
load_account_statuses(&self.pool),
|mut connection| async move { load_account_statuses(&mut *connection).await },
)
.await?;
let now = observed_at.into();
@@ -180,32 +181,39 @@ impl PgObservabilityRepository {
};
let (totals, trend, analytics, account_usage) = self
.query_budget
.run("load dashboard analytics", async {
let mut snapshot = analytics::AnalyticsSnapshot::new(&self.pool).await?;
let totals = snapshot.totals().await?;
let trend = snapshot
.trend(
range,
&UsageRecordFilter::default(),
query.granularity,
self.timezone,
false,
)
.await?;
let account_usage =
provider_account_usage(&mut snapshot.tx, account_usage_query).await?;
let metadata = snapshot.metadata();
snapshot.finish().await?;
Ok((totals, trend, metadata, account_usage))
})
.run(
&self.pool,
"load dashboard analytics",
|mut connection| async move {
let mut snapshot = analytics::AnalyticsSnapshot::new(&mut connection).await?;
let totals = snapshot.totals().await?;
let trend = snapshot
.trend(
range,
&UsageRecordFilter::default(),
query.granularity,
self.timezone,
false,
)
.await?;
let account_usage =
provider_account_usage(&mut snapshot.tx, account_usage_query).await?;
let metadata = snapshot.metadata();
snapshot.finish().await?;
Ok((totals, trend, metadata, account_usage))
},
)
.await?;
let (provider_accounts, normal_accounts) =
self.account_status_snapshot(observed_at).await?;
let recent_requests = self
.query_budget
.run(
&self.pool,
"load dashboard recent requests",
list_usage_record_items(&self.pool, &recent_query),
|mut connection| async move {
list_usage_record_items(&mut *connection, &recent_query, None).await
},
)
.await?;
let runtime_slots = self.dashboard_runtime_slots(&normal_accounts).await?;
@@ -227,20 +235,24 @@ impl PgObservabilityRepository {
granularity: admin_observability::Granularity,
) -> StoreResult<Vec<RequestMetricPoint>> {
self.query_budget
.run("load dashboard analytics trend", async {
let mut snapshot = analytics::AnalyticsSnapshot::new(&self.pool).await?;
let points = snapshot
.trend(
range,
&UsageRecordFilter::default(),
granularity,
self.timezone,
false,
)
.await?;
snapshot.finish().await?;
Ok(points)
})
.run(
&self.pool,
"load dashboard analytics trend",
|mut connection| async move {
let mut snapshot = analytics::AnalyticsSnapshot::new(&mut connection).await?;
let points = snapshot
.trend(
range,
&UsageRecordFilter::default(),
granularity,
self.timezone,
false,
)
.await?;
snapshot.finish().await?;
Ok(points)
},
)
.await
}
@@ -251,14 +263,18 @@ impl PgObservabilityRepository {
granularity: admin_observability::Granularity,
) -> StoreResult<Vec<RequestMetricPoint>> {
self.query_budget
.run("load analytics trend", async {
let mut snapshot = analytics::AnalyticsSnapshot::new(&self.pool).await?;
let points = snapshot
.trend(range, &filter, granularity, self.timezone, true)
.await?;
snapshot.finish().await?;
Ok(points)
})
.run(
&self.pool,
"load analytics trend",
|mut connection| async move {
let mut snapshot = analytics::AnalyticsSnapshot::new(&mut connection).await?;
let points = snapshot
.trend(range, &filter, granularity, self.timezone, true)
.await?;
snapshot.finish().await?;
Ok(points)
},
)
.await
}
@@ -269,23 +285,27 @@ impl PgObservabilityRepository {
granularity: admin_observability::Granularity,
) -> StoreResult<admin_observability::UsageInsightsObservation> {
self.query_budget
.run("load usage insights snapshot", async {
let mut snapshot = analytics::AnalyticsSnapshot::new(&self.pool).await?;
let mut overview = snapshot.overview(range, &filter).await?;
let trend = snapshot
.trend(range, &filter, granularity, self.timezone, true)
.await?;
let costs = snapshot
.projected_costs(range, &filter, granularity, self.timezone)
.await?;
overview.analytics = snapshot.metadata();
snapshot.finish().await?;
Ok(admin_observability::UsageInsightsObservation {
overview: admin_usage_overview(overview),
trend: trend.into_iter().map(admin_request_metric_point).collect(),
costs,
})
})
.run(
&self.pool,
"load usage insights snapshot",
|mut connection| async move {
let mut snapshot = analytics::AnalyticsSnapshot::new(&mut connection).await?;
let mut overview = snapshot.overview(range, &filter).await?;
let trend = snapshot
.trend(range, &filter, granularity, self.timezone, true)
.await?;
let costs = snapshot
.projected_costs(range, &filter, granularity, self.timezone)
.await?;
overview.analytics = snapshot.metadata();
snapshot.finish().await?;
Ok(admin_observability::UsageInsightsObservation {
overview: admin_usage_overview(overview),
trend: trend.into_iter().map(admin_request_metric_point).collect(),
costs,
})
},
)
.await
}
@@ -294,16 +314,20 @@ impl PgObservabilityRepository {
query: ProviderAccountUsageQuery,
) -> StoreResult<Vec<ProviderAccountUsageObservation>> {
self.query_budget
.run("load provider account usage", async {
let mut snapshot = analytics::AnalyticsSnapshot::new(&self.pool).await?;
snapshot.require_visible(Some(query.range)).await?;
if let Some(range) = query.request_bucket_range {
snapshot.require_visible(Some(range)).await?;
}
let usage = provider_account_usage(&mut snapshot.tx, query).await?;
snapshot.finish().await?;
Ok(usage)
})
.run(
&self.pool,
"load provider account usage",
|mut connection| async move {
let mut snapshot = analytics::AnalyticsSnapshot::new(&mut connection).await?;
snapshot.require_visible(Some(query.range)).await?;
if let Some(range) = query.request_bucket_range {
snapshot.require_visible(Some(range)).await?;
}
let usage = provider_account_usage(&mut snapshot.tx, query).await?;
snapshot.finish().await?;
Ok(usage)
},
)
.await
}
@@ -312,15 +336,20 @@ impl PgObservabilityRepository {
query: UsageRecordQuery,
) -> StoreResult<UsageRecordPage> {
self.query_budget
.run("list usage records", list_usage_records(&self.pool, query))
.run(
&self.pool,
"list usage records",
|mut connection| async move { list_usage_records(&mut connection, query).await },
)
.await
}
pub async fn usage_record_detail(&self, request_id: &str) -> StoreResult<UsageRecordDetail> {
self.query_budget
.run(
&self.pool,
"load usage record detail",
usage_record_detail(&self.pool, request_id),
|mut connection| async move { usage_record_detail(&mut connection, request_id).await },
)
.await
}
@@ -331,12 +360,16 @@ impl PgObservabilityRepository {
filter: UsageRecordFilter,
) -> StoreResult<UsageOverview> {
self.query_budget
.run("load usage analytics", async {
let mut snapshot = analytics::AnalyticsSnapshot::new(&self.pool).await?;
let overview = snapshot.overview(range, &filter).await?;
snapshot.finish().await?;
Ok(overview)
})
.run(
&self.pool,
"load usage analytics",
|mut connection| async move {
let mut snapshot = analytics::AnalyticsSnapshot::new(&mut connection).await?;
let overview = snapshot.overview(range, &filter).await?;
snapshot.finish().await?;
Ok(overview)
},
)
.await
}
@@ -348,20 +381,26 @@ impl PgObservabilityRepository {
limit: u16,
) -> StoreResult<DiagnosticsObservation> {
self.query_budget
.run("load analytics diagnostics", async {
let mut snapshot = analytics::AnalyticsSnapshot::new(&self.pool).await?;
let observation = snapshot
.diagnostics(range, &filter, dimension, limit)
.await?;
snapshot.finish().await?;
Ok(observation)
})
.run(
&self.pool,
"load analytics diagnostics",
|mut connection| async move {
let mut snapshot = analytics::AnalyticsSnapshot::new(&mut connection).await?;
let observation = snapshot
.diagnostics(range, &filter, dimension, limit)
.await?;
snapshot.finish().await?;
Ok(observation)
},
)
.await
}
pub async fn list_ops_errors(&self, query: OpsErrorQuery) -> StoreResult<OpsErrorPage> {
self.query_budget
.run("list ops errors", list_ops_errors(&self.pool, query))
.run(&self.pool, "list ops errors", |mut connection| async move {
list_ops_errors(&mut connection, query).await
})
.await
}
}
@@ -371,15 +410,21 @@ impl AdminObservabilityStore for PgAdminObservabilityStore {
async fn analytics_cache_generation(&self) -> AdminStoreResult<u64> {
self.repository
.query_budget
.run("read analytics cache generation", async {
let generation: i64 = sqlx::query_scalar(
"select generation from request_analytics_build_state where singleton",
)
.fetch_one(&self.repository.pool)
.await
.map_err(|error| postgres_unavailable("read analytics cache generation", error))?;
u64::try_from(generation).map_err(|_| invalid("invalid analytics generation"))
})
.run(
&self.repository.pool,
"read analytics cache generation",
|mut connection| async move {
let generation: i64 = sqlx::query_scalar(
"select generation from request_analytics_build_state where singleton",
)
.fetch_one(&mut *connection)
.await
.map_err(|error| {
postgres_unavailable("read analytics cache generation", error)
})?;
u64::try_from(generation).map_err(|_| invalid("invalid analytics generation"))
},
)
.await
.map_err(observability_error)
}
backend/crates/gateway-store/src/postgres/observability/analytics/mod.rs
+4 -2
View File
@@ -17,8 +17,10 @@ const SCHEMA_VERSION: i32 = 1;
const HISTOGRAM_VERSION: i32 = histogram::HISTOGRAM_VERSION;
const MAX_FACT_ROWS: usize = 100_000;
pub(super) async fn begin_snapshot(pool: &PgPool) -> StoreResult<sqlx::Transaction<'_, Postgres>> {
let mut transaction = pool
pub(super) async fn begin_snapshot(
connection: &mut PgConnection,
) -> StoreResult<sqlx::Transaction<'_, Postgres>> {
let mut transaction = connection
.begin()
.await
.map_err(|error| postgres_unavailable("begin analytics snapshot", error))?;
backend/crates/gateway-store/src/postgres/observability/analytics/query.rs
+2 -2
View File
@@ -16,8 +16,8 @@ struct ReadSelection<'a> {
}
impl<'a> AnalyticsSnapshot<'a> {
pub async fn new(pool: &'a PgPool) -> StoreResult<Self> {
let mut tx = begin_snapshot(pool).await?;
pub async fn new(connection: &'a mut PgConnection) -> StoreResult<Self> {
let mut tx = begin_snapshot(connection).await?;
// 时间跨度和分组选择性差异很大，通用计划会把长区间误估成短区间
// 仅当前只读事务使用实际参数规划，不改变连接池后续事务的计划策略
sqlx::query("set local plan_cache_mode = force_custom_plan")
backend/crates/gateway-store/src/postgres/observability/analytics/worker.rs
+15 -12
View File
@@ -53,15 +53,18 @@ impl PgRequestAnalyticsRepository {
pub async fn status(&self) -> StoreResult<AnalyticsStatus> {
self.query_budget
.run("analytics status", async {
let mut transaction = begin_snapshot(&self.pool).await?;
let status = status(&mut transaction).await?;
transaction
.commit()
.await
.map_err(|error| postgres_unavailable("complete analytics status", error))?;
Ok(status)
})
.run(
&self.pool,
"analytics status",
|mut connection| async move {
let mut transaction = begin_snapshot(&mut connection).await?;
let status = status(&mut transaction).await?;
transaction.commit().await.map_err(|error| {
postgres_unavailable("complete analytics status", error)
})?;
Ok(status)
},
)
.await
}
@@ -69,9 +72,9 @@ impl PgRequestAnalyticsRepository {
if batch_size == 0 || batch_size > 10_000 {
return Err(invalid("analytics batch size must be between 1 and 10000"));
}
self.query_budget.run("maintain request analytics", async {
self.query_budget.run(&self.pool, "maintain request analytics", |mut connection| async move {
// 数据库事务锁同时约束单副本内重复任务与过期 Worker，源写入不获取此锁
let mut transaction = begin_snapshot(&self.pool).await?;
let mut transaction = begin_snapshot(&mut connection).await?;
let acquired: bool = sqlx::query_scalar("select pg_try_advisory_xact_lock(hashtextextended(current_schema()::text,382032))")
.fetch_one(&mut *transaction).await.map_err(|error| postgres_unavailable("claim analytics maintenance", error))?;
if !acquired { return Ok(AnalyticsMaintenance::default()); }
@@ -100,7 +103,7 @@ impl PgRequestAnalyticsRepository {
.bind(ids).bind(versions).execute(&mut *transaction).await.map_err(|error| postgres_unavailable("confirm analytics request versions", error))?;
transaction.commit().await.map_err(|error| postgres_unavailable("publish analytics maintenance", error))?;
// 每个阶段使用独立短事务，聚合只锁自己的队列，不把读放大传递到请求终结
let mut transaction = begin_snapshot(&self.pool).await?;
let mut transaction = begin_snapshot(&mut connection).await?;
let acquired: bool = sqlx::query_scalar("select pg_try_advisory_xact_lock(hashtextextended(current_schema()::text,382032))")
.fetch_one(&mut *transaction).await.map_err(|error| postgres_unavailable("claim analytics bucket maintenance", error))?;
if !acquired { return Ok(result); }
backend/crates/gateway-store/src/postgres/observability/mod.rs
+1 -1
View File
@@ -11,7 +11,7 @@ use gateway_admin::{
model::observability as admin_observability,
ports::store::{AdminStoreResult, ObservabilityStore as AdminObservabilityStore},
};
use sqlx::{PgPool, Postgres, QueryBuilder, Row};
use sqlx::{Connection, PgConnection, PgPool, Postgres, QueryBuilder, Row};
use crate::{
DecimalAmount, StoreError, StoreResult, admin_store_error, postgres_unavailable,
backend/crates/gateway-store/src/postgres/observability/queries/ops.rs
+2 -2
View File
@@ -69,11 +69,11 @@ left join model_request_observations mr on mr.id = oe.model_request_id
where true";
pub(crate) async fn list_ops_errors(
pool: &PgPool,
connection: &mut PgConnection,
query: OpsErrorQuery,
) -> StoreResult<OpsErrorPage> {
validate_ops_error_filter(&query.filter)?;
let mut transaction = analytics::begin_snapshot(pool).await?;
let mut transaction = analytics::begin_snapshot(connection).await?;
let total = count_ops_errors(&mut transaction, query.range, &query.filter).await?;
let offset = observability_page_offset(query.current_page, query.page_size)?;
let mut statement =
backend/crates/gateway-store/src/postgres/observability/queries/usage.rs
+43 -11
View File
@@ -174,13 +174,25 @@ pub(crate) const USAGE_RECORD_DETAIL_SELECT: &str =
from model_request_observations mr";
pub(crate) async fn list_usage_records(
pool: &PgPool,
connection: &mut PgConnection,
query: UsageRecordQuery,
) -> StoreResult<UsageRecordPage> {
query.filter.validate()?;
let mut transaction = analytics::begin_snapshot(pool).await?;
let total = count_usage_records(&mut *transaction, query.range, &query.filter).await?;
let items = list_usage_record_items(&mut *transaction, &query).await?;
let mut transaction = analytics::begin_snapshot(connection).await?;
// 第一页未填满时，当前快照中的行数已是精确总数，无需再次扫描计数
let (total, items) = if query.current_page == 1 {
let items = list_usage_record_items(&mut *transaction, &query, None).await?;
let total = if items.len() < usize::from(query.page_size.get()) {
items.len() as u64
} else {
count_usage_records(&mut *transaction, query.range, &query.filter).await?
};
(total, items)
} else {
let total = count_usage_records(&mut *transaction, query.range, &query.filter).await?;
let items = list_usage_record_items(&mut *transaction, &query, Some(total)).await?;
(total, items)
};
transaction
.commit()
.await
@@ -196,9 +208,25 @@ pub(crate) async fn list_usage_records(
pub(crate) async fn list_usage_record_items<'e, E: sqlx::Executor<'e, Database = Postgres>>(
executor: E,
query: &UsageRecordQuery,
total: Option<u64>,
) -> StoreResult<Vec<UsageListRecord>> {
query.filter.validate()?;
let offset = observability_page_offset(query.current_page, query.page_size)?;
let mut offset = observability_page_offset(query.current_page, query.page_size)?;
let mut limit = i64::from(query.page_size.get());
let mut ascending = false;
if let Some(total) = total {
let total = i64::try_from(total).map_err(|_| invalid("usage count overflows i64"))?;
if offset >= total {
return Ok(Vec::new());
}
limit = limit.min(total - offset);
let reverse_offset = total - offset - limit;
// 精确总数与候选页共用快照，尾页可从时间索引另一端读取而不改变页码语义
if reverse_offset < offset {
offset = reverse_offset;
ascending = true;
}
}
let mut statement = QueryBuilder::<Postgres>::new(
"with candidate_page as materialized (select mr.id, mr.started_at from model_request_observations mr",
);
@@ -208,8 +236,12 @@ pub(crate) async fn list_usage_record_items<'e, E: sqlx::Executor<'e, Database =
statement.push_bind(query.range.end);
push_completed_usage_fact_filter(&mut statement, "mr");
push_usage_filter(&mut statement, &query.filter, "mr");
statement.push(" order by mr.started_at desc, mr.id desc limit ");
statement.push_bind(i64::from(query.page_size.get()));
statement.push(if ascending {
" order by mr.started_at, mr.id limit "
} else {
" order by mr.started_at desc, mr.id desc limit "
});
statement.push_bind(limit);
statement.push(" offset ");
statement.push_bind(offset);
statement.push(") ");
@@ -251,7 +283,7 @@ pub(crate) async fn count_usage_records<'e, E: sqlx::Executor<'e, Database = Pos
}
pub(crate) async fn usage_record_detail(
pool: &PgPool,
connection: &mut PgConnection,
request_id: &str,
) -> StoreResult<UsageRecordDetail> {
require_nonempty("model request", "id", request_id)?;
@@ -261,7 +293,7 @@ pub(crate) async fn usage_record_detail(
statement.push_bind(request_id.to_owned());
let row = statement
.build()
.fetch_optional(pool)
.fetch_optional(&mut *connection)
.await
.map_err(|source| postgres_unavailable("load usage record detail", source))?
.ok_or_else(|| StoreError::NotFound {
@@ -284,7 +316,7 @@ pub(crate) async fn usage_record_detail(
order by attempt_index, created_at, id",
)
.bind(request_id)
.fetch_all(pool)
.fetch_all(&mut *connection)
.await
.map_err(|source| postgres_unavailable("load usage attempt observations", source))?;
let mut attempts = rows
@@ -304,7 +336,7 @@ pub(crate) async fn usage_record_detail(
from model_request_observations current join model_request_observations related
on related.id = current.recovery_request_id or related.recovery_request_id = current.id
where current.id = $1 order by related.started_at limit 20"
).bind(request_id).fetch_all(pool).await
).bind(request_id).fetch_all(&mut *connection).await
.map_err(|source| postgres_unavailable("load related recovery requests", source))?;
Ok(UsageRecordDetail {
request,
backend/crates/gateway-store/src/postgres/observability/query_budget.rs
+59 -10
View File
@@ -2,13 +2,15 @@
use std::{
future::Future,
ops::{Deref, DerefMut},
sync::Arc,
time::{Duration, Instant},
};
use sqlx::{PgConnection, PgPool, Postgres, pool::PoolConnection};
use tokio::sync::{OwnedSemaphorePermit, Semaphore};
use crate::{StoreBackend, StoreError, StoreResult};
use crate::{StoreBackend, StoreError, StoreResult, postgres_unavailable};
/// 所有管理观测入口共享的 PostgreSQL 连接槽位预算
#[derive(Clone)]
@@ -17,6 +19,38 @@ pub struct ObservabilityQueryBudget {
wait_timeout: Duration,
}
/// 观测名额与实际 PostgreSQL 连接共同存活，取消后仍覆盖协议收尾和事务回滚
pub struct ObservabilityConnection {
connection: PoolConnection<Postgres>,
permit: Option<OwnedSemaphorePermit>,
}
impl Deref for ObservabilityConnection {
type Target = PgConnection;
fn deref(&self) -> &Self::Target {
&self.connection
}
}
impl DerefMut for ObservabilityConnection {
fn deref_mut(&mut self) -> &mut Self::Target {
&mut self.connection
}
}
impl Drop for ObservabilityConnection {
fn drop(&mut self) {
let returning = self.connection.return_to_pool();
let permit = self.permit.take();
// SQLx 在确认旧语句结束并刷新回滚后才归还连接，名额不能早于这个边界释放
tokio::spawn(async move {
returning.await;
drop(permit);
});
}
}
impl ObservabilityQueryBudget {
/// 预算表示观测类 SQL 最多可同时占用的连接数
pub fn try_new(max_connections: u32, wait_timeout: Duration) -> StoreResult<Self> {
@@ -43,18 +77,33 @@ impl ObservabilityQueryBudget {
/// # Errors
///
/// 预算在等待期限内没有空闲槽位、预算已关闭，或查询本身失败时返回错误
pub async fn run<T, F>(&self, operation: &'static str, query: F) -> StoreResult<T>
pub async fn run<T, F, Q>(
&self,
pool: &PgPool,
operation: &'static str,
query: F,
) -> StoreResult<T>
where
F: Future<Output = StoreResult<T>>,
F: FnOnce(ObservabilityConnection) -> Q,
Q: Future<Output = StoreResult<T>>,
{
let _slot = self.acquire(operation).await?;
tokio::time::timeout(Duration::from_secs(10), query)
let permit = self.acquire(operation).await?;
tokio::time::timeout(Duration::from_secs(10), async {
let connection = pool.acquire().await.map_err(|source| {
postgres_unavailable("acquire observability connection", source)
})?;
query(ObservabilityConnection {
connection,
permit: Some(permit),
})
.await
.map_err(|source| StoreError::Unavailable {
backend: StoreBackend::PostgreSql,
message: "observability PostgreSQL execution budget is exhausted".to_owned(),
source: Some(gateway_core::error::ErrorSource::new(source)),
})?
})
.await
.map_err(|source| StoreError::Unavailable {
backend: StoreBackend::PostgreSql,
message: "observability PostgreSQL execution budget is exhausted".to_owned(),
source: Some(gateway_core::error::ErrorSource::new(source)),
})?
}
async fn acquire(&self, operation: &'static str) -> StoreResult<OwnedSemaphorePermit> {
backend/crates/gateway-store/src/postgres/provider_accounts/admin_adapter.rs
+16 -12
View File
@@ -120,18 +120,22 @@ impl PgAdminAccountStore {
.collect::<Vec<_>>();
let rows = self
.query_budget
.run("load account usage windows", async {
sqlx::query(sqlx::AssertSqlSafe(account_usage_by_windows_sql()))
.bind(account_ids)
.bind(keys)
.bind(starts)
.bind(ends)
.fetch_all(&self.pool)
.await
.map_err(|source| {
postgres_unavailable("load provider account quota window usage", source)
})
})
.run(
&self.pool,
"load account usage windows",
|mut connection| async move {
sqlx::query(sqlx::AssertSqlSafe(account_usage_by_windows_sql()))
.bind(account_ids)
.bind(keys)
.bind(starts)
.bind(ends)
.fetch_all(&mut *connection)
.await
.map_err(|source| {
postgres_unavailable("load provider account quota window usage", source)
})
},
)
.await
.map_err(|error| admin_store_error(ENTITY, error))?;
let mut usage_rows = Vec::with_capacity(windows.len());
backend/crates/gateway-store/src/postgres/provider_accounts/quota_forecast.rs
+14 -10
View File
@@ -24,16 +24,20 @@ pub(super) async fn load_history(
));
}
let rows = budget
.run("load quota forecast history", async {
sqlx::query(sqlx::AssertSqlSafe(history_sql()))
.bind(&window.account_id)
.bind(range.start)
.bind(range.end)
.bind(MAX_FORECAST_HISTORY_POINTS as i32)
.fetch_all(pool)
.await
.map_err(|source| postgres_unavailable("load quota forecast history", source))
})
.run(
pool,
"load quota forecast history",
|mut connection| async move {
sqlx::query(sqlx::AssertSqlSafe(history_sql()))
.bind(&window.account_id)
.bind(range.start)
.bind(range.end)
.bind(MAX_FORECAST_HISTORY_POINTS as i32)
.fetch_all(&mut *connection)
.await
.map_err(|source| postgres_unavailable("load quota forecast history", source))
},
)
.await
.map_err(|error| admin_store_error(ENTITY, error))?;
let mut history = QuotaForecastHistory::default();
backend/crates/gateway-store/src/postgres/provider_accounts/runtime.rs
+3 -3
View File
@@ -67,8 +67,8 @@ pub(crate) struct ProviderAccountStatus {
pub facts: AccountStatusFacts,
}
pub(crate) async fn load_account_statuses(
pool: &sqlx::PgPool,
pub(crate) async fn load_account_statuses<'e, E: sqlx::Executor<'e, Database = sqlx::Postgres>>(
executor: E,
) -> crate::StoreResult<Vec<ProviderAccountStatus>> {
use super::rows::{account_status_facts_from_row, get};
let rows = sqlx::query(
@@ -77,7 +77,7 @@ pub(crate) async fn load_account_statuses(
last_error_reason, last_error_message
from provider_accounts",
)
.fetch_all(pool)
.fetch_all(executor)
.await
.map_err(|source| crate::postgres_unavailable("load account status facts", source))?;
rows.iter()
backend/crates/gateway-store/tests/postgres/mod.rs
+1
View File
@@ -38,6 +38,7 @@ mod runtime_settings;
mod schema_integrity;
mod snapshot;
mod snapshots;
mod usage_paging;
static TEST_MIGRATOR: sqlx::migrate::Migrator = sqlx::migrate!("../../migrations");
backend/crates/gateway-store/tests/postgres/query_budget.rs
+253 -17
View File
@@ -3,8 +3,20 @@
use std::time::Duration;
use gateway_store::{StoreError, postgres::ObservabilityQueryBudget};
use sqlx::{Connection, PgPool, postgres::PgPoolOptions};
use tokio::sync::{mpsc, oneshot};
async fn test_pool() -> Option<PgPool> {
let url = crate::support::test_env("CPR_TEST_DATABASE_URL")?;
Some(
PgPoolOptions::new()
.max_connections(5)
.connect(&url)
.await
.unwrap(),
)
}
#[test]
fn query_budget_rejects_an_empty_budget() {
assert!(ObservabilityQueryBudget::try_new(0, Duration::from_secs(1)).is_err());
@@ -12,25 +24,33 @@ fn query_budget_rejects_an_empty_budget() {
#[tokio::test]
async fn running_query_deadline_releases_the_slot() {
let Some(pool) = test_pool().await else {
return;
};
let budget = ObservabilityQueryBudget::try_new(1, Duration::from_millis(10)).unwrap();
let error = budget
.run(
"never completes",
std::future::pending::<Result<(), StoreError>>(),
)
.run(&pool, "never completes", |connection| async move {
let result = std::future::pending::<Result<(), StoreError>>().await;
drop(connection);
result
})
.await
.expect_err("whole operation must have a deadline");
assert!(
matches!(error, StoreError::Unavailable { message, .. } if message == "observability PostgreSQL execution budget is exhausted")
);
budget
.run("after timeout", async { Ok(()) })
.run(&pool, "after timeout", |_connection| async { Ok(()) })
.await
.expect("timeout releases slot");
pool.close().await;
}
#[tokio::test]
async fn query_budget_times_out_without_exceeding_its_connection_limit() {
let Some(pool) = test_pool().await else {
return;
};
let budget = ObservabilityQueryBudget::try_new(4, Duration::from_millis(10))
.expect("valid query budget");
let (started_tx, mut started_rx) = mpsc::unbounded_channel();
@@ -38,16 +58,22 @@ async fn query_budget_times_out_without_exceeding_its_connection_limit() {
let mut occupied = Vec::new();
for _ in 0..4 {
let occupied_budget = budget.clone();
let occupied_pool = pool.clone();
let started_tx = started_tx.clone();
let (release_tx, release_rx) = oneshot::channel();
releases.push(release_tx);
occupied.push(tokio::spawn(async move {
occupied_budget
.run("occupied test query", async move {
started_tx.send(()).expect("report occupied slot");
release_rx.await.expect("release occupied slot");
Ok::<_, StoreError>(())
})
.run(
&occupied_pool,
"occupied test query",
|connection| async move {
started_tx.send(()).expect("report occupied slot");
release_rx.await.expect("release occupied slot");
drop(connection);
Ok::<_, StoreError>(())
},
)
.await
}));
}
@@ -56,7 +82,9 @@ async fn query_budget_times_out_without_exceeding_its_connection_limit() {
}
let error = budget
.run("queued test query", async { Ok::<_, StoreError>(()) })
.run(&pool, "queued test query", |_connection| async {
Ok::<_, StoreError>(())
})
.await
.expect_err("exhausted query budget must time out");
@@ -73,10 +101,14 @@ async fn query_budget_times_out_without_exceeding_its_connection_limit() {
.expect("occupied task must not panic")
.expect("occupied query must succeed");
}
pool.close().await;
}
#[tokio::test]
async fn queued_query_runs_after_one_slot_is_released() {
let Some(pool) = test_pool().await else {
return;
};
let budget =
ObservabilityQueryBudget::try_new(4, Duration::from_secs(1)).expect("valid query budget");
let (started_tx, mut started_rx) = mpsc::unbounded_channel();
@@ -84,16 +116,22 @@ async fn queued_query_runs_after_one_slot_is_released() {
let mut occupied = Vec::new();
for _ in 0..4 {
let occupied_budget = budget.clone();
let occupied_pool = pool.clone();
let started_tx = started_tx.clone();
let (release_tx, release_rx) = oneshot::channel();
releases.push(release_tx);
occupied.push(tokio::spawn(async move {
occupied_budget
.run("occupied test query", async move {
started_tx.send(()).expect("report occupied slot");
release_rx.await.expect("release occupied slot");
Ok::<_, StoreError>(())
})
.run(
&occupied_pool,
"occupied test query",
|connection| async move {
started_tx.send(()).expect("report occupied slot");
release_rx.await.expect("release occupied slot");
drop(connection);
Ok::<_, StoreError>(())
},
)
.await
}));
}
@@ -101,9 +139,12 @@ async fn queued_query_runs_after_one_slot_is_released() {
started_rx.recv().await.expect("occupied slot started");
}
let queued_budget = budget.clone();
let queued_pool = pool.clone();
let queued = tokio::spawn(async move {
queued_budget
.run("queued test query", async { Ok::<_, StoreError>(()) })
.run(&queued_pool, "queued test query", |_connection| async {
Ok::<_, StoreError>(())
})
.await
});
@@ -126,4 +167,199 @@ async fn queued_query_runs_after_one_slot_is_released() {
.expect("occupied task must not panic")
.expect("occupied query must succeed");
}
pool.close().await;
}
fn database_error(source: sqlx::Error) -> StoreError {
StoreError::Unavailable {
backend: gateway_store::StoreBackend::PostgreSql,
message: "query budget test SQL failed".to_owned(),
source: Some(gateway_core::error::ErrorSource::new(source)),
}
}
#[tokio::test]
async fn operation_timeout_keeps_the_slot_until_the_database_query_finishes() {
let Some(pool) = test_pool().await else {
return;
};
let budget = ObservabilityQueryBudget::try_new(1, Duration::from_millis(100)).unwrap();
let (pid_tx, pid_rx) = oneshot::channel();
let error = budget
.run(
&pool,
"multi statement deadline",
|mut connection| async move {
let pid: i32 = sqlx::query_scalar("select pg_backend_pid()")
.fetch_one(&mut *connection)
.await
.map_err(database_error)?;
pid_tx.send(pid).unwrap();
let mut tx = connection.begin().await.map_err(database_error)?;
sqlx::query("set local statement_timeout = '10s'")
.execute(&mut *tx)
.await
.map_err(database_error)?;
sqlx::query("select pg_sleep(6)")
.execute(&mut *tx)
.await
.map_err(database_error)?;
sqlx::query("select pg_sleep(9)")
.execute(&mut *tx)
.await
.map_err(database_error)?;
tx.commit().await.map_err(database_error)?;
Ok(())
},
)
.await
.expect_err("operation deadline must expire before the second statement");
assert!(
matches!(error, StoreError::Unavailable { message, .. } if message.contains("execution budget"))
);
let pid = pid_rx.await.unwrap();
let state: String = sqlx::query_scalar("select state from pg_stat_activity where pid=$1")
.bind(pid)
.fetch_one(&pool)
.await
.unwrap();
assert_eq!(
state, "active",
"the timed-out statement is still executing"
);
let queued = budget
.run(&pool, "after operation timeout", |_connection| async {
Ok(())
})
.await;
assert!(
matches!(queued, Err(StoreError::Unavailable { message, .. }) if message.contains("connection budget is exhausted"))
);
tokio::time::timeout(
Duration::from_secs(1),
sqlx::query("select 1").execute(&pool),
)
.await
.expect("data plane capacity stays available")
.unwrap();
// 只取消本用例持有的后台查询，避免让睡眠耗尽测试时间
sqlx::query("select pg_cancel_backend($1)")
.bind(pid)
.execute(&pool)
.await
.unwrap();
tokio::time::timeout(Duration::from_secs(2), async {
loop {
if budget
.run(&pool, "after database cancellation", |_connection| async {
Ok(())
})
.await
.is_ok()
{
break;
}
}
})
.await
.expect("completed rollback releases the budget");
pool.close().await;
}
#[tokio::test]
async fn dropping_an_operation_keeps_the_slot_through_database_rollback() {
let Some(pool) = test_pool().await else {
return;
};
let budget = ObservabilityQueryBudget::try_new(1, Duration::from_millis(100)).unwrap();
let running_pool = pool.clone();
let running_budget = budget.clone();
let (pid_tx, pid_rx) = oneshot::channel();
let running = tokio::spawn(async move {
running_budget
.run(
&running_pool,
"cancelled operation",
|mut connection| async move {
let pid: i32 = sqlx::query_scalar("select pg_backend_pid()")
.fetch_one(&mut *connection)
.await
.map_err(database_error)?;
pid_tx.send(pid).unwrap();
let mut tx = connection.begin().await.map_err(database_error)?;
sqlx::query("set local statement_timeout = '10s'")
.execute(&mut *tx)
.await
.map_err(database_error)?;
sqlx::query("select pg_sleep(20)")
.execute(&mut *tx)
.await
.map_err(database_error)?;
tx.commit().await.map_err(database_error)?;
Ok(())
},
)
.await
});
let pid = pid_rx.await.unwrap();
tokio::time::timeout(Duration::from_secs(2), async {
loop {
let sleeping: bool = sqlx::query_scalar(
"select coalesce(wait_event='PgSleep',false) from pg_stat_activity where pid=$1",
)
.bind(pid)
.fetch_one(&pool)
.await
.unwrap();
if sleeping {
break;
}
tokio::time::sleep(Duration::from_millis(10)).await;
}
})
.await
.expect("query reaches the server before cancellation");
running.abort();
assert!(running.await.unwrap_err().is_cancelled());
let queued = budget
.run(&pool, "after caller cancellation", |_connection| async {
Ok(())
})
.await;
assert!(
matches!(queued, Err(StoreError::Unavailable { message, .. }) if message.contains("connection budget is exhausted"))
);
sqlx::query("select pg_cancel_backend($1)")
.bind(pid)
.execute(&pool)
.await
.unwrap();
tokio::time::timeout(Duration::from_secs(2), async {
loop {
if budget
.run(&pool, "after rollback", |mut connection| async move {
sqlx::query("select 1")
.execute(&mut *connection)
.await
.map_err(database_error)?;
Ok(())
})
.await
.is_ok()
{
break;
}
}
})
.await
.expect("cancelled operation eventually returns a usable connection");
let idle_transactions: i64 = sqlx::query_scalar(
"select count(*) from pg_stat_activity where pid=$1 and state='idle in transaction'",
)
.bind(pid)
.fetch_one(&pool)
.await
.unwrap();
assert_eq!(idle_transactions, 0);
pool.close().await;
}
backend/crates/gateway-store/tests/postgres/usage_paging.rs
+72
View File
@@ -0,0 +1,72 @@
//! 请求列表的精确总数、双向分页与筛选隔离
use chrono::{TimeDelta, Utc};
use gateway_store::postgres::{
ObservabilityPageSize, ObservabilityRange, UsageRecordFilter, UsageRecordQuery,
};
use super::{TestDatabase, analytics::seed, observability_repository};
#[tokio::test]
async fn usage_pages_preserve_order_and_exact_totals_from_both_ends() {
let Some(database) = TestDatabase::create("usage_bidirectional_pages").await else {
return;
};
let start = Utc::now() - TimeDelta::hours(1);
seed(&database.pool, "page_00", start, &[]).await;
sqlx::query("insert into model_requests select cloned.* from model_requests original cross join generate_series(1,22) n cross join lateral jsonb_populate_record(null::model_requests,to_jsonb(original)||jsonb_build_object('id','page_'||lpad(n::text,2,'0'),'client_api_key_ref',case when n%2=0 then 'key_analytics' else 'other_key' end)) cloned where original.id='page_00'")
.execute(&database.pool).await.unwrap();
let range =
ObservabilityRange::new(start - TimeDelta::seconds(1), start + TimeDelta::seconds(1))
.unwrap();
let repository = observability_repository(&database.pool);
for key in [
None,
Some("key_analytics"),
Some("other_key"),
Some("absent_key"),
] {
let expected = (0..23)
.rev()
.filter(|n| match key {
None => true,
Some("key_analytics") => n % 2 == 0,
Some("other_key") => n % 2 != 0,
Some(_) => false,
})
.map(|n| format!("page_{n:02}"))
.collect::<Vec<_>>();
for page in 1..=5 {
let result = repository
.list_usage_records(UsageRecordQuery {
range,
filter: UsageRecordFilter {
client_api_key_ref: key.map(str::to_owned),
..Default::default()
},
current_page: page,
page_size: ObservabilityPageSize::new(7).unwrap(),
})
.await
.unwrap();
let ids = result
.items
.iter()
.map(|item| item.id.as_str())
.collect::<Vec<_>>();
let expected_page = expected
.iter()
.skip((page as usize - 1) * 7)
.take(7)
.map(String::as_str)
.collect::<Vec<_>>();
assert_eq!(
result.total,
expected.len() as u64,
"key={key:?}, page={page}"
);
assert_eq!(ids, expected_page, "key={key:?}, page={page}");
}
}
database.close().await;
}
backend/migrations/.frozen-sha256
+1
View File
@@ -26,3 +26,4 @@ fa99dde0fe0b3d7da6693d76b2735629dde0dc255ea8c53f0354857a4e80669b 0024_preferred
d4b599066d7382a1c9cb6b3c003200046579e4fce3531f232cbd665a9d92c542 0026_codex_privacy_policy.sql
8c9bbecd8315a26e928213a82a2ec651c35976e0643b82caef1d1bcfeee2916f 0027_request_observation_projection.sql
31d6668707186d7c19cdd507e20162a08c7fbb8ba41ff4a4439d0a5e08369cad 0028_request_analytics.sql
1923f14095dac5f808a530db002c37e17168afa0778bc7f3da768c818133eec9 0029_usage_indexes_and_identity_snapshots.sql
backend/migrations/0029_usage_indexes_and_identity_snapshots.sql
+61
View File
@@ -0,0 +1,61 @@
-- Key 与账号用量列表的精确计数和双向分页只扫描已交付请求索引
create index model_requests_delivered_client on model_requests (client_api_key_ref,started_at,id)
where outcome='succeeded' and downstream_committed_at is not null
and (provider_kind is distinct from 'openai' or request_kind is distinct from 'prewarm')
and ((client_transport='websocket' and client_status_code is null) or client_status_code between 200 and 399)
and (requested_model_id is not null or upstream_model_id is not null or image_generation_requested
or input_tokens is not null or output_tokens is not null or cached_input_tokens is not null
or cache_write_input_tokens is not null or reasoning_output_tokens is not null
or image_input_tokens is not null or image_output_tokens is not null
or total_tokens is not null or cost_amount is not null);
create index model_requests_delivered_account on model_requests (provider_account_ref,started_at,id)
where outcome='succeeded' and downstream_committed_at is not null
and (provider_kind is distinct from 'openai' or request_kind is distinct from 'prewarm')
and ((client_transport='websocket' and client_status_code is null) or client_status_code between 200 and 399)
and (requested_model_id is not null or upstream_model_id is not null or image_generation_requested
or input_tokens is not null or output_tokens is not null or cached_input_tokens is not null
or cache_write_input_tokens is not null or reasoning_output_tokens is not null
or image_input_tokens is not null or image_output_tokens is not null
or total_tokens is not null or cost_amount is not null);
-- 只补齐快照启用前的记录，已有快照和启用后采集到的未知值保持不变
-- 补写值来自此刻仍存在的配置，不代表能够还原请求发生时的真实名称或订阅
update model_requests mr
set request_observation_json = jsonb_set(
mr.request_observation_json,
'{request}',
coalesce(mr.request_observation_json -> 'request', '{}'::jsonb)
|| jsonb_build_object('clientApiKeyNameSnapshot', client_key.name)
)
from client_api_keys client_key
where client_key.id = mr.client_api_key_ref
and mr.request_observation_json #> '{request,clientApiKeyNameSnapshot}' is null
and mr.started_at < (
select installed_on from _sqlx_migrations where version = 28 and success
);
update model_requests mr
set request_observation_json = jsonb_set(
mr.request_observation_json,
'{account}',
coalesce(mr.request_observation_json -> 'account', '{}'::jsonb)
|| jsonb_build_object('planTypeSnapshot', account.plan_type)
)
from provider_accounts account
where account.id = mr.provider_account_ref
and account.plan_type is not null
and mr.request_observation_json #> '{account,planTypeSnapshot}' is null
and mr.started_at < (
select installed_on from _sqlx_migrations where version = 28 and success
);
update ops_events event
set provider_account_plan_type_snapshot = account.plan_type
from provider_accounts account
where account.id = event.provider_account_ref
and account.plan_type is not null
and event.provider_account_plan_type_snapshot is null
and event.created_at < (
select installed_on from _sqlx_migrations where version = 28 and success
);
docs/api.md
+7 -6
View File
@@ -1824,15 +1824,16 @@ request/response/upstream ID、outcome 与搜索文本。诊断 `dimension` 可
`requestShare` 的分母为该维度筛选后、截取前的全部请求数；`failureClass` 只在带错误类型的请求内计算占比
`retryCount` 为额外执行尝试次数之和，`retryRate` 为发生过重试的请求数占该组请求数的比例，同一请求多次重试只计一次
账号维度诊断项、使用记录列表和错误排查列表的 `accountPlanType` / `accountPlanTypeDisplay` 返回账号当前订阅及展示名称，
按各记录的内部账号 ID 关联；订阅未知或账号已删除时为 `null`，不作为请求发生时的订阅快照
使用记录列表和错误排查列表的 `accountPlanType` / `accountPlanTypeDisplay` 来自记录保存的账号订阅快照。
账号维度诊断使用该账号最近请求的订阅快照；未保存订阅时为 `null`
管理端请求列表及 Dashboard 最近请求中的 `accountNotes` 为账号当前备注，按内部账号 ID 关联。
备注不写入请求历史快照；无备注或账号已删除时返回 `null`，修改备注不改变历史请求的账号归属
管理端请求列表及 Dashboard 最近请求中的 `accountNotes` 来自请求保存的账号备注快照，未保存时为 `null`
`/api/admin/usage/records` 与 `/api/admin/operations/errors` 的记录返回 `clientApiKeyName`，为关联 Key 的当前名称
`/api/admin/usage/records` 与 `/api/admin/operations/errors` 的 `clientApiKeyName` 来自请求保存的 Key 名称快照，
未保存时为 `null`，不影响记录返回，不包含密钥原文
Key 已删除或未关联时为 `null`，不影响记录返回，不包含密钥原文
查询不关联当前配置，改名、修改订阅或删除配置不改变已有快照。补写的历史 Key 名称与订阅取自补写时仍存在的配置，
不保证等于请求发生时的值；无法补齐的字段保持未知
运维错误记录的 `errorDetails` 为受控诊断文本，无详情时为 `null`。
错误来源快照以 JSON 文本保存：`causes.messages` 按外层到内层排列本地原因，`causes.truncated`
docs/architecture.md
+4
View File
@@ -853,6 +853,10 @@ OpenAI 订阅周期属于按需个人信息，不是额度事实。Admin 账号
## 10. 观测与后台任务
管理观测查询与统计 Worker 共用 PostgreSQL 连接预算，为请求执行保留连接容量。每项操作使用一个受预算约束的连接；
超时或调用方取消后，名额持续覆盖 SQLx 的协议收尾、事务回滚和连接归还，不能随查询 Future 提前释放。
请求列表的精确总数与候选页共用数据库快照，按距首尾的较短方向读取，返回顺序和页码语义保持一致。
### 额度预测
账号容量预测属于 Admin 的只读派生规则，不参与 quota 权威状态、调度或金额结算。Store 通过专用采样端口
frontend/pnpm-lock.yaml
Generated
+25
View File
@@ -6,6 +6,9 @@ importers:
.:
configDependencies: {}
packageManagerDependencies:
'@pnpm/exe':
specifier: 12.6.0
version: 12.6.0
pnpm:
specifier: 12.6.0
version: 12.6.0
@@ -89,6 +92,11 @@ packages:
cpu: [x64]
os: [win32]
'@pnpm/exe@12.6.0':
resolution: {integrity: sha512-CkNetlJZugjKp9vPvWqCm5ahORU6FfO46BlOqqSCKLJ+W6QcRgM0qmmPTvlgrXVvkAiH+9hE+mMp/wSaqcU5Aw==}
engines: {node: '>=18.*'}
hasBin: true
pnpm@12.6.0:
resolution: {integrity: sha512-PvaPlRyxEawgS0paFvCy3fDaVqluBBPoHYVdnwtV75JnFHCQKOHNAMQFwsX7e56OxNxGd3yAXQNzwvL/AP0g7A==}
engines: {node: '>=18.*'}
@@ -138,6 +146,23 @@ snapshots:
'@pnpm/exe.win32-x64@12.6.0':
optional: true
'@pnpm/exe@12.6.0':
optionalDependencies:
'@pnpm/exe.android-arm64': 12.6.0
'@pnpm/exe.android-x64': 12.6.0
'@pnpm/exe.darwin-arm64': 12.6.0
'@pnpm/exe.darwin-x64': 12.6.0
'@pnpm/exe.freebsd-x64': 12.6.0
'@pnpm/exe.linux-arm64': 12.6.0
'@pnpm/exe.linux-arm64-musl': 12.6.0
'@pnpm/exe.linux-ppc64': 12.6.0
'@pnpm/exe.linux-riscv64': 12.6.0
'@pnpm/exe.linux-s390x': 12.6.0
'@pnpm/exe.linux-x64': 12.6.0
'@pnpm/exe.linux-x64-musl': 12.6.0
'@pnpm/exe.win32-arm64': 12.6.0
'@pnpm/exe.win32-x64': 12.6.0
pnpm@12.6.0:
optionalDependencies:
'@pnpm/exe.android-arm64': 12.6.0
frontend/src/views/usage/composables/useUsageRecordsTable.ts
+11 -11
View File
@@ -23,12 +23,10 @@ type LoadingAnalytics<T> = Omit<T, 'analytics'> & { analytics: AnalyticsMetadata
interface UsageLoadOptions {
scope?: UsageLoadScope
background?: boolean
}
export function useUsageRecordsTable(options: UseUsageRecordsTableOptions) {
const analyticsLoading = shallowRef(true)
const analyticsError = shallowRef(false)
const summary = shallowRef(emptySummary())
const insights = shallowRef(emptyInsights())
const searchQuery = shallowRef('')
@@ -73,18 +71,18 @@ export function useUsageRecordsTable(options: UseUsageRecordsTableOptions) {
}
async function loadUsageRecords(loadOptions: UsageLoadOptions = {}) {
const { scope = 'all', background = false } = loadOptions
const { scope = 'all' } = loadOptions
const globalParams = scopedParams()
if (scope === 'all')
resetTable()
await Promise.all([
...(options.active.value ? [table.execute()] : []),
...(scope === 'all' ? [loadUsageAnalytics(globalParams, background)] : []),
...(scope === 'all' ? [loadUsageAnalytics(globalParams)] : []),
])
}
async function loadUsageAnalytics(globalParams: ReturnType<typeof scopedParams>, background: boolean) {
async function loadUsageAnalytics(globalParams: ReturnType<typeof scopedParams>) {
const requestId = ++analyticsRequestId
const diagnosticsId = ++diagnosticRequestId
analyticsController?.abort()
@@ -92,8 +90,9 @@ export function useUsageRecordsTable(options: UseUsageRecordsTableOptions) {
analyticsController = new AbortController()
const requestOptions = { signal: analyticsController.signal }
const dimension = diagnosticDimension.value
analyticsLoading.value = !background
analyticsError.value = false
analyticsLoading.value = true
summary.value = emptySummary()
insights.value = emptyInsights()
try {
const [nextSummary, overview, diagnostics] = await Promise.all([
getUsageRecordSummary(globalParams, requestOptions),
@@ -116,8 +115,10 @@ export function useUsageRecordsTable(options: UseUsageRecordsTableOptions) {
}
}
catch {
if (requestId === analyticsRequestId)
analyticsError.value = true
if (requestId === analyticsRequestId) {
summary.value = emptySummary()
insights.value = emptyInsights()
}
}
finally {
if (requestId === analyticsRequestId) {
@@ -190,7 +191,7 @@ export function useUsageRecordsTable(options: UseUsageRecordsTableOptions) {
})
watch(providerQuery, () => {
void loadUsageRecords({ background: true })
void loadUsageRecords()
})
watch(options.active, (active) => {
@@ -226,7 +227,6 @@ export function useUsageRecordsTable(options: UseUsageRecordsTableOptions) {
loading,
error,
analyticsLoading,
analyticsError,
records,
summary,
insights,
frontend/src/views/usage/index.vue
+7 -13
View File
@@ -33,7 +33,6 @@ const {
loading,
error,
analyticsLoading,
analyticsError,
records,
summary,
insights,
@@ -71,18 +70,13 @@ watch(timeRange, () => {
</template>
</ZPageHeader>
<p v-if="analyticsError" class="mt-4 text-cp-sm text-cp-warning-text" role="alert">
统计暂不可用，请稍后重试
</p>
<template v-else>
<UsageSummaryCards :summary="summary" />
<UsageInsightsGrid
v-model:diagnostic-dimension="diagnosticDimension"
:overview="insights.overview"
:diagnostics="insights.diagnostics"
:loading="analyticsLoading"
/>
</template>
<UsageSummaryCards :summary="summary" />
<UsageInsightsGrid
v-model:diagnostic-dimension="diagnosticDimension"
:overview="insights.overview"
:diagnostics="insights.diagnostics"
:loading="analyticsLoading"
/>
<ZCard
class="mt-5 flex flex-col"