From 3677482f9f073c4e3026fd2a73a13ef7f37a622e Mon Sep 17 00:00:00 2001 From: overtrue Date: Sat, 22 Aug 2026 02:45:51 +0800 Subject: [PATCH] fix(ecstore): fence rebalance commits and unblock stop --- .../bucket/lifecycle/bucket_lifecycle_ops.rs | 117 ++++++++- crates/ecstore/src/core/pools.rs | 12 +- crates/ecstore/src/data_movement/mod.rs | 65 +++-- .../ecstore/src/services/rebalance/control.rs | 48 +++- .../ecstore/src/services/rebalance/entry.rs | 175 +++++++++++++- .../src/services/rebalance/migration.rs | 19 +- .../rebalance/rebalance_unit_tests.rs | 224 +++++------------- crates/ecstore/src/set_disk/ops/object.rs | 1 + 8 files changed, 447 insertions(+), 214 deletions(-) diff --git a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs index 1162787f7..3fd0ea336 100644 --- a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs +++ b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs @@ -4324,6 +4324,16 @@ pub async fn expire_transitioned_object( lc_event: &lifecycle::Event, _src: &LcEventSrc, bucket_incarnation_id: Uuid, +) -> Result { + expire_transitioned_object_with_lock_lost_signal(api, oi, lc_event, bucket_incarnation_id, None).await +} + +async fn expire_transitioned_object_with_lock_lost_signal( + api: Arc, + oi: &ObjectInfo, + lc_event: &lifecycle::Event, + bucket_incarnation_id: Uuid, + lock_lost_signal: Option>, ) -> Result { let publication_guard = lifecycle_expiry_publication_guard(&api, oi, bucket_incarnation_id) .await @@ -4335,6 +4345,9 @@ pub async fn expire_transitioned_object( let mut opts = transitioned_object_delete_opts(oi, lc_event.action, versioned, version_suspended, bucket_incarnation_id) .map_err(std::io::Error::other)?; opts.add_namespace_lock_guard(&publication_guard); + if let Some(signal) = lock_lost_signal { + opts.add_namespace_lock_lost_signal(signal); + } opts.delete_replication_config_snapshot = Some(Arc::new(snapshot)); //let tags = LcAuditEvent::new(src, lcEvent).Tags(); if lc_event.action.delete_restored() { @@ -4993,12 +5006,32 @@ pub async fn apply_expiry_on_transitioned_object( lc_event: &lifecycle::Event, src: &LcEventSrc, bucket_incarnation_id: Uuid, +) -> bool { + apply_expiry_on_transitioned_object_with_lock_lost_signal(api, oi, lc_event, src, bucket_incarnation_id, None).await +} + +async fn apply_expiry_on_transitioned_object_with_lock_lost_signal( + api: Arc, + oi: &ObjectInfo, + lc_event: &lifecycle::Event, + src: &LcEventSrc, + bucket_incarnation_id: Uuid, + lock_lost_signal: Option>, ) -> bool { if lc_event.action.delete_all() { - return apply_expiry_on_non_transitioned_objects(api, oi, lc_event, src, bucket_incarnation_id).await; + return apply_expiry_on_non_transitioned_objects_with_lock_lost_signal( + api, + oi, + lc_event, + bucket_incarnation_id, + lock_lost_signal, + ) + .await; } let time_ilm = Metrics::time_ilm(lc_event.action); - if let Err(_err) = expire_transitioned_object(api, oi, lc_event, src, bucket_incarnation_id).await { + if let Err(_err) = + expire_transitioned_object_with_lock_lost_signal(api, oi, lc_event, bucket_incarnation_id, lock_lost_signal).await + { return false; } time_ilm(1)(); @@ -5012,6 +5045,16 @@ pub async fn apply_expiry_on_non_transitioned_objects( lc_event: &lifecycle::Event, _src: &LcEventSrc, bucket_incarnation_id: Uuid, +) -> bool { + apply_expiry_on_non_transitioned_objects_with_lock_lost_signal(api, oi, lc_event, bucket_incarnation_id, None).await +} + +async fn apply_expiry_on_non_transitioned_objects_with_lock_lost_signal( + api: Arc, + oi: &ObjectInfo, + lc_event: &lifecycle::Event, + bucket_incarnation_id: Uuid, + lock_lost_signal: Option>, ) -> bool { let Some(publication_guard) = lifecycle_expiry_publication_guard(&api, oi, bucket_incarnation_id).await else { return false; @@ -5042,6 +5085,9 @@ pub async fn apply_expiry_on_non_transitioned_objects( ..Default::default() }; opts.add_namespace_lock_guard(&publication_guard); + if let Some(signal) = lock_lost_signal { + opts.add_namespace_lock_lost_signal(signal); + } if lc_event.action.delete_versioned() { opts.version_id = oi.version_id.map(|v| v.to_string()); @@ -5123,6 +5169,61 @@ async fn enqueue_expiry_rule_with_incarnation( expiry_state.enqueue_by_days(oi, event, src, bucket_incarnation_id) } +fn lifecycle_expiry_object_matches(current: &ObjectInfo, expected: &ObjectInfo) -> bool { + current.version_id == expected.version_id + && current.data_dir == expected.data_dir + && current.mod_time == expected.mod_time + && current.etag == expected.etag + && current.delete_marker == expected.delete_marker + && current.transitioned_object.name == expected.transitioned_object.name + && current.transitioned_object.version_id == expected.transitioned_object.version_id + && current.transitioned_object.tier == expected.transitioned_object.tier + && current.transitioned_object.status == expected.transitioned_object.status + && current.restore_expires == expected.restore_expires +} + +pub(crate) async fn apply_expiry_rule_for_data_movement( + api: Arc, + event: &lifecycle::Event, + src: &LcEventSrc, + oi: &ObjectInfo, + lock_lost_signal: Option>, +) -> bool { + let Ok(_lifecycle_guard) = api.acquire_bucket_lifecycle_read_lock(&oi.bucket).await else { + return false; + }; + let Ok(bucket_incarnation_id) = api.bucket_incarnation_id_from_disk(&oi.bucket).await else { + return false; + }; + let current = match api + .get_object_info( + &oi.bucket, + &oi.name, + &ObjectOptions { + version_id: oi.version_id.map(|version_id| version_id.to_string()), + versioned: oi.version_id.is_some(), + expected_bucket_incarnation_id: Some(bucket_incarnation_id), + ..Default::default() + }, + ) + .await + { + Ok(current) => current, + Err(_) => return false, + }; + if !lifecycle_expiry_object_matches(¤t, oi) { + return false; + } + + if oi.transitioned_object.status.is_empty() { + apply_expiry_on_non_transitioned_objects_with_lock_lost_signal(api, oi, event, bucket_incarnation_id, lock_lost_signal) + .await + } else { + apply_expiry_on_transitioned_object_with_lock_lost_signal(api, oi, event, src, bucket_incarnation_id, lock_lost_signal) + .await + } +} + pub(crate) async fn apply_expiry_rule_in(api: Arc, event: &lifecycle::Event, src: &LcEventSrc, oi: &ObjectInfo) -> bool { let Ok(_lifecycle_guard) = api.acquire_bucket_lifecycle_read_lock(&oi.bucket).await else { return false; @@ -5146,17 +5247,7 @@ pub(crate) async fn apply_expiry_rule_in(api: Arc, event: &lifecycle::E Ok(current) => current, Err(_) => return false, }; - if current.version_id != oi.version_id - || current.data_dir != oi.data_dir - || current.mod_time != oi.mod_time - || current.etag != oi.etag - || current.delete_marker != oi.delete_marker - || current.transitioned_object.name != oi.transitioned_object.name - || current.transitioned_object.version_id != oi.transitioned_object.version_id - || current.transitioned_object.tier != oi.transitioned_object.tier - || current.transitioned_object.status != oi.transitioned_object.status - || current.restore_expires != oi.restore_expires - { + if !lifecycle_expiry_object_matches(¤t, oi) { return false; } enqueue_expiry_rule_with_incarnation(event, src, oi, bucket_incarnation_id).await diff --git a/crates/ecstore/src/core/pools.rs b/crates/ecstore/src/core/pools.rs index 50c882923..91cf38f66 100644 --- a/crates/ecstore/src/core/pools.rs +++ b/crates/ecstore/src/core/pools.rs @@ -19,7 +19,7 @@ use crate::bucket::{ LifecycleExpiryConfigs, bucket_lifecycle_audit::LcEventSrc, bucket_lifecycle_ops::{ - LifecycleOps, apply_expiry_on_transitioned_object, apply_expiry_rule_in, eval_action_from_lifecycle, + LifecycleOps, apply_expiry_rule_for_data_movement, apply_expiry_rule_in, eval_action_from_lifecycle, lifecycle_delete_all_versions_blocked_by_replication, }, get_expiry_configs, @@ -2446,6 +2446,7 @@ pub(crate) async fn should_skip_lifecycle_for_data_movement( object_lock_config: Option<&ObjectLockConfiguration>, apply_actions: bool, event_source: &LcEventSrc, + lock_lost_signal: Option>, ) -> Result { let Some(lifecycle_config) = lifecycle_config else { return Ok(false); @@ -2461,8 +2462,7 @@ pub(crate) async fn should_skip_lifecycle_for_data_movement( let Ok(bucket_incarnation_id) = store.bucket_incarnation_id_from_disk(bucket).await else { return Ok(false); }; - let _ = - apply_expiry_on_transitioned_object(store, &object_info, &event, event_source, bucket_incarnation_id).await; + let _ = apply_expiry_rule_for_data_movement(store, &event, event_source, &object_info, lock_lost_signal).await; } Ok(false) } @@ -2470,7 +2470,8 @@ pub(crate) async fn should_skip_lifecycle_for_data_movement( if lifecycle_delete_all_versions_blocked_by_replication(store.clone(), bucket, &object_info.name, action).await? { return Ok(false); } - let applied = !apply_actions || apply_expiry_rule_in(store, &event, event_source, &object_info).await; + let applied = !apply_actions + || apply_expiry_rule_for_data_movement(store, &event, event_source, &object_info, lock_lost_signal).await; resolve_data_movement_lifecycle_expiry_result(action, apply_actions, applied) } _ => Ok(false), @@ -3044,6 +3045,7 @@ impl ECStore { object_lock_config.as_ref(), true, &LcEventSrc::Decom, + None, ) .await .map_err(|err| with_decommission_entry_context("lifecycle_expiry", bucket.as_str(), version.name.as_str(), err))? @@ -3356,6 +3358,7 @@ impl ECStore { lifecycle_guard: bucket_incarnation_fence .as_ref() .and_then(|guard| guard.namespace_lock_guard()), + namespace_lock_lost_signal: None, }, "decommission", ) @@ -4333,6 +4336,7 @@ impl ECStore { object_lock_config.as_ref(), false, &LcEventSrc::Decom, + None, ) .await { diff --git a/crates/ecstore/src/data_movement/mod.rs b/crates/ecstore/src/data_movement/mod.rs index 1f4daa703..caccd1e94 100644 --- a/crates/ecstore/src/data_movement/mod.rs +++ b/crates/ecstore/src/data_movement/mod.rs @@ -1024,10 +1024,11 @@ pub(crate) enum SourceCleanupError { Storage(#[from] Error), } -#[derive(Clone, Copy, Default)] +#[derive(Clone, Default)] pub(crate) struct SourceCleanupBucketFence<'a> { pub(crate) expected_incarnation_id: Option, pub(crate) lifecycle_guard: Option<&'a rustfs_lock::NamespaceLockGuard>, + pub(crate) namespace_lock_lost_signal: Option>, } fn ensure_source_cleanup_versions_match( @@ -1186,6 +1187,9 @@ pub(crate) async fn cleanup_source_entry_if_unchanged( if let Some(bucket_lifecycle_guard) = bucket_fence.lifecycle_guard { opts.add_bucket_lifecycle_lock_guard(bucket_lifecycle_guard); } + if let Some(signal) = bucket_fence.namespace_lock_lost_signal { + opts.add_namespace_lock_lost_signal(signal); + } let result = set.delete_object(bucket, cleanup_key.as_str(), opts).await; if result.is_ok() { crate::store::list_objects::observe_scanner_namespace_mutations(bucket, 1); @@ -1337,6 +1341,19 @@ pub(crate) async fn migrate_object( rd: GetObjectReader, source_bucket_incarnation_id: Option, op_label: &str, +) -> Result<()> { + migrate_object_with_lock_lost_signal(store, pool_idx, bucket, rd, source_bucket_incarnation_id, op_label, None).await +} + +#[allow(clippy::too_many_arguments)] +pub(crate) async fn migrate_object_with_lock_lost_signal( + store: Arc, + pool_idx: usize, + bucket: String, + rd: GetObjectReader, + source_bucket_incarnation_id: Option, + op_label: &str, + lock_lost_signal: Option>, ) -> Result<()> { let object_info = rd.object_info.clone(); let has_part_checksums = object_info @@ -1349,6 +1366,9 @@ pub(crate) async fn migrate_object( if should_use_multipart_data_movement(&object_info, has_part_checksums) { let mut new_multipart_opts = data_movement_new_multipart_opts(&object_info, pool_idx); new_multipart_opts.expected_bucket_incarnation_id = source_bucket_incarnation_id; + if let Some(signal) = lock_lost_signal.as_ref() { + new_multipart_opts.add_namespace_lock_lost_signal(Arc::clone(signal)); + } let (res, target_pool_idx, expected_bucket_incarnation_id) = match store .handle_new_multipart_upload_with_pool_idx(&bucket, &object_info.name, &new_multipart_opts) .await @@ -1393,7 +1413,7 @@ pub(crate) async fn migrate_object( err, ) })?; - let part_opts = ObjectOptions { + let mut part_opts = ObjectOptions { part_number: Some(part.number), preserve_etag: Some(part.etag.clone()), data_movement: true, @@ -1401,6 +1421,9 @@ pub(crate) async fn migrate_object( expected_bucket_incarnation_id, ..Default::default() }; + if let Some(signal) = lock_lost_signal.as_ref() { + part_opts.add_namespace_lock_lost_signal(Arc::clone(signal)); + } let pi = match store .put_object_part_for_data_movement( target_pool_idx, @@ -1445,6 +1468,9 @@ pub(crate) async fn migrate_object( ) })?; complete_multipart_opts.expected_bucket_incarnation_id = expected_bucket_incarnation_id; + if let Some(signal) = lock_lost_signal.as_ref() { + complete_multipart_opts.add_namespace_lock_lost_signal(Arc::clone(signal)); + } if let Err(err) = store .clone() .complete_multipart_upload_for_data_movement( @@ -1493,18 +1519,18 @@ pub(crate) async fn migrate_object( if multipart_result.is_ok() && should_abort_multipart_upload(&abort_multipart_flag) { let abort_result = store - .abort_multipart_upload_for_data_movement( - target_pool_idx, - &bucket, - &object_info.name, - &res.upload_id, - &ObjectOptions { + .abort_multipart_upload_for_data_movement(target_pool_idx, &bucket, &object_info.name, &res.upload_id, &{ + let mut opts = ObjectOptions { data_movement: true, src_pool_idx: pool_idx, expected_bucket_incarnation_id, ..Default::default() - }, - ) + }; + if let Some(signal) = lock_lost_signal.as_ref() { + opts.add_namespace_lock_lost_signal(Arc::clone(signal)); + } + opts + }) .await; match abort_result { Ok(()) => return Ok(()), @@ -1562,18 +1588,18 @@ pub(crate) async fn migrate_object( if let Err(primary_err) = multipart_result { if should_abort_multipart_upload(&abort_multipart_flag) { return match store - .abort_multipart_upload_for_data_movement( - target_pool_idx, - &bucket, - &object_info.name, - &res.upload_id, - &ObjectOptions { + .abort_multipart_upload_for_data_movement(target_pool_idx, &bucket, &object_info.name, &res.upload_id, &{ + let mut opts = ObjectOptions { data_movement: true, src_pool_idx: pool_idx, expected_bucket_incarnation_id, ..Default::default() - }, - ) + }; + if let Some(signal) = lock_lost_signal.as_ref() { + opts.add_namespace_lock_lost_signal(Arc::clone(signal)); + } + opts + }) .await { Ok(()) => Err(primary_err), @@ -1608,6 +1634,9 @@ pub(crate) async fn migrate_object( let mut put_opts = data_movement_put_object_opts(&object_info, pool_idx); put_opts.expected_bucket_incarnation_id = source_bucket_incarnation_id; + if let Some(signal) = lock_lost_signal { + put_opts.add_namespace_lock_lost_signal(signal); + } let (target_pool_idx, put_result) = store .put_object_for_data_movement(&bucket, &object_info.name, &mut data, &put_opts) .await diff --git a/crates/ecstore/src/services/rebalance/control.rs b/crates/ecstore/src/services/rebalance/control.rs index a57c8bf3e..7e1860c6b 100644 --- a/crates/ecstore/src/services/rebalance/control.rs +++ b/crates/ecstore/src/services/rebalance/control.rs @@ -32,6 +32,16 @@ use time::OffsetDateTime; use tracing::{debug, info}; use uuid::Uuid; +#[cfg(test)] +static FAIL_NEXT_REBALANCE_ACTIVATION_SAVE: std::sync::Mutex> = std::sync::Mutex::new(None); + +#[cfg(test)] +pub(super) fn fail_next_rebalance_activation_save_for_test(rebalance_id: &str) { + *FAIL_NEXT_REBALANCE_ACTIVATION_SAVE + .lock() + .expect("rebalance activation save failure hook should not be poisoned") = Some(rebalance_id.to_string()); +} + fn ensure_rebalance_activation_pool_meta_allowed(meta: &PoolMeta) -> Result<()> { if pool_meta_has_active_decommission(meta) { return Err(Error::DecommissionAlreadyRunning); @@ -57,6 +67,10 @@ impl RebalanceRunGuard { } Ok(()) } + + pub(super) fn lock_lost_signal(&self) -> Option> { + self.persisted_guard.lock_lost_signal() + } } async fn acquire_persisted_rebalance_run_guard( @@ -145,7 +159,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() || !is_rebalance_conflicting_with_decommission(meta) { + if meta.stopped_at.is_some() + || 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}"))); } Ok(()) @@ -287,6 +307,16 @@ impl ECStore { where S: EcstoreObjectIO, { + #[cfg(test)] + { + let mut fail_id = FAIL_NEXT_REBALANCE_ACTIVATION_SAVE + .lock() + .expect("rebalance activation save failure hook should not be poisoned"); + if fail_id.as_deref() == Some(local_snapshot.id.as_str()) { + fail_id.take(); + return Err(Error::other("injected rebalance activation save failure")); + } + } merge_and_save_rebalance_meta_no_lock( pool, local_snapshot, @@ -885,7 +915,21 @@ impl ECStore { #[tracing::instrument(skip(self))] pub async fn stop_rebalance_for_id(self: &Arc, expected_id: Option<&str>) -> Result<()> { let _start_guard = self.start_gate.lock().await; - let _activation_guard = self.rebalance_activation_write_guard(expected_id, "stop rebalance").await?; + let activation_gate = { + let mut rebalance_meta = self.rebalance_meta.write().await; + if let Some(expected_id) = expected_id { + ensure_rebalance_run_id(rebalance_meta.as_ref(), expected_id, "stop rebalance")?; + } + rebalance_meta.as_mut().map(|meta| { + let cancel = meta.cancel.get_or_insert_with(tokio_util::sync::CancellationToken::new); + cancel.cancel(); + Arc::clone(&meta.activation_gate) + }) + }; + let _activation_guard = match activation_gate { + Some(gate) => Some(gate.write_owned().await), + None => None, + }; let meta_to_save = { let mut rebalance_meta = self.rebalance_meta.write().await; stop_rebalance_meta_snapshot_for_id(rebalance_meta.as_mut(), OffsetDateTime::now_utc(), expected_id) diff --git a/crates/ecstore/src/services/rebalance/entry.rs b/crates/ecstore/src/services/rebalance/entry.rs index de5015fd2..37370415e 100644 --- a/crates/ecstore/src/services/rebalance/entry.rs +++ b/crates/ecstore/src/services/rebalance/entry.rs @@ -43,6 +43,13 @@ use time::OffsetDateTime; use tokio_util::sync::CancellationToken; use tracing::{debug, error, warn}; +fn ensure_rebalance_entry_active(cancel: &CancellationToken) -> Result<()> { + if cancel.is_cancelled() { + return Err(Error::OperationCanceled); + } + Ok(()) +} + impl ECStore { async fn finish_rebalance_entry_after_cleanup( &self, @@ -52,11 +59,14 @@ impl ECStore { object: &str, stats_updates: &[&FileInfo], expected_id: &str, + cancel: &CancellationToken, cleanup: impl std::future::Future>, ) -> Result { // Persisted stats can complete a pool on restart, so source cleanup must resolve first. + ensure_rebalance_entry_active(cancel)?; run_guard.ensure_held("rebalance source cleanup")?; let cleanup_result = cleanup.await; + ensure_rebalance_entry_active(cancel)?; run_guard.ensure_held("rebalance source cleanup")?; let cleanup_result = resolve_rebalance_entry_cleanup_delete_result(cleanup_result, bucket, object); let RebalanceEntryCleanupResult::Completed { warning } = cleanup_result else { @@ -103,6 +113,7 @@ impl ECStore { set: Arc, bucket_configs: Arc, rebalance_id: Arc, + cancel: CancellationToken, // wk: Arc, ) -> Result { debug!( @@ -165,13 +176,16 @@ impl ECStore { // Entry lock order is bucket incarnation -> activation_gate -> rebalance.bin. // Stop waits for in-flight entries through cleanup, but not for entries admitted later. + ensure_rebalance_entry_active(&cancel)?; let run_guard = self.rebalance_run_guard(rebalance_id.as_ref(), "rebalance entry").await?; + let lock_lost_signal = run_guard.lock_lost_signal(); let mut rebalanced: usize = 0; let mut expired: usize = 0; let mut cleanup_preflight_allowed_missing = Vec::new(); let mut stats_updates = Vec::with_capacity(fivs.versions.len()); for version in fivs.versions.iter() { + ensure_rebalance_entry_active(&cancel)?; run_guard.ensure_held("rebalance lifecycle mutation")?; let lifecycle_result = crate::core::pools::should_skip_lifecycle_for_data_movement( self.clone(), @@ -181,8 +195,10 @@ impl ECStore { bucket_configs.object_lock_config.as_ref(), true, &crate::bucket::lifecycle::bucket_lifecycle_audit::LcEventSrc::Rebal, + lock_lost_signal.clone(), ) .await; + ensure_rebalance_entry_active(&cancel)?; run_guard.ensure_held("rebalance lifecycle mutation")?; let expired_by_lifecycle = lifecycle_result?; if expired_by_lifecycle { @@ -224,23 +240,29 @@ impl ECStore { let version_id = version.version_id.map(|v| v.to_string()); let expected_bucket_incarnation_id = bucket_configs.bucket_incarnation_id; + let transfer_lock_lost_signal = lock_lost_signal.clone(); let mut transfer = |src_pool_idx: usize, bucket: String, rd: GetObjectReader| { let store = self.clone(); + let lock_lost_signal = transfer_lock_lost_signal.clone(); async move { store - .rebalance_object(src_pool_idx, bucket, rd, expected_bucket_incarnation_id) + .rebalance_object(src_pool_idx, bucket, rd, expected_bucket_incarnation_id, lock_lost_signal) .await } }; // Route delete-marker migration through the store layer so it lands on the // cross-pool target (excluding the source pool), not back onto the source set. - let mut delete_marker = |bucket: String, object: String, opts: ObjectOptions| { + let delete_marker_lock_lost_signal = lock_lost_signal.clone(); + let mut delete_marker = |bucket: String, object: String, mut opts: ObjectOptions| { let store = self.clone(); + if let Some(signal) = delete_marker_lock_lost_signal.as_ref() { + opts.add_namespace_lock_lost_signal(Arc::clone(signal)); + } async move { store.delete_object(&bucket, &object, opts).await } }; run_guard.ensure_held("rebalance version migration")?; let result = migrate_entry_version( - &RebalanceMigrationBackend::new(set.as_ref(), self.as_ref()), + &RebalanceMigrationBackend::new(set.as_ref(), self.as_ref(), lock_lost_signal.clone()), bucket.clone(), pool_index, version, @@ -252,6 +274,7 @@ impl ECStore { &mut delete_marker, ) .await; + ensure_rebalance_entry_active(&cancel)?; run_guard.ensure_held("rebalance version migration")?; if result.ignored { @@ -355,6 +378,7 @@ impl ECStore { entry.name.as_str(), stats_updates.as_slice(), rebalance_id.as_ref(), + &cancel, data_movement::cleanup_source_entry_if_unchanged( set.clone(), bucket.as_str(), @@ -366,6 +390,7 @@ impl ECStore { lifecycle_guard: bucket_incarnation_fence .as_ref() .and_then(|guard| guard.namespace_lock_guard()), + namespace_lock_lost_signal: lock_lost_signal.clone(), }, "rebalance", ), @@ -441,6 +466,7 @@ impl ECStore { resolve_rebalance_stats_update_result(stats_result, pool_index, bucket.as_str(), entry.name.as_str())?; } + ensure_rebalance_entry_active(&cancel)?; run_guard.ensure_held("rebalance entry completion")?; Ok(RebalanceEntryOutcome::Completed) } @@ -452,8 +478,18 @@ impl ECStore { bucket: String, rd: GetObjectReader, expected_bucket_incarnation_id: Option, + lock_lost_signal: Option>, ) -> Result<()> { - data_movement::migrate_object(self, pool_idx, bucket, rd, expected_bucket_incarnation_id, "rebalance_object").await + data_movement::migrate_object_with_lock_lost_signal( + self, + pool_idx, + bucket, + rd, + expected_bucket_incarnation_id, + "rebalance_object", + lock_lost_signal, + ) + .await } async fn update_rebalance_last_error(&self, pool_idx: usize, message: String, expected_id: &str) -> Result<()> { @@ -576,7 +612,15 @@ impl ECStore { "Started rebalance entry task" ); let result = this - .rebalance_entry(bucket, pool_index, entry, set, bucket_configs, rebalance_id) + .rebalance_entry( + bucket, + pool_index, + entry, + set, + bucket_configs, + rebalance_id, + callback_rx.clone(), + ) .await; if let Err(err) = &result { error!("rebalance_entry: rebalance entry failed: {err}"); @@ -682,7 +726,10 @@ impl ECStore { mod tests { use super::*; use crate::services::rebalance::{RebalStatus, RebalanceInfo, RebalanceMeta, RebalanceStats}; - use rustfs_filemeta::FileInfo; + use crate::storage_api_contracts::bucket::{BucketOperations as _, MakeBucketOptions}; + use crate::storage_api_contracts::object::ObjectOperations as _; + use rustfs_filemeta::{FileInfo, FileMeta}; + use std::time::Duration as StdDuration; use time::OffsetDateTime; #[tokio::test] @@ -708,6 +755,7 @@ mod tests { version.is_latest = true; let warning_version = version.clone(); let (release_cleanup, cleanup_released) = tokio::sync::oneshot::channel(); + let cancel = CancellationToken::new(); let finish_store = Arc::clone(&store); let finish = tokio::spawn(async move { @@ -723,6 +771,7 @@ mod tests { "object.bin", &[&version], rebalance_id, + &cancel, async move { cleanup_released.await.expect("cleanup release sender should remain alive"); Ok(ObjectInfo::default()) @@ -782,6 +831,7 @@ mod tests { "object.bin", &[&warning_version], rebalance_id, + &CancellationToken::new(), async { Err(Error::SlowDown.into()) }, ) .await @@ -810,6 +860,7 @@ mod tests { "object.bin", &[&warning_version], rebalance_id, + &CancellationToken::new(), async { Err(data_movement::SourceCleanupError::SourceChanged) }, ) .await @@ -820,4 +871,116 @@ mod tests { assert_eq!(pool_stats.bytes, 0, "deferred cleanup must not commit completion stats"); assert_eq!(pool_stats.cleanup_warnings.count, 1, "deferred cleanup must not add a permanent warning"); } + + #[tokio::test] + #[serial_test::serial] + async fn stop_cancels_real_rebalance_entry_before_waiting_for_cleanup_guard() { + let rebalance_id = "rebalance-stop-entry"; + let cancel = CancellationToken::new(); + let (_temp_dirs, store) = crate::services::rebalance::test_store_with_persisted_rebalance_meta(RebalanceMeta { + id: rebalance_id.to_string(), + percent_free_goal: 1.0, + cancel: Some(cancel.clone()), + pool_stats: vec![RebalanceStats { + participating: true, + init_capacity: 100, + buckets: vec!["bucket".to_string()], + info: RebalanceInfo { + start_time: Some(OffsetDateTime::now_utc()), + status: RebalStatus::Started, + ..Default::default() + }, + ..Default::default() + }], + ..Default::default() + }) + .await; + let bucket = "bucket"; + let object = "delete-marker"; + let set = store.pools[0].get_disks_by_key(object); + set.make_bucket(bucket, &MakeBucketOptions::default()) + .await + .expect("source bucket should be created"); + set.delete_object( + bucket, + object, + ObjectOptions { + versioned: true, + ..Default::default() + }, + ) + .await + .expect("source delete marker should be created"); + let source_versions = set + .load_file_info_versions_exact(bucket, object) + .await + .expect("source metadata should be readable") + .expect("source delete marker should exist"); + assert_eq!(source_versions.versions.len(), 1); + assert!(source_versions.versions[0].deleted); + let mut file_meta = FileMeta::new(); + for version in &source_versions.versions { + file_meta + .add_version(version.clone()) + .expect("source version should encode into a metacache entry"); + } + let entry = MetaCacheEntry { + name: object.to_string(), + metadata: file_meta.marshal_msg().expect("source metadata should marshal"), + cached: Some(file_meta), + reusable: false, + }; + let barrier = data_movement::SourceCleanupDeleteBarrier::install(bucket, object); + let entry_store = Arc::clone(&store); + let entry_set = Arc::clone(&set); + let entry_cancel = cancel.clone(); + let mut entry_task = tokio::spawn(async move { + entry_store + .rebalance_entry( + bucket.to_string(), + 0, + entry, + entry_set, + Arc::new(RebalanceBucketConfigs::default()), + Arc::from(rebalance_id), + entry_cancel, + ) + .await + }); + barrier.wait_until_paused().await; + + let stop_store = Arc::clone(&store); + let mut stop_task = tokio::spawn(async move { stop_store.stop_rebalance_for_id(Some(rebalance_id)).await }); + tokio::time::timeout(StdDuration::from_secs(1), cancel.cancelled()) + .await + .expect("stop must cancel the in-flight entry before waiting for its guard"); + assert!( + tokio::time::timeout(StdDuration::from_millis(50), &mut stop_task) + .await + .is_err(), + "stop must wait for the real entry cleanup guard to drain" + ); + + barrier.release(); + let entry_error = tokio::time::timeout(StdDuration::from_secs(5), &mut entry_task) + .await + .expect("the cancelled entry should finish in bounded time") + .expect("entry task should not panic") + .expect_err("the entry must observe stop cancellation"); + assert!(matches!(entry_error, Error::OperationCanceled)); + tokio::time::timeout(StdDuration::from_secs(5), &mut stop_task) + .await + .expect("stop should finish after the entry guard drains") + .expect("stop task should not panic") + .expect("stop should persist the terminal state"); + assert!( + store + .rebalance_meta + .read() + .await + .as_ref() + .is_some_and(|meta| meta.stopped_at.is_some()), + "stop should publish the terminal state after entry cleanup drains" + ); + } } diff --git a/crates/ecstore/src/services/rebalance/migration.rs b/crates/ecstore/src/services/rebalance/migration.rs index c3e1e5e69..cf2dabd8f 100644 --- a/crates/ecstore/src/services/rebalance/migration.rs +++ b/crates/ecstore/src/services/rebalance/migration.rs @@ -102,11 +102,20 @@ pub(crate) trait MigrationBackend: Send + Sync { pub(crate) struct RebalanceMigrationBackend<'a> { source: &'a SetDisks, store: &'a ECStore, + lock_lost_signal: Option>, } impl<'a> RebalanceMigrationBackend<'a> { - pub(crate) fn new(source: &'a SetDisks, store: &'a ECStore) -> Self { - Self { source, store } + pub(crate) fn new( + source: &'a SetDisks, + store: &'a ECStore, + lock_lost_signal: Option>, + ) -> Self { + Self { + source, + store, + lock_lost_signal, + } } } @@ -130,7 +139,11 @@ impl MigrationBackend for RebalanceMigrationBackend<'_> { fi: &FileInfo, opts: &ObjectOptions, ) -> Result<()> { - self.store.decommission_tiered_object(bucket, object, fi, opts).await + let mut opts = opts.clone(); + if let Some(signal) = self.lock_lost_signal.as_ref() { + opts.add_namespace_lock_lost_signal(std::sync::Arc::clone(signal)); + } + self.store.decommission_tiered_object(bucket, object, fi, &opts).await } } diff --git a/crates/ecstore/src/services/rebalance/rebalance_unit_tests.rs b/crates/ecstore/src/services/rebalance/rebalance_unit_tests.rs index b3cd1cdbb..f63a56114 100644 --- a/crates/ecstore/src/services/rebalance/rebalance_unit_tests.rs +++ b/crates/ecstore/src/services/rebalance/rebalance_unit_tests.rs @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -use super::control::{merge_and_save_rebalance_meta_no_lock, validate_rebalance_disk_stats_coverage}; +use super::control::{fail_next_rebalance_activation_save_for_test, validate_rebalance_disk_stats_coverage}; use super::meta::{ RebalanceMetaMergeOutcome, RebalanceTerminalEvent, apply_rebalance_save_option, apply_rebalance_terminal_event, apply_stopped_at, classify_rebalance_terminal_event, clone_arc_by_index, clone_first_arc, clone_rebalance_pool_stats, @@ -73,7 +73,6 @@ use std::io::Cursor; use std::sync::Arc; use std::sync::Mutex; use std::sync::atomic::{AtomicUsize, Ordering}; -use std::task::{Context, Poll}; use time::OffsetDateTime; use tokio::sync::mpsc; use tokio::time::Duration; @@ -2767,10 +2766,10 @@ fn test_rebalance_activation_candidate_does_not_clobber_replacement_token() { } #[tokio::test] +#[serial_test::serial] async fn test_rebalance_start_save_failure_retries_persisted_completed_state() { - let (_temp_dirs, _disk_stores, pool) = crate::set_disk::hermetic_set_disks_isolated(4).await; - let mut local = RebalanceMeta { - id: "rebalance-a".to_string(), + let active = RebalanceMeta { + id: "rebalance-real-save-completed".to_string(), percent_free_goal: 0.5, pool_stats: vec![RebalanceStats { participating: true, @@ -2785,93 +2784,49 @@ async fn test_rebalance_start_save_failure_retries_persisted_completed_state() { }], ..Default::default() }; - local.save(pool.clone()).await.expect("active metadata should be persisted"); + let (_temp_dirs, store) = super::test_store_with_persisted_rebalance_meta(active).await; - let (failed_candidate, outcome, must_persist) = stage_local_rebalance_worker_activation( - &local, - "rebalance-a", - CancellationToken::new(), - OffsetDateTime::from_unix_timestamp(1_000).expect("test timestamp should be valid"), - ) - .expect("terminal activation candidate should be staged"); - assert_eq!(outcome, RebalanceLocalActivationOutcome::NotStartedTerminal); - assert!(must_persist); - assert_eq!(failed_candidate.pool_stats[0].info.status, RebalStatus::Completed); - assert_eq!(local.pool_stats[0].info.status, RebalStatus::Started); - - let err = merge_and_save_rebalance_meta_no_lock( - pool.clone(), - &failed_candidate, - "failed completed candidate", - ObjectOptions { - no_lock: true, - namespace_lock_fence: Some(crate::object_api::NamespaceLockFence::lost_for_test()), - ..Default::default() - }, - None, - Some(local.id.as_str()), - ) - .await - .expect_err("lost namespace quorum must reject the terminal candidate save"); - - assert!(matches!( - err, - Error::NamespaceLockQuorumUnavailable { - required: 3, - achieved: 2, - .. - } - )); - assert_eq!(local.pool_stats[0].info.status, RebalStatus::Started); - assert!(local.cancel.is_none(), "failed persistence must not publish a worker token"); + fail_next_rebalance_activation_save_for_test("rebalance-real-save-completed"); + let err = store + .start_rebalance_under_gate() + .await + .expect_err("the injected first activation save must fail through the real start path"); + assert!(err.to_string().contains("injected rebalance activation save failure")); + { + let local = store.rebalance_meta.read().await; + let local = local.as_ref().expect("local rebalance metadata should remain present"); + assert_eq!(local.pool_stats[0].info.status, RebalStatus::Started); + assert!(local.cancel.is_none(), "failed persistence must not publish a worker token"); + } let mut after_failure = RebalanceMeta::new(); after_failure - .load(pool.clone()) + .load(store.pools[0].clone()) .await .expect("active metadata should remain readable after the failed save"); assert_eq!(after_failure.pool_stats[0].info.status, RebalStatus::Started); - let expected_cancel = local.cancel.clone(); - let (retry_candidate, retry_outcome, retry_must_persist) = stage_local_rebalance_worker_activation( - &local, - "rebalance-a", - CancellationToken::new(), - OffsetDateTime::from_unix_timestamp(2_000).expect("test timestamp should be valid"), - ) - .expect("retry terminal activation candidate should be staged"); - assert_eq!(retry_outcome, RebalanceLocalActivationOutcome::NotStartedTerminal); - assert!(retry_must_persist); - merge_and_save_rebalance_meta_no_lock( - pool.clone(), - &retry_candidate, - "retry completed candidate", - ObjectOptions { - no_lock: true, - ..Default::default() - }, - None, - Some(local.id.as_str()), - ) - .await - .expect("retry must persist the completed candidate"); - commit_local_rebalance_worker_activation_candidate(&mut local, "rebalance-a", expected_cancel.as_ref(), retry_candidate) - .expect("successful persistence should publish the completed candidate locally"); + store + .start_rebalance_under_gate() + .await + .expect("the real start path must retry and persist the terminal candidate"); let mut persisted = RebalanceMeta::new(); persisted - .load(pool) + .load(store.pools[0].clone()) .await .expect("retry-persisted completed metadata should be readable"); assert_eq!(persisted.pool_stats[0].info.status, RebalStatus::Completed); + let local = store.rebalance_meta.read().await; + let local = local.as_ref().expect("local rebalance metadata should remain present"); assert_eq!(local.pool_stats[0].info.status, RebalStatus::Completed); assert!(local.cancel.is_none()); } #[tokio::test] +#[serial_test::serial] async fn test_rebalance_start_save_failure_retries_persisted_stopped_state() { - let (_temp_dirs, _disk_stores, pool) = crate::set_disk::hermetic_set_disks_isolated(4).await; let active = RebalanceMeta { - id: "rebalance-a".to_string(), + id: "rebalance-real-save-stopped".to_string(), pool_stats: vec![RebalanceStats { participating: true, info: RebalanceInfo { @@ -2882,71 +2837,51 @@ async fn test_rebalance_start_save_failure_retries_persisted_stopped_state() { }], ..Default::default() }; - active.save(pool.clone()).await.expect("active metadata should be persisted"); + let (_temp_dirs, store) = super::test_store_with_persisted_rebalance_meta(active).await; let stopped_at = OffsetDateTime::from_unix_timestamp(1_000).expect("test timestamp should be valid"); - let mut local = active.clone(); - local.stopped_at = Some(stopped_at); - local.pool_stats[0].info.status = RebalStatus::Stopped; - local.pool_stats[0].info.end_time = Some(stopped_at); + { + let mut local = store.rebalance_meta.write().await; + let local = local.as_mut().expect("local rebalance metadata should remain present"); + local.stopped_at = Some(stopped_at); + local.pool_stats[0].info.status = RebalStatus::Stopped; + local.pool_stats[0].info.end_time = Some(stopped_at); + } - let (failed_candidate, outcome, must_persist) = - stage_local_rebalance_worker_activation(&local, "rebalance-a", CancellationToken::new(), stopped_at) - .expect("stopped activation candidate should be staged"); - assert_eq!(outcome, RebalanceLocalActivationOutcome::NotStartedTerminal); - assert!(must_persist, "an already-terminal local state must still be persisted on retry"); - let err = merge_and_save_rebalance_meta_no_lock( - pool.clone(), - &failed_candidate, - "failed stopped candidate", - ObjectOptions { - no_lock: true, - namespace_lock_fence: Some(crate::object_api::NamespaceLockFence::lost_for_test()), - ..Default::default() - }, - None, - Some(local.id.as_str()), - ) - .await - .expect_err("lost namespace quorum must reject the stopped candidate save"); - assert!(matches!(err, Error::NamespaceLockQuorumUnavailable { .. })); + fail_next_rebalance_activation_save_for_test("rebalance-real-save-stopped"); + let err = store + .start_rebalance_under_gate() + .await + .expect_err("the injected first stopped-state save must fail through the real start path"); + assert!(err.to_string().contains("injected rebalance activation save failure")); let mut after_failure = RebalanceMeta::new(); after_failure - .load(pool.clone()) + .load(store.pools[0].clone()) .await .expect("active metadata should remain readable after the failed save"); assert_eq!(after_failure.pool_stats[0].info.status, RebalStatus::Started); - assert_eq!(local.pool_stats[0].info.status, RebalStatus::Stopped); + { + let local = store.rebalance_meta.read().await; + let local = local.as_ref().expect("local rebalance metadata should remain present"); + assert_eq!(local.pool_stats[0].info.status, RebalStatus::Stopped); + assert!(local.cancel.is_none(), "failed persistence must not publish a worker token"); + } - let expected_cancel = local.cancel.clone(); - let (retry_candidate, retry_outcome, retry_must_persist) = - stage_local_rebalance_worker_activation(&local, "rebalance-a", CancellationToken::new(), stopped_at) - .expect("retry stopped activation candidate should be staged"); - assert_eq!(retry_outcome, RebalanceLocalActivationOutcome::NotStartedTerminal); - assert!(retry_must_persist); - merge_and_save_rebalance_meta_no_lock( - pool.clone(), - &retry_candidate, - "retry stopped candidate", - ObjectOptions { - no_lock: true, - ..Default::default() - }, - None, - Some(local.id.as_str()), - ) - .await - .expect("retry must persist the stopped candidate"); - commit_local_rebalance_worker_activation_candidate(&mut local, "rebalance-a", expected_cancel.as_ref(), retry_candidate) - .expect("successful persistence should preserve the stopped candidate locally"); + store + .start_rebalance_under_gate() + .await + .expect("the real start path must retry and persist the stopped candidate"); let mut persisted = RebalanceMeta::new(); persisted - .load(pool) + .load(store.pools[0].clone()) .await .expect("retry-persisted stopped metadata should be readable"); assert_eq!(persisted.stopped_at, Some(stopped_at)); assert_eq!(persisted.pool_stats[0].info.status, RebalStatus::Stopped); + let local = store.rebalance_meta.read().await; + let local = local.as_ref().expect("local rebalance metadata should remain present"); assert_eq!(local.pool_stats[0].info.status, RebalStatus::Stopped); + assert!(local.cancel.is_none()); } #[tokio::test] @@ -3008,53 +2943,6 @@ async fn test_old_worker_cannot_mutate_replacement_rebalance_state() { assert!(meta.stopped_at.is_none()); } -#[tokio::test] -async fn test_stop_waits_for_active_rebalance_migration_guard() { - 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 (_temp_dirs, store) = super::test_store_with_persisted_rebalance_meta(meta).await; - let run_guard = store - .rebalance_run_guard("rebalance-a", "rebalance remote-tier migration") - .await - .expect("active run should admit the remote-tier migration"); - let mut stop = Box::pin(store.stop_rebalance_for_id(Some("rebalance-a"))); - let mut context = Context::from_waker(futures::task::noop_waker_ref()); - - assert!(matches!(stop.as_mut().poll(&mut context), Poll::Pending)); - assert!( - store - .rebalance_meta - .read() - .await - .as_ref() - .is_some_and(|meta| meta.stopped_at.is_none()), - "stop must not change run state while a fenced side effect is active" - ); - - drop(run_guard); - stop.await - .expect("stop should persist after the side-effect fence is released"); - assert!( - store - .rebalance_meta - .read() - .await - .as_ref() - .is_some_and(|meta| meta.stopped_at.is_some()), - "stop should commit after the production side-effect fence is released" - ); -} - #[tokio::test] async fn test_rebalance_metadata_reload_under_start_gate_does_not_reacquire_gate() { let store = test_store_with_rebalance_meta(RebalanceMeta::default()); diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index da809eea6..570a405a1 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/crates/ecstore/src/set_disk/ops/object.rs @@ -11634,6 +11634,7 @@ mod transition_upload_integrity_tests { crate::data_movement::SourceCleanupBucketFence { expected_incarnation_id: None, lifecycle_guard: Some(&bucket_guard), + namespace_lock_lost_signal: None, }, "test_data_movement", )