diff --git a/crates/ecstore/src/bucket/replication/README.md b/crates/ecstore/src/bucket/replication/README.md index ec29085cb..3a8781f52 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, bandwidth reader wrapping, and SetDisks lock timing. | +| `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_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. | @@ -22,7 +22,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. | -| `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. | +| `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. | | `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. | diff --git a/crates/ecstore/src/bucket/replication/replication_resyncer.rs b/crates/ecstore/src/bucket/replication/replication_resyncer.rs index 4c2f3c7d3..2190cd20b 100644 --- a/crates/ecstore/src/bucket/replication/replication_resyncer.rs +++ b/crates/ecstore/src/bucket/replication/replication_resyncer.rs @@ -19,12 +19,12 @@ 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_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::bucket_target_sys::{ +use super::replication_target_boundary; +use super::replication_target_boundary::{ AdvancedPutOptions, PutObjectOptions, PutObjectPartOptions, RemoveObjectOptions, TargetClient, }; +use super::replication_versioning_boundary as versioning_boundary; +use super::runtime_boundary as runtime_sources; use crate::bucket::replication::ResyncStatusType; use crate::bucket::replication::{ObjectOpts, ReplicationConfigurationExt as _}; use crate::bucket::target::BucketTargets; @@ -875,7 +875,7 @@ impl ReplicationResyncer { } }; - let targets = match target_boundary::list_bucket_targets(&opts.bucket).await { + let targets = match replication_target_boundary::list_bucket_targets(&opts.bucket).await { Ok(targets) => targets, Err(err) => { debug!( @@ -920,7 +920,7 @@ impl ReplicationResyncer { return; } - let Some(target_client) = target_boundary::remote_target_client(&opts.bucket, &target_arns[0]).await else { + let Some(target_client) = replication_target_boundary::remote_target_client(&opts.bucket, &target_arns[0]).await else { error!( event = EVENT_RESYNC_RUNTIME_SKIPPED, component = LOG_COMPONENT_ECSTORE, @@ -1675,7 +1675,7 @@ pub async fn check_replicate_delete( continue; } - let tgt = target_boundary::remote_target_client(bucket, &tgt_arn).await; + let tgt = replication_target_boundary::remote_target_client(bucket, &tgt_arn).await; // The target online status should not be used here while deciding // whether to replicate deletes as the target could be temporarily down let tgt_dsc = if let Some(tgt) = tgt { @@ -1800,7 +1800,7 @@ pub async fn must_replicate(bucket: &str, object: &str, mopts: MustReplicateOpti let mut dsc = ReplicateDecision::default(); for arn in arns { - let cli = target_boundary::remote_target_client(bucket, &arn).await; + let cli = replication_target_boundary::remote_target_client(bucket, &arn).await; let mut sopts = opts.clone(); sopts.target_arn = arn.clone(); @@ -2067,7 +2067,7 @@ pub async fn replicate_delete(dobj: DeletedObjectReplicat } // Get the remote target client - let Some(tgt_client) = target_boundary::remote_target_client(&bucket, &tgt_entry.arn).await else { + let Some(tgt_client) = replication_target_boundary::remote_target_client(&bucket, &tgt_entry.arn).await else { debug!( event = EVENT_REPLICATION_DELETE_SKIPPED, component = LOG_COMPONENT_ECSTORE, @@ -2293,7 +2293,7 @@ async fn replicate_delete_marker_purge_to_targets(bucket: &str, dobj: &DeletedOb if !dobj.target_arn.is_empty() && dobj.target_arn != tgt_entry.arn { continue; } - let Some(tgt_client) = target_boundary::remote_target_client(bucket, &tgt_entry.arn).await else { + let Some(tgt_client) = replication_target_boundary::remote_target_client(bucket, &tgt_entry.arn).await else { continue; }; @@ -2444,7 +2444,7 @@ async fn replicate_force_delete_to_targets(dobj: &Deleted let mut join_set = JoinSet::new(); for arn in tgt_arns { - let Some(tgt_client) = target_boundary::remote_target_client(bucket, &arn).await else { + let Some(tgt_client) = replication_target_boundary::remote_target_client(bucket, &arn).await else { debug!( event = EVENT_REPLICATION_FORCE_DELETE_SKIPPED, component = LOG_COMPONENT_ECSTORE, @@ -2473,7 +2473,7 @@ async fn replicate_force_delete_to_targets(dobj: &Deleted let object_name = object_name.clone(); join_set.spawn(async move { - if target_boundary::target_is_offline(&tgt_client).await { + if replication_target_boundary::target_is_offline(&tgt_client).await { error!( event = EVENT_REPLICATION_FORCE_DELETE_SKIPPED, component = LOG_COMPONENT_ECSTORE, @@ -2601,7 +2601,7 @@ async fn replicate_delete_to_target(dobj: &DeletedObjectReplicationInfo, tgt_cli return rinfo; } - if target_boundary::target_is_offline(&tgt_client).await { + if replication_target_boundary::target_is_offline(&tgt_client).await { if !is_version_purge { rinfo.replication_status = ReplicationStatusType::Failed; } else { @@ -2830,7 +2830,7 @@ pub async fn replicate_object(roi: ReplicateObjectInfo, s let mut join_set = JoinSet::new(); for arn in tgt_arns { - let Some(tgt_client) = target_boundary::remote_target_client(&bucket, &arn).await else { + let Some(tgt_client) = replication_target_boundary::remote_target_client(&bucket, &arn).await else { debug!( event = EVENT_RESYNC_RUNTIME_SKIPPED, component = LOG_COMPONENT_ECSTORE, @@ -2988,7 +2988,7 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { return rinfo; } - if target_boundary::target_is_offline(&tgt_client).await { + if replication_target_boundary::target_is_offline(&tgt_client).await { debug!( event = EVENT_RESYNC_RUNTIME_SKIPPED, component = LOG_COMPONENT_ECSTORE, @@ -3275,7 +3275,7 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { ..Default::default() }; - if target_boundary::target_is_offline(&tgt_client).await { + if replication_target_boundary::target_is_offline(&tgt_client).await { debug!( event = EVENT_RESYNC_RUNTIME_SKIPPED, component = LOG_COMPONENT_ECSTORE, diff --git a/crates/ecstore/src/bucket/replication/replication_target_boundary.rs b/crates/ecstore/src/bucket/replication/replication_target_boundary.rs index 20906a60b..ecf70821d 100644 --- a/crates/ecstore/src/bucket/replication/replication_target_boundary.rs +++ b/crates/ecstore/src/bucket/replication/replication_target_boundary.rs @@ -14,9 +14,13 @@ use std::sync::Arc; -use crate::bucket::bucket_target_sys::{BucketTargetError, BucketTargetSys, TargetClient}; +use crate::bucket::bucket_target_sys::{BucketTargetError, BucketTargetSys}; use crate::bucket::target::BucketTargets; +pub(crate) use crate::bucket::bucket_target_sys::{ + AdvancedPutOptions, PutObjectOptions, PutObjectPartOptions, RemoveObjectOptions, TargetClient, +}; + pub(crate) async fn list_bucket_targets(bucket: &str) -> Result { BucketTargetSys::get().list_bucket_targets(bucket).await } diff --git a/docs/architecture/ecstore-module-split-plan.md b/docs/architecture/ecstore-module-split-plan.md index c05f4a5e1..01b98ea43 100644 --- a/docs/architecture/ecstore-module-split-plan.md +++ b/docs/architecture/ecstore-module-split-plan.md @@ -106,8 +106,9 @@ Current coupling: 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 and bucket target systems - directly; +- resync and delete replication paths call metadata directly, while bucket + target system access and target operation types are concentrated behind the + replication target boundary; - lifecycle and heal paths schedule replication work through the current ECStore module; - global replication pool/stat initialization still lives with ECStore runtime @@ -118,8 +119,10 @@ Required contracts before crate movement: - `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. -- `ReplicationMetadataStore`: replication config, bucket targets, target reset - headers, MRF/resync state, and status persistence. +- `ReplicationMetadataStore`: replication config, target reset headers, + MRF/resync state, and status persistence. +- `ReplicationTargetStore`: bucket target listing, target client lookup, + target offline checks, and target operation option types. - `ReplicationRuntime`: pool, stats, worker admission, bucket monitor, local node identity, cancellation, and queue sizing. - `ReplicationBandwidthLimiter`: target reader wrapping for replication diff --git a/scripts/check_architecture_migration_rules.sh b/scripts/check_architecture_migration_rules.sh index 44c8e0227..b5ebcafff 100755 --- a/scripts/check_architecture_migration_rules.sh +++ b/scripts/check_architecture_migration_rules.sh @@ -2590,14 +2590,21 @@ fi ( cd "$ROOT_DIR" - rg -n --with-filename 'BucketTargetSys::get\(\)' \ - crates/ecstore/src/bucket/replication \ - --glob '*.rs' | + find crates/ecstore/src/bucket/replication -type f -name '*.rs' -print0 | + xargs -0 perl -0ne ' + while (/(?:(?:crate::bucket|super(?:::super)+)(?:::\{[^;]*\bbucket_target_sys\b|::bucket_target_sys\b)|BucketTargetSys::get\(\))/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_target_boundary\.rs:' || true ) >"$REPLICATION_TARGET_BOUNDARY_BYPASS_HITS_FILE" if [[ -s "$REPLICATION_TARGET_BOUNDARY_BYPASS_HITS_FILE" ]]; then - report_failure "replication bucket-target access must stay behind replication target boundary: $(paste -sd '; ' "$REPLICATION_TARGET_BOUNDARY_BYPASS_HITS_FILE")" + report_failure "replication bucket-target access and types must stay behind replication target boundary: $(paste -sd '; ' "$REPLICATION_TARGET_BOUNDARY_BYPASS_HITS_FILE")" fi (