From 380ec74ece7c1aa0000ca63c0d0fbd1ede303f0b Mon Sep 17 00:00:00 2001 From: cxymds Date: Mon, 3 Aug 2026 09:44:28 +0800 Subject: [PATCH] fix(replication): persist force-delete handoff state (#5641) * fix(replication): persist force-delete handoff state * fix(arch): route force-delete config access through boundary * style: format force-delete imports --------- Co-authored-by: Zhengchao An --- crates/ecstore/src/api/mod.rs | 11 +- crates/ecstore/src/bucket/replication/mod.rs | 5 +- .../replication/replication_config_store.rs | 14 + .../replication_metadata_boundary.rs | 5 + .../replication/replication_object_bridge.rs | 7 + .../replication/replication_object_config.rs | 10 + .../bucket/replication/replication_pool.rs | 267 +++++++++++++++++- .../replication/replication_resyncer.rs | 158 ++++++----- .../replication_storage_boundary.rs | 3 + crates/replication/src/config.rs | 70 ++++- crates/replication/src/delete.rs | 47 ++- crates/replication/src/filemeta.rs | 15 +- crates/replication/src/mrf.rs | 9 + crates/replication/src/storage_api.rs | 3 + crates/storage-api/src/object.rs | 3 + rustfs/src/admin/handlers/replication.rs | 3 + rustfs/src/app/object_usecase.rs | 60 +++- rustfs/src/app/storage_api.rs | 45 +++ rustfs/src/storage/storage_api.rs | 15 + 19 files changed, 662 insertions(+), 88 deletions(-) diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index 952409d1f..e67cffadc 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/crates/ecstore/src/api/mod.rs @@ -188,11 +188,12 @@ pub mod bucket { 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, + VersionPurgeStatusType, commit_force_delete_intent, complete_force_delete_intent, + 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, + persist_force_delete_intent, 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, unsupported_replication_config_field, validate_replication_config_target_arns, version_purge_status_to_filemeta, }; diff --git a/crates/ecstore/src/bucket/replication/mod.rs b/crates/ecstore/src/bucket/replication/mod.rs index faf24b3e4..19cb5f702 100644 --- a/crates/ecstore/src/bucket/replication/mod.rs +++ b/crates/ecstore/src/bucket/replication/mod.rs @@ -70,8 +70,9 @@ pub use replication_object_decision_boundary::{ should_use_existing_delete_replication_source, }; pub use replication_pool::{ - DurableMrfBacklog, DynReplicationPool, ReplicationPoolTrait, get_global_replication_pool, get_global_replication_stats, - init_background_replication, read_durable_mrf_backlog, resync_start_conflict_id, + DurableMrfBacklog, DynReplicationPool, ReplicationPoolTrait, commit_force_delete_intent, complete_force_delete_intent, + get_global_replication_pool, get_global_replication_stats, init_background_replication, persist_force_delete_intent, + read_durable_mrf_backlog, resync_start_conflict_id, }; pub use replication_queue_boundary::{ DeletedObjectReplicationInfo, ReplicationBatchAdmission, ReplicationHealQueueResult, ReplicationOperation, diff --git a/crates/ecstore/src/bucket/replication/replication_config_store.rs b/crates/ecstore/src/bucket/replication/replication_config_store.rs index 459f2e4b4..28a90a83b 100644 --- a/crates/ecstore/src/bucket/replication/replication_config_store.rs +++ b/crates/ecstore/src/bucket/replication/replication_config_store.rs @@ -30,10 +30,24 @@ impl ReplicationConfigStore { com::read_config(api, file).await } + pub(crate) async fn read_no_lock(api: Arc, file: &str) -> Result> + where + S: ReplicationObjectIO, + { + com::read_config_no_lock(api, file).await + } + pub(crate) async fn save(api: Arc, file: &str, data: Vec) -> Result<()> where S: ReplicationObjectIO, { com::save_config(api, file, data).await } + + pub(crate) async fn save_no_lock(api: Arc, file: &str, data: Vec) -> Result<()> + where + S: ReplicationObjectIO, + { + com::save_config_no_lock(api, file, data).await + } } diff --git a/crates/ecstore/src/bucket/replication/replication_metadata_boundary.rs b/crates/ecstore/src/bucket/replication/replication_metadata_boundary.rs index 2d2273b2b..64f6c362f 100644 --- a/crates/ecstore/src/bucket/replication/replication_metadata_boundary.rs +++ b/crates/ecstore/src/bucket/replication/replication_metadata_boundary.rs @@ -32,6 +32,7 @@ pub(crate) struct ReplicationMetadataStore; impl ReplicationMetadataStore { pub(crate) const MRF_REPLICATION_FILE: &'static str = "config/replication/mrf.bin"; + pub(crate) const FORCE_DELETE_REPLICATION_FILE: &'static str = "config/replication/force-delete.bin"; pub(crate) async fn replication_config(bucket: &str) -> Result<(ReplicationConfiguration, OffsetDateTime)> { metadata_sys::get_replication_config(bucket).await @@ -109,5 +110,9 @@ mod tests { "buckets/bucket-a/.replication/resync.bin" ); assert_eq!(ReplicationMetadataStore::MRF_REPLICATION_FILE, "config/replication/mrf.bin"); + assert_eq!( + ReplicationMetadataStore::FORCE_DELETE_REPLICATION_FILE, + "config/replication/force-delete.bin" + ); } } diff --git a/crates/ecstore/src/bucket/replication/replication_object_bridge.rs b/crates/ecstore/src/bucket/replication/replication_object_bridge.rs index 0f00a1658..07b126c42 100644 --- a/crates/ecstore/src/bucket/replication/replication_object_bridge.rs +++ b/crates/ecstore/src/bucket/replication/replication_object_bridge.rs @@ -89,6 +89,13 @@ impl ReplicationObjectBridge { snapshot.has_active_rule(object) } + pub fn force_delete_target_set( + snapshot: &DeleteReplicationConfigSnapshot, + prefix: &str, + ) -> Option<(Vec, time::OffsetDateTime)> { + snapshot.force_delete_target_set(prefix) + } + pub fn check_delete_with_snapshot( object: &ObjectToDelete, source: &ObjectInfo, diff --git a/crates/ecstore/src/bucket/replication/replication_object_config.rs b/crates/ecstore/src/bucket/replication/replication_object_config.rs index 43f8e77de..c09db31ce 100644 --- a/crates/ecstore/src/bucket/replication/replication_object_config.rs +++ b/crates/ecstore/src/bucket/replication/replication_object_config.rs @@ -18,6 +18,7 @@ use crate::bucket::metadata::BucketMetadata; use rustfs_utils::http::AMZ_BUCKET_REPLICATION_STATUS; use s3s::dto::{BucketVersioningStatus, ReplicationConfiguration, ReplicationRuleStatus, VersioningConfiguration}; use serde::{Deserialize, Serialize}; +use time::OffsetDateTime; use tracing::error; use super::replication_config_boundary::{ @@ -83,6 +84,15 @@ impl DeleteReplicationConfigSnapshot { .and_then(|metadata| metadata.replication_config.as_ref()) } + pub(crate) fn force_delete_target_set(&self, prefix: &str) -> Option<(Vec, OffsetDateTime)> { + self.metadata.as_ref().and_then(|metadata| { + metadata + .replication_config + .as_ref() + .map(|config| (config.filter_force_delete_target_arns(prefix), metadata.replication_config_updated_at)) + }) + } + pub(crate) fn has_active_rule(&self, object: &str) -> bool { self.replication_config() .is_some_and(|config| config.has_active_rules(object, true)) diff --git a/crates/ecstore/src/bucket/replication/replication_pool.rs b/crates/ecstore/src/bucket/replication/replication_pool.rs index f16d1eb32..3159452cf 100644 --- a/crates/ecstore/src/bucket/replication/replication_pool.rs +++ b/crates/ecstore/src/bucket/replication/replication_pool.rs @@ -136,6 +136,10 @@ static DURABLE_MRF_TARGET_BACKLOG: LazyLock> = LazyLock::new(|| StdRwLock::new(MrfBacklogObservabilityTracker::default())); +fn should_replay_force_delete_intent(entry: &MrfReplicateEntry) -> bool { + entry.force_delete_id.is_some() && entry.force_delete_local_commit && !entry.target_arns.is_empty() +} + #[derive(Debug, Clone, Default)] struct DurableMrfBacklogTracker { available: bool, @@ -302,6 +306,7 @@ where delete_marker: false, delete_marker_mtime: None, target_arns: Vec::new(), + ..Default::default() }); } tracker.into_snapshot() @@ -394,6 +399,82 @@ pub async fn read_durable_mrf_backlog(storage: Arc) - durable_mrf_backlog_from_read(ReplicationConfigStore::read(storage, ReplicationMetadataStore::MRF_REPLICATION_FILE).await) } +pub async fn persist_force_delete_intent( + storage: Arc, + mut entry: MrfReplicateEntry, +) -> Result<(), EcstoreError> { + entry.force_delete_local_commit = false; + let file = ReplicationMetadataStore::FORCE_DELETE_REPLICATION_FILE; + let lock = storage + .new_ns_lock(ReplicationMetadataStore::rustfs_meta_bucket(), file) + .await?; + let _guard = lock.get_write_lock(ReplicationLockTiming::acquire_timeout()).await?; + + let mut entries = match ReplicationConfigStore::read_no_lock(storage.clone(), file).await { + Ok(data) => decode_mrf_file(&data)?, + Err(EcstoreError::ConfigNotFound) => Vec::new(), + Err(err) => return Err(err), + }; + + if entries + .iter() + .any(|existing| existing.force_delete_id == entry.force_delete_id) + { + return Ok(()); + } + + entries.push(entry); + let data = encode_mrf_file(&entries)?; + ReplicationConfigStore::save_no_lock(storage, file, data).await +} + +pub async fn commit_force_delete_intent( + storage: Arc, + operation_id: uuid::Uuid, +) -> Result<(), EcstoreError> { + let file = ReplicationMetadataStore::FORCE_DELETE_REPLICATION_FILE; + let lock = storage + .new_ns_lock(ReplicationMetadataStore::rustfs_meta_bucket(), file) + .await?; + let _guard = lock.get_write_lock(ReplicationLockTiming::acquire_timeout()).await?; + + let data = ReplicationConfigStore::read_no_lock(storage.clone(), file).await?; + let mut entries = decode_mrf_file(&data)?; + let Some(entry) = entries.iter_mut().find(|entry| entry.force_delete_id == Some(operation_id)) else { + return Err(EcstoreError::ConfigNotFound); + }; + if entry.force_delete_local_commit { + return Ok(()); + } + entry.force_delete_local_commit = true; + ReplicationConfigStore::save_no_lock(storage, file, encode_mrf_file(&entries)?).await +} + +pub async fn complete_force_delete_intent( + storage: Arc, + operation_id: uuid::Uuid, +) -> Result<(), EcstoreError> { + let file = ReplicationMetadataStore::FORCE_DELETE_REPLICATION_FILE; + let lock = storage + .new_ns_lock(ReplicationMetadataStore::rustfs_meta_bucket(), file) + .await?; + let _guard = lock.get_write_lock(ReplicationLockTiming::acquire_timeout()).await?; + + let data = match ReplicationConfigStore::read_no_lock(storage.clone(), file).await { + Ok(data) => data, + Err(EcstoreError::ConfigNotFound) => return Ok(()), + Err(err) => return Err(err), + }; + let mut entries = decode_mrf_file(&data)?; + let original_len = entries.len(); + entries.retain(|entry| entry.force_delete_id != Some(operation_id)); + if entries.len() == original_len { + return Ok(()); + } + + ReplicationConfigStore::save_no_lock(storage, file, encode_mrf_file(&entries)?).await +} + #[derive(Debug, thiserror::Error)] #[error("replication resync {active_resync_id} is already active for {bucket}/{arn}")] struct ResyncActiveConflictError { @@ -497,6 +578,7 @@ impl ReplicationPool { // Start background tasks pool.start_mrf_processor().await; + pool.start_force_delete_processor().await; pool.start_mrf_persister().await; pool @@ -972,6 +1054,32 @@ impl ReplicationPool { for entry in entries.iter() { match entry.op { MrfOpKind::Delete => { + if should_replay_force_delete_intent(entry) { + let Some(operation_id) = entry.force_delete_id else { + continue; + }; + schedule_replication_delete(DeletedObjectReplicationInfo { + delete_object: ReplicationDeletedObject { + object_name: entry.object.clone(), + force_delete: true, + force_delete_id: Some(operation_id), + force_delete_target_arns: entry.target_arns.clone(), + force_delete_generation: entry.force_delete_generation, + ..Default::default() + }, + bucket: entry.bucket.clone(), + op_type: ReplicationType::Heal, + event_type: REPLICATE_HEAL_DELETE.to_string(), + ..Default::default() + }) + .await; + queued_count += 1; + continue; + } + if entry.force_delete_id.is_some() { + continue; + } + // Reconstruct a heal delete and re-queue it. We do NOT call // get_object_info here because the delete-marker or version may // already be absent from the local store — that is expected. @@ -1131,6 +1239,67 @@ impl ReplicationPool { self.task_handles.lock().await.push(handle); } + async fn start_force_delete_processor(&self) { + let storage = self.storage.clone(); + let handle = + tokio::spawn(async move { + let data = + match ReplicationConfigStore::read(storage.clone(), ReplicationMetadataStore::FORCE_DELETE_REPLICATION_FILE) + .await + { + Ok(data) => data, + Err(EcstoreError::ConfigNotFound) => return, + Err(error) => { + warn!( + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION, + error = %error, + "Failed to load durable force-delete intents" + ); + return; + } + }; + + let entries = match decode_mrf_file(&data) { + Ok(entries) => entries, + Err(error) => { + warn!( + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION, + error = %error, + "Failed to decode durable force-delete intents" + ); + return; + } + }; + + for entry in entries { + if !should_replay_force_delete_intent(&entry) { + continue; + } + let Some(operation_id) = entry.force_delete_id else { + continue; + }; + schedule_replication_delete(DeletedObjectReplicationInfo { + delete_object: ReplicationDeletedObject { + object_name: entry.object, + force_delete: true, + force_delete_id: Some(operation_id), + force_delete_target_arns: entry.target_arns, + force_delete_generation: entry.force_delete_generation, + ..Default::default() + }, + bucket: entry.bucket, + op_type: ReplicationType::Heal, + event_type: REPLICATE_HEAL_DELETE.to_string(), + ..Default::default() + }) + .await; + } + }); + self.task_handles.lock().await.push(handle); + } + /// Starts the MRF persister — ongoing background task. /// /// Drains `mrf_save_rx` (entries that overflowed the normal worker channels) and @@ -2216,7 +2385,7 @@ mod tests { _h: Self::HeaderMap, _opts: &Self::ObjectOptions, ) -> Result { - if !object.ends_with("/.replication/resync.bin") { + if !object.ends_with("/.replication/resync.bin") && !object.ends_with("config/replication/force-delete.bin") { return Err(EcstoreError::FileNotFound); } @@ -3104,6 +3273,7 @@ mod tests { delete_marker: false, delete_marker_mtime: None, target_arns: Vec::new(), + ..Default::default() }; let second = MrfReplicateEntry { object: "second".to_string(), @@ -3202,6 +3372,7 @@ mod tests { delete_marker: false, delete_marker_mtime: None, target_arns: Vec::new(), + ..Default::default() }, "test", ) @@ -3234,6 +3405,7 @@ mod tests { delete_marker: false, delete_marker_mtime: None, target_arns: Vec::new(), + ..Default::default() }; observe_mrf_pending(&entry); @@ -3289,6 +3461,7 @@ mod tests { delete_marker: false, delete_marker_mtime: None, target_arns: vec!["arn:rustfs:replication:target-a".to_string()], + ..Default::default() }; stats.inc_q(&entry.bucket, entry.size, false, ReplicationType::Heal); @@ -3312,6 +3485,7 @@ mod tests { delete_marker: false, delete_marker_mtime: None, target_arns: Vec::new(), + ..Default::default() }; let second = MrfReplicateEntry { object: "second".to_string(), @@ -3425,6 +3599,7 @@ mod tests { delete_marker: false, delete_marker_mtime: None, target_arns: Vec::new(), + ..Default::default() }; let encoded = encode_mrf_file(std::slice::from_ref(&entry)).expect("encode"); @@ -3460,6 +3635,7 @@ mod tests { delete_marker: true, delete_marker_mtime: Some(mtime_nanos), target_arns: Vec::new(), + ..Default::default() }; let encoded = encode_mrf_file(std::slice::from_ref(&entry)).expect("encode"); @@ -3495,6 +3671,7 @@ mod tests { delete_marker: false, delete_marker_mtime: None, target_arns: Vec::new(), + ..Default::default() }; let encoded = encode_mrf_file(&[entry]).expect("encode"); @@ -3525,6 +3702,7 @@ mod tests { delete_marker: false, delete_marker_mtime: None, target_arns: Vec::new(), + ..Default::default() }, MrfReplicateEntry { bucket: "b".to_string(), @@ -3538,6 +3716,7 @@ mod tests { delete_marker: true, delete_marker_mtime: None, target_arns: Vec::new(), + ..Default::default() }, ]; @@ -3569,6 +3748,7 @@ mod tests { delete_marker: false, delete_marker_mtime: None, target_arns: Vec::new(), + ..Default::default() }; assert_eq!(obj_entry.op, MrfOpKind::Object); @@ -3585,6 +3765,7 @@ mod tests { delete_marker: true, delete_marker_mtime: None, target_arns: Vec::new(), + ..Default::default() }; assert_eq!(del_entry.op, MrfOpKind::Delete); @@ -3602,6 +3783,7 @@ mod tests { delete_marker: false, delete_marker_mtime: None, target_arns: Vec::new(), + ..Default::default() }; assert_eq!(legacy_entry.op, MrfOpKind::Object, "legacy default must be Object"); } @@ -3668,6 +3850,7 @@ mod tests { delete_marker: false, delete_marker_mtime: None, target_arns: Vec::new(), + ..Default::default() }]; let encoded = encode_mrf_file(&entries).expect("durable MRF backlog should encode"); @@ -3716,6 +3899,7 @@ mod tests { delete_marker: false, delete_marker_mtime: None, target_arns: vec!["arn:target-a".to_string(), "arn:target-b".to_string()], + ..Default::default() }, MrfReplicateEntry { bucket: "b1".to_string(), @@ -3729,6 +3913,7 @@ mod tests { delete_marker: false, delete_marker_mtime: None, target_arns: vec!["arn:target-a".to_string()], + ..Default::default() }, MrfReplicateEntry { bucket: "b1".to_string(), @@ -3742,6 +3927,7 @@ mod tests { delete_marker: false, delete_marker_mtime: None, target_arns: Vec::new(), + ..Default::default() }, ]; @@ -3797,10 +3983,89 @@ mod tests { delete_marker: false, delete_marker_mtime: None, target_arns: Vec::new(), + ..Default::default() }]) .expect("invalid persisted entry should still encode for boundary testing"); let invalid = durable_mrf_backlog_from_read(Ok(negative)); assert!(!invalid.available); assert!(invalid.entries.is_empty()); } + + #[test] + fn force_delete_replay_requires_local_commit_and_keeps_persisted_targets() { + let operation_id = Uuid::new_v4(); + let pending = MrfReplicateEntry { + bucket: "source".to_string(), + object: "logs/".to_string(), + target_arns: vec!["arn:target:old-generation".to_string()], + force_delete_id: Some(operation_id), + force_delete_generation: Some(11), + force_delete_local_commit: false, + op: MrfOpKind::Delete, + ..Default::default() + }; + assert!(!should_replay_force_delete_intent(&pending)); + + let mut committed = pending; + committed.force_delete_local_commit = true; + let recovered = decode_mrf_file(&encode_mrf_file(&[committed.clone()]).expect("force-delete intent should encode")) + .expect("force-delete intent should decode"); + assert!(should_replay_force_delete_intent(&recovered[0])); + assert_eq!(recovered[0].target_arns, vec!["arn:target:old-generation"]); + assert_eq!(recovered[0].force_delete_generation, Some(11)); + + committed.target_arns.clear(); + assert!(!should_replay_force_delete_intent(&committed)); + } + + #[tokio::test] + async fn force_delete_intent_append_commit_and_cleanup_are_idempotent() { + let shared = empty_resync_shared_state(); + let storage = Arc::new(LoadResyncNodeStore::new("force-delete-journal", shared)); + let operation_id = Uuid::new_v4(); + let entry = MrfReplicateEntry { + bucket: "source".to_string(), + object: "logs/".to_string(), + target_arns: vec!["arn:target:stable".to_string()], + force_delete_id: Some(operation_id), + force_delete_generation: Some(12), + op: MrfOpKind::Delete, + ..Default::default() + }; + + persist_force_delete_intent(storage.clone(), entry.clone()) + .await + .expect("first journal append should succeed"); + persist_force_delete_intent(storage.clone(), entry) + .await + .expect("duplicate journal append should be a no-op"); + + let data = ReplicationConfigStore::read(storage.clone(), ReplicationMetadataStore::FORCE_DELETE_REPLICATION_FILE) + .await + .expect("journal should be readable"); + let entries = decode_mrf_file(&data).expect("journal should decode"); + assert_eq!(entries.len(), 1); + assert!(!entries[0].force_delete_local_commit); + assert!(!should_replay_force_delete_intent(&entries[0])); + + commit_force_delete_intent(storage.clone(), operation_id) + .await + .expect("commit marker should persist"); + commit_force_delete_intent(storage.clone(), operation_id) + .await + .expect("duplicate commit marker should be a no-op"); + let data = ReplicationConfigStore::read(storage.clone(), ReplicationMetadataStore::FORCE_DELETE_REPLICATION_FILE) + .await + .expect("committed journal should be readable"); + let entries = decode_mrf_file(&data).expect("committed journal should decode"); + assert!(should_replay_force_delete_intent(&entries[0])); + assert_eq!(entries[0].target_arns, vec!["arn:target:stable"]); + + complete_force_delete_intent(storage.clone(), operation_id) + .await + .expect("journal cleanup should succeed"); + complete_force_delete_intent(storage, operation_id) + .await + .expect("duplicate journal cleanup should be a no-op"); + } } diff --git a/crates/ecstore/src/bucket/replication/replication_resyncer.rs b/crates/ecstore/src/bucket/replication/replication_resyncer.rs index 240c356ad..b2f070de8 100644 --- a/crates/ecstore/src/bucket/replication/replication_resyncer.rs +++ b/crates/ecstore/src/bucket/replication/replication_resyncer.rs @@ -1642,54 +1642,62 @@ async fn replicate_delete_marker_purge_to_targets(bucket: &str, dobj: &DeletedOb async fn replicate_force_delete_to_targets(dobj: &DeletedObjectReplicationInfo, storage: Arc) { let bucket = &dobj.bucket; let object_name = &dobj.delete_object.object_name; + let admitted_target_arns = dobj.admitted_target_arns(); - let rcfg = match get_replication_config(bucket).await { - Ok(Some(config)) => config, - Ok(None) => { - debug!( - event = EVENT_REPLICATION_FORCE_DELETE_SKIPPED, - component = LOG_COMPONENT_ECSTORE, - subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, - bucket = %bucket, - reason = "replication_config_missing", - "Skipping replication force-delete because replication config is missing" - ); - send_local_event(EventArgs { - event_name: EventName::ObjectReplicationNotTracked.to_string(), - bucket_name: bucket.clone(), - object: ObjectInfo { - bucket: bucket.clone(), - name: object_name.clone(), - ..Default::default() - }, - user_agent: "Internal: [Replication]".to_string(), + let legacy_target_arns = if admitted_target_arns.is_empty() { + match get_replication_config(bucket).await { + Ok(Some(config)) => config.filter_target_arns(&ObjectOpts { + name: object_name.clone(), ..Default::default() - }); - return; - } - Err(err) => { - debug!( - event = EVENT_REPLICATION_FORCE_DELETE_SKIPPED, - component = LOG_COMPONENT_ECSTORE, - subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, - bucket = %bucket, - error = %err, - reason = "replication_config_lookup_failed", - "Skipping replication force-delete because replication config lookup failed" - ); - send_local_event(EventArgs { - event_name: EventName::ObjectReplicationNotTracked.to_string(), - bucket_name: bucket.clone(), - object: ObjectInfo { - bucket: bucket.clone(), - name: object_name.clone(), + }), + Ok(None) => { + debug!( + event = EVENT_REPLICATION_FORCE_DELETE_SKIPPED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, + bucket = %bucket, + reason = "replication_config_missing", + "Skipping replication force-delete because replication config is missing" + ); + send_local_event(EventArgs { + event_name: EventName::ObjectReplicationNotTracked.to_string(), + bucket_name: bucket.clone(), + object: ObjectInfo { + bucket: bucket.clone(), + name: object_name.clone(), + ..Default::default() + }, + user_agent: "Internal: [Replication]".to_string(), ..Default::default() - }, - user_agent: "Internal: [Replication]".to_string(), - ..Default::default() - }); - return; + }); + Vec::new() + } + Err(err) => { + debug!( + event = EVENT_REPLICATION_FORCE_DELETE_SKIPPED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, + bucket = %bucket, + error = %err, + reason = "replication_config_lookup_failed", + "Skipping replication force-delete because replication config lookup failed" + ); + send_local_event(EventArgs { + event_name: EventName::ObjectReplicationNotTracked.to_string(), + bucket_name: bucket.clone(), + object: ObjectInfo { + bucket: bucket.clone(), + name: object_name.clone(), + ..Default::default() + }, + user_agent: "Internal: [Replication]".to_string(), + ..Default::default() + }); + Vec::new() + } } + } else { + Vec::new() }; let ns_lock = match storage @@ -1751,22 +1759,18 @@ async fn replicate_force_delete_to_targets(dobj: &Deleted } }; - let tgt_arns = { - let admitted = dobj.admitted_target_arns(); - if admitted.is_empty() { - rcfg.filter_target_arns(&ObjectOpts { - name: object_name.clone(), - ..Default::default() - }) - } else { - admitted - } + let tgt_arns = if admitted_target_arns.is_empty() { + legacy_target_arns + } else { + admitted_target_arns }; let mut join_set = JoinSet::new(); + let mut all_succeeded = true; for arn in tgt_arns { let Some(tgt_client) = ReplicationTargetStore::remote_target_client(bucket, &arn).await else { + all_succeeded = false; debug!( event = EVENT_REPLICATION_FORCE_DELETE_SKIPPED, component = LOG_COMPONENT_ECSTORE, @@ -1816,7 +1820,7 @@ async fn replicate_force_delete_to_targets(dobj: &Deleted user_agent: "Internal: [Replication]".to_string(), ..Default::default() }); - return; + return false; } if let Err(e) = tgt_client @@ -1845,24 +1849,46 @@ async fn replicate_force_delete_to_targets(dobj: &Deleted user_agent: "Internal: [Replication]".to_string(), ..Default::default() }); + return false; } + + true }); } while let Some(result) = join_set.join_next().await { - if let Err(e) = result { - error!( - event = EVENT_RESYNC_TASK_FAILED, - component = LOG_COMPONENT_ECSTORE, - subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, - bucket = %bucket, - object = %object_name, - operation = "force_delete", - error = %e, - "Replication resync task failed" - ); + match result { + Ok(success) => all_succeeded &= success, + Err(error) => { + all_succeeded = false; + error!( + event = EVENT_RESYNC_TASK_FAILED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, + bucket = %bucket, + object = %object_name, + operation = "force_delete", + error = %error, + "Replication resync task failed" + ); + } } } + + if all_succeeded + && let Some(operation_id) = dobj.delete_object.force_delete_id + && let Err(error) = super::replication_pool::complete_force_delete_intent(storage, operation_id).await + { + warn!( + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, + bucket = %bucket, + object = %object_name, + operation_id = %operation_id, + error = %error, + "Force-delete replication completed but durable intent cleanup failed" + ); + } } fn target_delete_version_id(version_id: Uuid, version_purge: bool) -> Option { diff --git a/crates/ecstore/src/bucket/replication/replication_storage_boundary.rs b/crates/ecstore/src/bucket/replication/replication_storage_boundary.rs index 0b729d9ba..315b88e15 100644 --- a/crates/ecstore/src/bucket/replication/replication_storage_boundary.rs +++ b/crates/ecstore/src/bucket/replication/replication_storage_boundary.rs @@ -105,6 +105,9 @@ pub(crate) fn deleted_object_for_replication(delete_object: DeletedObject) -> Re replication_state: delete_object.replication_state.as_ref().map(replication_state_from_filemeta), found: delete_object.found, force_delete: delete_object.force_delete, + force_delete_id: delete_object.force_delete_id, + force_delete_target_arns: delete_object.force_delete_target_arns, + force_delete_generation: delete_object.force_delete_generation, } } diff --git a/crates/replication/src/config.rs b/crates/replication/src/config.rs index fad2b018a..ddfdad183 100644 --- a/crates/replication/src/config.rs +++ b/crates/replication/src/config.rs @@ -23,7 +23,7 @@ use s3s::dto::{ ReplicationRuleStatus, ReplicationRules, }; use serde::{Deserialize, Serialize}; -use std::collections::{HashMap, HashSet}; +use std::collections::{BTreeMap, HashMap, HashSet}; use uuid::Uuid; pub const REPLICATION_CAPABILITY_CONTRACT_VERSION: u32 = 1; @@ -93,6 +93,7 @@ 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; + fn filter_force_delete_target_arns(&self, prefix: &str) -> Vec; fn filter_target_replication_decisions(&self, obj: &ObjectOpts) -> Vec<(String, bool)> { self.filter_target_arns(obj) .into_iter() @@ -434,6 +435,51 @@ impl ReplicationConfigurationExt for ReplicationConfiguration { arns } + fn filter_force_delete_target_arns(&self, prefix: &str) -> Vec { + let role = self.role.trim(); + let mut selected = BTreeMap::::new(); + + for rule in &self.rules { + if rule.status == ReplicationRuleStatus::from_static(ReplicationRuleStatus::DISABLED) { + continue; + } + + let rule_prefix = rule.prefix(); + if !prefix.starts_with(rule_prefix) && !rule_prefix.starts_with(prefix) { + continue; + } + + let target = if role.is_empty() { + rule.destination.bucket.trim() + } else { + role + }; + if target.is_empty() { + continue; + } + + let delete_enabled = + rule.delete_replication.as_ref().is_some_and(|delete| { + delete.status == DeleteReplicationStatus::from_static(DeleteReplicationStatus::ENABLED) + }) || rule.delete_marker_replication.as_ref().is_some_and(|delete_marker| { + delete_marker.status + == Some(DeleteMarkerReplicationStatus::from_static(DeleteMarkerReplicationStatus::ENABLED)) + }); + + if selected + .get(target) + .is_none_or(|(current, _)| rule.priority > current.priority) + { + selected.insert(target.to_string(), (rule, delete_enabled)); + } + } + + selected + .into_iter() + .filter_map(|(target, (_, enabled))| enabled.then_some(target)) + .collect() + } + fn filter_target_replication_decisions(&self, obj: &ObjectOpts) -> Vec<(String, bool)> { let rules = self.filter_actionable_rules(obj); let role = self.role.trim(); @@ -1069,4 +1115,26 @@ mod tests { assert_eq!(decisions, vec![(target_a.to_string(), false), (target_b.to_string(), true)]); } + + #[test] + fn force_delete_targets_use_overlapping_rules_and_highest_priority_switch() { + let target_a = "arn:target:a"; + let target_b = "arn:target:b"; + let mut a_parent = delete_marker_rule("a-parent", target_a, "logs/", 1, true); + a_parent.delete_replication = Some(DeleteReplication { + status: DeleteReplicationStatus::from_static(DeleteReplicationStatus::ENABLED), + }); + let a_child_disabled = delete_marker_rule("a-child", target_a, "logs/2026/", 5, false); + let b_child = delete_marker_rule("b-child", target_b, "logs/2026/", 2, true); + let config = ReplicationConfiguration { + role: String::new(), + rules: vec![a_parent, a_child_disabled, b_child], + }; + + assert_eq!( + config.filter_force_delete_target_arns("logs/2026/app.log"), + vec![target_b.to_string()], + "the child rule must win for target A while the overlapping child target B remains eligible" + ); + } } diff --git a/crates/replication/src/delete.rs b/crates/replication/src/delete.rs index e4fbfd7ba..3a0c32bf8 100644 --- a/crates/replication/src/delete.rs +++ b/crates/replication/src/delete.rs @@ -33,12 +33,15 @@ impl DeletedObjectReplicationInfo { return vec![self.target_arn.clone()]; } - let mut target_arns = self - .delete_object - .replication_state - .as_ref() - .map(admitted_target_arns_from_replication_state) - .unwrap_or_default(); + let mut target_arns = if !self.delete_object.force_delete_target_arns.is_empty() { + self.delete_object.force_delete_target_arns.clone() + } else { + self.delete_object + .replication_state + .as_ref() + .map(admitted_target_arns_from_replication_state) + .unwrap_or_default() + }; target_arns.sort(); target_arns.dedup(); target_arns @@ -69,6 +72,9 @@ impl ReplicationWorkerOperation for DeletedObjectReplicationInfo { .delete_marker_mtime .and_then(|t| i64::try_from(t.unix_timestamp_nanos()).ok()), target_arns: self.admitted_target_arns(), + force_delete_id: self.delete_object.force_delete_id, + force_delete_generation: self.delete_object.force_delete_generation, + force_delete_local_commit: self.delete_object.force_delete, } } @@ -231,6 +237,35 @@ mod tests { ); } + #[test] + fn deleted_object_replication_info_preserves_force_delete_handoff() { + let operation_id = Uuid::new_v4(); + let info = DeletedObjectReplicationInfo { + bucket: "bucket".to_string(), + delete_object: DeletedObject { + object_name: "prefix/".to_string(), + force_delete: true, + force_delete_id: Some(operation_id), + force_delete_target_arns: vec![ + "arn:target-b".to_string(), + "arn:target-a".to_string(), + "arn:target-b".to_string(), + ], + force_delete_generation: Some(17), + ..Default::default() + }, + ..Default::default() + }; + + let entry = info.to_mrf_entry(); + + assert!(entry.force_delete); + assert_eq!(entry.force_delete_id, Some(operation_id)); + assert_eq!(entry.force_delete_generation, Some(17)); + assert!(entry.force_delete_local_commit); + assert_eq!(entry.target_arns, vec!["arn:target-a", "arn:target-b"]); + } + #[test] fn version_delete_replication_tracks_delete_marker_version_purge() { let dobj = DeletedObject { diff --git a/crates/replication/src/filemeta.rs b/crates/replication/src/filemeta.rs index cd4843a2a..2605353c9 100644 --- a/crates/replication/src/filemeta.rs +++ b/crates/replication/src/filemeta.rs @@ -572,7 +572,7 @@ impl MrfOpKind { } } -#[derive(Serialize, Deserialize, Debug, Clone)] +#[derive(Serialize, Deserialize, Debug, Clone, Default)] pub struct MrfReplicateEntry { #[serde(rename = "bucket")] pub bucket: String, @@ -619,6 +619,17 @@ pub struct MrfReplicateEntry { #[serde(rename = "targetARNs", skip_serializing_if = "Vec::is_empty", default)] pub target_arns: Vec, + + // Force-delete entries use the target ARN list above as their immutable target set. + // The id distinguishes a durable intent from legacy MRF delete entries. + #[serde(rename = "forceDeleteID", skip_serializing_if = "Option::is_none", default)] + pub force_delete_id: Option, + #[serde(rename = "forceDeleteGeneration", skip_serializing_if = "Option::is_none", default)] + pub force_delete_generation: Option, + + // Replay is allowed only after the source-side recursive delete has committed. + #[serde(rename = "forceDeleteLocalCommit", default)] + pub force_delete_local_commit: bool, } fn retry_count_to_mrf(retry_count: u32) -> i32 { @@ -865,6 +876,7 @@ impl ReplicationWorkerOperation for ReplicateObjectInfo { delete_marker: false, delete_marker_mtime: None, target_arns: self.admitted_target_arns(), + ..Default::default() } } @@ -945,6 +957,7 @@ impl ReplicateObjectInfo { delete_marker: false, delete_marker_mtime: None, target_arns: self.admitted_target_arns(), + ..Default::default() } } } diff --git a/crates/replication/src/mrf.rs b/crates/replication/src/mrf.rs index dc7ecefa0..0c30972b4 100644 --- a/crates/replication/src/mrf.rs +++ b/crates/replication/src/mrf.rs @@ -68,6 +68,9 @@ mod tests { retry_count: 1, size: 1024, op: MrfOpKind::Metadata, + force_delete_id: None, + force_delete_generation: None, + force_delete_local_commit: false, force_delete: false, delete_marker_version_id: None, delete_marker: false, @@ -81,6 +84,9 @@ mod tests { retry_count: 2, size: 1024, op: MrfOpKind::Object, + force_delete_id: None, + force_delete_generation: None, + force_delete_local_commit: false, force_delete: false, delete_marker_version_id: None, delete_marker: false, @@ -94,6 +100,9 @@ mod tests { retry_count: 0, size: 0, op: MrfOpKind::Delete, + force_delete_id: None, + force_delete_generation: None, + force_delete_local_commit: false, force_delete: true, delete_marker_version_id: Some(del_vid), delete_marker: true, diff --git a/crates/replication/src/storage_api.rs b/crates/replication/src/storage_api.rs index 2a4e4180a..66d4e6038 100644 --- a/crates/replication/src/storage_api.rs +++ b/crates/replication/src/storage_api.rs @@ -49,6 +49,9 @@ pub struct DeletedObject { pub replication_state: Option, pub found: bool, pub force_delete: bool, + pub force_delete_id: Option, + pub force_delete_target_arns: Vec, + pub force_delete_generation: Option, } impl DeletedObject { diff --git a/crates/storage-api/src/object.rs b/crates/storage-api/src/object.rs index 25523faa2..9f0957bde 100644 --- a/crates/storage-api/src/object.rs +++ b/crates/storage-api/src/object.rs @@ -213,6 +213,9 @@ pub struct DeletedObject { pub replication_state: Option, pub found: bool, pub force_delete: bool, + pub force_delete_id: Option, + pub force_delete_target_arns: Vec, + pub force_delete_generation: Option, } impl DeletedObject { diff --git a/rustfs/src/admin/handlers/replication.rs b/rustfs/src/admin/handlers/replication.rs index b5136762f..098778c58 100644 --- a/rustfs/src/admin/handlers/replication.rs +++ b/rustfs/src/admin/handlers/replication.rs @@ -1125,6 +1125,7 @@ mod tests { delete_marker: false, delete_marker_mtime: None, target_arns: vec!["arn-a".to_string(), "arn-durable-only".to_string()], + ..Default::default() }, MrfReplicateEntry { bucket: "other-bucket".to_string(), @@ -1138,6 +1139,7 @@ mod tests { delete_marker: false, delete_marker_mtime: None, target_arns: Vec::new(), + ..Default::default() }, ], }; @@ -1197,6 +1199,7 @@ mod tests { delete_marker: false, delete_marker_mtime: None, target_arns: Vec::new(), + ..Default::default() }], }; diff --git a/rustfs/src/app/object_usecase.rs b/rustfs/src/app/object_usecase.rs index 7e6d3c99e..62a804996 100644 --- a/rustfs/src/app/object_usecase.rs +++ b/rustfs/src/app/object_usecase.rs @@ -41,11 +41,11 @@ use super::storage_api::object_usecase::bucket::{ predict_lifecycle_expiration, quota::{QuotaCheckResult, QuotaError, QuotaOperation}, replication::{ - DeleteReplicationConfigSnapshot, REPLICATE_INCOMING_DELETE, ReplicationStatusType, delete_replication_state_from_config, - delete_replication_version_id, deleted_object_has_pending_replication_delete, has_active_delete_rule, - load_delete_config_snapshot, must_replicate_object, schedule_object_replication, schedule_replication_delete, - schedule_replication_deletes, set_deleted_object_replication_state, should_schedule_delete_replication, - should_use_existing_delete_replication_info, + DeleteReplicationConfigSnapshot, REPLICATE_INCOMING_DELETE, ReplicationStatusType, commit_force_delete_intent, + delete_replication_state_from_config, delete_replication_version_id, deleted_object_has_pending_replication_delete, + force_delete_target_set, has_active_delete_rule, load_delete_config_snapshot, must_replicate_object, + persist_force_delete_intent, schedule_object_replication, schedule_replication_delete, schedule_replication_deletes, + set_deleted_object_replication_state, should_schedule_delete_replication, should_use_existing_delete_replication_info, }, tagging::decode_tags, validate_restore_request, @@ -7137,6 +7137,7 @@ impl DefaultObjectUsecase { opts.expected_current_version_id = expected_current_version_id.clone(); let replicate_force_delete = force_delete && !replica && has_active_delete_rule(&delete_config_snapshot, &key); + let mut force_delete_intent = None; // Check Object Lock retention before deletion // TODO: Future optimization (separate PR) - If performance becomes critical under high delete load: @@ -7176,6 +7177,17 @@ impl DefaultObjectUsecase { let _ = invalidate_object_data_cache_before_mutation(&cache_adapter, &bucket, &key).await; } + if replicate_force_delete + && let Some((target_arns, generation)) = force_delete_target_set(&delete_config_snapshot, &key) + && !target_arns.is_empty() + { + let operation_id = + persist_force_delete_intent(store.clone(), bucket.clone(), key.clone(), target_arns.clone(), generation) + .await + .map_err(ApiError::from)?; + force_delete_intent = Some((operation_id, target_arns, generation)); + } + let obj_info = { match store .delete_object_with_tier_delete_journal(&bucket, &key, opts.clone()) @@ -7183,6 +7195,18 @@ impl DefaultObjectUsecase { { Ok(obj) => obj, Err(err) => { + if let Some((operation_id, _, _)) = force_delete_intent.as_ref() + && let Err(cleanup_error) = + crate::storage::storage_api::complete_force_delete_intent(store.clone(), *operation_id).await + { + warn!( + bucket = %bucket, + object = %key, + operation_id = %operation_id, + error = %cleanup_error, + "failed to remove uncommitted force-delete intent after local delete failure" + ); + } if is_err_bucket_not_found(&err) { return Err(S3Error::with_message(S3ErrorCode::NoSuchBucket, "Bucket not found".to_string())); } @@ -7211,7 +7235,31 @@ impl DefaultObjectUsecase { } if obj_info.name.is_empty() { - if replicate_force_delete { + if let Some((operation_id, target_arns, generation)) = force_delete_intent { + if let Err(error) = commit_force_delete_intent(store.clone(), operation_id).await { + warn!( + bucket = %bucket, + object = %key, + operation_id = %operation_id, + error = %error, + "failed to mark force-delete intent committed after local delete" + ); + } + let generation = i64::try_from(generation.unix_timestamp_nanos()).unwrap_or(i64::MAX); + schedule_replication_delete( + StorageDeletedObject { + object_name: key.clone(), + force_delete: true, + force_delete_id: Some(operation_id), + force_delete_target_arns: target_arns, + force_delete_generation: Some(generation), + ..Default::default() + }, + bucket.clone(), + REPLICATE_INCOMING_DELETE.to_string(), + ) + .await; + } else if replicate_force_delete { let mut delete_object = StorageDeletedObject { object_name: key.clone(), force_delete: true, diff --git a/rustfs/src/app/storage_api.rs b/rustfs/src/app/storage_api.rs index d487d2701..494ebed16 100644 --- a/rustfs/src/app/storage_api.rs +++ b/rustfs/src/app/storage_api.rs @@ -577,6 +577,44 @@ pub(crate) mod bucket { #[cfg(test)] pub(crate) use replication_contracts::replication_statuses_map; + pub(crate) async fn persist_force_delete_intent( + store: Arc, + bucket: String, + object: String, + target_arns: Vec, + generation: time::OffsetDateTime, + ) -> crate::storage::storage_api::Result { + let operation_id = Uuid::new_v4(); + crate::storage::storage_api::persist_force_delete_intent( + store, + replication_contracts::MrfReplicateEntry { + bucket, + object, + version_id: None, + retry_count: 0, + size: 0, + op: replication_contracts::MrfOpKind::Delete, + force_delete: true, + delete_marker_version_id: None, + delete_marker: false, + delete_marker_mtime: None, + target_arns, + force_delete_id: Some(operation_id), + force_delete_generation: Some(i64::try_from(generation.unix_timestamp_nanos()).unwrap_or(i64::MAX)), + force_delete_local_commit: false, + }, + ) + .await + .map(|_| operation_id) + } + + pub(crate) async fn commit_force_delete_intent( + store: Arc, + operation_id: Uuid, + ) -> crate::storage::storage_api::Result<()> { + crate::storage::storage_api::commit_force_delete_intent(store, operation_id).await + } + /// Test-only counter of `must_replicate_object` invocations. /// /// Used by white-box regression tests to assert that a single PUT @@ -614,6 +652,13 @@ pub(crate) mod bucket { ReplicationObjectBridge::has_active_delete_rule(snapshot, object) } + pub(crate) fn force_delete_target_set( + snapshot: &DeleteReplicationConfigSnapshot, + prefix: &str, + ) -> Option<(Vec, time::OffsetDateTime)> { + ReplicationObjectBridge::force_delete_target_set(snapshot, prefix) + } + pub(crate) fn delete_replication_version_id( replication_source: &crate::storage::storage_api::StorageObjectInfo, deleted_delete_marker_version: bool, diff --git a/rustfs/src/storage/storage_api.rs b/rustfs/src/storage/storage_api.rs index fa7a4e9b5..e0c292792 100644 --- a/rustfs/src/storage/storage_api.rs +++ b/rustfs/src/storage/storage_api.rs @@ -1441,6 +1441,21 @@ pub(crate) async fn get_bucket_replication_config( ecstore_bucket::metadata_sys::get_replication_config(bucket).await } +pub(crate) async fn persist_force_delete_intent( + api: Arc, + entry: ecstore_bucket::replication::MrfReplicateEntry, +) -> Result<()> { + ecstore_bucket::replication::persist_force_delete_intent(api, entry).await +} + +pub(crate) async fn commit_force_delete_intent(api: Arc, operation_id: uuid::Uuid) -> Result<()> { + ecstore_bucket::replication::commit_force_delete_intent(api, operation_id).await +} + +pub(crate) async fn complete_force_delete_intent(api: Arc, operation_id: uuid::Uuid) -> Result<()> { + ecstore_bucket::replication::complete_force_delete_intent(api, operation_id).await +} + pub(crate) async fn get_bucket_request_payment_config( bucket: &str, ) -> Result<(s3s::dto::RequestPaymentConfiguration, time::OffsetDateTime)> {