refactor(ecstore): add replication object bridge (#4140)

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

@@ -15,6 +15,7 @@ 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_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. |
 | `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. |
@@ -38,6 +39,7 @@ 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. |
+| `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. |
 
 ## Migration Rules
@@ -62,6 +64,9 @@ and lifecycle/heal scheduling paths.
    call semantics do not change.
 8. Keep the compatibility facade in `mod.rs` as an explicit symbol list. Do not
    reintroduce wildcard re-exports for replication implementation modules.
+9. Keep object write/delete replication helpers behind `ReplicationObjectBridge`;
+   do not export internal resyncer or pool scheduling helpers through the
+   compatibility facade.
 
 ## First Code-Bearing Step
 

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

@@ -23,6 +23,7 @@ mod replication_lifecycle_bridge;
 mod replication_lock_boundary;
 mod replication_metadata_boundary;
 mod replication_msgp_boundary;
+mod replication_object_bridge;
 pub(crate) mod replication_pool;
 mod replication_resyncer;
 mod replication_scanner_bridge;
@@ -37,13 +38,14 @@ mod runtime_boundary;
 pub use config::{ObjectOpts, ReplicationConfigurationExt};
 pub use datatypes::ResyncStatusType;
 pub(crate) use replication_lifecycle_bridge::{ReplicationLifecycleBridge, ReplicationLifecycleConfig};
+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, schedule_replication, schedule_replication_delete,
+    get_global_replication_stats, init_background_replication,
 };
 pub use replication_resyncer::{
     BucketReplicationResyncStatus, DeletedObjectReplicationInfo, MustReplicateOptions, ReplicationConfig, ResyncOpts,
-    TargetReplicationResyncStatus, check_replicate_delete, get_must_replicate_options, must_replicate,
+    TargetReplicationResyncStatus,
 };
 pub(crate) use replication_resyncer::{decode_resync_file, encode_resync_file};
 pub use replication_scanner_bridge::ReplicationScannerBridge;

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

