From 921ddef2c7426ec813cd449da64b6c8ceadd7c99 Mon Sep 17 00:00:00 2001 From: cxymds Date: Sun, 2 Aug 2026 19:27:39 +0800 Subject: [PATCH] fix(lifecycle): bind delete replication admission (#5621) --- .../bucket/lifecycle/bucket_lifecycle_ops.rs | 269 ++++++------------ .../src/bucket/lifecycle/replication_sink.rs | 120 +++++++- crates/ecstore/src/bucket/replication/mod.rs | 2 - .../src/client/object_handlers_common.rs | 50 +--- 4 files changed, 199 insertions(+), 242 deletions(-) diff --git a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs index 639e03f4c..75890bb15 100644 --- a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs +++ b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs @@ -33,8 +33,7 @@ use crate::bucket::lifecycle::manual_transition_job::{ }; use crate::bucket::lifecycle::replication_sink; use crate::bucket::lifecycle::replication_sink::{ - ReplicateDecision, ReplicationState, ReplicationStatusType, VersionPurgeStatusType, replication_state_to_filemeta, - replication_statuses_map, version_purge_statuses_map, + DeleteReplicationConfigSnapshot, ReplicationObjectBridge, ReplicationStatusType, replication_state_to_filemeta, }; use crate::bucket::lifecycle::tier_delete_journal::{process_tier_delete_journal_entry, run_tier_delete_journal_recovery_loop}; use crate::bucket::lifecycle::tier_free_version_recovery::{ @@ -43,6 +42,7 @@ use crate::bucket::lifecycle::tier_free_version_recovery::{ use crate::bucket::lifecycle::tier_last_day_stats::{DailyAllTierStats, LastDayTierStats}; use crate::bucket::lifecycle::tier_sweeper::{Jentry, delete_object_from_remote_tier_idempotent_with_manager_and_identity}; use crate::bucket::lifecycle::transition_transaction::run_transition_transaction_recovery_loop; +use crate::bucket::versioning::VersioningApi as _; use crate::bucket::versioning_sys::BucketVersioningSys; use crate::client::object_api_utils::new_getobjectreader; use crate::disk::error::DiskError; @@ -4079,12 +4079,12 @@ pub async fn expire_transitioned_object( lc_event: &lifecycle::Event, _src: &LcEventSrc, ) -> Result { - let opts = transitioned_object_delete_opts( - oi, - lc_event.action, - BucketVersioningSys::prefix_enabled(&oi.bucket, &oi.name).await, - BucketVersioningSys::prefix_suspended(&oi.bucket, &oi.name).await, - ); + let snapshot = lifecycle_delete_config_snapshot(&api, oi) + .await + .map_err(std::io::Error::other)?; + let (versioned, version_suspended) = snapshot.versioning_config().delete_state(&oi.name); + let mut opts = transitioned_object_delete_opts(oi, lc_event.action, versioned, version_suspended); + opts.delete_replication_config_snapshot = Some(Arc::new(snapshot)); //let tags = LcAuditEvent::new(src, lcEvent).Tags(); if lc_event.action.delete_restored() { return match api.delete_object(&oi.bucket, &oi.name, opts).await { @@ -4692,8 +4692,28 @@ pub async fn apply_expiry_on_non_transitioned_objects( lc_event: &lifecycle::Event, _src: &LcEventSrc, ) -> bool { + let snapshot = match lifecycle_delete_config_snapshot(&api, oi).await { + Ok(snapshot) => snapshot, + Err(err) => { + error!( + event = EVENT_LIFECYCLE_DELETE_FAILED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, + bucket = %oi.bucket, + object = %oi.name, + operation = "load_delete_config_snapshot", + error = ?err, + "Lifecycle delete admission failed" + ); + return false; + } + }; + let (versioned, version_suspended) = snapshot.versioning_config().delete_state(&oi.name); let mut opts = ObjectOptions { + versioned, + version_suspended, expiration: ExpirationOptions { expire: true }, + delete_replication_config_snapshot: Some(Arc::new(snapshot)), ..Default::default() }; @@ -4701,9 +4721,6 @@ pub async fn apply_expiry_on_non_transitioned_objects( opts.version_id = oi.version_id.map(|v| v.to_string()); } - opts.versioned = BucketVersioningSys::prefix_enabled(&oi.bucket, &oi.name).await; - opts.version_suspended = BucketVersioningSys::prefix_suspended(&oi.bucket, &oi.name).await; - if lc_event.action.delete_all() { opts.delete_prefix = true; opts.delete_prefix_object = true; @@ -4765,12 +4782,17 @@ pub async fn apply_expiry_rule(event: &lifecycle::Event, src: &LcEventSrc, oi: & } fn lifecycle_deleted_object(oi: &ObjectInfo, dobj: &ObjectInfo) -> DeletedObject { + let replication_state = dobj.replication_state(); + let replication_state = (!replication_state.targets.is_empty() || !replication_state.purge_targets.is_empty()) + .then(|| replication_state_to_filemeta(&replication_state)); + if dobj.delete_marker { return DeletedObject { object_name: oi.name.clone(), delete_marker: true, delete_marker_version_id: dobj.version_id, delete_marker_mtime: dobj.mod_time.or(oi.mod_time), + replication_state, ..Default::default() }; } @@ -4781,6 +4803,7 @@ fn lifecycle_deleted_object(oi: &ObjectInfo, dobj: &ObjectInfo) -> DeletedObject delete_marker: false, delete_marker_version_id: oi.version_id, delete_marker_mtime: oi.mod_time, + replication_state, ..Default::default() }; } @@ -4790,106 +4813,21 @@ fn lifecycle_deleted_object(oi: &ObjectInfo, dobj: &ObjectInfo) -> DeletedObject delete_marker: false, version_id: oi.version_id, delete_marker_mtime: oi.mod_time, + replication_state, ..Default::default() } } async fn schedule_lifecycle_replication_delete_if_needed(oi: &ObjectInfo, dobj: &ObjectInfo) { - let mut delete_object = lifecycle_deleted_object(oi, dobj); - let version_id = if delete_object.delete_marker { - None - } else if delete_object.delete_marker_version_id.is_some() { - delete_object.delete_marker_version_id - } else { - delete_object.version_id - }; - - let replication_state = lifecycle_delete_replication_state(oi, version_id).await; - if replication_state.is_none() { + let delete_object = lifecycle_deleted_object(oi, dobj); + if delete_object.replication_state.is_none() { return; } - - delete_object.replication_state = replication_state.as_ref().map(replication_state_to_filemeta); - replication_sink::schedule_delete(oi.bucket.clone(), delete_object).await; } -fn should_reuse_lifecycle_delete_replication_state(oi: &ObjectInfo, version_delete: bool) -> bool { - let state = oi.replication_state(); - if version_delete { - oi.version_purge_status == VersionPurgeStatusType::Pending && !state.purge_targets.is_empty() - } else { - oi.replication_status == ReplicationStatusType::Pending && !state.targets.is_empty() - } -} - -fn lifecycle_version_purge_state_from_completed_targets(oi: &ObjectInfo) -> Option { - if oi.replication_status != ReplicationStatusType::Completed { - return None; - } - - let targets = oi.replication_state().targets; - if targets.is_empty() { - return None; - } - - let pending_status = targets.keys().map(|arn| format!("{arn}=PENDING;")).collect::(); - - Some(ReplicationState { - replicate_decision_str: oi.replication_decision.clone(), - version_purge_status_internal: Some(pending_status.clone()), - purge_targets: version_purge_statuses_map(&pending_status), - ..Default::default() - }) -} - -async fn lifecycle_delete_replication_state(oi: &ObjectInfo, version_id: Option) -> Option { - if should_reuse_lifecycle_delete_replication_state(oi, version_id.is_some()) { - return Some(oi.replication_state()); - } - - if version_id.is_some() - && let Some(state) = lifecycle_version_purge_state_from_completed_targets(oi) - { - return Some(state); - } - - let dsc = replication_sink::check_delete_replication( - &oi.bucket, - ObjectToDelete { - object_name: oi.name.clone(), - version_id, - ..Default::default() - }, - oi, - &ObjectOptions { - version_id: version_id.map(|v| v.to_string()), - versioned: BucketVersioningSys::prefix_enabled(&oi.bucket, &oi.name).await, - ..Default::default() - }, - ) - .await; - if !dsc.replicate_any() { - return None; - } - - Some(replication_state_for_delete(dsc, version_id.is_some())) -} - -fn replication_state_for_delete(dsc: ReplicateDecision, version_delete: bool) -> ReplicationState { - let pending_status = dsc.pending_status(); - let mut state = ReplicationState { - replicate_decision_str: dsc.to_string(), - ..Default::default() - }; - if version_delete { - state.version_purge_status_internal = pending_status.clone(); - state.purge_targets = version_purge_statuses_map(pending_status.as_deref().unwrap_or_default()); - } else { - state.replication_status_internal = pending_status.clone(); - state.targets = replication_statuses_map(pending_status.as_deref().unwrap_or_default()); - } - state +async fn lifecycle_delete_config_snapshot(api: &ECStore, oi: &ObjectInfo) -> Result { + ReplicationObjectBridge::delete_request_config(api, &oi.bucket).await } pub async fn apply_lifecycle_action(event: &lifecycle::Event, src: &LcEventSrc, oi: &ObjectInfo) -> bool { @@ -4925,16 +4863,14 @@ mod tests { enqueue_transition_with_lifecycle, enqueue_transition_with_lifecycle_report, eval_action_from_lifecycle, jitter_tier_free_version_recovery_delay, lifecycle_action_blocked_by_replication, lifecycle_delete_all_versions_replication_scan, lifecycle_deleted_object, lifecycle_replication_blocks_action, - lifecycle_rule_has_date_expiration, lifecycle_version_purge_state_from_completed_targets, - manual_transition_duration_elapsed, manual_transition_has_more_after_limit, manual_transition_recovery_progress_sink, - manual_transition_version_marker, manual_transition_worker_failure_reason, + lifecycle_rule_has_date_expiration, manual_transition_duration_elapsed, manual_transition_has_more_after_limit, + manual_transition_recovery_progress_sink, manual_transition_version_marker, manual_transition_worker_failure_reason, mark_delete_opts_skip_decommissioned_on_remote_success, merge_stale_multipart_candidate, persist_manual_transition_job_progress, persist_manual_transition_page_checkpoint, recover_manual_transition_job, - recover_manual_transition_jobs, replication_state_for_delete, resolve_tier_free_version_recovery_enabled, - resolve_transition_queue_capacity, resolve_transition_queue_send_timeout, resolve_transition_worker_count, - resolve_transition_workers_absolute_max, run_tier_free_version_recovery_loop, select_restore_s3_location, - set_lifecycle_observability_observer, set_recovered_free_version_enqueue_observer, - should_defer_date_expiry_for_recent_config_update, should_reuse_lifecycle_delete_replication_state, + recover_manual_transition_jobs, resolve_tier_free_version_recovery_enabled, resolve_transition_queue_capacity, + resolve_transition_queue_send_timeout, resolve_transition_worker_count, resolve_transition_workers_absolute_max, + run_tier_free_version_recovery_loop, select_restore_s3_location, set_lifecycle_observability_observer, + set_recovered_free_version_enqueue_observer, should_defer_date_expiry_for_recent_config_update, transitioned_cleanup_tuple, transitioned_object_delete_opts, wait_for_tier_free_version_recovery, }; #[cfg(feature = "test-util")] @@ -4958,9 +4894,7 @@ mod tests { save_manual_transition_scope_admission_if_absent, save_manual_transition_scope_admission_if_current, save_manual_transition_task_if_absent, save_manual_transition_worker_result_if_absent, }; - use crate::bucket::lifecycle::replication_sink::{ - ReplicateDecision, ReplicateTargetDecision, ReplicationStatusType, VersionPurgeStatusType, - }; + use crate::bucket::lifecycle::replication_sink::{ReplicationStatusType, VersionPurgeStatusType}; use crate::bucket::lifecycle::runtime_boundary as runtime_sources; use crate::bucket::lifecycle::tier_free_version_recovery::{ FreeVersionRecoveryStats, RecoveryWalkTestAction, list_tier_free_versions, recover_tier_free_versions_with_cancel, @@ -7686,6 +7620,46 @@ mod tests { assert_eq!(deleted.object_name, "key"); } + #[test] + fn lifecycle_deleted_object_hands_off_only_persisted_delete_admission_state() { + let source = ObjectInfo { + bucket: "bucket".to_string(), + name: "key".to_string(), + ..Default::default() + }; + let marker_result = ObjectInfo { + delete_marker: true, + version_id: Some(Uuid::new_v4()), + replication_status_internal: Some("arn:target=PENDING;".to_string()), + replication_decision: "arn:target=true".to_string(), + ..Default::default() + }; + let marker_delete = lifecycle_deleted_object(&source, &marker_result); + let marker_state = marker_delete + .replication_state + .expect("persisted marker admission must be handed off"); + assert_eq!(marker_state.replication_status_internal.as_deref(), Some("arn:target=PENDING;")); + assert!(marker_state.version_purge_status_internal.is_none()); + + let version_result = ObjectInfo { + version_purge_status_internal: Some("arn:target=PENDING;".to_string()), + replication_decision: "arn:target=true".to_string(), + ..Default::default() + }; + let version_delete = lifecycle_deleted_object( + &ObjectInfo { + version_id: Some(Uuid::new_v4()), + ..source + }, + &version_result, + ); + let version_state = version_delete + .replication_state + .expect("persisted version purge admission must be handed off"); + assert!(version_state.replication_status_internal.is_none()); + assert_eq!(version_state.version_purge_status_internal.as_deref(), Some("arn:target=PENDING;")); + } + #[test] fn lifecycle_deleted_object_uses_version_id_for_noncurrent_version_purge() { let version_id = Uuid::new_v4(); @@ -7721,77 +7695,6 @@ mod tests { assert_eq!(deleted.version_id, None); } - #[test] - fn replication_state_for_delete_uses_replication_targets_for_current_delete() { - let arn = "arn:aws:s3:::target-bucket"; - let mut dsc = ReplicateDecision::default(); - dsc.set(ReplicateTargetDecision::new(arn.to_string(), true, false)); - - let state = replication_state_for_delete(dsc, false); - - assert_eq!(state.replication_status_internal.as_deref(), Some(format!("{arn}=PENDING;").as_str())); - assert!(state.version_purge_status_internal.is_none()); - assert!(state.targets.contains_key(arn)); - } - - #[test] - fn replication_state_for_delete_uses_purge_targets_for_version_delete() { - let arn = "arn:aws:s3:::target-bucket"; - let mut dsc = ReplicateDecision::default(); - dsc.set(ReplicateTargetDecision::new(arn.to_string(), true, false)); - - let state = replication_state_for_delete(dsc, true); - - assert_eq!(state.version_purge_status_internal.as_deref(), Some(format!("{arn}=PENDING;").as_str())); - assert!(state.replication_status_internal.is_none()); - assert!(state.purge_targets.contains_key(arn)); - } - - #[test] - fn lifecycle_delete_replication_state_reuses_only_pending_version_purge_state() { - let oi = ObjectInfo { - version_purge_status: VersionPurgeStatusType::Pending, - version_purge_status_internal: Some("arn:aws:s3:::target=PENDING;".to_string()), - replication_decision: "arn:aws:s3:::target=true;false;arn:aws:s3:::target;".to_string(), - ..Default::default() - }; - - assert!(should_reuse_lifecycle_delete_replication_state(&oi, true)); - assert!(!should_reuse_lifecycle_delete_replication_state(&oi, false)); - } - - #[test] - fn lifecycle_delete_replication_state_does_not_reuse_put_replication_for_version_delete() { - let oi = ObjectInfo { - replication_status: ReplicationStatusType::Completed, - replication_status_internal: Some("arn:aws:s3:::target=COMPLETED;".to_string()), - replication_decision: "arn:aws:s3:::target=true;false;arn:aws:s3:::target;".to_string(), - ..Default::default() - }; - - assert!( - !should_reuse_lifecycle_delete_replication_state(&oi, true), - "version purges must not reuse plain object replication state from prior PUT/delete-marker replication" - ); - } - - #[test] - fn lifecycle_version_purge_state_from_completed_targets_derives_pending_purge_targets() { - let oi = ObjectInfo { - replication_status: ReplicationStatusType::Completed, - replication_status_internal: Some("arn:aws:s3:::target=COMPLETED;".to_string()), - replication_decision: "arn:aws:s3:::target=true;false;arn:aws:s3:::target;".to_string(), - ..Default::default() - }; - - let state = lifecycle_version_purge_state_from_completed_targets(&oi) - .expect("completed replication targets should be convertible into version-purge targets"); - - 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, oi.replication_decision); - } - fn expired_delete_marker_lifecycle() -> BucketLifecycleConfiguration { BucketLifecycleConfiguration { expiry_updated_at: None, diff --git a/crates/ecstore/src/bucket/lifecycle/replication_sink.rs b/crates/ecstore/src/bucket/lifecycle/replication_sink.rs index 349472743..a965988e1 100644 --- a/crates/ecstore/src/bucket/lifecycle/replication_sink.rs +++ b/crates/ecstore/src/bucket/lifecycle/replication_sink.rs @@ -15,15 +15,14 @@ use rustfs_common::metrics::IlmAction; use crate::bucket::lifecycle::lifecycle::ObjectOpts; +pub(crate) use crate::bucket::replication::ReplicationStatusType; #[cfg(test)] -pub(crate) use crate::bucket::replication::ReplicateTargetDecision; +pub(crate) use crate::bucket::replication::VersionPurgeStatusType; pub(crate) use crate::bucket::replication::{ - ReplicateDecision, ReplicationState, ReplicationStatusType, VersionPurgeStatusType, replication_state_to_filemeta, - replication_statuses_map, version_purge_statuses_map, + DeleteReplicationConfigSnapshot, ReplicationObjectBridge, replication_state_to_filemeta, }; use crate::bucket::replication::{ReplicationLifecycleBridge, ReplicationLifecycleConfig}; -use crate::object_api::{ObjectInfo, ObjectOptions}; -use crate::storage_api_contracts::object::{DeletedObject, ObjectToDelete}; +use crate::storage_api_contracts::object::DeletedObject; pub(crate) type LifecycleReplicationConfig = ReplicationLifecycleConfig; @@ -57,15 +56,6 @@ pub(crate) fn lifecycle_action_waits_for_replication(action: IlmAction) -> bool ) } -pub(crate) async fn check_delete_replication( - bucket: &str, - object: ObjectToDelete, - source: &ObjectInfo, - opts: &ObjectOptions, -) -> ReplicateDecision { - ReplicationLifecycleBridge::check_delete_replication(bucket, &object, source, opts).await -} - pub(crate) async fn schedule_delete(bucket: String, delete_object: DeletedObject) { ReplicationLifecycleBridge::schedule_delete(bucket, delete_object).await; } @@ -74,7 +64,16 @@ pub(crate) async fn schedule_delete(bucket: String, delete_object: DeletedObject mod tests { use std::collections::HashMap; + use crate::bucket::replication::{DeleteReplicationConfigSnapshot, ReplicationObjectBridge}; + use crate::object_api::{ObjectInfo, ObjectOptions}; + use crate::storage_api_contracts::object::ObjectToDelete; use rustfs_common::metrics::IlmAction; + use s3s::dto::{ + BucketVersioningStatus, DeleteMarkerReplication, DeleteMarkerReplicationStatus, DeleteReplication, + DeleteReplicationStatus, Destination, ReplicationConfiguration, ReplicationRule, ReplicationRuleStatus, + VersioningConfiguration, + }; + use uuid::Uuid; use super::*; @@ -139,4 +138,97 @@ mod tests { assert!(lifecycle_action_waits_for_replication(IlmAction::TransitionVersionAction)); assert!(!lifecycle_action_waits_for_replication(IlmAction::NoneAction)); } + + #[test] + fn lifecycle_delete_admission_uses_marker_and_version_switches_for_all_purges() { + for marker_enabled in [false, true] { + for purge_enabled in [false, true] { + let snapshot = DeleteReplicationConfigSnapshot::from_configs_for_test( + VersioningConfiguration { + status: Some(BucketVersioningStatus::from_static(BucketVersioningStatus::ENABLED)), + ..Default::default() + }, + Some(ReplicationConfiguration { + role: String::new(), + rules: vec![ReplicationRule { + delete_marker_replication: Some(DeleteMarkerReplication { + status: Some(if marker_enabled { + DeleteMarkerReplicationStatus::from_static(DeleteMarkerReplicationStatus::ENABLED) + } else { + DeleteMarkerReplicationStatus::from_static(DeleteMarkerReplicationStatus::DISABLED) + }), + }), + delete_replication: Some(DeleteReplication { + status: if purge_enabled { + DeleteReplicationStatus::from_static(DeleteReplicationStatus::ENABLED) + } else { + DeleteReplicationStatus::from_static(DeleteReplicationStatus::DISABLED) + }, + }), + destination: Destination { + bucket: "arn:rustfs:replication:target".to_string(), + ..Default::default() + }, + existing_object_replication: None, + filter: None, + id: Some("lifecycle-delete-switches".to_string()), + prefix: Some(String::new()), + priority: Some(1), + source_selection_criteria: None, + status: ReplicationRuleStatus::from_static(ReplicationRuleStatus::ENABLED), + }], + }), + ); + let source = ObjectInfo { + bucket: "bucket".to_string(), + name: "logs/object".to_string(), + ..Default::default() + }; + let marker = ObjectToDelete { + object_name: source.name.clone(), + ..Default::default() + }; + let marker_opts = ObjectOptions { + versioned: true, + ..Default::default() + }; + assert_eq!( + ReplicationObjectBridge::check_delete_with_snapshot(&marker, &source, &marker_opts, false, &snapshot) + .replicate_any(), + marker_enabled + ); + + for delete_marker in [false, true] { + for version_id in [Uuid::new_v4(), Uuid::nil()] { + let purge = ObjectToDelete { + object_name: source.name.clone(), + version_id: Some(version_id), + ..Default::default() + }; + let purge_source = ObjectInfo { + delete_marker, + ..source.clone() + }; + let purge_opts = ObjectOptions { + version_id: Some(version_id.to_string()), + versioned: true, + ..Default::default() + }; + assert_eq!( + ReplicationObjectBridge::check_delete_with_snapshot( + &purge, + &purge_source, + &purge_opts, + false, + &snapshot, + ) + .replicate_any(), + purge_enabled, + "delete marker={delete_marker}, version_id={version_id}" + ); + } + } + } + } + } } diff --git a/crates/ecstore/src/bucket/replication/mod.rs b/crates/ecstore/src/bucket/replication/mod.rs index 741f0a258..346eff2ef 100644 --- a/crates/ecstore/src/bucket/replication/mod.rs +++ b/crates/ecstore/src/bucket/replication/mod.rs @@ -49,8 +49,6 @@ pub use replication_config_boundary::{ replication_target_arns, should_remove_replication_target, unsupported_replication_config_field, validate_replication_config_target_arns, }; -#[cfg(test)] -pub(crate) use replication_filemeta_boundary::ReplicateTargetDecision; pub(crate) use replication_filemeta_boundary::version_purge_statuses_map; pub use replication_filemeta_boundary::{ MrfOpKind, MrfReplicateEntry, REPLICATE_INCOMING_DELETE, ReplicateDecision, ReplicateObjectInfo, ReplicationState, diff --git a/crates/ecstore/src/client/object_handlers_common.rs b/crates/ecstore/src/client/object_handlers_common.rs index 7ac4595a4..44fa67db4 100644 --- a/crates/ecstore/src/client/object_handlers_common.rs +++ b/crates/ecstore/src/client/object_handlers_common.rs @@ -21,17 +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::{ReplicationLifecycleBridge, ReplicationState, replication_state_to_filemeta}; -use crate::bucket::versioning::VersioningApi; -use crate::bucket::versioning_sys::BucketVersioningSys; +use crate::bucket::replication::{ReplicationLifecycleBridge, ReplicationObjectBridge}; use crate::object_api::ObjectOptions; use crate::storage_api_contracts::object::{ObjectOperations as _, ObjectToDelete}; use crate::store::ECStore; use rustfs_lock::MAX_DELETE_LIST; 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(), + let delete_config_snapshot = match ReplicationObjectBridge::delete_request_config(api, bucket).await { + Ok(snapshot) => Arc::new(snapshot), Err(err) => { debug!( event = EVENT_LIFECYCLE_CLEANUP_SKIPPED, @@ -39,7 +37,7 @@ pub async fn delete_object_versions(api: &Arc, bucket: &str, to_del: &[ subsystem = LOG_SUBSYSTEM_LIFECYCLE, bucket, error = ?err, - reason = "versioning_config_unavailable", + reason = "delete_config_snapshot_unavailable", "Skipped lifecycle noncurrent version cleanup" ); return; @@ -55,45 +53,12 @@ pub async fn delete_object_versions(api: &Arc, bucket: &str, to_del: &[ remaining = &[]; } - let mut replication_candidates: Vec> = Vec::with_capacity(to_del.len()); - for object in to_del.iter() { - let version_id = object.version_id.map(|vid| vid.to_string()); - let opts = ObjectOptions { - version_id: version_id.clone(), - versioned: true, - version_suspended, - ..Default::default() - }; - let candidate = match api.get_object_info(bucket, &object.object_name, &opts).await { - Ok(info) => { - let dsc = ReplicationLifecycleBridge::check_delete_replication(bucket, object, &info, &opts).await; - dsc.replicate_any() - .then(|| ReplicationLifecycleBridge::version_delete_replication_state(&dsc)) - } - Err(err) => { - debug!( - event = EVENT_LIFECYCLE_CLEANUP_SKIPPED, - component = LOG_COMPONENT_ECSTORE, - subsystem = LOG_SUBSYSTEM_LIFECYCLE, - bucket, - object = %object.object_name, - version_id = ?version_id, - error = ?err, - reason = "object_info_unavailable", - "Skipped lifecycle delete replication scheduling" - ); - None - } - }; - replication_candidates.push(candidate); - } - let (mut deleted_objs, errors) = api .delete_objects( bucket, to_del.to_vec(), ObjectOptions { - version_suspended, + delete_replication_config_snapshot: Some(Arc::clone(&delete_config_snapshot)), ..Default::default() }, ) @@ -108,10 +73,9 @@ pub async fn delete_object_versions(api: &Arc, bucket: &str, to_del: &[ if let Some(target) = to_del.get(i) { crate::object_api::notify_object_mutation(bucket, &target.object_name).await; } - let Some(replication_state) = replication_candidates.get(i).and_then(|c| c.clone()) else { + if deleted_obj.replication_state.is_none() { continue; - }; - deleted_obj.replication_state = Some(replication_state_to_filemeta(&replication_state)); + } ReplicationLifecycleBridge::schedule_delete(bucket.to_string(), deleted_obj.clone()).await; }