From 8ce0f824cfd2501d1ccc71a2b179ef431060d1b9 Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Thu, 2 Jul 2026 11:40:49 +0800 Subject: [PATCH] refactor(replication): move queue contracts into crate (#4167) --- crates/ecstore/src/bucket/replication/mod.rs | 7 +- .../bucket/replication/replication_pool.rs | 116 +---------- .../replication/replication_scanner_bridge.rs | 5 +- crates/replication/src/lib.rs | 2 + crates/replication/src/queue.rs | 197 ++++++++++++++++++ scripts/check_architecture_migration_rules.sh | 16 ++ 6 files changed, 226 insertions(+), 117 deletions(-) create mode 100644 crates/replication/src/queue.rs diff --git a/crates/ecstore/src/bucket/replication/mod.rs b/crates/ecstore/src/bucket/replication/mod.rs index eb0088f8a..711a11b1a 100644 --- a/crates/ecstore/src/bucket/replication/mod.rs +++ b/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, }; diff --git a/crates/ecstore/src/bucket/replication/replication_pool.rs b/crates/ecstore/src/bucket/replication/replication_pool.rs index aa4ff37c5..a1d1e6fba 100644 --- a/crates/ecstore/src/bucket/replication/replication_pool.rs +++ b/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 { - 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), - Delete(Box), -} - -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 { diff --git a/crates/ecstore/src/bucket/replication/replication_scanner_bridge.rs b/crates/ecstore/src/bucket/replication/replication_scanner_bridge.rs index 9b6d0989e..714929b8b 100644 --- a/crates/ecstore/src/bucket/replication/replication_scanner_bridge.rs +++ b/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] diff --git a/crates/replication/src/lib.rs b/crates/replication/src/lib.rs index 0a81dd784..6883547c4 100644 --- a/crates/replication/src/lib.rs +++ b/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, diff --git a/crates/replication/src/queue.rs b/crates/replication/src/queue.rs new file mode 100644 index 000000000..20567b51a --- /dev/null +++ b/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 { + 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), + Delete(Box), +} + +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); + } +} diff --git a/scripts/check_architecture_migration_rules.sh b/scripts/check_architecture_migration_rules.sh index dd56ce47e..c8dae6cd4 100755 --- a/scripts/check_architecture_migration_rules.sh +++ b/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' \