diff --git a/backend/crates/gateway-admin/src/model/observability.rs b/backend/crates/gateway-admin/src/model/observability.rs index a6bdc171c..74125a7e9 100644 --- a/backend/crates/gateway-admin/src/model/observability.rs +++ b/backend/crates/gateway-admin/src/model/observability.rs @@ -38,13 +38,6 @@ impl TimeRange { } } -/// 观测记录的稳定键集游标。 -#[derive(Debug, Clone, PartialEq, Eq)] -pub struct ObservabilityCursor { - pub observed_at: DateTime, - pub stable_id: String, -} - /// 请求结果状态。 #[derive(Debug, Clone, PartialEq, Eq)] pub enum RequestOutcome { @@ -130,9 +123,8 @@ pub struct UsageFilter { pub struct UsageQuery { pub range: TimeRange, pub filter: UsageFilter, - pub cursor: Option, + pub current_page: u32, pub page_size: PageSize, - pub include_total: bool, } /// 运维错误过滤条件。 @@ -158,9 +150,8 @@ pub struct OpsErrorFilter { pub struct OpsErrorQuery { pub range: TimeRange, pub filter: OpsErrorFilter, - pub cursor: Option, + pub current_page: u32, pub page_size: PageSize, - pub include_total: bool, } /// 用量诊断维度。 @@ -667,8 +658,9 @@ pub struct UsageRecord { #[derive(Debug, Clone, PartialEq, Eq)] pub struct UsagePage { pub items: Vec, - pub total: Option, - pub next_cursor: Option, + pub current_page: u32, + pub page_size: u16, + pub total: u64, } /// 请求中的一次上游尝试或运维事件。 @@ -795,8 +787,9 @@ pub struct OpsError { #[derive(Debug, Clone, PartialEq, Eq)] pub struct OpsErrorPage { pub items: Vec, - pub total: Option, - pub next_cursor: Option, + pub current_page: u32, + pub page_size: u16, + pub total: u64, } /// 仪表盘趋势指标。 diff --git a/backend/crates/gateway-admin/tests/use_case/observability.rs b/backend/crates/gateway-admin/tests/use_case/observability.rs index 6fb251e53..8f5b4aff7 100644 --- a/backend/crates/gateway-admin/tests/use_case/observability.rs +++ b/backend/crates/gateway-admin/tests/use_case/observability.rs @@ -675,13 +675,14 @@ impl ObservabilityStore for FixtureObservabilityStore { .clone()) } - async fn list_usage_records(&self, _: UsageQuery) -> AdminStoreResult { + async fn list_usage_records(&self, query: UsageQuery) -> AdminStoreResult { let items = self.usage_records.lock().expect("usage records").clone(); let total = u64::try_from(items.len()).unwrap_or(u64::MAX); Ok(UsagePage { items, - total: Some(total), - next_cursor: None, + current_page: query.current_page, + page_size: query.page_size.get(), + total, }) } @@ -783,9 +784,8 @@ fn usage_query(now: DateTime) -> UsageQuery { UsageQuery { range: observation_range(now), filter: UsageFilter::default(), - cursor: None, + current_page: 1, page_size: PageSize::new(50).expect("page size"), - include_total: true, } } diff --git a/backend/crates/gateway-api/src/admin/observability/mod.rs b/backend/crates/gateway-api/src/admin/observability/mod.rs index 6f8746ccc..158d26298 100644 --- a/backend/crates/gateway-api/src/admin/observability/mod.rs +++ b/backend/crates/gateway-api/src/admin/observability/mod.rs @@ -3,7 +3,6 @@ use std::collections::BTreeMap; use axum::{Router, extract::State, http::StatusCode, response::IntoResponse, routing::get}; -use base64::engine::general_purpose::URL_SAFE_NO_PAD; use chrono::{DateTime, Duration, Utc}; use gateway_admin::model::{PageSize as DomainPageSize, observability as domain}; use serde::{Deserialize, Serialize}; diff --git a/backend/crates/gateway-api/src/admin/observability/presenter.rs b/backend/crates/gateway-api/src/admin/observability/presenter.rs index b94496a27..f3c012e18 100644 --- a/backend/crates/gateway-api/src/admin/observability/presenter.rs +++ b/backend/crates/gateway-api/src/admin/observability/presenter.rs @@ -515,18 +515,13 @@ pub(crate) fn usage_detail_view(detail: domain::UsageDetail) -> UsageRecordDetai } } -pub(crate) fn usage_page_view( - page: domain::UsagePage, -) -> Result, WireValidationError> { - Ok(PageData { +pub(crate) fn usage_page_view(page: domain::UsagePage) -> PageData { + PageData { items: page.items.into_iter().map(usage_list_record_view).collect(), + current_page: page.current_page, + page_size: page.page_size, total: page.total, - next_cursor: page - .next_cursor - .as_ref() - .map(encode_observability_cursor) - .transpose()?, - }) + } } pub(crate) fn ops_error_view(error: domain::OpsError) -> OpsErrorView { @@ -566,18 +561,13 @@ pub(crate) fn ops_error_view(error: domain::OpsError) -> OpsErrorView { } } -pub(crate) fn ops_page_view( - page: domain::OpsErrorPage, -) -> Result, WireValidationError> { - Ok(PageData { +pub(crate) fn ops_page_view(page: domain::OpsErrorPage) -> PageData { + PageData { items: page.items.into_iter().map(ops_error_view).collect(), + current_page: page.current_page, + page_size: page.page_size, total: page.total, - next_cursor: page - .next_cursor - .as_ref() - .map(encode_observability_cursor) - .transpose()?, - }) + } } pub(crate) fn trend_point_view(point: domain::TrendPoint) -> TrendPointView { diff --git a/backend/crates/gateway-api/src/admin/observability/query.rs b/backend/crates/gateway-api/src/admin/observability/query.rs index aedf62905..07bb18aaa 100644 --- a/backend/crates/gateway-api/src/admin/observability/query.rs +++ b/backend/crates/gateway-api/src/admin/observability/query.rs @@ -2,14 +2,10 @@ use super::*; -use base64::Engine as _; - /// 观测列表默认页大小。 pub const DEFAULT_PAGE_SIZE: u16 = 50; /// 观测列表允许的最大页大小。 pub const MAX_PAGE_SIZE: u16 = 100; -/// 游标编码在 HTTP 层解码前允许的最大长度。 -pub const MAX_CURSOR_BYTES: usize = 1024; /// Dashboard 查询参数。 #[derive(Clone, Default, Deserialize)] @@ -31,8 +27,8 @@ impl DashboardQuery { #[derive(Clone, Default, Deserialize)] #[serde(rename_all = "camelCase", deny_unknown_fields)] pub struct UsageQuery { + pub current_page: Option, pub page_size: Option, - pub cursor: Option, pub kind: Option, pub outcome: Option, pub client_api_key_id: Option, @@ -52,18 +48,17 @@ pub struct UsageQuery { } impl UsageQuery { - /// 校验分页字段,不改变仓储层的分页 owner。 - pub fn validate_page_size(&self) -> Result { + /// 校验 Element Plus 风格的页码分页字段。 + pub fn validate_pagination(&self) -> Result<(u32, u16), WireValidationError> { + let current_page = self.current_page.unwrap_or(1); + if current_page == 0 { + return Err(WireValidationError::new("currentPage")); + } let page_size = self.page_size.unwrap_or(DEFAULT_PAGE_SIZE); if page_size == 0 || page_size > MAX_PAGE_SIZE { return Err(WireValidationError::new("pageSize")); } - Ok(page_size) - } - - /// 校验游标的 wire 边界;编码和排序语义由应用层负责。 - pub fn validate_cursor(&self) -> Result<(), WireValidationError> { - validate_cursor(self.cursor.as_deref()) + Ok((current_page, page_size)) } } @@ -105,8 +100,8 @@ impl DiagnosticsQuery { #[derive(Clone, Default, Deserialize)] #[serde(rename_all = "camelCase", deny_unknown_fields)] pub struct OpsQuery { + pub current_page: Option, pub page_size: Option, - pub cursor: Option, pub kind: Option, pub client_api_key_id: Option, pub provider: Option, @@ -128,18 +123,17 @@ pub struct OpsQuery { } impl OpsQuery { - /// 校验分页字段。 - pub fn validate_page_size(&self) -> Result { + /// 校验 Element Plus 风格的页码分页字段。 + pub fn validate_pagination(&self) -> Result<(u32, u16), WireValidationError> { + let current_page = self.current_page.unwrap_or(1); + if current_page == 0 { + return Err(WireValidationError::new("currentPage")); + } let page_size = self.page_size.unwrap_or(DEFAULT_PAGE_SIZE); if page_size == 0 || page_size > MAX_PAGE_SIZE { return Err(WireValidationError::new("pageSize")); } - Ok(page_size) - } - - /// 校验游标的 wire 边界。 - pub fn validate_cursor(&self) -> Result<(), WireValidationError> { - validate_cursor(self.cursor.as_deref()) + Ok((current_page, page_size)) } } @@ -248,13 +242,6 @@ pub(crate) fn require_text(value: &str, field: &'static str) -> Result<(), WireV } } -pub(crate) fn validate_cursor(value: Option<&str>) -> Result<(), WireValidationError> { - if value.is_some_and(|value| value.is_empty() || value.len() > MAX_CURSOR_BYTES) { - return Err(WireValidationError::new("cursor")); - } - Ok(()) -} - pub(crate) fn trimmed(value: Option<&str>) -> Option<&str> { value.map(str::trim).filter(|value| !value.is_empty()) } @@ -336,22 +323,19 @@ pub(crate) fn usage_filter(query: &UsageQuery) -> Result Result { - let page_size = query.validate_page_size()?; - query.validate_cursor()?; + let (current_page, page_size) = query.validate_pagination()?; let page_size_value = DomainPageSize::new(page_size).map_err(|_| WireValidationError::new("pageSize"))?; Ok(domain::UsageQuery { range: usage_range(query.start_time.as_deref(), query.end_time.as_deref())?, filter: usage_filter(query)?, - cursor: decode_observability_cursor(query.cursor.as_deref())?, + current_page, page_size: page_size_value, - include_total: query.cursor.is_none(), }) } pub(crate) fn ops_command(query: &OpsQuery) -> Result { - let page_size = query.validate_page_size()?; - query.validate_cursor()?; + let (current_page, page_size) = query.validate_pagination()?; let page_size_value = DomainPageSize::new(page_size).map_err(|_| WireValidationError::new("pageSize"))?; let status_code = parse_status( @@ -377,47 +361,11 @@ pub(crate) fn ops_command(query: &OpsQuery) -> Result, -) -> Result, WireValidationError> { - value - .map(|encoded| { - if encoded.is_empty() || encoded.len() > MAX_CURSOR_BYTES { - return Err(WireValidationError::new("cursor")); - } - let bytes = URL_SAFE_NO_PAD - .decode(encoded) - .map_err(|_| WireValidationError::new("cursor"))?; - let wire: CursorWire = - serde_json::from_slice(&bytes).map_err(|_| WireValidationError::new("cursor"))?; - if wire.stable_id.trim().is_empty() { - return Err(WireValidationError::new("cursor")); - } - Ok(domain::ObservabilityCursor { - observed_at: wire.observed_at, - stable_id: wire.stable_id, - }) - }) - .transpose() -} - -pub(crate) fn encode_observability_cursor( - cursor: &domain::ObservabilityCursor, -) -> Result { - let bytes = serde_json::to_vec(&CursorWire { - observed_at: cursor.observed_at, - stable_id: cursor.stable_id.clone(), - }) - .map_err(|_| WireValidationError::new("cursor"))?; - Ok(URL_SAFE_NO_PAD.encode(bytes)) -} - pub(crate) fn non_empty(value: Option) -> Option { value .map(|value| value.trim().to_owned()) diff --git a/backend/crates/gateway-api/src/admin/observability/routes.rs b/backend/crates/gateway-api/src/admin/observability/routes.rs index a5a4fbc2d..5b71383f6 100644 --- a/backend/crates/gateway-api/src/admin/observability/routes.rs +++ b/backend/crates/gateway-api/src/admin/observability/routes.rs @@ -91,7 +91,7 @@ where .usage_records(command) .await .map_err(map_service_error)?; - let data = usage_page_view(result).map_err(|_| AdminError::internal())?; + let data = usage_page_view(result); Ok(AdminResponse::new(StatusCode::OK, AdminEnvelope::ok(data))) } @@ -207,6 +207,6 @@ where .ops_errors(command) .await .map_err(map_service_error)?; - let data = ops_page_view(result).map_err(|_| AdminError::internal())?; + let data = ops_page_view(result); Ok(AdminResponse::new(StatusCode::OK, AdminEnvelope::ok(data))) } diff --git a/backend/crates/gateway-api/src/admin/observability/wire.rs b/backend/crates/gateway-api/src/admin/observability/wire.rs index 1b159ca63..b796df6c4 100644 --- a/backend/crates/gateway-api/src/admin/observability/wire.rs +++ b/backend/crates/gateway-api/src/admin/observability/wire.rs @@ -2,21 +2,14 @@ use super::*; -/// 观测列表游标的稳定 wire 形状。 -#[derive(Clone, Serialize, serde::Deserialize)] -#[serde(rename_all = "camelCase", deny_unknown_fields)] -pub struct CursorWire { - pub observed_at: DateTime, - pub stable_id: String, -} - /// 观测列表响应数据。 #[derive(Debug, Serialize)] #[serde(rename_all = "camelCase")] pub struct PageData { pub items: Vec, - pub total: Option, - pub next_cursor: Option, + pub current_page: u32, + pub page_size: u16, + pub total: u64, } /// Token 详情展示。 diff --git a/backend/crates/gateway-api/tests/admin/mod.rs b/backend/crates/gateway-api/tests/admin/mod.rs index b64ffe2d9..9779c3312 100644 --- a/backend/crates/gateway-api/tests/admin/mod.rs +++ b/backend/crates/gateway-api/tests/admin/mod.rs @@ -858,9 +858,10 @@ impl ObservabilityStore for UnusedStore { async fn list_usage_records(&self, query: UsageQuery) -> AdminStoreResult { let items = self.usage_records.lock().expect("usage records").clone(); Ok(UsagePage { - total: query.include_total.then_some(items.len() as u64), + current_page: query.current_page, + page_size: query.page_size.get(), + total: items.len() as u64, items, - next_cursor: None, }) } @@ -888,9 +889,10 @@ impl ObservabilityStore for UnusedStore { async fn list_ops_errors(&self, query: OpsErrorQuery) -> AdminStoreResult { let items = self.ops_errors.lock().expect("ops errors").clone(); Ok(OpsErrorPage { - total: query.include_total.then_some(items.len() as u64), + current_page: query.current_page, + page_size: query.page_size.get(), + total: items.len() as u64, items, - next_cursor: None, }) } } diff --git a/backend/crates/gateway-api/tests/admin/observability/query.rs b/backend/crates/gateway-api/tests/admin/observability/query.rs index 0e9e082f0..bc2a10050 100644 --- a/backend/crates/gateway-api/tests/admin/observability/query.rs +++ b/backend/crates/gateway-api/tests/admin/observability/query.rs @@ -19,14 +19,13 @@ fn dashboard_query_should_reject_unknown_trend_kind() { } #[test] -fn usage_query_should_bound_page_size_and_cursor() { +fn usage_query_should_accept_element_plus_pagination() { let query: UsageQuery = serde_json::from_value(json!({ - "pageSize": 100, - "cursor": "opaque" + "currentPage": 129, + "pageSize": 10 })) .unwrap(); - assert_eq!(query.validate_page_size().unwrap(), 100); - assert!(query.validate_cursor().is_ok()); + assert_eq!(query.validate_pagination().unwrap(), (129, 10)); } #[test] @@ -35,14 +34,19 @@ fn usage_query_should_reject_removed_total_toggle() { } #[test] -fn usage_query_should_reject_removed_offset_page_contract() { +fn usage_query_should_reject_removed_cursor_contract() { + assert!(serde_json::from_value::(json!({"cursor": "opaque"})).is_err()); +} + +#[test] +fn usage_query_should_reject_nonstandard_page_name() { assert!(serde_json::from_value::(json!({"page": 1})).is_err()); } #[test] fn ops_query_should_reject_page_size_above_terminal_limit() { let query: OpsQuery = serde_json::from_value(json!({"pageSize": 101})).unwrap(); - assert_eq!(query.validate_page_size().unwrap_err().field(), "pageSize"); + assert_eq!(query.validate_pagination().unwrap_err().field(), "pageSize"); } #[test] diff --git a/backend/crates/gateway-api/tests/admin/observability/response.rs b/backend/crates/gateway-api/tests/admin/observability/response.rs index 1e0377434..a3e020fd7 100644 --- a/backend/crates/gateway-api/tests/admin/observability/response.rs +++ b/backend/crates/gateway-api/tests/admin/observability/response.rs @@ -3,10 +3,10 @@ use chrono::{TimeZone, Utc}; use gateway_admin::model::observability::DesktopReleaseStatus; use gateway_api::admin::observability::{ - BillingView, CostCoverageView, CursorWire, DashboardAccountRequestBucketView, - DashboardAccountUsageView, DashboardDesktopReleaseStatusView, DashboardWireAttributeView, - DashboardWireProfileView, DashboardWireTargetView, PageData, TokenDetailsView, TrendData, - TrendKind, TrendPointView, TrendSummaryView, + BillingView, CostCoverageView, DashboardAccountRequestBucketView, DashboardAccountUsageView, + DashboardDesktopReleaseStatusView, DashboardWireAttributeView, DashboardWireProfileView, + DashboardWireTargetView, PageData, TokenDetailsView, TrendData, TrendKind, TrendPointView, + TrendSummaryView, }; use serde_json::json; @@ -14,12 +14,15 @@ use serde_json::json; fn usage_page_should_keep_terminal_camel_case_shape() { let data = PageData { items: vec![json!({"id": "request_1"})], - total: Some(1), - next_cursor: Some("cursor".to_owned()), + current_page: 129, + page_size: 10, + total: 1287, }; let value = serde_json::to_value(data).unwrap(); - assert_eq!(value["total"], 1); - assert_eq!(value["nextCursor"], "cursor"); + assert_eq!(value["currentPage"], 129); + assert_eq!(value["pageSize"], 10); + assert_eq!(value["total"], 1287); + assert!(value.get("nextCursor").is_none()); } #[test] @@ -154,11 +157,7 @@ fn sensitive_response_views_do_not_require_debug_or_add_secret_fields() { image_output_tokens_display: "-".to_owned(), total_tokens_display: "3".to_owned(), }; - let cursor = CursorWire { - observed_at: Utc.timestamp_opt(0, 0).single().unwrap(), - stable_id: "request_1".to_owned(), - }; - let value = serde_json::to_value((&coverage, &token_details, &cursor)).unwrap(); + let value = serde_json::to_value((&coverage, &token_details)).unwrap(); assert!(value.to_string().contains("known")); assert!(!value.to_string().contains("secret")); } 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 92e396fcf..6e740d8c1 100644 --- a/backend/crates/gateway-store/src/postgres/observability/admin_adapter.rs +++ b/backend/crates/gateway-store/src/postgres/observability/admin_adapter.rs @@ -125,9 +125,8 @@ impl ObservabilityRepository for PgObservabilityRepository { outcome: Some("succeeded".to_owned()), ..UsageRecordFilter::default() }, - cursor: None, + current_page: 1, page_size: ObservabilityPageSize::new(10)?, - include_total: false, }; // 每条 SQL 独立取一个全局观测槽位,避免整包预留造成队头阻塞。 let (totals, (provider_accounts, _)) = futures::try_join!( @@ -148,7 +147,7 @@ impl ObservabilityRepository for PgObservabilityRepository { ), self.query_budget .run("load dashboard recent requests", async { - Ok(list_usage_records(&self.pool, recent_query).await?.items) + list_usage_record_items(&self.pool, &recent_query).await }), )?; Ok(DashboardObservation { diff --git a/backend/crates/gateway-store/src/postgres/observability/mapping.rs b/backend/crates/gateway-store/src/postgres/observability/mapping.rs index ee42d675e..eb1010274 100644 --- a/backend/crates/gateway-store/src/postgres/observability/mapping.rs +++ b/backend/crates/gateway-store/src/postgres/observability/mapping.rs @@ -30,22 +30,15 @@ pub(crate) fn store_request_outcome(outcome: admin_observability::RequestOutcome outcome.as_str().to_owned() } -pub(crate) fn store_cursor( - cursor: admin_observability::ObservabilityCursor, -) -> AdminStoreResult { - ObservabilityCursor::new(cursor.observed_at, cursor.stable_id).map_err(observability_error) -} - pub(crate) fn store_usage_query( query: admin_observability::UsageQuery, ) -> AdminStoreResult { Ok(UsageRecordQuery { range: store_range(query.range)?, filter: store_usage_filter(query.filter), - cursor: query.cursor.map(store_cursor).transpose()?, + current_page: query.current_page, page_size: ObservabilityPageSize::new(query.page_size.get()) .map_err(observability_error)?, - include_total: query.include_total, }) } @@ -75,10 +68,9 @@ pub(crate) fn store_ops_error_query( Ok(OpsErrorQuery { range: store_range(query.range)?, filter: store_ops_error_filter(query.filter), - cursor: query.cursor.map(store_cursor).transpose()?, + current_page: query.current_page, page_size: ObservabilityPageSize::new(query.page_size.get()) .map_err(observability_error)?, - include_total: query.include_total, }) } @@ -409,20 +401,12 @@ pub(crate) fn admin_usage_page( .into_iter() .map(admin_usage_list_record) .collect::>()?, + current_page: page.current_page, + page_size: page.page_size, total: page.total, - next_cursor: page.next_cursor.map(admin_cursor), }) } -pub(crate) fn admin_cursor( - cursor: ObservabilityCursor, -) -> admin_observability::ObservabilityCursor { - admin_observability::ObservabilityCursor { - observed_at: cursor.observed_at, - stable_id: cursor.stable_id, - } -} - pub(crate) fn admin_usage_list_record( record: UsageListRecord, ) -> AdminStoreResult { @@ -701,8 +685,9 @@ pub(crate) fn admin_ops_error_page( ) -> AdminStoreResult { Ok(admin_observability::OpsErrorPage { items: page.items.into_iter().map(admin_ops_error).collect(), + current_page: page.current_page, + page_size: page.page_size, total: page.total, - next_cursor: page.next_cursor.map(admin_cursor), }) } diff --git a/backend/crates/gateway-store/src/postgres/observability/model.rs b/backend/crates/gateway-store/src/postgres/observability/model.rs index c09f5223d..a6a0796e0 100644 --- a/backend/crates/gateway-store/src/postgres/observability/model.rs +++ b/backend/crates/gateway-store/src/postgres/observability/model.rs @@ -43,21 +43,17 @@ impl ObservabilityPageSize { } } -#[derive(Debug, Clone, PartialEq, Eq)] -pub struct ObservabilityCursor { - pub observed_at: DateTime, - pub stable_id: String, -} - -impl ObservabilityCursor { - pub fn new(observed_at: DateTime, stable_id: impl Into) -> StoreResult { - let stable_id = stable_id.into(); - validate_text(&stable_id, MAX_FILTER_BYTES, "cursor ID")?; - Ok(Self { - observed_at, - stable_id, - }) - } +pub(crate) fn observability_page_offset( + current_page: u32, + page_size: ObservabilityPageSize, +) -> StoreResult { + let page_index = current_page + .checked_sub(1) + .ok_or_else(|| invalid("current page must be positive"))?; + u64::from(page_index) + .checked_mul(u64::from(page_size.get())) + .and_then(|offset| i64::try_from(offset).ok()) + .ok_or_else(|| invalid("page offset is too large")) } #[derive(Debug, Clone, Default, PartialEq, Eq)] @@ -119,9 +115,8 @@ impl UsageRecordFilter { pub struct UsageRecordQuery { pub range: ObservabilityRange, pub filter: UsageRecordFilter, - pub cursor: Option, + pub current_page: u32, pub page_size: ObservabilityPageSize, - pub include_total: bool, } #[derive(Debug, Clone, Default, PartialEq, Eq)] @@ -183,9 +178,8 @@ impl OpsErrorFilter { pub struct OpsErrorQuery { pub range: ObservabilityRange, pub filter: OpsErrorFilter, - pub cursor: Option, + pub current_page: u32, pub page_size: ObservabilityPageSize, - pub include_total: bool, } #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -609,8 +603,9 @@ pub struct UsageRecord { #[derive(Debug, Clone, PartialEq, Eq)] pub struct UsageRecordPage { pub items: Vec, - pub total: Option, - pub next_cursor: Option, + pub current_page: u32, + pub page_size: u16, + pub total: u64, } #[derive(Debug, Clone, PartialEq, Eq)] @@ -723,8 +718,9 @@ pub struct OpsErrorRecord { #[derive(Debug, Clone, PartialEq, Eq)] pub struct OpsErrorPage { pub items: Vec, - pub total: Option, - pub next_cursor: Option, + pub current_page: u32, + pub page_size: u16, + pub total: u64, } #[async_trait] 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 7a4ef1564..b23be3531 100644 --- a/backend/crates/gateway-store/src/postgres/observability/queries/ops.rs +++ b/backend/crates/gateway-store/src/postgres/observability/queries/ops.rs @@ -44,54 +44,32 @@ pub(crate) async fn list_ops_errors( query: OpsErrorQuery, ) -> StoreResult { query.filter.validate()?; - let total = if query.include_total { - Some(count_ops_errors(pool, query.range, &query.filter).await?) - } else { - None - }; + let total = count_ops_errors(pool, query.range, &query.filter).await?; + let offset = observability_page_offset(query.current_page, query.page_size)?; let mut statement = QueryBuilder::::new("select * from ("); statement.push(REQUEST_ERROR_SELECT); - push_request_error_predicates( - &mut statement, - query.range, - &query.filter, - query.cursor.as_ref(), - ); + push_request_error_predicates(&mut statement, query.range, &query.filter); statement.push(" union all "); statement.push(OPS_EVENT_SELECT); - push_ops_event_predicates( - &mut statement, - query.range, - &query.filter, - query.cursor.as_ref(), - ); + push_ops_event_predicates(&mut statement, query.range, &query.filter); statement.push(") e order by occurred_at desc, stable_sort_id desc limit "); - statement.push_bind(i64::from(query.page_size.get()) + 1); + statement.push_bind(i64::from(query.page_size.get())); + statement.push(" offset "); + statement.push_bind(offset); let rows = statement .build() .fetch_all(pool) .await .map_err(|_| postgres_unavailable("list ops errors"))?; - let mut items = rows + let items = rows .iter() .map(ops_error_from_row) .collect::>>()?; - let has_more = items.len() > usize::from(query.page_size.get()); - if has_more { - items.pop(); - } - let next_cursor = if has_more { - items - .last() - .map(|item| ObservabilityCursor::new(item.occurred_at, item.stable_sort_id.clone())) - .transpose()? - } else { - None - }; Ok(OpsErrorPage { items, + current_page: query.current_page, + page_size: query.page_size.get(), total, - next_cursor, }) } @@ -103,11 +81,11 @@ pub(crate) async fn count_ops_errors( let mut statement = QueryBuilder::::new( "select coalesce(sum(source_count), 0)::bigint from (select count(*)::bigint as source_count from model_requests mr where mr.outcome = 'failed'", ); - push_request_error_predicates(&mut statement, range, filter, None); + push_request_error_predicates(&mut statement, range, filter); statement.push( " union all select count(*)::bigint as source_count from ops_events oe left join model_requests mr on mr.id = oe.model_request_id where true", ); - push_ops_event_predicates(&mut statement, range, filter, None); + push_ops_event_predicates(&mut statement, range, filter); statement.push(") counts"); let total = statement .build_query_scalar::() @@ -121,7 +99,6 @@ fn push_request_error_predicates( statement: &mut QueryBuilder, range: ObservabilityRange, filter: &OpsErrorFilter, - cursor: Option<&ObservabilityCursor>, ) { push_range(statement, "mr.completed_at", range); for (column, value) in [ @@ -162,19 +139,12 @@ fn push_request_error_predicates( search, ); } - push_cursor( - statement, - "mr.completed_at", - "'model_request:' || mr.id", - cursor, - ); } fn push_ops_event_predicates( statement: &mut QueryBuilder, range: ObservabilityRange, filter: &OpsErrorFilter, - cursor: Option<&ObservabilityCursor>, ) { push_range(statement, "oe.created_at", range); for (column, value) in [ @@ -217,7 +187,6 @@ fn push_ops_event_predicates( search, ); } - push_cursor(statement, "oe.created_at", "'ops_event:' || oe.id", cursor); } fn push_range(statement: &mut QueryBuilder, column: &str, range: ObservabilityRange) { @@ -262,18 +231,3 @@ fn push_prefix_search(statement: &mut QueryBuilder, columns: &[&str], } statement.push(")"); } - -fn push_cursor( - statement: &mut QueryBuilder, - occurred_at: &str, - stable_sort_id: &str, - cursor: Option<&ObservabilityCursor>, -) { - if let Some(cursor) = cursor { - statement.push(format!(" and ({occurred_at}, {stable_sort_id}) < (")); - statement.push_bind(cursor.observed_at); - statement.push(", "); - statement.push_bind(cursor.stable_id.clone()); - statement.push(")"); - } -} 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 020b03815..34e6422d6 100644 --- a/backend/crates/gateway-store/src/postgres/observability/queries/usage.rs +++ b/backend/crates/gateway-store/src/postgres/observability/queries/usage.rs @@ -156,11 +156,22 @@ pub(crate) async fn list_usage_records( query: UsageRecordQuery, ) -> StoreResult { query.filter.validate()?; - let total = if query.include_total { - Some(count_usage_records(pool, query.range, &query.filter).await?) - } else { - None - }; + let total = count_usage_records(pool, query.range, &query.filter).await?; + let items = list_usage_record_items(pool, &query).await?; + Ok(UsageRecordPage { + items, + current_page: query.current_page, + page_size: query.page_size.get(), + total, + }) +} + +pub(crate) async fn list_usage_record_items( + pool: &PgPool, + query: &UsageRecordQuery, +) -> StoreResult> { + query.filter.validate()?; + let offset = observability_page_offset(query.current_page, query.page_size)?; let mut statement = QueryBuilder::::new(USAGE_LIST_RECORD_SELECT); statement.push(" where mr.started_at >= "); statement.push_bind(query.range.start); @@ -168,41 +179,20 @@ pub(crate) async fn list_usage_records( statement.push_bind(query.range.end); push_completed_usage_fact_filter(&mut statement, "mr"); push_usage_filter(&mut statement, &query.filter, "mr"); - if let Some(cursor) = &query.cursor { - statement.push(" and (mr.started_at, mr.id) < ("); - statement.push_bind(cursor.observed_at); - statement.push(", "); - statement.push_bind(cursor.stable_id.clone()); - statement.push(")"); - } statement.push(" order by mr.started_at desc, mr.id desc limit "); - statement.push_bind(i64::from(query.page_size.get()) + 1); + statement.push_bind(i64::from(query.page_size.get())); + statement.push(" offset "); + statement.push_bind(offset); let rows = statement .build() .fetch_all(pool) .await .map_err(|_| postgres_unavailable("list usage records"))?; - let mut items = rows + let items = rows .iter() .map(usage_list_record_from_row) .collect::>>()?; - let has_more = items.len() > usize::from(query.page_size.get()); - if has_more { - items.pop(); - } - let next_cursor = if has_more { - items - .last() - .map(|item| ObservabilityCursor::new(item.started_at, item.id.clone())) - .transpose()? - } else { - None - }; - Ok(UsageRecordPage { - items, - total, - next_cursor, - }) + Ok(items) } pub(crate) async fn count_usage_records( diff --git a/backend/crates/gateway-store/tests/postgres/observability.rs b/backend/crates/gateway-store/tests/postgres/observability.rs index 94b57dd6a..b5c0d4735 100644 --- a/backend/crates/gateway-store/tests/postgres/observability.rs +++ b/backend/crates/gateway-store/tests/postgres/observability.rs @@ -106,13 +106,12 @@ async fn observability_preserves_and_filters_opaque_response_ids() { response_id: Some(response_id.clone()), ..UsageRecordFilter::default() }, - cursor: None, + current_page: 1, page_size: ObservabilityPageSize::new(10).expect("page size"), - include_total: true, }) .await .expect("filter opaque response ID"); - assert_eq!(records.total, Some(1)); + assert_eq!(records.total, 1); assert_eq!(records.items[0].id, "req_observe_success"); let detail = repository .usage_record_detail("req_observe_success") @@ -131,8 +130,8 @@ async fn observability_preserves_and_filters_opaque_response_ids() { } #[tokio::test] -async fn usage_page_should_skip_exact_count_when_total_is_not_requested() { - let Some(database) = TestDatabase::create("usage_page_without_total").await else { +async fn usage_page_should_always_return_total() { + let Some(database) = TestDatabase::create("usage_page_with_total").await else { return; }; let now = Utc::now(); @@ -146,14 +145,15 @@ async fn usage_page_should_skip_exact_count_when_total_is_not_requested() { .list_usage_records(UsageRecordQuery { range, filter: UsageRecordFilter::default(), - cursor: None, + current_page: 1, page_size: ObservabilityPageSize::new(10).expect("page size"), - include_total: false, }) .await - .expect("usage page without total"); + .expect("usage page with total"); - assert_eq!(page.total, None); + assert_eq!(page.total, 1); + assert_eq!(page.current_page, 1); + assert_eq!(page.page_size, 10); database.close().await; } @@ -176,14 +176,13 @@ async fn usage_search_should_match_literal_prefix_instead_of_substring() { search: Some("observe_success".to_owned()), ..UsageRecordFilter::default() }, - cursor: None, + current_page: 1, page_size: ObservabilityPageSize::new(10).expect("page size"), - include_total: true, }) .await .expect("usage substring search"); - assert_eq!(page.total, Some(0)); + assert_eq!(page.total, 0); database.close().await; } @@ -206,14 +205,13 @@ async fn ops_search_should_treat_sql_wildcards_as_literals() { search: Some("req%".to_owned()), ..OpsErrorFilter::default() }, - cursor: None, + current_page: 1, page_size: ObservabilityPageSize::new(10).expect("page size"), - include_total: true, }) .await .expect("ops literal wildcard search"); - assert_eq!(page.total, Some(0)); + assert_eq!(page.total, 0); database.close().await; } @@ -625,20 +623,34 @@ async fn admin_observability_adapter_preserves_utc_queries_metrics_costs_and_det .list_usage_records(admin_observability::UsageQuery { range, filter: admin_observability::UsageFilter::default(), - cursor: None, + current_page: 1, page_size: PageSize::new(1).expect("page size"), - include_total: true, }) .await .expect("first usage page"); - assert_eq!(first_page.total, Some(1)); + assert_eq!(first_page.total, 1); assert_eq!(first_page.items.len(), 1); - assert!(first_page.next_cursor.is_none()); + assert_eq!(first_page.current_page, 1); + assert_eq!(first_page.page_size, 1); assert_eq!( first_page.items[0].service_tier.as_deref(), Some("priority") ); + let deep_page = store + .list_usage_records(admin_observability::UsageQuery { + range, + filter: admin_observability::UsageFilter::default(), + current_page: 129, + page_size: PageSize::new(1).expect("page size"), + }) + .await + .expect("direct deep usage page"); + assert_eq!(deep_page.current_page, 129); + assert_eq!(deep_page.page_size, 1); + assert_eq!(deep_page.total, 1); + assert!(deep_page.items.is_empty()); + let filtered = store .list_usage_records(admin_observability::UsageQuery { range, @@ -657,13 +669,12 @@ async fn admin_observability_adapter_preserves_utc_queries_metrics_costs_and_det upstream_request_id: Some("upstream_req_success".to_owned()), search: Some("req_observe_success".to_owned()), }, - cursor: None, + current_page: 1, page_size: PageSize::new(10).expect("page size"), - include_total: true, }) .await .expect("fully filtered usage page"); - assert_eq!(filtered.total, Some(1)); + assert_eq!(filtered.total, 1); assert_eq!(filtered.items[0].id, "req_observe_success"); assert_eq!(filtered.items[0].service_tier.as_deref(), Some("priority")); @@ -677,13 +688,12 @@ async fn admin_observability_adapter_preserves_utc_queries_metrics_costs_and_det ), ..admin_observability::UsageFilter::default() }, - cursor: None, + current_page: 1, page_size: PageSize::new(10).expect("page size"), - include_total: true, }) .await .expect("other outcome filter should reach PostgreSQL"); - assert_eq!(other_outcome.total, Some(0)); + assert_eq!(other_outcome.total, 0); assert!(other_outcome.items.is_empty()); let detail = store @@ -757,15 +767,28 @@ async fn admin_observability_adapter_preserves_utc_queries_metrics_costs_and_det search: Some("req_observe_failed".to_owned()), ..admin_observability::OpsErrorFilter::default() }, - cursor: None, + current_page: 1, page_size: PageSize::new(10).expect("page size"), - include_total: true, }) .await .expect("admin ops errors"); - assert_eq!(errors.total, Some(2)); + assert_eq!(errors.total, 2); assert!(errors.items.iter().all(|item| item.occurred_at <= now)); + let deep_errors = store + .list_ops_errors(admin_observability::OpsErrorQuery { + range, + filter: admin_observability::OpsErrorFilter::default(), + current_page: 129, + page_size: PageSize::new(1).expect("page size"), + }) + .await + .expect("direct deep ops page"); + assert_eq!(deep_errors.current_page, 129); + assert_eq!(deep_errors.page_size, 1); + assert_eq!(deep_errors.total, 2); + assert!(deep_errors.items.is_empty()); + for filter in [ admin_observability::OpsErrorFilter { client_api_key_ref: Some("missing-key".to_owned()), @@ -796,13 +819,12 @@ async fn admin_observability_adapter_preserves_utc_queries_metrics_costs_and_det .list_ops_errors(admin_observability::OpsErrorQuery { range, filter, - cursor: None, + current_page: 1, page_size: PageSize::new(10).expect("page size"), - include_total: true, }) .await .expect("fully forwarded ops filter"); - assert_eq!(page.total, Some(0)); + assert_eq!(page.total, 0); } database.close().await; @@ -1052,13 +1074,12 @@ async fn observability_queries_preserve_request_account_cost_and_diagnostic_fact provider_account_ref: Some("acct_observe".to_owned()), ..UsageRecordFilter::default() }, - cursor: None, + current_page: 1, page_size: ObservabilityPageSize::new(10).expect("page size"), - include_total: true, }) .await .expect("usage records"); - assert_eq!(usage_page.total, Some(1)); + assert_eq!(usage_page.total, 1); let successful_image = usage_page .items .iter() @@ -1157,13 +1178,12 @@ async fn observability_queries_preserve_request_account_cost_and_diagnostic_fact .list_ops_errors(OpsErrorQuery { range, filter: OpsErrorFilter::default(), - cursor: None, + current_page: 1, page_size: ObservabilityPageSize::new(10).expect("page size"), - include_total: true, }) .await .expect("ops errors"); - assert_eq!(errors.total, Some(2)); + assert_eq!(errors.total, 2); let request_error = errors .items .iter() diff --git a/backend/crates/gateway-store/tests/postgres/snapshots.rs b/backend/crates/gateway-store/tests/postgres/snapshots.rs index be81fc451..0999cb5df 100644 --- a/backend/crates/gateway-store/tests/postgres/snapshots.rs +++ b/backend/crates/gateway-store/tests/postgres/snapshots.rs @@ -60,7 +60,7 @@ async fn completed_usage_projections_should_accept_statusless_websocket_but_reje assert_eq!( (page.total, request_ids), - (Some(1), vec!["req_statusless_websocket"]) + (1, vec!["req_statusless_websocket"]) ); let overview = repository @@ -399,9 +399,8 @@ async fn ops_errors_should_keep_request_and_event_snapshots_after_account_deleti .list_ops_errors(OpsErrorQuery { range: range_around(started_at), filter: OpsErrorFilter::default(), - cursor: None, + current_page: 1, page_size: ObservabilityPageSize::new(10).expect("page size"), - include_total: true, }) .await .expect("list ops errors after deletion"); @@ -886,9 +885,8 @@ fn usage_query(started_at: DateTime, filter: UsageRecordFilter) -> UsageRec UsageRecordQuery { range: range_around(started_at), filter, - cursor: None, + current_page: 1, page_size: ObservabilityPageSize::new(10).expect("page size"), - include_total: true, } } diff --git a/frontend/src/api/modules/usage.ts b/frontend/src/api/modules/usage.ts index 11800b913..84bc2766b 100644 --- a/frontend/src/api/modules/usage.ts +++ b/frontend/src/api/modules/usage.ts @@ -406,98 +406,26 @@ type OpsErrorPageQuery = UsagePageQuery & { route?: string } -interface CursorPageResponse { - items: Item[] - total: number | null - nextCursor: string | null -} - -type CursorPageQuery = Query & { - pageSize: number - cursor?: string -} - interface UsageDetailQuery { id: string } type UsageDiagnosticsQuery = UsageRangeQuery & { dimension: string } -export function createUsageRecordsPager() { - return createPageLoader(data => - request>({ - url: '/api/admin/usage/records', - method: 'GET', - params: data, - })) +export function getUsageRecords(data: UsagePageQuery) { + return request({ + url: '/api/admin/usage/records', + method: 'GET', + params: data, + }) } -export function createOpsErrorsPager() { - return createPageLoader>(data => - request>({ - url: '/api/admin/operations/errors', - method: 'GET', - params: data, - })) -} - -function createPageLoader( - fetchPage: (query: CursorPageQuery) => Promise>, -) { - const cursors = new Map([[1, undefined]]) - let queryKey = '' - let total = 0 - - function reset(nextQueryKey: string) { - cursors.clear() - cursors.set(1, undefined) - queryKey = nextQueryKey - total = 0 - } - - async function load(data: Query & PageQuery) { - const { currentPage, pageSize, ...query } = data - const nextQueryKey = JSON.stringify([pageSize, query]) - if (queryKey !== nextQueryKey) - reset(nextQueryKey) - - const requestedPage = Math.max(1, Math.trunc(currentPage)) - let page = [...cursors.keys()].reduce( - (known, candidate) => candidate <= requestedPage && candidate > known ? candidate : known, - 1, - ) - let result: CursorPageResponse - - while (page < requestedPage) { - result = await fetchPage({ - ...(query as Query), - pageSize, - cursor: cursors.get(page), - }) - if (result.total != null) - total = result.total - if (!result.nextCursor) - return { items: result.items, currentPage: page, pageSize, total } - cursors.set(page + 1, result.nextCursor) - page += 1 - } - - result = await fetchPage({ - ...(query as Query), - pageSize, - cursor: cursors.get(page), - }) - if (result.total != null) - total = result.total - if (result.nextCursor) - cursors.set(page + 1, result.nextCursor) - else - cursors.delete(page + 1) - - return { items: result.items, currentPage: page, pageSize, total } - } - - return { load } +export function getOpsErrors(data: OpsErrorPageQuery) { + return request({ + url: '/api/admin/operations/errors', + method: 'GET', + params: data, + }) } export function getUsageRecordDetail(data: UsageDetailQuery) { diff --git a/frontend/src/composables/useStablePagedQuery.ts b/frontend/src/composables/useStablePagedQuery.ts index 2e19d9cb5..c626af521 100644 --- a/frontend/src/composables/useStablePagedQuery.ts +++ b/frontend/src/composables/useStablePagedQuery.ts @@ -14,7 +14,7 @@ interface PageRequest { pageSize: number } -/** Element Plus 风格的页面分页状态;稳定深分页由 API pager 内部完成。 */ +/** Element Plus 风格的服务端页码分页;一次 execute 只发起一次列表请求。 */ export function useStablePagedQuery(options: { initialPageSize: number load: (pagination: PageRequest) => Promise diff --git a/frontend/src/views/usage/composables/useOpsErrorsTable.ts b/frontend/src/views/usage/composables/useOpsErrorsTable.ts index 25d742f18..be26287a5 100644 --- a/frontend/src/views/usage/composables/useOpsErrorsTable.ts +++ b/frontend/src/views/usage/composables/useOpsErrorsTable.ts @@ -3,7 +3,7 @@ import type { UsageTimeRangeParams } from './useUsageTimeRange' import { watchDebounced } from '@vueuse/core' import { computed, onMounted, shallowRef, watch } from 'vue' -import { createOpsErrorsPager } from '@/api' +import { getOpsErrors } from '@/api' import { toast } from '@/components/base/BaseToast' import { useStablePagedQuery } from '@/composables/useStablePagedQuery' import { errorMessage, withMinimumDuration } from '@/utils/async' @@ -13,11 +13,9 @@ export function useOpsErrorsTable(timeRangeParams: Readonly pager.load({ + load: ({ currentPage, pageSize }) => getOpsErrors({ currentPage, pageSize, search: searchQuery.value.trim() || undefined, diff --git a/frontend/src/views/usage/composables/useUsageRecordsTable.ts b/frontend/src/views/usage/composables/useUsageRecordsTable.ts index d8228a737..86dbcb5bf 100644 --- a/frontend/src/views/usage/composables/useUsageRecordsTable.ts +++ b/frontend/src/views/usage/composables/useUsageRecordsTable.ts @@ -5,9 +5,9 @@ import { watchDebounced } from '@vueuse/core' import { computed, onMounted, onScopeDispose, shallowRef, watch } from 'vue' import { - createUsageRecordsPager, getUsageRecordInsightsDiagnostics, getUsageRecordInsightsOverview, + getUsageRecords, getUsageRecordSummary, } from '@/api' import { toast } from '@/components/base/BaseToast' @@ -26,13 +26,6 @@ interface UsageLoadOptions { background?: boolean } -interface UsageTableParams { - startTime: string - endTime: string - provider?: string - search?: string -} - export function useUsageRecordsTable(options: UseUsageRecordsTableOptions) { const loading = shallowRef(true) const analyticsLoading = shallowRef(true) @@ -49,7 +42,6 @@ export function useUsageRecordsTable(options: UseUsageRecordsTableOptions) { }) const refreshingList = shallowRef(false) const diagnosticDimension = shallowRef('model') - const pager = createUsageRecordsPager() let loadRequestId = 0 let diagnosticRequestId = 0 const scopedParams = () => ({ @@ -70,21 +62,6 @@ export function useUsageRecordsTable(options: UseUsageRecordsTableOptions) { totalRecords.value = 0 } - async function loadTablePage( - params: UsageTableParams, - targetPage: number, - requestId: number, - ) { - const result = await pager.load({ - currentPage: targetPage, - pageSize: pageSize.value, - ...params, - }) - if (requestId !== loadRequestId) - return null - return { result, page: result.currentPage } - } - async function loadUsageRecords(loadOptions: UsageLoadOptions = {}) { const { scope = 'all', background = false } = loadOptions const requestId = ++loadRequestId @@ -114,8 +91,11 @@ export function useUsageRecordsTable(options: UseUsageRecordsTableOptions) { ...tableTimeRangeParams.value, ...filterParams(), } - const requestedPage = currentPage.value - const resultPromise = loadTablePage(tableParams, requestedPage, requestId) + const resultPromise = getUsageRecords({ + currentPage: currentPage.value, + pageSize: pageSize.value, + ...tableParams, + }) const analyticsPromise = scope === 'all' ? loadUsageAnalytics(globalParams) @@ -123,10 +103,9 @@ export function useUsageRecordsTable(options: UseUsageRecordsTableOptions) { summary: summary.value, insights: insights.value, }) - const [tableResult, nextAnalytics] = await Promise.all([resultPromise, analyticsPromise]) - if (requestId !== loadRequestId || tableResult == null) + const [result, nextAnalytics] = await Promise.all([resultPromise, analyticsPromise]) + if (requestId !== loadRequestId) return - const { result, page: resultPage } = tableResult records.value = result.items summary.value = nextAnalytics.summary @@ -137,8 +116,9 @@ export function useUsageRecordsTable(options: UseUsageRecordsTableOptions) { ? nextAnalytics.insights.diagnostics : insights.value.diagnostics, } + pageSize.value = result.pageSize totalRecords.value = result.total - currentPage.value = resultPage + currentPage.value = result.currentPage } catch (error: unknown) { if (requestId !== loadRequestId)