fix(storage): restore multipart disk compression and make the legacy decompressor resumable (#6044)

* fix(storage): restore multipart disk compression and make the legacy decompressor resumable

Multipart uploads have bypassed disk compression since #5169 removed the session marker as a stopgap for mid-stream GET failures. The actual root cause was never the multipart layout: the legacy DecompressReader reset its payload consumption state on every poll re-entry, so a Poll::Pending in the middle of a block payload (routine under the erasure duplex) desynchronized the block framing and surfaced as LZ4 frameType errors. This rewrites the decoder as a resumable state machine, restores the multipart session compression marker, reports logical part sizes in ListParts, and makes the rebalance migration read raw stored bytes so compressed and encrypted objects survive migration verbatim.

Fixes #5957. Internal tracking: backlog#1848, backlog#1850.

* 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.

* chore(compat): satisfy the cleanup-register guard for the multipart compression switch

The architecture guard requires every backticked identifier in a register entry to carry a RUSTFS_COMPAT_TODO source marker: keep only the entry slug in backticks, and add the marker (with its literal Remove-after condition) at the switch definition.

* chore(rio): drop a dead store in the poison guard and note the end-block branch

Review follow-up: the poison gate re-assigned an already-true flag, and the COMPRESS_TYPE_END branch reads as dead without stating that the writer never emits an end block — that absence is exactly what lets concatenated per-part streams decode as one.

* fix(s3): report empty compressed multipart part size

* fix(s3): report empty encrypted multipart part size

rustfs/src/app/multipart_usecase.rs

@@ -27,6 +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, 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};
@@ -39,7 +40,7 @@ use super::storage_api::multipart_usecase::error::{StorageError, is_err_object_n
 use super::storage_api::multipart_usecase::helper::OperationHelper;
 #[cfg(test)]
 use super::storage_api::multipart_usecase::io::{DecryptReader, EncryptReader, HardLimitReader, boxed_reader, wrap_reader};
-use super::storage_api::multipart_usecase::io::{HashReader, WriteEncryption, WritePlan};
+use super::storage_api::multipart_usecase::io::{HashReader, WriteEncryption, WritePlan, compression_metadata_value};
 use super::storage_api::multipart_usecase::object_utils::to_s3s_etag;
 use super::storage_api::multipart_usecase::options::{
     copy_src_opts, extract_metadata_from_mime, get_complete_multipart_upload_opts_with_replication_authorization,
@@ -210,6 +211,28 @@ fn create_multipart_upload_metadata(
     metadata
 }
 
+/// 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
+/// object-level compression again.
+///
+/// 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(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<()> {
     table_catalog::validate_bucket_object_mutation(bucket, key)
         .await
@@ -837,8 +860,17 @@ impl DefaultMultipartUsecase {
             None => (None, None),
         };
 
-        // Multipart parts are independent physical streams. Advertising object-level
-        // compression here would make GET decode the completed object as one stream.
+        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,
+                compression_metadata_value(CompressionAlgorithm::default()),
+            );
+        }
 
         let mt2 = metadata.clone();
         let mut opts: ObjectOptions =
@@ -1632,6 +1664,31 @@ mod tests {
         DefaultMultipartUsecase::without_context()
     }
 
