refactor: narrow replication storage consumers (#3485)

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

@@ -12,7 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use crate::StorageAPI;
 use crate::bucket::bucket_target_sys::BucketTargetSys;
 use crate::bucket::metadata_sys;
 use crate::bucket::replication::ResyncOpts;
@@ -21,14 +20,14 @@ 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, TargetReplicationResyncStatus, decode_mrf_file, decode_resync_file, encode_mrf_file,
+    ReplicationConfig, ReplicationResyncer, ReplicationStorage, TargetReplicationResyncStatus, decode_mrf_file,
-    get_heal_replicate_object_info, save_resync_status,
+    decode_resync_file, encode_mrf_file, get_heal_replicate_object_info, save_resync_status,
 };
 use crate::bucket::replication::replication_state::ReplicationStats;
 use crate::config::com::{read_config, save_config};
 use crate::disk::BUCKET_META_PREFIX;
 use crate::error::Error as EcstoreError;
-use crate::store_api::{NamespaceLocking, ObjectIO, ObjectInfo, ObjectOptions};
+use crate::store_api::{ObjectIO, ObjectInfo, ObjectOptions};
 use lazy_static::lazy_static;
 use rustfs_filemeta::MrfOpKind;
 use rustfs_filemeta::MrfReplicateEntry;
@@ -213,7 +212,7 @@ impl Default for ReplicationPoolOpts {
 }
 /// Main replication pool structure
 #[derive(Debug)]
