refactor(heal): split resume.rs into focused child modules

Split the 4242-line resume.rs (46% inline tests) into a canonical
foo.rs + foo/ module tree with zero behavior change:

- resume.rs (~1020): state file constants, PersistThrottle, ResumeState,
  ResumeManager core (constructors, load/discovery, progress mutators,
  ordinary persistence) plus root re-exports
- resume/replacement.rs (~690): replacement-intent/proof types and the
  ResumeManager replacement-lifecycle methods
- resume/checkpoint.rs (~350): ResumeCheckpoint + CheckpointManager
- resume/utils.rs (~310): ResumeUtils statics
- resume/tests.rs (~1980): the inline test module as a child module

All module paths are unchanged (heal::resume::CheckpointManager and
friends resolve through root re-exports), so no consumer inside or
outside the crate changes. Items defined in child modules keep
module-private visibility; only the ten cross-module helpers gain
pub(super), which is not part of the crate API. Code is moved verbatim
apart from those visibility markers, four super::storage_api path
fixes, and the new per-module import headers.

Co-Authored-By: heihutu <heihutu@gmail.com>

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

@@ -0,0 +1,351 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use crate::{Error, Result};
+use serde::{Deserialize, Serialize};
+use std::collections::HashSet;
+use std::path::Path;
+use std::sync::{Arc, Mutex};
+use std::time::{SystemTime, UNIX_EPOCH};
+use tokio::sync::RwLock;
+use tracing::{debug, warn};
+
+use super::super::{BUCKET_META_PREFIX, DiskStore, HealDiskExt as _, 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,
+};
+
+const EVENT_HEAL_CHECKPOINT_STATE: &str = "heal_checkpoint_state";
+
+/// Current on-disk schema version for `ResumeCheckpoint`. Same rationale as
+/// `CURRENT_RESUME_SCHEMA`: pre-per-version dedup identities are not comparable
+/// to the new `compose_key` identities, so a stale checkpoint is discarded.
+pub(super) const CURRENT_CHECKPOINT_SCHEMA: u32 = 5;
+
+/// resume checkpoint
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct ResumeCheckpoint {
+    /// on-disk schema version; absent in legacy snapshots (defaults to 0)
+    #[serde(default)]
+    pub schema_version: u32,
+    /// task id
+    pub task_id: String,
+    /// checkpoint time
+    pub checkpoint_time: u64,
+    /// current bucket index
+    pub current_bucket_index: usize,
+    /// current object index
+    pub current_object_index: usize,
+    /// Objects healed since the last completed page. HashSet: with the
+    /// previous Vec the per-object `contains` was O(n) and made large-bucket
+    /// heals O(N²). Only spans the in-flight page — completed pages are
+    /// covered by `current_object_index`, so `complete_page` prunes the sets.
+    pub processed_objects: HashSet<String>,
+    /// failed objects
+    pub failed_objects: HashSet<String>,
+    /// skipped objects
+    pub skipped_objects: HashSet<String>,
+}
+
+impl ResumeCheckpoint {
+    pub fn new(task_id: String) -> Self {
+        Self {
+            schema_version: CURRENT_CHECKPOINT_SCHEMA,
+            task_id,
+            checkpoint_time: SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_secs(),
+            current_bucket_index: 0,
+            current_object_index: 0,
+            processed_objects: HashSet::new(),
+            failed_objects: HashSet::new(),
+            skipped_objects: HashSet::new(),
+        }
+    }
+
+    pub fn update_position(&mut self, bucket_index: usize, object_index: usize) {
+        self.current_bucket_index = bucket_index;
+        self.current_object_index = object_index;
+        self.checkpoint_time = SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_secs();
+    }
+
+    pub fn add_processed_object(&mut self, object: String) {
+        self.processed_objects.insert(object);
+    }
+
+    pub fn add_failed_object(&mut self, object: String) {
+        self.failed_objects.insert(object);
+    }
+
+    pub fn add_skipped_object(&mut self, object: String) {
+        self.skipped_objects.insert(object);
+    }
+
+    /// Advance past a fully-processed page: objects below `object_index` are
+    /// skipped by position on resume, so the per-object sets no longer need
+    /// their entries and would otherwise grow with the whole bucket.
+    pub fn complete_page(&mut self, bucket_index: usize, object_index: usize) {
+        self.update_position(bucket_index, object_index);
+        self.processed_objects.clear();
+        self.skipped_objects.clear();
+        self.failed_objects.clear();
+    }
+
+    /// Reset the scan to the start and clear the per-object sets so a retry
+    /// re-scans the whole set.
+    pub fn reset_for_retry(&mut self) {
+        self.update_position(0, 0);
+        self.processed_objects.clear();
+        self.skipped_objects.clear();
+        self.failed_objects.clear();
+    }
+}
+
+/// resume checkpoint manager
+pub struct CheckpointManager {
+    disk: DiskStore,
+    checkpoint: Arc<RwLock<ResumeCheckpoint>>,
+    throttle: Mutex<PersistThrottle>,
+}
+
+impl CheckpointManager {
+    /// create new checkpoint manager
+    pub async fn new(disk: DiskStore, task_id: String) -> Result<Self> {
+        validate_resume_task_id(&task_id)?;
+        let checkpoint = ResumeCheckpoint::new(task_id);
+        let manager = Self {
+            disk,
+            checkpoint: Arc::new(RwLock::new(checkpoint)),
+            throttle: Mutex::new(PersistThrottle::new()),
+        };
+
+        // save initial checkpoint
+        if let Err(e) = manager.save_checkpoint().await {
+            warn!(
+                target: "rustfs::heal::resume",
+                event = EVENT_HEAL_CHECKPOINT_STATE,
+                component = LOG_COMPONENT_HEAL,
+                subsystem = LOG_SUBSYSTEM_RESUME,
+                state = "initial_save_failed",
+                error = %e,
+                "Heal checkpoint persistence failed"
+            );
+        }
+        Ok(manager)
+    }
+
+    /// load checkpoint from disk
+    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}"),
+            })?;
+        if checkpoint.task_id != task_id {
+            return Err(Error::TaskExecutionFailed {
+                message: "Resume checkpoint task id does not match filename".to_string(),
+            });
+        }
+
+        // A checkpoint from an older schema stored latest-only dedup identities
+        // that are not comparable to the new per-version `compose_key`
+        // identities. Discard the stale sets and position, then stamp the
+        // current schema so the scan restarts cleanly.
+        if checkpoint.schema_version > CURRENT_CHECKPOINT_SCHEMA {
+            return Err(Error::TaskExecutionFailed {
+                message: format!(
+                    "Checkpoint schema {} is newer than supported schema {CURRENT_CHECKPOINT_SCHEMA}",
+                    checkpoint.schema_version
+                ),
+            });
+        }
+        if checkpoint.schema_version < CURRENT_CHECKPOINT_SCHEMA {
+            warn!(
+                target: "rustfs::heal::resume",
+                event = EVENT_HEAL_CHECKPOINT_STATE,
+                component = LOG_COMPONENT_HEAL,
+                subsystem = LOG_SUBSYSTEM_RESUME,
+                task_id,
+                found_schema = checkpoint.schema_version,
+                current_schema = CURRENT_CHECKPOINT_SCHEMA,
+                state = "schema_discarded",
+                "Heal checkpoint schema is stale; discarding dedup sets and position"
+            );
+            checkpoint.processed_objects.clear();
+            checkpoint.failed_objects.clear();
+            checkpoint.skipped_objects.clear();
+            checkpoint.current_bucket_index = 0;
+            checkpoint.current_object_index = 0;
+            checkpoint.schema_version = CURRENT_CHECKPOINT_SCHEMA;
+        }
+
+        Ok(Self {
+            disk,
+            checkpoint: Arc::new(RwLock::new(checkpoint)),
+            throttle: Mutex::new(PersistThrottle::new()),
+        })
+    }
+
+    /// check if checkpoint exists
+    pub async fn has_checkpoint(disk: &DiskStore, task_id: &str) -> bool {
+        if validate_resume_task_id(task_id).is_err() {
+            return false;
+        }
+        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(data) => !data.is_empty(),
+                Err(_) => false,
+            },
+            Err(_) => false,
+        }
+    }
+
+    /// get current checkpoint
+    pub async fn get_checkpoint(&self) -> ResumeCheckpoint {
+        self.checkpoint.read().await.clone()
+    }
+
+    /// update position
+    pub async fn update_position(&self, bucket_index: usize, object_index: usize) -> Result<()> {
+        let mut checkpoint = self.checkpoint.write().await;
+        checkpoint.update_position(bucket_index, object_index);
+        drop(checkpoint);
+        self.save_checkpoint_throttled().await
+    }
+
+    /// Advance past a completed page and prune the per-object sets, then persist.
+    pub async fn complete_page(&self, bucket_index: usize, object_index: usize) -> Result<()> {
+        let mut checkpoint = self.checkpoint.write().await;
+        checkpoint.complete_page(bucket_index, object_index);
+        drop(checkpoint);
+        self.save_checkpoint_throttled().await
+    }
+
+    /// Reset the checkpoint to the start of the scan for a retry, then persist.
+    pub async fn reset_for_retry(&self) -> Result<()> {
+        let mut checkpoint = self.checkpoint.write().await;
+        checkpoint.reset_for_retry();
+        drop(checkpoint);
+        self.save_checkpoint_throttled().await
+    }
+
+    /// Add a processed object. Called once per healed object, so persistence
+    /// is batched (`PERSIST_EVERY_MUTATIONS` / `PERSIST_INTERVAL`); positions
+    /// and page boundaries still persist unconditionally.
+    pub async fn add_processed_object(&self, object: String) -> Result<()> {
+        let mut checkpoint = self.checkpoint.write().await;
+        checkpoint.add_processed_object(object);
+        drop(checkpoint);
+        self.save_checkpoint_if_due().await
+    }
+
+    /// add failed object (batched, see `add_processed_object`)
+    pub async fn add_failed_object(&self, object: String) -> Result<()> {
+        let mut checkpoint = self.checkpoint.write().await;
+        checkpoint.add_failed_object(object);
+        drop(checkpoint);
+        self.save_checkpoint_if_due().await
+    }
+
+    /// add skipped object (batched, see `add_processed_object`)
+    pub async fn add_skipped_object(&self, object: String) -> Result<()> {
+        let mut checkpoint = self.checkpoint.write().await;
+        checkpoint.add_skipped_object(object);
+        drop(checkpoint);
+        self.save_checkpoint_if_due().await
+    }
+
+    async fn save_checkpoint_if_due(&self) -> Result<()> {
+        let should_save = self.throttle.lock().map(|mut throttle| throttle.record()).unwrap_or(true);
+        if !should_save {
+            return Ok(());
+        }
+        self.save_checkpoint_throttled().await
+    }
+
+    async fn save_checkpoint_throttled(&self) -> Result<()> {
+        let result = self.save_checkpoint().await;
+        if result.is_ok()
+            && let Ok(mut throttle) = self.throttle.lock()
+        {
+            throttle.mark_saved();
+        }
+        result
+    }
+
+    /// cleanup checkpoint
+    pub async fn cleanup(&self) -> Result<()> {
+        let task_id = self.checkpoint.read().await.task_id.clone();
+        validate_resume_task_id(&task_id)?;
+
+        let checkpoint_file = Path::new(BUCKET_META_PREFIX).join(format!("{task_id}_{RESUME_CHECKPOINT_FILE}"));
+        delete_resume_file(&self.disk, &checkpoint_file).await?;
+
+        debug!(
+            target: "rustfs::heal::resume",
+            event = EVENT_HEAL_CHECKPOINT_STATE,
+            component = LOG_COMPONENT_HEAL,
+            subsystem = LOG_SUBSYSTEM_RESUME,
+            task_id,
+            state = "cleaned",
+            "Heal checkpoint cleaned"
+        );
+        Ok(())
+    }
+
+    /// save checkpoint to disk
+    async fn save_checkpoint(&self) -> Result<()> {
+        let checkpoint = self.checkpoint.read().await;
+        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 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())
+            .await
+            .map_err(|e| Error::TaskExecutionFailed {
+                message: format!("Failed to save checkpoint: {e}"),
+            })?;
+
+        debug!(
+            target: "rustfs::heal::resume",
+            event = EVENT_HEAL_CHECKPOINT_STATE,
+            component = LOG_COMPONENT_HEAL,
+            subsystem = LOG_SUBSYSTEM_RESUME,
+            task_id = %checkpoint.task_id,
+            state = "saved",
+            "Heal checkpoint persisted"
+        );
+        Ok(())
+    }
+
+    /// read checkpoint file from disk
+    async fn read_checkpoint_file(disk: &DiskStore, task_id: &str) -> Result<Vec<u8>> {
+        validate_resume_task_id(task_id)?;
+        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)
+            .await
+            .map(|bytes| bytes.to_vec())
+            .map_err(|e| Error::TaskExecutionFailed {
+                message: format!("Failed to read checkpoint file: {e}"),
+            })
+    }
+}

