From ec1cd606d31dbcd951132c2ba89fb89e816ea2f5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=94=90=E5=B0=8F=E9=B8=AD?= Date: Mon, 31 Aug 2026 02:59:55 +0800 Subject: [PATCH] fix(replication): surface object-lock denied purges and back off heal retries (#6900) --- .../replication_object_decision_boundary.rs | 8 +- .../bucket/replication/replication_pool.rs | 13 ++ .../replication/replication_resyncer.rs | 157 ++++++++++++++++-- crates/replication/src/delete.rs | 47 +++++- crates/replication/src/lib.rs | 6 +- 5 files changed, 204 insertions(+), 27 deletions(-) diff --git a/crates/ecstore/src/bucket/replication/replication_object_decision_boundary.rs b/crates/ecstore/src/bucket/replication/replication_object_decision_boundary.rs index 9b6136f68..da6b7e121 100644 --- a/crates/ecstore/src/bucket/replication/replication_object_decision_boundary.rs +++ b/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, }; diff --git a/crates/ecstore/src/bucket/replication/replication_pool.rs b/crates/ecstore/src/bucket/replication/replication_pool.rs index a6d5cf857..9b227c7c4 100644 --- a/crates/ecstore/src/bucket/replication/replication_pool.rs +++ b/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 { diff --git a/crates/ecstore/src/bucket/replication/replication_resyncer.rs b/crates/ecstore/src/bucket/replication/replication_resyncer.rs index ec4b888bf..3aa28ddfa 100644 --- a/crates/ecstore/src/bucket/replication/replication_resyncer.rs +++ b/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>> = 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, +} + +static OBJECT_LOCK_DENIED_PURGES: LazyLock>> = + 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(¤t, &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)); diff --git a/crates/replication/src/delete.rs b/crates/replication/src/delete.rs index 49d305b40..ffd014796 100644 --- a/crates/replication/src/delete.rs +++ b/crates/replication/src/delete.rs @@ -126,6 +126,27 @@ pub fn should_retry_delete_marker_purge(dobj: &DeletedObject) -> bool { dobj.delete_marker_version_id.is_some() } +/// True when the target denied a replicated delete because object-lock +/// retention or a legal hold protects that version on the replica (its +/// deletion gate answers `AccessDenied` with the lock reason, and a +/// replication request carries no governance bypass, rustfs#6850). Retrying +/// cannot succeed until the lock itself lapses, so callers treat this as a +/// policy denial rather than a transient fault. +/// +/// The reason text is the RustFS deletion-gate wording; a MinIO/AWS peer +/// phrases its WORM denial differently and simply stays unclassified — the +/// caller then falls back to plain retry behavior, never a wrong state. +pub fn is_object_lock_denied_delete(code: Option<&str>, message: Option<&str>) -> bool { + if !matches!(code, Some("AccessDenied")) { + return false; + } + let Some(message) = message else { + return false; + }; + let message = message.to_ascii_lowercase(); + message.contains("retention") || message.contains("legal hold") +} + fn admitted_target_arns_from_replication_state(state: &ReplicationState) -> Vec { let mut target_arns = state.targets.keys().cloned().collect::>(); target_arns.extend(state.purge_targets.keys().cloned()); @@ -237,9 +258,9 @@ mod tests { use super::{ DeletedObjectReplicationInfo, delete_marker_purge_mrf_entry, delete_marker_purge_version_id, - delete_replication_creates_marker, is_retryable_delete_replication_head_error, is_version_delete_replication, - replicate_delete_outcome, resync_existing_delete_replication_info, should_retry_delete_marker_purge, - target_delete_version_id, + delete_replication_creates_marker, is_object_lock_denied_delete, is_retryable_delete_replication_head_error, + is_version_delete_replication, replicate_delete_outcome, resync_existing_delete_replication_info, + should_retry_delete_marker_purge, target_delete_version_id, }; use crate::storage_api::DeletedObject; use crate::{ @@ -615,4 +636,24 @@ mod tests { corrupt.target_delete_marker_version_ids_corrupt = true; assert_eq!(delete_marker_purge_version_id(Some(&corrupt), arn, source), None); } + + #[test] + fn object_lock_denied_delete_is_recognized_by_code_and_reason() { + // The peer's deletion gate answers AccessDenied with the lock reason. + assert!(is_object_lock_denied_delete( + Some("AccessDenied"), + Some("Object is under GOVERNANCE retention and cannot be deleted until 2026-09-01T00:00:00Z") + )); + assert!(is_object_lock_denied_delete( + Some("AccessDenied"), + Some("Object has a legal hold and cannot be deleted. Remove the legal hold first.") + )); + + // A plain policy denial (misconfigured replicator) is not a lock denial. + assert!(!is_object_lock_denied_delete(Some("AccessDenied"), Some("Access Denied."))); + assert!(!is_object_lock_denied_delete(Some("AccessDenied"), None)); + // Other errors mentioning retention must not match. + assert!(!is_object_lock_denied_delete(Some("InternalError"), Some("retention lookup failed"))); + assert!(!is_object_lock_denied_delete(None, Some("legal hold"))); + } } diff --git a/crates/replication/src/lib.rs b/crates/replication/src/lib.rs index 6919f0a60..98464e7de 100644 --- a/crates/replication/src/lib.rs +++ b/crates/replication/src/lib.rs @@ -41,9 +41,9 @@ pub use config::{ }; pub use delete::{ DeletedObjectReplicationInfo, delete_marker_purge_mrf_entry, delete_marker_purge_version_id, - delete_replication_creates_marker, is_retryable_delete_replication_head_error, is_version_delete_replication, - replicate_delete_outcome, resync_existing_delete_replication_info, should_retry_delete_marker_purge, - target_delete_version_id, + delete_replication_creates_marker, is_object_lock_denied_delete, is_retryable_delete_replication_head_error, + is_version_delete_replication, replicate_delete_outcome, resync_existing_delete_replication_info, + should_retry_delete_marker_purge, target_delete_version_id, }; pub use filemeta::{ NULL_VERSION_ID, REPLICATE_EXISTING, REPLICATE_EXISTING_DELETE, REPLICATE_HEAL, REPLICATE_HEAL_DELETE, REPLICATE_INCOMING,