diff --git a/src/codex_config.rs b/src/codex_config.rs index 5839400..774373a 100644 --- a/src/codex_config.rs +++ b/src/codex_config.rs @@ -52,6 +52,10 @@ pub(crate) fn read_provider_from_config( .unwrap_or_else(|| DEFAULT_PROVIDER.to_string())) } +// Legacy single-store path resolver. The multi-store path now goes through +// `discover_stores` / `configured_sqlite_home`; this helper is only kept for the +// resolution unit tests, so it is gated to test builds to stay clippy-clean. +#[cfg(test)] pub(crate) fn resolve_sqlite_path( codex_home: &Path, profile_override: Option<&str>, @@ -188,6 +192,9 @@ fn trimmed_string(value: &str) -> Option { } } +// Test-only sibling of `configured_sqlite_home_from` that applies the +// `codex_home` default. Production code uses `configured_sqlite_home`. +#[cfg(test)] pub(crate) fn resolve_sqlite_home_from_config( codex_home: &Path, config_sqlite_home: Option<&str>, 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..bfa1d10 100644 --- a/src/output.rs +++ b/src/output.rs @@ -9,7 +9,6 @@ use crate::service::ServiceInstallSummary; use crate::service::ServiceManager; use crate::sync::MultiReconcileSummary; use crate::sync::ReconcileStatus; -use crate::sync::ReconcileSummary; use crate::sync::StatusSummary; use crate::sync::StoreOutcome; @@ -128,66 +127,6 @@ pub(crate) fn rollout_progress_message( } } -pub(crate) fn print_sync_summary(locale: Locale, title: &str, summary: &ReconcileSummary) { - println!("{title}"); - println!( - "{}: {}", - status_target_provider_label(locale), - summary.provider - ); - println!( - "{}: {}", - sync_rows_updated_label(locale), - summary.changed_rows - ); - if summary.checked_rollouts > 0 || summary.changed_rollouts > 0 { - println!( - "{}: {}", - sync_rollouts_checked_label(locale), - summary.checked_rollouts - ); - println!( - "{}: {}", - sync_rollouts_updated_label(locale), - summary.changed_rollouts - ); - if summary.prepared_rollouts > 0 { - println!( - "{}: {}", - sync_rollouts_prepared_label(locale), - summary.prepared_rollouts - ); - } - if summary.skipped_rollouts > 0 { - println!( - "{}: {}", - sync_rollouts_skipped_label(locale), - summary.skipped_rollouts - ); - } - } - println!( - "{}: {}", - status_total_threads_label(locale), - summary.total_rows - ); - println!( - "{}: {} ms", - sync_elapsed_label(locale), - summary.elapsed.as_millis() - ); - if let Some(backup_path) = &summary.backup_path { - println!("{}: {}", sync_backup_label(locale), backup_path.display()); - } - if let Some(journal_path) = &summary.rollout_journal_path { - println!( - "{}: {}", - sync_rollout_journal_label(locale), - journal_path.display() - ); - } -} - pub(crate) fn print_multi_sync_summary( locale: Locale, title: &str, @@ -252,6 +191,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 +883,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/rollout.rs b/src/rollout.rs index 7dbd99e..bb8e494 100644 --- a/src/rollout.rs +++ b/src/rollout.rs @@ -163,6 +163,7 @@ enum RolloutChangeMode { InPlace, } +#[cfg(test)] pub(crate) fn reconcile_rollout_metadata_from_sqlite_with_progress( sqlite_path: &Path, _codex_home: &Path, diff --git a/src/state_db.rs b/src/state_db.rs index 048a552..dcd983a 100644 --- a/src/state_db.rs +++ b/src/state_db.rs @@ -52,6 +52,13 @@ pub(crate) fn inspect_sqlite_distribution( /// callers can treat "no backfill machinery" and "complete" distinctly. Opens /// read-only to avoid contending with an in-progress rebuild. pub(crate) fn read_backfill_status(sqlite_path: &Path) -> Result> { + read_backfill_status_with_timeout(sqlite_path, Duration::from_millis(2_000)) +} + +pub(crate) fn read_backfill_status_with_timeout( + sqlite_path: &Path, + busy_timeout: Duration, +) -> Result> { if !sqlite_path.exists() { return Ok(None); } @@ -60,7 +67,7 @@ pub(crate) fn read_backfill_status(sqlite_path: &Path) -> Result> OpenFlags::SQLITE_OPEN_READ_ONLY | OpenFlags::SQLITE_OPEN_NO_MUTEX, ) .with_context(|| format!("failed to open {}", sqlite_path.display()))?; - connection.busy_timeout(Duration::from_millis(2_000))?; + connection.busy_timeout(busy_timeout)?; let has_table: Option = connection .query_row( "SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'backfill_state'", diff --git a/src/sync.rs b/src/sync.rs index 67444d7..a229413 100644 --- a/src/sync.rs +++ b/src/sync.rs @@ -1,19 +1,20 @@ 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; use crate::fs_sync::with_threadripper_lock; use crate::locale::detect_locale; use crate::rollout::RolloutProgressConfig; -use crate::rollout::RolloutReconcileSummary; use crate::rollout::RolloutScope; -use crate::rollout::reconcile_rollout_metadata_from_sqlite_with_progress; use crate::rollout::reconcile_rollouts_for_stores; use crate::service; use crate::service::ServiceStatus as BackgroundServiceStatus; @@ -21,6 +22,7 @@ use crate::state_db::ProviderDistribution; use crate::state_db::create_sqlite_backup_file_in; use crate::state_db::inspect_sqlite_distribution; use crate::state_db::read_backfill_status; +use crate::state_db::read_backfill_status_with_timeout; use crate::state_db::reconcile_sqlite_in_place; use crate::state_db::unix_timestamp_millis; use crate::stores::StoreKind; @@ -28,20 +30,6 @@ use crate::stores::StoreTarget; use crate::stores::discover_stores; use crate::stores::no_store_found_message; -#[derive(Debug)] -pub(crate) struct ReconcileSummary { - pub(crate) provider: String, - pub(crate) changed_rows: u64, - pub(crate) total_rows: u64, - pub(crate) changed_rollouts: u64, - pub(crate) checked_rollouts: u64, - pub(crate) prepared_rollouts: u64, - pub(crate) skipped_rollouts: u64, - pub(crate) elapsed: Duration, - pub(crate) backup_path: Option, - pub(crate) rollout_journal_path: Option, -} - /// Per-store status for a single discovered `state_5.sqlite` surface. #[derive(Debug)] pub(crate) struct StoreStatus { @@ -139,7 +127,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 +140,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,48 +183,135 @@ 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) + }) + } + + /// Total SQLite rows flipped across all updated stores. + pub(crate) fn total_changed_rows(&self) -> u64 { + total_changed_rows(&self.stores) + } +} + +/// 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(); + let read_timeout = Duration::from_millis(2_000).min(budget); + loop { + match read_backfill_status_with_timeout(db_path, read_timeout) { + 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, || { - reconcile_all_stores_with_backup_unlocked( + reconcile_stores_core( codex_home, provider_override, profile_override, rollout_scope, padding_bytes, + backfill_wait, + true, + progress, + ) + }) +} + +/// Like [`reconcile_all_stores_with_backup`] but without per-store backups: the +/// continuous `watch` service reconciles every poll, so backing up each store +/// every time would be wasteful. Supports the incremental `MismatchedRows` scope +/// (with an `AllRows` rollout followup when rows change), like the former +/// single-store watch path. +pub(crate) fn reconcile_all_stores( + codex_home: &Path, + provider_override: Option<&str>, + profile_override: Option<&str>, + rollout_scope: RolloutScope, + backfill_wait: Duration, + progress: Option, +) -> Result { + with_threadripper_lock(codex_home, || { + reconcile_stores_core( + codex_home, + provider_override, + profile_override, + rollout_scope, + DEFAULT_BUCKET_PADDING_BYTES, + backfill_wait, + false, progress, ) }) } -fn reconcile_all_stores_with_backup_unlocked( +#[allow(clippy::too_many_arguments)] +fn reconcile_stores_core( codex_home: &Path, provider_override: Option<&str>, profile_override: Option<&str>, rollout_scope: RolloutScope, padding_bytes: usize, + backfill_wait: Duration, + backup: bool, progress: Option, ) -> Result { let provider = match provider_override { @@ -246,27 +324,63 @@ fn reconcile_all_stores_with_backup_unlocked( } let started = Instant::now(); - // The multi-store path only supports whole-store scopes. MismatchedRows - // relies on the single-store followup pass (see reconcile_once_with_progress) - // that this path does not run; `sync` / `bucket switch` only pass AllRows or - // None. Guard against a future caller wiring MismatchedRows through here. - debug_assert!( - rollout_scope != RolloutScope::MismatchedRows, - "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 { - match create_store_backup(target) { - Ok(backup_path) => { - backup_paths.insert(target.db_path.clone(), backup_path); - } - Err(error) => { - backup_failed.insert(target.db_path.clone(), error.to_string()); + if backup { + 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); + } + Err(error) => { + backup_failed.insert(target.db_path.clone(), error.to_string()); + } } } } @@ -274,6 +388,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,60 +422,115 @@ 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(); - let rollout_journal_path = codex_home - .join("backups") - .join(format!("rollouts.{}.jsonl", unix_timestamp_millis()?)); + // The rollout change journal is part of the backup story: the one-shot + // commands write it next to the per-store backups, but the continuous + // `watch` service (backup = false) must not litter `backups/` every poll. + let rollout_journal_path = if backup { + Some( + codex_home + .join("backups") + .join(format!("rollouts.{}.jsonl", unix_timestamp_millis()?)), + ) + } else { + None + }; 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()), + rollout_journal_path.as_deref(), padding_bytes, progress, )?; - let rollout_summary = rollout_outcome.summary; + let mut rollout_summary = rollout_outcome.summary; 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 = if backup { + Some( + backup_paths + .remove(&target.db_path) + .expect("backup was prepared for ready store"), + ) + } else { + None }; + 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(); + // 3) MismatchedRows is incremental: the first pass only rewrote rollouts for + // rows the DB had marked mismatched. After flipping those rows, run a full + // AllRows rollout pass so any rollout that drifted is corrected too. (busy + // is empty here — the whole-round skip above already returned if a backfill + // was running under a rollout-rewriting scope.) + if rollout_scope == RolloutScope::MismatchedRows && total_changed_rows(&stores) > 0 { + let followup_db_paths: Vec = stores + .iter() + .filter(|store| matches!(store.outcome, StoreOutcome::Updated { .. })) + .map(|store| store.db_path.clone()) + .collect(); + let followup = reconcile_rollouts_for_stores( + followup_db_paths.as_slice(), + provider.as_str(), + RolloutScope::AllRows, + None, + padding_bytes, + None, + )?; + // These stores were just read successfully in the primary pass; a + // failure here would mean one became unreadable mid-run (rare). The + // primary DB + rollout writes are already consistent, so we only skip + // the optional drift repair, but flag it in debug builds. + debug_assert!( + followup.failed_stores.is_empty(), + "ready store became unreadable between rollout passes" + ); + rollout_summary.checked_files += followup.summary.checked_files; + rollout_summary.changed_files += followup.summary.changed_files; + rollout_summary.prepared_files += followup.summary.prepared_files; + rollout_summary.skipped_files += followup.summary.skipped_files; + } + Ok(MultiReconcileSummary { provider, stores, @@ -377,16 +553,27 @@ fn create_store_backup(target: &StoreTarget) -> Result { create_sqlite_backup_file_in(&target.db_path, &backups_dir) } +/// Total `changed_rows` across all updated stores. +fn total_changed_rows(stores: &[StoreReconcileResult]) -> u64 { + stores + .iter() + .map(|store| match &store.outcome { + StoreOutcome::Updated { changed_rows, .. } => *changed_rows, + _ => 0, + }) + .sum() +} + fn reconcile_single_store( target: &StoreTarget, provider: &str, - backup_path: PathBuf, + backup_path: Option, ) -> StoreReconcileResult { let outcome = match reconcile_single_store_inner(target, provider, backup_path) { Ok((changed_rows, total_rows, backup_path)) => StoreOutcome::Updated { changed_rows, total_rows, - backup_path: Some(backup_path), + backup_path, }, Err(error) => StoreOutcome::Failed { error: error.to_string(), @@ -402,98 +589,8 @@ fn reconcile_single_store( fn reconcile_single_store_inner( target: &StoreTarget, provider: &str, - backup_path: PathBuf, -) -> Result<(u64, u64, PathBuf)> { + backup_path: Option, +) -> Result<(u64, u64, Option)> { let (changed_rows, total_rows) = reconcile_sqlite_in_place(&target.db_path, provider)?; Ok((changed_rows, total_rows, backup_path)) } - -pub(crate) fn reconcile_once( - codex_home: &Path, - provider_override: Option<&str>, - profile_override: Option<&str>, - rollout_scope: RolloutScope, -) -> Result { - reconcile_once_with_progress( - codex_home, - provider_override, - profile_override, - rollout_scope, - None, - ) -} - -fn reconcile_once_with_progress( - codex_home: &Path, - provider_override: Option<&str>, - profile_override: Option<&str>, - rollout_scope: RolloutScope, - progress: Option, -) -> Result { - with_threadripper_lock(codex_home, || { - reconcile_once_with_progress_unlocked( - codex_home, - provider_override, - profile_override, - rollout_scope, - progress, - ) - }) -} - -fn reconcile_once_with_progress_unlocked( - codex_home: &Path, - provider_override: Option<&str>, - profile_override: Option<&str>, - rollout_scope: RolloutScope, - progress: Option, -) -> Result { - let provider = match provider_override { - Some(provider) => provider.to_string(), - None => read_provider_from_config(codex_home, profile_override)?, - }; - let sqlite_path = resolve_sqlite_path(codex_home, profile_override)?; - let started = Instant::now(); - let mut rollout_summary = reconcile_rollout_metadata_from_sqlite_with_progress( - &sqlite_path, - codex_home, - provider.as_str(), - rollout_scope, - None, - DEFAULT_BUCKET_PADDING_BYTES, - progress, - )?; - let (changed_rows, total_rows) = reconcile_sqlite_in_place(&sqlite_path, provider.as_str())?; - if rollout_scope == RolloutScope::MismatchedRows && changed_rows > 0 { - let followup_summary = reconcile_rollout_metadata_from_sqlite_with_progress( - &sqlite_path, - codex_home, - provider.as_str(), - RolloutScope::AllRows, - None, - DEFAULT_BUCKET_PADDING_BYTES, - None, - )?; - add_rollout_summary(&mut rollout_summary, followup_summary); - } - - Ok(ReconcileSummary { - provider, - changed_rows, - total_rows, - changed_rollouts: rollout_summary.changed_files, - checked_rollouts: rollout_summary.checked_files, - prepared_rollouts: rollout_summary.prepared_files, - skipped_rollouts: rollout_summary.skipped_files, - elapsed: started.elapsed(), - backup_path: None, - rollout_journal_path: rollout_summary.journal_path, - }) -} - -fn add_rollout_summary(summary: &mut RolloutReconcileSummary, extra: RolloutReconcileSummary) { - summary.checked_files += extra.checked_files; - summary.changed_files += extra.changed_files; - summary.prepared_files += extra.prepared_files; - summary.skipped_files += extra.skipped_files; -} diff --git a/src/tests.rs b/src/tests.rs index 4a9cfae..3a41c56 100644 --- a/src/tests.rs +++ b/src/tests.rs @@ -23,14 +23,17 @@ use crate::state_db::reconcile_sqlite_in_place; use crate::state_db::reconcile_sqlite_with_backup; use crate::stores::StoreKind; use crate::stores::discover_stores_with; +use crate::sync::MultiReconcileSummary; use crate::sync::ReconcileStatus; use crate::sync::StoreOutcome; +use crate::sync::StoreReconcileResult; use crate::sync::collect_status; +use crate::sync::reconcile_all_stores; use crate::sync::reconcile_all_stores_with_backup; -use crate::sync::reconcile_once; use crate::watch::WATCH_FULL_ROLLOUT_POLL_INTERVALS; use crate::watch::full_watch_rollout_scope; use crate::watch::periodic_watch_rollout_scope; +use crate::watch::watch_should_print_summary; use crate::watch::watched_config_paths; use anyhow::Result; use clap::FromArgMatches; @@ -42,6 +45,7 @@ use std::fs; use std::path::Path; use std::path::PathBuf; use std::time::Duration; +use std::time::Instant; #[test] fn parses_supported_locale_tags() { @@ -524,6 +528,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 +589,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 +636,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 +723,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 +776,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 +795,248 @@ 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 started = Instant::now(); + let summary = reconcile_all_stores_with_backup( + codex_home, + Some("openai"), + None, + RolloutScope::None, + DEFAULT_BUCKET_PADDING_BYTES, + Duration::from_millis(0), + None, + )?; + let elapsed = started.elapsed(); + lock.execute_batch("ROLLBACK;")?; + drop(lock); + + assert!(elapsed < Duration::from_millis(500)); + 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 reconcile_all_stores_writes_no_journal_without_backup() -> Result<()> { + let dir = tempfile::tempdir()?; + let codex_home = dir.path(); + isolate_process_sqlite_home(codex_home)?; + + 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"); + seed_store_with_provider(&cli_db, "cong")?; + { + let connection = Connection::open(&cli_db)?; + connection.execute( + "UPDATE threads SET rollout_path = ?1 WHERE id = '1'", + [rollout_path.display().to_string()], + )?; + } + + // The no-backup path (watch) rewrites the rollout but must not write a + // backups/rollouts.*.jsonl change journal, matching the old watch behaviour. + let summary = reconcile_all_stores( + codex_home, + Some("openai"), + None, + RolloutScope::AllRows, + Duration::from_millis(0), + None, + )?; + + assert_eq!(summary.changed_rollouts, 1); + assert!(fs::read_to_string(&rollout_path)?.contains("\"model_provider\":\"openai\"")); + assert!(summary.rollout_journal_path.is_none()); + + let backups_dir = codex_home.join("backups"); + let journal_count = if backups_dir.exists() { + fs::read_dir(&backups_dir)? + .filter_map(|entry| entry.ok()) + .filter(|entry| entry.file_name().to_string_lossy().starts_with("rollouts.")) + .count() + } else { + 0 + }; + assert_eq!(journal_count, 0); + Ok(()) +} + #[test] fn rejects_blank_provider_override() { let err = validate_provider_override(Locale::En, Some(" ")).unwrap_err(); @@ -829,18 +1080,35 @@ fn rejects_path_profile_override_in_cli_args_before_help_short_circuit() { } #[test] -fn reconcile_once_returns_error_when_db_missing() -> Result<()> { +fn reconcile_all_stores_reports_missing_configured_store_as_failed() -> Result<()> { let dir = tempfile::tempdir()?; - let sqlite_path = dir.path().join("state_5.sqlite"); fs::write( dir.path().join("config.toml"), - sqlite_home_config(dir.path()), + "model_provider = \"openai\"\nsqlite_home = \".threadripper-test-missing\"\n", )?; - let err = reconcile_once(dir.path(), Some("openai"), None, RolloutScope::None).unwrap_err(); + let summary = reconcile_all_stores( + dir.path(), + Some("openai"), + None, + RolloutScope::None, + Duration::from_millis(0), + None, + )?; - assert!(err.to_string().contains(&sqlite_path.display().to_string())); - assert!(!sqlite_path.exists()); + assert_eq!(summary.status(), ReconcileStatus::Failed); + assert_eq!(summary.stores.len(), 1); + assert_eq!(summary.stores[0].kind, StoreKind::Configured); + assert!(matches!( + summary.stores[0].outcome, + StoreOutcome::Failed { .. } + )); + assert!( + !dir.path() + .join(".threadripper-test-missing") + .join("state_5.sqlite") + .exists() + ); Ok(()) } @@ -1161,20 +1429,81 @@ fn mismatched_scope_followup_repairs_matching_sqlite_rollout_after_sqlite_change connection.execute("UPDATE threads SET rollout_path = '' WHERE id = '3'", [])?; drop(connection); - let summary = reconcile_once( + let summary = reconcile_all_stores( codex_home, Some("openai"), None, RolloutScope::MismatchedRows, + Duration::from_millis(0), + None, )?; - assert_eq!(summary.changed_rows, 1); + assert_eq!(summary.total_changed_rows(), 1); assert_eq!(summary.changed_rollouts, 2); assert!(fs::read_to_string(&rollout_a)?.contains("\"model_provider\":\"openai\"")); assert!(fs::read_to_string(&rollout_b)?.contains("\"model_provider\":\"openai\"")); Ok(()) } +#[test] +fn mismatched_scope_followup_excludes_failed_stores() -> Result<()> { + let dir = tempfile::tempdir()?; + let codex_home = dir.path(); + isolate_process_sqlite_home(codex_home)?; + + let rollout_cli = codex_home.join("sessions/2026/05/07/rollout-cli.jsonl"); + let rollout_app = codex_home.join("sessions/2026/05/07/rollout-app.jsonl"); + fs::create_dir_all(rollout_cli.parent().unwrap())?; + fs::write( + &rollout_cli, + "{\"type\":\"session_meta\",\"payload\":{\"id\":\"1\",\"model_provider\":\"vm\"}} \n", + )?; + fs::write( + &rollout_app, + "{\"type\":\"session_meta\",\"payload\":{\"id\":\"app\",\"model_provider\":\"cong\"}} \n", + )?; + + let cli_db = codex_home.join("state_5.sqlite"); + seed_store_with_provider(&cli_db, "vm")?; + { + let connection = Connection::open(&cli_db)?; + connection.execute( + "UPDATE threads SET rollout_path = ?1 WHERE id = '1'", + [rollout_cli.display().to_string()], + )?; + } + + let app_db = codex_home.join("sqlite").join("state_5.sqlite"); + fs::create_dir_all(app_db.parent().unwrap())?; + { + let connection = Connection::open(&app_db)?; + connection.execute( + "CREATE TABLE threads (id TEXT PRIMARY KEY, rollout_path TEXT NOT NULL)", + [], + )?; + connection.execute( + "INSERT INTO threads (id, rollout_path) VALUES ('app', ?1)", + [rollout_app.display().to_string()], + )?; + } + + let summary = reconcile_all_stores( + codex_home, + Some("openai"), + None, + RolloutScope::MismatchedRows, + Duration::from_millis(0), + None, + )?; + + assert_eq!(summary.status(), ReconcileStatus::Partial); + assert_eq!(summary.total_changed_rows(), 2); + assert!(fs::read_to_string(&rollout_cli)?.contains("\"model_provider\":\"openai\"")); + assert!(fs::read_to_string(&rollout_app)?.contains("\"model_provider\":\"cong\"")); + assert!(!fs::read_to_string(&rollout_app)?.contains("openai")); + Ok(()) +} + #[test] fn watch_uses_five_hundred_ms_by_default() -> Result<()> { let matches = @@ -1224,6 +1553,28 @@ fn watch_periodically_runs_full_rollout_scan() { ); } +#[test] +fn watch_prints_non_full_summary_even_when_counts_are_unchanged() { + let summary = MultiReconcileSummary { + provider: "openai".to_string(), + stores: vec![StoreReconcileResult { + kind: StoreKind::Cli, + db_path: PathBuf::from("/tmp/state_5.sqlite"), + outcome: StoreOutcome::Failed { + error: "missing store".to_string(), + }, + }], + changed_rollouts: 0, + checked_rollouts: 0, + prepared_rollouts: 0, + skipped_rollouts: 0, + rollout_journal_path: None, + elapsed: Duration::ZERO, + }; + + assert!(watch_should_print_summary(Some("openai"), &summary)); +} + #[test] fn watch_tracks_selected_profile_config_file() -> Result<()> { let dir = tempfile::tempdir()?; diff --git a/src/watch.rs b/src/watch.rs index 6dfd63f..1960f15 100644 --- a/src/watch.rs +++ b/src/watch.rs @@ -20,7 +20,7 @@ use crate::codex_config::selected_profile_config_path; use crate::locale::Locale; use crate::output::background_reconcile_title; use crate::output::config_change_title; -use crate::output::print_sync_summary; +use crate::output::print_multi_sync_summary; use crate::output::watch_already_running_error; use crate::output::watch_initial_reconcile_error_message; use crate::output::watch_reconcile_skipped_message; @@ -30,7 +30,9 @@ use crate::output::watch_stopped_message; use crate::output::watcher_disconnected_error; use crate::output::watcher_error_message; use crate::rollout::RolloutScope; -use crate::sync::reconcile_once; +use crate::sync::MultiReconcileSummary; +use crate::sync::ReconcileStatus; +use crate::sync::reconcile_all_stores; pub(crate) const WATCH_FULL_ROLLOUT_POLL_INTERVALS: u64 = 120; @@ -81,14 +83,16 @@ pub(crate) fn run_watch( } let mut last_provider = None; - match reconcile_once( + match reconcile_all_stores( codex_home, provider_override.as_deref(), profile_override.as_deref(), full_rollout_scope, + Duration::ZERO, + None, ) { Ok(summary) => { - print_sync_summary(locale, watch_started_title(locale), &summary); + print_multi_sync_summary(locale, watch_started_title(locale), &summary); last_provider = Some(summary.provider.clone()); } Err(err) => { @@ -108,18 +112,21 @@ pub(crate) fn run_watch( match rx.recv_timeout(timeout) { Ok(Ok(event)) => { if touches_config_file(&event, watched_paths.as_slice()) { - match reconcile_once( + match reconcile_all_stores( codex_home, provider_override.as_deref(), profile_override.as_deref(), full_rollout_scope, + Duration::ZERO, + None, ) { Ok(summary) => { - if last_provider.as_deref() != Some(summary.provider.as_str()) - || summary.changed_rows > 0 - || summary.changed_rollouts > 0 - { - print_sync_summary(locale, config_change_title(locale), &summary); + if watch_should_print_summary(last_provider.as_deref(), &summary) { + print_multi_sync_summary( + locale, + config_change_title(locale), + &summary, + ); } last_provider = Some(summary.provider.clone()); watched_paths = @@ -137,18 +144,17 @@ pub(crate) fn run_watch( } Err(mpsc::RecvTimeoutError::Timeout) => { poll_count = poll_count.wrapping_add(1); - match reconcile_once( + match reconcile_all_stores( codex_home, provider_override.as_deref(), profile_override.as_deref(), periodic_watch_rollout_scope(rollout_scope, poll_count), + Duration::ZERO, + None, ) { Ok(summary) => { - if last_provider.as_deref() != Some(summary.provider.as_str()) - || summary.changed_rows > 0 - || summary.changed_rollouts > 0 - { - print_sync_summary( + if watch_should_print_summary(last_provider.as_deref(), &summary) { + print_multi_sync_summary( locale, background_reconcile_title(locale), &summary, @@ -171,6 +177,16 @@ pub(crate) fn run_watch( Ok(()) } +pub(crate) fn watch_should_print_summary( + last_provider: Option<&str>, + summary: &MultiReconcileSummary, +) -> bool { + summary.status() != ReconcileStatus::Full + || last_provider != Some(summary.provider.as_str()) + || summary.total_changed_rows() > 0 + || summary.changed_rollouts > 0 +} + pub(crate) fn periodic_watch_rollout_scope( rollout_scope: RolloutScope, poll_count: u64,