crates/heal/src/heal/resume/replacement.rs

@@ -0,0 +1,688 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use crate::{Error, Result};
+use serde::{Deserialize, Serialize};
+use std::collections::HashSet;
+use std::time::{SystemTime, UNIX_EPOCH};
+
+use super::super::HealDiskExt as _;
+
+use super::super::storage_api::owner::{EcstoreConditionalFileUpdate, EcstoreDiskBytes};
+use super::{
+    DiskError, DiskStore, RUSTFS_META_BUCKET, ResumeManager, ResumeState, delete_resume_file, ensure_replacement_recovery_dir,
+    injected_replacement_proof_write_error, is_replacement_intent, legacy_replacement_completion_proof_path, path_to_str,
+    replacement_completion_proof_path, replacement_intent_seal_path, replacement_recovery_conflict,
+    replacement_recovery_corruption, validate_resume_task_id,
+};
+
+/// Durable-proof schema version.
+const CURRENT_REPLACEMENT_COMPLETION_PROOF_SCHEMA: u32 = 1;
+
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum ReplacementPhase {
+    #[default]
+    None,
+    Intent,
+    Rebuilding,
+    Verified,
+    CleanupPending,
+    Abandoned,
+}
+
+/// Target-specific state for a durable automatic replacement generation.
+///
+/// This is deliberately separate from the legacy background-heal status
+/// contract. Consumers must treat [`Self::Unknown`] as non-definitive.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum ReplacementRecoveryState {
+    WaitingForReplacement,
+    Running,
+    Incomplete,
+    Unrecoverable,
+    CleanupPending,
+    Completed,
+    Unknown,
+}
+
+/// Read-only status derived from one durable replacement generation.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct ReplacementRecoveryRecord {
+    pub task_id: String,
+    pub state: ReplacementRecoveryState,
+    pub generation: Option<String>,
+    pub set_disk_id: Option<String>,
+    pub target_slots: Vec<String>,
+    pub reason: Option<String>,
+    pub verified_at: Option<u64>,
+}
+
+impl ReplacementRecoveryRecord {
+    pub(super) fn from_state(state: ResumeState) -> Option<Self> {
+        if !is_replacement_intent(&state) {
+            return None;
+        }
+
+        let invariant_holds = state.replacement_generation.as_deref() == Some(state.task_id.as_str())
+            && replacement_targets_match_identities(&state.replacement_targets, &state.replacement_target_identities);
+        if !invariant_holds {
+            return Some(Self::unknown(
+                state.task_id,
+                "durable replacement state violates its generation or target identity binding",
+            ));
+        }
+
+        let (state_kind, reason) = if !state.completed && state.retry_count >= state.max_retries {
+            (
+                ReplacementRecoveryState::Unrecoverable,
+                Some("replacement retry budget exhausted".to_string()),
+            )
+        } else if let Some(reason) = state.error_message.clone() {
+            (ReplacementRecoveryState::Incomplete, Some(reason))
+        } else {
+            match state.replacement_phase {
+                ReplacementPhase::Intent => (ReplacementRecoveryState::WaitingForReplacement, None),
+                ReplacementPhase::Rebuilding => (ReplacementRecoveryState::Running, None),
+                ReplacementPhase::Verified | ReplacementPhase::CleanupPending => (ReplacementRecoveryState::CleanupPending, None),
+                ReplacementPhase::Abandoned => (
+                    ReplacementRecoveryState::Unrecoverable,
+                    Some("replacement generation was abandoned".to_string()),
+                ),
+                ReplacementPhase::None => (ReplacementRecoveryState::Unknown, Some("replacement phase is missing".to_string())),
+            }
+        };
+
+        Some(Self {
+            task_id: state.task_id,
+            state: state_kind,
+            generation: state.replacement_generation,
+            set_disk_id: Some(state.set_disk_id),
+            target_slots: state.replacement_targets,
+            reason,
+            verified_at: None,
+        })
+    }
+
+    pub(super) fn from_completion_proof(proof: &ReplacementCompletionProof) -> Self {
+        Self {
+            task_id: proof.task_id.clone(),
+            state: ReplacementRecoveryState::Completed,
+            generation: Some(proof.replacement_generation.clone()),
+            set_disk_id: Some(proof.set_disk_id.clone()),
+            target_slots: proof.replacement_targets.clone(),
+            reason: None,
+            verified_at: Some(proof.verified_at),
+        }
+    }
+
+    pub(super) fn unknown(task_id: String, reason: &str) -> Self {
+        Self {
+            task_id,
+            state: ReplacementRecoveryState::Unknown,
+            generation: None,
+            set_disk_id: None,
+            target_slots: Vec::new(),
+            reason: Some(reason.to_string()),
+            verified_at: None,
+        }
+    }
+}
+
+pub(super) fn replacement_targets_match_identities(targets: &[String], identities: &[ReplacementTargetIdentity]) -> bool {
+    !targets.is_empty()
+        && targets.len() == identities.len()
+        && targets.iter().collect::<HashSet<_>>().len() == targets.len()
+        && identities.iter().map(|identity| &identity.endpoint).eq(targets.iter())
+}
+
+/// Stable evidence for the mounted replacement instance that owns a repair
+/// generation. Endpoint text alone is not sufficient because a later disk can
+/// be mounted at the same configured path.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct ReplacementTargetIdentity {
+    pub endpoint: String,
+    pub canonical_path: String,
+    pub physical_device_ids: Vec<String>,
+    pub filesystem_identity: String,
+}
+
+/// Durable terminal evidence for one automatic replacement generation. This
+/// lives on the healthy non-target anchor rather than in the resumable state,
+/// because resume cleanup must not erase proof that the generation completed.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub(crate) struct ReplacementCompletionProof {
+    pub schema_version: u32,
+    pub task_id: String,
+    pub replacement_generation: String,
+    pub set_disk_id: String,
+    pub replacement_targets: Vec<String>,
+    pub replacement_target_identities: Vec<ReplacementTargetIdentity>,
+    pub verified_at: u64,
+}
+
+impl ReplacementCompletionProof {
+    pub(super) fn from_state(state: &ResumeState, verified_at: u64) -> Result<Self> {
+        let replacement_generation = state
+            .replacement_generation
+            .clone()
+            .ok_or_else(|| Error::TaskExecutionFailed {
+                message: format!("Replacement completion has no generation for task {}", state.task_id),
+            })?;
+        if replacement_generation != state.task_id
+            || state.replacement_targets.is_empty()
+            || state
+                .replacement_target_identities
+                .iter()
+                .map(|identity| &identity.endpoint)
+                .collect::<Vec<_>>()
+                != state.replacement_targets.iter().collect::<Vec<_>>()
+        {
+            return Err(Error::TaskExecutionFailed {
+                message: format!("Replacement completion identity does not match task {}", state.task_id),
+            });
+        }
+
+        Ok(Self {
+            schema_version: CURRENT_REPLACEMENT_COMPLETION_PROOF_SCHEMA,
+            task_id: state.task_id.clone(),
+            replacement_generation,
+            set_disk_id: state.set_disk_id.clone(),
+            replacement_targets: state.replacement_targets.clone(),
+            replacement_target_identities: state.replacement_target_identities.clone(),
+            verified_at,
+        })
+    }
+
+    fn matches_state(&self, state: &ResumeState) -> bool {
+        self.schema_version == CURRENT_REPLACEMENT_COMPLETION_PROOF_SCHEMA
+            && self.task_id == state.task_id
+            && state.replacement_generation.as_deref() == Some(self.replacement_generation.as_str())
+            && self.set_disk_id == state.set_disk_id
+            && self.replacement_targets == state.replacement_targets
+            && self.replacement_target_identities == state.replacement_target_identities
+    }
+
+    fn validate(&self, expected_task_id: &str) -> Result<()> {
+        if self.schema_version != CURRENT_REPLACEMENT_COMPLETION_PROOF_SCHEMA {
+            return Err(Error::TaskExecutionFailed {
+                message: format!("Replacement completion proof schema {} is unsupported", self.schema_version),
+            });
+        }
+        validate_resume_task_id(expected_task_id)?;
+        if self.task_id != expected_task_id
+            || self.replacement_generation != self.task_id
+            || self.set_disk_id.is_empty()
+            || self.verified_at == 0
+            || !replacement_targets_match_identities(&self.replacement_targets, &self.replacement_target_identities)
+        {
+            return Err(Error::TaskExecutionFailed {
+                message: format!("Replacement completion proof does not match task {expected_task_id}"),
+            });
+        }
+        Ok(())
+    }
+}
+
+pub(crate) fn replacement_target_identities_match(
+    expected: &[ReplacementTargetIdentity],
+    actual: &[ReplacementTargetIdentity],
+) -> bool {
+    let mut expected = expected.to_vec();
+    let mut actual = actual.to_vec();
+    expected.sort_by(|left, right| left.endpoint.cmp(&right.endpoint));
+    actual.sort_by(|left, right| left.endpoint.cmp(&right.endpoint));
+    expected == actual
+}
+
+/// Build the canonical, provably-injective dedup identity for an object
+/// version. Length-prefixing the object key makes the encoding injective: no
+/// two distinct `(object, version_id)` pairs can collide, even for adversarial
+/// keys containing `:` or embedded null bytes. This is the single source of
+/// truth for per-version dedup across the heal loop and the checkpoint sets.
+pub fn compose_key(object: &str, version_id: Option<&str>) -> String {
+    format!("{}:{}{}", object.len(), object, version_id.unwrap_or(""))
+}
+
+impl ResumeManager {
+    /// Seal a durably published intent before the caller may format a target.
+    /// A torn intent without this seal is known to have failed before its
+    /// creator returned and can be atomically recreated on retry.
+    pub(super) async fn ensure_replacement_intent_seal(&self) -> Result<()> {
+        let task_id = self.state.read().await.task_id.clone();
+        validate_resume_task_id(&task_id)?;
+        let path = replacement_intent_seal_path(&task_id);
+        let path = path_to_str(&path)?;
+        match self.disk.read_all(RUSTFS_META_BUCKET, path).await {
+            Ok(_) => return Ok(()),
+            Err(DiskError::FileNotFound) => {}
+            Err(error) => {
+                return Err(Error::TaskExecutionFailed {
+                    message: format!("Failed to read replacement intent seal: {error}"),
+                });
+            }
+        }
+        self.disk
+            .write_all(RUSTFS_META_BUCKET, path, b"sealed".as_slice().into())
+            .await
+            .map_err(|error| Error::TaskExecutionFailed {
+                message: format!("Failed to save replacement intent seal: {error}"),
+            })
+    }
+
+    pub async fn mark_replacement_rebuilding(
+        &self,
+        mut replacement_target_identities: Vec<ReplacementTargetIdentity>,
+    ) -> Result<()> {
+        replacement_target_identities.sort_by(|left, right| left.endpoint.cmp(&right.endpoint));
+        replacement_target_identities.dedup_by(|left, right| left.endpoint == right.endpoint);
+        let mut state = self.state.write().await;
+        if !matches!(state.replacement_phase, ReplacementPhase::Intent | ReplacementPhase::Rebuilding) {
+            return Err(Error::TaskExecutionFailed {
+                message: format!("Replacement intent is not active for task {}", state.task_id),
+            });
+        }
+        if replacement_target_identities
+            .iter()
+            .map(|identity| &identity.endpoint)
+            .collect::<Vec<_>>()
+            != state.replacement_targets.iter().collect::<Vec<_>>()
+        {
+            return Err(Error::TaskExecutionFailed {
+                message: format!("Replacement identities do not match targets for task {}", state.task_id),
+            });
+        }
+        if !replacement_target_identities_match(&state.replacement_target_identities, &replacement_target_identities) {
+            return Err(Error::TaskExecutionFailed {
+                message: format!("Replacement target changed after format for task {}", state.task_id),
+            });
+        }
+        state.replacement_phase = ReplacementPhase::Rebuilding;
+        state.last_update = SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_secs();
+        drop(state);
+        self.save_state_strict().await
+    }
+
+    /// Persist survivor-anchor completion proof before transitioning this
+    /// resumable state to `Verified`. If proof persistence fails, this state
+    /// stays rebuildable and the caller must retain the healing marker.
+    pub async fn mark_replacement_completed_and_verified(&self) -> Result<()> {
+        let state = self.state.read().await.clone();
+        if !matches!(state.replacement_phase, ReplacementPhase::Intent | ReplacementPhase::Rebuilding) {
+            return Err(Error::TaskExecutionFailed {
+                message: format!("Replacement verification is not active for task {}", state.task_id),
+            });
+        }
+        let proof = self.write_replacement_completion_proof(&state, None).await?;
+
+        let mut state = self.state.write().await;
+        if !matches!(state.replacement_phase, ReplacementPhase::Intent | ReplacementPhase::Rebuilding) {
+            return Err(Error::TaskExecutionFailed {
+                message: format!("Replacement verification changed for task {}", state.task_id),
+            });
+        }
+        state.mark_completed();
+        state.replacement_phase = ReplacementPhase::Verified;
+        state.last_update = proof.verified_at;
+        drop(state);
+        self.save_state_strict().await
+    }
+
+    /// Verify or backfill the terminal proof before marker removal or resume
+    /// cleanup. This supports restart recovery from a `Verified` state written
+    /// by a prior binary that did not yet have a separate proof record.
+    pub(crate) async fn ensure_replacement_completion_proof(&self) -> Result<ReplacementCompletionProof> {
+        let state = self.state.read().await.clone();
+        if !state.completed || !matches!(state.replacement_phase, ReplacementPhase::Verified | ReplacementPhase::CleanupPending) {
+            return Err(Error::TaskExecutionFailed {
+                message: format!("Replacement completion is not verified for task {}", state.task_id),
+            });
+        }
+        self.write_replacement_completion_proof(&state, Some(state.last_update)).await
+    }
+
+    /// Record that the healing markers have been removed, so a later retry can
+    /// safely delete the remaining resume artifacts without touching markers.
+    pub async fn mark_replacement_cleanup_pending(&self) -> Result<()> {
+        let mut state = self.state.write().await;
+        if !state.completed || !matches!(state.replacement_phase, ReplacementPhase::Verified | ReplacementPhase::CleanupPending) {
+            return Err(Error::TaskExecutionFailed {
+                message: format!("Replacement cleanup is not ready for task {}", state.task_id),
+            });
+        }
+        state.replacement_phase = ReplacementPhase::CleanupPending;
+        state.last_update = SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_secs();
+        drop(state);
+        self.save_state_strict().await
+    }
+
+    /// Load the durable terminal proof from the healthy survivor anchor.
+    pub(crate) async fn load_replacement_completion_proof(disk: DiskStore, task_id: &str) -> Result<ReplacementCompletionProof> {
+        Self::replacement_completion_proof_if_present(disk, task_id)
+            .await?
+            .ok_or_else(|| Error::TaskExecutionFailed {
+                message: format!("Failed to read replacement completion proof: proof is missing for task {task_id}"),
+            })
+    }
+
+    async fn replacement_completion_proof_if_present(
+        disk: DiskStore,
+        task_id: &str,
+    ) -> Result<Option<ReplacementCompletionProof>> {
+        validate_resume_task_id(task_id)?;
+        let mut proofs = Vec::new();
+        for path in [
+            replacement_completion_proof_path(task_id),
+            legacy_replacement_completion_proof_path(task_id),
+        ] {
+            let path_str = path_to_str(&path)?;
+            let bytes = match disk.read_all(RUSTFS_META_BUCKET, path_str).await {
+                Ok(bytes) => bytes,
+                Err(DiskError::FileNotFound) => continue,
+                Err(error) => {
+                    return Err(Error::TaskExecutionFailed {
+                        message: format!("Failed to read replacement completion proof: {error}"),
+                    });
+                }
+            };
+            let proof: ReplacementCompletionProof =
+                serde_json::from_slice(&bytes).map_err(|error| Error::TaskExecutionFailed {
+                    message: format!("Failed to deserialize replacement completion proof: {error}"),
+                })?;
+            proof.validate(task_id)?;
+            proofs.push(proof);
+        }
+
+        match proofs.as_slice() {
+            [] => Ok(None),
+            [proof] => Ok(Some(proof.clone())),
+            [proof, legacy_proof] if proof == legacy_proof => Ok(Some(proof.clone())),
+            _ => Err(replacement_recovery_conflict(format!(
+                "Replacement completion proof conflicts with legacy proof for task {task_id}"
+            ))),
+        }
+    }
+
+    /// Reconcile the proof-first publication order after a crash. A matching
+    /// proof is durable evidence that rebuilding finished, so it must win over
+    /// an older active state before a retry may format the target again.
+    pub(super) async fn reconcile_replacement_completion_proof(&self) -> Result<()> {
+        let task_id = self.state.read().await.task_id.clone();
+        let Some(proof) = Self::replacement_completion_proof_if_present(self.disk.clone(), &task_id).await? else {
+            return Ok(());
+        };
+
+        let mut state = self.state.write().await;
+        if !proof.matches_state(&state) {
+            return Err(Error::TaskExecutionFailed {
+                message: format!("Replacement completion proof does not match active intent for task {}", state.task_id),
+            });
+        }
+        if state.completed && matches!(state.replacement_phase, ReplacementPhase::Verified | ReplacementPhase::CleanupPending) {
+            return Ok(());
+        }
+        if state.completed || !matches!(state.replacement_phase, ReplacementPhase::Intent | ReplacementPhase::Rebuilding) {
+            return Err(replacement_recovery_conflict(format!(
+                "Replacement completion proof conflicts with state for task {}",
+                state.task_id
+            )));
+        }
+
+        state.mark_completed();
+        state.replacement_phase = ReplacementPhase::Verified;
+        state.last_update = proof.verified_at;
+        drop(state);
+        self.save_state_strict().await
+    }
+
+    pub(super) async fn migrate_legacy_replacement_completion_proof(disk: &DiskStore, task_id: &str) -> Result<bool> {
+        validate_resume_task_id(task_id)?;
+        let legacy_path = legacy_replacement_completion_proof_path(task_id);
+        let legacy_path_str = path_to_str(&legacy_path)?;
+        let legacy_bytes = match disk.read_all(RUSTFS_META_BUCKET, legacy_path_str).await {
+            Ok(bytes) => bytes,
+            Err(DiskError::FileNotFound) => return Ok(false),
+            Err(error) => {
+                return Err(Error::TaskExecutionFailed {
+                    message: format!("Failed to read legacy replacement completion proof: {error}"),
+                });
+            }
+        };
+        let legacy_proof: ReplacementCompletionProof = serde_json::from_slice(&legacy_bytes).map_err(|error| {
+            replacement_recovery_corruption(format!("Failed to deserialize legacy replacement completion proof: {error}"))
+        })?;
+        legacy_proof
+            .validate(task_id)
+            .map_err(|error| replacement_recovery_corruption(format!("Invalid legacy replacement completion proof: {error}")))?;
+
+        ensure_replacement_recovery_dir(disk)
+            .await
+            .map_err(|error| Error::TaskExecutionFailed {
+                message: format!("Failed to create replacement recovery directory: {error}"),
+            })?;
+        let path = replacement_completion_proof_path(task_id);
+        let path_str = path_to_str(&path)?;
+        for _ in 0..2 {
+            match disk.read_all(RUSTFS_META_BUCKET, path_str).await {
+                Ok(bytes) => {
+                    let proof: ReplacementCompletionProof =
+                        serde_json::from_slice(&bytes).map_err(|error| Error::TaskExecutionFailed {
+                            message: format!("Failed to deserialize replacement completion proof: {error}"),
+                        })?;
+                    proof.validate(task_id).map_err(|error| {
+                        replacement_recovery_corruption(format!("Invalid replacement completion proof: {error}"))
+                    })?;
+                    if proof != legacy_proof {
+                        return Err(replacement_recovery_conflict(format!(
+                            "Replacement completion proof conflicts with legacy proof for task {task_id}"
+                        )));
+                    }
+                    delete_resume_file(disk, &legacy_path).await?;
+                    return Ok(true);
+                }
+                Err(DiskError::FileNotFound) => {}
+                Err(error) => {
+                    return Err(Error::TaskExecutionFailed {
+                        message: format!("Failed to read replacement completion proof: {error}"),
+                    });
+                }
+            }
+
+            match super::super::storage_api::owner::EcstoreDiskAPI::compare_and_update_file(
+                disk.as_ref(),
+                RUSTFS_META_BUCKET,
+                path_str,
+                None,
+                Some(legacy_bytes.clone()),
+            )
+            .await
+            {
+                Ok(EcstoreConditionalFileUpdate::Updated) => {
+                    delete_resume_file(disk, &legacy_path).await?;
+                    return Ok(true);
+                }
+                Ok(EcstoreConditionalFileUpdate::Missing | EcstoreConditionalFileUpdate::Mismatch) => continue,
+                Err(error) => {
+                    return Err(Error::TaskExecutionFailed {
+                        message: format!("Failed to migrate replacement completion proof: {error}"),
+                    });
+                }
+            }
+        }
+
+        Err(Error::TaskExecutionFailed {
+            message: format!("Replacement completion proof changed while migrating task {task_id}"),
+        })
+    }
+
+    pub async fn abandon_replacement_intent(&self) -> Result<()> {
+        let mut state = self.state.write().await;
+        if matches!(state.replacement_phase, ReplacementPhase::Abandoned) {
+            return Ok(());
+        }
+        state.replacement_phase = ReplacementPhase::Abandoned;
+        state.last_update = SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_secs();
+        drop(state);
+        self.save_state_strict().await
+    }
+
+    pub async fn set_replacement_targets(&self, replacement_targets: Vec<String>) -> Result<()> {
+        {
+            let mut state = self.state.write().await;
+            state.replacement_targets = replacement_targets;
+        }
+        self.save_state().await
+    }
+
+    pub(super) async fn publish_new_replacement_intent(&self, expected: Option<EcstoreDiskBytes>) -> Result<()> {
+        let state = self.state.read().await.clone();
+        validate_resume_task_id(&state.task_id)?;
+        let state_data = EcstoreDiskBytes::from(serde_json::to_vec(&state).map_err(|error| Error::TaskExecutionFailed {
+            message: format!("Failed to serialize resume state: {error}"),
+        })?);
+        let path = self.state_file.path(&state.task_id);
+        let path = path_to_str(&path)?;
+
+        ensure_replacement_recovery_dir(&self.disk)
+            .await
+            .map_err(|error| Error::TaskExecutionFailed {
+                message: format!("Failed to create replacement recovery directory: {error}"),
+            })?;
+        match super::super::storage_api::owner::EcstoreDiskAPI::compare_and_update_file(
+            self.disk.as_ref(),
+            RUSTFS_META_BUCKET,
+            path,
+            expected,
+            Some(state_data),
+        )
+        .await
+        {
+            Ok(EcstoreConditionalFileUpdate::Updated) => Ok(()),
+            Ok(EcstoreConditionalFileUpdate::Missing | EcstoreConditionalFileUpdate::Mismatch) => {
+                Err(Error::TaskExecutionFailed {
+                    message: format!("Replacement intent changed before publication for task {}", state.task_id),
+                })
+            }
+            Err(error) => Err(Error::TaskExecutionFailed {
+                message: format!("Failed to save resume state: {error}"),
+            }),
+        }
+    }
+
+    async fn write_replacement_completion_proof(
+        &self,
+        state: &ResumeState,
+        verified_at: Option<u64>,
+    ) -> Result<ReplacementCompletionProof> {
+        ensure_replacement_recovery_dir(&self.disk)
+            .await
+            .map_err(|error| Error::TaskExecutionFailed {
+                message: format!("Failed to create replacement recovery directory: {error}"),
+            })?;
+        let path = replacement_completion_proof_path(&state.task_id);
+        let path_str = path_to_str(&path)?;
+        let proof = ReplacementCompletionProof::from_state(
+            state,
+            verified_at.unwrap_or_else(|| SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_secs()),
+        )?;
+        let proof_data = EcstoreDiskBytes::from(serde_json::to_vec(&proof).map_err(|e| Error::TaskExecutionFailed {
+            message: format!("Failed to serialize replacement completion proof: {e}"),
+        })?);
+        if let Some(error) = injected_replacement_proof_write_error(path_str) {
+            return Err(Error::TaskExecutionFailed {
+                message: format!("Failed to save replacement completion proof: {error}"),
+            });
+        }
+
+        // Publish through the disk CAS primitive: `write_all` can expose a
+        // partially written proof to a crash/restart reader. If a prior
+        // version left torn bytes behind, replace exactly the observed bytes;
+        // a concurrently published valid proof is never overwritten.
+        for _ in 0..2 {
+            let expected = match self.disk.read_all(RUSTFS_META_BUCKET, path_str).await {
+                Ok(existing) => match serde_json::from_slice::<ReplacementCompletionProof>(&existing) {
+                    Ok(existing_proof) => {
+                        existing_proof.validate(&state.task_id)?;
+                        if existing_proof.matches_state(state) {
+                            return Ok(existing_proof);
+                        }
+                        return Err(Error::TaskExecutionFailed {
+                            message: format!("Replacement completion proof does not match task {}", state.task_id),
+                        });
+                    }
+                    Err(_) => Some(existing),
+                },
+                Err(DiskError::FileNotFound) => None,
+                Err(error) => {
+                    return Err(Error::TaskExecutionFailed {
+                        message: format!("Failed to read replacement completion proof: {error}"),
+                    });
+                }
+            };
+
+            match super::super::storage_api::owner::EcstoreDiskAPI::compare_and_update_file(
+                self.disk.as_ref(),
+                RUSTFS_META_BUCKET,
+                path_str,
+                expected,
+                Some(proof_data.clone()),
+            )
+            .await
+            {
+                Ok(EcstoreConditionalFileUpdate::Updated) => return Ok(proof),
+                Ok(EcstoreConditionalFileUpdate::Missing | EcstoreConditionalFileUpdate::Mismatch) => continue,
+                Err(error) => {
+                    return Err(Error::TaskExecutionFailed {
+                        message: format!("Failed to save replacement completion proof: {error}"),
+                    });
+                }
+            }
+        }
+
+        Err(Error::TaskExecutionFailed {
+            message: format!("Replacement completion proof changed while publishing task {}", state.task_id),
+        })
+    }
+
+    pub(super) async fn write_replacement_intent_state(
+        &self,
+        path: &str,
+        state_data: EcstoreDiskBytes,
+    ) -> std::result::Result<(), DiskError> {
+        ensure_replacement_recovery_dir(&self.disk).await?;
+        for _ in 0..2 {
+            let expected = match self.disk.read_all(RUSTFS_META_BUCKET, path).await {
+                Ok(existing) => Some(existing),
+                Err(DiskError::FileNotFound) => None,
+                Err(error) => return Err(error),
+            };
+            match super::super::storage_api::owner::EcstoreDiskAPI::compare_and_update_file(
+                self.disk.as_ref(),
+                RUSTFS_META_BUCKET,
+                path,
+                expected,
+                Some(state_data.clone()),
+            )
+            .await
+            {
+                Ok(EcstoreConditionalFileUpdate::Updated) => return Ok(()),
+                Ok(EcstoreConditionalFileUpdate::Missing | EcstoreConditionalFileUpdate::Mismatch) => continue,
+                Err(error) => return Err(error),
+            }
+        }
+        Err(DiskError::other("replacement intent changed while publishing"))
+    }
+}

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

