refactor(replication): move resync classifiers into crate (#4170)

2026-07-26 16:28:15 +00:00 · 2026-07-02 12:57:59 +08:00
parent 953a080b37
commit d666028cdc
7 changed files with 194 additions and 49 deletions
@@ -33,8 +33,8 @@ use super::{BucketReplicationResyncStatus, ResyncOpts, TargetReplicationResyncSt
 use lazy_static::lazy_static;
 use rustfs_replication::{
    DeletedObjectReplicationInfo, LARGE_WORKER_COUNT, ReplicationHealQueueResult, ReplicationOperation, ReplicationPoolOpts,
-    ReplicationPriority, ReplicationQueueAdmission, WORKER_MAX_LIMIT, initial_worker_counts, next_large_worker_count,
-    next_mrf_worker_count, next_regular_worker_count, resized_worker_counts, should_grow_large_workers,
+    ReplicationPriority, ReplicationQueueAdmission, WORKER_MAX_LIMIT, initial_worker_counts, mrf_worker_size_to_count,
+    next_large_worker_count, next_mrf_worker_count, next_regular_worker_count, resized_worker_counts, should_grow_large_workers,
    should_queue_large_object,
 };
 use rustfs_utils::http::{SUFFIX_REPLICATION_TIMESTAMP, get_str};
@@ -366,7 +366,7 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
        max_l_workers: Option<usize>,
    ) {
        let current_workers = self.workers.read().await.len();
-        let current_mrf = usize::try_from(self.mrf_worker_size.load(Ordering::SeqCst)).unwrap_or_default();
+        let current_mrf = mrf_worker_size_to_count(self.mrf_worker_size.load(Ordering::SeqCst));
        let worker_counts = resized_worker_counts(&pri, max_workers, current_workers, current_mrf);

        if let Some(max_w) = max_workers {
@@ -57,6 +57,8 @@ use http_body_util::StreamBody;
 use rmp_serde;
 use rustfs_replication::{
    BucketReplicationResyncStatus, DeletedObjectReplicationInfo, MustReplicateOptions, ResyncOpts, TargetReplicationResyncStatus,
+    is_retryable_delete_replication_head_error, is_version_delete_replication, is_version_id_mismatch,
+    resync_state_accepts_update, should_count_head_proxy_failure, should_retry_delete_marker_purge,
 };
 use rustfs_s3_types::EventName;
 use rustfs_utils::http::{
@@ -106,20 +108,6 @@ const RESYNC_TIME_INTERVAL: TokioDuration = TokioDuration::from_secs(60);

 static WARNED_MONITOR_UNINIT: std::sync::Once = std::sync::Once::new();

-fn resync_state_accepts_update(state: &TargetReplicationResyncStatus, opts: &ResyncOpts) -> bool {
-    state.resync_id.is_empty() || opts.resync_id.is_empty() || state.resync_id == opts.resync_id
-}
-
-fn should_count_head_proxy_failure(is_not_found: bool, code: Option<&str>, raw_status: Option<u16>) -> bool {
-    if is_not_found || matches!(code, Some("MethodNotAllowed" | "405")) {
-        return false;
-    }
-    if matches!(raw_status, Some(404 | 405)) {
-        return false;
-    }
-    !is_version_id_mismatch(code, raw_status)
-}
-
 fn has_raw_status(err: &SdkError<HeadObjectError>, status: u16) -> bool {
    err.raw_response().is_some_and(|r| r.status().as_u16() == status)
 }
@@ -152,16 +140,6 @@ async fn head_object_with_proxy_stats(
    result
 }

-// AWS returns 400 for root callers and 403 for IAM users when a UUID version ID
-// is rejected. The 403 case is safe: a real auth failure also returns 403 on the
-// versionId-less fallback, propagating as a hard error instead of silently skipping.
-fn is_version_id_mismatch(code: Option<&str>, raw_status: Option<u16>) -> bool {
-    match code {
-        Some(c) if !c.is_empty() => c == "InvalidArgument",
-        _ => matches!(raw_status, Some(400) | Some(403)),
-    }
-}
-
 fn is_version_id_format_mismatch(err: &SdkError<HeadObjectError>) -> bool {
    let code = err.as_service_error().and_then(|se| se.code());
    let raw_status = err.raw_response().map(|r| r.status().as_u16());
@@ -2038,18 +2016,6 @@ async fn replicate_force_delete_to_targets<S: ReplicationStorage>(dobj: &Deleted
    }
 }

-fn is_version_delete_replication(dobj: &DeletedObject) -> bool {
-    dobj.version_id.is_some() || (dobj.delete_marker_version_id.is_some() && !dobj.delete_marker)
-}
-
-fn should_retry_delete_marker_purge(dobj: &DeletedObject) -> bool {
-    dobj.delete_marker_version_id.is_some()
-}
-
-fn is_retryable_delete_replication_head_error(is_not_found: bool, code: Option<&str>) -> bool {
-    !is_not_found && !matches!(code, Some("MethodNotAllowed" | "405"))
-}
-
 async fn replicate_delete_to_target(dobj: &DeletedObjectReplicationInfo, tgt_client: Arc<TargetClient>) -> ReplicatedTargetInfo {
    let version_id = if let Some(version_id) = &dobj.delete_object.delete_marker_version_id {
        version_id.to_owned()
@@ -67,9 +67,24 @@ impl ReplicationWorkerOperation for DeletedObjectReplicationInfo {
    }
 }

+pub fn is_version_delete_replication(dobj: &DeletedObject) -> bool {
+    dobj.version_id.is_some() || (dobj.delete_marker_version_id.is_some() && !dobj.delete_marker)
+}
+
+pub fn should_retry_delete_marker_purge(dobj: &DeletedObject) -> bool {
+    dobj.delete_marker_version_id.is_some()
+}
+
+pub fn is_retryable_delete_replication_head_error(is_not_found: bool, code: Option<&str>) -> bool {
+    !(is_not_found || matches!(code, Some("MethodNotAllowed" | "405")))
+}
+
 #[cfg(test)]
 mod tests {
-    use super::DeletedObjectReplicationInfo;
+    use super::{
+        DeletedObjectReplicationInfo, is_retryable_delete_replication_head_error, is_version_delete_replication,
+        should_retry_delete_marker_purge,
+    };
    use crate::{MrfOpKind, ReplicationType, ReplicationWorkerOperation};
    use rustfs_storage_api::DeletedObject;
    use uuid::Uuid;
@@ -101,4 +116,67 @@ mod tests {
        assert!(entry.delete_marker);
        assert_eq!(info.get_object(), "object");
    }
+
+    #[test]
+    fn version_delete_replication_tracks_delete_marker_version_purge() {
+        let dobj = DeletedObject {
+            delete_marker: false,
+            delete_marker_version_id: Some(Uuid::new_v4()),
+            ..Default::default()
+        };
+
+        assert!(is_version_delete_replication(&dobj));
+    }
+
+    #[test]
+    fn version_delete_replication_tracks_explicit_version_id() {
+        let dobj = DeletedObject {
+            version_id: Some(Uuid::new_v4()),
+            ..Default::default()
+        };
+
+        assert!(is_version_delete_replication(&dobj));
+    }
+
+    #[test]
+    fn version_delete_replication_keeps_delete_marker_creation_separate() {
+        let dobj = DeletedObject {
+            delete_marker: true,
+            delete_marker_version_id: Some(Uuid::new_v4()),
+            ..Default::default()
+        };
+
+        assert!(!is_version_delete_replication(&dobj));
+    }
+
+    #[test]
+    fn delete_marker_purge_retry_covers_version_purge_and_marker_creation() {
+        let version_purge = DeletedObject {
+            delete_marker: false,
+            delete_marker_version_id: Some(Uuid::new_v4()),
+            ..Default::default()
+        };
+        let marker_creation = DeletedObject {
+            delete_marker: true,
+            delete_marker_version_id: Some(Uuid::new_v4()),
+            ..Default::default()
+        };
+
+        assert!(should_retry_delete_marker_purge(&version_purge));
+        assert!(should_retry_delete_marker_purge(&marker_creation));
+
+        let marker_without_version = DeletedObject {
+            delete_marker: true,
+            ..Default::default()
+        };
+        assert!(!should_retry_delete_marker_purge(&marker_without_version));
+    }
+
+    #[test]
+    fn retryable_delete_replication_head_error_allows_expected_delete_marker_responses() {
+        assert!(!is_retryable_delete_replication_head_error(false, Some("405")));
+        assert!(!is_retryable_delete_replication_head_error(false, Some("MethodNotAllowed")));
+        assert!(!is_retryable_delete_replication_head_error(true, Some("NoSuchKey")));
+        assert!(is_retryable_delete_replication_head_error(false, Some("AccessDenied")));
+    }
 }
@@ -24,20 +24,23 @@ pub mod stats;
 pub mod tagging;

 pub use config::{ObjectOpts, ReplicationConfigurationExt};
-pub use delete::DeletedObjectReplicationInfo;
+pub use delete::{
+    DeletedObjectReplicationInfo, is_retryable_delete_replication_head_error, is_version_delete_replication,
+    should_retry_delete_marker_purge,
+};
 pub use mrf::{MrfOpKind, MrfReplicateEntry, decode_mrf_file, encode_mrf_file};
 pub use operation::{MustReplicateOptions, is_ssec_encrypted};
 pub use queue::{ReplicationHealQueueResult, ReplicationOperation, ReplicationPriority, ReplicationQueueAdmission};
 pub use resync::{
    BucketReplicationResyncStatus, Error, Result, ResyncOpts, ResyncStatusType, TargetReplicationResyncStatus,
-    decode_resync_file, encode_resync_file,
+    decode_resync_file, encode_resync_file, is_version_id_mismatch, resync_state_accepts_update, should_count_head_proxy_failure,
 };
 pub use rule::ReplicationRuleExt;
 pub use runtime::{
    LARGE_WORKER_COUNT, MIN_LARGE_OBJ_SIZE, MRF_WORKER_AUTO_DEFAULT, MRF_WORKER_MAX_LIMIT, MRF_WORKER_MIN_LIMIT,
    ReplicationPoolOpts, ReplicationWorkerCounts, WORKER_AUTO_DEFAULT, WORKER_MAX_LIMIT, WORKER_MIN_LIMIT, initial_worker_counts,
-    next_large_worker_count, next_mrf_worker_count, next_regular_worker_count, resized_worker_counts, should_grow_large_workers,
-    should_queue_large_object, worker_counts_for_priority,
+    mrf_worker_size_to_count, next_large_worker_count, next_mrf_worker_count, next_regular_worker_count, resized_worker_counts,
+    should_grow_large_workers, should_queue_large_object, worker_counts_for_priority,
 };
 pub use rustfs_filemeta::{
    REPLICATE_EXISTING, REPLICATE_EXISTING_DELETE, REPLICATE_HEAL, REPLICATE_HEAL_DELETE, REPLICATE_INCOMING_DELETE,
@@ -205,6 +205,30 @@ impl TargetReplicationResyncStatus {
    }
 }

+pub fn resync_state_accepts_update(state: &TargetReplicationResyncStatus, opts: &ResyncOpts) -> bool {
+    state.resync_id.is_empty() || opts.resync_id.is_empty() || state.resync_id == opts.resync_id
+}
+
+pub fn should_count_head_proxy_failure(is_not_found: bool, code: Option<&str>, raw_status: Option<u16>) -> bool {
+    if is_not_found || matches!(code, Some("MethodNotAllowed" | "405")) {
+        return false;
+    }
+    if matches!(raw_status, Some(404 | 405)) {
+        return false;
+    }
+    !is_version_id_mismatch(code, raw_status)
+}
+
+// AWS returns 400 for root callers and 403 for IAM users when a UUID version ID
+// is rejected. The 403 case is safe: a real auth failure also returns 403 on the
+// versionId-less fallback, propagating as a hard error instead of silently skipping.
+pub fn is_version_id_mismatch(code: Option<&str>, raw_status: Option<u16>) -> bool {
+    match code {
+        Some(c) if !c.is_empty() => c == "InvalidArgument",
+        _ => matches!(raw_status, Some(400 | 403)),
+    }
+}
+
 #[derive(Debug, Clone, Serialize, Deserialize, Default)]
 pub struct BucketReplicationResyncStatus {
    pub version: u16,
@@ -519,6 +543,61 @@ mod tests {
        assert_eq!(ResyncStatusType::NoResync.to_string(), "");
    }

+    #[test]
+    fn resync_state_accepts_update_only_for_matching_run() {
+        let current = TargetReplicationResyncStatus {
+            resync_id: "run-new".to_string(),
+            ..Default::default()
+        };
+        let matching = ResyncOpts {
+            bucket: "bucket".to_string(),
+            arn: "arn:replication::dest".to_string(),
+            resync_id: "run-new".to_string(),
+            resync_before: None,
+        };
+        let stale = ResyncOpts {
+            bucket: "bucket".to_string(),
+            arn: "arn:replication::dest".to_string(),
+            resync_id: "run-old".to_string(),
+            resync_before: None,
+        };
+
+        assert!(resync_state_accepts_update(&TargetReplicationResyncStatus::default(), &matching));
+        assert!(resync_state_accepts_update(&current, &matching));
+        assert!(!resync_state_accepts_update(&current, &stale));
+    }
+
+    #[test]
+    fn head_proxy_failure_ignores_expected_target_responses() {
+        assert!(!should_count_head_proxy_failure(true, Some("NoSuchKey"), Some(404)));
+        assert!(!should_count_head_proxy_failure(false, Some("MethodNotAllowed"), Some(405)));
+        assert!(!should_count_head_proxy_failure(false, Some("405"), Some(405)));
+        assert!(!should_count_head_proxy_failure(false, Some("InvalidArgument"), Some(400)));
+        assert!(!should_count_head_proxy_failure(false, None, Some(400)));
+        assert!(!should_count_head_proxy_failure(false, None, Some(403)));
+    }
+
+    #[test]
+    fn head_proxy_failure_counts_unexpected_errors() {
+        assert!(should_count_head_proxy_failure(false, Some("AccessDenied"), Some(403)));
+        assert!(should_count_head_proxy_failure(false, None, Some(500)));
+    }
+
+    #[test]
+    fn version_id_mismatch_detects_aws_rejections() {
+        assert!(is_version_id_mismatch(Some("InvalidArgument"), Some(400)));
+        assert!(is_version_id_mismatch(None, Some(400)));
+        assert!(is_version_id_mismatch(Some(""), Some(400)));
+        assert!(is_version_id_mismatch(None, Some(403)));
+        assert!(is_version_id_mismatch(Some(""), Some(403)));
+        assert!(!is_version_id_mismatch(Some("AccessDenied"), Some(403)));
+        assert!(!is_version_id_mismatch(Some("NoSuchKey"), Some(404)));
+        assert!(!is_version_id_mismatch(Some("MalformedXML"), Some(400)));
+        assert!(!is_version_id_mismatch(Some("EntityTooLarge"), Some(400)));
+        assert!(!is_version_id_mismatch(None, Some(500)));
+        assert!(!is_version_id_mismatch(None, Some(404)));
+    }
+
    #[test]
    fn resync_file_round_trips_status() {
        let mut status = BucketReplicationResyncStatus::new();
@@ -82,6 +82,14 @@ pub fn resized_worker_counts(
    }
 }

+pub fn mrf_worker_size_to_count(size: i32) -> usize {
+    let non_negative = size.max(0);
+    match usize::try_from(non_negative) {
+        Ok(size) => size,
+        Err(_) => 0,
+    }
+}
+
 pub fn worker_counts_for_priority(
    priority: &ReplicationPriority,
    current_workers: usize,
@@ -189,6 +197,13 @@ mod tests {
        );
    }

+    #[test]
+    fn mrf_worker_size_to_count_clamps_negative_values() {
+        assert_eq!(mrf_worker_size_to_count(-1), 0);
+        assert_eq!(mrf_worker_size_to_count(0), 0);
+        assert_eq!(mrf_worker_size_to_count(3), 3);
+    }
+
    #[test]
    fn auto_priority_grows_toward_defaults() {
        assert_eq!(
@@ -2582,7 +2582,7 @@ fi
 (
  cd "$ROOT_DIR"
  replication_delete_worker_status=0
-  rg -n --with-filename '^\s*(?:pub(?:\([^)]*\))?\s+)?struct\s+DeletedObjectReplicationInfo\b' \
+  rg -n --with-filename '^\s*(?:pub(?:\([^)]*\))?\s+)?(?:struct\s+DeletedObjectReplicationInfo|fn\s+(?:is_version_delete_replication|should_retry_delete_marker_purge|is_retryable_delete_replication_head_error))\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
@@ -2627,7 +2627,7 @@ fi
 (
  cd "$ROOT_DIR"
  replication_runtime_status=0
-  rg -n --with-filename '^\s*(?:pub(?:\([^)]*\))?\s+)?(?:(?:const\s+(?:WORKER_MAX_LIMIT|WORKER_MIN_LIMIT|WORKER_AUTO_DEFAULT|MRF_WORKER_MAX_LIMIT|MRF_WORKER_MIN_LIMIT|MRF_WORKER_AUTO_DEFAULT|LARGE_WORKER_COUNT|MIN_LARGE_OBJ_SIZE)\b)|(?:struct\s+(?:ReplicationPoolOpts|ReplicationWorkerCounts)\b)|(?:fn\s+(?:initial_worker_counts|resized_worker_counts|worker_counts_for_priority|should_queue_large_object|should_grow_large_workers|next_large_worker_count|next_regular_worker_count|next_mrf_worker_count)\b))' \
+  rg -n --with-filename '^\s*(?:pub(?:\([^)]*\))?\s+)?(?:(?:const\s+(?:WORKER_MAX_LIMIT|WORKER_MIN_LIMIT|WORKER_AUTO_DEFAULT|MRF_WORKER_MAX_LIMIT|MRF_WORKER_MIN_LIMIT|MRF_WORKER_AUTO_DEFAULT|LARGE_WORKER_COUNT|MIN_LARGE_OBJ_SIZE)\b)|(?:struct\s+(?:ReplicationPoolOpts|ReplicationWorkerCounts)\b)|(?:fn\s+(?:initial_worker_counts|resized_worker_counts|mrf_worker_size_to_count|worker_counts_for_priority|should_queue_large_object|should_grow_large_workers|next_large_worker_count|next_regular_worker_count|next_mrf_worker_count)\b))' \
    crates/ecstore/src/bucket/replication \
    --glob '*.rs' >"$REPLICATION_RUNTIME_CONTRACT_BACKSLIDE_HITS_FILE" || replication_runtime_status=$?
  if [[ "$replication_runtime_status" -ne 0 && "$replication_runtime_status" -ne 1 ]]; then
@@ -2641,10 +2641,14 @@ fi

 (
  cd "$ROOT_DIR"
-  rg -n --with-filename 'pub\s+(struct|enum)\s+(ResyncOpts|TargetReplicationResyncStatus|BucketReplicationResyncStatus|ResyncStatusType)\b' \
+  replication_resync_status=0
+  rg -n --with-filename '^\s*(?:pub(?:\([^)]*\))?\s+)?(?:(?:struct|enum)\s+(?:ResyncOpts|TargetReplicationResyncStatus|BucketReplicationResyncStatus|ResyncStatusType)|fn\s+(?:resync_state_accepts_update|should_count_head_proxy_failure|is_version_id_mismatch))\b' \
    crates/ecstore/src/bucket/replication \
-    --glob '*.rs' || true
-) >"$REPLICATION_RESYNC_CONTRACT_BACKSLIDE_HITS_FILE"
+    --glob '*.rs' >"$REPLICATION_RESYNC_CONTRACT_BACKSLIDE_HITS_FILE" || replication_resync_status=$?
+  if [[ "$replication_resync_status" -ne 0 && "$replication_resync_status" -ne 1 ]]; then
+    exit "$replication_resync_status"
+  fi
+)

 if [[ -s "$REPLICATION_RESYNC_CONTRACT_BACKSLIDE_HITS_FILE" ]]; then
  report_failure "resync DTO contracts must stay in crates/replication: $(paste -sd '; ' "$REPLICATION_RESYNC_CONTRACT_BACKSLIDE_HITS_FILE")"