diff --git a/crates/ecstore/src/bucket/replication/replication_pool.rs b/crates/ecstore/src/bucket/replication/replication_pool.rs index af2c2f575..9e7388033 100644 --- a/crates/ecstore/src/bucket/replication/replication_pool.rs +++ b/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, - get_heal_replicate_object_info, save_resync_status, + ReplicationConfig, ReplicationResyncer, ReplicationStorage, 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::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 { +pub struct ReplicationPool { // Atomic counters for active workers active_workers: Arc, active_lrg_workers: Arc, @@ -254,7 +253,7 @@ pub struct ReplicationPool { resyncer: Arc, } -impl ReplicationPool { +impl ReplicationPool { /// Creates a new replication pool with specified options pub async fn new(opts: ReplicationPoolOpts, stats: Arc, storage: Arc) -> Arc { 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 ReplicationPoolTrait for ReplicationPool { +impl ReplicationPoolTrait for ReplicationPool { fn active_workers(&self) -> i32 { ReplicationPool::::active_workers(self) } @@ -1382,7 +1381,7 @@ lazy_static! { } /// Initializes background replication with the given options -pub async fn init_background_replication(storage: Arc) { +pub async fn init_background_replication(storage: Arc) { 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> { GLOBAL_REPLICATION_POOL.get().cloned() } -pub async fn schedule_replication( +pub async fn schedule_replication( oi: ObjectInfo, o: Arc, dsc: ReplicateDecision, diff --git a/crates/ecstore/src/bucket/replication/replication_resyncer.rs b/crates/ecstore/src/bucket/replication/replication_resyncer.rs index 692dc8127..fabfd36a1 100644 --- a/crates/ecstore/src/bucket/replication/replication_resyncer.rs +++ b/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 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( + pub async fn resync_bucket( self: Arc, cancellation_token: CancellationToken, storage: Arc, @@ -1734,7 +1740,7 @@ pub async fn must_replicate(bucket: &str, object: &str, mopts: MustReplicateOpti dsc } -pub async fn replicate_delete(dobj: DeletedObjectReplicationInfo, storage: Arc) { +pub async fn replicate_delete(dobj: DeletedObjectReplicationInfo, storage: Arc) { if dobj.delete_object.force_delete { replicate_force_delete_to_targets(&dobj, storage).await; return; @@ -2177,7 +2183,7 @@ pub async fn replicate_delete(dobj: DeletedObj } } -async fn source_delete_marker_missing( +async fn source_delete_marker_missing( 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( - dobj: &DeletedObjectReplicationInfo, - storage: Arc, -) { +async fn replicate_force_delete_to_targets(dobj: &DeletedObjectReplicationInfo, storage: Arc) { 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(roi: ReplicateObjectInfo, storage: Arc) { +pub async fn replicate_object(roi: ReplicateObjectInfo, storage: Arc) { let bucket = roi.bucket.clone(); let object = roi.name.clone(); @@ -2872,13 +2875,13 @@ pub async fn replicate_object(roi: ReplicateOb } 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(); @@ -3173,7 +3176,7 @@ 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(); @@ -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, cli: Arc, src_bucket: &'a str, @@ -3907,7 +3910,7 @@ struct MultipartReplicationContext<'a, S: StorageAPI> { 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/docs/architecture/migration-progress.md b/docs/architecture/migration-progress.md index 49a38c8d1..321c10336 100644 --- a/docs/architecture/migration-progress.md +++ b/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` -- Baseline: `origin/main` at `307d788da1954ddaeac8a55d5040a7ce7d1a213f` +- Branch: `overtrue/arch-storage-namespace-lock-large-cleanup` +- 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 - metadata merge-save helper from full `StorageAPI` bounds to their actual +- Rust code changes: narrow replication pool, resync, delete replication, and + 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 - metadata helper consumers away from full `StorageAPI` where they only need - object I/O, object operations, list operations, and namespace locking. + - Completed cleanup slice: `rustfs/rustfs#3477` narrowed remaining table + catalog backend and rebalance metadata helper consumers away from full + `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. | -| Migration preservation | passed | Lock acquisition, table catalog object paths, optimistic preconditions, pagination, missing-object handling, and rebalance metadata save 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. | +| 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 | 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 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 test -p rustfs-ecstore rebalance --no-fail-fast`: passed. +- `cargo check -p rustfs-ecstore`: 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 - unchanged while narrowing the generic storage capabilities they require. -- Table catalog storage still depends on object metadata, object writes, - listing, and namespace locking; rebalance metadata merge-save only needs - object I/O plus namespace locking. +- This slice keeps the existing replication method bodies unchanged while + narrowing the generic storage capabilities they require. +- Replication storage still depends on object I/O, object metadata/delete + operations, bucket walking, and namespace locking; multipart replication + 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.