@@ -0,0 +1,311 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use crate::{Error, Result};
+use std::collections::HashSet;
+use std::time::{SystemTime, UNIX_EPOCH};
+use tracing::{debug, warn};
+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,
+    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,
+};
+
+/// resume utils
+pub struct ResumeUtils;
+
+impl ResumeUtils {
+    /// generate unique task id
+    pub fn generate_task_id() -> String {
+        Uuid::new_v4().to_string()
+    }
+
+    /// check if task can be resumed
+    pub async fn can_resume_task(disk: &DiskStore, task_id: &str) -> bool {
+        ResumeManager::has_resume_state(disk, task_id).await
+    }
+
+    /// get all resumable task ids
+    pub async fn get_resumable_tasks(disk: &DiskStore) -> Result<Vec<String>> {
+        // List all files in the buckets metadata directory
+        let entries = match disk.list_dir("", RUSTFS_META_BUCKET, BUCKET_META_PREFIX, -1).await {
+            Ok(entries) => entries,
+            Err(e) => {
+                debug!(
+                    target: "rustfs::heal::resume",
+                    event = EVENT_HEAL_RESUME_STATE,
+                    component = LOG_COMPONENT_HEAL,
+                    subsystem = LOG_SUBSYSTEM_RESUME,
+                    state = "list_failed",
+                    error = %e,
+                    "Heal resume state listing failed"
+                );
+                return Ok(Vec::new());
+            }
+        };
+
+        let mut task_ids = Vec::new();
+
+        // Filter files that end with ahm_resume_state.json and extract task IDs
+        for entry in entries {
+            if entry.ends_with(&format!("_{RESUME_STATE_FILE}")) {
+                // 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()
+                {
+                    task_ids.push(task_id.to_string());
+                }
+            }
+        }
+
+        debug!(
+            target: "rustfs::heal::resume",
+            event = EVENT_HEAL_RESUME_STATE,
+            component = LOG_COMPONENT_HEAL,
+            subsystem = LOG_SUBSYSTEM_RESUME,
+            task_count = task_ids.len(),
+            state = "listed",
+            "Heal resume states listed"
+        );
+        Ok(task_ids)
+    }
+
+    /// Return replacement intent task IDs from the dedicated recovery
+    /// directory. Periodic recovery must never enumerate the ordinary resume
+    /// directory, whose cardinality is unrelated to replacement work.
+    pub async fn get_replacement_intent_tasks(disk: &DiskStore) -> Result<Vec<String>> {
+        let entries = Self::replacement_recovery_entries(disk).await?;
+        let suffix = format!("_{REPLACEMENT_INTENT_FILE}");
+        let mut task_ids = HashSet::new();
+
+        for entry in entries {
+            if let Some(task_id) = entry.strip_suffix(&suffix)
+                && validate_resume_task_id(task_id).is_ok()
+            {
+                task_ids.insert(task_id.to_string());
+                continue;
+            }
+        }
+
+        let mut task_ids = task_ids.into_iter().collect::<Vec<_>>();
+        task_ids.sort_unstable();
+        Ok(task_ids)
+    }
+
+    async fn replacement_recovery_entries(disk: &DiskStore) -> Result<Vec<String>> {
+        let recovery_dir = replacement_recovery_dir();
+        let recovery_dir = path_to_str(&recovery_dir)?;
+        match disk.list_dir("", RUSTFS_META_BUCKET, recovery_dir, -1).await {
+            Ok(entries) => Ok(entries),
+            Err(DiskError::FileNotFound) => Ok(Vec::new()),
+            Err(error @ DiskError::UnformattedDisk) => Err(error.into()),
+            Err(error) => Err(Error::TaskExecutionFailed {
+                message: format!("Failed to list replacement recovery records: {error}"),
+            }),
+        }
+    }
+
+    /// Migrate flat replacement artifacts from earlier builds exactly once at
+    /// manager startup. The normal scanner only uses the dedicated directory;
+    /// ordinary resume JSON is never read on its periodic path.
+    pub async fn migrate_legacy_replacement_records(disk: &DiskStore) -> Result<()> {
+        let entries = disk
+            .list_dir("", RUSTFS_META_BUCKET, BUCKET_META_PREFIX, -1)
+            .await
+            .map_err(|error| Error::TaskExecutionFailed {
+                message: format!("Failed to list legacy replacement records: {error}"),
+            })?;
+        let ordinary_suffix = format!("_{RESUME_STATE_FILE}");
+        let intent_suffix = format!("_{REPLACEMENT_INTENT_FILE}");
+        let proof_suffix = format!("_{REPLACEMENT_COMPLETION_PROOF_FILE}");
+        let mut ordinary_task_ids = HashSet::new();
+        let mut intent_task_ids = HashSet::new();
+        let mut proof_task_ids = HashSet::new();
+
+        for entry in entries {
+            if let Some(task_id) = entry.strip_suffix(&intent_suffix)
+                && validate_resume_task_id(task_id).is_ok()
+            {
+                intent_task_ids.insert(task_id.to_string());
+                continue;
+            }
+            if let Some(task_id) = entry.strip_suffix(&ordinary_suffix)
+                && validate_resume_task_id(task_id).is_ok()
+            {
+                ordinary_task_ids.insert(task_id.to_string());
+                continue;
+            }
+            if let Some(task_id) = entry.strip_suffix(&proof_suffix)
+                && validate_resume_task_id(task_id).is_ok()
+            {
+                proof_task_ids.insert(task_id.to_string());
+            }
+        }
+
+        let mut state_task_ids = intent_task_ids.into_iter().collect::<Vec<_>>();
+        state_task_ids.extend(ordinary_task_ids);
+        state_task_ids.sort_unstable();
+        state_task_ids.dedup();
+        for task_id in state_task_ids {
+            let has_flat_intent = ResumeManager::has_state_file(disk, &task_id, ResumeStateFile::LegacyReplacementIntent).await;
+            if !has_flat_intent {
+                let manager = ResumeManager::load_from_disk(disk.clone(), &task_id).await.map_err(|error| {
+                    replacement_recovery_corruption_for_state_load(
+                        format!("Failed to load legacy replacement recovery candidate {task_id}"),
+                        error,
+                    )
+                })?;
+                if !is_replacement_intent(&manager.get_state().await) {
+                    continue;
+                }
+            }
+            ResumeManager::load_replacement_intent(disk.clone(), &task_id).await?;
+        }
+
+        for task_id in proof_task_ids {
+            ResumeManager::migrate_legacy_replacement_completion_proof(disk, &task_id).await?;
+        }
+        Ok(())
+    }
+
+    /// Return all durable replacement states and completion proofs stored on
+    /// one survivor disk. Unlike the legacy resumable-task helper, listing
+    /// failures are returned to the caller so an observability surface cannot
+    /// silently turn an unreadable durable record into a green result.
+    pub async fn get_replacement_recovery_records(disk: &DiskStore) -> Result<Vec<ReplacementRecoveryRecord>> {
+        let entries = Self::replacement_recovery_entries(disk).await?;
+        let proof_suffix = format!("_{REPLACEMENT_COMPLETION_PROOF_FILE}");
+        let mut records = Vec::new();
+        let mut intent_task_ids = HashSet::new();
+
+        for task_id in Self::get_replacement_intent_tasks(disk).await? {
+            let state = ResumeManager::load_replacement_intent(disk.clone(), &task_id)
+                .await?
+                .get_state()
+                .await;
+            intent_task_ids.insert(task_id.clone());
+            records.push(ReplacementRecoveryRecord::from_state(state).unwrap_or_else(|| {
+                ReplacementRecoveryRecord::unknown(
+                    task_id,
+                    "isolated replacement intent violates its generation or target identity binding",
+                )
+            }));
+        }
+
+        for entry in entries {
+            let Some(task_id) = entry.strip_suffix(&proof_suffix) else {
+                continue;
+            };
+            if validate_resume_task_id(task_id).is_err() {
+                continue;
+            }
+            if intent_task_ids.contains(task_id) {
+                continue;
+            }
+            let proof = ResumeManager::load_replacement_completion_proof(disk.clone(), task_id).await?;
+            records.push(ReplacementRecoveryRecord::from_completion_proof(&proof));
+        }
+
+        records.sort_by(|left, right| left.task_id.cmp(&right.task_id).then(left.state.cmp(&right.state)));
+        Ok(records)
+    }
+
+    /// cleanup expired resume states
+    pub async fn cleanup_expired_states(disk: &DiskStore, max_age_hours: u64) -> Result<()> {
+        let task_ids = Self::get_resumable_tasks(disk).await?;
+        let current_time = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_secs();
+
+        for task_id in task_ids {
+            if let Ok(resume_manager) = ResumeManager::load_from_disk(disk.clone(), &task_id).await {
+                let state = resume_manager.get_state().await;
+                let age_hours = current_time.saturating_sub(state.last_update) / 3600;
+
+                if !state.completed && matches!(state.replacement_phase, ReplacementPhase::Intent | ReplacementPhase::Rebuilding)
+                {
+                    continue;
+                }
+                if state.completed
+                    && matches!(state.replacement_phase, ReplacementPhase::Verified | ReplacementPhase::CleanupPending)
+                {
+                    continue;
+                }
+
+                if age_hours > max_age_hours {
+                    debug!(
+                        target: "rustfs::heal::resume",
+                        event = EVENT_HEAL_RESUME_STATE,
+                        component = LOG_COMPONENT_HEAL,
+                        subsystem = LOG_SUBSYSTEM_RESUME,
+                        task_id,
+                        age_hours,
+                        state = "expired_cleanup_started",
+                        "Heal resume cleanup started"
+                    );
+                    if let Err(e) = resume_manager.cleanup().await {
+                        warn!(
+                            target: "rustfs::heal::resume",
+                            event = EVENT_HEAL_RESUME_STATE,
+                            component = LOG_COMPONENT_HEAL,
+                            subsystem = LOG_SUBSYSTEM_RESUME,
+                            task_id,
+                            age_hours,
+                            state = "expired_cleanup_failed",
+                            error = %e,
+                            "Heal resume state cleanup failed"
+                        );
+                    }
+                }
+            }
+        }
+
+        for task_id in Self::get_replacement_intent_tasks(disk).await? {
+            if let Ok(resume_manager) = ResumeManager::load_replacement_intent(disk.clone(), &task_id).await {
+                let state = resume_manager.get_state().await;
+                let age_hours = current_time.saturating_sub(state.last_update) / 3600;
+
+                if !state.completed && matches!(state.replacement_phase, ReplacementPhase::Intent | ReplacementPhase::Rebuilding)
+                {
+                    continue;
+                }
+                if state.completed
+                    && matches!(state.replacement_phase, ReplacementPhase::Verified | ReplacementPhase::CleanupPending)
+                {
+                    continue;
+                }
+
+                if age_hours > max_age_hours
+                    && let Err(e) = resume_manager.cleanup().await
+                {
+                    warn!(
+                        target: "rustfs::heal::resume",
+                        event = EVENT_HEAL_RESUME_STATE,
+                        component = LOG_COMPONENT_HEAL,
+                        subsystem = LOG_SUBSYSTEM_RESUME,
+                        task_id,
+                        age_hours,
+                        state = "expired_cleanup_failed",
+                        error = %e,
+                        "Replacement intent cleanup failed"
+                    );
+                }
+            }
+        }
+
+        Ok(())
+    }
+}