fix(ecstore): preserve migrated reads and retry busy pool heals (#8821)

crates/ecstore/src/cluster/rpc/peer_s3_client.rs

@@ -417,8 +417,10 @@ impl S3PeerSys {
         let errs = join_all(futures).await;
 
         let mut pool_errs = Vec::new();
+        let mut all_pools_have_participants = true;
         for pool_idx in 0..self.pools_count {
             let per_pool_errs = pool_participant_errors(&self.clients, &errs, pool_idx);
+            all_pools_have_participants &= !per_pool_errs.is_empty();
             pool_errs.push(reduce_pool_write_quorum_errs(&per_pool_errs));
         }
 
@@ -451,6 +453,17 @@ impl S3PeerSys {
         for pool_idx in 0..self.pools_count {
             let per_pool_errs = pool_participant_errors(&self.clients, &errs, pool_idx);
             if let Some(pool_err) = reduce_pool_write_quorum_errs(&per_pool_errs) {
+                // A success/busy split can miss quorum without a majority error.
+                // Keep that refusal retryable without masking other peer failures.
+                let pool_err = if pool_err == Error::ErasureWriteQuorum
+                    && all_pools_have_participants
+                    && per_pool_errs.iter().any(|err| err.as_ref() == Some(&Error::TooManyOpenFiles))
+                    && errs.iter().flatten().all(|err| *err == Error::TooManyOpenFiles)
+                {
+                    Error::TooManyOpenFiles
+                } else {
+                    pool_err
+                };
                 tracing::error!("heal_bucket per_pool_errs: {per_pool_errs:?}");
                 tracing::error!("heal_bucket reduce_write_quorum_errs: {pool_err}");
                 return Err(pool_err);
@@ -1810,6 +1823,7 @@ mod tests {
     #[derive(Debug)]
     struct TestPeerS3Client {
         pools: Option<Vec<usize>>,
+        heal_bucket_result: Result<HealResultItem>,
         make_bucket_result: Result<()>,
         list_bucket_result: Result<Vec<BucketInfo>>,
         delete_bucket_result: Result<()>,
@@ -1819,7 +1833,7 @@ mod tests {
     #[async_trait]
     impl PeerS3Client for TestPeerS3Client {
         async fn heal_bucket(&self, _bucket: &str, _opts: &HealOpts) -> Result<HealResultItem> {
-            unreachable!("not used by quorum tests")
+            self.heal_bucket_result.clone()
         }
 
         async fn make_bucket(&self, _bucket: &str, _opts: &MakeBucketOptions) -> Result<()> {
@@ -1835,8 +1849,11 @@ mod tests {
             self.delete_bucket_result.clone()
         }
 
-        async fn get_bucket_info(&self, _bucket: &str, _opts: &BucketOptions) -> Result<BucketInfo> {
-            unreachable!("not used by quorum tests")
+        async fn get_bucket_info(&self, bucket: &str, _opts: &BucketOptions) -> Result<BucketInfo> {
+            Ok(BucketInfo {
+                name: bucket.to_string(),
+                ..Default::default()
+            })
         }
 
         fn get_pools(&self) -> Option<Vec<usize>> {
@@ -1848,6 +1865,17 @@ mod tests {
         test_peer_with_make_bucket(pools, Ok(()))
     }
 
+    fn test_peer_with_heal_bucket(pools: &[usize], heal_bucket_result: Result<HealResultItem>) -> Client {
+        Arc::new(Box::new(TestPeerS3Client {
+            pools: Some(pools.to_vec()),
+            heal_bucket_result,
+            make_bucket_result: Ok(()),
+            list_bucket_result: Ok(Vec::new()),
+            delete_bucket_result: Ok(()),
+            make_bucket_calls: Arc::new(AtomicUsize::new(0)),
+        }))
+    }
+
     fn test_peer_with_make_bucket(pools: &[usize], make_bucket_result: Result<()>) -> Client {
         test_peer_with_results(pools, make_bucket_result, Ok(Vec::new()))
     }
@@ -1885,6 +1913,7 @@ mod tests {
     ) -> Client {
         Arc::new(Box::new(TestPeerS3Client {
             pools: Some(pools.to_vec()),
+            heal_bucket_result: Ok(HealResultItem::default()),
             make_bucket_result,
             list_bucket_result,
             delete_bucket_result,
@@ -2634,6 +2663,119 @@ mod tests {
         assert_eq!(err, Error::VolumeExists);
     }
 
+    #[tokio::test]
+    async fn test_heal_bucket_preserves_busy_refusal_below_error_quorum() {
+        for busy_count in 2..=4 {
+            let peer_sys = S3PeerSys {
+                clients: (0..4)
+                    .map(|peer| {
+                        test_peer_with_heal_bucket(
+                            &[0],
+                            if peer < busy_count {
+                                Err(Error::TooManyOpenFiles)
+                            } else {
+                                Ok(HealResultItem::default())
+                            },
+                        )
+                    })
+                    .collect(),
+                pools_count: 1,
+            };
+
+            let error = peer_sys
+                .heal_bucket("bucket", &HealOpts::default())
+                .await
+                .expect_err("busy peers must not turn a failed heal into success");
+            assert_eq!(error, Error::TooManyOpenFiles, "busy peers: {busy_count}");
+        }
+    }
+
+    #[tokio::test]
+    async fn test_heal_bucket_busy_peer_preserves_success_quorum() {
+        let peer_sys = S3PeerSys {
+            clients: vec![
+                test_peer_with_heal_bucket(&[0], Err(Error::TooManyOpenFiles)),
+                test_peer(&[0]),
+                test_peer(&[0]),
+                test_peer(&[0]),
+            ],
+            pools_count: 1,
+        };
+
+        peer_sys
+            .heal_bucket("bucket", &HealOpts::default())
+            .await
+            .expect("three successful peers still satisfy the four-peer write quorum");
+    }
+
+    #[tokio::test]
+    async fn test_heal_bucket_busy_peers_do_not_hide_other_failures() {
+        for failure in [Error::FaultyDisk, Error::VolumeNotFound, Error::ErasureWriteQuorum] {
+            let peer_sys = S3PeerSys {
+                clients: vec![
+                    test_peer_with_heal_bucket(&[0], Err(Error::TooManyOpenFiles)),
+                    test_peer_with_heal_bucket(&[0], Err(Error::TooManyOpenFiles)),
+                    test_peer_with_heal_bucket(&[0], Err(failure.clone())),
+                    test_peer(&[0]),
+                ],
+                pools_count: 1,
+            };
+
+            let error = peer_sys
+                .heal_bucket("bucket", &HealOpts::default())
+                .await
+                .expect_err("a mixed failure must still fail write quorum");
+            assert_eq!(error, Error::ErasureWriteQuorum, "non-busy failure: {failure}");
+        }
+    }
+
+    #[tokio::test]
+    async fn test_heal_bucket_busy_pool_does_not_hide_another_pool_failure() {
+        let peer_sys = S3PeerSys {
+            clients: vec![
+                test_peer_with_heal_bucket(&[0], Err(Error::TooManyOpenFiles)),
+                test_peer_with_heal_bucket(&[0], Err(Error::TooManyOpenFiles)),
+                test_peer(&[0]),
+                test_peer(&[0]),
+                test_peer_with_heal_bucket(&[1], Err(Error::FaultyDisk)),
+            ],
+            pools_count: 2,
+        };
+
+        let error = peer_sys
+            .heal_bucket("bucket", &HealOpts::default())
+            .await
+            .expect_err("the busy split must not hide a non-busy failure from another pool");
+        assert_eq!(error, Error::ErasureWriteQuorum);
+    }
+
+    #[tokio::test]
+    async fn test_heal_bucket_missing_pool_participants_remains_quorum_failure() {
+        for occupied_pool in 0..2 {
+            let peer_sys = S3PeerSys {
+                clients: vec![
+                    test_peer_with_heal_bucket(&[occupied_pool], Err(Error::TooManyOpenFiles)),
+                    test_peer_with_heal_bucket(&[occupied_pool], Err(Error::TooManyOpenFiles)),
+                    test_peer(&[occupied_pool]),
+                    test_peer(&[occupied_pool]),
+                ],
+                pools_count: 2,
+            };
+
+            let error = peer_sys
+                .heal_bucket(
+                    "bucket",
+                    &HealOpts {
+                        recreate: true,
+                        ..Default::default()
+                    },
+                )
+                .await
+                .expect_err("a pool without participants cannot be treated as a busy gate");
+            assert_eq!(error, Error::ErasureWriteQuorum, "pool with participants: {occupied_pool}");
+        }
+    }
+
     #[test]
     fn heal_bucket_mode_fails_closed_on_incomplete_topology() {
         let mut opts = HealOpts::default();

crates/ecstore/src/core/pools.rs

@@ -33554,12 +33554,15 @@ mod pools_tests {
 
     #[test]
     fn test_resolve_decommission_preflight_heal_result_wraps_error_context() {
-        let err = resolve_decommission_preflight_heal_result::<()>("bucket-a", Err(Error::FaultyDisk))
-            .expect_err("heal failure should carry preflight context");
-        assert!(
-            err.to_string()
-                .contains("decommission preflight heal failed for bucket bucket-a")
-        );
+        for failure in [Error::FaultyDisk, Error::ErasureWriteQuorum] {
+            let err = resolve_decommission_preflight_heal_result::<()>("bucket-a", Err(failure))
+                .expect_err("heal failure should carry preflight context");
+            assert!(
+                err.to_string()
+                    .contains("decommission preflight heal failed for bucket bucket-a")
+            );
+            assert_ne!(err, Error::SlowDown, "non-busy failures must not become a retryable start");
+        }
     }
 
     #[test]

crates/ecstore/src/store/init.rs

@@ -10501,6 +10501,20 @@ mod tests {
             )
             .await
             .expect("seed data movement source");
+        store.pools[0]
+            .put_object(
+                &bucket,
+                object,
+                &mut PutObjReader::from_vec(b"newer source body".to_vec()),
+                &ObjectOptions {
+                    versioned: true,
+                    version_id: Some(uuid::Uuid::new_v4().to_string()),
+                    mod_time: Some(mod_time + time::Duration::SECOND),
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("retain an unmoved newer source version");
         *store.rebalance_meta.write().await = Some(active_rebalance_meta_for_pool(store.pools.len(), 0));
 
         let source_reader = store.pools[0]
@@ -10519,6 +10533,8 @@ mod tests {
             .await
             .expect("read data movement source");
         let source_data_dir = source_reader.object_info.data_dir;
+        assert!(!source_reader.object_info.is_latest, "the copied version is historical on the source");
+        assert_eq!(source_reader.object_info.num_versions, 2);
         crate::data_movement::migrate_object(store.clone(), 0, bucket.clone(), source_reader, None, "test_read_window")
             .await
             .expect("commit data movement target");
@@ -10535,6 +10551,8 @@ mod tests {
             )
             .await
             .expect("read committed data movement target");
+        assert!(target.is_latest, "the target has only the copied historical version");
+        assert_eq!(target.num_versions, 1);
         assert_ne!(source_data_dir, target.data_dir, "the target must use its own data directory");
         assert_eq!(
             rustfs_utils::http::get_consistent_str(&target.user_defined, rustfs_utils::http::SUFFIX_DATA_MOVED),
@@ -10555,6 +10573,27 @@ mod tests {
             .expect("the equal-time migration window must remain readable");
         assert_eq!(resolved.data_dir, target.data_dir);
 
+        let mut migrated = store
+            .get_object_reader(
+                &bucket,
+                object,
+                None,
+                HeaderMap::new(),
+                &ObjectOptions {
+                    version_id: Some(version.to_string()),
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("a partial target history must not hide the readable historical version");
+        let mut body = Vec::new();
+        migrated
+            .stream
+            .read_to_end(&mut body)
+            .await
+            .expect("read migrated historical body");
+        assert_eq!(body, b"source body");
+
         shutdown.cancel();
     }
 

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

@@ -186,21 +186,18 @@ fn same_user_defined_identity(left: &ObjectInfo, right: &ObjectInfo) -> bool {
     }
 }
 
-/// Pool-specific erasure geometry and data-movement rewrites are intentionally
-/// excluded. The selected winner still carries the chosen pool's layout, while
-/// the remaining read-visible fields must agree before the pool index can
-/// provide a deterministic tie-break.
+/// Pool-specific erasure geometry, version history, and data-movement rewrites
+/// are intentionally excluded. A copied version can be the target's latest
+/// while the source still holds newer versions. The selected winner keeps its
+/// pool's layout; the version's read-visible identity must still agree.
 pub(super) fn same_latest_object_info_identity(left: &ObjectInfo, right: &ObjectInfo) -> bool {
     let same_read_surface = left.bucket == right.bucket
         && left.name == right.name
         && left.is_dir == right.is_dir
         && left.restore_ongoing == right.restore_ongoing
         && left.restore_expires == right.restore_expires
-        && left.is_latest == right.is_latest
         && left.content_type == right.content_type
         && left.content_encoding == right.content_encoding
-        && left.num_versions == right.num_versions
-        && left.successor_mod_time == right.successor_mod_time
         && left.inlined == right.inlined
         && left.metadata_only == right.metadata_only
         && left.version_only == right.version_only
@@ -250,7 +247,13 @@ pub(super) fn same_latest_object_info_identity(left: &ObjectInfo, right: &Object
 }
 
 pub(super) fn validate_prepared_pool_refetch_identity(expected: &ObjectInfo, refetched: &ObjectInfo) -> Result<()> {
-    if same_latest_object_info_identity(expected, refetched) {
+    // A refetch stays in the same pool, so its history must still match the
+    // snapshot even though cross-pool copies may have different inventories.
+    if expected.is_latest == refetched.is_latest
+        && expected.num_versions == refetched.num_versions
+        && expected.successor_mod_time == refetched.successor_mod_time
+        && same_latest_object_info_identity(expected, refetched)
+    {
         Ok(())
     } else {
         Err(Error::ErasureReadQuorum)
@@ -348,6 +351,34 @@ mod tests {
         assert_eq!(error, Error::ErasureReadQuorum);
     }
 
+    #[test]
+    fn prepared_pool_refetch_identity_fails_closed_on_version_history_change() {
+        let expected = ObjectInfo {
+            mod_time: Some(OffsetDateTime::from_unix_timestamp(10).expect("test timestamp should be valid")),
+            version_id: Some(uuid::Uuid::from_u128(1)),
+            etag: Some("etag-a".to_string()),
+            ..Default::default()
+        };
+        for refetched in [
+            ObjectInfo {
+                is_latest: !expected.is_latest,
+                ..expected.clone()
+            },
+            ObjectInfo {
+                num_versions: expected.num_versions + 1,
+                ..expected.clone()
+            },
+            ObjectInfo {
+                successor_mod_time: Some(OffsetDateTime::from_unix_timestamp(20).expect("successor timestamp should be valid")),
+                ..expected.clone()
+            },
+        ] {
+            let error = validate_prepared_pool_refetch_identity(&expected, &refetched)
+                .expect_err("a changed same-pool history must invalidate the prepared snapshot");
+            assert_eq!(error, Error::ErasureReadQuorum);
+        }
+    }
+
     #[test]
     fn rebalance_delete_aggregation_preserves_typed_failures() {
         for expected in [

docs/testing/scanner-heal-topology-gates.md

@@ -58,6 +58,7 @@ The barrier directory is set for every node of a run; only the process whose ope
 |---|---|---|
 | movement active | 30 s | matches the existing decommission and rebalance cases |
 | movement complete | 240 s | the existing cases allow 180 s for 96 objects of 128-256 KiB; this inventory adds 1 MiB externals, multipart and version history (about 45 MiB) |
+| resumed decommission complete | 360 s | recovery includes the fleet-proof retry before migration resumes; a healthy CI restart completed after 253 s while still making progress, below the distributed lane's 720 s process limit |
 | movement barrier reach | 240 s (the movement budget) | a window can sit deep in the object order; on Linux CI a busy target capacity gate (160 inline retries, entries deferred to round end) delayed the first fault object by about 2 minutes. The wait fails at once if the operation completes, fails or is canceled without entering the window. Each status probe is bounded to 5 s and a slow probe counts as "still running", because the admin status request can wait on the held mover until the barrier itself times out. The server releases a held barrier on its own after 120 s |
 | pre-ACK window reach | 60 s | the decommission start normally reaches its durable save in about a second (POST to response ~0.8 s locally); exceeding 60 s is reported as a stalled start path, not retried with a longer budget |
 | cancel, stop and terminal transitions | 90 s | stop and cancel are acknowledged before every in-flight entry finishes its cleanup, so terminal state and placement are polled within this budget |