fix(heal): fail closed on tampered resume checkpoints

.github/workflows/ci.yml

@@ -400,7 +400,7 @@ jobs:
     if: github.event_name != 'pull_request' || github.event.action != 'closed'
     needs: [ quick-checks ]
     runs-on: sm-standard-4
-    timeout-minutes: 90
+    timeout-minutes: 45
     env:
       FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
     steps:
@@ -440,7 +440,7 @@ jobs:
     if: github.event_name != 'pull_request' || github.event.action != 'closed'
     needs: [ quick-checks ]
     runs-on: sm-standard-4
-    timeout-minutes: 90
+    timeout-minutes: 60
     env:
       FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
     steps:
@@ -470,7 +470,7 @@ jobs:
     if: github.event_name != 'pull_request' || github.event.action != 'closed'
     needs: [ quick-checks ]
     runs-on: sm-standard-4
-    timeout-minutes: 90
+    timeout-minutes: 60
     strategy:
       # On a PR, one failing protocol leg is enough to know the PR is not ready,
       # so stop the sibling leg instead of paying another ~40 minutes for it.

Cargo.lock

@@ -9587,6 +9587,7 @@ dependencies = [
  "serde",
  "serde_json",
  "serial_test",
+ "sha2 0.11.0",
  "temp-env",
  "tempfile",
  "thiserror 2.0.20",

crates/e2e_test/src/cluster_concurrency_test.rs

@@ -15,14 +15,12 @@
 use crate::common::RustFSTestClusterEnvironment;
 use aws_sdk_s3::Client;
 use aws_sdk_s3::error::SdkError;
-use aws_sdk_s3::types::{CorsConfiguration, CorsRule};
 use bytes::Bytes;
 use std::sync::Arc;
 use tokio::sync::Barrier;
 use tracing::{info, warn};
 
 const BUCKET: &str = "conditional-put-race-bucket";
-const BUCKET_METADATA_RELOAD_BUCKET: &str = "bucket-metadata-reload-barrier";
 
 async fn cleanup_object(client: &Client, key: &str) {
     if let Err(e) = client.delete_object().bucket(BUCKET).key(key).send().await {
@@ -30,16 +28,6 @@ async fn cleanup_object(client: &Client, key: &str) {
     }
 }
 
-async fn assert_bucket_cors_missing(client: &Client) {
-    let result = client.get_bucket_cors().bucket(BUCKET_METADATA_RELOAD_BUCKET).send().await;
-    match result {
-        Err(SdkError::ServiceError(error)) => {
-            assert_eq!(error.err().meta().code(), Some("NoSuchCORSConfiguration"));
-        }
-        result => panic!("expected the peer to report a missing CORS configuration: {result:?}"),
-    }
-}
-
 async fn conditional_put(
     client: &Client,
     key: &str,
@@ -248,48 +236,3 @@ async fn test_conditional_put_basic_cluster() -> Result<(), Box<dyn std::error::
     cleanup_object(&client, test_key).await;
     Ok(())
 }
-
-#[tokio::test]
-async fn test_bucket_cors_write_is_visible_on_peer_before_response() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
-    crate::common::init_logging();
-
-    let mut cluster = RustFSTestClusterEnvironment::new(2).await?;
-    cluster.start().await?;
-    cluster.create_test_bucket(BUCKET_METADATA_RELOAD_BUCKET).await?;
-
-    let writer = cluster.create_s3_client(0)?;
-    let reader = cluster.create_s3_client(1)?;
-    assert_bucket_cors_missing(&reader).await;
-
-    let rule = CorsRule::builder()
-        .allowed_methods("GET")
-        .allowed_origins("https://example.com")
-        .build()?;
-    let configuration = CorsConfiguration::builder().cors_rules(rule).build()?;
-
-    writer
-        .put_bucket_cors()
-        .bucket(BUCKET_METADATA_RELOAD_BUCKET)
-        .cors_configuration(configuration)
-        .send()
-        .await?;
-
-    let response = reader.get_bucket_cors().bucket(BUCKET_METADATA_RELOAD_BUCKET).send().await?;
-    let rules = response.cors_rules();
-    assert_eq!(
-        rules.len(),
-        1,
-        "peer should observe the committed CORS rule before the write response returns"
-    );
-    assert_eq!(rules[0].allowed_methods(), ["GET"]);
-    assert_eq!(rules[0].allowed_origins(), ["https://example.com"]);
-
-    writer
-        .delete_bucket_cors()
-        .bucket(BUCKET_METADATA_RELOAD_BUCKET)
-        .send()
-        .await?;
-    assert_bucket_cors_missing(&reader).await;
-    writer.delete_bucket().bucket(BUCKET_METADATA_RELOAD_BUCKET).send().await?;
-    Ok(())
-}

crates/ecstore/src/cluster/rpc/peer_rest_client.rs

@@ -85,7 +85,6 @@ const PEER_REST_RECOVERY_MAX_ATTEMPTS: u32 = 60;
 const PEER_REST_RECOVERY_MAX_BACKOFF: Duration = Duration::from_secs(30);
 const SCANNER_ACTIVITY_MAX_MESSAGE_SIZE: usize = 1024;
 const REPLICATION_STATS_MAX_MESSAGE_SIZE: usize = 8 * 1024 * 1024;
-const BUCKET_METADATA_RELOAD_TIMEOUT: Duration = Duration::from_secs(5);
 
 /// Error for a peer that reported `success = false` without an `error_info` payload.
 ///
@@ -1329,38 +1328,27 @@ impl PeerRestClient {
     }
 
     pub async fn load_bucket_metadata(&self, bucket: &str, scanner_maintenance_change: bool) -> Result<()> {
-        let result = tokio::time::timeout(BUCKET_METADATA_RELOAD_TIMEOUT, async {
-            let result = self.load_bucket_metadata_once(bucket, scanner_maintenance_change).await;
-            if let Err(err) = &result
-                && Self::is_network_like_error(err)
-            {
-                self.prepare_retry().await;
-                return self.load_bucket_metadata_once(bucket, scanner_maintenance_change).await;
+        self.finalize_result(
+            async {
+                let mut client = self.get_client().await?;
+                let mut request = Request::new(LoadBucketMetadataRequest {
+                    bucket: bucket.to_string(),
+                    scanner_maintenance_change,
+                });
+                set_tonic_mutation_body_digest(&mut request)?;
+
+                let response = client.load_bucket_metadata(request).await?.into_inner();
+                if !response.success {
+                    if let Some(msg) = response.error_info {
+                        return Err(Error::other(msg));
+                    }
+                    return Err(peer_failure_without_details("load_bucket_metadata", Some(bucket)));
+                }
+                Ok(())
             }
-            result
-        })
+            .await,
+        )
         .await
-        .unwrap_or_else(|_| Err(Error::other(format!("load_bucket_metadata({bucket}) timed out"))));
-        self.finalize_result(result).await
-    }
-
-    async fn load_bucket_metadata_once(&self, bucket: &str, scanner_maintenance_change: bool) -> Result<()> {
-        let mut client = self.get_client().await?;
-        let mut request = Request::new(LoadBucketMetadataRequest {
-            bucket: bucket.to_string(),
-            scanner_maintenance_change,
-        });
-        set_tonic_mutation_body_digest(&mut request)?;
-        request.set_timeout(BUCKET_METADATA_RELOAD_TIMEOUT);
-
-        let response = client.load_bucket_metadata(request).await?.into_inner();
-        if !response.success {
-            if let Some(msg) = response.error_info {
-                return Err(Error::other(msg));
-            }
-            return Err(peer_failure_without_details("load_bucket_metadata", Some(bucket)));
-        }
-        Ok(())
     }
 
     pub async fn delete_bucket_metadata(&self, bucket: &str) -> Result<()> {

crates/ecstore/src/core/pools.rs

@@ -36,8 +36,7 @@ use crate::disk::error::DiskError;
 use crate::disk::{BUCKET_META_PREFIX, RUSTFS_META_BUCKET};
 use crate::error::{Error, Result};
 use crate::error::{
-    StorageError, is_err_bucket_exists, is_err_bucket_not_found, is_err_object_not_found, is_err_operation_canceled,
-    is_err_version_not_found,
+    StorageError, is_err_bucket_exists, is_err_bucket_not_found, is_err_object_not_found, is_err_version_not_found,
 };
 use crate::layout::endpoints::EndpointServerPools;
 use crate::object_api::{GetObjectReader, ObjectOptions};
@@ -90,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;
@@ -640,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}")))
 }
@@ -759,76 +773,7 @@ async fn load_decommission_entry_exact_versions(
 }
 
 fn resolve_decommission_check_after_list_result(list_result: Result<()>, entry_error: Option<Error>) -> Result<()> {
-    match list_result {
-        Ok(()) => entry_error.map_or(Ok(()), Err),
-        Err(list_err) => resolve_decommission_listing_error(Some(list_err), entry_error).map_or(Ok(()), Err),
-    }
-}
-
-fn resolve_decommission_listing_error(listing_error: Option<Error>, entry_error: Option<Error>) -> Option<Error> {
-    match (listing_error, entry_error) {
-        (Some(listing_error), Some(entry_error)) if is_err_operation_canceled(&listing_error) => Some(entry_error),
-        (Some(listing_error), Some(entry_error)) if is_err_operation_canceled(&entry_error) => Some(listing_error),
-        (Some(listing_error), _) => Some(listing_error),
-        (None, entry_error) => entry_error,
-    }
-}
-
-fn decommission_unresolved_listing_error(
-    bucket: &str,
-    prefix: &str,
-    candidate: Option<&str>,
-    candidate_count: usize,
-    disk_error_count: usize,
-    pool_index: usize,
-    set_index: usize,
-) -> Error {
-    let location = candidate.unwrap_or(prefix);
-    Error::other(format!(
-        "decommission listing could not resolve metadata for {bucket}/{location} on pool {pool_index} set {set_index} ({candidate_count} candidate(s), {disk_error_count} disk error(s))"
-    ))
-}
-
-fn resolve_decommission_partial_listing_entry(
-    entries: MetaCacheEntries,
-    resolver: MetadataResolutionParams,
-    bucket: &str,
-    prefix: &str,
-    disk_error_count: usize,
-    pool_index: usize,
-    set_index: usize,
-) -> Result<MetaCacheEntry> {
-    let candidate_count = entries.as_ref().iter().flatten().count();
-    if let Some(entry) = entries.resolve(resolver) {
-        return Ok(entry);
-    }
-
-    let candidate = entries.as_ref().iter().flatten().map(|entry| entry.name.as_str()).next();
-    Err(decommission_unresolved_listing_error(
-        bucket,
-        prefix,
-        candidate,
-        candidate_count,
-        disk_error_count,
-        pool_index,
-        set_index,
-    ))
-}
-
-async fn record_decommission_entry_error(
-    entry_error: &Arc<tokio::sync::Mutex<Option<Error>>>,
-    rx: &CancellationToken,
-    err: Error,
-) {
-    if rx.is_cancelled() {
-        return;
-    }
-
-    let mut first_err = entry_error.lock().await;
-    if first_err.is_none() && !rx.is_cancelled() {
-        *first_err = Some(err);
-        rx.cancel();
-    }
+    if let Some(err) = entry_error { Err(err) } else { list_result }
 }
 
 fn resolve_decommission_pool_meta_reload_result(result: Result<()>, stage: &str) -> Result<()> {
@@ -1538,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,
         })
     }
 }
