diff --git a/src/main.rs b/src/main.rs index 69ac48d..79515a9 100644 --- a/src/main.rs +++ b/src/main.rs @@ -45,6 +45,7 @@ use output::uninstall_launchd_done; use rollout::RolloutProgressConfig; use rollout::RolloutScope; use rollout::prepare_bucket_padding; +use sync::DEFAULT_BACKFILL_WAIT; use sync::ReconcileStatus; use sync::collect_status; use sync::reconcile_all_stores_with_backup; @@ -91,10 +92,11 @@ fn run() -> Result { cli.profile.as_deref(), rollout_scope, DEFAULT_BUCKET_PADDING_BYTES, + DEFAULT_BACKFILL_WAIT, progress, )?; print_multi_sync_summary(locale, sync_complete_title(locale), &summary); - if sqlite_only && summary.touches_app_store(&codex_home) { + if sqlite_only && summary.app_store_updated(&codex_home) { eprintln!("{}", sqlite_only_app_warning(locale)); } Ok(exit_code_for(summary.status())) @@ -121,6 +123,7 @@ fn run() -> Result { cli.profile.as_deref(), RolloutScope::AllRows, padding_bytes, + DEFAULT_BACKFILL_WAIT, Some(RolloutProgressConfig { locale }), )?; print_multi_sync_summary(locale, bucket_switch_complete_title(locale), &summary); diff --git a/src/output.rs b/src/output.rs index 07ecc9c..bd550e3 100644 --- a/src/output.rs +++ b/src/output.rs @@ -252,6 +252,9 @@ pub(crate) fn print_multi_sync_summary( ); } } + StoreOutcome::Skipped => { + println!(" {}", sync_store_skipped_label(locale)); + } StoreOutcome::Failed { error } => { println!(" {}: {}", sync_store_failed_label(locale), error); } @@ -941,15 +944,26 @@ pub(crate) fn sync_store_failed_label(locale: Locale) -> &'static str { } } +pub(crate) fn sync_store_skipped_label(locale: Locale) -> &'static str { + match locale { + Locale::En => { + "Skipped — a Codex backfill has not completed; threadripper avoids racing the rebuild. Re-run once Codex finishes (if it keeps skipping, check whether the backfill is stuck)." + } + Locale::ZhHans => { + "已跳过 —— Codex backfill 尚未完成；threadripper 不与重建竞态。待 Codex 完成后重跑（若持续跳过，请检查 backfill 是否卡住）。" + } + } +} + pub(crate) fn reconcile_status_line(locale: Locale, status: ReconcileStatus) -> String { match (status, locale) { (ReconcileStatus::Full, Locale::En) => "Result: all stores updated.".to_string(), (ReconcileStatus::Full, Locale::ZhHans) => "结果：所有库均已更新。".to_string(), (ReconcileStatus::Partial, Locale::En) => { - "Result: PARTIAL — some stores updated, at least one failed (see above). Re-run after resolving the failure.".to_string() + "Result: PARTIAL — some stores updated, at least one was skipped or failed (see above). Re-run after it is resolved.".to_string() } (ReconcileStatus::Partial, Locale::ZhHans) => { - "结果：部分成功 —— 部分库已更新，至少一个失败（见上）。解决后请重跑。".to_string() + "结果：部分成功 —— 部分库已更新，至少一个被跳过或失败（见上）。解决后请重跑。".to_string() } (ReconcileStatus::Failed, Locale::En) => { "Result: FAILED — no store could be updated.".to_string() diff --git a/src/sync.rs b/src/sync.rs index 67444d7..bd74bdb 100644 --- a/src/sync.rs +++ b/src/sync.rs @@ -1,10 +1,14 @@ use anyhow::Result; use std::collections::HashMap; +use std::collections::HashSet; use std::path::Path; use std::path::PathBuf; use std::time::Duration; use std::time::Instant; +use rusqlite::Error as RusqliteError; +use rusqlite::ErrorCode; + use crate::cli::DEFAULT_BUCKET_PADDING_BYTES; use crate::codex_config::read_provider_from_config; use crate::codex_config::resolve_sqlite_path; @@ -139,7 +143,7 @@ pub(crate) fn collect_status( pub(crate) enum ReconcileStatus { /// Every selected store was updated. Exit code 0. Full, - /// Some stores were updated and at least one failed. Exit code 2. + /// Some stores were updated and at least one was skipped or failed. Exit code 2. Partial, /// No store could be updated. Exit code 1. Failed, @@ -152,6 +156,9 @@ pub(crate) enum StoreOutcome { total_rows: u64, backup_path: Option, }, + /// Left untouched because Codex's startup backfill was still running after + /// the bounded wait; the user should re-run once the rebuild finishes. + Skipped, Failed { error: String, }, @@ -192,28 +199,77 @@ impl MultiReconcileSummary { } } - /// True when any selected store is the Codex App surface — used to warn that + /// True when the Codex App store was actually updated — used to warn that /// `--sqlite-only` edits there may be reverted by Codex's rollout backfill. - pub(crate) fn touches_app_store(&self, codex_home: &Path) -> bool { + /// A skipped/failed App store did not change, so no warning is needed. + pub(crate) fn app_store_updated(&self, codex_home: &Path) -> bool { let app_db_path = codex_home .join(crate::stores::APP_SQLITE_SUBDIR) .join(crate::codex_config::STATE_DB_FILENAME); let app_db_path = app_db_path.canonicalize().unwrap_or(app_db_path); - self.stores - .iter() - .any(|store| store.kind == StoreKind::App || store.db_path == app_db_path) + self.stores.iter().any(|store| { + matches!(store.outcome, StoreOutcome::Updated { .. }) + && (store.kind == StoreKind::App || store.db_path == app_db_path) + }) } } +/// Default bounded wait for an in-progress Codex backfill before a store is +/// skipped. A one-shot `sync` can afford to pause briefly; if the rebuild is not +/// done by then the store is skipped and the user re-runs later. +pub(crate) const DEFAULT_BACKFILL_WAIT: Duration = Duration::from_secs(10); + +const BACKFILL_POLL_INTERVAL: Duration = Duration::from_millis(500); + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum BackfillReadiness { + Ready, + Busy, +} + +/// Wait up to `budget` for a store's Codex startup-backfill to finish before we +/// write to it, so threadripper never races Codex's rebuild. A store with no +/// `backfill_state` table (older Codex) or a `complete` status is ready +/// immediately; a status read error is treated as ready and the write phase +/// surfaces any real problem. +fn wait_for_store_backfill(db_path: &Path, budget: Duration) -> BackfillReadiness { + let started = Instant::now(); + loop { + match read_backfill_status(db_path) { + Ok(None) => return BackfillReadiness::Ready, + Ok(Some(status)) if status == "complete" => return BackfillReadiness::Ready, + Ok(Some(_)) => {} + Err(error) if is_sqlite_lock_error(&error) => {} + Err(_) => return BackfillReadiness::Ready, + } + if started.elapsed() >= budget { + return BackfillReadiness::Busy; + } + std::thread::sleep(BACKFILL_POLL_INTERVAL.min(budget)); + } +} + +fn is_sqlite_lock_error(error: &anyhow::Error) -> bool { + error.chain().any(|cause| { + matches!( + cause.downcast_ref::(), + Some(RusqliteError::SqliteFailure(error, _)) + if matches!(error.code, ErrorCode::DatabaseBusy | ErrorCode::DatabaseLocked) + ) + }) +} + /// Reconcile the provider across **all** discovered stores plus the shared /// rollout JSONL, backing up each store first. This is the multi-store write -/// path for the one-shot `sync` / `bucket switch` commands. +/// path for the one-shot `sync` / `bucket switch` commands. A store whose Codex +/// backfill is still running after `backfill_wait` is skipped, not written. pub(crate) fn reconcile_all_stores_with_backup( codex_home: &Path, provider_override: Option<&str>, profile_override: Option<&str>, rollout_scope: RolloutScope, padding_bytes: usize, + backfill_wait: Duration, progress: Option, ) -> Result { with_threadripper_lock(codex_home, || { @@ -223,6 +279,7 @@ pub(crate) fn reconcile_all_stores_with_backup( profile_override, rollout_scope, padding_bytes, + backfill_wait, progress, ) }) @@ -234,6 +291,7 @@ fn reconcile_all_stores_with_backup_unlocked( profile_override: Option<&str>, rollout_scope: RolloutScope, padding_bytes: usize, + backfill_wait: Duration, progress: Option, ) -> Result { let provider = match provider_override { @@ -255,12 +313,55 @@ fn reconcile_all_stores_with_backup_unlocked( "multi-store reconcile expects AllRows or None" ); - // 1) Take every per-store backup before touching shared rollout JSONL. If a - // rollout-writing command cannot back up one selected store, skip the - // whole round so no store is left with rewritten rollouts but an old DB. + // 0) Backfill guard: a store whose Codex startup-backfill is still running + // after the bounded wait is skipped entirely — we neither collect its + // (possibly partial) rollout targets nor write its DB, so we never race + // the rebuild. + let busy: HashSet = targets + .iter() + .filter(|target| { + wait_for_store_backfill(&target.db_path, backfill_wait) == BackfillReadiness::Busy + }) + .map(|target| target.db_path.clone()) + .collect(); + + // Rewriting any shared rollout JSONL (scope != None) while *any* store's + // backfill is running races Codex's rebuild on its own source of truth — the + // rollout files it is actively reading. Even a "ready" store's rollouts may + // be referenced by the busy store's session. So if we would touch rollouts + // and a backfill is in progress, skip the whole round and let the user re-run + // once it completes. `--sqlite-only` (RolloutScope::None) touches no rollout, + // so its ready stores can still be written below. + if rollout_scope != RolloutScope::None && !busy.is_empty() { + let stores = targets + .iter() + .map(|target| StoreReconcileResult { + kind: target.kind, + db_path: target.db_path.clone(), + outcome: StoreOutcome::Skipped, + }) + .collect(); + return Ok(MultiReconcileSummary { + provider, + stores, + changed_rollouts: 0, + checked_rollouts: 0, + prepared_rollouts: 0, + skipped_rollouts: 0, + rollout_journal_path: None, + elapsed: started.elapsed(), + }); + } + + // 1) Take every ready store's backup before touching shared rollout JSONL. + // If a rollout-writing command cannot back up one selected store, skip + // the whole round so no store is left with rewritten rollouts but an old DB. let mut backup_paths: HashMap = HashMap::new(); let mut backup_failed: HashMap = HashMap::new(); - for target in &targets { + for target in targets + .iter() + .filter(|target| !busy.contains(&target.db_path)) + { match create_store_backup(target) { Ok(backup_path) => { backup_paths.insert(target.db_path.clone(), backup_path); @@ -274,6 +375,13 @@ fn reconcile_all_stores_with_backup_unlocked( let stores = targets .iter() .map(|target| { + if busy.contains(&target.db_path) { + return StoreReconcileResult { + kind: target.kind, + db_path: target.db_path.clone(), + outcome: StoreOutcome::Skipped, + }; + } let error = backup_failed .get(&target.db_path) .cloned() @@ -301,10 +409,11 @@ fn reconcile_all_stores_with_backup_unlocked( } // 2) Rollout JSONL is the shared, durable source of truth. Collect targets - // across every backed-up store (deduped by canonical path) and rewrite - // once, before any SQLite row is flipped. - let store_db_paths: Vec = targets + // across the ready, backed-up stores (deduped by canonical path) and + // rewrite once, before any SQLite row is flipped. + let ready_db_paths: Vec = targets .iter() + .filter(|target| !busy.contains(&target.db_path)) .filter(|target| !backup_failed.contains_key(&target.db_path)) .map(|target| target.db_path.clone()) .collect(); @@ -312,7 +421,7 @@ fn reconcile_all_stores_with_backup_unlocked( .join("backups") .join(format!("rollouts.{}.jsonl", unix_timestamp_millis()?)); let rollout_outcome = reconcile_rollouts_for_stores( - store_db_paths.as_slice(), + ready_db_paths.as_slice(), provider.as_str(), rollout_scope, Some(rollout_journal_path.as_path()), @@ -323,35 +432,41 @@ fn reconcile_all_stores_with_backup_unlocked( let rollout_failed: HashMap = rollout_outcome.failed_stores.into_iter().collect(); - // 3) Reconcile each backed-up store's SQLite. A store whose rollouts could - // not be read is marked Failed and its DB is left untouched, so we never - // flip a DB while its rollouts stay stale; other per-store failures are - // likewise reported without aborting the healthy stores. + // 3) Reconcile each ready, backed-up store's SQLite. A store mid-backfill is + // Skipped; one whose rollouts could not be read is Failed and left + // untouched (so we never flip a DB while its rollouts stay stale); other + // per-store failures are likewise reported without aborting healthy stores. let stores: Vec = targets .iter() .map(|target| { - if let Some(error) = backup_failed.get(&target.db_path) { - return StoreReconcileResult { + if busy.contains(&target.db_path) { + StoreReconcileResult { + kind: target.kind, + db_path: target.db_path.clone(), + outcome: StoreOutcome::Skipped, + } + } else if let Some(error) = backup_failed.get(&target.db_path) { + StoreReconcileResult { kind: target.kind, db_path: target.db_path.clone(), outcome: StoreOutcome::Failed { error: error.clone(), }, - }; - } - if let Some(error) = rollout_failed.get(&target.db_path) { - return StoreReconcileResult { + } + } else if let Some(error) = rollout_failed.get(&target.db_path) { + StoreReconcileResult { kind: target.kind, db_path: target.db_path.clone(), outcome: StoreOutcome::Failed { error: error.clone(), }, - }; + } + } else { + let backup_path = backup_paths + .remove(&target.db_path) + .expect("backup was prepared for ready store"); + reconcile_single_store(target, provider.as_str(), backup_path) } - let backup_path = backup_paths - .remove(&target.db_path) - .expect("backup was prepared for ready store"); - reconcile_single_store(target, provider.as_str(), backup_path) }) .collect(); diff --git a/src/tests.rs b/src/tests.rs index 4a9cfae..6320d61 100644 --- a/src/tests.rs +++ b/src/tests.rs @@ -524,6 +524,7 @@ fn reconcile_all_stores_updates_every_surface_and_dedupes_rollout() -> Result<() None, RolloutScope::AllRows, DEFAULT_BUCKET_PADDING_BYTES, + Duration::from_millis(0), None, )?; @@ -584,6 +585,7 @@ fn reconcile_all_stores_reports_partial_when_a_store_is_unreadable() -> Result<( None, RolloutScope::None, DEFAULT_BUCKET_PADDING_BYTES, + Duration::from_millis(0), None, )?; @@ -630,13 +632,14 @@ fn sqlite_only_warning_detects_configured_app_alias() -> Result<()> { None, RolloutScope::None, DEFAULT_BUCKET_PADDING_BYTES, + Duration::from_millis(50), None, )?; assert_eq!(summary.status(), ReconcileStatus::Full); assert_eq!(summary.stores.len(), 1); assert_eq!(summary.stores[0].kind, StoreKind::Configured); - assert!(summary.touches_app_store(codex_home)); + assert!(summary.app_store_updated(codex_home)); Ok(()) } @@ -716,6 +719,7 @@ fn reconcile_all_stores_skips_rollout_when_store_backup_fails() -> Result<()> { None, RolloutScope::AllRows, DEFAULT_BUCKET_PADDING_BYTES, + Duration::from_millis(0), None, ); fs::set_permissions(app_dir, fs::Permissions::from_mode(original_mode))?; @@ -768,6 +772,7 @@ fn reconcile_all_stores_fails_store_when_rollout_targets_unreadable() -> Result< None, RolloutScope::AllRows, DEFAULT_BUCKET_PADDING_BYTES, + Duration::from_millis(0), None, )?; @@ -786,6 +791,192 @@ fn reconcile_all_stores_fails_store_when_rollout_targets_unreadable() -> Result< Ok(()) } +fn seed_store_with_provider(db: &Path, provider: &str) -> Result<()> { + seed_sqlite(db)?; + let connection = Connection::open(db)?; + connection.execute( + "UPDATE threads SET model_provider = ?1 WHERE id = '1'", + [provider], + )?; + Ok(()) +} + +fn set_backfill_status(db: &Path, status: &str) -> Result<()> { + let connection = Connection::open(db)?; + connection.execute( + "CREATE TABLE IF NOT EXISTS backfill_state (id INTEGER PRIMARY KEY, status TEXT NOT NULL)", + [], + )?; + connection.execute( + "INSERT OR REPLACE INTO backfill_state (id, status) VALUES (1, ?1)", + [status], + )?; + Ok(()) +} + +fn provider_of(db: &Path) -> Result { + let connection = Connection::open(db)?; + Ok(connection.query_row( + "SELECT model_provider FROM threads WHERE id = '1'", + [], + |row| row.get(0), + )?) +} + +#[test] +fn reconcile_all_stores_skips_busy_store_in_sqlite_only() -> Result<()> { + let dir = tempfile::tempdir()?; + let codex_home = dir.path(); + isolate_process_sqlite_home(codex_home)?; + + let cli_db = codex_home.join("state_5.sqlite"); + seed_store_with_provider(&cli_db, "cong")?; + let app_db = codex_home.join("sqlite").join("state_5.sqlite"); + fs::create_dir_all(app_db.parent().unwrap())?; + seed_store_with_provider(&app_db, "cong")?; + set_backfill_status(&app_db, "running")?; + + // --sqlite-only (RolloutScope::None) touches no rollout, so the ready CLI + // store is still written while the busy App store is skipped. + let summary = reconcile_all_stores_with_backup( + codex_home, + Some("openai"), + None, + RolloutScope::None, + DEFAULT_BUCKET_PADDING_BYTES, + Duration::from_millis(50), + None, + )?; + + assert_eq!(summary.status(), ReconcileStatus::Partial); + let cli = summary + .stores + .iter() + .find(|store| store.kind == StoreKind::Cli) + .expect("cli store present"); + assert!(matches!(cli.outcome, StoreOutcome::Updated { .. })); + let app = summary + .stores + .iter() + .find(|store| store.kind == StoreKind::App) + .expect("app store present"); + assert!(matches!(app.outcome, StoreOutcome::Skipped)); + assert_eq!(provider_of(&cli_db)?, "openai"); + assert_eq!(provider_of(&app_db)?, "cong"); + Ok(()) +} + +#[test] +fn reconcile_all_stores_treats_locked_backfill_status_as_busy() -> Result<()> { + let dir = tempfile::tempdir()?; + let codex_home = dir.path(); + fs::write(codex_home.join("config.toml"), sqlite_home_config(codex_home))?; + + let cli_db = codex_home.join("state_5.sqlite"); + seed_store_with_provider(&cli_db, "cong")?; + set_backfill_status(&cli_db, "running")?; + let lock = Connection::open(&cli_db)?; + lock.execute_batch("BEGIN EXCLUSIVE;")?; + + let summary = reconcile_all_stores_with_backup( + codex_home, + Some("openai"), + None, + RolloutScope::None, + DEFAULT_BUCKET_PADDING_BYTES, + Duration::from_millis(0), + None, + )?; + lock.execute_batch("ROLLBACK;")?; + drop(lock); + + assert_eq!(summary.status(), ReconcileStatus::Failed); + assert!(matches!(summary.stores[0].outcome, StoreOutcome::Skipped)); + assert_eq!(provider_of(&cli_db)?, "cong"); + Ok(()) +} + +#[test] +fn reconcile_all_stores_skips_whole_round_when_backfill_running() -> Result<()> { + let dir = tempfile::tempdir()?; + let codex_home = dir.path(); + isolate_process_sqlite_home(codex_home)?; + + // A rollout shared by both stores' thread "1". + let rollout_path = codex_home.join("sessions/2026/05/07/rollout-1.jsonl"); + fs::create_dir_all(rollout_path.parent().unwrap())?; + fs::write( + &rollout_path, + "{\"type\":\"session_meta\",\"payload\":{\"id\":\"1\",\"model_provider\":\"cong\"}} \n", + )?; + + let cli_db = codex_home.join("state_5.sqlite"); + let app_db = codex_home.join("sqlite").join("state_5.sqlite"); + fs::create_dir_all(app_db.parent().unwrap())?; + for db in [&cli_db, &app_db] { + seed_store_with_provider(db, "cong")?; + let connection = Connection::open(db)?; + connection.execute( + "UPDATE threads SET rollout_path = ?1 WHERE id = '1'", + [rollout_path.display().to_string()], + )?; + } + set_backfill_status(&app_db, "running")?; + + // A rollout-rewriting scope (AllRows) while App's backfill runs must skip the + // whole round: rewriting the shared rollout would race Codex's rebuild. + let summary = reconcile_all_stores_with_backup( + codex_home, + Some("openai"), + None, + RolloutScope::AllRows, + DEFAULT_BUCKET_PADDING_BYTES, + Duration::from_millis(50), + None, + )?; + + assert_eq!(summary.status(), ReconcileStatus::Failed); + for store in &summary.stores { + assert!(matches!(store.outcome, StoreOutcome::Skipped)); + } + // Nothing was touched: both DBs and the shared rollout are unchanged. + assert_eq!(provider_of(&cli_db)?, "cong"); + assert_eq!(provider_of(&app_db)?, "cong"); + let rollout = fs::read_to_string(&rollout_path)?; + assert!(rollout.contains("\"model_provider\":\"cong\"")); + assert!(!rollout.contains("openai")); + Ok(()) +} + +#[test] +fn reconcile_all_stores_treats_complete_backfill_as_ready() -> Result<()> { + let dir = tempfile::tempdir()?; + let codex_home = dir.path(); + isolate_process_sqlite_home(codex_home)?; + + let cli_db = codex_home.join("state_5.sqlite"); + seed_store_with_provider(&cli_db, "cong")?; + set_backfill_status(&cli_db, "complete")?; + + let summary = reconcile_all_stores_with_backup( + codex_home, + Some("openai"), + None, + RolloutScope::AllRows, + DEFAULT_BUCKET_PADDING_BYTES, + Duration::from_millis(50), + None, + )?; + + assert_eq!(summary.status(), ReconcileStatus::Full); + assert!(matches!( + summary.stores[0].outcome, + StoreOutcome::Updated { .. } + )); + assert_eq!(provider_of(&cli_db)?, "openai"); + Ok(()) +} + #[test] fn rejects_blank_provider_override() { let err = validate_provider_override(Locale::En, Some(" ")).unwrap_err();