fix(proxy): add stable usage keys and idempotent raw-response logging

Derive request ids from upstream envelope ids (Codex/OpenAI top-level
id, Gemini responseId, Claude message id with non-empty filtering)
scoped as session:{app_type}:{provider_id}:{id} for non-Claude sources;
Claude keeps bare session:{id} to preserve session-log convergence.

The logger now queries and conditionally writes under a single
connection guard: identical semantic replays return without writing or
notifying, session_log rows may be upgraded by proxy, and same-id
different-semantic responses land on a deterministic SHA-256 collision
fallback key instead of being overwritten. Fixes the random-UUID
INSERT OR REPLACE duplication behind #5496.
This commit is contained in:
Jason
2026-07-20 12:16:57 +08:00
parent df3e07edbb
commit c9ac6efd69
5 changed files with 385 additions and 54 deletions
+289 -34
View File
@@ -6,9 +6,59 @@ use crate::database::{Database, PRICING_SOURCE_REQUEST, PRICING_SOURCE_RESPONSE}
use crate::error::AppError;
use crate::services::sql_helpers::{INPUT_TOKEN_SEMANTICS_FRESH, INPUT_TOKEN_SEMANTICS_TOTAL};
use crate::services::usage_stats::{find_model_pricing_row, is_placeholder_pricing_model};
use rusqlite::OptionalExtension;
use rust_decimal::Decimal;
use sha2::{Digest, Sha256};
use std::str::FromStr;
#[derive(Debug, PartialEq, Eq)]
struct UsageSemantic {
app_type: String,
provider_id: String,
model: String,
input_token_semantics: i64,
input_tokens: u32,
output_tokens: u32,
cache_read_tokens: u32,
cache_creation_tokens: u32,
status_code: u16,
}
impl UsageSemantic {
fn from_log(log: &RequestLog, input_token_semantics: i64) -> Self {
Self {
app_type: log.app_type.clone(),
provider_id: log.provider_id.clone(),
model: log.model.clone(),
input_token_semantics,
input_tokens: log.usage.input_tokens,
output_tokens: log.usage.output_tokens,
cache_read_tokens: log.usage.cache_read_tokens,
cache_creation_tokens: log.usage.cache_creation_tokens,
status_code: log.status_code,
}
}
fn sha256(&self) -> String {
let encoded = serde_json::to_vec(&(
&self.app_type,
&self.provider_id,
&self.model,
self.input_token_semantics,
self.input_tokens,
self.output_tokens,
self.cache_read_tokens,
self.cache_creation_tokens,
self.status_code,
))
.expect("usage semantic tuple is serializable");
Sha256::digest(encoded)
.iter()
.map(|byte| format!("{byte:02x}"))
.collect()
}
}
/// 请求日志
#[derive(Debug, Clone)]
pub struct RequestLog {
@@ -77,52 +127,132 @@ impl<'a> UsageLogger<'a> {
} else {
INPUT_TOKEN_SEMANTICS_FRESH
};
let semantic = UsageSemantic::from_log(log, input_token_semantics);
let existing = Self::load_existing_semantic(&conn, &log.request_id)?;
conn.execute(
"INSERT OR REPLACE INTO proxy_request_logs (
let (request_id, replace_session_log, collision) = match existing {
None => (log.request_id.clone(), false, false),
Some((data_source, _existing_semantic))
if data_source.as_deref() == Some("session_log") =>
{
(log.request_id.clone(), true, false)
}
Some((data_source, existing_semantic))
if data_source.as_deref().unwrap_or("proxy") == "proxy"
&& existing_semantic == semantic =>
{
return Ok(());
}
Some(_) => {
let fallback = format!("{}:collision:{}", log.request_id, semantic.sha256());
if let Some((data_source, existing_semantic)) =
Self::load_existing_semantic(&conn, &fallback)?
{
if data_source.as_deref().unwrap_or("proxy") == "proxy"
&& existing_semantic == semantic
{
return Ok(());
}
return Err(AppError::Database(format!(
"usage collision fallback 主键发生 SHA-256 冲突: {fallback}"
)));
}
(fallback, false, true)
}
};
let insert_verb = if replace_session_log {
"INSERT OR REPLACE"
} else {
"INSERT OR IGNORE"
};
let sql = format!(
"{insert_verb} INTO proxy_request_logs (
request_id, provider_id, app_type, model, request_model, pricing_model,
input_tokens, output_tokens, cache_read_tokens, cache_creation_tokens,
input_token_semantics,
input_cost_usd, output_cost_usd, cache_read_cost_usd, cache_creation_cost_usd, total_cost_usd,
latency_ms, first_token_ms, status_code, error_message, session_id,
provider_type, is_streaming, cost_multiplier, created_at
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16, ?17, ?18, ?19, ?20, ?21, ?22, ?23, ?24, ?25)",
rusqlite::params![
log.request_id,
log.provider_id,
log.app_type,
log.model,
log.request_model,
log.pricing_model,
log.usage.input_tokens,
log.usage.output_tokens,
log.usage.cache_read_tokens,
log.usage.cache_creation_tokens,
input_token_semantics,
input_cost,
output_cost,
cache_read_cost,
cache_creation_cost,
total_cost,
log.latency_ms as i64,
log.first_token_ms.map(|v| v as i64),
log.status_code as i64,
log.error_message,
log.session_id,
log.provider_type,
log.is_streaming as i64,
log.cost_multiplier,
created_at,
],
)
.map_err(|e| AppError::Database(format!("记录请求日志失败: {e}")))?;
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16, ?17, ?18, ?19, ?20, ?21, ?22, ?23, ?24, ?25)"
);
let affected_rows = conn
.execute(
&sql,
rusqlite::params![
request_id,
log.provider_id,
log.app_type,
log.model,
log.request_model,
log.pricing_model,
log.usage.input_tokens,
log.usage.output_tokens,
log.usage.cache_read_tokens,
log.usage.cache_creation_tokens,
input_token_semantics,
input_cost,
output_cost,
cache_read_cost,
cache_creation_cost,
total_cost,
log.latency_ms as i64,
log.first_token_ms.map(|v| v as i64),
log.status_code as i64,
log.error_message,
log.session_id,
log.provider_type,
log.is_streaming as i64,
log.cost_multiplier,
created_at,
],
)
.map_err(|e| AppError::Database(format!("记录请求日志失败: {e}")))?;
// 通知前端使用统计有更新(200ms 防抖合并,不阻塞写入路径)
crate::usage_events::notify_log_recorded();
if affected_rows > 0 {
if collision {
log::warn!(
"usage request_id collision: primary={}, fallback={request_id}",
log.request_id
);
}
crate::usage_events::notify_log_recorded();
}
Ok(())
}
fn load_existing_semantic(
conn: &rusqlite::Connection,
request_id: &str,
) -> Result<Option<(Option<String>, UsageSemantic)>, AppError> {
conn.query_row(
"SELECT data_source, app_type, provider_id, model, input_token_semantics,
input_tokens, output_tokens, cache_read_tokens,
cache_creation_tokens, status_code
FROM proxy_request_logs WHERE request_id = ?1",
[request_id],
|row| {
Ok((
row.get(0)?,
UsageSemantic {
app_type: row.get(1)?,
provider_id: row.get(2)?,
model: row.get(3)?,
input_token_semantics: row.get(4)?,
input_tokens: row.get::<_, i64>(5)? as u32,
output_tokens: row.get::<_, i64>(6)? as u32,
cache_read_tokens: row.get::<_, i64>(7)? as u32,
cache_creation_tokens: row.get::<_, i64>(8)? as u32,
status_code: row.get::<_, i64>(9)? as u16,
},
))
},
)
.optional()
.map_err(|error| AppError::Database(format!("查询 usage request_id 失败: {error}")))
}
/// 记录失败的请求
///
/// 用于记录无法从上游获取 usage 信息的失败请求
@@ -375,6 +505,34 @@ impl<'a> UsageLogger<'a> {
mod tests {
use super::*;
fn request_log(request_id: &str, input_tokens: u32) -> RequestLog {
RequestLog {
request_id: request_id.to_string(),
provider_id: "provider-1".to_string(),
app_type: "codex".to_string(),
model: "gpt-5.6".to_string(),
request_model: "gpt-5.6".to_string(),
pricing_model: "gpt-5.6".to_string(),
usage: TokenUsage {
input_tokens,
output_tokens: 5,
cache_read_tokens: 2,
cache_creation_tokens: 0,
model: None,
message_id: Some("resp-1".to_string()),
},
cost: None,
latency_ms: 10,
first_token_ms: Some(2),
status_code: 200,
error_message: None,
session_id: None,
provider_type: Some("codex".to_string()),
is_streaming: true,
cost_multiplier: "1".to_string(),
}
}
#[test]
fn test_log_request() -> Result<(), AppError> {
let db = Database::memory()?;
@@ -432,6 +590,103 @@ mod tests {
Ok(())
}
#[test]
fn identical_replay_writes_and_notifies_once() -> Result<(), AppError> {
let db = Database::memory()?;
let logger = UsageLogger::new(&db);
crate::usage_events::take_test_notify_count();
let log = request_log("stable-id", 10);
logger.log_request(&log)?;
logger.log_request(&log)?;
let conn = crate::database::lock_conn!(db.conn);
let count: i64 = conn.query_row(
"SELECT COUNT(*) FROM proxy_request_logs WHERE request_id = 'stable-id'",
[],
|row| row.get(0),
)?;
assert_eq!(count, 1);
assert_eq!(crate::usage_events::take_test_notify_count(), 1);
Ok(())
}
#[test]
fn semantic_collision_uses_deterministic_idempotent_fallback() -> Result<(), AppError> {
let db = Database::memory()?;
let logger = UsageLogger::new(&db);
crate::usage_events::take_test_notify_count();
let first = request_log("shared-id", 10);
let second = request_log("shared-id", 20);
logger.log_request(&first)?;
logger.log_request(&second)?;
logger.log_request(&second)?;
let conn = crate::database::lock_conn!(db.conn);
let rows: Vec<(String, i64)> = conn
.prepare(
"SELECT request_id, input_tokens FROM proxy_request_logs ORDER BY input_tokens",
)?
.query_map([], |row| Ok((row.get(0)?, row.get(1)?)))?
.collect::<Result<_, _>>()?;
assert_eq!(rows.len(), 2);
assert_eq!(rows[0], ("shared-id".to_string(), 10));
assert!(rows[1].0.starts_with("shared-id:collision:"));
assert_eq!(rows[1].1, 20);
assert_eq!(crate::usage_events::take_test_notify_count(), 2);
Ok(())
}
#[test]
fn only_session_log_primary_rows_may_be_replaced() -> Result<(), AppError> {
let db = Database::memory()?;
{
let conn = crate::database::lock_conn!(db.conn);
for (request_id, data_source) in [
("session-primary", "session_log"),
("codex-primary", "codex_session"),
] {
conn.execute(
"INSERT INTO proxy_request_logs (
request_id, provider_id, app_type, model, input_tokens,
output_tokens, cache_read_tokens, cache_creation_tokens,
latency_ms, status_code, created_at, data_source
) VALUES (?1, '_session', 'claude', 'old', 1, 1, 0, 0, 0, 200, 1, ?2)",
rusqlite::params![request_id, data_source],
)?;
}
}
let logger = UsageLogger::new(&db);
let mut session_replacement = request_log("session-primary", 10);
session_replacement.app_type = "claude".to_string();
logger.log_request(&session_replacement)?;
logger.log_request(&request_log("codex-primary", 20))?;
let conn = crate::database::lock_conn!(db.conn);
let session_source: String = conn.query_row(
"SELECT data_source FROM proxy_request_logs WHERE request_id = 'session-primary'",
[],
|row| row.get(0),
)?;
let codex_input: i64 = conn.query_row(
"SELECT input_tokens FROM proxy_request_logs WHERE request_id = 'codex-primary'",
[],
|row| row.get(0),
)?;
let fallback_count: i64 = conn.query_row(
"SELECT COUNT(*) FROM proxy_request_logs
WHERE request_id LIKE 'codex-primary:collision:%'",
[],
|row| row.get(0),
)?;
assert_eq!(session_source, "proxy");
assert_eq!(codex_input, 1);
assert_eq!(fallback_count, 1);
Ok(())
}
#[test]
fn test_log_error() -> Result<(), AppError> {
let db = Database::memory()?;