fix(replication): harden live delete admission

crates/ecstore/src/api/mod.rs

@@ -178,15 +178,16 @@ pub mod bucket {
             mrf_backlog_observability_snapshot,
         };
         pub use crate::bucket::replication::{
-            BucketReplicationResyncStatus, BucketStats, DeletedObjectReplicationInfo, DurableMrfBacklog, DynReplicationPool,
-            MrfOpKind, MrfReplicateEntry, MustReplicateOptions, ObjectOpts, REPLICATE_INCOMING_DELETE, ReplicateDecision,
-            ReplicateObjectInfo, ReplicationConfig, ReplicationConfigurationExt, ReplicationDeleteScheduleInput,
-            ReplicationDeleteStateSource, ReplicationHealQueueResult, ReplicationObjectBridge, ReplicationObjectIO,
-            ReplicationOperation, ReplicationPoolTrait, ReplicationPriority, ReplicationQueueAdmission, ReplicationScannerBridge,
-            ReplicationState, ReplicationStats, ReplicationStatusType, ReplicationStorage, ReplicationTargetValidationError,
-            ReplicationType, ResyncOpts, ResyncStatusType, RuntimeReplicationTargetBacklog, TargetReplicationResyncStatus,
-            VersionPurgeStatusType, delete_replication_state_from_config, delete_replication_version_id,
-            get_global_replication_pool, get_global_replication_stats, init_background_replication, read_durable_mrf_backlog,
+            BucketReplicationResyncStatus, BucketStats, DeleteReplicationConfigSnapshot, DeletedObjectReplicationInfo,
+            DurableMrfBacklog, DynReplicationPool, MrfOpKind, MrfReplicateEntry, MustReplicateOptions, ObjectOpts,
+            REPLICATE_INCOMING_DELETE, ReplicateDecision, ReplicateObjectInfo, ReplicationConfig, ReplicationConfigurationExt,
+            ReplicationDeleteScheduleInput, ReplicationDeleteStateSource, ReplicationHealQueueResult, ReplicationObjectBridge,
+            ReplicationObjectIO, ReplicationOperation, ReplicationPoolTrait, ReplicationPriority, ReplicationQueueAdmission,
+            ReplicationScannerBridge, ReplicationState, ReplicationStats, ReplicationStatusType, ReplicationStorage,
+            ReplicationTargetValidationError, ReplicationType, ResyncOpts, ResyncStatusType, RuntimeReplicationTargetBacklog,
+            TargetReplicationResyncStatus, VersionPurgeStatusType, delete_replication_state_from_config,
+            delete_replication_version_id, get_global_replication_pool, get_global_replication_stats,
+            init_background_replication, invalid_replication_config_status_field, read_durable_mrf_backlog,
             replication_state_to_filemeta, replication_status_to_filemeta, replication_statuses_map, replication_target_arns,
             resync_start_conflict_id, should_remove_replication_target, should_schedule_delete_replication,
             should_use_existing_delete_replication_info, should_use_existing_delete_replication_source,

crates/ecstore/src/bucket/bucket_target_sys.rs

@@ -1995,6 +1995,49 @@ mod tests {
     use super::*;
     use rcgen::generate_simple_self_signed;
 
+    #[derive(Clone, Debug)]
+    struct RecordingHttpConnector {
+        request_uris: Arc<std::sync::Mutex<Vec<String>>>,
+    }
+
+    impl SmithyHttpConnector for RecordingHttpConnector {
+        fn call(&self, request: HttpRequest) -> HttpConnectorFuture {
+            self.request_uris
+                .lock()
+                .expect("recorded request lock should not be poisoned")
+                .push(request.uri().to_string());
+            HttpConnectorFuture::ready(Ok(HttpResponse::new(
+                aws_smithy_runtime_api::http::StatusCode::try_from(204_u16).expect("204 should be a valid response status"),
+                SdkBody::empty(),
+            )))
+        }
+    }
+
+    fn recording_target_client() -> (TargetClient, Arc<std::sync::Mutex<Vec<String>>>) {
+        let request_uris = Arc::new(std::sync::Mutex::new(Vec::new()));
+        let connector = SharedHttpConnector::new(RecordingHttpConnector {
+            request_uris: Arc::clone(&request_uris),
+        });
+        let http_client = http_client_fn(move |_settings, _components| connector.clone());
+        let client = s3_client_with_http_client(443, http_client);
+        (
+            TargetClient {
+                endpoint: "https://localhost:443".to_string(),
+                credentials: None,
+                bucket: "target-bucket".to_string(),
+                storage_class: String::new(),
+                disable_proxy: false,
+                arn: "arn:rustfs:replication:us-east-1:target:bucket".to_string(),
+                reset_id: String::new(),
+                secure: true,
+                health_check_duration: Duration::from_secs(5),
+                replicate_sync: false,
+                client: Arc::new(client),
+            },
+            request_uris,
+        )
+    }
+
     fn spawn_single_request_https_server(cert: &rcgen::CertifiedKey<rcgen::KeyPair>) -> (u16, std::thread::JoinHandle<()>) {
         use std::io::{Read, Write};
 
@@ -2285,6 +2328,32 @@ mod tests {
         assert_eq!(got.as_deref(), Some(vid.as_str()));
     }
 
+    #[tokio::test]
+    async fn remove_object_writes_null_purge_and_omits_marker_creation_version_queries() {
+        let (client, request_uris) = recording_target_client();
+        client
+            .remove_object("target-bucket", "object", Some("null".to_string()), remove_opts(true, false))
+            .await
+            .expect("explicit null version purge should reach the target client");
+        client
+            .remove_object("target-bucket", "object", Some(Uuid::new_v4().to_string()), remove_opts(true, true))
+            .await
+            .expect("delete marker creation should reach the target client");
+
+        let request_uris = request_uris.lock().expect("recorded request lock should not be poisoned");
+        assert_eq!(request_uris.len(), 2);
+        assert!(
+            request_uris[0].contains("versionId=null"),
+            "an explicit null purge must be emitted as a target versionId query: {}",
+            request_uris[0]
+        );
+        assert!(
+            !request_uris[1].contains("versionId="),
+            "delete marker creation must omit the target versionId query: {}",
+            request_uris[1]
+        );
+    }
+
     #[test]
     fn put_object_headers_include_non_empty_source_etag_only() {
         let mut opts = PutObjectOptions::default();

crates/ecstore/src/bucket/metadata.rs

@@ -16,6 +16,7 @@ use super::msgp_decode::{read_msgp_ext8_time, skip_msgp_value, write_msgp_time};
 use super::object_lock::ObjectLockApi;
 use super::versioning::VersioningApi;
 use super::{quota::BucketQuota, target::BucketTargets};
+use crate::bucket::replication::invalid_replication_config_status_field;
 use crate::bucket::utils::deserialize;
 use crate::config::com::{read_config, save_config};
 use crate::disk::BUCKET_META_PREFIX;
@@ -25,9 +26,9 @@ use crate::store::ECStore;
 use byteorder::{BigEndian, ByteOrder, LittleEndian};
 use rustfs_policy::policy::BucketPolicy;
 use s3s::dto::{
-    AccelerateConfiguration, BucketLifecycleConfiguration, BucketLoggingStatus, CORSConfiguration, NotificationConfiguration,
-    ObjectLockConfiguration, PublicAccessBlockConfiguration, ReplicationConfiguration, RequestPaymentConfiguration,
-    ServerSideEncryptionConfiguration, Tagging, VersioningConfiguration, WebsiteConfiguration,
+    AccelerateConfiguration, BucketLifecycleConfiguration, BucketLoggingStatus, BucketVersioningStatus, CORSConfiguration,
+    NotificationConfiguration, ObjectLockConfiguration, PublicAccessBlockConfiguration, ReplicationConfiguration,
+    RequestPaymentConfiguration, ServerSideEncryptionConfiguration, Tagging, VersioningConfiguration, WebsiteConfiguration,
 };
 use serde::Serializer;
 use sha2::{Digest, Sha256};
@@ -751,11 +752,33 @@ impl BucketMetadata {
                 self.object_lock_config_updated_at = updated;
             }
             BUCKET_VERSIONING_CONFIG => {
+                let config = if data.is_empty() {
+                    None
+                } else {
+                    let config = deserialize::<VersioningConfiguration>(&data)?;
+                    if config.status.as_ref().is_some_and(|status| {
+                        !matches!(status.as_str(), BucketVersioningStatus::ENABLED | BucketVersioningStatus::SUSPENDED)
+                    }) {
+                        return Err(Error::other("bucket versioning configuration has an invalid status"));
+                    }
+                    Some(config)
+                };
                 self.versioning_config_xml = data;
+                self.versioning_config = config;
                 self.versioning_config_updated_at = updated;
             }
             BUCKET_REPLICATION_CONFIG => {
+                let config = if data.is_empty() {
+                    None
+                } else {
+                    let config = deserialize::<ReplicationConfiguration>(&data)?;
+                    if let Some(field) = invalid_replication_config_status_field(&config) {
+                        return Err(Error::other(format!("replication field {field} has an invalid status")));
+                    }
+                    Some(config)
+                };
                 self.replication_config_xml = data;
+                self.replication_config = config;
                 self.replication_config_updated_at = updated;
             }
             BUCKET_TARGETS_FILE => {
@@ -825,7 +848,6 @@ impl BucketMetadata {
     /// ambient (first) one. [`BucketMetadata::save`] keeps the ambient default.
     pub async fn save_with_store(&mut self, store: std::sync::Arc<crate::store::ECStore>) -> Result<()> {
         self.parse_all_configs()?;
-
         let mut buf: Vec<u8> = vec![0; 4];
 
         LittleEndian::write_u16(&mut buf[0..2], BUCKET_METADATA_FORMAT);
@@ -906,6 +928,7 @@ impl BucketMetadata {
                 "Failed to parse bucket metadata config"
             );
         }
+        self.versioning_config = None;
         if !self.versioning_config_xml.is_empty()
             && let Err(e) =
                 deserialize::<VersioningConfiguration>(&self.versioning_config_xml).map(|c| self.versioning_config = Some(c))
@@ -960,6 +983,7 @@ impl BucketMetadata {
                 "Failed to parse bucket metadata config"
             );
         }
+        self.replication_config = None;
         if !self.replication_config_xml.is_empty()
             && let Err(e) =
                 deserialize::<ReplicationConfiguration>(&self.replication_config_xml).map(|c| self.replication_config = Some(c))
@@ -1345,6 +1369,66 @@ mod test {
         assert!(bm.tagging_config.is_none());
     }
 
+    #[test]
+    fn delete_admission_configs_update_parsed_state_atomically() {
+        let mut bm = BucketMetadata::new("test-bucket");
+        let versioning_xml = b"<VersioningConfiguration><Status>Enabled</Status></VersioningConfiguration>";
+        let replication_xml = b"<ReplicationConfiguration><Role>arn:aws:s3:::target-bucket</Role><Rule><ID>rule1</ID><Status>Enabled</Status><Prefix></Prefix><Destination><Bucket>arn:aws:s3:::target-bucket</Bucket></Destination></Rule></ReplicationConfiguration>";
+
+        bm.update_config(BUCKET_VERSIONING_CONFIG, versioning_xml.to_vec())
+            .expect("valid versioning config should update parsed state");
+        bm.update_config(BUCKET_REPLICATION_CONFIG, replication_xml.to_vec())
+            .expect("valid replication config should update parsed state");
+
+        assert!(bm.versioning_config.as_ref().is_some_and(VersioningConfiguration::enabled));
+        assert_eq!(
+            bm.replication_config.as_ref().map(|config| config.role.as_str()),
+            Some("arn:aws:s3:::target-bucket")
+        );
+
+        assert!(
+            bm.update_config(BUCKET_VERSIONING_CONFIG, b"<VersioningConfiguration>".to_vec())
+                .is_err()
+        );
+        assert!(
+            bm.update_config(BUCKET_REPLICATION_CONFIG, b"<ReplicationConfiguration>".to_vec())
+                .is_err()
+        );
+
+        assert_eq!(bm.versioning_config_xml, versioning_xml);
+        assert_eq!(bm.replication_config_xml, replication_xml);
+        assert!(bm.versioning_config.as_ref().is_some_and(VersioningConfiguration::enabled));
+        assert_eq!(
+            bm.replication_config.as_ref().map(|config| config.role.as_str()),
+            Some("arn:aws:s3:::target-bucket")
+        );
+
+        assert!(
+            bm.update_config(
+                BUCKET_VERSIONING_CONFIG,
+                b"<VersioningConfiguration><Status>Enabld</Status></VersioningConfiguration>".to_vec(),
+            )
+            .is_err()
+        );
+        assert!(
+            bm.update_config(
+                BUCKET_REPLICATION_CONFIG,
+                b"<ReplicationConfiguration><Role>arn:aws:s3:::target-bucket</Role><Rule><ID>rule1</ID><Status>Enabld</Status><Prefix></Prefix><Destination><Bucket>arn:aws:s3:::target-bucket</Bucket></Destination></Rule></ReplicationConfiguration>".to_vec(),
+            )
+            .is_err()
+        );
+        assert_eq!(bm.versioning_config_xml, versioning_xml);
+        assert_eq!(bm.replication_config_xml, replication_xml);
+
+        bm.versioning_config_xml = b"<VersioningConfiguration>".to_vec();
+        bm.replication_config_xml = b"<ReplicationConfiguration>".to_vec();
+        bm.parse_all_configs()
+            .expect("bulk config parsing reports malformed fields through cleared typed state");
+
+        assert!(bm.versioning_config.is_none());
+        assert!(bm.replication_config.is_none());
+    }
+
     #[tokio::test]
     async fn marshal_msg_complete_example() {
         // Create a complete BucketMetadata with various configurations

crates/ecstore/src/bucket/metadata_sys.rs

@@ -1027,7 +1027,6 @@ impl BucketMetadataSys {
     /// server's metadata never leaks into the ambient (first) instance.
     pub(crate) async fn persist_and_set(&self, bm: BucketMetadata) -> Result<()> {
         let mut bm = bm;
-
         bm.save_with_store(self.api.clone()).await?;
 
         self.set(bm.name.clone(), Arc::new(bm)).await;
@@ -1221,7 +1220,9 @@ impl BucketMetadataSys {
             }
         };
 
-        if let Some(config) = &bm.versioning_config {
+        if !bm.versioning_config_xml.is_empty() && bm.versioning_config.is_none() {
+            Err(Error::other("persisted bucket versioning configuration is invalid"))
+        } else if let Some(config) = &bm.versioning_config {
             Ok((config.clone(), bm.versioning_config_updated_at))
         } else {
             Ok((VersioningConfiguration::default(), bm.versioning_config_updated_at))
@@ -1407,7 +1408,9 @@ impl BucketMetadataSys {
     pub async fn get_replication_config(&self, bucket: &str) -> Result<(ReplicationConfiguration, OffsetDateTime)> {
         let (bm, _) = self.get_config(bucket).await?;
 
-        if let Some(config) = &bm.replication_config {
+        if !bm.replication_config_xml.is_empty() && bm.replication_config.is_none() {
+            Err(Error::other("persisted bucket replication configuration is invalid"))
+        } else if let Some(config) = &bm.replication_config {
             Ok((config.clone(), bm.replication_config_updated_at))
         } else {
             Err(Error::ConfigNotFound)
@@ -1481,6 +1484,28 @@ mod tests {
     use serial_test::serial;
     use tokio::time::timeout;
 
+    #[tokio::test]
+    async fn malformed_delete_configs_are_not_treated_as_absent() {
+        let (_dirs, ecstore) = isolated_store_over_temp_disks().await;
+        let sys = BucketMetadataSys::new(ecstore);
+        let bucket = "malformed-delete-config";
+        let mut metadata = BucketMetadata::new(bucket);
+        metadata.versioning_config_xml = b"<VersioningConfiguration>".to_vec();
+        metadata.versioning_config = None;
+        metadata.replication_config_xml = b"<ReplicationConfiguration>".to_vec();
+        metadata.replication_config = None;
+        sys.set(bucket.to_string(), Arc::new(metadata)).await;
+
+        assert!(
+            sys.get_versioning_config(bucket).await.is_err(),
+            "malformed versioning metadata must block destructive requests"
+        );
+        assert!(
+            sys.get_replication_config(bucket).await.is_err(),
+            "malformed replication metadata must not be reported as ConfigNotFound"
+        );
+    }
+
     /// Concurrent cache misses for one bucket must collapse into a single disk
     /// load.
     ///

crates/ecstore/src/bucket/replication/mod.rs

@@ -45,8 +45,9 @@ mod runtime_boundary;
 
 pub use datatypes::ResyncStatusType;
 pub use replication_config_boundary::{
-    ObjectOpts, ReplicationConfigurationExt, ReplicationTargetValidationError, replication_target_arns,
-    should_remove_replication_target, unsupported_replication_config_field, validate_replication_config_target_arns,
+    ObjectOpts, ReplicationConfigurationExt, ReplicationTargetValidationError, invalid_replication_config_status_field,
+    replication_target_arns, should_remove_replication_target, unsupported_replication_config_field,
+    validate_replication_config_target_arns,
 };
 #[cfg(test)]
 pub(crate) use replication_filemeta_boundary::ReplicateTargetDecision;
@@ -62,7 +63,7 @@ pub(crate) use replication_filemeta_boundary::{
 pub(crate) use replication_lifecycle_bridge::{ReplicationLifecycleBridge, ReplicationLifecycleConfig};
 pub(crate) use replication_migration_bridge::ReplicationMigrationBridge;
 pub use replication_object_bridge::ReplicationObjectBridge;
-pub use replication_object_config::ReplicationConfig;
+pub use replication_object_config::{DeleteReplicationConfigSnapshot, ReplicationConfig};
 pub use replication_object_decision_boundary::{
     MustReplicateOptions, ReplicationDeleteScheduleInput, ReplicationDeleteStateSource, delete_replication_state_from_config,
     delete_replication_version_id, should_schedule_delete_replication, should_use_existing_delete_replication_info,

crates/ecstore/src/bucket/replication/replication_config_boundary.rs

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 pub use rustfs_replication::{
-    ObjectOpts, ReplicationConfigurationExt, ReplicationTargetValidationError, replication_target_arns,
-    should_remove_replication_target, unsupported_replication_config_field, validate_replication_config_target_arns,
+    ObjectOpts, ReplicationConfigurationExt, ReplicationRuleExt, ReplicationTargetValidationError,
+    invalid_replication_config_status_field, replication_target_arns, should_remove_replication_target,
+    unsupported_replication_config_field, validate_replication_config_target_arns,
 };

crates/ecstore/src/bucket/replication/replication_filemeta_boundary.rs

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+pub(crate) use rustfs_filemeta::NULL_VERSION_ID;
 pub use rustfs_replication::{MrfOpKind, MrfReplicateEntry};
 pub(crate) use rustfs_replication::{
     REPLICATE_EXISTING, REPLICATE_EXISTING_DELETE, REPLICATE_HEAL_DELETE, ReplicateTargetDecision, ReplicatedInfos,

crates/ecstore/src/bucket/replication/replication_metadata_boundary.rs

@@ -12,14 +12,19 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use crate::bucket::metadata_sys;
+use std::sync::Arc;
+
+use crate::bucket::{metadata::BucketMetadata, metadata_sys};
 use crate::disk::{BUCKET_META_PREFIX, RUSTFS_META_BUCKET};
+use crate::runtime::instance::InstanceContext;
 use rustfs_utils::path::path_join_buf;
 use s3s::dto::ReplicationConfiguration;
 use time::OffsetDateTime;
 
 use super::replication_error_boundary::{Error, Result};
 
+pub(crate) type ReplicationInstanceContext = InstanceContext;
+
 const REPLICATION_DIR: &str = ".replication";
 const RESYNC_FILE_NAME: &str = "resync.bin";
 
@@ -45,6 +50,20 @@ impl ReplicationMetadataStore {
         Ok(config)
     }
 
+    pub(crate) async fn delete_metadata(bucket: &str) -> Result<Arc<BucketMetadata>> {
+        let sys = metadata_sys::get_bucket_metadata_sys()?;
+        let sys = sys.read().await;
+        Ok(sys.get_config(bucket).await?.0)
+    }
+
+    pub(crate) async fn delete_metadata_in(ctx: &ReplicationInstanceContext, bucket: &str) -> Result<Arc<BucketMetadata>> {
+        let sys = ctx
+            .bucket_metadata_sys()
+            .ok_or_else(|| Error::other("request instance bucket metadata system is not initialized"))?;
+        let sys = sys.read().await;
+        Ok(sys.get_config(bucket).await?.0)
+    }
+
     pub(crate) fn rustfs_meta_bucket() -> &'static str {
         RUSTFS_META_BUCKET
     }

crates/ecstore/src/bucket/replication/replication_object_bridge.rs

@@ -16,14 +16,17 @@ use std::{collections::HashMap, sync::Arc};
 
 use super::replication_error_boundary::Result;
 use super::replication_filemeta_boundary::{ReplicateDecision, ReplicationStatusType, ReplicationType};
+use super::replication_metadata_boundary::ReplicationInstanceContext;
 use super::replication_object_config::{
-    check_replicate_delete, check_replicate_delete_strict, get_must_replicate_options, must_replicate,
+    DeleteReplicationConfigSnapshot, check_replicate_delete, check_replicate_delete_strict, check_replicate_delete_with_snapshot,
+    get_must_replicate_options, load_delete_replication_config_in, load_delete_request_config_in, must_replicate,
 };
 use super::replication_object_decision_boundary::MustReplicateOptions;
 use super::replication_pool::{schedule_replication, schedule_replication_delete};
 use super::replication_queue_boundary::DeletedObjectReplicationInfo;
 use super::replication_storage_boundary::{
-    DeletedObject, ObjectInfo, ObjectOptions, ObjectToDelete, ReplicationStorage, deleted_object_for_replication,
+    DeletedObject, ObjectInfo, ObjectOptions, ObjectToDelete, ReplicationObjectStore, ReplicationStorage,
+    deleted_object_for_replication,
 };
 
 pub struct ReplicationObjectBridge;
@@ -63,6 +66,39 @@ impl ReplicationObjectBridge {
         check_replicate_delete_strict(bucket, object, source, opts, get_error).await
     }
 
+    pub async fn delete_request_config(api: &ReplicationObjectStore, bucket: &str) -> Result<DeleteReplicationConfigSnapshot> {
+        load_delete_request_config_in(&api.ctx, bucket).await
+    }
+
+    pub(crate) async fn delete_request_config_in(
+        ctx: &ReplicationInstanceContext,
+        bucket: &str,
+    ) -> Result<DeleteReplicationConfigSnapshot> {
+        load_delete_request_config_in(ctx, bucket).await
+    }
+
+    pub(crate) async fn delete_config_snapshot_in(
+        ctx: &ReplicationInstanceContext,
+        bucket: &str,
+        opts: &ObjectOptions,
+    ) -> Result<DeleteReplicationConfigSnapshot> {
+        load_delete_replication_config_in(ctx, bucket, opts).await
+    }
+
+    pub fn has_active_delete_rule(snapshot: &DeleteReplicationConfigSnapshot, object: &str) -> bool {
+        snapshot.has_active_rule(object)
+    }
+
+    pub fn check_delete_with_snapshot(
+        object: &ObjectToDelete,
+        source: &ObjectInfo,
+        opts: &ObjectOptions,
+        source_error: bool,
+        snapshot: &DeleteReplicationConfigSnapshot,
+    ) -> ReplicateDecision {
+        check_replicate_delete_with_snapshot(object, source, opts, source_error, snapshot)
+    }
+
     pub async fn schedule_object<S: ReplicationStorage>(
         object: ObjectInfo,
         storage: Arc<S>,

crates/ecstore/src/bucket/replication/replication_object_config.rs

@@ -12,23 +12,26 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::collections::HashMap;
+use std::{collections::HashMap, fmt, sync::Arc};
 
+use crate::bucket::metadata::BucketMetadata;
 use rustfs_utils::http::AMZ_BUCKET_REPLICATION_STATUS;
-use s3s::dto::ReplicationConfiguration;
+use s3s::dto::{BucketVersioningStatus, ReplicationConfiguration, ReplicationRuleStatus, VersioningConfiguration};
 use serde::{Deserialize, Serialize};
 use tracing::error;
 
-use super::replication_config_boundary::{ObjectOpts, ReplicationConfigurationExt as _};
+use super::replication_config_boundary::{
+    ObjectOpts, ReplicationConfigurationExt as _, ReplicationRuleExt as _, invalid_replication_config_status_field,
+};
 use super::replication_error_boundary::Result;
 use super::replication_filemeta_boundary::{
     ReplicateDecision, ReplicateTargetDecision, ReplicationStatusType, ReplicationType, ResyncDecision,
 };
 use super::replication_logging::{EVENT_RESYNC_CONFIG_LOOKUP_SKIPPED, LOG_COMPONENT_ECSTORE, LOG_SUBSYSTEM_REPLICATION_RESYNC};
-use super::replication_metadata_boundary::ReplicationMetadataStore;
+use super::replication_metadata_boundary::{ReplicationInstanceContext, ReplicationMetadataStore};
 use super::replication_object_decision_boundary::{
     MustReplicateOptions, ReplicationDeleteSource, ReplicationResyncTargetObject, delete_replication_missing_source_decision,
-    delete_replication_object_opts, resync_target_for_object,
+    delete_replication_object_opts, heal_uses_delete_replication_path, resync_target_for_object,
 };
 use super::replication_storage_boundary::{ObjectInfo, ObjectOptions, ObjectToDelete, object_to_delete_for_replication};
 use super::replication_target_boundary::{BucketTargets, ReplicationTargetStore};
@@ -36,7 +39,197 @@ use super::replication_versioning_boundary::ReplicationVersioningStore;
 use super::runtime_boundary as runtime_sources;
 
 pub(crate) async fn get_replication_config(bucket: &str) -> Result<Option<ReplicationConfiguration>> {
-    ReplicationMetadataStore::optional_replication_config(bucket).await
+    let config = ReplicationMetadataStore::optional_replication_config(bucket).await?;
+    validate_delete_replication_config(&VersioningConfiguration::default(), config.as_ref())?;
+    Ok(config)
+}
+
+#[derive(Default)]
+pub struct DeleteReplicationConfigSnapshot {
+    metadata: Option<Arc<BucketMetadata>>,
+    versioning: VersioningConfiguration,
+}
+
+impl fmt::Debug for DeleteReplicationConfigSnapshot {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("DeleteReplicationConfigSnapshot")
+            .field("has_replication_config", &self.replication_config().is_some())
+            .field("versioning_status", &self.versioning.status)
+            .finish()
+    }
+}
+
+impl DeleteReplicationConfigSnapshot {
+    #[cfg(test)]
+    pub(crate) fn from_configs_for_test(
+        versioning: VersioningConfiguration,
+        replication: Option<ReplicationConfiguration>,
+    ) -> Self {
+        let metadata = replication.map(|config| {
+            let mut metadata = BucketMetadata::new("test-bucket");
+            metadata.replication_config = Some(config);
+            Arc::new(metadata)
+        });
+        Self { metadata, versioning }
+    }
+
+    pub fn versioning_config(&self) -> &VersioningConfiguration {
+        &self.versioning
+    }
+
+    pub fn replication_config(&self) -> Option<&ReplicationConfiguration> {
+        self.metadata
+            .as_ref()
+            .and_then(|metadata| metadata.replication_config.as_ref())
+    }
+
+    pub(crate) fn has_active_rule(&self, object: &str) -> bool {
+        self.replication_config()
+            .is_some_and(|config| config.has_active_rules(object, true))
+    }
+
+    pub(crate) fn active_delete_marker_rules_require_tags(&self, object: &str) -> bool {
+        self.replication_config().is_some_and(|config| {
+            config.rules.iter().any(|rule| {
+                if rule.status == ReplicationRuleStatus::from_static(ReplicationRuleStatus::DISABLED) {
+                    return false;
+                }
+                if !object.starts_with(rule.prefix()) {
+                    return false;
+                }
+                rule.filter.as_ref().is_some_and(|filter| {
+                    filter.tag.is_some()
+                        || filter
+                            .and
+                            .as_ref()
+                            .and_then(|and| and.tags.as_ref())
+                            .is_some_and(|tags| !tags.is_empty())
+                })
+            })
+        })
+    }
+}
+
+fn validate_delete_replication_config(
+    versioning: &VersioningConfiguration,
+    config: Option<&ReplicationConfiguration>,
+) -> Result<()> {
+    if versioning
+        .status
+        .as_ref()
+        .is_some_and(|status| !matches!(status.as_str(), BucketVersioningStatus::ENABLED | BucketVersioningStatus::SUSPENDED))
+    {
+        return Err(super::replication_error_boundary::Error::other(
+            "bucket versioning configuration has an invalid status",
+        ));
+    }
+
+    if let Some(config) = config {
+        if let Some(field) = invalid_replication_config_status_field(config) {
+            return Err(super::replication_error_boundary::Error::other(format!(
+                "replication field {field} has an invalid status"
+            )));
+        }
+
+        let role = config.role.trim();
+        let mut role_destination = None;
+        for rule in &config.rules {
+            if rule.status.as_str() == ReplicationRuleStatus::ENABLED {
+                let destination = rule.destination.bucket.trim();
+                if role.is_empty() && destination.is_empty() {
+                    return Err(super::replication_error_boundary::Error::other(
+                        "enabled replication rule has no destination ARN",
+                    ));
+                }
+                if !role.is_empty() && !destination.is_empty() {
+                    match role_destination {
+                        Some(existing) if existing != destination => {
+                            return Err(super::replication_error_boundary::Error::other(
+                                "replication role cannot address multiple active destinations",
+                            ));
+                        }
+                        None => role_destination = Some(destination),
+                        _ => {}
+                    }
+                }
+            }
+        }
+    }
+
+    Ok(())
+}
+
+fn replication_config_from_metadata(metadata: &BucketMetadata) -> Result<Option<&ReplicationConfiguration>> {
+    if !metadata.replication_config_xml.is_empty() && metadata.replication_config.is_none() {
+        return Err(super::replication_error_boundary::Error::other(
+            "persisted bucket replication configuration is invalid",
+        ));
+    }
+    Ok(metadata.replication_config.as_ref())
+}
+
+fn delete_request_snapshot_from_metadata(metadata: Arc<BucketMetadata>) -> Result<DeleteReplicationConfigSnapshot> {
+    if !metadata.versioning_config_xml.is_empty() && metadata.versioning_config.is_none() {
+        return Err(super::replication_error_boundary::Error::other(
+            "persisted bucket versioning configuration is invalid",
+        ));
+    }
+
+    let versioning = metadata.versioning_config.clone().unwrap_or_default();
+    let has_config = {
+        let config = replication_config_from_metadata(&metadata)?;
+        if versioning.status.is_none() && config.is_some() {
+            return Err(super::replication_error_boundary::Error::other(
+                "bucket replication configuration requires versioning",
+            ));
+        }
+        validate_delete_replication_config(&versioning, config)?;
+        config.is_some()
+    };
+    Ok(DeleteReplicationConfigSnapshot {
+        metadata: has_config.then_some(metadata),
+        versioning,
+    })
+}
+
+fn delete_snapshot_from_metadata(metadata: Arc<BucketMetadata>) -> Result<DeleteReplicationConfigSnapshot> {
+    let has_config = {
+        let config = replication_config_from_metadata(&metadata)?;
+        validate_delete_replication_config(&VersioningConfiguration::default(), config)?;
+        config.is_some()
+    };
+    Ok(DeleteReplicationConfigSnapshot {
+        metadata: has_config.then_some(metadata),
+        versioning: VersioningConfiguration::default(),
+    })
+}
+
+pub(crate) async fn load_delete_request_config_in(
+    ctx: &ReplicationInstanceContext,
+    bucket: &str,
+) -> Result<DeleteReplicationConfigSnapshot> {
+    delete_request_snapshot_from_metadata(ReplicationMetadataStore::delete_metadata_in(ctx, bucket).await?)
+}
+
+pub(crate) async fn load_delete_replication_config(
+    bucket: &str,
+    opts: &ObjectOptions,
+) -> Result<DeleteReplicationConfigSnapshot> {
+    if opts.replication_request || (!opts.versioned && !opts.version_suspended) {
+        return Ok(DeleteReplicationConfigSnapshot::default());
+    }
+    delete_snapshot_from_metadata(ReplicationMetadataStore::delete_metadata(bucket).await?)
+}
+
+pub(crate) async fn load_delete_replication_config_in(
+    ctx: &ReplicationInstanceContext,
+    bucket: &str,
+    opts: &ObjectOptions,
+) -> Result<DeleteReplicationConfigSnapshot> {
+    if opts.replication_request || (!opts.versioned && !opts.version_suspended) {
+        return Ok(DeleteReplicationConfigSnapshot::default());
+    }
+    delete_snapshot_from_metadata(ReplicationMetadataStore::delete_metadata_in(ctx, bucket).await?)
 }
 
 #[derive(Debug, Clone, Default, Serialize, Deserialize)]
@@ -54,10 +247,31 @@ impl ReplicationConfig {
         self.config.is_none()
     }
 
+    pub(crate) fn validate(&self) -> Result<()> {
+        validate_delete_replication_config(&VersioningConfiguration::default(), self.config.as_ref())
+    }
+
     pub fn replicate(&self, obj: &ObjectOpts) -> bool {
         self.config.as_ref().is_some_and(|config| config.replicate(obj))
     }
 
+    pub(crate) fn check_delete_for_heal(
+        &self,
+        object: &ObjectToDelete,
+        source: &ObjectInfo,
+        opts: &ObjectOptions,
+    ) -> ReplicateDecision {
+        check_replicate_delete_with_config(
+            object,
+            source,
+            opts,
+            false,
+            self.config.as_ref(),
+            source.delete_marker && source.version_purge_status.is_empty(),
+            true,
+        )
+    }
+
     pub async fn resync(
         &self,
         oi: ObjectInfo,
@@ -70,30 +284,34 @@ impl ReplicationConfig {
 
         let mut dsc = dsc;
 
-        if oi.delete_marker {
+        if heal_uses_delete_replication_path(oi.delete_marker, &oi.version_purge_status) {
+            if !dsc.targets_map.is_empty() {
+                return self.resync_internal(oi, dsc, status);
+            }
             let opts = ObjectOpts {
                 name: oi.name.clone(),
-                version_id: oi.version_id,
-                delete_marker: true,
+                version_id: if oi.version_purge_status.is_empty() {
+                    None
+                } else {
+                    oi.version_id
+                },
+                delete_marker: oi.delete_marker,
                 op_type: ReplicationType::Delete,
                 existing_object: true,
                 ..Default::default()
             };
-            let arns = self
+            let targets = self
                 .config
                 .as_ref()
-                .map(|config| config.filter_target_arns(&opts))
+                .map(|config| config.filter_target_replication_decisions(&opts))
                 .unwrap_or_default();
 
-            if arns.is_empty() {
+            if targets.is_empty() {
                 return ResyncDecision::default();
             }
 
-            for arn in arns {
-                let mut opts = opts.clone();
-                opts.target_arn = arn;
-
-                dsc.set(ReplicateTargetDecision::new(opts.target_arn.clone(), self.replicate(&opts), false));
+            for (arn, replicate) in targets {
+                dsc.set(ReplicateTargetDecision::new(arn, replicate, false));
             }
 
             return self.resync_internal(oi, dsc, status);
@@ -169,8 +387,10 @@ pub(crate) async fn check_replicate_delete(
     del_opts: &ObjectOptions,
     gerr: Option<String>,
 ) -> ReplicateDecision {
-    match check_replicate_delete_strict(bucket, dobj, oi, del_opts, gerr).await {
-        Ok(decision) => decision,
+    match load_delete_replication_config(bucket, del_opts).await {
+        Ok(snapshot) => {
+            check_replicate_delete_with_config(dobj, oi, del_opts, gerr.is_some(), snapshot.replication_config(), false, false)
+        }
         Err(err) => {
             error!(
                 event = EVENT_RESYNC_CONFIG_LOOKUP_SKIPPED,
@@ -193,66 +413,101 @@ pub(crate) async fn check_replicate_delete_strict(
     del_opts: &ObjectOptions,
     gerr: Option<String>,
 ) -> Result<ReplicateDecision> {
-    let rcfg = match get_replication_config(bucket).await {
-        Ok(Some(config)) => config,
-        Ok(None) => return Ok(ReplicateDecision::default()),
-        Err(err) => return Err(err),
-    };
-
-    if del_opts.replication_request {
+    let Some(config) = get_replication_config(bucket).await? else {
         return Ok(ReplicateDecision::default());
+    };
+    let mut decision = check_replicate_delete_with_config(dobj, oi, del_opts, gerr.is_some(), Some(&config), false, false);
+    if gerr.is_some() {
+        return Ok(decision);
+    }
+
+    for target in decision.targets_map.values_mut() {
+        if let Some(client) = ReplicationTargetStore::remote_target_client(bucket, &target.arn).await {
+            target.synchronous = client.replicate_sync;
+        } else {
+            target.replicate = false;
+            target.synchronous = false;
+        }
+    }
+    Ok(decision)
+}
+
+pub(crate) fn check_replicate_delete_with_snapshot(
+    dobj: &ObjectToDelete,
+    oi: &ObjectInfo,
+    del_opts: &ObjectOptions,
+    source_error: bool,
+    snapshot: &DeleteReplicationConfigSnapshot,
+) -> ReplicateDecision {
+    check_replicate_delete_with_config(dobj, oi, del_opts, source_error, snapshot.replication_config(), false, false)
+}
+
+fn check_replicate_delete_with_config(
+    dobj: &ObjectToDelete,
+    oi: &ObjectInfo,
+    del_opts: &ObjectOptions,
+    source_error: bool,
+    config: Option<&ReplicationConfiguration>,
+    existing_delete_marker: bool,
+    trust_persisted_replica_status: bool,
+) -> ReplicateDecision {
+    if del_opts.replication_request {
+        return ReplicateDecision::default();
     }
 
     if !del_opts.versioned && !del_opts.version_suspended {
-        return Ok(ReplicateDecision::default());
+        return ReplicateDecision::default();
     }
 
+    let Some(rcfg) = config else {
+        return ReplicateDecision::default();
+    };
+
     let replication_delete = object_to_delete_for_replication(dobj);
-    let opts = delete_replication_object_opts(
+    let missing_source_marker = source_error && dobj.version_id.is_none();
+    let mut opts = delete_replication_object_opts(
         &replication_delete,
         &ReplicationDeleteSource {
             user_defined: oi.user_defined.as_ref(),
             user_tags: oi.user_tags.as_str(),
-            delete_marker: oi.delete_marker,
-            replication_status: oi.replication_status.clone(),
+            delete_marker: oi.delete_marker || missing_source_marker,
+            replication_status: if trust_persisted_replica_status {
+                oi.replication_status.clone()
+            } else {
+                ReplicationStatusType::Empty
+            },
         },
     );
-
-    let tgt_arns = rcfg.filter_target_arns(&opts);
-    let mut dsc = ReplicateDecision::new();
-
-    if tgt_arns.is_empty() {
-        return Ok(dsc);
+    if existing_delete_marker {
+        opts.version_id = None;
     }
 
-    for tgt_arn in tgt_arns {
-        let mut opts = opts.clone();
-        opts.target_arn = tgt_arn.clone();
-        let replicate = rcfg.replicate(&opts);
-        let sync = false;
+    let target_decisions = rcfg.filter_target_replication_decisions(&opts);
+    let mut dsc = ReplicateDecision::new();
 
-        if gerr.is_some() {
-            if let Some(replicate) = delete_replication_missing_source_decision(
-                oi.delete_marker,
+    if target_decisions.is_empty() {
+        return dsc;
+    }
+
+    for (tgt_arn, replicate) in target_decisions {
+        let effective_replicate = if source_error {
+            delete_replication_missing_source_decision(
+                oi.delete_marker || missing_source_marker,
                 oi.target_replication_status(&tgt_arn),
                 replicate,
                 &oi.version_purge_status,
-            ) {
-                dsc.set(ReplicateTargetDecision::new(tgt_arn, replicate, sync));
-            }
-            continue;
-        }
-
-        let tgt = ReplicationTargetStore::remote_target_client(bucket, &tgt_arn).await;
-        let tgt_dsc = if let Some(tgt) = tgt {
-            ReplicateTargetDecision::new(tgt_arn, replicate, tgt.replicate_sync)
+            )
         } else {
-            ReplicateTargetDecision::new(tgt_arn, false, false)
+            Some(replicate)
         };
-        dsc.set(tgt_dsc);
+        let Some(effective_replicate) = effective_replicate else {
+            continue;
+        };
+
+        dsc.set(ReplicateTargetDecision::new(tgt_arn, effective_replicate, false));
     }
 
-    Ok(dsc)
+    dsc
 }
 
 pub(crate) async fn must_replicate(bucket: &str, object: &str, mopts: MustReplicateOptions) -> ReplicateDecision {
@@ -312,8 +567,13 @@ pub(crate) async fn must_replicate(bucket: &str, object: &str, mopts: MustReplic
 
 #[cfg(test)]
 mod tests {
-    use s3s::dto::{Destination, ReplicationRule, ReplicationRuleStatus};
+    use s3s::dto::{
+        DeleteMarkerReplication, DeleteMarkerReplicationStatus, DeleteReplication, DeleteReplicationStatus, Destination,
+        ReplicaModifications, ReplicationRule, ReplicationRuleFilter, ReplicationRuleStatus, SourceSelectionCriteria, Tag,
+    };
 
+    use super::super::replication_filemeta_boundary::VersionPurgeStatusType;
+    use super::super::replication_target_boundary::BucketTarget;
     use super::*;
 
     fn replication_rule() -> ReplicationRule {
@@ -373,4 +633,379 @@ mod tests {
         assert!(options.is_replication_request());
         assert_eq!(options.user_tags(), "env=prod");
     }
+
+    #[test]
+    fn delete_snapshot_rejects_enabled_rules_without_a_destination() {
+        let mut rule = replication_rule();
+        rule.destination.bucket.clear();
+        let config = ReplicationConfiguration {
+            role: String::new(),
+            rules: vec![rule],
+        };
+
+        let err = validate_delete_replication_config(&VersioningConfiguration::default(), Some(&config))
+            .expect_err("an enabled rule without a destination must fail closed");
+
+        assert!(err.to_string().contains("destination ARN"));
+    }
+
+    #[test]
+    fn delete_snapshot_rejects_role_with_multiple_destinations() {
+        let first = replication_rule();
+        let mut second = replication_rule();
+        second.destination.bucket = "arn:aws:s3:::other-target".to_string();
+        let config = ReplicationConfiguration {
+            role: "arn:aws:s3:::role-target".to_string(),
+            rules: vec![first, second],
+        };
+
+        assert!(validate_delete_replication_config(&VersioningConfiguration::default(), Some(&config)).is_err());
+    }
+
+    #[test]
+    fn delete_snapshot_rejects_unknown_status_values() {
+        let invalid_versioning = VersioningConfiguration {
+            status: Some("Enabld".to_string().into()),
+            ..Default::default()
+        };
+        assert!(validate_delete_replication_config(&invalid_versioning, None).is_err());
+
+        let mut invalid_rule = replication_rule();
+        invalid_rule.status = "Enabld".to_string().into();
+        let config = ReplicationConfiguration {
+            role: String::new(),
+            rules: vec![invalid_rule],
+        };
+        assert!(validate_delete_replication_config(&VersioningConfiguration::default(), Some(&config)).is_err());
+
+        let mut invalid_delete = replication_rule();
+        invalid_delete.delete_replication = Some(DeleteReplication {
+            status: "Enabld".to_string().into(),
+        });
+        let config = ReplicationConfiguration {
+            role: String::new(),
+            rules: vec![invalid_delete],
+        };
+        assert!(validate_delete_replication_config(&VersioningConfiguration::default(), Some(&config)).is_err());
+
+        let mut invalid_delete_marker = replication_rule();
+        invalid_delete_marker.delete_marker_replication = Some(DeleteMarkerReplication {
+            status: Some("Enabld".to_string().into()),
+        });
+        let config = ReplicationConfiguration {
+            role: String::new(),
+            rules: vec![invalid_delete_marker],
+        };
+        assert!(validate_delete_replication_config(&VersioningConfiguration::default(), Some(&config)).is_err());
+
+        let mut invalid_replica_modifications = replication_rule();
+        invalid_replica_modifications.source_selection_criteria = Some(SourceSelectionCriteria {
+            replica_modifications: Some(ReplicaModifications {
+                status: "Enabld".to_string().into(),
+            }),
+            sse_kms_encrypted_objects: None,
+        });
+        let config = ReplicationConfiguration {
+            role: String::new(),
+            rules: vec![invalid_replica_modifications],
+        };
+        assert!(validate_delete_replication_config(&VersioningConfiguration::default(), Some(&config)).is_err());
+    }
+
+    #[test]
+    fn request_snapshot_borrows_cached_replication_config() {
+        let mut metadata = BucketMetadata::new("bucket");
+        metadata.versioning_config_xml = b"configured".to_vec();
+        metadata.versioning_config = Some(VersioningConfiguration {
+            status: Some(BucketVersioningStatus::from_static(BucketVersioningStatus::ENABLED)),
+            ..Default::default()
+        });
+        metadata.replication_config_xml = b"configured".to_vec();
+        metadata.replication_config = Some(ReplicationConfiguration {
+            role: String::new(),
+            rules: vec![replication_rule()],
+        });
+        let metadata = Arc::new(metadata);
+        let cached_config = metadata.replication_config.as_ref().expect("cached config") as *const _;
+
+        let snapshot = delete_request_snapshot_from_metadata(Arc::clone(&metadata)).expect("valid snapshot");
+
+        assert_eq!(snapshot.replication_config().expect("snapshot config") as *const _, cached_config);
+    }
+
+    #[test]
+    fn delete_snapshot_debug_redacts_bucket_target_credentials() {
+        let secret = "snapshot-secret-must-not-be-formatted";
+        let mut metadata = BucketMetadata::new("bucket");
+        metadata.bucket_targets_config_json = format!(r#"{{"secretKey":"{secret}"}}"#).into_bytes();
+        let opts = ObjectOptions {
+            delete_replication_config_snapshot: Some(Arc::new(DeleteReplicationConfigSnapshot {
+                metadata: Some(Arc::new(metadata)),
+                versioning: VersioningConfiguration::default(),
+            })),
+            ..Default::default()
+        };
+
+        let debug = format!("{opts:?}");
+        assert!(!debug.contains(secret), "delete tracing must not expose replication target credentials");
+        assert!(debug.contains("has_replication_config"));
+    }
+
+    #[test]
+    fn request_snapshot_rejects_replication_without_versioning_status() {
+        let mut malformed = BucketMetadata::new("bucket");
+        malformed.replication_config_xml = b"<ReplicationConfiguration>".to_vec();
+        assert!(
+            delete_request_snapshot_from_metadata(Arc::new(malformed)).is_err(),
+            "malformed replication metadata must fail closed even when versioning has no status"
+        );
+
+        let mut inconsistent = BucketMetadata::new("bucket");
+        inconsistent.replication_config = Some(ReplicationConfiguration {
+            role: String::new(),
+            rules: vec![replication_rule()],
+        });
+        assert!(
+            delete_request_snapshot_from_metadata(Arc::new(inconsistent)).is_err(),
+            "replication metadata without an enabled or suspended versioning state must fail closed"
+        );
+    }
+
+    #[test]
+    fn missing_source_marker_creation_is_still_admitted() {
+        let arn = "arn:rustfs:replication:us-east-1:target:bucket";
+        let mut rule = replication_rule();
+        rule.destination.bucket = arn.to_string();
+        rule.delete_marker_replication = Some(DeleteMarkerReplication {
+            status: Some(DeleteMarkerReplicationStatus::from_static(DeleteMarkerReplicationStatus::ENABLED)),
+        });
+        let mut metadata = BucketMetadata::new("bucket");
+        metadata.replication_config = Some(ReplicationConfiguration {
+            role: String::new(),
+            rules: vec![rule],
+        });
+        let snapshot = DeleteReplicationConfigSnapshot {
+            metadata: Some(Arc::new(metadata)),
+            ..Default::default()
+        };
+
+        let decision = check_replicate_delete_with_snapshot(
+            &ObjectToDelete {
+                object_name: "object".to_string(),
+                ..Default::default()
+            },
+            &ObjectInfo::default(),
+            &ObjectOptions {
+                versioned: true,
+                ..Default::default()
+            },
+            true,
+            &snapshot,
+        );
+
+        assert!(decision.replicate_any());
+        assert!(decision.targets_map.get(arn).is_some_and(|target| target.replicate));
+    }
+
+    #[test]
+    fn delete_marker_source_read_is_required_only_for_tag_filtered_rules() {
+        let mut prefix_rule = replication_rule();
+        prefix_rule.prefix = Some("logs/".to_string());
+        prefix_rule.delete_marker_replication = Some(DeleteMarkerReplication {
+            status: Some(DeleteMarkerReplicationStatus::from_static(DeleteMarkerReplicationStatus::ENABLED)),
+        });
+
+        let prefix_snapshot = DeleteReplicationConfigSnapshot::from_configs_for_test(
+            VersioningConfiguration {
+                status: Some(BucketVersioningStatus::from_static(BucketVersioningStatus::ENABLED)),
+                ..Default::default()
+            },
+            Some(ReplicationConfiguration {
+                role: String::new(),
+                rules: vec![prefix_rule],
+            }),
+        );
+
+        let mut tag_rule = replication_rule();
+        tag_rule.delete_marker_replication = Some(DeleteMarkerReplication {
+            status: Some(DeleteMarkerReplicationStatus::from_static(DeleteMarkerReplicationStatus::ENABLED)),
+        });
+        tag_rule.filter = Some(ReplicationRuleFilter {
+            tag: Some(Tag {
+                key: Some("class".to_string()),
+                value: Some("audit".to_string()),
+            }),
+            ..Default::default()
+        });
+        let tag_snapshot = DeleteReplicationConfigSnapshot::from_configs_for_test(
+            VersioningConfiguration {
+                status: Some(BucketVersioningStatus::from_static(BucketVersioningStatus::ENABLED)),
+                ..Default::default()
+            },
+            Some(ReplicationConfiguration {
+                role: String::new(),
+                rules: vec![tag_rule],
+            }),
+        );
+
+        assert!(!prefix_snapshot.active_delete_marker_rules_require_tags("logs/2026/app.log"));
+        assert!(tag_snapshot.active_delete_marker_rules_require_tags("logs/2026/app.log"));
+    }
+
+    #[test]
+    fn heal_uses_delete_switch_for_pending_purges_and_marker_switch_for_stored_markers() {
+        let arn = "arn:rustfs:replication:us-east-1:target:bucket";
+        let mut rule = replication_rule();
+        rule.destination.bucket = arn.to_string();
+        rule.delete_replication = Some(DeleteReplication {
+            status: DeleteReplicationStatus::from_static(DeleteReplicationStatus::ENABLED),
+        });
+        rule.delete_marker_replication = Some(DeleteMarkerReplication {
+            status: Some(DeleteMarkerReplicationStatus::from_static(DeleteMarkerReplicationStatus::DISABLED)),
+        });
+        let config = ReplicationConfig::new(
+            Some(ReplicationConfiguration {
+                role: String::new(),
+                rules: vec![rule],
+            }),
+            None,
+        );
+        let object = ObjectToDelete {
+            object_name: "object".to_string(),
+            version_id: Some(uuid::Uuid::new_v4()),
+            ..Default::default()
+        };
+        let opts = ObjectOptions {
+            versioned: true,
+            ..Default::default()
+        };
+
+        let purge = ObjectInfo {
+            version_id: object.version_id,
+            version_purge_status: VersionPurgeStatusType::Pending,
+            ..Default::default()
+        };
+        assert!(config.check_delete_for_heal(&object, &purge, &opts).replicate_any());
+
+        let marker = ObjectInfo {
+            delete_marker: true,
+            version_id: object.version_id,
+            ..Default::default()
+        };
+        assert!(!config.check_delete_for_heal(&object, &marker, &opts).replicate_any());
+    }
+
+    #[test]
+    fn live_delete_does_not_trust_persisted_replica_status() {
+        let mut rule = replication_rule();
+        rule.delete_replication = Some(DeleteReplication {
+            status: DeleteReplicationStatus::from_static(DeleteReplicationStatus::ENABLED),
+        });
+        rule.source_selection_criteria = Some(SourceSelectionCriteria {
+            replica_modifications: Some(ReplicaModifications {
+                status: s3s::dto::ReplicaModificationsStatus::from_static(s3s::dto::ReplicaModificationsStatus::DISABLED),
+            }),
+            sse_kms_encrypted_objects: None,
+        });
+        let replication = ReplicationConfiguration {
+            role: String::new(),
+            rules: vec![rule],
+        };
+        let snapshot = DeleteReplicationConfigSnapshot::from_configs_for_test(
+            VersioningConfiguration {
+                status: Some(BucketVersioningStatus::from_static(BucketVersioningStatus::ENABLED)),
+                ..Default::default()
+            },
+            Some(replication.clone()),
+        );
+        let object = ObjectToDelete {
+            object_name: "object".to_string(),
+            version_id: Some(uuid::Uuid::new_v4()),
+            ..Default::default()
+        };
+        let source = ObjectInfo {
+            replication_status: ReplicationStatusType::Replica,
+            ..Default::default()
+        };
+        let opts = ObjectOptions {
+            versioned: true,
+            ..Default::default()
+        };
+
+        assert!(
+            check_replicate_delete_with_snapshot(&object, &source, &opts, false, &snapshot).replicate_any(),
+            "an ordinary authenticated delete must not inherit replica identity from object metadata"
+        );
+        assert!(
+            !ReplicationConfig::new(Some(replication), None)
+                .check_delete_for_heal(&object, &source, &opts)
+                .replicate_any(),
+            "heal must still honor the persisted replica identity"
+        );
+    }
+
+    #[tokio::test]
+    async fn resync_keeps_marker_version_purges_separate_from_marker_creation() {
+        let arn = "arn:rustfs:replication:us-east-1:target:bucket";
+        let mut rule = replication_rule();
+        rule.destination.bucket = arn.to_string();
+        rule.delete_replication = Some(DeleteReplication {
+            status: DeleteReplicationStatus::from_static(DeleteReplicationStatus::ENABLED),
+        });
+        rule.delete_marker_replication = Some(DeleteMarkerReplication {
+            status: Some(DeleteMarkerReplicationStatus::from_static(DeleteMarkerReplicationStatus::DISABLED)),
+        });
+        let config = ReplicationConfig::new(
+            Some(ReplicationConfiguration {
+                role: String::new(),
+                rules: vec![rule],
+            }),
+            Some(BucketTargets {
+                targets: vec![BucketTarget {
+                    arn: arn.to_string(),
+                    ..Default::default()
+                }],
+            }),
+        );
+        let marker = ObjectInfo {
+            name: "object".to_string(),
+            delete_marker: true,
+            version_id: Some(uuid::Uuid::new_v4()),
+            ..Default::default()
+        };
+
+        let purge = config
+            .resync(
+                ObjectInfo {
+                    version_purge_status: VersionPurgeStatusType::Pending,
+                    ..marker.clone()
+                },
+                ReplicateDecision::default(),
+                &HashMap::new(),
+            )
+            .await;
+        assert!(purge.targets.get(arn).is_some_and(|target| target.replicate));
+
+        let mut object_purge_decision = ReplicateDecision::default();
+        object_purge_decision.set(ReplicateTargetDecision::new(arn.to_string(), true, false));
+        let object_purge = config
+            .resync(
+                ObjectInfo {
+                    name: "object".to_string(),
+                    version_id: Some(uuid::Uuid::new_v4()),
+                    version_purge_status: VersionPurgeStatusType::Pending,
+                    ..Default::default()
+                },
+                object_purge_decision,
+                &HashMap::new(),
+            )
+            .await;
+        assert!(
+            object_purge.targets.get(arn).is_some_and(|target| target.replicate),
+            "non-marker PENDING purges must stay on the delete resync path"
+        );
+
+        let stored_marker = config.resync(marker, ReplicateDecision::default(), &HashMap::new()).await;
+        assert!(!stored_marker.targets.contains_key(arn));
+    }
 }

crates/ecstore/src/bucket/replication/replication_pool.rs

@@ -1967,7 +1967,24 @@ pub(crate) async fn queue_replication_heal_internal(
         };
     }
 
-    roi = get_heal_replicate_object_info(&oi, &rcfg).await;
+    roi = match get_heal_replicate_object_info(&oi, &rcfg).await {
+        Ok(roi) => roi,
+        Err(err) => {
+            warn!(
+                event = EVENT_REPLICATION_CONFIG_LOOKUP_SKIPPED,
+                component = LOG_COMPONENT_ECSTORE,
+                subsystem = LOG_SUBSYSTEM_REPLICATION,
+                bucket = %oi.bucket,
+                object = %oi.name,
+                error = %err,
+                "Failed to classify object for replication heal"
+            );
+            return ReplicationHealQueueResult {
+                object_info: roi,
+                admission: ReplicationQueueAdmission::Missed,
+            };
+        }
+    };
     roi.retry_count = retry_count;
 
     match replication_heal_queue_action(&mut roi) {
@@ -2861,6 +2878,55 @@ mod tests {
         assert_eq!(admission, ReplicationQueueAdmission::Missed);
     }
 
+    #[tokio::test]
+    async fn heal_queue_marks_missing_versioning_state_as_missed() {
+        use super::super::replication_target_boundary::BucketTargets;
+        use s3s::dto::{
+            DeleteReplication, DeleteReplicationStatus, Destination, ReplicationConfiguration, ReplicationRule,
+            ReplicationRuleStatus,
+        };
+
+        let arn = "arn:rustfs:replication:us-east-1:target:bucket";
+        let result = queue_replication_heal_internal(
+            "missing-versioning-state",
+            ObjectInfo {
+                bucket: "missing-versioning-state".to_string(),
+                name: "object".to_string(),
+                version_id: Some(Uuid::new_v4()),
+                version_purge_status: super::super::replication_filemeta_boundary::VersionPurgeStatusType::Pending,
+                mod_time: Some(OffsetDateTime::now_utc()),
+                ..Default::default()
+            },
+            ReplicationConfig::new(
+                Some(ReplicationConfiguration {
+                    role: String::new(),
+                    rules: vec![ReplicationRule {
+                        delete_marker_replication: None,
+                        delete_replication: Some(DeleteReplication {
+                            status: DeleteReplicationStatus::from_static(DeleteReplicationStatus::ENABLED),
+                        }),
+                        destination: Destination {
+                            bucket: arn.to_string(),
+                            ..Default::default()
+                        },
+                        existing_object_replication: None,
+                        filter: None,
+                        id: Some("delete".to_string()),
+                        prefix: Some(String::new()),
+                        priority: Some(1),
+                        source_selection_criteria: None,
+                        status: ReplicationRuleStatus::from_static(ReplicationRuleStatus::ENABLED),
+                    }],
+                }),
+                Some(BucketTargets::default()),
+            ),
+            0,
+        )
+        .await;
+
+        assert_eq!(result.admission, ReplicationQueueAdmission::Missed);
+    }
+
     #[tokio::test]
     async fn queue_replica_task_counts_mrf_pending_backlog_when_worker_queue_is_full() {
         let shared = empty_resync_shared_state();

crates/ecstore/src/bucket/replication/replication_resyncer.rs

@@ -18,16 +18,16 @@ use super::replication_config_store::ReplicationConfigStore;
 use super::replication_error_boundary::{Result, is_err_object_not_found, is_err_version_not_found};
 use super::replication_event_sink::{EventArgs, send_event, send_local_event};
 use super::replication_filemeta_boundary::{
-    REPLICATE_EXISTING, REPLICATE_EXISTING_DELETE, ReplicateDecision, ReplicateObjectInfo, ReplicatedInfos, ReplicatedTargetInfo,
-    ReplicationAction, ReplicationStatusType, ReplicationType, VersionPurgeStatusType, get_replication_state,
-    parse_replicate_decision, replication_statuses_map, target_reset_header, version_purge_statuses_map,
+    NULL_VERSION_ID, REPLICATE_EXISTING, REPLICATE_EXISTING_DELETE, ReplicateDecision, ReplicateObjectInfo, ReplicatedInfos,
+    ReplicatedTargetInfo, ReplicationAction, ReplicationStatusType, ReplicationType, VersionPurgeStatusType,
+    get_replication_state, parse_replicate_decision, replication_statuses_map, target_reset_header, version_purge_statuses_map,
 };
 use super::replication_lock_boundary::ReplicationLockTiming;
 use super::replication_logging::{EVENT_RESYNC_CONFIG_LOOKUP_SKIPPED, LOG_COMPONENT_ECSTORE, LOG_SUBSYSTEM_REPLICATION_RESYNC};
 use super::replication_metadata_boundary::ReplicationMetadataStore;
 #[cfg(test)]
 use super::replication_msgp_boundary::ReplicationMsgpCodec;
-use super::replication_object_config::{ReplicationConfig, check_replicate_delete, get_replication_config, must_replicate};
+use super::replication_object_config::{ReplicationConfig, get_replication_config, must_replicate};
 use super::replication_object_decision_boundary::{
     MustReplicateOptions, ReplicationMultipartPartInput, heal_uses_delete_replication_path,
     is_retryable_delete_replication_head_error, is_version_delete_replication, replication_etags_match,
@@ -642,6 +642,21 @@ impl ReplicationResyncer {
         };
 
         let rcfg = ReplicationConfig::new(cfg.clone(), Some(targets));
+        if let Err(err) = rcfg.validate() {
+            error!(
+                event = EVENT_RESYNC_CONFIG_LOOKUP_SKIPPED,
+                component = LOG_COMPONENT_ECSTORE,
+                subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC,
+                bucket = %opts.bucket,
+                arn = %opts.arn,
+                error = %err,
+                reason = "replication_config_invalid",
+                "Replication resync config is invalid"
+            );
+            self.resync_bucket_mark_status(ResyncStatusType::ResyncFailed, opts.clone(), storage.clone())
+                .await;
+            return;
+        }
 
         let target_arns = if let Some(cfg) = cfg {
             cfg.filter_target_arns(&ObjectOpts {
@@ -982,7 +997,36 @@ impl ReplicationResyncer {
             }
             last_checkpoint = None;
 
-            let roi = get_heal_replicate_object_info(&object, &rcfg).await;
+            let roi = match get_heal_replicate_object_info(&object, &rcfg).await {
+                Ok(roi) => roi,
+                Err(err) => {
+                    error!(
+                        event = EVENT_RESYNC_CONFIG_LOOKUP_SKIPPED,
+                        component = LOG_COMPONENT_ECSTORE,
+                        subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC,
+                        bucket = %opts.bucket,
+                        arn = %opts.arn,
+                        object = %object.name,
+                        error = %err,
+                        "Failed to classify object for replication resync"
+                    );
+                    let worker_failed = finish_resync_workers(worker_txs, results_tx, futures, false).await;
+                    if worker_failed {
+                        error!(
+                            event = EVENT_RESYNC_TASK_FAILED,
+                            component = LOG_COMPONENT_ECSTORE,
+                            subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC,
+                            bucket = %opts.bucket,
+                            arn = %opts.arn,
+                            reason = "worker_join_failed_after_classification_error",
+                            "Replication resync worker cleanup observed task failure"
+                        );
+                    }
+                    self.resync_bucket_mark_status(ResyncStatusType::ResyncFailed, opts.clone(), storage.clone())
+                        .await;
+                    return;
+                }
+            };
             if !roi.existing_obj_resync.must_resync() {
                 continue;
             }
@@ -1037,18 +1081,25 @@ impl ReplicationResyncer {
     }
 }
 
-pub async fn get_heal_replicate_object_info(oi: &ObjectInfo, rcfg: &ReplicationConfig) -> ReplicateObjectInfo {
+pub async fn get_heal_replicate_object_info(oi: &ObjectInfo, rcfg: &ReplicationConfig) -> Result<ReplicateObjectInfo> {
     let mut oi = oi.clone();
     let mut user_defined = (*oi.user_defined).clone();
+    let delete_path = heal_uses_delete_replication_path(oi.delete_marker, &oi.version_purge_status);
+    let stored_delete_decision = if delete_path && !oi.replication_decision.is_empty() {
+        Some(parse_replicate_decision(&oi.bucket, &oi.replication_decision)?)
+    } else {
+        None
+    };
+    let has_stored_delete_decision = stored_delete_decision.is_some();
 
     if let Some(rc) = rcfg.config.as_ref()
         && !rc.role.is_empty()
     {
-        if !oi.version_purge_status.is_empty() {
+        if oi.version_purge_status_internal.is_none() && !oi.version_purge_status.is_empty() {
             oi.version_purge_status_internal = Some(format!("{}={};", rc.role, oi.version_purge_status.as_str()));
         }
 
-        if !oi.replication_status.is_empty() {
+        if oi.replication_status_internal.is_none() && !oi.replication_status.is_empty() {
             oi.replication_status_internal = Some(format!("{}={};", rc.role, oi.replication_status.as_str()));
         }
 
@@ -1064,23 +1115,31 @@ pub async fn get_heal_replicate_object_info(oi: &ObjectInfo, rcfg: &ReplicationC
         }
     }
 
-    let dsc = if heal_uses_delete_replication_path(oi.delete_marker, &oi.version_purge_status) {
-        check_replicate_delete(
-            oi.bucket.as_str(),
-            &ObjectToDelete {
-                object_name: oi.name.clone(),
-                version_id: oi.version_id,
-                ..Default::default()
-            },
-            &oi,
-            &ObjectOptions {
-                versioned: ReplicationVersioningStore::prefix_enabled(&oi.bucket, &oi.name).await,
-                version_suspended: ReplicationVersioningStore::prefix_suspended(&oi.bucket, &oi.name).await,
-                ..Default::default()
-            },
-            None,
-        )
-        .await
+    let delete_state = if delete_path && !has_stored_delete_decision {
+        ReplicationVersioningStore::prefix_state(&oi.bucket, &oi.name).await?
+    } else {
+        (false, false)
+    };
+    let dsc = if let Some(decision) = stored_delete_decision {
+        decision
+    } else if delete_path {
+        if !delete_state.0 && !delete_state.1 {
+            ReplicateDecision::default()
+        } else {
+            rcfg.check_delete_for_heal(
+                &ObjectToDelete {
+                    object_name: oi.name.clone(),
+                    version_id: oi.version_id,
+                    ..Default::default()
+                },
+                &oi,
+                &ObjectOptions {
+                    versioned: delete_state.0,
+                    version_suspended: delete_state.1,
+                    ..Default::default()
+                },
+            )
+        }
     } else {
         must_replicate(
             oi.bucket.as_str(),
@@ -1092,12 +1151,16 @@ pub async fn get_heal_replicate_object_info(oi: &ObjectInfo, rcfg: &ReplicationC
 
     let target_statuses = replication_statuses_map(&oi.replication_status_internal.clone().unwrap_or_default());
     let target_purge_statuses = version_purge_statuses_map(&oi.version_purge_status_internal.clone().unwrap_or_default());
-    let existing_obj_resync = rcfg.resync(oi.clone(), dsc.clone(), &target_statuses).await;
+    let existing_obj_resync = if delete_path && !has_stored_delete_decision && !delete_state.0 && !delete_state.1 {
+        Default::default()
+    } else {
+        rcfg.resync(oi.clone(), dsc.clone(), &target_statuses).await
+    };
     let mut replication_state = oi.replication_state();
     replication_state.replicate_decision_str = dsc.to_string();
     let actual_size = oi.get_actual_size().unwrap_or_default();
 
-    ReplicateObjectInfo {
+    Ok(ReplicateObjectInfo {
         name: oi.name.clone(),
         size: oi.size,
         actual_size,
@@ -1122,7 +1185,7 @@ pub async fn get_heal_replicate_object_info(oi: &ObjectInfo, rcfg: &ReplicationC
         user_tags: (*oi.user_tags).clone(),
         checksum: oi.checksum.clone(),
         retry_count: 0,
-    }
+    })
 }
 
 pub(crate) async fn save_resync_status<S: ReplicationObjectIO>(
@@ -1566,7 +1629,7 @@ async fn replicate_delete_marker_purge_to_targets(bucket: &str, dobj: &DeletedOb
             .remove_object(
                 &tgt_client.bucket,
                 &dobj.delete_object.object_name,
-                Some(delete_marker_version_id.to_string()),
+                target_delete_version_id(delete_marker_version_id, true),
                 replication_delete_marker_purge_remove_options(dobj.delete_object.delete_marker_mtime),
             )
             .await;
@@ -1796,6 +1859,14 @@ async fn replicate_force_delete_to_targets<S: ReplicationStorage>(dobj: &Deleted
     }
 }
 
+fn target_delete_version_id(version_id: Uuid, version_purge: bool) -> Option<String> {
+    if version_id.is_nil() {
+        version_purge.then(|| NULL_VERSION_ID.to_string())
+    } else {
+        Some(version_id.to_string())
+    }
+}
+
 async fn replicate_delete_to_target(dobj: &DeletedObjectReplicationInfo, tgt_client: Arc<TargetClient>) -> ReplicatedTargetInfo {
     let version_id = if let Some(version_id) = &dobj.delete_object.delete_marker_version_id {
         version_id.to_owned()
@@ -1835,11 +1906,7 @@ async fn replicate_delete_to_target(dobj: &DeletedObjectReplicationInfo, tgt_cli
         return rinfo;
     }
 
-    let version_id = if version_id.is_nil() {
-        None
-    } else {
-        Some(version_id.to_string())
-    };
+    let version_id = target_delete_version_id(version_id, is_version_purge);
 
     if dobj.delete_object.delete_marker && dobj.delete_object.delete_marker_version_id.is_some() {
         match head_object_with_proxy_stats(
@@ -3088,7 +3155,12 @@ async fn replicate_object_with_multipart<S: ReplicationObjectIO>(ctx: MultipartR
 
 #[cfg(test)]
 mod tests {
+    use super::super::replication_target_boundary::{BucketTarget, BucketTargets};
     use super::*;
+    use s3s::dto::{
+        BucketVersioningStatus, DeleteReplication, DeleteReplicationStatus, Destination, ExcludedPrefix, ReplicationRule,
+        ReplicationRuleStatus, VersioningConfiguration,
+    };
     use std::collections::HashMap;
     use time::OffsetDateTime;
     use uuid::Uuid;
@@ -3530,7 +3602,9 @@ mod tests {
             ..Default::default()
         };
         let rcfg = ReplicationConfig::new(None, None);
-        let roi = get_heal_replicate_object_info(&oi, &rcfg).await;
+        let roi = get_heal_replicate_object_info(&oi, &rcfg)
+            .await
+            .expect("non-delete heal classification should succeed");
 
         assert_eq!(roi.replication_status, ReplicationStatusType::Failed);
         assert_eq!(roi.op_type, ReplicationType::Heal);
@@ -3555,7 +3629,9 @@ mod tests {
         };
         let rcfg = ReplicationConfig::new(None, None);
 
-        let roi = get_heal_replicate_object_info(&oi, &rcfg).await;
+        let roi = get_heal_replicate_object_info(&oi, &rcfg)
+            .await
+            .expect("non-delete heal classification should succeed");
 
         assert!(roi.ssec);
         assert_eq!(roi.checksum, Some(checksum));
@@ -3571,6 +3647,7 @@ mod tests {
             version_purge_status: VersionPurgeStatusType::Pending,
             version_id: Some(Uuid::nil()),
             mod_time: Some(OffsetDateTime::now_utc()),
+            replication_decision: format!("{role}=true;false;{role};"),
             ..Default::default()
         };
         let rcfg = ReplicationConfig::new(
@@ -3580,13 +3657,176 @@ mod tests {
             }),
             None,
         );
-        let roi = get_heal_replicate_object_info(&oi, &rcfg).await;
+        let roi = get_heal_replicate_object_info(&oi, &rcfg)
+            .await
+            .expect("stored purge admission should classify without a live versioning lookup");
 
         assert_eq!(roi.replication_status_internal, None);
         assert_eq!(roi.version_purge_status_internal.as_deref(), Some(format!("{role}=PENDING;").as_str()));
         assert_eq!(roi.target_purge_statuses.get(role), Some(&VersionPurgeStatusType::Pending));
     }
 
+    #[tokio::test]
+    async fn heal_pending_purge_reads_one_versioning_generation() {
+        let bucket = format!("heal-versioning-snapshot-{}", Uuid::new_v4());
+        let object = "archive/object";
+        let arn = "arn:rustfs:replication:us-east-1:target:bucket";
+        ReplicationVersioningStore::install_prefix_state_test_config(
+            &bucket,
+            VersioningConfiguration {
+                status: Some(BucketVersioningStatus::from_static(BucketVersioningStatus::ENABLED)),
+                excluded_prefixes: Some(vec![ExcludedPrefix {
+                    prefix: Some("archive/".to_string()),
+                }]),
+                ..Default::default()
+            },
+        );
+        let rcfg = ReplicationConfig::new(
+            Some(ReplicationConfiguration {
+                role: String::new(),
+                rules: vec![ReplicationRule {
+                    delete_marker_replication: None,
+                    delete_replication: Some(DeleteReplication {
+                        status: DeleteReplicationStatus::from_static(DeleteReplicationStatus::ENABLED),
+                    }),
+                    destination: Destination {
+                        bucket: arn.to_string(),
+                        ..Default::default()
+                    },
+                    existing_object_replication: None,
+                    filter: None,
+                    id: Some("delete".to_string()),
+                    prefix: Some(String::new()),
+                    priority: Some(1),
+                    source_selection_criteria: None,
+                    status: ReplicationRuleStatus::from_static(ReplicationRuleStatus::ENABLED),
+                }],
+            }),
+            Some(BucketTargets {
+                targets: vec![BucketTarget {
+                    arn: arn.to_string(),
+                    ..Default::default()
+                }],
+            }),
+        );
+        let oi = ObjectInfo {
+            bucket,
+            name: object.to_string(),
+            version_id: Some(Uuid::nil()),
+            version_purge_status: VersionPurgeStatusType::Pending,
+            ..Default::default()
+        };
+
+        let roi = get_heal_replicate_object_info(&oi, &rcfg)
+            .await
+            .expect("pending null purge classification should succeed");
+
+        assert!(roi.dsc.targets_map.get(arn).is_some_and(|target| target.replicate));
+        assert!(
+            roi.existing_obj_resync
+                .targets
+                .get(arn)
+                .is_some_and(|target| target.replicate)
+        );
+    }
+
+    #[tokio::test]
+    async fn heal_pending_purge_preserves_the_persisted_admission_decision() {
+        let admitted_arn = "arn:rustfs:replication:us-east-1:target:admitted";
+        let current_role = "arn:rustfs:replication:us-east-1:target:current";
+        let rcfg = ReplicationConfig::new(
+            Some(ReplicationConfiguration {
+                role: current_role.to_string(),
+                rules: vec![ReplicationRule {
+                    delete_marker_replication: None,
+                    delete_replication: Some(DeleteReplication {
+                        status: DeleteReplicationStatus::from_static(DeleteReplicationStatus::DISABLED),
+                    }),
+                    destination: Destination {
+                        bucket: current_role.to_string(),
+                        ..Default::default()
+                    },
+                    existing_object_replication: None,
+                    filter: None,
+                    id: Some("delete".to_string()),
+                    prefix: Some(String::new()),
+                    priority: Some(1),
+                    source_selection_criteria: None,
+                    status: ReplicationRuleStatus::from_static(ReplicationRuleStatus::ENABLED),
+                }],
+            }),
+            Some(BucketTargets {
+                targets: vec![
+                    BucketTarget {
+                        arn: admitted_arn.to_string(),
+                        ..Default::default()
+                    },
+                    BucketTarget {
+                        arn: current_role.to_string(),
+                        ..Default::default()
+                    },
+                ],
+            }),
+        );
+        let oi = ObjectInfo {
+            bucket: "heal-persisted-delete-decision".to_string(),
+            name: "object".to_string(),
+            version_id: Some(Uuid::new_v4()),
+            version_purge_status: VersionPurgeStatusType::Pending,
+            version_purge_status_internal: Some(format!("{admitted_arn}=PENDING;")),
+            replication_decision: format!("{admitted_arn}=true;false;{admitted_arn};"),
+            ..Default::default()
+        };
+
+        let roi = get_heal_replicate_object_info(&oi, &rcfg)
+            .await
+            .expect("persisted delete admission should survive live rule disablement");
+
+        assert_eq!(
+            roi.version_purge_status_internal.as_deref(),
+            Some(format!("{admitted_arn}=PENDING;").as_str())
+        );
+        assert!(roi.dsc.targets_map.get(admitted_arn).is_some_and(|target| target.replicate));
+        assert!(!roi.dsc.targets_map.contains_key(current_role));
+        assert!(
+            roi.existing_obj_resync
+                .targets
+                .get(admitted_arn)
+                .is_some_and(|target| target.replicate)
+        );
+        assert!(!roi.existing_obj_resync.targets.contains_key(current_role));
+    }
+
+    #[tokio::test]
+    async fn heal_rejects_semantically_invalid_replication_config() {
+        let rcfg = ReplicationConfig::new(
+            Some(ReplicationConfiguration {
+                role: String::new(),
+                rules: vec![ReplicationRule {
+                    delete_marker_replication: None,
+                    delete_replication: None,
+                    destination: Destination {
+                        bucket: "arn:rustfs:replication:us-east-1:target:bucket".to_string(),
+                        ..Default::default()
+                    },
+                    existing_object_replication: None,
+                    filter: None,
+                    id: Some("invalid".to_string()),
+                    prefix: Some(String::new()),
+                    priority: Some(1),
+                    source_selection_criteria: None,
+                    status: ReplicationRuleStatus::from_static("Enabld"),
+                }],
+            }),
+            Some(BucketTargets::default()),
+        );
+        let err = rcfg
+            .validate()
+            .expect_err("invalid string-backed statuses must fail before heal classification loop");
+
+        assert!(err.to_string().contains("Rule.Status"));
+    }
+
     #[tokio::test]
     async fn test_cancel_marks_only_matching_bucket_target_token() {
         let resyncer = ReplicationResyncer::new().await;
@@ -3753,4 +3993,13 @@ mod tests {
         assert_eq!(resync_status_duration(ResyncStatusType::ResyncStarted, Some(start), end), None);
         assert_eq!(resync_status_duration(ResyncStatusType::ResyncFailed, None, end), None);
     }
+
+    #[test]
+    fn target_delete_version_id_preserves_explicit_null_purges() {
+        let version_id = Uuid::new_v4();
+
+        assert_eq!(target_delete_version_id(version_id, true), Some(version_id.to_string()));
+        assert_eq!(target_delete_version_id(Uuid::nil(), true).as_deref(), Some(NULL_VERSION_ID));
+        assert_eq!(target_delete_version_id(Uuid::nil(), false), None);
+    }
 }

crates/ecstore/src/bucket/replication/replication_target_boundary.rs

@@ -35,6 +35,8 @@ use time::format_description::well_known::Rfc3339;
 pub(crate) use crate::bucket::bucket_target_sys::{
     AdvancedPutOptions, PutObjectOptions, PutObjectPartOptions, RemoveObjectOptions, TargetClient,
 };
+#[cfg(test)]
+pub(crate) use crate::bucket::target::BucketTarget;
 pub(crate) use crate::bucket::target::BucketTargets;
 
 use super::replication_config_store::ReplicationConfigStore;

crates/ecstore/src/bucket/replication/replication_versioning_boundary.rs

@@ -12,11 +12,28 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use crate::bucket::versioning_sys::BucketVersioningSys;
+use super::replication_error_boundary::Result;
+use crate::bucket::{versioning::VersioningApi as _, versioning_sys::BucketVersioningSys};
+#[cfg(test)]
+use s3s::dto::VersioningConfiguration;
+#[cfg(test)]
+use std::{collections::HashMap, sync::LazyLock, sync::Mutex};
+
+#[cfg(test)]
+static PREFIX_STATE_TEST_CONFIGS: LazyLock<Mutex<HashMap<String, VersioningConfiguration>>> =
+    LazyLock::new(|| Mutex::new(HashMap::new()));
 
 pub(crate) struct ReplicationVersioningStore;
 
 impl ReplicationVersioningStore {
+    #[cfg(test)]
+    pub(crate) fn install_prefix_state_test_config(bucket: &str, config: VersioningConfiguration) {
+        PREFIX_STATE_TEST_CONFIGS
+            .lock()
+            .expect("replication versioning test config lock should not be poisoned")
+            .insert(bucket.to_string(), config);
+    }
+
     pub(crate) async fn prefix_enabled(bucket: &str, prefix: &str) -> bool {
         BucketVersioningSys::prefix_enabled(bucket, prefix).await
     }
@@ -24,4 +41,18 @@ impl ReplicationVersioningStore {
     pub(crate) async fn prefix_suspended(bucket: &str, prefix: &str) -> bool {
         BucketVersioningSys::prefix_suspended(bucket, prefix).await
     }
+
+    pub(crate) async fn prefix_state(bucket: &str, prefix: &str) -> Result<(bool, bool)> {
+        #[cfg(test)]
+        if let Some(config) = PREFIX_STATE_TEST_CONFIGS
+            .lock()
+            .expect("replication versioning test config lock should not be poisoned")
+            .remove(bucket)
+        {
+            return Ok((config.prefix_enabled(prefix), config.prefix_suspended(prefix)));
+        }
+
+        let config = BucketVersioningSys::get(bucket).await?;
+        Ok((config.prefix_enabled(prefix), config.prefix_suspended(prefix)))
+    }
 }

crates/ecstore/src/bucket/versioning/mod.rs

@@ -19,6 +19,9 @@ pub trait VersioningApi {
     fn enabled(&self) -> bool;
     fn prefix_enabled(&self, prefix: &str) -> bool;
     fn prefix_suspended(&self, prefix: &str) -> bool;
+    fn delete_state(&self, prefix: &str) -> (bool, bool) {
+        (self.prefix_enabled(prefix), self.suspended())
+    }
     fn versioned(&self, prefix: &str) -> bool;
     fn suspended(&self) -> bool;
 }
@@ -92,3 +95,60 @@ impl VersioningApi for VersioningConfiguration {
         self.status == Some(BucketVersioningStatus::from_static(BucketVersioningStatus::SUSPENDED))
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use s3s::dto::{BucketVersioningStatus, ExcludedPrefix};
+
+    use super::*;
+
+    struct LegacyVersioning;
+
+    impl VersioningApi for LegacyVersioning {
+        fn enabled(&self) -> bool {
+            false
+        }
+
+        fn prefix_enabled(&self, _prefix: &str) -> bool {
+            false
+        }
+
+        fn prefix_suspended(&self, _prefix: &str) -> bool {
+            false
+        }
+
+        fn versioned(&self, _prefix: &str) -> bool {
+            false
+        }
+
+        fn suspended(&self) -> bool {
+            false
+        }
+    }
+
+    #[test]
+    fn delete_state_has_a_backward_compatible_default() {
+        assert_eq!(LegacyVersioning.delete_state("object"), (false, false));
+    }
+
+    #[test]
+    fn delete_state_treats_excluded_prefixes_as_unversioned() {
+        let config = VersioningConfiguration {
+            status: Some(BucketVersioningStatus::from_static(BucketVersioningStatus::ENABLED)),
+            excluded_prefixes: Some(vec![ExcludedPrefix {
+                prefix: Some("archive/".to_string()),
+            }]),
+            ..Default::default()
+        };
+
+        assert_eq!(config.delete_state("archive/object"), (false, false));
+        assert!(config.prefix_suspended("archive/object"));
+        assert_eq!(config.delete_state("live/object"), (true, false));
+
+        let suspended = VersioningConfiguration {
+            status: Some(BucketVersioningStatus::from_static(BucketVersioningStatus::SUSPENDED)),
+            ..Default::default()
+        };
+        assert_eq!(suspended.delete_state("archive/object"), (false, true));
+    }
+}

crates/ecstore/src/core/sets.rs

@@ -615,7 +615,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for Sets {
         self.get_disks_by_key(object).delete_object(bucket, object, opts).await
     }
 
-    #[tracing::instrument(skip(self))]
+    #[tracing::instrument(skip(self, objects, opts))]
     async fn delete_objects(
         &self,
         bucket: &str,

crates/ecstore/src/object_api/mod.rs

@@ -17,8 +17,8 @@
 
 use crate::bucket::metadata_sys::get_versioning_config;
 use crate::bucket::replication::{
-    ReplicateDecision, ReplicationState, ReplicationStatusType, VersionPurgeStatusType, replication_status_from_filemeta,
-    replication_statuses_map, version_purge_status_from_filemeta, version_purge_statuses_map,
+    DeleteReplicationConfigSnapshot, ReplicateDecision, ReplicationState, ReplicationStatusType, VersionPurgeStatusType,
+    replication_status_from_filemeta, replication_statuses_map, version_purge_status_from_filemeta, version_purge_statuses_map,
 };
 use crate::bucket::versioning::VersioningApi as _;
 use crate::config::storageclass;

crates/ecstore/src/object_api/types.rs

@@ -20,6 +20,47 @@ use crate::storage_api_contracts::{
     },
 };
 
+#[derive(Clone, Default)]
+pub struct DeleteLockFence {
+    signals: Arc<Vec<Arc<rustfs_lock::distributed_lock::LockLostSignal>>>,
+    #[cfg(test)]
+    forced_lost: bool,
+}
+
+impl Debug for DeleteLockFence {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("DeleteLockFence")
+            .field("signal_count", &self.signals.len())
+            .finish()
+    }
+}
+
+impl DeleteLockFence {
+    pub(crate) fn new(signals: Vec<Arc<rustfs_lock::distributed_lock::LockLostSignal>>) -> Self {
+        Self {
+            signals: Arc::new(signals),
+            #[cfg(test)]
+            forced_lost: false,
+        }
+    }
+
+    pub(crate) fn is_lock_lost(&self) -> bool {
+        #[cfg(test)]
+        if self.forced_lost {
+            return true;
+        }
+        self.signals.iter().any(|signal| signal.is_lost())
+    }
+
+    #[cfg(test)]
+    pub(crate) fn lost_for_test() -> Self {
+        Self {
+            signals: Arc::default(),
+            forced_lost: true,
+        }
+    }
+}
+
 #[derive(Debug, Default, Clone)]
 pub struct ObjectOptions {
     // Use the maximum parity (N/2), used when saving server configuration files
@@ -54,8 +95,11 @@ pub struct ObjectOptions {
     pub http_preconditions: Option<HTTPPreconditions>,
 
     pub delete_replication: Option<ReplicationState>,
+    pub delete_replication_config_snapshot: Option<Arc<DeleteReplicationConfigSnapshot>>,
+    pub delete_lock_fence: Option<DeleteLockFence>,
     pub replication_request: bool,
     pub delete_marker: bool,
+    pub synthetic_version_id: bool,
 
     pub transition: TransitionOptions,
     pub expiration: ExpirationOptions,

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

@@ -35,14 +35,17 @@ use crate::bucket::lifecycle::{
         save_transition_transaction_record,
     },
 };
+use crate::bucket::replication::{DeleteReplicationConfigSnapshot, VersionPurgeStatusType, version_purge_status_to_filemeta};
 use crate::diagnostics::get::GetObjectFailureReason;
 use crate::disk::OldCurrentSize;
 use crate::error::is_err_invalid_upload_id;
+use crate::object_api::DeleteLockFence;
 use crate::object_api::{GetObjectBodySource, get_object_body_cache_hook_suppressed};
 use crate::services::tier::tier::{TierConfigMgr, TierOperationLease};
 use crate::store::ECStore;
 use futures::FutureExt as _;
 use http::HeaderValue;
+use rustfs_utils::path::decode_dir_object;
 use std::future::Future;
 
 fn erasure_from_file_info(fi: &FileInfo, uses_legacy: bool) -> Result<coding::Erasure> {
@@ -3111,25 +3114,24 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
         quorum_result
     }
 
-    #[tracing::instrument(skip(self))]
+    #[tracing::instrument(skip(self, objects, opts))]
     async fn delete_objects(
         &self,
         bucket: &str,
         objects: Vec<ObjectToDelete>,
         opts: ObjectOptions,
     ) -> (Vec<DeletedObject>, Vec<Option<Error>>) {
+        let mut del_objects = vec![DeletedObject::default(); objects.len()];
+        let delete_config_snapshot = opts
+            .delete_replication_config_snapshot
+            .clone()
+            .unwrap_or_else(|| Arc::new(DeleteReplicationConfigSnapshot::default()));
+
         for object in &objects {
             self.invalidate_get_object_metadata_cache(bucket, &object.object_name).await;
         }
 
-        // Default return value
-        let mut del_objects = vec![DeletedObject::default(); objects.len()];
-
-        let mut del_errs = Vec::with_capacity(objects.len());
-
-        for _ in 0..objects.len() {
-            del_errs.push(None)
-        }
+        let mut del_errs = (0..objects.len()).map(|_| None).collect::<Vec<_>>();
 
         // Acquire locks in batch mode (best effort, matching previous behavior)
         let mut batch = rustfs_lock::BatchLockRequest::new(self.locker_owner.as_str()).with_all_or_nothing(false);
@@ -3195,14 +3197,8 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
             }
         }
 
-        let ver_cfg = BucketVersioningSys::get_in(&self.ctx, bucket).await.unwrap_or_default();
-
-        // backlog#929 (HP-8): the per-object stat below exists solely to feed
-        // check_object_lock_delete (#4297). Resolve the bucket lock
-        // configuration once (in-memory cache) and skip the whole stat fanout
-        // for buckets without Object Lock; unknown metadata fails closed and
-        // keeps the stat, so the #4297 protection is preserved verbatim for
-        // every object-lock-enabled bucket.
+        // Resolve Object Lock once; replication still reads each matching
+        // object's tags below while the batch write lock is held.
         let object_lock_checks_required =
             object_lock_delete_check_required(metadata_sys::get_in(&self.ctx, bucket).await.ok().as_deref());
 
@@ -3213,31 +3209,75 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
                 continue;
             }
 
+            let replication_object_name = decode_dir_object(&dobj.object_name);
             let explicit_null_version = is_explicit_null_version(dobj.version_id);
             let version_id = delete_file_info_version_id(dobj.version_id);
-            if object_lock_checks_required {
-                let check_opts = ObjectOptions {
-                    version_id: version_id.map(|version_id| version_id.to_string()),
-                    versioned: ver_cfg.prefix_enabled(dobj.object_name.as_str()),
-                    version_suspended: ver_cfg.suspended(),
-                    object_lock_delete: opts.object_lock_delete.clone(),
-                    no_lock: true,
-                    ..Default::default()
-                };
+            let (versioned, version_suspended) = delete_config_snapshot
+                .versioning_config()
+                .delete_state(replication_object_name.as_str());
+            let check_opts = ObjectOptions {
+                version_id: version_id.map(|version_id| version_id.to_string()),
+                versioned,
+                version_suspended,
+                object_lock_delete: opts.object_lock_delete.clone(),
+                no_lock: true,
+                ..Default::default()
+            };
+            let replicate_delete = delete_config_snapshot.has_active_rule(&replication_object_name);
+            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_checks_required || replication_needs_source {
                 let (goi, _write_quorum, gerr) = self.get_object_info_and_quorum(bucket, &dobj.object_name, &check_opts).await;
-                if gerr.is_none()
-                    && let Err(err) = check_object_lock_delete(bucket, &dobj.object_name, &goi, &check_opts).await
-                {
-                    del_errs[i] = Some(err);
-                    continue;
-                }
+                (goi, gerr)
+            } else {
+                (ObjectInfo::default(), None)
+            };
+            let source_missing = gerr
+                .as_ref()
+                .is_some_and(|err| is_err_object_not_found(err) || is_err_version_not_found(err));
+            if let Some(err) = gerr.as_ref()
+                && !source_missing
+            {
+                del_errs[i] = Some(err.clone());
+                continue;
+            }
+            if object_lock_checks_required
+                && !source_missing
+                && let Err(err) = check_object_lock_delete(bucket, &dobj.object_name, &goi, &check_opts).await
+            {
+                del_errs[i] = Some(err);
+                continue;
             }
 
+            let mut admitted = dobj.clone();
+            admitted.object_name = replication_object_name;
+            if admitted.synthetic_version_id {
+                admitted.version_id = None;
+            }
+            if replicate_delete {
+                let dsc = ReplicationObjectBridge::check_delete_with_snapshot(
+                    &admitted,
+                    &goi,
+                    &check_opts,
+                    source_missing,
+                    &delete_config_snapshot,
+                );
+                if dsc.replicate_any() {
+                    if admitted.version_id.is_some() {
+                        admitted.version_purge_status = Some(version_purge_status_to_filemeta(VersionPurgeStatusType::Pending));
+                        admitted.version_purge_statuses = dsc.pending_status();
+                    } else {
+                        admitted.delete_marker_replication_status = dsc.pending_status();
+                    }
+                    admitted.replicate_decision_str = Some(dsc.to_string());
+                }
+            }
             let mut vr = FileInfo {
                 name: dobj.object_name.clone(),
                 version_id,
                 idx: i,
-                replication_state_internal: Some(dobj.replication_state()),
+                replication_state_internal: Some(admitted.replication_state()),
                 ..Default::default()
             };
 
@@ -3247,14 +3287,11 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
             // del_objects[i].object_name.clone_from(&vr.name);
             // del_objects[i].version_id = vr.version_id.map(|v| v.to_string());
 
-            if dobj.version_id.is_none() {
-                let (suspended, versioned) = (ver_cfg.suspended(), ver_cfg.prefix_enabled(dobj.object_name.as_str()));
-                if suspended || versioned {
-                    vr.mod_time = Some(OffsetDateTime::now_utc());
-                    vr.deleted = true;
-                    if versioned {
-                        vr.version_id = Some(Uuid::new_v4());
-                    }
+            if dobj.version_id.is_none() && (version_suspended || versioned) {
+                vr.mod_time = Some(OffsetDateTime::now_utc());
+                vr.deleted = true;
+                if versioned {
+                    vr.version_id = Some(Uuid::new_v4());
                 }
             }
 
@@ -3312,6 +3349,24 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
             vers.push(fi_vers);
         }
 
+        if opts.delete_lock_fence.as_ref().is_some_and(DeleteLockFence::is_lock_lost) {
+            if dist_erasure {
+                self.release_dist_delete_object_locks_batch(dist_batch_lock_ids).await;
+            }
+            for (index, object) in objects.iter().enumerate() {
+                if del_errs[index].is_none() {
+                    del_errs[index] = Some(Error::NamespaceLockQuorumUnavailable {
+                        mode: "delete_objects_commit",
+                        bucket: bucket.to_string(),
+                        object: decode_dir_object(&object.object_name),
+                        required: 1,
+                        achieved: 0,
+                    });
+                }
+            }
+            return (del_objects, del_errs);
+        }
+
         let rollback_dir = Uuid::new_v4();
 
         let disks = self.disks.read().await;
@@ -3490,6 +3545,16 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
 
     #[tracing::instrument(skip(self))]
     async fn delete_object(&self, bucket: &str, object: &str, mut opts: ObjectOptions) -> Result<ObjectInfo> {
+        let preserve_delete_replication_state = should_preserve_delete_replication_state(&opts);
+        let delete_config_snapshot = if opts.delete_prefix || opts.transition.expire_restored || preserve_delete_replication_state
+        {
+            None
+        } else if let Some(snapshot) = opts.delete_replication_config_snapshot.clone() {
+            Some(snapshot)
+        } else {
+            Some(Arc::new(DeleteReplicationConfigSnapshot::default()))
+        };
+
         self.invalidate_get_object_metadata_cache(bucket, object).await;
 
         // Guard lock for single object delete
@@ -3580,18 +3645,20 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
         }
 
         let otd = ObjectToDelete {
-            object_name: object.to_string(),
-            version_id: opts
-                .version_id
-                .clone()
-                .map(|v| Uuid::parse_str(v.as_str()).ok().unwrap_or_default()),
+            object_name: decode_dir_object(object),
+            version_id: if opts.synthetic_version_id {
+                None
+            } else {
+                opts.version_id.as_deref().map(Uuid::parse_str).transpose()?
+            },
+            synthetic_version_id: opts.synthetic_version_id,
             ..Default::default()
         };
 
-        let dsc = if should_preserve_delete_replication_state(&opts) {
-            ReplicateDecision::default()
+        let dsc = if let Some(snapshot) = delete_config_snapshot {
+            ReplicationObjectBridge::check_delete_with_snapshot(&otd, &goi, &opts, gerr.is_some(), &snapshot)
         } else {
-            ReplicationObjectBridge::check_delete(bucket, &otd, &goi, &opts, gerr.map(|e| e.to_string())).await
+            ReplicateDecision::default()
         };
 
         if dsc.replicate_any() {
@@ -3649,6 +3716,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
             self.record_capacity_scope_if_needed(opts.capacity_scope_token, &disks);
 
             let mut oi = ObjectInfo::from_file_info(&fi, bucket, object, opts.versioned || opts.version_suspended);
+            oi.user_tags = Arc::clone(&goi.user_tags);
             oi.replication_decision = goi.replication_decision;
             self.invalidate_get_object_metadata_cache(bucket, object).await;
             return Ok(oi);
@@ -3680,6 +3748,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
 
         let mut obj_info = ObjectInfo::from_file_info(&dfi, bucket, object, opts.versioned || opts.version_suspended);
         obj_info.size = goi.size;
+        obj_info.user_tags = Arc::clone(&goi.user_tags);
         self.invalidate_get_object_metadata_cache(bucket, object).await;
         Ok(obj_info)
     }
@@ -7801,6 +7870,56 @@ mod object_tagging_namespace_lock_tests {
         );
     }
 
+    #[tokio::test]
+    async fn version_delete_returns_tags_read_under_the_delete_lock() {
+        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
+        let bucket = "delete-locked-tags";
+        let object = "object";
+        for disk in &disk_stores {
+            disk.make_volume(bucket).await.expect("bucket volume should be created");
+        }
+        let mut reader = PutObjReader::from_vec(b"body".to_vec());
+        let uploaded = set_disks
+            .put_object(
+                bucket,
+                object,
+                &mut reader,
+                &ObjectOptions {
+                    versioned: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("versioned object should be written");
+        let version_id = uploaded.version_id.expect("versioned PUT should return an ID").to_string();
+        let version_opts = ObjectOptions {
+            versioned: true,
+            version_id: Some(version_id),
+            delete_replication_config_snapshot: Some(Arc::new(DeleteReplicationConfigSnapshot::default())),
+            ..Default::default()
+        };
+        set_disks
+            .put_object_tags(bucket, object, "generation=stale", &version_opts)
+            .await
+            .expect("initial tags should be written");
+        let advisory = set_disks
+            .get_object_info(bucket, object, &version_opts)
+            .await
+            .expect("advisory pre-read should succeed");
+        set_disks
+            .put_object_tags(bucket, object, "generation=locked", &version_opts)
+            .await
+            .expect("concurrent tag update should commit before delete");
+
+        let deleted = set_disks
+            .delete_object(bucket, object, version_opts)
+            .await
+            .expect("version delete should succeed");
+
+        assert_eq!(advisory.user_tags.as_str(), "generation=stale");
+        assert_eq!(deleted.user_tags.as_str(), "generation=locked");
+    }
+
     #[tokio::test]
     #[serial_test::serial]
     async fn tagging_serializes_with_put_and_delete_for_versioned_and_unversioned_objects() {
@@ -7815,6 +7934,7 @@ mod object_tagging_namespace_lock_tests {
 
                 let object_opts = ObjectOptions {
                     versioned,
+                    delete_replication_config_snapshot: Some(Arc::new(DeleteReplicationConfigSnapshot::default())),
                     ..Default::default()
                 };
                 let original_body = b"original body".to_vec();
@@ -8014,6 +8134,276 @@ mod delete_objects_lock_gating_tests {
         }
     }
 
+    #[tokio::test]
+    async fn delete_objects_derives_per_object_versioning_from_the_request_snapshot() {
+        use s3s::dto::{BucketVersioningStatus, ExcludedPrefix, VersioningConfiguration};
+
+        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
+        let bucket = "batch-versioning-snapshot-bucket";
+        for disk in &disk_stores {
+            disk.make_volume(bucket).await.expect("bucket volume should be created");
+        }
+
+        put_plain_object(&set_disks, bucket, "marker-object").await;
+        put_plain_object(&set_disks, bucket, "archive/unversioned-object").await;
+        let snapshot = Arc::new(DeleteReplicationConfigSnapshot::from_configs_for_test(
+            VersioningConfiguration {
+                status: Some(BucketVersioningStatus::from_static(BucketVersioningStatus::ENABLED)),
+                excluded_prefixes: Some(vec![ExcludedPrefix {
+                    prefix: Some("archive/".to_string()),
+                }]),
+                ..Default::default()
+            },
+            None,
+        ));
+        let objects = vec![
+            ObjectToDelete {
+                object_name: "marker-object".to_string(),
+                ..Default::default()
+            },
+            ObjectToDelete {
+                object_name: "archive/unversioned-object".to_string(),
+                ..Default::default()
+            },
+        ];
+
+        let (deleted, errors) = set_disks
+            .delete_objects(
+                bucket,
+                objects,
+                ObjectOptions {
+                    versioned: true,
+                    delete_replication_config_snapshot: Some(snapshot),
+                    ..Default::default()
+                },
+            )
+            .await;
+
+        assert!(
+            errors.iter().all(Option::is_none),
+            "snapshot-backed batch delete should succeed: {errors:?}"
+        );
+        assert!(deleted[0].delete_marker);
+        assert!(deleted[0].delete_marker_version_id.is_some());
+        assert!(!deleted[1].delete_marker);
+        set_disks
+            .get_object_info(bucket, "archive/unversioned-object", &ObjectOptions::default())
+            .await
+            .expect_err("the snapshot-excluded object should be removed without a delete marker");
+    }
+
+    #[tokio::test]
+    async fn batch_version_delete_uses_tags_read_under_the_delete_lock() {
+        use rustfs_utils::http::headers::AMZ_OBJECT_TAGGING;
+        use s3s::dto::{
+            BucketVersioningStatus, DeleteReplication, DeleteReplicationStatus, Destination, ReplicationConfiguration,
+            ReplicationRule, ReplicationRuleFilter, ReplicationRuleStatus, Tag, VersioningConfiguration,
+        };
+
+        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
+        let bucket = "batch-delete-locked-tags";
+        let object = "object";
+        let arn = "arn:rustfs:replication:us-east-1:target:bucket";
+        for disk in &disk_stores {
+            disk.make_volume(bucket).await.expect("bucket volume should be created");
+        }
+        let mut reader = PutObjReader::from_vec(b"body".to_vec());
+        let uploaded = set_disks
+            .put_object(
+                bucket,
+                object,
+                &mut reader,
+                &ObjectOptions {
+                    versioned: true,
+                    user_defined: HashMap::from([(AMZ_OBJECT_TAGGING.to_string(), "generation=stale".to_string())]),
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("versioned object should be written");
+        let version_id = uploaded.version_id.expect("versioned PUT should return an ID");
+        let version_opts = ObjectOptions {
+            versioned: true,
+            version_id: Some(version_id.to_string()),
+            ..Default::default()
+        };
+        set_disks
+            .put_object_tags(bucket, object, "generation=locked", &version_opts)
+            .await
+            .expect("tag update should commit before the batch delete");
+
+        let snapshot = Arc::new(DeleteReplicationConfigSnapshot::from_configs_for_test(
+            VersioningConfiguration {
+                status: Some(BucketVersioningStatus::from_static(BucketVersioningStatus::ENABLED)),
+                ..Default::default()
+            },
+            Some(ReplicationConfiguration {
+                role: String::new(),
+                rules: vec![ReplicationRule {
+                    delete_marker_replication: None,
+                    delete_replication: Some(DeleteReplication {
+                        status: DeleteReplicationStatus::from_static(DeleteReplicationStatus::ENABLED),
+                    }),
+                    destination: Destination {
+                        bucket: arn.to_string(),
+                        ..Default::default()
+                    },
+                    existing_object_replication: None,
+                    filter: Some(ReplicationRuleFilter {
+                        tag: Some(Tag {
+                            key: Some("generation".to_string()),
+                            value: Some("locked".to_string()),
+                        }),
+                        ..Default::default()
+                    }),
+                    id: Some("delete".to_string()),
+                    prefix: Some(String::new()),
+                    priority: Some(1),
+                    source_selection_criteria: None,
+                    status: ReplicationRuleStatus::from_static(ReplicationRuleStatus::ENABLED),
+                }],
+            }),
+        ));
+        let (deleted, errors) = set_disks
+            .delete_objects(
+                bucket,
+                vec![ObjectToDelete {
+                    object_name: object.to_string(),
+                    version_id: Some(version_id),
+                    ..Default::default()
+                }],
+                ObjectOptions {
+                    versioned: true,
+                    delete_replication_config_snapshot: Some(snapshot),
+                    ..Default::default()
+                },
+            )
+            .await;
+
+        assert!(errors.iter().all(Option::is_none), "batch version delete should succeed: {errors:?}");
+        let state = deleted[0]
+            .replication_state
+            .as_ref()
+            .expect("locked decision should be persisted");
+        assert_eq!(
+            state.version_purge_status_internal.as_deref(),
+            Some("arn:rustfs:replication:us-east-1:target:bucket=PENDING;")
+        );
+        assert!(state.replicate_decision_str.contains(arn));
+    }
+
+    #[tokio::test]
+    async fn synthetic_directory_delete_uses_decoded_prefix_and_marker_switch() {
+        use s3s::dto::{
+            BucketVersioningStatus, DeleteMarkerReplication, DeleteMarkerReplicationStatus, DeleteReplication,
+            DeleteReplicationStatus, Destination, ReplicationConfiguration, ReplicationRule, ReplicationRuleFilter,
+            ReplicationRuleStatus, VersioningConfiguration,
+        };
+
+        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
+        let bucket = "batch-directory-replication";
+        let object = encode_dir_object("photos/");
+        let arn = "arn:rustfs:replication:us-east-1:target:bucket";
+        for disk in &disk_stores {
+            disk.make_volume(bucket).await.expect("bucket volume should be created");
+        }
+        put_plain_object(&set_disks, bucket, &object).await;
+
+        let snapshot = Arc::new(DeleteReplicationConfigSnapshot::from_configs_for_test(
+            VersioningConfiguration {
+                status: Some(BucketVersioningStatus::from_static(BucketVersioningStatus::ENABLED)),
+                ..Default::default()
+            },
+            Some(ReplicationConfiguration {
+                role: String::new(),
+                rules: vec![ReplicationRule {
+                    delete_marker_replication: Some(DeleteMarkerReplication {
+                        status: Some(DeleteMarkerReplicationStatus::from_static(DeleteMarkerReplicationStatus::ENABLED)),
+                    }),
+                    delete_replication: Some(DeleteReplication {
+                        status: DeleteReplicationStatus::from_static(DeleteReplicationStatus::DISABLED),
+                    }),
+                    destination: Destination {
+                        bucket: arn.to_string(),
+                        ..Default::default()
+                    },
+                    existing_object_replication: None,
+                    filter: Some(ReplicationRuleFilter {
+                        prefix: Some("photos/".to_string()),
+                        ..Default::default()
+                    }),
+                    id: Some("directory-marker".to_string()),
+                    prefix: None,
+                    priority: Some(1),
+                    source_selection_criteria: None,
+                    status: ReplicationRuleStatus::from_static(ReplicationRuleStatus::ENABLED),
+                }],
+            }),
+        ));
+
+        let (deleted, errors) = set_disks
+            .delete_objects(
+                bucket,
+                vec![ObjectToDelete {
+                    object_name: object,
+                    version_id: Some(Uuid::nil()),
+                    synthetic_version_id: true,
+                    ..Default::default()
+                }],
+                ObjectOptions {
+                    versioned: true,
+                    delete_replication_config_snapshot: Some(snapshot),
+                    ..Default::default()
+                },
+            )
+            .await;
+
+        assert!(errors[0].is_none(), "directory delete should succeed: {:?}", errors[0]);
+        let state = deleted[0]
+            .replication_state
+            .as_ref()
+            .expect("marker replication decision should be persisted");
+        assert_eq!(state.replication_status_internal.as_deref(), Some(format!("{arn}=PENDING;").as_str()));
+        assert!(state.version_purge_status_internal.is_none());
+    }
+
+    #[tokio::test]
+    async fn delete_objects_aborts_before_disk_mutation_after_outer_lock_loss() {
+        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
+        let bucket = "batch-lost-outer-lock";
+        let object = "object";
+        for disk in &disk_stores {
+            disk.make_volume(bucket).await.expect("bucket volume should be created");
+        }
+        put_plain_object(&set_disks, bucket, object).await;
+
+        let (_deleted, errors) = set_disks
+            .delete_objects(
+                bucket,
+                vec![ObjectToDelete {
+                    object_name: object.to_string(),
+                    ..Default::default()
+                }],
+                ObjectOptions {
+                    no_lock: true,
+                    delete_lock_fence: Some(DeleteLockFence::lost_for_test()),
+                    delete_replication_config_snapshot: Some(Arc::new(DeleteReplicationConfigSnapshot::default())),
+                    ..Default::default()
+                },
+            )
+            .await;
+
+        assert!(
+            matches!(errors[0], Some(Error::NamespaceLockQuorumUnavailable { .. })),
+            "lost outer lock must fail the batch before disk mutation: {:?}",
+            errors[0]
+        );
+        set_disks
+            .get_object_info(bucket, object, &ObjectOptions::default())
+            .await
+            .expect("object must survive a lost-lock batch");
+    }
+
     #[tokio::test]
     async fn delete_objects_blocks_locked_object_and_deletes_the_rest() {
         let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;

crates/ecstore/src/store/mod.rs

@@ -519,7 +519,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for ECStore {
         result
     }
 
-    #[instrument(skip(self))]
+    #[instrument(skip(self, objects, opts))]
     async fn delete_objects(
         &self,
         bucket: &str,

crates/ecstore/src/store/object.rs

@@ -13,7 +13,9 @@
 // limitations under the License.
 
 use super::*;
+use crate::bucket::replication::ReplicationObjectBridge;
 use crate::disk::OldCurrentSize;
+use crate::object_api::DeleteLockFence;
 use crate::set_disk::{
     get_lock_acquire_timeout, get_object_lock_diag_slow_acquire_threshold, get_object_lock_diag_slow_hold_threshold,
     is_lock_optimization_enabled, is_object_lock_diag_enabled,
@@ -156,6 +158,13 @@ impl ObjectLockDiagGuard {
             acquired_at: Instant::now(),
         }
     }
+
+    fn lock_lost_signal(&self) -> Option<Arc<rustfs_lock::distributed_lock::LockLostSignal>> {
+        match &self.guard {
+            rustfs_lock::NamespaceLockGuard::Standard(guard) => Some(guard.lock_lost()),
+            rustfs_lock::NamespaceLockGuard::Fast(_) => None,
+        }
+    }
 }
 
 /// Opaque write-lock guard for the RestoreObject accept path; see
@@ -594,6 +603,9 @@ impl ECStore {
             guards.push(self.acquire_object_write_lock("delete_objects", bucket, object).await?);
         }
         opts.no_lock = true;
+        opts.delete_lock_fence = Some(DeleteLockFence::new(
+            guards.iter().filter_map(ObjectLockDiagGuard::lock_lost_signal).collect(),
+        ));
 
         Ok(guards)
     }
@@ -1222,7 +1234,7 @@ impl ECStore {
         Err(StorageError::ObjectNotFound(bucket.to_owned(), object.to_owned()))
     }
 
-    #[instrument(skip(self))]
+    #[instrument(skip(self, objects, opts))]
     pub(super) async fn handle_delete_objects(
         &self,
         bucket: &str,
@@ -1248,6 +1260,17 @@ impl ECStore {
         }
 
         let mut opts = opts;
+        if opts.delete_replication_config_snapshot.is_none() {
+            match ReplicationObjectBridge::delete_request_config_in(&self.ctx, bucket).await {
+                Ok(snapshot) => opts.delete_replication_config_snapshot = Some(Arc::new(snapshot)),
+                Err(err) => {
+                    let message = err.to_string();
+                    let errors = (0..objects.len()).map(|_| Some(Error::other(message.clone()))).collect();
+                    return (del_objects, errors);
+                }
+            }
+        }
+
         let _object_lock_guards = match self.acquire_delete_objects_write_locks(bucket, &objects, &mut opts).await {
             Ok(guards) => guards,
             Err(err) => return return_batch_delete_lock_error(objects.as_slice(), err),
@@ -2540,6 +2563,7 @@ mod tests {
 
         assert_eq!(guards.len(), 2, "duplicate object names should share one namespace lock");
         assert!(opts.no_lock, "set layer should not reacquire locks already held by ECStore");
+        assert!(opts.delete_lock_fence.is_some(), "set layer must receive the outer write-lock loss fence");
 
         let alpha_lock = store
             .handle_new_ns_lock("bucket", "alpha")

crates/ecstore/tests/replication_facade_compat_test.rs

@@ -15,6 +15,8 @@
 mod storage_api;
 
 use s3s::dto::ReplicationConfiguration;
+use std::future::Future;
+use storage_api::contract_compat::{Error, ObjectInfo, ObjectOptions, ObjectToDelete};
 use storage_api::replication_compat::{
     BucketStats, DeletedObjectReplicationInfo, DynReplicationPool, ObjectOpts, REPLICATE_INCOMING_DELETE, ReplicateDecision,
     ReplicationConfigurationExt, ReplicationDeleteScheduleInput, ReplicationDeleteStateSource, ReplicationObjectBridge,
@@ -34,6 +36,8 @@ fn type_name_unsized<T: ?Sized>() -> &'static str {
 
 fn assert_replication_config_ext<T: ReplicationConfigurationExt>() {}
 
+fn assert_strict_delete_future<F: Future<Output = Result<ReplicateDecision, Error>>>(_: &F) {}
+
 #[test]
 fn replication_facade_exports_config_extension_contract() {
     assert_replication_config_ext::<ReplicationConfiguration>();
@@ -58,6 +62,13 @@ fn replication_facade_exports_runtime_and_dto_types() {
     assert!(type_name::<DeletedObjectReplicationInfo>().contains("DeletedObjectReplicationInfo"));
     assert!(type_name::<ReplicationObjectBridge>().contains("ReplicationObjectBridge"));
     assert!(type_name_unsized::<DynReplicationPool>().contains("ReplicationPoolTrait"));
+
+    let object = ObjectToDelete::default();
+    let source = ObjectInfo::default();
+    let opts = ObjectOptions::default();
+    let future = ReplicationObjectBridge::check_delete_strict("bucket", &object, &source, &opts, None);
+    assert_strict_delete_future(&future);
+    assert!(!std::any::type_name_of_val(&future).is_empty());
 }
 
 #[test]

crates/replication/src/config.rs

@@ -18,9 +18,12 @@ use crate::rule::ReplicationRuleExt as _;
 use s3s::dto::DeleteMarkerReplicationStatus;
 use s3s::dto::DeleteReplicationStatus;
 use s3s::dto::Destination;
-use s3s::dto::{ExistingObjectReplicationStatus, ReplicationConfiguration, ReplicationRuleStatus, ReplicationRules};
+use s3s::dto::{
+    ExistingObjectReplicationStatus, ReplicaModificationsStatus, ReplicationConfiguration, ReplicationRule,
+    ReplicationRuleStatus, ReplicationRules,
+};
 use serde::{Deserialize, Serialize};
-use std::collections::HashSet;
+use std::collections::{HashMap, HashSet};
 use uuid::Uuid;
 
 #[derive(Debug, Clone, Serialize, Deserialize, Default)]
@@ -43,6 +46,44 @@ pub trait ReplicationConfigurationExt {
     fn get_destination(&self) -> Destination;
     fn has_active_rules(&self, prefix: &str, recursive: bool) -> bool;
     fn filter_target_arns(&self, obj: &ObjectOpts) -> Vec<String>;
+    fn filter_target_replication_decisions(&self, obj: &ObjectOpts) -> Vec<(String, bool)> {
+        self.filter_target_arns(obj)
+            .into_iter()
+            .map(|arn| {
+                let mut target = obj.clone();
+                target.target_arn = arn.clone();
+                (arn, self.replicate(&target))
+            })
+            .collect()
+    }
+}
+
+fn rule_replicates(rule: &ReplicationRule, obj: &ObjectOpts) -> bool {
+    if let Some(status) = &rule.existing_object_replication
+        && obj.existing_object
+        && status.status == ExistingObjectReplicationStatus::from_static(ExistingObjectReplicationStatus::DISABLED)
+    {
+        return false;
+    }
+
+    if obj.op_type != ReplicationType::Delete {
+        return rule.metadata_replicate(obj);
+    }
+
+    if !rule.metadata_replicate(obj) {
+        return false;
+    }
+
+    let version_purge = obj.version_id.is_some();
+    if version_purge {
+        rule.delete_replication
+            .as_ref()
+            .is_some_and(|delete| delete.status == DeleteReplicationStatus::from_static(DeleteReplicationStatus::ENABLED))
+    } else {
+        rule.delete_marker_replication.as_ref().is_some_and(|delete_marker| {
+            delete_marker.status == Some(DeleteMarkerReplicationStatus::from_static(DeleteMarkerReplicationStatus::ENABLED))
+        })
+    }
 }
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -73,6 +114,58 @@ pub fn unsupported_replication_config_field(config: &ReplicationConfiguration) -
     None
 }
 
+pub fn invalid_replication_config_status_field(config: &ReplicationConfiguration) -> Option<&'static str> {
+    for rule in &config.rules {
+        if !matches!(rule.status.as_str(), ReplicationRuleStatus::ENABLED | ReplicationRuleStatus::DISABLED) {
+            return Some("Rule.Status");
+        }
+        if rule.existing_object_replication.as_ref().is_some_and(|existing| {
+            !matches!(
+                existing.status.as_str(),
+                ExistingObjectReplicationStatus::ENABLED | ExistingObjectReplicationStatus::DISABLED
+            )
+        }) {
+            return Some("Rule.ExistingObjectReplication.Status");
+        }
+        if rule.delete_replication.as_ref().is_some_and(|delete| {
+            !matches!(
+                delete.status.as_str(),
+                DeleteReplicationStatus::ENABLED | DeleteReplicationStatus::DISABLED
+            )
+        }) {
+            return Some("Rule.DeleteReplication.Status");
+        }
+        if rule
+            .delete_marker_replication
+            .as_ref()
+            .and_then(|delete| delete.status.as_ref())
+            .is_some_and(|status| {
+                !matches!(
+                    status.as_str(),
+                    DeleteMarkerReplicationStatus::ENABLED | DeleteMarkerReplicationStatus::DISABLED
+                )
+            })
+        {
+            return Some("Rule.DeleteMarkerReplication.Status");
+        }
+        if rule
+            .source_selection_criteria
+            .as_ref()
+            .and_then(|criteria| criteria.replica_modifications.as_ref())
+            .is_some_and(|modifications| {
+                !matches!(
+                    modifications.status.as_str(),
+                    ReplicaModificationsStatus::ENABLED | ReplicaModificationsStatus::DISABLED
+                )
+            })
+        {
+            return Some("Rule.SourceSelectionCriteria.ReplicaModifications.Status");
+        }
+    }
+
+    None
+}
+
 pub fn active_replication_rule_destination_arns(config: &ReplicationConfiguration) -> HashSet<String> {
     let mut arns = HashSet::new();
 
@@ -220,40 +313,7 @@ impl ReplicationConfigurationExt for ReplicationConfiguration {
     /// Determine whether an object should be replicated
     fn replicate(&self, obj: &ObjectOpts) -> bool {
         let rules = self.filter_actionable_rules(obj);
-
-        for rule in rules.iter() {
-            if rule.status == ReplicationRuleStatus::from_static(ReplicationRuleStatus::DISABLED) {
-                continue;
-            }
-
-            if let Some(status) = &rule.existing_object_replication
-                && obj.existing_object
-                && status.status == ExistingObjectReplicationStatus::from_static(ExistingObjectReplicationStatus::DISABLED)
-            {
-                return false;
-            }
-
-            if obj.op_type == ReplicationType::Delete {
-                if !rule.metadata_replicate(obj) {
-                    return false;
-                }
-
-                if obj.version_id.is_some() {
-                    return rule
-                        .delete_replication
-                        .clone()
-                        .is_some_and(|d| d.status == DeleteReplicationStatus::from_static(DeleteReplicationStatus::ENABLED));
-                } else {
-                    return rule.delete_marker_replication.clone().is_some_and(|d| {
-                        d.status == Some(DeleteMarkerReplicationStatus::from_static(DeleteMarkerReplicationStatus::ENABLED))
-                    });
-                }
-            }
-
-            // Regular object/metadata replication
-            return rule.metadata_replicate(obj);
-        }
-        false
+        rules.first().is_some_and(|rule| rule_replicates(rule, obj))
     }
 
     /// Check for an active rule
@@ -317,6 +377,41 @@ impl ReplicationConfigurationExt for ReplicationConfiguration {
         }
         arns
     }
+
+    fn filter_target_replication_decisions(&self, obj: &ObjectOpts) -> Vec<(String, bool)> {
+        let rules = self.filter_actionable_rules(obj);
+        let role = self.role.trim();
+        if !role.is_empty() {
+            let mut selected = None;
+            for rule in &rules {
+                if selected.is_none_or(|current: &ReplicationRule| rule.priority > current.priority) {
+                    selected = Some(rule);
+                }
+            }
+            return vec![(role.to_string(), selected.is_some_and(|rule| rule_replicates(rule, obj)))];
+        }
+
+        let mut target_indexes: HashMap<&str, usize> = HashMap::new();
+        let mut selected_rules: Vec<(&str, &ReplicationRule)> = Vec::new();
+        for rule in &rules {
+            let arn = rule.destination.bucket.trim();
+            if arn.is_empty() {
+                continue;
+            }
+            if let Some(index) = target_indexes.get(arn).copied() {
+                if rule.priority > selected_rules[index].1.priority {
+                    selected_rules[index].1 = rule;
+                }
+            } else {
+                target_indexes.insert(arn, selected_rules.len());
+                selected_rules.push((arn, rule));
+            }
+        }
+        selected_rules
+            .into_iter()
+            .map(|(arn, rule)| (arn.to_string(), rule_replicates(rule, obj)))
+            .collect()
+    }
 }
 
 #[cfg(test)]
@@ -324,8 +419,8 @@ mod tests {
     use super::*;
     use s3s::dto::{
         DeleteMarkerReplication, DeleteReplication, Destination, EncryptionConfiguration, ExistingObjectReplication, Metrics,
-        MetricsStatus, ReplicationRule, ReplicationTime, ReplicationTimeStatus, ReplicationTimeValue, SourceSelectionCriteria,
-        SseKmsEncryptedObjects, SseKmsEncryptedObjectsStatus,
+        MetricsStatus, ReplicaModifications, ReplicationRule, ReplicationTime, ReplicationTimeStatus, ReplicationTimeValue,
+        SourceSelectionCriteria, SseKmsEncryptedObjects, SseKmsEncryptedObjectsStatus,
     };
 
     fn replication_rule(id: &str, arn: &str) -> ReplicationRule {
@@ -627,61 +722,75 @@ mod tests {
         );
     }
 
+    #[test]
+    fn role_delete_decision_follows_highest_priority_rule() {
+        let role = "arn:rustfs:replication:us-east-1:role-target:bucket";
+        let destination = "arn:rustfs:replication:us-east-1:target:bucket";
+        let config = ReplicationConfiguration {
+            role: role.to_string(),
+            rules: vec![
+                delete_marker_rule("low-priority-enabled", destination, "logs/", 1, true),
+                delete_marker_rule("high-priority-disabled", destination, "logs/2026/", 5, false),
+            ],
+        };
+        let opts = ObjectOpts {
+            name: "logs/2026/app.log".to_string(),
+            op_type: ReplicationType::Delete,
+            delete_marker: true,
+            ..Default::default()
+        };
+
+        assert_eq!(
+            config.filter_target_replication_decisions(&opts),
+            vec![(role.to_string(), false)],
+            "the role target must use the highest-priority matching rule"
+        );
+    }
+
     #[test]
     fn version_purge_uses_delete_replication_for_object_and_marker_versions() {
         let arn = "arn:rustfs:replication:us-east-1:target:bucket";
-        let mut rule = replication_rule("delete", arn);
-        rule.delete_marker_replication = Some(DeleteMarkerReplication {
-            status: Some(DeleteMarkerReplicationStatus::from_static(DeleteMarkerReplicationStatus::DISABLED)),
-        });
+        let mut rule = delete_marker_rule("delete-switches", arn, "", 1, true);
         rule.delete_replication = Some(DeleteReplication {
-            status: DeleteReplicationStatus::from_static(DeleteReplicationStatus::ENABLED),
+            status: DeleteReplicationStatus::from_static(DeleteReplicationStatus::DISABLED),
         });
         let mut config = ReplicationConfiguration {
             role: String::new(),
             rules: vec![rule],
         };
-        let version_id = Some(Uuid::new_v4());
 
-        for delete_marker in [false, true] {
-            assert!(config.replicate(&ObjectOpts {
-                name: "object".to_string(),
-                version_id,
-                delete_marker,
-                op_type: ReplicationType::Delete,
-                ..Default::default()
-            }));
+        for version_id in [Some(Uuid::new_v4()), Some(Uuid::nil())] {
+            for delete_marker in [false, true] {
+                assert!(!config.replicate(&ObjectOpts {
+                    name: "object".to_string(),
+                    op_type: ReplicationType::Delete,
+                    version_id,
+                    delete_marker,
+                    ..Default::default()
+                }));
+            }
         }
-        assert!(!config.replicate(&ObjectOpts {
+
+        let stored_marker = ObjectOpts {
             name: "object".to_string(),
-            delete_marker: true,
             op_type: ReplicationType::Delete,
+            delete_marker: true,
             ..Default::default()
-        }));
+        };
+        assert!(config.replicate(&stored_marker), "stored markers must use DeleteMarkerReplication");
+        assert_eq!(config.filter_target_replication_decisions(&stored_marker), vec![(arn.to_string(), true)]);
 
-        let rule = &mut config.rules[0];
-        rule.delete_marker_replication = Some(DeleteMarkerReplication {
-            status: Some(DeleteMarkerReplicationStatus::from_static(DeleteMarkerReplicationStatus::ENABLED)),
+        config.rules[0].delete_replication = Some(DeleteReplication {
+            status: DeleteReplicationStatus::from_static(DeleteReplicationStatus::ENABLED),
         });
-        rule.delete_replication = Some(DeleteReplication {
-            status: DeleteReplicationStatus::from_static(DeleteReplicationStatus::DISABLED),
-        });
-
-        for delete_marker in [false, true] {
-            assert!(!config.replicate(&ObjectOpts {
-                name: "object".to_string(),
-                version_id,
-                delete_marker,
-                op_type: ReplicationType::Delete,
-                ..Default::default()
-            }));
-        }
         assert!(config.replicate(&ObjectOpts {
             name: "object".to_string(),
-            delete_marker: true,
             op_type: ReplicationType::Delete,
+            version_id: Some(Uuid::nil()),
+            delete_marker: true,
             ..Default::default()
         }));
+        assert!(config.replicate(&stored_marker));
     }
 
     #[test]
@@ -721,4 +830,77 @@ mod tests {
         });
         assert_eq!(unsupported_replication_config_field(&config), Some("Destination.ReplicationTime"));
     }
+
+    #[test]
+    fn invalid_replication_status_fields_are_reported_before_persistence() {
+        let arn = "arn:rustfs:replication:us-east-1:target:bucket";
+        let mut config = ReplicationConfiguration {
+            role: String::new(),
+            rules: vec![replication_rule("invalid-status", arn)],
+        };
+
+        config.rules[0].status = ReplicationRuleStatus::from_static("Invalid");
+        assert_eq!(invalid_replication_config_status_field(&config), Some("Rule.Status"));
+
+        config.rules[0] = replication_rule("invalid-status", arn);
+        config.rules[0].existing_object_replication = Some(ExistingObjectReplication {
+            status: ExistingObjectReplicationStatus::from_static("Invalid"),
+        });
+        assert_eq!(
+            invalid_replication_config_status_field(&config),
+            Some("Rule.ExistingObjectReplication.Status")
+        );
+
+        config.rules[0] = replication_rule("invalid-status", arn);
+        config.rules[0].delete_replication = Some(DeleteReplication {
+            status: DeleteReplicationStatus::from_static("Invalid"),
+        });
+        assert_eq!(invalid_replication_config_status_field(&config), Some("Rule.DeleteReplication.Status"));
+
+        config.rules[0] = replication_rule("invalid-status", arn);
+        config.rules[0].delete_marker_replication = Some(DeleteMarkerReplication {
+            status: Some(DeleteMarkerReplicationStatus::from_static("Invalid")),
+        });
+        assert_eq!(
+            invalid_replication_config_status_field(&config),
+            Some("Rule.DeleteMarkerReplication.Status")
+        );
+
+        config.rules[0] = replication_rule("invalid-status", arn);
+        config.rules[0].source_selection_criteria = Some(SourceSelectionCriteria {
+            replica_modifications: Some(ReplicaModifications {
+                status: ReplicaModificationsStatus::from_static("Invalid"),
+            }),
+            sse_kms_encrypted_objects: None,
+        });
+        assert_eq!(
+            invalid_replication_config_status_field(&config),
+            Some("Rule.SourceSelectionCriteria.ReplicaModifications.Status")
+        );
+    }
+
+    #[test]
+    fn target_decisions_choose_highest_priority_rule_per_destination() {
+        let target_a = "arn:rustfs:replication:us-east-1:target:a";
+        let target_b = "arn:rustfs:replication:us-east-1:target:b";
+        let mut a_low = delete_marker_rule("a-low", target_a, "logs/", 1, true);
+        let b = delete_marker_rule("b", target_b, "logs/", 2, true);
+        let a_high = delete_marker_rule("a-high", target_a, "logs/2026/", 5, false);
+        a_low.delete_replication = Some(DeleteReplication {
+            status: DeleteReplicationStatus::from_static(DeleteReplicationStatus::ENABLED),
+        });
+        let config = ReplicationConfiguration {
+            role: String::new(),
+            rules: vec![a_low, b, a_high],
+        };
+
+        let decisions = config.filter_target_replication_decisions(&ObjectOpts {
+            name: "logs/2026/app.log".to_string(),
+            op_type: ReplicationType::Delete,
+            delete_marker: true,
+            ..Default::default()
+        });
+
+        assert_eq!(decisions, vec![(target_a.to_string(), false), (target_b.to_string(), true)]);
+    }
 }

crates/replication/src/lib.rs

@@ -30,8 +30,8 @@ pub mod tagging;
 
 pub use config::{
     ObjectOpts, ReplicationConfigurationExt, ReplicationTargetValidationError, active_replication_rule_destination_arns,
-    replication_target_arns, should_remove_replication_target, unsupported_replication_config_field,
-    validate_replication_config_target_arns,
+    invalid_replication_config_status_field, replication_target_arns, should_remove_replication_target,
+    unsupported_replication_config_field, validate_replication_config_target_arns,
 };
 pub use delete::{
     DeletedObjectReplicationInfo, is_retryable_delete_replication_head_error, is_version_delete_replication,

crates/replication/src/operation.rs

@@ -133,16 +133,13 @@ pub fn delete_replication_state_from_config(
         replica: source.replica,
         ..Default::default()
     };
-    let target_arns = config.filter_target_arns(&opts);
-    if target_arns.is_empty() {
+    let target_decisions = config.filter_target_replication_decisions(&opts);
+    if target_decisions.is_empty() {
         return None;
     }
 
     let mut decision = ReplicateDecision::new();
-    for target_arn in target_arns {
-        let mut target_opts = opts.clone();
-        target_opts.target_arn = target_arn.clone();
-        let replicate = config.replicate(&target_opts);
+    for (target_arn, replicate) in target_decisions {
         decision.set(ReplicateTargetDecision::new(target_arn, replicate, false));
     }
     if !decision.replicate_any() {
@@ -174,20 +171,13 @@ pub struct ReplicationDeleteScheduleInput<'a> {
     pub deleted_delete_marker_version: bool,
 }
 
-fn delete_version_purge_source_status(status: &ReplicationStatusType) -> bool {
-    status == &ReplicationStatusType::Replica
-        || status == &ReplicationStatusType::Pending
-        || status == &ReplicationStatusType::Completed
-        || status == &ReplicationStatusType::Failed
-}
-
 pub fn should_schedule_delete_replication(input: ReplicationDeleteScheduleInput<'_>) -> bool {
     if input.replication_request {
         return false;
     }
 
-    if input.version_id_requested && !input.deleted_delete_marker_version && !input.source_delete_marker {
-        return delete_version_purge_source_status(input.source_replication_status);
+    if input.version_id_requested {
+        return input.source_version_purge_status == &VersionPurgeStatusType::Pending;
     }
 
     input.source_replication_status == &ReplicationStatusType::Replica
@@ -433,9 +423,7 @@ mod tests {
             delete_marker_replication: Some(DeleteMarkerReplication {
                 status: Some(DeleteMarkerReplicationStatus::from_static(DeleteMarkerReplicationStatus::ENABLED)),
             }),
-            delete_replication: Some(DeleteReplication {
-                status: DeleteReplicationStatus::from_static(DeleteReplicationStatus::ENABLED),
-            }),
+            delete_replication: None,
             destination: Destination {
                 bucket: arn.to_string(),
                 ..Default::default()
@@ -500,11 +488,15 @@ mod tests {
     }
 
     #[test]
-    fn delete_replication_state_tracks_delete_marker_version_purges() {
+    fn delete_replication_state_uses_delete_switch_for_marker_version_purges() {
         let arn = "arn:aws:s3:::target-bucket";
-        let config = ReplicationConfiguration {
+        let mut rule = delete_replication_rule(arn, false);
+        rule.delete_replication = Some(DeleteReplication {
+            status: DeleteReplicationStatus::from_static(DeleteReplicationStatus::DISABLED),
+        });
+        let mut config = ReplicationConfiguration {
             role: arn.to_string(),
-            rules: vec![delete_replication_rule(arn, false)],
+            rules: vec![rule],
         };
         let source = ReplicationDeleteStateSource {
             name: "test/object.txt".to_string(),
@@ -514,8 +506,16 @@ mod tests {
             replica: false,
         };
 
+        assert!(
+            delete_replication_state_from_config(&config, &source).is_none(),
+            "delete-marker version purge must not use the enabled marker-creation switch"
+        );
+
+        config.rules[0].delete_replication = Some(DeleteReplication {
+            status: DeleteReplicationStatus::from_static(DeleteReplicationStatus::ENABLED),
+        });
         let state = delete_replication_state_from_config(&config, &source)
-            .expect("delete-marker version purge should honor delete replication rules");
+            .expect("delete-marker version purge should honor the enabled permanent-delete switch");
         let pending = format!("{arn}=PENDING;");
 
         assert_eq!(state.version_purge_status_internal.as_deref(), Some(pending.as_str()));
@@ -536,8 +536,8 @@ mod tests {
     }
 
     #[test]
-    fn delete_replication_schedule_keeps_marker_and_version_purges() {
-        assert!(should_schedule_delete_replication(ReplicationDeleteScheduleInput {
+    fn delete_replication_schedule_requires_current_admission_for_version_purges() {
+        assert!(!should_schedule_delete_replication(ReplicationDeleteScheduleInput {
             replication_request: false,
             version_id_requested: true,
             source_delete_marker: true,
@@ -549,8 +549,16 @@ mod tests {
             replication_request: false,
             version_id_requested: true,
             source_delete_marker: false,
-            source_replication_status: &ReplicationStatusType::Completed,
-            source_version_purge_status: &VersionPurgeStatusType::Empty,
+            source_replication_status: &ReplicationStatusType::Empty,
+            source_version_purge_status: &VersionPurgeStatusType::Pending,
+            deleted_delete_marker_version: true,
+        }));
+        assert!(should_schedule_delete_replication(ReplicationDeleteScheduleInput {
+            replication_request: false,
+            version_id_requested: true,
+            source_delete_marker: false,
+            source_replication_status: &ReplicationStatusType::Empty,
+            source_version_purge_status: &VersionPurgeStatusType::Pending,
             deleted_delete_marker_version: false,
         }));
         assert!(should_schedule_delete_replication(ReplicationDeleteScheduleInput {

crates/replication/src/queue.rs

@@ -410,6 +410,21 @@ mod tests {
         assert_eq!(delete_info.delete_object.delete_marker_version_id, None);
     }
 
+    #[test]
+    fn heal_queue_action_preserves_pending_null_version_purge() {
+        let mut roi = replicate_object_info(ReplicationStatusType::Completed);
+        roi.version_id = Some(Uuid::nil());
+        roi.version_purge_status = VersionPurgeStatusType::Pending;
+
+        let action = replication_heal_queue_action(&mut roi);
+
+        let ReplicationHealQueueAction::QueueDelete(delete_info) = action else {
+            panic!("expected null-version purge delete queue action");
+        };
+        assert_eq!(delete_info.delete_object.version_id, Some(Uuid::nil()));
+        assert_eq!(delete_info.delete_object.delete_marker_version_id, None);
+    }
+
     #[test]
     fn replication_priority_parses_known_values_and_defaults_unknown() {
         assert_eq!(ReplicationPriority::from_str("fast"), Ok(ReplicationPriority::Fast));

crates/scanner/src/scanner_folder.rs

@@ -1038,7 +1038,7 @@ impl ScannerItem {
     }
 
     async fn heal_replication(&mut self, oi: &ObjectInfo, size_summary: &mut SizeSummary) {
-        if oi.version_id.is_none_or(|v| v.is_nil()) {
+        if oi.version_id.is_none_or(|version| version.is_nil()) && !oi.delete_marker && oi.version_purge_status.is_empty() {
             return;
         }
 
@@ -3188,6 +3188,61 @@ mod tests {
         assert!(!ScannerItem::should_account_replication_stats(&purge_version));
     }
 
+    #[tokio::test]
+    #[serial]
+    async fn test_heal_replication_only_queues_pending_null_deletes() {
+        async fn replication_skipped_count() -> u64 {
+            global_metrics()
+                .report()
+                .await
+                .source_work
+                .iter()
+                .find(|work| work.source == ScannerWorkSource::BucketReplication.as_str())
+                .map(|work| work.skipped)
+                .unwrap_or_default()
+        }
+
+        let temp_dir = std::env::temp_dir();
+        let file_type = std::fs::metadata(&temp_dir)
+            .expect("temp dir metadata should be readable")
+            .file_type();
+        let mut item = ScannerItem {
+            path: temp_dir.join("object").to_string_lossy().to_string(),
+            bucket: "bucket".to_string(),
+            prefix: String::new(),
+            object_name: "object".to_string(),
+            file_type,
+            lifecycle: None,
+            object_lock: None,
+            replication: Some(Arc::new(ReplicationConfig::new(None, None))),
+            heal_enabled: false,
+            heal_bitrot: false,
+            debug: false,
+        };
+        let null_object = ObjectInfo {
+            bucket: "bucket".to_string(),
+            name: "object".to_string(),
+            version_id: Some(Uuid::nil()),
+            mod_time: Some(OffsetDateTime::now_utc()),
+            ..Default::default()
+        };
+        let mut size_summary = SizeSummary::default();
+        let before = replication_skipped_count().await;
+
+        item.heal_replication(&null_object, &mut size_summary).await;
+        assert_eq!(replication_skipped_count().await, before);
+
+        item.heal_replication(
+            &ObjectInfo {
+                version_purge_status: VersionPurgeStatusType::Pending,
+                ..null_object
+            },
+            &mut size_summary,
+        )
+        .await;
+        assert_eq!(replication_skipped_count().await, before + 1);
+    }
+
     #[tokio::test]
     async fn test_scanner_ilm_action_accounting_requires_enqueue_success() {
         let metrics = Metrics::new();

crates/storage-api/src/object.rs

@@ -183,6 +183,7 @@ pub enum WalkVersionsSortOrder {
 pub struct ObjectToDelete {
     pub object_name: String,
     pub version_id: Option<Uuid>,
+    pub synthetic_version_id: bool,
     pub delete_marker_replication_status: Option<String>,
     pub version_purge_status: Option<VersionPurgeStatusType>,
     pub version_purge_statuses: Option<String>,