refactor(ecstore): add replication owner bridges (#4141)

* refactor(ecstore): add replication owner bridges

* fix(ecstore): tighten replication owner guards

crates/ecstore/src/bucket/bucket_target_sys.rs

@@ -15,8 +15,7 @@
 use crate::bucket::metadata::BucketMetadata;
 use crate::bucket::metadata_sys::get_bucket_targets_config;
 use crate::bucket::metadata_sys::get_replication_config;
-use crate::bucket::replication::ObjectOpts;
-use crate::bucket::replication::ReplicationConfigurationExt;
+use crate::bucket::replication::ReplicationTargetConfigBridge;
 use crate::bucket::target::ARN;
 use crate::bucket::target::BucketTargetType;
 use crate::bucket::target::{self, BucketTarget, BucketTargets, Credentials};
@@ -50,7 +49,7 @@ use hyper_util::client::legacy::Client as HyperClient;
 use hyper_util::rt::{TokioExecutor, TokioTimer};
 use reqwest::Client as HttpClient;
 use rustfs_config::{DEFAULT_TRUST_LEAF_CERT_AS_CA, ENV_TRUST_LEAF_CERT_AS_CA, RUSTFS_CA_CERT, RUSTFS_TLS_CERT};
-use rustfs_filemeta::{ReplicationStatusType, ReplicationType};
+use rustfs_filemeta::ReplicationStatusType;
 use rustfs_utils::egress::{OutboundUrlError, validate_outbound_url};
 use rustfs_utils::http::{
     AMZ_BUCKET_REPLICATION_STATUS, AMZ_OBJECT_LOCK_BYPASS_GOVERNANCE, AMZ_OBJECT_LOCK_LEGAL_HOLD, AMZ_OBJECT_LOCK_MODE,
@@ -552,19 +551,13 @@ impl BucketTargetSys {
 
         if arn.arn_type == BucketTargetType::ReplicationService
             && let Ok((config, _)) = get_replication_config(bucket).await
+            && ReplicationTargetConfigBridge::target_is_used_by_rules(&config, arn_str)
         {
-            for rule in config.filter_target_arns(&ObjectOpts {
-                op_type: ReplicationType::All,
-                ..Default::default()
-            }) {
-                if rule == arn_str || config.role == arn_str {
-                    let arn_remotes_map = self.arn_remotes_map.read().await;
-                    if arn_remotes_map.get(arn_str).is_some() {
-                        return Err(BucketTargetError::BucketRemoteRemoveDisallowed {
-                            bucket: bucket.to_string(),
-                        });
-                    }
-                }
+            let arn_remotes_map = self.arn_remotes_map.read().await;
+            if arn_remotes_map.get(arn_str).is_some() {
+                return Err(BucketTargetError::BucketRemoteRemoveDisallowed {
+                    bucket: bucket.to_string(),
+                });
             }
         }
 

crates/ecstore/src/bucket/migration.rs

@@ -15,7 +15,7 @@
 //! Migration of bucket metadata and IAM config from legacy format to RustFS format.
 
 use crate::bucket::metadata::BUCKET_METADATA_FILE;
-use crate::bucket::replication::{decode_resync_file, encode_resync_file};
+use crate::bucket::replication::ReplicationMigrationBridge;
 use crate::disk::{BUCKET_META_PREFIX, MIGRATING_META_BUCKET, RUSTFS_META_BUCKET};
 use crate::error::Error;
 use crate::object_api::{GetObjectReader, ObjectInfo, ObjectOptions, PutObjReader};
@@ -179,8 +179,9 @@ fn normalize_bucket_meta_blob(path: &str, data: &[u8]) -> std::result::Result<Op
     if !is_resync_meta_path(path) {
         return Ok(None);
     }
-    let status = decode_resync_file(data).map_err(|err| format!("decode resync meta failed: {err}"))?;
-    encode_resync_file(&status)
+    let status =
+        ReplicationMigrationBridge::decode_resync_status(data).map_err(|err| format!("decode resync meta failed: {err}"))?;
+    ReplicationMigrationBridge::encode_resync_status(&status)
         .map(Some)
         .map_err(|err| format!("encode resync meta failed: {err}"))
 }
@@ -420,7 +421,7 @@ where
 mod tests {
     use super::{normalize_bucket_meta_blob, normalize_iam_config_blob};
     use crate::bucket::replication::{
-        BucketReplicationResyncStatus, ResyncStatusType, TargetReplicationResyncStatus, decode_resync_file, encode_resync_file,
+        BucketReplicationResyncStatus, ReplicationMigrationBridge, ResyncStatusType, TargetReplicationResyncStatus,
     };
     use std::collections::HashMap;
 
@@ -459,12 +460,12 @@ mod tests {
             },
         )]);
 
-        let input = encode_resync_file(&status).expect("encode should succeed");
+        let input = ReplicationMigrationBridge::encode_resync_status(&status).expect("encode should succeed");
         let output = normalize_bucket_meta_blob(path, &input)
             .expect("normalize should succeed")
             .expect("resync path should be normalized");
 
-        let decoded = decode_resync_file(&output).expect("decode should succeed");
+        let decoded = ReplicationMigrationBridge::decode_resync_status(&output).expect("decode should succeed");
         assert_eq!(decoded.id, 123);
         assert_eq!(decoded.targets_map["arn:replication::1:dest"].resync_id, "reset-1");
     }

