From 1511f9eacbd90062ca1e72c5c063a9d807d9ec2b Mon Sep 17 00:00:00 2001 From: cxymds Date: Mon, 20 Apr 2026 18:02:59 +0800 Subject: [PATCH] fix(lifecycle): correct delete replication fanout (#2609) Co-authored-by: loverustfs --- .../bucket/lifecycle/bucket_lifecycle_ops.rs | 254 ++++++++++++++-- .../ecstore/src/bucket/replication/config.rs | 4 + .../replication/replication_resyncer.rs | 119 +++++++- .../src/client/object_handlers_common.rs | 81 ++++- rustfs/src/app/object_usecase.rs | 285 +++++++++++++++++- 5 files changed, 693 insertions(+), 50 deletions(-) diff --git a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs index c734eab17..24d8517fd 100644 --- a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs +++ b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs @@ -1354,7 +1354,7 @@ pub async fn expire_transitioned_object( } }; - schedule_lifecycle_replication_delete_if_needed(oi).await; + schedule_lifecycle_replication_delete_if_needed(oi, &dobj).await; //defer auditLogLifecycle(ctx, *oi, ILMExpiry, tags, traceFn) @@ -1778,7 +1778,7 @@ pub async fn apply_expiry_on_non_transitioned_objects( return false; } }; - schedule_lifecycle_replication_delete_if_needed(oi).await; + schedule_lifecycle_replication_delete_if_needed(oi, &dobj).await; //debug!("dobj: {:?}", dobj); if dobj.name.is_empty() { dobj = oi.clone(); @@ -1819,25 +1819,55 @@ pub async fn apply_expiry_rule(event: &lifecycle::Event, src: &LcEventSrc, oi: & true } -async fn schedule_lifecycle_replication_delete_if_needed(oi: &ObjectInfo) { - if !oi.delete_marker || oi.version_id.is_none() { - return; +fn lifecycle_deleted_object(oi: &ObjectInfo, dobj: &ObjectInfo) -> crate::store_api::DeletedObject { + if dobj.delete_marker { + return crate::store_api::DeletedObject { + object_name: oi.name.clone(), + delete_marker: true, + delete_marker_version_id: dobj.version_id, + delete_marker_mtime: dobj.mod_time.or(oi.mod_time), + ..Default::default() + }; } - let replication_state = lifecycle_delete_replication_state(oi).await; + if oi.delete_marker && oi.version_id.is_some() { + return crate::store_api::DeletedObject { + object_name: oi.name.clone(), + delete_marker: false, + delete_marker_version_id: oi.version_id, + delete_marker_mtime: oi.mod_time, + ..Default::default() + }; + } + + crate::store_api::DeletedObject { + object_name: oi.name.clone(), + delete_marker: false, + version_id: oi.version_id, + delete_marker_mtime: oi.mod_time, + ..Default::default() + } +} + +async fn schedule_lifecycle_replication_delete_if_needed(oi: &ObjectInfo, dobj: &ObjectInfo) { + let mut delete_object = lifecycle_deleted_object(oi, dobj); + let version_id = if delete_object.delete_marker { + None + } else if delete_object.delete_marker_version_id.is_some() { + delete_object.delete_marker_version_id + } else { + delete_object.version_id + }; + + let replication_state = lifecycle_delete_replication_state(oi, version_id).await; if replication_state.is_none() { return; } + delete_object.replication_state = replication_state; + schedule_replication_delete(DeletedObjectReplicationInfo { - delete_object: crate::store_api::DeletedObject { - object_name: oi.name.clone(), - delete_marker_version_id: oi.version_id, - delete_marker: false, - delete_marker_mtime: oi.mod_time, - replication_state, - ..Default::default() - }, + delete_object, bucket: oi.bucket.clone(), event_type: REPLICATE_INCOMING_DELETE.to_string(), ..Default::default() @@ -1845,21 +1875,56 @@ async fn schedule_lifecycle_replication_delete_if_needed(oi: &ObjectInfo) { .await; } -async fn lifecycle_delete_replication_state(oi: &ObjectInfo) -> Option { - if !oi.replication_decision.is_empty() || oi.version_purge_status == VersionPurgeStatusType::Pending { +fn should_reuse_lifecycle_delete_replication_state(oi: &ObjectInfo, version_delete: bool) -> bool { + let state = oi.replication_state(); + if version_delete { + oi.version_purge_status == VersionPurgeStatusType::Pending && !state.purge_targets.is_empty() + } else { + oi.replication_status == rustfs_filemeta::ReplicationStatusType::Pending && !state.targets.is_empty() + } +} + +fn lifecycle_version_purge_state_from_completed_targets(oi: &ObjectInfo) -> Option { + if oi.replication_status != rustfs_filemeta::ReplicationStatusType::Completed { + return None; + } + + let targets = oi.replication_state().targets; + if targets.is_empty() { + return None; + } + + let pending_status = targets.keys().map(|arn| format!("{arn}=PENDING;")).collect::(); + + Some(ReplicationState { + replicate_decision_str: oi.replication_decision.clone(), + version_purge_status_internal: Some(pending_status.clone()), + purge_targets: rustfs_filemeta::version_purge_statuses_map(&pending_status), + ..Default::default() + }) +} + +async fn lifecycle_delete_replication_state(oi: &ObjectInfo, version_id: Option) -> Option { + if should_reuse_lifecycle_delete_replication_state(oi, version_id.is_some()) { return Some(oi.replication_state()); } + if version_id.is_some() + && let Some(state) = lifecycle_version_purge_state_from_completed_targets(oi) + { + return Some(state); + } + let dsc = check_replicate_delete( &oi.bucket, &ObjectToDelete { object_name: oi.name.clone(), - version_id: oi.version_id, + version_id, ..Default::default() }, oi, &ObjectOptions { - version_id: oi.version_id.map(|v| v.to_string()), + version_id: version_id.map(|v| v.to_string()), versioned: BucketVersioningSys::prefix_enabled(&oi.bucket, &oi.name).await, ..Default::default() }, @@ -1870,17 +1935,23 @@ async fn lifecycle_delete_replication_state(oi: &ObjectInfo) -> Option ReplicationState { +fn replication_state_for_delete(dsc: ReplicateDecision, version_delete: bool) -> ReplicationState { let pending_status = dsc.pending_status(); - ReplicationState { + let mut state = ReplicationState { replicate_decision_str: dsc.to_string(), - version_purge_status_internal: pending_status.clone(), - purge_targets: rustfs_filemeta::version_purge_statuses_map(pending_status.as_deref().unwrap_or_default()), ..Default::default() + }; + if version_delete { + state.version_purge_status_internal = pending_status.clone(); + state.purge_targets = rustfs_filemeta::version_purge_statuses_map(pending_status.as_deref().unwrap_or_default()); + } else { + state.replication_status_internal = pending_status.clone(); + state.targets = rustfs_filemeta::replication_statuses_map(pending_status.as_deref().unwrap_or_default()); } + state } pub async fn apply_lifecycle_action(event: &lifecycle::Event, src: &LcEventSrc, oi: &ObjectInfo) -> bool { @@ -1906,7 +1977,9 @@ pub async fn apply_lifecycle_action(event: &lifecycle::Event, src: &LcEventSrc, mod tests { use super::{ StaleMultipartUploadCandidate, cleanup_empty_multipart_sha_dirs_on_local_disks, cleanup_stale_multipart_uploads_once_at, - mark_delete_opts_skip_decommissioned_on_remote_success, merge_stale_multipart_candidate, + lifecycle_deleted_object, lifecycle_version_purge_state_from_completed_targets, + mark_delete_opts_skip_decommissioned_on_remote_success, merge_stale_multipart_candidate, replication_state_for_delete, + should_reuse_lifecycle_delete_replication_state, }; use crate::bucket::metadata::BUCKET_LIFECYCLE_CONFIG; use crate::bucket::metadata_sys; @@ -1917,8 +1990,9 @@ mod tests { use crate::set_disk::{RUSTFS_MULTIPART_BUCKET_KEY, RUSTFS_MULTIPART_OBJECT_KEY}; use crate::store::ECStore; use crate::store_api::{ - BucketOperations, BucketOptions, MakeBucketOptions, MultipartOperations, ObjectOptions, PutObjReader, + BucketOperations, BucketOptions, MakeBucketOptions, MultipartOperations, ObjectInfo, ObjectOptions, PutObjReader, }; + use rustfs_filemeta::{ReplicateDecision, VersionPurgeStatusType}; use serial_test::serial; use sha2::{Digest, Sha256}; use std::collections::HashMap; @@ -1960,6 +2034,136 @@ mod tests { assert!(opts.skip_decommissioned); } + #[test] + fn lifecycle_deleted_object_uses_delete_marker_created_by_expiry() { + let source = ObjectInfo { + bucket: "bucket".to_string(), + name: "key".to_string(), + ..Default::default() + }; + let delete_result = ObjectInfo { + bucket: "bucket".to_string(), + name: "key".to_string(), + delete_marker: true, + version_id: Some(Uuid::new_v4()), + mod_time: Some(OffsetDateTime::now_utc()), + ..Default::default() + }; + + let deleted = lifecycle_deleted_object(&source, &delete_result); + + assert!(deleted.delete_marker); + assert_eq!(deleted.delete_marker_version_id, delete_result.version_id); + assert_eq!(deleted.version_id, None); + assert_eq!(deleted.object_name, "key"); + } + + #[test] + fn lifecycle_deleted_object_uses_version_id_for_noncurrent_version_purge() { + let version_id = Uuid::new_v4(); + let source = ObjectInfo { + bucket: "bucket".to_string(), + name: "key".to_string(), + version_id: Some(version_id), + ..Default::default() + }; + + let deleted = lifecycle_deleted_object(&source, &ObjectInfo::default()); + + assert!(!deleted.delete_marker); + assert_eq!(deleted.version_id, Some(version_id)); + assert_eq!(deleted.delete_marker_version_id, None); + } + + #[test] + fn lifecycle_deleted_object_uses_delete_marker_version_for_marker_purge() { + let version_id = Uuid::new_v4(); + let source = ObjectInfo { + bucket: "bucket".to_string(), + name: "key".to_string(), + delete_marker: true, + version_id: Some(version_id), + ..Default::default() + }; + + let deleted = lifecycle_deleted_object(&source, &ObjectInfo::default()); + + assert!(!deleted.delete_marker); + assert_eq!(deleted.delete_marker_version_id, Some(version_id)); + assert_eq!(deleted.version_id, None); + } + + #[test] + fn replication_state_for_delete_uses_replication_targets_for_current_delete() { + let arn = "arn:aws:s3:::target-bucket"; + let mut dsc = ReplicateDecision::default(); + dsc.set(rustfs_filemeta::ReplicateTargetDecision::new(arn.to_string(), true, false)); + + let state = replication_state_for_delete(dsc, false); + + assert_eq!(state.replication_status_internal.as_deref(), Some(format!("{arn}=PENDING;").as_str())); + assert!(state.version_purge_status_internal.is_none()); + assert!(state.targets.contains_key(arn)); + } + + #[test] + fn replication_state_for_delete_uses_purge_targets_for_version_delete() { + let arn = "arn:aws:s3:::target-bucket"; + let mut dsc = ReplicateDecision::default(); + dsc.set(rustfs_filemeta::ReplicateTargetDecision::new(arn.to_string(), true, false)); + + let state = replication_state_for_delete(dsc, true); + + assert_eq!(state.version_purge_status_internal.as_deref(), Some(format!("{arn}=PENDING;").as_str())); + assert!(state.replication_status_internal.is_none()); + assert!(state.purge_targets.contains_key(arn)); + } + + #[test] + fn lifecycle_delete_replication_state_reuses_only_pending_version_purge_state() { + let oi = ObjectInfo { + version_purge_status: VersionPurgeStatusType::Pending, + version_purge_status_internal: Some("arn:aws:s3:::target=PENDING;".to_string()), + replication_decision: "arn:aws:s3:::target=true;false;arn:aws:s3:::target;".to_string(), + ..Default::default() + }; + + assert!(should_reuse_lifecycle_delete_replication_state(&oi, true)); + assert!(!should_reuse_lifecycle_delete_replication_state(&oi, false)); + } + + #[test] + fn lifecycle_delete_replication_state_does_not_reuse_put_replication_for_version_delete() { + let oi = ObjectInfo { + replication_status: rustfs_filemeta::ReplicationStatusType::Completed, + replication_status_internal: Some("arn:aws:s3:::target=COMPLETED;".to_string()), + replication_decision: "arn:aws:s3:::target=true;false;arn:aws:s3:::target;".to_string(), + ..Default::default() + }; + + assert!( + !should_reuse_lifecycle_delete_replication_state(&oi, true), + "version purges must not reuse plain object replication state from prior PUT/delete-marker replication" + ); + } + + #[test] + fn lifecycle_version_purge_state_from_completed_targets_derives_pending_purge_targets() { + let oi = ObjectInfo { + replication_status: rustfs_filemeta::ReplicationStatusType::Completed, + replication_status_internal: Some("arn:aws:s3:::target=COMPLETED;".to_string()), + replication_decision: "arn:aws:s3:::target=true;false;arn:aws:s3:::target;".to_string(), + ..Default::default() + }; + + let state = lifecycle_version_purge_state_from_completed_targets(&oi) + .expect("completed replication targets should be convertible into version-purge targets"); + + assert_eq!(state.version_purge_status_internal.as_deref(), Some("arn:aws:s3:::target=PENDING;")); + assert!(state.purge_targets.contains_key("arn:aws:s3:::target")); + assert_eq!(state.replicate_decision_str, oi.replication_decision); + } + static STALE_MULTIPART_TEST_ENV: OnceLock<(Vec, Arc)> = OnceLock::new(); async fn setup_test_env() -> (Vec, Arc) { diff --git a/crates/ecstore/src/bucket/replication/config.rs b/crates/ecstore/src/bucket/replication/config.rs index abc8cdc76..e150488ee 100644 --- a/crates/ecstore/src/bucket/replication/config.rs +++ b/crates/ecstore/src/bucket/replication/config.rs @@ -152,6 +152,10 @@ impl ReplicationConfigurationExt for ReplicationConfiguration { } if obj.op_type == ReplicationType::Delete { + if !rule.metadata_replicate(obj) { + return false; + } + if obj.version_id.is_some() { if obj.delete_marker { return rule.delete_marker_replication.clone().is_some_and(|d| { diff --git a/crates/ecstore/src/bucket/replication/replication_resyncer.rs b/crates/ecstore/src/bucket/replication/replication_resyncer.rs index d2b85cbde..ad30e259f 100644 --- a/crates/ecstore/src/bucket/replication/replication_resyncer.rs +++ b/crates/ecstore/src/bucket/replication/replication_resyncer.rs @@ -85,7 +85,7 @@ use tokio::task::JoinSet; use tokio::time::Duration as TokioDuration; use tokio_util::io::ReaderStream; use tokio_util::sync::CancellationToken; -use tracing::{error, info, instrument, warn}; +use tracing::{debug, error, info, instrument, warn}; use uuid::Uuid; pub(crate) const REPLICATION_DIR: &str = ".replication"; @@ -1272,15 +1272,7 @@ pub async fn check_replicate_delete( return ReplicateDecision::default(); } - let opts = ObjectOpts { - name: dobj.object_name.clone(), - ssec: is_ssec_encrypted(&oi.user_defined), - user_tags: oi.user_tags.clone(), - delete_marker: oi.delete_marker, - version_id: dobj.version_id, - op_type: ReplicationType::Delete, - ..Default::default() - }; + let opts = delete_replication_object_opts(dobj, oi); let tgt_arns = rcfg.filter_target_arns(&opts); let mut dsc = ReplicateDecision::new(); @@ -1332,6 +1324,19 @@ pub async fn check_replicate_delete( dsc } +fn delete_replication_object_opts(dobj: &ObjectToDelete, oi: &ObjectInfo) -> ObjectOpts { + ObjectOpts { + name: dobj.object_name.clone(), + ssec: is_ssec_encrypted(&oi.user_defined), + user_tags: oi.user_tags.clone(), + delete_marker: oi.delete_marker, + version_id: dobj.version_id, + op_type: ReplicationType::Delete, + replica: oi.replication_status == ReplicationStatusType::Replica, + ..Default::default() + } +} + /// Check if the user-defined metadata contains SSEC encryption headers fn is_ssec_encrypted(user_defined: &HashMap) -> bool { user_defined.contains_key(SSEC_ALGORITHM_HEADER) @@ -1703,7 +1708,7 @@ pub async fn replicate_delete(dobj: DeletedObjectReplicationInfo, let is_version_purge = is_version_delete_replication(&dobj.delete_object); - if !is_version_purge && dobj.delete_object.delete_marker && dobj.delete_object.delete_marker_version_id.is_some() { + if should_retry_delete_marker_purge(&dobj.delete_object) { let bucket_clone = bucket.clone(); let dobj_clone = dobj.clone(); let dsc_clone = dsc.clone(); @@ -2061,6 +2066,10 @@ fn is_version_delete_replication(dobj: &DeletedObject) -> bool { dobj.version_id.is_some() || (dobj.delete_marker_version_id.is_some() && !dobj.delete_marker) } +fn should_retry_delete_marker_purge(dobj: &DeletedObject) -> bool { + dobj.delete_marker_version_id.is_some() +} + fn is_retryable_delete_replication_head_error(is_not_found: bool, code: Option<&str>) -> bool { !is_not_found && !matches!(code, Some("MethodNotAllowed" | "405")) } @@ -2152,6 +2161,14 @@ async fn replicate_delete_to_target(dobj: &DeletedObjectReplicationInfo, tgt_cli .await { Ok(_) => { + debug!( + bucket = tgt_client.bucket, + object = dobj.delete_object.object_name, + version_id = ?version_id, + delete_marker = dobj.delete_object.delete_marker, + is_version_purge, + "replicate_delete_to_target succeeded" + ); if !is_version_purge { rinfo.replication_status = ReplicationStatusType::Completed; } else { @@ -2159,6 +2176,15 @@ async fn replicate_delete_to_target(dobj: &DeletedObjectReplicationInfo, tgt_cli } } Err(e) => { + warn!( + 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, + "replicate_delete_to_target failed" + ); rinfo.error = Some(e.to_string()); if !is_version_purge { rinfo.replication_status = ReplicationStatusType::Failed; @@ -3603,6 +3629,49 @@ mod tests { ); } + #[test] + fn test_delete_replication_object_opts_marks_replica_deletes() { + let dobj = ObjectToDelete { + object_name: "obj".to_string(), + version_id: Some(Uuid::new_v4()), + ..Default::default() + }; + let oi = ObjectInfo { + bucket: "b".to_string(), + name: "obj".to_string(), + replication_status: ReplicationStatusType::Replica, + ..Default::default() + }; + + let opts = delete_replication_object_opts(&dobj, &oi); + + assert!( + opts.replica, + "replica deletes must preserve replica status for downstream ReplicaModifications rules" + ); + assert_eq!(opts.version_id, dobj.version_id); + assert_eq!(opts.name, dobj.object_name); + assert_eq!(opts.op_type, ReplicationType::Delete); + } + + #[test] + fn test_delete_replication_object_opts_keeps_non_replica_deletes_local() { + let dobj = ObjectToDelete { + object_name: "obj".to_string(), + ..Default::default() + }; + let oi = ObjectInfo { + bucket: "b".to_string(), + name: "obj".to_string(), + replication_status: ReplicationStatusType::Completed, + ..Default::default() + }; + + let opts = delete_replication_object_opts(&dobj, &oi); + + assert!(!opts.replica, "source-originated deletes should not be treated as replica modifications"); + } + #[test] fn test_is_version_delete_replication_for_delete_marker_version_purge() { let dobj = DeletedObject { @@ -3631,6 +3700,34 @@ mod tests { ); } + #[test] + fn test_should_retry_delete_marker_purge_for_version_purge() { + let dobj = DeletedObject { + delete_marker: false, + delete_marker_version_id: Some(Uuid::new_v4()), + ..Default::default() + }; + + assert!( + should_retry_delete_marker_purge(&dobj), + "delete-marker version purge should schedule delayed target cleanup in case the target marker arrives late" + ); + } + + #[test] + fn test_should_retry_delete_marker_purge_for_delete_marker_creation() { + let dobj = DeletedObject { + delete_marker: true, + delete_marker_version_id: Some(Uuid::new_v4()), + ..Default::default() + }; + + assert!( + should_retry_delete_marker_purge(&dobj), + "delete-marker creation should keep the late-arrival cleanup path so downstream purges can catch up" + ); + } + #[test] fn test_is_retryable_delete_replication_head_error_allows_delete_marker_head_responses() { assert!( diff --git a/crates/ecstore/src/client/object_handlers_common.rs b/crates/ecstore/src/client/object_handlers_common.rs index 2fef505f4..cf57b9aa5 100644 --- a/crates/ecstore/src/client/object_handlers_common.rs +++ b/crates/ecstore/src/client/object_handlers_common.rs @@ -16,12 +16,26 @@ use std::sync::Arc; use tracing::warn; use crate::bucket::lifecycle::lifecycle; +use crate::bucket::replication::{DeletedObjectReplicationInfo, check_replicate_delete, schedule_replication_delete}; use crate::bucket::versioning::VersioningApi; use crate::bucket::versioning_sys::BucketVersioningSys; use crate::store::ECStore; use crate::store_api::{ObjectOperations, ObjectOptions, ObjectToDelete}; +use rustfs_filemeta::{REPLICATE_INCOMING_DELETE, ReplicationState, version_purge_statuses_map}; use rustfs_lock::MAX_DELETE_LIST; +fn lifecycle_version_delete_replication_state( + replicate_decision_str: String, + pending_status: Option, +) -> ReplicationState { + ReplicationState { + replicate_decision_str, + version_purge_status_internal: pending_status.clone(), + purge_targets: version_purge_statuses_map(pending_status.as_deref().unwrap_or_default()), + ..Default::default() + } +} + pub async fn delete_object_versions(api: &Arc, bucket: &str, to_del: &[ObjectToDelete], _lc_event: lifecycle::Event) { let version_suspended = match BucketVersioningSys::get(bucket).await { Ok(vc) => vc.suspended(), @@ -39,7 +53,37 @@ pub async fn delete_object_versions(api: &Arc, bucket: &str, to_del: &[ } else { remaining = &[]; } - let (_deleted_objs, errors) = api + + let mut replication_candidates: Vec> = Vec::with_capacity(to_del.len()); + for object in to_del.iter() { + let version_id = object.version_id.map(|vid| vid.to_string()); + let opts = ObjectOptions { + version_id: version_id.clone(), + versioned: true, + version_suspended, + ..Default::default() + }; + let candidate = match api.get_object_info(bucket, &object.object_name, &opts).await { + Ok(info) => { + let dsc = check_replicate_delete(bucket, object, &info, &opts, None).await; + dsc.replicate_any() + .then(|| lifecycle_version_delete_replication_state(dsc.to_string(), dsc.pending_status())) + } + Err(err) => { + warn!( + bucket, + object = %object.object_name, + version_id = ?version_id, + error = ?err, + "failed to get object info during lifecycle noncurrent version cleanup; skipping delete replication scheduling" + ); + None + } + }; + replication_candidates.push(candidate); + } + + let (mut deleted_objs, errors) = api .delete_objects( bucket, to_del.to_vec(), @@ -49,6 +93,24 @@ pub async fn delete_object_versions(api: &Arc, bucket: &str, to_del: &[ }, ) .await; + + for (i, deleted_obj) in deleted_objs.iter_mut().enumerate() { + if errors.get(i).and_then(|err| err.as_ref()).is_some() { + continue; + } + let Some(replication_state) = replication_candidates.get(i).and_then(|c| c.clone()) else { + continue; + }; + deleted_obj.replication_state = Some(replication_state); + schedule_replication_delete(DeletedObjectReplicationInfo { + delete_object: deleted_obj.clone(), + bucket: bucket.to_string(), + event_type: REPLICATE_INCOMING_DELETE.to_string(), + ..Default::default() + }) + .await; + } + for (i, err) in errors.iter().enumerate() { if let Some(e) = err { let obj_name = to_del.get(i).map(|o| o.object_name.as_str()).unwrap_or(""); @@ -65,3 +127,20 @@ pub async fn delete_object_versions(api: &Arc, bucket: &str, to_del: &[ } } } + +#[cfg(test)] +mod tests { + use super::lifecycle_version_delete_replication_state; + + #[test] + fn lifecycle_version_delete_replication_state_tracks_pending_purge_targets() { + let state = lifecycle_version_delete_replication_state( + "arn:aws:s3:::target=true;false;arn:aws:s3:::target;".to_string(), + Some("arn:aws:s3:::target=PENDING;".to_string()), + ); + + assert_eq!(state.version_purge_status_internal.as_deref(), Some("arn:aws:s3:::target=PENDING;")); + assert!(state.purge_targets.contains_key("arn:aws:s3:::target")); + assert_eq!(state.replicate_decision_str, "arn:aws:s3:::target=true;false;arn:aws:s3:::target;"); + } +} diff --git a/rustfs/src/app/object_usecase.rs b/rustfs/src/app/object_usecase.rs index 2711d9675..b4e584bfa 100644 --- a/rustfs/src/app/object_usecase.rs +++ b/rustfs/src/app/object_usecase.rs @@ -515,7 +515,8 @@ async fn enrich_delete_replication_state_if_needed( let Some(replication_state) = delete_object.replication_state.as_ref() else { return; }; - if !replication_state.replicate_decision_str.is_empty() + if obj_info.replication_status != ReplicationStatusType::Replica + && !replication_state.replicate_decision_str.is_empty() && (!replication_state.targets.is_empty() || !replication_state.purge_targets.is_empty()) { return; @@ -550,12 +551,61 @@ fn should_schedule_delete_replication( return false; } + if opts.version_id.is_some() && !deleted_delete_marker_version && !replication_source.delete_marker { + return matches!( + replication_source.replication_status, + ReplicationStatusType::Replica + | ReplicationStatusType::Pending + | ReplicationStatusType::Completed + | ReplicationStatusType::Failed + ); + } + replication_source.replication_status == ReplicationStatusType::Replica || replication_source.replication_status == ReplicationStatusType::Pending || replication_source.version_purge_status == VersionPurgeStatusType::Pending || (deleted_delete_marker_version && replication_source.replication_status == ReplicationStatusType::Completed) } +async fn should_schedule_replica_delete_replication( + bucket: &str, + replication_source: &ObjectInfo, + version_id: Option, +) -> bool { + let Ok((config, _)) = metadata_sys::get_replication_config(bucket).await else { + return false; + }; + + delete_replication_state_from_config(&config, replication_source, version_id, true).is_some() +} + +fn delete_replication_version_id(replication_source: &ObjectInfo, deleted_delete_marker_version: bool) -> Option { + if replication_source.delete_marker && !deleted_delete_marker_version { + None + } else { + replication_source.version_id + } +} + +fn should_use_existing_delete_replication_info(opts: &ObjectOptions) -> bool { + opts.version_id.is_some() && !opts.delete_marker +} + +fn delete_replication_state_source<'a>( + opts: &ObjectOptions, + existing_object_info: Option<&'a ObjectInfo>, + deleted_object_info: &'a ObjectInfo, +) -> &'a ObjectInfo { + if opts.replication_request + && deleted_object_info.delete_marker + && let Some(existing) = existing_object_info + { + return existing; + } + + deleted_object_info +} + const AMZ_SNOWBALL_EXTRACT_COMPAT: &str = "X-Amz-Snowball-Auto-Extract"; #[cfg(test)] const AMZ_SNOWBALL_PREFIX_INTERNAL: &str = "X-Amz-Meta-Rustfs-Snowball-Prefix"; @@ -3242,34 +3292,45 @@ impl DefaultObjectUsecase { return result; } - let deleted_replication_info = existing_object_info.as_ref().filter(|_| opts.version_id.is_some()); - let replication_source = deleted_replication_info.unwrap_or(&obj_info); + let deleted_replication_info = existing_object_info + .as_ref() + .filter(|_| should_use_existing_delete_replication_info(&opts)); + let deleted_object_source = deleted_replication_info.unwrap_or(&obj_info); + let replication_state_source = + delete_replication_state_source(&opts, existing_object_info.as_ref(), deleted_object_source); let deleted_delete_marker_version = deleted_replication_info.is_some_and(|info| info.delete_marker); - if should_schedule_delete_replication(&opts, replication_source, deleted_delete_marker_version) { + let delete_replication_version_id = delete_replication_version_id(deleted_object_source, deleted_delete_marker_version); + let schedule_delete_replication = if opts.replication_request && replica { + should_schedule_replica_delete_replication(&bucket, replication_state_source, delete_replication_version_id).await + } else { + should_schedule_delete_replication(&opts, deleted_object_source, deleted_delete_marker_version) + }; + + if schedule_delete_replication { let mut deleted_object = DeletedObjectReplicationInfo { delete_object: rustfs_ecstore::store_api::DeletedObject { - delete_marker: replication_source.delete_marker && !deleted_delete_marker_version, - delete_marker_version_id: if replication_source.delete_marker { - replication_source.version_id + delete_marker: deleted_object_source.delete_marker && !deleted_delete_marker_version, + delete_marker_version_id: if deleted_object_source.delete_marker { + deleted_object_source.version_id } else { None }, object_name: key.clone(), - version_id: if replication_source.delete_marker { + version_id: if deleted_object_source.delete_marker { None } else { - replication_source.version_id + deleted_object_source.version_id }, - delete_marker_mtime: replication_source.mod_time, - replication_state: Some(replication_source.replication_state()), + delete_marker_mtime: deleted_object_source.mod_time, + replication_state: Some(replication_state_source.replication_state()), ..Default::default() }, bucket: bucket.clone(), event_type: REPLICATE_INCOMING_DELETE.to_string(), ..Default::default() }; - enrich_delete_replication_state_if_needed(&bucket, &mut deleted_object.delete_object, replication_source).await; + enrich_delete_replication_state_if_needed(&bucket, &mut deleted_object.delete_object, replication_state_source).await; schedule_replication_delete(deleted_object).await; } @@ -4802,6 +4863,25 @@ mod tests { ); } + #[test] + fn should_schedule_delete_replication_keeps_object_version_purge_from_completed_source() { + let opts = ObjectOptions { + replication_request: false, + version_id: Some(Uuid::new_v4().to_string()), + ..Default::default() + }; + let replication_source = ObjectInfo { + delete_marker: false, + replication_status: ReplicationStatusType::Completed, + ..Default::default() + }; + + assert!( + should_schedule_delete_replication(&opts, &replication_source, false), + "source-side object version purge must still enqueue delete replication after the original PUT completed" + ); + } + #[tokio::test] async fn execute_get_object_attributes_returns_internal_error_when_store_uninitialized() { let input = GetObjectAttributesInput::builder() @@ -4990,7 +5070,12 @@ mod tests { id: Some("rule-1".to_string()), prefix: Some("test/".to_string()), priority: Some(1), - source_selection_criteria: None, + source_selection_criteria: Some(SourceSelectionCriteria { + replica_modifications: Some(ReplicaModifications { + status: ReplicaModificationsStatus::from_static(ReplicaModificationsStatus::ENABLED), + }), + sse_kms_encrypted_objects: None, + }), status: ReplicationRuleStatus::from_static(ReplicationRuleStatus::ENABLED), }], }; @@ -5011,6 +5096,45 @@ mod tests { assert!(state.targets.contains_key(&arn)); } + #[test] + fn delete_replication_state_from_config_skips_replica_delete_without_replica_modifications() { + let arn = "arn:aws:s3:::target-bucket".to_string(); + let config = ReplicationConfiguration { + role: arn.clone(), + rules: vec![ReplicationRule { + delete_marker_replication: Some(DeleteMarkerReplication { + status: Some(DeleteMarkerReplicationStatus::from_static(DeleteMarkerReplicationStatus::ENABLED)), + }), + delete_replication: None, + destination: Destination { + bucket: arn, + ..Default::default() + }, + existing_object_replication: Some(ExistingObjectReplication { + status: ExistingObjectReplicationStatus::from_static(ExistingObjectReplicationStatus::ENABLED), + }), + filter: None, + id: Some("rule-1".to_string()), + prefix: Some("test/".to_string()), + priority: Some(1), + source_selection_criteria: None, + status: ReplicationRuleStatus::from_static(ReplicationRuleStatus::ENABLED), + }], + }; + let obj_info = ObjectInfo { + bucket: "bucket".to_string(), + name: "test/object.txt".to_string(), + delete_marker: true, + replication_status: ReplicationStatusType::Replica, + ..Default::default() + }; + + assert!( + delete_replication_state_from_config(&config, &obj_info, None, true).is_none(), + "replica deletes must only fan out when ReplicaModifications are enabled" + ); + } + #[test] fn delete_replication_state_from_config_tracks_delete_marker_version_purges() { let arn = "arn:aws:s3:::target-bucket".to_string(); @@ -5053,4 +5177,139 @@ mod tests { assert_eq!(state.replicate_decision_str, format!("{arn}=true;false;{arn};")); assert!(state.purge_targets.contains_key(&arn)); } + + #[test] + fn delete_replication_state_source_prefers_existing_replica_for_replication_delete_marker_creation() { + let opts = ObjectOptions { + replication_request: true, + version_id: Some(Uuid::new_v4().to_string()), + ..Default::default() + }; + let existing = ObjectInfo { + name: "test/object.txt".to_string(), + replication_status: ReplicationStatusType::Completed, + ..Default::default() + }; + let deleted = ObjectInfo { + name: "test/object.txt".to_string(), + delete_marker: true, + ..Default::default() + }; + + let source = delete_replication_state_source(&opts, Some(&existing), &deleted); + + assert_eq!(source.replication_status, ReplicationStatusType::Completed); + assert!( + !source.delete_marker, + "downstream fanout should inherit replica identity from the pre-delete object" + ); + } + + #[test] + fn delete_replication_state_source_keeps_deleted_marker_for_non_replication_requests() { + let opts = ObjectOptions::default(); + let existing = ObjectInfo { + name: "test/object.txt".to_string(), + replication_status: ReplicationStatusType::Replica, + ..Default::default() + }; + let deleted = ObjectInfo { + name: "test/object.txt".to_string(), + delete_marker: true, + ..Default::default() + }; + + let source = delete_replication_state_source(&opts, Some(&existing), &deleted); + + assert!( + source.delete_marker, + "source-originated deletes should keep using the new delete marker state" + ); + } + + #[test] + fn replica_delete_enrichment_must_not_reuse_upstream_targets() { + let delete_object = rustfs_ecstore::store_api::DeletedObject { + replication_state: Some(ReplicationState { + replicate_decision_str: "arn:aws:s3:::upstream=true;false;arn:aws:s3:::upstream;".to_string(), + replication_status_internal: Some("arn:aws:s3:::upstream=COMPLETED;".to_string()), + targets: replication_statuses_map("arn:aws:s3:::upstream=COMPLETED;"), + ..Default::default() + }), + ..Default::default() + }; + let obj_info = ObjectInfo { + replication_status: ReplicationStatusType::Replica, + ..Default::default() + }; + + let should_keep_existing = delete_object.replication_state.as_ref().is_some_and(|state| { + obj_info.replication_status != ReplicationStatusType::Replica + && !state.replicate_decision_str.is_empty() + && (!state.targets.is_empty() || !state.purge_targets.is_empty()) + }); + + assert!( + !should_keep_existing, + "replica fanout deletes must recompute targets from the local bucket config instead of reusing upstream replication state" + ); + } + + #[test] + fn delete_replication_version_id_uses_none_for_delete_marker_creation() { + let source = ObjectInfo { + delete_marker: true, + version_id: Some(Uuid::new_v4()), + ..Default::default() + }; + + assert_eq!( + delete_replication_version_id(&source, false), + None, + "delete-marker creation must stay on the delete-marker replication path" + ); + } + + #[test] + fn delete_replication_version_id_keeps_version_for_marker_purge() { + let version_id = Uuid::new_v4(); + let source = ObjectInfo { + delete_marker: true, + version_id: Some(version_id), + ..Default::default() + }; + + assert_eq!( + delete_replication_version_id(&source, true), + Some(version_id), + "delete-marker version purge must preserve the concrete version id for downstream purge replication" + ); + } + + #[test] + fn should_use_existing_delete_replication_info_ignores_replication_delete_marker_creation() { + let opts = ObjectOptions { + version_id: Some(Uuid::new_v4().to_string()), + delete_marker: true, + ..Default::default() + }; + + assert!( + !should_use_existing_delete_replication_info(&opts), + "replicated delete-marker creation carries a source version id header but must not be treated as a version purge" + ); + } + + #[test] + fn should_use_existing_delete_replication_info_keeps_version_delete_requests() { + let opts = ObjectOptions { + version_id: Some(Uuid::new_v4().to_string()), + ..Default::default() + }; + + assert!( + should_use_existing_delete_replication_info(&opts), + "true version-delete requests should keep using the pre-delete object info" + ); + } }