test(ilm): add restore full-chain integration coverage (#4860)

* test(ilm): add restore full-chain integration coverage

* test(ilm): fix restore integration types

---------

Co-authored-by: houseme <housemecn@gmail.com>

crates/lifecycle/src/core.rs

@@ -3457,6 +3457,66 @@ mod tests {
         assert_eq!(event.action, IlmAction::DeleteAction);
     }
 
+    /// 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
+    /// the local restored copy. The restore branch fires independently of any
+    /// configured rules.
+    #[tokio::test]
+    async fn eval_inner_emits_delete_restored_action_when_restore_expired() {
+        let lc = BucketLifecycleConfiguration {
+            expiry_updated_at: None,
+            rules: vec![],
+        };
+        let now = datetime!(2025-06-01 12:00:00 UTC);
+
+        let current = ObjectOpts {
+            name: "docs/report.bin".to_string(),
+            mod_time: Some(datetime!(2025-05-01 00:00:00 UTC)),
+            is_latest: true,
+            transition_status: TRANSITION_COMPLETE.to_string(),
+            restore_expires: Some(now - Duration::hours(1)),
+            ..Default::default()
+        };
+        let event = lc.eval_inner(&current, now, 0).await;
+        assert_eq!(event.action, IlmAction::DeleteRestoredAction);
+
+        let noncurrent = ObjectOpts {
+            name: "docs/report.bin".to_string(),
+            mod_time: Some(datetime!(2025-05-01 00:00:00 UTC)),
+            is_latest: false,
+            version_id: Some(Uuid::from_u128(9)),
+            successor_mod_time: Some(datetime!(2025-05-02 00:00:00 UTC)),
+            transition_status: TRANSITION_COMPLETE.to_string(),
+            restore_expires: Some(now - Duration::hours(1)),
+            ..Default::default()
+        };
+        let event = lc.eval_inner(&noncurrent, now, 0).await;
+        assert_eq!(event.action, IlmAction::DeleteRestoredVersionAction);
+    }
+
+    /// A restored copy whose expiry is still in the future must not be
+    /// scheduled for cleanup.
+    #[tokio::test]
+    async fn eval_inner_keeps_unexpired_restore_copy() {
+        let lc = BucketLifecycleConfiguration {
+            expiry_updated_at: None,
+            rules: vec![],
+        };
+        let now = datetime!(2025-06-01 12:00:00 UTC);
+
+        let current = ObjectOpts {
+            name: "docs/report.bin".to_string(),
+            mod_time: Some(datetime!(2025-05-01 00:00:00 UTC)),
+            is_latest: true,
+            transition_status: TRANSITION_COMPLETE.to_string(),
+            restore_expires: Some(now + Duration::hours(1)),
+            ..Default::default()
+        };
+        let event = lc.eval_inner(&current, now, 0).await;
+        assert_eq!(event.action, IlmAction::NoneAction);
+    }
+
     /// Property-based tests for the rule evaluator (backlog#1148 ilm-14,
     /// follow-up to backlog#1030 / rustfs#4455).
     ///

crates/scanner/tests/lifecycle_integration_test.rs

@@ -1752,4 +1752,262 @@ mod serial_tests {
 
         assert!(deleted, "background scanner should delete zero-day exact-key lifecycle targets");
     }
+
+    /// Read the full object body through the regular GET path.
+    async fn read_object_fully(ecstore: &Arc<ECStore>, bucket: &str, object: &str) -> Vec<u8> {
+        let mut reader = ecstore
+            .get_object_reader(bucket, object, None, http::HeaderMap::new(), &ObjectOptions::default())
+            .await
+            .expect("Failed to open object reader");
+        let mut data = Vec::new();
+        reader
+            .stream
+            .read_to_end(&mut data)
+            .await
+            .expect("Failed to consume object stream");
+        data
+    }
+
+    /// backlog#1148 ilm-8: the full restore chain on a transitioned object.
+    ///
+    /// transition -> restore(days=1) -> the restored copy serves GET locally
+    /// (the mock tier records zero additional `get` calls) -> the scanner's
+    /// restore-expiry action (`DeleteRestoredAction`, driven directly like the
+    /// ilm-2 expiry test; the evaluator mapping from a past `restore_expires`
+    /// to this action is pinned by unit tests in crates/lifecycle) removes ONLY
+    /// the local restored copy -> the object is still transitioned, the remote
+    /// tier object is untouched (zero `remove` calls) -> GET streams from the
+    /// 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)"]
+    async fn test_restore_chain_local_read_expiry_keeps_remote_and_allows_re_restore() {
+        let (_disk_paths, ecstore) = setup_test_env().await;
+
+        let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase();
+        let backend = register_mock_tier(&tier_name).await;
+
+        let bucket_name = format!("test-restore-chain-{}", &Uuid::new_v4().simple().to_string()[..8]);
+        let object_name = "restore/report.bin";
+        // Position-dependent payload so a misaligned read is caught.
+        let payload: Vec<u8> = (0..256 * 1024).map(|i| (i % 251) as u8).collect();
+
+        create_test_bucket(&ecstore, bucket_name.as_str()).await;
+        set_bucket_lifecycle_transition_with_tier(bucket_name.as_str(), &tier_name)
+            .await
+            .expect("Failed to set lifecycle configuration");
+        upload_test_object(&ecstore, bucket_name.as_str(), object_name, &payload).await;
+
+        enqueue_transition_for_existing_objects(ecstore.clone(), bucket_name.as_str())
+            .await
+            .expect("Failed to enqueue transition for existing objects");
+        let transitioned = wait_for_transition(&ecstore, bucket_name.as_str(), object_name, TRANSITION_WAIT_TIMEOUT)
+            .await
+            .expect("object should transition before restore");
+        let remote_object = transitioned.transitioned_object.name.clone();
+
+        // Restore for one day.
+        let restore_opts = || ObjectOptions {
+            transition: TransitionOptions {
+                restore_request: RestoreRequest {
+                    days: Some(1),
+                    description: None,
+                    glacier_job_parameters: None,
+                    output_location: None,
+                    select_parameters: None,
+                    tier: None,
+                    type_: None,
+                },
+                ..Default::default()
+            },
+            ..Default::default()
+        };
+        ecstore
+            .clone()
+            .restore_transitioned_object(bucket_name.as_str(), object_name, &restore_opts())
+            .await
+            .expect("Failed to restore transitioned object");
+
+        let restored = ecstore
+            .get_object_info(bucket_name.as_str(), object_name, &ObjectOptions::default())
+            .await
+            .expect("Failed to load restored object info");
+        assert!(restored.restore_expires.is_some(), "restore must record an expiry date");
+        assert!(!restored.restore_ongoing, "restore must be complete");
+        assert_eq!(
+            restored.transitioned_object.status, "complete",
+            "restore must not clear the transitioned state"
+        );
+
+        // GET is served from the LOCAL restored copy: the mock tier records no
+        // further `get` calls.
+        let tier_gets_after_restore = backend.get_count().await;
+        let data = read_object_fully(&ecstore, bucket_name.as_str(), object_name).await;
+        assert_eq!(data, payload, "restored GET must return the original bytes");
+        assert_eq!(
+            backend.get_count().await,
+            tier_gets_after_restore,
+            "GET of a restored object must be served locally, not from the tier"
+        );
+
+        // The scanner's restore-expiry action removes only the local restored
+        // copy (same direct-drive pattern as the ilm-2 expiry test).
+        let lc_event = LcEvent {
+            action: IlmAction::DeleteRestoredAction,
+            ..Default::default()
+        };
+        expire_transitioned_object(ecstore.clone(), &restored, &lc_event, &LcEventSrc::Scanner)
+            .await
+            .expect("restore-expiry cleanup should succeed");
+
+        let after = ecstore
+            .get_object_info(bucket_name.as_str(), object_name, &ObjectOptions::default())
+            .await
+            .expect("object must still exist after restored-copy cleanup");
+        assert!(
+            after.restore_expires.is_none(),
+            "restore headers must be stripped by DeleteRestoredAction"
+        );
+        assert_eq!(
+            after.transitioned_object.status, "complete",
+            "the object must remain transitioned after restored-copy cleanup"
+        );
+        assert_eq!(backend.remove_count().await, 0, "restore-expiry must never remove the remote tier object");
+        assert!(
+            backend.contains(&remote_object).await,
+            "remote tier object must survive restored-copy cleanup"
+        );
+
+        // GET now streams from the tier again...
+        let tier_gets_before_remote_read = backend.get_count().await;
+        let data = read_object_fully(&ecstore, bucket_name.as_str(), object_name).await;
+        assert_eq!(data, payload, "post-cleanup GET must stream the original bytes from the tier");
+        assert!(
+            backend.get_count().await > tier_gets_before_remote_read,
+            "post-cleanup GET must hit the remote tier"
+        );
+
+        // ...and the object is restorable again.
+        ecstore
+            .clone()
+            .restore_transitioned_object(bucket_name.as_str(), object_name, &restore_opts())
+            .await
+            .expect("object must be restorable again after restored-copy cleanup");
+        let re_restored = ecstore
+            .get_object_info(bucket_name.as_str(), object_name, &ObjectOptions::default())
+            .await
+            .expect("Failed to load re-restored object info");
+        assert!(re_restored.restore_expires.is_some(), "second restore must record an expiry");
+    }
+
+    /// backlog#1148 ilm-8: restoring a transitioned MULTIPART object (>= 3
+    /// parts) must reassemble the exact part layout: part count and sizes,
+    /// the multipart ETag, and byte-identical content across part boundaries.
+    #[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)"]
+    async fn test_multipart_restore_preserves_parts_and_etag() {
+        let (_disk_paths, ecstore) = setup_test_env().await;
+
+        let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase();
+        let _backend = register_mock_tier(&tier_name).await;
+
+        let bucket_name = format!("test-restore-mpu3-{}", &Uuid::new_v4().simple().to_string()[..8]);
+        let object_name = "restore/multipart.bin";
+        // Three parts: 5 MiB + 5 MiB + small tail, with position-dependent
+        // bytes so any part-boundary mixup is caught.
+        let part_sizes = [5 * 1024 * 1024usize, 5 * 1024 * 1024, 4096];
+        let total: usize = part_sizes.iter().sum();
+        let expected: Vec<u8> = (0..total).map(|i| (i % 249) as u8).collect();
+
+        create_test_bucket(&ecstore, bucket_name.as_str()).await;
+        set_bucket_lifecycle_transition_with_tier(bucket_name.as_str(), &tier_name)
+            .await
+            .expect("Failed to set lifecycle configuration");
+
+        let upload = ecstore
+            .new_multipart_upload(bucket_name.as_str(), object_name, &ObjectOptions::default())
+            .await
+            .expect("Failed to create multipart upload");
+        let mut completed = Vec::new();
+        let mut offset = 0usize;
+        for (idx, part_size) in part_sizes.iter().enumerate() {
+            let mut reader = PutObjReader::from_vec(expected[offset..offset + part_size].to_vec());
+            let part = ecstore
+                .put_object_part(
+                    bucket_name.as_str(),
+                    object_name,
+                    &upload.upload_id,
+                    idx + 1,
+                    &mut reader,
+                    &ObjectOptions::default(),
+                )
+                .await
+                .expect("Failed to upload multipart part");
+            completed.push(CompletePart {
+                part_num: idx + 1,
+                etag: part.etag.clone(),
+                ..Default::default()
+            });
+            offset += part_size;
+        }
+        ecstore
+            .clone()
+            .complete_multipart_upload(bucket_name.as_str(), object_name, &upload.upload_id, completed, &ObjectOptions::default())
+            .await
+            .expect("Failed to complete multipart upload");
+
+        enqueue_transition_for_existing_objects(ecstore.clone(), bucket_name.as_str())
+            .await
+            .expect("Failed to enqueue transition for existing objects");
+        let transitioned = wait_for_transition(&ecstore, bucket_name.as_str(), object_name, TRANSITION_WAIT_TIMEOUT)
+            .await
+            .expect("multipart object should transition before restore");
+        assert_eq!(transitioned.parts.len(), part_sizes.len());
+        let etag_before = transitioned.etag.clone();
+        assert!(
+            etag_before.as_deref().is_some_and(|etag| etag.ends_with("-3")),
+            "expected a 3-part multipart etag, got {etag_before:?}"
+        );
+
+        ecstore
+            .clone()
+            .restore_transitioned_object(
+                bucket_name.as_str(),
+                object_name,
+                &ObjectOptions {
+                    transition: TransitionOptions {
+                        restore_request: RestoreRequest {
+                            days: Some(1),
+                            description: None,
+                            glacier_job_parameters: None,
+                            output_location: None,
+                            select_parameters: None,
+                            tier: None,
+                            type_: None,
+                        },
+                        ..Default::default()
+                    },
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("Failed to restore transitioned multipart object");
+
+        let restored = ecstore
+            .get_object_info(bucket_name.as_str(), object_name, &ObjectOptions::default())
+            .await
+            .expect("Failed to load restored multipart object info");
+        assert!(restored.restore_expires.is_some());
+        assert!(!restored.restore_ongoing);
+        assert_eq!(restored.parts.len(), part_sizes.len(), "restore must preserve the part count");
+        for (idx, part_size) in part_sizes.iter().enumerate() {
+            assert_eq!(restored.parts[idx].size, *part_size, "restore must preserve the size of part {}", idx + 1);
+        }
+        assert_eq!(restored.etag, etag_before, "restore must preserve the multipart ETag");
+
+        let data = read_object_fully(&ecstore, bucket_name.as_str(), object_name).await;
+        assert_eq!(data.len(), expected.len(), "restored multipart read-back length mismatch");
+        assert_eq!(data, expected, "restored multipart read-back must be byte-identical");
+    }
 }

rustfs/src/app/lifecycle_transition_api_test.rs

@@ -1520,3 +1520,127 @@ async fn put_bucket_lifecycle_configuration_rejects_zero_day_expiration() {
         "unexpected error message: {message}"
     );
 }
+
+/// backlog#1148 ilm-8: the RestoreObject API surface on a transitioned object.
+///
+/// POST restore(days=1) is accepted and immediately flips the object to
+/// `x-amz-restore: ongoing-request="true"` (the mock tier's injected GET
+/// latency keeps the background copy-back in flight); a second POST during
+/// that window is rejected with `ErrObjectRestoreAlreadyInProgress`; once the
+/// copy-back completes the object reports `ongoing-request="false"` with a
+/// future expiry-date; and a full GET is then served from the local restored
+/// copy (the mock tier records no further `get` calls).
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+#[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)"]
+async fn restore_object_usecase_reports_ongoing_conflict_and_completion() {
+    let (_disk_paths, ecstore) = setup_test_env().await;
+    let usecase = DefaultObjectUsecase::from_global();
+
+    let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase();
+    let backend = register_mock_tier(&tier_name).await;
+
+    let bucket = format!("test-api-restore-{}", &Uuid::new_v4().simple().to_string()[..8]);
+    let object = "restore/api-object.bin";
+    let payload: Vec<u8> = (0..128 * 1024).map(|i| (i % 251) as u8).collect();
+
+    create_test_bucket(&ecstore, bucket.as_str()).await;
+    set_bucket_lifecycle_transition_with_tier(bucket.as_str(), &tier_name)
+        .await
+        .expect("Failed to set lifecycle configuration");
+    let _ = upload_test_object(&ecstore, bucket.as_str(), object, &payload).await;
+
+    lifecycle::bucket_lifecycle_ops::enqueue_transition_for_existing_objects(ecstore.clone(), bucket.as_str())
+        .await
+        .expect("Failed to enqueue transitioned object");
+    let _ = wait_for_transition(&ecstore, bucket.as_str(), object, TRANSITION_WAIT_TIMEOUT)
+        .await
+        .expect("object should transition before the restore API runs");
+
+    // Slow the tier GET so the background copy-back stays in flight long
+    // enough to observe the ongoing state and the conflict rejection.
+    backend.set_latency(Some(Duration::from_millis(1500))).await;
+
+    let restore_request = || RestoreRequest {
+        days: Some(1),
+        description: None,
+        glacier_job_parameters: None,
+        output_location: None,
+        select_parameters: None,
+        tier: None,
+        type_: None,
+    };
+    let restore_input = || {
+        RestoreObjectInput::builder()
+            .bucket(bucket.clone())
+            .key(object.to_string())
+            .restore_request(Some(restore_request()))
+            .build()
+            .unwrap()
+    };
+
+    Box::pin(usecase.execute_restore_object(build_request(restore_input(), Method::POST)))
+        .await
+        .expect("restore request should be accepted");
+
+    // The accepted restore is immediately visible as ongoing (the metadata is
+    // written synchronously before the copy-back is spawned).
+    let ongoing = ecstore
+        .get_object_info(bucket.as_str(), object, &ObjectOptions::default())
+        .await
+        .expect("Failed to load object info during restore");
+    assert!(
+        ongoing.restore_ongoing,
+        "x-amz-restore must report ongoing-request=true right after the restore is accepted"
+    );
+
+    // A second restore while one is in flight is rejected.
+    let err = Box::pin(usecase.execute_restore_object(build_request(restore_input(), Method::POST)))
+        .await
+        .expect_err("a second restore while one is ongoing must be rejected");
+    assert_eq!(
+        err.code(),
+        &s3s::S3ErrorCode::Custom("ErrObjectRestoreAlreadyInProgress".into()),
+        "unexpected rejection for a repeated restore: {err:?}"
+    );
+
+    // Completion: ongoing flips to false and a future expiry-date appears.
+    let mut completed = None;
+    for _ in 0..40 {
+        let info = ecstore
+            .get_object_info(bucket.as_str(), object, &ObjectOptions::default())
+            .await
+            .expect("Failed to poll object info for restore completion");
+        if !info.restore_ongoing && info.restore_expires.is_some() {
+            completed = Some(info);
+            break;
+        }
+        tokio::time::sleep(Duration::from_millis(500)).await;
+    }
+    let completed = completed.expect("restore copy-back should complete within the poll window");
+    backend.clear_faults().await;
+
+    let now_secs = std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .expect("clock before unix epoch")
+        .as_secs() as i64;
+    let expires = completed.restore_expires.expect("completed restore carries an expiry");
+    assert!(
+        expires.unix_timestamp() > now_secs,
+        "restore expiry-date must be in the future, got {expires}"
+    );
+    assert_eq!(
+        completed.transitioned_object.status, "complete",
+        "restore must not clear the transitioned state"
+    );
+
+    // The restored copy serves GET locally: no further tier GETs.
+    let tier_gets_after_restore = backend.get_count().await;
+    let data = read_object_bytes(&ecstore, bucket.as_str(), object).await;
+    assert_eq!(data, payload, "restored GET must return the original bytes");
+    assert_eq!(
+        backend.get_count().await,
+        tier_gets_after_restore,
+        "GET of a restored object must be served locally, not from the tier"
+    );
+}