Merge branch 'main' into cxymds/fix-1925-heal-state-fence

crates/config/src/constants/runtime.rs

@@ -57,6 +57,13 @@ 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,7 +89,6 @@ 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;
@@ -639,6 +638,22 @@ 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}")))
 }
@@ -1468,7 +1483,6 @@ 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,
         })
     }
 }
@@ -1500,7 +1514,6 @@ 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,
         })
     }
 }
@@ -1614,82 +1627,6 @@ 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(());
@@ -1738,7 +1675,7 @@ impl PoolMeta {
         self.load_no_lock(pool).await
     }
 
-    async fn load_no_lock<S>(&mut self, pool: Arc<S>) -> Result<()>
+    pub(crate) async fn load_no_lock<S>(&mut self, pool: Arc<S>) -> Result<()>
     where
         S: EcstoreObjectIO,
     {
@@ -2050,9 +1987,30 @@ impl PoolMeta {
     }
 
     pub fn update_after(&mut self, idx: usize, duration: Duration) -> Result<bool> {
-        Ok(self
-            .decommission_progress_checkpoint(idx, duration, OffsetDateTime::now_utc())?
-            .is_some())
+        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)
     }
 
     pub fn validate(&self, pools: Vec<Arc<Sets>>) -> Result<bool> {
@@ -2193,16 +2151,6 @@ 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 {
@@ -2237,7 +2185,6 @@ 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) {
@@ -2542,40 +2489,6 @@ 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],
@@ -2958,7 +2871,7 @@ impl ECStore {
         Ok(())
     }
 
-    async fn track_decommission_entry_progress_stage(
+    async fn save_decommission_entry_progress_stage(
         &self,
         idx: usize,
         bucket: &str,
@@ -2969,6 +2882,22 @@ 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(())
@@ -3236,7 +3165,7 @@ impl ECStore {
                 let bucket_name = bucket.clone();
                 let object_name = rd.object_info.name.clone();
 
-                self.track_decommission_entry_progress_stage(
+                self.save_decommission_entry_progress_stage(
                     idx,
                     bucket_name.as_str(),
                     object_name.as_str(),
@@ -3330,7 +3259,7 @@ impl ECStore {
             }
             decommission_cancel_signal_result(rx.is_cancelled())?;
 
-            self.track_decommission_entry_progress_stage(
+            self.save_decommission_entry_progress_stage(
                 idx,
                 bucket.as_str(),
                 entry.name.as_str(),
@@ -3338,7 +3267,7 @@ impl ECStore {
             )
             .await?;
 
-            self.track_decommission_entry_progress_stage(
+            self.save_decommission_entry_progress_stage(
                 idx,
                 bucket.as_str(),
                 entry.name.as_str(),
@@ -3405,42 +3334,34 @@ impl ECStore {
             }
         };
 
-        self.track_decommission_entry_progress_stage(
-            idx,
-            bucket.as_str(),
-            entry.name.as_str(),
-            DECOMMISSION_STAGE_ENTRY_FINISHED,
-        )
-        .await?;
+        self.save_decommission_entry_progress_stage(idx, bucket.as_str(), entry.name.as_str(), DECOMMISSION_STAGE_ENTRY_FINISHED)
+            .await?;
 
         if should_save_progress {
-            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"
-                        );
-                    }
+            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}");
                 }
             }
         }
@@ -5343,11 +5264,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, 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_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_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,
@@ -5372,8 +5293,9 @@ 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,
-        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,
+        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,
     };
     use crate::data_movement;
     use crate::disk::endpoint::Endpoint;
@@ -6616,7 +6538,7 @@ mod pools_tests {
     }
 
     #[test]
-    fn test_track_decommission_stage_does_not_advance_checkpoint_state() {
+    fn test_touch_decommission_progress_updates_last_update_and_save_baseline() {
         let mut meta = PoolMeta {
             pools: vec![PoolStatus {
                 id: 0,
@@ -6631,13 +6553,11 @@ mod pools_tests {
             ..Default::default()
         };
 
-        track_decommission_current_object_stage(&mut meta, 0, "bucket", "object", "migrate_object")
-            .expect("valid decommission progress should be tracked");
+        touch_decommission_progress(&mut meta, 0).expect("valid decommission progress should be touched");
 
-        assert_eq!(meta.pools[0].last_update, OffsetDateTime::UNIX_EPOCH);
+        assert!(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(), 5);
-        assert_eq!(info.stage, "migrate_object");
+        assert_eq!(info.items_since_last_progress_save(), 0);
     }
 
     #[test]
@@ -6712,134 +6632,6 @@ 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();
-        tokio::task::spawn_blocking(move || fsync_dir_std(dir)).await?
+        fsync_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();
-    tokio::task::spawn_blocking(move || {
+    fsync_spawn_blocking(move || {
         #[cfg(test)]
         {
             if let Some(kind) = fsync_dir_recorder::take_grouped_failure(&dir) {
@@ -1080,6 +1080,44 @@ 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<()>;
@@ -1217,7 +1255,7 @@ where
     F: FnOnce() -> io::Result<T> + Send + 'static,
 {
     let (disk_permit, global_permit) = acquire_file_sync_permits(disk_permits).await?;
-    let result = tokio::task::spawn_blocking(move || {
+    let result = fsync_spawn_blocking(move || {
         let _disk_permit = disk_permit;
         work()
     })
@@ -2146,7 +2184,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 = tokio::task::spawn_blocking(move || {
+    let result = fsync_spawn_blocking(move || {
         let _lease = lease;
         let _disk_permit = disk_permit;
         operation()

crates/ecstore/src/store/heal.rs

@@ -13,7 +13,12 @@
 // 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";
@@ -30,7 +35,119 @@ 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![
@@ -52,8 +169,24 @@ impl ECStore {
         };
 
         let mut count_no_heal = 0;
+        let mut count_completed = 0;
         let mut first_error = None;
-        for pool in self.pools.iter() {
+        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;
+            }
+
             let (mut result, err) = pool.heal_format(dry_run).await?;
             if let Some(err) = err {
                 match err {
@@ -69,11 +202,18 @@ 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 == self.pools.len() {
+        if count_no_heal + count_completed == self.pools.len() {
             info!(
                 event = EVENT_HEAL_FORMAT_COMPLETED,
                 component = LOG_COMPONENT_ECSTORE,
@@ -300,6 +440,7 @@ 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> {
@@ -347,6 +488,164 @@ 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;
@@ -615,6 +914,18 @@ 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
@@ -628,5 +939,22 @@ 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,6 +231,10 @@ 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,
@@ -239,7 +243,7 @@ impl HealTask {
                             task_id = %self.id,
                             set_disk_id,
                             result = "format_failed",
-                            error = %e,
+                            error = %error,
                             "Heal erasure set failed"
                         );
                         {
@@ -247,7 +251,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}: {e}"),
+                            message: format!("Failed to heal disk format for {set_disk_id}: {error}"),
                         });
                     }
                 } else {
@@ -284,6 +288,9 @@ 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,6 +547,7 @@ 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>,
@@ -867,6 +868,9 @@ 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))))
@@ -2052,6 +2056,30 @@ 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,6 +245,18 @@ 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 {