refactor(scanner): hide replication queue DTOs (#4150)

crates/scanner/src/lib.rs

@@ -29,15 +29,15 @@ use storage_api::owner::{
     ECSTORE_STORAGECLASS_STANDARD, ECSTORE_TRANSITION_COMPLETE, EcstoreBucketTargetSys, EcstoreBucketVersioningSys, EcstoreDisk,
     EcstoreDiskAPI, EcstoreDiskBytes, EcstoreDiskError, EcstoreDiskInfo, EcstoreDiskInfoOptions, EcstoreDiskLocation,
     EcstoreDiskResult, EcstoreErrorType, EcstoreEvaluator, EcstoreEvent, EcstoreLcEventSrc, EcstoreLifecycle,
-    EcstoreListPathRawOptions, EcstoreObjectOpts, EcstoreReplicationConfig, EcstoreReplicationConfigurationExt,
-    EcstoreReplicationHealQueueResult, EcstoreReplicationQueueAdmission, EcstoreReplicationScannerBridge, EcstoreResultType,
-    EcstoreScanGuard, EcstoreSetDisks, EcstoreStorageError, EcstoreStore, EcstoreTierConfig, EcstoreVersioningApi, HTTPRangeSpec,
-    ObjectIO, ObjectOperations, ObjectToDelete, ecstore_apply_expiry_rule, ecstore_apply_transition_rule,
+    EcstoreListPathRawOptions, EcstoreObjectOpts, EcstoreReplicationConfigurationExt, EcstoreReplicationScannerBridge,
+    EcstoreResultType, EcstoreScanGuard, EcstoreSetDisks, EcstoreStorageError, EcstoreStore, EcstoreTierConfig,
+    EcstoreVersioningApi, HTTPRangeSpec, ObjectIO, ObjectOperations, ObjectToDelete, ScannerReplicationHealObject,
+    ScannerReplicationHealResult, ScannerReplicationQueueAdmission, 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,
+    ecstore_resolve_object_store_handle, ecstore_save_config, scanner_replication_config_for_lifecycle_eval,
 };
 #[cfg(test)]
 use storage_api::owner::{EcstoreDiskOption, EcstoreDiskStore, EcstoreEndpoint, ecstore_config_init, ecstore_new_disk};
@@ -61,6 +61,7 @@ pub use scanner::init_data_scanner;
 pub use scanner_io::{clear_dirty_usage_bucket, record_dirty_usage_bucket};
 pub use sleeper::{DynamicSleeper, SCANNER_IDLE_MODE, SCANNER_SLEEPER};
 use std::sync::atomic::{AtomicU64, Ordering};
+pub use storage_api::ScannerReplicationConfig as ReplicationConfig;
 
 static SCANNER_ACTIVE_WORK_UNITS: AtomicU64 = AtomicU64::new(0);
 static SCANNER_FOREGROUND_READ_ACTIVITY: AtomicU64 = AtomicU64::new(0);
@@ -150,9 +151,9 @@ pub(crate) type Evaluator = EcstoreEvaluator;
 pub(crate) type Event = EcstoreEvent;
 pub(crate) type LcEventSrc = EcstoreLcEventSrc;
 pub(crate) type ObjectOpts = EcstoreObjectOpts;
-pub(crate) type ReplicationConfig = EcstoreReplicationConfig;
-pub(crate) type ReplicationHealQueueResult = EcstoreReplicationHealQueueResult;
-pub(crate) type ReplicationQueueAdmission = EcstoreReplicationQueueAdmission;
+pub(crate) type ReplicationHealObject = ScannerReplicationHealObject;
+pub(crate) type ReplicationHealQueueResult = ScannerReplicationHealResult;
+pub(crate) type ReplicationQueueAdmission = ScannerReplicationQueueAdmission;
 pub(crate) type ScanGuard = EcstoreScanGuard;
 pub(crate) type SetDisks = EcstoreSetDisks;
 pub(crate) type StorageError = EcstoreStorageError;
@@ -310,7 +311,9 @@ pub(crate) async fn queue_replication_heal(
     rcfg: ReplicationConfig,
     retry_count: u32,
 ) -> ReplicationHealQueueResult {
-    EcstoreReplicationScannerBridge::queue_heal(bucket, oi, rcfg, retry_count).await
+    EcstoreReplicationScannerBridge::queue_heal(bucket, oi, rcfg.into_ecstore(), retry_count)
+        .await
+        .into()
 }
 
 pub(crate) fn resolve_scanner_object_store_handle() -> Option<Arc<ECStore>> {

crates/scanner/src/scanner_folder.rs

@@ -41,9 +41,7 @@ use rustfs_common::metrics::{
     IlmAction, Metric, Metrics, ScannerReplicationRepairKind, ScannerSourceWorkUpdate, ScannerWorkSource, UpdateCurrentPathFn,
     current_path_updater, global_metrics,
 };
-use rustfs_filemeta::{
-    MetaCacheEntries, MetaCacheEntry, MetadataResolutionParams, ReplicateObjectInfo, ReplicationStatusType, ReplicationType,
-};
+use rustfs_filemeta::{MetaCacheEntries, MetaCacheEntry, MetadataResolutionParams, ReplicationStatusType};
 use rustfs_utils::path::{SLASH_SEPARATOR, path_join_buf};
 use s3s::dto::{BucketLifecycleConfiguration, ObjectLockConfiguration};
 use time::OffsetDateTime;
@@ -54,10 +52,10 @@ use tracing::{debug, error, warn};
 
 use crate::{
     BucketVersioningSys, Disk, DiskError, DiskInfoOptions, Evaluator, Event, LcEventSrc, ListPathRawOptions, ObjectOpts,
-    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, scanner_is_erasure,
+    ReplicationConfig, ReplicationHealObject, ReplicationQueueAdmission, ScannerDiskExt as _, ScannerLifecycleConfigExt 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, scanner_is_erasure, scanner_replication_config_for_lifecycle_eval,
 };
 use crate::{ScannerObjectInfo as ObjectInfo, ScannerObjectToDelete as ObjectToDelete};
 
@@ -213,14 +211,14 @@ fn scanner_replication_work_update(admission: ReplicationQueueAdmission) -> Scan
     }
 }
 
