diff --git a/crates/ecstore/src/bucket/lifecycle/replication_sink.rs b/crates/ecstore/src/bucket/lifecycle/replication_sink.rs index 18f1f0057..1c5bbea77 100644 --- a/crates/ecstore/src/bucket/lifecycle/replication_sink.rs +++ b/crates/ecstore/src/bucket/lifecycle/replication_sink.rs @@ -12,26 +12,22 @@ // See the License for the specific language governing permissions and // limitations under the License. -use rustfs_filemeta::{REPLICATE_INCOMING_DELETE, ReplicateDecision}; +use rustfs_filemeta::ReplicateDecision; use s3s::dto::ReplicationConfiguration; use crate::bucket::lifecycle::lifecycle::ObjectOpts; -use crate::bucket::replication::{self, DeletedObjectReplicationInfo, ReplicationConfig, ReplicationConfigurationExt as _}; +use crate::bucket::replication::{ReplicationLifecycleBridge, ReplicationLifecycleConfig}; use crate::object_api::{ObjectInfo, ObjectOptions}; use crate::storage_api_contracts::object::{DeletedObject, ObjectToDelete}; -pub(crate) type LifecycleReplicationConfig = ReplicationConfig; +pub(crate) type LifecycleReplicationConfig = ReplicationLifecycleConfig; pub(crate) fn new_replication_config(config: ReplicationConfiguration) -> LifecycleReplicationConfig { - ReplicationConfig::new(Some(config), None) + ReplicationLifecycleBridge::new_config(config) } pub(crate) fn has_pending_version_purge(config: &LifecycleReplicationConfig, obj: &ObjectOpts) -> bool { - config - .config - .as_ref() - .is_some_and(|config| config.has_active_rules(obj.name.as_str(), true)) - && !obj.version_purge_status.is_empty() + ReplicationLifecycleBridge::has_pending_version_purge(config, obj.name.as_str(), !obj.version_purge_status.is_empty()) } pub(crate) async fn check_delete_replication( @@ -40,76 +36,9 @@ pub(crate) async fn check_delete_replication( source: &ObjectInfo, opts: &ObjectOptions, ) -> ReplicateDecision { - replication::check_replicate_delete(bucket, &object, source, opts, None).await + ReplicationLifecycleBridge::check_delete_replication(bucket, &object, source, opts).await } pub(crate) async fn schedule_delete(bucket: String, delete_object: DeletedObject) { - replication::schedule_replication_delete(DeletedObjectReplicationInfo { - delete_object, - bucket, - event_type: REPLICATE_INCOMING_DELETE.to_string(), - ..Default::default() - }) - .await; -} - -#[cfg(test)] -mod tests { - use std::collections::HashMap; - - use rustfs_filemeta::{ReplicationStatusType, VersionPurgeStatusType}; - use s3s::dto::{Destination, ReplicationRule, ReplicationRuleStatus}; - - use super::*; - - fn replication_rule() -> ReplicationRule { - ReplicationRule { - delete_marker_replication: None, - delete_replication: None, - destination: Destination { - bucket: "arn:aws:s3:::target-bucket".to_string(), - ..Default::default() - }, - existing_object_replication: None, - filter: None, - id: Some("rule".to_string()), - prefix: Some(String::new()), - priority: Some(1), - source_selection_criteria: None, - status: ReplicationRuleStatus::from_static(ReplicationRuleStatus::ENABLED), - } - } - - fn object_opts(version_purge_status: VersionPurgeStatusType) -> ObjectOpts { - ObjectOpts { - name: "logs/object".to_string(), - user_tags: String::new(), - mod_time: None, - size: 0, - version_id: None, - is_latest: true, - delete_marker: false, - num_versions: 1, - successor_mod_time: None, - transition_status: String::new(), - restore_ongoing: false, - restore_expires: None, - versioned: true, - version_suspended: false, - user_defined: HashMap::new(), - version_purge_status, - replication_status: ReplicationStatusType::default(), - } - } - - #[test] - fn has_pending_version_purge_preserves_replication_active_rule_behavior() { - let config = new_replication_config(ReplicationConfiguration { - role: String::new(), - rules: vec![replication_rule()], - }); - - assert!(has_pending_version_purge(&config, &object_opts(VersionPurgeStatusType::Pending))); - assert!(!has_pending_version_purge(&config, &object_opts(VersionPurgeStatusType::default()))); - } + ReplicationLifecycleBridge::schedule_delete(bucket, delete_object).await; } diff --git a/crates/ecstore/src/bucket/replication/README.md b/crates/ecstore/src/bucket/replication/README.md index 239a3a13b..2f75e1852 100644 --- a/crates/ecstore/src/bucket/replication/README.md +++ b/crates/ecstore/src/bucket/replication/README.md @@ -14,6 +14,8 @@ and lifecycle/heal scheduling paths. | `replication_pool.rs` | Replication queue, worker pool, MRF persistence, bucket stats, and delete/object scheduling. | Depends on bucket target sys, bucket metadata sys, metadata paths, and file metadata replication contracts through local boundaries, config storage, storage contracts through the replication storage boundary, runtime sources, and notification state. | | `replication_resyncer.rs` | Object replication, delete replication, resync, MRF encode/decode, target calls, and multipart target upload paths. | Depends on target calls and target config types through the replication target boundary, metadata paths and metadata systems through the replication metadata boundary, file metadata replication contracts through the filemeta boundary, error contracts through the error boundary, versioning systems, storage contracts through the replication storage boundary, config-derived storage class labels through the config store, runtime sources, notification events and local event host selection through the event sink, bandwidth reader wrapping, and SetDisks lock timing. | | `replication_state.rs` | Replication queue/stat state and worker accounting. | Reads runtime sources, file metadata replication contracts, error contracts, and bucket monitor handles through local boundaries, and owns shared replication pool/stat state. | +| `replication_lifecycle_bridge.rs` | Lifecycle-originated delete replication admission and version-purge state construction. | Depends on replication config/rule matching, delete-replication decisions, and replication delete scheduling through a local contract type. | +| `replication_scanner_bridge.rs` | Scanner-originated replication heal admission. | Keeps scanner-facing heal queueing behind a local contract type instead of exporting the internal queue function directly. | | `rule.rs` | Rule evaluation helpers for object replication options. | Depends on ECStore replication object option types. | | `mod.rs` | Compatibility re-export facade for the current ECStore owner. | Must stay stable until downstream scanner, lifecycle, heal, and metrics paths compile through replacement contracts. | @@ -35,7 +37,8 @@ and lifecycle/heal scheduling paths. | `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. | +| `ReplicationLifecycleBridge` | Lifecycle-originated delete and version-purge scheduling. | Lifecycle delete paths call the bridge contract in `replication_lifecycle_bridge.rs` instead of constructing replication delete work directly. | +| `ReplicationScannerBridge` | Scanner-originated replication heal scheduling. | Scanner heal paths call the bridge contract in `replication_scanner_bridge.rs` instead of importing the internal queue function directly. | ## Migration Rules diff --git a/crates/ecstore/src/bucket/replication/mod.rs b/crates/ecstore/src/bucket/replication/mod.rs index af09536ed..345235dfd 100644 --- a/crates/ecstore/src/bucket/replication/mod.rs +++ b/crates/ecstore/src/bucket/replication/mod.rs @@ -19,11 +19,13 @@ mod replication_config_store; mod replication_error_boundary; mod replication_event_sink; mod replication_filemeta_boundary; +mod replication_lifecycle_bridge; mod replication_lock_boundary; mod replication_metadata_boundary; mod replication_msgp_boundary; mod replication_pool; mod replication_resyncer; +mod replication_scanner_bridge; mod replication_state; mod replication_storage_boundary; mod replication_tagging_boundary; @@ -34,8 +36,10 @@ mod runtime_boundary; pub use config::*; pub use datatypes::*; +pub(crate) use replication_lifecycle_bridge::{ReplicationLifecycleBridge, ReplicationLifecycleConfig}; pub use replication_pool::*; pub use replication_resyncer::*; +pub use replication_scanner_bridge::ReplicationScannerBridge; pub use replication_state::{BucketStats, ReplicationStats}; pub use replication_storage_boundary::{ReplicationObjectIO, ReplicationStorage}; pub use rule::*; diff --git a/crates/ecstore/src/bucket/replication/replication_filemeta_boundary.rs b/crates/ecstore/src/bucket/replication/replication_filemeta_boundary.rs index 2b543c445..d51a6a26c 100644 --- a/crates/ecstore/src/bucket/replication/replication_filemeta_boundary.rs +++ b/crates/ecstore/src/bucket/replication/replication_filemeta_boundary.rs @@ -14,8 +14,8 @@ pub(crate) use rustfs_filemeta::{ MrfOpKind, MrfReplicateEntry, REPLICATE_EXISTING, REPLICATE_EXISTING_DELETE, REPLICATE_HEAL, REPLICATE_HEAL_DELETE, - ReplicateDecision, ReplicateObjectInfo, ReplicateTargetDecision, ReplicatedInfos, ReplicatedTargetInfo, ReplicationAction, - ReplicationState, ReplicationStatusType, ReplicationType, ReplicationWorkerOperation, ResyncDecision, ResyncTargetDecision, - VersionPurgeStatusType, get_replication_state, parse_replicate_decision, replication_statuses_map, target_reset_header, - version_purge_statuses_map, + REPLICATE_INCOMING_DELETE, ReplicateDecision, ReplicateObjectInfo, ReplicateTargetDecision, ReplicatedInfos, + ReplicatedTargetInfo, ReplicationAction, ReplicationState, ReplicationStatusType, ReplicationType, + ReplicationWorkerOperation, ResyncDecision, ResyncTargetDecision, VersionPurgeStatusType, get_replication_state, + parse_replicate_decision, replication_statuses_map, target_reset_header, version_purge_statuses_map, }; diff --git a/crates/ecstore/src/bucket/replication/replication_lifecycle_bridge.rs b/crates/ecstore/src/bucket/replication/replication_lifecycle_bridge.rs new file mode 100644 index 000000000..ebf1310af --- /dev/null +++ b/crates/ecstore/src/bucket/replication/replication_lifecycle_bridge.rs @@ -0,0 +1,123 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use s3s::dto::ReplicationConfiguration; + +use super::config::ReplicationConfigurationExt as _; +use super::replication_filemeta_boundary::{ + REPLICATE_INCOMING_DELETE, ReplicateDecision, ReplicationState, version_purge_statuses_map, +}; +use super::replication_resyncer::{DeletedObjectReplicationInfo, ReplicationConfig, check_replicate_delete}; +use super::replication_storage_boundary::{DeletedObject, ObjectInfo, ObjectOptions, ObjectToDelete}; + +pub(crate) type ReplicationLifecycleConfig = ReplicationConfig; + +pub(crate) struct ReplicationLifecycleBridge; + +impl ReplicationLifecycleBridge { + pub(crate) fn new_config(config: ReplicationConfiguration) -> ReplicationLifecycleConfig { + ReplicationConfig::new(Some(config), None) + } + + pub(crate) fn has_pending_version_purge( + config: &ReplicationLifecycleConfig, + object_name: &str, + version_purge_pending: bool, + ) -> bool { + version_purge_pending + && config + .config + .as_ref() + .is_some_and(|config| config.has_active_rules(object_name, true)) + } + + pub(crate) async fn check_delete_replication( + bucket: &str, + object: &ObjectToDelete, + source: &ObjectInfo, + opts: &ObjectOptions, + ) -> ReplicateDecision { + check_replicate_delete(bucket, object, source, opts, None).await + } + + pub(crate) fn version_delete_replication_state(decision: &ReplicateDecision) -> ReplicationState { + let pending_status = decision.pending_status(); + ReplicationState { + replicate_decision_str: decision.to_string(), + version_purge_status_internal: pending_status.clone(), + purge_targets: version_purge_statuses_map(pending_status.as_deref().unwrap_or_default()), + ..Default::default() + } + } + + pub(crate) async fn schedule_delete(bucket: String, delete_object: DeletedObject) { + super::replication_pool::schedule_replication_delete(DeletedObjectReplicationInfo { + delete_object, + bucket, + event_type: REPLICATE_INCOMING_DELETE.to_string(), + ..Default::default() + }) + .await; + } +} + +#[cfg(test)] +mod tests { + use s3s::dto::{Destination, ReplicationRule, ReplicationRuleStatus}; + + use super::super::replication_filemeta_boundary::{ReplicateTargetDecision, VersionPurgeStatusType}; + use super::*; + + fn replication_rule() -> ReplicationRule { + ReplicationRule { + delete_marker_replication: None, + delete_replication: None, + destination: Destination { + bucket: "arn:aws:s3:::target-bucket".to_string(), + ..Default::default() + }, + existing_object_replication: None, + filter: None, + id: Some("rule".to_string()), + prefix: Some(String::new()), + priority: Some(1), + source_selection_criteria: None, + status: ReplicationRuleStatus::from_static(ReplicationRuleStatus::ENABLED), + } + } + + #[test] + fn has_pending_version_purge_preserves_replication_active_rule_behavior() { + let config = ReplicationLifecycleBridge::new_config(ReplicationConfiguration { + role: String::new(), + rules: vec![replication_rule()], + }); + + assert!(ReplicationLifecycleBridge::has_pending_version_purge(&config, "logs/object", true)); + assert!(!ReplicationLifecycleBridge::has_pending_version_purge(&config, "logs/object", false)); + } + + #[test] + fn version_delete_replication_state_tracks_pending_purge_targets() { + let target = ReplicateTargetDecision::new("arn:aws:s3:::target".to_string(), true, false); + let mut decision = ReplicateDecision::new(); + decision.set(target); + + let state = ReplicationLifecycleBridge::version_delete_replication_state(&decision); + + assert_eq!(state.version_purge_status_internal.as_deref(), Some("arn:aws:s3:::target=PENDING;")); + assert!(state.purge_targets.contains_key("arn:aws:s3:::target")); + assert_eq!(state.purge_targets["arn:aws:s3:::target"], VersionPurgeStatusType::Pending); + } +} diff --git a/crates/ecstore/src/bucket/replication/replication_pool.rs b/crates/ecstore/src/bucket/replication/replication_pool.rs index fd30bc223..022c4c3e6 100644 --- a/crates/ecstore/src/bucket/replication/replication_pool.rs +++ b/crates/ecstore/src/bucket/replication/replication_pool.rs @@ -1528,7 +1528,7 @@ pub async fn queue_replication_heal(bucket: &str, oi: ObjectInfo, retry_count: u /// queue_replication_heal_internal enqueues objects that failed replication OR eligible for resyncing through /// an ongoing resync operation or via existing objects replication configuration setting. -pub async fn queue_replication_heal_internal( +pub(crate) async fn queue_replication_heal_internal( _bucket: &str, oi: ObjectInfo, rcfg: ReplicationConfig, diff --git a/crates/ecstore/src/bucket/replication/replication_scanner_bridge.rs b/crates/ecstore/src/bucket/replication/replication_scanner_bridge.rs new file mode 100644 index 000000000..9b6d0989e --- /dev/null +++ b/crates/ecstore/src/bucket/replication/replication_scanner_bridge.rs @@ -0,0 +1,43 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use super::replication_pool::{ReplicationHealQueueResult, queue_replication_heal_internal}; +use super::replication_resyncer::ReplicationConfig; +use super::replication_storage_boundary::ObjectInfo; + +pub struct ReplicationScannerBridge; + +impl ReplicationScannerBridge { + pub async fn queue_heal( + bucket: &str, + oi: ObjectInfo, + rcfg: ReplicationConfig, + retry_count: u32, + ) -> ReplicationHealQueueResult { + queue_replication_heal_internal(bucket, oi, rcfg, retry_count).await + } +} + +#[cfg(test)] +mod tests { + use super::super::replication_pool::ReplicationQueueAdmission; + use super::*; + + #[tokio::test] + async fn scanner_bridge_preserves_empty_config_skip_behavior() { + let result = ReplicationScannerBridge::queue_heal("bucket", ObjectInfo::default(), ReplicationConfig::default(), 0).await; + + assert_eq!(result.admission, ReplicationQueueAdmission::Skipped); + } +} diff --git a/crates/ecstore/src/client/object_handlers_common.rs b/crates/ecstore/src/client/object_handlers_common.rs index f4660430d..7a8e61e7e 100644 --- a/crates/ecstore/src/client/object_handlers_common.rs +++ b/crates/ecstore/src/client/object_handlers_common.rs @@ -21,27 +21,15 @@ const EVENT_LIFECYCLE_CLEANUP_SKIPPED: &str = "lifecycle_cleanup_skipped"; const EVENT_LIFECYCLE_CLEANUP_FAILED: &str = "lifecycle_cleanup_failed"; use crate::bucket::lifecycle::lifecycle; -use crate::bucket::replication::{DeletedObjectReplicationInfo, check_replicate_delete, schedule_replication_delete}; +use crate::bucket::replication::ReplicationLifecycleBridge; use crate::bucket::versioning::VersioningApi; use crate::bucket::versioning_sys::BucketVersioningSys; use crate::object_api::ObjectOptions; use crate::storage_api_contracts::object::{ObjectOperations as _, ObjectToDelete}; use crate::store::ECStore; -use rustfs_filemeta::{REPLICATE_INCOMING_DELETE, ReplicationState, version_purge_statuses_map}; +use rustfs_filemeta::ReplicationState; use rustfs_lock::MAX_DELETE_LIST; -fn lifecycle_version_delete_replication_state( - replicate_decision_str: String, - pending_status: Option, -) -> ReplicationState { - ReplicationState { - replicate_decision_str, - version_purge_status_internal: pending_status.clone(), - purge_targets: version_purge_statuses_map(pending_status.as_deref().unwrap_or_default()), - ..Default::default() - } -} - pub async fn delete_object_versions(api: &Arc, bucket: &str, to_del: &[ObjectToDelete], _lc_event: lifecycle::Event) { let version_suspended = match BucketVersioningSys::get(bucket).await { Ok(vc) => vc.suspended(), @@ -79,9 +67,9 @@ pub async fn delete_object_versions(api: &Arc, bucket: &str, to_del: &[ }; let candidate = match api.get_object_info(bucket, &object.object_name, &opts).await { Ok(info) => { - let dsc = check_replicate_delete(bucket, object, &info, &opts, None).await; + let dsc = ReplicationLifecycleBridge::check_delete_replication(bucket, object, &info, &opts).await; dsc.replicate_any() - .then(|| lifecycle_version_delete_replication_state(dsc.to_string(), dsc.pending_status())) + .then(|| ReplicationLifecycleBridge::version_delete_replication_state(&dsc)) } Err(err) => { debug!( @@ -120,13 +108,7 @@ pub async fn delete_object_versions(api: &Arc, bucket: &str, to_del: &[ continue; }; deleted_obj.replication_state = Some(replication_state); - schedule_replication_delete(DeletedObjectReplicationInfo { - delete_object: deleted_obj.clone(), - bucket: bucket.to_string(), - event_type: REPLICATE_INCOMING_DELETE.to_string(), - ..Default::default() - }) - .await; + ReplicationLifecycleBridge::schedule_delete(bucket.to_string(), deleted_obj.clone()).await; } for (i, err) in errors.iter().enumerate() { @@ -154,20 +136,3 @@ pub async fn delete_object_versions(api: &Arc, bucket: &str, to_del: &[ } } } - -#[cfg(test)] -mod tests { - use super::lifecycle_version_delete_replication_state; - - #[test] - fn lifecycle_version_delete_replication_state_tracks_pending_purge_targets() { - let state = lifecycle_version_delete_replication_state( - "arn:aws:s3:::target=true;false;arn:aws:s3:::target;".to_string(), - Some("arn:aws:s3:::target=PENDING;".to_string()), - ); - - assert_eq!(state.version_purge_status_internal.as_deref(), Some("arn:aws:s3:::target=PENDING;")); - assert!(state.purge_targets.contains_key("arn:aws:s3:::target")); - assert_eq!(state.replicate_decision_str, "arn:aws:s3:::target=true;false;arn:aws:s3:::target;"); - } -} diff --git a/crates/scanner/src/lib.rs b/crates/scanner/src/lib.rs index a0346d020..3c32e075b 100644 --- a/crates/scanner/src/lib.rs +++ b/crates/scanner/src/lib.rs @@ -30,13 +30,13 @@ use storage_api::owner::{ EcstoreDiskAPI, EcstoreDiskBytes, EcstoreDiskError, EcstoreDiskInfo, EcstoreDiskInfoOptions, EcstoreDiskLocation, EcstoreDiskResult, EcstoreErrorType, EcstoreEvaluator, EcstoreEvent, EcstoreLcEventSrc, EcstoreLifecycle, EcstoreListPathRawOptions, EcstoreObjectOpts, EcstoreReplicationConfig, EcstoreReplicationConfigurationExt, - EcstoreReplicationHealQueueResult, EcstoreReplicationQueueAdmission, EcstoreResultType, EcstoreScanGuard, EcstoreSetDisks, - EcstoreStorageError, EcstoreStore, EcstoreTierConfig, EcstoreVersioningApi, HTTPRangeSpec, ObjectIO, ObjectOperations, - ObjectToDelete, ecstore_apply_expiry_rule, ecstore_apply_transition_rule, ecstore_expiry_state_handle, - ecstore_get_global_tier_config_mgr, ecstore_get_lifecycle_config, ecstore_get_object_lock_config, - ecstore_get_replication_config, ecstore_is_erasure, ecstore_is_erasure_sd, ecstore_is_reserved_or_invalid_bucket, - ecstore_list_path_raw, ecstore_path2_bucket_object, ecstore_path2_bucket_object_with_base_path, - ecstore_queue_replication_heal_internal, ecstore_read_config, ecstore_replace_bucket_usage_memory_from_info, + EcstoreReplicationHealQueueResult, EcstoreReplicationQueueAdmission, EcstoreReplicationScannerBridge, EcstoreResultType, + EcstoreScanGuard, EcstoreSetDisks, EcstoreStorageError, EcstoreStore, EcstoreTierConfig, EcstoreVersioningApi, HTTPRangeSpec, + ObjectIO, ObjectOperations, ObjectToDelete, ecstore_apply_expiry_rule, ecstore_apply_transition_rule, + ecstore_expiry_state_handle, ecstore_get_global_tier_config_mgr, ecstore_get_lifecycle_config, + ecstore_get_object_lock_config, ecstore_get_replication_config, ecstore_is_erasure, ecstore_is_erasure_sd, + ecstore_is_reserved_or_invalid_bucket, ecstore_list_path_raw, ecstore_path2_bucket_object, + ecstore_path2_bucket_object_with_base_path, ecstore_read_config, ecstore_replace_bucket_usage_memory_from_info, ecstore_resolve_object_store_handle, ecstore_save_config, }; #[cfg(test)] @@ -304,13 +304,13 @@ pub(crate) async fn enqueue_runtime_newer_noncurrent( .await } -pub(crate) async fn queue_replication_heal_internal( +pub(crate) async fn queue_replication_heal( bucket: &str, oi: ScannerObjectInfo, rcfg: ReplicationConfig, retry_count: u32, ) -> ReplicationHealQueueResult { - ecstore_queue_replication_heal_internal(bucket, oi, rcfg, retry_count).await + EcstoreReplicationScannerBridge::queue_heal(bucket, oi, rcfg, retry_count).await } pub(crate) fn resolve_scanner_object_store_handle() -> Option> { diff --git a/crates/scanner/src/scanner_folder.rs b/crates/scanner/src/scanner_folder.rs index 334fde9a6..20ff08542 100644 --- a/crates/scanner/src/scanner_folder.rs +++ b/crates/scanner/src/scanner_folder.rs @@ -57,7 +57,7 @@ use crate::{ ReplicationConfig, ReplicationQueueAdmission, ScannerDiskExt as _, ScannerLifecycleConfigExt as _, ScannerReplicationConfigExt as _, ScannerVersioningConfigExt as _, StorageError, apply_expiry_rule, apply_transition_rule, enqueue_runtime_newer_noncurrent, is_reserved_or_invalid_bucket, list_path_raw, path2_bucket_object, - path2_bucket_object_with_base_path, queue_replication_heal_internal, scanner_is_erasure, + path2_bucket_object_with_base_path, queue_replication_heal, scanner_is_erasure, }; use crate::{ScannerObjectInfo as ObjectInfo, ScannerObjectToDelete as ObjectToDelete}; @@ -1009,7 +1009,7 @@ impl ScannerItem { }; let done_replication = Metrics::time(Metric::CheckReplication); - let replication_result = queue_replication_heal_internal(&oi.bucket, oi.clone(), (*replication).clone(), 0).await; + let replication_result = queue_replication_heal(&oi.bucket, oi.clone(), (*replication).clone(), 0).await; done_replication(); let roi = replication_result.object_info; record_scanner_replication_admission(global_metrics(), &roi, replication_result.admission); diff --git a/crates/scanner/src/storage_api.rs b/crates/scanner/src/storage_api.rs index e83a47a16..030d1adf9 100644 --- a/crates/scanner/src/storage_api.rs +++ b/crates/scanner/src/storage_api.rs @@ -29,8 +29,7 @@ pub(crate) use rustfs_ecstore::api::bucket::metadata_sys::{ pub(crate) use rustfs_ecstore::api::bucket::replication::{ ReplicationConfig as EcstoreReplicationConfig, ReplicationConfigurationExt as EcstoreReplicationConfigurationExt, ReplicationHealQueueResult as EcstoreReplicationHealQueueResult, - ReplicationQueueAdmission as EcstoreReplicationQueueAdmission, - queue_replication_heal_internal as ecstore_queue_replication_heal_internal, + ReplicationQueueAdmission as EcstoreReplicationQueueAdmission, ReplicationScannerBridge as EcstoreReplicationScannerBridge, }; pub(crate) use rustfs_ecstore::api::bucket::versioning::VersioningApi as EcstoreVersioningApi; pub(crate) use rustfs_ecstore::api::bucket::versioning_sys::BucketVersioningSys as EcstoreBucketVersioningSys; @@ -84,12 +83,12 @@ pub(crate) mod owner { EcstoreDiskLocation, EcstoreDiskResult, EcstoreErrorType, EcstoreEvaluator, EcstoreEvent, EcstoreLcEventSrc, EcstoreLifecycle, EcstoreListPathRawOptions, EcstoreObjectOpts, EcstoreReplicationConfig, EcstoreReplicationConfigurationExt, EcstoreReplicationHealQueueResult, EcstoreReplicationQueueAdmission, - EcstoreResultType, EcstoreScanGuard, EcstoreSetDisks, EcstoreStorageError, EcstoreStore, EcstoreTierConfig, - EcstoreVersioningApi, ecstore_apply_expiry_rule, ecstore_apply_transition_rule, ecstore_expiry_state_handle, - ecstore_get_global_tier_config_mgr, ecstore_get_lifecycle_config, ecstore_get_object_lock_config, - ecstore_get_replication_config, ecstore_is_erasure, ecstore_is_erasure_sd, ecstore_is_reserved_or_invalid_bucket, - ecstore_list_path_raw, ecstore_path2_bucket_object, ecstore_path2_bucket_object_with_base_path, - ecstore_queue_replication_heal_internal, ecstore_read_config, ecstore_replace_bucket_usage_memory_from_info, + EcstoreReplicationScannerBridge, EcstoreResultType, EcstoreScanGuard, EcstoreSetDisks, EcstoreStorageError, EcstoreStore, + EcstoreTierConfig, EcstoreVersioningApi, ecstore_apply_expiry_rule, ecstore_apply_transition_rule, + ecstore_expiry_state_handle, ecstore_get_global_tier_config_mgr, ecstore_get_lifecycle_config, + ecstore_get_object_lock_config, ecstore_get_replication_config, ecstore_is_erasure, ecstore_is_erasure_sd, + ecstore_is_reserved_or_invalid_bucket, ecstore_list_path_raw, ecstore_path2_bucket_object, + ecstore_path2_bucket_object_with_base_path, ecstore_read_config, ecstore_replace_bucket_usage_memory_from_info, ecstore_resolve_object_store_handle, ecstore_save_config, }; diff --git a/docs/architecture/ecstore-module-split-plan.md b/docs/architecture/ecstore-module-split-plan.md index 31a6f8b70..c6992d8e7 100644 --- a/docs/architecture/ecstore-module-split-plan.md +++ b/docs/architecture/ecstore-module-split-plan.md @@ -59,7 +59,8 @@ Current coupling: IDs, and local node names; - stale multipart cleanup depends on `SetDisks` internals and bucket metadata through the lifecycle metadata boundary; -- lifecycle expiry schedules bucket replication delete work directly; +- lifecycle expiry schedules bucket replication delete work through the + replication lifecycle bridge contract; - lifecycle evaluation shares S3 DTOs, object metadata, object lock, replication config reads through lifecycle-local boundaries, scanner metrics, notification/audit side effects, and tier services. @@ -112,8 +113,9 @@ Current coupling: - resync and delete replication paths call metadata paths through the metadata boundary, while bucket target system access, target config types, and target operation types are concentrated behind the replication target boundary; -- lifecycle and heal paths schedule replication work through the current ECStore - module; +- lifecycle delete paths schedule replication work through + `ReplicationLifecycleBridge`, while scanner heal paths schedule replication + work through `ReplicationScannerBridge`; - global replication pool/stat initialization still lives with ECStore runtime compatibility state; - modules inside `bucket/replication` use local relative paths rather than the @@ -167,7 +169,11 @@ Required contracts before crate movement: - `ReplicationEventSink`: notification/audit events for skipped, failed, and completed replication operations, including local event host selection. - `ReplicationLifecycleBridge`: lifecycle-originated delete and version-purge - scheduling without importing lifecycle internals. + scheduling is exposed through the contract type in + `crates/ecstore/src/bucket/replication/replication_lifecycle_bridge.rs`. +- `ReplicationScannerBridge`: scanner-originated replication heal scheduling is + exposed through the contract type in + `crates/ecstore/src/bucket/replication/replication_scanner_bridge.rs`. First safe PR: diff --git a/scripts/check_architecture_migration_rules.sh b/scripts/check_architecture_migration_rules.sh index eab6ffbc7..825cfc82a 100755 --- a/scripts/check_architecture_migration_rules.sh +++ b/scripts/check_architecture_migration_rules.sh @@ -209,6 +209,7 @@ REPLICATION_LOCK_BOUNDARY_BYPASS_HITS_FILE="${TMP_DIR}/replication_lock_boundary REPLICATION_METADATA_BOUNDARY_BYPASS_HITS_FILE="${TMP_DIR}/replication_metadata_boundary_bypass_hits.txt" REPLICATION_MSGP_BOUNDARY_BYPASS_HITS_FILE="${TMP_DIR}/replication_msgp_boundary_bypass_hits.txt" REPLICATION_RUNTIME_TYPE_BYPASS_HITS_FILE="${TMP_DIR}/replication_runtime_type_bypass_hits.txt" +REPLICATION_SCANNER_BRIDGE_BYPASS_HITS_FILE="${TMP_DIR}/replication_scanner_bridge_bypass_hits.txt" REPLICATION_STORAGE_BOUNDARY_BYPASS_HITS_FILE="${TMP_DIR}/replication_storage_boundary_bypass_hits.txt" REPLICATION_TARGET_BOUNDARY_BYPASS_HITS_FILE="${TMP_DIR}/replication_target_boundary_bypass_hits.txt" REPLICATION_TAGGING_BOUNDARY_BYPASS_HITS_FILE="${TMP_DIR}/replication_tagging_boundary_bypass_hits.txt" @@ -2669,6 +2670,17 @@ if [[ -s "$REPLICATION_RUNTIME_SOURCE_BYPASS_HITS_FILE" ]]; then report_failure "replication runtime-source access must stay behind replication runtime boundary: $(paste -sd '; ' "$REPLICATION_RUNTIME_SOURCE_BYPASS_HITS_FILE")" fi +( + cd "$ROOT_DIR" + rg -n --with-filename 'queue_replication_heal_internal' \ + crates/scanner/src crates/scanner/tests \ + --glob '*.rs' || true +) >"$REPLICATION_SCANNER_BRIDGE_BYPASS_HITS_FILE" + +if [[ -s "$REPLICATION_SCANNER_BRIDGE_BYPASS_HITS_FILE" ]]; then + report_failure "scanner replication heal queueing must stay behind ReplicationScannerBridge: $(paste -sd '; ' "$REPLICATION_SCANNER_BRIDGE_BYPASS_HITS_FILE")" +fi + ( cd "$ROOT_DIR" rg -n --with-filename 'crate::services::event_notification' \