diff --git a/crates/clickhousectl/src/local/mod.rs b/crates/clickhousectl/src/local/mod.rs index b8dae566..a944d7e5 100644 --- a/crates/clickhousectl/src/local/mod.rs +++ b/crates/clickhousectl/src/local/mod.rs @@ -210,6 +210,20 @@ fn remove(version: &str, force: bool, json: bool) -> Result<()> { } } + let versions_dir = paths::versions_dir()?; + let staging = version_manager::atomic::InstallStaging::create(&versions_dir)?; + let commit_lock = version_manager::atomic::CommitLock::acquire_blocking(&versions_dir)?; + if !version_dir.exists() { + return Err(Error::VersionNotFound(version.to_string())); + } + + version_manager::master::invalidate_version( + &commit_lock, + &versions_dir, + staging.path(), + version, + )?; + // Check if this is the default version if let Ok(default) = version_manager::get_default_version() && default == version @@ -221,14 +235,7 @@ fn remove(version: &str, force: bool, json: bool) -> Result<()> { } std::fs::remove_dir_all(&version_dir)?; - - // If the removed dir was recorded as the installed master build, clear the - // master sidecar record so a later `latest` resolve doesn't see a stale - // entry pointing at a now-deleted binary. Best-effort, like the install - // overwrite path: a sidecar failure must not fail a successful removal. - if let Ok(platform) = version_manager::platform::Platform::detect() { - let _ = version_manager::master::clear_record_for_version(&platform, version); - } + version_manager::atomic::sync_directory(&versions_dir)?; let out = output::RemoveOutput { version: version.to_string(), diff --git a/crates/clickhousectl/src/version_manager/atomic.rs b/crates/clickhousectl/src/version_manager/atomic.rs new file mode 100644 index 00000000..475d89f0 --- /dev/null +++ b/crates/clickhousectl/src/version_manager/atomic.rs @@ -0,0 +1,172 @@ +//! Cross-process install state. +//! +//! Staging creation briefly takes the cleanup lock and then keeps only its +//! owner lock. Cleanup probes owner locks without blocking. The commit lock is +//! acquired later, after the cleanup lock is gone, so there is no lock cycle. + +use crate::error::{Error, Result}; +use crate::paths; +use std::fs::{File, OpenOptions}; +use std::path::{Path, PathBuf}; +use std::time::{Duration, SystemTime}; +use uuid::Uuid; + +const COMMIT_LOCK: &str = ".install-commit.lock"; +const STAGING_DIR: &str = ".staging"; +const STAGING_CLEANUP_LOCK: &str = ".cleanup.lock"; +const STAGING_OWNER_LOCK: &str = ".owner.lock"; +const STAGING_PREFIX: &str = "install-"; +const STAGING_MAX_AGE: Duration = Duration::from_secs(24 * 60 * 60); + +pub(crate) struct CommitLock { + _file: File, +} + +impl CommitLock { + pub(crate) async fn acquire(versions_dir: &Path) -> Result { + let lock_path = versions_dir.join(COMMIT_LOCK); + let display_path = lock_path.clone(); + tokio::task::spawn_blocking(move || Self::acquire_at(&lock_path)) + .await + .map_err(|error| { + Error::Exec(format!( + "Failed to wait for install commit lock '{}': {error}", + display_path.display() + )) + })? + } + + fn acquire_at(lock_path: &Path) -> Result { + let file = open_lock_file(lock_path)?; + file.lock()?; + Ok(Self { _file: file }) + } + + pub(crate) fn acquire_blocking(versions_dir: &Path) -> Result { + Self::acquire_at(&versions_dir.join(COMMIT_LOCK)) + } +} + +pub(crate) struct InstallStaging { + path: PathBuf, + payload: PathBuf, + _owner: File, +} + +impl InstallStaging { + pub(crate) fn create(versions_dir: &Path) -> Result { + let staging_root = versions_dir.join(STAGING_DIR); + paths::ensure_dir(&staging_root)?; + + let cleanup_lock = open_lock_file(&staging_root.join(STAGING_CLEANUP_LOCK))?; + cleanup_lock.lock()?; + let stale_before = SystemTime::now() + .checked_sub(STAGING_MAX_AGE) + .unwrap_or(SystemTime::UNIX_EPOCH); + cleanup_staging_before_locked(&staging_root, stale_before); + + loop { + let path = staging_root.join(format!("{STAGING_PREFIX}{}", Uuid::new_v4())); + match std::fs::create_dir(&path) { + Ok(()) => { + let staging = (|| { + let owner = open_lock_file(&path.join(STAGING_OWNER_LOCK))?; + owner.lock()?; + let payload = path.join("payload"); + paths::ensure_dir(&payload)?; + Ok(Self { + path: path.clone(), + payload, + _owner: owner, + }) + })(); + if staging.is_err() { + let _ = std::fs::remove_dir_all(&path); + } + return staging; + } + Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => continue, + Err(error) => return Err(error.into()), + } + } + } + + pub(crate) fn path(&self) -> &Path { + &self.path + } + + pub(crate) fn payload(&self) -> &Path { + &self.payload + } + + pub(crate) fn binary_path(&self) -> PathBuf { + self.payload.join("clickhouse") + } +} + +impl Drop for InstallStaging { + fn drop(&mut self) { + let _ = std::fs::remove_dir_all(&self.path); + } +} + +fn open_lock_file(path: &Path) -> std::io::Result { + OpenOptions::new() + .read(true) + .write(true) + .create(true) + .truncate(false) + .open(path) +} + +fn cleanup_staging_before_locked(staging_root: &Path, stale_before: SystemTime) { + let Ok(entries) = std::fs::read_dir(staging_root) else { + return; + }; + + for entry in entries.flatten() { + let path = entry.path(); + let Some(name) = entry.file_name().to_str().map(str::to_owned) else { + continue; + }; + let Some(id) = name.strip_prefix(STAGING_PREFIX) else { + continue; + }; + let Ok(file_type) = entry.file_type() else { + continue; + }; + if Uuid::parse_str(id).is_err() || !file_type.is_dir() { + continue; + } + + let Ok(modified) = std::fs::metadata(&path).and_then(|metadata| metadata.modified()) else { + continue; + }; + if modified >= stale_before { + continue; + } + + let Ok(owner) = open_lock_file(&path.join(STAGING_OWNER_LOCK)) else { + continue; + }; + if owner.try_lock().is_err() { + continue; + } + let _ = std::fs::remove_dir_all(path); + } +} + +#[cfg(test)] +pub(crate) fn cleanup_staging_before(versions_dir: &Path, stale_before: SystemTime) -> Result<()> { + let staging_root = versions_dir.join(STAGING_DIR); + paths::ensure_dir(&staging_root)?; + let cleanup_lock = open_lock_file(&staging_root.join(STAGING_CLEANUP_LOCK))?; + cleanup_lock.lock()?; + cleanup_staging_before_locked(&staging_root, stale_before); + Ok(()) +} + +pub(crate) fn sync_directory(path: &Path) -> Result<()> { + File::open(path)?.sync_all()?; + Ok(()) +} diff --git a/crates/clickhousectl/src/version_manager/install.rs b/crates/clickhousectl/src/version_manager/install.rs index f4e9fadd..d8c89e7a 100644 --- a/crates/clickhousectl/src/version_manager/install.rs +++ b/crates/clickhousectl/src/version_manager/install.rs @@ -1,5 +1,6 @@ use crate::error::{Error, Result}; use crate::paths; +use crate::version_manager::atomic::{CommitLock, InstallStaging, sync_directory}; use crate::version_manager::download::download_from_source; use crate::version_manager::list::list_installed_versions; use crate::version_manager::master; @@ -56,6 +57,7 @@ pub async fn install_resolved( force: bool, ) -> Result { paths::ensure_dirs()?; + let versions_dir = paths::versions_dir()?; // The floating `latest`/master build has no version upfront and a stable // URL whose content moves. Use the HTTP etag to skip the ~153MB download @@ -81,11 +83,11 @@ pub async fn install_resolved( } // If we know the exact version upfront, check if already installed - if let Some(ref version) = resolved.exact_version { - let version_dir = paths::version_dir(version)?; - if version_dir.exists() && !force { - return Err(Error::VersionAlreadyInstalled(version.to_string())); - } + if let Some(ref version) = resolved.exact_version + && is_installed(&paths::binary_path(version)?) + && !force + { + return Err(Error::VersionAlreadyInstalled(version.to_string())); } // For builds source (minor versions like "25.12"), check if we already have @@ -107,22 +109,17 @@ pub async fn install_resolved( } } - // Download to a temp directory first - let temp_dir = paths::versions_dir()?.join(".download-temp"); - if temp_dir.exists() { - std::fs::remove_dir_all(&temp_dir)?; - } - paths::ensure_dir(&temp_dir)?; - - let binary_path = temp_dir.join("clickhouse"); + // Downloads stay outside the commit lock in invocation-owned staging. + let staging = InstallStaging::create(&versions_dir)?; + let binary_path = staging.binary_path(); eprintln!("Downloading ClickHouse {}...", resolved.display_version); if resolved.source.is_tarball(platform) { - let tarball_path = temp_dir.join("clickhouse.tgz"); + let tarball_path = staging.path().join("clickhouse.tgz"); download_from_source(&resolved.source, platform, &tarball_path).await?; eprintln!("Extracting..."); - extract_tarball_auto(&tarball_path, &temp_dir)?; + extract_tarball_auto(&tarball_path, staging.payload())?; } else { download_from_source(&resolved.source, platform, &binary_path).await?; } @@ -140,24 +137,18 @@ pub async fn install_resolved( detect_binary_version(&binary_path)? }; - // Check if this exact version is already installed (post-detection check for builds source). - // Skipped for master: a master build's version string is shared across commits, so an - // existing dir doesn't mean the content matches — we got here because the etag changed - // (or there was no record), so overwrite the existing binary in place and adopt the dir - // as the master install (the record write below re-points the master record at it). - let version_dir = paths::version_dir(&exact_version)?; - if version_dir.exists() && !force && !is_master { - let _ = std::fs::remove_dir_all(&temp_dir); - return Err(Error::VersionAlreadyInstalled(exact_version)); - } - - // Move to final location - let replaced_existing = version_dir.exists(); - if replaced_existing { - std::fs::remove_dir_all(&version_dir)?; - } - paths::ensure_dir(&version_dir)?; - std::fs::rename(&binary_path, version_dir.join("clickhouse"))?; + let commit_lock = CommitLock::acquire(&versions_dir).await?; + let replaced_existing = commit_staged_install_locked( + &commit_lock, + &versions_dir, + &staging, + &exact_version, + force, + is_master, + platform, + master_head.as_ref(), + |_| Ok(()), + )?; // Replacing a build on disk never affects already-running servers (they keep // executing the old binary) — just say so, so the swap isn't silent. @@ -168,24 +159,6 @@ pub async fn install_resolved( ); } - // Clean up temp dir - let _ = std::fs::remove_dir_all(&temp_dir); - - // Record the master etag so a later `latest` resolve can skip the download - // when master is unchanged. Best-effort: a sidecar write failure must not - // fail an otherwise-successful install. - if is_master { - if let Some(head) = &master_head { - let _ = master::record(platform, head, &exact_version); - } - } else { - // A non-master install just wrote versions//. If the - // sidecar recorded that dir as the installed master build, the record - // is now stale and would make a later `latest` resolve reuse this - // binary as if it were master. Best-effort, like `record`. - let _ = master::clear_record_for_version(platform, &exact_version); - } - let channel_suffix = match resolved.channel { Some(ch) => format!(" ({})", ch), None => String::new(), @@ -195,16 +168,85 @@ pub async fn install_resolved( Ok(exact_version) } +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum CommitCheckpoint { + SidecarInvalidated, + BinaryReplaced, +} + +#[allow(clippy::too_many_arguments)] +fn commit_staged_install_locked( + lock: &CommitLock, + versions_dir: &Path, + staging: &InstallStaging, + exact_version: &str, + force: bool, + is_master: bool, + platform: &Platform, + master_head: Option<&master::HeadInfo>, + mut checkpoint: impl FnMut(CommitCheckpoint) -> Result<()>, +) -> Result { + let version_dir = versions_dir.join(exact_version); + let target_binary = version_dir.join("clickhouse"); + let target_metadata = match std::fs::symlink_metadata(&version_dir) { + Ok(metadata) if metadata.file_type().is_dir() => Some(metadata), + Ok(_) => { + return Err(Error::Io(io::Error::new( + io::ErrorKind::AlreadyExists, + format!( + "install target '{}' exists and is not a directory", + version_dir.display() + ), + ))); + } + Err(error) if error.kind() == io::ErrorKind::NotFound => None, + Err(error) => return Err(error.into()), + }; + let replaced_existing = target_binary.exists(); + if replaced_existing && !force && !is_master { + return Err(Error::VersionAlreadyInstalled(exact_version.to_string())); + } + + File::open(staging.binary_path())?.sync_all()?; + sync_directory(staging.payload())?; + + master::invalidate_version(lock, versions_dir, staging.path(), exact_version)?; + checkpoint(CommitCheckpoint::SidecarInvalidated)?; + + if target_metadata.is_some() { + std::fs::rename(staging.binary_path(), &target_binary)?; + sync_directory(&version_dir)?; + } else { + std::fs::rename(staging.payload(), &version_dir)?; + } + sync_directory(versions_dir)?; + checkpoint(CommitCheckpoint::BinaryReplaced)?; + + if is_master && let Some(head) = master_head { + // The binary is already durably committed. A failed freshness record + // must only cause a later re-download, not report the install as failed. + let _ = master::record_install( + lock, + versions_dir, + staging.path(), + platform, + head, + exact_version, + ); + } + + Ok(replaced_existing) +} + /// Like `install_resolved`, but returns the existing version instead of erroring /// when already installed. Intended for cases like `server start --version` where /// the goal is "make sure this version is available" rather than "install this". pub async fn ensure_installed(resolved: &ResolvedVersion, platform: &Platform) -> Result { // If we know the exact version upfront, return it if already installed - if let Some(ref version) = resolved.exact_version { - let version_dir = paths::version_dir(version)?; - if version_dir.exists() { - return Ok(version.clone()); - } + if let Some(ref version) = resolved.exact_version + && is_installed(&paths::binary_path(version)?) + { + return Ok(version.clone()); } // For builds source (minor versions), check if a matching minor is installed @@ -231,6 +273,10 @@ pub async fn ensure_installed(resolved: &ResolvedVersion, platform: &Platform) - } } +fn is_installed(binary_path: &Path) -> bool { + binary_path.exists() +} + /// Whether a running managed server (in the current project) was started from /// this version. Recovers orphans first (like `local remove`) so a server that /// lost its metadata file is still counted. @@ -395,8 +441,14 @@ fn is_clickhouse_binary_path(path: &Path) -> bool { #[cfg(test)] mod tests { use super::*; + use crate::version_manager::atomic::cleanup_staging_before; + use crate::version_manager::platform::{Arch, Os}; use flate2::{Compression, write::GzEncoder}; use std::fs; + use std::path::PathBuf; + use std::process::{Child, Command, Stdio}; + use std::thread; + use std::time::{Duration, Instant, SystemTime}; use tar::{Builder, EntryType, Header}; fn write_archive(archive_path: &Path, entry_path: &str, contents: &[u8]) { @@ -443,6 +495,383 @@ mod tests { assert!(!temp.path().join(".clickhouse.extracting").exists()); } + fn test_platform() -> Platform { + Platform { + os: Os::Linux, + arch: Arch::X86_64, + } + } + + fn signal(path: &Path, value: &str) { + let temporary = path.with_extension(format!("{}.tmp", uuid::Uuid::new_v4())); + let mut file = File::create(&temporary).unwrap(); + file.write_all(value.as_bytes()).unwrap(); + file.sync_all().unwrap(); + drop(file); + fs::rename(temporary, path).unwrap(); + } + + fn wait_for_file(path: &Path) { + let deadline = Instant::now() + Duration::from_secs(10); + while !path.exists() { + assert!( + Instant::now() < deadline, + "timed out waiting for {}", + path.display() + ); + thread::sleep(Duration::from_millis(5)); + } + } + + fn wait_for_env_path(name: &str) { + if let Some(path) = std::env::var_os(name).map(PathBuf::from) { + wait_for_file(&path); + } + } + + fn signal_env_path(name: &str, value: &str) { + if let Some(path) = std::env::var_os(name).map(PathBuf::from) { + signal(&path, value); + } + } + + fn helper_command(versions_dir: &Path, version: &str, contents: &str, etag: &str) -> Command { + let mut command = Command::new(std::env::current_exe().unwrap()); + command + .args([ + "--exact", + "version_manager::install::tests::atomic_install_process_helper", + "--ignored", + "--nocapture", + ]) + .env("CHCTL_ATOMIC_HELPER", "1") + .env("CHCTL_ATOMIC_VERSIONS_DIR", versions_dir) + .env("CHCTL_ATOMIC_VERSION", version) + .env("CHCTL_ATOMIC_CONTENTS", contents) + .env("CHCTL_ATOMIC_ETAG", etag) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()); + command + } + + fn assert_child_success(child: Child) { + let output = child.wait_with_output().unwrap(); + assert!( + output.status.success(), + "helper failed with {:?}\nstdout:\n{}\nstderr:\n{}", + output.status, + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); + } + + fn seed_sidecar(versions_dir: &Path, etag: &str, version: &str) { + let sidecar = serde_json::json!({ + "builds": { + "amd64": { + "etag": etag, + "version": version + } + } + }); + fs::write( + versions_dir.join(".master-builds.json"), + serde_json::to_vec_pretty(&sidecar).unwrap(), + ) + .unwrap(); + } + + fn assert_master_record(versions_dir: &Path, expected: Option<(&str, &str)>) { + let sidecar: serde_json::Value = + serde_json::from_slice(&fs::read(versions_dir.join(".master-builds.json")).unwrap()) + .unwrap(); + let record = sidecar["builds"].get("amd64"); + match expected { + Some((etag, version)) => { + let record = record.expect("amd64 master record"); + assert_eq!(record["etag"], etag); + assert_eq!(record["version"], version); + } + None => assert!(record.is_none(), "unexpected master record: {record:?}"), + } + } + + #[test] + fn empty_version_directory_is_not_installed() { + let temp = tempfile::tempdir().unwrap(); + let version_dir = temp.path().join("26.5.1.1"); + fs::create_dir(&version_dir).unwrap(); + let binary = version_dir.join("clickhouse"); + + assert!(!is_installed(&binary)); + + fs::write(&binary, b"clickhouse").unwrap(); + assert!(is_installed(&binary)); + } + + #[test] + fn post_commit_sidecar_failure_does_not_fail_install() { + let temp = tempfile::tempdir().unwrap(); + let versions_dir = temp.path().join("versions"); + fs::create_dir(&versions_dir).unwrap(); + let staging = InstallStaging::create(&versions_dir).unwrap(); + fs::write(staging.binary_path(), b"complete-master-build").unwrap(); + fs::write(staging.path().join("master-builds.json.tmp"), b"occupied").unwrap(); + let lock = CommitLock::acquire_blocking(&versions_dir).unwrap(); + let head = master::HeadInfo { + etag: "etag-new".to_string(), + last_modified: None, + }; + + let replaced = commit_staged_install_locked( + &lock, + &versions_dir, + &staging, + "26.5.1.1", + true, + true, + &test_platform(), + Some(&head), + |_| Ok(()), + ) + .unwrap(); + + assert!(!replaced); + assert_eq!( + fs::read(versions_dir.join("26.5.1.1/clickhouse")).unwrap(), + b"complete-master-build" + ); + assert!(!versions_dir.join(".master-builds.json").exists()); + } + + #[test] + #[ignore = "subprocess helper for atomic install tests"] + fn atomic_install_process_helper() { + if std::env::var_os("CHCTL_ATOMIC_HELPER").is_none() { + return; + } + + let versions_dir = PathBuf::from(std::env::var_os("CHCTL_ATOMIC_VERSIONS_DIR").unwrap()); + let version = std::env::var("CHCTL_ATOMIC_VERSION").unwrap(); + let contents = std::env::var("CHCTL_ATOMIC_CONTENTS").unwrap(); + let etag = std::env::var("CHCTL_ATOMIC_ETAG").unwrap(); + let staging = InstallStaging::create(&versions_dir).unwrap(); + fs::write(staging.binary_path(), contents).unwrap(); + let mut permissions = fs::metadata(staging.binary_path()).unwrap().permissions(); + permissions.set_mode(0o755); + fs::set_permissions(staging.binary_path(), permissions).unwrap(); + + signal_env_path("CHCTL_ATOMIC_STAGED", &staging.path().to_string_lossy()); + wait_for_env_path("CHCTL_ATOMIC_BEFORE_LOCK_RELEASE"); + + let lock = CommitLock::acquire_blocking(&versions_dir).unwrap(); + signal_env_path("CHCTL_ATOMIC_LOCKED", "locked"); + wait_for_env_path("CHCTL_ATOMIC_LOCK_RELEASE"); + + let pause_at = std::env::var("CHCTL_ATOMIC_PAUSE_AT").ok(); + let head = master::HeadInfo { + etag, + last_modified: None, + }; + commit_staged_install_locked( + &lock, + &versions_dir, + &staging, + &version, + true, + true, + &test_platform(), + Some(&head), + |checkpoint| { + let checkpoint_name = match checkpoint { + CommitCheckpoint::SidecarInvalidated => "invalidated", + CommitCheckpoint::BinaryReplaced => "replaced", + }; + if pause_at.as_deref() == Some(checkpoint_name) { + signal_env_path("CHCTL_ATOMIC_CHECKPOINT", checkpoint_name); + wait_for_env_path("CHCTL_ATOMIC_CHECKPOINT_RELEASE"); + } + Ok(()) + }, + ) + .unwrap(); + } + + #[test] + fn atomic_same_version_race_commits_binary_and_matching_sidecar() { + let temp = tempfile::tempdir().unwrap(); + let versions_dir = temp.path().join("versions"); + fs::create_dir_all(&versions_dir).unwrap(); + let first_staged = temp.path().join("first-staged"); + let first_locked = temp.path().join("first-locked"); + let release_first = temp.path().join("release-first"); + let second_staged = temp.path().join("second-staged"); + let release_second = temp.path().join("release-second"); + + let mut first = helper_command(&versions_dir, "26.5.1.1", "master-a", "etag-a"); + first + .env("CHCTL_ATOMIC_STAGED", &first_staged) + .env("CHCTL_ATOMIC_LOCKED", &first_locked) + .env("CHCTL_ATOMIC_LOCK_RELEASE", &release_first); + let first = first.spawn().unwrap(); + wait_for_file(&first_locked); + + let mut second = helper_command(&versions_dir, "26.5.1.1", "master-b", "etag-b"); + second + .env("CHCTL_ATOMIC_STAGED", &second_staged) + .env("CHCTL_ATOMIC_BEFORE_LOCK_RELEASE", &release_second); + let second = second.spawn().unwrap(); + wait_for_file(&second_staged); + let first_stage = PathBuf::from(fs::read_to_string(&first_staged).unwrap()); + let second_stage = PathBuf::from(fs::read_to_string(&second_staged).unwrap()); + assert_ne!(first_stage, second_stage); + assert!(first_stage.exists()); + assert!(second_stage.exists()); + + signal(&release_second, "release"); + signal(&release_first, "release"); + assert_child_success(first); + assert_child_success(second); + + assert_eq!( + fs::read(versions_dir.join("26.5.1.1/clickhouse")).unwrap(), + b"master-b" + ); + assert_master_record(&versions_dir, Some(("etag-b", "26.5.1.1"))); + } + + #[test] + fn atomic_different_version_race_preserves_binaries_and_latest_sidecar() { + let temp = tempfile::tempdir().unwrap(); + let versions_dir = temp.path().join("versions"); + fs::create_dir_all(&versions_dir).unwrap(); + let first_locked = temp.path().join("first-locked"); + let release_first = temp.path().join("release-first"); + let second_staged = temp.path().join("second-staged"); + + let mut first = helper_command(&versions_dir, "26.5.1.1", "version-a", "etag-a"); + first + .env("CHCTL_ATOMIC_LOCKED", &first_locked) + .env("CHCTL_ATOMIC_LOCK_RELEASE", &release_first); + let first = first.spawn().unwrap(); + wait_for_file(&first_locked); + + let mut second = helper_command(&versions_dir, "26.6.2.2", "version-b", "etag-b"); + second.env("CHCTL_ATOMIC_STAGED", &second_staged); + let second = second.spawn().unwrap(); + wait_for_file(&second_staged); + + signal(&release_first, "release"); + assert_child_success(first); + assert_child_success(second); + + assert_eq!( + fs::read(versions_dir.join("26.5.1.1/clickhouse")).unwrap(), + b"version-a" + ); + assert_eq!( + fs::read(versions_dir.join("26.6.2.2/clickhouse")).unwrap(), + b"version-b" + ); + assert_master_record(&versions_dir, Some(("etag-b", "26.6.2.2"))); + } + + #[test] + fn interrupted_commit_keeps_valid_binary_and_invalidates_sidecar() { + let temp = tempfile::tempdir().unwrap(); + let versions_dir = temp.path().join("versions"); + let version_dir = versions_dir.join("26.5.1.1"); + fs::create_dir_all(&version_dir).unwrap(); + fs::write(version_dir.join("clickhouse"), b"known-good").unwrap(); + seed_sidecar(&versions_dir, "etag-old", "26.5.1.1"); + let checkpoint = temp.path().join("invalidated"); + let never_release = temp.path().join("never-release"); + + let mut command = helper_command(&versions_dir, "26.5.1.1", "partial-new", "etag-new"); + command + .env("CHCTL_ATOMIC_PAUSE_AT", "invalidated") + .env("CHCTL_ATOMIC_CHECKPOINT", &checkpoint) + .env("CHCTL_ATOMIC_CHECKPOINT_RELEASE", &never_release); + let mut child = command.spawn().unwrap(); + wait_for_file(&checkpoint); + child.kill().unwrap(); + let status = child.wait().unwrap(); + assert!(!status.success()); + + assert_eq!( + fs::read(version_dir.join("clickhouse")).unwrap(), + b"known-good" + ); + assert_master_record(&versions_dir, None); + } + + #[test] + fn interrupted_after_replacement_leaves_complete_binary_and_safe_sidecar() { + let temp = tempfile::tempdir().unwrap(); + let versions_dir = temp.path().join("versions"); + let version_dir = versions_dir.join("26.5.1.1"); + fs::create_dir_all(&version_dir).unwrap(); + fs::write(version_dir.join("clickhouse"), b"known-good").unwrap(); + seed_sidecar(&versions_dir, "etag-old", "26.5.1.1"); + let checkpoint = temp.path().join("replaced"); + let never_release = temp.path().join("never-release"); + + let mut command = helper_command(&versions_dir, "26.5.1.1", "complete-new", "etag-new"); + command + .env("CHCTL_ATOMIC_PAUSE_AT", "replaced") + .env("CHCTL_ATOMIC_CHECKPOINT", &checkpoint) + .env("CHCTL_ATOMIC_CHECKPOINT_RELEASE", &never_release); + let mut child = command.spawn().unwrap(); + wait_for_file(&checkpoint); + child.kill().unwrap(); + let status = child.wait().unwrap(); + assert!(!status.success()); + + assert_eq!( + fs::read(version_dir.join("clickhouse")).unwrap(), + b"complete-new" + ); + assert_master_record(&versions_dir, None); + } + + #[test] + fn stale_cleanup_removes_only_unowned_stages_during_live_install() { + let temp = tempfile::tempdir().unwrap(); + let versions_dir = temp.path().join("versions"); + let staging_root = versions_dir.join(".staging"); + fs::create_dir_all(&staging_root).unwrap(); + let stale = staging_root.join(format!("install-{}", uuid::Uuid::new_v4())); + fs::create_dir_all(stale.join("payload")).unwrap(); + fs::write(stale.join(".owner.lock"), b"").unwrap(); + fs::write(stale.join("payload/clickhouse"), b"abandoned").unwrap(); + let unknown = staging_root.join("install-not-owned-by-clickhousectl"); + fs::create_dir(&unknown).unwrap(); + + let live_staged = temp.path().join("live-staged"); + let release_live = temp.path().join("release-live"); + let mut command = helper_command(&versions_dir, "26.7.3.3", "live-build", "etag-live"); + command + .env("CHCTL_ATOMIC_STAGED", &live_staged) + .env("CHCTL_ATOMIC_BEFORE_LOCK_RELEASE", &release_live); + let child = command.spawn().unwrap(); + wait_for_file(&live_staged); + let live_stage = PathBuf::from(fs::read_to_string(&live_staged).unwrap()); + + cleanup_staging_before(&versions_dir, SystemTime::now() + Duration::from_secs(1)).unwrap(); + assert!(!stale.exists()); + assert!(unknown.exists()); + assert!(live_stage.exists()); + + signal(&release_live, "release"); + assert_child_success(child); + assert!(!live_stage.exists()); + assert_eq!( + fs::read(versions_dir.join("26.7.3.3/clickhouse")).unwrap(), + b"live-build" + ); + assert_master_record(&versions_dir, Some(("etag-live", "26.7.3.3"))); + } + #[test] fn test_parse_version_output_client() { let output = "ClickHouse client version 25.12.9.61 (official build)."; diff --git a/crates/clickhousectl/src/version_manager/master.rs b/crates/clickhousectl/src/version_manager/master.rs index 13de002c..085212d9 100644 --- a/crates/clickhousectl/src/version_manager/master.rs +++ b/crates/clickhousectl/src/version_manager/master.rs @@ -14,10 +14,13 @@ use crate::error::{NetworkCategory, NetworkFailure, NetworkStage, Result}; use crate::paths; +use crate::version_manager::atomic::{CommitLock, sync_directory}; use crate::version_manager::network; use crate::version_manager::platform::{DownloadSource, Platform}; use serde::{Deserialize, Serialize}; use std::collections::BTreeMap; +use std::io::Write; +use std::path::{Path, PathBuf}; /// One installed master build's change-detection state, per platform. #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] @@ -49,15 +52,12 @@ struct Sidecar { } /// Path to the sidecar file (`~/.clickhouse/versions/.master-builds.json`). -fn sidecar_path() -> Result { +fn sidecar_path() -> Result { Ok(paths::versions_dir()?.join(".master-builds.json")) } -fn load_sidecar() -> Sidecar { - let Ok(path) = sidecar_path() else { - return Sidecar::default(); - }; - let Ok(bytes) = std::fs::read(&path) else { +fn load_sidecar_at(path: &Path) -> Sidecar { + let Ok(bytes) = std::fs::read(path) else { return Sidecar::default(); }; // A corrupt/old-format sidecar is treated as absent -- worst case is one @@ -67,40 +67,59 @@ fn load_sidecar() -> Sidecar { /// Load the recorded master state for this platform, if any. fn load_record(platform: &Platform) -> Option { - load_sidecar().builds.remove(platform.builds_path()) + load_sidecar_at(&sidecar_path().ok()?) + .builds + .remove(platform.builds_path()) } -/// Pure core of [`clear_record_for_version`]: drop the platform's record iff it -/// records exactly `version`. Returns whether the sidecar changed. -fn clear_version_from(sidecar: &mut Sidecar, platform_key: &str, version: &str) -> bool { - if sidecar - .builds - .get(platform_key) - .is_some_and(|r| r.version == version) - { - sidecar.builds.remove(platform_key); - true - } else { - false - } +fn clear_version_from(sidecar: &mut Sidecar, version: &str) -> bool { + let previous_len = sidecar.builds.len(); + sidecar.builds.retain(|_, record| record.version != version); + sidecar.builds.len() != previous_len +} + +fn write_sidecar_atomic(versions_dir: &Path, scratch_dir: &Path, sidecar: &Sidecar) -> Result<()> { + let temporary_path = scratch_dir.join("master-builds.json.tmp"); + let mut temporary = std::fs::OpenOptions::new() + .write(true) + .create_new(true) + .open(&temporary_path)?; + temporary.write_all(&serde_json::to_vec_pretty(sidecar)?)?; + temporary.sync_all()?; + drop(temporary); + std::fs::rename(&temporary_path, versions_dir.join(".master-builds.json"))?; + sync_directory(versions_dir)?; + Ok(()) } -/// Invalidate the record when a non-master install overwrites `versions//`: -/// the recorded etag no longer describes the binary on disk, and a stale match -/// would make a later `latest` resolve silently reuse the wrong build. -pub fn clear_record_for_version(platform: &Platform, version: &str) -> Result<()> { - let mut sidecar = load_sidecar(); - if clear_version_from(&mut sidecar, platform.builds_path(), version) { - let json = serde_json::to_vec_pretty(&sidecar)?; - std::fs::write(sidecar_path()?, json)?; +/// Remove every record that points at a version about to be replaced. This is +/// committed before the binary swap so an interruption can only cause an extra +/// download, never reuse a binary that no longer matches its recorded etag. +pub(crate) fn invalidate_version( + _lock: &CommitLock, + versions_dir: &Path, + scratch_dir: &Path, + version: &str, +) -> Result<()> { + let path = versions_dir.join(".master-builds.json"); + let mut sidecar = load_sidecar_at(&path); + if clear_version_from(&mut sidecar, version) { + write_sidecar_atomic(versions_dir, scratch_dir, &sidecar)?; } Ok(()) } /// Persist the master state for this platform, merging into any existing /// sidecar so other platforms' records are preserved. -pub fn record(platform: &Platform, head: &HeadInfo, version: &str) -> Result<()> { - let mut sidecar = load_sidecar(); +pub(crate) fn record_install( + _lock: &CommitLock, + versions_dir: &Path, + scratch_dir: &Path, + platform: &Platform, + head: &HeadInfo, + version: &str, +) -> Result<()> { + let mut sidecar = load_sidecar_at(&versions_dir.join(".master-builds.json")); sidecar.builds.insert( platform.builds_path().to_string(), MasterRecord { @@ -109,10 +128,7 @@ pub fn record(platform: &Platform, head: &HeadInfo, version: &str) -> Result<()> version: version.to_string(), }, ); - let path = sidecar_path()?; - let json = serde_json::to_vec_pretty(&sidecar)?; - std::fs::write(&path, json)?; - Ok(()) + write_sidecar_atomic(versions_dir, scratch_dir, &sidecar) } /// HEAD the master URL and pull the change-detection headers. @@ -349,11 +365,7 @@ mod tests { sidecar .builds .insert("macos-aarch64".to_string(), rec("\"x-1\"", "26.5.1.1")); - assert!(clear_version_from( - &mut sidecar, - "macos-aarch64", - "26.5.1.1" - )); + assert!(clear_version_from(&mut sidecar, "26.5.1.1")); assert!(!sidecar.builds.contains_key("macos-aarch64")); } @@ -365,16 +377,12 @@ mod tests { sidecar .builds .insert("macos-aarch64".to_string(), rec("\"x-1\"", "26.5.1.1")); - assert!(!clear_version_from( - &mut sidecar, - "macos-aarch64", - "25.12.9.61" - )); + assert!(!clear_version_from(&mut sidecar, "25.12.9.61")); assert!(sidecar.builds.contains_key("macos-aarch64")); } #[test] - fn clear_version_keeps_other_platforms() { + fn clear_version_removes_every_platform_pointing_at_replaced_directory() { let mut sidecar = Sidecar::default(); sidecar .builds @@ -382,21 +390,13 @@ mod tests { sidecar .builds .insert("macos-aarch64".to_string(), rec("\"y-2\"", "26.5.1.1")); - assert!(clear_version_from( - &mut sidecar, - "macos-aarch64", - "26.5.1.1" - )); - assert!(sidecar.builds.contains_key("amd64")); + assert!(clear_version_from(&mut sidecar, "26.5.1.1")); + assert!(sidecar.builds.is_empty()); } #[test] fn clear_version_no_record_is_noop() { let mut sidecar = Sidecar::default(); - assert!(!clear_version_from( - &mut sidecar, - "macos-aarch64", - "26.5.1.1" - )); + assert!(!clear_version_from(&mut sidecar, "26.5.1.1")); } } diff --git a/crates/clickhousectl/src/version_manager/mod.rs b/crates/clickhousectl/src/version_manager/mod.rs index ed568503..8ccda75b 100644 --- a/crates/clickhousectl/src/version_manager/mod.rs +++ b/crates/clickhousectl/src/version_manager/mod.rs @@ -1,3 +1,4 @@ +pub(crate) mod atomic; pub mod download; pub mod install; pub mod list;