From ca6b4088e4c8ed55ec26808c86eb0ee0ade3e1d4 Mon Sep 17 00:00:00 2001 From: overtrue Date: Thu, 13 Aug 2026 08:42:16 +0800 Subject: [PATCH] feat(storage): stage multipart compression behind RUSTFS_COMPRESSION_MULTIPART_ENABLED Review follow-up: a rolling-upgrade window must not create new compressed multipart objects while pre-fix nodes (whose decompressor is not resumable) may still serve reads. The session marker is now additionally gated on RUSTFS_COMPRESSION_MULTIPART_ENABLED, default off, so the restored capability stays dark until the operator confirms fleet convergence. The default flips per the multipart-compression-default-off-window entry in docs/architecture/compat-cleanup-register.md once the minimum supported direct-upgrade release ships the resumable decoder. --- crates/e2e_test/src/compression_test.rs | 2 + .../src/inline_fast_path_cluster_test.rs | 2 + crates/ecstore/src/api/mod.rs | 4 +- crates/ecstore/src/io_support/compress.rs | 22 +++++++++ docs/architecture/compat-cleanup-register.md | 1 + rustfs/src/app/multipart_usecase.rs | 47 +++++++++++++------ rustfs/src/app/storage_api.rs | 4 +- rustfs/src/storage/storage_api.rs | 4 +- scripts/run.sh | 1 + 9 files changed, 69 insertions(+), 18 deletions(-) diff --git a/crates/e2e_test/src/compression_test.rs b/crates/e2e_test/src/compression_test.rs index 1f3bfa59e..775decfa6 100644 --- a/crates/e2e_test/src/compression_test.rs +++ b/crates/e2e_test/src/compression_test.rs @@ -72,6 +72,7 @@ async fn start_rustfs_with_compression(env: &mut RustFSTestEnvironment) -> Resul command .env("RUSTFS_CONSOLE_ENABLE", "false") .env("RUSTFS_COMPRESSION_ENABLED", "true") + .env("RUSTFS_COMPRESSION_MULTIPART_ENABLED", "true") .args([ "--address", &env.address, @@ -652,6 +653,7 @@ async fn start_rustfs_with_compression_and_sse( let process = Command::new(&binary_path) .env("RUSTFS_CONSOLE_ENABLE", "false") .env("RUSTFS_COMPRESSION_ENABLED", "true") + .env("RUSTFS_COMPRESSION_MULTIPART_ENABLED", "true") .env("RUSTFS_SSE_S3_MASTER_KEY", master_key) .env("RUST_LOG", "rustfs=info,rustfs_ecstore=info") .stdout(std::process::Stdio::from(server_log)) 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 bf8bbd024..e2890c4a8 100644 --- a/crates/e2e_test/src/inline_fast_path_cluster_test.rs +++ b/crates/e2e_test/src/inline_fast_path_cluster_test.rs @@ -1839,6 +1839,7 @@ async fn four_node_multipart_disk_compression_roundtrip() -> TestResult { let mut cluster = RustFSTestClusterEnvironment::new(4).await?; configure_reader_metric_cluster(&mut cluster, &collector); cluster.set_env("RUSTFS_COMPRESSION_ENABLED", "true"); + cluster.set_env("RUSTFS_COMPRESSION_MULTIPART_ENABLED", "true"); cluster.start().await?; let bucket = "inline-multipart-compression-roundtrip"; @@ -1873,6 +1874,7 @@ async fn four_node_mixed_msgpack_compat_mode_preserves_fallback_controls() -> Te let sse_master_key = base64::engine::general_purpose::STANDARD.encode([0x42u8; 32]); cluster.set_env("RUSTFS_SSE_S3_MASTER_KEY", sse_master_key); cluster.set_env("RUSTFS_COMPRESSION_ENABLED", "true"); + cluster.set_env("RUSTFS_COMPRESSION_MULTIPART_ENABLED", "true"); configure_mixed_msgpack_cluster(&mut cluster, &collector)?; cluster.start().await?; diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index 4b3162313..2fd79c3b3 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/crates/ecstore/src/api/mod.rs @@ -276,7 +276,9 @@ pub mod cluster { } pub mod compression { - pub use crate::io_support::compress::{MIN_DISK_COMPRESSIBLE_SIZE, is_disk_compressible, is_disk_compression_enabled}; + pub use crate::io_support::compress::{ + MIN_DISK_COMPRESSIBLE_SIZE, is_disk_compressible, is_disk_compression_enabled, is_multipart_disk_compression_enabled, + }; } pub mod config { diff --git a/crates/ecstore/src/io_support/compress.rs b/crates/ecstore/src/io_support/compress.rs index 356974afe..633aab0c5 100644 --- a/crates/ecstore/src/io_support/compress.rs +++ b/crates/ecstore/src/io_support/compress.rs @@ -31,6 +31,13 @@ pub const ENV_DISK_COMPRESSION_MIME_TYPES: &str = "RUSTFS_COMPRESSION_MIME_TYPES // Environment variable for additional extensions to exclude from compression (comma-separated, e.g. ".foo,.bar") pub const ENV_ADDED_EXCLUDE_COMPRESS_EXTENSIONS: &str = "RUSTFS_ADDED_EXCLUDE_COMPRESS_EXTENSIONS"; +// Environment variable to additionally enable disk compression for multipart uploads. +// Default off: nodes from before the resumable decompressor fix fail transient reads of +// compressed objects, so multipart compression stays dark until the operator confirms the +// fleet has converged on a fixed build (see docs/architecture/compat-cleanup-register.md, +// `multipart-compression-default-off-window`, for the default-flip condition). +pub const ENV_DISK_COMPRESSION_MULTIPART_ENABLED: &str = "RUSTFS_COMPRESSION_MULTIPART_ENABLED"; + pub const DEFAULT_DISK_COMPRESS_EXTENSIONS: &str = ".txt,.log,.csv,.json,.tar,.xml,.bin"; pub const DEFAULT_DISK_COMPRESS_MIME_TYPES: &str = "text/*,application/json,application/xml,binary/octet-stream"; @@ -171,6 +178,21 @@ pub fn is_disk_compression_enabled() -> bool { DISK_COMPRESSION_CONFIG.get_or_init(parse_disk_compression_config).enabled } +// Parsed once at first use, mirroring DISK_COMPRESSION_CONFIG. +static MULTIPART_DISK_COMPRESSION_ENABLED: OnceLock = 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 { diff --git a/docs/architecture/compat-cleanup-register.md b/docs/architecture/compat-cleanup-register.md index 880396b5f..51726d4c0 100644 --- a/docs/architecture/compat-cleanup-register.md +++ b/docs/architecture/compat-cleanup-register.md @@ -35,6 +35,7 @@ for later deletion. - `rustfs-5063` pre-beta.9 Local KMS recovery: persisted Local KMS configs from beta.8 and earlier predate the explicit insecure-development flag, and encrypted key files use the legacy SHA-256 KDF. Remove the config fallback after supported upgrades have rewritten or explicitly resaved all pre-beta.9 configs with the development-default field, and remove the legacy KDF after supported upgrades have rewritten all pre-beta.9 Local KMS key files with explicit at-rest protection. - `sse-local-dek-json-v1` legacy local SSE DEK decoding: releases before the JSON envelope wrote wrapped DEKs as `base64(nonce):base64(ciphertext)`, so readers retain that decoder while all new writes use the versioned JSON envelope. Remove the colon decoder after the minimum supported direct-upgrade release writes JSON envelopes and migration tooling has rewritten every retained legacy object. - `rio-v2-dormant-variant` dormant `rio-v2` build variant: `crates/rio-v2` and the `rio-v2` feature ship in no default or release build and exist only as the candidate MinIO stream-format implementation for the rustfs/backlog#1638 SSE-interop adjudication. Per-PR CI keeps only `test-and-lint-rio-v2` to guard the `#[cfg(feature = "rio-v2")]` seam; the full-suite lanes (`build-rustfs-debug-binary-rio-v2`, `e2e-tests-rio-v2` in `.github/workflows/ci.yml`) run on schedule/workflow_dispatch only — see the "`rio-v2` variant lifecycle" section in [minio-file-format-compat.md](minio-file-format-compat.md). While both implementations exist, DARE/S2 stream fixes must land in both `crates/rio` and `crates/rio-v2`. No `RUSTFS_COMPAT_TODO` source marker applies: the temporary surface is CI workflow YAML plus an entire feature-gated candidate crate, not a compatibility code path inside shipping code, and workflow files are outside the marker convention's Rust scope. Remove after the #1638 adjudication lands and converges on one implementation: delete the losing implementation, its feature seam, and the gating CI jobs. +- `multipart-compression-default-off-window` staged multipart disk-compression rollout: releases before the resumable legacy decompressor fail transient reads of compressed objects under mid-payload suspension, so multipart uploads advertise the compression marker only when `RUSTFS_COMPRESSION_MULTIPART_ENABLED` is set in addition to `RUSTFS_COMPRESSION_ENABLED`, keeping rolling upgrades from creating new compressed multipart objects while pre-fix nodes may still serve reads. Flip the default to enabled (and retire the extra switch) after the minimum supported direct-upgrade release ships the resumable `DecompressReader`. ## Review Checklist diff --git a/rustfs/src/app/multipart_usecase.rs b/rustfs/src/app/multipart_usecase.rs index 18295fc00..ade9e1e91 100644 --- a/rustfs/src/app/multipart_usecase.rs +++ b/rustfs/src/app/multipart_usecase.rs @@ -27,7 +27,7 @@ use super::storage_api::multipart_usecase::bucket::{ replication::{must_replicate_object, schedule_object_replication}, versioning_sys::BucketVersioningSys, }; -use super::storage_api::multipart_usecase::compression::is_disk_compressible; +use super::storage_api::multipart_usecase::compression::{is_disk_compressible, is_multipart_disk_compression_enabled}; #[cfg(test)] use super::storage_api::multipart_usecase::contract::http::HTTPPreconditions; use super::storage_api::multipart_usecase::contract::multipart::{CompletePart, MultipartOperations as _, MultipartUploadResult}; @@ -209,9 +209,16 @@ fn create_multipart_upload_metadata( metadata } -/// A multipart session advertises disk compression only when the object key/headers -/// qualify AND the session is not an SSE-C ciphertext-passthrough replication session, -/// which must preserve source bytes verbatim. +/// A multipart session advertises disk compression only when the staged-rollout +/// switch (`RUSTFS_COMPRESSION_MULTIPART_ENABLED`) is on, the object key/headers +/// qualify, AND the session is not an SSE-C ciphertext-passthrough replication +/// session, which must preserve source bytes verbatim. +/// +/// The rollout switch defaults to off so a rolling upgrade never creates new +/// compressed multipart objects while pre-fix nodes (whose decompressor is not +/// resumable) may still serve reads. Enable it once the fleet has converged on a +/// fixed build; the default flips per the `multipart-compression-default-off-window` +/// entry in docs/architecture/compat-cleanup-register.md. /// /// Each part is compressed as an independent stream; the GET path decodes across part /// boundaries (see `ReadTransform::Compressed`), so the session may advertise @@ -220,8 +227,8 @@ fn create_multipart_upload_metadata( /// Unlike single PUT there is no `MIN_DISK_COMPRESSIBLE_SIZE` floor here: the total /// object size is unknown at CreateMultipartUpload time, so tiny multipart objects pay /// the (harmless) framing overhead. This is a deliberate trade-off, not a bug. -fn should_advertise_session_compression(ciphertext_passthrough: bool, disk_compressible: bool) -> bool { - !ciphertext_passthrough && disk_compressible +fn should_advertise_session_compression(multipart_enabled: bool, ciphertext_passthrough: bool, disk_compressible: bool) -> bool { + multipart_enabled && !ciphertext_passthrough && disk_compressible } async fn validate_table_catalog_object_mutation(bucket: &str, key: &str) -> S3Result<()> { @@ -886,7 +893,11 @@ impl DefaultMultipartUsecase { None => (None, None), }; - if should_advertise_session_compression(ciphertext_passthrough, is_disk_compressible(&req.headers, &key)) { + if should_advertise_session_compression( + is_multipart_disk_compression_enabled(), + ciphertext_passthrough, + is_disk_compressible(&req.headers, &key), + ) { rustfs_utils::http::insert_str( &mut metadata, rustfs_utils::http::SUFFIX_COMPRESSION, @@ -1688,19 +1699,25 @@ mod tests { #[test] fn session_compression_is_advertised_only_for_non_passthrough_compressible_uploads() { - // (ciphertext_passthrough, disk_compressible, expected) + // (multipart_enabled, ciphertext_passthrough, disk_compressible, expected) let cases = [ - (false, false, false), - (false, true, true), - (true, false, false), - (true, true, false), + (true, false, false, false), + (true, false, true, true), + (true, true, false, false), + (true, true, true, false), + // The staged-rollout switch keeps multipart compression dark by + // default regardless of the other gates. + (false, false, true, false), + (false, false, false, false), + (false, true, true, false), + (false, true, false, false), ]; - for (ciphertext_passthrough, disk_compressible, expected) in cases { + for (multipart_enabled, ciphertext_passthrough, disk_compressible, expected) in cases { assert_eq!( - should_advertise_session_compression(ciphertext_passthrough, disk_compressible), + should_advertise_session_compression(multipart_enabled, ciphertext_passthrough, disk_compressible), expected, - "ciphertext_passthrough={ciphertext_passthrough} disk_compressible={disk_compressible}" + "multipart_enabled={multipart_enabled} ciphertext_passthrough={ciphertext_passthrough} disk_compressible={disk_compressible}" ); } } diff --git a/rustfs/src/app/storage_api.rs b/rustfs/src/app/storage_api.rs index 91933ddb0..672008b86 100644 --- a/rustfs/src/app/storage_api.rs +++ b/rustfs/src/app/storage_api.rs @@ -940,7 +940,9 @@ pub(crate) mod concurrency { } pub(crate) mod compression { - pub(crate) use crate::storage::storage_api::ecstore_compression::{MIN_DISK_COMPRESSIBLE_SIZE, is_disk_compressible}; + pub(crate) use crate::storage::storage_api::ecstore_compression::{ + MIN_DISK_COMPRESSIBLE_SIZE, is_disk_compressible, is_multipart_disk_compression_enabled, + }; } pub(crate) mod deadlock_detector { diff --git a/rustfs/src/storage/storage_api.rs b/rustfs/src/storage/storage_api.rs index b12169c71..dd64a5691 100644 --- a/rustfs/src/storage/storage_api.rs +++ b/rustfs/src/storage/storage_api.rs @@ -409,7 +409,9 @@ pub(crate) mod ecstore_client { } pub(crate) mod ecstore_compression { - pub(crate) use rustfs_ecstore::api::compression::{MIN_DISK_COMPRESSIBLE_SIZE, is_disk_compressible}; + pub(crate) use rustfs_ecstore::api::compression::{ + MIN_DISK_COMPRESSIBLE_SIZE, is_disk_compressible, is_multipart_disk_compression_enabled, + }; } pub(crate) mod ecstore_cluster { diff --git a/scripts/run.sh b/scripts/run.sh index 55c6a3d6b..7954c2be1 100755 --- a/scripts/run.sh +++ b/scripts/run.sh @@ -209,6 +209,7 @@ export RUSTFS_NS_SCANNER_INTERVAL=60 # Object scanning interval in seconds # Storage level compression (compression at object storage level) # export RUSTFS_COMPRESSION_ENABLED=true # Whether to enable storage-level compression for objects +# export RUSTFS_COMPRESSION_MULTIPART_ENABLED=true # Additionally compress multipart uploads (staged rollout switch: enable only after the whole fleet runs a build with the resumable decompressor; see docs/architecture/compat-cleanup-register.md) # HTTP Response Compression (whitelist-based, aligned with MinIO) # By default, HTTP response compression is DISABLED (aligned with MinIO behavior)