refactor(ecstore): route replication bandwidth wrapping (#4113)

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

@@ -12,7 +12,7 @@ and lifecycle/heal scheduling paths.
 | `config.rs` | Replication config helpers, rule matching, and tag filtering. | Uses replication-local tagging boundary and S3 DTOs directly. |
 | `datatypes.rs` | Replication status and operation DTOs. | Publicly re-exported through the ECStore replication facade. |
 | `replication_pool.rs` | Replication queue, worker pool, MRF persistence, bucket stats, and delete/object scheduling. | Depends on bucket target sys, bucket metadata sys, config storage, object API, runtime sources, notification state, and storage-api object IO contracts. |
-| `replication_resyncer.rs` | Object replication, delete replication, resync, MRF encode/decode, target calls, and multipart target upload paths. | Depends on bucket target clients, metadata/versioning systems, ECStore object readers/writers, disk paths, runtime sources, notification events, and SetDisks lock timing. |
+| `replication_resyncer.rs` | Object replication, delete replication, resync, MRF encode/decode, target calls, and multipart target upload paths. | Depends on bucket target clients, metadata/versioning systems, ECStore object readers/writers, disk paths, runtime sources, notification events, bandwidth reader wrapping, and SetDisks lock timing. |
 | `replication_state.rs` | Replication queue/stat state and worker accounting. | Reads runtime sources and owns shared replication pool/stat state. |
 | `rule.rs` | Rule evaluation helpers for object replication options. | Depends on ECStore replication object option types. |
 | `mod.rs` | Compatibility re-export facade for the current ECStore owner. | Must stay stable until downstream scanner, lifecycle, heal, and metrics paths compile through replacement contracts. |
@@ -24,6 +24,7 @@ and lifecycle/heal scheduling paths.
 | `ReplicationStorage` | Object read/write/delete, object walk, metadata update, and target object IO. | Direct ECStore object API and storage-api object IO coupling in worker paths. |
 | `ReplicationMetadataStore` | Replication config, bucket targets, MRF/resync state, target reset headers, and status persistence. | Direct bucket target sys, metadata sys, versioning sys, config storage, and file metadata imports. |
 | `ReplicationRuntime` | Worker pool, queue sizing, stats, bucket monitor, local node identity, cancellation, and admission state. | Direct runtime source/global access and shared replication pool/stat state. |
+| `ReplicationBandwidthLimiter` | Target reader wrapping for replication bandwidth accounting and throttling. | Direct bucket bandwidth reader imports from resyncer paths. |
 | `ReplicationEventSink` | Notification and audit events for skipped, failed, pending, and completed replication operations. | Direct event notification service calls from worker code. |
 | `ReplicationTagFilter` | Decode object tag strings for rule and metadata replication decisions. | Direct bucket tagging helper imports from replication workers. |
 | `ReplicationLifecycleBridge` | Lifecycle-originated delete and version-purge scheduling. | Direct coupling between lifecycle worker paths and replication delete scheduling. |
@@ -31,8 +32,8 @@ and lifecycle/heal scheduling paths.
 ## Migration Rules
 
 1. Do not move `bucket/replication` into a new crate while workers import
-   bucket target sys, metadata sys, runtime sources, notification services, or
+   bucket target sys, metadata sys, runtime sources, bandwidth reader,
-   SetDisks lock timing directly.
+   notification services, or SetDisks lock timing directly.
 2. Keep existing queue behavior, MRF persistence, resync state, target client
    semantics, notification/audit events, and scanner/heal classifications
    unchanged during inventory and contract PRs.

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

@@ -14,6 +14,7 @@
 
 mod config;
 pub mod datatypes;
+mod replication_bandwidth_boundary;
 mod replication_config_store;
 mod replication_event_sink;
 mod replication_lock_boundary;

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

@@ -0,0 +1,40 @@
+// 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::sync::Arc;
+
+use tokio::io::AsyncRead;
+
+use crate::bucket::bandwidth::monitor::Monitor;
+use crate::bucket::bandwidth::reader::{BucketOptions, MonitorReaderOptions, MonitoredReader};
+
+pub(crate) fn wrap_reader(
+    stream: Box<dyn AsyncRead + Unpin + Send + Sync>,
+    monitor: Arc<Monitor>,
+    bucket: &str,
+    arn: &str,
+    header_size: usize,
+) -> Box<dyn AsyncRead + Unpin + Send + Sync> {
+    Box::new(MonitoredReader::new(
+        monitor,
+        stream,
+        MonitorReaderOptions {
+            bucket_options: BucketOptions {
+                name: bucket.to_string(),
+                replication_arn: arn.to_string(),
+            },
+            header_size,
+        },
+    ))
+}

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

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use super::replication_bandwidth_boundary;
 use super::replication_config_store as config_store;
 use super::replication_event_sink::{EventArgs, send_event};
 use super::replication_lock_boundary as lock_boundary;
@@ -21,7 +22,6 @@ use super::replication_tagging_boundary as tagging_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::{
     AdvancedPutOptions, PutObjectOptions, PutObjectPartOptions, RemoveObjectOptions, TargetClient,
 };
@@ -3703,17 +3703,7 @@ fn wrap_with_bandwidth_monitor_with_header(
     header_size: usize,
 ) -> Box<dyn AsyncRead + Unpin + Send + Sync> {
     if let Some(monitor) = runtime_sources::bucket_monitor() {
-        Box::new(MonitoredReader::new(
+        replication_bandwidth_boundary::wrap_reader(stream, monitor, bucket, arn, header_size)
-            monitor,
-            stream,
-            MonitorReaderOptions {
-                bucket_options: BucketOptions {
-                    name: bucket.to_string(),
-                    replication_arn: arn.to_string(),
-                },
-                header_size,
-            },
-        ))
     } else {
         WARNED_MONITOR_UNINIT.call_once(|| {
             warn!(

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

@@ -104,7 +104,8 @@ Current coupling:
 - replication workers depend on `ReplicationStorage`, ECStore object APIs,
   bucket target clients, bucket metadata, file metadata replication state,
   scanner repair classification, runtime replication pool/stat handles, bucket
-  monitor, local node names, and notification events;
+  monitor and bandwidth reader access through local boundaries, local node
+  names, and notification events;
 - resync and delete replication paths call metadata and bucket target systems
   directly;
 - lifecycle and heal paths schedule replication work through the current ECStore
@@ -121,6 +122,8 @@ Required contracts before crate movement:
   headers, MRF/resync state, and status persistence.
 - `ReplicationRuntime`: pool, stats, worker admission, bucket monitor, local
   node identity, cancellation, and queue sizing.
+- `ReplicationBandwidthLimiter`: target reader wrapping for replication
+  bandwidth accounting and throttling.
 - `ReplicationEventSink`: notification/audit events for skipped, failed, and
   completed replication operations.
 - `ReplicationLifecycleBridge`: lifecycle-originated delete and version-purge

scripts/check_architecture_migration_rules.sh

@@ -198,6 +198,7 @@ EXTERNAL_TEST_ECSTORE_COMPAT_BYPASS_HITS_FILE="${TMP_DIR}/external_test_ecstore_
 FUZZ_ECSTORE_COMPAT_BYPASS_HITS_FILE="${TMP_DIR}/fuzz_ecstore_compat_bypass_hits.txt"
 EXTERNAL_ECSTORE_API_BOUNDARY_HITS_FILE="${TMP_DIR}/external_ecstore_api_boundary_hits.txt"
 REPLICATION_FACADE_BYPASS_HITS_FILE="${TMP_DIR}/replication_facade_bypass_hits.txt"
+REPLICATION_BANDWIDTH_BOUNDARY_BYPASS_HITS_FILE="${TMP_DIR}/replication_bandwidth_boundary_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"
@@ -2532,6 +2533,25 @@ 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"
+  find crates/ecstore/src/bucket/replication -type f -name '*.rs' -print0 |
+    xargs -0 perl -0ne '
+      while (/crate::bucket::bandwidth(?:::\{[^;]*\breader\b|::reader\b)/sg) {
+        my $prefix = substr($_, 0, $-[0]);
+        my $line = ($prefix =~ tr/\n//) + 1;
+        my $match = $&;
+        $match =~ s/\s+/ /g;
+        print "$ARGV:$line:$match\n";
+      }
+    ' |
+    rg -v '^crates/ecstore/src/bucket/replication/replication_bandwidth_boundary\.rs:' || true
+) >"$REPLICATION_BANDWIDTH_BOUNDARY_BYPASS_HITS_FILE"
+
+if [[ -s "$REPLICATION_BANDWIDTH_BOUNDARY_BYPASS_HITS_FILE" ]]; then
+  report_failure "replication bandwidth reader access must stay behind replication bandwidth boundary: $(paste -sd '; ' "$REPLICATION_BANDWIDTH_BOUNDARY_BYPASS_HITS_FILE")"
+fi
+
 (
   cd "$ROOT_DIR"
   rg -n --with-filename 'crate::set_disk::get_lock_acquire_timeout|\bget_lock_acquire_timeout\(' \