-pub struct ReplicationPool<S: StorageAPI + NamespaceLocking> {
+pub struct ReplicationPool<S: ReplicationStorage> {
     // Atomic counters for active workers
     active_workers: Arc<AtomicI32>,
     active_lrg_workers: Arc<AtomicI32>,
@@ -254,7 +253,7 @@ pub struct ReplicationPool<S: StorageAPI + NamespaceLocking> {
     resyncer: Arc<ReplicationResyncer>,
 }
 
-impl<S: StorageAPI + NamespaceLocking> ReplicationPool<S> {
+impl<S: ReplicationStorage> ReplicationPool<S> {
     /// Creates a new replication pool with specified options
     pub async fn new(opts: ReplicationPoolOpts, stats: Arc<ReplicationStats>, storage: Arc<S>) -> Arc<Self> {
         let max_workers = opts.max_workers.unwrap_or(WORKER_MAX_LIMIT);
@@ -1330,7 +1329,7 @@ pub trait ReplicationPoolTrait: std::fmt::Debug {
 
 // Implement the trait for ReplicationPool
 #[async_trait::async_trait]
-impl<S: StorageAPI + NamespaceLocking> ReplicationPoolTrait for ReplicationPool<S> {
+impl<S: ReplicationStorage> ReplicationPoolTrait for ReplicationPool<S> {
     fn active_workers(&self) -> i32 {
         ReplicationPool::<S>::active_workers(self)
     }
@@ -1382,7 +1381,7 @@ lazy_static! {
 }
 
 /// Initializes background replication with the given options
-pub async fn init_background_replication<S: StorageAPI + NamespaceLocking>(storage: Arc<S>) {
+pub async fn init_background_replication<S: ReplicationStorage>(storage: Arc<S>) {
     let stats = GLOBAL_REPLICATION_STATS
         .get_or_init(|| async {
             let stats = Arc::new(ReplicationStats::new());
@@ -1406,7 +1405,7 @@ pub fn get_global_replication_pool() -> Option<Arc<DynReplicationPool>> {
     GLOBAL_REPLICATION_POOL.get().cloned()
 }
 
-pub async fn schedule_replication<S: StorageAPI + NamespaceLocking>(
+pub async fn schedule_replication<S: ReplicationStorage>(
     oi: ObjectInfo,
     o: Arc<S>,
     dsc: ReplicateDecision,

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

@@ -31,11 +31,12 @@ use crate::error::{Error, Result, is_err_object_not_found, is_err_version_not_fo
 use crate::event_notification::{EventArgs, send_event};
 use crate::global::GLOBAL_LocalNodeName;
 use crate::global::get_global_bucket_monitor;
+use crate::resolve_object_store_handle;
 use crate::set_disk::get_lock_acquire_timeout;
 use crate::store_api::{
-    DeletedObject, HTTPRangeSpec, NamespaceLocking, ObjectIO, ObjectInfo, ObjectOptions, ObjectToDelete, WalkOptions,
+    DeletedObject, HTTPRangeSpec, ListOperations, NamespaceLocking, ObjectIO, ObjectInfo, ObjectOperations, ObjectOptions,
+    ObjectToDelete, WalkOptions,
 };
-use crate::{StorageAPI, resolve_object_store_handle};
 use aws_sdk_s3::error::{ProvideErrorMetadata, SdkError};
 use aws_sdk_s3::operation::head_object::{HeadObjectError, HeadObjectOutput};
 use aws_sdk_s3::primitives::ByteStream;
@@ -103,6 +104,11 @@ 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";
 
+/// Storage capabilities required by bucket replication workers.
+pub trait ReplicationStorage: ObjectIO + ObjectOperations + ListOperations + NamespaceLocking {}
+
+impl<T> ReplicationStorage for T where T: ObjectIO + ObjectOperations + ListOperations + NamespaceLocking {}
+
 pub(crate) const REPLICATION_DIR: &str = ".replication";
 pub(crate) const RESYNC_FILE_NAME: &str = "resync.bin";
 pub(crate) const RESYNC_META_FORMAT: u16 = 1;
@@ -707,7 +713,7 @@ impl ReplicationResyncer {
     }
 
     #[instrument(skip(cancellation_token, storage))]
-    pub async fn resync_bucket<S: StorageAPI + NamespaceLocking>(
+    pub async fn resync_bucket<S: ReplicationStorage>(
         self: Arc<Self>,
         cancellation_token: CancellationToken,
         storage: Arc<S>,
@@ -1734,7 +1740,7 @@ pub async fn must_replicate(bucket: &str, object: &str, mopts: MustReplicateOpti
     dsc
 }
 
-pub async fn replicate_delete<S: StorageAPI + NamespaceLocking>(dobj: DeletedObjectReplicationInfo, storage: Arc<S>) {
+pub async fn replicate_delete<S: ReplicationStorage>(dobj: DeletedObjectReplicationInfo, storage: Arc<S>) {
     if dobj.delete_object.force_delete {
         replicate_force_delete_to_targets(&dobj, storage).await;
         return;
@@ -2177,7 +2183,7 @@ pub async fn replicate_delete<S: StorageAPI + NamespaceLocking>(dobj: DeletedObj
     }
 }
 
-async fn source_delete_marker_missing<S: StorageAPI>(
+async fn source_delete_marker_missing<S: ObjectOperations>(
     storage: &S,
     bucket: &str,
     object_name: &str,
@@ -2236,10 +2242,7 @@ async fn replicate_delete_marker_purge_to_targets(bucket: &str, dobj: &DeletedOb
     }
 }
 
-async fn replicate_force_delete_to_targets<S: StorageAPI + NamespaceLocking>(
+async fn replicate_force_delete_to_targets<S: ReplicationStorage>(dobj: &DeletedObjectReplicationInfo, storage: Arc<S>) {
-    dobj: &DeletedObjectReplicationInfo,
-    storage: Arc<S>,
-) {
     let bucket = &dobj.bucket;
     let object_name = &dobj.delete_object.object_name;
 
@@ -2634,7 +2637,7 @@ async fn replicate_delete_to_target(dobj: &DeletedObjectReplicationInfo, tgt_cli
     rinfo
 }
 
-pub async fn replicate_object<S: StorageAPI + NamespaceLocking>(roi: ReplicateObjectInfo, storage: Arc<S>) {
+pub async fn replicate_object<S: ReplicationStorage>(roi: ReplicateObjectInfo, storage: Arc<S>) {
     let bucket = roi.bucket.clone();
     let object = roi.name.clone();
 
@@ -2872,13 +2875,13 @@ pub async fn replicate_object<S: StorageAPI + NamespaceLocking>(roi: ReplicateOb
 }
 
 trait ReplicateObjectInfoExt {
-    async fn replicate_object<S: StorageAPI>(&self, storage: Arc<S>, tgt_client: Arc<TargetClient>) -> ReplicatedTargetInfo;
+    async fn replicate_object<S: ObjectIO>(&self, storage: Arc<S>, tgt_client: Arc<TargetClient>) -> ReplicatedTargetInfo;
-    async fn replicate_all<S: StorageAPI>(&self, storage: Arc<S>, tgt_client: Arc<TargetClient>) -> ReplicatedTargetInfo;
+    async fn replicate_all<S: ObjectIO>(&self, storage: Arc<S>, tgt_client: Arc<TargetClient>) -> ReplicatedTargetInfo;
     fn to_object_info(&self) -> ObjectInfo;
 }
 
 impl ReplicateObjectInfoExt for ReplicateObjectInfo {
-    async fn replicate_object<S: StorageAPI>(&self, storage: Arc<S>, tgt_client: Arc<TargetClient>) -> ReplicatedTargetInfo {
+    async fn replicate_object<S: ObjectIO>(&self, storage: Arc<S>, tgt_client: Arc<TargetClient>) -> ReplicatedTargetInfo {
         let bucket = self.bucket.clone();
         let object = self.name.clone();
 
@@ -3173,7 +3176,7 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo {
         rinfo
     }
 
-    async fn replicate_all<S: StorageAPI>(&self, storage: Arc<S>, tgt_client: Arc<TargetClient>) -> ReplicatedTargetInfo {
+    async fn replicate_all<S: ObjectIO>(&self, storage: Arc<S>, tgt_client: Arc<TargetClient>) -> ReplicatedTargetInfo {
         let start_time = OffsetDateTime::now_utc();
 
         let bucket = self.bucket.clone();
@@ -3895,7 +3898,7 @@ fn part_range_spec_from_actual_size(offset: i64, part_size: i64) -> std::io::Res
     ))
 }
 
-struct MultipartReplicationContext<'a, S: StorageAPI> {
+struct MultipartReplicationContext<'a, S: ObjectIO> {
     storage: Arc<S>,
     cli: Arc<TargetClient>,
     src_bucket: &'a str,
@@ -3907,7 +3910,7 @@ struct MultipartReplicationContext<'a, S: StorageAPI> {
     put_opts: PutObjectOptions,
 }
 
-async fn replicate_object_with_multipart<S: StorageAPI>(ctx: MultipartReplicationContext<'_, S>) -> std::io::Result<()> {
+async fn replicate_object_with_multipart<S: ObjectIO>(ctx: MultipartReplicationContext<'_, S>) -> std::io::Result<()> {
     let MultipartReplicationContext {
         storage,
         cli,

docs/architecture/migration-progress.md

@@ -5,12 +5,12 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block
 ## Current Context
 
 - Issue: [`rustfs/backlog#660`](https://github.com/rustfs/backlog/issues/660)
-- Branch: `overtrue/arch-storage-namespace-consumer-cleanup`
+- Branch: `overtrue/arch-storage-namespace-lock-large-cleanup`
-- Baseline: `origin/main` at `307d788da1954ddaeac8a55d5040a7ce7d1a213f`
+- Baseline: `origin/main` at `49c0f131205035d125271bb5b87db5b0f5bc2a6d`
 - PR type for this branch: `consumer-migration`
 - Runtime behavior changes: no external behavior change expected.
-- Rust code changes: narrow the table catalog object backend and rebalance
+- Rust code changes: narrow replication pool, resync, delete replication, and
-  metadata merge-save helper from full `StorageAPI` bounds to their actual
+  object replication storage bounds away from full `StorageAPI` to their actual
   object I/O, object operation, list, and namespace-lock capabilities.
 - CI/script changes: none.
 - Docs changes: record the current `API-012` consumer cleanup slice and its
@@ -459,9 +459,14 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block
     namespace locking directly on ECStore storage types, and remove the
     temporary namespace-lock compatibility method from the full storage trait
     and cleanup register entry.
-  - Current cleanup slice: narrow remaining table catalog backend and rebalance
+  - Completed cleanup slice: `rustfs/rustfs#3477` narrowed remaining table
-    metadata helper consumers away from full `StorageAPI` where they only need
+    catalog backend and rebalance metadata helper consumers away from full
-    object I/O, object operations, list operations, and namespace locking.
+    `StorageAPI` where they only need object I/O, object operations, list
+    operations, and namespace locking.
+  - Current cleanup slice: narrow replication pool, resync leader-lock, delete
+    replication, object replication, and multipart replication helpers away
+    from full `StorageAPI` where they only need object I/O, object operations,
+    list operations, and namespace locking.
   - Acceptance: table catalog object backend contracts express the actual
     object read/write, metadata/delete, list, and namespace-lock capabilities
     they need; namespace-lock consumers depend on `NamespaceLocking` instead of
@@ -754,16 +759,18 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block
 
 | Expert | Status | Notes |
 |---|---|---|
-| Quality/architecture | passed | Trait bounds now express the actual object I/O, object operation, list, and namespace-lock capabilities without adding abstractions or changing method bodies. |
+| Quality/architecture | passed | Replication consumers now depend on a local replication storage capability boundary plus object-I/O-only helper bounds, without changing replication method bodies. |
-| Migration preservation | passed | Lock acquisition, table catalog object paths, optimistic preconditions, pagination, missing-object handling, and rebalance metadata save semantics are unchanged. |
+| Migration preservation | passed | Resync leader locks, per-object replication locks, delete replication, object reader flow, multipart upload flow, MRF recovery, and replication scheduling semantics are unchanged. |
-| Testing/verification | passed | Focused table catalog and rebalance tests, compile check, migration/layer guards, formatting, diff hygiene, Rust risk scan, and full `make pre-commit` passed. |
+| Testing/verification | passed | Focused replication tests, storage API compatibility test, compile checks, migration/layer guards, formatting, diff hygiene, Rust risk scan, and full `make pre-commit` passed. |
 
 ## Verification Notes
 
-Passed on `307d788da1954ddaeac8a55d5040a7ce7d1a213f`:
+Passed on `49c0f131205035d125271bb5b87db5b0f5bc2a6d`:
 
-- `cargo test -p rustfs table_catalog --no-fail-fast`: passed.
+- `cargo check -p rustfs-ecstore`: passed.
-- `cargo test -p rustfs-ecstore rebalance --no-fail-fast`: passed.
+- `cargo test -p rustfs-ecstore bucket::replication --no-fail-fast`: passed.
+- `cargo test -p rustfs-ecstore --test storage_api_compat_test --no-fail-fast`:
+  passed.
 - `cargo check -p rustfs -p rustfs-ecstore`: passed.
 - `./scripts/check_architecture_migration_rules.sh`: passed.
 - `./scripts/check_layer_dependencies.sh`: passed.
@@ -776,16 +783,17 @@ Passed on `307d788da1954ddaeac8a55d5040a7ce7d1a213f`:
 
 Notes:
 
-- This slice keeps the existing table catalog and rebalance method bodies
+- This slice keeps the existing replication method bodies unchanged while
-  unchanged while narrowing the generic storage capabilities they require.
+  narrowing the generic storage capabilities they require.
-- Table catalog storage still depends on object metadata, object writes,
+- Replication storage still depends on object I/O, object metadata/delete
-  listing, and namespace locking; rebalance metadata merge-save only needs
+  operations, bucket walking, and namespace locking; multipart replication
-  object I/O plus namespace locking.
+  helpers only need object reader access to stream source ranges.
 - The slice does not remove the full storage facade or move traits across crate
   boundaries.
 
 ## Handoff Notes
 
-- API-012 cleanup is locally verified and current with `origin/main`.
+- API-012 replication storage cleanup is locally verified and current with
+  `origin/main`.
 - Remaining namespace-lock cleanup can continue by migrating other consumers
   that no longer need the full storage facade.