diff --git a/crates/ecstore/src/core/pools.rs b/crates/ecstore/src/core/pools.rs index aef124c46..4d19b92ec 100644 --- a/crates/ecstore/src/core/pools.rs +++ b/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!( diff --git a/crates/ecstore/src/data_movement/mod.rs b/crates/ecstore/src/data_movement/mod.rs index 503483bd6..67bbfc50f 100644 --- a/crates/ecstore/src/data_movement/mod.rs +++ b/crates/ecstore/src/data_movement/mod.rs @@ -437,6 +437,15 @@ struct SourceCleanupPartIdentity { checksums: BTreeMap, } +#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)] +struct SourceCleanupErasureIdentity { + algorithm: String, + data_blocks: usize, + parity_blocks: usize, + block_size: usize, + distribution: Vec, +} + #[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)] pub(crate) struct SourceCleanupVersionIdentity { name: String, @@ -447,10 +456,28 @@ pub(crate) struct SourceCleanupVersionIdentity { etag: Option, checksum: Option>, data_dir: Option, + transition_status: String, + transitioned_objname: String, + transition_tier: String, + transition_version_id: Option, + transition_version: Option, + transition_version_state: u8, + expire_restored: bool, + erasure: SourceCleanupErasureIdentity, metadata: BTreeMap, parts: Vec, } +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 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, ¤t, allowed_missing) { - return Ok(()); - } - - Err(source_cleanup_preflight_error( - op_label, - bucket, - object, - "source versions changed after migration started", - )) + ensure_source_cleanup_versions_match(expected, ¤t, allowed_missing) } #[cfg(test)] @@ -651,10 +698,13 @@ pub(crate) async fn cleanup_source_entry_if_unchanged( expected: &FileInfoVersions, allowed_missing: &[SourceCleanupVersionIdentity], op_label: &str, -) -> Result { +) -> std::result::Result { 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, ¤t)); + assert!(source_cleanup_versions_match_with_allowed_missing(&expected, ¤t, &[])); } #[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, ¤t)); + let err = ensure_source_cleanup_versions_match(&expected, ¤t, &[]) + .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, ¤t, &[]) + .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, ¤t, &[]) + .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, ¤t, &[])); } #[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, ¤t)); + let err = ensure_source_cleanup_versions_match(&expected, ¤t, &[]) + .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, ¤t, &[])); + let err = ensure_source_cleanup_versions_match(&expected, ¤t, &[]) + .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, ¤t, &allowed_missing)); + let err = ensure_source_cleanup_versions_match(&expected, ¤t, &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] diff --git a/crates/ecstore/src/services/rebalance/entry.rs b/crates/ecstore/src/services/rebalance/entry.rs index 51398d531..b066a0235 100644 --- a/crates/ecstore/src/services/rebalance/entry.rs +++ b/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, - resolve_rebalance_entry_cleanup_delete_result, resolve_rebalance_file_info_versions_result, + RebalanceEntryCleanupResult, RebalanceEntryTask, load_rebalance_bucket_configs, rebalance_max_attempts, + 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>, - ) -> Result> { + cleanup: impl std::future::Future>, + ) -> Result { // 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)?; - if let Some(message) = cleanup_warning.as_ref() + let cleanup_result = resolve_rebalance_entry_cleanup_delete_result(cleanup.await, bucket, object); + 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 { - warn!( - event = EVENT_REBALANCE_ENTRY, - component = LOG_COMPONENT_ECSTORE, - subsystem = LOG_SUBSYSTEM_REBALANCE, - pool_index, - bucket = %bucket, - object = %entry.name, - stage = "cleanup_source", - cleanup_status = "failed_ignored", - error = %message, - "Ignored rebalance source cleanup failure" - ); - } else { - 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" - ); + match cleanup_result { + RebalanceEntryCleanupResult::Deferred { last_error } => { + debug!( + event = EVENT_REBALANCE_ENTRY, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REBALANCE, + pool_index, + bucket = %bucket, + object = %entry.name, + state = "deferred", + error = %last_error, + "Deferred rebalance entry after source cleanup conflict" + ); + return Ok(RebalanceEntryOutcome::Deferred { last_error }); + } + RebalanceEntryCleanupResult::Completed { warning: Some(message) } => { + warn!( + event = EVENT_REBALANCE_ENTRY, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REBALANCE, + pool_index, + bucket = %bucket, + 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") - .is_none(), - "successful cleanup should not produce a warning" + .expect("finish should succeed"), + RebalanceEntryCleanupResult::Completed { warning: None } ); 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 - .finish_rebalance_entry_after_cleanup(0, "bucket", "object.bin", &[&warning_version], async { Err(Error::SlowDown) }) + let warning_result = store + .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"); } } diff --git a/crates/ecstore/src/services/rebalance/mod.rs b/crates/ecstore/src/services/rebalance/mod.rs index 07295b76d..126cfa81c 100644 --- a/crates/ecstore/src/services/rebalance/mod.rs +++ b/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; diff --git a/crates/ecstore/src/services/rebalance/rebalance_unit_tests.rs b/crates/ecstore/src/services/rebalance/rebalance_unit_tests.rs index affafdc42..b4bd196fc 100644 --- a/crates/ecstore/src/services/rebalance/rebalance_unit_tests.rs +++ b/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, - 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, + 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, }; 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") - .expect("cleanup delete failures should be downgraded to warnings") - .expect("cleanup delete failure should return warning"); - let message = warning.as_str(); - assert!(message.contains("rebalance cleanup delete failed for bucket-a/obj.txt")); + let result = resolve_rebalance_entry_cleanup_delete_result(Err(Error::SlowDown.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_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] diff --git a/crates/ecstore/src/services/rebalance/runtime.rs b/crates/ecstore/src/services/rebalance/runtime.rs index 70d3df0f8..906e92922 100644 --- a/crates/ecstore/src/services/rebalance/runtime.rs +++ b/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, + 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, 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) -> 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, diff --git a/crates/ecstore/src/services/rebalance/worker.rs b/crates/ecstore/src/services/rebalance/worker.rs index 1c1afdf80..2ad077f30 100644 --- a/crates/ecstore/src/services/rebalance/worker.rs +++ b/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( pub(super) type RebalanceEntryTask = tokio::task::JoinHandle>; +#[derive(Debug, Clone, PartialEq, Eq)] +pub(super) enum RebalanceEntryCleanupResult { + Completed { warning: Option }, + Deferred { last_error: String }, +} + pub(super) async fn wait_rebalance_entry_tasks( set_idx: usize, tasks: Arc>>, @@ -145,14 +153,23 @@ where } pub(super) fn resolve_rebalance_entry_cleanup_delete_result( - result: Result, + result: std::result::Result, bucket: &str, object_name: &str, -) -> Result> { +) -> RebalanceEntryCleanupResult { match result { - Ok(_) => Ok(None), - Err(err) if is_err_object_not_found(&err) || is_err_version_not_found(&err) => Ok(None), - Err(err) => Ok(Some(format!("rebalance cleanup delete failed for {bucket}/{object_name}: {err}"))), + Ok(_) => RebalanceEntryCleanupResult::Completed { warning: None }, + Err(SourceCleanupError::Storage(err)) if is_err_object_not_found(&err) || is_err_version_not_found(&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}")), + }, } } diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index 8f9c4bbc1..e51247eed 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/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; }