diff --git a/crates/ecstore/src/bucket/replication/replication_object_config.rs b/crates/ecstore/src/bucket/replication/replication_object_config.rs index c09db31ce..e8dbe33e4 100644 --- a/crates/ecstore/src/bucket/replication/replication_object_config.rs +++ b/crates/ecstore/src/bucket/replication/replication_object_config.rs @@ -566,7 +566,7 @@ pub(crate) async fn must_replicate(bucket: &str, object: &str, mopts: MustReplic let mut sopts = opts.clone(); sopts.target_arn = arn.clone(); - let replicate = cfg.replicate(&sopts); + let replicate = cfg.replicate(&sopts) && mopts.metadata_target_is_eligible(&arn); let synchronous = if let Some(cli) = cli { cli.replicate_sync } else { false }; dsc.set(ReplicateTargetDecision::new(arn, replicate, synchronous)); diff --git a/crates/replication/src/http.rs b/crates/replication/src/http.rs index d11d62ee0..84e9e1a84 100644 --- a/crates/replication/src/http.rs +++ b/crates/replication/src/http.rs @@ -34,6 +34,7 @@ pub(crate) const SSEC_ALGORITHM_HEADER: &str = "x-amz-server-side-encryption-cus pub(crate) const SSEC_KEY_HEADER: &str = "x-amz-server-side-encryption-customer-key"; pub(crate) const SSEC_KEY_MD5_HEADER: &str = "x-amz-server-side-encryption-customer-key-md5"; pub(crate) const SUFFIX_ACTUAL_SIZE: &str = "actual-size"; +pub(crate) const SUFFIX_REPLICATION_STATUS: &str = "replication-status"; pub(crate) const SUFFIX_REPLICATION_RESET_STATUS: &str = "replication-reset-status"; fn internal_keys(suffix: &str) -> (String, String) { diff --git a/crates/replication/src/operation.rs b/crates/replication/src/operation.rs index 22a130432..b3476f771 100644 --- a/crates/replication/src/operation.rs +++ b/crates/replication/src/operation.rs @@ -16,7 +16,7 @@ use std::collections::HashMap; use crate::http::{ AMZ_BUCKET_REPLICATION_STATUS, AMZ_OBJECT_TAGGING, SSEC_ALGORITHM_HEADER, SSEC_KEY_HEADER, SSEC_KEY_MD5_HEADER, - SUFFIX_REPLICATION_RESET_STATUS, get_header_metadata, + SUFFIX_REPLICATION_RESET_STATUS, SUFFIX_REPLICATION_STATUS, get_header_metadata, get_internal_metadata, }; use s3s::dto::ReplicationConfiguration; use time::OffsetDateTime; @@ -82,6 +82,22 @@ impl MustReplicateOptions { pub fn user_tags(&self) -> &str { self.meta.get(AMZ_OBJECT_TAGGING).map(String::as_str).unwrap_or_default() } + + pub fn metadata_target_is_eligible(&self, arn: &str) -> bool { + if !self.is_metadata_replication() { + return true; + } + + get_internal_metadata(&self.meta, SUFFIX_REPLICATION_STATUS) + .as_deref() + .and_then(|statuses| { + statuses.split(';').find_map(|entry| { + let (target_arn, status) = entry.split_once('=')?; + (target_arn == arn).then(|| ReplicationStatusType::from(status)) + }) + }) + == Some(ReplicationStatusType::Completed) + } } pub fn is_ssec_encrypted(user_defined: &HashMap) -> bool { @@ -308,7 +324,9 @@ mod tests { is_ssec_encrypted, resync_target_for_object, should_schedule_delete_replication, should_use_existing_delete_replication_info, should_use_existing_delete_replication_source, }; - use crate::http::{AMZ_BUCKET_REPLICATION_STATUS, SSEC_ALGORITHM_HEADER}; + use crate::http::{ + AMZ_BUCKET_REPLICATION_STATUS, SSEC_ALGORITHM_HEADER, SUFFIX_REPLICATION_STATUS, insert_internal_metadata, + }; use crate::storage_api::ObjectToDelete; use crate::{ReplicationStatusType, ReplicationType, VersionPurgeStatusType, target_reset_header}; use s3s::dto::{ @@ -346,6 +364,29 @@ mod tests { assert_eq!(options.replication_status(), ReplicationStatusType::Replica); } + #[test] + fn metadata_replication_requires_completed_target_state() { + let arn = "arn:rustfs:replication:target"; + for status in ["PENDING", "FAILED"] { + let mut meta = HashMap::new(); + insert_internal_metadata(&mut meta, SUFFIX_REPLICATION_STATUS, format!("{arn}={status};")); + let options = MustReplicateOptions::new(&meta, String::new(), ReplicationType::Metadata, false); + + assert!(!options.metadata_target_is_eligible(arn)); + } + + let mut meta = HashMap::new(); + insert_internal_metadata(&mut meta, SUFFIX_REPLICATION_STATUS, format!("{arn}=COMPLETED;")); + let options = MustReplicateOptions::new(&meta, String::new(), ReplicationType::Metadata, false); + + assert!(options.metadata_target_is_eligible(arn)); + assert!(!options.metadata_target_is_eligible("arn:rustfs:replication:missing")); + assert!( + MustReplicateOptions::new(&HashMap::new(), String::new(), ReplicationType::Object, false) + .metadata_target_is_eligible(arn) + ); + } + #[test] fn ssec_detection_uses_existing_metadata_headers() { let meta = HashMap::from([(SSEC_ALGORITHM_HEADER.to_string(), "AES256".to_string())]);