feat: stream object zip downloads (#3380)

* feat: stream object zip downloads

* fix: stream zip downloads page by page

* fix: prepare zip downloads before streaming

---------

Signed-off-by: houseme <housemecn@gmail.com>
Co-authored-by: houseme <housemecn@gmail.com>

Cargo.lock

@@ -635,6 +635,7 @@ checksum = "e79b3f8a79cccc2898f31920fc69f304859b3bd567490f75ebf51ae1c792a9ac"
 dependencies = [
  "compression-codecs",
  "compression-core",
+ "futures-io",
  "pin-project-lite",
  "tokio",
 ]
@@ -778,6 +779,21 @@ dependencies = [
  "tungstenite",
 ]
 
+[[package]]
+name = "async_zip"
+version = "0.0.18"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0d8c50d65ce1b0e0cb65a785ff615f78860d7754290647d3b983208daa4f85e6"
+dependencies = [
+ "async-compression",
+ "crc32fast",
+ "futures-lite",
+ "pin-project",
+ "thiserror 2.0.18",
+ "tokio",
+ "tokio-util",
+]
+
 [[package]]
 name = "atoi"
 version = "2.0.0"
@@ -9031,6 +9047,7 @@ dependencies = [
  "anyhow",
  "astral-tokio-tar",
  "async-trait",
+ "async_zip",
  "atoi 3.0.0",
  "atomic_enum",
  "aws-config",
@@ -9044,6 +9061,7 @@ dependencies = [
  "datafusion",
  "flatbuffers",
  "futures",
+ "futures-lite",
  "futures-util",
  "hashbrown 0.17.1",
  "hex-simd",

Cargo.toml

@@ -125,6 +125,7 @@ rustfs-zip = { path = "./crates/zip", version = "1.0.0-beta.8" }
 
 # Async Runtime and Networking
 async-channel = "2.5.0"
+async_zip = { version = "0.0.18", default-features = false, features = ["tokio", "deflate"] }
 mysql_async = { version = "0.37", default-features = false, features = ["default-rustls", "tracing"] }
 async-compression = { version = "0.4.42" }
 async-recursion = "1.1.1"
@@ -133,6 +134,7 @@ async-nats = "0.49.1"
 axum = "0.8.9"
 futures = "0.3.32"
 futures-core = "0.3.32"
+futures-lite = "2.6.1"
 futures-util = "0.3.32"
 pollster = "0.4.0"
 pulsar = { version = "6.8.0", default-features = false, features = ["tokio-rustls-runtime"] }

rustfs/Cargo.toml

@@ -104,6 +104,7 @@ tempfile = { workspace = true }
 async-trait = { workspace = true }
 axum.workspace = true
 futures.workspace = true
+futures-lite.workspace = true
 futures-util.workspace = true
 hyper.workspace = true
 hyper-util.workspace = true
@@ -141,6 +142,7 @@ time = { workspace = true, features = ["parsing", "formatting", "serde"] }
 astral-tokio-tar = { workspace = true }
 atoi = { workspace = true }
 atomic_enum = { workspace = true }
+async_zip = { workspace = true }
 base64 = { workspace = true }
 hmac = { workspace = true }
 sha2 = { workspace = true }

rustfs/src/admin/handlers/mod.rs

@@ -31,6 +31,7 @@ pub mod kms_management;
 pub mod metrics;
 pub mod module_switch;
 mod notify_runtime_access;
+pub mod object_zip_download;
 pub mod oidc;
 pub mod plugins_catalog;
 pub mod plugins_instances;
@@ -75,6 +76,7 @@ mod tests {
         let _get_extension_catalog = extensions::GetExtensionCatalogHandler {};
         let _list_extension_instances = extensions::ListExtensionInstancesHandler {};
         let _get_plugin_catalog = plugins_catalog::GetPluginCatalogHandler {};
+        let _create_object_zip_download = object_zip_download::CreateObjectZipDownloadHandler {};
         let _list_plugin_instances = plugins_instances::ListPluginInstancesHandler {};
         let _get_plugin_instance = plugins_instances::GetPluginInstanceHandler {};
         let _put_plugin_instance = plugins_instances::PutPluginInstanceHandler {};

rustfs/src/admin/mod.rs

@@ -31,9 +31,9 @@ mod console_test;
 mod route_registration_test;
 
 use handlers::{
-    audit, bucket_meta, config_admin, extensions, heal, health, kms, module_switch, oidc, plugins_catalog, plugins_instances,
-    pools, profile_admin, quota, rebalance, replication, scanner, site_replication, sts, system, table_catalog, tier, tls_debug,
-    user,
+    audit, bucket_meta, config_admin, extensions, heal, health, kms, module_switch, object_zip_download, oidc, plugins_catalog,
+    plugins_instances, pools, profile_admin, quota, rebalance, replication, scanner, site_replication, sts, system,
+    table_catalog, tier, tls_debug, user,
 };
 use router::{AdminOperation, S3Router};
 use s3s::route::S3Route;
@@ -71,6 +71,7 @@ fn register_admin_routes(r: &mut S3Router<AdminOperation>) -> std::io::Result<()
     audit::register_audit_target_route(r)?;
     module_switch::register_module_switch_route(r)?;
     extensions::register_extension_route(r)?;
+    object_zip_download::register_object_zip_download_route(r)?;
     plugins_catalog::register_plugin_catalog_route(r)?;
     plugins_instances::register_plugin_instance_route(r)?;
 

rustfs/src/admin/route_policy.rs

@@ -981,6 +981,16 @@ pub const DEFERRED_ADMIN_ROUTE_POLICIES: &[DeferredAdminRoutePolicy] = &[
         "/rustfs/admin/v3/datausageinfo",
         DeferredRoutePolicyReason::MultipleActions,
     ),
+    deferred(
+        HttpMethod::Post,
+        "/rustfs/admin/v3/object-zip-downloads",
+        DeferredRoutePolicyReason::S3Action,
+    ),
+    deferred(
+        HttpMethod::Get,
+        "/rustfs/admin/v3/object-zip-downloads/{id}.zip",
+        DeferredRoutePolicyReason::CredentialOnly,
+    ),
     deferred(HttpMethod::Get, "/rustfs/admin/v3/metrics", DeferredRoutePolicyReason::CredentialOnly),
     deferred(HttpMethod::Get, "/rustfs/admin/v3/pools/list", DeferredRoutePolicyReason::MultipleActions),
     deferred(
@@ -1220,6 +1230,16 @@ mod tests {
             DeferredRoutePolicyReason::MultipleActions,
         );
         assert_deferred(HttpMethod::Get, "/rustfs/admin/v3/accountinfo", DeferredRoutePolicyReason::S3Action);
+        assert_deferred(
+            HttpMethod::Post,
+            "/rustfs/admin/v3/object-zip-downloads",
+            DeferredRoutePolicyReason::S3Action,
+        );
+        assert_deferred(
+            HttpMethod::Get,
+            "/rustfs/admin/v3/object-zip-downloads/{id}.zip",
+            DeferredRoutePolicyReason::CredentialOnly,
+        );
     }
 
     fn route_policy_inventory_keys() -> BTreeSet<String> {

rustfs/src/admin/route_registration_test.rs

@@ -225,6 +225,12 @@ fn expected_admin_route_matrix() -> Vec<RouteMatrixEntry> {
         admin_route(Method::PUT, "/v3/module-switches"),
         admin_route(Method::GET, "/v4/extensions/catalog"),
         admin_route(Method::GET, "/v4/extensions/instances"),
+        admin_route(Method::POST, "/v3/object-zip-downloads"),
+        admin_route_sample(
+            Method::GET,
+            "/v3/object-zip-downloads/{id}.zip",
+            "/v3/object-zip-downloads/example-id.zip",
+        ),
         admin_route(Method::GET, "/v4/plugins/catalog"),
         admin_route(Method::GET, "/v4/plugins/instances"),
         admin_route_sample(Method::GET, "/v4/plugins/instances/{id}", "/v4/plugins/instances/example-id"),
@@ -567,6 +573,7 @@ fn test_register_routes_cover_representative_admin_paths() {
     assert_route(&router, Method::PUT, &admin_path("/v3/module-switches"));
     assert_route(&router, Method::GET, &admin_path("/v4/extensions/catalog"));
     assert_route(&router, Method::GET, &admin_path("/v4/extensions/instances"));
+    assert_route(&router, Method::POST, &admin_path("/v3/object-zip-downloads"));
     assert_route(&router, Method::GET, &admin_path("/v4/plugins/catalog"));
     assert_route(&router, Method::GET, &admin_path("/v4/plugins/instances"));
     assert_route(&router, Method::GET, &admin_path("/v4/plugins/instances/example-id"));

rustfs/src/admin/router.rs

@@ -99,6 +99,7 @@ use tracing::{error, warn};
 use url::form_urlencoded;
 use uuid::Uuid;
 
+pub const ADMIN_OBJECT_ZIP_DOWNLOADS_PATH: &str = "/v3/object-zip-downloads";
 const LOG_COMPONENT_ADMIN: &str = "admin";
 const LOG_SUBSYSTEM_OBJECT_LAMBDA: &str = "object_lambda";
 const LOG_SUBSYSTEM_LIVE_EVENTS: &str = "live_events";
@@ -2298,6 +2299,17 @@ fn is_public_health_path(path: &str) -> bool {
     path == HEALTH_PREFIX || path == HEALTH_READY_PATH
 }
 
+fn is_object_zip_download_token_path(method: &Method, uri: &Uri) -> bool {
+    if method != Method::GET {
+        return false;
+    }
+
+    let path = canonicalize_admin_path(uri.path());
+    path.starts_with(&format!("{ADMIN_PREFIX}{ADMIN_OBJECT_ZIP_DOWNLOADS_PATH}/"))
+        && path.ends_with(".zip")
+        && query_value_exact(uri, "token").is_some_and(|token| !token.is_empty())
+}
+
 fn canonicalize_admin_path(path: &str) -> std::borrow::Cow<'_, str> {
     if is_admin_path(path)
         && let Some(suffix) = path.strip_prefix(MINIO_ADMIN_PREFIX)
@@ -2441,6 +2453,12 @@ where
             return Ok(());
         }
 
+        // Object ZIP downloads are browser-navigated with a short-lived token;
+        // the handler validates the token before returning any bytes.
+        if is_object_zip_download_token_path(&req.method, &req.uri) {
+            return Ok(());
+        }
+
         // Allow unauthenticated STS requests to POST / (AssumeRoleWithWebIdentity
         // doesn't use SigV4 — the JWT token in the request body is the authentication).
         // The handler dispatches on the Action parameter: AssumeRole will reject if
@@ -3871,6 +3889,77 @@ mod tests {
         assert_eq!(err.code(), &S3ErrorCode::AccessDenied);
     }
 
+    #[tokio::test]
+    async fn check_access_allows_object_zip_download_token_navigation() {
+        let router: S3Router<AdminOperation> = S3Router::new(false);
+        let mut req = S3Request {
+            input: Body::from(String::new()),
+            method: Method::GET,
+            uri: "/rustfs/admin/v3/object-zip-downloads/example.zip?token=abc"
+                .parse()
+                .expect("uri should parse"),
+            headers: HeaderMap::new(),
+            extensions: http::Extensions::new(),
+            credentials: None,
+            region: None,
+            service: None,
+            trailing_headers: None,
+        };
+
+        router
+            .check_access(&mut req)
+            .await
+            .expect("token download navigation should reach the handler");
+    }
+
+    #[tokio::test]
+    async fn check_access_rejects_object_zip_download_without_token() {
+        let router: S3Router<AdminOperation> = S3Router::new(false);
+        let mut req = S3Request {
+            input: Body::from(String::new()),
+            method: Method::GET,
+            uri: "/rustfs/admin/v3/object-zip-downloads/example.zip"
+                .parse()
+                .expect("uri should parse"),
+            headers: HeaderMap::new(),
+            extensions: http::Extensions::new(),
+            credentials: None,
+            region: None,
+            service: None,
+            trailing_headers: None,
+        };
+
+        let err = router
+            .check_access(&mut req)
+            .await
+            .expect_err("token download without token must be denied before handler");
+        assert_eq!(err.code(), &S3ErrorCode::AccessDenied);
+    }
+
+    #[tokio::test]
+    async fn check_access_rejects_anonymous_object_zip_download_post() {
+        let router: S3Router<AdminOperation> = S3Router::new(false);
+        let mut req = S3Request {
+            input: Body::from(String::new()),
+            method: Method::POST,
+            uri: "/rustfs/admin/v3/object-zip-downloads?token=abc"
+                .parse()
+                .expect("uri should parse"),
+            headers: HeaderMap::new(),
+            extensions: http::Extensions::new(),
+            credentials: None,
+            region: None,
+            service: None,
+            trailing_headers: None,
+        };
+
+        let err = router
+            .check_access(&mut req)
+            .await
+            .expect_err("token exception must not apply to POST");
+        assert_eq!(err.code(), &S3ErrorCode::AccessDenied);
+    }
+
     #[test]
     fn listen_notification_keepalive_plan_defaults_to_space_keepalive() {
         let uri: Uri = "/demo-bucket?events=s3:ObjectCreated:Put".parse().expect("uri should parse");

rustfs/src/server/http.rs

@@ -24,7 +24,7 @@ use crate::server::{
     layer::{
         BodylessStatusFixLayer, ConditionalCorsLayer, EmptyBodyContentLengthCompatLayer, HeadRequestBodyFixLayer,
         ObjectAttributesEtagFixLayer, PublicHealthEndpointLayer, RedirectLayer, RequestContextLayer, RequestLoggingLayer,
-        S3ErrorMessageCompatLayer,
+        S3ErrorMessageCompatLayer, redact_sensitive_uri_query,
     },
     tls_material::{
         TlsAcceptorHolder, TlsHandshakeFailureKind, build_acceptor_from_loaded, load_tls_material, spawn_reload_loop,
@@ -970,7 +970,7 @@ fn process_connection(
                             status_code = tracing::field::Empty,
                             method = %request.method(),
                             peer_addr = %peer_addr,
-                            uri = %request.uri(),
+                            uri = %redact_sensitive_uri_query(request.uri()),
                             version = ?request.version(),
                             user_agent = tracing::field::Empty,
                             content_type = tracing::field::Empty,

rustfs/src/server/layer.rs

@@ -43,11 +43,14 @@ use std::task::{Context, Poll};
 use std::time::Instant;
 use tower::{Layer, Service};
 use tracing::{debug, error, info};
+use url::form_urlencoded;
 
 const HTTP_REQUEST_COMPLETED_EVENT: &str = "http_request_completed";
 const HTTP_REQUEST_FAILED_EVENT: &str = "http_request_failed";
 const LOG_COMPONENT_SERVER: &str = "server";
 const LOG_SUBSYSTEM_HTTP: &str = "http";
+const REDACTED_QUERY_VALUE: &str = "redacted";
+const OBJECT_ZIP_DOWNLOADS_PATH: &str = "/v3/object-zip-downloads/";
 
 /// A carrier that adapts [`HeaderMap`] for OpenTelemetry trace context propagation.
 struct HeaderMapCarrier<'a>(&'a HeaderMap);
@@ -73,6 +76,55 @@ impl<'a> opentelemetry::propagation::Extractor for HeaderMapCarrier<'a> {
     }
 }
 
+pub(crate) fn redact_sensitive_uri_query(uri: &http::Uri) -> String {
+    let path = uri.path();
+    if !is_object_zip_download_path(path) {
+        return uri.to_string();
+    }
+
+    let Some(query) = uri.query() else {
+        return uri.to_string();
+    };
+
+    let mut redacted_token = false;
+    let mut serializer = form_urlencoded::Serializer::new(String::new());
+    for (key, value) in form_urlencoded::parse(query.as_bytes()) {
+        if key == "token" {
+            redacted_token = true;
+            serializer.append_pair(&key, REDACTED_QUERY_VALUE);
+        } else {
+            serializer.append_pair(&key, &value);
+        }
+    }
+
+    if !redacted_token {
+        return uri.to_string();
+    }
+
+    let redacted_query = serializer.finish();
+    let path_and_query = if redacted_query.is_empty() {
+        path.to_string()
+    } else {
+        format!("{path}?{redacted_query}")
+    };
+    let mut parts = uri.clone().into_parts();
+    match path_and_query.parse() {
+        Ok(path_and_query) => {
+            parts.path_and_query = Some(path_and_query);
+            http::Uri::from_parts(parts)
+                .map(|uri| uri.to_string())
+                .unwrap_or_else(|_| uri.to_string())
+        }
+        Err(_) => uri.to_string(),
+    }
+}
+
+fn is_object_zip_download_path(path: &str) -> bool {
+    (path.starts_with(ADMIN_PREFIX) || path.starts_with(MINIO_ADMIN_PREFIX))
+        && path.contains(OBJECT_ZIP_DOWNLOADS_PATH)
+        && path.ends_with(".zip")
+}
+
 /// Tower middleware layer that creates a canonical [`RequestContext`] from HTTP headers
 /// and injects it into `request.extensions()`.
 ///
@@ -199,7 +251,7 @@ impl RequestLogContext {
             span_id: request_context.as_ref().and_then(|ctx| ctx.span_id.clone()),
             peer_addr,
             method: req.method().to_string(),
-            uri: req.uri().to_string(),
+            uri: redact_sensitive_uri_query(req.uri()),
             request_started_at: request_context,
             fallback_start: Instant::now(),
         }
@@ -2667,6 +2719,41 @@ mod tests {
         assert_eq!(context.uri, "/bucket/object.txt");
     }
 
+    #[test]
+    fn request_log_context_redacts_object_zip_download_tokens() {
+        let request = Request::builder()
+            .method(Method::GET)
+            .uri("/rustfs/admin/v3/object-zip-downloads/download-id.zip?token=secret-token&part=1")
+            .body(())
+            .expect("request");
+
+        let context = RequestLogContext::from_request(&request);
+
+        assert_eq!(context.uri, "/rustfs/admin/v3/object-zip-downloads/download-id.zip?token=redacted&part=1");
+        assert!(!context.uri.contains("secret-token"));
+    }
+
+    #[test]
+    fn request_log_context_redacts_object_zip_download_tokens_for_minio_admin_prefix() {
+        let request = Request::builder()
+            .method(Method::GET)
+            .uri("/minio/admin/v3/object-zip-downloads/download-id.zip?token=secret-token&part=1")
+            .body(())
+            .expect("request");
+
+        let context = RequestLogContext::from_request(&request);
+
+        assert_eq!(context.uri, "/minio/admin/v3/object-zip-downloads/download-id.zip?token=redacted&part=1");
+        assert!(!context.uri.contains("secret-token"));
+    }
+
+    #[test]
+    fn redact_sensitive_uri_query_preserves_non_zip_download_uris() {
+        let uri: http::Uri = "/rustfs/admin/v3/users?token=not-a-download-token".parse().expect("uri");
+
+        assert_eq!(redact_sensitive_uri_query(&uri), "/rustfs/admin/v3/users?token=not-a-download-token");
+    }
+
     #[test]
     fn request_logging_layer_emits_single_completion_event_with_standard_fields() {
         let writer = SharedWriter::default();