fix(ecstore): surface prefix listing storage errors (#2940)

crates/ecstore/src/cache_value/metacache_set.rs

@@ -106,6 +106,7 @@ pub async fn list_path_raw(rx: CancellationToken, opts: ListPathRawOptions) -> d
     let mut jobs: Vec<tokio::task::JoinHandle<std::result::Result<(), DiskError>>> = Vec::new();
     let mut readers = Vec::with_capacity(opts.disks.len());
     let fds = opts.fallback_disks.iter().flatten().cloned().collect::<Vec<_>>();
+    let max_disk_failures = opts.disks.len().saturating_sub(opts.min_disks);
 
     let cancel_rx = CancellationToken::new();
 
@@ -144,15 +145,18 @@ pub async fn list_path_raw(rx: CancellationToken, opts: ListPathRawOptions) -> d
             };
 
             let mut need_fallback = false;
+            let mut last_err = None;
             if let Some(disk) = opdisk {
                 match disk.walk_dir(wakl_opts, &mut wr).await {
                     Ok(_res) => {}
                     Err(err) => {
                         info!("walk dir err {:?}", &err);
+                        last_err = Some(err);
                         need_fallback = true;
                     }
                 }
             } else {
+                last_err = Some(DiskError::DiskNotFound);
                 need_fallback = true;
             }
 
@@ -195,14 +199,20 @@ pub async fn list_path_raw(rx: CancellationToken, opts: ListPathRawOptions) -> d
                 {
                     Ok(_r) => {
                         need_fallback = false;
+                        last_err = None;
                     }
                     Err(err) => {
                         error!("walk dir2 err {:?}", &err);
+                        last_err = Some(err);
                         break;
                     }
                 }
             }
 
+            if need_fallback {
+                return Err(last_err.unwrap_or(DiskError::DiskNotFound));
+            }
+
             // warn!("list_path_raw: while need_fallback done");
             Ok(())
         }));
@@ -436,12 +446,25 @@ pub async fn list_path_raw(rx: CancellationToken, opts: ListPathRawOptions) -> d
     }
 
     let results = join_all(jobs).await;
+    let mut job_errs = Vec::new();
     for result in results {
-        if let Err(err) = result {
-            error!("list_path_raw err {:?}", err);
+        match result {
+            Ok(Ok(())) => {}
+            Ok(Err(err)) => {
+                error!("list_path_raw err {:?}", err);
+                job_errs.push(err);
+            }
+            Err(err) => {
+                error!("list_path_raw join err {:?}", err);
+                job_errs.push(err.into());
+            }
         }
     }
 
+    if job_errs.len() > max_disk_failures {
+        return Err(job_errs.remove(0));
+    }
+
     // warn!("list_path_raw: done");
     Ok(())
 }
@@ -449,6 +472,8 @@ pub async fn list_path_raw(rx: CancellationToken, opts: ListPathRawOptions) -> d
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::disk::endpoint::Endpoint;
+    use crate::disk::{DiskOption, new_disk};
     use rustfs_filemeta::MetacacheWriter;
 
     #[tokio::test]
@@ -510,4 +535,53 @@ mod tests {
             other => panic!("expected ready entry, got {other:?}"),
         }
     }
+
+    #[cfg(unix)]
+    #[tokio::test]
+    async fn list_path_raw_propagates_parent_scan_access_denied() {
+        use std::fs::Permissions;
+        use std::os::unix::fs::PermissionsExt;
+        use tempfile::tempdir;
+        use tokio::fs;
+
+        let dir = tempdir().expect("tempdir should be created");
+        let bucket = "test-bucket";
+        let parent = dir.path().join(bucket).join("shiplog");
+
+        fs::create_dir_all(parent.join("nano/a.txt"))
+            .await
+            .expect("test object directory should be created");
+        fs::write(parent.join("nano/a.txt/xl.meta"), b"meta")
+            .await
+            .expect("test metadata should be written");
+
+        std::fs::set_permissions(&parent, Permissions::from_mode(0o111)).expect("parent permissions should be changed");
+        if fs::read_dir(&parent).await.is_ok() {
+            std::fs::set_permissions(&parent, Permissions::from_mode(0o755)).expect("parent permissions should be restored");
+            return;
+        }
+
+        let endpoint = Endpoint::try_from(dir.path().to_str().expect("temp path should be valid utf-8"))
+            .expect("endpoint should be created");
+        let disk = new_disk(&endpoint, &DiskOption::default())
+            .await
+            .expect("local disk should be created");
+
+        let result = list_path_raw(
+            CancellationToken::new(),
+            ListPathRawOptions {
+                disks: vec![Some(disk)],
+                bucket: bucket.to_string(),
+                path: "shiplog/".to_string(),
+                min_disks: 1,
+                ..Default::default()
+            },
+        )
+        .await;
+
+        std::fs::set_permissions(&parent, Permissions::from_mode(0o755)).expect("parent permissions should be restored");
+
+        let err = result.expect_err("parent directory access failure must not be treated as an empty listing");
+        assert_eq!(err, DiskError::FileAccessDenied);
+    }
 }

crates/ecstore/src/disk/local.rs

@@ -1283,6 +1283,7 @@ impl LocalDisk {
             Err(e) => {
                 if e != DiskError::VolumeNotFound && e != Error::FileNotFound {
                     error!("scan list_dir {}, err {:?}", &current, &e);
+                    return Err(e);
                 }
 
                 if opts.report_notfound && e == Error::FileNotFound && current == opts.base_dir {

crates/ecstore/src/store_list_objects.rs

@@ -315,7 +315,7 @@ impl ECStore {
                 ..Default::default()
             });
 
-        if let Some(err) = list_result.err.clone()
+        if let Some(err) = list_result.err.take()
             && err != rustfs_filemeta::Error::Unexpected
         {
             return Err(to_object_err(err.into(), vec![bucket, prefix]));
@@ -427,7 +427,7 @@ impl ECStore {
                 ..Default::default()
             });
 
-        if let Some(err) = list_result.err.clone()
+        if let Some(err) = list_result.err.take()
             && err != rustfs_filemeta::Error::Unexpected
         {
             return Err(to_object_err(err.into(), vec![bucket, prefix]));
@@ -612,6 +612,11 @@ impl ECStore {
         // wait spawns exit
         join_all(vec![job1, job2]).await;
 
+        if let Ok(err) = err_rx.try_recv() {
+            error!("list_path err_rx.try_recv() ok {:?}", &err);
+            result.err = Some(err.as_ref().clone().into());
+        }
+
         if result.err.is_some() {
             return Ok(result);
         }
@@ -1340,7 +1345,7 @@ impl SetDisks {
             },
         )
         .await
-        .map_err(Error::other)
+        .map_err(Error::from)
     }
 }