From f8c70e017df71fda37a5519d71a791941af3ee9a Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Mon, 29 Jun 2026 23:52:36 +0800 Subject: [PATCH] refactor(replication): add versioning boundary (#4071) --- crates/ecstore/src/bucket/replication/mod.rs | 1 + .../replication/replication_resyncer.rs | 28 +++++++++---------- .../replication_versioning_boundary.rs | 23 +++++++++++++++ scripts/check_architecture_migration_rules.sh | 13 +++++++++ 4 files changed, 51 insertions(+), 14 deletions(-) create mode 100644 crates/ecstore/src/bucket/replication/replication_versioning_boundary.rs diff --git a/crates/ecstore/src/bucket/replication/mod.rs b/crates/ecstore/src/bucket/replication/mod.rs index 3fd583bb5..3873570e9 100644 --- a/crates/ecstore/src/bucket/replication/mod.rs +++ b/crates/ecstore/src/bucket/replication/mod.rs @@ -20,6 +20,7 @@ mod replication_pool; mod replication_resyncer; mod replication_state; mod replication_target_boundary; +mod replication_versioning_boundary; mod rule; mod runtime_boundary; diff --git a/crates/ecstore/src/bucket/replication/replication_resyncer.rs b/crates/ecstore/src/bucket/replication/replication_resyncer.rs index 9166cf6ca..ea268adf7 100644 --- a/crates/ecstore/src/bucket/replication/replication_resyncer.rs +++ b/crates/ecstore/src/bucket/replication/replication_resyncer.rs @@ -15,6 +15,7 @@ use super::replication_event_sink::{EventArgs, send_event}; use super::replication_metadata_boundary as metadata_boundary; use super::replication_target_boundary as target_boundary; +use super::replication_versioning_boundary as versioning_boundary; use super::runtime_boundary as runtime_sources; use crate::bucket::bandwidth::reader::{BucketOptions, MonitorReaderOptions, MonitoredReader}; use crate::bucket::bucket_target_sys::{ @@ -25,7 +26,6 @@ use crate::bucket::replication::ResyncStatusType; use crate::bucket::replication::{ObjectOpts, ReplicationConfigurationExt as _}; use crate::bucket::tagging::decode_tags_to_map; use crate::bucket::target::BucketTargets; -use crate::bucket::versioning_sys::BucketVersioningSys; use crate::client::api_get_options::{AdvancedGetOptions, StatObjectOptions}; use crate::config::com::save_config; use crate::disk::{BUCKET_META_PREFIX, RUSTFS_META_BUCKET}; @@ -1247,8 +1247,8 @@ pub async fn get_heal_replicate_object_info(oi: &ObjectInfo, rcfg: &ReplicationC }, &oi, &ObjectOptions { - versioned: BucketVersioningSys::prefix_enabled(&oi.bucket, &oi.name).await, - version_suspended: BucketVersioningSys::prefix_suspended(&oi.bucket, &oi.name).await, + versioned: versioning_boundary::prefix_enabled(&oi.bucket, &oi.name).await, + version_suspended: versioning_boundary::prefix_suspended(&oi.bucket, &oi.name).await, ..Default::default() }, None, @@ -1756,7 +1756,7 @@ pub async fn must_replicate(bucket: &str, object: &str, mopts: MustReplicateOpti return ReplicateDecision::default(); } - if !BucketVersioningSys::prefix_enabled(bucket, object).await { + if !versioning_boundary::prefix_enabled(bucket, object).await { return ReplicateDecision::default(); } @@ -1892,8 +1892,8 @@ pub async fn replicate_delete(dobj: DeletedObjectReplicat &dobj.delete_object.object_name, &ObjectOptions { version_id: Some(delete_marker_version_id.to_string()), - versioned: BucketVersioningSys::prefix_enabled(&bucket, &dobj.delete_object.object_name).await, - version_suspended: BucketVersioningSys::prefix_suspended(&bucket, &dobj.delete_object.object_name).await, + versioned: versioning_boundary::prefix_enabled(&bucket, &dobj.delete_object.object_name).await, + version_suspended: versioning_boundary::prefix_suspended(&bucket, &dobj.delete_object.object_name).await, ..Default::default() }, ) @@ -2214,8 +2214,8 @@ pub async fn replicate_delete(dobj: DeletedObjectReplicat version_id: version_id.map(|v| v.to_string()), mod_time: dobj.delete_object.delete_marker_mtime, delete_replication: Some(drs), - versioned: BucketVersioningSys::prefix_enabled(&bucket, &dobj.delete_object.object_name).await, - version_suspended: BucketVersioningSys::prefix_suspended(&bucket, &dobj.delete_object.object_name).await, + versioned: versioning_boundary::prefix_enabled(&bucket, &dobj.delete_object.object_name).await, + version_suspended: versioning_boundary::prefix_suspended(&bucket, &dobj.delete_object.object_name).await, ..Default::default() }, ) @@ -2269,8 +2269,8 @@ async fn source_delete_marker_missing( object_name, &ObjectOptions { version_id: Some(delete_marker_version_id.to_string()), - versioned: BucketVersioningSys::prefix_enabled(bucket, object_name).await, - version_suspended: BucketVersioningSys::prefix_suspended(bucket, object_name).await, + versioned: versioning_boundary::prefix_enabled(bucket, object_name).await, + version_suspended: versioning_boundary::prefix_suspended(bucket, object_name).await, ..Default::default() }, ) @@ -3010,8 +3010,8 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { return rinfo; } - let versioned = BucketVersioningSys::prefix_enabled(&bucket, &object).await; - let version_suspended = BucketVersioningSys::prefix_suspended(&bucket, &object).await; + let versioned = versioning_boundary::prefix_enabled(&bucket, &object).await; + let version_suspended = versioning_boundary::prefix_suspended(&bucket, &object).await; let obj_opts = ObjectOptions { version_id: self.version_id.map(|v| v.to_string()), @@ -3297,8 +3297,8 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { return rinfo; } - let versioned = BucketVersioningSys::prefix_enabled(&bucket, &object).await; - let version_suspended = BucketVersioningSys::prefix_suspended(&bucket, &object).await; + let versioned = versioning_boundary::prefix_enabled(&bucket, &object).await; + let version_suspended = versioning_boundary::prefix_suspended(&bucket, &object).await; let obj_opts = ObjectOptions { version_id: self.version_id.map(|v| v.to_string()), diff --git a/crates/ecstore/src/bucket/replication/replication_versioning_boundary.rs b/crates/ecstore/src/bucket/replication/replication_versioning_boundary.rs new file mode 100644 index 000000000..75835d441 --- /dev/null +++ b/crates/ecstore/src/bucket/replication/replication_versioning_boundary.rs @@ -0,0 +1,23 @@ +// 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 crate::bucket::versioning_sys::BucketVersioningSys; + +pub(crate) async fn prefix_enabled(bucket: &str, prefix: &str) -> bool { + BucketVersioningSys::prefix_enabled(bucket, prefix).await +} + +pub(crate) async fn prefix_suspended(bucket: &str, prefix: &str) -> bool { + BucketVersioningSys::prefix_suspended(bucket, prefix).await +} diff --git a/scripts/check_architecture_migration_rules.sh b/scripts/check_architecture_migration_rules.sh index 9bf1df923..2696acccb 100755 --- a/scripts/check_architecture_migration_rules.sh +++ b/scripts/check_architecture_migration_rules.sh @@ -186,6 +186,7 @@ REPLICATION_FACADE_BYPASS_HITS_FILE="${TMP_DIR}/replication_facade_bypass_hits.t REPLICATION_EVENT_SINK_BYPASS_HITS_FILE="${TMP_DIR}/replication_event_sink_bypass_hits.txt" REPLICATION_METADATA_BOUNDARY_BYPASS_HITS_FILE="${TMP_DIR}/replication_metadata_boundary_bypass_hits.txt" REPLICATION_TARGET_BOUNDARY_BYPASS_HITS_FILE="${TMP_DIR}/replication_target_boundary_bypass_hits.txt" +REPLICATION_VERSIONING_BOUNDARY_BYPASS_HITS_FILE="${TMP_DIR}/replication_versioning_boundary_bypass_hits.txt" REPLICATION_RUNTIME_SOURCE_BYPASS_HITS_FILE="${TMP_DIR}/replication_runtime_source_bypass_hits.txt" GLOBAL_REPLICATION_STATE_BYPASS_HITS_FILE="${TMP_DIR}/global_replication_state_bypass_hits.txt" GLOBAL_BUCKET_MONITOR_BYPASS_HITS_FILE="${TMP_DIR}/global_bucket_monitor_bypass_hits.txt" @@ -2377,6 +2378,18 @@ if [[ -s "$REPLICATION_METADATA_BOUNDARY_BYPASS_HITS_FILE" ]]; then report_failure "replication metadata access must stay behind replication metadata boundary: $(paste -sd '; ' "$REPLICATION_METADATA_BOUNDARY_BYPASS_HITS_FILE")" fi +( + cd "$ROOT_DIR" + rg -n --with-filename 'crate::bucket::versioning_sys|BucketVersioningSys::prefix_(enabled|suspended)' \ + crates/ecstore/src/bucket/replication \ + --glob '*.rs' | + rg -v '^crates/ecstore/src/bucket/replication/replication_versioning_boundary\.rs:' || true +) >"$REPLICATION_VERSIONING_BOUNDARY_BYPASS_HITS_FILE" + +if [[ -s "$REPLICATION_VERSIONING_BOUNDARY_BYPASS_HITS_FILE" ]]; then + report_failure "replication versioning access must stay behind replication versioning boundary: $(paste -sd '; ' "$REPLICATION_VERSIONING_BOUNDARY_BYPASS_HITS_FILE")" +fi + ( cd "$ROOT_DIR" rg -n --with-filename '\bGLOBAL_REPLICATION_(POOL|STATS)\b' \