diff --git a/Cargo.lock b/Cargo.lock
index f97711209..4fec6d622 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -9261,6 +9261,7 @@ name = "rustfs-heal"
 version = "1.0.0-beta.8"
 dependencies = [
  "async-trait",
+ "base64 0.22.1",
  "futures",
  "http 1.4.2",
  "metrics",
diff --git a/crates/heal/Cargo.toml b/crates/heal/Cargo.toml
index a0ad388c0..b60dfa118 100644
--- a/crates/heal/Cargo.toml
+++ b/crates/heal/Cargo.toml
@@ -44,6 +44,7 @@ uuid = { workspace = true, features = ["v4", "serde"] }
 async-trait = { workspace = true }
 futures = { workspace = true }
 metrics = { workspace = true }
+base64 = { workspace = true }
 
 [dev-dependencies]
 serde_json = { workspace = true }
diff --git a/crates/heal/src/heal/channel.rs b/crates/heal/src/heal/channel.rs
index e2ca51d9b..9be3813ef 100644
--- a/crates/heal/src/heal/channel.rs
+++ b/crates/heal/src/heal/channel.rs
@@ -606,7 +606,11 @@ mod tests {
         ) -> crate::Result<(rustfs_madmin::heal_commands::HealResultItem, Option<crate::Error>)> {
             Ok((rustfs_madmin::heal_commands::HealResultItem::default(), None))
         }
-        async fn list_objects_for_heal(&self, _bucket: &str, _prefix: &str) -> crate::Result<Vec<String>> {
+        async fn list_objects_for_heal(
+            &self,
+            _bucket: &str,
+            _prefix: &str,
+        ) -> crate::Result<Vec<crate::heal::storage::HealListItem>> {
             Ok(vec![])
         }
         async fn list_objects_for_heal_page(
@@ -614,7 +618,7 @@ mod tests {
             _bucket: &str,
             _prefix: &str,
             _continuation_token: Option<&str>,
-        ) -> crate::Result<(Vec<String>, Option<String>, bool)> {
+        ) -> crate::Result<(Vec<crate::heal::storage::HealListItem>, Option<String>, bool)> {
             Ok((vec![], None, false))
         }
         async fn get_disk_for_resume(&self, _set_disk_id: &str) -> crate::Result<DiskStore> {
diff --git a/crates/heal/src/heal/erasure_healer.rs b/crates/heal/src/heal/erasure_healer.rs
index e69576ba0..f09d8212e 100644
--- a/crates/heal/src/heal/erasure_healer.rs
+++ b/crates/heal/src/heal/erasure_healer.rs
@@ -14,12 +14,12 @@
 
 use crate::heal::{
     progress::HealProgress,
-    resume::{CheckpointManager, ResumeManager, ResumeUtils},
+    resume::{CheckpointManager, ResumeManager, ResumeUtils, compose_key},
     storage::{HealStorageAPI, next_heal_listing_token},
     task::is_missing_object_dir_heal_result,
 };
 use crate::{Error, Result};
-use futures::{StreamExt, future::join_all, stream::FuturesUnordered};
+use futures::{StreamExt, stream::FuturesUnordered};
 use metrics::gauge;
 use rustfs_common::heal_channel::{HealOpts, HealRequestSource, HealScanMode};
 use std::sync::{
@@ -27,7 +27,7 @@ use std::sync::{
     atomic::{AtomicUsize, Ordering},
 };
 use tokio::sync::{RwLock, Semaphore};
-use tracing::{debug, error, info, warn};
+use tracing::{debug, error, warn};
 
 use super::{DiskStore, EcstoreError};
 
@@ -447,6 +447,12 @@ impl ErasureSetHealer {
         // (backlog#855 / #799 B6).
         if failed_objects > 0 {
             if resume_manager.schedule_retry().await? {
+                // Both persistence layers must be reset together: schedule_retry
+                // rewinds the resume state (cursor + counters), and the
+                // checkpoint's per-version dedup sets + position must be cleared
+                // in lockstep, or the retry would skip the very versions it is
+                // meant to re-heal.
+                checkpoint_manager.reset_for_retry().await?;
                 // Retry budget remains: state has been reset for a full re-scan.
                 // Return Err so `heal_erasure_set` preserves (does not clean up)
                 // the resume/checkpoint state and the caller keeps the healing
@@ -549,15 +555,23 @@ impl ErasureSetHealer {
             }
         };
 
-        // 2. process objects with pagination to avoid loading all objects into memory
-        let mut continuation_token: Option<String> = None;
+        // 2. process object VERSIONS with pagination to avoid loading everything into memory.
+        //    The continuation token is the authoritative opaque (marker, version_marker)
+        //    cursor seeded from the resume state, so a resume continues exactly where the
+        //    previous pass stopped — including mid-object across version pages.
+        let mut continuation_token: Option<String> = resume_manager.resume_cursor().await;
         let mut global_obj_idx = 0usize;
+        // Anti-loop guard: the last (name, version_id) actually observed on the
+        // previous page. Comparing decoded item identities (not raw tokens)
+        // detects a non-advancing cursor even though next_marker embeds a
+        // transient cache id that changes between identical pages.
+        let mut previous_page_last: Option<(String, Option<String>)> = None;
         let page_concurrency_limit =
             Self::effective_heal_page_object_concurrency_for_source(self.source, self.heal_opts.scan_mode);
         let in_flight = Arc::new(AtomicUsize::new(0));
 
         loop {
-            // Get one page of objects
+            // Get one page of object versions
             let (objects, next_token, is_truncated) = self
                 .storage
                 .list_objects_for_heal_page(bucket, "", continuation_token.as_deref())
@@ -567,30 +581,33 @@ impl ErasureSetHealer {
             let semaphore = Arc::new(Semaphore::new(page_concurrency_limit));
             let mut page_tasks = FuturesUnordered::new();
 
-            for object in objects {
-                let object_idx = global_obj_idx;
+            // Capture the last version identity of this page for the anti-loop guard.
+            let page_last = objects.last().map(|item| (item.name.clone(), item.version_id.clone()));
+
+            for item in objects {
+                // current_object_index is now only a progress metric; the cursor
+                // drives resume, so we no longer skip by position.
                 global_obj_idx += 1;
 
-                if object_idx < *current_object_index {
-                    continue;
-                }
-
-                if checkpoint.processed_objects.contains(&object) || checkpoint.skipped_objects.contains(&object) {
+                // Per-version dedup identity — the single canonical key.
+                let key = compose_key(&item.name, item.version_id.as_deref());
+                if checkpoint.processed_objects.contains(&key) || checkpoint.skipped_objects.contains(&key) {
                     continue;
                 }
 
                 resume_manager
-                    .set_current_item(Some(bucket.to_string()), Some(object.clone()))
+                    .set_current_item(Some(bucket.to_string()), Some(item.name.clone()))
                     .await?;
 
                 let storage = self.storage.clone();
                 let bucket_name = bucket.to_string();
-                let object_name = object.clone();
+                let object_name = item.name.clone();
+                let version_id = item.version_id.clone();
+                let dedup_key = key;
                 let cancel_token = self.cancel_token.clone();
                 let in_flight = in_flight.clone();
                 let set_label = set_disk_id.to_string();
                 let heal_opts = self.heal_opts;
-                let deep_scan = matches!(heal_opts.scan_mode, HealScanMode::Deep);
                 let semaphore = semaphore.clone();
 
                 page_tasks.push(async move {
@@ -602,7 +619,7 @@ impl ErasureSetHealer {
 
                     let _permit = match permit {
                         Ok(permit) => permit,
-                        Err(err) => return (object_name, Err(err)),
+                        Err(err) => return (dedup_key, object_name, version_id, Err(err)),
                     };
 
                     let current_in_flight = in_flight.fetch_add(1, Ordering::SeqCst) + 1;
@@ -612,11 +629,20 @@ impl ErasureSetHealer {
                     )
                     .set(current_in_flight as f64);
 
+                    // Always go through heal_object. Genuine absence flows through
+                    // heal_object -> FileVersionNotFound/FileNotFound ->
+                    // classify_heal_object_error -> Absent, so gone versions are
+                    // recorded as skipped-ok rather than failed. The delete-marker
+                    // vs data path is chosen internally in ops/heal.rs.
                     let result = if cancel_token.is_cancelled() {
                         Err(Error::TaskCancelled)
-                    } else if deep_scan {
-                        match storage.heal_object(&bucket_name, &object_name, None, &heal_opts).await {
+                    } else {
+                        match storage
+                            .heal_object(&bucket_name, &object_name, version_id.as_deref(), &heal_opts)
+                            .await
+                        {
                             Ok((_result, None)) => Ok(true),
+                            Ok((_, Some(err))) if is_missing_object_dir_heal_result(&object_name, &err) => Ok(false),
                             Ok((_, Some(err))) | Err(err) => match Self::classify_heal_object_error(&err) {
                                 HealObjectOutcome::Absent => Ok(false),
                                 HealObjectOutcome::Transient => Err(Error::transient_skip(format!(
@@ -625,51 +651,6 @@ impl ErasureSetHealer {
                                 HealObjectOutcome::Failed => Err(err),
                             },
                         }
-                    } else {
-                        let object_exists = match storage.object_exists(&bucket_name, &object_name).await {
-                            Ok(exists) => exists,
-                            Err(err @ Error::TransientSkip { .. }) => {
-                                let current = in_flight.fetch_sub(1, Ordering::SeqCst) - 1;
-                                gauge!(
-                                    "rustfs_heal_page_concurrency_current",
-                                    "set" => set_label.clone()
-                                )
-                                .set(current as f64);
-                                return (object_name, Err(err));
-                            }
-                            Err(err) => {
-                                let object_name_for_error = object_name.clone();
-                                let current = in_flight.fetch_sub(1, Ordering::SeqCst) - 1;
-                                gauge!(
-                                    "rustfs_heal_page_concurrency_current",
-                                    "set" => set_label.clone()
-                                )
-                                .set(current as f64);
-                                return (
-                                    object_name,
-                                    Err(Error::other(format!(
-                                        "Failed to check existence of {}/{}: {}",
-                                        bucket_name, object_name_for_error, err
-                                    ))),
-                                );
-                            }
-                        };
-
-                        if !object_exists {
-                            Ok(false)
-                        } else {
-                            match storage.heal_object(&bucket_name, &object_name, None, &heal_opts).await {
-                                Ok((_result, None)) => Ok(true),
-                                Ok((_, Some(err))) if is_missing_object_dir_heal_result(&object_name, &err) => Ok(false),
-                                Ok((_, Some(err))) | Err(err) => match Self::classify_heal_object_error(&err) {
-                                    HealObjectOutcome::Absent => Ok(false),
-                                    HealObjectOutcome::Transient => Err(Error::transient_skip(format!(
-                                        "Skipped heal for {bucket_name}/{object_name} due to transient error: {err}"
-                                    ))),
-                                    HealObjectOutcome::Failed => Err(err),
-                                },
-                            }
-                        }
                     };
 
                     let current = in_flight.fetch_sub(1, Ordering::SeqCst) - 1;
@@ -679,16 +660,16 @@ impl ErasureSetHealer {
                     )
                     .set(current as f64);
 
-                    (object_name, result)
+                    (dedup_key, object_name, version_id, result)
                 });
             }
 
             let mut completed_in_page = 0usize;
-            while let Some((object, result)) = page_tasks.next().await {
+            while let Some((key, object, version_id, result)) = page_tasks.next().await {
                 match result {
                     Ok(true) => {
                         *successful_objects += 1;
-                        checkpoint_manager.add_processed_object(object.clone()).await?;
+                        checkpoint_manager.add_processed_object(key).await?;
                         debug!(
                             target: "rustfs::heal::erasure_healer",
                             event = EVENT_HEAL_ERASURE_OBJECT_STATE,
@@ -697,12 +678,13 @@ impl ErasureSetHealer {
                             set_disk_id,
                             bucket,
                             object = %object,
+                            version_id = ?version_id,
                             state = "healed",
                             "Erasure set object healed"
                         );
                     }
                     Ok(false) => {
-                        checkpoint_manager.add_processed_object(object.clone()).await?;
+                        checkpoint_manager.add_processed_object(key).await?;
                         *successful_objects += 1;
                         debug!(
                             target: "rustfs::heal::erasure_healer",
@@ -712,6 +694,7 @@ impl ErasureSetHealer {
                             set_disk_id,
                             bucket,
                             object = %object,
+                            version_id = ?version_id,
                             state = "missing_treated_as_ok",
                             "Erasure set missing object treated as ok"
                         );
@@ -726,7 +709,7 @@ impl ErasureSetHealer {
                     }
                     Err(Error::TransientSkip { message }) => {
                         *skipped_objects += 1;
-                        checkpoint_manager.add_skipped_object(object.clone()).await?;
+                        checkpoint_manager.add_skipped_object(key).await?;
                         warn!(
                             target: "rustfs::heal::erasure_healer",
                             event = EVENT_HEAL_ERASURE_OBJECT_STATE,
@@ -735,6 +718,7 @@ impl ErasureSetHealer {
                             set_disk_id,
                             bucket,
                             object = %object,
+                            version_id = ?version_id,
                             state = "transient_skip",
                             error = %message,
                             "Erasure set object heal skipped due to transient error"
@@ -742,7 +726,7 @@ impl ErasureSetHealer {
                     }
                     Err(err) => {
                         *failed_objects += 1;
-                        checkpoint_manager.add_failed_object(object.clone()).await?;
+                        checkpoint_manager.add_failed_object(key).await?;
                         warn!(
                             target: "rustfs::heal::erasure_healer",
                             event = EVENT_HEAL_ERASURE_OBJECT_STATE,
@@ -751,6 +735,7 @@ impl ErasureSetHealer {
                             set_disk_id,
                             bucket,
                             object = %object,
+                            version_id = ?version_id,
                             state = "failed",
                             error = %err,
                             "Erasure set object heal failed"
@@ -767,6 +752,12 @@ impl ErasureSetHealer {
             }
 
             *current_object_index = global_obj_idx;
+
+            // Persist the authoritative cursor FIRST (points at the next page
+            // boundary), then prune the per-version dedup sets. Both are
+            // idempotent under crash: heal_object re-heals safely.
+            let next_cursor = if is_truncated { next_token.clone() } else { None };
+            resume_manager.set_resume_cursor(next_cursor.clone()).await?;
             checkpoint_manager.complete_page(bucket_index, *current_object_index).await?;
             gauge!(
                 "rustfs_heal_page_concurrency_current",
@@ -779,7 +770,20 @@ impl ErasureSetHealer {
                 break;
             }
 
+            // Anti-loop guard: if the backend keeps reporting truncation but the
+            // last version identity did not advance, we would spin forever.
+            if page_last.is_some() && page_last == previous_page_last {
+                return Err(Error::other(format!(
+                    "Erasure set heal listing for bucket {bucket} is not advancing (repeated last version {page_last:?}); aborting to avoid an infinite loop"
+                )));
+            }
+            previous_page_last = page_last;
+
             continuation_token = next_heal_listing_token(bucket, "", next_token, is_truncated)?;
+            if continuation_token.is_none() {
+                // Truncated but no continuation token: treat as end of listing.
+                break;
+            }
         }
 
         Ok(())
@@ -794,264 +798,6 @@ impl ErasureSetHealer {
         progress.bytes_processed = 0; // set to 0 for now, can be extended later
         progress.set_current_object(state.current_object.clone());
     }
-
-    /// heal all buckets concurrently
-    #[allow(dead_code)]
-    async fn heal_buckets_concurrently(&self, buckets: &[String]) -> Vec<Result<()>> {
-        // use semaphore to control concurrency, avoid too many concurrent healings
-        let semaphore = Arc::new(tokio::sync::Semaphore::new(4)); // max 4 concurrent healings
-
-        let heal_futures = buckets.iter().map(|bucket| {
-            let bucket = bucket.clone();
-            let storage = self.storage.clone();
-            let progress = self.progress.clone();
-            let semaphore = semaphore.clone();
-            let cancel_token = self.cancel_token.clone();
-
-            async move {
-                let _permit = semaphore
-                    .acquire()
-                    .await
-                    .map_err(|e| Error::other(format!("Failed to acquire semaphore for bucket heal: {e}")))?;
-
-                if cancel_token.is_cancelled() {
-                    return Err(Error::TaskCancelled);
-                }
-
-                Self::heal_single_bucket(&storage, &bucket, &progress).await
-            }
-        });
-
-        // use join_all to process concurrently
-        join_all(heal_futures).await
-    }
-
-    /// heal single bucket
-    #[allow(dead_code)]
-    async fn heal_single_bucket(
-        storage: &Arc<dyn HealStorageAPI>,
-        bucket: &str,
-        progress: &Arc<RwLock<HealProgress>>,
-    ) -> Result<()> {
-        debug!(
-            target: "rustfs::heal::erasure_healer",
-            event = EVENT_HEAL_ERASURE_BUCKET_STATE,
-            component = LOG_COMPONENT_HEAL,
-            subsystem = LOG_SUBSYSTEM_ERASURE_HEALER,
-            bucket,
-            state = "started",
-            "Erasure set bucket started"
-        );
-
-        // 1. get bucket info
-        let _bucket_info = match storage.get_bucket_info(bucket).await? {
-            Some(info) => info,
-            None => {
-                warn!(
-                    target: "rustfs::heal::erasure_healer",
-                    event = EVENT_HEAL_ERASURE_BUCKET_STATE,
-                    component = LOG_COMPONENT_HEAL,
-                    subsystem = LOG_SUBSYSTEM_ERASURE_HEALER,
-                    bucket,
-                    state = "missing",
-                    "Erasure set bucket heal skipped because bucket is missing"
-                );
-                return Ok(());
-            }
-        };
-
-        // 2. process objects with pagination to avoid loading all objects into memory
-        let mut continuation_token: Option<String> = None;
-        let mut total_scanned = 0u64;
-        let mut total_success = 0u64;
-        let mut total_failed = 0u64;
-
-        let heal_opts = HealOpts {
-            scan_mode: HealScanMode::Normal,
-            remove: true,   // remove corrupted data
-            recreate: true, // recreate missing data
-            ..Default::default()
-        };
-
-        loop {
-            // Get one page of objects
-            let (objects, next_token, is_truncated) = storage
-                .list_objects_for_heal_page(bucket, "", continuation_token.as_deref())
-                .await?;
-
-            let page_count = objects.len() as u64;
-            total_scanned += page_count;
-
-            // 3. update progress
-            {
-                let mut p = progress.write().await;
-                p.objects_scanned = total_scanned;
-            }
-
-            // 4. heal objects concurrently for this page
-            let object_results = Self::heal_objects_concurrently(storage, bucket, &objects, &heal_opts, progress).await;
-
-            // 5. count results for this page
-            let (success_count, failure_count) =
-                object_results
-                    .into_iter()
-                    .fold((0, 0), |(success, failure), result| match result {
-                        Ok(_) => (success + 1, failure),
-                        Err(_) => (success, failure + 1),
-                    });
-
-            total_success += success_count;
-            total_failed += failure_count;
-
-            // 6. update progress
-            {
-                let mut p = progress.write().await;
-                p.objects_healed = total_success;
-                p.objects_failed = total_failed;
-                p.set_current_object(Some(format!("processing bucket: {bucket} (page)")));
-            }
-
-            // Check if there are more pages
-            if !is_truncated {
-                break;
-            }
-
-            continuation_token = next_heal_listing_token(bucket, "", next_token, is_truncated)?;
-        }
-
-        // 7. final progress update
-        {
-            let mut p = progress.write().await;
-            p.set_current_object(Some(format!("completed bucket: {bucket}")));
-        }
-
-        debug!(
-            target: "rustfs::heal::erasure_healer",
-            event = EVENT_HEAL_ERASURE_BUCKET_STATE,
-            component = LOG_COMPONENT_HEAL,
-            subsystem = LOG_SUBSYSTEM_ERASURE_HEALER,
-            bucket,
-            total_success,
-            total_failed,
-            total_scanned,
-            state = "completed",
-            "Erasure set bucket completed"
-        );
-
-        Ok(())
-    }
-
-    /// heal objects concurrently
-    #[allow(dead_code)]
-    async fn heal_objects_concurrently(
-        storage: &Arc<dyn HealStorageAPI>,
-        bucket: &str,
-        objects: &[String],
-        heal_opts: &HealOpts,
-        _progress: &Arc<RwLock<HealProgress>>,
-    ) -> Vec<Result<()>> {
-        // use semaphore to control object healing concurrency
-        let semaphore = Arc::new(tokio::sync::Semaphore::new(8)); // max 8 concurrent object healings
-
-        let heal_futures = objects.iter().map(|object| {
-            let object = object.clone();
-            let bucket = bucket.to_string();
-            let storage = storage.clone();
-            let heal_opts = *heal_opts;
-            let semaphore = semaphore.clone();
-
-            async move {
-                let _permit = semaphore
-                    .acquire()
-                    .await
-                    .map_err(|e| Error::other(format!("Failed to acquire semaphore for object heal: {e}")))?;
-
-                match storage.heal_object(&bucket, &object, None, &heal_opts).await {
-                    Ok((_result, None)) => {
-                        debug!(
-                            target: "rustfs::heal::erasure_healer",
-                            event = EVENT_HEAL_ERASURE_OBJECT_STATE,
-                            component = LOG_COMPONENT_HEAL,
-                            subsystem = LOG_SUBSYSTEM_ERASURE_HEALER,
-                            bucket,
-                            object = %object,
-                            state = "healed",
-                            "Erasure set object healed"
-                        );
-                        Ok(())
-                    }
-                    Ok((_, Some(err))) => {
-                        warn!(
-                            target: "rustfs::heal::erasure_healer",
-                            event = EVENT_HEAL_ERASURE_OBJECT_STATE,
-                            component = LOG_COMPONENT_HEAL,
-                            subsystem = LOG_SUBSYSTEM_ERASURE_HEALER,
-                            bucket,
-                            object = %object,
-                            state = "failed",
-                            error = %err,
-                            "Erasure set object heal failed"
-                        );
-                        Err(Error::other(err))
-                    }
-                    Err(err) => {
-                        warn!(
-                            target: "rustfs::heal::erasure_healer",
-                            event = EVENT_HEAL_ERASURE_OBJECT_STATE,
-                            component = LOG_COMPONENT_HEAL,
-                            subsystem = LOG_SUBSYSTEM_ERASURE_HEALER,
-                            bucket,
-                            object = %object,
-                            state = "failed",
-                            error = %err,
-                            "Erasure set object heal failed"
-                        );
-                        Err(err)
-                    }
-                }
-            }
-        });
-
-        join_all(heal_futures).await
-    }
-
-    /// process results
-    #[allow(dead_code)]
-    async fn process_results(&self, results: Vec<Result<()>>) -> Result<()> {
-        let (success_count, failure_count): (usize, usize) =
-            results.into_iter().fold((0, 0), |(success, failure), result| match result {
-                Ok(_) => (success + 1, failure),
-                Err(_) => (success, failure + 1),
-            });
-
-        let total = success_count + failure_count;
-
-        info!(
-            target: "rustfs::heal::erasure_healer",
-            event = EVENT_HEAL_ERASURE_RESUME_STATE,
-            component = LOG_COMPONENT_HEAL,
-            subsystem = LOG_SUBSYSTEM_ERASURE_HEALER,
-            success_count,
-            total,
-            state = "summary",
-            "Erasure set summary recorded"
-        );
-
-        if failure_count > 0 {
-            warn!(
-                target: "rustfs::heal::erasure_healer",
-                event = EVENT_HEAL_ERASURE_RESUME_STATE,
-                component = LOG_COMPONENT_HEAL,
-                subsystem = LOG_SUBSYSTEM_ERASURE_HEALER,
-                failure_count,
-                state = "summary_failed",
-                "Erasure set heal summary recorded failures"
-            );
-            return Err(Error::other(format!("{failure_count} buckets failed to heal")));
-        }
-
-        Ok(())
-    }
 }
 
 #[cfg(test)]
@@ -1190,3 +936,446 @@ mod tests {
         ));
     }
 }
+
+#[cfg(test)]
+mod resume_loop_tests {
+    //! Loop-level tests driving the private concurrent resume loop
+    //! (`heal_bucket_with_resume`) against a controllable fake `HealStorageAPI`
+    //! that emits programmable multi-version pages. These exercise the real loop
+    //! logic (cursor seeding, per-version dedup, anti-loop guard, absence
+    //! handling) — not merely a mock's own output.
+    use super::ErasureSetHealer;
+    use crate::heal::progress::HealProgress;
+    use crate::heal::resume::{CheckpointManager, ResumeManager, compose_key};
+    use crate::heal::storage::{DiskStatus, HealListItem, HealObjectInfo, HealStorageAPI};
+    use crate::heal::storage_api::status::BucketInfo;
+    use crate::heal::{
+        BUCKET_META_PREFIX, DiskOption, DiskStore, EcstoreError, Endpoint, HealDiskExt as _, RUSTFS_META_BUCKET, new_disk,
+    };
+    use crate::{Error, Result};
+    use rustfs_common::heal_channel::{HealOpts, HealRequestSource};
+    use rustfs_madmin::heal_commands::HealResultItem;
+    use std::collections::HashMap;
+    use std::sync::{Arc, Mutex};
+    use tempfile::TempDir;
+    use tokio::sync::RwLock;
+    use tokio_util::sync::CancellationToken;
+
+    fn item(name: &str, version: Option<&str>, delete_marker: bool) -> HealListItem {
+        HealListItem {
+            name: name.to_string(),
+            version_id: version.map(str::to_string),
+            is_delete_marker: delete_marker,
+        }
+    }
+
+    #[derive(Clone)]
+    struct Page {
+        items: Vec<HealListItem>,
+        next: Option<String>,
+        truncated: bool,
+    }
+
+    #[derive(Clone)]
+    enum HealOutcome {
+        Ok,
+        /// The version vanished before heal ran (deleted mid-heal).
+        VersionNotFound,
+    }
+
+    #[derive(Default)]
+    struct FakeStorage {
+        /// page keyed by the *incoming* continuation token
+        pages: Mutex<HashMap<Option<String>, Page>>,
+        /// per-`compose_key` heal outcome; default is `Ok`
+        outcomes: Mutex<HashMap<String, HealOutcome>>,
+        /// every heal_object call recorded as (name, version_id)
+        heal_calls: Mutex<Vec<(String, Option<String>)>>,
+    }
+
+    impl FakeStorage {
+        fn set_page(&self, token: Option<&str>, page: Page) {
+            self.pages.lock().unwrap().insert(token.map(str::to_string), page);
+        }
+        fn set_outcome(&self, name: &str, version: Option<&str>, outcome: HealOutcome) {
+            self.outcomes.lock().unwrap().insert(compose_key(name, version), outcome);
+        }
+        fn calls(&self) -> Vec<(String, Option<String>)> {
+            self.heal_calls.lock().unwrap().clone()
+        }
+    }
+
+    #[async_trait::async_trait]
+    impl HealStorageAPI for FakeStorage {
+        async fn get_object_meta(&self, _b: &str, _o: &str) -> Result<Option<HealObjectInfo>> {
+            Ok(None)
+        }
+        async fn get_object_data(&self, _b: &str, _o: &str) -> Result<Option<Vec<u8>>> {
+            Ok(None)
+        }
+        async fn put_object_data(&self, _b: &str, _o: &str, _d: &[u8]) -> Result<()> {
+            Ok(())
+        }
+        async fn delete_object(&self, _b: &str, _o: &str) -> Result<()> {
+            Ok(())
+        }
+        async fn verify_object_integrity(&self, _b: &str, _o: &str) -> Result<bool> {
+            Ok(true)
+        }
+        async fn ec_decode_rebuild(&self, _b: &str, _o: &str) -> Result<Vec<u8>> {
+            Ok(Vec::new())
+        }
+        async fn get_disk_status(&self, _e: &Endpoint) -> Result<DiskStatus> {
+            Ok(DiskStatus::Ok)
+        }
+        async fn format_disk(&self, _e: &Endpoint) -> Result<()> {
+            Ok(())
+        }
+        async fn get_bucket_info(&self, bucket: &str) -> Result<Option<BucketInfo>> {
+            Ok(Some(BucketInfo {
+                name: bucket.to_string(),
+                ..Default::default()
+            }))
+        }
+        async fn heal_bucket_metadata(&self, _b: &str) -> Result<()> {
+            Ok(())
+        }
+        async fn list_buckets(&self) -> Result<Vec<BucketInfo>> {
+            Ok(Vec::new())
+        }
+        async fn object_exists(&self, _b: &str, _o: &str) -> Result<bool> {
+            // Must never be consulted: the resume loop always goes through heal_object.
+            panic!("object_exists must not be called by the resume heal loop");
+        }
+        async fn get_object_size(&self, _b: &str, _o: &str) -> Result<Option<u64>> {
+            Ok(None)
+        }
+        async fn get_object_checksum(&self, _b: &str, _o: &str) -> Result<Option<String>> {
+            Ok(None)
+        }
+        async fn heal_object(
+            &self,
+            _bucket: &str,
+            object: &str,
+            version_id: Option<&str>,
+            _opts: &HealOpts,
+        ) -> Result<(HealResultItem, Option<Error>)> {
+            self.heal_calls
+                .lock()
+                .unwrap()
+                .push((object.to_string(), version_id.map(str::to_string)));
+            let key = compose_key(object, version_id);
+            let outcome = self.outcomes.lock().unwrap().get(&key).cloned().unwrap_or(HealOutcome::Ok);
+            match outcome {
+                HealOutcome::Ok => Ok((HealResultItem::default(), None)),
+                HealOutcome::VersionNotFound => {
+                    Ok((HealResultItem::default(), Some(Error::Storage(EcstoreError::FileVersionNotFound))))
+                }
+            }
+        }
+        async fn heal_bucket(&self, _b: &str, _o: &HealOpts) -> Result<HealResultItem> {
+            Ok(HealResultItem::default())
+        }
+        async fn heal_format(&self, _dry: bool) -> Result<(HealResultItem, Option<Error>)> {
+            Ok((HealResultItem::default(), None))
+        }
+        async fn list_objects_for_heal(&self, _b: &str, _p: &str) -> Result<Vec<HealListItem>> {
+            Ok(Vec::new())
+        }
+        async fn list_objects_for_heal_page(
+            &self,
+            _bucket: &str,
+            _prefix: &str,
+            continuation_token: Option<&str>,
+        ) -> Result<(Vec<HealListItem>, Option<String>, bool)> {
+            let key = continuation_token.map(str::to_string);
+            let page = self.pages.lock().unwrap().get(&key).cloned();
+            match page {
+                Some(p) => Ok((p.items, p.next, p.truncated)),
+                None => Ok((Vec::new(), None, false)),
+            }
+        }
+        async fn get_disk_for_resume(&self, _id: &str) -> Result<DiskStore> {
+            Err(Error::other("not implemented in tests"))
+        }
+    }
+
+    async fn make_disk(temp: &TempDir) -> DiskStore {
+        let disk_path = temp.path().join("test_disk");
+        std::fs::create_dir_all(&disk_path).unwrap();
+        let endpoint = Endpoint::try_from(disk_path.to_string_lossy().as_ref()).unwrap();
+        let disk = new_disk(
+            &endpoint,
+            &DiskOption {
+                cleanup: false,
+                health_check: false,
+            },
+        )
+        .await
+        .unwrap();
+        let _ = disk.make_volume(RUSTFS_META_BUCKET).await;
+        let _ = disk.make_volume(&format!("{RUSTFS_META_BUCKET}/{BUCKET_META_PREFIX}")).await;
+        disk
+    }
+
+    struct Env {
+        healer: ErasureSetHealer,
+        storage: Arc<FakeStorage>,
+        resume: ResumeManager,
+        checkpoint: CheckpointManager,
+        _temp: TempDir,
+    }
+
+    async fn make_env() -> Env {
+        let temp = TempDir::new().unwrap();
+        let disk = make_disk(&temp).await;
+        let storage = Arc::new(FakeStorage::default());
+        let healer = ErasureSetHealer::new(
+            storage.clone(),
+            Arc::new(RwLock::new(HealProgress::new())),
+            CancellationToken::new(),
+            disk.clone(),
+            HealOpts::default(),
+            HealRequestSource::Internal,
+        );
+        let resume = ResumeManager::new(
+            disk.clone(),
+            "task".to_string(),
+            "erasure_set".to_string(),
+            "pool_0_set_0".to_string(),
+            vec!["b".to_string()],
+        )
+        .await
+        .unwrap();
+        let checkpoint = CheckpointManager::new(disk, "task".to_string()).await.unwrap();
+        Env {
+            healer,
+            storage,
+            resume,
+            checkpoint,
+            _temp: temp,
+        }
+    }
+
+    /// Drive one bucket heal pass; returns (processed, successful, failed, skipped, result).
+    async fn run(env: &Env) -> (u64, u64, u64, u64, Result<()>) {
+        let mut current_object_index = 0usize;
+        let mut processed = 0u64;
+        let mut successful = 0u64;
+        let mut failed = 0u64;
+        let mut skipped = 0u64;
+        let result = env
+            .healer
+            .heal_bucket_with_resume(
+                "b",
+                "pool_0_set_0",
+                0,
+                &mut current_object_index,
+                &mut processed,
+                &mut successful,
+                &mut failed,
+                &mut skipped,
+                &env.resume,
+                &env.checkpoint,
+            )
+            .await;
+        (processed, successful, failed, skipped, result)
+    }
+
+    #[tokio::test]
+    async fn test_empty_bucket_no_panic() {
+        let env = make_env().await;
+        // no pages configured => empty, non-truncated page
+        let (processed, successful, failed, skipped, result) = run(&env).await;
+        result.expect("empty bucket must succeed");
+        assert_eq!(processed, 0);
+        assert_eq!(successful, 0);
+        assert_eq!(failed, 0);
+        assert_eq!(skipped, 0);
+        assert!(env.storage.calls().is_empty());
+        assert_eq!(env.resume.resume_cursor().await, None);
+    }
+
+    #[tokio::test]
+    async fn test_resume_across_page_boundary_no_drop_no_double() {
+        let env = make_env().await;
+        env.storage.set_page(
+            None,
+            Page {
+                items: vec![item("a", None, false), item("b", None, false)],
+                next: Some("t1".to_string()),
+                truncated: true,
+            },
+        );
+        env.storage.set_page(
+            Some("t1"),
+            Page {
+                items: vec![item("c", None, false), item("d", None, false)],
+                next: None,
+                truncated: false,
+            },
+        );
+
+        let (processed, successful, failed, _skipped, result) = run(&env).await;
+        result.expect("two-page heal must succeed");
+        assert_eq!(processed, 4);
+        assert_eq!(successful, 4);
+        assert_eq!(failed, 0);
+
+        let mut names: Vec<String> = env.storage.calls().into_iter().map(|(n, _)| n).collect();
+        names.sort();
+        assert_eq!(names, vec!["a", "b", "c", "d"], "every object exactly once, none dropped/doubled");
+        // Final page not truncated => cursor cleared.
+        assert_eq!(env.resume.resume_cursor().await, None);
+    }
+
+    #[tokio::test]
+    async fn test_object_with_versions_spanning_pages_advances() {
+        let env = make_env().await;
+        env.storage.set_page(
+            None,
+            Page {
+                items: vec![item("obj", Some("v1"), false)],
+                next: Some("t1".to_string()),
+                truncated: true,
+            },
+        );
+        env.storage.set_page(
+            Some("t1"),
+            Page {
+                items: vec![item("obj", Some("v2"), false)],
+                next: None,
+                truncated: false,
+            },
+        );
+
+        let (processed, _s, failed, _sk, result) = run(&env).await;
+        result.expect("object whose versions span pages must heal fully");
+        assert_eq!(processed, 2);
+        assert_eq!(failed, 0);
+        let calls = env.storage.calls();
+        assert!(calls.contains(&("obj".to_string(), Some("v1".to_string()))));
+        assert!(calls.contains(&("obj".to_string(), Some("v2".to_string()))));
+    }
+
+    #[tokio::test]
+    async fn test_non_advancing_cursor_aborts() {
+        let env = make_env().await;
+        // Both pages end on the same (name, version) even though the raw token
+        // advances (t1 -> t2): identity comparison must detect the stall.
+        env.storage.set_page(
+            None,
+            Page {
+                items: vec![item("a", None, false)],
+                next: Some("t1".to_string()),
+                truncated: true,
+            },
+        );
+        env.storage.set_page(
+            Some("t1"),
+            Page {
+                items: vec![item("a", None, false)],
+                next: Some("t2".to_string()),
+                truncated: true,
+            },
+        );
+
+        let (_p, _s, _f, _sk, result) = run(&env).await;
+        let err = result.expect_err("a non-advancing cursor must abort the loop");
+        assert!(err.to_string().contains("not advancing"), "unexpected error: {err}");
+    }
+
+    #[tokio::test]
+    async fn test_concurrent_page_dedup_exact_once() {
+        let env = make_env().await;
+        let items: Vec<HealListItem> = (0..50).map(|i| item(&format!("obj-{i}"), Some("v"), false)).collect();
+        env.storage.set_page(
+            None,
+            Page {
+                items,
+                next: None,
+                truncated: false,
+            },
+        );
+
+        let (processed, successful, failed, _sk, result) = run(&env).await;
+        result.expect("concurrent page must succeed");
+        assert_eq!(processed, 50);
+        assert_eq!(successful, 50);
+        assert_eq!(failed, 0);
+        let calls = env.storage.calls();
+        assert_eq!(calls.len(), 50, "each version healed exactly once under concurrency");
+        let unique: std::collections::HashSet<_> = calls.into_iter().collect();
+        assert_eq!(unique.len(), 50, "no version healed twice");
+    }
+
+    #[tokio::test]
+    async fn test_resume_after_version_deleted_midheal_no_skip() {
+        let env = make_env().await;
+        // Simulate a resumed pass: (a,v1) was already processed last time, and
+        // the cursor points at the in-flight page.
+        env.checkpoint
+            .add_processed_object(compose_key("a", Some("v1")))
+            .await
+            .unwrap();
+        env.resume.set_resume_cursor(Some("t0".to_string())).await.unwrap();
+
+        env.storage.set_page(
+            Some("t0"),
+            Page {
+                items: vec![
+                    item("a", Some("v1"), false), // already done -> deduped
+                    item("a", Some("v2"), false), // deleted mid-heal
+                    item("c", None, true),        // delete-marker latest, still healed
+                ],
+                next: None,
+                truncated: false,
+            },
+        );
+        // v2 vanished before heal ran.
+        env.storage.set_outcome("a", Some("v2"), HealOutcome::VersionNotFound);
+
+        let (_p, _s, failed, skipped, result) = run(&env).await;
+        result.expect("resume after mid-heal deletion must succeed");
+        // Genuine absence is handled (Ok), never counted as a failure.
+        assert_eq!(failed, 0, "a deleted version must not be a failure");
+        assert_eq!(skipped, 0, "absence is treated as healed-ok, not skipped");
+
+        let calls = env.storage.calls();
+        assert!(
+            !calls.contains(&("a".to_string(), Some("v1".to_string()))),
+            "already-processed version must be deduped, not re-healed"
+        );
+        assert!(
+            calls.contains(&("a".to_string(), Some("v2".to_string()))),
+            "the surviving-but-now-gone version must still be attempted, not skipped"
+        );
+        assert!(calls.contains(&("c".to_string(), None)), "delete-marker latest must be healed");
+    }
+
+    #[tokio::test]
+    async fn test_schedule_retry_resets_both_managers_and_reheals() {
+        let env = make_env().await;
+        // Seed some progress that a retry must discard.
+        env.checkpoint
+            .add_processed_object(compose_key("a", Some("v1")))
+            .await
+            .unwrap();
+        env.checkpoint.update_position(1, 42).await.unwrap();
+        env.resume.set_resume_cursor(Some("t9".to_string())).await.unwrap();
+        assert_eq!(env.resume.resume_cursor().await, Some("t9".to_string()));
+
+        // The retry branch calls BOTH together.
+        assert!(env.resume.schedule_retry().await.unwrap(), "retry budget should be available");
+        env.checkpoint.reset_for_retry().await.unwrap();
+
+        // Resume state: cursor cleared, retry counted, progress zeroed.
+        let state = env.resume.get_state().await;
+        assert_eq!(env.resume.resume_cursor().await, None, "cursor must be cleared for a full re-scan");
+        assert_eq!(state.retry_count, 1);
+        // Checkpoint: dedup sets and position cleared so the retry re-heals everything.
+        let checkpoint = env.checkpoint.get_checkpoint().await;
+        assert!(checkpoint.processed_objects.is_empty());
+        assert_eq!(checkpoint.current_object_index, 0);
+    }
+}
diff --git a/crates/heal/src/heal/manager.rs b/crates/heal/src/heal/manager.rs
index f62286b94..338073ce0 100644
--- a/crates/heal/src/heal/manager.rs
+++ b/crates/heal/src/heal/manager.rs
@@ -3039,7 +3039,7 @@ mod tests {
             Ok((HealResultItem::default(), None))
         }
 
-        async fn list_objects_for_heal(&self, _bucket: &str, _prefix: &str) -> Result<Vec<String>> {
+        async fn list_objects_for_heal(&self, _bucket: &str, _prefix: &str) -> Result<Vec<crate::heal::storage::HealListItem>> {
             Ok(Vec::new())
         }
 
@@ -3048,7 +3048,7 @@ mod tests {
             _bucket: &str,
             _prefix: &str,
             _continuation_token: Option<&str>,
-        ) -> Result<(Vec<String>, Option<String>, bool)> {
+        ) -> Result<(Vec<crate::heal::storage::HealListItem>, Option<String>, bool)> {
             Ok((Vec::new(), None, false))
         }
 
diff --git a/crates/heal/src/heal/resume.rs b/crates/heal/src/heal/resume.rs
index 757cf1378..7c44b25f6 100644
--- a/crates/heal/src/heal/resume.rs
+++ b/crates/heal/src/heal/resume.rs
@@ -34,6 +34,25 @@ const RESUME_STATE_FILE: &str = "ahm_resume_state.json";
 const RESUME_PROGRESS_FILE: &str = "ahm_progress.json";
 const RESUME_CHECKPOINT_FILE: &str = "ahm_checkpoint.json";
 
+/// Current on-disk schema version for `ResumeState`. Snapshots written by an
+/// older schema (which tracked latest-only object names and a positional
+/// cursor) are incompatible with the per-version resume cursor, so they are
+/// discarded on load and the scan restarts from the beginning.
+const CURRENT_RESUME_SCHEMA: u32 = 2;
+/// 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.
+const CURRENT_CHECKPOINT_SCHEMA: u32 = 2;
+
+/// 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(""))
+}
+
 /// Persistence throttle for per-object bookkeeping: flush after this many
 /// buffered mutations or once the interval elapses, whichever comes first.
 /// Object heal is idempotent, so a crash re-heals at most one throttle window.
@@ -76,6 +95,13 @@ fn path_to_str(path: &Path) -> Result<&str> {
 /// resume state
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct ResumeState {
+    /// on-disk schema version; absent in legacy snapshots (defaults to 0)
+    #[serde(default)]
+    pub schema_version: u32,
+    /// authoritative opaque `(marker, version_marker)` continuation token for
+    /// the version listing. `None` means "start from the beginning".
+    #[serde(default)]
+    pub resume_cursor: Option<String>,
     /// task id
     pub task_id: String,
     /// task type
@@ -118,6 +144,8 @@ pub struct ResumeState {
 impl ResumeState {
     pub fn new(task_id: String, task_type: String, set_disk_id: String, buckets: Vec<String>) -> Self {
         Self {
+            schema_version: CURRENT_RESUME_SCHEMA,
+            resume_cursor: None,
             task_id,
             task_type,
             set_disk_id,
@@ -153,6 +181,18 @@ impl ResumeState {
         self.last_update = SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_secs();
     }
 
+    /// Read the authoritative version-listing continuation cursor.
+    pub fn resume_cursor(&self) -> Option<String> {
+        self.resume_cursor.clone()
+    }
+
+    /// Persist the authoritative version-listing continuation cursor. `None`
+    /// resets the scan to the beginning of the current bucket.
+    pub fn set_resume_cursor(&mut self, cursor: Option<String>) {
+        self.resume_cursor = cursor;
+        self.last_update = SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_secs();
+    }
+
     pub fn complete_bucket(&mut self, bucket: &str) {
         if !self.completed_buckets.contains(&bucket.to_string()) {
             self.completed_buckets.push(bucket.to_string());
@@ -178,6 +218,9 @@ impl ResumeState {
         self.failed_objects = 0;
         self.skipped_objects = 0;
         self.completed = false;
+        // A retry re-scans every bucket from the beginning, so the version
+        // cursor must be cleared too — otherwise the retry would resume mid-scan.
+        self.resume_cursor = None;
         self.last_update = SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_secs();
     }
 
@@ -252,10 +295,34 @@ impl ResumeManager {
     /// load resume state from disk
     pub async fn load_from_disk(disk: DiskStore, task_id: &str) -> Result<Self> {
         let state_data = Self::read_state_file(&disk, task_id).await?;
-        let state: ResumeState = serde_json::from_slice(&state_data).map_err(|e| Error::TaskExecutionFailed {
+        let mut state: ResumeState = serde_json::from_slice(&state_data).map_err(|e| Error::TaskExecutionFailed {
             message: format!("Failed to deserialize resume state: {e}"),
         })?;
 
+        // A snapshot written by an older schema tracked a latest-only positional
+        // cursor that is meaningless under per-version resume. Discard the stale
+        // progress so the scan restarts cleanly, then stamp the current schema.
+        if state.schema_version < CURRENT_RESUME_SCHEMA {
+            warn!(
+                target: "rustfs::heal::resume",
+                event = EVENT_HEAL_RESUME_STATE,
+                component = LOG_COMPONENT_HEAL,
+                subsystem = LOG_SUBSYSTEM_RESUME,
+                task_id,
+                found_schema = state.schema_version,
+                current_schema = CURRENT_RESUME_SCHEMA,
+                state = "schema_discarded",
+                "Heal resume state schema is stale; discarding cursor and progress"
+            );
+            state.resume_cursor = None;
+            state.processed_objects = 0;
+            state.successful_objects = 0;
+            state.failed_objects = 0;
+            state.skipped_objects = 0;
+            state.completed = false;
+            state.schema_version = CURRENT_RESUME_SCHEMA;
+        }
+
         Ok(Self {
             disk,
             state: Arc::new(RwLock::new(state)),
@@ -312,6 +379,21 @@ impl ResumeManager {
         result
     }
 
+    /// Read the authoritative version-listing continuation cursor.
+    pub async fn resume_cursor(&self) -> Option<String> {
+        self.state.read().await.resume_cursor()
+    }
+
+    /// Persist the authoritative version-listing continuation cursor. This is
+    /// written unthrottled (once per completed page) so a crash always resumes
+    /// from a real page boundary.
+    pub async fn set_resume_cursor(&self, cursor: Option<String>) -> Result<()> {
+        let mut state = self.state.write().await;
+        state.set_resume_cursor(cursor);
+        drop(state);
+        self.save_state().await
+    }
+
     /// complete bucket
     pub async fn complete_bucket(&self, bucket: &str) -> Result<()> {
         let mut state = self.state.write().await;
@@ -450,6 +532,9 @@ impl ResumeManager {
 /// 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
@@ -472,6 +557,7 @@ pub struct ResumeCheckpoint {
 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,
@@ -555,9 +641,33 @@ impl CheckpointManager {
     /// load checkpoint from disk
     pub async fn load_from_disk(disk: DiskStore, task_id: &str) -> Result<Self> {
         let checkpoint_data = Self::read_checkpoint_file(&disk, task_id).await?;
-        let checkpoint: ResumeCheckpoint = serde_json::from_slice(&checkpoint_data).map_err(|e| Error::TaskExecutionFailed {
-            message: format!("Failed to deserialize checkpoint: {e}"),
-        })?;
+        let mut checkpoint: ResumeCheckpoint =
+            serde_json::from_slice(&checkpoint_data).map_err(|e| Error::TaskExecutionFailed {
+                message: format!("Failed to deserialize checkpoint: {e}"),
+            })?;
+
+        // 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 {
+            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_object_index = 0;
+            checkpoint.schema_version = CURRENT_CHECKPOINT_SCHEMA;
+        }
 
         Ok(Self {
             disk,
@@ -957,6 +1067,139 @@ mod tests {
         assert!(checkpoint.skipped_objects.contains("c"));
     }
 
+    #[test]
+    fn test_compose_key_injective_with_adversarial_keys() {
+        // Length-prefixing must keep the encoding injective even when keys
+        // contain the delimiter, embedded nulls, or look like a composed key.
+        assert_ne!(compose_key("a\0b", None), compose_key("a", Some("b")));
+        assert_ne!(compose_key("3:xy", None), compose_key("x", Some("y")));
+        assert_ne!(compose_key("a:b", None), compose_key("a", Some("b")));
+        assert_ne!(compose_key("", Some("x")), compose_key("x", None));
+        // Identical inputs must produce identical keys (stable identity).
+        assert_eq!(compose_key("obj", Some("v1")), compose_key("obj", Some("v1")));
+    }
+
+    #[test]
+    fn test_composite_key_dedup_distinguishes_versions() {
+        // Two versions of the same object must be distinct dedup identities, and
+        // the delete-marker/nil (None) version must not collide with a real one.
+        let mut checkpoint = ResumeCheckpoint::new("task".to_string());
+        checkpoint.add_processed_object(compose_key("obj", Some("v1")));
+        checkpoint.add_processed_object(compose_key("obj", Some("v2")));
+        checkpoint.add_processed_object(compose_key("obj", None));
+        assert_eq!(checkpoint.processed_objects.len(), 3);
+        assert!(checkpoint.processed_objects.contains(&compose_key("obj", Some("v1"))));
+        assert!(checkpoint.processed_objects.contains(&compose_key("obj", Some("v2"))));
+        assert!(checkpoint.processed_objects.contains(&compose_key("obj", None)));
+        // A different object with the same version id is still distinct.
+        assert!(!checkpoint.processed_objects.contains(&compose_key("other", Some("v1"))));
+    }
+
+    #[tokio::test]
+    async fn test_resumestate_schema_v0_discarded_on_load() {
+        use super::super::{DiskOption, Endpoint, new_disk};
+        use tempfile::TempDir;
+
+        let temp_dir = TempDir::new().unwrap();
+        let disk_path = temp_dir.path().join("test_disk");
+        std::fs::create_dir_all(&disk_path).unwrap();
+        let endpoint = Endpoint::try_from(disk_path.to_string_lossy().as_ref()).unwrap();
+        let disk = new_disk(
+            &endpoint,
+            &DiskOption {
+                cleanup: false,
+                health_check: false,
+            },
+        )
+        .await
+        .unwrap();
+        let _ = disk.make_volume(RUSTFS_META_BUCKET).await;
+        let _ = disk.make_volume(&format!("{RUSTFS_META_BUCKET}/{BUCKET_META_PREFIX}")).await;
+
+        // Legacy snapshot: no schema_version, a stale positional cursor and progress.
+        let legacy = r#"{
+            "task_id": "old-task",
+            "task_type": "erasure_set",
+            "set_disk_id": "pool_0_set_0",
+            "start_time": 1700000000,
+            "last_update": 1700000000,
+            "completed": true,
+            "total_objects": 100,
+            "processed_objects": 50,
+            "successful_objects": 40,
+            "failed_objects": 10,
+            "skipped_objects": 0,
+            "current_bucket": null,
+            "current_object": null,
+            "completed_buckets": ["b1"],
+            "pending_buckets": [],
+            "error_message": null,
+            "retry_count": 1,
+            "max_retries": 3,
+            "resume_cursor": "v1:stale-token"
+        }"#;
+        let file_path = format!("{BUCKET_META_PREFIX}/old-task_{RESUME_STATE_FILE}");
+        disk.write_all(RUSTFS_META_BUCKET, &file_path, legacy.as_bytes().to_vec().into())
+            .await
+            .unwrap();
+
+        let manager = ResumeManager::load_from_disk(disk.clone(), "old-task").await.unwrap();
+        let state = manager.get_state().await;
+        assert_eq!(state.schema_version, CURRENT_RESUME_SCHEMA, "schema must be stamped current");
+        assert_eq!(state.resume_cursor, None, "stale cursor must be cleared");
+        assert_eq!(state.processed_objects, 0);
+        assert_eq!(state.successful_objects, 0);
+        assert_eq!(state.failed_objects, 0);
+        assert!(!state.completed);
+        temp_dir.close().unwrap();
+    }
+
+    #[tokio::test]
+    async fn test_checkpoint_schema_v0_discarded_on_load() {
+        use super::super::{DiskOption, Endpoint, new_disk};
+        use tempfile::TempDir;
+
+        let temp_dir = TempDir::new().unwrap();
+        let disk_path = temp_dir.path().join("test_disk");
+        std::fs::create_dir_all(&disk_path).unwrap();
+        let endpoint = Endpoint::try_from(disk_path.to_string_lossy().as_ref()).unwrap();
+        let disk = new_disk(
+            &endpoint,
+            &DiskOption {
+                cleanup: false,
+                health_check: false,
+            },
+        )
+        .await
+        .unwrap();
+        let _ = disk.make_volume(RUSTFS_META_BUCKET).await;
+        let _ = disk.make_volume(&format!("{RUSTFS_META_BUCKET}/{BUCKET_META_PREFIX}")).await;
+
+        // Legacy checkpoint: no schema_version, stale position and dedup sets.
+        let legacy = r#"{
+            "task_id": "old-task",
+            "checkpoint_time": 1700000000,
+            "current_bucket_index": 2,
+            "current_object_index": 500,
+            "processed_objects": ["a", "b"],
+            "failed_objects": ["c"],
+            "skipped_objects": ["d"]
+        }"#;
+        let file_path = format!("{BUCKET_META_PREFIX}/old-task_{RESUME_CHECKPOINT_FILE}");
+        disk.write_all(RUSTFS_META_BUCKET, &file_path, legacy.as_bytes().to_vec().into())
+            .await
+            .unwrap();
+
+        let manager = CheckpointManager::load_from_disk(disk.clone(), "old-task").await.unwrap();
+        let checkpoint = manager.get_checkpoint().await;
+        assert_eq!(checkpoint.schema_version, CURRENT_CHECKPOINT_SCHEMA, "schema must be stamped current");
+        assert_eq!(checkpoint.current_object_index, 0, "stale position must be reset");
+        assert!(checkpoint.processed_objects.is_empty());
+        assert!(checkpoint.failed_objects.is_empty());
+        assert!(checkpoint.skipped_objects.is_empty());
+        temp_dir.close().unwrap();
+    }
+
     #[test]
     fn test_persist_throttle_batches_until_threshold() {
         let mut throttle = PersistThrottle::new();
diff --git a/crates/heal/src/heal/storage.rs b/crates/heal/src/heal/storage.rs
index 2077553b5..ac93fe778 100644
--- a/crates/heal/src/heal/storage.rs
+++ b/crates/heal/src/heal/storage.rs
@@ -14,8 +14,11 @@
 
 use crate::{Error, Result};
 use async_trait::async_trait;
+use base64::Engine as _;
+use base64::engine::general_purpose::URL_SAFE_NO_PAD;
 use rustfs_common::heal_channel::{HealOpts, HealScanMode};
 use rustfs_madmin::heal_commands::HealResultItem;
+use serde::{Deserialize, Serialize};
 use std::sync::Arc;
 use tracing::{debug, error, warn};
 
@@ -44,9 +47,156 @@ pub(crate) fn next_heal_listing_token(
         return Ok(None);
     }
 
-    next_token.map(Some).ok_or_else(|| Error::TaskExecutionFailed {
-        message: format!("Object listing for {bucket}/{prefix} was truncated without continuation token"),
-    })
+    match next_token {
+        Some(token) => Ok(Some(token)),
+        None => {
+            // A version listing legitimately reports the final page as truncated
+            // when the last object's versions land exactly on the page boundary
+            // yet the backend has nothing further to yield. Treat a missing
+            // continuation token as end-of-listing rather than a hard error so
+            // the heal pass terminates cleanly instead of failing the bucket.
+            warn!(
+                target: "rustfs::heal::storage",
+                event = EVENT_HEAL_STORAGE_ADMIN_OP,
+                component = LOG_COMPONENT_HEAL,
+                subsystem = LOG_SUBSYSTEM_STORAGE,
+                operation = "next_heal_listing_token",
+                bucket,
+                prefix,
+                state = "truncated_without_token",
+                "Heal storage object listing truncated without continuation token; treating as end of listing"
+            );
+            Ok(None)
+        }
+    }
+}
+
+/// Opaque continuation token payload for heal version listing. Encodes the
+/// `(marker, version_marker)` pair that `list_object_versions` needs to resume.
+#[derive(Debug, Serialize, Deserialize)]
+struct HealTokenPayload {
+    /// object key marker
+    #[serde(rename = "m")]
+    m: Option<String>,
+    /// version-id marker
+    #[serde(rename = "v")]
+    v: Option<String>,
+}
+
+const HEAL_TOKEN_PREFIX: &str = "v1:";
+
+/// Encode a `(marker, version_marker)` pair into an opaque heal continuation
+/// token. The token is `"v1:" + base64url_nopad(json)`.
+///
+/// Invariant: `list_object_versions` returns `NotImplemented` for
+/// `(None, Some(_))`, so callers must never produce that pair. This is checked
+/// with a `debug_assert!`.
+pub(crate) fn encode_heal_token(marker: Option<&str>, version_marker: Option<&str>) -> String {
+    debug_assert!(
+        !(marker.is_none() && version_marker.is_some()),
+        "encode_heal_token must never be called with (None, Some(_))"
+    );
+
+    let payload = HealTokenPayload {
+        m: marker.map(str::to_string),
+        v: version_marker.map(str::to_string),
+    };
+    // serde_json of a simple two-Option struct cannot fail; fall back to an
+    // empty object rather than panicking if it somehow does.
+    let json = serde_json::to_vec(&payload).unwrap_or_else(|_| b"{}".to_vec());
+    format!("{HEAL_TOKEN_PREFIX}{}", URL_SAFE_NO_PAD.encode(json))
+}
+
+/// Decode an opaque heal continuation token back into `(marker, version_marker)`.
+///
+/// TOTAL function: an empty token, a missing `"v1:"` prefix, invalid base64, or
+/// invalid JSON all decode to `(None, None)` (start from the beginning). A
+/// decoded `(None, Some(_))` is coerced to `(None, None)` to preserve the
+/// `list_object_versions` invariant.
+pub(crate) fn decode_heal_token(token: &str) -> (Option<String>, Option<String>) {
+    if token.is_empty() {
+        return (None, None);
+    }
+
+    let Some(encoded) = token.strip_prefix(HEAL_TOKEN_PREFIX) else {
+        warn!(
+            target: "rustfs::heal::storage",
+            event = EVENT_HEAL_STORAGE_ADMIN_OP,
+            component = LOG_COMPONENT_HEAL,
+            subsystem = LOG_SUBSYSTEM_STORAGE,
+            operation = "decode_heal_token",
+            state = "missing_prefix",
+            "Heal continuation token missing version prefix; restarting listing"
+        );
+        return (None, None);
+    };
+
+    let bytes = match URL_SAFE_NO_PAD.decode(encoded) {
+        Ok(bytes) => bytes,
+        Err(e) => {
+            warn!(
+                target: "rustfs::heal::storage",
+                event = EVENT_HEAL_STORAGE_ADMIN_OP,
+                component = LOG_COMPONENT_HEAL,
+                subsystem = LOG_SUBSYSTEM_STORAGE,
+                operation = "decode_heal_token",
+                state = "bad_base64",
+                error = %e,
+                "Heal continuation token has invalid base64; restarting listing"
+            );
+            return (None, None);
+        }
+    };
+
+    let payload: HealTokenPayload = match serde_json::from_slice(&bytes) {
+        Ok(payload) => payload,
+        Err(e) => {
+            warn!(
+                target: "rustfs::heal::storage",
+                event = EVENT_HEAL_STORAGE_ADMIN_OP,
+                component = LOG_COMPONENT_HEAL,
+                subsystem = LOG_SUBSYSTEM_STORAGE,
+                operation = "decode_heal_token",
+                state = "bad_json",
+                error = %e,
+                "Heal continuation token has invalid payload; restarting listing"
+            );
+            return (None, None);
+        }
+    };
+
+    // Preserve the list_object_versions invariant: (None, Some(_)) is illegal.
+    if payload.m.is_none() && payload.v.is_some() {
+        warn!(
+            target: "rustfs::heal::storage",
+            event = EVENT_HEAL_STORAGE_ADMIN_OP,
+            component = LOG_COMPONENT_HEAL,
+            subsystem = LOG_SUBSYSTEM_STORAGE,
+            operation = "decode_heal_token",
+            state = "illegal_version_only_marker",
+            "Heal continuation token had a version marker without an object marker; restarting listing"
+        );
+        return (None, None);
+    }
+
+    (payload.m, payload.v)
+}
+
+/// A single object version to heal.
+///
+/// `is_delete_marker` is OBSERVABILITY-ONLY (metrics / logging / e2e
+/// assertions); it MUST NOT gate healing logic. Whether the delete-marker path
+/// or the data path is taken is decided internally in `ops/heal.rs` from
+/// `latest_meta.deleted`. `version_id` is normalized (nil/absent UUID => `None`)
+/// at the single construction point in `list_objects_for_heal_page`.
+#[derive(Debug, Clone)]
+pub struct HealListItem {
+    /// object key
+    pub name: String,
+    /// normalized version id (`None` when the version is nil/absent)
+    pub version_id: Option<String>,
+    /// whether this version is a delete marker (observability only)
+    pub is_delete_marker: bool,
 }
 
 /// Disk status for heal operations
@@ -132,20 +282,21 @@ pub trait HealStorageAPI: Send + Sync {
     /// Heal format using ecstore
     async fn heal_format(&self, dry_run: bool) -> Result<(HealResultItem, Option<Error>)>;
 
-    /// List objects for healing (returns all objects, may use significant memory for large buckets)
+    /// List object versions for healing (returns all versions, may use significant memory for large buckets)
     ///
-    /// WARNING: This method loads all objects into memory at once. For buckets with many objects,
-    /// consider using `list_objects_for_heal_page` instead to process objects in pages.
-    async fn list_objects_for_heal(&self, bucket: &str, prefix: &str) -> Result<Vec<String>>;
+    /// WARNING: This method loads all object versions into memory at once. For buckets with many
+    /// objects/versions, consider using `list_objects_for_heal_page` instead to process versions in pages.
+    async fn list_objects_for_heal(&self, bucket: &str, prefix: &str) -> Result<Vec<HealListItem>>;
 
-    /// List objects for healing with pagination (returns one page and continuation token)
-    /// Returns (objects, next_continuation_token, is_truncated)
+    /// List object versions for healing with pagination (returns one page and continuation token)
+    /// Returns (versions, next_continuation_token, is_truncated). The continuation token is an
+    /// opaque composite `(marker, version_marker)` value — see `encode_heal_token`/`decode_heal_token`.
     async fn list_objects_for_heal_page(
         &self,
         bucket: &str,
         prefix: &str,
         continuation_token: Option<&str>,
-    ) -> Result<(Vec<String>, Option<String>, bool)>;
+    ) -> Result<(Vec<HealListItem>, Option<String>, bool)>;
 
     /// Get disk for resume functionality
     async fn get_disk_for_resume(&self, set_disk_id: &str) -> Result<DiskStore>;
@@ -1063,7 +1214,7 @@ impl HealStorageAPI for ECStoreHealStorage {
         }
     }
 
-    async fn list_objects_for_heal(&self, bucket: &str, prefix: &str) -> Result<Vec<String>> {
+    async fn list_objects_for_heal(&self, bucket: &str, prefix: &str) -> Result<Vec<HealListItem>> {
         debug!(
             target: "rustfs::heal::storage",
             event = EVENT_HEAL_STORAGE_ADMIN_OP,
@@ -1084,10 +1235,10 @@ impl HealStorageAPI for ECStoreHealStorage {
             bucket,
             prefix,
             state = "memory_heavy",
-            "Heal storage object listing loads all objects into memory"
+            "Heal storage version listing loads all versions into memory (footprint is per-version, not per-object)"
         );
 
-        let mut all_objects = Vec::new();
+        let mut all_objects: Vec<HealListItem> = Vec::new();
         let mut continuation_token: Option<String> = None;
 
         loop {
@@ -1102,6 +1253,9 @@ impl HealStorageAPI for ECStoreHealStorage {
             }
 
             continuation_token = next_heal_listing_token(bucket, prefix, next_token, is_truncated)?;
+            if continuation_token.is_none() {
+                break;
+            }
         }
 
         debug!(
@@ -1124,7 +1278,7 @@ impl HealStorageAPI for ECStoreHealStorage {
         bucket: &str,
         prefix: &str,
         continuation_token: Option<&str>,
-    ) -> Result<(Vec<String>, Option<String>, bool)> {
+    ) -> Result<(Vec<HealListItem>, Option<String>, bool)> {
         debug!(
             target: "rustfs::heal::storage",
             event = EVENT_HEAL_STORAGE_ADMIN_OP,
@@ -1139,13 +1293,17 @@ impl HealStorageAPI for ECStoreHealStorage {
         );
 
         const MAX_KEYS: i32 = 1000;
-        let continuation_token_opt = continuation_token.map(|s| s.to_string());
+        // Decode the opaque composite token into the (marker, version_marker)
+        // pair that list_object_versions consumes. Malformed tokens restart the
+        // listing from the beginning (decode_heal_token is total).
+        let (marker, version_marker) = decode_heal_token(continuation_token.unwrap_or(""));
 
-        // Use list_objects_v2 to get objects with pagination
+        // Enumerate EVERY version (not just the latest) so old versions and
+        // delete-marker-latest objects are healed too.
         let list_info = match self
             .ecstore
             .clone()
-            .list_objects_v2(bucket, prefix, continuation_token_opt, None, MAX_KEYS, false, None, false)
+            .list_object_versions(bucket, prefix, marker, version_marker, None, MAX_KEYS)
             .await
         {
             Ok(info) => info,
@@ -1166,25 +1324,43 @@ impl HealStorageAPI for ECStoreHealStorage {
             }
         };
 
-        // Collect objects from this page
-        let page_objects: Vec<String> = list_info.objects.into_iter().map(|obj| obj.name).collect();
+        // Collect versions from this page. version_id is normalized to Option<String>
+        // here at the single construction point: nil/absent UUID => None.
+        let page_objects: Vec<HealListItem> = list_info
+            .objects
+            .into_iter()
+            .map(|obj| HealListItem {
+                name: obj.name,
+                version_id: obj.version_id.filter(|u| !u.is_nil()).map(|u| u.to_string()),
+                is_delete_marker: obj.delete_marker,
+            })
+            .collect();
         let page_count = page_objects.len();
 
+        let next_token = if list_info.is_truncated {
+            Some(encode_heal_token(
+                list_info.next_marker.as_deref(),
+                list_info.next_version_idmarker.as_deref(),
+            ))
+        } else {
+            None
+        };
+
         debug!(
             target: "rustfs::heal::storage",
             event = EVENT_HEAL_STORAGE_ADMIN_OP,
             component = LOG_COMPONENT_HEAL,
             subsystem = LOG_SUBSYSTEM_STORAGE,
             operation = "list_objects_for_heal_page",
-                        bucket,
-                        prefix,
-                        object_count = page_count,
-                        is_truncated = list_info.is_truncated,
-                        state = "page_loaded",
-                        "Heal storage object listing page loaded"
+            bucket,
+            prefix,
+            version_count = page_count,
+            is_truncated = list_info.is_truncated,
+            state = "page_loaded",
+            "Heal storage version listing page loaded"
         );
 
-        Ok((page_objects, list_info.next_continuation_token, list_info.is_truncated))
+        Ok((page_objects, next_token, list_info.is_truncated))
     }
 
     async fn get_disk_for_resume(&self, set_disk_id: &str) -> Result<DiskStore> {
@@ -1234,7 +1410,11 @@ impl HealStorageAPI for ECStoreHealStorage {
 #[cfg(test)]
 mod tests {
     use super::super::StorageError;
-    use super::{is_transient_object_exists_error, is_transient_object_exists_message, next_heal_listing_token};
+    use super::{
+        decode_heal_token, encode_heal_token, is_transient_object_exists_error, is_transient_object_exists_message,
+        next_heal_listing_token,
+    };
+    use base64::Engine as _;
 
     #[test]
     fn next_heal_listing_token_returns_none_for_complete_page() {
@@ -1254,11 +1434,51 @@ mod tests {
     }
 
     #[test]
-    fn next_heal_listing_token_fails_for_truncated_page_without_token() {
-        let err = next_heal_listing_token("bucket", "prefix", None, true).expect_err("truncated page without token must fail");
+    fn next_heal_listing_token_treats_truncated_page_without_token_as_end() {
+        // A version listing can report the final page as truncated with no
+        // continuation token; that must terminate the scan cleanly, not error.
+        assert_eq!(
+            next_heal_listing_token("bucket", "prefix", None, true).expect("truncated without token ends listing"),
+            None
+        );
+    }
 
-        assert!(matches!(err, super::Error::TaskExecutionFailed { .. }));
-        assert!(err.to_string().contains("truncated without continuation token"));
+    #[test]
+    fn test_heal_token_roundtrip() {
+        let token = encode_heal_token(Some("obj/key"), Some("v-123"));
+        assert!(token.starts_with("v1:"));
+        assert_eq!(decode_heal_token(&token), (Some("obj/key".to_string()), Some("v-123".to_string())));
+
+        // marker only (no version marker) round-trips.
+        let token = encode_heal_token(Some("obj/key"), None);
+        assert_eq!(decode_heal_token(&token), (Some("obj/key".to_string()), None));
+
+        // (None, None) round-trips.
+        let token = encode_heal_token(None, None);
+        assert_eq!(decode_heal_token(&token), (None, None));
+    }
+
+    #[test]
+    fn test_heal_token_malformed_resets_to_start() {
+        // empty, wrong prefix, bad base64, and bad json all reset to (None, None).
+        assert_eq!(decode_heal_token(""), (None, None));
+        assert_eq!(decode_heal_token("no-prefix-here"), (None, None));
+        assert_eq!(decode_heal_token("v1:!!!not-base64!!!"), (None, None));
+        // valid base64 of non-JSON bytes.
+        let bad_json = format!("v1:{}", base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(b"not json"));
+        assert_eq!(decode_heal_token(&bad_json), (None, None));
+        // a raw v2-style list_objects_v2 token (no "v1:" prefix) resets cleanly.
+        assert_eq!(decode_heal_token("some-opaque-legacy-token"), (None, None));
+    }
+
+    #[test]
+    fn test_heal_token_none_and_marker_only() {
+        // A decoded payload must NEVER yield (None, Some(_)) because
+        // list_object_versions returns NotImplemented for that pairing.
+        // Craft a token whose JSON encodes (None, Some) directly and confirm coercion.
+        let json = br#"{"m":null,"v":"orphan-version"}"#;
+        let token = format!("v1:{}", base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(json));
+        assert_eq!(decode_heal_token(&token), (None, None), "version-only marker must coerce to (None, None)");
     }
 
     #[test]
diff --git a/crates/heal/src/heal/task.rs b/crates/heal/src/heal/task.rs
index a39155773..494d8d217 100644
--- a/crates/heal/src/heal/task.rs
+++ b/crates/heal/src/heal/task.rs
@@ -1315,8 +1315,9 @@ impl HealTask {
                 )
                 .await?;
 
-            for object in objects {
+            for item in objects {
                 self.check_control_flags().await?;
+                let object = item.name.as_str();
                 scanned += 1;
                 {
                     let mut progress = self.progress.write().await;
@@ -1324,109 +1325,95 @@ impl HealTask {
                     progress.update_progress(scanned, healed, failed, bytes);
                 }
 
-                match self.await_with_control(self.storage.object_exists(bucket, &object)).await {
-                    Ok(false) => {
+                // Heal each enumerated version directly. There is no latest-only
+                // existence gate: genuine absence surfaces through heal_object's
+                // not-found result and is handled below.
+                match self
+                    .await_with_control(
+                        self.storage
+                            .heal_object(bucket, object, item.version_id.as_deref(), &heal_opts),
+                    )
+                    .await
+                {
+                    Ok((result, None)) => {
                         healed += 1;
+                        bytes = bytes.saturating_add(result.object_size as u64);
+                        self.record_result_item(result).await;
                     }
-                    Ok(true) => match self
-                        .await_with_control(self.storage.heal_object(bucket, &object, None, &heal_opts))
-                        .await
-                    {
-                        Ok((result, None)) => {
+                    Ok((_, Some(err))) => {
+                        if is_missing_object_dir_heal_result(object, &err) {
                             healed += 1;
-                            bytes = bytes.saturating_add(result.object_size as u64);
-                            self.record_result_item(result).await;
+                            debug!(
+                                target: "rustfs::heal::task",
+                                event = EVENT_HEAL_BUCKET_RESULT,
+                                component = LOG_COMPONENT_HEAL,
+                                subsystem = LOG_SUBSYSTEM_TASK,
+                                task_id = %self.id,
+                                bucket,
+                                object = %object,
+                                result = "object_dir_not_found_skipped",
+                                "Heal bucket object-dir candidate skipped after not-found result"
+                            );
+                            continue;
                         }
-                        Ok((_, Some(err))) => {
-                            if is_missing_object_dir_heal_result(&object, &err) {
-                                healed += 1;
-                                debug!(
-                                    target: "rustfs::heal::task",
-                                    event = EVENT_HEAL_BUCKET_RESULT,
-                                    component = LOG_COMPONENT_HEAL,
-                                    subsystem = LOG_SUBSYSTEM_TASK,
-                                    task_id = %self.id,
-                                    bucket,
-                                    object = %object,
-                                    result = "object_dir_not_found_skipped",
-                                    "Heal bucket object-dir candidate skipped after not-found result"
-                                );
-                                continue;
-                            }
-                            if Self::should_skip_data_usage_cache_heal_error(bucket, &object, &err) {
-                                warn!(
-                                    target: "rustfs::heal::task",
-                                    event = EVENT_HEAL_BUCKET_RESULT,
-                                    component = LOG_COMPONENT_HEAL,
-                                    subsystem = LOG_SUBSYSTEM_TASK,
-                                    task_id = %self.id,
-                                    bucket,
-                                    object = %object,
-                                    result = "transient_skip",
-                                    error = %err,
-                                    "Heal bucket object repair skipped due to transient error"
-                                );
-                            } else {
-                                failed += 1;
-                                warn!(
-                                    target: "rustfs::heal::task",
-                                    event = EVENT_HEAL_BUCKET_RESULT,
-                                    component = LOG_COMPONENT_HEAL,
-                                    subsystem = LOG_SUBSYSTEM_TASK,
-                                    task_id = %self.id,
-                                    bucket,
-                                    object = %object,
-                                    result = "object_failed",
-                                    error = %err,
-                                    "Heal bucket object repair failed"
-                                );
-                            }
+                        if Self::should_skip_data_usage_cache_heal_error(bucket, object, &err) {
+                            warn!(
+                                target: "rustfs::heal::task",
+                                event = EVENT_HEAL_BUCKET_RESULT,
+                                component = LOG_COMPONENT_HEAL,
+                                subsystem = LOG_SUBSYSTEM_TASK,
+                                task_id = %self.id,
+                                bucket,
+                                object = %object,
+                                result = "transient_skip",
+                                error = %err,
+                                "Heal bucket object repair skipped due to transient error"
+                            );
+                        } else {
+                            failed += 1;
+                            warn!(
+                                target: "rustfs::heal::task",
+                                event = EVENT_HEAL_BUCKET_RESULT,
+                                component = LOG_COMPONENT_HEAL,
+                                subsystem = LOG_SUBSYSTEM_TASK,
+                                task_id = %self.id,
+                                bucket,
+                                object = %object,
+                                result = "object_failed",
+                                error = %err,
+                                "Heal bucket object repair failed"
+                            );
                         }
-                        Err(err) => {
-                            if Self::should_skip_data_usage_cache_heal_error(bucket, &object, &err) {
-                                warn!(
-                                    target: "rustfs::heal::task",
-                                    event = EVENT_HEAL_BUCKET_RESULT,
-                                    component = LOG_COMPONENT_HEAL,
-                                    subsystem = LOG_SUBSYSTEM_TASK,
-                                    task_id = %self.id,
-                                    bucket,
-                                    object = %object,
-                                    result = "transient_skip",
-                                    error = %err,
-                                    "Heal bucket object repair skipped due to transient error"
-                                );
-                            } else {
-                                failed += 1;
-                                warn!(
-                                    target: "rustfs::heal::task",
-                                    event = EVENT_HEAL_BUCKET_RESULT,
-                                    component = LOG_COMPONENT_HEAL,
-                                    subsystem = LOG_SUBSYSTEM_TASK,
-                                    task_id = %self.id,
-                                    bucket,
-                                    object = %object,
-                                    result = "object_failed",
-                                    error = %err,
-                                    "Heal bucket object repair failed"
-                                );
-                            }
-                        }
-                    },
+                    }
                     Err(err) => {
-                        failed += 1;
-                        warn!(
-                            target: "rustfs::heal::task",
-                            event = EVENT_HEAL_BUCKET_RESULT,
-                            component = LOG_COMPONENT_HEAL,
-                            subsystem = LOG_SUBSYSTEM_TASK,
-                            task_id = %self.id,
-                            bucket,
-                            object = %object,
-                            result = "precheck_failed",
-                            error = %err,
-                            "Heal bucket object precheck failed"
-                        );
+                        if Self::should_skip_data_usage_cache_heal_error(bucket, object, &err) {
+                            warn!(
+                                target: "rustfs::heal::task",
+                                event = EVENT_HEAL_BUCKET_RESULT,
+                                component = LOG_COMPONENT_HEAL,
+                                subsystem = LOG_SUBSYSTEM_TASK,
+                                task_id = %self.id,
+                                bucket,
+                                object = %object,
+                                result = "transient_skip",
+                                error = %err,
+                                "Heal bucket object repair skipped due to transient error"
+                            );
+                        } else {
+                            failed += 1;
+                            warn!(
+                                target: "rustfs::heal::task",
+                                event = EVENT_HEAL_BUCKET_RESULT,
+                                component = LOG_COMPONENT_HEAL,
+                                subsystem = LOG_SUBSYSTEM_TASK,
+                                task_id = %self.id,
+                                bucket,
+                                object = %object,
+                                result = "object_failed",
+                                error = %err,
+                                "Heal bucket object repair failed"
+                            );
+                        }
                     }
                 }
 
@@ -1441,6 +1428,10 @@ impl HealTask {
             }
 
             continuation_token = next_heal_listing_token(bucket, prefix, next_token, is_truncated)?;
+            if continuation_token.is_none() {
+                // Truncated but no continuation token: end of listing.
+                break;
+            }
         }
 
         if failed > 0 {
@@ -2230,7 +2221,7 @@ impl std::fmt::Debug for HealTask {
 mod tests {
     use super::super::{DiskStore, Endpoint};
     use super::*;
-    use crate::heal::storage::{DiskStatus, HealObjectInfo};
+    use crate::heal::storage::{DiskStatus, HealListItem, HealObjectInfo};
     use rustfs_madmin::heal_commands::HealResultItem;
     use std::collections::HashMap;
     use std::sync::Mutex;
@@ -2254,6 +2245,15 @@ mod tests {
         include_object_dir_candidate: Mutex<bool>,
     }
 
+    /// Build a latest, non-delete-marker heal list item with no version id.
+    fn heal_item(name: &str) -> HealListItem {
+        HealListItem {
+            name: name.to_string(),
+            version_id: None,
+            is_delete_marker: false,
+        }
+    }
+
     enum MockHealObjectOutcome {
         OkWithOtherError(&'static str),
         ErrOther(&'static str),
@@ -2407,8 +2407,8 @@ mod tests {
             }
         }
 
-        async fn list_objects_for_heal(&self, _bucket: &str, _prefix: &str) -> Result<Vec<String>> {
-            Ok(vec!["object-a".to_string(), "object-b".to_string()])
+        async fn list_objects_for_heal(&self, _bucket: &str, _prefix: &str) -> Result<Vec<HealListItem>> {
+            Ok(vec![heal_item("object-a"), heal_item("object-b")])
         }
 
         async fn list_objects_for_heal_page(
@@ -2416,10 +2416,10 @@ mod tests {
             bucket: &str,
             prefix: &str,
             continuation_token: Option<&str>,
-        ) -> Result<(Vec<String>, Option<String>, bool)> {
+        ) -> Result<(Vec<HealListItem>, Option<String>, bool)> {
             self.listed_prefixes.lock().unwrap().push(prefix.to_string());
             if *self.truncate_without_token.lock().unwrap() {
-                return Ok((vec!["object-a".to_string()], None, true));
+                return Ok((vec![heal_item("object-a")], None, true));
             }
 
             let mut listed = self.listed.lock().unwrap();
@@ -2427,15 +2427,15 @@ mod tests {
                 *listed = true;
                 let objects = if bucket == RUSTFS_META_BUCKET {
                     vec![
-                        format!("{BUCKET_META_PREFIX}/{DATA_USAGE_CACHE_NAME}"),
-                        format!("{BUCKET_META_PREFIX}/bucket-metadata.bin"),
+                        heal_item(&format!("{BUCKET_META_PREFIX}/{DATA_USAGE_CACHE_NAME}")),
+                        heal_item(&format!("{BUCKET_META_PREFIX}/bucket-metadata.bin")),
                     ]
                 } else if prefix == "logs/" {
-                    vec!["logs/object-a".to_string(), "logs/object-b".to_string()]
+                    vec![heal_item("logs/object-a"), heal_item("logs/object-b")]
                 } else if *self.include_object_dir_candidate.lock().unwrap() {
-                    vec!["object-a".to_string(), "object-dir/".to_string(), "object-b".to_string()]
+                    vec![heal_item("object-a"), heal_item("object-dir/"), heal_item("object-b")]
                 } else {
-                    vec!["object-a".to_string(), "object-b".to_string()]
+                    vec![heal_item("object-a"), heal_item("object-b")]
                 };
                 Ok((objects, None, false))
             } else {
@@ -2514,7 +2514,11 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn test_recursive_bucket_heal_fails_when_listing_lacks_continuation_token() {
+    async fn test_recursive_bucket_heal_treats_missing_continuation_token_as_end() {
+        // A version listing can report the final page as truncated with no
+        // continuation token. That is treated as end-of-listing (not an error),
+        // so the returned page is healed and the pass terminates cleanly instead
+        // of erroring or looping forever.
         let storage = Arc::new(MockStorage {
             truncate_without_token: Mutex::new(true),
             ..Default::default()
@@ -2532,17 +2536,14 @@ mod tests {
         );
         let task = HealTask::from_request(request, storage.clone());
 
-        let err = task
-            .heal_bucket("bucket-a")
+        task.heal_bucket("bucket-a")
             .await
-            .expect_err("recursive bucket heal must fail on incomplete pagination state");
+            .expect("truncated-without-token must terminate cleanly, not loop or error");
 
-        assert!(matches!(err, Error::TaskExecutionFailed { .. }));
-        assert!(err.to_string().contains("truncated without continuation token"));
         assert_eq!(
             storage.healed_objects.lock().unwrap().as_slice(),
             ["object-a".to_string()],
-            "the already returned page may be processed, but the task must not report success"
+            "the returned page is healed exactly once and the scan ends"
         );
     }
 
diff --git a/crates/heal/tests/heal_b5_versioned_regression_test.rs b/crates/heal/tests/heal_b5_versioned_regression_test.rs
new file mode 100644
index 000000000..6d53a513b
--- /dev/null
+++ b/crates/heal/tests/heal_b5_versioned_regression_test.rs
@@ -0,0 +1,683 @@
+// 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.
+
+//! B5-2 (rustfs/backlog#919): real-disk-wipe e2e regression suite proving the
+//! B5-1 version-aware heal enumeration physically repairs OLD non-latest
+//! versions and DELETE-MARKER-latest objects.
+//!
+//! These drive the REAL `ECStoreHealStorage` (not a mock) against a real 4-disk
+//! `ECStore`, mirroring `heal_integration_test.rs`. Every test is `#[serial]`
+//! and re-runs the full `setup_test_env` init (which sets the process-global
+//! bucket-metadata-sys OnceCell) — under `cargo nextest` each test runs
+//! in its own process so the OnceCell never collides.
+
+use http::HeaderMap;
+use rustfs_common::heal_channel::{HealOpts, HealScanMode};
+use rustfs_heal::heal::{
+    manager::{HealConfig, HealManager},
+    storage::{
+        ECStoreHealStorage, HealListItem, HealObjectOptions as ObjectOptions, HealPutObjReader as PutObjReader, HealStorageAPI,
+    },
+    task::{HealOptions, HealPriority, HealRequest, HealTaskStatus, HealType},
+};
+use serial_test::serial;
+use std::{
+    path::{Path, PathBuf},
+    sync::{Arc, Once},
+    time::Duration,
+};
+use tokio::fs;
+use tokio_util::sync::CancellationToken;
+use tracing::info;
+use walkdir::WalkDir;
+
+mod storage_api;
+
+use storage_api::integration::{
+    BucketOperations, BucketOptions, ECStore, Endpoint, EndpointServerPools, Endpoints, MakeBucketOptions, ObjectIO as _,
+    ObjectOperations as _, PoolEndpoints, init_bucket_metadata_sys, init_local_disks,
+};
+
+/// 256 KiB + change: large enough to be stored as non-inline erasure shards
+/// (so each data version materializes as an on-disk `part.*` file we can assert
+/// was physically restored on the wiped disk).
+const NON_INLINE_TEST_DATA_SIZE: usize = 256 * 1024 + 137;
+
+const SET_DISK_ID: &str = "pool_0_set_0";
+
+fn versioned_test_data(seed: u8) -> Vec<u8> {
+    (0..NON_INLINE_TEST_DATA_SIZE)
+        .map(|idx| ((idx + seed as usize) % 251) as u8)
+        .collect()
+}
+
+static INIT: Once = Once::new();
+
+fn init_tracing() {
+    INIT.call_once(|| {
+        let _ = tracing_subscriber::fmt()
+            .with_env_filter(tracing_subscriber::EnvFilter::from_default_env())
+            .with_timer(tracing_subscriber::fmt::time::UtcTime::rfc_3339())
+            .with_thread_names(true)
+            .try_init();
+    });
+}
+
+/// Build a real 4-disk `ECStore` + `ECStoreHealStorage`. Mirrors
+/// `heal_integration_test::setup_test_env`.
+async fn setup_test_env() -> (Vec<PathBuf>, Arc<ECStore>, Arc<ECStoreHealStorage>) {
+    init_tracing();
+
+    let test_base_dir = format!("/tmp/rustfs_heal_b5_test_{}", uuid::Uuid::new_v4());
+    let temp_dir = PathBuf::from(&test_base_dir);
+    if temp_dir.exists() {
+        fs::remove_dir_all(&temp_dir).await.ok();
+    }
+    fs::create_dir_all(&temp_dir).await.unwrap();
+
+    let disk_paths = vec![
+        temp_dir.join("disk1"),
+        temp_dir.join("disk2"),
+        temp_dir.join("disk3"),
+        temp_dir.join("disk4"),
+    ];
+    for disk_path in &disk_paths {
+        fs::create_dir_all(disk_path).await.unwrap();
+    }
+
+    let mut endpoints = Vec::new();
+    for (i, disk_path) in disk_paths.iter().enumerate() {
+        let mut endpoint = Endpoint::try_from(disk_path.to_str().unwrap()).unwrap();
+        endpoint.set_pool_index(0);
+        endpoint.set_set_index(0);
+        endpoint.set_disk_index(i);
+        endpoints.push(endpoint);
+    }
+
+    let pool_endpoints = PoolEndpoints {
+        legacy: false,
+        set_count: 1,
+        drives_per_set: 4,
+        endpoints: Endpoints::from(endpoints),
+        cmd_line: "test".to_string(),
+        platform: format!("OS: {} | Arch: {}", std::env::consts::OS, std::env::consts::ARCH),
+    };
+    let endpoint_pools = EndpointServerPools::from(vec![pool_endpoints]);
+
+    init_local_disks(endpoint_pools.clone()).await.unwrap();
+
+    let server_addr: std::net::SocketAddr = "127.0.0.1:0".parse().unwrap();
+    let ecstore = ECStore::new(server_addr, endpoint_pools, CancellationToken::new())
+        .await
+        .unwrap();
+
+    let buckets_list = ecstore
+        .list_bucket(&BucketOptions {
+            no_metadata: true,
+            ..Default::default()
+        })
+        .await
+        .unwrap();
+    let buckets = buckets_list.into_iter().map(|v| v.name).collect();
+    init_bucket_metadata_sys(ecstore.clone(), buckets).await;
+
+    let heal_storage = Arc::new(ECStoreHealStorage::new(ecstore.clone()));
+    (disk_paths, ecstore, heal_storage)
+}
+
+/// Create a bucket with S3 versioning ENABLED at creation time. Without this the
+/// second PUT overwrites in place and DELETE removes the object outright — no
+/// old versions and no delete-marker-latest would ever exist (empty-pass trap).
+async fn create_versioned_bucket(ecstore: &Arc<ECStore>, bucket: &str) {
+    (**ecstore)
+        .make_bucket(
+            bucket,
+            &MakeBucketOptions {
+                versioning_enabled: true,
+                ..Default::default()
+            },
+        )
+        .await
+        .expect("failed to create versioned bucket");
+}
+
+async fn create_unversioned_bucket(ecstore: &Arc<ECStore>, bucket: &str) {
+    (**ecstore)
+        .make_bucket(bucket, &MakeBucketOptions::default())
+        .await
+        .expect("failed to create unversioned bucket");
+}
+
+/// PUT a new version (versioned:true forces a fresh version id per write) and
+/// return the created version id as a String.
+async fn put_versioned(ecstore: &Arc<ECStore>, bucket: &str, object: &str, data: &[u8]) -> String {
+    let mut reader = PutObjReader::from_vec(data.to_vec());
+    let opts = ObjectOptions {
+        versioned: true,
+        ..Default::default()
+    };
+    let info = (**ecstore)
+        .put_object(bucket, object, &mut reader, &opts)
+        .await
+        .expect("versioned put_object failed");
+    info.version_id
+        .map(|u| u.to_string())
+        .expect("versioned put must return a version id")
+}
+
+async fn put_unversioned(ecstore: &Arc<ECStore>, bucket: &str, object: &str, data: &[u8]) {
+    let mut reader = PutObjReader::from_vec(data.to_vec());
+    (**ecstore)
+        .put_object(bucket, object, &mut reader, &ObjectOptions::default())
+        .await
+        .expect("unversioned put_object failed");
+}
+
+/// Create a delete-marker as the latest version (versioned:true, no version_id)
+/// and return the delete-marker's version id.
+async fn put_delete_marker(ecstore: &Arc<ECStore>, bucket: &str, object: &str) -> String {
+    let opts = ObjectOptions {
+        versioned: true,
+        ..Default::default()
+    };
+    let info = (**ecstore)
+        .delete_object(bucket, object, opts)
+        .await
+        .expect("versioned delete (delete-marker) failed");
+    assert!(info.delete_marker, "delete on a versioned bucket must yield a delete marker");
+    info.version_id
+        .map(|u| u.to_string())
+        .expect("delete marker must carry a version id")
+}
+
+/// On-disk object directory for a given disk: `<disk>/<bucket>/<object>/`.
+fn object_dir(disk: &Path, bucket: &str, object: &str) -> PathBuf {
+    disk.join(bucket).join(object)
+}
+
+/// Count `part.*` data-shard files two levels below the object dir
+/// (`<object>/<data-uuid>/part.N`). One data dir per non-delete-marker version.
+fn count_part_files(obj_dir: &Path) -> usize {
+    if !obj_dir.exists() {
+        return 0;
+    }
+    WalkDir::new(obj_dir)
+        .min_depth(2)
+        .max_depth(2)
+        .into_iter()
+        .filter_map(Result::ok)
+        .filter(|e| e.file_type().is_file() && e.file_name().to_str().map(|n| n.starts_with("part.")).unwrap_or(false))
+        .count()
+}
+
+fn xl_meta_path(obj_dir: &Path) -> PathBuf {
+    obj_dir.join("xl.meta")
+}
+
+fn recreate_heal_opts() -> HealOpts {
+    // Mirrors the proven-working object heal opts in
+    // `heal_integration_test::test_heal_format_with_data`.
+    HealOpts {
+        recreate: true,
+        remove: false,
+        ..Default::default()
+    }
+}
+
+/// Enumerate every version via the REAL B5-1 paged listing, walking all pages.
+async fn enumerate_all_versions(heal_storage: &Arc<ECStoreHealStorage>, bucket: &str) -> Vec<HealListItem> {
+    let mut items = Vec::new();
+    let mut token: Option<String> = None;
+    loop {
+        let (page, next, truncated) = heal_storage
+            .list_objects_for_heal_page(bucket, "", token.as_deref())
+            .await
+            .expect("list_objects_for_heal_page failed");
+        items.extend(page);
+        if !truncated {
+            break;
+        }
+        token = next;
+        if token.is_none() {
+            break;
+        }
+    }
+    items
+}
+
+/// Heal every enumerated version through the REAL `ECStoreHealStorage`, mirroring
+/// the production per-version loop. Returns (healed_ok, failed).
+async fn heal_all_versions(heal_storage: &Arc<ECStoreHealStorage>, bucket: &str) -> (usize, usize) {
+    let items = enumerate_all_versions(heal_storage, bucket).await;
+    let opts = recreate_heal_opts();
+    let mut healed = 0usize;
+    let mut failed = 0usize;
+    for item in items {
+        let (result, error) = heal_storage
+            .heal_object(bucket, &item.name, item.version_id.as_deref(), &opts)
+            .await
+            .expect("heal_object call itself must not error out");
+        if error.is_some() {
+            failed += 1;
+            info!(
+                "heal_object reported error for {}/{} v={:?} dm={}: {:?}",
+                bucket, item.name, item.version_id, item.is_delete_marker, error
+            );
+        } else {
+            healed += 1;
+        }
+        let _ = result;
+    }
+    (healed, failed)
+}
+
+async fn read_version(ecstore: &Arc<ECStore>, bucket: &str, object: &str, version_id: &str) -> Vec<u8> {
+    let opts = ObjectOptions {
+        version_id: Some(version_id.to_string()),
+        ..Default::default()
+    };
+    let mut reader = ecstore
+        .get_object_reader(bucket, object, None, HeaderMap::new(), &opts)
+        .await
+        .expect("failed to open version reader");
+    let mut buf = Vec::new();
+    tokio::io::copy(&mut reader, &mut buf)
+        .await
+        .expect("failed to read version data");
+    buf
+}
+
+mod serial_tests {
+    use super::*;
+
+    /// Directly exercises `ECStoreHealStorage::list_objects_for_heal_page` on a
+    /// real versioned fixture: two data versions + a delete-marker-latest. Proves
+    /// enumeration returns one `HealListItem` per version, flags the delete marker
+    /// via `is_delete_marker`, and carries the correct version ids.
+    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+    #[serial]
+    async fn test_enumeration_includes_all_versions_and_delete_marker_real_fixture() {
+        let (_disk_paths, ecstore, heal_storage) = setup_test_env().await;
+        let bucket = "b5-enum-versions";
+        let object = "obj.bin";
+        create_versioned_bucket(&ecstore, bucket).await;
+
+        let v1 = put_versioned(&ecstore, bucket, object, &versioned_test_data(1)).await;
+        let v2 = put_versioned(&ecstore, bucket, object, &versioned_test_data(2)).await;
+        let dm = put_delete_marker(&ecstore, bucket, object).await;
+        assert_ne!(v1, v2);
+        assert_ne!(v2, dm);
+
+        let items = enumerate_all_versions(&heal_storage, bucket).await;
+
+        // FIXTURE-NON-EMPTY GUARD: exactly the three versions we created.
+        assert_eq!(items.len(), 3, "must enumerate every version + the delete marker, got {items:?}");
+        assert!(items.iter().all(|it| it.name == object), "all entries belong to the same object");
+
+        let dm_items: Vec<&HealListItem> = items.iter().filter(|it| it.is_delete_marker).collect();
+        assert_eq!(dm_items.len(), 1, "exactly one entry must be flagged as a delete marker");
+        assert_eq!(
+            dm_items[0].version_id.as_deref(),
+            Some(dm.as_str()),
+            "the delete-marker entry must carry the delete marker's version id"
+        );
+
+        let ids: std::collections::HashSet<Option<String>> = items.iter().map(|it| it.version_id.clone()).collect();
+        assert!(ids.contains(&Some(v1.clone())), "old version v1 must be enumerated");
+        assert!(ids.contains(&Some(v2.clone())), "version v2 must be enumerated");
+        assert!(ids.contains(&Some(dm.clone())), "delete marker version must be enumerated");
+        // Every version id is a real (non-nil) id, so none normalized to None.
+        assert!(items.iter().all(|it| it.version_id.is_some()), "versioned entries must keep their ids");
+    }
+
+    /// Physical repair of an OLD non-latest version after wiping one disk.
+    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+    #[serial]
+    async fn test_heal_old_nonlatest_version_after_disk_wipe() {
+        let (disk_paths, ecstore, heal_storage) = setup_test_env().await;
+        let bucket = "b5-old-version-heal";
+        let object = "obj.bin";
+        create_versioned_bucket(&ecstore, bucket).await;
+
+        let data_v1 = versioned_test_data(10);
+        let data_v2 = versioned_test_data(20);
+        let v1 = put_versioned(&ecstore, bucket, object, &data_v1).await; // OLD, non-latest
+        let v2 = put_versioned(&ecstore, bucket, object, &data_v2).await; // latest
+
+        // ── Pre-wipe: prove the fixture actually has 2 versions on disk[0] ──
+        let obj_dir0 = object_dir(&disk_paths[0], bucket, object);
+        assert!(xl_meta_path(&obj_dir0).exists(), "xl.meta must exist before wipe");
+        assert_eq!(
+            count_part_files(&obj_dir0),
+            2,
+            "two data versions must each have an on-disk shard before wipe"
+        );
+        let pre_items = enumerate_all_versions(&heal_storage, bucket).await;
+        assert_eq!(pre_items.len(), 2, "fixture must expose both versions");
+
+        // ── Wipe disk[0]'s object dir (xl.meta + both data dirs) ──
+        std::fs::remove_dir_all(&obj_dir0).expect("failed to wipe object dir on disk[0]");
+        assert!(!obj_dir0.exists(), "object dir must be gone after wipe");
+        assert_eq!(count_part_files(&obj_dir0), 0, "no shards on disk[0] after wipe");
+
+        // ── Heal every enumerated version through the real heal storage ──
+        let (healed, failed) = heal_all_versions(&heal_storage, bucket).await;
+        assert_eq!(failed, 0, "no version may be recorded as failed");
+        assert!(healed >= 2, "both versions must be healed, healed={healed}");
+
+        // ── Post-heal: disk[0] physically restored, incl. the OLD version ──
+        assert!(xl_meta_path(&obj_dir0).exists(), "xl.meta must be restored on the wiped disk");
+        assert_eq!(
+            count_part_files(&obj_dir0),
+            2,
+            "both versions' data shards must be physically restored on disk[0]"
+        );
+
+        // Old non-latest version data must be intact and readable end-to-end.
+        assert_eq!(read_version(&ecstore, bucket, object, &v1).await, data_v1, "old version data corrupted");
+        assert_eq!(
+            read_version(&ecstore, bucket, object, &v2).await,
+            data_v2,
+            "latest version data corrupted"
+        );
+    }
+
+    /// Delete-marker-latest objects must be enumerated AND healed after a wipe.
+    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+    #[serial]
+    async fn test_heal_delete_marker_latest_enumerated_and_healed() {
+        let (disk_paths, ecstore, heal_storage) = setup_test_env().await;
+        let bucket = "b5-dm-latest-heal";
+        let object = "obj.bin";
+        create_versioned_bucket(&ecstore, bucket).await;
+
+        let data_v1 = versioned_test_data(30);
+        let v1 = put_versioned(&ecstore, bucket, object, &data_v1).await; // data version
+        let dm = put_delete_marker(&ecstore, bucket, object).await; // delete-marker latest
+
+        // ── Pre-wipe fixture proof: one data shard + a delete-marker latest ──
+        let obj_dir0 = object_dir(&disk_paths[0], bucket, object);
+        assert!(xl_meta_path(&obj_dir0).exists(), "xl.meta must exist before wipe");
+        assert_eq!(count_part_files(&obj_dir0), 1, "one data version => one shard before wipe");
+        let pre_items = enumerate_all_versions(&heal_storage, bucket).await;
+        assert_eq!(pre_items.len(), 2, "must enumerate the data version and the delete marker");
+        assert!(
+            pre_items
+                .iter()
+                .any(|it| it.is_delete_marker && it.version_id.as_deref() == Some(dm.as_str())),
+            "the delete-marker-latest must be enumerated as a heal unit"
+        );
+        // The latest, per get_object_info, is a delete marker (object reads as deleted).
+        let latest = ecstore.get_object_info(bucket, object, &ObjectOptions::default()).await;
+        assert!(
+            matches!(&latest, Ok(info) if info.delete_marker) || latest.is_err(),
+            "latest must resolve to a delete marker before heal"
+        );
+
+        // ── Wipe disk[0]'s object dir ──
+        std::fs::remove_dir_all(&obj_dir0).expect("failed to wipe object dir on disk[0]");
+        assert_eq!(count_part_files(&obj_dir0), 0, "no shards on disk[0] after wipe");
+
+        // ── Heal all enumerated versions (data version + delete marker) ──
+        let (healed, failed) = heal_all_versions(&heal_storage, bucket).await;
+        assert_eq!(failed, 0, "delete marker + data version must not be recorded as failed");
+        assert!(healed >= 2, "both the delete marker and the data version must heal, healed={healed}");
+
+        // ── Post-heal: xl.meta (with the tombstone) + the data shard restored ──
+        assert!(
+            xl_meta_path(&obj_dir0).exists(),
+            "xl.meta (carrying the delete-marker tombstone) must be restored on disk[0]"
+        );
+        assert_eq!(
+            count_part_files(&obj_dir0),
+            1,
+            "the underlying data version's shard must be physically restored on disk[0]"
+        );
+
+        // The old data version is still readable by id; the latest is still a DM.
+        assert_eq!(
+            read_version(&ecstore, bucket, object, &v1).await,
+            data_v1,
+            "healed data version corrupted"
+        );
+        let latest_after = ecstore.get_object_info(bucket, object, &ObjectOptions::default()).await;
+        assert!(
+            matches!(&latest_after, Ok(info) if info.delete_marker) || latest_after.is_err(),
+            "latest must remain a delete marker after heal"
+        );
+    }
+
+    /// Unversioned objects normalize to `version_id == None` and enumerate once
+    /// each (never a phantom `Some("null")`).
+    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+    #[serial]
+    async fn test_version_id_normalization_null_and_unversioned_real_fixture() {
+        let (_disk_paths, ecstore, heal_storage) = setup_test_env().await;
+        let bucket = "b5-unversioned-normalize";
+        create_unversioned_bucket(&ecstore, bucket).await;
+
+        put_unversioned(&ecstore, bucket, "a.bin", &versioned_test_data(1)).await;
+        put_unversioned(&ecstore, bucket, "b.bin", &versioned_test_data(2)).await;
+        // Overwrite b.bin: on an unversioned bucket this replaces in place, so
+        // there is still exactly one enumerable unit for it.
+        put_unversioned(&ecstore, bucket, "b.bin", &versioned_test_data(3)).await;
+
+        let items = enumerate_all_versions(&heal_storage, bucket).await;
+        assert_eq!(items.len(), 2, "unversioned bucket => exactly one heal unit per object, got {items:?}");
+        assert!(
+            items.iter().all(|it| it.version_id.is_none()),
+            "unversioned objects must normalize to version_id == None, never Some(\"null\"): {items:?}"
+        );
+        assert!(items.iter().all(|it| !it.is_delete_marker), "no delete markers expected");
+        let names: std::collections::HashSet<&str> = items.iter().map(|it| it.name.as_str()).collect();
+        assert!(names.contains("a.bin") && names.contains("b.bin"), "both objects enumerated once");
+
+        // NOTE: A literal MinIO-interop "null" *string* version id is only minted
+        // by the S3 request layer (versioning-suspended writes); it cannot be
+        // constructed through the internal ECStore put path used here (suspended/
+        // unversioned writes store no version id at all -> None). The nil-UUID ->
+        // None normalization itself is unit-covered in B5-1
+        // (crates/heal storage list_objects_for_heal_page maps
+        // `version_id.filter(|u| !u.is_nil())`). This test covers the real
+        // unversioned-object => None path end-to-end.
+    }
+
+    /// Unversioned bucket, full heal path via `HealManager` (recursive bucket
+    /// heal), after wiping one disk. Every object is healed exactly once and
+    /// remains readable; nothing is dropped or double-processed.
+    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+    #[serial]
+    async fn test_heal_unversioned_bucket_e2e() {
+        let (disk_paths, ecstore, heal_storage) = setup_test_env().await;
+        let bucket = "b5-unversioned-e2e";
+        create_unversioned_bucket(&ecstore, bucket).await;
+
+        let objects = ["o1.bin", "o2.bin", "o3.bin"];
+        let mut datas = Vec::new();
+        for (i, obj) in objects.iter().enumerate() {
+            let data = versioned_test_data(i as u8 + 40);
+            put_unversioned(&ecstore, bucket, obj, &data).await;
+            datas.push(data);
+        }
+
+        // Wipe every object dir on disk[0].
+        for obj in &objects {
+            let dir = object_dir(&disk_paths[0], bucket, obj);
+            std::fs::remove_dir_all(&dir).expect("failed to wipe object dir");
+            assert_eq!(count_part_files(&dir), 0);
+        }
+
+        // Enumeration returns exactly one unit per object (no duplicates).
+        let items = enumerate_all_versions(&heal_storage, bucket).await;
+        assert_eq!(items.len(), objects.len(), "one heal unit per object");
+        assert!(items.iter().all(|it| it.version_id.is_none()));
+
+        // Drive the real recursive bucket heal through the HealManager task loop.
+        let cfg = HealConfig {
+            heal_interval: Duration::from_millis(1),
+            ..Default::default()
+        };
+        let heal_manager = HealManager::new(heal_storage.clone(), Some(cfg));
+        heal_manager.start().await.unwrap();
+        let request = HealRequest::new(
+            HealType::Bucket {
+                bucket: bucket.to_string(),
+            },
+            HealOptions {
+                recursive: true,
+                recreate_missing: true,
+                scan_mode: HealScanMode::Normal,
+                timeout: Some(Duration::from_secs(300)),
+                ..Default::default()
+            },
+            HealPriority::Normal,
+        );
+        let task_id = request.id.clone();
+        let admission = heal_manager
+            .submit_heal_request(request)
+            .await
+            .expect("failed to submit unversioned bucket heal");
+        assert!(admission.is_admitted());
+
+        wait_for_task(&heal_manager, &task_id, Duration::from_secs(60)).await;
+
+        // Every object is physically restored on disk[0] and reads back intact.
+        for (obj, data) in objects.iter().zip(datas.iter()) {
+            let dir = object_dir(&disk_paths[0], bucket, obj);
+            assert!(xl_meta_path(&dir).exists(), "{obj}: xl.meta not restored on disk[0]");
+            assert_eq!(count_part_files(&dir), 1, "{obj}: data shard not restored on disk[0]");
+            let mut reader = ecstore
+                .get_object_reader(bucket, obj, None, HeaderMap::new(), &ObjectOptions::default())
+                .await
+                .expect("healed object must be readable");
+            let mut buf = Vec::new();
+            tokio::io::copy(&mut reader, &mut buf).await.unwrap();
+            assert_eq!(&buf, data, "{obj}: healed data mismatch");
+        }
+    }
+
+    /// End-to-end drive of the REAL resume/checkpoint-backed erasure-set healer
+    /// (`ErasureSet` heal type -> `ErasureSetHealer::heal_bucket_with_resume`)
+    /// against real disks with a multi-version fixture across a wiped disk.
+    ///
+    /// SCOPE NOTE: literally crossing the 1000-key `list_object_versions` page
+    /// boundary at e2e scale (1000+ real versions) is impractical for a unit-speed
+    /// test, so that exact boundary + mid-page cancel/resume behavior is covered
+    /// by the B5-1 in-crate loop tests
+    /// (`erasure_healer::resume_loop_tests::test_resume_across_page_boundary_no_drop_no_double`,
+    /// `test_object_with_versions_spanning_pages_advances`,
+    /// `test_non_advancing_cursor_aborts`,
+    /// `test_schedule_retry_resets_both_managers_and_reheals`). Here we prove the
+    /// same resume machinery drives a real ECStore heal to completion and repairs
+    /// every version, with the resume state cleaned up afterward.
+    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+    #[serial]
+    async fn test_heal_resume_across_page_boundary_e2e() {
+        let (disk_paths, ecstore, heal_storage) = setup_test_env().await;
+        let bucket = "b5-resume-e2e";
+        create_versioned_bucket(&ecstore, bucket).await;
+
+        // A small multi-object, multi-version fixture: 3 objects, 2 versions each,
+        // plus a delete-marker-latest on one of them.
+        let mut versions: Vec<(String, String, Vec<u8>)> = Vec::new(); // (object, version_id, data)
+        for obj_idx in 0..3u8 {
+            let object = format!("obj-{obj_idx}.bin");
+            let d1 = versioned_test_data(obj_idx + 50);
+            let d2 = versioned_test_data(obj_idx + 80);
+            let v1 = put_versioned(&ecstore, bucket, &object, &d1).await;
+            let v2 = put_versioned(&ecstore, bucket, &object, &d2).await;
+            versions.push((object.clone(), v1, d1));
+            versions.push((object.clone(), v2, d2));
+        }
+        // Delete-marker latest on obj-0.
+        let _dm = put_delete_marker(&ecstore, bucket, "obj-0.bin").await;
+
+        // Wipe disk[0]'s bucket dir entirely (all objects' shards + metadata).
+        let bucket_dir0 = disk_paths[0].join(bucket);
+        assert!(bucket_dir0.exists());
+        std::fs::remove_dir_all(&bucket_dir0).expect("failed to wipe bucket dir on disk[0]");
+        assert!(!bucket_dir0.exists());
+
+        // Drive the resume-backed erasure-set heal through the HealManager, which
+        // constructs the ErasureSetHealer and runs heal_bucket_with_resume.
+        let cfg = HealConfig {
+            heal_interval: Duration::from_millis(1),
+            ..Default::default()
+        };
+        let heal_manager = HealManager::new(heal_storage.clone(), Some(cfg));
+        heal_manager.start().await.unwrap();
+        let request = HealRequest::new(
+            HealType::ErasureSet {
+                buckets: vec![bucket.to_string()],
+                set_disk_id: SET_DISK_ID.to_string(),
+            },
+            HealOptions {
+                recursive: true,
+                recreate_missing: true,
+                scan_mode: HealScanMode::Normal,
+                timeout: Some(Duration::from_secs(300)),
+                ..Default::default()
+            },
+            HealPriority::Normal,
+        );
+        let task_id = request.id.clone();
+        let admission = heal_manager
+            .submit_heal_request(request)
+            .await
+            .expect("failed to submit erasure-set heal");
+        assert!(admission.is_admitted(), "erasure-set heal must be admitted");
+
+        wait_for_task(&heal_manager, &task_id, Duration::from_secs(120)).await;
+
+        // Every data version is readable end-to-end and physically restored on the
+        // wiped disk. (Resume machinery drove the full per-version heal.)
+        for (object, version_id, data) in &versions {
+            assert_eq!(
+                &read_version(&ecstore, bucket, object, version_id).await,
+                data,
+                "{object} v={version_id}: data not restored after resume-backed heal"
+            );
+            let dir = object_dir(&disk_paths[0], bucket, object);
+            assert!(xl_meta_path(&dir).exists(), "{object}: xl.meta not restored on disk[0]");
+        }
+        // Each object should have both of its data versions' shards back on disk[0].
+        for obj_idx in 0..3u8 {
+            let dir = object_dir(&disk_paths[0], bucket, &format!("obj-{obj_idx}.bin"));
+            assert_eq!(
+                count_part_files(&dir),
+                2,
+                "obj-{obj_idx}: both data versions' shards must be restored on disk[0]"
+            );
+        }
+    }
+
+    /// Poll a heal task to a terminal state, panicking on failure/timeout.
+    async fn wait_for_task(heal_manager: &HealManager, task_id: &str, timeout: Duration) {
+        let deadline = tokio::time::Instant::now() + timeout;
+        loop {
+            if let Ok(status) = heal_manager.get_task_status(task_id).await {
+                match status {
+                    HealTaskStatus::Completed => return,
+                    HealTaskStatus::Failed { ref error } => panic!("heal task failed: {error}"),
+                    HealTaskStatus::Cancelled => panic!("heal task was cancelled"),
+                    _ => {}
+                }
+            }
+            if tokio::time::Instant::now() >= deadline {
+                panic!("heal task {task_id} did not complete within {timeout:?}");
+            }
+            tokio::time::sleep(Duration::from_millis(100)).await;
+        }
+    }
+}
diff --git a/crates/heal/tests/heal_bug_fixes_test.rs b/crates/heal/tests/heal_bug_fixes_test.rs
index a543f50fb..7b8737f19 100644
--- a/crates/heal/tests/heal_bug_fixes_test.rs
+++ b/crates/heal/tests/heal_bug_fixes_test.rs
@@ -162,7 +162,7 @@ fn test_path_to_str_helper() {
 
 #[test]
 fn test_heal_task_status_atomic_update() {
-    use rustfs_heal::heal::storage::{HealObjectInfo, HealStorageAPI};
+    use rustfs_heal::heal::storage::{HealListItem, HealObjectInfo, HealStorageAPI};
     use rustfs_heal::heal::task::{HealOptions, HealRequest, HealTask, HealTaskStatus};
     use std::sync::Arc;
 
@@ -234,7 +234,7 @@ fn test_heal_task_status_atomic_update() {
         ) -> rustfs_heal::Result<(rustfs_madmin::heal_commands::HealResultItem, Option<rustfs_heal::Error>)> {
             Ok((rustfs_madmin::heal_commands::HealResultItem::default(), None))
         }
-        async fn list_objects_for_heal(&self, _bucket: &str, _prefix: &str) -> rustfs_heal::Result<Vec<String>> {
+        async fn list_objects_for_heal(&self, _bucket: &str, _prefix: &str) -> rustfs_heal::Result<Vec<HealListItem>> {
             Ok(vec![])
         }
         async fn list_objects_for_heal_page(
@@ -242,7 +242,7 @@ fn test_heal_task_status_atomic_update() {
             _bucket: &str,
             _prefix: &str,
             _continuation_token: Option<&str>,
-        ) -> rustfs_heal::Result<(Vec<String>, Option<String>, bool)> {
+        ) -> rustfs_heal::Result<(Vec<HealListItem>, Option<String>, bool)> {
             Ok((vec![], None, false))
         }
         async fn get_disk_for_resume(&self, _set_disk_id: &str) -> rustfs_heal::Result<DiskStore> {
@@ -277,7 +277,7 @@ fn test_heal_task_status_atomic_update() {
 
 #[tokio::test]
 async fn test_heal_task_transient_object_exists_skip_avoids_recreate() {
-    use rustfs_heal::heal::storage::{DiskStatus, HealObjectInfo, HealStorageAPI};
+    use rustfs_heal::heal::storage::{DiskStatus, HealListItem, HealObjectInfo, HealStorageAPI};
     use rustfs_heal::heal::task::{HealOptions, HealPriority, HealRequest, HealTask, HealTaskStatus, HealType};
     use std::sync::{
         Arc,
@@ -376,7 +376,7 @@ async fn test_heal_task_transient_object_exists_skip_avoids_recreate() {
             Ok((rustfs_madmin::heal_commands::HealResultItem::default(), None))
         }
 
-        async fn list_objects_for_heal(&self, _bucket: &str, _prefix: &str) -> rustfs_heal::Result<Vec<String>> {
+        async fn list_objects_for_heal(&self, _bucket: &str, _prefix: &str) -> rustfs_heal::Result<Vec<HealListItem>> {
             Ok(Vec::new())
         }
 
@@ -385,7 +385,7 @@ async fn test_heal_task_transient_object_exists_skip_avoids_recreate() {
             _bucket: &str,
             _prefix: &str,
             _continuation_token: Option<&str>,
-        ) -> rustfs_heal::Result<(Vec<String>, Option<String>, bool)> {
+        ) -> rustfs_heal::Result<(Vec<HealListItem>, Option<String>, bool)> {
             Ok((Vec::new(), None, false))
         }
 
diff --git a/crates/heal/tests/storage_api.rs b/crates/heal/tests/storage_api.rs
index 308e9d114..e0a01f801 100644
--- a/crates/heal/tests/storage_api.rs
+++ b/crates/heal/tests/storage_api.rs
@@ -26,6 +26,7 @@ pub(crate) mod integration {
     pub(crate) use rustfs_ecstore::api::storage::init_local_disks;
     pub(crate) use rustfs_storage_api::BucketOperations;
     pub(crate) use rustfs_storage_api::BucketOptions;
+    pub(crate) use rustfs_storage_api::MakeBucketOptions;
     pub(crate) use rustfs_storage_api::ObjectIO;
     pub(crate) use rustfs_storage_api::ObjectOperations;
 }