refactor(replication): isolate stats and app contract boundaries (#4235)

* refactor(replication): isolate stats boundary adapters

* refactor(replication): route app contracts through storage boundary

rustfs/src/app/bucket_usecase.rs

@@ -33,6 +33,10 @@ use super::storage_api::bucket_usecase::bucket::{
     },
     metadata_sys,
     policy_sys::PolicySys,
+    replication::{
+        ReplicationTargetValidationError, replication_target_arns, should_remove_replication_target,
+        validate_replication_config_target_arns,
+    },
     target::{BucketTargetType, BucketTargets},
     utils::serialize,
     versioning_sys::BucketVersioningSys,
@@ -252,14 +256,14 @@ fn validate_replication_config_targets(targets: &BucketTargets, config: &Replica
         .filter(|target| target.target_type == BucketTargetType::ReplicationService)
         .map(|target| target.arn.as_str());
 
-    match rustfs_replication::validate_replication_config_target_arns(configured_arns, config) {
+    match validate_replication_config_target_arns(configured_arns, config) {
         Ok(()) => Ok(()),
         Err(err) => {
             let message = match err {
-                rustfs_replication::ReplicationTargetValidationError::RoleWithMultipleDestinations => {
+                ReplicationTargetValidationError::RoleWithMultipleDestinations => {
                     "replication config with Role cannot define multiple destination targets"
                 }
-                rustfs_replication::ReplicationTargetValidationError::StaleTarget => "replication config has a stale target",
+                ReplicationTargetValidationError::StaleTarget => "replication config has a stale target",
             };
             Err(S3Error::with_message(S3ErrorCode::InvalidRequest, message))
         }
@@ -290,7 +294,7 @@ async fn replication_targets_without_config_targets(
     bucket: &str,
     config: &ReplicationConfiguration,
 ) -> S3Result<Option<(BucketTargets, usize)>> {
-    let target_arns = rustfs_replication::replication_target_arns(config);
+    let target_arns = replication_target_arns(config);
     if target_arns.is_empty() {
         return Ok(None);
     }
@@ -315,7 +319,7 @@ async fn replication_targets_without_config_targets(
 fn remove_replication_targets_from_config_targets(targets: &mut BucketTargets, target_arns: &HashSet<String>) -> usize {
     let original_len = targets.targets.len();
     targets.targets.retain(|target| {
-        !rustfs_replication::should_remove_replication_target(
+        !should_remove_replication_target(
             target.arn.as_str(),
             target.target_type == BucketTargetType::ReplicationService,
             target_arns,
@@ -2387,7 +2391,7 @@ mod tests {
             rules: vec![replication_rule_for_target(destination)],
         };
 
-        let arns = rustfs_replication::replication_target_arns(&config);
+        let arns = replication_target_arns(&config);
 
         assert!(arns.contains(role));
         assert!(!arns.contains(destination));
@@ -2401,7 +2405,7 @@ mod tests {
             rules: vec![replication_rule_for_target(destination)],
         };
 
-        let arns = rustfs_replication::replication_target_arns(&config);
+        let arns = replication_target_arns(&config);
 
         assert!(arns.contains(destination));
     }

rustfs/src/app/object_usecase.rs

@@ -22,6 +22,8 @@ use super::storage_api::object_usecase::access::{
     PostObjectRequestMarker, authorize_request, has_bypass_governance_header, req_info_mut,
 };
 use super::storage_api::object_usecase::bucket::quota::checker::QuotaChecker;
+#[cfg(test)]
+use super::storage_api::object_usecase::bucket::replication::{ReplicationState, replication_statuses_map};
 use super::storage_api::object_usecase::bucket::{
     VersioningConfigExt as _,
     lifecycle::{
@@ -38,8 +40,10 @@ use super::storage_api::object_usecase::bucket::{
     predict_lifecycle_expiration,
     quota::QuotaOperation,
     replication::{
-        DeletedObjectReplicationInfo, check_replicate_delete, delete_replication_state_from_config, must_replicate_object,
-        schedule_object_replication, schedule_replication_delete,
+        DeletedObjectReplicationInfo, REPLICATE_INCOMING_DELETE, ReplicationStatusType, VersionPurgeStatusType,
+        check_replicate_delete, delete_replication_state_from_config, delete_replication_version_id, must_replicate_object,
+        schedule_object_replication, schedule_replication_delete, should_schedule_delete_replication,
+        should_use_existing_delete_replication_info, should_use_existing_delete_replication_source,
     },
     tagging::decode_tags,
     validate_restore_request,
@@ -116,9 +120,6 @@ use rustfs_lock::NamespaceLockGuard;
 use rustfs_notify::EventArgsBuilder;
 use rustfs_object_capacity::capacity_manager::get_capacity_manager;
 use rustfs_policy::policy::action::{Action, S3Action};
-use rustfs_replication::{REPLICATE_INCOMING_DELETE, ReplicationStatusType, VersionPurgeStatusType};
-#[cfg(test)]
-use rustfs_replication::{ReplicationState, replication_statuses_map};
 use rustfs_s3_ops::{S3Operation, delete_event_name_for_marker, put_event_name_for_post_object};
 use rustfs_s3select_api::object_store::bytes_stream;
 use rustfs_targets::{
@@ -1523,21 +1524,6 @@ async fn enrich_delete_replication_state_if_needed(
     }
 }
 
-fn should_schedule_delete_replication(
-    opts: &ObjectOptions,
-    replication_source: &ObjectInfo,
-    deleted_delete_marker_version: bool,
-) -> bool {
-    rustfs_replication::should_schedule_delete_replication(rustfs_replication::ReplicationDeleteScheduleInput {
-        replication_request: opts.replication_request,
-        version_id_requested: opts.version_id.is_some(),
-        source_delete_marker: replication_source.delete_marker,
-        source_replication_status: &replication_source.replication_status,
-        source_version_purge_status: &replication_source.version_purge_status,
-        deleted_delete_marker_version,
-    })
-}
-
 async fn should_schedule_replica_delete_replication(
     bucket: &str,
     replication_source: &ObjectInfo,
@@ -1550,18 +1536,6 @@ async fn should_schedule_replica_delete_replication(
     delete_replication_state_from_config(&config, replication_source, version_id, true).is_some()
 }
 
-fn delete_replication_version_id(replication_source: &ObjectInfo, deleted_delete_marker_version: bool) -> Option<Uuid> {
-    rustfs_replication::delete_replication_version_id(
-        replication_source.delete_marker,
-        replication_source.version_id,
-        deleted_delete_marker_version,
-    )
-}
-
-fn should_use_existing_delete_replication_info(opts: &ObjectOptions) -> bool {
-    rustfs_replication::should_use_existing_delete_replication_info(opts.version_id.is_some(), opts.delete_marker)
-}
-
 fn internal_object_info_lookup_opts(mut opts: ObjectOptions) -> ObjectOptions {
     opts.http_preconditions = None;
     opts
@@ -1596,7 +1570,7 @@ fn delete_replication_state_source<'a>(
     existing_object_info: Option<&'a ObjectInfo>,
     deleted_object_info: &'a ObjectInfo,
 ) -> &'a ObjectInfo {
-    if rustfs_replication::should_use_existing_delete_replication_source(
+    if should_use_existing_delete_replication_source(
         opts.replication_request,
         deleted_object_info.delete_marker,
         existing_object_info.is_some(),

rustfs/src/app/storage_api.rs

@@ -605,6 +605,7 @@ pub(crate) mod bucket {
 
     pub(crate) mod replication {
         use std::collections::HashMap;
+        use std::collections::HashSet;
         use std::sync::Arc;
         use uuid::Uuid;
 
@@ -612,6 +613,14 @@ pub(crate) mod bucket {
             crate::storage::storage_api::ecstore_bucket::replication::DeletedObjectReplicationInfo;
         type ReplicationObjectBridge = crate::storage::storage_api::ecstore_bucket::replication::ReplicationObjectBridge;
         pub(crate) type ReplicateDecision = rustfs_replication::ReplicateDecision;
+        #[cfg(test)]
+        pub(crate) type ReplicationState = rustfs_replication::ReplicationState;
+        pub(crate) type ReplicationStatusType = rustfs_replication::ReplicationStatusType;
+        pub(crate) type ReplicationTargetValidationError = rustfs_replication::ReplicationTargetValidationError;
+        pub(crate) type VersionPurgeStatusType = rustfs_replication::VersionPurgeStatusType;
+        pub(crate) const REPLICATE_INCOMING_DELETE: &str = rustfs_replication::REPLICATE_INCOMING_DELETE;
+        #[cfg(test)]
+        pub(crate) use rustfs_replication::replication_statuses_map;
 
         pub(crate) async fn check_replicate_delete(
             bucket: &str,
@@ -623,12 +632,23 @@ pub(crate) mod bucket {
             ReplicationObjectBridge::check_delete(bucket, dobj, oi, del_opts, gerr).await
         }
 
+        pub(crate) fn delete_replication_version_id(
+            replication_source: &crate::storage::storage_api::StorageObjectInfo,
+            deleted_delete_marker_version: bool,
+        ) -> Option<Uuid> {
+            rustfs_replication::delete_replication_version_id(
+                replication_source.delete_marker,
+                replication_source.version_id,
+                deleted_delete_marker_version,
+            )
+        }
+
         pub(crate) async fn must_replicate_object(
             bucket: &str,
             object: &str,
             user_defined: &HashMap<String, String>,
             user_tags: String,
-            status: rustfs_replication::ReplicationStatusType,
+            status: ReplicationStatusType,
             opts: crate::storage::storage_api::StorageObjectOptions,
         ) -> ReplicateDecision {
             let mopts = ReplicationObjectBridge::must_replicate_options(
@@ -668,6 +688,58 @@ pub(crate) mod bucket {
             };
             rustfs_replication::delete_replication_state_from_config(config, &source)
         }
+
+        pub(crate) fn replication_target_arns(config: &s3s::dto::ReplicationConfiguration) -> HashSet<String> {
+            rustfs_replication::replication_target_arns(config)
+        }
+
+        pub(crate) fn should_remove_replication_target(
+            target_arn: &str,
+            is_replication_service: bool,
+            target_arns: &HashSet<String>,
+        ) -> bool {
+            rustfs_replication::should_remove_replication_target(target_arn, is_replication_service, target_arns)
+        }
+
+        pub(crate) fn should_schedule_delete_replication(
+            opts: &crate::storage::storage_api::StorageObjectOptions,
+            replication_source: &crate::storage::storage_api::StorageObjectInfo,
+            deleted_delete_marker_version: bool,
+        ) -> bool {
+            rustfs_replication::should_schedule_delete_replication(rustfs_replication::ReplicationDeleteScheduleInput {
+                replication_request: opts.replication_request,
+                version_id_requested: opts.version_id.is_some(),
+                source_delete_marker: replication_source.delete_marker,
+                source_replication_status: &replication_source.replication_status,
+                source_version_purge_status: &replication_source.version_purge_status,
+                deleted_delete_marker_version,
+            })
+        }
+
+        pub(crate) fn should_use_existing_delete_replication_info(
+            opts: &crate::storage::storage_api::StorageObjectOptions,
+        ) -> bool {
+            rustfs_replication::should_use_existing_delete_replication_info(opts.version_id.is_some(), opts.delete_marker)
+        }
+
+        pub(crate) fn should_use_existing_delete_replication_source(
+            replication_request: bool,
+            deleted_delete_marker: bool,
+            has_existing_info: bool,
+        ) -> bool {
+            rustfs_replication::should_use_existing_delete_replication_source(
+                replication_request,
+                deleted_delete_marker,
+                has_existing_info,
+            )
+        }
+
+        pub(crate) fn validate_replication_config_target_arns<'a>(
+            configured_arns: impl Iterator<Item = &'a str>,
+            config: &s3s::dto::ReplicationConfiguration,
+        ) -> Result<(), ReplicationTargetValidationError> {
+            rustfs_replication::validate_replication_config_target_arns(configured_arns, config)
+        }
     }
 
     pub(crate) mod tagging {