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 058840def..ca6f46c97 100644 --- a/backend/crates/gateway-store/src/postgres/observability/admin_adapter.rs +++ b/backend/crates/gateway-store/src/postgres/observability/admin_adapter.rs @@ -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> { 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> { 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 { 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> { 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 { 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 { 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 { 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 { 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 { 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 { 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) } diff --git a/backend/crates/gateway-store/src/postgres/observability/analytics/mod.rs b/backend/crates/gateway-store/src/postgres/observability/analytics/mod.rs index c22ebbba1..44616c02e 100644 --- a/backend/crates/gateway-store/src/postgres/observability/analytics/mod.rs +++ b/backend/crates/gateway-store/src/postgres/observability/analytics/mod.rs @@ -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> { - let mut transaction = pool +pub(super) async fn begin_snapshot( + connection: &mut PgConnection, +) -> StoreResult> { + let mut transaction = connection .begin() .await .map_err(|error| postgres_unavailable("begin analytics snapshot", error))?; diff --git a/backend/crates/gateway-store/src/postgres/observability/analytics/query.rs b/backend/crates/gateway-store/src/postgres/observability/analytics/query.rs index b9472eab2..4d54f421f 100644 --- a/backend/crates/gateway-store/src/postgres/observability/analytics/query.rs +++ b/backend/crates/gateway-store/src/postgres/observability/analytics/query.rs @@ -16,8 +16,8 @@ struct ReadSelection<'a> { } impl<'a> AnalyticsSnapshot<'a> { - pub async fn new(pool: &'a PgPool) -> StoreResult { - let mut tx = begin_snapshot(pool).await?; + pub async fn new(connection: &'a mut PgConnection) -> StoreResult { + let mut tx = begin_snapshot(connection).await?; // 时间跨度和分组选择性差异很大,通用计划会把长区间误估成短区间 // 仅当前只读事务使用实际参数规划,不改变连接池后续事务的计划策略 sqlx::query("set local plan_cache_mode = force_custom_plan") diff --git a/backend/crates/gateway-store/src/postgres/observability/analytics/worker.rs b/backend/crates/gateway-store/src/postgres/observability/analytics/worker.rs index f746a3553..83a6b2e6e 100644 --- a/backend/crates/gateway-store/src/postgres/observability/analytics/worker.rs +++ b/backend/crates/gateway-store/src/postgres/observability/analytics/worker.rs @@ -53,15 +53,18 @@ impl PgRequestAnalyticsRepository { pub async fn status(&self) -> StoreResult { 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); } diff --git a/backend/crates/gateway-store/src/postgres/observability/mod.rs b/backend/crates/gateway-store/src/postgres/observability/mod.rs index a071accd6..d3db2308e 100644 --- a/backend/crates/gateway-store/src/postgres/observability/mod.rs +++ b/backend/crates/gateway-store/src/postgres/observability/mod.rs @@ -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, diff --git a/backend/crates/gateway-store/src/postgres/observability/queries/ops.rs b/backend/crates/gateway-store/src/postgres/observability/queries/ops.rs index 4b36e7d9b..661b4a754 100644 --- a/backend/crates/gateway-store/src/postgres/observability/queries/ops.rs +++ b/backend/crates/gateway-store/src/postgres/observability/queries/ops.rs @@ -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 { 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 = diff --git a/backend/crates/gateway-store/src/postgres/observability/queries/usage.rs b/backend/crates/gateway-store/src/postgres/observability/queries/usage.rs index c80d87661..4c49d60be 100644 --- a/backend/crates/gateway-store/src/postgres/observability/queries/usage.rs +++ b/backend/crates/gateway-store/src/postgres/observability/queries/usage.rs @@ -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 { 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, ) -> StoreResult> { 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::::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 { 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, diff --git a/backend/crates/gateway-store/src/postgres/observability/query_budget.rs b/backend/crates/gateway-store/src/postgres/observability/query_budget.rs index 58373a6d3..d8ee898cd 100644 --- a/backend/crates/gateway-store/src/postgres/observability/query_budget.rs +++ b/backend/crates/gateway-store/src/postgres/observability/query_budget.rs @@ -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, + permit: Option, +} + +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 { @@ -43,18 +77,33 @@ impl ObservabilityQueryBudget { /// # Errors /// /// 预算在等待期限内没有空闲槽位、预算已关闭,或查询本身失败时返回错误 - pub async fn run(&self, operation: &'static str, query: F) -> StoreResult + pub async fn run( + &self, + pool: &PgPool, + operation: &'static str, + query: F, + ) -> StoreResult where - F: Future>, + F: FnOnce(ObservabilityConnection) -> Q, + Q: Future>, { - 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 { 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 a91ee7a2f..25b09138f 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 @@ -120,18 +120,22 @@ impl PgAdminAccountStore { .collect::>(); 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()); diff --git a/backend/crates/gateway-store/src/postgres/provider_accounts/quota_forecast.rs b/backend/crates/gateway-store/src/postgres/provider_accounts/quota_forecast.rs index a3fa6b23f..b8810e04c 100644 --- a/backend/crates/gateway-store/src/postgres/provider_accounts/quota_forecast.rs +++ b/backend/crates/gateway-store/src/postgres/provider_accounts/quota_forecast.rs @@ -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(); 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 a8551fb5b..26eaedaff 100644 --- a/backend/crates/gateway-store/src/postgres/provider_accounts/runtime.rs +++ b/backend/crates/gateway-store/src/postgres/provider_accounts/runtime.rs @@ -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> { 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() diff --git a/backend/crates/gateway-store/tests/postgres/mod.rs b/backend/crates/gateway-store/tests/postgres/mod.rs index f459c79d0..500050d03 100644 --- a/backend/crates/gateway-store/tests/postgres/mod.rs +++ b/backend/crates/gateway-store/tests/postgres/mod.rs @@ -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"); diff --git a/backend/crates/gateway-store/tests/postgres/query_budget.rs b/backend/crates/gateway-store/tests/postgres/query_budget.rs index 886d2775a..a2c98c2bf 100644 --- a/backend/crates/gateway-store/tests/postgres/query_budget.rs +++ b/backend/crates/gateway-store/tests/postgres/query_budget.rs @@ -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 { + 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::>(), - ) + .run(&pool, "never completes", |connection| async move { + let result = std::future::pending::>().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; } diff --git a/backend/crates/gateway-store/tests/postgres/usage_paging.rs b/backend/crates/gateway-store/tests/postgres/usage_paging.rs new file mode 100644 index 000000000..87073e93b --- /dev/null +++ b/backend/crates/gateway-store/tests/postgres/usage_paging.rs @@ -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::>(); + 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::>(); + let expected_page = expected + .iter() + .skip((page as usize - 1) * 7) + .take(7) + .map(String::as_str) + .collect::>(); + assert_eq!( + result.total, + expected.len() as u64, + "key={key:?}, page={page}" + ); + assert_eq!(ids, expected_page, "key={key:?}, page={page}"); + } + } + database.close().await; +} diff --git a/backend/migrations/.frozen-sha256 b/backend/migrations/.frozen-sha256 index 5280d9c9c..392662d06 100644 --- a/backend/migrations/.frozen-sha256 +++ b/backend/migrations/.frozen-sha256 @@ -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 diff --git a/backend/migrations/0029_usage_indexes_and_identity_snapshots.sql b/backend/migrations/0029_usage_indexes_and_identity_snapshots.sql new file mode 100644 index 000000000..a2a594853 --- /dev/null +++ b/backend/migrations/0029_usage_indexes_and_identity_snapshots.sql @@ -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 + ); diff --git a/docs/api.md b/docs/api.md index 45d3be36c..b133c64e6 100644 --- a/docs/api.md +++ b/docs/api.md @@ -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` diff --git a/docs/architecture.md b/docs/architecture.md index 69b4bcdaa..73c254cfc 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -853,6 +853,10 @@ OpenAI 订阅周期属于按需个人信息,不是额度事实。Admin 账号 ## 10. 观测与后台任务 +管理观测查询与统计 Worker 共用 PostgreSQL 连接预算,为请求执行保留连接容量。每项操作使用一个受预算约束的连接; +超时或调用方取消后,名额持续覆盖 SQLx 的协议收尾、事务回滚和连接归还,不能随查询 Future 提前释放。 +请求列表的精确总数与候选页共用数据库快照,按距首尾的较短方向读取,返回顺序和页码语义保持一致。 + ### 额度预测 账号容量预测属于 Admin 的只读派生规则,不参与 quota 权威状态、调度或金额结算。Store 通过专用采样端口 diff --git a/frontend/pnpm-lock.yaml b/frontend/pnpm-lock.yaml index 19c5d66bc..866a5b258 100644 --- a/frontend/pnpm-lock.yaml +++ b/frontend/pnpm-lock.yaml @@ -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 diff --git a/frontend/src/views/usage/composables/useUsageRecordsTable.ts b/frontend/src/views/usage/composables/useUsageRecordsTable.ts index cf004b809..97136a4f2 100644 --- a/frontend/src/views/usage/composables/useUsageRecordsTable.ts +++ b/frontend/src/views/usage/composables/useUsageRecordsTable.ts @@ -23,12 +23,10 @@ type LoadingAnalytics = Omit & { 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, background: boolean) { + async function loadUsageAnalytics(globalParams: ReturnType) { 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, diff --git a/frontend/src/views/usage/index.vue b/frontend/src/views/usage/index.vue index 5499066db..f33615c92 100644 --- a/frontend/src/views/usage/index.vue +++ b/frontend/src/views/usage/index.vue @@ -33,7 +33,6 @@ const { loading, error, analyticsLoading, - analyticsError, records, summary, insights, @@ -71,18 +70,13 @@ watch(timeRange, () => { - - + +