// Copyright 2024 RustFS Team // // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. // You may obtain a copy of the License at // // http://www.apache.org/licenses/LICENSE-2.0 // // Unless required by applicable law or agreed to in writing, software // distributed under the License is distributed on an "AS IS" BASIS, // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. // See the License for the specific language governing permissions and // limitations under the License. use futures::FutureExt; use rustfs_config::ENV_TEST_FORCE_IMMEDIATE_TRANSITION_ENQUEUE_TIMEOUT; use rustfs_scanner::scanner_folder::ScannerItem; use rustfs_scanner::scanner_io::ScannerIODisk; use rustfs_scanner::{ ScannerObjectInfo as ObjectInfo, ScannerObjectOptions as ObjectOptions, ScannerPutObjReader as PutObjReader, scanner::init_data_scanner, }; use s3s::dto::RestoreRequest; use serial_test::serial; use std::{ collections::HashMap, env, path::{Path, PathBuf}, sync::{Arc, Once, OnceLock}, time::Duration, }; use tokio::fs; use tokio::io::AsyncReadExt; use tokio_util::sync::CancellationToken; use tracing::info; use uuid::Uuid; mod storage_api; use storage_api::lifecycle::{ BUCKET_LIFECYCLE_CONFIG, BucketOperations, BucketOptions, BucketVersioningSys, CompletePart, DiskOption, ECStore, EcstoreError, Endpoint, EndpointServerPools, Endpoints, IlmAction, LcEvent, LcEventSrc, ListOperations as _, MakeBucketOptions, MockWarmBackend, MultipartOperations as _, ObjectIO as _, ObjectOperations as _, PoolEndpoints, STORAGE_FORMAT_FILE, TRANSITION_PENDING, TransitionCleanupStoreBarrier, TransitionOptions, assert_transition_meta_consistent, enqueue_transition_for_existing_objects, expire_transitioned_object, free_version_count, get_bucket_metadata, get_global_tier_config_mgr, init_background_expiry, init_bucket_metadata_sys, init_local_disks, is_err_object_not_found, is_err_version_not_found, new_disk, path2_bucket_object_with_base_path, recover_tier_delete_journal_entries, register_mock_tier_util, update_bucket_metadata, wait_for_free_version_absence, }; static GLOBAL_ENV: OnceLock<(Vec, Arc)> = OnceLock::new(); static INIT: Once = Once::new(); const TRANSITION_WAIT_TIMEOUT: Duration = Duration::from_secs(15); fn init_tracing() { INIT.call_once(|| { let _ = tracing_subscriber::fmt::try_init(); }); } /// Test helper: Create test environment with ECStore async fn setup_test_env() -> (Vec, Arc) { init_tracing(); // Fast path: already initialized, just clone and return if let Some((paths, ecstore)) = GLOBAL_ENV.get() { return (paths.clone(), ecstore.clone()); } // create temp dir as 4 disks with unique base dir let test_base_dir = format!("/tmp/rustfs_scanner_lifecycle_test_{}", uuid::Uuid::new_v4()); let temp_dir = std::path::PathBuf::from(&test_base_dir); if temp_dir.exists() { fs::remove_dir_all(&temp_dir).await.ok(); } fs::create_dir_all(&temp_dir).await.unwrap(); // create 4 disk dirs let disk_paths = vec![ temp_dir.join("disk1"), temp_dir.join("disk2"), temp_dir.join("disk3"), temp_dir.join("disk4"), ]; for disk_path in &disk_paths { fs::create_dir_all(disk_path).await.unwrap(); } // create EndpointServerPools let mut endpoints = Vec::new(); for (i, disk_path) in disk_paths.iter().enumerate() { let mut endpoint = Endpoint::try_from(disk_path.to_str().unwrap()).unwrap(); // set correct index endpoint.set_pool_index(0); endpoint.set_set_index(0); endpoint.set_disk_index(i); endpoints.push(endpoint); } let pool_endpoints = PoolEndpoints { legacy: false, set_count: 1, drives_per_set: 4, endpoints: Endpoints::from(endpoints), cmd_line: "test".to_string(), platform: format!("OS: {} | Arch: {}", std::env::consts::OS, std::env::consts::ARCH), }; let endpoint_pools = EndpointServerPools::from(vec![pool_endpoints]); // format disks (only first time) init_local_disks(endpoint_pools.clone()).await.unwrap(); // create ECStore with dynamic port 0 (let OS assign) or fixed 9002 if free let port = 9002; // for simplicity let server_addr: std::net::SocketAddr = format!("127.0.0.1:{port}").parse().unwrap(); let ecstore = ECStore::new(server_addr, endpoint_pools, CancellationToken::new()) .await .unwrap(); // init bucket metadata system let buckets_list = ecstore .list_bucket(&BucketOptions { no_metadata: true, ..Default::default() }) .await .unwrap(); let buckets = buckets_list.into_iter().map(|v| v.name).collect(); init_bucket_metadata_sys(ecstore.clone(), buckets).await; // Initialize background expiry workers init_background_expiry(ecstore.clone()).await; // Store in global once lock let _ = GLOBAL_ENV.set((disk_paths.clone(), ecstore.clone())); (disk_paths, ecstore) } async fn setup_isolated_test_env(init_expiry: bool) -> (Vec, Arc) { init_tracing(); let test_base_dir = format!("/tmp/rustfs_scanner_lifecycle_test_{}", uuid::Uuid::new_v4()); let temp_dir = std::path::PathBuf::from(&test_base_dir); if temp_dir.exists() { fs::remove_dir_all(&temp_dir).await.ok(); } fs::create_dir_all(&temp_dir).await.unwrap(); let disk_paths = vec![ temp_dir.join("disk1"), temp_dir.join("disk2"), temp_dir.join("disk3"), temp_dir.join("disk4"), ]; for disk_path in &disk_paths { fs::create_dir_all(disk_path).await.unwrap(); } let mut endpoints = Vec::new(); for (i, disk_path) in disk_paths.iter().enumerate() { let mut endpoint = Endpoint::try_from(disk_path.to_str().unwrap()).unwrap(); endpoint.set_pool_index(0); endpoint.set_set_index(0); endpoint.set_disk_index(i); endpoints.push(endpoint); } let pool_endpoints = PoolEndpoints { legacy: false, set_count: 1, drives_per_set: 4, endpoints: Endpoints::from(endpoints), cmd_line: "test".to_string(), platform: format!("OS: {} | Arch: {}", std::env::consts::OS, std::env::consts::ARCH), }; let endpoint_pools = EndpointServerPools::from(vec![pool_endpoints]); init_local_disks(endpoint_pools.clone()).await.unwrap(); let server_addr: std::net::SocketAddr = "127.0.0.1:0".parse().unwrap(); let ecstore = ECStore::new(server_addr, endpoint_pools, CancellationToken::new()) .await .unwrap(); let buckets_list = ecstore .list_bucket(&BucketOptions { no_metadata: true, ..Default::default() }) .await .unwrap(); let buckets = buckets_list.into_iter().map(|v| v.name).collect(); init_bucket_metadata_sys(ecstore.clone(), buckets).await; if init_expiry { init_background_expiry(ecstore.clone()).await; } (disk_paths, ecstore) } /// Test helper: Create a test bucket #[allow(dead_code)] async fn create_test_bucket(ecstore: &Arc, bucket_name: &str) { (**ecstore) .make_bucket(bucket_name, &Default::default()) .await .expect("Failed to create test bucket"); info!("Created test bucket: {}", bucket_name); } /// Test helper: Create a test lock bucket async fn create_test_lock_bucket(ecstore: &Arc, bucket_name: &str) { (**ecstore) .make_bucket( bucket_name, &MakeBucketOptions { lock_enabled: true, versioning_enabled: true, ..Default::default() }, ) .await .expect("Failed to create test bucket"); info!("Created test bucket: {}", bucket_name); } /// Test helper: Upload test object async fn upload_test_object(ecstore: &Arc, bucket: &str, object: &str, data: &[u8]) { let mut reader = PutObjReader::from_vec(data.to_vec()); let object_info = (**ecstore) .put_object(bucket, object, &mut reader, &ObjectOptions::default()) .await .expect("Failed to upload test object"); info!("Uploaded test object: {}/{} ({} bytes)", bucket, object, object_info.size); } async fn modeled_versioned_delete_opts(bucket: &str, object: &str) -> ObjectOptions { ObjectOptions { versioned: BucketVersioningSys::prefix_enabled(bucket, object).await, version_suspended: BucketVersioningSys::prefix_suspended(bucket, object).await, ..Default::default() } } /// Test helper: Set bucket lifecycle configuration #[allow(dead_code)] async fn set_bucket_lifecycle(bucket_name: &str) -> Result<(), Box> { // Create a simple lifecycle configuration XML with 0 days expiry for immediate testing let lifecycle_xml = r#" test-rule Enabled test/ 0 "#; update_bucket_metadata(bucket_name, BUCKET_LIFECYCLE_CONFIG, lifecycle_xml.as_bytes().to_vec()).await?; Ok(()) } /// Test helper: Set bucket lifecycle configuration #[allow(dead_code)] async fn set_bucket_lifecycle_deletemarker(bucket_name: &str) -> Result<(), Box> { // Create lifecycle rule that targets delete-marker cleanup only. // Keep Expiration.Days unset to avoid expiring live transitioned object versions. let lifecycle_xml = r#" test-rule Enabled test/ true "#; update_bucket_metadata(bucket_name, BUCKET_LIFECYCLE_CONFIG, lifecycle_xml.as_bytes().to_vec()).await?; Ok(()) } #[allow(dead_code)] async fn set_bucket_lifecycle_delmarker_expiration(bucket_name: &str, days: i64) -> Result<(), Box> { let lifecycle_xml = format!( r#" test-rule Enabled test/ {days} "# ); update_bucket_metadata(bucket_name, BUCKET_LIFECYCLE_CONFIG, lifecycle_xml.into_bytes()).await?; Ok(()) } #[allow(dead_code)] async fn set_bucket_lifecycle_transition_with_tier( bucket_name: &str, storage_class: &str, ) -> Result<(), Box> { // Create a simple lifecycle configuration XML with 0 days expiry for immediate testing let lifecycle_xml = format!( r#" test-rule Enabled test/ 0 {storage_class} test-rule2 Disabled test/ 0 {storage_class} "# ); update_bucket_metadata(bucket_name, BUCKET_LIFECYCLE_CONFIG, lifecycle_xml.into_bytes()).await?; Ok(()) } /// Test helper: Check if object exists async fn object_exists(ecstore: &Arc, bucket: &str, object: &str) -> bool { match (**ecstore).get_object_info(bucket, object, &ObjectOptions::default()).await { Ok(info) => !info.delete_marker, Err(_) => false, } } /// Test helper: Check if object exists #[allow(dead_code)] async fn object_is_delete_marker(ecstore: &Arc, bucket: &str, object: &str) -> bool { if let Ok(oi) = (**ecstore).get_object_info(bucket, object, &ObjectOptions::default()).await { println!("oi: {oi:?}"); oi.delete_marker } else { println!("object_is_delete_marker is error"); panic!("object_is_delete_marker is error"); } } #[allow(dead_code)] async fn wait_for_object_absence(ecstore: &Arc, bucket: &str, object: &str, timeout: Duration) -> bool { let deadline = tokio::time::Instant::now() + timeout; loop { if !object_exists(ecstore, bucket, object).await { return true; } if tokio::time::Instant::now() >= deadline { return false; } tokio::time::sleep(Duration::from_millis(200)).await; } } async fn object_version_count(ecstore: &Arc, bucket: &str, object: &str) -> usize { let mut marker = None; let mut version_marker = None; let mut count = 0; loop { let Ok(page) = ecstore .clone() .list_object_versions(bucket, object, marker.clone(), version_marker.clone(), None, 1000) .await else { return 0; }; count += page.objects.iter().filter(|version| version.name == object).count(); if !page.is_truncated { return count; } marker = page.next_marker; version_marker = page.next_version_idmarker; } } async fn wait_for_version_count(ecstore: &Arc, bucket: &str, object: &str, expected: usize, timeout: Duration) -> bool { let deadline = tokio::time::Instant::now() + timeout; loop { if object_version_count(ecstore, bucket, object).await == expected { return true; } if tokio::time::Instant::now() >= deadline { return false; } tokio::time::sleep(Duration::from_millis(50)).await; } } async fn scan_object_with_lifecycle(disk_path: &Path, bucket: &str, object: &str) { let mut endpoint = Endpoint::try_from(disk_path.to_str().unwrap()).unwrap(); endpoint.set_pool_index(0); endpoint.set_set_index(0); endpoint.set_disk_index(0); let disk = new_disk( &endpoint, &DiskOption { cleanup: false, health_check: false, }, ) .await .expect("failed to open local disk"); let metadata_path = disk_path.join(bucket).join(object).join(STORAGE_FORMAT_FILE); let relative_path = metadata_path.to_string_lossy().to_string(); let (_, scanner_path) = path2_bucket_object_with_base_path(disk_path.to_string_lossy().as_ref(), relative_path.as_str()); let file_type = fs::metadata(&metadata_path) .await .expect("failed to stat object metadata") .file_type(); let lifecycle = get_bucket_metadata(bucket) .await .expect("failed to load bucket metadata") .lifecycle_config .clone() .map(Arc::new); let item = ScannerItem { path: scanner_path.clone(), bucket: bucket.to_string(), prefix: object.to_string(), object_name: STORAGE_FORMAT_FILE.to_string(), file_type, lifecycle, object_lock: None, replication: None, heal_enabled: false, heal_bitrot: false, debug: false, }; disk.get_size(item).await.expect("scanner get_size should succeed"); } async fn scan_object_metadata(disk_path: &Path, bucket: &str, object: &str) { let mut endpoint = Endpoint::try_from(disk_path.to_str().unwrap()).unwrap(); endpoint.set_pool_index(0); endpoint.set_set_index(0); endpoint.set_disk_index(0); let disk = new_disk( &endpoint, &DiskOption { cleanup: false, health_check: false, }, ) .await .expect("failed to open local disk"); let metadata_path = disk_path.join(bucket).join(object).join(STORAGE_FORMAT_FILE); let relative_path = metadata_path.to_string_lossy().to_string(); let (_, scanner_path) = path2_bucket_object_with_base_path(disk_path.to_string_lossy().as_ref(), relative_path.as_str()); let file_type = fs::metadata(&metadata_path) .await .expect("failed to stat object metadata") .file_type(); let item = ScannerItem { path: scanner_path.clone(), bucket: bucket.to_string(), prefix: object.to_string(), object_name: STORAGE_FORMAT_FILE.to_string(), file_type, lifecycle: None, object_lock: None, replication: None, heal_enabled: false, heal_bitrot: false, debug: false, }; disk.get_size(item).await.expect("scanner get_size should succeed"); } /// Register the shared [`MockWarmBackend`] into the global tier config manager /// used by the scanner integration tests. Thin wrapper over the shared /// `register_mock_tier` helper (rustfs/backlog#1148 ilm-6). async fn register_mock_tier(tier_name: &str) -> MockWarmBackend { register_mock_tier_util(&get_global_tier_config_mgr(), tier_name).await } async fn wait_for_transition(ecstore: &Arc, bucket: &str, object: &str, timeout: Duration) -> Option { let deadline = tokio::time::Instant::now() + timeout; loop { if let Ok(info) = (**ecstore).get_object_info(bucket, object, &ObjectOptions::default()).await && info.transitioned_object.status == "complete" { return Some(info); } if tokio::time::Instant::now() >= deadline { return None; } tokio::time::sleep(Duration::from_millis(50)).await; } } // SAFETY: this helper is used only by `#[serial]` tests and runs under the single-threaded Tokio // runtime (`worker_threads = 1`), so no concurrent test can mutate process environment during the // `env::set_var` / `env::remove_var` window. #[allow(unsafe_code)] async fn with_forced_immediate_enqueue_timeout(test_fn: F) where F: FnOnce() -> Fut, Fut: std::future::Future, { let original = env::var_os(ENV_TEST_FORCE_IMMEDIATE_TRANSITION_ENQUEUE_TIMEOUT); unsafe { env::set_var(ENV_TEST_FORCE_IMMEDIATE_TRANSITION_ENQUEUE_TIMEOUT, "1"); } let result = std::panic::AssertUnwindSafe(test_fn()).catch_unwind().await; match original { Some(value) => unsafe { env::set_var(ENV_TEST_FORCE_IMMEDIATE_TRANSITION_ENQUEUE_TIMEOUT, value); }, None => unsafe { env::remove_var(ENV_TEST_FORCE_IMMEDIATE_TRANSITION_ENQUEUE_TIMEOUT); }, } if let Err(err) = result { std::panic::resume_unwind(err); } } mod serial_tests { use super::*; /// Regression for rustfs#3491 (backlog#1148 ilm-2): the expire/GET race on a /// transitioned (tiered) object. /// /// Before #3491, `expire_transitioned_object` removed the remote tier /// version **before** deleting local metadata. A GET that raced into the /// window between those two steps read local metadata still pointing at a /// remote version that was already gone, so the tier fetch failed and the /// client saw a spurious, user-visible error (`NoSuchVersion`). #3491 /// flipped the ordering: local metadata is deleted **first** (the object /// becomes atomically unreachable) and remote-tier cleanup is deferred to /// persisted free-version recovery -- so no live local metadata ever points /// at an already-removed remote version. /// /// This test pins the FIXED contract two complementary ways, both /// revert-proof (reverting to remote-first ordering turns them red): /// /// 1. Ordering (deterministic): immediately after /// `expire_transitioned_object` returns, the remote tier object is still /// present and the mock recorded **zero** remote `remove` calls -- /// proving the local delete happened with no synchronous remote removal /// (local-first). Remote-first ordering loses the object and records a /// `remove`. /// 2. Concurrent GET (user-visible): a tight GET loop runs concurrently /// with the expiry; every observation must be either a full, correct /// body (GET won) or a clean object/version-not-found (expiry won). A /// tier-fetch failure -- the #3491 symptom -- is never tolerated. #[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-2)"] async fn test_expire_transitioned_object_never_races_concurrent_get() { let (_disk_paths, ecstore) = setup_isolated_test_env(false).await; let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase(); let backend = register_mock_tier(&tier_name).await; backend.set_put_remote_version(Some(String::new())).await; let bucket_name = format!("test-expire-get-race-{}", &Uuid::new_v4().simple().to_string()[..8]); let object_name = "test/race-object.bin"; // Multi-block payload so the transitioned GET streams from the remote // tier rather than serving an inline fast-path. let payload: Vec = (0..64 * 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 expiry"); let remote_object = transitioned.transitioned_object.name.clone(); assert!( backend.contains(&remote_object).await, "transitioned remote object should exist in the mock tier before expiry" ); assert_eq!( backend.remove_count().await, 0, "no remote-tier removal should have happened before expiry" ); // The ObjectInfo the scanner would hand to the expiry action. let oi = ecstore .get_object_info(bucket_name.as_str(), object_name, &ObjectOptions::default()) .await .expect("Failed to load transitioned object info"); assert!( oi.transitioned_object.version_id.is_empty(), "the regression must exercise an unversioned remote tier" ); // Concurrent GET loop: hammer GET while the expiry runs. Every outcome // must be a full correct body or a clean not-found -- never a tier-fetch // failure. let get_store = ecstore.clone(); let get_bucket = bucket_name.clone(); let get_object = object_name.to_string(); let expected = payload.clone(); let get_loop = tokio::spawn(async move { let mut saw_full_body = 0usize; let mut saw_not_found = 0usize; for _ in 0..400 { match get_store .get_object_reader( get_bucket.as_str(), get_object.as_str(), None, http::HeaderMap::new(), &ObjectOptions::default(), ) .await { Ok(mut reader) => { let mut data = Vec::new(); match reader.stream.read_to_end(&mut data).await { Ok(_) => { assert_eq!( data, expected, "a successful GET during expiry must return the complete, correct body" ); saw_full_body += 1; } Err(err) => { panic!("GET during expiry streamed a truncated/failed body (expire/GET race regression): {err:?}") } } } Err(err) => { let ec: &EcstoreError = &err; assert!( is_err_object_not_found(ec) || is_err_version_not_found(ec), "GET during expiry may only fail with a clean object/version-not-found (expiry won \ the race); a tier-fetch failure is the #3491 regression: {err:?}" ); saw_not_found += 1; } } tokio::task::yield_now().await; } (saw_full_body, saw_not_found) }); // Run the exact expiry action the scanner drives for a transitioned // current version. let lc_event = LcEvent { action: IlmAction::DeleteAction, ..Default::default() }; let bucket_incarnation_id = ecstore .bucket_incarnation_id(bucket_name.as_str()) .await .expect("read bucket incarnation"); expire_transitioned_object(ecstore.clone(), &oi, &lc_event, &LcEventSrc::Scanner, bucket_incarnation_id) .await .expect("expire_transitioned_object should succeed"); // --- Ordering contract (deterministic revert-proof) ---------------- // #3491 defers remote cleanup to free-version recovery, so immediately // after expiry the remote object is still present and NO synchronous // remote `remove` was issued. Reverting to remote-first ordering makes // both assertions fail. assert_eq!( backend.remove_count().await, 0, "expire_transitioned_object must NOT issue a synchronous remote-tier removal (local-first \ ordering, #3491); remote cleanup is deferred to free-version recovery" ); assert!( backend.contains(&remote_object).await, "remote tier object must still exist immediately after expiry (deferred cleanup, #3491)" ); // Local metadata is gone: the object is atomically unreachable. assert!( wait_for_object_absence(&ecstore, bucket_name.as_str(), object_name, Duration::from_secs(5)).await, "local metadata for the expired transitioned object should be gone" ); // Drain the concurrent GET loop; its internal asserts already guarantee // no #3491-style tier-fetch failure was ever observed. let (saw_full_body, saw_not_found) = get_loop.await.expect("concurrent GET loop task panicked"); assert!( saw_full_body + saw_not_found > 0, "the concurrent GET loop should have observed at least one GET outcome" ); } #[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"] async fn rejected_transition_candidate_is_recovered_from_persisted_delete_journal() { let (_disk_paths, ecstore) = setup_isolated_test_env(false).await; let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase(); let backend = register_mock_tier(&tier_name).await; backend.set_put_read_limit(Some(4096)).await; backend.set_put_remote_version(Some(String::new())).await; backend.set_remove_failure(true); let bucket_name = format!("test-transition-cleanup-journal-{}", &Uuid::new_v4().simple().to_string()[..8]); let object_name = "test/rejected-candidate.bin"; let payload = b"rejected remote candidate must not replace the local source".repeat(1024); create_test_bucket(&ecstore, bucket_name.as_str()).await; upload_test_object(&ecstore, bucket_name.as_str(), object_name, &payload).await; let original = ecstore .get_object_info(bucket_name.as_str(), object_name, &ObjectOptions::default()) .await .expect("source object metadata should resolve before transition"); let opts = ObjectOptions { no_lock: true, transition: TransitionOptions { status: TRANSITION_PENDING.to_string(), tier: tier_name, etag: original.etag.clone().expect("uploaded source object should have an ETag"), ..Default::default() }, version_id: original.version_id.map(|version| version.to_string()), mod_time: original.mod_time, ..Default::default() }; ecstore .transition_object(bucket_name.as_str(), object_name, &opts) .await .expect_err("a remotely accepted partial upload must not commit transition metadata"); let put_versions = backend.put_versions().await; assert_eq!(put_versions.len(), 1, "transition should create exactly one remote candidate"); assert!(put_versions[0].1.is_empty()); assert_eq!( backend.object_count().await, 1, "failed cleanup must retain the remote candidate for recovery" ); assert!( backend.remove_versions().await.is_empty(), "no cleanup path may delete the candidate while remove failures are enabled" ); let retained = recover_tier_delete_journal_entries(ecstore.clone(), 100, None) .await .expect("tier delete journal recovery should scan the persisted candidate"); assert_eq!(retained.scanned, 1); assert_eq!(retained.deleted, 0); assert_eq!(retained.failed, 1); assert_eq!(backend.object_count().await, 1, "failed recovery must retain the remote candidate"); backend.set_remove_failure(false); let recovered = recover_tier_delete_journal_entries(ecstore.clone(), 100, None) .await .expect("tier delete journal recovery should delete the retained candidate"); assert_eq!(recovered.scanned, 1); assert_eq!(recovered.deleted, 1); assert_eq!(recovered.failed, 0); let removed_versions = backend.remove_versions().await; assert!(!removed_versions.is_empty(), "recovery must issue at least one successful delete"); assert!( removed_versions.iter().all(|removed| removed == &put_versions[0]), "every idempotent cleanup must delete the exact PUT object and version" ); assert_eq!(backend.object_count().await, 0, "recovery should remove the rejected remote candidate"); let empty = recover_tier_delete_journal_entries(ecstore.clone(), 100, None) .await .expect("a removed tier delete journal entry should no longer be listed"); assert_eq!(empty.scanned, 0, "successful recovery must remove the persisted journal entry"); assert_eq!( read_object_fully(&ecstore, bucket_name.as_str(), object_name).await, payload, "rejected transition cleanup must leave the source object readable" ); } #[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"] async fn cancelled_before_cleanup_store_resolution_persists_journal() { let (_disk_paths, ecstore) = setup_isolated_test_env(false).await; let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase(); let backend = register_mock_tier(&tier_name).await; backend.set_put_remote_version(Some(Uuid::new_v4().to_string())).await; backend.reject_next_non_empty_remote_version_validation(); backend.set_remove_failure(true); let cleanup_store_barrier = TransitionCleanupStoreBarrier::install(); let bucket_name = format!("test-transition-cancel-cleanup-{}", &Uuid::new_v4().simple().to_string()[..8]); let object_name = "test/rejected-candidate.bin"; let payload = b"cancelled rejected cleanup must retain durable recovery evidence".repeat(1024); create_test_bucket(&ecstore, bucket_name.as_str()).await; upload_test_object(&ecstore, bucket_name.as_str(), object_name, &payload).await; let original = ecstore .get_object_info(bucket_name.as_str(), object_name, &ObjectOptions::default()) .await .expect("source object metadata should resolve before transition"); let opts = ObjectOptions { no_lock: true, transition: TransitionOptions { status: TRANSITION_PENDING.to_string(), tier: tier_name, etag: original.etag.clone().expect("uploaded source object should have an ETag"), ..Default::default() }, version_id: original.version_id.map(|version| version.to_string()), mod_time: original.mod_time, ..Default::default() }; let transition_store = ecstore.clone(); let transition_bucket = bucket_name.clone(); let transition = tokio::spawn(async move { transition_store .transition_object(transition_bucket.as_str(), object_name, &opts) .await }); cleanup_store_barrier.wait_until_paused().await; transition.abort(); assert!( transition .await .expect_err("aborted transition task should be cancelled") .is_cancelled() ); let retained = tokio::time::timeout(Duration::from_secs(30), async { loop { let recovery = recover_tier_delete_journal_entries(ecstore.clone(), 100, None) .await .expect("the cancelled transition journal should be readable"); if recovery.scanned > 0 { break recovery; } tokio::task::yield_now().await; } }) .await .expect("Drop should persist the rejected candidate through the saved instance context"); assert_eq!((retained.scanned, retained.deleted, retained.failed), (1, 0, 1)); tokio::time::timeout(Duration::from_secs(5), async { while backend.exact_remove_count() < 2 { tokio::task::yield_now().await; } }) .await .expect("Drop cleanup and failed journal recovery must both preserve the exact version constraint"); let failed_exact_attempts = backend.exact_remove_count(); assert_eq!(backend.object_count().await, 1); assert!(backend.remove_versions().await.is_empty()); backend.set_remove_failure(false); let recovered = recover_tier_delete_journal_entries(ecstore.clone(), 100, None) .await .expect("recovery should delete the candidate retained by the cancelled transition"); assert_eq!((recovered.scanned, recovered.deleted, recovered.failed), (1, 1, 0)); assert_eq!(backend.remove_versions().await, backend.put_versions().await); assert_eq!(backend.exact_remove_count(), failed_exact_attempts + 1); assert_eq!(backend.object_count().await, 0); let empty = recover_tier_delete_journal_entries(ecstore.clone(), 100, None) .await .expect("successful recovery should remove the cancellation journal"); assert_eq!(empty.scanned, 0); assert_eq!( read_object_fully(&ecstore, bucket_name.as_str(), object_name).await, payload, "cancelled transition cleanup must preserve the local source" ); } #[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"] async fn rejected_transition_cleanup_durability_matrix() { #[derive(Clone, Copy)] enum CleanupCase { Persisted, DeleteFallback, RetryPersisted, FullyFailed, } let (_disk_paths, ecstore) = setup_isolated_test_env(false).await; for case in [ CleanupCase::Persisted, CleanupCase::DeleteFallback, CleanupCase::RetryPersisted, CleanupCase::FullyFailed, ] { let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase(); let backend = register_mock_tier(&tier_name).await; let remote_version = Uuid::new_v4().to_string(); backend.set_put_remote_version(Some(remote_version.clone())).await; backend.reject_next_non_empty_remote_version_validation(); backend.set_remove_failure(matches!(case, CleanupCase::FullyFailed)); let put_barrier = backend.arm_put_barrier().await; let remove_barrier = if matches!(case, CleanupCase::RetryPersisted) { Some(backend.arm_failing_remove_barrier().await) } else { None }; let bucket_name = format!("test-transition-journal-failure-{}", &Uuid::new_v4().simple().to_string()[..8]); let object_name = "test/rejected-candidate.bin"; let payload = b"journal failure must either delete the exact candidate or report both failures".repeat(1024); create_test_bucket(&ecstore, bucket_name.as_str()).await; upload_test_object(&ecstore, bucket_name.as_str(), object_name, &payload).await; let original = ecstore .get_object_info(bucket_name.as_str(), object_name, &ObjectOptions::default()) .await .expect("source object metadata should resolve before transition"); let opts = ObjectOptions { no_lock: true, transition: TransitionOptions { status: TRANSITION_PENDING.to_string(), tier: tier_name, etag: original.etag.clone().expect("uploaded source object should have an ETag"), ..Default::default() }, version_id: original.version_id.map(|version| version.to_string()), mod_time: original.mod_time, ..Default::default() }; let transition_store = ecstore.clone(); let transition_bucket = bucket_name.clone(); let transition = tokio::spawn(async move { transition_store .transition_object(transition_bucket.as_str(), object_name, &opts) .await }); put_barrier.wait_until_paused().await; let set = ecstore.pools[0].get_disks(0); let mut saved_disks = if matches!(case, CleanupCase::Persisted) { None } else { let mut disks = set.disks.write().await; let saved = std::mem::take(&mut *disks); *disks = vec![None; saved.len()]; Some(saved) }; put_barrier.release(); if let Some(remove_barrier) = remove_barrier.as_ref() { remove_barrier.wait_until_paused().await; *set.disks.write().await = saved_disks.take().expect("offline disks should be restorable"); backend.set_remove_failure(true); remove_barrier.release(); } let transition_result = tokio::time::timeout(Duration::from_secs(30), transition).await; if let Some(saved_disks) = saved_disks { *set.disks.write().await = saved_disks; } let err = transition_result .expect("transition should finish while validating cleanup durability") .expect("transition task should not panic") .expect_err("a versioned candidate must not commit to an unversioned tier"); match case { CleanupCase::Persisted | CleanupCase::DeleteFallback => { assert_eq!(backend.remove_versions().await, backend.put_versions().await); assert_eq!(backend.object_count().await, 0, "cleanup must remove the exact candidate"); let recovery = recover_tier_delete_journal_entries(ecstore.clone(), 100, None) .await .expect("successful cleanup must not retain a journal entry"); assert_eq!(recovery.scanned, 0); } CleanupCase::RetryPersisted => { assert!( !err.to_string().contains("journal retry error"), "a successful journal retry must preserve the original version-constraint error" ); assert_eq!(backend.object_count().await, 1); let retained = recover_tier_delete_journal_entries(ecstore.clone(), 100, None) .await .expect("the retried journal should be recoverable"); assert_eq!((retained.scanned, retained.deleted, retained.failed), (1, 0, 1)); backend.set_remove_failure(false); let recovered = recover_tier_delete_journal_entries(ecstore.clone(), 100, None) .await .expect("recovery should delete the exact retried candidate"); assert_eq!((recovered.scanned, recovered.deleted, recovered.failed), (1, 1, 0)); assert_eq!(backend.remove_versions().await, backend.put_versions().await); assert_eq!(backend.object_count().await, 0); let empty = recover_tier_delete_journal_entries(ecstore.clone(), 100, None) .await .expect("successful recovery must remove the retried journal"); assert_eq!(empty.scanned, 0); } CleanupCase::FullyFailed => { let message = err.to_string(); assert!(message.contains("initial journal error"), "{message}"); assert!(message.contains("cleanup error"), "{message}"); assert!(message.contains("journal retry error"), "{message}"); assert_eq!(backend.object_count().await, 1, "both failed safeguards must leave the candidate visible"); assert!(backend.remove_versions().await.is_empty()); let recovery = recover_tier_delete_journal_entries(ecstore.clone(), 100, None) .await .expect("failed journal writes must not create partial recovery entries"); assert_eq!(recovery.scanned, 0); } } assert_eq!( read_object_fully(&ecstore, bucket_name.as_str(), object_name).await, payload, "rejected transition cleanup must preserve the local source" ); } } #[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-1)"] async fn test_transition_and_restore_flows() { 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 put_bucket = format!("test-immediate-put-{}", &Uuid::new_v4().simple().to_string()[..8]); let put_object = "test/object.txt"; let put_payload = b"Hello, immediate transition!"; create_test_bucket(&ecstore, put_bucket.as_str()).await; set_bucket_lifecycle_transition_with_tier(put_bucket.as_str(), &tier_name) .await .expect("Failed to set lifecycle configuration"); let mut reader = PutObjReader::from_vec(put_payload.to_vec()); let mut metadata = HashMap::new(); metadata.insert("content-type".to_string(), "text/plain".to_string()); ecstore .put_object( put_bucket.as_str(), put_object, &mut reader, &ObjectOptions { user_defined: metadata, ..Default::default() }, ) .await .expect("Failed to upload transition metadata test object"); enqueue_transition_for_existing_objects(ecstore.clone(), put_bucket.as_str()) .await .expect("Failed to enqueue transitioned put object"); let put_info = wait_for_transition(&ecstore, put_bucket.as_str(), put_object, TRANSITION_WAIT_TIMEOUT) .await .expect("object should transition after enqueueing existing objects"); assert_eq!(put_info.transitioned_object.status, "complete"); assert_eq!(put_info.transitioned_object.tier, tier_name); assert!(backend.contains(&put_info.transitioned_object.name).await); { let transitioned = backend .stored(&put_info.transitioned_object.name) .await .expect("transitioned object should be present in mock backend"); assert_eq!(transitioned.metadata.get("content-type"), Some(&"text/plain".to_string())); assert!( !transitioned.metadata.contains_key("x-amz-replication-status"), "transitioned objects must not inherit replication status defaults" ); assert!( !transitioned.metadata.contains_key("x-amz-object-lock-legal-hold"), "transitioned objects must not invent object lock headers" ); } // Cross-shard xl.meta transition assertion helper (rustfs/backlog#1148 ilm-6): // every disk must agree on the transition tuple for the object. let put_meta = assert_transition_meta_consistent(&disk_paths, put_bucket.as_str(), put_object).await; assert_eq!(put_meta.status, "complete"); assert_eq!(put_meta.tier, tier_name); let multipart_bucket = format!("test-immediate-mpu-{}", &Uuid::new_v4().simple().to_string()[..8]); let multipart_object = "test/multipart.txt"; create_test_bucket(&ecstore, multipart_bucket.as_str()).await; set_bucket_lifecycle_transition_with_tier(multipart_bucket.as_str(), &tier_name) .await .expect("Failed to set lifecycle configuration"); let upload = ecstore .new_multipart_upload(multipart_bucket.as_str(), multipart_object, &ObjectOptions::default()) .await .expect("Failed to create multipart upload"); let part_data = b"multipart immediate transition"; let mut reader = PutObjReader::from_vec(part_data.to_vec()); let part = ecstore .put_object_part( multipart_bucket.as_str(), multipart_object, &upload.upload_id, 1, &mut reader, &ObjectOptions::default(), ) .await .expect("Failed to upload multipart part"); ecstore .clone() .complete_multipart_upload( multipart_bucket.as_str(), multipart_object, &upload.upload_id, vec![CompletePart { part_num: 1, etag: part.etag.clone(), ..Default::default() }], &ObjectOptions::default(), ) .await .expect("Failed to complete multipart upload"); enqueue_transition_for_existing_objects(ecstore.clone(), multipart_bucket.as_str()) .await .expect("Failed to enqueue transitioned multipart object"); let multipart_info = wait_for_transition(&ecstore, multipart_bucket.as_str(), multipart_object, TRANSITION_WAIT_TIMEOUT) .await .expect("object should transition after enqueueing existing objects"); assert_eq!(multipart_info.transitioned_object.status, "complete"); assert_eq!(multipart_info.transitioned_object.tier, tier_name); assert!(backend.contains(&multipart_info.transitioned_object.name).await); let src_bucket = format!("test-immediate-copy-src-{}", &Uuid::new_v4().simple().to_string()[..8]); let dst_bucket = format!("test-immediate-copy-dst-{}", &Uuid::new_v4().simple().to_string()[..8]); let src_object = "test/source.txt"; let dst_object = "test/copied.txt"; let payload = b"copy object immediate transition"; create_test_bucket(&ecstore, src_bucket.as_str()).await; create_test_bucket(&ecstore, dst_bucket.as_str()).await; set_bucket_lifecycle_transition_with_tier(dst_bucket.as_str(), &tier_name) .await .expect("Failed to set destination lifecycle configuration"); upload_test_object(&ecstore, src_bucket.as_str(), src_object, payload).await; let mut src_info = ecstore .get_object_info(src_bucket.as_str(), src_object, &ObjectOptions::default()) .await .expect("Failed to load source object info"); src_info.put_object_reader = Some(PutObjReader::from_vec(payload.to_vec())); ecstore .copy_object( src_bucket.as_str(), src_object, dst_bucket.as_str(), dst_object, &mut src_info, &ObjectOptions::default(), &ObjectOptions::default(), ) .await .expect("Failed to copy object"); enqueue_transition_for_existing_objects(ecstore.clone(), dst_bucket.as_str()) .await .expect("Failed to enqueue transitioned copied object"); let copy_info = wait_for_transition(&ecstore, dst_bucket.as_str(), dst_object, TRANSITION_WAIT_TIMEOUT) .await .expect("copied object should transition after enqueueing existing objects"); assert_eq!(copy_info.transitioned_object.status, "complete"); assert_eq!(copy_info.transitioned_object.tier, tier_name); assert!(backend.contains(©_info.transitioned_object.name).await); let bucket_name = format!("test-lifecycle-update-{}", &Uuid::new_v4().simple().to_string()[..8]); let object_name = "test/existing.txt"; let payload = b"existing object before lifecycle"; create_test_bucket(&ecstore, bucket_name.as_str()).await; upload_test_object(&ecstore, bucket_name.as_str(), object_name, payload).await; set_bucket_lifecycle_transition_with_tier(bucket_name.as_str(), &tier_name) .await .expect("Failed to set lifecycle configuration"); enqueue_transition_for_existing_objects(ecstore.clone(), bucket_name.as_str()) .await .expect("Failed to enqueue transition for existing objects"); let info = wait_for_transition(&ecstore, bucket_name.as_str(), object_name, TRANSITION_WAIT_TIMEOUT) .await .expect("existing object should transition after lifecycle update"); assert_eq!(info.transitioned_object.status, "complete"); assert_eq!(info.transitioned_object.tier, tier_name); assert!(backend.contains(&info.transitioned_object.name).await); let bucket_name = format!("test-restore-mpu-{}", &Uuid::new_v4().simple().to_string()[..8]); let object_name = "test/restore.txt"; let part1 = vec![b'a'; 5 * 1024 * 1024]; let part2 = b"restored-tail".to_vec(); let expected = [part1.clone(), part2.clone()].concat(); 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 part1_reader = PutObjReader::from_vec(part1); let uploaded_part1 = ecstore .put_object_part( bucket_name.as_str(), object_name, &upload.upload_id, 1, &mut part1_reader, &ObjectOptions::default(), ) .await .expect("Failed to upload first multipart part"); let mut part2_reader = PutObjReader::from_vec(part2); let uploaded_part2 = ecstore .put_object_part( bucket_name.as_str(), object_name, &upload.upload_id, 2, &mut part2_reader, &ObjectOptions::default(), ) .await .expect("Failed to upload second multipart part"); ecstore .clone() .complete_multipart_upload( bucket_name.as_str(), object_name, &upload.upload_id, vec![ CompletePart { part_num: 1, etag: uploaded_part1.etag.clone(), ..Default::default() }, CompletePart { part_num: 2, etag: uploaded_part2.etag.clone(), ..Default::default() }, ], &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 transitioned restore object"); let transitioned = wait_for_transition(&ecstore, bucket_name.as_str(), object_name, TRANSITION_WAIT_TIMEOUT) .await .expect("multipart object should transition after enqueueing existing objects"); assert_eq!(transitioned.parts.len(), 2); 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 object info"); assert_eq!(restored.parts.len(), 2); assert!(restored.restore_expires.is_some()); assert!(!restored.restore_ongoing); let mut reader = ecstore .get_object_reader(bucket_name.as_str(), object_name, None, http::HeaderMap::new(), &ObjectOptions::default()) .await .expect("Failed to read restored object"); let mut data = Vec::new(); reader .stream .read_to_end(&mut data) .await .expect("Failed to consume restored object stream"); assert_eq!(data, expected); } #[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-1)"] async fn test_scanner_enqueues_free_version_cleanup_for_stale_transitioned_object() { let (disk_paths, ecstore) = setup_isolated_test_env(false).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-scanner-free-version-{}", &Uuid::new_v4().simple().to_string()[..8]); let object_name = "test/object.txt"; let initial_payload = b"scanner should clean stale transitioned null version"; 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, initial_payload).await; enqueue_transition_for_existing_objects(ecstore.clone(), bucket_name.as_str()) .await .expect("Failed to enqueue transitioned object"); let transitioned = wait_for_transition(&ecstore, bucket_name.as_str(), object_name, TRANSITION_WAIT_TIMEOUT) .await .expect("object should transition before overwrite"); let stale_remote_object = transitioned.transitioned_object.name.clone(); assert!(backend.contains(&stale_remote_object).await); ecstore .delete_object(bucket_name.as_str(), object_name, ObjectOptions::default()) .await .expect("Failed to delete transitioned object without expiry workers"); assert!( free_version_count(&disk_paths[0], bucket_name.as_str(), object_name).await > 0, "deleting a transitioned null version should leave a free version for async cleanup" ); assert!( backend.contains(&stale_remote_object).await, "stale transitioned remote object should still exist before scanner fallback runs" ); init_background_expiry(ecstore.clone()).await; scan_object_metadata(&disk_paths[0], bucket_name.as_str(), object_name).await; assert!( backend .wait_for_remote_absence(&stale_remote_object, TRANSITION_WAIT_TIMEOUT) .await, "scanner should enqueue stale free-version cleanup for the transitioned remote object" ); assert!( wait_for_free_version_absence(&disk_paths[0], bucket_name.as_str(), object_name, TRANSITION_WAIT_TIMEOUT).await, "free-version metadata should be removed after scanner-triggered cleanup" ); assert!( wait_for_object_absence(&ecstore, bucket_name.as_str(), object_name, Duration::from_secs(1)).await, "deleted object should remain absent after scanner cleanup" ); } #[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-1)"] async fn test_scanner_cleanup_still_works_after_immediate_compensation_transition() { let (disk_paths, ecstore) = setup_isolated_test_env(false).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-scanner-after-compensation-{}", &Uuid::new_v4().simple().to_string()[..8]); let object_name = "test/object.txt"; let payload = b"scanner cleanup should still work after immediate compensation"; 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"); with_forced_immediate_enqueue_timeout(|| async { upload_test_object(&ecstore, bucket_name.as_str(), object_name, payload).await; }) .await; let transitioned = wait_for_transition(&ecstore, bucket_name.as_str(), object_name, TRANSITION_WAIT_TIMEOUT) .await .expect("object should transition after compensation backfill"); let stale_remote_object = transitioned.transitioned_object.name.clone(); assert!(backend.contains(&stale_remote_object).await); ecstore .delete_object(bucket_name.as_str(), object_name, ObjectOptions::default()) .await .expect("Failed to delete transitioned object after compensation-driven transition"); assert!( free_version_count(&disk_paths[0], bucket_name.as_str(), object_name).await > 0, "deleting a compensation-transitioned null version should leave a free version for async cleanup" ); assert!( backend.contains(&stale_remote_object).await, "stale transitioned remote object should still exist before scanner cleanup runs" ); init_background_expiry(ecstore.clone()).await; scan_object_metadata(&disk_paths[0], bucket_name.as_str(), object_name).await; assert!( backend .wait_for_remote_absence(&stale_remote_object, TRANSITION_WAIT_TIMEOUT) .await, "scanner should clean stale remote object even after immediate compensation transitioned it" ); assert!( wait_for_free_version_absence(&disk_paths[0], bucket_name.as_str(), object_name, TRANSITION_WAIT_TIMEOUT).await, "free-version metadata should be removed after scanner cleanup" ); } #[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-1)"] async fn test_existing_object_backfill_is_idempotent_after_immediate_compensation_transition() { let (_disk_paths, ecstore) = setup_isolated_test_env(false).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-backfill-after-compensation-{}", &Uuid::new_v4().simple().to_string()[..8]); let object_name = "test/object.txt"; let payload = b"existing-object backfill should be idempotent after compensation transition"; 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"); with_forced_immediate_enqueue_timeout(|| async { upload_test_object(&ecstore, bucket_name.as_str(), object_name, payload).await; }) .await; let transitioned = wait_for_transition(&ecstore, bucket_name.as_str(), object_name, TRANSITION_WAIT_TIMEOUT) .await .expect("object should transition after immediate compensation backfill"); let remote_object = transitioned.transitioned_object.name.clone(); assert!(backend.contains(&remote_object).await); enqueue_transition_for_existing_objects(ecstore.clone(), bucket_name.as_str()) .await .expect("existing-object backfill should succeed after compensation transition"); let info = wait_for_transition(&ecstore, bucket_name.as_str(), object_name, TRANSITION_WAIT_TIMEOUT) .await .expect("object should remain transitioned after existing-object backfill rerun"); assert_eq!(info.transitioned_object.status, "complete"); assert_eq!(info.transitioned_object.tier, tier_name); assert_eq!(info.transitioned_object.name, remote_object); assert!(backend.contains(&remote_object).await); } #[tokio::test(flavor = "multi_thread", worker_threads = 1)] #[serial] #[ignore = "FAILING on main: excluded from the serial ILM lane pending a fix, see rustfs/backlog#1148 (ilm-1 partial)"] async fn test_noncurrent_expiry_still_works_after_immediate_compensation_transition() { let (disk_paths, ecstore) = setup_isolated_test_env(true).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-versioned-compensation-{}", &Uuid::new_v4().simple().to_string()[..8]); let object_name = "test/object.txt"; create_test_lock_bucket(&ecstore, bucket_name.as_str()).await; let lifecycle_xml = format!( r#" test-rule Enabled test/ 0 {tier_name} 0 "# ); update_bucket_metadata(bucket_name.as_str(), BUCKET_LIFECYCLE_CONFIG, lifecycle_xml.into_bytes()) .await .expect("Failed to set lifecycle configuration"); let mut reader = PutObjReader::from_vec(b"v1".to_vec()); ecstore .put_object( bucket_name.as_str(), object_name, &mut reader, &ObjectOptions { versioned: true, ..Default::default() }, ) .await .expect("failed to upload v1"); with_forced_immediate_enqueue_timeout(|| async { let mut reader = PutObjReader::from_vec(b"v2".to_vec()); ecstore .put_object( bucket_name.as_str(), object_name, &mut reader, &ObjectOptions { versioned: true, ..Default::default() }, ) .await .expect("failed to upload v2"); }) .await; let info = wait_for_transition(&ecstore, bucket_name.as_str(), object_name, TRANSITION_WAIT_TIMEOUT) .await .expect("current version should transition after compensation backfill"); assert_eq!(info.transitioned_object.status, "complete"); assert_eq!(info.transitioned_object.tier, tier_name); assert!(backend.contains(&info.transitioned_object.name).await); scan_object_with_lifecycle(&disk_paths[0], bucket_name.as_str(), object_name).await; assert!( wait_for_version_count(&ecstore, bucket_name.as_str(), object_name, 1, Duration::from_secs(3)).await, "noncurrent expiry should still remove the previous version after compensation transition" ); } #[tokio::test(flavor = "multi_thread", worker_threads = 1)] #[serial] #[ignore = "FAILING on main: excluded from the serial ILM lane pending a fix, see rustfs/backlog#1148 (ilm-1 partial)"] async fn test_noncurrent_transition_still_works_after_immediate_compensation_transition() { let (disk_paths, ecstore) = setup_isolated_test_env(true).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-noncurrent-transition-comp-{}", &Uuid::new_v4().simple().to_string()[..8]); let object_name = "test/object.txt"; create_test_lock_bucket(&ecstore, bucket_name.as_str()).await; let lifecycle_xml = format!( r#" test-rule Enabled test/ 0 {tier_name} 0 {tier_name} "# ); update_bucket_metadata(bucket_name.as_str(), BUCKET_LIFECYCLE_CONFIG, lifecycle_xml.into_bytes()) .await .expect("Failed to set lifecycle configuration"); let mut reader = PutObjReader::from_vec(b"v1".to_vec()); ecstore .put_object( bucket_name.as_str(), object_name, &mut reader, &ObjectOptions { versioned: true, ..Default::default() }, ) .await .expect("failed to upload v1"); with_forced_immediate_enqueue_timeout(|| async { let mut reader = PutObjReader::from_vec(b"v2".to_vec()); ecstore .put_object( bucket_name.as_str(), object_name, &mut reader, &ObjectOptions { versioned: true, ..Default::default() }, ) .await .expect("failed to upload v2"); }) .await; let info = wait_for_transition(&ecstore, bucket_name.as_str(), object_name, TRANSITION_WAIT_TIMEOUT) .await .expect("current version should transition after compensation backfill"); assert_eq!(info.transitioned_object.status, "complete"); assert_eq!(info.transitioned_object.tier, tier_name); scan_object_with_lifecycle(&disk_paths[0], bucket_name.as_str(), object_name).await; assert!( backend.wait_for_object_count(2, TRANSITION_WAIT_TIMEOUT).await, "noncurrent transition should still move the previous version into the remote tier" ); } #[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-1)"] async fn test_modeled_versioned_delete_creates_delete_marker_after_immediate_compensation_transition() { let (_disk_paths, ecstore) = setup_isolated_test_env(true).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-modeled-versioned-delete-{}", &Uuid::new_v4().simple().to_string()[..8]); let object_name = "test/object.txt"; let payload = b"modeled versioned delete should create delete marker after compensation"; create_test_lock_bucket(&ecstore, bucket_name.as_str()).await; set_bucket_lifecycle_transition_with_tier(bucket_name.as_str(), &tier_name) .await .expect("Failed to set transition lifecycle configuration"); with_forced_immediate_enqueue_timeout(|| async { upload_test_object(&ecstore, bucket_name.as_str(), object_name, payload).await; }) .await; let transitioned = wait_for_transition(&ecstore, bucket_name.as_str(), object_name, TRANSITION_WAIT_TIMEOUT) .await .expect("current version should transition after compensation backfill"); let remote_object = transitioned.transitioned_object.name.clone(); assert!(backend.contains(&remote_object).await); ecstore .delete_object( bucket_name.as_str(), object_name, modeled_versioned_delete_opts(bucket_name.as_str(), object_name).await, ) .await .expect("modeled versioned delete should succeed"); assert!( object_is_delete_marker(&ecstore, bucket_name.as_str(), object_name).await, "versioned delete modeled with versioned flags should create a delete marker" ); assert!( backend.contains(&remote_object).await, "creating a delete marker should not remove the transitioned remote object version" ); } #[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-1)"] async fn test_modeled_delete_marker_cleanup_after_immediate_compensation_transition() { let (disk_paths, ecstore) = setup_isolated_test_env(true).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-modeled-del-marker-cleanup-{}", &Uuid::new_v4().simple().to_string()[..8]); let object_name = "test/object.txt"; let payload = b"modeled delete-marker cleanup should converge after compensation transition"; create_test_lock_bucket(&ecstore, bucket_name.as_str()).await; set_bucket_lifecycle_transition_with_tier(bucket_name.as_str(), &tier_name) .await .expect("Failed to set transition lifecycle configuration"); with_forced_immediate_enqueue_timeout(|| async { upload_test_object(&ecstore, bucket_name.as_str(), object_name, payload).await; }) .await; let transitioned = wait_for_transition(&ecstore, bucket_name.as_str(), object_name, TRANSITION_WAIT_TIMEOUT) .await .expect("current version should transition after compensation backfill"); let remote_object = transitioned.transitioned_object.name.clone(); assert!(backend.contains(&remote_object).await); ecstore .delete_object( bucket_name.as_str(), object_name, modeled_versioned_delete_opts(bucket_name.as_str(), object_name).await, ) .await .expect("modeled versioned delete should succeed"); assert!( object_is_delete_marker(&ecstore, bucket_name.as_str(), object_name).await, "modeled versioned delete should create delete marker before cleanup" ); assert!( backend.contains(&remote_object).await, "delete marker creation should not remove transitioned remote object" ); set_bucket_lifecycle_delmarker_expiration(bucket_name.as_str(), 1) .await .expect("Failed to set delete marker expiration lifecycle configuration"); scan_object_with_lifecycle(&disk_paths[0], bucket_name.as_str(), object_name).await; assert!( object_is_delete_marker(&ecstore, bucket_name.as_str(), object_name).await, "delete marker should remain before DelMarkerExpiration due time" ); assert!( backend.contains(&remote_object).await, "pre-due delete marker lifecycle scan should not remove transitioned remote object" ); set_bucket_lifecycle_deletemarker(bucket_name.as_str()) .await .expect("Failed to set expired object delete marker lifecycle configuration"); scan_object_with_lifecycle(&disk_paths[0], bucket_name.as_str(), object_name).await; assert!( wait_for_object_absence(&ecstore, bucket_name.as_str(), object_name, Duration::from_secs(5)).await, "expired object delete marker lifecycle should eventually clean up the delete marker" ); assert!( backend.contains(&remote_object).await, "delete marker lifecycle cleanup should not remove transitioned remote object" ); } #[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-1)"] async fn test_scanner_expires_zero_day_current_version() { let (disk_paths, ecstore) = setup_isolated_test_env(false).await; let bucket_name = format!("test-zero-day-expire-{}", &Uuid::new_v4().simple().to_string()[..8]); let object_name = "test/object.txt"; create_test_bucket(&ecstore, bucket_name.as_str()).await; upload_test_object(&ecstore, bucket_name.as_str(), object_name, b"expire immediately").await; set_bucket_lifecycle(bucket_name.as_str()) .await .expect("Failed to set lifecycle configuration"); assert!(object_exists(&ecstore, bucket_name.as_str(), object_name).await); init_background_expiry(ecstore.clone()).await; scan_object_with_lifecycle(&disk_paths[0], bucket_name.as_str(), object_name).await; assert!( wait_for_object_absence(&ecstore, bucket_name.as_str(), object_name, Duration::from_secs(3)).await, "scanner should delete zero-day current version after enqueueing expiry" ); } #[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-1)"] async fn test_put_object_immediately_enqueues_zero_day_current_expiry() { let (_disk_paths, ecstore) = setup_isolated_test_env(true).await; let bucket_name = format!("test-put-zero-day-expire-{}", &Uuid::new_v4().simple().to_string()[..8]); let object_name = "expire-now.txt"; create_test_bucket(&ecstore, bucket_name.as_str()).await; let lifecycle_xml = format!( r#" test-rule Enabled {object_name} 0 "# ); update_bucket_metadata(bucket_name.as_str(), BUCKET_LIFECYCLE_CONFIG, lifecycle_xml.into_bytes()) .await .expect("Failed to set lifecycle configuration"); upload_test_object(&ecstore, bucket_name.as_str(), object_name, b"expire immediately").await; assert!( wait_for_object_absence(&ecstore, bucket_name.as_str(), object_name, Duration::from_secs(2)).await, "put_object should enqueue zero-day current expiry without waiting for scanner" ); } #[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-1)"] async fn test_scanner_expires_zero_day_noncurrent_version() { let (disk_paths, ecstore) = setup_isolated_test_env(false).await; let bucket_name = format!("test-zero-day-noncurrent-{}", &Uuid::new_v4().simple().to_string()[..8]); let object_name = "test/object.txt"; create_test_lock_bucket(&ecstore, bucket_name.as_str()).await; let mut reader = PutObjReader::from_vec(b"v1".to_vec()); ecstore .put_object( bucket_name.as_str(), object_name, &mut reader, &ObjectOptions { versioned: true, ..Default::default() }, ) .await .expect("failed to upload v1"); let mut reader = PutObjReader::from_vec(b"v2".to_vec()); ecstore .put_object( bucket_name.as_str(), object_name, &mut reader, &ObjectOptions { versioned: true, ..Default::default() }, ) .await .expect("failed to upload v2"); assert_eq!(object_version_count(&ecstore, bucket_name.as_str(), object_name).await, 2); let lifecycle_xml = r#" test-rule Enabled test/ 0 "#; update_bucket_metadata(bucket_name.as_str(), BUCKET_LIFECYCLE_CONFIG, lifecycle_xml.as_bytes().to_vec()) .await .expect("Failed to set noncurrent lifecycle configuration"); init_background_expiry(ecstore.clone()).await; scan_object_with_lifecycle(&disk_paths[0], bucket_name.as_str(), object_name).await; assert!( wait_for_version_count(&ecstore, bucket_name.as_str(), object_name, 1, Duration::from_secs(3)).await, "scanner should delete zero-day noncurrent versions after enqueueing expiry" ); } #[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-1)"] async fn test_put_object_immediately_enqueues_zero_day_noncurrent_expiry() { let (_disk_paths, ecstore) = setup_isolated_test_env(true).await; let bucket_name = format!("test-put-zero-day-noncurrent-{}", &Uuid::new_v4().simple().to_string()[..8]); let object_name = "test/object.txt"; create_test_lock_bucket(&ecstore, bucket_name.as_str()).await; let lifecycle_xml = r#" test-rule Enabled test/ 0 "#; update_bucket_metadata(bucket_name.as_str(), BUCKET_LIFECYCLE_CONFIG, lifecycle_xml.as_bytes().to_vec()) .await .expect("Failed to set noncurrent lifecycle configuration"); let mut reader = PutObjReader::from_vec(b"v1".to_vec()); ecstore .put_object( bucket_name.as_str(), object_name, &mut reader, &ObjectOptions { versioned: true, ..Default::default() }, ) .await .expect("failed to upload v1"); let mut reader = PutObjReader::from_vec(b"v2".to_vec()); ecstore .put_object( bucket_name.as_str(), object_name, &mut reader, &ObjectOptions { versioned: true, ..Default::default() }, ) .await .expect("failed to upload v2"); assert!( wait_for_version_count(&ecstore, bucket_name.as_str(), object_name, 1, Duration::from_secs(2)).await, "put_object should enqueue zero-day noncurrent expiry without waiting for scanner" ); } #[tokio::test(flavor = "multi_thread", worker_threads = 1)] #[serial] async fn test_background_scanner_expires_zero_day_current_version() { let (_disk_paths, ecstore) = setup_isolated_test_env(true).await; let bucket_name = format!("test-bg-zero-day-expire-{}", &Uuid::new_v4().simple().to_string()[..8]); let object_name = "test/object.txt"; create_test_bucket(&ecstore, bucket_name.as_str()).await; set_bucket_lifecycle(bucket_name.as_str()) .await .expect("Failed to set lifecycle configuration"); upload_test_object(&ecstore, bucket_name.as_str(), object_name, b"expire immediately").await; let ctx = CancellationToken::new(); init_data_scanner(ctx.clone(), ecstore.clone()).await; let deleted = wait_for_object_absence(&ecstore, bucket_name.as_str(), object_name, Duration::from_secs(12)).await; ctx.cancel(); assert!(deleted, "background scanner should delete zero-day current version after startup delay"); } #[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-1)"] async fn test_background_scanner_expires_zero_day_current_version_for_exact_key_prefix() { let (_disk_paths, ecstore) = setup_isolated_test_env(true).await; let bucket_name = format!("test-bg-zero-day-exact-{}", &Uuid::new_v4().simple().to_string()[..8]); let object_name = "expire-now.txt"; create_test_bucket(&ecstore, bucket_name.as_str()).await; let lifecycle_xml = format!( r#" test-rule Enabled {object_name} 0 "# ); update_bucket_metadata(bucket_name.as_str(), BUCKET_LIFECYCLE_CONFIG, lifecycle_xml.into_bytes()) .await .expect("Failed to set lifecycle configuration"); upload_test_object(&ecstore, bucket_name.as_str(), object_name, b"expire immediately").await; let ctx = CancellationToken::new(); init_data_scanner(ctx.clone(), ecstore.clone()).await; let deleted = wait_for_object_absence(&ecstore, bucket_name.as_str(), object_name, Duration::from_secs(12)).await; ctx.cancel(); 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]); // Must live under the `test/` prefix: `set_bucket_lifecycle_transition_with_tier` // installs a transition rule filtered on `test/`, so any other key never // transitions and `wait_for_transition` below times out. let object_name = "test/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() }; let bucket_incarnation_id = ecstore .bucket_incarnation_id(bucket_name.as_str()) .await .expect("read bucket incarnation"); expire_transitioned_object(ecstore.clone(), &restored, &lc_event, &LcEventSrc::Scanner, bucket_incarnation_id) .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]); // Must live under the `test/` prefix (see the transition rule filter in // `set_bucket_lifecycle_transition_with_tier`) or the object never // transitions and `wait_for_transition` below times out. let object_name = "test/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"); } }