fix(rebalance): defer retryable source cleanup failures (#7977)

Source cleanup treated a retryable conflict such as a lock acquisition
timeout as a permanent cleanup warning. The warning was persisted, mapped
the pool to failed_ignored, and blocked pool completion for the rest of
the run while the source replica stayed on disk. Retryable storage
failures now defer the bucket instead, and only non-retryable failures
keep a permanent warning, so completion still requires a resolved delete
or a successful not-found check.

Recording a bucket deferral no longer overwrites the pool's pending
migration-deferral marker, because that marker is the only signal keeping
a pool from completing at the free-space goal while an entry is retried.

Teach the transient matchers the rendered lock timeout text, unwrap
data-movement stage context before classifying not-found and transient
cleanup failures, and raise the bounded source cleanup deferral budget.

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

@@ -9,11 +9,12 @@ use super::meta::{
 };
 use super::worker::{
     rebalance_max_attempts, rebalance_meta_lock_error, resolve_load_rebalance_stats_update_result,
-    resolve_rebalance_meta_load_result, resolve_rebalance_meta_save_result, retry_rebalance_metadata_access,
+    resolve_rebalance_deferred_last_error, resolve_rebalance_meta_load_result, resolve_rebalance_meta_save_result,
+    retry_rebalance_metadata_access,
 };
 use super::{
     DiskStat, EVENT_REBALANCE_BUCKET, EVENT_REBALANCE_STATE, LOG_COMPONENT_ECSTORE, LOG_SUBSYSTEM_REBALANCE, REBAL_META_NAME,
-    RebalStatus, RebalanceInfo, RebalanceMeta, RebalanceStats, RebalanceStopPropagationRecord,
+    RebalStatus, RebalanceDeferKind, RebalanceInfo, RebalanceMeta, RebalanceStats, RebalanceStopPropagationRecord,
     encode_rebalance_stop_propagation_record,
 };
 use crate::core::pools::{
@@ -1103,6 +1104,7 @@ impl ECStore {
         bucket: String,
         last_error: String,
         expected_id: &str,
+        kind: RebalanceDeferKind,
     ) -> Result<()> {
         let mut rebalance_meta = self.rebalance_meta.write().await;
         ensure_rebalance_worker_active(rebalance_meta.as_ref(), expected_id, "defer rebalance bucket")?;
@@ -1116,7 +1118,8 @@ impl ECStore {
         };
 
         defer_bucket_in_rebalance_queue(pool_stat, &bucket)?;
-        pool_stat.info.last_error = Some(last_error);
+        let pending_entry_defer = pool_stat.info.last_error.clone();
+        pool_stat.info.last_error = resolve_rebalance_deferred_last_error(kind, pending_entry_defer.as_deref(), &last_error);
         meta.last_refreshed_at = Some(OffsetDateTime::now_utc());
         Ok(())
     }

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

@@ -1209,7 +1209,7 @@ mod tests {
             cancel: &warning_cancel,
         };
         let warning_result = store
-            .finish_rebalance_entry_after_cleanup(&warning_cleanup_context, async { Err(Error::SlowDown.into()) })
+            .finish_rebalance_entry_after_cleanup(&warning_cleanup_context, async { Err(Error::FileAccessDenied.into()) })
             .await
             .expect("cleanup warnings should not fail the completed migration");
         assert!(matches!(warning_result, RebalanceEntryCleanupResult::Completed { warning: Some(_) }));
@@ -1250,6 +1250,35 @@ mod tests {
         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");
+        drop(meta);
+
+        let transient_guard = store
+            .rebalance_run_guard(rebalance_id, "rebalance transient cleanup deferral test")
+            .await
+            .expect("rebalance transient cleanup deferral test guard should be acquired");
+        let transient_cancel = CancellationToken::new();
+        let transient_stats_updates = [&warning_version];
+        let transient_cleanup_context = RebalanceEntryCleanupContext {
+            run_guard: &transient_guard,
+            pool_index: 0,
+            bucket: "bucket",
+            object: "object.bin",
+            stats_updates: &transient_stats_updates,
+            expected_id: rebalance_id,
+            cancel: &transient_cancel,
+        };
+        let transient = store
+            .finish_rebalance_entry_after_cleanup(&transient_cleanup_context, async { Err(Error::SlowDown.into()) })
+            .await
+            .expect("retryable cleanup failures should defer without failing the worker");
+        assert!(matches!(transient, 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, "retryable cleanup failures must not commit completion stats");
+        assert_eq!(
+            pool_stats.cleanup_warnings.count, 1,
+            "retryable cleanup failures must not add a permanent warning"
+        );
     }
 
     #[tokio::test]

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

@@ -29,7 +29,7 @@ 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;
+pub(crate) const REBALANCE_SOURCE_CLEANUP_MAX_DEFERS: usize = 8;
 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);
@@ -55,7 +55,7 @@ pub use types::{
     DiskStat, RebalSaveOpt, RebalStatus, RebalanceCleanupWarningEntry, RebalanceCleanupWarnings, RebalanceInfo, RebalanceMeta,
     RebalanceStats, RebalanceStopPropagationRecord,
 };
-use types::{RebalanceBucketConfigs, RebalanceBucketOutcome, RebalanceEntryOutcome};
+use types::{RebalanceBucketConfigs, RebalanceBucketOutcome, RebalanceDeferKind, RebalanceEntryOutcome};
 
 #[cfg(any(test, feature = "test-util"))]
 pub async fn test_store_with_persisted_rebalance_meta(

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

@@ -34,27 +34,29 @@ use super::migration::{
 };
 use super::runtime::{
     RebalanceLocalActivationOutcome, commit_local_rebalance_worker_activation,
-    commit_local_rebalance_worker_activation_candidate, should_fail_repeated_rebalance_bucket_defer,
-    source_cleanup_defer_attempt, stage_local_rebalance_worker_activation,
+    commit_local_rebalance_worker_activation_candidate, reached_rebalance_source_cleanup_defer_limit,
+    should_fail_repeated_rebalance_bucket_defer, source_cleanup_defer_attempt, stage_local_rebalance_worker_activation,
 };
 use super::worker::{
     RebalanceEntryCleanupResult, ensure_rebalance_listing_disks_available, is_transient_rebalance_error,
     parse_rebalance_max_attempts, rebalance_listing_retry_delay, rebalance_migration_retry_delay,
     resolve_load_rebalance_stats_update_result, resolve_rebalance_bucket_error, resolve_rebalance_bucket_result,
-    resolve_rebalance_entry_cleanup_delete_result, resolve_rebalance_file_info_versions_result,
-    resolve_rebalance_meta_load_result, resolve_rebalance_meta_save_result, resolve_rebalance_migrate_result_error,
-    resolve_rebalance_optional_bucket_config_result, resolve_rebalance_save_task_result, resolve_rebalance_stats_update_result,
-    resolve_rebalance_terminal_error, resolve_rebalance_worker_result, run_rebalance_listing_with_retry,
-    send_rebalance_done_signal, should_cleanup_rebalance_source_entry, should_count_rebalance_version_complete,
-    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,
+    resolve_rebalance_deferred_last_error, resolve_rebalance_entry_cleanup_delete_result,
+    resolve_rebalance_file_info_versions_result, resolve_rebalance_meta_load_result, resolve_rebalance_meta_save_result,
+    resolve_rebalance_migrate_result_error, resolve_rebalance_optional_bucket_config_result, resolve_rebalance_save_task_result,
+    resolve_rebalance_stats_update_result, resolve_rebalance_terminal_error, resolve_rebalance_worker_result,
+    run_rebalance_listing_with_retry, send_rebalance_done_signal, should_cleanup_rebalance_source_entry,
+    should_count_rebalance_version_complete, 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,
-    RebalanceStopPropagationRecord,
+    RebalanceBucketOutcome, RebalanceCleanupWarnings, RebalanceDeferKind, RebalanceEntryOutcome, RebalanceInfo, RebalanceMeta,
+    RebalanceStats, RebalanceStopPropagationRecord,
+};
+use super::{
+    REBALANCE_DEFERRED_ENTRY_ERROR_PREFIX, REBALANCE_SOURCE_CLEANUP_DEFERRED_ERROR_PREFIX, REBALANCE_SOURCE_CLEANUP_MAX_DEFERS,
 };
-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;
@@ -1775,7 +1777,7 @@ fn test_resolve_rebalance_entry_cleanup_delete_result_ignores_not_found() {
 
 #[test]
 fn test_resolve_rebalance_entry_cleanup_delete_result_returns_warning_for_failures() {
-    let result = resolve_rebalance_entry_cleanup_delete_result(Err(Error::SlowDown.into()), "bucket-a", "obj.txt");
+    let result = resolve_rebalance_entry_cleanup_delete_result(Err(Error::FileAccessDenied.into()), "bucket-a", "obj.txt");
     assert!(matches!(
         result,
         RebalanceEntryCleanupResult::Completed { warning: Some(ref message) }
@@ -1783,6 +1785,122 @@ fn test_resolve_rebalance_entry_cleanup_delete_result_returns_warning_for_failur
     ));
 }
 
+#[test]
+fn test_resolve_rebalance_entry_cleanup_delete_result_defers_transient_failures() {
+    let cases = [
+        (Error::SlowDown, "slow down"),
+        (
+            Error::Lock(rustfs_lock::LockError::timeout("bucket-a/obj.txt@latest", Duration::from_secs(5))),
+            "object lock timeout",
+        ),
+        (
+            Error::Lock(rustfs_lock::LockError::network(
+                "peer unavailable",
+                std::io::Error::from(std::io::ErrorKind::ConnectionReset),
+            )),
+            "object lock network failure",
+        ),
+        (Error::ErasureWriteQuorum, "write quorum"),
+        (Error::Io(std::io::Error::from(std::io::ErrorKind::TimedOut)), "io timeout"),
+        (
+            Error::other("Lock error: Lock acquisition timeout for resource 'bucket-a/obj.txt@latest' after 5s"),
+            "rendered lock timeout text",
+        ),
+    ];
+
+    for (err, label) in cases {
+        match resolve_rebalance_entry_cleanup_delete_result(Err(err.into()), "bucket-a", "obj.txt") {
+            RebalanceEntryCleanupResult::Deferred { last_error } => {
+                assert!(
+                    last_error.starts_with(REBALANCE_SOURCE_CLEANUP_DEFERRED_ERROR_PREFIX),
+                    "{label}: {last_error}"
+                );
+                assert!(last_error.contains("bucket-a/obj.txt"), "{label}: {last_error}");
+            }
+            RebalanceEntryCleanupResult::Completed { warning } => {
+                panic!("{label} must defer source cleanup instead of completing the entry with warning {warning:?}")
+            }
+        }
+    }
+}
+
+#[test]
+fn test_resolve_rebalance_entry_cleanup_delete_result_defers_stage_wrapped_lock_timeout() {
+    let err = data_movement::data_movement_stage_error_for_test(
+        "rebalance",
+        "delete_object",
+        "bucket-a",
+        "obj.txt",
+        Error::Lock(rustfs_lock::LockError::timeout("bucket-a/obj.txt@latest", Duration::from_secs(5))),
+    );
+
+    assert!(matches!(
+        resolve_rebalance_entry_cleanup_delete_result(Err(err.into()), "bucket-a", "obj.txt"),
+        RebalanceEntryCleanupResult::Deferred { .. }
+    ));
+}
+
+#[test]
+fn test_resolve_rebalance_entry_cleanup_delete_result_ignores_stage_wrapped_not_found() {
+    let err = data_movement::data_movement_stage_error_for_test(
+        "rebalance",
+        "delete_object",
+        "bucket-a",
+        "obj.txt",
+        Error::ObjectNotFound("bucket-a".to_string(), "obj.txt".to_string()),
+    );
+
+    assert_eq!(
+        resolve_rebalance_entry_cleanup_delete_result(Err(err.into()), "bucket-a", "obj.txt"),
+        RebalanceEntryCleanupResult::Completed { warning: None }
+    );
+}
+
+#[test]
+fn test_resolve_rebalance_deferred_last_error_hides_retryable_cleanup_conflicts() {
+    let entry_error = format!("{REBALANCE_DEFERRED_ENTRY_ERROR_PREFIX} timeout");
+    assert_eq!(
+        resolve_rebalance_deferred_last_error(RebalanceDeferKind::Entry, None, entry_error.as_str()),
+        Some(entry_error.clone()),
+        "transient migration deferrals must stay visible to the completion guards"
+    );
+
+    let cleanup_error = format!("{REBALANCE_SOURCE_CLEANUP_DEFERRED_ERROR_PREFIX} lock acquisition timeout");
+    assert_eq!(
+        resolve_rebalance_deferred_last_error(RebalanceDeferKind::SourceCleanup, None, cleanup_error.as_str()),
+        None,
+        "a retryable source cleanup conflict is progress, not a pool failure"
+    );
+
+    assert_eq!(
+        resolve_rebalance_deferred_last_error(
+            RebalanceDeferKind::SourceCleanup,
+            Some(entry_error.as_str()),
+            cleanup_error.as_str()
+        ),
+        Some(entry_error.clone()),
+        "a cleanup deferral for one bucket must not erase an unresolved migration deferral of the same pool"
+    );
+    assert_eq!(
+        resolve_rebalance_deferred_last_error(
+            RebalanceDeferKind::SourceCleanup,
+            Some(cleanup_error.as_str()),
+            cleanup_error.as_str()
+        ),
+        None,
+        "a stale retryable cleanup message must not survive as a pool failure"
+    );
+    assert_eq!(
+        resolve_rebalance_deferred_last_error(
+            RebalanceDeferKind::SourceCleanup,
+            Some(entry_error.as_str()),
+            entry_error.as_str()
+        ),
+        Some(entry_error),
+        "repeated cleanup deferrals must keep the pending migration deferral visible"
+    );
+}
+
 #[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");
@@ -1817,6 +1935,89 @@ fn test_source_cleanup_defer_does_not_fail_repeated_bucket_retry() {
     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);
+
+    let mut bounded_attempts = std::collections::HashMap::new();
+    for expected in 1..REBALANCE_SOURCE_CLEANUP_MAX_DEFERS {
+        assert_eq!(source_cleanup_defer_attempt(&mut bounded_attempts, "bucket-d"), expected);
+        assert!(
+            !reached_rebalance_source_cleanup_defer_limit(expected),
+            "a retryable cleanup conflict must stay retryable at deferral {expected}"
+        );
+    }
+    assert_eq!(
+        source_cleanup_defer_attempt(&mut bounded_attempts, "bucket-d"),
+        REBALANCE_SOURCE_CLEANUP_MAX_DEFERS
+    );
+    assert!(
+        reached_rebalance_source_cleanup_defer_limit(REBALANCE_SOURCE_CLEANUP_MAX_DEFERS),
+        "an unreclaimable source replica must fail the bucket after the bounded deferral budget"
+    );
+}
+
+#[tokio::test]
+async fn test_defer_rebalance_bucket_keeps_pending_entry_defer_without_surfacing_cleanup_conflicts() {
+    let id = "rebalance-defer-last-error";
+    let meta = RebalanceMeta {
+        id: id.to_string(),
+        pool_stats: vec![RebalanceStats {
+            participating: true,
+            buckets: vec!["bucket-a".to_string(), "bucket-b".to_string()],
+            info: RebalanceInfo {
+                status: RebalStatus::Started,
+                ..Default::default()
+            },
+            ..Default::default()
+        }],
+        ..Default::default()
+    };
+    let store = test_store_with_rebalance_meta(meta);
+    let cleanup_error = format!("{REBALANCE_SOURCE_CLEANUP_DEFERRED_ERROR_PREFIX} lock acquisition timeout");
+
+    store
+        .defer_rebalance_bucket(0, "bucket-a".to_string(), cleanup_error.clone(), id, RebalanceDeferKind::SourceCleanup)
+        .await
+        .expect("a retryable cleanup conflict must be deferrable");
+    {
+        let meta = store.rebalance_meta.read().await;
+        let pool_stat = &meta.as_ref().expect("rebalance metadata should exist").pool_stats[0];
+        assert_eq!(
+            pool_stat.info.last_error, None,
+            "a retryable cleanup conflict is progress and must not surface as a pool failure"
+        );
+        assert_eq!(pool_stat.buckets, vec!["bucket-b".to_string(), "bucket-a".to_string()]);
+    }
+
+    let entry_error = format!("{REBALANCE_DEFERRED_ENTRY_ERROR_PREFIX} slow down");
+    {
+        let mut meta = store.rebalance_meta.write().await;
+        meta.as_mut().expect("rebalance metadata should exist").pool_stats[0]
+            .info
+            .last_error = Some(entry_error.clone());
+    }
+    store
+        .defer_rebalance_bucket(0, "bucket-b".to_string(), cleanup_error, id, RebalanceDeferKind::SourceCleanup)
+        .await
+        .expect("a cleanup deferral must be accepted while another bucket still defers an entry");
+    {
+        let meta = store.rebalance_meta.read().await;
+        let pool_stat = &meta.as_ref().expect("rebalance metadata should exist").pool_stats[0];
+        assert_eq!(
+            pool_stat.info.last_error,
+            Some(entry_error),
+            "a cleanup deferral must not erase the pending migration deferral that blocks goal completion"
+        );
+        assert!(has_deferred_rebalance_error(pool_stat));
+        assert_eq!(pool_stat.buckets, vec!["bucket-a".to_string(), "bucket-b".to_string()]);
+    }
+
+    let migration_error = format!("{REBALANCE_DEFERRED_ENTRY_ERROR_PREFIX} i/o timeout");
+    store
+        .defer_rebalance_bucket(0, "bucket-a".to_string(), migration_error.clone(), id, RebalanceDeferKind::Entry)
+        .await
+        .expect("migration deferrals must keep their last error");
+    let meta = store.rebalance_meta.read().await;
+    let pool_stat = &meta.as_ref().expect("rebalance metadata should exist").pool_stats[0];
+    assert_eq!(pool_stat.info.last_error, Some(migration_error));
 }
 
 #[test]

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

@@ -12,8 +12,8 @@ use super::worker::{
 };
 use super::{
     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,
+    REBALANCE_LISTING_RETRY_BASE_DELAY, REBALANCE_SOURCE_CLEANUP_DEFERRED_ERROR_PREFIX, REBALANCE_SOURCE_CLEANUP_MAX_DEFERS,
+    RebalSaveOpt, RebalStatus, RebalanceBucketOutcome, RebalanceDeferKind,
 };
 use crate::error::{Error, Result};
 use crate::runtime::sources as runtime_sources;
@@ -40,6 +40,12 @@ pub(super) fn source_cleanup_defer_attempt(deferred_attempts: &mut HashMap<Strin
     *attempts
 }
 
+/// Retryable source cleanup conflicts are bounded per run, so a source replica that stays
+/// unreclaimable fails the bucket explicitly instead of deferring forever.
+pub(super) fn reached_rebalance_source_cleanup_defer_limit(attempt: usize) -> bool {
+    attempt >= REBALANCE_SOURCE_CLEANUP_MAX_DEFERS
+}
+
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
 pub(super) enum RebalanceLocalActivationOutcome {
     Started,
@@ -623,6 +629,11 @@ impl ECStore {
                     } else {
                         0
                     };
+                    let defer_kind = if source_cleanup_deferred {
+                        RebalanceDeferKind::SourceCleanup
+                    } else {
+                        RebalanceDeferKind::Entry
+                    };
                     warn!(
                         event = EVENT_REBALANCE_BUCKET,
                         component = LOG_COMPONENT_ECSTORE,
@@ -634,7 +645,7 @@ impl ECStore {
                         "Deferred rebalance bucket after transient object failures"
                     );
                     if let Err(err) = self
-                        .defer_rebalance_bucket(pool_index, bucket.clone(), last_error.clone(), rebalance_id.as_ref())
+                        .defer_rebalance_bucket(pool_index, bucket.clone(), last_error.clone(), rebalance_id.as_ref(), defer_kind)
                         .await
                     {
                         error!(
@@ -654,10 +665,10 @@ impl ECStore {
                         break;
                     }
                     if source_cleanup_deferred {
-                        if source_cleanup_attempt >= super::REBALANCE_SOURCE_CLEANUP_MAX_DEFERS {
+                        if reached_rebalance_source_cleanup_defer_limit(source_cleanup_attempt) {
                             let err = Error::other(format!(
                                 "rebalance bucket {bucket} source cleanup remained unstable after {} deferrals: {last_error}",
-                                super::REBALANCE_SOURCE_CLEANUP_MAX_DEFERS
+                                REBALANCE_SOURCE_CLEANUP_MAX_DEFERS
                             ));
                             warn!(
                                 event = EVENT_REBALANCE_BUCKET,

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

@@ -51,6 +51,18 @@ pub(super) enum RebalanceEntryOutcome {
     Deferred { last_error: String },
 }
 
+/// Why a rebalance bucket was put back at the end of the queue.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(super) enum RebalanceDeferKind {
+    /// A transient object-migration failure. Persisting it as `lastError` keeps the pool
+    /// from being completed at the free-space goal while the entry is still retried.
+    Entry,
+    /// A retryable source-cleanup conflict. The bucket stays queued and is retried, so a
+    /// transient lock conflict must not be recorded as a permanent warning that would block
+    /// pool completion for the rest of the run.
+    SourceCleanup,
+}
+
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
 pub enum RebalStatus {
     #[default]

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

@@ -1,9 +1,9 @@
 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, REBALANCE_SOURCE_CLEANUP_DEFERRED_ERROR_PREFIX, RebalanceBucketConfigs,
-    RebalanceBucketOutcome, RebalanceEntryOutcome, Result,
+    REBALANCE_DEFERRED_ENTRY_ERROR_PREFIX, REBALANCE_LISTING_RETRY_BASE_DELAY, REBALANCE_MAX_ATTEMPTS_ENV,
+    REBALANCE_MIGRATION_LOCK_RETRY_CAP, REBALANCE_MIGRATION_RETRY_BASE_DELAY, REBALANCE_SOURCE_CLEANUP_DEFERRED_ERROR_PREFIX,
+    RebalanceBucketConfigs, RebalanceBucketOutcome, RebalanceDeferKind, RebalanceEntryOutcome, Result,
 };
 use crate::cache_value::metacache_set::{ListPathRawOptions, list_path_raw};
 use crate::core::pools::ListCallback;
@@ -177,7 +177,7 @@ pub(super) fn resolve_rebalance_entry_cleanup_delete_result(
 ) -> RebalanceEntryCleanupResult {
     match result {
         Ok(_) => RebalanceEntryCleanupResult::Completed { warning: None },
-        Err(SourceCleanupError::Storage(err)) if is_err_object_not_found(&err) || is_err_version_not_found(&err) => {
+        Err(SourceCleanupError::Storage(err)) if is_source_cleanup_not_found(&err) => {
             RebalanceEntryCleanupResult::Completed { warning: None }
         }
         Err(SourceCleanupError::SourceChanged) => RebalanceEntryCleanupResult::Deferred {
@@ -185,12 +185,25 @@ pub(super) fn resolve_rebalance_entry_cleanup_delete_result(
                 "{REBALANCE_SOURCE_CLEANUP_DEFERRED_ERROR_PREFIX} source changed during cleanup preflight for {bucket}/{object_name}"
             ),
         },
+        // A transient cleanup failure is not evidence that the source replica is gone, so the
+        // entry stays incomplete and the bucket is retried instead of recording a permanent
+        // cleanup warning that would block pool completion.
+        Err(SourceCleanupError::Storage(err)) if is_transient_rebalance_error(&err) => RebalanceEntryCleanupResult::Deferred {
+            last_error: format!(
+                "{REBALANCE_SOURCE_CLEANUP_DEFERRED_ERROR_PREFIX} transient source cleanup failure for {bucket}/{object_name} will be retried: {err}"
+            ),
+        },
         Err(SourceCleanupError::Storage(err)) => RebalanceEntryCleanupResult::Completed {
             warning: Some(format!("rebalance cleanup delete failed for {bucket}/{object_name}: {err}")),
         },
     }
 }
 
+fn is_source_cleanup_not_found(err: &Error) -> bool {
+    let err = rebalance_error_source(err);
+    is_err_object_not_found(err) || is_err_version_not_found(err)
+}
+
 pub(super) fn resolve_rebalance_migrate_result_error(
     err: Option<Error>,
     pool_idx: usize,
@@ -210,6 +223,23 @@ pub(super) fn should_defer_rebalance_entry_failure(err: &Error) -> bool {
     is_transient_rebalance_error(err)
 }
 
+pub(super) fn resolve_rebalance_deferred_last_error(
+    kind: RebalanceDeferKind,
+    pending_entry_defer: Option<&str>,
+    last_error: &str,
+) -> Option<String> {
+    match kind {
+        RebalanceDeferKind::Entry => Some(last_error.to_string()),
+        // A retryable cleanup conflict is progress, not a pool failure, so it must not surface as
+        // `lastError`. It also must not erase an unresolved migration deferral: that marker is the
+        // only signal keeping the pool from completing at the free-space goal while an entry is
+        // still retried, and the two deferrals can be reported by different buckets of one pool.
+        RebalanceDeferKind::SourceCleanup => pending_entry_defer
+            .filter(|pending| pending.starts_with(REBALANCE_DEFERRED_ENTRY_ERROR_PREFIX))
+            .map(str::to_string),
+    }
+}
+
 pub(super) fn resolve_load_rebalance_stats_update_result(result: Result<()>) -> Result<()> {
     result.map_err(|err| Error::other(format!("rebalance metadata stats refresh failed after load: {err}")))
 }
@@ -299,7 +329,10 @@ fn is_rebalance_transient_io_error(err: &std::io::Error) -> bool {
 
 fn is_rebalance_transient_message(message: &str) -> bool {
     let message = message.to_ascii_lowercase();
-    message.contains("lock acquisition timed out")
+    // `LockError::Timeout` renders "Lock acquisition timeout for resource ...", while the
+    // namespace-lock layer renders "lock acquisition timed out on ..."; both are retryable.
+    message.contains("lock acquisition timeout")
+        || message.contains("lock acquisition timed out")
         || message.contains("remote lock rpc timed out")
         || message.contains("keepalivetimedout")
         || message.contains("i/o timeout")
@@ -380,7 +413,8 @@ fn is_rebalance_lock_or_rpc_timeout(err: &Error) -> bool {
 
 fn is_rebalance_lock_or_rpc_timeout_message(message: &str) -> bool {
     let message = message.to_ascii_lowercase();
-    message.contains("lock acquisition timed out")
+    message.contains("lock acquisition timeout")
+        || message.contains("lock acquisition timed out")
         || message.contains("remote lock rpc timed out")
         || message.contains("keepalivetimedout")
 }
@@ -807,4 +841,44 @@ mod error_source_tests {
             }
         }
     }
+    #[test]
+    fn rendered_lock_timeout_text_selects_the_lock_backoff() {
+        // The lock backend renders a timeout as "Lock acquisition timeout for resource ...",
+        // so the message matcher must recognize that text when the error arrives re-rendered
+        // instead of as a typed `Error::Lock`.
+        let rendered = rustfs_lock::LockError::timeout("bucket/object@latest", Duration::from_secs(5)).to_string();
+        assert!(
+            rendered.contains("Lock acquisition timeout for resource"),
+            "unexpected lock timeout text: {rendered}"
+        );
+
+        // The lock policy jitters the delay inside its own cap, so a far-out attempt identifies
+        // the selected policy: the linear fallback would return `base * (attempt + 1)`.
+        let far_attempt = 100;
+        assert!(REBALANCE_MIGRATION_RETRY_BASE_DELAY * 101 > REBALANCE_MIGRATION_LOCK_RETRY_CAP);
+
+        let mut error = Error::other(format!("Lock error: {rendered}"));
+        for depth in 0..=3 {
+            assert!(
+                is_transient_rebalance_error(&error),
+                "rendered lock timeout lost retryability at depth {depth}: {error:?}"
+            );
+            assert!(
+                is_rebalance_lock_or_rpc_timeout(&error),
+                "rendered lock timeout lost the lock backoff at depth {depth}: {error:?}"
+            );
+            let delay = rebalance_migration_retry_delay(far_attempt, &error);
+            assert!(
+                delay <= REBALANCE_MIGRATION_LOCK_RETRY_CAP && delay >= Duration::from_millis(1),
+                "rendered lock timeout must stay inside the lock backoff cap at depth {depth}: {delay:?}"
+            );
+            error = crate::data_movement::data_movement_stage_error_for_test(
+                "rebalance_object",
+                "put_object",
+                "bucket",
+                "baseline/00042.bin",
+                error,
+            );
+        }
+    }
 }