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.

crates/e2e_test/src/compression_test.rs

@@ -72,6 +72,7 @@ async fn start_rustfs_with_compression(env: &mut RustFSTestEnvironment) -> Resul
     command
         .env("RUSTFS_CONSOLE_ENABLE", "false")
         .env("RUSTFS_COMPRESSION_ENABLED", "true")
+        .env("RUSTFS_COMPRESSION_MULTIPART_ENABLED", "true")
         .args([
             "--address",
             &env.address,
@@ -652,6 +653,7 @@ async fn start_rustfs_with_compression_and_sse(
     let process = Command::new(&binary_path)
         .env("RUSTFS_CONSOLE_ENABLE", "false")
         .env("RUSTFS_COMPRESSION_ENABLED", "true")
+        .env("RUSTFS_COMPRESSION_MULTIPART_ENABLED", "true")
         .env("RUSTFS_SSE_S3_MASTER_KEY", master_key)
         .env("RUST_LOG", "rustfs=info,rustfs_ecstore=info")
         .stdout(std::process::Stdio::from(server_log))

crates/e2e_test/src/inline_fast_path_cluster_test.rs

@@ -1839,6 +1839,7 @@ async fn four_node_multipart_disk_compression_roundtrip() -> TestResult {
     let mut cluster = RustFSTestClusterEnvironment::new(4).await?;
     configure_reader_metric_cluster(&mut cluster, &collector);
     cluster.set_env("RUSTFS_COMPRESSION_ENABLED", "true");
+    cluster.set_env("RUSTFS_COMPRESSION_MULTIPART_ENABLED", "true");
     cluster.start().await?;
 
     let bucket = "inline-multipart-compression-roundtrip";
@@ -1873,6 +1874,7 @@ async fn four_node_mixed_msgpack_compat_mode_preserves_fallback_controls() -> Te
     let sse_master_key = base64::engine::general_purpose::STANDARD.encode([0x42u8; 32]);
     cluster.set_env("RUSTFS_SSE_S3_MASTER_KEY", sse_master_key);
     cluster.set_env("RUSTFS_COMPRESSION_ENABLED", "true");
+    cluster.set_env("RUSTFS_COMPRESSION_MULTIPART_ENABLED", "true");
     configure_mixed_msgpack_cluster(&mut cluster, &collector)?;
     cluster.start().await?;
 

crates/ecstore/src/api/mod.rs

@@ -276,7 +276,9 @@ pub mod cluster {
 }
 
 pub mod compression {
-    pub use crate::io_support::compress::{MIN_DISK_COMPRESSIBLE_SIZE, is_disk_compressible, is_disk_compression_enabled};
+    pub use crate::io_support::compress::{
+        MIN_DISK_COMPRESSIBLE_SIZE, is_disk_compressible, is_disk_compression_enabled, is_multipart_disk_compression_enabled,
+    };
 }
 
 pub mod config {

crates/ecstore/src/io_support/compress.rs

@@ -31,6 +31,13 @@ pub const ENV_DISK_COMPRESSION_MIME_TYPES: &str = "RUSTFS_COMPRESSION_MIME_TYPES
 // Environment variable for additional extensions to exclude from compression (comma-separated, e.g. ".foo,.bar")
 pub const ENV_ADDED_EXCLUDE_COMPRESS_EXTENSIONS: &str = "RUSTFS_ADDED_EXCLUDE_COMPRESS_EXTENSIONS";
 
+// Environment variable to additionally enable disk compression for multipart uploads.
+// Default off: nodes from before the resumable decompressor fix fail transient reads of
+// compressed objects, so multipart compression stays dark until the operator confirms the
+// fleet has converged on a fixed build (see docs/architecture/compat-cleanup-register.md,
+// `multipart-compression-default-off-window`, for the default-flip condition).
+pub const ENV_DISK_COMPRESSION_MULTIPART_ENABLED: &str = "RUSTFS_COMPRESSION_MULTIPART_ENABLED";
+
 pub const DEFAULT_DISK_COMPRESS_EXTENSIONS: &str = ".txt,.log,.csv,.json,.tar,.xml,.bin";
 pub const DEFAULT_DISK_COMPRESS_MIME_TYPES: &str = "text/*,application/json,application/xml,binary/octet-stream";
 
@@ -171,6 +178,21 @@ pub fn is_disk_compression_enabled() -> bool {
     DISK_COMPRESSION_CONFIG.get_or_init(parse_disk_compression_config).enabled
 }
 
+// Parsed once at first use, mirroring DISK_COMPRESSION_CONFIG.
+static MULTIPART_DISK_COMPRESSION_ENABLED: OnceLock<bool> = OnceLock::new();
+
+/// Whether multipart uploads may advertise disk compression. Requires the
+/// regular disk-compression gates to pass as well; this is the staged-rollout
+/// switch that keeps multipart compression dark during rolling upgrades from
+/// builds whose decompressor was not yet resumable.
+pub fn is_multipart_disk_compression_enabled() -> bool {
+    *MULTIPART_DISK_COMPRESSION_ENABLED.get_or_init(|| {
+        env::var(ENV_DISK_COMPRESSION_MULTIPART_ENABLED)
+            .map(|s| matches!(s.to_ascii_lowercase().as_str(), "true" | "on" | "1"))
+            .unwrap_or(false)
+    })
+}
+
 fn is_disk_compressible_with_config(headers: &http::HeaderMap, object_name: &str, config: &DiskCompressionConfig) -> bool {
     // Check if disk compression is enabled (read once at first use, then fixed for process lifetime)
     if !config.enabled {

docs/architecture/compat-cleanup-register.md

@@ -35,6 +35,7 @@ for later deletion.
 - `rustfs-5063` pre-beta.9 Local KMS recovery: persisted Local KMS configs from beta.8 and earlier predate the explicit insecure-development flag, and encrypted key files use the legacy SHA-256 KDF. Remove the config fallback after supported upgrades have rewritten or explicitly resaved all pre-beta.9 configs with the development-default field, and remove the legacy KDF after supported upgrades have rewritten all pre-beta.9 Local KMS key files with explicit at-rest protection.
 - `sse-local-dek-json-v1` legacy local SSE DEK decoding: releases before the JSON envelope wrote wrapped DEKs as `base64(nonce):base64(ciphertext)`, so readers retain that decoder while all new writes use the versioned JSON envelope. Remove the colon decoder after the minimum supported direct-upgrade release writes JSON envelopes and migration tooling has rewritten every retained legacy object.
 - `rio-v2-dormant-variant` dormant `rio-v2` build variant: `crates/rio-v2` and the `rio-v2` feature ship in no default or release build and exist only as the candidate MinIO stream-format implementation for the rustfs/backlog#1638 SSE-interop adjudication. Per-PR CI keeps only `test-and-lint-rio-v2` to guard the `#[cfg(feature = "rio-v2")]` seam; the full-suite lanes (`build-rustfs-debug-binary-rio-v2`, `e2e-tests-rio-v2` in `.github/workflows/ci.yml`) run on schedule/workflow_dispatch only — see the "`rio-v2` variant lifecycle" section in [minio-file-format-compat.md](minio-file-format-compat.md). While both implementations exist, DARE/S2 stream fixes must land in both `crates/rio` and `crates/rio-v2`. No `RUSTFS_COMPAT_TODO` source marker applies: the temporary surface is CI workflow YAML plus an entire feature-gated candidate crate, not a compatibility code path inside shipping code, and workflow files are outside the marker convention's Rust scope. Remove after the #1638 adjudication lands and converges on one implementation: delete the losing implementation, its feature seam, and the gating CI jobs.
+- `multipart-compression-default-off-window` staged multipart disk-compression rollout: releases before the resumable legacy decompressor fail transient reads of compressed objects under mid-payload suspension, so multipart uploads advertise the compression marker only when `RUSTFS_COMPRESSION_MULTIPART_ENABLED` is set in addition to `RUSTFS_COMPRESSION_ENABLED`, keeping rolling upgrades from creating new compressed multipart objects while pre-fix nodes may still serve reads. Flip the default to enabled (and retire the extra switch) after the minimum supported direct-upgrade release ships the resumable `DecompressReader`.
 
 ## Review Checklist
 

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};
@@ -209,9 +209,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
@@ -220,8 +227,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<()> {
@@ -886,7 +893,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,
@@ -1688,19 +1699,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

@@ -940,7 +940,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 {

scripts/run.sh

@@ -209,6 +209,7 @@ export RUSTFS_NS_SCANNER_INTERVAL=60  # Object scanning interval in seconds
 
 # Storage level compression (compression at object storage level)
 # export RUSTFS_COMPRESSION_ENABLED=true # Whether to enable storage-level compression for objects
+# export RUSTFS_COMPRESSION_MULTIPART_ENABLED=true # Additionally compress multipart uploads (staged rollout switch: enable only after the whole fleet runs a build with the resumable decompressor; see docs/architecture/compat-cleanup-register.md)
 
 # HTTP Response Compression (whitelist-based, aligned with MinIO)
 # By default, HTTP response compression is DISABLED (aligned with MinIO behavior)