Merge branch 'main' into cxymds/fix-1926-owner-selection

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

@@ -297,10 +297,16 @@ impl ECStore {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::bucket::metadata_sys;
     use crate::core::pools::{PoolDecommissionInfo, PoolStatus};
-    use crate::disk::{DiskOption, format::FormatV3, new_disk};
-    use crate::layout::endpoints::{Endpoints, PoolEndpoints};
+    use crate::disk::{DeleteOptions, DiskOption, format::FormatV3, new_disk};
+    use crate::layout::endpoints::{EndpointServerPools, Endpoints, PoolEndpoints};
+    use crate::runtime::instance::InstanceContext;
+    use crate::storage_api_contracts::bucket::{BucketOperations, MakeBucketOptions};
+    use crate::storage_api_contracts::object::{ObjectIO as _, ObjectOperations};
     use crate::store::init_format::{load_format_erasure, save_format_file};
+    use crate::store::init_local_disks_with_instance_ctx;
+    use tokio_util::sync::CancellationToken;
 
     async fn minimal_heal_pool(pool_idx: usize) -> Arc<Sets> {
         let format = FormatV3::new(1, 1);
@@ -347,6 +353,51 @@ mod tests {
         }
     }
 
+    async fn multi_pool_heal_store() -> (tempfile::TempDir, Arc<ECStore>, CancellationToken) {
+        let temp_dir = tempfile::tempdir().expect("multi-pool heal test directory should be created");
+        let mut pool_endpoints = Vec::new();
+        for pool_index in 0..2 {
+            let mut endpoints = Vec::new();
+            for disk_index in 0..4 {
+                let disk_path = temp_dir.path().join(format!("pool{pool_index}-disk{disk_index}"));
+                tokio::fs::create_dir_all(&disk_path)
+                    .await
+                    .expect("multi-pool heal test disk should be created");
+                let mut endpoint = Endpoint::try_from(disk_path.to_str().expect("disk path should be utf8"))
+                    .expect("test endpoint should parse");
+                endpoint.set_pool_index(pool_index);
+                endpoint.set_set_index(0);
+                endpoint.set_disk_index(disk_index);
+                endpoints.push(endpoint);
+            }
+            pool_endpoints.push(PoolEndpoints {
+                legacy: false,
+                set_count: 1,
+                drives_per_set: 4,
+                endpoints: Endpoints::from(endpoints),
+                cmd_line: format!("heal-owner-pool-{pool_index}"),
+                platform: "test".to_string(),
+            });
+        }
+
+        let endpoint_pools = EndpointServerPools::from(pool_endpoints);
+        let instance_ctx = Arc::new(InstanceContext::new());
+        init_local_disks_with_instance_ctx(&instance_ctx, endpoint_pools.clone())
+            .await
+            .expect("multi-pool local disks should initialize");
+        let shutdown = CancellationToken::new();
+        let store = ECStore::new_with_instance_ctx(
+            "127.0.0.1:0".parse().expect("test address should parse"),
+            endpoint_pools,
+            shutdown.clone(),
+            instance_ctx,
+        )
+        .await
+        .expect("multi-pool test store should initialize");
+        metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await;
+        (temp_dir, store, shutdown)
+    }
+
     #[tokio::test]
     async fn heal_object_pool_scope_selects_only_requested_pool() {
         let store = minimal_heal_store().await;
@@ -506,6 +557,204 @@ mod tests {
         }
     }
 
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn unscoped_heal_object_suspended_owner_semantics() {
+        let (_temp_dir, store, shutdown) = multi_pool_heal_store().await;
+        let bucket = format!("heal-owner-{}", Uuid::new_v4().simple());
+        let active_object = "active-owner";
+        let suspended_only_object = "suspended-only";
+        let duplicate_object = "duplicate-owner";
+        let marker_object = "marker-owner";
+        let quorum_object = "quorum-owner";
+        store
+            .make_bucket(&bucket, &MakeBucketOptions::default())
+            .await
+            .expect("bucket should be created in all pools");
+
+        let mut active_reader = PutObjReader::from_vec(b"active owner".to_vec());
+        store.pools[0]
+            .put_object(&bucket, active_object, &mut active_reader, &ObjectOptions::default())
+            .await
+            .expect("active owner object should be written");
+        let active_disks = store.pools[0].disk_set[0].disks.read().await.clone();
+        let missing_active_disk = active_disks[0].clone().expect("active disk should be online");
+        missing_active_disk
+            .delete(
+                &bucket,
+                active_object,
+                DeleteOptions {
+                    recursive: true,
+                    immediate: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("active owner shard should be removed for repair");
+        assert!(
+            missing_active_disk.read_xl(&bucket, active_object, false).await.is_err(),
+            "the active owner fixture must start with one missing metadata copy"
+        );
+
+        let mut suspended_reader = PutObjReader::from_vec(b"suspended owner".to_vec());
+        store.pools[1]
+            .put_object(&bucket, suspended_only_object, &mut suspended_reader, &ObjectOptions::default())
+            .await
+            .expect("suspended owner object should be written");
+        for (pool_index, mod_time) in [1_i64, 2_i64].into_iter().enumerate() {
+            let mut duplicate_reader = PutObjReader::from_vec(format!("duplicate-pool-{pool_index}").into_bytes());
+            store.pools[pool_index]
+                .put_object(
+                    &bucket,
+                    duplicate_object,
+                    &mut duplicate_reader,
+                    &ObjectOptions {
+                        mod_time: Some(OffsetDateTime::UNIX_EPOCH + time::Duration::seconds(mod_time)),
+                        ..Default::default()
+                    },
+                )
+                .await
+                .expect("duplicate owner object should be written");
+        }
+        let history_version = Uuid::new_v4();
+        let mut history_reader = PutObjReader::from_vec(b"marker history".to_vec());
+        store.pools[0]
+            .put_object(
+                &bucket,
+                marker_object,
+                &mut history_reader,
+                &ObjectOptions {
+                    versioned: true,
+                    version_id: Some(history_version.to_string()),
+                    mod_time: Some(OffsetDateTime::UNIX_EPOCH + time::Duration::seconds(1)),
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("versioned marker history should be written");
+        store.pools[0]
+            .delete_object(
+                &bucket,
+                marker_object,
+                ObjectOptions {
+                    versioned: true,
+                    mod_time: Some(OffsetDateTime::UNIX_EPOCH + time::Duration::seconds(2)),
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("delete marker should be written");
+        let mut quorum_reader = PutObjReader::from_vec(b"quorum boundary".to_vec());
+        store.pools[0]
+            .put_object(&bucket, quorum_object, &mut quorum_reader, &ObjectOptions::default())
+            .await
+            .expect("quorum boundary object should be written");
+        {
+            let mut pool_meta = store.pool_meta.write().await;
+            let mut next = PoolMeta::new(&store.pools, &pool_meta);
+            next.pools[1].decommission = Some(PoolDecommissionInfo {
+                start_time: Some(OffsetDateTime::UNIX_EPOCH),
+                ..Default::default()
+            });
+            *pool_meta = next;
+        }
+
+        let (_, duplicate_owner) = store
+            .get_latest_object_info_with_idx(&bucket, duplicate_object, &ObjectOptions::default())
+            .await
+            .expect("duplicate owner should resolve");
+        assert_eq!(duplicate_owner, 1, "latest duplicate must win when all pools are eligible");
+        let (_, active_duplicate_owner) = store
+            .get_latest_object_info_with_idx(
+                &bucket,
+                duplicate_object,
+                &ObjectOptions {
+                    skip_decommissioned: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("active duplicate owner should resolve");
+        assert_eq!(
+            active_duplicate_owner, 0,
+            "suspended duplicate must be excluded from active owner selection"
+        );
+        let (marker_info, marker_owner) = store
+            .get_latest_object_info_with_idx(
+                &bucket,
+                marker_object,
+                &ObjectOptions {
+                    skip_decommissioned: true,
+                    versioned: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("latest delete marker should resolve");
+        assert_eq!(marker_owner, 0);
+        assert!(marker_info.delete_marker, "latest version must preserve delete-marker semantics");
+
+        let (active_result, active_err) = store
+            .handle_heal_object(&bucket, active_object, "", &HealOpts::default())
+            .await
+            .expect("unscoped active-owner heal should complete");
+        assert_eq!(active_result.object, active_object);
+        assert!(active_err.is_none(), "active owner must be selected even with a suspended pool");
+        assert!(
+            missing_active_disk.read_xl(&bucket, active_object, false).await.is_ok(),
+            "active owner heal must write the missing disk metadata: result={active_result:?}, err={active_err:?}"
+        );
+        assert!(
+            store.pools[1]
+                .get_object_info(&bucket, active_object, &ObjectOptions::default())
+                .await
+                .is_err(),
+            "the suspended pool must not be written for an active-owner object"
+        );
+
+        let (suspended_result, suspended_err) = store
+            .handle_heal_object(&bucket, suspended_only_object, "", &HealOpts::default())
+            .await
+            .expect("unscoped suspended-only heal should return a terminal result");
+        assert!(suspended_result.object.is_empty());
+        assert!(matches!(suspended_err, Some(Error::FileNotFound)));
+        assert!(
+            store.pools[1]
+                .get_object_info(&bucket, suspended_only_object, &ObjectOptions::default())
+                .await
+                .is_ok(),
+            "suspended-only data must remain untouched when unscoped heal reports absent"
+        );
+
+        let (_, explicit_err) = store
+            .handle_heal_object(
+                &bucket,
+                suspended_only_object,
+                "",
+                &HealOpts {
+                    pool: Some(1),
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("explicit suspended-owner heal should return a mapped error");
+        assert!(matches!(explicit_err, Some(Error::SlowDown)));
+
+        let original_quorum_disks = store.pools[0].disk_set[0].disks.read().await.clone();
+        let surviving_quorum_disk = original_quorum_disks[3].clone();
+        *store.pools[0].disk_set[0].disks.write().await = vec![None, None, None, surviving_quorum_disk];
+        let (_, quorum_err) = store
+            .handle_heal_object(&bucket, quorum_object, "", &HealOpts::default())
+            .await
+            .expect("quorum boundary heal should return a mapped result");
+        *store.pools[0].disk_set[0].disks.write().await = original_quorum_disks;
+        assert!(
+            matches!(quorum_err, Some(Error::ErasureReadQuorum)),
+            "quorum-boundary heal must preserve quorum error, got {quorum_err:?}"
+        );
+        shutdown.cancel();
+    }
+
     #[tokio::test]
     async fn handle_heal_format_continues_after_a_pool_error() {
         let canonical_format = FormatV3::new(1, 3);