From 236142a6823358840261d9101f95d82d386634cb Mon Sep 17 00:00:00 2001 From: cxymds Date: Mon, 23 Mar 2026 12:29:13 +0800 Subject: [PATCH] fix(ecstore): repair lifecycle transition and restore flows (#2240) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: heihutu Co-authored-by: 安正超 Co-authored-by: houseme Co-authored-by: loverustfs Co-authored-by: weisd --- .../bucket/lifecycle/bucket_lifecycle_ops.rs | 56 ++- .../ecstore/src/bucket/lifecycle/lifecycle.rs | 41 +- .../replication/replication_resyncer.rs | 6 +- crates/ecstore/src/client/api_get_options.rs | 77 ++- crates/ecstore/src/client/transition_api.rs | 2 +- crates/ecstore/src/endpoints.rs | 85 ++-- crates/ecstore/src/set_disk.rs | 59 ++- crates/ecstore/src/store.rs | 2 +- crates/ecstore/src/store/bucket.rs | 1 + crates/ecstore/src/store/rebalance.rs | 1 + .../tests/lifecycle_integration_test.rs | 427 ++++++++++++++++- rustfs/src/app/bucket_usecase.rs | 84 +++- .../src/app/lifecycle_transition_api_test.rs | 439 ++++++++++++++++++ rustfs/src/app/mod.rs | 3 + rustfs/src/app/multipart_usecase.rs | 7 + rustfs/src/app/object_usecase.rs | 13 +- 16 files changed, 1209 insertions(+), 94 deletions(-) create mode 100644 rustfs/src/app/lifecycle_transition_api_test.rs diff --git a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs index d70742a07..21483e742 100644 --- a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs +++ b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs @@ -33,7 +33,9 @@ use crate::global::GLOBAL_LocalNodeName; use crate::global::{GLOBAL_LifecycleSys, GLOBAL_TierConfigMgr, get_global_deployment_id}; use crate::store::ECStore; use crate::store_api::StorageAPI; -use crate::store_api::{GetObjectReader, HTTPRangeSpec, ObjectInfo, ObjectOperations, ObjectOptions, ObjectToDelete}; +use crate::store_api::{ + GetObjectReader, HTTPRangeSpec, ListOperations, ObjectInfo, ObjectOperations, ObjectOptions, ObjectToDelete, +}; use crate::tier::warm_backend::WarmBackendGetOpts; use async_channel::{Receiver as A_Receiver, Sender as A_Sender, bounded}; use bytes::BytesMut; @@ -646,18 +648,48 @@ pub async fn validate_transition_tier(lc: &BucketLifecycleConfiguration) -> Resu } pub async fn enqueue_transition_immediate(oi: &ObjectInfo, src: LcEventSrc) { - let lc = GLOBAL_LifecycleSys.get(&oi.bucket).await; - if !lc.is_none() { - let event = lc.expect("err").eval(&oi.to_lifecycle_opts()).await; - match event.action { - IlmAction::TransitionAction | IlmAction::TransitionVersionAction => { - if oi.delete_marker || oi.is_dir { - return; - } - GLOBAL_TransitionState.queue_transition_task(oi, &event, &src).await; - } - _ => (), + if let Some(lc) = GLOBAL_LifecycleSys.get(&oi.bucket).await { + enqueue_transition_with_lifecycle(oi, &lc, &src).await; + } +} + +pub async fn enqueue_transition_for_existing_objects(api: Arc, bucket: &str) -> Result<(), Error> { + let Some(lc) = GLOBAL_LifecycleSys.get(bucket).await else { + return Ok(()); + }; + let mut marker = None; + let mut version_marker = None; + let src = LcEventSrc::Scanner; + + loop { + let page = api + .clone() + .list_object_versions(bucket, "", marker.clone(), version_marker.clone(), None, 1000) + .await?; + + for object in &page.objects { + enqueue_transition_with_lifecycle(object, &lc, &src).await; } + + if !page.is_truncated { + return Ok(()); + } + + marker = page.next_marker; + version_marker = page.next_version_idmarker; + } +} + +async fn enqueue_transition_with_lifecycle(oi: &ObjectInfo, lc: &BucketLifecycleConfiguration, src: &LcEventSrc) { + let event = lc.eval(&oi.to_lifecycle_opts()).await; + match event.action { + IlmAction::TransitionAction | IlmAction::TransitionVersionAction => { + if oi.delete_marker || oi.is_dir { + return; + } + GLOBAL_TransitionState.queue_transition_task(oi, &event, src).await; + } + _ => (), } } diff --git a/crates/ecstore/src/bucket/lifecycle/lifecycle.rs b/crates/ecstore/src/bucket/lifecycle/lifecycle.rs index 0cbe7f6e7..55787b684 100644 --- a/crates/ecstore/src/bucket/lifecycle/lifecycle.rs +++ b/crates/ecstore/src/bucket/lifecycle/lifecycle.rs @@ -768,7 +768,7 @@ impl LifecycleCalculate for NoncurrentVersionTransition { #[async_trait::async_trait] impl LifecycleCalculate for Transition { fn next_due(&self, obj: &ObjectOpts) -> Option { - if !obj.is_latest || self.days.is_none() { + if !obj.is_latest { return None; } @@ -1296,6 +1296,45 @@ mod tests { assert_eq!(event.storage_class, "COLDTIER44"); } + #[tokio::test] + #[serial] + async fn eval_inner_transitions_latest_object_after_date_due() { + let base_time = OffsetDateTime::from_unix_timestamp(1_000_000).unwrap(); + let transition_date = base_time - Duration::days(1); + let lc = BucketLifecycleConfiguration { + expiry_updated_at: None, + rules: vec![LifecycleRule { + status: ExpirationStatus::from_static(ExpirationStatus::ENABLED), + expiration: None, + abort_incomplete_multipart_upload: None, + del_marker_expiration: None, + filter: None, + id: Some("transition-date".to_string()), + noncurrent_version_expiration: None, + noncurrent_version_transitions: None, + prefix: None, + transitions: Some(vec![Transition { + days: None, + date: Some(transition_date.into()), + storage_class: Some(TransitionStorageClass::from_static("WARM")), + }]), + }], + }; + + let opts = ObjectOpts { + name: "obj".to_string(), + mod_time: Some(base_time), + is_latest: true, + transition_status: "".to_string(), + ..Default::default() + }; + let event = lc.eval_inner(&opts, base_time + Duration::days(1), 0).await; + + assert_eq!(event.action, IlmAction::TransitionAction); + assert_eq!(event.rule_id, "transition-date"); + assert_eq!(event.storage_class, "WARM"); + } + #[tokio::test] #[serial] async fn eval_inner_expires_noncurrent_version_after_due() { diff --git a/crates/ecstore/src/bucket/replication/replication_resyncer.rs b/crates/ecstore/src/bucket/replication/replication_resyncer.rs index d043018a8..32e3ebf63 100644 --- a/crates/ecstore/src/bucket/replication/replication_resyncer.rs +++ b/crates/ecstore/src/bucket/replication/replication_resyncer.rs @@ -2453,7 +2453,7 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { return rinfo; } - let sopts = StatObjectOptions { + let mut sopts = StatObjectOptions { version_id: object_info.version_id.map(|v| v.to_string()).unwrap_or_default(), internal: AdvancedGetOptions { replication_proxy_request: "false".to_string(), @@ -2462,7 +2462,9 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { ..Default::default() }; - sopts.set(AMZ_TAGGING_DIRECTIVE, "ACCESS"); + if let Err(err) = sopts.set(AMZ_TAGGING_DIRECTIVE, "ACCESS") { + warn!("failed to set replication tagging directive header: {err}"); + } match tgt_client .head_object(&tgt_client.bucket, &object, self.version_id.map(|v| v.to_string())) diff --git a/crates/ecstore/src/client/api_get_options.rs b/crates/ecstore/src/client/api_get_options.rs index 3025018b7..622c5a4c2 100644 --- a/crates/ecstore/src/client/api_get_options.rs +++ b/crates/ecstore/src/client/api_get_options.rs @@ -64,20 +64,32 @@ impl GetObjectOptions { pub fn header(&self) -> HeaderMap { let mut headers: HeaderMap = HeaderMap::with_capacity(self.headers.len()); for (k, v) in &self.headers { - if let Ok(header_name) = HeaderName::from_bytes(k.as_bytes()) { - headers.insert(header_name, v.parse().expect("err")); - } else { - warn!("Invalid header name: {}", k); + match (HeaderName::from_bytes(k.as_bytes()), HeaderValue::from_str(v)) { + (Ok(header_name), Ok(header_value)) => { + headers.insert(header_name, header_value); + } + (Err(_), _) => { + warn!("Invalid header name: {}", k); + } + (_, Err(_)) => { + warn!("Invalid header value for {}: {:?}", k, v); + } } } if self.checksum { - headers.insert("x-amz-checksum-mode", "ENABLED".parse().expect("err")); + headers.insert(HeaderName::from_static("x-amz-checksum-mode"), HeaderValue::from_static("ENABLED")); } headers } - pub fn set(&self, key: &str, value: &str) { - //self.headers[http.CanonicalHeaderKey(key)] = value; + pub fn set(&mut self, key: &str, value: &str) -> Result<(), std::io::Error> { + let header_name = HeaderName::from_bytes(key.as_bytes()) + .map_err(|err| std::io::Error::other(err_invalid_argument(&format!("Invalid header name {key}: {err}"))))?; + HeaderValue::from_str(value) + .map_err(|err| std::io::Error::other(err_invalid_argument(&format!("Invalid header value for {key}: {err}"))))?; + + self.headers.insert(header_name.as_str().to_string(), value.to_string()); + Ok(()) } pub fn set_req_param(&mut self, key: &str, value: &str) { @@ -89,12 +101,12 @@ impl GetObjectOptions { } pub fn set_match_etag(&mut self, etag: &str) -> Result<(), std::io::Error> { - self.set("If-Match", &format!("\"{etag}\"")); + self.set("If-Match", &format!("\"{etag}\""))?; Ok(()) } pub fn set_match_etag_except(&mut self, etag: &str) -> Result<(), std::io::Error> { - self.set("If-None-Match", &format!("\"{etag}\"")); + self.set("If-None-Match", &format!("\"{etag}\""))?; Ok(()) } @@ -102,7 +114,7 @@ impl GetObjectOptions { if mod_time.unix_timestamp() == 0 { return Err(std::io::Error::other(err_invalid_argument("Modified since cannot be empty."))); } - self.set("If-Unmodified-Since", &mod_time.to_string()); + self.set("If-Unmodified-Since", &mod_time.to_string())?; Ok(()) } @@ -110,17 +122,17 @@ impl GetObjectOptions { if mod_time.unix_timestamp() == 0 { return Err(std::io::Error::other(err_invalid_argument("Modified since cannot be empty."))); } - self.set("If-Modified-Since", &mod_time.to_string()); + self.set("If-Modified-Since", &mod_time.to_string())?; Ok(()) } pub fn set_range(&mut self, start: i64, end: i64) -> Result<(), std::io::Error> { if start == 0 && end < 0 { - self.set("Range", &format!("bytes={}", end)); + self.set("Range", &format!("bytes={}", end))?; } else if 0 < start && end == 0 { - self.set("Range", &format!("bytes={}-", start)); + self.set("Range", &format!("bytes={}-", start))?; } else if 0 <= start && start <= end { - self.set("Range", &format!("bytes={}-{}", start, end)); + self.set("Range", &format!("bytes={}-{}", start, end))?; } else { return Err(std::io::Error::other(err_invalid_argument(&format!( "Invalid range specified: start={} end={}", @@ -146,3 +158,40 @@ impl GetObjectOptions { url_values } } + +#[cfg(test)] +mod tests { + use super::GetObjectOptions; + + #[test] + fn set_range_populates_range_header() { + let mut opts = GetObjectOptions::default(); + opts.set_range(5, 9).expect("valid range should succeed"); + + let headers = opts.header(); + let range = headers.get("range").expect("range header should be present"); + assert_eq!(range.to_str().expect("range header must be valid ascii"), "bytes=5-9"); + } + + #[test] + fn set_rejects_invalid_header_value() { + let mut opts = GetObjectOptions::default(); + + let err = opts + .set("Range", "bytes=5-\n9") + .expect_err("invalid header value should fail"); + + assert!(err.to_string().contains("Invalid header value")); + assert!(opts.headers.is_empty(), "invalid headers must not be stored"); + } + + #[test] + fn header_skips_invalid_prepopulated_header_value() { + let mut opts = GetObjectOptions::default(); + opts.headers.insert("Range".to_string(), "bytes=5-\n9".to_string()); + + let headers = opts.header(); + + assert!(headers.get("range").is_none(), "invalid stored header values should be ignored"); + } +} diff --git a/crates/ecstore/src/client/transition_api.rs b/crates/ecstore/src/client/transition_api.rs index d23c6fb3a..a733aef26 100644 --- a/crates/ecstore/src/client/transition_api.rs +++ b/crates/ecstore/src/client/transition_api.rs @@ -375,7 +375,7 @@ impl TransitionClient { //debug!("http_resp_body: {}", String::from_utf8(b).unwrap()); //if self.is_trace_enabled && !(self.trace_errors_only && resp.status() == StatusCode::OK) { - if resp.status() != StatusCode::OK { + if !resp.status().is_success() { //self.dump_http(&cloned_req, &resp)?; let mut body_vec = Vec::new(); let mut body = resp.into_body(); diff --git a/crates/ecstore/src/endpoints.rs b/crates/ecstore/src/endpoints.rs index 129618fb3..a24b49d9c 100644 --- a/crates/ecstore/src/endpoints.rs +++ b/crates/ecstore/src/endpoints.rs @@ -791,14 +791,17 @@ mod test { panic!("No non-loop back IP address found for this host"); } let non_loop_back_ip = non_loop_back_i_ps[0]; + let remote_ip1 = "192.0.2.10"; + let remote_ip2 = "192.0.2.11"; + let remote_ip3 = "192.0.2.12"; let case1_endpoint1 = format!("http://{non_loop_back_ip}/d1"); let case1_endpoint2 = format!("http://{non_loop_back_ip}/d2"); let args = vec![ format!("http://{}:10000/d1", non_loop_back_ip), format!("http://{}:10000/d2", non_loop_back_ip), - "http://example.org:10000/d3".to_string(), - "http://example.com:10000/d4".to_string(), + format!("http://{remote_ip1}:10000/d3"), + format!("http://{remote_ip2}:10000/d4"), ]; let (case1_ur_ls, case1_local_flags) = get_expected_endpoints(args, format!("http://{non_loop_back_ip}:10000/")); @@ -807,26 +810,26 @@ mod test { let args = vec![ format!("http://{}:10000/d1", non_loop_back_ip), format!("http://{}:9000/d2", non_loop_back_ip), - "http://example.org:10000/d3".to_string(), - "http://example.com:10000/d4".to_string(), + format!("http://{remote_ip1}:10000/d3"), + format!("http://{remote_ip2}:10000/d4"), ]; let (case2_ur_ls, case2_local_flags) = get_expected_endpoints(args, format!("http://{non_loop_back_ip}:10000/")); let case3_endpoint1 = format!("http://{non_loop_back_ip}/d1"); let args = vec![ format!("http://{}:80/d1", non_loop_back_ip), - "http://example.org:9000/d2".to_string(), - "http://example.com:80/d3".to_string(), - "http://example.net:80/d4".to_string(), + format!("http://{remote_ip1}:9000/d2"), + format!("http://{remote_ip2}:80/d3"), + format!("http://{remote_ip3}:80/d4"), ]; let (case3_ur_ls, case3_local_flags) = get_expected_endpoints(args, format!("http://{non_loop_back_ip}:80/")); let case4_endpoint1 = format!("http://{non_loop_back_ip}/d1"); let args = vec![ format!("http://{}:9000/d1", non_loop_back_ip), - "http://example.org:9000/d2".to_string(), - "http://example.com:9000/d3".to_string(), - "http://example.net:9000/d4".to_string(), + format!("http://{remote_ip1}:9000/d2"), + format!("http://{remote_ip2}:9000/d3"), + format!("http://{remote_ip3}:9000/d4"), ]; let (case4_ur_ls, case4_local_flags) = get_expected_endpoints(args, format!("http://{non_loop_back_ip}:9000/")); @@ -844,8 +847,8 @@ mod test { let case6_endpoint1 = format!("http://{non_loop_back_ip}:9003/d4"); let args = vec![ - "http://localhost:9000/d1".to_string(), - "http://localhost:9001/d2".to_string(), + "http://127.0.0.1:9000/d1".to_string(), + "http://127.0.0.1:9001/d2".to_string(), "http://127.0.0.1:9002/d3".to_string(), case6_endpoint1.clone(), ]; @@ -864,8 +867,8 @@ mod test { // Erasure Single Drive TestCase { num: 2, - server_addr: "localhost:9000", - args: vec!["http://localhost/d1"], + server_addr: "127.0.0.1:9000", + args: vec!["http://127.0.0.1/d1"], expected_err: Some(Error::other("use path style endpoint for single node setup")), ..Default::default() }, @@ -885,7 +888,7 @@ mod test { }, TestCase { num: 4, - server_addr: "localhost:10000", + server_addr: "127.0.0.1:10000", args: vec!["/d1"], expected_endpoints: Some(Endpoints(vec![Endpoint { url: must_file_path("/d1"), @@ -899,12 +902,12 @@ mod test { }, TestCase { num: 5, - server_addr: "localhost:9000", + server_addr: "127.0.0.1:9000", args: vec![ "https://127.0.0.1:9000/d1", - "https://localhost:9001/d1", - "https://example.com/d1", - "https://example.com/d2", + "https://127.0.0.1:9001/d1", + "https://192.0.2.1/d1", + "https://192.0.2.1/d2", ], expected_err: Some(Error::other("same path '/d1' can not be served by different port on same address")), ..Default::default() @@ -952,35 +955,35 @@ mod test { num: 7, server_addr: "0.0.0.0:9000", args: vec![ - "http://localhost/d1", - "http://localhost/d2", - "http://localhost/d3", - "http://localhost/d4", + "http://127.0.0.1/d1", + "http://127.0.0.1/d2", + "http://127.0.0.1/d3", + "http://127.0.0.1/d4", ], expected_endpoints: Some(Endpoints(vec![ Endpoint { - url: must_url("http://localhost:9000/d1"), + url: must_url("http://127.0.0.1:9000/d1"), is_local: true, pool_idx: 0, set_idx: 0, disk_idx: 0, }, Endpoint { - url: must_url("http://localhost:9000/d2"), + url: must_url("http://127.0.0.1:9000/d2"), is_local: true, pool_idx: 0, set_idx: 0, disk_idx: 0, }, Endpoint { - url: must_url("http://localhost:9000/d3"), + url: must_url("http://127.0.0.1:9000/d3"), is_local: true, pool_idx: 0, set_idx: 0, disk_idx: 0, }, Endpoint { - url: must_url("http://localhost:9000/d4"), + url: must_url("http://127.0.0.1:9000/d4"), is_local: true, pool_idx: 0, set_idx: 0, @@ -995,8 +998,8 @@ mod test { num: 8, server_addr: "127.0.0.1:10000", args: vec![ - "http://localhost/d1", - "http://localhost/d2", + "http://[::1]/d1", + "http://[::1]/d2", "http://127.0.0.1/d3", "http://127.0.0.1/d4", ], @@ -1034,8 +1037,8 @@ mod test { args: vec![ case1_endpoint1.as_str(), case1_endpoint2.as_str(), - "http://example.org/d3", - "http://example.com/d4", + "http://192.0.2.10/d3", + "http://192.0.2.11/d4", ], expected_endpoints: Some(Endpoints(vec![ Endpoint { @@ -1076,8 +1079,8 @@ mod test { args: vec![ case2_endpoint1.as_str(), case2_endpoint2.as_str(), - "http://example.org/d3", - "http://example.com/d4", + "http://192.0.2.10/d3", + "http://192.0.2.11/d4", ], expected_endpoints: Some(Endpoints(vec![ Endpoint { @@ -1117,9 +1120,9 @@ mod test { server_addr: "0.0.0.0:80", args: vec![ case3_endpoint1.as_str(), - "http://example.org:9000/d2", - "http://example.com/d3", - "http://example.net/d4", + "http://192.0.2.10:9000/d2", + "http://192.0.2.11/d3", + "http://192.0.2.12/d4", ], expected_endpoints: Some(Endpoints(vec![ Endpoint { @@ -1159,9 +1162,9 @@ mod test { server_addr: "0.0.0.0:9000", args: vec![ case4_endpoint1.as_str(), - "http://example.org/d2", - "http://example.com/d3", - "http://example.net/d4", + "http://192.0.2.10/d2", + "http://192.0.2.11/d3", + "http://192.0.2.12/d4", ], expected_endpoints: Some(Endpoints(vec![ Endpoint { @@ -1242,8 +1245,8 @@ mod test { num: 16, server_addr: "0.0.0.0:9003", args: vec![ - "http://localhost:9000/d1", - "http://localhost:9001/d2", + "http://127.0.0.1:9000/d1", + "http://127.0.0.1:9001/d2", "http://127.0.0.1:9002/d3", case6_endpoint1.as_str(), ], diff --git a/crates/ecstore/src/set_disk.rs b/crates/ecstore/src/set_disk.rs index cb0f68992..24d46ebf7 100644 --- a/crates/ecstore/src/set_disk.rs +++ b/crates/ecstore/src/set_disk.rs @@ -1943,19 +1943,54 @@ impl ObjectOperations for SetDisks { //} let mut uploaded_parts: Vec = vec![]; - let rs: Option = None; - let gr = get_transitioned_object_reader(bucket, object, &rs, &HeaderMap::new(), &oi, opts).await; - if let Err(err) = gr { - return set_restore_header_fn(&mut oi, Some(StorageError::Io(err))).await; - } - let gr = gr.unwrap(); - - for part_info in &oi.parts { - let reader = BufReader::new(Cursor::new(vec![] /*gr.stream*/)); + let parts = oi.parts.clone(); + let mut part_offset: i64 = 0; + for part_info in &parts { + let mut part_opts = opts.clone(); + part_opts.part_number = Some(part_info.number); + if part_info.actual_size <= 0 { + return set_restore_header_fn( + &mut oi, + Some(Error::other(format!("invalid multipart restore part size {}", part_info.actual_size))), + ) + .await; + } + let part_end = match part_offset.checked_add(part_info.actual_size - 1) { + Some(end) => end, + None => { + return set_restore_header_fn( + &mut oi, + Some(Error::other("multipart restore part range overflow".to_string())), + ) + .await; + } + }; + let rs = Some(HTTPRangeSpec { + is_suffix_length: false, + start: part_offset, + end: part_end, + }); + part_offset = match part_end.checked_add(1) { + Some(next) => next, + None => { + return set_restore_header_fn( + &mut oi, + Some(Error::other("multipart restore part offset overflow".to_string())), + ) + .await; + } + }; + let gr = match get_transitioned_object_reader(bucket, object, &rs, &HeaderMap::new(), &oi, &part_opts).await { + Ok(reader) => reader, + Err(err) => { + return set_restore_header_fn(&mut oi, Some(StorageError::Io(err))).await; + } + }; + let reader = BufReader::new(gr.stream); let hash_reader = HashReader::new( Box::new(WarpReader::new(reader)), - part_info.size as i64, - part_info.size as i64, + part_info.actual_size, + part_info.actual_size, None, None, false, @@ -1968,7 +2003,7 @@ impl ObjectOperations for SetDisks { //if let Err(err) = p_info { // return set_restore_header_fn(&mut oi, err).await; //} - if p_info.size != part_info.size { + if p_info.size as i64 != part_info.actual_size { return set_restore_header_fn( &mut oi, Some(Error::other(ObjectApiError::InvalidObjectState(GenericError { diff --git a/crates/ecstore/src/store.rs b/crates/ecstore/src/store.rs index c58955a24..aafed2a8b 100644 --- a/crates/ecstore/src/store.rs +++ b/crates/ecstore/src/store.rs @@ -27,7 +27,7 @@ use crate::bucket::utils::check_new_multipart_args; use crate::bucket::utils::check_object_args; use crate::bucket::utils::check_put_object_args; use crate::bucket::utils::check_put_object_part_args; -use crate::bucket::utils::{check_valid_bucket_name, check_valid_bucket_name_strict, is_meta_bucketname}; +use crate::bucket::utils::{check_valid_bucket_name, check_valid_bucket_name_strict}; use crate::config::GLOBAL_STORAGE_CLASS; use crate::config::storageclass; use crate::disk::endpoint::{Endpoint, EndpointType}; diff --git a/crates/ecstore/src/store/bucket.rs b/crates/ecstore/src/store/bucket.rs index ff1fec359..484e6b795 100644 --- a/crates/ecstore/src/store/bucket.rs +++ b/crates/ecstore/src/store/bucket.rs @@ -13,6 +13,7 @@ // limitations under the License. use super::*; +use crate::bucket::utils::is_meta_bucketname; fn should_override_created_from_metadata(created: OffsetDateTime) -> bool { created != OffsetDateTime::UNIX_EPOCH diff --git a/crates/ecstore/src/store/rebalance.rs b/crates/ecstore/src/store/rebalance.rs index 1915f573d..a437cd811 100644 --- a/crates/ecstore/src/store/rebalance.rs +++ b/crates/ecstore/src/store/rebalance.rs @@ -13,6 +13,7 @@ // limitations under the License. use super::*; +use crate::bucket::utils::is_meta_bucketname; impl ECStore { #[instrument(level = "debug", skip(self))] diff --git a/crates/scanner/tests/lifecycle_integration_test.rs b/crates/scanner/tests/lifecycle_integration_test.rs index 2cc3683b1..91b5995f2 100644 --- a/crates/scanner/tests/lifecycle_integration_test.rs +++ b/crates/scanner/tests/lifecycle_integration_test.rs @@ -13,28 +13,42 @@ // limitations under the License. use rustfs_ecstore::{ + bucket::lifecycle::lifecycle::TransitionOptions, bucket::metadata::BUCKET_LIFECYCLE_CONFIG, - bucket::metadata_sys, + bucket::{lifecycle::bucket_lifecycle_ops::enqueue_transition_for_existing_objects, metadata_sys}, + client::transition_api::{ReadCloser, ReaderImpl}, disk::endpoint::Endpoint, endpoints::{EndpointServerPools, Endpoints, PoolEndpoints}, global::GLOBAL_TierConfigMgr, store::ECStore, - store_api::{BucketOperations, MakeBucketOptions, ObjectIO, ObjectOperations, ObjectOptions, PutObjReader}, - tier::tier_config::{TierConfig, TierMinIO, TierType}, + store_api::{ + BucketOperations, MakeBucketOptions, MultipartOperations, ObjectIO, ObjectOperations, ObjectOptions, PutObjReader, + }, + tier::{ + tier_config::{TierConfig, TierMinIO, TierType}, + warm_backend::{WarmBackend, WarmBackendGetOpts}, + }, }; use rustfs_scanner::scanner::init_data_scanner; +use s3s::dto::RestoreRequest; use serial_test::serial; use std::{ + collections::HashMap, + io::Cursor, path::PathBuf, sync::{Arc, Once, OnceLock}, time::Duration, }; use tokio::fs; +use tokio::io::AsyncReadExt; +use tokio::sync::Mutex; use tokio_util::sync::CancellationToken; use tracing::info; +use uuid::Uuid; 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(|| { @@ -209,8 +223,17 @@ async fn set_bucket_lifecycle_deletemarker(bucket_name: &str) -> Result<(), Box< #[allow(dead_code)] async fn set_bucket_lifecycle_transition(bucket_name: &str) -> Result<(), Box> { + set_bucket_lifecycle_transition_with_tier(bucket_name, "COLDTIER44").await +} + +#[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 = r#" + let lifecycle_xml = format!( + r#" test-rule @@ -220,7 +243,7 @@ async fn set_bucket_lifecycle_transition(bucket_name: &str) -> Result<(), Box 0 - COLDTIER44 + {storage_class} @@ -231,12 +254,13 @@ async fn set_bucket_lifecycle_transition(bucket_name: &str) -> Result<(), Box 0 - COLDTIER44 + {storage_class} -"#; +"# + ); - metadata_sys::update(bucket_name, BUCKET_LIFECYCLE_CONFIG, lifecycle_xml.as_bytes().to_vec()).await?; + metadata_sys::update(bucket_name, BUCKET_LIFECYCLE_CONFIG, lifecycle_xml.into_bytes()).await?; Ok(()) } @@ -350,6 +374,114 @@ async fn wait_for_object_absence(ecstore: &Arc, bucket: &str, object: & } } +#[derive(Clone, Default)] +struct MockWarmBackend { + objects: Arc>>>, +} + +impl MockWarmBackend { + async fn put_bytes(&self, object: &str, bytes: Vec) -> String { + self.objects.lock().await.insert(object.to_string(), bytes); + Uuid::new_v4().to_string() + } + + async fn read_bytes(&self, reader: ReaderImpl) -> Result, std::io::Error> { + match reader { + ReaderImpl::Body(bytes) => Ok(bytes.to_vec()), + ReaderImpl::ObjectBody(mut reader) => { + let mut buf = Vec::new(); + reader.stream.read_to_end(&mut buf).await?; + Ok(buf) + } + } + } +} + +#[async_trait::async_trait] +impl WarmBackend for MockWarmBackend { + async fn put(&self, object: &str, r: ReaderImpl, _length: i64) -> Result { + let bytes = self.read_bytes(r).await?; + Ok(self.put_bytes(object, bytes).await) + } + + async fn put_with_meta( + &self, + object: &str, + r: ReaderImpl, + _length: i64, + _meta: HashMap, + ) -> Result { + let bytes = self.read_bytes(r).await?; + Ok(self.put_bytes(object, bytes).await) + } + + async fn get(&self, object: &str, _rv: &str, opts: WarmBackendGetOpts) -> Result { + let objects = self.objects.lock().await; + let Some(bytes) = objects.get(object) else { + return Err(std::io::Error::new(std::io::ErrorKind::NotFound, "mock object not found")); + }; + + let start = opts.start_offset.max(0) as usize; + let end = if opts.length > 0 { + start.saturating_add(opts.length as usize).min(bytes.len()) + } else { + bytes.len() + }; + + Ok(tokio::io::BufReader::new(Cursor::new(bytes[start.min(bytes.len())..end].to_vec()))) + } + + async fn remove(&self, object: &str, _rv: &str) -> Result<(), std::io::Error> { + self.objects.lock().await.remove(object); + Ok(()) + } + + async fn in_use(&self) -> Result { + Ok(false) + } +} + +async fn register_mock_tier(tier_name: &str) -> MockWarmBackend { + let backend = MockWarmBackend::default(); + let mut tier_config_mgr = GLOBAL_TierConfigMgr.write().await; + tier_config_mgr.tiers.insert( + tier_name.to_string(), + TierConfig { + version: "v1".to_string(), + tier_type: TierType::MinIO, + name: tier_name.to_string(), + ..Default::default() + }, + ); + tier_config_mgr + .driver_cache + .insert(tier_name.to_string(), Box::new(backend.clone())); + backend +} + +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; + } +} + mod serial_tests { use super::*; @@ -443,4 +575,283 @@ mod serial_tests { println!("Lifecycle transition basic test completed"); } + + #[tokio::test(flavor = "multi_thread", worker_threads = 1)] + #[serial] + 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"); + + upload_test_object(&ecstore, put_bucket.as_str(), put_object, put_payload).await; + + 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.objects.lock().await.contains_key(&put_info.transitioned_object.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![rustfs_ecstore::store_api::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 + .objects + .lock() + .await + .contains_key(&multipart_info.transitioned_object.name) + ); + + 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.objects.lock().await.contains_key(©_info.transitioned_object.name)); + + 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.objects.lock().await.contains_key(&info.transitioned_object.name)); + + 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![ + rustfs_ecstore::store_api::CompletePart { + part_num: 1, + etag: uploaded_part1.etag.clone(), + ..Default::default() + }, + rustfs_ecstore::store_api::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); + } } diff --git a/rustfs/src/app/bucket_usecase.rs b/rustfs/src/app/bucket_usecase.rs index 07e57d1a7..71fbc01ce 100644 --- a/rustfs/src/app/bucket_usecase.rs +++ b/rustfs/src/app/bucket_usecase.rs @@ -28,7 +28,7 @@ use http::StatusCode; use metrics::counter; use rustfs_config::RUSTFS_REGION; use rustfs_ecstore::bucket::{ - lifecycle::bucket_lifecycle_ops::validate_transition_tier, + lifecycle::bucket_lifecycle_ops::{enqueue_transition_for_existing_objects, validate_transition_tier}, metadata::{ BUCKET_CORS_CONFIG, BUCKET_LIFECYCLE_CONFIG, BUCKET_NOTIFICATION_CONFIG, BUCKET_POLICY_CONFIG, BUCKET_PUBLIC_ACCESS_BLOCK_CONFIG, BUCKET_REPLICATION_CONFIG, BUCKET_SSECONFIG, BUCKET_TAGGING_CONFIG, @@ -126,6 +126,27 @@ fn validate_lifecycle_rule_status(rules: &[LifecycleRule]) -> Result<(), &'stati Ok(()) } +fn lifecycle_has_transition_rules(config: &BucketLifecycleConfiguration) -> bool { + config.rules.iter().any(|rule| { + rule.status == ExpirationStatus::from_static(ExpirationStatus::ENABLED) + && (rule.transitions.as_ref().is_some_and(|transitions| { + transitions.iter().any(|transition| { + transition + .storage_class + .as_ref() + .is_some_and(|storage_class| !storage_class.as_str().is_empty()) + }) + }) || rule.noncurrent_version_transitions.as_ref().is_some_and(|transitions| { + transitions.iter().any(|transition| { + transition + .storage_class + .as_ref() + .is_some_and(|storage_class| !storage_class.as_str().is_empty()) + }) + })) + }) +} + #[derive(Clone, Default)] pub struct DefaultBucketUsecase { context: Option>, @@ -1051,6 +1072,17 @@ impl DefaultBucketUsecase { .await .map_err(ApiError::from)?; + if lifecycle_has_transition_rules(&input_cfg) + && let Some(store) = new_object_layer_fn() + { + let bucket_name = bucket.clone(); + tokio::spawn(async move { + if let Err(err) = enqueue_transition_for_existing_objects(store, &bucket_name).await { + warn!(bucket = %bucket_name, error = ?err, "failed to enqueue transition for existing objects"); + } + }); + } + Ok(S3Response::new(PutBucketLifecycleConfigurationOutput::default())) } @@ -1902,6 +1934,56 @@ mod tests { assert_eq!(validate_lifecycle_rule_status(&rules).unwrap_err(), ERR_LIFECYCLE_RULE_STATUS); } + #[test] + fn lifecycle_has_transition_rules_ignores_disabled_rules() { + let config = BucketLifecycleConfiguration { + expiry_updated_at: None, + rules: vec![LifecycleRule { + status: ExpirationStatus::from_static(ExpirationStatus::DISABLED), + expiration: None, + abort_incomplete_multipart_upload: None, + del_marker_expiration: None, + filter: None, + id: Some("disabled-transition".to_string()), + noncurrent_version_expiration: None, + noncurrent_version_transitions: None, + prefix: None, + transitions: Some(vec![Transition { + days: Some(1), + date: None, + storage_class: Some(TransitionStorageClass::from_static("WARM")), + }]), + }], + }; + + assert!(!lifecycle_has_transition_rules(&config)); + } + + #[test] + fn lifecycle_has_transition_rules_accepts_enabled_noncurrent_transitions() { + let config = BucketLifecycleConfiguration { + expiry_updated_at: None, + rules: vec![LifecycleRule { + status: ExpirationStatus::from_static(ExpirationStatus::ENABLED), + expiration: None, + abort_incomplete_multipart_upload: None, + del_marker_expiration: None, + filter: None, + id: Some("enabled-noncurrent-transition".to_string()), + noncurrent_version_expiration: None, + noncurrent_version_transitions: Some(vec![NoncurrentVersionTransition { + noncurrent_days: Some(1), + newer_noncurrent_versions: None, + storage_class: Some(TransitionStorageClass::from_static("WARM")), + }]), + prefix: None, + transitions: None, + }], + }; + + assert!(lifecycle_has_transition_rules(&config)); + } + #[tokio::test] async fn execute_list_buckets_returns_internal_error_when_store_uninitialized() { let input = ListBucketsInput::builder().build().unwrap(); diff --git a/rustfs/src/app/lifecycle_transition_api_test.rs b/rustfs/src/app/lifecycle_transition_api_test.rs new file mode 100644 index 000000000..74d13a45e --- /dev/null +++ b/rustfs/src/app/lifecycle_transition_api_test.rs @@ -0,0 +1,439 @@ +// 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 super::{multipart_usecase::DefaultMultipartUsecase, object_usecase::DefaultObjectUsecase}; +use crate::storage::ecfs::FS; +use bytes::Bytes; +use futures::stream; +use http::{Extensions, HeaderMap, Method, Uri}; +use rustfs_ecstore::{ + bucket::metadata::BUCKET_LIFECYCLE_CONFIG, + bucket::metadata_sys, + client::object_api_utils::to_s3s_etag, + client::transition_api::{ReadCloser, ReaderImpl}, + disk::endpoint::Endpoint, + endpoints::{EndpointServerPools, Endpoints, PoolEndpoints}, + global::GLOBAL_TierConfigMgr, + store::ECStore, + store_api::{ + BucketOperations, BucketOptions, MakeBucketOptions, MultipartOperations, ObjectIO, ObjectOperations, ObjectOptions, + PutObjReader, + }, + tier::{ + tier_config::{TierConfig, TierType}, + warm_backend::{WarmBackend, WarmBackendGetOpts}, + }, +}; +use s3s::{S3Request, dto::*}; +use serial_test::serial; +use std::{ + collections::HashMap, + convert::Infallible, + io::Cursor, + path::PathBuf, + sync::{Arc, Once, OnceLock}, + time::Duration, +}; +use tokio::fs; +use tokio::io::AsyncReadExt; +use tokio::sync::Mutex; +use tokio_util::sync::CancellationToken; +use uuid::Uuid; + +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(|| {}); +} + +async fn setup_test_env() -> (Vec, Arc) { + init_tracing(); + + if let Some((paths, ecstore)) = GLOBAL_ENV.get() { + return (paths.clone(), ecstore.clone()); + } + + let test_base_dir = format!("/tmp/rustfs_app_lifecycle_test_{}", Uuid::new_v4()); + let temp_dir = 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(vec![pool_endpoints]); + + rustfs_ecstore::store::init_local_disks(endpoint_pools.clone()).await.unwrap(); + + let server_addr: std::net::SocketAddr = "127.0.0.1:9003".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(); + metadata_sys::init_bucket_metadata_sys(ecstore.clone(), buckets).await; + + rustfs_ecstore::bucket::lifecycle::bucket_lifecycle_ops::init_background_expiry(ecstore.clone()).await; + + let _ = GLOBAL_ENV.set((disk_paths.clone(), ecstore.clone())); + + (disk_paths, ecstore) +} + +async fn create_test_bucket(ecstore: &Arc, bucket_name: &str) { + (**ecstore) + .make_bucket( + bucket_name, + &MakeBucketOptions { + versioning_enabled: true, + ..Default::default() + }, + ) + .await + .expect("Failed to create test bucket"); +} + +async fn upload_test_object(ecstore: &Arc, bucket: &str, object: &str, data: &[u8]) { + let mut reader = PutObjReader::from_vec(data.to_vec()); + (**ecstore) + .put_object(bucket, object, &mut reader, &ObjectOptions::default()) + .await + .expect("Failed to upload test object"); +} + +async fn set_bucket_lifecycle_transition_with_tier( + bucket_name: &str, + storage_class: &str, +) -> Result<(), Box> { + let lifecycle_xml = format!( + r#" + + + test-rule + Enabled + + test/ + + + 0 + {storage_class} + + +"# + ); + + metadata_sys::update(bucket_name, BUCKET_LIFECYCLE_CONFIG, lifecycle_xml.into_bytes()).await?; + Ok(()) +} + +#[derive(Clone, Default)] +struct MockWarmBackend { + objects: Arc>>>, +} + +impl MockWarmBackend { + async fn put_bytes(&self, object: &str, bytes: Vec) -> String { + self.objects.lock().await.insert(object.to_string(), bytes); + Uuid::new_v4().to_string() + } + + async fn read_bytes(&self, reader: ReaderImpl) -> Result, std::io::Error> { + match reader { + ReaderImpl::Body(bytes) => Ok(bytes.to_vec()), + ReaderImpl::ObjectBody(mut reader) => { + let mut buf = Vec::new(); + reader.stream.read_to_end(&mut buf).await?; + Ok(buf) + } + } + } +} + +#[async_trait::async_trait] +impl WarmBackend for MockWarmBackend { + async fn put(&self, object: &str, r: ReaderImpl, _length: i64) -> Result { + let bytes = self.read_bytes(r).await?; + Ok(self.put_bytes(object, bytes).await) + } + + async fn put_with_meta( + &self, + object: &str, + r: ReaderImpl, + _length: i64, + _meta: HashMap, + ) -> Result { + let bytes = self.read_bytes(r).await?; + Ok(self.put_bytes(object, bytes).await) + } + + async fn get(&self, object: &str, _rv: &str, opts: WarmBackendGetOpts) -> Result { + let objects = self.objects.lock().await; + let Some(bytes) = objects.get(object) else { + return Err(std::io::Error::new(std::io::ErrorKind::NotFound, "mock object not found")); + }; + + let start = opts.start_offset.max(0) as usize; + let end = if opts.length > 0 { + start.saturating_add(opts.length as usize).min(bytes.len()) + } else { + bytes.len() + }; + + Ok(tokio::io::BufReader::new(Cursor::new(bytes[start.min(bytes.len())..end].to_vec()))) + } + + async fn remove(&self, object: &str, _rv: &str) -> Result<(), std::io::Error> { + self.objects.lock().await.remove(object); + Ok(()) + } + + async fn in_use(&self) -> Result { + Ok(false) + } +} + +async fn register_mock_tier(tier_name: &str) -> MockWarmBackend { + let backend = MockWarmBackend::default(); + let mut tier_config_mgr = GLOBAL_TierConfigMgr.write().await; + tier_config_mgr.tiers.insert( + tier_name.to_string(), + TierConfig { + version: "v1".to_string(), + tier_type: TierType::MinIO, + name: tier_name.to_string(), + ..Default::default() + }, + ); + tier_config_mgr + .driver_cache + .insert(tier_name.to_string(), Box::new(backend.clone())); + backend +} + +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; + } +} + +fn build_request(input: T, method: Method) -> S3Request { + S3Request { + input, + method, + uri: Uri::from_static("/"), + headers: HeaderMap::new(), + extensions: Extensions::new(), + credentials: None, + region: None, + service: None, + trailing_headers: None, + } +} + +fn streaming_blob_from_bytes(data: &[u8]) -> StreamingBlob { + let body = Bytes::copy_from_slice(data); + StreamingBlob::wrap::<_, Infallible>(stream::once(async move { Ok(body) })) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 1)] +#[serial] +#[ignore = "requires isolated global object layer state"] +async fn put_and_copy_object_transition_immediately_via_usecases() { + let (_disk_paths, ecstore) = setup_test_env().await; + let fs = FS::new(); + let usecase = DefaultObjectUsecase::without_context(); + + 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-api-put-{}", &Uuid::new_v4().simple().to_string()[..8]); + let put_object = "test/object.txt"; + let put_payload = b"Hello, immediate transition through put API!"; + + 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 put_input = PutObjectInput::builder() + .bucket(put_bucket.clone()) + .key(put_object.to_string()) + .body(Some(streaming_blob_from_bytes(put_payload))) + .content_length(Some(put_payload.len() as i64)) + .build() + .unwrap(); + + usecase + .execute_put_object(&fs, build_request(put_input, Method::PUT)) + .await + .expect("Failed to put object through usecase"); + + let put_info = wait_for_transition(&ecstore, put_bucket.as_str(), put_object, TRANSITION_WAIT_TIMEOUT) + .await + .expect("object should transition immediately after put usecase"); + + assert_eq!(put_info.transitioned_object.status, "complete"); + assert_eq!(put_info.transitioned_object.tier, tier_name); + assert!(backend.objects.lock().await.contains_key(&put_info.transitioned_object.name)); + + let src_bucket = format!("test-api-copy-src-{}", &Uuid::new_v4().simple().to_string()[..8]); + let dst_bucket = format!("test-api-copy-dst-{}", &Uuid::new_v4().simple().to_string()[..8]); + let src_object = "test/source.txt"; + let dst_object = "test/copied.txt"; + let copy_payload = b"copy object immediate transition through copy API"; + + 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, copy_payload).await; + + let copy_input = CopyObjectInput::builder() + .copy_source(CopySource::Bucket { + bucket: src_bucket.clone().into(), + key: src_object.to_string().into(), + version_id: None, + }) + .bucket(dst_bucket.clone()) + .key(dst_object.to_string()) + .build() + .unwrap(); + + usecase + .execute_copy_object(build_request(copy_input, Method::PUT)) + .await + .expect("Failed to copy object through usecase"); + + let copy_info = wait_for_transition(&ecstore, dst_bucket.as_str(), dst_object, TRANSITION_WAIT_TIMEOUT) + .await + .expect("copied object should transition immediately after copy usecase"); + + assert_eq!(copy_info.transitioned_object.status, "complete"); + assert_eq!(copy_info.transitioned_object.tier, tier_name); + assert!(backend.objects.lock().await.contains_key(©_info.transitioned_object.name)); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 1)] +#[serial] +#[ignore = "requires isolated global object layer state"] +async fn complete_multipart_upload_transitions_immediately_via_usecase() { + let (_disk_paths, ecstore) = setup_test_env().await; + let usecase = DefaultMultipartUsecase::without_context(); + + let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase(); + let backend = register_mock_tier(&tier_name).await; + + let bucket = format!("test-api-mpu-{}", &Uuid::new_v4().simple().to_string()[..8]); + let object = "test/multipart.txt"; + let payload = b"multipart immediate transition through complete API"; + + create_test_bucket(&ecstore, bucket.as_str()).await; + set_bucket_lifecycle_transition_with_tier(bucket.as_str(), &tier_name) + .await + .expect("Failed to set lifecycle configuration"); + + let upload = ecstore + .new_multipart_upload(bucket.as_str(), object, &ObjectOptions::default()) + .await + .expect("Failed to create multipart upload"); + + let mut reader = PutObjReader::from_vec(payload.to_vec()); + let uploaded_part = ecstore + .put_object_part(bucket.as_str(), object, &upload.upload_id, 1, &mut reader, &ObjectOptions::default()) + .await + .expect("Failed to upload multipart part"); + + let complete_input = CompleteMultipartUploadInput::builder() + .bucket(bucket.clone()) + .key(object.to_string()) + .upload_id(upload.upload_id.clone()) + .multipart_upload(Some(CompletedMultipartUpload { + parts: Some(vec![CompletedPart { + part_number: Some(1), + e_tag: uploaded_part.etag.clone().map(|etag| to_s3s_etag(&etag)), + ..Default::default() + }]), + })) + .build() + .unwrap(); + + usecase + .execute_complete_multipart_upload(build_request(complete_input, Method::POST)) + .await + .expect("Failed to complete multipart upload through usecase"); + + let info = wait_for_transition(&ecstore, bucket.as_str(), object, TRANSITION_WAIT_TIMEOUT) + .await + .expect("multipart object should transition immediately after complete usecase"); + + assert_eq!(info.transitioned_object.status, "complete"); + assert_eq!(info.transitioned_object.tier, tier_name); + assert!(backend.objects.lock().await.contains_key(&info.transitioned_object.name)); +} diff --git a/rustfs/src/app/mod.rs b/rustfs/src/app/mod.rs index 8fbfed090..ab8391102 100644 --- a/rustfs/src/app/mod.rs +++ b/rustfs/src/app/mod.rs @@ -20,3 +20,6 @@ pub mod bucket_usecase; pub mod context; pub mod multipart_usecase; pub mod object_usecase; + +#[cfg(test)] +mod lifecycle_transition_api_test; diff --git a/rustfs/src/app/multipart_usecase.rs b/rustfs/src/app/multipart_usecase.rs index b2df090bd..800efa3de 100644 --- a/rustfs/src/app/multipart_usecase.rs +++ b/rustfs/src/app/multipart_usecase.rs @@ -30,6 +30,7 @@ use futures::StreamExt; use http::{HeaderMap, Uri}; use rustfs_ecstore::bucket::quota::checker::QuotaChecker; use rustfs_ecstore::bucket::{ + lifecycle::{bucket_lifecycle_audit::LcEventSrc, bucket_lifecycle_ops::enqueue_transition_immediate}, metadata_sys, quota::QuotaOperation, replication::{get_must_replicate_options, must_replicate, schedule_replication}, @@ -60,6 +61,10 @@ use tokio_util::io::StreamReader; use tracing::{info, instrument, warn}; use urlencoding::encode; +async fn maybe_enqueue_transition_immediate(obj_info: &rustfs_ecstore::store_api::ObjectInfo, src: LcEventSrc) { + enqueue_transition_immediate(obj_info, src).await; +} + /// Returns InvalidRange error if CopySourceRange end exceeds the source object size. /// Used by execute_upload_part_copy to reject out-of-bounds ranges per S3 spec. fn validate_copy_source_range_not_exceeds(range_spec: &HTTPRangeSpec, object_size: i64) -> S3Result<()> { @@ -356,6 +361,8 @@ impl DefaultMultipartUsecase { } } + maybe_enqueue_transition_immediate(&obj_info, LcEventSrc::S3CompleteMultipartUpload).await; + // Invalidate cache for the completed multipart object let manager = get_concurrency_manager(); let mpu_bucket = bucket.clone(); diff --git a/rustfs/src/app/object_usecase.rs b/rustfs/src/app/object_usecase.rs index b8b9dd185..c633cc0f6 100644 --- a/rustfs/src/app/object_usecase.rs +++ b/rustfs/src/app/object_usecase.rs @@ -42,7 +42,8 @@ use metrics::{counter, histogram}; use rustfs_ecstore::bucket::quota::checker::QuotaChecker; use rustfs_ecstore::bucket::{ lifecycle::{ - bucket_lifecycle_ops::{RestoreRequestOps, post_restore_opts}, + bucket_lifecycle_audit::LcEventSrc, + bucket_lifecycle_ops::{RestoreRequestOps, enqueue_transition_immediate, post_restore_opts}, lifecycle::{self, Lifecycle, TransitionOptions}, }, metadata::{BUCKET_VERSIONING_CONFIG, OBJECT_LOCK_CONFIG}, @@ -116,6 +117,10 @@ use tokio_util::io::{ReaderStream, StreamReader}; use tracing::{debug, error, info, instrument, warn}; use uuid::Uuid; +async fn maybe_enqueue_transition_immediate(obj_info: &ObjectInfo, src: LcEventSrc) { + enqueue_transition_immediate(obj_info, src).await; +} + /// Extract trailing-header checksum values, overriding the corresponding input fields. fn apply_trailing_checksums( algorithm: Option<&str>, @@ -509,6 +514,8 @@ impl DefaultObjectUsecase { .await .map_err(ApiError::from)?; + maybe_enqueue_transition_immediate(&obj_info, LcEventSrc::S3PutObject).await; + // Fast in-memory update for immediate quota consistency rustfs_ecstore::data_usage::increment_bucket_usage_memory(&bucket, obj_info.size as u64).await; @@ -2307,6 +2314,8 @@ impl DefaultObjectUsecase { .await .map_err(ApiError::from)?; + maybe_enqueue_transition_immediate(&oi, LcEventSrc::S3CopyObject).await; + // Update quota tracking after successful copy if has_bucket_metadata { rustfs_ecstore::data_usage::increment_bucket_usage_memory(&bucket, oi.size as u64).await; @@ -3607,6 +3616,8 @@ impl DefaultObjectUsecase { .await .map_err(ApiError::from)?; + maybe_enqueue_transition_immediate(&obj_info, LcEventSrc::S3PutObject).await; + let manager = get_concurrency_manager(); let fpath_clone = fpath.clone(); let bucket_clone = bucket.clone();