fix(replication): let replicated version purges pass the peer WORM gate (#6960)

A replicated version purge reaches the peer without the governance
bypass header, so a GOVERNANCE-retained version deleted on the source
with x-amz-bypass-governance-retention was rejected by the peer's WORM
deletion gate forever: retryStats ended at a permanent failed count and
the sites stayed diverged (issue #6850).

The source is authoritative for such a purge: the same WORM gate
already ran there, and GOVERNANCE retention with an authorized bypass
is the only lock state it can purge through. The peer's commit-time
deletion gate now treats an authorized replication delete addressed to
an explicit version as carrying that judged bypass, reusing the same
trust judgment as the replication write exemption
(ObjectOptions::replication_request, set only after the handler
authorized ReplicateDeleteAction). COMPLIANCE retention and legal hold
keep blocking replicated purges, and a plain client delete without the
bypass header stays rejected.

crates/ecstore/src/bucket/object_lock/objectlock_sys.rs

@@ -177,6 +177,28 @@ pub fn replication_write_may_pass_worm_gate(
     Ok(!(retention_locked && opts.replication_retention_timestamp.is_none()))
 }
 
+/// Whether an authorized replication delete (`ObjectOptions::replication_request`)
+/// addressed to an explicit version may bypass GOVERNANCE retention on the
+/// local replica, exactly as an `x-amz-bypass-governance-retention` caller
+/// with the bypass permission would.
+///
+/// The source is authoritative for a replicated version purge (issue #6850):
+/// the same WORM deletion gate already ran there, and GOVERNANCE retention
+/// with an authorized bypass is the only lock state it can purge through.
+/// Requiring the bypass header again here makes the purge permanently
+/// undeliverable — replication senders never carry it — and the sites diverge
+/// forever. COMPLIANCE retention and legal hold stay blocking: the source
+/// gate can never purge through them, so a replication purge that meets one
+/// here is divergence or forgery and fails closed.
+///
+/// The trust judgment is the same one the write-path exemption uses:
+/// `replication_request` is only set once the receiving handler has
+/// authorized the caller for the replication action
+/// (`ReplicateDeleteAction`), never straight from request headers.
+pub fn replication_delete_may_bypass_governance(opts: &ObjectOptions) -> bool {
+    opts.replication_request && opts.version_id.is_some()
+}
+
 /// Check if an object is locked based on its metadata.
 /// This is a common function used by both lifecycle evaluation and deletion checks.
 ///
@@ -680,6 +702,32 @@ mod tests {
         assert!(err.to_string().contains("modification time"));
     }
 
+    /// The replicated-purge GOVERNANCE bypass (#6850) applies only to an
+    /// authorized replication delete addressed to an explicit version: a
+    /// local delete never gets it, and a replicated delete without a version
+    /// id creates a delete marker rather than purging anything.
+    #[test]
+    fn replication_delete_bypasses_governance_only_for_authorized_version_purges() {
+        let version_purge = ObjectOptions {
+            replication_request: true,
+            version_id: Some("6b6ffbc0-b0d3-4a86-8f6c-fe19163b8dcd".to_string()),
+            ..Default::default()
+        };
+        assert!(replication_delete_may_bypass_governance(&version_purge));
+
+        let local_version_delete = ObjectOptions {
+            replication_request: false,
+            ..version_purge.clone()
+        };
+        assert!(!replication_delete_may_bypass_governance(&local_version_delete));
+
+        let replicated_marker_creation = ObjectOptions {
+            version_id: None,
+            ..version_purge
+        };
+        assert!(!replication_delete_may_bypass_governance(&replicated_marker_creation));
+    }
+
     /// A local PutObjectRetention / PutObjectLegalHold "clear" persists the
     /// lock keys as empty strings (the MinIO on-disk shape, see
     /// `parse_object_lock_retention`); that is "no lock", not corruption, and

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

@@ -196,13 +196,17 @@ 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.
+/// Version purges the peer denied under object lock (#6850). A RustFS peer
+/// with the replicated-purge GOVERNANCE exemption
+/// (`replication_delete_may_bypass_governance`) no longer produces this for
+/// governance retention, but COMPLIANCE retention, legal hold, and targets
+/// without the exemption (older RustFS, MinIO, generic S3) still deny — and
+/// such a purge cannot succeed until the lock on the replica lapses, so
+/// 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);
@@ -2830,11 +2834,12 @@ async fn replicate_delete_to_target(dobj: &DeletedObjectReplicationInfo, tgt_cli
             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.
+                // this version under COMPLIANCE retention or legal hold, or
+                // is a target without the replicated-purge GOVERNANCE
+                // exemption (#6850), so the sites stay diverged until the
+                // lock 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,

crates/ecstore/src/set_disk/mod.rs

@@ -42,7 +42,8 @@ use crate::bucket::lifecycle::lifecycle::TRANSITION_COMPLETE;
 use crate::bucket::metadata_sys;
 use crate::bucket::metadata_sys::ObjectLockConfigState;
 use crate::bucket::object_lock::objectlock_sys::{
-    check_object_lock_for_deletion_with_state, check_retention_for_modification, replication_write_may_pass_worm_gate,
+    check_object_lock_for_deletion_with_state, check_retention_for_modification, replication_delete_may_bypass_governance,
+    replication_write_may_pass_worm_gate,
 };
 #[cfg(test)]
 use crate::bucket::replication::ReplicationState;
@@ -5578,10 +5579,15 @@ async fn check_object_lock_delete(
         return Ok(());
     }
 
+    // An authorized replicated version purge already passed this gate on the
+    // source with the bypass it carried there, so it clears GOVERNANCE
+    // retention here without the header; COMPLIANCE and legal hold still
+    // block below (see `replication_delete_may_bypass_governance`, #6850).
     let bypass_governance = opts
         .object_lock_delete
         .as_ref()
-        .is_some_and(|delete_opts| delete_opts.bypass_governance);
+        .is_some_and(|delete_opts| delete_opts.bypass_governance)
+        || replication_delete_may_bypass_governance(opts);
     let blocked = match opts.object_lock_config_snapshot.as_deref() {
         Some(snapshot) => check_object_lock_for_deletion_with_state(snapshot.state(), obj_info, bypass_governance)?.is_some(),
         None => {
@@ -11582,6 +11588,100 @@ mod tests {
             .expect("versioned delete marker creation should not delete the locked version");
     }
 
+    fn governance_retained_obj_info() -> ObjectInfo {
+        let retain_until = OffsetDateTime::now_utc() + Duration::from_secs(60 * 60 * 24 * 60);
+        let mut user_defined = HashMap::new();
+        user_defined.insert(
+            X_AMZ_OBJECT_LOCK_MODE.as_str().to_string(),
+            s3s::dto::ObjectLockRetentionMode::GOVERNANCE.to_string(),
+        );
+        user_defined.insert(
+            X_AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE.as_str().to_string(),
+            retain_until.format(&time::format_description::well_known::Rfc3339).unwrap(),
+        );
+        ObjectInfo {
+            user_defined: Arc::new(user_defined),
+            ..Default::default()
+        }
+    }
+
+    fn explicit_version_delete_opts(replication_request: bool) -> ObjectOptions {
+        ObjectOptions {
+            version_id: Some(Uuid::new_v4().to_string()),
+            versioned: true,
+            replication_request,
+            object_lock_config_snapshot: Some(Arc::new(ObjectLockConfigSnapshot::new(ObjectLockConfigState::ConfirmedAbsent))),
+            ..Default::default()
+        }
+    }
+
+    // Issue #6850: a replicated version purge carries no bypass header, so the
+    // GOVERNANCE gate must honor the source's already-judged bypass instead of
+    // keeping the sites permanently diverged.
+    #[tokio::test]
+    async fn test_check_object_lock_delete_allows_replicated_governance_version_purge() {
+        let obj_info = governance_retained_obj_info();
+        let opts = explicit_version_delete_opts(true);
+
+        check_object_lock_delete(&bootstrap_ctx(), "bucket", "object", &obj_info, &opts)
+            .await
+            .expect("an authorized replicated version purge must pass GOVERNANCE retention (#6850)");
+    }
+
+    #[tokio::test]
+    async fn test_check_object_lock_delete_blocks_plain_governance_version_delete_without_bypass() {
+        let obj_info = governance_retained_obj_info();
+        let opts = explicit_version_delete_opts(false);
+
+        let err = check_object_lock_delete(&bootstrap_ctx(), "bucket", "object", &obj_info, &opts)
+            .await
+            .expect_err("a plain client delete without the bypass header must stay blocked by GOVERNANCE retention");
+
+        assert!(matches!(err, StorageError::PrefixAccessDenied(_, _)));
+    }
+
+    #[tokio::test]
+    async fn test_check_object_lock_delete_blocks_replicated_compliance_version_purge() {
+        let retain_until = OffsetDateTime::now_utc() + Duration::from_secs(60 * 60 * 24 * 60);
+        let mut user_defined = HashMap::new();
+        user_defined.insert(
+            X_AMZ_OBJECT_LOCK_MODE.as_str().to_string(),
+            s3s::dto::ObjectLockRetentionMode::COMPLIANCE.to_string(),
+        );
+        user_defined.insert(
+            X_AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE.as_str().to_string(),
+            retain_until.format(&time::format_description::well_known::Rfc3339).unwrap(),
+        );
+        let obj_info = ObjectInfo {
+            user_defined: Arc::new(user_defined),
+            ..Default::default()
+        };
+        let opts = explicit_version_delete_opts(true);
+
+        let err = check_object_lock_delete(&bootstrap_ctx(), "bucket", "object", &obj_info, &opts)
+            .await
+            .expect_err("the source gate can never purge through COMPLIANCE, so a replicated purge fails closed");
+
+        assert!(matches!(err, StorageError::PrefixAccessDenied(_, _)));
+    }
+
+    #[tokio::test]
+    async fn test_check_object_lock_delete_blocks_replicated_legal_hold_version_purge() {
+        let mut user_defined = HashMap::new();
+        user_defined.insert(X_AMZ_OBJECT_LOCK_LEGAL_HOLD.as_str().to_string(), "ON".to_string());
+        let obj_info = ObjectInfo {
+            user_defined: Arc::new(user_defined),
+            ..Default::default()
+        };
+        let opts = explicit_version_delete_opts(true);
+
+        let err = check_object_lock_delete(&bootstrap_ctx(), "bucket", "object", &obj_info, &opts)
+            .await
+            .expect_err("the source gate can never purge through a legal hold, so a replicated purge fails closed");
+
+        assert!(matches!(err, StorageError::PrefixAccessDenied(_, _)));
+    }
+
     // backlog#929 (HP-8): the delete_objects per-object stat is gated on the
     // bucket object-lock configuration. Lock-enabled buckets (either legacy
     // lock_enabled flag or an enabled ObjectLockConfiguration) and unknown