Move replication operation contracts into crate (#4165)

refactor(replication): move operation contracts into crate

Cargo.lock

@@ -9897,6 +9897,7 @@ dependencies = [
  "rmp",
  "rmp-serde",
  "rustfs-filemeta",
+ "rustfs-utils",
  "s3s",
  "serde",
  "time",

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};

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;
 

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_LEGALHOLD_TIMESTAMP, SUFFIX_OBJECTLOCK_RETENTION_TIMESTAMP, SUFFIX_REPLICATION_RESET,
-    SUFFIX_OBJECTLOCK_RETENTION_TIMESTAMP, SUFFIX_REPLICATION_RESET, SUFFIX_REPLICATION_STATUS, SUFFIX_TAGGING_TIMESTAMP,
+    SUFFIX_REPLICATION_STATUS, SUFFIX_TAGGING_TIMESTAMP, headers,
-    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(
+            MustReplicateOptions::new(&user_defined, (*oi.user_tags).clone(), ReplicationType::Heal, false),
-                &user_defined,
-                (*oi.user_tags).clone(),
-                ReplicationStatusType::Empty,
-                ReplicationType::Heal,
-                ObjectOptions::default(),
-            ),
         )
         .await
     };
@@ -1089,13 +1082,7 @@ impl ReplicationConfig {
         let dsc = must_replicate(
             oi.bucket.as_str(),
             &oi.name,
-            MustReplicateOptions::new(
+            MustReplicateOptions::new(&user_defined, (*oi.user_tags).clone(), ReplicationType::ExistingObject, false),
-                &user_defined,
-                (*oi.user_tags).clone(),
-                ReplicationStatusType::Empty,
-                ReplicationType::ExistingObject,
-                ObjectOptions::default(),
-            ),
         )
         .await;
 
@@ -1192,62 +1179,14 @@ pub fn resync_target(
     dec
 }
 
-pub struct MustReplicateOptions {
-    meta: HashMap<String, String>,
-    status: ReplicationStatusType,
-    op_type: ReplicationType,
-    replication_request: bool,
-}
-
-impl MustReplicateOptions {
-    pub fn new(
-        meta: &HashMap<String, String>,
-        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<String, String>,
     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<String, String>) -> 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() {

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

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,

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<String, String>,
+    op_type: ReplicationType,
+    replication_request: bool,
+}
+
+impl MustReplicateOptions {
+    pub fn new(meta: &HashMap<String, String>, 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<String, String>) -> 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()));
+    }
+}

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' \