diff --git a/crates/ecstore/src/set_disk.rs b/crates/ecstore/src/set_disk.rs index 382035e37..125a17d71 100644 --- a/crates/ecstore/src/set_disk.rs +++ b/crates/ecstore/src/set_disk.rs @@ -1171,7 +1171,7 @@ impl ObjectIO for SetDisks { object_lock_guard = Some(self.acquire_write_lock_diag("put_object_commit", bucket, object).await?); } - let (online_disks, _, op_old_dir) = Self::rename_data( + let (online_disks, _, op_old_dir, cleanup_disks) = Self::rename_data( &shuffle_disks, RUSTFS_META_TMP_BUCKET, tmp_dir.as_str(), @@ -1183,7 +1183,7 @@ impl ObjectIO for SetDisks { .await?; if let Some(old_dir) = op_old_dir { - self.commit_rename_data_dir(&online_disks, bucket, object, &old_dir.to_string(), write_quorum) + self.commit_rename_data_dir(&cleanup_disks, bucket, object, &old_dir.to_string(), write_quorum) .await?; } @@ -3840,7 +3840,7 @@ impl MultipartOperations for SetDisks { self.cleanup_multipart_path(&parts).await; - let (online_disks, versions, op_old_dir) = Self::rename_data( + let (online_disks, versions, op_old_dir, cleanup_disks) = Self::rename_data( &shuffle_disks, RUSTFS_META_MULTIPART_BUCKET, &upload_id_path, @@ -3852,7 +3852,7 @@ impl MultipartOperations for SetDisks { .await?; if let Some(old_dir) = op_old_dir { - self.commit_rename_data_dir(&online_disks, bucket, object, &old_dir.to_string(), write_quorum) + self.commit_rename_data_dir(&cleanup_disks, bucket, object, &old_dir.to_string(), write_quorum) .await?; } @@ -6267,6 +6267,10 @@ mod tests { let data_dirs = vec![Some(uuid1), Some(uuid2), None]; let result = SetDisks::reduce_common_data_dir(&data_dirs, 2); assert_eq!(result, None); // No UUID meets quorum of 2 + + let data_dirs = vec![Some(uuid1), Some(uuid1), None, None]; + let result = SetDisks::reduce_common_data_dir(&data_dirs, 2); + assert_eq!(result, Some(uuid1)); // Ignore None votes; uuid1 should still meet quorum } #[test] diff --git a/crates/ecstore/src/set_disk/metadata.rs b/crates/ecstore/src/set_disk/metadata.rs index 13295a95b..0eba4dcd0 100644 --- a/crates/ecstore/src/set_disk/metadata.rs +++ b/crates/ecstore/src/set_disk/metadata.rs @@ -37,10 +37,10 @@ impl SetDisks { }) } - pub(super) fn reduce_common_data_dir(data_dirs: &Vec>, write_quorum: usize) -> Option { + pub(super) fn reduce_common_data_dir(data_dirs: &[Option], write_quorum: usize) -> Option { let mut data_dirs_count = HashMap::new(); - for ddir in data_dirs { + for ddir in data_dirs.iter().flatten().copied() { *data_dirs_count.entry(ddir).or_insert(0) += 1; } @@ -49,7 +49,7 @@ impl SetDisks { for (ddir, count) in data_dirs_count { if count > max { max = count; - data_dir = *ddir; + data_dir = Some(ddir); } } diff --git a/crates/ecstore/src/set_disk/write.rs b/crates/ecstore/src/set_disk/write.rs index 38c52c76a..9da1b97fd 100644 --- a/crates/ecstore/src/set_disk/write.rs +++ b/crates/ecstore/src/set_disk/write.rs @@ -38,7 +38,7 @@ impl SetDisks { dst_bucket: &str, dst_object: &str, write_quorum: usize, - ) -> disk::error::Result<(Vec>, Option>, Option)> { + ) -> disk::error::Result<(Vec>, Option>, Option, Vec>)> { let mut futures = Vec::with_capacity(disks.len()); let mut errs = Vec::with_capacity(disks.len()); @@ -179,6 +179,23 @@ impl SetDisks { // TODO: reduceCommonVersions let data_dir = Self::reduce_common_data_dir(&data_dirs, write_quorum); + let online_disks = Self::eval_disks(disks, &errs); + let cleanup_disks = if let Some(data_dir) = data_dir { + disks + .iter() + .zip(errs.iter()) + .zip(data_dirs.iter()) + .map(|((disk, err), old_data_dir)| { + if err.is_none() && *old_data_dir == Some(data_dir) { + disk.clone() + } else { + None + } + }) + .collect() + } else { + vec![None; disks.len()] + }; // // TODO: reduce_common_data_dir // if let Some(old_dir) = rename_ress @@ -193,7 +210,7 @@ impl SetDisks { // self.delete_all(RUSTFS_META_TMP_BUCKET, &tmp_dir).await?; - Ok((Self::eval_disks(disks, &errs), versions, data_dir)) + Ok((online_disks, versions, data_dir, cleanup_disks)) } #[allow(dead_code)]