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

fix(observability): restore direct page navigation

Browse Source 
Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
Zyy
2026-08-31 04:09:31 +08:00
co-authored by Codex
parent 8cfd7f53be
commit c46910df22
21 changed files with 222 additions and 446 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/crates/gateway-admin/src/model/observability.rs
+8 -15
View File
@@ -38,13 +38,6 @@ impl TimeRange {
}
}
/// 观测记录的稳定键集游标。
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ObservabilityCursor {
pub observed_at: DateTime<Utc>,
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<ObservabilityCursor>,
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<ObservabilityCursor>,
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<UsageListRecord>,
pub total: Option<u64>,
pub next_cursor: Option<ObservabilityCursor>,
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<OpsError>,
pub total: Option<u64>,
pub next_cursor: Option<ObservabilityCursor>,
pub current_page: u32,
pub page_size: u16,
pub total: u64,
}
/// 仪表盘趋势指标。
backend/crates/gateway-admin/tests/use_case/observability.rs
+5 -5
View File
@@ -675,13 +675,14 @@ impl ObservabilityStore for FixtureObservabilityStore {
.clone())
}
async fn list_usage_records(&self, _: UsageQuery) -> AdminStoreResult<UsagePage> {
async fn list_usage_records(&self, query: UsageQuery) -> AdminStoreResult<UsagePage> {
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<Utc>) -> 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,
}
}
backend/crates/gateway-api/src/admin/observability/mod.rs
-1
View File
@@ -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};
backend/crates/gateway-api/src/admin/observability/presenter.rs
+10 -20
View File
@@ -515,18 +515,13 @@ pub(crate) fn usage_detail_view(detail: domain::UsageDetail) -> UsageRecordDetai
}
}
pub(crate) fn usage_page_view(
page: domain::UsagePage,
) -> Result<PageData<UsageListRecordView>, WireValidationError> {
Ok(PageData {
pub(crate) fn usage_page_view(page: domain::UsagePage) -> PageData<UsageListRecordView> {
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<PageData<OpsErrorView>, WireValidationError> {
Ok(PageData {
pub(crate) fn ops_page_view(page: domain::OpsErrorPage) -> PageData<OpsErrorView> {
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 {
backend/crates/gateway-api/src/admin/observability/query.rs
+20 -72
View File
@@ -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<u32>,
pub page_size: Option<u16>,
pub cursor: Option<String>,
pub kind: Option<String>,
pub outcome: Option<String>,
pub client_api_key_id: Option<String>,
@@ -52,18 +48,17 @@ pub struct UsageQuery {
}
impl UsageQuery {
/// 校验分页字段，不改变仓储层的分页 owner。
pub fn validate_page_size(&self) -> Result<u16, WireValidationError> {
/// 校验 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<u32>,
pub page_size: Option<u16>,
pub cursor: Option<String>,
pub kind: Option<String>,
pub client_api_key_id: Option<String>,
pub provider: Option<String>,
@@ -128,18 +123,17 @@ pub struct OpsQuery {
}
impl OpsQuery {
/// 校验分页字段。
pub fn validate_page_size(&self) -> Result<u16, WireValidationError> {
/// 校验 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<domain::UsageFilter, Wi
}
pub(crate) fn usage_command(query: &UsageQuery) -> Result<domain::UsageQuery, WireValidationError> {
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<domain::OpsErrorQuery, WireValidationError> {
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<domain::OpsErrorQuery, Wir
status_code,
search: non_empty(query.search.clone()),
},
cursor: decode_observability_cursor(query.cursor.as_deref())?,
current_page,
page_size: page_size_value,
include_total: query.cursor.is_none(),
})
}
pub(crate) fn decode_observability_cursor(
value: Option<&str>,
) -> Result<Option<domain::ObservabilityCursor>, 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<String, WireValidationError> {
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<String>) -> Option<String> {
value
.map(|value| value.trim().to_owned())
backend/crates/gateway-api/src/admin/observability/routes.rs
+2 -2
View File
@@ -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)))
}
backend/crates/gateway-api/src/admin/observability/wire.rs
+3 -10
View File
@@ -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<Utc>,
pub stable_id: String,
}
/// 观测列表响应数据。
#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct PageData<T> {
pub items: Vec<T>,
pub total: Option<u64>,
pub next_cursor: Option<String>,
pub current_page: u32,
pub page_size: u16,
pub total: u64,
}
/// Token 详情展示。
backend/crates/gateway-api/tests/admin/mod.rs
+6 -4
View File
@@ -858,9 +858,10 @@ impl ObservabilityStore for UnusedStore {
async fn list_usage_records(&self, query: UsageQuery) -> AdminStoreResult<UsagePage> {
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<OpsErrorPage> {
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,
})
}
}
backend/crates/gateway-api/tests/admin/observability/query.rs
+11 -7
View File
@@ -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::<UsageQuery>(json!({"cursor": "opaque"})).is_err());
}
#[test]
fn usage_query_should_reject_nonstandard_page_name() {
assert!(serde_json::from_value::<UsageQuery>(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]
backend/crates/gateway-api/tests/admin/observability/response.rs
+12 -13
View File
@@ -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"));
}
backend/crates/gateway-store/src/postgres/observability/admin_adapter.rs
+2 -3
View File
@@ -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 {
backend/crates/gateway-store/src/postgres/observability/mapping.rs
+6 -21
View File
@@ -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> {
ObservabilityCursor::new(cursor.observed_at, cursor.stable_id).map_err(observability_error)
}
pub(crate) fn store_usage_query(
query: admin_observability::UsageQuery,
) -> AdminStoreResult<UsageRecordQuery> {
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::<AdminStoreResult<_>>()?,
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<admin_observability::UsageListRecord> {
@@ -701,8 +685,9 @@ pub(crate) fn admin_ops_error_page(
) -> AdminStoreResult<admin_observability::OpsErrorPage> {
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),
})
}
backend/crates/gateway-store/src/postgres/observability/model.rs
+19 -23
View File
@@ -43,21 +43,17 @@ impl ObservabilityPageSize {
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ObservabilityCursor {
pub observed_at: DateTime<Utc>,
pub stable_id: String,
}
impl ObservabilityCursor {
pub fn new(observed_at: DateTime<Utc>, stable_id: impl Into<String>) -> StoreResult<Self> {
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<i64> {
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<ObservabilityCursor>,
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<ObservabilityCursor>,
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<UsageListRecord>,
pub total: Option<u64>,
pub next_cursor: Option<ObservabilityCursor>,
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<OpsErrorRecord>,
pub total: Option<u64>,
pub next_cursor: Option<ObservabilityCursor>,
pub current_page: u32,
pub page_size: u16,
pub total: u64,
}
#[async_trait]
backend/crates/gateway-store/src/postgres/observability/queries/ops.rs
+12 -58
View File
@@ -44,54 +44,32 @@ pub(crate) async fn list_ops_errors(
query: OpsErrorQuery,
) -> StoreResult<OpsErrorPage> {
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::<Postgres>::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::<StoreResult<Vec<_>>>()?;
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::<Postgres>::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::<i64>()
@@ -121,7 +99,6 @@ fn push_request_error_predicates(
statement: &mut QueryBuilder<Postgres>,
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<Postgres>,
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<Postgres>, column: &str, range: ObservabilityRange) {
@@ -262,18 +231,3 @@ fn push_prefix_search(statement: &mut QueryBuilder<Postgres>, columns: &[&str],
}
statement.push(")");
}
fn push_cursor(
statement: &mut QueryBuilder<Postgres>,
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(")");
}
}
backend/crates/gateway-store/src/postgres/observability/queries/usage.rs
+21 -31
View File
@@ -156,11 +156,22 @@ pub(crate) async fn list_usage_records(
query: UsageRecordQuery,
) -> StoreResult<UsageRecordPage> {
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<Vec<UsageListRecord>> {
query.filter.validate()?;
let offset = observability_page_offset(query.current_page, query.page_size)?;
let mut statement = QueryBuilder::<Postgres>::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::<StoreResult<Vec<_>>>()?;
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(
backend/crates/gateway-store/tests/postgres/observability.rs
+57 -37
View File
@@ -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()
backend/crates/gateway-store/tests/postgres/snapshots.rs
+3 -5
View File
@@ -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<Utc>, 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,
}
}
frontend/src/api/modules/usage.ts
+12 -84
View File
@@ -406,98 +406,26 @@ type OpsErrorPageQuery = UsagePageQuery & {
route?: string
}
interface CursorPageResponse<Item> {
items: Item[]
total: number | null
nextCursor: string | null
}
type CursorPageQuery<Query> = Query & {
pageSize: number
cursor?: string
}
interface UsageDetailQuery {
id: string
}
type UsageDiagnosticsQuery = UsageRangeQuery & { dimension: string }
export function createUsageRecordsPager() {
return createPageLoader<UsageListRecord, UsageRangeQuery>(data =>
request<CursorPageResponse<UsageListRecord>>({
url: '/api/admin/usage/records',
method: 'GET',
params: data,
}))
export function getUsageRecords(data: UsagePageQuery) {
return request<UsageRecordsResponse>({
url: '/api/admin/usage/records',
method: 'GET',
params: data,
})
}
export function createOpsErrorsPager() {
return createPageLoader<OpsError, Omit<OpsErrorPageQuery, keyof PageQuery>>(data =>
request<CursorPageResponse<OpsError>>({
url: '/api/admin/operations/errors',
method: 'GET',
params: data,
}))
}
function createPageLoader<Item, Query extends object>(
fetchPage: (query: CursorPageQuery<Query>) => Promise<CursorPageResponse<Item>>,
) {
const cursors = new Map<number, string | undefined>([[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<Item>
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<OpsErrorsResponse>({
url: '/api/admin/operations/errors',
method: 'GET',
params: data,
})
}
export function getUsageRecordDetail(data: UsageDetailQuery) {
frontend/src/composables/useStablePagedQuery.ts
+1 -1
View File
@@ -14,7 +14,7 @@ interface PageRequest {
pageSize: number
}
/** Element Plus 风格的页面分页状态；稳定深分页由 API pager 内部完成。 */
/** Element Plus 风格的服务端页码分页；一次 execute 只发起一次列表请求。 */
export function useStablePagedQuery<Result extends PageResult>(options: {
initialPageSize: number
load: (pagination: PageRequest) => Promise<Result>
frontend/src/views/usage/composables/useOpsErrorsTable.ts
+2 -4
View File
@@ -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<Ref<UsageTimeRangePa
const searchQuery = shallowRef('')
const failureClass = shallowRef('')
const route = shallowRef('')
const pager = createOpsErrorsPager()
const query = useStablePagedQuery({
initialPageSize: 10,
load: ({ currentPage, pageSize }) => pager.load({
load: ({ currentPage, pageSize }) => getOpsErrors({
currentPage,
pageSize,
search: searchQuery.value.trim() || undefined,
frontend/src/views/usage/composables/useUsageRecordsTable.ts
+10 -30
View File
@@ -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)