feat(storage): stage multipart compression behind RUSTFS_COMPRESSION_MULTIPART_ENABLED

Review follow-up: a rolling-upgrade window must not create new compressed multipart objects while pre-fix nodes (whose decompressor is not resumable) may still serve reads. The session marker is now additionally gated on RUSTFS_COMPRESSION_MULTIPART_ENABLED, default off, so the restored capability stays dark until the operator confirms fleet convergence. The default flips per the multipart-compression-default-off-window entry in docs/architecture/compat-cleanup-register.md once the minimum supported direct-upgrade release ships the resumable decoder.

rustfs/src/app/multipart_usecase.rs

@@ -27,7 +27,7 @@ use super::storage_api::multipart_usecase::bucket::{
     replication::{must_replicate_object, schedule_object_replication},
     versioning_sys::BucketVersioningSys,
 };
-use super::storage_api::multipart_usecase::compression::is_disk_compressible;
+use super::storage_api::multipart_usecase::compression::{is_disk_compressible, is_multipart_disk_compression_enabled};
 #[cfg(test)]
 use super::storage_api::multipart_usecase::contract::http::HTTPPreconditions;
 use super::storage_api::multipart_usecase::contract::multipart::{CompletePart, MultipartOperations as _, MultipartUploadResult};
@@ -211,9 +211,16 @@ fn create_multipart_upload_metadata(
     metadata
 }
 
-/// A multipart session advertises disk compression only when the object key/headers
-/// qualify AND the session is not an SSE-C ciphertext-passthrough replication session,
-/// which must preserve source bytes verbatim.
+/// A multipart session advertises disk compression only when the staged-rollout
+/// switch (`RUSTFS_COMPRESSION_MULTIPART_ENABLED`) is on, the object key/headers
+/// qualify, AND the session is not an SSE-C ciphertext-passthrough replication
+/// session, which must preserve source bytes verbatim.
+///
+/// The rollout switch defaults to off so a rolling upgrade never creates new
+/// compressed multipart objects while pre-fix nodes (whose decompressor is not
+/// resumable) may still serve reads. Enable it once the fleet has converged on a
+/// fixed build; the default flips per the `multipart-compression-default-off-window`
+/// entry in docs/architecture/compat-cleanup-register.md.
 ///
 /// Each part is compressed as an independent stream; the GET path decodes across part
 /// boundaries (see `ReadTransform::Compressed`), so the session may advertise
@@ -222,8 +229,8 @@ fn create_multipart_upload_metadata(
 /// Unlike single PUT there is no `MIN_DISK_COMPRESSIBLE_SIZE` floor here: the total
 /// object size is unknown at CreateMultipartUpload time, so tiny multipart objects pay
 /// the (harmless) framing overhead. This is a deliberate trade-off, not a bug.
-fn should_advertise_session_compression(ciphertext_passthrough: bool, disk_compressible: bool) -> bool {
-    !ciphertext_passthrough && disk_compressible
+fn should_advertise_session_compression(multipart_enabled: bool, ciphertext_passthrough: bool, disk_compressible: bool) -> bool {
+    multipart_enabled && !ciphertext_passthrough && disk_compressible
 }
 
 async fn validate_table_catalog_object_mutation(bucket: &str, key: &str) -> S3Result<()> {
@@ -853,7 +860,11 @@ impl DefaultMultipartUsecase {
             None => (None, None),
         };
 
-        if should_advertise_session_compression(ciphertext_passthrough, is_disk_compressible(&req.headers, &key)) {
+        if should_advertise_session_compression(
+            is_multipart_disk_compression_enabled(),
+            ciphertext_passthrough,
+            is_disk_compressible(&req.headers, &key),
+        ) {
             rustfs_utils::http::insert_str(
                 &mut metadata,
                 rustfs_utils::http::SUFFIX_COMPRESSION,
@@ -1655,19 +1666,25 @@ mod tests {
 
     #[test]
     fn session_compression_is_advertised_only_for_non_passthrough_compressible_uploads() {
-        // (ciphertext_passthrough, disk_compressible, expected)
+        // (multipart_enabled, ciphertext_passthrough, disk_compressible, expected)
         let cases = [
-            (false, false, false),
-            (false, true, true),
-            (true, false, false),
-            (true, true, false),
+            (true, false, false, false),
+            (true, false, true, true),
+            (true, true, false, false),
+            (true, true, true, false),
+            // The staged-rollout switch keeps multipart compression dark by
+            // default regardless of the other gates.
+            (false, false, true, false),
+            (false, false, false, false),
+            (false, true, true, false),
+            (false, true, false, false),
         ];
 
-        for (ciphertext_passthrough, disk_compressible, expected) in cases {
+        for (multipart_enabled, ciphertext_passthrough, disk_compressible, expected) in cases {
             assert_eq!(
-                should_advertise_session_compression(ciphertext_passthrough, disk_compressible),
+                should_advertise_session_compression(multipart_enabled, ciphertext_passthrough, disk_compressible),
                 expected,
-                "ciphertext_passthrough={ciphertext_passthrough} disk_compressible={disk_compressible}"
+                "multipart_enabled={multipart_enabled} ciphertext_passthrough={ciphertext_passthrough} disk_compressible={disk_compressible}"
             );
         }
     }

rustfs/src/app/storage_api.rs

@@ -942,7 +942,9 @@ pub(crate) mod concurrency {
 }
 
 pub(crate) mod compression {
-    pub(crate) use crate::storage::storage_api::ecstore_compression::{MIN_DISK_COMPRESSIBLE_SIZE, is_disk_compressible};
+    pub(crate) use crate::storage::storage_api::ecstore_compression::{
+        MIN_DISK_COMPRESSIBLE_SIZE, is_disk_compressible, is_multipart_disk_compression_enabled,
+    };
 }
 
 pub(crate) mod deadlock_detector {

rustfs/src/storage/storage_api.rs

@@ -409,7 +409,9 @@ pub(crate) mod ecstore_client {
 }
 
 pub(crate) mod ecstore_compression {
-    pub(crate) use rustfs_ecstore::api::compression::{MIN_DISK_COMPRESSIBLE_SIZE, is_disk_compressible};
+    pub(crate) use rustfs_ecstore::api::compression::{
+        MIN_DISK_COMPRESSIBLE_SIZE, is_disk_compressible, is_multipart_disk_compression_enabled,
+    };
 }
 
 pub(crate) mod ecstore_cluster {