test(heal): cover stale current delete marker rejoin (#8021)

Co-authored-by: zhi22915 <qiuzgang@gmail.com>

crates/ecstore/src/io_support/shard_integrity.rs

@@ -587,7 +587,11 @@ pub(crate) async fn verify_deep_parts(
             return Err(DiskError::FileCorrupt);
         }
         let part_status = statuses.get_mut(&part_index).ok_or(DiskError::FileCorrupt)?;
-        let length = erasure.shard_file_offset(0, part.size, part.size);
+        // Deep verification reads the complete encoded shard. `shard_file_offset`
+        // is a range-read helper and can stop before the final encoded byte for
+        // a partial data block, which would let a one-byte tail truncation pass
+        // as healthy. Use the physical shard length for the integrity proof.
+        let length = usize::try_from(erasure.shard_file_size(part.size as i64)).map_err(|_| DiskError::FileCorrupt)?;
         let mut readers = Vec::with_capacity(disks.len());
         for (index, disk) in disks.iter().enumerate() {
             if part_status[index] != CHECK_PART_UNKNOWN {

crates/ecstore/src/services/notification_sys.rs

@@ -42,6 +42,7 @@ use std::sync::{
     atomic::{AtomicBool, AtomicUsize, Ordering},
 };
 use std::time::{Duration, Instant, SystemTime};
+use tokio::sync::Notify;
 use tokio::time::{sleep, timeout};
 use tokio_util::sync::CancellationToken;
 use tracing::{debug, error, info, warn};
@@ -61,6 +62,14 @@ const TIER_CONFIG_RELOAD_RETRY_CAP: Duration = Duration::from_secs(5);
 const REMOTE_VERSION_STATE_PROBE_INTERVAL: Duration = Duration::from_secs(10);
 const REMOTE_VERSION_STATE_PROBE_TIMEOUT: Duration = Duration::from_secs(5);
 const REMOTE_VERSION_STATE_PROOF_TTL: Duration = Duration::from_secs(30);
+/// Fallback poll cadence while startup has not yet published the notification
+/// system. IAM finalization starts the probe before `init_notification_runtime`
+/// runs, so the first pass always fails closed; waiting a full probe interval
+/// there left a single node without its durable quota capability for ten
+/// seconds after `/health` reported ok (rustfs/rustfs#8014). Publication wakes
+/// the probe immediately through `NOTIFICATION_SYS_PUBLISHED`; this bound only
+/// covers a wakeup that races the availability check.
+const REMOTE_VERSION_STATE_PROBE_BOOTSTRAP_INTERVAL: Duration = Duration::from_millis(100);
 const CROSS_POOL_FENCE_SUPPORTED_VERSION: u32 = 2;
 const TIER_DELETE_JOURNAL_POLICY_SUPPORTED_VERSION: u32 = 3;
 const DECOMMISSION_TARGET_FENCE_POLICY_SUPPORTED_VERSION: u32 = 4;
@@ -338,6 +347,9 @@ static LEGACY_TRANSITION_STATE_RECONCILE_FLEET_PROOF: OnceLock<std::sync::RwLock
 static ILM_RECOVERY_EXPORT_FLEET_PROOF: OnceLock<std::sync::RwLock<FleetCapabilityProofState>> = OnceLock::new();
 static TRANSITION_TRANSACTION_COMPACTION_FLEET_PROOF: OnceLock<std::sync::RwLock<FleetCapabilityProofState>> = OnceLock::new();
 static REMOTE_VERSION_STATE_PROBE_TOPOLOGY: OnceLock<String> = OnceLock::new();
+/// Signalled once `GLOBAL_NOTIFICATION_SYS` is published so the fleet probe can
+/// run its first real pass without waiting for the bootstrap poll.
+static NOTIFICATION_SYS_PUBLISHED: Notify = Notify::const_new();
 static ILM_RECOVERY_EXPORT_LOCAL_PROCESS_EPOCH: LazyLock<Uuid> = LazyLock::new(Uuid::new_v4);
 
 fn cross_pool_fence_fleet_proof_slot() -> &'static std::sync::RwLock<FleetCapabilityProofState> {
@@ -368,6 +380,24 @@ fn transition_transaction_compaction_fleet_proof_slot() -> &'static std::sync::R
     TRANSITION_TRANSACTION_COMPACTION_FLEET_PROOF.get_or_init(|| std::sync::RwLock::new(FleetCapabilityProofState::default()))
 }
 
+fn all_fleet_capability_proof_slots() -> [&'static std::sync::RwLock<FleetCapabilityProofState>; 7] {
+    [
+        remote_version_state_fleet_proof_slot(),
+        cross_pool_fence_fleet_proof_slot(),
+        tier_delete_journal_fleet_proof_slot(),
+        decommission_target_fence_fleet_proof_slot(),
+        legacy_transition_state_reconcile_fleet_proof_slot(),
+        ilm_recovery_export_fleet_proof_slot(),
+        transition_transaction_compaction_fleet_proof_slot(),
+    ]
+}
+
+fn revoke_all_fleet_capability_proofs() {
+    for slot in all_fleet_capability_proof_slots() {
+        revoke_fleet_capability_proof(slot);
+    }
+}
+
 fn revoke_fleet_capability_proof_state(state: &mut FleetCapabilityProofState) {
     if let Some(proof) = state.proof.take() {
         proof.generation.revoke();
@@ -1226,15 +1256,7 @@ fn insert_remote_version_state_peer(peer_epochs: &mut BTreeMap<String, Uuid>, pe
 pub fn start_remote_version_state_fleet_probe(topology_fingerprint: String) {
     if REMOTE_VERSION_STATE_PROBE_TOPOLOGY.set(topology_fingerprint.clone()).is_err() {
         if REMOTE_VERSION_STATE_PROBE_TOPOLOGY.get() != Some(&topology_fingerprint) {
-            for slot in [
-                remote_version_state_fleet_proof_slot(),
-                cross_pool_fence_fleet_proof_slot(),
-                tier_delete_journal_fleet_proof_slot(),
-                decommission_target_fence_fleet_proof_slot(),
-                legacy_transition_state_reconcile_fleet_proof_slot(),
-                ilm_recovery_export_fleet_proof_slot(),
-                transition_transaction_compaction_fleet_proof_slot(),
-            ] {
+            for slot in all_fleet_capability_proof_slots() {
                 mark_fleet_capability_topology_conflict(slot);
             }
         }
@@ -1242,53 +1264,59 @@ pub fn start_remote_version_state_fleet_probe(topology_fingerprint: String) {
     }
 
     tokio::spawn(async move {
+        let mut notification_sys_unavailable_logged = false;
         loop {
-            let notification_sys = get_global_notification_sys();
-            let remote_version_state_probe = async {
-                match notification_sys.as_ref() {
-                    Some(notification_sys) => timeout(
-                        REMOTE_VERSION_STATE_PROBE_TIMEOUT,
-                        notification_sys.probe_remote_version_state_fleet(&topology_fingerprint),
-                    )
-                    .await
-                    .unwrap_or_else(|_| Err(Error::other("remote version state fleet capability probe timed out"))),
-                    None => Err(Error::other("remote version state fleet capability notification system is unavailable")),
+            let Some(notification_sys) = get_global_notification_sys() else {
+                // Startup publishes the notification system after this probe
+                // is started. Stay failed closed, but retry on the bootstrap
+                // cadence so a single node gains its capabilities as soon as
+                // the system appears instead of one full probe interval later.
+                revoke_all_fleet_capability_proofs();
+                if !notification_sys_unavailable_logged {
+                    notification_sys_unavailable_logged = true;
+                    debug!(
+                        event = EVENT_NOTIFICATION_CAPABILITY_PROBE,
+                        component = LOG_COMPONENT_ECSTORE,
+                        subsystem = LOG_SUBSYSTEM_NOTIFICATION,
+                        state = "waiting_for_notification_system",
+                        "notification capability probe"
+                    );
                 }
+                let _ = timeout(REMOTE_VERSION_STATE_PROBE_BOOTSTRAP_INTERVAL, NOTIFICATION_SYS_PUBLISHED.notified()).await;
+                continue;
+            };
+            notification_sys_unavailable_logged = false;
+            let remote_version_state_probe = async {
+                timeout(
+                    REMOTE_VERSION_STATE_PROBE_TIMEOUT,
+                    notification_sys.probe_remote_version_state_fleet(&topology_fingerprint),
+                )
+                .await
+                .unwrap_or_else(|_| Err(Error::other("remote version state fleet capability probe timed out")))
             };
             let cross_pool_fence_probe = async {
-                match notification_sys.as_ref() {
-                    Some(notification_sys) => timeout(
-                        REMOTE_VERSION_STATE_PROBE_TIMEOUT,
-                        notification_sys.probe_cross_pool_fence_fleet(&topology_fingerprint),
-                    )
-                    .await
-                    .unwrap_or_else(|_| Err(Error::other("cross-pool fence fleet capability probe timed out"))),
-                    None => Err(Error::other("cross-pool fence fleet capability notification system is unavailable")),
-                }
+                timeout(
+                    REMOTE_VERSION_STATE_PROBE_TIMEOUT,
+                    notification_sys.probe_cross_pool_fence_fleet(&topology_fingerprint),
+                )
+                .await
+                .unwrap_or_else(|_| Err(Error::other("cross-pool fence fleet capability probe timed out")))
             };
             let recovery_export_probe = async {
-                match notification_sys.as_ref() {
-                    Some(notification_sys) => timeout(
-                        REMOTE_VERSION_STATE_PROBE_TIMEOUT,
-                        notification_sys.probe_ilm_recovery_export_fleet(&topology_fingerprint),
-                    )
-                    .await
-                    .unwrap_or_else(|_| Err(Error::other("ILM recovery export fleet capability probe timed out"))),
-                    None => Err(Error::other("ILM recovery export fleet capability notification system is unavailable")),
-                }
+                timeout(
+                    REMOTE_VERSION_STATE_PROBE_TIMEOUT,
+                    notification_sys.probe_ilm_recovery_export_fleet(&topology_fingerprint),
+                )
+                .await
+                .unwrap_or_else(|_| Err(Error::other("ILM recovery export fleet capability probe timed out")))
             };
             let transition_transaction_compaction_probe = async {
-                match notification_sys.as_ref() {
-                    Some(notification_sys) => timeout(
-                        REMOTE_VERSION_STATE_PROBE_TIMEOUT,
-                        notification_sys.probe_transition_transaction_compaction_fleet(&topology_fingerprint),
-                    )
-                    .await
-                    .unwrap_or_else(|_| Err(Error::other("transition transaction compaction fleet capability probe timed out"))),
-                    None => Err(Error::other(
-                        "transition transaction compaction fleet capability notification system is unavailable",
-                    )),
-                }
+                timeout(
+                    REMOTE_VERSION_STATE_PROBE_TIMEOUT,
+                    notification_sys.probe_transition_transaction_compaction_fleet(&topology_fingerprint),
+                )
+                .await
+                .unwrap_or_else(|_| Err(Error::other("transition transaction compaction fleet capability probe timed out")))
             };
             let (result, fence_probe, recovery_export_result, transition_transaction_compaction_result) = tokio::join!(
                 remote_version_state_probe,
@@ -1313,13 +1341,7 @@ pub fn start_remote_version_state_fleet_probe(topology_fingerprint: String) {
                 .unwrap_or_else(std::sync::PoisonError::into_inner)
                 .topology_conflict;
             if topology_conflict {
-                revoke_fleet_capability_proof(remote_version_state_fleet_proof_slot());
-                revoke_fleet_capability_proof(cross_pool_fence_fleet_proof_slot());
-                revoke_fleet_capability_proof(tier_delete_journal_fleet_proof_slot());
-                revoke_fleet_capability_proof(decommission_target_fence_fleet_proof_slot());
-                revoke_fleet_capability_proof(legacy_transition_state_reconcile_fleet_proof_slot());
-                revoke_fleet_capability_proof(ilm_recovery_export_fleet_proof_slot());
-                revoke_fleet_capability_proof(transition_transaction_compaction_fleet_proof_slot());
+                revoke_all_fleet_capability_proofs();
             } else if let Some(err) = publish_fleet_capability_probe_result(
                 remote_version_state_fleet_proof_slot(),
                 &topology_fingerprint,
@@ -1442,9 +1464,11 @@ pub fn start_remote_version_state_fleet_probe(topology_fingerprint: String) {
 }
 
 pub async fn new_global_notification_sys(eps: EndpointServerPools) -> Result<()> {
-    let _ = GLOBAL_NOTIFICATION_SYS
-        .set(Arc::new(NotificationSys::new(eps).await))
-        .map_err(|_| Error::other("init notification_sys fail"));
+    if GLOBAL_NOTIFICATION_SYS.set(Arc::new(NotificationSys::new(eps).await)).is_ok() {
+        // `notify_one` stores a permit, so a probe that checks availability
+        // just before this publication still wakes without losing the signal.
+        NOTIFICATION_SYS_PUBLISHED.notify_one();
+    }
     Ok(())
 }
 
@@ -3919,6 +3943,49 @@ mod tests {
         stats
     }
 
+    /// rustfs/rustfs#8014: IAM finalization starts the fleet probe before the
+    /// notification system is published. The first probe therefore fails
+    /// closed, and a single node used to wait a full probe interval before the
+    /// durable quota capability appeared, even though `/health` was already
+    /// reporting ok. The probe must retry promptly while it waits for startup
+    /// to publish the notification system.
+    #[tokio::test]
+    async fn fleet_probe_publishes_promptly_once_the_notification_system_appears() {
+        if REMOTE_VERSION_STATE_PROBE_TOPOLOGY.get().is_some() || GLOBAL_NOTIFICATION_SYS.get().is_some() {
+            // Process-wide state was bound by another test in this binary. The
+            // startup ordering under test needs a fresh process; nextest
+            // (the authoritative runner) always provides one.
+            eprintln!("skipping: fleet probe globals already bound in this process");
+            return;
+        }
+
+        // Share the fingerprint every other proof-installing test in this crate
+        // binds, so the plain `cargo test` fallback cannot see two topologies.
+        start_remote_version_state_fleet_probe("object-transaction-fencing-test".to_string());
+        sleep(Duration::from_millis(300)).await;
+        assert!(
+            acquire_cross_pool_fence_fleet_proof().is_none(),
+            "no capability proof may exist before the notification system is published"
+        );
+
+        let published_at = Instant::now();
+        new_global_notification_sys(EndpointServerPools::default())
+            .await
+            .expect("single-node notification system initializes");
+
+        let observed = timeout(Duration::from_secs(2), async {
+            while acquire_cross_pool_fence_fleet_proof().is_none() {
+                sleep(Duration::from_millis(20)).await;
+            }
+        })
+        .await;
+        assert!(
+            observed.is_ok(),
+            "single-node cross-pool fence proof should publish within 2s of the notification system, waited {:?}",
+            published_at.elapsed()
+        );
+    }
+
     #[test]
     fn merging_peer_rings_sums_each_tier_without_double_counting() {
         let mut cluster = DailyAllTierStats::from([("WARM".to_string(), ring(10))]);

crates/ecstore/src/set_disk/mod.rs

@@ -3915,6 +3915,72 @@ pub struct SetDisks {
     >,
 }
 
+fn marker_purge_receipt_path(
+    bucket: &str,
+    object: &str,
+    version: Uuid,
+    purge: &rustfs_common::mrf_channel::MrfDeleteMarkerPurge,
+) -> String {
+    let object_hash = rustfs_utils::crypto::hex(Sha256::digest(object.as_bytes()));
+    let marker_hash = rustfs_utils::crypto::hex(purge.marker_identity);
+    format!(
+        "marker-purge-receipts/{}/{}/{}-{}-{}.json",
+        bucket, purge.bucket_incarnation_id, version, object_hash, marker_hash
+    )
+}
+
+impl SetDisks {
+    /// Persist a quorum DELETE receipt outside the object marker itself. The
+    /// receipt is written to the metadata bucket so it survives a node restart
+    /// while an inaccessible member still holds the stale marker.
+    pub(crate) async fn persist_marker_purge_receipt(
+        &self,
+        bucket: &str,
+        object: &str,
+        version: Uuid,
+        purge: &rustfs_common::mrf_channel::MrfDeleteMarkerPurge,
+    ) -> bool {
+        let path = marker_purge_receipt_path(bucket, object, version, purge);
+        let api = Arc::new(self.clone());
+        crate::config::com::save_config_with_opts(
+            api,
+            &path,
+            b"rustfs-marker-purge-receipt-v1".to_vec(),
+            &ObjectOptions {
+                max_parity: true,
+                no_lock: true,
+                ..Default::default()
+            },
+        )
+        .await
+        .is_ok()
+    }
+
+    pub(crate) async fn has_marker_purge_receipt(
+        &self,
+        bucket: &str,
+        object: &str,
+        version: Uuid,
+        purge: &rustfs_common::mrf_channel::MrfDeleteMarkerPurge,
+    ) -> bool {
+        let path = marker_purge_receipt_path(bucket, object, version, purge);
+        crate::config::com::read_config_limited_preserve_empty(Arc::new(self.clone()), &path, 128)
+            .await
+            .is_ok()
+    }
+
+    pub(crate) async fn consume_marker_purge_receipt(
+        &self,
+        bucket: &str,
+        object: &str,
+        version: Uuid,
+        purge: &rustfs_common::mrf_channel::MrfDeleteMarkerPurge,
+    ) {
+        let path = marker_purge_receipt_path(bucket, object, version, purge);
+        let _ = crate::config::com::delete_config_no_lock(Arc::new(self.clone()), &path).await;
+    }
+}
+
 /// Read every physical copy before selecting a version quorum. A minority
 /// legacy record is still evidence and must not disappear behind a majority
 /// not-found result. Only an explicit file/volume absence produces `None`;

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

@@ -38,6 +38,7 @@ use crate::disk::DiskAPI;
 use crate::disk::local::{DELETE_DATA_DIR_MARKER_PREFIX, metadata_less_part_file};
 use crate::io_support::bitrot::object_mmap_read_enabled;
 use crate::storage_api_contracts::namespace::NamespaceLocking as _;
+use rustfs_common::mrf_channel::MrfDeleteMarkerPurge;
 use rustfs_common::trace_bus::{TraceEvent, TraceFunc, TraceKind, trace_emit};
 use tracing::trace;
 
@@ -1469,7 +1470,12 @@ impl SetDisks {
                                     })
                                     .transpose()
                                     .map_err(DiskError::from)?;
-                                let till_offset = erasure.shard_file_offset(0, part.size, part.size);
+                                // This reader covers the whole encoded shard. A range
+                                // offset can exclude the final encoded byte when the
+                                // logical part ends inside a data block, allowing a
+                                // one-byte tail truncation to evade deep Heal.
+                                let till_offset = usize::try_from(erasure.shard_file_size(part.size as i64))
+                                    .map_err(|_| DiskError::FileCorrupt)?;
                                 let use_mmap_read = object_mmap_read_enabled();
 
                                 let mut readers = Vec::with_capacity(latest_disks.len());
@@ -1907,13 +1913,26 @@ impl SetDisks {
                         .zip(&errs)
                         .any(|(info, error)| error.is_none() && info.is_canonical_delete_marker())
                 {
-                    let cleanup = match Self::retired_marker_candidate(&parts_metadata, &errs) {
-                        Ok(marker) => {
-                            self.remove_retired_marker(retirement, bucket, object, marker, opts, &disks)
-                                .await
-                        }
-                        Err(error) => Err(error),
-                    };
+                    // A current-generation delete marker can be left behind on
+                    // an inaccessible member after an acknowledged S3 DELETE.
+                    // That case has no bucket-retirement record, so evaluate its
+                    // stricter same-incarnation proof before the retired-marker
+                    // path. The proof requires an exact marker identity, at
+                    // least one replica already absent, and every target online.
+                    let cleanup =
+                        match Self::same_incarnation_marker_candidate(&parts_metadata, &errs, retirement.current_incarnation) {
+                            Ok(marker) => {
+                                self.remove_current_marker(retirement, bucket, object, marker, opts, &disks)
+                                    .await
+                            }
+                            Err(_) => match Self::retired_marker_candidate(&parts_metadata, &errs) {
+                                Ok(marker) => {
+                                    self.remove_retired_marker(retirement, bucket, object, marker, opts, &disks)
+                                        .await
+                                }
+                                Err(error) => Err(error),
+                            },
+                        };
                     let absent = cleanup.is_ok();
                     let item = self
                         .dangling_heal_result(FileInfo::default(), &errs, bucket, object, version_id, absent)
@@ -2027,6 +2046,41 @@ impl SetDisks {
         candidate.ok_or_else(|| DiskError::retired_marker_deferred("no exact marker candidate"))
     }
 
+    fn same_incarnation_marker_candidate(
+        metadata: &[FileInfo],
+        errors: &[Option<DiskError>],
+        current_incarnation: Option<Uuid>,
+    ) -> disk::error::Result<rustfs_filemeta::MetaDeleteMarker> {
+        let defer = DiskError::retired_marker_deferred;
+        let Some(current) = current_incarnation.filter(|id| !id.is_nil()) else {
+            return Err(defer("current bucket incarnation is unavailable"));
+        };
+        let mut candidate = None;
+        let mut missing_replica = false;
+        for (info, error) in metadata.iter().zip(errors) {
+            match error {
+                Some(DiskError::FileNotFound | DiskError::FileVersionNotFound) => {
+                    missing_replica = true;
+                    continue;
+                }
+                Some(_) => return Err(defer("not every replica is readable")),
+                None => {}
+            }
+            if !info.is_canonical_delete_marker() || info.delete_marker_incarnation() != Some(current) {
+                return Err(defer("marker is not a canonical current-generation marker"));
+            }
+            let marker = rustfs_filemeta::MetaDeleteMarker::from(info.clone());
+            if candidate.as_ref().is_some_and(|previous| previous != &marker) {
+                return Err(defer("surviving marker identities conflict"));
+            }
+            candidate = Some(marker);
+        }
+        if !missing_replica {
+            return Err(defer("marker is present on every replica"));
+        }
+        candidate.ok_or_else(|| defer("no exact current-generation marker candidate"))
+    }
+
     async fn remove_retired_marker(
         &self,
         context: &MarkerRetirementContext<'_>,
@@ -2065,6 +2119,72 @@ impl SetDisks {
         if context.lifecycle_guard.is_lock_lost() {
             return Err(defer("bucket lifecycle fence was lost"));
         }
+        let result = self
+            .remove_marker_exact_under_fence(bucket, object, marker, version, disks)
+            .await;
+        if context.lifecycle_guard.is_lock_lost() {
+            return Err(defer("bucket lifecycle fence was lost"));
+        }
+        result
+    }
+
+    async fn remove_current_marker(
+        &self,
+        context: &MarkerRetirementContext<'_>,
+        bucket: &str,
+        object: &str,
+        marker: rustfs_filemeta::MetaDeleteMarker,
+        opts: &HealOpts,
+        disks: &[Option<DiskStore>],
+    ) -> disk::error::Result<()> {
+        let defer = DiskError::retired_marker_deferred;
+        if opts.dry_run || !opts.remove {
+            return Err(defer("cleanup requires remove=true and dry_run=false"));
+        }
+        if disks.len() != self.set_drive_count || disks.iter().any(Option::is_none) {
+            return Err(defer("every target disk must be online"));
+        }
+        let Some(current) = context.current_incarnation.filter(|id| !id.is_nil()) else {
+            return Err(defer("current bucket incarnation is unavailable"));
+        };
+        if context.lifecycle_guard.is_lock_lost() {
+            return Err(defer("bucket lifecycle fence was lost"));
+        }
+        let incarnation = marker.into_fileinfo(bucket, object, false)?.delete_marker_incarnation();
+        if incarnation != Some(current) {
+            return Err(defer("marker is not from the current bucket incarnation"));
+        }
+        let Some(version) = marker.version_id.filter(|id| !id.is_nil()) else {
+            return Err(defer("cleanup requires an explicit non-null version"));
+        };
+        let marker_identity = marker.stable_identity();
+        let marker_bytes = marker.marshal_msg().map_err(DiskError::other)?;
+        let Some(purge) = MrfDeleteMarkerPurge::new(current, current, marker_identity, marker_bytes) else {
+            return Err(defer("marker purge proof payload is invalid"));
+        };
+        if !self.has_marker_purge_receipt(bucket, object, version, &purge).await {
+            return Err(defer("same-incarnation marker lacks an acknowledged delete receipt"));
+        }
+        let result = self
+            .remove_marker_exact_under_fence(bucket, object, marker, version, disks)
+            .await;
+        if context.lifecycle_guard.is_lock_lost() {
+            return Err(defer("bucket lifecycle fence was lost"));
+        }
+        if result.is_ok() {
+            self.consume_marker_purge_receipt(bucket, object, version, &purge).await;
+        }
+        result
+    }
+
+    async fn remove_marker_exact_under_fence(
+        &self,
+        bucket: &str,
+        object: &str,
+        marker: rustfs_filemeta::MetaDeleteMarker,
+        version: Uuid,
+        disks: &[Option<DiskStore>],
+    ) -> disk::error::Result<()> {
         let request = FileInfo {
             volume: bucket.to_owned(),
             name: object.to_owned(),
@@ -2105,13 +2225,14 @@ impl SetDisks {
         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 context.lifecycle_guard.is_lock_lost()
-            || !same_targets
+        if !same_targets
             || !errors
                 .iter()
                 .all(|error| matches!(error, Some(DiskError::FileNotFound | DiskError::FileVersionNotFound)))
         {
-            return Err(defer("complete marker absence could not be verified under the original fence"));
+            return Err(DiskError::retired_marker_deferred(
+                "complete marker absence could not be verified under the original target set",
+            ));
         }
         self.invalidate_get_object_metadata_cache(bucket, object).await;
         Ok(())
@@ -3319,7 +3440,9 @@ mod heal_result_report_tests {
     use crate::disk::{DiskAPI as _, DiskOption, DiskStore, RUSTFS_META_TMP_BUCKET, ReadOptions, STORAGE_FORMAT_FILE, new_disk};
     use crate::error::Error;
     use crate::object_api::{ObjectOptions, PutObjReader};
-    use crate::set_disk::ops::object::hermetic_set_disks_support::hermetic_set_disks_isolated;
+    use crate::set_disk::ops::object::hermetic_set_disks_support::{
+        hermetic_set_disks_for_pool_with_default_parity_isolated, hermetic_set_disks_isolated,
+    };
     use crate::storage_api_contracts::bucket::{BucketOperations as _, DeleteBucketOptions, MakeBucketOptions};
     use crate::storage_api_contracts::heal::HealOperations as _;
     use crate::storage_api_contracts::object::{ObjectIO as _, ObjectOperations as _};
@@ -4100,6 +4223,78 @@ mod heal_result_report_tests {
         );
     }
 
+    #[tokio::test]
+    async fn deep_heal_rebuilds_a_stale_current_delete_marker_metadata_replica() {
+        let (_temp_dirs, disks, set) = hermetic_set_disks_isolated(4).await;
+        let bucket = "deep-heal-stale-current-marker";
+        let object = "object.bin";
+        for disk in &disks {
+            disk.make_volume(bucket).await.expect("test bucket should be created");
+        }
+
+        let versioned = ObjectOptions {
+            no_lock: true,
+            versioned: true,
+            ..Default::default()
+        };
+        let original = set
+            .put_object(bucket, object, &mut PutObjReader::from_vec(vec![0x41; 1024]), &versioned)
+            .await
+            .expect("versioned source object should be written");
+        let stale_metadata = disks[0]
+            .read_version("", bucket, object, "", &ReadOptions::default())
+            .await
+            .expect("pre-marker metadata should be readable");
+
+        let marker = set
+            .delete_object(bucket, object, versioned)
+            .await
+            .expect("versioned delete should create a current marker");
+        let marker_version = marker.version_id.expect("delete marker should have a version id");
+
+        // Simulate a member that was offline while the acknowledged marker was
+        // committed on the other members: its physical root still contains the
+        // pre-delete metadata and the original data version.
+        let mut stale_metadata = stale_metadata;
+        stale_metadata.fresh = true;
+        disks[0]
+            .write_metadata("", bucket, object, stale_metadata)
+            .await
+            .expect("stale member metadata should be restored for the fixture");
+        assert_eq!(
+            disks[0]
+                .read_version("", bucket, object, &marker_version.to_string(), &ReadOptions::default())
+                .await
+                .expect_err("stale member must not contain the current marker")
+                .to_string(),
+            "file version not found"
+        );
+
+        let (result, error) = set
+            .heal_object(
+                bucket,
+                object,
+                &marker_version.to_string(),
+                &HealOpts {
+                    no_lock: true,
+                    scan_mode: HealScanMode::Deep,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("deep heal should report the stale marker result");
+
+        assert!(error.is_none(), "deep heal should repair stale marker metadata: {error:?}");
+        assert_eq!(result.drives_healed(), Some(1));
+        let repaired = disks[0]
+            .read_version("", bucket, object, &marker_version.to_string(), &ReadOptions::default())
+            .await
+            .expect("rejoined member should contain the marker");
+        assert!(repaired.deleted, "repaired metadata must remain a delete marker");
+        assert_eq!(repaired.version_id, Some(marker_version));
+        assert!(original.version_id.is_some(), "source object should have a version id");
+    }
+
     #[tokio::test]
     async fn deep_heal_rebuilds_near_tail_truncated_part() {
         let (temp_dirs, disks, set) = hermetic_set_disks_isolated(4).await;
@@ -4183,6 +4378,89 @@ mod heal_result_report_tests {
         );
     }
 
+    #[tokio::test]
+    async fn deep_heal_rebuilds_h2_near_tail_truncated_part() {
+        let (temp_dirs, disks, set) = hermetic_set_disks_for_pool_with_default_parity_isolated(16, 0, 4).await;
+        let bucket = "deep-heal-h2-near-tail-truncation";
+        let object = "object.bin";
+        for disk in &disks {
+            disk.make_volume(bucket).await.expect("bucket volume should be created");
+        }
+
+        let expected_payload = vec![0x6d; 5 * 1024 * 1024 + 123];
+        set.put_object(
+            bucket,
+            object,
+            &mut PutObjReader::from_vec(expected_payload.clone()),
+            &ObjectOptions {
+                no_lock: true,
+                ..Default::default()
+            },
+        )
+        .await
+        .expect("source object should be written before shard truncation");
+        let source = disks[2]
+            .read_version("", bucket, object, "", &ReadOptions::default())
+            .await
+            .expect("source metadata should be readable");
+        let data_dir = source.data_dir.expect("non-inline source should have a data directory");
+        let truncated_part = temp_dirs[1]
+            .path()
+            .join(bucket)
+            .join(object)
+            .join(data_dir.to_string())
+            .join("part.1");
+        let original_len = tokio::fs::metadata(&truncated_part)
+            .await
+            .expect("target shard should exist")
+            .len();
+        assert!(original_len > 1, "test shard must be large enough to truncate");
+        let file = tokio::fs::OpenOptions::new()
+            .write(true)
+            .open(&truncated_part)
+            .await
+            .expect("target shard should be writable");
+        file.set_len(original_len - 1)
+            .await
+            .expect("target shard should be truncated");
+
+        let mut reader = set
+            .get_object_reader(bucket, object, None, Default::default(), &ObjectOptions::default())
+            .await
+            .expect("GET should remain readable after a one-byte shard truncation");
+        let mut read_back = Vec::new();
+        tokio::io::copy(&mut reader, &mut read_back)
+            .await
+            .expect("GET should reconstruct the truncated shard through EC");
+        assert_eq!(read_back, expected_payload, "EC GET must preserve the object bytes");
+
+        let (result, error) = set
+            .heal_object(
+                bucket,
+                object,
+                "",
+                &HealOpts {
+                    no_lock: true,
+                    scan_mode: HealScanMode::Deep,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("deep heal should finish after a one-byte H2 tail truncation");
+
+        assert!(error.is_none(), "deep heal should recover the H2 truncated shard: {error:?}");
+        assert_eq!(result.drives_healed(), Some(1));
+        assert_eq!(result.before.drives[1].state, DriveState::Corrupt.to_string());
+        assert_eq!(
+            tokio::fs::metadata(&truncated_part)
+                .await
+                .expect("repaired H2 shard should exist")
+                .len(),
+            original_len,
+            "deep heal must restore the complete H2 shard length"
+        );
+    }
+
     #[tokio::test]
     async fn deep_heal_inline_bitrot_rebuilds_physical_data_and_parity_shards() {
         use crate::storage_api_contracts::range::HTTPRangeSpec;

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

@@ -7703,10 +7703,11 @@ impl SetDisks {
         drop(namespace_owner);
         if quorum_result.is_ok()
             && errs.iter().any(Option::is_some)
-            && let Some(purge) = delete_marker_purge
+            && let Some(purge) = delete_marker_purge.as_ref()
             && let Some(version) = fi.version_id.filter(|version| !version.is_nil())
         {
-            let _ = self.persist_delete_marker_purge(bucket, object, version, purge).await;
+            let _ = self.persist_marker_purge_receipt(bucket, object, version, purge).await;
+            let _ = self.persist_delete_marker_purge(bucket, object, version, purge.clone()).await;
         }
         // An explicit purge can carry deleted=true for the existing marker.
         // It must not create a repair intent that could reintroduce that marker.

crates/ecstore/src/store/object.rs

@@ -9124,6 +9124,100 @@ mod tests {
         assert_eq!(dirs.len(), 16);
     }
 
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn current_marker_delete_ack_removes_stale_rejoined_copy() {
+        let bucket = "current-marker-stale-rejoin";
+        let object = "marker.bin";
+        let ctx = Arc::new(crate::runtime::instance::InstanceContext::new());
+        let (_dirs, original) = make_local_set_disks_with_ctx(16, 4, ctx.clone()).await;
+        let store = Arc::new(new_prepared_reader_test_store_with_ctx(&[original], ctx).await);
+        crate::bucket::metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await;
+        store
+            .handle_make_bucket(
+                bucket,
+                &MakeBucketOptions {
+                    versioning_enabled: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("create versioned bucket");
+        let marker = store
+            .delete_object(
+                bucket,
+                object,
+                ObjectOptions {
+                    versioned: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("create current marker")
+            .version_id
+            .expect("marker has explicit version");
+        let current = store
+            .bucket_incarnation_id_from_disk(bucket)
+            .await
+            .expect("bucket incarnation");
+        let set = &store.pools[0].disk_set[0];
+        let before = set.disks.read().await.clone();
+        for slot in 12..16 {
+            set.disks.write().await[slot] = None;
+        }
+        store
+            .delete_object(
+                bucket,
+                object,
+                ObjectOptions {
+                    versioned: true,
+                    version_id: Some(marker.to_string()),
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("acknowledged delete should remove the marker from the online quorum");
+        *set.disks.write().await = before.clone();
+
+        for disk in before.iter().take(12).flatten() {
+            assert!(matches!(
+                disk.read_version("", bucket, object, &marker.to_string(), &crate::disk::ReadOptions::default())
+                    .await,
+                Err(crate::disk::error::DiskError::FileVersionNotFound | crate::disk::error::DiskError::FileNotFound)
+            ));
+        }
+        for disk in before.iter().skip(12).flatten() {
+            assert!(
+                disk.read_version("", bucket, object, &marker.to_string(), &crate::disk::ReadOptions::default())
+                    .await
+                    .is_ok(),
+                "offline member must retain the stale current marker"
+            );
+        }
+        let healed = store
+            .heal_object_at_incarnation(
+                bucket,
+                object,
+                &marker.to_string(),
+                current,
+                &rustfs_heal_contracts::heal_channel::HealOpts {
+                    remove: true,
+                    scan_mode: rustfs_heal_contracts::heal_channel::HealScanMode::Deep,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("stale current marker heal should complete");
+        assert!(healed.error.is_none(), "stale current marker heal: {:?}", healed.error);
+        assert!(healed.absence.as_ref().is_some_and(|proof| proof.removed));
+        for disk in before.iter().flatten() {
+            assert!(matches!(
+                disk.read_version("", bucket, object, &marker.to_string(), &crate::disk::ReadOptions::default())
+                    .await,
+                Err(crate::disk::error::DiskError::FileVersionNotFound | crate::disk::error::DiskError::FileNotFound)
+            ));
+        }
+    }
+
     #[tokio::test]
     async fn multipool_delete_marker_stays_with_existing_versions() {
         let bucket = "multipool-marker-routing";

crates/scanner/src/remote_scanner/stream.rs

@@ -16,8 +16,8 @@
 use crate::RUSTFS_META_BUCKET;
 use crate::scanner_budget::{ScannerCycleBudget, ScannerCycleBudgetConfig};
 use crate::scanner_io::{
-    DataUsageCacheReuseOptions, DataUsageCacheScanState, ScannerCheckpointPersistResult, ScannerDiskScanOptions,
-    ScannerDiskScanOutcome, ScannerIODisk, acquire_scanner_cache_locks, cache_root_entry_info,
+    DataUsageCacheReuseOptions, DataUsageCacheScanState, ScannerCheckpointPersistContext, ScannerCheckpointPersistResult,
+    ScannerDiskScanOptions, ScannerDiskScanOutcome, ScannerIODisk, acquire_scanner_cache_locks, cache_root_entry_info,
     current_cache_root_or_prepare_with_generation, persist_scanner_checkpoint, scanner_set_disk_inventory,
 };
 use crate::storage_api::owner::NS_SCANNER_PROTOCOL_VERSION;
@@ -840,12 +840,15 @@ async fn scan_and_persist_local_bucket(
                 }
                 match persist_scanner_checkpoint(
                     set.clone(),
+                    ScannerCheckpointPersistContext {
+                        ctx: &scan_ctx,
+                        expected_publication_epoch,
+                        cycle: next_cycle,
+                        leader_epoch,
+                    },
                     &cache_name,
                     &checkpoint,
                     &mut revisions,
-                    expected_publication_epoch,
-                    next_cycle,
-                    leader_epoch,
                 )
                 .await
                 {

crates/scanner/src/scanner.rs

@@ -1778,9 +1778,7 @@ where
             mark_scan_cycle_idle(cycle_info, &mut cycle_metrics_guard).await;
             return ScannerCycleOutcome::Failed;
         }
-        BackgroundHealInfoReadStatus::ErasureSd
-        | BackgroundHealInfoReadStatus::Loaded
-        | BackgroundHealInfoReadStatus::Missing => {}
+        BackgroundHealInfoReadStatus::Loaded | BackgroundHealInfoReadStatus::Missing => {}
     }
     let mut background_heal_info = background_heal_read.info;
     let background_heal_epoch = background_heal_read.expected_epoch;

crates/scanner/src/scanner/heal_info.rs

@@ -28,7 +28,6 @@ pub struct BackgroundHealInfo {
 
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
 pub(super) enum BackgroundHealInfoReadStatus {
-    ErasureSd,
     Loaded,
     Missing,
     Blocked,
@@ -66,15 +65,6 @@ pub(super) async fn read_background_heal_info_with_epoch<S>(storeapi: Arc<S>) ->
 where
     S: ScannerStorage,
 {
-    // Skip for ErasureSD setup
-    if storeapi.setup_is_erasure_sd().await {
-        return BackgroundHealInfoRead {
-            info: BackgroundHealInfo::default(),
-            expected_epoch: None,
-            status: BackgroundHealInfoReadStatus::ErasureSd,
-        };
-    }
-
     let expected_epoch = scanner_publication_epoch(storeapi.clone()).await;
     if expected_epoch.is_none() {
         return BackgroundHealInfoRead {
@@ -146,11 +136,6 @@ pub(super) async fn save_background_heal_info_for_epoch<S>(
 ) where
     S: ScannerStorage,
 {
-    // Skip for ErasureSD setup
-    if storeapi.setup_is_erasure_sd().await {
-        return;
-    }
-
     // Serialize to JSON
     let data = match serde_json::to_vec(&info) {
         Ok(data) => data,

crates/scanner/src/scanner_folder.rs

@@ -1806,6 +1806,14 @@ impl FolderScanner {
                     continue;
                 }
 
+                // Do not start another metadata read while foreground work is
+                // active. The post-object timer protects the next request only
+                // after the read has already been dispatched; this admission
+                // point keeps the scanner from extending a single-disk I/O
+                // burst across foreground requests.
+                if crate::workload_admission::foreground_workload_activity() > 0 {
+                    self.sleeper.sleep_folder().await;
+                }
                 let timer = self.sleeper.timer();
 
                 let heal_enabled = this_hash.mod_alt(

crates/scanner/src/scanner_io.rs

@@ -1609,8 +1609,8 @@ use dirty_usage::*;
 use guards::*;
 
 pub(crate) use cache::{
-    DataUsageCacheReuseOptions, DataUsageCacheScanState, ScannerCheckpointPersistResult, acquire_scanner_cache_locks,
-    current_cache_root_or_prepare_with_generation, persist_scanner_checkpoint,
+    DataUsageCacheReuseOptions, DataUsageCacheScanState, ScannerCheckpointPersistContext, ScannerCheckpointPersistResult,
+    acquire_scanner_cache_locks, current_cache_root_or_prepare_with_generation, persist_scanner_checkpoint,
 };
 pub use dirty_usage::{
     ScannerDirtyUsageAckError, ScannerDirtyUsageBucket, ScannerDirtyUsageClearObserver, ScannerDirtyUsageMutationObserver,

crates/scanner/src/scanner_io/cache.rs

@@ -31,6 +31,41 @@ pub(crate) enum ScannerCheckpointPersistResult {
     Failed(StorageError),
 }
 
+pub(crate) struct ScannerCheckpointPersistContext<'a> {
+    pub(crate) ctx: &'a CancellationToken,
+    pub(crate) expected_publication_epoch: u64,
+    pub(crate) cycle: u64,
+    pub(crate) leader_epoch: u64,
+}
+
+const CHECKPOINT_FOREGROUND_QUIET_WAIT: Duration = Duration::from_secs(1);
+
+async fn wait_for_checkpoint_foreground_quiet(ctx: &CancellationToken) -> bool {
+    let deadline = tokio::time::Instant::now() + CHECKPOINT_FOREGROUND_QUIET_WAIT;
+    loop {
+        if crate::workload_admission::foreground_workload_activity() == 0 {
+            return true;
+        }
+
+        let remaining = deadline.saturating_duration_since(tokio::time::Instant::now());
+        if remaining.is_zero() {
+            return false;
+        }
+
+        let backoff = Duration::from_millis(
+            crate::workload_admission::foreground_workload_activity()
+                .saturating_mul(10)
+                .min(250),
+        )
+        .max(Duration::from_millis(10))
+        .min(remaining);
+        tokio::select! {
+            _ = ctx.cancelled() => return false,
+            _ = tokio::time::sleep(backoff) => {}
+        }
+    }
+}
+
 /// Persist one bounded checkpoint and refresh its CAS revisions.
 ///
 /// Local and remote workers share the same publication/leader fencing and
@@ -38,23 +73,37 @@ pub(crate) enum ScannerCheckpointPersistResult {
 /// at the caller because those guards have different concrete types.
 pub(crate) async fn persist_scanner_checkpoint<S>(
     store: Arc<S>,
+    context: ScannerCheckpointPersistContext<'_>,
     cache_name: &str,
     checkpoint: &DataUsageCache,
     revisions: &mut DataUsageCacheRevisions,
-    expected_publication_epoch: u64,
-    cycle: u64,
-    leader_epoch: u64,
 ) -> ScannerCheckpointPersistResult
 where
     S: ScannerObjectIO + ScannerConfigObjectDelete,
 {
-    if crate::remote_scanner::validate_remote_scanner_request_fence_with_store(cycle, leader_epoch, store.clone())
+    let foreground_quiet = wait_for_checkpoint_foreground_quiet(context.ctx).await;
+    if !foreground_quiet && context.ctx.is_cancelled() {
+        return ScannerCheckpointPersistResult::FenceChanged;
+    }
+    if !foreground_quiet {
+        debug!(
+            target: "rustfs::scanner::io",
+            event = EVENT_SCANNER_CACHE_PERSIST_STATE,
+            component = LOG_COMPONENT_SCANNER,
+            subsystem = LOG_SUBSYSTEM_IO,
+            cache_name,
+            state = "checkpoint_foreground_wait_expired",
+            "Scanner checkpoint foreground quiet wait expired; preserving bounded progress"
+        );
+    }
+
+    if crate::remote_scanner::validate_remote_scanner_request_fence_with_store(context.cycle, context.leader_epoch, store.clone())
         .await
         .is_err()
     {
         return ScannerCheckpointPersistResult::FenceChanged;
     }
-    if scanner_publication_admission_for_epoch(store.clone(), expected_publication_epoch)
+    if scanner_publication_admission_for_epoch(store.clone(), context.expected_publication_epoch)
         .await
         .is_none()
     {
@@ -62,16 +111,16 @@ where
     }
 
     if let Err(error) = checkpoint
-        .save_with_revisions_for_epoch(store.clone(), cache_name, revisions, expected_publication_epoch)
+        .save_with_revisions_for_epoch(store.clone(), cache_name, revisions, context.expected_publication_epoch)
         .await
     {
         return ScannerCheckpointPersistResult::Failed(error);
     }
 
-    if crate::remote_scanner::validate_remote_scanner_request_fence_with_store(cycle, leader_epoch, store.clone())
+    if crate::remote_scanner::validate_remote_scanner_request_fence_with_store(context.cycle, context.leader_epoch, store.clone())
         .await
         .is_err()
-        || scanner_publication_admission_for_epoch(store.clone(), expected_publication_epoch)
+        || scanner_publication_admission_for_epoch(store.clone(), context.expected_publication_epoch)
             .await
             .is_none()
     {

crates/scanner/src/scanner_io/io_cache.rs

@@ -1191,12 +1191,15 @@ impl ScannerIOCache for SetDisks {
                                 }
                                 match persist_scanner_checkpoint(
                                     store_clone_clone.clone(),
+                                    ScannerCheckpointPersistContext {
+                                        ctx: &ctx_clone,
+                                        expected_publication_epoch: expected_publication_epoch_clone,
+                                        cycle: want_cycle,
+                                        leader_epoch,
+                                    },
                                     cache_name.as_str(),
                                     &checkpoint,
                                     &mut revisions,
-                                    expected_publication_epoch_clone,
-                                    want_cycle,
-                                    leader_epoch,
                                 )
                                 .await
                                 {

rustfs/src/admin/handlers/module_switch.rs

@@ -26,7 +26,7 @@ use crate::server::{
     apply_audit_module_switch_for_context, current_module_switch_snapshot, mark_event_notifier_reconciled,
     mark_event_notifier_unreconciled, refresh_audit_module_enabled, refresh_notify_module_enabled,
     refresh_persisted_module_switches_from, refresh_persisted_module_switches_from_store, save_persisted_module_switches_to,
-    validate_module_switch_update,
+    validate_module_switch_update, with_notify_runtime_reconcile_lock,
 };
 use http::{HeaderMap, StatusCode};
 use hyper::Method;
@@ -138,48 +138,53 @@ async fn refresh_module_switch_snapshot() -> S3Result<ModuleSwitchSnapshot> {
 async fn apply_module_switch_update(context: Arc<AppContext>, switches: PersistedModuleSwitches) -> S3Result<()> {
     preflight_dynamic_config_reload_for_context(Some(context.as_ref()), MODULE_SWITCHES_SIGNAL_SUBSYSTEM).await?;
     let store = context.object_store();
-    mark_event_notifier_unreconciled();
-    let notification_system = rustfs_notify::ensure_live_events();
-    if switches.notify_enabled {
-        notification_system
-            .reload_persisted_config_from_store(store.clone())
-            .await
-            .map_err(|err| {
-                tracing::warn!(error = %err, "Failed to load notification config for module switch update");
-                s3_error!(InternalError, "failed to load notification config")
-            })?;
-    }
+    let mut failures = with_notify_runtime_reconcile_lock(async {
+        mark_event_notifier_unreconciled();
+        let notification_system = rustfs_notify::ensure_live_events();
+        if switches.notify_enabled {
+            notification_system
+                .reload_persisted_config_from_store(store.clone())
+                .await
+                .map_err(|err| {
+                    tracing::warn!(error = %err, "Failed to load notification config for module switch update");
+                    s3_error!(InternalError, "failed to load notification config")
+                })?;
+        }
 
-    let transition_system = notification_system.clone();
-    let notify_transition = save_persisted_module_switches_to(store.clone(), switches, move || {
-        let enabled = refresh_notify_module_enabled();
-        transition_system.publish_targets_enabled(enabled, None)
+        let transition_system = notification_system.clone();
+        let notify_transition = save_persisted_module_switches_to(store.clone(), switches, move || {
+            let enabled = refresh_notify_module_enabled();
+            transition_system.publish_targets_enabled(enabled, None)
+        })
+        .await
+        .map_err(|err| {
+            tracing::warn!(error = %err, "Failed to save module switches");
+            s3_error!(InternalError, "failed to save module switches")
+        })?;
+
+        let mut failures = Vec::new();
+        let mut notify_converged = true;
+        if let Err(err) = notify_transition.wait().await {
+            tracing::warn!(error = %err, "Local notification runtime failed to apply module switch update");
+            notify_converged = false;
+            failures.push("local notify");
+        }
+        if !switches.notify_enabled
+            && let Err(err) = notification_system.reload_persisted_config_from_store(store).await
+        {
+            tracing::warn!(error = %err, "Local notification config cache failed to reload after module disable");
+            notify_converged = false;
+            failures.push("local notify config cache");
+        }
+        if notify_converged && notification_system.runtime_lifecycle_is_converged() {
+            mark_event_notifier_reconciled();
+        } else if notify_converged {
+            failures.push("local notify convergence");
+        }
+
+        Ok::<_, s3s::S3Error>(failures)
     })
-    .await
-    .map_err(|err| {
-        tracing::warn!(error = %err, "Failed to save module switches");
-        s3_error!(InternalError, "failed to save module switches")
-    })?;
-
-    let mut failures = Vec::new();
-    let mut notify_converged = true;
-    if let Err(err) = notify_transition.wait().await {
-        tracing::warn!(error = %err, "Local notification runtime failed to apply module switch update");
-        notify_converged = false;
-        failures.push("local notify");
-    }
-    if !switches.notify_enabled
-        && let Err(err) = notification_system.reload_persisted_config_from_store(store).await
-    {
-        tracing::warn!(error = %err, "Local notification config cache failed to reload after module disable");
-        notify_converged = false;
-        failures.push("local notify config cache");
-    }
-    if notify_converged && notification_system.runtime_lifecycle_is_converged() {
-        mark_event_notifier_reconciled();
-    } else if notify_converged {
-        failures.push("local notify convergence");
-    }
+    .await?;
 
     if apply_audit_module_switch_for_context(Some(context.as_ref())).await.is_err() {
         tracing::warn!(reason = "apply_failed", "Local audit runtime failed to apply module switch update");

rustfs/src/server/event.rs

@@ -29,6 +29,7 @@ use std::sync::OnceLock;
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::time::Duration;
 use tokio::spawn;
+use tokio::sync::Mutex as AsyncMutex;
 use tokio::task::JoinHandle;
 use tokio::time::{Instant, MissedTickBehavior};
 use tokio_util::sync::CancellationToken;
@@ -38,6 +39,12 @@ static NOTIFY_RUNTIME_RECONCILED: AtomicBool = AtomicBool::new(false);
 static NOTIFY_BUCKET_RULES_RECONCILED: AtomicBool = AtomicBool::new(false);
 static ECSTORE_EVENT_DISPATCH_HOOK: OnceLock<()> = OnceLock::new();
 
+// Serialize local notification lifecycle publication with the periodic and
+// peer-triggered reconciler. Lock order is this guard before the module-switch
+// RMW/object locks and the server-config read lock; the module-switch handler's
+// preliminary server-config read is released before it takes the RMW lock.
+static NOTIFY_RUNTIME_RECONCILE_LOCK: AsyncMutex<()> = AsyncMutex::const_new(());
+
 const EVENT_NOTIFIER_RECONCILE_INTERVAL: Duration = Duration::from_secs(5);
 const EVENT_NOTIFIER_RECONCILE_ATTEMPT_TIMEOUT: Duration = Duration::from_secs(120);
 const EVENT_NOTIFY_RUNTIME_RECONCILE: &str = "notify_runtime_reconcile";
@@ -55,6 +62,14 @@ pub(crate) fn mark_event_notifier_unreconciled() {
     NOTIFY_BUCKET_RULES_RECONCILED.store(false, Ordering::Release);
 }
 
+pub(crate) async fn with_notify_runtime_reconcile_lock<Fut, T>(operation: Fut) -> T
+where
+    Fut: Future<Output = T>,
+{
+    let _guard = NOTIFY_RUNTIME_RECONCILE_LOCK.lock().await;
+    operation.await
+}
+
 fn are_bucket_notification_rules_reconciled() -> bool {
     NOTIFY_BUCKET_RULES_RECONCILED.load(Ordering::Acquire)
 }
@@ -161,6 +176,12 @@ fn ensure_event_notifier_converged(system: &NotificationSystem) -> Result<(), No
 
 pub(crate) async fn reconcile_event_notifier_from_store(
     store: std::sync::Arc<rustfs_notify::NotifyStore>,
+) -> Result<(), NotificationError> {
+    with_notify_runtime_reconcile_lock(reconcile_event_notifier_from_store_unlocked(store)).await
+}
+
+async fn reconcile_event_notifier_from_store_unlocked(
+    store: std::sync::Arc<rustfs_notify::NotifyStore>,
 ) -> Result<(), NotificationError> {
     let result = async {
         validate_notify_module_env().map_err(NotificationError::Initialization)?;
@@ -310,7 +331,7 @@ pub async fn shutdown_event_notifier() -> Result<(), NotificationError> {
 
 #[instrument]
 pub async fn init_event_notifier() -> Result<(), NotificationError> {
-    init_event_notifier_with_store(runtime_sources::current_object_store_handle).await
+    with_notify_runtime_reconcile_lock(init_event_notifier_with_store(runtime_sources::current_object_store_handle)).await
 }
 
 async fn init_event_notifier_with_store<CurrentStore>(current_store: CurrentStore) -> Result<(), NotificationError>
@@ -408,6 +429,7 @@ mod tests {
     use super::{
         convert_ecstore_object_info, init_event_notifier_with_store, mark_bucket_notification_rules_reconciled,
         parse_host_and_port, run_persisted_event_notifier_reconciler, should_reconcile_bucket_notification_rules,
+        with_notify_runtime_reconcile_lock,
     };
     use crate::server::is_event_notifier_reconciled;
     use crate::storage_api::server::event::StorageObjectInfo;
@@ -471,6 +493,43 @@ mod tests {
         .await;
     }
 
+    #[tokio::test]
+    #[serial]
+    async fn notify_runtime_reconcile_lock_serializes_overlapping_operations() {
+        let entered = Arc::new(Notify::new());
+        let release = Arc::new(Notify::new());
+        let second_attempted = Arc::new(Notify::new());
+        let second_started = Arc::new(AtomicBool::new(false));
+
+        let first_entered = entered.clone();
+        let first_release = release.clone();
+        let first = tokio::spawn(with_notify_runtime_reconcile_lock(async move {
+            first_entered.notify_one();
+            first_release.notified().await;
+        }));
+        entered.notified().await;
+
+        let second_attempted_signal = second_attempted.clone();
+        let second_started_flag = second_started.clone();
+        let second = tokio::spawn(async move {
+            second_attempted_signal.notify_one();
+            with_notify_runtime_reconcile_lock(async move {
+                second_started_flag.store(true, Ordering::Release);
+            })
+            .await;
+        });
+        second_attempted.notified().await;
+        assert!(
+            !second_started.load(Ordering::Acquire),
+            "overlapping reconcile work must wait for the first operation"
+        );
+
+        release.notify_one();
+        first.await.expect("first reconcile operation should finish");
+        second.await.expect("second reconcile operation should finish");
+        assert!(second_started.load(Ordering::Acquire));
+    }
+
     #[test]
     fn parse_host_and_port_with_ipv4_and_port() {
         let (host, port) = parse_host_and_port("127.0.0.1:9000".to_string());

rustfs/src/server/mod.rs

@@ -48,7 +48,7 @@ pub use service_state::wait_for_shutdown;
 // Items only used within the library crate (admin handlers, server/http.rs, etc.).
 pub(crate) use event::{
     is_event_notifier_reconciled, mark_event_notifier_reconciled, mark_event_notifier_unreconciled,
-    reconcile_event_notifier_from_store, start_persisted_event_notifier_reconciler,
+    reconcile_event_notifier_from_store, start_persisted_event_notifier_reconciler, with_notify_runtime_reconcile_lock,
 };
 #[cfg(test)]
 pub(crate) use health::{

scripts/release/package_versions.sh

@@ -37,6 +37,14 @@ esac
 
 semver_core='(0|[1-9][0-9]*)\.(0|[1-9][0-9]*)\.(0|[1-9][0-9]*)'
 prerelease_id='(alpha|beta|rc)\.(0|[1-9][0-9]*)'
+# Every SemVer pre-release separator becomes '~' in both package formats.
+# dpkg and rpm (>= 4.10) both treat '~' as "sorts before anything, even the
+# empty string", so 1.0.0~rc.5 < 1.0.0 and 1.0.0~rc.5~preview.2 < 1.0.0~rc.5.
+# Neither '_' (an ordinary rpm segment separator, which makes 1.0.0_rc.5 sort
+# above 1.0.0) nor a second '-' (which dpkg reads as the start of the Debian
+# revision, so 1.0.0~rc.5-preview.2 sorts above 1.0.0~rc.5) preserves the
+# SemVer ordering. GitHub stores '~' in asset names as '.'; package.yml
+# accounts for that when it writes the release checksums.
 tilde='~'
 
 case "$build_type" in
@@ -65,8 +73,8 @@ case "$build_type" in
     [[ "$source_version" =~ ^${semver_core}-${prerelease_id}$ ]] ||
       fail "prerelease version must be strict alpha, beta, or rc SemVer"
 
-    deb_version=${source_version/-/$tilde}
-    rpm_version=${source_version//-/_}
+    deb_version=${source_version//-/$tilde}
+    rpm_version=${source_version//-/$tilde}
     rpm_release=1
     ;;
   preview)
@@ -74,8 +82,8 @@ case "$build_type" in
     [[ "$source_version" =~ ^${semver_core}-${prerelease_id}-preview\.(0|[1-9][0-9]*)$ ]] ||
       fail "preview version must be strict prerelease-preview SemVer"
 
-    deb_version=${source_version/-/$tilde}
-    rpm_version=${source_version//-/_}
+    deb_version=${source_version//-/$tilde}
+    rpm_version=${source_version//-/$tilde}
     rpm_release=1
     ;;
   *) fail "unsupported build type" ;;

scripts/test_package_versions.sh

@@ -74,16 +74,16 @@ run_success stable-amd64 \
   $'deb_version=1.2.3\nrpm_version=1.2.3\nrpm_release=1\ndeb_file=rustfs_1.2.3_amd64.deb\nrpm_file=rustfs-1.2.3-1.x86_64.rpm' \
   release 1.2.3 '' amd64 x86_64
 run_success alpha-arm64 \
-  $'deb_version=1.2.3~alpha.1\nrpm_version=1.2.3_alpha.1\nrpm_release=1\ndeb_file=rustfs_1.2.3~alpha.1_arm64.deb\nrpm_file=rustfs-1.2.3_alpha.1-1.aarch64.rpm' \
+  $'deb_version=1.2.3~alpha.1\nrpm_version=1.2.3~alpha.1\nrpm_release=1\ndeb_file=rustfs_1.2.3~alpha.1_arm64.deb\nrpm_file=rustfs-1.2.3~alpha.1-1.aarch64.rpm' \
   prerelease 1.2.3-alpha.1 '' arm64 aarch64
 run_success beta-amd64 \
-  $'deb_version=1.2.3~beta.2\nrpm_version=1.2.3_beta.2\nrpm_release=1\ndeb_file=rustfs_1.2.3~beta.2_amd64.deb\nrpm_file=rustfs-1.2.3_beta.2-1.x86_64.rpm' \
+  $'deb_version=1.2.3~beta.2\nrpm_version=1.2.3~beta.2\nrpm_release=1\ndeb_file=rustfs_1.2.3~beta.2_amd64.deb\nrpm_file=rustfs-1.2.3~beta.2-1.x86_64.rpm' \
   prerelease 1.2.3-beta.2 '' amd64 x86_64
 run_success rc-amd64 \
-  $'deb_version=1.2.3~rc.4\nrpm_version=1.2.3_rc.4\nrpm_release=1\ndeb_file=rustfs_1.2.3~rc.4_amd64.deb\nrpm_file=rustfs-1.2.3_rc.4-1.x86_64.rpm' \
+  $'deb_version=1.2.3~rc.4\nrpm_version=1.2.3~rc.4\nrpm_release=1\ndeb_file=rustfs_1.2.3~rc.4_amd64.deb\nrpm_file=rustfs-1.2.3~rc.4-1.x86_64.rpm' \
   prerelease 1.2.3-rc.4 '' amd64 x86_64
 run_success preview-amd64 \
-  $'deb_version=1.0.0~rc.5-preview.2\nrpm_version=1.0.0_rc.5_preview.2\nrpm_release=1\ndeb_file=rustfs_1.0.0~rc.5-preview.2_amd64.deb\nrpm_file=rustfs-1.0.0_rc.5_preview.2-1.x86_64.rpm' \
+  $'deb_version=1.0.0~rc.5~preview.2\nrpm_version=1.0.0~rc.5~preview.2\nrpm_release=1\ndeb_file=rustfs_1.0.0~rc.5~preview.2_amd64.deb\nrpm_file=rustfs-1.0.0~rc.5~preview.2-1.x86_64.rpm' \
   preview 1.0.0-rc.5-preview.2 '' amd64 x86_64
 run_success development-amd64 \
   "deb_version=0~dev.7463.${sha}
@@ -126,23 +126,43 @@ run_failure newline release $'1.2.3\nforged=1' '' amd64 x86_64
 run_failure unsupported-deb-arch release 1.2.3 '' x86_64 x86_64
 run_failure mismatched-arch release 1.2.3 '' amd64 aarch64
 
+# Ordering contract shared by both package managers: every pre-release sorts
+# below its final release, every preview sorts below the pre-release it
+# previews, and pre-release kinds/numbers keep their SemVer order.
 if command -v dpkg >/dev/null 2>&1; then
   dpkg --compare-versions "0~dev.7462.${sha}" lt "0~dev.7463.${sha}"
   dpkg --compare-versions "0~dev.7463.${sha}" lt 0.1.0
   dpkg --compare-versions 1.2.3~rc.4 lt 1.2.3
-  passed=$((passed + 3))
+  dpkg --compare-versions 1.2.3~alpha.1 lt 1.2.3~beta.2
+  dpkg --compare-versions 1.2.3~beta.2 lt 1.2.3~rc.4
+  dpkg --compare-versions 1.2.3~rc.9 lt 1.2.3~rc.10
+  dpkg --compare-versions 1.0.0~rc.5~preview.2 lt 1.0.0~rc.5
+  dpkg --compare-versions 1.0.0~rc.5~preview.1 lt 1.0.0~rc.5~preview.2
+  dpkg --compare-versions 1.0.0~rc.5~preview.2 lt 1.0.0~rc.6
+  dpkg --compare-versions 1.0.0~rc.5 lt 1.0.1
+  passed=$((passed + 10))
 elif [[ $require_package_managers == true ]]; then
   printf 'FAIL package ordering: dpkg is required\n' >&2
   exit 1
 fi
 
 if command -v rpm >/dev/null 2>&1; then
-  rpm_old="0-0.dev.7462.${sha}"
-  rpm_new="0-0.dev.7463.${sha}"
-  rpm_release=0.1.0-1
-  [[ $(rpm --eval "%{lua: print(rpm.vercmp('${rpm_old}', '${rpm_new}'))}") == -1 ]]
-  [[ $(rpm --eval "%{lua: print(rpm.vercmp('${rpm_new}', '${rpm_release}'))}") == -1 ]]
-  passed=$((passed + 2))
+  rpm_lt() {
+    [[ $(rpm --eval "%{lua: print(rpm.vercmp('$1', '$2'))}") == -1 ]] ||
+      { printf 'FAIL rpm ordering: expected %s < %s\n' "$1" "$2" >&2; exit 1; }
+    passed=$((passed + 1))
+  }
+  rpm_lt "0-0.dev.7462.${sha}" "0-0.dev.7463.${sha}"
+  rpm_lt "0-0.dev.7463.${sha}" 0.1.0-1
+  # The former 1.2.3_rc.4 spelling compared as newer than 1.2.3 (issue #8012).
+  rpm_lt 1.2.3~rc.4-1 1.2.3-1
+  rpm_lt 1.2.3~alpha.1-1 1.2.3~beta.2-1
+  rpm_lt 1.2.3~beta.2-1 1.2.3~rc.4-1
+  rpm_lt 1.2.3~rc.9-1 1.2.3~rc.10-1
+  rpm_lt 1.0.0~rc.5~preview.2-1 1.0.0~rc.5-1
+  rpm_lt 1.0.0~rc.5~preview.1-1 1.0.0~rc.5~preview.2-1
+  rpm_lt 1.0.0~rc.5~preview.2-1 1.0.0~rc.6-1
+  rpm_lt 1.0.0~rc.5-1 1.0.1-1
 elif [[ $require_package_managers == true ]]; then
   printf 'FAIL package ordering: rpm is required\n' >&2
   exit 1