diff --git a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_audit.rs b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_audit.rs index e7dde4652..55e5bf63b 100644 --- a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_audit.rs +++ b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_audit.rs @@ -16,6 +16,7 @@ use super::runtime_boundary as runtime_sources; use crate::bucket::lifecycle::lifecycle; use crate::object_api::ObjectInfo; use crate::services::event_notification::{EventArgs, send_event}; +use crate::storage_api_contracts::object::{DeletedObject, ObjectToDelete}; use rustfs_s3_types::EventName; use rustfs_scanner_metrics::metrics::IlmAction; @@ -76,6 +77,60 @@ pub(crate) fn emit_non_transitioned_expiration_event(action: IlmAction, source: emit_lifecycle_event(event_name, deleted, LIFECYCLE_EXPIRY_USER_AGENT); } +/// Emit the lifecycle expiration event for one version removed by the batch +/// `NewerNoncurrentVersions` expiry path. +/// +/// That path never sent events, so a successful noncurrent-version expiry was +/// invisible to notification subscribers even though the equivalent +/// current-version path emits one (backlog#2202). +pub(crate) fn emit_noncurrent_expiration_event(bucket: &str, target: &ObjectToDelete, deleted: &DeletedObject, failed: bool) { + if let Some((event_name, object)) = noncurrent_expiration_event(bucket, target, deleted, failed) { + emit_lifecycle_event(event_name, object, LIFECYCLE_EXPIRY_USER_AGENT); + } +} + +/// Decide which event a single batch entry earned, if any. +/// +/// Only an entry that mutated something may be announced. "No error" is not +/// enough, and neither is `found`: the disk layer skips an absent version and +/// reports success (`delete_versions_internal` in `disk/local.rs` continues +/// past `FileVersionNotFound`), so a batch entry for a version that was +/// already gone comes back indistinguishable from a committed delete. The +/// delete plan's own source lookup is the signal that survives that, and the +/// lifecycle batch path always performs it because every target carries an +/// exact version identity. +fn noncurrent_expiration_event( + bucket: &str, + target: &ObjectToDelete, + deleted: &DeletedObject, + failed: bool, +) -> Option<(EventName, ObjectInfo)> { + if failed || !deleted.found || deleted.source_missing { + return None; + } + // A version removed by explicit version id is a plain versioned delete + // even when that version is itself a delete marker; only a request that + // carried no version id can publish a new delete marker. This is the rule + // the S3 DeleteObjects path applies (issue #6745). `delete_object_versions` + // now refuses targets without an exact version identity, so the + // marker-creation shape is unreachable from that caller; the mapping stays + // here so a future caller cannot silently announce the wrong mutation. + let created_delete_marker = deleted.delete_marker && target.version_id.is_none(); + let (event_name, version_id) = if created_delete_marker { + (EventName::LifecycleExpirationDeleteMarkerCreated, deleted.delete_marker_version_id) + } else { + (EventName::LifecycleExpirationDelete, deleted.version_id.or(target.version_id)) + }; + let object = ObjectInfo { + bucket: bucket.to_string(), + name: target.object_name.clone(), + version_id, + delete_marker: deleted.delete_marker, + ..Default::default() + }; + Some((event_name, object)) +} + fn emit_lifecycle_event(event_name: EventName, object: ObjectInfo, user_agent: &str) { send_event(EventArgs { event_name: event_name.to_string(), @@ -113,6 +168,7 @@ fn non_transitioned_expiration_event_name( #[cfg(test)] mod tests { use super::*; + use uuid::Uuid; #[test] fn transitioned_expiration_event_marks_delete_marker_creation() { @@ -129,4 +185,87 @@ mod tests { EventName::LifecycleExpirationDelete ); } + + fn deleted_version(version_id: Uuid) -> DeletedObject { + DeletedObject { + object_name: "object".to_string(), + version_id: Some(version_id), + found: true, + ..Default::default() + } + } + + fn target_version(version_id: Option) -> ObjectToDelete { + ObjectToDelete { + object_name: "object".to_string(), + version_id, + ..Default::default() + } + } + + #[test] + fn noncurrent_expiration_emits_versioned_delete_with_exact_identity() { + let version_id = Uuid::new_v4(); + let (event_name, object) = + noncurrent_expiration_event("bucket", &target_version(Some(version_id)), &deleted_version(version_id), false) + .expect("a committed delete must emit"); + assert_eq!(event_name, EventName::LifecycleExpirationDelete); + assert_eq!(object.bucket, "bucket"); + assert_eq!(object.name, "object"); + assert_eq!(object.version_id, Some(version_id)); + } + + /// Removing a noncurrent version that happens to be a delete marker is a + /// plain versioned delete, not a delete-marker creation. + #[test] + fn noncurrent_expiration_of_a_delete_marker_version_is_a_plain_delete() { + let version_id = Uuid::new_v4(); + let deleted = DeletedObject { + delete_marker: true, + ..deleted_version(version_id) + }; + let (event_name, object) = noncurrent_expiration_event("bucket", &target_version(Some(version_id)), &deleted, false) + .expect("a committed delete must emit"); + assert_eq!(event_name, EventName::LifecycleExpirationDelete); + assert_eq!(object.version_id, Some(version_id)); + } + + #[test] + fn noncurrent_expiration_reports_a_created_delete_marker() { + let marker_version_id = Uuid::new_v4(); + let deleted = DeletedObject { + object_name: "object".to_string(), + delete_marker: true, + delete_marker_version_id: Some(marker_version_id), + found: true, + ..Default::default() + }; + let (event_name, object) = + noncurrent_expiration_event("bucket", &target_version(None), &deleted, false).expect("a committed delete must emit"); + assert_eq!(event_name, EventName::LifecycleExpirationDeleteMarkerCreated); + assert_eq!(object.version_id, Some(marker_version_id)); + } + + /// A batch mixes successes with failures and versions that were already + /// gone; only a real mutation may produce an event. A version that was + /// already gone comes back with no error and `found` set, so + /// `source_missing` is the signal that keeps it silent. + #[test] + fn noncurrent_expiration_skips_failed_and_missing_versions() { + let version_id = Uuid::new_v4(); + let target = target_version(Some(version_id)); + assert!(noncurrent_expiration_event("bucket", &target, &deleted_version(version_id), true).is_none()); + + let absent = DeletedObject { + source_missing: true, + ..deleted_version(version_id) + }; + assert!(noncurrent_expiration_event("bucket", &target, &absent, false).is_none()); + + let not_found = DeletedObject { + found: false, + ..deleted_version(version_id) + }; + assert!(noncurrent_expiration_event("bucket", &target, ¬_found, false).is_none()); + } } diff --git a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs index d9e352e26..7a8b65f75 100644 --- a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs +++ b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs @@ -11336,6 +11336,120 @@ mod tests { } } + /// backlog#2202: the batch `NewerNoncurrentVersions` path used to delete + /// noncurrent versions without telling notification subscribers anything, + /// while the current-version path emitted a lifecycle expiration event. + /// Only versions this batch actually removed may produce an event. + #[tokio::test] + #[serial] + async fn lifecycle_noncurrent_batch_expiry_emits_events_only_for_committed_deletes() { + use crate::services::event_notification::test_recorder; + use rustfs_s3_types::EventName; + + let (_disk_paths, ecstore) = setup_test_env().await; + let bucket = format!("lifecycle-noncurrent-events-{}", 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 be enabled"); + + let now = OffsetDateTime::now_utc(); + let mut noncurrent_reader = PutObjReader::from_vec(b"noncurrent".to_vec()); + let noncurrent = ecstore + .put_object( + &bucket, + object, + &mut noncurrent_reader, + &ObjectOptions { + versioned: true, + mod_time: Some(now - time::Duration::days(40)), + ..Default::default() + }, + ) + .await + .expect("the noncurrent version should be created"); + let noncurrent_version_id = noncurrent.version_id.expect("a versioned PUT has an exact identity"); + let mut current_reader = PutObjReader::from_vec(b"current".to_vec()); + let current = ecstore + .put_object( + &bucket, + object, + &mut current_reader, + &ObjectOptions { + versioned: true, + mod_time: Some(now - time::Duration::days(2)), + ..Default::default() + }, + ) + .await + .expect("the current version should be created"); + let current_version_id = current.version_id.expect("a versioned PUT has an exact identity"); + + let incarnation = ecstore + .bucket_incarnation_id_from_disk(&bucket) + .await + .expect("bucket incarnation should be available"); + + test_recorder::install(); + + // One version that exists and one that never did: `delete_objects` + // suppresses the not-found error, so only the committed delete may be + // announced. + let missing_version_id = Uuid::new_v4(); + let targets = vec![ + ObjectToDelete { + object_name: object.to_string(), + version_id: Some(noncurrent_version_id), + ..Default::default() + }, + ObjectToDelete { + object_name: object.to_string(), + version_id: Some(missing_version_id), + ..Default::default() + }, + ]; + let failed = crate::bucket::lifecycle::object_handlers_common::delete_object_versions( + &ecstore, + &bucket, + &targets, + lifecycle::Event::default(), + incarnation, + ) + .await; + assert_eq!(failed, 0, "a missing version is not a batch failure"); + + let events = test_recorder::recorded_for_bucket(&bucket); + let announced = events.iter().map(|event| event.version_id).collect::>(); + assert_eq!( + announced, + vec![Some(noncurrent_version_id)], + "only the committed noncurrent delete should be announced \ + (noncurrent={noncurrent_version_id}, current={current_version_id}, missing={missing_version_id}), got {events:?}" + ); + assert_eq!(events[0].event_name, EventName::LifecycleExpirationDelete.to_string()); + assert_eq!(events[0].object, object); + assert!(!events[0].delete_marker); + + let remaining = ecstore + .clone() + .list_object_versions(&bucket, object, None, None, None, 10) + .await + .expect("remaining versions should be listable") + .objects + .into_iter() + .filter(|candidate| candidate.name == object) + .collect::>(); + assert_eq!(remaining.len(), 1); + assert_eq!(remaining[0].version_id, Some(current_version_id)); + } + #[tokio::test] #[serial] async fn lifecycle_deletes_only_the_historical_null_version_after_versioning_is_reenabled() { diff --git a/crates/ecstore/src/bucket/lifecycle/object_handlers_common.rs b/crates/ecstore/src/bucket/lifecycle/object_handlers_common.rs index 3f4d1d8ac..6b57afcec 100644 --- a/crates/ecstore/src/bucket/lifecycle/object_handlers_common.rs +++ b/crates/ecstore/src/bucket/lifecycle/object_handlers_common.rs @@ -20,6 +20,7 @@ const LOG_SUBSYSTEM_LIFECYCLE: &str = "lifecycle"; const EVENT_LIFECYCLE_CLEANUP_SKIPPED: &str = "lifecycle_cleanup_skipped"; const EVENT_LIFECYCLE_CLEANUP_FAILED: &str = "lifecycle_cleanup_failed"; +use crate::bucket::lifecycle::bucket_lifecycle_audit::emit_noncurrent_expiration_event; use crate::bucket::lifecycle::lifecycle; use crate::bucket::lifecycle::replication_sink::{self, ReplicationObjectBridge}; use crate::object_api::ObjectOptions; @@ -98,6 +99,12 @@ pub async fn delete_object_versions( // version so it does not sit resident until TTL (ODC-26). if let Some(target) = to_del.get(i) { crate::object_api::notify_object_mutation(bucket, &target.object_name).await; + // Announce the version this batch actually removed. Cache + // eviction and replication scheduling keep their existing + // order and admission; the event is derived from the committed + // result, and a send failure never rolls back a delete that + // already happened (backlog#2202). + emit_noncurrent_expiration_event(bucket, target, deleted_obj, false); } if deleted_obj.replication_state.is_none() { continue; diff --git a/crates/ecstore/src/services/event_notification.rs b/crates/ecstore/src/services/event_notification.rs index d9f5267be..8e37f1c6b 100644 --- a/crates/ecstore/src/services/event_notification.rs +++ b/crates/ecstore/src/services/event_notification.rs @@ -117,26 +117,85 @@ pub fn send_event(args: EventArgs) { ); } +/// Shared event recorder for this crate's tests. +/// +/// [`register_event_dispatch_hook`] backs a `OnceLock`, so only the first +/// caller in a test binary can install a hook. Every test that needs to +/// observe dispatched events must therefore go through this single recorder +/// instead of registering its own. +#[cfg(test)] +pub(crate) mod test_recorder { + use super::register_event_dispatch_hook; + use std::sync::{Mutex, OnceLock}; + use uuid::Uuid; + + /// The fields a test needs from a dispatched event. `EventArgs` itself is + /// not `Clone`, and recording a reduced shape keeps this test seam from + /// constraining the production type. + #[derive(Clone, Debug)] + pub(crate) struct RecordedEvent { + pub(crate) event_name: String, + pub(crate) bucket: String, + pub(crate) object: String, + pub(crate) version_id: Option, + pub(crate) delete_marker: bool, + } + + static RECORDED: OnceLock>> = OnceLock::new(); + + fn recorded() -> &'static Mutex> { + RECORDED.get_or_init(|| Mutex::new(Vec::new())) + } + + pub(crate) fn install() { + static INSTALLED: OnceLock<()> = OnceLock::new(); + INSTALLED.get_or_init(|| { + assert!( + register_event_dispatch_hook(|args| { + recorded().lock().unwrap_or_else(|err| err.into_inner()).push(RecordedEvent { + event_name: args.event_name, + bucket: args.bucket_name, + object: args.object.name, + version_id: args.object.version_id, + delete_marker: args.object.delete_marker, + }); + }), + "the test event recorder must own this binary's dispatch hook" + ); + }); + } + + /// Everything recorded for one bucket. Tests select by their own unique + /// bucket name rather than draining, because tests that are not + /// `#[serial]` may dispatch events concurrently. + pub(crate) fn recorded_for_bucket(bucket: &str) -> Vec { + recorded() + .lock() + .unwrap_or_else(|err| err.into_inner()) + .iter() + .filter(|event| event.bucket == bucket) + .cloned() + .collect() + } +} + #[cfg(test)] mod tests { use super::*; - use std::sync::atomic::{AtomicUsize, Ordering}; - - static DISPATCH_COUNT: AtomicUsize = AtomicUsize::new(0); #[test] fn send_event_dispatches_to_registered_hook() { - let _ = register_event_dispatch_hook(|_args| { - DISPATCH_COUNT.fetch_add(1, Ordering::Relaxed); - }); - let before = DISPATCH_COUNT.load(Ordering::Relaxed); + test_recorder::install(); + let bucket = format!("event-dispatch-{}", uuid::Uuid::new_v4().simple()); send_event(EventArgs { event_name: "s3:ObjectCreated:Put".to_string(), - bucket_name: "demo".to_string(), + bucket_name: bucket.clone(), ..Default::default() }); - assert_eq!(DISPATCH_COUNT.load(Ordering::Relaxed), before + 1); + let dispatched = test_recorder::recorded_for_bucket(&bucket); + assert_eq!(dispatched.len(), 1); + assert_eq!(dispatched[0].event_name, "s3:ObjectCreated:Put"); } } diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index 40e45c902..5a220e432 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/crates/ecstore/src/set_disk/ops/object.rs @@ -7935,6 +7935,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { delete_marker_mtime: vr.mod_time.or(goi.mod_time), object_name: vr.name.clone(), replication_state: vr.replication_state_internal.clone(), + source_missing, ..Default::default() } } else { @@ -7946,6 +7947,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { vr.version_id }, replication_state: vr.replication_state_internal.clone(), + source_missing, ..Default::default() }; accounting[i] = Some(DeleteAccounting { diff --git a/crates/storage-api/src/object.rs b/crates/storage-api/src/object.rs index b2a090941..b7aaf28dc 100644 --- a/crates/storage-api/src/object.rs +++ b/crates/storage-api/src/object.rs @@ -243,6 +243,16 @@ impl ObjectToDelete { #[derive(Debug, Default, Clone)] pub struct DeletedObject { pub delete_marker: bool, + /// True when the delete plan looked the target up and found no such + /// object or version. + /// + /// The lookup only runs when the plan needs the source (Object Lock + /// check, replication decision, tier journal, or an expected identity), + /// so this proves absence and never proves presence: it stays false when + /// no lookup ran. Callers that must not announce a delete that removed + /// nothing need this, because the disk layer treats an absent version as + /// an idempotent success and reports `found` regardless. + pub source_missing: bool, pub delete_marker_version_id: Option, pub object_name: String, pub version_id: Option,