fix(ecstore): stabilize decommission config and retry tests (#7091)

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<Option<NotificationConfiguration>> {
     let bucket_meta_sys_lock = get_bucket_metadata_sys()?;
     let bucket_meta_sys = bucket_meta_sys_lock.read().await;

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<VersioningConfiguration> {
         if bucket == RUSTFS_META_BUCKET || bucket.starts_with(RUSTFS_META_BUCKET) {
             return Ok(VersioningConfiguration::default());

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());

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()
                         },
                     ));