refactor(replication): route app storage through ecstore (#4251)

Cargo.lock

@@ -8836,7 +8836,6 @@ dependencies = [
  "rustfs-policy",
  "rustfs-protocols",
  "rustfs-protos",
- "rustfs-replication",
  "rustfs-rio",
  "rustfs-s3-ops",
  "rustfs-s3-types",
@@ -9764,7 +9763,6 @@ dependencies = [
  "rustfs-data-usage",
  "rustfs-ecstore",
  "rustfs-filemeta",
- "rustfs-replication",
  "rustfs-storage-api",
  "rustfs-utils",
  "s3s",

crates/ecstore/src/bucket/replication/mod.rs

@@ -43,18 +43,26 @@ mod replication_versioning_boundary;
 mod runtime_boundary;
 
 pub use datatypes::ResyncStatusType;
-pub use replication_config_boundary::{ObjectOpts, ReplicationConfigurationExt};
+pub use replication_config_boundary::{
+    ObjectOpts, ReplicationConfigurationExt, ReplicationTargetValidationError, replication_target_arns,
+    should_remove_replication_target, validate_replication_config_target_arns,
+};
 #[cfg(test)]
 pub(crate) use replication_filemeta_boundary::ReplicateTargetDecision;
-pub(crate) use replication_filemeta_boundary::{
-    ReplicateDecision, ReplicationState, replication_statuses_map, version_purge_statuses_map,
+pub(crate) use replication_filemeta_boundary::version_purge_statuses_map;
+pub use replication_filemeta_boundary::{
+    REPLICATE_INCOMING_DELETE, ReplicateDecision, ReplicateObjectInfo, ReplicationState, ReplicationStatusType, ReplicationType,
+    VersionPurgeStatusType, replication_statuses_map,
 };
-pub use replication_filemeta_boundary::{ReplicateObjectInfo, ReplicationStatusType, ReplicationType, VersionPurgeStatusType};
 pub(crate) use replication_lifecycle_bridge::{ReplicationLifecycleBridge, ReplicationLifecycleConfig};
 pub(crate) use replication_migration_bridge::ReplicationMigrationBridge;
 pub use replication_object_bridge::ReplicationObjectBridge;
 pub use replication_object_config::ReplicationConfig;
-pub use replication_object_decision_boundary::MustReplicateOptions;
+pub use replication_object_decision_boundary::{
+    MustReplicateOptions, ReplicationDeleteScheduleInput, ReplicationDeleteStateSource, delete_replication_state_from_config,
+    delete_replication_version_id, should_schedule_delete_replication, should_use_existing_delete_replication_info,
+    should_use_existing_delete_replication_source,
+};
 pub use replication_pool::{
     DynReplicationPool, ReplicationPoolTrait, get_global_replication_pool, get_global_replication_stats,
     init_background_replication,

crates/ecstore/src/bucket/replication/replication_config_boundary.rs

@@ -12,4 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-pub use rustfs_replication::{ObjectOpts, ReplicationConfigurationExt};
+pub use rustfs_replication::{
+    ObjectOpts, ReplicationConfigurationExt, ReplicationTargetValidationError, replication_target_arns,
+    should_remove_replication_target, validate_replication_config_target_arns,
+};

crates/ecstore/src/bucket/replication/replication_filemeta_boundary.rs

@@ -14,9 +14,11 @@
 
 pub(crate) use rustfs_replication::{MrfOpKind, MrfReplicateEntry};
 pub(crate) use rustfs_replication::{
-    REPLICATE_EXISTING, REPLICATE_EXISTING_DELETE, REPLICATE_HEAL_DELETE, REPLICATE_INCOMING_DELETE, ReplicateDecision,
-    ReplicateTargetDecision, ReplicatedInfos, ReplicatedTargetInfo, ReplicationAction, ReplicationState,
-    ReplicationWorkerOperation, ResyncDecision, get_replication_state, parse_replicate_decision, replication_statuses_map,
-    target_reset_header, version_purge_statuses_map,
+    REPLICATE_EXISTING, REPLICATE_EXISTING_DELETE, REPLICATE_HEAL_DELETE, ReplicateTargetDecision, ReplicatedInfos,
+    ReplicatedTargetInfo, ReplicationAction, ReplicationWorkerOperation, ResyncDecision, get_replication_state,
+    parse_replicate_decision, target_reset_header, version_purge_statuses_map,
+};
+pub use rustfs_replication::{
+    REPLICATE_INCOMING_DELETE, ReplicateDecision, ReplicateObjectInfo, ReplicationState, ReplicationStatusType, ReplicationType,
+    VersionPurgeStatusType, replication_statuses_map,
 };
-pub use rustfs_replication::{ReplicateObjectInfo, ReplicationStatusType, ReplicationType, VersionPurgeStatusType};

crates/ecstore/src/bucket/replication/replication_object_decision_boundary.rs

@@ -12,7 +12,11 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-pub use rustfs_replication::MustReplicateOptions;
+pub use rustfs_replication::{
+    MustReplicateOptions, ReplicationDeleteScheduleInput, ReplicationDeleteStateSource, delete_replication_state_from_config,
+    delete_replication_version_id, should_schedule_delete_replication, should_use_existing_delete_replication_info,
+    should_use_existing_delete_replication_source,
+};
 pub(crate) use rustfs_replication::{
     ReplicationDeleteSource, ReplicationMultipartPartInput, ReplicationResyncTargetObject,
     delete_replication_missing_source_decision, delete_replication_object_opts, heal_uses_delete_replication_path,

crates/ecstore/tests/replication_facade_compat_test.rs

@@ -16,8 +16,12 @@ mod storage_api;
 
 use s3s::dto::ReplicationConfiguration;
 use storage_api::replication_compat::{
-    BucketStats, DeletedObjectReplicationInfo, DynReplicationPool, ObjectOpts, ReplicationConfigurationExt,
-    ReplicationObjectBridge, ReplicationStats, ResyncStatusType,
+    BucketStats, DeletedObjectReplicationInfo, DynReplicationPool, ObjectOpts, REPLICATE_INCOMING_DELETE, ReplicateDecision,
+    ReplicationConfigurationExt, ReplicationDeleteScheduleInput, ReplicationDeleteStateSource, ReplicationObjectBridge,
+    ReplicationState, ReplicationStats, ReplicationTargetValidationError, ResyncStatusType, delete_replication_state_from_config,
+    delete_replication_version_id, replication_statuses_map, replication_target_arns, should_remove_replication_target,
+    should_schedule_delete_replication, should_use_existing_delete_replication_info,
+    should_use_existing_delete_replication_source, validate_replication_config_target_arns,
 };
 
 fn type_name<T>() -> &'static str {
@@ -55,3 +59,43 @@ fn replication_facade_exports_runtime_and_dto_types() {
     assert!(type_name::<ReplicationObjectBridge>().contains("ReplicationObjectBridge"));
     assert!(type_name_unsized::<DynReplicationPool>().contains("ReplicationPoolTrait"));
 }
+
+#[test]
+fn replication_facade_exports_app_storage_helper_contracts() {
+    let config = ReplicationConfiguration::default();
+    let target_arns = replication_target_arns(&config);
+    assert!(target_arns.is_empty());
+
+    let source = ReplicationDeleteStateSource {
+        name: String::new(),
+        user_tags: String::new(),
+        version_id: None,
+        delete_marker: false,
+        replica: false,
+    };
+    assert!(delete_replication_state_from_config(&config, &source).is_none());
+    assert!(replication_statuses_map("").is_empty());
+
+    let source_replication_status = Default::default();
+    let source_version_purge_status = Default::default();
+    let input = ReplicationDeleteScheduleInput {
+        replication_request: false,
+        version_id_requested: false,
+        source_delete_marker: false,
+        source_replication_status: &source_replication_status,
+        source_version_purge_status: &source_version_purge_status,
+        deleted_delete_marker_version: false,
+    };
+    assert!(!should_schedule_delete_replication(input));
+
+    assert!(delete_replication_version_id(false, None, false).is_none());
+    assert!(!should_remove_replication_target("arn", false, &target_arns));
+    assert!(!should_use_existing_delete_replication_info(false, false));
+    assert!(!should_use_existing_delete_replication_source(false, false, false));
+    assert!(validate_replication_config_target_arns(std::iter::empty(), &config).is_ok());
+
+    assert_eq!(REPLICATE_INCOMING_DELETE, "replicate:incoming:delete");
+    assert!(type_name::<ReplicateDecision>().contains("ReplicateDecision"));
+    assert!(type_name::<ReplicationState>().contains("ReplicationState"));
+    assert!(type_name::<ReplicationTargetValidationError>().contains("ReplicationTargetValidationError"));
+}

crates/ecstore/tests/storage_api.rs

@@ -22,8 +22,12 @@ pub(crate) mod contract_compat {
 
 pub(crate) mod replication_compat {
     pub(crate) use rustfs_ecstore::api::bucket::replication::{
-        BucketStats, DeletedObjectReplicationInfo, DynReplicationPool, ObjectOpts, ReplicationConfigurationExt,
-        ReplicationObjectBridge, ReplicationStats, ResyncStatusType,
+        BucketStats, DeletedObjectReplicationInfo, DynReplicationPool, ObjectOpts, REPLICATE_INCOMING_DELETE, ReplicateDecision,
+        ReplicationConfigurationExt, ReplicationDeleteScheduleInput, ReplicationDeleteStateSource, ReplicationObjectBridge,
+        ReplicationState, ReplicationStats, ReplicationTargetValidationError, ResyncStatusType,
+        delete_replication_state_from_config, delete_replication_version_id, replication_statuses_map, replication_target_arns,
+        should_remove_replication_target, should_schedule_delete_replication, should_use_existing_delete_replication_info,
+        should_use_existing_delete_replication_source, validate_replication_config_target_arns,
     };
 }
 

crates/scanner/Cargo.toml

@@ -44,7 +44,6 @@ time = { workspace = true }
 chrono = { workspace = true }
 rmp-serde = { workspace = true }
 rustfs-filemeta = { workspace = true }
-rustfs-replication = { workspace = true }
 tokio-util = { workspace = true }
 rustfs-ecstore = { workspace = true }
 rustfs-storage-api = { workspace = true }

docs/architecture/ecstore-module-split-plan.md

@@ -137,8 +137,8 @@ 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;
-- scanner, admin, and storage-owner replication status/DTO consumers import
-  those contracts through the ECStore replication facade;
+- scanner, admin, storage-owner, and app storage replication status/DTO/helper
+  consumers import those contracts through the ECStore replication facade;
 - 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;
@@ -187,10 +187,11 @@ Required contracts before crate movement:
   `crates/replication/src/storage_api.rs` inside `rustfs-replication`.
 - `EcstoreReplicationBoundaryImports`: ECStore-side imports from
   `rustfs-replication` are concentrated in replication `*_boundary.rs` modules.
-- `RuntimeReplicationFacadeConsumers`: scanner, admin, and storage-owner
-  replication status/DTO consumers import through `rustfs-ecstore`, while app
-  storage keeps direct object/delete helper calls inside its storage API
-  boundary.
+- `RuntimeReplicationFacadeConsumers`: scanner, admin, storage-owner, and app
+  storage replication status/DTO/helper consumers import through
+  `rustfs-ecstore`; runtime code under `rustfs/src` does not import
+  `rustfs-replication` directly, and the RustFS runtime/scanner crates do not
+  depend on it.
 - `StorageApiReplicationContracts`: storage-api delete DTO replication
   state/status helpers are concentrated in `crates/storage-api/src/replication.rs`
   until the underlying wire contracts can move without a

rustfs/Cargo.toml

@@ -85,7 +85,6 @@ rustfs-policy = { workspace = true }
 rustfs-protocols = { workspace = true }
 rustfs-protos = { workspace = true }
 rustfs-rio = { workspace = true }
-rustfs-replication = { workspace = true }
 rustfs-s3-types = { workspace = true }
 rustfs-s3-ops = { workspace = true }
 rustfs-security-governance = { workspace = true }

rustfs/src/app/storage_api.rs

@@ -609,18 +609,20 @@ pub(crate) mod bucket {
         use std::sync::Arc;
         use uuid::Uuid;
 
+        use crate::storage::storage_api::ecstore_bucket::replication as replication_contracts;
+
         pub(crate) type DeletedObjectReplicationInfo =
             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;
+        pub(crate) type ReplicateDecision = replication_contracts::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;
+        pub(crate) type ReplicationState = replication_contracts::ReplicationState;
+        pub(crate) type ReplicationStatusType = replication_contracts::ReplicationStatusType;
+        pub(crate) type ReplicationTargetValidationError = replication_contracts::ReplicationTargetValidationError;
+        pub(crate) type VersionPurgeStatusType = replication_contracts::VersionPurgeStatusType;
+        pub(crate) const REPLICATE_INCOMING_DELETE: &str = replication_contracts::REPLICATE_INCOMING_DELETE;
         #[cfg(test)]
-        pub(crate) use rustfs_replication::replication_statuses_map;
+        pub(crate) use replication_contracts::replication_statuses_map;
 
         pub(crate) async fn check_replicate_delete(
             bucket: &str,
@@ -636,7 +638,7 @@ pub(crate) mod bucket {
             replication_source: &crate::storage::storage_api::StorageObjectInfo,
             deleted_delete_marker_version: bool,
         ) -> Option<Uuid> {
-            rustfs_replication::delete_replication_version_id(
+            replication_contracts::delete_replication_version_id(
                 replication_source.delete_marker,
                 replication_source.version_id,
                 deleted_delete_marker_version,
@@ -655,7 +657,7 @@ pub(crate) mod bucket {
                 user_defined,
                 user_tags,
                 status,
-                rustfs_replication::ReplicationType::Object,
+                replication_contracts::ReplicationType::Object,
                 opts,
             );
             ReplicationObjectBridge::must_replicate(bucket, object, mopts).await
@@ -666,7 +668,7 @@ pub(crate) mod bucket {
             store: Arc<crate::storage::storage_api::ECStore>,
             dsc: ReplicateDecision,
         ) {
-            ReplicationObjectBridge::schedule_object(oi, store, dsc, rustfs_replication::ReplicationType::Object).await;
+            ReplicationObjectBridge::schedule_object(oi, store, dsc, replication_contracts::ReplicationType::Object).await;
         }
 
         pub(crate) async fn schedule_replication_delete(dv: DeletedObjectReplicationInfo) {
@@ -678,19 +680,19 @@ pub(crate) mod bucket {
             obj_info: &crate::storage::storage_api::StorageObjectInfo,
             version_id: Option<Uuid>,
             replica: bool,
-        ) -> Option<rustfs_replication::ReplicationState> {
-            let source = rustfs_replication::ReplicationDeleteStateSource {
+        ) -> Option<replication_contracts::ReplicationState> {
+            let source = replication_contracts::ReplicationDeleteStateSource {
                 name: obj_info.name.clone(),
                 user_tags: (*obj_info.user_tags).clone(),
                 version_id,
                 delete_marker: obj_info.delete_marker,
                 replica,
             };
-            rustfs_replication::delete_replication_state_from_config(config, &source)
+            replication_contracts::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)
+            replication_contracts::replication_target_arns(config)
         }
 
         pub(crate) fn should_remove_replication_target(
@@ -698,7 +700,7 @@ pub(crate) mod bucket {
             is_replication_service: bool,
             target_arns: &HashSet<String>,
         ) -> bool {
-            rustfs_replication::should_remove_replication_target(target_arn, is_replication_service, target_arns)
+            replication_contracts::should_remove_replication_target(target_arn, is_replication_service, target_arns)
         }
 
         pub(crate) fn should_schedule_delete_replication(
@@ -706,7 +708,7 @@ pub(crate) mod bucket {
             replication_source: &crate::storage::storage_api::StorageObjectInfo,
             deleted_delete_marker_version: bool,
         ) -> bool {
-            rustfs_replication::should_schedule_delete_replication(rustfs_replication::ReplicationDeleteScheduleInput {
+            replication_contracts::should_schedule_delete_replication(replication_contracts::ReplicationDeleteScheduleInput {
                 replication_request: opts.replication_request,
                 version_id_requested: opts.version_id.is_some(),
                 source_delete_marker: replication_source.delete_marker,
@@ -719,7 +721,7 @@ pub(crate) mod bucket {
         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)
+            replication_contracts::should_use_existing_delete_replication_info(opts.version_id.is_some(), opts.delete_marker)
         }
 
         pub(crate) fn should_use_existing_delete_replication_source(
@@ -727,7 +729,7 @@ pub(crate) mod bucket {
             deleted_delete_marker: bool,
             has_existing_info: bool,
         ) -> bool {
-            rustfs_replication::should_use_existing_delete_replication_source(
+            replication_contracts::should_use_existing_delete_replication_source(
                 replication_request,
                 deleted_delete_marker,
                 has_existing_info,
@@ -738,7 +740,7 @@ pub(crate) mod bucket {
             configured_arns: impl Iterator<Item = &'a str>,
             config: &s3s::dto::ReplicationConfiguration,
         ) -> Result<(), ReplicationTargetValidationError> {
-            rustfs_replication::validate_replication_config_target_arns(configured_arns, config)
+            replication_contracts::validate_replication_config_target_arns(configured_arns, config)
         }
     }
 

scripts/check_architecture_migration_rules.sh

@@ -210,6 +210,7 @@ REPLICATION_FACADE_WILDCARD_EXPORT_HITS_FILE="${TMP_DIR}/replication_facade_wild
 ECSTORE_REPLICATION_BOUNDARY_BYPASS_HITS_FILE="${TMP_DIR}/ecstore_replication_boundary_bypass_hits.txt"
 SCANNER_REPLICATION_FACADE_BYPASS_HITS_FILE="${TMP_DIR}/scanner_replication_facade_bypass_hits.txt"
 RUSTFS_REPLICATION_FACADE_BYPASS_HITS_FILE="${TMP_DIR}/rustfs_replication_facade_bypass_hits.txt"
+RUNTIME_REPLICATION_DEPENDENCY_BYPASS_HITS_FILE="${TMP_DIR}/runtime_replication_dependency_bypass_hits.txt"
 REPLICATION_CRATE_FILEMETA_BYPASS_HITS_FILE="${TMP_DIR}/replication_crate_filemeta_bypass_hits.txt"
 REPLICATION_CRATE_STORAGE_API_BYPASS_HITS_FILE="${TMP_DIR}/replication_crate_storage_api_bypass_hits.txt"
 REPLICATION_CONFIG_RULE_CONTRACT_BACKSLIDE_HITS_FILE="${TMP_DIR}/replication_config_rule_contract_backslide_hits.txt"
@@ -2643,12 +2644,22 @@ fi
   cd "$ROOT_DIR"
   rg -n --with-filename 'rustfs_replication::|use\s+rustfs_replication\b' \
     rustfs/src \
-    --glob '*.rs' |
-    rg -v '^rustfs/src/app/storage_api\.rs:' || true
+    --glob '*.rs' || true
 ) >"$RUSTFS_REPLICATION_FACADE_BYPASS_HITS_FILE"
 
 if [[ -s "$RUSTFS_REPLICATION_FACADE_BYPASS_HITS_FILE" ]]; then
-  report_failure "RustFS runtime replication contracts must stay behind storage API facades: $(paste -sd '; ' "$RUSTFS_REPLICATION_FACADE_BYPASS_HITS_FILE")"
+  report_failure "RustFS runtime replication contracts must come through storage API / ECStore replication facades: $(paste -sd '; ' "$RUSTFS_REPLICATION_FACADE_BYPASS_HITS_FILE")"
+fi
+
+(
+  cd "$ROOT_DIR"
+  rg -n --with-filename '^rustfs-replication\b' \
+    rustfs/Cargo.toml \
+    crates/scanner/Cargo.toml || true
+) >"$RUNTIME_REPLICATION_DEPENDENCY_BYPASS_HITS_FILE"
+
+if [[ -s "$RUNTIME_REPLICATION_DEPENDENCY_BYPASS_HITS_FILE" ]]; then
+  report_failure "RustFS runtime and scanner crates must not depend on rustfs-replication directly: $(paste -sd '; ' "$RUNTIME_REPLICATION_DEPENDENCY_BYPASS_HITS_FILE")"
 fi
 
 (

scripts/s3-tests/run.sh

@@ -744,6 +744,29 @@ envsubst < "${TEMPLATE_PATH}" > "${CONF_OUTPUT_PATH}" || {
 log_info "Provisioning s3-tests alt user..."
 
 # Helper function to install Python packages with fallback for externally-managed environments
+ensure_python_pip() {
+    if python3 -m pip --version >/dev/null 2>&1; then
+        return 0
+    fi
+
+    log_info "Installing pip for Python package setup..."
+    if python3 -m ensurepip --upgrade >/dev/null 2>&1; then
+        return 0
+    fi
+
+    if command -v apt-get >/dev/null 2>&1; then
+        sudo apt-get update && sudo apt-get install -y python3-pip || {
+            log_warn "Failed to install python3-pip via apt-get"
+        }
+    elif command -v brew >/dev/null 2>&1; then
+        brew install python || {
+            log_warn "Failed to install Python via brew"
+        }
+    fi
+
+    python3 -m pip --version >/dev/null 2>&1
+}
+
 install_python_package() {
     local package=$1
     local error_output
@@ -759,6 +782,11 @@ install_python_package() {
         log_warn "Failed to install ${package} with uv, falling back to python3 -m pip"
     fi
 
+    ensure_python_pip || {
+        log_error "python3 -m pip is unavailable"
+        return 1
+    }
+
     # Try --user first (works on most Linux systems)
     error_output=$(python3 -m pip install --user --upgrade pip "${package}" 2>&1)
     if [ $? -eq 0 ]; then