From a285066737c28399b83a0e6b0884c777059fb701 Mon Sep 17 00:00:00 2001 From: Dmitriy Kovalenko Date: Thu, 24 Sep 2026 16:49:50 +0000 Subject: [PATCH] Preserve diagnostic logs when SQLite logging fails (#47886) ## Why SQLite write failures can leave persisted logs incomplete, while log snapshots taken before collection miss errors raised during that collection. ## What changed - Report SQLite flush failures directly to the in-memory log buffer without re-entering tracing, using an error category and batch size instead of raw error details. - Track whether the log writer has lost a batch and make app-server diagnostic uploads use the memory buffer when SQLite logs are incomplete. - Refresh log bytes after flushing and querying SQLite so the same submission includes collection failures, while preserving captured metadata and thread identity. ## Testing Add regression tests for failure reporting without recursion, subsequent write recovery, metadata preservation, and healthy and failing SQLite collection. Add an upload integration test that captures attachments locally and checks failure diagnostics, memory-buffer fallback, and omission of raw SQLite error details. GitOrigin-RevId: 51b0c4b69c977d9f4e74f2f245a9fd90fdb13761 --- codex-rs/Cargo.lock | 3 + codex-rs/app-server/Cargo.toml | 2 + codex-rs/app-server/src/lib.rs | 4 +- .../feedback_log_snapshot_tests.rs | 82 ++++++++ .../request_processors/feedback_processor.rs | 95 ++++++---- .../app-server/tests/suite/v2/feedback.rs | 177 ++++++++++++++++++ codex-rs/core/tests/suite/sqlite_state.rs | 3 +- codex-rs/feedback/Cargo.toml | 1 + codex-rs/feedback/src/lib.rs | 22 +++ codex-rs/feedback/src/metadata_tests.rs | 26 +++ codex-rs/state/src/lib.rs | 1 + codex-rs/state/src/log_db.rs | 110 +++++++++-- codex-rs/state/src/log_db_failure_tests.rs | 86 +++++++++ codex-rs/state/src/log_db_filter_tests.rs | 5 +- codex-rs/state/src/telemetry.rs | 2 +- codex-rs/tui/src/startup_orchestration.rs | 4 +- 16 files changed, 563 insertions(+), 60 deletions(-) create mode 100644 codex-rs/app-server/src/request_processors/feedback_log_snapshot_tests.rs create mode 100644 codex-rs/state/src/log_db_failure_tests.rs diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index 871d3a2cc4..dc457a186a 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -2126,6 +2126,7 @@ dependencies = [ "opentelemetry", "opentelemetry_sdk", "pretty_assertions", + "rcgen", "rmcp", "serde", "serde_json", @@ -2140,6 +2141,7 @@ dependencies = [ "tikv-jemallocator", "time", "tokio", + "tokio-rustls", "tokio-tungstenite", "tokio-util", "toml 0.9.11+spec-1.1.0", @@ -3358,6 +3360,7 @@ dependencies = [ "codex-login", "codex-protocol", "codex-rollout", + "codex-state", "flate2", "http 1.4.0", "httpdate", diff --git a/codex-rs/app-server/Cargo.toml b/codex-rs/app-server/Cargo.toml index 73ae57d673..ad0a5fbab3 100644 --- a/codex-rs/app-server/Cargo.toml +++ b/codex-rs/app-server/Cargo.toml @@ -142,6 +142,7 @@ http = { workspace = true } opentelemetry = { workspace = true } opentelemetry_sdk = { workspace = true } pretty_assertions = { workspace = true } +rcgen = { workspace = true } rmcp = { workspace = true, default-features = false, features = [ "elicitation", "server", @@ -152,6 +153,7 @@ shlex = { workspace = true } sqlx = { workspace = true } tar = { workspace = true } test-case = "3.3.1" +tokio-rustls = { workspace = true } tokio-tungstenite = { workspace = true } tracing-opentelemetry = { workspace = true } wiremock = { workspace = true } diff --git a/codex-rs/app-server/src/lib.rs b/codex-rs/app-server/src/lib.rs index fc7868bbac..4b7e526085 100644 --- a/codex-rs/app-server/src/lib.rs +++ b/codex-rs/app-server/src/lib.rs @@ -712,7 +712,9 @@ pub async fn run_main_with_transport_options( let feedback_layer = feedback.logger_layer(); let feedback_metadata_layer = feedback.metadata_layer(); - let log_db = state_db.clone().map(log_db::start); + let log_db = state_db + .clone() + .map(|state_db| log_db::start(state_db, std::sync::Arc::new(feedback.clone()))); let log_db_layer = log_db .clone() .map(|layer| layer.with_filter(log_db::default_filter())); diff --git a/codex-rs/app-server/src/request_processors/feedback_log_snapshot_tests.rs b/codex-rs/app-server/src/request_processors/feedback_log_snapshot_tests.rs new file mode 100644 index 0000000000..1a447050c3 --- /dev/null +++ b/codex-rs/app-server/src/request_processors/feedback_log_snapshot_tests.rs @@ -0,0 +1,82 @@ +//! Exercises the real SQLite writer, feedback ring buffer, and snapshot ordering. +use super::*; +use codex_state::SqliteConfig; +use codex_state::StateRuntime; +use codex_utils_absolute_path::test_support::PathExt; +use pretty_assertions::assert_eq; +use tracing_subscriber::prelude::*; + +#[derive(PartialEq)] +enum Scenario { + Healthy, + WriteFailure, + Corruption, +} + +#[tokio::test] +async fn feedback_includes_flush_and_query_failures_in_the_same_submission() -> anyhow::Result<()> { + for scenario in [ + Scenario::Healthy, + Scenario::WriteFailure, + Scenario::Corruption, + ] { + let home = tempfile::tempdir()?; + let sqlite = SqliteConfig::new_for_testing(home.path().abs()); + let runtime = StateRuntime::init(sqlite.clone(), "test-provider".to_string()).await?; + let pool = sqlite.open_read_write_pool(&sqlite.logs_db_path()).await?; + let feedback = CodexFeedback::new(); + let writer = + codex_state::log_db::start(runtime.clone(), std::sync::Arc::new(feedback.clone())); + let _guard = tracing_subscriber::registry() + .with(feedback.logger_layer()) + .with(writer.clone()) + .set_default(); + let thread_id = ThreadId::new(); + tracing::info_span!("thread", thread_id = %thread_id) + .in_scope(|| tracing::info!("persisted log")); + writer.flush().await; + match scenario { + Scenario::WriteFailure => { + sqlx::query("CREATE TRIGGER fail_log_insert BEFORE INSERT ON logs BEGIN SELECT RAISE(ABORT, 'synthetic-secret'); END").execute(&pool).await?; + } + Scenario::Corruption => { + sqlx::raw_sql("PRAGMA writable_schema = ON; UPDATE sqlite_schema SET rootpage = 2147483647 WHERE name = 'logs'; PRAGMA schema_version = 1000000;").execute(&pool).await?; + } + Scenario::Healthy => {} + } + let snapshot = feedback.snapshot(Some(thread_id)); + tracing::info_span!("thread", thread_id = %thread_id) + .in_scope(|| tracing::info!("latest log")); + let (snapshot, sqlite_logs) = collect_feedback_logs( + &feedback, + Some(&writer), + Some(&runtime), + snapshot, + &[thread_id], + ) + .await; + assert_eq!(sqlite_logs.is_some(), scenario == Scenario::Healthy); + let logs = String::from_utf8(snapshot.log_attachment(sqlite_logs).buffer)?; + assert!(logs.contains("persisted log")); + assert!(logs.contains("latest log")); + let expected_error = match scenario { + Scenario::Healthy => None, + Scenario::WriteFailure => Some("constraint"), + Scenario::Corruption => Some("corrupt"), + }; + assert_eq!( + logs.matches("failed to flush logs to SQLite").count(), + usize::from(expected_error.is_some()) + ); + if let Some(error) = expected_error { + assert!(logs.contains(&format!("failed to flush logs to SQLite error={error:?}"))); + } + assert_eq!( + logs.contains("failed to query feedback logs from sqlite"), + scenario == Scenario::Corruption + ); + pool.close().await; + runtime.close().await; + } + Ok(()) +} diff --git a/codex-rs/app-server/src/request_processors/feedback_processor.rs b/codex-rs/app-server/src/request_processors/feedback_processor.rs index 6c9cef24ed..9bb2b461ec 100644 --- a/codex-rs/app-server/src/request_processors/feedback_processor.rs +++ b/codex-rs/app-server/src/request_processors/feedback_processor.rs @@ -6,6 +6,7 @@ use codex_connectors::ConnectorDirectoryCacheKey; use codex_connectors::connector_runtime_cache_path; use codex_feedback::CODEX_APP_DIRECTORY_CACHE_ATTACHMENT_FILENAME; use codex_feedback::CODEX_APPS_TOOLS_CACHE_ATTACHMENT_FILENAME; +use codex_feedback::FeedbackSnapshot; #[cfg(target_os = "windows")] use codex_feedback::WINDOWS_SANDBOX_LOG_ATTACHMENT_FILENAME; use codex_feedback::guardian_review_failures; @@ -124,13 +125,9 @@ impl FeedbackRequestProcessor { tracing::info!(target: "feedback_tags", account_id); } let snapshot = self.feedback.snapshot(conversation_id); - let thread_id = snapshot.thread_id.clone(); let mut extra_attachments = Vec::new(); let mut feedback_index = None; - let (sqlite_feedback_logs, state_db_ctx) = if include_logs { - if let Some(log_db) = self.log_db.as_ref() { - log_db.flush().await; - } + let (snapshot, sqlite_feedback_logs, state_db_ctx) = if include_logs { let state_db_ctx = self.state_db.clone(); let feedback_thread_ids = match conversation_id { Some(conversation_id) => match self @@ -161,38 +158,19 @@ impl FeedbackRequestProcessor { feedback_index = Some(index); } extra_attachments.extend(failures.attachment); - let sqlite_feedback_logs = if let Some(state_db_ctx) = state_db_ctx.as_ref() - && !feedback_thread_ids.is_empty() - { - let thread_id_texts = feedback_thread_ids - .iter() - .map(ToString::to_string) - .collect::>(); - let thread_id_refs = thread_id_texts - .iter() - .map(String::as_str) - .collect::>(); - match state_db_ctx - .query_feedback_logs_for_threads(&thread_id_refs) - .await - { - Ok(logs) if logs.is_empty() => None, - Ok(logs) => Some(logs), - Err(err) => { - let thread_ids = thread_id_texts.join(", "); - warn!( - "failed to query feedback logs from sqlite for thread_ids=[{thread_ids}]: {err}" - ); - None - } - } - } else { - None - }; - (sqlite_feedback_logs, state_db_ctx) + let (snapshot, sqlite_feedback_logs) = collect_feedback_logs( + &self.feedback, + self.log_db.as_ref(), + state_db_ctx.as_ref(), + snapshot, + &feedback_thread_ids, + ) + .await; + (snapshot, sqlite_feedback_logs, state_db_ctx) } else { - (None, None) + (snapshot, None, None) }; + let thread_id = snapshot.thread_id.clone(); let mut attachment_paths = Vec::new(); let mut seen_attachment_paths = HashSet::new(); @@ -503,6 +481,49 @@ fn windows_sandbox_log_attachment(_codex_home: &Path) -> Option, + state_db: Option<&StateDbHandle>, + mut snapshot: FeedbackSnapshot, + thread_ids: &[ThreadId], +) -> (FeedbackSnapshot, Option>) { + if let Some(log_db) = log_db { + log_db.flush().await; + } + let mut sqlite_logs = None; + if let Some(state_db) = state_db + && !thread_ids.is_empty() + { + let thread_ids = thread_ids + .iter() + .map(ToString::to_string) + .collect::>(); + let thread_id_refs = thread_ids.iter().map(String::as_str).collect::>(); + match state_db + .query_feedback_logs_for_threads(&thread_id_refs) + .await + { + Ok(logs) if logs.is_empty() => {} + Ok(logs) => sqlite_logs = Some(logs), + Err(err) => { + let thread_ids = thread_ids.join(", "); + tracing::warn!( + "failed to query feedback logs from sqlite for thread_ids=[{thread_ids}]: {err}" + ); + } + } + } + if log_db.is_some_and(LogDbLayer::has_write_failure) { + sqlite_logs = None; + } + + // new logs might arrive to the buffer while collecting metadata e.g. if log db flush fails + snapshot.refresh_logs(feedback); + (snapshot, sqlite_logs) +} + #[cfg(test)] mod tests { use super::*; @@ -885,3 +906,7 @@ mod tests { ); } } + +#[cfg(test)] +#[path = "feedback_log_snapshot_tests.rs"] +mod feedback_log_snapshot_tests; diff --git a/codex-rs/app-server/tests/suite/v2/feedback.rs b/codex-rs/app-server/tests/suite/v2/feedback.rs index fd6a2ef936..c594d6237f 100644 --- a/codex-rs/app-server/tests/suite/v2/feedback.rs +++ b/codex-rs/app-server/tests/suite/v2/feedback.rs @@ -90,3 +90,180 @@ async fn feedback_upload_limits_concurrency_and_releases_failed_uploads() -> Res assert_eq!(error.error.code, -32603); Ok(()) } + +#[tokio::test] +async fn feedback_upload_includes_sqlite_flush_and_query_failures() -> Result<()> { + use std::io::Read; + use std::sync::Arc; + + use app_test_support::MockResponsesConfig; + use codex_app_server_protocol::ClientNotification; + use codex_app_server_protocol::ThreadStartParams; + use codex_state::SqliteConfig; + use codex_utils_absolute_path::test_support::PathExt; + use tokio::io::AsyncReadExt; + use tokio_rustls::TlsAcceptor; + use tokio_rustls::rustls; + + // Terminate HTTPS locally and capture real Sentry envelopes. Never forward uploads. + let home = tempfile::tempdir()?; + let cert = rcgen::generate_simple_self_signed(vec!["o33249.ingest.us.sentry.io".to_string()])?; + let cert_path = home.path().join("feedback-ca.pem"); + std::fs::write(&cert_path, cert.cert.pem())?; + let tls = rustls::ServerConfig::builder_with_provider(Arc::new( + rustls::crypto::aws_lc_rs::default_provider(), + )) + .with_safe_default_protocol_versions()? + .with_no_client_auth() + .with_single_cert( + vec![cert.cert.der().clone()], + rustls::pki_types::PrivatePkcs8KeyDer::from(cert.signing_key.serialize_der()).into(), + )?; + let acceptor = TlsAcceptor::from(Arc::new(tls)); + let listener = TcpListener::bind("127.0.0.1:0").await?; + let proxy_uri = format!("http://{}", listener.local_addr()?); + let (uploads_tx, mut uploads_rx) = tokio::sync::mpsc::unbounded_channel(); + let _proxy = tokio_util::task::AbortOnDropHandle::new(tokio::spawn(async move { + while let Ok((stream, _)) = listener.accept().await { + let mut stream = BufReader::new(stream); + loop { + let mut line = String::new(); + anyhow::ensure!(stream.read_line(&mut line).await? > 0, "missing CONNECT"); + if line == "\r\n" { + break; + } + } + stream + .write_all(b"HTTP/1.1 200 Connection Established\r\n\r\n") + .await?; + let mut stream = BufReader::new(acceptor.accept(stream.into_inner()).await?); + let mut length = None; + loop { + let mut line = String::new(); + anyhow::ensure!( + stream.read_line(&mut line).await? > 0, + "missing HTTP headers" + ); + if line == "\r\n" { + break; + } + if let Some((name, value)) = line.split_once(':') + && name.eq_ignore_ascii_case("content-length") + { + length = Some(value.trim().parse::()?); + } + } + let length = length.ok_or_else(|| anyhow::anyhow!("missing content length"))?; + anyhow::ensure!(length < 16 * 1024 * 1024, "unexpected envelope size"); + let mut body = vec![0; length]; + stream.read_exact(&mut body).await?; + if body.starts_with(&[0x1f, 0x8b]) { + let mut decoded = Vec::new(); + flate2::read::GzDecoder::new(body.as_slice()).read_to_end(&mut decoded)?; + body = decoded; + } + uploads_tx.send(body)?; + stream + .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: close\r\n\r\n{}") + .await?; + stream.flush().await?; + } + Ok::<(), anyhow::Error>(()) + })); + + let models = wiremock::MockServer::start().await; + MockResponsesConfig::new(&models.uri()).write(home.path())?; + let mut app_server = TestAppServer::builder() + .with_codex_home(home.path()) + .with_env_overrides(&[ + ("HTTPS_PROXY", Some(proxy_uri.as_str())), + ("https_proxy", Some(proxy_uri.as_str())), + ("NO_PROXY", Some("127.0.0.1,localhost")), + ("no_proxy", Some("127.0.0.1,localhost")), + ("SSL_CERT_FILE", cert_path.to_str()), + ]) + .build_initialized() + .await?; + let request = app_server + .send_thread_start_request_with_auto_env(ThreadStartParams { + ephemeral: Some(true), + ..Default::default() + }) + .await?; + let response = app_server + .read_stream_until_response_message(RequestId::Integer(request)) + .await?; + let thread_id = response.result["thread"]["id"].as_str().unwrap(); + let sqlite = SqliteConfig::new_for_testing(home.path().abs()); + let pool = sqlite.open_read_write_pool(&sqlite.logs_db_path()).await?; + // This old history exists only in SQLite, as if it had already left the memory ring. + sqlx::query("INSERT INTO logs (ts, ts_nanos, level, target, feedback_log_body, thread_id) VALUES (1, 0, 'INFO', 'test', 'sqlite-only-history', ?)") + .bind(thread_id).execute(&pool).await?; + sqlx::query("CREATE TRIGGER fail_log_insert BEFORE INSERT ON logs BEGIN SELECT RAISE(ABORT, 'synthetic-secret'); END") + .execute(&pool).await?; + + #[derive(Debug, PartialEq)] + enum Phase { + WriteFailure, + Corrupt, + } + for phase in [Phase::WriteFailure, Phase::Corrupt] { + match phase { + Phase::Corrupt => { + sqlx::raw_sql("PRAGMA writable_schema = ON; UPDATE sqlite_schema SET rootpage = 2147483647 WHERE name = 'logs'; PRAGMA schema_version = 1000000;").execute(&pool).await?; + } + Phase::WriteFailure => {} + } + // Notifications emit an INFO log, ensuring feedback/upload has a batch to flush. + app_server + .send_notification(ClientNotification::Initialized) + .await?; + let request = app_server + .send_raw_request( + "feedback/upload", + Some(json!({ + "classification": "bug", "includeLogs": true, "threadId": thread_id + })), + ) + .await?; + timeout( + Duration::from_secs(/*secs*/ 30), + app_server.read_stream_until_response_message(RequestId::Integer(request)), + ) + .await??; + let mut attachments = std::collections::BTreeMap::new(); + while let Ok(envelope) = uploads_rx.try_recv() { + let mut bytes = envelope.as_slice(); + // Skip the envelope header, then decode each length-delimited item. + let header_end = bytes.iter().position(|byte| *byte == b'\n').unwrap(); + bytes = &bytes[header_end + 1..]; + while !bytes.is_empty() { + let header_end = bytes.iter().position(|byte| *byte == b'\n').unwrap(); + let header: serde_json::Value = serde_json::from_slice(&bytes[..header_end])?; + let length = header["length"].as_u64().unwrap() as usize; + bytes = &bytes[header_end + 1..]; + if let Some(filename) = header["filename"].as_str() { + attachments.insert( + filename.to_string(), + String::from_utf8_lossy(&bytes[..length]).into_owned(), + ); + } + bytes = &bytes[length..]; + if bytes.starts_with(b"\n") { + bytes = &bytes[1..]; + } + } + } + let logs = &attachments["codex-logs.log"]; + if phase == Phase::Corrupt { + assert!(logs.contains("failed to flush logs to SQLite error=\"corrupt\"")); + assert!(logs.contains("failed to query feedback logs from sqlite")); + } else { + assert!(logs.contains("failed to flush logs to SQLite error=\"constraint\"")); + assert!(!logs.contains("sqlite-only-history")); + assert!(!logs.contains("synthetic-secret")); + } + } + pool.close().await; + Ok(()) +} diff --git a/codex-rs/core/tests/suite/sqlite_state.rs b/codex-rs/core/tests/suite/sqlite_state.rs index 54432b9ca3..b7b3121eb9 100644 --- a/codex-rs/core/tests/suite/sqlite_state.rs +++ b/codex-rs/core/tests/suite/sqlite_state.rs @@ -987,7 +987,8 @@ async fn tool_call_logs_include_thread_id() -> Result<()> { test.submit_turn("run a shell command").await?; - let log_db_layer = codex_state::log_db::start(db.clone()); + let log_db_layer = + codex_state::log_db::start(db.clone(), Arc::new(codex_feedback::CodexFeedback::new())); let subscriber = tracing_subscriber::registry().with(log_db_layer.clone()); let dispatch = tracing::Dispatch::new(subscriber); tracing::dispatcher::with_default(&dispatch, || { diff --git a/codex-rs/feedback/Cargo.toml b/codex-rs/feedback/Cargo.toml index 4e66442884..2fab1dd906 100644 --- a/codex-rs/feedback/Cargo.toml +++ b/codex-rs/feedback/Cargo.toml @@ -14,6 +14,7 @@ codex-http-client = { workspace = true } codex-login = { workspace = true } codex-protocol = { workspace = true } codex-rollout = { workspace = true } +codex-state = { workspace = true } flate2 = { workspace = true } http = { workspace = true } httpdate = { workspace = true } diff --git a/codex-rs/feedback/src/lib.rs b/codex-rs/feedback/src/lib.rs index a8c48c1450..5f6519368a 100644 --- a/codex-rs/feedback/src/lib.rs +++ b/codex-rs/feedback/src/lib.rs @@ -21,6 +21,7 @@ use codex_http_client::RouteAwareClientPool; use codex_login::AuthEnvTelemetry; use codex_protocol::ThreadId; use codex_protocol::protocol::SessionSource; +use codex_state::LogWriteFailureReporter; use tracing::Event; use tracing::Level; use tracing::field::Visit; @@ -196,6 +197,17 @@ pub struct CodexFeedback { inner: Arc, } +impl LogWriteFailureReporter for CodexFeedback { + fn report_failure(&self, diagnostic: &str) { + // Bypass tracing so this diagnostic cannot return to the SQLite writer. + self.inner + .ring + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .push_bytes(diagnostic.as_bytes()); + } +} + impl Default for CodexFeedback { fn default() -> Self { Self::new() @@ -489,6 +501,16 @@ pub struct FeedbackUploadOptions<'a> { } impl FeedbackSnapshot { + /// Refreshes log bytes while preserving the captured metadata and thread identity. + pub fn refresh_logs(&mut self, feedback: &CodexFeedback) { + self.bytes = feedback + .inner + .ring + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .snapshot_bytes(); + } + fn feedback_event( &self, classification: &str, diff --git a/codex-rs/feedback/src/metadata_tests.rs b/codex-rs/feedback/src/metadata_tests.rs index 583f560698..83f7cddbac 100644 --- a/codex-rs/feedback/src/metadata_tests.rs +++ b/codex-rs/feedback/src/metadata_tests.rs @@ -8,6 +8,32 @@ use sentry::protocol::EnvelopeItem; use tracing_subscriber::layer::SubscriberExt; use tracing_subscriber::util::SubscriberInitExt; +#[test] +fn refreshing_logs_preserves_captured_metadata() { + let feedback = CodexFeedback::new(); + let _guard = tracing_subscriber::registry() + .with(feedback.metadata_layer()) + .set_default(); + let thread_id = ThreadId::new(); + tracing::info!(target: FEEDBACK_TAGS_TARGET, approval_policy = "on-request"); + feedback.report_failure("initial log\n"); + let mut snapshot = feedback.snapshot(Some(thread_id)); + + // Another conversation changes the process-wide tags while log collection runs. + tracing::info!(target: FEEDBACK_TAGS_TARGET, approval_policy = "never"); + feedback.report_failure("flush failed\n"); + snapshot.refresh_logs(&feedback); + + assert_eq!( + (snapshot.tags, snapshot.thread_id, snapshot.bytes), + ( + BTreeMap::from([("approval_policy".to_string(), "on-request".to_string())]), + thread_id.to_string(), + b"initial log\nflush failed\n".to_vec(), + ) + ); +} + #[test] fn dynamic_tags_survive_upload_alongside_static_fields() { let feedback = CodexFeedback::new(); diff --git a/codex-rs/state/src/lib.rs b/codex-rs/state/src/lib.rs index 6d88091b2f..fa5a5e2803 100644 --- a/codex-rs/state/src/lib.rs +++ b/codex-rs/state/src/lib.rs @@ -19,6 +19,7 @@ mod runtime; mod sqlite; mod telemetry; +pub use log_db::LogWriteFailureReporter; pub use model::CreatedProject; pub use model::LogEntry; pub use model::LogQuery; diff --git a/codex-rs/state/src/log_db.rs b/codex-rs/state/src/log_db.rs index 06387efc4c..4b2d8a6c90 100644 --- a/codex-rs/state/src/log_db.rs +++ b/codex-rs/state/src/log_db.rs @@ -4,8 +4,8 @@ //! formats each one into a `LogEntry`, and sends entries to a bounded background //! queue. The background task inserts into the dedicated `logs` SQLite database //! in batches to keep logging overhead low. -//! SQLx diagnostics are always excluded so database writes cannot generate more -//! database writes, even without an external subscriber filter. +//! SQLx events are excluded so database writes cannot generate more writes. +//! Flush failures go to the supplied failure reporter without re-entering tracing. //! //! ## Usage //! @@ -13,8 +13,8 @@ //! use codex_state::log_db; //! use tracing_subscriber::prelude::*; //! -//! # async fn example(state_db: std::sync::Arc) { -//! let layer = log_db::start(state_db); +//! # async fn example(state_db: std::sync::Arc, reporter: std::sync::Arc) { +//! let layer = log_db::start(state_db, reporter); //! let _ = tracing_subscriber::registry() //! .with(layer) //! .try_init(); @@ -22,7 +22,10 @@ //! ``` use std::future::Future; +use std::sync::Arc; use std::sync::OnceLock; +use std::sync::atomic::AtomicBool; +use std::sync::atomic::Ordering; use std::time::Duration; use std::time::Instant; use std::time::SystemTime; @@ -124,42 +127,71 @@ where fn flush(&self) -> impl Future + Send + '_; } +/// Receives a formatted, redacted diagnostic when a SQLite log batch is lost +/// this propagatest to telemetry and feedback memory buffer to avoid lost reports +pub trait LogWriteFailureReporter: Send + Sync { + fn report_failure(&self, diagnostic: &str); +} + pub struct LogDbLayer { sender: mpsc::Sender, + has_write_failure: Arc, process_uuid: String, } -pub fn start(state_db: std::sync::Arc) -> LogDbLayer { - LogDbLayer::start(state_db) +/// Starts the SQLite log writer with a shared, independent failure reporter. +pub fn start( + state_db: std::sync::Arc, + failure_reporter: Arc, +) -> LogDbLayer { + LogDbLayer::start(state_db, failure_reporter) } impl Clone for LogDbLayer { fn clone(&self) -> Self { Self { sender: self.sender.clone(), + has_write_failure: self.has_write_failure.clone(), process_uuid: self.process_uuid.clone(), } } } impl LogDbLayer { - pub fn start(state_db: std::sync::Arc) -> Self { - Self::start_with_config(state_db, LogSinkQueueConfig::default()) + pub fn start( + state_db: std::sync::Arc, + failure_reporter: Arc, + ) -> Self { + Self::start_with_config(state_db, failure_reporter, LogSinkQueueConfig::default()) } pub fn start_with_config( state_db: std::sync::Arc, + failure_reporter: Arc, config: LogSinkQueueConfig, ) -> Self { let config = config.normalized(); let (sender, receiver) = mpsc::channel(config.queue_capacity); - tokio::spawn(run_inserter(state_db, receiver, config)); + let has_write_failure = Arc::new(AtomicBool::new(false)); + tokio::spawn(run_inserter( + state_db, + receiver, + config, + failure_reporter, + has_write_failure.clone(), + )); Self { sender, + has_write_failure, process_uuid: current_process_log_uuid().to_string(), } } + /// Whether this writer has lost logs, making SQLite incomplete for feedback. + pub fn has_write_failure(&self) -> bool { + self.has_write_failure.load(Ordering::Relaxed) + } + pub async fn flush(&self) { let (tx, rx) = oneshot::channel(); if self.sender.send(LogDbCommand::Flush(tx)).await.is_ok() { @@ -447,6 +479,8 @@ async fn run_inserter( state_db: std::sync::Arc, mut receiver: mpsc::Receiver, config: LogSinkQueueConfig, + failure_reporter: Arc, + has_write_failure: Arc, ) { let mut buffer = Vec::with_capacity(config.batch_size); let mut ticker = tokio::time::interval(config.flush_interval); @@ -459,36 +493,56 @@ async fn run_inserter( Some(LogDbCommand::Entry(entry)) => { buffer.push(*entry); if buffer.len() >= config.batch_size { - flush(&state_db, &mut buffer).await; + flush(&state_db, &mut buffer, failure_reporter.as_ref(), &has_write_failure).await; } } Some(LogDbCommand::Flush(reply)) => { - flush(&state_db, &mut buffer).await; + flush(&state_db, &mut buffer, failure_reporter.as_ref(), &has_write_failure).await; let _ = reply.send(()); } None => { - flush(&state_db, &mut buffer).await; + flush(&state_db, &mut buffer, failure_reporter.as_ref(), &has_write_failure).await; break; } } } _ = ticker.tick() => { - flush(&state_db, &mut buffer).await; + flush(&state_db, &mut buffer, failure_reporter.as_ref(), &has_write_failure).await; } } } } -async fn flush(state_db: &StateRuntime, buffer: &mut Vec) { +async fn flush( + state_db: &StateRuntime, + buffer: &mut Vec, + failure_reporter: &dyn LogWriteFailureReporter, + has_write_failure: &AtomicBool, +) { if buffer.is_empty() { return; } let entries = buffer.split_off(0); let started = Instant::now(); let result = state_db.insert_logs(entries.as_slice()).await; + let duration = started.elapsed(); + // if log flushing failed it means that likely something is wrong with the logs database, + // so store the logs in the memory buffer to avoid them being lost during /feedback + // the entries are dropped because the in memory buffer has it's own space-constrainted rules for preserving logs + if let Err(error) = &result { + let error = crate::telemetry::classify_error(error); + let timestamp = + chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Millis, /*use_z*/ true); + let entry_count = entries.len(); + failure_reporter.report_failure(&format!( + "{timestamp} ERROR failed to flush logs to SQLite error={error:?} entries={entry_count}\n" + )); + has_write_failure.store(true, Ordering::Relaxed); + } + crate::telemetry::record_log_write( /*telemetry*/ None, - started.elapsed(), + duration, entries.as_slice(), &result, ); @@ -545,11 +599,16 @@ impl Visit for MessageVisitor { #[path = "log_db_filter_tests.rs"] mod filter_tests; +#[cfg(test)] +#[path = "log_db_failure_tests.rs"] +mod failure_tests; + #[cfg(test)] mod tests { use std::io; use std::sync::Arc; use std::sync::Mutex; + use std::sync::atomic::AtomicBool; use codex_utils_absolute_path::test_support::PathExt; use pretty_assertions::assert_eq; @@ -600,21 +659,28 @@ mod tests { } #[derive(Clone, Default)] - struct SharedWriter { + pub(super) struct SharedWriter { bytes: Arc>>, } impl SharedWriter { - fn snapshot(&self) -> String { + pub(super) fn snapshot(&self) -> String { String::from_utf8(self.bytes.lock().expect("writer mutex poisoned").clone()) .expect("valid utf-8") } } - struct SharedWriterGuard { + pub(super) struct SharedWriterGuard { bytes: Arc>>, } + impl LogWriteFailureReporter for SharedWriter { + fn report_failure(&self, diagnostic: &str) { + std::io::Write::write_all(&mut self.make_writer(), diagnostic.as_bytes()) + .expect("write diagnostic to test buffer"); + } + } + impl<'a> MakeWriter<'a> for SharedWriter { type Writer = SharedWriterGuard; @@ -649,7 +715,7 @@ mod tests { .await .expect("initialize runtime"); let writer = SharedWriter::default(); - let layer = start(runtime.clone()); + let layer = start(runtime.clone(), Arc::new(SharedWriter::default())); let subscriber = tracing_subscriber::registry() .with( @@ -709,7 +775,7 @@ mod tests { ) .await .expect("initialize runtime"); - let layer = start(runtime.clone()); + let layer = start(runtime.clone(), Arc::new(SharedWriter::default())); let guard = tracing_subscriber::registry() .with( @@ -745,6 +811,7 @@ mod tests { .expect("initialize runtime"); let layer = LogDbLayer::start_with_config( runtime.clone(), + Arc::new(SharedWriter::default()), LogSinkQueueConfig { queue_capacity: 8, batch_size: 2, @@ -797,6 +864,7 @@ mod tests { .expect("initialize runtime"); let layer = LogDbLayer::start_with_config( runtime.clone(), + Arc::new(SharedWriter::default()), LogSinkQueueConfig { queue_capacity: 8, batch_size: 128, @@ -827,6 +895,7 @@ mod tests { let (sender, mut receiver) = mpsc::channel(1); let layer = LogDbLayer { sender, + has_write_failure: Arc::new(AtomicBool::new(false)), process_uuid: "process-1".to_string(), }; @@ -847,6 +916,7 @@ mod tests { let (sender, mut receiver) = mpsc::channel(1); let layer = LogDbLayer { sender, + has_write_failure: Arc::new(AtomicBool::new(false)), process_uuid: "process-1".to_string(), }; diff --git a/codex-rs/state/src/log_db_failure_tests.rs b/codex-rs/state/src/log_db_failure_tests.rs new file mode 100644 index 0000000000..de14ab709e --- /dev/null +++ b/codex-rs/state/src/log_db_failure_tests.rs @@ -0,0 +1,86 @@ +//! Regression coverage for flush diagnostics, recovery, and sink recursion prevention. + +use codex_utils_absolute_path::test_support::PathExt; +use pretty_assertions::assert_eq; +use tracing_subscriber::layer::SubscriberExt; +use tracing_subscriber::util::SubscriberInitExt; + +use super::tests::SharedWriter; +use super::*; + +#[tokio::test] +async fn failed_batch_is_reported_once_and_later_writes_recover() { + let codex_home = + std::env::temp_dir().join(format!("codex-state-log-failure-{}", Uuid::new_v4())); + let _cleanup = scopeguard::guard(codex_home.clone(), |codex_home| { + let _ = std::fs::remove_dir_all(codex_home); + }); + let sqlite = crate::SqliteConfig::new_for_testing(codex_home.as_path().abs()); + let runtime = StateRuntime::init(sqlite.clone(), "test-provider".to_string()) + .await + .expect("initialize runtime"); + let pool = sqlite + .open_read_write_pool(&sqlite.logs_db_path()) + .await + .expect("open logs database"); + sqlx::query("CREATE TRIGGER fail_log_insert BEFORE INSERT ON logs BEGIN SELECT RAISE(ABORT, 'synthetic-secret'); END") + .execute(&pool) + .await + .expect("install failure trigger"); + let diagnostics = SharedWriter::default(); + let layer = start(runtime.clone(), Arc::new(diagnostics.clone())); + assert_eq!(layer.has_write_failure(), false); + let writer = SharedWriter::default(); + let guard = tracing_subscriber::registry() + .with( + tracing_subscriber::fmt::layer() + .with_writer(writer.clone()) + .without_time() + .with_level(/*display_level*/ false) + .with_target(/*display_target*/ false) + .with_ansi(/*ansi*/ false) + .with_filter(LevelFilter::ERROR), + ) + // No diagnostic filter is needed: the diagnostic writer bypasses tracing. + .with(layer.clone()) + .set_default(); + + tracing::info!("sensitive log body"); + layer.flush().await; + assert_eq!(layer.has_write_failure(), true); + let diagnostic = diagnostics.snapshot(); + assert_eq!( + diagnostic.split_once(" ERROR ").unwrap().1, + "failed to flush logs to SQLite error=\"constraint\" entries=1\n" + ); + assert_eq!(writer.snapshot(), ""); + + // Reporting the first failure must not enqueue a second failing batch. + layer.flush().await; + assert_eq!(diagnostics.snapshot(), diagnostic); + assert_eq!(writer.snapshot(), ""); + + sqlx::query("DROP TRIGGER fail_log_insert") + .execute(&pool) + .await + .expect("remove failure trigger"); + tracing::info!("recovered log"); + layer.flush().await; + assert_eq!(diagnostics.snapshot(), diagnostic); + assert_eq!(writer.snapshot(), ""); + assert_eq!(layer.has_write_failure(), true); + drop(guard); + + let logs = runtime + .query_logs(&crate::LogQuery::default()) + .await + .expect("query logs"); + assert_eq!( + logs.iter() + .map(|log| log.message.as_deref()) + .collect::>(), + vec![Some("recovered log")] + ); + pool.close().await; + runtime.close().await; +} diff --git a/codex-rs/state/src/log_db_filter_tests.rs b/codex-rs/state/src/log_db_filter_tests.rs index 28a36c706f..d628dd834a 100644 --- a/codex-rs/state/src/log_db_filter_tests.rs +++ b/codex-rs/state/src/log_db_filter_tests.rs @@ -19,7 +19,10 @@ async fn sqlite_sink_filters_noisy_targets_without_dropping_useful_diagnostics() ) .await .expect("initialize runtime"); - let layer = start(runtime.clone()); + let layer = start( + runtime.clone(), + Arc::new(super::tests::SharedWriter::default()), + ); let guard = tracing_subscriber::registry() .with(layer.clone().with_filter(default_filter())) diff --git a/codex-rs/state/src/telemetry.rs b/codex-rs/state/src/telemetry.rs index ac32d90672..0d9495ba79 100644 --- a/codex-rs/state/src/telemetry.rs +++ b/codex-rs/state/src/telemetry.rs @@ -177,7 +177,7 @@ impl DbOutcomeTags { } } -fn classify_error(err: &anyhow::Error) -> &'static str { +pub(crate) fn classify_error(err: &anyhow::Error) -> &'static str { for cause in err.chain() { if let Some(sqlx_err) = cause.downcast_ref::() { return classify_sqlx_error(sqlx_err); diff --git a/codex-rs/tui/src/startup_orchestration.rs b/codex-rs/tui/src/startup_orchestration.rs index bef842a8aa..cbba29ac3f 100644 --- a/codex-rs/tui/src/startup_orchestration.rs +++ b/codex-rs/tui/src/startup_orchestration.rs @@ -828,7 +828,9 @@ pub(super) async fn run_main_inner( let otel_tracing_layer = otel.as_ref().and_then(|o| o.tracing_layer()); - let log_db = state_db.clone().map(log_db::start); + let log_db = state_db + .clone() + .map(|state_db| log_db::start(state_db, std::sync::Arc::new(feedback.clone()))); let log_db_layer = log_db .clone() .map(|layer| layer.with_filter(log_db::default_filter()));