Move replication delete worker contract into crate (#4166)

refactor(replication): move delete worker contract into crate

Cargo.lock

@@ -9897,6 +9897,7 @@ dependencies = [
  "rmp",
  "rmp-serde",
  "rustfs-filemeta",
+ "rustfs-storage-api",
  "rustfs-utils",
  "s3s",
  "serde",

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

@@ -45,9 +45,11 @@ pub use replication_pool::{
     DynReplicationPool, ReplicationHealQueueResult, ReplicationPoolTrait, ReplicationQueueAdmission, get_global_replication_pool,
     get_global_replication_stats, init_background_replication,
 };
-pub use replication_resyncer::{DeletedObjectReplicationInfo, ReplicationConfig};
+pub use replication_resyncer::ReplicationConfig;
 pub use replication_scanner_bridge::ReplicationScannerBridge;
 pub use replication_state::{BucketStats, ReplicationStats};
 pub use replication_storage_boundary::{ReplicationObjectIO, ReplicationStorage};
 pub(crate) use replication_target_config_bridge::ReplicationTargetConfigBridge;
-pub use rustfs_replication::{BucketReplicationResyncStatus, MustReplicateOptions, ResyncOpts, TargetReplicationResyncStatus};
+pub use rustfs_replication::{
+    BucketReplicationResyncStatus, DeletedObjectReplicationInfo, MustReplicateOptions, ResyncOpts, TargetReplicationResyncStatus,
+};

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

@@ -18,8 +18,9 @@ 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_resyncer::{ReplicationConfig, check_replicate_delete};
 use super::replication_storage_boundary::{DeletedObject, ObjectInfo, ObjectOptions, ObjectToDelete};
+use rustfs_replication::DeletedObjectReplicationInfo;
 
 pub(crate) type ReplicationLifecycleConfig = ReplicationConfig;
 

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

@@ -16,11 +16,9 @@ use std::{collections::HashMap, sync::Arc};
 
 use super::replication_filemeta_boundary::{ReplicateDecision, ReplicationStatusType, ReplicationType};
 use super::replication_pool::{schedule_replication, schedule_replication_delete};
-use super::replication_resyncer::{
-    DeletedObjectReplicationInfo, check_replicate_delete, get_must_replicate_options, must_replicate,
-};
+use super::replication_resyncer::{check_replicate_delete, get_must_replicate_options, must_replicate};
 use super::replication_storage_boundary::{ObjectInfo, ObjectOptions, ObjectToDelete, ReplicationStorage};
-use rustfs_replication::MustReplicateOptions;
+use rustfs_replication::{DeletedObjectReplicationInfo, MustReplicateOptions};
 
 pub struct ReplicationObjectBridge;
 

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

@@ -22,8 +22,8 @@ use super::replication_filemeta_boundary::{
 };
 use super::replication_metadata_boundary::ReplicationMetadataStore;
 use super::replication_resyncer::{
-    DeletedObjectReplicationInfo, ReplicationConfig, ReplicationResyncer, decode_mrf_file, decode_resync_file, encode_mrf_file,
-    get_heal_replicate_object_info, replicate_delete, replicate_object, save_resync_status,
+    ReplicationConfig, ReplicationResyncer, decode_mrf_file, decode_resync_file, encode_mrf_file, get_heal_replicate_object_info,
+    replicate_delete, replicate_object, save_resync_status,
 };
 use super::replication_state::ReplicationStats;
 use super::replication_storage_boundary::{DeletedObject, ObjectInfo, ObjectOptions, ReplicationObjectIO, ReplicationStorage};
@@ -31,6 +31,7 @@ use super::replication_target_boundary::ReplicationTargetStore;
 use super::runtime_boundary as runtime_sources;
 use super::{BucketReplicationResyncStatus, ResyncOpts, TargetReplicationResyncStatus};
 use lazy_static::lazy_static;
+use rustfs_replication::DeletedObjectReplicationInfo;
 use rustfs_utils::http::{SUFFIX_REPLICATION_TIMESTAMP, get_str};
 use std::any::Any;
 use std::sync::Arc;

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

@@ -19,10 +19,10 @@ use super::replication_config_store::ReplicationConfigStore;
 use super::replication_error_boundary::{Error, Result, is_err_object_not_found, is_err_version_not_found};
 use super::replication_event_sink::{EventArgs, send_event, send_local_event};
 use super::replication_filemeta_boundary::{
-    MrfOpKind, MrfReplicateEntry, REPLICATE_EXISTING, REPLICATE_EXISTING_DELETE, ReplicateDecision, ReplicateObjectInfo,
+    MrfReplicateEntry, REPLICATE_EXISTING, REPLICATE_EXISTING_DELETE, ReplicateDecision, ReplicateObjectInfo,
     ReplicateTargetDecision, ReplicatedInfos, ReplicatedTargetInfo, ReplicationAction, ReplicationStatusType, ReplicationType,
-    ReplicationWorkerOperation, ResyncDecision, ResyncTargetDecision, VersionPurgeStatusType, get_replication_state,
-    parse_replicate_decision, replication_statuses_map, target_reset_header, version_purge_statuses_map,
+    ResyncDecision, ResyncTargetDecision, VersionPurgeStatusType, get_replication_state, parse_replicate_decision,
+    replication_statuses_map, target_reset_header, version_purge_statuses_map,
 };
 use super::replication_lock_boundary::ReplicationLockTiming;
 use super::replication_metadata_boundary::ReplicationMetadataStore;
@@ -55,7 +55,9 @@ use http_body::Frame;
 use http_body_util::StreamBody;
 #[cfg(test)]
 use rmp_serde;
-use rustfs_replication::{BucketReplicationResyncStatus, MustReplicateOptions, ResyncOpts, TargetReplicationResyncStatus};
+use rustfs_replication::{
+    BucketReplicationResyncStatus, DeletedObjectReplicationInfo, MustReplicateOptions, ResyncOpts, TargetReplicationResyncStatus,
+};
 use rustfs_s3_types::EventName;
 use rustfs_utils::http::{
     AMZ_BUCKET_REPLICATION_STATUS, AMZ_OBJECT_TAGGING, AMZ_TAGGING_DIRECTIVE, CONTENT_ENCODING, HeaderExt as _,
@@ -71,7 +73,6 @@ use rustfs_utils::{DEFAULT_SIP_HASH_KEY, sip_hash};
 use s3s::dto::ReplicationConfiguration;
 use serde::Deserialize;
 use serde::Serialize;
-use std::any::Any;
 use std::collections::HashMap;
 use std::sync::Arc;
 use time::OffsetDateTime;
@@ -965,57 +966,6 @@ async fn get_replication_config(bucket: &str) -> Result<Option<ReplicationConfig
     ReplicationMetadataStore::optional_replication_config(bucket).await
 }
 
-#[derive(Debug, Clone, Default)]
-pub struct DeletedObjectReplicationInfo {
-    pub delete_object: DeletedObject,
-    pub bucket: String,
-    pub event_type: String,
-    pub op_type: ReplicationType,
-    pub reset_id: String,
-    pub target_arn: String,
-}
-
-impl ReplicationWorkerOperation for DeletedObjectReplicationInfo {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
-    fn to_mrf_entry(&self) -> MrfReplicateEntry {
-        MrfReplicateEntry {
-            bucket: self.bucket.clone(),
-            object: self.delete_object.object_name.clone(),
-            // version_id here is the version being purged (if any); the delete-marker
-            // version is stored separately in delete_marker_version_id.
-            version_id: self.delete_object.version_id,
-            retry_count: 0,
-            size: 0,
-            op: MrfOpKind::Delete,
-            delete_marker_version_id: self.delete_object.delete_marker_version_id,
-            delete_marker: self.delete_object.delete_marker,
-        }
-    }
-
-    fn get_bucket(&self) -> &str {
-        &self.bucket
-    }
-
-    fn get_object(&self) -> &str {
-        &self.delete_object.object_name
-    }
-
-    fn get_size(&self) -> i64 {
-        0
-    }
-
-    fn is_delete_marker(&self) -> bool {
-        true
-    }
-
-    fn get_op_type(&self) -> ReplicationType {
-        self.op_type
-    }
-}
-
 #[derive(Debug, Clone, Default, Serialize, Deserialize)]
 pub struct ReplicationConfig {
     pub config: Option<ReplicationConfiguration>,

crates/replication/Cargo.toml

@@ -30,6 +30,7 @@ byteorder.workspace = true
 rmp.workspace = true
 rmp-serde.workspace = true
 rustfs-filemeta.workspace = true
+rustfs-storage-api.workspace = true
 rustfs-utils = { workspace = true, features = ["http"] }
 s3s.workspace = true
 serde.workspace = true

crates/replication/src/delete.rs

@@ -0,0 +1,104 @@
+// 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 std::any::Any;
+
+use rustfs_storage_api::DeletedObject;
+
+use crate::{MrfOpKind, MrfReplicateEntry, ReplicationType, ReplicationWorkerOperation};
+
+#[derive(Debug, Clone, Default)]
+pub struct DeletedObjectReplicationInfo {
+    pub delete_object: DeletedObject,
+    pub bucket: String,
+    pub event_type: String,
+    pub op_type: ReplicationType,
+    pub reset_id: String,
+    pub target_arn: String,
+}
+
+impl ReplicationWorkerOperation for DeletedObjectReplicationInfo {
+    fn as_any(&self) -> &dyn Any {
+        self
+    }
+
+    fn to_mrf_entry(&self) -> MrfReplicateEntry {
+        MrfReplicateEntry {
+            bucket: self.bucket.clone(),
+            object: self.delete_object.object_name.clone(),
+            version_id: self.delete_object.version_id,
+            retry_count: 0,
+            size: 0,
+            op: MrfOpKind::Delete,
+            delete_marker_version_id: self.delete_object.delete_marker_version_id,
+            delete_marker: self.delete_object.delete_marker,
+        }
+    }
+
+    fn get_bucket(&self) -> &str {
+        &self.bucket
+    }
+
+    fn get_object(&self) -> &str {
+        &self.delete_object.object_name
+    }
+
+    fn get_size(&self) -> i64 {
+        0
+    }
+
+    fn is_delete_marker(&self) -> bool {
+        true
+    }
+
+    fn get_op_type(&self) -> ReplicationType {
+        self.op_type
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::DeletedObjectReplicationInfo;
+    use crate::{MrfOpKind, ReplicationType, ReplicationWorkerOperation};
+    use rustfs_storage_api::DeletedObject;
+    use uuid::Uuid;
+
+    #[test]
+    fn deleted_object_replication_info_encodes_delete_mrf_entry() {
+        let version_id = Uuid::new_v4();
+        let delete_marker_version_id = Uuid::new_v4();
+        let info = DeletedObjectReplicationInfo {
+            bucket: "bucket".to_string(),
+            op_type: ReplicationType::Delete,
+            delete_object: DeletedObject {
+                object_name: "object".to_string(),
+                version_id: Some(version_id),
+                delete_marker_version_id: Some(delete_marker_version_id),
+                delete_marker: true,
+                ..Default::default()
+            },
+            ..Default::default()
+        };
+
+        let entry = info.to_mrf_entry();
+
+        assert_eq!(entry.bucket, "bucket");
+        assert_eq!(entry.object, "object");
+        assert_eq!(entry.version_id, Some(version_id));
+        assert_eq!(entry.delete_marker_version_id, Some(delete_marker_version_id));
+        assert_eq!(entry.op, MrfOpKind::Delete);
+        assert!(entry.delete_marker);
+        assert_eq!(info.get_object(), "object");
+    }
+}

crates/replication/src/lib.rs

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 pub mod config;
+pub mod delete;
 pub mod mrf;
 pub mod operation;
 pub mod resync;
@@ -20,6 +21,7 @@ pub mod rule;
 pub mod tagging;
 
 pub use config::{ObjectOpts, ReplicationConfigurationExt};
+pub use delete::DeletedObjectReplicationInfo;
 pub use mrf::{MrfOpKind, MrfReplicateEntry, decode_mrf_file, encode_mrf_file};
 pub use operation::{MustReplicateOptions, is_ssec_encrypted};
 pub use resync::{

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"
 REPLICATION_CONFIG_RULE_CONTRACT_BACKSLIDE_HITS_FILE="${TMP_DIR}/replication_config_rule_contract_backslide_hits.txt"
+REPLICATION_DELETE_WORKER_CONTRACT_BACKSLIDE_HITS_FILE="${TMP_DIR}/replication_delete_worker_contract_backslide_hits.txt"
 REPLICATION_OPERATION_CONTRACT_BACKSLIDE_HITS_FILE="${TMP_DIR}/replication_operation_contract_backslide_hits.txt"
 REPLICATION_RESYNC_CONTRACT_BACKSLIDE_HITS_FILE="${TMP_DIR}/replication_resync_contract_backslide_hits.txt"
 REPLICATION_MRF_WIRE_FORMAT_BACKSLIDE_HITS_FILE="${TMP_DIR}/replication_mrf_wire_format_backslide_hits.txt"
@@ -2575,6 +2576,21 @@ if [[ -s "$REPLICATION_OPERATION_CONTRACT_BACKSLIDE_HITS_FILE" ]]; then
   report_failure "replication operation contracts must stay in crates/replication: $(paste -sd '; ' "$REPLICATION_OPERATION_CONTRACT_BACKSLIDE_HITS_FILE")"
 fi
 
+(
+  cd "$ROOT_DIR"
+  replication_delete_worker_status=0
+  rg -n --with-filename '^\s*(?:pub(?:\([^)]*\))?\s+)?struct\s+DeletedObjectReplicationInfo\b' \
+    crates/ecstore/src/bucket/replication \
+    --glob '*.rs' >"$REPLICATION_DELETE_WORKER_CONTRACT_BACKSLIDE_HITS_FILE" || replication_delete_worker_status=$?
+  if [[ "$replication_delete_worker_status" -ne 0 && "$replication_delete_worker_status" -ne 1 ]]; then
+    exit "$replication_delete_worker_status"
+  fi
+)
+
+if [[ -s "$REPLICATION_DELETE_WORKER_CONTRACT_BACKSLIDE_HITS_FILE" ]]; then
+  report_failure "replication delete worker contracts must stay in crates/replication: $(paste -sd '; ' "$REPLICATION_DELETE_WORKER_CONTRACT_BACKSLIDE_HITS_FILE")"
+fi
+
 (
   cd "$ROOT_DIR"
   rg -n --with-filename 'pub\s+(struct|enum)\s+(ResyncOpts|TargetReplicationResyncStatus|BucketReplicationResyncStatus|ResyncStatusType)\b' \