From 24ff339c976d7ecc8e0bfa96d39fbffce24e99a5 Mon Sep 17 00:00:00 2001 From: zyycn Date: Mon, 27 Jul 2026 02:26:13 +0800 Subject: [PATCH] fix(admin): tolerate bad records in usage billing and stop showing invalid accounts as active MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit One record with an unparseable provider_kind no longer makes the whole usage list return 500; the failing record is skipped and the list still renders. An account whose credential availability is Invalid — excluded from scheduling — was displayed as Active in the directory; it now derives the Expired status so it reads as unavailable, consistent with how a dead credential surfaces to the operator. Co-authored-by: Claude --- .../gateway-admin/src/use_case/accounts.rs | 3 +- .../src/use_case/observability.rs | 25 ++- .../tests/use_case/observability.rs | 147 +++++++++++++++++- .../src/postgres/provider_accounts.rs | 1 + .../tests/postgres/provider_accounts.rs | 2 +- 5 files changed, 156 insertions(+), 22 deletions(-) diff --git a/backend/crates/gateway-admin/src/use_case/accounts.rs b/backend/crates/gateway-admin/src/use_case/accounts.rs index c8a2be3ed..2477bbc53 100644 --- a/backend/crates/gateway-admin/src/use_case/accounts.rs +++ b/backend/crates/gateway-admin/src/use_case/accounts.rs @@ -474,7 +474,8 @@ fn account_status(account: &AccountRecord, now: chrono::DateTime) -> Accoun crate::model::accounts::AccountAvailability::QuotaExhausted => { AccountStatus::QuotaExhausted } - crate::model::accounts::AccountAvailability::Expired => AccountStatus::Expired, + crate::model::accounts::AccountAvailability::Expired + | crate::model::accounts::AccountAvailability::Invalid => AccountStatus::Expired, _ if account .access_token_expires_at .is_some_and(|expires_at| expires_at <= now) => diff --git a/backend/crates/gateway-admin/src/use_case/observability.rs b/backend/crates/gateway-admin/src/use_case/observability.rs index e6571ee28..7fc9812f0 100644 --- a/backend/crates/gateway-admin/src/use_case/observability.rs +++ b/backend/crates/gateway-admin/src/use_case/observability.rs @@ -104,7 +104,7 @@ impl ObservabilityService for DefaultObservabilityService { self.store.dashboard_runtime_slots(range.end), ) .map_err(|error| map_store_error(error, "dashboard"))?; - self.enrich_billing(&mut observation.recent_requests)?; + self.enrich_billing(&mut observation.recent_requests); self.enrich_dashboard_quotas(&mut observation.account_usage) .await; let today_start = china_day_start(observation.range.end); @@ -176,7 +176,7 @@ impl ObservabilityService for DefaultObservabilityService { .list_usage_records(query) .await .map_err(|error| map_store_error(error, "usage records"))?; - self.enrich_billing(&mut page.items)?; + self.enrich_billing(&mut page.items); Ok(page) } @@ -189,7 +189,7 @@ impl ObservabilityService for DefaultObservabilityService { .usage_record_detail(request_id) .await .map_err(|error| map_store_error(error, "usage record"))?; - self.enrich_billing(std::slice::from_mut(&mut detail.request))?; + self.enrich_billing(std::slice::from_mut(&mut detail.request)); Ok(detail) } @@ -491,10 +491,9 @@ impl DefaultObservabilityService { } } - fn enrich_billing( - &self, - records: &mut [crate::model::observability::UsageRecord], - ) -> Result<(), AdminError> { + /// 逐条尽力把 calculated 总额升级为完整分解；单条脏数据（非法 Provider kind + /// 或费用规则失败）只保留该条已存的总额，不影响整页返回。 + fn enrich_billing(&self, records: &mut [crate::model::observability::UsageRecord]) { for record in records { let Some(UsageBilling::Total { source, total }) = record.billing.as_ref() else { continue; @@ -508,8 +507,9 @@ impl DefaultObservabilityService { ) else { continue; }; - let provider_kind = ProviderKind::new(provider.to_owned()) - .map_err(|_| AdminError::internal("Stored Provider kind is invalid"))?; + let Ok(provider_kind) = ProviderKind::new(provider.to_owned()) else { + continue; + }; let input = ProviderBillingInput { upstream_model_id: upstream_model_id.clone(), input_tokens: record.input_tokens, @@ -518,15 +518,10 @@ impl DefaultObservabilityService { cache_write_tokens: record.cache_write_tokens, total: total.clone(), }; - if let Some(breakdown) = self - .providers - .calculated_billing(&provider_kind, &input) - .map_err(|error| map_provider_error(error, "usage billing"))? - { + if let Ok(Some(breakdown)) = self.providers.calculated_billing(&provider_kind, &input) { record.billing = Some(UsageBilling::Calculated(Box::new(breakdown))); } } - Ok(()) } } diff --git a/backend/crates/gateway-admin/tests/use_case/observability.rs b/backend/crates/gateway-admin/tests/use_case/observability.rs index 6c79a2dcf..2f8230560 100644 --- a/backend/crates/gateway-admin/tests/use_case/observability.rs +++ b/backend/crates/gateway-admin/tests/use_case/observability.rs @@ -1,4 +1,5 @@ use std::{ + num::NonZeroU32, str::FromStr as _, sync::{Arc, Mutex}, }; @@ -9,13 +10,14 @@ use chrono::{DateTime, Duration, Timelike as _, Utc}; use gateway_admin::{ AdminServices, model::{ - MutationContext, Revision, + MutationContext, PageSize, Revision, observability::{ AccountPoolMetrics, AttemptMetrics, CostCoverage, CurrencyCost, DashboardObservation, DashboardRuntimeSlots, DiagnosticDimension, DiagnosticObservation, Granularity, - HealthStatus, OpsErrorPage, OpsErrorQuery, RequestMetricPoint, RequestMetrics, - TimeRange, TrendKind, UsageCalculatedBillingFact, UsageDetail, UsageFilter, - UsageOverview, UsagePage, UsageQuery, china_day_start, + HealthStatus, OpsErrorPage, OpsErrorQuery, PageNumber, RequestMetricPoint, + RequestMetrics, RequestOutcome, TimeRange, TrendKind, UsageBilling, + UsageCalculatedBillingFact, UsageDetail, UsageFilter, UsageOverview, UsagePage, + UsageQuery, UsageRecord, china_day_start, }, settings::{ AdminApiKey, AdminApiKeyMutation, ReplaceRuntimeSettings, RotationStrategy, @@ -318,12 +320,46 @@ async fn observability_services_should_calculate_usage_insights_and_diagnostic_s assert_eq!(diagnostics.items[1].request_share, 0.25); } +#[tokio::test] +async fn usage_records_should_tolerate_records_that_fail_billing_enrichment() { + let now = Utc::now(); + let store = Arc::new(FixtureObservabilityStore::new(observation_range(now))); + let invalid_kind = "x".repeat(65); + store.replace_usage_records(vec![ + calculated_total_record("request_invalid_kind", Some(&invalid_kind), now), + calculated_total_record("request_unregistered_kind", Some("anthropic"), now), + calculated_total_record("request_enriched", Some("openai"), now), + ]); + let services = observability_services_with_calculated_billing(store).await; + + let page = services + .observability() + .usage_records(usage_query(now)) + .await + .expect("one bad record must not fail the whole usage list"); + + assert_eq!(page.items.len(), 3); + assert!( + matches!(page.items[0].billing, Some(UsageBilling::Total { .. })), + "invalid Provider kind keeps the stored total" + ); + assert!( + matches!(page.items[1].billing, Some(UsageBilling::Total { .. })), + "unregistered Provider kind keeps the stored total" + ); + assert!( + matches!(page.items[2].billing, Some(UsageBilling::Calculated(_))), + "healthy record is still enriched" + ); +} + struct FixtureObservabilityStore { trend: Mutex>, overview: Mutex, calculated_billing_facts: Mutex>, diagnostics: Mutex>, runtime_slots: Mutex>, + usage_records: Mutex>, } impl FixtureObservabilityStore { @@ -339,6 +375,7 @@ impl FixtureObservabilityStore { calculated_billing_facts: Mutex::new(Vec::new()), diagnostics: Mutex::new(Vec::new()), runtime_slots: Mutex::new(None), + usage_records: Mutex::new(Vec::new()), } } @@ -364,6 +401,10 @@ impl FixtureObservabilityStore { fn replace_runtime_slots(&self, runtime_slots: Option) { *self.runtime_slots.lock().expect("runtime slots") = runtime_slots; } + + fn replace_usage_records(&self, records: Vec) { + *self.usage_records.lock().expect("usage records") = records; + } } #[async_trait] @@ -412,7 +453,13 @@ impl ObservabilityStore for FixtureObservabilityStore { } async fn list_usage_records(&self, _: UsageQuery) -> AdminStoreResult { - Err(super::unavailable("usage records")) + 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, + next_cursor: None, + }) } async fn usage_record_detail(&self, _: &str) -> AdminStoreResult { @@ -511,6 +558,96 @@ fn observation_range(end: DateTime) -> TimeRange { TimeRange::new(end - Duration::hours(24), end).expect("observation range") } +fn usage_query(now: DateTime) -> UsageQuery { + UsageQuery { + range: observation_range(now), + filter: UsageFilter::default(), + cursor: None, + page: PageNumber::new(NonZeroU32::new(1).expect("page number")), + page_size: PageSize::new(50).expect("page size"), + } +} + +fn calculated_total_record( + id: &str, + provider_kind: Option<&str>, + now: DateTime, +) -> UsageRecord { + UsageRecord { + id: id.to_owned(), + client_api_key_ref: "key_billing".to_owned(), + config_revision: 1, + protocol: "openai".to_owned(), + operation: "generate".to_owned(), + endpoint: "/v1/responses".to_owned(), + client_transport: "http_sse".to_owned(), + requested_model_id: "gpt-5.5".to_owned(), + provider_kind: provider_kind.map(str::to_owned), + provider_account_ref: None, + provider_account_name: None, + provider_account_email: None, + provider_account_authentication_kind: None, + upstream_model_id: Some("gpt-5.5".to_owned()), + upstream_transport: None, + http_version: None, + websocket_pool: None, + provider_metadata_json: None, + attempt_count: 1, + upstream_send_state: "sent".to_owned(), + downstream_committed_at: None, + outcome: RequestOutcome::Succeeded, + client_status_code: Some(200), + upstream_status_code: Some(200), + client_response_id: None, + upstream_request_id: None, + upstream_response_id: None, + error_kind: None, + provider_error_code: None, + error_message: None, + retry_after_ms: None, + input_tokens: Some(800), + output_tokens: Some(200), + cached_tokens: Some(0), + cache_write_tokens: Some(0), + reasoning_tokens: Some(0), + image_input_tokens: None, + image_output_tokens: None, + total_tokens: Some(1_000), + cost_source: "calculated".to_owned(), + cost_amount: None, + cost_currency: None, + billing: Some(UsageBilling::Total { + source: "calculated".to_owned(), + total: CurrencyCost { + currency: "USD".to_owned(), + amount: gateway_admin::model::observability::DecimalAmount::from_str("1.25") + .expect("billing total"), + }, + }), + transport_decision_wait_ms: None, + connect_ms: None, + headers_ms: None, + first_event_ms: None, + first_reasoning_ms: None, + first_text_ms: None, + first_token_ms: None, + provider_processing_ms: None, + latency_ms: Some(100), + client_ip: None, + user_agent: None, + reasoning_effort: None, + reasoning_preset: None, + request_kind: None, + subagent_kind: None, + compact: false, + image_generation_requested: false, + image_generation_succeeded: None, + started_at: now, + deadline_at: now, + completed_at: Some(now), + } +} + fn health_metric_point(bucket_start: DateTime, metrics: RequestMetrics) -> RequestMetricPoint { RequestMetricPoint { bucket_start, diff --git a/backend/crates/gateway-store/src/postgres/provider_accounts.rs b/backend/crates/gateway-store/src/postgres/provider_accounts.rs index b35246ab3..a7c67367d 100644 --- a/backend/crates/gateway-store/src/postgres/provider_accounts.rs +++ b/backend/crates/gateway-store/src/postgres/provider_accounts.rs @@ -1661,6 +1661,7 @@ fn admin_account_status( } else if account.availability == ProviderAccountAvailability::QuotaExhausted { AdminAccountStatus::QuotaExhausted } else if account.availability == ProviderAccountAvailability::Expired + || account.availability == ProviderAccountAvailability::Invalid || account .access_token_expires_at .is_some_and(|expires_at| expires_at <= now) diff --git a/backend/crates/gateway-store/tests/postgres/provider_accounts.rs b/backend/crates/gateway-store/tests/postgres/provider_accounts.rs index e80223f7f..74fb30bbb 100644 --- a/backend/crates/gateway-store/tests/postgres/provider_accounts.rs +++ b/backend/crates/gateway-store/tests/postgres/provider_accounts.rs @@ -214,7 +214,7 @@ async fn terminal_admin_list_filters_and_sorts_before_pagination_with_retained_u assert_eq!(usage_page.config_revision.get(), 1); assert_eq!(usage_page.total, 5); assert_eq!(usage_page.summary.total, 5); - assert_eq!(usage_page.summary.active, 3); + assert_eq!(usage_page.summary.active, 2); assert_eq!(usage_page.summary.quota_exhausted, 1); assert_eq!( usage_page