From a269f8df0595adeffe6b7e14df79d78f1f774c27 Mon Sep 17 00:00:00 2001 From: GatewayJ <835269233@qq.com> Date: Fri, 10 Jul 2026 15:57:09 +0800 Subject: [PATCH] fix(ecstore): require write quorum for metadata early stop (#4300) --- crates/ecstore/src/disk/disk_store.rs | 1 + crates/ecstore/src/disk/local.rs | 984 +++++++++++++++++- crates/ecstore/src/disk/mod.rs | 4 + .../src/set_disk/core/io_primitives.rs | 38 +- crates/ecstore/src/set_disk/mod.rs | 63 ++ crates/ecstore/src/set_disk/ops/object.rs | 177 +++- crates/ecstore/src/set_disk/read.rs | 42 +- 7 files changed, 1241 insertions(+), 68 deletions(-) diff --git a/crates/ecstore/src/disk/disk_store.rs b/crates/ecstore/src/disk/disk_store.rs index 9c71b0c6c..5a7b71d77 100644 --- a/crates/ecstore/src/disk/disk_store.rs +++ b/crates/ecstore/src/disk/disk_store.rs @@ -812,6 +812,7 @@ impl LocalDiskWrapper { recursive: false, immediate: false, undo_write: false, + undo_delete: false, old_data_dir: None, }, ) diff --git a/crates/ecstore/src/disk/local.rs b/crates/ecstore/src/disk/local.rs index c8781d8c1..095c338d1 100644 --- a/crates/ecstore/src/disk/local.rs +++ b/crates/ecstore/src/disk/local.rs @@ -71,11 +71,166 @@ use uuid::Uuid; const DELETED_OBJECTS_CLEANUP_INTERVAL: Duration = Duration::from_secs(60 * 5); const STALE_TMP_OBJECT_EXPIRY: Duration = Duration::from_secs(24 * 60 * 60); const RUSTFS_META_TMP_OLD_BUCKET: &str = ".rustfs.sys/tmp-old"; +const INLINE_METADATA_ROLLBACK_DIR_XOR: u128 = 0x7275737466735f696e6c696e655f7262; +const DELETE_MARKER_ROLLBACK_FILE: &str = "xl.meta.delete-marker.rollback"; const STARTUP_CLEANUP_WAIT_TIMEOUT: Duration = Duration::from_secs(2); const ENV_BITROT_SIZE_MISMATCH_RETRY_COUNT: &str = "RUSTFS_BITROT_SIZE_MISMATCH_RETRY_COUNT"; const ENV_BITROT_SIZE_MISMATCH_RETRY_DELAY_MS: &str = "RUSTFS_BITROT_SIZE_MISMATCH_RETRY_DELAY_MS"; const DEFAULT_BITROT_SIZE_MISMATCH_RETRY_COUNT: u64 = 2; const DEFAULT_BITROT_SIZE_MISMATCH_RETRY_DELAY_MS: u64 = 100; + +fn inline_metadata_rollback_dir(version_id: Uuid, meta: &FileMeta) -> Uuid { + let used_data_dirs: HashSet = meta.get_data_dirs().unwrap_or_default().into_iter().flatten().collect(); + let base = version_id.as_u128() ^ INLINE_METADATA_ROLLBACK_DIR_XOR; + let mut salt = 0u128; + + loop { + let candidate = Uuid::from_u128(base ^ salt); + if !candidate.is_nil() && !used_data_dirs.contains(&candidate) { + return candidate; + } + salt = salt.wrapping_add(1); + } +} + +fn remove_file_if_exists(path: &Path) -> std::io::Result<()> { + match std::fs::remove_file(path) { + Ok(()) => Ok(()), + Err(err) if err.kind() == std::io::ErrorKind::NotFound => Ok(()), + Err(err) => Err(err), + } +} + +fn remove_dir_all_if_exists(path: &Path) -> std::io::Result<()> { + match std::fs::remove_dir_all(path) { + Ok(()) => Ok(()), + Err(err) if err.kind() == std::io::ErrorKind::NotFound => Ok(()), + Err(err) => Err(err), + } +} + +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(std::io::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(()) +} + +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)?; + fs::write(backup_dir.join(STORAGE_FORMAT_FILE_BACKUP), data) + .await + .map_err(to_file_error)?; + Ok(()) +} + +async fn restore_metadata_backup(object_dir: &Path, xl_path: &Path, rollback_dir: Uuid) -> Result<()> { + let backup_path = object_dir.join(rollback_dir.to_string()).join(STORAGE_FORMAT_FILE_BACKUP); + rename_all(&backup_path, xl_path, object_dir).await +} + +async fn restore_delete_rollback(object_dir: &Path, xl_path: &Path, rollback_dir: Uuid) -> Result<()> { + let rollback_path = object_dir.join(rollback_dir.to_string()); + let mut staged_paths = Vec::new(); + let mut remove_new_metadata = false; + match fs::read_dir(&rollback_path).await { + Ok(mut entries) => { + while let Some(entry) = entries.next_entry().await.map_err(to_file_error)? { + let name = entry.file_name(); + if name == DELETE_MARKER_ROLLBACK_FILE { + remove_new_metadata = true; + } else if name != STORAGE_FORMAT_FILE_BACKUP { + staged_paths.push((entry.path(), object_dir.join(name))); + } + } + } + Err(err) if err.kind() == ErrorKind::NotFound => return Ok(()), + Err(err) => return Err(to_file_error(err).into()), + } + + let had_staged_paths = !staged_paths.is_empty(); + for (src, dst) in staged_paths { + rename_all(&src, &dst, object_dir).await?; + } + + let backup_path = rollback_path.join(STORAGE_FORMAT_FILE_BACKUP); + match rename_all(&backup_path, xl_path, object_dir).await { + Ok(()) => { + let _ = fs::remove_dir(&rollback_path).await; + Ok(()) + } + // A missing backup only means "remove the newly-created delete marker" + // when the marker proves there was no old metadata to restore. + Err(DiskError::FileNotFound) if remove_new_metadata => match fs::remove_file(xl_path).await { + Ok(()) => { + let _ = fs::remove_file(rollback_path.join(DELETE_MARKER_ROLLBACK_FILE)).await; + let _ = fs::remove_dir(&rollback_path).await; + Ok(()) + } + Err(err) if err.kind() == ErrorKind::NotFound => { + let _ = fs::remove_file(rollback_path.join(DELETE_MARKER_ROLLBACK_FILE)).await; + let _ = fs::remove_dir(&rollback_path).await; + Ok(()) + } + Err(err) => Err(to_file_error(err).into()), + }, + Err(DiskError::FileNotFound) if had_staged_paths => Err(DiskError::FileNotFound), + Err(DiskError::FileNotFound) => match fs::metadata(xl_path).await { + Ok(_) => { + let _ = fs::remove_dir(&rollback_path).await; + Ok(()) + } + Err(err) if err.kind() == ErrorKind::NotFound => Err(DiskError::FileNotFound), + Err(err) => Err(to_file_error(err).into()), + }, + Err(err) => Err(err), + } +} + +async fn restore_delete_rollback_after_error( + object_dir: &Path, + xl_path: &Path, + rollback_dir: Option, + volume: &str, + path: &str, + stage: &'static str, + err: DiskError, +) -> DiskError { + let Some(rollback_dir) = rollback_dir else { + return err; + }; + + if let Err(restore_err) = restore_delete_rollback(object_dir, xl_path, rollback_dir).await { + warn!( + volume, + path, + rollback_dir = %rollback_dir, + stage, + cause = ?err, + error = ?restore_err, + "failed to restore delete rollback after local delete error" + ); + } + + err +} + const LOG_COMPONENT_ECSTORE: &str = "ecstore"; const LOG_SUBSYSTEM_DISK_LOCAL: &str = "disk_local"; const EVENT_DISK_LOCAL_STARTUP_CLEANUP: &str = "disk_local_startup_cleanup"; @@ -1171,6 +1326,10 @@ fn mmap_page_size() -> Result { #[cfg(test)] static RENAME_DATA_FAIL_BEFORE_OLD_METADATA_BACKUP: std::sync::Mutex> = std::sync::Mutex::new(None); +#[cfg(test)] +static RENAME_DATA_FAIL_AFTER_METADATA_COMMIT: std::sync::Mutex> = std::sync::Mutex::new(None); +#[cfg(test)] +static DELETE_VERSION_FAIL_AFTER_DATA_STAGED: std::sync::Mutex> = std::sync::Mutex::new(Vec::new()); #[cfg(test)] fn set_rename_data_fail_before_old_metadata_backup(dst_path: &str) { @@ -1179,6 +1338,21 @@ fn set_rename_data_fail_before_old_metadata_backup(dst_path: &str) { .expect("test failpoint lock should not be poisoned") = Some(dst_path.to_string()); } +#[cfg(test)] +fn set_rename_data_fail_after_metadata_commit(dst_path: &str) { + *RENAME_DATA_FAIL_AFTER_METADATA_COMMIT + .lock() + .expect("test failpoint lock should not be poisoned") = Some(dst_path.to_string()); +} + +#[cfg(test)] +fn set_delete_version_fail_after_data_staged(path: &str) { + DELETE_VERSION_FAIL_AFTER_DATA_STAGED + .lock() + .expect("test failpoint lock should not be poisoned") + .push(path.to_string()); +} + #[cfg(test)] fn should_fail_before_old_metadata_backup(dst_path: &str) -> bool { let mut target = RENAME_DATA_FAIL_BEFORE_OLD_METADATA_BACKUP @@ -1192,6 +1366,32 @@ fn should_fail_before_old_metadata_backup(dst_path: &str) -> bool { } } +#[cfg(test)] +fn should_fail_after_metadata_commit(dst_path: &str) -> bool { + let mut target = RENAME_DATA_FAIL_AFTER_METADATA_COMMIT + .lock() + .expect("test failpoint lock should not be poisoned"); + if target.as_deref() == Some(dst_path) { + target.take(); + true + } else { + false + } +} + +#[cfg(test)] +fn should_fail_after_delete_data_staged(path: &str) -> bool { + let mut targets = DELETE_VERSION_FAIL_AFTER_DATA_STAGED + .lock() + .expect("test failpoint lock should not be poisoned"); + if let Some(index) = targets.iter().position(|target| target == path) { + targets.remove(index); + true + } else { + false + } +} + #[cfg(not(test))] fn should_fail_before_old_metadata_backup(_dst_path: &str) -> bool { false @@ -1258,6 +1458,16 @@ fn should_crash_rename_data_at(_point: RenameDataCrashPoint, _dst_path: &str) -> false } +#[cfg(not(test))] +fn should_fail_after_metadata_commit(_dst_path: &str) -> bool { + false +} + +#[cfg(not(test))] +fn should_fail_after_delete_data_staged(_path: &str) -> bool { + false +} + fn log_startup_disk_io_error(stage: &str, path: &Path, err: &IoError) { warn!( event = EVENT_DISK_LOCAL_STARTUP_CLEANUP, @@ -3731,9 +3941,22 @@ impl LocalDisk { Ok((bytes, modtime)) } - async fn delete_versions_internal(&self, volume: &str, path: &str, fis: &[FileInfo]) -> Result<()> { + async fn delete_versions_internal(&self, volume: &str, path: &str, fis: &[FileInfo], opts: &DeleteOptions) -> Result<()> { let volume_dir = self.get_bucket_path(volume)?; let xlpath = self.get_object_path(volume, format!("{path}/{STORAGE_FORMAT_FILE}").as_str())?; + let object_dir = xlpath + .parent() + .ok_or_else(|| DiskError::other("missing object metadata parent"))?; + + if let Some(rollback_dir) = opts.old_data_dir + && opts.undo_write + { + if opts.undo_delete { + return restore_delete_rollback(object_dir, &xlpath, rollback_dir).await; + } + + return restore_metadata_backup(object_dir, &xlpath, rollback_dir).await; + } let (data, _) = self.read_all_data_with_dmtime(volume, volume_dir.as_path(), &xlpath).await?; @@ -3744,6 +3967,10 @@ impl LocalDisk { let mut fm = FileMeta::default(); fm.unmarshal_msg(&data)?; + let rollback_dir = opts.old_data_dir; + if let Some(rollback_dir) = rollback_dir { + write_metadata_rollback_backup(object_dir, rollback_dir, &data).await?; + } for fi in fis.iter() { let data_dir = match fm.delete_version(fi) { @@ -3754,7 +3981,16 @@ impl LocalDisk { continue; } - return Err(err); + return Err(restore_delete_rollback_after_error( + object_dir, + &xlpath, + rollback_dir, + volume, + path, + "delete_versions_metadata_update", + err, + ) + .await); } }; @@ -3762,8 +3998,64 @@ impl LocalDisk { let vid = fi.version_id.unwrap_or_default(); let _ = fm.data.remove(vec![vid, dir]); - let dir_path = self.get_object_path(volume, format!("{path}/{dir}").as_str())?; - if let Err(err) = self.move_to_trash(&dir_path, true, false).await + let dir_path = match self.get_object_path(volume, format!("{path}/{dir}").as_str()) { + Ok(dir_path) => dir_path, + Err(err) => { + return Err(restore_delete_rollback_after_error( + object_dir, + &xlpath, + rollback_dir, + volume, + path, + "delete_versions_data_path", + err, + ) + .await); + } + }; + if let Some(rollback_dir) = rollback_dir { + let rollback_path = object_dir.join(rollback_dir.to_string()); + if let Err(err) = fs::create_dir_all(&rollback_path).await { + let err: DiskError = to_file_error(err).into(); + return Err(restore_delete_rollback_after_error( + object_dir, + &xlpath, + Some(rollback_dir), + volume, + path, + "delete_versions_rollback_dir", + err, + ) + .await); + } + let rollback_data_path = rollback_path.join(dir.to_string()); + if let Err(err) = rename_all(&dir_path, &rollback_data_path, &rollback_path).await + && !(err == DiskError::FileNotFound || err == DiskError::VolumeNotFound) + { + return Err(restore_delete_rollback_after_error( + object_dir, + &xlpath, + Some(rollback_dir), + volume, + path, + "delete_versions_stage_data", + err, + ) + .await); + } + if should_fail_after_delete_data_staged(path) { + return Err(restore_delete_rollback_after_error( + object_dir, + &xlpath, + Some(rollback_dir), + volume, + path, + "delete_versions_test_after_stage", + DiskError::Unexpected, + ) + .await); + } + } else if let Err(err) = self.move_to_trash(&dir_path, true, false).await && !(err == DiskError::FileNotFound || err == DiskError::VolumeNotFound) { return Err(err); @@ -3773,16 +4065,55 @@ impl LocalDisk { // Remove xl.meta when no versions remain if fm.versions.is_empty() { - self.delete_file(&volume_dir, &xlpath, true, false).await?; + if let Err(err) = self.delete_file(&volume_dir, &xlpath, true, false).await { + return Err(restore_delete_rollback_after_error( + object_dir, + &xlpath, + rollback_dir, + volume, + path, + "delete_versions_commit_delete", + err, + ) + .await); + } return Ok(()); } // Update xl.meta atomically: a concurrent reader or crash mid-write must // never observe a truncated xl.meta for versions that were not deleted. - let buf = fm.marshal_msg()?; + let buf = match fm.marshal_msg() { + Ok(buf) => buf, + Err(err) => { + let err: DiskError = err.into(); + return Err(restore_delete_rollback_after_error( + object_dir, + &xlpath, + rollback_dir, + volume, + path, + "delete_versions_metadata_encode", + err, + ) + .await); + } + }; - self.write_all_meta(volume, format!("{path}/{STORAGE_FORMAT_FILE}").as_str(), &buf, true) - .await?; + if let Err(err) = self + .write_all_meta(volume, format!("{path}/{STORAGE_FORMAT_FILE}").as_str(), &buf, true) + .await + { + return Err(restore_delete_rollback_after_error( + object_dir, + &xlpath, + rollback_dir, + volume, + path, + "delete_versions_commit_write", + err, + ) + .await); + } Ok(()) } @@ -5672,6 +6003,14 @@ impl DiskAPI for LocalDisk { 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); } @@ -5775,7 +6114,7 @@ impl DiskAPI for LocalDisk { } else { SyncMode::None }; - if let Some(old_data_dir) = has_old_data_dir + if let Some(old_data_dir) = rollback_data_dir && let Some(dst_buf) = has_dst_buf.as_ref() && let Err(err) = self .write_all_private( @@ -5826,6 +6165,13 @@ impl DiskAPI for LocalDisk { 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); + } + // 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). @@ -5833,6 +6179,8 @@ impl DiskAPI for LocalDisk { && let Some(parent) = dst_file_path.parent() && let Err(err) = os::fsync_dir(parent).await { + rollback_committed_rename_std(&dst_file_path, committed_new_data_path, rollback_data_dir) + .map_err(to_file_error)?; return Err(to_file_error(err).into()); } @@ -5852,7 +6200,11 @@ impl DiskAPI for LocalDisk { if !dir.starts_with(&dst_volume_dir) { break; } - os::fsync_dir(dir).await.map_err(to_file_error)?; + if let Err(err) = os::fsync_dir(dir).await { + rollback_committed_rename_std(&dst_file_path, committed_new_data_path, rollback_data_dir) + .map_err(to_file_error)?; + return Err(to_file_error(err).into()); + } if dir == dst_volume_dir.as_path() { break; } @@ -5871,7 +6223,7 @@ impl DiskAPI for LocalDisk { } Ok(RenameDataResp { - old_data_dir: has_old_data_dir, + old_data_dir: rollback_data_dir, sign: version_signature, old_current_size, }) @@ -5887,6 +6239,7 @@ impl DiskAPI for LocalDisk { None }; + let dst_path_for_failpoint = dst_path.to_string(); let (old_data_dir, version_signature, old_current_size) = tokio::task::spawn_blocking(move || { // Read existing xl.meta let has_dst_buf = match std::fs::read(&dst) { @@ -5918,6 +6271,14 @@ impl DiskAPI for LocalDisk { 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 + } + }); if let Some(d) = old_data_dir.as_ref() { let _ = xlmeta.data.remove_two(version_id, *d); } @@ -5943,7 +6304,7 @@ impl DiskAPI for LocalDisk { if sync { f.sync_data()?; } - if let Some(old_dir) = old_data_dir.as_ref() + if let Some(old_dir) = rollback_data_dir.as_ref() && let Some(ref buf) = has_dst_buf && let Some(dst_parent) = dst.parent() { @@ -5986,9 +6347,18 @@ impl DiskAPI for LocalDisk { Err(err) => Err(to_file_error(err)), }?; + if should_fail_after_metadata_commit(&dst_path_for_failpoint) { + rollback_committed_rename_std(&dst, None, rollback_data_dir)?; + return Err(std::io::Error::other("test fail after metadata commit")); + } + // Persist the commit rename's directory entry across power loss. - if sync && let Some(dst_parent) = dst.parent() { - os::fsync_dir_std(dst_parent)?; + if sync + && let Some(dst_parent) = dst.parent() + && let Err(err) = os::fsync_dir_std(dst_parent) + { + rollback_committed_rename_std(&dst, None, rollback_data_dir)?; + return Err(err); } // Same power-loss gap as the non-inline path (rustfs/backlog#922 @@ -6004,7 +6374,10 @@ impl DiskAPI for LocalDisk { if !ancestor_dir.starts_with(&bucket_dir) { break; } - os::fsync_dir_std(ancestor_dir)?; + if let Err(err) = os::fsync_dir_std(ancestor_dir) { + rollback_committed_rename_std(&dst, None, rollback_data_dir)?; + return Err(err); + } if ancestor_dir == bucket_dir.as_path() { break; } @@ -6013,7 +6386,7 @@ impl DiskAPI for LocalDisk { } Ok::<(Option, Option>, Option), std::io::Error>(( - old_data_dir, + rollback_data_dir, version_signature, old_current_size, )) @@ -6349,6 +6722,32 @@ impl DiskAPI for LocalDisk { check_path_length(file_path.to_string_lossy().as_ref())?; let xl_path = path_join(&[file_path.as_path(), Path::new(STORAGE_FORMAT_FILE)]); + if let Some(old_data_dir) = opts.old_data_dir + && opts.undo_write + { + if opts.undo_delete { + restore_delete_rollback(file_path.as_path(), &xl_path, old_data_dir).await?; + } else { + restore_metadata_backup(file_path.as_path(), &xl_path, old_data_dir).await?; + } + + if !opts.undo_delete + && let Some(new_data_dir) = fi.data_dir + { + let new_data_path = path_join(&[file_path.as_path(), Path::new(new_data_dir.to_string().as_str())]); + check_path_length(new_data_path.to_string_lossy().as_ref())?; + if let Err(err) = self.move_to_trash(&new_data_path, true, false).await + && err != DiskError::FileNotFound + && err != DiskError::VolumeNotFound + { + return Err(err); + } + } + + return Ok(()); + } + + let rollback_dir = opts.old_data_dir; let buf = match self.read_all_data(volume, &volume_dir, &xl_path).await { Ok(res) => res, Err(err) => { @@ -6357,7 +6756,28 @@ impl DiskAPI for LocalDisk { } if fi.deleted && force_del_marker { - return self.write_metadata("", volume, path, fi).await; + if let Some(rollback_dir) = rollback_dir { + let rollback_path = file_path.join(rollback_dir.to_string()); + fs::create_dir_all(&rollback_path).await.map_err(to_file_error)?; + fs::write(rollback_path.join(DELETE_MARKER_ROLLBACK_FILE), []) + .await + .map_err(to_file_error)?; + } + if let Err(err) = self.write_metadata("", volume, path, fi).await { + if let Some(rollback_dir) = rollback_dir + && let Err(restore_err) = restore_delete_rollback(file_path.as_path(), &xl_path, rollback_dir).await + { + warn!( + volume, + path, + rollback_dir = %rollback_dir, + error = ?restore_err, + "failed to restore metadata after delete marker commit error" + ); + } + return Err(err); + } + return Ok(()); } return if fi.version_id.is_some() { @@ -6370,15 +6790,84 @@ impl DiskAPI for LocalDisk { let mut meta = FileMeta::load(&buf)?; let old_dir = meta.delete_version(&fi)?; + if let Some(rollback_dir) = rollback_dir { + write_metadata_rollback_backup(file_path.as_path(), rollback_dir, &buf).await?; + } if let Some(uuid) = old_dir { let vid = fi.version_id.unwrap_or_default(); - let _ = meta.data.remove(vec![vid, uuid])?; + if let Err(err) = meta.data.remove(vec![vid, uuid]) { + let err: DiskError = err.into(); + return Err(restore_delete_rollback_after_error( + file_path.as_path(), + &xl_path, + rollback_dir, + volume, + path, + "delete_version_metadata_update", + err, + ) + .await); + } let old_path = path_join(&[file_path.as_path(), Path::new(uuid.to_string().as_str())]); - check_path_length(old_path.to_string_lossy().as_ref())?; + if let Err(err) = check_path_length(old_path.to_string_lossy().as_ref()) { + return Err(restore_delete_rollback_after_error( + file_path.as_path(), + &xl_path, + rollback_dir, + volume, + path, + "delete_version_data_path", + err, + ) + .await); + } - if let Err(err) = self.move_to_trash(&old_path, true, false).await + if let Some(rollback_dir) = rollback_dir { + let rollback_path = file_path.join(rollback_dir.to_string()); + if let Err(err) = fs::create_dir_all(&rollback_path).await { + let err: DiskError = to_file_error(err).into(); + return Err(restore_delete_rollback_after_error( + file_path.as_path(), + &xl_path, + Some(rollback_dir), + volume, + path, + "delete_version_rollback_dir", + err, + ) + .await); + } + let rollback_data_path = rollback_path.join(uuid.to_string()); + if let Err(err) = rename_all(&old_path, &rollback_data_path, &rollback_path).await + && err != DiskError::FileNotFound + && err != DiskError::VolumeNotFound + { + return Err(restore_delete_rollback_after_error( + file_path.as_path(), + &xl_path, + Some(rollback_dir), + volume, + path, + "delete_version_stage_data", + err, + ) + .await); + } + if should_fail_after_delete_data_staged(path) { + return Err(restore_delete_rollback_after_error( + file_path.as_path(), + &xl_path, + Some(rollback_dir), + volume, + path, + "delete_version_test_after_stage", + DiskError::Unexpected, + ) + .await); + } + } else if let Err(err) = self.move_to_trash(&old_path, true, false).await && err != DiskError::FileNotFound && err != DiskError::VolumeNotFound { @@ -6386,35 +6875,56 @@ impl DiskAPI for LocalDisk { } } - if let Some(old_data_dir) = opts.old_data_dir - && opts.undo_write - { - let src_path = path_join(&[ + let commit_result = if !meta.versions.is_empty() { + let buf = match meta.marshal_msg() { + Ok(buf) => buf, + Err(err) => { + let err: DiskError = err.into(); + return Err(restore_delete_rollback_after_error( + file_path.as_path(), + &xl_path, + rollback_dir, + volume, + path, + "delete_version_metadata_encode", + err, + ) + .await); + } + }; + self.write_all_meta(volume, format!("{path}{SLASH_SEPARATOR}{STORAGE_FORMAT_FILE}").as_str(), &buf, true) + .await + } else { + self.delete_file(&volume_dir, &xl_path, true, false).await + }; + + if let Err(err) = commit_result { + return Err(restore_delete_rollback_after_error( file_path.as_path(), - Path::new(format!("{old_data_dir}{SLASH_SEPARATOR}{STORAGE_FORMAT_FILE_BACKUP}").as_str()), - ]); - let dst_path = path_join(&[file_path.as_path(), Path::new(STORAGE_FORMAT_FILE)]); - return rename_all(&src_path, &dst_path, file_path).await; + &xl_path, + rollback_dir, + volume, + path, + "delete_version_commit", + err, + ) + .await); } - if !meta.versions.is_empty() { - let buf = meta.marshal_msg()?; - return self - .write_all_meta(volume, format!("{path}{SLASH_SEPARATOR}{STORAGE_FORMAT_FILE}").as_str(), &buf, true) - .await; - } - - self.delete_file(&volume_dir, &xl_path, true, false).await + Ok(()) } #[tracing::instrument(level = "debug", skip(self))] - async fn delete_versions(&self, volume: &str, versions: Vec, _opts: DeleteOptions) -> Vec> { + async fn delete_versions(&self, volume: &str, versions: Vec, opts: DeleteOptions) -> Vec> { let mut errs = Vec::with_capacity(versions.len()); for _ in 0..versions.len() { errs.push(None); } for (i, ver) in versions.iter().enumerate() { - if let Err(e) = self.delete_versions_internal(volume, ver.name.as_str(), &ver.versions).await { + if let Err(e) = self + .delete_versions_internal(volume, ver.name.as_str(), &ver.versions, &opts) + .await + { errs[i] = Some(e); } else { errs[i] = None; @@ -6625,6 +7135,21 @@ mod test { meta.marshal_msg().expect("test metadata should encode") } + #[test] + fn inline_metadata_rollback_dir_avoids_real_data_dir_collision() { + let target_version = Uuid::parse_str("11111111-2222-3333-4444-555555555555").expect("version id should parse"); + let colliding_dir = Uuid::from_u128(target_version.as_u128() ^ INLINE_METADATA_ROLLBACK_DIR_XOR); + let other_version = Uuid::parse_str("66666666-7777-8888-9999-aaaaaaaaaaaa").expect("version id should parse"); + + let mut meta = FileMeta::new(); + meta.add_version(test_file_info("object", other_version, Some(colliding_dir), None)) + .expect("test metadata should accept file info"); + + let rollback_dir = inline_metadata_rollback_dir(target_version, &meta); + assert_ne!(rollback_dir, colliding_dir); + assert!(!rollback_dir.is_nil()); + } + async fn ensure_test_volume(disk: &LocalDisk, volume: &str) { match disk.make_volume(volume).await { Ok(()) | Err(DiskError::VolumeExists) => {} @@ -8650,6 +9175,50 @@ mod test { } } + #[tokio::test] + async fn test_rename_data_inline_post_commit_error_restores_old_metadata() { + use tempfile::tempdir; + + let dir = tempdir().expect("temp dir should be created"); + let endpoint = Endpoint::try_from(dir.path().to_str().expect("temp dir should be utf8")).expect("endpoint should parse"); + let disk = LocalDisk::new(&endpoint, false).await.expect("local disk should be created"); + + let bucket = "bucket"; + let object = "inline-post-commit-object"; + let tmp_object = "tmp-inline-post-commit-write"; + let version_id = Uuid::parse_str("99999999-9999-9999-9999-999999999999").expect("version id should parse"); + + ensure_test_volume(&disk, bucket).await; + ensure_test_volume(&disk, RUSTFS_META_TMP_BUCKET).await; + + let old_fi = test_file_info(object, version_id, None, Some(Bytes::from_static(b"inline-old"))); + let old_meta = test_meta(old_fi); + let dst_object_dir = dir.path().join(bucket).join(object); + fs::create_dir_all(&dst_object_dir) + .await + .expect("object dir should be created"); + fs::write(dst_object_dir.join(STORAGE_FORMAT_FILE), old_meta.clone()) + .await + .expect("old metadata should be written"); + + let tmp_object_dir = dir.path().join(RUSTFS_META_TMP_BUCKET).join(tmp_object); + fs::create_dir_all(&tmp_object_dir) + .await + .expect("tmp object dir should be created"); + + set_rename_data_fail_after_metadata_commit(object); + let new_fi = test_file_info(object, version_id, None, Some(Bytes::from_static(b"inline-new"))); + let result = disk + .rename_data(RUSTFS_META_TMP_BUCKET, tmp_object, new_fi, bucket, object) + .await; + + assert!(result.is_err()); + let restored_meta = fs::read(dst_object_dir.join(STORAGE_FORMAT_FILE)) + .await + .expect("old metadata should still be readable"); + assert_eq!(restored_meta, old_meta); + } + #[tokio::test] async fn test_delete_version_undo_restores_backup_to_object_root() { use tempfile::tempdir; @@ -8815,7 +9384,7 @@ mod test { .await .expect("existing metadata should be written"); - disk.delete_versions_internal(bucket, object, &[missing_fi, existing_fi]) + disk.delete_versions_internal(bucket, object, &[missing_fi, existing_fi], &DeleteOptions::default()) .await .expect("missing non-deleted version should not abort deletion"); @@ -8832,6 +9401,342 @@ mod test { ); } + #[tokio::test] + async fn test_delete_version_rollback_restores_staged_data_dir() { + use tempfile::tempdir; + + let dir = tempdir().expect("temp dir should be created"); + let endpoint = Endpoint::try_from(dir.path().to_str().expect("temp dir should be utf8")).expect("endpoint should parse"); + let disk = LocalDisk::new(&endpoint, false).await.expect("local disk should be created"); + + let bucket = "bucket"; + let object = "dir/object"; + let version_id = Uuid::parse_str("99999999-1111-2222-3333-444444444444").expect("version id should parse"); + let data_dir = Uuid::parse_str("88888888-1111-2222-3333-444444444444").expect("data dir should parse"); + let rollback_dir = Uuid::parse_str("77777777-1111-2222-3333-444444444444").expect("rollback dir should parse"); + + ensure_test_volume(&disk, bucket).await; + + let object_dir = dir.path().join(bucket).join("dir/object"); + let data_path = object_dir.join(data_dir.to_string()); + fs::create_dir_all(&data_path).await.expect("data dir should be created"); + fs::write(data_path.join("part.1"), b"old-data") + .await + .expect("part data should be written"); + + let old_fi = test_file_info(object, version_id, Some(data_dir), None); + let old_meta = test_meta(old_fi.clone()); + fs::write(object_dir.join(STORAGE_FORMAT_FILE), old_meta.clone()) + .await + .expect("old metadata should be written"); + + disk.delete_version( + bucket, + object, + old_fi.clone(), + false, + DeleteOptions { + old_data_dir: Some(rollback_dir), + ..Default::default() + }, + ) + .await + .expect("delete should stage rollback state"); + + assert!(!object_dir.join(STORAGE_FORMAT_FILE).exists()); + assert!(!data_path.exists()); + assert!( + object_dir + .join(rollback_dir.to_string()) + .join(STORAGE_FORMAT_FILE_BACKUP) + .exists() + ); + assert!( + object_dir + .join(rollback_dir.to_string()) + .join(data_dir.to_string()) + .join("part.1") + .exists() + ); + + disk.delete_version( + bucket, + object, + old_fi.clone(), + false, + DeleteOptions { + undo_write: true, + undo_delete: true, + old_data_dir: Some(rollback_dir), + ..Default::default() + }, + ) + .await + .expect("undo should restore metadata and data"); + + let restored_meta = fs::read(object_dir.join(STORAGE_FORMAT_FILE)) + .await + .expect("metadata should be restored"); + assert_eq!(restored_meta, old_meta); + assert_eq!( + fs::read(data_path.join("part.1")) + .await + .expect("part data should be restored"), + b"old-data" + ); + assert!(!object_dir.join(rollback_dir.to_string()).exists()); + + disk.delete_version( + bucket, + object, + old_fi.clone(), + false, + DeleteOptions { + undo_write: true, + undo_delete: true, + old_data_dir: Some(rollback_dir), + ..Default::default() + }, + ) + .await + .expect("repeated undo should be a no-op after rollback state is consumed"); + + let restored_meta = fs::read(object_dir.join(STORAGE_FORMAT_FILE)) + .await + .expect("metadata should remain restored after repeated undo"); + assert_eq!(restored_meta, old_meta); + assert_eq!( + fs::read(data_path.join("part.1")) + .await + .expect("part data should remain restored after repeated undo"), + b"old-data" + ); + + fs::create_dir_all(object_dir.join(rollback_dir.to_string())) + .await + .expect("stale empty rollback dir should be created"); + disk.delete_version( + bucket, + object, + old_fi, + false, + DeleteOptions { + undo_write: true, + undo_delete: true, + old_data_dir: Some(rollback_dir), + ..Default::default() + }, + ) + .await + .expect("undo with consumed backup and no delete-marker marker should be a no-op"); + + let restored_meta = fs::read(object_dir.join(STORAGE_FORMAT_FILE)) + .await + .expect("metadata should remain restored after stale-dir undo"); + assert_eq!(restored_meta, old_meta); + assert_eq!( + fs::read(data_path.join("part.1")) + .await + .expect("part data should remain restored after stale-dir undo"), + b"old-data" + ); + assert!(!object_dir.join(rollback_dir.to_string()).exists()); + } + + #[tokio::test] + async fn test_delete_version_error_after_staging_restores_data_dir() { + use tempfile::tempdir; + + let dir = tempdir().expect("temp dir should be created"); + let endpoint = Endpoint::try_from(dir.path().to_str().expect("temp dir should be utf8")).expect("endpoint should parse"); + let disk = LocalDisk::new(&endpoint, false).await.expect("local disk should be created"); + + let bucket = "bucket"; + let object = "dir/delete-version-error"; + let version_id = Uuid::parse_str("21212121-1111-2222-3333-444444444444").expect("version id should parse"); + let data_dir = Uuid::parse_str("22222222-1111-2222-3333-444444444444").expect("data dir should parse"); + let rollback_dir = Uuid::parse_str("23232323-1111-2222-3333-444444444444").expect("rollback dir should parse"); + + ensure_test_volume(&disk, bucket).await; + + let object_dir = dir.path().join(bucket).join(object); + let data_path = object_dir.join(data_dir.to_string()); + fs::create_dir_all(&data_path).await.expect("data dir should be created"); + fs::write(data_path.join("part.1"), b"old-data") + .await + .expect("part data should be written"); + + let old_fi = test_file_info(object, version_id, Some(data_dir), None); + let old_meta = test_meta(old_fi.clone()); + fs::write(object_dir.join(STORAGE_FORMAT_FILE), old_meta.clone()) + .await + .expect("old metadata should be written"); + + set_delete_version_fail_after_data_staged(object); + let err = disk + .delete_version( + bucket, + object, + old_fi, + false, + DeleteOptions { + old_data_dir: Some(rollback_dir), + ..Default::default() + }, + ) + .await + .expect_err("delete should fail after staging data"); + assert_eq!(err, DiskError::Unexpected); + + let restored_meta = fs::read(object_dir.join(STORAGE_FORMAT_FILE)) + .await + .expect("metadata should be restored"); + assert_eq!(restored_meta, old_meta); + assert_eq!( + fs::read(data_path.join("part.1")) + .await + .expect("part data should be restored"), + b"old-data" + ); + assert!(!object_dir.join(rollback_dir.to_string()).exists()); + } + + #[tokio::test] + async fn test_delete_versions_error_after_staging_restores_data_dir() { + use tempfile::tempdir; + + let dir = tempdir().expect("temp dir should be created"); + let endpoint = Endpoint::try_from(dir.path().to_str().expect("temp dir should be utf8")).expect("endpoint should parse"); + let disk = LocalDisk::new(&endpoint, false).await.expect("local disk should be created"); + + let bucket = "bucket"; + let object = "dir/delete-versions-error"; + let version_id = Uuid::parse_str("24242424-1111-2222-3333-444444444444").expect("version id should parse"); + let data_dir = Uuid::parse_str("25252525-1111-2222-3333-444444444444").expect("data dir should parse"); + let rollback_dir = Uuid::parse_str("26262626-1111-2222-3333-444444444444").expect("rollback dir should parse"); + + ensure_test_volume(&disk, bucket).await; + + let object_dir = dir.path().join(bucket).join(object); + let data_path = object_dir.join(data_dir.to_string()); + fs::create_dir_all(&data_path).await.expect("data dir should be created"); + fs::write(data_path.join("part.1"), b"old-data") + .await + .expect("part data should be written"); + + let old_fi = test_file_info(object, version_id, Some(data_dir), None); + let old_meta = test_meta(old_fi.clone()); + fs::write(object_dir.join(STORAGE_FORMAT_FILE), old_meta.clone()) + .await + .expect("old metadata should be written"); + + set_delete_version_fail_after_data_staged(object); + let errs = disk + .delete_versions( + bucket, + vec![FileInfoVersions { + name: object.to_string(), + versions: vec![old_fi], + ..Default::default() + }], + DeleteOptions { + old_data_dir: Some(rollback_dir), + ..Default::default() + }, + ) + .await; + assert_eq!(errs, vec![Some(DiskError::Unexpected)]); + + let restored_meta = fs::read(object_dir.join(STORAGE_FORMAT_FILE)) + .await + .expect("metadata should be restored"); + assert_eq!(restored_meta, old_meta); + assert_eq!( + fs::read(data_path.join("part.1")) + .await + .expect("part data should be restored"), + b"old-data" + ); + assert!(!object_dir.join(rollback_dir.to_string()).exists()); + } + + #[tokio::test] + async fn test_delete_marker_rollback_removes_new_metadata_without_backup() { + use tempfile::tempdir; + + let dir = tempdir().expect("temp dir should be created"); + let endpoint = Endpoint::try_from(dir.path().to_str().expect("temp dir should be utf8")).expect("endpoint should parse"); + let disk = LocalDisk::new(&endpoint, false).await.expect("local disk should be created"); + + let bucket = "bucket"; + let object = "missing/object"; + let rollback_dir = Uuid::parse_str("66666666-1111-2222-3333-444444444444").expect("rollback dir should parse"); + + ensure_test_volume(&disk, bucket).await; + + let object_dir = dir.path().join(bucket).join("missing/object"); + let delete_marker = FileInfo { + name: object.to_string(), + version_id: Some(Uuid::parse_str("55555555-1111-2222-3333-444444444444").expect("version id should parse")), + deleted: true, + mark_deleted: true, + mod_time: Some(OffsetDateTime::now_utc()), + ..Default::default() + }; + + disk.delete_version( + bucket, + object, + delete_marker.clone(), + true, + DeleteOptions { + old_data_dir: Some(rollback_dir), + ..Default::default() + }, + ) + .await + .expect("delete marker should be written"); + assert!(object_dir.join(STORAGE_FORMAT_FILE).exists()); + let rollback_path = object_dir.join(rollback_dir.to_string()); + assert!( + rollback_path.join(DELETE_MARKER_ROLLBACK_FILE).exists(), + "delete-marker rollback should carry an explicit no-backup marker" + ); + + disk.delete_version( + bucket, + object, + delete_marker.clone(), + true, + DeleteOptions { + undo_write: true, + undo_delete: true, + old_data_dir: Some(rollback_dir), + ..Default::default() + }, + ) + .await + .expect("undo should remove new delete marker metadata"); + assert!(!object_dir.join(STORAGE_FORMAT_FILE).exists()); + assert!(!rollback_path.exists()); + + disk.delete_version( + bucket, + object, + delete_marker, + true, + DeleteOptions { + undo_write: true, + undo_delete: true, + old_data_dir: Some(rollback_dir), + ..Default::default() + }, + ) + .await + .expect("repeated delete-marker undo should be a no-op"); + assert!(!object_dir.join(STORAGE_FORMAT_FILE).exists()); + } + #[tokio::test] async fn test_rename_data_failure_before_metadata_commit_preserves_old_metadata() { use tempfile::tempdir; @@ -10609,6 +11514,7 @@ mod test { recursive: false, immediate: true, undo_write: false, + undo_delete: false, old_data_dir: None, }; disk.delete("test-volume", "test-file.txt", delete_opts) diff --git a/crates/ecstore/src/disk/mod.rs b/crates/ecstore/src/disk/mod.rs index 66287e6dc..d8069ed58 100644 --- a/crates/ecstore/src/disk/mod.rs +++ b/crates/ecstore/src/disk/mod.rs @@ -883,6 +883,8 @@ pub struct DeleteOptions { pub recursive: bool, pub immediate: bool, pub undo_write: bool, + #[serde(default)] + pub undo_delete: bool, pub old_data_dir: Option, } @@ -1122,12 +1124,14 @@ mod tests { recursive: true, immediate: false, undo_write: true, + undo_delete: false, old_data_dir: Some(Uuid::new_v4()), }; assert!(opts.recursive); assert!(!opts.immediate); assert!(opts.undo_write); + assert!(!opts.undo_delete); assert!(opts.old_data_dir.is_some()); } diff --git a/crates/ecstore/src/set_disk/core/io_primitives.rs b/crates/ecstore/src/set_disk/core/io_primitives.rs index 994c29068..d3506770a 100644 --- a/crates/ecstore/src/set_disk/core/io_primitives.rs +++ b/crates/ecstore/src/set_disk/core/io_primitives.rs @@ -357,7 +357,7 @@ impl MetadataQuorumAccumulator { if !self.allow_early_stop { return None; } - if self.delete_marker_votes >= self.missing_response_quorum() { + if self.delete_marker_votes >= self.default_write_quorum() { return Some(MetadataEarlyStopDecision { reason: GET_METADATA_EARLY_STOP_REASON_DELETE_MARKER, }); @@ -373,8 +373,8 @@ impl MetadataQuorumAccumulator { if self .candidate .as_ref() - .and_then(|candidate| self.candidate_read_quorum(candidate)) - .is_some_and(|read_quorum| self.candidate_votes >= read_quorum) + .and_then(|candidate| self.candidate_latest_quorum(candidate)) + .is_some_and(|latest_quorum| self.candidate_votes >= latest_quorum) { return Some(MetadataEarlyStopDecision { reason: GET_METADATA_EARLY_STOP_REASON_VALID_QUORUM, @@ -441,14 +441,26 @@ impl MetadataQuorumAccumulator { GET_METADATA_EARLY_STOP_REASON_INSUFFICIENT_QUORUM } - pub(in crate::set_disk) fn candidate_read_quorum(&self, candidate: &FileInfo) -> Option { + pub(in crate::set_disk) fn candidate_latest_quorum(&self, candidate: &FileInfo) -> Option { if self.default_parity_count == 0 { return Some(self.total_disks); } if candidate.deleted || candidate.size == 0 || candidate.erasure.parity_blocks >= self.total_disks { return None; } - Some(self.total_disks.saturating_sub(candidate.erasure.parity_blocks)) + Some(candidate.write_quorum(self.default_write_quorum())) + } + + pub(in crate::set_disk) fn default_write_quorum(&self) -> usize { + if self.default_parity_count == 0 { + return self.total_disks; + } + let data_blocks = self.total_disks.saturating_sub(self.default_parity_count); + if data_blocks == self.default_parity_count { + data_blocks.saturating_add(1) + } else { + data_blocks + } } pub(in crate::set_disk) fn missing_response_quorum(&self) -> usize { @@ -1772,7 +1784,7 @@ pub(in crate::set_disk) fn should_allow_metadata_early_stop( } (is_get_metadata_early_stop_enabled() && version_id.is_empty() && !healing && !incl_free_versions) - || (is_version_early_stop_enabled() && !version_id.is_empty() && !healing) + || (is_version_early_stop_enabled() && !version_id.is_empty() && !healing && !incl_free_versions) } /// Final gate for the metadata early-stop fast path. @@ -1885,7 +1897,7 @@ impl SetDisks { read_data: bool, healing: bool, incl_free_versions: bool, - allow_early_stop: bool, + caller_allows_early_stop: bool, default_parity_count: usize, ) -> disk::error::Result<(Vec, Vec>, MetadataFanoutDiagnostics)> { Self::read_all_fileinfo_inner( @@ -1898,7 +1910,7 @@ impl SetDisks { healing, incl_free_versions, true, - allow_early_stop, + caller_allows_early_stop, default_parity_count, ) .await @@ -4064,24 +4076,24 @@ mod tests { } #[test] - fn metadata_quorum_accumulator_candidate_quorum_handles_zero_parity_and_invalid_candidates() { + fn metadata_quorum_accumulator_candidate_latest_quorum_handles_zero_parity_and_invalid_candidates() { let accumulator = MetadataQuorumAccumulator::new(4, 0, true); let candidate = metadata_test_fileinfo("object"); - assert_eq!(accumulator.candidate_read_quorum(&candidate), Some(4)); + assert_eq!(accumulator.candidate_latest_quorum(&candidate), Some(4)); assert_eq!(accumulator.missing_response_quorum(), 4); let accumulator = MetadataQuorumAccumulator::new(4, 2, true); let mut deleted = candidate.clone(); deleted.deleted = true; - assert_eq!(accumulator.candidate_read_quorum(&deleted), None); + assert_eq!(accumulator.candidate_latest_quorum(&deleted), None); let mut empty = candidate.clone(); empty.size = 0; - assert_eq!(accumulator.candidate_read_quorum(&empty), None); + assert_eq!(accumulator.candidate_latest_quorum(&empty), None); let mut impossible_parity = candidate; impossible_parity.erasure.parity_blocks = 4; - assert_eq!(accumulator.candidate_read_quorum(&impossible_parity), None); + assert_eq!(accumulator.candidate_latest_quorum(&impossible_parity), None); } #[test] diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index 2e2daa570..60af7e9a6 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/crates/ecstore/src/set_disk/mod.rs @@ -3903,6 +3903,69 @@ mod tests { ); } + #[tokio::test] + async fn test_rename_data_inline_quorum_failure_rolls_back_destination_object() { + let dir = tempfile::tempdir().expect("tempdir should be created"); + let disk_root = dir.path().join("disk0"); + fs::create_dir_all(&disk_root).await.expect("disk root should be created"); + let endpoint = Endpoint::try_from(disk_root.to_str().expect("disk path should be utf8")).expect("endpoint should parse"); + let disk = new_disk( + &endpoint, + &DiskOption { + cleanup: false, + health_check: false, + }, + ) + .await + .expect("disk should be created"); + + let bucket = "bucket"; + let object = "inline-object"; + let tmp_object = "tmp-inline-object"; + let version_id = Uuid::parse_str("aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa").expect("version id should parse"); + + match disk.make_volume(bucket).await { + Ok(()) | Err(DiskError::VolumeExists) => {} + Err(err) => panic!("bucket should be available: {err:?}"), + } + match disk.make_volume(RUSTFS_META_TMP_BUCKET).await { + Ok(()) | Err(DiskError::VolumeExists) => {} + Err(err) => panic!("tmp bucket should be available: {err:?}"), + } + + let object_dir = disk_root.join(bucket).join(object); + fs::create_dir_all(&object_dir).await.expect("object dir should be created"); + let mut old_fi = FileInfo::new(&format!("{bucket}/{object}"), 1, 1); + old_fi.name = object.to_string(); + old_fi.version_id = Some(version_id); + old_fi.data = Some(Bytes::from_static(b"old-inline")); + old_fi.size = 10; + old_fi.mod_time = Some(OffsetDateTime::now_utc()); + let mut old_meta = FileMeta::default(); + old_meta.add_version(old_fi).expect("old metadata should accept file info"); + let old_meta_buf = old_meta.marshal_msg().expect("old metadata should encode"); + fs::write(object_dir.join(STORAGE_FORMAT_FILE), old_meta_buf.clone()) + .await + .expect("old metadata should be written"); + + let mut new_fi = FileInfo::new(&format!("{bucket}/{object}"), 1, 1); + new_fi.name = object.to_string(); + new_fi.version_id = Some(version_id); + new_fi.data = Some(Bytes::from_static(b"new-inline")); + new_fi.size = 10; + new_fi.mod_time = Some(OffsetDateTime::now_utc()); + + let disks = vec![Some(disk), None]; + let file_infos = vec![new_fi.clone(), new_fi]; + let result = SetDisks::rename_data(&disks, RUSTFS_META_TMP_BUCKET, tmp_object, &file_infos, bucket, object, 2).await; + + assert!(result.is_err()); + let restored_meta = fs::read(object_dir.join(STORAGE_FORMAT_FILE)) + .await + .expect("destination metadata should remain readable"); + assert_eq!(restored_meta, old_meta_buf); + } + #[test] fn disk_health_entry_returns_cached_value_within_ttl() { let entry = DiskHealthEntry { diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index 91b162e18..469ad1976 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/crates/ecstore/src/set_disk/ops/object.rs @@ -78,7 +78,7 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks { }; let metadata_stage_start = Instant::now(); - let (fi, files, disks) = match self.get_object_fileinfo(bucket, object, opts, true).await { + let (fi, files, disks) = match self.get_object_fileinfo(bucket, object, opts, true, true).await { Ok(result) => result, Err(err) => { rustfs_io_metrics::record_get_object_metadata_phase_duration(metadata_stage_start.elapsed().as_secs_f64()); @@ -1312,6 +1312,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { async fn delete_object_version(&self, bucket: &str, object: &str, fi: &FileInfo, force_del_marker: bool) -> Result<()> { let disks = self.disk_inventory().await; let write_quorum = disks.len() / 2 + 1; + let rollback_dir = Uuid::new_v4(); let mut futures = Vec::with_capacity(disks.len()); let mut errs = Vec::with_capacity(disks.len()); @@ -1320,7 +1321,16 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { futures.push(async move { if let Some(disk) = disk { match disk - .delete_version(bucket, object, fi.clone(), force_del_marker, DeleteOptions::default()) + .delete_version( + bucket, + object, + fi.clone(), + force_del_marker, + DeleteOptions { + old_data_dir: Some(rollback_dir), + ..Default::default() + }, + ) .await { Ok(r) => Ok(r), @@ -1344,7 +1354,77 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { } } - resolve_tiered_decommission_write_quorum_result(&errs, write_quorum, bucket, object) + let quorum_result = resolve_tiered_decommission_write_quorum_result(&errs, write_quorum, bucket, object); + let should_rollback = quorum_result.is_err(); + let mut rollback_futures = Vec::new(); + for (index, err) in errs.iter().enumerate() { + if err.is_some() { + continue; + } + + let Some(disk) = disks[index].as_ref() else { + continue; + }; + + let disk = disk.clone(); + let bucket = bucket.to_string(); + let object = object.to_string(); + let fi = fi.clone(); + rollback_futures.push(async move { + if should_rollback { + if let Err(err) = disk + .delete_version( + &bucket, + &object, + fi, + force_del_marker, + DeleteOptions { + undo_write: true, + undo_delete: true, + old_data_dir: Some(rollback_dir), + ..Default::default() + }, + ) + .await + { + warn!( + bucket = %bucket, + object = %object, + rollback_dir = %rollback_dir, + error = ?err, + "failed to roll back delete after write quorum failure" + ); + } + } else { + let rollback_path = format!("{object}/{rollback_dir}"); + if let Err(err) = disk + .delete( + &bucket, + &rollback_path, + DeleteOptions { + recursive: true, + immediate: true, + ..Default::default() + }, + ) + .await + && err != DiskError::FileNotFound + && err != DiskError::VolumeNotFound + { + warn!( + bucket = %bucket, + object = %object, + rollback_dir = %rollback_dir, + error = ?err, + "failed to clean delete rollback state after quorum success" + ); + } + } + }); + } + + join_all(rollback_futures).await; + quorum_result } #[tracing::instrument(skip(self))] @@ -1542,6 +1622,8 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { vers.push(fi_vers); } + let rollback_dir = Uuid::new_v4(); + let disks = self.disks.read().await; let disks = disks.clone(); @@ -1554,7 +1636,15 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { let vers = vers.clone(); futures.push(async move { if let Some(disk) = disk { - disk.delete_versions(bucket, vers, DeleteOptions::default()).await + disk.delete_versions( + bucket, + vers, + DeleteOptions { + old_data_dir: Some(rollback_dir), + ..Default::default() + }, + ) + .await } else { let mut errs = Vec::with_capacity(vers.len()); for _ in 0..vers.len() { @@ -1618,6 +1708,77 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { record_capacity_scope_if_needed(opts.capacity_scope_token, &disks); + let mut rollback_futures = Vec::new(); + for fi_vers in &vers { + // delete_versions commits one xl.meta per object group, so rollback must use the same boundary. + let should_rollback = fi_vers.versions.iter().any(|fi| del_errs[fi.idx].is_some()); + for (disk_idx, disk) in disks.iter().enumerate() { + if fi_vers.versions.iter().any(|fi| del_obj_errs[disk_idx][fi.idx].is_some()) { + continue; + } + + let Some(disk) = disk.as_ref() else { + continue; + }; + + let disk = disk.clone(); + let bucket = bucket.to_string(); + let object = fi_vers.name.clone(); + let versions = fi_vers.clone(); + rollback_futures.push(async move { + if should_rollback { + let errs = disk + .delete_versions( + &bucket, + vec![versions], + DeleteOptions { + undo_write: true, + undo_delete: true, + old_data_dir: Some(rollback_dir), + ..Default::default() + }, + ) + .await; + if let Some(err) = errs.into_iter().flatten().next() { + warn!( + bucket = %bucket, + object = %object, + rollback_dir = %rollback_dir, + error = ?err, + "failed to roll back batch delete after write quorum failure" + ); + } + } else { + let rollback_path = format!("{object}/{rollback_dir}"); + if let Err(err) = disk + .delete( + &bucket, + &rollback_path, + DeleteOptions { + recursive: true, + immediate: true, + ..Default::default() + }, + ) + .await + && err != DiskError::FileNotFound + && err != DiskError::VolumeNotFound + { + warn!( + bucket = %bucket, + object = %object, + rollback_dir = %rollback_dir, + error = ?err, + "failed to clean batch delete rollback state after quorum success" + ); + } + } + }); + } + } + + join_all(rollback_futures).await; + // TODO: add_partial if dist_erasure { @@ -1789,7 +1950,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { // This avoids HEAD/GetObject metadata visibility skew immediately after // PutObject/CompleteMultipartUpload. let (fi, _, _) = self - .get_object_fileinfo(bucket, object, opts, true) + .get_object_fileinfo(bucket, object, opts, true, false) .await .map_err(|e| to_object_err(e, vec![bucket, object]))?; @@ -1935,7 +2096,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { // _lock_guard = guard_opt; // } - let (mut fi, meta_arr, online_disks) = self.get_object_fileinfo(bucket, object, opts, true).await?; + let (mut fi, meta_arr, online_disks) = self.get_object_fileinfo(bucket, object, opts, true, false).await?; /*if err != nil { return Err(to_object_err(err, vec![bucket, object])); }*/ @@ -1969,7 +2130,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { if let Err(err) = self.heal_object(bucket, object, "", &HealOpts {no_lock: true, ..Default::default()}) { return err.expect("err"); } - (fi, meta_arr, online_disks) = self.get_object_fileinfo(&bucket, &object, &opts, true); + (fi, meta_arr, online_disks) = self.get_object_fileinfo(&bucket, &object, &opts, true, false); if err != nil { return to_object_err(err, vec![bucket, object]); } @@ -2116,7 +2277,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { Err(rerr.unwrap()) }; let mut oi = ObjectInfo::default(); - let fi = self_.clone().get_object_fileinfo(bucket, object, opts, true).await; + let fi = self_.clone().get_object_fileinfo(bucket, object, opts, true, false).await; if let Err(err) = fi { return set_restore_header_fn(&mut oi, Some(to_object_err(err, vec![bucket, object]))).await; } diff --git a/crates/ecstore/src/set_disk/read.rs b/crates/ecstore/src/set_disk/read.rs index 9f8042837..febe1fc9a 100644 --- a/crates/ecstore/src/set_disk/read.rs +++ b/crates/ecstore/src/set_disk/read.rs @@ -170,10 +170,10 @@ impl SetDisks { object: &str, opts: &ObjectOptions, read_data: bool, + caller_allows_early_stop: bool, ) -> Result<(FileInfo, Vec, Vec>)> { - // Read-only callers (GET/HEAD/tag read) may use the metadata early-stop - // fast path. - self.get_object_fileinfo_gated(bucket, object, opts, read_data, true).await + self.get_object_fileinfo_gated(bucket, object, opts, read_data, caller_allows_early_stop) + .await } /// Like `get_object_fileinfo`, but `allow_early_stop=false` forces the full @@ -329,7 +329,7 @@ impl SetDisks { object: &str, opts: &ObjectOptions, ) -> (ObjectInfo, usize, Option) { - let fi = match self.get_object_fileinfo(bucket, object, opts, false).await { + let fi = match self.get_object_fileinfo(bucket, object, opts, false, false).await { Ok((fi, _, _)) => fi, Err(e) => return (ObjectInfo::default(), 0, Some(e)), }; @@ -2783,6 +2783,8 @@ mod tests { accumulator.observe_file_info(&metadata_early_stop_candidate("object", 1)); assert!(accumulator.early_stop_decision().is_none()); accumulator.observe_file_info(&metadata_early_stop_candidate("object", 2)); + assert!(accumulator.early_stop_decision().is_none()); + accumulator.observe_file_info(&metadata_early_stop_candidate("object", 3)); assert_eq!( accumulator.early_stop_decision(), @@ -2790,7 +2792,7 @@ mod tests { reason: GET_METADATA_EARLY_STOP_REASON_VALID_QUORUM }) ); - assert_eq!(accumulator.valid_responses, 2); + assert_eq!(accumulator.valid_responses, 3); } #[test] @@ -2799,25 +2801,31 @@ mod tests { accumulator.observe_file_info(&metadata_early_stop_candidate("object", 1)); accumulator.observe_file_info(&metadata_early_stop_candidate("object", 2)); + assert!(accumulator.early_stop_decision().is_none()); + accumulator.observe_file_info(&metadata_early_stop_candidate("object", 3)); assert!(accumulator.early_stop_decision().is_some()); - assert_eq!(accumulator.candidate_votes, 2); + assert_eq!(accumulator.candidate_votes, 3); } #[test] fn metadata_quorum_accumulator_partial_result_remains_quorum_compatible() { let first = metadata_early_stop_candidate("object", 1); let second = metadata_early_stop_candidate("object", 2); - let parts_metadata = vec![first.clone(), second, FileInfo::default(), FileInfo::default()]; + let third = metadata_early_stop_candidate("object", 3); + let parts_metadata = vec![first.clone(), second, third, FileInfo::default()]; let errs = vec![None, None, None, None]; - let (read_quorum, _) = SetDisks::object_quorum_from_meta(&parts_metadata, &errs, 2) + let (read_quorum, write_quorum) = SetDisks::object_quorum_from_meta(&parts_metadata, &errs, 2) .expect("partial early-stop metadata should preserve read quorum"); let read_quorum = usize::try_from(read_quorum).expect("read quorum should be non-negative"); + let write_quorum = usize::try_from(write_quorum).expect("write quorum should be non-negative"); + let selection_quorum = SetDisks::latest_fileinfo_selection_quorum("", &parts_metadata, &errs, read_quorum, write_quorum); let selected = SetDisks::pick_valid_fileinfo(&parts_metadata, None, Some("etag-1".to_string()), read_quorum) .expect("partial early-stop metadata should preserve selected FileInfo"); assert_eq!(read_quorum, 2); + assert_eq!(selection_quorum, 3); assert_eq!(selected.name, first.name); assert_eq!(selected.get_etag(), first.get_etag()); } @@ -2870,6 +2878,8 @@ mod tests { accumulator.observe_file_info(&deleted); accumulator.observe_file_info(&deleted); + assert!(accumulator.early_stop_decision().is_none()); + accumulator.observe_file_info(&deleted); assert_eq!( accumulator.early_stop_decision(), @@ -3008,6 +3018,22 @@ mod tests { ); } + #[test] + fn metadata_early_stop_rejects_healing_and_free_version_requests() { + temp_env::with_vars( + [ + (ENV_RUSTFS_GET_METADATA_EARLY_STOP_ENABLE, Some("true")), + (ENV_RUSTFS_GET_METADATA_VERSION_EARLY_STOP_ENABLE, Some("true")), + ], + || { + assert!(!should_allow_metadata_early_stop(false, "", true, false)); + assert!(!should_allow_metadata_early_stop(false, "", false, true)); + assert!(!should_allow_metadata_early_stop(false, "version-id", true, false)); + assert!(!should_allow_metadata_early_stop(false, "version-id", false, true)); + }, + ); + } + #[test] fn version_early_stop_gate_defaults_to_disabled() { temp_env::with_var(ENV_RUSTFS_GET_METADATA_VERSION_EARLY_STOP_ENABLE, None::<&str>, || {