crates/ecstore/src/bucket/replication/README.md

@@ -15,8 +15,10 @@ and lifecycle/heal scheduling paths.
 | `replication_resyncer.rs` | Object replication, delete replication, resync, MRF encode/decode, target calls, and multipart target upload paths. | Depends on target calls and target config types through the replication target boundary, metadata paths and metadata systems through the replication metadata boundary, file metadata replication contracts through the filemeta boundary, error contracts through the error boundary, versioning systems, storage contracts through the replication storage boundary, config-derived storage class labels through the config store, runtime sources, notification events and local event host selection through the event sink, bandwidth reader wrapping, and SetDisks lock timing. |
 | `replication_state.rs` | Replication queue/stat state and worker accounting. | Reads runtime sources, file metadata replication contracts, error contracts, and bucket monitor handles through local boundaries, and owns shared replication pool/stat state. |
 | `replication_lifecycle_bridge.rs` | Lifecycle-originated delete replication admission and version-purge state construction. | Depends on replication config/rule matching, delete-replication decisions, and replication delete scheduling through a local contract type. |
+| `replication_migration_bridge.rs` | Bucket migration access to persisted replication resync codec helpers. | Keeps migration normalization behind a bridge instead of re-exporting resyncer codec helpers. |
 | `replication_object_bridge.rs` | App and SetDisks object replication decisions plus object/delete scheduling. | Keeps object write/delete replication call sites behind a bridge instead of exporting low-level resyncer and pool helpers. |
 | `replication_scanner_bridge.rs` | Scanner-originated replication heal admission. | Keeps scanner-facing heal queueing behind a local contract type instead of exporting the internal queue function directly. |
+| `replication_target_config_bridge.rs` | Bucket target removal checks against replication target rules. | Keeps bucket target sys from importing replication config helper types directly. |
 | `rule.rs` | Rule evaluation helpers for object replication options. | Depends on ECStore replication object option types. |
 | `mod.rs` | Explicit compatibility re-export facade for the current ECStore owner. | Wildcard re-exports are guarded so internal helpers do not leak back into the public facade. |
 
@@ -39,8 +41,10 @@ and lifecycle/heal scheduling paths.
 | `ReplicationMsgpCodec` | MessagePack time encode/decode and unknown value skipping for persisted resync/MRF state. | Bucket MessagePack helpers are exposed through the contract type in `replication_msgp_boundary.rs`. |
 | `ReplicationTagFilter` | Decode object tag strings for rule and metadata replication decisions. | Bucket tagging helper access is exposed through the contract type in `replication_tagging_boundary.rs`. |
 | `ReplicationLifecycleBridge` | Lifecycle-originated delete and version-purge scheduling. | Lifecycle delete paths call the bridge contract in `replication_lifecycle_bridge.rs` instead of constructing replication delete work directly. |
+| `ReplicationMigrationBridge` | Persisted resync status decode/encode access for bucket metadata migration. | Bucket migration calls the bridge contract in `replication_migration_bridge.rs` instead of importing internal resyncer codec helpers. |
 | `ReplicationObjectBridge` | Object write/delete replication decision and scheduling entry point for app storage and SetDisks paths. | App and SetDisks object paths call the bridge contract in `replication_object_bridge.rs` instead of importing internal resyncer/pool helpers. |
 | `ReplicationScannerBridge` | Scanner-originated replication heal scheduling. | Scanner heal paths call the bridge contract in `replication_scanner_bridge.rs` instead of importing the internal queue function directly. |
