From 36828b1a7ed1c86728b65cf31b23e87c8292aa88 Mon Sep 17 00:00:00 2001 From: zyycn Date: Wed, 15 Jul 2026 03:09:39 +0800 Subject: [PATCH] fix: harden websocket liveness and usage reporting Enforce pump-owned Pong deadlines without adding request-path probes, and pause watchdogs during inbound backpressure. Refresh usage snapshots against the current time, expose account identities, preserve full plan badges, and source dashboard totals from global request buckets. Co-authored-by: Codex --- backend/src/api/admin/dashboard_routes.rs | 8 +- backend/src/telemetry/account_usage/query.rs | 53 +---- backend/src/telemetry/account_usage/store.rs | 64 ------ backend/src/telemetry/buckets/query.rs | 21 ++ backend/src/telemetry/dashboard.rs | 19 +- backend/src/telemetry/usage/insights.rs | 37 +++- .../openai/transport/websocket_pool.rs | 5 +- .../openai/transport/websocket_pump.rs | 194 +++++++++++------- backend/tests/api/admin/dashboard_routes.rs | 75 ++----- backend/tests/api/admin/usage_routes.rs | 27 +++ .../upstream/openai/transport/websocket.rs | 55 +++-- .../openai/transport/websocket_pool.rs | 141 ++++++++++--- docs/architecture.md | 2 +- .../components/AccountIdentityCell.vue | 4 +- .../accounts/components/AccountPlanBadge.vue | 8 +- .../usage/components/UsageDiagnosticCard.vue | 2 +- .../usage/composables/useUsageRecordsTable.ts | 21 +- .../usage/composables/useUsageTimeRange.ts | 40 ++++ frontend/src/views/usage/index.vue | 30 +-- 19 files changed, 453 insertions(+), 353 deletions(-) create mode 100644 frontend/src/views/usage/composables/useUsageTimeRange.ts diff --git a/backend/src/api/admin/dashboard_routes.rs b/backend/src/api/admin/dashboard_routes.rs index 2c3cc737d..295ff799e 100644 --- a/backend/src/api/admin/dashboard_routes.rs +++ b/backend/src/api/admin/dashboard_routes.rs @@ -205,12 +205,12 @@ pub(crate) async fn dashboard_summary( let capacity = state.services.account_pool.capacity_summary_now().await; let now = Utc::now(); let today_filter = today_usage_record_filter(now); - let lifetime_usage = state + let retained_usage = state .services .usage - .summary() + .retained_totals() .await - .map_err(|error| dashboard_data_error("account usage summary", &error))?; + .map_err(|error| dashboard_data_error("retained usage summary", &error))?; let account_usage_records = state .services .usage_records @@ -256,7 +256,7 @@ pub(crate) async fn dashboard_summary( cards: dashboard_cards( dashboard_account_counts(&accounts), &time_buckets, - &lifetime_usage, + &retained_usage, ), trend, health_timeline: dashboard_health_timeline_data(&time_buckets), diff --git a/backend/src/telemetry/account_usage/query.rs b/backend/src/telemetry/account_usage/query.rs index e5fc2280b..2122d547b 100644 --- a/backend/src/telemetry/account_usage/query.rs +++ b/backend/src/telemetry/account_usage/query.rs @@ -6,7 +6,7 @@ use thiserror::Error; use crate::infra::format::nonnegative_i64_to_u64; use crate::telemetry::{ - account_usage::store::{PgAccountUsageStore, UsageListRecord, UsageSummary}, + account_usage::store::{PgAccountUsageStore, UsageListRecord}, billing, buckets::query::{ ModelBucketUsage, ModelUsageWindow, PgRequestBucketQuery, UsageBucketTotals, @@ -42,13 +42,12 @@ impl AccountUsageQueryService { .map_err(|_| AccountUsageQueryError::List) } - /// 汇总账号用量。 - pub async fn summary(&self) -> Result { - self.store - .usage_summary() + /// 汇总请求时间桶保留期内的全局用量。 + pub async fn retained_totals(&self) -> Result { + self.buckets + .retained_totals() .await - .map(AccountUsageSummary::from) - .map_err(|_| AccountUsageQueryError::Summary) + .map_err(|_| AccountUsageQueryError::RetainedTotals) } /// 列出指定时间范围内的聚合时间桶。 @@ -95,8 +94,8 @@ impl AccountUsageQueryService { pub enum AccountUsageQueryError { #[error("failed to list account usage")] List, - #[error("failed to summarize account usage")] - Summary, + #[error("failed to summarize retained usage")] + RetainedTotals, #[error("failed to list usage time buckets")] TimeBuckets, } @@ -129,23 +128,6 @@ pub struct AccountUsageRecord { pub last_used_at: Option>, } -/// 管理端账号用量汇总。 -#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] -pub struct AccountUsageSummary { - pub account_count: i64, - pub request_count: i64, - pub empty_response_count: i64, - pub input_tokens: i64, - pub output_tokens: i64, - pub cached_tokens: i64, - pub reasoning_tokens: i64, - pub total_tokens: i64, - pub image_input_tokens: i64, - pub image_output_tokens: i64, - pub image_request_count: i64, - pub image_request_failed_count: i64, -} - /// 管理端时间桶用量记录。 #[derive(Debug, Clone, PartialEq)] pub struct AccountUsageTimeBucket { @@ -166,25 +148,6 @@ pub struct AccountUsageTimeBucket { pub billing_amount_usd: Option, } -impl From for AccountUsageSummary { - fn from(s: UsageSummary) -> Self { - Self { - account_count: s.account_count, - request_count: s.request_count, - empty_response_count: s.empty_response_count, - input_tokens: s.input_tokens, - output_tokens: s.output_tokens, - cached_tokens: s.cached_tokens, - reasoning_tokens: s.reasoning_tokens, - total_tokens: s.total_tokens, - image_input_tokens: s.image_input_tokens, - image_output_tokens: s.image_output_tokens, - image_request_count: s.image_request_count, - image_request_failed_count: s.image_request_failed_count, - } - } -} - impl From for AccountUsageTimeBucket { fn from(record: UsageTimeBucketRecord) -> Self { let billing_amount_usd = time_bucket_billing_amount_usd(&record); diff --git a/backend/src/telemetry/account_usage/store.rs b/backend/src/telemetry/account_usage/store.rs index a3cee77a1..62b83eef0 100644 --- a/backend/src/telemetry/account_usage/store.rs +++ b/backend/src/telemetry/account_usage/store.rs @@ -129,22 +129,6 @@ select from account_usage au left join accounts a on a.id = au.account_id"; -const USAGE_SUMMARY_SQL: &str = r" -select - count(*) as account_count, - coalesce(sum(request_count), 0)::bigint as request_count, - coalesce(sum(empty_response_count), 0)::bigint as empty_response_count, - coalesce(sum(input_tokens), 0)::bigint as input_tokens, - coalesce(sum(output_tokens), 0)::bigint as output_tokens, - coalesce(sum(cached_tokens), 0)::bigint as cached_tokens, - coalesce(sum(reasoning_tokens), 0)::bigint as reasoning_tokens, - coalesce(sum(total_tokens), 0)::bigint as total_tokens, - coalesce(sum(image_input_tokens), 0)::bigint as image_input_tokens, - coalesce(sum(image_output_tokens), 0)::bigint as image_output_tokens, - coalesce(sum(image_request_count), 0)::bigint as image_request_count, - coalesce(sum(image_request_failed_count), 0)::bigint as image_request_failed_count -from account_usage"; - /// PostgreSQL 用量存储错误。 #[derive(Debug, Error)] pub enum PgAccountUsageStoreError { @@ -312,35 +296,6 @@ pub struct UsageListRecord { pub last_used_at: Option>, } -/// 账号用量汇总。 -#[derive(Debug, Clone)] -pub struct UsageSummary { - /// 账号数。 - pub account_count: i64, - /// 总请求数。 - pub request_count: i64, - /// 总空响应数。 - pub empty_response_count: i64, - /// 总输入 token。 - pub input_tokens: i64, - /// 总输出 token。 - pub output_tokens: i64, - /// 总缓存 token。 - pub cached_tokens: i64, - /// 总 reasoning token。 - pub reasoning_tokens: i64, - /// 总 token。 - pub total_tokens: i64, - /// 总图片输入 token。 - pub image_input_tokens: i64, - /// 总图片输出 token。 - pub image_output_tokens: i64, - /// 总图片请求成功数。 - pub image_request_count: i64, - /// 总图片请求失败数。 - pub image_request_failed_count: i64, -} - /// PostgreSQL 用量存储。 #[derive(Clone)] pub struct PgAccountUsageStore { @@ -492,25 +447,6 @@ where account_id in (", let rows = builder.build().fetch_all(&self.pool).await?; rows.iter().map(usage_list_from_row).collect() } - - /// 汇总账号用量。 - pub async fn usage_summary(&self) -> PgAccountUsageStoreResult { - let row = sqlx::query(USAGE_SUMMARY_SQL).fetch_one(&self.pool).await?; - Ok(UsageSummary { - account_count: row.get("account_count"), - request_count: row.get("request_count"), - empty_response_count: row.get("empty_response_count"), - input_tokens: row.get("input_tokens"), - output_tokens: row.get("output_tokens"), - cached_tokens: row.get("cached_tokens"), - reasoning_tokens: row.get("reasoning_tokens"), - total_tokens: row.get("total_tokens"), - image_input_tokens: row.get("image_input_tokens"), - image_output_tokens: row.get("image_output_tokens"), - image_request_count: row.get("image_request_count"), - image_request_failed_count: row.get("image_request_failed_count"), - }) - } } #[async_trait] diff --git a/backend/src/telemetry/buckets/query.rs b/backend/src/telemetry/buckets/query.rs index 9b21b3f64..7bdf299fd 100644 --- a/backend/src/telemetry/buckets/query.rs +++ b/backend/src/telemetry/buckets/query.rs @@ -29,6 +29,14 @@ where bucket_start >= $1 and bucket_start <= $2 group by bucket_start, model, service_tier order by bucket_start asc, model asc, service_tier asc"; +const RETAINED_USAGE_TOTALS_SQL: &str = r" +select + coalesce(sum(success_count + error_count), 0)::bigint as request_count, + coalesce(sum(input_tokens), 0)::bigint as input_tokens, + coalesce(sum(output_tokens), 0)::bigint as output_tokens, + coalesce(sum(cached_tokens), 0)::bigint as cached_tokens +from request_time_buckets"; + /// 请求时间桶查询错误。 #[derive(Debug, Error)] pub enum PgRequestBucketQueryError { @@ -127,6 +135,19 @@ impl PgRequestBucketQuery { Ok(rows.iter().map(usage_time_bucket_from_row).collect()) } + /// 汇总请求时间桶保留期内的全局用量，不受当前账号集合影响。 + pub async fn retained_totals(&self) -> Result { + let row = sqlx::query(RETAINED_USAGE_TOTALS_SQL) + .fetch_one(&self.pool) + .await?; + Ok(UsageBucketTotals { + request_count: row.get("request_count"), + input_tokens: row.get("input_tokens"), + output_tokens: row.get("output_tokens"), + cached_tokens: row.get("cached_tokens"), + }) + } + pub async fn usage_by_windows( &self, windows: &[UsageBucketWindow], diff --git a/backend/src/telemetry/dashboard.rs b/backend/src/telemetry/dashboard.rs index 8cdc21eba..716661083 100644 --- a/backend/src/telemetry/dashboard.rs +++ b/backend/src/telemetry/dashboard.rs @@ -11,7 +11,7 @@ use crate::infra::{ time::{china_day_start, china_quarter_hour_start}, }; -use super::account_usage::query::{AccountUsageSummary, AccountUsageTimeBucket}; +use super::{account_usage::query::AccountUsageTimeBucket, buckets::query::UsageBucketTotals}; const HEALTH_TIMELINE_SLOT_MINUTES: i64 = 15; const HEALTH_TIMELINE_SLOTS: i64 = 24 * 4; @@ -174,15 +174,15 @@ impl UsageWindow { pub fn dashboard_cards( accounts: DashboardAccountCounts, buckets: &[AccountUsageTimeBucket], - lifetime_usage: &AccountUsageSummary, + retained_usage: &UsageBucketTotals, ) -> DashboardCardsData { - dashboard_cards_at(accounts, buckets, lifetime_usage, Utc::now()) + dashboard_cards_at(accounts, buckets, retained_usage, Utc::now()) } pub fn dashboard_cards_at( accounts: DashboardAccountCounts, buckets: &[AccountUsageTimeBucket], - lifetime_usage: &AccountUsageSummary, + retained_usage: &UsageBucketTotals, now: DateTime, ) -> DashboardCardsData { let today_start = china_day_start(now); @@ -190,9 +190,10 @@ pub fn dashboard_cards_at( let today = usage_window(buckets, today_start, now); let yesterday = usage_window(buckets, yesterday_start, today_start); let today_billing_amount = billing_window(buckets, today_start, now).unwrap_or(0.0); - let total_requests = nonnegative_i64_to_u64(lifetime_usage.request_count); - let total_tokens = nonnegative_i64_to_u64(lifetime_usage.total_tokens); - let total_cached_tokens = nonnegative_i64_to_u64(lifetime_usage.cached_tokens); + let total_requests = nonnegative_i64_to_u64(retained_usage.request_count); + let total_tokens = nonnegative_i64_to_u64(retained_usage.input_tokens) + .saturating_add(nonnegative_i64_to_u64(retained_usage.output_tokens)); + let total_cached_tokens = nonnegative_i64_to_u64(retained_usage.cached_tokens); DashboardCardsData { accounts: DashboardAccountsCardData { @@ -220,7 +221,7 @@ pub fn dashboard_cards_at( today_hit_rate: format_rate(today.cache_hit_rate()), today_hit_rate_value: today.cache_hit_rate(), yesterday_hit_rate_value: yesterday.cache_hit_rate(), - total_hit_rate: format_rate(summary_cache_hit_rate(lifetime_usage)), + total_hit_rate: format_rate(summary_cache_hit_rate(retained_usage)), total_cached_tokens: format_tokens(total_cached_tokens), average_first_token_latency_ms: format_optional_duration_ms( today.avg_first_token_latency(), @@ -540,7 +541,7 @@ fn health_tone(bucket: UsageWindow, success_rate: f64, is_future: bool) -> char } } -fn summary_cache_hit_rate(summary: &AccountUsageSummary) -> Option { +fn summary_cache_hit_rate(summary: &UsageBucketTotals) -> Option { let input_tokens = nonnegative_i64_to_u64(summary.input_tokens); (input_tokens > 0) .then(|| nonnegative_i64_to_u64(summary.cached_tokens) as f64 / input_tokens as f64) diff --git a/backend/src/telemetry/usage/insights.rs b/backend/src/telemetry/usage/insights.rs index cc83dcc9f..55e4b6873 100644 --- a/backend/src/telemetry/usage/insights.rs +++ b/backend/src/telemetry/usage/insights.rs @@ -162,12 +162,27 @@ impl UsageDiagnosticsDimension { fn expression(self) -> &'static str { match self { - Self::Model => "model_name", - Self::Account => "coalesce(nullif(account_id, ''), '未知账号')", - Self::ApiKey => "coalesce(nullif(client_api_key_id, ''), '未知 API Key')", - Self::Provider => "coalesce(nullif(provider, ''), '未知 Provider')", - Self::Transport => "coalesce(nullif(transport, ''), '未知传输')", - Self::FailureClass => "failure_class", + Self::Model => "terminal_requests.model_name", + Self::Account => { + "coalesce(nullif(concat_ws(' → ', nullif(diagnostic_accounts.email, ''), nullif(terminal_requests.account_id, '')), ''), '未知账号')" + } + Self::ApiKey => { + "coalesce(nullif(terminal_requests.client_api_key_id, ''), '未知 API Key')" + } + Self::Provider => "coalesce(nullif(terminal_requests.provider, ''), '未知 Provider')", + Self::Transport => "coalesce(nullif(terminal_requests.transport, ''), '未知传输')", + Self::FailureClass => "terminal_requests.failure_class", + } + } + + fn source(self) -> &'static str { + match self { + Self::Account => { + "terminal_requests left join accounts diagnostic_accounts on diagnostic_accounts.id = terminal_requests.account_id" + } + Self::Model | Self::ApiKey | Self::Provider | Self::Transport | Self::FailureClass => { + "terminal_requests" + } } } } @@ -307,8 +322,9 @@ pub async fn diagnostics( dimension: UsageDiagnosticsDimension, ) -> PgUsageRecordStoreResult { let expression = dimension.expression(); + let source = dimension.source(); let failure_filter = if dimension == UsageDiagnosticsDimension::FailureClass { - "where not is_success" + "where not terminal_requests.is_success" } else { "" }; @@ -321,7 +337,7 @@ pub async fn diagnostics( percentile_cont(0.95) within group (order by latency_ms)\n\ filter (where is_success and latency_ms is not null) as latency_p95_ms,\n\ (select count(*) from terminal_requests)::bigint as total_requests\n\ - from terminal_requests {failure_filter}\n\ + from {source} {failure_filter}\n\ group by name\n\ order by request_count desc, name asc\n\ limit 8" @@ -610,6 +626,7 @@ async fn load_diagnostic_costs( dimension: UsageDiagnosticsDimension, ) -> PgUsageRecordStoreResult> { let expression = dimension.expression(); + let source = dimension.source(); let sql = format!( "with {TERMINAL_REQUESTS_CTE}\n\ select {expression} as name, billing_model, service_tier,\n\ @@ -622,8 +639,8 @@ async fn load_diagnostic_costs( coalesce(sum(input_tokens), 0)::bigint as input_tokens,\n\ coalesce(sum(output_tokens), 0)::bigint as output_tokens,\n\ coalesce(sum(least(coalesce(cached_tokens, 0), coalesce(input_tokens, 0))), 0)::bigint as cached_tokens\n\ - from terminal_requests\n\ - where is_success\n\ + from {source}\n\ + where terminal_requests.is_success\n\ group by name, billing_model, service_tier, is_long" ); // SAFETY: the grouping expression comes from the closed diagnostics dimension enum. diff --git a/backend/src/upstream/openai/transport/websocket_pool.rs b/backend/src/upstream/openai/transport/websocket_pool.rs index fd6c39fa1..9a68cba14 100644 --- a/backend/src/upstream/openai/transport/websocket_pool.rs +++ b/backend/src/upstream/openai/transport/websocket_pool.rs @@ -84,7 +84,7 @@ pub struct CodexWebSocketPoolConfig { pub maintenance_interval: Option, /// idle socket 探活 ping 间隔；`None` 表示维护时只做过期清理。 pub ping_interval: Option, - /// 发送 ping 后等待上游响应的超时时间。 + /// 发送 ping 后等待上游响应的超时时间；零值表示不校验 Pong deadline。 pub ping_timeout: Duration, /// idle socket 无活动多久后视为失活。 pub liveness_timeout: Option, @@ -110,10 +110,11 @@ impl Default for CodexWebSocketPoolConfig { } impl CodexWebSocketPoolConfig { - /// pump 后台任务的保活策略：从连接池配置派生出 ping 间隔与 liveness 超时。 + /// pump 后台任务的保活策略：从连接池配置派生出 ping/pong 与 liveness 策略。 pub(crate) fn keepalive(&self) -> PumpKeepalive { PumpKeepalive { ping_interval: self.ping_interval, + ping_timeout: (!self.ping_timeout.is_zero()).then_some(self.ping_timeout), liveness_timeout: self.liveness_timeout, } } diff --git a/backend/src/upstream/openai/transport/websocket_pump.rs b/backend/src/upstream/openai/transport/websocket_pump.rs index 563c422e9..492d19e88 100644 --- a/backend/src/upstream/openai/transport/websocket_pump.rs +++ b/backend/src/upstream/openai/transport/websocket_pump.rs @@ -2,7 +2,7 @@ //! //! 每条上游 WebSocket 连接都由一个后台 pump 任务独占： //! - 持续读取 socket：一旦观察到 `Close` / EOF / 传输错误，立即把连接标记为 `closed`。 -//! - 自动回应上游 `Ping`（保活），并按 `ping_interval` 主动 `Ping`（穿透 NAT/中间盒空闲计时器）。 +//! - 自动回应上游 `Ping`，并按 `ping_interval` 主动 `Ping`；匹配的 `Pong` 必须在 deadline 内返回。 //! - 可选 `liveness_timeout`：长时间无任何入站活动时判定连接失活并退出。 //! //! 因此空闲连接的“死没死”在后台被实时感知；复用方只需零成本读取 [`PumpedWebSocket::is_closed`]， @@ -11,7 +11,7 @@ //! 收发都通过 channel 与 pump 任务交互： //! - 发送：`send` 走 command channel，等待 pump 回执。 //! - 接收：`next` 从 message channel 取出 pump 转发的入站帧（`Ping`/`Pong` 已被 pump 吞掉）。 -//! - 入站缓冲满时暂停读取 socket，并继续处理关闭命令与保活；消费恢复后按原顺序继续转发。 +//! - 入站缓冲满时暂停读取 socket 和主动探活，仅继续处理发送/关闭命令；消费恢复后按原顺序继续转发。 use std::sync::{ Arc, Mutex, @@ -39,6 +39,8 @@ const PUMP_COMMAND_BUFFER: usize = 32; pub(crate) struct PumpKeepalive { /// 主动 `Ping` 间隔；`None` 表示 pump 不主动 ping（仅被动读取 + 回应上游 ping）。 pub(crate) ping_interval: Option, + /// 主动 `Ping` 发出后等待匹配 `Pong` 的 deadline；`None` 表示不校验响应。 + pub(crate) ping_timeout: Option, /// 无入站活动多久后判定失活；`None` 表示只靠显式 close/传输错误发现死亡。 pub(crate) liveness_timeout: Option, } @@ -48,6 +50,7 @@ impl PumpKeepalive { pub(crate) fn disabled() -> Self { Self { ping_interval: None, + ping_timeout: None, liveness_timeout: None, } } @@ -70,6 +73,7 @@ pub(crate) enum PumpExitReason { UpstreamCloseFrame { frame: Option }, UpstreamEof, InboundTransportError { message: String }, + PongTimeout { timeout: Duration }, LivenessTimeout { timeout: Duration }, KeepaliveTransportError { message: String }, MessageReceiverClosed, @@ -84,6 +88,7 @@ impl PumpExitReason { Self::UpstreamCloseFrame { .. } => "upstream_close_frame", Self::UpstreamEof => "upstream_eof", Self::InboundTransportError { .. } => "inbound_transport_error", + Self::PongTimeout { .. } => "pong_timeout", Self::LivenessTimeout { .. } => "liveness_timeout", Self::KeepaliveTransportError { .. } => "keepalive_transport_error", Self::MessageReceiverClosed => "message_receiver_closed", @@ -96,7 +101,9 @@ impl PumpExitReason { | Self::InboundTransportError { message } | Self::KeepaliveTransportError { message } => Some(message.clone()), Self::UpstreamCloseFrame { frame } => frame.clone(), - Self::LivenessTimeout { timeout } => Some(format!("{timeout:?}")), + Self::PongTimeout { timeout } | Self::LivenessTimeout { timeout } => { + Some(format!("{timeout:?}")) + } Self::CommandChannelClosed | Self::LocalClose | Self::UpstreamEof @@ -110,6 +117,7 @@ impl PumpExitReason { Self::OutboundTransportError { .. } | Self::UpstreamEof | Self::InboundTransportError { .. } + | Self::PongTimeout { .. } | Self::LivenessTimeout { .. } | Self::KeepaliveTransportError { .. } ) @@ -120,6 +128,7 @@ impl PumpExitReason { self, Self::CommandChannelClosed | Self::UpstreamCloseFrame { .. } + | Self::PongTimeout { .. } | Self::LivenessTimeout { .. } | Self::MessageReceiverClosed ) @@ -131,6 +140,12 @@ struct PendingInbound { exit_reason: Option, } +#[derive(Debug, Clone, Copy)] +struct PendingPing { + payload: [u8; 8], + deadline: Instant, +} + /// 由后台 pump 任务托管的 WebSocket 连接句柄。 pub(crate) struct PumpedWebSocket { connection_id: Uuid, @@ -248,16 +263,9 @@ async fn pump_loop( ) { let mut last_activity = Instant::now(); let ping_interval = keepalive.ping_interval.filter(|d| !d.is_zero()); + let ping_timeout = keepalive.ping_timeout.filter(|d| !d.is_zero()); let liveness_timeout = keepalive.liveness_timeout.filter(|d| !d.is_zero()); - // ticker 由 ping 间隔或 liveness 检查间隔中较小者驱动：即使不主动 ping， - // 也要周期性醒来检查 liveness。两者都未配置时不启动 ticker。 - let tick_interval = match (ping_interval, liveness_timeout) { - (Some(ping), Some(live)) => Some(ping.min(live)), - (Some(ping), None) => Some(ping), - (None, Some(live)) => Some(live), - (None, None) => None, - }; - let mut ticker = tick_interval.map(|d| { + let mut ping_ticker = ping_interval.map(|d| { // 首个 tick 推迟一整个间隔：tokio::time::interval 默认会让首个 tick 立即就绪， // 否则连接一建立就会立刻发一帧 Ping，与首个请求 Text 抢跑、打乱帧序。 let mut ticker = tokio::time::interval_at(tokio::time::Instant::now() + d, d); @@ -266,6 +274,8 @@ async fn pump_loop( }); let mut pending_inbound: Option = None; + let mut pending_ping: Option = None; + let mut ping_sequence = 0_u64; let mut backpressure_events = 0_u64; let reason = 'pump: loop { @@ -282,6 +292,8 @@ async fn pump_loop( if let Some(reason) = pending.exit_reason { break 'pump reason; } + // 背压期间无法读取 socket，不能把这段本地下游阻塞误算成上游静默。 + last_activity = Instant::now(); } command = rx_command.recv() => { let Some(command) = command else { @@ -291,21 +303,12 @@ async fn pump_loop( break 'pump reason; } } - _ = tick(&mut ticker) => { - if let Some(reason) = handle_tick( - &mut inner, - last_activity, - ping_interval, - liveness_timeout, - false, - ).await { - break 'pump reason; - } - } } continue; } + let pong_deadline = pending_ping.map(|ping| ping.deadline); + let liveness_deadline = liveness_timeout.map(|timeout| last_activity + timeout); tokio::select! { command = rx_command.recv() => { let Some(command) = command else { @@ -318,36 +321,38 @@ async fn pump_loop( message = inner.next() => { match message { None => break 'pump PumpExitReason::UpstreamEof, - Some(Ok(Message::Ping(payload))) => { - last_activity = Instant::now(); - if let Err(error) = inner.send(Message::Pong(payload)).await { - break 'pump PumpExitReason::KeepaliveTransportError { - message: error.to_string(), - }; - } - } - Some(Ok(Message::Pong(_))) => { - last_activity = Instant::now(); - } Some(Ok(message)) => { last_activity = Instant::now(); - let terminal_reason = match &message { - Message::Close(frame) => Some(PumpExitReason::UpstreamCloseFrame { - frame: frame.as_ref().map(|frame| format!("{frame:?}")), - }), - _ => None, - }; - if let Some(reason) = enqueue_inbound( - &tx_message, - &mut pending_inbound, - connection_id, - &mut backpressure_events, - PendingInbound { - item: Ok(message), - exit_reason: terminal_reason, - }, - ) { - break 'pump reason; + observe_inbound(connection_id, &mut pending_ping, &message); + match message { + Message::Ping(payload) => { + if let Err(error) = inner.send(Message::Pong(payload)).await { + break 'pump PumpExitReason::KeepaliveTransportError { + message: error.to_string(), + }; + } + } + Message::Pong(_) => {} + message => { + let terminal_reason = match &message { + Message::Close(frame) => Some(PumpExitReason::UpstreamCloseFrame { + frame: frame.as_ref().map(|frame| format!("{frame:?}")), + }), + _ => None, + }; + if let Some(reason) = enqueue_inbound( + &tx_message, + &mut pending_inbound, + connection_id, + &mut backpressure_events, + PendingInbound { + item: Ok(message), + exit_reason: terminal_reason, + }, + ) { + break 'pump reason; + } + } } } Some(Err(err)) => { @@ -369,17 +374,28 @@ async fn pump_loop( } } } - _ = tick(&mut ticker) => { - if let Some(reason) = handle_tick( - &mut inner, - last_activity, - ping_interval, - liveness_timeout, - true, - ).await { - break 'pump reason; + _ = tick(&mut ping_ticker) => { + if pending_ping.is_none() { + match send_keepalive_ping( + &mut inner, + &mut ping_sequence, + ping_timeout, + ).await { + Ok(pending) => pending_ping = pending, + Err(reason) => break 'pump reason, + } } } + _ = wait_until(pong_deadline) => { + break 'pump PumpExitReason::PongTimeout { + timeout: ping_timeout.expect("pending ping must have a timeout"), + }; + } + _ = wait_until(liveness_deadline) => { + break 'pump PumpExitReason::LivenessTimeout { + timeout: liveness_timeout.expect("liveness deadline must have a timeout"), + }; + } } }; @@ -433,27 +449,39 @@ async fn handle_command(inner: &mut RawWsStream, command: PumpCommand) -> Option } } -async fn handle_tick( +fn observe_inbound(connection_id: Uuid, pending_ping: &mut Option, message: &Message) { + let Some(pending) = pending_ping else { + return; + }; + if let Message::Pong(payload) = message + && payload.as_ref() != pending.payload + { + tracing::debug!( + websocket_connection_id = %connection_id, + "Responses WebSocket pump received a non-matching Pong" + ); + } + // 匹配 Pong 是探针的直接回执；其他入站帧同样证明上游链路仍可达。 + *pending_ping = None; +} + +async fn send_keepalive_ping( inner: &mut RawWsStream, - last_activity: Instant, - ping_interval: Option, - liveness_timeout: Option, - check_liveness: bool, -) -> Option { - if check_liveness - && let Some(timeout) = liveness_timeout - && last_activity.elapsed() >= timeout - { - return Some(PumpExitReason::LivenessTimeout { timeout }); - } - if ping_interval.is_some() - && let Err(error) = inner.send(Message::Ping(Vec::new().into())).await - { - return Some(PumpExitReason::KeepaliveTransportError { + sequence: &mut u64, + ping_timeout: Option, +) -> Result, PumpExitReason> { + let payload = sequence.to_be_bytes(); + *sequence = sequence.wrapping_add(1); + inner + .send(Message::Ping(payload.to_vec().into())) + .await + .map_err(|error| PumpExitReason::KeepaliveTransportError { message: error.to_string(), - }); - } - None + })?; + Ok(ping_timeout.map(|timeout| PendingPing { + payload, + deadline: Instant::now() + timeout, + })) } fn log_pump_exit(connection_id: Uuid, reason: &PumpExitReason, backpressure_events: u64) { @@ -494,3 +522,11 @@ async fn tick(ticker: &mut Option) { None => std::future::pending::<()>().await, } } + +/// 等待可选 deadline；`None` 时永远挂起，让 `select!` 分支实际禁用。 +async fn wait_until(deadline: Option) { + match deadline { + Some(deadline) => tokio::time::sleep_until(deadline).await, + None => std::future::pending::<()>().await, + } +} diff --git a/backend/tests/api/admin/dashboard_routes.rs b/backend/tests/api/admin/dashboard_routes.rs index 4e25b615d..1b7021bbe 100644 --- a/backend/tests/api/admin/dashboard_routes.rs +++ b/backend/tests/api/admin/dashboard_routes.rs @@ -117,7 +117,7 @@ async fn dashboard_summary_should_calculate_today_traffic_and_average_first_toke .append(&usage_record_with_tokens(yesterday_record_at, 30)) .await .unwrap(); - seed_lifetime_usage_summary(&pool, 7, 400, 500, 100).await; + seed_retained_usage_bucket(&pool, today_start - Duration::days(8), 4, 340, 500, 100).await; let body = dashboard_summary(app).await; @@ -441,23 +441,6 @@ async fn dashboard_summary_should_match_versioned_model_names_for_billing() { assert_eq!(billing_amount, "$1.75"); } -#[tokio::test] -async fn dashboard_summary_should_not_price_usage_summary_without_model_dimension() { - let (app, _store, pool, _dir) = dashboard_test_app( - "dashboard-billing-summary-fallback", - crate::support::fingerprint::test_fingerprint(), - ) - .await; - seed_usage_summary(&pool, 100_000, 100_000).await; - - let body = dashboard_summary(app).await; - - assert_eq!( - body["data"]["cards"]["tokens"]["todayBillingAmountUsd"], - "$0.00" - ); -} - #[tokio::test] async fn dashboard_summary_should_return_backend_formatted_time_fields() { let (app, store, _pool, _dir) = dashboard_test_app( @@ -812,62 +795,30 @@ async fn dashboard_trend(app: axum::Router, kind: &str) -> Value { response_json(response).await } -async fn seed_usage_summary(pool: &PgPool, input_tokens: u64, output_tokens: u64) { - seed_usage_summary_with_last_used(pool, input_tokens, output_tokens, "2026-06-18T00:00:00Z") - .await; -} - -async fn seed_lifetime_usage_summary( +async fn seed_retained_usage_bucket( pool: &PgPool, + bucket_start: chrono::DateTime, request_count: u64, input_tokens: u64, output_tokens: u64, cached_tokens: u64, ) { - seed_dashboard_account(pool, "acct_lifetime", "lifetime@example.com").await; sqlx::query( - "insert into account_usage - (account_id, request_count, input_tokens, output_tokens, cached_tokens, total_tokens) - values ($1, $2, $3, $4, $5, $6)", + "insert into request_time_buckets + (bucket_start, provider, account_id, model, service_tier, success_count, + input_tokens, output_tokens, cached_tokens, updated_at) + values ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)", ) - .bind("acct_lifetime") + .bind(china_quarter_hour_start(bucket_start)) + .bind("openai") + .bind("acct_retained") + .bind("gpt-5.5") + .bind("__unknown__") .bind(request_count as i64) .bind(input_tokens as i64) .bind(output_tokens as i64) .bind(cached_tokens as i64) - .bind((input_tokens + output_tokens) as i64) - .execute(pool) - .await - .unwrap(); -} - -async fn seed_usage_summary_with_last_used( - pool: &PgPool, - input_tokens: u64, - output_tokens: u64, - last_used_at: &str, -) { - sqlx::query( - "insert into accounts (id, email, access_token, status, added_at, updated_at) values ($1, $2, $3, $4, $5, $6)", - ) - .bind("acct_usage") - .bind("acct-usage@example.com") - .bind("cipher") - .bind("active") - .bind(crate::support::storage::timestamp("2026-06-18T00:00:00Z")) - .bind(crate::support::storage::timestamp("2026-06-18T00:00:00Z")) - .execute(pool) - .await - .unwrap(); - sqlx::query( - "insert into account_usage (account_id, request_count, input_tokens, output_tokens, total_tokens, last_used_at) values ($1, $2, $3, $4, $5, $6)", - ) - .bind("acct_usage") - .bind(1_i64) - .bind(input_tokens as i64) - .bind(output_tokens as i64) - .bind((input_tokens + output_tokens) as i64) - .bind(crate::support::storage::timestamp(last_used_at)) + .bind(Utc::now()) .execute(pool) .await .unwrap(); diff --git a/backend/tests/api/admin/usage_routes.rs b/backend/tests/api/admin/usage_routes.rs index 9285d1a11..bce142617 100644 --- a/backend/tests/api/admin/usage_routes.rs +++ b/backend/tests/api/admin/usage_routes.rs @@ -124,6 +124,20 @@ async fn usage_summary_and_trend_expose_only_real_success_metrics() { async fn usage_insights_should_collapse_retry_errors_to_terminal_requests() { let (app, usage, ops, _guard) = monitoring_test_app("monitoring-insights", true).await; let now = chrono::Utc::now(); + sqlx::query( + "insert into accounts + (id, email, access_token, status, added_at, updated_at) + values ($1, $2, $3, $4, $5, $6)", + ) + .bind("acct_usage") + .bind("usage-account@example.com") + .bind("cipher") + .bind("active") + .bind(now) + .bind(now) + .execute(usage.pool()) + .await + .unwrap(); let mut success = success_record("eventually succeeded"); success.id = "usage_insights_success".to_string(); @@ -184,6 +198,19 @@ async fn usage_insights_should_collapse_retry_errors_to_terminal_requests() { .unwrap() .is_empty() ); + + let account_diagnostics = admin_get( + &app, + "/api/admin/usage/records/insights/diagnostics?dimension=account", + ) + .await; + assert!( + account_diagnostics["data"]["items"] + .as_array() + .unwrap() + .iter() + .any(|item| item["name"] == "usage-account@example.com → acct_usage") + ); } #[tokio::test] diff --git a/backend/tests/upstream/openai/transport/websocket.rs b/backend/tests/upstream/openai/transport/websocket.rs index 1d29c0a26..e5dbdcc84 100644 --- a/backend/tests/upstream/openai/transport/websocket.rs +++ b/backend/tests/upstream/openai/transport/websocket.rs @@ -1327,21 +1327,33 @@ async fn codex_backend_client_stream_should_preserve_burst_during_downstream_bac let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); let addr = listener.local_addr().unwrap(); + let (terminal_drained_tx, terminal_drained_rx) = tokio::sync::oneshot::channel(); let server = tokio::spawn(async move { let (stream, _) = listener.accept().await.unwrap(); let mut websocket = accept_codex_test_websocket(stream).await; let _message = websocket.next().await.unwrap().unwrap(); - websocket - .send(Message::Text( - json!({ - "type": "response.output_text.delta", - "delta": "initial;" - }) - .to_string() - .into(), - )) - .await - .unwrap(); + let (mut sink, mut source) = websocket.split(); + // 服务端持续读取控制帧，让 tungstenite 正常回 Pong。若客户端在本地入站背压 + // 期间仍执行 Pong deadline，它仍会因为暂时读不到已返回的 Pong 而误杀连接。 + let control_frames = tokio::spawn(async move { + while let Some(message) = source.next().await { + match message.unwrap() { + Message::Ping(_) | Message::Pong(_) => {} + Message::Close(_) => break, + other => panic!("unexpected client frame during response: {other:?}"), + } + } + }); + sink.send(Message::Text( + json!({ + "type": "response.output_text.delta", + "delta": "initial;" + }) + .to_string() + .into(), + )) + .await + .unwrap(); for index in 0..BURST_FRAMES { let frame = json!({ @@ -1349,22 +1361,34 @@ async fn codex_backend_client_stream_should_preserve_burst_during_downstream_bac "delta": format!("chunk-{index};") }) .to_string(); - if websocket.send(Message::Text(frame.into())).await.is_err() { + if sink.send(Message::Text(frame.into())).await.is_err() { + control_frames.abort(); return; } } - let _ = websocket + let _ = sink .send(Message::Text( completed_websocket_response("resp_backpressure", 3, 257).into(), )) .await; - let _ = websocket.close(None).await; + terminal_drained_rx + .await + .expect("client should drain the terminal response before server close"); + let _ = sink.send(Message::Close(None)).await; + control_frames.abort(); }); + let pool = Arc::new(CodexWebSocketPool::with_config(CodexWebSocketPoolConfig { + ping_interval: Some(Duration::from_millis(5)), + ping_timeout: Duration::from_millis(20), + liveness_timeout: None, + ..websocket_pool_config_for_tests(None, None, None) + })); let backend = CodexBackendClient::new( reqwest::Client::builder().no_proxy().build().unwrap(), format!("http://{addr}"), crate::support::fingerprint::runtime_test_fingerprint(), - ); + ) + .with_websocket_pool(pool); let request = pooled_websocket_request("conversation-backpressure"); let mut response = backend @@ -1386,6 +1410,7 @@ async fn codex_backend_client_stream_should_preserve_burst_during_downstream_bac }) .await .expect("backpressured websocket stream should finish"); + terminal_drained_tx.send(()).unwrap(); server.await.unwrap(); assert_eq!( diff --git a/backend/tests/upstream/openai/transport/websocket_pool.rs b/backend/tests/upstream/openai/transport/websocket_pool.rs index eaf943de3..71bd9b60d 100644 --- a/backend/tests/upstream/openai/transport/websocket_pool.rs +++ b/backend/tests/upstream/openai/transport/websocket_pool.rs @@ -714,29 +714,48 @@ async fn websocket_pool_should_use_http_for_new_keys_after_account_cap() { } #[tokio::test] -async fn websocket_pool_should_evict_idle_connection_when_liveness_lapses_despite_pings() { - // 不变量：pump 自己发出的 keepalive ping 不算“入站活动”，因此一个只收 ping、 - // 从不回 pong / 从不发帧的静默连接，仍会被 liveness watchdog 判定失活并驱逐； - // 复用前 acquire 读到 is_closed 即开新连接。 +async fn websocket_pool_should_replace_idle_connection_after_pong_deadline() { + // 服务端读取 Ping 后暂不继续 poll，避免 tungstenite 自动回 Pong；pump 必须在独立 + // deadline 到期时主动关闭连接，acquire 随后只读 closed 状态并直接新建连接。 let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); let addr = listener.local_addr().unwrap(); let accepted_connections = Arc::new(AtomicUsize::new(0)); let accepted_connections_for_server = Arc::clone(&accepted_connections); - let (release_first_tx, release_first_rx) = tokio::sync::oneshot::channel(); + let (ping_seen_tx, ping_seen_rx) = tokio::sync::oneshot::channel(); + let (inspect_close_tx, inspect_close_rx) = tokio::sync::oneshot::channel(); + let (pong_timeout_closed_tx, pong_timeout_closed_rx) = tokio::sync::oneshot::channel(); let server = tokio::spawn(async move { let (first_stream, _) = listener.accept().await.unwrap(); accepted_connections_for_server.fetch_add(1, Ordering::SeqCst); - let first_connection = tokio::spawn(async move { - let mut first_websocket = accept_codex_test_websocket(first_stream).await; - let _first_message = first_websocket.next().await.unwrap().unwrap(); - first_websocket - .send(Message::Text( - completed_websocket_response("resp_no_pong_first", 2, 1).into(), - )) - .await - .unwrap(); - let _ = release_first_rx.await; - }); + let mut first_websocket = accept_codex_test_websocket(first_stream).await; + let _first_message = first_websocket.next().await.unwrap().unwrap(); + first_websocket + .send(Message::Text( + completed_websocket_response("resp_no_pong_first", 2, 1).into(), + )) + .await + .unwrap(); + let ping = timeout(Duration::from_secs(1), first_websocket.next()) + .await + .expect("pump should send a keepalive ping") + .expect("keepalive ping should be present") + .expect("keepalive ping should be valid"); + let Message::Ping(payload) = ping else { + panic!("expected keepalive ping, got {ping:?}"); + }; + assert_eq!(payload.len(), 8); + ping_seen_tx.send(()).unwrap(); + + inspect_close_rx + .await + .expect("test should release the server after the Pong deadline"); + let close = timeout(Duration::from_secs(1), first_websocket.next()) + .await + .expect("Pong deadline should close the idle websocket") + .expect("Pong deadline should send a close frame") + .expect("close frame should be valid"); + std::assert_matches!(close, Message::Close(_)); + pong_timeout_closed_tx.send(()).unwrap(); let (second_stream, _) = listener.accept().await.unwrap(); accepted_connections_for_server.fetch_add(1, Ordering::SeqCst); @@ -749,11 +768,11 @@ async fn websocket_pool_should_evict_idle_connection_when_liveness_lapses_despit .await .unwrap(); second_websocket.close(None).await.unwrap(); - first_connection.await.unwrap(); }); let pool = Arc::new(CodexWebSocketPool::with_config(CodexWebSocketPoolConfig { - ping_interval: Some(Duration::from_millis(1)), - liveness_timeout: Some(Duration::from_millis(20)), + ping_interval: Some(Duration::from_millis(10)), + ping_timeout: Duration::from_millis(30), + liveness_timeout: None, maintenance_interval: None, ..websocket_pool_config_for_tests(None, None, None) })); @@ -772,12 +791,16 @@ async fn websocket_pool_should_evict_idle_connection_when_liveness_lapses_despit ) .await .expect("first websocket response should succeed"); - tokio::time::pause(); - tokio::time::advance(Duration::from_millis(25)).await; - tokio::task::yield_now().await; - // 只加速首条 idle 连接的失活窗口；恢复实时时钟，避免新连接在发送 - // response.create 前被测试时钟自动推进到下一次 liveness timeout。 - tokio::time::resume(); + timeout(Duration::from_secs(1), ping_seen_rx) + .await + .expect("server should observe the keepalive ping") + .expect("server should report the keepalive ping"); + tokio::time::sleep(Duration::from_millis(80)).await; + inspect_close_tx.send(()).unwrap(); + timeout(Duration::from_secs(1), pong_timeout_closed_rx) + .await + .expect("server should observe the Pong-timeout close") + .expect("server should report the Pong-timeout close"); let second = backend .create_response( &request, @@ -785,7 +808,6 @@ async fn websocket_pool_should_evict_idle_connection_when_liveness_lapses_despit ) .await .expect("second websocket response should use a fresh connection"); - release_first_tx.send(()).unwrap(); server.await.unwrap(); assert!(first.body.contains("resp_no_pong_first")); @@ -901,6 +923,7 @@ async fn codex_backend_client_should_keep_idle_pooled_websocket_alive_across_rep // pump 会在 idle 期间反复发送 keepalive ping；服务端计数并回 pong， // 直到下一个业务请求（response.create）到达为止。 + let mut previous_ping = None; loop { let message = timeout(Duration::from_secs(1), websocket.next()) .await @@ -909,6 +932,11 @@ async fn codex_backend_client_should_keep_idle_pooled_websocket_alive_across_rep .expect("frame should be valid"); match message { Message::Ping(payload) => { + assert_eq!(payload.len(), 8); + if let Some(previous) = &previous_ping { + assert_ne!(previous, &payload, "keepalive ping sequence must be unique"); + } + previous_ping = Some(payload.clone()); ping_count_for_server.fetch_add(1, Ordering::SeqCst); websocket.send(Message::Pong(payload)).await.unwrap(); } @@ -978,7 +1006,66 @@ async fn codex_backend_client_should_keep_idle_pooled_websocket_alive_across_rep assert!(first.body.contains("resp_pool_background_first")); assert!(second.body.contains("resp_pool_background_second")); - assert!(ping_count.load(Ordering::SeqCst) >= 1); + assert!(ping_count.load(Ordering::SeqCst) >= 2); +} + +#[tokio::test] +async fn codex_backend_client_should_treat_active_business_frames_as_ping_liveness() { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let server = tokio::spawn(async move { + let (stream, _) = listener.accept().await.unwrap(); + let mut websocket = accept_codex_test_websocket(stream).await; + let _request = websocket.next().await.unwrap().unwrap(); + + for index in 0..8 { + tokio::time::sleep(Duration::from_millis(20)).await; + websocket + .send(Message::Text( + json!({ + "type": "response.output_text.delta", + "delta": format!("active-{index};") + }) + .to_string() + .into(), + )) + .await + .unwrap(); + } + websocket + .send(Message::Text( + completed_websocket_response("resp_active_without_pong", 3, 8).into(), + )) + .await + .unwrap(); + websocket.close(None).await.unwrap(); + }); + let pool = Arc::new(CodexWebSocketPool::with_config(CodexWebSocketPoolConfig { + ping_interval: Some(Duration::from_millis(10)), + ping_timeout: Duration::from_millis(60), + liveness_timeout: None, + ..websocket_pool_config_for_tests(None, None, None) + })); + let backend = CodexBackendClient::new( + reqwest::Client::builder().no_proxy().build().unwrap(), + format!("http://{addr}"), + crate::support::fingerprint::runtime_test_fingerprint(), + ) + .with_websocket_pool(Arc::clone(&pool)); + let request = pooled_websocket_request("conversation-active-without-pong"); + + let response = backend + .create_response( + &request, + request_context("req_active_without_pong", Some("chatgpt-account")), + ) + .await + .expect("active business frames should satisfy the Ping liveness probe"); + server.await.unwrap(); + + assert!(response.body.contains("active-0;")); + assert!(response.body.contains("active-7;")); + assert!(response.body.contains("resp_active_without_pong")); } #[tokio::test] diff --git a/docs/architecture.md b/docs/architecture.md index fa16a34f2..6820d6a21 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -277,7 +277,7 @@ Pool key 为 `(base_url, account_id, local_conversation_id)`，slot 状态为： 每条连接记录最新 completed response ID。Exact continuation 只有 ID 完全匹配时才能租用；socket 缺失、busy、失活或 ID 不匹配时不新建连接碰运气，由 HistoryController 决定 full replay 或失败。 -池化 socket 默认 25 秒 ping、5 秒探活、55 分钟最大寿命；后台 pump 处理 ping/pong、EOF、close 和 liveness，复用前只读取连接状态。 +池化 socket 默认每 25 秒发送带唯一序列的 Ping，匹配 Pong 的 deadline 为 5 秒，最大寿命为 55 分钟。任意业务入站帧同样证明链路存活；下游背压导致 pump 无法读取 socket 时暂停主动探活。后台 pump 统一处理 ping/pong、EOF、close 和 liveness，复用前只读取原子连接状态，不增加网络往返。 ### Origin breaker diff --git a/frontend/src/views/accounts/components/AccountIdentityCell.vue b/frontend/src/views/accounts/components/AccountIdentityCell.vue index c43f53e93..0c7e6e98a 100644 --- a/frontend/src/views/accounts/components/AccountIdentityCell.vue +++ b/frontend/src/views/accounts/components/AccountIdentityCell.vue @@ -61,9 +61,9 @@ const avatarClass = computed(() => { > {{ initial }} -
+
- + {{ displayTitle }} diff --git a/frontend/src/views/accounts/components/AccountPlanBadge.vue b/frontend/src/views/accounts/components/AccountPlanBadge.vue index d7988ba83..a7ba22c7d 100644 --- a/frontend/src/views/accounts/components/AccountPlanBadge.vue +++ b/frontend/src/views/accounts/components/AccountPlanBadge.vue @@ -23,8 +23,8 @@ const label = computed(() => props.planType?.trim() || 'Free') const sizeClass = computed(() => props.size === 'sm' - ? 'h-5 max-w-24 rounded-full px-1.75 text-[11px] font-[720]' - : 'h-5.5 max-w-full rounded-full px-2 text-[11px] font-[760]', + ? 'h-5 rounded-full px-1.75 text-[11px] font-[720]' + : 'h-5.5 rounded-full px-2 text-[11px] font-[760]', ) const paletteClass = computed(() => { @@ -39,9 +39,9 @@ const paletteClass = computed(() => { diff --git a/frontend/src/views/usage/components/UsageDiagnosticCard.vue b/frontend/src/views/usage/components/UsageDiagnosticCard.vue index 38b43e56f..2497de8b0 100644 --- a/frontend/src/views/usage/components/UsageDiagnosticCard.vue +++ b/frontend/src/views/usage/components/UsageDiagnosticCard.vue @@ -80,7 +80,7 @@ const tableRows = computed(() => (isStale.value ? [] : displayItems.value)) function diagnosticNameDisplay(name: string) { const full = name.trim() || '未知' - if (dimension.value !== 'model') { + if (dimension.value !== 'model' && dimension.value !== 'account') { return { primary: full, secondary: '', full } } diff --git a/frontend/src/views/usage/composables/useUsageRecordsTable.ts b/frontend/src/views/usage/composables/useUsageRecordsTable.ts index 0621da1c2..45b70cfce 100644 --- a/frontend/src/views/usage/composables/useUsageRecordsTable.ts +++ b/frontend/src/views/usage/composables/useUsageRecordsTable.ts @@ -1,5 +1,6 @@ import { clamp } from 'es-toolkit' import { onMounted, shallowRef, watch } from 'vue' +import type { Ref } from 'vue' import { getUsageRecordInsightsDiagnostics, @@ -10,7 +11,20 @@ import { import { toast } from '@/components/base/BaseToast' import { withMinimumDuration } from '@/utils/async' -export function useUsageRecordsTable(options: any) { +import type { UsageTimeRangeParams } from './useUsageTimeRange' + +interface UseUsageRecordsTableOptions { + page: Ref + pageSize: Ref + searchQuery: Ref + timeRangeParams: Readonly> + totalRecords: Ref + refreshTimeRangeEnd: () => void +} + +type UsageLoadScope = 'all' | 'table' + +export function useUsageRecordsTable(options: UseUsageRecordsTableOptions) { const loading = shallowRef(true) const analyticsLoading = shallowRef(true) const records = shallowRef([]) @@ -26,7 +40,7 @@ export function useUsageRecordsTable(options: any) { search: options.searchQuery.value || undefined, }) - async function loadUsageRecords(scope = 'all') { + async function loadUsageRecords(scope: UsageLoadScope = 'all') { const requestId = ++loadRequestId try { loading.value = true @@ -135,7 +149,8 @@ export function useUsageRecordsTable(options: any) { if (refreshingList.value || loading.value) return refreshingList.value = true try { - await withMinimumDuration(() => loadUsageRecords('table')) + options.refreshTimeRangeEnd() + await withMinimumDuration(() => loadUsageRecords('all')) } finally { refreshingList.value = false } diff --git a/frontend/src/views/usage/composables/useUsageTimeRange.ts b/frontend/src/views/usage/composables/useUsageTimeRange.ts new file mode 100644 index 000000000..be19ccb32 --- /dev/null +++ b/frontend/src/views/usage/composables/useUsageTimeRange.ts @@ -0,0 +1,40 @@ +import dayjs, { type Dayjs } from 'dayjs' +import { computed, shallowRef } from 'vue' + +export type UsageTimeRange = 'today' | '7d' | '30d' + +export interface UsageTimeRangeParams extends Record { + startTime: string + endTime: string +} + +export function useUsageTimeRange(initialRange: UsageTimeRange = 'today') { + const timeRange = shallowRef(initialRange) + const rangeEnd = shallowRef(dayjs()) + const timeRangeParams = computed(() => buildTimeRange(timeRange.value, rangeEnd.value)) + + function refreshTimeRangeEnd() { + rangeEnd.value = dayjs() + } + + return { + timeRange, + timeRangeParams, + refreshTimeRangeEnd, + } +} + +function buildTimeRange(range: UsageTimeRange, end: Dayjs): UsageTimeRangeParams { + if (range === 'today') { + return { + startTime: end.startOf('day').toISOString(), + endTime: end.toISOString(), + } + } + + const days = range === '30d' ? 29 : 6 + return { + startTime: end.subtract(days, 'day').startOf('day').toISOString(), + endTime: end.toISOString(), + } +} diff --git a/frontend/src/views/usage/index.vue b/frontend/src/views/usage/index.vue index f0cd655d1..ba7a03902 100644 --- a/frontend/src/views/usage/index.vue +++ b/frontend/src/views/usage/index.vue @@ -1,7 +1,6 @@