From 465eafacbc2db4ff828cd6d18ed8f25d22e48f53 Mon Sep 17 00:00:00 2001 From: Owen Lin Date: Mon, 24 Aug 2026 23:55:51 +0000 Subject: [PATCH] Harden startup rollout migration against concurrent updates (#40499) ## Why Another Codex process can write, archive, or compress a rollout while startup migration is inspecting it. This can leave a discovered path stale or make an in-progress rollout look empty or busy. ## What changed - Wait for rollout maintenance to finish before starting background migration. - Re-read empty rollouts under their writer lock and retry busy rollouts on a later startup. - Find archived or compressed rollouts again when their paths change after discovery, while keeping terminal failures from blocking the startup cursor. ## Testing Added coverage for maintenance contention, writer-owned empty rollouts, busy rollout retries after archive and compression moves, permanent failure skips, pending recovery, and paths that move after discovery. GitOrigin-RevId: 9708f212f2bbff88bd47a01ec4872bf6ce9af535 --- .../src/local/rollout_migration.rs | 159 ++++++--- .../src/local/rollout_migration/startup.rs | 294 +++++++++++------ .../local/rollout_migration/startup_tests.rs | 304 +++++++++++++++--- .../src/local/rollout_migration_tests.rs | 33 ++ 4 files changed, 599 insertions(+), 191 deletions(-) diff --git a/codex-rs/thread-store/src/local/rollout_migration.rs b/codex-rs/thread-store/src/local/rollout_migration.rs index 028b6d0094..46d71d5f11 100644 --- a/codex-rs/thread-store/src/local/rollout_migration.rs +++ b/codex-rs/thread-store/src/local/rollout_migration.rs @@ -81,6 +81,11 @@ enum CanonicalizationAttempt { NeedsRollbackPlan, } +enum RolloutMigrationPaths { + Discover, + Known(Vec), +} + struct CanonicalizationSource<'a> { thread_id: ThreadId, source_path: &'a Path, @@ -268,6 +273,7 @@ impl LocalThreadStore { options, |_| {}, RolloutMigrationTrigger::Manual, + RolloutMigrationPaths::Discover, ) .await } @@ -282,6 +288,7 @@ impl LocalThreadStore { options, on_progress, RolloutMigrationTrigger::Manual, + RolloutMigrationPaths::Discover, ) .await } @@ -291,10 +298,11 @@ impl LocalThreadStore { options: RolloutMigrationOptions, mut on_progress: impl FnMut(RolloutMigrationProgress), trigger: RolloutMigrationTrigger, + paths: RolloutMigrationPaths, ) -> ThreadStoreResult { let telemetry = RolloutMigrationTelemetry::new(trigger, &options); let result = self - .migrate_rollouts_with_progress_inner(options, &mut on_progress) + .migrate_rollouts_with_progress_inner(options, &mut on_progress, paths) .await; telemetry.finish(&result); result @@ -304,6 +312,7 @@ impl LocalThreadStore { &self, options: RolloutMigrationOptions, on_progress: &mut impl FnMut(RolloutMigrationProgress), + paths: RolloutMigrationPaths, ) -> ThreadStoreResult { let mut limiter = RolloutMigrationRateLimiter::new(options.max_mib_per_second)?; let _maintenance_guard = match options.mode { @@ -317,18 +326,12 @@ impl LocalThreadStore { })?, ), }; - let mut paths = - find_rollout_paths(&self.config.codex_home.join(codex_rollout::SESSIONS_SUBDIR)) - .await?; - paths.extend( - find_rollout_paths( - &self - .config - .codex_home - .join(codex_rollout::ARCHIVED_SESSIONS_SUBDIR), - ) - .await?, - ); + let mut paths = match paths { + RolloutMigrationPaths::Discover => { + find_all_rollout_paths(&self.config.codex_home).await? + } + RolloutMigrationPaths::Known(paths) => paths, + }; if options.mode == RolloutMigrationMode::Apply { let pending_thread_ids = pending_migration_thread_ids(&self.config.codex_home).await?; paths.sort_by_key(|path| { @@ -376,36 +379,76 @@ impl LocalThreadStore { legacy_names: &HashMap, limiter: &mut RolloutMigrationRateLimiter, ) -> ThreadStoreResult> { - let metadata = match codex_rollout::read_session_meta_line(&path).await { - Ok(metadata) => metadata, - Err(error) => { - let thread_id = thread_id_from_rollout_filename(&path); - if !matches_selection(&options.thread_ids, thread_id) { - return Ok(None); + let mut retried_moved_path = false; + let metadata = loop { + let error = match codex_rollout::read_session_meta_line(&path).await { + Ok(metadata) => break metadata, + Err(error) if !retried_moved_path && error.kind() == io::ErrorKind::NotFound => { + // A different Codex process can archive or compress this rollout after path + // discovery but before we take its writer lock. Retry the same rollout under + // its current root/suffix before treating the missing snapshot path as failed. + if let Some(current_path) = + find_current_rollout_path(&self.config.codex_home, &path).await? + { + path = current_path; + retried_moved_path = true; + continue; + } + error } - let empty = tokio::fs::metadata(&path) - .await - .is_ok_and(|metadata| metadata.len() == 0); - let failure_reason = if empty { - None - } else if error.kind() != io::ErrorKind::Other { - Some(RolloutMigrationFailureReason::RolloutReadFailed) - } else { - Some(RolloutMigrationFailureReason::InvalidSessionMetadata) - }; - return Ok(Some(RolloutMigrationOutcome { - thread_id, - rollout_path: path, - status: if empty { - RolloutMigrationStatus::SkippedEmpty - } else { - RolloutMigrationStatus::Failed - }, - failure_reason, - bytes_processed: 0, - message: (!empty).then(|| error.to_string()), - })); + Err(error) => error, + }; + let thread_id = thread_id_from_rollout_filename(&path); + if !matches_selection(&options.thread_ids, thread_id) { + return Ok(None); } + let initially_empty = tokio::fs::metadata(&path) + .await + .is_ok_and(|metadata| metadata.len() == 0); + // Writers create the rollout path before SessionMeta is durable. Re-read under the + // writer lock so a writer that just finished does not become a permanent empty skip. + let error = if initially_empty + && options.mode == RolloutMigrationMode::Apply + && let Some(thread_id) = thread_id + { + let _writer_guard = match self.writer_lock_coordinator.acquire(thread_id) { + Ok(guard) => guard, + Err(ThreadStoreError::Conflict { message }) => { + return Ok(Some(skipped_busy_outcome( + thread_id, path, message, /*bytes_processed*/ 0, + ))); + } + Err(error) => return Err(error), + }; + match codex_rollout::read_session_meta_line(&path).await { + Ok(metadata) => break metadata, + Err(error) => error, + } + } else { + error + }; + let empty = tokio::fs::metadata(&path) + .await + .is_ok_and(|metadata| metadata.len() == 0); + let failure_reason = if empty { + None + } else if error.kind() != io::ErrorKind::Other { + Some(RolloutMigrationFailureReason::RolloutReadFailed) + } else { + Some(RolloutMigrationFailureReason::InvalidSessionMetadata) + }; + return Ok(Some(RolloutMigrationOutcome { + thread_id, + rollout_path: path, + status: if empty { + RolloutMigrationStatus::SkippedEmpty + } else { + RolloutMigrationStatus::Failed + }, + failure_reason, + bytes_processed: 0, + message: (!empty).then(|| error.to_string()), + })); }; let thread_id = metadata.meta.id; if !matches_selection(&options.thread_ids, Some(thread_id)) { @@ -511,6 +554,16 @@ impl LocalThreadStore { ))); } }; + // SessionMeta gives us the writer-lock id, so archiving can win between that read and + // lock acquisition. Once the lock is ours, follow the same rollout to its current path. + if !tokio::fs::try_exists(&path) + .await + .map_err(migration_error)? + && let Some(current_path) = + find_current_rollout_path(&self.config.codex_home, &path).await? + { + path = current_path; + } let bytes_before = limiter.bytes_processed; let result = match self .migrate_one_rollout(thread_id, &path, &journal_path, kind, legacy_names, limiter) @@ -1228,6 +1281,30 @@ async fn find_rollout_paths(root: &Path) -> ThreadStoreResult> { Ok(paths) } +async fn find_all_rollout_paths(codex_home: &Path) -> ThreadStoreResult> { + let mut paths = find_rollout_paths(&codex_home.join(codex_rollout::SESSIONS_SUBDIR)).await?; + paths.extend( + find_rollout_paths(&codex_home.join(codex_rollout::ARCHIVED_SESSIONS_SUBDIR)).await?, + ); + Ok(paths) +} + +async fn find_current_rollout_path( + codex_home: &Path, + stale_path: &Path, +) -> ThreadStoreResult> { + let plain_path = codex_rollout::plain_rollout_path(stale_path); + let Some(file_name) = plain_path.file_name() else { + return Ok(None); + }; + Ok(find_all_rollout_paths(codex_home) + .await? + .into_iter() + .find(|candidate| { + codex_rollout::plain_rollout_path(candidate).file_name() == Some(file_name) + })) +} + fn matches_selection(selected: &[ThreadId], actual: Option) -> bool { selected.is_empty() || actual.is_some_and(|thread_id| selected.contains(&thread_id)) } diff --git a/codex-rs/thread-store/src/local/rollout_migration/startup.rs b/codex-rs/thread-store/src/local/rollout_migration/startup.rs index 18c467526c..dbc48ed8c0 100644 --- a/codex-rs/thread-store/src/local/rollout_migration/startup.rs +++ b/codex-rs/thread-store/src/local/rollout_migration/startup.rs @@ -4,13 +4,15 @@ //! rollout files on later launches. When it finds legacy history or a pending recovery marker, it //! invokes the existing full migration path. //! -//! Empty or malformed rollouts are fingerprinted so they do not block the cursor forever. If one -//! changes later, startup retries it instead of trusting the old skip. +//! Rollouts that background migration cannot finish are remembered so they do not hold the cursor +//! back forever. Ordinary failures stay skipped until a manual migration retries them; busy +//! rollouts are retried on later startups because the writer may have gone away. use std::collections::HashSet; -use std::io::ErrorKind; +use std::ffi::OsString; use std::path::Path; use std::path::PathBuf; +use std::time::Duration; use std::time::SystemTime; use chrono::NaiveDateTime; @@ -23,19 +25,26 @@ use codex_state::RolloutMigrationSkippedRollout; use super::LocalThreadStore; use super::RolloutMigrationMode; use super::RolloutMigrationOptions; +use super::RolloutMigrationReport; use super::RolloutMigrationStatus; -use super::find_rollout_paths; +use super::find_all_rollout_paths; use super::migration_error; +use super::publish::migration_journal_path; use super::publish::pending_migration_thread_ids; use super::telemetry::RolloutMigrationTrigger; +use super::thread_id_from_rollout_filename; +use crate::ThreadStoreError; use crate::ThreadStoreResult; const LEGACY_TO_PAGINATED_MIGRATION_ID: &str = "legacy_to_paginated_v1"; const EMPTY_SKIP_REASON: &str = "empty"; +const FAILED_SKIP_REASON: &str = "failed"; const MALFORMED_SESSION_META_SKIP_REASON: &str = "malformed_session_meta"; +const BUSY_SKIP_REASON: &str = "busy"; const CURSOR_LOOKBACK_SECONDS: i64 = 48 * 60 * 60; +const MAINTENANCE_RETRY_DELAY: Duration = Duration::from_secs(1); -#[derive(Clone, Copy, Debug, Eq, PartialEq)] +#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] struct RolloutFingerprint { size_bytes: i64, modified_at_ns: i64, @@ -44,6 +53,7 @@ struct RolloutFingerprint { enum StartupInspection { Paginated, Legacy, + NeedsMigration, Skipped, Unresolved, } @@ -52,8 +62,13 @@ pub(super) async fn migrate_rollouts_on_startup(store: &LocalThreadStore) -> Thr let Some(state_db) = store.state_db.as_ref() else { return Ok(()); }; - let paths = find_all_rollout_paths(store).await?; - let skipped_rollouts = state_db + let paths = find_all_rollout_paths(&store.config.codex_home).await?; + let mut skipped_rollouts = state_db + .list_rollout_migration_skipped_rollouts(LEGACY_TO_PAGINATED_MIGRATION_ID) + .await + .map_err(migration_error)?; + retry_busy_rollouts(store, skipped_rollouts.as_slice(), paths.as_slice()).await?; + skipped_rollouts = state_db .list_rollout_migration_skipped_rollouts(LEGACY_TO_PAGINATED_MIGRATION_ID) .await .map_err(migration_error)?; @@ -63,14 +78,13 @@ pub(super) async fn migrate_rollouts_on_startup(store: &LocalThreadStore) -> Thr { return migrate_all_rollouts(store, paths, skipped_rollouts.as_slice()).await; } - let (unchanged_skips, invalidated_skip) = - revalidate_skipped_rollouts(store, skipped_rollouts.as_slice()).await?; + let skipped_file_names = skipped_rollout_file_names(store, skipped_rollouts.as_slice()); let state = state_db .get_rollout_migration_state(LEGACY_TO_PAGINATED_MIGRATION_ID) .await .map_err(migration_error)?; - if state.is_none() || invalidated_skip { + if state.is_none() { return migrate_all_rollouts(store, paths, skipped_rollouts.as_slice()).await; } @@ -83,8 +97,8 @@ pub(super) async fn migrate_rollouts_on_startup(store: &LocalThreadStore) -> Thr let candidates = paths .iter() .filter(|path| { - let relative_path = relative_rollout_path(store, path); - !unchanged_skips.contains(relative_path.as_str()) + !plain_rollout_file_name(path) + .is_some_and(|file_name| skipped_file_names.contains(&file_name)) && thread_creation_cursor(path).is_none_or(|cursor| { lookback_created_at.is_none_or(|lookback_created_at| { cursor.thread_created_at >= lookback_created_at @@ -100,7 +114,7 @@ pub(super) async fn migrate_rollouts_on_startup(store: &LocalThreadStore) -> Thr for path in candidates { match inspect_rollout_path(store, path).await? { StartupInspection::Paginated | StartupInspection::Skipped => {} - StartupInspection::Legacy => { + StartupInspection::Legacy | StartupInspection::NeedsMigration => { return migrate_all_rollouts(store, paths, skipped_rollouts.as_slice()).await; } StartupInspection::Unresolved => unresolved = true, @@ -118,57 +132,134 @@ async fn migrate_all_rollouts( paths_before_migration: Vec, existing_skips: &[RolloutMigrationSkippedRollout], ) -> ThreadStoreResult<()> { - let report = store - .migrate_rollouts_with_progress_for_trigger( - RolloutMigrationOptions { - mode: RolloutMigrationMode::Apply, - thread_ids: Vec::new(), - max_mib_per_second: None, - }, - |_| {}, - RolloutMigrationTrigger::Startup, - ) - .await?; - let existing_skip_paths = existing_skips + let skipped_file_names = skipped_rollout_file_names(store, existing_skips); + let pending_thread_ids = pending_migration_thread_ids(&store.config.codex_home).await?; + let paths_to_migrate = paths_before_migration .iter() - .map(|skipped_rollout| skipped_rollout.rollout_path.as_str()) - .collect::>(); - let mut terminal = true; - let mut reported_paths = HashSet::new(); + .filter(|path| { + !plain_rollout_file_name(path) + .is_some_and(|file_name| skipped_file_names.contains(&file_name)) + || thread_id_from_rollout_filename(path) + .is_some_and(|thread_id| pending_thread_ids.contains(&thread_id)) + }) + .cloned() + .collect(); + let report = run_startup_migration(store, paths_to_migrate).await?; for outcome in &report.outcomes { - let relative_path = relative_rollout_path(store, &outcome.rollout_path); - reported_paths.insert(relative_path.clone()); - match outcome.status { - RolloutMigrationStatus::Migrated | RolloutMigrationStatus::AlreadyPaginated => { - if existing_skip_paths.contains(relative_path.as_str()) { - remove_skip(store, relative_path.as_str()).await?; - } - } - RolloutMigrationStatus::SkippedEmpty | RolloutMigrationStatus::Failed => { - if !matches!( - inspect_rollout_path(store, &outcome.rollout_path).await?, - StartupInspection::Skipped - ) { - terminal = false; - } - } - RolloutMigrationStatus::Eligible | RolloutMigrationStatus::SkippedBusy => { - terminal = false - } - } - } - if !terminal { - return Ok(()); - } - for skipped_rollout in existing_skips { - if !reported_paths.contains(skipped_rollout.rollout_path.as_str()) { - remove_skip(store, skipped_rollout.rollout_path.as_str()).await?; - } + update_skip_after_outcome(store, outcome).await?; } // Only mark the pre-migration snapshot; newer rollouts wait for the next startup check. advance_last_checked_thread(store, paths_before_migration.as_slice()).await } +async fn retry_busy_rollouts( + store: &LocalThreadStore, + skipped_rollouts: &[RolloutMigrationSkippedRollout], + discovered_paths: &[PathBuf], +) -> ThreadStoreResult<()> { + let mut paths = Vec::new(); + let mut moved_skip_paths = Vec::new(); + for skipped_rollout in skipped_rollouts + .iter() + .filter(|skipped_rollout| skipped_rollout.skip_reason == BUSY_SKIP_REASON) + { + let stored_path = store.config.codex_home.join(&skipped_rollout.rollout_path); + let path = if tokio::fs::try_exists(&stored_path) + .await + .map_err(migration_error)? + { + Some(stored_path.clone()) + } else { + // Archive/unarchive moves one rollout between roots, while compression swaps between + // its plain and compressed filenames. Match the plain basename across both. + let file_name = plain_rollout_file_name(&stored_path); + discovered_paths + .iter() + .find(|path| plain_rollout_file_name(path) == file_name) + .cloned() + }; + let Some(path) = path else { + continue; + }; + if path != stored_path { + moved_skip_paths.push(skipped_rollout.rollout_path.as_str()); + } + paths.push(path); + } + if paths.is_empty() { + return Ok(()); + } + let report = run_startup_migration(store, paths).await?; + for moved_skip_path in moved_skip_paths { + remove_skip(store, moved_skip_path).await?; + } + for outcome in &report.outcomes { + update_skip_after_outcome(store, outcome).await?; + } + Ok(()) +} + +async fn run_startup_migration( + store: &LocalThreadStore, + paths: Vec, +) -> ThreadStoreResult { + loop { + let Some(maintenance_guard) = + codex_rollout::try_acquire_rollout_maintenance_lock(&store.config.codex_home) + .map_err(migration_error)? + else { + tokio::time::sleep(MAINTENANCE_RETRY_DELAY).await; + continue; + }; + // Avoid counting expected compression contention as a failed migration run. The migration + // path takes the real lock below, so retry if another maintainer wins this small gap. + drop(maintenance_guard); + match store + .migrate_rollouts_with_progress_for_trigger( + RolloutMigrationOptions { + mode: RolloutMigrationMode::Apply, + thread_ids: Vec::new(), + max_mib_per_second: None, + }, + |_| {}, + RolloutMigrationTrigger::Startup, + super::RolloutMigrationPaths::Known(paths.clone()), + ) + .await + { + Err(ThreadStoreError::Conflict { .. }) => continue, + result => return result, + } + } +} + +async fn update_skip_after_outcome( + store: &LocalThreadStore, + outcome: &super::RolloutMigrationOutcome, +) -> ThreadStoreResult<()> { + let relative_path = relative_rollout_path(store, &outcome.rollout_path); + match outcome.status { + RolloutMigrationStatus::Migrated | RolloutMigrationStatus::AlreadyPaginated => { + remove_skip(store, relative_path.as_str()).await + } + RolloutMigrationStatus::SkippedEmpty => { + record_current_skip(store, &outcome.rollout_path, EMPTY_SKIP_REASON).await + } + RolloutMigrationStatus::SkippedBusy => { + record_current_skip(store, &outcome.rollout_path, BUSY_SKIP_REASON).await + } + RolloutMigrationStatus::Failed => { + if outcome.thread_id.is_some_and(|thread_id| { + migration_journal_path(&store.config.codex_home, thread_id).exists() + }) { + return Ok(()); + } + record_current_skip(store, &outcome.rollout_path, FAILED_SKIP_REASON).await + } + RolloutMigrationStatus::Eligible => Ok(()), + } +} + async fn inspect_rollout_path( store: &LocalThreadStore, path: &Path, @@ -179,33 +270,45 @@ async fn inspect_rollout_path( Ok(StartupInspection::Legacy) } Ok(_) => Ok(StartupInspection::Paginated), - Err(error) => { + Err(_) => { let after = rollout_fingerprint(path).await?; - if before != after || !matches!(error.kind(), ErrorKind::Other | ErrorKind::InvalidData) - { + if before != after { return Ok(StartupInspection::Unresolved); } - record_skip(store, path, before).await?; + // The migration path re-reads empty files under the writer lock before deciding + // whether they are terminally empty or just waiting for SessionMeta. + if before.size_bytes == 0 { + return Ok(StartupInspection::NeedsMigration); + } + record_skip(store, path, before, MALFORMED_SESSION_META_SKIP_REASON).await?; Ok(StartupInspection::Skipped) } } } +async fn record_current_skip( + store: &LocalThreadStore, + path: &Path, + skip_reason: &str, +) -> ThreadStoreResult<()> { + // These fields remain in the generic schema, but background skips are permanent now. Keep + // recording the best available fingerprint for humans inspecting SQLite. + let fingerprint = rollout_fingerprint(path).await.unwrap_or_default(); + record_skip(store, path, fingerprint, skip_reason).await +} + async fn record_skip( store: &LocalThreadStore, path: &Path, fingerprint: RolloutFingerprint, + skip_reason: &str, ) -> ThreadStoreResult<()> { let state_db = startup_state_db(store)?; let skipped_rollout = RolloutMigrationSkippedRollout { rollout_path: relative_rollout_path(store, path), rollout_size_bytes: fingerprint.size_bytes, rollout_modified_at_ns: fingerprint.modified_at_ns, - skip_reason: if fingerprint.size_bytes == 0 { - EMPTY_SKIP_REASON.to_string() - } else { - MALFORMED_SESSION_META_SKIP_REASON.to_string() - }, + skip_reason: skip_reason.to_string(), }; state_db .record_rollout_migration_skip(LEGACY_TO_PAGINATED_MIGRATION_ID, &skipped_rollout) @@ -220,32 +323,6 @@ async fn remove_skip(store: &LocalThreadStore, rollout_path: &str) -> ThreadStor .map_err(migration_error) } -async fn revalidate_skipped_rollouts( - store: &LocalThreadStore, - skipped_rollouts: &[RolloutMigrationSkippedRollout], -) -> ThreadStoreResult<(HashSet, bool)> { - let mut unchanged_skips = HashSet::new(); - let mut invalidated_skip = false; - for skipped_rollout in skipped_rollouts { - let path = store.config.codex_home.join(&skipped_rollout.rollout_path); - let fingerprint = match rollout_fingerprint(&path).await { - Ok(fingerprint) => fingerprint, - Err(_) => { - invalidated_skip = true; - continue; - } - }; - if fingerprint.size_bytes == skipped_rollout.rollout_size_bytes - && fingerprint.modified_at_ns == skipped_rollout.rollout_modified_at_ns - { - unchanged_skips.insert(skipped_rollout.rollout_path.clone()); - } else { - invalidated_skip = true; - } - } - Ok((unchanged_skips, invalidated_skip)) -} - async fn advance_last_checked_thread( store: &LocalThreadStore, paths: &[PathBuf], @@ -270,21 +347,6 @@ fn startup_state_db(store: &LocalThreadStore) -> ThreadStoreResult<&StateDbHandl .ok_or_else(|| migration_error("startup migration requires state db")) } -async fn find_all_rollout_paths(store: &LocalThreadStore) -> ThreadStoreResult> { - let mut paths = - find_rollout_paths(&store.config.codex_home.join(codex_rollout::SESSIONS_SUBDIR)).await?; - paths.extend( - find_rollout_paths( - &store - .config - .codex_home - .join(codex_rollout::ARCHIVED_SESSIONS_SUBDIR), - ) - .await?, - ); - Ok(paths) -} - fn thread_creation_cursor(path: &Path) -> Option { let name = path.file_name()?.to_str()?; let stem = name @@ -311,6 +373,24 @@ fn relative_rollout_path(store: &LocalThreadStore, path: &Path) -> String { .replace('\\', "/") } +fn skipped_rollout_file_names( + store: &LocalThreadStore, + skipped_rollouts: &[RolloutMigrationSkippedRollout], +) -> HashSet { + skipped_rollouts + .iter() + .filter_map(|skipped_rollout| { + plain_rollout_file_name(&store.config.codex_home.join(&skipped_rollout.rollout_path)) + }) + .collect() +} + +fn plain_rollout_file_name(path: &Path) -> Option { + codex_rollout::plain_rollout_path(path) + .file_name() + .map(std::ffi::OsStr::to_os_string) +} + async fn rollout_fingerprint(path: &Path) -> ThreadStoreResult { let metadata = tokio::fs::metadata(path).await.map_err(migration_error)?; let size_bytes = i64::try_from(metadata.len()).map_err(migration_error)?; diff --git a/codex-rs/thread-store/src/local/rollout_migration/startup_tests.rs b/codex-rs/thread-store/src/local/rollout_migration/startup_tests.rs index 041dd1df45..7a4512bfcd 100644 --- a/codex-rs/thread-store/src/local/rollout_migration/startup_tests.rs +++ b/codex-rs/thread-store/src/local/rollout_migration/startup_tests.rs @@ -199,20 +199,33 @@ async fn checks_rollouts_within_the_cursor_lookback() { } #[tokio::test] -async fn recovers_pending_migrations_behind_the_checked_thread() { +async fn recovers_pending_migrations_after_retrying_busy_rollouts() { let home = TempDir::new().expect("create Codex home"); - let thread_id = ThreadId::new(); - let path = write_rollout(home.path(), thread_id, ThreadHistoryMode::Legacy); + let pending_thread_id = ThreadId::new(); + write_rollout(home.path(), pending_thread_id, ThreadHistoryMode::Legacy); + let busy_thread_id = ThreadId::new(); + let busy_path = move_to_timestamp( + home.path(), + write_rollout(home.path(), busy_thread_id, ThreadHistoryMode::Legacy), + "2025/01/04", + "2025-01-04T12-00-00", + ); let store = indexed_store(home.path()).await; + let state_db = store.state_db().await.expect("state db"); + let writer_guard = store + .writer_lock_coordinator + .acquire(busy_thread_id) + .expect("hold cross-process writer lock"); store .migrate_rollouts_on_startup() .await - .expect("migrate and advance startup cursor"); - thread_history::delete_thread(&store, thread_id) + .expect("migrate and record busy rollout"); + drop(writer_guard); + thread_history::delete_thread(&store, pending_thread_id) .await .expect("simulate missing projection"); - let journal_path = migration_journal_path(home.path(), thread_id); + let journal_path = migration_journal_path(home.path(), pending_thread_id); write_migration_journal(&journal_path) .await .expect("simulate pending migration marker"); @@ -220,23 +233,30 @@ async fn recovers_pending_migrations_behind_the_checked_thread() { store .migrate_rollouts_on_startup() .await - .expect("recover pending migration behind cursor"); + .expect("retry busy rollout before recovery"); - assert!(!journal_path.exists()); - assert!( - thread_history::projection_state(&store, thread_id) - .await - .expect("read repaired projection") - .is_some() - ); assert_eq!( - codex_rollout::read_session_meta_line(&path) + codex_rollout::read_session_meta_line(&busy_path) .await .expect("read migrated metadata") .meta .history_mode, ThreadHistoryMode::Paginated ); + assert!(!journal_path.exists()); + assert!( + thread_history::projection_state(&store, pending_thread_id) + .await + .expect("read repaired projection") + .is_some() + ); + assert!( + state_db + .list_rollout_migration_skipped_rollouts(super::LEGACY_TO_PAGINATED_MIGRATION_ID) + .await + .expect("read skipped rollouts") + .is_empty() + ); } #[tokio::test] @@ -274,46 +294,180 @@ async fn waits_for_a_live_writer_before_migrating() { } #[tokio::test] -async fn rechecks_changed_empty_rollouts() { +async fn waits_for_rollout_maintenance_before_migrating() { let home = TempDir::new().expect("create Codex home"); - write_rollout(home.path(), ThreadId::new(), ThreadHistoryMode::Legacy); - let empty_thread_id = ThreadId::new(); - let empty_path = move_to_timestamp( + let thread_id = ThreadId::new(); + let path = write_rollout(home.path(), thread_id, ThreadHistoryMode::Legacy); + let store = indexed_store(home.path()).await; + let maintenance_guard = codex_rollout::try_acquire_rollout_maintenance_lock(home.path()) + .expect("acquire rollout maintenance lock") + .expect("claim rollout maintenance lock"); + let migration_store = store.clone(); + let mut migration = tokio::spawn(async move { + migration_store + .migrate_rollouts_on_startup() + .await + .expect("migrate startup rollouts"); + }); + + assert!( + tokio::time::timeout(Duration::from_millis(50), &mut migration) + .await + .is_err(), + "migration should wait for rollout maintenance" + ); + drop(maintenance_guard); + tokio::time::timeout(Duration::from_secs(2), migration) + .await + .expect("migration should retry rollout maintenance") + .expect("join startup migration"); + + assert_eq!( + codex_rollout::read_session_meta_line(&path) + .await + .expect("read migrated metadata") + .meta + .history_mode, + ThreadHistoryMode::Paginated + ); +} + +#[tokio::test] +async fn permanently_skips_failed_rollouts_without_blocking_the_cursor() { + let home = TempDir::new().expect("create Codex home"); + let failed_thread_id = ThreadId::new(); + let failed_path = write_rollout(home.path(), failed_thread_id, ThreadHistoryMode::Legacy); + let store = indexed_store(home.path()).await; + let state_db = store.state_db().await.expect("state db"); + let metadata = state_db + .get_thread(failed_thread_id) + .await + .expect("read thread metadata") + .expect("thread metadata"); + state_db + .delete_thread(failed_thread_id) + .await + .expect("remove thread metadata"); + + store + .migrate_rollouts_on_startup() + .await + .expect("record failed rollout"); + assert_eq!( + state_db + .get_rollout_migration_state(super::LEGACY_TO_PAGINATED_MIGRATION_ID) + .await + .expect("read migration state") + .expect("migration state") + .last_checked_thread + .expect("checked thread") + .thread_id, + failed_thread_id.to_string() + ); + assert_eq!( + state_db + .list_rollout_migration_skipped_rollouts(super::LEGACY_TO_PAGINATED_MIGRATION_ID) + .await + .expect("read skipped rollouts") + .into_iter() + .map(|skipped_rollout| skipped_rollout.skip_reason) + .collect::>(), + vec![super::FAILED_SKIP_REASON.to_string()] + ); + + state_db + .insert_thread_if_absent(&metadata) + .await + .expect("restore thread metadata"); + let archived_directory = home.path().join(codex_rollout::ARCHIVED_SESSIONS_SUBDIR); + fs::create_dir_all(&archived_directory).expect("create archived directory"); + let archived_path = archived_directory.join(failed_path.file_name().expect("rollout filename")); + fs::rename(&failed_path, &archived_path).expect("archive failed rollout"); + + store + .migrate_rollouts_on_startup() + .await + .expect("skip failed rollout again"); + + assert_eq!( + codex_rollout::read_session_meta_line(&archived_path) + .await + .expect("read failed rollout metadata") + .meta + .history_mode, + ThreadHistoryMode::Legacy + ); +} + +#[tokio::test] +async fn retries_busy_rollouts_after_archive_and_compression_move() { + let home = TempDir::new().expect("create Codex home"); + let thread_id = ThreadId::new(); + let path = move_to_timestamp( home.path(), - write_rollout(home.path(), empty_thread_id, ThreadHistoryMode::Legacy), + write_rollout(home.path(), thread_id, ThreadHistoryMode::Legacy), + "2025/01/01", + "2025-01-01T12-00-00", + ); + let newest_thread_id = ThreadId::new(); + move_to_timestamp( + home.path(), + write_rollout(home.path(), newest_thread_id, ThreadHistoryMode::Paginated), "2025/01/04", "2025-01-04T12-00-00", ); - let restored_contents = fs::read(&empty_path).expect("read rollout before emptying"); - fs::write(&empty_path, []).expect("empty rollout"); let store = indexed_store(home.path()).await; + let state_db = store.state_db().await.expect("state db"); + let writer_guard = store + .writer_lock_coordinator + .acquire(thread_id) + .expect("hold cross-process writer lock"); store .migrate_rollouts_on_startup() .await - .expect("record empty rollout"); - fs::write(&empty_path, restored_contents).expect("restore rollout"); - let (items, _, _) = RolloutRecorder::load_rollout_items(&empty_path) - .await - .expect("load restored rollout"); - let metadata = codex_rollout::builder_from_items(items.as_slice(), &empty_path) - .expect("build restored metadata") - .build("test-provider"); - store - .state_db() - .await - .expect("state db") - .upsert_thread(&metadata) - .await - .expect("seed restored metadata"); + .expect("record busy rollout"); + assert_eq!( + state_db + .get_rollout_migration_state(super::LEGACY_TO_PAGINATED_MIGRATION_ID) + .await + .expect("read migration state") + .expect("migration state") + .last_checked_thread + .expect("checked thread") + .thread_id, + newest_thread_id.to_string() + ); + assert_eq!( + state_db + .list_rollout_migration_skipped_rollouts(super::LEGACY_TO_PAGINATED_MIGRATION_ID) + .await + .expect("read skipped rollouts") + .into_iter() + .map(|skipped_rollout| skipped_rollout.skip_reason) + .collect::>(), + vec![super::BUSY_SKIP_REASON.to_string()] + ); + drop(writer_guard); + let archived_directory = home.path().join(codex_rollout::ARCHIVED_SESSIONS_SUBDIR); + fs::create_dir_all(&archived_directory).expect("create archived directory"); + let archived_path = archived_directory.join(path.file_name().expect("rollout filename")); + fs::rename(&path, &archived_path).expect("archive busy rollout"); + let compressed_path = archived_path.with_extension("jsonl.zst"); + let mut input = fs::File::open(&archived_path).expect("open archived rollout"); + let output = fs::File::create(&compressed_path).expect("create compressed rollout"); + let mut encoder = zstd::stream::write::Encoder::new(output, 0).expect("create encoder"); + std::io::copy(&mut input, &mut encoder).expect("compress archived rollout"); + encoder.finish().expect("finish compressed rollout"); + fs::remove_file(&archived_path).expect("remove plain archived rollout"); store .migrate_rollouts_on_startup() .await - .expect("migrate changed rollout"); + .expect("retry no-longer-busy rollout"); assert_eq!( - codex_rollout::read_session_meta_line(&empty_path) + codex_rollout::read_session_meta_line(&compressed_path) .await .expect("read migrated metadata") .meta @@ -321,13 +475,77 @@ async fn rechecks_changed_empty_rollouts() { ThreadHistoryMode::Paginated ); assert!( - store - .state_db() - .await - .expect("state db") + state_db .list_rollout_migration_skipped_rollouts(super::LEGACY_TO_PAGINATED_MIGRATION_ID) .await .expect("read skipped rollouts") .is_empty() ); } + +#[tokio::test] +async fn treats_writer_owned_empty_rollouts_as_busy() { + let home = TempDir::new().expect("create Codex home"); + write_rollout(home.path(), ThreadId::new(), ThreadHistoryMode::Paginated); + let store = indexed_store(home.path()).await; + store + .migrate_rollouts_on_startup() + .await + .expect("seed startup cursor"); + + let thread_id = ThreadId::new(); + let path = move_to_timestamp( + home.path(), + write_rollout(home.path(), thread_id, ThreadHistoryMode::Legacy), + "2025/01/04", + "2025-01-04T12-00-00", + ); + let (items, _, _) = RolloutRecorder::load_rollout_items(&path) + .await + .expect("load rollout items"); + let metadata = codex_rollout::builder_from_items(items.as_slice(), &path) + .expect("build thread metadata") + .build("test-provider"); + let contents = fs::read(&path).expect("read rollout before emptying"); + fs::write(&path, []).expect("empty rollout"); + let state_db = store.state_db().await.expect("state db"); + state_db + .upsert_thread(&metadata) + .await + .expect("seed thread metadata"); + let writer_guard = store + .writer_lock_coordinator + .acquire(thread_id) + .expect("hold cross-process writer lock"); + + store + .migrate_rollouts_on_startup() + .await + .expect("record empty rollout as busy"); + assert_eq!( + state_db + .list_rollout_migration_skipped_rollouts(super::LEGACY_TO_PAGINATED_MIGRATION_ID) + .await + .expect("read skipped rollouts") + .into_iter() + .map(|skipped_rollout| skipped_rollout.skip_reason) + .collect::>(), + vec![super::BUSY_SKIP_REASON.to_string()] + ); + + fs::write(&path, contents).expect("restore rollout"); + drop(writer_guard); + store + .migrate_rollouts_on_startup() + .await + .expect("retry no-longer-empty rollout"); + + assert_eq!( + codex_rollout::read_session_meta_line(&path) + .await + .expect("read migrated metadata") + .meta + .history_mode, + ThreadHistoryMode::Paginated + ); +} diff --git a/codex-rs/thread-store/src/local/rollout_migration_tests.rs b/codex-rs/thread-store/src/local/rollout_migration_tests.rs index 96ee8cb0c2..9e7a380ac1 100644 --- a/codex-rs/thread-store/src/local/rollout_migration_tests.rs +++ b/codex-rs/thread-store/src/local/rollout_migration_tests.rs @@ -48,11 +48,13 @@ use super::LocalThreadStore; use super::RolloutMigrationFailureReason; use super::RolloutMigrationMode; use super::RolloutMigrationOptions; +use super::RolloutMigrationPaths; use super::RolloutMigrationProgress; use super::RolloutMigrationStatus; #[cfg(unix)] use super::decompress_rollout_to_path; use super::migration_journal_path; +use super::telemetry::RolloutMigrationTrigger; use super::thread_history; use super::write_migration_journal; use crate::ItemSortKey; @@ -1786,6 +1788,37 @@ async fn migration_migrates_archived_rollouts_without_unarchiving_them() { assert_eq!(turns.turns[0].items.len(), 2); } +#[tokio::test] +async fn migration_retries_a_rollout_moved_after_path_discovery() { + let home = TempDir::new().expect("create Codex home"); + let thread_id = ThreadId::new(); + let active_path = write_rollout( + home.path(), + thread_id, + SessionSource::Cli, + vec![user_message("question"), agent_message("answer")], + ); + let store = indexed_store(home.path()).await; + let archived_path = move_to_archived(home.path(), active_path.clone()); + + let report = store + .migrate_rollouts_with_progress_for_trigger( + apply_options(), + |_| {}, + RolloutMigrationTrigger::Startup, + RolloutMigrationPaths::Known(vec![active_path]), + ) + .await + .expect("migrate moved rollout"); + + assert_eq!(report.outcomes[0].status, RolloutMigrationStatus::Migrated); + assert!(matches!( + &read_rollout(&archived_path)[0].item, + RolloutItem::SessionMeta(metadata) + if metadata.meta.history_mode == ThreadHistoryMode::Paginated + )); +} + #[tokio::test] async fn migration_preserves_legacy_displayed_thread_names() { let home = TempDir::new().expect("create Codex home");