Merge branch 'apache:main' into wip_opfs_3

Author: jccampagne

core/Cargo.lock

@@ -51,7 +51,6 @@ logforth = { version = "0.29.1", features = ["starter-log"] }
 mea = "0.6"
 quick-xml = { version = "0.38", default-features = false }
 rand = "0.8"
-reqsign = { version = "0.16.5", default-features = false }
 serde = { version = "1", default-features = false }
 serde_json = "1"
 sha2 = "0.10"

core/Cargo.toml

@@ -103,9 +103,31 @@ where
                         );
                         continue;
                     }
+                    Err(e) if e.kind() == ErrorKind::NotFound => {
+                        // Skip directories that are deleted while listing.
+                        log::warn!(
+                            "FlatLister skipping directory due to not found during listing: {}",
+                            de.path()
+                        );
+                        continue;
+                    }
                     Err(e) => return Err(e),
                 };
-                if let Some(v) = l.next().await? {
+                let first = loop {
+                    match l.next().await {
+                        Ok(v) => break v,
+                        Err(e) if e.kind() == ErrorKind::NotFound => {
+                            // Skip entries that are deleted during listing.
+                            log::warn!(
+                                "FlatLister skipping entry due to not found during listing: {}",
+                                de.path()
+                            );
+                            continue;
+                        }
+                        Err(e) => return Err(e),
+                    }
+                };
+                if let Some(v) = first {
                     self.active_lister.push((Some(de.clone()), l));
 
                     if v.mode().is_dir() {
@@ -120,21 +142,40 @@ where
                 }
             }
 
+            if matches!(self.active_lister.last(), Some((None, _))) {
+                let _ = self.active_lister.pop();
+                continue;
+            }
+
             let (de, lister) = match self.active_lister.last_mut() {
                 Some((de, lister)) => (de, lister),
                 None => return Ok(None),
             };
 
-            match lister.next().await? {
-                Some(v) if v.mode().is_dir() => {
+            match lister.next().await {
+                Err(e) if e.kind() == ErrorKind::NotFound => {
+                    let path = de.as_ref().map(|entry| entry.path()).unwrap_or("<unknown>");
+                    log::warn!(
+                        "FlatLister skipping entry due to not found during recursive listing: {}",
+                        path
+                    );
+                    continue;
+                }
+                Err(e) => return Err(e),
+                Ok(Some(v)) if v.mode().is_dir() => {
                     // should not loop itself again
-                    if v.path() != de.as_ref().expect("de should not be none here").path() {
+                    if v.path()
+                        != de
+                            .as_ref()
+                            .expect("de must be present before listing")
+                            .path()
+                    {
                         self.next_dir = Some(v);
                         continue;
                     }
                 }
-                Some(v) => return Ok(Some(v)),
-                None => match de.take() {
+                Ok(Some(v)) => return Ok(Some(v)),
+                Ok(None) => match de.take() {
                     Some(de) => {
                         return Ok(Some(de));
                     }

core/core/src/raw/oio/list/flat_list.rs

@@ -31,7 +31,7 @@ version = { workspace = true }
 all-features = true
 
 [dependencies]
-fastmetrics = "0.6"
+fastmetrics = "0.7"
 opendal-core = { path = "../../core", version = "0.55.0", default-features = false }
 opendal-layer-observe-metrics-common = { path = "../observe-metrics-common", version = "0.55.0", default-features = false }