fix(ecstore): propagate walk listing errors (#2949)

Co-authored-by: houseme <housemecn@gmail.com>

crates/ecstore/src/store_list_objects.rs

@@ -67,6 +67,26 @@ fn ensure_non_empty_listing_disks(bucket: &str, path: &str, disks: &[DiskStore])
     Ok(())
 }
 
+fn walk_result_from_set_errors(errs: &[Option<Error>]) -> Result<()> {
+    if is_all_not_found(errs) {
+        if is_all_volume_not_found(errs) {
+            return Err(StorageError::VolumeNotFound);
+        }
+
+        return Ok(());
+    }
+
+    for err in errs.iter().flatten() {
+        if err == &Error::Unexpected || err.is_not_found() {
+            continue;
+        }
+
+        return Err(err.clone());
+    }
+
+    Ok(())
+}
+
 pub fn max_keys_plus_one(max_keys: i32, add_one: bool) -> i32 {
     let mut max_keys = max_keys;
     if !(0..=MAX_OBJECT_LIST).contains(&max_keys) {
@@ -967,19 +987,26 @@ impl ECStore {
         tokio::spawn(async move { merge_entry_channels(rx, inputs, merge_tx, 1).await });
 
         let walk_results = join_all(futures).await;
-        for (idx, walk_result) in walk_results.into_iter().enumerate() {
-            if let Err(err) = walk_result {
-                error!(
-                    bucket = %bucket,
-                    prefix = %prefix,
-                    set_task_index = idx,
-                    error = ?err,
-                    "walk_internal list_path_raw task failed"
-                );
+        let mut errs = Vec::new();
+        for walk_result in walk_results {
+            match walk_result {
+                Ok(()) => errs.push(None),
+                Err(err) => errs.push(Some(err.into())),
             }
         }
 
-        Ok(())
+        let result = walk_result_from_set_errors(&errs);
+        if let Err(err) = &result {
+            error!(
+                bucket = %bucket,
+                prefix = %prefix,
+                error = ?err,
+                set_errors = ?errs,
+                "walk_internal list_path_raw tasks failed"
+            );
+        }
+
+        result
     }
 }
 
@@ -1416,7 +1443,8 @@ fn calc_common_counter(infos: &[DiskInfo], read_quorum: usize) -> u64 {
 
 #[cfg(test)]
 mod test {
-    use super::{ListPathOptions, MAX_OBJECT_LIST, max_keys_plus_one};
+    use super::{ListPathOptions, MAX_OBJECT_LIST, max_keys_plus_one, walk_result_from_set_errors};
+    use crate::error::StorageError;
     use uuid::Uuid;
 
     #[test]
@@ -1527,6 +1555,34 @@ mod test {
         assert!(!parsed.create);
     }
 
+    #[test]
+    fn walk_result_from_set_errors_returns_non_eof_error() {
+        let err = walk_result_from_set_errors(&[Some(StorageError::Unexpected), Some(StorageError::FileAccessDenied)])
+            .expect_err("walk should fail when any set reports a real listing error");
+
+        assert_eq!(err, StorageError::FileAccessDenied);
+    }
+
+    #[test]
+    fn walk_result_from_set_errors_prefers_real_error_over_not_found() {
+        let err = walk_result_from_set_errors(&[
+            Some(StorageError::VolumeNotFound),
+            Some(StorageError::DiskNotFound),
+            Some(StorageError::FileAccessDenied),
+        ])
+        .expect_err("walk should report the real listing error");
+
+        assert_eq!(err, StorageError::FileAccessDenied);
+    }
+
+    #[test]
+    fn walk_result_from_set_errors_preserves_volume_not_found() {
+        let err = walk_result_from_set_errors(&[Some(StorageError::VolumeNotFound), Some(StorageError::VolumeNotFound)])
+            .expect_err("all volume-not-found set errors should remain visible");
+
+        assert_eq!(err, StorageError::VolumeNotFound);
+    }
+
     // use std::sync::Arc;
 
     // use crate::cache_value::metacache_set::list_path_raw;