diff --git a/crates/ecstore/src/services/rebalance/control.rs b/crates/ecstore/src/services/rebalance/control.rs index 2a35d732b..75c189788 100644 --- a/crates/ecstore/src/services/rebalance/control.rs +++ b/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(()) } diff --git a/crates/ecstore/src/services/rebalance/entry.rs b/crates/ecstore/src/services/rebalance/entry.rs index a58d76edc..6a9987707 100644 --- a/crates/ecstore/src/services/rebalance/entry.rs +++ b/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] diff --git a/crates/ecstore/src/services/rebalance/mod.rs b/crates/ecstore/src/services/rebalance/mod.rs index 1fb063237..1c32c07b3 100644 --- a/crates/ecstore/src/services/rebalance/mod.rs +++ b/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( diff --git a/crates/ecstore/src/services/rebalance/rebalance_unit_tests.rs b/crates/ecstore/src/services/rebalance/rebalance_unit_tests.rs index ed7c06a5f..9add0a0b8 100644 --- a/crates/ecstore/src/services/rebalance/rebalance_unit_tests.rs +++ b/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] diff --git a/crates/ecstore/src/services/rebalance/runtime.rs b/crates/ecstore/src/services/rebalance/runtime.rs index f3f730d76..2795c4a00 100644 --- a/crates/ecstore/src/services/rebalance/runtime.rs +++ b/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 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, diff --git a/crates/ecstore/src/services/rebalance/types.rs b/crates/ecstore/src/services/rebalance/types.rs index f3bcdbab3..40b5163c4 100644 --- a/crates/ecstore/src/services/rebalance/types.rs +++ b/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] diff --git a/crates/ecstore/src/services/rebalance/worker.rs b/crates/ecstore/src/services/rebalance/worker.rs index c61a78339..7b6dc85f0 100644 --- a/crates/ecstore/src/services/rebalance/worker.rs +++ b/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, 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 { + 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, + ); + } + } }