fix(heal): persist explicit delete-marker purges (#7908)

* fix(heal): persist explicit delete-marker purges

* fix(ecstore): remove unused object info wrapper

* test(rpc): use valid incarnation in digest gate

---------

Signed-off-by: Hauser <housemecn@gmail.com>
Co-authored-by: Hauser <housemecn@gmail.com>

crates/common/src/mrf_channel.rs

@@ -22,8 +22,9 @@
 //! hint is always preferred over stalling an IO path. Durable replay of
 //! unconsumed intents is the consumer's job (see `rustfs-heal`
 //! `heal::mrf_queue`), mirroring MinIO's `.heal/mrf/list.bin`. Committed partial
-//! writes use a separate bounded channel and await a checkpoint receipt from
-//! [`persist_partial_write_intent`]; ordinary complete writes do not use it.
+//! writes and committed partial delete-marker purges use a separate bounded
+//! channel and await a checkpoint receipt; ordinary complete writes do not use
+//! it.
 
 use std::collections::HashMap;
 use std::collections::hash_map::RandomState;
@@ -46,10 +47,10 @@ const MRF_COALESCER_MAX_KEYS: usize = 8192;
 const MRF_COALESCER_MAX_BYTES: usize = 16 * 1024 * 1024;
 const MRF_COALESCER_TTL: Duration = Duration::from_secs(60);
 const MRF_MAX_IDENTITY_COMPONENT: usize = 1024;
+pub const MRF_MAX_DELETE_MARKER_BYTES: usize = 1024 * 1024;
 
 /// Why an intent was produced. Drives the heal priority mapping on the
-/// consumer side (DecodeFailure -> Urgent, MetadataCorruption -> High,
-/// PartialWrite -> Normal).
+/// consumer side.
 #[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
 pub enum MrfKind {
     /// Erasure decode failed while serving a read (read path).
@@ -58,6 +59,8 @@ pub enum MrfKind {
     MetadataCorruption,
     /// A write left the object with fewer committed shards than the set size.
     PartialWrite,
+    /// A committed explicit delete-marker purge missed at least one member.
+    DeleteMarkerPurge,
 }
 
 impl MrfKind {
@@ -66,8 +69,60 @@ impl MrfKind {
             MrfKind::DecodeFailure => "decode-failure",
             MrfKind::MetadataCorruption => "metadata-corruption",
             MrfKind::PartialWrite => "partial-write",
+            MrfKind::DeleteMarkerPurge => "delete-marker-purge",
         }
     }
+
+    pub const fn is_durable(self) -> bool {
+        matches!(self, Self::PartialWrite | Self::DeleteMarkerPurge)
+    }
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
+pub struct MrfDeleteMarkerPurgeIdentity {
+    pub bucket_incarnation_id: Uuid,
+    pub marker_incarnation_id: Uuid,
+    pub marker_identity: [u8; 32],
+}
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct MrfDeleteMarkerPurge {
+    pub bucket_incarnation_id: Uuid,
+    pub marker_incarnation_id: Uuid,
+    pub marker_identity: [u8; 32],
+    pub marker: Arc<[u8]>,
+}
+
+impl MrfDeleteMarkerPurge {
+    pub fn new(
+        bucket_incarnation_id: Uuid,
+        marker_incarnation_id: Uuid,
+        marker_identity: [u8; 32],
+        marker: Vec<u8>,
+    ) -> Option<Self> {
+        let payload = Self {
+            bucket_incarnation_id,
+            marker_incarnation_id,
+            marker_identity,
+            marker: Arc::from(marker),
+        };
+        payload.is_valid().then_some(payload)
+    }
+
+    pub const fn identity(&self) -> MrfDeleteMarkerPurgeIdentity {
+        MrfDeleteMarkerPurgeIdentity {
+            bucket_incarnation_id: self.bucket_incarnation_id,
+            marker_incarnation_id: self.marker_incarnation_id,
+            marker_identity: self.marker_identity,
+        }
+    }
+
+    pub fn is_valid(&self) -> bool {
+        !self.bucket_incarnation_id.is_nil()
+            && self.marker_incarnation_id == self.bucket_incarnation_id
+            && !self.marker.is_empty()
+            && self.marker.len() <= MRF_MAX_DELETE_MARKER_BYTES
+    }
 }
 
 /// One repair intent. Kept deliberately small so the in-memory queue and the
@@ -80,6 +135,10 @@ pub struct MrfIntent {
     /// Version the intent targets, as raw UUID bytes.
     pub version_id: Option<[u8; 16]>,
     pub kind: MrfKind,
+    /// Present only for [`MrfKind::DeleteMarkerPurge`]. The encoded marker is a
+    /// destructive-operation precondition, while `marker_identity` is its
+    /// deterministic queue and proof identity.
+    pub delete_marker_purge: Option<MrfDeleteMarkerPurge>,
     /// Stable erasure-set scope when the producer has it. Kept optional so
     /// metadata corruption and legacy producers do not invent a scope.
     pub scope: Option<MrfScope>,
@@ -99,6 +158,7 @@ pub struct MrfDurableRepairAnchor {
     pub object: Arc<str>,
     pub version_id: Option<[u8; 16]>,
     pub scope: Option<MrfScope>,
+    pub delete_marker_purge: Option<MrfDeleteMarkerPurgeIdentity>,
     pub lease: MrfIngressLease,
     pub bucket_incarnation_id: Uuid,
 }
@@ -108,17 +168,22 @@ impl MrfDurableRepairAnchor {
     /// incarnation and the original ingress lease. Legacy replay records lack
     /// both pieces and therefore remain fail-closed.
     pub fn from_intent(intent: &MrfIntent, bucket_incarnation_id: Uuid) -> Option<Self> {
+        let lease = intent.lease?;
+        let (version_id, scope) = canonical_identity(intent.kind, intent.version_id, intent.scope);
+        let delete_marker_purge = canonical_delete_marker_purge_identity(intent)?;
+        let bucket_incarnation_id = delete_marker_purge
+            .map(|identity| identity.bucket_incarnation_id)
+            .unwrap_or(bucket_incarnation_id);
         if bucket_incarnation_id.is_nil() {
             return None;
         }
-        let lease = intent.lease?;
-        let (version_id, scope) = canonical_identity(intent.kind, intent.version_id, intent.scope);
         Some(Self {
             kind: intent.kind,
             bucket: intent.bucket.clone(),
             object: intent.object.clone(),
             version_id,
             scope,
+            delete_marker_purge,
             lease,
             bucket_incarnation_id,
         })
@@ -133,6 +198,7 @@ impl MrfDurableRepairAnchor {
             && self.object == event.object
             && self.version_id == event.version_id
             && self.scope == event.scope
+            && self.delete_marker_purge == event.delete_marker_purge
             && self.lease == lease
             && self.bucket_incarnation_id == event.bucket_incarnation_id
     }
@@ -203,6 +269,7 @@ pub enum MrfDropReason {
     Uninitialized,
     Full,
     OversizedIdentity,
+    InvalidPayload,
     CoalescerFull,
 }
 
@@ -221,7 +288,14 @@ impl MrfIntent {
     pub fn estimated_bytes(&self) -> usize {
         // Struct + strings + version bytes; buckets and objects are usually
         // far below this bound, so rounding up keeps the budget conservative.
-        64 + self.bucket.len() + self.object.len()
+        64usize
+            .saturating_add(self.bucket.len())
+            .saturating_add(self.object.len())
+            .saturating_add(
+                self.delete_marker_purge
+                    .as_ref()
+                    .map_or(0, |purge| std::mem::size_of::<MrfDeleteMarkerPurge>().saturating_add(purge.marker.len())),
+            )
     }
 }
 
@@ -245,8 +319,9 @@ pub enum MrfDurableAdmissionError {
     Timeout,
 }
 
-/// A committed partial write awaiting a durable checkpoint, separate from the
-/// best-effort read/scanner channel. Dropping the reply does not cancel repair.
+/// A committed storage responsibility awaiting a durable checkpoint, separate
+/// from the best-effort read/scanner channel. Dropping the reply does not
+/// cancel repair.
 pub struct MrfDurableSubmission {
     pub intent: MrfIntent,
     pub response: oneshot::Sender<Result<(), MrfDurableAdmissionError>>,
@@ -260,36 +335,70 @@ pub fn init_durable_mrf_channel() -> Result<mpsc::Receiver<MrfDurableSubmission>
     Ok(receiver)
 }
 
-/// Wait for checkpoint publication only on a committed partial-write path.
-/// Every write gets a fresh lease: a previous repair of the same unversioned
-/// key cannot discharge a later overwrite. An error is not durable admission.
+/// Wait for checkpoint publication on a committed partial-write path. Every
+/// write gets a fresh lease: a previous repair of the same unversioned key
+/// cannot discharge a later overwrite. An error is not durable admission.
 pub async fn persist_partial_write_intent(
     bucket: &str,
     object: &str,
     version_id: Option<Uuid>,
     scope: MrfScope,
 ) -> Result<(), MrfDurableAdmissionError> {
-    if !mrf_delivery_enabled() {
-        return Err(MrfDurableAdmissionError::Disabled);
-    }
-    if bucket.is_empty()
-        || object.is_empty()
-        || bucket.len() > MRF_MAX_IDENTITY_COMPONENT
-        || object.len() > MRF_MAX_IDENTITY_COMPONENT
-    {
-        return Err(MrfDurableAdmissionError::InvalidIdentity);
-    }
-    let sender = GLOBAL_DURABLE_MRF_SENDER.get().ok_or(MrfDurableAdmissionError::Unavailable)?;
-    let mut intent = MrfIntent {
+    let intent = MrfIntent {
         bucket: Arc::from(bucket),
         object: Arc::from(object),
         version_id: canonical_version(version_id),
         kind: MrfKind::PartialWrite,
+        delete_marker_purge: None,
         scope: Some(scope),
         lease: None,
         enqueued_at_ms: unix_now_ms(),
         attempts: 0,
     };
+    persist_durable_intent(intent).await
+}
+
+pub async fn persist_delete_marker_purge_intent(
+    bucket: &str,
+    object: &str,
+    version_id: Uuid,
+    scope: MrfScope,
+    delete_marker_purge: MrfDeleteMarkerPurge,
+) -> Result<(), MrfDurableAdmissionError> {
+    if version_id.is_nil() {
+        return Err(MrfDurableAdmissionError::InvalidIdentity);
+    }
+    let intent = MrfIntent {
+        bucket: Arc::from(bucket),
+        object: Arc::from(object),
+        version_id: canonical_version(Some(version_id)),
+        kind: MrfKind::DeleteMarkerPurge,
+        delete_marker_purge: Some(delete_marker_purge),
+        scope: Some(scope),
+        lease: None,
+        enqueued_at_ms: unix_now_ms(),
+        attempts: 0,
+    };
+    persist_durable_intent_unconditionally(intent).await
+}
+
+async fn persist_durable_intent(intent: MrfIntent) -> Result<(), MrfDurableAdmissionError> {
+    if !mrf_delivery_enabled() {
+        return Err(MrfDurableAdmissionError::Disabled);
+    }
+    persist_durable_intent_unconditionally(intent).await
+}
+
+async fn persist_durable_intent_unconditionally(mut intent: MrfIntent) -> Result<(), MrfDurableAdmissionError> {
+    if intent.bucket.is_empty()
+        || intent.object.is_empty()
+        || intent.bucket.len() > MRF_MAX_IDENTITY_COMPONENT
+        || intent.object.len() > MRF_MAX_IDENTITY_COMPONENT
+        || !valid_intent_payload(&intent)
+    {
+        return Err(MrfDurableAdmissionError::InvalidIdentity);
+    }
+    let sender = GLOBAL_DURABLE_MRF_SENDER.get().ok_or(MrfDurableAdmissionError::Unavailable)?;
     if try_rearm_mrf_replay_intent(&mut intent) != MrfIngressResult::Enqueued {
         return Err(MrfDurableAdmissionError::InvalidIdentity);
     }
@@ -359,10 +468,27 @@ fn canonical_identity(
     let version_id = version_id.filter(|bytes| *bytes != [0; 16]);
     match kind {
         MrfKind::MetadataCorruption => (None, None),
-        MrfKind::DecodeFailure | MrfKind::PartialWrite => (version_id, scope),
+        MrfKind::DecodeFailure | MrfKind::PartialWrite | MrfKind::DeleteMarkerPurge => (version_id, scope),
     }
 }
 
+fn canonical_delete_marker_purge_identity(intent: &MrfIntent) -> Option<Option<MrfDeleteMarkerPurgeIdentity>> {
+    match intent.kind {
+        MrfKind::DeleteMarkerPurge => intent
+            .delete_marker_purge
+            .as_ref()
+            .filter(|payload| payload.is_valid())
+            .map(|payload| Some(payload.identity())),
+        _ if intent.delete_marker_purge.is_none() => Some(None),
+        _ => None,
+    }
+}
+
+fn valid_intent_payload(intent: &MrfIntent) -> bool {
+    canonical_delete_marker_purge_identity(intent).is_some()
+        && (intent.kind != MrfKind::DeleteMarkerPurge || (intent.version_id.is_some() && intent.scope.is_some()))
+}
+
 fn identity_estimated_bytes(key: &MrfIdentityKey) -> usize {
     64usize
         .saturating_add(key.bucket.len())
@@ -512,6 +638,9 @@ pub fn try_send_mrf_intent_typed(
     if !mrf_delivery_enabled() {
         return MrfIngressResult::Dropped(MrfDropReason::Disabled);
     }
+    if kind == MrfKind::DeleteMarkerPurge {
+        return MrfIngressResult::Dropped(MrfDropReason::InvalidPayload);
+    }
     let Some(sender) = GLOBAL_MRF_SENDER.get() else {
         return MrfIngressResult::Dropped(MrfDropReason::Uninitialized);
     };
@@ -535,6 +664,7 @@ pub fn try_send_mrf_intent_typed(
         object: key.object.clone(),
         version_id: key.version_id,
         kind,
+        delete_marker_purge: None,
         scope,
         lease: Some(lease),
         enqueued_at_ms: unix_now_ms(),
@@ -562,15 +692,18 @@ pub fn try_send_mrf_intent_typed(
 /// producer coalescer key: the replay queue first deduplicates legacy records,
 /// then manager admission owns task-level deduplication with live producers.
 pub fn try_rearm_mrf_replay_intent(intent: &mut MrfIntent) -> MrfIngressResult {
-    if intent.lease.is_some() {
-        return MrfIngressResult::Enqueued;
-    }
     if intent.bucket.len() > MRF_MAX_IDENTITY_COMPONENT || intent.object.len() > MRF_MAX_IDENTITY_COMPONENT {
         return MrfIngressResult::Dropped(MrfDropReason::OversizedIdentity);
     }
     let (version_id, scope) = canonical_identity(intent.kind, intent.version_id, intent.scope);
     intent.version_id = version_id;
     intent.scope = scope;
+    if !valid_intent_payload(intent) {
+        return MrfIngressResult::Dropped(MrfDropReason::InvalidPayload);
+    }
+    if intent.lease.is_some() {
+        return MrfIngressResult::Enqueued;
+    }
     intent.lease = Some(MrfIngressLease::new(NEXT_MRF_LEASE.fetch_add(1, Ordering::Relaxed)));
     MrfIngressResult::Enqueued
 }
@@ -635,6 +768,7 @@ pub struct MrfVerifiedRepairEvent {
     pub object: Arc<str>,
     pub version_id: Option<[u8; 16]>,
     pub scope: Option<MrfScope>,
+    pub delete_marker_purge: Option<MrfDeleteMarkerPurgeIdentity>,
     pub lease: Option<MrfIngressLease>,
     pub bucket_incarnation_id: Uuid,
     pub disposition: MrfVerifiedRepairDisposition,
@@ -736,6 +870,7 @@ mod tests {
             object: Arc::from("object"),
             version_id: Some([0u8; 16]),
             kind: MrfKind::DecodeFailure,
+            delete_marker_purge: None,
             scope: None,
             lease: None,
             enqueued_at_ms: 0,
@@ -800,6 +935,7 @@ mod tests {
             object: Arc::from("object"),
             version_id: Some([0; 16]),
             kind: MrfKind::PartialWrite,
+            delete_marker_purge: None,
             scope: Some(MrfScope {
                 pool_index: 1,
                 set_index: 2,
@@ -839,6 +975,7 @@ mod tests {
             object: Arc::from("object"),
             version_id: Some([0; 16]),
             kind: MrfKind::PartialWrite,
+            delete_marker_purge: None,
             scope: Some(MrfScope {
                 pool_index: 2,
                 set_index: 3,
@@ -873,6 +1010,7 @@ mod tests {
                 pool_index: 4,
                 set_index: 5,
             }),
+            delete_marker_purge: None,
             lease,
             bucket_incarnation_id: incarnation,
         };
@@ -882,6 +1020,7 @@ mod tests {
             object,
             version_id: anchor.version_id,
             scope: anchor.scope,
+            delete_marker_purge: None,
             lease: Some(lease),
             bucket_incarnation_id: incarnation,
             disposition: MrfVerifiedRepairDisposition::Repaired,
@@ -949,6 +1088,7 @@ mod tests {
                 pool_index: 1,
                 set_index: 2,
             }),
+            delete_marker_purge: None,
             lease,
             bucket_incarnation_id: incarnation,
         };
@@ -963,6 +1103,7 @@ mod tests {
             object: retained_anchor.object.clone(),
             version_id: retained_anchor.version_id,
             scope: retained_anchor.scope,
+            delete_marker_purge: None,
             lease: Some(retained_anchor.lease),
             bucket_incarnation_id: retained_anchor.bucket_incarnation_id,
             disposition: MrfVerifiedRepairDisposition::Repaired,
@@ -1076,6 +1217,7 @@ mod tests {
                 pool_index: 2,
                 set_index: 3,
             }),
+            delete_marker_purge: None,
             lease: Some(MrfIngressLease::new(42)),
             bucket_incarnation_id,
             disposition: MrfVerifiedRepairDisposition::Repaired,
@@ -1120,4 +1262,42 @@ async fn partial_write_durable_admission_waits_for_persistence_and_renews_lease(
         .send(Ok(()))
         .expect("second producer should await receipt");
     assert_eq!(second.await.expect("second producer should complete"), Ok(()));
+
+    let bucket_incarnation_id = Uuid::new_v4();
+    let marker_identity = [7; 32];
+    let purge = MrfDeleteMarkerPurge::new(bucket_incarnation_id, bucket_incarnation_id, marker_identity, vec![1, 2, 3])
+        .expect("valid purge payload");
+    let version_id = Uuid::new_v4();
+    let purge_submit =
+        tokio::spawn(async move { persist_delete_marker_purge_intent("bucket", "marker", version_id, scope, purge).await });
+    let submission = receiver.recv().await.expect("purge submission should arrive");
+    assert_eq!(submission.intent.kind, MrfKind::DeleteMarkerPurge);
+    assert_eq!(submission.intent.version_id, Some(*version_id.as_bytes()));
+    assert_eq!(submission.intent.scope, Some(scope));
+    assert_eq!(
+        submission
+            .intent
+            .delete_marker_purge
+            .as_ref()
+            .map(MrfDeleteMarkerPurge::identity),
+        Some(MrfDeleteMarkerPurgeIdentity {
+            bucket_incarnation_id,
+            marker_incarnation_id: bucket_incarnation_id,
+            marker_identity,
+        })
+    );
+    assert!(submission.intent.lease.is_some());
+    submission.response.send(Ok(())).expect("purge producer should await receipt");
+    assert_eq!(purge_submit.await.expect("purge producer should complete"), Ok(()));
+
+    let invalid = MrfDeleteMarkerPurge {
+        bucket_incarnation_id,
+        marker_incarnation_id: Uuid::new_v4(),
+        marker_identity,
+        marker: Arc::from([1u8]),
+    };
+    assert_eq!(
+        persist_delete_marker_purge_intent("bucket", "marker", version_id, scope, invalid).await,
+        Err(MrfDurableAdmissionError::InvalidIdentity)
+    );
 }

crates/ecstore/src/set_disk/mod.rs

@@ -4208,6 +4208,59 @@ impl DiskHealthEntry {
 }
 
 impl SetDisks {
+    pub(in crate::set_disk) async fn persist_delete_marker_purge(
+        &self,
+        bucket: &str,
+        object: &str,
+        version: Uuid,
+        purge: rustfs_common::mrf_channel::MrfDeleteMarkerPurge,
+    ) -> bool {
+        use rustfs_common::mrf_channel::{MrfDurableAdmissionError, MrfScope, persist_delete_marker_purge_intent};
+        let scope = (|| {
+            Ok::<_, MrfDurableAdmissionError>(MrfScope {
+                pool_index: u32::try_from(self.pool_index).map_err(|_| MrfDurableAdmissionError::InvalidIdentity)?,
+                set_index: u32::try_from(self.set_index).map_err(|_| MrfDurableAdmissionError::InvalidIdentity)?,
+            })
+        })();
+        let result = match scope {
+            Ok(scope) => persist_delete_marker_purge_intent(bucket, object, version, scope, purge).await,
+            Err(error) => Err(error),
+        };
+        match result {
+            Ok(()) => {
+                tracing::trace!(
+                    event = EVENT_SET_DISK_HEAL,
+                    component = LOG_COMPONENT_ECSTORE,
+                    subsystem = LOG_SUBSYSTEM_SET_DISK,
+                    state = "delete_marker_purge_durably_admitted",
+                    bucket,
+                    object,
+                    version_id = %version,
+                    pool_index = self.pool_index,
+                    set_index = self.set_index,
+                    "Delete-marker purge responsibility persisted"
+                );
+                true
+            }
+            Err(error) => {
+                warn!(
+                    event = EVENT_SET_DISK_HEAL,
+                    component = LOG_COMPONENT_ECSTORE,
+                    subsystem = LOG_SUBSYSTEM_SET_DISK,
+                    state = "delete_marker_purge_admission_failed",
+                    bucket,
+                    object,
+                    version_id = %version,
+                    pool_index = self.pool_index,
+                    set_index = self.set_index,
+                    error = %error,
+                    "Delete-marker purge responsibility could not be persisted"
+                );
+                false
+            }
+        }
+    }
+
     pub(in crate::set_disk) async fn persist_partial_write(&self, bucket: &str, object: &str, version_id: Option<&str>) -> bool {
         use rustfs_common::mrf_channel::{
             MrfDurableAdmissionError, MrfScope, mrf_delivery_enabled, persist_partial_write_intent,
@@ -15155,8 +15208,8 @@ mod tests {
             "a stale null-marker lifecycle target must not delete its replacement: {stale_errors:?}"
         );
 
-        let (current_marker, _, current_error) = set_disks
-            .get_object_info_and_quorum(
+        let (current_marker, _, _, current_error) = set_disks
+            .get_object_info_fileinfo_and_quorum(
                 bucket,
                 object,
                 &ObjectOptions {

crates/ecstore/src/set_disk/ops/heal.rs

@@ -521,6 +521,77 @@ fn warn_heal_writer_failures(
 }
 
 impl SetDisks {
+    pub(crate) async fn purge_delete_marker_exact(
+        &self,
+        bucket: &str,
+        object: &str,
+        version: Uuid,
+        marker: rustfs_filemeta::MetaDeleteMarker,
+    ) -> Result<bool> {
+        let disks = self.get_disks_internal().await;
+        if disks.len() != self.set_drive_count || disks.iter().any(Option::is_none) {
+            return Err(StorageError::SlowDown);
+        }
+        let _object_guard = self.acquire_write_lock_diag("delete_marker_purge", bucket, object).await?;
+        let scope = crate::store::bucket_heal_scope(bucket).ok_or(StorageError::PreconditionFailed)?;
+        scope.check()?;
+
+        let request = FileInfo {
+            volume: bucket.to_owned(),
+            name: object.to_owned(),
+            version_id: Some(version),
+            ..Default::default()
+        };
+        let results = join_all(disks.iter().enumerate().map(|(disk_index, disk)| {
+            let request = request.clone();
+            let marker = marker.clone();
+            async move {
+                if let Some(error) = injected_dangling_delete_error(bucket, object, disk_index) {
+                    return Err(error);
+                }
+                let Some(disk) = disk else { return Err(DiskError::DiskNotFound) };
+                disk.delete_version(
+                    bucket,
+                    object,
+                    request,
+                    false,
+                    DeleteOptions {
+                        expected_delete_marker: Some(marker),
+                        ..Default::default()
+                    },
+                )
+                .await
+            }
+        }))
+        .await;
+        let mut removed = false;
+        for result in results {
+            match result {
+                Ok(()) => removed = true,
+                Err(DiskError::FileNotFound | DiskError::FileVersionNotFound) => {}
+                Err(error) => return Err(error.into()),
+            }
+        }
+
+        scope.check()?;
+        let (_, errors) =
+            Self::read_all_fileinfo(disks.as_slice(), "", bucket, object, &version.to_string(), false, false, false).await?;
+        let selected = self.get_disks_internal().await;
+        let same_targets = selected.len() == disks.len() && selected.iter().zip(&disks).all(|(current, original)| {
+            matches!((current, original), (Some(current), Some(original)) if std::sync::Arc::ptr_eq(current, original))
+        });
+        if !same_targets
+            || !errors
+                .iter()
+                .all(|error| matches!(error, Some(DiskError::FileNotFound | DiskError::FileVersionNotFound)))
+        {
+            return Err(StorageError::SlowDown);
+        }
+        scope.check()?;
+        self.invalidate_get_object_metadata_cache(bucket, object).await;
+        Ok(removed)
+    }
+
     /// Read back one healed version from every explicitly admitted replacement
     /// target. This is intentionally separate from the normal heal result: a
     /// successful result describes the transaction attempt, while automatic

crates/ecstore/src/set_disk/ops/object.rs

@@ -20,6 +20,7 @@
 //! SetDisks core (io_primitives) via inherent calls.
 
 use crate::core::pools::DecommissionCapacityAdmission;
+use rustfs_common::mrf_channel::MrfDeleteMarkerPurge;
 use rustfs_filemeta::metadata_keys;
 
 #[cfg(test)]
@@ -63,6 +64,27 @@ use super::super::{
     take_prepared_get_object_metadata, to_object_err, try_read_inline_data_shards_direct, warn,
 };
 use super::bitrot_self_verify::{BitrotSelfVerifyTarget, drop_failed_writer_disks, verify_written_bitrot_shards};
+
+fn delete_marker_purge_candidate(
+    file_info: &FileInfo,
+    expected_bucket_incarnation_id: Option<Uuid>,
+) -> Option<MrfDeleteMarkerPurge> {
+    let bucket_incarnation_id = expected_bucket_incarnation_id.filter(|id| !id.is_nil())?;
+    let version_id = file_info.version_id.filter(|id| !id.is_nil())?;
+    if !file_info.is_canonical_delete_marker() {
+        return None;
+    }
+    let marker_incarnation_id = file_info.delete_marker_incarnation()?;
+    if marker_incarnation_id != bucket_incarnation_id {
+        return None;
+    }
+    let marker = rustfs_filemeta::MetaDeleteMarker::from(file_info.clone());
+    if marker.version_id != Some(version_id) {
+        return None;
+    }
+    let marker_identity = marker.stable_identity();
+    MrfDeleteMarkerPurge::new(bucket_incarnation_id, marker_incarnation_id, marker_identity, marker.marshal_msg().ok()?)
+}
 use crate::api::config::storageclass;
 use crate::bucket::lifecycle::bucket_lifecycle_ops::LifecycleOps;
 use crate::bucket::utils::is_meta_bucketname;
@@ -7529,6 +7551,168 @@ impl SetDisks {
 
         Ok(ObjectInfo::from_file_info(&fi, bucket, object, opts.versioned || opts.version_suspended))
     }
+    #[tracing::instrument(skip(self, delete_marker_purge))]
+    async fn delete_object_version_with_purge(
+        &self,
+        bucket: &str,
+        object: &str,
+        fi: &FileInfo,
+        force_del_marker: bool,
+        delete_marker_purge: Option<MrfDeleteMarkerPurge>,
+    ) -> Result<()> {
+        let transported = delete_file_info_with_replication_transport_metadata(fi);
+        let fi = &transported;
+        let disks = self.disk_inventory().await;
+        let namespace_owner = (!is_meta_bucketname(bucket)).then(|| self.ctx.begin_namespace_commit());
+        let write_quorum = disks.len() / 2 + 1;
+        let rollback_dir = Uuid::new_v4();
+
+        let mut futures = Vec::with_capacity(disks.len());
+        let mut errs = Vec::with_capacity(disks.len());
+
+        for disk in disks.iter() {
+            let disk_namespace_owner = namespace_owner.clone().map(|owner| owner as Arc<dyn Send + Sync>);
+            futures.push(async move {
+                if let Some(disk) = disk {
+                    match disk
+                        .delete_version_with_namespace_owner(
+                            bucket,
+                            object,
+                            fi.clone(),
+                            force_del_marker,
+                            DeleteOptions {
+                                old_data_dir: Some(rollback_dir),
+                                ..Default::default()
+                            },
+                            disk_namespace_owner,
+                        )
+                        .await
+                    {
+                        Ok(r) => Ok(r),
+                        Err(e) => Err(e),
+                    }
+                } else {
+                    Err(DiskError::DiskNotFound)
+                }
+            });
+        }
+
+        let results = join_all(futures).await;
+        for result in results {
+            match result {
+                Ok(_) => {
+                    errs.push(None);
+                }
+                Err(e) => {
+                    errs.push(Some(e));
+                }
+            }
+        }
+
+        let quorum_result = resolve_tiered_decommission_write_quorum_result(&errs, write_quorum, bucket, object);
+        let should_rollback = quorum_result.is_err();
+        let mut rollback_futures = Vec::new();
+        for (index, err) in errs.iter().enumerate() {
+            // backlog#1158: when rolling back, fan the idempotent undo out to every
+            // online disk (each self-decides from its staged backup: restore if the
+            // rollback dir is present, no-op otherwise). This covers a disk that
+            // staged + applied the delete and *then* errored, which the plain
+            // `err.is_some()` skip would leave deleted while its peers were restored.
+            // On success only the successful disks' backup dirs need cleaning; errored
+            // disks' residue is reclaimed by heal/scanner.
+            if !should_rollback && err.is_some() {
+                continue;
+            }
+
+            let Some(disk) = disks[index].as_ref() else {
+                continue;
+            };
+
+            let disk = disk.clone();
+            let bucket = bucket.to_string();
+            let object = object.to_string();
+            let fi = fi.clone();
+            let disk_namespace_owner = namespace_owner.clone().map(|owner| owner as Arc<dyn Send + Sync>);
+            rollback_futures.push(async move {
+                if should_rollback {
+                    // The dedicated undo path never forwards the marker-only creation flag.
+                    if let Err(err) = disk
+                        .undo_write_with_namespace_owner(
+                            &bucket,
+                            &object,
+                            fi,
+                            DeleteOptions {
+                                undo_write: true,
+                                undo_delete: true,
+                                old_data_dir: Some(rollback_dir),
+                                ..Default::default()
+                            },
+                            disk_namespace_owner,
+                        )
+                        .await
+                    {
+                        warn!(
+                            bucket = %bucket,
+                            object = %object,
+                            rollback_dir = %rollback_dir,
+                            error = ?err,
+                            "failed to roll back delete after write quorum failure"
+                        );
+                    }
+                } else {
+                    let rollback_path = format!("{object}/{rollback_dir}");
+                    if let Err(err) = disk
+                        .delete_with_namespace_owner(
+                            &bucket,
+                            &rollback_path,
+                            DeleteOptions {
+                                recursive: true,
+                                immediate: true,
+                                ..Default::default()
+                            },
+                            disk_namespace_owner,
+                        )
+                        .await
+                        && err != DiskError::FileNotFound
+                        && err != DiskError::VolumeNotFound
+                    {
+                        warn!(
+                            bucket = %bucket,
+                            object = %object,
+                            rollback_dir = %rollback_dir,
+                            error = ?err,
+                            "failed to clean delete rollback state after quorum success"
+                        );
+                    }
+                }
+            });
+        }
+
+        join_all(rollback_futures).await;
+        drop(namespace_owner);
+        if quorum_result.is_ok()
+            && errs.iter().any(Option::is_some)
+            && let Some(purge) = delete_marker_purge
+            && let Some(version) = fi.version_id.filter(|version| !version.is_nil())
+        {
+            let _ = self.persist_delete_marker_purge(bucket, object, version, purge).await;
+        }
+        // An explicit purge can carry deleted=true for the existing marker.
+        // It must not create a repair intent that could reintroduce that marker.
+        if quorum_result.is_ok()
+            && fi.deleted
+            && (fi.mark_deleted || force_del_marker)
+            && !fi.tier_free_version()
+            && version_purge_status_from_filemeta(fi.version_purge_status()) != VersionPurgeStatusType::Complete
+            && errs.iter().any(Option::is_some)
+        {
+            let version_id = fi.version_id.map(|version| version.to_string());
+            let _ = self
+                .add_partial(bucket, object, version_id.as_deref().unwrap_or_default())
+                .await;
+        }
+        quorum_result
+    }
 }
 
 #[async_trait::async_trait]
@@ -7818,151 +8002,8 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
     }
     #[tracing::instrument(skip(self))]
     async fn delete_object_version(&self, bucket: &str, object: &str, fi: &FileInfo, force_del_marker: bool) -> Result<()> {
-        let transported = delete_file_info_with_replication_transport_metadata(fi);
-        let fi = &transported;
-        let disks = self.disk_inventory().await;
-        let namespace_owner = (!is_meta_bucketname(bucket)).then(|| self.ctx.begin_namespace_commit());
-        let write_quorum = disks.len() / 2 + 1;
-        let rollback_dir = Uuid::new_v4();
-
-        let mut futures = Vec::with_capacity(disks.len());
-        let mut errs = Vec::with_capacity(disks.len());
-
-        for disk in disks.iter() {
-            let disk_namespace_owner = namespace_owner.clone().map(|owner| owner as Arc<dyn Send + Sync>);
-            futures.push(async move {
-                if let Some(disk) = disk {
-                    match disk
-                        .delete_version_with_namespace_owner(
-                            bucket,
-                            object,
-                            fi.clone(),
-                            force_del_marker,
-                            DeleteOptions {
-                                old_data_dir: Some(rollback_dir),
-                                ..Default::default()
-                            },
-                            disk_namespace_owner,
-                        )
-                        .await
-                    {
-                        Ok(r) => Ok(r),
-                        Err(e) => Err(e),
-                    }
-                } else {
-                    Err(DiskError::DiskNotFound)
-                }
-            });
-        }
-
-        let results = join_all(futures).await;
-        for result in results {
-            match result {
-                Ok(_) => {
-                    errs.push(None);
-                }
-                Err(e) => {
-                    errs.push(Some(e));
-                }
-            }
-        }
-
-        let quorum_result = resolve_tiered_decommission_write_quorum_result(&errs, write_quorum, bucket, object);
-        let should_rollback = quorum_result.is_err();
-        let mut rollback_futures = Vec::new();
-        for (index, err) in errs.iter().enumerate() {
-            // backlog#1158: when rolling back, fan the idempotent undo out to every
-            // online disk (each self-decides from its staged backup: restore if the
-            // rollback dir is present, no-op otherwise). This covers a disk that
-            // staged + applied the delete and *then* errored, which the plain
-            // `err.is_some()` skip would leave deleted while its peers were restored.
-            // On success only the successful disks' backup dirs need cleaning; errored
-            // disks' residue is reclaimed by heal/scanner.
-            if !should_rollback && err.is_some() {
-                continue;
-            }
-
-            let Some(disk) = disks[index].as_ref() else {
-                continue;
-            };
-
-            let disk = disk.clone();
-            let bucket = bucket.to_string();
-            let object = object.to_string();
-            let fi = fi.clone();
-            let disk_namespace_owner = namespace_owner.clone().map(|owner| owner as Arc<dyn Send + Sync>);
-            rollback_futures.push(async move {
-                if should_rollback {
-                    // The dedicated undo path never forwards the marker-only creation flag.
-                    if let Err(err) = disk
-                        .undo_write_with_namespace_owner(
-                            &bucket,
-                            &object,
-                            fi,
-                            DeleteOptions {
-                                undo_write: true,
-                                undo_delete: true,
-                                old_data_dir: Some(rollback_dir),
-                                ..Default::default()
-                            },
-                            disk_namespace_owner,
-                        )
-                        .await
-                    {
-                        warn!(
-                            bucket = %bucket,
-                            object = %object,
-                            rollback_dir = %rollback_dir,
-                            error = ?err,
-                            "failed to roll back delete after write quorum failure"
-                        );
-                    }
-                } else {
-                    let rollback_path = format!("{object}/{rollback_dir}");
-                    if let Err(err) = disk
-                        .delete_with_namespace_owner(
-                            &bucket,
-                            &rollback_path,
-                            DeleteOptions {
-                                recursive: true,
-                                immediate: true,
-                                ..Default::default()
-                            },
-                            disk_namespace_owner,
-                        )
-                        .await
-                        && err != DiskError::FileNotFound
-                        && err != DiskError::VolumeNotFound
-                    {
-                        warn!(
-                            bucket = %bucket,
-                            object = %object,
-                            rollback_dir = %rollback_dir,
-                            error = ?err,
-                            "failed to clean delete rollback state after quorum success"
-                        );
-                    }
-                }
-            });
-        }
-
-        join_all(rollback_futures).await;
-        drop(namespace_owner);
-        // An explicit purge can carry deleted=true for the existing marker.
-        // It must not create a repair intent that could reintroduce that marker.
-        if quorum_result.is_ok()
-            && fi.deleted
-            && (fi.mark_deleted || force_del_marker)
-            && !fi.tier_free_version()
-            && version_purge_status_from_filemeta(fi.version_purge_status()) != VersionPurgeStatusType::Complete
-            && errs.iter().any(Option::is_some)
-        {
-            let version_id = fi.version_id.map(|version| version.to_string());
-            let _ = self
-                .add_partial(bucket, object, version_id.as_deref().unwrap_or_default())
-                .await;
-        }
-        quorum_result
+        self.delete_object_version_with_purge(bucket, object, fi, force_del_marker, None)
+            .await
     }
 
     #[tracing::instrument(skip(self, objects, opts))]
@@ -8081,6 +8122,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
         let mut vers_map: HashMap<&String, FileInfoVersions> = HashMap::new();
         let mut tier_reference_leases: Vec<(usize, String, Option<TierDestinationId>)> = Vec::new();
         let mut tier_free_version_receipt_candidates: HashMap<usize, TierFreeVersionReceiptCandidate> = HashMap::new();
+        let mut delete_marker_purge_candidates = vec![None; objects.len()];
 
         for (i, dobj) in objects.iter().enumerate() {
             if del_errs[i].is_some() {
@@ -8111,15 +8153,18 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
             let marker_delete = dobj.version_id.is_none() || dobj.synthetic_version_id;
             let replication_needs_source = replicate_delete
                 && (!marker_delete || delete_config_snapshot.active_delete_marker_rules_require_tags(&replication_object_name));
-            let (goi, gerr) = if object_lock_check_required
+            let (goi, authoritative_file_info, gerr) = if object_lock_check_required
                 || replication_needs_source
                 || opts.tier_delete_journal_api.is_some()
                 || dobj.expected_identity.is_some()
+                || dobj.version_id.is_some()
             {
-                let (goi, _write_quorum, gerr) = self.get_object_info_and_quorum(bucket, &dobj.object_name, &check_opts).await;
-                (goi, gerr)
+                let (goi, file_info, _write_quorum, gerr) = self
+                    .get_object_info_fileinfo_and_quorum(bucket, &dobj.object_name, &check_opts)
+                    .await;
+                (goi, file_info, gerr)
             } else {
-                (ObjectInfo::default(), None)
+                (ObjectInfo::default(), FileInfo::default(), None)
             };
             let source_missing = gerr
                 .as_ref()
@@ -8253,6 +8298,15 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
             // accounted as an object deletion (issue #6745).
             let removed_delete_marker =
                 goi.delete_marker && dobj.version_id.is_some() && delete_file_info_version_id(goi.version_id) == version_id;
+            if removed_delete_marker
+                && goi.version_purge_status.is_empty()
+                && vr.version_purge_status().is_empty()
+                && vr.delete_marker_replication_status().is_empty()
+                && version_id.is_some_and(|version| authoritative_file_info.version_id == Some(version))
+            {
+                delete_marker_purge_candidates[i] =
+                    delete_marker_purge_candidate(&authoritative_file_info, opts.expected_bucket_incarnation_id);
+            }
             // Response-side only for the null identity: a delete request marked
             // `deleted` with `version_id == None` makes `FileMeta::delete_version`
             // re-create the marker it just removed (the suspended-bucket
@@ -8616,6 +8670,19 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
             self.release_dist_delete_object_locks_batch(dist_batch_lock_ids).await;
         }
 
+        let purge_submissions = delete_marker_purge_candidates
+            .into_iter()
+            .enumerate()
+            .filter_map(|(index, purge)| {
+                let purge = purge?;
+                if del_errs[index].is_some() || !del_obj_errs.iter().any(|errors| errors[index].is_some()) {
+                    return None;
+                }
+                let version = objects[index].version_id.filter(|version| !version.is_nil())?;
+                Some(self.persist_delete_marker_purge(bucket, objects[index].object_name.as_str(), version, purge))
+            });
+        let _ = join_all(purge_submissions).await;
+
         for (object, err) in objects.iter().zip(del_errs.iter()) {
             if err.is_none() {
                 self.invalidate_get_object_metadata_cache(bucket, &object.object_name).await;
@@ -8848,7 +8915,8 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
         let mut version_found = true;
         // delete_object_version below derives its own majority quorum from the
         // disk array, so the object-derived quorum here is unused.
-        let (mut goi, _write_quorum, gerr) = self.get_object_info_and_quorum(bucket, object, &opts).await;
+        let (mut goi, authoritative_file_info, _write_quorum, gerr) =
+            self.get_object_info_fileinfo_and_quorum(bucket, object, &opts).await;
         if let Some(err) = &gerr
             && goi.name.is_empty()
         {
@@ -8863,6 +8931,12 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
             opts.precondition_check(&goi)?;
             check_object_lock_delete(&self.ctx, bucket, object, &goi, &opts).await?;
         }
+        let delete_marker_purge_candidate = opts
+            .version_id
+            .as_deref()
+            .and_then(|value| Uuid::parse_str(value).ok())
+            .filter(|version| !version.is_nil() && authoritative_file_info.version_id == Some(*version))
+            .and_then(|_| delete_marker_purge_candidate(&authoritative_file_info, opts.expected_bucket_incarnation_id));
 
         if opts.transition.expire_restored {
             // Restore-expiry (DeleteRestoredAction / DeleteRestoredVersionAction)
@@ -8921,6 +8995,9 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
         }
 
         let (mark_delete, mut delete_marker) = resolve_delete_version_state(&opts, &goi, version_found);
+        let delete_marker_purge = explicit_delete_removed_marker(&opts, &goi, version_found)
+            .then_some(delete_marker_purge_candidate)
+            .flatten();
 
         let mod_time = if let Some(mt) = opts.mod_time {
             mt
@@ -8969,9 +9046,15 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
             if opts.skip_free_version {
                 fi.set_skip_tier_free_version();
             }
-            self.delete_object_version(bucket, object, &fi, should_force_delete_marker_for_missing_version(&opts))
-                .await
-                .map_err(|e| to_object_err(e, vec![bucket, object]))?;
+            self.delete_object_version_with_purge(
+                bucket,
+                object,
+                &fi,
+                should_force_delete_marker_for_missing_version(&opts),
+                delete_marker_purge.clone(),
+            )
+            .await
+            .map_err(|e| to_object_err(e, vec![bucket, object]))?;
             #[cfg(test)]
             pause_delete_object_commit_after_publish(bucket, object).await;
 
@@ -9021,7 +9104,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
         if opts.skip_free_version {
             dfi.set_skip_tier_free_version();
         }
-        self.delete_object_version(bucket, object, &dfi, opts.delete_marker)
+        self.delete_object_version_with_purge(bucket, object, &dfi, opts.delete_marker, delete_marker_purge)
             .await
             .map_err(|e| to_object_err(e, vec![bucket, object]))?;
         #[cfg(test)]

crates/ecstore/src/set_disk/read.rs

@@ -625,26 +625,26 @@ impl SetDisks {
         }
     }
 
-    #[hotpath::measure(impl_type = "SetDisks")]
-    pub(super) async fn get_object_info_and_quorum(
+    pub(super) async fn get_object_info_fileinfo_and_quorum(
         &self,
         bucket: &str,
         object: &str,
         opts: &ObjectOptions,
-    ) -> (ObjectInfo, usize, Option<StorageError>) {
+    ) -> (ObjectInfo, FileInfo, usize, Option<StorageError>) {
         let snapshot = match self.get_object_fileinfo(bucket, object, opts, false, false).await {
             Ok(snapshot) => snapshot,
-            Err(e) => return (ObjectInfo::default(), 0, Some(e)),
+            Err(e) => return (ObjectInfo::default(), FileInfo::default(), 0, Some(e)),
         };
-        let fi = snapshot.fi();
+        let fi = snapshot.fi().clone();
 
         let write_quorum = fi.write_quorum(self.default_write_quorum());
 
-        let oi = ObjectInfo::from_file_info(fi, bucket, object, opts.versioned || opts.version_suspended);
+        let oi = ObjectInfo::from_file_info(&fi, bucket, object, opts.versioned || opts.version_suspended);
 
         if !fi.version_purge_status().is_empty() && opts.version_id.is_some() {
             return (
                 oi,
+                fi,
                 write_quorum,
                 Some(to_object_err(StorageError::MethodNotAllowed, vec![bucket, object])),
             );
@@ -652,20 +652,26 @@ impl SetDisks {
 
         if fi.deleted {
             if opts.incl_free_versions && fi.tier_free_version() && opts.version_id.is_some() {
-                return (oi, write_quorum, None);
+                return (oi, fi, write_quorum, None);
             }
             return if opts.version_id.is_none() || opts.delete_marker {
-                (oi, write_quorum, Some(to_object_err(StorageError::FileNotFound, vec![bucket, object])))
+                (
+                    oi,
+                    fi,
+                    write_quorum,
+                    Some(to_object_err(StorageError::FileNotFound, vec![bucket, object])),
+                )
             } else {
                 (
                     oi,
+                    fi,
                     write_quorum,
                     Some(to_object_err(StorageError::MethodNotAllowed, vec![bucket, object])),
                 )
             };
         }
 
-        (oi, write_quorum, None)
+        (oi, fi, write_quorum, None)
     }
 
     #[allow(clippy::too_many_arguments)]
@@ -2802,7 +2808,7 @@ mod metadata_cache_tests {
     }
 
     #[tokio::test]
-    async fn get_object_info_and_quorum_maps_delete_marker_and_purge_states() {
+    async fn get_object_info_fileinfo_and_quorum_maps_delete_marker_and_purge_states() {
         let bucket = "get-object-info-marker-bucket";
         let (_dir, disk) = new_read_version_test_disk(bucket).await;
         let set = SetDisks::new(
@@ -2829,8 +2835,8 @@ mod metadata_cache_tests {
         disk.write_metadata(bucket, bucket, "latest-delete-marker", latest_marker)
             .await
             .expect("latest marker metadata should be written");
-        let (_, _, latest_err) = set
-            .get_object_info_and_quorum(bucket, "latest-delete-marker", &ObjectOptions::default())
+        let (_, _, _, latest_err) = set
+            .get_object_info_fileinfo_and_quorum(bucket, "latest-delete-marker", &ObjectOptions::default())
             .await;
         assert!(
             matches!(latest_err, Some(StorageError::ObjectNotFound(_, _))),
@@ -2849,8 +2855,8 @@ mod metadata_cache_tests {
         disk.write_metadata(bucket, bucket, "version-delete-marker", version_marker)
             .await
             .expect("version marker metadata should be written");
-        let (_, _, version_err) = set
-            .get_object_info_and_quorum(
+        let (_, _, _, version_err) = set
+            .get_object_info_fileinfo_and_quorum(
                 bucket,
                 "version-delete-marker",
                 &ObjectOptions {

crates/ecstore/src/store/heal.rs

@@ -19,6 +19,7 @@ use crate::services::rebalance::{REBAL_META_NAME, RebalStatus};
 use crate::set_disk::get_lock_acquire_timeout;
 use crate::storage_api_contracts::heal::HealOperations as _;
 use crate::storage_api_contracts::namespace::NamespaceLocking as _;
+use rustfs_common::mrf_channel::MrfDeleteMarkerPurge;
 use rustfs_lock::NamespaceLockGuard;
 use std::collections::BTreeSet;
 use tracing::trace;
@@ -143,6 +144,72 @@ fn heal_format_fence_lost_error() -> Error {
 }
 
 impl ECStore {
+    pub async fn purge_delete_marker_with_proof(
+        self: &Arc<Self>,
+        bucket: &str,
+        object: &str,
+        version_id: &str,
+        purge: &MrfDeleteMarkerPurge,
+        opts: &HealOpts,
+    ) -> Result<bool> {
+        if opts.dry_run || opts.no_lock || !purge.is_valid() {
+            return Err(StorageError::PreconditionFailed);
+        }
+        let version = Uuid::parse_str(version_id)
+            .ok()
+            .filter(|version| !version.is_nil())
+            .ok_or(StorageError::PreconditionFailed)?;
+        let (Some(pool_index), Some(set_index)) = (opts.pool, opts.set) else {
+            return Err(StorageError::PreconditionFailed);
+        };
+        let mut marker = rustfs_filemeta::MetaDeleteMarker::default();
+        let consumed = marker.unmarshal_msg(&purge.marker)?;
+        if usize::try_from(consumed).ok() != Some(purge.marker.len())
+            || marker.version_id != Some(version)
+            || marker.stable_identity() != purge.marker_identity
+        {
+            return Err(StorageError::PreconditionFailed);
+        }
+        let marker_info = marker.clone().into_fileinfo(bucket, object, false)?;
+        if !marker_info.is_canonical_delete_marker()
+            || marker_info.delete_marker_incarnation() != Some(purge.marker_incarnation_id)
+        {
+            return Err(StorageError::PreconditionFailed);
+        }
+
+        let current = self.bucket_incarnation_id_from_disk(bucket).await?;
+        if current.is_nil() {
+            return Err(StorageError::BucketNotFound(bucket.to_owned()));
+        }
+        let original = purge.bucket_incarnation_id;
+        let object = object.to_owned();
+        self.run_bucket_heal_at_incarnation(bucket, current, opts, move |store, bucket, _opts| async move {
+            let scope = bucket_heal_scope(&bucket).ok_or(StorageError::PreconditionFailed)?;
+            scope.check()?;
+            if original != current {
+                let retirement_store = crate::bucket::metadata_sys::object_store_if_initialized_in(&store.ctx)
+                    .await
+                    .ok_or(StorageError::PreconditionFailed)?;
+                if !crate::bucket::retirement::is_retired(retirement_store, &bucket, original).await? {
+                    return Err(StorageError::PreconditionFailed);
+                }
+            }
+            let pool = store
+                .pools
+                .get(pool_index)
+                .ok_or_else(|| invalid_heal_pool_index(pool_index, store.pools.len()))?;
+            let set = pool.disk_set.get(set_index).cloned().ok_or_else(|| {
+                StorageError::InvalidArgument(
+                    "heal".to_string(),
+                    "set".to_string(),
+                    format!("invalid heal set index {set_index} for pool {pool_index}"),
+                )
+            })?;
+            set.purge_delete_marker_exact(&bucket, &object, version, marker).await
+        })
+        .await
+    }
+
     pub(super) async fn run_bucket_heal_at_incarnation<T, F, Fut>(
         self: &Arc<Self>,
         bucket: &str,

crates/filemeta/src/filemeta/version.rs

@@ -37,6 +37,7 @@ use rustfs_utils::http::{
     insert_bytes, is_internal_key, remove_bytes, strip_internal_prefix, strip_internal_prefix_preserving_case,
     target_delete_marker_versions,
 };
+use sha2::{Digest as _, Sha256};
 
 const MSGPACK_EXT8: u8 = 0xc7;
 const MSGPACK_EXT16: u8 = 0xc8;
@@ -3010,6 +3011,40 @@ impl TryFrom<LegacyMetaV2DeleteMarker> for MetaDeleteMarker {
 }
 
 impl MetaDeleteMarker {
+    /// Return a deterministic identity for an exact delete-marker body.
+    /// MessagePack map order is intentionally excluded from the identity.
+    pub fn stable_identity(&self) -> [u8; 32] {
+        const DOMAIN: &[u8] = b"rustfs-delete-marker-identity-v1\0";
+
+        let mut hasher = Sha256::new();
+        hasher.update(DOMAIN);
+        match self.version_id {
+            Some(version_id) => {
+                hasher.update([1]);
+                hasher.update(version_id.as_bytes());
+            }
+            None => hasher.update([0]),
+        }
+        match self.mod_time {
+            Some(mod_time) => {
+                hasher.update([1]);
+                hasher.update(mod_time.unix_timestamp_nanos().to_be_bytes());
+            }
+            None => hasher.update([0]),
+        }
+
+        let mut metadata = self.meta_sys.iter().collect::<Vec<_>>();
+        metadata.sort_unstable_by(|(left, _), (right, _)| left.as_bytes().cmp(right.as_bytes()));
+        hasher.update(u64::try_from(metadata.len()).unwrap_or(u64::MAX).to_be_bytes());
+        for (key, value) in metadata {
+            hasher.update(u64::try_from(key.len()).unwrap_or(u64::MAX).to_be_bytes());
+            hasher.update(key.as_bytes());
+            hasher.update(u64::try_from(value.len()).unwrap_or(u64::MAX).to_be_bytes());
+            hasher.update(value);
+        }
+        hasher.finalize().into()
+    }
+
     pub fn free_version(&self) -> bool {
         contains_key_bytes(&self.meta_sys, SUFFIX_FREE_VERSION)
     }
@@ -5625,6 +5660,40 @@ mod tests {
         assert_eq!(base.get_signature(), base.get_signature());
     }
 
+    #[test]
+    fn delete_marker_stable_identity_is_order_independent_and_exact() {
+        let version_id = sample_version_id();
+        let mod_time = sample_mod_time();
+        let mut first = MetaDeleteMarker {
+            version_id: Some(version_id),
+            mod_time: Some(mod_time),
+            meta_sys: HashMap::new(),
+        };
+        first.meta_sys.insert("alpha".to_string(), vec![1, 2]);
+        first.meta_sys.insert("beta".to_string(), vec![3, 4]);
+
+        let mut reordered = MetaDeleteMarker {
+            version_id: Some(version_id),
+            mod_time: Some(mod_time),
+            meta_sys: HashMap::new(),
+        };
+        reordered.meta_sys.insert("beta".to_string(), vec![3, 4]);
+        reordered.meta_sys.insert("alpha".to_string(), vec![1, 2]);
+        assert_eq!(first.stable_identity(), reordered.stable_identity());
+
+        let mut changed = first.clone();
+        changed.meta_sys.insert("beta".to_string(), vec![3, 5]);
+        assert_ne!(first.stable_identity(), changed.stable_identity());
+
+        changed = first.clone();
+        changed.version_id = Some(Uuid::new_v4());
+        assert_ne!(first.stable_identity(), changed.stable_identity());
+
+        changed = first.clone();
+        changed.mod_time = changed.mod_time.map(|value| value + time::Duration::NANOSECOND);
+        assert_ne!(first.stable_identity(), changed.stable_identity());
+    }
+
     #[test]
     fn invalid_version_uses_error_sentinel_not_zero() {
         // A version whose inner body is missing must not masquerade as an

crates/heal/src/heal/manager.rs

@@ -134,6 +134,7 @@ pub(super) struct MrfRepairNoticeTarget {
     pub(super) version_id: Option<[u8; 16]>,
     pub(super) kind: rustfs_common::mrf_channel::MrfKind,
     pub(super) scope: Option<rustfs_common::mrf_channel::MrfScope>,
+    pub(super) delete_marker_purge: Option<rustfs_common::mrf_channel::MrfDeleteMarkerPurgeIdentity>,
     pub(super) lease: Option<rustfs_common::mrf_channel::MrfIngressLease>,
 }
 
@@ -493,7 +494,8 @@ fn heal_type_matches_path(heal_type: &HealType, heal_path: &str) -> bool {
         HealType::Cluster => false,
         HealType::Object { bucket, object, .. }
         | HealType::Metadata { bucket, object }
-        | HealType::ECDecode { bucket, object, .. } => heal_path_matches_bucket_child(heal_path, bucket, object),
+        | HealType::ECDecode { bucket, object, .. }
+        | HealType::DeleteMarkerPurge { bucket, object, .. } => heal_path_matches_bucket_child(heal_path, bucket, object),
         HealType::Bucket { bucket } => heal_path == bucket,
         HealType::Prefix { bucket, prefix } => heal_path_matches_bucket_child(heal_path, bucket, prefix),
         HealType::ErasureSet { set_disk_id, .. } => heal_path == set_disk_id,
@@ -629,7 +631,8 @@ fn heal_type_path_view(heal_type: &HealType) -> Option<(&str, &str, bool)> {
         HealType::Prefix { bucket, prefix } => Some((bucket, prefix, true)),
         HealType::Object { bucket, object, .. }
         | HealType::Metadata { bucket, object }
-        | HealType::ECDecode { bucket, object, .. } => Some((bucket, object, false)),
+        | HealType::ECDecode { bucket, object, .. }
+        | HealType::DeleteMarkerPurge { bucket, object, .. } => Some((bucket, object, false)),
         HealType::Cluster | HealType::ErasureSet { .. } => None,
     }
 }
@@ -1754,6 +1757,7 @@ impl HealManager {
         let kind = match &request.heal_type {
             HealType::Metadata { .. } => rustfs_common::mrf_channel::MrfKind::MetadataCorruption,
             HealType::ECDecode { .. } => rustfs_common::mrf_channel::MrfKind::DecodeFailure,
+            HealType::DeleteMarkerPurge { .. } => rustfs_common::mrf_channel::MrfKind::DeleteMarkerPurge,
             _ => rustfs_common::mrf_channel::MrfKind::PartialWrite,
         };
         self.submit_mrf_heal_request_with_receipt_and_identity(
@@ -1764,6 +1768,7 @@ impl HealManager {
                 version_id,
                 kind,
                 scope: None,
+                delete_marker_purge: None,
                 lease: None,
             },
         )
@@ -1775,7 +1780,7 @@ impl HealManager {
         request: HealRequest,
         mrf_notice_target: MrfRepairNoticeTarget,
     ) -> Result<HealAdmissionReceipt> {
-        let preserve_alias = mrf_notice_target.kind != rustfs_common::mrf_channel::MrfKind::PartialWrite;
+        let preserve_alias = !mrf_notice_target.kind.is_durable();
         self.submit_heal_request_with_receipt_alias_and_mrf_notice(request, preserve_alias, true, Some(mrf_notice_target))
             .await
     }
@@ -1784,6 +1789,9 @@ impl HealManager {
         &self,
         intent: &rustfs_common::mrf_channel::MrfIntent,
     ) -> Option<rustfs_common::mrf_channel::MrfDurableRepairAnchor> {
+        if let Some(purge) = intent.delete_marker_purge.as_ref() {
+            return rustfs_common::mrf_channel::MrfDurableRepairAnchor::from_intent(intent, purge.bucket_incarnation_id);
+        }
         match self.storage.mrf_bucket_incarnation_id(intent.bucket.as_ref()).await {
             Ok(Some(bucket_incarnation_id)) => {
                 rustfs_common::mrf_channel::MrfDurableRepairAnchor::from_intent(intent, bucket_incarnation_id)
@@ -2030,10 +2038,7 @@ impl HealManager {
             // of the silent merge (MinIO's ErrHealAlreadyRunning).
             let admission = if (request.source == HealRequestSource::Admin
                 && config.overlap_policy == HealOverlapPolicy::MinioError)
-                || (duplicate_state != "queued"
-                    && mrf_notice_target
-                        .as_ref()
-                        .is_some_and(|target| target.kind == rustfs_common::mrf_channel::MrfKind::PartialWrite))
+                || (duplicate_state != "queued" && mrf_notice_target.as_ref().is_some_and(|target| target.kind.is_durable()))
             {
                 // A running/retrying task may have observed the object before
                 // this write committed. Its receipt cannot prove the new lease.

crates/heal/src/heal/manager/queue.rs

@@ -126,6 +126,17 @@ impl CompletedHealStatus {
                 add(object.capacity());
                 add(version_id.as_ref().map_or(0, String::capacity));
             }
+            HealType::DeleteMarkerPurge {
+                bucket,
+                object,
+                version_id,
+                purge,
+            } => {
+                add(bucket.capacity());
+                add(object.capacity());
+                add(version_id.capacity());
+                add(purge.marker.len());
+            }
             HealType::Prefix { bucket, prefix } => {
                 add(bucket.capacity());
                 add(prefix.capacity());
@@ -490,6 +501,17 @@ impl PriorityHealQueue {
             } => {
                 format!("ecdecode:{}:{}:{}", bucket, object, version_id.as_deref().unwrap_or(""))
             }
+            HealType::DeleteMarkerPurge {
+                bucket,
+                object,
+                version_id,
+                purge,
+            } => format!(
+                "delete-marker-purge:{bucket}:{object}:{version_id}:{}:{}:{}",
+                purge.bucket_incarnation_id,
+                purge.marker_incarnation_id,
+                base64_simd::URL_SAFE_NO_PAD.encode_to_string(purge.marker_identity)
+            ),
         }
     }
 

crates/heal/src/heal/manager/root_recovery.rs

@@ -149,9 +149,11 @@ impl RecoveryHealType {
     }
 }
 
-impl From<&HealType> for RecoveryHealType {
-    fn from(heal_type: &HealType) -> Self {
-        match heal_type {
+impl TryFrom<&HealType> for RecoveryHealType {
+    type Error = Error;
+
+    fn try_from(heal_type: &HealType) -> Result<Self> {
+        Ok(match heal_type {
             HealType::Cluster => Self::Cluster,
             HealType::Bucket { bucket } => Self::Bucket { bucket: bucket.clone() },
             HealType::Object {
@@ -185,7 +187,10 @@ impl From<&HealType> for RecoveryHealType {
                 object: object.clone(),
                 version_id: version_id.clone(),
             },
-        }
+            HealType::DeleteMarkerPurge { .. } => {
+                return Err(Error::Other("Delete-marker purge recovery is owned by the MRF journal".to_string()));
+            }
+        })
     }
 }
 
@@ -309,28 +314,28 @@ impl RootHealTerminal {
         Ok(())
     }
 
-    fn from_completed(task_id: &str, completed: &CompletedHealStatus) -> Self {
-        Self {
+    fn from_completed(task_id: &str, completed: &CompletedHealStatus) -> Result<Self> {
+        Ok(Self {
             schema: ROOT_TERMINAL_SCHEMA,
             task_id: task_id.to_owned(),
-            heal_type: RecoveryHealType::from(&completed.heal_type),
+            heal_type: RecoveryHealType::try_from(&completed.heal_type)?,
             status: completed.status.clone(),
             options: completed.options.clone(),
             progress: completed.progress.clone(),
             completed_at: completed.completed_at,
-        }
+        })
     }
 
-    fn cancelled(task_id: &str, heal_type: &HealType, options: HealOptions) -> Self {
-        Self {
+    fn cancelled(task_id: &str, heal_type: &HealType, options: HealOptions) -> Result<Self> {
+        Ok(Self {
             schema: ROOT_TERMINAL_SCHEMA,
             task_id: task_id.to_owned(),
-            heal_type: RecoveryHealType::from(heal_type),
+            heal_type: RecoveryHealType::try_from(heal_type)?,
             status: HealTaskStatus::Cancelled,
             options,
             progress: None,
             completed_at: SystemTime::now(),
-        }
+        })
     }
 
     fn into_completed(self) -> CompletedHealStatus {
@@ -357,17 +362,17 @@ impl RootHealTerminal {
 }
 
 impl RootHealIntent {
-    fn from_request(request: &HealRequest) -> Self {
-        Self {
+    fn from_request(request: &HealRequest) -> Result<Self> {
+        Ok(Self {
             schema: ROOT_RECOVERY_SCHEMA,
             task_id: request.id.clone(),
-            heal_type: RecoveryHealType::from(&request.heal_type),
+            heal_type: RecoveryHealType::try_from(&request.heal_type)?,
             bucket_incarnation_id: request.bucket_incarnation_id,
             options: request.options.clone(),
             priority: request.priority,
             retry_attempts: request.retry_attempts,
             created_at: request.created_at,
-        }
+        })
     }
 
     fn into_request(self) -> HealRequest {
@@ -919,7 +924,7 @@ impl RootHealRecovery {
         if request.options.no_lock {
             return Err(Error::Other("Administrator root heal cannot skip namespace locking".to_string()));
         }
-        let intent = RootHealIntent::from_request(request);
+        let intent = RootHealIntent::from_request(request)?;
         intent.heal_type.validate()?;
         let bytes =
             serde_json::to_vec(&intent).map_err(|error| Error::Other(format!("Serialize root heal recovery record: {error}")))?;
@@ -1076,7 +1081,7 @@ impl RootHealRecovery {
         };
         let pending = decode_intent(task_id, &bytes)?;
         let heal_type = HealType::from(pending.heal_type);
-        let terminal = RootHealTerminal::cancelled(task_id, &heal_type, pending.options);
+        let terminal = RootHealTerminal::cancelled(task_id, &heal_type, pending.options)?;
         let completed = terminal.clone().into_completed();
         let _ = Self::persist_terminal_locked(&disks, task_id, terminal, &completed).await?;
         match EcstoreDiskAPI::compare_and_update_file(
@@ -1110,7 +1115,7 @@ impl RootHealRecovery {
         let _guard = self.mutation.lock().await;
         let disks = self.disks().await?;
         let pending =
-            Self::persist_terminal_locked(&disks, task_id, RootHealTerminal::from_completed(task_id, completed), completed)
+            Self::persist_terminal_locked(&disks, task_id, RootHealTerminal::from_completed(task_id, completed)?, completed)
                 .await?;
         #[cfg(test)]
         if self
@@ -1373,7 +1378,7 @@ impl HealManager {
                 {
                     Ok(()) => {}
                     Err(error @ Error::StaleBucketIncarnation { .. }) => {
-                        let mut terminal = RootHealTerminal::cancelled(&request.id, &request.heal_type, request.options.clone());
+                        let mut terminal = RootHealTerminal::cancelled(&request.id, &request.heal_type, request.options.clone())?;
                         terminal.status = HealTaskStatus::Failed {
                             error: error.to_string(),
                         };

crates/heal/src/heal/manager/root_recovery/report.rs

@@ -203,7 +203,8 @@ mod tests {
     #[test]
     fn root_report_encoding_rejects_oversized_payload() {
         let task_id = Uuid::new_v4().to_string();
-        let terminal = RootHealTerminal::cancelled(&task_id, &HealType::Cluster, HealOptions::default());
+        let terminal = RootHealTerminal::cancelled(&task_id, &HealType::Cluster, HealOptions::default())
+            .expect("cluster is an administrator recovery type");
         let mut completed = terminal.clone().into_completed();
         completed.progress = Some(HealProgress {
             current_object: Some("x".repeat(MAX_ROOT_REPORT_BYTES)),

crates/heal/src/heal/manager/scheduler.rs

@@ -818,6 +818,7 @@ pub(super) fn mrf_verified_repair_event_for_target(
         MrfKind::DecodeFailure => HealObjectKind::Decode,
         MrfKind::MetadataCorruption => HealObjectKind::Metadata,
         MrfKind::PartialWrite => HealObjectKind::Object,
+        MrfKind::DeleteMarkerPurge => HealObjectKind::DeleteMarkerPurge,
     };
     if outcome.identity.kind != expected_kind
         || outcome.identity.bucket.as_str() != target.bucket.as_ref()
@@ -848,6 +849,7 @@ pub(super) fn mrf_verified_repair_event_for_target(
         object: target.object.clone(),
         version_id,
         scope,
+        delete_marker_purge: target.delete_marker_purge,
         lease: target.lease,
         bucket_incarnation_id,
         disposition,

crates/heal/src/heal/manager/tests.rs

@@ -1226,6 +1226,7 @@ fn mrf_verified_repair_event_requires_positive_exact_identity() {
             pool_index: 1,
             set_index: 2,
         }),
+        delete_marker_purge: None,
         lease: None,
     };
     let matching = HealObjectOutcome {
@@ -1352,6 +1353,7 @@ fn completed_mrf_notice_publishes_only_verified_positive_events() {
             pool_index: 1,
             set_index: 2,
         }),
+        delete_marker_purge: None,
         lease: None,
     };
     let mismatch_target = MrfRepairNoticeTarget {

crates/heal/src/heal/mrf_queue.rs

@@ -26,8 +26,8 @@ use super::{DiskStore, HealDiskExt as _, local_disk_map_read};
 use crate::heal::manager::{HealManager, MrfRepairNoticeTarget};
 use metrics::{counter, gauge};
 use rustfs_common::mrf_channel::{
-    MRF_MAX_ATTEMPTS, MrfDurableAdmissionError, MrfDurableRepairAnchor, MrfDurableSubmission, MrfIngressResult, MrfIntent,
-    MrfKind,
+    MRF_MAX_ATTEMPTS, MRF_MAX_DELETE_MARKER_BYTES, MrfDeleteMarkerPurge, MrfDeleteMarkerPurgeIdentity, MrfDurableAdmissionError,
+    MrfDurableRepairAnchor, MrfDurableSubmission, MrfIngressResult, MrfIntent, MrfKind,
 };
 use rustfs_heal_contracts::heal_channel::{HealAdmissionDropReason, HealAdmissionResult};
 use std::collections::{HashSet, VecDeque};
@@ -48,7 +48,7 @@ pub mod snapshot;
 /// layout.
 pub(crate) const MRF_JOURNAL_PATH: &str = "buckets/.heal/mrf/journal.bin";
 /// The scoped path is the authoritative snapshot for new readers and carries
-/// both v1 and v2 records. The legacy path is only a v1 compatibility mirror;
+/// v1, v2, and v3 records. The legacy path is only a v1 compatibility mirror;
 /// older readers ignore the authoritative path, while new readers never merge
 /// the two files. This prevents a partial two-file flush from fabricating a
 /// mixed epoch.
@@ -59,6 +59,7 @@ const MRF_JOURNAL_FORMAT: u8 = 1;
 /// Record layout version.
 const MRF_JOURNAL_VERSION: u8 = 1;
 const MRF_JOURNAL_VERSION_SCOPED: u8 = 2;
+const MRF_JOURNAL_VERSION_DELETE_MARKER_PURGE: u8 = 3;
 
 /// Fixed header size: format, version, kind, attempts, enqueued_at_ms,
 /// has_version flag.
@@ -125,6 +126,7 @@ struct MrfQueueKey {
     object: Arc<str>,
     version_id: Option<[u8; 16]>,
     scope: Option<rustfs_common::mrf_channel::MrfScope>,
+    delete_marker_purge: Option<MrfDeleteMarkerPurgeIdentity>,
 }
 
 fn queue_key(intent: &MrfIntent) -> MrfQueueKey {
@@ -138,6 +140,7 @@ fn queue_key(intent: &MrfIntent) -> MrfQueueKey {
         object: intent.object.clone(),
         version_id,
         scope,
+        delete_marker_purge: intent.delete_marker_purge.as_ref().map(MrfDeleteMarkerPurge::identity),
     }
 }
 
@@ -234,9 +237,19 @@ pub(crate) fn encode_intent(intent: &MrfIntent, out: &mut Vec<u8>) -> bool {
         .then_some(intent.scope)
         .flatten();
     let version_id = intent.version_id.filter(|bytes| *bytes != [0; 16]);
+    let purge = match intent.kind {
+        MrfKind::DeleteMarkerPurge => match intent.delete_marker_purge.as_ref().filter(|payload| payload.is_valid()) {
+            Some(payload) if version_id.is_some() && scope.is_some() => Some(payload),
+            _ => return false,
+        },
+        _ if intent.delete_marker_purge.is_none() => None,
+        _ => return false,
+    };
     let start = out.len();
     out.push(MRF_JOURNAL_FORMAT);
-    out.push(if scope.is_some() {
+    out.push(if purge.is_some() {
+        MRF_JOURNAL_VERSION_DELETE_MARKER_PURGE
+    } else if scope.is_some() {
         MRF_JOURNAL_VERSION_SCOPED
     } else {
         MRF_JOURNAL_VERSION
@@ -245,6 +258,7 @@ pub(crate) fn encode_intent(intent: &MrfIntent, out: &mut Vec<u8>) -> bool {
         rustfs_common::mrf_channel::MrfKind::DecodeFailure => 1,
         rustfs_common::mrf_channel::MrfKind::MetadataCorruption => 2,
         rustfs_common::mrf_channel::MrfKind::PartialWrite => 3,
+        rustfs_common::mrf_channel::MrfKind::DeleteMarkerPurge => 4,
     });
     out.push(intent.attempts);
     out.extend_from_slice(&intent.enqueued_at_ms.to_le_bytes());
@@ -263,6 +277,17 @@ pub(crate) fn encode_intent(intent: &MrfIntent, out: &mut Vec<u8>) -> bool {
     out.extend_from_slice(&object_len.to_le_bytes());
     out.extend_from_slice(intent.bucket.as_bytes());
     out.extend_from_slice(intent.object.as_bytes());
+    if let Some(purge) = purge {
+        let Ok(marker_len) = u32::try_from(purge.marker.len()) else {
+            out.truncate(start);
+            return false;
+        };
+        out.extend_from_slice(purge.bucket_incarnation_id.as_bytes());
+        out.extend_from_slice(purge.marker_incarnation_id.as_bytes());
+        out.extend_from_slice(&purge.marker_identity);
+        out.extend_from_slice(&marker_len.to_le_bytes());
+        out.extend_from_slice(&purge.marker);
+    }
     let mut hasher = crc_fast::Digest::new(crc_fast::CrcAlgorithm::Crc32IsoHdlc);
     hasher.update(&out[start..]);
     let Ok(checksum) = u32::try_from(hasher.finalize()) else {
@@ -277,15 +302,24 @@ fn decode_one(data: &[u8]) -> Option<(MrfIntent, usize)> {
     if data.len() < MRF_RECORD_FIXED_HEAD + 8 {
         return None;
     }
-    if data[0] != MRF_JOURNAL_FORMAT || !matches!(data[1], MRF_JOURNAL_VERSION | MRF_JOURNAL_VERSION_SCOPED) {
+    if data[0] != MRF_JOURNAL_FORMAT
+        || !matches!(
+            data[1],
+            MRF_JOURNAL_VERSION | MRF_JOURNAL_VERSION_SCOPED | MRF_JOURNAL_VERSION_DELETE_MARKER_PURGE
+        )
+    {
         return None;
     }
     let kind = match data[2] {
         1 => rustfs_common::mrf_channel::MrfKind::DecodeFailure,
         2 => rustfs_common::mrf_channel::MrfKind::MetadataCorruption,
         3 => rustfs_common::mrf_channel::MrfKind::PartialWrite,
+        4 => rustfs_common::mrf_channel::MrfKind::DeleteMarkerPurge,
         _ => return None,
     };
+    if (kind == MrfKind::DeleteMarkerPurge) != (data[1] == MRF_JOURNAL_VERSION_DELETE_MARKER_PURGE) {
+        return None;
+    }
     let attempts = data[3];
     let enqueued_at_ms = u64::from_le_bytes(data[4..12].try_into().ok()?);
     let has_version = match data[12] {
@@ -304,7 +338,10 @@ fn decode_one(data: &[u8]) -> Option<(MrfIntent, usize)> {
     } else {
         None
     };
-    let scope = if data[1] == MRF_JOURNAL_VERSION_SCOPED {
+    if kind == MrfKind::DeleteMarkerPurge && version_id.is_none_or(|bytes| bytes == [0; 16]) {
+        return None;
+    }
+    let scope = if matches!(data[1], MRF_JOURNAL_VERSION_SCOPED | MRF_JOURNAL_VERSION_DELETE_MARKER_PURGE) {
         if data.len() < cursor + 8 {
             return None;
         }
@@ -324,7 +361,43 @@ fn decode_one(data: &[u8]) -> Option<(MrfIntent, usize)> {
         return None;
     }
     cursor += 8;
-    let body_end = cursor.checked_add(bucket_len)?.checked_add(object_len)?;
+    let identity_end = cursor.checked_add(bucket_len)?.checked_add(object_len)?;
+    if data.len() < identity_end {
+        return None;
+    }
+    let bucket = std::sync::Arc::from(std::str::from_utf8(&data[cursor..cursor + bucket_len]).ok()?);
+    let object = std::sync::Arc::from(std::str::from_utf8(&data[cursor + bucket_len..identity_end]).ok()?);
+    let (delete_marker_purge, body_end) = if kind == MrfKind::DeleteMarkerPurge {
+        const PURGE_FIXED: usize = 16 + 16 + 32 + 4;
+        if data.len() < identity_end.checked_add(PURGE_FIXED)? {
+            return None;
+        }
+        let mut purge_cursor = identity_end;
+        let bucket_incarnation_id = Uuid::from_slice(&data[purge_cursor..purge_cursor + 16]).ok()?;
+        purge_cursor += 16;
+        let marker_incarnation_id = Uuid::from_slice(&data[purge_cursor..purge_cursor + 16]).ok()?;
+        purge_cursor += 16;
+        let marker_identity = data[purge_cursor..purge_cursor + 32].try_into().ok()?;
+        purge_cursor += 32;
+        let marker_len = usize::try_from(u32::from_le_bytes(data[purge_cursor..purge_cursor + 4].try_into().ok()?)).ok()?;
+        purge_cursor += 4;
+        if marker_len == 0 || marker_len > MRF_MAX_DELETE_MARKER_BYTES {
+            return None;
+        }
+        let body_end = purge_cursor.checked_add(marker_len)?;
+        if data.len() < body_end {
+            return None;
+        }
+        let payload = MrfDeleteMarkerPurge::new(
+            bucket_incarnation_id,
+            marker_incarnation_id,
+            marker_identity,
+            data[purge_cursor..body_end].to_vec(),
+        )?;
+        (Some(payload), body_end)
+    } else {
+        (None, identity_end)
+    };
     let record_end = body_end.checked_add(4)?;
     if data.len() < record_end {
         return None;
@@ -334,14 +407,13 @@ fn decode_one(data: &[u8]) -> Option<(MrfIntent, usize)> {
     if u32::try_from(hasher.finalize()).ok()? != u32::from_le_bytes(data[body_end..record_end].try_into().ok()?) {
         return None;
     }
-    let bucket = std::sync::Arc::from(std::str::from_utf8(&data[cursor..cursor + bucket_len]).ok()?);
-    let object = std::sync::Arc::from(std::str::from_utf8(&data[cursor + bucket_len..body_end]).ok()?);
     Some((
         MrfIntent {
             bucket,
             object,
             version_id,
             kind,
+            delete_marker_purge,
             scope: if matches!(kind, rustfs_common::mrf_channel::MrfKind::MetadataCorruption) {
                 None
             } else {
@@ -482,6 +554,18 @@ pub(crate) fn build_heal_request(intent: &MrfIntent) -> HealRequest {
             },
             HealPriority::Normal,
         ),
+        rustfs_common::mrf_channel::MrfKind::DeleteMarkerPurge => (
+            HealType::DeleteMarkerPurge {
+                bucket,
+                object,
+                version_id: version_id.expect("validated delete-marker purge requires a version"),
+                purge: intent
+                    .delete_marker_purge
+                    .clone()
+                    .expect("validated delete-marker purge requires a payload"),
+            },
+            HealPriority::Normal,
+        ),
     };
     let mut options = HealOptions::default();
     if matches!(intent.kind, rustfs_common::mrf_channel::MrfKind::PartialWrite) {
@@ -510,6 +594,7 @@ async fn submit_mrf_heal_request(manager: &HealManager, intent: &MrfIntent) -> c
                 version_id: intent.version_id,
                 kind: intent.kind,
                 scope: intent.scope,
+                delete_marker_purge: intent.delete_marker_purge.as_ref().map(MrfDeleteMarkerPurge::identity),
                 lease: intent.lease,
             },
         )
@@ -536,7 +621,7 @@ struct MrfRuntime {
     /// True when replay observed a responsibility that cannot be discharged by
     /// a complete verified repair proof in this process.
     retain_replay_journal: bool,
-    /// Partial-write responsibilities accepted from replay and waiting for an
+    /// Durable responsibilities accepted from replay and waiting for an
     /// exact storage-owned proof before the startup journal can be deleted.
     durable_replay_anchors: Vec<MrfDurableRepairAnchor>,
     /// Startup replay source to remove after the retained replay
@@ -568,7 +653,7 @@ impl MrfRuntime {
         // later write replaces this lease before its old proof arrives.
         let adopted_leases: HashSet<_> = self.partial_writes.intents().filter_map(|intent| intent.lease).collect();
         self.durable_replay_anchors
-            .retain(|anchor| anchor.kind != MrfKind::PartialWrite || !adopted_leases.contains(&anchor.lease));
+            .retain(|anchor| !anchor.kind.is_durable() || !adopted_leases.contains(&anchor.lease));
     }
 
     fn admit_partial_write(&mut self, intent: MrfIntent) -> Result<(), MrfDurableAdmissionError> {
@@ -664,7 +749,7 @@ impl MrfRuntime {
             self.backoff_until = None;
         }
         while let Some(mut intent) = self.queue.pop_front() {
-            if intent.kind == MrfKind::PartialWrite {
+            if intent.kind.is_durable() {
                 if self.admit_partial_write(intent.clone()).is_err() {
                     self.queue.push_back(intent);
                     break;
@@ -780,9 +865,10 @@ impl MrfRuntime {
     }
 }
 
-/// Initialize the global MRF channel (honoring `RUSTFS_HEAL_MRF_ENABLE`) and
-/// spawn the consumer task. Called once from the heal runtime bootstrap right
-/// after the manager started; a disabled feature or a double call is a no-op.
+/// Initialize the global MRF channels and spawn the consumer task. The
+/// `RUSTFS_HEAL_MRF_ENABLE` switch only disables best-effort ingress; durable
+/// responsibility replay remains active so committed marker purges survive the
+/// ordinary MRF kill-switch.
 /// Public for integration tests that drive the real consumer loop.
 pub fn spawn_mrf_consumer(manager: Arc<HealManager>) {
     let enabled = rustfs_utils::get_env_bool(rustfs_config::ENV_HEAL_MRF_ENABLE, rustfs_config::DEFAULT_HEAL_MRF_ENABLE);
@@ -790,9 +876,8 @@ pub fn spawn_mrf_consumer(manager: Arc<HealManager>) {
     if !enabled {
         tracing::info!(
             target: "rustfs::heal::mrf",
-            "MRF intent pipeline disabled by configuration; producers will not deliver"
+            "Best-effort MRF ingress disabled by configuration; durable responsibilities remain active"
         );
-        return;
     }
     let (receiver, durable_receiver) = match rustfs_common::mrf_channel::init_mrf_channel().and_then(|receiver| {
         rustfs_common::mrf_channel::init_durable_mrf_channel().map(|durable_receiver| (receiver, durable_receiver))
@@ -995,7 +1080,7 @@ async fn replay_into(
                 *backoff_until = Some(tokio::time::Instant::now());
                 break;
             }
-            if intent.kind == MrfKind::PartialWrite {
+            if intent.kind.is_durable() {
                 // Preserve the executable record as well as its proof anchor:
                 // a target that is still offline during replay needs live retries.
                 if let Some(anchor) = manager.durable_mrf_repair_anchor(&intent).await {
@@ -1146,7 +1231,7 @@ async fn run_mrf_consumer(
                     return;
                 }
                 for intent in batch.drain(..) {
-                    if intent.kind == MrfKind::PartialWrite {
+                    if intent.kind.is_durable() {
                         if runtime.admit_partial_write(intent.clone()).is_err() {
                             rustfs_common::mrf_channel::release_mrf_intent(&intent);
                             counter!("rustfs_heal_mrf_dropped_total", "reason" => "queue_overflow").increment(1);
@@ -1273,6 +1358,7 @@ mod tests {
             object: StdArc::from(object),
             version_id: Some([7u8; 16]),
             kind: MrfKind::DecodeFailure,
+            delete_marker_purge: None,
             scope: None,
             lease: None,
             enqueued_at_ms: 1_700_000_000_000,
@@ -1280,6 +1366,38 @@ mod tests {
         }
     }
 
+    fn purge_intent(marker: Vec<u8>) -> MrfIntent {
+        let bucket_incarnation_id = Uuid::new_v4();
+        MrfIntent {
+            bucket: StdArc::from("purge-bucket"),
+            object: StdArc::from("markers/object.bin"),
+            version_id: Some(*Uuid::new_v4().as_bytes()),
+            kind: MrfKind::DeleteMarkerPurge,
+            delete_marker_purge: MrfDeleteMarkerPurge::new(bucket_incarnation_id, bucket_incarnation_id, [9; 32], marker),
+            scope: Some(rustfs_common::mrf_channel::MrfScope {
+                pool_index: 2,
+                set_index: 3,
+            }),
+            lease: None,
+            enqueued_at_ms: 1_700_000_000_001,
+            attempts: 2,
+        }
+    }
+
+    fn rewrite_record_crc(record: &mut [u8]) {
+        let crc_offset = record.len() - 4;
+        let mut hasher = crc_fast::Digest::new(crc_fast::CrcAlgorithm::Crc32IsoHdlc);
+        hasher.update(&record[..crc_offset]);
+        let checksum = u32::try_from(hasher.finalize()).expect("CRC32 fits");
+        record[crc_offset..].copy_from_slice(&checksum.to_le_bytes());
+    }
+
+    fn assert_journal_record_rejected(record: &[u8]) {
+        let (decoded, truncated) = decode_journal(record);
+        assert!(decoded.is_empty());
+        assert_eq!(truncated, record.len());
+    }
+
     fn encoded_payload(intent: &MrfIntent) -> Vec<u8> {
         let mut payload = Vec::new();
         assert!(encode_intent(intent, &mut payload), "fixture intent must encode");
@@ -1432,6 +1550,7 @@ mod tests {
             object: anchor.object.clone(),
             version_id: anchor.version_id,
             scope: anchor.scope,
+            delete_marker_purge: anchor.delete_marker_purge,
             lease: Some(anchor.lease),
             bucket_incarnation_id: anchor.bucket_incarnation_id,
             disposition: MrfVerifiedRepairDisposition::Repaired,
@@ -2113,6 +2232,7 @@ mod tests {
             object: intent.object.clone(),
             version_id: intent.version_id,
             scope: intent.scope,
+            delete_marker_purge: intent.delete_marker_purge.as_ref().map(MrfDeleteMarkerPurge::identity),
             lease: intent.lease,
             bucket_incarnation_id,
             disposition: MrfVerifiedRepairDisposition::Repaired,
@@ -2368,6 +2488,7 @@ mod tests {
             object: anchor.object.clone(),
             version_id: anchor.version_id,
             scope: anchor.scope,
+            delete_marker_purge: anchor.delete_marker_purge,
             lease: Some(anchor.lease),
             bucket_incarnation_id: anchor.bucket_incarnation_id,
             disposition: MrfVerifiedRepairDisposition::Repaired,
@@ -2629,6 +2750,7 @@ mod tests {
                 object: StdArc::from("object/c"),
                 version_id: None,
                 kind: MrfKind::MetadataCorruption,
+                delete_marker_purge: None,
                 scope: None,
                 lease: None,
                 enqueued_at_ms: 5,
@@ -2651,6 +2773,76 @@ mod tests {
         }
     }
 
+    #[test]
+    fn journal_v3_roundtrip_preserves_delete_marker_purge_payload() {
+        let intent = purge_intent(vec![1, 2, 3, 4, 5]);
+        let mut buf = Vec::new();
+        assert!(encode_intent(&intent, &mut buf));
+        assert_eq!(buf[1], MRF_JOURNAL_VERSION_DELETE_MARKER_PURGE);
+        assert_eq!(buf[2], 4);
+
+        let (decoded, truncated) = decode_journal(&buf);
+        assert_eq!(truncated, 0);
+        assert_eq!(decoded.len(), 1);
+        let decoded = &decoded[0];
+        assert_eq!(decoded.bucket, intent.bucket);
+        assert_eq!(decoded.object, intent.object);
+        assert_eq!(decoded.version_id, intent.version_id);
+        assert_eq!(decoded.kind, MrfKind::DeleteMarkerPurge);
+        assert_eq!(decoded.scope, intent.scope);
+        assert_eq!(decoded.delete_marker_purge, intent.delete_marker_purge);
+        assert_eq!(decoded.attempts, intent.attempts);
+    }
+
+    #[test]
+    fn journal_v3_rejects_nil_version_wrong_layout_corruption_and_truncation() {
+        let intent = purge_intent(vec![1, 2, 3, 4, 5]);
+        let mut encoded = Vec::new();
+        assert!(encode_intent(&intent, &mut encoded));
+
+        let mut nil_version = encoded.clone();
+        nil_version[13..29].fill(0);
+        rewrite_record_crc(&mut nil_version);
+        assert_journal_record_rejected(&nil_version);
+
+        let mut wrong_kind = encoded.clone();
+        wrong_kind[2] = 3;
+        rewrite_record_crc(&mut wrong_kind);
+        assert_journal_record_rejected(&wrong_kind);
+
+        let mut wrong_layout = encoded.clone();
+        wrong_layout[1] = MRF_JOURNAL_VERSION_SCOPED;
+        rewrite_record_crc(&mut wrong_layout);
+        assert_journal_record_rejected(&wrong_layout);
+
+        let marker_byte = encoded.len() - 5;
+        encoded[marker_byte] ^= 0xff;
+        assert_journal_record_rejected(&encoded);
+
+        let torn = &encoded[..encoded.len() - 1];
+        assert_journal_record_rejected(torn);
+    }
+
+    #[test]
+    fn journal_v3_rejects_missing_or_oversized_payload_without_mutating_output() {
+        let mut missing = purge_intent(vec![1]);
+        missing.delete_marker_purge = None;
+        let mut output = vec![7];
+        assert!(!encode_intent(&missing, &mut output));
+        assert_eq!(output, vec![7]);
+
+        let bucket_incarnation_id = Uuid::new_v4();
+        let mut oversized = purge_intent(vec![1]);
+        oversized.delete_marker_purge = Some(MrfDeleteMarkerPurge {
+            bucket_incarnation_id,
+            marker_incarnation_id: bucket_incarnation_id,
+            marker_identity: [3; 32],
+            marker: StdArc::from(vec![0; MRF_MAX_DELETE_MARKER_BYTES + 1]),
+        });
+        assert!(!encode_intent(&oversized, &mut output));
+        assert_eq!(output, vec![7]);
+    }
+
     #[test]
     fn journal_torn_tail_is_truncated() {
         let mut buf = Vec::new();
@@ -2701,6 +2893,7 @@ mod tests {
             object: StdArc::from("o"),
             version_id: None,
             kind: MrfKind::MetadataCorruption,
+            delete_marker_purge: None,
             scope: None,
             lease: None,
             enqueued_at_ms: 0,
@@ -2714,6 +2907,7 @@ mod tests {
             object: StdArc::from("o"),
             version_id: None,
             kind: MrfKind::PartialWrite,
+            delete_marker_purge: None,
             scope: None,
             lease: None,
             enqueued_at_ms: 0,

crates/heal/src/heal/mrf_queue/partial_write.rs

@@ -24,7 +24,7 @@ struct Responsibility {
     next_attempt: Instant,
 }
 
-/// Partial writes have no rediscovery producer when scanning is disabled.
+/// Durable storage responsibilities have no guaranteed rediscovery producer.
 /// Admission, task failure and retry exhaustion cannot release their records.
 #[derive(Default)]
 pub(super) struct PartialWrites {
@@ -169,6 +169,7 @@ mod tests {
             object: Arc::from(object),
             version_id: None,
             kind: MrfKind::PartialWrite,
+            delete_marker_purge: None,
             scope: Some(MrfScope {
                 pool_index: 0,
                 set_index: 0,

crates/heal/src/heal/mrf_queue/snapshot.rs

@@ -701,6 +701,7 @@ mod tests {
             object: Arc::from(object),
             version_id: None,
             kind: MrfKind::PartialWrite,
+            delete_marker_purge: None,
             scope: None,
             lease: None,
             enqueued_at_ms: 1234,
@@ -719,6 +720,7 @@ mod tests {
                 object: Arc::from(format!("o-{index:06}")),
                 version_id: None,
                 kind: MrfKind::PartialWrite,
+                delete_marker_purge: None,
                 scope: None,
                 lease: None,
                 enqueued_at_ms: 1234,
@@ -1699,6 +1701,7 @@ mod tests {
                 object: Arc::from("a"),
                 version_id: None,
                 kind: MrfKind::PartialWrite,
+                delete_marker_purge: None,
                 scope: Some(MrfScope {
                     pool_index: 0,
                     set_index,

crates/heal/src/heal/mrf_queue/snapshot/migration.rs

@@ -584,6 +584,7 @@ mod tests {
                 object: Arc::from(object),
                 version_id: None,
                 kind,
+                delete_marker_purge: None,
                 scope,
                 lease: None,
                 enqueued_at_ms: 1,

crates/heal/src/heal/outcome.rs

@@ -29,6 +29,7 @@ pub enum HealObjectKind {
     Object,
     Metadata,
     Decode,
+    DeleteMarkerPurge,
 }
 
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]

crates/heal/src/heal/storage.rs

@@ -15,6 +15,7 @@
 use crate::{Error, Result};
 use async_trait::async_trait;
 use base64_simd::URL_SAFE_NO_PAD;
+use rustfs_common::mrf_channel::MrfDeleteMarkerPurge;
 use rustfs_heal_contracts::heal_channel::{DriveState, HealOpts, HealScanMode};
 use rustfs_madmin::heal_commands::HealResultItem;
 use serde::{Deserialize, Serialize};
@@ -401,6 +402,18 @@ pub struct HealListItem {
 /// Heal storage layer interface
 #[async_trait]
 pub trait HealStorageAPI: Send + Sync {
+    /// Delete one exact marker from every member of its original erasure set.
+    async fn purge_delete_marker(
+        &self,
+        _bucket: &str,
+        _object: &str,
+        _version_id: &str,
+        _purge: &MrfDeleteMarkerPurge,
+        _opts: &HealOpts,
+    ) -> Result<HealStorageObjectResult> {
+        Err(Error::other("delete-marker purge is unsupported by this heal storage backend"))
+    }
+
     /// Get object meta
     ///
     /// Reserved for HS-01 MRF wiring (rustfs/backlog#1865): MRF intents
@@ -846,6 +859,41 @@ fn is_transient_object_exists_error(err: &StorageError) -> bool {
 
 #[async_trait]
 impl HealStorageAPI for ECStoreHealStorage {
+    async fn purge_delete_marker(
+        &self,
+        bucket: &str,
+        object: &str,
+        version_id: &str,
+        purge: &MrfDeleteMarkerPurge,
+        opts: &HealOpts,
+    ) -> Result<HealStorageObjectResult> {
+        let removed = self
+            .ecstore
+            .purge_delete_marker_with_proof(bucket, object, version_id, purge, opts)
+            .await
+            .map_err(Error::Storage)?;
+        Ok(HealStorageObjectResult {
+            item: HealResultItem::default(),
+            error: None,
+            receipt: Some(HealObjectReceipt {
+                identity: HealObjectIdentity {
+                    kind: HealObjectKind::DeleteMarkerPurge,
+                    bucket: bucket.to_owned(),
+                    object: object.to_owned(),
+                    version_id: Some(version_id.to_owned()),
+                    bucket_incarnation_id: Some(purge.bucket_incarnation_id),
+                    pool_index: opts.pool,
+                    set_index: opts.set,
+                },
+                disposition: if removed {
+                    HealObjectDisposition::Repaired
+                } else {
+                    HealObjectDisposition::AuthoritativelyAbsent
+                },
+            }),
+        })
+    }
+
     async fn admit_bucket_incarnation(&self, bucket: &str) -> Result<Uuid> {
         match self.ecstore.bucket_incarnation_id_from_disk(bucket).await {
             Ok(id) if !id.is_nil() => Ok(id),

crates/heal/src/heal/task.rs

@@ -130,6 +130,13 @@ pub enum HealType {
         object: String,
         version_id: Option<String>,
     },
+    /// Complete an explicit delete-marker purge on its original erasure set.
+    DeleteMarkerPurge {
+        bucket: String,
+        object: String,
+        version_id: String,
+        purge: rustfs_common::mrf_channel::MrfDeleteMarkerPurge,
+    },
 }
 
 impl HealType {
@@ -142,6 +149,7 @@ impl HealType {
             Self::ErasureSet { .. } => "erasure_set",
             Self::Metadata { .. } => "metadata",
             Self::ECDecode { .. } => "ec_decode",
+            Self::DeleteMarkerPurge { .. } => "delete_marker_purge",
         }
     }
 
@@ -152,7 +160,10 @@ impl HealType {
     /// cannot amplify into per-object `info!`/`warn!` lines; aggregate kinds
     /// (cluster/bucket/prefix/erasure-set) keep operator-visible levels.
     pub(crate) fn is_per_object(&self) -> bool {
-        matches!(self, Self::Object { .. } | Self::Metadata { .. } | Self::ECDecode { .. })
+        matches!(
+            self,
+            Self::Object { .. } | Self::Metadata { .. } | Self::ECDecode { .. } | Self::DeleteMarkerPurge { .. }
+        )
     }
 }
 
@@ -598,6 +609,7 @@ impl HealTask {
             kind: match self.heal_type {
                 HealType::Metadata { .. } => HealObjectKind::Metadata,
                 HealType::ECDecode { .. } => HealObjectKind::Decode,
+                HealType::DeleteMarkerPurge { .. } => HealObjectKind::DeleteMarkerPurge,
                 _ => HealObjectKind::Object,
             },
             bucket: bucket.to_owned(),
@@ -634,6 +646,12 @@ impl HealTask {
                 version_id,
             } => (bucket, object, version_id.as_deref()),
             HealType::Metadata { bucket, object } => (bucket, object, None),
+            HealType::DeleteMarkerPurge {
+                bucket,
+                object,
+                version_id,
+                ..
+            } => (bucket, object, Some(version_id.as_str())),
             _ => return None,
         };
         Some(self.outcome_identity(bucket, object, version, self.options.pool_index, self.options.set_index))
@@ -752,6 +770,15 @@ impl HealTask {
                         None => event,
                     }
                 }
+                HealType::DeleteMarkerPurge {
+                    bucket,
+                    object,
+                    version_id,
+                    ..
+                } => event
+                    .with_bucket(bucket.as_str())
+                    .with_object(object.as_str())
+                    .with_attr("version_id", version_id.as_str()),
             };
 
             match error {
@@ -1091,6 +1118,12 @@ impl HealTask {
                     object,
                     version_id,
                 } => self.heal_ec_decode(bucket, object, version_id.as_deref()).await,
+                HealType::DeleteMarkerPurge {
+                    bucket,
+                    object,
+                    version_id,
+                    purge,
+                } => self.heal_delete_marker_purge(bucket, object, version_id, purge).await,
                 HealType::ErasureSet { buckets, set_disk_id } => {
                     self.heal_erasure_set(buckets.clone(), set_disk_id.clone()).await
                 }

crates/heal/src/heal/task/heal_object.rs

@@ -15,6 +15,47 @@
 use super::*;
 
 impl HealTask {
+    pub(super) async fn heal_delete_marker_purge(
+        &self,
+        bucket: &str,
+        object: &str,
+        version_id: &str,
+        purge: &rustfs_common::mrf_channel::MrfDeleteMarkerPurge,
+    ) -> Result<()> {
+        self.check_control_flags().await?;
+        let heal_opts = HealOpts {
+            recursive: false,
+            dry_run: self.options.dry_run,
+            remove: true,
+            recreate: false,
+            scan_mode: HealScanMode::Deep,
+            update_parity: false,
+            no_lock: self.options.no_lock,
+            read_repair: false,
+            pool: self.options.pool_index,
+            set: self.options.set_index,
+        };
+        let mut expected =
+            self.outcome_identity(bucket, object, Some(version_id), self.options.pool_index, self.options.set_index);
+        expected.bucket_incarnation_id = Some(purge.bucket_incarnation_id);
+        let storage_result = self
+            .await_with_control(
+                self.storage
+                    .purge_delete_marker(bucket, object, version_id, purge, &heal_opts),
+            )
+            .await?;
+        if let Some(error) = storage_result.error {
+            return Err(error);
+        }
+        if !self.record_verified_storage_receipt(expected, storage_result.receipt).await {
+            return Err(Error::TaskExecutionFailed {
+                message: format!("delete-marker purge returned no exact storage proof for {bucket}/{object}/{version_id}"),
+            });
+        }
+        self.progress.write().await.update_object_progress(1, 1, 0, 0, 0);
+        Ok(())
+    }
+
     // specific heal implementation method
     #[tracing::instrument(skip(self), fields(bucket = %bucket, object = %object, version_id = ?version_id))]
     #[hotpath::measure]

rustfs/src/storage/rpc/node_service.rs

@@ -1783,6 +1783,14 @@ impl Node for NodeService {
         &self,
         request: Request<DeleteVersionRequest>,
     ) -> Result<Response<DeleteVersionResponse>, Status> {
+        require_incarnation_body_digest(&request)?;
+        if Uuid::from_slice(&request.get_ref().bucket_incarnation_id)
+            .ok()
+            .filter(|id| !id.is_nil())
+            .is_none()
+        {
+            return Err(Status::invalid_argument("bucket incarnation must be a non-nil UUID"));
+        }
         self.handle_delete_version(request, true).await
     }
 
@@ -4423,7 +4431,7 @@ mod tests {
         assert_gated!(
             delete_retired_marker,
             DeleteVersionRequest {
-                bucket_incarnation_id: Default::default(),
+                bucket_incarnation_id: vec![1; 16].into(),
                 disk: disk.clone(),
                 volume: "v".into(),
                 path: "p".into(),
@@ -7382,7 +7390,8 @@ mod tests {
             (vec![0; 16], Some(vec![0; 16]), tonic::Code::InvalidArgument),
             (vec![1; 15], Some(vec![1; 15]), tonic::Code::InvalidArgument),
             (vec![2; 16], Some(vec![1; 16]), tonic::Code::PermissionDenied),
-            (Vec::new(), Some(vec![1; 16]), tonic::Code::PermissionDenied),
+            (Vec::new(), Some(vec![1; 16]), tonic::Code::InvalidArgument),
+            (Vec::new(), Some(Vec::new()), tonic::Code::InvalidArgument),
             (vec![1; 16], Some(vec![1; 16]), tonic::Code::FailedPrecondition),
         ] {
             let mut request = Request::new(message.clone());

rustfs/src/storage/rpc/node_service/disk.rs

@@ -753,6 +753,14 @@ impl NodeService {
             rustfs_protos::canonical_delete_version_request_body(request.get_ref()),
             "delete_version",
         )?;
+        if require_marker_condition
+            && Uuid::from_slice(&request.get_ref().bucket_incarnation_id)
+                .ok()
+                .filter(|id| !id.is_nil())
+                .is_none()
+        {
+            return Err(Status::invalid_argument("retired marker deletion requires a non-nil bucket incarnation"));
+        }
         let request = request.into_inner();
         let file_info = match decode_msgpack_or_json::<FileInfo>(&request.file_info_bin, &request.file_info, "FileInfo") {
             Ok(file_info) => file_info,