feat(ecstore): expose read version stage metrics (#6073)

Record local read_version path resolution, path length check, xl.meta read, and metadata decode durations through the existing GET stage metrics channel. The new samples are gated by GET stage metrics so metrics-off reads avoid timer and recorder work.

Co-authored-by: heihutu <heihutu@gmail.com>

.config/nextest.toml

@@ -252,16 +252,10 @@ test-group = 'ecstore-serial-flaky'
 # cluster, so it keeps the lane's parallel-safe / no-external-dependency
 # properties. The RustFS warm backend has no loopback guard (that guard is
 # replication-only), so it needs no opt-in env for its 127.0.0.1 tier target.
-#
-# Disk compression (backlog#1848): the `compression` module joins the smoke
-# lane so the multipart disk-compression roundtrips (restored after
-# rustfs/rustfs#5169 disabled them) have PR-lane signal, not just merge-gate.
-# Single-node servers on random ports with isolated temp dirs — meets the
-# admission criteria unchanged.
 [profile.e2e-smoke]
 default-filter = """
     package(e2e_test) & (
-        test(/^(delete_marker_migration_semantics|version_id_regression|list_objects_v2_pagination|list_object_versions_regression|list_objects_duplicates|list_buckets_double_slash|list_buckets_auth|list_buckets_iam_filter|leading_slash_key|special_chars|create_bucket_region|delete_objects_versioning|head_object_consistency|head_object_range|copy_object_metadata|copy_object_tagging|copy_source_invalid_date|content_encoding|compression|multipart_storage_class|storage_class_capability|ssec_copy|anonymous_access|bucket_policy_check|presigned_negative|negative_sigv4|admin_auth|notification_webhook|tls_hot_reload|console_smoke|admin_iam_crud|admin_pools|sts_query_compat)_test::|^fake_s3_target::/)
+        test(/^(delete_marker_migration_semantics|version_id_regression|list_objects_v2_pagination|list_object_versions_regression|list_objects_duplicates|list_buckets_double_slash|list_buckets_auth|list_buckets_iam_filter|leading_slash_key|special_chars|create_bucket_region|delete_objects_versioning|head_object_consistency|head_object_range|copy_object_metadata|copy_object_tagging|copy_source_invalid_date|content_encoding|multipart_storage_class|storage_class_capability|ssec_copy|anonymous_access|bucket_policy_check|presigned_negative|negative_sigv4|admin_auth|notification_webhook|tls_hot_reload|console_smoke|admin_iam_crud|admin_pools|sts_query_compat)_test::|^fake_s3_target::/)
         | test(/^replication_extension_test::(test_replication_check_succeeds_with_remote_target|test_replication_check_rejects_target_without_object_lock|test_set_remote_target_rejects_unversioned_source_bucket|test_replication_check_rejects_unversioned_source_bucket|test_replication_check_rejects_missing_replication_config|test_replication_check_rejects_invalid_bucket|test_set_remote_target_rejects_same_bucket_on_same_deployment|test_set_remote_target_rejects_unversioned_target_bucket|test_set_remote_target_update_requires_arn|test_set_remote_target_update_rejects_missing_target|test_set_remote_target_rejects_invalid_target_url|test_set_remote_target_rejects_self_signed_https_target_without_skip_tls_verify|test_set_remote_target_rejects_private_ca_https_target_without_ca_cert_pem|test_list_remote_targets_rejects_empty_bucket|test_list_remote_targets_rejects_invalid_bucket|test_remove_remote_target_rejects_missing_target|test_remove_remote_target_rejects_missing_arn|test_remove_remote_target_rejects_invalid_bucket|test_remove_remote_target_rejects_target_used_by_replication|test_delete_bucket_replication_removes_remote_target)$/)
         | test(/^reliant::lifecycle::/)
         | test(/^reliant::tiering::/)

crates/e2e_test/src/common.rs

@@ -67,10 +67,7 @@ fn configured_capture_log_path(temp_dir: &str) -> Option<String> {
     capture_log_path(Path::new(&log_dir), temp_dir).map(|path| path.to_string_lossy().into_owned())
 }
 
-pub(crate) fn capture_command_logs(
-    command: &mut Command,
-    log_path: Option<&str>,
-) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+fn capture_command_logs(command: &mut Command, log_path: Option<&str>) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
     let Some(log_path) = log_path else {
         return Ok(());
     };

crates/e2e_test/src/compression_test.rs

@@ -2,7 +2,6 @@
 
 use crate::common::{RustFSTestEnvironment, init_logging, rustfs_binary_path};
 use aws_sdk_s3::primitives::ByteStream;
-use aws_sdk_s3::types::{CompletedMultipartUpload, CompletedPart};
 use serial_test::serial;
 use std::fs;
 use std::path::PathBuf;
@@ -26,15 +25,6 @@ fn generate_compressible_data(size: usize) -> Vec<u8> {
     data
 }
 
-/// Deterministic 2048-byte-period binary pattern that compresses extremely well: every part
-/// yields many compressed blocks, which is exactly the shape that reproduced the mid-payload
-/// Pending truncation (rustfs/rustfs#5957).
-fn generate_high_ratio_binary_data(size: usize, seed: u8) -> Vec<u8> {
-    (0..size)
-        .map(|i| ((i as u64).wrapping_mul(2_654_435_761).wrapping_add(seed as u64) >> 3) as u8)
-        .collect()
-}
-
 fn find_part_files(temp_dir: &str, bucket: &str, object_key: &str) -> Vec<PathBuf> {
     let bucket_path = PathBuf::from(temp_dir).join(bucket);
     let mut part_files = Vec::new();
@@ -65,14 +55,9 @@ async fn start_rustfs_with_compression(env: &mut RustFSTestEnvironment) -> Resul
     env.cleanup_existing_processes().await?;
 
     let binary_path = rustfs_binary_path();
-    // Route the child's stdout/stderr through the shared RUSTFS_E2E_LOG_DIR
-    // capture (survives the temp-dir cleanup on Drop and is uploaded as a CI
-    // artifact); without the env var the child inherits stdio as before.
-    let mut command = Command::new(&binary_path);
-    command
+    let process = Command::new(&binary_path)
         .env("RUSTFS_CONSOLE_ENABLE", "false")
         .env("RUSTFS_COMPRESSION_ENABLED", "true")
-        .env("RUSTFS_COMPRESSION_MULTIPART_ENABLED", "true")
         .args([
             "--address",
             &env.address,
@@ -81,9 +66,8 @@ async fn start_rustfs_with_compression(env: &mut RustFSTestEnvironment) -> Resul
             "--secret-key",
             &env.secret_key,
             &env.temp_dir,
-        ]);
-    crate::common::capture_command_logs(&mut command, env.capture_log_path.as_deref())?;
-    let process = command.spawn()?;
+        ])
+        .spawn()?;
 
     env.process = Some(process);
 
@@ -170,647 +154,3 @@ async fn test_compression_roundtrip() -> Result<(), Box<dyn std::error::Error +
     env.stop_server();
     Ok(())
 }
-
-const MULTIPART_COMPRESSION_BUCKET: &str = "compression-multipart-bucket";
-const MPU_PART1_SIZE: usize = 5 * 1024 * 1024;
-const MPU_PART2_SIZE: usize = 1024 * 1024;
-
-async fn multipart_upload(
-    client: &aws_sdk_s3::Client,
-    bucket: &str,
-    key: &str,
-    parts: &[&[u8]],
-) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
-    let create = client.create_multipart_upload().bucket(bucket).key(key).send().await?;
-    let upload_id = create.upload_id().ok_or("missing upload id")?.to_string();
-
-    let mut completed_parts = Vec::with_capacity(parts.len());
-    for (i, part) in parts.iter().enumerate() {
-        let part_number = (i + 1) as i32;
-        let upload = client
-            .upload_part()
-            .bucket(bucket)
-            .key(key)
-            .upload_id(&upload_id)
-            .part_number(part_number)
-            .body(ByteStream::from(part.to_vec()))
-            .send()
-            .await?;
-        completed_parts.push(
-            CompletedPart::builder()
-                .part_number(part_number)
-                .e_tag(upload.e_tag().unwrap_or_default())
-                .build(),
-        );
-    }
-
-    client
-        .complete_multipart_upload()
-        .bucket(bucket)
-        .key(key)
-        .upload_id(&upload_id)
-        .multipart_upload(CompletedMultipartUpload::builder().set_parts(Some(completed_parts)).build())
-        .send()
-        .await?;
-    Ok(())
-}
-
-async fn fetch_range(
-    client: &aws_sdk_s3::Client,
-    bucket: &str,
-    key: &str,
-    range: &str,
-) -> Result<Vec<u8>, Box<dyn std::error::Error + Send + Sync>> {
-    let response = client.get_object().bucket(bucket).key(key).range(range).send().await?;
-    Ok(response.body.collect().await?.into_bytes().to_vec())
-}
-
-/// Multipart disk compression roundtrip: parts are written as independent
-/// compressed streams and every GET shape must reassemble the original bytes
-/// (rustfs/rustfs#5957: multipart uploads previously bypassed disk compression
-/// entirely).
-#[tokio::test]
-#[serial]
-async fn test_compression_multipart_roundtrip() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
-    init_logging();
-    info!("Starting multipart compression roundtrip test");
-
-    let mut env = RustFSTestEnvironment::new().await?;
-    start_rustfs_with_compression(&mut env).await?;
-
-    let client = env.create_s3_client();
-    env.create_test_bucket(MULTIPART_COMPRESSION_BUCKET).await?;
-
-    let object_key = "multipart-compressible.txt";
-    let part1 = generate_compressible_data(MPU_PART1_SIZE);
-    let part2 = generate_compressible_data(MPU_PART2_SIZE);
-    let mut original_data = part1.clone();
-    original_data.extend_from_slice(&part2);
-    let total_size = original_data.len();
-
-    multipart_upload(&client, MULTIPART_COMPRESSION_BUCKET, object_key, &[&part1, &part2]).await?;
-
-    let head_response = client
-        .head_object()
-        .bucket(MULTIPART_COMPRESSION_BUCKET)
-        .key(object_key)
-        .send()
-        .await?;
-    assert_eq!(
-        head_response.content_length().unwrap_or(0) as usize,
-        total_size,
-        "Content-Length should be the logical object size"
-    );
-
-    let part_files = find_part_files(&env.temp_dir, MULTIPART_COMPRESSION_BUCKET, object_key);
-    assert!(!part_files.is_empty(), "expected on-disk part files for the multipart object");
-    let total_physical_size: u64 = part_files.iter().filter_map(|p| fs::metadata(p).ok()).map(|m| m.len()).sum();
-    assert!(
-        total_physical_size < (total_size / 2) as u64,
-        "Physical size {total_physical_size} should be well below original size {total_size} (multipart compression applied)"
-    );
-    info!("Multipart physical storage size: {total_physical_size} bytes (compressed from {total_size} bytes)");
-
-    // Full GET must reassemble both independently compressed parts.
-    let get_response = client
-        .get_object()
-        .bucket(MULTIPART_COMPRESSION_BUCKET)
-        .key(object_key)
-        .send()
-        .await?;
-    let downloaded = get_response.body.collect().await?.into_bytes();
-    assert_eq!(downloaded.len(), total_size);
-    assert_eq!(&downloaded[..], &original_data[..], "full GET data mismatch");
-
-    // Range fully inside part 1.
-    let range_inside_part1 = fetch_range(&client, MULTIPART_COMPRESSION_BUCKET, object_key, "bytes=1024-999423").await?;
-    assert_eq!(&range_inside_part1[..], &original_data[1024..999424], "part-1 range mismatch");
-
-    // Range crossing the part boundary.
-    let boundary_start = MPU_PART1_SIZE - 128 * 1024;
-    let boundary_end = MPU_PART1_SIZE + 128 * 1024 - 1;
-    let range_crossing = fetch_range(
-        &client,
-        MULTIPART_COMPRESSION_BUCKET,
-        object_key,
-        &format!("bytes={boundary_start}-{boundary_end}"),
-    )
-    .await?;
-    assert_eq!(
-        &range_crossing[..],
-        &original_data[boundary_start..boundary_end + 1],
-        "boundary-crossing range mismatch"
-    );
-
-    // Range fully inside part 2.
-    let part2_start = MPU_PART1_SIZE + 4096;
-    let part2_end = MPU_PART1_SIZE + 256 * 1024 - 1;
-    let range_inside_part2 = fetch_range(
-        &client,
-        MULTIPART_COMPRESSION_BUCKET,
-        object_key,
-        &format!("bytes={part2_start}-{part2_end}"),
-    )
-    .await?;
-    assert_eq!(
-        &range_inside_part2[..],
-        &original_data[part2_start..part2_end + 1],
-        "part-2 range mismatch"
-    );
-
-    // Suffix range (last 128 KiB, entirely in part 2).
-    let suffix_len = 128 * 1024;
-    let suffix = fetch_range(&client, MULTIPART_COMPRESSION_BUCKET, object_key, &format!("bytes=-{suffix_len}")).await?;
-    assert_eq!(&suffix[..], &original_data[total_size - suffix_len..], "suffix range mismatch");
-
-    // partNumber GETs must return each original part.
-    for (part_number, expected) in [(1, &part1), (2, &part2)] {
-        let response = client
-            .get_object()
-            .bucket(MULTIPART_COMPRESSION_BUCKET)
-            .key(object_key)
-            .part_number(part_number)
-            .send()
-            .await?;
-        let body = response.body.collect().await?.into_bytes();
-        assert_eq!(&body[..], &expected[..], "partNumber={part_number} GET mismatch");
-    }
-
-    info!("Multipart compression roundtrip test passed");
-    env.delete_test_bucket(MULTIPART_COMPRESSION_BUCKET).await?;
-    env.stop_server();
-    Ok(())
-}
-
-const MPU_HIGH_RATIO_BUCKET: &str = "compression-mpu-high-ratio-bucket";
-
-/// High-ratio binary multipart payload: the object key is on the compression allow-list, so the
-/// disk-compression path runs and each part is stored as many compressed blocks — the shape that
-/// reproduced the mid-payload Pending truncation (rustfs/rustfs#5957). Every GET shape must return
-/// the exact original bytes, and the stored size must show the data really was compressed.
-#[tokio::test]
-#[serial]
-async fn test_compression_multipart_high_ratio_binary_roundtrip() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
-    init_logging();
-    info!("Starting multipart high-ratio binary compression roundtrip test");
-
-    let mut env = RustFSTestEnvironment::new().await?;
-    start_rustfs_with_compression(&mut env).await?;
-
-    let client = env.create_s3_client();
-    env.create_test_bucket(MPU_HIGH_RATIO_BUCKET).await?;
-
-    let object_key = "multipart-high-ratio.txt";
-    let part1 = generate_high_ratio_binary_data(MPU_PART1_SIZE, 7);
-    let part2 = generate_high_ratio_binary_data(MPU_PART2_SIZE, 61);
-    let mut original_data = part1.clone();
-    original_data.extend_from_slice(&part2);
-    let total_size = original_data.len();
-
-    multipart_upload(&client, MPU_HIGH_RATIO_BUCKET, object_key, &[&part1, &part2]).await?;
-
-    let head_response = client
-        .head_object()
-        .bucket(MPU_HIGH_RATIO_BUCKET)
-        .key(object_key)
-        .send()
-        .await?;
-    assert_eq!(
-        head_response.content_length().unwrap_or(0) as usize,
-        total_size,
-        "Content-Length should be the logical object size"
-    );
-
-    // This pattern compresses to roughly 1/50 of its logical size, so a comfortably loose 2x
-    // margin still proves the parts were stored compressed rather than raw or double-encoded.
-    let part_files = find_part_files(&env.temp_dir, MPU_HIGH_RATIO_BUCKET, object_key);
-    assert!(!part_files.is_empty(), "expected on-disk part files for the multipart object");
-    let total_physical_size: u64 = part_files.iter().filter_map(|p| fs::metadata(p).ok()).map(|m| m.len()).sum();
-    assert!(
-        total_physical_size < (total_size as u64) / 2,
-        "Physical size {total_physical_size} should be far below the logical size {total_size} for high-ratio data"
-    );
-    info!("High-ratio multipart physical storage size: {total_physical_size} bytes (logical {total_size} bytes)");
-
-    info!("step: full GET");
-    let get_response = client
-        .get_object()
-        .bucket(MPU_HIGH_RATIO_BUCKET)
-        .key(object_key)
-        .send()
-        .await?;
-    let downloaded = get_response.body.collect().await?.into_bytes();
-    assert_eq!(downloaded.len(), total_size);
-    assert_eq!(&downloaded[..], &original_data[..], "full GET data mismatch");
-
-    // Range crossing the part boundary.
-    info!("step: boundary range GET");
-    let boundary_start = MPU_PART1_SIZE - 128 * 1024;
-    let boundary_end = MPU_PART1_SIZE + 128 * 1024 - 1;
-    let range_crossing = fetch_range(
-        &client,
-        MPU_HIGH_RATIO_BUCKET,
-        object_key,
-        &format!("bytes={boundary_start}-{boundary_end}"),
-    )
-    .await?;
-    assert_eq!(
-        &range_crossing[..],
-        &original_data[boundary_start..boundary_end + 1],
-        "boundary-crossing range mismatch"
-    );
-
-    // partNumber GET for the trailing part.
-    info!("step: partNumber GET");
-    let part2_response = client
-        .get_object()
-        .bucket(MPU_HIGH_RATIO_BUCKET)
-        .key(object_key)
-        .part_number(2)
-        .send()
-        .await?;
-    let part2_body = part2_response.body.collect().await?.into_bytes();
-    assert_eq!(&part2_body[..], &part2[..], "partNumber=2 GET mismatch");
-
-    info!("Multipart high-ratio binary compression roundtrip test passed");
-    env.delete_test_bucket(MPU_HIGH_RATIO_BUCKET).await?;
-    env.stop_server();
-    Ok(())
-}
-
-const MPU_COPY_COMPRESSION_BUCKET: &str = "compression-mpu-copy-bucket";
-const MPU_COPY_SOURCE_SIZE: usize = 6 * 1024 * 1024;
-const MPU_COPY_RANGE_LEN: usize = 5 * 1024 * 1024;
-
-/// UploadPartCopy feeds a part from an already stored (and already compressed) object. The copied
-/// range must be decompressed on read and re-compressed into the destination part, so the final
-/// object has to match "source prefix + uploaded tail" byte for byte.
-#[tokio::test]
-#[serial]
-async fn test_compression_multipart_upload_part_copy_roundtrip() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
-    init_logging();
-    info!("Starting multipart upload-part-copy compression roundtrip test");
-
-    let mut env = RustFSTestEnvironment::new().await?;
-    start_rustfs_with_compression(&mut env).await?;
-
-    let client = env.create_s3_client();
-    env.create_test_bucket(MPU_COPY_COMPRESSION_BUCKET).await?;
-
-    // Source object: a plain PUT that goes through the single-stream compression path.
-    let source_key = "copy-source.txt";
-    let source_data = generate_compressible_data(MPU_COPY_SOURCE_SIZE);
-    client
-        .put_object()
-        .bucket(MPU_COPY_COMPRESSION_BUCKET)
-        .key(source_key)
-        .body(ByteStream::from(source_data.clone()))
-        .send()
-        .await?;
-
-    // Destination object: part 1 copied from the source, part 2 uploaded directly.
-    let target_key = "copy-target.txt";
-    let part2 = generate_compressible_data(MPU_PART2_SIZE);
-    let mut expected_data = source_data[..MPU_COPY_RANGE_LEN].to_vec();
-    expected_data.extend_from_slice(&part2);
-    let total_size = expected_data.len();
-
-    let create = client
-        .create_multipart_upload()
-        .bucket(MPU_COPY_COMPRESSION_BUCKET)
-        .key(target_key)
-        .send()
-        .await?;
-    let upload_id = create.upload_id().ok_or("missing upload id")?.to_string();
-
-    let copy_part = client
-        .upload_part_copy()
-        .bucket(MPU_COPY_COMPRESSION_BUCKET)
-        .key(target_key)
-        .upload_id(&upload_id)
-        .part_number(1)
-        .copy_source(format!("{MPU_COPY_COMPRESSION_BUCKET}/{source_key}"))
-        .copy_source_range(format!("bytes=0-{}", MPU_COPY_RANGE_LEN - 1))
-        .send()
-        .await?;
-    let copy_etag = copy_part
-        .copy_part_result()
-        .and_then(|r| r.e_tag())
-        .ok_or("missing copy part etag")?
-        .to_string();
-
-    let uploaded_part = client
-        .upload_part()
-        .bucket(MPU_COPY_COMPRESSION_BUCKET)
-        .key(target_key)
-        .upload_id(&upload_id)
-        .part_number(2)
-        .body(ByteStream::from(part2.clone()))
-        .send()
-        .await?;
-
-    client
-        .complete_multipart_upload()
-        .bucket(MPU_COPY_COMPRESSION_BUCKET)
-        .key(target_key)
-        .upload_id(&upload_id)
-        .multipart_upload(
-            CompletedMultipartUpload::builder()
-                .parts(CompletedPart::builder().part_number(1).e_tag(copy_etag).build())
-                .parts(
-                    CompletedPart::builder()
-                        .part_number(2)
-                        .e_tag(uploaded_part.e_tag().unwrap_or_default())
-                        .build(),
-                )
-                .build(),
-        )
-        .send()
-        .await?;
-
-    let head_response = client
-        .head_object()
-        .bucket(MPU_COPY_COMPRESSION_BUCKET)
-        .key(target_key)
-        .send()
-        .await?;
-    assert_eq!(
-        head_response.content_length().unwrap_or(0) as usize,
-        total_size,
-        "Content-Length should be the logical object size"
-    );
-
-    let part_files = find_part_files(&env.temp_dir, MPU_COPY_COMPRESSION_BUCKET, target_key);
-    assert!(!part_files.is_empty(), "expected on-disk part files for the copied object");
-    let total_physical_size: u64 = part_files.iter().filter_map(|p| fs::metadata(p).ok()).map(|m| m.len()).sum();
-    assert!(
-        total_physical_size < (total_size / 2) as u64,
-        "Physical size {total_physical_size} should be well below original size {total_size} (copied part compression applied)"
-    );
-
-    let get_response = client
-        .get_object()
-        .bucket(MPU_COPY_COMPRESSION_BUCKET)
-        .key(target_key)
-        .send()
-        .await?;
-    let downloaded = get_response.body.collect().await?.into_bytes();
-    assert_eq!(downloaded.len(), total_size);
-    assert_eq!(&downloaded[..], &expected_data[..], "copied multipart GET data mismatch");
-
-    info!("Multipart upload-part-copy compression roundtrip test passed");
-    env.delete_test_bucket(MPU_COPY_COMPRESSION_BUCKET).await?;
-    env.stop_server();
-    Ok(())
-}
-
-const MPU_THREE_PARTS_BUCKET: &str = "compression-mpu-three-parts-bucket";
-const MPU_THREE_PARTS_TAIL_SIZE: usize = 512 * 1024;
-
-/// Three-part upload with uneven part sizes: each partNumber GET must map back to exactly one
-/// compressed part stream, and a suffix range must resolve inside the trailing part.
-#[tokio::test]
-#[serial]
-async fn test_compression_multipart_three_parts_part_number_gets() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
-    init_logging();
-    info!("Starting three-part multipart compression partNumber test");
-
-    let mut env = RustFSTestEnvironment::new().await?;
-    start_rustfs_with_compression(&mut env).await?;
-
-    let client = env.create_s3_client();
-    env.create_test_bucket(MPU_THREE_PARTS_BUCKET).await?;
-
-    let object_key = "multipart-three-parts.txt";
-    let part1 = generate_compressible_data(MPU_PART1_SIZE);
-    let part2 = generate_compressible_data(MPU_PART1_SIZE);
-    let part3 = generate_compressible_data(MPU_THREE_PARTS_TAIL_SIZE);
-    let mut original_data = part1.clone();
-    original_data.extend_from_slice(&part2);
-    original_data.extend_from_slice(&part3);
-    let total_size = original_data.len();
-
-    multipart_upload(&client, MPU_THREE_PARTS_BUCKET, object_key, &[&part1, &part2, &part3]).await?;
-
-    let head_response = client
-        .head_object()
-        .bucket(MPU_THREE_PARTS_BUCKET)
-        .key(object_key)
-        .send()
-        .await?;
-    assert_eq!(
-        head_response.content_length().unwrap_or(0) as usize,
-        total_size,
-        "Content-Length should be the logical object size"
-    );
-
-    let part_files = find_part_files(&env.temp_dir, MPU_THREE_PARTS_BUCKET, object_key);
-    assert!(!part_files.is_empty(), "expected on-disk part files for the multipart object");
-    let total_physical_size: u64 = part_files.iter().filter_map(|p| fs::metadata(p).ok()).map(|m| m.len()).sum();
-    assert!(
-        total_physical_size < (total_size / 2) as u64,
-        "Physical size {total_physical_size} should be well below original size {total_size} (multipart compression applied)"
-    );
-
-    // Every partNumber GET must return exactly the bytes of the corresponding uploaded part.
-    for (part_number, expected) in [(1, &part1), (2, &part2), (3, &part3)] {
-        let response = client
-            .get_object()
-            .bucket(MPU_THREE_PARTS_BUCKET)
-            .key(object_key)
-            .part_number(part_number)
-            .send()
-            .await?;
-        let body = response.body.collect().await?.into_bytes();
-        assert_eq!(&body[..], &expected[..], "partNumber={part_number} GET mismatch");
-    }
-
-    // Suffix range (last 64 KiB) resolves inside the trailing part.
-    let suffix_len = 64 * 1024;
-    let suffix = fetch_range(&client, MPU_THREE_PARTS_BUCKET, object_key, &format!("bytes=-{suffix_len}")).await?;
-    assert_eq!(&suffix[..], &original_data[total_size - suffix_len..], "suffix range mismatch");
-
-    info!("Three-part multipart compression partNumber test passed");
-    env.delete_test_bucket(MPU_THREE_PARTS_BUCKET).await?;
-    env.stop_server();
-    Ok(())
-}
-
-const MPU_SSE_COMPRESSION_BUCKET: &str = "compression-mpu-sse-bucket";
-
-async fn start_rustfs_with_compression_and_sse(
-    env: &mut RustFSTestEnvironment,
-) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
-    use base64::Engine;
-    env.cleanup_existing_processes().await?;
-
-    let binary_path = rustfs_binary_path();
-    let master_key = base64::engine::general_purpose::STANDARD.encode([0x42u8; 32]);
-    // Server output goes to a file inside the per-test temp dir so a failing
-    // run can be diagnosed from the child's logs.
-    let server_log = std::fs::File::create(format!("{}/server.log", env.temp_dir))?;
-    let server_log_err = server_log.try_clone()?;
-    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))
-        .stderr(std::process::Stdio::from(server_log_err))
-        .args([
-            "--address",
-            &env.address,
-            "--access-key",
-            &env.access_key,
-            "--secret-key",
-            &env.secret_key,
-            &env.temp_dir,
-        ])
-        .spawn()?;
-
-    env.process = Some(process);
-
-    info!("Waiting for RustFS server with compression + SSE-S3 enabled on {}", env.address);
-    for i in 0..30 {
-        if TcpStream::connect(&env.address).await.is_ok() {
-            info!("RustFS server is ready after {} attempts", i + 1);
-            return Ok(());
-        }
-        if i == 29 {
-            return Err("RustFS server failed to become ready".into());
-        }
-        sleep(Duration::from_secs(1)).await;
-    }
-    Ok(())
-}
-
-/// SSE-S3 + disk compression multipart: each part is compressed and then encrypted, and every GET
-/// shape must still return the original plaintext bytes. Physical size must shrink because the
-/// compression runs before encryption.
-#[tokio::test]
-#[serial]
-async fn test_compression_multipart_sse_s3_roundtrip() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
-    use aws_sdk_s3::types::ServerSideEncryption;
-
-    init_logging();
-    info!("Starting SSE-S3 multipart compression roundtrip test");
-
-    let mut env = RustFSTestEnvironment::new().await?;
-    start_rustfs_with_compression_and_sse(&mut env).await?;
-
-    let client = env.create_s3_client();
-    env.create_test_bucket(MPU_SSE_COMPRESSION_BUCKET).await?;
-
-    let object_key = "multipart-sse-compressible.txt";
-    let part1 = generate_compressible_data(MPU_PART1_SIZE);
-    let part2 = generate_compressible_data(MPU_PART2_SIZE);
-    let mut original_data = part1.clone();
-    original_data.extend_from_slice(&part2);
-    let total_size = original_data.len();
-
-    let create = client
-        .create_multipart_upload()
-        .bucket(MPU_SSE_COMPRESSION_BUCKET)
-        .key(object_key)
-        .server_side_encryption(ServerSideEncryption::Aes256)
-        .send()
-        .await?;
-    let upload_id = create.upload_id().ok_or("missing upload id")?.to_string();
-
-    let mut completed_parts = Vec::new();
-    for (i, part) in [&part1, &part2].into_iter().enumerate() {
-        let part_number = (i + 1) as i32;
-        let upload = client
-            .upload_part()
-            .bucket(MPU_SSE_COMPRESSION_BUCKET)
-            .key(object_key)
-            .upload_id(&upload_id)
-            .part_number(part_number)
-            .body(ByteStream::from(part.clone()))
-            .send()
-            .await?;
-        completed_parts.push(
-            CompletedPart::builder()
-                .part_number(part_number)
-                .e_tag(upload.e_tag().unwrap_or_default())
-                .build(),
-        );
-    }
-
-    client
-        .complete_multipart_upload()
-        .bucket(MPU_SSE_COMPRESSION_BUCKET)
-        .key(object_key)
-        .upload_id(&upload_id)
-        .multipart_upload(CompletedMultipartUpload::builder().set_parts(Some(completed_parts)).build())
-        .send()
-        .await?;
-
-    let head_response = client
-        .head_object()
-        .bucket(MPU_SSE_COMPRESSION_BUCKET)
-        .key(object_key)
-        .send()
-        .await?;
-    assert_eq!(
-        head_response.content_length().unwrap_or(0) as usize,
-        total_size,
-        "Content-Length should be the logical object size"
-    );
-    assert_eq!(
-        head_response.server_side_encryption(),
-        Some(&ServerSideEncryption::Aes256),
-        "HEAD must report SSE-S3"
-    );
-
-    let part_files = find_part_files(&env.temp_dir, MPU_SSE_COMPRESSION_BUCKET, object_key);
-    assert!(!part_files.is_empty(), "expected on-disk part files for the multipart object");
-    let total_physical_size: u64 = part_files.iter().filter_map(|p| fs::metadata(p).ok()).map(|m| m.len()).sum();
-    assert!(
-        total_physical_size < (total_size / 2) as u64,
-        "Physical size {total_physical_size} should be well below original size {total_size} (compress-then-encrypt applied)"
-    );
-
-    let get_response = client
-        .get_object()
-        .bucket(MPU_SSE_COMPRESSION_BUCKET)
-        .key(object_key)
-        .send()
-        .await?;
-    let downloaded = get_response.body.collect().await?.into_bytes();
-    assert_eq!(downloaded.len(), total_size);
-    assert_eq!(&downloaded[..], &original_data[..], "SSE-S3 multipart full GET data mismatch");
-
-    // Range crossing the part boundary must decrypt and decompress across parts.
-    let boundary_start = MPU_PART1_SIZE - 64 * 1024;
-    let boundary_end = MPU_PART1_SIZE + 64 * 1024 - 1;
-    let range_crossing = fetch_range(
-        &client,
-        MPU_SSE_COMPRESSION_BUCKET,
-        object_key,
-        &format!("bytes={boundary_start}-{boundary_end}"),
-    )
-    .await?;
-    assert_eq!(
-        &range_crossing[..],
-        &original_data[boundary_start..boundary_end + 1],
-        "SSE-S3 boundary-crossing range mismatch"
-    );
-
-    // partNumber GET for the trailing part.
-    let part2_response = client
-        .get_object()
-        .bucket(MPU_SSE_COMPRESSION_BUCKET)
-        .key(object_key)
-        .part_number(2)
-        .send()
-        .await?;
-    let part2_body = part2_response.body.collect().await?.into_bytes();
-    assert_eq!(&part2_body[..], &part2[..], "SSE-S3 partNumber=2 GET mismatch");
-
-    info!("SSE-S3 multipart compression roundtrip test passed");
-    env.delete_test_bucket(MPU_SSE_COMPRESSION_BUCKET).await?;
-    env.stop_server();
-    Ok(())
-}

crates/e2e_test/src/inline_fast_path_cluster_test.rs

@@ -1828,36 +1828,33 @@ async fn four_node_compressed_inline_fallback() -> TestResult {
     Ok(())
 }
 
-/// Multipart disk compression is live again, so a compression-enabled cluster classifies multipart objects as compressed and the roundtrip (full GET plus partNumber GET) must still return the original bytes.
-/// Reverting the multipart compression fix must fail this test.
 #[tokio::test]
 #[serial]
-async fn four_node_multipart_disk_compression_roundtrip() -> TestResult {
+async fn four_node_multipart_ignores_disk_compression_fallback() -> TestResult {
     init_logging();
 
     let collector = OtlpMetricCollector::start().await?;
     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";
+    let bucket = "inline-multipart-compression-fallback";
     cluster.create_test_bucket(bucket).await?;
     let client = cluster.create_s3_client(0)?;
-    let key = "multipart/compressed.txt";
+    let key = "multipart/compression-disabled.txt";
     let (body, second_part, etag) = put_two_part_multipart(&client, bucket, key).await?;
 
     assert_reader_path(
         &collector,
         &client,
-        ReaderPathExpectation::for_class(ReaderObject::new(bucket, key, &body, etag.as_deref(), None), LEGACY_DUPLEX, COMPRESSED),
+        ReaderPathExpectation::for_class(ReaderObject::new(bucket, key, &body, etag.as_deref(), None), LEGACY_DUPLEX, MULTIPART),
     )
     .await?;
     assert_part_number_reader_path(
         &collector,
         &client,
-        PartNumberReaderPathExpectation::new(bucket, key, &second_part, body.len(), COMPRESSED, LEGACY_DUPLEX),
+        PartNumberReaderPathExpectation::new(bucket, key, &second_part, body.len(), MULTIPART, LEGACY_DUPLEX),
     )
     .await?;
 
@@ -1874,7 +1871,6 @@ 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?;
 
@@ -1894,21 +1890,14 @@ async fn four_node_mixed_msgpack_compat_mode_preserves_fallback_controls() -> Te
         ReaderPathExpectation::for_class(
             ReaderObject::new(bucket, multipart_key, &multipart_body, multipart_etag.as_deref(), None),
             LEGACY_DUPLEX,
-            COMPRESSED,
+            MULTIPART,
         ),
     )
     .await?;
     assert_part_number_reader_path(
         &collector,
         &client,
-        PartNumberReaderPathExpectation::new(
-            bucket,
-            multipart_key,
-            &second_part,
-            multipart_body.len(),
-            COMPRESSED,
-            LEGACY_DUPLEX,
-        ),
+        PartNumberReaderPathExpectation::new(bucket, multipart_key, &second_part, multipart_body.len(), MULTIPART, LEGACY_DUPLEX),
     )
     .await?;
     assert_msgpack_decode_observed(&collector, &decode_before).await?;
@@ -2364,11 +2353,7 @@ async fn four_node_mixed_msgpack_compat_mode_preserves_fallback_controls_during_
     hot_client.create_bucket().bucket(bucket).send().await?;
     put_lifecycle_with_transition_retry(&hot_client, bucket, &tier_name).await?;
 
-    // `.zip` sits on the disk-compression exclusion list: this test pins
-    // msgpack compat controls across ILM transition, and a compressed object
-    // would classify as `compressed` instead of `remote` (and the warm-tier
-    // read path does not decode compression — tracked separately).
-    let key = "transition/mixed-multipart.zip";
+    let key = "transition/mixed-multipart.bin";
     let (body, second_part, etag) = put_two_part_multipart(&hot_client, bucket, key).await?;
     wait_for_transition(&hot_client, bucket, key, &tier_name).await?;
     assert!(

crates/ecstore/src/api/mod.rs

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

crates/ecstore/src/diagnostics/get.rs

@@ -84,6 +84,10 @@ pub(crate) const GET_STAGE_READER_STREAM_FIRST_READ: &str = "reader_stream_first
 pub(crate) const GET_STAGE_READER_TASK_BITROT_READER_INIT: &str = "reader_task_bitrot_reader_init";
 pub(crate) const GET_STAGE_READER_TASK_FILE_OPEN: &str = "reader_task_file_open";
 pub(crate) const GET_STAGE_READER_TASK_READER_CONSTRUCTION: &str = "reader_task_reader_construction";
+pub(crate) const GET_STAGE_READ_VERSION_DECODE: &str = "read_version_decode";
+pub(crate) const GET_STAGE_READ_VERSION_PATH_CHECK: &str = "read_version_path_check";
+pub(crate) const GET_STAGE_READ_VERSION_PATH_RESOLVE: &str = "read_version_path_resolve";
+pub(crate) const GET_STAGE_READ_VERSION_XLMETA_READ: &str = "read_version_xlmeta_read";
 pub(crate) const GET_STAGE_RECONSTRUCT: &str = "reconstruct";
 pub(crate) const GET_STAGE_RESPONSE_HANDOFF: &str = "response_handoff";
 pub(crate) const GET_STAGE_SLOWEST_METADATA_RESPONSE: &str = "slowest_metadata_response";
@@ -442,6 +446,10 @@ mod tests {
         assert_eq!(GET_STAGE_QUORUM_REACHED, "quorum_reached");
         assert_eq!(GET_STAGE_RANGE, "range");
         assert_eq!(GET_STAGE_READER_SETUP, "reader_setup");
+        assert_eq!(GET_STAGE_READ_VERSION_DECODE, "read_version_decode");
+        assert_eq!(GET_STAGE_READ_VERSION_PATH_CHECK, "read_version_path_check");
+        assert_eq!(GET_STAGE_READ_VERSION_PATH_RESOLVE, "read_version_path_resolve");
+        assert_eq!(GET_STAGE_READ_VERSION_XLMETA_READ, "read_version_xlmeta_read");
         assert_eq!(GET_STAGE_RECONSTRUCT, "reconstruct");
         assert_eq!(GET_STAGE_RESPONSE_HANDOFF, "response_handoff");
         assert_eq!(GET_STAGE_SLOWEST_METADATA_RESPONSE, "slowest_metadata_response");

crates/ecstore/src/disk/local.rs

@@ -15,6 +15,11 @@
 use crate::config::storageclass::DEFAULT_INLINE_BLOCK;
 use crate::crash_inject::{self, CrashPoint};
 use crate::data_usage::local_snapshot::ensure_data_usage_layout;
+use crate::diagnostics::get::{
+    GET_OBJECT_PATH_INTERNAL_META, GET_OBJECT_PATH_LEGACY_DUPLEX, GET_STAGE_READ_VERSION_DECODE,
+    GET_STAGE_READ_VERSION_PATH_CHECK, GET_STAGE_READ_VERSION_PATH_RESOLVE, GET_STAGE_READ_VERSION_XLMETA_READ,
+    get_stage_timer_if_enabled, record_get_stage_duration_if_enabled,
+};
 #[cfg(test)]
 use crate::disk::HEALING_MARKER_PATH;
 use crate::disk::disk_store::{get_drive_walkdir_stall_timeout, get_object_disk_read_timeout};
@@ -9840,6 +9845,12 @@ impl DiskAPI for LocalDisk {
         opts: &ReadOptions,
     ) -> Result<FileInfo> {
         crate::hp_guard!("LocalDisk::read_version");
+        let stage_metrics_enabled = rustfs_io_metrics::get_stage_metrics_enabled();
+        let metrics_path = if stage_metrics_enabled && crate::bucket::utils::is_meta_bucketname(volume) {
+            GET_OBJECT_PATH_INTERNAL_META
+        } else {
+            GET_OBJECT_PATH_LEGACY_DUPLEX
+        };
         if !org_volume.is_empty() {
             let org_volume_path = self.io_get_bucket_path(org_volume)?;
             if !skip_access_checks(org_volume) {
@@ -9849,37 +9860,46 @@ impl DiskAPI for LocalDisk {
             }
         }
 
+        let path_resolve_start = get_stage_timer_if_enabled(stage_metrics_enabled);
         let file_path = self.io_get_object_path(volume, path)?;
         let volume_dir = self.io_get_bucket_path(volume)?;
+        record_get_stage_duration_if_enabled(metrics_path, GET_STAGE_READ_VERSION_PATH_RESOLVE, path_resolve_start);
 
+        let path_check_start = get_stage_timer_if_enabled(stage_metrics_enabled);
         check_path_length(file_path.to_string_lossy().as_ref())?;
+        record_get_stage_duration_if_enabled(metrics_path, GET_STAGE_READ_VERSION_PATH_CHECK, path_check_start);
 
         let read_data = opts.read_data;
 
-        let (data, _) = self
-            .read_raw(volume, volume_dir.clone(), file_path, read_data)
-            .await
-            .map_err(|e| {
-                if e == DiskError::FileNotFound && !version_id.is_empty() {
-                    DiskError::FileVersionNotFound
-                } else {
-                    e
-                }
-            })?;
+        let xlmeta_read_start = get_stage_timer_if_enabled(stage_metrics_enabled);
+        let raw_read_result = self.read_raw(volume, volume_dir.clone(), file_path, read_data).await;
+        record_get_stage_duration_if_enabled(metrics_path, GET_STAGE_READ_VERSION_XLMETA_READ, xlmeta_read_start);
+        let (data, _) = raw_read_result.map_err(|e| {
+            if e == DiskError::FileNotFound && !version_id.is_empty() {
+                DiskError::FileVersionNotFound
+            } else {
+                e
+            }
+        })?;
 
-        let mut fi = get_file_info(
-            &data,
-            volume,
-            path,
-            version_id,
-            FileInfoOpts {
-                data: read_data,
-                include_free_versions: opts.incl_free_versions,
-                include_part_checksums: false,
-            },
-        )?;
-
-        fi.validate_for_metadata_read()?;
+        let decode_start = get_stage_timer_if_enabled(stage_metrics_enabled);
+        let file_info_result: Result<FileInfo> = (|| {
+            let fi = get_file_info(
+                &data,
+                volume,
+                path,
+                version_id,
+                FileInfoOpts {
+                    data: read_data,
+                    include_free_versions: opts.incl_free_versions,
+                    include_part_checksums: false,
+                },
+            )?;
+            fi.validate_for_metadata_read()?;
+            Ok(fi)
+        })();
+        record_get_stage_duration_if_enabled(metrics_path, GET_STAGE_READ_VERSION_DECODE, decode_start);
+        let mut fi = file_info_result?;
         if fi.is_canonical_delete_marker() {
             return Ok(fi);
         }
@@ -10562,6 +10582,108 @@ mod test {
         meta.marshal_msg().expect("test metadata should encode")
     }
 
+    #[test]
+    #[serial_test::serial]
+    fn read_version_records_local_metadata_stage_breakdown() {
+        let runtime = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .expect("test runtime should be created");
+        let recorder = crate::test_metrics::CapturingRecorder::default();
+        let previous_gate = rustfs_io_metrics::get_stage_metrics_enabled();
+        rustfs_io_metrics::set_get_stage_metrics_enabled(true);
+
+        metrics::with_local_recorder(&recorder, || {
+            runtime.block_on(async {
+                let dir = tempfile::tempdir().expect("temp dir should be created");
+                let endpoint =
+                    Endpoint::try_from(dir.path().to_str().expect("temp dir should be utf8")).expect("endpoint should parse");
+                let disk = LocalDisk::new(&endpoint, false).await.expect("local disk should be created");
+                let bucket = "bucket";
+                let object = "stage-breakdown";
+                ensure_test_volume(&disk, bucket).await;
+
+                let object_dir = dir.path().join(bucket).join(object);
+                fs::create_dir_all(&object_dir)
+                    .await
+                    .expect("object directory should be created");
+                fs::write(
+                    object_dir.join(STORAGE_FORMAT_FILE),
+                    test_meta(test_file_info(object, Uuid::new_v4(), None, Some(Bytes::from_static(b"inline")))),
+                )
+                .await
+                .expect("object metadata should be written");
+
+                disk.read_version(
+                    "",
+                    bucket,
+                    object,
+                    "",
+                    &ReadOptions {
+                        read_data: true,
+                        ..Default::default()
+                    },
+                )
+                .await
+                .expect("read_version should succeed");
+
+                let meta_object = "stage-breakdown-meta";
+                let meta_object_dir = dir.path().join(RUSTFS_META_BUCKET).join(meta_object);
+                fs::create_dir_all(&meta_object_dir)
+                    .await
+                    .expect("internal metadata object directory should be created");
+                fs::write(
+                    meta_object_dir.join(STORAGE_FORMAT_FILE),
+                    test_meta(test_file_info(meta_object, Uuid::new_v4(), None, Some(Bytes::from_static(b"meta")))),
+                )
+                .await
+                .expect("internal metadata should be written");
+
+                disk.read_version(
+                    "",
+                    RUSTFS_META_BUCKET,
+                    meta_object,
+                    "",
+                    &ReadOptions {
+                        read_data: true,
+                        ..Default::default()
+                    },
+                )
+                .await
+                .expect("internal metadata read_version should succeed");
+            });
+        });
+        rustfs_io_metrics::set_get_stage_metrics_enabled(previous_gate);
+
+        for stage in [
+            GET_STAGE_READ_VERSION_PATH_RESOLVE,
+            GET_STAGE_READ_VERSION_PATH_CHECK,
+            GET_STAGE_READ_VERSION_XLMETA_READ,
+            GET_STAGE_READ_VERSION_DECODE,
+        ] {
+            assert_eq!(
+                recorder
+                    .histogram_values(
+                        "rustfs_io_get_object_stage_duration_seconds",
+                        &[("path", GET_OBJECT_PATH_LEGACY_DUPLEX), ("stage", stage)]
+                    )
+                    .len(),
+                1,
+                "{stage} should be recorded once for user-bucket LocalDisk::read_version"
+            );
+            assert_eq!(
+                recorder
+                    .histogram_values(
+                        "rustfs_io_get_object_stage_duration_seconds",
+                        &[("path", GET_OBJECT_PATH_INTERNAL_META), ("stage", stage)]
+                    )
+                    .len(),
+                1,
+                "{stage} should be recorded once for internal-meta LocalDisk::read_version"
+            );
+        }
+    }
+
     #[test]
     fn inline_metadata_rollback_dir_avoids_real_data_dir_collision() {
         let target_version = Uuid::parse_str("11111111-2222-3333-4444-555555555555").expect("version id should parse");

crates/ecstore/src/io_support/compress.rs

@@ -31,13 +31,6 @@ 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.
-// RUSTFS_COMPAT_TODO(multipart-compression-default-off-window): staged rollout switch for restored multipart compression, flipping the default to enabled on retirement. Remove after the minimum supported direct-upgrade release ships the resumable DecompressReader.
-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";
 
@@ -178,21 +171,6 @@ 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 {

crates/ecstore/src/object_api/readers.rs

@@ -1665,423 +1665,6 @@ mod tests {
         assert_eq!(actual, b"fghijkl");
     }
 
-    /// Compresses one multipart part exactly like the write path does
-    /// (`WritePlan::with_compression` wraps each part in its own
-    /// `compression_reader`), returning the on-disk bytes and the storage-format
-    /// compression index.
-    async fn compressed_part_fixture(data: &[u8]) -> (Vec<u8>, Option<Bytes>) {
-        use crate::io_support::rio::TryGetIndex as _;
-        let mut compressor =
-            crate::io_support::rio::compression_reader(Cursor::new(data.to_vec()), CompressionAlgorithm::default(), false);
-        let mut compressed = Vec::new();
-        compressor.read_to_end(&mut compressed).await.expect("compress part stream");
-        let index = compressor
-            .try_get_index()
-            .map(crate::io_support::rio::compression_index_storage_bytes);
-        (compressed, index)
-    }
-
-    struct CompressedMultipartFixture {
-        object_info: ObjectInfo,
-        stored: Vec<u8>,
-        plaintext: Vec<u8>,
-    }
-
-    /// Builds the on-disk representation of a compressed multipart object: each
-    /// part is an independent compressed stream and the storage layer serves
-    /// their concatenation.
-    async fn compressed_multipart_fixture(part_sizes: &[usize]) -> CompressedMultipartFixture {
-        let pattern = b"compressed multipart read path fixture data ";
-        let mut plaintext = Vec::new();
-        let mut stored = Vec::new();
-        let mut parts = Vec::with_capacity(part_sizes.len());
-
-        for (i, part_size) in part_sizes.iter().enumerate() {
-            let mut part_plaintext = Vec::with_capacity(*part_size);
-            while part_plaintext.len() < *part_size {
-                part_plaintext.extend_from_slice(pattern);
-                part_plaintext.push(i as u8);
-            }
-            part_plaintext.truncate(*part_size);
-
-            let (compressed, index) = compressed_part_fixture(&part_plaintext).await;
-            parts.push(ObjectPartInfo {
-                number: i + 1,
-                size: compressed.len(),
-                actual_size: *part_size as i64,
-                index,
-                ..Default::default()
-            });
-            stored.extend_from_slice(&compressed);
-            plaintext.extend_from_slice(&part_plaintext);
-        }
-
-        let mut user_defined = HashMap::new();
-        rustfs_utils::http::insert_str(
-            &mut user_defined,
-            rustfs_utils::http::SUFFIX_COMPRESSION,
-            crate::io_support::rio::compression_metadata_value(CompressionAlgorithm::default()),
-        );
-        rustfs_utils::http::insert_str(&mut user_defined, rustfs_utils::http::SUFFIX_ACTUAL_SIZE, plaintext.len().to_string());
-
-        let object_info = ObjectInfo {
-            bucket: "test-bucket".to_string(),
-            name: "compressed-multipart".to_string(),
-            size: stored.len() as i64,
-            etag: Some(format!("6bcf86bed8807b8e78f0fc6e0a53079d-{}", part_sizes.len())),
-            parts: Arc::new(parts),
-            user_defined: Arc::new(user_defined),
-            ..Default::default()
-        };
-
-        CompressedMultipartFixture {
-            object_info,
-            stored,
-            plaintext,
-        }
-    }
-
-    /// Plans the read once to learn the storage window, then serves exactly that
-    /// window — mirroring how `set_disk` feeds the erasure read into the
-    /// returned reader.
-    async fn read_compressed_multipart(
-        fixture: &CompressedMultipartFixture,
-        rs: Option<HTTPRangeSpec>,
-        opts: &ObjectOptions,
-    ) -> Vec<u8> {
-        let headers = HeaderMap::new();
-        let (_, offset, length) =
-            GetObjectReader::new(Box::new(Cursor::new(Vec::new())), rs.clone(), &fixture.object_info, opts, &headers)
-                .await
-                .expect("plan compressed multipart read");
-
-        let end = offset + usize::try_from(length).expect("storage window length must be non-negative");
-        assert!(
-            end <= fixture.stored.len(),
-            "planned storage window {offset}..{end} exceeds stored stream of {} bytes",
-            fixture.stored.len()
-        );
-        let window = fixture.stored[offset..end].to_vec();
-
-        let (mut reader, replay_offset, replay_length) =
-            GetObjectReader::new(Box::new(Cursor::new(window)), rs, &fixture.object_info, opts, &headers)
-                .await
-                .expect("build compressed multipart reader");
-        assert_eq!((replay_offset, replay_length), (offset, length), "read plan must be deterministic");
-
-        reader.read_all().await.expect("read compressed multipart stream")
-    }
-
-    /// Byte pattern with a 2 KiB period: it compresses extremely well while
-    /// looking nothing like ASCII fixtures. Mirrors the e2e generator that
-    /// exposed a truncated full GET on high-ratio multipart payloads.
-    fn high_ratio_binary_payload(size: usize, seed: u8) -> Vec<u8> {
-        (0..size)
-            .map(|i| ((i as u64).wrapping_mul(2_654_435_761).wrapping_add(seed as u64) >> 3) as u8)
-            .collect()
-    }
-
-    #[tokio::test]
-    async fn compressed_multipart_full_get_handles_high_ratio_binary_payload() {
-        let part_sizes = [5 * 1024 * 1024_usize, 1024 * 1024];
-        let mut plaintext = Vec::new();
-        let mut stored = Vec::new();
-        let mut parts = Vec::with_capacity(part_sizes.len());
-
-        for (i, part_size) in part_sizes.iter().enumerate() {
-            let part_plaintext = high_ratio_binary_payload(*part_size, if i == 0 { 7 } else { 61 });
-            let (compressed, index) = compressed_part_fixture(&part_plaintext).await;
-            parts.push(ObjectPartInfo {
-                number: i + 1,
-                size: compressed.len(),
-                actual_size: *part_size as i64,
-                index,
-                ..Default::default()
-            });
-            stored.extend_from_slice(&compressed);
-            plaintext.extend_from_slice(&part_plaintext);
-        }
-
-        let mut user_defined = HashMap::new();
-        rustfs_utils::http::insert_str(
-            &mut user_defined,
-            rustfs_utils::http::SUFFIX_COMPRESSION,
-            crate::io_support::rio::compression_metadata_value(CompressionAlgorithm::default()),
-        );
-        rustfs_utils::http::insert_str(&mut user_defined, rustfs_utils::http::SUFFIX_ACTUAL_SIZE, plaintext.len().to_string());
-        let fixture = CompressedMultipartFixture {
-            object_info: ObjectInfo {
-                bucket: "test-bucket".to_string(),
-                name: "high-ratio-multipart".to_string(),
-                size: stored.len() as i64,
-                etag: Some("6bcf86bed8807b8e78f0fc6e0a53079d-2".to_string()),
-                parts: Arc::new(parts),
-                user_defined: Arc::new(user_defined),
-                ..Default::default()
-            },
-            stored,
-            plaintext,
-        };
-
-        let read = read_compressed_multipart(&fixture, None, &ObjectOptions::default()).await;
-
-        assert_eq!(read.len(), fixture.plaintext.len(), "full GET must return the logical size");
-        assert_eq!(read, fixture.plaintext, "high-ratio multipart payload must survive the roundtrip");
-    }
-
-    /// Full GET over a compressed multipart object must decode across part
-    /// boundaries: every part is an independent compressed stream (this is also
-    /// the on-disk shape written by builds before rustfs/rustfs#5169 disabled
-    /// multipart compression, so this pins legacy-object readability).
-    #[tokio::test]
-    async fn compressed_multipart_full_get_decodes_across_part_boundaries() {
-        let fixture = compressed_multipart_fixture(&[3 * 1024 * 1024, 2 * 1024 * 1024, 512 * 1024]).await;
-
-        let read = read_compressed_multipart(&fixture, None, &ObjectOptions::default()).await;
-
-        assert_eq!(read.len(), fixture.plaintext.len(), "full GET must return the logical size");
-        assert_eq!(read, fixture.plaintext, "full GET must reassemble all parts");
-    }
-
-    #[tokio::test]
-    async fn compressed_multipart_range_get_crosses_part_boundary() {
-        let fixture = compressed_multipart_fixture(&[3 * 1024 * 1024, 2 * 1024 * 1024]).await;
-        let boundary = 3 * 1024 * 1024_i64;
-        let rs = HTTPRangeSpec {
-            is_suffix_length: false,
-            start: boundary - 100_000,
-            end: boundary + 100_000 - 1,
-        };
-
-        let read = read_compressed_multipart(&fixture, Some(rs), &ObjectOptions::default()).await;
-
-        let expected = &fixture.plaintext[(boundary - 100_000) as usize..(boundary + 100_000) as usize];
-        assert_eq!(read, expected, "boundary-crossing range must splice both parts");
-    }
-
-    #[tokio::test]
-    async fn compressed_multipart_range_get_seeks_into_later_part() {
-        let fixture = compressed_multipart_fixture(&[3 * 1024 * 1024, 4 * 1024 * 1024]).await;
-        // Deep inside part 2 so the plan skips part 1 entirely and (when the
-        // part carries an index) seeks within part 2.
-        let start = 3 * 1024 * 1024_i64 + 2 * 1024 * 1024_i64 + 137;
-        let rs = HTTPRangeSpec {
-            is_suffix_length: false,
-            start,
-            end: start + 64 * 1024 - 1,
-        };
-
-        let read = read_compressed_multipart(&fixture, Some(rs), &ObjectOptions::default()).await;
-
-        let expected = &fixture.plaintext[start as usize..(start + 64 * 1024) as usize];
-        assert_eq!(read, expected, "range inside a later part must decode from that part");
-    }
-
-    /// Parts written without a compression index (small parts skip the index in
-    /// the rio-v2 backend) must still be rangeable: the plan starts at the part
-    /// boundary and skips decompressed bytes.
-    #[tokio::test]
-    async fn compressed_multipart_range_get_works_without_part_indexes() {
-        let mut fixture = compressed_multipart_fixture(&[1024 * 1024, 1024 * 1024]).await;
-        let parts = fixture
-            .object_info
-            .parts
-            .iter()
-            .map(|part| ObjectPartInfo {
-                index: None,
-                ..part.clone()
-            })
-            .collect::<Vec<_>>();
-        fixture.object_info.parts = Arc::new(parts);
-
-        let start = 1024 * 1024_i64 + 4096;
-        let rs = HTTPRangeSpec {
-            is_suffix_length: false,
-            start,
-            end: start + 32 * 1024 - 1,
-        };
-
-        let read = read_compressed_multipart(&fixture, Some(rs), &ObjectOptions::default()).await;
-
-        let expected = &fixture.plaintext[start as usize..(start + 32 * 1024) as usize];
-        assert_eq!(read, expected, "index-less parts must fall back to part-boundary skip");
-    }
-
-    #[tokio::test]
-    async fn compressed_multipart_part_number_get_returns_single_part() {
-        let part_sizes = [3 * 1024 * 1024, 2 * 1024 * 1024, 512 * 1024];
-        let fixture = compressed_multipart_fixture(&part_sizes).await;
-
-        let mut logical_offset = 0_usize;
-        for (i, part_size) in part_sizes.iter().enumerate() {
-            let opts = ObjectOptions {
-                part_number: Some(i + 1),
-                ..Default::default()
-            };
-
-            let read = read_compressed_multipart(&fixture, None, &opts).await;
-
-            let expected = &fixture.plaintext[logical_offset..logical_offset + part_size];
-            assert_eq!(read.len(), *part_size, "partNumber={} GET must return the part's logical size", i + 1);
-            assert_eq!(read, expected, "partNumber={} GET must return the original part bytes", i + 1);
-            logical_offset += part_size;
-        }
-    }
-
-    #[tokio::test]
-    async fn compressed_multipart_suffix_range_reads_tail() {
-        let fixture = compressed_multipart_fixture(&[3 * 1024 * 1024, 1024 * 1024]).await;
-        let suffix_len = 128 * 1024_i64;
-        let rs = HTTPRangeSpec {
-            is_suffix_length: true,
-            start: suffix_len,
-            end: -1,
-        };
-
-        let read = read_compressed_multipart(&fixture, Some(rs), &ObjectOptions::default()).await;
-
-        let expected = &fixture.plaintext[fixture.plaintext.len() - suffix_len as usize..];
-        assert_eq!(read, expected, "suffix range must return the tail of the last part");
-    }
-
-    /// Builds an SSE-C + disk-compression multipart object exactly like the
-    /// write path: each part is compressed into its own stream and then
-    /// encrypted with the per-part key schedule. The fixture is
-    /// legacy-encryption-specific (`rustfs_rio::EncryptReader`), matching the
-    /// pre-existing `build_legacy_ssec_multipart_fixture` shape, while the
-    /// compression layer follows the active backend feature.
-    async fn compressed_encrypted_multipart_fixture(key_bytes: [u8; 32], part_sizes: &[usize]) -> CompressedMultipartFixture {
-        let pattern = b"compressed encrypted multipart fixture data ";
-        let mut plaintext = Vec::new();
-        let mut stored = Vec::new();
-        let mut parts = Vec::with_capacity(part_sizes.len());
-
-        for (i, part_size) in part_sizes.iter().enumerate() {
-            let part_number = i + 1;
-            let mut part_plaintext = Vec::with_capacity(*part_size);
-            while part_plaintext.len() < *part_size {
-                part_plaintext.extend_from_slice(pattern);
-                part_plaintext.push(part_number as u8);
-            }
-            part_plaintext.truncate(*part_size);
-
-            let (compressed, index) = compressed_part_fixture(&part_plaintext).await;
-            let mut part_cipher = Vec::new();
-            rustfs_rio::EncryptReader::new_multipart(Cursor::new(compressed), key_bytes, LEGACY_FIXTURE_BASE_NONCE, part_number)
-                .read_to_end(&mut part_cipher)
-                .await
-                .expect("encrypt compressed fixture part");
-
-            parts.push(ObjectPartInfo {
-                number: part_number,
-                size: part_cipher.len(),
-                actual_size: *part_size as i64,
-                index,
-                ..Default::default()
-            });
-            stored.extend_from_slice(&part_cipher);
-            plaintext.extend_from_slice(&part_plaintext);
-        }
-
-        let mut user_defined = legacy_ssec_multipart_metadata(key_bytes, plaintext.len());
-        rustfs_utils::http::insert_str(
-            &mut user_defined,
-            rustfs_utils::http::SUFFIX_COMPRESSION,
-            crate::io_support::rio::compression_metadata_value(CompressionAlgorithm::default()),
-        );
-        rustfs_utils::http::insert_str(&mut user_defined, rustfs_utils::http::SUFFIX_ACTUAL_SIZE, plaintext.len().to_string());
-
-        let object_info = ObjectInfo {
-            bucket: "test-bucket".to_string(),
-            name: "compressed-encrypted-multipart".to_string(),
-            size: stored.len() as i64,
-            etag: Some(format!("6bcf86bed8807b8e78f0fc6e0a53079d-{}", part_sizes.len())),
-            parts: Arc::new(parts),
-            user_defined: Arc::new(user_defined),
-            ..Default::default()
-        };
-
-        CompressedMultipartFixture {
-            object_info,
-            stored,
-            plaintext,
-        }
-    }
-
-    async fn read_compressed_encrypted_multipart(
-        fixture: &CompressedMultipartFixture,
-        key_bytes: [u8; 32],
-        rs: Option<HTTPRangeSpec>,
-        opts: &ObjectOptions,
-    ) -> Vec<u8> {
-        let headers = ssec_headers_from_key(key_bytes);
-        let (_, offset, length) =
-            GetObjectReader::new(Box::new(Cursor::new(Vec::new())), rs.clone(), &fixture.object_info, opts, &headers)
-                .await
-                .expect("plan compressed encrypted multipart read");
-
-        let end = offset + usize::try_from(length).expect("storage window length must be non-negative");
-        assert!(
-            end <= fixture.stored.len(),
-            "planned storage window {offset}..{end} exceeds stored stream of {} bytes",
-            fixture.stored.len()
-        );
-        let window = fixture.stored[offset..end].to_vec();
-
-        let (mut reader, replay_offset, replay_length) =
-            GetObjectReader::new(Box::new(Cursor::new(window)), rs, &fixture.object_info, opts, &headers)
-                .await
-                .expect("build compressed encrypted multipart reader");
-        assert_eq!((replay_offset, replay_length), (offset, length), "read plan must be deterministic");
-
-        reader.read_all().await.expect("read compressed encrypted multipart stream")
-    }
-
-    #[tokio::test]
-    async fn compressed_encrypted_multipart_full_get_roundtrip() {
-        let key_bytes = [0x6Eu8; 32];
-        let fixture = compressed_encrypted_multipart_fixture(key_bytes, &[3 * 1024 * 1024, 1024 * 1024]).await;
-
-        let read = read_compressed_encrypted_multipart(&fixture, key_bytes, None, &ObjectOptions::default()).await;
-
-        assert_eq!(read.len(), fixture.plaintext.len(), "full GET must return the logical size");
-        assert_eq!(read, fixture.plaintext, "SSE-C + compression full GET must reassemble all parts");
-    }
-
-    #[tokio::test]
-    async fn compressed_encrypted_multipart_range_crosses_part_boundary() {
-        let key_bytes = [0x6Eu8; 32];
-        let fixture = compressed_encrypted_multipart_fixture(key_bytes, &[3 * 1024 * 1024, 1024 * 1024]).await;
-        let boundary = 3 * 1024 * 1024_i64;
-        let rs = HTTPRangeSpec {
-            is_suffix_length: false,
-            start: boundary - 65_536,
-            end: boundary + 65_536 - 1,
-        };
-
-        let read = read_compressed_encrypted_multipart(&fixture, key_bytes, Some(rs), &ObjectOptions::default()).await;
-
-        let expected = &fixture.plaintext[(boundary - 65_536) as usize..(boundary + 65_536) as usize];
-        assert_eq!(read, expected, "SSE-C + compression boundary-crossing range must splice both parts");
-    }
-
-    #[tokio::test]
-    async fn compressed_encrypted_multipart_part_number_get_returns_single_part() {
-        let key_bytes = [0x6Eu8; 32];
-        let part_sizes = [3 * 1024 * 1024, 1024 * 1024];
-        let fixture = compressed_encrypted_multipart_fixture(key_bytes, &part_sizes).await;
-
-        let opts = ObjectOptions {
-            part_number: Some(2),
-            ..Default::default()
-        };
-        let read = read_compressed_encrypted_multipart(&fixture, key_bytes, None, &opts).await;
-
-        let expected = &fixture.plaintext[part_sizes[0]..];
-        assert_eq!(read.len(), part_sizes[1], "partNumber=2 GET must return the part's logical size");
-        assert_eq!(read, expected, "partNumber=2 GET must return the original part bytes");
-    }
-
     #[tokio::test]
     async fn test_get_object_reader_rejects_ssec_read_without_headers() {
         let object_info = ObjectInfo {

crates/ecstore/src/set_disk/mod.rs

@@ -880,12 +880,44 @@ mod prepared_get_object_metadata_tests {
     use super::*;
     use crate::ecstore_validation_blackbox::make_local_set_disks;
     use crate::object_api::{BLOCK_SIZE_V2, PutObjReader};
-    use crate::set_disk::core::io_primitives::disk_call_counters;
+    use crate::set_disk::core::io_primitives::{bounded_metadata_fanout_order, disk_call_counters, rename_fanout_barrier};
     use crate::storage_api_contracts::bucket::{BucketOperations as _, MakeBucketOptions};
     use crate::storage_api_contracts::object::{ObjectIO as _, ObjectOperations as _};
+    use crate::test_metrics::CapturingRecorder;
     use http::HeaderMap;
     use tokio::io::AsyncReadExt;
 
+    const READ_VERSION_BARRIER_GUARD: std::time::Duration = std::time::Duration::from_secs(10);
+
+    fn object_with_initial_data_shards(bucket: &str, prefix: &str) -> String {
+        (0..1000)
+            .map(|index| format!("{prefix}-{index}.bin"))
+            .find(|name| {
+                let order = bounded_metadata_fanout_order(bucket, name, 4, 2);
+                let distribution = FileInfo::new(&[bucket, name].join("/"), 2, 2).erasure.distribution;
+                let mut seen = [false; 2];
+                for disk_index in order.into_iter().take(3) {
+                    if let Some(block_index @ 1..=2) = distribution.get(disk_index).copied() {
+                        seen[block_index - 1] = true;
+                    }
+                }
+                seen.into_iter().all(|seen| seen)
+            })
+            .expect("test should find an object whose initial fanout covers both data shards")
+    }
+
+    fn bounded_spare_disk_index(bucket: &str, object: &str) -> usize {
+        *bounded_metadata_fanout_order(bucket, object, 4, 2)
+            .get(3)
+            .expect("4-disk test geometry should leave one bounded spare disk")
+    }
+
+    fn bounded_slow_initial_disk_index(bucket: &str, object: &str) -> usize {
+        *bounded_metadata_fanout_order(bucket, object, 4, 2)
+            .get(2)
+            .expect("4-disk test geometry should include a third initial metadata disk")
+    }
+
     #[tokio::test]
     async fn prepared_metadata_is_consumed_exactly_once() {
         let snapshot = GetObjectFileInfo::owned(FileInfo::default(), Vec::new(), Vec::new());
@@ -1002,6 +1034,307 @@ mod prepared_get_object_metadata_tests {
         );
     }
 
+    #[test]
+    #[serial_test::serial(body_cache_hook)]
+    fn inline_data_read_early_stop_reader_returns_exact_body() {
+        let runtime = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .expect("current-thread runtime should build");
+        let bucket = "inline-data-read-early-stop-reader";
+        let object = object_with_initial_data_shards(bucket, "inline-data-read-early-stop-reader-object");
+        let payload = b"inline early-stop reader payload".repeat(256);
+        let recorder = CapturingRecorder::default();
+        let previous_gate = rustfs_io_metrics::get_stage_metrics_enabled();
+        rustfs_io_metrics::set_get_stage_metrics_enabled(true);
+
+        let (restored, object_size, calls_total) = metrics::with_local_recorder(&recorder, || {
+            runtime.block_on(async {
+                let (_dirs, set_disks) = make_local_set_disks(4, 2).await;
+                let opts = ObjectOptions {
+                    no_lock: true,
+                    ..Default::default()
+                };
+
+                set_disks
+                    .make_bucket(bucket, &MakeBucketOptions::default())
+                    .await
+                    .expect("bucket should be created");
+                let mut put_reader = PutObjReader::from_vec(payload.clone());
+                set_disks
+                    .put_object(bucket, &object, &mut put_reader, &opts)
+                    .await
+                    .expect("inline object should be written");
+
+                temp_env::async_with_vars(
+                    [
+                        ("RUSTFS_GET_METADATA_EARLY_STOP_ENABLE", Some("true")),
+                        ("RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE", Some("true")),
+                        ("RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT", Some("true")),
+                    ],
+                    async {
+                        let slow_initial_disk = bounded_slow_initial_disk_index(bucket, &object);
+                        let barrier =
+                            rename_fanout_barrier::arm(&object, slow_initial_disk, rename_fanout_barrier::PHASE_READ_VERSION);
+                        let calls = disk_call_counters::observe(&object);
+                        let set_disks_for_read = Arc::clone(&set_disks);
+                        let opts_for_read = opts.clone();
+                        let object_for_read = object.clone();
+                        let mut open_reader = tokio::spawn(async move {
+                            set_disks_for_read
+                                .get_object_reader(bucket, &object_for_read, None, HeaderMap::new(), &opts_for_read)
+                                .await
+                        });
+
+                        tokio::time::timeout(READ_VERSION_BARRIER_GUARD, barrier.wait_until_paused())
+                            .await
+                            .expect("bounded inline GET should pause a slow initial metadata read");
+                        let mut reader = tokio::time::timeout(READ_VERSION_BARRIER_GUARD, &mut open_reader)
+                            .await
+                            .expect("production inline GET should return before the paused metadata response")
+                            .expect("inline GET reader task should not panic")
+                            .expect("inline GET reader should open");
+                        let object_size = reader.object_info.size;
+                        let mut restored = Vec::new();
+                        reader
+                            .stream
+                            .read_to_end(&mut restored)
+                            .await
+                            .expect("inline GET body should stream");
+
+                        (restored, object_size, calls.total(disk_call_counters::KIND_READ_VERSION))
+                    },
+                )
+                .await
+            })
+        });
+        rustfs_io_metrics::set_get_stage_metrics_enabled(previous_gate);
+
+        assert_eq!(object_size, payload.len() as i64);
+        assert_eq!(restored, payload);
+        assert_eq!(calls_total, 4, "bounded production GET should schedule the initial quorum plus one spare");
+        assert_eq!(
+            recorder.histogram_values(
+                "rustfs_io_get_object_metadata_fanout_scheduled",
+                &[("path", GET_OBJECT_PATH_LEGACY_DUPLEX)]
+            ),
+            vec![4.0],
+            "bounded production GET should record all scheduled metadata tasks"
+        );
+        assert_eq!(
+            recorder.histogram_values(
+                "rustfs_io_get_object_metadata_fanout_completed",
+                &[("path", GET_OBJECT_PATH_LEGACY_DUPLEX)]
+            ),
+            vec![3.0],
+            "bounded production GET should record only observed metadata responses as completed"
+        );
+        assert_eq!(
+            recorder.histogram_values(
+                "rustfs_io_get_object_metadata_fanout_cancelled",
+                &[("path", GET_OBJECT_PATH_LEGACY_DUPLEX)]
+            ),
+            vec![1.0],
+            "bounded production GET should record the aborted slow metadata task"
+        );
+    }
+
+    #[test]
+    #[serial_test::serial(body_cache_hook)]
+    fn prepared_metadata_uses_full_fanout_even_when_data_read_early_stop_is_enabled() {
+        let runtime = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .expect("current-thread runtime should build");
+        let bucket = "prepared-metadata-early-stop-enabled";
+        let object = object_with_initial_data_shards(bucket, "prepared-metadata-early-stop-enabled-object");
+        let payload = b"prepared metadata early-stop enabled payload".repeat(16);
+        let recorder = CapturingRecorder::default();
+        let previous_gate = rustfs_io_metrics::get_stage_metrics_enabled();
+        rustfs_io_metrics::set_get_stage_metrics_enabled(true);
+
+        let (restored, calls_total) = metrics::with_local_recorder(&recorder, || {
+            runtime.block_on(async {
+                let (_dirs, set_disks) = make_local_set_disks(4, 2).await;
+                let opts = ObjectOptions {
+                    no_lock: true,
+                    ..Default::default()
+                };
+
+                set_disks
+                    .make_bucket(bucket, &MakeBucketOptions::default())
+                    .await
+                    .expect("bucket should be created");
+                let mut put_reader = PutObjReader::from_vec(payload.clone());
+                set_disks
+                    .put_object(bucket, &object, &mut put_reader, &opts)
+                    .await
+                    .expect("object should be written");
+
+                temp_env::async_with_vars(
+                    [
+                        ("RUSTFS_GET_METADATA_EARLY_STOP_ENABLE", Some("true")),
+                        ("RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE", Some("true")),
+                        ("RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT", Some("true")),
+                    ],
+                    async {
+                        let calls = disk_call_counters::observe(&object);
+                        let metadata = set_disks
+                            .prepare_get_object_metadata(bucket, &object, &opts)
+                            .await
+                            .expect("prepared metadata should resolve");
+                        let calls_total = calls.total(disk_call_counters::KIND_READ_VERSION);
+
+                        let mut reader = set_disks
+                            .get_object_reader_with_prepared_metadata(bucket, &object, None, HeaderMap::new(), &opts, metadata)
+                            .await
+                            .expect("prepared body reader should open");
+                        let mut restored = Vec::new();
+                        reader
+                            .stream
+                            .read_to_end(&mut restored)
+                            .await
+                            .expect("prepared body should stream");
+                        (restored, calls_total)
+                    },
+                )
+                .await
+            })
+        });
+        rustfs_io_metrics::set_get_stage_metrics_enabled(previous_gate);
+
+        assert_eq!(restored, payload);
+        assert_eq!(
+            calls_total, 4,
+            "prepared metadata must opt out of data-read early-stop until the read shape is known"
+        );
+        assert_eq!(
+            recorder.histogram_values(
+                "rustfs_io_get_object_metadata_fanout_scheduled",
+                &[("path", GET_OBJECT_PATH_LEGACY_DUPLEX)]
+            ),
+            vec![4.0],
+            "prepared metadata should schedule the full metadata fanout"
+        );
+        assert_eq!(
+            recorder.histogram_values(
+                "rustfs_io_get_object_metadata_fanout_completed",
+                &[("path", GET_OBJECT_PATH_LEGACY_DUPLEX)]
+            ),
+            vec![4.0],
+            "prepared metadata must wait for every scheduled metadata response"
+        );
+        assert_eq!(
+            recorder.histogram_values(
+                "rustfs_io_get_object_metadata_fanout_cancelled",
+                &[("path", GET_OBJECT_PATH_LEGACY_DUPLEX)]
+            ),
+            vec![0.0],
+            "prepared metadata must not cancel metadata responses"
+        );
+    }
+
+    #[test]
+    #[serial_test::serial(body_cache_hook)]
+    fn data_read_early_stop_request_shapes_full_wait_in_production_reader() {
+        let runtime = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .expect("current-thread runtime should build");
+        let bucket = "data-read-early-stop-shape-reader";
+        let payload = b"shape-gated inline reader payload".repeat(256);
+
+        for (object_prefix, range, configure_opts, expected_body) in [
+            (
+                "data-read-early-stop-range-reader-object",
+                Some(HTTPRangeSpec {
+                    start: 0,
+                    end: 3,
+                    is_suffix_length: false,
+                }),
+                None,
+                payload[..4].to_vec(),
+            ),
+            ("data-read-early-stop-part-reader-object", None, Some(1), payload.clone()),
+        ] {
+            let recorder = CapturingRecorder::default();
+            let previous_gate = rustfs_io_metrics::get_stage_metrics_enabled();
+            rustfs_io_metrics::set_get_stage_metrics_enabled(true);
+            let (restored, calls_total) = metrics::with_local_recorder(&recorder, || {
+                runtime.block_on(async {
+                    let (_dirs, set_disks) = make_local_set_disks(4, 2).await;
+                    let object = object_with_initial_data_shards(bucket, object_prefix);
+                    let mut opts = ObjectOptions {
+                        no_lock: true,
+                        ..Default::default()
+                    };
+                    opts.part_number = configure_opts;
+
+                    set_disks
+                        .make_bucket(bucket, &MakeBucketOptions::default())
+                        .await
+                        .expect("bucket should be created");
+                    let mut put_reader = PutObjReader::from_vec(payload.clone());
+                    set_disks
+                        .put_object(bucket, &object, &mut put_reader, &opts)
+                        .await
+                        .expect("inline object should be written");
+
+                    temp_env::async_with_vars(
+                        [
+                            ("RUSTFS_GET_METADATA_EARLY_STOP_ENABLE", Some("true")),
+                            ("RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE", Some("true")),
+                            ("RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT", Some("true")),
+                        ],
+                        async {
+                            let calls = disk_call_counters::observe(&object);
+                            let mut reader = set_disks
+                                .get_object_reader(bucket, &object, range, HeaderMap::new(), &opts)
+                                .await
+                                .expect("shape-gated GET reader should open");
+                            let mut restored = Vec::new();
+                            reader
+                                .stream
+                                .read_to_end(&mut restored)
+                                .await
+                                .expect("shape-gated GET body should stream");
+                            (restored, calls.total(disk_call_counters::KIND_READ_VERSION))
+                        },
+                    )
+                    .await
+                })
+            });
+            rustfs_io_metrics::set_get_stage_metrics_enabled(previous_gate);
+
+            assert_eq!(restored, expected_body);
+            assert_eq!(calls_total, 4, "shape-gated production GET should keep full metadata fanout");
+            assert_eq!(
+                recorder.histogram_values(
+                    "rustfs_io_get_object_metadata_fanout_scheduled",
+                    &[("path", GET_OBJECT_PATH_LEGACY_DUPLEX)]
+                ),
+                vec![4.0],
+                "shape-gated production GET should schedule the full metadata fanout"
+            );
+            assert_eq!(
+                recorder.histogram_values(
+                    "rustfs_io_get_object_metadata_fanout_completed",
+                    &[("path", GET_OBJECT_PATH_LEGACY_DUPLEX)]
+                ),
+                vec![4.0],
+                "shape-gated production GET must wait for every scheduled metadata response"
+            );
+            assert_eq!(
+                recorder.histogram_values(
+                    "rustfs_io_get_object_metadata_fanout_cancelled",
+                    &[("path", GET_OBJECT_PATH_LEGACY_DUPLEX)]
+                ),
+                vec![0.0],
+                "shape-gated production GET must not cancel metadata responses"
+            );
+        }
+    }
+
     #[tokio::test]
     #[serial_test::serial(body_cache_hook)]
     async fn prepared_reader_rebuilds_object_info_when_precomputed_value_is_absent() {
@@ -1105,7 +1438,7 @@ impl SetDisks {
         object: &str,
         opts: &ObjectOptions,
     ) -> Result<PreparedGetObjectMetadata> {
-        let snapshot = self.get_object_fileinfo(bucket, object, opts, true, true).await?;
+        let snapshot = self.get_object_fileinfo(bucket, object, opts, true, false).await?;
         let object_info = build_get_object_info(snapshot.fi(), bucket, object, opts.versioned || opts.version_suspended);
         Ok(PreparedGetObjectMetadata {
             snapshot,
@@ -3411,9 +3744,15 @@ fn collect_inline_data_shard_fileinfos_by_index<'a>(
         if block_index == 0 || block_index > data_shards {
             continue;
         }
+        if file_info.erasure.index != block_index {
+            continue;
+        }
         if !file_info.has_valid_erasure_geometry() {
             continue;
         }
+        if !core::io_primitives::metadata_early_stop_candidate_matches(file_info, fi) {
+            continue;
+        }
         if file_info.data.as_ref().is_none_or(|data| data.is_empty()) {
             continue;
         }
@@ -9221,6 +9560,9 @@ mod tests {
             HashAlgorithm::HighwayHash256S
         };
         let shards = erasure.encode_data(payload).expect("payload should encode");
+        let version_id = Some(Uuid::new_v4());
+        let data_dir = Some(Uuid::new_v4());
+        let mod_time = Some(OffsetDateTime::now_utc());
         let mut files = Vec::with_capacity(shards.len());
 
         for shard in shards {
@@ -9233,6 +9575,16 @@ mod tests {
             writer.shutdown().await.expect("inline writer should shutdown");
             let data = writer.into_inline_data().expect("inline data should be retained");
             let mut file = FileInfo::new("bucket/object", erasure.data_shards, erasure.parity_shards);
+            file.volume = "bucket".to_string();
+            file.name = "object".to_string();
+            file.size = i64::try_from(payload.len()).expect("test payload should fit i64");
+            file.is_latest = true;
+            file.version_id = version_id;
+            file.data_dir = data_dir;
+            file.mod_time = mod_time;
+            file.metadata.insert("etag".to_string(), "etag-inline".to_string());
+            file.add_object_part(1, "part-etag-inline".to_string(), payload.len(), file.mod_time, file.size, None, None);
+            file.set_inline_data();
             file.erasure.index = files.len() + 1;
             file.data = Some(Bytes::from(data));
             files.push(file);
@@ -9245,16 +9597,40 @@ mod tests {
         inline_bitrot_files_for_payload_with_mode(payload, false).await
     }
 
+    fn disk_ordered_fileinfos(files: &[FileInfo]) -> Vec<FileInfo> {
+        let distribution = &files
+            .first()
+            .expect("inline data shard fixture should include metadata")
+            .erasure
+            .distribution;
+        distribution
+            .iter()
+            .map(|block_index| {
+                files
+                    .get(block_index.checked_sub(1).expect("erasure block indexes are one-based"))
+                    .expect("inline data shard fixture should include every distributed shard")
+                    .clone()
+            })
+            .collect()
+    }
+
     fn inline_data_shard_fileinfo(
-        name: &str,
         data_blocks: usize,
         parity_blocks: usize,
         erasure_index: usize,
         distribution: &[usize],
         data: Option<&'static [u8]>,
     ) -> FileInfo {
-        let mut fi = FileInfo::new(name, data_blocks, parity_blocks);
-        fi.name = name.to_string();
+        let mut fi = FileInfo::new("object", data_blocks, parity_blocks);
+        fi.name = "object".to_string();
+        fi.volume = "bucket".to_string();
+        fi.size = 4;
+        fi.is_latest = true;
+        fi.data_dir = Some(Uuid::nil());
+        fi.mod_time = Some(OffsetDateTime::UNIX_EPOCH);
+        fi.metadata.insert("etag".to_string(), "etag-inline".to_string());
+        fi.add_object_part(1, "part-etag-inline".to_string(), 4, fi.mod_time, 4, None, None);
+        fi.set_inline_data();
         fi.erasure.index = erasure_index;
         fi.erasure.distribution = distribution.to_vec();
         fi.data = data.map(Bytes::from_static);
@@ -9264,36 +9640,41 @@ mod tests {
     #[test]
     fn collect_inline_data_shards_by_index_uses_distribution_order() {
         let distribution = vec![3, 1, 5, 2, 4, 6];
-        let mut fi = FileInfo::new("object", 4, 2);
+        let mut fi = inline_data_shard_fileinfo(4, 2, 1, &distribution, Some(b"x"));
+        fi.erasure.index = 1;
         fi.erasure.distribution = distribution.clone();
         let files = vec![
-            inline_data_shard_fileinfo("block-3", 4, 2, 3, &distribution, Some(b"c")),
-            inline_data_shard_fileinfo("block-1", 4, 2, 1, &distribution, Some(b"a")),
-            inline_data_shard_fileinfo("parity-5", 4, 2, 5, &distribution, Some(b"p")),
-            inline_data_shard_fileinfo("block-2", 4, 2, 2, &distribution, Some(b"b")),
-            inline_data_shard_fileinfo("block-4", 4, 2, 4, &distribution, Some(b"d")),
-            inline_data_shard_fileinfo("parity-6", 4, 2, 6, &distribution, Some(b"q")),
+            inline_data_shard_fileinfo(4, 2, 3, &distribution, Some(b"c")),
+            inline_data_shard_fileinfo(4, 2, 1, &distribution, Some(b"a")),
+            inline_data_shard_fileinfo(4, 2, 5, &distribution, Some(b"p")),
+            inline_data_shard_fileinfo(4, 2, 2, &distribution, Some(b"b")),
+            inline_data_shard_fileinfo(4, 2, 4, &distribution, Some(b"d")),
+            inline_data_shard_fileinfo(4, 2, 6, &distribution, Some(b"q")),
         ];
 
         let data_files =
             collect_inline_data_shard_fileinfos_by_index(&files, &fi, 4, |_| true).expect("all data shards should be collected");
 
         assert_eq!(
-            data_files.iter().map(|file| file.name.as_str()).collect::<Vec<_>>(),
-            ["block-1", "block-2", "block-3", "block-4"]
+            data_files
+                .iter()
+                .map(|file| file.data.as_deref().expect("fixture carries inline bytes"))
+                .collect::<Vec<_>>(),
+            [b"a".as_slice(), b"b".as_slice(), b"c".as_slice(), b"d".as_slice()]
         );
     }
 
     #[test]
     fn collect_inline_data_shards_by_index_rejects_missing_data_shard() {
         let distribution = vec![1, 2, 3, 4];
-        let mut fi = FileInfo::new("object", 2, 2);
+        let mut fi = inline_data_shard_fileinfo(2, 2, 1, &distribution, Some(b"x"));
+        fi.erasure.index = 1;
         fi.erasure.distribution = distribution.clone();
         let files = vec![
-            inline_data_shard_fileinfo("block-1", 2, 2, 1, &distribution, Some(b"a")),
-            inline_data_shard_fileinfo("block-2", 2, 2, 2, &distribution, None),
-            inline_data_shard_fileinfo("parity-3", 2, 2, 3, &distribution, Some(b"p")),
-            inline_data_shard_fileinfo("parity-4", 2, 2, 4, &distribution, Some(b"q")),
+            inline_data_shard_fileinfo(2, 2, 1, &distribution, Some(b"a")),
+            inline_data_shard_fileinfo(2, 2, 2, &distribution, None),
+            inline_data_shard_fileinfo(2, 2, 3, &distribution, Some(b"p")),
+            inline_data_shard_fileinfo(2, 2, 4, &distribution, Some(b"q")),
         ];
 
         assert!(collect_inline_data_shard_fileinfos_by_index(&files, &fi, 2, |_| true).is_none());
@@ -9426,10 +9807,8 @@ mod tests {
 
         let payload = vec![b'i'; 192 * 1024];
         let (erasure, files, _read_length, _checksum_algo) = inline_bitrot_files_for_payload(&payload).await;
-        let mut fi = FileInfo::new("bucket/object", erasure.data_shards, erasure.parity_shards);
-        fi.size = payload.len() as i64;
-        fi.data = files[0].data.clone();
-        fi.add_object_part(1, String::new(), payload.len(), None, payload.len() as i64, None, None);
+        let fi = files[0].clone();
+        let disk_files = disk_ordered_fileinfos(&files);
 
         let disks = vec![Some(disk); erasure.total_shard_count()];
         let metrics_size_bucket = rustfs_io_metrics::get_object_size_bucket(fi.size);
@@ -9438,7 +9817,7 @@ mod tests {
             "bucket",
             "object",
             &fi,
-            &files,
+            &disk_files,
             &disks,
             true,
             GET_CODEC_STREAMING_OBJECT_CLASS_PLAIN_SINGLE_PART,
@@ -9469,10 +9848,8 @@ mod tests {
         let payload = vec![b'v'; 64 * 1024];
         let payload_size = i64::try_from(payload.len()).expect("test payload size should fit i64");
         let (erasure, files, _read_length, _checksum_algo) = inline_bitrot_files_for_payload(&payload).await;
-        let mut fi = FileInfo::new("bucket/object", erasure.data_shards, erasure.parity_shards);
-        fi.size = payload_size;
-        fi.data = files[0].data.clone();
-        fi.add_object_part(1, String::new(), payload.len(), None, payload_size, None, None);
+        let fi = files[0].clone();
+        let disk_files = disk_ordered_fileinfos(&files);
 
         let mut object_info = ObjectInfo {
             size: payload_size,
@@ -9503,7 +9880,7 @@ mod tests {
             "bucket",
             "object",
             &fi,
-            &files,
+            &disk_files,
             &vec![Some(disk); erasure.total_shard_count()],
             true,
             GET_CODEC_STREAMING_OBJECT_CLASS_PLAIN_SINGLE_PART,

crates/ecstore/src/set_disk/ops/object.rs

@@ -188,6 +188,74 @@ async fn get_object_reader_with_context(
     GetObjectReader::new_with_resolver(reader, range, object_info, opts, headers, ctx.object_encryption_resolver()).await
 }
 
+fn data_read_metadata_early_stop_request_shape_allowed(range: &Option<HTTPRangeSpec>, opts: &ObjectOptions) -> bool {
+    range.is_none()
+        && opts.part_number.is_none()
+        && opts.version_id.is_none()
+        && !opts.incl_free_versions
+        && !opts.skip_free_version
+        && !opts.raw_data_movement_read
+        && !opts.data_movement
+        && !crate::object_api::restore_request_active(opts)
+}
+
+#[cfg(test)]
+mod data_read_metadata_early_stop_request_shape_tests {
+    use super::*;
+
+    #[test]
+    fn data_read_metadata_early_stop_only_allows_whole_latest_plain_get_shape() {
+        assert!(data_read_metadata_early_stop_request_shape_allowed(&None, &ObjectOptions::default()));
+
+        let range = Some(HTTPRangeSpec {
+            is_suffix_length: false,
+            start: 0,
+            end: 0,
+        });
+        assert!(!data_read_metadata_early_stop_request_shape_allowed(&range, &ObjectOptions::default()));
+
+        let part_opts = ObjectOptions {
+            part_number: Some(1),
+            ..Default::default()
+        };
+        assert!(!data_read_metadata_early_stop_request_shape_allowed(&None, &part_opts));
+
+        let version_opts = ObjectOptions {
+            version_id: Some(Uuid::new_v4().to_string()),
+            ..Default::default()
+        };
+        assert!(!data_read_metadata_early_stop_request_shape_allowed(&None, &version_opts));
+
+        let incl_free_opts = ObjectOptions {
+            incl_free_versions: true,
+            ..Default::default()
+        };
+        assert!(!data_read_metadata_early_stop_request_shape_allowed(&None, &incl_free_opts));
+
+        let skip_free_opts = ObjectOptions {
+            skip_free_version: true,
+            ..Default::default()
+        };
+        assert!(!data_read_metadata_early_stop_request_shape_allowed(&None, &skip_free_opts));
+
+        let data_movement_opts = ObjectOptions {
+            data_movement: true,
+            ..Default::default()
+        };
+        assert!(!data_read_metadata_early_stop_request_shape_allowed(&None, &data_movement_opts));
+
+        let raw_data_movement_opts = ObjectOptions {
+            raw_data_movement_read: true,
+            ..Default::default()
+        };
+        assert!(!data_read_metadata_early_stop_request_shape_allowed(&None, &raw_data_movement_opts));
+
+        let mut restore_opts = ObjectOptions::default();
+        restore_opts.transition.restore_request.days = Some(1);
+        assert!(!data_read_metadata_early_stop_request_shape_allowed(&None, &restore_opts));
+    }
+}
+
 /// Length of the full plaintext body when — and only when — this read's output
 /// is exactly the object's complete plaintext, so the app-layer body cache may
 /// serve it in place of the erasure read.
@@ -431,7 +499,16 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
         let (snapshot, prepared_object_info) = if let Some(prepared) = take_prepared_get_object_metadata() {
             (prepared.snapshot, prepared.object_info)
         } else {
-            match self.get_object_fileinfo(bucket, object, opts, true, true).await {
+            match self
+                .get_object_fileinfo(
+                    bucket,
+                    object,
+                    opts,
+                    true,
+                    data_read_metadata_early_stop_request_shape_allowed(&range, opts),
+                )
+                .await
+            {
                 Ok(snapshot) => (snapshot, None),
                 Err(err) => {
                     rustfs_io_metrics::record_get_object_metadata_phase_duration(metadata_stage_start.elapsed().as_secs_f64());
@@ -6102,7 +6179,10 @@ mod inline_put_commit_path_tests {
 mod get_object_downstream_close_accounting_tests {
     use super::hermetic_set_disks_support::hermetic_set_disks;
     use super::*;
-    use crate::diagnostics::get::{GET_OBJECT_PATH_INTERNAL_META, GET_STAGE_DECODE, GET_STAGE_EMIT, GetObjectFailureReason};
+    use crate::diagnostics::get::{
+        GET_METADATA_EARLY_STOP_REASON_UNSAFE_REQUEST, GET_OBJECT_PATH_INTERNAL_META, GET_STAGE_DECODE, GET_STAGE_EMIT,
+        GetObjectFailureReason,
+    };
     use crate::disk::RUSTFS_META_BUCKET;
     use crate::storage_api_contracts::bucket::{BucketOperations as _, MakeBucketOptions};
     use crate::storage_api_contracts::object::{ObjectIO as _, ObjectOperations as _};
@@ -6221,7 +6301,22 @@ mod get_object_downstream_close_accounting_tests {
         let previous_gate = rustfs_io_metrics::get_stage_metrics_enabled();
         rustfs_io_metrics::set_get_stage_metrics_enabled(true);
 
-        let (internal_missing, legacy_unknown, internal_fanout, legacy_fanout) = metrics::with_local_recorder(&recorder, || {
+        let (
+            internal_missing,
+            legacy_unknown,
+            internal_fanout,
+            legacy_fanout,
+            internal_scheduled,
+            legacy_scheduled,
+            internal_completed,
+            legacy_completed,
+            internal_cancelled,
+            legacy_cancelled,
+            internal_unsafe_miss,
+            legacy_unsafe_miss,
+            internal_saved,
+            legacy_saved,
+        ) = metrics::with_local_recorder(&recorder, || {
             runtime.block_on(async {
                 let (_temp_dirs, _disk_stores, set_disks) = hermetic_set_disks(4).await;
                 let options = ObjectOptions {
@@ -6265,6 +6360,54 @@ mod get_object_downstream_close_accounting_tests {
                         "rustfs_io_get_object_metadata_fanout_error_responses",
                         &[("path", GET_OBJECT_PATH_LEGACY_DUPLEX)],
                     ),
+                    recorder.histogram_values(
+                        "rustfs_io_get_object_metadata_fanout_scheduled",
+                        &[("path", GET_OBJECT_PATH_INTERNAL_META)],
+                    ),
+                    recorder.histogram_values(
+                        "rustfs_io_get_object_metadata_fanout_scheduled",
+                        &[("path", GET_OBJECT_PATH_LEGACY_DUPLEX)],
+                    ),
+                    recorder.histogram_values(
+                        "rustfs_io_get_object_metadata_fanout_completed",
+                        &[("path", GET_OBJECT_PATH_INTERNAL_META)],
+                    ),
+                    recorder.histogram_values(
+                        "rustfs_io_get_object_metadata_fanout_completed",
+                        &[("path", GET_OBJECT_PATH_LEGACY_DUPLEX)],
+                    ),
+                    recorder.histogram_values(
+                        "rustfs_io_get_object_metadata_fanout_cancelled",
+                        &[("path", GET_OBJECT_PATH_INTERNAL_META)],
+                    ),
+                    recorder.histogram_values(
+                        "rustfs_io_get_object_metadata_fanout_cancelled",
+                        &[("path", GET_OBJECT_PATH_LEGACY_DUPLEX)],
+                    ),
+                    recorder.counter_value(
+                        "rustfs_io_get_object_metadata_early_stop_total",
+                        &[
+                            ("path", GET_OBJECT_PATH_INTERNAL_META),
+                            ("decision", "miss"),
+                            ("reason", GET_METADATA_EARLY_STOP_REASON_UNSAFE_REQUEST),
+                        ],
+                    ),
+                    recorder.counter_value(
+                        "rustfs_io_get_object_metadata_early_stop_total",
+                        &[
+                            ("path", GET_OBJECT_PATH_LEGACY_DUPLEX),
+                            ("decision", "miss"),
+                            ("reason", GET_METADATA_EARLY_STOP_REASON_UNSAFE_REQUEST),
+                        ],
+                    ),
+                    recorder.histogram_values(
+                        "rustfs_io_get_object_metadata_early_stop_saved_responses",
+                        &[("path", GET_OBJECT_PATH_INTERNAL_META)],
+                    ),
+                    recorder.histogram_values(
+                        "rustfs_io_get_object_metadata_early_stop_saved_responses",
+                        &[("path", GET_OBJECT_PATH_LEGACY_DUPLEX)],
+                    ),
                 )
             })
         });
@@ -6277,6 +6420,50 @@ mod get_object_downstream_close_accounting_tests {
         );
         assert_eq!(internal_fanout, vec![4.0], "internal metadata fanout must retain its path label");
         assert!(legacy_fanout.is_empty(), "internal metadata fanout must not leak into legacy_duplex");
+        assert_eq!(
+            internal_scheduled,
+            vec![4.0],
+            "internal metadata lifecycle scheduled count must retain its path label"
+        );
+        assert!(
+            legacy_scheduled.is_empty(),
+            "internal metadata lifecycle scheduled count must not leak into legacy_duplex"
+        );
+        assert_eq!(
+            internal_completed,
+            vec![4.0],
+            "internal metadata lifecycle completed count must retain its path label"
+        );
+        assert!(
+            legacy_completed.is_empty(),
+            "internal metadata lifecycle completed count must not leak into legacy_duplex"
+        );
+        assert_eq!(
+            internal_cancelled,
+            vec![0.0],
+            "internal metadata full-wait lifecycle must record zero cancellations"
+        );
+        assert!(
+            legacy_cancelled.is_empty(),
+            "internal metadata lifecycle cancelled count must not leak into legacy_duplex"
+        );
+        assert_eq!(
+            internal_unsafe_miss, 1,
+            "internal metadata unsafe early-stop miss must retain its path label"
+        );
+        assert_eq!(
+            legacy_unsafe_miss, 0,
+            "internal metadata unsafe early-stop miss must not leak into legacy_duplex"
+        );
+        assert_eq!(
+            internal_saved,
+            vec![0.0],
+            "internal metadata unsafe miss must record zero saved responses on internal_meta"
+        );
+        assert!(
+            legacy_saved.is_empty(),
+            "internal metadata unsafe miss saved responses must not leak into legacy_duplex"
+        );
     }
 }
 

crates/heal/tests/endpoint_index_test.rs

@@ -14,6 +14,8 @@
 
 //! test endpoint index settings
 
+#![recursion_limit = "256"]
+
 use std::net::SocketAddr;
 use tempfile::TempDir;
 use tokio_util::sync::CancellationToken;

crates/heal/tests/heal_b5_versioned_regression_test.rs

@@ -22,6 +22,8 @@
 //! bucket-metadata-sys OnceCell) — under `cargo nextest` each test runs
 //! in its own process so the OnceCell never collides.
 
+#![recursion_limit = "256"]
+
 use http::HeaderMap;
 use rustfs_common::heal_channel::{HealOpts, HealScanMode};
 use rustfs_heal::heal::{

crates/heal/tests/heal_b920_subquorum_union_test.rs

@@ -21,6 +21,8 @@
 //! These drive the REAL `ECStoreHealStorage` + `ECStore` against real disks.
 //! Every test is `#[serial]`; under `cargo nextest` each runs in its own process.
 
+#![recursion_limit = "256"]
+
 use http::HeaderMap;
 use rustfs_common::heal_channel::{HealOpts, HealScanMode};
 use rustfs_heal::heal::storage::{

crates/heal/tests/heal_integration_test.rs

@@ -12,6 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+#![recursion_limit = "256"]
+
 use http::HeaderMap;
 use rustfs_common::heal_channel::{HealOpts, HealScanMode};
 use rustfs_heal::heal::{

crates/io-metrics/src/lib.rs

@@ -812,6 +812,17 @@ pub fn record_get_object_metadata_fanout_shape(path: &'static str, total: usize,
         .record(metadata_fanout_count_to_f64(non_valid));
 }
 
+/// Record task lifecycle shape for one GetObject metadata fanout.
+#[inline(always)]
+pub fn record_get_object_metadata_fanout_lifecycle(path: &'static str, scheduled: usize, completed: usize, cancelled: usize) {
+    if !get_stage_metrics_enabled() {
+        return;
+    }
+    histogram!("rustfs_io_get_object_metadata_fanout_scheduled", "path" => path).record(metadata_fanout_count_to_f64(scheduled));
+    histogram!("rustfs_io_get_object_metadata_fanout_completed", "path" => path).record(metadata_fanout_count_to_f64(completed));
+    histogram!("rustfs_io_get_object_metadata_fanout_cancelled", "path" => path).record(metadata_fanout_count_to_f64(cancelled));
+}
+
 /// Record a guarded metadata early-stop hit for GetObject.
 #[inline(always)]
 pub fn record_get_object_metadata_early_stop_hit(path: &'static str, reason: &'static str) {
@@ -2692,12 +2703,17 @@ mod tests {
         record_get_object_reader_prefetch_wait("codec_streaming", 0.0002);
         record_get_object_response_handoff("standard", "selected", 8192, 1024, 0.0001);
         record_get_object_metadata_fanout_duration("legacy_duplex", 0.001);
+        record_get_object_stage_duration("legacy_duplex", "read_version_path_resolve", 0.0001);
+        record_get_object_stage_duration("legacy_duplex", "read_version_path_check", 0.0001);
+        record_get_object_stage_duration("legacy_duplex", "read_version_xlmeta_read", 0.0005);
+        record_get_object_stage_duration("legacy_duplex", "read_version_decode", 0.0002);
         record_get_object_first_metadata_response_latency("legacy_duplex", 0.001);
         record_get_object_first_valid_metadata_response_latency("legacy_duplex", 0.001);
         record_get_object_slowest_metadata_response_latency("legacy_duplex", 0.003);
         record_get_object_quorum_reached_latency("legacy_duplex", 0.002);
         record_get_object_metadata_response("legacy_duplex", "valid");
         record_get_object_metadata_fanout_shape("legacy_duplex", 4, 3, 1, 1);
+        record_get_object_metadata_fanout_lifecycle("legacy_duplex", 4, 3, 1);
         record_get_object_metadata_early_stop_hit("legacy_duplex", "valid_quorum");
         record_get_object_metadata_early_stop_miss("legacy_duplex", "insufficient_quorum");
         record_get_object_metadata_early_stop_saved_responses("legacy_duplex", 1);
@@ -2768,6 +2784,38 @@ mod tests {
         assert!(remote_scheduled >= remote_avoid_potential);
     }
 
+    #[test]
+    fn metadata_fanout_lifecycle_records_named_histograms() {
+        let _guard = METRICS_FLAG_LOCK.lock().unwrap_or_else(|e| e.into_inner());
+        let recorder = DebuggingRecorder::new();
+        let snapshotter = recorder.snapshotter();
+
+        metrics::with_local_recorder(&recorder, || {
+            set_get_stage_metrics_enabled(true);
+            record_get_object_metadata_fanout_lifecycle("legacy_duplex", 4, 3, 1);
+            set_get_stage_metrics_enabled(false);
+        });
+
+        let metrics = snapshotter.snapshot().into_vec();
+        for (name, expected) in [
+            ("rustfs_io_get_object_metadata_fanout_scheduled", 4.0),
+            ("rustfs_io_get_object_metadata_fanout_completed", 3.0),
+            ("rustfs_io_get_object_metadata_fanout_cancelled", 1.0),
+        ] {
+            let value = metrics.iter().find_map(|(composite, _, _, value)| {
+                let has_path = composite
+                    .key()
+                    .labels()
+                    .any(|label| label.key() == "path" && label.value() == "legacy_duplex");
+                (composite.kind() == MetricKind::Histogram && composite.key().name() == name && has_path).then_some(value)
+            });
+            assert!(
+                matches!(value, Some(DebugValue::Histogram(values)) if values.len() == 1 && values[0].0 == expected),
+                "{name} must record the exact fanout lifecycle sample"
+            );
+        }
+    }
+
     #[test]
     fn test_record_get_object_fill_metrics() {
         record_get_object_fill_queued("codec_streaming", "single_inflight", 1);

crates/kms/src/config.rs

@@ -41,6 +41,7 @@ pub const ENV_KMS_AWS_ENDPOINT_URL: &str = "RUSTFS_KMS_AWS_ENDPOINT_URL";
 /// unset leaves rotation readiness unreported. Read once when the manager is
 /// built, by [`crate::manager::KmsManager`].
 pub const ENV_KMS_ROTATION_MAX_AGE_SECS: &str = "RUSTFS_KMS_ROTATION_MAX_AGE_SECS";
+pub const ENV_KMS_ROTATION_MAX_WRAPS: &str = "RUSTFS_KMS_ROTATION_MAX_WRAPS";
 pub const DEFAULT_VAULT_TRANSIT_METADATA_KV_MOUNT: &str = "secret";
 pub const DEFAULT_VAULT_TRANSIT_METADATA_KEY_PREFIX: &str = "rustfs/kms/transit-metadata";
 pub const DEFAULT_VAULT_APPROLE_MOUNT: &str = "approle";

crates/kms/src/manager.rs

@@ -17,7 +17,7 @@
 use crate::audit::{KmsAuditOperation, KmsAuditRecord, KmsAuditSink};
 use crate::backends::KmsBackend;
 use crate::cache::{KmsCache, KmsCacheStats};
-use crate::config::{ENV_KMS_ALLOW_IMMEDIATE_DELETION, ENV_KMS_ROTATION_MAX_AGE_SECS, KmsConfig};
+use crate::config::{ENV_KMS_ALLOW_IMMEDIATE_DELETION, ENV_KMS_ROTATION_MAX_AGE_SECS, ENV_KMS_ROTATION_MAX_WRAPS, KmsConfig};
 use crate::deletion_worker::DeletionReferenceChecker;
 use crate::error::{KmsError, Result};
 use crate::types::{
@@ -42,6 +42,13 @@ use tracing::warn;
 /// after it was rotated, which trains operators to ignore the signal.
 const MIN_ROTATION_MAX_AGE: Duration = Duration::from_secs(3600);
 
+/// Smallest wrap budget that can be configured.
+///
+/// Wraps are accounted in reserved blocks, so any threshold below one block
+/// would be crossed by a single reservation and report a key that has barely
+/// wrapped anything as overdue.
+const MIN_ROTATION_MAX_WRAPS: u64 = 1_000_000;
+
 /// Rotation age from the environment, or `None` when the signal is off.
 ///
 /// Unset leaves it off rather than guessing a policy: how often a deployment
@@ -68,6 +75,33 @@ fn parse_rotation_max_age(value: Option<&str>) -> Option<Duration> {
     Some(Duration::from_secs(seconds).max(MIN_ROTATION_MAX_AGE))
 }
 
+/// Wrap budget from the environment, or `None` when the signal is off.
+///
+/// Same discipline as the age threshold: unset means unreported rather than a
+/// guessed policy, and an unparsable value is refused loudly instead of
+/// falling back to a number the operator did not write. Clamped to
+/// [`MIN_ROTATION_MAX_WRAPS`] because the backend accounts for wraps in
+/// reserved blocks, so a threshold below one block would trip on the first
+/// reservation regardless of how many wraps actually happened.
+fn configured_rotation_max_wraps() -> Option<u64> {
+    parse_rotation_max_wraps(std::env::var(ENV_KMS_ROTATION_MAX_WRAPS).ok().as_deref())
+}
+
+fn parse_rotation_max_wraps(value: Option<&str>) -> Option<u64> {
+    let value = value?;
+    let Ok(wraps) = value.trim().parse::<u64>() else {
+        warn!(
+            variable = ENV_KMS_ROTATION_MAX_WRAPS,
+            "ignoring unparsable KMS rotation wrap budget; rotation readiness stays unreported"
+        );
+        return None;
+    };
+    if wraps == 0 {
+        return None;
+    }
+    Some(wraps.max(MIN_ROTATION_MAX_WRAPS))
+}
+
 #[derive(Clone)]
 pub struct KmsManager {
     backend: Arc<dyn KmsBackend>,
@@ -82,6 +116,7 @@ pub struct KmsManager {
     /// the verdict unreported. Read once at construction so a listing cannot
     /// change its answer halfway through.
     rotation_max_age: Option<Duration>,
+    rotation_max_wraps: Option<u64>,
 }
 
 impl KmsManager {
@@ -103,6 +138,7 @@ impl KmsManager {
             allow_immediate_deletion: config.allow_immediate_deletion,
             reference_checker: None,
             rotation_max_age: configured_rotation_max_age(),
+            rotation_max_wraps: configured_rotation_max_wraps(),
         }
     }
 
@@ -314,9 +350,22 @@ impl KmsManager {
             key.rotation_due_reason = Some(RotationDueReason::Unsupported);
             return;
         }
+        key.rotation_due = false;
+        key.rotation_due_reason = None;
+
+        // The wrap budget is checked first: it is the cryptographic bound (the
+        // AES-GCM random-nonce ceiling), whereas the age threshold is a policy
+        // choice, so when both are crossed the reason an operator most needs to
+        // see is the one they cannot negotiate.
+        if let (Some(max_wraps), Some(wraps)) = (self.rotation_max_wraps, key.wrap_budget_reserved)
+            && wraps >= max_wraps
+        {
+            key.rotation_due = true;
+            key.rotation_due_reason = Some(RotationDueReason::Wraps);
+            return;
+        }
+
         let Some(max_age) = self.rotation_max_age else {
-            key.rotation_due = false;
-            key.rotation_due_reason = None;
             return;
         };
 
@@ -333,9 +382,6 @@ impl KmsManager {
         if age >= max_age {
             key.rotation_due = true;
             key.rotation_due_reason = Some(reason);
-        } else {
-            key.rotation_due = false;
-            key.rotation_due_reason = None;
         }
     }
 
@@ -1685,10 +1731,15 @@ mod tests {
     }
 
     fn readiness_manager(rotation_max_age: Option<Duration>) -> KmsManager {
+        readiness_manager_with(rotation_max_age, None)
+    }
+
+    fn readiness_manager_with(rotation_max_age: Option<Duration>, rotation_max_wraps: Option<u64>) -> KmsManager {
         let temp_dir = tempfile::tempdir().expect("temp dir");
         let config = KmsConfig::local(temp_dir.path().to_path_buf()).with_insecure_development_defaults();
         let mut manager = KmsManager::new(Arc::new(ScriptedBackend::succeeding()), config);
         manager.rotation_max_age = rotation_max_age;
+        manager.rotation_max_wraps = rotation_max_wraps;
         manager
     }
 
@@ -1765,6 +1816,85 @@ mod tests {
         assert!(!key.rotation_due, "clock skew must not manufacture an overdue key");
     }
 
+    /// The wrap-budget half of the verdict: the cryptographic bound, checked
+    /// independently of the age policy and reported under its own reason.
+    #[test]
+    fn rotation_readiness_reports_an_exhausted_wrap_budget() {
+        let now = Zoned::now();
+        let recently = &now - jiff::Span::new().hours(1);
+        let long_ago = &now - jiff::Span::new().days(400);
+        let day = Duration::from_secs(86_400);
+        let budget = 2_000_000;
+
+        let with_wraps = |manager: &KmsManager, wraps: Option<u64>, rotated_at: Option<Zoned>| {
+            let mut key = aged_key(rotated_at, recently.clone());
+            key.wrap_budget_reserved = wraps;
+            manager.apply_rotation_readiness(&mut key, true, &now);
+            (key.rotation_due, key.rotation_due_reason)
+        };
+
+        // Budget configured and exceeded on a freshly rotated key: due, and the
+        // reason names the wrap budget rather than an age nobody crossed.
+        let manager = readiness_manager_with(Some(day), Some(budget));
+        assert_eq!(
+            with_wraps(&manager, Some(budget), Some(recently.clone())),
+            (true, Some(RotationDueReason::Wraps))
+        );
+        // At the threshold exactly, not only past it: the bound is a ceiling.
+        assert_eq!(
+            with_wraps(&manager, Some(budget + 1), Some(recently.clone())),
+            (true, Some(RotationDueReason::Wraps))
+        );
+        // Under the threshold: no verdict from the wrap half.
+        assert_eq!(with_wraps(&manager, Some(budget - 1), Some(recently.clone())), (false, None));
+
+        // The cryptographic bound outranks the policy one when both are crossed.
+        let mut key = aged_key(Some(long_ago.clone()), long_ago);
+        key.wrap_budget_reserved = Some(budget);
+        manager.apply_rotation_readiness(&mut key, true, &now);
+        assert_eq!(key.rotation_due_reason, Some(RotationDueReason::Wraps));
+
+        // No wrap threshold configured: an enormous count reports nothing, the
+        // same way an unset age threshold does.
+        let age_only = readiness_manager_with(Some(day), None);
+        assert_eq!(with_wraps(&age_only, Some(u64::MAX), Some(recently.clone())), (false, None));
+
+        // Backend reports no count (Transit, AWS, or a pre-accounting record):
+        // the wrap half stays silent instead of guessing, and the age half
+        // still decides.
+        let wraps_only = readiness_manager_with(None, Some(budget));
+        assert_eq!(with_wraps(&wraps_only, None, Some(recently.clone())), (false, None));
+        assert_eq!(
+            with_wraps(&wraps_only, Some(budget), Some(recently.clone())),
+            (true, Some(RotationDueReason::Wraps))
+        );
+
+        // A backend that cannot rotate is never told to, whatever it wrapped.
+        let mut key = aged_key(None, recently);
+        key.wrap_budget_reserved = Some(u64::MAX);
+        wraps_only.apply_rotation_readiness(&mut key, false, &now);
+        assert!(!key.rotation_due);
+        assert_eq!(key.rotation_due_reason, Some(RotationDueReason::Unsupported));
+    }
+
+    /// Threshold parsing matches the age threshold's discipline: unset and
+    /// unparsable both disable the signal rather than inventing a policy.
+    #[test]
+    fn rotation_wrap_threshold_parsing_refuses_to_guess() {
+        assert_eq!(parse_rotation_max_wraps(None), None);
+        assert_eq!(parse_rotation_max_wraps(Some("not-a-number")), None);
+        assert_eq!(parse_rotation_max_wraps(Some("")), None);
+        assert_eq!(parse_rotation_max_wraps(Some("-1")), None);
+        assert_eq!(parse_rotation_max_wraps(Some("0")), None);
+        // Clamped: below one reservation block the first reservation would trip it.
+        assert_eq!(parse_rotation_max_wraps(Some("1")), Some(MIN_ROTATION_MAX_WRAPS));
+        assert_eq!(
+            parse_rotation_max_wraps(Some(" 5000000 ")),
+            Some(5_000_000),
+            "a configured budget above the floor is honored verbatim"
+        );
+    }
+
     /// The two fields are additive on the wire: a payload written before they
     /// existed still deserializes, and a key with no verdict serializes exactly
     /// as it did before.

crates/kms/src/types.rs

@@ -217,6 +217,12 @@ pub enum RotationDueReason {
     /// The key has never been rotated and has existed longer than the
     /// configured maximum age.
     NeverRotated,
+    /// The key has wrapped more data keys than the configured maximum.
+    ///
+    /// Counted per key-material version, so a rotation restarts the budget.
+    /// The count is an over-estimate by construction (see the backend's
+    /// reservation accounting), so this verdict errs toward rotating early.
+    Wraps,
     /// The backend cannot rotate keys at all, so no age makes one due.
     Unsupported,
 }

crates/protocols/tests/swift_metadata_persistence.rs

@@ -23,6 +23,7 @@
 //! two are tested together because a reload is the only way to tell a real
 //! merge from one that happened to look right in the cache.
 
+#![recursion_limit = "256"]
 #![cfg(feature = "swift")]
 
 use std::collections::HashMap;

crates/rio/src/compress_reader.rs

@@ -71,7 +71,6 @@ where
 
     /// Optional: allow users to customize block_size
     pub fn with_block_size(inner: R, block_size: usize, compression_algorithm: CompressionAlgorithm) -> Self {
-        debug_assert!(block_size > 0, "CompressReader block_size must be non-zero");
         Self {
             inner,
             buffer: Vec::new(),
@@ -184,21 +183,11 @@ pin_project! {
         buffer: Vec<u8>,
         buffer_pos: usize,
         finished: bool,
-        // A previously surfaced stream error is sticky: without this, a caller
-        // that polls again after an error would restart at the header phase and
-        // read a truncated tail as a clean EOF, converting the error into a
-        // silently short body.
-        poisoned: bool,
         // Fields for saving header read progress across polls
         header_buf: [u8; 8],
         header_read: usize,
-        // Fields for saving compressed block read progress across polls.
-        // `compressed_len > 0` means a block payload is in flight: the header has
-        // been fully parsed and `compressed_read` bytes of the payload are already
-        // consumed from the inner stream. The header phase must not run again (and
-        // must not reset `compressed_read`) until this block completes, or a
-        // `Poll::Pending` in the middle of a payload would silently drop the bytes
-        // read so far and desynchronize the block framing.
+        header_done: bool,
+        // Fields for saving compressed block read progress across polls
         compressed_buf: Vec<u8>,
         compressed_read: usize,
         compressed_len: usize,
@@ -216,9 +205,9 @@ where
             buffer: Vec::new(),
             buffer_pos: 0,
             finished: false,
-            poisoned: false,
             header_buf: [0u8; 8],
             header_read: 0,
+            header_done: false,
             compressed_buf: Vec::new(),
             compressed_read: 0,
             compressed_len: 0,
@@ -247,72 +236,54 @@ where
         if *this.finished {
             return Poll::Ready(Ok(()));
         }
-        if *this.poisoned {
-            *this.poisoned = true;
-            return Poll::Ready(Err(io::Error::new(io::ErrorKind::InvalidData, "decompress reader previously failed")));
-        }
-
-        if *this.compressed_len == 0 {
-            // Read the 8-byte block header, resuming across polls via `header_read`.
-            while *this.header_read < HEADER_LEN {
-                let mut temp = [0u8; HEADER_LEN];
-                let mut temp_buf = ReadBuf::new(&mut temp[0..HEADER_LEN - *this.header_read]);
-                match this.inner.as_mut().poll_read(cx, &mut temp_buf) {
-                    Poll::Pending => return Poll::Pending,
-                    Poll::Ready(Ok(())) => {
-                        let n = temp_buf.filled().len();
-                        if n == 0 {
-                            if *this.header_read == 0 {
-                                // Clean EOF on a block boundary.
-                                *this.finished = true;
-                                return Poll::Ready(Ok(()));
-                            }
-                            *this.poisoned = true;
-                            return Poll::Ready(Err(io::Error::new(
-                                io::ErrorKind::UnexpectedEof,
-                                "unexpected EOF while reading compressed block header",
-                            )));
-                        }
-                        this.header_buf[*this.header_read..*this.header_read + n].copy_from_slice(&temp_buf.filled()[..n]);
-                        *this.header_read += n;
-                    }
-                    Poll::Ready(Err(e)) => {
-                        // error!("DecompressReader poll_read: read header error: {e}");
-                        *this.poisoned = true;
-                        return Poll::Ready(Err(e));
+        // Read header
+        while !*this.header_done && *this.header_read < HEADER_LEN {
+            let mut temp = [0u8; HEADER_LEN];
+            let mut temp_buf = ReadBuf::new(&mut temp[0..HEADER_LEN - *this.header_read]);
+            match this.inner.as_mut().poll_read(cx, &mut temp_buf) {
+                Poll::Pending => return Poll::Pending,
+                Poll::Ready(Ok(())) => {
+                    let n = temp_buf.filled().len();
+                    if n == 0 {
+                        break;
                     }
+                    this.header_buf[*this.header_read..*this.header_read + n].copy_from_slice(&temp_buf.filled()[..n]);
+                    *this.header_read += n;
+                }
+                Poll::Ready(Err(e)) => {
+                    // error!("DecompressReader poll_read: read header error: {e}");
+                    return Poll::Ready(Err(e));
                 }
             }
+            if *this.header_read < HEADER_LEN {
+                return Poll::Pending;
+            }
+        }
+        if !*this.header_done && *this.header_read == 0 {
+            return Poll::Ready(Ok(()));
+        }
+        let typ = this.header_buf[0];
+        let len = (this.header_buf[1] as usize) | ((this.header_buf[2] as usize) << 8) | ((this.header_buf[3] as usize) << 16);
+        let crc = (this.header_buf[4] as u32)
+            | ((this.header_buf[5] as u32) << 8)
+            | ((this.header_buf[6] as u32) << 16)
+            | ((this.header_buf[7] as u32) << 24);
+        *this.header_read = 0;
+        *this.header_done = true;
 
-            let typ = this.header_buf[0];
-            let len =
-                (this.header_buf[1] as usize) | ((this.header_buf[2] as usize) << 8) | ((this.header_buf[3] as usize) << 16);
-            *this.header_read = 0;
-
-            if typ == COMPRESS_TYPE_END {
-                *this.compressed_read = 0;
-                *this.compressed_len = 0;
-                *this.finished = true;
-                return Poll::Ready(Ok(()));
-            }
-            if typ != COMPRESS_TYPE_COMPRESSED && typ != COMPRESS_TYPE_UNCOMPRESSED {
-                // error!("DecompressReader unknown compression type: {typ}");
-                *this.poisoned = true;
-                return Poll::Ready(Err(io::Error::new(io::ErrorKind::InvalidData, "Unknown compression type")));
-            }
-            if len == 0 {
-                *this.poisoned = true;
-                return Poll::Ready(Err(io::Error::new(io::ErrorKind::InvalidData, "Invalid compressed block length")));
-            }
-
-            if this.compressed_buf.len() < len {
-                this.compressed_buf.resize(len, 0);
-            }
-            *this.compressed_len = len;
+        if typ == COMPRESS_TYPE_END {
             *this.compressed_read = 0;
+            *this.compressed_len = 0;
+            *this.finished = true;
+            return Poll::Ready(Ok(()));
         }
 
-        // Fill the in-flight block payload, resuming across polls via `compressed_read`.
+        if this.compressed_buf.len() < len {
+            this.compressed_buf.resize(len, 0);
+        }
+        *this.compressed_len = len;
+        *this.compressed_read = 0;
+
         while *this.compressed_read < *this.compressed_len {
             let mut temp_buf = ReadBuf::new(&mut this.compressed_buf[*this.compressed_read..*this.compressed_len]);
             match this.inner.as_mut().poll_read(cx, &mut temp_buf) {
@@ -320,13 +291,7 @@ where
                 Poll::Ready(Ok(())) => {
                     let n = temp_buf.filled().len();
                     if n == 0 {
-                        *this.compressed_read = 0;
-                        *this.compressed_len = 0;
-                        *this.poisoned = true;
-                        return Poll::Ready(Err(io::Error::new(
-                            io::ErrorKind::UnexpectedEof,
-                            "unexpected EOF while reading compressed block payload",
-                        )));
+                        break;
                     }
                     *this.compressed_read += n;
                 }
@@ -334,17 +299,10 @@ where
                     // error!("DecompressReader poll_read: read compressed block error: {e}");
                     *this.compressed_read = 0;
                     *this.compressed_len = 0;
-                    *this.poisoned = true;
                     return Poll::Ready(Err(e));
                 }
             }
         }
-
-        let typ = this.header_buf[0];
-        let crc = (this.header_buf[4] as u32)
-            | ((this.header_buf[5] as u32) << 8)
-            | ((this.header_buf[6] as u32) << 16)
-            | ((this.header_buf[7] as u32) << 24);
         let compressed_buf = &this.compressed_buf[..*this.compressed_len];
         // `compressed_buf`'s length comes from the untrusted 24-bit header length field, so it
         // can be shorter than 16 bytes. `uvarint` is safe on any slice length (reads at most 10
@@ -358,7 +316,6 @@ where
         if uvarint <= 0 || uvarint as usize > compressed_buf.len() {
             *this.compressed_read = 0;
             *this.compressed_len = 0;
-            *this.poisoned = true;
             return Poll::Ready(Err(io::Error::new(io::ErrorKind::InvalidData, "Invalid compressed block length prefix")));
         }
         let compressed_data = &compressed_buf[uvarint as usize..];
@@ -369,29 +326,21 @@ where
                     // error!("DecompressReader decompress_block error: {e}");
                     *this.compressed_read = 0;
                     *this.compressed_len = 0;
-                    *this.poisoned = true;
                     return Poll::Ready(Err(e));
                 }
             }
-        } else {
-            // The header phase already rejected every type other than
-            // COMPRESS_TYPE_COMPRESSED / COMPRESS_TYPE_UNCOMPRESSED.
+        } else if typ == COMPRESS_TYPE_UNCOMPRESSED {
             compressed_data.to_vec()
-        };
-        if decompressed.is_empty() {
-            // The writer never emits zero-length plaintext blocks; an empty
-            // decode surfacing as Ready(Ok) with no bytes would read as EOF and
-            // silently truncate the stream.
-            *this.poisoned = true;
+        } else {
+            // error!("DecompressReader unknown compression type: {typ}");
             *this.compressed_read = 0;
             *this.compressed_len = 0;
-            return Poll::Ready(Err(io::Error::new(io::ErrorKind::InvalidData, "Empty compressed block")));
-        }
+            return Poll::Ready(Err(io::Error::new(io::ErrorKind::InvalidData, "Unknown compression type")));
+        };
         if decompressed.len() != uncompress_len as usize {
             // error!("DecompressReader decompressed length mismatch: {} != {}", decompressed.len(), uncompress_len);
             *this.compressed_read = 0;
             *this.compressed_len = 0;
-            *this.poisoned = true;
             return Poll::Ready(Err(io::Error::new(io::ErrorKind::InvalidData, "Decompressed length mismatch")));
         }
         let actual_crc = {
@@ -403,13 +352,13 @@ where
             // error!("DecompressReader CRC32 mismatch: actual {actual_crc} != expected {crc}");
             *this.compressed_read = 0;
             *this.compressed_len = 0;
-            *this.poisoned = true;
             return Poll::Ready(Err(io::Error::new(io::ErrorKind::InvalidData, "CRC32 mismatch")));
         }
         *this.buffer = decompressed;
         *this.buffer_pos = 0;
         *this.compressed_read = 0;
         *this.compressed_len = 0;
+        *this.header_done = false;
         let to_copy = min(buf.remaining(), this.buffer.len());
         buf.put_slice(&this.buffer[..to_copy]);
         *this.buffer_pos += to_copy;
@@ -544,184 +493,6 @@ mod tests {
         assert_eq!(&decompressed, &data);
     }
 
-    /// Wraps a reader so every other poll returns `Poll::Pending` and every
-    /// `Ready` poll serves at most `chunk` bytes. This is the shape a duplex
-    /// pipe produces when the erasure writer is slower than the decoder, which
-    /// is exactly what desynchronized the block framing before the resumable
-    /// payload state was added (rustfs/rustfs#5957 multipart GET truncation).
-    struct PendingChunkReader<R> {
-        inner: R,
-        chunk: usize,
-        pending_next: bool,
-    }
-
-    impl<R> PendingChunkReader<R> {
-        fn new(inner: R, chunk: usize) -> Self {
-            Self {
-                inner,
-                chunk,
-                pending_next: true,
-            }
-        }
-    }
-
-    impl<R: AsyncRead + Unpin> AsyncRead for PendingChunkReader<R> {
-        fn poll_read(
-            mut self: std::pin::Pin<&mut Self>,
-            cx: &mut std::task::Context<'_>,
-            buf: &mut tokio::io::ReadBuf<'_>,
-        ) -> std::task::Poll<std::io::Result<()>> {
-            if self.pending_next {
-                self.pending_next = false;
-                cx.waker().wake_by_ref();
-                return std::task::Poll::Pending;
-            }
-            self.pending_next = true;
-            let cap = self.chunk.min(buf.remaining());
-            let mut scratch = vec![0u8; cap];
-            let mut inner_buf = tokio::io::ReadBuf::new(&mut scratch);
-            match std::pin::Pin::new(&mut self.inner).poll_read(cx, &mut inner_buf) {
-                std::task::Poll::Ready(Ok(())) => {
-                    buf.put_slice(inner_buf.filled());
-                    std::task::Poll::Ready(Ok(()))
-                }
-                other => other,
-            }
-        }
-    }
-
-    fn patterned_payload(size: usize, seed: u8) -> Vec<u8> {
-        (0..size)
-            .map(|i| ((i as u64).wrapping_mul(2_654_435_761).wrapping_add(seed as u64) >> 3) as u8)
-            .collect()
-    }
-
-    /// Root-cause regression for the multipart compressed GET truncation: a
-    /// `Poll::Pending` in the middle of a block payload must not drop the bytes
-    /// already consumed. Before the resumable payload state, the decoder reset
-    /// `compressed_read` on every re-poll and surfaced
-    /// `LZ4 error: ERROR_frameType_unknown` mid-stream.
-    #[tokio::test]
-    async fn test_decompress_reader_survives_pending_mid_payload() {
-        let data = patterned_payload(100 * 1024, 7);
-        let mut compress_reader =
-            CompressReader::with_block_size(Cursor::new(data.clone()), 8192, CompressionAlgorithm::default());
-        let mut compressed = Vec::new();
-        compress_reader.read_to_end(&mut compressed).await.unwrap();
-
-        for chunk in [1usize, 3, 7, 8, 17, 1000, 8192] {
-            let inner = PendingChunkReader::new(Cursor::new(compressed.clone()), chunk);
-            let mut decompress_reader = DecompressReader::new(inner, CompressionAlgorithm::default());
-            let mut decompressed = Vec::new();
-            decompress_reader.read_to_end(&mut decompressed).await.unwrap();
-            assert_eq!(decompressed, data, "pending-chunked decode must be byte-exact for chunk={chunk}");
-        }
-    }
-
-    /// Two independently compressed streams concatenated back to back — the
-    /// on-disk shape of a compressed multipart object — must decode across the
-    /// stream boundary even when every poll can suspend mid-block.
-    #[tokio::test]
-    async fn test_decompress_reader_survives_pending_across_concatenated_streams() {
-        let part1 = patterned_payload(64 * 1024, 7);
-        let part2 = patterned_payload(24 * 1024, 61);
-
-        let mut stored = Vec::new();
-        for part in [&part1, &part2] {
-            let mut compress_reader =
-                CompressReader::with_block_size(Cursor::new(part.clone()), 8192, CompressionAlgorithm::default());
-            let mut compressed = Vec::new();
-            compress_reader.read_to_end(&mut compressed).await.unwrap();
-            stored.extend_from_slice(&compressed);
-        }
-
-        let mut expected = part1;
-        expected.extend_from_slice(&part2);
-
-        for chunk in [1usize, 5, 8, 13, 4096] {
-            let inner = PendingChunkReader::new(Cursor::new(stored.clone()), chunk);
-            let mut decompress_reader = DecompressReader::new(inner, CompressionAlgorithm::default());
-            let mut decompressed = Vec::new();
-            decompress_reader.read_to_end(&mut decompressed).await.unwrap();
-            assert_eq!(
-                decompressed, expected,
-                "concatenated part streams must decode byte-exact for chunk={chunk}"
-            );
-        }
-    }
-
-    /// After the first stream error, every further poll must keep failing.
-    /// Without the sticky poison a retrying caller would restart at the header
-    /// phase and read the truncated tail as a clean EOF — converting a hard
-    /// error into a silently short body.
-    #[tokio::test]
-    async fn test_decompress_reader_error_is_sticky() {
-        let data = patterned_payload(32 * 1024, 7);
-        let mut compress_reader = CompressReader::with_block_size(Cursor::new(data), 8192, CompressionAlgorithm::default());
-        let mut compressed = Vec::new();
-        compress_reader.read_to_end(&mut compressed).await.unwrap();
-        compressed.truncate(compressed.len() - 3);
-
-        let mut decompress_reader = DecompressReader::new(Cursor::new(compressed), CompressionAlgorithm::default());
-        let mut out = Vec::new();
-        let first = decompress_reader
-            .read_to_end(&mut out)
-            .await
-            .expect_err("truncated payload must error");
-        assert_eq!(first.kind(), std::io::ErrorKind::UnexpectedEof);
-
-        let mut retry = Vec::new();
-        let second = decompress_reader
-            .read_to_end(&mut retry)
-            .await
-            .expect_err("a poll after the first error must not turn into a clean EOF");
-        assert_eq!(second.kind(), std::io::ErrorKind::InvalidData);
-        assert!(retry.is_empty(), "no bytes may be produced after the stream failed");
-    }
-
-    /// A stream cut off in the middle of a block payload must fail with a clean
-    /// UnexpectedEof instead of decoding a short buffer.
-    #[tokio::test]
-    async fn test_decompress_reader_truncated_payload_is_unexpected_eof() {
-        let data = patterned_payload(32 * 1024, 7);
-        let mut compress_reader = CompressReader::with_block_size(Cursor::new(data), 8192, CompressionAlgorithm::default());
-        let mut compressed = Vec::new();
-        compress_reader.read_to_end(&mut compressed).await.unwrap();
-
-        compressed.truncate(compressed.len() - 3);
-        let mut decompress_reader = DecompressReader::new(Cursor::new(compressed), CompressionAlgorithm::default());
-        let mut out = Vec::new();
-        let err = decompress_reader
-            .read_to_end(&mut out)
-            .await
-            .expect_err("truncated payload must error");
-        assert_eq!(err.kind(), std::io::ErrorKind::UnexpectedEof);
-    }
-
-    /// A stream cut off in the middle of a block header must fail with a clean
-    /// UnexpectedEof instead of parsing a garbage header.
-    #[tokio::test]
-    async fn test_decompress_reader_truncated_header_is_unexpected_eof() {
-        let data = patterned_payload(12 * 1024, 7);
-        let mut compress_reader = CompressReader::with_block_size(Cursor::new(data), 8192, CompressionAlgorithm::default());
-        let mut compressed = Vec::new();
-        compress_reader.read_to_end(&mut compressed).await.unwrap();
-
-        // Keep the first full block plus 3 bytes of the next header.
-        let ln = (compressed[1] as usize) | ((compressed[2] as usize) << 8) | ((compressed[3] as usize) << 16);
-        let first_block_end = 8 + ln;
-        assert!(compressed.len() > first_block_end, "fixture must contain more than one block");
-        compressed.truncate(first_block_end + 3);
-
-        let mut decompress_reader = DecompressReader::new(Cursor::new(compressed), CompressionAlgorithm::default());
-        let mut out = Vec::new();
-        let err = decompress_reader
-            .read_to_end(&mut out)
-            .await
-            .expect_err("truncated header must error");
-        assert_eq!(err.kind(), std::io::ErrorKind::UnexpectedEof);
-    }
-
     // Regression: a corrupted block whose 24-bit length field is < 16 must not panic.
     // Header layout (HEADER_LEN = 8): [type, len_lo, len_mid, len_hi, crc0..crc3], then `len`
     // bytes of block body. Pre-fix, poll_read sliced `compressed_buf[0..16]` unconditionally,
@@ -747,85 +518,6 @@ mod tests {
         assert_eq!(res.unwrap_err().kind(), std::io::ErrorKind::InvalidData);
     }
 
-    // Header-level fail-closed matrix, built by hand so the decoder is exercised against bytes no
-    // encoder in this crate can produce. Header layout (HEADER_LEN = 8):
-    // [type, len_lo, len_mid, len_hi, crc0..crc3], then `len` body bytes = uvarint(plain_len) + data.
-    #[tokio::test]
-    async fn test_decompress_reader_header_validation_matrix() {
-        // Build a block whose body is `uvarint(plain.len()) + plain` (i.e. the
-        // COMPRESS_TYPE_UNCOMPRESSED shape), with the header CRC taken over the plaintext exactly
-        // like the production writer does.
-        fn build_raw_block(typ: u8, plain: &[u8], len_override: Option<usize>) -> Vec<u8> {
-            let crc = {
-                let mut hasher = crc_fast::Digest::new(crc_fast::CrcAlgorithm::Crc32IsoHdlc);
-                hasher.update(plain);
-                hasher.finalize() as u32
-            };
-            let mut uvarint_buf = [0u8; 10];
-            let int_len = put_uvarint(&mut uvarint_buf[..], plain.len() as u64);
-            let body_len = int_len + plain.len();
-            let len = len_override.unwrap_or(body_len);
-
-            let mut out = Vec::with_capacity(HEADER_LEN + body_len);
-            out.push(typ);
-            out.push((len & 0xFF) as u8);
-            out.push(((len >> 8) & 0xFF) as u8);
-            out.push(((len >> 16) & 0xFF) as u8);
-            out.extend_from_slice(&crc.to_le_bytes());
-            out.extend_from_slice(&uvarint_buf[..int_len]);
-            out.extend_from_slice(plain);
-            out
-        }
-
-        let plain = b"uncompressed passthrough payload";
-
-        // (a) A well-formed uncompressed block decodes to the plaintext verbatim.
-        let mut out = Vec::new();
-        DecompressReader::new(
-            Cursor::new(build_raw_block(COMPRESS_TYPE_UNCOMPRESSED, plain, None)),
-            CompressionAlgorithm::default(),
-        )
-        .read_to_end(&mut out)
-        .await
-        .expect("a well-formed uncompressed block must decode");
-        assert_eq!(out.as_slice(), plain.as_slice());
-
-        // (b) An unknown block type must be rejected instead of being treated as passthrough.
-        let mut out = Vec::new();
-        let err = DecompressReader::new(Cursor::new(build_raw_block(0x7E, plain, None)), CompressionAlgorithm::default())
-            .read_to_end(&mut out)
-            .await
-            .expect_err("unknown compression type must error");
-        assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
-        assert!(err.to_string().contains("Unknown compression type"), "got: {err}");
-
-        // (c) A zero-length block would stall the decoder, so it must be rejected up front.
-        let mut out = Vec::new();
-        let err = DecompressReader::new(
-            Cursor::new(build_raw_block(COMPRESS_TYPE_UNCOMPRESSED, plain, Some(0))),
-            CompressionAlgorithm::default(),
-        )
-        .read_to_end(&mut out)
-        .await
-        .expect_err("zero-length block must error");
-        assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
-        assert!(err.to_string().contains("Invalid compressed block length"), "got: {err}");
-
-        // (d) A block that decodes to zero plaintext bytes must be rejected: the
-        // writer never emits empty blocks, and an empty decode surfacing as
-        // Ready(Ok) with no bytes would read as EOF and silently truncate.
-        let mut out = Vec::new();
-        let err = DecompressReader::new(
-            Cursor::new(build_raw_block(COMPRESS_TYPE_UNCOMPRESSED, b"", None)),
-            CompressionAlgorithm::default(),
-        )
-        .read_to_end(&mut out)
-        .await
-        .expect_err("empty block must error");
-        assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
-        assert!(err.to_string().contains("Empty compressed block"), "got: {err}");
-    }
-
     // Directly exercises the length-prefix guard: an unterminated varint (all continuation bytes)
     // makes `uvarint` return 0, which must be rejected as an invalid length prefix.
     #[tokio::test]

crates/scanner/src/lib.rs

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+#![recursion_limit = "256"]
 #![cfg_attr(docsrs, feature(doc_auto_cfg))]
 #![warn(
     // missing_docs,

crates/scanner/tests/lifecycle_integration_test.rs

@@ -12,6 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+#![recursion_limit = "256"]
+
 use futures::FutureExt;
 use rustfs_config::ENV_TEST_FORCE_IMMEDIATE_TRANSITION_ENQUEUE_TIMEOUT;
 use rustfs_scanner::scanner_folder::ScannerItem;

crates/utils/src/http/object_encryption_keys.rs

@@ -25,7 +25,8 @@
 // The lowercase stored forms, matching exactly what encryption_material_to_metadata
 // persists. The read-path SSE-C check is case-sensitive, so restoring under any
 // other casing would classify the replica as managed-SSE and reject SSE-C GETs.
-use super::headers::{SSEC_ALGORITHM_HEADER, SSEC_KEY_HEADER, SSEC_KEY_MD5_HEADER};
+use super::headers::{AMZ_ENCRYPTION_KMS, SSEC_ALGORITHM_HEADER, SSEC_KEY_HEADER, SSEC_KEY_MD5_HEADER};
+use std::collections::HashMap;
 
 pub const INTERNAL_ENCRYPTION_KEY_ID_HEADER: &str = "x-rustfs-encryption-key-id";
 pub const INTERNAL_ENCRYPTION_KEY_HEADER: &str = "x-rustfs-encryption-key";
@@ -165,6 +166,143 @@ pub fn is_replication_stripped_encryption_key(key: &str) -> bool {
         || super::starts_with_ignore_ascii_case(key, RUSTFS_INTERNAL_ENCRYPTION_PREFIX)
 }
 
+// ============================================================================
+// Managed-SSE attribution (shared classifier)
+// ============================================================================
+//
+// Single source of truth for classifying stored managed-SSE (SSE-S3 / SSE-KMS)
+// object metadata. These live here — rather than in the `rustfs` binary
+// crate's SSE module — so lower-layer consumers such as the scanner can
+// attribute encrypted objects without growing a second copy of the
+// normalization/classification logic (backlog#1643 PR-B0). The binary crate
+// re-exports them from `rustfs::storage::sse`, and a source-scan test there
+// pins that no second definition reappears.
+//
+// Every metadata lookup below is a case-SENSITIVE exact match on the stored
+// `HashMap<String, String>` keys, mirroring the SSE read path. Do not
+// "harmonize" these with the lowercase-normalizing helpers in
+// `header_compat.rs`: the lowercase `x-amz-*` stored forms and the TitleCase
+// MinIO-internal names are load-bearing exactly as written.
+
+/// Type of encryption used
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum SSEType {
+    /// SSE-S3 (AES256)
+    SseS3,
+    /// SSE-KMS (aws:kms)
+    SseKms,
+    /// SSE-C (customer-provided key)
+    SseC,
+}
+
+impl SSEType {
+    /// Stable scheme name for audit consumers.
+    pub fn audit_label(self) -> &'static str {
+        match self {
+            SSEType::SseS3 => "SSE-S3",
+            SSEType::SseKms => "SSE-KMS",
+            SSEType::SseC => "SSE-C",
+        }
+    }
+}
+
+/// Recodes a stored MinIO KMS context value — base64-wrapped JSON under
+/// [`MINIO_INTERNAL_ENCRYPTION_KMS_CONTEXT_HEADER`] — into the plain-JSON form
+/// RustFS stores under [`INTERNAL_ENCRYPTION_CONTEXT_HEADER`].
+///
+/// Injected by callers because this crate deliberately carries no JSON codec.
+/// Returning `None` skips the context mapping, matching the historical
+/// silent-skip on a value that fails to decode.
+pub type KmsContextRecoder = fn(&str) -> Option<String>;
+
+/// True when the stored metadata carries a managed-SSE (SSE-S3 / SSE-KMS)
+/// encryption envelope, under either the RustFS-branded or the MinIO-branded
+/// internal keys.
+pub fn contains_managed_encryption_metadata(metadata: &HashMap<String, String>) -> bool {
+    metadata.contains_key(INTERNAL_ENCRYPTION_KEY_HEADER)
+        || metadata.contains_key(MINIO_INTERNAL_ENCRYPTION_S3_SEALED_KEY_HEADER)
+        || metadata.contains_key(MINIO_INTERNAL_ENCRYPTION_KMS_SEALED_KEY_HEADER)
+        || metadata.contains_key(MINIO_INTERNAL_ENCRYPTION_KMS_DATA_KEY_HEADER)
+        || metadata.contains_key(MINIO_INTERNAL_ENCRYPTION_KMS_CONTEXT_HEADER)
+}
+
+/// Maps the MinIO-branded internal SSE keys onto the RustFS-branded stored
+/// keys (the dual internal metadata keys invariant). RustFS-branded keys
+/// already present always win; every source lookup is a case-sensitive exact
+/// match on the specific TitleCase MinIO names.
+pub fn normalize_managed_metadata(
+    metadata: &HashMap<String, String>,
+    recode_kms_context: Option<KmsContextRecoder>,
+) -> HashMap<String, String> {
+    let mut normalized = metadata.clone();
+
+    if !normalized.contains_key(INTERNAL_ENCRYPTION_KEY_HEADER)
+        && let Some(value) = metadata
+            .get(MINIO_INTERNAL_ENCRYPTION_KMS_DATA_KEY_HEADER)
+            .or_else(|| metadata.get(MINIO_INTERNAL_ENCRYPTION_KMS_SEALED_KEY_HEADER))
+            .or_else(|| metadata.get(MINIO_INTERNAL_ENCRYPTION_S3_SEALED_KEY_HEADER))
+            .or_else(|| metadata.get(MINIO_INTERNAL_ENCRYPTION_SSEC_SEALED_KEY_HEADER))
+    {
+        normalized.insert(INTERNAL_ENCRYPTION_KEY_HEADER.to_string(), value.clone());
+    }
+
+    if !normalized.contains_key(INTERNAL_ENCRYPTION_IV_HEADER)
+        && let Some(value) = metadata.get(MINIO_INTERNAL_ENCRYPTION_IV_HEADER)
+    {
+        normalized.insert(INTERNAL_ENCRYPTION_IV_HEADER.to_string(), value.clone());
+    }
+
+    if !normalized.contains_key(INTERNAL_ENCRYPTION_ALGORITHM_HEADER)
+        && let Some(value) = metadata.get(MINIO_INTERNAL_ENCRYPTION_ALGORITHM_HEADER)
+    {
+        normalized.insert(INTERNAL_ENCRYPTION_ALGORITHM_HEADER.to_string(), value.clone());
+    }
+
+    if !normalized.contains_key(INTERNAL_ENCRYPTION_KEY_ID_HEADER)
+        && let Some(value) = metadata.get(MINIO_INTERNAL_ENCRYPTION_KMS_KEY_ID_HEADER)
+    {
+        normalized.insert(INTERNAL_ENCRYPTION_KEY_ID_HEADER.to_string(), value.clone());
+    }
+
+    if !normalized.contains_key(INTERNAL_ENCRYPTION_CONTEXT_HEADER)
+        && let Some(value) = metadata.get(MINIO_INTERNAL_ENCRYPTION_KMS_CONTEXT_HEADER)
+        && let Some(recode) = recode_kms_context
+        && let Some(encoded) = recode(value)
+    {
+        normalized.insert(INTERNAL_ENCRYPTION_CONTEXT_HEADER.to_string(), encoded);
+    }
+
+    normalized
+}
+
+/// Resolve the scheme and KMS key a stored managed-SSE object was wrapped with.
+///
+/// Mirrors the lookup `apply_managed_decryption_material` performs, so both agree on
+/// which key a read is authorized against.
+///
+/// No [`KmsContextRecoder`] is taken: the context mapping only ever inserts
+/// [`INTERNAL_ENCRYPTION_CONTEXT_HEADER`], which this lookup never reads, so
+/// the result is identical with or without it.
+pub fn stored_managed_encryption_key(metadata: &HashMap<String, String>) -> Option<(SSEType, String)> {
+    if !contains_managed_encryption_metadata(metadata) {
+        return None;
+    }
+
+    // Case-sensitive: the SSE writer stores the scheme under the lowercase
+    // `x-amz-server-side-encryption` key; other casings are not stored forms.
+    let sse_type = match metadata.get("x-amz-server-side-encryption")?.as_str() {
+        AMZ_ENCRYPTION_KMS => SSEType::SseKms,
+        _ => SSEType::SseS3,
+    };
+    let key_id = normalize_managed_metadata(metadata, None)
+        .get(INTERNAL_ENCRYPTION_KEY_ID_HEADER)
+        .or_else(|| metadata.get("x-amz-server-side-encryption-aws-kms-key-id"))
+        .cloned()
+        .unwrap_or_else(|| "default".to_string());
+
+    Some((sse_type, key_id))
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -273,6 +411,134 @@ mod tests {
         assert!(!is_replication_stripped_encryption_key("content-type"));
     }
 
+    #[test]
+    fn managed_envelope_predicate_matches_both_key_families() {
+        assert!(!contains_managed_encryption_metadata(&HashMap::new()));
+
+        for key in [
+            INTERNAL_ENCRYPTION_KEY_HEADER,
+            MINIO_INTERNAL_ENCRYPTION_S3_SEALED_KEY_HEADER,
+            MINIO_INTERNAL_ENCRYPTION_KMS_SEALED_KEY_HEADER,
+            MINIO_INTERNAL_ENCRYPTION_KMS_DATA_KEY_HEADER,
+            MINIO_INTERNAL_ENCRYPTION_KMS_CONTEXT_HEADER,
+        ] {
+            let single = HashMap::from([(key.to_string(), "value".to_string())]);
+            assert!(contains_managed_encryption_metadata(&single), "{key} must classify as managed SSE");
+        }
+
+        // SSE-C material alone is not a managed envelope.
+        let ssec_only = HashMap::from([(SSEC_ALGORITHM_HEADER.to_string(), "AES256".to_string())]);
+        assert!(!contains_managed_encryption_metadata(&ssec_only));
+    }
+
+    #[test]
+    fn normalize_maps_minio_keys_onto_missing_rustfs_keys_only() {
+        let metadata = HashMap::from([
+            (MINIO_INTERNAL_ENCRYPTION_KMS_DATA_KEY_HEADER.to_string(), "minio-dek".to_string()),
+            (MINIO_INTERNAL_ENCRYPTION_IV_HEADER.to_string(), "minio-iv".to_string()),
+            (MINIO_INTERNAL_ENCRYPTION_ALGORITHM_HEADER.to_string(), "DAREv2-HMAC-SHA256".to_string()),
+            (MINIO_INTERNAL_ENCRYPTION_KMS_KEY_ID_HEADER.to_string(), "minio-key".to_string()),
+        ]);
+
+        let normalized = normalize_managed_metadata(&metadata, None);
+        assert_eq!(normalized.get(INTERNAL_ENCRYPTION_KEY_HEADER).map(String::as_str), Some("minio-dek"));
+        assert_eq!(normalized.get(INTERNAL_ENCRYPTION_IV_HEADER).map(String::as_str), Some("minio-iv"));
+        assert_eq!(
+            normalized.get(INTERNAL_ENCRYPTION_ALGORITHM_HEADER).map(String::as_str),
+            Some("DAREv2-HMAC-SHA256")
+        );
+        assert_eq!(normalized.get(INTERNAL_ENCRYPTION_KEY_ID_HEADER).map(String::as_str), Some("minio-key"));
+
+        // Existing RustFS-branded keys always win over the MinIO twins.
+        let mut both = metadata;
+        both.insert(INTERNAL_ENCRYPTION_KEY_ID_HEADER.to_string(), "rustfs-key".to_string());
+        assert_eq!(
+            normalize_managed_metadata(&both, None)
+                .get(INTERNAL_ENCRYPTION_KEY_ID_HEADER)
+                .map(String::as_str),
+            Some("rustfs-key")
+        );
+
+        // The mapping is a case-sensitive exact match on the TitleCase MinIO
+        // names; a lowercased twin must not normalize.
+        let lowercased = HashMap::from([(MINIO_INTERNAL_ENCRYPTION_KMS_KEY_ID_HEADER.to_lowercase(), "minio-key".to_string())]);
+        assert!(!normalize_managed_metadata(&lowercased, None).contains_key(INTERNAL_ENCRYPTION_KEY_ID_HEADER));
+    }
+
+    #[test]
+    fn normalize_recodes_kms_context_only_through_the_injected_codec() {
+        let metadata = HashMap::from([(MINIO_INTERNAL_ENCRYPTION_KMS_CONTEXT_HEADER.to_string(), "encoded-context".to_string())]);
+
+        // Without a codec the context stays unnormalized.
+        assert!(!normalize_managed_metadata(&metadata, None).contains_key(INTERNAL_ENCRYPTION_CONTEXT_HEADER));
+
+        // A codec that fails to decode also leaves it unnormalized.
+        fn reject(_value: &str) -> Option<String> {
+            None
+        }
+        assert!(!normalize_managed_metadata(&metadata, Some(reject)).contains_key(INTERNAL_ENCRYPTION_CONTEXT_HEADER));
+
+        fn recode(value: &str) -> Option<String> {
+            Some(format!("recoded:{value}"))
+        }
+        assert_eq!(
+            normalize_managed_metadata(&metadata, Some(recode))
+                .get(INTERNAL_ENCRYPTION_CONTEXT_HEADER)
+                .map(String::as_str),
+            Some("recoded:encoded-context")
+        );
+
+        // A stored RustFS context wins without invoking the codec.
+        let mut both = metadata;
+        both.insert(INTERNAL_ENCRYPTION_CONTEXT_HEADER.to_string(), "stored-context".to_string());
+        assert_eq!(
+            normalize_managed_metadata(&both, Some(recode))
+                .get(INTERNAL_ENCRYPTION_CONTEXT_HEADER)
+                .map(String::as_str),
+            Some("stored-context")
+        );
+    }
+
+    #[test]
+    fn stored_managed_encryption_key_attributes_scheme_and_key() {
+        // Plaintext metadata carries no managed envelope.
+        assert!(stored_managed_encryption_key(&HashMap::new()).is_none());
+
+        // A managed envelope without the stored SSE marker cannot be attributed.
+        let envelope_only = HashMap::from([(INTERNAL_ENCRYPTION_KEY_HEADER.to_string(), "dek".to_string())]);
+        assert!(stored_managed_encryption_key(&envelope_only).is_none());
+
+        // The stored SSE marker is the lowercase form; a TitleCase key is not
+        // a stored form and must not be recognized.
+        let mut titlecase = envelope_only.clone();
+        titlecase.insert("X-Amz-Server-Side-Encryption".to_string(), "aws:kms".to_string());
+        assert!(stored_managed_encryption_key(&titlecase).is_none());
+
+        let mut sse_s3 = envelope_only.clone();
+        sse_s3.insert("x-amz-server-side-encryption".to_string(), "AES256".to_string());
+        assert_eq!(stored_managed_encryption_key(&sse_s3), Some((SSEType::SseS3, "default".to_string())));
+
+        let mut sse_kms = envelope_only;
+        sse_kms.insert("x-amz-server-side-encryption".to_string(), "aws:kms".to_string());
+        assert_eq!(stored_managed_encryption_key(&sse_kms), Some((SSEType::SseKms, "default".to_string())));
+
+        // Key-id precedence: RustFS stored key id, then the MinIO twin, then
+        // the lowercase amz key id, then "default".
+        sse_kms.insert("x-amz-server-side-encryption-aws-kms-key-id".to_string(), "amz-key".to_string());
+        assert_eq!(stored_managed_encryption_key(&sse_kms), Some((SSEType::SseKms, "amz-key".to_string())));
+        sse_kms.insert(MINIO_INTERNAL_ENCRYPTION_KMS_KEY_ID_HEADER.to_string(), "minio-key".to_string());
+        assert_eq!(stored_managed_encryption_key(&sse_kms), Some((SSEType::SseKms, "minio-key".to_string())));
+        sse_kms.insert(INTERNAL_ENCRYPTION_KEY_ID_HEADER.to_string(), "rustfs-key".to_string());
+        assert_eq!(stored_managed_encryption_key(&sse_kms), Some((SSEType::SseKms, "rustfs-key".to_string())));
+    }
+
+    #[test]
+    fn sse_type_audit_labels_are_stable() {
+        assert_eq!(SSEType::SseS3.audit_label(), "SSE-S3");
+        assert_eq!(SSEType::SseKms.audit_label(), "SSE-KMS");
+        assert_eq!(SSEType::SseC.audit_label(), "SSE-C");
+    }
+
     #[test]
     fn transport_prefixes_cover_every_transport_value_key() {
         // Every transport key that carries material must match a redaction

docs/architecture/compat-cleanup-register.md

@@ -34,7 +34,6 @@ for later deletion.
 - `tonic-013-status-render` peer RPC failure classification: internode failures that reach a node only as text (a peer's error_info payload, a status flattened through format!) are classified by matching the rendering of an Unavailable gRPC status. Releases up to 1.0.0-alpha.38 shipped tonic 0.13, which rendered that status as "status: Unavailable, message: ..."; tonic 0.14 renders it as "code: 'The service is currently unavailable', message: ...". Both forms are matched so an older peer's relayed text still marks an unreachable peer offline. Remove the tonic 0.13 form after the minimum supported RustFS peer version ships tonic 0.14 or later.
 - `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.
-- `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 decompressor.
 
 ## Review Checklist
 

docs/operations/kms-backend-security.md

@@ -112,6 +112,8 @@ RustFS does not rotate keys on a schedule. There is no built-in rotation worker,
 
 Set `RUSTFS_KMS_ROTATION_MAX_AGE_SECS` to that period in whole seconds. Unset — the default — leaves the verdict unreported rather than assuming a policy: how often keys must be rotated is a compliance decision, and a built-in default would report keys as overdue against a rule nobody wrote. An unparsable value is treated the same way, with a warning, instead of silently falling back to a number the operator did not choose. Values below one hour are raised to one hour, because a threshold of seconds reports every key as overdue moments after it was rotated and teaches operators to ignore the signal.
 
+A second, independent threshold covers the cryptographic bound rather than the policy one. `RUSTFS_KMS_ROTATION_MAX_WRAPS` is the number of data keys one key's material may wrap before the verdict reports `rotation_due` with reason `wraps`. It follows the same discipline — unset or unparsable leaves the verdict unreported, and values below one million are raised to one million because wraps are accounted in reserved blocks of that size, so a smaller threshold would trip on the first reservation. Only backends where RustFS wraps locally and can rotate report a count (Vault KV2 today); Transit and AWS wrap externally and report none, so the wrap half stays silent there rather than guessing. When both thresholds are crossed the reported reason is `wraps`: the AES-GCM random-nonce ceiling is not negotiable, while the age period is a policy an operator chose.
+
 `GET /rustfs/admin/v3/kms/keys` then carries two additional fields per key:
 
 - `rotation_due` — whether the key has outlived the configured period.

docs/operations/kms-observability-runbook.md

@@ -204,7 +204,7 @@ Meaning: `rustfs_kms_oldest_key_rotation_age_seconds` — seconds since the leas
 
 Investigation:
 
-1. Find which keys are due. The gauge deliberately names no key — a per-key label would carry key identifiers into the metric stream — so read the per-key verdict from the listing: `GET /rustfs/admin/v3/kms/keys` carries `rotation_due` and `rotation_due_reason` (`age`, `never_rotated`, or `unsupported`) per key, computed against `RUSTFS_KMS_ROTATION_MAX_AGE_SECS`. The verdict appears only on the listing, not on single-key describe. If `RUSTFS_KMS_ROTATION_MAX_AGE_SECS` is unset, set it to your policy's rotation period so the per-key verdict and this alert agree on what "overdue" means.
+1. Find which keys are due. The gauge deliberately names no key — a per-key label would carry key identifiers into the metric stream — so read the per-key verdict from the listing: `GET /rustfs/admin/v3/kms/keys` carries `rotation_due` and `rotation_due_reason` (`age`, `never_rotated`, `wraps`, or `unsupported`) per key, computed against `RUSTFS_KMS_ROTATION_MAX_AGE_SECS` and `RUSTFS_KMS_ROTATION_MAX_WRAPS`. A `wraps` reason means the key's material has wrapped more data keys than the configured budget — the AES-GCM random-nonce ceiling rather than an age policy, so it is not satisfied by relaxing the age threshold. The verdict appears only on the listing, not on single-key describe. If `RUSTFS_KMS_ROTATION_MAX_AGE_SECS` is unset, set it to your policy's rotation period so the per-key verdict and this alert agree on what "overdue" means.
 2. If the reason is `unsupported`, the backend cannot rotate at all (Local, Static). There is no key-level response; the decision is a backend migration, and the wrap ceiling above is the reason it cannot be deferred forever. See the [rotation drivers and scheduling matrix](kms-backend-security.md#rotation-drivers-and-scheduling-per-backend).
 3. On a backend that can rotate, act per the driver matrix: on **Vault KV2**, check why your external rotation scheduler did not run (or set one up — RustFS deliberately ships none) and satisfy the [pre-rotation checklist](kms-backend-security.md#rotation-drivers-and-scheduling-per-backend) before rotating, above all the [upgrade-ordering hard constraint](kms-backend-security.md#upgrade-before-first-rotation-hard-constraint) — never respond to this alert by rotating in the middle of a rolling upgrade. On **Vault Transit**, check `auto_rotate_period` on the key in Vault. On **AWS KMS**, check the key's automatic rotation status in AWS — and do not schedule rotation through the RustFS endpoint, which maps to quota-limited `RotateKeyOnDemand`.
 4. Know the gauge's blind spot on Transit and AWS before chasing a rotation that already happened: only KV2 persists a rotation timestamp, so Transit and AWS keys age from creation permanently and this alert will not clear after a rotation there. Confirm the real cadence at the owning system — the Transit key's version history in Vault, or the key's rotation status in AWS — and treat a confirmed-healthy cadence as a known overstatement of this gauge rather than an overdue key.

rustfs/src/app/multipart_usecase.rs

@@ -27,7 +27,6 @@ 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};
@@ -40,7 +39,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, compression_metadata_value};
+use super::storage_api::multipart_usecase::io::{HashReader, WriteEncryption, WritePlan};
 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,
@@ -208,28 +207,6 @@ 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
@@ -865,17 +842,8 @@ impl DefaultMultipartUsecase {
             None => (None, None),
         };
 
-        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()),
-            );
-        }
+        // Multipart parts are independent physical streams. Advertising object-level
+        // compression here would make GET decode the completed object as one stream.
 
         let mt2 = metadata.clone();
         let mut opts: ObjectOptions =
@@ -1669,31 +1637,6 @@ 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/object_usecase.rs

@@ -86,7 +86,7 @@ use super::storage_api::object_usecase::options::{
     namespace_reserved_user_metadata, normalize_content_encoding_for_storage, preserve_unclassified_user_metadata,
     put_opts_with_replication_authorization, validate_archive_content_encoding,
 };
-use super::storage_api::object_usecase::request_context::{self, spawn_traced};
+use super::storage_api::object_usecase::request_context::{self, spawn_traced, spawn_traced_join};
 use super::storage_api::object_usecase::s3_api::multipart::parse_list_parts_params;
 use super::storage_api::object_usecase::set_disk::{
     get_lock_acquire_timeout, get_object_disk_read_timeout, is_valid_storage_class,
@@ -7506,31 +7506,50 @@ impl DefaultObjectUsecase {
         let cache_adapter = self.object_data_cache();
         let _ = invalidate_object_data_cache_before_mutation(&cache_adapter, &bucket, &key).await;
 
-        let oi = store
-            .copy_object(&src_bucket, &src_key, &bucket, &key, &mut src_info, &src_opts, &dst_opts)
-            .await
-            .map_err(ApiError::from)?;
-        drop(_self_copy_lock_guard);
+        let copy_commit = spawn_traced_join({
+            let store = Arc::clone(&store);
+            let src_bucket = src_bucket.clone();
+            let src_key = src_key.clone();
+            let bucket = bucket.clone();
+            let key = key.clone();
+            let src_opts = src_opts.clone();
+            let dst_opts = dst_opts.clone();
+            async move {
+                let _source_bucket_lifecycle_guard = source_bucket_lifecycle_guard;
+                let _destination_bucket_lifecycle_guard_storage = destination_bucket_lifecycle_guard_storage;
+                let _self_copy_lock_guard = _self_copy_lock_guard;
 
-        // Reuse the single pre-commit replication decision (see `dsc` above) so
-        // the persisted pending marker and the schedule always agree, mirroring
-        // the PUT path.
-        if dsc.replicate_any() {
-            schedule_object_replication(oi.clone(), store.clone(), dsc).await;
-        }
+                let oi = store
+                    .copy_object(&src_bucket, &src_key, &bucket, &key, &mut src_info, &src_opts, &dst_opts)
+                    .await
+                    .map_err(ApiError::from)?;
 
-        maybe_enqueue_transition_immediate(&oi, LcEventSrc::S3CopyObject).await;
-        let _ = invalidate_object_data_cache_after_copy_success(&cache_adapter, &bucket, &key).await;
+                // Reuse the single pre-commit replication decision (see `dsc` above) so
+                // the persisted pending marker and the schedule always agree, mirroring
+                // the PUT path.
+                if dsc.replicate_any() {
+                    schedule_object_replication(oi.clone(), Arc::clone(&store), dsc).await;
+                }
 
-        let dest_versioned = BucketVersioningSys::prefix_enabled(&bucket, &key).await;
-        // Update quota tracking after successful copy
-        if has_bucket_metadata {
-            if dest_versioned {
-                record_bucket_object_version_write_memory(&bucket, previous_current_size, oi.size.max(0) as u64).await;
-            } else {
-                record_bucket_object_write_memory(&bucket, previous_current_size, oi.size.max(0) as u64).await;
+                maybe_enqueue_transition_immediate(&oi, LcEventSrc::S3CopyObject).await;
+                let _ = invalidate_object_data_cache_after_copy_success(&cache_adapter, &bucket, &key).await;
+
+                let dest_versioned = BucketVersioningSys::prefix_enabled(&bucket, &key).await;
+                if has_bucket_metadata {
+                    if dest_versioned {
+                        record_bucket_object_version_write_memory(&bucket, previous_current_size, oi.size.max(0) as u64).await;
+                    } else {
+                        record_bucket_object_write_memory(&bucket, previous_current_size, oi.size.max(0) as u64).await;
+                    }
+                }
+
+                rustfs_scanner::record_dirty_usage_bucket(&bucket);
+                Ok::<_, S3Error>((oi, dest_versioned))
             }
-        }
+        });
+        let (oi, dest_versioned) = copy_commit.await.map_err(|err| {
+            S3Error::with_message(S3ErrorCode::InternalError, format!("copy object commit owner task failed: {err}"))
+        })??;
 
         let raw_dest_version = oi.version_id.map(|v| v.to_string());
         let dest_version = if dest_versioned { raw_dest_version } else { None };
@@ -7578,7 +7597,7 @@ impl DefaultObjectUsecase {
             }
         }
         let copy_object_result = CopyObjectResult {
-            e_tag: oi.etag.map(|etag| to_s3s_etag(&etag)),
+            e_tag: oi.etag.as_ref().map(|etag| to_s3s_etag(etag)),
             last_modified: oi.mod_time.map(Timestamp::from),
             checksum_crc32: response_checksums.crc32,
             checksum_crc32c: response_checksums.crc32c,
@@ -7609,7 +7628,6 @@ impl DefaultObjectUsecase {
 
         let result = Ok(S3Response::new(output));
         let _ = helper.complete(&result);
-        rustfs_scanner::record_dirty_usage_bucket(&bucket);
         result
     }
 

rustfs/src/app/storage_api.rs

@@ -940,9 +940,7 @@ pub(crate) mod concurrency {
 }
 
 pub(crate) mod compression {
-    pub(crate) use crate::storage::storage_api::ecstore_compression::{
-        MIN_DISK_COMPRESSIBLE_SIZE, is_disk_compressible, is_multipart_disk_compression_enabled,
-    };
+    pub(crate) use crate::storage::storage_api::ecstore_compression::{MIN_DISK_COMPRESSIBLE_SIZE, is_disk_compressible};
 }
 
 pub(crate) mod deadlock_detector {
@@ -1000,7 +998,7 @@ pub(crate) mod options {
 }
 
 pub(crate) mod request_context {
-    pub(crate) use crate::storage::storage_api::request_context_consumer::{RequestContext, spawn_traced};
+    pub(crate) use crate::storage::storage_api::request_context_consumer::{RequestContext, spawn_traced, spawn_traced_join};
 }
 
 pub(crate) mod sse {
@@ -1152,9 +1150,7 @@ pub(crate) mod multipart_usecase {
         }
     }
 
-    pub(crate) use super::{
-        access, bucket, compression, data_usage, error, helper, io, object_utils, options, s3_api, set_disk, sse,
-    };
+    pub(crate) use super::{access, bucket, data_usage, error, helper, io, object_utils, options, s3_api, set_disk, sse};
     pub(crate) use crate::storage::storage_api::{ECStore, StorageObjectInfo, StorageObjectOptions, StoragePutObjReader};
 }
 

rustfs/src/storage/request_context.rs

@@ -257,6 +257,15 @@ where
     tokio::spawn(tracing::Instrument::instrument(fut, tracing::Span::current()));
 }
 
+/// Spawn a request-internal task and return its join handle to the caller.
+pub fn spawn_traced_join<F>(fut: F) -> tokio::task::JoinHandle<F::Output>
+where
+    F: std::future::Future + Send + 'static,
+    F::Output: Send + 'static,
+{
+    tokio::spawn(tracing::Instrument::instrument(fut, tracing::Span::current()))
+}
+
 #[cfg(test)]
 #[allow(unused_imports)]
 mod tests {

rustfs/src/storage/s3_api/multipart.rs

@@ -51,14 +51,7 @@ 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(),
-                    // 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 {
-                        Some(p.actual_size)
-                    } else {
-                        p.size.try_into().ok()
-                    },
+                    size: p.size.try_into().ok(),
                     ..Default::default()
                 })
                 .collect(),
@@ -254,63 +247,6 @@ 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_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/sse.rs

@@ -119,8 +119,14 @@ use rustfs_utils::http::object_encryption_keys::{
     MINIO_INTERNAL_ENCRYPTION_KMS_CONTEXT_HEADER, MINIO_INTERNAL_ENCRYPTION_KMS_DATA_KEY_HEADER,
     MINIO_INTERNAL_ENCRYPTION_KMS_KEY_ID_HEADER, MINIO_INTERNAL_ENCRYPTION_KMS_SEALED_KEY_HEADER,
     MINIO_INTERNAL_ENCRYPTION_MULTIPART_HEADER, MINIO_INTERNAL_ENCRYPTION_S3_SEALED_KEY_HEADER,
-    MINIO_INTERNAL_ENCRYPTION_SSEC_SEALED_KEY_HEADER, SSEC_ORIGINAL_SIZE_HEADER,
+    MINIO_INTERNAL_ENCRYPTION_SSEC_SEALED_KEY_HEADER, SSEC_ORIGINAL_SIZE_HEADER, normalize_managed_metadata,
+    stored_managed_encryption_key,
 };
+// The managed-SSE classifier lives in the shared encryption-keys module so the
+// scanner can reuse it (backlog#1643 PR-B0); these re-exports keep the
+// historical `crate::storage::sse` paths compiling.
+pub use rustfs_utils::http::object_encryption_keys::SSEType;
+pub(crate) use rustfs_utils::http::object_encryption_keys::contains_managed_encryption_metadata;
 #[cfg(feature = "rio-v2")]
 const MINIO_INTERNAL_ENCRYPTION_SEAL_ALGORITHM: &str = "DAREv2-HMAC-SHA256";
 #[cfg(feature = "rio-v2")]
@@ -783,28 +789,6 @@ pub struct DecryptionMaterial {
     pub key_kind: EncryptionKeyKind,
 }
 
-/// Type of encryption used
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
-pub enum SSEType {
-    /// SSE-S3 (AES256)
-    SseS3,
-    /// SSE-KMS (aws:kms)
-    SseKms,
-    /// SSE-C (customer-provided key)
-    SseC,
-}
-
-impl SSEType {
-    /// Stable scheme name for audit consumers.
-    fn audit_label(self) -> &'static str {
-        match self {
-            SSEType::SseS3 => "SSE-S3",
-            SSEType::SseKms => "SSE-KMS",
-            SSEType::SseC => "SSE-C",
-        }
-    }
-}
-
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum EncryptionKeyKind {
     Direct,
@@ -1064,28 +1048,6 @@ pub async fn authorize_sse_kms_object_read(
     result
 }
 
-/// Resolve the scheme and KMS key a stored managed-SSE object was wrapped with.
-///
-/// Mirrors the lookup `apply_managed_decryption_material` performs, so both agree on
-/// which key a read is authorized against.
-fn stored_managed_encryption_key(metadata: &HashMap<String, String>) -> Option<(SSEType, String)> {
-    if !contains_managed_encryption_metadata(metadata) {
-        return None;
-    }
-
-    let sse_type = match metadata.get("x-amz-server-side-encryption")?.as_str() {
-        ServerSideEncryption::AWS_KMS => SSEType::SseKms,
-        _ => SSEType::SseS3,
-    };
-    let key_id = normalize_managed_metadata(metadata)
-        .get(INTERNAL_ENCRYPTION_KEY_ID_HEADER)
-        .or_else(|| metadata.get("x-amz-server-side-encryption-aws-kms-key-id"))
-        .cloned()
-        .unwrap_or_else(|| "default".to_string());
-
-    Some((sse_type, key_id))
-}
-
 // ============================================================================
 // Data-plane KMS audit attachment (SSE-S3 / SSE-KMS)
 // ============================================================================
@@ -1339,7 +1301,9 @@ fn envelope_master_key_version(envelope_bytes: &[u8]) -> Option<u32> {
 /// Master-key version of the envelope stored on an object, for the audit
 /// summary of a read against that object.
 fn stored_envelope_master_key_version(metadata: &HashMap<String, String>) -> Option<u32> {
-    let encoded = normalize_managed_metadata(metadata);
+    // No context recoder: the recode only ever inserts the context key, which
+    // this lookup never reads, so the normalized result is identical without it.
+    let encoded = normalize_managed_metadata(metadata, None);
     let encoded = encoded.get(INTERNAL_ENCRYPTION_KEY_HEADER)?;
     let envelope = BASE64_STANDARD.decode(encoded).ok()?;
     envelope_master_key_version(&envelope)
@@ -2487,7 +2451,7 @@ async fn apply_managed_decryption_material_inner(
 
     // Safe: presence is guaranteed by the contains_key check above.
     let server_side_encryption = metadata.get("x-amz-server-side-encryption").cloned().unwrap_or_default();
-    let normalized_metadata = normalize_managed_metadata(metadata);
+    let normalized_metadata = normalize_managed_metadata(metadata, Some(recode_minio_kms_context));
 
     let encryption_type = match server_side_encryption.as_str() {
         ServerSideEncryption::AES256 => SSEType::SseS3,
@@ -3229,14 +3193,6 @@ pub fn mark_encrypted_multipart_metadata(metadata: &mut HashMap<String, String>)
     metadata.insert(MINIO_INTERNAL_ENCRYPTION_MULTIPART_HEADER.to_string(), String::new());
 }
 
-pub(crate) fn contains_managed_encryption_metadata(metadata: &HashMap<String, String>) -> bool {
-    metadata.contains_key(INTERNAL_ENCRYPTION_KEY_HEADER)
-        || metadata.contains_key(MINIO_INTERNAL_ENCRYPTION_S3_SEALED_KEY_HEADER)
-        || metadata.contains_key(MINIO_INTERNAL_ENCRYPTION_KMS_SEALED_KEY_HEADER)
-        || metadata.contains_key(MINIO_INTERNAL_ENCRYPTION_KMS_DATA_KEY_HEADER)
-        || metadata.contains_key(MINIO_INTERNAL_ENCRYPTION_KMS_CONTEXT_HEADER)
-}
-
 #[cfg(feature = "rio-v2")]
 fn is_legacy_rustfs_managed_metadata(metadata: &HashMap<String, String>) -> bool {
     metadata.contains_key(INTERNAL_ENCRYPTION_KEY_HEADER)
@@ -3282,47 +3238,16 @@ fn parse_minio_managed_sealed_key(
     Ok(Some(ManagedSealedKey { iv, sealed_key }))
 }
 
-fn normalize_managed_metadata(metadata: &HashMap<String, String>) -> HashMap<String, String> {
-    let mut normalized = metadata.clone();
-
-    if !normalized.contains_key(INTERNAL_ENCRYPTION_KEY_HEADER)
-        && let Some(value) = metadata
-            .get(MINIO_INTERNAL_ENCRYPTION_KMS_DATA_KEY_HEADER)
-            .or_else(|| metadata.get(MINIO_INTERNAL_ENCRYPTION_KMS_SEALED_KEY_HEADER))
-            .or_else(|| metadata.get(MINIO_INTERNAL_ENCRYPTION_S3_SEALED_KEY_HEADER))
-            .or_else(|| metadata.get(MINIO_INTERNAL_ENCRYPTION_SSEC_SEALED_KEY_HEADER))
-    {
-        normalized.insert(INTERNAL_ENCRYPTION_KEY_HEADER.to_string(), value.clone());
-    }
-
-    if !normalized.contains_key(INTERNAL_ENCRYPTION_IV_HEADER)
-        && let Some(value) = metadata.get(MINIO_INTERNAL_ENCRYPTION_IV_HEADER)
-    {
-        normalized.insert(INTERNAL_ENCRYPTION_IV_HEADER.to_string(), value.clone());
-    }
-
-    if !normalized.contains_key(INTERNAL_ENCRYPTION_ALGORITHM_HEADER)
-        && let Some(value) = metadata.get(MINIO_INTERNAL_ENCRYPTION_ALGORITHM_HEADER)
-    {
-        normalized.insert(INTERNAL_ENCRYPTION_ALGORITHM_HEADER.to_string(), value.clone());
-    }
-
-    if !normalized.contains_key(INTERNAL_ENCRYPTION_KEY_ID_HEADER)
-        && let Some(value) = metadata.get(MINIO_INTERNAL_ENCRYPTION_KMS_KEY_ID_HEADER)
-    {
-        normalized.insert(INTERNAL_ENCRYPTION_KEY_ID_HEADER.to_string(), value.clone());
-    }
-
-    if !normalized.contains_key(INTERNAL_ENCRYPTION_CONTEXT_HEADER)
-        && let Some(value) = metadata.get(MINIO_INTERNAL_ENCRYPTION_KMS_CONTEXT_HEADER)
-        && let Ok(decoded) = BASE64_STANDARD.decode(value)
-        && let Ok(context) = serde_json::from_slice::<HashMap<String, String>>(&decoded)
-        && let Ok(encoded) = serde_json::to_string(&context)
-    {
-        normalized.insert(INTERNAL_ENCRYPTION_CONTEXT_HEADER.to_string(), encoded);
-    }
-
-    normalized
+/// Recodes a stored MinIO KMS context value (base64-wrapped JSON) into the
+/// plain-JSON form RustFS stores under [`INTERNAL_ENCRYPTION_CONTEXT_HEADER`].
+///
+/// Injected into the shared [`normalize_managed_metadata`] because the shared
+/// crate carries no JSON codec; any decode failure returns `None`, which skips
+/// the context mapping exactly like the historical inline `if let Ok` chain.
+fn recode_minio_kms_context(value: &str) -> Option<String> {
+    let decoded = BASE64_STANDARD.decode(value).ok()?;
+    let context = serde_json::from_slice::<HashMap<String, String>>(&decoded).ok()?;
+    serde_json::to_string(&context).ok()
 }
 
 // ============================================================================
@@ -3473,9 +3398,9 @@ mod tests {
         encryption_material_to_metadata, extract_server_side_encryption_from_headers, extract_ssec_params_from_headers,
         extract_ssekms_context_from_headers, generate_ssec_nonce, is_managed_sse, kms_operation_error,
         map_get_object_reader_error, mark_encrypted_multipart_metadata, md5_base64, normalize_managed_metadata,
-        reset_sse_dek_provider, resolve_effective_kms_key_id, sse_decryption, sse_encryption, sse_prepare_encryption,
-        strip_managed_encryption_metadata, validate_sse_headers_for_read, validate_sse_headers_for_write, validate_ssec_for_read,
-        validate_ssec_params, verify_ssec_key_match,
+        recode_minio_kms_context, reset_sse_dek_provider, resolve_effective_kms_key_id, sse_decryption, sse_encryption,
+        sse_prepare_encryption, strip_managed_encryption_metadata, validate_sse_headers_for_read, validate_sse_headers_for_write,
+        validate_ssec_for_read, validate_ssec_params, verify_ssec_key_match,
     };
     #[cfg(feature = "rio-v2")]
     use super::{
@@ -3484,6 +3409,38 @@ mod tests {
     };
     use rustfs_utils::http::headers::SSEC_ALGORITHM_HEADER;
 
+    /// backlog#1643 PR-B0 acceptance guard: the managed-SSE classifier must
+    /// have exactly one definition — in the shared encryption-keys module —
+    /// so the scanner and the S3 layer can never disagree on attribution.
+    /// This module may only re-export or call it.
+    #[test]
+    fn managed_sse_classifier_has_exactly_one_definition() {
+        let classifier_fns = [
+            "contains_managed_encryption_metadata",
+            "normalize_managed_metadata",
+            "stored_managed_encryption_key",
+        ];
+
+        let sse_src = include_str!("sse.rs");
+        let shared_src =
+            std::fs::read_to_string(concat!(env!("CARGO_MANIFEST_DIR"), "/../crates/utils/src/http/object_encryption_keys.rs"))
+                .expect("shared encryption-keys module should be readable");
+
+        for name in classifier_fns {
+            // Built at runtime so this test's own source cannot satisfy the scan.
+            let definition = format!("fn {name}(");
+            assert!(
+                !sse_src.contains(&definition),
+                "{name} must not be redefined in storage/sse.rs; call the shared rustfs_utils::http::object_encryption_keys implementation instead"
+            );
+            assert_eq!(
+                shared_src.matches(&definition).count(),
+                1,
+                "{name} must be defined exactly once, in the shared encryption-keys module"
+            );
+        }
+    }
+
     #[test]
     fn ssec_read_headers_are_sensitive() {
         let headers = super::build_ssec_read_headers(
@@ -4869,7 +4826,7 @@ mod tests {
             (MINIO_INTERNAL_ENCRYPTION_KMS_KEY_ID_HEADER.to_string(), "default".to_string()),
         ]);
 
-        let normalized = normalize_managed_metadata(&metadata);
+        let normalized = normalize_managed_metadata(&metadata, Some(recode_minio_kms_context));
 
         assert_eq!(
             normalized.get(INTERNAL_ENCRYPTION_KEY_HEADER),

rustfs/src/storage/storage_api.rs

@@ -203,7 +203,9 @@ pub(crate) mod options_consumer {
 }
 
 pub(crate) mod request_context_consumer {
-    pub(crate) use super::super::request_context::{RequestContext, extract_request_id_from_headers, spawn_traced};
+    pub(crate) use super::super::request_context::{
+        RequestContext, extract_request_id_from_headers, spawn_traced, spawn_traced_join,
+    };
 }
 
 pub(crate) mod rpc_consumer {
@@ -407,9 +409,7 @@ pub(crate) mod ecstore_client {
 }
 
 pub(crate) mod ecstore_compression {
-    pub(crate) use rustfs_ecstore::api::compression::{
-        MIN_DISK_COMPRESSIBLE_SIZE, is_disk_compressible, is_multipart_disk_compression_enabled,
-    };
+    pub(crate) use rustfs_ecstore::api::compression::{MIN_DISK_COMPRESSIBLE_SIZE, is_disk_compressible};
 }
 
 pub(crate) mod ecstore_cluster {

scripts/run.sh

@@ -209,7 +209,6 @@ 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)