diff --git a/.config/nextest.toml b/.config/nextest.toml index d543d0c4a..ade76a58c 100644 --- a/.config/nextest.toml +++ b/.config/nextest.toml @@ -252,10 +252,16 @@ 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|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|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(/^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::/) diff --git a/crates/e2e_test/src/common.rs b/crates/e2e_test/src/common.rs index a4f4dfedb..f1fcaa20a 100644 --- a/crates/e2e_test/src/common.rs +++ b/crates/e2e_test/src/common.rs @@ -67,7 +67,10 @@ fn configured_capture_log_path(temp_dir: &str) -> Option { capture_log_path(Path::new(&log_dir), temp_dir).map(|path| path.to_string_lossy().into_owned()) } -fn capture_command_logs(command: &mut Command, log_path: Option<&str>) -> Result<(), Box> { +pub(crate) fn capture_command_logs( + command: &mut Command, + log_path: Option<&str>, +) -> Result<(), Box> { let Some(log_path) = log_path else { return Ok(()); }; diff --git a/crates/e2e_test/src/compression_test.rs b/crates/e2e_test/src/compression_test.rs index f01e78320..1f3bfa59e 100644 --- a/crates/e2e_test/src/compression_test.rs +++ b/crates/e2e_test/src/compression_test.rs @@ -2,6 +2,7 @@ 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; @@ -25,6 +26,15 @@ fn generate_compressible_data(size: usize) -> Vec { 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 { + (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 { let bucket_path = PathBuf::from(temp_dir).join(bucket); let mut part_files = Vec::new(); @@ -55,7 +65,11 @@ async fn start_rustfs_with_compression(env: &mut RustFSTestEnvironment) -> Resul env.cleanup_existing_processes().await?; let binary_path = rustfs_binary_path(); - let process = Command::new(&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 .env("RUSTFS_CONSOLE_ENABLE", "false") .env("RUSTFS_COMPRESSION_ENABLED", "true") .args([ @@ -66,8 +80,9 @@ async fn start_rustfs_with_compression(env: &mut RustFSTestEnvironment) -> Resul "--secret-key", &env.secret_key, &env.temp_dir, - ]) - .spawn()?; + ]); + crate::common::capture_command_logs(&mut command, env.capture_log_path.as_deref())?; + let process = command.spawn()?; env.process = Some(process); @@ -154,3 +169,646 @@ async fn test_compression_roundtrip() -> Result<(), Box Result<(), Box> { + 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, Box> { + 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> { + 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> { + 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> { + 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> { + 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> { + 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_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> { + 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(()) +} diff --git a/crates/e2e_test/src/inline_fast_path_cluster_test.rs b/crates/e2e_test/src/inline_fast_path_cluster_test.rs index 68f37b3a2..bf8bbd024 100644 --- a/crates/e2e_test/src/inline_fast_path_cluster_test.rs +++ b/crates/e2e_test/src/inline_fast_path_cluster_test.rs @@ -1828,9 +1828,11 @@ 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_ignores_disk_compression_fallback() -> TestResult { +async fn four_node_multipart_disk_compression_roundtrip() -> TestResult { init_logging(); let collector = OtlpMetricCollector::start().await?; @@ -1839,22 +1841,22 @@ async fn four_node_multipart_ignores_disk_compression_fallback() -> TestResult { cluster.set_env("RUSTFS_COMPRESSION_ENABLED", "true"); cluster.start().await?; - let bucket = "inline-multipart-compression-fallback"; + let bucket = "inline-multipart-compression-roundtrip"; cluster.create_test_bucket(bucket).await?; let client = cluster.create_s3_client(0)?; - let key = "multipart/compression-disabled.txt"; + let key = "multipart/compressed.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, MULTIPART), + ReaderPathExpectation::for_class(ReaderObject::new(bucket, key, &body, etag.as_deref(), None), LEGACY_DUPLEX, COMPRESSED), ) .await?; assert_part_number_reader_path( &collector, &client, - PartNumberReaderPathExpectation::new(bucket, key, &second_part, body.len(), MULTIPART, LEGACY_DUPLEX), + PartNumberReaderPathExpectation::new(bucket, key, &second_part, body.len(), COMPRESSED, LEGACY_DUPLEX), ) .await?; @@ -1890,14 +1892,21 @@ 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, - MULTIPART, + COMPRESSED, ), ) .await?; assert_part_number_reader_path( &collector, &client, - PartNumberReaderPathExpectation::new(bucket, multipart_key, &second_part, multipart_body.len(), MULTIPART, LEGACY_DUPLEX), + PartNumberReaderPathExpectation::new( + bucket, + multipart_key, + &second_part, + multipart_body.len(), + COMPRESSED, + LEGACY_DUPLEX, + ), ) .await?; assert_msgpack_decode_observed(&collector, &decode_before).await?; @@ -2353,7 +2362,11 @@ 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?; - let key = "transition/mixed-multipart.bin"; + // `.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 (body, second_part, etag) = put_two_part_multipart(&hot_client, bucket, key).await?; wait_for_transition(&hot_client, bucket, key, &tier_name).await?; assert!( diff --git a/crates/ecstore/src/object_api/readers.rs b/crates/ecstore/src/object_api/readers.rs index cc3a1986a..cc0481938 100644 --- a/crates/ecstore/src/object_api/readers.rs +++ b/crates/ecstore/src/object_api/readers.rs @@ -1727,6 +1727,423 @@ 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, Option) { + 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, + plaintext: Vec, + } + + /// 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, + opts: &ObjectOptions, + ) -> Vec { + 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 { + (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::>(); + 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, + opts: &ObjectOptions, + ) -> Vec { + 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 { diff --git a/crates/rio/src/compress_reader.rs b/crates/rio/src/compress_reader.rs index b8b1c985a..037bb38f1 100644 --- a/crates/rio/src/compress_reader.rs +++ b/crates/rio/src/compress_reader.rs @@ -71,6 +71,7 @@ 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(), @@ -183,11 +184,21 @@ pin_project! { buffer: Vec, 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, - header_done: bool, - // Fields for saving compressed block read progress across polls + // 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. compressed_buf: Vec, compressed_read: usize, compressed_len: usize, @@ -205,9 +216,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, @@ -236,54 +247,72 @@ where if *this.finished { return Poll::Ready(Ok(())); } - // 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; + 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)); } - 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; - if typ == COMPRESS_TYPE_END { + 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; *this.compressed_read = 0; - *this.compressed_len = 0; - *this.finished = true; - return Poll::Ready(Ok(())); } - if this.compressed_buf.len() < len { - this.compressed_buf.resize(len, 0); - } - *this.compressed_len = len; - *this.compressed_read = 0; - + // Fill the in-flight block payload, resuming across polls via `compressed_read`. 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) { @@ -291,7 +320,13 @@ where Poll::Ready(Ok(())) => { let n = temp_buf.filled().len(); if n == 0 { - break; + *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", + ))); } *this.compressed_read += n; } @@ -299,10 +334,17 @@ 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 @@ -316,6 +358,7 @@ 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..]; @@ -326,21 +369,29 @@ where // error!("DecompressReader decompress_block error: {e}"); *this.compressed_read = 0; *this.compressed_len = 0; + *this.poisoned = true; return Poll::Ready(Err(e)); } } - } else if typ == COMPRESS_TYPE_UNCOMPRESSED { - compressed_data.to_vec() } else { - // error!("DecompressReader unknown compression type: {typ}"); + // The header phase already rejected every type other than + // COMPRESS_TYPE_COMPRESSED / 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; *this.compressed_read = 0; *this.compressed_len = 0; - return Poll::Ready(Err(io::Error::new(io::ErrorKind::InvalidData, "Unknown compression type"))); - }; + return Poll::Ready(Err(io::Error::new(io::ErrorKind::InvalidData, "Empty compressed block"))); + } 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 = { @@ -352,13 +403,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; @@ -493,6 +544,184 @@ 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 { + inner: R, + chunk: usize, + pending_next: bool, + } + + impl PendingChunkReader { + fn new(inner: R, chunk: usize) -> Self { + Self { + inner, + chunk, + pending_next: true, + } + } + } + + impl AsyncRead for PendingChunkReader { + fn poll_read( + mut self: std::pin::Pin<&mut Self>, + cx: &mut std::task::Context<'_>, + buf: &mut tokio::io::ReadBuf<'_>, + ) -> std::task::Poll> { + 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 { + (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, @@ -518,6 +747,85 @@ 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) -> Vec { + 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] diff --git a/rustfs/src/app/multipart_usecase.rs b/rustfs/src/app/multipart_usecase.rs index 8dd561da1..18295fc00 100644 --- a/rustfs/src/app/multipart_usecase.rs +++ b/rustfs/src/app/multipart_usecase.rs @@ -27,6 +27,7 @@ use super::storage_api::multipart_usecase::bucket::{ replication::{must_replicate_object, schedule_object_replication}, versioning_sys::BucketVersioningSys, }; +use super::storage_api::multipart_usecase::compression::is_disk_compressible; #[cfg(test)] use super::storage_api::multipart_usecase::contract::http::HTTPPreconditions; use super::storage_api::multipart_usecase::contract::multipart::{CompletePart, MultipartOperations as _, MultipartUploadResult}; @@ -39,7 +40,7 @@ use super::storage_api::multipart_usecase::error::{StorageError, is_err_object_n use super::storage_api::multipart_usecase::helper::OperationHelper; #[cfg(test)] use super::storage_api::multipart_usecase::io::{DecryptReader, EncryptReader, HardLimitReader, boxed_reader, wrap_reader}; -use super::storage_api::multipart_usecase::io::{HashReader, WriteEncryption, WritePlan}; +use super::storage_api::multipart_usecase::io::{HashReader, WriteEncryption, WritePlan, compression_metadata_value}; use super::storage_api::multipart_usecase::object_utils::to_s3s_etag; use super::storage_api::multipart_usecase::options::{ copy_src_opts, extract_metadata_from_mime, get_complete_multipart_upload_opts_with_replication_authorization, @@ -208,6 +209,21 @@ fn create_multipart_upload_metadata( metadata } +/// A multipart session advertises disk compression only when the object key/headers +/// qualify AND the session is not an SSE-C ciphertext-passthrough replication session, +/// which must preserve source bytes verbatim. +/// +/// 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(ciphertext_passthrough: bool, disk_compressible: bool) -> bool { + !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 @@ -870,8 +886,13 @@ impl DefaultMultipartUsecase { None => (None, None), }; - // Multipart parts are independent physical streams. Advertising object-level - // compression here would make GET decode the completed object as one stream. + if should_advertise_session_compression(ciphertext_passthrough, is_disk_compressible(&req.headers, &key)) { + rustfs_utils::http::insert_str( + &mut metadata, + rustfs_utils::http::SUFFIX_COMPRESSION, + compression_metadata_value(CompressionAlgorithm::default()), + ); + } let mt2 = metadata.clone(); let mut opts: ObjectOptions = @@ -1665,6 +1686,25 @@ mod tests { DefaultMultipartUsecase::without_context() } + #[test] + fn session_compression_is_advertised_only_for_non_passthrough_compressible_uploads() { + // (ciphertext_passthrough, disk_compressible, expected) + let cases = [ + (false, false, false), + (false, true, true), + (true, false, false), + (true, true, false), + ]; + + for (ciphertext_passthrough, disk_compressible, expected) in cases { + assert_eq!( + should_advertise_session_compression(ciphertext_passthrough, disk_compressible), + expected, + "ciphertext_passthrough={ciphertext_passthrough} disk_compressible={disk_compressible}" + ); + } + } + #[test] fn quota_accounting_uses_logical_size_when_available() { let mut metadata = HashMap::new(); diff --git a/rustfs/src/app/storage_api.rs b/rustfs/src/app/storage_api.rs index 39d7d7685..91933ddb0 100644 --- a/rustfs/src/app/storage_api.rs +++ b/rustfs/src/app/storage_api.rs @@ -1151,7 +1151,8 @@ pub(crate) mod multipart_usecase { } pub(crate) use super::{ - access, bucket, data_usage, error, helper, io, object_utils, options, request_context, s3_api, set_disk, sse, + access, bucket, compression, data_usage, error, helper, io, object_utils, options, request_context, s3_api, + set_disk, sse, }; pub(crate) use crate::storage::storage_api::{ECStore, StorageObjectInfo, StorageObjectOptions, StoragePutObjReader}; } diff --git a/rustfs/src/storage/s3_api/multipart.rs b/rustfs/src/storage/s3_api/multipart.rs index ffc1a4692..3bd355b75 100644 --- a/rustfs/src/storage/s3_api/multipart.rs +++ b/rustfs/src/storage/s3_api/multipart.rs @@ -51,7 +51,14 @@ pub(crate) fn build_list_parts_output(res: ListPartsInfo) -> ListPartsOutput { e_tag: p.etag.map(|etag| to_s3s_etag(&etag)), last_modified: p.last_mod.map(Timestamp::from), part_number: p.part_num.try_into().ok(), - size: p.size.try_into().ok(), + // Compressed parts store fewer bytes than the client sent; S3 + // semantics report the uploaded (logical) size, matching + // GetObjectAttributes ObjectParts. + size: if p.actual_size > 0 { + Some(p.actual_size) + } else { + p.size.try_into().ok() + }, ..Default::default() }) .collect(), @@ -247,6 +254,63 @@ 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 {