refactor(protocols): replace tar with astral-tokio-tar for async processing (#2099)

Co-authored-by: Copilot <198982749+Copilot@users.noreply.github.com>
Co-authored-by: houseme <4829346+houseme@users.noreply.github.com>

crates/protocols/src/swift/bulk.rs

@@ -76,10 +76,10 @@
 
 use super::{SwiftError, SwiftResult, container, object};
 use axum::http::{Response, StatusCode};
+use futures::StreamExt;
 use rustfs_credentials::Credentials;
 use s3s::Body;
 use serde::{Deserialize, Serialize};
-use std::io::Read;
 use tracing::{debug, error};
 
 /// Result of a single delete operation
@@ -340,8 +340,8 @@ pub async fn handle_bulk_extract(
         return Err(SwiftError::NotFound(format!("Container not found: {}", container)));
     }
 
-    // Parse archive and collect entries (without holding the archive)
-    let entries = extract_tar_entries(format, body)?;
+    // Parse archive and collect all entries into memory (entire archive and file contents are buffered)
+    let entries = extract_tar_entries(format, body).await?;
 
     // Now upload each entry (async operations)
     for (path_str, contents) in entries {
@@ -352,7 +352,7 @@ pub async fn handle_bulk_extract(
             &path_str,
             credentials,
             std::io::Cursor::new(contents),
-            &axum::http::HeaderMap::new(),
+            &http::HeaderMap::new(),
         )
         .await
         {
@@ -395,30 +395,30 @@ pub async fn handle_bulk_extract(
         .map_err(|e| SwiftError::InternalServerError(format!("Failed to build response: {}", e)))
 }
 
-/// Extract tar entries synchronously to avoid Send issues
-fn extract_tar_entries(format: ArchiveFormat, body: Vec<u8>) -> SwiftResult<Vec<(String, Vec<u8>)>> {
+/// Extract tar entries using async I/O and return them as in-memory buffers
+async fn extract_tar_entries(format: ArchiveFormat, body: Vec<u8>) -> SwiftResult<Vec<(String, Vec<u8>)>> {
     // Create appropriate reader based on format
-    let reader: Box<dyn Read> = match format {
+    let reader: Box<dyn tokio::io::AsyncRead + Unpin + Send> = match format {
         ArchiveFormat::Tar => Box::new(std::io::Cursor::new(body)),
         ArchiveFormat::TarGz => {
             let cursor = std::io::Cursor::new(body);
-            Box::new(flate2::read::GzDecoder::new(cursor))
+            Box::new(async_compression::tokio::bufread::GzipDecoder::new(tokio::io::BufReader::new(cursor)))
         }
         ArchiveFormat::TarBz2 => {
             let cursor = std::io::Cursor::new(body);
-            Box::new(bzip2::read::BzDecoder::new(cursor))
+            Box::new(async_compression::tokio::bufread::BzDecoder::new(tokio::io::BufReader::new(cursor)))
         }
     };
 
     // Parse tar archive
-    let mut archive = tar::Archive::new(reader);
+    let mut archive = tokio_tar::Archive::new(reader);
     let mut entries = Vec::new();
+    let mut entries_iter = archive
+        .entries()
+        .map_err(|e| SwiftError::BadRequest(format!("Failed to read tar archive: {}", e)))?;
 
     // Extract each entry
-    for entry in archive
-        .entries()
-        .map_err(|e| SwiftError::BadRequest(format!("Failed to read tar archive: {}", e)))?
-    {
+    while let Some(entry) = entries_iter.next().await {
         let mut entry = entry.map_err(|e| SwiftError::BadRequest(format!("Failed to read tar entry: {}", e)))?;
 
         // Get entry path
@@ -436,7 +436,7 @@ fn extract_tar_entries(format: ArchiveFormat, body: Vec<u8>) -> SwiftResult<Vec<
 
         // Read file contents
         let mut contents = Vec::new();
-        if let Err(e) = entry.read_to_end(&mut contents) {
+        if let Err(e) = tokio::io::AsyncReadExt::read_to_end(&mut entry, &mut contents).await {
             error!("Failed to read tar entry {}: {}", path_str, e);
             continue;
         }
@@ -552,4 +552,117 @@ mod tests {
 
         assert_eq!(paths.len(), 3);
     }
+
+    /// Tests for the `extract_tar_entries` async function.
+    ///
+    /// Conditionally compiled with the `swift` feature, which gates the
+    /// `tokio_tar` (astral-tokio-tar) and `async_compression` dependencies.
+    #[cfg(feature = "swift")]
+    mod tar_extraction {
+        use super::*;
+        use tokio::io::AsyncWriteExt;
+
+        /// Builds an uncompressed tar archive in memory.
+        ///
+        /// * `files` – `(path, content)` pairs added as regular files.
+        /// * `dirs`  – paths added as directory entries.
+        async fn make_tar(files: &[(&str, &[u8])], dirs: &[&str]) -> Vec<u8> {
+            let buf = std::io::Cursor::new(Vec::new());
+            let mut builder = tokio_tar::Builder::new(buf);
+
+            for &dir in dirs {
+                let mut header = tokio_tar::Header::new_gnu();
+                header.set_entry_type(tokio_tar::EntryType::Directory);
+                header.set_size(0);
+                header.set_mode(0o755);
+                header.set_cksum();
+                builder
+                    .append_data(&mut header, dir, std::io::Cursor::new(&[] as &[u8]))
+                    .await
+                    .unwrap();
+            }
+
+            for &(name, data) in files {
+                let mut header = tokio_tar::Header::new_gnu();
+                header.set_size(data.len() as u64);
+                header.set_mode(0o644);
+                header.set_cksum();
+                builder
+                    .append_data(&mut header, name, std::io::Cursor::new(data))
+                    .await
+                    .unwrap();
+            }
+
+            builder.into_inner().await.unwrap().into_inner()
+        }
+
+        #[tokio::test]
+        async fn test_extract_plain_tar_paths_and_contents() {
+            let tar_bytes = make_tar(&[("file1.txt", b"hello"), ("dir/file2.txt", b"world")], &[]).await;
+            let entries = extract_tar_entries(ArchiveFormat::Tar, tar_bytes).await.unwrap();
+            assert_eq!(entries.len(), 2);
+            assert_eq!(entries[0].0, "file1.txt");
+            assert_eq!(entries[0].1, b"hello");
+            assert_eq!(entries[1].0, "dir/file2.txt");
+            assert_eq!(entries[1].1, b"world");
+        }
+
+        #[tokio::test]
+        async fn test_extract_tar_skips_directories() {
+            let tar_bytes = make_tar(&[("file.txt", b"content")], &["subdir/", "another/"]).await;
+            let entries = extract_tar_entries(ArchiveFormat::Tar, tar_bytes).await.unwrap();
+            // Directory entries must be filtered out; only the regular file is returned.
+            assert_eq!(entries.len(), 1);
+            assert_eq!(entries[0].0, "file.txt");
+        }
+
+        #[tokio::test]
+        async fn test_extract_tar_gz() {
+            let tar_bytes = make_tar(&[("file.txt", b"compressed content")], &[]).await;
+
+            // Compress with gzip.
+            let cursor = std::io::Cursor::new(Vec::new());
+            let mut encoder = async_compression::tokio::write::GzipEncoder::new(cursor);
+            encoder.write_all(&tar_bytes).await.unwrap();
+            encoder.shutdown().await.unwrap();
+            let gz_bytes = encoder.into_inner().into_inner();
+
+            let entries = extract_tar_entries(ArchiveFormat::TarGz, gz_bytes).await.unwrap();
+            assert_eq!(entries.len(), 1);
+            assert_eq!(entries[0].0, "file.txt");
+            assert_eq!(entries[0].1, b"compressed content");
+        }
+
+        #[tokio::test]
+        async fn test_extract_tar_bz2() {
+            let tar_bytes = make_tar(&[("file.txt", b"bzip2 content")], &[]).await;
+
+            // Compress with bzip2.
+            let cursor = std::io::Cursor::new(Vec::new());
+            let mut encoder = async_compression::tokio::write::BzEncoder::new(cursor);
+            encoder.write_all(&tar_bytes).await.unwrap();
+            encoder.shutdown().await.unwrap();
+            let bz2_bytes = encoder.into_inner().into_inner();
+
+            let entries = extract_tar_entries(ArchiveFormat::TarBz2, bz2_bytes).await.unwrap();
+            assert_eq!(entries.len(), 1);
+            assert_eq!(entries[0].0, "file.txt");
+            assert_eq!(entries[0].1, b"bzip2 content");
+        }
+
+        #[tokio::test]
+        async fn test_extract_invalid_tar_returns_error() {
+            // Fewer than 512 bytes → parser cannot read a complete tar header block.
+            let bad_bytes = b"this is not a valid tar archive".to_vec();
+            let result = extract_tar_entries(ArchiveFormat::Tar, bad_bytes).await;
+            assert!(result.is_err(), "expected error for invalid tar data");
+        }
+
+        #[tokio::test]
+        async fn test_extract_empty_tar() {
+            let tar_bytes = make_tar(&[], &[]).await;
+            let entries = extract_tar_entries(ArchiveFormat::Tar, tar_bytes).await.unwrap();
+            assert!(entries.is_empty());
+        }
+    }
 }

crates/protocols/src/swift/encryption.rs

@@ -260,7 +260,7 @@ impl EncryptionMetadata {
 }
 
 /// Check if object should be encrypted based on configuration and headers
-pub fn should_encrypt(config: &EncryptionConfig, headers: &axum::http::HeaderMap) -> bool {
+pub fn should_encrypt(config: &EncryptionConfig, headers: &http::HeaderMap) -> bool {
     // Check if encryption is globally enabled
     if !config.enabled {
         return false;
@@ -447,7 +447,7 @@ mod tests {
         let key = vec![0u8; 32];
         let config = EncryptionConfig::new(true, "test".to_string(), key).unwrap();
 
-        let headers = axum::http::HeaderMap::new();
+        let headers = http::HeaderMap::new();
         assert!(should_encrypt(&config, &headers));
 
         // Test with disabled config

crates/protocols/src/swift/expiration.rs

@@ -86,7 +86,7 @@ pub fn parse_delete_after(value: &str) -> SwiftResult<u64> {
 ///
 /// Checks both X-Delete-At and X-Delete-After headers.
 /// X-Delete-After takes precedence and is converted to X-Delete-At.
-pub fn extract_expiration(headers: &axum::http::HeaderMap) -> SwiftResult<Option<u64>> {
+pub fn extract_expiration(headers: &http::HeaderMap) -> SwiftResult<Option<u64>> {
     // Check X-Delete-After first (takes precedence)
     if let Some(delete_after) = headers.get("x-delete-after")
         && let Ok(value_str) = delete_after.to_str()
@@ -239,7 +239,7 @@ mod tests {
 
     #[test]
     fn test_extract_expiration_delete_at() {
-        let mut headers = axum::http::HeaderMap::new();
+        let mut headers = http::HeaderMap::new();
         headers.insert("x-delete-at", "1740000000".parse().unwrap());
 
         let result = extract_expiration(&headers);
@@ -249,7 +249,7 @@ mod tests {
 
     #[test]
     fn test_extract_expiration_delete_after() {
-        let mut headers = axum::http::HeaderMap::new();
+        let mut headers = http::HeaderMap::new();
         headers.insert("x-delete-after", "3600".parse().unwrap());
 
         let result = extract_expiration(&headers);
@@ -263,7 +263,7 @@ mod tests {
 
     #[test]
     fn test_extract_expiration_delete_after_precedence() {
-        let mut headers = axum::http::HeaderMap::new();
+        let mut headers = http::HeaderMap::new();
         headers.insert("x-delete-at", "1740000000".parse().unwrap());
         headers.insert("x-delete-after", "3600".parse().unwrap());
 
@@ -280,7 +280,7 @@ mod tests {
 
     #[test]
     fn test_extract_expiration_none() {
-        let headers = axum::http::HeaderMap::new();
+        let headers = http::HeaderMap::new();
 
         let result = extract_expiration(&headers);
         assert!(result.is_ok());

crates/protocols/src/swift/formpost.rs

@@ -364,7 +364,7 @@ pub async fn handle_formpost(
     body: Vec<u8>,
     tempurl_key: &str,
     credentials: &rustfs_credentials::Credentials,
-) -> SwiftResult<axum::http::Response<s3s::Body>> {
+) -> SwiftResult<http::Response<s3s::Body>> {
     use axum::http::{Response, StatusCode};
 
     // Parse multipart boundary
@@ -424,9 +424,9 @@ pub async fn handle_formpost(
         let reader = std::io::Cursor::new(file.contents.clone());
 
         // Create headers for upload
-        let mut upload_headers = axum::http::HeaderMap::new();
+        let mut upload_headers = http::HeaderMap::new();
         if let Some(ct) = &file.content_type
-            && let Ok(header_value) = axum::http::HeaderValue::from_str(ct)
+            && let Ok(header_value) = http::HeaderValue::from_str(ct)
         {
             upload_headers.insert("content-type", header_value);
         }

crates/protocols/src/swift/handler.rs

@@ -132,7 +132,7 @@ async fn handle_swift_request(
         // Get account TempURL key
         let tempurl_key = super::account::get_tempurl_key(account, &credentials).await?;
 
-        if let Some(key) = tempurl_key {
+        return if let Some(key) = tempurl_key {
             // Validate TempURL signature
             let tempurl = tempurl::TempURL::new(key);
             let path = uri.path();
@@ -144,11 +144,11 @@ async fn handle_swift_request(
 
             // Reconstruct request for object operation
             let req = Request::from_parts(parts, body);
-            return handle_tempurl_object_request(req, route).await;
+            handle_tempurl_object_request(req, route).await
         } else {
             // No TempURL key configured for this account
-            return Err(SwiftError::Unauthorized("TempURL key not configured for this account".to_string()));
-        }
+            Err(SwiftError::Unauthorized("TempURL key not configured for this account".to_string()))
+        };
     }
 
     // No TempURL or TempURL validation failed - require normal authentication
@@ -476,7 +476,7 @@ async fn handle_authenticated_request(
                         // FormPost upload - get TempURL key for signature validation
                         let tempurl_key = super::account::get_tempurl_key(&account, &Some(credentials.clone())).await?;
 
-                        if let Some(key) = tempurl_key {
+                        return if let Some(key) = tempurl_key {
                             // Collect body for multipart parsing
                             use http_body_util::BodyExt;
                             let body_bytes = body
@@ -489,19 +489,11 @@ async fn handle_authenticated_request(
                             // Build path for signature validation
                             let path = format!("/v1/{}/{}", account, container);
 
-                            return super::formpost::handle_formpost(
-                                &account,
-                                &container,
-                                &path,
-                                ct_str,
-                                body_bytes,
-                                &key,
-                                &credentials,
-                            )
-                            .await;
+                            super::formpost::handle_formpost(&account, &container, &path, ct_str, body_bytes, &key, &credentials)
+                                .await
                         } else {
-                            return Err(SwiftError::Unauthorized("TempURL key not configured for FormPost".to_string()));
-                        }
+                            Err(SwiftError::Unauthorized("TempURL key not configured for FormPost".to_string()))
+                        };
                     }
 
                     // Check for versioning headers first
@@ -1018,9 +1010,8 @@ async fn handle_authenticated_request(
 }
 
 // Type alias for complex symlink resolution future
-type SymlinkResolutionFuture<'a> = std::pin::Pin<
-    Box<dyn std::future::Future<Output = Result<(String, String, String, Option<String>), SwiftError>> + Send + 'a>,
->;
+type SymlinkResolutionFuture<'a> =
+    Pin<Box<dyn std::future::Future<Output = Result<(String, String, String, Option<String>), SwiftError>> + Send + 'a>>;
 
 /// Resolve symlink chain recursively
 ///
@@ -1073,7 +1064,7 @@ async fn handle_object_get(
     account: &str,
     container: &str,
     object: &str,
-    headers: &axum::http::HeaderMap,
+    headers: &http::HeaderMap,
     credentials: &Option<Credentials>,
 ) -> Result<Response<Body>, SwiftError> {
     // For TempURL requests, credentials will be None
@@ -1269,7 +1260,7 @@ async fn handle_object_put(
     container: &str,
     object: &str,
     body: Body,
-    headers: &axum::http::HeaderMap,
+    headers: &http::HeaderMap,
     credentials: &Option<Credentials>,
 ) -> Result<Response<Body>, SwiftError> {
     let creds = credentials
@@ -1323,7 +1314,7 @@ async fn check_container_acl(
     container: &str,
     credentials: &Credentials,
     is_write: bool,
-    headers: &axum::http::HeaderMap,
+    headers: &http::HeaderMap,
 ) -> Result<(), SwiftError> {
     // Get container ACLs
     let acl = container::get_container_acl(account, container, credentials).await?;
@@ -1396,7 +1387,7 @@ async fn inject_cors_headers(
     container: Option<&str>,
     account: &str,
     credentials: &Credentials,
-    request_headers: &axum::http::HeaderMap,
+    request_headers: &http::HeaderMap,
 ) -> Response<Body> {
     // Only inject CORS for container/object routes
     if let Some(container_name) = container {

crates/protocols/src/swift/symlink.rs

@@ -128,7 +128,7 @@ impl SymlinkTarget {
 }
 
 /// Extract symlink target from request headers
-pub fn extract_symlink_target(headers: &axum::http::HeaderMap) -> SwiftResult<Option<SymlinkTarget>> {
+pub fn extract_symlink_target(headers: &http::HeaderMap) -> SwiftResult<Option<SymlinkTarget>> {
     if let Some(target_header) = headers.get("x-object-symlink-target") {
         let target_str = target_header
             .to_str()
@@ -248,7 +248,7 @@ mod tests {
 
     #[test]
     fn test_extract_symlink_target_present() {
-        let mut headers = axum::http::HeaderMap::new();
+        let mut headers = http::HeaderMap::new();
         headers.insert("x-object-symlink-target", "target.txt".parse().unwrap());
 
         let result = extract_symlink_target(&headers).unwrap();
@@ -261,7 +261,7 @@ mod tests {
 
     #[test]
     fn test_extract_symlink_target_absent() {
-        let headers = axum::http::HeaderMap::new();
+        let headers = http::HeaderMap::new();
         let result = extract_symlink_target(&headers).unwrap();
         assert!(result.is_none());
     }

crates/protocols/src/swift/tempurl.rs

@@ -1,3 +1,17 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
 //! TempURL (Temporary URL) support for OpenStack Swift
 //!
 //! TempURLs provide time-limited access to objects without requiring authentication.

crates/protocols/src/swift/versioning.rs

@@ -1,3 +1,17 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
 //! Object Versioning Support for Swift API
 //!
 //! Implements Swift object versioning where old versions are automatically