fix: handle empty multipart list-parts (#2765)

Co-authored-by: loverustfs <hello@rustfs.com>

crates/ecstore/src/set_disk/multipart.rs

@@ -17,6 +17,16 @@ use std::future::Future;
 use std::time::Duration;
 use tokio::task::JoinSet;
 
+fn empty_upload_fallback_possible(successful_responses: usize, errs: &[Option<DiskError>]) -> bool {
+    successful_responses == 0
+        && errs.iter().any(|err| matches!(err, Some(DiskError::FileNotFound)))
+        && errs.iter().all(|err| match err {
+            Some(DiskError::FileNotFound) => true,
+            Some(err) => OBJECT_OP_IGNORED_ERRS.contains(err),
+            None => false,
+        })
+}
+
 async fn collect_list_parts_results<F>(
     tasks: Vec<F>,
     read_quorum: usize,
@@ -49,11 +59,19 @@ where
             Err(_) => {}
         }
 
-        if successful_responses + pending < read_quorum {
+        if successful_responses + pending < read_quorum && !empty_upload_fallback_possible(successful_responses, &errs) {
             return Err(DiskError::ErasureReadQuorum);
         }
     }
 
+    if successful_responses < read_quorum {
+        if empty_upload_fallback_possible(successful_responses, &errs) {
+            return Err(DiskError::FileNotFound);
+        }
+
+        return Err(DiskError::ErasureReadQuorum);
+    }
+
     Ok((errs, object_parts))
 }
 
@@ -232,6 +250,50 @@ mod tests {
         assert_eq!(object_parts.iter().filter(|parts| !parts.is_empty()).count(), 2);
     }
 
+    #[tokio::test]
+    async fn collect_list_parts_results_returns_file_not_found_for_empty_upload_dirs() {
+        let tasks: Vec<_> = vec![
+            (5_u64, Err(DiskError::FileNotFound)),
+            (10, Err(DiskError::DiskNotFound)),
+            (12, Err(DiskError::DiskNotFound)),
+        ]
+        .into_iter()
+        .map(|(delay_ms, outcome)| async move {
+            tokio::time::sleep(Duration::from_millis(delay_ms)).await;
+            outcome
+        })
+        .collect();
+
+        let err = collect_list_parts_results(tasks, 2)
+            .await
+            .expect_err("missing multipart directories should be treated as empty uploads");
+
+        assert_eq!(err, DiskError::FileNotFound);
+    }
+
+    #[tokio::test]
+    async fn collect_list_parts_results_fails_early_when_file_not_found_fallback_is_impossible() {
+        let started = std::time::Instant::now();
+        let tasks: Vec<_> = vec![
+            (5_u64, Err(DiskError::FileNotFound)),
+            (10, Err(DiskError::FileCorrupt)),
+            (250, Err(DiskError::DiskNotFound)),
+        ]
+        .into_iter()
+        .map(|(delay_ms, outcome)| async move {
+            tokio::time::sleep(Duration::from_millis(delay_ms)).await;
+            outcome
+        })
+        .collect();
+
+        let err = collect_list_parts_results(tasks, 2)
+            .await
+            .expect_err("non-ignored errors should preserve early quorum failure");
+
+        assert_eq!(err, DiskError::ErasureReadQuorum);
+        assert!(started.elapsed() < Duration::from_millis(120));
+    }
+
     #[test]
     fn reduce_quorum_part_numbers_only_keeps_parts_present_on_quorum_of_drives() {
         let object_parts = vec![