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 @@