Keep thread storage available when diagnostic logs fail (#60466)

## Why

Failure to open the diagnostic logs database should not disable thread storage or database-only history listing. Thread deletion must still preserve saved conversations when diagnostic log cleanup cannot complete.

## What changed

- Allow state initialization to continue after non-corruption errors opening the logs database, while preserving corruption recovery.
- Report the original diagnostic database error after tracing is installed in the TUI and app server, and return it from log operations.
- Reject thread deletion before destructive cleanup when diagnostic storage failed to initialize.
- Continue periodic cleanup of legacy logs when the current logs database is unavailable.

## Testing

Extend coverage for unavailable log directories, database-only history listing, startup diagnostics, deletion preserving saved conversations, corrupt log headers, and periodic legacy log cleanup during a diagnostic storage outage.

GitOrigin-RevId: a5184580ae6c39ed0c4eefa8cb48ee7731a991c3
This commit is contained in:
Dmitriy Kovalenko
2026-11-09 23:22:08 +00:00
committed by copyberry
parent e1cc96712f
commit caae161dd0
11 changed files with 257 additions and 90 deletions
+8
View File
@@ -774,6 +774,14 @@ pub async fn run_main_with_transport_options(
.with(log_db_layer)
.with(otel_layers)
.try_init();
if let Some(error) = state_db.as_ref().and_then(|db| db.diagnostic_logs_error()) {
tracing::warn!(
event.name = "diagnostic_database_unavailable",
error,
"diagnostic database unavailable"
);
}
#[cfg(unix)]
if let Some(Err(err)) = nofile_limit_result {
warn!(%err, "failed to raise managed app-server file descriptor limit");
@@ -1,4 +1,4 @@
//! Verifies startup recovery preserves corrupt databases and allows thread creation.
//! Verifies database failures preserve saved threads and report actionable diagnostics.
use anyhow::Result;
use app_test_support::TestAppServer;
@@ -93,6 +93,24 @@ async fn startup_reports_shared_and_fallback_recovery() -> Result<()> {
backup_pool.close().await;
server.start_thread(ThreadStartParams::default()).await?;
timeout(read_timeout, server.shutdown_gracefully()).await??;
// An unreadable logs header must also reach the outer recovery path.
tokio::fs::write(&logs_path, b"damaged logs database").await?;
let mut server = TestAppServer::builder()
.with_codex_home(sqlite.home())
.build_initialized_with_timeout(read_timeout)
.await?;
let warning: ConfigWarningNotification =
timeout(read_timeout, server.read_notification("configWarning")).await??;
assert_eq!(warning.summary, "Codex rebuilt its local database");
assert!(
warning
.details
.expect("recovery details")
.contains(&logs_path.display().to_string())
);
server.start_thread(ThreadStartParams::default()).await?;
timeout(read_timeout, server.shutdown_gracefully()).await??;
}
Ok(())
}
@@ -142,6 +160,64 @@ async fn state_recovery_restores_saved_threads() -> Result<()> {
let before: ThreadReadResponse = timeout(read_timeout, server.read_response(request)).await??;
timeout(read_timeout, server.shutdown_gracefully()).await??;
// A logs outage must be visible after startup installs tracing, and must
// reject deletion before touching the saved conversation.
let config_path = home.path().join("config.toml");
let config = std::fs::read_to_string(&config_path)?;
let blocked_logs = home.path().join("blocked-logs");
std::fs::write(&blocked_logs, b"not a directory")?;
std::fs::write(
&config_path,
format!("log_dir = {blocked_logs:?}\n{config}"),
)?;
let mut server = TestAppServer::builder()
.with_codex_home(sqlite.home())
.with_json_logging("warn")
.build_initialized_with_timeout(read_timeout)
.await?;
let warning = server
.wait_for_json_log_event("diagnostic_database_unavailable", read_timeout)
.await?;
let cause = warning["fields"]["error"]
.as_str()
.expect("original startup error");
assert!(cause.contains(&blocked_logs.display().to_string()));
assert!(
cause.contains(
&std::fs::create_dir_all(&blocked_logs)
.unwrap_err()
.to_string()
)
);
let rollout_path = before.thread.path.as_ref().expect("saved rollout");
let rollout = std::fs::read(rollout_path)?;
let request = server
.send_thread_delete_request(codex_app_server_protocol::ThreadDeleteParams {
thread_id: thread_id.clone(),
})
.await?;
let error = timeout(
read_timeout,
server.read_stream_until_error_message(codex_app_server_protocol::RequestId::Integer(
request,
)),
)
.await??;
assert_eq!(error.error.code, -32603);
assert!(error.error.message.contains(cause));
assert_eq!(std::fs::read(rollout_path)?, rollout);
let request = server
.send_thread_read_request(ThreadReadParams {
thread_id: thread_id.clone(),
include_turns: true,
})
.await?;
let retained: ThreadReadResponse =
timeout(read_timeout, server.read_response(request)).await??;
assert_eq!(retained.thread.turns, before.thread.turns);
timeout(read_timeout, server.shutdown_gracefully()).await??;
std::fs::write(config_path, config)?;
// Introduce corruption only after the owning process has closed its pools.
let pool = sqlite.open_read_write_pool(&sqlite.state_db_path()).await?;
sqlx::raw_sql(
@@ -1176,14 +1176,17 @@ async fn thread_list_db_only_errors_distinguish_unavailable_database_from_empty_
let other_home = TempDir::new()?;
create_minimal_config(home.path())?;
let loader_overrides = LoaderOverrides::without_managed_config_for_tests();
let config = Arc::new(
ConfigBuilder::default()
.codex_home(home.path().to_path_buf())
.fallback_cwd(Some(home.path().to_path_buf()))
.loader_overrides(loader_overrides.clone())
.build()
.await?,
);
let mut config = ConfigBuilder::default()
.codex_home(home.path().to_path_buf())
.fallback_cwd(Some(home.path().to_path_buf()))
.loader_overrides(loader_overrides.clone())
.build()
.await?;
// Diagnostic storage failure must not disable database-only history listing.
let logs_home = home.path().join("unavailable-logs");
std::fs::write(&logs_home, "not a directory")?;
config.sqlite = config.sqlite.clone().with_local_logs_home(logs_home);
let config = Arc::new(config);
let right_db = codex_state::StateRuntime::init(
config.sqlite_config().clone(),
config.model_provider_id.clone(),
+55 -25
View File
@@ -102,7 +102,8 @@ pub struct StateRuntime {
sqlite: SqliteConfig,
default_provider: String,
pool: Arc<sqlx::SqlitePool>,
logs_pool: Arc<sqlx::SqlitePool>,
logs_pool: Option<Arc<sqlx::SqlitePool>>,
diagnostic_logs_error: Option<String>,
thread_goals: GoalStore,
memories: MemoryStore,
memories_v2: Arc<tokio::sync::OnceCell<MemoryStore>>,
@@ -159,16 +160,22 @@ impl StateRuntime {
}
};
let has_memories_v2 = tokio::fs::try_exists(sqlite.memories_v2_db_path()).await?;
let logs_pool = match sqlite
let (logs_pool, diagnostic_logs_error) = match sqlite
.open_logs_db(&logs_migrator, telemetry_override)
.await
{
Ok(db) => Arc::new(db),
Err(err) => {
warn!("failed to open logs db at {}: {err}", logs_path.display());
close_sqlite_pools(&[pool.as_ref()]).await;
Ok(db) => (Some(Arc::new(db)), None),
Err(err) if is_sqlite_corruption_error(&err) => {
pool.close().await;
return Err(err);
}
Err(err) => (
None,
Some(format!(
"failed to open logs db at {}: {err:#}",
logs_path.display()
)),
),
};
let goals_pool = match sqlite
.open_goals_db(&goals_migrator, telemetry_override)
@@ -177,7 +184,7 @@ impl StateRuntime {
Ok(db) => Arc::new(db),
Err(err) => {
warn!("failed to open goals db at {}: {err}", goals_path.display());
close_sqlite_pools(&[pool.as_ref(), logs_pool.as_ref()]).await;
close_sqlite_pools(&[Some(pool.as_ref()), logs_pool.as_deref()]).await;
return Err(err);
}
};
@@ -191,7 +198,12 @@ impl StateRuntime {
"failed to open memories db at {}: {err}",
memories_path.display()
);
close_sqlite_pools(&[pool.as_ref(), logs_pool.as_ref(), goals_pool.as_ref()]).await;
close_sqlite_pools(&[
Some(pool.as_ref()),
logs_pool.as_deref(),
Some(goals_pool.as_ref()),
])
.await;
return Err(err);
}
};
@@ -203,10 +215,10 @@ impl StateRuntime {
Err(err) => {
warn!("failed to open queue db at {}: {err}", queue_path.display());
close_sqlite_pools(&[
pool.as_ref(),
logs_pool.as_ref(),
goals_pool.as_ref(),
memories_pool.as_ref(),
Some(pool.as_ref()),
logs_pool.as_deref(),
Some(goals_pool.as_ref()),
Some(memories_pool.as_ref()),
])
.await;
return Err(err);
@@ -223,11 +235,11 @@ impl StateRuntime {
);
if let Err(err) = backfill_state_result {
close_sqlite_pools(&[
pool.as_ref(),
logs_pool.as_ref(),
goals_pool.as_ref(),
memories_pool.as_ref(),
queue_pool.as_ref(),
Some(pool.as_ref()),
logs_pool.as_deref(),
Some(goals_pool.as_ref()),
Some(memories_pool.as_ref()),
Some(queue_pool.as_ref()),
])
.await;
return Err(err);
@@ -252,11 +264,11 @@ impl StateRuntime {
Ok(value) => value,
Err(err) => {
close_sqlite_pools(&[
pool.as_ref(),
logs_pool.as_ref(),
goals_pool.as_ref(),
memories_pool.as_ref(),
queue_pool.as_ref(),
Some(pool.as_ref()),
logs_pool.as_deref(),
Some(goals_pool.as_ref()),
Some(memories_pool.as_ref()),
Some(queue_pool.as_ref()),
])
.await;
return Err(err);
@@ -272,6 +284,7 @@ impl StateRuntime {
thread_queue: SqliteQueueStore::new(queue_pool),
pool,
logs_pool,
diagnostic_logs_error,
sqlite,
default_provider,
thread_updated_at_millis: Arc::new(AtomicI64::new(thread_updated_at_millis)),
@@ -309,6 +322,21 @@ impl StateRuntime {
&self.thread_queue
}
/// Startup failure to report after tracing is installed and before destructive cleanup.
pub fn diagnostic_logs_error(&self) -> Option<&str> {
self.diagnostic_logs_error.as_deref()
}
fn diagnostic_logs_pool(&self) -> anyhow::Result<&SqlitePool> {
self.logs_pool.as_deref().ok_or_else(|| {
anyhow::anyhow!(
"diagnostic database unavailable: {}",
self.diagnostic_logs_error()
.unwrap_or("logs pool was not initialized")
)
})
}
/// Close all SQLite pools and wait for outstanding pool workers to exit.
pub async fn close(&self) {
self.reclamation.close().await;
@@ -318,7 +346,9 @@ impl StateRuntime {
memories.close().await;
}
self.thread_goals.close().await;
self.logs_pool.close().await;
if let Some(pool) = &self.logs_pool {
pool.close().await;
}
self.pool.close().await;
}
@@ -355,8 +385,8 @@ impl StateRuntime {
}
}
async fn close_sqlite_pools(pools: &[&SqlitePool]) {
for pool in pools {
async fn close_sqlite_pools(pools: &[Option<&SqlitePool>]) {
for pool in pools.iter().flatten() {
pool.close().await;
}
}
+7 -4
View File
@@ -11,7 +11,7 @@ impl StateRuntime {
return Ok(());
}
let mut tx = self.logs_pool.begin().await?;
let mut tx = self.diagnostic_logs_pool()?.begin().await?;
let mut builder = QueryBuilder::<Sqlite>::new(
"INSERT INTO logs (ts, ts_nanos, level, target, feedback_log_body, thread_id, process_uuid, module_path, file, line, estimated_bytes) ",
);
@@ -300,7 +300,7 @@ WHERE id IN (
let rows = builder
.build_query_as::<LogRow>()
.fetch_all(self.logs_pool.as_ref())
.fetch_all(self.diagnostic_logs_pool()?)
.await?;
Ok(rows)
}
@@ -310,7 +310,7 @@ WHERE id IN (
&self,
thread_ids: &[&str],
) -> anyhow::Result<Vec<u8>> {
Self::query_feedback_logs_from_pool(self.logs_pool.as_ref(), thread_ids).await
Self::query_feedback_logs_from_pool(self.diagnostic_logs_pool()?, thread_ids).await
}
/// Read logs from the previous directory without creating or migrating its database.
@@ -441,7 +441,10 @@ WHERE cumulative_estimated_bytes <=
let mut builder =
QueryBuilder::<Sqlite>::new("SELECT MAX(id) AS max_id FROM logs WHERE 1 = 1");
push_log_filters(&mut builder, query);
let row = builder.build().fetch_one(self.logs_pool.as_ref()).await?;
let row = builder
.build()
.fetch_one(self.diagnostic_logs_pool()?)
.await?;
let max_id: Option<i64> = row.try_get("max_id")?;
Ok(max_id.unwrap_or(0))
}
@@ -1,4 +1,5 @@
//! Prune current and legacy diagnostic logs immediately, then on a timer.
//! Legacy cleanup continues when current diagnostics are unavailable.
//! SQLite keeps freed pages for reuse.
use super::StateRuntime;
@@ -26,7 +27,8 @@ impl StateRuntime {
let Some(runtime) = runtime.upgrade() else {
break;
};
if runtime.logs_pool.is_closed() {
let logs_pool = runtime.logs_pool.as_deref();
if runtime.pool.is_closed() || logs_pool.is_some_and(SqlitePool::is_closed) {
break;
}
// Older versions write logs in sqlite_home. Keep the same retention policy
@@ -38,8 +40,7 @@ impl StateRuntime {
None
}
};
for pool in std::iter::once(runtime.logs_pool.as_ref()).chain(legacy_pool.as_ref())
{
for pool in logs_pool.into_iter().chain(legacy_pool.as_ref()) {
if let Err(err) = prune_by_age_and_size(
pool,
Utc::now().timestamp(),
@@ -94,55 +94,70 @@ async fn stops_at_one_second_when_recent_rows_exceed_budget() {
#[tokio::test]
async fn periodic_cleanup_prunes_without_new_log_writes_or_restart() {
let directory = super::super::test_support::unique_temp_dir();
let sqlite = SqliteConfig::new_for_testing(directory.as_path().abs());
let runtime = StateRuntime::init(
sqlite.clone().with_local_logs_home(directory.join("log")),
"test-provider".to_string(),
)
.await
.unwrap();
// An older process can create the legacy database after the new runtime starts.
let legacy_pool = sqlite
.open_logs_db(&LOGS_MIGRATOR, /*telemetry_override*/ None)
for current_logs_available in [true, false] {
let directory = super::super::test_support::unique_temp_dir();
let sqlite = SqliteConfig::new_for_testing(directory.as_path().abs());
let logs_home = directory.join("log");
if !current_logs_available {
std::fs::create_dir_all(&directory).unwrap();
std::fs::write(&logs_home, "not a directory").unwrap();
}
let runtime = StateRuntime::init(
sqlite.clone().with_local_logs_home(logs_home),
"test-provider".to_string(),
)
.await
.unwrap();
runtime.start_periodic_logs_maintenance(Duration::from_millis(10));
let now = Utc::now().timestamp();
for round in 1..=2 {
for pool in [runtime.logs_pool.as_ref(), &legacy_pool] {
for timestamp in [now - 11 * DAY, now] {
sqlx::query(
"INSERT INTO logs (ts, ts_nanos, level, target) VALUES (?, 0, 'INFO', 'test')",
)
.bind(timestamp)
.execute(pool)
.await
.unwrap();
}
}
tokio::time::timeout(Duration::from_secs(5), async {
loop {
let current: Vec<i64> = sqlx::query_scalar("SELECT ts FROM logs ORDER BY ts")
.fetch_all(runtime.logs_pool.as_ref())
assert_eq!(runtime.logs_pool.is_some(), current_logs_available);
// An older process can create the legacy database after the new runtime starts.
let legacy_pool = sqlite
.open_logs_db(&LOGS_MIGRATOR, /*telemetry_override*/ None)
.await
.unwrap();
runtime.start_periodic_logs_maintenance(Duration::from_millis(10));
let now = Utc::now().timestamp();
for round in 1..=2 {
let pools: Vec<_> = runtime
.logs_pool
.as_deref()
.into_iter()
.chain([&legacy_pool])
.collect();
for pool in &pools {
for timestamp in [now - 11 * DAY, now] {
sqlx::query(
"INSERT INTO logs (ts, ts_nanos, level, target) VALUES (?, 0, 'INFO', 'test')",
)
.bind(timestamp)
.execute(*pool)
.await
.unwrap();
let legacy: Vec<i64> = sqlx::query_scalar("SELECT ts FROM logs ORDER BY ts")
.fetch_all(&legacy_pool)
.await
.unwrap();
if (current, legacy) == (vec![now; round], vec![now; round]) {
break;
}
tokio::time::sleep(Duration::from_millis(10)).await;
}
})
.await
.expect("periodic cleanup should remove expired logs from both locations while idle");
tokio::time::timeout(Duration::from_secs(5), async {
loop {
let mut retained = Vec::new();
for pool in &pools {
retained.push(
sqlx::query_scalar::<_, i64>("SELECT ts FROM logs ORDER BY ts")
.fetch_all(*pool)
.await
.unwrap(),
);
}
if retained == vec![vec![now; round]; pools.len()] {
break;
}
tokio::time::sleep(Duration::from_millis(10)).await;
}
})
.await
.expect("periodic cleanup should remove expired logs from both locations while idle");
}
legacy_pool.close().await;
runtime.close().await;
tokio::fs::remove_dir_all(directory).await.unwrap();
}
legacy_pool.close().await;
runtime.close().await;
tokio::fs::remove_dir_all(directory).await.unwrap();
}
#[tokio::test]
@@ -183,12 +198,12 @@ async fn legacy_writer_does_not_block_current_cleanup_and_is_retried() -> anyhow
)
.await?;
sqlx::query("INSERT INTO logs (ts, ts_nanos, level, target) VALUES (0, 0, 'INFO', 'test')")
.execute(runtime.logs_pool.as_ref())
.execute(runtime.diagnostic_logs_pool().unwrap())
.await?;
runtime.start_periodic_logs_maintenance(Duration::from_millis(10));
tokio::time::timeout(Duration::from_secs(5), async {
while sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM logs")
.fetch_one(runtime.logs_pool.as_ref())
.fetch_one(runtime.diagnostic_logs_pool().unwrap())
.await
.unwrap()
!= 0
@@ -263,7 +278,7 @@ async fn startup_cleanup_runs_without_waiting_for_the_period() {
tokio::time::timeout(Duration::from_secs(5), async {
loop {
let current: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM logs")
.fetch_one(runtime.logs_pool.as_ref())
.fetch_one(runtime.diagnostic_logs_pool().unwrap())
.await
.unwrap();
let legacy: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM logs")
+3 -3
View File
@@ -1265,7 +1265,7 @@ ON CONFLICT(id) DO UPDATE SET
for (thread_id, thread_id_string) in thread_ids.iter().zip(&thread_id_strings) {
sqlx::query("DELETE FROM logs WHERE thread_id = ?")
.bind(thread_id_string)
.execute_with_busy_retry(self.logs_pool.as_ref())
.execute_with_busy_retry(self.diagnostic_logs_pool()?)
.await?;
self.thread_queue.delete_thread_queue(*thread_id).await?;
self.delete_versioned_thread_memory(*thread_id).await?;
@@ -2254,7 +2254,7 @@ mod tests {
.await?;
seed_thread_cleanup_state(&runtime, thread_id, child_thread_id).await?;
runtime.logs_pool.close().await;
runtime.diagnostic_logs_pool().unwrap().close().await;
runtime
.delete_thread(thread_id)
.await
@@ -2291,7 +2291,7 @@ mod tests {
.await?;
sqlx::query("INSERT INTO logs (ts, ts_nanos, level, target, feedback_log_body, thread_id) VALUES (1, 0, 'INFO', 'test', 'feedback log', ?)")
.bind(thread_id.to_string())
.execute(runtime.logs_pool.as_ref())
.execute(runtime.diagnostic_logs_pool().unwrap())
.await?;
Ok(())
}
+8 -1
View File
@@ -88,11 +88,18 @@ async fn configured_log_directory_is_initialized_on_use_and_retries_after_failur
// Opening unrelated state must not require the configured logs directory.
let pool = sqlite.open_read_write_pool(&sqlite.state_db_path()).await?;
pool.close().await;
let runtime = crate::StateRuntime::init(sqlite.clone(), "openai".to_string()).await?;
runtime.get_backfill_state().await?;
assert!(
crate::StateRuntime::init(sqlite.clone(), "openai".to_string())
runtime
.max_log_id(&crate::LogQuery::default())
.await
.is_err()
);
assert!(sqlite.goals_db_path().is_file());
assert!(sqlite.memories_db_path().is_file());
assert!(sqlite.queue_db_path().is_file());
runtime.close().await;
assert!(!fallback_logs_path.exists());
// A failed initialization must remain retryable without replacing the config.
@@ -5,6 +5,7 @@
//! associated rollout is removed, under the same lifecycle lock, so queued artifact mutations cannot
//! use deleted thread metadata. Host-owned data is cleaned after reference and writer checks,
//! before removing rollouts, so cleanup failures remain retryable even without a state DB.
//! Known diagnostic database failures must be rejected before any destructive cleanup.
use std::collections::HashMap;
use std::collections::HashSet;
@@ -41,11 +42,25 @@ impl LocalThreadStore {
&self,
thread_ids: &[ThreadId],
) -> ThreadStoreResult<()> {
ensure_diagnostic_logs_available(self)?;
let reference_index = scan_reference_index(self).await?;
check_attached_children(&reference_index, thread_ids)
}
}
fn ensure_diagnostic_logs_available(store: &LocalThreadStore) -> ThreadStoreResult<()> {
if let Some(error) = store
.state_db
.as_ref()
.and_then(|db| db.diagnostic_logs_error())
{
return Err(ThreadStoreError::Internal {
message: format!("cannot delete thread while diagnostic logs are unavailable: {error}"),
});
}
Ok(())
}
fn check_attached_children(
reference_index: &RolloutReferenceIndex,
thread_ids: &[ThreadId],
@@ -94,6 +109,7 @@ pub(super) async fn delete_thread(
store: &LocalThreadStore,
params: DeleteThreadParams,
) -> ThreadStoreResult<()> {
ensure_diagnostic_logs_available(store)?;
let thread_id = params.thread_id;
let _lifecycle_guard = store.live_writer_locks.lock_lifecycle(thread_id).await;
let _live_writer_guard = store.live_writer_locks.lock(thread_id).await;
@@ -129,6 +145,7 @@ pub(super) async fn delete_threads(
return Ok(());
}
ensure_diagnostic_logs_available(store)?;
let mut lock_thread_ids = thread_ids.clone();
lock_thread_ids.sort_unstable_by_key(ToString::to_string);
lock_thread_ids.dedup();
@@ -988,6 +988,13 @@ pub(super) async fn run_main_inner(
.with(otel_logger_layer)
.with(otel_tracing_layer)
.try_init();
if let Some(error) = state_db.as_ref().and_then(|db| db.diagnostic_logs_error()) {
tracing::warn!(
event.name = "diagnostic_database_unavailable",
error,
"diagnostic database unavailable"
);
}
if let Err(err) = screen_reader_result {
tracing::warn!("Could not save screen-reader detection: {err}");