openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-09-08 15:50:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

Harden startup rollout migration against concurrent updates (#40499)

Browse Source 
## 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
This commit is contained in:
Owen Lin
2026-08-24 23:55:51 +00:00
committed by copyberry
parent 4ef1d4b89b
commit 465eafacbc
4 changed files with 599 additions and 191 deletions
Download Patch File Download Diff File Expand all files Collapse all files

159
codex-rs/thread-store/src/local/rollout_migration.rs
View File

@@ -81,6 +81,11 @@ enum CanonicalizationAttempt {
NeedsRollbackPlan,
}
enum RolloutMigrationPaths {
Discover,
Known(Vec<PathBuf>),
}
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<RolloutMigrationReport> {
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<RolloutMigrationReport> {
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<ThreadId, String>,
limiter: &mut RolloutMigrationRateLimiter,
) -> ThreadStoreResult<Option<RolloutMigrationOutcome>> {
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<Vec<PathBuf>> {
Ok(paths)
}
async fn find_all_rollout_paths(codex_home: &Path) -> ThreadStoreResult<Vec<PathBuf>> {
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<Option<PathBuf>> {
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<ThreadId>) -> bool {
selected.is_empty() || actual.is_some_and(|thread_id| selected.contains(&thread_id))
}

294
codex-rs/thread-store/src/local/rollout_migration/startup.rs
View File

@@ -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<PathBuf>,
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::<HashSet<_>>();
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<PathBuf>,
) -> ThreadStoreResult<RolloutMigrationReport> {
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<String>, 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<Vec<PathBuf>> {
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<RolloutMigrationCursor> {
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<OsString> {
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<OsString> {
codex_rollout::plain_rollout_path(path)
.file_name()
.map(std::ffi::OsStr::to_os_string)
}
async fn rollout_fingerprint(path: &Path) -> ThreadStoreResult<RolloutFingerprint> {
let metadata = tokio::fs::metadata(path).await.map_err(migration_error)?;
let size_bytes = i64::try_from(metadata.len()).map_err(migration_error)?;

304
codex-rs/thread-store/src/local/rollout_migration/startup_tests.rs
View File

@@ -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<_>>(),
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<_>>(),
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<_>>(),
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
);
}

33
codex-rs/thread-store/src/local/rollout_migration_tests.rs
View File

@@ -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");