gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-09-08 04:58:12 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(rebalance): preserve explicit stop intent and real failures (#7410)

Browse Source
This commit is contained in:
cxymds
2026-09-07 22:30:51 +08:00
committed by GitHub
parent f11697d6e2
commit 6855640192
7 changed files with 252 additions and 72 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/services/rebalance/control.rs
+150 -9
View File
@@ -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(&current.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]
crates/ecstore/src/services/rebalance/entry.rs
+10 -17
View File
@@ -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();
crates/ecstore/src/services/rebalance/meta.rs
+2 -6
View File
@@ -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<bool> {
@@ -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);
crates/ecstore/src/services/rebalance/rebalance_unit_tests.rs
+63 -25
View File
@@ -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 {
crates/ecstore/src/services/rebalance/runtime.rs
+1 -1
View File
@@ -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);
crates/ecstore/src/services/rebalance/types.rs
+4
View File
@@ -143,6 +143,10 @@ pub struct DiskStat {
pub struct RebalanceMeta {
#[serde(skip)]
pub cancel: Option<CancellationToken>, // 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<tokio::sync::RwLock<()>>,
#[serde(skip)]
crates/ecstore/src/services/rebalance/worker.rs
+22 -14
View File
@@ -38,6 +38,17 @@ pub(super) fn resolve_rebalance_worker_result<T>(
pub(super) type RebalanceEntryTask = tokio::task::JoinHandle<Result<RebalanceEntryOutcome>>;
/// 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<Error>, 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<String> },
@@ -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<Error>, worker_error: Option<Error>) -> Result<()> {
if let Some(err) = entry_error {
return Err(err);
}
pub(super) fn resolve_rebalance_bucket_error(mut entry_error: Option<Error>, worker_error: Option<Error>) -> 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}"))
}