fix(services/s3): List with deleted should contain latest

core/src/services/s3/lister.rs

@@ -222,28 +222,34 @@ impl oio::PageList for S3ObjectVersionsLister {
             ctx.entries.push_back(de);
         }
 
-        if self.args.versions() {
-            for version_object in output.version {
-                let mut path = build_rel_path(&self.core.root, &version_object.key);
-                if path.is_empty() {
-                    path = "/".to_owned();
-                }
+        for version_object in output.version {
+            // `list` must be additive, so we need to include the latest version object
+            // even if `versions` is not enabled.
+            //
+            // Here we skip all non-latest version objects if `versions` is not enabled.
+            if !(self.args.versions() || version_object.is_latest) {
+                continue;
+            }
 
-                let mut meta = Metadata::new(EntryMode::from_path(&path));
-                meta.set_version(&version_object.version_id);
-                meta.set_is_current(version_object.is_latest);
-                meta.set_content_length(version_object.size);
-                meta.set_last_modified(parse_datetime_from_rfc3339(
-                    version_object.last_modified.as_str(),
-                )?);
-                if let Some(etag) = version_object.etag {
-                    meta.set_etag(&etag);
-                    meta.set_content_md5(etag.trim_matches('"'));
-                }
+            let mut path = build_rel_path(&self.core.root, &version_object.key);
+            if path.is_empty() {
+                path = "/".to_owned();
+            }
 
-                let entry = oio::Entry::new(&path, meta);
-                ctx.entries.push_back(entry);
+            let mut meta = Metadata::new(EntryMode::from_path(&path));
+            meta.set_version(&version_object.version_id);
+            meta.set_is_current(version_object.is_latest);
+            meta.set_content_length(version_object.size);
+            meta.set_last_modified(parse_datetime_from_rfc3339(
+                version_object.last_modified.as_str(),
+            )?);
+            if let Some(etag) = version_object.etag {
+                meta.set_etag(&etag);
+                meta.set_content_md5(etag.trim_matches('"'));
             }
+
+            let entry = oio::Entry::new(&path, meta);
+            ctx.entries.push_back(entry);
         }
 
         if self.args.deleted() {

core/tests/behavior/async_list.rs

@@ -610,10 +610,19 @@ pub async fn test_list_files_with_deleted(op: Operator) -> Result<()> {
     let file_name = TEST_FIXTURE.new_file_path();
     let file_path = format!("{}{}", parent, file_name);
     op.write(file_path.as_str(), "1").await?;
+
+    // List with deleted should include self too.
+    let ds = op.list_with(&file_path).deleted(true).await?;
+    assert_eq!(
+        ds.len(),
+        1,
+        "list with deleted should contain current active file version"
+    );
+
     op.write(file_path.as_str(), "2").await?;
     op.delete(file_path.as_str()).await?;
 
-    // This file has been deleted
+    // This file has been deleted, list with deleted should contain its versions and delete marker.
     let mut ds = op.list_with(&file_path).deleted(true).await?;
     ds.retain(|de| de.path() == file_path && de.metadata().is_deleted());