diff --git a/crates/ecstore/src/bucket/replication/replication_resyncer.rs b/crates/ecstore/src/bucket/replication/replication_resyncer.rs index 1da8b7175..f8ac16aed 100644 --- a/crates/ecstore/src/bucket/replication/replication_resyncer.rs +++ b/crates/ecstore/src/bucket/replication/replication_resyncer.rs @@ -20,9 +20,9 @@ use super::replication_error_boundary::{Error, Result, is_err_object_not_found, use super::replication_event_sink::{EventArgs, send_event, send_local_event}; use super::replication_filemeta_boundary::{ MrfOpKind, MrfReplicateEntry, REPLICATE_EXISTING, REPLICATE_EXISTING_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, + ReplicateTargetDecision, ReplicatedInfos, ReplicatedTargetInfo, ReplicationAction, ReplicationStatusType, ReplicationType, + ReplicationWorkerOperation, ResyncDecision, ResyncTargetDecision, 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; @@ -53,18 +53,19 @@ use headers::{ use http::HeaderMap; use http_body::Frame; use http_body_util::StreamBody; -use regex::Regex; +#[cfg(test)] +use rmp_serde; use rustfs_replication::{BucketReplicationResyncStatus, ResyncOpts, TargetReplicationResyncStatus}; use rustfs_s3_types::EventName; use rustfs_utils::http::{ AMZ_BUCKET_REPLICATION_STATUS, AMZ_OBJECT_TAGGING, AMZ_TAGGING_DIRECTIVE, CONTENT_ENCODING, HeaderExt as _, SSEC_ALGORITHM_HEADER, SSEC_KEY_HEADER, SSEC_KEY_MD5_HEADER, SUFFIX_OBJECTLOCK_LEGALHOLD_TIMESTAMP, - SUFFIX_OBJECTLOCK_RETENTION_TIMESTAMP, SUFFIX_REPLICATION_RESET, SUFFIX_REPLICATION_RESET_ARN_PREFIX, - SUFFIX_REPLICATION_STATUS, SUFFIX_TAGGING_TIMESTAMP, headers, + SUFFIX_OBJECTLOCK_RETENTION_TIMESTAMP, SUFFIX_REPLICATION_RESET, 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, internal_key_strip_suffix_prefix, is_internal_key, + has_internal_suffix, insert_header_map, insert_str, is_internal_key, }; use rustfs_utils::string::strings_has_prefix_fold; use rustfs_utils::{DEFAULT_SIP_HASH_KEY, sip_hash}; @@ -1359,48 +1360,6 @@ fn is_ssec_encrypted(user_defined: &HashMap) -> bool { || user_defined.contains_key(SSEC_KEY_MD5_HEADER) } -/// Extension trait for ObjectInfo to add replication-related methods -pub trait ObjectInfoExt { - fn target_replication_status(&self, arn: &str) -> ReplicationStatusType; - fn replication_state(&self) -> ReplicationState; -} - -impl ObjectInfoExt for ObjectInfo { - /// Returns replication status of a target - fn target_replication_status(&self, arn: &str) -> ReplicationStatusType { - lazy_static::lazy_static! { - static ref REPL_STATUS_REGEX: Regex = Regex::new(r"([^=].*?)=([^,].*?);").unwrap(); - } - - let binding = self.replication_status_internal.clone().unwrap_or_default(); - let captures = REPL_STATUS_REGEX.captures_iter(&binding); - for cap in captures { - if cap.len() == 3 && &cap[1] == arn { - return ReplicationStatusType::from(&cap[2]); - } - } - ReplicationStatusType::default() - } - - fn replication_state(&self) -> ReplicationState { - ReplicationState { - replication_status_internal: self.replication_status_internal.clone(), - version_purge_status_internal: self.version_purge_status_internal.clone(), - replicate_decision_str: self.replication_decision.clone(), - targets: replication_statuses_map(&self.replication_status_internal.clone().unwrap_or_default()), - purge_targets: version_purge_statuses_map(&self.version_purge_status_internal.clone().unwrap_or_default()), - reset_statuses_map: self - .user_defined - .iter() - .filter_map(|(k, v)| { - internal_key_strip_suffix_prefix(k, SUFFIX_REPLICATION_RESET_ARN_PREFIX).map(|arn| (arn, v.clone())) - }) - .collect(), - ..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(); diff --git a/crates/ecstore/src/object_api/types.rs b/crates/ecstore/src/object_api/types.rs index a37fea71f..2b91e6807 100644 --- a/crates/ecstore/src/object_api/types.rs +++ b/crates/ecstore/src/object_api/types.rs @@ -747,6 +747,13 @@ impl ObjectInfo { } } + pub fn target_replication_status(&self, arn: &str) -> ReplicationStatusType { + replication_statuses_map(self.replication_status_internal.as_deref().unwrap_or_default()) + .get(arn) + .cloned() + .unwrap_or_default() + } + pub fn decrypt_checksums(&self, part: usize, _headers: &HeaderMap) -> Result<(HashMap, bool)> { if part > 0 && let Some(checksums) = self.parts.iter().find(|p| p.number == part).and_then(|p| p.checksums.clone()) @@ -823,6 +830,25 @@ mod tests { assert_eq!(state.composite_replication_status(), ReplicationStatusType::Replica); } + #[test] + fn object_info_replication_helpers_parse_target_status_and_reset_headers() { + let reset_key = rustfs_utils::http::internal_key_rustfs("replication-reset-arn:target-a"); + let object = ObjectInfo { + replication_status_internal: Some("arn:target-a=COMPLETED;arn:target-b=FAILED;".to_string()), + version_purge_status_internal: Some("arn:target-a=PENDING;".to_string()), + user_defined: Arc::new(HashMap::from([(reset_key, "reset-id".to_string())])), + ..Default::default() + }; + + let state = object.replication_state(); + + assert_eq!(object.target_replication_status("arn:target-a"), ReplicationStatusType::Completed); + assert_eq!(object.target_replication_status("arn:missing"), ReplicationStatusType::Empty); + assert_eq!(state.targets.get("arn:target-b"), Some(&ReplicationStatusType::Failed)); + assert_eq!(state.purge_targets.get("arn:target-a"), Some(&VersionPurgeStatusType::Pending)); + assert_eq!(state.reset_statuses_map.get("arn:target-a"), Some(&"reset-id".to_string())); + } + #[test] fn versions_after_marker_handles_uuid_version_marker() { let first_version = Uuid::parse_str("11111111-2222-3333-4444-555555555555").unwrap();