diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index 9febf4071..652ccae02 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/crates/ecstore/src/api/mod.rs @@ -49,7 +49,7 @@ pub mod bucket { apply_transition_rule, enqueue_expiry_for_existing_objects, enqueue_transition_for_existing_objects, enqueue_transition_for_existing_objects_scoped, enqueue_transition_for_existing_objects_scoped_with_cancel, enqueue_transition_immediate, expire_transitioned_object, get_global_expiry_state, get_global_transition_state, - init_background_expiry, manual_transition_queue_snapshot, post_restore_opts, + init_background_expiry, lifecycle_version_delete_target, manual_transition_queue_snapshot, post_restore_opts, run_stale_multipart_upload_cleanup_once, validate_transition_tier, }; } diff --git a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs index 89ccf5166..1649552c6 100644 --- a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs +++ b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs @@ -1141,7 +1141,7 @@ impl ExpiryState { let version_count = u64::try_from(v.versions.len()).unwrap_or(u64::MAX); let trace = LifecycleExpiryTrace::for_batch(&v.bucket, &v.event, &v.src, version_count); trace.emit(EVENT_LIFECYCLE_DELETE_DISPATCHED, "delete_dispatched", None); - crate::bucket::lifecycle::object_handlers_common::delete_object_versions( + let failed = crate::bucket::lifecycle::object_handlers_common::delete_object_versions( &api, &v.bucket, &v.versions, @@ -1149,7 +1149,19 @@ impl ExpiryState { v.bucket_incarnation_id, ) .await; - trace.emit(EVENT_LIFECYCLE_DELETE_COMPLETED, "delete_completed", None); + if failed == 0 { + trace.emit(EVENT_LIFECYCLE_DELETE_COMPLETED, "delete_completed", None); + } else { + record_scanner_lifecycle_expiry_delete_failed( + &v.src, + u64::try_from(failed).unwrap_or(u64::MAX), + ); + trace.emit( + EVENT_LIFECYCLE_DELETE_FAILED, + "delete_failed", + Some("delete_operation_failed"), + ); + } } else if v.as_any().is::() { let v = v.as_any().downcast_ref::().expect("Jentry downcast failed"); @@ -3680,13 +3692,11 @@ pub async fn enqueue_immediate_expiry(oi: &ObjectInfo, src: LcEventSrc) { enqueue_expiry_rule_with_incarnation(event, &src, object, configs.bucket_incarnation_id).await; } IlmAction::DeleteVersionAction => { - to_delete_objs.push(ObjectToDelete { - object_name: object.name.clone(), - version_id: object.version_id, - ..Default::default() - }); - if noncurrent_event.is_none() { - noncurrent_event = Some(event.clone()); + if let Some(target) = lifecycle_version_delete_target(object) { + to_delete_objs.push(target); + if noncurrent_event.is_none() { + noncurrent_event = Some(event.clone()); + } } } _ => {} @@ -4214,13 +4224,11 @@ async fn enqueue_expiry_for_existing_object_group( } if event.action == IlmAction::DeleteVersionAction { - to_delete_objs.push(ObjectToDelete { - object_name: object.name.clone(), - version_id: object.version_id, - ..Default::default() - }); - if noncurrent_event.is_none() { - noncurrent_event = Some(event.clone()); + if let Some(target) = lifecycle_version_delete_target(object) { + to_delete_objs.push(target); + if noncurrent_event.is_none() { + noncurrent_event = Some(event.clone()); + } } } else { let blocked_by_replication = match lifecycle_delete_all_versions_blocked_by_replication( @@ -4452,16 +4460,15 @@ fn transitioned_object_delete_opts( version_suspended: bool, bucket_incarnation_id: Uuid, ) -> crate::error::Result { + let version_id = lifecycle_expiry_target_version_id(oi, action, versioned, version_suspended)?; let mut opts = ObjectOptions { + version_id, versioned, version_suspended, expiration: ExpirationOptions { expire: true }, expected_bucket_incarnation_id: Some(bucket_incarnation_id), ..Default::default() }; - if action.delete_versioned() { - opts.version_id = oi.version_id.map(|id| id.to_string()); - } if action.delete_restored() { let etag = oi .etag @@ -4491,6 +4498,34 @@ fn transitioned_object_delete_opts( Ok(opts) } +fn lifecycle_expiry_target_version_id( + oi: &ObjectInfo, + action: IlmAction, + versioned: bool, + version_suspended: bool, +) -> crate::error::Result> { + // Delete-all revalidates the authoritative current trigger under the + // object write lock; queued snapshots do not always carry `is_latest`. + if !action.delete_all() + && !lifecycle::expiration_action_has_valid_target(action, oi.version_id, oi.is_latest, oi.delete_marker) + { + return Err(Error::other("lifecycle expiry action does not match the evaluated object identity")); + } + if matches!(action, IlmAction::DeleteAction | IlmAction::DeleteRestoredAction) + && (versioned || version_suspended) + && oi.version_id.is_none() + { + return Err(Error::other("current-version lifecycle expiry is missing its version identity")); + } + if action.delete_versioned() { + return oi + .version_id + .ok_or_else(|| Error::other("exact-version lifecycle expiry is missing its version identity")) + .map(|version_id| Some(version_id.to_string())); + } + Ok(None) +} + pub async fn expire_transitioned_object( api: Arc, oi: &ObjectInfo, @@ -4971,12 +5006,37 @@ impl RestoreRequestOps for RestoreRequest { const _MAX_RESTORE_OBJECT_REQUEST_SIZE: i64 = 2 << 20; +/// Builds an exact-version lifecycle delete target with the concrete +/// generation observed by the evaluator. The null S3 version ID is reusable, +/// so null data versions additionally require a write-unique data directory. +pub fn lifecycle_version_delete_target(oi: &ObjectInfo) -> Option { + let version_id = oi.version_id?; + if version_id.is_nil() + && ((!oi.delete_marker && oi.data_dir.is_none_or(|data_dir| data_dir.is_nil())) + || (oi.delete_marker && oi.mod_time.is_none_or(|mod_time| mod_time == OffsetDateTime::UNIX_EPOCH))) + { + return None; + } + + let target = ObjectToDelete { + object_name: oi.name.clone(), + version_id: Some(version_id), + ..Default::default() + }; + Some(if version_id.is_nil() { + target.with_expected_identity(oi.data_dir, oi.mod_time, oi.delete_marker) + } else { + target + }) +} + pub async fn eval_action_from_lifecycle( lc: &BucketLifecycleConfiguration, lock_config: Option<&ObjectLockConfiguration>, oi: &ObjectInfo, ) -> lifecycle::Event { - let event = lc.eval(&oi.to_lifecycle_opts()).await; + let lifecycle_opts = oi.to_lifecycle_opts(); + let event = lc.eval(&lifecycle_opts).await; debug!( event = EVENT_LIFECYCLE_SCAN_SKIPPED, component = LOG_COMPONENT_ECSTORE, @@ -4986,6 +5046,15 @@ pub async fn eval_action_from_lifecycle( "Evaluated lifecycle action during secondary scan" ); + if !lifecycle::expiration_action_has_valid_target( + event.action, + lifecycle_opts.version_id, + lifecycle_opts.is_latest, + lifecycle_opts.delete_marker, + ) { + return lifecycle::Event::default(); + } + let lock_enabled = lock_config.is_some_and(ObjectLockApi::enabled); let object_locked = object_lock_boundary::is_object_locked_by_metadata(&oi.user_defined, oi.delete_marker); @@ -4996,44 +5065,38 @@ pub async fn eval_action_from_lifecycle( IlmAction::DeleteAction | IlmAction::DeleteRestoredAction | IlmAction::DeleteVersionAction - | IlmAction::DeleteRestoredVersionAction => { - if matches!(event.action, IlmAction::DeleteVersionAction | IlmAction::DeleteRestoredVersionAction) - && oi.version_id.is_none() - { - return lifecycle::Event::default(); - } - // Destructive expiry never bypasses retention. Restore expiry only - // removes the local copy; the retained logical version remains. + | IlmAction::DeleteRestoredVersionAction if !event.action.delete_restored() && (object_locked || !matches!( object_lock_boundary::check_object_lock_for_deletion_with_config(lock_config, oi, false), Ok(None) - )) - { - //if serverDebugLog { - if oi.version_id.is_some() { - debug!( - event = EVENT_LIFECYCLE_SCAN_SKIPPED, - component = LOG_COMPONENT_ECSTORE, - subsystem = LOG_SUBSYSTEM_LIFECYCLE, - object = %oi.name, - version_id = %oi.version_id.map(|v| v.to_string()).unwrap_or_default(), - reason = "object_locked", - "Skipped lifecycle delete because object version is locked" - ); - } else { - debug!( - event = EVENT_LIFECYCLE_SCAN_SKIPPED, - component = LOG_COMPONENT_ECSTORE, - subsystem = LOG_SUBSYSTEM_LIFECYCLE, - object = %oi.name, - reason = "object_locked", - "Skipped lifecycle delete because object is locked" - ); - } - return lifecycle::Event::default(); + )) => + { + // Destructive expiry never bypasses retention. Restore expiry only + // removes the local copy; the retained logical version remains. + //if serverDebugLog { + if oi.version_id.is_some() { + debug!( + event = EVENT_LIFECYCLE_SCAN_SKIPPED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, + object = %oi.name, + version_id = %oi.version_id.map(|v| v.to_string()).unwrap_or_default(), + reason = "object_locked", + "Skipped lifecycle delete because object version is locked" + ); + } else { + debug!( + event = EVENT_LIFECYCLE_SCAN_SKIPPED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, + object = %oi.name, + reason = "object_locked", + "Skipped lifecycle delete because object is locked" + ); } + return lifecycle::Event::default(); } _ => (), } @@ -5250,7 +5313,12 @@ async fn apply_expiry_on_non_transitioned_objects_with_lock_lost_signal( } }; let (versioned, version_suspended) = snapshot.versioning_config().delete_state(&oi.name); + let version_id = match lifecycle_expiry_target_version_id(oi, lc_event.action, versioned, version_suspended) { + Ok(version_id) => version_id, + Err(_) => return false, + }; let mut opts = ObjectOptions { + version_id, versioned, version_suspended, expiration: ExpirationOptions { expire: true }, @@ -5263,10 +5331,6 @@ async fn apply_expiry_on_non_transitioned_objects_with_lock_lost_signal( opts.add_namespace_lock_lost_signal(signal); } - if lc_event.action.delete_versioned() { - opts.version_id = oi.version_id.map(|v| v.to_string()); - } - if lc_event.action.delete_all() { opts.delete_prefix = true; opts.delete_prefix_object = true; @@ -5521,14 +5585,14 @@ mod tests { enqueue_transition_with_lifecycle, enqueue_transition_with_lifecycle_report, eval_action_from_lifecycle, get_lock_acquire_timeout, 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, 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_if_owned, persist_manual_transition_page_checkpoint, - recover_manual_transition_job, 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, + lifecycle_rule_has_date_expiration, lifecycle_version_delete_target, 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_if_owned, + persist_manual_transition_page_checkpoint, recover_manual_transition_job, 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, }; @@ -5539,6 +5603,8 @@ mod tests { decode_manual_transition_continuation_token, encode_manual_transition_continuation_token, }; use crate::bucket::lifecycle::config_boundary; + use crate::bucket::lifecycle::evaluator::Evaluator; + use crate::bucket::lifecycle::lifecycle; use crate::bucket::lifecycle::manual_transition_job::{ ManualTransitionJobCasBarrier, ManualTransitionJobRecord, ManualTransitionJobState, ManualTransitionScopeAdmission, ManualTransitionScopeAdmissionClaim, ManualTransitionTaskRecord, ManualTransitionWorkerFailureReason, @@ -5583,7 +5649,7 @@ mod tests { lifecycle::ExpirationOptions, list::ListOperations as _, multipart::MultipartOperations as _, - object::{ObjectIO as _, ObjectOperations as _}, + object::{ObjectIO as _, ObjectOperations as _, ObjectToDelete}, }; use crate::store::ECStore; #[cfg(feature = "test-util")] @@ -5600,8 +5666,8 @@ mod tests { use rustfs_scanner_metrics::metrics::{IlmAction, global_metrics}; use s3s::dto::{ BucketLifecycleConfiguration, DefaultRetention, ExpirationStatus, LifecycleExpiration, LifecycleRule, MetadataEntry, - ObjectLockConfiguration, ObjectLockEnabled, ObjectLockRetentionMode, ObjectLockRule, OutputLocation, RestoreRequest, - RestoreRequestType, S3Location, Timestamp, Transition, TransitionStorageClass, + NoncurrentVersionExpiration, ObjectLockConfiguration, ObjectLockEnabled, ObjectLockRetentionMode, ObjectLockRule, + OutputLocation, RestoreRequest, RestoreRequestType, S3Location, Timestamp, Transition, TransitionStorageClass, }; use s3s::header::{X_AMZ_OBJECT_LOCK_LEGAL_HOLD, X_AMZ_OBJECT_LOCK_MODE, X_AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE}; use serial_test::serial; @@ -6537,6 +6603,7 @@ mod tests { bucket: "bucket".to_string(), name: "object".to_string(), version_id: Some(vid), + is_latest: true, data_dir: Some(Uuid::new_v4()), etag: Some("etag".to_string()), restore_expires: Some(OffsetDateTime::now_utc() - StdDuration::from_secs(1)), @@ -6548,10 +6615,14 @@ mod tests { }, ..Default::default() }; + let noncurrent = ObjectInfo { + is_latest: false, + ..oi.clone() + }; // Plain version expiry: exact version, real delete. let incarnation = Uuid::new_v4(); - let opts = transitioned_object_delete_opts(&oi, IlmAction::DeleteVersionAction, true, false, incarnation) + let opts = transitioned_object_delete_opts(&noncurrent, IlmAction::DeleteVersionAction, true, false, incarnation) .expect("build version expiry options"); assert_eq!(opts.version_id.as_deref(), Some(vid_str.as_str())); assert_eq!(opts.expected_bucket_incarnation_id, Some(incarnation)); @@ -6567,7 +6638,7 @@ mod tests { // Restore-expiry of a noncurrent version: restored-copy cleanup of the // exact version. Routing this through the full transitioned-object // delete instead would remove the remote tier data. - let opts = transitioned_object_delete_opts(&oi, IlmAction::DeleteRestoredVersionAction, true, false, incarnation) + let opts = transitioned_object_delete_opts(&noncurrent, IlmAction::DeleteRestoredVersionAction, true, false, incarnation) .expect("build restored-version expiry options"); assert_eq!(opts.version_id.as_deref(), Some(vid_str.as_str())); assert!(opts.transition.expire_restored); @@ -6577,6 +6648,45 @@ mod tests { .expect("build object expiry options"); assert!(opts.version_id.is_none()); assert!(!opts.transition.expire_restored); + + let historical_null = ObjectInfo { + version_id: Some(Uuid::nil()), + ..noncurrent + }; + let opts = transitioned_object_delete_opts(&historical_null, IlmAction::DeleteVersionAction, true, false, incarnation) + .expect("an explicit null version should remain an exact delete target"); + let null_version_id = Uuid::nil().to_string(); + assert_eq!(opts.version_id.as_deref(), Some(null_version_id.as_str())); + assert!( + transitioned_object_delete_opts(&historical_null, IlmAction::DeleteAction, true, false, incarnation).is_err(), + "a historical null version must never be routed as a current-object delete" + ); + + let versioned_current_without_identity = ObjectInfo { + version_id: None, + is_latest: true, + ..oi + }; + assert!( + transitioned_object_delete_opts( + &versioned_current_without_identity, + IlmAction::DeleteAction, + true, + false, + incarnation, + ) + .is_err(), + "a versioned current object without an identity is ambiguous" + ); + let opts = transitioned_object_delete_opts( + &versioned_current_without_identity, + IlmAction::DeleteAction, + false, + false, + incarnation, + ) + .expect("a truly unversioned current object should remain deletable"); + assert!(opts.version_id.is_none()); } #[tokio::test] @@ -8567,6 +8677,30 @@ mod tests { } } + fn current_and_noncurrent_expiration_lifecycle() -> BucketLifecycleConfiguration { + BucketLifecycleConfiguration { + expiry_updated_at: None, + rules: vec![LifecycleRule { + status: ExpirationStatus::from_static(ExpirationStatus::ENABLED), + expiration: Some(LifecycleExpiration { + days: Some(30), + ..Default::default() + }), + abort_incomplete_multipart_upload: None, + del_marker_expiration: None, + filter: None, + id: Some("expire-current-and-noncurrent".to_string()), + noncurrent_version_expiration: Some(NoncurrentVersionExpiration { + noncurrent_days: Some(1), + newer_noncurrent_versions: None, + }), + noncurrent_version_transitions: None, + prefix: None, + transitions: None, + }], + } + } + fn all_versions_expiration_lifecycle() -> BucketLifecycleConfiguration { BucketLifecycleConfiguration { expiry_updated_at: None, @@ -11074,6 +11208,315 @@ mod tests { assert_eq!(event.action, IlmAction::DeleteAction); } + #[tokio::test] + async fn lifecycle_evaluators_agree_on_null_version_identity_and_guards() { + let lifecycle = current_and_noncurrent_expiration_lifecycle(); + let now = OffsetDateTime::now_utc(); + let historical_null = ObjectInfo { + bucket: "bucket".to_string(), + name: "logs/object".to_string(), + mod_time: Some(now - time::Duration::days(40)), + successor_mod_time: Some(now - time::Duration::days(3)), + version_id: Some(Uuid::nil()), + is_latest: false, + num_versions: 1, + replication_status: ReplicationStatusType::Completed, + ..Default::default() + }; + let cases = [ + ("explicit historical null", historical_null.clone(), None, IlmAction::DeleteVersionAction), + ( + "missing historical identity", + ObjectInfo { + version_id: None, + ..historical_null.clone() + }, + None, + IlmAction::NoneAction, + ), + ( + "pending null replication", + ObjectInfo { + replication_status: ReplicationStatusType::Pending, + ..historical_null.clone() + }, + None, + IlmAction::NoneAction, + ), + ( + "locked historical null", + ObjectInfo { + user_defined: Arc::new(HashMap::from([( + X_AMZ_OBJECT_LOCK_LEGAL_HOLD.as_str().to_string(), + "ON".to_string(), + )])), + ..historical_null.clone() + }, + Some(lock_enabled_without_default_retention()), + IlmAction::NoneAction, + ), + ( + "unversioned current", + ObjectInfo { + version_id: None, + is_latest: true, + successor_mod_time: None, + ..historical_null + }, + None, + IlmAction::DeleteAction, + ), + ]; + + for (name, object, lock_config, expected) in cases { + let object_opts = lifecycle::object_opts_from_object_info(&object); + let batch_event = Evaluator::new(Arc::new(lifecycle.clone())) + .with_lock_retention(lock_config.clone().map(Arc::new)) + .eval(&[object_opts]) + .await + .expect("batch lifecycle evaluation should succeed") + .remove(0); + let secondary_event = eval_action_from_lifecycle(&lifecycle, lock_config.as_ref(), &object).await; + + assert_eq!(batch_event.action, expected, "batch evaluator mismatch for {name}"); + assert_eq!(secondary_event.action, expected, "secondary evaluator mismatch for {name}"); + } + } + + #[tokio::test] + #[serial] + async fn lifecycle_deletes_only_the_historical_null_version_after_versioning_is_reenabled() { + let (_disk_paths, ecstore) = setup_test_env().await; + let bucket = format!("lifecycle-null-history-{}", Uuid::new_v4().simple()); + let object = "logs/object"; + create_test_bucket(&ecstore, &bucket).await; + + metadata_sys::update_in( + &ecstore.ctx, + &bucket, + BUCKET_VERSIONING_CONFIG, + b"Enabled".to_vec(), + ) + .await + .expect("bucket versioning should first be enabled"); + metadata_sys::update_in( + &ecstore.ctx, + &bucket, + BUCKET_VERSIONING_CONFIG, + b"Suspended".to_vec(), + ) + .await + .expect("bucket versioning should be suspended"); + + let now = OffsetDateTime::now_utc(); + let mut null_reader = PutObjReader::from_vec(b"historical-null".to_vec()); + let null_version = ecstore + .put_object( + &bucket, + object, + &mut null_reader, + &ObjectOptions { + version_suspended: true, + mod_time: Some(now - time::Duration::days(40)), + ..Default::default() + }, + ) + .await + .expect("suspended PUT should create a null version"); + assert_eq!(null_version.version_id, Some(Uuid::nil())); + + metadata_sys::update_in( + &ecstore.ctx, + &bucket, + BUCKET_VERSIONING_CONFIG, + b"Enabled".to_vec(), + ) + .await + .expect("bucket versioning should be re-enabled"); + let active_bytes = b"active-version".to_vec(); + let active_mod_time = now - time::Duration::days(2); + let mut active_reader = PutObjReader::from_vec(active_bytes.clone()); + let active = ecstore + .put_object( + &bucket, + object, + &mut active_reader, + &ObjectOptions { + versioned: true, + mod_time: Some(active_mod_time), + ..Default::default() + }, + ) + .await + .expect("re-enabled PUT should create the active version"); + let active_version_id = active.version_id.expect("active version should have an exact identity"); + assert!(!active_version_id.is_nil()); + + let mut object_infos = ecstore + .clone() + .list_object_versions(&bucket, object, None, None, None, 10) + .await + .expect("version history should be listable") + .objects + .into_iter() + .filter(|candidate| candidate.name == object) + .collect::>(); + object_infos.sort_by_key(|candidate| !candidate.is_latest); + assert_eq!(object_infos.len(), 2); + let null_index = object_infos + .iter() + .position(|candidate| candidate.version_id == Some(Uuid::nil())) + .expect("history should expose an explicit null identity"); + assert!(!object_infos[null_index].is_latest); + + let lifecycle = current_and_noncurrent_expiration_lifecycle(); + let object_opts = object_infos + .iter() + .map(lifecycle::object_opts_from_object_info) + .collect::>(); + let events = Evaluator::new(Arc::new(lifecycle.clone())) + .eval(&object_opts) + .await + .expect("version group should evaluate"); + let current_index = object_infos + .iter() + .position(|candidate| candidate.is_latest) + .expect("version history should expose one current version"); + assert_eq!(events[null_index].action, IlmAction::DeleteVersionAction); + assert_eq!(events[current_index].action, IlmAction::NoneAction); + let secondary_event = eval_action_from_lifecycle(&lifecycle, None, &object_infos[null_index]).await; + assert_eq!(secondary_event.action, events[null_index].action); + let null_target = lifecycle_version_delete_target(&object_infos[null_index]) + .expect("the historical null generation should be an exact lifecycle target"); + assert!(null_target.expected_identity.is_some()); + + let incarnation = ecstore + .bucket_incarnation_id_from_disk(&bucket) + .await + .expect("bucket incarnation should be available"); + let failed = crate::bucket::lifecycle::object_handlers_common::delete_object_versions( + &ecstore, + &bucket, + std::slice::from_ref(&null_target), + events[null_index].clone(), + incarnation, + ) + .await; + assert_eq!(failed, 0, "the unchanged historical null generation should be deleted"); + + let failed = crate::bucket::lifecycle::object_handlers_common::delete_object_versions( + &ecstore, + &bucket, + &[ObjectToDelete { + object_name: object.to_string(), + version_id: None, + ..Default::default() + }], + events[null_index].clone(), + incarnation, + ) + .await; + assert_eq!(failed, 1, "a lifecycle batch target without a version identity must fail closed"); + + let remaining = ecstore + .clone() + .list_object_versions(&bucket, object, None, None, None, 10) + .await + .expect("remaining version should be listable") + .objects + .into_iter() + .filter(|candidate| candidate.name == object) + .collect::>(); + assert_eq!(remaining.len(), 1); + assert_eq!(remaining[0].version_id, Some(active_version_id)); + assert!(remaining[0].is_latest); + + let mut reader = ecstore + .get_object_reader(&bucket, object, None, http::HeaderMap::new(), &ObjectOptions::default()) + .await + .expect("active version should remain readable"); + let actual = reader.read_all().await.expect("active version bytes should be readable"); + assert_eq!(actual, active_bytes); + + metadata_sys::update_in( + &ecstore.ctx, + &bucket, + BUCKET_VERSIONING_CONFIG, + b"Suspended".to_vec(), + ) + .await + .expect("bucket versioning should be suspended again"); + let replacement_bytes = b"replacement-null-version".to_vec(); + let mut replacement_reader = PutObjReader::from_vec(replacement_bytes.clone()); + let replacement = ecstore + .put_object( + &bucket, + object, + &mut replacement_reader, + &ObjectOptions { + version_suspended: true, + ..Default::default() + }, + ) + .await + .expect("a later suspended PUT should reuse the null version ID"); + assert_eq!(replacement.version_id, Some(Uuid::nil())); + assert_ne!( + replacement.data_dir, object_infos[null_index].data_dir, + "the replacement must be a distinct persisted generation" + ); + + let failed = crate::bucket::lifecycle::object_handlers_common::delete_object_versions( + &ecstore, + &bucket, + &[ObjectToDelete { + object_name: object.to_string(), + version_id: Some(Uuid::nil()), + ..Default::default() + }], + events[null_index].clone(), + incarnation, + ) + .await; + assert_eq!(failed, 1, "a reusable null ID without its observed generation must fail closed"); + + let failed = crate::bucket::lifecycle::object_handlers_common::delete_object_versions( + &ecstore, + &bucket, + &[null_target], + events[null_index].clone(), + incarnation, + ) + .await; + assert_eq!(failed, 1, "a stale null-generation target must fail its write-lock CAS"); + + let remaining = ecstore + .clone() + .list_object_versions(&bucket, object, None, None, None, 10) + .await + .expect("replacement history should be listable") + .objects + .into_iter() + .filter(|candidate| candidate.name == object) + .collect::>(); + assert_eq!(remaining.len(), 2); + assert!( + remaining + .iter() + .any(|candidate| candidate.version_id == Some(active_version_id)) + ); + assert!(remaining.iter().any(|candidate| candidate.version_id == Some(Uuid::nil()))); + let mut reader = ecstore + .get_object_reader(&bucket, object, None, http::HeaderMap::new(), &ObjectOptions::default()) + .await + .expect("the replacement null version should remain current and readable"); + let actual = reader + .read_all() + .await + .expect("replacement null-version bytes should be readable"); + assert_eq!(actual, replacement_bytes); + } + #[tokio::test] async fn existing_object_lifecycle_skips_current_expiration_for_bucket_default_retention() { let lc = latest_expiration_lifecycle(); diff --git a/crates/ecstore/src/bucket/lifecycle/core.rs b/crates/ecstore/src/bucket/lifecycle/core.rs index c6c970da3..897874a25 100644 --- a/crates/ecstore/src/bucket/lifecycle/core.rs +++ b/crates/ecstore/src/bucket/lifecycle/core.rs @@ -17,6 +17,7 @@ use crate::object_api::ObjectInfo; pub use rustfs_lifecycle::{ Event, ExpirationOptions, IlmAction, Lifecycle, LifecycleCalculate, ObjectOpts, RuleValidate, TRANSITION_COMPLETE, TRANSITION_PENDING, TransitionOptions, abort_incomplete_multipart_upload_due, expected_expiry_time, + expiration_action_has_valid_target, }; pub fn object_opts_from_object_info(oi: &ObjectInfo) -> ObjectOpts { diff --git a/crates/ecstore/src/bucket/lifecycle/object_handlers_common.rs b/crates/ecstore/src/bucket/lifecycle/object_handlers_common.rs index 95f332e64..3f4d1d8ac 100644 --- a/crates/ecstore/src/bucket/lifecycle/object_handlers_common.rs +++ b/crates/ecstore/src/bucket/lifecycle/object_handlers_common.rs @@ -34,7 +34,23 @@ pub async fn delete_object_versions( to_del: &[ObjectToDelete], _lc_event: lifecycle::Event, bucket_incarnation_id: Uuid, -) { +) -> usize { + if to_del.iter().any(|target| { + target.version_id.is_none() + || (target.version_id.is_some_and(|version_id| version_id.is_nil()) && target.expected_identity.is_none()) + }) { + debug!( + event = EVENT_LIFECYCLE_CLEANUP_SKIPPED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, + bucket, + target_count = to_del.len(), + reason = "incomplete_version_identity", + "Skipped lifecycle noncurrent version cleanup" + ); + return to_del.len(); + } + let delete_config_snapshot = match ReplicationObjectBridge::delete_request_config(api, bucket).await { Ok(snapshot) => Arc::new(snapshot), Err(err) => { @@ -47,10 +63,11 @@ pub async fn delete_object_versions( reason = "delete_config_snapshot_unavailable", "Skipped lifecycle noncurrent version cleanup" ); - return; + return to_del.len(); } }; let mut remaining = to_del; + let mut failed = 0; loop { let mut to_del = remaining; if to_del.len() > MAX_DELETE_LIST { @@ -71,6 +88,7 @@ pub async fn delete_object_versions( }, ) .await; + failed += errors.iter().filter(|err| err.is_some()).count(); for (i, deleted_obj) in deleted_objs.iter_mut().enumerate() { if errors.get(i).and_then(|err| err.as_ref()).is_some() { @@ -111,4 +129,5 @@ pub async fn delete_object_versions( break; } } + failed } diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index a258c8fb2..1bc87619d 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/crates/ecstore/src/set_disk/mod.rs @@ -14560,12 +14560,66 @@ mod tests { .expect("suspended-version object should be written"); let marker = set_disks - .delete_object(bucket, object, opts.clone()) + .delete_object( + bucket, + object, + ObjectOptions { + mod_time: Some(OffsetDateTime::now_utc()), + ..opts.clone() + }, + ) .await .expect("version-suspended delete should create a null marker"); assert!(marker.delete_marker); assert_eq!(marker.version_id, Some(Uuid::nil())); + let stale_target = crate::bucket::lifecycle::bucket_lifecycle_ops::lifecycle_version_delete_target(&marker) + .expect("the observed null marker should produce an exact lifecycle target"); + let replacement_mod_time = + marker.mod_time.expect("the first null marker must have a modification time") + time::Duration::seconds(1); + + let mut replacement_reader = PutObjReader::from_vec(b"replacement suspended version body".to_vec()); + set_disks + .put_object(bucket, object, &mut replacement_reader, &opts) + .await + .expect("a suspended-version PUT should replace the first null marker"); + let replacement_marker = set_disks + .delete_object( + bucket, + object, + ObjectOptions { + mod_time: Some(replacement_mod_time), + ..opts.clone() + }, + ) + .await + .expect("a second suspended-version delete should create a replacement null marker"); + assert!(replacement_marker.delete_marker); + assert_eq!(replacement_marker.version_id, Some(Uuid::nil())); + assert_eq!(replacement_marker.mod_time, Some(replacement_mod_time)); + + let (_deleted, stale_errors) = set_disks.delete_objects(bucket, vec![stale_target], opts.clone()).await; + assert!( + matches!(stale_errors.as_slice(), [Some(StorageError::PreconditionFailed)]), + "a stale null-marker lifecycle target must not delete its replacement: {stale_errors:?}" + ); + + let (current_marker, _, current_error) = set_disks + .get_object_info_and_quorum( + bucket, + object, + &ObjectOptions { + version_id: Some(Uuid::nil().to_string()), + version_suspended: true, + no_lock: true, + ..Default::default() + }, + ) + .await; + assert!(matches!(current_error, Some(StorageError::MethodNotAllowed))); + assert!(current_marker.delete_marker); + assert_eq!(current_marker.mod_time, Some(replacement_mod_time)); + let (_deleted, errs) = set_disks .delete_objects( bucket, diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index 2b4785f00..40e45c902 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/crates/ecstore/src/set_disk/ops/object.rs @@ -7752,7 +7752,10 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { let marker_delete = dobj.version_id.is_none() || dobj.synthetic_version_id; let replication_needs_source = replicate_delete && (!marker_delete || delete_config_snapshot.active_delete_marker_rules_require_tags(&replication_object_name)); - let (goi, gerr) = if object_lock_check_required || replication_needs_source || opts.tier_delete_journal_api.is_some() + let (goi, gerr) = if object_lock_check_required + || replication_needs_source + || opts.tier_delete_journal_api.is_some() + || dobj.expected_identity.is_some() { let (goi, _write_quorum, gerr) = self.get_object_info_and_quorum(bucket, &dobj.object_name, &check_opts).await; (goi, gerr) @@ -7762,6 +7765,19 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { let source_missing = gerr .as_ref() .is_some_and(|err| is_err_object_not_found(err) || is_err_version_not_found(err)); + // A pool-local miss keeps the existing idempotent batch-delete + // semantics. Exact-version deletes fan out to every pool, and a + // retry may also arrive after the intended generation is gone. + // Only a concrete generation with a different identity is stale. + if let Some(expected) = dobj.expected_identity + && (gerr.is_none() || matches!(gerr.as_ref(), Some(StorageError::MethodNotAllowed))) + && (goi.data_dir != expected.data_dir + || goi.mod_time != expected.mod_time + || goi.delete_marker != expected.delete_marker) + { + del_errs[i] = Some(StorageError::PreconditionFailed); + continue; + } // Resolve accounting from the generation selected under this // object's write lock. A request-layer pre-stat is only an // optimization and cannot identify a concurrent overwrite. diff --git a/crates/ecstore/src/store/object.rs b/crates/ecstore/src/store/object.rs index 4ebcf73be..c3e150964 100644 --- a/crates/ecstore/src/store/object.rs +++ b/crates/ecstore/src/store/object.rs @@ -7551,6 +7551,96 @@ mod tests { assert!(error.is_none()); } + #[tokio::test] + async fn lifecycle_null_version_cas_distinguishes_absent_and_replacement_pools() { + use s3s::dto::{BucketVersioningStatus, VersioningConfiguration}; + + let ctx = Arc::new(crate::runtime::instance::InstanceContext::new()); + let (_first_dirs, first_set) = make_local_set_disks_with_ctx(4, 2, Arc::clone(&ctx)).await; + let (_second_dirs, second_set) = make_local_set_disks_with_ctx(4, 2, Arc::clone(&ctx)).await; + let store = new_prepared_reader_test_store_with_ctx(&[Arc::clone(&first_set), Arc::clone(&second_set)], ctx).await; + let bucket = RUSTFS_META_BUCKET; + let suspended_snapshot = Arc::new(DeleteReplicationConfigSnapshot::from_configs_for_test( + VersioningConfiguration { + status: Some(BucketVersioningStatus::from_static(BucketVersioningStatus::SUSPENDED)), + ..Default::default() + }, + None, + )); + let write_opts = ObjectOptions { + no_lock: true, + version_suspended: true, + ..Default::default() + }; + let delete_opts = ObjectOptions { + version_suspended: true, + delete_replication_config_snapshot: Some(Arc::clone(&suspended_snapshot)), + ..Default::default() + }; + + let success_object = "lifecycle-null-only-in-first-pool"; + let mut success_reader = PutObjReader::from_vec(b"single-pool null generation".to_vec()); + let success_source = first_set + .put_object(bucket, success_object, &mut success_reader, &write_opts) + .await + .expect("the historical null generation should be written to one pool"); + let success_target = crate::bucket::lifecycle::bucket_lifecycle_ops::lifecycle_version_delete_target(&success_source) + .expect("the null generation should produce an exact lifecycle target"); + + let (deleted, errors) = store + .handle_delete_objects(bucket, vec![success_target.clone()], delete_opts.clone()) + .await; + assert!( + errors.iter().all(Option::is_none), + "an absent peer pool must not override success: {errors:?}" + ); + assert!(deleted[0].found, "the pool containing the exact null generation must report success"); + + let (deleted, errors) = store + .handle_delete_objects(bucket, vec![success_target], delete_opts.clone()) + .await; + assert!( + errors.iter().all(Option::is_none), + "an exact lifecycle delete retry must be idempotent: {errors:?}" + ); + assert!(!deleted[0].found, "the retried generation should already be absent from every pool"); + + let mismatch_object = "lifecycle-null-replaced-in-second-pool"; + let mut original_reader = PutObjReader::from_vec(b"original null generation".to_vec()); + let original = first_set + .put_object(bucket, mismatch_object, &mut original_reader, &write_opts) + .await + .expect("the original null generation should be written to the first pool"); + let stale_target = crate::bucket::lifecycle::bucket_lifecycle_ops::lifecycle_version_delete_target(&original) + .expect("the original null generation should produce an exact lifecycle target"); + let mut replacement_reader = PutObjReader::from_vec(b"replacement null generation".to_vec()); + let replacement = second_set + .put_object(bucket, mismatch_object, &mut replacement_reader, &write_opts) + .await + .expect("the replacement null generation should be written to the second pool"); + assert_ne!(original.data_dir, replacement.data_dir); + + let (_deleted, errors) = store.handle_delete_objects(bucket, vec![stale_target], delete_opts).await; + assert!( + matches!(errors.as_slice(), [Some(StorageError::PreconditionFailed)]), + "a different null generation in any pool must fail the aggregate CAS: {errors:?}" + ); + let retained = second_set + .get_object_info( + bucket, + mismatch_object, + &ObjectOptions { + version_id: Some(Uuid::nil().to_string()), + version_suspended: true, + no_lock: true, + ..Default::default() + }, + ) + .await + .expect("the replacement null generation must remain after a stale CAS"); + assert_eq!(retained.data_dir, replacement.data_dir); + } + #[test] fn data_movement_pool_lookup_opts_keeps_no_lock_for_tiered_moves() { let lookup_opts = data_movement_pool_lookup_opts( diff --git a/crates/lifecycle/src/core.rs b/crates/lifecycle/src/core.rs index 544cdd5ba..fc959b92a 100644 --- a/crates/lifecycle/src/core.rs +++ b/crates/lifecycle/src/core.rs @@ -481,7 +481,7 @@ impl Lifecycle for BucketLifecycleConfiguration { return Event::default(); }; - if obj.delete_marker || !(obj.is_latest || obj.version_id.is_none_or(|v| v.is_nil())) { + if obj.delete_marker || !obj.is_latest { return Event::default(); } @@ -676,10 +676,11 @@ impl Lifecycle for BucketLifecycleConfiguration { obj.is_latest, obj.delete_marker, obj.version_id, - (obj.is_latest || obj.version_id.is_none_or(|v| v.is_nil())) && !obj.delete_marker + obj.is_latest && !obj.delete_marker ); - // Allow expiration for latest objects OR non-versioned objects (empty version_id) - if (obj.is_latest || obj.version_id.is_none_or(|v| v.is_nil())) && !obj.delete_marker { + // Current-version expiration is selected by the authoritative + // latest flag. An explicit null version can also be historical. + if obj.is_latest && !obj.delete_marker { debug!("eval_inner: entering expiration check"); if let Some(ref expiration) = rule.expiration { if let Some(ref date) = expiration.date { @@ -1043,6 +1044,26 @@ impl ObjectOpts { } } +/// Returns whether an expiry action has enough identity to target the object +/// it was evaluated against. A nil UUID is an explicit S3 null-version +/// identity; only an absent version ID is ambiguous for an exact-version +/// action. +pub fn expiration_action_has_valid_target( + action: IlmAction, + version_id: Option, + is_latest: bool, + delete_marker: bool, +) -> bool { + match action { + IlmAction::DeleteAction | IlmAction::DeleteRestoredAction => is_latest && !delete_marker, + IlmAction::DeleteVersionAction => version_id.is_some() && (!is_latest || delete_marker), + IlmAction::DeleteRestoredVersionAction => version_id.is_some() && !is_latest && !delete_marker, + IlmAction::DeleteAllVersionsAction => is_latest && !delete_marker, + IlmAction::DelMarkerDeleteAllVersionsAction => is_latest && delete_marker, + _ => true, + } +} + /// Total-order rank for lifecycle actions used to break `due` ties. /// /// Delete-type actions rank before every other action so that, when two events @@ -4008,6 +4029,76 @@ mod tests { assert_eq!(event.action, IlmAction::DeleteAction); } + #[tokio::test] + async fn current_expiration_requires_latest_even_for_null_version_identity() { + let base_time = datetime!(2025-01-01 00:00:00 UTC); + let lc = BucketLifecycleConfiguration { + expiry_updated_at: None, + rules: vec![enabled_rule( + Some(LifecycleExpiration { + days: Some(1), + ..Default::default() + }), + None, + Some("expire-current"), + )], + }; + let now = base_time + Duration::days(2); + + let unversioned_current = ObjectOpts { + name: "object".to_string(), + mod_time: Some(base_time), + is_latest: true, + version_id: None, + ..Default::default() + }; + assert_eq!( + lc.eval_inner(&unversioned_current, now, 0).await.action, + IlmAction::DeleteAction, + "a truly unversioned current object must remain eligible" + ); + assert_eq!(lc.predict_expiration(&unversioned_current).await.action, IlmAction::DeleteAction); + assert!(expiration_action_has_valid_target( + IlmAction::DeleteAction, + unversioned_current.version_id, + unversioned_current.is_latest, + unversioned_current.delete_marker, + )); + assert!(!expiration_action_has_valid_target( + IlmAction::DeleteVersionAction, + unversioned_current.version_id, + unversioned_current.is_latest, + unversioned_current.delete_marker, + )); + + let historical_null = ObjectOpts { + name: "object".to_string(), + mod_time: Some(base_time), + version_id: Some(Uuid::nil()), + is_latest: false, + versioned: true, + ..Default::default() + }; + assert_eq!( + lc.eval_inner(&historical_null, now, 0).await.action, + IlmAction::NoneAction, + "an explicit null identity does not make a historical version current" + ); + assert_eq!(lc.predict_expiration(&historical_null).await.action, IlmAction::NoneAction); + assert!(!expiration_action_has_valid_target( + IlmAction::DeleteAction, + historical_null.version_id, + historical_null.is_latest, + historical_null.delete_marker, + )); + assert!(expiration_action_has_valid_target( + IlmAction::DeleteVersionAction, + historical_null.version_id, + historical_null.is_latest, + historical_null.delete_marker, + )); + } + /// backlog#1148 ilm-8: once a restored copy's expiry passes, the evaluator /// must emit `DeleteRestoredAction` for the current version (and the /// `...Version` variant for a noncurrent one) so the scanner removes only diff --git a/crates/lifecycle/src/evaluator.rs b/crates/lifecycle/src/evaluator.rs index a6646b33d..80bd61de6 100644 --- a/crates/lifecycle/src/evaluator.rs +++ b/crates/lifecycle/src/evaluator.rs @@ -22,7 +22,7 @@ use rustfs_replication::ReplicationStatusType; use rustfs_scanner_metrics::metrics::IlmAction; use crate::object_lock; -use crate::{Event, Lifecycle, ObjectOpts}; +use crate::{Event, Lifecycle, ObjectOpts, expiration_action_has_valid_target}; const LOG_COMPONENT_ECSTORE: &str = "ecstore"; const LOG_SUBSYSTEM_LIFECYCLE: &str = "lifecycle"; @@ -80,6 +80,9 @@ impl Evaluator { 'top_loop: { for (i, obj) in objs.iter().enumerate() { let mut event = self.policy.eval_inner(obj, now, newer_noncurrent_versions).await; + if !expiration_action_has_valid_target(event.action, obj.version_id, obj.is_latest, obj.delete_marker) { + event = Event::default(); + } if lifecycle_action_waits_for_replication(event.action) && self.is_pending_replication(obj) { event = Event::default(); } @@ -116,16 +119,10 @@ impl Evaluator { IlmAction::DeleteAction | IlmAction::DeleteRestoredAction | IlmAction::DeleteVersionAction - | IlmAction::DeleteRestoredVersionAction => { - // Defensive code, should never happen - if matches!(event.action, IlmAction::DeleteVersionAction | IlmAction::DeleteRestoredVersionAction) - && obj.version_id.is_none_or(|v| v.is_nil()) - { - event.action = IlmAction::NoneAction; - } - if self.is_object_locked(obj) { - event = Event::default(); - } + | IlmAction::DeleteRestoredVersionAction + if self.is_object_locked(obj) => + { + event = Event::default(); } _ => {} } @@ -336,6 +333,30 @@ mod tests { }) } + fn current_and_noncurrent_expiration_lifecycle() -> Arc { + Arc::new(BucketLifecycleConfiguration { + expiry_updated_at: None, + rules: vec![LifecycleRule { + status: ExpirationStatus::from_static(ExpirationStatus::ENABLED), + expiration: Some(LifecycleExpiration { + days: Some(30), + ..Default::default() + }), + abort_incomplete_multipart_upload: None, + del_marker_expiration: None, + filter: None, + id: Some("expire-current-and-noncurrent".to_string()), + noncurrent_version_expiration: Some(NoncurrentVersionExpiration { + noncurrent_days: Some(1), + newer_noncurrent_versions: None, + }), + noncurrent_version_transitions: None, + prefix: None, + transitions: None, + }], + }) + } + fn object_opts(replication_status: ReplicationStatusType, version_purge_status: VersionPurgeStatusType) -> ObjectOpts { ObjectOpts { name: "logs/object".to_string(), @@ -541,6 +562,68 @@ mod tests { assert_eq!(events[0].action, IlmAction::DeleteVersionAction); } + #[tokio::test] + async fn evaluator_treats_explicit_null_as_an_exact_noncurrent_identity() { + let lifecycle = current_and_noncurrent_expiration_lifecycle(); + let now = OffsetDateTime::now_utc(); + let version_group = + |version_id: Option, replication_status: ReplicationStatusType, user_defined: HashMap| { + vec![ + ObjectOpts { + name: "logs/object".to_string(), + mod_time: Some(now), + version_id: Some(Uuid::new_v4()), + is_latest: true, + num_versions: 2, + replication_status: ReplicationStatusType::Completed, + ..Default::default() + }, + ObjectOpts { + name: "logs/object".to_string(), + mod_time: Some(now - time::Duration::days(40)), + successor_mod_time: Some(now - time::Duration::days(3)), + version_id, + is_latest: false, + num_versions: 2, + versioned: true, + replication_status, + user_defined, + ..Default::default() + }, + ] + }; + + let events = Evaluator::new(lifecycle.clone()) + .eval(&version_group(Some(Uuid::nil()), ReplicationStatusType::Completed, HashMap::new())) + .await + .expect("explicit null-version lifecycle evaluation should succeed"); + assert_eq!(events[0].action, IlmAction::NoneAction); + assert_eq!(events[1].action, IlmAction::DeleteVersionAction); + + let events = Evaluator::new(lifecycle.clone()) + .eval(&version_group(None, ReplicationStatusType::Completed, HashMap::new())) + .await + .expect("missing historical identity should fail closed without aborting evaluation"); + assert_eq!(events[1].action, IlmAction::NoneAction); + + let events = Evaluator::new(lifecycle.clone()) + .eval(&version_group(Some(Uuid::nil()), ReplicationStatusType::Pending, HashMap::new())) + .await + .expect("pending null-version replication should fail closed without aborting evaluation"); + assert_eq!(events[1].action, IlmAction::NoneAction); + + let events = Evaluator::new(lifecycle) + .with_lock_retention(Some(lock_enabled_without_default_retention())) + .eval(&version_group( + Some(Uuid::nil()), + ReplicationStatusType::Completed, + HashMap::from([(X_AMZ_OBJECT_LOCK_LEGAL_HOLD.as_str().to_string(), "ON".to_string())]), + )) + .await + .expect("locked null-version lifecycle evaluation should fail closed without aborting evaluation"); + assert_eq!(events[1].action, IlmAction::NoneAction); + } + #[tokio::test] async fn evaluator_skips_transition_while_replication_pending() { let evaluator = Evaluator::new(latest_transition_lifecycle()); diff --git a/crates/scanner/src/lib.rs b/crates/scanner/src/lib.rs index 393392a01..6d0b84edb 100644 --- a/crates/scanner/src/lib.rs +++ b/crates/scanner/src/lib.rs @@ -43,9 +43,9 @@ use storage_api::owner::{ 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_invalidate_admin_data_usage_snapshot_cache, ecstore_invalidate_data_usage_snapshot_cache, ecstore_is_erasure, - ecstore_is_reserved_or_invalid_bucket, ecstore_list_path_raw, ecstore_object_opts_from_object_info, - 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_is_reserved_or_invalid_bucket, ecstore_lifecycle_version_delete_target, ecstore_list_path_raw, + ecstore_object_opts_from_object_info, 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)] diff --git a/crates/scanner/src/scanner_folder.rs b/crates/scanner/src/scanner_folder.rs index a7abfe975..0e54b69d0 100644 --- a/crates/scanner/src/scanner_folder.rs +++ b/crates/scanner/src/scanner_folder.rs @@ -65,9 +65,9 @@ use crate::{ Disk, DiskError, DiskInfoOptions, Evaluator, Event, LcEventSrc, ListPathRawOptions, ObjectOpts, ReplicationConfig, ReplicationHealObject, ReplicationQueueAdmission, ReplicationStatusType, STORAGE_FORMAT_FILE, ScannerDiskExt as _, ScannerLifecycleConfigExt as _, ScannerVersioningConfigExt as _, StorageError, TierRegistrySnapshot, 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, runtime_tier_registry_for_cycle, scanner_is_erasure, - scanner_replication_config_for_lifecycle_eval, + apply_transition_rule, ecstore_lifecycle_version_delete_target, enqueue_runtime_newer_noncurrent, + is_reserved_or_invalid_bucket, list_path_raw, path2_bucket_object, path2_bucket_object_with_base_path, + queue_replication_heal, runtime_tier_registry_for_cycle, scanner_is_erasure, scanner_replication_config_for_lifecycle_eval, }; use crate::{ScannerObjectInfo as ObjectInfo, ScannerObjectToDelete as ObjectToDelete}; diff --git a/crates/scanner/src/scanner_folder/item_actions.rs b/crates/scanner/src/scanner_folder/item_actions.rs index 524482b07..5120f8181 100644 --- a/crates/scanner/src/scanner_folder/item_actions.rs +++ b/crates/scanner/src/scanner_folder/item_actions.rs @@ -819,14 +819,15 @@ impl ScannerItem { } } IlmAction::DeleteVersionAction => { - if let Some(opt) = object_opts.get(i) { - to_delete_objs.push(ObjectToDelete { - object_name: opt.name.clone(), - version_id: opt.version_id, - ..Default::default() - }); + if let Some(target) = ecstore_lifecycle_version_delete_target(oi) { + to_delete_objs.push(target); noncurrent_events.push(event.clone()); noncurrent_unknown.push(oi); + } else { + if let SizeResolution::Unknown { physical, .. } = &resolved_sizes[i] { + self.heal_actions(oi, *physical, size_summary).await; + } + size_summary.actions_accounting_unknown(oi); } } IlmAction::TransitionAction | IlmAction::TransitionVersionAction => { @@ -933,12 +934,8 @@ impl ScannerItem { } } IlmAction::DeleteVersionAction => { - if let Some(opt) = object_opts.get(i) { - to_delete_objs.push(ObjectToDelete { - object_name: opt.name.clone(), - version_id: opt.version_id, - ..Default::default() - }); + if let Some(target) = ecstore_lifecycle_version_delete_target(oi) { + to_delete_objs.push(target); if let Some(actual_size) = known_size { noncurrent_accounting.push(PendingScannerAccounting { object: oi, @@ -947,8 +944,8 @@ impl ScannerItem { }); } account_now = false; + noncurrent_events.push(event.clone()); } - noncurrent_events.push(event.clone()); } IlmAction::TransitionAction | IlmAction::TransitionVersionAction => { debug!( diff --git a/crates/scanner/src/storage_api.rs b/crates/scanner/src/storage_api.rs index 9efd990be..3c410deed 100644 --- a/crates/scanner/src/storage_api.rs +++ b/crates/scanner/src/storage_api.rs @@ -32,6 +32,7 @@ pub(crate) use rustfs_ecstore::api::bucket::bucket_target_sys::BucketTargetSys a 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::{ apply_expiry_rule as ecstore_apply_expiry_rule, apply_transition_rule as ecstore_apply_transition_rule, + lifecycle_version_delete_target as ecstore_lifecycle_version_delete_target, }; pub(crate) use rustfs_ecstore::api::bucket::lifecycle::lifecycle::object_opts_from_object_info as ecstore_object_opts_from_object_info; #[cfg(test)] @@ -142,10 +143,10 @@ pub(crate) mod owner { ecstore_get_global_tier_config_mgr, ecstore_get_lifecycle_config, ecstore_get_object_lock_config, ecstore_get_replication_config, ecstore_invalidate_admin_data_usage_snapshot_cache, ecstore_invalidate_data_usage_snapshot_cache, ecstore_is_erasure, ecstore_is_reserved_or_invalid_bucket, - ecstore_list_path_raw, ecstore_object_opts_from_object_info, 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_send_event, - scanner_replication_config_for_lifecycle_eval, + ecstore_lifecycle_version_delete_target, ecstore_list_path_raw, ecstore_object_opts_from_object_info, + 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_send_event, scanner_replication_config_for_lifecycle_eval, }; #[cfg(test)] diff --git a/crates/scanner/tests/lifecycle_integration_test.rs b/crates/scanner/tests/lifecycle_integration_test.rs index 26ef8496c..2911d9c49 100644 --- a/crates/scanner/tests/lifecycle_integration_test.rs +++ b/crates/scanner/tests/lifecycle_integration_test.rs @@ -40,7 +40,7 @@ use uuid::Uuid; mod storage_api; use storage_api::lifecycle::{ - BUCKET_LIFECYCLE_CONFIG, BucketOperations, BucketOptions, BucketVersioningSys, CompletePart, + BUCKET_LIFECYCLE_CONFIG, BUCKET_VERSIONING_CONFIG, BucketOperations, BucketOptions, BucketVersioningSys, CompletePart, DeleteAfterObjectLockSnapshotBarrier, DiskOption, ECStore, EcstoreError, Endpoint, EndpointServerPools, Endpoints, ExpiryState, IlmAction, LcEvent, LcEventSrc, ListOperations as _, MakeBucketOptions, MockWarmBackend, MultipartOperations as _, ObjectIO as _, ObjectOperations as _, PoolEndpoints, STORAGE_FORMAT_FILE, TRANSITION_PENDING, @@ -2112,6 +2112,116 @@ mod serial_tests { ); } + #[tokio::test(flavor = "multi_thread", worker_threads = 1)] + #[serial] + #[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#2198"] + async fn test_scanner_expires_historical_null_without_deleting_active_version() { + let (disk_paths, ecstore) = setup_isolated_test_env(false).await; + let bucket_name = format!("test-historical-null-expire-{}", &Uuid::new_v4().simple().to_string()[..8]); + let object_name = "test/object.txt"; + + create_test_bucket(&ecstore, bucket_name.as_str()).await; + update_bucket_metadata( + bucket_name.as_str(), + BUCKET_VERSIONING_CONFIG, + b"Enabled".to_vec(), + ) + .await + .expect("bucket versioning should first be enabled"); + update_bucket_metadata( + bucket_name.as_str(), + BUCKET_VERSIONING_CONFIG, + b"Suspended".to_vec(), + ) + .await + .expect("bucket versioning should be suspended"); + + let mut null_reader = PutObjReader::from_vec(b"historical null body".to_vec()); + let historical_null = ecstore + .put_object( + bucket_name.as_str(), + object_name, + &mut null_reader, + &ObjectOptions { + version_suspended: true, + ..Default::default() + }, + ) + .await + .expect("suspended PUT should create a null version"); + assert_eq!(historical_null.version_id, Some(Uuid::nil())); + + update_bucket_metadata( + bucket_name.as_str(), + BUCKET_VERSIONING_CONFIG, + b"Enabled".to_vec(), + ) + .await + .expect("bucket versioning should be re-enabled"); + let active_body = b"active uuid body".to_vec(); + let mut active_reader = PutObjReader::from_vec(active_body.clone()); + let active = ecstore + .put_object( + bucket_name.as_str(), + object_name, + &mut active_reader, + &ObjectOptions { + versioned: true, + ..Default::default() + }, + ) + .await + .expect("re-enabled PUT should create an active UUID version"); + let active_version_id = active.version_id.expect("the active version should have an identity"); + assert!(!active_version_id.is_nil()); + + let lifecycle_xml = r#" + + + expire-historical-null + Enabled + + test/ + + + 0 + + +"#; + update_bucket_metadata(bucket_name.as_str(), BUCKET_LIFECYCLE_CONFIG, lifecycle_xml.as_bytes().to_vec()) + .await + .expect("noncurrent expiration should be configured"); + init_background_expiry(ecstore.clone()).await; + + assert_eq!(object_version_count(&ecstore, bucket_name.as_str(), object_name).await, 2); + scan_object_with_lifecycle(&disk_paths[0], bucket_name.as_str(), object_name).await; + assert!( + wait_for_version_count(&ecstore, bucket_name.as_str(), object_name, 1, Duration::from_secs(3)).await, + "scanner should delete only the historical null generation" + ); + + let remaining = ecstore + .get_object_info(bucket_name.as_str(), object_name, &ObjectOptions::default()) + .await + .expect("the active UUID version must remain visible"); + assert_eq!(remaining.version_id, Some(active_version_id)); + assert_eq!(read_object_fully(&ecstore, bucket_name.as_str(), object_name).await, active_body); + + let null_error = ecstore + .get_object_info( + bucket_name.as_str(), + object_name, + &ObjectOptions { + version_id: Some(Uuid::nil().to_string()), + versioned: true, + ..Default::default() + }, + ) + .await + .expect_err("the historical null generation should be gone"); + assert!(is_err_object_not_found(&null_error) || is_err_version_not_found(&null_error)); + } + #[tokio::test(flavor = "multi_thread", worker_threads = 1)] #[serial] #[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#1148 (ilm-1)"] diff --git a/crates/scanner/tests/storage_api/mod.rs b/crates/scanner/tests/storage_api/mod.rs index c71522dc8..1bf2ffaf3 100644 --- a/crates/scanner/tests/storage_api/mod.rs +++ b/crates/scanner/tests/storage_api/mod.rs @@ -22,7 +22,7 @@ pub(crate) use rustfs_ecstore::api::bucket::lifecycle::{ }, lifecycle::{Event as LcEvent, IlmAction, TRANSITION_PENDING, TransitionOptions}, }; -pub(crate) use rustfs_ecstore::api::bucket::metadata::BUCKET_LIFECYCLE_CONFIG; +pub(crate) use rustfs_ecstore::api::bucket::metadata::{BUCKET_LIFECYCLE_CONFIG, BUCKET_VERSIONING_CONFIG}; pub(crate) use rustfs_ecstore::api::bucket::metadata_sys::{ get as get_bucket_metadata, init_bucket_metadata_sys, update as update_bucket_metadata, }; @@ -50,13 +50,13 @@ pub(crate) mod lifecycle { }; pub(crate) use super::{ - BUCKET_LIFECYCLE_CONFIG, BucketVersioningSys, DeleteAfterObjectLockSnapshotBarrier, DiskOption, ECStore, EcstoreError, - Endpoint, EndpointServerPools, Endpoints, ExpiryState, IlmAction, LcEvent, LcEventSrc, MockWarmBackend, PoolEndpoints, - STORAGE_FORMAT_FILE, TRANSITION_PENDING, TransitionCleanupStoreBarrier, TransitionOptions, - assert_transition_meta_consistent, enqueue_transition_for_existing_objects, expire_transitioned_object, - free_version_count, get_bucket_metadata, get_global_tier_config_mgr, init_background_expiry, init_bucket_metadata_sys, - init_local_disks, is_err_object_not_found, is_err_version_not_found, new_disk, path2_bucket_object_with_base_path, - recover_transition_transaction_records, recover_transition_transaction_records_at, register_mock_tier_util, - update_bucket_metadata, wait_for_free_version_absence, + BUCKET_LIFECYCLE_CONFIG, BUCKET_VERSIONING_CONFIG, BucketVersioningSys, DeleteAfterObjectLockSnapshotBarrier, DiskOption, + ECStore, EcstoreError, Endpoint, EndpointServerPools, Endpoints, ExpiryState, IlmAction, LcEvent, LcEventSrc, + MockWarmBackend, PoolEndpoints, STORAGE_FORMAT_FILE, TRANSITION_PENDING, TransitionCleanupStoreBarrier, + TransitionOptions, assert_transition_meta_consistent, enqueue_transition_for_existing_objects, + expire_transitioned_object, free_version_count, get_bucket_metadata, get_global_tier_config_mgr, init_background_expiry, + init_bucket_metadata_sys, init_local_disks, is_err_object_not_found, is_err_version_not_found, new_disk, + path2_bucket_object_with_base_path, recover_transition_transaction_records, recover_transition_transaction_records_at, + register_mock_tier_util, update_bucket_metadata, wait_for_free_version_absence, }; } diff --git a/crates/storage-api/src/lib.rs b/crates/storage-api/src/lib.rs index f6dd9f8d3..b16a47152 100644 --- a/crates/storage-api/src/lib.rs +++ b/crates/storage-api/src/lib.rs @@ -86,6 +86,7 @@ pub use error::{StorageErrorCode, StorageResult}; pub use multipart::{CompletePart, ListMultipartsInfo, ListPartsInfo, MultipartInfo, MultipartUploadResult, PartInfo}; pub use object::DeleteAccounting; pub use object::ObjectLockDeleteOptions; +pub use object::ObjectToDeleteIdentity; pub use object::{DeletedObject, ObjectToDelete}; pub use object::{ExpirationOptions, TransitionedObject}; pub use object::{HTTPPreconditions, HTTPRangeError, HTTPRangeSpec, ObjectLockRetentionOptions}; diff --git a/crates/storage-api/src/object.rs b/crates/storage-api/src/object.rs index 8452e62cd..b2a090941 100644 --- a/crates/storage-api/src/object.rs +++ b/crates/storage-api/src/object.rs @@ -187,10 +187,25 @@ pub enum WalkVersionsSortOrder { Descending, } +/// In-memory compare-and-set identity for a queued exact-version delete. +/// +/// `version_id` identifies the S3 version while these fields identify the +/// concrete generation observed before the delete was queued. This matters for +/// the reusable null version ID in versioning-suspended buckets. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct ObjectToDeleteIdentity { + pub data_dir: Option, + pub mod_time: Option, + pub delete_marker: bool, +} + #[derive(Debug, Default, Clone)] pub struct ObjectToDelete { pub object_name: String, pub version_id: Option, + /// RustFS-only precondition checked under the object write lock. + #[doc(hidden)] + pub expected_identity: Option, pub synthetic_version_id: bool, pub delete_marker_replication_status: Option, pub version_purge_status: Option, @@ -199,6 +214,20 @@ pub struct ObjectToDelete { } impl ObjectToDelete { + pub fn with_expected_identity( + mut self, + data_dir: Option, + mod_time: Option, + delete_marker: bool, + ) -> Self { + self.expected_identity = Some(ObjectToDeleteIdentity { + data_dir, + mod_time, + delete_marker, + }); + self + } + pub fn replication_state(&self) -> ReplicationState { ReplicationState { replication_status_internal: self.delete_marker_replication_status.clone(),