diff --git a/crates/ecstore/src/bucket/replication/replication_pool.rs b/crates/ecstore/src/bucket/replication/replication_pool.rs index 27f933441..edfef7e99 100644 --- a/crates/ecstore/src/bucket/replication/replication_pool.rs +++ b/crates/ecstore/src/bucket/replication/replication_pool.rs @@ -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 ReplicationPool { max_l_workers: Option, ) { 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 { diff --git a/crates/ecstore/src/bucket/replication/replication_resyncer.rs b/crates/ecstore/src/bucket/replication/replication_resyncer.rs index 52e402637..cc325eb93 100644 --- a/crates/ecstore/src/bucket/replication/replication_resyncer.rs +++ b/crates/ecstore/src/bucket/replication/replication_resyncer.rs @@ -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) -> 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, 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) -> 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) -> 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(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) -> ReplicatedTargetInfo { let version_id = if let Some(version_id) = &dobj.delete_object.delete_marker_version_id { version_id.to_owned() diff --git a/crates/replication/src/delete.rs b/crates/replication/src/delete.rs index 8cd6f7670..319d1f1d7 100644 --- a/crates/replication/src/delete.rs +++ b/crates/replication/src/delete.rs @@ -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"))); + } } diff --git a/crates/replication/src/lib.rs b/crates/replication/src/lib.rs index dc77f4920..1f8984c3e 100644 --- a/crates/replication/src/lib.rs +++ b/crates/replication/src/lib.rs @@ -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, diff --git a/crates/replication/src/resync.rs b/crates/replication/src/resync.rs index 3d2b350e2..2bd5b2896 100644 --- a/crates/replication/src/resync.rs +++ b/crates/replication/src/resync.rs @@ -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) -> 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) -> 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(¤t, &matching)); + assert!(!resync_state_accepts_update(¤t, &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(); diff --git a/crates/replication/src/runtime.rs b/crates/replication/src/runtime.rs index 898e362b1..6876df660 100644 --- a/crates/replication/src/runtime.rs +++ b/crates/replication/src/runtime.rs @@ -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!( diff --git a/scripts/check_architecture_migration_rules.sh b/scripts/check_architecture_migration_rules.sh index 911501ccd..5f09d9fef 100755 --- a/scripts/check_architecture_migration_rules.sh +++ b/scripts/check_architecture_migration_rules.sh @@ -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")"