From f5f727e077563c27561ace1d413b77eb26d3f54f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=AE=89=E6=AD=A3=E8=B6=85?= Date: Sat, 20 Jun 2026 16:31:58 +0800 Subject: [PATCH] refactor: move ecstore rebalance metadata helpers (#3655) --- crates/ecstore/src/rebalance.rs | 663 ++---------------------- crates/ecstore/src/rebalance/meta.rs | 611 ++++++++++++++++++++++ docs/architecture/migration-progress.md | 52 +- 3 files changed, 688 insertions(+), 638 deletions(-) create mode 100644 crates/ecstore/src/rebalance/meta.rs diff --git a/crates/ecstore/src/rebalance.rs b/crates/ecstore/src/rebalance.rs index 00635ba05..02a224079 100644 --- a/crates/ecstore/src/rebalance.rs +++ b/crates/ecstore/src/rebalance.rs @@ -61,6 +61,20 @@ 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:"; +mod meta; +use meta::{ + apply_rebalance_save_option, apply_rebalance_terminal_event, classify_rebalance_terminal_event, clone_arc_by_index, + clone_first_arc, clone_rebalance_pool_stats, complete_rebalance_pools_at_goal, complete_rebalance_pools_with_empty_queue, + defer_bucket_in_rebalance_queue, ensure_valid_rebalance_pool_index, has_deferred_rebalance_error, + invalid_rebalance_pool_index_error, is_rebalance_conflicting_with_decommission, is_rebalance_in_progress, + mark_rebalance_bucket_done, merge_rebalance_meta, percent_free_ratio, rebalance_goal_reached, + rebalance_meta_load_no_data_error, rebalance_meta_load_unknown_format_error, rebalance_meta_load_unknown_version_error, + rebalance_metadata_not_initialized_error, record_rebalance_cleanup_warning_in_meta, resolve_next_rebalance_bucket, + resolve_rebalance_participants, should_accept_rebalance_stats_update, should_ignore_rebalance_data_usage_cache, + should_pool_participate, should_preserve_rebalance_stopped_state, should_skip_start_rebalance, stop_rebalance_meta_snapshot, + validate_start_rebalance_state, +}; + #[derive(Debug, Default, Clone, Serialize, Deserialize)] pub struct RebalanceStats { #[serde(rename = "ifs")] @@ -556,38 +570,6 @@ pub struct RebalanceMeta { pub pool_stats: Vec, // Per-pool rebalance stats keyed by pool index } -fn is_rebalance_pool_started(pool_stat: &RebalanceStats) -> bool { - pool_stat.participating && pool_stat.info.status == RebalStatus::Started -} - -fn is_rebalance_in_progress(meta: &RebalanceMeta) -> bool { - if meta.stopped_at.is_some() { - return false; - } - - meta.pool_stats.iter().any(is_rebalance_pool_started) -} - -fn is_rebalance_conflicting_with_decommission(meta: &RebalanceMeta) -> bool { - is_rebalance_in_progress(meta) -} - -fn first_rebalance_bucket(pool_stat: &RebalanceStats) -> Option { - pool_stat.buckets.first().cloned() -} - -fn rebalance_meta_load_no_data_error() -> Error { - Error::other("rebalance metadata load failed: metadata payload is too short") -} - -fn rebalance_meta_load_unknown_format_error(fmt: u16) -> Error { - Error::other(format!("rebalance metadata load failed: unknown format {fmt}")) -} - -fn rebalance_meta_load_unknown_version_error(ver: u16) -> Error { - Error::other(format!("rebalance metadata load failed: unknown version {ver}")) -} - impl RebalanceMeta { pub fn new() -> Self { Self::default() @@ -1615,252 +1597,6 @@ impl ECStore { } } -fn rebalance_goal_reached(init_free_space: u64, init_capacity: u64, bytes: u64, percent_free_goal: f64) -> bool { - if init_capacity == 0 { - return false; - } - - let pfi = (init_free_space + bytes) as f64 / init_capacity as f64; - pfi + f64::EPSILON >= percent_free_goal -} - -fn percent_free_ratio(total_free: u64, total_cap: u64) -> f64 { - if total_cap == 0 { - return 0.0; - } - total_free as f64 / total_cap as f64 -} - -fn next_rebal_bucket_from_stat(pool_stat: &RebalanceStats) -> Option { - if pool_stat.buckets.is_empty() { - return None; - } - - first_rebalance_bucket(pool_stat) -} - -fn rebalance_metadata_not_initialized_error(operation: &str) -> Error { - Error::other(format!("failed to {operation}: rebalance metadata not initialized")) -} - -fn invalid_rebalance_pool_index_error(pool_index: usize, pool_count: usize) -> Error { - Error::other(format!("invalid rebalance pool index {pool_index} for {pool_count} pools")) -} - -fn clone_rebalance_pool_stats(meta: Option<&RebalanceMeta>) -> Result> { - let Some(meta) = meta else { - return Err(rebalance_metadata_not_initialized_error("clone rebalance pool stats")); - }; - Ok(meta.pool_stats.clone()) -} - -fn should_accept_rebalance_stats_update(meta: &RebalanceMeta, pool_index: usize) -> bool { - if meta.stopped_at.is_some() { - return false; - } - - meta.pool_stats - .get(pool_index) - .is_some_and(|pool_stat| pool_stat.info.status == RebalStatus::Started) -} - -fn resolve_next_rebalance_bucket(meta: Option<&RebalanceMeta>, pool_index: usize) -> Result> { - let Some(meta) = meta else { - return Err(rebalance_metadata_not_initialized_error("resolve next rebalance bucket")); - }; - - ensure_valid_rebalance_pool_index(meta.pool_stats.len(), pool_index)?; - let Some(pool_stat) = meta.pool_stats.get(pool_index) else { - return Err(invalid_rebalance_pool_index_error(pool_index, meta.pool_stats.len())); - }; - - if pool_stat.info.status == RebalStatus::Completed || !pool_stat.participating { - debug!( - event = EVENT_REBALANCE_BUCKET, - component = LOG_COMPONENT_ECSTORE, - subsystem = LOG_SUBSYSTEM_REBALANCE, - pool_index, - state = "unavailable", - reason = "completed_or_not_participating", - "No rebalance bucket available" - ); - return Ok(None); - } - - if pool_stat.buckets.is_empty() { - debug!( - event = EVENT_REBALANCE_BUCKET, - component = LOG_COMPONENT_ECSTORE, - subsystem = LOG_SUBSYSTEM_REBALANCE, - pool_index, - state = "unavailable", - reason = "bucket_queue_empty", - "No rebalance bucket available" - ); - return Ok(None); - } - - if let Some(bucket) = next_rebal_bucket_from_stat(pool_stat) { - debug!( - event = EVENT_REBALANCE_BUCKET, - component = LOG_COMPONENT_ECSTORE, - subsystem = LOG_SUBSYSTEM_REBALANCE, - pool_index, - bucket = %bucket, - state = "selected", - "Selected rebalance bucket" - ); - return Ok(Some(bucket)); - } - - debug!( - event = EVENT_REBALANCE_BUCKET, - component = LOG_COMPONENT_ECSTORE, - subsystem = LOG_SUBSYSTEM_REBALANCE, - pool_index, - state = "unavailable", - reason = "selection_returned_none", - "No rebalance bucket available" - ); - Ok(None) -} - -fn mark_rebalance_bucket_done(meta: Option<&mut RebalanceMeta>, pool_index: usize, bucket: &str) -> Result<()> { - let Some(meta) = meta else { - return Err(rebalance_metadata_not_initialized_error("mark rebalance bucket done")); - }; - - ensure_valid_rebalance_pool_index(meta.pool_stats.len(), pool_index)?; - let Some(pool_stat) = meta.pool_stats.get_mut(pool_index) else { - return Err(invalid_rebalance_pool_index_error(pool_index, meta.pool_stats.len())); - }; - - if take_bucket_from_rebalance_queue(pool_stat, bucket) { - debug!( - event = EVENT_REBALANCE_BUCKET, - component = LOG_COMPONENT_ECSTORE, - subsystem = LOG_SUBSYSTEM_REBALANCE, - pool_index, - bucket = %bucket, - state = "queue_removed", - remaining_bucket_count = pool_stat.buckets.len(), - "Removed bucket from rebalance queue" - ); - if has_deferred_rebalance_error(pool_stat) { - pool_stat.info.last_error = None; - } - Ok(()) - } else { - Err(Error::other(format!( - "failed to mark rebalance bucket done: bucket {bucket} was not queued for pool {pool_index}" - ))) - } -} - -fn record_rebalance_cleanup_warning_in_meta( - meta: Option<&mut RebalanceMeta>, - pool_index: usize, - bucket: &str, - object: &str, - message: String, - now: OffsetDateTime, -) -> Result<()> { - let Some(meta) = meta else { - return Err(rebalance_metadata_not_initialized_error("record rebalance cleanup warning")); - }; - - ensure_valid_rebalance_pool_index(meta.pool_stats.len(), pool_index)?; - let Some(pool_stat) = meta.pool_stats.get_mut(pool_index) else { - return Err(invalid_rebalance_pool_index_error(pool_index, meta.pool_stats.len())); - }; - - pool_stat.cleanup_warnings.count = pool_stat.cleanup_warnings.count.saturating_add(1); - pool_stat.cleanup_warnings.last_message = Some(message); - pool_stat.cleanup_warnings.last_bucket = Some(bucket.to_string()); - pool_stat.cleanup_warnings.last_object = Some(object.to_string()); - pool_stat.cleanup_warnings.last_at = Some(now); - meta.last_refreshed_at = Some(now); - Ok(()) -} - -fn take_bucket_from_rebalance_queue(pool_stat: &mut RebalanceStats, bucket: &str) -> bool { - let mut found = false; - pool_stat.buckets.retain(|name| { - if name == bucket { - found = true; - pool_stat.rebalanced_buckets.push(name.clone()); - false - } else { - true - } - }); - - found -} - -fn defer_bucket_in_rebalance_queue(pool_stat: &mut RebalanceStats, bucket: &str) -> Result<()> { - let Some(pos) = pool_stat.buckets.iter().position(|name| name == bucket) else { - return Err(Error::other(format!("failed to defer rebalance bucket {bucket}: bucket was not queued"))); - }; - - let bucket = pool_stat.buckets.remove(pos); - pool_stat.buckets.push(bucket); - Ok(()) -} - -fn should_pool_participate(init_free_space: u64, init_capacity: u64, percent_free_goal: f64) -> bool { - init_capacity > 0 && percent_free_ratio(init_free_space, init_capacity) < percent_free_goal -} - -fn complete_rebalance_pools_at_goal(meta: &mut RebalanceMeta, now: OffsetDateTime) -> bool { - let mut changed = false; - - for pool_stat in meta.pool_stats.iter_mut() { - if !is_rebalance_pool_started(pool_stat) || has_deferred_rebalance_error(pool_stat) { - continue; - } - - if rebalance_goal_reached( - pool_stat.init_free_space, - pool_stat.init_capacity, - pool_stat.bytes, - meta.percent_free_goal, - ) { - pool_stat.info.status = RebalStatus::Completed; - pool_stat.info.end_time = Some(now); - pool_stat.info.last_error = None; - changed = true; - } - } - - changed -} - -fn complete_rebalance_pools_with_empty_queue(meta: &mut RebalanceMeta, now: OffsetDateTime) -> bool { - let mut changed = false; - - for pool_stat in meta.pool_stats.iter_mut() { - if !is_rebalance_pool_started(pool_stat) || !pool_stat.buckets.is_empty() { - continue; - } - - pool_stat.info.status = RebalStatus::Completed; - pool_stat.info.end_time = Some(now); - pool_stat.info.last_error = None; - changed = true; - } - - changed -} - -fn has_deferred_rebalance_error(pool_stat: &RebalanceStats) -> bool { - pool_stat - .info - .last_error - .as_deref() - .is_some_and(|last_error| last_error.starts_with(REBALANCE_DEFERRED_ENTRY_ERROR_PREFIX)) -} - fn resolve_rebalance_worker_result( set_idx: usize, worker_result: std::result::Result, tokio::task::JoinError>, @@ -2222,329 +1958,6 @@ async fn load_rebalance_bucket_configs(bucket: &str) -> Result(values: &[Arc], err_msg: &str) -> Result> { - values.first().cloned().ok_or_else(|| Error::other(err_msg)) -} - -fn clone_arc_by_index(values: &[Arc], idx: usize, err_prefix: &str) -> Result> { - values - .get(idx) - .cloned() - .ok_or_else(|| Error::other(format!("{err_prefix}: {idx}"))) -} - -fn ensure_valid_rebalance_pool_index(pool_count: usize, idx: usize) -> Result<()> { - if idx >= pool_count { - return Err(invalid_rebalance_pool_index_error(idx, pool_count)); - } - - Ok(()) -} - -enum RebalanceTerminalEvent { - Completed { msg: String }, - Stopped { msg: String }, - Failed { msg: String, last_error: String }, - ChannelClosed { msg: String, last_error: String }, -} - -impl RebalanceTerminalEvent { - fn message(&self) -> &str { - match self { - RebalanceTerminalEvent::Completed { msg } - | RebalanceTerminalEvent::Stopped { msg } - | RebalanceTerminalEvent::Failed { msg, .. } - | RebalanceTerminalEvent::ChannelClosed { msg, .. } => msg, - } - } -} - -fn apply_rebalance_terminal_event( - status: &mut RebalStatus, - end_time: &mut Option, - last_error: &mut Option, - terminal_event: RebalanceTerminalEvent, - now: OffsetDateTime, -) { - match terminal_event { - RebalanceTerminalEvent::Completed { .. } => { - *status = RebalStatus::Completed; - *end_time = Some(now); - *last_error = None; - } - RebalanceTerminalEvent::Stopped { .. } => { - *status = RebalStatus::Stopped; - *end_time = Some(now); - *last_error = None; - } - RebalanceTerminalEvent::Failed { last_error: err, .. } - | RebalanceTerminalEvent::ChannelClosed { last_error: err, .. } => { - *status = RebalStatus::Failed; - *end_time = Some(now); - *last_error = Some(err); - } - } -} - -fn classify_rebalance_terminal_event(signal: Option>, now: OffsetDateTime) -> RebalanceTerminalEvent { - match signal { - Some(Ok(())) => RebalanceTerminalEvent::Completed { - msg: format!("Rebalance completed at {now:?}"), - }, - Some(Err(err)) => { - if is_err_operation_canceled(&err) { - RebalanceTerminalEvent::Stopped { - msg: format!("Rebalance stopped at {now:?}"), - } - } else { - RebalanceTerminalEvent::Failed { - msg: format!("Rebalance failed at {now:?} with err {err:?}"), - last_error: err.to_string(), - } - } - } - None => RebalanceTerminalEvent::ChannelClosed { - msg: format!("Rebalance save task channel closed unexpectedly at {now:?}"), - last_error: format!("rebalance save channel closed before terminal event at {now:?}"), - }, - } -} - -fn ensure_rebalance_not_decommissioning(decommission_running: bool) -> bool { - !decommission_running -} - -fn validate_start_rebalance_state(decommission_running: bool, meta_loaded: bool) -> Result<()> { - if !ensure_rebalance_not_decommissioning(decommission_running) { - return Err(Error::DecommissionAlreadyRunning); - } - if !meta_loaded { - return Err(Error::ConfigNotFound); - } - - Ok(()) -} - -fn should_skip_start_rebalance(cancel_attached: bool, in_progress: bool) -> bool { - cancel_attached && in_progress -} - -fn is_rebalance_stopped_terminal_event(terminal_event: &RebalanceTerminalEvent) -> bool { - matches!(terminal_event, RebalanceTerminalEvent::Stopped { .. }) -} - -fn should_preserve_rebalance_stopped_state( - meta_stopped: bool, - status: RebalStatus, - terminal_event: &RebalanceTerminalEvent, -) -> bool { - (meta_stopped || status == RebalStatus::Stopped) && !is_rebalance_stopped_terminal_event(terminal_event) -} - -fn resolve_rebalance_participants(pool_stats: &[RebalanceStats], pool_count: usize) -> Vec { - let mut participants = vec![false; pool_count]; - - for (idx, pool_stat) in pool_stats.iter().enumerate() { - if idx >= participants.len() { - break; - } - - if pool_stat.info.status == RebalStatus::Started { - participants[idx] = pool_stat.participating; - } - } - - participants -} - -fn is_rebalance_actively_running(meta: &RebalanceMeta) -> bool { - meta.cancel.is_some() && is_rebalance_in_progress(meta) -} - -fn should_ignore_rebalance_data_usage_cache(bucket: &str) -> bool { - bucket == crate::disk::RUSTFS_META_BUCKET -} - -fn apply_rebalance_save_option(meta: &mut RebalanceMeta, pool_idx: usize, opt: RebalSaveOpt, now: OffsetDateTime) { - match opt { - RebalSaveOpt::Stats => { - if pool_idx >= meta.pool_stats.len() { - info!("save_rebalance_stats: pool_idx {pool_idx} out of range for pool_stats"); - } - } - RebalSaveOpt::StoppedAt => { - apply_stopped_at(meta, now); - } - } - - meta.last_refreshed_at = Some(now); -} - -fn is_rebalance_terminal_status(status: RebalStatus) -> bool { - matches!(status, RebalStatus::Completed | RebalStatus::Stopped | RebalStatus::Failed) -} - -fn merge_rebalance_bucket_lists(remote: &mut Vec, local: &[String]) { - let mut existing: HashSet = remote.iter().cloned().collect(); - for bucket in local { - if existing.insert(bucket.clone()) { - remote.push(bucket.clone()); - } - } -} - -fn remove_rebalanced_buckets_from_queue(pool_stat: &mut RebalanceStats) { - let rebalanced_buckets: HashSet = pool_stat.rebalanced_buckets.iter().cloned().collect(); - pool_stat.buckets.retain(|bucket| !rebalanced_buckets.contains(bucket)); -} - -fn merge_rebalance_cleanup_warnings(remote: &mut RebalanceCleanupWarnings, local: &RebalanceCleanupWarnings) { - remote.count = remote.count.max(local.count); - - if should_replace_rebalance_cleanup_warning(remote.last_at, local.last_at) { - remote.last_message = local.last_message.clone(); - remote.last_bucket = local.last_bucket.clone(); - remote.last_object = local.last_object.clone(); - remote.last_at = local.last_at; - } -} - -fn should_replace_rebalance_cleanup_warning(remote_at: Option, local_at: Option) -> bool { - match (remote_at, local_at) { - (_, None) => false, - (None, Some(_)) => true, - (Some(remote), Some(local)) => local >= remote, - } -} - -fn merge_rebalance_pool_stats(remote: &mut RebalanceStats, local: &RebalanceStats) { - remote.init_free_space = remote.init_free_space.max(local.init_free_space); - remote.init_capacity = remote.init_capacity.max(local.init_capacity); - remote.participating |= local.participating; - - merge_rebalance_bucket_lists(&mut remote.buckets, &local.buckets); - merge_rebalance_bucket_lists(&mut remote.rebalanced_buckets, &local.rebalanced_buckets); - remove_rebalanced_buckets_from_queue(remote); - - let local_is_newer = local.num_versions >= remote.num_versions; - remote.num_objects = remote.num_objects.max(local.num_objects); - remote.num_versions = remote.num_versions.max(local.num_versions); - remote.bytes = remote.bytes.max(local.bytes); - merge_rebalance_cleanup_warnings(&mut remote.cleanup_warnings, &local.cleanup_warnings); - - if local_is_newer { - remote.bucket = local.bucket.clone(); - remote.object = local.object.clone(); - } - - if remote.info.start_time.is_none() { - remote.info.start_time = local.info.start_time; - } - - if is_rebalance_terminal_status(remote.info.status) && local.info.status == RebalStatus::Started { - return; - } - - if remote.info.status == RebalStatus::Stopped && matches!(local.info.status, RebalStatus::Started | RebalStatus::Completed) { - return; - } - - match local.info.status { - RebalStatus::Failed => { - remote.info.status = RebalStatus::Failed; - remote.info.end_time = local.info.end_time.or(remote.info.end_time); - remote.info.last_error = local.info.last_error.clone().or_else(|| remote.info.last_error.clone()); - } - RebalStatus::Stopped => { - if remote.info.status != RebalStatus::Failed { - remote.info.status = RebalStatus::Stopped; - remote.info.end_time = local.info.end_time.or(remote.info.end_time); - remote.info.last_error = None; - } - } - RebalStatus::Completed => { - if !matches!(remote.info.status, RebalStatus::Failed | RebalStatus::Stopped) { - remote.info.status = RebalStatus::Completed; - remote.info.end_time = local.info.end_time.or(remote.info.end_time); - remote.info.last_error = None; - } - } - RebalStatus::Started => { - if !is_rebalance_terminal_status(remote.info.status) { - remote.info.status = RebalStatus::Started; - remote.info.last_error = local.info.last_error.clone(); - } - } - RebalStatus::None => {} - } -} - -fn merge_rebalance_meta(remote: &mut RebalanceMeta, local: &RebalanceMeta) { - if remote.id.is_empty() { - *remote = local.clone(); - return; - } - - if !local.id.is_empty() && remote.id != local.id { - *remote = local.clone(); - return; - } - - remote.percent_free_goal = local.percent_free_goal; - remote.last_refreshed_at = Some(OffsetDateTime::now_utc()); - if remote.stopped_at.is_none() { - remote.stopped_at = local.stopped_at; - } - - if remote.pool_stats.len() < local.pool_stats.len() { - remote.pool_stats.resize_with(local.pool_stats.len(), RebalanceStats::default); - } - - for (idx, local_pool_stat) in local.pool_stats.iter().enumerate() { - if let Some(remote_pool_stat) = remote.pool_stats.get_mut(idx) { - merge_rebalance_pool_stats(remote_pool_stat, local_pool_stat); - } - } -} - -fn mark_started_rebalance_pools_stopped(meta: &mut RebalanceMeta, stop_time: OffsetDateTime) { - for pool_stat in meta.pool_stats.iter_mut() { - if pool_stat.info.status == RebalStatus::Started { - pool_stat.info.status = RebalStatus::Stopped; - pool_stat.info.end_time.get_or_insert(stop_time); - pool_stat.info.last_error = None; - } - } -} - -fn apply_stopped_at(meta: &mut RebalanceMeta, now: OffsetDateTime) { - meta.stopped_at = Some(now); - mark_started_rebalance_pools_stopped(meta, now); -} - -fn stop_rebalance_state(meta: &mut RebalanceMeta, now: OffsetDateTime) { - if let Some(cancel_tx) = meta.cancel.take() { - cancel_tx.cancel(); - } - - let stop_time = meta.stopped_at.unwrap_or(now); - if meta.stopped_at.is_none() && is_rebalance_in_progress(meta) { - meta.stopped_at = Some(stop_time); - } - - if meta.stopped_at.is_some() { - mark_started_rebalance_pools_stopped(meta, stop_time); - } -} - -fn stop_rebalance_meta_snapshot(meta: Option<&mut RebalanceMeta>, now: OffsetDateTime) -> Option { - meta.map(|meta| { - stop_rebalance_state(meta, now); - meta.last_refreshed_at = Some(now); - meta.clone() - }) -} - impl ECStore { #[allow(unused_assignments)] #[tracing::instrument(skip(self, set))] @@ -3205,35 +2618,33 @@ impl SetDisks { #[cfg(test)] mod rebalance_unit_tests { use super::REBALANCE_DEFERRED_ENTRY_ERROR_PREFIX; - use super::first_rebalance_bucket; - use super::is_rebalance_actively_running; - use super::is_rebalance_conflicting_with_decommission; - use super::is_rebalance_in_progress; - use super::percent_free_ratio; - use super::rebalance_goal_reached; + use super::meta::{ + RebalanceTerminalEvent, apply_rebalance_save_option, apply_rebalance_terminal_event, apply_stopped_at, + classify_rebalance_terminal_event, clone_arc_by_index, clone_first_arc, clone_rebalance_pool_stats, + complete_rebalance_pools_at_goal, complete_rebalance_pools_with_empty_queue, defer_bucket_in_rebalance_queue, + ensure_rebalance_not_decommissioning, ensure_valid_rebalance_pool_index, first_rebalance_bucket, + has_deferred_rebalance_error, is_rebalance_actively_running, is_rebalance_conflicting_with_decommission, + is_rebalance_in_progress, is_rebalance_stopped_terminal_event, mark_rebalance_bucket_done, merge_rebalance_bucket_lists, + merge_rebalance_meta, next_rebal_bucket_from_stat, percent_free_ratio, rebalance_goal_reached, + rebalance_meta_load_no_data_error, rebalance_meta_load_unknown_format_error, rebalance_meta_load_unknown_version_error, + record_rebalance_cleanup_warning_in_meta, remove_rebalanced_buckets_from_queue, resolve_next_rebalance_bucket, + resolve_rebalance_participants, should_accept_rebalance_stats_update, should_ignore_rebalance_data_usage_cache, + should_pool_participate, should_preserve_rebalance_stopped_state, should_skip_start_rebalance, + stop_rebalance_meta_snapshot, stop_rebalance_state, take_bucket_from_rebalance_queue, validate_start_rebalance_state, + }; use super::{ DiskError, GetObjectReader, MigrationBackend, MigrationVersionResult, ObjectInfo, ObjectOptions, RebalSaveOpt, RebalStatus, RebalanceBucketOutcome, RebalanceCleanupWarnings, RebalanceEntryOutcome, RebalanceInfo, RebalanceMeta, - RebalanceStats, RebalanceTerminalEvent, apply_rebalance_save_option, apply_rebalance_terminal_event, apply_stopped_at, - classify_rebalance_terminal_event, clone_arc_by_index, clone_first_arc, clone_rebalance_pool_stats, - complete_rebalance_pools_at_goal, complete_rebalance_pools_with_empty_queue, defer_bucket_in_rebalance_queue, - ensure_rebalance_listing_disks_available, ensure_rebalance_not_decommissioning, ensure_valid_rebalance_pool_index, - has_deferred_rebalance_error, is_rebalance_stopped_terminal_event, is_transient_rebalance_error, - load_rebalance_bucket_configs, mark_rebalance_bucket_done, merge_rebalance_meta, migrate_entry_version, - migrate_entry_version_with_retry_wait, next_rebal_bucket_from_stat, parse_rebalance_max_attempts, - rebalance_delete_marker_opts, rebalance_listing_retry_delay, rebalance_meta_load_no_data_error, - rebalance_meta_load_unknown_format_error, rebalance_meta_load_unknown_version_error, rebalance_migration_retry_delay, - record_rebalance_cleanup_warning_in_meta, resolve_load_rebalance_stats_update_result, resolve_next_rebalance_bucket, + RebalanceStats, ensure_rebalance_listing_disks_available, is_transient_rebalance_error, load_rebalance_bucket_configs, + migrate_entry_version, migrate_entry_version_with_retry_wait, parse_rebalance_max_attempts, rebalance_delete_marker_opts, + 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_participants, + 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, send_rebalance_done_signal, should_accept_rebalance_stats_update, - should_cleanup_rebalance_source_entry, should_count_rebalance_version_complete, should_defer_rebalance_entry_failure, - should_ignore_rebalance_data_usage_cache, should_pool_participate, should_preserve_rebalance_stopped_state, - should_retry_rebalance_listing, should_skip_rebalance_delete_marker, should_skip_start_rebalance, - stop_rebalance_meta_snapshot, stop_rebalance_state, take_bucket_from_rebalance_queue, validate_start_rebalance_state, - wait_rebalance_listing_retry, with_rebalance_entry_context, + resolve_rebalance_worker_result, 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_listing_retry, with_rebalance_entry_context, }; use crate::data_movement; use crate::data_usage::DATA_USAGE_CACHE_NAME; @@ -5555,7 +4966,7 @@ mod rebalance_unit_tests { "bucket-d".to_string(), ]; - super::merge_rebalance_bucket_lists(&mut remote, &local); + merge_rebalance_bucket_lists(&mut remote, &local); assert_eq!( remote, @@ -5581,7 +4992,7 @@ mod rebalance_unit_tests { ..Default::default() }; - super::remove_rebalanced_buckets_from_queue(&mut pool_stat); + remove_rebalanced_buckets_from_queue(&mut pool_stat); assert_eq!(pool_stat.buckets, vec!["bucket-a".to_string(), "bucket-c".to_string()]); } diff --git a/crates/ecstore/src/rebalance/meta.rs b/crates/ecstore/src/rebalance/meta.rs new file mode 100644 index 000000000..727d497e3 --- /dev/null +++ b/crates/ecstore/src/rebalance/meta.rs @@ -0,0 +1,611 @@ +use super::{ + EVENT_REBALANCE_BUCKET, Error, LOG_COMPONENT_ECSTORE, LOG_SUBSYSTEM_REBALANCE, REBALANCE_DEFERRED_ENTRY_ERROR_PREFIX, + RebalSaveOpt, RebalStatus, RebalanceCleanupWarnings, RebalanceMeta, RebalanceStats, Result, +}; +use crate::error::is_err_operation_canceled; +use std::collections::HashSet; +use std::sync::Arc; +use time::OffsetDateTime; +use tracing::{debug, info}; + +pub(super) fn is_rebalance_pool_started(pool_stat: &RebalanceStats) -> bool { + pool_stat.participating && pool_stat.info.status == RebalStatus::Started +} + +pub(super) fn is_rebalance_in_progress(meta: &RebalanceMeta) -> bool { + if meta.stopped_at.is_some() { + return false; + } + + meta.pool_stats.iter().any(is_rebalance_pool_started) +} + +pub(super) fn is_rebalance_conflicting_with_decommission(meta: &RebalanceMeta) -> bool { + is_rebalance_in_progress(meta) +} + +pub(super) fn first_rebalance_bucket(pool_stat: &RebalanceStats) -> Option { + pool_stat.buckets.first().cloned() +} + +pub(super) fn rebalance_meta_load_no_data_error() -> Error { + Error::other("rebalance metadata load failed: metadata payload is too short") +} + +pub(super) fn rebalance_meta_load_unknown_format_error(fmt: u16) -> Error { + Error::other(format!("rebalance metadata load failed: unknown format {fmt}")) +} + +pub(super) fn rebalance_meta_load_unknown_version_error(ver: u16) -> Error { + Error::other(format!("rebalance metadata load failed: unknown version {ver}")) +} +pub(super) fn rebalance_goal_reached(init_free_space: u64, init_capacity: u64, bytes: u64, percent_free_goal: f64) -> bool { + if init_capacity == 0 { + return false; + } + + let pfi = (init_free_space + bytes) as f64 / init_capacity as f64; + pfi + f64::EPSILON >= percent_free_goal +} + +pub(super) fn percent_free_ratio(total_free: u64, total_cap: u64) -> f64 { + if total_cap == 0 { + return 0.0; + } + total_free as f64 / total_cap as f64 +} + +pub(super) fn next_rebal_bucket_from_stat(pool_stat: &RebalanceStats) -> Option { + if pool_stat.buckets.is_empty() { + return None; + } + + first_rebalance_bucket(pool_stat) +} + +pub(super) fn rebalance_metadata_not_initialized_error(operation: &str) -> Error { + Error::other(format!("failed to {operation}: rebalance metadata not initialized")) +} + +pub(super) fn invalid_rebalance_pool_index_error(pool_index: usize, pool_count: usize) -> Error { + Error::other(format!("invalid rebalance pool index {pool_index} for {pool_count} pools")) +} + +pub(super) fn clone_rebalance_pool_stats(meta: Option<&RebalanceMeta>) -> Result> { + let Some(meta) = meta else { + return Err(rebalance_metadata_not_initialized_error("clone rebalance pool stats")); + }; + Ok(meta.pool_stats.clone()) +} + +pub(super) fn should_accept_rebalance_stats_update(meta: &RebalanceMeta, pool_index: usize) -> bool { + if meta.stopped_at.is_some() { + return false; + } + + meta.pool_stats + .get(pool_index) + .is_some_and(|pool_stat| pool_stat.info.status == RebalStatus::Started) +} + +pub(super) fn resolve_next_rebalance_bucket(meta: Option<&RebalanceMeta>, pool_index: usize) -> Result> { + let Some(meta) = meta else { + return Err(rebalance_metadata_not_initialized_error("resolve next rebalance bucket")); + }; + + ensure_valid_rebalance_pool_index(meta.pool_stats.len(), pool_index)?; + let Some(pool_stat) = meta.pool_stats.get(pool_index) else { + return Err(invalid_rebalance_pool_index_error(pool_index, meta.pool_stats.len())); + }; + + if pool_stat.info.status == RebalStatus::Completed || !pool_stat.participating { + debug!( + event = EVENT_REBALANCE_BUCKET, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REBALANCE, + pool_index, + state = "unavailable", + reason = "completed_or_not_participating", + "No rebalance bucket available" + ); + return Ok(None); + } + + if pool_stat.buckets.is_empty() { + debug!( + event = EVENT_REBALANCE_BUCKET, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REBALANCE, + pool_index, + state = "unavailable", + reason = "bucket_queue_empty", + "No rebalance bucket available" + ); + return Ok(None); + } + + if let Some(bucket) = next_rebal_bucket_from_stat(pool_stat) { + debug!( + event = EVENT_REBALANCE_BUCKET, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REBALANCE, + pool_index, + bucket = %bucket, + state = "selected", + "Selected rebalance bucket" + ); + return Ok(Some(bucket)); + } + + debug!( + event = EVENT_REBALANCE_BUCKET, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REBALANCE, + pool_index, + state = "unavailable", + reason = "selection_returned_none", + "No rebalance bucket available" + ); + Ok(None) +} + +pub(super) fn mark_rebalance_bucket_done(meta: Option<&mut RebalanceMeta>, pool_index: usize, bucket: &str) -> Result<()> { + let Some(meta) = meta else { + return Err(rebalance_metadata_not_initialized_error("mark rebalance bucket done")); + }; + + ensure_valid_rebalance_pool_index(meta.pool_stats.len(), pool_index)?; + let Some(pool_stat) = meta.pool_stats.get_mut(pool_index) else { + return Err(invalid_rebalance_pool_index_error(pool_index, meta.pool_stats.len())); + }; + + if take_bucket_from_rebalance_queue(pool_stat, bucket) { + debug!( + event = EVENT_REBALANCE_BUCKET, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REBALANCE, + pool_index, + bucket = %bucket, + state = "queue_removed", + remaining_bucket_count = pool_stat.buckets.len(), + "Removed bucket from rebalance queue" + ); + if has_deferred_rebalance_error(pool_stat) { + pool_stat.info.last_error = None; + } + Ok(()) + } else { + Err(Error::other(format!( + "failed to mark rebalance bucket done: bucket {bucket} was not queued for pool {pool_index}" + ))) + } +} + +pub(super) fn record_rebalance_cleanup_warning_in_meta( + meta: Option<&mut RebalanceMeta>, + pool_index: usize, + bucket: &str, + object: &str, + message: String, + now: OffsetDateTime, +) -> Result<()> { + let Some(meta) = meta else { + return Err(rebalance_metadata_not_initialized_error("record rebalance cleanup warning")); + }; + + ensure_valid_rebalance_pool_index(meta.pool_stats.len(), pool_index)?; + let Some(pool_stat) = meta.pool_stats.get_mut(pool_index) else { + return Err(invalid_rebalance_pool_index_error(pool_index, meta.pool_stats.len())); + }; + + pool_stat.cleanup_warnings.count = pool_stat.cleanup_warnings.count.saturating_add(1); + pool_stat.cleanup_warnings.last_message = Some(message); + pool_stat.cleanup_warnings.last_bucket = Some(bucket.to_string()); + pool_stat.cleanup_warnings.last_object = Some(object.to_string()); + pool_stat.cleanup_warnings.last_at = Some(now); + meta.last_refreshed_at = Some(now); + Ok(()) +} + +pub(super) fn take_bucket_from_rebalance_queue(pool_stat: &mut RebalanceStats, bucket: &str) -> bool { + let mut found = false; + pool_stat.buckets.retain(|name| { + if name == bucket { + found = true; + pool_stat.rebalanced_buckets.push(name.clone()); + false + } else { + true + } + }); + + found +} + +pub(super) fn defer_bucket_in_rebalance_queue(pool_stat: &mut RebalanceStats, bucket: &str) -> Result<()> { + let Some(pos) = pool_stat.buckets.iter().position(|name| name == bucket) else { + return Err(Error::other(format!("failed to defer rebalance bucket {bucket}: bucket was not queued"))); + }; + + let bucket = pool_stat.buckets.remove(pos); + pool_stat.buckets.push(bucket); + Ok(()) +} + +pub(super) fn should_pool_participate(init_free_space: u64, init_capacity: u64, percent_free_goal: f64) -> bool { + init_capacity > 0 && percent_free_ratio(init_free_space, init_capacity) < percent_free_goal +} + +pub(super) fn complete_rebalance_pools_at_goal(meta: &mut RebalanceMeta, now: OffsetDateTime) -> bool { + let mut changed = false; + + for pool_stat in meta.pool_stats.iter_mut() { + if !is_rebalance_pool_started(pool_stat) || has_deferred_rebalance_error(pool_stat) { + continue; + } + + if rebalance_goal_reached( + pool_stat.init_free_space, + pool_stat.init_capacity, + pool_stat.bytes, + meta.percent_free_goal, + ) { + pool_stat.info.status = RebalStatus::Completed; + pool_stat.info.end_time = Some(now); + pool_stat.info.last_error = None; + changed = true; + } + } + + changed +} + +pub(super) fn complete_rebalance_pools_with_empty_queue(meta: &mut RebalanceMeta, now: OffsetDateTime) -> bool { + let mut changed = false; + + for pool_stat in meta.pool_stats.iter_mut() { + if !is_rebalance_pool_started(pool_stat) || !pool_stat.buckets.is_empty() { + continue; + } + + pool_stat.info.status = RebalStatus::Completed; + pool_stat.info.end_time = Some(now); + pool_stat.info.last_error = None; + changed = true; + } + + changed +} + +pub(super) fn has_deferred_rebalance_error(pool_stat: &RebalanceStats) -> bool { + pool_stat + .info + .last_error + .as_deref() + .is_some_and(|last_error| last_error.starts_with(REBALANCE_DEFERRED_ENTRY_ERROR_PREFIX)) +} +pub(super) fn clone_first_arc(values: &[Arc], err_msg: &str) -> Result> { + values.first().cloned().ok_or_else(|| Error::other(err_msg)) +} + +pub(super) fn clone_arc_by_index(values: &[Arc], idx: usize, err_prefix: &str) -> Result> { + values + .get(idx) + .cloned() + .ok_or_else(|| Error::other(format!("{err_prefix}: {idx}"))) +} + +pub(super) fn ensure_valid_rebalance_pool_index(pool_count: usize, idx: usize) -> Result<()> { + if idx >= pool_count { + return Err(invalid_rebalance_pool_index_error(idx, pool_count)); + } + + Ok(()) +} + +pub(super) enum RebalanceTerminalEvent { + Completed { msg: String }, + Stopped { msg: String }, + Failed { msg: String, last_error: String }, + ChannelClosed { msg: String, last_error: String }, +} + +impl RebalanceTerminalEvent { + pub(super) fn message(&self) -> &str { + match self { + RebalanceTerminalEvent::Completed { msg } + | RebalanceTerminalEvent::Stopped { msg } + | RebalanceTerminalEvent::Failed { msg, .. } + | RebalanceTerminalEvent::ChannelClosed { msg, .. } => msg, + } + } +} + +pub(super) fn apply_rebalance_terminal_event( + status: &mut RebalStatus, + end_time: &mut Option, + last_error: &mut Option, + terminal_event: RebalanceTerminalEvent, + now: OffsetDateTime, +) { + match terminal_event { + RebalanceTerminalEvent::Completed { .. } => { + *status = RebalStatus::Completed; + *end_time = Some(now); + *last_error = None; + } + RebalanceTerminalEvent::Stopped { .. } => { + *status = RebalStatus::Stopped; + *end_time = Some(now); + *last_error = None; + } + RebalanceTerminalEvent::Failed { last_error: err, .. } + | RebalanceTerminalEvent::ChannelClosed { last_error: err, .. } => { + *status = RebalStatus::Failed; + *end_time = Some(now); + *last_error = Some(err); + } + } +} + +pub(super) fn classify_rebalance_terminal_event(signal: Option>, now: OffsetDateTime) -> RebalanceTerminalEvent { + match signal { + Some(Ok(())) => RebalanceTerminalEvent::Completed { + msg: format!("Rebalance completed at {now:?}"), + }, + Some(Err(err)) => { + if is_err_operation_canceled(&err) { + RebalanceTerminalEvent::Stopped { + msg: format!("Rebalance stopped at {now:?}"), + } + } else { + RebalanceTerminalEvent::Failed { + msg: format!("Rebalance failed at {now:?} with err {err:?}"), + last_error: err.to_string(), + } + } + } + None => RebalanceTerminalEvent::ChannelClosed { + msg: format!("Rebalance save task channel closed unexpectedly at {now:?}"), + last_error: format!("rebalance save channel closed before terminal event at {now:?}"), + }, + } +} + +pub(super) fn ensure_rebalance_not_decommissioning(decommission_running: bool) -> bool { + !decommission_running +} + +pub(super) fn validate_start_rebalance_state(decommission_running: bool, meta_loaded: bool) -> Result<()> { + if !ensure_rebalance_not_decommissioning(decommission_running) { + return Err(Error::DecommissionAlreadyRunning); + } + if !meta_loaded { + return Err(Error::ConfigNotFound); + } + + Ok(()) +} + +pub(super) fn should_skip_start_rebalance(cancel_attached: bool, in_progress: bool) -> bool { + cancel_attached && in_progress +} + +pub(super) fn is_rebalance_stopped_terminal_event(terminal_event: &RebalanceTerminalEvent) -> bool { + matches!(terminal_event, RebalanceTerminalEvent::Stopped { .. }) +} + +pub(super) fn should_preserve_rebalance_stopped_state( + meta_stopped: bool, + status: RebalStatus, + terminal_event: &RebalanceTerminalEvent, +) -> bool { + (meta_stopped || status == RebalStatus::Stopped) && !is_rebalance_stopped_terminal_event(terminal_event) +} + +pub(super) fn resolve_rebalance_participants(pool_stats: &[RebalanceStats], pool_count: usize) -> Vec { + let mut participants = vec![false; pool_count]; + + for (idx, pool_stat) in pool_stats.iter().enumerate() { + if idx >= participants.len() { + break; + } + + if pool_stat.info.status == RebalStatus::Started { + participants[idx] = pool_stat.participating; + } + } + + participants +} + +pub(super) fn is_rebalance_actively_running(meta: &RebalanceMeta) -> bool { + meta.cancel.is_some() && is_rebalance_in_progress(meta) +} + +pub(super) fn should_ignore_rebalance_data_usage_cache(bucket: &str) -> bool { + bucket == crate::disk::RUSTFS_META_BUCKET +} + +pub(super) fn apply_rebalance_save_option(meta: &mut RebalanceMeta, pool_idx: usize, opt: RebalSaveOpt, now: OffsetDateTime) { + match opt { + RebalSaveOpt::Stats => { + if pool_idx >= meta.pool_stats.len() { + info!("save_rebalance_stats: pool_idx {pool_idx} out of range for pool_stats"); + } + } + RebalSaveOpt::StoppedAt => { + apply_stopped_at(meta, now); + } + } + + meta.last_refreshed_at = Some(now); +} + +pub(super) fn is_rebalance_terminal_status(status: RebalStatus) -> bool { + matches!(status, RebalStatus::Completed | RebalStatus::Stopped | RebalStatus::Failed) +} + +pub(super) fn merge_rebalance_bucket_lists(remote: &mut Vec, local: &[String]) { + let mut existing: HashSet = remote.iter().cloned().collect(); + for bucket in local { + if existing.insert(bucket.clone()) { + remote.push(bucket.clone()); + } + } +} + +pub(super) fn remove_rebalanced_buckets_from_queue(pool_stat: &mut RebalanceStats) { + let rebalanced_buckets: HashSet = pool_stat.rebalanced_buckets.iter().cloned().collect(); + pool_stat.buckets.retain(|bucket| !rebalanced_buckets.contains(bucket)); +} + +pub(super) fn merge_rebalance_cleanup_warnings(remote: &mut RebalanceCleanupWarnings, local: &RebalanceCleanupWarnings) { + remote.count = remote.count.max(local.count); + + if should_replace_rebalance_cleanup_warning(remote.last_at, local.last_at) { + remote.last_message = local.last_message.clone(); + remote.last_bucket = local.last_bucket.clone(); + remote.last_object = local.last_object.clone(); + remote.last_at = local.last_at; + } +} + +pub(super) fn should_replace_rebalance_cleanup_warning( + remote_at: Option, + local_at: Option, +) -> bool { + match (remote_at, local_at) { + (_, None) => false, + (None, Some(_)) => true, + (Some(remote), Some(local)) => local >= remote, + } +} + +pub(super) fn merge_rebalance_pool_stats(remote: &mut RebalanceStats, local: &RebalanceStats) { + remote.init_free_space = remote.init_free_space.max(local.init_free_space); + remote.init_capacity = remote.init_capacity.max(local.init_capacity); + remote.participating |= local.participating; + + merge_rebalance_bucket_lists(&mut remote.buckets, &local.buckets); + merge_rebalance_bucket_lists(&mut remote.rebalanced_buckets, &local.rebalanced_buckets); + remove_rebalanced_buckets_from_queue(remote); + + let local_is_newer = local.num_versions >= remote.num_versions; + remote.num_objects = remote.num_objects.max(local.num_objects); + remote.num_versions = remote.num_versions.max(local.num_versions); + remote.bytes = remote.bytes.max(local.bytes); + merge_rebalance_cleanup_warnings(&mut remote.cleanup_warnings, &local.cleanup_warnings); + + if local_is_newer { + remote.bucket = local.bucket.clone(); + remote.object = local.object.clone(); + } + + if remote.info.start_time.is_none() { + remote.info.start_time = local.info.start_time; + } + + if is_rebalance_terminal_status(remote.info.status) && local.info.status == RebalStatus::Started { + return; + } + + if remote.info.status == RebalStatus::Stopped && matches!(local.info.status, RebalStatus::Started | RebalStatus::Completed) { + return; + } + + match local.info.status { + RebalStatus::Failed => { + remote.info.status = RebalStatus::Failed; + remote.info.end_time = local.info.end_time.or(remote.info.end_time); + remote.info.last_error = local.info.last_error.clone().or_else(|| remote.info.last_error.clone()); + } + RebalStatus::Stopped => { + if remote.info.status != RebalStatus::Failed { + remote.info.status = RebalStatus::Stopped; + remote.info.end_time = local.info.end_time.or(remote.info.end_time); + remote.info.last_error = None; + } + } + RebalStatus::Completed => { + if !matches!(remote.info.status, RebalStatus::Failed | RebalStatus::Stopped) { + remote.info.status = RebalStatus::Completed; + remote.info.end_time = local.info.end_time.or(remote.info.end_time); + remote.info.last_error = None; + } + } + RebalStatus::Started => { + if !is_rebalance_terminal_status(remote.info.status) { + remote.info.status = RebalStatus::Started; + remote.info.last_error = local.info.last_error.clone(); + } + } + RebalStatus::None => {} + } +} + +pub(super) fn merge_rebalance_meta(remote: &mut RebalanceMeta, local: &RebalanceMeta) { + if remote.id.is_empty() { + *remote = local.clone(); + return; + } + + if !local.id.is_empty() && remote.id != local.id { + *remote = local.clone(); + return; + } + + remote.percent_free_goal = local.percent_free_goal; + remote.last_refreshed_at = Some(OffsetDateTime::now_utc()); + if remote.stopped_at.is_none() { + remote.stopped_at = local.stopped_at; + } + + if remote.pool_stats.len() < local.pool_stats.len() { + remote.pool_stats.resize_with(local.pool_stats.len(), RebalanceStats::default); + } + + for (idx, local_pool_stat) in local.pool_stats.iter().enumerate() { + if let Some(remote_pool_stat) = remote.pool_stats.get_mut(idx) { + merge_rebalance_pool_stats(remote_pool_stat, local_pool_stat); + } + } +} + +pub(super) fn mark_started_rebalance_pools_stopped(meta: &mut RebalanceMeta, stop_time: OffsetDateTime) { + for pool_stat in meta.pool_stats.iter_mut() { + if pool_stat.info.status == RebalStatus::Started { + pool_stat.info.status = RebalStatus::Stopped; + pool_stat.info.end_time.get_or_insert(stop_time); + pool_stat.info.last_error = None; + } + } +} + +pub(super) fn apply_stopped_at(meta: &mut RebalanceMeta, now: OffsetDateTime) { + meta.stopped_at = Some(now); + mark_started_rebalance_pools_stopped(meta, now); +} + +pub(super) fn stop_rebalance_state(meta: &mut RebalanceMeta, now: OffsetDateTime) { + if let Some(cancel_tx) = meta.cancel.take() { + cancel_tx.cancel(); + } + + let stop_time = meta.stopped_at.unwrap_or(now); + if meta.stopped_at.is_none() && is_rebalance_in_progress(meta) { + meta.stopped_at = Some(stop_time); + } + + if meta.stopped_at.is_some() { + mark_started_rebalance_pools_stopped(meta, stop_time); + } +} + +pub(super) fn stop_rebalance_meta_snapshot(meta: Option<&mut RebalanceMeta>, now: OffsetDateTime) -> Option { + meta.map(|meta| { + stop_rebalance_state(meta, now); + meta.last_refreshed_at = Some(now); + meta.clone() + }) +} diff --git a/docs/architecture/migration-progress.md b/docs/architecture/migration-progress.md index b89746f5b..6e8b51a9e 100644 --- a/docs/architecture/migration-progress.md +++ b/docs/architecture/migration-progress.md @@ -5,18 +5,19 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block ## Current Context - Issue: [`rustfs/backlog#660`](https://github.com/rustfs/backlog/issues/660) -- Branch: `overtrue/arch-ecstore-layout-pool-space-builder` -- Baseline: completed `E-006/E-REBALANCE-001`. +- Branch: `overtrue/arch-ecstore-rebalance-meta-helpers` +- Baseline: completed `E-007/E-LAYOUT-005`. - Stacked on: merged ECStore layout foundation, format layout ownership, and - endpoint layout move slices plus the set-format heal and pool-space layout - helper slices and the rebalance support helper boundary slice. + endpoint layout move slices plus the set-format heal, pool-space layout + helper, rebalance support helper, and pool-space builder slices. - PR type for this branch: `pure-move` - Runtime behavior changes: none. -- Rust code changes: move ECStore capacity checks and pool-space builder logic - into the layout pool-space owner while preserving the old `store` export path - and rebalance orchestration. +- Rust code changes: move ECStore rebalance metadata state, bucket queue, + terminal event, participant, and metadata merge helpers into `rebalance::meta` + while preserving the `RebalanceMeta`/`RebalanceStats` wire structs and ECStore + orchestration. - CI/script changes: none. -- Docs changes: record the ECStore pool-space builder layout slice. +- Docs changes: record the ECStore rebalance metadata helper slice. ## Phase 0 Tasks @@ -2314,10 +2315,26 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block diff hygiene, Rust risk scan, branch freshness check, pre-commit quality gate, and three-expert review. +- [x] `E-008/E-REBALANCE-002` Move ECStore rebalance metadata helpers. + - Do: move rebalance metadata status, bucket queue, terminal event, + participant, cleanup-warning, metadata merge, and stop-state helpers into + `rebalance::meta` while keeping wire structs and ECStore orchestration in + `rebalance.rs`. + - Acceptance: `rebalance::meta` owns the helper functions, `rebalance.rs` + keeps save/load and object-flow orchestration, and focused rebalance tests + keep covering the moved behavior. + - Must preserve: metadata wire shape, stopped/completed/failed precedence, + bucket queue ordering, cleanup-warning merge semantics, participant + resolution, data-usage cache filtering, start/stop validation, and + percent-free goal math. + - Verification: focused ECStore rebalance tests, ECStore/RustFS/Heal compile + checks, migration/layer guards, formatting, diff hygiene, Rust risk scan, + branch freshness check, pre-commit quality gate, and three-expert review. + ## Next PRs 1. `pure-move`: continue moving ECStore rebalance/layout support helpers once - E-007/E-LAYOUT-005 lands. + E-008/E-REBALANCE-002 lands. 2. `pure-move`: continue pruning residual embedded startup-only orchestration once the lifecycle helpers are merged. @@ -2325,9 +2342,9 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block | Expert | Status | Notes | |---|---|---| -| Quality/architecture | passed | E-007/E-LAYOUT-005 moves capacity and pool-space construction into the layout owner while leaving rebalance orchestration in place. | -| Migration preservation | passed | Capacity math, inode/free-space guards, meta-bucket bypass, pool ordering, and old `store::has_space_for` imports remain preserved. | -| Testing/verification | passed | Focused pool-space/rebalance tests, compile checks, guards, formatting, diff hygiene, Rust risk scan, and full pre-commit passed. | +| Quality/architecture | passed | E-008/E-REBALANCE-002 moves rebalance metadata helpers into a child module while leaving ECStore orchestration in place. | +| Migration preservation | passed | Metadata wire structs, bucket queues, terminal-state precedence, metadata merge behavior, and stop-state handling remain preserved. | +| Testing/verification | passed | Focused rebalance tests, compile checks, guards, formatting, diff hygiene, Rust risk scan, and full pre-commit passed. | ## Verification Notes @@ -2465,6 +2482,17 @@ Passed before push: - `make pre-commit`: passed. - Three-expert review: passed. +- Issue #660 E-008/E-REBALANCE-002 current slice: + - `cargo test -p rustfs-ecstore rebalance::rebalance_unit_tests -- --nocapture`: passed. + - `cargo check -p rustfs-ecstore -p rustfs -p rustfs-heal`: passed. + - `./scripts/check_architecture_migration_rules.sh`: passed. + - `./scripts/check_layer_dependencies.sh`: passed. + - `cargo fmt --all --check`: passed. + - `git diff --check`: passed. + - Rust risk scan on changed Rust files: passed. + - `make pre-commit`: passed. + - Three-expert review: passed. + - Issue #660 X-012 current slice: - `cargo test -p rustfs-extension-schema`: passed. - `cargo check -p rustfs-extension-schema`: passed.