+    #[test]
+    fn session_compression_is_advertised_only_for_non_passthrough_compressible_uploads() {
+        // (multipart_enabled, ciphertext_passthrough, disk_compressible, expected)
+        let cases = [
+            (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 (multipart_enabled, ciphertext_passthrough, disk_compressible, expected) in cases {
+            assert_eq!(
+                should_advertise_session_compression(multipart_enabled, ciphertext_passthrough, disk_compressible),
+                expected,
+                "multipart_enabled={multipart_enabled} ciphertext_passthrough={ciphertext_passthrough} disk_compressible={disk_compressible}"
+            );
+        }
+    }
+
     #[test]
     fn quota_accounting_uses_logical_size_when_available() {
         let mut metadata = HashMap::new();

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 {
@@ -1153,7 +1155,7 @@ pub(crate) mod multipart_usecase {
     }
 
     pub(crate) use super::{
-        access, bucket, data_usage, error, helper, io, object_utils, options, request_context, s3_api, set_disk, sse,
+        access, bucket, compression, data_usage, error, helper, io, object_utils, options, request_context, s3_api, set_disk, sse,
     };
     pub(crate) use crate::storage::storage_api::{ECStore, StorageObjectInfo, StorageObjectOptions, StoragePutObjReader};
 }

rustfs/src/storage/s3_api/multipart.rs

@@ -39,6 +39,11 @@ pub(crate) struct ListMultipartUploadsParams {
 pub(crate) fn build_list_parts_output(res: ListPartsInfo) -> ListPartsOutput {
     let owner = rustfs_owner();
     let initiator = rustfs_initiator();
+    let transformed_parts = rustfs_utils::http::contains_key_str(&res.user_defined, rustfs_utils::http::SUFFIX_COMPRESSION)
+        || res
+            .user_defined
+            .keys()
+            .any(|key| rustfs_utils::http::is_object_encryption_marker(key));
 
     ListPartsOutput {
         bucket: Some(res.bucket),
@@ -51,7 +56,14 @@ pub(crate) fn build_list_parts_output(res: ListPartsInfo) -> ListPartsOutput {
                     e_tag: p.etag.map(|etag| to_s3s_etag(&etag)),
                     last_modified: p.last_mod.map(Timestamp::from),
                     part_number: p.part_num.try_into().ok(),
-                    size: p.size.try_into().ok(),
+                    // Compressed parts store fewer bytes than the client sent; S3
+                    // semantics report the uploaded (logical) size, matching
+                    // GetObjectAttributes ObjectParts.
+                    size: if p.actual_size > 0 || (transformed_parts && p.actual_size == 0) {
+                        Some(p.actual_size)
+                    } else {
+                        p.size.try_into().ok()
+                    },
                     ..Default::default()
                 })
                 .collect(),
@@ -247,6 +259,116 @@ mod tests {
         assert_eq!(output.initiator, Some(rustfs_initiator()));
     }
 
+    #[test]
+    fn test_list_parts_output_reports_logical_size_for_compressed_parts() {
+        let input = ListPartsInfo {
+            bucket: "bucket-a".to_string(),
+            object: "obj-a".to_string(),
+            upload_id: "upload-a".to_string(),
+            parts: vec![PartInfo {
+                part_num: 1,
+                // Stored (compressed) bytes on disk vs. the logical size the client uploaded.
+                size: 1_024,
+                actual_size: 8_388_608,
+                ..Default::default()
+            }],
+            ..Default::default()
+        };
+
+        let output = build_list_parts_output(input);
+        let parts = output.parts.as_ref().expect("parts should be present");
+
+        assert_eq!(parts.len(), 1);
+        assert_eq!(
+            parts[0].size,
+            Some(8_388_608),
+            "compressed parts must report the uploaded logical size, not the stored size"
+        );
+    }
+
+    #[test]
+    fn test_list_parts_output_reports_zero_logical_size_for_compressed_parts() {
+        let mut user_defined = std::collections::HashMap::new();
+        rustfs_utils::http::insert_str(&mut user_defined, rustfs_utils::http::SUFFIX_COMPRESSION, "S2".to_string());
+        let input = ListPartsInfo {
+            user_defined,
+            parts: vec![PartInfo {
+                part_num: 1,
+                // Legacy SSE writes an 8-byte end record for an empty part.
+                size: 8,
+                actual_size: 0,
+                ..Default::default()
+            }],
+            ..Default::default()
+        };
+
+        let output = build_list_parts_output(input);
+        let parts = output.parts.as_ref().expect("parts should be present");
+
+        assert_eq!(parts[0].size, Some(0));
+    }
+
+    #[test]
+    fn test_list_parts_output_reports_zero_logical_size_for_encrypted_parts() {
+        let input = ListPartsInfo {
+            user_defined: std::collections::HashMap::from([(
+                rustfs_utils::http::AMZ_SERVER_SIDE_ENCRYPTION.to_string(),
+                "AES256".to_string(),
+            )]),
+            parts: vec![
+                PartInfo {
+                    part_num: 1,
+                    size: 8,
+                    actual_size: 0,
+                    ..Default::default()
+                },
+                PartInfo {
+                    part_num: 2,
+                    size: 8,
+                    actual_size: -1,
+                    ..Default::default()
+                },
+            ],
+            ..Default::default()
+        };
+
+        let output = build_list_parts_output(input);
+        let parts = output.parts.as_ref().expect("parts should be present");
+
+        assert_eq!(parts[0].size, Some(0));
+        assert_eq!(parts[1].size, Some(8));
+    }
+
+    #[test]
+    fn test_list_parts_output_falls_back_to_stored_size_when_actual_size_unknown() {
+        let input = ListPartsInfo {
+            parts: vec![
+                PartInfo {
+                    part_num: 1,
+                    size: 1_024,
+                    // Uncompressed parts leave actual_size unset.
+                    actual_size: 0,
+                    ..Default::default()
+                },
+                PartInfo {
+                    part_num: 2,
+                    size: 1_024,
+                    // Legacy/unknown sentinel must not leak a negative size to clients.
+                    actual_size: -1,
+                    ..Default::default()
+                },
+            ],
+            ..Default::default()
+        };
+
+        let output = build_list_parts_output(input);
+        let parts = output.parts.as_ref().expect("parts should be present");
+
+        assert_eq!(parts.len(), 2);
+        assert_eq!(parts[0].size, Some(1024));
+        assert_eq!(parts[1].size, Some(1024));
+    }
+
     #[test]
     fn test_list_parts_output_normalizes_legacy_storage_class_and_handles_overflow_markers() {
         let input = ListPartsInfo {

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 {