fix(ilm): delete historical null versions by exact identity (#7109)

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,

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.