fix(ilm): implement expire_restored delete semantics for restore expiry (#4950)

DeleteRestoredAction is supposed to demote a restored object back to its
pure transitioned state: remove only the local restored copy, strip the
x-amz-restore headers, and leave the version (and the remote tier data)
untouched. expire_transitioned_object set
opts.transition.expire_restored accordingly, but no delete path ever
read the flag, so delete_object ran an ordinary delete: on unversioned
buckets the whole object vanished and the free-version record scheduled
remote tier cleanup (tier data loss); on versioned buckets the latest
version got a spurious delete marker that replication propagated.

Route expire_restored explicitly in SetDisks::delete_object before
delete-marker resolution and replication dispatch: target the found
version with FileInfo.expire_restored=true and return early. The
FileMeta::delete_version layer already implements the semantics (strip
restore headers, keep the version, hand back the local data dir); this
wires it up.

Also fix the action matching in expire_transitioned_object (extracted
into transitioned_object_delete_opts): DeleteRestoredVersionAction
previously fell through to the full transitioned-object delete, which
removed the remote tier data of a noncurrent restored version. It now
routes through the same restored-copy cleanup with the exact version id,
matching MinIO's Action.DeleteVersioned()/DeleteRestored() dispatch.

Re-enable test_restore_chain_local_read_expiry_keeps_remote_and_allows_
re_restore in the ILM Integration (serial) lane; add unit tests pinning
the event->options routing and the filemeta expire_restored branch.

Closes rustfs/backlog#1302

.github/workflows/ci.yml

@@ -233,16 +233,11 @@ jobs:
       # they keep #[ignore] with a backlog reference):
       #   - test_noncurrent_{expiry,transition}_still_works_after_immediate_compensation_transition:
       #     noncurrent transition/expiry after an immediate compensation transition.
-      #   - test_restore_chain_local_read_expiry_keeps_remote_and_allows_re_restore:
-      #     DeleteRestoredAction sets opts.transition.expire_restored, but no
-      #     delete path reads that flag, so cleanup deletes the whole object
-      #     instead of only the local restored copy (ObjectNotFound afterwards).
-      #     The expire_restored delete semantics are unimplemented.
       - name: Run ignored ILM integration tests serially
         run: |
           cargo nextest run -j1 --run-ignored ignored-only \
             -p rustfs-scanner -p rustfs \
-            -E '(binary(lifecycle_integration_test) or (package(rustfs) and test(lifecycle_transition_api_test))) and not (test(test_noncurrent_expiry_still_works_after_immediate_compensation_transition) or test(test_noncurrent_transition_still_works_after_immediate_compensation_transition) or test(test_restore_chain_local_read_expiry_keeps_remote_and_allows_re_restore))'
+            -E '(binary(lifecycle_integration_test) or (package(rustfs) and test(lifecycle_transition_api_test))) and not (test(test_noncurrent_expiry_still_works_after_immediate_compensation_transition) or test(test_noncurrent_transition_still_works_after_immediate_compensation_transition))'
 
   test-and-lint-rio-v2:
     name: Test and Lint (rio-v2)

crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs

@@ -2294,24 +2294,47 @@ async fn enqueue_transition_with_lifecycle(oi: &ObjectInfo, lc: &BucketLifecycle
     false
 }
 
+/// Build the delete options for a lifecycle expiry event on a transitioned
+/// object. Versioned events target the exact version; restore-expiry events
+/// (`DeleteRestoredAction`/`DeleteRestoredVersionAction`) set
+/// `transition.expire_restored` so the set-layer delete strips only the
+/// `x-amz-restore` headers and the local restored copy while the version stays
+/// transitioned (rustfs/backlog#1302).
+fn transitioned_object_delete_opts(
+    oi: &ObjectInfo,
+    action: IlmAction,
+    versioned: bool,
+    version_suspended: bool,
+) -> ObjectOptions {
+    let mut opts = ObjectOptions {
+        versioned,
+        version_suspended,
+        expiration: ExpirationOptions { expire: true },
+        ..Default::default()
+    };
+    if action.delete_versioned() {
+        opts.version_id = oi.version_id.map(|id| id.to_string());
+    }
+    if action.delete_restored() {
+        opts.transition.expire_restored = true;
+    }
+    opts
+}
+
 pub async fn expire_transitioned_object(
     api: Arc<ECStore>,
     oi: &ObjectInfo,
     lc_event: &lifecycle::Event,
     _src: &LcEventSrc,
 ) -> Result<ObjectInfo, std::io::Error> {
-    let mut opts = ObjectOptions {
-        versioned: BucketVersioningSys::prefix_enabled(&oi.bucket, &oi.name).await,
-        version_suspended: BucketVersioningSys::prefix_suspended(&oi.bucket, &oi.name).await,
-        expiration: ExpirationOptions { expire: true },
-        ..Default::default()
-    };
-    if lc_event.action == IlmAction::DeleteVersionAction {
-        opts.version_id = oi.version_id.map(|id| id.to_string());
-    }
+    let opts = transitioned_object_delete_opts(
+        oi,
+        lc_event.action,
+        BucketVersioningSys::prefix_enabled(&oi.bucket, &oi.name).await,
+        BucketVersioningSys::prefix_suspended(&oi.bucket, &oi.name).await,
+    );
     //let tags = LcAuditEvent::new(src, lcEvent).Tags();
-    if lc_event.action == IlmAction::DeleteRestoredAction {
-        opts.transition.expire_restored = true;
+    if lc_event.action.delete_restored() {
         return match api.delete_object(&oi.bucket, &oi.name, opts).await {
             Ok(dobj) => {
                 // Drop any cached restored-copy body so it does not sit resident
@@ -3056,7 +3079,7 @@ mod tests {
         merge_stale_multipart_candidate, replication_state_for_delete, resolve_transition_queue_capacity,
         resolve_transition_queue_send_timeout, resolve_transition_worker_count, resolve_transition_workers_absolute_max,
         select_restore_s3_location, should_defer_date_expiry_for_recent_config_update,
-        should_reuse_lifecycle_delete_replication_state, transitioned_cleanup_tuple,
+        should_reuse_lifecycle_delete_replication_state, transitioned_cleanup_tuple, transitioned_object_delete_opts,
     };
     use crate::bucket::lifecycle::bucket_lifecycle_audit::LcEventSrc;
     use crate::bucket::lifecycle::replication_sink::{
@@ -3098,6 +3121,45 @@ mod tests {
     use tokio_util::sync::CancellationToken;
     use uuid::Uuid;
 
+    /// Pins the expiry-event routing for transitioned objects
+    /// (rustfs/backlog#1302): restore-expiry events must set
+    /// `transition.expire_restored` (strip-restored-copy semantics, never a
+    /// full delete), and versioned events must target the exact version.
+    #[test]
+    fn transitioned_object_delete_opts_routes_expiry_actions() {
+        let vid = Uuid::new_v4();
+        let vid_str = vid.to_string();
+        let oi = ObjectInfo {
+            bucket: "bucket".to_string(),
+            name: "object".to_string(),
+            version_id: Some(vid),
+            ..Default::default()
+        };
+
+        // Plain version expiry: exact version, real delete.
+        let opts = transitioned_object_delete_opts(&oi, IlmAction::DeleteVersionAction, true, false);
+        assert_eq!(opts.version_id.as_deref(), Some(vid_str.as_str()));
+        assert!(!opts.transition.expire_restored);
+        assert!(opts.expiration.expire);
+
+        // Restore-expiry of the latest version: restored-copy cleanup only.
+        let opts = transitioned_object_delete_opts(&oi, IlmAction::DeleteRestoredAction, true, false);
+        assert!(opts.version_id.is_none());
+        assert!(opts.transition.expire_restored);
+
+        // 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);
+        assert_eq!(opts.version_id.as_deref(), Some(vid_str.as_str()));
+        assert!(opts.transition.expire_restored);
+
+        // Whole-object expiry stays a real delete.
+        let opts = transitioned_object_delete_opts(&oi, IlmAction::DeleteAction, false, false);
+        assert!(opts.version_id.is_none());
+        assert!(!opts.transition.expire_restored);
+    }
+
     #[tokio::test]
     #[serial]
     async fn expiry_enqueue_reports_missed_without_worker_channel() {

crates/ecstore/src/set_disk/ops/object.rs

@@ -1930,6 +1930,31 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
             check_object_lock_delete(bucket, object, &goi, &opts).await?;
         }
 
+        if opts.transition.expire_restored {
+            // Restore-expiry (DeleteRestoredAction / DeleteRestoredVersionAction)
+            // must only drop the local restored copy and strip the x-amz-restore
+            // headers; the version itself stays transitioned (status=complete)
+            // and keeps serving GETs from the tier. Route it before delete-marker
+            // resolution and replication dispatch: a delete marker would hide the
+            // version, a replicated delete would remove it on the target, and a
+            // free-version record would schedule remote tier cleanup.
+            if !version_found {
+                return Err(gerr.unwrap_or_else(|| StorageError::ObjectNotFound(bucket.to_string(), object.to_string())));
+            }
+            let dfi = FileInfo {
+                name: object.to_string(),
+                version_id: goi.version_id,
+                mod_time: Some(opts.mod_time.unwrap_or_else(OffsetDateTime::now_utc)),
+                expire_restored: true,
+                ..Default::default()
+            };
+            self.delete_object_version(bucket, object, &dfi, false)
+                .await
+                .map_err(|e| to_object_err(e, vec![bucket, object]))?;
+            self.invalidate_get_object_metadata_cache(bucket, object).await;
+            return Ok(ObjectInfo::from_file_info(&dfi, bucket, object, opts.versioned || opts.version_suspended));
+        }
+
         let otd = ObjectToDelete {
             object_name: object.to_string(),
             version_id: opts

crates/filemeta/src/filemeta.rs

@@ -1140,6 +1140,53 @@ mod test {
         assert_eq!(fm, newfm)
     }
 
+    /// Regression for rustfs/backlog#1302: `delete_version` with
+    /// `expire_restored` must only strip the `x-amz-restore` headers and hand
+    /// back the local data dir for cleanup; the version itself must survive
+    /// with its transition metadata (and thus the remote tier copy) intact.
+    #[test]
+    fn test_delete_version_expire_restored_keeps_transitioned_version() {
+        use rustfs_utils::http::headers::{AMZ_RESTORE, AMZ_RESTORE_EXPIRY_DAYS, AMZ_RESTORE_REQUEST_DATE};
+
+        let mut fm = FileMeta::new();
+        let vid = Uuid::new_v4();
+        let data_dir = Uuid::new_v4();
+
+        let mut fi = FileInfo::new("restored.bin", 3, 2);
+        fi.version_id = Some(vid);
+        fi.data_dir = Some(data_dir);
+        fi.mod_time = Some(OffsetDateTime::now_utc());
+        fi.transition_status = TRANSITION_COMPLETE.to_string();
+        fi.transitioned_objname = "remote/obj".to_string();
+        fi.transition_tier = "COLDTIER".to_string();
+        fi.metadata.insert(
+            AMZ_RESTORE.to_string(),
+            "ongoing-request=\"false\", expiry-date=\"Fri, 17 Jul 2026 00:00:00 GMT\"".to_string(),
+        );
+        fi.metadata.insert(AMZ_RESTORE_EXPIRY_DAYS.to_string(), "1".to_string());
+        fi.metadata
+            .insert(AMZ_RESTORE_REQUEST_DATE.to_string(), "Thu, 16 Jul 2026 00:00:00 GMT".to_string());
+        fm.add_version(fi).unwrap();
+
+        let expire_fi = FileInfo {
+            name: "restored.bin".to_string(),
+            version_id: Some(vid),
+            expire_restored: true,
+            ..Default::default()
+        };
+        let freed = fm.delete_version(&expire_fi).unwrap();
+        assert_eq!(freed, Some(data_dir), "the restored copy's local data dir must be handed back");
+
+        assert_eq!(fm.versions.len(), 1, "the version must survive restored-copy expiry");
+        let after = fm.into_fileinfo("vol", "restored.bin", "", false, false, true).unwrap();
+        assert!(!after.metadata.contains_key(AMZ_RESTORE), "x-amz-restore must be stripped");
+        assert!(!after.metadata.contains_key(AMZ_RESTORE_EXPIRY_DAYS));
+        assert!(!after.metadata.contains_key(AMZ_RESTORE_REQUEST_DATE));
+        assert_eq!(after.transition_status, TRANSITION_COMPLETE);
+        assert_eq!(after.transitioned_objname, "remote/obj");
+        assert_eq!(after.transition_tier, "COLDTIER");
+    }
+
     #[test]
     fn test_get_idx_out_of_bounds_returns_error_without_panic() {
         let mut fm = FileMeta::new();

crates/scanner/tests/lifecycle_integration_test.rs

@@ -1780,7 +1780,7 @@ mod serial_tests {
     /// tier again -> a second restore succeeds.
     #[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-8); currently excluded there - DeleteRestoredAction/expire_restored delete semantics are unimplemented, so cleanup removes the whole object"]
+    #[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-8)"]
     async fn test_restore_chain_local_read_expiry_keeps_remote_and_allows_re_restore() {
         let (_disk_paths, ecstore) = setup_test_env().await;