diff --git a/backend/crates/gateway-core/src/diagnostics/mod.rs b/backend/crates/gateway-core/src/diagnostics/mod.rs index 1f5492ea1..1882fd21a 100644 --- a/backend/crates/gateway-core/src/diagnostics/mod.rs +++ b/backend/crates/gateway-core/src/diagnostics/mod.rs @@ -9,4 +9,4 @@ mod trace; pub use capture::{body_fingerprint, diagnostic_headers, diagnostic_json, json_fingerprint}; pub use failure::{OperationalDiagnostics, OperationalFailure}; pub use stream::{StreamCapture, StreamFormat}; -pub use trace::TraceContext; +pub use trace::{MAX_TRACE_SNAPSHOT_BYTES, TraceContext}; diff --git a/backend/crates/gateway-core/src/diagnostics/trace.rs b/backend/crates/gateway-core/src/diagnostics/trace.rs index 4f7c9fe7a..366ef46d5 100644 --- a/backend/crates/gateway-core/src/diagnostics/trace.rs +++ b/backend/crates/gateway-core/src/diagnostics/trace.rs @@ -15,13 +15,16 @@ use crate::error::ProviderError; use super::capture::{bounded, diagnostic_event_json, diagnostic_event_type}; use super::{body_fingerprint, diagnostic_headers}; +/// 完整诊断快照的编码上限,生产者与持久化适配器共用 +pub const MAX_TRACE_SNAPSHOT_BYTES: usize = 128 * 1024; + const MAX_EVENTS: usize = 128; const HEAD_EVENTS: usize = 8; const MAX_DATA_BYTES: usize = 4096; -const MAX_BUFFER_BYTES: usize = 64 * 1024; +const MAX_BUFFER_BYTES: usize = MAX_TRACE_SNAPSHOT_BYTES / 2; const MAX_EVIDENCE_EVENTS: usize = 64; const MAX_EVIDENCE_DATA_BYTES: usize = 16 * 1024; -const MAX_EVIDENCE_BUFFER_BYTES: usize = 64 * 1024; +const MAX_EVIDENCE_BUFFER_BYTES: usize = MAX_TRACE_SNAPSHOT_BYTES / 2; /// 显式传给异步任务的关联上下文;默认值禁用捕获,避免全局请求映射 #[derive(Clone, Default)] @@ -284,27 +287,22 @@ impl TraceContext { /// 与请求终态原子持久化;旧记录没有此字段,不伪造历史时间线 #[must_use] pub fn snapshot(&self) -> Option { - let state = self + let mut state = self .state .as_ref()? .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); - Some(json!({ - "schemaVersion": 3, "requestId": state.request_id, - "startedAtUnixMs": state.started_at_ms, "totalEvents": state.sequence, - "droppedEvents": state.dropped_events, "maxEvents": MAX_EVENTS, - "captureMode": "sanitized", "events": state.events, - "wireDumpEnabled": tracing::enabled!(target: "request_dump", tracing::Level::INFO), - "wireFrames": state.wire_frames, "wireBytes": state.wire_bytes, - "maxBufferBytes": MAX_BUFFER_BYTES + MAX_EVIDENCE_BUFFER_BYTES, "headEvents": HEAD_EVENTS, - "maxEvidenceEvents": MAX_EVIDENCE_EVENTS, - "maxEvidenceDataBytes": MAX_EVIDENCE_DATA_BYTES, - "maxEvidenceBufferBytes": MAX_EVIDENCE_BUFFER_BYTES, - "droppedEvidenceEvents": state.dropped_evidence_events, - "stageCounts": state.stage_counts, - "stageCountOverflow": state.stage_count_overflow, - "snapshotBoundary": "execution_finalization", - })) + loop { + let snapshot = state.snapshot(); + if snapshot.to_string().len() <= MAX_TRACE_SNAPSHOT_BYTES { + return Some(snapshot); + } + // 完整编码还包含计数与外层字段,超限时先淘汰普通事件并保留丢弃计数 + // 元数据本身无法容纳时省略诊断,不能让辅助观测阻断请求终态写入 + let index = + regular_eviction_index(&state).or_else(|| evidence_eviction_index(&state))?; + remove_event(&mut state, index); + } } fn push(&self, stage: &'static str, data: Value, coalesce: bool) { @@ -439,26 +437,14 @@ impl TraceContext { if state.evidence_bytes > MAX_EVIDENCE_BUFFER_BYTES || state.evidence_events > MAX_EVIDENCE_EVENTS { - if let Some(index) = state - .events - .iter() - .position(|event| event.evidence && event.stage != "request.routed") - .or_else(|| state.events.iter().position(|event| event.evidence)) - { + if let Some(index) = evidence_eviction_index(&state) { remove_event(&mut state, index); } continue; } // 相比流中间的常规事件,优先保留失败与重试决策 // 关键事实使用独立预算,只有自身超限才会淘汰 - let index = (HEAD_EVENTS..state.events.len().saturating_sub(8)) - .find(|index| !state.events[*index].evidence) - .or_else(|| { - (HEAD_EVENTS..state.events.len()) - .chain(0..HEAD_EVENTS.min(state.events.len())) - .find(|index| !state.events[*index].evidence) - }); - if let Some(index) = index { + if let Some(index) = regular_eviction_index(&state) { remove_event(&mut state, index); } else { break; @@ -467,6 +453,45 @@ impl TraceContext { } } +impl TraceState { + fn snapshot(&self) -> Value { + json!({ + "schemaVersion": 3, "requestId": self.request_id, + "startedAtUnixMs": self.started_at_ms, "totalEvents": self.sequence, + "droppedEvents": self.dropped_events, "maxEvents": MAX_EVENTS, + "captureMode": "sanitized", "events": self.events, + "wireDumpEnabled": tracing::enabled!(target: "request_dump", tracing::Level::INFO), + "wireFrames": self.wire_frames, "wireBytes": self.wire_bytes, + "maxBufferBytes": MAX_BUFFER_BYTES + MAX_EVIDENCE_BUFFER_BYTES, "headEvents": HEAD_EVENTS, + "maxEvidenceEvents": MAX_EVIDENCE_EVENTS, + "maxEvidenceDataBytes": MAX_EVIDENCE_DATA_BYTES, + "maxEvidenceBufferBytes": MAX_EVIDENCE_BUFFER_BYTES, + "droppedEvidenceEvents": self.dropped_evidence_events, + "stageCounts": self.stage_counts, + "stageCountOverflow": self.stage_count_overflow, + "snapshotBoundary": "execution_finalization", + }) + } +} + +fn regular_eviction_index(state: &TraceState) -> Option { + (HEAD_EVENTS..state.events.len().saturating_sub(8)) + .find(|index| !state.events[*index].evidence) + .or_else(|| { + (HEAD_EVENTS..state.events.len()) + .chain(0..HEAD_EVENTS.min(state.events.len())) + .find(|index| !state.events[*index].evidence) + }) +} + +fn evidence_eviction_index(state: &TraceState) -> Option { + state + .events + .iter() + .position(|event| event.evidence && event.stage != "request.routed") + .or_else(|| state.events.iter().position(|event| event.evidence)) +} + fn remove_event(state: &mut TraceState, index: usize) { if let Some(event) = state.events.remove(index) { state.bytes = state.bytes.saturating_sub(event.bytes); diff --git a/backend/crates/gateway-core/tests/diagnostics/trace.rs b/backend/crates/gateway-core/tests/diagnostics/trace.rs index a62974cd3..5dcd6fc0e 100644 --- a/backend/crates/gateway-core/tests/diagnostics/trace.rs +++ b/backend/crates/gateway-core/tests/diagnostics/trace.rs @@ -5,7 +5,7 @@ use std::{ sync::{Arc, Mutex}, }; -use gateway_core::diagnostics::{TraceContext, body_fingerprint}; +use gateway_core::diagnostics::{MAX_TRACE_SNAPSHOT_BYTES, TraceContext, body_fingerprint}; use gateway_core::error::{ProviderDiagnostic, ProviderError, ProviderErrorKind, RawUpstreamError}; use gateway_core::upstream::UpstreamSendState; use serde_json::json; @@ -47,6 +47,7 @@ fn critical_evidence_survives_stream_pressure_and_preserves_stage_coverage() { ); assert_eq!(snapshot["droppedEvidenceEvents"], 0); assert!(snapshot["droppedEvents"].as_u64().unwrap() > 0); + assert!(snapshot.to_string().len() <= MAX_TRACE_SNAPSHOT_BYTES); } #[test] @@ -67,8 +68,42 @@ fn critical_evidence_has_an_independent_bound_and_reports_its_own_loss() { events.len() as u64 + snapshot["droppedEvents"].as_u64().unwrap(), 100 ); - assert!(snapshot.to_string().len() < 128 * 1024); + assert!(snapshot.to_string().len() <= MAX_TRACE_SNAPSHOT_BYTES); } + +#[test] +fn snapshot_budget_includes_stage_coverage_and_event_envelopes() { + let trace = TraceContext::new("req_snapshot_budget"); + trace.record("request.routed", json!({"scope": "all"})); + for index in 0..10 { + trace.record_evidence( + "provider.request.input", + json!({"index": index, "hashes": "a".repeat(6000)}), + ); + } + for index in 0..64 { + let stage = Box::leak(format!("stage.{index}.{}", "x".repeat(512)).into_boxed_str()); + trace.record(stage, json!({"detail": "x".repeat(2000)})); + } + trace.record("request.finished", json!({"outcome": "succeeded"})); + let snapshot = trace.snapshot().unwrap(); + let events = snapshot["events"].as_array().unwrap(); + assert!(snapshot.to_string().len() <= MAX_TRACE_SNAPSHOT_BYTES); + assert_eq!(events.first().unwrap()["stage"], "request.routed"); + assert_eq!(events.last().unwrap()["stage"], "request.finished"); + assert_eq!(snapshot["droppedEvidenceEvents"], 0); + assert!(snapshot["droppedEvents"].as_u64().unwrap() > 0); + let retained: u64 = events + .iter() + .map(|event| event["count"].as_u64().unwrap()) + .sum(); + assert_eq!( + retained + snapshot["droppedEvents"].as_u64().unwrap(), + snapshot["totalEvents"] + ); + assert_eq!(trace.snapshot(), Some(snapshot)); +} + #[test] fn bounded_history_preserves_start_failure_and_final_result() { let trace = TraceContext::new("req_trace"); diff --git a/backend/crates/gateway-store/src/postgres/execution/mod.rs b/backend/crates/gateway-store/src/postgres/execution/mod.rs index 242213910..4f13734ad 100644 --- a/backend/crates/gateway-store/src/postgres/execution/mod.rs +++ b/backend/crates/gateway-store/src/postgres/execution/mod.rs @@ -8,6 +8,7 @@ use serde_json::Value; use sqlx::PgPool; use uuid::Uuid; +use gateway_core::diagnostics::MAX_TRACE_SNAPSHOT_BYTES; use gateway_core::engine::{ AttemptRecord as CoreAttemptRecord, ExecutionStore, IntermediateFailure, ModelRequestFinalization as CoreModelRequestFinalization, ModelRequestId, @@ -326,13 +327,11 @@ impl ModelRequestFinalization { { return Err(invalid("provider observation must be a JSON object")); } - if self - .diagnostic_trace_json - .as_ref() - .is_some_and(|trace| !trace.is_object() || trace.to_string().len() > 64 * 1024) - { + if self.diagnostic_trace_json.as_ref().is_some_and(|trace| { + !trace.is_object() || trace.to_string().len() > MAX_TRACE_SNAPSHOT_BYTES + }) { return Err(invalid( - "diagnostic trace must be a JSON object within 64 KiB", + "diagnostic trace must be a JSON object within the snapshot size limit", )); } validate_optional_stable_reason( diff --git a/backend/crates/gateway-store/tests/postgres/execution/mod.rs b/backend/crates/gateway-store/tests/postgres/execution/mod.rs index bee0edf67..2f67c66a2 100644 --- a/backend/crates/gateway-store/tests/postgres/execution/mod.rs +++ b/backend/crates/gateway-store/tests/postgres/execution/mod.rs @@ -8,7 +8,7 @@ use chrono::{DateTime, Duration, Utc}; use gateway_admin::{ model::observability as admin_observability, ports::store::ObservabilityStore as _, }; -use gateway_core::diagnostics::TraceContext; +use gateway_core::diagnostics::{MAX_TRACE_SNAPSHOT_BYTES, TraceContext}; use gateway_core::engine::{ ExecutionOutcome, ExecutionStore, ModelRequestFailureObservation, ModelRequestFinalization as CoreModelRequestFinalization, ModelRequestId, @@ -1429,54 +1429,100 @@ async fn seed_running_request(pool: &sqlx::PgPool, id: &str) -> Result<(), sqlx: } #[tokio::test] -async fn diagnostic_trace_is_finalized_atomically_and_available_for_failed_request_details() { +async fn diagnostic_trace_pressure_preserves_terminal_outcomes_and_usage() { let Some(database) = TestDatabase::create("diagnostic_trace").await else { return; }; - seed_running_request(&database.pool, "req_diagnostic_failed") - .await - .unwrap(); let store = PgExecutionStore::new(database.pool.clone()); - let mut finalization = successful_core_finalization("req_diagnostic_failed"); - finalization.outcome = ExecutionOutcome::Failed; - finalization.client_status_code = Some(502); - let diagnostic = gateway_core::error::ProviderDiagnostic::new( - "OpenAI WebSocket receive idle timeout after 300s", - ) - .with_classification("receive", "receive_idle_timeout"); - let provider_error = gateway_core::error::ProviderError::new( - gateway_core::error::ProviderErrorKind::Timeout, - UpstreamSendState::Ambiguous, - ) - .with_diagnostic(diagnostic); - finalization.error = Some(GatewayError::from_provider(&provider_error)); - let trace = serde_json::json!({"schemaVersion": 1, "requestId": "req_diagnostic_failed", "events": [ - {"attemptIndex": 1, "stage": "attempt.failed", "data": {"diagnostic": { - "stage": "receive", "code": "receive_idle_timeout", "message": "OpenAI WebSocket receive idle timeout after 300s" - }}}, - ]}); - finalization.diagnostic_trace_json = Some(trace.to_string()); - ExecutionStore::finalize_model_request(&store, finalization) - .await - .unwrap(); - let persisted: (String, serde_json::Value, String) = sqlx::query_as( - "select outcome, diagnostic_trace_json, error_message from model_request_observations where id = 'req_diagnostic_failed'" - ).fetch_one(&database.pool).await.unwrap(); - assert_eq!( - persisted, - ( - "failed".to_owned(), - trace.clone(), - "OpenAI WebSocket receive idle timeout after 300s".to_owned() - ) - ); let repository = super::observability_repository(&database.pool); - let detail = repository - .usage_record_detail("req_diagnostic_failed") + for (id, outcome, expected_outcome) in [ + ( + "req_diagnostic_success", + ExecutionOutcome::Succeeded, + "succeeded", + ), + ("req_diagnostic_failed", ExecutionOutcome::Failed, "failed"), + ] { + seed_running_request(&database.pool, id).await.unwrap(); + let mut finalization = successful_core_finalization(id); + finalization.outcome = outcome; + finalization.usage.input_tokens = Some(100_000); + finalization.usage.output_tokens = Some(2000); + finalization.usage.total_tokens = Some(102_000); + finalization.cost = CalculatedCost::from_usd_ticks(10_000_000_000) + .unwrap() + .into_estimate(); + finalization.timings.first_token_ms = Some(120); + finalization.timings.latency_ms = Some(42_000); + let trace = TraceContext::new(id); + trace.record("request.routed", json!({"scope": "all"})); + for stage in ["provider.request.input", "provider.request.outbound"] { + trace.record_evidence(stage, json!({"hashes": "a".repeat(6000)})); + } + for index in 0..500 { + trace.record( + "stream.sample", + json!({"index": index, "detail": "x".repeat(2000)}), + ); + } + let error_message = if outcome == ExecutionOutcome::Failed { + finalization.client_status_code = Some(502); + let message = "OpenAI WebSocket receive idle timeout after 300s"; + let provider_error = gateway_core::error::ProviderError::new( + gateway_core::error::ProviderErrorKind::Timeout, + UpstreamSendState::Ambiguous, + ) + .with_diagnostic( + gateway_core::error::ProviderDiagnostic::new(message) + .with_classification("receive", "receive_idle_timeout"), + ); + trace.record_provider_failure(&provider_error); + finalization.error = Some(GatewayError::from_provider(&provider_error)); + Some(message) + } else { + None + }; + trace.record("request.finished", json!({"outcome": expected_outcome})); + let snapshot = trace.snapshot().unwrap(); + let encoded = snapshot.to_string(); + assert!(encoded.len() > 64 * 1024); + assert!(encoded.len() <= MAX_TRACE_SNAPSHOT_BYTES); + assert!(snapshot["droppedEvents"].as_u64().unwrap() > 0); + finalization.diagnostic_trace_json = Some(encoded); + ExecutionStore::finalize_model_request(&store, finalization) + .await + .expect("trace pressure must not discard the terminal result"); + let persisted: Value = sqlx::query_scalar( + "select jsonb_build_object( + 'outcome', outcome, 'input', input_tokens, 'output', output_tokens, + 'cost', cost_amount::text, 'firstToken', first_token_ms, + 'latency', latency_ms, 'error', error_message) + from model_request_observations where id = $1", + ) + .bind(id) + .fetch_one(&database.pool) .await .unwrap(); - assert_eq!(detail.trace, Some(trace)); - assert!(detail.related_requests.is_empty()); + assert_eq!( + persisted, + json!({ + "outcome": expected_outcome, "input": 100_000, "output": 2000, + "cost": "1.0000000000", "firstToken": 120, "latency": 42_000, + "error": error_message, + }) + ); + let detail = repository.usage_record_detail(id).await.unwrap(); + assert_eq!(detail.trace, Some(snapshot)); + assert!(detail.related_requests.is_empty()); + } + assert_eq!( + store + .recover_expired_model_requests(Utc::now() + Duration::minutes(20)) + .await + .unwrap() + .requests, + 0, + ); database.close().await; } @@ -1784,7 +1830,7 @@ async fn zero_attempt_finalization_rejects_sent_or_missing_attempt_facts_atomica for invalid_trace in [ "not JSON".to_owned(), "[]".to_owned(), - json!({"data": "x".repeat(65_536)}).to_string(), + json!({"data": "x".repeat(MAX_TRACE_SNAPSHOT_BYTES)}).to_string(), ] { let mut finalization = early_failure(&request); finalization.diagnostic_trace_json = Some(invalid_trace); diff --git a/docs/api.md b/docs/api.md index 1fa94f310..bf9654b9a 100644 --- a/docs/api.md +++ b/docs/api.md @@ -2188,7 +2188,9 @@ WebSocket 握手不主动添加专项计时请求头,客户端显式提供的 已有的 `provider.precommit.released` 数据保留释放原因、预取字节数与等待时间,不等同于下游已提交 Trace v3 为路由、账号选择、失败、重试、响应边界及终态事实预留独立的 64 KiB 预算, -普通事件另有 64 KiB 预算,总事件数最多 128。`stageCounts` 记录至多 64 类已采集阶段的次数, +普通事件另有 64 KiB 预算,总事件数最多 128。包含元数据的完整 JSON 快照最多 128 KiB, +生成与持久化共用同一上限;完整编码超限时先淘汰普通事件,再淘汰关键证据。 +`stageCounts` 记录至多 64 类已采集阶段的次数, 事件淘汰不抹除阶段计数;`droppedEvidenceEvents` 与 `stageCountOverflow` 分别报告关键证据淘汰和阶段统计超限 OpenAI Responses 的 `provider.request.input` 与 `provider.request.outbound` 分别记录 Provider 收到的