refactor(replication): move queue contracts into crate (#4167)

crates/ecstore/src/bucket/replication/mod.rs

@@ -42,8 +42,8 @@ pub(crate) use replication_lifecycle_bridge::{ReplicationLifecycleBridge, Replic
 pub(crate) use replication_migration_bridge::ReplicationMigrationBridge;
 pub use replication_object_bridge::ReplicationObjectBridge;
 pub use replication_pool::{
-    DynReplicationPool, ReplicationHealQueueResult, ReplicationPoolTrait, ReplicationQueueAdmission, get_global_replication_pool,
-    get_global_replication_stats, init_background_replication,
+    DynReplicationPool, ReplicationPoolTrait, get_global_replication_pool, get_global_replication_stats,
+    init_background_replication,
 };
 pub use replication_resyncer::ReplicationConfig;
 pub use replication_scanner_bridge::ReplicationScannerBridge;
@@ -51,5 +51,6 @@ 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, DeletedObjectReplicationInfo, MustReplicateOptions, ResyncOpts, TargetReplicationResyncStatus,
+    BucketReplicationResyncStatus, DeletedObjectReplicationInfo, MustReplicateOptions, ReplicationHealQueueResult,
+    ReplicationOperation, ReplicationPriority, ReplicationQueueAdmission, ResyncOpts, TargetReplicationResyncStatus,
 };

crates/ecstore/src/bucket/replication/replication_pool.rs

@@ -31,9 +31,11 @@ use super::replication_target_boundary::ReplicationTargetStore;
 use super::runtime_boundary as runtime_sources;
 use super::{BucketReplicationResyncStatus, ResyncOpts, TargetReplicationResyncStatus};
 use lazy_static::lazy_static;
-use rustfs_replication::DeletedObjectReplicationInfo;
+use rustfs_replication::{
+    DeletedObjectReplicationInfo, ReplicationHealQueueResult, ReplicationOperation, ReplicationPriority,
+    ReplicationQueueAdmission,
+};
 use rustfs_utils::http::{SUFFIX_REPLICATION_TIMESTAMP, get_str};
-use std::any::Any;
 use std::sync::Arc;
 use std::sync::atomic::AtomicI32;
 use std::sync::atomic::Ordering;
@@ -73,116 +75,6 @@ pub const MRF_WORKER_AUTO_DEFAULT: usize = 4;
 pub const LARGE_WORKER_COUNT: usize = 10;
 pub const MIN_LARGE_OBJ_SIZE: i64 = 128 * 1024 * 1024; // 128MiB
 
-#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
-pub enum ReplicationQueueAdmission {
-    #[default]
-    Skipped,
-    Queued,
-    Missed,
-}
-
-impl ReplicationQueueAdmission {
-    fn merge(&mut self, other: Self) {
-        *self = match (*self, other) {
-            (Self::Missed, _) | (_, Self::Missed) => Self::Missed,
-            (Self::Queued, _) | (_, Self::Queued) => Self::Queued,
-            (Self::Skipped, Self::Skipped) => Self::Skipped,
-        };
-    }
-}
-
-#[derive(Debug, Clone, Default)]
-pub struct ReplicationHealQueueResult {
-    pub object_info: ReplicateObjectInfo,
-    pub admission: ReplicationQueueAdmission,
-}
-
-/// Priority levels for replication
-#[derive(Debug, Clone, PartialEq)]
-pub enum ReplicationPriority {
-    Fast,
-    Slow,
-    Auto,
-}
-
-impl std::str::FromStr for ReplicationPriority {
-    type Err = ();
-
-    fn from_str(s: &str) -> Result<Self, Self::Err> {
-        match s {
-            "fast" => Ok(ReplicationPriority::Fast),
-            "slow" => Ok(ReplicationPriority::Slow),
-            "auto" => Ok(ReplicationPriority::Auto),
-            _ => Ok(ReplicationPriority::Auto), // Default to Auto for unknown values
-        }
-    }
-}
-
-impl ReplicationPriority {
-    pub fn as_str(&self) -> &'static str {
-        match self {
-            ReplicationPriority::Fast => "fast",
-            ReplicationPriority::Slow => "slow",
-            ReplicationPriority::Auto => "auto",
-        }
-    }
-}
-
-/// Enum for different types of replication operations
-#[derive(Debug)]
-pub enum ReplicationOperation {
-    Object(Box<ReplicateObjectInfo>),
-    Delete(Box<DeletedObjectReplicationInfo>),
-}
-
-impl ReplicationWorkerOperation for ReplicationOperation {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
-    fn to_mrf_entry(&self) -> MrfReplicateEntry {
-        match self {
-            ReplicationOperation::Object(obj) => obj.to_mrf_entry(),
-            ReplicationOperation::Delete(del) => del.to_mrf_entry(),
-        }
-    }
-
-    fn get_bucket(&self) -> &str {
-        match self {
-            ReplicationOperation::Object(obj) => obj.get_bucket(),
-            ReplicationOperation::Delete(del) => del.get_bucket(),
-        }
-    }
-
-    fn get_object(&self) -> &str {
-        match self {
-            ReplicationOperation::Object(obj) => obj.get_object(),
-            ReplicationOperation::Delete(del) => del.get_object(),
-        }
-    }
-
-    fn get_size(&self) -> i64 {
-        match self {
-            ReplicationOperation::Object(obj) => obj.get_size(),
-            ReplicationOperation::Delete(del) => del.get_size(),
-        }
-    }
-
-    fn is_delete_marker(&self) -> bool {
-        match self {
-            ReplicationOperation::Object(obj) => obj.is_delete_marker(),
-            ReplicationOperation::Delete(del) => del.is_delete_marker(),
-        }
-    }
-
-    fn get_op_type(&self) -> ReplicationType {
-        match self {
-            ReplicationOperation::Object(obj) => obj.get_op_type(),
-            ReplicationOperation::Delete(del) => del.get_op_type(),
-        }
-    }
-}
-
 /// Replication pool options
 #[derive(Debug, Clone)]
 pub struct ReplicationPoolOpts {

crates/ecstore/src/bucket/replication/replication_scanner_bridge.rs

@@ -12,7 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use super::replication_pool::{ReplicationHealQueueResult, queue_replication_heal_internal};
+use super::ReplicationHealQueueResult;
+use super::replication_pool::queue_replication_heal_internal;
 use super::replication_resyncer::ReplicationConfig;
 use super::replication_storage_boundary::ObjectInfo;
 
@@ -31,7 +32,7 @@ impl ReplicationScannerBridge {
 
 #[cfg(test)]
 mod tests {
-    use super::super::replication_pool::ReplicationQueueAdmission;
+    use super::super::ReplicationQueueAdmission;
     use super::*;
 
     #[tokio::test]

crates/replication/src/lib.rs

@@ -16,6 +16,7 @@ pub mod config;
 pub mod delete;
 pub mod mrf;
 pub mod operation;
+pub mod queue;
 pub mod resync;
 pub mod rule;
 pub mod tagging;
@@ -24,6 +25,7 @@ pub use config::{ObjectOpts, ReplicationConfigurationExt};
 pub use delete::DeletedObjectReplicationInfo;
 pub use mrf::{MrfOpKind, MrfReplicateEntry, decode_mrf_file, encode_mrf_file};
 pub use operation::{MustReplicateOptions, is_ssec_encrypted};
+pub use queue::{ReplicationHealQueueResult, ReplicationOperation, ReplicationPriority, ReplicationQueueAdmission};
 pub use resync::{
     BucketReplicationResyncStatus, Error, Result, ResyncOpts, ResyncStatusType, TargetReplicationResyncStatus,
     decode_resync_file, encode_resync_file,

crates/replication/src/queue.rs

@@ -0,0 +1,197 @@
+// 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::any::Any;
+
+use crate::{DeletedObjectReplicationInfo, MrfReplicateEntry, ReplicateObjectInfo, ReplicationType, ReplicationWorkerOperation};
+
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
+pub enum ReplicationQueueAdmission {
+    #[default]
+    Skipped,
+    Queued,
+    Missed,
+}
+
+impl ReplicationQueueAdmission {
+    pub fn merge(&mut self, other: Self) {
+        *self = match (*self, other) {
+            (Self::Missed, _) | (_, Self::Missed) => Self::Missed,
+            (Self::Queued, _) | (_, Self::Queued) => Self::Queued,
+            (Self::Skipped, Self::Skipped) => Self::Skipped,
+        };
+    }
+}
+
+#[derive(Debug, Clone, Default)]
+pub struct ReplicationHealQueueResult {
+    pub object_info: ReplicateObjectInfo,
+    pub admission: ReplicationQueueAdmission,
+}
+
+#[derive(Debug, Clone, PartialEq)]
+pub enum ReplicationPriority {
+    Fast,
+    Slow,
+    Auto,
+}
+
+impl std::str::FromStr for ReplicationPriority {
+    type Err = ();
+
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        match s {
+            "fast" => Ok(ReplicationPriority::Fast),
+            "slow" => Ok(ReplicationPriority::Slow),
+            "auto" => Ok(ReplicationPriority::Auto),
+            _ => Ok(ReplicationPriority::Auto),
+        }
+    }
+}
+
+impl ReplicationPriority {
+    pub fn as_str(&self) -> &'static str {
+        match self {
+            ReplicationPriority::Fast => "fast",
+            ReplicationPriority::Slow => "slow",
+            ReplicationPriority::Auto => "auto",
+        }
+    }
+}
+
+#[derive(Debug)]
+pub enum ReplicationOperation {
+    Object(Box<ReplicateObjectInfo>),
+    Delete(Box<DeletedObjectReplicationInfo>),
+}
+
+impl ReplicationWorkerOperation for ReplicationOperation {
+    fn as_any(&self) -> &dyn Any {
+        self
+    }
+
+    fn to_mrf_entry(&self) -> MrfReplicateEntry {
+        match self {
+            ReplicationOperation::Object(obj) => obj.to_mrf_entry(),
+            ReplicationOperation::Delete(del) => del.to_mrf_entry(),
+        }
+    }
+
+    fn get_bucket(&self) -> &str {
+        match self {
+            ReplicationOperation::Object(obj) => obj.get_bucket(),
+            ReplicationOperation::Delete(del) => del.get_bucket(),
+        }
+    }
+
+    fn get_object(&self) -> &str {
+        match self {
+            ReplicationOperation::Object(obj) => obj.get_object(),
+            ReplicationOperation::Delete(del) => del.get_object(),
+        }
+    }
+
+    fn get_size(&self) -> i64 {
+        match self {
+            ReplicationOperation::Object(obj) => obj.get_size(),
+            ReplicationOperation::Delete(del) => del.get_size(),
+        }
+    }
+
+    fn is_delete_marker(&self) -> bool {
+        match self {
+            ReplicationOperation::Object(obj) => obj.is_delete_marker(),
+            ReplicationOperation::Delete(del) => del.is_delete_marker(),
+        }
+    }
+
+    fn get_op_type(&self) -> ReplicationType {
+        match self {
+            ReplicationOperation::Object(obj) => obj.get_op_type(),
+            ReplicationOperation::Delete(del) => del.get_op_type(),
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::str::FromStr;
+
+    use rustfs_storage_api::DeletedObject;
+
+    use super::{ReplicationOperation, ReplicationPriority, ReplicationQueueAdmission};
+    use crate::{DeletedObjectReplicationInfo, ReplicateObjectInfo, ReplicationType, ReplicationWorkerOperation};
+
+    #[test]
+    fn replication_queue_admission_combines_target_results() {
+        let mut admission = ReplicationQueueAdmission::Skipped;
+
+        admission.merge(ReplicationQueueAdmission::Queued);
+        assert_eq!(admission, ReplicationQueueAdmission::Queued);
+
+        admission.merge(ReplicationQueueAdmission::Missed);
+        assert_eq!(admission, ReplicationQueueAdmission::Missed);
+
+        admission.merge(ReplicationQueueAdmission::Skipped);
+        assert_eq!(admission, ReplicationQueueAdmission::Missed);
+    }
+
+    #[test]
+    fn replication_priority_parses_known_values_and_defaults_unknown() {
+        assert_eq!(ReplicationPriority::from_str("fast"), Ok(ReplicationPriority::Fast));
+        assert_eq!(ReplicationPriority::from_str("slow"), Ok(ReplicationPriority::Slow));
+        assert_eq!(ReplicationPriority::from_str("auto"), Ok(ReplicationPriority::Auto));
+        assert_eq!(ReplicationPriority::from_str("unexpected"), Ok(ReplicationPriority::Auto));
+        assert_eq!(ReplicationPriority::Fast.as_str(), "fast");
+        assert_eq!(ReplicationPriority::Slow.as_str(), "slow");
+        assert_eq!(ReplicationPriority::Auto.as_str(), "auto");
+    }
+
+    #[test]
+    fn replication_operation_delegates_delete_contract() {
+        let info = DeletedObjectReplicationInfo {
+            bucket: "bucket".to_string(),
+            op_type: ReplicationType::Delete,
+            delete_object: DeletedObject {
+                object_name: "object".to_string(),
+                delete_marker: true,
+                ..Default::default()
+            },
+            ..Default::default()
+        };
+        let operation = ReplicationOperation::Delete(Box::new(info));
+
+        assert_eq!(operation.get_bucket(), "bucket");
+        assert_eq!(operation.get_object(), "object");
+        assert_eq!(operation.get_op_type(), ReplicationType::Delete);
+        assert!(operation.is_delete_marker());
+    }
+
+    #[test]
+    fn replication_operation_delegates_object_contract() {
+        let info = ReplicateObjectInfo {
+            bucket: "bucket".to_string(),
+            name: "object".to_string(),
+            size: 128,
+            op_type: ReplicationType::Object,
+            ..Default::default()
+        };
+        let operation = ReplicationOperation::Object(Box::new(info));
+
+        assert_eq!(operation.get_bucket(), "bucket");
+        assert_eq!(operation.get_object(), "object");
+        assert_eq!(operation.get_size(), 128);
+        assert_eq!(operation.get_op_type(), ReplicationType::Object);
+    }
+}

scripts/check_architecture_migration_rules.sh

@@ -202,6 +202,7 @@ REPLICATION_FACADE_WILDCARD_EXPORT_HITS_FILE="${TMP_DIR}/replication_facade_wild
 REPLICATION_CONFIG_RULE_CONTRACT_BACKSLIDE_HITS_FILE="${TMP_DIR}/replication_config_rule_contract_backslide_hits.txt"
 REPLICATION_DELETE_WORKER_CONTRACT_BACKSLIDE_HITS_FILE="${TMP_DIR}/replication_delete_worker_contract_backslide_hits.txt"
 REPLICATION_OPERATION_CONTRACT_BACKSLIDE_HITS_FILE="${TMP_DIR}/replication_operation_contract_backslide_hits.txt"
+REPLICATION_QUEUE_CONTRACT_BACKSLIDE_HITS_FILE="${TMP_DIR}/replication_queue_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"
@@ -2591,6 +2592,21 @@ if [[ -s "$REPLICATION_DELETE_WORKER_CONTRACT_BACKSLIDE_HITS_FILE" ]]; then
   report_failure "replication delete worker contracts must stay in crates/replication: $(paste -sd '; ' "$REPLICATION_DELETE_WORKER_CONTRACT_BACKSLIDE_HITS_FILE")"
 fi
 
+(
+  cd "$ROOT_DIR"
+  replication_queue_status=0
+  rg -n --with-filename '^\s*(?:pub(?:\([^)]*\))?\s+)?(?:enum\s+ReplicationQueueAdmission|struct\s+ReplicationHealQueueResult|enum\s+ReplicationPriority|enum\s+ReplicationOperation)\b' \
+    crates/ecstore/src/bucket/replication \
+    --glob '*.rs' >"$REPLICATION_QUEUE_CONTRACT_BACKSLIDE_HITS_FILE" || replication_queue_status=$?
+  if [[ "$replication_queue_status" -ne 0 && "$replication_queue_status" -ne 1 ]]; then
+    exit "$replication_queue_status"
+  fi
+)
+
+if [[ -s "$REPLICATION_QUEUE_CONTRACT_BACKSLIDE_HITS_FILE" ]]; then
+  report_failure "replication queue contracts must stay in crates/replication: $(paste -sd '; ' "$REPLICATION_QUEUE_CONTRACT_BACKSLIDE_HITS_FILE")"
+fi
+
 (
   cd "$ROOT_DIR"
   rg -n --with-filename 'pub\s+(struct|enum)\s+(ResyncOpts|TargetReplicationResyncStatus|BucketReplicationResyncStatus|ResyncStatusType)\b' \