diff --git a/crates/ecstore/src/bucket/metadata_sys.rs b/crates/ecstore/src/bucket/metadata_sys.rs index 46962d211..1d3c43133 100644 --- a/crates/ecstore/src/bucket/metadata_sys.rs +++ b/crates/ecstore/src/bucket/metadata_sys.rs @@ -1161,6 +1161,16 @@ pub async fn get_replication_config(bucket: &str) -> Result<(ReplicationConfigur bucket_meta_sys.get_replication_config(bucket).await } +pub(crate) async fn get_replication_config_in( + ctx: &crate::runtime::instance::InstanceContext, + bucket: &str, +) -> Result<(ReplicationConfiguration, OffsetDateTime)> { + let bucket_meta_sys_lock = bucket_metadata_sys_of(ctx)?; + let bucket_meta_sys = bucket_meta_sys_lock.read().await; + + bucket_meta_sys.get_replication_config(bucket).await +} + pub async fn get_notification_config(bucket: &str) -> Result> { let bucket_meta_sys_lock = get_bucket_metadata_sys()?; let bucket_meta_sys = bucket_meta_sys_lock.read().await; diff --git a/crates/ecstore/src/bucket/versioning_sys.rs b/crates/ecstore/src/bucket/versioning_sys.rs index abc6fee6d..685e95af9 100644 --- a/crates/ecstore/src/bucket/versioning_sys.rs +++ b/crates/ecstore/src/bucket/versioning_sys.rs @@ -90,10 +90,6 @@ impl BucketVersioningSys { /// caller's own instance context so a second in-process store never /// answers with the first instance's versioning state; falls back to the /// ambient system when the instance cell is not initialized. - #[allow( - dead_code, - reason = "instance-scoped seam (backlog#1052) with no caller in this port (backlog#1823)" - )] pub(crate) async fn get_in(ctx: &crate::runtime::instance::InstanceContext, bucket: &str) -> Result { if bucket == RUSTFS_META_BUCKET || bucket.starts_with(RUSTFS_META_BUCKET) { return Ok(VersioningConfiguration::default()); diff --git a/crates/ecstore/src/core/pools.rs b/crates/ecstore/src/core/pools.rs index 8e2c54bee..956b3c3d1 100644 --- a/crates/ecstore/src/core/pools.rs +++ b/crates/ecstore/src/core/pools.rs @@ -15,6 +15,7 @@ use crate::bucket::replication::replication_state_from_filemeta; #[cfg(all(test, feature = "test-util"))] use crate::bucket::utils::is_meta_bucketname; +use crate::bucket::versioning::VersioningApi as _; use crate::bucket::versioning_sys::BucketVersioningSys; use crate::bucket::{ lifecycle::{ @@ -8812,7 +8813,21 @@ pub(crate) async fn should_skip_lifecycle_for_data_movement( return Ok(false); }; - let versioned = BucketVersioningSys::prefix_enabled(bucket, &version.name).await; + let versioned = match BucketVersioningSys::get_in(&store.ctx, bucket).await { + Ok(config) => config.prefix_enabled(&version.name), + Err(err) => { + warn!( + event = EVENT_DECOMMISSION_ENTRY, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_POOLS, + state = "versioning_config_unavailable", + bucket = %bucket, + error = %err, + "Decommission lifecycle versioning config unavailable; treating object as unversioned" + ); + false + } + }; let object_info = crate::object_api::ObjectInfo::from_file_info(version, bucket, &version.name, versioned); let event = eval_action_from_lifecycle(lifecycle_config, object_lock_config, &object_info).await; @@ -13293,7 +13308,7 @@ impl ECStore { let _ = resolve_decommission_optional_bucket_config_result( &bi.name, "versioning", - BucketVersioningSys::get(&bi.name).await, + BucketVersioningSys::get_in(&self.ctx, &bi.name).await, )?; let expiry_configs = get_expiry_configs(self, &bi.name).await?; lifecycle_config = expiry_configs.lifecycle.map(|config| (*config).clone()); @@ -13301,7 +13316,7 @@ impl ECStore { replication_config = resolve_decommission_optional_bucket_config_result( &bi.name, "replication", - metadata_sys::get_replication_config(&bi.name).await, + metadata_sys::get_replication_config_in(&self.ctx, &bi.name).await, )?; } @@ -15635,7 +15650,7 @@ impl ECStore { replication_configured = resolve_decommission_optional_bucket_config_result( &bucket_info.name, "replication", - metadata_sys::get_replication_config(&bucket_info.name).await, + metadata_sys::get_replication_config_in(&self.ctx, &bucket_info.name).await, )? .is_some(); } @@ -17200,7 +17215,14 @@ mod tests { .decommission .as_ref() .expect("the blocked decommission state should remain present"); - assert!(!info.complete && !info.failed && !info.canceled); + assert!( + !info.complete && !info.failed && !info.canceled, + "capacity-blocked state unexpectedly became terminal: complete={}, failed={}, canceled={}, blocked_reason={:?}", + info.complete, + info.failed, + info.canceled, + info.capacity_blocked_reason + ); assert_eq!(info.items_decommission_failed, 0); assert_eq!(info.bytes_failed, 0); assert!(info.capacity_blocked_reason.is_some()); diff --git a/crates/ecstore/src/store/init.rs b/crates/ecstore/src/store/init.rs index 334ace3da..f1505de4d 100644 --- a/crates/ecstore/src/store/init.rs +++ b/crates/ecstore/src/store/init.rs @@ -5210,15 +5210,25 @@ mod tests { let fault_bucket = other_bucket.clone(); let _fault_guard = crate::core::pools::DecommissionTestFaultGuard::install(Arc::new( move |stage, bucket, object, attempt, succeeded| { - let injected = succeeded + let candidate = succeeded && stage == DECOMMISSION_TEST_FAULT_STAGE_MIGRATE_OBJECT && bucket == fault_bucket.as_str() - && object == other_object - && attempt < crate::core::pools::DECOMMISSION_VERSION_COPY_ATTEMPTS; - if injected { - ordinary_faults_for_hook.fetch_add(1, Ordering::SeqCst); + && object == other_object; + if !candidate { + return false; } - injected + + // Keep the fault budget global across any + // entry-level re-list; its inner attempt counter + // restarts after SourceChanged. + ordinary_faults_for_hook + .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |faults| { + let next_fault = faults.saturating_add(1); + (faults < crate::core::pools::DECOMMISSION_VERSION_COPY_ATTEMPTS.saturating_sub(1) + && attempt == next_fault) + .then_some(next_fault) + }) + .is_ok() }, ));