From 268baf285003fe86bbfbfba24a6c87b88a7a3427 Mon Sep 17 00:00:00 2001 From: wangnov <48670012+Wangnov@users.noreply.github.com> Date: Sat, 13 Jun 2026 11:40:25 +0800 Subject: [PATCH 1/2] feat: write provider across all stores in sync / bucket switch PR2 of the multi-store adaptation (builds on #16). The one-shot write commands now reconcile every discovered store plus the shared rollout JSONL, instead of a single resolved DB. - sync.rs: reconcile_all_stores_with_backup discovers all stores, rewrites the shared rollout JSONL once (deduped by canonical path, before any DB row is flipped), then backs up and reconciles each store's SQLite. A per-store failure is reported (StoreOutcome::Failed) without aborting healthy stores. - A store whose rollout targets cannot be read is marked Failed and its DB is left untouched, so a DB is never flipped while its rollouts stay stale. - Backups are namespaced per store: /backups//. - main returns ExitCode: Full -> 0, Partial -> 2, Failed -> 1. - --sqlite-only warns that App-store SQLite-only edits may be reverted by Codex's rollout backfill (rollout JSONL is the source of truth). - watch still uses the single-store path (MismatchedRows followup); a debug_assert guards the multi-store path against MismatchedRows misuse. - Remove the now-unused single-store backup helpers. Reviewed in parallel by Codex and a Claude subagent. Codex caught a real consistency hole (a rollout-collection failure was swallowed while the DB write could still succeed, reporting Full while the rollout stayed stale); fixed by failing such stores and skipping their DB write, with a regression test. 50 tests, clippy and fmt clean. --- src/main.rs | 49 +++++-- src/output.rs | 133 +++++++++++++++++++ src/rollout.rs | 100 ++++++++++++++- src/state_db.rs | 22 ++-- src/sync.rs | 330 ++++++++++++++++++++++++++++++++++++++---------- src/tests.rs | 311 ++++++++++++++++++++++++++++++++++++++++++++- 6 files changed, 853 insertions(+), 92 deletions(-) diff --git a/src/main.rs b/src/main.rs index 6458d33..69ac48d 100644 --- a/src/main.rs +++ b/src/main.rs @@ -2,6 +2,7 @@ use anyhow::Context; use anyhow::Result; use std::path::Path; use std::path::PathBuf; +use std::process::ExitCode; use std::time::Duration; mod cli; @@ -20,6 +21,7 @@ mod watch; use cli::BucketCommand; use cli::Command; +use cli::DEFAULT_BUCKET_PADDING_BYTES; use cli::parse_cli; use cli::validate_profile_override; use cli::validate_provider_override; @@ -34,20 +36,31 @@ use output::next_steps_heading; use output::no_launchd_plist_message; use output::print_bucket_prepare_summary; use output::print_install_service_summary; +use output::print_multi_sync_summary; use output::print_status; -use output::print_sync_summary; use output::run_status_next_step; +use output::sqlite_only_app_warning; use output::sync_complete_title; use output::uninstall_launchd_done; use rollout::RolloutProgressConfig; use rollout::RolloutScope; use rollout::prepare_bucket_padding; +use sync::ReconcileStatus; use sync::collect_status; -use sync::reconcile_once_with_backup_and_padding; -use sync::reconcile_once_with_backup_progress; +use sync::reconcile_all_stores_with_backup; use watch::run_watch; -fn main() -> Result<()> { +fn main() -> ExitCode { + match run() { + Ok(code) => code, + Err(err) => { + eprintln!("{err:?}"); + ExitCode::FAILURE + } + } +} + +fn run() -> Result { let locale = detect_locale(); let cli = parse_cli(locale)?; validate_provider_override(locale, cli.provider.as_deref())?; @@ -59,6 +72,7 @@ fn main() -> Result<()> { let summary = collect_status(&codex_home, cli.provider.as_deref(), cli.profile.as_deref())?; print_status(locale, &summary); + Ok(ExitCode::SUCCESS) } Command::Sync { sqlite_only } => { let rollout_scope = if sqlite_only { @@ -71,20 +85,26 @@ fn main() -> Result<()> { } else { Some(RolloutProgressConfig { locale }) }; - let summary = reconcile_once_with_backup_progress( + let summary = reconcile_all_stores_with_backup( &codex_home, cli.provider.as_deref(), cli.profile.as_deref(), rollout_scope, + DEFAULT_BUCKET_PADDING_BYTES, progress, )?; - print_sync_summary(locale, sync_complete_title(locale), &summary); + print_multi_sync_summary(locale, sync_complete_title(locale), &summary); + if sqlite_only && summary.touches_app_store(&codex_home) { + eprintln!("{}", sqlite_only_app_warning(locale)); + } + Ok(exit_code_for(summary.status())) } Command::Bucket { command } => match command { BucketCommand::Prepare { padding_bytes } => { let summary = prepare_bucket_padding(&codex_home, cli.profile.as_deref(), padding_bytes)?; print_bucket_prepare_summary(locale, &summary); + Ok(ExitCode::SUCCESS) } BucketCommand::Switch { target_provider, @@ -95,7 +115,7 @@ fn main() -> Result<()> { Some(provider) => Some(provider), None => cli.provider.clone(), }; - let summary = reconcile_once_with_backup_and_padding( + let summary = reconcile_all_stores_with_backup( &codex_home, provider.as_deref(), cli.profile.as_deref(), @@ -103,7 +123,8 @@ fn main() -> Result<()> { padding_bytes, Some(RolloutProgressConfig { locale }), )?; - print_sync_summary(locale, bucket_switch_complete_title(locale), &summary); + print_multi_sync_summary(locale, bucket_switch_complete_title(locale), &summary); + Ok(exit_code_for(summary.status())) } }, Command::Watch { @@ -122,6 +143,7 @@ fn main() -> Result<()> { }, Duration::from_millis(poll_interval_ms), )?; + Ok(ExitCode::SUCCESS) } Command::PrintServiceConfig { poll_interval_ms } => { let exe_path = std::env::current_exe().context(current_exe_error(locale))?; @@ -133,6 +155,7 @@ fn main() -> Result<()> { Duration::from_millis(poll_interval_ms), )?; println!("{config}"); + Ok(ExitCode::SUCCESS) } Command::InstallService { poll_interval_ms } => { install_service( @@ -142,13 +165,21 @@ fn main() -> Result<()> { cli.profile.as_deref(), Duration::from_millis(poll_interval_ms), )?; + Ok(ExitCode::SUCCESS) } Command::UninstallService => { uninstall_service(locale, &codex_home)?; + Ok(ExitCode::SUCCESS) } } +} - Ok(()) +fn exit_code_for(status: ReconcileStatus) -> ExitCode { + match status { + ReconcileStatus::Full => ExitCode::SUCCESS, + ReconcileStatus::Partial => ExitCode::from(2), + ReconcileStatus::Failed => ExitCode::FAILURE, + } } fn resolve_codex_home(cli_codex_home: Option) -> Result { diff --git a/src/output.rs b/src/output.rs index 31b57f4..07ecc9c 100644 --- a/src/output.rs +++ b/src/output.rs @@ -7,8 +7,11 @@ use crate::rollout::BucketPrepareSummary; use crate::service; 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; #[derive(Clone, Copy, Debug)] pub(crate) struct RolloutProgressSnapshot { @@ -185,6 +188,92 @@ pub(crate) fn print_sync_summary(locale: Locale, title: &str, summary: &Reconcil } } +pub(crate) fn print_multi_sync_summary( + locale: Locale, + title: &str, + summary: &MultiReconcileSummary, +) { + println!("{title}"); + println!( + "{}: {}", + status_target_provider_label(locale), + summary.provider + ); + 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!(); + println!("{}", sync_stores_heading(locale)); + for store in &summary.stores { + println!(); + println!(" [{}] {}", store.kind.slug(), store.kind.label(locale)); + println!( + " {}: {}", + status_sqlite_file_label(locale), + store.db_path.display() + ); + match &store.outcome { + StoreOutcome::Updated { + changed_rows, + total_rows, + backup_path, + } => { + println!(" {}: {}", sync_rows_updated_label(locale), changed_rows); + println!(" {}: {}", status_total_threads_label(locale), total_rows); + if let Some(backup_path) = backup_path { + println!( + " {}: {}", + sync_backup_label(locale), + backup_path.display() + ); + } + } + StoreOutcome::Failed { error } => { + println!(" {}: {}", sync_store_failed_label(locale), error); + } + } + } + + println!(); + if let Some(journal_path) = &summary.rollout_journal_path { + println!( + "{}: {}", + sync_rollout_journal_label(locale), + journal_path.display() + ); + } + println!( + "{}: {} ms", + sync_elapsed_label(locale), + summary.elapsed.as_millis() + ); + println!("{}", reconcile_status_line(locale, summary.status())); +} + pub(crate) fn print_bucket_prepare_summary(locale: Locale, summary: &BucketPrepareSummary) { println!("{}", bucket_prepare_complete_title(locale)); println!( @@ -838,6 +927,50 @@ pub(crate) fn status_split_note(locale: Locale) -> &'static str { } } +pub(crate) fn sync_stores_heading(locale: Locale) -> &'static str { + match locale { + Locale::En => "Stores:", + Locale::ZhHans => "存储面：", + } +} + +pub(crate) fn sync_store_failed_label(locale: Locale) -> &'static str { + match locale { + Locale::En => "Failed", + Locale::ZhHans => "失败", + } +} + +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() + } + (ReconcileStatus::Partial, Locale::ZhHans) => { + "结果：部分成功 —— 部分库已更新，至少一个失败（见上）。解决后请重跑。".to_string() + } + (ReconcileStatus::Failed, Locale::En) => { + "Result: FAILED — no store could be updated.".to_string() + } + (ReconcileStatus::Failed, Locale::ZhHans) => { + "结果：失败 —— 没有任何库被更新。".to_string() + } + } +} + +pub(crate) fn sqlite_only_app_warning(locale: Locale) -> &'static str { + match locale { + Locale::En => { + "Warning: --sqlite-only edits to the Codex App store take effect immediately but may be reverted by Codex's startup backfill, because the rollout JSONL is the source of truth. Run a full sync (without --sqlite-only) to persist the change." + } + Locale::ZhHans => { + "警告：--sqlite-only 对 Codex App 库的改动会立即生效，但可能被 Codex 启动时的 backfill 从 rollout 还原（rollout 才是事实源）。要持久化请运行完整 sync（不加 --sqlite-only）。" + } + } +} + pub(crate) fn status_background_service_heading(locale: Locale) -> &'static str { match locale { Locale::En => "Background service:", diff --git a/src/rollout.rs b/src/rollout.rs index 8372ae7..7dbd99e 100644 --- a/src/rollout.rs +++ b/src/rollout.rs @@ -5,6 +5,7 @@ use filetime::set_file_times; use rusqlite::Connection; use serde_json::Value; use serde_json::json; +use std::collections::HashSet; use std::fs; use std::fs::File; use std::fs::OpenOptions; @@ -21,7 +22,6 @@ use std::time::Duration; use std::time::Instant; use crate::codex_config::DEFAULT_PROVIDER; -use crate::codex_config::resolve_sqlite_path; use crate::fs_sync::sync_dir; use crate::fs_sync::with_threadripper_lock; use crate::locale::Locale; @@ -29,6 +29,7 @@ use crate::output::RolloutProgressSnapshot; use crate::output::rollout_progress_message; use crate::state_db::ensure_sqlite_exists; use crate::state_db::unix_timestamp_millis; +use crate::stores::discover_stores; const ROLLOUT_PROGRESS_INTERVAL: Duration = Duration::from_millis(500); @@ -184,6 +185,81 @@ pub(crate) fn reconcile_rollout_metadata_from_sqlite_with_progress( ) } +/// Result of a multi-store rollout reconcile: the aggregate rewrite summary plus +/// the stores whose rollout targets could not be read. +#[derive(Debug, Default)] +pub(crate) struct MultiStoreRolloutOutcome { + pub(crate) summary: RolloutReconcileSummary, + /// `(db_path, error)` for each store whose rollout targets failed to load. + /// Callers must treat these stores as failed and skip writing their SQLite, + /// otherwise the DB could be flipped while its rollouts stayed stale. + pub(crate) failed_stores: Vec<(PathBuf, String)>, +} + +/// Reconcile rollout metadata across multiple store DBs. +/// +/// CLI and App surfaces share `CODEX_HOME/sessions`, so the same rollout JSONL +/// can be referenced by more than one `state_5.sqlite`. Collect targets from +/// every store, de-duplicate by canonical rollout path, and rewrite each file's +/// provider exactly once (the rewrite is idempotent, but de-duping avoids +/// re-reading and double-counting shared files). +/// +/// A store whose targets can't be read is recorded in `failed_stores` rather +/// than aborting the whole rewrite: healthy stores are still reconciled, and the +/// caller must skip the failed store's SQLite write so we never flip a DB whose +/// rollouts were left untouched. +pub(crate) fn reconcile_rollouts_for_stores( + store_db_paths: &[PathBuf], + provider: &str, + scope: RolloutScope, + journal_path: Option<&Path>, + padding_bytes: usize, + progress: Option, +) -> Result { + if scope == RolloutScope::None { + return Ok(MultiStoreRolloutOutcome::default()); + } + let (targets, failed_stores) = rollout_targets_for_store_paths(store_db_paths, provider, scope); + let summary = reconcile_rollout_metadata_files( + targets.as_slice(), + provider, + journal_path, + padding_bytes, + progress, + )?; + Ok(MultiStoreRolloutOutcome { + summary, + failed_stores, + }) +} + +fn rollout_targets_for_store_paths( + store_db_paths: &[PathBuf], + provider: &str, + scope: RolloutScope, +) -> (Vec, Vec<(PathBuf, String)>) { + let mut seen: HashSet = HashSet::new(); + let mut targets: Vec = Vec::new(); + let mut failed_stores: Vec<(PathBuf, String)> = Vec::new(); + for db_path in store_db_paths { + match rollout_targets_for_scope(db_path, provider, scope) { + Ok(store_targets) => { + for target in store_targets { + let key = target + .path + .canonicalize() + .unwrap_or_else(|_| target.path.clone()); + if seen.insert(key) { + targets.push(target); + } + } + } + Err(error) => failed_stores.push((db_path.clone(), error.to_string())), + } + } + (targets, failed_stores) +} + fn rollout_targets_for_scope( sqlite_path: &Path, provider: &str, @@ -337,9 +413,27 @@ fn prepare_bucket_padding_unlocked( profile_override: Option<&str>, padding_bytes: usize, ) -> Result { - let sqlite_path = resolve_sqlite_path(codex_home, profile_override)?; let started = Instant::now(); - let targets = rollout_targets_for_scope(&sqlite_path, DEFAULT_PROVIDER, RolloutScope::AllRows)?; + let store_db_paths = discover_stores(codex_home, profile_override)? + .into_iter() + .map(|store| store.db_path) + .collect::>(); + if store_db_paths.is_empty() { + anyhow::bail!( + "no Codex state database found for bucket prepare under {}", + codex_home.display() + ); + } + let (targets, failed_stores) = + rollout_targets_for_store_paths(&store_db_paths, DEFAULT_PROVIDER, RolloutScope::AllRows); + if !failed_stores.is_empty() { + let details = failed_stores + .into_iter() + .map(|(path, error)| format!("{}: {error}", path.display())) + .collect::>() + .join("; "); + anyhow::bail!("failed to read rollout targets for bucket prepare: {details}"); + } let journal_path = codex_home .join("backups") .join(format!("bucket-prepare.{}.jsonl", unix_timestamp_millis()?)); diff --git a/src/state_db.rs b/src/state_db.rs index 6cc782b..048a552 100644 --- a/src/state_db.rs +++ b/src/state_db.rs @@ -103,17 +103,23 @@ pub(crate) fn reconcile_sqlite_with_backup( sqlite_path: &Path, provider: &str, ) -> Result<(u64, u64, PathBuf)> { - let backup_path = create_sqlite_backup_file(sqlite_path)?; - let (changed_rows, total_rows) = reconcile_sqlite_in_place(sqlite_path, provider)?; - Ok((changed_rows, total_rows, backup_path)) -} - -pub(crate) fn create_sqlite_backup_file(sqlite_path: &Path) -> Result { let backups_dir = sqlite_path .parent() .unwrap_or_else(|| Path::new(".")) .join("backups"); - fs::create_dir_all(&backups_dir) + let backup_path = create_sqlite_backup_file_in(sqlite_path, &backups_dir)?; + let (changed_rows, total_rows) = reconcile_sqlite_in_place(sqlite_path, provider)?; + Ok((changed_rows, total_rows, backup_path)) +} + +/// Back up `sqlite_path` into an explicit `backups_dir`. Multi-store sync uses +/// this with a per-store namespaced directory (`/backups//`) +/// so concurrent surfaces never clobber each other's backups. +pub(crate) fn create_sqlite_backup_file_in( + sqlite_path: &Path, + backups_dir: &Path, +) -> Result { + fs::create_dir_all(backups_dir) .with_context(|| format!("failed to create {}", backups_dir.display()))?; let timestamp = unix_timestamp_millis()?; @@ -133,7 +139,7 @@ pub(crate) fn create_sqlite_backup_file(sqlite_path: &Path) -> Result { create_sqlite_backup(sqlite_path, &backup_temp_path)?; fs::rename(&backup_temp_path, &backup_path) .with_context(|| format!("failed to finalize {}", backup_path.display()))?; - sync_dir(&backups_dir)?; + sync_dir(backups_dir)?; Ok(backup_path) } diff --git a/src/sync.rs b/src/sync.rs index 953016b..67444d7 100644 --- a/src/sync.rs +++ b/src/sync.rs @@ -1,4 +1,5 @@ use anyhow::Result; +use std::collections::HashMap; use std::path::Path; use std::path::PathBuf; use std::time::Duration; @@ -13,14 +14,17 @@ 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; use crate::state_db::ProviderDistribution; -use crate::state_db::create_sqlite_backup_file; +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::reconcile_sqlite_in_place; +use crate::state_db::unix_timestamp_millis; use crate::stores::StoreKind; +use crate::stores::StoreTarget; use crate::stores::discover_stores; use crate::stores::no_store_found_message; @@ -130,132 +134,318 @@ pub(crate) fn collect_status( }) } -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, - ) +/// Status of a multi-store reconcile run, mapped to a process exit code. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +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. + Partial, + /// No store could be updated. Exit code 1. + Failed, } -fn reconcile_once_with_progress( +#[derive(Debug)] +pub(crate) enum StoreOutcome { + Updated { + changed_rows: u64, + total_rows: u64, + backup_path: Option, + }, + Failed { + error: String, + }, +} + +#[derive(Debug)] +pub(crate) struct StoreReconcileResult { + pub(crate) kind: StoreKind, + pub(crate) db_path: PathBuf, + pub(crate) outcome: StoreOutcome, +} + +#[derive(Debug)] +pub(crate) struct MultiReconcileSummary { + pub(crate) provider: String, + pub(crate) stores: Vec, + pub(crate) changed_rollouts: u64, + pub(crate) checked_rollouts: u64, + pub(crate) prepared_rollouts: u64, + pub(crate) skipped_rollouts: u64, + pub(crate) rollout_journal_path: Option, + pub(crate) elapsed: Duration, +} + +impl MultiReconcileSummary { + pub(crate) fn status(&self) -> ReconcileStatus { + let updated = self + .stores + .iter() + .filter(|store| matches!(store.outcome, StoreOutcome::Updated { .. })) + .count(); + if updated == self.stores.len() { + ReconcileStatus::Full + } else if updated == 0 { + ReconcileStatus::Failed + } else { + ReconcileStatus::Partial + } + } + + /// True when any selected store is the Codex App surface — 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 { + 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) + } +} + +/// 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. +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, progress: Option, -) -> Result { +) -> Result { with_threadripper_lock(codex_home, || { - reconcile_once_with_progress_unlocked( + reconcile_all_stores_with_backup_unlocked( codex_home, provider_override, profile_override, rollout_scope, + padding_bytes, progress, ) }) } -fn reconcile_once_with_progress_unlocked( +fn reconcile_all_stores_with_backup_unlocked( codex_home: &Path, provider_override: Option<&str>, profile_override: Option<&str>, rollout_scope: RolloutScope, + padding_bytes: usize, progress: Option, -) -> Result { +) -> 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 targets = discover_stores(codex_home, profile_override)?; + if targets.is_empty() { + anyhow::bail!(no_store_found_message(detect_locale(), codex_home)); + } let started = Instant::now(); - let mut rollout_summary = reconcile_rollout_metadata_from_sqlite_with_progress( - &sqlite_path, - codex_home, + + // 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. + 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 rollout_scope != RolloutScope::None && !backup_failed.is_empty() { + let stores = targets + .iter() + .map(|target| { + let error = backup_failed + .get(&target.db_path) + .cloned() + .unwrap_or_else(|| { + "skipped because another store could not be backed up before rollout rewrite" + .to_string() + }); + StoreReconcileResult { + kind: target.kind, + db_path: target.db_path.clone(), + outcome: StoreOutcome::Failed { error }, + } + }) + .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(), + }); + } + + // 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 + .iter() + .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()?)); + let rollout_outcome = reconcile_rollouts_for_stores( + store_db_paths.as_slice(), provider.as_str(), rollout_scope, - None, - DEFAULT_BUCKET_PADDING_BYTES, + Some(rollout_journal_path.as_path()), + 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); - } + let rollout_summary = rollout_outcome.summary; + let rollout_failed: HashMap = + rollout_outcome.failed_stores.into_iter().collect(); - Ok(ReconcileSummary { + // 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. + let stores: Vec = targets + .iter() + .map(|target| { + if let Some(error) = backup_failed.get(&target.db_path) { + return 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 { + kind: target.kind, + db_path: target.db_path.clone(), + outcome: StoreOutcome::Failed { + error: error.clone(), + }, + }; + } + 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(); + + Ok(MultiReconcileSummary { provider, - changed_rows, - total_rows, + stores, 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, + elapsed: started.elapsed(), }) } -pub(crate) fn reconcile_once_with_backup_progress( +fn create_store_backup(target: &StoreTarget) -> Result { + let backups_dir = target + .db_path + .parent() + .unwrap_or_else(|| Path::new(".")) + .join("backups") + .join(target.kind.slug()); + create_sqlite_backup_file_in(&target.db_path, &backups_dir) +} + +fn reconcile_single_store( + target: &StoreTarget, + provider: &str, + backup_path: PathBuf, +) -> 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), + }, + Err(error) => StoreOutcome::Failed { + error: error.to_string(), + }, + }; + StoreReconcileResult { + kind: target.kind, + db_path: target.db_path.clone(), + outcome, + } +} + +fn reconcile_single_store_inner( + target: &StoreTarget, + provider: &str, + backup_path: PathBuf, +) -> Result<(u64, u64, PathBuf)> { + 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, - progress: Option, ) -> Result { - reconcile_once_with_backup_and_padding( + reconcile_once_with_progress( codex_home, provider_override, profile_override, rollout_scope, - DEFAULT_BUCKET_PADDING_BYTES, - progress, + None, ) } -pub(crate) fn reconcile_once_with_backup_and_padding( +fn reconcile_once_with_progress( codex_home: &Path, provider_override: Option<&str>, profile_override: Option<&str>, rollout_scope: RolloutScope, - padding_bytes: usize, progress: Option, ) -> Result { with_threadripper_lock(codex_home, || { - reconcile_once_with_backup_and_padding_unlocked( + reconcile_once_with_progress_unlocked( codex_home, provider_override, profile_override, rollout_scope, - padding_bytes, progress, ) }) } -fn reconcile_once_with_backup_and_padding_unlocked( +fn reconcile_once_with_progress_unlocked( codex_home: &Path, provider_override: Option<&str>, profile_override: Option<&str>, rollout_scope: RolloutScope, - padding_bytes: usize, progress: Option, ) -> Result { let provider = match provider_override { @@ -264,28 +454,28 @@ fn reconcile_once_with_backup_and_padding_unlocked( }; let sqlite_path = resolve_sqlite_path(codex_home, profile_override)?; let started = Instant::now(); - let backup_path = create_sqlite_backup_file(&sqlite_path)?; - let rollout_journal_path = - backup_path - .parent() - .unwrap_or_else(|| Path::new(".")) - .join(format!( - "rollouts.{}.jsonl", - backup_path - .file_name() - .and_then(|name| name.to_str()) - .unwrap_or("state-db.bak") - )); - let rollout_summary = reconcile_rollout_metadata_from_sqlite_with_progress( + let mut rollout_summary = reconcile_rollout_metadata_from_sqlite_with_progress( &sqlite_path, codex_home, provider.as_str(), rollout_scope, - Some(rollout_journal_path.as_path()), - padding_bytes, + 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, @@ -296,7 +486,7 @@ fn reconcile_once_with_backup_and_padding_unlocked( prepared_rollouts: rollout_summary.prepared_files, skipped_rollouts: rollout_summary.skipped_files, elapsed: started.elapsed(), - backup_path: Some(backup_path), + backup_path: None, rollout_journal_path: rollout_summary.journal_path, }) } diff --git a/src/tests.rs b/src/tests.rs index 7107312..4a9cfae 100644 --- a/src/tests.rs +++ b/src/tests.rs @@ -15,6 +15,7 @@ use crate::locale::parse_apple_languages; use crate::locale::parse_locale_tag; use crate::output::install_next_steps; use crate::rollout::RolloutScope; +use crate::rollout::prepare_bucket_padding; use crate::rollout::reconcile_rollout_metadata_from_sqlite_with_progress; use crate::service::ServiceManager; use crate::state_db::inspect_sqlite_distribution; @@ -22,7 +23,10 @@ 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::ReconcileStatus; +use crate::sync::StoreOutcome; use crate::sync::collect_status; +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; @@ -479,7 +483,7 @@ fn status_reports_missing_configured_store_alongside_default_store() -> Result<( fn status_errors_when_every_store_is_broken() -> Result<()> { let dir = tempfile::tempdir()?; let home = dir.path(); - isolate_process_sqlite_home(home)?; + fs::write(home.join("config.toml"), sqlite_home_config(home))?; fs::write(home.join("state_5.sqlite"), b"not a sqlite database")?; let err = collect_status(home, Some("openai"), None).unwrap_err(); @@ -488,6 +492,300 @@ fn status_errors_when_every_store_is_broken() -> Result<()> { Ok(()) } +#[test] +fn reconcile_all_stores_updates_every_surface_and_dedupes_rollout() -> Result<()> { + let dir = tempfile::tempdir()?; + let codex_home = dir.path(); + isolate_process_sqlite_home(codex_home)?; + + // A single rollout JSONL 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_sqlite(db)?; + let connection = Connection::open(db)?; + connection.execute( + "UPDATE threads SET rollout_path = ?1, model_provider = 'cong' WHERE id = '1'", + [rollout_path.display().to_string()], + )?; + } + + let summary = reconcile_all_stores_with_backup( + codex_home, + Some("openai"), + None, + RolloutScope::AllRows, + DEFAULT_BUCKET_PADDING_BYTES, + None, + )?; + + assert_eq!(summary.status(), ReconcileStatus::Full); + let kinds: Vec = summary.stores.iter().map(|store| store.kind).collect(); + assert!(kinds.contains(&StoreKind::App)); + assert!(kinds.contains(&StoreKind::Cli)); + for store in &summary.stores { + assert!(matches!(store.outcome, StoreOutcome::Updated { .. })); + } + + // Both DBs now report the target provider for thread "1". + for db in [&cli_db, &app_db] { + let connection = Connection::open(db)?; + let provider: String = connection.query_row( + "SELECT model_provider FROM threads WHERE id = '1'", + [], + |row| row.get(0), + )?; + assert_eq!(provider, "openai"); + } + + // The shared rollout was rewritten exactly once (deduped across stores). + assert_eq!(summary.changed_rollouts, 1); + let rewritten = fs::read_to_string(&rollout_path)?; + assert!(rewritten.contains("\"model_provider\":\"openai\"")); + + // Backups are namespaced per store. + assert!(codex_home.join("backups/cli").exists()); + assert!(codex_home.join("sqlite/backups/app").exists()); + Ok(()) +} + +#[test] +fn reconcile_all_stores_reports_partial_when_a_store_is_unreadable() -> Result<()> { + let dir = tempfile::tempdir()?; + let codex_home = dir.path(); + isolate_process_sqlite_home(codex_home)?; + + // Healthy CLI store with a mismatched provider. + let cli_db = codex_home.join("state_5.sqlite"); + seed_sqlite(&cli_db)?; + { + let connection = Connection::open(&cli_db)?; + connection.execute( + "UPDATE threads SET model_provider = 'cong' WHERE id = '1'", + [], + )?; + } + // App store exists but is not a valid SQLite database. + let app_db = codex_home.join("sqlite").join("state_5.sqlite"); + fs::create_dir_all(app_db.parent().unwrap())?; + fs::write(&app_db, b"not a sqlite database")?; + + let summary = reconcile_all_stores_with_backup( + codex_home, + Some("openai"), + None, + RolloutScope::None, + DEFAULT_BUCKET_PADDING_BYTES, + 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::Failed { .. })); + + // The healthy store was still updated despite the broken one. + let connection = Connection::open(&cli_db)?; + let provider: String = connection.query_row( + "SELECT model_provider FROM threads WHERE id = '1'", + [], + |row| row.get(0), + )?; + assert_eq!(provider, "openai"); + Ok(()) +} + +#[test] +fn sqlite_only_warning_detects_configured_app_alias() -> Result<()> { + let dir = tempfile::tempdir()?; + let codex_home = dir.path(); + let app_home = codex_home.join("sqlite"); + fs::create_dir_all(&app_home)?; + fs::write( + codex_home.join("config.toml"), + sqlite_home_config(&app_home), + )?; + seed_sqlite(&app_home.join("state_5.sqlite"))?; + + let summary = reconcile_all_stores_with_backup( + codex_home, + Some("openai"), + None, + RolloutScope::None, + DEFAULT_BUCKET_PADDING_BYTES, + 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)); + Ok(()) +} + +#[test] +fn bucket_prepare_checks_rollouts_from_all_stores() -> Result<()> { + let dir = tempfile::tempdir()?; + let codex_home = dir.path(); + fs::write( + codex_home.join("config.toml"), + sqlite_home_config(codex_home), + )?; + + let cli_rollout = codex_home.join("sessions/2026/05/07/rollout-cli.jsonl"); + let app_rollout = codex_home.join("sessions/2026/05/07/rollout-app.jsonl"); + fs::create_dir_all(cli_rollout.parent().unwrap())?; + fs::write( + &cli_rollout, + "{\"type\":\"session_meta\",\"payload\":{\"id\":\"1\",\"model_provider\":\"cong\"}}\n", + )?; + fs::write( + &app_rollout, + "{\"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, rollout) in [(&cli_db, &cli_rollout), (&app_db, &app_rollout)] { + seed_sqlite(db)?; + let connection = Connection::open(db)?; + connection.execute( + "UPDATE threads SET rollout_path = ?1 WHERE id = '1'", + [rollout.display().to_string()], + )?; + } + + let summary = prepare_bucket_padding(codex_home, None, DEFAULT_BUCKET_PADDING_BYTES)?; + + assert_eq!(summary.checked_rollouts, 2); + Ok(()) +} + +#[cfg(unix)] +#[test] +fn reconcile_all_stores_skips_rollout_when_store_backup_fails() -> Result<()> { + use std::os::unix::fs::PermissionsExt; + + 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"); + 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_sqlite(db)?; + let connection = Connection::open(db)?; + connection.execute( + "UPDATE threads SET rollout_path = ?1, model_provider = 'cong' WHERE id = '1'", + [rollout_path.display().to_string()], + )?; + } + + let app_dir = app_db.parent().unwrap(); + let original_mode = fs::metadata(app_dir)?.permissions().mode(); + fs::set_permissions(app_dir, fs::Permissions::from_mode(0o500))?; + let summary = reconcile_all_stores_with_backup( + codex_home, + Some("openai"), + None, + RolloutScope::AllRows, + DEFAULT_BUCKET_PADDING_BYTES, + None, + ); + fs::set_permissions(app_dir, fs::Permissions::from_mode(original_mode))?; + let summary = summary?; + + assert_eq!(summary.status(), ReconcileStatus::Failed); + assert_eq!(summary.changed_rollouts, 0); + assert!(fs::read_to_string(&rollout_path)?.contains("\"model_provider\":\"cong\"")); + for db in [&cli_db, &app_db] { + let connection = Connection::open(db)?; + let provider: String = connection.query_row( + "SELECT model_provider FROM threads WHERE id = '1'", + [], + |row| row.get(0), + )?; + assert_eq!(provider, "cong"); + } + Ok(()) +} + +#[test] +fn reconcile_all_stores_fails_store_when_rollout_targets_unreadable() -> Result<()> { + let dir = tempfile::tempdir()?; + let codex_home = dir.path(); + fs::write( + codex_home.join("config.toml"), + sqlite_home_config(codex_home), + )?; + + // A valid SQLite DB whose `threads` table has `model_provider` but no + // `rollout_path` column: `UPDATE model_provider` would succeed, yet rollout + // target collection fails. The store must be reported Failed (not silently + // Updated), and its DB must be left untouched. + let cli_db = codex_home.join("state_5.sqlite"); + { + let connection = Connection::open(&cli_db)?; + connection.execute( + "CREATE TABLE threads (id TEXT PRIMARY KEY, model_provider TEXT NOT NULL)", + [], + )?; + connection.execute( + "INSERT INTO threads (id, model_provider) VALUES ('1', 'cong')", + [], + )?; + } + + let summary = reconcile_all_stores_with_backup( + codex_home, + Some("openai"), + None, + RolloutScope::AllRows, + DEFAULT_BUCKET_PADDING_BYTES, + None, + )?; + + assert_eq!(summary.status(), ReconcileStatus::Failed); + assert!(matches!( + summary.stores[0].outcome, + StoreOutcome::Failed { .. } + )); + let connection = Connection::open(&cli_db)?; + let provider: String = connection.query_row( + "SELECT model_provider FROM threads WHERE id = '1'", + [], + |row| row.get(0), + )?; + assert_eq!(provider, "cong"); + Ok(()) +} + #[test] fn rejects_blank_provider_override() { let err = validate_provider_override(Locale::En, Some(" ")).unwrap_err(); @@ -1097,9 +1395,18 @@ fn sqlite_home_config(path: &Path) -> String { } fn isolate_process_sqlite_home(codex_home: &Path) -> Result<()> { + let configured_home = codex_home.join(".threadripper-test-configured"); + fs::create_dir_all(&configured_home)?; + let configured_db = configured_home.join("state_5.sqlite"); + seed_sqlite(&configured_db)?; + let connection = Connection::open(&configured_db)?; + connection.execute( + "UPDATE threads SET model_provider = 'openai', rollout_path = ''", + [], + )?; fs::write( codex_home.join("config.toml"), - "model_provider = \"openai\"\nsqlite_home = \".threadripper-test-missing\"\n", + "model_provider = \"openai\"\nsqlite_home = \".threadripper-test-configured\"\n", )?; Ok(()) } From b6dfe0bdc51de36276276c297894122e0cd9cd07 Mon Sep 17 00:00:00 2001 From: wangnov <48670012+Wangnov@users.noreply.github.com> Date: Sat, 13 Jun 2026 11:55:23 +0800 Subject: [PATCH 2/2] feat: skip stores during an in-progress Codex backfill PR3 of the multi-store adaptation (builds on #17). Before writing, the multi-store path now waits (bounded) for each store's Codex startup-backfill to finish, so threadripper never races Codex's rebuild. - wait_for_store_backfill polls backfill_state (read-only): no table or a `complete` status is ready immediately; otherwise it waits up to backfill_wait (default 10s) and then reports the store busy. - A busy store is reported StoreOutcome::Skipped, and neither its rollout targets nor its DB are touched. - Because rewriting the shared rollout JSONL would race a running backfill's reads (the rollout files are its source of truth), a rollout-rewriting scope (AllRows) combined with any in-progress backfill skips the whole round; --sqlite-only (RolloutScope::None) still writes its ready stores. - main returns Partial(2) / Failed(1) accordingly; the --sqlite-only App warning now fires only when the App store was actually updated. Reviewed in parallel by Codex and a Claude subagent. Codex caught that the first cut still rewrote the shared rollout JSONL (racing the backfill) even while skipping busy stores' DBs; fixed with the whole-round skip above, plus a regression test asserting that a shared rollout and both DBs stay untouched while a backfill runs. 53 tests, clippy and fmt clean. --- src/main.rs | 5 +- src/output.rs | 18 ++++- src/sync.rs | 175 ++++++++++++++++++++++++++++++++++++-------- src/tests.rs | 196 +++++++++++++++++++++++++++++++++++++++++++++++++- 4 files changed, 360 insertions(+), 34 deletions(-) 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..03d926c 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,195 @@ 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();