From df6ce41acd0f3eb9027da90b09de8e1ab24ce530 Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Thu, 2 Jul 2026 10:48:02 +0800 Subject: [PATCH] Move replication operation contracts into crate (#4165) refactor(replication): move operation contracts into crate --- Cargo.lock | 1 + crates/ecstore/src/bucket/replication/mod.rs | 4 +- .../replication/replication_object_bridge.rs | 3 +- .../replication/replication_resyncer.rs | 90 ++------------- crates/replication/Cargo.toml | 1 + crates/replication/src/lib.rs | 2 + crates/replication/src/operation.rs | 105 ++++++++++++++++++ scripts/check_architecture_migration_rules.sh | 16 +++ 8 files changed, 140 insertions(+), 82 deletions(-) create mode 100644 crates/replication/src/operation.rs diff --git a/Cargo.lock b/Cargo.lock index 446dbcc44..b5df4b90f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -9897,6 +9897,7 @@ dependencies = [ "rmp", "rmp-serde", "rustfs-filemeta", + "rustfs-utils", "s3s", "serde", "time", diff --git a/crates/ecstore/src/bucket/replication/mod.rs b/crates/ecstore/src/bucket/replication/mod.rs index 61af5ea7b..5b54f4d56 100644 --- a/crates/ecstore/src/bucket/replication/mod.rs +++ b/crates/ecstore/src/bucket/replication/mod.rs @@ -45,9 +45,9 @@ pub use replication_pool::{ DynReplicationPool, ReplicationHealQueueResult, ReplicationPoolTrait, ReplicationQueueAdmission, get_global_replication_pool, get_global_replication_stats, init_background_replication, }; -pub use replication_resyncer::{DeletedObjectReplicationInfo, MustReplicateOptions, ReplicationConfig}; +pub use replication_resyncer::{DeletedObjectReplicationInfo, ReplicationConfig}; pub use replication_scanner_bridge::ReplicationScannerBridge; pub use replication_state::{BucketStats, ReplicationStats}; pub use replication_storage_boundary::{ReplicationObjectIO, ReplicationStorage}; pub(crate) use replication_target_config_bridge::ReplicationTargetConfigBridge; -pub use rustfs_replication::{BucketReplicationResyncStatus, ResyncOpts, TargetReplicationResyncStatus}; +pub use rustfs_replication::{BucketReplicationResyncStatus, MustReplicateOptions, ResyncOpts, TargetReplicationResyncStatus}; diff --git a/crates/ecstore/src/bucket/replication/replication_object_bridge.rs b/crates/ecstore/src/bucket/replication/replication_object_bridge.rs index ea72e9665..60a946f35 100644 --- a/crates/ecstore/src/bucket/replication/replication_object_bridge.rs +++ b/crates/ecstore/src/bucket/replication/replication_object_bridge.rs @@ -17,9 +17,10 @@ use std::{collections::HashMap, sync::Arc}; use super::replication_filemeta_boundary::{ReplicateDecision, ReplicationStatusType, ReplicationType}; use super::replication_pool::{schedule_replication, schedule_replication_delete}; use super::replication_resyncer::{ - DeletedObjectReplicationInfo, MustReplicateOptions, check_replicate_delete, get_must_replicate_options, must_replicate, + DeletedObjectReplicationInfo, check_replicate_delete, get_must_replicate_options, must_replicate, }; use super::replication_storage_boundary::{ObjectInfo, ObjectOptions, ObjectToDelete, ReplicationStorage}; +use rustfs_replication::MustReplicateOptions; pub struct ReplicationObjectBridge; diff --git a/crates/ecstore/src/bucket/replication/replication_resyncer.rs b/crates/ecstore/src/bucket/replication/replication_resyncer.rs index f8ac16aed..c9ab0fc08 100644 --- a/crates/ecstore/src/bucket/replication/replication_resyncer.rs +++ b/crates/ecstore/src/bucket/replication/replication_resyncer.rs @@ -55,13 +55,12 @@ use http_body::Frame; use http_body_util::StreamBody; #[cfg(test)] use rmp_serde; -use rustfs_replication::{BucketReplicationResyncStatus, ResyncOpts, TargetReplicationResyncStatus}; +use rustfs_replication::{BucketReplicationResyncStatus, MustReplicateOptions, 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_STATUS, SUFFIX_TAGGING_TIMESTAMP, - headers, + SUFFIX_OBJECTLOCK_LEGALHOLD_TIMESTAMP, 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, @@ -909,13 +908,7 @@ pub async fn get_heal_replicate_object_info(oi: &ObjectInfo, rcfg: &ReplicationC must_replicate( oi.bucket.as_str(), &oi.name, - MustReplicateOptions::new( - &user_defined, - (*oi.user_tags).clone(), - ReplicationStatusType::Empty, - ReplicationType::Heal, - ObjectOptions::default(), - ), + MustReplicateOptions::new(&user_defined, (*oi.user_tags).clone(), ReplicationType::Heal, false), ) .await }; @@ -1089,13 +1082,7 @@ impl ReplicationConfig { let dsc = must_replicate( oi.bucket.as_str(), &oi.name, - MustReplicateOptions::new( - &user_defined, - (*oi.user_tags).clone(), - ReplicationStatusType::Empty, - ReplicationType::ExistingObject, - ObjectOptions::default(), - ), + MustReplicateOptions::new(&user_defined, (*oi.user_tags).clone(), ReplicationType::ExistingObject, false), ) .await; @@ -1192,62 +1179,14 @@ pub fn resync_target( dec } -pub struct MustReplicateOptions { - meta: HashMap, - status: ReplicationStatusType, - op_type: ReplicationType, - replication_request: bool, -} - -impl MustReplicateOptions { - pub fn new( - meta: &HashMap, - user_tags: String, - status: ReplicationStatusType, - op_type: ReplicationType, - opts: ObjectOptions, - ) -> Self { - let mut meta = meta.clone(); - if !user_tags.is_empty() { - meta.insert(AMZ_OBJECT_TAGGING.to_string(), user_tags); - } - - Self { - meta, - status, - op_type, - replication_request: opts.replication_request, - } - } - - pub fn from_object_info(oi: &ObjectInfo, op_type: ReplicationType, opts: ObjectOptions) -> Self { - Self::new(&oi.user_defined, (*oi.user_tags).clone(), oi.replication_status.clone(), op_type, opts) - } - - pub fn replication_status(&self) -> ReplicationStatusType { - if let Some(rs) = self.meta.get(AMZ_BUCKET_REPLICATION_STATUS) { - return ReplicationStatusType::from(rs.as_str()); - } - ReplicationStatusType::default() - } - - pub fn is_existing_object_replication(&self) -> bool { - self.op_type == ReplicationType::ExistingObject - } - - pub fn is_metadata_replication(&self) -> bool { - self.op_type == ReplicationType::Metadata - } -} - pub(crate) fn get_must_replicate_options( user_defined: &HashMap, user_tags: String, - status: ReplicationStatusType, + _status: ReplicationStatusType, op_type: ReplicationType, opts: ObjectOptions, ) -> MustReplicateOptions { - MustReplicateOptions::new(user_defined, user_tags, status, op_type, opts) + MustReplicateOptions::new(user_defined, user_tags, op_type, opts.replication_request) } /// Returns whether object version is a delete marker and if object qualifies for replication @@ -1343,7 +1282,7 @@ pub(crate) async fn check_replicate_delete( fn delete_replication_object_opts(dobj: &ObjectToDelete, oi: &ObjectInfo) -> ObjectOpts { ObjectOpts { name: dobj.object_name.clone(), - ssec: is_ssec_encrypted(&oi.user_defined), + 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, @@ -1353,13 +1292,6 @@ fn delete_replication_object_opts(dobj: &ObjectToDelete, oi: &ObjectInfo) -> Obj } } -/// 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) - || user_defined.contains_key(SSEC_KEY_HEADER) - || user_defined.contains_key(SSEC_KEY_MD5_HEADER) -} - pub(crate) async fn must_replicate(bucket: &str, object: &str, mopts: MustReplicateOptions) -> ReplicateDecision { if runtime_sources::object_store_handle().is_none() { return ReplicateDecision::default(); @@ -1375,7 +1307,7 @@ pub(crate) async fn must_replicate(bucket: &str, object: &str, mopts: MustReplic return ReplicateDecision::default(); } - if mopts.replication_request { + if mopts.is_replication_request() { return ReplicateDecision::default(); } @@ -1396,7 +1328,7 @@ pub(crate) async fn must_replicate(bucket: &str, object: &str, mopts: MustReplic name: object.to_string(), replica: replication_status == ReplicationStatusType::Replica, existing_object: mopts.is_existing_object_replication(), - user_tags: mopts.meta.get(AMZ_OBJECT_TAGGING).map(|s| s.to_string()).unwrap_or_default(), + user_tags: mopts.user_tags().to_string(), ..Default::default() }; @@ -3359,7 +3291,7 @@ fn put_replication_opts(sc: &str, object_info: &ObjectInfo) -> Result<(PutObject }; let mut meta = HashMap::new(); - let is_ssec = is_ssec_encrypted(&object_info.user_defined); + let is_ssec = rustfs_replication::is_ssec_encrypted(&object_info.user_defined); // Process user-defined metadata for (k, v) in object_info.user_defined.iter() { diff --git a/crates/replication/Cargo.toml b/crates/replication/Cargo.toml index 9e6bf9dff..96b5f402a 100644 --- a/crates/replication/Cargo.toml +++ b/crates/replication/Cargo.toml @@ -30,6 +30,7 @@ byteorder.workspace = true rmp.workspace = true rmp-serde.workspace = true rustfs-filemeta.workspace = true +rustfs-utils = { workspace = true, features = ["http"] } s3s.workspace = true serde.workspace = true time.workspace = true diff --git a/crates/replication/src/lib.rs b/crates/replication/src/lib.rs index b138c5ed2..ba7e8a4ca 100644 --- a/crates/replication/src/lib.rs +++ b/crates/replication/src/lib.rs @@ -14,12 +14,14 @@ pub mod config; pub mod mrf; +pub mod operation; pub mod resync; pub mod rule; pub mod tagging; pub use config::{ObjectOpts, ReplicationConfigurationExt}; pub use mrf::{MrfOpKind, MrfReplicateEntry, decode_mrf_file, encode_mrf_file}; +pub use operation::{MustReplicateOptions, is_ssec_encrypted}; pub use resync::{ BucketReplicationResyncStatus, Error, Result, ResyncOpts, ResyncStatusType, TargetReplicationResyncStatus, decode_resync_file, encode_resync_file, diff --git a/crates/replication/src/operation.rs b/crates/replication/src/operation.rs new file mode 100644 index 000000000..5ca8e074e --- /dev/null +++ b/crates/replication/src/operation.rs @@ -0,0 +1,105 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use std::collections::HashMap; + +use rustfs_utils::http::{ + AMZ_BUCKET_REPLICATION_STATUS, AMZ_OBJECT_TAGGING, SSEC_ALGORITHM_HEADER, SSEC_KEY_HEADER, SSEC_KEY_MD5_HEADER, +}; + +use crate::{ReplicationStatusType, ReplicationType}; + +#[derive(Debug, Clone, Default)] +pub struct MustReplicateOptions { + meta: HashMap, + op_type: ReplicationType, + replication_request: bool, +} + +impl MustReplicateOptions { + pub fn new(meta: &HashMap, user_tags: String, op_type: ReplicationType, replication_request: bool) -> Self { + let mut meta = meta.clone(); + if !user_tags.is_empty() { + meta.insert(AMZ_OBJECT_TAGGING.to_string(), user_tags); + } + + Self { + meta, + op_type, + replication_request, + } + } + + pub fn replication_status(&self) -> ReplicationStatusType { + if let Some(rs) = self.meta.get(AMZ_BUCKET_REPLICATION_STATUS) { + return ReplicationStatusType::from(rs.as_str()); + } + ReplicationStatusType::default() + } + + pub fn is_existing_object_replication(&self) -> bool { + self.op_type == ReplicationType::ExistingObject + } + + pub fn is_metadata_replication(&self) -> bool { + self.op_type == ReplicationType::Metadata + } + + pub fn is_replication_request(&self) -> bool { + self.replication_request + } + + pub fn user_tags(&self) -> &str { + self.meta.get(AMZ_OBJECT_TAGGING).map(String::as_str).unwrap_or_default() + } +} + +pub fn is_ssec_encrypted(user_defined: &HashMap) -> bool { + user_defined.contains_key(SSEC_ALGORITHM_HEADER) + || user_defined.contains_key(SSEC_KEY_HEADER) + || user_defined.contains_key(SSEC_KEY_MD5_HEADER) +} + +#[cfg(test)] +mod tests { + use super::{MustReplicateOptions, is_ssec_encrypted}; + use crate::{ReplicationStatusType, ReplicationType}; + use rustfs_utils::http::{AMZ_BUCKET_REPLICATION_STATUS, SSEC_ALGORITHM_HEADER}; + use std::collections::HashMap; + + #[test] + fn must_replicate_options_preserves_user_tags_and_operation_type() { + let options = MustReplicateOptions::new(&HashMap::new(), "env=prod".to_string(), ReplicationType::ExistingObject, true); + + assert_eq!(options.user_tags(), "env=prod"); + assert!(options.is_existing_object_replication()); + assert!(options.is_replication_request()); + } + + #[test] + fn must_replicate_options_reads_replication_status_header() { + let meta = HashMap::from([(AMZ_BUCKET_REPLICATION_STATUS.to_string(), "COMPLETED".to_string())]); + let options = MustReplicateOptions::new(&meta, String::new(), ReplicationType::Object, false); + + assert_eq!(options.replication_status(), ReplicationStatusType::Completed); + } + + #[test] + fn ssec_detection_uses_existing_metadata_headers() { + let meta = HashMap::from([(SSEC_ALGORITHM_HEADER.to_string(), "AES256".to_string())]); + + assert!(is_ssec_encrypted(&meta)); + assert!(!is_ssec_encrypted(&HashMap::new())); + } +} diff --git a/scripts/check_architecture_migration_rules.sh b/scripts/check_architecture_migration_rules.sh index 3c3f12166..2e30a944c 100755 --- a/scripts/check_architecture_migration_rules.sh +++ b/scripts/check_architecture_migration_rules.sh @@ -200,6 +200,7 @@ EXTERNAL_ECSTORE_API_BOUNDARY_HITS_FILE="${TMP_DIR}/external_ecstore_api_boundar REPLICATION_FACADE_BYPASS_HITS_FILE="${TMP_DIR}/replication_facade_bypass_hits.txt" REPLICATION_FACADE_WILDCARD_EXPORT_HITS_FILE="${TMP_DIR}/replication_facade_wildcard_export_hits.txt" REPLICATION_CONFIG_RULE_CONTRACT_BACKSLIDE_HITS_FILE="${TMP_DIR}/replication_config_rule_contract_backslide_hits.txt" +REPLICATION_OPERATION_CONTRACT_BACKSLIDE_HITS_FILE="${TMP_DIR}/replication_operation_contract_backslide_hits.txt" REPLICATION_RESYNC_CONTRACT_BACKSLIDE_HITS_FILE="${TMP_DIR}/replication_resync_contract_backslide_hits.txt" REPLICATION_MRF_WIRE_FORMAT_BACKSLIDE_HITS_FILE="${TMP_DIR}/replication_mrf_wire_format_backslide_hits.txt" STORAGE_REPLICATION_HANDLE_BOUNDARY_BYPASS_HITS_FILE="${TMP_DIR}/storage_replication_handle_boundary_bypass_hits.txt" @@ -2559,6 +2560,21 @@ if [[ -s "$REPLICATION_CONFIG_RULE_CONTRACT_BACKSLIDE_HITS_FILE" ]]; then report_failure "replication config/rule/tag contracts must stay in crates/replication: $(paste -sd '; ' "$REPLICATION_CONFIG_RULE_CONTRACT_BACKSLIDE_HITS_FILE")" fi +( + cd "$ROOT_DIR" + replication_operation_status=0 + rg -n --with-filename '^\s*(?:pub(?:\([^)]*\))?\s+)?(?:struct\s+MustReplicateOptions|fn\s+is_ssec_encrypted)\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 + exit "$replication_operation_status" + fi +) + +if [[ -s "$REPLICATION_OPERATION_CONTRACT_BACKSLIDE_HITS_FILE" ]]; then + report_failure "replication operation contracts must stay in crates/replication: $(paste -sd '; ' "$REPLICATION_OPERATION_CONTRACT_BACKSLIDE_HITS_FILE")" +fi + ( cd "$ROOT_DIR" rg -n --with-filename 'pub\s+(struct|enum)\s+(ResyncOpts|TargetReplicationResyncStatus|BucketReplicationResyncStatus|ResyncStatusType)\b' \