fix: clean old data dirs on object overwrite (#3244)

Co-authored-by: cxymds <Cxymds@qq.com>

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]

crates/ecstore/src/set_disk/metadata.rs

@@ -37,10 +37,10 @@ impl SetDisks {
             })
     }
 
-    pub(super) fn reduce_common_data_dir(data_dirs: &Vec<Option<Uuid>>, write_quorum: usize) -> Option<Uuid> {
+    pub(super) fn reduce_common_data_dir(data_dirs: &[Option<Uuid>], write_quorum: usize) -> Option<Uuid> {
         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);
             }
         }
 

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<DiskStore>>, Option<Vec<u8>>, Option<Uuid>)> {
+    ) -> disk::error::Result<(Vec<Option<DiskStore>>, Option<Vec<u8>>, Option<Uuid>, Vec<Option<DiskStore>>)> {
         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)]