From 93dc417b79e6f21449513588a9dcc965e116d0ac Mon Sep 17 00:00:00 2001 From: Zyy Date: Tue, 8 Sep 2026 17:15:42 +0800 Subject: [PATCH] fix(observability): preserve failure causes and complete log retention Preserve safe failure stages and causes through OpenAI and xAI transport mapping, Core traces and persisted errors without changing replay safety. Distinguish missing WebSocket close handshakes from TCP resets and surface connection facts in the borderless diagnostics panel. Retain complete UTC date groups instead of limiting segment counts. Default application and OAuth recovery logs to seven days and full payload logs to one day. Compress closed segments, drain bounded file queues on shutdown and report write or maintenance failures through the logging health probe. Normalize file names and expose an independent OAuth recovery switch that defaults to off. Remove max_files; leave old file names for manual cleanup. Validation: workspace backend tests with isolated PostgreSQL, final Host and application tests (104 passed), OpenAI tests (542 passed), xAI tests (371 passed), rustfmt and strict Clippy. Frontend lint/build, light/dark and mobile browser checks, and Compose validation passed. Redis integration tests were not configured and followed the existing skip convention. Co-authored-by: Codex --- backend/apps/gateway/src/bootstrap.rs | 1 + .../gateway-core/src/engine/coordinator.rs | 3 + backend/crates/gateway-core/src/error.rs | 77 +++- .../gateway-core/tests/engine/coordinator.rs | 18 +- backend/crates/gateway-core/tests/error.rs | 13 +- backend/crates/gateway-host/src/config.rs | 26 +- backend/crates/gateway-host/src/lib.rs | 12 +- backend/crates/gateway-host/src/logging.rs | 385 ------------------ .../crates/gateway-host/src/logging/mod.rs | 230 +++++++++++ .../crates/gateway-host/src/logging/sink.rs | 154 +++++++ .../crates/gateway-host/src/logging/writer.rs | 246 +++++++++++ backend/crates/gateway-host/tests/config.rs | 48 ++- .../tests/{logging.rs => logging/mod.rs} | 178 ++++++-- .../crates/gateway-host/tests/logging/sink.rs | 108 +++++ .../gateway-host/tests/logging/writer.rs | 118 ++++++ .../gateway-store/tests/postgres/execution.rs | 31 +- .../openai/src/credential/recovery_log.rs | 6 +- .../providers/openai/src/provider/failure.rs | 150 ++++++- .../src/transport/websocket/coordinator.rs | 15 +- .../openai/src/transport/websocket/error.rs | 44 ++ .../transport/websocket/exchange/stream.rs | 7 +- .../openai/tests/provider/contract.rs | 168 ++++++++ .../tests/transport/websocket/diagnostics.rs | 107 +++++ .../openai/tests/transport/websocket/mod.rs | 1 + .../providers/xai/src/provider/failure.rs | 12 + .../providers/xai/src/transport/inference.rs | 17 +- .../providers/xai/src/transport/network.rs | 25 +- .../providers/xai/tests/provider/contract.rs | 29 ++ .../providers/xai/tests/transport/network.rs | 16 + deploy/README.md | 43 +- deploy/config.example.yaml | 6 +- docs/architecture.md | 12 +- frontend/src/components/base/BaseButton.vue | 4 +- .../usage/components/OpsErrorDetailModal.vue | 2 +- .../components/RequestDiagnosticsPanel.vue | 32 +- .../components/RequestTransportFailure.vue | 115 ++++++ 36 files changed, 1967 insertions(+), 492 deletions(-) delete mode 100644 backend/crates/gateway-host/src/logging.rs create mode 100644 backend/crates/gateway-host/src/logging/mod.rs create mode 100644 backend/crates/gateway-host/src/logging/sink.rs create mode 100644 backend/crates/gateway-host/src/logging/writer.rs rename backend/crates/gateway-host/tests/{logging.rs => logging/mod.rs} (58%) create mode 100644 backend/crates/gateway-host/tests/logging/sink.rs create mode 100644 backend/crates/gateway-host/tests/logging/writer.rs create mode 100644 backend/crates/providers/openai/tests/transport/websocket/diagnostics.rs create mode 100644 frontend/src/views/usage/components/RequestTransportFailure.vue diff --git a/backend/apps/gateway/src/bootstrap.rs b/backend/apps/gateway/src/bootstrap.rs index aaa97bc3e..582fe4c52 100644 --- a/backend/apps/gateway/src/bootstrap.rs +++ b/backend/apps/gateway/src/bootstrap.rs @@ -87,6 +87,7 @@ pub async fn run() -> Result<(), BootstrapError> { let mut probes = store.health_probes(); probes.extend(core.health_probes()); + probes.push(host.logging_health_probe()); let api = gateway_api::initialize( api, core.execution_service(), diff --git a/backend/crates/gateway-core/src/engine/coordinator.rs b/backend/crates/gateway-core/src/engine/coordinator.rs index ff3ce8b45..f9366dc6e 100644 --- a/backend/crates/gateway-core/src/engine/coordinator.rs +++ b/backend/crates/gateway-core/src/engine/coordinator.rs @@ -1854,6 +1854,9 @@ fn duration_ms(duration: Duration) -> u64 { fn record_trace_error(trace: &TraceContext, error: &ProviderError) { trace.record("attempt.failed", json!({ "kind": error.kind().as_str(), "sendState": format!("{:?}", error.send_state()), + "diagnostic": error.diagnostic().map(|diagnostic| json!({ + "stage": diagnostic.stage(), "code": diagnostic.code(), "message": diagnostic.as_str(), + })), "upstreamStatus": error.upstream_status(), "upstreamCode": error.upstream_code().map(|code| code.as_str()), "rawError": error.raw_upstream_error().map(|raw| { diff --git a/backend/crates/gateway-core/src/error.rs b/backend/crates/gateway-core/src/error.rs index 948c75c79..4530f132c 100644 --- a/backend/crates/gateway-core/src/error.rs +++ b/backend/crates/gateway-core/src/error.rs @@ -172,17 +172,72 @@ impl fmt::Debug for ClientVisibleUpstreamError { /// Adapter 只能放入不含凭据、cookie、原始请求正文和 opaque response ID 的有界摘要； /// 原始上游错误返回由 [`RawUpstreamError`] 单独承载。 #[derive(Clone, PartialEq, Eq)] -pub struct ProviderDiagnostic(String); +pub struct ProviderDiagnostic { + message: String, + stage: Option<&'static str>, + code: Option<&'static str>, +} impl ProviderDiagnostic { #[must_use] pub fn new(message: impl Into) -> Self { - Self(message.into()) + Self { + message: message.into(), + stage: None, + code: None, + } + } + + /// 附着本地定义的阶段与原因码；独立于发送状态和重试策略。 + #[must_use] + pub const fn with_classification(mut self, stage: &'static str, code: &'static str) -> Self { + self.stage = Some(stage); + self.code = Some(code); + self + } + + /// 从错误链补充可证明的 I/O 原因；只保留本地原因码，不格式化原始错误。 + #[must_use] + pub fn with_io_cause(mut self, error: &(dyn std::error::Error + 'static)) -> Self { + let mut source = Some(error); + while let Some(error) = source { + if let Some(error) = error.downcast_ref::() { + let code = match error.kind() { + std::io::ErrorKind::ConnectionRefused => Some("connection_refused"), + std::io::ErrorKind::ConnectionReset => Some("tcp_reset"), + std::io::ErrorKind::ConnectionAborted => Some("connection_aborted"), + std::io::ErrorKind::BrokenPipe => Some("broken_pipe"), + std::io::ErrorKind::UnexpectedEof => Some("unexpected_eof"), + std::io::ErrorKind::TimedOut => Some("io_timeout"), + std::io::ErrorKind::NetworkUnreachable => Some("network_unreachable"), + std::io::ErrorKind::HostUnreachable => Some("host_unreachable"), + _ => None, + }; + if let Some(code) = code { + self.message.push_str("; I/O cause: "); + self.message.push_str(code); + self.code = Some(code); + break; + } + } + source = error.source(); + } + self + } + + #[must_use] + pub const fn stage(&self) -> Option<&'static str> { + self.stage + } + + #[must_use] + pub const fn code(&self) -> Option<&'static str> { + self.code } #[must_use] pub fn as_str(&self) -> &str { - &self.0 + &self.message } } @@ -817,7 +872,10 @@ impl fmt::Debug for ProviderError { "sensitive_context", &self.sensitive_context_redacted.then_some(""), ) - .field("diagnostic", &self.diagnostic.as_ref().map(|_| "")) + .field( + "diagnostic", + &self.diagnostic.as_deref().map(|_| ""), + ) .field( "raw_upstream_error", &self.raw_upstream_error.as_ref().map(|_| ""), @@ -915,7 +973,7 @@ impl GatewayErrorKind { pub struct GatewayError { kind: GatewayErrorKind, message: &'static str, - diagnostic: Option, + diagnostic: Option>, client_visible_upstream_error: Option, } @@ -993,7 +1051,7 @@ impl GatewayError { /// 附加只供运维观测使用的脱敏 Provider 摘要。 #[must_use] pub fn with_diagnostic(mut self, diagnostic: ProviderDiagnostic) -> Self { - self.diagnostic = Some(diagnostic); + self.diagnostic = Some(Box::new(diagnostic)); self } @@ -1019,7 +1077,7 @@ impl GatewayError { /// 返回可安全持久化和展示的 Provider 诊断摘要。 #[must_use] pub fn diagnostic(&self) -> Option<&ProviderDiagnostic> { - self.diagnostic.as_ref() + self.diagnostic.as_deref() } /// 返回优先用于原客户端协议响应的 message；持久化和日志必须继续使用 @@ -1054,7 +1112,10 @@ impl fmt::Debug for GatewayError { .debug_struct("GatewayError") .field("kind", &self.kind) .field("message", &self.message) - .field("diagnostic", &self.diagnostic.as_ref().map(|_| "")) + .field( + "diagnostic", + &self.diagnostic.as_deref().map(|_| ""), + ) .field( "client_visible_upstream_error", &self diff --git a/backend/crates/gateway-core/tests/engine/coordinator.rs b/backend/crates/gateway-core/tests/engine/coordinator.rs index a9bad8108..d4ea8825b 100644 --- a/backend/crates/gateway-core/tests/engine/coordinator.rs +++ b/backend/crates/gateway-core/tests/engine/coordinator.rs @@ -1292,7 +1292,13 @@ fn discarded_attempt_observation_does_not_leak_into_retry_result() { Ok(ProviderEvent::observation(first_observation)), Err( ProviderError::new(ProviderErrorKind::Unavailable, UpstreamSendState::Sent) - .with_replay_safe(), + .with_replay_safe() + .with_diagnostic( + gateway_core::error::ProviderDiagnostic::new( + "response event idle timeout after 300s", + ) + .with_classification("receive", "receive_idle_timeout"), + ), ), ], }, @@ -1329,6 +1335,16 @@ fn discarded_attempt_observation_does_not_leak_into_retry_result() { .iter() .any(|event| event["stage"] == "attempt.started" && event["attemptIndex"] == 2) ); + let failed = events + .iter() + .find(|event| event["stage"] == "attempt.failed") + .unwrap(); + assert_eq!( + failed["data"]["diagnostic"], + serde_json::json!({ + "stage": "receive", "code": "receive_idle_timeout", "message": "response event idle timeout after 300s" + }) + ); assert_eq!(events.last().unwrap()["stage"], "request.finished"); assert_eq!(finalization.upstream_transport.as_deref(), Some("http_sse")); diff --git a/backend/crates/gateway-core/tests/error.rs b/backend/crates/gateway-core/tests/error.rs index 7c3dab529..e302bc387 100644 --- a/backend/crates/gateway-core/tests/error.rs +++ b/backend/crates/gateway-core/tests/error.rs @@ -30,7 +30,10 @@ fn provider_error_debug_should_not_print_classified_upstream_values() { fn classified_provider_diagnostic_survives_clone_without_entering_debug() { let message = "OpenAI WebSocket closed before terminal response (close code 1000)"; let error = ProviderError::new(ProviderErrorKind::Transport, UpstreamSendState::Ambiguous) - .with_diagnostic(ProviderDiagnostic::new(message)); + .with_diagnostic( + ProviderDiagnostic::new(message) + .with_classification("receive", "closed_before_terminal"), + ); let cloned = error.clone(); let gateway = GatewayError::from_provider(&cloned); @@ -38,6 +41,14 @@ fn classified_provider_diagnostic_survives_clone_without_entering_debug() { gateway.diagnostic().map(ProviderDiagnostic::as_str), Some(message) ); + assert_eq!( + gateway.diagnostic().and_then(ProviderDiagnostic::stage), + Some("receive") + ); + assert_eq!( + gateway.diagnostic().and_then(ProviderDiagnostic::code), + Some("closed_before_terminal") + ); assert!(!format!("{error:?}").contains(message)); assert!(!format!("{gateway:?}").contains(message)); } diff --git a/backend/crates/gateway-host/src/config.rs b/backend/crates/gateway-host/src/config.rs index 5eef78b2f..97a6770c8 100644 --- a/backend/crates/gateway-host/src/config.rs +++ b/backend/crates/gateway-host/src/config.rs @@ -118,16 +118,31 @@ pub struct LoggingConfig { pub level: String, pub stdout: bool, pub file: FileLoggingConfig, + /// 将 OAuth 原始 AT/RT 写入独立恢复日志；默认关闭，沿用 file 的目录与保留期。 + #[serde(default)] + pub oauth_recovery: bool, /// 将完整请求原文写入独立诊断日志；默认关闭，因为内容包含凭据。 + /// 请求报文的留存窗口由 request_dump_retention_days 独立控制。 #[serde(default)] pub request_dump: bool, + /// 完整报文的最少保留天数；按 UTC 日期整组清理，默认 1 天。 + #[serde(default = "default_request_dump_retention_days")] + pub request_dump_retention_days: usize, +} + +fn default_file_retention_days() -> usize { + 7 +} + +fn default_request_dump_retention_days() -> usize { + 1 } impl LoggingConfig { fn resolve_and_validate(&mut self, source_dir: &Path) -> Result<(), ConfigError> { EnvFilter::try_new(&self.level) .map_err(|_| ConfigError::InvalidField("host.logging.level"))?; - if !self.stdout && !self.file.enabled { + if !self.stdout && !self.file.enabled && !self.oauth_recovery && !self.request_dump { return Err(ConfigError::InvalidField("host.logging")); } if self.file.directory.as_os_str().is_empty() { @@ -143,8 +158,10 @@ impl LoggingConfig { "host.logging.file.max_file_size_mb", )); } - if self.file.max_files == 0 { - return Err(ConfigError::InvalidField("host.logging.file.max_files")); + if self.request_dump_retention_days == 0 { + return Err(ConfigError::InvalidField( + "host.logging.request_dump_retention_days", + )); } resolve_relative_path(source_dir, &mut self.file.directory); Ok(()) @@ -157,9 +174,10 @@ impl LoggingConfig { pub struct FileLoggingConfig { pub enabled: bool, pub directory: PathBuf, + /// 至少保留的完整天数，默认 7 天；文件数量不参与清理。 + #[serde(default = "default_file_retention_days")] pub retention_days: usize, pub max_file_size_mb: u64, - pub max_files: usize, } #[derive(Debug, thiserror::Error)] diff --git a/backend/crates/gateway-host/src/lib.rs b/backend/crates/gateway-host/src/lib.rs index 5f17f4dff..ad9d214e8 100644 --- a/backend/crates/gateway-host/src/lib.rs +++ b/backend/crates/gateway-host/src/lib.rs @@ -13,7 +13,7 @@ use axum::Router; use gateway_admin::ports::{ client_distribution::ClientDistributionResolver, system::SystemOperations, }; -use gateway_core::health::WorkerHealthSource; +use gateway_core::health::{HealthProbe, WorkerHealthSource}; use gateway_core::lifecycle::CancellationToken; use gateway_core::lifecycle::ConnectionLifecycle; use gateway_core::task::{WorkerContribution, WorkerLeaderLeasePort}; @@ -29,7 +29,7 @@ use self::workers::WorkerSupervisor; /// Host 初始化的能力集；字段全部私有，不暴露内部监督器或进程状态。 pub struct HostBundle { config: HostConfig, - _log_guard: LogGuard, + log_guard: LogGuard, cancellation: CancellationToken, connections: Arc, workers: WorkerSupervisor, @@ -50,7 +50,7 @@ pub async fn initialize(config: HostConfig) -> Result { let client_distribution = Arc::new(RgAdguardClientDistribution::new()); Ok(HostBundle { config, - _log_guard: log_guard, + log_guard, cancellation, connections, workers, @@ -86,6 +86,12 @@ impl HostBundle { self.connections.clone() } + /// 报告文件日志写入或归档失败，避免将缺失日志误报为完整。 + #[must_use] + pub fn logging_health_probe(&self) -> Arc { + self.log_guard.health_probe() + } + #[must_use] pub fn worker_health(&self) -> Arc { self.workers.health_source() diff --git a/backend/crates/gateway-host/src/logging.rs b/backend/crates/gateway-host/src/logging.rs deleted file mode 100644 index c49525525..000000000 --- a/backend/crates/gateway-host/src/logging.rs +++ /dev/null @@ -1,385 +0,0 @@ -//! 进程结构化日志与按日期/大小轮转的文件 writer。 - -use std::env; -use std::fs::{self, File, OpenOptions}; -use std::io::{self, Write}; -use std::path::{Path, PathBuf}; - -use chrono::Utc; -use tracing_appender::non_blocking::{NonBlocking, NonBlockingBuilder, WorkerGuard}; -use tracing_subscriber::EnvFilter; -use tracing_subscriber::{Layer as _, layer::SubscriberExt as _, util::SubscriberInitExt as _}; - -use crate::config::{FileLoggingConfig, LoggingConfig}; - -const LOG_FILE_PREFIX: &str = "codex-proxy-rs"; -const OAUTH_RECOVERY_LOG_FILE_PREFIX: &str = "codex-proxy-rs-oath"; -/// OAuth 恢复记录含原始 AT/RT，文件日志开启时必须落到独立文件。 -const OAUTH_RECOVERY_LOG_TARGET: &str = "oauth_recovery"; -const REQUEST_DUMP_LOG_FILE_PREFIX: &str = "codex-proxy-rs-request-dump"; -/// 请求转储含完整凭据与正文，只允许在显式开启时写入独立文件。 -const REQUEST_DUMP_LOG_TARGET: &str = "request_dump"; - -/// non-blocking 日志 writer 的进程级守卫。 -pub struct LogGuard { - _writers: Vec, -} - -#[derive(Debug, thiserror::Error)] -pub enum LogError { - #[error("log IO failed")] - Io(#[from] io::Error), - #[error("logging filter is invalid")] - InvalidFilter, - #[error("global tracing subscriber is already initialized")] - AlreadyInitialized, - #[error("logging size limit is too large")] - SizeOverflow, -} - -/// 按自然日、单文件大小、保留天数和文件总数初始化结构化日志。 -pub fn initialize_logging(config: &LoggingConfig) -> Result { - let directive = env::var("RUST_LOG").unwrap_or_else(|_| config.level.clone()); - let file_filter = application_file_filter(&directive)?; - let recovery_file_filter = oauth_recovery_file_filter(); - let request_dump_file_filter = request_dump_file_filter(); - let stdout_filter = - EnvFilter::try_new(stdout_filter_directive(&directive, config.file.enabled)) - .map_err(|_| LogError::InvalidFilter)?; - let mut guards = Vec::new(); - - let stdout_writer = config.stdout.then(|| { - let (writer, guard) = NonBlockingBuilder::default() - .thread_name("gateway-log-stdout") - .finish(io::stdout()); - guards.push(guard); - writer - }); - let (file_writer, recovery_file_writer) = if config.file.enabled { - let (file_writer, file_guard) = - rotating_file_writer(&config.file, LOG_FILE_PREFIX, "gateway-log-file")?; - let (recovery_file_writer, recovery_file_guard) = rotating_file_writer( - &config.file, - OAUTH_RECOVERY_LOG_FILE_PREFIX, - "gateway-log-oauth-recovery-file", - )?; - guards.push(file_guard); - guards.push(recovery_file_guard); - (Some(file_writer), Some(recovery_file_writer)) - } else { - (None, None) - }; - let request_dump_file_writer = if config.request_dump { - let (writer, guard) = rotating_file_writer( - &config.file, - REQUEST_DUMP_LOG_FILE_PREFIX, - "gateway-log-request-dump-file", - )?; - guards.push(guard); - Some(writer) - } else { - None - }; - - let stdout_layer = stdout_writer.map(|writer| { - tracing_subscriber::fmt::layer() - .compact() - .with_writer(writer) - .with_target(false) - .with_ansi(false) - .with_filter(stdout_filter) - }); - let file_layer = file_writer.map(|writer| { - tracing_subscriber::fmt::layer() - .json() - .with_writer(writer) - .with_target(true) - .with_file(true) - .with_line_number(true) - .with_thread_ids(true) - .with_thread_names(true) - .with_current_span(true) - .with_span_list(true) - .with_filter(file_filter) - }); - let recovery_file_layer = recovery_file_writer.map(|writer| { - tracing_subscriber::fmt::layer() - .json() - .with_writer(writer) - .with_target(true) - .with_file(true) - .with_line_number(true) - .with_thread_ids(true) - .with_thread_names(true) - .with_current_span(true) - .with_span_list(true) - .with_filter(recovery_file_filter) - }); - let request_dump_file_layer = request_dump_file_writer.map(|writer| { - tracing_subscriber::fmt::layer() - .json() - .with_writer(writer) - .with_target(true) - .with_file(true) - .with_line_number(true) - .with_thread_ids(true) - .with_thread_names(true) - .with_current_span(true) - .with_span_list(true) - .with_filter(request_dump_file_filter) - }); - tracing_subscriber::registry() - .with(stdout_layer) - .with(file_layer) - .with(recovery_file_layer) - .with(request_dump_file_layer) - .try_init() - .map_err(|_| LogError::AlreadyInitialized)?; - Ok(LogGuard { _writers: guards }) -} - -fn application_file_filter(directive: &str) -> Result { - let recovery_directive = format!("{OAUTH_RECOVERY_LOG_TARGET}=off") - .parse() - .expect("static OAuth recovery log directive is valid"); - let request_dump_directive = format!("{REQUEST_DUMP_LOG_TARGET}=off") - .parse() - .expect("static request dump log directive is valid"); - EnvFilter::try_new(directive) - .map(|filter| filter.add_directive(recovery_directive)) - .map(|filter| filter.add_directive(request_dump_directive)) - .map_err(|_| LogError::InvalidFilter) -} - -fn oauth_recovery_file_filter() -> EnvFilter { - format!("off,{OAUTH_RECOVERY_LOG_TARGET}=info") - .parse() - .expect("static OAuth recovery log filter is valid") -} - -fn request_dump_file_filter() -> EnvFilter { - format!("off,{REQUEST_DUMP_LOG_TARGET}=info") - .parse() - .expect("static request dump log filter is valid") -} - -fn stdout_filter_directive(directive: &str, persistent_log_enabled: bool) -> String { - if !persistent_log_enabled { - format!("{directive},{REQUEST_DUMP_LOG_TARGET}=off") - } else { - "off,gateway_startup=info".to_owned() - } -} - -fn rotating_file_writer( - config: &FileLoggingConfig, - file_prefix: &'static str, - thread_name: &'static str, -) -> Result<(NonBlocking, WorkerGuard), LogError> { - let maximum_bytes = config - .max_file_size_mb - .checked_mul(1024 * 1024) - .ok_or(LogError::SizeOverflow)?; - let writer = RotatingLogWriter::open( - config.directory.clone(), - file_prefix, - maximum_bytes, - config.retention_days, - config.max_files, - )?; - Ok(NonBlockingBuilder::default() - .thread_name(thread_name) - .finish(writer)) -} - -struct RotatingLogWriter { - directory: PathBuf, - file_prefix: &'static str, - maximum_bytes: u64, - retention_days: usize, - maximum_files: usize, - date: chrono::NaiveDate, - segment: usize, - bytes_written: u64, - file: File, -} - -impl RotatingLogWriter { - fn open( - directory: PathBuf, - file_prefix: &'static str, - maximum_bytes: u64, - retention_days: usize, - maximum_files: usize, - ) -> io::Result { - fs::create_dir_all(&directory)?; - let date = Utc::now().date_naive(); - cleanup_log_files(&directory, file_prefix, date, retention_days, maximum_files)?; - let (segment, bytes_written) = - writable_log_segment(&directory, file_prefix, date, maximum_bytes)?; - let file = open_log_segment(&directory, file_prefix, date, segment)?; - Ok(Self { - directory, - file_prefix, - maximum_bytes, - retention_days, - maximum_files, - date, - segment, - bytes_written, - file, - }) - } - - fn rotate_if_required(&mut self, incoming_bytes: usize) -> io::Result<()> { - let date = Utc::now().date_naive(); - let incoming_bytes = u64::try_from(incoming_bytes).unwrap_or(u64::MAX); - let day_changed = date != self.date; - let size_exceeded = !day_changed - && self.bytes_written > 0 - && self.bytes_written.saturating_add(incoming_bytes) > self.maximum_bytes; - if !day_changed && !size_exceeded { - return Ok(()); - } - self.file.flush()?; - if day_changed { - self.date = date; - self.segment = 0; - } else { - self.segment = self.segment.saturating_add(1); - } - self.file = open_log_segment(&self.directory, self.file_prefix, self.date, self.segment)?; - self.bytes_written = self.file.metadata()?.len(); - cleanup_log_files( - &self.directory, - self.file_prefix, - self.date, - self.retention_days, - self.maximum_files, - ) - } -} - -impl Write for RotatingLogWriter { - fn write(&mut self, buffer: &[u8]) -> io::Result { - self.rotate_if_required(buffer.len())?; - let written = self.file.write(buffer)?; - self.bytes_written = self - .bytes_written - .saturating_add(u64::try_from(written).unwrap_or(u64::MAX)); - Ok(written) - } - - fn flush(&mut self) -> io::Result<()> { - self.file.flush() - } -} - -#[derive(Debug)] -struct ManagedLogFile { - date: chrono::NaiveDate, - segment: usize, - path: PathBuf, -} - -fn writable_log_segment( - directory: &Path, - file_prefix: &str, - date: chrono::NaiveDate, - maximum_bytes: u64, -) -> io::Result<(usize, u64)> { - let latest = managed_log_files(directory, file_prefix)? - .into_iter() - .filter(|entry| entry.date == date) - .max_by_key(|entry| entry.segment); - let Some(latest) = latest else { - return Ok((0, 0)); - }; - let length = latest.path.metadata()?.len(); - if length >= maximum_bytes { - Ok((latest.segment.saturating_add(1), 0)) - } else { - Ok((latest.segment, length)) - } -} - -fn open_log_segment( - directory: &Path, - file_prefix: &str, - date: chrono::NaiveDate, - segment: usize, -) -> io::Result { - OpenOptions::new() - .create(true) - .append(true) - .open(directory.join(log_file_name(file_prefix, date, segment))) -} - -fn log_file_name(file_prefix: &str, date: chrono::NaiveDate, segment: usize) -> String { - if segment == 0 { - format!("{file_prefix}.{date}.log") - } else { - format!("{file_prefix}.{date}.{segment}.log") - } -} - -fn managed_log_files(directory: &Path, file_prefix: &str) -> io::Result> { - Ok(fs::read_dir(directory)? - .filter_map(Result::ok) - .filter_map(|entry| { - parse_log_file_name(entry.file_name().to_string_lossy().as_ref(), file_prefix).map( - |(date, segment)| ManagedLogFile { - date, - segment, - path: entry.path(), - }, - ) - }) - .collect()) -} - -fn parse_log_file_name(name: &str, file_prefix: &str) -> Option<(chrono::NaiveDate, usize)> { - let body = name - .strip_prefix(&format!("{file_prefix}."))? - .strip_suffix(".log")?; - let (date, segment) = body.split_once('.').map_or((body, 0), |(date, segment)| { - (date, segment.parse().ok().unwrap_or(usize::MAX)) - }); - if segment == usize::MAX { - return None; - } - chrono::NaiveDate::parse_from_str(date, "%Y-%m-%d") - .ok() - .map(|date| (date, segment)) -} - -fn cleanup_log_files( - directory: &Path, - file_prefix: &str, - today: chrono::NaiveDate, - retention_days: usize, - maximum_files: usize, -) -> io::Result<()> { - let retention_offset = i64::try_from(retention_days.saturating_sub(1)).unwrap_or(i64::MAX); - let cutoff = today - .checked_sub_signed(chrono::Duration::days(retention_offset)) - .unwrap_or(chrono::NaiveDate::MIN); - let mut retained = Vec::new(); - for entry in managed_log_files(directory, file_prefix)? { - if entry.date < cutoff { - fs::remove_file(entry.path)?; - } else { - retained.push(entry); - } - } - retained.sort_by(|left, right| { - right - .date - .cmp(&left.date) - .then_with(|| right.segment.cmp(&left.segment)) - }); - for entry in retained.into_iter().skip(maximum_files) { - fs::remove_file(entry.path)?; - } - Ok(()) -} diff --git a/backend/crates/gateway-host/src/logging/mod.rs b/backend/crates/gateway-host/src/logging/mod.rs new file mode 100644 index 000000000..cf5018703 --- /dev/null +++ b/backend/crates/gateway-host/src/logging/mod.rs @@ -0,0 +1,230 @@ +//! 进程结构化日志与按日期/大小轮转的文件 writer。 + +use std::env; +use std::io; +use std::sync::Arc; + +use gateway_core::health::HealthProbe; +use tracing_appender::non_blocking::{NonBlockingBuilder, WorkerGuard}; +use tracing_subscriber::EnvFilter; +use tracing_subscriber::{Layer as _, layer::SubscriberExt as _, util::SubscriberInitExt as _}; + +use crate::config::LoggingConfig; + +mod sink; +mod writer; + +use sink::{FileLogGuard, FileLogSink, LogHealth}; +use writer::RotatingLogWriter; + +// 文件前缀统一为 codex-proxy-rs-；专用 tracing target 使用 snake_case。 +const APPLICATION_LOG_FILE_PREFIX: &str = "codex-proxy-rs-application"; + +/// 内部分类标识；由 host.logging.oauth_recovery 控制是否写入独立文件。 +const OAUTH_RECOVERY_LOG_TARGET: &str = "oauth_recovery"; +const OAUTH_RECOVERY_LOG_FILE_PREFIX: &str = "codex-proxy-rs-oauth-recovery"; + +/// 请求转储含完整凭据与正文，只允许在显式开启时写入独立文件。 +const REQUEST_DUMP_LOG_TARGET: &str = "request_dump"; +const REQUEST_DUMP_LOG_FILE_PREFIX: &str = "codex-proxy-rs-request-dump"; + +/// 文件日志在正常退出时排空队列并等待落盘；控制台日志单独持有守卫。 +pub struct LogGuard { + _stdout: Vec, + _files: Vec, + health: Arc, +} + +#[derive(Debug, thiserror::Error)] +pub enum LogError { + #[error("log IO failed")] + Io(#[from] io::Error), + #[error("logging filter is invalid")] + InvalidFilter, + #[error("global tracing subscriber is already initialized")] + AlreadyInitialized, + #[error("logging size limit is too large")] + SizeOverflow, +} + +/// 初始化日志；文件按完整日期留存，不以分片数量淘汰窗口内记录。 +pub fn initialize_logging(config: &LoggingConfig) -> Result { + let directive = env::var("RUST_LOG").unwrap_or_else(|_| config.level.clone()); + let file_filter = application_file_filter(&directive)?; + let recovery_file_filter = oauth_recovery_file_filter(); + let request_dump_file_filter = request_dump_file_filter(); + let stdout_filter = + EnvFilter::try_new(stdout_filter_directive(&directive, config.file.enabled)) + .map_err(|_| LogError::InvalidFilter)?; + let mut guards = Vec::new(); + let mut file_guards = Vec::new(); + let health = Arc::new(LogHealth::default()); + + let stdout_writer = config.stdout.then(|| { + let (writer, guard) = NonBlockingBuilder::default() + .thread_name("gateway-log-stdout") + .finish(io::stdout()); + guards.push(guard); + writer + }); + let file_writer = if config.file.enabled { + let (writer, guard) = create_file_writer( + config, + config.file.retention_days, + APPLICATION_LOG_FILE_PREFIX, + "gateway-log-application-file", + Arc::clone(&health), + )?; + file_guards.push(guard); + Some(writer) + } else { + None + }; + let recovery_file_writer = if config.oauth_recovery { + let (writer, guard) = create_file_writer( + config, + config.file.retention_days, + OAUTH_RECOVERY_LOG_FILE_PREFIX, + "gateway-log-oauth-recovery-file", + Arc::clone(&health), + )?; + file_guards.push(guard); + Some(writer) + } else { + None + }; + let request_dump_file_writer = if config.request_dump { + let (writer, guard) = create_file_writer( + config, + config.request_dump_retention_days, + REQUEST_DUMP_LOG_FILE_PREFIX, + "gateway-log-request-dump-file", + Arc::clone(&health), + )?; + file_guards.push(guard); + Some(writer) + } else { + None + }; + + let stdout_layer = stdout_writer.map(|writer| { + tracing_subscriber::fmt::layer() + .compact() + .with_writer(writer) + .with_target(false) + .with_ansi(false) + .with_filter(stdout_filter) + }); + let file_layer = file_writer.map(|writer| { + tracing_subscriber::fmt::layer() + .json() + .with_writer(writer) + .with_target(true) + .with_file(true) + .with_line_number(true) + .with_thread_ids(true) + .with_thread_names(true) + .with_current_span(true) + .with_span_list(true) + .with_filter(file_filter) + }); + let recovery_file_layer = recovery_file_writer.map(|writer| { + tracing_subscriber::fmt::layer() + .json() + .with_writer(writer) + .with_target(true) + .with_file(true) + .with_line_number(true) + .with_thread_ids(true) + .with_thread_names(true) + .with_current_span(true) + .with_span_list(true) + .with_filter(recovery_file_filter) + }); + let request_dump_file_layer = request_dump_file_writer.map(|writer| { + tracing_subscriber::fmt::layer() + .json() + .with_writer(writer) + .with_target(true) + .with_file(true) + .with_line_number(true) + .with_thread_ids(true) + .with_thread_names(true) + .with_current_span(true) + .with_span_list(true) + .with_filter(request_dump_file_filter) + }); + tracing_subscriber::registry() + .with(stdout_layer) + .with(file_layer) + .with(recovery_file_layer) + .with(request_dump_file_layer) + .try_init() + .map_err(|_| LogError::AlreadyInitialized)?; + Ok(LogGuard { + _stdout: guards, + _files: file_guards, + health, + }) +} + +fn application_file_filter(directive: &str) -> Result { + let recovery_directive = format!("{OAUTH_RECOVERY_LOG_TARGET}=off") + .parse() + .expect("static OAuth recovery log directive is valid"); + let request_dump_directive = format!("{REQUEST_DUMP_LOG_TARGET}=off") + .parse() + .expect("static request dump log directive is valid"); + EnvFilter::try_new(directive) + .map(|filter| filter.add_directive(recovery_directive)) + .map(|filter| filter.add_directive(request_dump_directive)) + .map_err(|_| LogError::InvalidFilter) +} + +fn oauth_recovery_file_filter() -> EnvFilter { + format!("off,{OAUTH_RECOVERY_LOG_TARGET}=info") + .parse() + .expect("static OAuth recovery log filter is valid") +} + +fn request_dump_file_filter() -> EnvFilter { + format!("off,{REQUEST_DUMP_LOG_TARGET}=info") + .parse() + .expect("static request dump log filter is valid") +} + +fn stdout_filter_directive(directive: &str, persistent_log_enabled: bool) -> String { + if !persistent_log_enabled { + format!("{directive},{OAUTH_RECOVERY_LOG_TARGET}=off,{REQUEST_DUMP_LOG_TARGET}=off") + } else { + "off,gateway_startup=info".to_owned() + } +} + +impl LogGuard { + pub(crate) fn health_probe(&self) -> Arc { + self.health.clone() + } +} + +fn create_file_writer( + config: &LoggingConfig, + retention_days: usize, + prefix: &'static str, + thread_name: &'static str, + health: Arc, +) -> Result<(FileLogSink, FileLogGuard), LogError> { + let maximum_bytes = config + .file + .max_file_size_mb + .checked_mul(1024 * 1024) + .ok_or(LogError::SizeOverflow)?; + let writer = RotatingLogWriter::open( + config.file.directory.clone(), + prefix, + maximum_bytes, + retention_days, + Arc::clone(&health), + )?; + Ok(FileLogSink::spawn(writer, thread_name, health)?) +} diff --git a/backend/crates/gateway-host/src/logging/sink.rs b/backend/crates/gateway-host/src/logging/sink.rs new file mode 100644 index 000000000..47b4a9030 --- /dev/null +++ b/backend/crates/gateway-host/src/logging/sink.rs @@ -0,0 +1,154 @@ +//! 有界文件日志队列：拥堵时背压，正常退出时排空；写入失败显式影响健康状态。 + +use std::io::{self, Write}; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::{Arc, mpsc}; +use std::thread::{self, JoinHandle}; + +use futures::future::BoxFuture; +use gateway_core::health::{HealthProbe, HealthState}; +use tracing_subscriber::fmt::MakeWriter; + +use super::writer::RotatingLogWriter; + +const QUEUE_CAPACITY: usize = 64; + +enum Command { + Write(Vec), + Shutdown, +} + +#[derive(Default)] +pub(super) struct LogHealth { + write_failures: AtomicU64, + maintenance_failures: AtomicU64, +} + +impl LogHealth { + pub(super) fn write_failed(&self, kind: io::ErrorKind) { + if self.write_failures.fetch_add(1, Ordering::Relaxed) == 0 { + eprintln!("file_logging: log completeness compromised; write failure ({kind:?})"); + } + } + + pub(super) fn maintenance_failed(&self, kind: io::ErrorKind) { + if self.maintenance_failures.fetch_add(1, Ordering::Relaxed) == 0 { + eprintln!("file_logging: archive/retention maintenance failed ({kind:?})"); + } + } +} + +impl HealthProbe for LogHealth { + fn name(&self) -> &'static str { + "file_logging" + } + + fn check(&self) -> BoxFuture<'_, HealthState> { + Box::pin(async move { + let failures = self.write_failures.load(Ordering::Relaxed); + if failures > 0 { + return HealthState::Unhealthy(format!( + "log completeness compromised: {failures} write/flush failures since startup" + )); + } + let failures = self.maintenance_failures.load(Ordering::Relaxed); + if failures > 0 { + return HealthState::Degraded(format!( + "log archive/retention maintenance failed {failures} times since startup" + )); + } + HealthState::Healthy + }) + } +} + +#[derive(Clone)] +pub(super) struct FileLogSink { + sender: mpsc::SyncSender, + health: Arc, +} + +pub(super) struct FileLogGuard { + sender: mpsc::SyncSender, + worker: Option>, + health: Arc, +} + +impl FileLogSink { + pub(super) fn spawn( + mut writer: RotatingLogWriter, + name: &'static str, + health: Arc, + ) -> io::Result<(Self, FileLogGuard)> { + let (sender, receiver) = mpsc::sync_channel(QUEUE_CAPACITY); + let worker_health = Arc::clone(&health); + let worker = thread::Builder::new() + .name(name.to_owned()) + .spawn(move || { + while let Ok(command) = receiver.recv() { + match command { + Command::Write(bytes) => { + if let Err(error) = writer.write_all(&bytes) { + worker_health.write_failed(error.kind()); + } + } + Command::Shutdown => break, + } + } + if let Err(error) = writer.sync() { + worker_health.write_failed(error.kind()); + } + })?; + Ok(( + Self { + sender: sender.clone(), + health: Arc::clone(&health), + }, + FileLogGuard { + sender, + worker: Some(worker), + health, + }, + )) + } +} + +impl<'a> MakeWriter<'a> for FileLogSink { + type Writer = Self; + + fn make_writer(&'a self) -> Self::Writer { + self.clone() + } +} + +impl Write for FileLogSink { + fn write(&mut self, buffer: &[u8]) -> io::Result { + self.sender + .send(Command::Write(buffer.to_vec())) + .map_err(|_| { + self.health.write_failed(io::ErrorKind::BrokenPipe); + io::Error::new(io::ErrorKind::BrokenPipe, "file log worker unavailable") + })?; + Ok(buffer.len()) + } + + fn flush(&mut self) -> io::Result<()> { + // Files are unbuffered. The guard is the lifecycle owner of queue drain and sync. + Ok(()) + } +} + +impl Drop for FileLogGuard { + fn drop(&mut self) { + // Shutdown follows every accepted record. Unlike a timed guard, joining cannot + // silently abandon queued log records during a normal process shutdown. + let _ = self.sender.send(Command::Shutdown); + if self + .worker + .take() + .is_some_and(|worker| worker.join().is_err()) + { + self.health.write_failed(io::ErrorKind::BrokenPipe); + } + } +} diff --git a/backend/crates/gateway-host/src/logging/writer.rs b/backend/crates/gateway-host/src/logging/writer.rs new file mode 100644 index 000000000..74e112cfd --- /dev/null +++ b/backend/crates/gateway-host/src/logging/writer.rs @@ -0,0 +1,246 @@ +//! 按大小轮转、压缩已关闭分片、按完整 UTC 日期组清理。 + +use std::collections::BTreeMap; +use std::fs::{self, File, OpenOptions}; +use std::io::{self, Write}; +use std::path::{Path, PathBuf}; +use std::sync::Arc; + +use chrono::{NaiveDate, Utc}; +use flate2::{Compression, write::GzEncoder}; + +use super::sink::LogHealth; + +pub(super) struct RotatingLogWriter { + directory: PathBuf, + prefix: &'static str, + maximum_bytes: u64, + retention_days: usize, + date: NaiveDate, + segment: usize, + bytes_written: u64, + file: File, + health: Arc, +} + +impl RotatingLogWriter { + pub(super) fn open( + directory: PathBuf, + prefix: &'static str, + maximum_bytes: u64, + retention_days: usize, + health: Arc, + ) -> io::Result { + fs::create_dir_all(&directory)?; + let date = Utc::now().date_naive(); + cleanup_log_files(&directory, prefix, date, retention_days)?; + let latest = managed_log_files(&directory, prefix)? + .into_iter() + .filter(|entry| entry.date == date) + .max_by_key(|entry| (entry.segment, entry.compressed)); + let (segment, bytes_written) = match latest { + Some(entry) if !entry.compressed && entry.path.metadata()?.len() < maximum_bytes => { + (entry.segment, entry.path.metadata()?.len()) + } + Some(entry) => (next_segment(entry.segment)?, 0), + None => (0, 0), + }; + let path = directory.join(log_file_name(prefix, date, segment)); + let file = open_log_segment(&path)?; + let writer = Self { + directory, + prefix, + maximum_bytes, + retention_days, + date, + segment, + bytes_written, + file, + health, + }; + // Recover unfinished archive work after restart; the active file is never compressed. + for entry in managed_log_files(&writer.directory, prefix)? { + if !entry.compressed + && entry.path != path + && let Err(error) = compress_log_file(&entry.path) + { + writer.health.maintenance_failed(error.kind()); + } + } + Ok(writer) + } + + fn rotate_if_required(&mut self, incoming_bytes: usize) -> io::Result<()> { + let date = Utc::now().date_naive(); + let day_changed = date != self.date; + let size_exceeded = self.bytes_written > 0 + && self.bytes_written.saturating_add(incoming_bytes as u64) > self.maximum_bytes; + if !day_changed && !size_exceeded { + return Ok(()); + } + let previous = self + .directory + .join(log_file_name(self.prefix, self.date, self.segment)); + self.file.sync_all()?; + let segment = if day_changed { + managed_log_files(&self.directory, self.prefix)? + .into_iter() + .filter(|entry| entry.date == date) + .map(|entry| entry.segment) + .max() + .map_or(Ok(0), next_segment)? + } else { + next_segment(self.segment)? + }; + let path = self + .directory + .join(log_file_name(self.prefix, date, segment)); + let file = open_log_segment(&path)?; + // Commit rotation state only after opening the new file succeeds. + self.file = file; + self.date = date; + self.segment = segment; + self.bytes_written = self.file.metadata()?.len(); + if let Err(error) = compress_log_file(&previous) { + self.health.maintenance_failed(error.kind()); + } + // Maintenance errors must not discard the record that triggered rotation. + if let Err(error) = + cleanup_log_files(&self.directory, self.prefix, date, self.retention_days) + { + self.health.maintenance_failed(error.kind()); + } + Ok(()) + } + + pub(super) fn sync(&mut self) -> io::Result<()> { + self.file.sync_all() + } +} + +impl Write for RotatingLogWriter { + fn write(&mut self, buffer: &[u8]) -> io::Result { + self.rotate_if_required(buffer.len())?; + if let Err(error) = self.file.write_all(buffer) { + self.bytes_written = self.file.metadata()?.len(); + return Err(error); + } + self.bytes_written = self.bytes_written.saturating_add(buffer.len() as u64); + Ok(buffer.len()) + } + + fn flush(&mut self) -> io::Result<()> { + self.file.flush() + } +} + +fn next_segment(segment: usize) -> io::Result { + segment + .checked_add(1) + .ok_or_else(|| io::Error::other("log segment sequence exhausted")) +} + +fn open_log_segment(path: &Path) -> io::Result { + OpenOptions::new().create(true).append(true).open(path) +} + +fn log_file_name(prefix: &str, date: NaiveDate, segment: usize) -> String { + if segment == 0 { + format!("{prefix}.{date}.log") + } else { + format!("{prefix}.{date}.{segment}.log") + } +} + +struct ManagedLogFile { + date: NaiveDate, + segment: usize, + compressed: bool, + path: PathBuf, +} + +fn managed_log_files(directory: &Path, prefix: &str) -> io::Result> { + let mut files = Vec::new(); + for entry in fs::read_dir(directory)? { + let entry = entry?; + let name = entry.file_name(); + let name = name.to_string_lossy(); + let Some(body) = name.strip_prefix(&format!("{prefix}.")) else { + continue; + }; + let (body, compressed) = match body.strip_suffix(".log.gz") { + Some(body) => (body, true), + None => match body.strip_suffix(".log") { + Some(body) => (body, false), + None => continue, + }, + }; + let (date, segment) = body + .split_once('.') + .map_or((body, Some(0)), |(date, segment)| { + (date, segment.parse::().ok()) + }); + if let (Ok(date), Some(segment)) = (NaiveDate::parse_from_str(date, "%Y-%m-%d"), segment) { + files.push(ManagedLogFile { + date, + segment, + compressed, + path: entry.path(), + }); + } + } + Ok(files) +} + +fn cleanup_log_files( + directory: &Path, + prefix: &str, + today: NaiveDate, + retention_days: usize, +) -> io::Result<()> { + let days = + u64::try_from(retention_days).map_err(|_| io::Error::other("log retention overflow"))?; + let cutoff = today + .checked_sub_days(chrono::Days::new(days)) + .unwrap_or(NaiveDate::MIN); + let mut dates = BTreeMap::>::new(); + for entry in managed_log_files(directory, prefix)? { + dates.entry(entry.date).or_default().push(entry.path); + } + for (date, paths) in dates { + if date >= cutoff { + continue; + } + // Keep a whole date if any segment was written more recently than its filename. + // This also protects archives recovered after clock corrections or manual restoration. + let mut latest = date; + for path in &paths { + let modified: chrono::DateTime = path.metadata()?.modified()?.into(); + latest = latest.max(modified.date_naive()); + } + if latest >= cutoff { + continue; + } + for path in paths { + fs::remove_file(path)?; + } + } + Ok(()) +} + +fn compress_log_file(path: &Path) -> io::Result<()> { + let mut source = File::open(path)?; + let modified = source.metadata()?.modified()?; + let archive = path.with_extension("log.gz"); + let temporary = path.with_extension("log.gz.tmp"); + // Only a closed, owner-managed segment can have this temporary archive. + let output = File::create(&temporary)?; + let mut encoder = GzEncoder::new(output, Compression::fast()); + io::copy(&mut source, &mut encoder)?; + let output = encoder.finish()?; + output.set_modified(modified)?; + output.sync_all()?; + fs::rename(&temporary, &archive)?; + // The original survives until a complete, synced archive has been published. + fs::remove_file(path) +} diff --git a/backend/crates/gateway-host/tests/config.rs b/backend/crates/gateway-host/tests/config.rs index 273c23cd7..4df0c4db0 100644 --- a/backend/crates/gateway-host/tests/config.rs +++ b/backend/crates/gateway-host/tests/config.rs @@ -80,6 +80,51 @@ fn host_config_rejects_zero_drain_window() { ); } +#[test] +fn logging_rejects_zero_retention_windows() { + for request_dump in [false, true] { + let mut config = valid_config(); + if request_dump { + config.logging.request_dump_retention_days = 0; + } else { + config.logging.file.retention_days = 0; + } + assert!( + config + .resolve_and_validate(std::path::Path::new("/srv/gateway")) + .is_err() + ); + } +} + +#[test] +fn logging_defaults_to_seven_days_and_one_day_for_payloads_and_rejects_count_retention() { + let mut value = serde_json::json!({ + "level": "info", "stdout": true, + "file": { "enabled": true, "directory": "logs", "max_file_size_mb": 20 } + }); + let config: LoggingConfig = serde_json::from_value(value.clone()).unwrap(); + assert_eq!(config.request_dump_retention_days, 1); + assert_eq!(config.file.retention_days, 7); + assert!(!config.oauth_recovery); + value["file"]["max_files"] = serde_json::json!(20); + assert!(serde_json::from_value::(value).is_err()); +} + +#[test] +fn dedicated_file_channels_can_be_enabled_without_application_sinks() { + for oauth_recovery in [false, true] { + let mut config = valid_config(); + config.logging.stdout = false; + config.logging.file.enabled = false; + config.logging.oauth_recovery = oauth_recovery; + config.logging.request_dump = !oauth_recovery; + config + .resolve_and_validate(std::path::Path::new("/srv/gateway")) + .unwrap(); + } +} + fn valid_config() -> HostConfig { let system_update = SystemUpdateConfig { update_state_file: PathBuf::from("update-state.json"), @@ -101,9 +146,10 @@ fn valid_config() -> HostConfig { directory: PathBuf::from(".runtime/logs"), retention_days: 7, max_file_size_mb: 100, - max_files: 30, }, + oauth_recovery: false, request_dump: false, + request_dump_retention_days: 1, }, system_update, drain_timeout_seconds: 30, diff --git a/backend/crates/gateway-host/tests/logging.rs b/backend/crates/gateway-host/tests/logging/mod.rs similarity index 58% rename from backend/crates/gateway-host/tests/logging.rs rename to backend/crates/gateway-host/tests/logging/mod.rs index 3f727751a..52b7ffcb5 100644 --- a/backend/crates/gateway-host/tests/logging.rs +++ b/backend/crates/gateway-host/tests/logging/mod.rs @@ -1,16 +1,20 @@ use std::env; use std::fs; +use std::io::Read; use std::path::{Path, PathBuf}; use std::process::Command; use gateway_host::config::{FileLoggingConfig, HostConfig, ListenConfig, LoggingConfig}; use gateway_host::system_update::SystemUpdateConfig; +mod sink; +mod writer; + const LOG_DIRECTORY_ENV: &str = "CPR_LOGGING_TEST_DIRECTORY"; const CHILD_PROCESS_ENV: &str = "CPR_LOGGING_TEST_CHILD"; const REQUEST_DUMP_ENABLED_ENV: &str = "CPR_LOGGING_TEST_REQUEST_DUMP_ENABLED"; -const APPLICATION_LOG_FILE_PREFIX: &str = "codex-proxy-rs."; -const OAUTH_RECOVERY_LOG_FILE_PREFIX: &str = "codex-proxy-rs-oath."; +const APPLICATION_LOG_FILE_PREFIX: &str = "codex-proxy-rs-application."; +const OAUTH_RECOVERY_LOG_FILE_PREFIX: &str = "codex-proxy-rs-oauth-recovery."; const REQUEST_DUMP_LOG_FILE_PREFIX: &str = "codex-proxy-rs-request-dump."; const APPLICATION_LOG_TARGET: &str = "logging_test_application"; const APPLICATION_LOG_MARKER: &str = "application-file-filter-test"; @@ -37,9 +41,10 @@ fn logging_requires_at_least_one_sink() { directory: PathBuf::from("logs"), retention_days: 7, max_file_size_mb: 100, - max_files: 30, }, + oauth_recovery: false, request_dump: false, + request_dump_retention_days: 1, }, system_update: SystemUpdateConfig::default(), drain_timeout_seconds: 30, @@ -143,8 +148,115 @@ fn sensitive_file_logging_is_separate_and_overrides_global_log_level() { assert!(!application_log.contains(REQUEST_DUMP_SECRET)); } +#[test] +fn oauth_recovery_switch_controls_only_its_dedicated_file() { + const OAUTH_ENABLED_ENV: &str = "CPR_LOGGING_TEST_OAUTH_ENABLED"; + const APPLICATION_ENABLED_ENV: &str = "CPR_LOGGING_TEST_APPLICATION_ENABLED"; + const TOKEN: &str = "synthetic-oauth-token-must-stay-in-recovery-file"; + if env::var_os(CHILD_PROCESS_ENV).is_some() { + let mut config = + logging_config(PathBuf::from(env::var_os(LOG_DIRECTORY_ENV).unwrap()), true); + config.logging.oauth_recovery = env::var(OAUTH_ENABLED_ENV).unwrap() == "true"; + config.logging.file.enabled = env::var(APPLICATION_ENABLED_ENV).unwrap() == "true"; + config.logging.stdout = true; + with_logging(config, || { + tracing::info!(target: APPLICATION_LOG_TARGET, marker = APPLICATION_LOG_MARKER); + tracing::info!(target: OAUTH_RECOVERY_LOG_TARGET, access_token = TOKEN); + tracing::info!(target: REQUEST_DUMP_LOG_TARGET, marker = REQUEST_DUMP_LOG_MARKER); + }); + return; + } + for application_enabled in [false, true] { + for oauth_enabled in [false, true] { + let directory = tempfile::tempdir().unwrap(); + let output = Command::new(env::current_exe().unwrap()) + .args([ + "--exact", + "logging::oauth_recovery_switch_controls_only_its_dedicated_file", + "--nocapture", + ]) + .env(CHILD_PROCESS_ENV, "1") + .env(LOG_DIRECTORY_ENV, directory.path()) + .env(OAUTH_ENABLED_ENV, oauth_enabled.to_string()) + .env(APPLICATION_ENABLED_ENV, application_enabled.to_string()) + .env("RUST_LOG", "info,oauth_recovery=trace,request_dump=trace") + .output() + .unwrap(); + assert!( + output.status.success(), + "{}", + String::from_utf8_lossy(&output.stderr) + ); + assert!(!String::from_utf8_lossy(&output.stdout).contains(TOKEN)); + assert!(!String::from_utf8_lossy(&output.stderr).contains(TOKEN)); + assert_eq!( + log_file_set_exists(directory.path(), OAUTH_RECOVERY_LOG_FILE_PREFIX), + oauth_enabled + ); + assert_eq!( + log_file_set_exists(directory.path(), APPLICATION_LOG_FILE_PREFIX), + application_enabled + ); + if oauth_enabled { + assert!( + read_log_file_set(directory.path(), OAUTH_RECOVERY_LOG_FILE_PREFIX) + .contains(TOKEN) + ); + } + if application_enabled { + let body = read_log_file_set(directory.path(), APPLICATION_LOG_FILE_PREFIX); + assert!(body.contains(APPLICATION_LOG_MARKER)); + assert!(!body.contains(TOKEN)); + } else { + assert!(String::from_utf8_lossy(&output.stdout).contains(APPLICATION_LOG_MARKER)); + } + let dump = read_log_file_set(directory.path(), REQUEST_DUMP_LOG_FILE_PREFIX); + assert!(dump.contains(REQUEST_DUMP_LOG_MARKER)); + assert!(!dump.contains(TOKEN)); + } + } +} + fn write_sensitive_logs(directory: PathBuf, request_dump: bool) { - let config = HostConfig { + with_file_logging(directory, request_dump, || { + tracing::info!( + target: APPLICATION_LOG_TARGET, + marker = APPLICATION_LOG_MARKER, + "application file test record" + ); + tracing::info!( + target: OAUTH_RECOVERY_LOG_TARGET, + provider = OAUTH_RECOVERY_PROVIDER, + marker = OAUTH_RECOVERY_LOG_MARKER, + "OAuth recovery test record" + ); + tracing::info!( + target: REQUEST_DUMP_LOG_TARGET, + marker = REQUEST_DUMP_LOG_MARKER, + authorization = REQUEST_DUMP_SECRET, + "request dump test record" + ); + }); +} + +fn with_file_logging(directory: PathBuf, request_dump: bool, write: impl FnOnce()) { + with_logging(logging_config(directory, request_dump), write); +} + +fn with_logging(config: HostConfig, write: impl FnOnce()) { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("create logging runtime"); + let bundle = runtime + .block_on(gateway_host::initialize(config)) + .expect("initialize logging"); + write(); + drop(bundle); +} + +fn logging_config(directory: PathBuf, request_dump: bool) -> HostConfig { + HostConfig { listen: ListenConfig { host: "127.0.0.1".to_owned(), port: 8080, @@ -156,42 +268,17 @@ fn write_sensitive_logs(directory: PathBuf, request_dump: bool) { file: FileLoggingConfig { enabled: true, directory, - retention_days: 1, + retention_days: 7, max_file_size_mb: 1, - max_files: 1, }, + oauth_recovery: true, request_dump, + request_dump_retention_days: 1, }, system_update: SystemUpdateConfig::default(), drain_timeout_seconds: 30, worker_shutdown_timeout_seconds: 30, - }; - let runtime = tokio::runtime::Builder::new_current_thread() - .enable_all() - .build() - .expect("create logging runtime"); - let bundle = runtime - .block_on(gateway_host::initialize(config)) - .expect("initialize logging"); - - tracing::info!( - target: APPLICATION_LOG_TARGET, - marker = APPLICATION_LOG_MARKER, - "application file test record" - ); - tracing::info!( - target: OAUTH_RECOVERY_LOG_TARGET, - provider = OAUTH_RECOVERY_PROVIDER, - marker = OAUTH_RECOVERY_LOG_MARKER, - "OAuth recovery test record" - ); - tracing::info!( - target: REQUEST_DUMP_LOG_TARGET, - marker = REQUEST_DUMP_LOG_MARKER, - authorization = REQUEST_DUMP_SECRET, - "request dump test record" - ); - drop(bundle); + } } fn read_log_file_set(directory: &Path, file_prefix: &str) -> String { @@ -201,7 +288,10 @@ fn read_log_file_set(directory: &Path, file_prefix: &str) -> String { .filter(|path| { path.file_name() .and_then(|name| name.to_str()) - .is_some_and(|name| name.starts_with(file_prefix)) + .is_some_and(|name| { + name.starts_with(file_prefix) + && (name.ends_with(".log") || name.ends_with(".log.gz")) + }) }) .collect::>(); files.sort(); @@ -209,7 +299,17 @@ fn read_log_file_set(directory: &Path, file_prefix: &str) -> String { assert!(!files.is_empty(), "expected log file set {file_prefix}"); files .into_iter() - .map(|path| fs::read_to_string(path).expect("read log file")) + .map(|path| { + if path.extension().is_some_and(|extension| extension == "gz") { + let mut body = String::new(); + flate2::read::GzDecoder::new(fs::File::open(path).expect("open archive")) + .read_to_string(&mut body) + .expect("read complete archive"); + body + } else { + fs::read_to_string(path).expect("read log file") + } + }) .collect() } @@ -218,10 +318,10 @@ fn log_file_set_exists(directory: &Path, file_prefix: &str) -> bool { .expect("read log directory") .filter_map(Result::ok) .any(|entry| { - entry - .file_name() - .to_str() - .is_some_and(|name| name.starts_with(file_prefix)) + entry.file_name().to_str().is_some_and(|name| { + name.starts_with(file_prefix) + && (name.ends_with(".log") || name.ends_with(".log.gz")) + }) }) } diff --git a/backend/crates/gateway-host/tests/logging/sink.rs b/backend/crates/gateway-host/tests/logging/sink.rs new file mode 100644 index 000000000..7c9ede72a --- /dev/null +++ b/backend/crates/gateway-host/tests/logging/sink.rs @@ -0,0 +1,108 @@ +use super::*; +use gateway_core::health::HealthState; +use std::time::Duration; + +#[test] +fn file_queue_drains_every_record_before_normal_shutdown() { + if env::var_os(CHILD_PROCESS_ENV).is_some() { + with_file_logging( + PathBuf::from(env::var_os(LOG_DIRECTORY_ENV).unwrap()), + true, + || { + let payload = "x".repeat(16 * 1024); + for sequence in 0..2000 { + tracing::info!(target: REQUEST_DUMP_LOG_TARGET, sequence, payload, "queue record"); + } + }, + ); + return; + } + let directory = tempfile::tempdir().unwrap(); + let output = Command::new(env::current_exe().unwrap()) + .args([ + "--exact", + "logging::sink::file_queue_drains_every_record_before_normal_shutdown", + ]) + .env(CHILD_PROCESS_ENV, "1") + .env(LOG_DIRECTORY_ENV, directory.path()) + .env("RUST_LOG", "off") + .output() + .unwrap(); + assert!( + output.status.success(), + "{}", + String::from_utf8_lossy(&output.stderr) + ); + let body = read_log_file_set(directory.path(), REQUEST_DUMP_LOG_FILE_PREFIX); + let mut sequences = body + .lines() + .map(|line| { + serde_json::from_str::(line).unwrap()["fields"]["sequence"] + .as_u64() + .unwrap() + }) + .collect::>(); + sequences.sort_unstable(); + assert_eq!(sequences, (0..2000).collect::>()); +} + +#[test] +fn write_failure_marks_log_completeness_unhealthy_even_after_writes_recover() { + if env::var_os(CHILD_PROCESS_ENV).is_some() { + let directory = PathBuf::from(env::var_os(LOG_DIRECTORY_ENV).unwrap()); + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap(); + let bundle = runtime + .block_on(gateway_host::initialize(logging_config( + directory.clone(), + true, + ))) + .unwrap(); + let health = bundle.logging_health_probe(); + assert_eq!(runtime.block_on(health.check()), HealthState::Healthy); + let blocked = directory.join(format!( + "{REQUEST_DUMP_LOG_FILE_PREFIX}{}.1.log", + chrono::Utc::now().date_naive() + )); + fs::create_dir(&blocked).unwrap(); + let payload = "x".repeat(1024 * 1024); + tracing::info!(target: REQUEST_DUMP_LOG_TARGET, payload, marker="before-failure"); + tracing::info!(target: REQUEST_DUMP_LOG_TARGET, payload, marker="cannot-write"); + runtime.block_on(async { + tokio::time::timeout(Duration::from_secs(5), async { + loop { + if matches!(health.check().await, HealthState::Unhealthy(_)) { + break; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .unwrap(); + }); + fs::remove_dir(&blocked).unwrap(); + tracing::info!(target: REQUEST_DUMP_LOG_TARGET, marker="after-recovery"); + drop(bundle); + assert!(matches!( + runtime.block_on(health.check()), + HealthState::Unhealthy(_) + )); + return; + } + let directory = tempfile::tempdir().unwrap(); + let output = Command::new(env::current_exe().unwrap()) + .args(["--nocapture", "--exact", "logging::sink::write_failure_marks_log_completeness_unhealthy_even_after_writes_recover"]) + .env(CHILD_PROCESS_ENV, "1").env(LOG_DIRECTORY_ENV, directory.path()) + .env("RUST_LOG", "off").output().unwrap(); + assert!( + output.status.success(), + "{}", + String::from_utf8_lossy(&output.stderr) + ); + assert!(String::from_utf8_lossy(&output.stderr).contains("log completeness compromised")); + let body = read_log_file_set(directory.path(), REQUEST_DUMP_LOG_FILE_PREFIX); + assert!(body.contains("before-failure") && body.contains("after-recovery")); + assert!(!body.contains("cannot-write")); +} diff --git a/backend/crates/gateway-host/tests/logging/writer.rs b/backend/crates/gateway-host/tests/logging/writer.rs new file mode 100644 index 000000000..0ffbe0c2a --- /dev/null +++ b/backend/crates/gateway-host/tests/logging/writer.rs @@ -0,0 +1,118 @@ +use super::*; +use std::time::SystemTime; + +#[test] +fn retention_preserves_complete_dates_across_compression_rotation_and_restart() { + if env::var_os(CHILD_PROCESS_ENV).is_some() { + with_file_logging( + PathBuf::from(env::var_os(LOG_DIRECTORY_ENV).unwrap()), + true, + || { + let payload = "x".repeat(1024 * 1024); + for sequence in 0..25 { + tracing::info!(target: REQUEST_DUMP_LOG_TARGET, sequence, payload, "retention record"); + tracing::info!(target: APPLICATION_LOG_TARGET, sequence, payload, "retention record"); + tracing::info!(target: OAUTH_RECOVERY_LOG_TARGET, sequence, payload, "retention record"); + } + }, + ); + return; + } + let directory = tempfile::tempdir().unwrap(); + let today = chrono::Utc::now().date_naive(); + let mut retained = Vec::new(); + let mut expired = Vec::new(); + for (prefix, days) in [ + (APPLICATION_LOG_FILE_PREFIX, 7), + (OAUTH_RECOVERY_LOG_FILE_PREFIX, 7), + (REQUEST_DUMP_LOG_FILE_PREFIX, 1), + ] { + let boundary = today - chrono::Days::new(days); + for segment in 1..=25 { + let path = directory + .path() + .join(format!("{prefix}{boundary}.{segment}.log")); + seed_log(&path, boundary, &format!("boundary-{segment}\n")); + retained.push(path); + } + let old_date = boundary - chrono::Days::new(1); + let path = directory.path().join(format!("{prefix}{old_date}.log")); + seed_log(&path, old_date, "expired date\n"); + expired.push(path); + // If any old segment was recently written, protect its whole UTC date. + let restored_date = old_date - chrono::Days::new(1); + for segment in 0..=1 { + let path = directory + .path() + .join(format!("{prefix}{restored_date}.{segment}.log")); + seed_log( + &path, + if segment == 0 { today } else { restored_date }, + "restored date\n", + ); + retained.push(path); + } + } + let unrelated = directory.path().join("unmanaged.log"); + fs::write(&unrelated, "unmanaged\n").unwrap(); + // A crash before archive publication leaves the original and possibly a partial temporary file. + fs::write( + retained[0].with_extension("log.gz.tmp"), + "unfinished archive", + ) + .unwrap(); + for run in 1..=2 { + let output = Command::new(env::current_exe().unwrap()) + .args(["--exact", "logging::writer::retention_preserves_complete_dates_across_compression_rotation_and_restart"]) + .env(CHILD_PROCESS_ENV, "1").env(LOG_DIRECTORY_ENV, directory.path()) + .env("RUST_LOG", "off,logging_test_application=info").output().unwrap(); + assert!( + output.status.success(), + "{}", + String::from_utf8_lossy(&output.stderr) + ); + assert!( + retained + .iter() + .all(|path| path.with_extension("log.gz").exists()) + ); + assert!( + expired + .iter() + .all(|path| !path.exists() && !path.with_extension("log.gz").exists()) + ); + assert!(unrelated.exists()); + + for prefix in [ + APPLICATION_LOG_FILE_PREFIX, + OAUTH_RECOVERY_LOG_FILE_PREFIX, + REQUEST_DUMP_LOG_FILE_PREFIX, + ] { + let body = read_log_file_set(directory.path(), prefix); + for sequence in 0..25 { + assert_eq!( + body.matches(&format!(r#""sequence":{sequence},"#)).count(), + run + ); + } + for segment in 1..=25 { + assert!( + body.contains(&format!("boundary-{segment}\n")), + "whole boundary date must survive" + ); + } + } + } +} + +fn seed_log(path: &Path, date: chrono::NaiveDate, body: &str) { + fs::write(path, body).unwrap(); + let modified = chrono::DateTime::::from_naive_utc_and_offset( + date.and_hms_opt(0, 0, 0).unwrap(), + chrono::Utc, + ); + fs::File::open(path) + .unwrap() + .set_modified(SystemTime::from(modified)) + .unwrap(); +} diff --git a/backend/crates/gateway-store/tests/postgres/execution.rs b/backend/crates/gateway-store/tests/postgres/execution.rs index f6c0267ee..d35f77ebf 100644 --- a/backend/crates/gateway-store/tests/postgres/execution.rs +++ b/backend/crates/gateway-store/tests/postgres/execution.rs @@ -1111,21 +1111,36 @@ async fn diagnostic_trace_is_finalized_atomically_and_available_for_failed_reque let mut finalization = successful_core_finalization("req_diagnostic_failed"); finalization.outcome = ExecutionOutcome::Failed; finalization.client_status_code = Some(502); - finalization.error = Some(GatewayError::new( - GatewayErrorKind::UpstreamUnavailable, - "upstream closed", - )); + 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": "upstream.close", "data": {"code": 1000}}, + {"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) = sqlx::query_as( - "select outcome, diagnostic_trace_json from model_requests where id = 'req_diagnostic_failed'" + let persisted: (String, serde_json::Value, String) = sqlx::query_as( + "select outcome, diagnostic_trace_json, error_message from model_requests where id = 'req_diagnostic_failed'" ).fetch_one(&database.pool).await.unwrap(); - assert_eq!(persisted, ("failed".to_owned(), trace.clone())); + 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") diff --git a/backend/crates/providers/openai/src/credential/recovery_log.rs b/backend/crates/providers/openai/src/credential/recovery_log.rs index fbb2d4337..301db9386 100644 --- a/backend/crates/providers/openai/src/credential/recovery_log.rs +++ b/backend/crates/providers/openai/src/credential/recovery_log.rs @@ -1,7 +1,7 @@ //! 成功取得 OAuth token 后写入独立恢复日志的记录。 //! //! 此日志会刻意保存原始 AT/RT，避免后续数据库写入或校验失败时，已成功 -//! 交换的账号无法恢复。记录独立落盘，但沿用普通结构化日志的滚动和清理策略。 +//! 交换的账号无法恢复。Host 在 oauth_recovery 配置开启时独立落盘，沿用文件日志的滚动和清理策略。 const OAUTH_RECOVERY_LOG_TARGET: &str = "oauth_recovery"; const OAUTH_RECOVERY_PROVIDER: &str = "openai"; @@ -36,8 +36,8 @@ impl CodexOAuthRecoveryOperation { /// 将已取得的 AT/RT 写入独立恢复结构化日志。 /// /// 调用方必须在任何可能拒绝响应的本地校验、画像补全或数据库写入之前调用。 -/// 该调用不执行额外 I/O，也不返回业务错误。Host 负责恢复日志文件的分割、清理和 -/// 输出；日志系统不可用时按其既有策略处理，不能阻断 OAuth 业务流程。 +/// 此处只发出结构化事件，不返回业务错误。Host 的 oauth_recovery 配置控制落盘， +/// 并负责文件分割、清理和写入队列背压。 pub(crate) fn record_oauth_recovery( operation: CodexOAuthRecoveryOperation, account_id: Option<&str>, diff --git a/backend/crates/providers/openai/src/provider/failure.rs b/backend/crates/providers/openai/src/provider/failure.rs index b11781931..7500ca02a 100644 --- a/backend/crates/providers/openai/src/provider/failure.rs +++ b/backend/crates/providers/openai/src/provider/failure.rs @@ -703,10 +703,7 @@ pub(super) fn map_client_error( uncertain_state: UpstreamSendState, observe_transport: bool, ) -> MappedProviderFailure { - let websocket_diagnostic = match &error { - CodexClientError::WebSocket(error) => Some(websocket_diagnostic(error)), - _ => None, - }; + let diagnostic = client_diagnostic(&error); let raw_upstream_error = match &error { CodexClientError::WebSocket(error) => websocket_raw_error(error), _ => None, @@ -838,7 +835,7 @@ pub(super) fn map_client_error( ContinuationRecoveryDisposition::ClientReplayRequired, ); } - if let Some(diagnostic) = websocket_diagnostic { + if let Some(diagnostic) = diagnostic { failure.error = failure.error.with_diagnostic(diagnostic); } if let Some(raw) = raw_upstream_error { @@ -853,7 +850,119 @@ pub(super) fn map_client_error( failure } +/// 在消费 transport 错误前提取可持久化事实，不能使用可能携带 URL/凭据的 Display。 +fn client_diagnostic(error: &CodexClientError) -> Option { + let (stage, code, message) = match error { + CodexClientError::WebSocket(error) => return Some(websocket_diagnostic(error)), + CodexClientError::Http(error) | CodexClientError::HttpJson(error) => { + let (stage, code) = if error.is_builder() { + ("prepare", "http_request_build_failed") + } else if error.is_connect() && error.is_timeout() { + ("connect", "http_connect_timeout") + } else if error.is_connect() { + ("connect", "http_connect_failed") + } else if error.is_timeout() { + ("exchange", "http_timeout") + } else if error.is_decode() { + ("decode", "http_decode_failed") + } else if error.is_body() { + ("receive", "http_body_failed") + } else { + ("exchange", "http_transport_failed") + }; + return Some( + ProviderDiagnostic::new(format!( + "OpenAI HTTP failure: stage={stage}, cause={code}" + )) + .with_classification(stage, code) + .with_io_cause(error), + ); + } + CodexClientError::StreamIdleTimeout { timeout } => ( + "receive", + "receive_idle_timeout", + format!("OpenAI HTTP response event idle timeout after {timeout:?}"), + ), + CodexClientError::InvalidSse(_) => ( + "decode", + "invalid_sse", + "OpenAI HTTP returned an invalid Responses event stream".to_owned(), + ), + CodexClientError::InvalidHeaderName(_) => ( + "prepare", + "invalid_header_name", + "OpenAI request header name is invalid".to_owned(), + ), + CodexClientError::InvalidHeaderValue(_) => ( + "prepare", + "invalid_header_value", + "OpenAI request header value is invalid".to_owned(), + ), + CodexClientError::WebSocketEncode(_) | CodexClientError::RequestBodyEncode(_) => ( + "prepare", + "request_encode_failed", + "OpenAI request encoding failed".to_owned(), + ), + CodexClientError::RequestCompression(_) => ( + "prepare", + "request_compression_failed", + "OpenAI request compression failed".to_owned(), + ), + CodexClientError::CustomCa(_) => ( + "prepare", + "custom_ca_invalid", + "OpenAI custom CA configuration is invalid".to_owned(), + ), + CodexClientError::ModelCatalog(_) => ( + "prepare", + "model_catalog_invalid", + "OpenAI model catalog validation failed".to_owned(), + ), + CodexClientError::Upstream { .. } => return None, + }; + Some(ProviderDiagnostic::new(message).with_classification(stage, code)) +} + fn websocket_diagnostic(error: &CodexWebSocketExchangeError) -> ProviderDiagnostic { + let (stage, code) = match error.diagnostic_cause() { + CodexWebSocketExchangeError::Connect(_) => ( + "connect", + error + .transport_failure_reason() + .unwrap_or("websocket_connect_failed"), + ), + CodexWebSocketExchangeError::ConnectTimeout { .. } => ("connect", "connect_timeout"), + CodexWebSocketExchangeError::SharedConnectFailed => ("connect", "shared_connect_failed"), + CodexWebSocketExchangeError::OriginCircuitOpen => ("connect", "origin_circuit_open"), + CodexWebSocketExchangeError::OriginHalfOpenBusy => ("connect", "origin_half_open_busy"), + CodexWebSocketExchangeError::InvalidRequest(_) => ("prepare", "invalid_request"), + CodexWebSocketExchangeError::SendTimeout { .. } => ("send", "send_timeout"), + CodexWebSocketExchangeError::ReceiveIdleTimeout { .. } => { + ("receive", "receive_idle_timeout") + } + CodexWebSocketExchangeError::InvalidSse(_) => ("decode", "invalid_sse"), + CodexWebSocketExchangeError::UnexpectedBinaryEvent => ("decode", "unexpected_binary_event"), + CodexWebSocketExchangeError::ClosedBeforeTerminal(_) => { + ("receive", "closed_before_terminal") + } + CodexWebSocketExchangeError::ConnectionLimitReached => { + ("upstream", "connection_limit_reached") + } + CodexWebSocketExchangeError::ContinuationUnavailable { .. } => { + ("prepare", "continuation_unavailable") + } + CodexWebSocketExchangeError::Upstream(_) => ("connect", "websocket_opening_rejected"), + _ => ( + "exchange", + error + .transport_failure_reason() + .unwrap_or("websocket_transport_failed"), + ), + }; + websocket_diagnostic_message(error).with_classification(stage, code) +} + +fn websocket_diagnostic_message(error: &CodexWebSocketExchangeError) -> ProviderDiagnostic { if let Some(close) = error.close_before_terminal() { let mut message = close.code().map_or_else( || "OpenAI WebSocket closed before a terminal response".to_owned(), @@ -868,7 +977,25 @@ fn websocket_diagnostic(error: &CodexWebSocketExchangeError) -> ProviderDiagnost return ProviderDiagnostic::new(message); } - ProviderDiagnostic::new(match error.classified() { + if let Some(reason) = error.transport_failure_reason() { + if matches!( + error.diagnostic_cause(), + CodexWebSocketExchangeError::Connect(_) + ) { + return ProviderDiagnostic::new(format!( + "OpenAI WebSocket connection failed before payload send ({reason})" + )); + } + return ProviderDiagnostic::new(if reason == "reset_without_closing_handshake" { + "OpenAI WebSocket disconnected without a closing handshake after payload send; result is ambiguous".to_owned() + } else { + format!( + "OpenAI WebSocket transport failed after payload send ({reason}); result is ambiguous" + ) + }); + } + + ProviderDiagnostic::new(match error.diagnostic_cause() { CodexWebSocketExchangeError::InvalidRequest(_) => { "OpenAI WebSocket request could not be constructed".to_owned() } @@ -1041,10 +1168,13 @@ pub(super) fn map_upstream_failure( .upstream_code() .map_or_else(|| "none".to_owned(), |code| code.as_str().to_owned()); let kind = error.kind().as_str(); - error = error.with_diagnostic(ProviderDiagnostic::new(format!( - "OpenAI upstream failure: kind={}, status={status}, code={code}", - kind - ))); + error = error.with_diagnostic( + ProviderDiagnostic::new(format!( + "OpenAI upstream failure: kind={}, status={status}, code={code}", + kind + )) + .with_classification("upstream", "upstream_rejected"), + ); MappedProviderFailure { error, websocket_transport_retryable: false, diff --git a/backend/crates/providers/openai/src/transport/websocket/coordinator.rs b/backend/crates/providers/openai/src/transport/websocket/coordinator.rs index d8e4d67f8..e01665a4e 100644 --- a/backend/crates/providers/openai/src/transport/websocket/coordinator.rs +++ b/backend/crates/providers/openai/src/transport/websocket/coordinator.rs @@ -597,13 +597,20 @@ pub(crate) async fn execute_prepared_response_create_request_stream( })); trace.capture("upstream.request.body", request.payload_text().as_bytes()); if let Err(error) = send_websocket_request(&websocket, request.payload_text()).await { - trace.record( - "upstream.send.failed", - serde_json::json!({"phase": "websocket_payload", "sendState": "ambiguous"}), - ); let observation = websocket .observation() .with_exit_reason("outbound_transport_error"); + trace.record( + "upstream.send.failed", + serde_json::json!({ + "phase": "websocket_payload", "sendState": "ambiguous", + "failureReason": error.transport_failure_reason().unwrap_or(observation.exit_reason()), + "connectionId": observation.connection_id().to_string(), + "connectionAgeMs": observation.age_ms(), + "connectionIdleMs": observation.idle_ms(), + "reused": reused, + }), + ); discard_after_send(websocket, lease, observation.clone()).await; return Err(post_send_ambiguous( error.with_connection_observation(observation), diff --git a/backend/crates/providers/openai/src/transport/websocket/error.rs b/backend/crates/providers/openai/src/transport/websocket/error.rs index f73da7213..907c9037a 100644 --- a/backend/crates/providers/openai/src/transport/websocket/error.rs +++ b/backend/crates/providers/openai/src/transport/websocket/error.rs @@ -223,6 +223,50 @@ impl fmt::Display for CodexWebSocketUpstreamError { } impl CodexWebSocketExchangeError { + /// 仅供诊断展示穿透发送状态与连接快照包装；重试分类仍使用 classified。 + pub(crate) fn diagnostic_cause(&self) -> &Self { + match self { + Self::PostSendAmbiguous { + source: Some(source), + .. + } + | Self::ReusedConnectionDiedBeforeFirstEvent { + source: Some(source), + .. + } + | Self::ConnectionObserved { source, .. } => source.diagnostic_cause(), + _ => self, + } + } + + /// 返回可持久化的传输错误分类，不包含底层错误中的地址、报文或凭据。 + /// 与已有指标粗分类分开：缺少关闭握手本身不能证明收到 TCP RST。 + pub(crate) fn transport_failure_reason(&self) -> Option<&'static str> { + match self.diagnostic_cause() { + Self::Transport(error) | Self::Connect(error) => Some(match error { + tungstenite::Error::Protocol( + tungstenite::error::ProtocolError::ResetWithoutClosingHandshake, + ) => "reset_without_closing_handshake", + tungstenite::Error::Io(error) => match error.kind() { + std::io::ErrorKind::ConnectionRefused => "connection_refused", + std::io::ErrorKind::NetworkUnreachable => "network_unreachable", + std::io::ErrorKind::HostUnreachable => "host_unreachable", + std::io::ErrorKind::ConnectionReset => "tcp_reset", + std::io::ErrorKind::ConnectionAborted => "connection_aborted", + std::io::ErrorKind::BrokenPipe => "broken_pipe", + std::io::ErrorKind::UnexpectedEof => "unexpected_eof", + std::io::ErrorKind::TimedOut => "transport_timeout", + _ => "io_error", + }, + tungstenite::Error::Tls(_) => "tls_error", + tungstenite::Error::Protocol(_) => "protocol_error", + tungstenite::Error::Capacity(_) => "capacity_error", + _ => "transport_error", + }), + _ => None, + } + } + pub(crate) fn closed_before_terminal_on( connection_id: Uuid, code: Option, diff --git a/backend/crates/providers/openai/src/transport/websocket/exchange/stream.rs b/backend/crates/providers/openai/src/transport/websocket/exchange/stream.rs index 429cf449a..379add7d2 100644 --- a/backend/crates/providers/openai/src/transport/websocket/exchange/stream.rs +++ b/backend/crates/providers/openai/src/transport/websocket/exchange/stream.rs @@ -185,14 +185,19 @@ async fn forward_websocket_response_stream(state: WebSocketStreamForwardState) { let message = match message { Ok(message) => message, Err(error) => { + let observation = connection_observation(&websocket, &error); trace.record( "upstream.read.failed", json!({ "lastEventType": last_event_type, + "failureReason": error.transport_failure_reason().unwrap_or_else(|| observation.exit_reason()), + "connectionId": observation.connection_id().to_string(), + "connectionAgeMs": observation.age_ms(), + "connectionIdleMs": observation.idle_ms(), + "reused": reused_connection, "error": diagnostic_json(&json!({"message": error.to_string()})), }), ); - let observation = connection_observation(&websocket, &error); discard_stream_websocket( websocket, pool_return, diff --git a/backend/crates/providers/openai/tests/provider/contract.rs b/backend/crates/providers/openai/tests/provider/contract.rs index a64b8ed54..92b3a48f9 100644 --- a/backend/crates/providers/openai/tests/provider/contract.rs +++ b/backend/crates/providers/openai/tests/provider/contract.rs @@ -423,6 +423,7 @@ fn planned_provider_endpoint_request(provider_name: &str, operation: Operation) fn contract_account_scope() -> Arc { let provider = ProviderKind::new("openai").expect("provider"); let accounts = [ + "acct_abrupt_disconnect", "acct_affinity", "acct_affinity_switch_a", "acct_affinity_switch_b", @@ -913,6 +914,11 @@ async fn truncated_http_stream_allows_account_rotation_only_before_client_delive assert_eq!(error.kind(), ProviderErrorKind::Transport); assert_eq!(error.send_state(), UpstreamSendState::Sent); + assert_eq!( + error.diagnostic().and_then(|diagnostic| diagnostic.code()), + Some("unexpected_eof") + ); + assert!(!error.diagnostic().unwrap().as_str().contains("127.0.0.1")); assert!(error.allows_pre_delivery_retry()); assert!(!error.replay_is_safe()); } @@ -2211,6 +2217,168 @@ async fn websocket_fast_path_miss_uses_http_and_keeps_background_preconnect() { server.await.expect("HTTP and WebSocket upstream server"); } +#[tokio::test] +async fn connection_failure_preserves_io_cause_without_claiming_payload_was_sent() { + for websocket in [false, true] { + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, "acct_abrupt_disconnect").await; + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let base_url = format!("http://{}", listener.local_addr().unwrap()); + drop(listener); + let operation = if websocket { + Operation::Generate(generate_with_persisted_session_context( + "acct_abrupt_disconnect", + "conversation-refused", + "session-refused", + "turn-refused", + )) + } else { + http_generate_operation() + }; + let mut stream = provider_with_base_url(&store, base_url) + .execute( + planned_request("openai", operation), + context("req_connect_refused", CancellationToken::new()), + ) + .await + .expect("prepare provider stream"); + let error = loop { + match stream.next().await { + Some(Ok(_)) => {} + Some(Err(error)) => break error, + None => panic!("closed listener must fail"), + } + }; + let diagnostic = error.diagnostic().expect("connection diagnosis"); + assert_eq!(diagnostic.code(), Some("connection_refused")); + assert_eq!(diagnostic.stage(), Some("connect")); + assert_eq!(error.send_state(), UpstreamSendState::NotSent); + assert!(!diagnostic.as_str().contains("after payload send")); + assert!(!diagnostic.as_str().contains("127.0.0.1")); + } +} + +#[tokio::test] +async fn abrupt_websocket_disconnect_preserves_diagnosis_and_ambiguous_send_state() { + const ACCOUNT_ID: &str = "acct_abrupt_disconnect"; + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, ACCOUNT_ID).await; + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let base_url = format!("http://{}", 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; + websocket.next().await.unwrap().unwrap(); + // The peer disappears after receiving the payload, without sending a Close frame. + }); + let operation = Operation::Generate(generate_with_persisted_session_context( + ACCOUNT_ID, + "conversation-abrupt-disconnect", + "session-abrupt-disconnect", + "turn-abrupt-disconnect", + )); + let mut stream = provider_with_base_url(&store, base_url) + .execute( + planned_request("openai", operation), + context("req_abrupt_disconnect", CancellationToken::new()), + ) + .await + .expect("prepare WebSocket provider stream"); + let error = loop { + match stream.next().await { + Some(Ok(_)) => {} + Some(Err(error)) => break error, + None => panic!("abrupt disconnect must surface a provider error"), + } + }; + server.await.unwrap(); + assert_eq!(error.kind(), ProviderErrorKind::Transport); + assert_eq!(error.send_state(), UpstreamSendState::Ambiguous); + assert!(!error.replay_is_safe()); + assert_eq!(error.pre_delivery_retry(), None); + assert_eq!( + error.diagnostic().map(|diagnostic| diagnostic.as_str()), + Some( + "OpenAI WebSocket disconnected without a closing handshake after payload send; result is ambiguous" + ), + ); +} + +#[tokio::test] +async fn websocket_idle_timeout_diagnosis_survives_ambiguous_send_wrapping() { + const ACCOUNT_ID: &str = "acct_abrupt_disconnect"; + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, ACCOUNT_ID).await; + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let base_url = format!("http://{}", 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; + websocket.next().await.unwrap().unwrap(); + websocket.send(Message::Text(json!({ + "type": "response.created", "response": {"id": "resp_idle", "model": "gpt-5.4"}, + }).to_string().into())).await.unwrap(); + websocket + .send(Message::Text( + json!({"type": "response.output_text.delta", "delta": "partial"}) + .to_string() + .into(), + )) + .await + .unwrap(); + futures::future::pending::<()>().await; + }); + let operation = Operation::Generate(generate_with_persisted_session_context( + ACCOUNT_ID, + "conversation-idle-timeout", + "session-idle-timeout", + "turn-idle-timeout", + )); + let mut stream = provider_with_base_url(&store, base_url) + .execute( + planned_request("openai", operation), + context("req_idle_timeout", CancellationToken::new()), + ) + .await + .expect("prepare WebSocket provider stream"); + loop { + let event = stream + .next() + .await + .expect("partial response") + .expect("valid partial response"); + if event.has_client_event() { + break; + } + } + tokio::time::pause(); + tokio::time::advance(Duration::from_secs(300)).await; + let error = loop { + match stream.next().await { + Some(Ok(_)) => {} + Some(Err(error)) => break error, + None => panic!("idle timeout must surface a provider error"), + } + }; + server.abort(); + assert_eq!(error.send_state(), UpstreamSendState::Ambiguous); + assert!(!error.replay_is_safe()); + assert_eq!(error.pre_delivery_retry(), None); + assert_eq!( + error.diagnostic().and_then(|diagnostic| diagnostic.code()), + Some("receive_idle_timeout") + ); + assert_eq!( + error.diagnostic().and_then(|diagnostic| diagnostic.stage()), + Some("receive") + ); + assert_eq!( + error.diagnostic().map(|diagnostic| diagnostic.as_str()), + Some("OpenAI WebSocket receive idle timeout after 300s"), + ); +} + #[tokio::test] async fn ambiguous_websocket_close_reconnects_and_retains_native_continuation() { const ACCOUNT_ID: &str = "acct_websocket_close"; diff --git a/backend/crates/providers/openai/tests/transport/websocket/diagnostics.rs b/backend/crates/providers/openai/tests/transport/websocket/diagnostics.rs new file mode 100644 index 000000000..add2877b9 --- /dev/null +++ b/backend/crates/providers/openai/tests/transport/websocket/diagnostics.rs @@ -0,0 +1,107 @@ +use super::*; +use gateway_core::diagnostics::TraceContext; + +#[tokio::test] +async fn diagnostics_preserve_abrupt_disconnect_facts_without_response_content() { + for (reuse, send_event) in [(false, false), (false, true), (true, false), (true, true)] { + 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; + websocket.next().await.unwrap().unwrap(); + if reuse { + websocket + .send(Message::Text( + completed_websocket_response("resp_warmup", 1, 1).into(), + )) + .await + .unwrap(); + websocket.next().await.unwrap().unwrap(); + } + if send_event { + websocket + .send(Message::Text( + json!({"type": "codex.response.metadata", "private": "private-response-body"}) + .to_string() + .into(), + )) + .await + .unwrap(); + } + // Drop the transport without the WebSocket close handshake. + }); + let trace = TraceContext::new("req_abrupt_disconnect"); + let attempt = trace.attempt(1); + let mut request = codex_request("gpt-5.5", "private-request-body", Vec::new()); + request.set_previous_response_id(Some("resp_previous".to_owned())); + request.previous_response_scope = Some(PreviousResponseScope::Persisted); + request.force_http_sse = false; + request.local_conversation_id = Some("diagnostics-conversation".to_owned()); + let backend = CodexBackendClient::new( + reqwest::Client::builder().no_proxy().build().unwrap(), + format!("http://{addr}"), + test_wire_profile(), + ) + .with_websocket_pool(Arc::new(CodexWebSocketPool::new(Duration::from_mins(1)))); + if reuse { + backend + .create_response( + &request, + request_context("req_warmup", Some("chatgpt-account")), + ) + .await + .expect("warmup response must complete"); + } + let result = backend + .create_response_stream( + &request, + request_context("req_abrupt_disconnect", Some("chatgpt-account")) + .with_trace(&attempt), + ) + .await; + let mut failed = result.is_err(); + if let Ok(response) = result { + let mut body = response.body; + while let Some(chunk) = body.next().await { + if chunk.is_err() { + failed = true; + break; + } + } + } + assert!(failed, "an abrupt disconnect must fail the exchange"); + server.await.unwrap(); + let snapshot = trace.snapshot().unwrap(); + let events = snapshot["events"].as_array().unwrap(); + let connection = events + .iter() + .find(|event| event["stage"] == "upstream.connection") + .unwrap(); + let failure = events + .iter() + .find(|event| event["stage"] == "upstream.read.failed") + .unwrap(); + assert_eq!( + failure["data"]["failureReason"], + "reset_without_closing_handshake" + ); + assert_eq!( + failure["data"]["connectionId"], + connection["data"]["connectionId"] + ); + assert_eq!(failure["data"]["reused"], reuse); + assert_eq!( + failure["data"]["lastEventType"], + if send_event { + json!("codex.response.metadata") + } else { + Value::Null + } + ); + assert!(failure["data"]["connectionAgeMs"].is_u64()); + assert!(failure["data"]["connectionIdleMs"].is_u64()); + assert!(!snapshot.to_string().contains("private-response-body")); + assert!(!snapshot.to_string().contains("private-request-body")); + } +} diff --git a/backend/crates/providers/openai/tests/transport/websocket/mod.rs b/backend/crates/providers/openai/tests/transport/websocket/mod.rs index 88be4357a..ea738bf2b 100644 --- a/backend/crates/providers/openai/tests/transport/websocket/mod.rs +++ b/backend/crates/providers/openai/tests/transport/websocket/mod.rs @@ -4,6 +4,7 @@ use provider_openai::transport::{ websocket::write_websocket_audit_artifact_for_dir, }; +mod diagnostics; mod metadata; fn rate_limit_event(used_percent: u64) -> String { diff --git a/backend/crates/providers/xai/src/provider/failure.rs b/backend/crates/providers/xai/src/provider/failure.rs index ba01de4ef..7284a1322 100644 --- a/backend/crates/providers/xai/src/provider/failure.rs +++ b/backend/crates/providers/xai/src/provider/failure.rs @@ -1,6 +1,7 @@ //! xAI 上游失败分类、账号反馈与恢复决策。 use super::*; +use gateway_core::error::ProviderDiagnostic; pub(super) fn is_invalid_encrypted_content_failure(error: &GrokInferenceTransportError) -> bool { error.kind() == GrokInferenceTransportErrorKind::InvalidRequest @@ -509,6 +510,17 @@ pub(super) fn map_transport_error_with_state( if let Some(detail) = error.client_visible_upstream_error().cloned() { mapped = mapped.with_client_visible_upstream_error(detail); } + let diagnostic = error.diagnostic().cloned().unwrap_or_else(|| { + let status = error + .status() + .map_or_else(|| "none".to_owned(), |value| value.to_string()); + ProviderDiagnostic::new(format!( + "xAI upstream failure: kind={}, status={status}", + kind.as_str() + )) + .with_classification("upstream", "upstream_failure") + }); + mapped = mapped.with_diagnostic(diagnostic); if error.requires_credential_recovery() { mapped = mapped.with_credential_recovery().with_replay_safe(); } diff --git a/backend/crates/providers/xai/src/transport/inference.rs b/backend/crates/providers/xai/src/transport/inference.rs index fb7f83a1d..bc55fe80e 100644 --- a/backend/crates/providers/xai/src/transport/inference.rs +++ b/backend/crates/providers/xai/src/transport/inference.rs @@ -4,7 +4,7 @@ use std::pin::Pin; use std::time::Duration; use futures::Stream; -use gateway_core::error::{ClientVisibleUpstreamError, OpaqueUpstreamValue}; +use gateway_core::error::{ClientVisibleUpstreamError, OpaqueUpstreamValue, ProviderDiagnostic}; use gateway_core::event::UpstreamHttpVersion; use gateway_core::upstream::UpstreamSendState; use url::Url; @@ -318,6 +318,7 @@ pub enum GrokInferenceTransportErrorKind { #[derive(Clone, PartialEq, Eq)] pub struct GrokInferenceTransportError { kind: GrokInferenceTransportErrorKind, + diagnostic: Option>, send_state: UpstreamSendState, status: Option, retry_after: Option, @@ -336,6 +337,7 @@ impl GrokInferenceTransportError { pub const fn new(kind: GrokInferenceTransportErrorKind, send_state: UpstreamSendState) -> Self { Self { kind, + diagnostic: None, send_state, status: None, retry_after: None, @@ -353,6 +355,18 @@ impl GrokInferenceTransportError { } } + /// 保存 transport 提取的安全诊断，避免 Provider 映射时丢失原因。 + #[must_use] + pub fn with_diagnostic(mut self, diagnostic: ProviderDiagnostic) -> Self { + self.diagnostic = Some(Box::new(diagnostic)); + self + } + + #[must_use] + pub fn diagnostic(&self) -> Option<&ProviderDiagnostic> { + self.diagnostic.as_deref() + } + /// 附着合法的 HTTP 状态码。 #[must_use] pub fn with_status(mut self, status: u16) -> Self { @@ -482,6 +496,7 @@ impl fmt::Debug for GrokInferenceTransportError { formatter .debug_struct("GrokInferenceTransportError") .field("kind", &self.kind) + .field("diagnostic", &self.diagnostic) .field("send_state", &self.send_state) .field("status", &self.status) .field("retry_after", &self.retry_after) diff --git a/backend/crates/providers/xai/src/transport/network.rs b/backend/crates/providers/xai/src/transport/network.rs index 1b37bf05b..64648d8a9 100644 --- a/backend/crates/providers/xai/src/transport/network.rs +++ b/backend/crates/providers/xai/src/transport/network.rs @@ -30,7 +30,7 @@ use crate::{ HttpMethod, MAX_GROK_BILLING_BYTES, MAX_GROK_MODEL_CATALOG_BYTES, OAuthHttpRequest, OAuthHttpResponse, OAuthHttpTransport, TransportFailure, TransportFailureKind, TransportFuture, }; -use gateway_core::error::{ClientVisibleUpstreamError, OpaqueUpstreamValue}; +use gateway_core::error::{ClientVisibleUpstreamError, OpaqueUpstreamValue, ProviderDiagnostic}; use gateway_core::event::UpstreamHttpVersion; use gateway_core::upstream::UpstreamSendState; @@ -954,6 +954,27 @@ fn classify_oauth_reqwest_error(error: reqwest::Error) -> TransportFailure { TransportFailure::new(kind) } +fn inference_http_diagnostic(error: &reqwest::Error) -> ProviderDiagnostic { + let (stage, code) = if error.is_builder() { + ("prepare", "http_request_build_failed") + } else if error.is_connect() && error.is_timeout() { + ("connect", "http_connect_timeout") + } else if error.is_connect() { + ("connect", "http_connect_failed") + } else if error.is_timeout() { + ("exchange", "http_timeout") + } else if error.is_decode() { + ("decode", "http_decode_failed") + } else if error.is_body() { + ("receive", "http_body_failed") + } else { + ("exchange", "http_transport_failed") + }; + ProviderDiagnostic::new(format!("xAI HTTP failure: stage={stage}, cause={code}")) + .with_classification(stage, code) + .with_io_cause(error) +} + fn classify_inference_reqwest_error(error: reqwest::Error) -> GrokInferenceTransportError { let (kind, send_state) = if error.is_builder() { ( @@ -977,6 +998,7 @@ fn classify_inference_reqwest_error(error: reqwest::Error) -> GrokInferenceTrans ) }; GrokInferenceTransportError::new(kind, send_state) + .with_diagnostic(inference_http_diagnostic(&error)) } fn classify_model_catalog_reqwest_error(error: reqwest::Error) -> GrokModelCatalogTransportError { @@ -1010,6 +1032,7 @@ fn classify_inference_stream_error(error: &reqwest::Error) -> GrokInferenceTrans }, UpstreamSendState::Sent, ) + .with_diagnostic(inference_http_diagnostic(error)) } async fn classify_inference_status( diff --git a/backend/crates/providers/xai/tests/provider/contract.rs b/backend/crates/providers/xai/tests/provider/contract.rs index da29ff124..236a28ab7 100644 --- a/backend/crates/providers/xai/tests/provider/contract.rs +++ b/backend/crates/providers/xai/tests/provider/contract.rs @@ -1704,6 +1704,35 @@ async fn inference_request_uses_oauth_headers_and_no_api_key() { drop(provider); } +#[tokio::test] +async fn transport_diagnostic_survives_provider_mapping_without_enabling_replay() { + let transport = StubInferenceTransport::stream_error( + GrokInferenceTransportError::new( + GrokInferenceTransportErrorKind::Transport, + UpstreamSendState::Sent, + ) + .with_diagnostic( + gateway_core::error::ProviderDiagnostic::new("xAI HTTP body read failed") + .with_classification("receive", "http_body_failed"), + ), + ); + let provider = provider(StubSelector::success(), transport).await; + let mut stream = provider + .execute( + provider_request("xai"), + context(CancellationToken::new(), None), + ) + .await + .expect("stream"); + let error = next_provider_error(&mut stream).await; + let diagnostic = error.diagnostic().expect("preserved transport diagnosis"); + assert_eq!(diagnostic.code(), Some("http_body_failed")); + assert_eq!(diagnostic.stage(), Some("receive")); + assert_eq!(diagnostic.as_str(), "xAI HTTP body read failed"); + assert_eq!(error.send_state(), UpstreamSendState::Sent); + assert!(!error.replay_is_safe()); +} + #[tokio::test] async fn unauthorized_transport_feedback_is_bound_to_selected_account() { let selector = StubSelector::success(); diff --git a/backend/crates/providers/xai/tests/transport/network.rs b/backend/crates/providers/xai/tests/transport/network.rs index 0b9787fdd..8609eddcf 100644 --- a/backend/crates/providers/xai/tests/transport/network.rs +++ b/backend/crates/providers/xai/tests/transport/network.rs @@ -252,6 +252,22 @@ async fn inference_transport_should_build_distinct_cold_account_clients_concurre assert!(concurrency.max_active.load(Ordering::SeqCst) >= 2); } +#[tokio::test] +async fn inference_connect_failure_keeps_safe_diagnosis_without_endpoint_or_payload() { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let origin = Url::parse(&format!("http://{}", listener.local_addr().unwrap())).unwrap(); + drop(listener); + let error = inference_transport(&origin) + .execute(inference_request(&origin)) + .await + .expect_err("closed listener"); + let diagnostic = error.diagnostic().expect("transport cause"); + assert_eq!(diagnostic.stage(), Some("connect")); + assert_eq!(diagnostic.code(), Some("connection_refused")); + assert!(!diagnostic.as_str().contains("127.0.0.1")); + assert_eq!(error.send_state(), UpstreamSendState::NotSent); +} + #[tokio::test] async fn inference_transport_should_classify_http_failures_without_retaining_bodies() { let cases = [ diff --git a/deploy/README.md b/deploy/README.md index c0ff143c6..aeb05eb2c 100644 --- a/deploy/README.md +++ b/deploy/README.md @@ -242,10 +242,40 @@ OpenAI 主动额度重置卡及其消费结果由上游持有，不写入 Postgr 数据库可用管理端的 S3/R2 逻辑备份，或停库后备份 `.runtime/postgres`。 不要在 PostgreSQL 写入期间直接复制数据目录作为一致性备份。要保留 OAuth 的 AT/RT 恢复记录，需保持 -`host.logging.file.enabled: true` 并备份 `.runtime/logs`；恢复记录位于独立文件集，仍按普通日志的 -`retention_days` 和 `max_files` 分别约束。若希望保留短期 Redis 状态和会话锚点，也同时备份 `.runtime/redis` 与 +`host.logging.oauth_recovery: true` 并备份 `.runtime/logs`；恢复记录位于独立文件集，仍按普通日志的 +`retention_days` 按完整日期保留。若希望保留短期 Redis 状态和会话锚点，也同时备份 `.runtime/redis` 与 `.runtime/data`。 +文件日志以时间完整性为清理依据，**没有文件数量淘汰上限**： + +| 文件集 | 保留配置 | 默认完整窗口 | +| --- | --- | --- | +| 普通日志、OAuth 恢复日志 | `host.logging.file.retention_days` | 至少 7×24 小时 | +| 全量请求/响应报文 | `host.logging.request_dump_retention_days` | 至少 24 小时（默认 1） | + +文件名统一为 `codex-proxy-rs-<类别>.YYYY-MM-DD[.N].log[.gz]`，类别分别为 +`application`、`oauth-recovery`、`request-dump`。专用 tracing target 为 `oauth_recovery` 和 +`request_dump`；普通日志保留各 Rust 模块的 target，便于按模块过滤。 +程序只管理上述规范名称的日志，旧命名文件由运维手动清理。 +普通日志未配置 `retention_days` 时默认使用 7 天，显式配置优先。 + +按 UTC 日期整组保留：例如 9 月 8 日配置 1 天，会保留 9 月 7 日全天及 9 月 8 日的所有分片， +到 9 月 9 日才允许清理 9 月 7 日。这会略多保留，保证跨午夜及高流量时不留下半天日志。 +配置 7 天同理，保留前 7 个完整 UTC 日期及当天；若旧日期分片近期又被写入，整组延后删除。 +`max_file_size_mb` 仅决定分片大小（默认 20 MiB），不决定保存时长，单条大记录不会被截断。 +关闭的分片压缩为 `.log.gz`；成功压缩、同步并发布归档后才删除原文件，保留原修改时间。 +清理发生在启动和轮转时；空闲期间过期文件可能暂时多保留。检索时须同时读取 `.log` 与 `.log.gz`。 + +升级旧配置时删除 `host.logging.file.max_files`；该字段已移除，旧配置会校验失败而不会继续按数量删日志。 +报文开关仍为 `host.logging.request_dump`，默认关闭；开启时原始报文按块完整写入独立文件， +数据库请求诊断仍是有界摘要，不能用摘要事件数代替全量报文完整性。 + +文件写入使用有界队列背压，正常退出会排空队列并同步文件。`file_logging` 健康探针在写入/同步失败后 +报告 `Unhealthy`，本次进程内恢复写入也不会清除已有缺口；压缩或清理失败报告 `Degraded`。 +这不是断电、强杀或磁盘故障下的零丢失承诺；已删除的历史文件也不能靠升级恢复。 +容量不足时不会提前删除保留窗口内日志，必须根据完整日期的压缩后实际用量规划空间，并监控磁盘余量和 +健康探针。Docker stdout 的独立轮转不承担应用文件日志的完整保留承诺。 + 完整运行时、Provider、revision 与恢复边界见 [架构文档](../docs/architecture.md)。 ## 密码语义 @@ -340,10 +370,11 @@ Release 必须提供当前 OS/架构的 `codex-proxy-rs___.ta 用户可读写）。部署卷至少预留一个最大数据库归档的空间。 - OAuth 恢复记录通道使用 `oauth_recovery` 结构化日志事件；当前 OpenAI Provider 每次成功取得 AT（以及存在时的 RT）后，都会在账号资料补全、过期时间计算和数据库写入前写入独立的 - `codex-proxy-rs-oath.YYYY-MM-DD[.N].log` 文件集。事件含原始 AT/RT，并以 `provider` - 字段标记来源，且与普通 `.runtime/logs` 完全相同地按日、按大小分割，并分别按相同 `retention_days` 与 `max_files` - 清理；文件日志开启时不会被普通日志级别筛掉。它遵循普通日志现有的非阻塞写入机制，不会阻断 - 导入、授权或刷新。 + `codex-proxy-rs-oauth-recovery.YYYY-MM-DD[.N].log` 文件集。事件含原始 AT/RT，并以 `provider` + 字段标记来源，与普通日志一样按日、按大小分割，按 `retention_days` 保留完整日期。 + `host.logging.oauth_recovery` 默认关闭，显式开启后才写独立文件，与普通文件日志开关分别控制。 + 开启时不会被普通日志级别筛掉；关闭时不会写入普通文件或 stdout，即使 `RUST_LOG` 提高此 target 的级别。 + 有界队列满时会等待写入，避免拥堵时静默丢记录。 `.runtime/logs` 因此属于敏感数据，必须按现有普通日志的访问控制和加密备份策略处理。 - S3/R2 存储、Cron 计划、保留策略与备份记录都保存在 PostgreSQL（`backup_settings` / `backup_records`），备份记录行在删除成功后硬删除，操作历史进入 `admin_audit_events`。 diff --git a/deploy/config.example.yaml b/deploy/config.example.yaml index 904599851..cf6c88657 100644 --- a/deploy/config.example.yaml +++ b/deploy/config.example.yaml @@ -20,15 +20,17 @@ host: logging: level: 'info' stdout: true + # 将原始 AT/RT 写入独立 OAuth 恢复文件，沿用 file 的目录、大小与保留期。 + oauth_recovery: false # 将完整请求与响应头、正文写入独立文件；包含凭据，仅排障时开启。 request_dump: false + request_dump_retention_days: 1 file: enabled: true # 相对路径以 deploy/config.yaml 所在目录为基准。 directory: '../.runtime/logs' - retention_days: 14 + retention_days: 7 max_file_size_mb: 20 - max_files: 20 drain_timeout_seconds: 30 worker_shutdown_timeout_seconds: 30 diff --git a/docs/architecture.md b/docs/architecture.md index cb862f0ef..9c7b9551a 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -318,9 +318,19 @@ HTTP Client 构造失败也不会阻断网关启动。外部解析在已认证 安全边界： - Provider credential 以 Provider schema 的明文 JSON 保存在 PostgreSQL；数据库和备份必须按敏感数据保护。 -- OAuth 恢复日志包含原始 AT/RT，`.runtime/logs` 同样属于敏感数据。 +- `host.logging.oauth_recovery` 默认关闭，开启后将 OAuth 原始 AT/RT 写入独立文件； + 与普通文件日志开关分别控制，不输出到普通日志或 stdout。`.runtime/logs` 同样属于敏感数据。 - `host.logging.request_dump` 默认关闭；开启后独立请求转储包含原始请求头和正文， 可能包含密钥及用户内容，只能在明确的排障范围内使用，不能作为普通日志公开。 + 普通及 OAuth 文件按 `file.retention_days`（默认 7）保留；报文按独立的 + `request_dump_retention_days`（默认 1）保留。保留当天及前 N 个完整 UTC 日期的所有分片， + 仅整组清理更早日期；近期被写入的旧日期组延后清理。不存在文件数量淘汰。 + `file.max_file_size_mb` 只控制轮转；已关闭分片压缩后才删除原文件，启动和轮转时清理过期日期。 + 文件队列满时背压，正常关闭时排空并同步。`file_logging` 健康探针持续报告本次进程的写入缺口， + 不将压缩/清理失败伪装成保留完成。容量规划须覆盖完整时间窗口，不能靠提前删窗口内分片解决磁盘不足。 +- Provider 将安全诊断的阶段、原因码与消息独立于错误大类和发送状态传入 Core； + `attempt.failed` 保存这些字段，最终错误记录优先持久化诊断消息。重试包装不能覆盖底层诊断， + 也不能因为错误更详细而改变已有重试安全边界。 - 真实 secret 不进入普通日志、Debug、fixture 或 audit details；明文只能通过账号导出、Key reveal、 备份设置等明确的敏感 Admin 合同返回。 - OAuth pending flow 使用有期限、带 owner 的一次性 claim；事务成功后才消费，失败释放 claim。 diff --git a/frontend/src/components/base/BaseButton.vue b/frontend/src/components/base/BaseButton.vue index 72c39e491..0e2bd669f 100644 --- a/frontend/src/components/base/BaseButton.vue +++ b/frontend/src/components/base/BaseButton.vue @@ -2,7 +2,7 @@ import { LoaderCircle } from '@lucide/vue' import { computed } from 'vue' -type ButtonVariant = 'primary' | 'secondary' | 'ghost' | 'destructive' +type ButtonVariant = 'primary' | 'secondary' | 'soft' | 'ghost' | 'destructive' type ButtonSize = 'sm' | 'md' | 'lg' const props = withDefaults( @@ -33,6 +33,8 @@ const variantClasses: Record = { 'bg-(--cp-button-primary-bg) text-(--cp-button-primary-color) shadow-cp-tertiary hover:bg-(--cp-button-primary-hover-bg) active:bg-(--cp-button-primary-active-bg)', secondary: 'bg-cp-fill-tertiary text-cp-text shadow-cp-tertiary hover:bg-cp-bg-text-active active:bg-cp-bg-text-active', + soft: + 'bg-cp-primary-container text-cp-primary-on-container shadow-none hover:bg-cp-primary-container-hover active:bg-cp-primary-container-hover', ghost: 'bg-transparent text-cp-text-secondary shadow-none hover:bg-cp-fill-quaternary hover:text-cp-text active:bg-cp-fill-tertiary', destructive: 'bg-cp-error-container text-cp-error-on-container shadow-none hover:bg-cp-error-container-hover active:bg-cp-error-container-active', diff --git a/frontend/src/views/usage/components/OpsErrorDetailModal.vue b/frontend/src/views/usage/components/OpsErrorDetailModal.vue index ef6f1641f..569e27b74 100644 --- a/frontend/src/views/usage/components/OpsErrorDetailModal.vue +++ b/frontend/src/views/usage/components/OpsErrorDetailModal.vue @@ -130,7 +130,7 @@ function visibleFields(items: DetailField[]) { size="xl" >