fix(data): 同步模型调用次数至 global_models 读模型

- 模型列表改为读取 global_models.usage_count,避免每次扫描 usage 明细表
- 通过历史 backfill 与 usage upsert delta 维护该读模型
- 详情页保留实时 facts 兜底,统一排除 pending/streaming 状态
This commit is contained in:
mayrain
2026-05-05 17:34:02 +08:00
parent 53fa33b0c5
commit f0197b685f
6 changed files with 347 additions and 11 deletions

View File

@@ -89,6 +89,41 @@ impl ApiKeyUsageDelta {
}
}
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub(crate) struct ModelUsageContribution {
pub model: String,
pub request_count: i64,
}
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub(crate) struct ModelUsageDelta {
pub request_count: i64,
}
impl ModelUsageDelta {
pub(crate) fn between(before: &ModelUsageContribution, after: &ModelUsageContribution) -> Self {
Self {
request_count: after.request_count - before.request_count,
}
}
pub(crate) fn addition(after: &ModelUsageContribution) -> Self {
Self {
request_count: after.request_count,
}
}
pub(crate) fn removal(before: &ModelUsageContribution) -> Self {
Self {
request_count: -before.request_count,
}
}
pub(crate) fn is_noop(&self) -> bool {
self.request_count == 0
}
}
#[derive(Debug, Clone, PartialEq, Default)]
pub(crate) struct ProviderApiKeyUsageContribution {
pub key_id: String,
@@ -265,6 +300,23 @@ pub(crate) fn provider_api_key_usage_contribution(
})
}
pub(crate) fn model_usage_contribution(
usage: &StoredRequestUsageAudit,
) -> Option<ModelUsageContribution> {
if matches!(usage.status.as_str(), "pending" | "streaming") {
return None;
}
let model = usage.model.trim();
if model.is_empty() {
return None;
}
Some(ModelUsageContribution {
model: model.to_string(),
request_count: 1,
})
}
pub(crate) fn api_key_usage_contribution(
usage: &StoredRequestUsageAudit,
) -> Option<ApiKeyUsageContribution> {
@@ -292,9 +344,10 @@ pub(crate) fn api_key_usage_contribution(
mod tests {
use super::{
api_key_usage_contribution, incoming_usage_can_recover_terminal_failure,
provider_api_key_usage_contribution, provider_api_key_usage_is_error,
provider_api_key_usage_is_success, strip_deprecated_usage_display_fields,
usage_can_recover_terminal_failure, StoredRequestUsageAudit, UpsertUsageRecord,
model_usage_contribution, provider_api_key_usage_contribution,
provider_api_key_usage_is_error, provider_api_key_usage_is_success,
strip_deprecated_usage_display_fields, usage_can_recover_terminal_failure, ModelUsageDelta,
StoredRequestUsageAudit, UpsertUsageRecord,
};
#[test]
@@ -570,4 +623,75 @@ mod tests {
assert_eq!(contribution.total_cost_usd, 0.25);
assert_eq!(contribution.last_used_at_unix_secs, Some(123));
}
#[test]
fn model_usage_contribution_tracks_terminal_requests_only() {
let completed = StoredRequestUsageAudit::new(
"usage-1".to_string(),
"request-1".to_string(),
Some("user-1".to_string()),
Some("api-key-1".to_string()),
None,
None,
"OpenAI".to_string(),
" gpt-5.5 ".to_string(),
None,
Some("provider-1".to_string()),
None,
None,
None,
None,
None,
None,
None,
None,
None,
false,
false,
12,
8,
20,
0.25,
0.25,
Some(200),
None,
None,
Some(120),
None,
"completed".to_string(),
"settled".to_string(),
123,
124,
Some(125),
)
.expect("usage should build");
let contribution =
model_usage_contribution(&completed).expect("completed usage should count");
assert_eq!(contribution.model, "gpt-5.5");
assert_eq!(contribution.request_count, 1);
let mut streaming = completed.clone();
streaming.status = "streaming".to_string();
assert!(model_usage_contribution(&streaming).is_none());
let mut pending = completed;
pending.status = "pending".to_string();
assert!(model_usage_contribution(&pending).is_none());
}
#[test]
fn model_usage_delta_handles_model_changes() {
let before = super::ModelUsageContribution {
model: "gpt-5.4".to_string(),
request_count: 1,
};
let after = super::ModelUsageContribution {
model: "gpt-5.5".to_string(),
request_count: 1,
};
assert_eq!(ModelUsageDelta::removal(&before).request_count, -1);
assert_eq!(ModelUsageDelta::addition(&after).request_count, 1);
assert!(ModelUsageDelta::between(&before, &before).is_noop());
}
}

View File

@@ -36,7 +36,8 @@ use uuid::Uuid;
use super::{
api_key_usage_contribution, incoming_usage_can_recover_terminal_failure,
provider_api_key_usage_contribution, strip_deprecated_usage_display_fields, ApiKeyUsageDelta,
model_usage_contribution, provider_api_key_usage_contribution,
strip_deprecated_usage_display_fields, ApiKeyUsageDelta, ModelUsageDelta,
ProviderApiKeyUsageDelta, StoredProviderApiKeyUsageSummary, StoredProviderUsageSummary,
StoredRequestUsageAudit, StoredUsageDailySummary, UpsertUsageRecord, UsageAuditListQuery,
UsageDailyHeatmapQuery, UsageReadRepository, UsageWriteRepository,
@@ -1219,6 +1220,9 @@ const REBUILD_API_KEY_USAGE_STATS_SQL: &str =
const APPLY_PROVIDER_API_KEY_USAGE_DELTA_SQL: &str =
include_str!("queries/apply_provider_api_key_usage_delta_sql.sql");
const APPLY_GLOBAL_MODEL_USAGE_DELTA_SQL: &str =
include_str!("queries/apply_global_model_usage_delta_sql.sql");
const RESET_PROVIDER_API_KEY_USAGE_STATS_SQL: &str =
include_str!("queries/reset_provider_api_key_usage_stats_sql.sql");
@@ -7268,6 +7272,40 @@ ORDER BY "usage".user_id ASC
}
}
}
let before_model_contribution =
previous_usage.as_ref().and_then(model_usage_contribution);
let after_model_contribution = model_usage_contribution(&stored);
match (
before_model_contribution.as_ref(),
after_model_contribution.as_ref(),
) {
(Some(before), Some(after)) if before.model == after.model => {
let delta = ModelUsageDelta::between(before, after);
apply_global_model_usage_delta_in_tx(tx, before.model.as_str(), &delta)
.await?;
}
_ => {
if let Some(before) = before_model_contribution.as_ref() {
let delta = ModelUsageDelta::removal(before);
apply_global_model_usage_delta_in_tx(
tx,
before.model.as_str(),
&delta,
)
.await?;
}
if let Some(after) = after_model_contribution.as_ref() {
let delta = ModelUsageDelta::addition(after);
apply_global_model_usage_delta_in_tx(
tx,
after.model.as_str(),
&delta,
)
.await?;
}
}
}
Ok(stored)
}) as BoxFuture<'_, Result<StoredRequestUsageAudit, DataLayerError>>
})
@@ -7686,6 +7724,33 @@ async fn apply_provider_api_key_usage_delta_in_tx(
Ok(())
}
async fn apply_global_model_usage_delta_in_tx(
tx: &mut sqlx::Transaction<'_, Postgres>,
model: &str,
delta: &ModelUsageDelta,
) -> Result<(), DataLayerError> {
let model = model.trim();
if model.is_empty() {
return Ok(());
}
if delta.is_noop() {
return Ok(());
}
sqlx::query(APPLY_GLOBAL_MODEL_USAGE_DELTA_SQL)
.bind(model)
.bind(i32::try_from(delta.request_count).map_err(|_| {
DataLayerError::UnexpectedValue(format!(
"global_models.usage_count delta exceeds i32: {}",
delta.request_count
))
})?)
.execute(&mut **tx)
.await
.map_postgres_err()?;
Ok(())
}
// Build the usage read model from the split storage layout.
//
// Query projections already prefer the newer audit/snapshot owners and only fall back to

View File

@@ -0,0 +1,5 @@
UPDATE global_models
SET
usage_count = GREATEST(COALESCE(usage_count, 0) + $2, 0),
updated_at = NOW()
WHERE name = $1