From a118d7e4fd07be9aa2ccc705f463c639308981a8 Mon Sep 17 00:00:00 2001 From: houseme Date: Sun, 16 Aug 2026 14:38:51 +0800 Subject: [PATCH] perf(ecstore): enable inline data read early-stop by default (#6140) * perf(ecstore): enable inline data read early-stop by default Co-Authored-By: heihutu * test(scanner): box large ILM transition flow future Co-Authored-By: heihutu * test(ecstore): align internal meta early-stop miss Co-Authored-By: heihutu --------- Co-authored-by: heihutu --- .../src/set_disk/core/io_primitives.rs | 4 +- crates/ecstore/src/set_disk/mod.rs | 43 +- crates/ecstore/src/set_disk/ops/object.rs | 22 +- crates/ecstore/src/set_disk/read.rs | 36 +- .../tests/lifecycle_integration_test.rs | 563 +++++++++--------- 5 files changed, 349 insertions(+), 319 deletions(-) diff --git a/crates/ecstore/src/set_disk/core/io_primitives.rs b/crates/ecstore/src/set_disk/core/io_primitives.rs index a251644ea..750c9d0ba 100644 --- a/crates/ecstore/src/set_disk/core/io_primitives.rs +++ b/crates/ecstore/src/set_disk/core/io_primitives.rs @@ -7141,7 +7141,7 @@ mod tests { } #[tokio::test] - async fn bounded_metadata_early_stop_defaults_keep_data_get_full_fanout() { + async fn bounded_metadata_early_stop_defaults_keep_non_inline_data_get_full_fanout() { const DISKS: usize = 4; let bucket = "bounded-data-get-default-bucket"; let object = "bounded-data-get-default-object"; @@ -7164,7 +7164,7 @@ mod tests { assert_eq!( calls.total(disk_call_counters::KIND_READ_VERSION), DISKS as u64, - "default GET data-read metadata must keep full fanout for read-failure tolerance" + "default non-inline GET data-read metadata must keep full fanout for read-failure tolerance" ); assert_eq!(diagnostics.total_responses(), DISKS); assert_eq!(parts_metadata.iter().filter(|fi| fi.name == object).count(), DISKS); diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index 7f3a37ea0..5ac05f207 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/crates/ecstore/src/set_disk/mod.rs @@ -692,8 +692,9 @@ const DEFAULT_RUSTFS_GET_SMALL_OBJECT_DIRECT_MEMORY_THRESHOLD: usize = 128 * 102 const ENV_RUSTFS_GET_METADATA_EARLY_STOP_ENABLE: &str = "RUSTFS_GET_METADATA_EARLY_STOP_ENABLE"; // Enabled by default (backlog#872): the early-stop path only engages for // requests `should_allow_metadata_early_stop` classifies as safe (latest-version -// metadata-only reads by default, without version_id / healing / free-version -// needs) and still requires a full read-quorum agreement before stopping. Set +// reads by default, without version_id / healing / free-version needs) and still +// requires a full read-quorum agreement before stopping. Data-read requests add +// a separate inline-shard verifier before cancelling the remaining fanout. Set // the env var to `false` to fall back to full-wait metadata fanout. const DEFAULT_RUSTFS_GET_METADATA_EARLY_STOP_ENABLE: bool = true; @@ -704,7 +705,7 @@ const ENV_RUSTFS_GET_METADATA_VERSION_EARLY_STOP_ENABLE: &str = "RUSTFS_GET_META const DEFAULT_RUSTFS_GET_METADATA_VERSION_EARLY_STOP_ENABLE: bool = false; const ENV_RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE: &str = "RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE"; -const DEFAULT_RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE: bool = false; +const DEFAULT_RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE: bool = true; const ENV_RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT: &str = "RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT"; const DEFAULT_RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT: bool = false; @@ -915,12 +916,6 @@ mod prepared_get_object_metadata_tests { .expect("4-disk test geometry should leave one bounded spare disk") } - fn bounded_slow_initial_disk_index(bucket: &str, object: &str) -> usize { - *bounded_metadata_fanout_order(bucket, object, 4, 2) - .get(2) - .expect("4-disk test geometry should include a third initial metadata disk") - } - #[tokio::test] async fn prepared_metadata_is_consumed_exactly_once() { let snapshot = GetObjectFileInfo::owned(FileInfo::default(), Vec::new(), Vec::new()); @@ -1039,7 +1034,7 @@ mod prepared_get_object_metadata_tests { #[test] #[serial_test::serial(body_cache_hook)] - fn inline_data_read_early_stop_reader_returns_exact_body() { + fn inline_data_read_early_stop_defaults_return_exact_body() { let runtime = tokio::runtime::Builder::new_current_thread() .enable_all() .build() @@ -1071,14 +1066,14 @@ mod prepared_get_object_metadata_tests { temp_env::async_with_vars( [ - ("RUSTFS_GET_METADATA_EARLY_STOP_ENABLE", Some("true")), - ("RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE", Some("true")), - ("RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT", Some("true")), + ("RUSTFS_GET_METADATA_EARLY_STOP_ENABLE", None::<&str>), + ("RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE", None::<&str>), + ("RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT", None::<&str>), ], async { - let slow_initial_disk = bounded_slow_initial_disk_index(bucket, &object); + let slow_parity_disk = bounded_spare_disk_index(bucket, &object); let barrier = - rename_fanout_barrier::arm(&object, slow_initial_disk, rename_fanout_barrier::PHASE_READ_VERSION); + rename_fanout_barrier::arm(&object, slow_parity_disk, rename_fanout_barrier::PHASE_READ_VERSION); let calls = disk_call_counters::observe(&object); let set_disks_for_read = Arc::clone(&set_disks); let opts_for_read = opts.clone(); @@ -1091,10 +1086,10 @@ mod prepared_get_object_metadata_tests { tokio::time::timeout(READ_VERSION_BARRIER_GUARD, barrier.wait_until_paused()) .await - .expect("bounded inline GET should pause a slow initial metadata read"); + .expect("default inline GET should pause a slow parity metadata read"); let mut reader = tokio::time::timeout(READ_VERSION_BARRIER_GUARD, &mut open_reader) .await - .expect("production inline GET should return before the paused metadata response") + .expect("default production inline GET should return before the paused parity metadata response") .expect("inline GET reader task should not panic") .expect("inline GET reader should open"); let object_size = reader.object_info.size; @@ -1115,14 +1110,14 @@ mod prepared_get_object_metadata_tests { assert_eq!(object_size, payload.len() as i64); assert_eq!(restored, payload); - assert_eq!(calls_total, 4, "bounded production GET should schedule the initial quorum plus one spare"); + assert_eq!(calls_total, 4, "default production GET should eagerly schedule the full metadata fanout"); assert_eq!( recorder.histogram_values( "rustfs_io_get_object_metadata_fanout_scheduled", &[("path", GET_OBJECT_PATH_LEGACY_DUPLEX)] ), vec![4.0], - "bounded production GET should record all scheduled metadata tasks" + "default production GET should record all scheduled metadata tasks" ); assert_eq!( recorder.histogram_values( @@ -1130,7 +1125,7 @@ mod prepared_get_object_metadata_tests { &[("path", GET_OBJECT_PATH_LEGACY_DUPLEX)] ), vec![3.0], - "bounded production GET should record only observed metadata responses as completed" + "default production GET should record only observed metadata responses as completed" ); assert_eq!( recorder.histogram_values( @@ -1138,7 +1133,7 @@ mod prepared_get_object_metadata_tests { &[("path", GET_OBJECT_PATH_LEGACY_DUPLEX)] ), vec![1.0], - "bounded production GET should record the aborted slow metadata task" + "default production GET should record the aborted slow parity metadata task" ); } @@ -1285,9 +1280,9 @@ mod prepared_get_object_metadata_tests { temp_env::async_with_vars( [ - ("RUSTFS_GET_METADATA_EARLY_STOP_ENABLE", Some("true")), - ("RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE", Some("true")), - ("RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT", Some("true")), + ("RUSTFS_GET_METADATA_EARLY_STOP_ENABLE", None::<&str>), + ("RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE", None::<&str>), + ("RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT", None::<&str>), ], async { let calls = disk_call_counters::observe(&object); diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index 1c64de30d..3a2949140 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/crates/ecstore/src/set_disk/ops/object.rs @@ -7505,7 +7505,7 @@ mod get_object_downstream_close_accounting_tests { use super::hermetic_set_disks_support::hermetic_set_disks; use super::*; use crate::diagnostics::get::{ - GET_METADATA_EARLY_STOP_REASON_UNSAFE_REQUEST, GET_OBJECT_PATH_INTERNAL_META, GET_STAGE_DECODE, GET_STAGE_EMIT, + GET_METADATA_EARLY_STOP_REASON_NOT_FOUND, GET_OBJECT_PATH_INTERNAL_META, GET_STAGE_DECODE, GET_STAGE_EMIT, GetObjectFailureReason, }; use crate::disk::RUSTFS_META_BUCKET; @@ -7637,8 +7637,8 @@ mod get_object_downstream_close_accounting_tests { legacy_completed, internal_cancelled, legacy_cancelled, - internal_unsafe_miss, - legacy_unsafe_miss, + internal_not_found_miss, + legacy_not_found_miss, internal_saved, legacy_saved, ) = metrics::with_local_recorder(&recorder, || { @@ -7714,7 +7714,7 @@ mod get_object_downstream_close_accounting_tests { &[ ("path", GET_OBJECT_PATH_INTERNAL_META), ("decision", "miss"), - ("reason", GET_METADATA_EARLY_STOP_REASON_UNSAFE_REQUEST), + ("reason", GET_METADATA_EARLY_STOP_REASON_NOT_FOUND), ], ), recorder.counter_value( @@ -7722,7 +7722,7 @@ mod get_object_downstream_close_accounting_tests { &[ ("path", GET_OBJECT_PATH_LEGACY_DUPLEX), ("decision", "miss"), - ("reason", GET_METADATA_EARLY_STOP_REASON_UNSAFE_REQUEST), + ("reason", GET_METADATA_EARLY_STOP_REASON_NOT_FOUND), ], ), recorder.histogram_values( @@ -7773,21 +7773,21 @@ mod get_object_downstream_close_accounting_tests { "internal metadata lifecycle cancelled count must not leak into legacy_duplex" ); assert_eq!( - internal_unsafe_miss, 1, - "internal metadata unsafe early-stop miss must retain its path label" + internal_not_found_miss, 1, + "internal metadata not-found early-stop miss must retain its path label" ); assert_eq!( - legacy_unsafe_miss, 0, - "internal metadata unsafe early-stop miss must not leak into legacy_duplex" + legacy_not_found_miss, 0, + "internal metadata not-found early-stop miss must not leak into legacy_duplex" ); assert_eq!( internal_saved, vec![0.0], - "internal metadata unsafe miss must record zero saved responses on internal_meta" + "internal metadata not-found miss must record zero saved responses on internal_meta" ); assert!( legacy_saved.is_empty(), - "internal metadata unsafe miss saved responses must not leak into legacy_duplex" + "internal metadata not-found miss saved responses must not leak into legacy_duplex" ); } } diff --git a/crates/ecstore/src/set_disk/read.rs b/crates/ecstore/src/set_disk/read.rs index d0011cd11..4a3de7e93 100644 --- a/crates/ecstore/src/set_disk/read.rs +++ b/crates/ecstore/src/set_disk/read.rs @@ -3886,13 +3886,15 @@ mod tests { assert!(metadata_early_stop_permitted(true, true, false, "", false, false)); // observe=false (non-observed fanout) also disables early-stop. assert!(!metadata_early_stop_permitted(true, false, false, "", false, false)); - assert!(!metadata_early_stop_permitted(true, true, true, "", false, false)); + // Whole/latest data-read metadata is now allowed by default; + // the inline verifier still decides whether it can stop early. + assert!(metadata_early_stop_permitted(true, true, true, "", false, false)); }, ); } #[test] - fn metadata_early_stop_keeps_data_reads_opt_in_by_default() { + fn metadata_early_stop_allows_safe_data_reads_by_default() { temp_env::with_vars( [ (ENV_RUSTFS_GET_METADATA_EARLY_STOP_ENABLE, Some("true")), @@ -3900,7 +3902,7 @@ mod tests { (ENV_RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE, None), ], || { - assert!(!should_allow_metadata_early_stop(true, "", false, false)); + assert!(should_allow_metadata_early_stop(true, "", false, false)); assert!(!should_allow_metadata_early_stop(true, "version-id", false, false)); assert!(should_allow_metadata_early_stop(false, "", false, false)); assert!(!should_allow_metadata_early_stop(false, "version-id", false, false)); @@ -3932,6 +3934,34 @@ mod tests { ); } + #[test] + fn metadata_early_stop_bounded_fanout_defaults_to_disabled() { + temp_env::with_vars( + [ + (ENV_RUSTFS_GET_METADATA_EARLY_STOP_ENABLE, Some("true")), + (ENV_RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE, None), + (ENV_RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT, None), + ], + || { + assert!(is_get_metadata_data_read_early_stop_enabled()); + assert!(!is_get_metadata_early_stop_bounded_fanout_enabled()); + }, + ); + temp_env::with_vars([(ENV_RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT, Some("true"))], || { + assert!(is_get_metadata_early_stop_bounded_fanout_enabled()); + }); + temp_env::with_vars( + [ + (ENV_RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE, Some("false")), + (ENV_RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT, Some("false")), + ], + || { + assert!(!is_get_metadata_data_read_early_stop_enabled()); + assert!(!is_get_metadata_early_stop_bounded_fanout_enabled()); + }, + ); + } + #[test] fn metadata_early_stop_rejects_healing_and_free_version_requests() { temp_env::with_vars( diff --git a/crates/scanner/tests/lifecycle_integration_test.rs b/crates/scanner/tests/lifecycle_integration_test.rs index 0a43a8060..7b2130316 100644 --- a/crates/scanner/tests/lifecycle_integration_test.rs +++ b/crates/scanner/tests/lifecycle_integration_test.rs @@ -1069,309 +1069,314 @@ mod serial_tests { #[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; + async move { + 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 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!"; + 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) + create_test_bucket(&ecstore, put_bucket.as_str()).await; + set_bucket_lifecycle_transition_with_tier(put_bucket.as_str(), &tier_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" - ); - } + .expect("Failed to set lifecycle configuration"); - // 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 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"); - let multipart_bucket = format!("test-immediate-mpu-{}", &Uuid::new_v4().simple().to_string()[..8]); - let multipart_object = "test/multipart.txt"; + enqueue_transition_for_existing_objects(ecstore.clone(), put_bucket.as_str()) + .await + .expect("Failed to enqueue transitioned put object"); - 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 put_info = wait_for_transition(&ecstore, put_bucket.as_str(), put_object, TRANSITION_WAIT_TIMEOUT) + .await + .expect("object should transition after enqueueing existing objects"); - let upload = ecstore - .new_multipart_upload(multipart_bucket.as_str(), multipart_object, &ObjectOptions::default()) - .await - .expect("Failed to create multipart upload"); + 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" + ); + } - 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"); + // 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); - 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"); + let multipart_bucket = format!("test-immediate-mpu-{}", &Uuid::new_v4().simple().to_string()[..8]); + let multipart_object = "test/multipart.txt"; - enqueue_transition_for_existing_objects(ecstore.clone(), multipart_bucket.as_str()) - .await - .expect("Failed to enqueue transitioned multipart object"); + 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 multipart_info = wait_for_transition(&ecstore, multipart_bucket.as_str(), multipart_object, TRANSITION_WAIT_TIMEOUT) - .await - .expect("object should transition after enqueueing existing objects"); + let upload = ecstore + .new_multipart_upload(multipart_bucket.as_str(), multipart_object, &ObjectOptions::default()) + .await + .expect("Failed to create multipart upload"); - 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 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"); - 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 { + ecstore + .clone() + .complete_multipart_upload( + multipart_bucket.as_str(), + multipart_object, + &upload.upload_id, + vec![CompletePart { part_num: 1, - etag: uploaded_part1.etag.clone(), + etag: part.etag.clone(), ..Default::default() - }, - CompletePart { - part_num: 2, - etag: uploaded_part2.etag.clone(), - ..Default::default() - }, - ], - &ObjectOptions::default(), - ) - .await - .expect("Failed to complete multipart upload"); + }], + &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"); + enqueue_transition_for_existing_objects(ecstore.clone(), multipart_bucket.as_str()) + .await + .expect("Failed to enqueue transitioned multipart 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); + 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"); - 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, + 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() }, - ..Default::default() - }, - ) - .await - .expect("Failed to restore transitioned multipart object"); + ) + .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 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); + 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); + } + .boxed_local() + .await; } #[tokio::test(flavor = "multi_thread", worker_threads = 1)]