diff --git a/crates/ecstore/src/rebalance.rs b/crates/ecstore/src/rebalance.rs index e81f5b8d0..46d063530 100644 --- a/crates/ecstore/src/rebalance.rs +++ b/crates/ecstore/src/rebalance.rs @@ -14,9 +14,7 @@ use crate::config::com::{read_config_with_metadata, save_config_with_opts}; use crate::data_movement; -use crate::data_usage::DATA_USAGE_CACHE_NAME; use crate::error::{Error, Result}; -use crate::error::{is_err_object_not_found, is_err_version_not_found}; use crate::global::get_global_endpoints; use crate::object_api::{GetObjectReader, ObjectInfo, ObjectOptions, PutObjReader}; use crate::pools::ListCallback; @@ -32,7 +30,6 @@ use rustfs_utils::path::encode_dir_object; use serde::{Deserialize, Serialize}; use std::collections::HashSet; use std::fmt; -use std::future::Future; use std::io::Cursor; use std::sync::Arc; use time::OffsetDateTime; @@ -59,6 +56,7 @@ 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; +mod migration; mod worker; use meta::{ apply_rebalance_save_option, apply_rebalance_terminal_event, classify_rebalance_terminal_event, clone_arc_by_index, @@ -72,16 +70,16 @@ use meta::{ should_pool_participate, should_preserve_rebalance_stopped_state, should_skip_start_rebalance, stop_rebalance_meta_snapshot, validate_start_rebalance_state, }; +use migration::migrate_entry_version; use worker::{ - RebalanceEntryTask, is_transient_rebalance_error, load_rebalance_bucket_configs, rebalance_max_attempts, - rebalance_meta_lock_error, 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_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_skip_rebalance_delete_marker, sleep_rebalance_migration_retry, - wait_rebalance_entry_tasks, with_rebalance_entry_context, + RebalanceEntryTask, load_rebalance_bucket_configs, rebalance_max_attempts, rebalance_meta_lock_error, + 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_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_skip_rebalance_delete_marker, wait_rebalance_entry_tasks, with_rebalance_entry_context, }; #[derive(Debug, Default, Clone, Serialize, Deserialize)] @@ -151,16 +149,6 @@ struct RebalanceBucketConfigs { replication_config: Option<(s3s::dto::ReplicationConfiguration, OffsetDateTime)>, } -#[derive(Debug, Default, Clone)] -pub(crate) struct MigrationVersionResult { - pub moved: bool, - pub ignored: bool, - pub cleanup_ignored: bool, - pub failed: bool, - pub stage: Option<&'static str>, - pub error: Option, -} - #[derive(Debug, Clone, PartialEq, Eq)] enum RebalanceBucketOutcome { Completed, @@ -173,321 +161,6 @@ enum RebalanceEntryOutcome { Deferred { last_error: String }, } -fn rebalance_delete_marker_opts(version: &FileInfo, version_id: Option, src_pool_idx: usize) -> ObjectOptions { - ObjectOptions { - versioned: true, - version_id, - mod_time: version.mod_time, - src_pool_idx, - data_movement: true, - delete_marker: true, - skip_decommissioned: true, - delete_replication: version.replication_state_internal.clone(), - ..Default::default() - } -} - -fn rebalance_remote_tiered_opts(version: &FileInfo, version_id: Option, src_pool_idx: usize) -> ObjectOptions { - ObjectOptions { - versioned: version_id.is_some(), - version_id, - mod_time: version.mod_time, - user_defined: version.metadata.clone(), - src_pool_idx, - data_movement: true, - ..Default::default() - } -} - -#[async_trait::async_trait] -pub(crate) trait MigrationBackend: Send + Sync { - async fn get_object_reader_for_migration( - &self, - bucket: &str, - object: &str, - range: Option, - h: HeaderMap, - opts: &ObjectOptions, - ) -> Result; - - async fn delete_object_for_migration(&self, bucket: &str, object: &str, opts: ObjectOptions) -> Result; - - async fn move_remote_version_for_migration( - &self, - bucket: &str, - object: &str, - fi: &FileInfo, - opts: &ObjectOptions, - ) -> Result<()>; -} - -#[async_trait::async_trait] -impl MigrationBackend for SetDisks { - async fn get_object_reader_for_migration( - &self, - bucket: &str, - object: &str, - range: Option, - h: HeaderMap, - opts: &ObjectOptions, - ) -> Result { - self.get_object_reader(bucket, object, range, h, opts).await - } - - async fn delete_object_for_migration(&self, bucket: &str, object: &str, opts: ObjectOptions) -> Result { - self.delete_object(bucket, object, opts).await - } - - async fn move_remote_version_for_migration( - &self, - bucket: &str, - object: &str, - fi: &FileInfo, - opts: &ObjectOptions, - ) -> Result<()> { - self.decommission_tiered_object(bucket, object, fi, opts).await - } -} - -#[allow(clippy::too_many_arguments)] -pub(crate) async fn migrate_entry_version( - set: &Backend, - bucket: String, - pool_index: usize, - version: &FileInfo, - version_id: Option, - max_attempts: usize, - ignore_data_usage_cache: bool, - transfer: F, -) -> MigrationVersionResult -where - Backend: MigrationBackend + ?Sized, - F: FnMut(usize, String, GetObjectReader) -> Fut + Send, - Fut: Future> + Send, -{ - migrate_entry_version_with_retry_wait( - set, - bucket, - pool_index, - version, - version_id, - max_attempts, - ignore_data_usage_cache, - transfer, - sleep_rebalance_migration_retry, - ) - .await -} - -#[allow(clippy::too_many_arguments)] -async fn migrate_entry_version_with_retry_wait( - set: &Backend, - bucket: String, - pool_index: usize, - version: &FileInfo, - version_id: Option, - max_attempts: usize, - ignore_data_usage_cache: bool, - mut transfer: F, - mut wait_retry: W, -) -> MigrationVersionResult -where - Backend: MigrationBackend + ?Sized, - F: FnMut(usize, String, GetObjectReader) -> Fut + Send, - Fut: Future> + Send, - W: FnMut(Duration) -> WFut + Send, - WFut: Future + Send, -{ - let max_attempts = max_attempts.max(1); - - if ignore_data_usage_cache && bucket == crate::disk::RUSTFS_META_BUCKET && version.name.contains(DATA_USAGE_CACHE_NAME) { - return MigrationVersionResult { - moved: false, - ignored: true, - cleanup_ignored: false, - failed: false, - stage: None, - error: None, - }; - } - - if version.is_remote() { - if let Err(err) = set - .move_remote_version_for_migration( - &bucket, - &version.name, - version, - &rebalance_remote_tiered_opts(version, version_id, pool_index), - ) - .await - { - if is_err_object_not_found(&err) || is_err_version_not_found(&err) { - return MigrationVersionResult { - moved: false, - ignored: true, - cleanup_ignored: true, - failed: false, - stage: Some("move_remote_version"), - error: None, - }; - } - - return MigrationVersionResult { - moved: false, - ignored: false, - cleanup_ignored: false, - failed: true, - stage: Some("move_remote_version"), - error: Some(err), - }; - } - - return MigrationVersionResult { - moved: true, - ignored: false, - cleanup_ignored: false, - failed: false, - stage: None, - error: None, - }; - } - - if version.deleted { - if let Err(err) = set - .delete_object_for_migration(&bucket, &version.name, rebalance_delete_marker_opts(version, version_id, pool_index)) - .await - { - if is_err_object_not_found(&err) || is_err_version_not_found(&err) { - return MigrationVersionResult { - moved: false, - ignored: true, - cleanup_ignored: true, - failed: false, - stage: Some("delete_marker"), - error: None, - }; - } - - return MigrationVersionResult { - moved: false, - ignored: false, - cleanup_ignored: false, - failed: true, - stage: Some("delete_marker"), - error: Some(err), - }; - } - - return MigrationVersionResult { - moved: true, - ignored: false, - cleanup_ignored: false, - failed: false, - stage: None, - error: None, - }; - } - - let mut last_error: Option = None; - for attempt in 0..max_attempts { - let rd = match set - .get_object_reader_for_migration( - &bucket, - &encode_dir_object(&version.name), - None, - HeaderMap::new(), - &ObjectOptions { - version_id: version_id.clone(), - no_lock: true, - ..Default::default() - }, - ) - .await - { - Ok(rd) => rd, - Err(err) => { - if is_err_object_not_found(&err) || is_err_version_not_found(&err) { - return MigrationVersionResult { - moved: false, - ignored: true, - cleanup_ignored: true, - failed: false, - stage: Some("read_source"), - error: None, - }; - } - - last_error = Some(err); - let Some(err) = last_error.as_ref() else { - continue; - }; - if attempt + 1 >= max_attempts || !is_transient_rebalance_error(err) { - return MigrationVersionResult { - moved: false, - ignored: false, - cleanup_ignored: false, - failed: true, - stage: Some("read_source"), - error: last_error, - }; - } - - wait_retry(rebalance_migration_retry_delay(attempt, err)).await; - continue; - } - }; - - if let Err(err) = transfer(pool_index, bucket.clone(), rd).await { - if is_err_object_not_found(&err) || is_err_version_not_found(&err) { - return MigrationVersionResult { - moved: false, - ignored: true, - cleanup_ignored: true, - failed: false, - stage: Some("write_target"), - error: None, - }; - } - - last_error = Some(err); - let Some(err) = last_error.as_ref() else { - continue; - }; - if attempt + 1 >= max_attempts || !is_transient_rebalance_error(err) { - return MigrationVersionResult { - moved: false, - ignored: false, - cleanup_ignored: false, - failed: true, - stage: Some("write_target"), - error: last_error, - }; - } - - wait_retry(rebalance_migration_retry_delay(attempt, err)).await; - continue; - } - - return MigrationVersionResult { - moved: true, - ignored: false, - cleanup_ignored: false, - failed: false, - stage: None, - error: None, - }; - } - - MigrationVersionResult { - moved: false, - ignored: false, - cleanup_ignored: false, - failed: true, - stage: Some("migrate"), - error: last_error, - } -} - #[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)] pub enum RebalStatus { #[default] @@ -2125,6 +1798,10 @@ mod rebalance_unit_tests { 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::migration::{ + MigrationBackend, MigrationVersionResult, migrate_entry_version, migrate_entry_version_with_retry_wait, + rebalance_delete_marker_opts, + }; use super::worker::{ ensure_rebalance_listing_disks_available, is_transient_rebalance_error, load_rebalance_bucket_configs, parse_rebalance_max_attempts, rebalance_listing_retry_delay, rebalance_migration_retry_delay, @@ -2138,9 +1815,8 @@ mod rebalance_unit_tests { wait_rebalance_listing_retry, with_rebalance_entry_context, }; use super::{ - GetObjectReader, MigrationBackend, MigrationVersionResult, ObjectInfo, ObjectOptions, RebalSaveOpt, RebalStatus, - RebalanceBucketOutcome, RebalanceCleanupWarnings, RebalanceEntryOutcome, RebalanceInfo, RebalanceMeta, RebalanceStats, - migrate_entry_version, migrate_entry_version_with_retry_wait, rebalance_delete_marker_opts, + GetObjectReader, ObjectInfo, ObjectOptions, RebalSaveOpt, RebalStatus, RebalanceBucketOutcome, RebalanceCleanupWarnings, + RebalanceEntryOutcome, RebalanceInfo, RebalanceMeta, RebalanceStats, }; use crate::data_movement; use crate::data_usage::DATA_USAGE_CACHE_NAME; diff --git a/crates/ecstore/src/rebalance/migration.rs b/crates/ecstore/src/rebalance/migration.rs new file mode 100644 index 000000000..a17637f23 --- /dev/null +++ b/crates/ecstore/src/rebalance/migration.rs @@ -0,0 +1,336 @@ +use super::worker::{is_transient_rebalance_error, rebalance_migration_retry_delay, sleep_rebalance_migration_retry}; +use crate::data_usage::DATA_USAGE_CACHE_NAME; +use crate::error::{Error, Result, is_err_object_not_found, is_err_version_not_found}; +use crate::object_api::{GetObjectReader, ObjectInfo, ObjectOptions}; +use crate::set_disk::SetDisks; +use http::HeaderMap; +use rustfs_filemeta::FileInfo; +use rustfs_storage_api::{HTTPRangeSpec, ObjectIO, ObjectOperations as _}; +use rustfs_utils::path::encode_dir_object; +use std::future::Future; +use tokio::time::Duration; + +#[derive(Debug, Default, Clone)] +pub(crate) struct MigrationVersionResult { + pub moved: bool, + pub ignored: bool, + pub cleanup_ignored: bool, + pub failed: bool, + pub stage: Option<&'static str>, + pub error: Option, +} + +pub(super) fn rebalance_delete_marker_opts(version: &FileInfo, version_id: Option, src_pool_idx: usize) -> ObjectOptions { + ObjectOptions { + versioned: true, + version_id, + mod_time: version.mod_time, + src_pool_idx, + data_movement: true, + delete_marker: true, + skip_decommissioned: true, + delete_replication: version.replication_state_internal.clone(), + ..Default::default() + } +} + +fn rebalance_remote_tiered_opts(version: &FileInfo, version_id: Option, src_pool_idx: usize) -> ObjectOptions { + ObjectOptions { + versioned: version_id.is_some(), + version_id, + mod_time: version.mod_time, + user_defined: version.metadata.clone(), + src_pool_idx, + data_movement: true, + ..Default::default() + } +} + +#[async_trait::async_trait] +pub(crate) trait MigrationBackend: Send + Sync { + async fn get_object_reader_for_migration( + &self, + bucket: &str, + object: &str, + range: Option, + h: HeaderMap, + opts: &ObjectOptions, + ) -> Result; + + async fn delete_object_for_migration(&self, bucket: &str, object: &str, opts: ObjectOptions) -> Result; + + async fn move_remote_version_for_migration( + &self, + bucket: &str, + object: &str, + fi: &FileInfo, + opts: &ObjectOptions, + ) -> Result<()>; +} + +#[async_trait::async_trait] +impl MigrationBackend for SetDisks { + async fn get_object_reader_for_migration( + &self, + bucket: &str, + object: &str, + range: Option, + h: HeaderMap, + opts: &ObjectOptions, + ) -> Result { + self.get_object_reader(bucket, object, range, h, opts).await + } + + async fn delete_object_for_migration(&self, bucket: &str, object: &str, opts: ObjectOptions) -> Result { + self.delete_object(bucket, object, opts).await + } + + async fn move_remote_version_for_migration( + &self, + bucket: &str, + object: &str, + fi: &FileInfo, + opts: &ObjectOptions, + ) -> Result<()> { + self.decommission_tiered_object(bucket, object, fi, opts).await + } +} + +#[allow(clippy::too_many_arguments)] +pub(crate) async fn migrate_entry_version( + set: &Backend, + bucket: String, + pool_index: usize, + version: &FileInfo, + version_id: Option, + max_attempts: usize, + ignore_data_usage_cache: bool, + transfer: F, +) -> MigrationVersionResult +where + Backend: MigrationBackend + ?Sized, + F: FnMut(usize, String, GetObjectReader) -> Fut + Send, + Fut: Future> + Send, +{ + migrate_entry_version_with_retry_wait( + set, + bucket, + pool_index, + version, + version_id, + max_attempts, + ignore_data_usage_cache, + transfer, + sleep_rebalance_migration_retry, + ) + .await +} + +#[allow(clippy::too_many_arguments)] +pub(super) async fn migrate_entry_version_with_retry_wait( + set: &Backend, + bucket: String, + pool_index: usize, + version: &FileInfo, + version_id: Option, + max_attempts: usize, + ignore_data_usage_cache: bool, + mut transfer: F, + mut wait_retry: W, +) -> MigrationVersionResult +where + Backend: MigrationBackend + ?Sized, + F: FnMut(usize, String, GetObjectReader) -> Fut + Send, + Fut: Future> + Send, + W: FnMut(Duration) -> WFut + Send, + WFut: Future + Send, +{ + let max_attempts = max_attempts.max(1); + + if ignore_data_usage_cache && bucket == crate::disk::RUSTFS_META_BUCKET && version.name.contains(DATA_USAGE_CACHE_NAME) { + return MigrationVersionResult { + moved: false, + ignored: true, + cleanup_ignored: false, + failed: false, + stage: None, + error: None, + }; + } + + if version.is_remote() { + if let Err(err) = set + .move_remote_version_for_migration( + &bucket, + &version.name, + version, + &rebalance_remote_tiered_opts(version, version_id, pool_index), + ) + .await + { + if is_err_object_not_found(&err) || is_err_version_not_found(&err) { + return MigrationVersionResult { + moved: false, + ignored: true, + cleanup_ignored: true, + failed: false, + stage: Some("move_remote_version"), + error: None, + }; + } + + return MigrationVersionResult { + moved: false, + ignored: false, + cleanup_ignored: false, + failed: true, + stage: Some("move_remote_version"), + error: Some(err), + }; + } + + return MigrationVersionResult { + moved: true, + ignored: false, + cleanup_ignored: false, + failed: false, + stage: None, + error: None, + }; + } + + if version.deleted { + if let Err(err) = set + .delete_object_for_migration(&bucket, &version.name, rebalance_delete_marker_opts(version, version_id, pool_index)) + .await + { + if is_err_object_not_found(&err) || is_err_version_not_found(&err) { + return MigrationVersionResult { + moved: false, + ignored: true, + cleanup_ignored: true, + failed: false, + stage: Some("delete_marker"), + error: None, + }; + } + + return MigrationVersionResult { + moved: false, + ignored: false, + cleanup_ignored: false, + failed: true, + stage: Some("delete_marker"), + error: Some(err), + }; + } + + return MigrationVersionResult { + moved: true, + ignored: false, + cleanup_ignored: false, + failed: false, + stage: None, + error: None, + }; + } + + let mut last_error: Option = None; + for attempt in 0..max_attempts { + let rd = match set + .get_object_reader_for_migration( + &bucket, + &encode_dir_object(&version.name), + None, + HeaderMap::new(), + &ObjectOptions { + version_id: version_id.clone(), + no_lock: true, + ..Default::default() + }, + ) + .await + { + Ok(rd) => rd, + Err(err) => { + if is_err_object_not_found(&err) || is_err_version_not_found(&err) { + return MigrationVersionResult { + moved: false, + ignored: true, + cleanup_ignored: true, + failed: false, + stage: Some("read_source"), + error: None, + }; + } + + last_error = Some(err); + let Some(err) = last_error.as_ref() else { + continue; + }; + if attempt + 1 >= max_attempts || !is_transient_rebalance_error(err) { + return MigrationVersionResult { + moved: false, + ignored: false, + cleanup_ignored: false, + failed: true, + stage: Some("read_source"), + error: last_error, + }; + } + + wait_retry(rebalance_migration_retry_delay(attempt, err)).await; + continue; + } + }; + + if let Err(err) = transfer(pool_index, bucket.clone(), rd).await { + if is_err_object_not_found(&err) || is_err_version_not_found(&err) { + return MigrationVersionResult { + moved: false, + ignored: true, + cleanup_ignored: true, + failed: false, + stage: Some("write_target"), + error: None, + }; + } + + last_error = Some(err); + let Some(err) = last_error.as_ref() else { + continue; + }; + if attempt + 1 >= max_attempts || !is_transient_rebalance_error(err) { + return MigrationVersionResult { + moved: false, + ignored: false, + cleanup_ignored: false, + failed: true, + stage: Some("write_target"), + error: last_error, + }; + } + + wait_retry(rebalance_migration_retry_delay(attempt, err)).await; + continue; + } + + return MigrationVersionResult { + moved: true, + ignored: false, + cleanup_ignored: false, + failed: false, + stage: None, + error: None, + }; + } + + MigrationVersionResult { + moved: false, + ignored: false, + cleanup_ignored: false, + failed: true, + stage: Some("migrate"), + error: last_error, + } +} diff --git a/crates/ecstore/src/rebalance/worker.rs b/crates/ecstore/src/rebalance/worker.rs index 801ad36b5..dabda47a1 100644 --- a/crates/ecstore/src/rebalance/worker.rs +++ b/crates/ecstore/src/rebalance/worker.rs @@ -1,8 +1,8 @@ +use super::migration::MigrationVersionResult; use super::{ - DEFAULT_REBALANCE_MAX_ATTEMPTS, EVENT_REBALANCE_LISTING, LOG_COMPONENT_ECSTORE, LOG_SUBSYSTEM_REBALANCE, - MigrationVersionResult, 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, + 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, }; use crate::cache_value::metacache_set::{ListPathRawOptions, list_path_raw}; use crate::disk::error::DiskError; diff --git a/docs/architecture/migration-progress.md b/docs/architecture/migration-progress.md index 4a67b3e34..c9f6d4950 100644 --- a/docs/architecture/migration-progress.md +++ b/docs/architecture/migration-progress.md @@ -5,20 +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-rebalance-worker-helpers` -- Baseline: completed `E-008/E-REBALANCE-002`. +- Branch: `overtrue/arch-ecstore-rebalance-migration-helpers` +- Baseline: completed `E-009/E-REBALANCE-003`. - Stacked on: merged ECStore layout foundation, format layout ownership, and endpoint layout move slices plus the set-format heal, pool-space layout - helper, rebalance support helper, pool-space builder, and rebalance metadata - helper slices. + helper, rebalance support helper, pool-space builder, rebalance metadata + helper, and rebalance worker helper slices. - PR type for this branch: `pure-move` - Runtime behavior changes: none. -- Rust code changes: move ECStore rebalance worker result handling, retry, - listing, bucket config, and source cleanup helper boundaries into - `rebalance::worker` while preserving the `RebalanceMeta`/`RebalanceStats` - wire structs and ECStore orchestration. +- Rust code changes: move ECStore rebalance migration backend, version-result, + delete-marker option, remote-tier option, and version migration retry helpers + into `rebalance::migration` while preserving ECStore orchestration. - CI/script changes: none. -- Docs changes: record the ECStore rebalance worker helper slice. +- Docs changes: record the ECStore rebalance migration helper slice. ## Phase 0 Tasks @@ -2348,10 +2347,26 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block checks, migration/layer guards, formatting, diff hygiene, Rust risk scan, branch freshness check, pre-commit quality gate, and three-expert review. +- [x] `E-010/E-REBALANCE-004` Move ECStore rebalance migration helpers. + - Do: move migration backend abstraction, migration version result, + delete-marker/remote-tier option builders, and version migration retry flow + into `rebalance::migration` while keeping high-level rebalance orchestration + in `rebalance.rs`. + - Acceptance: `rebalance::migration` owns migration helper functions and + result types, `rebalance.rs` keeps orchestration and wire structs, and + focused rebalance tests keep covering moved behavior. + - Must preserve: remote-tier object movement, delete-marker replication + state, data-usage cache skip behavior, source read/write retry semantics, + transient/non-transient classification, retry backoff, not-found handling, + migration stage labels, and cleanup accounting. + - 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-009/E-REBALANCE-003 lands. + E-010/E-REBALANCE-004 lands. 2. `pure-move`: continue pruning residual embedded startup-only orchestration once the lifecycle helpers are merged. @@ -2359,8 +2374,8 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block | Expert | Status | Notes | |---|---|---| -| Quality/architecture | passed | E-009/E-REBALANCE-003 moves rebalance worker helpers into a child module while leaving ECStore orchestration in place. | -| Migration preservation | passed | Worker join error context, retry/backoff classification, listing retry cancellation, bucket config handling, and cleanup decisions remain preserved. | +| Quality/architecture | passed | E-010/E-REBALANCE-004 moves migration helpers into a child module while leaving ECStore orchestration in place. | +| Migration preservation | passed | Remote-tier movement, delete-marker options, retry/backoff classification, not-found handling, stage labels, and cleanup accounting remain preserved. | | Testing/verification | passed | Focused rebalance tests, compile checks, guards, formatting, diff hygiene, Rust risk scan, and full pre-commit passed. | ## Verification Notes