perf(ecstore): reduce small PUT fixed costs (#5987)

crates/ecstore/src/erasure/coding/encode.rs

@@ -91,6 +91,11 @@ fn use_bytesmut_ingest() -> bool {
     })
 }
 
+fn small_ingest_capacity(erasure: &Erasure, size_hint: usize) -> usize {
+    let data_len = size_hint.min(erasure.block_size);
+    erasure.encoded_capacity_for_data_len(data_len).min(erasure.block_size)
+}
+
 /// Keeps the encoder producer scoped to its parent future. Tokio detaches a
 /// task when its `JoinHandle` is dropped, so the producer must be aborted when
 /// an upload is cancelled before the encode pipeline finishes.
@@ -540,13 +545,14 @@ impl Erasure {
         writers: &mut [Option<BitrotWriterWrapper>],
         quorum: usize,
         require_single_block: bool,
+        size_hint: usize,
     ) -> std::io::Result<(R, usize)>
     where
         R: AsyncRead + Send + Sync + Unpin,
     {
         use tokio::io::AsyncReadExt;
 
-        let mut buf = Vec::with_capacity(self.block_size);
+        let mut buf = Vec::with_capacity(small_ingest_capacity(&self, size_hint));
         let total = if require_single_block {
             let read_limit = self
                 .block_size
@@ -880,7 +886,24 @@ impl Erasure {
     where
         R: AsyncRead + Send + Sync + Unpin,
     {
-        self.encode_small_direct(reader, writers, quorum, false).await
+        let size_hint = self.block_size;
+        self.encode_small_direct(reader, writers, quorum, false, size_hint).await
+    }
+
+    /// Size-aware inline fast path. `size_hint` only controls the bounded initial
+    /// allocation; reads remain authoritative.
+    #[hotpath::measure(impl_type = "Erasure")]
+    pub async fn encode_inline_small_with_size_hint<R>(
+        self: Arc<Self>,
+        reader: R,
+        writers: &mut [Option<BitrotWriterWrapper>],
+        quorum: usize,
+        size_hint: usize,
+    ) -> std::io::Result<(R, usize)>
+    where
+        R: AsyncRead + Send + Sync + Unpin,
+    {
+        self.encode_small_direct(reader, writers, quorum, false, size_hint).await
     }
 
     /// Fast path for single-block non-inline objects: avoids the producer/consumer
@@ -895,7 +918,24 @@ impl Erasure {
     where
         R: AsyncRead + Send + Sync + Unpin,
     {
-        self.encode_small_direct(reader, writers, quorum, true).await
+        let size_hint = self.block_size;
+        self.encode_small_direct(reader, writers, quorum, true, size_hint).await
+    }
+
+    /// Size-aware single-block fast path. `size_hint` only controls the bounded
+    /// initial allocation; reads remain authoritative.
+    #[hotpath::measure(impl_type = "Erasure")]
+    pub async fn encode_single_block_non_inline_with_size_hint<R>(
+        self: Arc<Self>,
+        reader: R,
+        writers: &mut [Option<BitrotWriterWrapper>],
+        quorum: usize,
+        size_hint: usize,
+    ) -> std::io::Result<(R, usize)>
+    where
+        R: AsyncRead + Send + Sync + Unpin,
+    {
+        self.encode_small_direct(reader, writers, quorum, true, size_hint).await
     }
 }
 
@@ -2293,7 +2333,10 @@ mod tests {
 
         let erasure = Arc::new(Erasure::new(1, 0, 16));
         let reader = tokio::io::BufReader::new(Cursor::new(Vec::<u8>::new()));
-        let (_reader, total) = erasure.encode_inline_small(reader, &mut writers, 1).await.unwrap();
+        let (_reader, total) = erasure
+            .encode_inline_small_with_size_hint(reader, &mut writers, 1, 0)
+            .await
+            .unwrap();
 
         assert_eq!(total, 0);
         // No shutdown was called, so nothing should be committed
@@ -2325,7 +2368,10 @@ mod tests {
         let payload = b"hello inline small";
         let erasure = Arc::new(Erasure::new(DATA_SHARDS, PARITY_SHARDS, BLOCK_SIZE));
         let reader = tokio::io::BufReader::new(Cursor::new(payload.to_vec()));
-        let (_reader, total) = erasure.encode_inline_small(reader, &mut writers, DATA_SHARDS).await.unwrap();
+        let (_reader, total) = erasure
+            .encode_inline_small_with_size_hint(reader, &mut writers, DATA_SHARDS, 1)
+            .await
+            .unwrap();
 
         assert_eq!(total, payload.len());
         // All shards must have received data (shutdown flushed the bitrot header + shard bytes)
@@ -2392,7 +2438,7 @@ mod tests {
         let erasure = Arc::new(Erasure::new(DATA_SHARDS, PARITY_SHARDS, BLOCK_SIZE));
         let reader = tokio::io::BufReader::new(Cursor::new(payload));
         let err = erasure
-            .encode_single_block_non_inline(reader, &mut writers, DATA_SHARDS)
+            .encode_single_block_non_inline_with_size_hint(reader, &mut writers, DATA_SHARDS, BLOCK_SIZE)
             .await
             .expect_err("single-block fast path must reject oversized readers");
 
@@ -2403,6 +2449,21 @@ mod tests {
         }
     }
 
+    #[test]
+    fn small_ingest_capacity_uses_bounded_size_hint() {
+        let erasure = Erasure::new(4, 2, 1024 * 1024);
+        assert_eq!(small_ingest_capacity(&erasure, 0), 0);
+        assert_eq!(small_ingest_capacity(&erasure, 4 * 1024), 6 * 1024);
+        assert_eq!(small_ingest_capacity(&erasure, 16 * 1024), 24 * 1024);
+        assert_eq!(small_ingest_capacity(&erasure, usize::MAX), 1024 * 1024);
+
+        let legacy = Erasure::new_with_options(4, 2, 1024 * 1024, true);
+        assert_eq!(small_ingest_capacity(&legacy, 4 * 1024), 6 * 1024);
+
+        let high_parity = Erasure::new(4, 12, 1024 * 1024);
+        assert_eq!(small_ingest_capacity(&high_parity, usize::MAX), 1024 * 1024);
+    }
+
     #[tokio::test]
     async fn read_full_buf_or_eof_returns_none_on_empty_reader() {
         let mut reader = Cursor::new(Vec::<u8>::new());

crates/ecstore/src/erasure/coding/erasure.rs

@@ -968,6 +968,15 @@ impl Erasure {
         self.data_shards + self.parity_shards
     }
 
+    pub(crate) fn encoded_capacity_for_data_len(&self, data_len: usize) -> usize {
+        let shard_size_fn = if self.uses_legacy {
+            calc_shard_size_legacy
+        } else {
+            calc_shard_size
+        };
+        shard_size_fn(data_len, self.data_shards).saturating_mul(self.total_shard_count())
+    }
+
     /// Whether the erasure dimensions are safe for the shard/offset arithmetic.
     ///
     /// `block_size` and `data_shards` come straight from on-disk metadata; a

crates/ecstore/src/set_disk/core/io_primitives.rs

@@ -58,6 +58,7 @@ use crate::io_support::bitrot::{
     create_deferred_bitrot_reader_with_stripe_handle, object_mmap_read_enabled, object_mmap_read_max_length,
 };
 use crate::set_disk::shard_source::ShardReadCost;
+use futures::FutureExt as _;
 use futures::stream::{FuturesUnordered, StreamExt};
 use metrics::counter;
 use std::{
@@ -2856,8 +2857,6 @@ impl SetDisks {
                 file_info.validate_for_erasure_write()?;
             }
         }
-        let mut futures = Vec::with_capacity(disks.len());
-
         let mut errs = Vec::with_capacity(disks.len());
 
         let src_bucket = Arc::new(src_bucket.to_string());
@@ -2865,48 +2864,65 @@ impl SetDisks {
         let dst_bucket = Arc::new(dst_bucket.to_string());
         let dst_object = Arc::new(dst_object.to_string());
 
-        for (i, (disk, file_info)) in disks.iter().zip(file_infos.iter()).enumerate() {
-            let mut file_info = file_info.clone();
-            let disk = disk.clone();
-            let src_bucket = src_bucket.clone();
-            let src_object = src_object.clone();
-            let dst_object = dst_object.clone();
-            let dst_bucket = dst_bucket.clone();
+        let disk_count = disks.len();
+        let fanout_disks = disks.to_vec();
+        let fanout_file_infos = file_infos.to_vec();
+        let fanout_src_bucket = src_bucket.clone();
+        let fanout_src_object = src_object.clone();
+        let fanout_dst_bucket = dst_bucket.clone();
+        let fanout_dst_object = dst_object.clone();
+        // Keep one coordinator task so a cancelled caller cannot drop partially
+        // completed disk mutations. Per-disk futures stay ordered in `join_all`,
+        // preserving slot-indexed quorum and convergence accounting without a
+        // scheduler task for every disk.
+        let fanout = tokio::spawn(async move {
+            let futures = fanout_disks
+                .into_iter()
+                .zip(fanout_file_infos)
+                .enumerate()
+                .map(|(i, (disk, mut file_info))| {
+                    let src_bucket = fanout_src_bucket.clone();
+                    let src_object = fanout_src_object.clone();
+                    let dst_object = fanout_dst_object.clone();
+                    let dst_bucket = fanout_dst_bucket.clone();
 
-            futures.push(tokio::spawn(async move {
-                // Test-only introspection guard: counts this task as in-flight for
-                // the whole body. Compiles to `()` in production (no behavior).
-                #[allow(clippy::let_unit_value)]
-                let _fanout_task_guard = Self::rename_fanout_task_guard(&dst_object);
+                    std::panic::AssertUnwindSafe(async move {
+                        // Test-only introspection guard: counts this operation as
+                        // in-flight for the whole body. Compiles to `()` in production.
+                        #[allow(clippy::let_unit_value)]
+                        let _fanout_task_guard = Self::rename_fanout_task_guard(&dst_object);
 
-                let Some(disk) = disk else {
-                    return Err(DiskError::DiskNotFound);
-                };
+                        let Some(disk) = disk else {
+                            return Err(DiskError::DiskNotFound);
+                        };
 
-                let is_delete_marker = file_info.is_canonical_delete_marker();
-                if file_info.erasure.index == 0 {
-                    file_info.erasure.index = i + 1;
-                }
+                        let is_delete_marker = file_info.is_canonical_delete_marker();
+                        if file_info.erasure.index == 0 {
+                            file_info.erasure.index = i + 1;
+                        }
 
-                if !is_delete_marker && !file_info.has_valid_erasure_geometry() {
-                    return Err(DiskError::FileCorrupt);
-                }
+                        if !is_delete_marker && !file_info.has_valid_erasure_geometry() {
+                            return Err(DiskError::FileCorrupt);
+                        }
 
-                // Test-only awaitable pause point right before the disk rename.
-                // A no-op immediately-ready future in production.
-                Self::rename_fanout_barrier(&dst_object, i, rename_fanout_barrier_phase::RENAME).await;
+                        // Test-only awaitable pause point right before the disk rename.
+                        // A no-op immediately-ready future in production.
+                        Self::rename_fanout_barrier(&dst_object, i, rename_fanout_barrier_phase::RENAME).await;
 
-                disk.rename_data(&src_bucket, &src_object, file_info, &dst_bucket, &dst_object)
-                    .await
-            }));
-        }
+                        disk.rename_data(&src_bucket, &src_object, file_info, &dst_bucket, &dst_object)
+                            .await
+                    })
+                    .catch_unwind()
+                });
+            join_all(futures).await
+        });
 
-        let mut disk_versions = vec![None; disks.len()];
-        let mut data_dirs = vec![None; disks.len()];
-        let mut cleanup_data_dirs = vec![None; disks.len()];
-        let mut old_current_sizes = vec![None; disks.len()];
+        let mut disk_versions = vec![None; disk_count];
+        let mut data_dirs = vec![None; disk_count];
+        let mut cleanup_data_dirs = vec![None; disk_count];
+        let mut old_current_sizes = vec![None; disk_count];
 
-        let results = join_all(futures).await;
+        let results = fanout.await.map_err(|_| DiskError::Unexpected)?;
 
         for (idx, result) in results.iter().enumerate() {
             match result.as_ref().map_err(|_| DiskError::Unexpected)? {
@@ -5867,6 +5883,51 @@ mod tests {
         drop(dirs);
     }
 
+    #[tokio::test]
+    async fn rename_fanout_drains_after_caller_cancellation() {
+        const DISKS: usize = 4;
+        let bucket = "rename-cancel-bucket";
+        let object = "rename-cancel-object";
+        let (dirs, disks) = call_counter_local_disks(bucket, DISKS).await;
+        let marker = metadata_test_delete_marker(object, Uuid::new_v4(), OffsetDateTime::now_utc());
+        let file_infos = vec![marker; DISKS];
+        let tracker = rename_fanout_barrier::observe_tasks(object);
+        let barrier = rename_fanout_barrier::arm(object, 0, rename_fanout_barrier::PHASE_RENAME);
+
+        let rename =
+            tokio::spawn(
+                async move { SetDisks::rename_data(&disks, bucket, object, &file_infos, bucket, object, DISKS - 1).await },
+            );
+        tokio::time::timeout(BARRIER_PAUSE_GUARD, barrier.wait_until_paused())
+            .await
+            .expect("rename fan-out must reach the armed barrier");
+        rename.abort();
+        assert!(
+            rename
+                .await
+                .expect_err("aborted caller should report cancellation")
+                .is_cancelled(),
+            "caller task should be cancelled, not panic"
+        );
+        assert!(tracker.running() >= 1, "the coordinator must retain in-flight disk mutations");
+
+        barrier.release();
+        tokio::time::timeout(BARRIER_PAUSE_GUARD, async {
+            while tracker.running() != 0 {
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .expect("cancelled caller's disk mutations must drain");
+
+        for (idx, dir) in dirs.iter().enumerate() {
+            assert!(
+                dir.path().join(bucket).join(object).join(STORAGE_FORMAT_FILE).exists(),
+                "disk {idx} must finish the rename after caller cancellation"
+            );
+        }
+    }
+
     /// Demo / regression guard for the barrier on the commit (old-data-dir)
     /// cleanup fan-out. Serves the same #1312/#1319 "no background disk write
     /// after release" shape, on the reclamation path that runs *after* a write is

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

@@ -954,12 +954,17 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
 
             let write_path = classify_multipart_part_write_path(multipart_part_size, fi.erasure.block_size);
             rustfs_io_metrics::record_put_object_path(write_path.multipart_metric_label());
+            let small_size_hint = if matches!(write_path, SmallWritePath::SingleBlockNonInline) {
+                usize::try_from(multipart_part_size).map_err(Error::other)?
+            } else {
+                0
+            };
             let encode_stage_start = rustfs_io_metrics::put_stage_metrics_enabled().then(Instant::now);
 
             let (reader, w_size) = match write_path {
                 SmallWritePath::SingleBlockNonInline => {
                     Arc::clone(&erasure)
-                        .encode_single_block_non_inline(stream, &mut writers, write_quorum)
+                        .encode_single_block_non_inline_with_size_hint(stream, &mut writers, write_quorum, small_size_hint)
                         .await?
                 }
                 SmallWritePath::PipelineBatchedLarge => {

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

@@ -56,6 +56,22 @@ use http::HeaderValue;
 use rustfs_utils::path::decode_dir_object;
 use std::future::Future;
 
+#[inline]
+fn duration_millis_f64(duration: std::time::Duration) -> f64 {
+    duration.as_secs_f64() * 1000.0
+}
+
+#[cfg(test)]
+mod duration_metrics_tests {
+    use super::duration_millis_f64;
+    use std::time::Duration;
+
+    #[test]
+    fn duration_millis_preserves_sub_millisecond_precision() {
+        assert_eq!(duration_millis_f64(Duration::from_micros(125)), 0.125);
+    }
+}
+
 fn is_restore_control_metadata(key: &str) -> bool {
     key.eq_ignore_ascii_case(X_AMZ_RESTORE.as_str())
         || key.eq_ignore_ascii_case(rustfs_utils::http::headers::AMZ_RESTORE_EXPIRY_DAYS)
@@ -1107,8 +1123,12 @@ impl SetDisks {
                 writers.push(w);
                 errors.push(e);
             }
-            let writer_setup_ms = writer_setup_stage_start.elapsed().as_millis() as u64;
-            rustfs_io_metrics::record_put_object_stage_duration("set_disk_writer_setup", writer_setup_ms as f64);
+            let writer_setup_elapsed = writer_setup_stage_start.elapsed();
+            let writer_setup_ms = writer_setup_elapsed.as_millis() as u64;
+            rustfs_io_metrics::record_put_object_stage_duration(
+                "set_disk_writer_setup",
+                duration_millis_f64(writer_setup_elapsed),
+            );
 
             let nil_count = errors.iter().filter(|&e| e.is_none()).count();
             if nil_count < write_quorum {
@@ -1138,11 +1158,16 @@ impl SetDisks {
 
             let write_path = classify_put_write_path(is_inline_buffer, put_object_size, fi.erasure.block_size);
             rustfs_io_metrics::record_put_object_path(write_path.metric_label());
+            let small_size_hint = if matches!(write_path, SmallWritePath::Inline | SmallWritePath::SingleBlockNonInline) {
+                usize::try_from(put_object_size).map_err(Error::other)?
+            } else {
+                0
+            };
 
             let encode_stage_start = Instant::now();
             let (reader, w_size) = match write_path {
                 SmallWritePath::Inline => match Arc::clone(&erasure)
-                    .encode_inline_small(stream, &mut writers, write_quorum)
+                    .encode_inline_small_with_size_hint(stream, &mut writers, write_quorum, small_size_hint)
                     .await
                 {
                     Ok((r, w)) => (r, w),
@@ -1152,7 +1177,7 @@ impl SetDisks {
                     }
                 },
                 SmallWritePath::SingleBlockNonInline => match Arc::clone(&erasure)
-                    .encode_single_block_non_inline(stream, &mut writers, write_quorum)
+                    .encode_single_block_non_inline_with_size_hint(stream, &mut writers, write_quorum, small_size_hint)
                     .await
                 {
                     Ok((r, w)) => (r, w),
@@ -1178,8 +1203,9 @@ impl SetDisks {
                     }
                 },
             };
-            let encode_ms = encode_stage_start.elapsed().as_millis() as u64;
-            rustfs_io_metrics::record_put_object_stage_duration("set_disk_encode", encode_ms as f64);
+            let encode_elapsed = encode_stage_start.elapsed();
+            let encode_ms = encode_elapsed.as_millis() as u64;
+            rustfs_io_metrics::record_put_object_stage_duration("set_disk_encode", duration_millis_f64(encode_elapsed));
 
             let _ = mem::replace(&mut data.stream, reader);
             // if let Err(err) = close_bitrot_writers(&mut writers).await {
@@ -1497,8 +1523,9 @@ impl SetDisks {
                     let _ = rustfs_common::heal_channel::send_heal_request(request).await;
                 });
             }
-            let rename_stage_ms = rename_stage_start.elapsed().as_millis() as u64;
-            rustfs_io_metrics::record_put_object_stage_duration("set_disk_rename", rename_stage_ms as f64);
+            let rename_stage_elapsed = rename_stage_start.elapsed();
+            let rename_stage_ms = rename_stage_elapsed.as_millis() as u64;
+            rustfs_io_metrics::record_put_object_stage_duration("set_disk_rename", duration_millis_f64(rename_stage_elapsed));
             if (rename_stage_ms as u128) >= SET_DISK_COMMIT_TAIL_WARN_THRESHOLD_MS {
                 warn!(
                     event = EVENT_SET_DISK_COMMIT_TAIL_SLOW,
@@ -1527,9 +1554,13 @@ impl SetDisks {
                 let cleanup = self
                     .commit_rename_data_dir(&cleanup_disks, bucket, object, &old_dir.to_string(), &committed_dir, write_quorum)
                     .await;
-                let cleanup_ms = cleanup_stage_start.elapsed().as_millis() as u64;
+                let cleanup_elapsed = cleanup_stage_start.elapsed();
+                let cleanup_ms = cleanup_elapsed.as_millis() as u64;
                 cleanup_stage_ms = Some(cleanup_ms);
-                rustfs_io_metrics::record_put_object_stage_duration("set_disk_old_data_cleanup", cleanup_ms as f64);
+                rustfs_io_metrics::record_put_object_stage_duration(
+                    "set_disk_old_data_cleanup",
+                    duration_millis_f64(cleanup_elapsed),
+                );
                 self.report_old_data_dir_cleanup(bucket, object, &old_dir.to_string(), &cleanup)
                     .await;
                 if (cleanup_ms as u128) >= SET_DISK_COMMIT_TAIL_WARN_THRESHOLD_MS {