@@ -0,0 +1,93 @@
+// 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::{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, MustReplicateOptions, check_replicate_delete, get_must_replicate_options, must_replicate,
+};
+use super::replication_storage_boundary::{ObjectInfo, ObjectOptions, ObjectToDelete, ReplicationStorage};
+
+pub struct ReplicationObjectBridge;
+
+impl ReplicationObjectBridge {
+    pub fn must_replicate_options(
+        user_defined: &HashMap<String, String>,
+        user_tags: String,
+        status: ReplicationStatusType,
+        op_type: ReplicationType,
+        opts: ObjectOptions,
+    ) -> MustReplicateOptions {
+        get_must_replicate_options(user_defined, user_tags, status, op_type, opts)
+    }
+
+    pub async fn must_replicate(bucket: &str, object: &str, options: MustReplicateOptions) -> ReplicateDecision {
+        must_replicate(bucket, object, options).await
+    }
+
+    pub async fn check_delete(
+        bucket: &str,
+        object: &ObjectToDelete,
+        source: &ObjectInfo,
+        opts: &ObjectOptions,
+        get_error: Option<String>,
+    ) -> ReplicateDecision {
+        check_replicate_delete(bucket, object, source, opts, get_error).await
+    }
+
+    pub async fn schedule_object<S: ReplicationStorage>(
+        object: ObjectInfo,
+        storage: Arc<S>,
+        decision: ReplicateDecision,
+        op_type: ReplicationType,
+    ) {
+        schedule_replication(object, storage, decision, op_type).await;
+    }
+
+    pub async fn schedule_delete(delete_object: DeletedObjectReplicationInfo) {
+        schedule_replication_delete(delete_object).await;
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn object_bridge_builds_operation_specific_options() {
+        let user_defined = HashMap::new();
+
+        let metadata = ReplicationObjectBridge::must_replicate_options(
+            &user_defined,
+            String::new(),
+            ReplicationStatusType::Empty,
+            ReplicationType::Metadata,
+            ObjectOptions::default(),
+        );
+        assert!(metadata.is_metadata_replication());
+        assert!(!metadata.is_existing_object_replication());
+
+        let existing = ReplicationObjectBridge::must_replicate_options(
+            &user_defined,
+            String::new(),
+            ReplicationStatusType::Empty,
+            ReplicationType::ExistingObject,
+            ObjectOptions::default(),
+        );
+        assert!(existing.is_existing_object_replication());
+        assert!(!existing.is_metadata_replication());
+    }
+}

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

@@ -1409,7 +1409,7 @@ pub fn get_global_replication_stats() -> Option<Arc<ReplicationStats>> {
     runtime_sources::replication_stats()
 }
 
-pub async fn schedule_replication<S: ReplicationStorage>(
+pub(crate) async fn schedule_replication<S: ReplicationStorage>(
     oi: ObjectInfo,
     o: Arc<S>,
     dsc: ReplicateDecision,
@@ -1461,7 +1461,7 @@ pub async fn schedule_replication<S: ReplicationStorage>(
     }
 }
 
-pub async fn schedule_replication_delete(dv: DeletedObjectReplicationInfo) {
+pub(crate) async fn schedule_replication_delete(dv: DeletedObjectReplicationInfo) {
     if let Some(pool) = runtime_sources::replication_pool() {
         let _ = pool.queue_replica_delete_task(dv.clone()).await;
     }

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

@@ -1506,7 +1506,7 @@ impl MustReplicateOptions {
     }
 }
 
-pub fn get_must_replicate_options(
+pub(crate) fn get_must_replicate_options(
     user_defined: &HashMap<String, String>,
     user_tags: String,
     status: ReplicationStatusType,
@@ -1517,7 +1517,7 @@ pub fn get_must_replicate_options(
 }
 
 /// Returns whether object version is a delete marker and if object qualifies for replication
-pub async fn check_replicate_delete(
+pub(crate) async fn check_replicate_delete(
     bucket: &str,
     dobj: &ObjectToDelete,
     oi: &ObjectInfo,
@@ -1668,7 +1668,7 @@ impl ObjectInfoExt for ObjectInfo {
     }
 }
 
-pub async fn must_replicate(bucket: &str, object: &str, mopts: MustReplicateOptions) -> ReplicateDecision {
+pub(crate) async fn must_replicate(bucket: &str, object: &str, mopts: MustReplicateOptions) -> ReplicateDecision {
     if runtime_sources::object_store_handle().is_none() {
         return ReplicateDecision::default();
     }

crates/ecstore/src/set_disk/mod.rs

@@ -20,7 +20,7 @@
 use crate::bucket::lifecycle::lifecycle::TRANSITION_COMPLETE;
 use crate::bucket::metadata_sys;
 use crate::bucket::object_lock::objectlock_sys::check_retention_for_modification;
-use crate::bucket::replication::check_replicate_delete;
+use crate::bucket::replication::ReplicationObjectBridge;
 use crate::bucket::versioning::VersioningApi;
 use crate::bucket::versioning_sys::BucketVersioningSys;
 use crate::client::{object_api_utils::get_raw_etag, transition_api::ReaderImpl};
@@ -3916,7 +3916,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
         let dsc = if should_preserve_delete_replication_state(&opts) {
             ReplicateDecision::default()
         } else {
-            check_replicate_delete(bucket, &otd, &goi, &opts, gerr.map(|e| e.to_string())).await
+            ReplicationObjectBridge::check_delete(bucket, &otd, &goi, &opts, gerr.map(|e| e.to_string())).await
         };
 
         if dsc.replicate_any() {

crates/ecstore/tests/replication_facade_compat_test.rs

@@ -16,8 +16,8 @@ mod storage_api;
 
 use s3s::dto::ReplicationConfiguration;
 use storage_api::replication_compat::{
-    BucketStats, DeletedObjectReplicationInfo, DynReplicationPool, ObjectOpts, ReplicationConfigurationExt, ReplicationStats,
-    ResyncStatusType,
+    BucketStats, DeletedObjectReplicationInfo, DynReplicationPool, ObjectOpts, ReplicationConfigurationExt,
+    ReplicationObjectBridge, ReplicationStats, ResyncStatusType,
 };
 
 fn type_name<T>() -> &'static str {
@@ -52,5 +52,6 @@ fn replication_facade_exports_runtime_and_dto_types() {
     assert!(type_name::<ReplicationStats>().contains("ReplicationStats"));
     assert!(type_name::<BucketStats>().contains("BucketStats"));
     assert!(type_name::<DeletedObjectReplicationInfo>().contains("DeletedObjectReplicationInfo"));
+    assert!(type_name::<ReplicationObjectBridge>().contains("ReplicationObjectBridge"));
     assert!(type_name_unsized::<DynReplicationPool>().contains("ReplicationPoolTrait"));
 }

crates/ecstore/tests/storage_api.rs

@@ -22,8 +22,8 @@ pub(crate) mod contract_compat {
 
 pub(crate) mod replication_compat {
     pub(crate) use rustfs_ecstore::api::bucket::replication::{
-        BucketStats, DeletedObjectReplicationInfo, DynReplicationPool, ObjectOpts, ReplicationConfigurationExt, ReplicationStats,
-        ResyncStatusType,
+        BucketStats, DeletedObjectReplicationInfo, DynReplicationPool, ObjectOpts, ReplicationConfigurationExt,
+        ReplicationObjectBridge, ReplicationStats, ResyncStatusType,
     };
 }
 

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

@@ -115,7 +115,8 @@ Current coupling:
   operation types are concentrated behind the replication target boundary;
 - lifecycle delete paths schedule replication work through
   `ReplicationLifecycleBridge`, while scanner heal paths schedule replication
-  work through `ReplicationScannerBridge`;
+  work through `ReplicationScannerBridge`, and app/SetDisks object write/delete
+  paths use `ReplicationObjectBridge`;
 - global replication pool/stat initialization still lives with ECStore runtime
   compatibility state;
 - modules inside `bucket/replication` use local relative paths rather than the
@@ -173,6 +174,9 @@ 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`.
+- `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`.

rustfs/src/app/storage_api.rs

@@ -630,6 +630,8 @@ pub(crate) mod bucket {
             crate::storage::storage_api::ecstore_bucket::replication::DeletedObjectReplicationInfo;
         pub(crate) type MustReplicateOptions = crate::storage::storage_api::ecstore_bucket::replication::MustReplicateOptions;
         pub(crate) type ObjectOpts = crate::storage::storage_api::ecstore_bucket::replication::ObjectOpts;
+        pub(crate) type ReplicationObjectBridge =
+            crate::storage::storage_api::ecstore_bucket::replication::ReplicationObjectBridge;
         pub(crate) type ReplicateDecision = rustfs_filemeta::ReplicateDecision;
 
         pub(crate) async fn check_replicate_delete(
@@ -639,8 +641,7 @@ pub(crate) mod bucket {
             del_opts: &crate::storage::storage_api::StorageObjectOptions,
             gerr: Option<String>,
         ) -> ReplicateDecision {
-            crate::storage::storage_api::ecstore_bucket::replication::check_replicate_delete(bucket, dobj, oi, del_opts, gerr)
-                .await
+            ReplicationObjectBridge::check_delete(bucket, dobj, oi, del_opts, gerr).await
         }
 
         pub(crate) fn get_must_replicate_options(
@@ -650,17 +651,11 @@ pub(crate) mod bucket {
             op_type: rustfs_filemeta::ReplicationType,
             opts: crate::storage::storage_api::StorageObjectOptions,
         ) -> MustReplicateOptions {
-            crate::storage::storage_api::ecstore_bucket::replication::get_must_replicate_options(
-                user_defined,
-                user_tags,
-                status,
-                op_type,
-                opts,
-            )
+            ReplicationObjectBridge::must_replicate_options(user_defined, user_tags, status, op_type, opts)
         }
 
         pub(crate) async fn must_replicate(bucket: &str, object: &str, mopts: MustReplicateOptions) -> ReplicateDecision {
-            crate::storage::storage_api::ecstore_bucket::replication::must_replicate(bucket, object, mopts).await
+            ReplicationObjectBridge::must_replicate(bucket, object, mopts).await
         }
 
         pub(crate) async fn schedule_replication(
@@ -669,11 +664,11 @@ pub(crate) mod bucket {
             dsc: ReplicateDecision,
             op_type: rustfs_filemeta::ReplicationType,
         ) {
-            crate::storage::storage_api::ecstore_bucket::replication::schedule_replication(oi, store, dsc, op_type).await;
+            ReplicationObjectBridge::schedule_object(oi, store, dsc, op_type).await;
         }
 
         pub(crate) async fn schedule_replication_delete(dv: DeletedObjectReplicationInfo) {
-            crate::storage::storage_api::ecstore_bucket::replication::schedule_replication_delete(dv).await;
+            ReplicationObjectBridge::schedule_delete(dv).await;
         }
     }
 

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_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"
 REPLICATION_TARGET_BOUNDARY_BYPASS_HITS_FILE="${TMP_DIR}/replication_target_boundary_bypass_hits.txt"
@@ -2515,6 +2516,21 @@ if [[ -s "$REPLICATION_FACADE_WILDCARD_EXPORT_HITS_FILE" ]]; then
   report_failure "replication facade must use explicit compatibility exports instead of wildcard re-exports: $(paste -sd '; ' "$REPLICATION_FACADE_WILDCARD_EXPORT_HITS_FILE")"
 fi
 
+(
+  cd "$ROOT_DIR"
+  {
+    rg -n --with-filename 'ecstore_bucket::replication::(check_replicate_delete|get_must_replicate_options|must_replicate|schedule_replication|schedule_replication_delete)|crate::bucket::replication::(check_replicate_delete|get_must_replicate_options|must_replicate|schedule_replication|schedule_replication_delete)' \
+      rustfs/src crates/ecstore/src \
+      --glob '*.rs'
+    rg -n --with-filename '\b(check_replicate_delete|get_must_replicate_options|must_replicate|schedule_replication|schedule_replication_delete)\b' \
+      crates/ecstore/src/bucket/replication/mod.rs
+  } || true
+) >"$REPLICATION_OBJECT_BRIDGE_BYPASS_HITS_FILE"
+
+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"
   find crates/ecstore/src/bucket/replication -type f -name '*.rs' -print0 |