diff --git a/crates/e2e_test/src/compression_test.rs b/crates/e2e_test/src/compression_test.rs index f9df6167d..cb22811be 100644 --- a/crates/e2e_test/src/compression_test.rs +++ b/crates/e2e_test/src/compression_test.rs @@ -152,9 +152,29 @@ async fn test_compression_roundtrip() -> Result<(), Box TestResult { configure_reader_metric_cluster(&mut cluster, &collector); let sse_master_key = base64_simd::STANDARD.encode_to_string([0x42u8; 32]); cluster.set_env("RUSTFS_SSE_S3_MASTER_KEY", &sse_master_key); + // Inspect every disk only after the PUT rename fanout has drained. + cluster.set_env("RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE", "false"); cluster.start().await?; let bucket = "inline-fallback-controls"; @@ -1839,6 +1841,10 @@ async fn four_node_inline_fallback_controls() -> TestResult { ), ) .await?; + // 16 KiB of plaintext is 8 KiB per data shard on EC 2+2, inside the inline + // budget: an SSE-S3 PUT is admitted inline by its plaintext size even though + // the ciphertext length is unknown up front (backlog#2393 STOR-115). + assert_storage_layout(&cluster, bucket, encrypted_key, None, true)?; Ok(()) } @@ -1851,6 +1857,8 @@ async fn four_node_compressed_inline_fallback() -> TestResult { let mut cluster = RustFSTestClusterEnvironment::new(4).await?; configure_reader_metric_cluster(&mut cluster, &collector); cluster.set_env("RUSTFS_COMPRESSION_ENABLED", "true"); + // Inspect every disk only after the PUT rename fanout has drained. + cluster.set_env("RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE", "false"); cluster.start().await?; let bucket = "inline-compressed-fallback"; @@ -1871,6 +1879,33 @@ async fn four_node_compressed_inline_fallback() -> TestResult { ReaderPathExpectation::for_class(ReaderObject::new(bucket, key, &body, put.e_tag(), None), LEGACY_DUPLEX, COMPRESSED), ) .await?; + // 64 KiB of plaintext is 32 KiB per data shard on EC 2+2, inside the inline + // budget: a compressed PUT is admitted inline by its plaintext size even + // though its stored size is unknown up front (backlog#2393 STOR-115). + assert_storage_layout(&cluster, bucket, key, None, true)?; + + // A server-side copy onto another compressible key re-compresses the + // decompressed source stream and lands inline the same way (STOR-125). + let copy_key = "compressed/copy.txt"; + let copy = client + .copy_object() + .bucket(bucket) + .key(copy_key) + .copy_source(format!("{bucket}/{key}")) + .send() + .await?; + let copy_etag = copy.copy_object_result().and_then(|result| result.e_tag()).map(str::to_owned); + assert_reader_path( + &collector, + &client, + ReaderPathExpectation::for_class( + ReaderObject::new(bucket, copy_key, &body, copy_etag.as_deref(), None), + LEGACY_DUPLEX, + COMPRESSED, + ), + ) + .await?; + assert_storage_layout(&cluster, bucket, copy_key, None, true)?; Ok(()) } diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index ed1c689f6..4a0927627 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/crates/ecstore/src/set_disk/mod.rs @@ -5053,6 +5053,24 @@ fn known_put_object_storage_size(data_size: i64) -> i64 { } } +/// Shard size the inline admission check evaluates for a single PUT. +/// +/// A compressed or encrypted stream reports `SIZE_PRESERVE_LAYER` as its +/// stored size because the transformed length is only known after the write. +/// MinIO's `putObject` sizes such objects for inline admission by their +/// plaintext `ActualSize`; without that fallback every transformed object, +/// however small, lands in `part.1` files. A stream with neither size known +/// yields a negative shard size, which `should_inline` rejects. +fn inline_admission_shard_size(erasure: &coding::Erasure, stored_size: i64, actual_size: i64) -> i64 { + if stored_size >= 0 { + return erasure.shard_file_size(stored_size); + } + if actual_size > 0 { + return erasure.shard_file_size(actual_size); + } + HashReader::SIZE_PRESERVE_LAYER +} + #[allow(clippy::too_many_arguments)] async fn build_inline_bitrot_readers( files: &[FileInfo], @@ -13170,6 +13188,29 @@ mod tests { )); } + #[test] + fn inline_admission_falls_back_to_actual_size_for_transformed_streams() { + let erasure = coding::Erasure::new(2, 2, 1024 * 1024); + let unknown = HashReader::SIZE_PRESERVE_LAYER; + + // A known stored size is authoritative, whatever the plaintext size says. + assert_eq!( + inline_admission_shard_size(&erasure, 16 * 1024, 4 * 1024 * 1024), + erasure.shard_file_size(16 * 1024) + ); + assert_eq!(inline_admission_shard_size(&erasure, 0, 4 * 1024), 0); + + // A transformed stream is sized by its plaintext length (MinIO parity). + assert_eq!( + inline_admission_shard_size(&erasure, unknown, 16 * 1024), + erasure.shard_file_size(16 * 1024) + ); + + // Neither size known, or an empty transformed stream, cannot be admitted. + assert!(inline_admission_shard_size(&erasure, unknown, 0) < 0); + assert!(inline_admission_shard_size(&erasure, unknown, unknown) < 0); + } + #[test] fn put_object_part_fast_path_selection_matches_single_block_non_inline_rules() { assert!(should_use_single_block_non_inline_fast_path(false, 4096, 4096)); diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index f8b595e87..532b6d165 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/crates/ecstore/src/set_disk/ops/object.rs @@ -48,16 +48,16 @@ use super::super::{ disk, ensure_delete_commit_locks_held, error, explicit_delete_removed_marker, finish_set_disk_read_lock, get_codec_streaming_reader_gate_with_plan, get_object_body_cache_hook, get_raw_etag, get_small_object_direct_memory_decision_with_threshold_and_plan, get_small_object_direct_memory_threshold, - get_stage_timer_if_enabled, get_str, get_transitioned_object_reader_with_tier_manager, inline_erasure_shard_file_offset, - inline_erasure_shard_size, insert_str, is_deadlock_detection_enabled, is_err_object_not_found, is_err_version_not_found, - is_explicit_null_version, is_get_codec_streaming_base_enabled, is_get_small_object_direct_memory_enabled, - is_lock_optimization_enabled, issue3031_diag_enabled, join_all, known_put_object_storage_size, path_join_buf, - put_restore_opts, record_compression_total_memory, record_get_codec_streaming_gate_decision, - record_get_direct_memory_decision, record_get_object_pipeline_failure, record_get_object_pipeline_failure_for_path, - record_get_object_reader_path_observation, record_get_stage_duration_if_enabled, record_lock_acquire, - reduce_write_quorum_errs, release_materialized_read_lock, replication_write_may_pass_worm_gate, require_restore_operation_id, - resolve_delete_version_state, resolve_tiered_decommission_write_quorum_result, resolve_write_layout, - restore_commit_operation_id_from_metadata, restore_operation_id_from_metadata, send_event, + get_stage_timer_if_enabled, get_str, get_transitioned_object_reader_with_tier_manager, inline_admission_shard_size, + inline_erasure_shard_file_offset, inline_erasure_shard_size, insert_str, is_deadlock_detection_enabled, + is_err_object_not_found, is_err_version_not_found, is_explicit_null_version, is_get_codec_streaming_base_enabled, + is_get_small_object_direct_memory_enabled, is_lock_optimization_enabled, issue3031_diag_enabled, join_all, + known_put_object_storage_size, path_join_buf, put_restore_opts, record_compression_total_memory, + record_get_codec_streaming_gate_decision, record_get_direct_memory_decision, record_get_object_pipeline_failure, + record_get_object_pipeline_failure_for_path, record_get_object_reader_path_observation, record_get_stage_duration_if_enabled, + record_lock_acquire, reduce_write_quorum_errs, release_materialized_read_lock, replication_write_may_pass_worm_gate, + require_restore_operation_id, resolve_delete_version_state, resolve_tiered_decommission_write_quorum_result, + resolve_write_layout, restore_commit_operation_id_from_metadata, restore_operation_id_from_metadata, send_event, set_disk_delete_creates_delete_marker, should_force_delete_marker_for_missing_version, should_persist_encryption_original_size, should_preserve_delete_replication_state, should_use_inline_fast_path_with_plan, take_prepared_get_object_metadata, to_object_err, try_read_inline_data_shards_direct, warn, @@ -3591,7 +3591,14 @@ impl SetDisks { let put_object_size = known_put_object_storage_size(data.size()); let shard_file_size_raw = erasure.shard_file_size(put_object_size); - let is_inline_buffer = storage_class_config.should_inline(shard_file_size_raw, erasure.data_shards, opts.versioned); + // Transformed streams (unknown stored size) are admitted by their + // plaintext size and then take the streaming encode with inline + // buffer writers, since the single-block fast path needs a known length. + let is_inline_buffer = storage_class_config.should_inline( + inline_admission_shard_size(&erasure, put_object_size, data.actual_size()), + erasure.data_shards, + opts.versioned, + ); let collect_stage_timing = rustfs_io_metrics::put_stage_metrics_enabled() || issue3031_diag_enabled(); let shard_file_size = shard_file_size_raw; @@ -11452,6 +11459,152 @@ mod inline_put_commit_path_tests { assert_eq!(restored, payload); } + /// Counts `part.*` files under every hermetic disk root. + fn count_part_files(temp_dirs: &[tempfile::TempDir]) -> usize { + fn walk(dir: &std::path::Path, hits: &mut usize) { + let Ok(entries) = std::fs::read_dir(dir) else { + return; + }; + for entry in entries.flatten() { + let path = entry.path(); + if path.is_dir() { + walk(&path, hits); + } else if path + .file_name() + .and_then(|name| name.to_str()) + .is_some_and(|name| name.starts_with("part.")) + { + *hits += 1; + } + } + } + + let mut hits = 0; + for dir in temp_dirs { + walk(dir.path(), &mut hits); + } + hits + } + + /// The reader shape the app layer hands to `put_object` for a compressed + /// single PUT: the stored size is unknown (`SIZE_PRESERVE_LAYER`), the + /// plaintext size is known, and the metadata marks the object compressed so + /// GET decompresses it. Encrypted PUTs present the same shape. + fn transformed_put(plaintext: Vec) -> (PutObjReader, ObjectOptions) { + let actual_size = plaintext.len() as i64; + let plain = HashReader::from_stream(Cursor::new(plaintext), actual_size, actual_size, None, None, false) + .expect("hash reader over plaintext"); + let compressed = crate::io_support::rio::compression_reader(plain, rustfs_utils::CompressionAlgorithm::default(), false); + let stream = HashReader::from_reader(compressed, HashReader::SIZE_PRESERVE_LAYER, actual_size, None, None, false) + .expect("hash reader over transformed stream"); + let mut user_defined = HashMap::new(); + insert_str( + &mut user_defined, + SUFFIX_COMPRESSION, + crate::io_support::rio::compression_metadata_value(rustfs_utils::CompressionAlgorithm::default()), + ); + insert_str(&mut user_defined, SUFFIX_ACTUAL_SIZE, actual_size.to_string()); + let opts = ObjectOptions { + no_lock: true, + user_defined, + ..Default::default() + }; + (PutObjReader::new(stream), opts) + } + + fn compressible_payload(size: usize) -> Vec { + b"inline transformed stream payload. " + .iter() + .copied() + .cycle() + .take(size) + .collect() + } + + async fn read_back(set_disks: &Arc, bucket: &str, object: &str) -> Vec { + let mut object_reader = set_disks + .get_object_reader(bucket, object, None, HeaderMap::new(), &ObjectOptions::default()) + .await + .expect("committed object should be readable"); + let mut restored = Vec::new(); + object_reader + .stream + .read_to_end(&mut restored) + .await + .expect("object should stream"); + restored + } + + #[tokio::test] + async fn transformed_small_put_is_stored_inline_and_round_trips() { + let (temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; + let bucket = "inline-transformed-small"; + let object = "object.txt"; + // 16 KiB plaintext over EC 2+2 is 8 KiB per data shard, inside the + // default 128 KiB inline budget; the stored size is unknown up front. + let plaintext = compressible_payload(16 * 1024); + make_bucket(&disk_stores, bucket).await; + + let (mut reader, opts) = transformed_put(plaintext.clone()); + set_disks + .put_object(bucket, object, &mut reader, &opts) + .await + .expect("transformed PUT should commit"); + + let read_data = ReadOptions { + read_data: true, + ..Default::default() + }; + for (disk_index, disk) in disk_stores.iter().enumerate() { + let file_info = disk + .read_version("", bucket, object, "", &read_data) + .await + .unwrap_or_else(|err| panic!("disk {disk_index} should persist metadata: {err}")); + assert!(file_info.inline_data(), "disk {disk_index} must mark the transformed shard inline"); + assert!( + file_info.data.as_ref().is_some_and(|data| !data.is_empty()), + "disk {disk_index} must embed the shard in xl.meta" + ); + assert!(file_info.is_compressed(), "disk {disk_index} must keep the compression marker"); + } + assert_eq!(count_part_files(&temp_dirs), 0, "an inline transformed object must not leave part files"); + + assert_eq!(read_back(&set_disks, bucket, object).await, plaintext); + } + + #[tokio::test] + async fn transformed_put_above_inline_budget_keeps_part_files() { + let (temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; + let bucket = "inline-transformed-large"; + let object = "object.txt"; + // 512 KiB plaintext is 256 KiB per data shard, above the 128 KiB budget, + // even though the compressed bytes would fit: the plaintext size is the + // admission input, exactly like a known-size PUT of the same object. + let plaintext = compressible_payload(512 * 1024); + make_bucket(&disk_stores, bucket).await; + + let (mut reader, opts) = transformed_put(plaintext.clone()); + set_disks + .put_object(bucket, object, &mut reader, &opts) + .await + .expect("transformed PUT should commit"); + + let read_data = ReadOptions { + read_data: true, + ..Default::default() + }; + for (disk_index, disk) in disk_stores.iter().enumerate() { + let file_info = disk + .read_version("", bucket, object, "", &read_data) + .await + .unwrap_or_else(|err| panic!("disk {disk_index} should persist metadata: {err}")); + assert!(!file_info.inline_data(), "disk {disk_index} must keep the shard outside xl.meta"); + } + assert_eq!(count_part_files(&temp_dirs), disk_stores.len(), "every disk must hold one part file"); + + assert_eq!(read_back(&set_disks, bucket, object).await, plaintext); + } + #[tokio::test] #[serial] async fn get_object_reader_wires_mid_size_to_single_inflight() { diff --git a/docs/architecture/erasure-coding.md b/docs/architecture/erasure-coding.md index 921f845e7..d2aca3cb1 100644 --- a/docs/architecture/erasure-coding.md +++ b/docs/architecture/erasure-coding.md @@ -221,7 +221,7 @@ Fields: `version_id`, `mod_time`, `signature: [u8;4]`, `version_type`, `flags: u ### 6.6 Inline data -Small objects store their payload inline after the container CRC ([filemeta_inline.rs](../../crates/filemeta/src/filemeta_inline.rs)): 1 version byte (`INLINE_DATA_VER = 1`) then a msgpack map of `version-key → bin`. **INVARIANT — the map key** is the version-id string, `"null"` (`NULL_VERSION_ID`, [fileinfo.rs](../../crates/filemeta/src/fileinfo.rs)) for the null/None version, else the lowercase hyphenated UUID. Presence is determined **on read** solely by the `meta_sys[inline-data]` body marker (`FileInfo::inline_data`); the read path gates inline extraction on that marker alone. The header `InlineData` flag is **written** (mirrored from the body on marshal) but is **not** consulted on read, and a disagreement is tolerated — MinIO may leave the header flag unset while inline data is present, so a reader must **not** require the flag and the marker to agree. The inline threshold is `should_inline` ([storageclass.rs](../../crates/ecstore/src/config/storageclass.rs)): inline if `shard_size ≤ inline_block/8` for versioned buckets, else `≤ inline_block`; `DEFAULT_INLINE_BLOCK = 128 KiB`. +Small objects store their payload inline after the container CRC ([filemeta_inline.rs](../../crates/filemeta/src/filemeta_inline.rs)): 1 version byte (`INLINE_DATA_VER = 1`) then a msgpack map of `version-key → bin`. **INVARIANT — the map key** is the version-id string, `"null"` (`NULL_VERSION_ID`, [fileinfo.rs](../../crates/filemeta/src/fileinfo.rs)) for the null/None version, else the lowercase hyphenated UUID. Presence is determined **on read** solely by the `meta_sys[inline-data]` body marker (`FileInfo::inline_data`); the read path gates inline extraction on that marker alone. The header `InlineData` flag is **written** (mirrored from the body on marshal) but is **not** consulted on read, and a disagreement is tolerated — MinIO may leave the header flag unset while inline data is present, so a reader must **not** require the flag and the marker to agree. The inline threshold is `should_inline` ([storageclass.rs](../../crates/ecstore/src/config/storageclass.rs)): inline if `shard_size ≤ inline_block/8` for versioned buckets, else `≤ inline_block`; `DEFAULT_INLINE_BLOCK = 128 KiB`. A compressed or encrypted single PUT has no known stored size up front, so its shard size is derived from the plaintext `actual_size` (`inline_admission_shard_size`, MinIO `putObject` parity); such objects are inlined through the streaming encoder with in-memory bitrot writers rather than the single-block fast path. ---