fix(ecstore): prefer active pool reads during decommission (#6475)

crates/ecstore/src/store/init.rs

@@ -4316,6 +4316,8 @@ mod tests {
         let bucket = format!("reverse-decom-fixed-target-{}", uuid::Uuid::new_v4());
         let object = "ordinary.bin";
         let object_body = b"reverse ordinary generation".to_vec();
+        let self_copy_object = "source-only-self-copy.bin";
+        let self_copy_body = b"source only copy generation".to_vec();
         let multipart_object = "multipart.bin";
         let first_part = vec![b'm'; 5 * 1024 * 1024];
         let second_part = b"reverse multipart tail".to_vec();
@@ -4331,6 +4333,11 @@ mod tests {
             .put_object(&bucket, object, &mut source, &ObjectOptions::default())
             .await
             .expect("write ordinary source object to pool 1");
+        let mut self_copy_source = PutObjReader::from_vec(self_copy_body.clone());
+        store.pools[1]
+            .put_object(&bucket, self_copy_object, &mut self_copy_source, &ObjectOptions::default())
+            .await
+            .expect("write self-copy source object to pool 1");
 
         let upload = store.pools[1]
             .new_multipart_upload(&bucket, multipart_object, &ObjectOptions::default())
@@ -4371,6 +4378,69 @@ mod tests {
         }
         assert!(store.is_suspended(1).await, "pool 1 must be the reverse decommission source");
 
+        let self_copy_opts = ObjectOptions {
+            no_lock: true,
+            ..Default::default()
+        };
+        let mut self_copy_reader = store
+            .get_object_reader(&bucket, self_copy_object, None, HeaderMap::new(), &self_copy_opts)
+            .await
+            .expect("read the suspended source before its target is committed");
+        let mut self_copy_info = self_copy_reader.object_info.clone();
+        let mut self_copy_source_body = Vec::new();
+        self_copy_reader
+            .stream
+            .read_to_end(&mut self_copy_source_body)
+            .await
+            .expect("drain the suspended self-copy source");
+        assert_eq!(self_copy_source_body, self_copy_body);
+        self_copy_info.metadata_only = true;
+        self_copy_info.put_object_reader = Some(PutObjReader::from_vec(self_copy_source_body));
+        store
+            .copy_object(
+                &bucket,
+                self_copy_object,
+                &bucket,
+                self_copy_object,
+                &mut self_copy_info,
+                &self_copy_opts,
+                &self_copy_opts,
+            )
+            .await
+            .expect("self-copy should read the suspended source and commit to an active pool");
+        assert_pool_object_present(&store.pools[0], &bucket, self_copy_object).await;
+        assert_pool_object_present(&store.pools[1], &bucket, self_copy_object).await;
+
+        let mut active_copy_reader = store
+            .get_object_reader(&bucket, self_copy_object, None, HeaderMap::new(), &self_copy_opts)
+            .await
+            .expect("read the active self-copy target while the source remains");
+        let active_copy_data_dir = active_copy_reader.object_info.data_dir;
+        let mut active_copy_info = active_copy_reader.object_info.clone();
+        let mut active_copy_body = Vec::new();
+        active_copy_reader
+            .stream
+            .read_to_end(&mut active_copy_body)
+            .await
+            .expect("drain the active self-copy target");
+        assert_eq!(active_copy_body, self_copy_body);
+        active_copy_info.metadata_only = true;
+        active_copy_info.put_object_reader = Some(PutObjReader::from_vec(active_copy_body));
+        let active_copy_result = store
+            .copy_object(
+                &bucket,
+                self_copy_object,
+                &bucket,
+                self_copy_object,
+                &mut active_copy_info,
+                &self_copy_opts,
+                &self_copy_opts,
+            )
+            .await
+            .expect("self-copy should keep using the committed active target");
+        assert_eq!(active_copy_result.data_dir, active_copy_data_dir);
+
+        let cleanup_barrier = crate::data_movement::SourceCleanupDeleteBarrier::install(&bucket, object);
         let commit_barrier = crate::set_disk::PutObjectCommitBarrier::install(
             &bucket,
             object,
@@ -4419,7 +4489,50 @@ mod tests {
         drop(delete_barrier);
 
         commit_barrier.release();
+        cleanup_barrier.wait_until_paused().await;
         drop(commit_barrier);
+
+        let target_info = store.pools[0]
+            .get_object_info(&bucket, object, &ObjectOptions::default())
+            .await
+            .expect("read the committed active target before source cleanup");
+        assert_pool_object_present(&store.pools[1], &bucket, object).await;
+
+        let read_opts = ObjectOptions {
+            no_lock: true,
+            ..Default::default()
+        };
+        let routed_info = store
+            .get_object_info(&bucket, object, &read_opts)
+            .await
+            .expect("HEAD routing should prefer the active target before source cleanup");
+        assert_eq!(routed_info.data_dir, target_info.data_dir);
+
+        let mut ranged_reader = store
+            .get_object_reader(
+                &bucket,
+                object,
+                Some(HTTPRangeSpec {
+                    is_suffix_length: false,
+                    start: 8,
+                    end: 15,
+                }),
+                HeaderMap::new(),
+                &read_opts,
+            )
+            .await
+            .expect("ranged GET routing should prefer the active target before source cleanup");
+        assert_eq!(ranged_reader.object_info.data_dir, target_info.data_dir);
+        let mut ranged_body = Vec::new();
+        ranged_reader
+            .stream
+            .read_to_end(&mut ranged_body)
+            .await
+            .expect("drain the routed target range");
+        assert_eq!(ranged_body, object_body[8..=15]);
+
+        cleanup_barrier.release();
+        drop(cleanup_barrier);
         tokio::time::timeout(Duration::from_secs(60), worker)
             .await
             .expect("reverse ordinary decommission must not self-deadlock on the fixed target set")

crates/ecstore/src/store/object.rs

@@ -2708,13 +2708,13 @@ impl ECStore {
         };
 
         if cp_src_dst_same {
-            let pool_idx = self
-                .get_pool_info_existing_with_opts(src_bucket, &src_object, &writer_pool_lookup_opts(src_opts, true))
-                .await?
-                .0
-                .index;
+            let (_, pool_idx) = self
+                .get_latest_accessible_object_info_with_idx(src_bucket, &src_object, &version_aware_lookup_opts(src_opts, true))
+                .await?;
+            let source_pool_writable = !self.is_suspended(pool_idx).await && !self.is_pool_rebalancing(pool_idx).await;
 
-            if let (Some(src_vid), Some(dst_vid)) = (&src_opts.version_id, &dst_opts.version_id)
+            if source_pool_writable
+                && let (Some(src_vid), Some(dst_vid)) = (&src_opts.version_id, &dst_opts.version_id)
                 && src_vid == dst_vid
             {
                 return self.pools[pool_idx]
@@ -2722,7 +2722,7 @@ impl ECStore {
                     .await;
             }
 
-            if !dst_opts.versioned && src_opts.version_id.is_none() {
+            if source_pool_writable && !dst_opts.versioned && src_opts.version_id.is_none() {
                 if src_info.metadata_only {
                     // Zero-copy update: only xl.meta is rewritten, the data blocks stay as they
                     // are. The caller must therefore guarantee that the destination metadata
@@ -2765,7 +2765,7 @@ impl ECStore {
                 };
             }
 
-            if dst_opts.versioned && src_opts.version_id != dst_opts.version_id {
+            if source_pool_writable && dst_opts.versioned && src_opts.version_id != dst_opts.version_id {
                 // Restoring a specific historical version onto the current key creates a NEW
                 // version. When the caller supplies a reader (S3 CopyObject), write the fetched
                 // bytes through put_object so any re-encryption/compression applied to the reader

crates/ecstore/src/store/rebalance.rs

@@ -23,10 +23,13 @@ pub(in crate::store) mod support;
 const LOG_COMPONENT_ECSTORE: &str = "ecstore";
 const LOG_SUBSYSTEM_POOLS: &str = "pools";
 const EVENT_POOL_META_RELOAD: &str = "pool_meta_reload";
+#[cfg(test)]
+use support::resolve_latest_object_info_candidates;
 use support::{
     LatestObjectInfoCandidate, PoolErr, PoolObjInfo, RebalanceDeletePoolResult, pool_lookup_not_found_error,
-    rebalance_disk_set_lookup_error, resolve_latest_object_info_candidates, resolve_rebalance_delete_from_all_pools_result,
-    resolve_rebalance_delete_from_all_pools_results, resolve_store_rebalance_pool_meta_reload_result,
+    rebalance_disk_set_lookup_error, resolve_latest_object_info_candidates_with_pool_state,
+    resolve_rebalance_delete_from_all_pools_result, resolve_rebalance_delete_from_all_pools_results,
+    resolve_store_rebalance_pool_meta_reload_result,
 };
 
 #[derive(Debug, Default, Eq, PartialEq)]
@@ -611,9 +614,19 @@ impl ECStore {
         object: &str,
         opts: &ObjectOptions,
     ) -> Result<(ObjectInfo, usize)> {
+        let suspended_pools = if opts.skip_decommissioned {
+            let pool_meta = self.pool_meta.read().await;
+            Some(
+                (0..self.pools.len())
+                    .map(|idx| pool_meta.is_suspended(idx))
+                    .collect::<Vec<_>>(),
+            )
+        } else {
+            None
+        };
         let mut futures = Vec::with_capacity(self.pools.len());
         for (idx, pool) in self.pools.iter().enumerate() {
-            if opts.skip_decommissioned && self.is_suspended(idx).await {
+            if suspended_pools.as_ref().is_some_and(|pools| pools[idx]) {
                 continue;
             }
 
@@ -646,10 +659,20 @@ impl ECStore {
             }
         }
 
+        let suspended_pools = match suspended_pools {
+            Some(pools) => pools,
+            None => {
+                let pool_meta = self.pool_meta.read().await;
+                (0..self.pools.len())
+                    .map(|idx| pool_meta.is_suspended(idx))
+                    .collect::<Vec<_>>()
+            }
+        };
+
         // 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)
+        resolve_latest_object_info_candidates_with_pool_state(candidates, &suspended_pools, bucket, object, opts)
     }
 
     pub(super) async fn delete_object_from_all_pools(
@@ -1526,6 +1549,83 @@ mod tests {
         assert_eq!(idx, 1);
     }
 
+    #[test]
+    fn resolve_latest_object_info_candidates_prefers_active_target_over_higher_suspended_source() {
+        let source = object_info_with_identity(10, false, Uuid::from_u128(1), Some("etag-a".to_string()));
+        let target = source.clone();
+
+        let (info, idx) = resolve_latest_object_info_candidates_with_pool_state(
+            vec![
+                LatestObjectInfoCandidate {
+                    info: Some(target),
+                    idx: 0,
+                    err: None,
+                },
+                LatestObjectInfoCandidate {
+                    info: Some(source),
+                    idx: 1,
+                    err: None,
+                },
+            ],
+            &[false, true],
+            "bucket",
+            "object",
+            &ObjectOptions::default(),
+        )
+        .expect("an active committed target should fence an equivalent suspended source");
+
+        assert_eq!(idx, 0);
+        assert_eq!(info.version_id, Some(Uuid::from_u128(1)));
+    }
+
+    #[test]
+    fn resolve_latest_object_info_candidates_keeps_lone_suspended_source_readable() {
+        let source = object_info_with_identity(10, false, Uuid::from_u128(1), Some("etag-a".to_string()));
+
+        let (_, idx) = resolve_latest_object_info_candidates_with_pool_state(
+            vec![LatestObjectInfoCandidate {
+                info: Some(source),
+                idx: 1,
+                err: None,
+            }],
+            &[false, true],
+            "bucket",
+            "object",
+            &ObjectOptions::default(),
+        )
+        .expect("a source-only object must remain readable before migration commits");
+
+        assert_eq!(idx, 1);
+    }
+
+    #[test]
+    fn resolve_latest_object_info_candidates_rejects_suspended_source_identity_conflict() {
+        let target = object_info_with_identity(10, false, Uuid::from_u128(1), Some("etag-new".to_string()));
+        let source = object_info_with_identity(10, false, Uuid::from_u128(1), Some("etag-old".to_string()));
+
+        let err = resolve_latest_object_info_candidates_with_pool_state(
+            vec![
+                LatestObjectInfoCandidate {
+                    info: Some(target),
+                    idx: 0,
+                    err: None,
+                },
+                LatestObjectInfoCandidate {
+                    info: Some(source),
+                    idx: 1,
+                    err: None,
+                },
+            ],
+            &[false, true],
+            "bucket",
+            "object",
+            &ObjectOptions::default(),
+        )
+        .expect_err("pool state must not mask an equal-time identity conflict");
+
+        assert_eq!(err, Error::ErasureReadQuorum);
+    }
+
     #[test]
     fn resolve_latest_object_info_candidates_keeps_index_fallback_for_fully_equivalent_identities() {
         let candidates = vec![

crates/ecstore/src/store/rebalance/support.rs

@@ -277,11 +277,22 @@ fn same_latest_object_info_identity(left: &ObjectInfo, right: &ObjectInfo) -> bo
     }
 }
 
+#[cfg(test)]
 pub(super) fn resolve_latest_object_info_candidates(
     candidates: Vec<LatestObjectInfoCandidate>,
     bucket: &str,
     object: &str,
     opts: &ObjectOptions,
+) -> Result<(ObjectInfo, usize)> {
+    resolve_latest_object_info_candidates_with_pool_state(candidates, &[], bucket, object, opts)
+}
+
+pub(super) fn resolve_latest_object_info_candidates_with_pool_state(
+    candidates: Vec<LatestObjectInfoCandidate>,
+    suspended_pools: &[bool],
+    bucket: &str,
+    object: &str,
+    opts: &ObjectOptions,
 ) -> Result<(ObjectInfo, usize)> {
     let latest_mod_time = candidates.iter().filter_map(latest_candidate_mod_time).max();
 
@@ -291,7 +302,15 @@ pub(super) fn resolve_latest_object_info_candidates(
             .filter(|candidate| latest_candidate_mod_time(candidate) == Some(latest_mod_time))
             .collect::<Vec<_>>();
 
-        latest_candidates.sort_by_key(|candidate| std::cmp::Reverse(candidate.idx));
+        // A decommission source remains readable until its target commits. Once
+        // equivalent copies exist, prefer the active target without hiding a
+        // same-time identity conflict behind pool state.
+        latest_candidates.sort_by_key(|candidate| {
+            (
+                suspended_pools.get(candidate.idx).copied().unwrap_or(false),
+                std::cmp::Reverse(candidate.idx),
+            )
+        });
 
         let Some(winner) = latest_candidates.first() else {
             return Err(Error::ErasureReadQuorum);