perf(proxy): trim per-request hot-path work and db wait

- Guard debug body serialization with `log::log_enabled!`; previously
  serialized the filtered body to a throwaway String on every forward,
  even with debug logging off.
- Skip SSE parse + UTF-8 buffer loop when no usage collector and debug
  is off; the per-chunk `serde_json::from_str::<Value>` ran even in
  pure passthrough mode.
- Add cheap per-app SSE event pre-filter (string `contains`) so usage
  collectors only parse events that could contain usage (e.g. Claude
  `message_start` / `message_delta`).
- Skip non-streaming response body JSON parse when usage logging is
  disabled.
- Move `ProviderRouter::record_result` off the success response path
  via `tokio::spawn` for non-HalfOpen state; that call internally does
  `get_proxy_config_for_app` + `update_provider_health`, two SQLite
  ops that previously blocked TTFB.

Also: dedupe `usage_logging_enabled` (was duplicated in handlers.rs)
and merge `SseUsageCollector::{new, new_filtered}` into a single
constructor that takes `Option<StreamUsageEventFilter>`.
This commit is contained in:
Jason
2026-05-11 09:35:56 +08:00
parent 00a789e7a3
commit 4b384dfe55
4 changed files with 290 additions and 187 deletions
+43 -32
View File
@@ -10,7 +10,8 @@
use super::{
error_mapper::{get_error_message, map_proxy_error_to_status},
handler_config::{
CLAUDE_PARSER_CONFIG, CODEX_PARSER_CONFIG, GEMINI_PARSER_CONFIG, OPENAI_PARSER_CONFIG,
claude_stream_usage_event_filter, CLAUDE_PARSER_CONFIG, CODEX_PARSER_CONFIG,
GEMINI_PARSER_CONFIG, OPENAI_PARSER_CONFIG,
},
handler_context::RequestContext,
providers::{
@@ -22,7 +23,7 @@ use super::{
response_processor::{
create_logged_passthrough_stream, process_response, read_decoded_body,
strip_entity_headers_for_rebuilt_body, strip_hop_by_hop_response_headers,
SseUsageCollector,
usage_logging_enabled, SseUsageCollector,
},
server::ProxyState,
sse::{strip_sse_field, take_sse_block},
@@ -270,8 +271,8 @@ async fn handle_claude_transform(
Box::new(Box::pin(create_anthropic_sse_stream(stream)))
};
// 创建使用量收集器
let usage_collector = {
// 创建使用量收集器;关闭 usage logging 时不要再解析转换后的 SSE。
let usage_collector = if usage_logging_enabled(state) {
let state = state.clone();
let provider_id = ctx.provider.id.clone();
let model = ctx.request_model.clone();
@@ -279,34 +280,40 @@ async fn handle_claude_transform(
let start_time = ctx.start_time;
let session_id = ctx.session_id.clone();
SseUsageCollector::new(start_time, move |events, first_token_ms| {
if let Some(usage) = TokenUsage::from_claude_stream_events(&events) {
let latency_ms = start_time.elapsed().as_millis() as u64;
let state = state.clone();
let provider_id = provider_id.clone();
let model = model.clone();
let session_id = session_id.clone();
Some(SseUsageCollector::new(
start_time,
Some(claude_stream_usage_event_filter),
move |events, first_token_ms| {
if let Some(usage) = TokenUsage::from_claude_stream_events(&events) {
let latency_ms = start_time.elapsed().as_millis() as u64;
let state = state.clone();
let provider_id = provider_id.clone();
let model = model.clone();
let session_id = session_id.clone();
tokio::spawn(async move {
log_usage(
&state,
&provider_id,
"claude",
&model,
&model,
usage,
latency_ms,
first_token_ms,
true,
status_code,
Some(session_id),
)
.await;
});
} else {
log::debug!("[Claude] OpenRouter 流式响应缺少 usage 统计,跳过消费记录");
}
})
tokio::spawn(async move {
log_usage(
&state,
&provider_id,
"claude",
&model,
&model,
usage,
latency_ms,
first_token_ms,
true,
status_code,
Some(session_id),
)
.await;
});
} else {
log::debug!("[Claude] OpenRouter 流式响应缺少 usage 统计,跳过消费记录");
}
},
))
} else {
None
};
// 获取流式超时配置
@@ -315,7 +322,7 @@ async fn handle_claude_transform(
let logged_stream = create_logged_passthrough_stream(
sse_stream,
"Claude/OpenRouter",
Some(usage_collector),
usage_collector,
timeout_config,
);
@@ -798,6 +805,10 @@ async fn log_usage(
) {
use super::usage::logger::UsageLogger;
if !usage_logging_enabled(state) {
return;
}
let logger = UsageLogger::new(&state.db);
let (multiplier, pricing_model_source) =