fix(ecstore): admit transformed small objects inline by plaintext size (#7916)

* fix(ecstore): admit transformed small objects inline by plaintext size

* test(e2e): expect the small compressed object inline in compression roundtrip

---------

Co-authored-by: Hauser <housemecn@gmail.com>

crates/e2e_test/src/compression_test.rs

@@ -152,9 +152,29 @@ async fn test_compression_roundtrip() -> Result<(), Box<dyn std::error::Error +
     let content_length = head_response.content_length().unwrap_or(0);
     assert_eq!(content_length as usize, original_size, "Content-Length should be original size");
 
+    // A 5 KiB compressed object sits inside the inline budget, so its shard is
+    // embedded in xl.meta rather than written as a part file; the stored size
+    // recorded in the metadata is the compressed length.
     let part_files = find_part_files(&env.temp_dir, COMPRESSION_TEST_BUCKET, object_key)?;
-    assert!(!part_files.is_empty(), "expected on-disk part files for the compressed object");
-    let total_physical_size = part_files_total_size(&part_files)?;
+    assert!(
+        part_files.is_empty(),
+        "small compressed object must be stored inline, found part files {part_files:?}"
+    );
+    let xl_meta_path = PathBuf::from(&env.temp_dir)
+        .join(COMPRESSION_TEST_BUCKET)
+        .join(object_key)
+        .join("xl.meta");
+    let file_info = rustfs_filemeta::FileMeta::load(&fs::read(&xl_meta_path)?)?.into_fileinfo(
+        COMPRESSION_TEST_BUCKET,
+        object_key,
+        "",
+        true,
+        false,
+        true,
+    )?;
+    assert!(file_info.inline_data(), "xl.meta must carry the inline marker for the compressed object");
+    assert!(file_info.is_compressed(), "xl.meta must carry the compression marker");
+    let total_physical_size = u64::try_from(file_info.size)?;
 
     assert!(
         total_physical_size < original_size as u64,

crates/e2e_test/src/inline_fast_path_cluster_test.rs

@@ -1788,6 +1788,8 @@ async fn four_node_inline_fallback_controls() -> 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(())
 }

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));

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<u8>) -> (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<u8> {
+        b"inline transformed stream payload. "
+            .iter()
+            .copied()
+            .cycle()
+            .take(size)
+            .collect()
+    }
+
+    async fn read_back(set_disks: &Arc<SetDisks>, bucket: &str, object: &str) -> Vec<u8> {
+        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() {

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