feat(ecstore): implement decommission and rebalance (#2281)

Co-authored-by: weisd <im@weisd.in>
Co-authored-by: houseme <housemecn@gmail.com>

crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs

@@ -716,6 +716,12 @@ pub async fn validate_transition_tier(lc: &BucketLifecycleConfiguration) -> Resu
     Ok(())
 }
 
+fn mark_delete_opts_skip_decommissioned_on_remote_success(opts: &mut ObjectOptions, remote_delete_succeeded: bool) {
+    if remote_delete_succeeded {
+        opts.skip_decommissioned = true;
+    }
+}
+
 pub async fn enqueue_transition_immediate(oi: &ObjectInfo, src: LcEventSrc) {
     if let Some(lc) = GLOBAL_LifecycleSys.get(&oi.bucket).await {
         enqueue_transition_with_lifecycle(oi, &lc, &src).await;
@@ -795,11 +801,10 @@ pub async fn expire_transitioned_object(
         &oi.transitioned_object.tier,
     )
     .await;
-    if ret.is_ok() {
-        opts.skip_decommissioned = true;
-    } else {
+    if ret.is_err() {
         //transitionLogIf(ctx, err);
     }
+    mark_delete_opts_skip_decommissioned_on_remote_success(&mut opts, ret.is_ok());
 
     let dobj = match api.delete_object(&oi.bucket, &oi.name, opts).await {
         Ok(obj) => obj,
@@ -1278,3 +1283,39 @@ pub async fn apply_lifecycle_action(event: &lifecycle::Event, src: &LcEventSrc,
     }
     success
 }
+
+#[cfg(test)]
+mod tests {
+    use super::mark_delete_opts_skip_decommissioned_on_remote_success;
+    use crate::store_api::ObjectOptions;
+
+    #[test]
+    fn mark_delete_opts_skip_decommissioned_on_remote_success_sets_flag_on_success() {
+        let mut opts = ObjectOptions::default();
+
+        mark_delete_opts_skip_decommissioned_on_remote_success(&mut opts, true);
+
+        assert!(opts.skip_decommissioned);
+    }
+
+    #[test]
+    fn mark_delete_opts_skip_decommissioned_on_remote_success_preserves_false_on_failure() {
+        let mut opts = ObjectOptions::default();
+
+        mark_delete_opts_skip_decommissioned_on_remote_success(&mut opts, false);
+
+        assert!(!opts.skip_decommissioned);
+    }
+
+    #[test]
+    fn mark_delete_opts_skip_decommissioned_on_remote_success_preserves_existing_true_on_failure() {
+        let mut opts = ObjectOptions {
+            skip_decommissioned: true,
+            ..ObjectOptions::default()
+        };
+
+        mark_delete_opts_skip_decommissioned_on_remote_success(&mut opts, false);
+
+        assert!(opts.skip_decommissioned);
+    }
+}

crates/ecstore/src/data_movement.rs

@@ -0,0 +1,401 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use crate::error::{Error, Result};
+use crate::store::ECStore;
+use crate::store_api::{CompletePart, GetObjectReader, MultipartOperations, ObjectIO, ObjectInfo, ObjectOptions, PutObjReader};
+use bytes::Bytes;
+use rustfs_rio::{EtagResolvable, HashReader, HashReaderDetector, Index, Reader, TryGetIndex, WarpReader};
+use std::io::Cursor;
+use std::pin::Pin;
+use std::sync::{
+    Arc,
+    atomic::{AtomicBool, Ordering},
+};
+use std::task::{Context, Poll};
+use tokio::io::{AsyncRead, AsyncReadExt, BufReader, ReadBuf};
+use tracing::error;
+
+pub struct IndexedDataMovementReader<R> {
+    inner: R,
+    index: Option<Index>,
+}
+
+impl<R> IndexedDataMovementReader<R> {
+    pub fn new(inner: R, index: Option<Index>) -> Self {
+        Self { inner, index }
+    }
+}
+
+impl<R: AsyncRead + Unpin + Send + Sync> AsyncRead for IndexedDataMovementReader<R> {
+    fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<std::io::Result<()>> {
+        Pin::new(&mut self.inner).poll_read(cx, buf)
+    }
+}
+
+impl<R: AsyncRead + Unpin + Send + Sync> EtagResolvable for IndexedDataMovementReader<R> {}
+
+impl<R: AsyncRead + Unpin + Send + Sync> HashReaderDetector for IndexedDataMovementReader<R> {}
+
+impl<R: AsyncRead + Unpin + Send + Sync> TryGetIndex for IndexedDataMovementReader<R> {
+    fn try_get_index(&self) -> Option<&Index> {
+        self.index.as_ref()
+    }
+}
+
+impl<R: AsyncRead + Unpin + Send + Sync> Reader for IndexedDataMovementReader<R> {}
+
+pub fn decode_part_index(index: Option<&Bytes>) -> Option<Index> {
+    let bytes = index?;
+    let mut decoded = Index::new();
+    if decoded.load(bytes.as_ref()).is_ok() {
+        Some(decoded)
+    } else {
+        None
+    }
+}
+
+pub fn put_obj_reader_from_chunk(chunk: Vec<u8>, size: i64, actual_size: i64, index: Option<Index>) -> Result<PutObjReader> {
+    use sha2::{Digest, Sha256};
+
+    let sha256hex = if !chunk.is_empty() {
+        Some(hex_simd::encode_to_string(Sha256::digest(&chunk), hex_simd::AsciiCase::Lower))
+    } else {
+        None
+    };
+
+    let reader = IndexedDataMovementReader::new(WarpReader::new(Cursor::new(chunk)), index);
+    let hash_reader = HashReader::new(Box::new(reader), size, actual_size, None, sha256hex, false)?;
+    Ok(PutObjReader::new(hash_reader))
+}
+
+pub fn new_multipart_abort_flag() -> Arc<AtomicBool> {
+    Arc::new(AtomicBool::new(true))
+}
+
+pub fn should_abort_multipart_upload(flag: &Arc<AtomicBool>) -> bool {
+    flag.load(Ordering::Relaxed)
+}
+
+pub fn mark_multipart_upload_completed(flag: &Arc<AtomicBool>) {
+    flag.store(false, Ordering::Relaxed);
+}
+
+fn data_movement_new_multipart_opts(object_info: &ObjectInfo, src_pool_idx: usize) -> ObjectOptions {
+    ObjectOptions {
+        versioned: object_info.version_id.is_some(),
+        version_id: object_info.version_id.as_ref().map(|v| v.to_string()),
+        user_defined: object_info.user_defined.clone(),
+        preserve_etag: object_info.etag.clone(),
+        src_pool_idx,
+        data_movement: true,
+        ..Default::default()
+    }
+}
+
+fn data_movement_complete_multipart_opts(object_info: &ObjectInfo) -> ObjectOptions {
+    ObjectOptions {
+        versioned: object_info.version_id.is_some(),
+        version_id: object_info.version_id.as_ref().map(|v| v.to_string()),
+        data_movement: true,
+        mod_time: object_info.mod_time,
+        preserve_etag: object_info.etag.clone(),
+        ..Default::default()
+    }
+}
+
+fn data_movement_put_object_opts(object_info: &ObjectInfo, src_pool_idx: usize) -> ObjectOptions {
+    ObjectOptions {
+        versioned: object_info.version_id.is_some(),
+        src_pool_idx,
+        data_movement: true,
+        version_id: object_info.version_id.as_ref().map(|v| v.to_string()),
+        mod_time: object_info.mod_time,
+        user_defined: object_info.user_defined.clone(),
+        preserve_etag: object_info.etag.clone(),
+        ..Default::default()
+    }
+}
+
+fn resolve_data_movement_abort_result(
+    op_label: &str,
+    bucket: &str,
+    object: &str,
+    upload_id: &str,
+    primary_err: Error,
+    abort_err: Error,
+) -> Error {
+    Error::other(format!(
+        "{op_label}: abort_multipart_upload failed for {bucket}/{object} upload {upload_id} after error {primary_err}: {abort_err}"
+    ))
+}
+
+pub(crate) async fn migrate_object(
+    store: Arc<ECStore>,
+    pool_idx: usize,
+    bucket: String,
+    rd: GetObjectReader,
+    op_label: &str,
+) -> Result<()> {
+    let object_info = rd.object_info.clone();
+
+    if object_info.is_multipart() {
+        let res = match store
+            .new_multipart_upload(&bucket, &object_info.name, &data_movement_new_multipart_opts(&object_info, pool_idx))
+            .await
+        {
+            Ok(res) => res,
+            Err(err) => {
+                error!("{op_label}: new_multipart_upload err {:?}", &err);
+                return Err(err);
+            }
+        };
+
+        let abort_multipart_flag = new_multipart_abort_flag();
+        let multipart_result: Result<()> = async {
+            let mut parts = vec![CompletePart::default(); object_info.parts.len()];
+            let mut reader = rd.stream;
+
+            for (i, part) in object_info.parts.iter().enumerate() {
+                let mut chunk = vec![0u8; part.size];
+                reader.read_exact(&mut chunk).await?;
+
+                let part_size = i64::try_from(part.size).map_err(|_| Error::other("part size overflow"))?;
+                let part_actual_size = if part.actual_size > 0 { part.actual_size } else { part_size };
+                let index = decode_part_index(part.index.as_ref());
+                let mut data = put_obj_reader_from_chunk(chunk, part_size, part_actual_size, index)?;
+
+                let pi = match store
+                    .put_object_part(
+                        &bucket,
+                        &object_info.name,
+                        &res.upload_id,
+                        part.number,
+                        &mut data,
+                        &ObjectOptions {
+                            preserve_etag: Some(part.etag.clone()),
+                            ..Default::default()
+                        },
+                    )
+                    .await
+                {
+                    Ok(pi) => pi,
+                    Err(err) => {
+                        error!("{op_label}: put_object_part {i} err {:?}", &err);
+                        return Err(err);
+                    }
+                };
+
+                parts[i] = CompletePart {
+                    part_num: pi.part_num,
+                    etag: pi.etag,
+                    ..Default::default()
+                };
+            }
+
+            if let Err(err) = store
+                .clone()
+                .complete_multipart_upload(
+                    &bucket,
+                    &object_info.name,
+                    &res.upload_id,
+                    parts,
+                    &data_movement_complete_multipart_opts(&object_info),
+                )
+                .await
+            {
+                error!("{op_label}: complete_multipart_upload err {:?}", &err);
+                return Err(err);
+            }
+
+            mark_multipart_upload_completed(&abort_multipart_flag);
+            Ok(())
+        }
+        .await;
+
+        if let Err(primary_err) = multipart_result {
+            if should_abort_multipart_upload(&abort_multipart_flag) {
+                return match store
+                    .abort_multipart_upload(&bucket, &object_info.name, &res.upload_id, &ObjectOptions::default())
+                    .await
+                {
+                    Ok(()) => Err(primary_err),
+                    Err(abort_err) => {
+                        error!("{op_label}: abort_multipart_upload err {:?}", &abort_err);
+                        Err(resolve_data_movement_abort_result(
+                            op_label,
+                            bucket.as_str(),
+                            object_info.name.as_str(),
+                            res.upload_id.as_str(),
+                            primary_err,
+                            abort_err,
+                        ))
+                    }
+                };
+            }
+            return Err(primary_err);
+        }
+
+        return Ok(());
+    }
+
+    let actual_size = object_info.get_actual_size()?;
+    let index = object_info
+        .parts
+        .first()
+        .and_then(|part| decode_part_index(part.index.as_ref()));
+    let reader = IndexedDataMovementReader::new(WarpReader::new(BufReader::new(rd.stream)), index);
+    let hrd = HashReader::new(Box::new(reader), object_info.size, actual_size, object_info.etag.clone(), None, false)?;
+    let mut data = PutObjReader::new(hrd);
+
+    if let Err(err) = store
+        .put_object(
+            &bucket,
+            &object_info.name,
+            &mut data,
+            &data_movement_put_object_opts(&object_info, pool_idx),
+        )
+        .await
+    {
+        error!("{op_label}: put_object err {:?}", &err);
+        return Err(err);
+    }
+
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use time::OffsetDateTime;
+    use uuid::Uuid;
+
+    #[test]
+    fn test_new_multipart_abort_flag_defaults_to_abort_enabled() {
+        let flag = new_multipart_abort_flag();
+        assert!(should_abort_multipart_upload(&flag));
+    }
+
+    #[test]
+    fn test_mark_multipart_upload_completed_disables_abort_cleanup() {
+        let flag = new_multipart_abort_flag();
+        mark_multipart_upload_completed(&flag);
+        assert!(!should_abort_multipart_upload(&flag));
+    }
+
+    #[test]
+    fn test_resolve_data_movement_abort_result_wraps_abort_context() {
+        let err = resolve_data_movement_abort_result(
+            "rebalance_object",
+            "bucket-a",
+            "object-a",
+            "upload-1",
+            Error::SlowDown,
+            Error::OperationCanceled,
+        );
+        let message = err.to_string();
+        assert!(message.contains("rebalance_object: abort_multipart_upload failed"));
+        assert!(message.contains("bucket-a/object-a"));
+        assert!(message.contains("upload upload-1"));
+        assert!(message.contains(Error::SlowDown.to_string().as_str()));
+    }
+
+    #[test]
+    fn test_decode_part_index_returns_none_when_absent() {
+        assert!(decode_part_index(None).is_none());
+    }
+
+    #[test]
+    fn test_decode_part_index_returns_none_for_invalid_payload() {
+        let invalid = Bytes::from_static(b"not-a-valid-index");
+        assert!(decode_part_index(Some(&invalid)).is_none());
+    }
+
+    #[test]
+    fn test_decode_part_index_returns_some_for_valid_payload() {
+        let mut index = Index::new();
+        index.add(0, 0).expect("first index entry should be accepted");
+        index
+            .add(2_097_152, 2_097_152)
+            .expect("second index entry should advance totals");
+
+        let encoded = index.into_vec();
+        let decoded = decode_part_index(Some(&encoded)).expect("valid index payload should decode");
+
+        assert_eq!(decoded.total_uncompressed, 2_097_152);
+        assert_eq!(decoded.total_compressed, 2_097_152);
+    }
+
+    #[test]
+    fn test_data_movement_new_multipart_opts_preserves_etag_and_version() {
+        let version_id = Uuid::nil();
+        let object_info = ObjectInfo {
+            version_id: Some(version_id),
+            etag: Some("etag-value".to_string()),
+            user_defined: std::collections::HashMap::from([("x-amz-meta-key".to_string(), "value".to_string())]),
+            ..Default::default()
+        };
+
+        let opts = data_movement_new_multipart_opts(&object_info, 7);
+
+        assert!(opts.versioned);
+        assert_eq!(opts.version_id.as_deref(), Some(version_id.to_string().as_str()));
+        assert_eq!(opts.preserve_etag.as_deref(), Some("etag-value"));
+        assert_eq!(opts.user_defined.get("x-amz-meta-key").map(String::as_str), Some("value"));
+        assert_eq!(opts.src_pool_idx, 7);
+        assert!(opts.data_movement);
+    }
+
+    #[test]
+    fn test_data_movement_complete_multipart_opts_preserves_mod_time_version_and_etag() {
+        let mod_time = OffsetDateTime::now_utc();
+        let version_id = Uuid::nil();
+        let object_info = ObjectInfo {
+            version_id: Some(version_id),
+            mod_time: Some(mod_time),
+            etag: Some("etag-value".to_string()),
+            ..Default::default()
+        };
+
+        let opts = data_movement_complete_multipart_opts(&object_info);
+
+        assert!(opts.versioned);
+        assert!(opts.data_movement);
+        assert_eq!(opts.mod_time, Some(mod_time));
+        assert_eq!(opts.version_id.as_deref(), Some(version_id.to_string().as_str()));
+        assert_eq!(opts.preserve_etag.as_deref(), Some("etag-value"));
+    }
+
+    #[test]
+    fn test_data_movement_put_object_opts_preserves_version_and_etag() {
+        let version_id = Uuid::nil();
+        let object_info = ObjectInfo {
+            version_id: Some(version_id),
+            mod_time: Some(OffsetDateTime::UNIX_EPOCH),
+            etag: Some("etag-value".to_string()),
+            user_defined: std::collections::HashMap::from([("x-amz-meta-key".to_string(), "value".to_string())]),
+            ..Default::default()
+        };
+
+        let opts = data_movement_put_object_opts(&object_info, 9);
+
+        assert!(opts.versioned);
+        assert_eq!(opts.version_id.as_deref(), Some(version_id.to_string().as_str()));
+        assert_eq!(opts.preserve_etag.as_deref(), Some("etag-value"));
+        assert_eq!(opts.user_defined.get("x-amz-meta-key").map(String::as_str), Some("value"));
+        assert_eq!(opts.src_pool_idx, 9);
+        assert!(opts.data_movement);
+        assert_eq!(opts.mod_time, object_info.mod_time);
+    }
+}

crates/ecstore/src/error.rs

@@ -144,6 +144,10 @@ pub enum StorageError {
     DecommissionNotStarted,
     #[error("Decommission already running")]
     DecommissionAlreadyRunning,
+    #[error("Rebalance already running")]
+    RebalanceAlreadyRunning,
+    #[error("Operation canceled")]
+    OperationCanceled,
     #[error("No heal required")]
     NoHealRequired,
     #[error("DoneForNow")]
@@ -414,6 +418,8 @@ impl Clone for StorageError {
             StorageError::EntityTooSmall(a, b, c) => StorageError::EntityTooSmall(*a, *b, *c),
             StorageError::DoneForNow => StorageError::DoneForNow,
             StorageError::DecommissionAlreadyRunning => StorageError::DecommissionAlreadyRunning,
+            StorageError::RebalanceAlreadyRunning => StorageError::RebalanceAlreadyRunning,
+            StorageError::OperationCanceled => StorageError::OperationCanceled,
             StorageError::ErasureReadQuorum => StorageError::ErasureReadQuorum,
             StorageError::ErasureWriteQuorum => StorageError::ErasureWriteQuorum,
             StorageError::NotFirstDisk => StorageError::NotFirstDisk,
@@ -482,6 +488,8 @@ impl StorageError {
             StorageError::InvalidPart(_, _, _) => 0x2E,
             StorageError::DoneForNow => 0x2F,
             StorageError::DecommissionAlreadyRunning => 0x30,
+            StorageError::RebalanceAlreadyRunning => 0x40,
+            StorageError::OperationCanceled => 0x41,
             StorageError::ErasureReadQuorum => 0x31,
             StorageError::ErasureWriteQuorum => 0x32,
             StorageError::NotFirstDisk => 0x33,
@@ -554,6 +562,8 @@ impl StorageError {
             0x2E => Some(StorageError::InvalidPart(Default::default(), Default::default(), Default::default())),
             0x2F => Some(StorageError::DoneForNow),
             0x30 => Some(StorageError::DecommissionAlreadyRunning),
+            0x40 => Some(StorageError::RebalanceAlreadyRunning),
+            0x41 => Some(StorageError::OperationCanceled),
             0x31 => Some(StorageError::ErasureReadQuorum),
             0x32 => Some(StorageError::ErasureWriteQuorum),
             0x33 => Some(StorageError::NotFirstDisk),
@@ -682,6 +692,22 @@ pub fn is_err_data_movement_overwrite(err: &Error) -> bool {
     matches!(err, &StorageError::DataMovementOverwriteErr(_, _, _))
 }
 
+pub fn is_err_decommission_running(err: &Error) -> bool {
+    matches!(err, &StorageError::DecommissionAlreadyRunning)
+}
+
+pub fn is_err_rebalance_running(err: &Error) -> bool {
+    matches!(err, &StorageError::RebalanceAlreadyRunning)
+}
+
+pub fn is_err_operation_canceled(err: &Error) -> bool {
+    matches!(err, &StorageError::OperationCanceled)
+}
+
+pub fn is_err_not_initialized(err: &Error) -> bool {
+    err.to_string().contains("errServerNotInitialized") || err.to_string().contains("ServerNotInitialized")
+}
+
 pub fn is_err_io(err: &Error) -> bool {
     matches!(err, &StorageError::Io(_))
 }
@@ -944,6 +970,8 @@ mod tests {
         assert_eq!(StorageError::VolumeExists.to_u32(), 0x05);
         assert_eq!(StorageError::FileNotFound.to_u32(), 0x06);
         assert_eq!(StorageError::DecommissionAlreadyRunning.to_u32(), 0x30);
+        assert_eq!(StorageError::RebalanceAlreadyRunning.to_u32(), 0x40);
+        assert_eq!(StorageError::OperationCanceled.to_u32(), 0x41);
     }
 
     #[test]
@@ -956,6 +984,8 @@ mod tests {
         assert!(matches!(StorageError::from_u32(0x03), Some(StorageError::DiskFull)));
         assert!(matches!(StorageError::from_u32(0x04), Some(StorageError::VolumeNotFound)));
         assert!(matches!(StorageError::from_u32(0x30), Some(StorageError::DecommissionAlreadyRunning)));
+        assert!(matches!(StorageError::from_u32(0x40), Some(StorageError::RebalanceAlreadyRunning)));
+        assert!(matches!(StorageError::from_u32(0x41), Some(StorageError::OperationCanceled)));
 
         // Test invalid code returns None
         assert!(StorageError::from_u32(0xFF).is_none());
@@ -980,6 +1010,20 @@ mod tests {
         assert_ne!(bucket1, disk_error);
     }
 
+    #[test]
+    fn test_error_running_state_helpers() {
+        assert!(is_err_decommission_running(&StorageError::DecommissionAlreadyRunning));
+        assert!(!is_err_decommission_running(&StorageError::RebalanceAlreadyRunning));
+
+        assert!(is_err_rebalance_running(&StorageError::RebalanceAlreadyRunning));
+        assert!(!is_err_rebalance_running(&StorageError::DecommissionAlreadyRunning));
+        assert!(is_err_operation_canceled(&StorageError::OperationCanceled));
+        assert!(!is_err_operation_canceled(&StorageError::RebalanceAlreadyRunning));
+        assert!(is_err_not_initialized(&StorageError::other("errServerNotInitialized")));
+        assert!(is_err_not_initialized(&StorageError::other("ServerNotInitialized")));
+        assert!(!is_err_not_initialized(&StorageError::DecommissionAlreadyRunning));
+    }
+
     #[test]
     fn test_storage_error_from_disk_error() {
         // Test conversion from DiskError
@@ -1067,6 +1111,8 @@ mod tests {
             StorageError::BucketExists("test".to_string()),
             StorageError::ObjectNotFound("bucket".to_string(), "object".to_string()),
             StorageError::DecommissionAlreadyRunning,
+            StorageError::RebalanceAlreadyRunning,
+            StorageError::OperationCanceled,
         ];
 
         for original_error in test_errors {

crates/ecstore/src/lib.rs

@@ -22,6 +22,7 @@ pub mod bucket;
 pub mod cache_value;
 pub mod compress;
 pub mod config;
+mod data_movement;
 pub mod data_usage;
 pub mod disk;
 pub mod disks_layout;

crates/ecstore/src/notification_sys.rs

@@ -17,6 +17,7 @@ use crate::admin_server_info::get_commit_id;
 use crate::error::{Error, Result};
 use crate::global::{GLOBAL_BOOT_TIME, get_global_endpoints};
 use crate::metrics_realtime::{CollectMetricsOpts, MetricType};
+use crate::rebalance::RebalSaveOpt;
 use crate::rpc::PeerRestClient;
 use crate::{endpoints::EndpointServerPools, new_object_layer_fn};
 use futures::future::join_all;
@@ -376,67 +377,112 @@ impl NotificationSys {
         join_all(futures).await
     }
 
-    pub async fn reload_pool_meta(&self) {
+    pub async fn reload_pool_meta(&self) -> Result<()> {
+        let mut failures = Vec::new();
         let mut futures = Vec::with_capacity(self.peer_clients.len());
-        for client in self.peer_clients.iter().flatten() {
-            futures.push(client.reload_pool_meta());
-        }
-
-        let results = join_all(futures).await;
-        for result in results {
-            if let Err(err) = result {
-                error!("notification reload_pool_meta err {:?}", err);
+        for (idx, client) in self.peer_clients.iter().enumerate() {
+            if let Some(client) = client {
+                let host = client.grid_host.clone();
+                futures.push(async move { client.reload_pool_meta().await.map_err(|err| (host, err)) });
+            } else {
+                failures.push(format!("peer[{idx}] reload_pool_meta failed: peer is not reachable"));
             }
         }
+
+        for result in join_all(futures).await {
+            if let Err((host, err)) = result {
+                let failure = format!("peer {host} reload_pool_meta failed: {err}");
+                error!("notification reload_pool_meta err {}", failure);
+                failures.push(failure);
+            }
+        }
+
+        aggregate_notification_failures("reload_pool_meta", failures)
     }
 
     #[tracing::instrument(skip(self))]
-    pub async fn load_rebalance_meta(&self, start: bool) {
+    pub async fn load_rebalance_meta(&self, start: bool) -> Result<()> {
+        let operation = format!("load_rebalance_meta(start={start})");
+        let mut failures = Vec::new();
         let mut futures = Vec::with_capacity(self.peer_clients.len());
-        for (i, client) in self.peer_clients.iter().flatten().enumerate() {
-            warn!(
-                "notification load_rebalance_meta start: {}, index: {}, client: {:?}",
-                start, i, client.host
-            );
-            futures.push(client.load_rebalance_meta(start));
+        for (idx, client) in self.peer_clients.iter().enumerate() {
+            if let Some(client) = client {
+                warn!(
+                    "notification load_rebalance_meta start: {}, index: {}, client: {:?}",
+                    start, idx, client.host
+                );
+                let host = client.grid_host.clone();
+                futures.push(async move { client.load_rebalance_meta(start).await.map_err(|err| (host, err)) });
+            } else {
+                failures.push(format!("peer[{idx}] {operation} failed: peer is not reachable"));
+            }
         }
 
-        let results = join_all(futures).await;
-        for result in results {
-            if let Err(err) = result {
-                error!("notification load_rebalance_meta err {:?}", err);
+        for result in join_all(futures).await {
+            if let Err((host, err)) = result {
+                let failure = format!("peer {host} {operation} failed: {err}");
+                error!("notification load_rebalance_meta err {}", failure);
+                failures.push(failure);
             } else {
                 warn!("notification load_rebalance_meta success");
             }
         }
+
+        aggregate_notification_failures("load_rebalance_meta", failures)
     }
 
-    pub async fn stop_rebalance(&self) {
+    pub async fn stop_rebalance(&self) -> Result<()> {
         warn!("notification stop_rebalance start");
         let Some(store) = new_object_layer_fn() else {
             error!("stop_rebalance: not init");
-            return;
+            return Err(Error::other("stop_rebalance: object layer not initialized"));
         };
 
         // warn!("notification stop_rebalance load_rebalance_meta");
         // self.load_rebalance_meta(false).await;
         // warn!("notification stop_rebalance load_rebalance_meta done");
 
+        let mut failures = Vec::new();
+
         let mut futures = Vec::with_capacity(self.peer_clients.len());
-        for client in self.peer_clients.iter().flatten() {
-            futures.push(client.stop_rebalance());
+        for (idx, client) in self.peer_clients.iter().enumerate() {
+            if let Some(client) = client {
+                let host = client.grid_host.clone();
+                futures.push(async move { client.stop_rebalance().await.map_err(|err| (host, err)) });
+            } else {
+                failures.push(format!("peer[{idx}] stop_rebalance failed: peer is not reachable"));
+            }
         }
 
-        let results = join_all(futures).await;
-        for result in results {
-            if let Err(err) = result {
-                error!("notification stop_rebalance err {:?}", err);
+        for result in join_all(futures).await {
+            if let Err((host, err)) = result {
+                let failure = format!("peer {host} stop_rebalance failed: {err}");
+                error!("notification stop_rebalance err {}", failure);
+                failures.push(failure);
             }
         }
 
         warn!("notification stop_rebalance stop_rebalance start");
-        let _ = store.stop_rebalance().await;
+        match store.stop_rebalance().await {
+            Ok(_) => {
+                if let Err(err) = store.save_rebalance_stats(usize::MAX, RebalSaveOpt::StoppedAt).await {
+                    error!("notification stop_rebalance local save err {:?}", err);
+                    return Err(Error::other(format!(
+                        "local stop_rebalance save_rebalance_stats(stopped_at) failed: {err}"
+                    )));
+                }
+            }
+            Err(err) => {
+                error!("notification stop_rebalance local stop err {:?}", err);
+                return Err(Error::other(format!("local stop_rebalance stop failed: {err}")));
+            }
+        }
+
+        if let Err(err) = aggregate_notification_failures("stop_rebalance", failures) {
+            warn!("{err}");
+        }
         warn!("notification stop_rebalance stop_rebalance done");
+        Ok(())
     }
 
     pub async fn load_bucket_metadata(&self, bucket: &str) -> Vec<NotificationPeerErr> {
@@ -773,6 +819,18 @@ fn get_offline_disks(offline_host: &str, endpoints: &EndpointServerPools) -> Vec
     offline_disks
 }
 
+fn aggregate_notification_failures(operation: &str, failures: Vec<String>) -> Result<()> {
+    if failures.is_empty() {
+        return Ok(());
+    }
+
+    Err(Error::other(format!(
+        "{operation} encountered {} failure(s): {}",
+        failures.len(),
+        failures.join(" | ")
+    )))
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -825,4 +883,24 @@ mod tests {
 
         assert_eq!(result.endpoint, "fallback");
     }
+
+    #[test]
+    fn aggregate_notification_failures_returns_ok_when_empty() {
+        assert!(aggregate_notification_failures("stop_rebalance", Vec::new()).is_ok());
+    }
+
+    #[test]
+    fn aggregate_notification_failures_returns_joined_error_when_non_empty() {
+        let err = aggregate_notification_failures(
+            "load_rebalance_meta",
+            vec!["peer-1 failed".to_string(), "local save failed".to_string()],
+        )
+        .expect_err("non-empty failures should return error");
+
+        let msg = err.to_string();
+        assert!(msg.contains("load_rebalance_meta"));
+        assert!(msg.contains("2 failure(s)"));
+        assert!(msg.contains("peer-1 failed"));
+        assert!(msg.contains("local save failed"));
+    }
 }

crates/ecstore/src/set_disk.rs

@@ -246,6 +246,19 @@ fn build_tiered_decommission_file_info(
     (updated, write_quorum)
 }
 
+fn resolve_tiered_decommission_write_quorum_result(
+    errs: &[Option<DiskError>],
+    write_quorum: usize,
+    bucket: &str,
+    object: &str,
+) -> Result<()> {
+    if let Some(err) = reduce_write_quorum_errs(errs, OBJECT_OP_IGNORED_ERRS, write_quorum) {
+        return Err(to_object_err(err.into(), vec![bucket, object]));
+    }
+
+    Ok(())
+}
+
 #[derive(Clone, Debug)]
 pub struct SetDisks {
     pub locker_owner: String,
@@ -1200,10 +1213,7 @@ impl ObjectOperations for SetDisks {
             }
         }
 
-        if let Some(err) = reduce_write_quorum_errs(&errs, OBJECT_OP_IGNORED_ERRS, write_quorum) {
-            return Err(err.into());
-        }
-        Ok(())
+        resolve_tiered_decommission_write_quorum_result(&errs, write_quorum, bucket, object)
     }
 
     #[tracing::instrument(skip(self))]
@@ -2158,11 +2168,7 @@ impl SetDisks {
             }
         }
 
-        if let Some(err) = reduce_write_quorum_errs(&errs, OBJECT_OP_IGNORED_ERRS, write_quorum) {
-            return Err(to_object_err(err.into(), vec![bucket, object]));
-        }
-
-        Ok(())
+        resolve_tiered_decommission_write_quorum_result(&errs, write_quorum, bucket, object)
     }
 }
 
@@ -4463,6 +4469,31 @@ mod tests {
         assert_ne!(updated.erasure.distribution, original.erasure.distribution);
     }
 
+    #[test]
+    fn test_resolve_tiered_decommission_write_quorum_result_allows_successful_quorum() {
+        let errs = vec![None, None, Some(DiskError::DiskNotFound), None];
+
+        let result = resolve_tiered_decommission_write_quorum_result(&errs, 3, "bucket", "object");
+
+        assert!(result.is_ok());
+    }
+
+    #[test]
+    fn test_resolve_tiered_decommission_write_quorum_result_wraps_object_context() {
+        let errs = vec![
+            Some(DiskError::DiskNotFound),
+            Some(DiskError::DiskNotFound),
+            Some(DiskError::DiskNotFound),
+            Some(DiskError::DiskNotFound),
+        ];
+
+        let err = resolve_tiered_decommission_write_quorum_result(&errs, 3, "bucket", "object").expect_err("expected error");
+        let rendered = err.to_string();
+
+        assert!(rendered.contains("bucket"), "{rendered}");
+        assert!(rendered.contains("object"), "{rendered}");
+    }
+
     #[test]
     fn test_should_prevent_write() {
         let oi = ObjectInfo {

crates/ecstore/src/store/init.rs

@@ -13,8 +13,84 @@
 // limitations under the License.
 
 use super::*;
+use crate::error::is_err_decommission_running;
 use crate::global::is_first_cluster_node_local;
 
+fn should_resume_local_decommission(endpoints: &EndpointServerPools, idx: usize) -> Result<bool> {
+    let pool = endpoints.as_ref().get(idx).ok_or_else(|| {
+        Error::other(format!(
+            "store init failed to resolve decommission resume pool index {idx} from current endpoints"
+        ))
+    })?;
+    let endpoint = pool.endpoints.as_ref().first().ok_or_else(|| {
+        Error::other(format!(
+            "store init failed to resolve decommission resume pool index {idx}: no endpoints available"
+        ))
+    })?;
+
+    Ok(endpoint.is_local)
+}
+
+const LOCAL_DECOMMISSION_RESUME_MAX_CONFIG_RETRIES: usize = 6;
+const LOCAL_DECOMMISSION_INITIAL_RESUME_DELAY: Duration = Duration::from_secs(60 * 3);
+const LOCAL_DECOMMISSION_RESUME_RETRY_DELAY: Duration = Duration::from_secs(30);
+
+fn should_retry_local_decommission_resume(err: &Error, attempt: usize) -> bool {
+    matches!(err, Error::ConfigNotFound) && attempt < LOCAL_DECOMMISSION_RESUME_MAX_CONFIG_RETRIES
+}
+
+async fn wait_for_local_decommission_resume_delay(rx: &CancellationToken, delay: Duration) -> bool {
+    tokio::select! {
+        _ = rx.cancelled() => false,
+        _ = tokio::time::sleep(delay) => true,
+    }
+}
+
+fn resolve_store_init_stage_result(result: Result<()>, stage: &str) -> Result<()> {
+    result.map_err(|err| Error::other(format!("store init failed during {stage}: {err}")))
+}
+
+async fn resume_local_decommission_after_init(store: Arc<ECStore>, rx: CancellationToken, pool_indices: Vec<usize>) {
+    for attempt in 0..=LOCAL_DECOMMISSION_RESUME_MAX_CONFIG_RETRIES {
+        if rx.is_cancelled() {
+            return;
+        }
+
+        match store.decommission(rx.clone(), pool_indices.clone()).await {
+            Ok(()) => return,
+            Err(err) if is_err_decommission_running(&err) => {
+                if let Err(spawn_err) = store
+                    .spawn_decommission_routines(store.clone(), rx.clone(), pool_indices.clone())
+                    .await
+                {
+                    error!(
+                        "store init failed to resume decommission workers for pools {:?}: {}",
+                        pool_indices, spawn_err
+                    );
+                }
+                return;
+            }
+            Err(err) if should_retry_local_decommission_resume(&err, attempt) => {
+                warn!(
+                    "store init decommission resume missing config for pools {:?}, retry {}/{}: {}",
+                    pool_indices,
+                    attempt + 1,
+                    LOCAL_DECOMMISSION_RESUME_MAX_CONFIG_RETRIES + 1,
+                    err
+                );
+                tokio::select! {
+                    _ = rx.cancelled() => return,
+                    _ = tokio::time::sleep(LOCAL_DECOMMISSION_RESUME_RETRY_DELAY) => {}
+                }
+            }
+            Err(err) => {
+                error!("store init failed to resume decommission for pools {:?}: {}", pool_indices, err);
+                return;
+            }
+        }
+    }
+}
+
 impl ECStore {
     #[allow(clippy::new_ret_no_self)]
     #[instrument(level = "debug", skip(endpoint_pools))]
@@ -87,7 +163,7 @@ impl ECStore {
                         Ok(fm) => break Ok(fm),
                         // Wrap the final error if we are giving up
                         Err(e) if times >= 10 => {
-                            break Err(Error::other(format!("can not get formats after {} retries, last error: {e}", times)));
+                            break Err(Error::other(format!("store init failed to load formats after {times} retries: {e}")));
                         }
                         // Retrying so just drop the error
                         Err(_) => {}
@@ -118,10 +194,10 @@ impl ECStore {
             }
 
             if deployment_id != Some(fm.id) {
-                return Err(Error::other("deployment_id not same in one pool"));
+                return Err(Error::other("store init failed: deployment IDs do not match across pools"));
             }
 
-            if deployment_id.is_some() && deployment_id.unwrap().is_nil() {
+            if deployment_id.is_some_and(|id| id.is_nil()) {
                 deployment_id = Some(Uuid::new_v4());
             }
 
@@ -152,7 +228,7 @@ impl ECStore {
 
         let decommission_cancelers = RwLock::new(vec![None; pools.len()]);
         let ec = Arc::new(ECStore {
-            id: deployment_id.unwrap(),
+            id: deployment_id.ok_or_else(|| Error::other("store init failed: deployment id is not initialized"))?,
             disk_map,
             pools,
             peer_sys,
@@ -177,7 +253,7 @@ impl ECStore {
                 sleep(Duration::from_secs(wait_sec)).await;
 
                 if exit_count > 10 {
-                    return Err(Error::other("ec init failed"));
+                    return Err(Error::other("store init failed: init retry budget exhausted"));
                 }
 
                 exit_count += 1;
@@ -197,13 +273,25 @@ impl ECStore {
     pub async fn init(self: &Arc<Self>, rx: CancellationToken) -> Result<()> {
         GLOBAL_BOOT_TIME.get_or_init(|| async { SystemTime::now() }).await;
 
-        if self.load_rebalance_meta().await.is_ok() {
-            self.start_rebalance().await;
+        resolve_store_init_stage_result(self.load_rebalance_meta().await, "load_rebalance_meta")?;
+        if self.rebalance_meta.read().await.is_some() {
+            resolve_store_init_stage_result(self.start_rebalance().await, "start_rebalance")?;
         }
 
         let mut meta = PoolMeta::default();
-        meta.load(self.pools[0].clone(), self.pools.clone()).await?;
+        resolve_store_init_stage_result(
+            meta.load(
+                self.pools
+                    .first()
+                    .cloned()
+                    .ok_or_else(|| Error::other("store init failed: no storage pools available"))?,
+                self.pools.clone(),
+            )
+            .await,
+            "load_pool_meta",
+        )?;
         let update = meta.validate(self.pools.clone())?;
+        let endpoints = get_global_endpoints();
         let should_persist_pool_meta = is_first_cluster_node_local().await;
 
         if !update {
@@ -213,8 +301,10 @@ impl ECStore {
             }
         } else {
             let new_meta = PoolMeta::new(&self.pools, &meta);
+            // Only one local node should persist validated pool metadata here; otherwise
+            // distributed startup can race on the same lock and replay the prior init bug.
             if should_persist_pool_meta {
-                new_meta.save(self.pools.clone()).await?;
+                resolve_store_init_stage_result(new_meta.save(self.pools.clone()).await, "save_validated_pool_meta")?;
             }
             {
                 let mut pool_meta = self.pool_meta.write().await;
@@ -225,14 +315,12 @@ impl ECStore {
         let pools = meta.return_resumable_pools();
         let mut pool_indices = Vec::with_capacity(pools.len());
 
-        let endpoints = get_global_endpoints();
-
         for p in pools.iter() {
             if let Some(idx) = endpoints.get_pool_idx(&p.cmd_line) {
                 pool_indices.push(idx);
             } else {
                 return Err(Error::other(format!(
-                    "unexpected state present for decommission status pool({}) not found",
+                    "store init failed to resolve resumable decommission pool `{}` from current endpoints",
                     p.cmd_line
                 )));
             }
@@ -240,25 +328,14 @@ impl ECStore {
 
         if !pool_indices.is_empty() {
             let idx = pool_indices[0];
-            if endpoints.as_ref()[idx].endpoints.as_ref()[0].is_local {
+            if should_resume_local_decommission(&endpoints, idx)? {
                 let store = self.clone();
 
                 tokio::spawn(async move {
-                    // wait  3 minutes for cluster init
-                    tokio::time::sleep(Duration::from_secs(60 * 3)).await;
-
-                    if let Err(err) = store.decommission(rx.clone(), pool_indices.clone()).await {
-                        if err == StorageError::DecommissionAlreadyRunning {
-                            for i in pool_indices.iter() {
-                                store.do_decommission_in_routine(rx.clone(), *i).await;
-                            }
-                            return;
-                        }
-
-                        error!("store init decommission err: {}", err);
-
-                        // TODO: check config err
+                    if !wait_for_local_decommission_resume_delay(&rx, LOCAL_DECOMMISSION_INITIAL_RESUME_DELAY).await {
+                        return;
                     }
+                    resume_local_decommission_after_init(store, rx, pool_indices).await;
                 });
             }
         }
@@ -284,3 +361,106 @@ impl ECStore {
         self.pools.len() == 1
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::{
+        LOCAL_DECOMMISSION_RESUME_MAX_CONFIG_RETRIES, resolve_store_init_stage_result, should_resume_local_decommission,
+        should_retry_local_decommission_resume, wait_for_local_decommission_resume_delay,
+    };
+    use crate::{
+        disk::endpoint::Endpoint,
+        endpoints::{EndpointServerPools, Endpoints, PoolEndpoints},
+        error::StorageError,
+    };
+    use std::time::Duration;
+    use tokio_util::sync::CancellationToken;
+
+    #[test]
+    fn test_should_resume_local_decommission_respects_local_flag() {
+        let mut local_endpoint = Endpoint::try_from("http://127.0.0.1:9000/data").expect("endpoint should parse");
+        local_endpoint.is_local = true;
+        let endpoints = EndpointServerPools::from(vec![PoolEndpoints {
+            legacy: false,
+            set_count: 1,
+            drives_per_set: 1,
+            endpoints: Endpoints::from(vec![local_endpoint]),
+            cmd_line: "pool-0".to_string(),
+            platform: String::new(),
+        }]);
+
+        assert!(should_resume_local_decommission(&endpoints, 0).expect("local endpoint should resume"));
+    }
+
+    #[test]
+    fn test_should_resume_local_decommission_rejects_unresolvable_pool() {
+        let endpoints = EndpointServerPools::default();
+        let err = should_resume_local_decommission(&endpoints, 0).expect_err("missing pool should error");
+        assert_eq!(
+            err.to_string(),
+            "Io error: store init failed to resolve decommission resume pool index 0 from current endpoints"
+        );
+    }
+
+    #[test]
+    fn test_should_resume_local_decommission_rejects_missing_endpoint() {
+        let endpoints = EndpointServerPools::from(vec![PoolEndpoints {
+            legacy: false,
+            set_count: 1,
+            drives_per_set: 1,
+            endpoints: Endpoints::from(Vec::<Endpoint>::new()),
+            cmd_line: "pool-0".to_string(),
+            platform: String::new(),
+        }]);
+        let err = should_resume_local_decommission(&endpoints, 0).expect_err("missing endpoint should error");
+        assert_eq!(
+            err.to_string(),
+            "Io error: store init failed to resolve decommission resume pool index 0: no endpoints available"
+        );
+    }
+
+    #[test]
+    fn test_should_retry_local_decommission_resume_accepts_config_not_found_before_retry_limit() {
+        assert!(should_retry_local_decommission_resume(&StorageError::ConfigNotFound, 0));
+    }
+
+    #[test]
+    fn test_should_retry_local_decommission_resume_rejects_config_not_found_at_retry_limit() {
+        assert!(!should_retry_local_decommission_resume(
+            &StorageError::ConfigNotFound,
+            LOCAL_DECOMMISSION_RESUME_MAX_CONFIG_RETRIES
+        ));
+    }
+
+    #[test]
+    fn test_should_retry_local_decommission_resume_rejects_non_config_errors() {
+        assert!(!should_retry_local_decommission_resume(&StorageError::SlowDown, 0));
+    }
+
+    #[test]
+    fn test_resolve_store_init_stage_result_passthrough_ok() {
+        resolve_store_init_stage_result(Ok(()), "load_rebalance_meta").expect("successful stage should pass through");
+    }
+
+    #[test]
+    fn test_resolve_store_init_stage_result_wraps_error_context() {
+        let err = resolve_store_init_stage_result(Err(StorageError::SlowDown), "start_rebalance")
+            .expect_err("failed stage should be wrapped");
+        let err_message = err.to_string();
+        assert!(err_message.contains("store init failed during start_rebalance"));
+        assert!(err_message.contains(&StorageError::SlowDown.to_string()));
+    }
+
+    #[tokio::test]
+    async fn test_wait_for_local_decommission_resume_delay_returns_true_after_delay() {
+        let rx = CancellationToken::new();
+        assert!(wait_for_local_decommission_resume_delay(&rx, Duration::from_millis(1)).await);
+    }
+
+    #[tokio::test]
+    async fn test_wait_for_local_decommission_resume_delay_returns_false_when_cancelled() {
+        let rx = CancellationToken::new();
+        rx.cancel();
+        assert!(!wait_for_local_decommission_resume_delay(&rx, Duration::from_secs(1)).await);
+    }
+}

crates/ecstore/src/store/object.rs

@@ -40,7 +40,97 @@ fn select_data_movement_target_pool(
     }
 }
 
+fn latest_object_access_delete_marker_error(
+    bucket: &str,
+    object: &str,
+    info: &ObjectInfo,
+    opts: &ObjectOptions,
+) -> Option<Error> {
+    if !info.delete_marker {
+        return None;
+    }
+
+    Some(if opts.version_id.is_none() || opts.delete_marker {
+        to_object_err(StorageError::FileNotFound, vec![bucket, object])
+    } else {
+        to_object_err(StorageError::MethodNotAllowed, vec![bucket, object])
+    })
+}
+
+fn resolve_latest_object_access(
+    bucket: &str,
+    object: &str,
+    info: ObjectInfo,
+    idx: usize,
+    opts: &ObjectOptions,
+) -> Result<(ObjectInfo, usize)> {
+    if let Some(err) = latest_object_access_delete_marker_error(bucket, object, &info, opts) {
+        return Err(err);
+    }
+
+    Ok((info, idx))
+}
+
+fn version_aware_lookup_opts(opts: &ObjectOptions, no_lock: bool) -> ObjectOptions {
+    let mut lookup_opts = opts.clone();
+    lookup_opts.no_lock = no_lock;
+    if lookup_opts.version_id.is_some() {
+        lookup_opts.metadata_chg = true;
+    }
+
+    lookup_opts
+}
+
+fn data_movement_pool_lookup_opts(opts: &ObjectOptions, no_lock: bool) -> ObjectOptions {
+    let mut lookup_opts = version_aware_lookup_opts(opts, no_lock);
+    lookup_opts.skip_decommissioned = true;
+    lookup_opts.skip_rebalancing = true;
+
+    lookup_opts
+}
+
 impl ECStore {
+    async fn get_latest_accessible_object_info_with_idx(
+        &self,
+        bucket: &str,
+        object: &str,
+        opts: &ObjectOptions,
+    ) -> Result<(ObjectInfo, usize)> {
+        let (info, idx) = self.get_latest_object_info_with_idx(bucket, object, opts).await?;
+        resolve_latest_object_access(bucket, object, info, idx, opts)
+    }
+
+    pub(super) async fn select_data_movement_pool_idx(
+        &self,
+        bucket: &str,
+        object: &str,
+        size: i64,
+        opts: &ObjectOptions,
+        no_lock: bool,
+    ) -> Result<usize> {
+        match self
+            .get_pool_info_existing_with_opts(bucket, object, &data_movement_pool_lookup_opts(opts, no_lock))
+            .await
+        {
+            Ok((pinfo, _)) => Ok(pinfo.index),
+            Err(err) => {
+                if !is_err_object_not_found(&err) && !is_err_version_not_found(&err) {
+                    return Err(err);
+                }
+
+                self.get_available_pool_idx(bucket, object, size).await.ok_or(Error::DiskFull)
+            }
+        }
+    }
+
+    fn resolve_decommission_target_pool_idx_result(result: Result<usize>, bucket: &str, object: &str) -> Result<usize> {
+        result.map_err(|err| Error::other(format!("failed to select decommission target pool for {bucket}/{object}: {err}")))
+    }
+
+    fn resolve_decommission_tiered_object_result(result: Result<()>, bucket: &str, object: &str) -> Result<()> {
+        result.map_err(|err| Error::other(format!("failed to decommission tiered object for {bucket}/{object}: {err}")))
+    }
+
     #[instrument(skip(self, fi, opts))]
     pub(crate) async fn decommission_tiered_object(
         &self,
@@ -54,10 +144,26 @@ impl ECStore {
         let object = encode_dir_object(object);
 
         if self.single_pool() {
-            return Err(Error::other(format!("error decommissioning {bucket}/{object}")));
+            return Self::resolve_decommission_tiered_object_result(
+                Err(Error::other("single pool deployments cannot decommission tiered objects")),
+                bucket,
+                &object,
+            );
         }
 
-        let idx = self.get_pool_idx_no_lock(bucket, &object, fi.size).await?;
+        let idx = if opts.data_movement && opts.version_id.is_some() {
+            Self::resolve_decommission_target_pool_idx_result(
+                self.select_data_movement_pool_idx(bucket, &object, fi.size, opts, true).await,
+                bucket,
+                &object,
+            )?
+        } else {
+            Self::resolve_decommission_target_pool_idx_result(
+                self.get_pool_idx_no_lock(bucket, &object, fi.size).await,
+                bucket,
+                &object,
+            )?
+        };
         if opts.data_movement && idx == opts.src_pool_idx {
             return Err(StorageError::DataMovementOverwriteErr(
                 bucket.to_owned(),
@@ -66,10 +172,14 @@ impl ECStore {
             ));
         }
 
-        self.pools[idx]
-            .get_disks_by_key(&object)
-            .decommission_tiered_object(bucket, &object, fi, opts)
-            .await
+        Self::resolve_decommission_tiered_object_result(
+            self.pools[idx]
+                .get_disks_by_key(&object)
+                .decommission_tiered_object(bucket, &object, fi, opts)
+                .await,
+            bucket,
+            &object,
+        )
     }
 
     #[instrument(level = "debug", skip(self))]
@@ -95,9 +205,9 @@ impl ECStore {
 
         opts.no_lock = true;
 
-        // TODO: check if DeleteMarker
-        let (_oi, idx) = self.get_latest_object_info_with_idx(bucket, &object, &opts).await?;
-
+        let (_, idx) = self
+            .get_latest_accessible_object_info_with_idx(bucket, &object, &opts)
+            .await?;
         self.pools[idx]
             .get_object_reader(bucket, object.as_str(), range, h, &opts)
             .await
@@ -119,7 +229,12 @@ impl ECStore {
             return self.pools[0].put_object(bucket, object.as_str(), data, opts).await;
         }
 
-        let idx = self.get_pool_idx(bucket, &object, data.size()).await?;
+        let idx = if opts.data_movement && opts.version_id.is_some() {
+            self.select_data_movement_pool_idx(bucket, &object, data.size(), opts, false)
+                .await?
+        } else {
+            self.get_pool_idx(bucket, &object, data.size()).await?
+        };
 
         if opts.data_movement && idx == opts.src_pool_idx {
             return Err(StorageError::DataMovementOverwriteErr(
@@ -144,8 +259,9 @@ impl ECStore {
 
         // TODO: nslock
 
-        let (info, _) = self.get_latest_object_info_with_idx(bucket, object.as_str(), opts).await?;
-
+        let (info, _) = self
+            .get_latest_accessible_object_info_with_idx(bucket, object.as_str(), opts)
+            .await?;
         opts.precondition_check(&info)?;
         Ok(info)
     }
@@ -172,7 +288,11 @@ impl ECStore {
 
         // TODO: nslock
 
-        let pool_idx = self.get_pool_idx_no_lock(src_bucket, &src_object, src_info.size).await?;
+        let pool_idx = self
+            .get_pool_info_existing_with_opts(src_bucket, &src_object, &version_aware_lookup_opts(src_opts, true))
+            .await?
+            .0
+            .index;
 
         if cp_src_dst_same {
             if let (Some(src_vid), Some(dst_vid)) = (&src_opts.version_id, &dst_opts.version_id)
@@ -233,8 +353,7 @@ impl ECStore {
         let object = encode_dir_object(object);
         let object = object.as_str();
 
-        let mut gopts = opts.clone();
-        gopts.no_lock = true;
+        let gopts = version_aware_lookup_opts(&opts, true);
 
         if opts.data_movement {
             let existing_pool_idx = self
@@ -505,8 +624,12 @@ impl ECStore {
             return Ok(());
         }
 
-        let idx = self
-            .get_pool_idx_existing_with_opts(bucket, object.as_str(), &ObjectOptions::default())
+        let opts = ObjectOptions {
+            version_id: Some(version_id.to_string()),
+            ..Default::default()
+        };
+        let (_, idx) = self
+            .get_latest_accessible_object_info_with_idx(bucket, object.as_str(), &opts)
             .await?;
 
         let _ = self.pools[idx].add_partial(bucket, object.as_str(), version_id).await;
@@ -522,7 +645,7 @@ impl ECStore {
 
         //opts.skip_decommissioned = true;
         //opts.no_lock = true;
-        let idx = self.get_pool_idx_existing_with_opts(bucket, &object, opts).await?;
+        let (_, idx) = self.get_latest_accessible_object_info_with_idx(bucket, &object, opts).await?;
 
         self.pools[idx].transition_object(bucket, &object, opts).await
     }
@@ -541,7 +664,9 @@ impl ECStore {
 
         //opts.skip_decommissioned = true;
         //opts.nolock = true;
-        let idx = self.get_pool_idx_existing_with_opts(bucket, object.as_str(), opts).await?;
+        let (_, idx) = self
+            .get_latest_accessible_object_info_with_idx(bucket, object.as_str(), opts)
+            .await?;
 
         self.pools[idx]
             .clone()
@@ -564,7 +689,9 @@ impl ECStore {
         let mut opts = opts.clone();
         opts.metadata_chg = true;
 
-        let idx = self.get_pool_idx_existing_with_opts(bucket, object.as_str(), &opts).await?;
+        let (_, idx) = self
+            .get_latest_accessible_object_info_with_idx(bucket, object.as_str(), &opts)
+            .await?;
 
         self.pools[idx].put_object_metadata(bucket, object.as_str(), &opts).await
     }
@@ -577,8 +704,7 @@ impl ECStore {
             return self.pools[0].get_object_tags(bucket, object.as_str(), opts).await;
         }
 
-        let (oi, _) = self.get_latest_object_info_with_idx(bucket, &object, opts).await?;
-
+        let (oi, _) = self.get_latest_accessible_object_info_with_idx(bucket, &object, opts).await?;
         Ok(oi.user_tags)
     }
 
@@ -596,7 +722,9 @@ impl ECStore {
             return self.pools[0].put_object_tags(bucket, object.as_str(), tags, opts).await;
         }
 
-        let idx = self.get_pool_idx_existing_with_opts(bucket, object.as_str(), opts).await?;
+        let (_, idx) = self
+            .get_latest_accessible_object_info_with_idx(bucket, object.as_str(), opts)
+            .await?;
 
         self.pools[idx].put_object_tags(bucket, object.as_str(), tags, opts).await
     }
@@ -629,7 +757,9 @@ impl ECStore {
             return self.pools[0].delete_object_tags(bucket, object.as_str(), opts).await;
         }
 
-        let idx = self.get_pool_idx_existing_with_opts(bucket, object.as_str(), opts).await?;
+        let (_, idx) = self
+            .get_latest_accessible_object_info_with_idx(bucket, object.as_str(), opts)
+            .await?;
 
         self.pools[idx].delete_object_tags(bucket, object.as_str(), opts).await
     }
@@ -665,4 +795,193 @@ mod tests {
         let target = select_data_movement_target_pool(Ok(0), 1, false).unwrap();
         assert_eq!(target, Some(0));
     }
+
+    #[test]
+    fn latest_object_access_delete_marker_error_returns_none_for_live_object() {
+        let info = ObjectInfo::default();
+        let opts = ObjectOptions::default();
+
+        assert!(latest_object_access_delete_marker_error("bucket", "object", &info, &opts).is_none());
+    }
+
+    #[test]
+    fn latest_object_access_delete_marker_error_returns_not_found_without_version_id() {
+        let info = ObjectInfo {
+            delete_marker: true,
+            ..Default::default()
+        };
+        let opts = ObjectOptions::default();
+
+        let err = latest_object_access_delete_marker_error("bucket", "object", &info, &opts)
+            .expect("delete marker should stop latest-object reads");
+
+        assert!(crate::error::is_err_object_not_found(&err));
+    }
+
+    #[test]
+    fn latest_object_access_delete_marker_error_returns_method_not_allowed_for_version_read() {
+        let info = ObjectInfo {
+            delete_marker: true,
+            ..Default::default()
+        };
+        let opts = ObjectOptions {
+            version_id: Some("vid-1".to_string()),
+            ..Default::default()
+        };
+
+        let err = latest_object_access_delete_marker_error("bucket", "object", &info, &opts)
+            .expect("delete marker version reads should be rejected");
+
+        assert!(matches!(err, Error::MethodNotAllowed));
+    }
+
+    #[test]
+    fn latest_object_access_delete_marker_error_returns_not_found_for_delete_marker_lookup() {
+        let info = ObjectInfo {
+            delete_marker: true,
+            ..Default::default()
+        };
+        let opts = ObjectOptions {
+            version_id: Some("vid-1".to_string()),
+            delete_marker: true,
+            ..Default::default()
+        };
+
+        let err = latest_object_access_delete_marker_error("bucket", "object", &info, &opts)
+            .expect("delete marker lookup should keep not-found semantics");
+
+        assert!(crate::error::is_err_object_not_found(&err));
+    }
+
+    #[test]
+    fn resolve_latest_object_access_returns_live_object_and_pool_idx() {
+        let info = ObjectInfo::default();
+        let opts = ObjectOptions::default();
+
+        let (resolved, idx) = resolve_latest_object_access("bucket", "object", info, 7, &opts).unwrap();
+
+        assert_eq!(idx, 7);
+        assert!(!resolved.delete_marker);
+    }
+
+    #[test]
+    fn resolve_latest_object_access_rejects_delete_marker_without_version_id() {
+        let info = ObjectInfo {
+            delete_marker: true,
+            ..Default::default()
+        };
+        let opts = ObjectOptions::default();
+
+        let err = resolve_latest_object_access("bucket", "object", info, 2, &opts).unwrap_err();
+
+        assert!(crate::error::is_err_object_not_found(&err));
+    }
+
+    #[test]
+    fn resolve_latest_object_access_rejects_delete_marker_version_read() {
+        let info = ObjectInfo {
+            delete_marker: true,
+            ..Default::default()
+        };
+        let opts = ObjectOptions {
+            version_id: Some("vid-1".to_string()),
+            ..Default::default()
+        };
+
+        let err = resolve_latest_object_access("bucket", "object", info, 2, &opts).unwrap_err();
+
+        assert!(matches!(err, Error::MethodNotAllowed));
+    }
+
+    #[test]
+    fn resolve_decommission_target_pool_idx_result_passthrough_ok() {
+        let idx = ECStore::resolve_decommission_target_pool_idx_result(Ok(3), "bucket", "object").unwrap();
+
+        assert_eq!(idx, 3);
+    }
+
+    #[test]
+    fn resolve_decommission_target_pool_idx_result_wraps_error_context() {
+        let err = ECStore::resolve_decommission_target_pool_idx_result(Err(Error::other("boom")), "bucket", "object")
+            .expect_err("expected contextual error");
+        let rendered = err.to_string();
+
+        assert!(rendered.contains("failed to select decommission target pool"), "{rendered}");
+        assert!(rendered.contains("bucket"), "{rendered}");
+        assert!(rendered.contains("object"), "{rendered}");
+        assert!(rendered.contains("boom"), "{rendered}");
+    }
+
+    #[test]
+    fn resolve_decommission_tiered_object_result_passthrough_ok() {
+        ECStore::resolve_decommission_tiered_object_result(Ok(()), "bucket", "object")
+            .expect("successful decommission result should pass through");
+    }
+
+    #[test]
+    fn resolve_decommission_tiered_object_result_wraps_error_context() {
+        let err = ECStore::resolve_decommission_tiered_object_result(Err(Error::other("boom")), "bucket", "object")
+            .expect_err("expected contextual error");
+        let rendered = err.to_string();
+
+        assert!(rendered.contains("failed to decommission tiered object"), "{rendered}");
+        assert!(rendered.contains("bucket"), "{rendered}");
+        assert!(rendered.contains("object"), "{rendered}");
+        assert!(rendered.contains("boom"), "{rendered}");
+    }
+
+    #[test]
+    fn version_aware_lookup_opts_enables_version_aware_lookup() {
+        let opts = ObjectOptions {
+            version_id: Some("vid-1".to_string()),
+            ..Default::default()
+        };
+
+        let lookup_opts = version_aware_lookup_opts(&opts, true);
+
+        assert!(lookup_opts.no_lock);
+        assert!(lookup_opts.metadata_chg);
+        assert_eq!(lookup_opts.version_id.as_deref(), Some("vid-1"));
+    }
+
+    #[test]
+    fn version_aware_lookup_opts_keeps_latest_lookup_for_unversioned_requests() {
+        let lookup_opts = version_aware_lookup_opts(&ObjectOptions::default(), true);
+
+        assert!(lookup_opts.no_lock);
+        assert!(!lookup_opts.metadata_chg);
+        assert!(lookup_opts.version_id.is_none());
+    }
+
+    #[test]
+    fn data_movement_pool_lookup_opts_enables_version_aware_lookup_and_skip_flags() {
+        let opts = ObjectOptions {
+            version_id: Some("vid-1".to_string()),
+            ..Default::default()
+        };
+
+        let lookup_opts = data_movement_pool_lookup_opts(&opts, false);
+
+        assert!(!lookup_opts.no_lock);
+        assert!(lookup_opts.metadata_chg);
+        assert!(lookup_opts.skip_decommissioned);
+        assert!(lookup_opts.skip_rebalancing);
+        assert_eq!(lookup_opts.version_id.as_deref(), Some("vid-1"));
+    }
+
+    #[test]
+    fn data_movement_pool_lookup_opts_keeps_no_lock_for_tiered_moves() {
+        let lookup_opts = data_movement_pool_lookup_opts(
+            &ObjectOptions {
+                version_id: Some("vid-1".to_string()),
+                ..Default::default()
+            },
+            true,
+        );
+
+        assert!(lookup_opts.no_lock);
+        assert!(lookup_opts.metadata_chg);
+        assert!(lookup_opts.skip_decommissioned);
+        assert!(lookup_opts.skip_rebalancing);
+    }
 }

crates/ecstore/src/store/rebalance.rs

@@ -13,7 +13,133 @@
 // limitations under the License.
 
 use super::*;
-use crate::bucket::utils::is_meta_bucketname;
+
+struct LatestObjectInfoCandidate {
+    info: Option<ObjectInfo>,
+    idx: usize,
+    err: Option<Error>,
+}
+
+fn pool_lookup_not_found_error(bucket: &str, object: &str, opts: &ObjectOptions) -> Error {
+    let object = decode_dir_object(object);
+
+    if let Some(version_id) = &opts.version_id {
+        StorageError::VersionNotFound(bucket.to_owned(), object.to_owned(), version_id.clone())
+    } else {
+        StorageError::ObjectNotFound(bucket.to_owned(), object.to_owned())
+    }
+}
+
+fn resolve_store_rebalance_pool_meta_reload_result(result: Result<()>, stage: &str) -> Result<()> {
+    result.map_err(|err| Error::other(format!("store rebalance pool meta reload failed during {stage}: {err}")))
+}
+
+fn resolve_rebalance_delete_from_all_pools_result(result: Result<ObjectInfo>, bucket: &str, object: &str) -> Result<ObjectInfo> {
+    result.map_err(|err| Error::other(format!("failed to delete rebalance source object {bucket}/{object}: {err}")))
+}
+
+fn rebalance_disk_set_lookup_error(pool_idx: usize, set_idx: usize, pool_count: usize) -> Error {
+    Error::other(format!(
+        "failed to resolve rebalance disk set: pool index {pool_idx}, set index {set_idx}, pool count {pool_count}",
+    ))
+}
+
+fn resolve_latest_object_info_candidates(
+    mut candidates: Vec<LatestObjectInfoCandidate>,
+    bucket: &str,
+    object: &str,
+    opts: &ObjectOptions,
+) -> Result<(ObjectInfo, usize)> {
+    candidates.sort_by(|a, b| {
+        let a_mod = if let Some(info) = &a.info {
+            info.mod_time.unwrap_or(OffsetDateTime::UNIX_EPOCH)
+        } else {
+            OffsetDateTime::UNIX_EPOCH
+        };
+
+        let b_mod = if let Some(info) = &b.info {
+            info.mod_time.unwrap_or(OffsetDateTime::UNIX_EPOCH)
+        } else {
+            OffsetDateTime::UNIX_EPOCH
+        };
+
+        if a_mod == b_mod {
+            return if a.idx < b.idx { Ordering::Greater } else { Ordering::Less };
+        }
+
+        b_mod.cmp(&a_mod)
+    });
+
+    for candidate in candidates {
+        if let Some(info) = candidate.info {
+            return Ok((info, candidate.idx));
+        }
+
+        if let Some(err) = candidate.err
+            && !is_err_object_not_found(&err)
+            && !is_err_version_not_found(&err)
+        {
+            return Err(err);
+        }
+    }
+
+    Err(pool_lookup_not_found_error(bucket, object, opts))
+}
+
+async fn build_server_pools_available_space(
+    bucket: &str,
+    size: i64,
+    n_sets: &[usize],
+    infos: &[Vec<Option<DiskInfo>>],
+) -> ServerPoolsAvailableSpace {
+    let mut server_pools = vec![PoolAvailableSpace::default(); infos.len()];
+
+    for (i, zinfo) in infos.iter().enumerate() {
+        if zinfo.is_empty() {
+            server_pools[i] = PoolAvailableSpace {
+                index: i,
+                ..Default::default()
+            };
+
+            continue;
+        }
+
+        if !is_meta_bucketname(bucket) && !has_space_for(zinfo, size).await.unwrap_or_default() {
+            server_pools[i] = PoolAvailableSpace {
+                index: i,
+                ..Default::default()
+            };
+
+            continue;
+        }
+
+        let mut available = 0;
+        let mut max_used_pct = 0;
+        for disk in zinfo.iter().flatten() {
+            if disk.total == 0 {
+                continue;
+            }
+
+            available += disk.total - disk.used;
+
+            let pct_used = disk.used * 100 / disk.total;
+
+            if pct_used > max_used_pct {
+                max_used_pct = pct_used;
+            }
+        }
+
+        available *= n_sets[i] as u64;
+
+        server_pools[i] = PoolAvailableSpace {
+            index: i,
+            available,
+            max_used_pct,
+        }
+    }
+
+    ServerPoolsAvailableSpace(server_pools)
+}
 
 impl ECStore {
     #[instrument(level = "debug", skip(self))]
@@ -72,7 +198,7 @@ impl ECStore {
         Ok(())
     }
 
-    async fn get_available_pool_idx(&self, bucket: &str, object: &str, size: i64) -> Option<usize> {
+    pub(super) async fn get_available_pool_idx(&self, bucket: &str, object: &str, size: i64) -> Option<usize> {
         // // Return a random one first
 
         let mut server_pools = self.get_server_pools_available_space(bucket, object, size).await;
@@ -102,67 +228,26 @@ impl ECStore {
     async fn get_server_pools_available_space(&self, bucket: &str, object: &str, size: i64) -> ServerPoolsAvailableSpace {
         let mut n_sets = vec![0; self.pools.len()];
         let mut infos = vec![Vec::new(); self.pools.len()];
-
-        // TODO: add concurrency
-        for (idx, pool) in self.pools.iter().enumerate() {
+        let pool_inputs = join_all(self.pools.iter().enumerate().map(|(idx, pool)| async move {
             if self.is_suspended(idx).await || self.is_pool_rebalancing(idx).await {
-                continue;
+                return (idx, 0, Vec::new());
             }
 
-            n_sets[idx] = pool.set_count;
+            let disk_infos = match pool.get_disks_by_key(object).get_disks(0, 0).await {
+                Ok(disks) => get_disk_infos(&disks).await,
+                Err(_) => Vec::new(),
+            };
 
-            if let Ok(disks) = pool.get_disks_by_key(object).get_disks(0, 0).await {
-                let disk_infos = get_disk_infos(&disks).await;
-                infos[idx] = disk_infos;
-            }
+            (idx, pool.set_count, disk_infos)
+        }))
+        .await;
+
+        for (idx, set_count, disk_infos) in pool_inputs {
+            n_sets[idx] = set_count;
+            infos[idx] = disk_infos;
         }
 
-        let mut server_pools = vec![PoolAvailableSpace::default(); self.pools.len()];
-        for (i, zinfo) in infos.iter().enumerate() {
-            if zinfo.is_empty() {
-                server_pools[i] = PoolAvailableSpace {
-                    index: i,
-                    ..Default::default()
-                };
-
-                continue;
-            }
-
-            if !is_meta_bucketname(bucket) && !has_space_for(zinfo, size).await.unwrap_or_default() {
-                server_pools[i] = PoolAvailableSpace {
-                    index: i,
-                    ..Default::default()
-                };
-
-                continue;
-            }
-
-            let mut available = 0;
-            let mut max_used_pct = 0;
-            for disk in zinfo.iter().flatten() {
-                if disk.total == 0 {
-                    continue;
-                }
-
-                available += disk.total - disk.used;
-
-                let pct_used = disk.used * 100 / disk.total;
-
-                if pct_used > max_used_pct {
-                    max_used_pct = pct_used;
-                }
-            }
-
-            available *= n_sets[i] as u64;
-
-            server_pools[i] = PoolAvailableSpace {
-                index: i,
-                available,
-                max_used_pct,
-            }
-        }
-
-        ServerPoolsAvailableSpace(server_pools)
+        build_server_pools_available_space(bucket, size, &n_sets, &infos).await
     }
 
     pub(super) async fn is_suspended(&self, idx: usize) -> bool {
@@ -349,7 +434,7 @@ impl ECStore {
             return Ok((def_pool, Vec::new()));
         }
 
-        Err(Error::ObjectNotFound(bucket.to_owned(), object.to_owned()))
+        Err(pool_lookup_not_found_error(bucket, object, opts))
     }
 
     async fn pools_with_object(&self, pools: &[PoolObjInfo], opts: &ObjectOptions) -> Vec<PoolErr> {
@@ -393,27 +478,20 @@ impl ECStore {
         }
 
         let results = join_all(futures).await;
-
-        struct IndexRes {
-            res: Option<ObjectInfo>,
-            idx: usize,
-            err: Option<Error>,
-        }
-
-        let mut idx_res = Vec::with_capacity(self.pools.len());
+        let mut candidates = Vec::with_capacity(self.pools.len());
 
         for (idx, result) in results.into_iter().enumerate() {
             match result {
                 Ok(res) => {
-                    idx_res.push(IndexRes {
-                        res: Some(res),
+                    candidates.push(LatestObjectInfoCandidate {
+                        info: Some(res),
                         idx,
                         err: None,
                     });
                 }
                 Err(e) => {
-                    idx_res.push(IndexRes {
-                        res: None,
+                    candidates.push(LatestObjectInfoCandidate {
+                        info: None,
                         idx,
                         err: Some(e),
                     });
@@ -421,53 +499,10 @@ impl ECStore {
             }
         }
 
-        // TODO: test order
-        idx_res.sort_by(|a, b| {
-            let a_mod = if let Some(o1) = &a.res {
-                o1.mod_time.unwrap_or(OffsetDateTime::UNIX_EPOCH)
-            } else {
-                OffsetDateTime::UNIX_EPOCH
-            };
-
-            let b_mod = if let Some(o2) = &b.res {
-                o2.mod_time.unwrap_or(OffsetDateTime::UNIX_EPOCH)
-            } else {
-                OffsetDateTime::UNIX_EPOCH
-            };
-
-            if a_mod == b_mod {
-                return if a.idx < b.idx { Ordering::Greater } else { Ordering::Less };
-            }
-
-            b_mod.cmp(&a_mod)
-        });
-
-        for res in idx_res.into_iter() {
-            if let Some(obj) = res.res {
-                return Ok((obj, res.idx));
-            }
-
-            if let Some(err) = res.err
-                && !is_err_object_not_found(&err)
-                && !is_err_version_not_found(&err)
-            {
-                return Err(err);
-            }
-
-            // TODO: delete marker
-        }
-
-        let object = decode_dir_object(object);
-
-        if opts.version_id.is_none() {
-            Err(StorageError::ObjectNotFound(bucket.to_owned(), object.to_owned()))
-        } else {
-            Err(StorageError::VersionNotFound(
-                bucket.to_owned(),
-                object.to_owned(),
-                opts.version_id.clone().unwrap_or_default(),
-            ))
-        }
+        // Delete markers are returned as latest object infos here. Higher-level
+        // access paths are responsible for translating them into read/write
+        // semantics such as object-not-found or method-not-allowed.
+        resolve_latest_object_info_candidates(candidates, bucket, object, opts)
     }
 
     pub(super) async fn delete_object_from_all_pools(
@@ -505,15 +540,18 @@ impl ECStore {
         }
 
         if let Some(e) = &derrs[0] {
-            return Err(e.clone());
+            return resolve_rebalance_delete_from_all_pools_result(Err(e.clone()), bucket, object);
         }
 
-        Ok(objs[0].as_ref().unwrap().clone())
+        resolve_rebalance_delete_from_all_pools_result(Ok(objs[0].as_ref().unwrap().clone()), bucket, object)
     }
 
     pub async fn reload_pool_meta(&self) -> Result<()> {
         let mut meta = PoolMeta::default();
-        meta.load(self.pools[0].clone(), self.pools.clone()).await?;
+        resolve_store_rebalance_pool_meta_reload_result(
+            meta.load(self.pools[0].clone(), self.pools.clone()).await,
+            "reload_pool_meta",
+        )?;
 
         let mut pool_meta = self.pool_meta.write().await;
         *pool_meta = meta;
@@ -670,7 +708,7 @@ impl ECStore {
         if pool_idx < self.pools.len() && set_idx < self.pools[pool_idx].disk_set.len() {
             self.pools[pool_idx].disk_set[set_idx].get_disks(0, 0).await
         } else {
-            Err(Error::other(format!("pool idx {pool_idx}, set idx {set_idx}, not found")))
+            Err(rebalance_disk_set_lookup_error(pool_idx, set_idx, self.pools.len()))
         }
     }
 
@@ -699,3 +737,216 @@ impl ECStore {
         Err(Error::DiskNotFound)
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::disk::DiskInfo;
+
+    fn object_info_with_mod_time(unix_ts: i64, delete_marker: bool) -> ObjectInfo {
+        ObjectInfo {
+            mod_time: Some(OffsetDateTime::from_unix_timestamp(unix_ts).unwrap()),
+            delete_marker,
+            ..Default::default()
+        }
+    }
+
+    fn disk_info(total: u64, used: u64, free: u64) -> DiskInfo {
+        DiskInfo {
+            total,
+            used,
+            free,
+            free_inodes: 1_024,
+            ..Default::default()
+        }
+    }
+
+    #[test]
+    fn resolve_latest_object_info_candidates_returns_latest_delete_marker() {
+        let candidates = vec![
+            LatestObjectInfoCandidate {
+                info: Some(object_info_with_mod_time(10, false)),
+                idx: 0,
+                err: None,
+            },
+            LatestObjectInfoCandidate {
+                info: Some(object_info_with_mod_time(20, true)),
+                idx: 1,
+                err: None,
+            },
+        ];
+
+        let (info, idx) =
+            resolve_latest_object_info_candidates(candidates, "bucket", "object", &ObjectOptions::default()).unwrap();
+
+        assert_eq!(idx, 1);
+        assert!(info.delete_marker);
+    }
+
+    #[test]
+    fn resolve_latest_object_info_candidates_prefers_higher_pool_idx_on_equal_mod_time() {
+        let candidates = vec![
+            LatestObjectInfoCandidate {
+                info: Some(object_info_with_mod_time(10, false)),
+                idx: 0,
+                err: None,
+            },
+            LatestObjectInfoCandidate {
+                info: Some(object_info_with_mod_time(10, false)),
+                idx: 1,
+                err: None,
+            },
+        ];
+
+        let (_, idx) = resolve_latest_object_info_candidates(candidates, "bucket", "object", &ObjectOptions::default()).unwrap();
+
+        assert_eq!(idx, 1);
+    }
+
+    #[test]
+    fn resolve_latest_object_info_candidates_returns_non_not_found_error() {
+        let err = resolve_latest_object_info_candidates(
+            vec![LatestObjectInfoCandidate {
+                info: None,
+                idx: 0,
+                err: Some(Error::ErasureReadQuorum),
+            }],
+            "bucket",
+            "object",
+            &ObjectOptions::default(),
+        )
+        .unwrap_err();
+
+        assert_eq!(err, Error::ErasureReadQuorum);
+    }
+
+    #[test]
+    fn resolve_latest_object_info_candidates_returns_version_not_found_for_versioned_lookups() {
+        let err = resolve_latest_object_info_candidates(
+            vec![LatestObjectInfoCandidate {
+                info: None,
+                idx: 0,
+                err: Some(Error::ObjectNotFound("bucket".to_string(), "object".to_string())),
+            }],
+            "bucket",
+            "object",
+            &ObjectOptions {
+                version_id: Some("vid-1".to_string()),
+                ..Default::default()
+            },
+        )
+        .unwrap_err();
+
+        assert_eq!(
+            err,
+            Error::VersionNotFound("bucket".to_string(), "object".to_string(), "vid-1".to_string())
+        );
+    }
+
+    #[test]
+    fn pool_lookup_not_found_error_returns_object_not_found_for_latest_lookup() {
+        let err = pool_lookup_not_found_error("bucket", "object", &ObjectOptions::default());
+
+        assert_eq!(err, Error::ObjectNotFound("bucket".to_string(), "object".to_string()));
+    }
+
+    #[test]
+    fn pool_lookup_not_found_error_returns_version_not_found_for_versioned_lookup() {
+        let err = pool_lookup_not_found_error(
+            "bucket",
+            "object",
+            &ObjectOptions {
+                version_id: Some("vid-1".to_string()),
+                ..Default::default()
+            },
+        );
+
+        assert_eq!(
+            err,
+            Error::VersionNotFound("bucket".to_string(), "object".to_string(), "vid-1".to_string())
+        );
+    }
+
+    #[test]
+    fn resolve_store_rebalance_pool_meta_reload_result_passthrough_ok() {
+        resolve_store_rebalance_pool_meta_reload_result(Ok(()), "reload_pool_meta")
+            .expect("successful pool meta reload should pass through");
+    }
+
+    #[test]
+    fn resolve_store_rebalance_pool_meta_reload_result_wraps_error_context() {
+        let err = resolve_store_rebalance_pool_meta_reload_result(Err(Error::SlowDown), "reload_pool_meta")
+            .expect_err("failed pool meta reload should be wrapped");
+        let err_message = err.to_string();
+        assert!(err_message.contains("store rebalance pool meta reload failed during reload_pool_meta"));
+        assert!(err_message.contains(&Error::SlowDown.to_string()));
+    }
+
+    #[test]
+    fn resolve_rebalance_delete_from_all_pools_result_passthrough_ok() {
+        let info = ObjectInfo {
+            bucket: "bucket".to_string(),
+            name: "object".to_string(),
+            ..Default::default()
+        };
+
+        let resolved = resolve_rebalance_delete_from_all_pools_result(Ok(info.clone()), "bucket", "object")
+            .expect("successful rebalance delete should pass through");
+
+        assert_eq!(resolved.bucket, info.bucket);
+        assert_eq!(resolved.name, info.name);
+    }
+
+    #[test]
+    fn resolve_rebalance_delete_from_all_pools_result_wraps_object_context() {
+        let err = resolve_rebalance_delete_from_all_pools_result(Err(Error::SlowDown), "bucket", "object")
+            .expect_err("failed rebalance delete should be wrapped");
+        let rendered = err.to_string();
+
+        assert!(rendered.contains("failed to delete rebalance source object bucket/object"), "{rendered}");
+        assert!(rendered.contains(&Error::SlowDown.to_string()), "{rendered}");
+    }
+
+    #[test]
+    fn rebalance_disk_set_lookup_error_formats_pool_and_set_context() {
+        let err = rebalance_disk_set_lookup_error(2, 7, 3);
+
+        assert!(
+            err.to_string()
+                .contains("failed to resolve rebalance disk set: pool index 2, set index 7, pool count 3")
+        );
+    }
+
+    #[tokio::test]
+    async fn build_server_pools_available_space_returns_zero_for_empty_pool_info() {
+        let spaces = build_server_pools_available_space("bucket-a", 64, &[1], &[Vec::new()]).await;
+
+        assert_eq!(spaces.0.len(), 1);
+        assert_eq!(spaces.0[0].index, 0);
+        assert_eq!(spaces.0[0].available, 0);
+        assert_eq!(spaces.0[0].max_used_pct, 0);
+    }
+
+    #[tokio::test]
+    async fn build_server_pools_available_space_computes_available_capacity_and_max_used_pct() {
+        let infos = vec![vec![Some(disk_info(1_000, 100, 900)), Some(disk_info(1_000, 200, 800))]];
+
+        let spaces = build_server_pools_available_space("bucket-a", 64, &[2], &infos).await;
+
+        assert_eq!(spaces.0.len(), 1);
+        assert_eq!(spaces.0[0].index, 0);
+        assert_eq!(spaces.0[0].available, 3_400);
+        assert_eq!(spaces.0[0].max_used_pct, 20);
+    }
+
+    #[tokio::test]
+    async fn build_server_pools_available_space_skips_capacity_guard_for_meta_bucket() {
+        let infos = vec![vec![Some(disk_info(10, 9, 1)), Some(disk_info(10, 9, 1))]];
+
+        let spaces = build_server_pools_available_space(crate::disk::RUSTFS_META_BUCKET, 1_024, &[1], &infos).await;
+
+        assert_eq!(spaces.0.len(), 1);
+        assert_eq!(spaces.0[0].available, 2);
+        assert_eq!(spaces.0[0].max_used_pct, 90);
+    }
+}

crates/ecstore/src/tier/tier.rs

@@ -49,7 +49,7 @@ use crate::{
     StorageAPI,
     config::com::{CONFIG_PREFIX, read_config},
     disk::{MIGRATING_META_BUCKET, RUSTFS_META_BUCKET},
-    global::{get_global_endpoints, is_first_cluster_node_local},
+    global::is_first_cluster_node_local,
     store::ECStore,
     store_api::{ObjectIO as _, ObjectOptions, PutObjReader},
 };
@@ -1006,7 +1006,7 @@ impl TierConfigMgr {
     #[tracing::instrument(level = "debug", name = "tier_save", skip(self))]
     pub async fn save(&self) -> std::result::Result<(), std::io::Error> {
         let Some(api) = new_object_layer_fn() else {
-            return Err(std::io::Error::other("errServerNotInitialized"));
+            return Err(tier_config_not_initialized_error("save tiering config"));
         };
         //let (pr, opts) = GLOBAL_TierConfigMgr.write().config_reader()?;
 
@@ -1231,6 +1231,10 @@ pub fn is_err_config_not_found(err: &StorageError) -> bool {
     matches!(err, StorageError::ObjectNotFound(_, _) | StorageError::BucketNotFound(_)) || err == &StorageError::ConfigNotFound
 }
 
+fn tier_config_not_initialized_error(operation: &str) -> std::io::Error {
+    std::io::Error::other(format!("failed to {operation}: object layer not initialized"))
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -1335,4 +1339,13 @@ mod tests {
             Some("bucket-a")
         );
     }
+
+    #[test]
+    fn test_tier_config_not_initialized_error_formats_operation_context() {
+        let err = tier_config_not_initialized_error("save tiering config");
+        let rendered = err.to_string();
+
+        assert!(rendered.contains("failed to save tiering config"), "{rendered}");
+        assert!(rendered.contains("object layer not initialized"), "{rendered}");
+    }
 }

crates/filemeta/src/metacache.rs

@@ -399,7 +399,13 @@ impl MetaCacheEntries {
             return None;
         }
 
-        let metadata = match cached.marshal_msg() {
+        let merged_cached = FileMeta {
+            meta_ver: cached.meta_ver,
+            versions,
+            ..Default::default()
+        };
+
+        let metadata = match merged_cached.marshal_msg() {
             Ok(meta) => meta,
             Err(e) => {
                 warn!("decommission_pool: entries resolve entry marshal_msg {:?}", e);
@@ -411,11 +417,7 @@ impl MetaCacheEntries {
         // Create a new merged result.
         let new_selected = MetaCacheEntry {
             name: selected.name.clone(),
-            cached: Some(FileMeta {
-                meta_ver: cached.meta_ver,
-                versions,
-                ..Default::default()
-            }),
+            cached: Some(merged_cached),
             reusable: true,
             metadata,
         };
@@ -871,7 +873,12 @@ impl<T: Clone + Debug + Send + 'static> Cache<T> {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::test_data::create_real_xlmeta;
+    use crate::{FileMetaVersion, MetaDeleteMarker};
+    use std::collections::HashMap;
     use std::io::Cursor;
+    use time::OffsetDateTime;
+    use uuid::Uuid;
 
     #[tokio::test]
     async fn test_writer() {
@@ -900,4 +907,61 @@ mod tests {
 
         assert_eq!(objs, nobjs);
     }
+
+    #[test]
+    fn test_resolve_rebuilds_metadata_from_merged_versions() {
+        let base_metadata = create_real_xlmeta().expect("base xl.meta");
+        let base = FileMeta::load(&base_metadata).expect("load base xl.meta");
+
+        let extra_version = FileMetaVersion {
+            version_type: VersionType::Delete,
+            object: None,
+            delete_marker: Some(MetaDeleteMarker {
+                version_id: Some(Uuid::from_u128(0x22222222333344445555666666666666)),
+                mod_time: Some(OffsetDateTime::from_unix_timestamp(1_705_312_400).expect("valid timestamp")),
+                meta_sys: HashMap::new(),
+            }),
+            write_version: 99,
+            uses_legacy_checksum: false,
+        };
+
+        let extra_shallow = FileMetaShallowVersion::try_from(extra_version).expect("build shallow delete version");
+
+        let mut extended = base.clone();
+        extended.versions.insert(0, extra_shallow);
+
+        let base_versions = base.versions.len();
+        let extended_versions = extended.versions.len();
+        let extended_metadata = extended.marshal_msg().expect("serialize extended xl.meta");
+
+        let resolved = MetaCacheEntries(vec![
+            Some(MetaCacheEntry {
+                name: "bucket/object".to_string(),
+                metadata: extended_metadata,
+                cached: Some(extended),
+                reusable: false,
+            }),
+            Some(MetaCacheEntry {
+                name: "bucket/object".to_string(),
+                metadata: base_metadata,
+                cached: Some(base),
+                reusable: false,
+            }),
+        ])
+        .resolve(MetadataResolutionParams {
+            obj_quorum: 2,
+            requested_versions: extended_versions,
+            strict: true,
+            ..Default::default()
+        })
+        .expect("merged entry should resolve");
+
+        let cached = resolved.cached.expect("resolved entry should keep merged cached metadata");
+        let decoded = FileMeta::load(&resolved.metadata).expect("resolved metadata should decode");
+
+        assert_eq!(cached.versions.len(), base_versions);
+        assert_eq!(decoded.versions.len(), base_versions);
+        assert_eq!(decoded.versions, cached.versions);
+        assert_ne!(extended_versions, cached.versions.len());
+    }
 }

crates/rio/src/compress_index.rs

@@ -18,6 +18,7 @@ use std::io::{self, Read, Seek, SeekFrom};
 
 const S2_INDEX_HEADER: &[u8] = b"s2idx\x00";
 const S2_INDEX_TRAILER: &[u8] = b"\x00xdi2s";
+const LEGACY_INDEX_HEADER_PADDING: &[u8] = &[0, 0, 0];
 const MAX_INDEX_ENTRIES: usize = 1 << 16;
 const MIN_INDEX_DIST: i64 = 1 << 20;
 // const MIN_INDEX_DIST: i64 = 0;
@@ -76,10 +77,14 @@ impl Index {
     }
 
     fn alloc_infos(&mut self, n: usize) {
-        if n > MAX_INDEX_ENTRIES {
-            panic!("n > MAX_INDEX_ENTRIES");
-        }
-        self.info = Vec::with_capacity(n);
+        debug_assert!(n <= MAX_INDEX_ENTRIES, "n > MAX_INDEX_ENTRIES");
+        self.info = vec![
+            IndexInfo {
+                compressed_offset: 0,
+                uncompressed_offset: 0,
+            };
+            n
+        ];
     }
 
     pub fn add(&mut self, compressed_offset: i64, uncompressed_offset: i64) -> io::Result<()> {
@@ -217,9 +222,8 @@ impl Index {
         self.reduce();
         let init_size = b.len();
 
-        // Add skippable header
-        b.extend_from_slice(&[0x50, 0x2A, 0x4D, 0x18]); // ChunkTypeIndex
-        b.extend_from_slice(&[0, 0, 0]); // Placeholder for chunk length
+        // Add skippable header (1-byte marker + 24-bit length placeholder)
+        b.extend_from_slice(&[0x50, 0x2A, 0x4D, 0x18]); // length is written back into bytes 1..=3
 
         // Add header
         b.extend_from_slice(S2_INDEX_HEADER);
@@ -295,7 +299,7 @@ impl Index {
             return Err(io::Error::new(io::ErrorKind::UnexpectedEof, "buffer too small"));
         }
 
-        if b[0] != 0x50 || b[1] != 0x2A || b[2] != 0x4D || b[3] != 0x18 {
+        if b[0] != 0x50 {
             return Err(io::Error::other("invalid chunk type"));
         }
 
@@ -306,6 +310,10 @@ impl Index {
             return Err(io::Error::new(io::ErrorKind::UnexpectedEof, "buffer too small"));
         }
 
+        if b.starts_with(LEGACY_INDEX_HEADER_PADDING) {
+            b = &b[LEGACY_INDEX_HEADER_PADDING.len()..];
+        }
+
         if !b.starts_with(S2_INDEX_HEADER) {
             return Err(io::Error::other("invalid header"));
         }
@@ -687,4 +695,72 @@ mod tests {
 
         Ok(())
     }
+
+    #[test]
+    fn test_index_into_vec_round_trip_via_load() -> io::Result<()> {
+        let mut source = Index::new();
+        source.add(100, 1_000)?;
+        source.add(300, 1_000 + MIN_INDEX_DIST)?;
+
+        let encoded = source.clone().into_vec();
+
+        let mut decoded = Index::new();
+        let rest = decoded.load(encoded.as_ref())?;
+
+        assert!(rest.is_empty());
+        assert_eq!(decoded.total_uncompressed, source.total_uncompressed);
+        assert_eq!(decoded.total_compressed, source.total_compressed);
+        assert_eq!(decoded.info.len(), source.info.len());
+        assert_eq!(decoded.info[0].compressed_offset, source.info[0].compressed_offset);
+        assert_eq!(decoded.info[0].uncompressed_offset, source.info[0].uncompressed_offset);
+        assert_eq!(decoded.info[1].uncompressed_offset, source.info[1].uncompressed_offset);
+        assert!(decoded.info[1].compressed_offset > decoded.info[0].compressed_offset);
+
+        Ok(())
+    }
+
+    #[test]
+    fn test_index_load_rejects_invalid_chunk_type_marker() -> io::Result<()> {
+        let mut source = Index::new();
+        source.add(100, 1_000)?;
+        source.add(300, 1_000 + MIN_INDEX_DIST)?;
+        let mut encoded = source.into_vec().to_vec();
+
+        encoded[0] = 0x51;
+
+        let mut decoded = Index::new();
+        let err = decoded
+            .load(encoded.as_slice())
+            .expect_err("invalid marker should be rejected");
+        assert_eq!(err.kind(), io::ErrorKind::Other);
+        assert_eq!(err.to_string(), "invalid chunk type");
+
+        Ok(())
+    }
+
+    #[test]
+    fn test_index_load_accepts_legacy_zero_padded_header() -> io::Result<()> {
+        let mut source = Index::new();
+        source.add(100, 1_000)?;
+        source.add(300, 1_000 + MIN_INDEX_DIST)?;
+
+        let mut encoded = source.clone().into_vec().to_vec();
+        let chunk_len = (encoded[1] as usize) | ((encoded[2] as usize) << 8) | ((encoded[3] as usize) << 16);
+        let legacy_chunk_len = chunk_len + LEGACY_INDEX_HEADER_PADDING.len();
+
+        encoded[1] = legacy_chunk_len as u8;
+        encoded[2] = (legacy_chunk_len >> 8) as u8;
+        encoded[3] = (legacy_chunk_len >> 16) as u8;
+        encoded.splice(4..4, LEGACY_INDEX_HEADER_PADDING.iter().copied());
+
+        let mut decoded = Index::new();
+        let rest = decoded.load(encoded.as_slice())?;
+
+        assert!(rest.is_empty());
+        assert_eq!(decoded.total_uncompressed, source.total_uncompressed);
+        assert_eq!(decoded.total_compressed, source.total_compressed);
+        assert_eq!(decoded.info.len(), source.info.len());
+
+        Ok(())
+    }
 }

crates/rio/src/http_reader.rs

@@ -22,6 +22,7 @@ use rustfs_common::internode_metrics::global_internode_metrics;
 use rustfs_utils::get_env_opt_str;
 use std::io::IoSlice;
 use std::io::{self, Error};
+use std::net::IpAddr;
 use std::ops::Not as _;
 use std::pin::Pin;
 use std::sync::LazyLock;
@@ -91,25 +92,48 @@ fn load_optional_mtls_identity_from_tls_path() -> Option<Identity> {
     }
 }
 
-fn get_http_client() -> Client {
-    // Reuse the HTTP connection pool in the global `reqwest::Client` instance
-    // TODO: interact with load balancing?
-    static CLIENT: LazyLock<Client> = LazyLock::new(|| {
-        let mut builder = Client::builder()
-            .connect_timeout(std::time::Duration::from_secs(5))
-            .tcp_keepalive(std::time::Duration::from_secs(10))
-            .http2_keep_alive_interval(std::time::Duration::from_secs(5))
-            .http2_keep_alive_timeout(std::time::Duration::from_secs(3))
-            .http2_keep_alive_while_idle(true);
+fn build_http_client(disable_proxy: bool) -> Client {
+    let mut builder = Client::builder()
+        .connect_timeout(std::time::Duration::from_secs(5))
+        .tcp_keepalive(std::time::Duration::from_secs(10))
+        .http2_keep_alive_interval(std::time::Duration::from_secs(5))
+        .http2_keep_alive_timeout(std::time::Duration::from_secs(3))
+        .http2_keep_alive_while_idle(true);
 
-        // HTTPS root trust + optional mTLS identity from RUSTFS_TLS_PATH
-        builder = load_ca_roots_from_tls_path(builder);
-        if let Some(id) = load_optional_mtls_identity_from_tls_path() {
-            builder = builder.identity(id);
-        }
+    if disable_proxy {
+        builder = builder.no_proxy();
+    }
+
+    builder = load_ca_roots_from_tls_path(builder);
+    if let Some(id) = load_optional_mtls_identity_from_tls_path() {
+        builder = builder.identity(id);
+    }
+
+    builder.build().expect("Failed to create global HTTP client")
+}
+
+fn should_bypass_proxy_for_url(url: &str) -> bool {
+    let Some(host) = reqwest::Url::parse(url)
+        .ok()
+        .and_then(|url| url.host_str().map(str::to_owned))
+    else {
+        return false;
+    };
+    let host = host.trim_matches(['[', ']']);
+
+    host.eq_ignore_ascii_case("localhost") || host.parse::<IpAddr>().is_ok_and(|addr| addr.is_loopback())
+}
+
+fn get_http_client(url: &str) -> Client {
+    // Reuse HTTP connection pools while keeping loopback traffic away from
+    // system proxies so local RPC/tests do not leak to proxy listeners.
+    static CLIENT: LazyLock<Client> = LazyLock::new(|| build_http_client(false));
+    static LOCAL_CLIENT: LazyLock<Client> = LazyLock::new(|| build_http_client(true));
+
+    if should_bypass_proxy_for_url(url) {
+        return LOCAL_CLIENT.clone();
+    }
 
-        builder.build().expect("Failed to create global HTTP client")
-    });
     CLIENT.clone()
 }
 
@@ -138,7 +162,7 @@ impl HttpReader {
         _read_buf_size: usize,
     ) -> io::Result<Self> {
         let track_internode_metrics = is_internode_rpc_url(&url);
-        let client = get_http_client();
+        let client = get_http_client(&url);
         let mut request: RequestBuilder = client.request(method.clone(), url.clone()).headers(headers.clone());
         if let Some(body) = body {
             request = request.body(body);
@@ -302,7 +326,7 @@ impl HttpWriter {
             //     "[HttpWriter::spawn] sending HTTP request: url={url_clone}, method={method_clone:?}, headers={headers_clone:?}"
             // );
 
-            let client = get_http_client();
+            let client = get_http_client(&url_clone);
             let request = client
                 .request(method_clone, url_clone.clone())
                 .headers(headers_clone.clone())
@@ -664,4 +688,14 @@ mod tests {
 
         handle.abort();
     }
+
+    #[test]
+    fn loopback_urls_bypass_proxy_selection() {
+        assert!(should_bypass_proxy_for_url("http://127.0.0.1:9000/stream"));
+        assert!(should_bypass_proxy_for_url("http://localhost:9000/stream"));
+        assert!(should_bypass_proxy_for_url("http://[::1]:9000/stream"));
+        assert!(!should_bypass_proxy_for_url("http://192.168.1.10:9000/stream"));
+        assert!(!should_bypass_proxy_for_url("http://example.com/stream"));
+        assert!(!should_bypass_proxy_for_url("not-a-url"));
+    }
 }