test(ecstore): cover final sweep cancel fence

crates/ecstore/src/core/pools.rs

@@ -5014,10 +5014,7 @@ impl ECStore {
                     state = "verifying_completion",
                     "Decommission completion verification started"
                 );
-                self.ensure_decommission_generation_current(idx, generation).await?;
-                let operation_gate = self.ctx.decommission_operation_gate();
-                if let Err(err) = run_decommission_side_effect(&rx, &operation_gate, || self.check_after_decommission(idx)).await
-                {
+                if let Err(err) = self.check_after_decommission(idx, &rx, generation).await {
                     if is_err_operation_canceled(&err) {
                         return Err(err);
                     }
@@ -6516,7 +6513,18 @@ impl ECStore {
         self.cleanup_decommission_durable_ilm_receipts(source_pool_idx).await
     }
 
-    async fn check_after_decommission(self: &Arc<Self>, idx: usize) -> Result<()> {
+    async fn check_after_decommission(
+        self: &Arc<Self>,
+        idx: usize,
+        rx: &CancellationToken,
+        generation: OffsetDateTime,
+    ) -> Result<()> {
+        self.ensure_decommission_generation_current(idx, generation).await?;
+        let operation_gate = self.ctx.decommission_operation_gate();
+        run_decommission_side_effect(rx, &operation_gate, || self.check_after_decommission_unfenced(idx)).await
+    }
+
+    async fn check_after_decommission_unfenced(self: &Arc<Self>, idx: usize) -> Result<()> {
         let buckets = self.get_buckets_to_decommission().await?;
         let pool = self.pools[idx].clone();
 
@@ -6690,7 +6698,9 @@ impl ECStore {
 
     #[cfg(test)]
     pub(crate) async fn check_after_decommission_for_test(self: &Arc<Self>, idx: usize) -> Result<()> {
-        self.check_after_decommission(idx).await
+        let generation = self.active_decommission_generation(idx).await?;
+        self.check_after_decommission(idx, &CancellationToken::new(), generation)
+            .await
     }
 
     #[tracing::instrument(skip(self, rd))]

crates/ecstore/src/store/init.rs

@@ -4575,6 +4575,70 @@ mod tests {
         assert!(backend.remove_versions().await.contains(&(entry.obj_name, entry.version_id)));
     }
 
+    #[cfg(feature = "test-util")]
+    #[tokio::test]
+    #[serial_test::serial(storage_class_env)]
+    async fn decommission_final_sweep_blocks_cancel_until_source_cleanup_finishes() {
+        let temp_dir = tempfile::tempdir().expect("create final sweep gate store dir");
+        let (_ctx, store, _shutdown) =
+            without_storage_class_env(build_isolated_test_store(temp_dir.path(), "durable-ilm-final-sweep-gate", &[4, 4])).await;
+        let job_id = uuid::Uuid::new_v4();
+        let job = ManualTransitionJobRecord::new(job_id, "final-sweep-gate", &ManualTransitionRunOptions::default(), "owner");
+        let path = manual_transition_job_record_object_name(job_id).expect("manual job path should build");
+        let data = job.encode().expect("manual job should encode");
+        for pool in &store.pools {
+            com::save_config(pool.clone(), &path, data.clone())
+                .await
+                .expect("manual job fixture should persist in both pools");
+        }
+        let active_pool_meta = {
+            let mut pool_meta = store.pool_meta.write().await;
+            pool_meta.pools[0].decommission = Some(PoolDecommissionInfo {
+                start_time: Some(OffsetDateTime::now_utc()),
+                ..Default::default()
+            });
+            pool_meta.clone()
+        };
+        active_pool_meta
+            .save(store.pools.clone())
+            .await
+            .expect("active decommission run identity should persist");
+
+        let barrier = SourceCleanupDeleteBarrier::install(RUSTFS_META_BUCKET, &path);
+        let final_sweep = tokio::spawn({
+            let store = store.clone();
+            async move { store.check_after_decommission_for_test(0).await }
+        });
+        barrier.wait_until_paused().await;
+
+        let mut cancel = tokio::spawn({
+            let store = store.clone();
+            async move { store.decommission_cancel(0).await }
+        });
+        assert!(
+            tokio::time::timeout(StdDuration::from_millis(100), &mut cancel)
+                .await
+                .is_err(),
+            "cancel must wait for the final sweep source cleanup"
+        );
+
+        barrier.release();
+        final_sweep
+            .await
+            .expect("final sweep task should not panic")
+            .expect("final sweep should finish after the barrier releases");
+        cancel
+            .await
+            .expect("cancel task should not panic")
+            .expect("cancel should complete after the final sweep releases the operation gate");
+        assert!(
+            store.pool_meta.read().await.pools[0]
+                .decommission
+                .as_ref()
+                .is_some_and(|info| info.canceled)
+        );
+    }
+
     #[cfg(feature = "test-util")]
     #[tokio::test]
     #[serial_test::serial(storage_class_env)]