From fe079cd2e8aea42b2898d7f8d49ab944758abbdc Mon Sep 17 00:00:00 2001 From: overtrue Date: Sat, 5 Sep 2026 09:37:50 +0800 Subject: [PATCH] refactor(ecstore): isolate local object rename commit --- crates/ecstore/src/disk/local.rs | 1145 +-------------------- crates/ecstore/src/disk/local/commit.rs | 1205 +++++++++++++++++++++++ 2 files changed, 1212 insertions(+), 1138 deletions(-) create mode 100644 crates/ecstore/src/disk/local/commit.rs diff --git a/crates/ecstore/src/disk/local.rs b/crates/ecstore/src/disk/local.rs index 6baa92a3e..adef17e85 100644 --- a/crates/ecstore/src/disk/local.rs +++ b/crates/ecstore/src/disk/local.rs @@ -12,6 +12,11 @@ // See the License for the specific language governing permissions and // limitations under the License. +#[cfg(test)] +use self::commit::lock_rename_commit_directories; + +mod commit; + use crate::crash_inject::{self, CrashPoint}; use crate::data_usage::local_snapshot::ensure_data_usage_layout; use crate::diagnostics::get::{ @@ -184,65 +189,6 @@ fn restore_part_transaction_file(current: &Path, backup: &Path, absent: &Path, r } } -fn rollback_committed_rename_std( - dst_file_path: &Path, - new_data_path: Option<&Path>, - rollback_data_dir: Option, -) -> std::io::Result<()> { - if let Some(old_data_dir) = rollback_data_dir { - let Some(dst_parent) = dst_file_path.parent() else { - return Err(std::io::Error::new(ErrorKind::InvalidInput, "missing object metadata parent")); - }; - let backup_path = dst_parent.join(old_data_dir.to_string()).join(STORAGE_FORMAT_FILE_BACKUP); - std::fs::rename(backup_path, dst_file_path)?; - } else { - remove_file_if_exists(dst_file_path)?; - } - - if let Some(new_data_path) = new_data_path { - remove_dir_all_if_exists(new_data_path)?; - } - - Ok(()) -} - -fn rollback_inline_metadata_commit_std( - dst_file_path: &Path, - rollback_data_dir: Option, - local_rollback_path: Option<&Path>, -) -> std::io::Result<()> { - if let Some(backup_path) = local_rollback_path { - // The commit immediately before this rollback renamed the staged - // xl.meta from the same directory as `backup_path` onto - // `dst_file_path`, proving both paths are on the same filesystem. - // Unix rename atomically replaces the committed destination; never - // unlink it first or an interrupted rollback could lose xl.meta. - std::fs::rename(backup_path, dst_file_path)?; - } else { - rollback_committed_rename_std(dst_file_path, None, rollback_data_dir)?; - } - Ok(()) -} - -fn create_local_inline_rollback_backup( - dst_file_path: &Path, - staging_file_path: &Path, - old_metadata: &[u8], -) -> std::io::Result { - let Some(staging_parent) = staging_file_path.parent() else { - return Err(std::io::Error::new(ErrorKind::InvalidInput, "missing staging metadata parent")); - }; - let backup_path = staging_parent.join(STORAGE_FORMAT_FILE_BACKUP); - remove_file_if_exists(&backup_path)?; - if (should_fail_local_inline_rollback_hardlink(dst_file_path) || std::fs::hard_link(dst_file_path, &backup_path).is_err()) - && let Err(err) = std::fs::write(&backup_path, old_metadata) - { - let _ = remove_file_if_exists(&backup_path); - return Err(err); - } - Ok(backup_path) -} - async fn write_metadata_rollback_backup(object_dir: &Path, rollback_dir: Uuid, data: &[u8]) -> Result<()> { let backup_dir = object_dir.join(rollback_dir.to_string()); fs::create_dir_all(&backup_dir).await.map_err(to_file_error)?; @@ -269,126 +215,6 @@ async fn restore_metadata_backup( Ok(()) } -async fn lock_rename_commit_directories( - source_parent: &Path, - destination_parent: &Path, - base_dir: &Path, - publication_root: &os::PublicationRoot, - mutation_lease: Arc, -) -> Result { - #[cfg(windows)] - let result = { - let source_parent = source_parent.to_path_buf(); - let destination_parent = destination_parent.to_path_buf(); - let base_dir = base_dir.to_path_buf(); - let publication_root = publication_root.clone(); - os::run_blocking_namespace_operation(mutation_lease, move || { - let result = os::prepare_rename_commit_guard(&source_parent, &destination_parent, &base_dir, &publication_root); - #[cfg(test)] - if result.is_ok() { - run_destination_commit_directory_preparation(&destination_parent); - } - result - }) - .await - }; - #[cfg(not(windows))] - let result = { - let _ = mutation_lease; - os::prepare_rename_commit_guard(source_parent, destination_parent, base_dir, publication_root) - }; - - let result = result.map_err(|err| match std::fs::symlink_metadata(base_dir) { - Err(base_err) if base_err.kind() == ErrorKind::NotFound => base_err, - _ => err, - }); - - result.map_err(to_file_error).map_err(DiskError::from) -} - -async fn read_rename_destination_metadata( - file_path: &Path, - rename_commit_guard: &os::RenameCommitGuard, - mutation_lease: Arc, -) -> Result> { - #[cfg(windows)] - let result = { - let file_path = file_path.to_path_buf(); - let rename_commit_guard = rename_commit_guard.clone(); - os::run_blocking_namespace_operation(mutation_lease, move || { - os::read_destination_file_with_commit_guard(&file_path, &rename_commit_guard) - }) - .await - }; - #[cfg(not(windows))] - let _ = (rename_commit_guard, mutation_lease); - #[cfg(not(windows))] - let result = match super::fs::read_file(file_path).await { - Ok(data) => Ok(Some(data)), - Err(err) if err.kind() == ErrorKind::NotFound => Ok(None), - Err(err) => Err(err), - }; - - result - .map(|data| data.map(Bytes::from)) - .map_err(to_file_error) - .map_err(DiskError::from) -} - -async fn restore_renamed_data_source( - src_volume_dir: &Path, - src_data_path: &Path, - dst_data_path: &Path, - publication_root: &os::PublicationRoot, - mutation_lease: Arc, -) -> Result<()> { - if fs::symlink_metadata(src_data_path).await.is_ok() { - return Ok(()); - } - let result = - match os::rename_all_with_lease(dst_data_path, src_data_path, src_volume_dir, publication_root, mutation_lease).await { - Ok(()) => Ok(()), - Err(DiskError::FileNotFound) => { - let source_exists = fs::symlink_metadata(src_data_path).await.is_ok(); - let destination_missing = matches!( - fs::symlink_metadata(dst_data_path).await, - Err(err) if err.kind() == ErrorKind::NotFound - ); - if source_exists && destination_missing { - Ok(()) - } else { - Err(DiskError::FileNotFound) - } - } - Err(err) => Err(err), - }; - if let Err(err) = &result { - warn!( - event = EVENT_DISK_LOCAL_RENAME_REJECTED, - component = LOG_COMPONENT_ECSTORE, - subsystem = LOG_SUBSYSTEM_DISK_LOCAL, - reason = "restore_staged_data_source_failed", - src_path = ?src_data_path, - dst_path = ?dst_data_path, - error = ?err, - "Failed to restore staged data after a metadata commit was rejected" - ); - } - result -} - -async fn restore_published_data_source( - data_paths: Option<&(PathBuf, PathBuf)>, - src_volume_dir: &Path, - publication_root: &os::PublicationRoot, - mutation_lease: Arc, -) -> Result<()> { - let Some((src_data_path, dst_data_path)) = data_paths else { - return Ok(()); - }; - restore_renamed_data_source(src_volume_dir, src_data_path, dst_data_path, publication_root, mutation_lease).await -} - async fn restore_delete_rollback( object_dir: &Path, xl_path: &Path, @@ -9050,965 +8876,7 @@ impl DiskAPI for LocalDisk { dst_path: &str, ) -> Result { crate::hp_guard!("LocalDisk::rename_data"); - let mut fi = fi; - // A non-force DeleteBucket must not remove a directory while a local - // object commit is publishing into it. The peer's empty scan remains - // optimistic; this lease establishes the local commit/delete order and - // remains owned by any blocking syscall that outlives async cancellation. - let destination_object_path = self.io_get_object_path(dst_volume, dst_path)?; - let quota_fence_token = - match rustfs_utils::http::metadata_compat::get_consistent_str(&fi.metadata, QUOTA_MUTATION_FENCE_METADATA_SUFFIX) { - Some(value) => { - let token = Uuid::parse_str(value).map_err(|_| DiskError::FileCorrupt)?; - Some(SnapshotLeaseToken::from_slice(token.as_bytes())?) - } - None if rustfs_utils::http::metadata_compat::contains_key_str( - &fi.metadata, - QUOTA_MUTATION_FENCE_METADATA_SUFFIX, - ) => - { - return Err(DiskError::FileCorrupt); - } - None => None, - }; - rustfs_utils::http::metadata_compat::remove_str(&mut fi.metadata, QUOTA_MUTATION_FENCE_METADATA_SUFFIX); - let quota_fence_claim = match quota_fence_token { - Some(token) => Some(self.claim_quota_mutation_fence(dst_volume, dst_path, token).await?), - None => None, - }; - let mutation_lease = os::acquire_rename_data_mutation_lease(&self.root, dst_volume, &destination_object_path).await; - if let Some(claim) = quota_fence_claim { - mutation_lease.attach_external_guard(claim); - } - if fi.is_legacy_indexed_delete_marker() { - fi.erasure.index = 0; - } - fi.validate_for_metadata_read()?; - // Snapshot the destination part paths before `fi` is consumed below. These - // are the descriptors a reader may hold for the version this call is about - // to replace (backlog#1145); readers build the identical string in - // `io_primitives`. An inline-data version has no parts and yields none. - let invalidate_part_paths: Vec = { - let data_dir = fi.data_dir.unwrap_or_default(); - fi.parts - .iter() - .map(|part| format!("{dst_path}/{data_dir}/part.{}", part.number)) - .collect() - }; - let src_volume_dir = self.io_get_bucket_path(src_volume)?; - if !skip_access_checks(src_volume) - && let Err(e) = super::fs::access_std(&src_volume_dir) - { - info!( - event = EVENT_DISK_LOCAL_ACCESS_FAILED, - component = LOG_COMPONENT_ECSTORE, - subsystem = LOG_SUBSYSTEM_DISK_LOCAL, - path = ?src_volume_dir, - operation = "rename_data_src_access", - error = %e, - "Disk local access check failed" - ); - return Err(to_access_error(e, DiskError::VolumeAccessDenied).into()); - } - - let dst_volume_dir = self.io_get_bucket_path(dst_volume)?; - if !skip_access_checks(dst_volume) - && let Err(e) = super::fs::access_std(&dst_volume_dir) - { - info!( - event = EVENT_DISK_LOCAL_ACCESS_FAILED, - component = LOG_COMPONENT_ECSTORE, - subsystem = LOG_SUBSYSTEM_DISK_LOCAL, - path = ?dst_volume_dir, - operation = "rename_data_dst_access", - error = %e, - "Disk local access check failed" - ); - return Err(to_access_error(e, DiskError::VolumeAccessDenied).into()); - } - - // xl.meta path - let src_file_path = self.io_get_object_path(src_volume, format!("{}/{}", src_path, STORAGE_FORMAT_FILE).as_str())?; - let dst_file_path = self.io_get_object_path(dst_volume, format!("{}/{}", dst_path, STORAGE_FORMAT_FILE).as_str())?; - - // data_dir path - let has_data_dir_path = { - let has_data_dir = { - if !fi.is_remote() { - fi.data_dir - .map(|dir| rustfs_utils::path::retain_slash(dir.to_string().as_str())) - } else { - None - } - }; - - if let Some(data_dir) = has_data_dir { - let src_data_path = self.io_get_object_path( - src_volume, - rustfs_utils::path::retain_slash(format!("{}/{}", src_path, data_dir).as_str()).as_str(), - )?; - let dst_data_path = self.io_get_object_path( - dst_volume, - rustfs_utils::path::retain_slash(format!("{}/{}", dst_path, data_dir).as_str()).as_str(), - )?; - - Some((src_data_path, dst_data_path)) - } else { - None - } - }; - - check_path_length(src_file_path.to_string_lossy().to_string().as_str())?; - check_path_length(dst_file_path.to_string_lossy().to_string().as_str())?; - - let no_inline = fi.data.is_none() && fi.size > 0; - // Captured before `fi` is consumed by add_version; gates the stale - // destination purge below. - let fi_healing = fi.is_healing(); - - // Resolved once for the whole commit so a concurrent configuration - // change can never leave a single rename_data half-synced. The tier is - // keyed on the destination volume: user data staged in scratch - // namespaces follows the configured tier, while commits into - // system-critical namespaces (IAM, config, bucket metadata) stay - // pinned to strict. - let durability = effective_durability(dst_volume); - - let src_file_parent = src_file_path - .parent() - .ok_or_else(|| DiskError::other("missing staged metadata parent"))?; - let dst_file_parent = dst_file_path - .parent() - .ok_or_else(|| DiskError::other("missing object metadata parent"))?; - if !no_inline { - fs::create_dir_all(src_file_parent).await.map_err(to_file_error)?; - } - // Acquire the common trees before reading destination metadata. On - // Windows this pins the object directory identity across metadata - // preparation, data publication, rollback backup, and final commit. - let rename_commit_guard = lock_rename_commit_directories( - src_file_parent, - dst_file_parent, - &dst_volume_dir, - &self.publication_root, - mutation_lease.clone(), - ) - .await?; - let has_dst_buf = read_rename_destination_metadata(&dst_file_path, &rename_commit_guard, mutation_lease.clone()).await?; - - if no_inline { - // Non-inline: read xl.meta, parse, write, rename data dir, rename xl.meta - let mut xlmeta = FileMeta::new(); - // An existing dst xl.meta that fails to parse leaves `xlmeta` empty - // and gets overwritten by the commit below (pre-existing behavior); - // track that so the old-size observation reports unknown instead of - // a false `Absent` (rustfs/backlog#1009). - let mut dst_meta_unparsable = false; - if let Some(dst_buf) = has_dst_buf.as_ref() { - if FileMeta::is_xl2_v1_format(dst_buf) - && let Ok(nmeta) = FileMeta::load(dst_buf) - { - xlmeta = nmeta - } else { - dst_meta_unparsable = true; - } - } - - let old_current_size = if dst_meta_unparsable { - None - } else { - observe_old_current_size(has_dst_buf.is_some(), &xlmeta) - }; - - let mut skip_parent = dst_volume_dir.clone(); - if has_dst_buf.as_ref().is_some() - && let Some(parent) = dst_file_path.parent() - { - skip_parent = parent.to_path_buf(); - } - - let version_id = fi.version_id.unwrap_or_default(); - let has_old_data_dir = xlmeta.find_unshared_data_dir_for_version(Some(version_id)); - let old_version_exists = xlmeta.find_version(Some(version_id)).is_ok(); - let rollback_data_dir = has_old_data_dir.or_else(|| { - if old_version_exists && has_dst_buf.is_some() { - Some(inline_metadata_rollback_dir(version_id, &xlmeta)) - } else { - None - } - }); - if let Some(old_data_dir) = has_old_data_dir.as_ref() { - let _ = xlmeta.data.remove_two(version_id, *old_data_dir); - } - xlmeta.add_version(fi)?; - let version_signature = rename_data_versions_signature(&xlmeta); - let new_dst_buf = xlmeta.marshal_msg()?; - - // This tmp xl.meta is renamed onto dst_file_path at the commit - // point below, so only its contents must be durable before the - // rename (SyncMode::FileOnly); the dst parent directory is fsynced - // after the commit rename, and a crash before the rename means the - // PUT was never acknowledged. A metadata commit: relaxed tiers - // leave it to the page cache. - let tmp_meta_sync = if durability.syncs_commit_metadata() { - SyncMode::FileOnly - } else { - SyncMode::None - }; - // The tmp xl.meta write and the shard-file fdatasync are independent - // (disjoint paths) and both only need to be durable before the commit - // renames below, so run them concurrently to drop a blocking - // round-trip from the PUT commit critical path (rustfs/backlog#922 - // step 2). The "contents durable -> rename -> dst dir fsync" ordering - // is unchanged — both futures complete before any rename — which the - // rename_data crash-consistency harness (backlog#935) exercises. - // - // Shard durability: once rename_data succeeds the write is - // acknowledged, so data must not live only in the page cache. - // Multipart parts were already synced during rename_part, so their - // fdatasync here is a cheap no-op. A missing source dir is left for the - // rename below to report through the existing rollback path. Payload - // durability is kept by both strict and relaxed. - let tmp_meta_write = { - let src_file_path = src_file_path.clone(); - let dst_file_path = dst_file_path.clone(); - let rename_commit_guard = rename_commit_guard.clone(); - let mutation_lease = mutation_lease.clone(); - async move { - os::run_blocking_namespace_operation(mutation_lease, move || { - #[cfg(test)] - run_owned_file_write_before_open(&src_file_path); - let mut prepared_metadata_source = os::create_prepared_rename_source_with_commit_guard( - &src_file_path, - &dst_file_path, - &rename_commit_guard, - )?; - prepared_metadata_source.write_all(&new_dst_buf, tmp_meta_sync != SyncMode::None)?; - Ok(prepared_metadata_source) - }) - .await - .map_err(to_file_error) - .map_err(DiskError::from) - } - }; - let shard_sync = async { - if durability.syncs_data_shards() - && let Some((src_data_path, _)) = has_data_dir_path.as_ref() - && let Err(err) = os::sync_dir_files_with_limiter(src_data_path, self.file_sync_permits.clone()).await - && err.kind() != ErrorKind::NotFound - { - return Err::<(), DiskError>(to_file_error(err).into()); - } - Ok(()) - }; - let (tmp_meta_res, shard_sync_res) = tokio::join!(tmp_meta_write, shard_sync); - // Surface a tmp-meta failure first (its prior serial position), then a - // shard-sync failure; either aborts before any rename, exactly as the - // sequential version did. - let prepared_metadata_source = tmp_meta_res?; - shard_sync_res?; - let rename_commit_guard = remove_dst_base_before_commit( - dst_path, - rename_commit_guard, - src_file_parent, - dst_file_parent, - &dst_volume_dir, - &self.publication_root, - mutation_lease.clone(), - ) - .await?; - if should_remove_staged_meta_before_commit(dst_path) { - drop(prepared_metadata_source); - std::fs::remove_file(&src_file_path).map_err(to_file_error)?; - return Err(DiskError::FileNotFound); - } - - // Heal reuses the version's data_dir, so for in-place corruption - // the destination dir still exists — and rename(2) cannot replace - // a non-empty directory (EEXIST on XFS, ENOTEMPTY on ext4). Purge - // it first, healing commits only; fresh PUTs mint a new data_dir - // and never collide. Best effort: a real failure surfaces in the - // rename below. - if fi_healing - && let Some((_, dst_data_path)) = has_data_dir_path.as_ref() - && let Err(err) = self.move_to_trash(dst_data_path, true, false).await - { - warn!( - event = EVENT_DISK_LOCAL_HEAL_PURGE_FAILED, - component = LOG_COMPONENT_ECSTORE, - subsystem = LOG_SUBSYSTEM_DISK_LOCAL, - dst_path = ?dst_data_path, - error = ?err, - "Healing commit could not purge the stale destination data dir" - ); - } - if let Some((src_data_path, dst_data_path)) = has_data_dir_path.as_ref() - && let Err(err) = os::rename_all_with_commit_guard( - src_data_path, - dst_data_path, - &skip_parent, - &self.publication_root, - &rename_commit_guard, - mutation_lease.clone(), - ) - .await - { - info!( - event = EVENT_DISK_LOCAL_RENAME_REJECTED, - component = LOG_COMPONENT_ECSTORE, - subsystem = LOG_SUBSYSTEM_DISK_LOCAL, - reason = "rename_all_data_path_failed", - src_path = ?src_data_path, - dst_path = ?dst_data_path, - error = ?err, - "Disk local rename flow failed" - ); - restore_published_data_source( - has_data_dir_path.as_ref(), - &src_volume_dir, - &self.publication_root, - mutation_lease.clone(), - ) - .await?; - return Err(err); - } - #[cfg(test)] - if has_data_dir_path.is_some() { - run_rename_data_after_first_publication(&self.root, dst_volume, dst_path); - } - - // Crash-consistency injection: hard power loss after the data dir - // is in place but before xl.meta commits. No cleanup — the harness - // reopens the disk and asserts the object still reads as the old - // version (the staged data dir is a harmless orphan for GC). - if crash_inject::should_crash_at(CrashPoint::RenameAfterDataRename, dst_path) { - return Err(DiskError::Unexpected); - } - - if should_fail_before_old_metadata_backup(dst_path) { - info!( - event = EVENT_DISK_LOCAL_RENAME_REJECTED, - component = LOG_COMPONENT_ECSTORE, - subsystem = LOG_SUBSYSTEM_DISK_LOCAL, - reason = "test_fail_before_old_metadata_backup", - "Disk local rename flow failed before metadata commit" - ); - restore_published_data_source( - has_data_dir_path.as_ref(), - &src_volume_dir, - &self.publication_root, - mutation_lease.clone(), - ) - .await?; - return Err(DiskError::Unexpected); - } - - // The rollback backup stays where it is written (no rename) and is - // the sole restore source for a later undo_write, so under strict - // it keeps SyncMode::FileAndDir: contents and directory entry both - // durable. It is part of the metadata commit machinery, so relaxed - // tiers leave it to the page cache like the xl.meta it mirrors. - let backup_sync = if durability.syncs_commit_metadata() { - SyncMode::FileAndDir - } else { - SyncMode::None - }; - if let (Some(old_data_dir), Some(dst_buf)) = (rollback_data_dir, has_dst_buf.as_ref()) { - let backup_parent = dst_file_parent.join(old_data_dir.to_string()); - #[cfg(not(windows))] - if let Err(err) = os::make_dir_all(&backup_parent, &skip_parent).await { - restore_published_data_source( - has_data_dir_path.as_ref(), - &src_volume_dir, - &self.publication_root, - mutation_lease.clone(), - ) - .await?; - return Err(err); - } - let backup_path_guard = match rename_commit_guard.create_destination_directory_for_path_access(&backup_parent) { - Ok(guard) => guard, - Err(err) => { - restore_published_data_source( - has_data_dir_path.as_ref(), - &src_volume_dir, - &self.publication_root, - mutation_lease.clone(), - ) - .await?; - return Err(DiskError::from(to_file_error(err))); - } - }; - let backup_path = backup_parent.join(STORAGE_FORMAT_FILE_BACKUP); - if let Err(err) = check_path_length(backup_path.to_string_lossy().as_ref()) { - #[cfg(windows)] - drop(backup_path_guard); - restore_published_data_source( - has_data_dir_path.as_ref(), - &src_volume_dir, - &self.publication_root, - mutation_lease.clone(), - ) - .await?; - return Err(err); - } - let backup_bytes = dst_buf.clone(); - // Keep the volume, commit-tree, and exact destination-path - // guards in this task until the backup write and durability - // sync finish. A detached spawn_blocking writer could survive - // cancellation and later truncate a newer transaction's - // deterministic rollback backup. - let write_result = os::run_blocking_namespace_operation(mutation_lease.clone(), move || { - #[cfg(test)] - run_owned_file_write_before_open(&backup_path); - backup_path_guard.write_file_for_path_access( - &backup_path, - backup_bytes.as_ref(), - backup_sync != SyncMode::None, - backup_sync == SyncMode::FileAndDir, - ) - }) - .await - .map_err(to_file_error) - .map_err(DiskError::from); - if let Err(err) = write_result { - info!( - event = EVENT_DISK_LOCAL_RENAME_REJECTED, - component = LOG_COMPONENT_ECSTORE, - subsystem = LOG_SUBSYSTEM_DISK_LOCAL, - reason = "write_old_metadata_backup_failed", - error = ?err, - "Disk local rename flow failed" - ); - restore_published_data_source( - has_data_dir_path.as_ref(), - &src_volume_dir, - &self.publication_root, - mutation_lease.clone(), - ) - .await?; - return Err(err); - } - } - - // Crash-consistency injection: hard power loss after the rollback - // backup is durable but before the xl.meta commit rename. No - // cleanup — the harness asserts the object still reads as the old - // version, since the destination xl.meta is untouched here. - if crash_inject::should_crash_at(CrashPoint::RenameAfterBackupBeforeMetaCommit, dst_path) { - return Err(DiskError::Unexpected); - } - - if let Err(err) = os::rename_all_with_prepared_source( - prepared_metadata_source, - &src_file_path, - &dst_file_path, - &skip_parent, - &self.publication_root, - &rename_commit_guard, - mutation_lease.clone(), - ) - .await - { - info!( - event = EVENT_DISK_LOCAL_RENAME_REJECTED, - component = LOG_COMPONENT_ECSTORE, - subsystem = LOG_SUBSYSTEM_DISK_LOCAL, - reason = "rename_all_metadata_failed", - src_path = ?src_file_path, - dst_path = ?dst_file_path, - error = ?err, - "Disk local rename flow failed" - ); - restore_published_data_source( - has_data_dir_path.as_ref(), - &src_volume_dir, - &self.publication_root, - mutation_lease.clone(), - ) - .await?; - return Err(err); - } - - let committed_new_data_path = has_data_dir_path.as_ref().map(|(_, dst_data_path)| dst_data_path.as_path()); - if should_fail_after_metadata_commit(dst_path) { - rollback_committed_rename_std(&dst_file_path, committed_new_data_path, rollback_data_dir) - .map_err(to_file_error)?; - return Err(DiskError::Unexpected); - } - - // Crash-consistency injection: hard power loss immediately after the - // xl.meta commit rename but before the durability fsync. Unlike the - // graceful failpoint above, no rollback runs — the commit rename is - // already on disk, so the harness asserts the object reads back as - // the new version. - if crash_inject::should_crash_at(CrashPoint::RenameAfterMetaCommit, dst_path) { - return Err(DiskError::Unexpected); - } - - // Persist the directory entries for both the data dir and xl.meta renames; - // without this the commit itself can vanish on power loss. Relaxed tiers - // accept that window (documented in docs/operations/durability-modes.md). - if durability.syncs_commit_metadata() - && let Some(parent) = dst_file_path.parent() - { - let fsync_started = rustfs_io_metrics::put_stage_timer(); - if let Err(err) = os::fsync_dst_dir_group_commit(parent).await { - rustfs_io_metrics::record_put_object_stage_duration_from( - rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_DST_DIR_FSYNC, - fsync_started, - ); - rollback_committed_rename_std(&dst_file_path, committed_new_data_path, rollback_data_dir) - .map_err(to_file_error)?; - // The commit rename changed the dst part inodes before this fsync - // failed and rolled them back; drop any fd cached during that - // window so readers re-open the restored inode (rustfs/backlog#1177). - for part_path in &invalidate_part_paths { - self.io_backend.invalidate_cached_fd(dst_volume, part_path).await; - } - return Err(to_file_error(err).into()); - } - rustfs_io_metrics::record_put_object_stage_duration_from( - rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_DST_DIR_FSYNC, - fsync_started, - ); - } - - // First PUT of an object creates its directory (and any missing prefix - // dirs) via reliable_mkdir_all, which never fsyncs the parent chain. The - // commit fsync above persists the object dir's *contents*, not its own - // entry in the bucket/prefix dir, so on power loss after ack the whole - // object dir could vanish (rustfs/backlog#922 step 4). For a new object - // (no prior xl.meta) fsync the ancestor chain from the object dir's - // parent up to and including the bucket so those new directory entries - // are durable. Overwrites already have a durable object dir. The - // starts_with guard bounds the walk to the bucket subtree. Relaxed/none - // accept the wider window, like the commit fsync above. - if has_dst_buf.is_none() && durability.syncs_commit_metadata() { - let mut ancestor = dst_file_path.parent().and_then(|object_dir| object_dir.parent()); - while let Some(dir) = ancestor { - if !dir.starts_with(&dst_volume_dir) { - break; - } - let fsync_started = rustfs_io_metrics::put_stage_timer(); - if let Err(err) = os::fsync_dir(dir).await { - rustfs_io_metrics::record_put_object_stage_duration_from( - rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_ANCESTOR_DIR_FSYNC, - fsync_started, - ); - rollback_committed_rename_std(&dst_file_path, committed_new_data_path, rollback_data_dir) - .map_err(to_file_error)?; - // Same post-commit rollback window as above — drop cached - // dst part fds so readers re-open the restored inode - // (rustfs/backlog#1177). - for part_path in &invalidate_part_paths { - self.io_backend.invalidate_cached_fd(dst_volume, part_path).await; - } - return Err(to_file_error(err).into()); - } - rustfs_io_metrics::record_put_object_stage_duration_from( - rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_ANCESTOR_DIR_FSYNC, - fsync_started, - ); - if dir == dst_volume_dir.as_path() { - break; - } - ancestor = dir.parent(); - } - } - - // Publication and every rollback-capable durability step are now - // complete. Do not retain the Windows object identity guard while - // cleaning staging paths or invalidating cached descriptors. - #[cfg(windows)] - drop(rename_commit_guard); - - if let Some(src_file_path_parent) = src_file_path.parent() { - if src_volume != super::RUSTFS_META_MULTIPART_BUCKET { - let _ = std::fs::remove_dir(src_file_path_parent); - } else { - let _ = self - .delete_file(&dst_volume_dir, &src_file_path_parent.to_path_buf(), true, false) - .await; - } - } - - // Heal reuses a version's `data_dir` and lands the rebuilt shard on - // the SAME `//part.N` path. Without this, a cached - // descriptor would keep serving the pre-heal inode, defeating the heal - // and eroding read quorum (backlog#1145). - // - // The exact keys are derivable here, and this runs on every write, so - // use them rather than registering a predicate the read path would then - // have to evaluate. Readers build the same string - // (`{object}/{data_dir}/part.{n}`), and `fi.parts` enumerates every - // part of the version now at `dst_path` — any part path absent from it - // no longer exists for readers to ask for. - for part_path in &invalidate_part_paths { - self.io_backend.invalidate_cached_fd(dst_volume, part_path).await; - } - - Ok(RenameDataResp { - old_data_dir: has_old_data_dir, - rollback_data_dir, - cleanup_data_dir: has_old_data_dir, - sign: version_signature, - old_current_size, - }) - } else { - // Inline metadata preparation is blocking. The transaction lease is - // moved into that work so a timeout can release the async waiter without - // allowing a retry to reuse the deterministic staging path too early. - let src = src_file_path.clone(); - let dst = dst_file_path.clone(); - let cleanup_path = if src_volume == super::RUSTFS_META_MULTIPART_BUCKET { - src_file_path.parent().map(|p| p.to_path_buf()) - } else { - None - }; - let dst_path_for_failpoint = dst_path.to_string(); - #[cfg(windows)] - let source_parent = src_file_parent.to_path_buf(); - let rename_commit_guard_for_preparation = rename_commit_guard.clone(); - let sync = durability.syncs_commit_metadata(); - #[cfg(test)] - run_inline_before_file_sync_admission(dst_path); - let mut file_sync_admission = if sync { - Some( - os::acquire_file_sync_admission(self.file_sync_permits.clone()) - .await - .map_err(to_file_error) - .map_err(DiskError::from)?, - ) - } else { - None - }; - let prepare_inline_metadata = move || { - let mut prepared_metadata_source = - os::create_prepared_rename_source_with_commit_guard(&src, &dst, &rename_commit_guard_for_preparation)?; - #[cfg(windows)] - let source_metadata_guard = - rename_commit_guard_for_preparation.lock_source_directory_for_path_access(&source_parent)?; - let mut xlmeta = FileMeta::new(); - // Same as the non-inline branch: an unparsable existing dst - // xl.meta must surface as unknown, not `Absent` - // (rustfs/backlog#1009). - let mut dst_meta_unparsable = false; - if let Some(ref buf) = has_dst_buf { - if FileMeta::is_xl2_v1_format(buf) - && let Ok(nmeta) = FileMeta::load(buf) - { - xlmeta = nmeta - } else { - dst_meta_unparsable = true; - } - } - - let old_current_size = if dst_meta_unparsable { - None - } else { - observe_old_current_size(has_dst_buf.is_some(), &xlmeta) - }; - - let version_id = fi.version_id.unwrap_or_default(); - let old_data_dir = xlmeta.find_unshared_data_dir_for_version(Some(version_id)); - let old_version_exists = xlmeta.find_version(Some(version_id)).is_ok(); - let rollback_data_dir = old_data_dir.or_else(|| { - if old_version_exists && has_dst_buf.is_some() { - Some(inline_metadata_rollback_dir(version_id, &xlmeta)) - } else { - None - } - }); - let mut staged_rollback_path = None; - if let Some(d) = old_data_dir.as_ref() { - let _ = xlmeta.data.remove_two(version_id, *d); - } - xlmeta.add_version(fi)?; - let version_signature = rename_data_versions_signature(&xlmeta); - let new_buf = xlmeta.marshal_msg()?; - // Write the staged xl.meta. Inline objects carry their data inside - // xl.meta, so this is the durable preparation for the metadata commit: - // relaxed tiers do no per-object fsync here at all (aligned - // with MinIO's default), trading a documented power-loss - // window for latency. - prepared_metadata_source.write_all(&new_buf, sync)?; - run_inline_preparation_before_backup(&dst_path_for_failpoint); - if let Some(ref old_metadata) = has_dst_buf - && (rollback_data_dir.is_some() || sync || cfg!(test)) - { - #[cfg(windows)] - let backup_path = { - let backup_path = src - .parent() - .ok_or_else(|| std::io::Error::new(ErrorKind::InvalidInput, "missing staging metadata parent"))? - .join(STORAGE_FORMAT_FILE_BACKUP); - source_metadata_guard.write_file_for_path_access(&backup_path, old_metadata, sync, false)?; - backup_path - }; - #[cfg(not(windows))] - let backup_path = create_local_inline_rollback_backup(&dst, &src, old_metadata)?; - #[cfg(not(windows))] - if sync { - std::fs::File::open(&backup_path)?.sync_data()?; - } - staged_rollback_path = Some(backup_path); - } - - Ok::<_, std::io::Error>(( - rollback_data_dir, - old_data_dir, - version_signature, - old_current_size, - staged_rollback_path, - has_dst_buf.is_none(), - prepared_metadata_source, - )) - }; - let inline_preparation = if let Some(admission) = file_sync_admission.as_ref() { - os::run_blocking_namespace_file_sync_operation(mutation_lease.clone(), admission, prepare_inline_metadata).await - } else { - os::run_blocking_namespace_operation(mutation_lease.clone(), prepare_inline_metadata).await - } - .map_err(to_file_error) - .map_err(DiskError::from); - - let ( - rollback_data_dir, - cleanup_data_dir, - version_signature, - old_current_size, - mut local_rollback_path, - destination_was_absent, - prepared_metadata_source, - ) = match inline_preparation { - Ok(prepared) => prepared, - Err(err) => { - for part_path in &invalidate_part_paths { - self.io_backend.invalidate_cached_fd(dst_volume, part_path).await; - } - return Err(err); - } - }; - - let rename_commit_guard = remove_dst_base_before_commit( - dst_path, - rename_commit_guard, - src_file_parent, - dst_file_parent, - &dst_volume_dir, - &self.publication_root, - mutation_lease.clone(), - ) - .await?; - - if should_remove_staged_meta_before_commit(dst_path) { - drop(prepared_metadata_source); - let remove_result = std::fs::remove_file(&src_file_path); - if let Some(backup_path) = local_rollback_path.as_deref() { - let _ = remove_file_if_exists(backup_path); - } - remove_result.map_err(to_file_error)?; - return Err(DiskError::FileNotFound); - } - - if let (Some(rollback_data_dir), Some(staged_backup)) = (rollback_data_dir, local_rollback_path.as_deref()) { - let Some(dst_parent) = dst_file_path.parent() else { - return Err(DiskError::other("missing object metadata parent")); - }; - let backup_path = dst_parent - .join(rollback_data_dir.to_string()) - .join(STORAGE_FORMAT_FILE_BACKUP); - // rename_all acquires the backup path's namespace lease. Do not - // hold a disk admission while acquiring another namespace lock. - drop(file_sync_admission.take()); - if let Err(err) = rename_all(staged_backup, &backup_path, &dst_volume_dir, &self.publication_root).await { - let _ = remove_file_if_exists(staged_backup); - return Err(err); - } - #[cfg(test)] - run_rename_data_after_first_publication(&self.root, dst_volume, dst_path); - if sync { - file_sync_admission = Some( - os::acquire_file_sync_admission(self.file_sync_permits.clone()) - .await - .map_err(to_file_error) - .map_err(DiskError::from)?, - ); - } - if let Some(admission) = file_sync_admission.as_ref() - && let Some(backup_parent) = backup_path.parent() - { - let fsync_started = rustfs_io_metrics::put_stage_timer(); - if let Err(err) = - os::fsync_dir_with_namespace_file_sync_limit(backup_parent, mutation_lease.clone(), admission).await - { - rustfs_io_metrics::record_put_object_stage_duration_from( - rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_BACKUP_DIR_FSYNC, - fsync_started, - ); - return Err(DiskError::from(to_file_error(err))); - } - rustfs_io_metrics::record_put_object_stage_duration_from( - rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_BACKUP_DIR_FSYNC, - fsync_started, - ); - } - local_rollback_path = None; - } - - let commit_result = if should_fail_commit_rename(dst_path) { - Err(DiskError::other("test fail during metadata commit rename")) - } else { - os::rename_all_with_prepared_source( - prepared_metadata_source, - &src_file_path, - &dst_file_path, - &dst_volume_dir, - &self.publication_root, - &rename_commit_guard, - mutation_lease.clone(), - ) - .await - }; - if let Err(err) = commit_result { - if let Some(backup_path) = local_rollback_path.as_deref() { - let _ = remove_file_if_exists(backup_path); - } - for part_path in &invalidate_part_paths { - self.io_backend.invalidate_cached_fd(dst_volume, part_path).await; - } - return Err(err); - } - - let post_commit = async { - if should_fail_after_metadata_commit(dst_path) { - rollback_inline_metadata_commit_std(&dst_file_path, rollback_data_dir, local_rollback_path.as_deref())?; - return Err(std::io::Error::other("test fail after metadata commit")); - } - - // Persist the commit rename's directory entry across power loss. - if let Some(admission) = file_sync_admission.as_ref() - && let Some(dst_parent) = dst_file_path.parent() - { - let fsync_started = rustfs_io_metrics::put_stage_timer(); - if let Err(err) = - os::fsync_dst_dir_group_commit_or_namespace_file_sync_limit(dst_parent, mutation_lease.clone(), admission) - .await - { - rustfs_io_metrics::record_put_object_stage_duration_from( - rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_DST_DIR_FSYNC, - fsync_started, - ); - rollback_inline_metadata_commit_std(&dst_file_path, rollback_data_dir, local_rollback_path.as_deref())?; - return Err(err); - } - rustfs_io_metrics::record_put_object_stage_duration_from( - rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_DST_DIR_FSYNC, - fsync_started, - ); - } - - // Same power-loss gap as the non-inline path (rustfs/backlog#922 - // step 4): a first PUT creates the object dir (and any missing - // prefix dirs) whose entry in the bucket/prefix dir reliable_mkdir_all - // never fsynced. The fsync above persists the object dir's contents, - // not its own entry, so for a new inline object fsync the ancestor - // chain up to and including the bucket. Overwrites already have a - // durable object dir; the starts_with guard bounds the walk. - if let Some(admission) = file_sync_admission.as_ref() - && destination_was_absent - { - let mut ancestor = dst_file_path.parent().and_then(|object_dir| object_dir.parent()); - while let Some(ancestor_dir) = ancestor { - if !ancestor_dir.starts_with(&dst_volume_dir) { - break; - } - let fsync_started = rustfs_io_metrics::put_stage_timer(); - if let Err(err) = - os::fsync_dir_with_namespace_file_sync_limit(ancestor_dir, mutation_lease.clone(), admission).await - { - rustfs_io_metrics::record_put_object_stage_duration_from( - rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_ANCESTOR_DIR_FSYNC, - fsync_started, - ); - rollback_inline_metadata_commit_std( - &dst_file_path, - rollback_data_dir, - local_rollback_path.as_deref(), - )?; - return Err(err); - } - rustfs_io_metrics::record_put_object_stage_duration_from( - rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_ANCESTOR_DIR_FSYNC, - fsync_started, - ); - if ancestor_dir == dst_volume_dir.as_path() { - break; - } - ancestor = ancestor_dir.parent(); - } - } - - Ok::<(), std::io::Error>(()) - } - .await; - - // The disk admission protects the durability chain, not staging - // cleanup or cache invalidation after that chain has completed. - drop(file_sync_admission.take()); - - // A post-commit rollback (for example, a commit-metadata fsync - // failure under strict durability) restores the old metadata; drop any - // descriptors cached during the committed window before propagating the - // error (rustfs/backlog#1177). Inline objects carry data in xl.meta, so - // this is mostly defensive and keeps both commit branches consistent. - if let Err(err) = post_commit { - for part_path in &invalidate_part_paths { - self.io_backend.invalidate_cached_fd(dst_volume, part_path).await; - } - return Err(DiskError::from(err)); - } - - // The commit no longer has a rollback path. Release the Windows - // object identity guard before best-effort staging cleanup. - #[cfg(windows)] - drop(rename_commit_guard); - - if let Some(backup_path) = local_rollback_path.as_deref() { - let _ = remove_file_if_exists(backup_path); - } - - // Cleanup - if let Some(ref cleanup) = cleanup_path { - let _ = self.delete_file(&dst_volume_dir, cleanup, true, false).await; - } else if let Some(parent) = src_file_path.parent() { - let _ = std::fs::remove_dir(parent); - } - - // Heal reuses a version's `data_dir` and lands the rebuilt shard on - // the SAME `//part.N` path. Without this, a cached - // descriptor would keep serving the pre-heal inode, defeating the heal - // and eroding read quorum (backlog#1145). - // - // The exact keys are derivable here, and this runs on every write, so - // use them rather than registering a predicate the read path would then - // have to evaluate. Readers build the same string - // (`{object}/{data_dir}/part.{n}`), and `fi.parts` enumerates every - // part of the version now at `dst_path` — any part path absent from it - // no longer exists for readers to ask for. - for part_path in &invalidate_part_paths { - self.io_backend.invalidate_cached_fd(dst_volume, part_path).await; - } - - Ok(RenameDataResp { - old_data_dir: cleanup_data_dir, - rollback_data_dir, - cleanup_data_dir, - sign: version_signature, - old_current_size, - }) - } + self.rename_data_commit(src_volume, src_path, fi, dst_volume, dst_path).await } #[tracing::instrument(level = "trace", skip_all)] @@ -11055,6 +9923,7 @@ async fn get_disk_info(drive_path: PathBuf) -> Result<(rustfs_utils::os::DiskInf #[cfg(test)] mod test { + use super::commit::create_local_inline_rollback_backup; use super::*; use rustfs_filemeta::ErasureInfo; use std::io::{self, Write}; diff --git a/crates/ecstore/src/disk/local/commit.rs b/crates/ecstore/src/disk/local/commit.rs new file mode 100644 index 000000000..12663bc5f --- /dev/null +++ b/crates/ecstore/src/disk/local/commit.rs @@ -0,0 +1,1205 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Single-disk object rename publication and rollback. The caller retains the +//! DiskAPI instrumentation; mutation leases and commit guards follow the syscall. + +#[cfg(all(test, windows))] +use super::run_destination_commit_directory_preparation; +use super::{ + EVENT_DISK_LOCAL_ACCESS_FAILED, EVENT_DISK_LOCAL_HEAL_PURGE_FAILED, EVENT_DISK_LOCAL_RENAME_REJECTED, LOG_COMPONENT_ECSTORE, + LOG_SUBSYSTEM_DISK_LOCAL, LocalDisk, SyncMode, effective_durability, inline_metadata_rollback_dir, observe_old_current_size, + remove_dir_all_if_exists, remove_dst_base_before_commit, remove_file_if_exists, rename_data_versions_signature, + run_inline_preparation_before_backup, should_fail_after_metadata_commit, should_fail_before_old_metadata_backup, + should_fail_commit_rename, should_fail_local_inline_rollback_hardlink, should_remove_staged_meta_before_commit, + skip_access_checks, +}; +#[cfg(test)] +use super::{run_inline_before_file_sync_admission, run_owned_file_write_before_open, run_rename_data_after_first_publication}; +use crate::crash_inject::{self, CrashPoint}; +use crate::disk::{ + QUOTA_MUTATION_FENCE_METADATA_SUFFIX, RenameDataResp, STORAGE_FORMAT_FILE, STORAGE_FORMAT_FILE_BACKUP, SnapshotLeaseToken, + error::{DiskError, Result}, + error_conv::{to_access_error, to_file_error}, + os, + os::{check_path_length, rename_all}, +}; +use bytes::Bytes; +use rustfs_filemeta::{FileInfo, FileMeta}; +use std::{ + io::ErrorKind, + path::{Path, PathBuf}, + sync::Arc, +}; +use tokio::fs; +use tracing::{info, warn}; +use uuid::Uuid; + +fn rollback_committed_rename_std( + dst_file_path: &Path, + new_data_path: Option<&Path>, + rollback_data_dir: Option, +) -> std::io::Result<()> { + if let Some(old_data_dir) = rollback_data_dir { + let Some(dst_parent) = dst_file_path.parent() else { + return Err(std::io::Error::new(ErrorKind::InvalidInput, "missing object metadata parent")); + }; + let backup_path = dst_parent.join(old_data_dir.to_string()).join(STORAGE_FORMAT_FILE_BACKUP); + std::fs::rename(backup_path, dst_file_path)?; + } else { + remove_file_if_exists(dst_file_path)?; + } + + if let Some(new_data_path) = new_data_path { + remove_dir_all_if_exists(new_data_path)?; + } + + Ok(()) +} + +fn rollback_inline_metadata_commit_std( + dst_file_path: &Path, + rollback_data_dir: Option, + local_rollback_path: Option<&Path>, +) -> std::io::Result<()> { + if let Some(backup_path) = local_rollback_path { + // The commit immediately before this rollback renamed the staged + // xl.meta from the same directory as `backup_path` onto + // `dst_file_path`, proving both paths are on the same filesystem. + // Unix rename atomically replaces the committed destination; never + // unlink it first or an interrupted rollback could lose xl.meta. + std::fs::rename(backup_path, dst_file_path)?; + } else { + rollback_committed_rename_std(dst_file_path, None, rollback_data_dir)?; + } + Ok(()) +} + +pub(super) fn create_local_inline_rollback_backup( + dst_file_path: &Path, + staging_file_path: &Path, + old_metadata: &[u8], +) -> std::io::Result { + let Some(staging_parent) = staging_file_path.parent() else { + return Err(std::io::Error::new(ErrorKind::InvalidInput, "missing staging metadata parent")); + }; + let backup_path = staging_parent.join(STORAGE_FORMAT_FILE_BACKUP); + remove_file_if_exists(&backup_path)?; + if (should_fail_local_inline_rollback_hardlink(dst_file_path) || std::fs::hard_link(dst_file_path, &backup_path).is_err()) + && let Err(err) = std::fs::write(&backup_path, old_metadata) + { + let _ = remove_file_if_exists(&backup_path); + return Err(err); + } + Ok(backup_path) +} + +pub(super) async fn lock_rename_commit_directories( + source_parent: &Path, + destination_parent: &Path, + base_dir: &Path, + publication_root: &os::PublicationRoot, + mutation_lease: Arc, +) -> Result { + #[cfg(windows)] + let result = { + let source_parent = source_parent.to_path_buf(); + let destination_parent = destination_parent.to_path_buf(); + let base_dir = base_dir.to_path_buf(); + let publication_root = publication_root.clone(); + os::run_blocking_namespace_operation(mutation_lease, move || { + let result = os::prepare_rename_commit_guard(&source_parent, &destination_parent, &base_dir, &publication_root); + #[cfg(test)] + if result.is_ok() { + run_destination_commit_directory_preparation(&destination_parent); + } + result + }) + .await + }; + #[cfg(not(windows))] + let result = { + let _ = mutation_lease; + os::prepare_rename_commit_guard(source_parent, destination_parent, base_dir, publication_root) + }; + + let result = result.map_err(|err| match std::fs::symlink_metadata(base_dir) { + Err(base_err) if base_err.kind() == ErrorKind::NotFound => base_err, + _ => err, + }); + + result.map_err(to_file_error).map_err(DiskError::from) +} + +async fn read_rename_destination_metadata( + file_path: &Path, + rename_commit_guard: &os::RenameCommitGuard, + mutation_lease: Arc, +) -> Result> { + #[cfg(windows)] + let result = { + let file_path = file_path.to_path_buf(); + let rename_commit_guard = rename_commit_guard.clone(); + os::run_blocking_namespace_operation(mutation_lease, move || { + os::read_destination_file_with_commit_guard(&file_path, &rename_commit_guard) + }) + .await + }; + #[cfg(not(windows))] + let _ = (rename_commit_guard, mutation_lease); + #[cfg(not(windows))] + let result = match super::super::fs::read_file(file_path).await { + Ok(data) => Ok(Some(data)), + Err(err) if err.kind() == ErrorKind::NotFound => Ok(None), + Err(err) => Err(err), + }; + + result + .map(|data| data.map(Bytes::from)) + .map_err(to_file_error) + .map_err(DiskError::from) +} + +async fn restore_renamed_data_source( + src_volume_dir: &Path, + src_data_path: &Path, + dst_data_path: &Path, + publication_root: &os::PublicationRoot, + mutation_lease: Arc, +) -> Result<()> { + if fs::symlink_metadata(src_data_path).await.is_ok() { + return Ok(()); + } + let result = + match os::rename_all_with_lease(dst_data_path, src_data_path, src_volume_dir, publication_root, mutation_lease).await { + Ok(()) => Ok(()), + Err(DiskError::FileNotFound) => { + let source_exists = fs::symlink_metadata(src_data_path).await.is_ok(); + let destination_missing = matches!( + fs::symlink_metadata(dst_data_path).await, + Err(err) if err.kind() == ErrorKind::NotFound + ); + if source_exists && destination_missing { + Ok(()) + } else { + Err(DiskError::FileNotFound) + } + } + Err(err) => Err(err), + }; + if let Err(err) = &result { + warn!( + target: "rustfs_ecstore::disk::local", + event = EVENT_DISK_LOCAL_RENAME_REJECTED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_DISK_LOCAL, + reason = "restore_staged_data_source_failed", + src_path = ?src_data_path, + dst_path = ?dst_data_path, + error = ?err, + "Failed to restore staged data after a metadata commit was rejected" + ); + } + result +} + +async fn restore_published_data_source( + data_paths: Option<&(PathBuf, PathBuf)>, + src_volume_dir: &Path, + publication_root: &os::PublicationRoot, + mutation_lease: Arc, +) -> Result<()> { + let Some((src_data_path, dst_data_path)) = data_paths else { + return Ok(()); + }; + restore_renamed_data_source(src_volume_dir, src_data_path, dst_data_path, publication_root, mutation_lease).await +} + +impl LocalDisk { + pub(super) async fn rename_data_commit( + &self, + src_volume: &str, + src_path: &str, + fi: FileInfo, + dst_volume: &str, + dst_path: &str, + ) -> Result { + let mut fi = fi; + // A non-force DeleteBucket must not remove a directory while a local + // object commit is publishing into it. The peer's empty scan remains + // optimistic; this lease establishes the local commit/delete order and + // remains owned by any blocking syscall that outlives async cancellation. + let destination_object_path = self.io_get_object_path(dst_volume, dst_path)?; + let quota_fence_token = + match rustfs_utils::http::metadata_compat::get_consistent_str(&fi.metadata, QUOTA_MUTATION_FENCE_METADATA_SUFFIX) { + Some(value) => { + let token = Uuid::parse_str(value).map_err(|_| DiskError::FileCorrupt)?; + Some(SnapshotLeaseToken::from_slice(token.as_bytes())?) + } + None if rustfs_utils::http::metadata_compat::contains_key_str( + &fi.metadata, + QUOTA_MUTATION_FENCE_METADATA_SUFFIX, + ) => + { + return Err(DiskError::FileCorrupt); + } + None => None, + }; + rustfs_utils::http::metadata_compat::remove_str(&mut fi.metadata, QUOTA_MUTATION_FENCE_METADATA_SUFFIX); + let quota_fence_claim = match quota_fence_token { + Some(token) => Some(self.claim_quota_mutation_fence(dst_volume, dst_path, token).await?), + None => None, + }; + let mutation_lease = os::acquire_rename_data_mutation_lease(&self.root, dst_volume, &destination_object_path).await; + if let Some(claim) = quota_fence_claim { + mutation_lease.attach_external_guard(claim); + } + if fi.is_legacy_indexed_delete_marker() { + fi.erasure.index = 0; + } + fi.validate_for_metadata_read()?; + // Snapshot the destination part paths before `fi` is consumed below. These + // are the descriptors a reader may hold for the version this call is about + // to replace (backlog#1145); readers build the identical string in + // `io_primitives`. An inline-data version has no parts and yields none. + let invalidate_part_paths: Vec = { + let data_dir = fi.data_dir.unwrap_or_default(); + fi.parts + .iter() + .map(|part| format!("{dst_path}/{data_dir}/part.{}", part.number)) + .collect() + }; + let src_volume_dir = self.io_get_bucket_path(src_volume)?; + if !skip_access_checks(src_volume) + && let Err(e) = super::super::fs::access_std(&src_volume_dir) + { + info!( + target: "rustfs_ecstore::disk::local", + event = EVENT_DISK_LOCAL_ACCESS_FAILED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_DISK_LOCAL, + path = ?src_volume_dir, + operation = "rename_data_src_access", + error = %e, + "Disk local access check failed" + ); + return Err(to_access_error(e, DiskError::VolumeAccessDenied).into()); + } + + let dst_volume_dir = self.io_get_bucket_path(dst_volume)?; + if !skip_access_checks(dst_volume) + && let Err(e) = super::super::fs::access_std(&dst_volume_dir) + { + info!( + target: "rustfs_ecstore::disk::local", + event = EVENT_DISK_LOCAL_ACCESS_FAILED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_DISK_LOCAL, + path = ?dst_volume_dir, + operation = "rename_data_dst_access", + error = %e, + "Disk local access check failed" + ); + return Err(to_access_error(e, DiskError::VolumeAccessDenied).into()); + } + + // xl.meta path + let src_file_path = self.io_get_object_path(src_volume, format!("{}/{}", src_path, STORAGE_FORMAT_FILE).as_str())?; + let dst_file_path = self.io_get_object_path(dst_volume, format!("{}/{}", dst_path, STORAGE_FORMAT_FILE).as_str())?; + + // data_dir path + let has_data_dir_path = { + let has_data_dir = { + if !fi.is_remote() { + fi.data_dir + .map(|dir| rustfs_utils::path::retain_slash(dir.to_string().as_str())) + } else { + None + } + }; + + if let Some(data_dir) = has_data_dir { + let src_data_path = self.io_get_object_path( + src_volume, + rustfs_utils::path::retain_slash(format!("{}/{}", src_path, data_dir).as_str()).as_str(), + )?; + let dst_data_path = self.io_get_object_path( + dst_volume, + rustfs_utils::path::retain_slash(format!("{}/{}", dst_path, data_dir).as_str()).as_str(), + )?; + + Some((src_data_path, dst_data_path)) + } else { + None + } + }; + + check_path_length(src_file_path.to_string_lossy().to_string().as_str())?; + check_path_length(dst_file_path.to_string_lossy().to_string().as_str())?; + + let no_inline = fi.data.is_none() && fi.size > 0; + // Captured before `fi` is consumed by add_version; gates the stale + // destination purge below. + let fi_healing = fi.is_healing(); + + // Resolved once for the whole commit so a concurrent configuration + // change can never leave a single rename_data half-synced. The tier is + // keyed on the destination volume: user data staged in scratch + // namespaces follows the configured tier, while commits into + // system-critical namespaces (IAM, config, bucket metadata) stay + // pinned to strict. + let durability = effective_durability(dst_volume); + + let src_file_parent = src_file_path + .parent() + .ok_or_else(|| DiskError::other("missing staged metadata parent"))?; + let dst_file_parent = dst_file_path + .parent() + .ok_or_else(|| DiskError::other("missing object metadata parent"))?; + if !no_inline { + fs::create_dir_all(src_file_parent).await.map_err(to_file_error)?; + } + // Acquire the common trees before reading destination metadata. On + // Windows this pins the object directory identity across metadata + // preparation, data publication, rollback backup, and final commit. + let rename_commit_guard = lock_rename_commit_directories( + src_file_parent, + dst_file_parent, + &dst_volume_dir, + &self.publication_root, + mutation_lease.clone(), + ) + .await?; + let has_dst_buf = read_rename_destination_metadata(&dst_file_path, &rename_commit_guard, mutation_lease.clone()).await?; + + if no_inline { + // Non-inline: read xl.meta, parse, write, rename data dir, rename xl.meta + let mut xlmeta = FileMeta::new(); + // An existing dst xl.meta that fails to parse leaves `xlmeta` empty + // and gets overwritten by the commit below (pre-existing behavior); + // track that so the old-size observation reports unknown instead of + // a false `Absent` (rustfs/backlog#1009). + let mut dst_meta_unparsable = false; + if let Some(dst_buf) = has_dst_buf.as_ref() { + if FileMeta::is_xl2_v1_format(dst_buf) + && let Ok(nmeta) = FileMeta::load(dst_buf) + { + xlmeta = nmeta + } else { + dst_meta_unparsable = true; + } + } + + let old_current_size = if dst_meta_unparsable { + None + } else { + observe_old_current_size(has_dst_buf.is_some(), &xlmeta) + }; + + let mut skip_parent = dst_volume_dir.clone(); + if has_dst_buf.as_ref().is_some() + && let Some(parent) = dst_file_path.parent() + { + skip_parent = parent.to_path_buf(); + } + + let version_id = fi.version_id.unwrap_or_default(); + let has_old_data_dir = xlmeta.find_unshared_data_dir_for_version(Some(version_id)); + let old_version_exists = xlmeta.find_version(Some(version_id)).is_ok(); + let rollback_data_dir = has_old_data_dir.or_else(|| { + if old_version_exists && has_dst_buf.is_some() { + Some(inline_metadata_rollback_dir(version_id, &xlmeta)) + } else { + None + } + }); + if let Some(old_data_dir) = has_old_data_dir.as_ref() { + let _ = xlmeta.data.remove_two(version_id, *old_data_dir); + } + xlmeta.add_version(fi)?; + let version_signature = rename_data_versions_signature(&xlmeta); + let new_dst_buf = xlmeta.marshal_msg()?; + + // This tmp xl.meta is renamed onto dst_file_path at the commit + // point below, so only its contents must be durable before the + // rename (SyncMode::FileOnly); the dst parent directory is fsynced + // after the commit rename, and a crash before the rename means the + // PUT was never acknowledged. A metadata commit: relaxed tiers + // leave it to the page cache. + let tmp_meta_sync = if durability.syncs_commit_metadata() { + SyncMode::FileOnly + } else { + SyncMode::None + }; + // The tmp xl.meta write and the shard-file fdatasync are independent + // (disjoint paths) and both only need to be durable before the commit + // renames below, so run them concurrently to drop a blocking + // round-trip from the PUT commit critical path (rustfs/backlog#922 + // step 2). The "contents durable -> rename -> dst dir fsync" ordering + // is unchanged — both futures complete before any rename — which the + // rename_data crash-consistency harness (backlog#935) exercises. + // + // Shard durability: once rename_data succeeds the write is + // acknowledged, so data must not live only in the page cache. + // Multipart parts were already synced during rename_part, so their + // fdatasync here is a cheap no-op. A missing source dir is left for the + // rename below to report through the existing rollback path. Payload + // durability is kept by both strict and relaxed. + let tmp_meta_write = { + let src_file_path = src_file_path.clone(); + let dst_file_path = dst_file_path.clone(); + let rename_commit_guard = rename_commit_guard.clone(); + let mutation_lease = mutation_lease.clone(); + async move { + os::run_blocking_namespace_operation(mutation_lease, move || { + #[cfg(test)] + run_owned_file_write_before_open(&src_file_path); + let mut prepared_metadata_source = os::create_prepared_rename_source_with_commit_guard( + &src_file_path, + &dst_file_path, + &rename_commit_guard, + )?; + prepared_metadata_source.write_all(&new_dst_buf, tmp_meta_sync != SyncMode::None)?; + Ok(prepared_metadata_source) + }) + .await + .map_err(to_file_error) + .map_err(DiskError::from) + } + }; + let shard_sync = async { + if durability.syncs_data_shards() + && let Some((src_data_path, _)) = has_data_dir_path.as_ref() + && let Err(err) = os::sync_dir_files_with_limiter(src_data_path, self.file_sync_permits.clone()).await + && err.kind() != ErrorKind::NotFound + { + return Err::<(), DiskError>(to_file_error(err).into()); + } + Ok(()) + }; + let (tmp_meta_res, shard_sync_res) = tokio::join!(tmp_meta_write, shard_sync); + // Surface a tmp-meta failure first (its prior serial position), then a + // shard-sync failure; either aborts before any rename, exactly as the + // sequential version did. + let prepared_metadata_source = tmp_meta_res?; + shard_sync_res?; + let rename_commit_guard = remove_dst_base_before_commit( + dst_path, + rename_commit_guard, + src_file_parent, + dst_file_parent, + &dst_volume_dir, + &self.publication_root, + mutation_lease.clone(), + ) + .await?; + if should_remove_staged_meta_before_commit(dst_path) { + drop(prepared_metadata_source); + std::fs::remove_file(&src_file_path).map_err(to_file_error)?; + return Err(DiskError::FileNotFound); + } + + // Heal reuses the version's data_dir, so for in-place corruption + // the destination dir still exists — and rename(2) cannot replace + // a non-empty directory (EEXIST on XFS, ENOTEMPTY on ext4). Purge + // it first, healing commits only; fresh PUTs mint a new data_dir + // and never collide. Best effort: a real failure surfaces in the + // rename below. + if fi_healing + && let Some((_, dst_data_path)) = has_data_dir_path.as_ref() + && let Err(err) = self.move_to_trash(dst_data_path, true, false).await + { + warn!( + target: "rustfs_ecstore::disk::local", + event = EVENT_DISK_LOCAL_HEAL_PURGE_FAILED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_DISK_LOCAL, + dst_path = ?dst_data_path, + error = ?err, + "Healing commit could not purge the stale destination data dir" + ); + } + if let Some((src_data_path, dst_data_path)) = has_data_dir_path.as_ref() + && let Err(err) = os::rename_all_with_commit_guard( + src_data_path, + dst_data_path, + &skip_parent, + &self.publication_root, + &rename_commit_guard, + mutation_lease.clone(), + ) + .await + { + info!( + target: "rustfs_ecstore::disk::local", + event = EVENT_DISK_LOCAL_RENAME_REJECTED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_DISK_LOCAL, + reason = "rename_all_data_path_failed", + src_path = ?src_data_path, + dst_path = ?dst_data_path, + error = ?err, + "Disk local rename flow failed" + ); + restore_published_data_source( + has_data_dir_path.as_ref(), + &src_volume_dir, + &self.publication_root, + mutation_lease.clone(), + ) + .await?; + return Err(err); + } + #[cfg(test)] + if has_data_dir_path.is_some() { + run_rename_data_after_first_publication(&self.root, dst_volume, dst_path); + } + + // Crash-consistency injection: hard power loss after the data dir + // is in place but before xl.meta commits. No cleanup — the harness + // reopens the disk and asserts the object still reads as the old + // version (the staged data dir is a harmless orphan for GC). + if crash_inject::should_crash_at(CrashPoint::RenameAfterDataRename, dst_path) { + return Err(DiskError::Unexpected); + } + + if should_fail_before_old_metadata_backup(dst_path) { + info!( + target: "rustfs_ecstore::disk::local", + event = EVENT_DISK_LOCAL_RENAME_REJECTED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_DISK_LOCAL, + reason = "test_fail_before_old_metadata_backup", + "Disk local rename flow failed before metadata commit" + ); + restore_published_data_source( + has_data_dir_path.as_ref(), + &src_volume_dir, + &self.publication_root, + mutation_lease.clone(), + ) + .await?; + return Err(DiskError::Unexpected); + } + + // The rollback backup stays where it is written (no rename) and is + // the sole restore source for a later undo_write, so under strict + // it keeps SyncMode::FileAndDir: contents and directory entry both + // durable. It is part of the metadata commit machinery, so relaxed + // tiers leave it to the page cache like the xl.meta it mirrors. + let backup_sync = if durability.syncs_commit_metadata() { + SyncMode::FileAndDir + } else { + SyncMode::None + }; + if let (Some(old_data_dir), Some(dst_buf)) = (rollback_data_dir, has_dst_buf.as_ref()) { + let backup_parent = dst_file_parent.join(old_data_dir.to_string()); + #[cfg(not(windows))] + if let Err(err) = os::make_dir_all(&backup_parent, &skip_parent).await { + restore_published_data_source( + has_data_dir_path.as_ref(), + &src_volume_dir, + &self.publication_root, + mutation_lease.clone(), + ) + .await?; + return Err(err); + } + let backup_path_guard = match rename_commit_guard.create_destination_directory_for_path_access(&backup_parent) { + Ok(guard) => guard, + Err(err) => { + restore_published_data_source( + has_data_dir_path.as_ref(), + &src_volume_dir, + &self.publication_root, + mutation_lease.clone(), + ) + .await?; + return Err(DiskError::from(to_file_error(err))); + } + }; + let backup_path = backup_parent.join(STORAGE_FORMAT_FILE_BACKUP); + if let Err(err) = check_path_length(backup_path.to_string_lossy().as_ref()) { + #[cfg(windows)] + drop(backup_path_guard); + restore_published_data_source( + has_data_dir_path.as_ref(), + &src_volume_dir, + &self.publication_root, + mutation_lease.clone(), + ) + .await?; + return Err(err); + } + let backup_bytes = dst_buf.clone(); + // Keep the volume, commit-tree, and exact destination-path + // guards in this task until the backup write and durability + // sync finish. A detached spawn_blocking writer could survive + // cancellation and later truncate a newer transaction's + // deterministic rollback backup. + let write_result = os::run_blocking_namespace_operation(mutation_lease.clone(), move || { + #[cfg(test)] + run_owned_file_write_before_open(&backup_path); + backup_path_guard.write_file_for_path_access( + &backup_path, + backup_bytes.as_ref(), + backup_sync != SyncMode::None, + backup_sync == SyncMode::FileAndDir, + ) + }) + .await + .map_err(to_file_error) + .map_err(DiskError::from); + if let Err(err) = write_result { + info!( + target: "rustfs_ecstore::disk::local", + event = EVENT_DISK_LOCAL_RENAME_REJECTED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_DISK_LOCAL, + reason = "write_old_metadata_backup_failed", + error = ?err, + "Disk local rename flow failed" + ); + restore_published_data_source( + has_data_dir_path.as_ref(), + &src_volume_dir, + &self.publication_root, + mutation_lease.clone(), + ) + .await?; + return Err(err); + } + } + + // Crash-consistency injection: hard power loss after the rollback + // backup is durable but before the xl.meta commit rename. No + // cleanup — the harness asserts the object still reads as the old + // version, since the destination xl.meta is untouched here. + if crash_inject::should_crash_at(CrashPoint::RenameAfterBackupBeforeMetaCommit, dst_path) { + return Err(DiskError::Unexpected); + } + + if let Err(err) = os::rename_all_with_prepared_source( + prepared_metadata_source, + &src_file_path, + &dst_file_path, + &skip_parent, + &self.publication_root, + &rename_commit_guard, + mutation_lease.clone(), + ) + .await + { + info!( + target: "rustfs_ecstore::disk::local", + event = EVENT_DISK_LOCAL_RENAME_REJECTED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_DISK_LOCAL, + reason = "rename_all_metadata_failed", + src_path = ?src_file_path, + dst_path = ?dst_file_path, + error = ?err, + "Disk local rename flow failed" + ); + restore_published_data_source( + has_data_dir_path.as_ref(), + &src_volume_dir, + &self.publication_root, + mutation_lease.clone(), + ) + .await?; + return Err(err); + } + + let committed_new_data_path = has_data_dir_path.as_ref().map(|(_, dst_data_path)| dst_data_path.as_path()); + if should_fail_after_metadata_commit(dst_path) { + rollback_committed_rename_std(&dst_file_path, committed_new_data_path, rollback_data_dir) + .map_err(to_file_error)?; + return Err(DiskError::Unexpected); + } + + // Crash-consistency injection: hard power loss immediately after the + // xl.meta commit rename but before the durability fsync. Unlike the + // graceful failpoint above, no rollback runs — the commit rename is + // already on disk, so the harness asserts the object reads back as + // the new version. + if crash_inject::should_crash_at(CrashPoint::RenameAfterMetaCommit, dst_path) { + return Err(DiskError::Unexpected); + } + + // Persist the directory entries for both the data dir and xl.meta renames; + // without this the commit itself can vanish on power loss. Relaxed tiers + // accept that window (documented in docs/operations/durability-modes.md). + if durability.syncs_commit_metadata() + && let Some(parent) = dst_file_path.parent() + { + let fsync_started = rustfs_io_metrics::put_stage_timer(); + if let Err(err) = os::fsync_dst_dir_group_commit(parent).await { + rustfs_io_metrics::record_put_object_stage_duration_from( + rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_DST_DIR_FSYNC, + fsync_started, + ); + rollback_committed_rename_std(&dst_file_path, committed_new_data_path, rollback_data_dir) + .map_err(to_file_error)?; + // The commit rename changed the dst part inodes before this fsync + // failed and rolled them back; drop any fd cached during that + // window so readers re-open the restored inode (rustfs/backlog#1177). + for part_path in &invalidate_part_paths { + self.io_backend.invalidate_cached_fd(dst_volume, part_path).await; + } + return Err(to_file_error(err).into()); + } + rustfs_io_metrics::record_put_object_stage_duration_from( + rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_DST_DIR_FSYNC, + fsync_started, + ); + } + + // First PUT of an object creates its directory (and any missing prefix + // dirs) via reliable_mkdir_all, which never fsyncs the parent chain. The + // commit fsync above persists the object dir's *contents*, not its own + // entry in the bucket/prefix dir, so on power loss after ack the whole + // object dir could vanish (rustfs/backlog#922 step 4). For a new object + // (no prior xl.meta) fsync the ancestor chain from the object dir's + // parent up to and including the bucket so those new directory entries + // are durable. Overwrites already have a durable object dir. The + // starts_with guard bounds the walk to the bucket subtree. Relaxed/none + // accept the wider window, like the commit fsync above. + if has_dst_buf.is_none() && durability.syncs_commit_metadata() { + let mut ancestor = dst_file_path.parent().and_then(|object_dir| object_dir.parent()); + while let Some(dir) = ancestor { + if !dir.starts_with(&dst_volume_dir) { + break; + } + let fsync_started = rustfs_io_metrics::put_stage_timer(); + if let Err(err) = os::fsync_dir(dir).await { + rustfs_io_metrics::record_put_object_stage_duration_from( + rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_ANCESTOR_DIR_FSYNC, + fsync_started, + ); + rollback_committed_rename_std(&dst_file_path, committed_new_data_path, rollback_data_dir) + .map_err(to_file_error)?; + // Same post-commit rollback window as above — drop cached + // dst part fds so readers re-open the restored inode + // (rustfs/backlog#1177). + for part_path in &invalidate_part_paths { + self.io_backend.invalidate_cached_fd(dst_volume, part_path).await; + } + return Err(to_file_error(err).into()); + } + rustfs_io_metrics::record_put_object_stage_duration_from( + rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_ANCESTOR_DIR_FSYNC, + fsync_started, + ); + if dir == dst_volume_dir.as_path() { + break; + } + ancestor = dir.parent(); + } + } + + // Publication and every rollback-capable durability step are now + // complete. Do not retain the Windows object identity guard while + // cleaning staging paths or invalidating cached descriptors. + #[cfg(windows)] + drop(rename_commit_guard); + + if let Some(src_file_path_parent) = src_file_path.parent() { + if src_volume != super::super::RUSTFS_META_MULTIPART_BUCKET { + let _ = std::fs::remove_dir(src_file_path_parent); + } else { + let _ = self + .delete_file(&dst_volume_dir, &src_file_path_parent.to_path_buf(), true, false) + .await; + } + } + + // Heal reuses a version's `data_dir` and lands the rebuilt shard on + // the SAME `//part.N` path. Without this, a cached + // descriptor would keep serving the pre-heal inode, defeating the heal + // and eroding read quorum (backlog#1145). + // + // The exact keys are derivable here, and this runs on every write, so + // use them rather than registering a predicate the read path would then + // have to evaluate. Readers build the same string + // (`{object}/{data_dir}/part.{n}`), and `fi.parts` enumerates every + // part of the version now at `dst_path` — any part path absent from it + // no longer exists for readers to ask for. + for part_path in &invalidate_part_paths { + self.io_backend.invalidate_cached_fd(dst_volume, part_path).await; + } + + Ok(RenameDataResp { + old_data_dir: has_old_data_dir, + rollback_data_dir, + cleanup_data_dir: has_old_data_dir, + sign: version_signature, + old_current_size, + }) + } else { + // Inline metadata preparation is blocking. The transaction lease is + // moved into that work so a timeout can release the async waiter without + // allowing a retry to reuse the deterministic staging path too early. + let src = src_file_path.clone(); + let dst = dst_file_path.clone(); + let cleanup_path = if src_volume == super::super::RUSTFS_META_MULTIPART_BUCKET { + src_file_path.parent().map(|p| p.to_path_buf()) + } else { + None + }; + let dst_path_for_failpoint = dst_path.to_string(); + #[cfg(windows)] + let source_parent = src_file_parent.to_path_buf(); + let rename_commit_guard_for_preparation = rename_commit_guard.clone(); + let sync = durability.syncs_commit_metadata(); + #[cfg(test)] + run_inline_before_file_sync_admission(dst_path); + let mut file_sync_admission = if sync { + Some( + os::acquire_file_sync_admission(self.file_sync_permits.clone()) + .await + .map_err(to_file_error) + .map_err(DiskError::from)?, + ) + } else { + None + }; + let prepare_inline_metadata = move || { + let mut prepared_metadata_source = + os::create_prepared_rename_source_with_commit_guard(&src, &dst, &rename_commit_guard_for_preparation)?; + #[cfg(windows)] + let source_metadata_guard = + rename_commit_guard_for_preparation.lock_source_directory_for_path_access(&source_parent)?; + let mut xlmeta = FileMeta::new(); + // Same as the non-inline branch: an unparsable existing dst + // xl.meta must surface as unknown, not `Absent` + // (rustfs/backlog#1009). + let mut dst_meta_unparsable = false; + if let Some(ref buf) = has_dst_buf { + if FileMeta::is_xl2_v1_format(buf) + && let Ok(nmeta) = FileMeta::load(buf) + { + xlmeta = nmeta + } else { + dst_meta_unparsable = true; + } + } + + let old_current_size = if dst_meta_unparsable { + None + } else { + observe_old_current_size(has_dst_buf.is_some(), &xlmeta) + }; + + let version_id = fi.version_id.unwrap_or_default(); + let old_data_dir = xlmeta.find_unshared_data_dir_for_version(Some(version_id)); + let old_version_exists = xlmeta.find_version(Some(version_id)).is_ok(); + let rollback_data_dir = old_data_dir.or_else(|| { + if old_version_exists && has_dst_buf.is_some() { + Some(inline_metadata_rollback_dir(version_id, &xlmeta)) + } else { + None + } + }); + let mut staged_rollback_path = None; + if let Some(d) = old_data_dir.as_ref() { + let _ = xlmeta.data.remove_two(version_id, *d); + } + xlmeta.add_version(fi)?; + let version_signature = rename_data_versions_signature(&xlmeta); + let new_buf = xlmeta.marshal_msg()?; + // Write the staged xl.meta. Inline objects carry their data inside + // xl.meta, so this is the durable preparation for the metadata commit: + // relaxed tiers do no per-object fsync here at all (aligned + // with MinIO's default), trading a documented power-loss + // window for latency. + prepared_metadata_source.write_all(&new_buf, sync)?; + run_inline_preparation_before_backup(&dst_path_for_failpoint); + if let Some(ref old_metadata) = has_dst_buf + && (rollback_data_dir.is_some() || sync || cfg!(test)) + { + #[cfg(windows)] + let backup_path = { + let backup_path = src + .parent() + .ok_or_else(|| std::io::Error::new(ErrorKind::InvalidInput, "missing staging metadata parent"))? + .join(STORAGE_FORMAT_FILE_BACKUP); + source_metadata_guard.write_file_for_path_access(&backup_path, old_metadata, sync, false)?; + backup_path + }; + #[cfg(not(windows))] + let backup_path = create_local_inline_rollback_backup(&dst, &src, old_metadata)?; + #[cfg(not(windows))] + if sync { + std::fs::File::open(&backup_path)?.sync_data()?; + } + staged_rollback_path = Some(backup_path); + } + + Ok::<_, std::io::Error>(( + rollback_data_dir, + old_data_dir, + version_signature, + old_current_size, + staged_rollback_path, + has_dst_buf.is_none(), + prepared_metadata_source, + )) + }; + let inline_preparation = if let Some(admission) = file_sync_admission.as_ref() { + os::run_blocking_namespace_file_sync_operation(mutation_lease.clone(), admission, prepare_inline_metadata).await + } else { + os::run_blocking_namespace_operation(mutation_lease.clone(), prepare_inline_metadata).await + } + .map_err(to_file_error) + .map_err(DiskError::from); + + let ( + rollback_data_dir, + cleanup_data_dir, + version_signature, + old_current_size, + mut local_rollback_path, + destination_was_absent, + prepared_metadata_source, + ) = match inline_preparation { + Ok(prepared) => prepared, + Err(err) => { + for part_path in &invalidate_part_paths { + self.io_backend.invalidate_cached_fd(dst_volume, part_path).await; + } + return Err(err); + } + }; + + let rename_commit_guard = remove_dst_base_before_commit( + dst_path, + rename_commit_guard, + src_file_parent, + dst_file_parent, + &dst_volume_dir, + &self.publication_root, + mutation_lease.clone(), + ) + .await?; + + if should_remove_staged_meta_before_commit(dst_path) { + drop(prepared_metadata_source); + let remove_result = std::fs::remove_file(&src_file_path); + if let Some(backup_path) = local_rollback_path.as_deref() { + let _ = remove_file_if_exists(backup_path); + } + remove_result.map_err(to_file_error)?; + return Err(DiskError::FileNotFound); + } + + if let (Some(rollback_data_dir), Some(staged_backup)) = (rollback_data_dir, local_rollback_path.as_deref()) { + let Some(dst_parent) = dst_file_path.parent() else { + return Err(DiskError::other("missing object metadata parent")); + }; + let backup_path = dst_parent + .join(rollback_data_dir.to_string()) + .join(STORAGE_FORMAT_FILE_BACKUP); + // rename_all acquires the backup path's namespace lease. Do not + // hold a disk admission while acquiring another namespace lock. + drop(file_sync_admission.take()); + if let Err(err) = rename_all(staged_backup, &backup_path, &dst_volume_dir, &self.publication_root).await { + let _ = remove_file_if_exists(staged_backup); + return Err(err); + } + #[cfg(test)] + run_rename_data_after_first_publication(&self.root, dst_volume, dst_path); + if sync { + file_sync_admission = Some( + os::acquire_file_sync_admission(self.file_sync_permits.clone()) + .await + .map_err(to_file_error) + .map_err(DiskError::from)?, + ); + } + if let Some(admission) = file_sync_admission.as_ref() + && let Some(backup_parent) = backup_path.parent() + { + let fsync_started = rustfs_io_metrics::put_stage_timer(); + if let Err(err) = + os::fsync_dir_with_namespace_file_sync_limit(backup_parent, mutation_lease.clone(), admission).await + { + rustfs_io_metrics::record_put_object_stage_duration_from( + rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_BACKUP_DIR_FSYNC, + fsync_started, + ); + return Err(DiskError::from(to_file_error(err))); + } + rustfs_io_metrics::record_put_object_stage_duration_from( + rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_BACKUP_DIR_FSYNC, + fsync_started, + ); + } + local_rollback_path = None; + } + + let commit_result = if should_fail_commit_rename(dst_path) { + Err(DiskError::other("test fail during metadata commit rename")) + } else { + os::rename_all_with_prepared_source( + prepared_metadata_source, + &src_file_path, + &dst_file_path, + &dst_volume_dir, + &self.publication_root, + &rename_commit_guard, + mutation_lease.clone(), + ) + .await + }; + if let Err(err) = commit_result { + if let Some(backup_path) = local_rollback_path.as_deref() { + let _ = remove_file_if_exists(backup_path); + } + for part_path in &invalidate_part_paths { + self.io_backend.invalidate_cached_fd(dst_volume, part_path).await; + } + return Err(err); + } + + let post_commit = async { + if should_fail_after_metadata_commit(dst_path) { + rollback_inline_metadata_commit_std(&dst_file_path, rollback_data_dir, local_rollback_path.as_deref())?; + return Err(std::io::Error::other("test fail after metadata commit")); + } + + // Persist the commit rename's directory entry across power loss. + if let Some(admission) = file_sync_admission.as_ref() + && let Some(dst_parent) = dst_file_path.parent() + { + let fsync_started = rustfs_io_metrics::put_stage_timer(); + if let Err(err) = + os::fsync_dst_dir_group_commit_or_namespace_file_sync_limit(dst_parent, mutation_lease.clone(), admission) + .await + { + rustfs_io_metrics::record_put_object_stage_duration_from( + rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_DST_DIR_FSYNC, + fsync_started, + ); + rollback_inline_metadata_commit_std(&dst_file_path, rollback_data_dir, local_rollback_path.as_deref())?; + return Err(err); + } + rustfs_io_metrics::record_put_object_stage_duration_from( + rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_DST_DIR_FSYNC, + fsync_started, + ); + } + + // Same power-loss gap as the non-inline path (rustfs/backlog#922 + // step 4): a first PUT creates the object dir (and any missing + // prefix dirs) whose entry in the bucket/prefix dir reliable_mkdir_all + // never fsynced. The fsync above persists the object dir's contents, + // not its own entry, so for a new inline object fsync the ancestor + // chain up to and including the bucket. Overwrites already have a + // durable object dir; the starts_with guard bounds the walk. + if let Some(admission) = file_sync_admission.as_ref() + && destination_was_absent + { + let mut ancestor = dst_file_path.parent().and_then(|object_dir| object_dir.parent()); + while let Some(ancestor_dir) = ancestor { + if !ancestor_dir.starts_with(&dst_volume_dir) { + break; + } + let fsync_started = rustfs_io_metrics::put_stage_timer(); + if let Err(err) = + os::fsync_dir_with_namespace_file_sync_limit(ancestor_dir, mutation_lease.clone(), admission).await + { + rustfs_io_metrics::record_put_object_stage_duration_from( + rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_ANCESTOR_DIR_FSYNC, + fsync_started, + ); + rollback_inline_metadata_commit_std( + &dst_file_path, + rollback_data_dir, + local_rollback_path.as_deref(), + )?; + return Err(err); + } + rustfs_io_metrics::record_put_object_stage_duration_from( + rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_ANCESTOR_DIR_FSYNC, + fsync_started, + ); + if ancestor_dir == dst_volume_dir.as_path() { + break; + } + ancestor = ancestor_dir.parent(); + } + } + + Ok::<(), std::io::Error>(()) + } + .await; + + // The disk admission protects the durability chain, not staging + // cleanup or cache invalidation after that chain has completed. + drop(file_sync_admission.take()); + + // A post-commit rollback (for example, a commit-metadata fsync + // failure under strict durability) restores the old metadata; drop any + // descriptors cached during the committed window before propagating the + // error (rustfs/backlog#1177). Inline objects carry data in xl.meta, so + // this is mostly defensive and keeps both commit branches consistent. + if let Err(err) = post_commit { + for part_path in &invalidate_part_paths { + self.io_backend.invalidate_cached_fd(dst_volume, part_path).await; + } + return Err(DiskError::from(err)); + } + + // The commit no longer has a rollback path. Release the Windows + // object identity guard before best-effort staging cleanup. + #[cfg(windows)] + drop(rename_commit_guard); + + if let Some(backup_path) = local_rollback_path.as_deref() { + let _ = remove_file_if_exists(backup_path); + } + + // Cleanup + if let Some(ref cleanup) = cleanup_path { + let _ = self.delete_file(&dst_volume_dir, cleanup, true, false).await; + } else if let Some(parent) = src_file_path.parent() { + let _ = std::fs::remove_dir(parent); + } + + // Heal reuses a version's `data_dir` and lands the rebuilt shard on + // the SAME `//part.N` path. Without this, a cached + // descriptor would keep serving the pre-heal inode, defeating the heal + // and eroding read quorum (backlog#1145). + // + // The exact keys are derivable here, and this runs on every write, so + // use them rather than registering a predicate the read path would then + // have to evaluate. Readers build the same string + // (`{object}/{data_dir}/part.{n}`), and `fi.parts` enumerates every + // part of the version now at `dst_path` — any part path absent from it + // no longer exists for readers to ask for. + for part_path in &invalidate_part_paths { + self.io_backend.invalidate_cached_fd(dst_volume, part_path).await; + } + + Ok(RenameDataResp { + old_data_dir: cleanup_data_dir, + rollback_data_dir, + cleanup_data_dir, + sign: version_signature, + old_current_size, + }) + } + } +}