Merge branch 'main' into cxymds/perf-decommission-checkpoint

crates/config/src/constants/runtime.rs

@@ -57,13 +57,6 @@ pub const DEFAULT_MAX_IO_EVENTS_PER_TICK: usize = 1024;
 pub const DEFAULT_EVENT_INTERVAL: u32 = 61;
 pub const DEFAULT_RNG_SEED: Option<u64> = None; // None means random
 
-/// Dedicated blocking thread pool for fsync/fdatasync operations.
-/// When > 1, fsync operations are isolated from the main blocking pool to
-/// prevent device-bound fsync from starving read operations (pread/stat/open).
-/// Default 0 means auto (no isolation, use main runtime).
-pub const ENV_FSYNC_BLOCKING_THREADS: &str = "RUSTFS_RUNTIME_FSYNC_BLOCKING_THREADS";
-pub const DEFAULT_FSYNC_BLOCKING_THREADS: usize = 0;
-
 // Dial9 Tokio Telemetry Default values
 pub const DEFAULT_RUNTIME_DIAL9_ENABLED: bool = false; // Disabled by default
 pub const DEFAULT_RUNTIME_DIAL9_OUTPUT_DIR: &str = "/var/log/rustfs/telemetry";

crates/ecstore/src/core/pools.rs

@@ -89,6 +89,7 @@ const DECOMMISSION_STAGE_SOURCE_CLEANUP: &str = "source_cleanup";
 const DECOMMISSION_STAGE_ENTRY_FINISHED: &str = "entry_finished";
 const DECOMMISSION_PROGRESS_SAVE_INTERVAL: Duration = Duration::seconds(30);
 const DECOMMISSION_PROGRESS_SAVE_ITEM_THRESHOLD: usize = 1000;
+const DECOMMISSION_PROGRESS_SAVE_RETRY_BACKOFF: Duration = Duration::seconds(1);
 const DECOMMISSION_BUCKET_CONCURRENCY_ENV: &str = "RUSTFS_DECOMMISSION_BUCKET_CONCURRENCY";
 const DECOMMISSION_BUCKET_CONCURRENCY_DEFAULT_CAP: usize = 4;
 const DECOMMISSION_TARGET_CAPACITY_OVERHEAD_PERCENT: usize = 30;
@@ -638,22 +639,6 @@ fn track_decommission_current_object(meta: &mut PoolMeta, idx: usize, bucket: &s
     track_decommission_current_object_stage(meta, idx, bucket, object, "")
 }
 
-fn touch_decommission_progress(meta: &mut PoolMeta, idx: usize) -> Result<()> {
-    let pool_count = meta.pools.len();
-    ensure_valid_decommission_pool_index(pool_count, idx)?;
-
-    let Some(pool) = meta.pools.get_mut(idx) else {
-        return Err(invalid_decommission_pool_index_error(pool_count, idx));
-    };
-    let Some(info) = pool.decommission.as_mut() else {
-        return Err(decommission_metadata_not_initialized_error("touch decommission progress"));
-    };
-
-    pool.last_update = OffsetDateTime::now_utc();
-    info.mark_progress_saved();
-    Ok(())
-}
-
 fn resolve_decommission_update_after_result(result: Result<bool>) -> Result<bool> {
     result.map_err(|err| Error::other(format!("decommission metadata update failed: {err}")))
 }
@@ -1483,6 +1468,7 @@ impl TryFrom<PersistedPoolDecommissionInfo> for PoolDecommissionInfo {
             terminal_reload_attempt_at: value.terminal_reload_attempt_at,
             terminal_reload_failures: value.terminal_reload_failures,
             progress_save_item_baseline: value.items_decommissioned.saturating_add(value.items_decommission_failed),
+            progress_save_retry_after: None,
         })
     }
 }
@@ -1514,6 +1500,7 @@ impl TryFrom<LegacyPoolDecommissionInfo> for PoolDecommissionInfo {
             terminal_reload_attempt_at: None,
             terminal_reload_failures: Vec::new(),
             progress_save_item_baseline: value.items_decommissioned.saturating_add(value.items_decommission_failed),
+            progress_save_retry_after: None,
         })
     }
 }
@@ -1627,6 +1614,82 @@ impl PoolMeta {
         }
     }
 