-fn scanner_replication_repair_kind(roi: &ReplicateObjectInfo) -> Option<ScannerReplicationRepairKind> {
-    if roi.bucket.is_empty() && roi.name.is_empty() {
+fn scanner_replication_repair_kind(roi: &ReplicationHealObject) -> Option<ScannerReplicationRepairKind> {
+    if roi.is_empty_identity() {
         return None;
     }
 
-    if roi.op_type == ReplicationType::ExistingObject || roi.existing_obj_resync.must_resync() {
+    if roi.is_existing_object_repair() {
         Some(ScannerReplicationRepairKind::BucketExistingObject)
-    } else if !roi.version_purge_status.is_empty() {
+    } else if roi.has_version_purge_status() {
         Some(ScannerReplicationRepairKind::BucketVersionPurge)
     } else if roi.delete_marker {
         Some(ScannerReplicationRepairKind::BucketDeleteMarker)
@@ -229,7 +227,7 @@ fn scanner_replication_repair_kind(roi: &ReplicateObjectInfo) -> Option<ScannerR
     }
 }
 
-fn record_scanner_replication_admission(metrics: &Metrics, roi: &ReplicateObjectInfo, admission: ReplicationQueueAdmission) {
+fn record_scanner_replication_admission(metrics: &Metrics, roi: &ReplicationHealObject, admission: ReplicationQueueAdmission) {
     let work = scanner_replication_work_update(admission);
     metrics.record_scanner_source_work(ScannerWorkSource::BucketReplication, work);
     if let Some(kind) = scanner_replication_repair_kind(roi) {
@@ -801,7 +799,7 @@ impl ScannerItem {
 
         let events = match Evaluator::new(lifecycle.clone())
             .with_lock_retention(lock_retention)
-            .with_replication_config(self.replication.clone())
+            .with_replication_config(scanner_replication_config_for_lifecycle_eval(self.replication.clone()))
             .eval(&object_opts)
             .await
         {
@@ -1765,14 +1763,17 @@ impl FolderScanner {
                 None
             };
 
-            let active_replication =
-                if self.old_cache.info.replication.as_ref().is_some_and(|v| {
-                    !v.is_empty() && v.config.as_ref().is_some_and(|config| config.has_active_rules(&prefix, true))
-                }) {
-                    self.old_cache.info.replication.clone()
-                } else {
-                    None
-                };
+            let active_replication = if self
+                .old_cache
+                .info
+                .replication
+                .as_ref()
+                .is_some_and(|v| v.has_active_rules(&prefix, true))
+            {
+                self.old_cache.info.replication.clone()
+            } else {
+                None
+            };
 
             self.sleeper.sleep_folder().await;
 
@@ -2835,9 +2836,7 @@ mod tests {
 
     use super::*;
     use crate::{DiskOption, Endpoint, new_disk};
-    use rustfs_filemeta::{
-        FileInfo, FileMeta, ReplicateObjectInfo, ReplicationType, ResyncDecision, ResyncTargetDecision, VersionPurgeStatusType,
-    };
+    use rustfs_filemeta::{FileInfo, FileMeta, VersionPurgeStatusType};
     use serial_test::serial;
     #[cfg(unix)]
     use std::os::unix::fs::{PermissionsExt, symlink};
@@ -3089,11 +3088,7 @@ mod tests {
     #[tokio::test]
     async fn test_scanner_replication_admission_accounting_maps_source_work() {
         let metrics = Metrics::new();
-        let object = ReplicateObjectInfo {
-            bucket: "bucket-a".to_string(),
-            name: "object-a".to_string(),
-            ..Default::default()
-        };
+        let object = ReplicationHealObject::new("bucket-a", "object-a");
 
         record_scanner_replication_admission(&metrics, &object, ReplicationQueueAdmission::Skipped);
         record_scanner_replication_admission(&metrics, &object, ReplicationQueueAdmission::Queued);
@@ -3124,70 +3119,36 @@ mod tests {
 
     #[test]
     fn test_scanner_replication_repair_kind_maps_bucket_variants() {
-        let object = ReplicateObjectInfo {
-            bucket: "bucket-a".to_string(),
-            name: "object-a".to_string(),
-            replication_status: ReplicationStatusType::Pending,
-            ..Default::default()
-        };
+        let object = ReplicationHealObject::new("bucket-a", "object-a");
         assert_eq!(scanner_replication_repair_kind(&object), Some(ScannerReplicationRepairKind::BucketObject));
 
-        let delete_marker = ReplicateObjectInfo {
-            bucket: "bucket-a".to_string(),
-            name: "delete-marker-a".to_string(),
-            delete_marker: true,
-            replication_status: ReplicationStatusType::Failed,
-            ..Default::default()
-        };
+        let delete_marker = ReplicationHealObject::new("bucket-a", "delete-marker-a").with_delete_marker();
         assert_eq!(
             scanner_replication_repair_kind(&delete_marker),
             Some(ScannerReplicationRepairKind::BucketDeleteMarker)
         );
 
-        let version_purge = ReplicateObjectInfo {
-            bucket: "bucket-a".to_string(),
-            name: "version-purge-a".to_string(),
-            version_purge_status: VersionPurgeStatusType::Pending,
-            ..Default::default()
-        };
+        let version_purge = ReplicationHealObject::new("bucket-a", "version-purge-a").with_pending_version_purge();
         assert_eq!(
             scanner_replication_repair_kind(&version_purge),
             Some(ScannerReplicationRepairKind::BucketVersionPurge)
         );
 
-        let existing_object = ReplicateObjectInfo {
-            bucket: "bucket-a".to_string(),
-            name: "existing-object-a".to_string(),
-            op_type: ReplicationType::ExistingObject,
-            ..Default::default()
-        };
+        let existing_object = ReplicationHealObject::new("bucket-a", "existing-object-a").with_existing_object();
         assert_eq!(
             scanner_replication_repair_kind(&existing_object),
             Some(ScannerReplicationRepairKind::BucketExistingObject)
         );
 
-        let mut existing_obj_resync = ResyncDecision::new();
-        existing_obj_resync.targets.insert(
-            "arn:minio:replication:::target".to_string(),
-            ResyncTargetDecision {
-                replicate: true,
-                ..Default::default()
-            },
-        );
-        let existing_delete_marker = ReplicateObjectInfo {
-            bucket: "bucket-a".to_string(),
-            name: "existing-delete-marker-a".to_string(),
-            delete_marker: true,
-            replication_status: ReplicationStatusType::Completed,
-            existing_obj_resync,
-            ..Default::default()
-        };
+        let existing_delete_marker = ReplicationHealObject::new("bucket-a", "existing-delete-marker-a")
+            .with_delete_marker()
+            .with_existing_object_resync();
         assert_eq!(
             scanner_replication_repair_kind(&existing_delete_marker),
             Some(ScannerReplicationRepairKind::BucketExistingObject)
         );
 
-        assert_eq!(scanner_replication_repair_kind(&ReplicateObjectInfo::default()), None);
+        assert_eq!(scanner_replication_repair_kind(&ReplicationHealObject::default()), None);
     }
 
     #[tokio::test]

crates/scanner/src/scanner_io.rs

@@ -1518,10 +1518,7 @@ impl ScannerIODisk for Disk {
         if replication_config.has_active_rules("", true)
             && let Ok(targets) = BucketTargetSys::get().list_bucket_targets(&cache.info.name).await
         {
-            cache.info.replication = Some(Arc::new(ReplicationConfig {
-                config: Some(replication_config),
-                remotes: Some(targets),
-            }));
+            cache.info.replication = Some(Arc::new(ReplicationConfig::new(Some(replication_config), Some(targets))));
         }
 
         // TODO: object lock

crates/scanner/src/storage_api.rs

@@ -12,6 +12,12 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::collections::HashMap;
+use std::sync::Arc;
+
+use rustfs_filemeta::{ReplicateObjectInfo, ReplicationStatusType, ReplicationType, VersionPurgeStatusType};
+use serde::{Deserialize, Serialize};
+
 pub(crate) use rustfs_ecstore::api::bucket::bucket_target_sys::BucketTargetSys as EcstoreBucketTargetSys;
 pub(crate) use rustfs_ecstore::api::bucket::lifecycle::bucket_lifecycle_audit::LcEventSrc as EcstoreLcEventSrc;
 pub(crate) use rustfs_ecstore::api::bucket::lifecycle::bucket_lifecycle_ops::{
@@ -31,6 +37,7 @@ pub(crate) use rustfs_ecstore::api::bucket::replication::{
     ReplicationHealQueueResult as EcstoreReplicationHealQueueResult,
     ReplicationQueueAdmission as EcstoreReplicationQueueAdmission, ReplicationScannerBridge as EcstoreReplicationScannerBridge,
 };
+pub(crate) use rustfs_ecstore::api::bucket::target::BucketTargets as EcstoreBucketTargets;
 pub(crate) use rustfs_ecstore::api::bucket::versioning::VersioningApi as EcstoreVersioningApi;
 pub(crate) use rustfs_ecstore::api::bucket::versioning_sys::BucketVersioningSys as EcstoreBucketVersioningSys;
 pub(crate) use rustfs_ecstore::api::cache::{
@@ -81,21 +88,159 @@ pub(crate) mod owner {
         ECSTORE_STORAGECLASS_STANDARD, ECSTORE_TRANSITION_COMPLETE, EcstoreBucketTargetSys, EcstoreBucketVersioningSys,
         EcstoreDisk, EcstoreDiskAPI, EcstoreDiskBytes, EcstoreDiskError, EcstoreDiskInfo, EcstoreDiskInfoOptions,
         EcstoreDiskLocation, EcstoreDiskResult, EcstoreErrorType, EcstoreEvaluator, EcstoreEvent, EcstoreLcEventSrc,
-        EcstoreLifecycle, EcstoreListPathRawOptions, EcstoreObjectOpts, EcstoreReplicationConfig,
-        EcstoreReplicationConfigurationExt, EcstoreReplicationHealQueueResult, EcstoreReplicationQueueAdmission,
+        EcstoreLifecycle, EcstoreListPathRawOptions, EcstoreObjectOpts, EcstoreReplicationConfigurationExt,
         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,
+        EcstoreTierConfig, EcstoreVersioningApi, ScannerReplicationHealObject, ScannerReplicationHealResult,
+        ScannerReplicationQueueAdmission, 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,
+        scanner_replication_config_for_lifecycle_eval,
     };
 
     #[cfg(test)]
     pub(crate) use super::{EcstoreDiskOption, EcstoreDiskStore, EcstoreEndpoint, ecstore_config_init, ecstore_new_disk};
 }
 
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct ScannerReplicationConfig(EcstoreReplicationConfig);
+
+impl ScannerReplicationConfig {
+    pub(crate) fn new(config: Option<s3s::dto::ReplicationConfiguration>, remotes: Option<EcstoreBucketTargets>) -> Self {
+        Self(EcstoreReplicationConfig::new(config, remotes))
+    }
+
+    pub(crate) fn has_active_rules(&self, prefix: &str, recursive: bool) -> bool {
+        !self.0.is_empty()
+            && self
+                .0
+                .config
+                .as_ref()
+                .is_some_and(|config| EcstoreReplicationConfigurationExt::has_active_rules(config, prefix, recursive))
+    }
+
+    pub(crate) fn into_ecstore(self) -> EcstoreReplicationConfig {
+        self.0
+    }
+}
+
+pub(crate) fn scanner_replication_config_for_lifecycle_eval(
+    config: Option<Arc<ScannerReplicationConfig>>,
+) -> Option<Arc<EcstoreReplicationConfig>> {
+    config.map(|config| match Arc::try_unwrap(config) {
+        Ok(config) => Arc::new(config.into_ecstore()),
+        Err(config) => Arc::new(config.0.clone()),
+    })
+}
+
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
+pub(crate) enum ScannerReplicationQueueAdmission {
+    #[default]
+    Skipped,
+    Queued,
+    Missed,
+}
+
+impl From<EcstoreReplicationQueueAdmission> for ScannerReplicationQueueAdmission {
+    fn from(admission: EcstoreReplicationQueueAdmission) -> Self {
+        match admission {
+            EcstoreReplicationQueueAdmission::Skipped => Self::Skipped,
+            EcstoreReplicationQueueAdmission::Queued => Self::Queued,
+            EcstoreReplicationQueueAdmission::Missed => Self::Missed,
+        }
+    }
+}
+
+#[derive(Debug, Clone, Default)]
+pub(crate) struct ScannerReplicationHealObject {
+    pub(crate) bucket: String,
+    pub(crate) name: String,
+    pub(crate) size: i64,
+    pub(crate) delete_marker: bool,
+    pub(crate) target_statuses: HashMap<String, ReplicationStatusType>,
+    version_purge_status: VersionPurgeStatusType,
+    existing_object: bool,
+    existing_object_resync: bool,
+}
+
+impl ScannerReplicationHealObject {
+    #[cfg(test)]
+    pub(crate) fn new(bucket: impl Into<String>, name: impl Into<String>) -> Self {
+        Self {
+            bucket: bucket.into(),
+            name: name.into(),
+            ..Default::default()
+        }
+    }
+
+    pub(crate) fn is_empty_identity(&self) -> bool {
+        self.bucket.is_empty() && self.name.is_empty()
+    }
+
+    pub(crate) fn is_existing_object_repair(&self) -> bool {
+        self.existing_object || self.existing_object_resync
+    }
+
+    pub(crate) fn has_version_purge_status(&self) -> bool {
+        !self.version_purge_status.is_empty()
+    }
+
+    #[cfg(test)]
+    pub(crate) fn with_delete_marker(mut self) -> Self {
+        self.delete_marker = true;
+        self
+    }
+
+    #[cfg(test)]
+    pub(crate) fn with_pending_version_purge(mut self) -> Self {
+        self.version_purge_status = VersionPurgeStatusType::Pending;
+        self
+    }
+
+    #[cfg(test)]
+    pub(crate) fn with_existing_object(mut self) -> Self {
+        self.existing_object = true;
+        self
+    }
+
+    #[cfg(test)]
+    pub(crate) fn with_existing_object_resync(mut self) -> Self {
+        self.existing_object_resync = true;
+        self
+    }
+}
+
+impl From<ReplicateObjectInfo> for ScannerReplicationHealObject {
+    fn from(object: ReplicateObjectInfo) -> Self {
+        Self {
+            bucket: object.bucket,
+            name: object.name,
+            size: object.size,
+            delete_marker: object.delete_marker,
+            target_statuses: object.target_statuses,
+            version_purge_status: object.version_purge_status,
+            existing_object: object.op_type == ReplicationType::ExistingObject,
+            existing_object_resync: object.existing_obj_resync.must_resync(),
+        }
+    }
+}
+
+#[derive(Debug, Clone, Default)]
+pub(crate) struct ScannerReplicationHealResult {
+    pub(crate) object_info: ScannerReplicationHealObject,
+    pub(crate) admission: ScannerReplicationQueueAdmission,
+}
+
+impl From<EcstoreReplicationHealQueueResult> for ScannerReplicationHealResult {
+    fn from(result: EcstoreReplicationHealQueueResult) -> Self {
+        Self {
+            object_info: result.object_info.into(),
+            admission: result.admission.into(),
+        }
+    }
+}
+
 pub(crate) mod scan {
     pub(crate) use super::storage_contracts::{BucketOperations, BucketOptions, NamespaceLocking};
 }