diff --git a/crates/ecstore/src/bucket/replication/README.md b/crates/ecstore/src/bucket/replication/README.md index d0dbac759..ec29085cb 100644 --- a/crates/ecstore/src/bucket/replication/README.md +++ b/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 - SetDisks lock timing directly. + bucket target sys, metadata sys, runtime sources, bandwidth reader, + 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. diff --git a/crates/ecstore/src/bucket/replication/mod.rs b/crates/ecstore/src/bucket/replication/mod.rs index 998d23086..f77157b1c 100644 --- a/crates/ecstore/src/bucket/replication/mod.rs +++ b/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; diff --git a/crates/ecstore/src/bucket/replication/replication_bandwidth_boundary.rs b/crates/ecstore/src/bucket/replication/replication_bandwidth_boundary.rs new file mode 100644 index 000000000..c307c2f4a --- /dev/null +++ b/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, + monitor: Arc, + bucket: &str, + arn: &str, + header_size: usize, +) -> Box { + Box::new(MonitoredReader::new( + monitor, + stream, + MonitorReaderOptions { + bucket_options: BucketOptions { + name: bucket.to_string(), + replication_arn: arn.to_string(), + }, + header_size, + }, + )) +} diff --git a/crates/ecstore/src/bucket/replication/replication_resyncer.rs b/crates/ecstore/src/bucket/replication/replication_resyncer.rs index 1f7fa776c..4c2f3c7d3 100644 --- a/crates/ecstore/src/bucket/replication/replication_resyncer.rs +++ b/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 { if let Some(monitor) = runtime_sources::bucket_monitor() { - Box::new(MonitoredReader::new( - monitor, - stream, - MonitorReaderOptions { - bucket_options: BucketOptions { - name: bucket.to_string(), - replication_arn: arn.to_string(), - }, - header_size, - }, - )) + replication_bandwidth_boundary::wrap_reader(stream, monitor, bucket, arn, header_size) } else { WARNED_MONITOR_UNINIT.call_once(|| { warn!( diff --git a/docs/architecture/ecstore-module-split-plan.md b/docs/architecture/ecstore-module-split-plan.md index 8a4ffdd25..c05f4a5e1 100644 --- a/docs/architecture/ecstore-module-split-plan.md +++ b/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 diff --git a/scripts/check_architecture_migration_rules.sh b/scripts/check_architecture_migration_rules.sh index ed9f6369d..44c8e0227 100755 --- a/scripts/check_architecture_migration_rules.sh +++ b/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\(' \