diff --git a/docs/architecture/ecstore-module-split-plan.md b/docs/architecture/ecstore-module-split-plan.md index 125bf4c06..4338952f8 100644 --- a/docs/architecture/ecstore-module-split-plan.md +++ b/docs/architecture/ecstore-module-split-plan.md @@ -123,6 +123,9 @@ Current coupling: - admin replication extension target filtering and resync request construction stay behind the admin storage boundary instead of exposing replication work DTO construction to handlers; +- app object and multipart writes call object-replication boundary helpers + instead of constructing replication work DTOs or choosing object replication + operation types at the use-case layer; - global replication pool/stat initialization still lives with ECStore runtime compatibility state; - modules inside `bucket/replication` use local relative paths rather than the diff --git a/rustfs/src/app/multipart_usecase.rs b/rustfs/src/app/multipart_usecase.rs index b179f718e..3c130ae4c 100644 --- a/rustfs/src/app/multipart_usecase.rs +++ b/rustfs/src/app/multipart_usecase.rs @@ -21,7 +21,7 @@ use super::storage_api::multipart_usecase::bucket::{ lifecycle::{bucket_lifecycle_audit::LcEventSrc, bucket_lifecycle_ops::enqueue_transition_immediate}, metadata_sys, quota::QuotaOperation, - replication::{get_must_replicate_options, must_replicate, schedule_replication}, + replication::{must_replicate_object, schedule_object_replication}, versioning_sys::BucketVersioningSys, }; use super::storage_api::multipart_usecase::compression::is_disk_compressible; @@ -63,7 +63,6 @@ use crate::table_catalog; use bytes::Bytes; use futures::StreamExt; use http::{HeaderMap, Uri}; -use rustfs_filemeta::ReplicationType; use rustfs_s3_ops::S3Operation; use rustfs_targets::EventName; use rustfs_utils::CompressionAlgorithm; @@ -560,18 +559,13 @@ impl DefaultMultipartUsecase { ..Default::default() }; let mt2 = obj_info.user_defined.clone(); - let replicate_options = get_must_replicate_options( - &mt2, - "".to_string(), - opts.delete_marker_replication_status(), - ReplicationType::Object, - opts.clone(), - ); - let dsc = must_replicate(&bucket, &key, replicate_options).await; + let dsc = + must_replicate_object(&bucket, &key, &mt2, "".to_string(), opts.delete_marker_replication_status(), opts.clone()) + .await; if dsc.replicate_any() { warn!("need multipart replication"); - schedule_replication(obj_info.clone(), store, dsc, ReplicationType::Object).await; + schedule_object_replication(obj_info.clone(), store, dsc).await; } // Set object info for event notification @@ -699,14 +693,9 @@ impl DefaultMultipartUsecase { .await .map_err(ApiError::from)?; - let replicate_options = get_must_replicate_options( - &mt2, - "".to_string(), - opts.delete_marker_replication_status(), - ReplicationType::Object, - opts.clone(), - ); - let dsc = must_replicate(&bucket, &key, replicate_options).await; + let dsc = + must_replicate_object(&bucket, &key, &mt2, "".to_string(), opts.delete_marker_replication_status(), opts.clone()) + .await; if dsc.replicate_any() { insert_str(&mut opts.user_defined, SUFFIX_REPLICATION_TIMESTAMP, jiff::Zoned::now().to_string()); insert_str( diff --git a/rustfs/src/app/object_usecase.rs b/rustfs/src/app/object_usecase.rs index 1794b1a6f..1262d3f3d 100644 --- a/rustfs/src/app/object_usecase.rs +++ b/rustfs/src/app/object_usecase.rs @@ -23,7 +23,7 @@ use super::storage_api::object_usecase::access::{ }; use super::storage_api::object_usecase::bucket::quota::checker::QuotaChecker; use super::storage_api::object_usecase::bucket::{ - ReplicationConfigExt as _, VersioningConfigExt as _, + VersioningConfigExt as _, lifecycle::{ bucket_lifecycle_audit::LcEventSrc, bucket_lifecycle_ops::{enqueue_transition_immediate, post_restore_opts}, @@ -38,8 +38,8 @@ use super::storage_api::object_usecase::bucket::{ predict_lifecycle_expiration, quota::QuotaOperation, replication::{ - DeletedObjectReplicationInfo, ObjectOpts as ReplicationObjectOpts, check_replicate_delete, get_must_replicate_options, - must_replicate, schedule_replication, schedule_replication_delete, + DeletedObjectReplicationInfo, check_replicate_delete, delete_replication_state_from_config, must_replicate_object, + schedule_object_replication, schedule_replication_delete, }, tagging::decode_tags, validate_restore_request, @@ -109,10 +109,10 @@ use pin_project_lite::pin_project; use rustfs_concurrency::GetObjectQueueSnapshot; use rustfs_config::MI_B; use rustfs_filemeta::{ - REPLICATE_INCOMING_DELETE, ReplicateDecision, ReplicateTargetDecision, ReplicationState, ReplicationStatusType, - ReplicationType, RestoreStatusOps, VersionPurgeStatusType, parse_restore_obj_status, replication_statuses_map, - version_purge_statuses_map, + REPLICATE_INCOMING_DELETE, ReplicationStatusType, RestoreStatusOps, VersionPurgeStatusType, parse_restore_obj_status, }; +#[cfg(test)] +use rustfs_filemeta::{ReplicationState, replication_statuses_map}; use rustfs_io_core::{BytesPool, PooledBuffer}; use rustfs_io_metrics; use rustfs_lock::NamespaceLockGuard; @@ -147,9 +147,9 @@ use s3s::dto::{ DeleteObjectsOutput, DeletedObject, ETag, GetObjectAttributesInput, GetObjectAttributesOutput, GetObjectAttributesParts, GetObjectInput, GetObjectOutput, HeadObjectInput, HeadObjectOutput, MetadataDirective, ObjectAttributes, ObjectLockLegalHold, ObjectLockLegalHoldStatus, ObjectLockMode, ObjectLockRetention, ObjectLockRetentionMode, ObjectPart, PutObjectInput, - PutObjectOutput, Range, ReplicationConfiguration, RequestCharged, RestoreObjectInput, RestoreObjectOutput, RestoreStatus, - SSECustomerAlgorithm, SSECustomerKeyMD5, SSEKMSKeyId, SelectObjectContentInput, SelectObjectContentOutput, - ServerSideEncryption, StorageClass, StreamingBlob, TaggingHeader, Timestamp, TimestampFormat, WebsiteRedirectLocation, + PutObjectOutput, Range, RequestCharged, RestoreObjectInput, RestoreObjectOutput, RestoreStatus, SSECustomerAlgorithm, + SSECustomerKeyMD5, SSEKMSKeyId, SelectObjectContentInput, SelectObjectContentOutput, ServerSideEncryption, StorageClass, + StreamingBlob, TaggingHeader, Timestamp, TimestampFormat, WebsiteRedirectLocation, }; use s3s::header::{X_AMZ_RESTORE, X_AMZ_RESTORE_OUTPUT_PATH}; use s3s::stream::{ByteStream, RemainingLength}; @@ -1403,51 +1403,6 @@ fn build_put_object_expiration_header(event: &lifecycle::Event) -> Option, - replica: bool, -) -> Option { - let opts = ReplicationObjectOpts { - name: obj_info.name.clone(), - user_tags: (*obj_info.user_tags).clone(), - version_id, - delete_marker: obj_info.delete_marker, - op_type: ReplicationType::Delete, - replica, - ..Default::default() - }; - let target_arns = config.filter_target_arns(&opts); - if target_arns.is_empty() { - return None; - } - - let mut decision = ReplicateDecision::new(); - for target_arn in target_arns { - let mut target_opts = opts.clone(); - target_opts.target_arn = target_arn.clone(); - decision.set(ReplicateTargetDecision::new(target_arn, config.replicate(&target_opts), false)); - } - if !decision.replicate_any() { - return None; - } - - let pending_status = decision.pending_status(); - let mut state = ReplicationState { - replicate_decision_str: decision.to_string(), - ..Default::default() - }; - if version_id.is_some() { - state.version_purge_status_internal = pending_status.clone(); - state.purge_targets = version_purge_statuses_map(pending_status.as_deref().unwrap_or_default()); - } else { - state.replication_status_internal = pending_status.clone(); - state.targets = replication_statuses_map(pending_status.as_deref().unwrap_or_default()); - } - Some(state) -} - async fn enrich_delete_replication_state_if_needed( bucket: &str, delete_object: &mut StorageDeletedObject, @@ -3168,14 +3123,9 @@ impl DefaultObjectUsecase { .map(|ctx| ctx.request_id.clone()) .unwrap_or_else(|| request_context::RequestContext::fallback().request_id); - let repoptions = get_must_replicate_options( - &mt2, - "".to_string(), - opts.delete_marker_replication_status(), - ReplicationType::Object, - opts.clone(), - ); - let dsc = must_replicate(&bucket, &key, repoptions).await; + let dsc = + must_replicate_object(&bucket, &key, &mt2, "".to_string(), opts.delete_marker_replication_status(), opts.clone()) + .await; if dsc.replicate_any() { insert_str(&mut opts.user_defined, SUFFIX_REPLICATION_TIMESTAMP, jiff::Zoned::now().to_string()); @@ -3284,19 +3234,11 @@ impl DefaultObjectUsecase { let e_tag = obj_info.etag.clone().map(|etag| to_s3s_etag(&etag)); - let repoptions = get_must_replicate_options( - &mt2, - "".to_string(), - opts.delete_marker_replication_status(), - ReplicationType::Object, - opts, - ); - - let dsc = must_replicate(&bucket, &key, repoptions).await; + let dsc = must_replicate_object(&bucket, &key, &mt2, "".to_string(), opts.delete_marker_replication_status(), opts).await; let expiration = resolve_put_object_expiration(&bucket, &obj_info).await; if dsc.replicate_any() { - schedule_replication(obj_info.clone(), store, dsc, ReplicationType::Object).await; + schedule_object_replication(obj_info.clone(), store, dsc).await; } let mut checksums = PutObjectChecksums { diff --git a/rustfs/src/app/storage_api.rs b/rustfs/src/app/storage_api.rs index 0b22c22e5..7bdf552dd 100644 --- a/rustfs/src/app/storage_api.rs +++ b/rustfs/src/app/storage_api.rs @@ -237,25 +237,6 @@ pub(crate) mod bucket { } } - pub(crate) trait ReplicationConfigExt { - fn filter_target_arns(&self, obj: &replication::ObjectOpts) -> Vec; - fn replicate(&self, opts: &replication::ObjectOpts) -> bool; - } - - impl ReplicationConfigExt for s3s::dto::ReplicationConfiguration { - fn filter_target_arns(&self, obj: &replication::ObjectOpts) -> Vec { - ::filter_target_arns( - self, obj, - ) - } - - fn replicate(&self, opts: &replication::ObjectOpts) -> bool { - ::replicate( - self, opts, - ) - } - } - pub(crate) trait VersioningConfigExt { fn enabled(&self) -> bool; fn prefix_enabled(&self, prefix: &str) -> bool; @@ -625,13 +606,12 @@ pub(crate) mod bucket { pub(crate) mod replication { use std::collections::HashMap; use std::sync::Arc; + use uuid::Uuid; pub(crate) type DeletedObjectReplicationInfo = crate::storage::storage_api::ecstore_bucket::replication::DeletedObjectReplicationInfo; - pub(crate) type MustReplicateOptions = crate::storage::storage_api::ecstore_bucket::replication::MustReplicateOptions; - pub(crate) type ObjectOpts = crate::storage::storage_api::ecstore_bucket::replication::ObjectOpts; - pub(crate) type ReplicationObjectBridge = - crate::storage::storage_api::ecstore_bucket::replication::ReplicationObjectBridge; + type ObjectOpts = crate::storage::storage_api::ecstore_bucket::replication::ObjectOpts; + type ReplicationObjectBridge = crate::storage::storage_api::ecstore_bucket::replication::ReplicationObjectBridge; pub(crate) type ReplicateDecision = rustfs_filemeta::ReplicateDecision; pub(crate) async fn check_replicate_delete( @@ -644,32 +624,87 @@ pub(crate) mod bucket { ReplicationObjectBridge::check_delete(bucket, dobj, oi, del_opts, gerr).await } - pub(crate) fn get_must_replicate_options( + pub(crate) async fn must_replicate_object( + bucket: &str, + object: &str, user_defined: &HashMap, user_tags: String, status: rustfs_filemeta::ReplicationStatusType, - op_type: rustfs_filemeta::ReplicationType, opts: crate::storage::storage_api::StorageObjectOptions, - ) -> MustReplicateOptions { - ReplicationObjectBridge::must_replicate_options(user_defined, user_tags, status, op_type, opts) - } - - pub(crate) async fn must_replicate(bucket: &str, object: &str, mopts: MustReplicateOptions) -> ReplicateDecision { + ) -> ReplicateDecision { + let mopts = ReplicationObjectBridge::must_replicate_options( + user_defined, + user_tags, + status, + rustfs_filemeta::ReplicationType::Object, + opts, + ); ReplicationObjectBridge::must_replicate(bucket, object, mopts).await } - pub(crate) async fn schedule_replication( + pub(crate) async fn schedule_object_replication( oi: crate::storage::storage_api::StorageObjectInfo, store: Arc, dsc: ReplicateDecision, - op_type: rustfs_filemeta::ReplicationType, ) { - ReplicationObjectBridge::schedule_object(oi, store, dsc, op_type).await; + ReplicationObjectBridge::schedule_object(oi, store, dsc, rustfs_filemeta::ReplicationType::Object).await; } pub(crate) async fn schedule_replication_delete(dv: DeletedObjectReplicationInfo) { ReplicationObjectBridge::schedule_delete(dv).await; } + + pub(crate) fn delete_replication_state_from_config( + config: &s3s::dto::ReplicationConfiguration, + obj_info: &crate::storage::storage_api::StorageObjectInfo, + version_id: Option, + replica: bool, + ) -> Option { + let opts = ObjectOpts { + name: obj_info.name.clone(), + user_tags: (*obj_info.user_tags).clone(), + version_id, + delete_marker: obj_info.delete_marker, + op_type: rustfs_filemeta::ReplicationType::Delete, + replica, + ..Default::default() + }; + let target_arns = + crate::storage::storage_api::ecstore_bucket::replication::ReplicationConfigurationExt::filter_target_arns( + config, &opts, + ); + if target_arns.is_empty() { + return None; + } + + let mut decision = ReplicateDecision::new(); + for target_arn in target_arns { + let mut target_opts = opts.clone(); + target_opts.target_arn = target_arn.clone(); + let replicate = crate::storage::storage_api::ecstore_bucket::replication::ReplicationConfigurationExt::replicate( + config, + &target_opts, + ); + decision.set(rustfs_filemeta::ReplicateTargetDecision::new(target_arn, replicate, false)); + } + if !decision.replicate_any() { + return None; + } + + let pending_status = decision.pending_status(); + let mut state = rustfs_filemeta::ReplicationState { + replicate_decision_str: decision.to_string(), + ..Default::default() + }; + if version_id.is_some() { + state.version_purge_status_internal = pending_status.clone(); + state.purge_targets = rustfs_filemeta::version_purge_statuses_map(pending_status.as_deref().unwrap_or_default()); + } else { + state.replication_status_internal = pending_status.clone(); + state.targets = rustfs_filemeta::replication_statuses_map(pending_status.as_deref().unwrap_or_default()); + } + Some(state) + } } pub(crate) mod tagging { diff --git a/scripts/check_architecture_migration_rules.sh b/scripts/check_architecture_migration_rules.sh index 2e1cd8546..29665abab 100755 --- a/scripts/check_architecture_migration_rules.sh +++ b/scripts/check_architecture_migration_rules.sh @@ -200,6 +200,7 @@ EXTERNAL_ECSTORE_API_BOUNDARY_HITS_FILE="${TMP_DIR}/external_ecstore_api_boundar REPLICATION_FACADE_BYPASS_HITS_FILE="${TMP_DIR}/replication_facade_bypass_hits.txt" REPLICATION_FACADE_WILDCARD_EXPORT_HITS_FILE="${TMP_DIR}/replication_facade_wildcard_export_hits.txt" ADMIN_REPLICATION_DTO_BOUNDARY_BYPASS_HITS_FILE="${TMP_DIR}/admin_replication_dto_boundary_bypass_hits.txt" +APP_REPLICATION_DTO_BOUNDARY_BYPASS_HITS_FILE="${TMP_DIR}/app_replication_dto_boundary_bypass_hits.txt" REPLICATION_BANDWIDTH_BOUNDARY_BYPASS_HITS_FILE="${TMP_DIR}/replication_bandwidth_boundary_bypass_hits.txt" REPLICATION_CONFIG_STORE_BYPASS_HITS_FILE="${TMP_DIR}/replication_config_store_bypass_hits.txt" REPLICATION_ERROR_BOUNDARY_BYPASS_HITS_FILE="${TMP_DIR}/replication_error_boundary_bypass_hits.txt" @@ -2555,6 +2556,24 @@ if [[ -s "$REPLICATION_OBJECT_BRIDGE_BYPASS_HITS_FILE" ]]; then report_failure "object replication work entry points must stay behind ReplicationObjectBridge: $(paste -sd '; ' "$REPLICATION_OBJECT_BRIDGE_BYPASS_HITS_FILE")" fi +( + cd "$ROOT_DIR" + { + rg -n --with-filename '\b(get_must_replicate_options|must_replicate|schedule_replication|MustReplicateOptions|ReplicationObjectBridge|ReplicationObjectOpts)\b|ObjectOpts as ReplicationObjectOpts|ReplicationType::Object' \ + rustfs/src/app \ + --glob '*.rs' \ + --glob '!rustfs/src/app/storage_api.rs' || true + rg -n --with-filename '(?:ecstore_bucket::replication|crate::storage::storage_api::ecstore_bucket::replication)::(?:ObjectOpts|MustReplicateOptions|ReplicationObjectBridge|ReplicationConfigurationExt)' \ + rustfs/src/app \ + --glob '*.rs' \ + --glob '!rustfs/src/app/storage_api.rs' || true + } +) >"$APP_REPLICATION_DTO_BOUNDARY_BYPASS_HITS_FILE" + +if [[ -s "$APP_REPLICATION_DTO_BOUNDARY_BYPASS_HITS_FILE" ]]; then + report_failure "app replication ObjectOpts/MustReplicateOptions/bridge access must stay behind rustfs/src/app/storage_api.rs: $(paste -sd '; ' "$APP_REPLICATION_DTO_BOUNDARY_BYPASS_HITS_FILE")" +fi + ( cd "$ROOT_DIR" {