@@ -1570,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,
         })
     }
 }
@@ -1684,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(());
@@ -2120,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> {
@@ -2263,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 {
@@ -2307,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) {
@@ -2612,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],
@@ -3028,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,
@@ -3039,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(())
@@ -3306,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(),
@@ -3400,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(),
@@ -3408,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(),
@@ -3475,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}");
                 }
             }
         }
@@ -3687,7 +3538,6 @@ impl ECStore {
             let rx_clone = rx.clone();
             let bi = bi.clone();
             let set_id = set_idx;
-            let listing_entry_error = entry_error.clone();
             let worker = tokio::spawn(async move {
                 let _listing_permit = listing_permit;
                 run_decommission_listing_with_retry(
@@ -3701,11 +3551,7 @@ impl ECStore {
                         let set = set.clone();
                         let rx = rx_clone.clone();
                         let bucket = bi.clone();
-                        let entry_error = listing_entry_error.clone();
-                        async move {
-                            set.list_objects_to_decommission(rx, bucket, callback, entry_error.clone(), idx, set_id)
-                                .await
-                        }
+                        async move { set.list_objects_to_decommission(rx, bucket, callback).await }
                     },
                 )
                 .await
@@ -3735,7 +3581,11 @@ impl ECStore {
 
         wait_decommission_worker_drain(&workers, worker_limit).await?;
 
-        if let Some(err) = resolve_decommission_listing_error(listing_worker_error, entry_error.lock().await.clone()) {
+        if let Some(err) = listing_worker_error {
+            return Err(err);
+        }
+
+        if let Some(err) = entry_error.lock().await.clone() {
             return Err(err);
         }
 
@@ -4341,7 +4191,7 @@ impl ECStore {
         let buckets = self.get_buckets_to_decommission().await?;
         let pool = self.pools[idx].clone();
 
-        for (set_index, set) in pool.disk_set.iter().enumerate() {
+        for set in &pool.disk_set {
             for bucket_info in &buckets {
                 let mut lifecycle_config = None;
                 let mut object_lock_config = None;
@@ -4436,7 +4286,7 @@ impl ECStore {
                 });
 
                 let list_result = set
-                    .list_objects_to_decommission(callback_rx, bucket_info.clone(), callback, entry_error.clone(), idx, set_index)
+                    .list_objects_to_decommission(callback_rx, bucket_info.clone(), callback)
                     .await;
                 let entry_error = entry_error.lock().await.clone();
                 resolve_decommission_check_after_list_result(list_result, entry_error)?;
@@ -5171,15 +5021,12 @@ mod tests {
 pub type ListCallback = Arc<dyn Fn(MetaCacheEntry) -> BoxFuture<'static, ()> + Send + Sync + 'static>;
 
 impl SetDisks {
-    #[tracing::instrument(skip(self, rx, cb_func, entry_error))]
+    #[tracing::instrument(skip(self, rx, cb_func))]
     async fn list_objects_to_decommission(
         self: &Arc<Self>,
         rx: CancellationToken,
         bucket_info: DecomBucketInfo,
         cb_func: ListCallback,
-        entry_error: Arc<tokio::sync::Mutex<Option<Error>>>,
-        pool_index: usize,
-        set_index: usize,
     ) -> Result<()> {
         let (disks, _) = self.get_online_disks_with_healing(false).await;
         ensure_decommission_listing_disks_available(!disks.is_empty(), &bucket_info.name)?;
@@ -5194,12 +5041,6 @@ impl SetDisks {
         };
 
         let cb1 = cb_func.clone();
-        let unresolved_error = entry_error.clone();
-        let unresolved_rx = rx.clone();
-        let unresolved_bucket = bucket_info.name.clone();
-        let unresolved_prefix = bucket_info.prefix.clone();
-        let unresolved_pool_index = pool_index;
-        let unresolved_set_index = set_index;
 
         list_path_raw(
             rx,
@@ -5212,51 +5053,20 @@ impl SetDisks {
                 skip_walkdir_total_timeout: true,
                 walkdir_stall_timeout: Some(DECOMMISSION_BACKGROUND_WALKDIR_STALL_TIMEOUT),
                 agreed: Some(Box::new(move |entry: MetaCacheEntry| Box::pin(cb1(entry)))),
-                partial: Some(Box::new(move |entries: MetaCacheEntries, errs: &[Option<DiskError>]| {
+                partial: Some(Box::new(move |entries: MetaCacheEntries, _: &[Option<DiskError>]| {
                     let resolver = resolver.clone();
                     let cb_func = cb_func.clone();
-                    let bucket = unresolved_bucket.clone();
-                    let prefix = unresolved_prefix.clone();
-                    let unresolved_error = unresolved_error.clone();
-                    let unresolved_rx = unresolved_rx.clone();
-                    let pool_index = unresolved_pool_index;
-                    let set_index = unresolved_set_index;
-                    let disk_error_count = errs.iter().flatten().count();
-                    if unresolved_rx.is_cancelled() {
-                        return Box::pin(async {});
-                    }
-
-                    match resolve_decommission_partial_listing_entry(
-                        entries,
-                        resolver,
-                        &bucket,
-                        &prefix,
-                        disk_error_count,
-                        pool_index,
-                        set_index,
-                    ) {
-                        Ok(entry) => {
+                    match entries.resolve(resolver) {
+                        Some(entry) => {
                             warn!("decommission_pool: list_objects_to_decommission get {}", &entry.name);
                             Box::pin(async move {
                                 cb_func(entry).await;
                             })
                         }
-                        Err(err) => Box::pin(async move {
-                            if unresolved_rx.is_cancelled() {
-                                return;
-                            }
-                            warn!(
-                                event = EVENT_DECOMMISSION_BUCKET,
-                                component = LOG_COMPONENT_ECSTORE,
-                                subsystem = LOG_SUBSYSTEM_POOLS,
-                                bucket = %bucket,
-                                prefix = %prefix,
-                                state = "unresolved_entry",
-                                error = %err,
-                                "Decommission listing failed closed on unresolved metadata"
-                            );
-                            record_decommission_entry_error(&unresolved_error, &unresolved_rx, err).await;
-                        }),
+                        None => {
+                            warn!("decommission_pool: list_objects_to_decommission get none");
+                            Box::pin(async {})
+                        }
                     }
                 })),
                 ..Default::default()
@@ -5264,10 +5074,6 @@ impl SetDisks {
         )
         .await?;
 
-        if let Some(err) = entry_error.lock().await.clone() {
-            return Err(err);
-        }
-
         Ok(())
     }
 }
@@ -5458,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,
@@ -5473,12 +5279,11 @@ mod pools_tests {
         has_active_decommission_canceler, is_decommission_active, is_decommission_cancel_requested,
         load_decommission_entry_versions, local_decommission_queue_prefix, mark_decommission_bucket_done,
         merge_pool_status_refresh, missing_decommission_worker_prefix, observe_decommission_terminal_reload_result,
-        pool_meta_has_active_decommission, record_decommission_entry_error, require_decommission_store,
-        resolve_decommission_bucket_done_save_result, resolve_decommission_bucket_state,
-        resolve_decommission_check_after_list_result, resolve_decommission_entry_cleanup_delete_result,
-        resolve_decommission_entry_exact_versions, resolve_decommission_entry_reload_result, resolve_decommission_listing_error,
-        resolve_decommission_listing_worker_result, resolve_decommission_optional_bucket_config_result,
-        resolve_decommission_partial_listing_entry, resolve_decommission_pool_meta_reload_result,
+        pool_meta_has_active_decommission, require_decommission_store, resolve_decommission_bucket_done_save_result,
+        resolve_decommission_bucket_state, resolve_decommission_check_after_list_result,
+        resolve_decommission_entry_cleanup_delete_result, resolve_decommission_entry_exact_versions,
+        resolve_decommission_entry_reload_result, resolve_decommission_listing_worker_result,
+        resolve_decommission_optional_bucket_config_result, resolve_decommission_pool_meta_reload_result,
         resolve_decommission_preflight_heal_result, resolve_decommission_progress_save_result,
         resolve_decommission_spawn_failure_result, resolve_decommission_terminal_mark_after_error_result,
         resolve_decommission_terminal_mark_result, resolve_decommission_update_after_result,
@@ -5488,17 +5293,16 @@ 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;
     use crate::error::{Error, StorageError};
     use crate::layout::endpoints::{EndpointServerPools, Endpoints, PoolEndpoints};
     use crate::services::rebalance::{RebalStatus, RebalanceInfo, RebalanceMeta, RebalanceStats};
-    use rustfs_filemeta::{
-        FileInfo, FileInfoVersions, MetaCacheEntries, MetaCacheEntry, MetadataResolutionParams, ObjectPartInfo,
-    };
+    use rustfs_filemeta::{FileInfo, FileInfoVersions, MetaCacheEntry, ObjectPartInfo};
     use rustfs_rio::Index;
     use std::sync::{
         Arc,
@@ -6517,65 +6321,6 @@ mod pools_tests {
         assert!(matches!(err, Error::SlowDown));
     }
 
-    #[test]
-    fn test_resolve_decommission_partial_listing_entry_rejects_unresolved_metadata() {
-        let err = resolve_decommission_partial_listing_entry(
-            MetaCacheEntries(vec![None]),
-            MetadataResolutionParams {
-                dir_quorum: 2,
-                obj_quorum: 2,
-                bucket: "bucket-a".to_string(),
-                ..Default::default()
-            },
-            "bucket-a",
-            "prefix/",
-            1,
-            2,
-            3,
-        )
-        .expect_err("unresolved partial listing must fail closed");
-
-        let message = err.to_string();
-        assert!(message.contains("decommission listing could not resolve metadata"));
-        assert!(message.contains("bucket-a/prefix/"));
-        assert!(message.contains("pool 2 set 3"));
-        assert!(message.contains("1 disk error(s)"));
-    }
-
-    #[tokio::test]
-    async fn test_record_decommission_entry_error_cancels_listing_and_preserves_first_error() {
-        let entry_error = Arc::new(tokio::sync::Mutex::new(None));
-        let rx = CancellationToken::new();
-
-        record_decommission_entry_error(&entry_error, &rx, Error::SlowDown).await;
-        record_decommission_entry_error(&entry_error, &rx, Error::OperationCanceled).await;
-
-        assert!(rx.is_cancelled());
-        assert!(matches!(*entry_error.lock().await, Some(Error::SlowDown)));
-    }
-
-    #[tokio::test]
-    async fn test_record_decommission_entry_error_ignores_already_canceled_listing() {
-        let entry_error = Arc::new(tokio::sync::Mutex::new(None));
-        let rx = CancellationToken::new();
-        rx.cancel();
-
-        record_decommission_entry_error(&entry_error, &rx, Error::SlowDown).await;
-
-        assert!(entry_error.lock().await.is_none());
-    }
-
-    #[test]
-    fn test_resolve_decommission_listing_error_preserves_real_listing_failure() {
-        let err = resolve_decommission_listing_error(Some(Error::SlowDown), Some(Error::OperationCanceled))
-            .expect("listing failure should be returned");
-        assert!(matches!(err, Error::SlowDown));
-
-        let err = resolve_decommission_listing_error(Some(Error::OperationCanceled), Some(Error::SlowDown))
-            .expect("entry failure should be returned");
-        assert!(matches!(err, Error::SlowDown));
-    }
-
     #[test]
     fn test_resolve_decommission_check_after_list_result_returns_list_result_without_entry_error() {
         let err = resolve_decommission_check_after_list_result(Err(Error::OperationCanceled), None)
@@ -6793,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,
@@ -6808,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]
@@ -6889,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/heal/Cargo.toml

@@ -91,6 +91,7 @@ metrics = { workspace = true }
 base64 = { workspace = true }
 bytes = { workspace = true }
 crc-fast = { workspace = true }
+sha2 = { workspace = true }
 
 [dev-dependencies]
 serde_json = { workspace = true, features = ["raw_value"] }

crates/heal/src/heal/erasure_healer.rs

@@ -373,6 +373,11 @@ impl ErasureSetHealer {
         set_disk_id: &str,
         buckets: &[String],
     ) -> Result<(ResumeManager, CheckpointManager)> {
+        if self.replacement_task_id.is_none() && CheckpointManager::is_blocked(&self.disk, task_id).await {
+            return Err(Error::TaskExecutionFailed {
+                message: format!("Resume task {task_id} has a blocked checkpoint"),
+            });
+        }
         // check if resume state exists
         let has_resume_state = if self.replacement_task_id.is_some() {
             ResumeManager::has_replacement_intent(&self.disk, task_id).await

crates/heal/src/heal/resume.rs

@@ -51,6 +51,7 @@ const RESUME_STATE_FILE: &str = "ahm_resume_state.json";
 const REPLACEMENT_INTENT_FILE: &str = "ahm_replacement_intent.json";
 const RESUME_PROGRESS_FILE: &str = "ahm_progress.json";
 pub(super) const RESUME_CHECKPOINT_FILE: &str = "ahm_checkpoint.json";
+pub(super) const RESUME_CHECKPOINT_BLOCKED_FILE: &str = "ahm_checkpoint.blocked";
 const REPLACEMENT_COMPLETION_PROOF_FILE: &str = "ahm_replacement_completion_proof.json";
 const REPLACEMENT_RECOVERY_DIR: &str = "ahm-replacement";
 const REPLACEMENT_INTENT_SEAL_FILE: &str = "ahm_replacement_intent_seal";

crates/heal/src/heal/resume/checkpoint.rs

@@ -13,21 +13,25 @@
 // limitations under the License.
 
 use crate::{Error, Result};
+use base64::Engine as _;
 use serde::{Deserialize, Serialize};
+use sha2::{Digest, Sha256};
 use std::collections::HashSet;
 use std::path::Path;
 use std::sync::{Arc, Mutex};
 use std::time::{SystemTime, UNIX_EPOCH};
-use tokio::sync::RwLock;
+use tokio::sync::{Mutex as AsyncMutex, RwLock};
 use tracing::{debug, warn};
 
-use super::super::{BUCKET_META_PREFIX, DiskStore, HealDiskExt as _, RUSTFS_META_BUCKET};
+use super::super::storage_api::owner::{EcstoreConditionalFileUpdate, EcstoreDiskAPI, EcstoreDiskBytes};
+use super::super::{BUCKET_META_PREFIX, DiskStore, HealDiskExt, RUSTFS_META_BUCKET};
 use super::{
-    LOG_COMPONENT_HEAL, LOG_SUBSYSTEM_RESUME, PersistThrottle, RESUME_CHECKPOINT_FILE, delete_resume_file, path_to_str,
-    validate_resume_task_id,
+    LOG_COMPONENT_HEAL, LOG_SUBSYSTEM_RESUME, PersistThrottle, RESUME_CHECKPOINT_BLOCKED_FILE, RESUME_CHECKPOINT_FILE,
+    delete_resume_file, path_to_str, validate_resume_task_id,
 };
 
 const EVENT_HEAL_CHECKPOINT_STATE: &str = "heal_checkpoint_state";
+const RESUME_CHECKPOINT_DIGEST_FILE: &str = "ahm_checkpoint.sha256";
 
 /// Current on-disk schema version for `ResumeCheckpoint`. Same rationale as
 /// `CURRENT_RESUME_SCHEMA`: pre-per-version dedup identities are not comparable
@@ -116,17 +120,111 @@ pub struct CheckpointManager {
     disk: DiskStore,
     checkpoint: Arc<RwLock<ResumeCheckpoint>>,
     throttle: Mutex<PersistThrottle>,
+    save_lock: AsyncMutex<()>,
+    last_saved: Mutex<Option<EcstoreDiskBytes>>,
 }
 
 impl CheckpointManager {
+    fn blocked_path(task_id: &str) -> std::path::PathBuf {
+        Path::new(BUCKET_META_PREFIX).join(format!("{task_id}_{RESUME_CHECKPOINT_BLOCKED_FILE}"))
+    }
+
+    /// Return whether a checkpoint was permanently isolated after a malformed
+    /// or unsupported snapshot was observed.
+    pub(crate) async fn is_blocked(disk: &DiskStore, task_id: &str) -> bool {
+        if validate_resume_task_id(task_id).is_err() {
+            return false;
+        }
+        let blocked_path = Self::blocked_path(task_id);
+        let Ok(path) = path_to_str(&blocked_path) else {
+            return false;
+        };
+        match HealDiskExt::read_all(disk.as_ref(), RUSTFS_META_BUCKET, path).await {
+            Ok(_) => true,
+            Err(crate::heal::DiskError::FileNotFound) => false,
+            Err(_) => true,
+        }
+    }
+
+    /// Validate the checkpoint while enumerating resumable state. This reads
+    /// the checkpoint once and also isolates malformed or unsupported data.
+    pub(crate) async fn is_resumable(disk: &DiskStore, task_id: &str) -> Result<bool> {
+        validate_resume_task_id(task_id)?;
+        if Self::is_blocked(disk, task_id).await {
+            return Err(Error::InvalidCheckpoint(format!("Resume task {task_id} has a blocked checkpoint")));
+        }
+        let file_path = Path::new(BUCKET_META_PREFIX).join(format!("{task_id}_{RESUME_CHECKPOINT_FILE}"));
+        let Ok(path) = path_to_str(&file_path) else {
+            return Err(Error::InvalidCheckpoint("Resume checkpoint path is not valid UTF-8".to_string()));
+        };
+        match HealDiskExt::read_all(disk.as_ref(), RUSTFS_META_BUCKET, path).await {
+            Ok(bytes) if bytes.is_empty() => Ok(true),
+            Ok(bytes) => Self::load_from_data(disk.clone(), task_id, bytes.to_vec())
+                .await
+                .map(|_| true),
+            Err(crate::heal::DiskError::FileNotFound) => Ok(true),
+            Err(error) => Err(error.into()),
+        }
+    }
+
+    async fn block_invalid_snapshot(disk: &DiskStore, task_id: &str) {
+        // This marker is intentionally version-agnostic: an unsupported reader
+        // must stop selector retries until an operator cleans up the snapshot.
+        let blocked_path = Self::blocked_path(task_id);
+        let Ok(path) = path_to_str(&blocked_path) else {
+            return;
+        };
+        let result = EcstoreDiskAPI::compare_and_update_file(
+            disk.as_ref(),
+            RUSTFS_META_BUCKET,
+            path,
+            None,
+            Some(EcstoreDiskBytes::from_static(b"blocked")),
+        )
+        .await;
+        match result {
+            Ok(EcstoreConditionalFileUpdate::Updated | EcstoreConditionalFileUpdate::Mismatch) => {}
+            Ok(EcstoreConditionalFileUpdate::Missing) => warn!(
+                target: "rustfs::heal::resume",
+                event = EVENT_HEAL_CHECKPOINT_STATE,
+                component = LOG_COMPONENT_HEAL,
+                subsystem = LOG_SUBSYSTEM_RESUME,
+                task_id,
+                state = "blocked_marker_write_failed",
+                error = "marker target disappeared",
+                "Heal checkpoint could not persist its blocked marker"
+            ),
+            Err(error) => warn!(
+                target: "rustfs::heal::resume",
+                event = EVENT_HEAL_CHECKPOINT_STATE,
+                component = LOG_COMPONENT_HEAL,
+                subsystem = LOG_SUBSYSTEM_RESUME,
+                task_id,
+                state = "blocked_marker_write_failed",
+                error = %error,
+                "Heal checkpoint could not persist its blocked marker"
+            ),
+        }
+    }
+
     /// create new checkpoint manager
     pub async fn new(disk: DiskStore, task_id: String) -> Result<Self> {
         validate_resume_task_id(&task_id)?;
+        let checkpoint_volume = format!("{RUSTFS_META_BUCKET}/{BUCKET_META_PREFIX}");
+        if let Err(error) = EcstoreDiskAPI::make_volume(disk.as_ref(), &checkpoint_volume).await
+            && error != crate::heal::DiskError::VolumeExists
+        {
+            return Err(Error::TaskExecutionFailed {
+                message: format!("Failed to create checkpoint volume: {error}"),
+            });
+        }
         let checkpoint = ResumeCheckpoint::new(task_id);
         let manager = Self {
             disk,
             checkpoint: Arc::new(RwLock::new(checkpoint)),
             throttle: Mutex::new(PersistThrottle::new()),
+            save_lock: AsyncMutex::new(()),
+            last_saved: Mutex::new(None),
         };
 
         // save initial checkpoint
@@ -140,6 +238,7 @@ impl CheckpointManager {
                 error = %e,
                 "Heal checkpoint persistence failed"
             );
+            return Err(e);
         }
         Ok(manager)
     }
@@ -148,11 +247,22 @@ impl CheckpointManager {
     pub async fn load_from_disk(disk: DiskStore, task_id: &str) -> Result<Self> {
         validate_resume_task_id(task_id)?;
         let checkpoint_data = Self::read_checkpoint_file(&disk, task_id).await?;
-        let mut checkpoint: ResumeCheckpoint =
-            serde_json::from_slice(&checkpoint_data).map_err(|e| Error::TaskExecutionFailed {
-                message: format!("Failed to deserialize checkpoint: {e}"),
-            })?;
+        Self::load_from_data(disk, task_id, checkpoint_data).await
+    }
+
+    async fn load_from_data(disk: DiskStore, task_id: &str, checkpoint_data: Vec<u8>) -> Result<Self> {
+        validate_resume_task_id(task_id)?;
+        let mut checkpoint: ResumeCheckpoint = match serde_json::from_slice(&checkpoint_data) {
+            Ok(checkpoint) => checkpoint,
+            Err(error) => {
+                Self::block_invalid_snapshot(&disk, task_id).await;
+                return Err(Error::TaskExecutionFailed {
+                    message: format!("Failed to deserialize checkpoint: {error}"),
+                });
+            }
+        };
         if checkpoint.task_id != task_id {
+            Self::block_invalid_snapshot(&disk, task_id).await;
             return Err(Error::TaskExecutionFailed {
                 message: "Resume checkpoint task id does not match filename".to_string(),
             });
@@ -163,6 +273,7 @@ impl CheckpointManager {
         // identities. Discard the stale sets and position, then stamp the
         // current schema so the scan restarts cleanly.
         if checkpoint.schema_version > CURRENT_CHECKPOINT_SCHEMA {
+            Self::block_invalid_snapshot(&disk, task_id).await;
             return Err(Error::TaskExecutionFailed {
                 message: format!(
                     "Checkpoint schema {} is newer than supported schema {CURRENT_CHECKPOINT_SCHEMA}",
@@ -194,6 +305,8 @@ impl CheckpointManager {
             disk,
             checkpoint: Arc::new(RwLock::new(checkpoint)),
             throttle: Mutex::new(PersistThrottle::new()),
+            save_lock: AsyncMutex::new(()),
+            last_saved: Mutex::new(Some(EcstoreDiskBytes::from(checkpoint_data))),
         })
     }
 
@@ -204,7 +317,7 @@ impl CheckpointManager {
         }
         let file_path = Path::new(BUCKET_META_PREFIX).join(format!("{task_id}_{RESUME_CHECKPOINT_FILE}"));
         match path_to_str(&file_path) {
-            Ok(path_str) => match disk.read_all(RUSTFS_META_BUCKET, path_str).await {
+            Ok(path_str) => match HealDiskExt::read_all(disk.as_ref(), RUSTFS_META_BUCKET, path_str).await {
                 Ok(data) => !data.is_empty(),
                 Err(_) => false,
             },
@@ -292,6 +405,8 @@ impl CheckpointManager {
 
         let checkpoint_file = Path::new(BUCKET_META_PREFIX).join(format!("{task_id}_{RESUME_CHECKPOINT_FILE}"));
         delete_resume_file(&self.disk, &checkpoint_file).await?;
+        delete_resume_file(&self.disk, &Self::digest_path(&task_id)).await?;
+        delete_resume_file(&self.disk, &Self::blocked_path(&task_id)).await?;
 
         debug!(
             target: "rustfs::heal::resume",
@@ -307,21 +422,139 @@ impl CheckpointManager {
 
     /// save checkpoint to disk
     async fn save_checkpoint(&self) -> Result<()> {
-        let checkpoint = self.checkpoint.read().await;
+        // Serialize saves and take the snapshot only after acquiring the lock:
+        // a slower writer must not publish a snapshot taken before a newer one.
+        let _save_guard = self.save_lock.lock().await;
+        let checkpoint = self.checkpoint.read().await.clone();
         validate_resume_task_id(&checkpoint.task_id)?;
-        let checkpoint_data = serde_json::to_vec(&*checkpoint).map_err(|e| Error::TaskExecutionFailed {
-            message: format!("Failed to serialize checkpoint: {e}"),
-        })?;
+        let checkpoint_data =
+            EcstoreDiskBytes::from(serde_json::to_vec(&checkpoint).map_err(|e| Error::TaskExecutionFailed {
+                message: format!("Failed to serialize checkpoint: {e}"),
+            })?);
 
         let file_path = Path::new(BUCKET_META_PREFIX).join(format!("{}_{}", checkpoint.task_id, RESUME_CHECKPOINT_FILE));
 
         let path_str = path_to_str(&file_path)?;
-        self.disk
-            .write_all(RUSTFS_META_BUCKET, path_str, checkpoint_data.into())
+        let last_saved = self
+            .last_saved
+            .lock()
+            .map_err(|_| Error::TaskExecutionFailed {
+                message: "Checkpoint save state lock is poisoned; refusing to save".to_string(),
+            })?
+            .clone();
+        let update = EcstoreDiskAPI::compare_and_update_file(
+            self.disk.as_ref(),
+            RUSTFS_META_BUCKET,
+            path_str,
+            last_saved.clone(),
+            Some(checkpoint_data.clone()),
+        )
+        .await
+        .map_err(|e| Error::TaskExecutionFailed {
+            message: format!("Failed to save checkpoint: {e}"),
+        })?;
+
+        let expected = match update {
+            EcstoreConditionalFileUpdate::Updated => None,
+            EcstoreConditionalFileUpdate::Missing => {
+                return Err(Error::TaskExecutionFailed {
+                    message: "Checkpoint was removed after this manager saved it; refusing to recreate it".to_string(),
+                });
+            }
+            EcstoreConditionalFileUpdate::Mismatch => {
+                // A healthy manager normally completes the CAS above without
+                // another read or JSON parse. Inspect only after a mismatch so
+                // corruption and future schemas cannot be overwritten blindly.
+                let existing = match HealDiskExt::read_all(self.disk.as_ref(), RUSTFS_META_BUCKET, path_str).await {
+                    Ok(existing) => existing,
+                    Err(crate::heal::DiskError::FileNotFound) => {
+                        return Err(Error::TaskExecutionFailed {
+                            message: "Checkpoint was removed after this manager saved it; refusing to recreate it".to_string(),
+                        });
+                    }
+                    Err(error) => {
+                        return Err(Error::TaskExecutionFailed {
+                            message: format!("Failed to inspect checkpoint after CAS mismatch: {error}"),
+                        });
+                    }
+                };
+
+                if existing.is_empty() && last_saved.is_none() {
+                    Some(existing)
+                } else {
+                    let current: ResumeCheckpoint = match serde_json::from_slice(&existing) {
+                        Ok(current) => current,
+                        Err(error) => {
+                            Self::block_invalid_snapshot(&self.disk, &checkpoint.task_id).await;
+                            return Err(Error::TaskExecutionFailed {
+                                message: format!("Existing checkpoint is corrupt: {error}"),
+                            });
+                        }
+                    };
+                    if current.task_id != checkpoint.task_id {
+                        Self::block_invalid_snapshot(&self.disk, &checkpoint.task_id).await;
+                        return Err(Error::TaskExecutionFailed {
+                            message: "Existing checkpoint task id does not match filename".to_string(),
+                        });
+                    }
+                    if current.schema_version > CURRENT_CHECKPOINT_SCHEMA {
+                        Self::block_invalid_snapshot(&self.disk, &checkpoint.task_id).await;
+                        return Err(Error::TaskExecutionFailed {
+                            message: format!(
+                                "Existing checkpoint schema {} is newer than supported schema {CURRENT_CHECKPOINT_SCHEMA}",
+                                current.schema_version
+                            ),
+                        });
+                    }
+                    if last_saved.as_ref().is_none_or(|saved| saved.as_ref() != existing.as_ref()) {
+                        return Err(Error::TaskExecutionFailed {
+                            message: "Checkpoint changed since this manager loaded it; refusing to overwrite newer progress"
+                                .to_string(),
+                        });
+                    }
+                    Some(existing)
+                }
+            }
+        };
+
+        if let Some(expected) = expected {
+            match EcstoreDiskAPI::compare_and_update_file(
+                self.disk.as_ref(),
+                RUSTFS_META_BUCKET,
+                path_str,
+                Some(expected),
+                Some(checkpoint_data.clone()),
+            )
             .await
             .map_err(|e| Error::TaskExecutionFailed {
-                message: format!("Failed to save checkpoint: {e}"),
-            })?;
+                message: format!("Failed to save checkpoint after CAS mismatch: {e}"),
+            })? {
+                EcstoreConditionalFileUpdate::Updated => {}
+                EcstoreConditionalFileUpdate::Missing | EcstoreConditionalFileUpdate::Mismatch => {
+                    return Err(Error::TaskExecutionFailed {
+                        message: "Checkpoint changed while saving; refusing to overwrite newer progress".to_string(),
+                    });
+                }
+            }
+        }
+
+        let digest_path = Self::digest_path(&checkpoint.task_id);
+        let digest = base64::engine::general_purpose::STANDARD.encode(Sha256::digest(checkpoint_data.as_ref()));
+        HealDiskExt::write_all(
+            self.disk.as_ref(),
+            RUSTFS_META_BUCKET,
+            path_to_str(&digest_path)?,
+            EcstoreDiskBytes::from(digest.into_bytes()),
+        )
+        .await
+        .map_err(|e| Error::TaskExecutionFailed {
+            message: format!("Failed to save checkpoint digest: {e}"),
+        })?;
+
+        let mut last_saved = self.last_saved.lock().map_err(|_| Error::TaskExecutionFailed {
+            message: "Checkpoint save state lock is poisoned after save".to_string(),
+        })?;
+        *last_saved = Some(checkpoint_data);
 
         debug!(
             target: "rustfs::heal::resume",
@@ -341,11 +574,35 @@ impl CheckpointManager {
         let file_path = Path::new(BUCKET_META_PREFIX).join(format!("{task_id}_{RESUME_CHECKPOINT_FILE}"));
 
         let path_str = path_to_str(&file_path)?;
-        disk.read_all(RUSTFS_META_BUCKET, path_str)
+        let checkpoint = HealDiskExt::read_all(disk.as_ref(), RUSTFS_META_BUCKET, path_str)
             .await
             .map(|bytes| bytes.to_vec())
             .map_err(|e| Error::TaskExecutionFailed {
                 message: format!("Failed to read checkpoint file: {e}"),
-            })
+            })?;
+        let digest_path = Self::digest_path(task_id);
+        let digest_path = path_to_str(&digest_path)?;
+        match HealDiskExt::read_all(disk.as_ref(), RUSTFS_META_BUCKET, digest_path).await {
+            Ok(expected) => {
+                let actual = base64::engine::general_purpose::STANDARD.encode(Sha256::digest(&checkpoint));
+                if expected.as_ref() != actual.as_bytes() {
+                    Self::block_invalid_snapshot(disk, task_id).await;
+                    return Err(Error::InvalidCheckpoint(format!(
+                        "Resume checkpoint digest does not match task {task_id}"
+                    )));
+                }
+            }
+            Err(crate::heal::DiskError::FileNotFound) => {}
+            Err(error) => {
+                return Err(Error::TaskExecutionFailed {
+                    message: format!("Failed to read checkpoint digest: {error}"),
+                });
+            }
+        }
+        Ok(checkpoint)
+    }
+
+    fn digest_path(task_id: &str) -> std::path::PathBuf {
+        Path::new(BUCKET_META_PREFIX).join(format!("{task_id}_{RESUME_CHECKPOINT_DIGEST_FILE}"))
     }
 }

crates/heal/src/heal/resume/tests.rs

@@ -1675,6 +1675,285 @@ async fn future_resume_and_checkpoint_schemas_are_rejected() {
     temp_dir.close().expect("remove schema test directory");
 }
 
+#[tokio::test]
+async fn checkpoint_save_does_not_replace_a_non_empty_truncated_snapshot() {
+    let (temp_dir, disk) = schema_test_disk().await;
+    let task_id = ResumeUtils::generate_task_id();
+    let manager = CheckpointManager::new(disk.clone(), task_id.clone())
+        .await
+        .expect("create checkpoint manager");
+    let checkpoint_path = format!("{BUCKET_META_PREFIX}/{task_id}_{RESUME_CHECKPOINT_FILE}");
+    let truncated = b"{\"schema_version\":5,\"task_id\":";
+    disk.write_all(RUSTFS_META_BUCKET, &checkpoint_path, truncated.as_slice().into())
+        .await
+        .expect("write truncated checkpoint fixture");
+
+    let error = manager
+        .update_position(2, 7)
+        .await
+        .expect_err("a truncated checkpoint must fail closed during save");
+    assert!(error.to_string().contains("Existing checkpoint is corrupt"));
+    assert_eq!(
+        disk.read_all(RUSTFS_META_BUCKET, &checkpoint_path)
+            .await
+            .expect("read truncated checkpoint fixture"),
+        truncated.as_slice()
+    );
+    assert!(CheckpointManager::is_blocked(&disk, &task_id).await);
+    temp_dir.close().expect("remove checkpoint save test directory");
+}
+
+#[tokio::test]
+async fn checkpoint_save_does_not_replace_a_future_schema_snapshot() {
+    let (temp_dir, disk) = schema_test_disk().await;
+    let task_id = ResumeUtils::generate_task_id();
+    let manager = CheckpointManager::new(disk.clone(), task_id.clone())
+        .await
+        .expect("create checkpoint manager");
+    let checkpoint_path = format!("{BUCKET_META_PREFIX}/{task_id}_{RESUME_CHECKPOINT_FILE}");
+    let mut future = ResumeCheckpoint::new(task_id.clone());
+    future.schema_version = CURRENT_CHECKPOINT_SCHEMA + 1;
+    let future_bytes = serde_json::to_vec(&future).expect("serialize future checkpoint fixture");
+    disk.write_all(RUSTFS_META_BUCKET, &checkpoint_path, future_bytes.clone().into())
+        .await
+        .expect("write future checkpoint fixture");
+
+    let error = manager
+        .update_position(2, 7)
+        .await
+        .expect_err("a future schema must fail closed during save");
+    assert!(error.to_string().contains("Existing checkpoint schema"));
+    assert_eq!(
+        disk.read_all(RUSTFS_META_BUCKET, &checkpoint_path)
+            .await
+            .expect("read future checkpoint fixture"),
+        future_bytes
+    );
+    assert!(CheckpointManager::is_blocked(&disk, &task_id).await);
+    temp_dir.close().expect("remove future schema test directory");
+}
+
+#[tokio::test]
+async fn checkpoint_digest_rejects_same_length_progress_tampering() {
+    let (temp_dir, disk) = schema_test_disk().await;
+    let task_id = ResumeUtils::generate_task_id();
+    let manager = CheckpointManager::new(disk.clone(), task_id.clone())
+        .await
+        .expect("create checkpoint manager");
+    manager
+        .add_processed_object("victim-a".to_string())
+        .await
+        .expect("persist checkpoint progress");
+    manager.update_position(1, 1).await.expect("flush checkpoint progress");
+    let checkpoint_path = format!("{BUCKET_META_PREFIX}/{task_id}_{RESUME_CHECKPOINT_FILE}");
+    let original = disk
+        .read_all(RUSTFS_META_BUCKET, &checkpoint_path)
+        .await
+        .expect("read checkpoint fixture");
+    let tampered = original
+        .windows(b"victim-a".len())
+        .position(|window| window == b"victim-a")
+        .map(|index| {
+            let mut bytes = original.to_vec();
+            bytes[index..index + b"victim-a".len()].copy_from_slice(b"victim-b");
+            bytes
+        })
+        .expect("checkpoint should contain the processed object");
+    disk.write_all(RUSTFS_META_BUCKET, &checkpoint_path, tampered.into())
+        .await
+        .expect("write tampered checkpoint fixture");
+
+    assert!(CheckpointManager::load_from_disk(disk.clone(), &task_id).await.is_err());
+    assert!(CheckpointManager::is_blocked(&disk, &task_id).await);
+    temp_dir.close().expect("remove digest test directory");
+}
+
+#[tokio::test]
+async fn new_checkpoint_manager_rebuilds_an_empty_snapshot() {
+    let (temp_dir, disk) = schema_test_disk().await;
+    let task_id = ResumeUtils::generate_task_id();
+    let checkpoint_path = format!("{BUCKET_META_PREFIX}/{task_id}_{RESUME_CHECKPOINT_FILE}");
+    disk.write_all(RUSTFS_META_BUCKET, &checkpoint_path, EcstoreDiskBytes::new())
+        .await
+        .expect("write empty checkpoint fixture");
+
+    let manager = CheckpointManager::new(disk.clone(), task_id.clone())
+        .await
+        .expect("a new manager must rebuild an empty checkpoint");
+    manager
+        .update_position(3, 11)
+        .await
+        .expect("rebuilt checkpoint must remain writable");
+    assert!(CheckpointManager::has_checkpoint(&disk, &task_id).await);
+    temp_dir.close().expect("remove empty checkpoint test directory");
+}
+
+#[tokio::test]
+async fn deleted_checkpoint_is_not_recreated_by_an_old_manager() {
+    let (temp_dir, disk) = schema_test_disk().await;
+    let task_id = ResumeUtils::generate_task_id();
+    let manager = CheckpointManager::new(disk.clone(), task_id.clone())
+        .await
+        .expect("create checkpoint manager");
+    manager.cleanup().await.expect("delete checkpoint fixture");
+
+    let error = manager
+        .update_position(1, 2)
+        .await
+        .expect_err("an old manager must not resurrect a deleted checkpoint");
+    assert!(error.to_string().contains("removed after this manager saved it"));
+    assert!(!CheckpointManager::has_checkpoint(&disk, &task_id).await);
+    temp_dir.close().expect("remove deleted checkpoint test directory");
+}
+
+#[tokio::test]
+async fn an_empty_blocked_marker_still_blocks_resume_selection() {
+    let (temp_dir, disk) = schema_test_disk().await;
+    let task_id = ResumeUtils::generate_task_id();
+    let manager = CheckpointManager::new(disk.clone(), task_id.clone())
+        .await
+        .expect("create checkpoint manager");
+    let blocked_path = format!("{BUCKET_META_PREFIX}/{task_id}_{RESUME_CHECKPOINT_BLOCKED_FILE}");
+    disk.write_all(RUSTFS_META_BUCKET, &blocked_path, EcstoreDiskBytes::new())
+        .await
+        .expect("write empty blocked marker fixture");
+
+    assert!(CheckpointManager::is_blocked(&disk, &task_id).await);
+    assert!(CheckpointManager::is_resumable(&disk, &task_id).await.is_err());
+    // Recovery requires replacing/cleaning the snapshot, then removing the
+    // marker; ordinary selector retries are intentionally not an unlock path.
+    manager.cleanup().await.expect("clean blocked checkpoint");
+    assert!(!CheckpointManager::is_blocked(&disk, &task_id).await);
+    temp_dir.close().expect("remove empty blocked marker test directory");
+}
+
+#[tokio::test]
+async fn resumable_selector_skips_healthy_tasks_with_blocked_markers() {
+    let (temp_dir, disk) = schema_test_disk().await;
+    let tasks = [
+        (ResumeUtils::generate_task_id(), EcstoreDiskBytes::new()),
+        (ResumeUtils::generate_task_id(), EcstoreDiskBytes::from_static(b"blocked")),
+    ];
+    for (task_id, marker) in &tasks {
+        ResumeManager::new(
+            disk.clone(),
+            task_id.clone(),
+            "erasure_set".to_string(),
+            "pool_0_set_0".to_string(),
+            vec!["bucket".to_string()],
+        )
+        .await
+        .expect("create healthy resume state");
+        CheckpointManager::new(disk.clone(), task_id.clone())
+            .await
+            .expect("create healthy checkpoint");
+        let checkpoint_path = format!("{BUCKET_META_PREFIX}/{task_id}_{RESUME_CHECKPOINT_FILE}");
+        let checkpoint_bytes = disk
+            .read_all(RUSTFS_META_BUCKET, &checkpoint_path)
+            .await
+            .expect("read healthy checkpoint before blocking");
+        let marker_path = format!("{BUCKET_META_PREFIX}/{task_id}_{RESUME_CHECKPOINT_BLOCKED_FILE}");
+        disk.write_all(RUSTFS_META_BUCKET, &marker_path, marker.clone())
+            .await
+            .expect("write blocked marker");
+
+        assert!(ResumeUtils::get_resumable_tasks(&disk).await.is_err());
+        assert_eq!(
+            disk.read_all(RUSTFS_META_BUCKET, &checkpoint_path)
+                .await
+                .expect("read healthy checkpoint after blocking"),
+            checkpoint_bytes
+        );
+    }
+    temp_dir.close().expect("remove blocked selector test directory");
+}
+
+#[tokio::test]
+async fn stale_checkpoint_manager_cannot_overwrite_newer_progress() {
+    let (temp_dir, disk) = schema_test_disk().await;
+    let task_id = ResumeUtils::generate_task_id();
+    let first = CheckpointManager::new(disk.clone(), task_id.clone())
+        .await
+        .expect("create first checkpoint manager");
+    let second = CheckpointManager::load_from_disk(disk.clone(), &task_id)
+        .await
+        .expect("load second checkpoint manager");
+
+    second
+        .update_position(4, 20)
+        .await
+        .expect("persist newer checkpoint progress");
+    let error = first
+        .update_position(1, 3)
+        .await
+        .expect_err("stale checkpoint manager must not overwrite newer progress");
+    assert!(error.to_string().contains("newer progress"));
+
+    let persisted = CheckpointManager::load_from_disk(disk.clone(), &task_id)
+        .await
+        .expect("load newer checkpoint progress")
+        .get_checkpoint()
+        .await;
+    assert_eq!(persisted.current_bucket_index, 4);
+    assert_eq!(persisted.current_object_index, 20);
+    temp_dir.close().expect("remove stale manager test directory");
+}
+
+#[tokio::test]
+async fn resumable_selector_isolates_future_and_corrupt_checkpoints() {
+    let (temp_dir, disk) = schema_test_disk().await;
+    let future_task = ResumeUtils::generate_task_id();
+    let corrupt_task = ResumeUtils::generate_task_id();
+    for task_id in [&future_task, &corrupt_task] {
+        ResumeManager::new(
+            disk.clone(),
+            task_id.to_string(),
+            "erasure_set".to_string(),
+            "pool_0_set_0".to_string(),
+            vec!["bucket".to_string()],
+        )
+        .await
+        .expect("create resumable state fixture");
+    }
+
+    let future_path = format!("{BUCKET_META_PREFIX}/{future_task}_{RESUME_CHECKPOINT_FILE}");
+    let mut future = ResumeCheckpoint::new(future_task.clone());
+    future.schema_version = CURRENT_CHECKPOINT_SCHEMA + 1;
+    let future_bytes = serde_json::to_vec(&future).expect("serialize future checkpoint fixture");
+    disk.write_all(RUSTFS_META_BUCKET, &future_path, future_bytes.clone().into())
+        .await
+        .expect("write future checkpoint fixture");
+    let corrupt_path = format!("{BUCKET_META_PREFIX}/{corrupt_task}_{RESUME_CHECKPOINT_FILE}");
+    let corrupt_bytes = b"{truncated";
+    disk.write_all(RUSTFS_META_BUCKET, &corrupt_path, corrupt_bytes.as_slice().into())
+        .await
+        .expect("write corrupt checkpoint fixture");
+
+    assert!(CheckpointManager::is_resumable(&disk, &future_task).await.is_err());
+    assert!(CheckpointManager::is_resumable(&disk, &corrupt_task).await.is_err());
+    assert!(ResumeUtils::get_resumable_tasks(&disk).await.is_err());
+    for (task_id, path, bytes) in [
+        (&future_task, future_path, future_bytes),
+        (&corrupt_task, corrupt_path, corrupt_bytes.to_vec()),
+    ] {
+        assert_eq!(
+            disk.read_all(RUSTFS_META_BUCKET, &path)
+                .await
+                .expect("read isolated checkpoint bytes"),
+            bytes
+        );
+        let blocked_path = format!("{BUCKET_META_PREFIX}/{task_id}_{RESUME_CHECKPOINT_BLOCKED_FILE}");
+        assert!(
+            !disk
+                .read_all(RUSTFS_META_BUCKET, &blocked_path)
+                .await
+                .expect("read checkpoint blocked marker")
+                .is_empty()
+        );
+    }
+    temp_dir.close().expect("remove selector isolation test directory");
+}
+
 #[test]
 fn test_persist_throttle_batches_until_threshold() {
     let mut throttle = PersistThrottle::new();

crates/heal/src/heal/resume/utils.rs

@@ -21,7 +21,7 @@ use uuid::Uuid;
 use super::super::{BUCKET_META_PREFIX, DiskError, DiskStore, HealDiskExt as _, RUSTFS_META_BUCKET};
 use super::replacement::{ReplacementPhase, ReplacementRecoveryRecord};
 use super::{
-    EVENT_HEAL_RESUME_STATE, LOG_COMPONENT_HEAL, LOG_SUBSYSTEM_RESUME, REPLACEMENT_COMPLETION_PROOF_FILE,
+    CheckpointManager, EVENT_HEAL_RESUME_STATE, LOG_COMPONENT_HEAL, LOG_SUBSYSTEM_RESUME, REPLACEMENT_COMPLETION_PROOF_FILE,
     REPLACEMENT_INTENT_FILE, RESUME_STATE_FILE, ResumeManager, ResumeStateFile, is_replacement_intent, path_to_str,
     replacement_recovery_corruption_for_state_load, replacement_recovery_dir, validate_resume_task_id,
 };
@@ -67,6 +67,7 @@ impl ResumeUtils {
                 // Extract task ID from filename: {task_id}_ahm_resume_state.json
                 if let Some(task_id) = entry.strip_suffix(&format!("_{RESUME_STATE_FILE}"))
                     && validate_resume_task_id(task_id).is_ok()
+                    && CheckpointManager::is_resumable(disk, task_id).await?
                 {
                     task_ids.push(task_id.to_string());
                 }

crates/protos/src/lib.rs

@@ -2106,9 +2106,6 @@ pub enum ChannelClass {
     Bulk,
 }
 
-// Keep multiplexed unary RPCs below h2's per-connection small-frame budget.
-const INTERNODE_RPC_CONCURRENCY_LIMIT: usize = 64;
-
 /// Whether control/bulk channel isolation is enabled (env-gated, default off for safe rollout).
 fn channel_isolation_enabled() -> bool {
     rustfs_utils::get_env_bool(
@@ -2191,7 +2188,6 @@ async fn build_channel(dial_addr: &str, cache_key: &str) -> Result<Channel, Box<
     let mut connector = Endpoint::from_shared(dial_addr.to_string())?
         // Fast connection timeout for dead peer detection
         .connect_timeout(connect_timeout)
-        .concurrency_limit(INTERNODE_RPC_CONCURRENCY_LIMIT)
         // TCP-level keepalive - OS will probe connection
         .tcp_keepalive(Some(tcp_keepalive))
         // Disable Nagle so latency-sensitive control-plane RPCs (locks/health) are not batched

rustfs/src/app/bucket_usecase.rs

@@ -513,15 +513,13 @@ fn sr_bucket_meta_item(bucket: String, item_type: &str) -> SRBucketMeta {
     }
 }
 
-async fn notify_bucket_metadata_reload(
+fn notify_bucket_metadata_reload(
     bucket: String,
     operation: &'static str,
     request_context: Option<request_context::RequestContext>,
     scanner_maintenance_change: bool,
 ) {
     record_local_scanner_maintenance_reload(&bucket, scanner_maintenance_change);
-    // Keep reload detached across request cancellation, but wait before a healthy peer can serve the previous config.
-    let (completed_tx, completed_rx) = tokio::sync::oneshot::channel();
     spawn_background_with_context(request_context, async move {
         if let Some(notification_sys) = current_notification_system() {
             let result = if scanner_maintenance_change {
@@ -533,9 +531,7 @@ async fn notify_bucket_metadata_reload(
                 warn!(bucket = %bucket, error = %err, "failed to notify peers after {operation}");
             }
         }
-        let _ = completed_tx.send(());
     });
-    let _ = completed_rx.await;
 }
 
 fn record_local_scanner_maintenance_reload(bucket: &str, scanner_maintenance_change: bool) {
@@ -1480,7 +1476,7 @@ impl DefaultBucketUsecase {
             .await
             .map_err(ApiError::from)?;
 
-        notify_bucket_metadata_reload(bucket.clone(), "delete bucket encryption", request_context, false).await;
+        notify_bucket_metadata_reload(bucket.clone(), "delete bucket encryption", request_context, false);
 
         let item = sr_bucket_meta_item(bucket.clone(), "sse-config");
         if let Err(err) = site_replication_bucket_meta_hook(item).await {
@@ -1512,7 +1508,7 @@ impl DefaultBucketUsecase {
             .await
             .map_err(ApiError::from)?;
 
-        notify_bucket_metadata_reload(bucket.clone(), "delete bucket cors", request_context, false).await;
+        notify_bucket_metadata_reload(bucket.clone(), "delete bucket cors", request_context, false);
 
         let item = sr_bucket_meta_item(bucket.clone(), "cors-config");
         if let Err(err) = site_replication_bucket_meta_hook(item).await {
@@ -1544,7 +1540,7 @@ impl DefaultBucketUsecase {
             .await
             .map_err(ApiError::from)?;
 
-        notify_bucket_metadata_reload(bucket.clone(), "delete bucket lifecycle", request_context, true).await;
+        notify_bucket_metadata_reload(bucket.clone(), "delete bucket lifecycle", request_context, true);
 
         let item = sr_bucket_meta_item(bucket.clone(), "lc-config");
         if let Err(err) = site_replication_bucket_meta_hook(item).await {
@@ -1576,7 +1572,7 @@ impl DefaultBucketUsecase {
             .await
             .map_err(ApiError::from)?;
 
-        notify_bucket_metadata_reload(bucket.clone(), "delete bucket policy", request_context, false).await;
+        notify_bucket_metadata_reload(bucket.clone(), "delete bucket policy", request_context, false);
 
         let item = sr_bucket_meta_item(bucket.clone(), "policy");
         if let Err(err) = site_replication_bucket_meta_hook(item).await {
@@ -1634,7 +1630,7 @@ impl DefaultBucketUsecase {
         }
         drop(targets_guard);
 
-        notify_bucket_metadata_reload(bucket.clone(), "delete bucket replication", request_context, true).await;
+        notify_bucket_metadata_reload(bucket.clone(), "delete bucket replication", request_context, true);
 
         let item = sr_bucket_meta_item(bucket.clone(), "replication-config");
         if let Err(err) = site_replication_bucket_meta_hook(item).await {
@@ -1659,7 +1655,7 @@ impl DefaultBucketUsecase {
             .await
             .map_err(ApiError::from)?;
 
-        notify_bucket_metadata_reload(bucket.clone(), "delete bucket tagging", request_context, false).await;
+        notify_bucket_metadata_reload(bucket.clone(), "delete bucket tagging", request_context, false);
 
         let item = sr_bucket_meta_item(bucket.clone(), "tags");
         if let Err(err) = site_replication_bucket_meta_hook(item).await {
@@ -1692,7 +1688,7 @@ impl DefaultBucketUsecase {
             .await
             .map_err(ApiError::from)?;
 
-        notify_bucket_metadata_reload(bucket.clone(), "delete public access block", request_context, false).await;
+        notify_bucket_metadata_reload(bucket.clone(), "delete public access block", request_context, false);
 
         Ok(S3Response::with_status(DeletePublicAccessBlockOutput::default(), StatusCode::NO_CONTENT))
     }
@@ -2147,7 +2143,7 @@ impl DefaultBucketUsecase {
             .await
             .map_err(ApiError::from)?;
 
-        notify_bucket_metadata_reload(bucket.clone(), "put bucket encryption", request_context, false).await;
+        notify_bucket_metadata_reload(bucket.clone(), "put bucket encryption", request_context, false);
 
         let mut item = sr_bucket_meta_item(bucket.clone(), "sse-config");
         item.sse_config = Some(
@@ -2226,7 +2222,7 @@ impl DefaultBucketUsecase {
             .await
             .map_err(ApiError::from)?;
 
-        notify_bucket_metadata_reload(bucket.clone(), "put bucket lifecycle", request_context, true).await;
+        notify_bucket_metadata_reload(bucket.clone(), "put bucket lifecycle", request_context, true);
 
         let mut item = sr_bucket_meta_item(bucket.clone(), "lc-config");
         item.expiry_lc_config =
@@ -2311,7 +2307,7 @@ impl DefaultBucketUsecase {
             .await
             .map_err(ApiError::from)?;
 
-        notify_bucket_metadata_reload(bucket.clone(), "put bucket notification", request_context, false).await;
+        notify_bucket_metadata_reload(bucket.clone(), "put bucket notification", request_context, false);
 
         let region = resolve_notification_region(self.global_region(), request_region);
         let notify = current_notify_interface_for_context(self.context.as_deref());
@@ -2416,7 +2412,7 @@ impl DefaultBucketUsecase {
             .await
             .map_err(ApiError::from)?;
 
-        notify_bucket_metadata_reload(bucket.clone(), "put bucket policy", request_context, false).await;
+        notify_bucket_metadata_reload(bucket.clone(), "put bucket policy", request_context, false);
 
         let mut item = sr_bucket_meta_item(bucket.clone(), "policy");
         item.policy = Some(serde_json::from_str(&policy).map_err(|e| s3_error!(InvalidArgument, "parse policy failed {:?}", e))?);
@@ -2451,7 +2447,7 @@ impl DefaultBucketUsecase {
             .await
             .map_err(ApiError::from)?;
 
-        notify_bucket_metadata_reload(bucket.clone(), "put bucket cors", request_context, false).await;
+        notify_bucket_metadata_reload(bucket.clone(), "put bucket cors", request_context, false);
 
         let mut item = sr_bucket_meta_item(bucket.clone(), "cors-config");
         item.cors =
@@ -2495,7 +2491,7 @@ impl DefaultBucketUsecase {
             .map_err(ApiError::from)?;
         drop(targets_guard);
 
-        notify_bucket_metadata_reload(bucket.clone(), "put bucket replication", request_context, true).await;
+        notify_bucket_metadata_reload(bucket.clone(), "put bucket replication", request_context, true);
 
         let mut item = sr_bucket_meta_item(bucket.clone(), "replication-config");
         item.replication_config = Some(
@@ -2535,7 +2531,7 @@ impl DefaultBucketUsecase {
             .await
             .map_err(ApiError::from)?;
 
-        notify_bucket_metadata_reload(bucket.clone(), "put public access block", request_context, false).await;
+        notify_bucket_metadata_reload(bucket.clone(), "put public access block", request_context, false);
 
         Ok(S3Response::new(PutPublicAccessBlockOutput::default()))
     }
@@ -2564,7 +2560,7 @@ impl DefaultBucketUsecase {
             .await
             .map_err(ApiError::from)?;
 
-        notify_bucket_metadata_reload(bucket.clone(), "put bucket tagging", request_context, false).await;
+        notify_bucket_metadata_reload(bucket.clone(), "put bucket tagging", request_context, false);
 
         let mut item = sr_bucket_meta_item(bucket.clone(), "tags");
         item.tags = Some(serialize_config(&tagging).and_then(|bytes| String::from_utf8(bytes).map_err(to_internal_error))?);
@@ -2597,7 +2593,7 @@ impl DefaultBucketUsecase {
             .await
             .map_err(ApiError::from)?;
 
-        notify_bucket_metadata_reload(bucket.clone(), "put bucket versioning", request_context, false).await;
+        notify_bucket_metadata_reload(bucket.clone(), "put bucket versioning", request_context, false);
 
         let mut item = sr_bucket_meta_item(bucket.clone(), "version-config");
         item.versioning = Some(
@@ -3048,7 +3044,7 @@ mod tests {
                 "{method} should identify the bucket metadata operation in reload logs"
             );
             let expected_reload = format!(
-                "notify_bucket_metadata_reload(bucket.clone(), \"{operation}\", request_context, {scanner_maintenance_change}).await;"
+                "notify_bucket_metadata_reload(bucket.clone(), \"{operation}\", request_context, {scanner_maintenance_change});"
             );
             assert!(
                 body.contains(&expected_reload),