openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-19 17:28:46 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add rollout migration tooling and background migration (#37348)

Browse Source 
## What changed

- Add `codex migrate-rollouts` with dry-run inspection by default, explicit `--apply`, thread filtering, optional I/O throttling, progress output, and JSON or verbose reports.
- Add the disabled-by-default `background_paginated_rollout_migration` feature to migrate legacy local sessions at startup before rollout compression begins.
- Persist a migration cursor and skipped-file fingerprints so later startups avoid full rescans while retrying changed files and recovering pending migrations.
- Coordinate migration with live writers and emit metrics for manual and startup runs.

## Testing

- Cover startup cursor advancement and lookback, pending migration recovery, live-writer coordination, changed empty rollouts, and progress reporting.

GitOrigin-RevId: 276ac506c50ebec5140fd319faca1bb998172061
This commit is contained in:
Owen Lin
2026-08-07 01:26:42 +00:00
committed by copyberry
parent 69b6152c17
commit 4bb7ee3472
21 changed files with 1569 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/thread-store/src/lib.rs
+1
View File
@@ -27,6 +27,7 @@ pub use local::LocalThreadStoreConfig;
pub use local::RolloutMigrationMode;
pub use local::RolloutMigrationOptions;
pub use local::RolloutMigrationOutcome;
pub use local::RolloutMigrationProgress;
pub use local::RolloutMigrationReport;
pub use local::RolloutMigrationStatus;
pub use queue_store::LocalQueueStore;
codex-rs/thread-store/src/local/mod.rs
+1
View File
@@ -82,6 +82,7 @@ use crate::local::writer_lock::WriterLockGuard;
pub use rollout_migration::RolloutMigrationMode;
pub use rollout_migration::RolloutMigrationOptions;
pub use rollout_migration::RolloutMigrationOutcome;
pub use rollout_migration::RolloutMigrationProgress;
pub use rollout_migration::RolloutMigrationReport;
pub use rollout_migration::RolloutMigrationStatus;
codex-rs/thread-store/src/local/rollout_migration.rs
+93 -14
View File
@@ -45,7 +45,9 @@ mod publish;
mod rollback;
mod rollback_plan;
mod rollback_replay;
mod startup;
mod subagent;
mod telemetry;
use canonicalizer::LegacyRolloutCanonicalizer;
use publish::compress_rollout_to_path;
@@ -62,6 +64,8 @@ use publish::sync_parent_directory;
use publish::write_migration_journal;
use rollback_plan::RollbackPlan;
use rollback_plan::RollbackPlanner;
use telemetry::RolloutMigrationTelemetry;
use telemetry::RolloutMigrationTrigger;
const PROJECTION_BATCH_BYTES: u64 = 256 * 1024;
const MAX_ROLLOUT_LINE_BYTES: usize = 16 * 1024 * 1024;
@@ -98,7 +102,9 @@ pub enum RolloutMigrationMode {
pub struct RolloutMigrationOptions {
pub mode: RolloutMigrationMode,
pub thread_ids: Vec<ThreadId>,
pub max_mib_per_second: u64,
/// Optional aggregate rollout I/O limit. Without one, migration runs as fast as local I/O
/// allows while still yielding between projection-sized batches.
pub max_mib_per_second: Option<u64>,
}
impl Default for RolloutMigrationOptions {
@@ -106,7 +112,7 @@ impl Default for RolloutMigrationOptions {
Self {
mode: RolloutMigrationMode::DryRun,
thread_ids: Vec::new(),
max_mib_per_second: 8,
max_mib_per_second: None,
}
}
}
@@ -139,10 +145,19 @@ pub struct RolloutMigrationReport {
pub outcomes: Vec<RolloutMigrationOutcome>,
}
/// Incremental progress emitted while rollout migration scans discovered paths.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct RolloutMigrationProgress {
pub processed_paths: usize,
pub total_paths: usize,
/// None means the scanned path was excluded by the requested thread filter.
pub outcome_status: Option<RolloutMigrationStatus>,
}
struct RolloutMigrationRateLimiter {
started_at: Instant,
bytes_processed: u64,
bytes_per_second: u64,
bytes_per_second: Option<u64>,
bytes_since_yield: u64,
}
@@ -158,13 +173,17 @@ enum RolloutMigrationKind {
}
impl RolloutMigrationRateLimiter {
fn new(max_mib_per_second: u64) -> ThreadStoreResult<Self> {
fn new(max_mib_per_second: Option<u64>) -> ThreadStoreResult<Self> {
let bytes_per_second = max_mib_per_second
.checked_mul(1024 * 1024)
.filter(|rate| *rate > 0)
.ok_or_else(|| ThreadStoreError::InvalidRequest {
message: "--max-mib-per-second must be a positive supported integer".to_string(),
})?;
.map(|rate| {
rate.checked_mul(1024 * 1024)
.filter(|rate| *rate > 0)
.ok_or_else(|| ThreadStoreError::InvalidRequest {
message: "--max-mib-per-second must be a positive supported integer"
.to_string(),
})
})
.transpose()?;
Ok(Self {
started_at: Instant::now(),
bytes_processed: 0,
@@ -180,8 +199,12 @@ impl RolloutMigrationRateLimiter {
return;
}
self.bytes_since_yield = 0;
let Some(bytes_per_second) = self.bytes_per_second else {
tokio::task::yield_now().await;
return;
};
let expected =
Duration::from_secs_f64(self.bytes_processed as f64 / self.bytes_per_second as f64);
Duration::from_secs_f64(self.bytes_processed as f64 / bytes_per_second as f64);
if let Some(remaining) = expected.checked_sub(self.started_at.elapsed()) {
tokio::time::sleep(remaining).await;
} else {
@@ -191,10 +214,57 @@ impl RolloutMigrationRateLimiter {
}
impl LocalThreadStore {
/// Check whether startup needs to migrate legacy rollouts, then migrate in the background
/// when it does.
pub async fn migrate_rollouts_on_startup(&self) -> ThreadStoreResult<()> {
startup::migrate_rollouts_on_startup(self).await
}
/// Inspect or migrate eligible legacy rollout files beneath active and archived sessions.
pub async fn migrate_rollouts(
&self,
options: RolloutMigrationOptions,
) -> ThreadStoreResult<RolloutMigrationReport> {
self.migrate_rollouts_with_progress_for_trigger(
options,
|_| {},
RolloutMigrationTrigger::Manual,
)
.await
}
/// Inspect or migrate rollouts while reporting each discovered path after it is processed.
pub async fn migrate_rollouts_with_progress(
&self,
options: RolloutMigrationOptions,
on_progress: impl FnMut(RolloutMigrationProgress),
) -> ThreadStoreResult<RolloutMigrationReport> {
self.migrate_rollouts_with_progress_for_trigger(
options,
on_progress,
RolloutMigrationTrigger::Manual,
)
.await
}
async fn migrate_rollouts_with_progress_for_trigger(
&self,
options: RolloutMigrationOptions,
mut on_progress: impl FnMut(RolloutMigrationProgress),
trigger: RolloutMigrationTrigger,
) -> ThreadStoreResult<RolloutMigrationReport> {
let telemetry = RolloutMigrationTelemetry::new(trigger, &options);
let result = self
.migrate_rollouts_with_progress_inner(options, &mut on_progress)
.await;
telemetry.finish(&result);
result
}
async fn migrate_rollouts_with_progress_inner(
&self,
options: RolloutMigrationOptions,
on_progress: &mut impl FnMut(RolloutMigrationProgress),
) -> ThreadStoreResult<RolloutMigrationReport> {
let mut limiter = RolloutMigrationRateLimiter::new(options.max_mib_per_second)?;
let _maintenance_guard = match options.mode {
@@ -227,15 +297,22 @@ impl LocalThreadStore {
.is_some_and(|thread_id| pending_thread_ids.contains(&thread_id))
});
}
let total_paths = paths.len();
let mut report = RolloutMigrationReport::default();
for path in paths {
if let Some(outcome) = self
for (index, path) in paths.into_iter().enumerate() {
let outcome = self
.migrate_rollout_path(path, &options, &mut limiter)
.await?
{
.await?;
let outcome_status = outcome.as_ref().map(|outcome| outcome.status);
if let Some(outcome) = outcome {
report.outcomes.push(outcome);
}
on_progress(RolloutMigrationProgress {
processed_paths: index + 1,
total_paths,
outcome_status,
});
}
Ok(report)
@@ -301,6 +378,7 @@ impl LocalThreadStore {
if metadata.meta.history_mode == ThreadHistoryMode::Paginated {
let bytes_before = limiter.bytes_processed;
let result = if pending_published_migration {
let _live_writer_guard = self.live_writer_locks.lock(thread_id).await;
match self
.recover_published_migration(thread_id, &path, &journal_path, limiter)
.await
@@ -341,6 +419,7 @@ impl LocalThreadStore {
)));
}
let _live_writer_guard = self.live_writer_locks.lock(thread_id).await;
let _writer_guard = match self.writer_lock_coordinator.acquire(thread_id) {
Ok(guard) => guard,
Err(ThreadStoreError::Conflict { message }) => {
codex-rs/thread-store/src/local/rollout_migration/startup.rs
+332
View File
@@ -0,0 +1,332 @@
//! Decides whether startup needs to invoke legacy -> paginated rollout migration.
//!
//! It keeps startup cheap by storing a creation-ordered cursor in SQLite and checking only newer
//! 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.
use std::collections::HashSet;
use std::io::ErrorKind;
use std::path::Path;
use std::path::PathBuf;
use std::time::SystemTime;
use chrono::NaiveDateTime;
use codex_protocol::ThreadId;
use codex_protocol::protocol::ThreadHistoryMode;
use codex_rollout::StateDbHandle;
use codex_state::RolloutMigrationCursor;
use codex_state::RolloutMigrationSkippedRollout;
use super::LocalThreadStore;
use super::RolloutMigrationMode;
use super::RolloutMigrationOptions;
use super::RolloutMigrationStatus;
use super::find_rollout_paths;
use super::migration_error;
use super::publish::pending_migration_thread_ids;
use super::telemetry::RolloutMigrationTrigger;
use crate::ThreadStoreResult;
const LEGACY_TO_PAGINATED_MIGRATION_ID: &str = "legacy_to_paginated_v1";
const EMPTY_SKIP_REASON: &str = "empty";
const MALFORMED_SESSION_META_SKIP_REASON: &str = "malformed_session_meta";
const CURSOR_LOOKBACK_SECONDS: i64 = 48 * 60 * 60;
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
struct RolloutFingerprint {
size_bytes: i64,
modified_at_ns: i64,
}
enum StartupInspection {
Paginated,
Legacy,
Skipped,
Unresolved,
}
pub(super) async fn migrate_rollouts_on_startup(store: &LocalThreadStore) -> ThreadStoreResult<()> {
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
.list_rollout_migration_skipped_rollouts(LEGACY_TO_PAGINATED_MIGRATION_ID)
.await
.map_err(migration_error)?;
if !pending_migration_thread_ids(&store.config.codex_home)
.await?
.is_empty()
{
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 state = state_db
.get_rollout_migration_state(LEGACY_TO_PAGINATED_MIGRATION_ID)
.await
.map_err(migration_error)?;
if state.is_none() || invalidated_skip {
return migrate_all_rollouts(store, paths, skipped_rollouts.as_slice()).await;
}
let last_checked_thread = state.and_then(|state| state.last_checked_thread);
let lookback_created_at = last_checked_thread.as_ref().map(|cursor| {
cursor
.thread_created_at
.saturating_sub(CURSOR_LOOKBACK_SECONDS)
});
let candidates = paths
.iter()
.filter(|path| {
let relative_path = relative_rollout_path(store, path);
!unchanged_skips.contains(relative_path.as_str())
&& thread_creation_cursor(path).is_none_or(|cursor| {
lookback_created_at.is_none_or(|lookback_created_at| {
cursor.thread_created_at >= lookback_created_at
})
})
})
.collect::<Vec<_>>();
if candidates.is_empty() {
return Ok(());
}
let mut unresolved = false;
for path in candidates {
match inspect_rollout_path(store, path).await? {
StartupInspection::Paginated | StartupInspection::Skipped => {}
StartupInspection::Legacy => {
return migrate_all_rollouts(store, paths, skipped_rollouts.as_slice()).await;
}
StartupInspection::Unresolved => unresolved = true,
}
}
if unresolved {
return Ok(());
}
advance_last_checked_thread(store, paths.as_slice()).await
}
async fn migrate_all_rollouts(
store: &LocalThreadStore,
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
.iter()
.map(|skipped_rollout| skipped_rollout.rollout_path.as_str())
.collect::<HashSet<_>>();
let mut terminal = true;
let mut reported_paths = HashSet::new();
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?;
}
}
// 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 inspect_rollout_path(
store: &LocalThreadStore,
path: &Path,
) -> ThreadStoreResult<StartupInspection> {
let before = rollout_fingerprint(path).await?;
match codex_rollout::read_session_meta_line(path).await {
Ok(metadata) if metadata.meta.history_mode == ThreadHistoryMode::Legacy => {
Ok(StartupInspection::Legacy)
}
Ok(_) => Ok(StartupInspection::Paginated),
Err(error) => {
let after = rollout_fingerprint(path).await?;
if before != after || !matches!(error.kind(), ErrorKind::Other | ErrorKind::InvalidData)
{
return Ok(StartupInspection::Unresolved);
}
record_skip(store, path, before).await?;
Ok(StartupInspection::Skipped)
}
}
}
async fn record_skip(
store: &LocalThreadStore,
path: &Path,
fingerprint: RolloutFingerprint,
) -> 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()
},
};
state_db
.record_rollout_migration_skip(LEGACY_TO_PAGINATED_MIGRATION_ID, &skipped_rollout)
.await
.map_err(migration_error)
}
async fn remove_skip(store: &LocalThreadStore, rollout_path: &str) -> ThreadStoreResult<()> {
startup_state_db(store)?
.remove_rollout_migration_skip(LEGACY_TO_PAGINATED_MIGRATION_ID, rollout_path)
.await
.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],
) -> ThreadStoreResult<()> {
let last_checked_thread = paths
.iter()
.filter_map(|path| thread_creation_cursor(path))
.max();
startup_state_db(store)?
.advance_rollout_migration_state(
LEGACY_TO_PAGINATED_MIGRATION_ID,
last_checked_thread.as_ref(),
)
.await
.map_err(migration_error)
}
fn startup_state_db(store: &LocalThreadStore) -> ThreadStoreResult<&StateDbHandle> {
store
.state_db
.as_ref()
.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
.strip_suffix(".jsonl.zst")
.or_else(|| name.strip_suffix(".jsonl"))?
.strip_prefix("rollout-")?;
let separator = stem.len().checked_sub(37)?;
let thread_id = stem.get(separator + 1..)?;
ThreadId::from_string(thread_id).ok()?;
let timestamp = NaiveDateTime::parse_from_str(stem.get(..separator)?, "%Y-%m-%dT%H-%M-%S")
.ok()?
.and_utc()
.timestamp();
Some(RolloutMigrationCursor {
thread_created_at: timestamp,
thread_id: thread_id.to_string(),
})
}
fn relative_rollout_path(store: &LocalThreadStore, path: &Path) -> String {
path.strip_prefix(&store.config.codex_home)
.unwrap_or(path)
.to_string_lossy()
.replace('\\', "/")
}
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)?;
let modified_at_ns = metadata
.modified()
.map_err(migration_error)?
.duration_since(SystemTime::UNIX_EPOCH)
.map_err(migration_error)?
.as_nanos();
let modified_at_ns = i64::try_from(modified_at_ns).map_err(migration_error)?;
Ok(RolloutFingerprint {
size_bytes,
modified_at_ns,
})
}
#[cfg(test)]
#[path = "startup_tests.rs"]
mod tests;
codex-rs/thread-store/src/local/rollout_migration/startup_tests.rs
+333
View File
@@ -0,0 +1,333 @@
use std::fs;
use std::io::Write;
use std::path::Path;
use std::path::PathBuf;
use std::time::Duration;
use codex_protocol::ThreadId;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::RolloutItem;
use codex_protocol::protocol::RolloutLine;
use codex_protocol::protocol::SessionMeta;
use codex_protocol::protocol::SessionMetaLine;
use codex_protocol::protocol::SessionSource;
use codex_protocol::protocol::ThreadHistoryMode;
use codex_protocol::protocol::UserMessageEvent;
use codex_rollout::RolloutConfig;
use codex_rollout::RolloutRecorder;
use pretty_assertions::assert_eq;
use tempfile::TempDir;
use super::super::publish::migration_journal_path;
use super::super::publish::write_migration_journal;
use super::super::thread_history;
use super::LocalThreadStore;
use crate::local::test_support::test_config;
const TIMESTAMP: &str = "2025-01-03T12:00:00Z";
fn write_rollout(home: &Path, thread_id: ThreadId, history_mode: ThreadHistoryMode) -> PathBuf {
let directory = home.join("sessions/2025/01/03");
fs::create_dir_all(&directory).expect("create rollout directory");
let path = directory.join(format!("rollout-2025-01-03T12-00-00-{thread_id}.jsonl"));
let mut file = fs::File::create(&path).expect("create legacy rollout");
let paginated = history_mode == ThreadHistoryMode::Paginated;
let metadata = SessionMeta {
session_id: thread_id.into(),
id: thread_id,
timestamp: TIMESTAMP.to_string(),
cwd: home.to_path_buf(),
originator: "test-originator".to_string(),
cli_version: "0.0.0".to_string(),
source: SessionSource::Cli,
model_provider: Some("test-provider".to_string()),
history_mode,
..SessionMeta::default()
};
for (ordinal, item) in [
RolloutItem::SessionMeta(SessionMetaLine {
meta: metadata,
git: None,
}),
RolloutItem::EventMsg(EventMsg::UserMessage(UserMessageEvent {
message: "question".to_string(),
..UserMessageEvent::default()
})),
]
.into_iter()
.enumerate()
{
let line = RolloutLine {
timestamp: TIMESTAMP.to_string(),
ordinal: paginated.then_some(ordinal as u64),
item,
};
writeln!(
file,
"{}",
serde_json::to_string(&line).expect("serialize legacy record")
)
.expect("write legacy record");
}
path
}
fn move_to_timestamp(
home: &Path,
path: PathBuf,
session_day: &str,
filename_timestamp: &str,
) -> PathBuf {
let directory = home.join(format!("sessions/{session_day}"));
fs::create_dir_all(&directory).expect("create rollout directory");
let stem = path
.file_name()
.and_then(|name| name.to_str())
.and_then(|name| name.strip_suffix(".jsonl"))
.expect("rollout filename");
let thread_id = stem
.get(stem.len().checked_sub(36).expect("thread id offset")..)
.expect("thread id in rollout filename");
let moved_path = directory.join(format!("rollout-{filename_timestamp}-{thread_id}.jsonl"));
fs::rename(path, &moved_path).expect("move rollout timestamp");
moved_path
}
async fn indexed_store(home: &Path) -> LocalThreadStore {
let config = test_config(home);
let rollout_config = RolloutConfig {
codex_home: config.codex_home.clone(),
sqlite: config.sqlite.clone(),
cwd: home.to_path_buf(),
model_provider_id: config.default_model_provider_id.clone(),
generate_memories: false,
};
let state_db = codex_rollout::state_db::try_init(&rollout_config)
.await
.expect("backfill legacy thread metadata");
LocalThreadStore::new(config, Some(state_db))
}
#[tokio::test]
async fn records_and_advances_checked_thread() {
let home = TempDir::new().expect("create Codex home");
let legacy_thread_id = ThreadId::new();
let legacy_path = write_rollout(home.path(), legacy_thread_id, ThreadHistoryMode::Legacy);
let store = indexed_store(home.path()).await;
store
.migrate_rollouts_on_startup()
.await
.expect("migrate startup rollouts");
assert_eq!(
codex_rollout::read_session_meta_line(&legacy_path)
.await
.expect("read migrated metadata")
.meta
.history_mode,
ThreadHistoryMode::Paginated
);
let newer_thread_id = ThreadId::new();
move_to_timestamp(
home.path(),
write_rollout(home.path(), newer_thread_id, ThreadHistoryMode::Paginated),
"2025/01/04",
"2025-01-04T12-00-00",
);
store
.migrate_rollouts_on_startup()
.await
.expect("check newer paginated rollout");
let state = store
.state_db()
.await
.expect("state db")
.get_rollout_migration_state(super::LEGACY_TO_PAGINATED_MIGRATION_ID)
.await
.expect("read migration state")
.expect("migration state");
assert_eq!(
state
.last_checked_thread
.expect("last checked thread")
.thread_id,
newer_thread_id.to_string()
);
}
#[tokio::test]
async fn checks_rollouts_within_the_cursor_lookback() {
let home = TempDir::new().expect("create Codex home");
let older_thread_id = ThreadId::new();
let older_path = move_to_timestamp(
home.path(),
write_rollout(home.path(), older_thread_id, ThreadHistoryMode::Legacy),
"2025/01/02",
"2025-01-02T12-00-00",
);
let newer_path = move_to_timestamp(
home.path(),
write_rollout(home.path(), ThreadId::new(), ThreadHistoryMode::Paginated),
"2025/01/03",
"2025-01-03T12-00-00",
);
let store = indexed_store(home.path()).await;
let cursor = super::thread_creation_cursor(&newer_path).expect("newer rollout cursor");
store
.state_db()
.await
.expect("state db")
.advance_rollout_migration_state(super::LEGACY_TO_PAGINATED_MIGRATION_ID, Some(&cursor))
.await
.expect("seed migration cursor");
store
.migrate_rollouts_on_startup()
.await
.expect("check rollout behind cursor");
assert_eq!(
codex_rollout::read_session_meta_line(&older_path)
.await
.expect("read migrated metadata")
.meta
.history_mode,
ThreadHistoryMode::Paginated
);
}
#[tokio::test]
async fn recovers_pending_migrations_behind_the_checked_thread() {
let home = TempDir::new().expect("create Codex home");
let thread_id = ThreadId::new();
let path = write_rollout(home.path(), thread_id, ThreadHistoryMode::Legacy);
let store = indexed_store(home.path()).await;
store
.migrate_rollouts_on_startup()
.await
.expect("migrate and advance startup cursor");
thread_history::delete_thread(&store, thread_id)
.await
.expect("simulate missing projection");
let journal_path = migration_journal_path(home.path(), thread_id);
write_migration_journal(&journal_path)
.await
.expect("simulate pending migration marker");
store
.migrate_rollouts_on_startup()
.await
.expect("recover pending migration behind cursor");
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)
.await
.expect("read migrated metadata")
.meta
.history_mode,
ThreadHistoryMode::Paginated
);
}
#[tokio::test]
async fn waits_for_a_live_writer_before_migrating() {
let home = TempDir::new().expect("create Codex home");
let thread_id = ThreadId::new();
let path = write_rollout(home.path(), thread_id, ThreadHistoryMode::Legacy);
let store = indexed_store(home.path()).await;
let live_writer_guard = store.live_writer_locks.lock(thread_id).await;
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 the live writer"
);
drop(live_writer_guard);
migration.await.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 rechecks_changed_empty_rollouts() {
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(
home.path(),
write_rollout(home.path(), empty_thread_id, ThreadHistoryMode::Legacy),
"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;
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");
store
.migrate_rollouts_on_startup()
.await
.expect("migrate changed rollout");
assert_eq!(
codex_rollout::read_session_meta_line(&empty_path)
.await
.expect("read migrated metadata")
.meta
.history_mode,
ThreadHistoryMode::Paginated
);
assert!(
store
.state_db()
.await
.expect("state db")
.list_rollout_migration_skipped_rollouts(super::LEGACY_TO_PAGINATED_MIGRATION_ID)
.await
.expect("read skipped rollouts")
.is_empty()
);
}
codex-rs/thread-store/src/local/rollout_migration/telemetry.rs
+133
View File
@@ -0,0 +1,133 @@
//! Records low-cardinality migration metrics shared by manual and startup runs.
//!
//! Migration owns the outcome stream, so it is the one place that can count successful,
//! skipped, and failed rollouts consistently. Callers only tell it why the run started.
use std::time::Instant;
use super::RolloutMigrationMode;
use super::RolloutMigrationOptions;
use super::RolloutMigrationReport;
use super::RolloutMigrationStatus;
use crate::ThreadStoreResult;
const RUN_METRIC: &str = "codex.rollout_migration.run";
const RUN_DURATION_METRIC: &str = "codex.rollout_migration.run.duration_ms";
const RUN_IO_BYTES_METRIC: &str = "codex.rollout_migration.run.io_bytes";
const THREAD_METRIC: &str = "codex.rollout_migration.thread";
#[derive(Clone, Copy)]
pub(super) enum RolloutMigrationTrigger {
Manual,
Startup,
}
#[derive(Clone, Copy)]
enum RolloutMigrationScope {
All,
Selected,
}
pub(super) struct RolloutMigrationTelemetry {
trigger: RolloutMigrationTrigger,
mode: RolloutMigrationMode,
scope: RolloutMigrationScope,
started_at: Instant,
}
impl RolloutMigrationTelemetry {
pub(super) fn new(trigger: RolloutMigrationTrigger, options: &RolloutMigrationOptions) -> Self {
Self {
trigger,
mode: options.mode,
scope: if options.thread_ids.is_empty() {
RolloutMigrationScope::All
} else {
RolloutMigrationScope::Selected
},
started_at: Instant::now(),
}
}
pub(super) fn finish(&self, result: &ThreadStoreResult<RolloutMigrationReport>) {
let Some(metrics) = codex_otel::global() else {
return;
};
let mut io_bytes = 0_u64;
if let Ok(report) = result {
for outcome in &report.outcomes {
let tags = [
("trigger", self.trigger.tag()),
("mode", self.mode.tag()),
("scope", self.scope.tag()),
("status", outcome.status.tag()),
];
let _ = metrics.counter(THREAD_METRIC, /*inc*/ 1, &tags);
io_bytes = io_bytes.saturating_add(outcome.bytes_processed);
}
}
let result = match result {
Err(_) => "error",
Ok(report)
if report
.outcomes
.iter()
.any(|outcome| outcome.status == RolloutMigrationStatus::Failed) =>
{
"partial_failure"
}
Ok(_) => "success",
};
let tags = [
("trigger", self.trigger.tag()),
("mode", self.mode.tag()),
("scope", self.scope.tag()),
("result", result),
];
let duration_ms = i64::try_from(self.started_at.elapsed().as_millis()).unwrap_or(i64::MAX);
let io_bytes = i64::try_from(io_bytes).unwrap_or(i64::MAX);
let _ = metrics.counter(RUN_METRIC, /*inc*/ 1, &tags);
let _ = metrics.histogram(RUN_DURATION_METRIC, duration_ms, &tags);
let _ = metrics.histogram(RUN_IO_BYTES_METRIC, io_bytes, &tags);
}
}
impl RolloutMigrationTrigger {
fn tag(self) -> &'static str {
match self {
Self::Manual => "manual",
Self::Startup => "startup",
}
}
}
impl RolloutMigrationMode {
fn tag(self) -> &'static str {
match self {
Self::DryRun => "dry_run",
Self::Apply => "apply",
}
}
}
impl RolloutMigrationScope {
fn tag(self) -> &'static str {
match self {
Self::All => "all",
Self::Selected => "selected",
}
}
}
impl RolloutMigrationStatus {
fn tag(self) -> &'static str {
match self {
Self::Eligible => "eligible",
Self::Migrated => "migrated",
Self::AlreadyPaginated => "already_paginated",
Self::SkippedEmpty => "skipped_empty",
Self::SkippedBusy => "skipped_busy",
Self::Failed => "failed",
}
}
}
codex-rs/thread-store/src/local/rollout_migration_tests.rs
+14 -2
View File
@@ -41,6 +41,7 @@ use tempfile::TempDir;
use super::LocalThreadStore;
use super::RolloutMigrationMode;
use super::RolloutMigrationOptions;
use super::RolloutMigrationProgress;
use super::RolloutMigrationStatus;
#[cfg(unix)]
use super::decompress_rollout_to_path;
@@ -241,7 +242,7 @@ fn read_rollout(path: &Path) -> Vec<RolloutLine> {
fn apply_options() -> RolloutMigrationOptions {
RolloutMigrationOptions {
mode: RolloutMigrationMode::Apply,
max_mib_per_second: 1024,
max_mib_per_second: Some(1024),
..RolloutMigrationOptions::default()
}
}
@@ -1516,8 +1517,11 @@ async fn dry_run_reports_migration_order() {
let original = fs::read(&root).expect("read original root rollout");
let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None);
let mut progress = Vec::new();
let report = store
.migrate_rollouts(RolloutMigrationOptions::default())
.migrate_rollouts_with_progress(RolloutMigrationOptions::default(), |update| {
progress.push(update);
})
.await
.expect("inspect legacy rollouts");
@@ -1540,6 +1544,14 @@ async fn dry_run_reports_migration_order() {
.collect::<Vec<_>>(),
expected,
);
assert_eq!(
progress.last(),
Some(&RolloutMigrationProgress {
processed_paths: 5,
total_paths: 5,
outcome_status: Some(RolloutMigrationStatus::Eligible),
})
);
assert_eq!(fs::read(&root).expect("read inspected rollout"), original);
let selected = store