fix(ilm): notify on batch noncurrent version expiry (#7116)

The batch `NewerNoncurrentVersions` expiry path took a lifecycle event
argument and ignored it: after `delete_objects` committed it only evicted
the cache and scheduled replication deletes, so a successful noncurrent
version expiry was invisible to notification subscribers while the
equivalent current-version path emitted a lifecycle expiration event.

Emit that event from the batch path too, reusing the existing lifecycle
audit sink and event contract. Only entries that actually mutated
something are announced, and 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.

"No error" is not enough to prove a mutation: the disk layer skips an
absent version and reports success, so a batch entry for a version that
was already gone came back indistinguishable from a committed delete.
The delete plan already resolves whether the source exists, so carry that
`source_missing` result on `DeletedObject` and let the lifecycle path
stay silent for versions it did not remove.

backlog#2202

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<Uuid>) -> 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, &not_found, false).is_none());
+    }
 }

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"<VersioningConfiguration><Status>Enabled</Status></VersioningConfiguration>".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::<Vec<_>>();
+        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::<Vec<_>>();
+        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() {

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;

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<Uuid>,
+        pub(crate) delete_marker: bool,
+    }
+
+    static RECORDED: OnceLock<Mutex<Vec<RecordedEvent>>> = OnceLock::new();
+
+    fn recorded() -> &'static Mutex<Vec<RecordedEvent>> {
+        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<RecordedEvent> {
+        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");
     }
 }

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 {

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<Uuid>,
     pub object_name: String,
     pub version_id: Option<Uuid>,