From b8b61bc692517adcd18622df260f2ddd80635122 Mon Sep 17 00:00:00 2001 From: Felipe Coury Date: Sat, 18 Jul 2026 16:07:25 +0000 Subject: [PATCH] Handle compressed rollouts in doctor thread inventory (#34038) ## Why The thread inventory check compared state database rows only with plain `.jsonl` rollout files. Once a rollout was compressed to `.jsonl.zst`, the check could report its canonical database row as stale and omit the rollout from parity results. ## What changed - Scan completed compressed rollouts and match them to database rows using their canonical `.jsonl` paths. - Prefer a plain rollout when both plain and compressed siblings exist, and ignore compression temporary files. - Keep unreadable compressed rollouts in the set of existing paths so they produce scan errors without also producing misleading stale-row reports. - Use canonical paths when deriving legacy thread IDs and archived status. ## Testing Added coverage for active and archived compressed rollouts, plain/compressed siblings, metadata and legacy filename IDs, corrupt compressed data, and compression temporary files. GitOrigin-RevId: 2461376432194a4062beedbe875aa7002dd2049f --- codex-rs/Cargo.lock | 1 + codex-rs/cli/Cargo.toml | 1 + codex-rs/cli/src/doctor/thread_inventory.rs | 218 +++++++++++++++++++- 3 files changed, 210 insertions(+), 10 deletions(-) diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index 1e3bae8936..f16cf6c938 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -2378,6 +2378,7 @@ dependencies = [ "which 8.0.0", "windows-sys 0.52.0", "wiremock", + "zstd", ] [[package]] diff --git a/codex-rs/cli/Cargo.toml b/codex-rs/cli/Cargo.toml index 46702be159..f223629148 100644 --- a/codex-rs/cli/Cargo.toml +++ b/codex-rs/cli/Cargo.toml @@ -104,6 +104,7 @@ predicates = { workspace = true } pretty_assertions = { workspace = true } sqlx = { workspace = true } wiremock = { workspace = true } +zstd = { workspace = true } [package.metadata.cargo-shear] # These Rust sources are intentionally Bazel-only macrobenchmarks rather than diff --git a/codex-rs/cli/src/doctor/thread_inventory.rs b/codex-rs/cli/src/doctor/thread_inventory.rs index 2eb9003ba5..61f69ed49b 100644 --- a/codex-rs/cli/src/doctor/thread_inventory.rs +++ b/codex-rs/cli/src/doctor/thread_inventory.rs @@ -36,6 +36,7 @@ struct RolloutAuditFile { #[derive(Default)] struct RolloutScan { files: Vec, + existing_keys: HashSet, scan_errors: Vec, malformed_names: Vec, reached_scan_cap: bool, @@ -231,7 +232,7 @@ fn parity_check_from_scan_and_rows( let mut rows_by_key: HashMap> = HashMap::new(); for row in &rows { rows_by_key - .entry(path_key(&row.rollout_path)) + .entry(rollout_path_key(&row.rollout_path)) .or_default() .push(row); } @@ -241,7 +242,12 @@ fn parity_check_from_scan_and_rows( let scan_complete = !scan.reached_scan_cap; let stale_rows = if scan_complete { rows.iter() - .filter(|row| !row.rollout_path.is_file()) + .filter(|row| { + !scan + .existing_keys + .contains(&rollout_path_key(&row.rollout_path)) + && !row.rollout_path.is_file() + }) .collect::>() } else { Vec::new() @@ -250,13 +256,15 @@ fn parity_check_from_scan_and_rows( rows.iter() .filter_map(|row| { let expected_archived = rollout_by_key - .get(&path_key(&row.rollout_path)) + .get(&rollout_path_key(&row.rollout_path)) .map(|file| file.archived) .or_else(|| { - row.rollout_path - .is_file() - .then(|| archived_from_rollout_path(codex_home, &row.rollout_path)) - .flatten() + (scan + .existing_keys + .contains(&rollout_path_key(&row.rollout_path)) + || row.rollout_path.is_file()) + .then(|| archived_from_rollout_path(codex_home, &row.rollout_path)) + .flatten() })?; (expected_archived != row.archived).then_some(row) }) @@ -482,13 +490,19 @@ async fn scan_rollout_root(root: &Path, archived: bool, scan: &mut RolloutScan) dirs.push(path); continue; } - if !file_type.is_file() || !is_rollout_file(&path) { + let logical_path = codex_rollout::plain_rollout_path(&path); + if !file_type.is_file() + || !is_rollout_file(&logical_path) + || (path != logical_path && logical_path.is_file()) + { continue; } if scan.candidate_count() >= MAX_PARITY_SCAN_FILES { scan.reached_scan_cap = true; return; } + let key = path_key(&logical_path); + scan.existing_keys.insert(key.clone()); let thread_id = match thread_id_from_rollout(&path).await { RolloutThreadId::Id(thread_id) => thread_id, RolloutThreadId::MalformedName => { @@ -501,7 +515,7 @@ async fn scan_rollout_root(root: &Path, archived: bool, scan: &mut RolloutScan) } }; scan.files.push(RolloutAuditFile { - key: path_key(&path), + key, path, archived, thread_id, @@ -558,7 +572,8 @@ async fn thread_id_from_rollout(path: &Path) -> RolloutThreadId { } // Legacy rollouts can omit session metadata, so use the validated filename fallback after // the bounded prefix without retaining the first item or loading the full history. - codex_rollout::builder_from_items(&[], path) + let logical_path = codex_rollout::plain_rollout_path(path); + codex_rollout::builder_from_items(&[], &logical_path) .map(|builder| RolloutThreadId::Id(builder.id.to_string())) .unwrap_or(RolloutThreadId::MalformedName) } @@ -583,6 +598,10 @@ fn path_key(path: &Path) -> PathBuf { normalize_for_path_comparison(path).unwrap_or_else(|_| path.to_path_buf()) } +fn rollout_path_key(path: &Path) -> PathBuf { + path_key(&codex_rollout::plain_rollout_path(path)) +} + fn archived_from_rollout_path(codex_home: &Path, path: &Path) -> Option { let key = path_key(path); if key.starts_with(path_key(&codex_home.join("archived_sessions"))) { @@ -892,6 +911,171 @@ mod tests { assert_detail(&check, "rollout DB missing active rows", "0"); } + #[tokio::test] + async fn thread_inventory_check_matches_compressed_rollouts_to_canonical_db_paths() { + let fixture = Fixture::new().await; + let active_id = "00000000-0000-0000-0000-000000000001"; + let archived_id = "00000000-0000-0000-0000-000000000002"; + let active_path = + fixture.write_rollout(/*archived*/ false, "2025-01-02T10-00-00", active_id); + let archived_path = + fixture.write_rollout(/*archived*/ true, "2025-01-02T11-00-00", archived_id); + compress_rollout(&active_path); + compress_rollout(&archived_path); + fixture + .insert_thread_row(active_id, active_path.as_path(), /*archived*/ false) + .await; + fixture + .insert_thread_row(archived_id, archived_path.as_path(), /*archived*/ true) + .await; + + let check = thread_inventory_check_for_roots( + fixture.codex_home.path(), + fixture.sqlite_home.path(), + "test-provider", + ) + .await; + + assert_eq!(check.status, CheckStatus::Ok); + assert_detail(&check, "rollout DB active files", "1"); + assert_detail(&check, "rollout DB archived files", "1"); + assert_detail(&check, "rollout DB scan errors", "0"); + assert_detail(&check, "rollout DB missing active rows", "0"); + assert_detail(&check, "rollout DB missing archived rows", "0"); + assert_detail(&check, "rollout DB stale rows", "0"); + } + + #[tokio::test] + async fn thread_inventory_check_prefers_plain_rollout_over_compressed_sibling() { + let fixture = Fixture::new().await; + let thread_id = "00000000-0000-0000-0000-000000000001"; + let path = fixture.write_rollout(/*archived*/ false, "2025-01-02T10-00-00", thread_id); + compress_rollout(&path); + fixture.write_rollout(/*archived*/ false, "2025-01-02T10-00-00", thread_id); + fixture + .insert_thread_row(thread_id, path.as_path(), /*archived*/ false) + .await; + + let check = thread_inventory_check_for_roots( + fixture.codex_home.path(), + fixture.sqlite_home.path(), + "test-provider", + ) + .await; + + assert_eq!(check.status, CheckStatus::Ok); + assert_detail(&check, "rollout DB active files", "1"); + assert_detail(&check, "rollout DB duplicate rollout thread ids", "0"); + assert_detail(&check, "rollout DB stale rows", "0"); + } + + #[tokio::test] + async fn thread_inventory_check_uses_compressed_metadata_id_and_legacy_filename_fallback() { + let fixture = Fixture::new().await; + let filename_id = "00000000-0000-0000-0000-000000000001"; + let metadata_id = ThreadId::from_string("00000000-0000-0000-0000-000000000002") + .expect("metadata thread id"); + let metadata_path = + fixture.write_rollout(/*archived*/ false, "2025-01-02T10-00-00", filename_id); + let contents = std::fs::read_to_string(&metadata_path).expect("rollout file"); + let mut rollout_line = + serde_json::from_str::(contents.trim()).expect("rollout line"); + let RolloutItem::SessionMeta(session_meta) = &mut rollout_line.item else { + panic!("expected session metadata"); + }; + session_meta.meta.session_id = metadata_id.into(); + session_meta.meta.id = metadata_id; + let contents = serde_json::to_string(&rollout_line).expect("rollout line"); + std::fs::write(&metadata_path, format!("{contents}\n")).expect("rollout file"); + compress_rollout(&metadata_path); + fixture + .insert_thread_row( + filename_id, + metadata_path.as_path(), + /*archived*/ false, + ) + .await; + + let legacy_id = "00000000-0000-0000-0000-000000000003"; + let legacy_path = fixture.codex_home.path().join(format!( + "sessions/2025/01/02/rollout-2025-01-02T11-00-00-{legacy_id}.jsonl" + )); + std::fs::write( + &legacy_path, + "{\"timestamp\":\"2025-01-02T11:00:00Z\",\"type\":\"compacted\",\"payload\":{\"message\":\"legacy history\"}}\n", + ) + .expect("legacy rollout"); + compress_rollout(&legacy_path); + fixture + .insert_thread_row(legacy_id, legacy_path.as_path(), /*archived*/ false) + .await; + + let check = thread_inventory_check_for_roots( + fixture.codex_home.path(), + fixture.sqlite_home.path(), + "test-provider", + ) + .await; + + assert_eq!(check.status, CheckStatus::Warning); + assert_detail(&check, "rollout DB active files", "2"); + assert_detail(&check, "rollout DB malformed file names", "0"); + assert_detail(&check, "rollout DB scan errors", "0"); + assert_detail(&check, "rollout DB missing active rows", "1"); + assert_detail(&check, "rollout DB stale rows", "0"); + } + + #[tokio::test] + async fn thread_inventory_check_reports_corrupt_compressed_rollout_without_stale_row() { + let fixture = Fixture::new().await; + let thread_id = "00000000-0000-0000-0000-000000000001"; + let path = fixture.write_rollout(/*archived*/ false, "2025-01-02T10-00-00", thread_id); + let compressed_path = path.with_file_name(format!( + "{}.zst", + path.file_name() + .expect("rollout file name") + .to_string_lossy() + )); + std::fs::write(&compressed_path, "not zstd").expect("corrupt compressed rollout"); + std::fs::remove_file(&path).expect("remove plain rollout"); + fixture + .insert_thread_row(thread_id, path.as_path(), /*archived*/ false) + .await; + + let check = thread_inventory_check_for_roots( + fixture.codex_home.path(), + fixture.sqlite_home.path(), + "test-provider", + ) + .await; + + assert_eq!(check.status, CheckStatus::Warning); + assert_detail(&check, "rollout DB scan errors", "1"); + assert_detail(&check, "rollout DB stale rows", "0"); + } + + #[tokio::test] + async fn thread_inventory_check_ignores_compression_temp_files() { + let fixture = Fixture::new().await; + let temp_path = fixture.codex_home.path().join( + "sessions/2025/01/02/rollout-2025-01-02T10-00-00-00000000-0000-0000-0000-000000000001.jsonl.zst.compress.1.0.tmp", + ); + std::fs::create_dir_all(temp_path.parent().expect("rollout temp parent")) + .expect("rollout temp dir"); + std::fs::write(temp_path, "not a completed rollout").expect("rollout temp file"); + + let check = thread_inventory_check_for_roots( + fixture.codex_home.path(), + fixture.sqlite_home.path(), + "test-provider", + ) + .await; + + assert_eq!(check.status, CheckStatus::Ok); + assert_detail(&check, "rollout DB active files", "0"); + assert_detail(&check, "rollout DB scan errors", "0"); + } + #[tokio::test] async fn thread_id_from_rollout_falls_back_for_legacy_non_meta_record() { let home = TempDir::new().expect("temp dir"); @@ -1100,6 +1284,20 @@ mod tests { sqlite_home: TempDir, } + fn compress_rollout(path: &Path) { + let compressed_path = path.with_file_name(format!( + "{}.zst", + path.file_name() + .expect("rollout file name") + .to_string_lossy() + )); + let contents = std::fs::read(path).expect("rollout file"); + let compressed = + zstd::stream::encode_all(contents.as_slice(), /*level*/ 3).expect("compress rollout"); + std::fs::write(compressed_path, compressed).expect("compressed rollout"); + std::fs::remove_file(path).expect("remove plain rollout"); + } + impl Fixture { async fn new() -> Self { let codex_home = TempDir::new().expect("codex home");