fix(replication): surface object-lock denied purges and back off heal retries (#6900)

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

@@ -20,8 +20,8 @@ pub use rustfs_replication::{
 pub(crate) use rustfs_replication::{
     ReplicationDeleteSource, ReplicationMultipartPartInput, ReplicationResyncTargetObject, delete_marker_purge_mrf_entry,
     delete_marker_purge_version_id, delete_replication_creates_marker, delete_replication_missing_source_decision,
-    delete_replication_object_opts, heal_uses_delete_replication_path, is_retryable_delete_replication_head_error,
-    is_version_delete_replication, replicate_delete_outcome, replication_etags_match, replication_multipart_complete_actual_size,
-    replication_multipart_part_plan, resync_existing_delete_replication_info, resync_target_for_object,
-    should_retry_delete_marker_purge, single_part_replica_etag_mismatch, target_delete_version_id,
+    delete_replication_object_opts, heal_uses_delete_replication_path, is_object_lock_denied_delete,
+    is_retryable_delete_replication_head_error, is_version_delete_replication, replicate_delete_outcome, replication_etags_match,
+    replication_multipart_complete_actual_size, replication_multipart_part_plan, resync_existing_delete_replication_info,
+    resync_target_for_object, should_retry_delete_marker_purge, single_part_replica_etag_mismatch, target_delete_version_id,
 };

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

@@ -3177,6 +3177,19 @@ pub(crate) async fn queue_replication_heal_internal(
             }
         }
         ReplicationHealQueueAction::QueueDelete(dv) => {
+            // A purge the peer denied under object lock cannot succeed until
+            // the lock lapses (#6850); requeuing it every heal cycle only
+            // burns bandwidth and failure counters. The backoff expires on
+            // its own, so the purge is probed again — and converges — once
+            // the retention window has a chance of being over.
+            if super::replication_object_decision_boundary::is_version_delete_replication(&dv.delete_object)
+                && super::replication_resyncer::object_lock_denied_purge_backoff_active(&dv)
+            {
+                return ReplicationHealQueueResult {
+                    object_info: roi,
+                    admission: ReplicationQueueAdmission::Skipped,
+                };
+            }
             let admission = if let Some(pool) = runtime_sources::replication_pool() {
                 pool.queue_replica_delete_task(dv).await
             } else {

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

@@ -30,10 +30,10 @@ use super::replication_msgp_boundary::ReplicationMsgpCodec;
 use super::replication_object_config::{ReplicationConfig, get_replication_config, must_replicate};
 use super::replication_object_decision_boundary::{
     MustReplicateOptions, ReplicationMultipartPartInput, delete_marker_purge_mrf_entry, delete_marker_purge_version_id,
-    delete_replication_creates_marker, heal_uses_delete_replication_path, is_retryable_delete_replication_head_error,
-    is_version_delete_replication, replicate_delete_outcome, replication_etags_match, replication_multipart_complete_actual_size,
-    replication_multipart_part_plan, resync_existing_delete_replication_info, should_retry_delete_marker_purge,
-    single_part_replica_etag_mismatch, target_delete_version_id,
+    delete_replication_creates_marker, heal_uses_delete_replication_path, is_object_lock_denied_delete,
+    is_retryable_delete_replication_head_error, is_version_delete_replication, replicate_delete_outcome, replication_etags_match,
+    replication_multipart_complete_actual_size, replication_multipart_part_plan, resync_existing_delete_replication_info,
+    should_retry_delete_marker_purge, single_part_replica_etag_mismatch, target_delete_version_id,
 };
 use super::replication_queue_boundary::{DeletedObjectReplicationInfo, ReplicationQueueAdmission};
 use super::replication_resync_boundary::ResyncStatusType;
@@ -118,6 +118,7 @@ const EVENT_DELETE_MARKER_PURGE_FAILED: &str = "replication_delete_marker_purge_
 const EVENT_DELETE_MARKER_PURGE_MRF: &str = "replication_delete_marker_purge_mrf";
 const METRIC_DELETE_MARKER_PURGE_TOTAL: &str = "rustfs_replication_delete_marker_purge_total";
 const EVENT_REPLICATION_VERSION_IDENTITY_DRIFT: &str = "replication_version_identity_drift";
+const EVENT_REPLICATION_PURGE_OBJECT_LOCK_DENIED: &str = "replication_purge_object_lock_denied";
 
 #[allow(
     dead_code,
@@ -195,6 +196,76 @@ const METRIC_VERSION_IDENTITY_DRIFT_TOTAL: &str = "rustfs_replication_version_id
 /// after a restart is acceptable.
 static VERSION_IDENTITY_WARNED_ARNS: LazyLock<StdMutex<HashSet<String>>> = LazyLock::new(|| StdMutex::new(HashSet::new()));
 
+/// Version purges the peer denied under object lock (#6850). Replication
+/// carries no governance bypass, so such a purge cannot succeed until the
+/// lock on the replica lapses — retrying every heal cycle only burns
+/// bandwidth and failure counters. Entries suppress heal requeues for the
+/// backoff window; after it expires one probe runs again, so the purge still
+/// converges on its own once retention ends. In-process only: a restart
+/// costs at most one extra probe per entry.
+const OBJECT_LOCK_DENIED_PURGE_BACKOFF: std::time::Duration = std::time::Duration::from_secs(60 * 60);
+const OBJECT_LOCK_DENIED_PURGE_CACHE_MAX: usize = 4096;
+type ObjectLockDeniedPurgeKey = (String, String, String);
+
+struct ObjectLockDeniedPurge {
+    denied_at: std::time::Instant,
+    denied_arns: HashSet<String>,
+}
+
+static OBJECT_LOCK_DENIED_PURGES: LazyLock<StdMutex<HashMap<ObjectLockDeniedPurgeKey, ObjectLockDeniedPurge>>> =
+    LazyLock::new(|| StdMutex::new(HashMap::new()));
+
+fn object_lock_denied_purge_key(dobj: &DeletedObjectReplicationInfo) -> ObjectLockDeniedPurgeKey {
+    let version_id = dobj
+        .delete_object
+        .delete_marker_version_id
+        .or(dobj.delete_object.version_id)
+        .unwrap_or_default();
+    (dobj.bucket.clone(), dobj.delete_object.object_name.clone(), version_id.to_string())
+}
+
+fn record_object_lock_denied_purge(dobj: &DeletedObjectReplicationInfo, arn: &str) {
+    let mut denied = OBJECT_LOCK_DENIED_PURGES
+        .lock()
+        .unwrap_or_else(|poisoned| poisoned.into_inner());
+    if denied.len() >= OBJECT_LOCK_DENIED_PURGE_CACHE_MAX {
+        denied.retain(|_, entry| entry.denied_at.elapsed() < OBJECT_LOCK_DENIED_PURGE_BACKOFF);
+    }
+    let key = object_lock_denied_purge_key(dobj);
+    if denied.len() < OBJECT_LOCK_DENIED_PURGE_CACHE_MAX || denied.contains_key(&key) {
+        let entry = denied.entry(key).or_insert_with(|| ObjectLockDeniedPurge {
+            denied_at: std::time::Instant::now(),
+            denied_arns: HashSet::new(),
+        });
+        entry.denied_at = std::time::Instant::now();
+        entry.denied_arns.insert(arn.to_string());
+    }
+    // Still full after dropping expired entries: skip recording — the purge
+    // then simply keeps retrying, which is the pre-#6850 behavior.
+}
+
+/// Whether a heal requeue of this delete can only reach targets that denied
+/// it under object lock within the backoff window. A target the entry does
+/// not cover (another peer, or one whose denial expired) keeps the requeue
+/// flowing — suppressing it would delay a purge that could succeed there.
+pub(crate) fn object_lock_denied_purge_backoff_active(dobj: &DeletedObjectReplicationInfo) -> bool {
+    let key = object_lock_denied_purge_key(dobj);
+    let mut denied = OBJECT_LOCK_DENIED_PURGES
+        .lock()
+        .unwrap_or_else(|poisoned| poisoned.into_inner());
+    match denied.get(&key) {
+        Some(entry) if entry.denied_at.elapsed() < OBJECT_LOCK_DENIED_PURGE_BACKOFF => {
+            let admitted = dobj.admitted_target_arns();
+            !admitted.is_empty() && admitted.iter().all(|arn| entry.denied_arns.contains(arn))
+        }
+        Some(_) => {
+            denied.remove(&key);
+            false
+        }
+        None => false,
+    }
+}
+
 const REPLICA_ETAG_VERIFY_ENV: &str = "RUSTFS_REPLICATION_REPLICA_ETAG_VERIFY";
 
 /// Escape hatch for a target whose 32-hex ETags are legitimately not the
@@ -2756,19 +2827,42 @@ async fn replicate_delete_to_target(dobj: &DeletedObjectReplicationInfo, tgt_cli
             }
         }
         Err(e) => {
-            warn!(
-                event = EVENT_RESYNC_TARGET_OPERATION_FAILED,
-                component = LOG_COMPONENT_ECSTORE,
-                subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC,
-                bucket = tgt_client.bucket,
-                object = dobj.delete_object.object_name,
-                version_id = ?version_id,
-                delete_marker = dobj.delete_object.delete_marker,
-                is_version_purge,
-                error = %e,
-                operation = "replicate_delete_to_target",
-                "Replication target operation failed"
-            );
+            let object_lock_denied = is_version_purge && is_object_lock_denied_delete(e.code.as_deref(), e.message.as_deref());
+            if object_lock_denied {
+                // Terminal for as long as the lock holds: the peer retains
+                // this version and replication carries no governance bypass
+                // (#6850), so the sites stay diverged until the retention or
+                // legal hold on the replica lapses. Surface it loudly instead
+                // of letting a silent failed counter and a hot heal-retry
+                // loop stand in for the divergence.
+                record_object_lock_denied_purge(dobj, &tgt_client.arn);
+                error!(
+                    event = EVENT_REPLICATION_PURGE_OBJECT_LOCK_DENIED,
+                    component = LOG_COMPONENT_ECSTORE,
+                    subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC,
+                    bucket = tgt_client.bucket,
+                    object = dobj.delete_object.object_name,
+                    version_id = ?version_id,
+                    arn = %tgt_client.arn,
+                    error = %e,
+                    operation = "replicate_delete_to_target",
+                    "Replicated version purge denied by object lock on the target; the sites stay diverged until the lock lapses"
+                );
+            } else {
+                warn!(
+                    event = EVENT_RESYNC_TARGET_OPERATION_FAILED,
+                    component = LOG_COMPONENT_ECSTORE,
+                    subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC,
+                    bucket = tgt_client.bucket,
+                    object = dobj.delete_object.object_name,
+                    version_id = ?version_id,
+                    delete_marker = dobj.delete_object.delete_marker,
+                    is_version_purge,
+                    error = %e,
+                    operation = "replicate_delete_to_target",
+                    "Replication target operation failed"
+                );
+            }
             rinfo.error = Some(e.to_string());
             if !is_version_purge {
                 rinfo.replication_status = ReplicationStatusType::Failed;
@@ -5511,6 +5605,35 @@ mod tests {
         assert!(!resync_state_accepts_update(&current, &stale));
     }
 
+    #[test]
+    fn object_lock_denied_purge_backoff_tracks_version_and_target() {
+        let denied = DeletedObjectReplicationInfo {
+            bucket: "worm-backoff-test-bucket".to_string(),
+            target_arn: "arn:rustfs:replication::worm-test:t1".to_string(),
+            delete_object: ReplicationDeletedObject {
+                object_name: "locked-object".to_string(),
+                version_id: Some(uuid::Uuid::new_v4()),
+                ..Default::default()
+            },
+            ..Default::default()
+        };
+        assert!(!object_lock_denied_purge_backoff_active(&denied));
+
+        record_object_lock_denied_purge(&denied, "arn:rustfs:replication::worm-test:t1");
+        assert!(object_lock_denied_purge_backoff_active(&denied));
+
+        // A requeue that can also reach a target this denial does not cover
+        // must keep flowing: the purge may succeed there.
+        let mut other_target = denied.clone();
+        other_target.target_arn = "arn:rustfs:replication::worm-test:t2".to_string();
+        assert!(!object_lock_denied_purge_backoff_active(&other_target));
+
+        // A different version of the same object must not be suppressed.
+        let mut other_version = denied;
+        other_version.delete_object.version_id = Some(uuid::Uuid::new_v4());
+        assert!(!object_lock_denied_purge_backoff_active(&other_version));
+    }
+
     #[tokio::test]
     async fn abort_multipart_on_failure_skips_abort_when_transfer_succeeded() {
         let aborted = Arc::new(AtomicBool::new(false));