perf(ecstore): reduce inline PUT commit overhead (#6033)

* perf(metrics): attribute PUT stage costs

Co-Authored-By: heihutu <heihutu@gmail.com>

* perf(ecstore): move PUT metadata during shuffle

Co-Authored-By: heihutu <heihutu@gmail.com>

* perf(s3): reuse PUT object lock state

Co-Authored-By: heihutu <heihutu@gmail.com>

* perf(ecstore): trim PUT metadata fanout clones

Build per-disk PUT metadata only for committed writer slots, move the response metadata out of the fanout vector, and preserve fresh FileInfo shuffle semantics.

Co-Authored-By: heihutu <heihutu@gmail.com>

* perf(metrics): make PUT stage attribution opt-in

Co-Authored-By: heihutu <heihutu@gmail.com>

* perf(ecstore): commit inline PUT shards directly

Co-Authored-By: heihutu <heihutu@gmail.com>

* perf(ecstore): streamline rename staging cleanup

Use the directory-specific removal operation for rename_data staging parents. This avoids a guaranteed failed file-removal probe on Unix-like hosts and lets Windows remove the empty directory directly while preserving best-effort non-empty handling.

Co-Authored-By: heihutu <heihutu@gmail.com>

* test(ecstore): cover inline PUT rename failures

Cache the detailed stage metrics gate once per PUT and exercise exact-quorum and quorum-minus-one failures after inline shard encoding.

Co-Authored-By: heihutu <heihutu@gmail.com>

---------

Co-authored-by: heihutu <heihutu@gmail.com>

crates/ecstore/src/set_disk/metadata.rs

@@ -1079,6 +1079,25 @@ impl SetDisks {
         shuffled_disks
     }
 
+    pub(super) fn shuffle_disks_owned(mut disks: Vec<Option<DiskStore>>, distribution: &[usize]) -> Vec<Option<DiskStore>> {
+        if distribution.is_empty() {
+            return disks;
+        }
+
+        let mut shuffled_disks = vec![None; disks.len()];
+        for (index, disk) in disks.iter_mut().enumerate() {
+            let Some(slot) = distribution
+                .get(index)
+                .and_then(|block_index| block_index.checked_sub(1))
+                .filter(|slot| *slot < shuffled_disks.len())
+            else {
+                continue;
+            };
+            shuffled_disks[slot] = disk.take();
+        }
+        shuffled_disks
+    }
+
     pub(super) fn shuffle_check_parts(parts_errs: &[usize], distribution: &[usize]) -> Vec<usize> {
         if distribution.is_empty() {
             return parts_errs.to_vec();
@@ -1390,6 +1409,23 @@ mod tests {
         assert_eq!(owned_slots, expected_slots, "fallback disk slots must match the borrowing variant");
     }
 
+    #[tokio::test]
+    async fn owned_shuffle_preserves_fresh_put_metadata() {
+        let tempdir = tempfile::tempdir().expect("tempdir should be created");
+        let fi = FileInfo::new("bucket/object", 2, 1);
+        let parts = vec![fi.clone(); fi.erasure.distribution.len()];
+        let disks = shuffle_test_disks(&tempdir, parts.len()).await;
+
+        let (owned_disks, owned_parts) = SetDisks::shuffle_disks_and_parts_metadata_by_index_owned(disks, parts, &fi);
+
+        assert!(owned_disks.iter().all(Option::is_some), "fresh PUT must retain every online disk");
+        assert_eq!(
+            owned_parts,
+            vec![fi; owned_disks.len()],
+            "fresh PUT metadata with pending shard indexes must survive init fallback"
+        );
+    }
+
     // backlog#949: corrupt/adversarial distribution values (0 or > N) must not
     // trigger a `usize` underflow / out-of-bounds panic in the shuffle helpers.
     #[test]
@@ -1419,6 +1455,22 @@ mod tests {
         assert_eq!(result.len(), disks.len(), "output length must be preserved");
     }
 
+    #[tokio::test]
+    async fn owned_disk_shuffle_matches_borrowing_variant() {
+        let tempdir = tempfile::tempdir().expect("tempdir should be created");
+        let mut disks = shuffle_test_disks(&tempdir, 4).await;
+        disks[1] = None;
+        disks[3] = None;
+        let distribution = [3, 1, 4, 2];
+
+        let expected = SetDisks::shuffle_disks(&disks, &distribution);
+        let actual = SetDisks::shuffle_disks_owned(disks, &distribution);
+
+        let expected_slots = expected.iter().map(Option::is_some).collect::<Vec<_>>();
+        let actual_slots = actual.iter().map(Option::is_some).collect::<Vec<_>>();
+        assert_eq!(actual_slots, expected_slots, "owned shuffle must preserve disk placement");
+    }
+
     #[tokio::test]
     async fn shuffle_disks_and_parts_metadata_survives_corrupt_distribution() {
         let tempdir = tempfile::tempdir().expect("tempdir should be created");