diff --git a/crates/ecstore/src/bucket/replication/README.md b/crates/ecstore/src/bucket/replication/README.md index 312a02925..239a3a13b 100644 --- a/crates/ecstore/src/bucket/replication/README.md +++ b/crates/ecstore/src/bucket/replication/README.md @@ -23,15 +23,18 @@ and lifecycle/heal scheduling 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. | Metadata sys access and replication metadata path constants are concentrated in `replication_metadata_boundary.rs`; versioning sys and config storage imports remain. | -| `ReplicationConfigStore` | Replication config persistence and config-derived labels used by target options. | Config read/save helpers and storage class labels are concentrated in `replication_config_store.rs`. | +| `ReplicationMetadataStore` | Replication config, MRF/resync state, target reset headers, and status persistence. | Metadata sys access and replication metadata path constants are exposed through the contract type in `replication_metadata_boundary.rs`; versioning sys and config storage imports remain separate contracts. | +| `ReplicationConfigStore` | Replication config persistence and config-derived labels used by target options. | Config read/save helpers and storage class labels are exposed through the contract type in `replication_config_store.rs`. | | `ReplicationFileMeta` | Replication status, decisions, MRF entries, resync decisions, and target reset helpers. | `rustfs_filemeta` replication contracts are concentrated in `replication_filemeta_boundary.rs`; `FileInfo` remains in the storage boundary for storage trait bindings and walk options. | | `ReplicationErrorBoundary` | ECStore error/result contracts and replication-specific error classifiers. | `crate::error` imports are concentrated in `replication_error_boundary.rs`. | -| `ReplicationTargetStore` | Bucket target listing, target client lookup, target offline checks, target config types, and target operation option types. | Bucket target sys access, `BucketTargets`, and target operation types are concentrated in `replication_target_boundary.rs`. | +| `ReplicationTargetStore` | Bucket target listing, target client lookup, target offline checks, target config types, and target operation option types. | Bucket target sys access, `BucketTargets`, and target operation types are exposed through the contract type 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; ECStore object store and bucket monitor implementation types stay behind local storage/bandwidth boundaries. | | `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. | Event notification service calls and local event host selection are concentrated in `replication_event_sink.rs`. | -| `ReplicationTagFilter` | Decode object tag strings for rule and metadata replication decisions. | Direct bucket tagging helper imports from replication workers. | +| `ReplicationVersioningStore` | Versioning state checks for object and delete replication decisions. | Bucket versioning sys access is exposed through the contract type in `replication_versioning_boundary.rs`. | +| `ReplicationLockTiming` | Namespace lock timing for replication resync, object replication, and delete replication locks. | SetDisks lock timeout access is exposed through the contract type in `replication_lock_boundary.rs`. | +| `ReplicationMsgpCodec` | MessagePack time encode/decode and unknown value skipping for persisted resync/MRF state. | Bucket MessagePack helpers are exposed through the contract type in `replication_msgp_boundary.rs`. | +| `ReplicationTagFilter` | Decode object tag strings for rule and metadata replication decisions. | Bucket tagging helper access is exposed through the contract type in `replication_tagging_boundary.rs`. | | `ReplicationLifecycleBridge` | Lifecycle-originated delete and version-purge scheduling. | Direct coupling between lifecycle worker paths and replication delete scheduling. | ## Migration Rules @@ -50,8 +53,10 @@ and lifecycle/heal scheduling paths. replication module, not `crate::bucket::replication::*` self paths. 6. Keep runtime source access from importing ECStore object store or bucket monitor implementation types directly; use local boundary-owned aliases. -7. Move at most one contract boundary per code-bearing PR and verify it with - focused replication tests before broad gates. +7. Move at most one owner boundary per code-bearing PR and verify it with + focused replication tests before broad gates. Non-behavioral contract-shape + cleanup may batch already-established boundary wrappers when the owner and + call semantics do not change. ## First Code-Bearing Step diff --git a/crates/ecstore/src/bucket/replication/config.rs b/crates/ecstore/src/bucket/replication/config.rs index d7cf40f92..807a25141 100644 --- a/crates/ecstore/src/bucket/replication/config.rs +++ b/crates/ecstore/src/bucket/replication/config.rs @@ -13,7 +13,7 @@ // limitations under the License. use super::replication_filemeta_boundary::ReplicationType; -use super::replication_tagging_boundary as tagging_boundary; +use super::replication_tagging_boundary::ReplicationTagFilter; use super::rule::ReplicationRuleExt as _; use s3s::dto::DeleteMarkerReplicationStatus; use s3s::dto::DeleteReplicationStatus; @@ -102,7 +102,7 @@ impl ReplicationConfigurationExt for ReplicationConfiguration { } if let Some(filter) = &rule.filter { - let object_tags = tagging_boundary::decode_tags_to_map(&obj.user_tags); + let object_tags = ReplicationTagFilter::decode_tags_to_map(&obj.user_tags); if filter.test_tags(&object_tags) { rules.push(rule.clone()); } diff --git a/crates/ecstore/src/bucket/replication/replication_config_store.rs b/crates/ecstore/src/bucket/replication/replication_config_store.rs index b0e163425..459f2e4b4 100644 --- a/crates/ecstore/src/bucket/replication/replication_config_store.rs +++ b/crates/ecstore/src/bucket/replication/replication_config_store.rs @@ -14,20 +14,26 @@ use super::replication_error_boundary::Result; use super::replication_storage_boundary::ReplicationObjectIO; -use crate::config::com; -pub(crate) use crate::config::storageclass::{RRS, STANDARD}; +use crate::config::{com, storageclass}; use std::sync::Arc; -pub(crate) async fn read(api: Arc, file: &str) -> Result> -where - S: ReplicationObjectIO, -{ - com::read_config(api, file).await -} +pub(crate) struct ReplicationConfigStore; -pub(crate) async fn save(api: Arc, file: &str, data: Vec) -> Result<()> -where - S: ReplicationObjectIO, -{ - com::save_config(api, file, data).await +impl ReplicationConfigStore { + pub(crate) const RRS: &'static str = storageclass::RRS; + pub(crate) const STANDARD: &'static str = storageclass::STANDARD; + + pub(crate) async fn read(api: Arc, file: &str) -> Result> + where + S: ReplicationObjectIO, + { + com::read_config(api, file).await + } + + pub(crate) async fn save(api: Arc, file: &str, data: Vec) -> Result<()> + where + S: ReplicationObjectIO, + { + com::save_config(api, file, data).await + } } diff --git a/crates/ecstore/src/bucket/replication/replication_lock_boundary.rs b/crates/ecstore/src/bucket/replication/replication_lock_boundary.rs index c78a15e52..903811225 100644 --- a/crates/ecstore/src/bucket/replication/replication_lock_boundary.rs +++ b/crates/ecstore/src/bucket/replication/replication_lock_boundary.rs @@ -14,6 +14,10 @@ use std::time::Duration; -pub(crate) fn acquire_timeout() -> Duration { - crate::set_disk::get_lock_acquire_timeout() +pub(crate) struct ReplicationLockTiming; + +impl ReplicationLockTiming { + pub(crate) fn acquire_timeout() -> Duration { + crate::set_disk::get_lock_acquire_timeout() + } } diff --git a/crates/ecstore/src/bucket/replication/replication_metadata_boundary.rs b/crates/ecstore/src/bucket/replication/replication_metadata_boundary.rs index 6d46db0c3..192082133 100644 --- a/crates/ecstore/src/bucket/replication/replication_metadata_boundary.rs +++ b/crates/ecstore/src/bucket/replication/replication_metadata_boundary.rs @@ -20,42 +20,47 @@ use time::OffsetDateTime; use super::replication_error_boundary::{Error, Result}; -pub(crate) const REPLICATION_DIR: &str = ".replication"; -pub(crate) const RESYNC_FILE_NAME: &str = "resync.bin"; -pub(crate) const MRF_REPLICATION_FILE: &str = "config/replication/mrf.bin"; +const REPLICATION_DIR: &str = ".replication"; +const RESYNC_FILE_NAME: &str = "resync.bin"; -pub(crate) async fn replication_config(bucket: &str) -> Result<(ReplicationConfiguration, OffsetDateTime)> { - metadata_sys::get_replication_config(bucket).await -} +pub(crate) struct ReplicationMetadataStore; -pub(crate) async fn optional_replication_config(bucket: &str) -> Result> { - let config = match replication_config(bucket).await { - Ok((config, _)) => Some(config), - Err(err) => { - if err != Error::ConfigNotFound { - return Err(err); +impl ReplicationMetadataStore { + pub(crate) const MRF_REPLICATION_FILE: &'static str = "config/replication/mrf.bin"; + + pub(crate) async fn replication_config(bucket: &str) -> Result<(ReplicationConfiguration, OffsetDateTime)> { + metadata_sys::get_replication_config(bucket).await + } + + pub(crate) async fn optional_replication_config(bucket: &str) -> Result> { + let config = match Self::replication_config(bucket).await { + Ok((config, _)) => Some(config), + Err(err) => { + if err != Error::ConfigNotFound { + return Err(err); + } + None } - None - } - }; - Ok(config) -} + }; + Ok(config) + } -pub(crate) fn rustfs_meta_bucket() -> &'static str { - RUSTFS_META_BUCKET -} + pub(crate) fn rustfs_meta_bucket() -> &'static str { + RUSTFS_META_BUCKET + } -pub(crate) fn resync_lock_key(bucket: &str, arn: &str) -> String { - format!("{REPLICATION_DIR}/{bucket}/{arn}") -} + pub(crate) fn resync_lock_key(bucket: &str, arn: &str) -> String { + format!("{REPLICATION_DIR}/{bucket}/{arn}") + } -pub(crate) fn bucket_resync_dir_path(bucket: &str) -> String { - path_join_buf(&[BUCKET_META_PREFIX, bucket, REPLICATION_DIR]) -} + pub(crate) fn bucket_resync_dir_path(bucket: &str) -> String { + path_join_buf(&[BUCKET_META_PREFIX, bucket, REPLICATION_DIR]) + } -pub(crate) fn bucket_resync_file_path(bucket: &str) -> String { - let resync_dir_path = bucket_resync_dir_path(bucket); - path_join_buf(&[&resync_dir_path, RESYNC_FILE_NAME]) + pub(crate) fn bucket_resync_file_path(bucket: &str) -> String { + let resync_dir_path = Self::bucket_resync_dir_path(bucket); + path_join_buf(&[&resync_dir_path, RESYNC_FILE_NAME]) + } } #[cfg(test)] @@ -64,9 +69,18 @@ mod tests { #[test] fn replication_metadata_paths_match_existing_layout() { - assert_eq!(resync_lock_key("bucket-a", "arn-a"), ".replication/bucket-a/arn-a"); - assert_eq!(bucket_resync_dir_path("bucket-a"), "buckets/bucket-a/.replication"); - assert_eq!(bucket_resync_file_path("bucket-a"), "buckets/bucket-a/.replication/resync.bin"); - assert_eq!(MRF_REPLICATION_FILE, "config/replication/mrf.bin"); + assert_eq!( + ReplicationMetadataStore::resync_lock_key("bucket-a", "arn-a"), + ".replication/bucket-a/arn-a" + ); + assert_eq!( + ReplicationMetadataStore::bucket_resync_dir_path("bucket-a"), + "buckets/bucket-a/.replication" + ); + assert_eq!( + ReplicationMetadataStore::bucket_resync_file_path("bucket-a"), + "buckets/bucket-a/.replication/resync.bin" + ); + assert_eq!(ReplicationMetadataStore::MRF_REPLICATION_FILE, "config/replication/mrf.bin"); } } diff --git a/crates/ecstore/src/bucket/replication/replication_msgp_boundary.rs b/crates/ecstore/src/bucket/replication/replication_msgp_boundary.rs index 9d9362843..48f0685b9 100644 --- a/crates/ecstore/src/bucket/replication/replication_msgp_boundary.rs +++ b/crates/ecstore/src/bucket/replication/replication_msgp_boundary.rs @@ -12,4 +12,25 @@ // See the License for the specific language governing permissions and // limitations under the License. -pub(crate) use crate::bucket::msgp_decode::{read_msgp_ext8_time, skip_msgp_value, write_msgp_time}; +use std::io::{Read, Write}; + +use time::OffsetDateTime; + +use super::replication_error_boundary::Result; +use crate::bucket::msgp_decode; + +pub(crate) struct ReplicationMsgpCodec; + +impl ReplicationMsgpCodec { + pub(crate) fn read_ext8_time(rd: &mut R) -> Result { + msgp_decode::read_msgp_ext8_time(rd) + } + + pub(crate) fn skip_value(rd: &mut R) -> Result<()> { + msgp_decode::skip_msgp_value(rd) + } + + pub(crate) fn write_time(wr: &mut W, time: OffsetDateTime) -> Result<()> { + msgp_decode::write_msgp_time(wr, time) + } +} diff --git a/crates/ecstore/src/bucket/replication/replication_pool.rs b/crates/ecstore/src/bucket/replication/replication_pool.rs index 15e72a8d7..fd30bc223 100644 --- a/crates/ecstore/src/bucket/replication/replication_pool.rs +++ b/crates/ecstore/src/bucket/replication/replication_pool.rs @@ -13,14 +13,14 @@ // limitations under the License. use super::datatypes::ResyncStatusType; -use super::replication_config_store as config_store; +use super::replication_config_store::ReplicationConfigStore; use super::replication_error_boundary::Error as EcstoreError; use super::replication_filemeta_boundary::{ MrfOpKind, MrfReplicateEntry, REPLICATE_EXISTING, REPLICATE_HEAL, REPLICATE_HEAL_DELETE, ReplicateDecision, ReplicateObjectInfo, ReplicatedTargetInfo, ReplicationStatusType, ReplicationType, ReplicationWorkerOperation, ResyncDecision, VersionPurgeStatusType, replication_statuses_map, version_purge_statuses_map, }; -use super::replication_metadata_boundary as metadata_boundary; +use super::replication_metadata_boundary::ReplicationMetadataStore; use super::replication_resyncer::{ BucketReplicationResyncStatus, DeletedObjectReplicationInfo, ReplicationConfig, ReplicationResyncer, ResyncOpts, TargetReplicationResyncStatus, decode_mrf_file, decode_resync_file, encode_mrf_file, get_heal_replicate_object_info, @@ -28,7 +28,7 @@ use super::replication_resyncer::{ }; use super::replication_state::ReplicationStats; use super::replication_storage_boundary::{DeletedObject, ObjectInfo, ObjectOptions, ReplicationObjectIO, ReplicationStorage}; -use super::replication_target_boundary as target_boundary; +use super::replication_target_boundary::ReplicationTargetStore; use super::runtime_boundary as runtime_sources; use lazy_static::lazy_static; use rustfs_utils::http::{SUFFIX_REPLICATION_TIMESTAMP, get_str}; @@ -788,7 +788,7 @@ impl ReplicationPool { let storage = self.storage.clone(); let handle = tokio::spawn(async move { - let data = match config_store::read(storage.clone(), metadata_boundary::MRF_REPLICATION_FILE).await { + let data = match ReplicationConfigStore::read(storage.clone(), ReplicationMetadataStore::MRF_REPLICATION_FILE).await { Ok(d) => d, Err(EcstoreError::ConfigNotFound) => return, // no file yet — normal on first start Err(e) => { @@ -812,9 +812,9 @@ impl ReplicationPool { "Failed to decode MRF recovery file — discarding corrupt data" ); // Overwrite the corrupt file so we don't fail again on next restart. - let _ = config_store::save( + let _ = ReplicationConfigStore::save( storage, - metadata_boundary::MRF_REPLICATION_FILE, + ReplicationMetadataStore::MRF_REPLICATION_FILE, encode_mrf_file(&[]).unwrap_or_default(), ) .await; @@ -876,9 +876,12 @@ impl ReplicationPool { // Clear AFTER all entries are processed so a crash mid-replay causes at-most-twice // delivery (idempotent) rather than entry loss. - if let Err(e) = - config_store::save(storage, metadata_boundary::MRF_REPLICATION_FILE, encode_mrf_file(&[]).unwrap_or_default()) - .await + if let Err(e) = ReplicationConfigStore::save( + storage, + ReplicationMetadataStore::MRF_REPLICATION_FILE, + encode_mrf_file(&[]).unwrap_or_default(), + ) + .await { warn!( component = LOG_COMPONENT_ECSTORE, @@ -1251,7 +1254,9 @@ impl ReplicationPool { 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(), metadata_boundary::MRF_REPLICATION_FILE, data).await { + if let Err(e) = + ReplicationConfigStore::save(storage.clone(), ReplicationMetadataStore::MRF_REPLICATION_FILE, data).await + { warn!( component = LOG_COMPONENT_ECSTORE, subsystem = LOG_SUBSYSTEM_REPLICATION, @@ -1283,9 +1288,9 @@ async fn load_bucket_resync_metadata( ) -> Result { let mut brs = BucketReplicationResyncStatus::new(); - let resync_file_path = metadata_boundary::bucket_resync_file_path(bucket); + let resync_file_path = ReplicationMetadataStore::bucket_resync_file_path(bucket); - let data = match config_store::read(obj_api, &resync_file_path).await { + let data = match ReplicationConfigStore::read(obj_api, &resync_file_path).await { Ok(data) => data, Err(EcstoreError::ConfigNotFound) => return Ok(brs), Err(err) => return Err(err), @@ -1484,7 +1489,7 @@ pub async fn queue_replication_heal(bucket: &str, oi: ObjectInfo, retry_count: u return; } - let rcfg = match metadata_boundary::replication_config(bucket).await { + let rcfg = match ReplicationMetadataStore::replication_config(bucket).await { Ok((config, _)) => config, Err(err) => { debug!( @@ -1501,7 +1506,7 @@ pub async fn queue_replication_heal(bucket: &str, oi: ObjectInfo, retry_count: u } }; - let tgts = match target_boundary::list_bucket_targets(bucket).await { + let tgts = match ReplicationTargetStore::list_bucket_targets(bucket).await { Ok(targets) => Some(targets), Err(err) => { debug!( diff --git a/crates/ecstore/src/bucket/replication/replication_resyncer.rs b/crates/ecstore/src/bucket/replication/replication_resyncer.rs index 00cc6001d..2b7090072 100644 --- a/crates/ecstore/src/bucket/replication/replication_resyncer.rs +++ b/crates/ecstore/src/bucket/replication/replication_resyncer.rs @@ -15,7 +15,7 @@ use super::config::{ObjectOpts, ReplicationConfigurationExt as _}; use super::datatypes::ResyncStatusType; use super::replication_bandwidth_boundary; -use super::replication_config_store as config_store; +use super::replication_config_store::ReplicationConfigStore; use super::replication_error_boundary::{Error, Result, is_err_object_not_found, is_err_version_not_found}; use super::replication_event_sink::{EventArgs, send_event, send_local_event}; use super::replication_filemeta_boundary::{ @@ -24,19 +24,19 @@ use super::replication_filemeta_boundary::{ ReplicationType, ReplicationWorkerOperation, ResyncDecision, ResyncTargetDecision, VersionPurgeStatusType, get_replication_state, parse_replicate_decision, replication_statuses_map, target_reset_header, version_purge_statuses_map, }; -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_lock_boundary::ReplicationLockTiming; +use super::replication_metadata_boundary::ReplicationMetadataStore; +use super::replication_msgp_boundary::ReplicationMsgpCodec; 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_tagging_boundary::ReplicationTagFilter; use super::replication_target_boundary::{ - AdvancedPutOptions, BucketTargets, PutObjectOptions, PutObjectPartOptions, RemoveObjectOptions, TargetClient, + AdvancedPutOptions, BucketTargets, PutObjectOptions, PutObjectPartOptions, RemoveObjectOptions, ReplicationTargetStore, + TargetClient, }; -use super::replication_versioning_boundary as versioning_boundary; +use super::replication_versioning_boundary::ReplicationVersioningStore; use super::runtime_boundary as runtime_sources; use aws_sdk_s3::error::{ProvideErrorMetadata, SdkError}; use aws_sdk_s3::operation::head_object::{HeadObjectError, HeadObjectOutput}; @@ -267,7 +267,7 @@ impl BucketReplicationResyncStatus { rmp::encode::write_str(&mut wr, "id")?; rmp::encode::write_i32(&mut wr, self.id)?; rmp::encode::write_str(&mut wr, "lu")?; - write_msgp_time(&mut wr, wire_time_or_default(self.last_update))?; + ReplicationMsgpCodec::write_time(&mut wr, wire_time_or_default(self.last_update))?; Ok(wr) } @@ -300,7 +300,7 @@ impl BucketReplicationResyncStatus { "lu" => { out.last_update = normalize_wire_time(read_msgp_time_or_nil(&mut rd)?); } - _ => skip_msgp_value(&mut rd)?, + _ => ReplicationMsgpCodec::skip_value(&mut rd)?, } } Ok(out) @@ -385,13 +385,13 @@ impl TargetReplicationResyncStatus { fn marshal_wire_msg(&self, wr: &mut Vec) -> Result<()> { rmp::encode::write_map_len(wr, 11)?; rmp::encode::write_str(wr, "st")?; - write_msgp_time(wr, wire_time_or_default(self.start_time))?; + ReplicationMsgpCodec::write_time(wr, wire_time_or_default(self.start_time))?; rmp::encode::write_str(wr, "lst")?; - write_msgp_time(wr, wire_time_or_default(self.last_update))?; + ReplicationMsgpCodec::write_time(wr, wire_time_or_default(self.last_update))?; rmp::encode::write_str(wr, "id")?; rmp::encode::write_str(wr, &self.resync_id)?; rmp::encode::write_str(wr, "rdt")?; - write_msgp_time(wr, wire_time_or_default(self.resync_before_date))?; + ReplicationMsgpCodec::write_time(wr, wire_time_or_default(self.resync_before_date))?; rmp::encode::write_str(wr, "rst")?; rmp::encode::write_i32(wr, resync_status_to_i32(self.resync_status))?; rmp::encode::write_str(wr, "fs")?; @@ -431,7 +431,7 @@ impl TargetReplicationResyncStatus { "rrc" => out.replicated_count = rmp::decode::read_int(rd)?, "bkt" => out.bucket = read_msgp_str(rd)?, "obj" => out.object = read_msgp_str(rd)?, - _ => skip_msgp_value(rd)?, + _ => ReplicationMsgpCodec::skip_value(rd)?, } } Ok(out) @@ -449,7 +449,7 @@ fn read_msgp_time_or_nil(rd: &mut R) -> Result> let marker = rmp::decode::read_marker(rd).map_err(|e| Error::other(format!("{e:?}")))?; match marker { rmp::Marker::Null => Ok(None), - rmp::Marker::Ext8 => Ok(Some(read_msgp_ext8_time(rd)?)), + rmp::Marker::Ext8 => Ok(Some(ReplicationMsgpCodec::read_ext8_time(rd)?)), other => Err(Error::other(format!("expected time ext or nil, got marker: {other:?}"))), } } @@ -737,9 +737,9 @@ impl ReplicationResyncer { // Acquire a cluster-wide leader lock for this (bucket, ARN) pair so that only // one node runs the resync scan at a time. Without this, every cluster node would // scan and replicate every object independently, causing N-fold duplicate traffic. - let resync_lock_key = metadata_boundary::resync_lock_key(&opts.bucket, &opts.arn); + let resync_lock_key = ReplicationMetadataStore::resync_lock_key(&opts.bucket, &opts.arn); let resync_ns_lock = match storage - .new_ns_lock(metadata_boundary::rustfs_meta_bucket(), &resync_lock_key) + .new_ns_lock(ReplicationMetadataStore::rustfs_meta_bucket(), &resync_lock_key) .await { Ok(l) => l, @@ -757,7 +757,7 @@ impl ReplicationResyncer { return; } }; - let _resync_leader_guard = match resync_ns_lock.get_write_lock(lock_boundary::acquire_timeout()).await { + let _resync_leader_guard = match resync_ns_lock.get_write_lock(ReplicationLockTiming::acquire_timeout()).await { Ok(g) => g, Err(_) => { debug!( @@ -792,7 +792,7 @@ impl ReplicationResyncer { } }; - let targets = match replication_target_boundary::list_bucket_targets(&opts.bucket).await { + let targets = match ReplicationTargetStore::list_bucket_targets(&opts.bucket).await { Ok(targets) => targets, Err(err) => { debug!( @@ -837,7 +837,7 @@ impl ReplicationResyncer { return; } - let Some(target_client) = replication_target_boundary::remote_target_client(&opts.bucket, &target_arns[0]).await else { + let Some(target_client) = ReplicationTargetStore::remote_target_client(&opts.bucket, &target_arns[0]).await else { error!( event = EVENT_RESYNC_RUNTIME_SKIPPED, component = LOG_COMPONENT_ECSTORE, @@ -1164,8 +1164,8 @@ pub async fn get_heal_replicate_object_info(oi: &ObjectInfo, rcfg: &ReplicationC }, &oi, &ObjectOptions { - versioned: versioning_boundary::prefix_enabled(&oi.bucket, &oi.name).await, - version_suspended: versioning_boundary::prefix_suspended(&oi.bucket, &oi.name).await, + versioned: ReplicationVersioningStore::prefix_enabled(&oi.bucket, &oi.name).await, + version_suspended: ReplicationVersioningStore::prefix_suspended(&oi.bucket, &oi.name).await, ..Default::default() }, None, @@ -1228,14 +1228,14 @@ pub(crate) async fn save_resync_status( ) -> Result<()> { let data = encode_resync_file(status)?; - let config_file = metadata_boundary::bucket_resync_file_path(bucket); - config_store::save(api, &config_file, data).await?; + let config_file = ReplicationMetadataStore::bucket_resync_file_path(bucket); + ReplicationConfigStore::save(api, &config_file, data).await?; Ok(()) } async fn get_replication_config(bucket: &str) -> Result> { - metadata_boundary::optional_replication_config(bucket).await + ReplicationMetadataStore::optional_replication_config(bucket).await } #[derive(Debug, Clone, Default)] @@ -1592,7 +1592,7 @@ pub async fn check_replicate_delete( continue; } - let tgt = replication_target_boundary::remote_target_client(bucket, &tgt_arn).await; + let tgt = ReplicationTargetStore::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 { @@ -1673,7 +1673,7 @@ pub async fn must_replicate(bucket: &str, object: &str, mopts: MustReplicateOpti return ReplicateDecision::default(); } - if !versioning_boundary::prefix_enabled(bucket, object).await { + if !ReplicationVersioningStore::prefix_enabled(bucket, object).await { return ReplicateDecision::default(); } @@ -1717,7 +1717,7 @@ pub async fn must_replicate(bucket: &str, object: &str, mopts: MustReplicateOpti let mut dsc = ReplicateDecision::default(); for arn in arns { - let cli = replication_target_boundary::remote_target_client(bucket, &arn).await; + let cli = ReplicationTargetStore::remote_target_client(bucket, &arn).await; let mut sopts = opts.clone(); sopts.target_arn = arn.clone(); @@ -1807,8 +1807,9 @@ pub async fn replicate_delete(dobj: DeletedObjectReplicat &dobj.delete_object.object_name, &ObjectOptions { version_id: Some(delete_marker_version_id.to_string()), - versioned: versioning_boundary::prefix_enabled(&bucket, &dobj.delete_object.object_name).await, - version_suspended: versioning_boundary::prefix_suspended(&bucket, &dobj.delete_object.object_name).await, + versioned: ReplicationVersioningStore::prefix_enabled(&bucket, &dobj.delete_object.object_name).await, + version_suspended: ReplicationVersioningStore::prefix_suspended(&bucket, &dobj.delete_object.object_name) + .await, ..Default::default() }, ) @@ -1928,7 +1929,7 @@ pub async fn replicate_delete(dobj: DeletedObjectReplicat } }; - let _lock_guard = match ns_lock.get_write_lock(lock_boundary::acquire_timeout()).await { + let _lock_guard = match ns_lock.get_write_lock(ReplicationLockTiming::acquire_timeout()).await { Ok(lock_guard) => lock_guard, Err(e) => { debug!( @@ -1979,7 +1980,7 @@ pub async fn replicate_delete(dobj: DeletedObjectReplicat } // Get the remote target client - let Some(tgt_client) = replication_target_boundary::remote_target_client(&bucket, &tgt_entry.arn).await else { + let Some(tgt_client) = ReplicationTargetStore::remote_target_client(&bucket, &tgt_entry.arn).await else { debug!( event = EVENT_REPLICATION_DELETE_SKIPPED, component = LOG_COMPONENT_ECSTORE, @@ -2125,8 +2126,8 @@ pub async fn replicate_delete(dobj: DeletedObjectReplicat version_id: version_id.map(|v| v.to_string()), mod_time: dobj.delete_object.delete_marker_mtime, delete_replication: Some(drs), - versioned: versioning_boundary::prefix_enabled(&bucket, &dobj.delete_object.object_name).await, - version_suspended: versioning_boundary::prefix_suspended(&bucket, &dobj.delete_object.object_name).await, + versioned: ReplicationVersioningStore::prefix_enabled(&bucket, &dobj.delete_object.object_name).await, + version_suspended: ReplicationVersioningStore::prefix_suspended(&bucket, &dobj.delete_object.object_name).await, ..Default::default() }, ) @@ -2180,8 +2181,8 @@ async fn source_delete_marker_missing( object_name, &ObjectOptions { version_id: Some(delete_marker_version_id.to_string()), - versioned: versioning_boundary::prefix_enabled(bucket, object_name).await, - version_suspended: versioning_boundary::prefix_suspended(bucket, object_name).await, + versioned: ReplicationVersioningStore::prefix_enabled(bucket, object_name).await, + version_suspended: ReplicationVersioningStore::prefix_suspended(bucket, object_name).await, ..Default::default() }, ) @@ -2204,7 +2205,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) = replication_target_boundary::remote_target_client(bucket, &tgt_entry.arn).await else { + let Some(tgt_client) = ReplicationTargetStore::remote_target_client(bucket, &tgt_entry.arn).await else { continue; }; @@ -2311,7 +2312,7 @@ async fn replicate_force_delete_to_targets(dobj: &Deleted } }; - let _lock_guard = match ns_lock.get_write_lock(lock_boundary::acquire_timeout()).await { + let _lock_guard = match ns_lock.get_write_lock(ReplicationLockTiming::acquire_timeout()).await { Ok(guard) => guard, Err(e) => { warn!( @@ -2351,7 +2352,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) = replication_target_boundary::remote_target_client(bucket, &arn).await else { + let Some(tgt_client) = ReplicationTargetStore::remote_target_client(bucket, &arn).await else { debug!( event = EVENT_REPLICATION_FORCE_DELETE_SKIPPED, component = LOG_COMPONENT_ECSTORE, @@ -2379,7 +2380,7 @@ async fn replicate_force_delete_to_targets(dobj: &Deleted let object_name = object_name.clone(); join_set.spawn(async move { - if replication_target_boundary::target_is_offline(&tgt_client).await { + if ReplicationTargetStore::target_is_offline(&tgt_client).await { error!( event = EVENT_REPLICATION_FORCE_DELETE_SKIPPED, component = LOG_COMPONENT_ECSTORE, @@ -2505,7 +2506,7 @@ async fn replicate_delete_to_target(dobj: &DeletedObjectReplicationInfo, tgt_cli return rinfo; } - if replication_target_boundary::target_is_offline(&tgt_client).await { + if ReplicationTargetStore::target_is_offline(&tgt_client).await { if !is_version_purge { rinfo.replication_status = ReplicationStatusType::Failed; } else { @@ -2703,7 +2704,7 @@ pub async fn replicate_object(roi: ReplicateObjectInfo, s return; } }; - let _obj_lock_guard = match obj_ns_lock.get_write_lock(lock_boundary::acquire_timeout()).await { + let _obj_lock_guard = match obj_ns_lock.get_write_lock(ReplicationLockTiming::acquire_timeout()).await { Ok(g) => g, Err(e) => { debug!( @@ -2730,7 +2731,7 @@ pub async fn replicate_object(roi: ReplicateObjectInfo, s let mut join_set = JoinSet::new(); for arn in tgt_arns { - let Some(tgt_client) = replication_target_boundary::remote_target_client(&bucket, &arn).await else { + let Some(tgt_client) = ReplicationTargetStore::remote_target_client(&bucket, &arn).await else { debug!( event = EVENT_RESYNC_RUNTIME_SKIPPED, component = LOG_COMPONENT_ECSTORE, @@ -2894,7 +2895,7 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { return rinfo; } - if replication_target_boundary::target_is_offline(&tgt_client).await { + if ReplicationTargetStore::target_is_offline(&tgt_client).await { debug!( event = EVENT_RESYNC_RUNTIME_SKIPPED, component = LOG_COMPONENT_ECSTORE, @@ -2915,8 +2916,8 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { return rinfo; } - let versioned = versioning_boundary::prefix_enabled(&bucket, &object).await; - let version_suspended = versioning_boundary::prefix_suspended(&bucket, &object).await; + let versioned = ReplicationVersioningStore::prefix_enabled(&bucket, &object).await; + let version_suspended = ReplicationVersioningStore::prefix_suspended(&bucket, &object).await; let obj_opts = ObjectOptions { version_id: self.version_id.map(|v| v.to_string()), @@ -3180,7 +3181,7 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { ..Default::default() }; - if replication_target_boundary::target_is_offline(&tgt_client).await { + if ReplicationTargetStore::target_is_offline(&tgt_client).await { debug!( event = EVENT_RESYNC_RUNTIME_SKIPPED, component = LOG_COMPONENT_ECSTORE, @@ -3201,8 +3202,8 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { return rinfo; } - let versioned = versioning_boundary::prefix_enabled(&bucket, &object).await; - let version_suspended = versioning_boundary::prefix_suspended(&bucket, &object).await; + let versioned = ReplicationVersioningStore::prefix_enabled(&bucket, &object).await; + let version_suspended = ReplicationVersioningStore::prefix_suspended(&bucket, &object).await; let obj_opts = ObjectOptions { version_id: self.version_id.map(|v| v.to_string()), @@ -3726,7 +3727,7 @@ fn put_replication_opts(sc: &str, object_info: &ObjectInfo) -> Result<(PutObject // Handle storage class default let storage_class = if sc.is_empty() { let obj_sc = object_info.storage_class.as_deref().unwrap_or_default(); - if obj_sc == config_store::STANDARD || obj_sc == config_store::RRS { + if obj_sc == ReplicationConfigStore::STANDARD || obj_sc == ReplicationConfigStore::RRS { obj_sc.to_string() } else { sc.to_string() @@ -3753,7 +3754,7 @@ fn put_replication_opts(sc: &str, object_info: &ObjectInfo) -> Result<(PutObject }; if !object_info.user_tags.is_empty() { - let tags = tagging_boundary::decode_tags_to_map(&object_info.user_tags); + let tags = ReplicationTagFilter::decode_tags_to_map(&object_info.user_tags); if !tags.is_empty() { put_op.user_tags = tags; @@ -4029,8 +4030,9 @@ fn get_replication_action(oi1: &ObjectInfo, oi2: &HeadObjectOutput, op_type: Rep } } - let oi1_tags = tagging_boundary::decode_tags_to_map(&oi1.user_tags); - let oi2_tags = tagging_boundary::decode_tags_to_map(metadata.get(AMZ_OBJECT_TAGGING).cloned().unwrap_or_default().as_str()); + let oi1_tags = ReplicationTagFilter::decode_tags_to_map(&oi1.user_tags); + let oi2_tags = + ReplicationTagFilter::decode_tags_to_map(metadata.get(AMZ_OBJECT_TAGGING).cloned().unwrap_or_default().as_str()); if (oi2.tag_count.unwrap_or_default() > 0 && oi1_tags != oi2_tags) || oi2.tag_count.unwrap_or_default() != oi1_tags.len() as i32 @@ -4125,13 +4127,13 @@ mod tests { rmp::encode::write_str(&mut payload, "arn:replication::1:dest").expect("write arn"); rmp::encode::write_map_len(&mut payload, 11).expect("write target"); rmp::encode::write_str(&mut payload, "st").expect("write key"); - write_msgp_time(&mut payload, start).expect("write time"); + ReplicationMsgpCodec::write_time(&mut payload, start).expect("write time"); rmp::encode::write_str(&mut payload, "lst").expect("write key"); - write_msgp_time(&mut payload, last).expect("write time"); + ReplicationMsgpCodec::write_time(&mut payload, last).expect("write time"); rmp::encode::write_str(&mut payload, "id").expect("write key"); rmp::encode::write_str(&mut payload, "resync-1").expect("write id"); rmp::encode::write_str(&mut payload, "rdt").expect("write key"); - write_msgp_time(&mut payload, before).expect("write time"); + ReplicationMsgpCodec::write_time(&mut payload, before).expect("write time"); rmp::encode::write_str(&mut payload, "rst").expect("write key"); rmp::encode::write_i32(&mut payload, 3).expect("write status"); rmp::encode::write_str(&mut payload, "fs").expect("write key"); @@ -4149,7 +4151,7 @@ mod tests { rmp::encode::write_str(&mut payload, "id").expect("write key"); rmp::encode::write_i32(&mut payload, 42).expect("write id"); rmp::encode::write_str(&mut payload, "lu").expect("write key"); - write_msgp_time(&mut payload, bucket_last).expect("write lu"); + ReplicationMsgpCodec::write_time(&mut payload, bucket_last).expect("write lu"); let got = BucketReplicationResyncStatus::unmarshal_msg(&payload).expect("decode"); assert_eq!(got.version, 1); diff --git a/crates/ecstore/src/bucket/replication/replication_tagging_boundary.rs b/crates/ecstore/src/bucket/replication/replication_tagging_boundary.rs index 03e57def1..69b81383f 100644 --- a/crates/ecstore/src/bucket/replication/replication_tagging_boundary.rs +++ b/crates/ecstore/src/bucket/replication/replication_tagging_boundary.rs @@ -14,17 +14,21 @@ use std::collections::HashMap; -pub(crate) fn decode_tags_to_map(tags: &str) -> HashMap { - crate::bucket::tagging::decode_tags_to_map(tags) +pub(crate) struct ReplicationTagFilter; + +impl ReplicationTagFilter { + pub(crate) fn decode_tags_to_map(tags: &str) -> HashMap { + crate::bucket::tagging::decode_tags_to_map(tags) + } } #[cfg(test)] mod tests { - use super::decode_tags_to_map; + use super::ReplicationTagFilter; #[test] fn decode_tags_to_map_preserves_bucket_tagging_parser_behavior() { - let tags = decode_tags_to_map("env=prod&encoded=a%2Fb&=ignored"); + let tags = ReplicationTagFilter::decode_tags_to_map("env=prod&encoded=a%2Fb&=ignored"); assert_eq!(tags.get("env").map(String::as_str), Some("prod")); assert_eq!(tags.get("encoded").map(String::as_str), Some("a/b")); diff --git a/crates/ecstore/src/bucket/replication/replication_target_boundary.rs b/crates/ecstore/src/bucket/replication/replication_target_boundary.rs index 31c3c76a8..981853924 100644 --- a/crates/ecstore/src/bucket/replication/replication_target_boundary.rs +++ b/crates/ecstore/src/bucket/replication/replication_target_boundary.rs @@ -21,14 +21,18 @@ pub(crate) use crate::bucket::bucket_target_sys::{ }; pub(crate) use crate::bucket::target::BucketTargets; -pub(crate) async fn list_bucket_targets(bucket: &str) -> Result { - BucketTargetSys::get().list_bucket_targets(bucket).await -} +pub(crate) struct ReplicationTargetStore; -pub(crate) async fn remote_target_client(bucket: &str, arn: &str) -> Option> { - BucketTargetSys::get().get_remote_target_client(bucket, arn).await -} +impl ReplicationTargetStore { + pub(crate) async fn list_bucket_targets(bucket: &str) -> Result { + BucketTargetSys::get().list_bucket_targets(bucket).await + } -pub(crate) async fn target_is_offline(target_client: &TargetClient) -> bool { - BucketTargetSys::get().is_offline(&target_client.to_url()).await + pub(crate) async fn remote_target_client(bucket: &str, arn: &str) -> Option> { + BucketTargetSys::get().get_remote_target_client(bucket, arn).await + } + + pub(crate) async fn target_is_offline(target_client: &TargetClient) -> bool { + BucketTargetSys::get().is_offline(&target_client.to_url()).await + } } diff --git a/crates/ecstore/src/bucket/replication/replication_versioning_boundary.rs b/crates/ecstore/src/bucket/replication/replication_versioning_boundary.rs index 75835d441..fa42d43ab 100644 --- a/crates/ecstore/src/bucket/replication/replication_versioning_boundary.rs +++ b/crates/ecstore/src/bucket/replication/replication_versioning_boundary.rs @@ -14,10 +14,14 @@ use crate::bucket::versioning_sys::BucketVersioningSys; -pub(crate) async fn prefix_enabled(bucket: &str, prefix: &str) -> bool { - BucketVersioningSys::prefix_enabled(bucket, prefix).await -} +pub(crate) struct ReplicationVersioningStore; -pub(crate) async fn prefix_suspended(bucket: &str, prefix: &str) -> bool { - BucketVersioningSys::prefix_suspended(bucket, prefix).await +impl ReplicationVersioningStore { + pub(crate) async fn prefix_enabled(bucket: &str, prefix: &str) -> bool { + BucketVersioningSys::prefix_enabled(bucket, prefix).await + } + + pub(crate) async fn prefix_suspended(bucket: &str, prefix: &str) -> bool { + BucketVersioningSys::prefix_suspended(bucket, prefix).await + } } diff --git a/docs/architecture/ecstore-module-split-plan.md b/docs/architecture/ecstore-module-split-plan.md index 95e6dfe7a..31a6f8b70 100644 --- a/docs/architecture/ecstore-module-split-plan.md +++ b/docs/architecture/ecstore-module-split-plan.md @@ -134,11 +134,11 @@ Required contracts before crate movement: `crates/ecstore/src/bucket/replication/replication_storage_boundary.rs`. - `ReplicationMetadataStore`: replication config, target reset headers, MRF/resync state, and status persistence. Metadata sys access and replication - metadata path constants are concentrated in + metadata path constants are exposed through the contract type in `crates/ecstore/src/bucket/replication/replication_metadata_boundary.rs`. - `ReplicationConfigStore`: replication config persistence and config-derived labels used by target options. Config read/save helpers and storage class - labels are concentrated in + labels are exposed through the contract type in `crates/ecstore/src/bucket/replication/replication_config_store.rs`. - `ReplicationFileMeta`: replication status, decisions, MRF entries, resync decisions, and target reset helpers. `rustfs_filemeta` replication contracts @@ -153,13 +153,17 @@ Required contracts before crate movement: - `ReplicationTargetStore`: bucket target listing, target client lookup, target offline checks, target config types, and target operation option types. Bucket target sys access, `BucketTargets`, and target operation types - are concentrated in + are exposed through the contract type in `crates/ecstore/src/bucket/replication/replication_target_boundary.rs`. - `ReplicationRuntime`: pool, stats, worker admission, bucket monitor, local node identity, cancellation, and queue sizing. Concrete ECStore object store and bucket monitor types stay behind local storage/bandwidth boundaries. - `ReplicationBandwidthLimiter`: target reader wrapping for replication bandwidth accounting and throttling. +- `ReplicationVersioningStore`, `ReplicationLockTiming`, `ReplicationMsgpCodec`, + and `ReplicationTagFilter`: smaller state/codec/filter contracts that keep + bucket versioning, SetDisks lock timing, MessagePack helpers, and bucket + tagging helper access behind local replication boundary types. - `ReplicationEventSink`: notification/audit events for skipped, failed, and completed replication operations, including local event host selection. - `ReplicationLifecycleBridge`: lifecycle-originated delete and version-purge