fix(core): enforce list file content length in complete layer (#7201)

* fix(core): enforce list file content length in complete layer

* fix(services/etcd): fill content length from list response

* fix(etcd): keep list keys-only and skip stale entries

* refactor(core): short-circuit list content length completion

* refactor(core): inline list completion fast-path check

Author: Xuanwo

core/core/src/layers/complete.rs

@@ -57,7 +57,7 @@ impl<A: Access> LayeredAccess for CompleteAccessor<A> {
     type Inner = A;
     type Reader = CompleteReader<A::Reader>;
     type Writer = CompleteWriter<A::Writer>;
-    type Lister = A::Lister;
+    type Lister = CompleteLister<A>;
     type Deleter = A::Deleter;
 
     fn inner(&self) -> &Self::Inner {
@@ -95,14 +95,76 @@ impl<A: Access> LayeredAccess for CompleteAccessor<A> {
     }
 
     async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
-        self.inner.list(path, args).await
+        let (rp, lister) = self.inner.list(path, args).await?;
+        let lister = CompleteLister::new(self.inner.clone(), self.info.clone(), lister);
+        Ok((rp, lister))
     }
 
     async fn presign(&self, path: &str, args: OpPresign) -> Result<RpPresign> {
         self.inner.presign(path, args).await
     }
 }
 
+pub struct CompleteLister<A: Access> {
+    inner: A::Lister,
+    acc: Arc<A>,
+    info: Arc<AccessorInfo>,
+}
+
+impl<A: Access> CompleteLister<A> {
+    fn new(acc: Arc<A>, info: Arc<AccessorInfo>, inner: A::Lister) -> Self {
+        Self { inner, acc, info }
+    }
+
+    async fn ensure_file_content_length(&self, entry: &mut oio::Entry) -> Result<()> {
+        let path = entry.path().to_string();
+        let version = entry.metadata().version().map(str::to_owned);
+        let mut op = OpStat::new();
+        if let Some(version) = version.as_deref() {
+            op = op.with_version(version);
+        }
+
+        let stat_metadata = self.acc.stat(&path, op).await?.into_metadata();
+        if !stat_metadata.has_content_length() {
+            return Err(Error::new(
+                ErrorKind::Unexpected,
+                "content length is required for list file entries",
+            )
+            .with_operation("CompleteLister::ensure_file_content_length")
+            .with_context("service", self.info.scheme().to_string())
+            .with_context("path", path));
+        }
+
+        entry
+            .metadata_mut()
+            .set_content_length(stat_metadata.content_length());
+        Ok(())
+    }
+}
+
+impl<A: Access> oio::List for CompleteLister<A> {
+    async fn next(&mut self) -> Result<Option<oio::Entry>> {
+        loop {
+            let Some(mut entry) = self.inner.next().await? else {
+                return Ok(None);
+            };
+
+            if !entry.mode().is_file()
+                || entry.metadata().is_deleted()
+                || entry.metadata().has_content_length()
+            {
+                return Ok(Some(entry));
+            }
+
+            match self.ensure_file_content_length(&mut entry).await {
+                Ok(()) => return Ok(Some(entry)),
+                Err(err) if err.kind() == ErrorKind::NotFound => continue,
+                Err(err) => return Err(err),
+            }
+        }
+    }
+}
+
 pub struct CompleteReader<R> {
     inner: R,
     size: Option<u64>,

core/core/src/raw/oio/entry.rs

@@ -86,4 +86,14 @@ impl Entry {
     pub(crate) fn into_entry(self) -> crate::Entry {
         crate::Entry::new(self.path, self.meta)
     }
+
+    /// Get metadata of entry.
+    pub(crate) fn metadata(&self) -> &Metadata {
+        &self.meta
+    }
+
+    /// Get mutable metadata of entry.
+    pub(crate) fn metadata_mut(&mut self) -> &mut Metadata {
+        &mut self.meta
+    }
 }

core/core/src/types/metadata.rs

@@ -222,6 +222,11 @@ impl Metadata {
         self.content_length.unwrap_or_default()
     }
 
+    /// Returns `true` if this metadata contains an explicit content length.
+    pub(crate) fn has_content_length(&self) -> bool {
+        self.content_length.is_some()
+    }
+
     /// Set content length of this entry.
     pub fn set_content_length(&mut self, v: u64) -> &mut Self {
         self.content_length = Some(v);

core/tests/behavior/async_list.rs

@@ -79,10 +79,9 @@ pub async fn test_list_dir(op: Operator) -> Result<()> {
     let mut obs = op.lister(&format!("{parent}/")).await?;
     let mut found = false;
     while let Some(de) = obs.try_next().await? {
-        let meta = op.stat(de.path()).await?;
         if de.path() == path {
+            let meta = de.metadata();
             assert_eq!(meta.mode(), EntryMode::FILE);
-
             assert_eq!(meta.content_length(), size as u64);
 
             found = true