gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-23 20:59:05 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(ecstore): fence rebalance commits and unblock stop

Browse Source
This commit is contained in:
overtrue
2026-08-22 02:45:51 +08:00
parent ce3cd2d890
commit 3677482f9f
8 changed files with 447 additions and 214 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs
+104 -13
View File
@@ -4324,6 +4324,16 @@ pub async fn expire_transitioned_object(
lc_event: &lifecycle::Event,
_src: &LcEventSrc,
bucket_incarnation_id: Uuid,
) -> Result<ObjectInfo, std::io::Error> {
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<ECStore>,
oi: &ObjectInfo,
lc_event: &lifecycle::Event,
bucket_incarnation_id: Uuid,
lock_lost_signal: Option<Arc<rustfs_lock::distributed_lock::LockLostSignal>>,
) -> Result<ObjectInfo, std::io::Error> {
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<ECStore>,
oi: &ObjectInfo,
lc_event: &lifecycle::Event,
src: &LcEventSrc,
bucket_incarnation_id: Uuid,
lock_lost_signal: Option<Arc<rustfs_lock::distributed_lock::LockLostSignal>>,
) -> 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<ECStore>,
oi: &ObjectInfo,
lc_event: &lifecycle::Event,
bucket_incarnation_id: Uuid,
lock_lost_signal: Option<Arc<rustfs_lock::distributed_lock::LockLostSignal>>,
) -> 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<ECStore>,
event: &lifecycle::Event,
src: &LcEventSrc,
oi: &ObjectInfo,
lock_lost_signal: Option<Arc<rustfs_lock::distributed_lock::LockLostSignal>>,
) -> 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(&current, 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<ECStore>, 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<ECStore>, 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(&current, oi) {
return false;
}
enqueue_expiry_rule_with_incarnation(event, src, oi, bucket_incarnation_id).await
crates/ecstore/src/core/pools.rs
+8 -4
View File
@@ -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<Arc<rustfs_lock::distributed_lock::LockLostSignal>>,
) -> Result<bool> {
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
{
crates/ecstore/src/data_movement/mod.rs
+47 -18
View File
@@ -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<uuid::Uuid>,
pub(crate) lifecycle_guard: Option<&'a rustfs_lock::NamespaceLockGuard>,
pub(crate) namespace_lock_lost_signal: Option<Arc<rustfs_lock::distributed_lock::LockLostSignal>>,
}
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<uuid::Uuid>,
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<ECStore>,
pool_idx: usize,
bucket: String,
rd: GetObjectReader,
source_bucket_incarnation_id: Option<uuid::Uuid>,
op_label: &str,
lock_lost_signal: Option<Arc<rustfs_lock::distributed_lock::LockLostSignal>>,
) -> 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
crates/ecstore/src/services/rebalance/control.rs
+46 -2
View File
@@ -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<Option<String>> = 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<Arc<rustfs_lock::distributed_lock::LockLostSignal>> {
self.persisted_guard.lock_lost_signal()
}
}
async fn acquire_persisted_rebalance_run_guard<S>(
@@ -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<Self>, 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)
crates/ecstore/src/services/rebalance/entry.rs
+169 -6
View File
@@ -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<Output = std::result::Result<ObjectInfo, data_movement::SourceCleanupError>>,
) -> Result<RebalanceEntryCleanupResult> {
// 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<SetDisks>,
bucket_configs: Arc<RebalanceBucketConfigs>,
rebalance_id: Arc<str>,
cancel: CancellationToken,
// wk: Arc<Workers>,
) -> Result<RebalanceEntryOutcome> {
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<uuid::Uuid>,
lock_lost_signal: Option<Arc<rustfs_lock::distributed_lock::LockLostSignal>>,
) -> 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"
);
}
}
crates/ecstore/src/services/rebalance/migration.rs
+16 -3
View File
@@ -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<std::sync::Arc<rustfs_lock::distributed_lock::LockLostSignal>>,
}
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<std::sync::Arc<rustfs_lock::distributed_lock::LockLostSignal>>,
) -> 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
}
}
crates/ecstore/src/services/rebalance/rebalance_unit_tests.rs
+56 -168
View File
@@ -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());
crates/ecstore/src/set_disk/ops/object.rs
+1
View File
@@ -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",
)