+    fn decommission_progress_checkpoint(
+        &self,
+        idx: usize,
+        duration: Duration,
+        now: OffsetDateTime,
+    ) -> Result<Option<DecommissionProgressCheckpoint>> {
+        let pool_count = self.pools.len();
+        ensure_valid_decommission_pool_index(pool_count, idx)?;
+
+        let Some(pool) = self.pools.get(idx) else {
+            return Err(invalid_decommission_pool_index_error(pool_count, idx));
+        };
+        let Some(info) = pool.decommission.as_ref() else {
+            return Err(decommission_metadata_not_initialized_error("update decommission metadata timestamp"));
+        };
+
+        if info.progress_save_retry_after.is_some_and(|retry_after| now < retry_after) {
+            return Ok(None);
+        }
+
+        let time_threshold_reached = now.unix_timestamp() - pool.last_update.unix_timestamp() >= duration.whole_seconds();
+        let item_threshold_reached = info.items_since_last_progress_save() >= DECOMMISSION_PROGRESS_SAVE_ITEM_THRESHOLD;
+        if !time_threshold_reached && !item_threshold_reached {
+            return Ok(None);
+        }
+
+        Ok(Some(DecommissionProgressCheckpoint {
+            start_time: info.start_time,
+            queued: info.queued,
+            counted_items: info.counted_items(),
+            checkpoint_at: now,
+        }))
+    }
+
+    fn commit_decommission_progress_checkpoint(&mut self, idx: usize, checkpoint: DecommissionProgressCheckpoint) -> bool {
+        let Some(pool) = self.pools.get_mut(idx) else {
+            return false;
+        };
+        let Some(info) = pool.decommission.as_mut() else {
+            return false;
+        };
+
+        if info.start_time != checkpoint.start_time
+            || info.queued != checkpoint.queued
+            || !is_decommission_active(info.complete, info.failed, info.canceled)
+        {
+            return false;
+        }
+
+        info.progress_save_item_baseline = info.progress_save_item_baseline.max(checkpoint.counted_items);
+        info.progress_save_retry_after = None;
+        pool.last_update = pool.last_update.max(checkpoint.checkpoint_at);
+        true
+    }
+
+    fn defer_decommission_progress_checkpoint(
+        &mut self,
+        idx: usize,
+        checkpoint: DecommissionProgressCheckpoint,
+        retry_after: OffsetDateTime,
+    ) {
+        let Some(pool) = self.pools.get_mut(idx) else {
+            return;
+        };
+        let Some(info) = pool.decommission.as_mut() else {
+            return;
+        };
+
+        if info.start_time == checkpoint.start_time
+            && info.queued == checkpoint.queued
+            && is_decommission_active(info.complete, info.failed, info.canceled)
+        {
+            info.progress_save_retry_after = Some(retry_after);
+        }
+    }
+
     fn load_from_config_data(&mut self, data: Vec<u8>) -> Result<()> {
         if data.is_empty() {
             return Ok(());
@@ -1675,7 +1738,7 @@ impl PoolMeta {
         self.load_no_lock(pool).await
     }
 
-    pub(crate) async fn load_no_lock<S>(&mut self, pool: Arc<S>) -> Result<()>
+    async fn load_no_lock<S>(&mut self, pool: Arc<S>) -> Result<()>
     where
         S: EcstoreObjectIO,
     {
@@ -1987,30 +2050,9 @@ impl PoolMeta {
     }
 
     pub fn update_after(&mut self, idx: usize, duration: Duration) -> Result<bool> {
-        let pool_count = self.pools.len();
-        ensure_valid_decommission_pool_index(pool_count, idx)?;
-
-        let (last_update, item_threshold_reached) = match self.pools.get(idx) {
-            Some(pool) if let Some(info) = pool.decommission.as_ref() => (
-                pool.last_update,
-                info.items_since_last_progress_save() >= DECOMMISSION_PROGRESS_SAVE_ITEM_THRESHOLD,
-            ),
-            Some(_) => {
-                return Err(decommission_metadata_not_initialized_error("update decommission metadata timestamp"));
-            }
-            None => return Err(invalid_decommission_pool_index_error(pool_count, idx)),
-        };
-        let now = OffsetDateTime::now_utc();
-
-        if now.unix_timestamp() - last_update.unix_timestamp() >= duration.whole_seconds() || item_threshold_reached {
-            let Some(pool) = self.pools.get_mut(idx) else {
-                return Err(invalid_decommission_pool_index_error(pool_count, idx));
-            };
-            pool.last_update = now;
-            return Ok(true);
-        }
-
-        Ok(false)
+        Ok(self
+            .decommission_progress_checkpoint(idx, duration, OffsetDateTime::now_utc())?
+            .is_some())
     }
 
     pub fn validate(&self, pools: Vec<Arc<Sets>>) -> Result<bool> {
@@ -2151,6 +2193,16 @@ pub struct PoolDecommissionInfo {
     pub terminal_reload_failures: Vec<String>,
     #[serde(skip)]
     pub progress_save_item_baseline: usize,
+    #[serde(skip)]
+    pub progress_save_retry_after: Option<OffsetDateTime>,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+struct DecommissionProgressCheckpoint {
+    start_time: Option<OffsetDateTime>,
+    queued: bool,
+    counted_items: usize,
+    checkpoint_at: OffsetDateTime,
 }
 
 impl PoolDecommissionInfo {
@@ -2185,6 +2237,7 @@ impl PoolDecommissionInfo {
 
     fn mark_progress_saved(&mut self) {
         self.progress_save_item_baseline = self.counted_items();
+        self.progress_save_retry_after = None;
     }
 
     pub fn bucket_push(&mut self, bucket: &DecomBucketInfo) {
@@ -2489,6 +2542,40 @@ impl ECStore {
         snapshot.save(self.pools.clone()).await
     }
 
+    async fn save_decommission_progress_checkpoint(&self, idx: usize) -> Result<bool> {
+        // Lock order: save gate, then the short pool metadata read/write sections. Peer
+        // reloads are intentionally performed by the caller after both locks are released.
+        let _save_guard = self.pool_meta_save_gate.lock().await;
+        let (snapshot, checkpoint) = {
+            let pool_meta = self.pool_meta.read().await;
+            let Some(checkpoint) = pool_meta.decommission_progress_checkpoint(
+                idx,
+                DECOMMISSION_PROGRESS_SAVE_INTERVAL,
+                OffsetDateTime::now_utc(),
+            )?
+            else {
+                return Ok(false);
+            };
+
+            let mut snapshot = pool_meta.clone();
+            let Some(pool) = snapshot.pools.get_mut(idx) else {
+                return Err(invalid_decommission_pool_index_error(snapshot.pools.len(), idx));
+            };
+            pool.last_update = checkpoint.checkpoint_at;
+            (snapshot, checkpoint)
+        };
+
+        if let Err(err) = snapshot.save(self.pools.clone()).await {
+            let retry_after = OffsetDateTime::now_utc() + DECOMMISSION_PROGRESS_SAVE_RETRY_BACKOFF;
+            let mut pool_meta = self.pool_meta.write().await;
+            pool_meta.defer_decommission_progress_checkpoint(idx, checkpoint, retry_after);
+            return Err(err);
+        }
+
+        let mut pool_meta = self.pool_meta.write().await;
+        Ok(pool_meta.commit_decommission_progress_checkpoint(idx, checkpoint))
+    }
+
     async fn save_current_pool_meta_for_decommission_start(
         &self,
         indices: &[usize],
@@ -2871,7 +2958,7 @@ impl ECStore {
         Ok(())
     }
 
-    async fn save_decommission_entry_progress_stage(
+    async fn track_decommission_entry_progress_stage(
         &self,
         idx: usize,
         bucket: &str,
@@ -2882,22 +2969,6 @@ impl ECStore {
             let mut pool_meta = self.pool_meta.write().await;
             track_decommission_current_object_stage(&mut pool_meta, idx, bucket, object, stage)
                 .map_err(|err| with_decommission_entry_context(stage, bucket, object, err))?;
-            touch_decommission_progress(&mut pool_meta, idx)
-                .map_err(|err| with_decommission_entry_context(stage, bucket, object, err))?;
-        }
-
-        if let Some(err) = resolve_decommission_progress_save_result(self.save_current_pool_meta().await) {
-            warn!(
-                event = EVENT_DECOMMISSION_ENTRY,
-                component = LOG_COMPONENT_ECSTORE,
-                subsystem = LOG_SUBSYSTEM_POOLS,
-                pool_index = idx,
-                bucket = %bucket,
-                object = %object,
-                stage,
-                error = ?err,
-                "Decommission progress stage save failed"
-            );
         }
 
         Ok(())
@@ -3165,7 +3236,7 @@ impl ECStore {
                 let bucket_name = bucket.clone();
                 let object_name = rd.object_info.name.clone();
 
-                self.save_decommission_entry_progress_stage(
+                self.track_decommission_entry_progress_stage(
                     idx,
                     bucket_name.as_str(),
                     object_name.as_str(),
@@ -3259,7 +3330,7 @@ impl ECStore {
             }
             decommission_cancel_signal_result(rx.is_cancelled())?;
 
-            self.save_decommission_entry_progress_stage(
+            self.track_decommission_entry_progress_stage(
                 idx,
                 bucket.as_str(),
                 entry.name.as_str(),
@@ -3267,7 +3338,7 @@ impl ECStore {
             )
             .await?;
 
-            self.save_decommission_entry_progress_stage(
+            self.track_decommission_entry_progress_stage(
                 idx,
                 bucket.as_str(),
                 entry.name.as_str(),
@@ -3334,34 +3405,42 @@ impl ECStore {
             }
         };
 
-        self.save_decommission_entry_progress_stage(idx, bucket.as_str(), entry.name.as_str(), DECOMMISSION_STAGE_ENTRY_FINISHED)
-            .await?;
+        self.track_decommission_entry_progress_stage(
+            idx,
+            bucket.as_str(),
+            entry.name.as_str(),
+            DECOMMISSION_STAGE_ENTRY_FINISHED,
+        )
+        .await?;
 
         if should_save_progress {
-            let save_result = self.save_current_pool_meta().await;
-            if let Some(err) = resolve_decommission_progress_save_result(save_result) {
-                warn!(
-                    event = EVENT_DECOMMISSION_ENTRY,
-                    component = LOG_COMPONENT_ECSTORE,
-                    subsystem = LOG_SUBSYSTEM_POOLS,
-                    pool_index = idx,
-                    bucket = %bucket,
-                    object = %entry.name,
-                    state = "progress_save_failed",
-                    error = %err,
-                    "Decommission progress save failed; continuing and will retry at the next checkpoint"
-                );
-            } else {
-                let mut pool_meta = self.pool_meta.write().await;
-                pool_meta.mark_decommission_progress_saved();
-                if let Some(notification_sys) = runtime_sources::notification_sys()
-                    && let Err(err) = resolve_decommission_entry_reload_result(
-                        notification_sys.reload_pool_meta().await,
-                        bucket.as_str(),
-                        entry.name.as_str(),
-                    )
-                {
-                    warn!("{err}");
+            match self.save_decommission_progress_checkpoint(idx).await {
+                Ok(true) => {
+                    if let Some(notification_sys) = runtime_sources::notification_sys()
+                        && let Err(err) = resolve_decommission_entry_reload_result(
+                            notification_sys.reload_pool_meta().await,
+                            bucket.as_str(),
+                            entry.name.as_str(),
+                        )
+                    {
+                        warn!("{err}");
+                    }
+                }
+                Ok(false) => {}
+                Err(err) => {
+                    if let Some(err) = resolve_decommission_progress_save_result(Err(err)) {
+                        warn!(
+                            event = EVENT_DECOMMISSION_ENTRY,
+                            component = LOG_COMPONENT_ECSTORE,
+                            subsystem = LOG_SUBSYSTEM_POOLS,
+                            pool_index = idx,
+                            bucket = %bucket,
+                            object = %entry.name,
+                            state = "progress_save_failed",
+                            error = %err,
+                            "Decommission progress save failed; continuing and will retry at the next checkpoint"
+                        );
+                    }
                 }
             }
         }
@@ -5264,11 +5343,11 @@ pub(crate) fn fallback_free_capacity_dedup(disks: &[rustfs_madmin::Disk]) -> usi
 #[cfg(test)]
 mod pools_tests {
     use super::{
-        DECOMMISSION_PROGRESS_SAVE_INTERVAL, DECOMMISSION_PROGRESS_SAVE_ITEM_THRESHOLD, DecomBucketInfo,
-        DecommissionStartPoolState, DecommissionTerminalState, ListCallback, PoolDecommissionInfo, PoolMeta, PoolSpaceInfo,
-        PoolStatus, apply_decommission_status_space_info, bind_decommission_cancelers, bind_missing_decommission_cancelers,
-        cancel_decommission_canceler, classify_decommission_terminal_state, count_decommission_item,
-        decommission_cancel_signal_result, decommission_item_size, decommission_meta_bucket_options,
+        DECOMMISSION_PROGRESS_SAVE_INTERVAL, DECOMMISSION_PROGRESS_SAVE_ITEM_THRESHOLD, DECOMMISSION_PROGRESS_SAVE_RETRY_BACKOFF,
+        DecomBucketInfo, DecommissionStartPoolState, DecommissionTerminalState, ListCallback, PoolDecommissionInfo, PoolMeta,
+        PoolSpaceInfo, PoolStatus, apply_decommission_status_space_info, bind_decommission_cancelers,
+        bind_missing_decommission_cancelers, cancel_decommission_canceler, classify_decommission_terminal_state,
+        count_decommission_item, decommission_cancel_signal_result, decommission_item_size, decommission_meta_bucket_options,
         decommission_start_pool_state, dedup_indices, default_decommission_bucket_concurrency,
         ensure_decommission_cancel_allowed, ensure_decommission_clear_allowed, ensure_decommission_listing_disks_available,
         ensure_decommission_not_rebalancing, ensure_decommission_start_allowed, ensure_decommission_start_keeps_active_pool,
@@ -5293,9 +5372,8 @@ mod pools_tests {
         should_preserve_decommission_canceled_state, should_reject_decommission_cancel_as_terminal,
         should_retry_decommission_cancel_reload, should_retry_decommission_listing, should_skip_canceled_decommission_routine,
         split_decommission_buckets, take_and_cancel_decommission_canceler, take_decommission_canceler,
-        touch_decommission_progress, track_decommission_current_object, track_decommission_current_object_stage,
-        validate_start_decommission_request, wait_decommission_listing_retry, wait_decommission_worker_drain,
-        with_decommission_entry_context,
+        track_decommission_current_object, track_decommission_current_object_stage, validate_start_decommission_request,
+        wait_decommission_listing_retry, wait_decommission_worker_drain, with_decommission_entry_context,
     };
     use crate::data_movement;
     use crate::disk::endpoint::Endpoint;
@@ -6538,7 +6616,7 @@ mod pools_tests {
     }
 
     #[test]
-    fn test_touch_decommission_progress_updates_last_update_and_save_baseline() {
+    fn test_track_decommission_stage_does_not_advance_checkpoint_state() {
         let mut meta = PoolMeta {
             pools: vec![PoolStatus {
                 id: 0,
@@ -6553,11 +6631,13 @@ mod pools_tests {
             ..Default::default()
         };
 
-        touch_decommission_progress(&mut meta, 0).expect("valid decommission progress should be touched");
+        track_decommission_current_object_stage(&mut meta, 0, "bucket", "object", "migrate_object")
+            .expect("valid decommission progress should be tracked");
 
-        assert!(meta.pools[0].last_update > OffsetDateTime::UNIX_EPOCH);
+        assert_eq!(meta.pools[0].last_update, OffsetDateTime::UNIX_EPOCH);
         let info = meta.pools[0].decommission.as_ref().expect("decommission info should exist");
-        assert_eq!(info.items_since_last_progress_save(), 0);
+        assert_eq!(info.items_since_last_progress_save(), 5);
+        assert_eq!(info.stage, "migrate_object");
     }
 
     #[test]
@@ -6632,6 +6712,134 @@ mod pools_tests {
         assert_eq!(info.items_since_last_progress_save(), 1);
     }
 
+    #[test]
+    fn test_pool_meta_update_after_does_not_advance_last_update_before_save() {
+        let last_update = OffsetDateTime::UNIX_EPOCH;
+        let mut meta = PoolMeta {
+            pools: vec![PoolStatus {
+                id: 0,
+                cmd_line: "pool-0".to_string(),
+                last_update,
+                decommission: Some(PoolDecommissionInfo {
+                    start_time: Some(last_update),
+                    items_decommissioned: DECOMMISSION_PROGRESS_SAVE_ITEM_THRESHOLD,
+                    ..Default::default()
+                }),
+            }],
+            ..Default::default()
+        };
+
+        assert!(
+            meta.update_after(0, DECOMMISSION_PROGRESS_SAVE_INTERVAL)
+                .expect("item threshold should request a checkpoint")
+        );
+        assert_eq!(meta.pools[0].last_update, last_update);
+    }
+
+    #[test]
+    fn test_decommission_progress_checkpoint_commits_exact_snapshot_watermark() {
+        let start_time = OffsetDateTime::UNIX_EPOCH;
+        let checkpoint_at = start_time + Duration::seconds(30);
+        let mut meta = PoolMeta {
+            pools: vec![PoolStatus {
+                id: 0,
+                cmd_line: "pool-0".to_string(),
+                last_update: start_time,
+                decommission: Some(PoolDecommissionInfo {
+                    start_time: Some(start_time),
+                    items_decommissioned: DECOMMISSION_PROGRESS_SAVE_ITEM_THRESHOLD,
+                    ..Default::default()
+                }),
+            }],
+            ..Default::default()
+        };
+
+        let checkpoint = meta
+            .decommission_progress_checkpoint(0, DECOMMISSION_PROGRESS_SAVE_INTERVAL, checkpoint_at)
+            .expect("valid decommission state should produce a checkpoint")
+            .expect("item threshold should produce a checkpoint");
+        meta.count_item(0, 1, false);
+
+        assert!(meta.commit_decommission_progress_checkpoint(0, checkpoint));
+        let info = meta.pools[0].decommission.as_ref().expect("decommission info should exist");
+        assert_eq!(info.progress_save_item_baseline, checkpoint.counted_items);
+        assert_eq!(info.items_since_last_progress_save(), 1);
+        assert_eq!(meta.pools[0].last_update, checkpoint_at);
+    }
+
+    #[test]
+    fn test_decommission_progress_checkpoint_backoff_does_not_advance_baseline() {
+        let start_time = OffsetDateTime::UNIX_EPOCH;
+        let checkpoint_at = start_time + Duration::seconds(30);
+        let retry_after = checkpoint_at + DECOMMISSION_PROGRESS_SAVE_RETRY_BACKOFF;
+        let mut meta = PoolMeta {
+            pools: vec![PoolStatus {
+                id: 0,
+                cmd_line: "pool-0".to_string(),
+                last_update: start_time,
+                decommission: Some(PoolDecommissionInfo {
+                    start_time: Some(start_time),
+                    items_decommissioned: DECOMMISSION_PROGRESS_SAVE_ITEM_THRESHOLD,
+                    ..Default::default()
+                }),
+            }],
+            ..Default::default()
+        };
+
+        let checkpoint = meta
+            .decommission_progress_checkpoint(0, DECOMMISSION_PROGRESS_SAVE_INTERVAL, checkpoint_at)
+            .expect("valid decommission state should produce a checkpoint")
+            .expect("item threshold should produce a checkpoint");
+        meta.defer_decommission_progress_checkpoint(0, checkpoint, retry_after);
+
+        assert!(
+            meta.decommission_progress_checkpoint(0, DECOMMISSION_PROGRESS_SAVE_INTERVAL, checkpoint_at)
+                .expect("retry backoff check should succeed")
+                .is_none()
+        );
+        assert_eq!(meta.pools[0].last_update, start_time);
+        assert_eq!(
+            meta.pools[0]
+                .decommission
+                .as_ref()
+                .expect("decommission info should exist")
+                .progress_save_item_baseline,
+            0
+        );
+    }
+
+    #[test]
+    fn test_decommission_progress_checkpoint_count_scales_with_threshold() {
+        let start_time = OffsetDateTime::UNIX_EPOCH;
+        let checkpoint_at = start_time;
+        let mut meta = PoolMeta {
+            pools: vec![PoolStatus {
+                id: 0,
+                cmd_line: "pool-0".to_string(),
+                last_update: start_time,
+                decommission: Some(PoolDecommissionInfo {
+                    start_time: Some(start_time),
+                    ..Default::default()
+                }),
+            }],
+            ..Default::default()
+        };
+        let mut checkpoint_count = 0;
+
+        for _ in 0..(DECOMMISSION_PROGRESS_SAVE_ITEM_THRESHOLD * 10) {
+            meta.count_item(0, 1, false);
+            if let Some(checkpoint) = meta
+                .decommission_progress_checkpoint(0, DECOMMISSION_PROGRESS_SAVE_INTERVAL, checkpoint_at)
+                .expect("valid decommission state should produce a checkpoint")
+            {
+                checkpoint_count += 1;
+                assert!(meta.commit_decommission_progress_checkpoint(0, checkpoint));
+            }
+        }
+
+        assert_eq!(checkpoint_count, 10);
+    }
+
     #[test]
     fn test_ensure_decommission_not_rebalancing_rejects_running_rebalance() {
         let err = ensure_decommission_not_rebalancing(true).expect_err("rebalance running should be rejected");

crates/ecstore/src/disk/os.rs

@@ -315,7 +315,7 @@ pub async fn fsync_dir(dir: impl AsRef<Path>) -> io::Result<()> {
     #[cfg(unix)]
     {
         let dir = dir.as_ref().to_path_buf();
-        fsync_spawn_blocking(move || fsync_dir_std(dir)).await?
+        tokio::task::spawn_blocking(move || fsync_dir_std(dir)).await?
     }
 
     #[cfg(not(unix))]
@@ -683,7 +683,7 @@ async fn fsync_open_dst_dir_group(group: &DstDirFsyncGroup) -> io::Result<()> {
     #[cfg(test)]
     let dir = group.dir.clone();
     let dir_file = group.dir_file.clone();
-    fsync_spawn_blocking(move || {
+    tokio::task::spawn_blocking(move || {
         #[cfg(test)]
         {
             if let Some(kind) = fsync_dir_recorder::take_grouped_failure(&dir) {
@@ -1080,44 +1080,6 @@ const TEST_GLOBAL_FILE_SYNCS: usize = 64;
 
 static FILE_SYNC_PERMITS: LazyLock<Semaphore> = LazyLock::new(|| Semaphore::new(global_file_sync_limit()));
 static DISK_FILE_SYNC_LIMITERS: LazyLock<Mutex<HashMap<PathBuf, Weak<Semaphore>>>> = LazyLock::new(|| Mutex::new(HashMap::new()));
-
-/// Dedicated tokio runtime for fsync/fdatasync blocking operations. When
-/// configured with >1 threads, isolates device-bound fsync from the main
-/// blocking pool so reads (pread/stat/open) are not starved. `None` means
-/// fall back to the main runtime (zero behavior change).
-static FSYNC_RUNTIME: LazyLock<Option<tokio::runtime::Runtime>> = LazyLock::new(|| {
-    let threads =
-        rustfs_utils::get_env_usize(rustfs_config::ENV_FSYNC_BLOCKING_THREADS, rustfs_config::DEFAULT_FSYNC_BLOCKING_THREADS);
-    if threads <= 1 {
-        return None;
-    }
-    let mut builder = tokio::runtime::Builder::new_multi_thread();
-    builder
-        .worker_threads(num_cpus::get().min(8))
-        .max_blocking_threads(threads)
-        .thread_name("rustfs-fsync")
-        .thread_stack_size(512 * 1024)
-        .enable_all();
-    match builder.build() {
-        Ok(rt) => {
-            tracing::info!(threads, "fsync dedicated blocking pool enabled");
-            Some(rt)
-        }
-        Err(err) => {
-            tracing::warn!(%err, "failed to build fsync runtime, falling back to main pool");
-            None
-        }
-    }
-});
-
-/// Spawn a blocking task on the fsync-dedicated runtime if configured,
-/// otherwise fall back to the main tokio blocking pool.
-fn fsync_spawn_blocking<T: Send + 'static>(f: impl FnOnce() -> T + Send + 'static) -> tokio::task::JoinHandle<T> {
-    match FSYNC_RUNTIME.as_ref() {
-        Some(rt) => rt.spawn_blocking(f),
-        None => tokio::task::spawn_blocking(f),
-    }
-}
 static DISK_VOLUME_MUTATION_LOCKS: LazyLock<Mutex<HashMap<PathBuf, Weak<RwLock<()>>>>> =
     LazyLock::new(|| Mutex::new(HashMap::new()));
 type NamespaceMutationLock = AsyncMutex<()>;
@@ -1255,7 +1217,7 @@ where
     F: FnOnce() -> io::Result<T> + Send + 'static,
 {
     let (disk_permit, global_permit) = acquire_file_sync_permits(disk_permits).await?;
-    let result = fsync_spawn_blocking(move || {
+    let result = tokio::task::spawn_blocking(move || {
         let _disk_permit = disk_permit;
         work()
     })
@@ -2184,7 +2146,7 @@ async fn run_blocking_namespace_file_sync_operation_with_global<T: Send + 'stati
         wait_started,
     );
     let disk_permit = admission.disk_permit.clone();
-    let result = fsync_spawn_blocking(move || {
+    let result = tokio::task::spawn_blocking(move || {
         let _lease = lease;
         let _disk_permit = disk_permit;
         operation()

crates/ecstore/src/store/heal.rs

@@ -13,12 +13,7 @@
 // limitations under the License.
 
 use super::*;
-use crate::core::pools::POOL_META_NAME;
-use crate::services::rebalance::{REBAL_META_NAME, RebalStatus};
-use crate::set_disk::get_lock_acquire_timeout;
 use crate::storage_api_contracts::heal::HealOperations as _;
-use crate::storage_api_contracts::namespace::NamespaceLocking as _;
-use rustfs_lock::NamespaceLockGuard;
 use tracing::trace;
 
 const LOG_COMPONENT_ECSTORE: &str = "ecstore";
@@ -35,119 +30,7 @@ fn invalid_heal_pool_index(pool_idx: usize, pool_count: usize) -> Error {
     )
 }
 
-#[derive(Debug, Clone, Copy)]
-enum HealFormatPoolSkip {
-    Completed,
-    Retryable,
-}
-
-fn classify_heal_format_pool(
-    pool_idx: usize,
-    pool_cmd_line: &str,
-    pool_meta: &PoolMeta,
-    rebalance_meta: Option<&RebalanceMeta>,
-) -> Option<HealFormatPoolSkip> {
-    let Some(pool) = pool_meta.pools.get(pool_idx) else {
-        return Some(HealFormatPoolSkip::Retryable);
-    };
-
-    if pool.id != pool_idx || pool_cmd_line.is_empty() || pool.cmd_line.is_empty() || pool.cmd_line != pool_cmd_line {
-        return Some(HealFormatPoolSkip::Retryable);
-    }
-
-    if let Some(decommission) = pool.decommission.as_ref() {
-        if decommission.complete {
-            return Some(HealFormatPoolSkip::Completed);
-        }
-        if decommission.failed || decommission.canceled || decommission.queued || pool_meta.is_suspended(pool_idx) {
-            return Some(HealFormatPoolSkip::Retryable);
-        }
-    }
-
-    if let Some(meta) = rebalance_meta {
-        let Some(pool_stats) = meta.pool_stats.get(pool_idx) else {
-            return Some(HealFormatPoolSkip::Retryable);
-        };
-        if pool_stats.info.stopping || (pool_stats.participating && pool_stats.info.status == RebalStatus::Started) {
-            return Some(HealFormatPoolSkip::Retryable);
-        }
-    }
-
-    None
-}
-
-fn heal_format_pool_skip_error(skip: HealFormatPoolSkip) -> Error {
-    match skip {
-        HealFormatPoolSkip::Completed => StorageError::NoHealRequired,
-        HealFormatPoolSkip::Retryable => StorageError::SlowDown,
-    }
-}
-
-fn heal_format_fence_lost_error() -> Error {
-    StorageError::SlowDown
-}
-
 impl ECStore {
-    async fn acquire_heal_format_fence(
-        &self,
-    ) -> Result<(NamespaceLockGuard, NamespaceLockGuard, PoolMeta, Option<RebalanceMeta>)> {
-        let metadata_pool = self
-            .pools
-            .first()
-            .cloned()
-            .ok_or_else(|| Error::other("heal format requires at least one storage pool"))?;
-
-        // Metadata fence order is part of the decommission/rebalance protocol:
-        // pool.bin must always be acquired before rebalance.bin.
-        let pool_lock = metadata_pool.new_ns_lock(RUSTFS_META_BUCKET, POOL_META_NAME).await?;
-        let pool_guard = pool_lock.get_write_lock(get_lock_acquire_timeout()).await?;
-        let rebalance_lock = metadata_pool.new_ns_lock(RUSTFS_META_BUCKET, REBAL_META_NAME).await?;
-        let rebalance_guard = rebalance_lock.get_write_lock(get_lock_acquire_timeout()).await?;
-
-        if pool_guard.is_lock_lost() || rebalance_guard.is_lock_lost() {
-            return Err(heal_format_fence_lost_error());
-        }
-
-        let mut pool_meta = PoolMeta::default();
-        pool_meta.load_no_lock(metadata_pool.clone()).await?;
-        if pool_meta.pools.len() != self.pools.len()
-            || pool_meta.pools.iter().enumerate().any(|(pool_idx, pool)| {
-                pool.id != pool_idx || pool.cmd_line.is_empty() || pool.cmd_line != self.pools[pool_idx].endpoints.cmd_line
-            })
-        {
-            return Err(heal_format_fence_lost_error());
-        }
-
-        let mut rebalance_meta = RebalanceMeta::new();
-        let rebalance_meta = match rebalance_meta
-            .load_with_opts(
-                metadata_pool,
-                ObjectOptions {
-                    no_lock: true,
-                    ..Default::default()
-                },
-            )
-            .await
-        {
-            Ok(()) => Some(rebalance_meta),
-            Err(Error::ConfigNotFound) => None,
-            Err(err) => return Err(err),
-        };
-
-        if rebalance_meta
-            .as_ref()
-            .is_some_and(|meta| meta.pool_stats.len() != self.pools.len())
-        {
-            return Err(heal_format_fence_lost_error());
-        }
-
-        if pool_guard.is_lock_lost() || rebalance_guard.is_lock_lost() {
-            return Err(heal_format_fence_lost_error());
-        }
-
-        Ok((pool_guard, rebalance_guard, pool_meta, rebalance_meta))
-    }
-
     fn get_pools_for_heal_object(&self, opts: &HealOpts) -> Result<Vec<Arc<Sets>>> {
         match opts.pool {
             Some(pool_idx) => Ok(vec![
@@ -169,24 +52,8 @@ impl ECStore {
         };
 
         let mut count_no_heal = 0;
-        let mut count_completed = 0;
         let mut first_error = None;
-        for (pool_idx, pool) in self.pools.iter().enumerate() {
-            let (pool_guard, rebalance_guard, pool_meta, rebalance_meta) = self.acquire_heal_format_fence().await?;
-            if pool_guard.is_lock_lost() || rebalance_guard.is_lock_lost() {
-                first_error.get_or_insert(heal_format_fence_lost_error());
-                break;
-            }
-            if let Some(skip) = classify_heal_format_pool(pool_idx, &pool.endpoints.cmd_line, &pool_meta, rebalance_meta.as_ref())
-            {
-                if matches!(skip, HealFormatPoolSkip::Completed) {
-                    count_completed += 1;
-                } else {
-                    first_error.get_or_insert(heal_format_pool_skip_error(skip));
-                }
-                continue;
-            }
-
+        for pool in self.pools.iter() {
             let (mut result, err) = pool.heal_format(dry_run).await?;
             if let Some(err) = err {
                 match err {
@@ -202,18 +69,11 @@ impl ECStore {
             r.set_count += result.set_count;
             r.before.drives.append(&mut result.before.drives);
             r.after.drives.append(&mut result.after.drives);
-
-            // Sets::heal_format cannot observe this guard before each disk write;
-            // fail closed after the call if the lease was lost during format IO.
-            if pool_guard.is_lock_lost() || rebalance_guard.is_lock_lost() {
-                first_error.get_or_insert(heal_format_fence_lost_error());
-                break;
-            }
         }
         if let Some(err) = first_error {
             return Ok((r, Some(err)));
         }
-        if count_no_heal + count_completed == self.pools.len() {
+        if count_no_heal == self.pools.len() {
             info!(
                 event = EVENT_HEAL_FORMAT_COMPLETED,
                 component = LOG_COMPONENT_ECSTORE,
@@ -440,7 +300,6 @@ mod tests {
     use crate::core::pools::{PoolDecommissionInfo, PoolStatus};
     use crate::disk::{DiskOption, format::FormatV3, new_disk};
     use crate::layout::endpoints::{Endpoints, PoolEndpoints};
-    use crate::services::rebalance::{RebalanceInfo, RebalanceStats};
     use crate::store::init_format::{load_format_erasure, save_format_file};
 
     async fn minimal_heal_pool(pool_idx: usize) -> Arc<Sets> {
@@ -488,164 +347,6 @@ mod tests {
         }
     }
 
-    fn pool_meta_with_decommission(info: PoolDecommissionInfo) -> PoolMeta {
-        PoolMeta {
-            pools: vec![PoolStatus {
-                id: 0,
-                cmd_line: "pool-0".to_string(),
-                last_update: OffsetDateTime::UNIX_EPOCH,
-                decommission: Some(info),
-            }],
-            ..Default::default()
-        }
-    }
-
-    #[test]
-    fn heal_format_pool_state_barriers_are_classified() {
-        let active = pool_meta_with_decommission(PoolDecommissionInfo {
-            start_time: Some(OffsetDateTime::UNIX_EPOCH),
-            ..Default::default()
-        });
-        assert!(matches!(
-            classify_heal_format_pool(0, "pool-0", &active, None),
-            Some(HealFormatPoolSkip::Retryable)
-        ));
-
-        for info in [
-            PoolDecommissionInfo {
-                failed: true,
-                ..Default::default()
-            },
-            PoolDecommissionInfo {
-                canceled: true,
-                ..Default::default()
-            },
-        ] {
-            assert!(matches!(
-                classify_heal_format_pool(0, "pool-0", &pool_meta_with_decommission(info), None),
-                Some(HealFormatPoolSkip::Retryable)
-            ));
-        }
-
-        let completed = pool_meta_with_decommission(PoolDecommissionInfo {
-            complete: true,
-            ..Default::default()
-        });
-        assert!(matches!(
-            classify_heal_format_pool(0, "pool-0", &completed, None),
-            Some(HealFormatPoolSkip::Completed)
-        ));
-    }
-
-    #[test]
-    fn heal_format_pool_rebalance_barriers_and_identity_are_fail_closed() {
-        let identity_meta = pool_meta_with_decommission(PoolDecommissionInfo::default());
-        let rebalance = RebalanceMeta {
-            pool_stats: vec![RebalanceStats {
-                participating: true,
-                info: RebalanceInfo {
-                    status: RebalStatus::Started,
-                    ..Default::default()
-                },
-                ..Default::default()
-            }],
-            ..Default::default()
-        };
-        assert!(matches!(
-            classify_heal_format_pool(0, "pool-0", &identity_meta, Some(&rebalance)),
-            Some(HealFormatPoolSkip::Retryable)
-        ));
-
-        let stopping = RebalanceMeta {
-            pool_stats: vec![RebalanceStats {
-                info: RebalanceInfo {
-                    stopping: true,
-                    ..Default::default()
-                },
-                ..Default::default()
-            }],
-            ..Default::default()
-        };
-        assert!(matches!(
-            classify_heal_format_pool(0, "pool-0", &identity_meta, Some(&stopping)),
-            Some(HealFormatPoolSkip::Retryable)
-        ));
-
-        let identity = pool_meta_with_decommission(PoolDecommissionInfo::default());
-        assert!(matches!(
-            classify_heal_format_pool(0, "pool-new", &identity, None),
-            Some(HealFormatPoolSkip::Retryable)
-        ));
-
-        let identity_without_decommission = PoolMeta {
-            pools: vec![PoolStatus {
-                id: 0,
-                cmd_line: "pool-0".to_string(),
-                last_update: OffsetDateTime::UNIX_EPOCH,
-                decommission: None,
-            }],
-            ..Default::default()
-        };
-        assert!(matches!(
-            classify_heal_format_pool(0, "pool-new", &identity_without_decommission, None),
-            Some(HealFormatPoolSkip::Retryable)
-        ));
-
-        assert!(matches!(
-            classify_heal_format_pool(0, "", &identity_meta, None),
-            Some(HealFormatPoolSkip::Retryable)
-        ));
-
-        assert!(matches!(
-            classify_heal_format_pool(0, "pool-0", &PoolMeta::default(), None),
-            Some(HealFormatPoolSkip::Retryable)
-        ));
-
-        let stopped = RebalanceMeta {
-            stopped_at: Some(OffsetDateTime::UNIX_EPOCH),
-            pool_stats: vec![RebalanceStats {
-                participating: true,
-                info: RebalanceInfo {
-                    status: RebalStatus::Stopped,
-                    ..Default::default()
-                },
-                ..Default::default()
-            }],
-            ..Default::default()
-        };
-        assert!(classify_heal_format_pool(0, "pool-0", &identity_meta, Some(&stopped)).is_none());
-
-        let stopping_after_stop = RebalanceMeta {
-            stopped_at: Some(OffsetDateTime::UNIX_EPOCH),
-            pool_stats: vec![RebalanceStats {
-                participating: true,
-                info: RebalanceInfo {
-                    status: RebalStatus::Started,
-                    stopping: true,
-                    ..Default::default()
-                },
-                ..Default::default()
-            }],
-            ..Default::default()
-        };
-        assert!(matches!(
-            classify_heal_format_pool(0, "pool-0", &identity_meta, Some(&stopping_after_stop)),
-            Some(HealFormatPoolSkip::Retryable)
-        ));
-    }
-
-    #[test]
-    fn skipped_heal_format_pool_is_never_reported_as_success() {
-        assert!(matches!(
-            heal_format_pool_skip_error(HealFormatPoolSkip::Retryable),
-            StorageError::SlowDown
-        ));
-        assert!(matches!(
-            heal_format_pool_skip_error(HealFormatPoolSkip::Completed),
-            StorageError::NoHealRequired
-        ));
-    }
-
     #[tokio::test]
     async fn heal_object_pool_scope_selects_only_requested_pool() {
         let store = minimal_heal_store().await;
@@ -914,18 +615,6 @@ mod tests {
             bucket_fence_registry: std::sync::Arc::default(),
         };
 
-        let err = store
-            .handle_heal_format(false)
-            .await
-            .expect_err("missing pool metadata must fail closed before format writes");
-        assert!(matches!(err, StorageError::SlowDown));
-
-        let pool_meta = PoolMeta::new(&store.pools, &PoolMeta::default());
-        pool_meta
-            .save(store.pools.clone())
-            .await
-            .expect("pool metadata should be persisted before format heal");
-
         let (result, err) = store
             .handle_heal_format(false)
             .await
@@ -939,22 +628,5 @@ mod tests {
             .await
             .expect("the later pool should be healed despite the first pool error");
         assert_eq!(healed.erasure.this, recoverable_format.erasure.sets[0][2]);
-
-        let mut completed_meta = PoolMeta::new(&store.pools, &PoolMeta::default());
-        for status in &mut completed_meta.pools {
-            status.decommission = Some(PoolDecommissionInfo {
-                complete: true,
-                ..Default::default()
-            });
-        }
-        completed_meta
-            .save(store.pools.clone())
-            .await
-            .expect("completed pool metadata should be persisted");
-        let (_, err) = store
-            .handle_heal_format(false)
-            .await
-            .expect("completed pools should be reported as a no-op");
-        assert!(matches!(err, Some(StorageError::NoHealRequired)));
     }
 }

crates/heal/src/heal/task/heal_erasure_set.rs

@@ -231,10 +231,6 @@ impl HealTask {
                             "Heal erasure set format repair skipped because no format heal was required"
                         );
                     } else {
-                        let error = e;
-                        if error.is_recoverable_heal() {
-                            return Err(error);
-                        }
                         error!(
                             target: "rustfs::heal::task",
                             event = EVENT_HEAL_ERASURE_SET_RESULT,
@@ -243,7 +239,7 @@ impl HealTask {
                             task_id = %self.id,
                             set_disk_id,
                             result = "format_failed",
-                            error = %error,
+                            error = %e,
                             "Heal erasure set failed"
                         );
                         {
@@ -251,7 +247,7 @@ impl HealTask {
                             progress.update_progress(4, 4, 0, 0);
                         }
                         return Err(Error::TaskExecutionFailed {
-                            message: format!("Failed to heal disk format for {set_disk_id}: {error}"),
+                            message: format!("Failed to heal disk format for {set_disk_id}: {e}"),
                         });
                     }
                 } else {
@@ -288,9 +284,6 @@ impl HealTask {
             Err(Error::TaskCancelled) => return Err(Error::TaskCancelled),
             Err(Error::TaskTimeout) => return Err(Error::TaskTimeout),
             Err(e) => {
-                if e.is_recoverable_heal() {
-                    return Err(e);
-                }
                 error!(
                     target: "rustfs::heal::task",
                     event = EVENT_HEAL_ERASURE_SET_RESULT,

crates/heal/src/heal/task/tests.rs

@@ -547,7 +547,6 @@ struct MockStorage {
     heal_object_outcome: Mutex<Option<MockHealObjectOutcome>>,
     heal_object_outcomes: Mutex<HashMap<String, VecDeque<MockHealObjectOutcome>>>,
     format_no_heal_required: Mutex<bool>,
-    format_error: Mutex<Option<Error>>,
     global_format_calls: Mutex<u32>,
     replacement_format_calls: Mutex<Vec<(usize, usize, Vec<String>)>>,
     replacement_targets_ready: Mutex<bool>,
@@ -868,9 +867,6 @@ impl HealStorageAPI for MockStorage {
 
     async fn heal_format(&self, _dry_run: bool) -> Result<(HealResultItem, Option<Error>)> {
         *self.global_format_calls.lock().unwrap() += 1;
-        if let Some(error) = self.format_error.lock().unwrap().take() {
-            return Err(error);
-        }
         let no_heal_required = *self.format_no_heal_required.lock().unwrap();
         if no_heal_required {
             Ok((HealResultItem::default(), Some(Error::Storage(EcstoreError::NoHealRequired))))
@@ -2056,30 +2052,6 @@ async fn test_erasure_set_heal_continues_after_format_no_heal_required() {
     );
 }
 
-#[tokio::test]
-async fn erasure_set_format_slowdown_is_propagated() {
-    let storage = Arc::new(MockStorage {
-        format_error: Mutex::new(Some(Error::Storage(EcstoreError::SlowDown))),
-        ..Default::default()
-    });
-    let request = HealRequest::new(
-        HealType::ErasureSet {
-            buckets: Vec::new(),
-            set_disk_id: "pool_0_set_0".to_string(),
-        },
-        HealOptions::default(),
-        HealPriority::Normal,
-    );
-    let task = HealTask::from_request(request, storage);
-
-    let error = task
-        .execute()
-        .await
-        .expect_err("format SlowDown must remain recoverable for the task manager");
-
-    assert!(matches!(error, Error::Storage(EcstoreError::SlowDown)));
-}
-
 #[tokio::test]
 async fn erasure_set_bucket_prepass_failure_stops_before_object_heal() {
     let temp = TempDir::new().expect("temporary directory should be created");

crates/test-utils/src/lib.rs

@@ -245,18 +245,6 @@ impl TestECStoreEnvBuilder {
             .await
             .expect("build test ECStore");
 
-        // The production bootstrap only persists pool.bin from the elected
-        // first cluster node.  Test stores intentionally have no cluster
-        // election, but heal-format still requires that durable fence before
-        // it can write any disk format.  Materialize the validated topology
-        // here so the shared fixture models a ready single-node store.
-        let mut pool_meta = ecstore.pool_meta.read().await.clone();
-        pool_meta.dont_save = false;
-        pool_meta
-            .save(ecstore.pools.clone())
-            .await
-            .expect("persist test pool metadata");
-
         if self.init_bucket_metadata {
             let buckets_list = ecstore
                 .list_bucket(&BucketOptions {