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()));