From da7f421f69f35949741e28f8a37e7d62a0252b56 Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Tue, 30 Jun 2026 20:12:06 +0800 Subject: [PATCH] refactor(ecstore): route replication storage contracts (#4118) --- .../ecstore/src/bucket/replication/README.md | 7 +- crates/ecstore/src/bucket/replication/mod.rs | 2 + .../replication/replication_config_store.rs | 24 +--- .../bucket/replication/replication_pool.rs | 12 +- .../replication/replication_resyncer.rs | 130 +++++------------- .../replication_storage_boundary.rs | 121 ++++++++++++++++ .../architecture/ecstore-module-split-plan.md | 19 ++- scripts/check_architecture_migration_rules.sh | 20 +++ 8 files changed, 201 insertions(+), 134 deletions(-) create mode 100644 crates/ecstore/src/bucket/replication/replication_storage_boundary.rs diff --git a/crates/ecstore/src/bucket/replication/README.md b/crates/ecstore/src/bucket/replication/README.md index 3a8781f52..f085967c5 100644 --- a/crates/ecstore/src/bucket/replication/README.md +++ b/crates/ecstore/src/bucket/replication/README.md @@ -11,8 +11,8 @@ 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 target calls through the replication target boundary, metadata/versioning systems, ECStore object readers/writers, disk paths, runtime sources, notification events, bandwidth reader wrapping, and SetDisks lock timing. | +| `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, storage contracts through the replication storage boundary, runtime sources, and notification state. | +| `replication_resyncer.rs` | Object replication, delete replication, resync, MRF encode/decode, target calls, and multipart target upload paths. | Depends on target calls through the replication target boundary, metadata/versioning systems, storage contracts through the replication storage boundary, 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. | @@ -21,7 +21,8 @@ and lifecycle/heal scheduling paths. | Contract | Responsibility | Current dependency to remove | |---|---|---| -| `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. | +| `ReplicationObjectIO` | Object read/write primitives used by config, MRF, resync status, and multipart replication paths. | ECStore object API reader/writer types and storage-api object IO contracts are concentrated in `replication_storage_boundary.rs`. | +| `ReplicationStorage` | Object read/write/delete, object walk, metadata update, and target object IO. | ECStore object API, storage-api contracts, and read option types are concentrated in `replication_storage_boundary.rs`. | | `ReplicationMetadataStore` | Replication config, MRF/resync state, target reset headers, and status persistence. | Direct metadata sys, versioning sys, config storage, and file metadata imports. | | `ReplicationTargetStore` | Bucket target listing, target client lookup, target offline checks, and target operation option types. | Bucket target sys access and target operation types are concentrated in `replication_target_boundary.rs`. | | `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. | diff --git a/crates/ecstore/src/bucket/replication/mod.rs b/crates/ecstore/src/bucket/replication/mod.rs index f77157b1c..082183844 100644 --- a/crates/ecstore/src/bucket/replication/mod.rs +++ b/crates/ecstore/src/bucket/replication/mod.rs @@ -23,6 +23,7 @@ mod replication_msgp_boundary; mod replication_pool; mod replication_resyncer; mod replication_state; +mod replication_storage_boundary; mod replication_tagging_boundary; mod replication_target_boundary; mod replication_versioning_boundary; @@ -34,4 +35,5 @@ pub use datatypes::*; pub use replication_pool::*; pub use replication_resyncer::*; pub use replication_state::{BucketStats, ReplicationStats}; +pub use replication_storage_boundary::{ReplicationObjectIO, ReplicationStorage}; pub use rule::*; diff --git a/crates/ecstore/src/bucket/replication/replication_config_store.rs b/crates/ecstore/src/bucket/replication/replication_config_store.rs index 68f054fd5..0ee48f0e1 100644 --- a/crates/ecstore/src/bucket/replication/replication_config_store.rs +++ b/crates/ecstore/src/bucket/replication/replication_config_store.rs @@ -12,39 +12,21 @@ // See the License for the specific language governing permissions and // limitations under the License. +use super::replication_storage_boundary::ReplicationObjectIO; use crate::config::com; use crate::error::Result; -use crate::object_api::{GetObjectReader, ObjectInfo, ObjectOptions, PutObjReader}; -use crate::storage_api_contracts::{object::ObjectIO, range::HTTPRangeSpec}; -use http::HeaderMap; use std::sync::Arc; pub(crate) async fn read(api: Arc, file: &str) -> Result> where - S: ObjectIO< - Error = crate::error::Error, - RangeSpec = HTTPRangeSpec, - HeaderMap = HeaderMap, - ObjectOptions = ObjectOptions, - ObjectInfo = ObjectInfo, - GetObjectReader = GetObjectReader, - PutObjectReader = PutObjReader, - >, + S: ReplicationObjectIO, { com::read_config(api, file).await } pub(crate) async fn save(api: Arc, file: &str, data: Vec) -> Result<()> where - S: ObjectIO< - Error = crate::error::Error, - RangeSpec = HTTPRangeSpec, - HeaderMap = HeaderMap, - ObjectOptions = ObjectOptions, - ObjectInfo = ObjectInfo, - GetObjectReader = GetObjectReader, - PutObjectReader = PutObjReader, - >, + S: ReplicationObjectIO, { com::save_config(api, file, data).await } diff --git a/crates/ecstore/src/bucket/replication/replication_pool.rs b/crates/ecstore/src/bucket/replication/replication_pool.rs index 40d3fdf53..ab9e75f86 100644 --- a/crates/ecstore/src/bucket/replication/replication_pool.rs +++ b/crates/ecstore/src/bucket/replication/replication_pool.rs @@ -14,6 +14,7 @@ use super::replication_config_store as config_store; use super::replication_metadata_boundary as metadata_boundary; +use super::replication_storage_boundary::{DeletedObject, ObjectInfo, ObjectOptions, ReplicationObjectIO, ReplicationStorage}; use super::replication_target_boundary as target_boundary; use super::runtime_boundary as runtime_sources; use crate::bucket::replication::ResyncOpts; @@ -22,15 +23,12 @@ use crate::bucket::replication::replicate_delete; use crate::bucket::replication::replicate_object; use crate::bucket::replication::replication_resyncer::{ BucketReplicationResyncStatus, DeletedObjectReplicationInfo, MRF_REPLICATION_FILE, REPLICATION_DIR, RESYNC_FILE_NAME, - ReplicationConfig, ReplicationResyncer, ReplicationStorage, TargetReplicationResyncStatus, decode_mrf_file, - decode_resync_file, encode_mrf_file, get_heal_replicate_object_info, save_resync_status, + ReplicationConfig, ReplicationResyncer, TargetReplicationResyncStatus, decode_mrf_file, decode_resync_file, encode_mrf_file, + get_heal_replicate_object_info, save_resync_status, }; use crate::bucket::replication::replication_state::ReplicationStats; use crate::disk::BUCKET_META_PREFIX; use crate::error::Error as EcstoreError; -use crate::object_api::{ObjectInfo, ObjectOptions}; -use crate::storage_api_contracts::object::DeletedObject; -use crate::storage_api_contracts::object::EcstoreObjectIO; use lazy_static::lazy_static; use rustfs_filemeta::MrfOpKind; use rustfs_filemeta::MrfReplicateEntry; @@ -1254,7 +1252,7 @@ impl ReplicationPool { /// Returns `true` on success; on failure logs the error and returns `false`. /// Callers must NOT clear their in-memory buffer on `false` so the next tick /// can retry — otherwise a transient storage error permanently drops the batch. -async fn flush_mrf_to_disk(entries: &[MrfReplicateEntry], storage: &Arc) -> bool { +async fn flush_mrf_to_disk(entries: &[MrfReplicateEntry], storage: &Arc) -> bool { match encode_mrf_file(entries) { Ok(data) => { if let Err(e) = config_store::save(storage.clone(), MRF_REPLICATION_FILE, data).await { @@ -1283,7 +1281,7 @@ async fn flush_mrf_to_disk(entries: &[MrfReplicateEntry], st } /// Load bucket resync metadata from disk -async fn load_bucket_resync_metadata( +async fn load_bucket_resync_metadata( bucket: &str, obj_api: Arc, ) -> Result { diff --git a/crates/ecstore/src/bucket/replication/replication_resyncer.rs b/crates/ecstore/src/bucket/replication/replication_resyncer.rs index 2190cd20b..0f8f6fb7a 100644 --- a/crates/ecstore/src/bucket/replication/replication_resyncer.rs +++ b/crates/ecstore/src/bucket/replication/replication_resyncer.rs @@ -18,6 +18,10 @@ 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_msgp_boundary::{read_msgp_ext8_time, skip_msgp_value, write_msgp_time}; +use super::replication_storage_boundary::{ + AdvancedGetOptions, DeletedObject, EcstoreObjectOperations, HTTPRangeSpec, ObjectInfo, ObjectOptions, ObjectToDelete, + ReplicationObjectIO, ReplicationStorage, StatObjectOptions, WalkOptions, +}; use super::replication_tagging_boundary as tagging_boundary; use super::replication_target_boundary; use super::replication_target_boundary::{ @@ -28,16 +32,8 @@ use super::runtime_boundary as runtime_sources; use crate::bucket::replication::ResyncStatusType; use crate::bucket::replication::{ObjectOpts, ReplicationConfigurationExt as _}; use crate::bucket::target::BucketTargets; -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::storage_api_contracts::{ - list::{ListOperations, StorageListObjectVersionsInfo, StorageListObjectsV2Info, StorageObjectInfoOrErr, StorageWalkOptions}, - namespace::NamespaceLocking as StorageNamespaceLocking, - object::{DeletedObject, EcstoreObjectIO, EcstoreObjectOperations, ObjectIO, ObjectOperations, ObjectToDelete}, - range::HTTPRangeSpec, -}; use aws_sdk_s3::error::{ProvideErrorMetadata, SdkError}; use aws_sdk_s3::operation::head_object::{HeadObjectError, HeadObjectOutput}; use aws_sdk_s3::primitives::ByteStream; @@ -56,7 +52,7 @@ use http_body_util::StreamBody; use regex::Regex; use rmp_serde; use rustfs_filemeta::{ - FileInfo, MrfReplicateEntry, REPLICATE_EXISTING, REPLICATE_EXISTING_DELETE, ReplicateDecision, ReplicateObjectInfo, + MrfReplicateEntry, REPLICATE_EXISTING, REPLICATE_EXISTING_DELETE, ReplicateDecision, ReplicateObjectInfo, ReplicateTargetDecision, ReplicatedInfos, ReplicatedTargetInfo, ReplicationAction, ReplicationState, ReplicationStatusType, ReplicationType, ReplicationWorkerOperation, ResyncDecision, ResyncTargetDecision, VersionPurgeStatusType, get_replication_state, parse_replicate_decision, replication_statuses_map, target_reset_header, version_purge_statuses_map, @@ -105,68 +101,6 @@ const EVENT_RESYNC_TASK_FAILED: &str = "replication_resync_task_failed"; const EVENT_RESYNC_TARGET_OPERATION_FAILED: &str = "replication_resync_target_operation_failed"; const EVENT_RESYNC_RUNTIME_CHANNEL_FAILED: &str = "replication_resync_runtime_channel_failed"; -type ListObjectsV2Info = StorageListObjectsV2Info; -type ListObjectVersionsInfo = StorageListObjectVersionsInfo; -type ObjectInfoOrErr = StorageObjectInfoOrErr; -type WalkOptions = StorageWalkOptions bool>; - -/// Storage capabilities required by bucket replication workers. -pub trait ReplicationStorage: - ObjectIO< - Error = Error, - RangeSpec = HTTPRangeSpec, - HeaderMap = HeaderMap, - ObjectOptions = ObjectOptions, - ObjectInfo = ObjectInfo, - GetObjectReader = GetObjectReader, - PutObjectReader = PutObjReader, - > + ObjectOperations< - Error = Error, - ObjectInfo = ObjectInfo, - ObjectOptions = ObjectOptions, - FileInfo = FileInfo, - ObjectToDelete = ObjectToDelete, - DeletedObject = DeletedObject, - > + ListOperations< - Error = Error, - ListObjectsV2Info = ListObjectsV2Info, - ListObjectVersionsInfo = ListObjectVersionsInfo, - ObjectInfoOrErr = ObjectInfoOrErr, - WalkOptions = WalkOptions, - WalkCancellation = CancellationToken, - WalkResultSender = tokio::sync::mpsc::Sender, - > + StorageNamespaceLocking -{ -} - -impl ReplicationStorage for T where - T: ObjectIO< - Error = Error, - RangeSpec = HTTPRangeSpec, - HeaderMap = HeaderMap, - ObjectOptions = ObjectOptions, - ObjectInfo = ObjectInfo, - GetObjectReader = GetObjectReader, - PutObjectReader = PutObjReader, - > + ObjectOperations< - Error = Error, - ObjectInfo = ObjectInfo, - ObjectOptions = ObjectOptions, - FileInfo = FileInfo, - ObjectToDelete = ObjectToDelete, - DeletedObject = DeletedObject, - > + ListOperations< - Error = Error, - ListObjectsV2Info = ListObjectsV2Info, - ListObjectVersionsInfo = ListObjectVersionsInfo, - ObjectInfoOrErr = ObjectInfoOrErr, - WalkOptions = WalkOptions, - WalkCancellation = CancellationToken, - WalkResultSender = tokio::sync::mpsc::Sender, - > + StorageNamespaceLocking -{ -} - pub(crate) const REPLICATION_DIR: &str = ".replication"; pub(crate) const RESYNC_FILE_NAME: &str = "resync.bin"; pub(crate) const RESYNC_META_FORMAT: u16 = 1; @@ -590,15 +524,7 @@ impl ReplicationResyncer { pub async fn mark_status(&self, status: ResyncStatusType, opts: ResyncOpts, obj_layer: Arc) -> Result<()> where - S: ObjectIO< - Error = Error, - RangeSpec = HTTPRangeSpec, - HeaderMap = HeaderMap, - ObjectOptions = ObjectOptions, - ObjectInfo = ObjectInfo, - GetObjectReader = GetObjectReader, - PutObjectReader = PutObjReader, - >, + S: ReplicationObjectIO, { let bucket_status = { let mut status_map = self.status_map.write().await; @@ -720,15 +646,7 @@ impl ReplicationResyncer { pub async fn persist_to_disk(&self, cancel_token: CancellationToken, api: Arc) where - S: ObjectIO< - Error = Error, - RangeSpec = HTTPRangeSpec, - HeaderMap = HeaderMap, - ObjectOptions = ObjectOptions, - ObjectInfo = ObjectInfo, - GetObjectReader = GetObjectReader, - PutObjectReader = PutObjReader, - >, + S: ReplicationObjectIO, { let mut interval = tokio::time::interval(RESYNC_TIME_INTERVAL); @@ -782,7 +700,12 @@ impl ReplicationResyncer { } } - async fn resync_bucket_mark_status(&self, status: ResyncStatusType, opts: ResyncOpts, storage: Arc) { + async fn resync_bucket_mark_status( + &self, + status: ResyncStatusType, + opts: ResyncOpts, + storage: Arc, + ) { if let Err(err) = self.mark_status(status, opts.clone(), storage.clone()).await { error!( event = EVENT_RESYNC_STATUS_UPDATE_SKIPPED, @@ -1304,7 +1227,7 @@ pub async fn get_heal_replicate_object_info(oi: &ObjectInfo, rcfg: &ReplicationC } } -pub(crate) async fn save_resync_status( +pub(crate) async fn save_resync_status( bucket: &str, status: &BucketReplicationResyncStatus, api: Arc, @@ -2955,13 +2878,22 @@ pub async fn replicate_object(roi: ReplicateObjectInfo, s } trait ReplicateObjectInfoExt { - async fn replicate_object(&self, storage: Arc, tgt_client: Arc) -> ReplicatedTargetInfo; - async fn replicate_all(&self, storage: Arc, tgt_client: Arc) -> ReplicatedTargetInfo; + async fn replicate_object( + &self, + storage: Arc, + tgt_client: Arc, + ) -> ReplicatedTargetInfo; + async fn replicate_all(&self, storage: Arc, tgt_client: Arc) + -> ReplicatedTargetInfo; fn to_object_info(&self) -> ObjectInfo; } impl ReplicateObjectInfoExt for ReplicateObjectInfo { - async fn replicate_object(&self, storage: Arc, tgt_client: Arc) -> ReplicatedTargetInfo { + async fn replicate_object( + &self, + storage: Arc, + tgt_client: Arc, + ) -> ReplicatedTargetInfo { let bucket = self.bucket.clone(); let object = self.name.clone(); @@ -3256,7 +3188,11 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { rinfo } - async fn replicate_all(&self, storage: Arc, tgt_client: Arc) -> ReplicatedTargetInfo { + async fn replicate_all( + &self, + storage: Arc, + tgt_client: Arc, + ) -> ReplicatedTargetInfo { let start_time = OffsetDateTime::now_utc(); let bucket = self.bucket.clone(); @@ -3973,7 +3909,7 @@ fn part_range_spec_from_actual_size(offset: i64, part_size: i64) -> std::io::Res )) } -struct MultipartReplicationContext<'a, S: EcstoreObjectIO> { +struct MultipartReplicationContext<'a, S: ReplicationObjectIO> { storage: Arc, cli: Arc, src_bucket: &'a str, @@ -3985,7 +3921,7 @@ struct MultipartReplicationContext<'a, S: EcstoreObjectIO> { put_opts: PutObjectOptions, } -async fn replicate_object_with_multipart(ctx: MultipartReplicationContext<'_, S>) -> std::io::Result<()> { +async fn replicate_object_with_multipart(ctx: MultipartReplicationContext<'_, S>) -> std::io::Result<()> { let MultipartReplicationContext { storage, cli, diff --git a/crates/ecstore/src/bucket/replication/replication_storage_boundary.rs b/crates/ecstore/src/bucket/replication/replication_storage_boundary.rs new file mode 100644 index 000000000..6de5aa128 --- /dev/null +++ b/crates/ecstore/src/bucket/replication/replication_storage_boundary.rs @@ -0,0 +1,121 @@ +// 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 crate::error::Error; +use rustfs_filemeta::FileInfo; +use tokio_util::sync::CancellationToken; + +pub(crate) use crate::client::api_get_options::{AdvancedGetOptions, StatObjectOptions}; +pub(crate) use crate::object_api::{GetObjectReader, ObjectInfo, ObjectOptions, PutObjReader}; +pub(crate) use crate::storage_api_contracts::list::{ + ListOperations, StorageListObjectVersionsInfo, StorageListObjectsV2Info, StorageObjectInfoOrErr, StorageWalkOptions, +}; +pub(crate) use crate::storage_api_contracts::namespace::NamespaceLocking as StorageNamespaceLocking; +pub(crate) use crate::storage_api_contracts::object::{ + DeletedObject, EcstoreObjectOperations, ObjectIO, ObjectOperations, ObjectToDelete, +}; +pub(crate) use crate::storage_api_contracts::range::HTTPRangeSpec; + +type ListObjectsV2Info = StorageListObjectsV2Info; +type ListObjectVersionsInfo = StorageListObjectVersionsInfo; +type ObjectInfoOrErr = StorageObjectInfoOrErr; +pub(crate) type WalkOptions = StorageWalkOptions bool>; + +pub trait ReplicationObjectIO: + ObjectIO< + Error = Error, + RangeSpec = HTTPRangeSpec, + HeaderMap = http::HeaderMap, + ObjectOptions = ObjectOptions, + ObjectInfo = ObjectInfo, + GetObjectReader = GetObjectReader, + PutObjectReader = PutObjReader, + > + Send + + Sync + + std::fmt::Debug + + 'static +{ +} + +impl ReplicationObjectIO for T where + T: ObjectIO< + Error = Error, + RangeSpec = HTTPRangeSpec, + HeaderMap = http::HeaderMap, + ObjectOptions = ObjectOptions, + ObjectInfo = ObjectInfo, + GetObjectReader = GetObjectReader, + PutObjectReader = PutObjReader, + > + Send + + Sync + + std::fmt::Debug + + 'static +{ +} + +pub trait ReplicationStorage: + ObjectIO< + Error = Error, + RangeSpec = HTTPRangeSpec, + HeaderMap = http::HeaderMap, + ObjectOptions = ObjectOptions, + ObjectInfo = ObjectInfo, + GetObjectReader = GetObjectReader, + PutObjectReader = PutObjReader, + > + ObjectOperations< + Error = Error, + ObjectInfo = ObjectInfo, + ObjectOptions = ObjectOptions, + FileInfo = FileInfo, + ObjectToDelete = ObjectToDelete, + DeletedObject = DeletedObject, + > + ListOperations< + Error = Error, + ListObjectsV2Info = ListObjectsV2Info, + ListObjectVersionsInfo = ListObjectVersionsInfo, + ObjectInfoOrErr = ObjectInfoOrErr, + WalkOptions = WalkOptions, + WalkCancellation = CancellationToken, + WalkResultSender = tokio::sync::mpsc::Sender, + > + StorageNamespaceLocking +{ +} + +impl ReplicationStorage for T where + T: ObjectIO< + Error = Error, + RangeSpec = HTTPRangeSpec, + HeaderMap = http::HeaderMap, + ObjectOptions = ObjectOptions, + ObjectInfo = ObjectInfo, + GetObjectReader = GetObjectReader, + PutObjectReader = PutObjReader, + > + ObjectOperations< + Error = Error, + ObjectInfo = ObjectInfo, + ObjectOptions = ObjectOptions, + FileInfo = FileInfo, + ObjectToDelete = ObjectToDelete, + DeletedObject = DeletedObject, + > + ListOperations< + Error = Error, + ListObjectsV2Info = ListObjectsV2Info, + ListObjectVersionsInfo = ListObjectVersionsInfo, + ObjectInfoOrErr = ObjectInfoOrErr, + WalkOptions = WalkOptions, + WalkCancellation = CancellationToken, + WalkResultSender = tokio::sync::mpsc::Sender, + > + StorageNamespaceLocking +{ +} diff --git a/docs/architecture/ecstore-module-split-plan.md b/docs/architecture/ecstore-module-split-plan.md index 01b98ea43..5ca3cdeb4 100644 --- a/docs/architecture/ecstore-module-split-plan.md +++ b/docs/architecture/ecstore-module-split-plan.md @@ -101,11 +101,12 @@ Focused verification for the first code-bearing lifecycle PR: 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 and bandwidth reader access through local boundaries, local node - names, and notification events; +- replication workers depend on `ReplicationStorage`, ECStore object APIs and + storage-api contracts through the replication storage boundary, bucket target + clients, bucket metadata, file metadata replication state, scanner repair + classification, runtime replication pool/stat handles, bucket monitor and + bandwidth reader access through local boundaries, local node names, and + notification events; - resync and delete replication paths call metadata directly, while bucket target system access and target operation types are concentrated behind the replication target boundary; @@ -116,9 +117,15 @@ Current coupling: Required contracts before crate movement: +- `ReplicationObjectIO`: object read/write primitives for config, MRF, resync + status, and multipart replication paths. ECStore object API reader/writer + types and storage-api object IO contracts are concentrated in + `crates/ecstore/src/bucket/replication/replication_storage_boundary.rs`. - `ReplicationStorage`: keep the existing trait as the starting point, then split object read/write/delete, walk, and metadata update responsibilities - only when call sites prove a narrower shape. + only when call sites prove a narrower shape. ECStore object API, + storage-api contracts, and read option types are concentrated in + `crates/ecstore/src/bucket/replication/replication_storage_boundary.rs`. - `ReplicationMetadataStore`: replication config, target reset headers, MRF/resync state, and status persistence. - `ReplicationTargetStore`: bucket target listing, target client lookup, diff --git a/scripts/check_architecture_migration_rules.sh b/scripts/check_architecture_migration_rules.sh index b5ebcafff..9c7203f3f 100755 --- a/scripts/check_architecture_migration_rules.sh +++ b/scripts/check_architecture_migration_rules.sh @@ -204,6 +204,7 @@ REPLICATION_EVENT_SINK_BYPASS_HITS_FILE="${TMP_DIR}/replication_event_sink_bypas 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_MSGP_BOUNDARY_BYPASS_HITS_FILE="${TMP_DIR}/replication_msgp_boundary_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" REPLICATION_TAGGING_BOUNDARY_BYPASS_HITS_FILE="${TMP_DIR}/replication_tagging_boundary_bypass_hits.txt" REPLICATION_VERSIONING_BOUNDARY_BYPASS_HITS_FILE="${TMP_DIR}/replication_versioning_boundary_bypass_hits.txt" @@ -2533,6 +2534,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::(?:client::api_get_options|object_api|storage_api_contracts)\b|crate::\{[^;]*\b(?:api_get_options|object_api|storage_api_contracts)\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_storage_boundary\.rs:' || true +) >"$REPLICATION_STORAGE_BOUNDARY_BYPASS_HITS_FILE" + +if [[ -s "$REPLICATION_STORAGE_BOUNDARY_BYPASS_HITS_FILE" ]]; then + report_failure "replication storage contracts must stay behind replication storage boundary: $(paste -sd '; ' "$REPLICATION_STORAGE_BOUNDARY_BYPASS_HITS_FILE")" +fi + ( cd "$ROOT_DIR" find crates/ecstore/src/bucket/replication -type f -name '*.rs' -print0 |