+| `ReplicationTargetConfigBridge` | Bucket target removal checks against replication target rules. | Bucket target sys calls the bridge contract in `replication_target_config_bridge.rs` instead of importing replication config helper types directly. |
 
 ## Migration Rules
 
@@ -67,6 +71,8 @@ and lifecycle/heal scheduling paths.
 9. Keep object write/delete replication helpers behind `ReplicationObjectBridge`;
    do not export internal resyncer or pool scheduling helpers through the
    compatibility facade.
+10. Keep ECStore owner modules outside `bucket/replication` behind bridge
+    contracts when they need replication codec or config helper behavior.
 
 ## First Code-Bearing Step
 

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

@@ -22,6 +22,7 @@ mod replication_filemeta_boundary;
 mod replication_lifecycle_bridge;
 mod replication_lock_boundary;
 mod replication_metadata_boundary;
+mod replication_migration_bridge;
 mod replication_msgp_boundary;
 mod replication_object_bridge;
 pub(crate) mod replication_pool;
@@ -31,6 +32,7 @@ mod replication_state;
 mod replication_storage_boundary;
 mod replication_tagging_boundary;
 mod replication_target_boundary;
+mod replication_target_config_bridge;
 mod replication_versioning_boundary;
 mod rule;
 mod runtime_boundary;
@@ -38,6 +40,7 @@ mod runtime_boundary;
 pub use config::{ObjectOpts, ReplicationConfigurationExt};
 pub use datatypes::ResyncStatusType;
 pub(crate) use replication_lifecycle_bridge::{ReplicationLifecycleBridge, ReplicationLifecycleConfig};
+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,
@@ -47,7 +50,7 @@ pub use replication_resyncer::{
     BucketReplicationResyncStatus, DeletedObjectReplicationInfo, MustReplicateOptions, ReplicationConfig, ResyncOpts,
     TargetReplicationResyncStatus,
 };
-pub(crate) use replication_resyncer::{decode_resync_file, encode_resync_file};
 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;

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

@@ -0,0 +1,28 @@
+// 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 super::replication_error_boundary::Result;
+use super::replication_resyncer::{BucketReplicationResyncStatus, decode_resync_file, encode_resync_file};
+
+pub(crate) struct ReplicationMigrationBridge;
+
+impl ReplicationMigrationBridge {
+    pub(crate) fn decode_resync_status(data: &[u8]) -> Result<BucketReplicationResyncStatus> {
+        decode_resync_file(data)
+    }
+
+    pub(crate) fn encode_resync_status(status: &BucketReplicationResyncStatus) -> Result<Vec<u8>> {
+        encode_resync_file(status)
+    }
+}

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

