fix(rebalance): defer changed source cleanup (#5829)

crates/ecstore/src/core/pools.rs

@@ -3017,7 +3017,14 @@ impl ECStore {
                 &cleanup_preflight_allowed_missing,
                 "decommission",
             )
-            .await;
+            .await
+            .map_err(|err| match err {
+                data_movement::SourceCleanupError::SourceChanged => Error::other(format!(
+                    "decommission: source cleanup preflight failed for {}/{}: source versions changed after migration started",
+                    bucket, entry.name
+                )),
+                data_movement::SourceCleanupError::Storage(err) => err,
+            });
             resolve_decommission_entry_cleanup_delete_result(cleanup_result, bucket.as_str(), entry.name.as_str())?
         } else if decommissioned != fivs.versions.len() || expired > 0 {
             warn!(

crates/ecstore/src/data_movement/mod.rs

@@ -437,6 +437,15 @@ struct SourceCleanupPartIdentity {
     checksums: BTreeMap<String, String>,
 }
 
+#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
+struct SourceCleanupErasureIdentity {
+    algorithm: String,
+    data_blocks: usize,
+    parity_blocks: usize,
+    block_size: usize,
+    distribution: Vec<usize>,
+}
+
 #[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
 pub(crate) struct SourceCleanupVersionIdentity {
     name: String,
@@ -447,10 +456,28 @@ pub(crate) struct SourceCleanupVersionIdentity {
     etag: Option<String>,
     checksum: Option<Vec<u8>>,
     data_dir: Option<uuid::Uuid>,
+    transition_status: String,
+    transitioned_objname: String,
+    transition_tier: String,
+    transition_version_id: Option<uuid::Uuid>,
+    transition_version: Option<String>,
+    transition_version_state: u8,
+    expire_restored: bool,
+    erasure: SourceCleanupErasureIdentity,
     metadata: BTreeMap<String, String>,
     parts: Vec<SourceCleanupPartIdentity>,
 }
 
+fn source_cleanup_erasure_identity(erasure: &rustfs_filemeta::ErasureInfo) -> SourceCleanupErasureIdentity {
+    SourceCleanupErasureIdentity {
+        algorithm: erasure.algorithm.clone(),
+        data_blocks: erasure.data_blocks,
+        parity_blocks: erasure.parity_blocks,
+        block_size: erasure.block_size,
+        distribution: erasure.distribution.clone(),
+    }
+}
+
 fn source_cleanup_part_identity(part: &ObjectPartInfo) -> SourceCleanupPartIdentity {
     SourceCleanupPartIdentity {
         number: part.number,
@@ -480,6 +507,19 @@ pub(crate) fn source_cleanup_version_identity(version: &FileInfo) -> SourceClean
         etag: version.get_etag(),
         checksum: version.checksum.as_ref().map(|checksum| checksum.to_vec()),
         data_dir: version.data_dir,
+        transition_status: version.transition_status.clone(),
+        transitioned_objname: version.transitioned_objname.clone(),
+        transition_tier: version.transition_tier.clone(),
+        transition_version_id: version.transition_version_id,
+        transition_version: version.transition_version.clone(),
+        transition_version_state: match version.transition_version_state {
+            rustfs_filemeta::TransitionVersionState::Unknown => 0,
+            rustfs_filemeta::TransitionVersionState::KnownDisabled => 1,
+            rustfs_filemeta::TransitionVersionState::SuspendedNull => 2,
+            rustfs_filemeta::TransitionVersionState::Exact => 3,
+        },
+        expire_restored: version.expire_restored,
+        erasure: source_cleanup_erasure_identity(&version.erasure),
         metadata: version
             .metadata
             .iter()
@@ -495,10 +535,6 @@ fn source_cleanup_version_identities(fivs: &FileInfoVersions) -> Vec<SourceClean
     identities
 }
 
-fn source_cleanup_versions_match(expected: &FileInfoVersions, current: &FileInfoVersions) -> bool {
-    source_cleanup_versions_match_with_allowed_missing(expected, current, &[])
-}
-
 fn source_cleanup_versions_match_with_allowed_missing(
     expected: &FileInfoVersions,
     current: &FileInfoVersions,
@@ -530,6 +566,26 @@ fn source_cleanup_versions_match_with_allowed_missing(
         .all(|(identity, count)| allowed_counts.get(&identity).copied().unwrap_or_default() >= count)
 }
 
+#[derive(Debug, thiserror::Error)]
+pub(crate) enum SourceCleanupError {
+    #[error("source versions changed after migration started")]
+    SourceChanged,
+    #[error(transparent)]
+    Storage(#[from] Error),
+}
+
+fn ensure_source_cleanup_versions_match(
+    expected: &FileInfoVersions,
+    current: &FileInfoVersions,
+    allowed_missing: &[SourceCleanupVersionIdentity],
+) -> std::result::Result<(), SourceCleanupError> {
+    if source_cleanup_versions_match_with_allowed_missing(expected, current, allowed_missing) {
+        Ok(())
+    } else {
+        Err(SourceCleanupError::SourceChanged)
+    }
+}
+
 fn source_cleanup_preflight_error(op_label: &str, bucket: &str, object: &str, err: impl std::fmt::Display) -> Error {
     Error::other(format!("{op_label}: source cleanup preflight failed for {bucket}/{object}: {err}"))
 }
@@ -552,21 +608,12 @@ pub(crate) async fn ensure_source_cleanup_versions_unchanged(
     expected: &FileInfoVersions,
     allowed_missing: &[SourceCleanupVersionIdentity],
     op_label: &str,
-) -> Result<()> {
+) -> std::result::Result<(), SourceCleanupError> {
     let Some(current) = load_source_cleanup_versions(set, bucket, object, op_label).await? else {
         return Ok(());
     };
 
-    if source_cleanup_versions_match_with_allowed_missing(expected, &current, allowed_missing) {
+    ensure_source_cleanup_versions_match(expected, &current, allowed_missing)
-        return Ok(());
-    }
-
-    Err(source_cleanup_preflight_error(
-        op_label,
-        bucket,
-        object,
-        "source versions changed after migration started",
-    ))
 }
 
 #[cfg(test)]
@@ -651,10 +698,13 @@ pub(crate) async fn cleanup_source_entry_if_unchanged(
     expected: &FileInfoVersions,
     allowed_missing: &[SourceCleanupVersionIdentity],
     op_label: &str,
-) -> Result<ObjectInfo> {
+) -> std::result::Result<ObjectInfo, SourceCleanupError> {
     let cleanup_key = encode_dir_object(object);
     let ns_lock = set.new_ns_lock(bucket, cleanup_key.as_str()).await?;
-    let _guard = ns_lock.get_write_lock(get_lock_acquire_timeout()).await?;
+    let _guard = ns_lock
+        .get_write_lock(get_lock_acquire_timeout())
+        .await
+        .map_err(Error::from)?;
 
     ensure_source_cleanup_versions_unchanged(set.clone(), bucket, object, expected, allowed_missing, op_label).await?;
 
@@ -673,7 +723,7 @@ pub(crate) async fn cleanup_source_entry_if_unchanged(
     if result.is_ok() {
         crate::store::list_objects::observe_scanner_namespace_mutations(bucket, 1);
     }
-    result
+    result.map_err(SourceCleanupError::from)
 }
 
 fn should_check_data_movement_resume_target(src_pool_idx: usize, target_pool_idx: usize) -> bool {
@@ -1182,7 +1232,7 @@ mod tests {
         let expected = cleanup_test_versions(vec![first.clone(), second.clone()]);
         let current = cleanup_test_versions(vec![second, first]);
 
-        assert!(source_cleanup_versions_match(&expected, &current));
+        assert!(source_cleanup_versions_match_with_allowed_missing(&expected, &current, &[]));
     }
 
     #[test]
@@ -1190,7 +1240,40 @@ mod tests {
         let expected = cleanup_test_versions(vec![cleanup_test_file_info("object.txt", Uuid::from_u128(1), "source")]);
         let current = cleanup_test_versions(vec![cleanup_test_file_info("object.txt", Uuid::from_u128(1), "changed")]);
 
-        assert!(!source_cleanup_versions_match(&expected, &current));
+        let err = ensure_source_cleanup_versions_match(&expected, &current, &[])
+            .expect_err("changed source metadata must defer cleanup");
+        assert!(matches!(err, SourceCleanupError::SourceChanged));
+    }
+
+    #[test]
+    fn test_source_cleanup_preflight_rejects_changed_transition_or_erasure() {
+        let expected = cleanup_test_versions(vec![cleanup_test_file_info("object.txt", Uuid::from_u128(1), "source")]);
+        let mut current = expected.clone();
+        current.versions[0].transition_tier = "COLD".to_string();
+        let err = ensure_source_cleanup_versions_match(&expected, &current, &[])
+            .expect_err("transition metadata changes must defer cleanup");
+        assert!(matches!(err, SourceCleanupError::SourceChanged));
+
+        let mut current = expected.clone();
+        current.versions[0].erasure.algorithm = "changed".to_string();
+        let err = ensure_source_cleanup_versions_match(&expected, &current, &[])
+            .expect_err("erasure metadata changes must defer cleanup");
+        assert!(matches!(err, SourceCleanupError::SourceChanged));
+    }
+
+    #[test]
+    fn test_source_cleanup_preflight_ignores_per_disk_erasure_fields() {
+        let mut expected = cleanup_test_versions(vec![cleanup_test_file_info("object.txt", Uuid::from_u128(1), "source")]);
+        expected.versions[0].erasure.checksums = vec![rustfs_filemeta::ChecksumInfo {
+            part_number: 1,
+            hash: Bytes::from_static(b"disk-a-checksum"),
+            ..Default::default()
+        }];
+        let mut current = expected.clone();
+        current.versions[0].erasure.index = 7;
+        current.versions[0].erasure.checksums[0].hash = Bytes::from_static(b"disk-b-checksum");
+
+        assert!(source_cleanup_versions_match_with_allowed_missing(&expected, &current, &[]));
     }
 
     #[test]
@@ -1201,7 +1284,9 @@ mod tests {
             cleanup_test_file_info("object.txt", Uuid::from_u128(2), "new-version"),
         ]);
 
-        assert!(!source_cleanup_versions_match(&expected, &current));
+        let err = ensure_source_cleanup_versions_match(&expected, &current, &[])
+            .expect_err("an added source version must defer cleanup");
+        assert!(matches!(err, SourceCleanupError::SourceChanged));
     }
 
     #[test]
@@ -1222,7 +1307,9 @@ mod tests {
         let expected = cleanup_test_versions(vec![migrated.clone(), protected]);
         let current = cleanup_test_versions(vec![migrated]);
 
-        assert!(!source_cleanup_versions_match_with_allowed_missing(&expected, &current, &[]));
+        let err = ensure_source_cleanup_versions_match(&expected, &current, &[])
+            .expect_err("an unexpected missing version must defer cleanup");
+        assert!(matches!(err, SourceCleanupError::SourceChanged));
     }
 
     #[test]
@@ -1234,7 +1321,9 @@ mod tests {
         let current = cleanup_test_versions(vec![migrated, new_version]);
         let allowed_missing = vec![source_cleanup_version_identity(&expired)];
 
-        assert!(!source_cleanup_versions_match_with_allowed_missing(&expected, &current, &allowed_missing));
+        let err = ensure_source_cleanup_versions_match(&expected, &current, &allowed_missing)
+            .expect_err("a new source version must defer cleanup even when an expired version may be missing");
+        assert!(matches!(err, SourceCleanupError::SourceChanged));
     }
 
     #[test]

crates/ecstore/src/services/rebalance/entry.rs

@@ -18,8 +18,8 @@ use super::meta::{
 };
 use super::migration::migrate_entry_version;
 use super::worker::{
-    RebalanceEntryTask, load_rebalance_bucket_configs, rebalance_max_attempts, resolve_rebalance_bucket_error,
+    RebalanceEntryCleanupResult, RebalanceEntryTask, load_rebalance_bucket_configs, rebalance_max_attempts,
-    resolve_rebalance_entry_cleanup_delete_result, resolve_rebalance_file_info_versions_result,
+    resolve_rebalance_bucket_error, resolve_rebalance_entry_cleanup_delete_result, resolve_rebalance_file_info_versions_result,
     resolve_rebalance_migrate_result_error, resolve_rebalance_stats_update_result, resolve_rebalance_worker_result,
     run_rebalance_listing_with_retry, should_cleanup_rebalance_source_entry, should_count_rebalance_version_complete,
     should_defer_rebalance_entry_failure, should_skip_rebalance_delete_marker, wait_rebalance_entry_tasks,
@@ -50,11 +50,14 @@ impl ECStore {
         bucket: &str,
         object: &str,
         stats_updates: &[&FileInfo],
-        cleanup: impl std::future::Future<Output = Result<ObjectInfo>>,
+        cleanup: impl std::future::Future<Output = std::result::Result<ObjectInfo, data_movement::SourceCleanupError>>,
-    ) -> Result<Option<String>> {
+    ) -> Result<RebalanceEntryCleanupResult> {
         // Persisted stats can complete a pool on restart, so source cleanup must resolve first.
-        let cleanup_warning = resolve_rebalance_entry_cleanup_delete_result(cleanup.await, bucket, object)?;
+        let cleanup_result = resolve_rebalance_entry_cleanup_delete_result(cleanup.await, bucket, object);
-        if let Some(message) = cleanup_warning.as_ref()
+        let RebalanceEntryCleanupResult::Completed { warning } = cleanup_result else {
+            return Ok(cleanup_result);
+        };
+        if let Some(message) = warning.as_ref()
             && let Err(err) = self
                 .record_rebalance_cleanup_warning(pool_index, bucket, object, message.clone())
                 .await
@@ -80,7 +83,7 @@ impl ECStore {
             object,
         )?;
 
-        Ok(cleanup_warning)
+        Ok(RebalanceEntryCleanupResult::Completed { warning })
     }
 
     #[allow(unused_assignments)]
@@ -255,7 +258,7 @@ impl ECStore {
                 );
                 if should_defer_rebalance_entry_failure(&err) {
                     let deferred_error = format!("{REBALANCE_DEFERRED_ENTRY_ERROR_PREFIX} {err}");
-                    warn!(
+                    debug!(
                         event = EVENT_REBALANCE_ENTRY,
                         component = LOG_COMPONENT_ECSTORE,
                         subsystem = LOG_SUBSYSTEM_REBALANCE,
@@ -300,7 +303,7 @@ impl ECStore {
         }
 
         if should_cleanup_rebalance_source_entry(rebalanced, fivs.versions.len(), expired) {
-            let cleanup_warning = self
+            let cleanup_result = self
                 .finish_rebalance_entry_after_cleanup(
                     pool_index,
                     bucket.as_str(),
@@ -316,30 +319,47 @@ impl ECStore {
                     ),
                 )
                 .await?;
-            if let Some(message) = cleanup_warning {
+            match cleanup_result {
-                warn!(
+                RebalanceEntryCleanupResult::Deferred { last_error } => {
-                    event = EVENT_REBALANCE_ENTRY,
+                    debug!(
-                    component = LOG_COMPONENT_ECSTORE,
+                        event = EVENT_REBALANCE_ENTRY,
-                    subsystem = LOG_SUBSYSTEM_REBALANCE,
+                        component = LOG_COMPONENT_ECSTORE,
-                    pool_index,
+                        subsystem = LOG_SUBSYSTEM_REBALANCE,
-                    bucket = %bucket,
+                        pool_index,
-                    object = %entry.name,
+                        bucket = %bucket,
-                    stage = "cleanup_source",
+                        object = %entry.name,
-                    cleanup_status = "failed_ignored",
+                        state = "deferred",
-                    error = %message,
+                        error = %last_error,
-                    "Ignored rebalance source cleanup failure"
+                        "Deferred rebalance entry after source cleanup conflict"
-                );
+                    );
-            } else {
+                    return Ok(RebalanceEntryOutcome::Deferred { last_error });
-                debug!(
+                }
-                    event = EVENT_REBALANCE_ENTRY,
+                RebalanceEntryCleanupResult::Completed { warning: Some(message) } => {
-                    component = LOG_COMPONENT_ECSTORE,
+                    warn!(
-                    subsystem = LOG_SUBSYSTEM_REBALANCE,
+                        event = EVENT_REBALANCE_ENTRY,
-                    pool_index,
+                        component = LOG_COMPONENT_ECSTORE,
-                    bucket = %bucket,
+                        subsystem = LOG_SUBSYSTEM_REBALANCE,
-                    object = %entry.name,
+                        pool_index,
-                    state = "source_deleted",
+                        bucket = %bucket,
-                    "Deleted rebalance source entry"
+                        object = %entry.name,
-                );
+                        stage = "cleanup_source",
+                        cleanup_status = "failed_ignored",
+                        error = %message,
+                        "Ignored rebalance source cleanup failure"
+                    );
+                }
+                RebalanceEntryCleanupResult::Completed { warning: None } => {
+                    debug!(
+                        event = EVENT_REBALANCE_ENTRY,
+                        component = LOG_COMPONENT_ECSTORE,
+                        subsystem = LOG_SUBSYSTEM_REBALANCE,
+                        pool_index,
+                        bucket = %bucket,
+                        object = %entry.name,
+                        state = "source_deleted",
+                        "Deleted rebalance source entry"
+                    );
+                }
             }
         } else if rebalanced != fivs.versions.len() || expired > 0 {
             warn!(
@@ -655,13 +675,12 @@ mod tests {
         );
 
         release_cleanup.send(()).expect("cleanup waiter should remain alive");
-        assert!(
+        assert_eq!(
             finish
                 .await
                 .expect("finish task should not panic")
-                .expect("finish should succeed")
+                .expect("finish should succeed"),
-                .is_none(),
+            RebalanceEntryCleanupResult::Completed { warning: None }
-            "successful cleanup should not produce a warning"
         );
         assert!(
             store
@@ -680,14 +699,33 @@ mod tests {
             let mut meta = store.rebalance_meta.write().await;
             meta.as_mut().expect("rebalance metadata should exist").pool_stats[0].bytes = 0;
         }
-        let warning = store
+        let warning_result = store
-            .finish_rebalance_entry_after_cleanup(0, "bucket", "object.bin", &[&warning_version], async { Err(Error::SlowDown) })
+            .finish_rebalance_entry_after_cleanup(0, "bucket", "object.bin", &[&warning_version], async {
+                Err(Error::SlowDown.into())
+            })
             .await
             .expect("cleanup warnings should not fail the completed migration");
-        assert!(warning.is_some(), "cleanup failure should return a warning");
+        assert!(matches!(warning_result, RebalanceEntryCleanupResult::Completed { warning: Some(_) }));
         let meta = store.rebalance_meta.read().await;
         let pool_stats = &meta.as_ref().expect("rebalance metadata should exist").pool_stats[0];
         assert_eq!(pool_stats.cleanup_warnings.count, 1, "cleanup warning must block pool completion");
         assert!(pool_stats.bytes > 0, "completed migration bytes should still be recorded");
+        drop(meta);
+
+        {
+            let mut meta = store.rebalance_meta.write().await;
+            meta.as_mut().expect("rebalance metadata should exist").pool_stats[0].bytes = 0;
+        }
+        let deferred = store
+            .finish_rebalance_entry_after_cleanup(0, "bucket", "object.bin", &[&warning_version], async {
+                Err(data_movement::SourceCleanupError::SourceChanged)
+            })
+            .await
+            .expect("source changes should defer cleanup without failing the worker");
+        assert!(matches!(deferred, RebalanceEntryCleanupResult::Deferred { .. }));
+        let meta = store.rebalance_meta.read().await;
+        let pool_stats = &meta.as_ref().expect("rebalance metadata should exist").pool_stats[0];
+        assert_eq!(pool_stats.bytes, 0, "deferred cleanup must not commit completion stats");
+        assert_eq!(pool_stats.cleanup_warnings.count, 1, "deferred cleanup must not add a permanent warning");
     }
 }

crates/ecstore/src/services/rebalance/mod.rs

@@ -27,12 +27,14 @@ const REBAL_META_FMT: u16 = 1; // Replace with actual format value
 const REBAL_META_VER: u16 = 1; // Replace with actual version value
 pub(crate) const REBAL_META_NAME: &str = "rebalance.bin";
 const DEFAULT_REBALANCE_MAX_ATTEMPTS: usize = 3;
+pub(crate) const REBALANCE_SOURCE_CLEANUP_MAX_DEFERS: usize = 3;
 const REBALANCE_MAX_ATTEMPTS_ENV: &str = "RUSTFS_REBALANCE_MAX_ATTEMPTS";
 const REBALANCE_STOP_PROPAGATION_ERROR_PREFIX: &str = "rebalance stop propagation incomplete: ";
 const REBALANCE_LISTING_RETRY_BASE_DELAY: Duration = Duration::from_millis(250);
 const REBALANCE_MIGRATION_RETRY_BASE_DELAY: Duration = Duration::from_millis(250);
 const REBALANCE_MIGRATION_LOCK_RETRY_CAP: Duration = Duration::from_secs(10);
 const REBALANCE_DEFERRED_ENTRY_ERROR_PREFIX: &str = "deferred transient rebalance entry failure:";
+pub(crate) const REBALANCE_SOURCE_CLEANUP_DEFERRED_ERROR_PREFIX: &str = "deferred rebalance source cleanup conflict:";
 const REBALANCE_CLEANUP_WARNING_ENTRY_LIMIT: usize = 10;
 
 mod control;

crates/ecstore/src/services/rebalance/rebalance_unit_tests.rs

@@ -12,7 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use super::REBALANCE_DEFERRED_ENTRY_ERROR_PREFIX;
 use super::control::validate_rebalance_disk_stats_coverage;
 use super::meta::{
     RebalanceMetaMergeOutcome, RebalanceTerminalEvent, apply_rebalance_save_option, apply_rebalance_terminal_event,
@@ -33,23 +32,27 @@ use super::migration::{
     MigrationBackend, MigrationVersionResult, migrate_entry_version, migrate_entry_version_with_retry_wait,
     rebalance_delete_marker_opts,
 };
+use super::runtime::{should_fail_repeated_rebalance_bucket_defer, source_cleanup_defer_attempt};
 use super::worker::{
-    ensure_rebalance_listing_disks_available, is_transient_rebalance_error, parse_rebalance_max_attempts,
+    RebalanceEntryCleanupResult, ensure_rebalance_listing_disks_available, is_transient_rebalance_error,
-    rebalance_listing_retry_delay, rebalance_migration_retry_delay, resolve_load_rebalance_stats_update_result,
+    parse_rebalance_max_attempts, rebalance_listing_retry_delay, rebalance_migration_retry_delay,
-    resolve_rebalance_bucket_error, resolve_rebalance_bucket_result, resolve_rebalance_entry_cleanup_delete_result,
+    resolve_load_rebalance_stats_update_result, resolve_rebalance_bucket_error, resolve_rebalance_bucket_result,
-    resolve_rebalance_file_info_versions_result, resolve_rebalance_meta_load_result, resolve_rebalance_meta_save_result,
+    resolve_rebalance_entry_cleanup_delete_result, resolve_rebalance_file_info_versions_result,
-    resolve_rebalance_migrate_result_error, resolve_rebalance_optional_bucket_config_result, resolve_rebalance_save_task_result,
+    resolve_rebalance_meta_load_result, resolve_rebalance_meta_save_result, resolve_rebalance_migrate_result_error,
-    resolve_rebalance_stats_update_result, resolve_rebalance_terminal_error, resolve_rebalance_worker_result,
+    resolve_rebalance_optional_bucket_config_result, resolve_rebalance_save_task_result, resolve_rebalance_stats_update_result,
-    run_rebalance_listing_with_retry, send_rebalance_done_signal, should_cleanup_rebalance_source_entry,
+    resolve_rebalance_terminal_error, resolve_rebalance_worker_result, run_rebalance_listing_with_retry,
-    should_count_rebalance_version_complete, should_defer_rebalance_entry_failure, should_retry_rebalance_listing,
+    send_rebalance_done_signal, should_cleanup_rebalance_source_entry, should_count_rebalance_version_complete,
-    should_skip_rebalance_delete_marker, wait_rebalance_entry_tasks, wait_rebalance_listing_retry, with_rebalance_entry_context,
+    should_defer_rebalance_entry_failure, should_retry_rebalance_listing, should_skip_rebalance_delete_marker,
+    wait_rebalance_entry_tasks, wait_rebalance_listing_retry, with_rebalance_entry_context,
 };
 use super::{
     DiskStat, GetObjectReader, ObjectInfo, ObjectOptions, RebalSaveOpt, RebalStatus, RebalanceBucketConfigs,
     RebalanceBucketOutcome, RebalanceCleanupWarnings, RebalanceEntryOutcome, RebalanceInfo, RebalanceMeta, RebalanceStats,
 };
+use super::{REBALANCE_DEFERRED_ENTRY_ERROR_PREFIX, REBALANCE_SOURCE_CLEANUP_DEFERRED_ERROR_PREFIX};
 use crate::bucket::replication::{ReplicationState, ReplicationStatusType, replication_state_to_filemeta};
 use crate::data_movement;
+use crate::data_movement::SourceCleanupError;
 use crate::data_usage::DATA_USAGE_CACHE_NAME;
 use crate::disk::RUSTFS_META_BUCKET;
 use crate::disk::error::DiskError;
@@ -1665,26 +1668,63 @@ fn test_resolve_rebalance_meta_load_result_wraps_error_context() {
 #[test]
 fn test_resolve_rebalance_entry_cleanup_delete_result_passthrough() {
     let result = resolve_rebalance_entry_cleanup_delete_result(Ok(ObjectInfo::default()), "bucket-a", "obj.txt");
-    assert_eq!(result.expect("successful cleanup should pass through"), None);
+    assert_eq!(result, RebalanceEntryCleanupResult::Completed { warning: None });
 }
 
 #[test]
 fn test_resolve_rebalance_entry_cleanup_delete_result_ignores_not_found() {
     let result = resolve_rebalance_entry_cleanup_delete_result(
-        Err(Error::ObjectNotFound("bucket-a".to_string(), "obj.txt".to_string())),
+        Err(Error::ObjectNotFound("bucket-a".to_string(), "obj.txt".to_string()).into()),
         "bucket-a",
         "obj.txt",
     );
-    assert_eq!(result.expect("missing cleanup source should be ignored"), None);
+    assert_eq!(result, RebalanceEntryCleanupResult::Completed { warning: None });
 }
 
 #[test]
 fn test_resolve_rebalance_entry_cleanup_delete_result_returns_warning_for_failures() {
-    let warning = resolve_rebalance_entry_cleanup_delete_result(Err(Error::SlowDown), "bucket-a", "obj.txt")
+    let result = resolve_rebalance_entry_cleanup_delete_result(Err(Error::SlowDown.into()), "bucket-a", "obj.txt");
-        .expect("cleanup delete failures should be downgraded to warnings")
+    assert!(matches!(
-        .expect("cleanup delete failure should return warning");
+        result,
-    let message = warning.as_str();
+        RebalanceEntryCleanupResult::Completed { warning: Some(ref message) }
-    assert!(message.contains("rebalance cleanup delete failed for bucket-a/obj.txt"));
+            if message.contains("rebalance cleanup delete failed for bucket-a/obj.txt")
+    ));
+}
+
+#[test]
+fn test_resolve_rebalance_entry_cleanup_delete_result_defers_source_change() {
+    let result = resolve_rebalance_entry_cleanup_delete_result(Err(SourceCleanupError::SourceChanged), "bucket-a", "obj.txt");
+    assert!(matches!(
+        result,
+        RebalanceEntryCleanupResult::Deferred { ref last_error }
+            if last_error.starts_with(REBALANCE_SOURCE_CLEANUP_DEFERRED_ERROR_PREFIX)
+                && last_error.contains("source changed during cleanup preflight for bucket-a/obj.txt")
+    ));
+}
+
+#[test]
+fn test_resolve_rebalance_entry_cleanup_delete_result_does_not_defer_other_precondition_failure() {
+    let result = resolve_rebalance_entry_cleanup_delete_result(Err(Error::PreconditionFailed.into()), "bucket-a", "obj.txt");
+    assert!(matches!(
+        result,
+        RebalanceEntryCleanupResult::Completed { warning: Some(ref message) }
+            if message.contains("rebalance cleanup delete failed for bucket-a/obj.txt")
+    ));
+}
+
+#[test]
+fn test_source_cleanup_defer_does_not_fail_repeated_bucket_retry() {
+    let mut deferred_buckets = std::collections::HashSet::new();
+
+    assert!(!should_fail_repeated_rebalance_bucket_defer(&mut deferred_buckets, "bucket-a", true));
+    assert!(!should_fail_repeated_rebalance_bucket_defer(&mut deferred_buckets, "bucket-a", true));
+    assert!(!should_fail_repeated_rebalance_bucket_defer(&mut deferred_buckets, "bucket-b", false));
+    assert!(should_fail_repeated_rebalance_bucket_defer(&mut deferred_buckets, "bucket-b", false));
+
+    let mut source_attempts = std::collections::HashMap::new();
+    assert_eq!(source_cleanup_defer_attempt(&mut source_attempts, "bucket-c"), 1);
+    assert_eq!(source_cleanup_defer_attempt(&mut source_attempts, "bucket-c"), 2);
+    assert_eq!(source_cleanup_defer_attempt(&mut source_attempts, "bucket-c"), 3);
 }
 
 #[test]

crates/ecstore/src/services/rebalance/runtime.rs

@@ -10,12 +10,14 @@ use super::worker::{
     resolve_rebalance_terminal_error, send_rebalance_done_signal,
 };
 use super::{
-    EVENT_REBALANCE_BUCKET, EVENT_REBALANCE_STATE, LOG_COMPONENT_ECSTORE, LOG_SUBSYSTEM_REBALANCE, RebalSaveOpt, RebalStatus,
+    EVENT_REBALANCE_BUCKET, EVENT_REBALANCE_STATE, LOG_COMPONENT_ECSTORE, LOG_SUBSYSTEM_REBALANCE,
+    REBALANCE_LISTING_RETRY_BASE_DELAY, REBALANCE_SOURCE_CLEANUP_DEFERRED_ERROR_PREFIX, RebalSaveOpt, RebalStatus,
     RebalanceBucketOutcome,
 };
 use crate::error::{Error, Result};
 use crate::runtime::sources as runtime_sources;
 use crate::store::ECStore;
+use std::collections::HashMap;
 use std::collections::HashSet;
 use std::sync::Arc;
 use time::OffsetDateTime;
@@ -23,6 +25,20 @@ use tokio::time::{Duration, Instant};
 use tokio_util::sync::CancellationToken;
 use tracing::{debug, error, info, warn};
 
+pub(super) fn should_fail_repeated_rebalance_bucket_defer(
+    deferred_buckets: &mut HashSet<String>,
+    bucket: &str,
+    source_cleanup_deferred: bool,
+) -> bool {
+    !source_cleanup_deferred && !deferred_buckets.insert(bucket.to_string())
+}
+
+pub(super) fn source_cleanup_defer_attempt(deferred_attempts: &mut HashMap<String, usize>, bucket: &str) -> usize {
+    let attempts = deferred_attempts.entry(bucket.to_string()).or_default();
+    *attempts = attempts.saturating_add(1);
+    *attempts
+}
+
 impl ECStore {
     #[tracing::instrument(skip_all)]
     pub async fn start_rebalance(self: &Arc<Self>) -> Result<()> {
@@ -298,6 +314,7 @@ impl ECStore {
         );
         let mut final_result: Result<()> = Ok(());
         let mut deferred_buckets = HashSet::new();
+        let mut source_cleanup_deferred_attempts = HashMap::new();
 
         loop {
             if rx.is_cancelled() {
@@ -375,7 +392,8 @@ impl ECStore {
                 };
 
                 if let RebalanceBucketOutcome::Deferred { last_error } = outcome {
-                    if !deferred_buckets.insert(bucket.clone()) {
+                    let source_cleanup_deferred = last_error.starts_with(REBALANCE_SOURCE_CLEANUP_DEFERRED_ERROR_PREFIX);
+                    if should_fail_repeated_rebalance_bucket_defer(&mut deferred_buckets, &bucket, source_cleanup_deferred) {
                         let err = Error::other(format!(
                             "rebalance bucket {bucket} deferred repeatedly due to transient object failures: {last_error}"
                         ));
@@ -396,6 +414,11 @@ impl ECStore {
                         break;
                     }
 
+                    let source_cleanup_attempt = if source_cleanup_deferred {
+                        source_cleanup_defer_attempt(&mut source_cleanup_deferred_attempts, &bucket)
+                    } else {
+                        0
+                    };
                     warn!(
                         event = EVENT_REBALANCE_BUCKET,
                         component = LOG_COMPONENT_ECSTORE,
@@ -406,7 +429,10 @@ impl ECStore {
                         error = %last_error,
                         "Deferred rebalance bucket after transient object failures"
                     );
-                    if let Err(err) = self.defer_rebalance_bucket(pool_index, bucket.clone(), last_error).await {
+                    if let Err(err) = self
+                        .defer_rebalance_bucket(pool_index, bucket.clone(), last_error.clone())
+                        .await
+                    {
                         error!(
                             event = EVENT_REBALANCE_BUCKET,
                             component = LOG_COMPONENT_ECSTORE,
@@ -423,6 +449,38 @@ impl ECStore {
                         ));
                         break;
                     }
+                    if source_cleanup_deferred {
+                        if source_cleanup_attempt >= super::REBALANCE_SOURCE_CLEANUP_MAX_DEFERS {
+                            let err = Error::other(format!(
+                                "rebalance bucket {bucket} source cleanup remained unstable after {} deferrals: {last_error}",
+                                super::REBALANCE_SOURCE_CLEANUP_MAX_DEFERS
+                            ));
+                            warn!(
+                                event = EVENT_REBALANCE_BUCKET,
+                                component = LOG_COMPONENT_ECSTORE,
+                                subsystem = LOG_SUBSYSTEM_REBALANCE,
+                                pool_index,
+                                bucket = %bucket,
+                                state = "source_cleanup_defer_limit",
+                                error = ?err,
+                                "Rebalance bucket failed after repeated source cleanup conflicts"
+                            );
+                            final_result = Err(resolve_rebalance_terminal_error(
+                                err.clone(),
+                                send_rebalance_done_signal(&done_tx, Err(err.clone()), pool_index).await,
+                            ));
+                            break;
+                        }
+                        if let Err(err) =
+                            super::worker::wait_rebalance_listing_retry(&rx, REBALANCE_LISTING_RETRY_BASE_DELAY).await
+                        {
+                            final_result = Err(resolve_rebalance_terminal_error(
+                                err.clone(),
+                                send_rebalance_done_signal(&done_tx, Err(err.clone()), pool_index).await,
+                            ));
+                            break;
+                        }
+                    }
                     continue;
                 }
 
@@ -435,6 +493,7 @@ impl ECStore {
                     state = "completed",
                     "Completed rebalance bucket"
                 );
+                source_cleanup_deferred_attempts.remove(&bucket);
                 if let Err(err) = self.bucket_rebalance_done(pool_index, bucket).await {
                     error!(
                         event = EVENT_REBALANCE_BUCKET,

crates/ecstore/src/services/rebalance/worker.rs

@@ -2,10 +2,12 @@ use super::migration::MigrationVersionResult;
 use super::{
     DEFAULT_REBALANCE_MAX_ATTEMPTS, EVENT_REBALANCE_LISTING, LOG_COMPONENT_ECSTORE, LOG_SUBSYSTEM_REBALANCE, REBAL_META_NAME,
     REBALANCE_LISTING_RETRY_BASE_DELAY, REBALANCE_MAX_ATTEMPTS_ENV, REBALANCE_MIGRATION_LOCK_RETRY_CAP,
-    REBALANCE_MIGRATION_RETRY_BASE_DELAY, RebalanceBucketConfigs, RebalanceBucketOutcome, RebalanceEntryOutcome, Result,
+    REBALANCE_MIGRATION_RETRY_BASE_DELAY, REBALANCE_SOURCE_CLEANUP_DEFERRED_ERROR_PREFIX, RebalanceBucketConfigs,
+    RebalanceBucketOutcome, RebalanceEntryOutcome, Result,
 };
 use crate::cache_value::metacache_set::{ListPathRawOptions, list_path_raw};
 use crate::core::pools::ListCallback;
+use crate::data_movement::SourceCleanupError;
 use crate::disk::error::DiskError;
 use crate::error::{
     Error, is_err_object_not_found, is_err_operation_canceled, is_err_version_not_found, is_network_or_host_down,
@@ -36,6 +38,12 @@ pub(super) fn resolve_rebalance_worker_result<T>(
 
 pub(super) type RebalanceEntryTask = tokio::task::JoinHandle<Result<RebalanceEntryOutcome>>;
 
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub(super) enum RebalanceEntryCleanupResult {
+    Completed { warning: Option<String> },
+    Deferred { last_error: String },
+}
+
 pub(super) async fn wait_rebalance_entry_tasks(
     set_idx: usize,
     tasks: Arc<tokio::sync::Mutex<Vec<RebalanceEntryTask>>>,
@@ -145,14 +153,23 @@ where
 }
 
 pub(super) fn resolve_rebalance_entry_cleanup_delete_result(
-    result: Result<crate::object_api::ObjectInfo>,
+    result: std::result::Result<crate::object_api::ObjectInfo, SourceCleanupError>,
     bucket: &str,
     object_name: &str,
-) -> Result<Option<String>> {
+) -> RebalanceEntryCleanupResult {
     match result {
-        Ok(_) => Ok(None),
+        Ok(_) => RebalanceEntryCleanupResult::Completed { warning: None },
-        Err(err) if is_err_object_not_found(&err) || is_err_version_not_found(&err) => Ok(None),
+        Err(SourceCleanupError::Storage(err)) if is_err_object_not_found(&err) || is_err_version_not_found(&err) => {
-        Err(err) => Ok(Some(format!("rebalance cleanup delete failed for {bucket}/{object_name}: {err}"))),
+            RebalanceEntryCleanupResult::Completed { warning: None }
+        }
+        Err(SourceCleanupError::SourceChanged) => RebalanceEntryCleanupResult::Deferred {
+            last_error: format!(
+                "{REBALANCE_SOURCE_CLEANUP_DEFERRED_ERROR_PREFIX} source changed during cleanup preflight for {bucket}/{object_name}"
+            ),
+        },
+        Err(SourceCleanupError::Storage(err)) => RebalanceEntryCleanupResult::Completed {
+            warning: Some(format!("rebalance cleanup delete failed for {bucket}/{object_name}: {err}")),
+        },
     }
 }
 

crates/ecstore/src/set_disk/ops/object.rs

@@ -7626,7 +7626,10 @@ mod transition_upload_integrity_tests {
             .await
             .expect("cleanup task should not panic")
             .expect_err("cleanup must fail after its outer namespace lock loses refresh quorum");
-        assert!(matches!(error, StorageError::NamespaceLockQuorumUnavailable { .. }));
+        assert!(matches!(
+            error,
+            crate::data_movement::SourceCleanupError::Storage(StorageError::NamespaceLockQuorumUnavailable { .. })
+        ));
         assert_local_source_intact(&set_disks, bucket, object, &payload).await;
     }