fix(ecstore): harden multipart part metadata visibility (#3042)

* fix(ecstore): harden multipart part metadata visibility

* fix(ecstore): address PR3042 review follow-ups

crates/ecstore/src/set_disk/read.rs

@@ -144,18 +144,24 @@ impl SetDisks {
         for (part_idx, part_info) in part_meta_paths.iter().enumerate() {
             let mut part_meta_quorum = HashMap::new();
             let mut part_infos = Vec::new();
-            for (j, parts) in object_parts.iter().enumerate() {
+            let mut present_count = 0usize;
+            let mut missing_or_empty_count = 0usize;
+            let mut mismatched_response_count = 0usize;
+            for parts in object_parts.iter() {
                 if parts.len() != part_meta_paths.len() {
+                    mismatched_response_count += 1;
                     *part_meta_quorum.entry(part_info.clone()).or_insert(0) += 1;
                     continue;
                 }
 
                 if !parts[part_idx].etag.is_empty() {
+                    present_count += 1;
                     *part_meta_quorum.entry(parts[part_idx].etag.clone()).or_insert(0) += 1;
                     part_infos.push(parts[part_idx].clone());
                     continue;
                 }
 
+                missing_or_empty_count += 1;
                 *part_meta_quorum.entry(part_info.clone()).or_insert(0) += 1;
             }
 
@@ -194,6 +200,22 @@ impl SetDisks {
             {
                 ret[part_idx] = found.clone();
             } else {
+                if issue3031_diag_enabled() {
+                    warn!(
+                        target: "rustfs_ecstore::set_disk",
+                        bucket = %bucket,
+                        part_meta_path = %part_info,
+                        part_id = part_numbers[part_idx],
+                        read_quorum = read_quorum,
+                        max_quorum = max_quorum,
+                        disk_response_count = object_parts.len(),
+                        present_count = present_count,
+                        missing_or_empty_count = missing_or_empty_count,
+                        mismatched_response_count = mismatched_response_count,
+                        max_vote_is_missing_marker = max_etag.map(|etag| etag == part_info).unwrap_or(false),
+                        "issue3031_read_parts_part_quorum"
+                    );
+                }
                 ret[part_idx] = ObjectPartInfo {
                     number: part_numbers[part_idx],
                     error: Some(format!("part.{} not found", part_numbers[part_idx])),

crates/ecstore/src/set_disk/write.rs

@@ -202,6 +202,9 @@ impl SetDisks {
 
     #[tracing::instrument(skip(self))]
     pub(super) async fn cleanup_multipart_path(&self, paths: &[String]) {
+        if paths.is_empty() {
+            return;
+        }
         let disks = self.get_disks_internal().await;
 
         let mut errs = Vec::with_capacity(disks.len());
@@ -290,6 +293,27 @@ impl SetDisks {
             }
         }
 
+        if issue3031_diag_enabled() {
+            let success_count = errs.iter().filter(|err| err.is_none()).count();
+            let error_count = errs.len().saturating_sub(success_count);
+            let disk_not_found_count = errs.iter().filter(|err| matches!(err, Some(DiskError::DiskNotFound))).count();
+            let file_not_found_count = errs.iter().filter(|err| matches!(err, Some(DiskError::FileNotFound))).count();
+            warn!(
+                target: "rustfs_ecstore::set_disk",
+                src_bucket = %src_bucket,
+                src_object = %src_object,
+                dst_bucket = %dst_bucket,
+                dst_object = %dst_object,
+                write_quorum = write_quorum,
+                disk_count = errs.len(),
+                success_count = success_count,
+                error_count = error_count,
+                disk_not_found_count = disk_not_found_count,
+                file_not_found_count = file_not_found_count,
+                "issue3031_rename_part_context"
+            );
+        }
+
         if let Some(err) = reduce_write_quorum_errs(&errs, OBJECT_OP_IGNORED_ERRS, write_quorum) {
             warn!("rename_part errs {:?}", &errs);
             self.cleanup_multipart_path(&[dst_object.to_string(), format!("{dst_object}.meta")])