diff --git a/crates/ecstore/src/services/rebalance/control.rs b/crates/ecstore/src/services/rebalance/control.rs index b5d9702ea..09e1993e7 100644 --- a/crates/ecstore/src/services/rebalance/control.rs +++ b/crates/ecstore/src/services/rebalance/control.rs @@ -327,16 +327,8 @@ impl ECStore { #[tracing::instrument(skip(self, bucktes))] pub async fn init_and_start_rebalance(self: &Arc, bucktes: Vec) -> Result { - let _start_guard = self.start_gate.lock().await; - - let decommission_running = self.is_decommission_running().await; - { - let rebalance_meta = self.rebalance_meta.read().await; - validate_init_rebalance_state(decommission_running, rebalance_meta.as_ref())?; - } - - let id = self.init_rebalance_meta(bucktes).await?; - if let Err(start_err) = self.start_rebalance().await { + let id = self.init_rebalance_start(bucktes).await?; + if let Err(start_err) = self.start_rebalance_for_id(&id).await { if let Err(rollback_err) = self .rollback_rebalance_start_without_worker_for_id(Some(&id), start_err.to_string()) .await @@ -354,6 +346,47 @@ impl ECStore { Ok(id) } + #[tracing::instrument(skip(self, bucktes))] + pub async fn init_rebalance_start(self: &Arc, bucktes: Vec) -> Result { + let _start_guard = self.start_gate.lock().await; + + let decommission_running = self.is_decommission_running().await; + { + let rebalance_meta = self.rebalance_meta.read().await; + validate_init_rebalance_state(decommission_running, rebalance_meta.as_ref())?; + } + + self.init_rebalance_meta(bucktes).await + } + + #[tracing::instrument(skip(self))] + pub async fn start_rebalance_for_id(self: &Arc, expected_id: &str) -> Result<()> { + let _start_guard = self.start_gate.lock().await; + + { + let rebalance_meta = self.rebalance_meta.read().await; + let Some(meta) = rebalance_meta.as_ref() else { + return Err(Error::ConfigNotFound); + }; + if meta.id != expected_id { + return Err(Error::other(format!( + "rebalance metadata changed before start: expected {expected_id}, found {}", + meta.id + ))); + } + if meta.stopped_at.is_some() { + return Err(Error::other(format!("rebalance {expected_id} was stopped before start"))); + } + } + + self.start_rebalance().await + } + + pub async fn rollback_rebalance_start_for_id(self: &Arc, expected_id: Option<&str>, start_error: String) -> Result<()> { + self.rollback_rebalance_start_without_worker_for_id(expected_id, start_error) + .await + } + #[tracing::instrument(skip(self, fi))] pub async fn update_pool_stats(&self, pool_index: usize, bucket: String, fi: &FileInfo) -> Result<()> { self.update_pool_stats_batch(pool_index, bucket, &[fi]).await diff --git a/crates/ecstore/src/services/rebalance/rebalance_unit_tests.rs b/crates/ecstore/src/services/rebalance/rebalance_unit_tests.rs index b4bd196fc..aade4cb28 100644 --- a/crates/ecstore/src/services/rebalance/rebalance_unit_tests.rs +++ b/crates/ecstore/src/services/rebalance/rebalance_unit_tests.rs @@ -2584,21 +2584,7 @@ async fn test_init_and_start_rebalance_rejects_second_start_after_gate() { }], ..Default::default() }; - - let endpoint_pools: crate::layout::endpoints::EndpointServerPools = Vec::new().into(); - let store = Arc::new(crate::store::ECStore { - id: uuid::Uuid::new_v4(), - disk_map: std::collections::HashMap::new(), - pools: Vec::new(), - peer_sys: crate::cluster::rpc::S3PeerSys::new(&endpoint_pools), - pool_meta: tokio::sync::RwLock::new(crate::core::pools::PoolMeta::default()), - rebalance_meta: tokio::sync::RwLock::new(Some(active_meta)), - decommission_cancelers: tokio::sync::RwLock::new(Vec::new()), - start_gate: tokio::sync::Mutex::new(()), - pool_meta_save_gate: tokio::sync::Mutex::new(()), - ctx: crate::runtime::instance::bootstrap_ctx(), - bucket_fence_registry: std::sync::Arc::default(), - }); + let store = test_store_with_rebalance_meta(active_meta); let err = store .init_and_start_rebalance(vec!["bucket".to_string()]) @@ -2608,6 +2594,72 @@ async fn test_init_and_start_rebalance_rejects_second_start_after_gate() { assert!(matches!(err, Error::RebalanceAlreadyRunning)); } +#[tokio::test] +async fn test_start_rebalance_for_id_rejects_changed_metadata() { + let meta = RebalanceMeta { + id: "rebalance-a".to_string(), + pool_stats: vec![RebalanceStats { + participating: true, + info: RebalanceInfo { + status: RebalStatus::Started, + ..Default::default() + }, + ..Default::default() + }], + ..Default::default() + }; + let store = test_store_with_rebalance_meta(meta); + + let err = store + .start_rebalance_for_id("rebalance-b") + .await + .expect_err("staged start must not start changed metadata"); + + assert!(err.to_string().contains("rebalance metadata changed before start")); +} + +#[tokio::test] +async fn test_start_rebalance_for_id_rejects_stopped_metadata() { + let meta = RebalanceMeta { + id: "rebalance-a".to_string(), + stopped_at: Some(OffsetDateTime::now_utc()), + pool_stats: vec![RebalanceStats { + participating: true, + info: RebalanceInfo { + status: RebalStatus::Started, + ..Default::default() + }, + ..Default::default() + }], + ..Default::default() + }; + let store = test_store_with_rebalance_meta(meta); + + let err = store + .start_rebalance_for_id("rebalance-a") + .await + .expect_err("staged start must not restart stopped metadata"); + + assert!(err.to_string().contains("was stopped before start")); +} + +fn test_store_with_rebalance_meta(meta: RebalanceMeta) -> Arc { + let endpoint_pools: crate::layout::endpoints::EndpointServerPools = Vec::new().into(); + Arc::new(crate::store::ECStore { + id: uuid::Uuid::new_v4(), + disk_map: std::collections::HashMap::new(), + pools: Vec::new(), + peer_sys: crate::cluster::rpc::S3PeerSys::new(&endpoint_pools), + pool_meta: tokio::sync::RwLock::new(crate::core::pools::PoolMeta::default()), + rebalance_meta: tokio::sync::RwLock::new(Some(meta)), + decommission_cancelers: tokio::sync::RwLock::new(Vec::new()), + start_gate: tokio::sync::Mutex::new(()), + pool_meta_save_gate: tokio::sync::Mutex::new(()), + ctx: crate::runtime::instance::bootstrap_ctx(), + bucket_fence_registry: std::sync::Arc::default(), + }) +} + #[test] fn test_percent_free_ratio_zero_capacity_is_zero() { assert_eq!(percent_free_ratio(100, 0), 0.0); diff --git a/rustfs/src/admin/handlers/rebalance.rs b/rustfs/src/admin/handlers/rebalance.rs index 5685d3f17..81493dab5 100644 --- a/rustfs/src/admin/handlers/rebalance.rs +++ b/rustfs/src/admin/handlers/rebalance.rs @@ -39,7 +39,7 @@ use rustfs_utils::{ http::{AMZ_REQUEST_ID, REQUEST_ID_HEADER}, }; use s3s::{ - Body, S3Request, S3Response, S3Result, + Body, S3Error, S3ErrorCode, S3Request, S3Response, S3Result, header::{CONTENT_LENGTH, CONTENT_TYPE}, s3_error, }; @@ -102,6 +102,10 @@ fn rebalance_start_rollback_error(start_err: &str, rollback_result: &Result<(), } } +fn rebalance_internal_error(message: impl Into) -> S3Error { + S3Error::with_message(S3ErrorCode::InternalError, message.into()) +} + fn rebalance_rollback_stop_failure_message(rebalance_id: &str, failures: &[String]) -> String { format!("cluster stop_rebalance rollback for {rebalance_id} partial: {}", failures.join("; ")) } @@ -128,6 +132,28 @@ fn rebalance_rollback_failure_message( failures.join("; ") } +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum RebalanceStartStep { + PropagateFence, + StartLocal, + PropagateWorkers, +} + +const DISTRIBUTED_REBALANCE_START_STEPS: [RebalanceStartStep; 3] = [ + RebalanceStartStep::PropagateFence, + RebalanceStartStep::StartLocal, + RebalanceStartStep::PropagateWorkers, +]; +const LOCAL_REBALANCE_START_STEPS: [RebalanceStartStep; 1] = [RebalanceStartStep::StartLocal]; + +fn rebalance_start_steps(has_notification_sys: bool) -> &'static [RebalanceStartStep] { + if has_notification_sys { + &DISTRIBUTED_REBALANCE_START_STEPS + } else { + &LOCAL_REBALANCE_START_STEPS + } +} + async fn rollback_cluster_rebalance_start( store: &Arc, notification_sys: Option<&NotificationSys>, @@ -177,6 +203,50 @@ async fn rollback_cluster_rebalance_start( Ok(()) } +async fn rollback_rebalance_start_for_admin( + store: &Arc, + notification_sys: Option<&NotificationSys>, + rebalance_id: &str, + start_err: &str, + request_id: &str, + actor: &str, + remote_addr: &str, +) -> S3Result<()> { + let rollback_result = rollback_cluster_rebalance_start(store, notification_sys, rebalance_id).await; + let rollback_label = rollback_result_label(&rollback_result); + match &rollback_result { + Ok(_) => info!( + event = EVENT_ADMIN_REBALANCE_STATE, + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_REBALANCE, + action = "start", + result = rollback_label, + request_id = %request_id, + actor = %actor, + remote_addr = %remote_addr, + rebalance_id = %rebalance_id, + propagation_error = %start_err, + "admin rebalance state" + ), + Err(rollback_err) => error!( + event = EVENT_ADMIN_REBALANCE_STATE, + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_REBALANCE, + action = "start", + result = rollback_label, + request_id = %request_id, + actor = %actor, + remote_addr = %remote_addr, + rebalance_id = %rebalance_id, + propagation_error = %start_err, + rollback_error = %rollback_err, + "admin rebalance state" + ), + } + + Err(rebalance_internal_error(rebalance_start_rollback_error(start_err, &rollback_result))) +} + pub fn register_rebalance_route(r: &mut S3Router) -> std::io::Result<()> { r.insert( Method::POST, @@ -473,7 +543,7 @@ impl Operation for RebalanceStart { let buckets: Vec = bucket_infos.into_iter().map(|bucket| bucket.name).collect(); - let id = match store.init_and_start_rebalance(buckets).await { + let id = match store.init_rebalance_start(buckets).await { Ok(id) => id, Err(StorageError::DecommissionAlreadyRunning) => { log_rebalance_request_rejected("start", "decommission_in_progress", &request_id, &actor, &remote_addr); @@ -484,7 +554,7 @@ impl Operation for RebalanceStart { return Err(s3_error!(OperationAborted, "rebalance is already in progress")); } Err(e) => { - return Err(s3_error!(InternalError, "failed to start rebalance: {}", e)); + return Err(s3_error!(InternalError, "failed to initialize rebalance: {}", e)); } }; @@ -493,80 +563,186 @@ impl Operation for RebalanceStart { component = LOG_COMPONENT_ADMIN, subsystem = LOG_SUBSYSTEM_REBALANCE, action = "start", - state = "started", + state = "metadata_initialized", request_id = %request_id, actor = %actor, remote_addr = %remote_addr, rebalance_id = %id, "admin rebalance state" ); - if let Some(notification_sys) = current_notification_system() { - info!( - event = EVENT_ADMIN_REBALANCE_STATE, - component = LOG_COMPONENT_ADMIN, - subsystem = LOG_SUBSYSTEM_REBALANCE, - action = "start", - state = "propagation_started", - request_id = %request_id, - actor = %actor, - remote_addr = %remote_addr, - rebalance_id = %id, - "admin rebalance state" - ); - if let Err(err) = notification_sys.load_rebalance_meta(true).await { - error!( - event = EVENT_ADMIN_REBALANCE_STATE, - component = LOG_COMPONENT_ADMIN, - subsystem = LOG_SUBSYSTEM_REBALANCE, - action = "start", - result = "propagation_failed", - request_id = %request_id, - actor = %actor, - remote_addr = %remote_addr, - rebalance_id = %id, - error = %err, - "admin rebalance state" - ); + let notification_sys = current_notification_system(); + for step in rebalance_start_steps(notification_sys.is_some()) { + match step { + RebalanceStartStep::PropagateFence => { + if let Some(notification_sys) = notification_sys.as_ref() { + info!( + event = EVENT_ADMIN_REBALANCE_STATE, + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_REBALANCE, + action = "start", + state = "fence_propagation_started", + request_id = %request_id, + actor = %actor, + remote_addr = %remote_addr, + rebalance_id = %id, + "admin rebalance state" + ); + if let Err(err) = notification_sys.load_rebalance_meta(false).await { + error!( + event = EVENT_ADMIN_REBALANCE_STATE, + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_REBALANCE, + action = "start", + result = "fence_propagation_failed", + request_id = %request_id, + actor = %actor, + remote_addr = %remote_addr, + rebalance_id = %id, + error = %err, + "admin rebalance state" + ); - let start_err = err.to_string(); - let rollback_result = rollback_cluster_rebalance_start(&store, Some(¬ification_sys), &id).await; - let rollback_label = rollback_result_label(&rollback_result); - match &rollback_result { - Ok(_) => info!( - event = EVENT_ADMIN_REBALANCE_STATE, - component = LOG_COMPONENT_ADMIN, - subsystem = LOG_SUBSYSTEM_REBALANCE, - action = "start", - result = rollback_label, - request_id = %request_id, - actor = %actor, - remote_addr = %remote_addr, - rebalance_id = %id, - propagation_error = %start_err, - "admin rebalance state" - ), - Err(rollback_err) => error!( - event = EVENT_ADMIN_REBALANCE_STATE, - component = LOG_COMPONENT_ADMIN, - subsystem = LOG_SUBSYSTEM_REBALANCE, - action = "start", - result = rollback_label, - request_id = %request_id, - actor = %actor, - remote_addr = %remote_addr, - rebalance_id = %id, - propagation_error = %start_err, - rollback_error = %rollback_err, - "admin rebalance state" - ), + let start_err = err.to_string(); + rollback_rebalance_start_for_admin( + &store, + Some(notification_sys), + &id, + &start_err, + &request_id, + &actor, + &remote_addr, + ) + .await?; + } + } } + RebalanceStartStep::StartLocal => { + if let Err(err) = store.start_rebalance_for_id(&id).await { + error!( + event = EVENT_ADMIN_REBALANCE_STATE, + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_REBALANCE, + action = "start", + result = "local_start_failed", + request_id = %request_id, + actor = %actor, + remote_addr = %remote_addr, + rebalance_id = %id, + error = %err, + "admin rebalance state" + ); - return Err(s3_error!( - InternalError, - "{}", - rebalance_start_rollback_error(&start_err, &rollback_result) - )); + let start_err = err.to_string(); + if let Err(rollback_err) = store.rollback_rebalance_start_for_id(Some(&id), start_err.clone()).await { + error!( + event = EVENT_ADMIN_REBALANCE_STATE, + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_REBALANCE, + action = "start", + result = "local_start_rollback_failed", + request_id = %request_id, + actor = %actor, + remote_addr = %remote_addr, + rebalance_id = %id, + start_error = %start_err, + rollback_error = %rollback_err, + "admin rebalance state" + ); + return Err(rebalance_internal_error(format!( + "failed to start rebalance after metadata initialized for {id}; rollback failed: {rollback_err}" + ))); + } + + info!( + event = EVENT_ADMIN_REBALANCE_STATE, + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_REBALANCE, + action = "start", + result = "local_start_rollback_success", + request_id = %request_id, + actor = %actor, + remote_addr = %remote_addr, + rebalance_id = %id, + start_error = %start_err, + "admin rebalance state" + ); + if let Some(notification_sys) = notification_sys.as_ref() { + let terminal_reload_attempt_at = OffsetDateTime::now_utc(); + let terminal_reload_failures = match notification_sys.load_rebalance_meta_failures(false).await { + Ok(failures) => failures, + Err(err) => vec![format!("terminal rebalance reload rollback for {id} failed: {err}")], + }; + if !terminal_reload_failures.is_empty() { + let record = RebalanceStopPropagationRecord { + stop_attempt_at: None, + stop_failures: Vec::new(), + terminal_reload_attempt_at: Some(terminal_reload_attempt_at), + terminal_reload_failures: terminal_reload_failures.clone(), + }; + store.record_rebalance_stop_propagation(record).await.map_err(|err| { + rebalance_internal_error(format!( + "failed to persist rebalance local-start rollback propagation metadata: {err}" + )) + })?; + return Err(rebalance_internal_error(format!( + "failed to start rebalance after metadata initialized for {}; local metadata was finalized as failed, but terminal peer reload was incomplete: {}", + id, + rebalance_rollback_terminal_reload_failure_message(&id, &terminal_reload_failures) + ))); + } + } + return Err(rebalance_internal_error(format!( + "failed to start rebalance after metadata initialized for {id}; local metadata was finalized as failed: {start_err}" + ))); + } + } + RebalanceStartStep::PropagateWorkers => { + if let Some(notification_sys) = notification_sys.as_ref() { + info!( + event = EVENT_ADMIN_REBALANCE_STATE, + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_REBALANCE, + action = "start", + state = "worker_propagation_started", + request_id = %request_id, + actor = %actor, + remote_addr = %remote_addr, + rebalance_id = %id, + "admin rebalance state" + ); + if let Err(err) = notification_sys.load_rebalance_meta(true).await { + error!( + event = EVENT_ADMIN_REBALANCE_STATE, + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_REBALANCE, + action = "start", + result = "worker_propagation_failed", + request_id = %request_id, + actor = %actor, + remote_addr = %remote_addr, + rebalance_id = %id, + error = %err, + "admin rebalance state" + ); + + let start_err = err.to_string(); + rollback_rebalance_start_for_admin( + &store, + Some(notification_sys), + &id, + &start_err, + &request_id, + &actor, + &remote_addr, + ) + .await?; + } + } + } } + } + + if notification_sys.is_some() { info!( event = EVENT_ADMIN_REBALANCE_STATE, component = LOG_COMPONENT_ADMIN, @@ -902,10 +1078,11 @@ mod rebalance_handler_tests { use super::build_rebalance_pool_progress; use super::calculate_rebalance_progress; use super::{ - RebalPoolProgress, RebalanceAdminStatus, RebalancePoolStatus, RebalanceStopPropagationStatus, + RebalPoolProgress, RebalanceAdminStatus, RebalancePoolStatus, RebalanceStartStep, RebalanceStopPropagationStatus, build_rebalance_admin_status, build_rebalance_pool_statuses, build_rebalance_stop_propagation_status, rebalance_pool_used, rebalance_query_present, rebalance_remaining_buckets, rebalance_rollback_failure_message, - rebalance_rollback_stop_failure_message, rebalance_start_rollback_error, rebalance_used_pct, rollback_result_label, + rebalance_rollback_stop_failure_message, rebalance_start_rollback_error, rebalance_start_steps, rebalance_used_pct, + rollback_result_label, }; use crate::admin::storage_api::rebalance::{ DiskStat, RebalStatus, RebalanceCleanupWarningEntry, RebalanceCleanupWarnings, RebalanceInfo, RebalanceMeta, @@ -994,6 +1171,23 @@ mod rebalance_handler_tests { assert_eq!(rollback_result_label(&rollback_result), "rollback_failed"); } + #[test] + fn test_distributed_rebalance_start_fences_peers_before_workers() { + assert_eq!( + rebalance_start_steps(true), + &[ + RebalanceStartStep::PropagateFence, + RebalanceStartStep::StartLocal, + RebalanceStartStep::PropagateWorkers + ] + ); + } + + #[test] + fn test_local_rebalance_start_has_no_peer_propagation_steps() { + assert_eq!(rebalance_start_steps(false), &[RebalanceStartStep::StartLocal]); + } + #[test] fn test_calculate_rebalance_progress_stopped_by_end_time() { let start = OffsetDateTime::from_unix_timestamp(1_000).unwrap(); diff --git a/scripts/check_s3s_footprint.sh b/scripts/check_s3s_footprint.sh index 31a58c2e3..255c084fa 100755 --- a/scripts/check_s3s_footprint.sh +++ b/scripts/check_s3s_footprint.sh @@ -24,7 +24,7 @@ cd "$(dirname "$0")/.." # Baselines verified on 2026-08-06. Lower-only; see header. S3S_IMPORT_FILES_BASELINE=236 -S3_ERROR_LINES_BASELINE=1679 +S3_ERROR_LINES_BASELINE=1678 S3S_PATH_PATTERN='(^|[^"[:alnum:]_])s3s::' TMP_DIR="$(mktemp -d)"