From 2cf856a51d14d9a7f1e821bcec2d014a678fbdab Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Thu, 2 Jul 2026 11:03:24 +0800 Subject: [PATCH] Move replication delete worker contract into crate (#4166) refactor(replication): move delete worker contract into crate --- Cargo.lock | 1 + crates/ecstore/src/bucket/replication/mod.rs | 6 +- .../replication_lifecycle_bridge.rs | 3 +- .../replication/replication_object_bridge.rs | 6 +- .../bucket/replication/replication_pool.rs | 5 +- .../replication/replication_resyncer.rs | 62 +---------- crates/replication/Cargo.toml | 1 + crates/replication/src/delete.rs | 104 ++++++++++++++++++ crates/replication/src/lib.rs | 2 + scripts/check_architecture_migration_rules.sh | 16 +++ 10 files changed, 141 insertions(+), 65 deletions(-) create mode 100644 crates/replication/src/delete.rs diff --git a/Cargo.lock b/Cargo.lock index b5df4b90f..0d930dc7f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -9897,6 +9897,7 @@ dependencies = [ "rmp", "rmp-serde", "rustfs-filemeta", + "rustfs-storage-api", "rustfs-utils", "s3s", "serde", diff --git a/crates/ecstore/src/bucket/replication/mod.rs b/crates/ecstore/src/bucket/replication/mod.rs index 5b54f4d56..eb0088f8a 100644 --- a/crates/ecstore/src/bucket/replication/mod.rs +++ b/crates/ecstore/src/bucket/replication/mod.rs @@ -45,9 +45,11 @@ pub use replication_pool::{ DynReplicationPool, ReplicationHealQueueResult, ReplicationPoolTrait, ReplicationQueueAdmission, get_global_replication_pool, get_global_replication_stats, init_background_replication, }; -pub use replication_resyncer::{DeletedObjectReplicationInfo, ReplicationConfig}; +pub use replication_resyncer::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, MustReplicateOptions, ResyncOpts, TargetReplicationResyncStatus}; +pub use rustfs_replication::{ + BucketReplicationResyncStatus, DeletedObjectReplicationInfo, MustReplicateOptions, ResyncOpts, TargetReplicationResyncStatus, +}; diff --git a/crates/ecstore/src/bucket/replication/replication_lifecycle_bridge.rs b/crates/ecstore/src/bucket/replication/replication_lifecycle_bridge.rs index ebf1310af..98c6d1ff7 100644 --- a/crates/ecstore/src/bucket/replication/replication_lifecycle_bridge.rs +++ b/crates/ecstore/src/bucket/replication/replication_lifecycle_bridge.rs @@ -18,8 +18,9 @@ use super::config::ReplicationConfigurationExt as _; use super::replication_filemeta_boundary::{ REPLICATE_INCOMING_DELETE, ReplicateDecision, ReplicationState, version_purge_statuses_map, }; -use super::replication_resyncer::{DeletedObjectReplicationInfo, ReplicationConfig, check_replicate_delete}; +use super::replication_resyncer::{ReplicationConfig, check_replicate_delete}; use super::replication_storage_boundary::{DeletedObject, ObjectInfo, ObjectOptions, ObjectToDelete}; +use rustfs_replication::DeletedObjectReplicationInfo; pub(crate) type ReplicationLifecycleConfig = ReplicationConfig; diff --git a/crates/ecstore/src/bucket/replication/replication_object_bridge.rs b/crates/ecstore/src/bucket/replication/replication_object_bridge.rs index 60a946f35..8d2e6c81c 100644 --- a/crates/ecstore/src/bucket/replication/replication_object_bridge.rs +++ b/crates/ecstore/src/bucket/replication/replication_object_bridge.rs @@ -16,11 +16,9 @@ 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, check_replicate_delete, get_must_replicate_options, must_replicate, -}; +use super::replication_resyncer::{check_replicate_delete, get_must_replicate_options, must_replicate}; use super::replication_storage_boundary::{ObjectInfo, ObjectOptions, ObjectToDelete, ReplicationStorage}; -use rustfs_replication::MustReplicateOptions; +use rustfs_replication::{DeletedObjectReplicationInfo, MustReplicateOptions}; pub struct ReplicationObjectBridge; diff --git a/crates/ecstore/src/bucket/replication/replication_pool.rs b/crates/ecstore/src/bucket/replication/replication_pool.rs index fa56cd2ad..aa4ff37c5 100644 --- a/crates/ecstore/src/bucket/replication/replication_pool.rs +++ b/crates/ecstore/src/bucket/replication/replication_pool.rs @@ -22,8 +22,8 @@ use super::replication_filemeta_boundary::{ }; use super::replication_metadata_boundary::ReplicationMetadataStore; use super::replication_resyncer::{ - DeletedObjectReplicationInfo, ReplicationConfig, ReplicationResyncer, decode_mrf_file, decode_resync_file, encode_mrf_file, - get_heal_replicate_object_info, replicate_delete, replicate_object, save_resync_status, + ReplicationConfig, ReplicationResyncer, decode_mrf_file, decode_resync_file, encode_mrf_file, get_heal_replicate_object_info, + replicate_delete, replicate_object, save_resync_status, }; use super::replication_state::ReplicationStats; use super::replication_storage_boundary::{DeletedObject, ObjectInfo, ObjectOptions, ReplicationObjectIO, ReplicationStorage}; @@ -31,6 +31,7 @@ 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_utils::http::{SUFFIX_REPLICATION_TIMESTAMP, get_str}; use std::any::Any; use std::sync::Arc; diff --git a/crates/ecstore/src/bucket/replication/replication_resyncer.rs b/crates/ecstore/src/bucket/replication/replication_resyncer.rs index c9ab0fc08..52e402637 100644 --- a/crates/ecstore/src/bucket/replication/replication_resyncer.rs +++ b/crates/ecstore/src/bucket/replication/replication_resyncer.rs @@ -19,10 +19,10 @@ use super::replication_config_store::ReplicationConfigStore; use super::replication_error_boundary::{Error, Result, is_err_object_not_found, is_err_version_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, + MrfReplicateEntry, REPLICATE_EXISTING, REPLICATE_EXISTING_DELETE, ReplicateDecision, ReplicateObjectInfo, 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, + 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; @@ -55,7 +55,9 @@ use http_body::Frame; use http_body_util::StreamBody; #[cfg(test)] use rmp_serde; -use rustfs_replication::{BucketReplicationResyncStatus, MustReplicateOptions, ResyncOpts, TargetReplicationResyncStatus}; +use rustfs_replication::{ + BucketReplicationResyncStatus, DeletedObjectReplicationInfo, 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 _, @@ -71,7 +73,6 @@ use rustfs_utils::{DEFAULT_SIP_HASH_KEY, sip_hash}; use s3s::dto::ReplicationConfiguration; use serde::Deserialize; use serde::Serialize; -use std::any::Any; use std::collections::HashMap; use std::sync::Arc; use time::OffsetDateTime; @@ -965,57 +966,6 @@ async fn get_replication_config(bucket: &str) -> Result &dyn Any { - self - } - - fn to_mrf_entry(&self) -> MrfReplicateEntry { - MrfReplicateEntry { - bucket: self.bucket.clone(), - object: self.delete_object.object_name.clone(), - // version_id here is the version being purged (if any); the delete-marker - // version is stored separately in delete_marker_version_id. - version_id: self.delete_object.version_id, - retry_count: 0, - size: 0, - op: MrfOpKind::Delete, - delete_marker_version_id: self.delete_object.delete_marker_version_id, - delete_marker: self.delete_object.delete_marker, - } - } - - fn get_bucket(&self) -> &str { - &self.bucket - } - - fn get_object(&self) -> &str { - &self.delete_object.object_name - } - - fn get_size(&self) -> i64 { - 0 - } - - fn is_delete_marker(&self) -> bool { - true - } - - fn get_op_type(&self) -> ReplicationType { - self.op_type - } -} - #[derive(Debug, Clone, Default, Serialize, Deserialize)] pub struct ReplicationConfig { pub config: Option, diff --git a/crates/replication/Cargo.toml b/crates/replication/Cargo.toml index 96b5f402a..f45409eb6 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-storage-api.workspace = true rustfs-utils = { workspace = true, features = ["http"] } s3s.workspace = true serde.workspace = true diff --git a/crates/replication/src/delete.rs b/crates/replication/src/delete.rs new file mode 100644 index 000000000..8cd6f7670 --- /dev/null +++ b/crates/replication/src/delete.rs @@ -0,0 +1,104 @@ +// 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 rustfs_storage_api::DeletedObject; + +use crate::{MrfOpKind, MrfReplicateEntry, ReplicationType, ReplicationWorkerOperation}; + +#[derive(Debug, Clone, Default)] +pub struct DeletedObjectReplicationInfo { + pub delete_object: DeletedObject, + pub bucket: String, + pub event_type: String, + pub op_type: ReplicationType, + pub reset_id: String, + pub target_arn: String, +} + +impl ReplicationWorkerOperation for DeletedObjectReplicationInfo { + fn as_any(&self) -> &dyn Any { + self + } + + fn to_mrf_entry(&self) -> MrfReplicateEntry { + MrfReplicateEntry { + bucket: self.bucket.clone(), + object: self.delete_object.object_name.clone(), + version_id: self.delete_object.version_id, + retry_count: 0, + size: 0, + op: MrfOpKind::Delete, + delete_marker_version_id: self.delete_object.delete_marker_version_id, + delete_marker: self.delete_object.delete_marker, + } + } + + fn get_bucket(&self) -> &str { + &self.bucket + } + + fn get_object(&self) -> &str { + &self.delete_object.object_name + } + + fn get_size(&self) -> i64 { + 0 + } + + fn is_delete_marker(&self) -> bool { + true + } + + fn get_op_type(&self) -> ReplicationType { + self.op_type + } +} + +#[cfg(test)] +mod tests { + use super::DeletedObjectReplicationInfo; + use crate::{MrfOpKind, ReplicationType, ReplicationWorkerOperation}; + use rustfs_storage_api::DeletedObject; + use uuid::Uuid; + + #[test] + fn deleted_object_replication_info_encodes_delete_mrf_entry() { + let version_id = Uuid::new_v4(); + let delete_marker_version_id = Uuid::new_v4(); + let info = DeletedObjectReplicationInfo { + bucket: "bucket".to_string(), + op_type: ReplicationType::Delete, + delete_object: DeletedObject { + object_name: "object".to_string(), + version_id: Some(version_id), + delete_marker_version_id: Some(delete_marker_version_id), + delete_marker: true, + ..Default::default() + }, + ..Default::default() + }; + + let entry = info.to_mrf_entry(); + + assert_eq!(entry.bucket, "bucket"); + assert_eq!(entry.object, "object"); + assert_eq!(entry.version_id, Some(version_id)); + assert_eq!(entry.delete_marker_version_id, Some(delete_marker_version_id)); + assert_eq!(entry.op, MrfOpKind::Delete); + assert!(entry.delete_marker); + assert_eq!(info.get_object(), "object"); + } +} diff --git a/crates/replication/src/lib.rs b/crates/replication/src/lib.rs index ba7e8a4ca..0a81dd784 100644 --- a/crates/replication/src/lib.rs +++ b/crates/replication/src/lib.rs @@ -13,6 +13,7 @@ // limitations under the License. pub mod config; +pub mod delete; pub mod mrf; pub mod operation; pub mod resync; @@ -20,6 +21,7 @@ pub mod rule; pub mod tagging; 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 resync::{ diff --git a/scripts/check_architecture_migration_rules.sh b/scripts/check_architecture_migration_rules.sh index 2e30a944c..dd56ce47e 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_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_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" @@ -2575,6 +2576,21 @@ 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" + replication_delete_worker_status=0 + rg -n --with-filename '^\s*(?:pub(?:\([^)]*\))?\s+)?struct\s+DeletedObjectReplicationInfo\b' \ + crates/ecstore/src/bucket/replication \ + --glob '*.rs' >"$REPLICATION_DELETE_WORKER_CONTRACT_BACKSLIDE_HITS_FILE" || replication_delete_worker_status=$? + if [[ "$replication_delete_worker_status" -ne 0 && "$replication_delete_worker_status" -ne 1 ]]; then + exit "$replication_delete_worker_status" + fi +) + +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" rg -n --with-filename 'pub\s+(struct|enum)\s+(ResyncOpts|TargetReplicationResyncStatus|BucketReplicationResyncStatus|ResyncStatusType)\b' \