diff --git a/README.md b/README.md index cd099f7..ecb4d4a 100644 --- a/README.md +++ b/README.md @@ -103,6 +103,16 @@ codex-threadripper install-service - `status`：查看当前 provider、SQLite 里各 provider 的线程分布，以及后台服务的运行状态 - `sync`：立即执行一次收敛。执行前会在 SQLite 状态库旁边的 `backups/` 目录里写一份带时间戳的备份；如果 rollout 首行需要改写，也会写 compact journal +- `sync --dry-run`：预览将要变更的内容（SQLite 行数和 rollout 文件数），不实际写入任何数据 +- `restore`：从备份恢复 SQLite 状态库 + - 不带参数：列出所有可用备份并退出 + - `restore `：从指定备份文件恢复 + - `restore --latest`：恢复最近一次备份（按文件名时间戳） + - `--force`：跳过确认提示 +- `prune-backups`：删除旧备份，保留最近的 N 个（默认 5） + - `--keep N`：指定保留数量 + - `--force`：跳过确认提示 + - 无法解析时间戳的 `.bak` 文件会被跳过并发出警告 - `bucket switch `：把全部历史切到指定 provider 可见桶；首行有 padding 时只做原地 patch - `bucket prepare`：给 rollout 首行补足 JSON 空白 padding，让后续 provider 桶切换保持快速 - `watch`：持续监听 `config.toml` 的变化，同时定时收敛新增的线程记录 @@ -220,6 +230,16 @@ codex-threadripper install-service - `status`: show the active provider, the per-provider thread counts from SQLite, and whether the background service is running - `sync`: reconcile once right now, writing a timestamped backup to the `backups/` directory beside the SQLite state DB; rollout first-line changes also get a compact journal +- `sync --dry-run`: preview what would change (SQLite rows and rollout files) without writing anything +- `restore`: restore the SQLite state DB from a backup file + - No argument: list all available backups and exit + - `restore `: restore from the specified backup file + - `restore --latest`: restore the most recent backup (by filename timestamp) + - `--force`: skip the confirmation prompt +- `prune-backups`: remove old backups, keeping the N most recent (default 5) + - `--keep N`: number of backups to keep + - `--force`: skip the confirmation prompt + - `.bak` files with unparseable timestamps are skipped with a warning - `bucket switch `: move all history into the requested provider visibility bucket; prepared first lines are patched in place - `bucket prepare`: reserve JSON whitespace padding in rollout first lines so future provider bucket switches stay fast - `watch`: keep watching `config.toml` for provider changes and periodically reconcile any newly added thread rows diff --git a/src/cli.rs b/src/cli.rs index 1230156..266881a 100644 --- a/src/cli.rs +++ b/src/cli.rs @@ -48,6 +48,8 @@ pub(crate) enum Command { Sync { #[arg(long, help = "placeholder")] sqlite_only: bool, + #[arg(long, help = "placeholder")] + dry_run: bool, }, /// placeholder Bucket { @@ -92,6 +94,27 @@ pub(crate) enum Command { /// placeholder #[command(name = "uninstall-service", alias = "uninstall-launchd")] UninstallService, + /// placeholder + Restore { + #[arg( + value_name = "BACKUP_PATH", + help = "placeholder", + conflicts_with = "latest" + )] + backup_path: Option, + #[arg(long, help = "placeholder")] + latest: bool, + #[arg(long, help = "placeholder")] + force: bool, + }, + /// placeholder + #[command(name = "prune-backups")] + PruneBackups { + #[arg(long, default_value_t = 5, value_name = "N", help = "placeholder")] + keep: usize, + #[arg(long, help = "placeholder")] + force: bool, + }, } #[derive(Subcommand, Debug)] @@ -184,6 +207,10 @@ pub(crate) fn localized_command(locale: Locale) -> clap::Command { arg.help(sqlite_only_help(locale)) .help_heading(options_heading(locale)) }) + .mut_arg("dry_run", |arg| { + arg.help(dry_run_help(locale)) + .help_heading(options_heading(locale)) + }) }); command = command.mut_subcommand("bucket", |sub| { sub.about(bucket_about(locale)) @@ -274,6 +301,42 @@ pub(crate) fn localized_command(locale: Locale) -> clap::Command { .disable_help_subcommand(true) .help_template(help_template(locale)) }); + command = command.mut_subcommand("restore", |sub| { + sub.about(restore_about(locale)) + .version(APP_VERSION) + .disable_help_flag(true) + .disable_version_flag(true) + .disable_help_subcommand(true) + .help_template(help_template(locale)) + .mut_arg("backup_path", |arg| { + arg.help(restore_path_help(locale)) + .value_name(restore_path_value_name(locale)) + }) + .mut_arg("latest", |arg| { + arg.help(restore_latest_help(locale)) + .help_heading(options_heading(locale)) + }) + .mut_arg("force", |arg| { + arg.help(restore_force_help(locale)) + .help_heading(options_heading(locale)) + }) + }); + command = command.mut_subcommand("prune-backups", |sub| { + sub.about(prune_backups_about(locale)) + .version(APP_VERSION) + .disable_help_flag(true) + .disable_version_flag(true) + .disable_help_subcommand(true) + .help_template(help_template(locale)) + .mut_arg("keep", |arg| { + arg.help(prune_keep_help(locale)) + .help_heading(options_heading(locale)) + }) + .mut_arg("force", |arg| { + arg.help(prune_force_help(locale)) + .help_heading(options_heading(locale)) + }) + }); command } diff --git a/src/main.rs b/src/main.rs index 224b542..ab5c2b0 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,5 +1,6 @@ use anyhow::Context; use anyhow::Result; +use std::io::Write; use std::path::Path; use std::path::PathBuf; use std::time::Duration; @@ -32,16 +33,30 @@ use output::launchd_plist_message; use output::next_steps_heading; use output::no_launchd_plist_message; use output::print_bucket_prepare_summary; +use output::print_dry_run_summary; use output::print_install_service_summary; use output::print_status; use output::print_sync_summary; +use output::prune_cancelled; +use output::prune_complete; +use output::prune_confirm_prompt; +use output::prune_skipped_warning; +use output::restore_cancelled; +use output::restore_complete; +use output::restore_confirm_prompt; +use output::restore_list_title; +use output::restore_no_backups; use output::run_status_next_step; use output::sync_complete_title; use output::uninstall_launchd_done; use rollout::RolloutProgressConfig; use rollout::RolloutScope; use rollout::prepare_bucket_padding; +use state_db::list_backups; +use state_db::prune_backups; +use state_db::restore_sqlite_from_backup; use sync::collect_status; +use sync::dry_run_sync; use sync::reconcile_once_with_backup_and_padding; use sync::reconcile_once_with_backup_progress; use watch::run_watch; @@ -59,25 +74,38 @@ fn main() -> Result<()> { collect_status(&codex_home, cli.provider.as_deref(), cli.profile.as_deref())?; print_status(locale, &summary); } - Command::Sync { sqlite_only } => { + Command::Sync { + sqlite_only, + dry_run, + } => { let rollout_scope = if sqlite_only { RolloutScope::None } else { RolloutScope::AllRows }; - let progress = if sqlite_only { - None + if dry_run { + let summary = dry_run_sync( + &codex_home, + cli.provider.as_deref(), + cli.profile.as_deref(), + rollout_scope, + )?; + print_dry_run_summary(locale, &summary); } else { - Some(RolloutProgressConfig { locale }) - }; - let summary = reconcile_once_with_backup_progress( - &codex_home, - cli.provider.as_deref(), - cli.profile.as_deref(), - rollout_scope, - progress, - )?; - print_sync_summary(locale, sync_complete_title(locale), &summary); + let progress = if sqlite_only { + None + } else { + Some(RolloutProgressConfig { locale }) + }; + let summary = reconcile_once_with_backup_progress( + &codex_home, + cli.provider.as_deref(), + cli.profile.as_deref(), + rollout_scope, + progress, + )?; + print_sync_summary(locale, sync_complete_title(locale), &summary); + } } Command::Bucket { command } => match command { BucketCommand::Prepare { padding_bytes } => { @@ -145,6 +173,23 @@ fn main() -> Result<()> { Command::UninstallService => { uninstall_service(locale, &codex_home)?; } + Command::Restore { + backup_path, + latest, + force, + } => { + run_restore( + locale, + &codex_home, + cli.profile.as_deref(), + backup_path, + latest, + force, + )?; + } + Command::PruneBackups { keep, force } => { + run_prune_backups(locale, &codex_home, cli.profile.as_deref(), keep, force)?; + } } Ok(()) @@ -212,3 +257,110 @@ fn uninstall_service(locale: Locale, codex_home: &Path) -> Result<()> { } Ok(()) } + +fn run_restore( + locale: Locale, + codex_home: &Path, + profile_override: Option<&str>, + backup_path: Option, + latest: bool, + force: bool, +) -> Result<()> { + use crate::codex_config::resolve_sqlite_path; + + let sqlite_path = resolve_sqlite_path(codex_home, profile_override)?; + let backups = list_backups(&sqlite_path)?; + + let chosen_path = if let Some(path) = backup_path { + path + } else if latest { + let entry = backups + .iter() + .find(|e| e.timestamp_ms.is_some()) + .ok_or_else(|| anyhow::anyhow!(restore_no_backups(locale)))?; + entry.path.clone() + } else { + if backups.is_empty() { + println!("{}", restore_no_backups(locale)); + return Ok(()); + } + println!("{}", restore_list_title(locale)); + for entry in &backups { + let name = entry + .path + .file_name() + .and_then(|n| n.to_str()) + .unwrap_or("?"); + if entry.timestamp_ms.is_some() { + println!(" {name}"); + } else { + println!(" {name} (unparseable timestamp)"); + } + } + return Ok(()); + }; + + if !force { + print!("{}", restore_confirm_prompt(locale, &chosen_path)); + std::io::stdout().flush()?; + let mut input = String::new(); + std::io::stdin().read_line(&mut input)?; + if !input.trim().eq_ignore_ascii_case("y") { + println!("{}", restore_cancelled(locale)); + return Ok(()); + } + } + + restore_sqlite_from_backup(&sqlite_path, &chosen_path)?; + println!("{}", restore_complete(locale, &chosen_path)); + Ok(()) +} + +fn run_prune_backups( + locale: Locale, + codex_home: &Path, + profile_override: Option<&str>, + keep: usize, + force: bool, +) -> Result<()> { + use crate::codex_config::resolve_sqlite_path; + + let sqlite_path = resolve_sqlite_path(codex_home, profile_override)?; + let backups = list_backups(&sqlite_path)?; + + for entry in &backups { + if entry.timestamp_ms.is_none() { + let name = entry + .path + .file_name() + .and_then(|n| n.to_str()) + .unwrap_or("?"); + eprintln!("{}", prune_skipped_warning(locale, name)); + } + } + + let parseable: Vec<_> = backups + .iter() + .filter(|e| e.timestamp_ms.is_some()) + .collect(); + if parseable.len() <= keep { + println!("{}", prune_complete(locale, 0, parseable.len())); + return Ok(()); + } + + let to_delete_count = parseable.len() - keep; + if !force { + print!("{}", prune_confirm_prompt(locale, to_delete_count)); + std::io::stdout().flush()?; + let mut input = String::new(); + std::io::stdin().read_line(&mut input)?; + if !input.trim().eq_ignore_ascii_case("y") { + println!("{}", prune_cancelled(locale)); + return Ok(()); + } + } + + let (deleted, kept) = prune_backups(&backups, keep)?; + println!("{}", prune_complete(locale, deleted, kept)); + Ok(()) +} diff --git a/src/output.rs b/src/output.rs index 9237537..531e541 100644 --- a/src/output.rs +++ b/src/output.rs @@ -886,3 +886,230 @@ pub(crate) fn service_log_message(locale: Locale, path: &Path) -> String { Locale::ZhHans => format!("日志路径：{}", path.display()), } } + +pub(crate) fn dry_run_help(locale: Locale) -> &'static str { + match locale { + Locale::En => "Preview what would change without writing anything.", + Locale::ZhHans => "预览将要变更的内容，不实际写入。", + } +} + +pub(crate) fn restore_about(locale: Locale) -> &'static str { + match locale { + Locale::En => "Restore the SQLite state DB from a backup file.", + Locale::ZhHans => "从备份文件恢复 SQLite 状态库。", + } +} + +pub(crate) fn restore_path_help(locale: Locale) -> &'static str { + match locale { + Locale::En => "Path to the .bak backup file to restore from.", + Locale::ZhHans => "要恢复的 .bak 备份文件路径。", + } +} + +pub(crate) fn restore_path_value_name(locale: Locale) -> &'static str { + match locale { + Locale::En => "BACKUP_PATH", + Locale::ZhHans => "备份路径", + } +} + +pub(crate) fn restore_latest_help(locale: Locale) -> &'static str { + match locale { + Locale::En => "Restore the most recent backup (by filename timestamp).", + Locale::ZhHans => "恢复最近一次备份（按文件名时间戳）。", + } +} + +pub(crate) fn restore_force_help(locale: Locale) -> &'static str { + match locale { + Locale::En => "Skip confirmation prompt.", + Locale::ZhHans => "跳过确认提示。", + } +} + +pub(crate) fn prune_backups_about(locale: Locale) -> &'static str { + match locale { + Locale::En => "Remove old backups, keeping the N most recent.", + Locale::ZhHans => "删除旧备份，保留最近的 N 个。", + } +} + +pub(crate) fn prune_keep_help(locale: Locale) -> &'static str { + match locale { + Locale::En => "Number of most recent backups to keep.", + Locale::ZhHans => "保留最近备份的数量。", + } +} + +pub(crate) fn prune_force_help(locale: Locale) -> &'static str { + match locale { + Locale::En => "Skip confirmation prompt before deleting.", + Locale::ZhHans => "删除前跳过确认提示。", + } +} + +pub(crate) fn restore_list_title(locale: Locale) -> &'static str { + match locale { + Locale::En => "Available backups (newest first):", + Locale::ZhHans => "可用备份（最新在前）：", + } +} + +pub(crate) fn restore_no_backups(locale: Locale) -> &'static str { + match locale { + Locale::En => "No backups found in the backups directory.", + Locale::ZhHans => "备份目录中没有找到备份文件。", + } +} + +pub(crate) fn restore_confirm_prompt(locale: Locale, path: &Path) -> String { + match locale { + Locale::En => format!( + "This will overwrite the current SQLite state DB with:\n {}\nContinue? [y/N] ", + path.display() + ), + Locale::ZhHans => format!( + "这将用以下备份覆盖当前 SQLite 状态库：\n {}\n继续？[y/N] ", + path.display() + ), + } +} + +pub(crate) fn restore_cancelled(locale: Locale) -> &'static str { + match locale { + Locale::En => "Restore cancelled.", + Locale::ZhHans => "恢复已取消。", + } +} + +pub(crate) fn restore_complete(locale: Locale, path: &Path) -> String { + match locale { + Locale::En => format!("Restored from: {}", path.display()), + Locale::ZhHans => format!("已从备份恢复：{}", path.display()), + } +} + +pub(crate) fn prune_confirm_prompt(locale: Locale, count: usize) -> String { + match locale { + Locale::En => format!("Will delete {count} backup(s). Continue? [y/N] "), + Locale::ZhHans => format!("将删除 {count} 个备份。继续？[y/N] "), + } +} + +pub(crate) fn prune_cancelled(locale: Locale) -> &'static str { + match locale { + Locale::En => "Prune cancelled.", + Locale::ZhHans => "清理已取消。", + } +} + +pub(crate) fn prune_complete(locale: Locale, deleted: usize, kept: usize) -> String { + match locale { + Locale::En => format!("Deleted {deleted} backup(s), kept {kept}."), + Locale::ZhHans => format!("已删除 {deleted} 个备份，保留 {kept} 个。"), + } +} + +pub(crate) fn prune_skipped_warning(locale: Locale, name: &str) -> String { + match locale { + Locale::En => format!(" skipped (unparseable timestamp): {name}"), + Locale::ZhHans => format!(" 已跳过（无法解析时间戳）：{name}"), + } +} + +pub(crate) fn dry_run_title(locale: Locale) -> &'static str { + match locale { + Locale::En => "Dry-run preview", + Locale::ZhHans => "Dry-run 预览", + } +} + +pub(crate) fn dry_run_sqlite_rows_label(locale: Locale) -> &'static str { + match locale { + Locale::En => "Rows that would be updated", + Locale::ZhHans => "将被更新的行数", + } +} + +pub(crate) fn dry_run_rollouts_label(locale: Locale) -> &'static str { + match locale { + Locale::En => "Rollouts that would be updated", + Locale::ZhHans => "将被更新的 rollout 数", + } +} + +pub(crate) fn dry_run_rollouts_prepare_label(locale: Locale) -> &'static str { + match locale { + Locale::En => "Rollouts that would be rewritten (padding)", + Locale::ZhHans => "将被重写的 rollout 数（padding）", + } +} + +pub(crate) fn dry_run_rollouts_skip_label(locale: Locale) -> &'static str { + match locale { + Locale::En => "Rollouts that would be skipped", + Locale::ZhHans => "将被跳过的 rollout 数", + } +} + +pub(crate) fn dry_run_no_changes(locale: Locale) -> &'static str { + match locale { + Locale::En => "No changes needed — everything is already aligned.", + Locale::ZhHans => "无需变更——一切已对齐。", + } +} + +pub(crate) fn dry_run_has_no_changes(summary: &crate::sync::DryRunSummary) -> bool { + summary.mismatched_rows == 0 + && summary.rollout_would_update == 0 + && summary.rollout_would_prepare == 0 + && summary.rollout_would_skip == 0 +} + +pub(crate) fn print_dry_run_summary(locale: Locale, summary: &crate::sync::DryRunSummary) { + println!("{}", dry_run_title(locale)); + println!( + "{}: {}", + status_target_provider_label(locale), + summary.provider + ); + println!( + "{}: {}", + dry_run_sqlite_rows_label(locale), + summary.mismatched_rows + ); + println!( + "{}: {}", + status_total_threads_label(locale), + summary.total_rows + ); + if summary.rollout_would_update > 0 + || summary.rollout_would_prepare > 0 + || summary.rollout_would_skip > 0 + { + println!( + "{}: {}", + dry_run_rollouts_label(locale), + summary.rollout_would_update + ); + if summary.rollout_would_prepare > 0 { + println!( + "{}: {}", + dry_run_rollouts_prepare_label(locale), + summary.rollout_would_prepare + ); + } + if summary.rollout_would_skip > 0 { + println!( + "{}: {}", + dry_run_rollouts_skip_label(locale), + summary.rollout_would_skip + ); + } + } + if dry_run_has_no_changes(summary) { + println!("{}", dry_run_no_changes(locale)); + } +} diff --git a/src/rollout.rs b/src/rollout.rs index 170e195..b0a44d1 100644 --- a/src/rollout.rs +++ b/src/rollout.rs @@ -62,6 +62,51 @@ pub(crate) struct BucketPrepareSummary { pub(crate) journal_path: Option, } +#[derive(Debug, Default)] +pub(crate) struct RolloutDryRunSummary { + pub(crate) would_update: u64, + pub(crate) would_prepare: u64, + pub(crate) would_skip: u64, +} + +pub(crate) fn scan_rollout_metadata_dry_run( + sqlite_path: &Path, + provider: &str, + scope: RolloutScope, + padding_bytes: usize, +) -> Result { + if scope == RolloutScope::None { + return Ok(RolloutDryRunSummary::default()); + } + let targets = rollout_targets_for_scope(sqlite_path, provider, scope)?; + let mut summary = RolloutDryRunSummary::default(); + for target in &targets { + if !target.path.exists() { + continue; + } + let first_line = read_first_line(&target.path)?; + let Some(mut value) = + parse_matching_session_meta(first_line.content.as_slice(), target.thread_id.as_str())? + else { + summary.would_skip += 1; + continue; + }; + if !set_session_meta_provider(&mut value, provider) { + continue; + } + let rendered = serde_json::to_vec(&value) + .with_context(|| format!("failed to render {}", target.path.display()))?; + if rendered.len() <= first_line.content.len() { + summary.would_update += 1; + } else { + summary.would_update += 1; + summary.would_prepare += 1; + let _ = padding_bytes; + } + } + Ok(summary) +} + #[derive(Clone, Copy, Debug)] pub(crate) struct RolloutProgressConfig { pub(crate) locale: Locale, diff --git a/src/state_db.rs b/src/state_db.rs index f0aff25..31c408d 100644 --- a/src/state_db.rs +++ b/src/state_db.rs @@ -1,10 +1,13 @@ use anyhow::Context; use anyhow::Result; use rusqlite::Connection; +use rusqlite::OpenFlags; use rusqlite::TransactionBehavior; use rusqlite::backup::Backup; use rusqlite::backup::StepResult; use std::fs; +#[cfg(unix)] +use std::os::unix::fs::MetadataExt; use std::path::Path; use std::path::PathBuf; use std::time::Duration; @@ -151,6 +154,143 @@ pub(crate) fn ensure_sqlite_exists(sqlite_path: &Path) -> Result<()> { anyhow::bail!(sqlite_missing_error(detect_locale(), sqlite_path)); } +#[derive(Debug)] +pub(crate) struct BackupEntry { + pub(crate) path: PathBuf, + pub(crate) timestamp_ms: Option, +} + +pub(crate) fn list_backups(sqlite_path: &Path) -> Result> { + let backups_dir = sqlite_path + .parent() + .unwrap_or_else(|| Path::new(".")) + .join("backups"); + if !backups_dir.exists() { + return Ok(Vec::new()); + } + let sqlite_name = sqlite_path + .file_name() + .and_then(|name| name.to_str()) + .unwrap_or(STATE_DB_FILENAME); + let backup_prefix = format!("{sqlite_name}."); + let mut entries: Vec = Vec::new(); + for entry in fs::read_dir(&backups_dir) + .with_context(|| format!("failed to read {}", backups_dir.display()))? + { + let entry = entry?; + let path = entry.path(); + if path.extension().and_then(|ext| ext.to_str()) != Some("bak") { + continue; + } + let Some(name) = path.file_name().and_then(|name| name.to_str()) else { + continue; + }; + if !name.starts_with(&backup_prefix) { + continue; + } + let timestamp_ms = parse_backup_timestamp(&path); + entries.push(BackupEntry { path, timestamp_ms }); + } + entries.sort_by_key(|b| std::cmp::Reverse(b.timestamp_ms)); + Ok(entries) +} + +fn parse_backup_timestamp(path: &Path) -> Option { + let name = path.file_stem()?.to_str()?; + let last_dot = name.rfind('.')?; + let ts_str = &name[last_dot + 1..]; + ts_str.parse::().ok() +} + +pub(crate) fn restore_sqlite_from_backup(sqlite_path: &Path, backup_path: &Path) -> Result<()> { + if !backup_path.exists() { + anyhow::bail!("backup file not found: {}", backup_path.display()); + } + if paths_refer_to_same_file(sqlite_path, backup_path)? { + anyhow::bail!( + "backup path points to the live database: {}", + backup_path.display() + ); + } + let source = Connection::open_with_flags(backup_path, OpenFlags::SQLITE_OPEN_READ_ONLY) + .with_context(|| format!("failed to open backup {}", backup_path.display()))?; + source.query_row("SELECT COUNT(*) FROM threads", [], |_| Ok(()))?; + + let mut destination = Connection::open(sqlite_path) + .with_context(|| format!("failed to open {}", sqlite_path.display()))?; + destination.busy_timeout(Duration::from_millis(5_000))?; + let backup = Backup::new(&source, &mut destination)?; + let started = Instant::now(); + let timeout = Duration::from_secs(30); + + loop { + if started.elapsed() >= timeout { + anyhow::bail!( + "sqlite restore timed out after {} seconds for {}", + timeout.as_secs(), + sqlite_path.display() + ); + } + + match backup.step(100)? { + StepResult::Done => break, + StepResult::More => {} + StepResult::Busy | StepResult::Locked => { + std::thread::sleep(Duration::from_millis(50)); + } + _ => {} + } + } + + drop(backup); + drop(destination); + sync_file(sqlite_path)?; + let parent = sqlite_path.parent().unwrap_or_else(|| Path::new(".")); + sync_dir(parent)?; + Ok(()) +} + +fn paths_refer_to_same_file(left: &Path, right: &Path) -> Result { + if !left.exists() || !right.exists() { + return Ok(false); + } + + #[cfg(unix)] + { + let left = + fs::metadata(left).with_context(|| format!("failed to stat {}", left.display()))?; + let right = + fs::metadata(right).with_context(|| format!("failed to stat {}", right.display()))?; + Ok(left.dev() == right.dev() && left.ino() == right.ino()) + } + + #[cfg(not(unix))] + { + let left = fs::canonicalize(left) + .with_context(|| format!("failed to resolve {}", left.display()))?; + let right = fs::canonicalize(right) + .with_context(|| format!("failed to resolve {}", right.display()))?; + Ok(left == right) + } +} + +pub(crate) fn prune_backups(entries: &[BackupEntry], keep: usize) -> Result<(usize, usize)> { + let parseable: Vec<&BackupEntry> = entries + .iter() + .filter(|e| e.timestamp_ms.is_some()) + .collect(); + if parseable.len() <= keep { + return Ok((0, parseable.len())); + } + let to_delete = &parseable[keep..]; + let deleted = to_delete.len(); + for entry in to_delete { + fs::remove_file(&entry.path) + .with_context(|| format!("failed to delete {}", entry.path.display()))?; + } + Ok((deleted, keep)) +} + fn sqlite_missing_error(locale: Locale, path: &Path) -> String { match locale { Locale::En => format!( diff --git a/src/sync.rs b/src/sync.rs index 05874c2..274f7dd 100644 --- a/src/sync.rs +++ b/src/sync.rs @@ -10,6 +10,7 @@ use crate::codex_config::resolve_sqlite_path; use crate::rollout::RolloutProgressConfig; use crate::rollout::RolloutScope; use crate::rollout::reconcile_rollout_metadata_from_sqlite_with_progress; +use crate::rollout::scan_rollout_metadata_dry_run; use crate::service; use crate::service::ServiceStatus as BackgroundServiceStatus; use crate::state_db::ProviderDistribution; @@ -17,6 +18,16 @@ use crate::state_db::create_sqlite_backup_file; use crate::state_db::inspect_sqlite_distribution; use crate::state_db::reconcile_sqlite_in_place; +#[derive(Debug)] +pub(crate) struct DryRunSummary { + pub(crate) provider: String, + pub(crate) total_rows: u64, + pub(crate) mismatched_rows: u64, + pub(crate) rollout_would_update: u64, + pub(crate) rollout_would_prepare: u64, + pub(crate) rollout_would_skip: u64, +} + #[derive(Debug)] pub(crate) struct ReconcileSummary { pub(crate) provider: String, @@ -123,6 +134,35 @@ fn reconcile_once_with_progress( }) } +pub(crate) fn dry_run_sync( + codex_home: &Path, + provider_override: Option<&str>, + profile_override: Option<&str>, + rollout_scope: RolloutScope, +) -> 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 (total_rows, mismatched_rows, _) = + inspect_sqlite_distribution(&sqlite_path, provider.as_str())?; + let rollout_preview = scan_rollout_metadata_dry_run( + &sqlite_path, + provider.as_str(), + rollout_scope, + DEFAULT_BUCKET_PADDING_BYTES, + )?; + Ok(DryRunSummary { + provider, + total_rows, + mismatched_rows, + rollout_would_update: rollout_preview.would_update, + rollout_would_prepare: rollout_preview.would_prepare, + rollout_would_skip: rollout_preview.would_skip, + }) +} + pub(crate) fn reconcile_once_with_backup_progress( codex_home: &Path, provider_override: Option<&str>, diff --git a/src/tests.rs b/src/tests.rs index 94b5aae..e575ea6 100644 --- a/src/tests.rs +++ b/src/tests.rs @@ -13,13 +13,18 @@ use crate::locale::Locale; use crate::locale::detect_locale_from_sources; use crate::locale::parse_apple_languages; use crate::locale::parse_locale_tag; +use crate::output::dry_run_has_no_changes; use crate::output::install_next_steps; use crate::rollout::RolloutScope; use crate::rollout::reconcile_rollout_metadata_from_sqlite_with_progress; use crate::service::ServiceManager; +use crate::state_db::create_sqlite_backup_file; use crate::state_db::inspect_sqlite_distribution; +use crate::state_db::list_backups; use crate::state_db::reconcile_sqlite_in_place; use crate::state_db::reconcile_sqlite_with_backup; +use crate::state_db::restore_sqlite_from_backup; +use crate::sync::DryRunSummary; use crate::sync::reconcile_once; use crate::watch::WATCH_FULL_ROLLOUT_POLL_INTERVALS; use crate::watch::full_watch_rollout_scope; @@ -32,6 +37,8 @@ use filetime::FileTime; use filetime::set_file_times; use rusqlite::Connection; use std::fs; +#[cfg(unix)] +use std::os::unix::fs::PermissionsExt; use std::path::Path; use std::path::PathBuf; use std::time::Duration; @@ -364,6 +371,181 @@ fn sync_backup_preserves_pre_reconcile_state() -> Result<()> { Ok(()) } +#[test] +fn restore_sqlite_from_backup_ignores_stale_live_wal() -> Result<()> { + let dir = tempfile::tempdir()?; + let sqlite_path = dir.path().join("state_5.sqlite"); + let connection = Connection::open(&sqlite_path)?; + connection.pragma_update(None, "journal_mode", "WAL")?; + connection.execute( + "CREATE TABLE threads(id TEXT PRIMARY KEY, model_provider TEXT)", + [], + )?; + connection.execute("INSERT INTO threads VALUES('old', 'old')", [])?; + connection.query_row("PRAGMA wal_checkpoint(FULL)", [], |_| Ok(()))?; + + let backup_path = create_sqlite_backup_file(&sqlite_path)?; + connection.execute("INSERT INTO threads VALUES('new', 'new')", [])?; + + let wal_path = sqlite_sidecar_path(&sqlite_path, "-wal"); + let shm_path = sqlite_sidecar_path(&sqlite_path, "-shm"); + assert!(wal_path.exists()); + assert!(shm_path.exists()); + + restore_sqlite_from_backup(&sqlite_path, &backup_path)?; + + let restored = Connection::open(&sqlite_path)?; + let restored_ids = restored + .prepare("SELECT id FROM threads ORDER BY id")? + .query_map([], |row| row.get::<_, String>(0))? + .collect::>>()?; + + assert_eq!(restored_ids, vec!["old"]); + Ok(()) +} + +#[test] +fn restore_sqlite_from_backup_keeps_existing_connections_on_live_db() -> Result<()> { + let dir = tempfile::tempdir()?; + let sqlite_path = dir.path().join("state_5.sqlite"); + let connection = Connection::open(&sqlite_path)?; + connection.pragma_update(None, "journal_mode", "WAL")?; + connection.execute( + "CREATE TABLE threads(id TEXT PRIMARY KEY, model_provider TEXT)", + [], + )?; + connection.execute("INSERT INTO threads VALUES('old', 'old')", [])?; + let backup_path = create_sqlite_backup_file(&sqlite_path)?; + connection.execute( + "INSERT INTO threads VALUES('live_before_restore', 'openai')", + [], + )?; + + restore_sqlite_from_backup(&sqlite_path, &backup_path)?; + connection.execute("INSERT INTO threads VALUES('after_restore', 'openai')", [])?; + drop(connection); + + let restored = Connection::open(&sqlite_path)?; + let restored_ids = restored + .prepare("SELECT id FROM threads ORDER BY id")? + .query_map([], |row| row.get::<_, String>(0))? + .collect::>>()?; + + assert_eq!(restored_ids, vec!["after_restore", "old"]); + Ok(()) +} + +#[cfg(unix)] +#[test] +fn restore_sqlite_from_backup_preserves_live_database_permissions() -> Result<()> { + let dir = tempfile::tempdir()?; + let sqlite_path = dir.path().join("state_5.sqlite"); + seed_sqlite(&sqlite_path)?; + fs::set_permissions(&sqlite_path, fs::Permissions::from_mode(0o600))?; + let backup_path = create_sqlite_backup_file(&sqlite_path)?; + + restore_sqlite_from_backup(&sqlite_path, &backup_path)?; + + let mode = fs::metadata(&sqlite_path)?.permissions().mode() & 0o777; + assert_eq!(mode, 0o600); + Ok(()) +} + +#[cfg(unix)] +#[test] +fn restore_sqlite_from_backup_accepts_read_only_backup() -> Result<()> { + let dir = tempfile::tempdir()?; + let sqlite_path = dir.path().join("state_5.sqlite"); + seed_sqlite(&sqlite_path)?; + let backup_path = create_sqlite_backup_file(&sqlite_path)?; + fs::set_permissions(&backup_path, fs::Permissions::from_mode(0o444))?; + + restore_sqlite_from_backup(&sqlite_path, &backup_path)?; + + let connection = Connection::open(&sqlite_path)?; + let rows: u64 = connection.query_row("SELECT COUNT(*) FROM threads", [], |row| row.get(0))?; + assert_eq!(rows, 3); + Ok(()) +} + +#[test] +fn restore_sqlite_from_backup_rejects_live_database_path() -> Result<()> { + let dir = tempfile::tempdir()?; + let sqlite_path = dir.path().join("state_5.sqlite"); + let connection = Connection::open(&sqlite_path)?; + connection.execute( + "CREATE TABLE threads(id TEXT PRIMARY KEY, model_provider TEXT)", + [], + )?; + connection.execute("INSERT INTO threads VALUES('live', 'openai')", [])?; + drop(connection); + + let err = restore_sqlite_from_backup(&sqlite_path, &sqlite_path).unwrap_err(); + assert!(err.to_string().contains("live database")); + let connection = Connection::open(&sqlite_path)?; + let rows: u64 = connection.query_row("SELECT COUNT(*) FROM threads", [], |row| row.get(0))?; + assert_eq!(rows, 1); + Ok(()) +} + +#[test] +fn restore_rejects_latest_with_backup_path() { + let err = localized_command(Locale::En) + .try_get_matches_from([ + "codex-threadripper", + "restore", + "--latest", + "state_5.sqlite.1.bak", + ]) + .unwrap_err(); + + assert_eq!(err.kind(), ErrorKind::ArgumentConflict); +} + +#[test] +fn list_backups_filters_to_selected_database_name() -> Result<()> { + let dir = tempfile::tempdir()?; + let sqlite_path = dir.path().join("state_5.sqlite"); + let backups_dir = dir.path().join("backups"); + fs::create_dir_all(&backups_dir)?; + fs::write(backups_dir.join("state_5.sqlite.300.bak"), b"")?; + fs::write(backups_dir.join("state_5.sqlite.unparseable.bak"), b"")?; + fs::write(backups_dir.join("other.sqlite.400.bak"), b"")?; + + let backups = list_backups(&sqlite_path)?; + let names = backups + .iter() + .map(|entry| { + entry + .path + .file_name() + .unwrap() + .to_string_lossy() + .into_owned() + }) + .collect::>(); + + assert_eq!( + names, + vec!["state_5.sqlite.300.bak", "state_5.sqlite.unparseable.bak"] + ); + Ok(()) +} + +#[test] +fn dry_run_no_changes_excludes_skipped_rollouts() { + let summary = DryRunSummary { + provider: "openai".to_string(), + total_rows: 3, + mismatched_rows: 0, + rollout_would_update: 0, + rollout_would_prepare: 0, + rollout_would_skip: 1, + }; + + assert!(!dry_run_has_no_changes(&summary)); +} + #[test] fn seed_sqlite_fixture_includes_current_codex_thread_columns() -> Result<()> { let dir = tempfile::tempdir()?; @@ -437,6 +619,12 @@ fn seed_sqlite_fixture_includes_current_codex_thread_columns() -> Result<()> { Ok(()) } +fn sqlite_sidecar_path(sqlite_path: &Path, suffix: &str) -> PathBuf { + let mut path = sqlite_path.as_os_str().to_os_string(); + path.push(suffix); + PathBuf::from(path) +} + #[test] fn durable_sync_updates_matching_rollout_session_meta() -> Result<()> { let dir = tempfile::tempdir()?;