feat(services/azblob): implement list_with_deleted operation in azblob

core/services/azblob/src/backend.rs

@@ -237,6 +237,15 @@ impl AzblobBuilder {
         self
     }
 
+    /// Set the soft delete feature for this backend.
+    ///
+    /// If enabled, deleted blobs will be retained for the configured retention period
+    /// and can be listed using list_with_deleted.
+    pub fn enable_soft_deletes(mut self, enabled: bool) -> Self {
+        self.config.enable_soft_deletes = enabled;
+        self
+    }
+
     /// from_connection_string will make a builder from connection string
     ///
     /// connection string looks like:
@@ -402,6 +411,7 @@ impl Builder for AzblobBuilder {
 
                             list: true,
                             list_with_recursive: true,
+                            list_with_deleted: self.config.enable_soft_deletes,
 
                             presign: self.config.sas_token.is_some(),
                             presign_stat: self.config.sas_token.is_some(),
@@ -514,6 +524,7 @@ impl Access for AzblobBackend {
             path.to_string(),
             args.recursive(),
             args.limit(),
+            args.deleted(),
         );
 
         Ok((RpList::default(), oio::PageLister::new(l)))

core/services/azblob/src/config.rs

@@ -76,6 +76,10 @@ pub struct AzblobConfig {
 
     /// The maximum batch operations of Azblob service backend.
     pub batch_max_operations: Option<usize>,
+
+    /// Enable soft deletes for this storage account.
+    #[serde(default)]
+    pub enable_soft_deletes: bool,
 }
 
 impl Debug for AzblobConfig {
@@ -84,6 +88,7 @@ impl Debug for AzblobConfig {
             .field("root", &self.root)
             .field("container", &self.container)
             .field("endpoint", &self.endpoint)
+            .field("enable_soft_deletes", &self.enable_soft_deletes)
             .finish_non_exhaustive()
     }
 }

core/services/azblob/src/core.rs

@@ -599,6 +599,7 @@ impl AzblobCore {
         next_marker: &str,
         delimiter: &str,
         limit: Option<usize>,
+        include_deleted: bool,
     ) -> Result<Response<Buffer>> {
         let p = build_abs_path(&self.root, path);
         let mut url = QueryPairsWriter::new(&format!("{}/{}", self.endpoint, self.container))
@@ -618,6 +619,10 @@ impl AzblobCore {
             url = url.push("marker", next_marker);
         }
 
+        if include_deleted {
+            url = url.push("include", "deleted");
+        }
+
         let mut req = Request::get(url.finish())
             .extension(Operation::List)
             .body(Buffer::new())
@@ -685,6 +690,8 @@ pub struct BlobPrefix {
 pub struct Blob {
     pub properties: Properties,
     pub name: String,
+    #[serde(rename = "Deleted")]
+    pub deleted: Option<bool>,
 }
 
 #[derive(Default, Debug, Deserialize)]

core/services/azblob/src/lister.rs

@@ -32,17 +32,25 @@ pub struct AzblobLister {
     path: String,
     delimiter: &'static str,
     limit: Option<usize>,
+    deleted: bool,
 }
 
 impl AzblobLister {
-    pub fn new(core: Arc<AzblobCore>, path: String, recursive: bool, limit: Option<usize>) -> Self {
+    pub fn new(
+        core: Arc<AzblobCore>,
+        path: String,
+        recursive: bool,
+        limit: Option<usize>,
+        deleted: bool,
+    ) -> Self {
         let delimiter = if recursive { "" } else { "/" };
 
         Self {
             core,
             path,
             delimiter,
             limit,
+            deleted,
         }
     }
 }
@@ -51,7 +59,13 @@ impl oio::PageList for AzblobLister {
     async fn next_page(&self, ctx: &mut oio::PageContext) -> Result<()> {
         let resp = self
             .core
-            .azblob_list_blobs(&self.path, &ctx.token, self.delimiter, self.limit)
+            .azblob_list_blobs(
+                &self.path,
+                &ctx.token,
+                self.delimiter,
+                self.limit,
+                self.deleted,
+            )
             .await?;
 
         if resp.status() != http::StatusCode::OK {
@@ -88,7 +102,7 @@ impl oio::PageList for AzblobLister {
                 path = "/".to_string();
             }
 
-            let meta = Metadata::new(EntryMode::from_path(&path))
+            let mut meta = Metadata::new(EntryMode::from_path(&path))
                 // Keep fit with ETag header.
                 .with_etag(format!("\"{}\"", object.properties.etag.as_str()))
                 .with_content_length(object.properties.content_length)
@@ -98,6 +112,11 @@ impl oio::PageList for AzblobLister {
                     object.properties.last_modified.as_str(),
                 )?);
 
+            // Mark as deleted if this is a delete marker
+            if object.deleted.unwrap_or(false) {
+                meta = meta.with_is_deleted(true);
+            }
+
             let de = oio::Entry::with(path, meta);
             ctx.entries.push_back(de);
         }