@@ -0,0 +1,48 @@
+// 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 s3s::dto::ReplicationConfiguration;
+
+use super::config::{ObjectOpts, ReplicationConfigurationExt};
+use super::replication_filemeta_boundary::ReplicationType;
+
+pub(crate) struct ReplicationTargetConfigBridge;
+
+impl ReplicationTargetConfigBridge {
+    pub(crate) fn target_is_used_by_rules(config: &ReplicationConfiguration, arn: &str) -> bool {
+        config
+            .filter_target_arns(&ObjectOpts {
+                op_type: ReplicationType::All,
+                ..Default::default()
+            })
+            .into_iter()
+            .any(|rule| rule == arn)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn target_config_bridge_matches_role_target() {
+        let config = ReplicationConfiguration {
+            role: " arn:target ".to_string(),
+            rules: Vec::new(),
+        };
+
+        assert!(ReplicationTargetConfigBridge::target_is_used_by_rules(&config, "arn:target"));
+        assert!(!ReplicationTargetConfigBridge::target_is_used_by_rules(&config, "arn:other"));
+    }
+}

docs/architecture/ecstore-module-split-plan.md

@@ -117,6 +117,9 @@ Current coupling:
   `ReplicationLifecycleBridge`, while scanner heal paths schedule replication
   work through `ReplicationScannerBridge`, and app/SetDisks object write/delete
   paths use `ReplicationObjectBridge`;
+- bucket metadata migration and bucket target removal checks use local
+  replication bridges instead of importing resyncer codec or config helper
+  internals;
 - global replication pool/stat initialization still lives with ECStore runtime
   compatibility state;
 - modules inside `bucket/replication` use local relative paths rather than the
@@ -174,12 +177,18 @@ Required contracts before crate movement:
 - `ReplicationLifecycleBridge`: lifecycle-originated delete and version-purge
   scheduling is exposed through the contract type in
   `crates/ecstore/src/bucket/replication/replication_lifecycle_bridge.rs`.
+- `ReplicationMigrationBridge`: persisted resync status decode/encode access
+  for bucket metadata migration is exposed through the contract type in
+  `crates/ecstore/src/bucket/replication/replication_migration_bridge.rs`.
 - `ReplicationObjectBridge`: app and SetDisks object write/delete replication
   decisions and scheduling are exposed through the contract type in
   `crates/ecstore/src/bucket/replication/replication_object_bridge.rs`.
 - `ReplicationScannerBridge`: scanner-originated replication heal scheduling is
   exposed through the contract type in
   `crates/ecstore/src/bucket/replication/replication_scanner_bridge.rs`.
+- `ReplicationTargetConfigBridge`: bucket target removal checks against
+  replication target rules are exposed through the contract type in
+  `crates/ecstore/src/bucket/replication/replication_target_config_bridge.rs`.
 - `ReplicationFacade`: the current `rustfs_ecstore::api::bucket::replication`
   compatibility surface is an explicit symbol list guarded against wildcard
   re-exports while downstream owners migrate to narrower contracts.

scripts/check_architecture_migration_rules.sh

@@ -210,6 +210,7 @@ REPLICATION_LOCK_BOUNDARY_BYPASS_HITS_FILE="${TMP_DIR}/replication_lock_boundary
 REPLICATION_METADATA_BOUNDARY_BYPASS_HITS_FILE="${TMP_DIR}/replication_metadata_boundary_bypass_hits.txt"
 REPLICATION_MSGP_BOUNDARY_BYPASS_HITS_FILE="${TMP_DIR}/replication_msgp_boundary_bypass_hits.txt"
 REPLICATION_RUNTIME_TYPE_BYPASS_HITS_FILE="${TMP_DIR}/replication_runtime_type_bypass_hits.txt"
+REPLICATION_ECSTORE_OWNER_BRIDGE_BYPASS_HITS_FILE="${TMP_DIR}/replication_ecstore_owner_bridge_bypass_hits.txt"
 REPLICATION_OBJECT_BRIDGE_BYPASS_HITS_FILE="${TMP_DIR}/replication_object_bridge_bypass_hits.txt"
 REPLICATION_SCANNER_BRIDGE_BYPASS_HITS_FILE="${TMP_DIR}/replication_scanner_bridge_bypass_hits.txt"
 REPLICATION_STORAGE_BOUNDARY_BYPASS_HITS_FILE="${TMP_DIR}/replication_storage_boundary_bypass_hits.txt"
@@ -2531,6 +2532,26 @@ if [[ -s "$REPLICATION_OBJECT_BRIDGE_BYPASS_HITS_FILE" ]]; then
   report_failure "object replication work entry points must stay behind ReplicationObjectBridge: $(paste -sd '; ' "$REPLICATION_OBJECT_BRIDGE_BYPASS_HITS_FILE")"
 fi
 
+(
+  cd "$ROOT_DIR"
+  {
+    rg -n --with-filename 'crate::bucket::replication::(decode_resync_file|encode_resync_file|ObjectOpts|ReplicationConfigurationExt)' \
+      crates/ecstore/src \
+      --glob '*.rs'
+    rg -n -U --with-filename 'use\s+crate::bucket::replication::\{[^}]*\b(decode_resync_file|encode_resync_file|ObjectOpts|ReplicationConfigurationExt)\b' \
+      crates/ecstore/src \
+      --glob '*.rs'
+    rg -n -U --with-filename 'crate::\{[^;]*bucket::replication::\{[^}]*\b(decode_resync_file|encode_resync_file|ObjectOpts|ReplicationConfigurationExt)\b' \
+      crates/ecstore/src \
+      --glob '*.rs'
+  } |
+    rg -v '^crates/ecstore/src/bucket/replication/' || true
+) >"$REPLICATION_ECSTORE_OWNER_BRIDGE_BYPASS_HITS_FILE"
+
+if [[ -s "$REPLICATION_ECSTORE_OWNER_BRIDGE_BYPASS_HITS_FILE" ]]; then
+  report_failure "ECStore owner modules must use replication bridge contracts instead of internal replication helpers: $(paste -sd '; ' "$REPLICATION_ECSTORE_OWNER_BRIDGE_BYPASS_HITS_FILE")"
+fi
+
 (
   cd "$ROOT_DIR"
   find crates/ecstore/src/bucket/replication -type f -name '*.rs' -print0 |