diff --git a/crates/ecstore/src/bucket/replication/replication_filemeta_boundary.rs b/crates/ecstore/src/bucket/replication/replication_filemeta_boundary.rs index 01d8a5ce9..e83a8a690 100644 --- a/crates/ecstore/src/bucket/replication/replication_filemeta_boundary.rs +++ b/crates/ecstore/src/bucket/replication/replication_filemeta_boundary.rs @@ -16,7 +16,6 @@ pub(crate) use rustfs_replication::{MrfOpKind, MrfReplicateEntry}; pub(crate) use rustfs_replication::{ REPLICATE_EXISTING, REPLICATE_EXISTING_DELETE, REPLICATE_HEAL, REPLICATE_HEAL_DELETE, REPLICATE_INCOMING_DELETE, ReplicateDecision, ReplicateObjectInfo, ReplicateTargetDecision, ReplicatedInfos, ReplicatedTargetInfo, ReplicationAction, - ReplicationState, ReplicationStatusType, ReplicationType, ReplicationWorkerOperation, ResyncDecision, ResyncTargetDecision, - VersionPurgeStatusType, get_replication_state, parse_replicate_decision, replication_statuses_map, target_reset_header, - version_purge_statuses_map, + ReplicationState, ReplicationStatusType, ReplicationType, ReplicationWorkerOperation, ResyncDecision, VersionPurgeStatusType, + get_replication_state, parse_replicate_decision, replication_statuses_map, target_reset_header, version_purge_statuses_map, }; diff --git a/crates/ecstore/src/bucket/replication/replication_resyncer.rs b/crates/ecstore/src/bucket/replication/replication_resyncer.rs index e2b004a2c..c9943e794 100644 --- a/crates/ecstore/src/bucket/replication/replication_resyncer.rs +++ b/crates/ecstore/src/bucket/replication/replication_resyncer.rs @@ -21,8 +21,8 @@ use super::replication_event_sink::{EventArgs, send_event, send_local_event}; use super::replication_filemeta_boundary::{ MrfReplicateEntry, REPLICATE_EXISTING, REPLICATE_EXISTING_DELETE, ReplicateDecision, ReplicateObjectInfo, ReplicateTargetDecision, ReplicatedInfos, ReplicatedTargetInfo, ReplicationAction, ReplicationStatusType, ReplicationType, - ResyncDecision, ResyncTargetDecision, VersionPurgeStatusType, get_replication_state, parse_replicate_decision, - replication_statuses_map, target_reset_header, version_purge_statuses_map, + ResyncDecision, VersionPurgeStatusType, get_replication_state, parse_replicate_decision, replication_statuses_map, + target_reset_header, version_purge_statuses_map, }; use super::replication_lock_boundary::ReplicationLockTiming; use super::replication_metadata_boundary::ReplicationMetadataStore; @@ -55,12 +55,15 @@ use http_body::Frame; use http_body_util::StreamBody; #[cfg(test)] use rmp_serde; +#[cfg(test)] +use rustfs_replication::content_matches_by_etag; use rustfs_replication::{ - BucketReplicationResyncStatus, DeletedObjectReplicationInfo, MustReplicateOptions, ReplicationSourceObject, - ReplicationTargetObject, ResyncOpts, TargetReplicationResyncStatus, content_matches_by_etag, + BucketReplicationResyncStatus, DeletedObjectReplicationInfo, MustReplicateOptions, ReplicationDeleteSource, + ReplicationResyncTargetObject, ReplicationSourceObject, ReplicationTargetObject, ResyncOpts, TargetReplicationResyncStatus, + delete_replication_missing_source_decision, delete_replication_object_opts, heal_uses_delete_replication_path, is_retryable_delete_replication_head_error, is_version_delete_replication, is_version_id_mismatch, - replication_action_for_target, resync_state_accepts_update, should_count_head_proxy_failure, - should_retry_delete_marker_purge, target_is_newer_than_source_null_version, + replication_action_for_target, replication_etags_match, resync_state_accepts_update, resync_target_for_object, + should_count_head_proxy_failure, should_retry_delete_marker_purge, target_is_newer_than_source_null_version, }; use rustfs_s3_types::EventName; use rustfs_utils::http::{ @@ -69,8 +72,8 @@ use rustfs_utils::http::{ SUFFIX_REPLICATION_STATUS, SUFFIX_TAGGING_TIMESTAMP, headers, }; use rustfs_utils::http::{ - SUFFIX_REPLICATION_ACTUAL_OBJECT_SIZE, SUFFIX_REPLICATION_RESET_STATUS, SUFFIX_REPLICATION_SSEC_CRC, get_header_map, get_str, - has_internal_suffix, insert_header_map, insert_str, is_internal_key, + SUFFIX_REPLICATION_ACTUAL_OBJECT_SIZE, SUFFIX_REPLICATION_SSEC_CRC, get_str, has_internal_suffix, insert_header_map, + insert_str, is_internal_key, }; use rustfs_utils::{DEFAULT_SIP_HASH_KEY, sip_hash}; use s3s::dto::ReplicationConfiguration; @@ -860,10 +863,6 @@ impl ReplicationResyncer { } } -fn heal_should_use_check_replicate_delete(oi: &ObjectInfo) -> bool { - oi.delete_marker || !oi.version_purge_status.is_empty() -} - pub async fn get_heal_replicate_object_info(oi: &ObjectInfo, rcfg: &ReplicationConfig) -> ReplicateObjectInfo { let mut oi = oi.clone(); let mut user_defined = (*oi.user_defined).clone(); @@ -891,7 +890,7 @@ pub async fn get_heal_replicate_object_info(oi: &ObjectInfo, rcfg: &ReplicationC } } - let dsc = if heal_should_use_check_replicate_delete(&oi) { + let dsc = if heal_uses_delete_replication_path(oi.delete_marker, &oi.version_purge_status) { check_replicate_delete( oi.bucket.as_str(), &ObjectToDelete { @@ -1064,8 +1063,11 @@ impl ReplicationConfig { { resync_decision.targets.insert( decision.arn.clone(), - resync_target( - &oi, + resync_target_for_object( + &ReplicationResyncTargetObject { + mod_time: oi.mod_time, + user_defined: oi.user_defined.as_ref(), + }, &target.arn, &target.reset_id, target.reset_before_date, @@ -1079,59 +1081,6 @@ impl ReplicationConfig { } } -pub fn resync_target( - oi: &ObjectInfo, - arn: &str, - reset_id: &str, - reset_before_date: Option, - status: ReplicationStatusType, -) -> ResyncTargetDecision { - let rs = oi - .user_defined - .get(target_reset_header(arn).as_str()) - .cloned() - .or_else(|| get_header_map(&oi.user_defined, SUFFIX_REPLICATION_RESET_STATUS)); - - let mut dec = ResyncTargetDecision::default(); - - let mod_time = oi.mod_time.unwrap_or(OffsetDateTime::UNIX_EPOCH); - - if rs.is_none() { - let reset_before_date = reset_before_date.unwrap_or(OffsetDateTime::UNIX_EPOCH); - if !reset_id.is_empty() && mod_time <= reset_before_date { - dec.replicate = true; - return dec; - } - - dec.replicate = status == ReplicationStatusType::Empty; - - return dec; - } - - if reset_id.is_empty() || reset_before_date.is_none() { - return dec; - } - - let rs = rs.unwrap(); - let reset_before_date = reset_before_date.unwrap(); - - let parts: Vec<&str> = rs.splitn(2, ';').collect(); - - if parts.len() != 2 { - return dec; - } - - let new_reset = parts[1] != reset_id; - - if !new_reset && status == ReplicationStatusType::Completed { - return dec; - } - - dec.replicate = new_reset && mod_time <= reset_before_date; - - dec -} - pub(crate) fn get_must_replicate_options( user_defined: &HashMap, user_tags: String, @@ -1180,7 +1129,15 @@ pub(crate) async fn check_replicate_delete( return ReplicateDecision::default(); } - let opts = delete_replication_object_opts(dobj, oi); + let opts = delete_replication_object_opts( + dobj, + &ReplicationDeleteSource { + user_defined: oi.user_defined.as_ref(), + user_tags: oi.user_tags.as_str(), + delete_marker: oi.delete_marker, + replication_status: oi.replication_status.clone(), + }, + ); let tgt_arns = rcfg.filter_target_arns(&opts); let mut dsc = ReplicateDecision::new(); @@ -1198,21 +1155,12 @@ pub(crate) async fn check_replicate_delete( // When incoming delete is removal of a delete marker (a.k.a versioned delete), // GetObjectInfo returns extra information even though it returns errFileNotFound if gerr.is_some() { - let valid_repl_status = matches!( + if let Some(replicate) = delete_replication_missing_source_decision( + oi.delete_marker, oi.target_replication_status(&tgt_arn), - ReplicationStatusType::Pending | ReplicationStatusType::Completed | ReplicationStatusType::Failed - ); - - if oi.delete_marker && (valid_repl_status || replicate) { - dsc.set(ReplicateTargetDecision::new(tgt_arn, replicate, sync)); - continue; - } - - // Can be the case that other cluster is down and duplicate `mc rm --vid` - // is issued - this still needs to be replicated back to the other target - if oi.version_purge_status != VersionPurgeStatusType::default() { - let replicate = oi.version_purge_status == VersionPurgeStatusType::Pending - || oi.version_purge_status == VersionPurgeStatusType::Failed; + replicate, + &oi.version_purge_status, + ) { dsc.set(ReplicateTargetDecision::new(tgt_arn, replicate, sync)); } continue; @@ -1232,19 +1180,6 @@ pub(crate) async fn check_replicate_delete( dsc } -fn delete_replication_object_opts(dobj: &ObjectToDelete, oi: &ObjectInfo) -> ObjectOpts { - ObjectOpts { - name: dobj.object_name.clone(), - ssec: rustfs_replication::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() - } -} - pub(crate) async fn must_replicate(bucket: &str, object: &str, mopts: MustReplicateOptions) -> ReplicateDecision { if runtime_sources::object_store_handle().is_none() { return ReplicateDecision::default(); @@ -2601,12 +2536,7 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { } else if is_version_id_format_mismatch(&e) { // Version-ID format mismatch: retry without versionId and compare ETags. match head_object_fallback(&bucket, &tgt_client, &object).await { - Ok(Some(oi)) - if content_matches_by_etag( - &replication_source_object(&object_info), - &replication_target_object(&oi), - ) => - { + Ok(Some(oi)) if replication_etags_match(object_info.etag.as_deref(), oi.e_tag.as_deref()) => { rinfo.replication_status = ReplicationStatusType::Completed; rinfo.replication_resynced = true; rinfo.replication_action = ReplicationAction::None; @@ -2973,10 +2903,7 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { // Version-ID format mismatch: retry without versionId and compare ETags. match head_object_fallback(&bucket, &tgt_client, &object).await { Ok(Some(oi)) => { - replication_action = if content_matches_by_etag( - &replication_source_object(&object_info), - &replication_target_object(&oi), - ) { + replication_action = if replication_etags_match(object_info.etag.as_deref(), oi.e_tag.as_deref()) { ReplicationAction::None } else { ReplicationAction::All @@ -3757,81 +3684,6 @@ mod tests { assert_eq!(target.resync_before_date, None); } - #[test] - fn test_heal_should_use_check_replicate_delete_failed_non_delete_marker() { - let oi = ObjectInfo { - bucket: "b".to_string(), - name: "obj".to_string(), - delete_marker: false, - replication_status: ReplicationStatusType::Failed, - ..Default::default() - }; - assert!( - !heal_should_use_check_replicate_delete(&oi), - "Failed non-delete-marker object must use must_replicate path so it can be re-queued for heal" - ); - } - - #[test] - fn test_heal_should_use_check_replicate_delete_pending_non_delete_marker_uses_must_replicate_path() { - let oi = ObjectInfo { - bucket: "b".to_string(), - name: "obj".to_string(), - delete_marker: false, - replication_status: ReplicationStatusType::Pending, - ..Default::default() - }; - assert!( - !heal_should_use_check_replicate_delete(&oi), - "Pending non-delete-marker object must use must_replicate path to evaluate current target set" - ); - } - - #[test] - fn test_heal_should_use_check_replicate_delete_delete_marker() { - let oi = ObjectInfo { - bucket: "b".to_string(), - name: "obj".to_string(), - delete_marker: true, - replication_status: ReplicationStatusType::Failed, - ..Default::default() - }; - assert!( - heal_should_use_check_replicate_delete(&oi), - "Delete marker always uses check_replicate_delete path" - ); - } - - #[test] - fn test_heal_should_use_check_replicate_delete_version_purge_status_uses_delete_path() { - let oi = ObjectInfo { - bucket: "b".to_string(), - name: "obj".to_string(), - delete_marker: false, - version_purge_status: VersionPurgeStatusType::Pending, - ..Default::default() - }; - assert!( - heal_should_use_check_replicate_delete(&oi), - "Version purge entries must use check_replicate_delete path" - ); - } - - #[test] - fn test_heal_should_use_check_replicate_delete_completed_non_delete_marker_uses_must_replicate_path() { - let oi = ObjectInfo { - bucket: "b".to_string(), - name: "obj".to_string(), - delete_marker: false, - replication_status: ReplicationStatusType::Completed, - ..Default::default() - }; - assert!( - !heal_should_use_check_replicate_delete(&oi), - "Completed non-delete-marker object must use must_replicate path so new targets can be evaluated" - ); - } - #[test] fn test_get_replication_action_existing_object_source_newer_null_version_requires_replication() { let source = ObjectInfo { @@ -3872,56 +3724,6 @@ mod tests { ); } - #[test] - fn test_resync_target_includes_object_at_reset_before_boundary() { - let reset_before = OffsetDateTime::UNIX_EPOCH + Duration::seconds(30); - let oi = ObjectInfo { - mod_time: Some(reset_before), - ..Default::default() - }; - - let decision = resync_target(&oi, "arn:target", "reset-1", Some(reset_before), ReplicationStatusType::Completed); - - assert!( - decision.replicate, - "objects whose mod_time equals reset_before must be included in the reset window" - ); - } - - #[test] - fn test_resync_target_replicates_when_reset_id_changes() { - let reset_before = OffsetDateTime::UNIX_EPOCH + Duration::seconds(30); - let oi = ObjectInfo { - mod_time: Some(OffsetDateTime::UNIX_EPOCH + Duration::seconds(10)), - user_defined: Arc::new(HashMap::from([( - target_reset_header("arn:target"), - "1970-01-01T00:00:20Z;old-reset".to_string(), - )])), - ..Default::default() - }; - - let decision = resync_target(&oi, "arn:target", "new-reset", Some(reset_before), ReplicationStatusType::Completed); - - assert!(decision.replicate, "a new reset id must resync objects marked by an older reset"); - } - - #[test] - fn test_resync_target_skips_completed_object_for_same_reset_id() { - let reset_before = OffsetDateTime::UNIX_EPOCH + Duration::seconds(30); - let oi = ObjectInfo { - mod_time: Some(OffsetDateTime::UNIX_EPOCH + Duration::seconds(10)), - user_defined: Arc::new(HashMap::from([( - target_reset_header("arn:target"), - "1970-01-01T00:00:20Z;same-reset".to_string(), - )])), - ..Default::default() - }; - - let decision = resync_target(&oi, "arn:target", "same-reset", Some(reset_before), ReplicationStatusType::Completed); - - assert!(!decision.replicate, "the same completed reset id must not resync again"); - } - #[test] fn test_replicate_object_info_to_object_info_preserves_delete_marker_flag() { let live = ReplicateObjectInfo { @@ -3941,49 +3743,6 @@ mod tests { assert!(delete_marker.to_object_info().delete_marker); } - #[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 { diff --git a/crates/replication/src/lib.rs b/crates/replication/src/lib.rs index b5634c7fa..51ec64dbe 100644 --- a/crates/replication/src/lib.rs +++ b/crates/replication/src/lib.rs @@ -32,9 +32,12 @@ pub use delete::{ pub use mrf::{MrfOpKind, MrfReplicateEntry, decode_mrf_file, encode_mrf_file}; pub use object::{ ReplicationSourceObject, ReplicationTargetObject, content_matches_by_etag, replication_action_for_target, - target_is_newer_than_source_null_version, + replication_etags_match, target_is_newer_than_source_null_version, +}; +pub use operation::{ + MustReplicateOptions, ReplicationDeleteSource, ReplicationResyncTargetObject, delete_replication_missing_source_decision, + delete_replication_object_opts, heal_uses_delete_replication_path, is_ssec_encrypted, resync_target_for_object, }; -pub use operation::{MustReplicateOptions, is_ssec_encrypted}; pub use queue::{ReplicationHealQueueResult, ReplicationOperation, ReplicationPriority, ReplicationQueueAdmission}; pub use resync::{ BucketReplicationResyncStatus, Error, Result, ResyncOpts, ResyncStatusType, TargetReplicationResyncStatus, diff --git a/crates/replication/src/object.rs b/crates/replication/src/object.rs index 7596df784..16bb5ff0e 100644 --- a/crates/replication/src/object.rs +++ b/crates/replication/src/object.rs @@ -24,6 +24,7 @@ use std::collections::HashMap; use time::OffsetDateTime; const AMZ_META_PREFIX: &str = "X-Amz-Meta-"; +const CONTENT_ENCODING_LOWER: &str = "content-encoding"; const REPLICATION_METADATA_COMPARE_KEYS: [&str; 9] = [ EXPIRES, CACHE_CONTROL, @@ -62,8 +63,12 @@ pub struct ReplicationTargetObject<'a> { } pub fn content_matches_by_etag(source: &ReplicationSourceObject<'_>, target: &ReplicationTargetObject<'_>) -> bool { - let source_etag = source.etag.map(trim_etag); - let target_etag = target.etag.map(trim_etag); + replication_etags_match(source.etag, target.etag) +} + +pub fn replication_etags_match(source: Option<&str>, target: Option<&str>) -> bool { + let source_etag = source.map(trim_etag); + let target_etag = target.map(trim_etag); source_etag.is_some() && source_etag == target_etag } @@ -121,7 +126,7 @@ fn content_encoding_differs(source: &ReplicationSourceObject<'_>, target: &Repli .and_then(|metadata| { metadata .get(CONTENT_ENCODING) - .or_else(|| metadata.get(&CONTENT_ENCODING.to_lowercase())) + .or_else(|| metadata.get(CONTENT_ENCODING_LOWER)) }) .is_none_or(|enc| enc != content_encoding); } @@ -130,14 +135,11 @@ fn content_encoding_differs(source: &ReplicationSourceObject<'_>, target: &Repli fn tag_metadata_differs(source: &ReplicationSourceObject<'_>, target: &ReplicationTargetObject<'_>) -> bool { let source_tags = ReplicationTagFilter::decode_tags_to_map(source.user_tags); - let target_tags = ReplicationTagFilter::decode_tags_to_map( - target - .metadata - .and_then(|metadata| metadata.get(AMZ_OBJECT_TAGGING)) - .cloned() - .unwrap_or_default() - .as_str(), - ); + let target_tagging = target + .metadata + .and_then(|metadata| metadata.get(AMZ_OBJECT_TAGGING).map(String::as_str)) + .unwrap_or_default(); + let target_tags = ReplicationTagFilter::decode_tags_to_map(target_tagging); let source_tag_count = match i32::try_from(source_tags.len()) { Ok(count) => count, Err(_) => i32::MAX, @@ -163,7 +165,7 @@ fn comparable_metadata(metadata: Option<&HashMap>) -> HashMap) -> bool { || user_defined.contains_key(SSEC_KEY_MD5_HEADER) } +#[derive(Debug, Clone)] +pub struct ReplicationDeleteSource<'a> { + pub user_defined: &'a HashMap, + pub user_tags: &'a str, + pub delete_marker: bool, + pub replication_status: ReplicationStatusType, +} + +pub fn delete_replication_object_opts(dobj: &ObjectToDelete, source: &ReplicationDeleteSource<'_>) -> ObjectOpts { + ObjectOpts { + name: dobj.object_name.clone(), + ssec: is_ssec_encrypted(source.user_defined), + user_tags: source.user_tags.to_string(), + delete_marker: source.delete_marker, + version_id: dobj.version_id, + op_type: ReplicationType::Delete, + replica: source.replication_status == ReplicationStatusType::Replica, + ..Default::default() + } +} + +pub fn heal_uses_delete_replication_path(delete_marker: bool, version_purge_status: &VersionPurgeStatusType) -> bool { + delete_marker || !version_purge_status.is_empty() +} + +pub fn delete_replication_missing_source_decision( + delete_marker: bool, + target_status: ReplicationStatusType, + rule_replicates: bool, + version_purge_status: &VersionPurgeStatusType, +) -> Option { + let valid_replication_status = matches!( + target_status, + ReplicationStatusType::Pending | ReplicationStatusType::Completed | ReplicationStatusType::Failed + ); + + if delete_marker && (valid_replication_status || rule_replicates) { + return Some(rule_replicates); + } + + if !version_purge_status.is_empty() { + return Some(matches!( + version_purge_status, + &VersionPurgeStatusType::Pending | &VersionPurgeStatusType::Failed + )); + } + + None +} + +#[derive(Debug, Clone)] +pub struct ReplicationResyncTargetObject<'a> { + pub mod_time: Option, + pub user_defined: &'a HashMap, +} + +pub fn resync_target_for_object( + object: &ReplicationResyncTargetObject<'_>, + arn: &str, + reset_id: &str, + reset_before_date: Option, + status: ReplicationStatusType, +) -> ResyncTargetDecision { + let rs = object + .user_defined + .get(target_reset_header(arn).as_str()) + .cloned() + .or_else(|| get_header_map(object.user_defined, SUFFIX_REPLICATION_RESET_STATUS)); + + let mut dec = ResyncTargetDecision::default(); + + let mod_time = object.mod_time.unwrap_or(OffsetDateTime::UNIX_EPOCH); + + let Some(rs) = rs else { + let reset_before_date = reset_before_date.unwrap_or(OffsetDateTime::UNIX_EPOCH); + if !reset_id.is_empty() && mod_time <= reset_before_date { + dec.replicate = true; + return dec; + } + + dec.replicate = status == ReplicationStatusType::Empty; + + return dec; + }; + + let Some(reset_before_date) = reset_before_date else { + return dec; + }; + + if reset_id.is_empty() { + return dec; + } + + let parts: Vec<&str> = rs.splitn(2, ';').collect(); + + if parts.len() != 2 { + return dec; + } + + let new_reset = parts[1] != reset_id; + + if !new_reset && status == ReplicationStatusType::Completed { + return dec; + } + + dec.replicate = new_reset && mod_time <= reset_before_date; + + dec +} + #[cfg(test)] mod tests { - use super::{MustReplicateOptions, is_ssec_encrypted}; - use crate::{ReplicationStatusType, ReplicationType}; + use super::{ + MustReplicateOptions, ReplicationDeleteSource, ReplicationResyncTargetObject, delete_replication_missing_source_decision, + delete_replication_object_opts, heal_uses_delete_replication_path, is_ssec_encrypted, resync_target_for_object, + }; + use crate::{ReplicationStatusType, ReplicationType, VersionPurgeStatusType, target_reset_header}; + use rustfs_storage_api::ObjectToDelete; use rustfs_utils::http::{AMZ_BUCKET_REPLICATION_STATUS, SSEC_ALGORITHM_HEADER}; use std::collections::HashMap; + use time::{Duration, OffsetDateTime}; + use uuid::Uuid; #[test] fn must_replicate_options_preserves_user_tags_and_operation_type() { @@ -102,4 +223,113 @@ mod tests { assert!(is_ssec_encrypted(&meta)); assert!(!is_ssec_encrypted(&HashMap::new())); } + + #[test] + fn heal_delete_path_is_limited_to_delete_markers_and_version_purges() { + assert!(heal_uses_delete_replication_path(true, &VersionPurgeStatusType::Empty)); + assert!(heal_uses_delete_replication_path(false, &VersionPurgeStatusType::Pending)); + assert!(!heal_uses_delete_replication_path(false, &VersionPurgeStatusType::Empty)); + } + + #[test] + fn missing_source_delete_decision_keeps_delete_marker_and_purge_rules() { + assert_eq!( + delete_replication_missing_source_decision( + true, + ReplicationStatusType::Completed, + false, + &VersionPurgeStatusType::Empty, + ), + Some(false) + ); + assert_eq!( + delete_replication_missing_source_decision( + false, + ReplicationStatusType::Empty, + false, + &VersionPurgeStatusType::Pending, + ), + Some(true) + ); + assert_eq!( + delete_replication_missing_source_decision( + false, + ReplicationStatusType::Empty, + false, + &VersionPurgeStatusType::Empty + ), + None + ); + } + + #[test] + fn delete_replication_object_opts_marks_replica_deletes() { + let version_id = Uuid::new_v4(); + let user_defined = HashMap::from([(SSEC_ALGORITHM_HEADER.to_string(), "AES256".to_string())]); + let dobj = ObjectToDelete { + object_name: "obj".to_string(), + version_id: Some(version_id), + ..Default::default() + }; + let source = ReplicationDeleteSource { + user_defined: &user_defined, + user_tags: "env=prod", + delete_marker: true, + replication_status: ReplicationStatusType::Replica, + }; + + let opts = delete_replication_object_opts(&dobj, &source); + + assert!(opts.replica); + assert!(opts.ssec); + assert_eq!(opts.user_tags, "env=prod"); + assert_eq!(opts.version_id, Some(version_id)); + assert_eq!(opts.name, "obj"); + assert_eq!(opts.op_type, ReplicationType::Delete); + } + + #[test] + fn resync_target_includes_object_at_reset_before_boundary() { + let reset_before = OffsetDateTime::UNIX_EPOCH + Duration::seconds(30); + let user_defined = HashMap::new(); + let object = ReplicationResyncTargetObject { + mod_time: Some(reset_before), + user_defined: &user_defined, + }; + + let decision = + resync_target_for_object(&object, "arn:target", "reset-1", Some(reset_before), ReplicationStatusType::Completed); + + assert!(decision.replicate); + } + + #[test] + fn resync_target_replicates_when_reset_id_changes() { + let reset_before = OffsetDateTime::UNIX_EPOCH + Duration::seconds(30); + let user_defined = HashMap::from([(target_reset_header("arn:target"), "1970-01-01T00:00:20Z;old-reset".to_string())]); + let object = ReplicationResyncTargetObject { + mod_time: Some(OffsetDateTime::UNIX_EPOCH + Duration::seconds(10)), + user_defined: &user_defined, + }; + + let decision = + resync_target_for_object(&object, "arn:target", "new-reset", Some(reset_before), ReplicationStatusType::Completed); + + assert!(decision.replicate); + } + + #[test] + fn resync_target_skips_completed_object_for_same_reset_id() { + let reset_before = OffsetDateTime::UNIX_EPOCH + Duration::seconds(30); + let user_defined = HashMap::from([(target_reset_header("arn:target"), "1970-01-01T00:00:20Z;same-reset".to_string())]); + let object = ReplicationResyncTargetObject { + mod_time: Some(OffsetDateTime::UNIX_EPOCH + Duration::seconds(10)), + user_defined: &user_defined, + }; + + let decision = + resync_target_for_object(&object, "arn:target", "same-reset", Some(reset_before), ReplicationStatusType::Completed); + + assert!(!decision.replicate); + } } diff --git a/scripts/check_architecture_migration_rules.sh b/scripts/check_architecture_migration_rules.sh index b182331eb..7b2ab7c85 100755 --- a/scripts/check_architecture_migration_rules.sh +++ b/scripts/check_architecture_migration_rules.sh @@ -2568,7 +2568,7 @@ fi ( cd "$ROOT_DIR" replication_operation_status=0 - rg -n --with-filename '^\s*(?:pub(?:\([^)]*\))?\s+)?(?:struct\s+MustReplicateOptions|fn\s+is_ssec_encrypted)\b' \ + rg -n --with-filename '^\s*(?:pub(?:\([^)]*\))?\s+)?(?:(?:struct)\s+(?:MustReplicateOptions|ReplicationDeleteSource|ReplicationResyncTargetObject)|fn\s+(?:is_ssec_encrypted|delete_replication_object_opts|heal_uses_delete_replication_path|delete_replication_missing_source_decision|resync_target_for_object|resync_target))\b' \ crates/ecstore/src/bucket/replication \ --glob '*.rs' >"$REPLICATION_OPERATION_CONTRACT_BACKSLIDE_HITS_FILE" || replication_operation_status=$? if [[ "$replication_operation_status" -ne 0 && "$replication_operation_status" -ne 1 ]]; then @@ -2658,7 +2658,7 @@ fi ( cd "$ROOT_DIR" replication_object_compare_status=0 - rg -n --with-filename '^\s*(?:pub(?:\([^)]*\))?\s+)?(?:(?:struct)\s+(?:ReplicationSourceObject|ReplicationTargetObject)|fn\s+(?:content_matches_by_etag|content_matches|target_is_newer_than_source_null_version|replication_action_for_target|get_replication_action))\b' \ + rg -n --with-filename '^\s*(?:pub(?:\([^)]*\))?\s+)?(?:(?:struct)\s+(?:ReplicationSourceObject|ReplicationTargetObject)|fn\s+(?:replication_etags_match|content_matches_by_etag|content_matches|target_is_newer_than_source_null_version|replication_action_for_target|get_replication_action))\b' \ crates/ecstore/src/bucket/replication \ --glob '*.rs' >"$REPLICATION_OBJECT_COMPARE_CONTRACT_BACKSLIDE_HITS_FILE" || replication_object_compare_status=$? if [[ "$replication_object_compare_status" -ne 0 && "$replication_object_compare_status" -ne 1 ]]; then