diff --git a/backend/crates/gateway-admin/src/ports/store.rs b/backend/crates/gateway-admin/src/ports/store.rs index 411f0072d..7a5f6eda9 100644 --- a/backend/crates/gateway-admin/src/ports/store.rs +++ b/backend/crates/gateway-admin/src/ports/store.rs @@ -218,7 +218,12 @@ pub trait ClientKeyStore: Send + Sync { /// 用量、趋势、诊断与运维错误的只读能力。 #[async_trait] pub trait ObservabilityStore: Send + Sync { - async fn dashboard_summary(&self, range: TimeRange) -> AdminStoreResult; + /// 返回历史统计区间和指定观测时刻下的实时账号状态。 + async fn dashboard_summary( + &self, + range: TimeRange, + observed_at: DateTime, + ) -> AdminStoreResult; /// 返回 Dashboard 可选的实时槽位事实。 /// diff --git a/backend/crates/gateway-admin/src/use_case/observability.rs b/backend/crates/gateway-admin/src/use_case/observability.rs index a116f81b6..8939e6426 100644 --- a/backend/crates/gateway-admin/src/use_case/observability.rs +++ b/backend/crates/gateway-admin/src/use_case/observability.rs @@ -98,10 +98,11 @@ impl ObservabilityService for DefaultObservabilityService { range: TimeRange, kind: TrendKind, ) -> Result { + let observed_at = Utc::now(); let (mut observation, settings, runtime_slots) = futures::try_join!( - self.store.dashboard_summary(range), + self.store.dashboard_summary(range, observed_at), self.settings.load_runtime_settings(), - self.store.dashboard_runtime_slots(range.end), + self.store.dashboard_runtime_slots(observed_at), ) .map_err(|error| map_store_error(error, "dashboard"))?; self.enrich_billing(&mut observation.recent_requests); diff --git a/backend/crates/gateway-admin/tests/use_case/mod.rs b/backend/crates/gateway-admin/tests/use_case/mod.rs index c7b1b196e..cda8de9db 100644 --- a/backend/crates/gateway-admin/tests/use_case/mod.rs +++ b/backend/crates/gateway-admin/tests/use_case/mod.rs @@ -14,6 +14,7 @@ use std::{ }; use async_trait::async_trait; +use chrono::{DateTime, Utc}; use futures::future::BoxFuture; use gateway_admin::{ AdminConfig, AdminServices, InitialAdminPassword, @@ -391,7 +392,11 @@ impl ClientKeyStore for UnavailableStore { #[async_trait] impl ObservabilityStore for UnavailableStore { - async fn dashboard_summary(&self, _: TimeRange) -> AdminStoreResult { + async fn dashboard_summary( + &self, + _: TimeRange, + _: DateTime, + ) -> AdminStoreResult { Err(unavailable("dashboard")) } diff --git a/backend/crates/gateway-admin/tests/use_case/observability.rs b/backend/crates/gateway-admin/tests/use_case/observability.rs index 3deec85ab..15a91e5d6 100644 --- a/backend/crates/gateway-admin/tests/use_case/observability.rs +++ b/backend/crates/gateway-admin/tests/use_case/observability.rs @@ -200,6 +200,25 @@ async fn dashboard_summary_should_project_rebuildable_runtime_slots() { assert_eq!(capacity.available_slots, Some(1)); } +#[tokio::test] +async fn dashboard_summary_should_share_one_current_observation_time_across_runtime_facts() { + let historical_end = Utc::now() - Duration::hours(1); + let store = Arc::new(FixtureObservabilityStore::new(observation_range( + historical_end, + ))); + let services = observability_services(store.clone()).await; + + services + .observability() + .dashboard_summary(observation_range(historical_end), TrendKind::Usage) + .await + .expect("dashboard summary"); + + let (summary_observed_at, slots_observed_at) = store.observed_times(); + assert_eq!(summary_observed_at, slots_observed_at); + assert!(summary_observed_at.is_some_and(|value| value > historical_end)); +} + #[tokio::test] async fn observability_services_should_calculate_usage_insights_and_diagnostic_shares() { let now = Utc::now(); @@ -394,6 +413,8 @@ struct FixtureObservabilityStore { calculated_billing_facts: Mutex>, diagnostics: Mutex>, runtime_slots: Mutex>, + summary_observed_at: Mutex>>, + slots_observed_at: Mutex>>, usage_records: Mutex>, } @@ -410,6 +431,8 @@ impl FixtureObservabilityStore { calculated_billing_facts: Mutex::new(Vec::new()), diagnostics: Mutex::new(Vec::new()), runtime_slots: Mutex::new(None), + summary_observed_at: Mutex::new(None), + slots_observed_at: Mutex::new(None), usage_records: Mutex::new(Vec::new()), } } @@ -437,6 +460,16 @@ impl FixtureObservabilityStore { *self.runtime_slots.lock().expect("runtime slots") = runtime_slots; } + fn observed_times(&self) -> (Option>, Option>) { + ( + *self + .summary_observed_at + .lock() + .expect("summary observed at"), + *self.slots_observed_at.lock().expect("slots observed at"), + ) + } + fn replace_usage_records(&self, records: Vec) { *self.usage_records.lock().expect("usage records") = records; } @@ -444,7 +477,15 @@ impl FixtureObservabilityStore { #[async_trait] impl ObservabilityStore for FixtureObservabilityStore { - async fn dashboard_summary(&self, range: TimeRange) -> AdminStoreResult { + async fn dashboard_summary( + &self, + range: TimeRange, + observed_at: DateTime, + ) -> AdminStoreResult { + *self + .summary_observed_at + .lock() + .expect("summary observed at") = Some(observed_at); Ok(DashboardObservation { range, requests: RequestMetrics::default(), @@ -459,8 +500,9 @@ impl ObservabilityStore for FixtureObservabilityStore { async fn dashboard_runtime_slots( &self, - _: DateTime, + observed_at: DateTime, ) -> AdminStoreResult> { + *self.slots_observed_at.lock().expect("slots observed at") = Some(observed_at); Ok(*self.runtime_slots.lock().expect("runtime slots")) } diff --git a/backend/crates/gateway-api/tests/admin/mod.rs b/backend/crates/gateway-api/tests/admin/mod.rs index 08c148842..eab9d45d5 100644 --- a/backend/crates/gateway-api/tests/admin/mod.rs +++ b/backend/crates/gateway-api/tests/admin/mod.rs @@ -7,7 +7,7 @@ use std::{ }; use async_trait::async_trait; -use chrono::{Duration, Utc}; +use chrono::{DateTime, Duration, Utc}; use futures::future::BoxFuture; use gateway_admin::{ AdminConfig, AdminServices, InitialAdminPassword, @@ -528,7 +528,11 @@ impl AccountStore for UnusedStore { #[async_trait] impl ObservabilityStore for UnusedStore { - async fn dashboard_summary(&self, range: TimeRange) -> AdminStoreResult { + async fn dashboard_summary( + &self, + range: TimeRange, + _: DateTime, + ) -> AdminStoreResult { *self .dashboard_summary_range .lock() 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 98b438f66..fa4171470 100644 --- a/backend/crates/gateway-store/src/postgres/observability/admin_adapter.rs +++ b/backend/crates/gateway-store/src/postgres/observability/admin_adapter.rs @@ -98,6 +98,7 @@ impl ObservabilityRepository for PgObservabilityRepository { async fn dashboard_summary( &self, range: ObservabilityRange, + observed_at: DateTime, ) -> StoreResult { let filter = UsageRecordFilter::default(); let account_usage_range = ObservabilityRange::new( @@ -123,7 +124,7 @@ impl ObservabilityRepository for PgObservabilityRepository { attempt_metrics(&self.pool, range, &filter), dashboard_totals(&self.pool), )?; - let (provider_accounts, _) = self.account_status_snapshot(range.end).await?; + let (provider_accounts, _) = self.account_status_snapshot(observed_at).await?; let (trend, account_usage, recent_requests) = futures::try_join!( request_metric_series(&self.pool, range, &filter), provider_account_usage(&self.pool, account_usage_query), @@ -215,10 +216,11 @@ impl AdminObservabilityStore for PgAdminObservabilityStore { async fn dashboard_summary( &self, range: admin_observability::TimeRange, + observed_at: DateTime, ) -> AdminStoreResult { let observation = self .repository - .dashboard_summary(store_range(range)?) + .dashboard_summary(store_range(range)?, observed_at) .await .map_err(observability_error)?; admin_dashboard_observation(observation) diff --git a/backend/crates/gateway-store/src/postgres/observability/model.rs b/backend/crates/gateway-store/src/postgres/observability/model.rs index 33d8083d3..cb43aac33 100644 --- a/backend/crates/gateway-store/src/postgres/observability/model.rs +++ b/backend/crates/gateway-store/src/postgres/observability/model.rs @@ -683,6 +683,7 @@ pub trait ObservabilityRepository: Send + Sync { async fn dashboard_summary( &self, range: ObservabilityRange, + observed_at: DateTime, ) -> StoreResult; async fn dashboard_trend( &self, diff --git a/backend/crates/gateway-store/tests/postgres/observability.rs b/backend/crates/gateway-store/tests/postgres/observability.rs index c506487bc..ab23dc1fa 100644 --- a/backend/crates/gateway-store/tests/postgres/observability.rs +++ b/backend/crates/gateway-store/tests/postgres/observability.rs @@ -166,7 +166,7 @@ async fn dashboard_account_metrics_should_partition_account_statuses() { .expect("dashboard range"); let metrics = repository - .dashboard_summary(range) + .dashboard_summary(range, now) .await .expect("dashboard summary") .provider_accounts; @@ -252,7 +252,7 @@ async fn dashboard_account_metrics_with_cooldowns_should_only_reclassify_eligibl .expect("dashboard range"); let metrics = repository - .dashboard_summary(range) + .dashboard_summary(range, now) .await .expect("dashboard summary") .provider_accounts; @@ -416,7 +416,7 @@ async fn admin_observability_adapter_preserves_utc_queries_metrics_costs_and_det let store = PgAdminObservabilityStore::new(database.pool.clone(), None, None); let dashboard = store - .dashboard_summary(range) + .dashboard_summary(range, now) .await .expect("admin dashboard summary"); assert_eq!(dashboard.range, range); @@ -714,7 +714,7 @@ async fn dashboard_summary_totals_include_history_outside_selected_range() { let repository = PgObservabilityRepository::new(database.pool.clone(), None); let dashboard = repository - .dashboard_summary(range) + .dashboard_summary(range, now) .await .expect("dashboard summary"); assert_eq!(dashboard.requests.request_count, 2); @@ -753,7 +753,7 @@ async fn observability_queries_preserve_request_account_cost_and_diagnostic_fact let repository = PgObservabilityRepository::new(database.pool.clone(), None); let dashboard = repository - .dashboard_summary(range) + .dashboard_summary(range, now) .await .expect("dashboard summary"); assert_eq!(dashboard.requests.request_count, 3);