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" >