diff --git a/crates/lifecycle/src/core.rs b/crates/lifecycle/src/core.rs index 61c3b06b0..f137e47a7 100644 --- a/crates/lifecycle/src/core.rs +++ b/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(¤t, 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(¤t, 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). /// diff --git a/crates/scanner/tests/lifecycle_integration_test.rs b/crates/scanner/tests/lifecycle_integration_test.rs index faa7a72e1..c843961ea 100644 --- a/crates/scanner/tests/lifecycle_integration_test.rs +++ b/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, bucket: &str, object: &str) -> Vec { + 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 = (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 = (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"); + } } diff --git a/rustfs/src/app/lifecycle_transition_api_test.rs b/rustfs/src/app/lifecycle_transition_api_test.rs index 6e2ac47fa..651d509d6 100644 --- a/rustfs/src/app/lifecycle_transition_api_test.rs +++ b/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 = (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" + ); +}