From 685564019297c8d1f0f91b5d79fc27801035f73d Mon Sep 17 00:00:00 2001 From: cxymds Date: Mon, 7 Sep 2026 22:30:51 +0800 Subject: [PATCH] fix(rebalance): preserve explicit stop intent and real failures (#7410) --- .../ecstore/src/services/rebalance/control.rs | 159 +++++++++++++++++- .../ecstore/src/services/rebalance/entry.rs | 27 ++- crates/ecstore/src/services/rebalance/meta.rs | 8 +- .../rebalance/rebalance_unit_tests.rs | 88 +++++++--- .../ecstore/src/services/rebalance/runtime.rs | 2 +- .../ecstore/src/services/rebalance/types.rs | 4 + .../ecstore/src/services/rebalance/worker.rs | 36 ++-- 7 files changed, 252 insertions(+), 72 deletions(-) diff --git a/crates/ecstore/src/services/rebalance/control.rs b/crates/ecstore/src/services/rebalance/control.rs index 1f95f4f31..6d875bd9d 100644 --- a/crates/ecstore/src/services/rebalance/control.rs +++ b/crates/ecstore/src/services/rebalance/control.rs @@ -300,11 +300,13 @@ pub(super) fn ensure_rebalance_worker_active(meta: Option<&RebalanceMeta>, expec let Some(meta) = meta else { return Err(rebalance_metadata_not_initialized_error(stage)); }; - if meta.stopped_at.is_some() - || meta - .cancel - .as_ref() - .is_some_and(tokio_util::sync::CancellationToken::is_cancelled) + if meta.stopped_at.is_some() || meta.stop_requested { + return Err(Error::OperationCanceled); + } + if meta + .cancel + .as_ref() + .is_some_and(tokio_util::sync::CancellationToken::is_cancelled) || !is_rebalance_conflicting_with_decommission(meta) { return Err(Error::other(format!("inactive rebalance worker rejected during {stage}: {expected_id}"))); @@ -629,6 +631,7 @@ impl ECStore { if let Some(meta) = rebalance_meta.as_mut() && is_rebalance_conflicting_with_decommission(meta) { + meta.stop_requested = true; meta.cancel .get_or_insert_with(tokio_util::sync::CancellationToken::new) .cancel(); @@ -643,12 +646,13 @@ impl ECStore { let Some(meta) = rebalance_meta.as_mut() else { return Ok(None); }; - if !is_rebalance_conflicting_with_decommission(meta) { + if meta.stopped_at.is_some() || (!is_rebalance_conflicting_with_decommission(meta) && !meta.stop_requested) { return Ok(None); } if meta.id.is_empty() { return Err(Error::other("active rebalance metadata has no activation id")); } + meta.stop_requested = true; meta.cancel .get_or_insert_with(tokio_util::sync::CancellationToken::new) .cancel(); @@ -673,7 +677,13 @@ impl ECStore { let movement_changed = rebalance_movement_snapshot_changed(self.rebalance_meta.read().await.as_ref(), &meta); { let mut rebalance_meta = self.rebalance_meta.write().await; - + if let Some(current) = rebalance_meta.as_ref() + && current.id == meta.id + { + meta.cancel = current.cancel.clone(); + meta.activation_gate = Arc::clone(¤t.activation_gate); + meta.stop_requested = current.stop_requested; + } *rebalance_meta = Some(meta); drop(rebalance_meta); @@ -1188,11 +1198,12 @@ impl ECStore { let meta = rebalance_meta .as_mut() .ok_or_else(|| rebalance_metadata_not_initialized_error("cancel rebalance admission"))?; - if meta.stopped_at.is_some() || !is_rebalance_conflicting_with_decommission(meta) { + if meta.stopped_at.is_some() || (!is_rebalance_conflicting_with_decommission(meta) && !meta.stop_requested) { return Err(Error::other(format!( "inactive rebalance rejected while cancelling admission: {expected_id}" ))); } + meta.stop_requested = true; meta.cancel .get_or_insert_with(tokio_util::sync::CancellationToken::new) .cancel(); @@ -1213,6 +1224,7 @@ impl ECStore { ensure_rebalance_run_id(rebalance_meta.as_ref(), expected_id, "stop rebalance")?; } rebalance_meta.as_mut().map(|meta| { + meta.stop_requested |= is_rebalance_conflicting_with_decommission(meta); let cancel = meta.cancel.get_or_insert_with(tokio_util::sync::CancellationToken::new); cancel.cancel(); Arc::clone(&meta.activation_gate) @@ -1377,6 +1389,79 @@ mod tests { probe.wait_until_attempted().await; } + #[test] + fn rebalance_stop_classification_checks_identity_and_explicit_intent() { + let mut meta = RebalanceMeta { + id: "current".to_string(), + cancel: Some(tokio_util::sync::CancellationToken::new()), + pool_stats: vec![RebalanceStats { + participating: true, + info: RebalanceInfo { + status: RebalStatus::Started, + ..Default::default() + }, + ..Default::default() + }], + ..Default::default() + }; + ensure_rebalance_worker_active(Some(&meta), "current", "test").expect("active worker"); + meta.cancel.as_ref().unwrap().cancel(); + assert!( + !matches!( + ensure_rebalance_worker_active(Some(&meta), "current", "test"), + Err(Error::OperationCanceled) + ), + "a sibling failure is not an operator stop" + ); + meta.stop_requested = true; + assert!(matches!( + ensure_rebalance_worker_active(Some(&meta), "current", "test"), + Err(Error::OperationCanceled) + )); + assert!( + !matches!(ensure_rebalance_worker_active(Some(&meta), "old", "test"), Err(Error::OperationCanceled)), + "stale identity remains a failure even during stop" + ); + assert!(!matches!( + ensure_rebalance_worker_active(None, "current", "test"), + Err(Error::OperationCanceled) + )); + } + + #[tokio::test] + async fn rebalance_stop_intent_does_not_survive_replacement_run_reload() { + let (_temp_dirs, store) = crate::services::rebalance::test_store_with_persisted_rebalance_meta(RebalanceMeta { + id: "replacement".to_string(), + pool_stats: vec![RebalanceStats { + participating: true, + info: RebalanceInfo { + status: RebalStatus::Completed, + ..Default::default() + }, + ..Default::default() + }], + ..Default::default() + }) + .await; + let previous_gate = { + let mut meta = store.rebalance_meta.write().await; + let meta = meta.as_mut().unwrap(); + meta.id = "previous".to_string(); + meta.stop_requested = true; + let cancel = tokio_util::sync::CancellationToken::new(); + cancel.cancel(); + meta.cancel = Some(cancel); + Arc::clone(&meta.activation_gate) + }; + store.load_rebalance_meta().await.expect("reload replacement run"); + let meta = store.rebalance_meta.read().await; + let meta = meta.as_ref().unwrap(); + assert_eq!(meta.id, "replacement"); + assert!(!meta.stop_requested); + assert!(meta.cancel.is_none()); + assert!(!Arc::ptr_eq(&previous_gate, &meta.activation_gate)); + } + #[tokio::test] async fn cancel_rebalance_admission_is_id_checked_and_idempotent() { let rebalance_id = "rebalance-admission-current"; @@ -1415,6 +1500,59 @@ mod tests { .await .expect("retrying admission cancellation should be idempotent"); assert!(cancel.is_cancelled()); + let err = store + .update_pool_stats_batch_for_rebalance(0, "bucket".to_string(), &[&FileInfo::default()], rebalance_id) + .await + .expect_err("stop racing with a final stats update must cancel that update"); + assert!(matches!(err, Error::OperationCanceled), "operator stop lost its cancellation type: {err}"); + } + + #[tokio::test] + async fn prepare_rebalance_stop_preserves_intent_when_worker_stops_before_reload() { + let id = "stop-worker-before-reload"; + let cancel = tokio_util::sync::CancellationToken::new(); + let (_temp_dirs, store) = crate::services::rebalance::test_store_with_persisted_rebalance_meta(RebalanceMeta { + id: id.to_string(), + cancel: Some(cancel.clone()), + pool_stats: vec![RebalanceStats { + participating: true, + info: RebalanceInfo { + status: RebalStatus::Stopped, + ..Default::default() + }, + ..Default::default() + }], + ..Default::default() + }) + .await; + let gate = { + let mut meta = store.rebalance_meta.write().await; + let meta = meta.as_mut().expect("local rebalance metadata"); + meta.pool_stats[0].info.status = RebalStatus::Started; + Arc::clone(&meta.activation_gate) + }; + assert_eq!( + store.prepare_rebalance_stop().await.expect("prepare the same run stop"), + Some(id.to_string()) + ); + { + let meta = store.rebalance_meta.read().await; + let meta = meta.as_ref().expect("reloaded stop target"); + assert!(Arc::ptr_eq(&gate, &meta.activation_gate), "reload must retain the drained run's gate"); + assert!(meta.cancel.as_ref().is_some_and(|token| token.is_cancelled())); + } + store + .stop_rebalance_for_id(Some(id)) + .await + .expect("finish the stop after the worker's terminal event"); + store + .load_rebalance_meta() + .await + .expect("reload the acknowledged durable stop"); + let meta = store.rebalance_meta.read().await; + let meta = meta.as_ref().expect("durable stopped metadata"); + assert!(meta.stopped_at.is_some(), "a successful stop must retain its durable timestamp"); + assert_eq!(meta.pool_stats[0].info.status, RebalStatus::Stopped); } #[tokio::test] @@ -2031,7 +2169,10 @@ mod tests { let err = acquire_persisted_rebalance_run_guard(set_disks, active.id.as_str(), "cross-node stale snapshot") .await .expect_err("persisted stop must fence a node that missed stop propagation"); - assert!(err.to_string().contains("inactive rebalance worker rejected")); + assert!( + matches!(err, Error::OperationCanceled), + "a durable remote stop cancels the same run: {err}" + ); } #[test] diff --git a/crates/ecstore/src/services/rebalance/entry.rs b/crates/ecstore/src/services/rebalance/entry.rs index 882abde01..6adb64618 100644 --- a/crates/ecstore/src/services/rebalance/entry.rs +++ b/crates/ecstore/src/services/rebalance/entry.rs @@ -19,11 +19,11 @@ use super::meta::{ use super::migration::{RebalanceMigrationBackend, migrate_entry_version}; use super::worker::{ RebalanceEntryCleanupResult, RebalanceEntryTask, load_rebalance_bucket_configs, rebalance_max_attempts, - resolve_rebalance_bucket_error, resolve_rebalance_entry_cleanup_delete_result, resolve_rebalance_file_info_versions_result, - resolve_rebalance_migrate_result_error, resolve_rebalance_stats_update_result, resolve_rebalance_worker_result, - run_rebalance_listing_with_retry, should_cleanup_rebalance_source_entry, should_count_rebalance_version_complete, - should_defer_rebalance_entry_failure, should_skip_rebalance_delete_marker, wait_rebalance_entry_tasks, - with_rebalance_entry_context, + record_rebalance_error, resolve_rebalance_bucket_error, resolve_rebalance_entry_cleanup_delete_result, + resolve_rebalance_file_info_versions_result, resolve_rebalance_migrate_result_error, resolve_rebalance_stats_update_result, + resolve_rebalance_worker_result, run_rebalance_listing_with_retry, should_cleanup_rebalance_source_entry, + should_count_rebalance_version_complete, should_defer_rebalance_entry_failure, should_skip_rebalance_delete_marker, + wait_rebalance_entry_tasks, with_rebalance_entry_context, }; use super::{ EVENT_REBALANCE_BUCKET, EVENT_REBALANCE_ENTRY, EVENT_REBALANCE_STATE, LOG_COMPONENT_ECSTORE, LOG_SUBSYSTEM_REBALANCE, @@ -676,10 +676,8 @@ impl ECStore { } error!("rebalance_entry: data movement admission failed: {err}"); let mut first_err = entry_error.lock().await; - if first_err.is_none() { - *first_err = Some(err); - callback_rx.cancel(); - } + record_rebalance_error(&mut first_err, err); + callback_rx.cancel(); return; } @@ -721,10 +719,8 @@ impl ECStore { if let Err(err) = &result { error!("rebalance_entry: rebalance entry failed: {err}"); let mut first_err = entry_error.lock().await; - if first_err.is_none() { - *first_err = Some(err.clone()); - callback_rx.cancel(); - } + record_rebalance_error(&mut first_err, err.clone()); + callback_rx.cancel(); } debug!( event = EVENT_REBALANCE_ENTRY, @@ -793,10 +789,7 @@ impl ECStore { deferred_error = Some(last_error); } Ok(_) => {} - Err(err) if worker_error.is_none() => { - worker_error = Some(err); - } - Err(_) => {} + Err(err) => record_rebalance_error(&mut worker_error, err), } } let entry_error = entry_error.lock().await.clone(); diff --git a/crates/ecstore/src/services/rebalance/meta.rs b/crates/ecstore/src/services/rebalance/meta.rs index 678be730a..8e383065d 100644 --- a/crates/ecstore/src/services/rebalance/meta.rs +++ b/crates/ecstore/src/services/rebalance/meta.rs @@ -643,16 +643,12 @@ pub(super) fn should_skip_start_rebalance(cancel_attached: bool, in_progress: bo 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) + (meta_stopped || status == RebalStatus::Stopped) && matches!(terminal_event, RebalanceTerminalEvent::Completed { .. }) } pub(super) fn resolve_rebalance_participants(pool_stats: &[RebalanceStats], pool_count: usize) -> Vec { @@ -920,7 +916,7 @@ pub(super) fn clear_rebalance_cancel_token(meta: Option<&mut RebalanceMeta>) -> pub(super) fn stop_rebalance_state(meta: &mut RebalanceMeta, now: OffsetDateTime) { clear_rebalance_cancel_token(Some(meta)); - if meta.stopped_at.is_none() && is_rebalance_in_progress(meta) { + if meta.stopped_at.is_none() && (meta.stop_requested || is_rebalance_in_progress(meta)) { apply_stopped_at(meta, now); } else if meta.stopped_at.is_some() { mark_started_rebalance_pools_stopping(meta); diff --git a/crates/ecstore/src/services/rebalance/rebalance_unit_tests.rs b/crates/ecstore/src/services/rebalance/rebalance_unit_tests.rs index a544ed733..529ed5914 100644 --- a/crates/ecstore/src/services/rebalance/rebalance_unit_tests.rs +++ b/crates/ecstore/src/services/rebalance/rebalance_unit_tests.rs @@ -19,14 +19,14 @@ use super::meta::{ 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_meta_replaceable_for_new_id, 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, rebalance_requires_worker_activation, 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_init_rebalance_state, validate_start_rebalance_state, + is_rebalance_in_progress, is_rebalance_meta_replaceable_for_new_id, 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, + rebalance_requires_worker_activation, 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_init_rebalance_state, validate_start_rebalance_state, }; use super::migration::{ MigrationBackend, MigrationVersionResult, migrate_entry_version, migrate_entry_version_with_retry_wait, @@ -1676,6 +1676,30 @@ fn test_resolve_rebalance_stats_update_result_passthrough() { assert!(resolve_rebalance_stats_update_result(Ok(()), 0, "bucket", "object").is_ok()); } +#[test] +fn test_rebalance_stop_preserves_cancellation_through_entry_context() { + let err = resolve_rebalance_stats_update_result(Err(Error::OperationCanceled), 0, "bucket", "object") + .expect_err("canceled stats update"); + let err = with_rebalance_entry_context("stats", "bucket", "object", err); + assert!(matches!(err, Error::OperationCanceled)); + assert!(matches!( + classify_rebalance_terminal_event(Some(Err(err)), OffsetDateTime::now_utc()), + RebalanceTerminalEvent::Stopped { .. } + )); +} + +#[tokio::test] +async fn test_rebalance_stop_does_not_hide_later_entry_failure() { + let tasks = Arc::new(tokio::sync::Mutex::new(vec![ + tokio::spawn(async { Err(Error::OperationCanceled) }), + tokio::spawn(async { Err(Error::ErasureWriteQuorum) }), + ])); + let err = wait_rebalance_entry_tasks(0, tasks) + .await + .expect_err("entry I/O failure must survive sibling cancellation"); + assert!(matches!(err, Error::ErasureWriteQuorum)); +} + #[test] fn test_resolve_rebalance_stats_update_result_wraps_error_context() { let err = resolve_rebalance_stats_update_result(Err(Error::SlowDown), 2, "bucket-a", "obj.txt") @@ -2365,9 +2389,9 @@ fn test_resolve_rebalance_terminal_error_wraps_signal_failure_context() { } #[test] -fn test_resolve_rebalance_bucket_error_prefers_entry_error() { +fn test_resolve_rebalance_bucket_error_prefers_real_failure_over_entry_cancellation() { let err = resolve_rebalance_bucket_error(Some(Error::OperationCanceled), Some(Error::SlowDown)).unwrap_err(); - assert!(matches!(err, Error::OperationCanceled)); + assert!(matches!(err, Error::SlowDown)); } #[test] @@ -2512,19 +2536,6 @@ fn test_apply_rebalance_terminal_event_stopped_clears_error() { assert_eq!(last_error, None); } -#[test] -fn test_is_rebalance_stopped_terminal_event_only_matches_stopped_variant() { - let stopped = RebalanceTerminalEvent::Stopped { - msg: "stopped".to_string(), - }; - let completed = RebalanceTerminalEvent::Completed { - msg: "completed".to_string(), - }; - - assert!(is_rebalance_stopped_terminal_event(&stopped)); - assert!(!is_rebalance_stopped_terminal_event(&completed)); -} - #[test] fn test_should_preserve_rebalance_stopped_state_when_meta_marked_stopped() { let event = RebalanceTerminalEvent::Completed { @@ -2535,13 +2546,14 @@ fn test_should_preserve_rebalance_stopped_state_when_meta_marked_stopped() { } #[test] -fn test_should_preserve_rebalance_stopped_state_when_pool_already_stopped() { +fn test_rebalance_stop_does_not_hide_real_terminal_failure() { let event = RebalanceTerminalEvent::Failed { msg: "failed".to_string(), last_error: "boom".to_string(), }; - assert!(should_preserve_rebalance_stopped_state(false, RebalStatus::Stopped, &event)); + assert!(!should_preserve_rebalance_stopped_state(false, RebalStatus::Stopped, &event)); + assert!(!should_preserve_rebalance_stopped_state(true, RebalStatus::Started, &event)); } #[test] @@ -2716,6 +2728,32 @@ async fn test_start_rebalance_for_id_rejects_stopped_metadata() { assert!(err.to_string().contains("was stopped before start")); } +#[test] +fn test_rebalance_stop_intent_blocks_activation_before_durable_timestamp() { + let mut meta = RebalanceMeta { + id: "stopping".to_string(), + stop_requested: true, + pool_stats: vec![RebalanceStats { + participating: true, + buckets: vec!["pending".to_string()], + info: RebalanceInfo { + status: RebalStatus::Started, + ..Default::default() + }, + ..Default::default() + }], + ..Default::default() + }; + let outcome = commit_local_rebalance_worker_activation(&mut meta, "stopping", CancellationToken::new()) + .expect("stop must prevent activation without a new error"); + assert_eq!(outcome, RebalanceLocalActivationOutcome::NotStartedTerminal); + assert!(meta.cancel.is_none()); + assert!(meta.stopped_at.is_none()); + let bytes = rmp_serde::to_vec_named(&meta).expect("encode legacy-compatible metadata"); + let reloaded: RebalanceMeta = rmp_serde::from_slice(&bytes).expect("decode metadata"); + assert!(!reloaded.stop_requested, "operator intent is local, not a new persisted field"); +} + #[test] fn test_stopped_activation_state_prevents_worker_token_commit() { let mut meta = RebalanceMeta { diff --git a/crates/ecstore/src/services/rebalance/runtime.rs b/crates/ecstore/src/services/rebalance/runtime.rs index f85184742..f3f730d76 100644 --- a/crates/ecstore/src/services/rebalance/runtime.rs +++ b/crates/ecstore/src/services/rebalance/runtime.rs @@ -57,7 +57,7 @@ pub(super) fn commit_local_rebalance_worker_activation( meta.id ))); } - if meta.stopped_at.is_some() || !is_rebalance_in_progress(meta) { + if meta.stopped_at.is_some() || meta.stop_requested || !is_rebalance_in_progress(meta) { return Ok(RebalanceLocalActivationOutcome::NotStartedTerminal); } meta.cancel = Some(cancel); diff --git a/crates/ecstore/src/services/rebalance/types.rs b/crates/ecstore/src/services/rebalance/types.rs index a0b04bd4d..f3bcdbab3 100644 --- a/crates/ecstore/src/services/rebalance/types.rs +++ b/crates/ecstore/src/services/rebalance/types.rs @@ -143,6 +143,10 @@ pub struct DiskStat { pub struct RebalanceMeta { #[serde(skip)] pub cancel: Option, // To be invoked on rebalance-stop + /// Local operator intent, scoped to this run ID; a worker failure also cancels + /// `cancel`, so the token alone cannot identify an administrative stop. + #[serde(skip)] + pub stop_requested: bool, #[serde(skip)] pub activation_gate: std::sync::Arc>, #[serde(skip)] diff --git a/crates/ecstore/src/services/rebalance/worker.rs b/crates/ecstore/src/services/rebalance/worker.rs index 38f15bc78..3f10c7b5e 100644 --- a/crates/ecstore/src/services/rebalance/worker.rs +++ b/crates/ecstore/src/services/rebalance/worker.rs @@ -38,6 +38,17 @@ pub(super) fn resolve_rebalance_worker_result( pub(super) type RebalanceEntryTask = tokio::task::JoinHandle>; +/// Preserve the first real failure even when another task observes cancellation +/// first. Cancellation is an outcome only when no entry or worker failed. +pub(super) fn record_rebalance_error(first_error: &mut Option, err: Error) { + if first_error + .as_ref() + .is_none_or(|first| is_err_operation_canceled(first) && !is_err_operation_canceled(&err)) + { + *first_error = Some(err); + } +} + #[derive(Debug, Clone, PartialEq, Eq)] pub(super) enum RebalanceEntryCleanupResult { Completed { warning: Option }, @@ -65,16 +76,12 @@ pub(super) async fn wait_rebalance_entry_tasks( } Ok(Err(err)) => { error!("rebalance entry task failed for set {}: {}", set_idx, err); - if first_error.is_none() { - first_error = Some(err); - } + record_rebalance_error(&mut first_error, err); } Err(err) => { let err = Error::other(format!("rebalance entry task join error for set {set_idx}: {err}")); error!("{}", err); - if first_error.is_none() { - first_error = Some(err); - } + record_rebalance_error(&mut first_error, err); } } } @@ -135,6 +142,9 @@ pub(super) fn resolve_rebalance_stats_update_result( object_name: &str, ) -> Result<()> { result.map_err(|err| { + if is_err_operation_canceled(&err) { + return err; + } Error::other(format!( "rebalance stats update failed for pool {pool_idx} bucket {bucket} object {object_name}: {err}" )) @@ -214,16 +224,11 @@ pub(super) fn resolve_rebalance_terminal_error(primary_err: Error, signal_result } } -pub(super) fn resolve_rebalance_bucket_error(entry_error: Option, worker_error: Option) -> Result<()> { - if let Some(err) = entry_error { - return Err(err); - } - +pub(super) fn resolve_rebalance_bucket_error(mut entry_error: Option, worker_error: Option) -> Result<()> { if let Some(err) = worker_error { - return Err(err); + record_rebalance_error(&mut entry_error, err); } - - Ok(()) + entry_error.map_or(Ok(()), Err) } pub(super) fn resolve_rebalance_bucket_result( @@ -362,6 +367,9 @@ pub(super) fn ensure_rebalance_listing_disks_available(has_disks: bool, bucket: } pub(super) fn with_rebalance_entry_context(stage: &str, bucket: &str, object_name: &str, err: Error) -> Error { + if is_err_operation_canceled(&err) { + return err; + } Error::other(format!("rebalance entry {stage} failed for {bucket}/{object_name}: {err}")) }