From 1d9fd3d7ac2d1bc71479cda4032481701794c543 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 13:34:23 +0800 Subject: [PATCH] refactor: move ecstore rebalance support helpers (#3650) --- crates/ecstore/src/store.rs | 26 --- crates/ecstore/src/store/rebalance.rs | 129 +------------ crates/ecstore/src/store/rebalance/support.rs | 170 ++++++++++++++++++ docs/architecture/migration-progress.md | 51 ++++-- 4 files changed, 216 insertions(+), 160 deletions(-) create mode 100644 crates/ecstore/src/store/rebalance/support.rs diff --git a/crates/ecstore/src/store.rs b/crates/ecstore/src/store.rs index f8bbc1fb6..bde53b519 100644 --- a/crates/ecstore/src/store.rs +++ b/crates/ecstore/src/store.rs @@ -84,7 +84,6 @@ use rustfs_storage_api::{ }; use rustfs_utils::path::{decode_dir_object, encode_dir_object, path_join_buf}; use s3s::dto::{BucketVersioningStatus, ObjectLockConfiguration, ObjectLockEnabled, VersioningConfiguration}; -use std::cmp::Ordering; use std::net::SocketAddr; use std::process::exit; use std::time::SystemTime; @@ -319,31 +318,6 @@ impl ECStore { // } // } -// init_local_disks must succeed before the server starts -/// create unique lock clients for the endpoints and store them globally -#[derive(Debug, Default)] -struct PoolErr { - index: Option, - err: Option, -} - -#[derive(Debug, Default)] -pub struct PoolObjInfo { - pub index: usize, - pub object_info: ObjectInfo, - pub err: Option, -} - -impl Clone for PoolObjInfo { - fn clone(&self) -> Self { - Self { - index: self.index, - object_info: self.object_info.clone(), - err: self.err.clone(), - } - } -} - // #[derive(Debug, Default, Clone)] // pub struct ListPathOptions { // pub id: String, diff --git a/crates/ecstore/src/store/rebalance.rs b/crates/ecstore/src/store/rebalance.rs index 475739663..7e977b19a 100644 --- a/crates/ecstore/src/store/rebalance.rs +++ b/crates/ecstore/src/store/rebalance.rs @@ -15,129 +15,12 @@ use super::*; use crate::config::get_global_storage_class; use rustfs_storage_api::{NamespaceLocking as _, ObjectOperations as _, StorageAdminApi}; - -struct LatestObjectInfoCandidate { - info: Option, - idx: usize, - err: Option, -} - -struct RebalanceDeletePoolResult { - pool_idx: usize, - result: Result, -} - -fn pool_lookup_not_found_error(bucket: &str, object: &str, opts: &ObjectOptions) -> Error { - let object = decode_dir_object(object); - - if let Some(version_id) = &opts.version_id { - StorageError::VersionNotFound(bucket.to_owned(), object, version_id.clone()) - } else { - StorageError::ObjectNotFound(bucket.to_owned(), object) - } -} - -fn resolve_store_rebalance_pool_meta_reload_result(result: Result<()>, stage: &str) -> Result<()> { - result.map_err(|err| Error::other(format!("store rebalance pool meta reload failed during {stage}: {err}"))) -} - -fn resolve_rebalance_delete_from_all_pools_result(result: Result, bucket: &str, object: &str) -> Result { - result.map_err(|err| Error::other(format!("failed to delete rebalance source object {bucket}/{object}: {err}"))) -} - -fn is_ignorable_rebalance_delete_error(err: &Error) -> bool { - is_err_object_not_found(err) || is_err_version_not_found(err) -} - -fn rebalance_delete_pool_error(pool_idx: usize, bucket: &str, object: &str, err: Error) -> Error { - Error::other(format!("pool {pool_idx} delete failed for {bucket}/{object}: {err}")) -} - -fn resolve_rebalance_delete_from_all_pools_results( - results: Vec, - bucket: &str, - object: &str, -) -> Result { - let mut deleted = None; - let mut ignored_error = None; - - for pool_result in results { - let pool_idx = pool_result.pool_idx; - match pool_result.result { - Ok(info) => { - if deleted.is_none() { - deleted = Some(info); - } - } - Err(err) if is_ignorable_rebalance_delete_error(&err) => { - ignored_error = Some((pool_idx, err)); - } - Err(err) => { - return Err(rebalance_delete_pool_error(pool_idx, bucket, object, err)); - } - } - } - - if let Some(info) = deleted { - return Ok(info); - } - - if let Some((pool_idx, err)) = ignored_error { - return Err(rebalance_delete_pool_error(pool_idx, bucket, object, err)); - } - - Err(Error::other(format!( - "failed to delete rebalance source object {bucket}/{object}: no pools were attempted" - ))) -} - -fn rebalance_disk_set_lookup_error(pool_idx: usize, set_idx: usize, pool_count: usize) -> Error { - Error::other(format!( - "failed to resolve rebalance disk set: pool index {pool_idx}, set index {set_idx}, pool count {pool_count}", - )) -} - -fn resolve_latest_object_info_candidates( - mut candidates: Vec, - bucket: &str, - object: &str, - opts: &ObjectOptions, -) -> Result<(ObjectInfo, usize)> { - candidates.sort_by(|a, b| { - let a_mod = if let Some(info) = &a.info { - info.mod_time.unwrap_or(OffsetDateTime::UNIX_EPOCH) - } else { - OffsetDateTime::UNIX_EPOCH - }; - - let b_mod = if let Some(info) = &b.info { - info.mod_time.unwrap_or(OffsetDateTime::UNIX_EPOCH) - } else { - OffsetDateTime::UNIX_EPOCH - }; - - if a_mod == b_mod { - return if a.idx < b.idx { Ordering::Greater } else { Ordering::Less }; - } - - b_mod.cmp(&a_mod) - }); - - for candidate in candidates { - if let Some(info) = candidate.info { - return Ok((info, candidate.idx)); - } - - if let Some(err) = candidate.err - && !is_err_object_not_found(&err) - && !is_err_version_not_found(&err) - { - return Err(err); - } - } - - Err(pool_lookup_not_found_error(bucket, object, opts)) -} +pub(in crate::store) mod support; +use support::{ + LatestObjectInfoCandidate, PoolErr, PoolObjInfo, RebalanceDeletePoolResult, pool_lookup_not_found_error, + rebalance_disk_set_lookup_error, resolve_latest_object_info_candidates, resolve_rebalance_delete_from_all_pools_result, + resolve_rebalance_delete_from_all_pools_results, resolve_store_rebalance_pool_meta_reload_result, +}; async fn build_server_pools_available_space( bucket: &str, diff --git a/crates/ecstore/src/store/rebalance/support.rs b/crates/ecstore/src/store/rebalance/support.rs new file mode 100644 index 000000000..9c653e5db --- /dev/null +++ b/crates/ecstore/src/store/rebalance/support.rs @@ -0,0 +1,170 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use std::cmp::Ordering; + +use crate::error::{Error, Result, StorageError, is_err_object_not_found, is_err_version_not_found}; +use crate::object_api::{ObjectInfo, ObjectOptions}; +use rustfs_utils::path::decode_dir_object; +use time::OffsetDateTime; + +#[derive(Debug, Default)] +pub(in crate::store) struct PoolErr { + pub(in crate::store) index: Option, + pub(in crate::store) err: Option, +} + +#[derive(Debug, Default)] +pub(in crate::store) struct PoolObjInfo { + pub(in crate::store) index: usize, + pub(in crate::store) object_info: ObjectInfo, + pub(in crate::store) err: Option, +} + +impl Clone for PoolObjInfo { + fn clone(&self) -> Self { + Self { + index: self.index, + object_info: self.object_info.clone(), + err: self.err.clone(), + } + } +} + +pub(super) struct LatestObjectInfoCandidate { + pub(super) info: Option, + pub(super) idx: usize, + pub(super) err: Option, +} + +pub(super) struct RebalanceDeletePoolResult { + pub(super) pool_idx: usize, + pub(super) result: Result, +} + +pub(super) fn pool_lookup_not_found_error(bucket: &str, object: &str, opts: &ObjectOptions) -> Error { + let object = decode_dir_object(object); + + if let Some(version_id) = &opts.version_id { + StorageError::VersionNotFound(bucket.to_owned(), object, version_id.clone()) + } else { + StorageError::ObjectNotFound(bucket.to_owned(), object) + } +} + +pub(super) fn resolve_store_rebalance_pool_meta_reload_result(result: Result<()>, stage: &str) -> Result<()> { + result.map_err(|err| Error::other(format!("store rebalance pool meta reload failed during {stage}: {err}"))) +} + +pub(super) fn resolve_rebalance_delete_from_all_pools_result( + result: Result, + bucket: &str, + object: &str, +) -> Result { + result.map_err(|err| Error::other(format!("failed to delete rebalance source object {bucket}/{object}: {err}"))) +} + +fn is_ignorable_rebalance_delete_error(err: &Error) -> bool { + is_err_object_not_found(err) || is_err_version_not_found(err) +} + +fn rebalance_delete_pool_error(pool_idx: usize, bucket: &str, object: &str, err: Error) -> Error { + Error::other(format!("pool {pool_idx} delete failed for {bucket}/{object}: {err}")) +} + +pub(super) fn resolve_rebalance_delete_from_all_pools_results( + results: Vec, + bucket: &str, + object: &str, +) -> Result { + let mut deleted = None; + let mut ignored_error = None; + + for pool_result in results { + let pool_idx = pool_result.pool_idx; + match pool_result.result { + Ok(info) => { + if deleted.is_none() { + deleted = Some(info); + } + } + Err(err) if is_ignorable_rebalance_delete_error(&err) => { + ignored_error = Some((pool_idx, err)); + } + Err(err) => { + return Err(rebalance_delete_pool_error(pool_idx, bucket, object, err)); + } + } + } + + if let Some(info) = deleted { + return Ok(info); + } + + if let Some((pool_idx, err)) = ignored_error { + return Err(rebalance_delete_pool_error(pool_idx, bucket, object, err)); + } + + Err(Error::other(format!( + "failed to delete rebalance source object {bucket}/{object}: no pools were attempted" + ))) +} + +pub(super) fn rebalance_disk_set_lookup_error(pool_idx: usize, set_idx: usize, pool_count: usize) -> Error { + Error::other(format!( + "failed to resolve rebalance disk set: pool index {pool_idx}, set index {set_idx}, pool count {pool_count}", + )) +} + +pub(super) fn resolve_latest_object_info_candidates( + mut candidates: Vec, + bucket: &str, + object: &str, + opts: &ObjectOptions, +) -> Result<(ObjectInfo, usize)> { + candidates.sort_by(|a, b| { + let a_mod = if let Some(info) = &a.info { + info.mod_time.unwrap_or(OffsetDateTime::UNIX_EPOCH) + } else { + OffsetDateTime::UNIX_EPOCH + }; + + let b_mod = if let Some(info) = &b.info { + info.mod_time.unwrap_or(OffsetDateTime::UNIX_EPOCH) + } else { + OffsetDateTime::UNIX_EPOCH + }; + + if a_mod == b_mod { + return if a.idx < b.idx { Ordering::Greater } else { Ordering::Less }; + } + + b_mod.cmp(&a_mod) + }); + + for candidate in candidates { + if let Some(info) = candidate.info { + return Ok((info, candidate.idx)); + } + + if let Some(err) = candidate.err + && !is_err_object_not_found(&err) + && !is_err_version_not_found(&err) + { + return Err(err); + } + } + + Err(pool_lookup_not_found_error(bucket, object, opts)) +} diff --git a/docs/architecture/migration-progress.md b/docs/architecture/migration-progress.md index bc3471073..a97948cea 100644 --- a/docs/architecture/migration-progress.md +++ b/docs/architecture/migration-progress.md @@ -5,16 +5,18 @@ 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` -- Baseline: completed `E-004/E-LAYOUT-003`. +- Branch: `overtrue/arch-ecstore-rebalance-helper-boundary` +- Baseline: completed `E-005/E-LAYOUT-004`. - Stacked on: merged ECStore layout foundation, format layout ownership, and - endpoint layout move slices plus the set-format heal helper layout slice. + endpoint layout move slices plus the set-format heal and pool-space layout + helper slices. - PR type for this branch: `pure-move` - Runtime behavior changes: none. -- Rust code changes: move ECStore pool-space selection helper types into the - internal layout bucket while preserving the old `store` export path. +- Rust code changes: move ECStore rebalance support helper types and pure + result reducers into a local rebalance support boundary while preserving + store orchestration. - CI/script changes: none. -- Docs changes: record the ECStore pool-space helper layout slice. +- Docs changes: record the ECStore rebalance support helper boundary slice. ## Phase 0 Tasks @@ -2282,10 +2284,25 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block scan, branch freshness check, pre-commit quality gate, and three-expert review. +- [x] `E-006/E-REBALANCE-001` Move ECStore rebalance support helpers. + - Do: move rebalance-only helper DTOs, pool lookup error classification, and + delete/latest-object result reducers into `store::rebalance::support` + while keeping async store orchestration in the existing modules. + - Acceptance: rebalance callers keep the same `PoolObjInfo`/`PoolErr` + access inside `store`, delete aggregation and latest-object selection keep + the same behavior, and the moved helpers remain private to the rebalance + boundary. + - Must preserve: latest-object tie-breaks, delete result aggregation, pool + lookup not-found/version-not-found classification, rebalance disk-set lookup + error context, object delete flows, and existing rebalance control flow. + - 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 runtime-neutral pool/set layout helpers once - E-005/E-LAYOUT-004 lands. +1. `pure-move`: continue moving ECStore rebalance/layout support helpers once + E-006/E-REBALANCE-001 lands. 2. `pure-move`: continue pruning residual embedded startup-only orchestration once the lifecycle helpers are merged. @@ -2293,9 +2310,9 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block | Expert | Status | Notes | |---|---|---| -| Quality/architecture | passed | E-005/E-LAYOUT-004 extracts runtime-neutral pool-space helper types into ECStore layout without moving rebalance or store orchestration. | -| Migration preservation | passed | Pool index ordering, available-space summation, max-used-percent filtering, excluded-pool zeroing, and old `store` import compatibility remain preserved. | -| Testing/verification | passed | Focused pool-space helper tests, compile checks, guards, formatting, diff hygiene, Rust risk scan, and full pre-commit passed. | +| Quality/architecture | passed | E-006/E-REBALANCE-001 extracts rebalance support helpers into a local boundary without moving async store orchestration. | +| Migration preservation | passed | Latest-object tie-breaks, delete aggregation, pool lookup error classification, and sibling `store` visibility remain preserved. | +| Testing/verification | passed | Focused rebalance tests, compile checks, guards, formatting, diff hygiene, Rust risk scan, and full pre-commit passed. | ## Verification Notes @@ -2408,6 +2425,18 @@ Passed before push: - `make pre-commit`: passed. - Three-expert review: passed. +- Issue #660 E-006/E-REBALANCE-001 current slice: + - `cargo test -p rustfs-ecstore store::rebalance -- --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; no risky added lines were + introduced. + - `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.