refactor(replication): add lock boundary (#4075)

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

@@ -16,6 +16,7 @@ mod config;
 pub mod datatypes;
 mod replication_config_store;
 mod replication_event_sink;
+mod replication_lock_boundary;
 mod replication_metadata_boundary;
 mod replication_pool;
 mod replication_resyncer;

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

@@ -0,0 +1,19 @@
+// 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::time::Duration;
+
+pub(crate) fn acquire_timeout() -> Duration {
+    crate::set_disk::get_lock_acquire_timeout()
+}

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

@@ -14,6 +14,7 @@
 
 use super::replication_config_store as config_store;
 use super::replication_event_sink::{EventArgs, send_event};
+use super::replication_lock_boundary as lock_boundary;
 use super::replication_metadata_boundary as metadata_boundary;
 use super::replication_target_boundary as target_boundary;
 use super::replication_versioning_boundary as versioning_boundary;
@@ -31,7 +32,6 @@ use crate::client::api_get_options::{AdvancedGetOptions, StatObjectOptions};
 use crate::disk::{BUCKET_META_PREFIX, RUSTFS_META_BUCKET};
 use crate::error::{Error, Result, is_err_object_not_found, is_err_version_not_found};
 use crate::object_api::{GetObjectReader, ObjectInfo, ObjectOptions, PutObjReader};
-use crate::set_disk::get_lock_acquire_timeout;
 use crate::storage_api_contracts::{
     list::{ListOperations, StorageListObjectVersionsInfo, StorageListObjectsV2Info, StorageObjectInfoOrErr, StorageWalkOptions},
     namespace::NamespaceLocking as StorageNamespaceLocking,
@@ -840,7 +840,7 @@ impl ReplicationResyncer {
                 return;
             }
         };
-        let _resync_leader_guard = match resync_ns_lock.get_write_lock(get_lock_acquire_timeout()).await {
+        let _resync_leader_guard = match resync_ns_lock.get_write_lock(lock_boundary::acquire_timeout()).await {
             Ok(g) => g,
             Err(_) => {
                 debug!(
@@ -2015,7 +2015,7 @@ pub async fn replicate_delete<S: ReplicationStorage>(dobj: DeletedObjectReplicat
         }
     };
 
-    let _lock_guard = match ns_lock.get_write_lock(get_lock_acquire_timeout()).await {
+    let _lock_guard = match ns_lock.get_write_lock(lock_boundary::acquire_timeout()).await {
         Ok(lock_guard) => lock_guard,
         Err(e) => {
             debug!(
@@ -2403,7 +2403,7 @@ async fn replicate_force_delete_to_targets<S: ReplicationStorage>(dobj: &Deleted
         }
     };
 
-    let _lock_guard = match ns_lock.get_write_lock(get_lock_acquire_timeout()).await {
+    let _lock_guard = match ns_lock.get_write_lock(lock_boundary::acquire_timeout()).await {
         Ok(guard) => guard,
         Err(e) => {
             warn!(
@@ -2802,7 +2802,7 @@ pub async fn replicate_object<S: ReplicationStorage>(roi: ReplicateObjectInfo, s
             return;
         }
     };
-    let _obj_lock_guard = match obj_ns_lock.get_write_lock(get_lock_acquire_timeout()).await {
+    let _obj_lock_guard = match obj_ns_lock.get_write_lock(lock_boundary::acquire_timeout()).await {
         Ok(g) => g,
         Err(e) => {
             debug!(

scripts/check_architecture_migration_rules.sh

@@ -185,6 +185,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_CONFIG_STORE_BYPASS_HITS_FILE="${TMP_DIR}/replication_config_store_bypass_hits.txt"
 REPLICATION_EVENT_SINK_BYPASS_HITS_FILE="${TMP_DIR}/replication_event_sink_bypass_hits.txt"
+REPLICATION_LOCK_BOUNDARY_BYPASS_HITS_FILE="${TMP_DIR}/replication_lock_boundary_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"
@@ -2367,6 +2368,18 @@ if [[ -s "$REPLICATION_CONFIG_STORE_BYPASS_HITS_FILE" ]]; then
   report_failure "replication config persistence must stay behind replication config store: $(paste -sd '; ' "$REPLICATION_CONFIG_STORE_BYPASS_HITS_FILE")"
 fi
 
+(
+  cd "$ROOT_DIR"
+  rg -n --with-filename 'crate::set_disk::get_lock_acquire_timeout|\bget_lock_acquire_timeout\(' \
+    crates/ecstore/src/bucket/replication \
+    --glob '*.rs' |
+    rg -v '^crates/ecstore/src/bucket/replication/replication_lock_boundary\.rs:' || true
+) >"$REPLICATION_LOCK_BOUNDARY_BYPASS_HITS_FILE"
+
+if [[ -s "$REPLICATION_LOCK_BOUNDARY_BYPASS_HITS_FILE" ]]; then
+  report_failure "replication lock timeout access must stay behind replication lock boundary: $(paste -sd '; ' "$REPLICATION_LOCK_BOUNDARY_BYPASS_HITS_FILE")"
+fi
+
 (
   cd "$ROOT_DIR"
   rg -n --with-filename 'BucketTargetSys::get\(\)' \