fix(ecstore): optimize ObjectInfo clone and fix critical TODOs (#3149)

* fix(ecstore): optimize ObjectInfo clone and fix critical TODOs

## ObjectInfo hot-path clone optimization (issue #653 item 2)
- Wrap user_defined (HashMap), user_tags (String), parts (Vec) in Arc
- Clone cost reduced from ~20 heap allocations to O(1) ref count bump
- Updated 11 downstream access sites with explicit deref where needed

## Critical TODO/FIXME fixes (issue #653 item 7)
- rpc/peer_s3_client.rs: descriptive error message for empty peer response
- set_disk/write.rs: concurrent rollback deletes via tokio::spawn + join_all
- store_list_objects.rs: resolved FIXME with explanation + 4 regression tests
- store/bucket.rs: namespace write locks for make_bucket and delete_bucket

Skipped TODOs (too risky without broader context):
- multipart.rs:30 nslock — causes lock timeout in existing tests
- bucket.rs:88 cached list_bucket — needs cache invalidation strategy
- bucket.rs:149 replication delete — needs replication subsystem integration

1134 tests pass (1 flaky under parallel execution, passes individually).

* fix: address PR #3149 review comments

- Preserve NamespaceLockQuorumUnavailable variant in bucket create/delete
- Fix cancellation test to actually test cancellation (keep senders open)
- Update ObjectInfo consumers in app/ for Arc-backed fields
- Update bucket_usecase.rs user_tags to Arc<String>

* fix: update ObjectInfo Arc consumers

* fix: address ecstore merge review comments

* fix: cancel blocked merge output sends

* fix(ecstore): satisfy ObjectInfo clone clippy

crates/ecstore/src/set_disk/replication.rs

@@ -25,7 +25,7 @@ impl SetDisks {
         let mut oi = obj_info.clone();
         oi.metadata_only = true;
 
-        oi.user_defined.remove(X_AMZ_RESTORE.as_str());
+        Arc::make_mut(&mut oi.user_defined).remove(X_AMZ_RESTORE.as_str());
 
         let version_id = oi.version_id.map(|v| v.to_string());
         let _obj = self

crates/ecstore/src/set_disk/write.rs

@@ -383,30 +383,35 @@ impl SetDisks {
         }
 
         if let Some(err) = reduce_write_quorum_errs(&errs, OBJECT_OP_IGNORED_ERRS, write_quorum) {
-            // TODO: add concurrency
+            let mut revert_futures = Vec::with_capacity(disks.len());
             for (i, err) in errs.iter().enumerate() {
                 if err.is_some() {
                     continue;
                 }
 
                 if let Some(disk) = disks[i].as_ref() {
-                    let _ = disk
-                        .delete(
-                            bucket,
-                            &path_join_buf(&[prefix, STORAGE_FORMAT_FILE]),
-                            DeleteOptions {
-                                recursive: true,
-                                ..Default::default()
-                            },
-                        )
-                        .await
-                        .map_err(|e| {
-                            warn!("write meta revert err {:?}", e);
-                            e
-                        });
+                    let disk = disk.clone();
+                    let bucket = bucket.to_string();
+                    let path = path_join_buf(&[prefix, STORAGE_FORMAT_FILE]);
+                    revert_futures.push(async move {
+                        if let Err(err) = disk
+                            .delete(
+                                &bucket,
+                                &path,
+                                DeleteOptions {
+                                    recursive: true,
+                                    ..Default::default()
+                                },
+                            )
+                            .await
+                        {
+                            warn!("write meta revert err {:?}", err);
+                        }
+                    });
                 }
             }
 
+            join_all(revert_futures).await;
             return Err(err);
         }
         Ok(())