diff --git a/crates/e2e_test/src/common.rs b/crates/e2e_test/src/common.rs index 67aef5f97..7fb344bb4 100644 --- a/crates/e2e_test/src/common.rs +++ b/crates/e2e_test/src/common.rs @@ -622,6 +622,7 @@ pub struct RustFSTestClusterEnvironment { pub temp_dir: String, pub access_key: String, pub secret_key: String, + pub extra_env: Vec<(String, String)>, } impl RustFSTestClusterEnvironment { @@ -670,9 +671,19 @@ impl RustFSTestClusterEnvironment { temp_dir, access_key: DEFAULT_ACCESS_KEY.to_string(), secret_key: DEFAULT_SECRET_KEY.to_string(), + extra_env: Vec::new(), }) } + /// Add an extra environment variable applied to every cluster node process. + pub fn set_env(&mut self, key: K, value: V) + where + K: Into, + V: Into, + { + self.extra_env.push((key.into(), value.into())); + } + /// Build the volumes argument string for RustFS binary (internal helper method). /// /// Concatenates the address and data directory of all cluster nodes into a single string @@ -703,15 +714,20 @@ impl RustFSTestClusterEnvironment { for (i, node) in self.nodes.iter_mut().enumerate() { info!("Starting cluster node {} on {}", i, node.address); - let process = Command::new(&binary_path) + let mut command = Command::new(&binary_path); + command .env("RUSTFS_VOLUMES", &volumes_arg) .env("RUSTFS_ADDRESS", &node.address) .env("RUSTFS_ACCESS_KEY", &self.access_key) .env("RUSTFS_SECRET_KEY", &self.secret_key) .env("RUSTFS_CONSOLE_ENABLE", "false") - .env("RUST_LOG", "rustfs=info,rustfs_notify=debug") - .current_dir(&node.data_dir) - .spawn()?; + .env("RUST_LOG", "rustfs=info,rustfs_notify=debug"); + + for (key, value) in &self.extra_env { + command.env(key, value); + } + + let process = command.current_dir(&node.data_dir).spawn()?; node.process = Some(process); } diff --git a/crates/e2e_test/src/lib.rs b/crates/e2e_test/src/lib.rs index 458c453ba..1e642f2a6 100644 --- a/crates/e2e_test/src/lib.rs +++ b/crates/e2e_test/src/lib.rs @@ -116,6 +116,9 @@ mod bucket_logging_test; #[cfg(test)] mod multipart_auth_test; +#[cfg(test)] +mod stale_multipart_cleanup_cluster_test; + // Object lambda end-to-end regression tests #[cfg(test)] mod object_lambda_test; diff --git a/crates/e2e_test/src/stale_multipart_cleanup_cluster_test.rs b/crates/e2e_test/src/stale_multipart_cleanup_cluster_test.rs new file mode 100644 index 000000000..618b79b61 --- /dev/null +++ b/crates/e2e_test/src/stale_multipart_cleanup_cluster_test.rs @@ -0,0 +1,155 @@ +// 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 crate::common::{RustFSTestClusterEnvironment, init_logging}; +use aws_sdk_s3::error::SdkError; +use aws_sdk_s3::primitives::ByteStream; +use aws_sdk_s3::types::CompletedMultipartUpload; +use serial_test::serial; +use tokio::time::{Duration, sleep}; +use tracing::info; +use uuid::Uuid; + +const CLEANUP_BUCKET: &str = "stale-multipart-cleanup-cluster"; + +async fn list_parts_reports_missing_upload( + client: &aws_sdk_s3::Client, + bucket: &str, + key: &str, + upload_id: &str, +) -> Result> { + let result = client.list_parts().bucket(bucket).key(key).upload_id(upload_id).send().await; + match result { + Ok(_) => Ok(false), + Err(SdkError::ServiceError(err)) => { + let code = err.err().meta().code().unwrap_or(""); + if code == "NoSuchUpload" { + Ok(true) + } else { + Err(format!("unexpected list_parts service error: code={code}, err={err:?}").into()) + } + } + Err(err) => Err(format!("unexpected list_parts error: {err:?}").into()), + } +} + +async fn complete_reports_missing_upload( + client: &aws_sdk_s3::Client, + bucket: &str, + key: &str, + upload_id: &str, +) -> Result> { + let result = client + .complete_multipart_upload() + .bucket(bucket) + .key(key) + .upload_id(upload_id) + .multipart_upload(CompletedMultipartUpload::builder().build()) + .send() + .await; + match result { + Ok(_) => Ok(false), + Err(SdkError::ServiceError(err)) => { + let code = err.err().meta().code().unwrap_or(""); + if code == "NoSuchUpload" { + Ok(true) + } else { + Err(format!("unexpected complete_multipart_upload service error: code={code}, err={err:?}").into()) + } + } + Err(err) => Err(format!("unexpected complete_multipart_upload error: {err:?}").into()), + } +} + +async fn wait_for_cleanup_on_all_nodes( + clients: &[aws_sdk_s3::Client], + bucket: &str, + key: &str, + upload_id: &str, +) -> Result<(), Box> { + for attempt in 0..30 { + let mut all_cleaned = true; + for (idx, client) in clients.iter().enumerate() { + let list_parts_missing = list_parts_reports_missing_upload(client, bucket, key, upload_id).await?; + let complete_missing = complete_reports_missing_upload(client, bucket, key, upload_id).await?; + if !(list_parts_missing && complete_missing) { + info!("stale multipart still visible on node {} at attempt {}", idx, attempt + 1); + all_cleaned = false; + break; + } + } + + if all_cleaned { + return Ok(()); + } + + sleep(Duration::from_secs(1)).await; + } + + Err("stale multipart upload was not cleaned up on all nodes within timeout".into()) +} + +#[tokio::test] +#[serial] +async fn test_stale_multipart_cleanup_removes_incomplete_upload_across_cluster() +-> Result<(), Box> { + init_logging(); + + let mut cluster = RustFSTestClusterEnvironment::new(4).await?; + cluster.set_env("RUSTFS_API_STALE_UPLOADS_EXPIRY", "5s"); + cluster.set_env("RUSTFS_API_STALE_UPLOADS_CLEANUP_INTERVAL", "1s"); + cluster.start().await?; + cluster.create_test_bucket(CLEANUP_BUCKET).await?; + + let clients = cluster.create_all_clients()?; + let key = format!("multipart/stale-{}.txt", Uuid::new_v4().simple()); + + let create_output = clients[0] + .create_multipart_upload() + .bucket(CLEANUP_BUCKET) + .key(&key) + .send() + .await?; + let upload_id = create_output + .upload_id() + .ok_or("create_multipart_upload response missing upload_id")? + .to_string(); + + clients[1] + .upload_part() + .bucket(CLEANUP_BUCKET) + .key(&key) + .upload_id(&upload_id) + .part_number(1) + .body(ByteStream::from_static(b"stale multipart part")) + .send() + .await?; + + let parts_before_cleanup = clients[2] + .list_parts() + .bucket(CLEANUP_BUCKET) + .key(&key) + .upload_id(&upload_id) + .send() + .await?; + assert_eq!( + parts_before_cleanup.parts().len(), + 1, + "multipart upload should be visible before background cleanup" + ); + + wait_for_cleanup_on_all_nodes(&clients, CLEANUP_BUCKET, &key, &upload_id).await?; + + Ok(()) +} diff --git a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs index 22c08b544..08efbed26 100644 --- a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs +++ b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs @@ -19,22 +19,28 @@ #![allow(clippy::all)] use crate::bucket::lifecycle::bucket_lifecycle_audit::{LcAuditEvent, LcEventSrc}; -use crate::bucket::lifecycle::lifecycle::{self, ExpirationOptions, Lifecycle, TransitionOptions}; +use crate::bucket::lifecycle::lifecycle::{ + self, ExpirationOptions, Lifecycle, ObjectOpts, TransitionOptions, abort_incomplete_multipart_upload_due, +}; use crate::bucket::lifecycle::tier_last_day_stats::{DailyAllTierStats, LastDayTierStats}; use crate::bucket::lifecycle::tier_sweeper::{Jentry, delete_object_from_remote_tier}; use crate::bucket::object_lock::objectlock_sys::check_object_lock_for_deletion; -use crate::bucket::{metadata_sys::get_lifecycle_config, versioning_sys::BucketVersioningSys}; +use crate::bucket::{metadata_sys, metadata_sys::get_lifecycle_config, versioning_sys::BucketVersioningSys}; use crate::client::object_api_utils::new_getobjectreader; +use crate::disk::error::DiskError; +use crate::disk::{DeleteOptions, Disk, DiskAPI, RUSTFS_META_MULTIPART_BUCKET, STORAGE_FORMAT_FILE}; use crate::error::Error; use crate::error::StorageError; use crate::error::{error_resp_to_object_err, is_err_object_not_found, is_err_version_not_found, is_network_or_host_down}; use crate::event_notification::{EventArgs, send_event}; use crate::global::GLOBAL_LocalNodeName; use crate::global::{GLOBAL_LifecycleSys, GLOBAL_TierConfigMgr, get_global_deployment_id}; +use crate::set_disk::{MAX_PARTS_COUNT, RUSTFS_MULTIPART_BUCKET_KEY, RUSTFS_MULTIPART_OBJECT_KEY, SetDisks}; use crate::store::ECStore; use crate::store_api::StorageAPI; use crate::store_api::{ - GetObjectReader, HTTPRangeSpec, ListOperations, ObjectInfo, ObjectOperations, ObjectOptions, ObjectToDelete, + GetObjectReader, HTTPRangeSpec, ListOperations, MultipartOperations, ObjectInfo, ObjectOperations, ObjectOptions, + ObjectToDelete, }; use crate::tier::warm_backend::WarmBackendGetOpts; use async_channel::{Receiver as A_Receiver, Sender as A_Sender, bounded}; @@ -45,7 +51,7 @@ use lazy_static::lazy_static; use rustfs_common::data_usage::TierStats; use rustfs_common::heal_channel::rep_has_active_rules; use rustfs_common::metrics::{IlmAction, Metrics}; -use rustfs_filemeta::{FileInfo, NULL_VERSION_ID, RestoreStatusOps, is_restored_object_on_disk}; +use rustfs_filemeta::{FileInfo, FileInfoOpts, NULL_VERSION_ID, RestoreStatusOps, get_file_info, is_restored_object_on_disk}; use rustfs_s3_common::EventName; use rustfs_utils::{get_env_i64, get_env_usize, path::encode_dir_object, string::strings_has_prefix_fold}; use s3s::Body; @@ -61,7 +67,8 @@ use std::env; use std::io::Write; use std::pin::Pin; use std::sync::atomic::{AtomicI64, Ordering}; -use std::sync::{Arc, Mutex}; +use std::sync::{Arc, Mutex, Weak}; +use std::time::Duration as StdDuration; use time::OffsetDateTime; use tokio::select; use tokio::sync::mpsc::{Receiver, Sender}; @@ -84,6 +91,10 @@ pub const AMZ_ENCRYPTION_AES: &str = "AES256"; pub const AMZ_ENCRYPTION_KMS: &str = "aws:kms"; pub const ERR_INVALID_STORAGECLASS: &str = "invalid tier."; +const ENV_STALE_UPLOADS_EXPIRY: &str = "RUSTFS_API_STALE_UPLOADS_EXPIRY"; +const ENV_STALE_UPLOADS_CLEANUP_INTERVAL: &str = "RUSTFS_API_STALE_UPLOADS_CLEANUP_INTERVAL"; +const DEFAULT_STALE_UPLOADS_EXPIRY: StdDuration = StdDuration::from_secs(24 * 60 * 60); +const DEFAULT_STALE_UPLOADS_CLEANUP_INTERVAL: StdDuration = StdDuration::from_secs(6 * 60 * 60); lazy_static! { pub static ref GLOBAL_ExpiryState: Arc> = ExpiryState::new(); @@ -686,6 +697,334 @@ pub async fn init_background_expiry(api: Arc) { ExpiryState::resize_workers(workers, api).await; } +#[derive(Debug, Clone)] +struct StaleMultipartUploadCandidate { + path: String, + initiated: OffsetDateTime, + metadata: Option>, +} + +fn parse_stale_uploads_duration(env_key: &str, default: StdDuration) -> StdDuration { + env::var(env_key) + .ok() + .and_then(|value| rustfs_madmin::utils::parse_duration(&value).ok()) + .filter(|duration| !duration.is_zero()) + .unwrap_or(default) +} + +fn stale_uploads_expiry() -> StdDuration { + parse_stale_uploads_duration(ENV_STALE_UPLOADS_EXPIRY, DEFAULT_STALE_UPLOADS_EXPIRY) +} + +fn stale_uploads_cleanup_interval() -> StdDuration { + parse_stale_uploads_duration(ENV_STALE_UPLOADS_CLEANUP_INTERVAL, DEFAULT_STALE_UPLOADS_CLEANUP_INTERVAL) +} + +fn encode_stale_upload_id(upload_uuid: &str) -> String { + base64_simd::URL_SAFE_NO_PAD + .encode_to_string(format!("{}.{}", get_global_deployment_id().unwrap_or_default(), upload_uuid).as_bytes()) +} + +fn initiated_from_upload_dir(upload_dir: &str, fallback: Option) -> OffsetDateTime { + upload_dir + .split_once('x') + .and_then(|(_, nanos)| nanos.parse::().ok()) + .and_then(|nanos| OffsetDateTime::from_unix_timestamp_nanos(nanos).ok()) + .or(fallback) + .unwrap_or_else(OffsetDateTime::now_utc) +} + +fn stale_upload_default_due(initiated: OffsetDateTime, default_expiry: StdDuration) -> OffsetDateTime { + initiated + time::Duration::seconds(default_expiry.as_secs() as i64) +} + +async fn stale_upload_current_size(set: &Arc, metadata: &HashMap, upload_dir: &str) -> Option { + let bucket = metadata.get(RUSTFS_MULTIPART_BUCKET_KEY)?; + let object = metadata.get(RUSTFS_MULTIPART_OBJECT_KEY)?; + let upload_id = encode_stale_upload_id(upload_dir); + let parts = set + .list_object_parts(bucket, object, &upload_id, None, MAX_PARTS_COUNT, &ObjectOptions::default()) + .await + .ok()?; + + Some( + parts + .parts + .iter() + .map(|part| part.actual_size.max(part.size as i64).max(0) as usize) + .sum(), + ) +} + +async fn stale_upload_lifecycle_due( + set: &Arc, + metadata: &HashMap, + initiated: OffsetDateTime, + upload_dir: &str, +) -> Option { + let bucket = metadata.get(RUSTFS_MULTIPART_BUCKET_KEY)?; + let object = metadata.get(RUSTFS_MULTIPART_OBJECT_KEY)?; + + let lifecycle = match metadata_sys::get_lifecycle_config(bucket).await { + Ok((lifecycle, _)) => lifecycle, + Err(_) => return None, + }; + + let object_opts = ObjectOpts { + name: object.clone(), + user_tags: metadata.get(AMZ_OBJECT_TAGGING).cloned().unwrap_or_default(), + mod_time: Some(initiated), + size: stale_upload_current_size(set, metadata, upload_dir).await.unwrap_or_default(), + is_latest: true, + delete_marker: false, + user_defined: metadata.clone(), + ..Default::default() + }; + + abort_incomplete_multipart_upload_due(&lifecycle, &object_opts) + .await + .map(|(due, _)| due) +} + +async fn read_stale_multipart_candidate( + disk: &Disk, + sha_dir: &str, + upload_dir: &str, +) -> Result { + let metadata_path = format!("{sha_dir}/{upload_dir}/{STORAGE_FORMAT_FILE}"); + let metadata_bytes = disk.read_metadata(RUSTFS_META_MULTIPART_BUCKET, &metadata_path).await?; + + let (metadata, mod_time) = match get_file_info( + &metadata_bytes, + RUSTFS_META_MULTIPART_BUCKET, + &metadata_path, + "", + FileInfoOpts { + data: false, + include_free_versions: false, + }, + ) { + Ok(file_info) => (Some(file_info.metadata), file_info.mod_time), + Err(err) => { + warn!(path = %metadata_path, error = ?err, "failed to parse multipart metadata during stale cleanup"); + (None, None) + } + }; + + let initiated = initiated_from_upload_dir(upload_dir, mod_time); + + Ok(StaleMultipartUploadCandidate { + path: format!("{sha_dir}/{upload_dir}"), + initiated, + metadata, + }) +} + +fn merge_stale_multipart_candidate( + candidates: &mut HashMap, + candidate: StaleMultipartUploadCandidate, +) { + match candidates.get(&candidate.path) { + Some(existing) if existing.metadata.is_some() => {} + Some(existing) if existing.metadata.is_none() && candidate.metadata.is_none() => {} + _ => { + candidates.insert(candidate.path.clone(), candidate); + } + } +} + +async fn cleanup_empty_multipart_sha_dirs_on_local_disks(set: &Arc) { + for disk in set.get_local_disks().await.into_iter().flatten() { + if !disk.is_online().await { + continue; + } + + let sha_dirs = match disk + .list_dir(RUSTFS_META_MULTIPART_BUCKET, RUSTFS_META_MULTIPART_BUCKET, "", -1) + .await + { + Ok(entries) => entries, + Err(err) => { + if err != DiskError::FileNotFound && err != DiskError::VolumeNotFound { + warn!(error = ?err, "failed to list multipart root during empty sha cleanup"); + } + continue; + } + }; + + for sha_dir in sha_dirs { + let sha_dir = sha_dir.trim_end_matches('/').to_string(); + let upload_dirs = match disk + .list_dir(RUSTFS_META_MULTIPART_BUCKET, RUSTFS_META_MULTIPART_BUCKET, &sha_dir, -1) + .await + { + Ok(entries) => entries, + Err(err) => { + if err != DiskError::FileNotFound && err != DiskError::VolumeNotFound { + warn!(sha_dir = %sha_dir, error = ?err, "failed to list multipart sha dir during empty sha cleanup"); + } + continue; + } + }; + + if !upload_dirs.is_empty() { + continue; + } + + if let Err(err) = disk + .delete(RUSTFS_META_MULTIPART_BUCKET, &sha_dir, DeleteOptions::default()) + .await + && err != DiskError::FileNotFound + && err != DiskError::VolumeNotFound + { + warn!(sha_dir = %sha_dir, error = ?err, "failed to remove empty multipart sha dir"); + } + } + } +} + +async fn cleanup_stale_multipart_uploads_in_set(set: &Arc, now: OffsetDateTime, default_expiry: StdDuration) -> usize { + let mut deleted = 0usize; + let mut candidates = HashMap::new(); + + for disk in set.get_local_disks().await.into_iter().flatten() { + if !disk.is_online().await { + continue; + } + + let sha_dirs = match disk + .list_dir(RUSTFS_META_MULTIPART_BUCKET, RUSTFS_META_MULTIPART_BUCKET, "", -1) + .await + { + Ok(entries) => entries, + Err(err) => { + if err != DiskError::FileNotFound && err != DiskError::VolumeNotFound { + warn!(error = ?err, "failed to list multipart root during stale cleanup"); + } + continue; + } + }; + + for sha_dir in sha_dirs { + let sha_dir = sha_dir.trim_end_matches('/').to_string(); + let upload_dirs = match disk + .list_dir(RUSTFS_META_MULTIPART_BUCKET, RUSTFS_META_MULTIPART_BUCKET, &sha_dir, -1) + .await + { + Ok(entries) => entries, + Err(err) => { + if err != DiskError::FileNotFound && err != DiskError::VolumeNotFound { + warn!(sha_dir = %sha_dir, error = ?err, "failed to list multipart sha dir during stale cleanup"); + } + continue; + } + }; + + for upload_dir in upload_dirs { + let upload_dir = upload_dir.trim_end_matches('/').to_string(); + let candidate_path = format!("{sha_dir}/{upload_dir}"); + if candidates + .get(&candidate_path) + .is_some_and(|existing: &StaleMultipartUploadCandidate| existing.metadata.is_some()) + { + continue; + } + + let candidate = match read_stale_multipart_candidate(disk.as_ref(), &sha_dir, &upload_dir).await { + Ok(candidate) => candidate, + Err(err) => { + if err != DiskError::FileNotFound { + warn!(path = %candidate_path, error = ?err, "failed to read multipart metadata during stale cleanup"); + } + let initiated = initiated_from_upload_dir(&upload_dir, None); + StaleMultipartUploadCandidate { + path: candidate_path, + initiated, + metadata: None, + } + } + }; + merge_stale_multipart_candidate(&mut candidates, candidate); + } + } + } + + for candidate in candidates.into_values() { + let upload_dir = candidate.path.rsplit('/').next().unwrap_or_default().to_string(); + let mut due = stale_upload_default_due(candidate.initiated, default_expiry); + if let Some(metadata) = candidate.metadata.as_ref() + && let Some(lifecycle_due) = stale_upload_lifecycle_due(set, metadata, candidate.initiated, &upload_dir).await + && lifecycle_due < due + { + due = lifecycle_due; + } + + if now < due { + continue; + } + + match set.delete_all(RUSTFS_META_MULTIPART_BUCKET, &candidate.path).await { + Ok(()) => { + deleted += 1; + let upload_id = encode_stale_upload_id(&upload_dir); + if let Some(metadata) = candidate.metadata.as_ref() { + info!( + bucket = metadata.get(RUSTFS_MULTIPART_BUCKET_KEY).cloned().unwrap_or_default(), + object = metadata.get(RUSTFS_MULTIPART_OBJECT_KEY).cloned().unwrap_or_default(), + upload_id = %upload_id, + due = ?due, + "removed stale multipart upload" + ); + } else { + info!(path = %candidate.path, upload_id = %upload_id, due = ?due, "removed stale multipart upload"); + } + } + Err(err) => warn!(path = %candidate.path, error = ?err, "failed to remove stale multipart upload"), + } + } + + cleanup_empty_multipart_sha_dirs_on_local_disks(set).await; + + deleted +} + +async fn cleanup_stale_multipart_uploads_once_at(api: Arc, now: OffsetDateTime, default_expiry: StdDuration) -> usize { + let mut deleted = 0usize; + for pool in &api.pools { + for set in &pool.disk_set { + deleted += cleanup_stale_multipart_uploads_in_set(set, now, default_expiry).await; + } + } + deleted +} + +pub async fn run_stale_multipart_upload_cleanup_once(api: Arc) -> usize { + cleanup_stale_multipart_uploads_once_at(api, OffsetDateTime::now_utc(), stale_uploads_expiry()).await +} + +pub fn init_background_stale_multipart_upload_cleanup(api: Arc) { + let cleanup_interval = stale_uploads_cleanup_interval(); + let default_expiry = stale_uploads_expiry(); + let api = Arc::downgrade(&api); + + tokio::spawn(async move { + let mut interval = tokio::time::interval(cleanup_interval); + + loop { + interval.tick().await; + + let Some(api) = Weak::upgrade(&api) else { + return; + }; + + let deleted = cleanup_stale_multipart_uploads_once_at(api, OffsetDateTime::now_utc(), default_expiry).await; + if deleted > 0 { + info!(deleted, "completed stale multipart cleanup pass"); + } + } + }); +} + pub async fn validate_transition_tier(lc: &BucketLifecycleConfiguration) -> Result<(), std::io::Error> { for rule in &lc.rules { if let Some(transitions) = &rule.transitions { @@ -1288,8 +1627,31 @@ pub async fn apply_lifecycle_action(event: &lifecycle::Event, src: &LcEventSrc, #[cfg(test)] mod tests { - use super::mark_delete_opts_skip_decommissioned_on_remote_success; - use crate::store_api::ObjectOptions; + use super::{ + StaleMultipartUploadCandidate, cleanup_empty_multipart_sha_dirs_on_local_disks, cleanup_stale_multipart_uploads_once_at, + mark_delete_opts_skip_decommissioned_on_remote_success, merge_stale_multipart_candidate, + }; + use crate::bucket::metadata::BUCKET_LIFECYCLE_CONFIG; + use crate::bucket::metadata_sys; + use crate::disk::RUSTFS_META_MULTIPART_BUCKET; + use crate::disk::endpoint::Endpoint; + use crate::endpoints::{EndpointServerPools, Endpoints, PoolEndpoints}; + use crate::error::is_err_invalid_upload_id; + use crate::set_disk::{RUSTFS_MULTIPART_BUCKET_KEY, RUSTFS_MULTIPART_OBJECT_KEY}; + use crate::store::ECStore; + use crate::store_api::{ + BucketOperations, BucketOptions, MakeBucketOptions, MultipartOperations, ObjectOptions, PutObjReader, + }; + use serial_test::serial; + use sha2::{Digest, Sha256}; + use std::collections::HashMap; + use std::path::PathBuf; + use std::sync::{Arc, OnceLock}; + use std::time::Duration as StdDuration; + use time::OffsetDateTime; + use tokio::fs; + use tokio_util::sync::CancellationToken; + use uuid::Uuid; #[test] fn mark_delete_opts_skip_decommissioned_on_remote_success_sets_flag_on_success() { @@ -1320,4 +1682,334 @@ mod tests { assert!(opts.skip_decommissioned); } + + static STALE_MULTIPART_TEST_ENV: OnceLock<(Vec, Arc)> = OnceLock::new(); + + async fn setup_test_env() -> (Vec, Arc) { + if let Some((paths, ecstore)) = STALE_MULTIPART_TEST_ENV.get() { + return (paths.clone(), ecstore.clone()); + } + + let test_base_dir = format!("/tmp/rustfs_stale_multipart_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 endpoint_pools = EndpointServerPools(vec![PoolEndpoints { + legacy: false, + set_count: 1, + drives_per_set: 4, + endpoints: Endpoints::from(endpoints), + cmd_line: "stale-multipart-test".to_string(), + platform: format!("OS: {} | Arch: {}", std::env::consts::OS, std::env::consts::ARCH), + }]); + + crate::store::init_local_disks(endpoint_pools.clone()).await.unwrap(); + + let ecstore = ECStore::new("127.0.0.1:0".parse().unwrap(), endpoint_pools, CancellationToken::new()) + .await + .unwrap(); + + let buckets = ecstore + .list_bucket(&BucketOptions { + no_metadata: true, + ..Default::default() + }) + .await + .unwrap() + .into_iter() + .map(|bucket| bucket.name) + .collect(); + metadata_sys::init_bucket_metadata_sys(ecstore.clone(), buckets).await; + + let _ = STALE_MULTIPART_TEST_ENV.set((disk_paths.clone(), ecstore.clone())); + + (disk_paths, ecstore) + } + + async fn create_test_bucket(ecstore: &Arc, bucket: &str) { + ecstore + .make_bucket(bucket, &MakeBucketOptions::default()) + .await + .expect("bucket should be created"); + } + + async fn set_abort_incomplete_lifecycle(bucket: &str, prefix: &str, days_after_initiation: i32) { + let lifecycle_xml = format!( + r#" + + + abort-multipart + Enabled + + {prefix} + + + {days_after_initiation} + + +"# + ); + + metadata_sys::update(bucket, BUCKET_LIFECYCLE_CONFIG, lifecycle_xml.into_bytes()) + .await + .expect("lifecycle metadata should be stored"); + } + + async fn set_abort_incomplete_lifecycle_with_size( + bucket: &str, + prefix: &str, + days_after_initiation: i32, + object_size_greater_than: usize, + ) { + let lifecycle_xml = format!( + r#" + + + abort-multipart-size + Enabled + + + {prefix} + {object_size_greater_than} + + + + {days_after_initiation} + + +"# + ); + + metadata_sys::update(bucket, BUCKET_LIFECYCLE_CONFIG, lifecycle_xml.into_bytes()) + .await + .expect("lifecycle metadata should be stored"); + } + + fn multipart_sha_dir(bucket: &str, object: &str) -> String { + hex_simd::encode_to_string(Sha256::digest(format!("{bucket}/{object}").as_bytes()), hex_simd::AsciiCase::Lower) + } + + #[test] + fn merge_stale_multipart_candidate_prefers_metadata_over_fallback() { + let mut candidates = HashMap::new(); + + merge_stale_multipart_candidate( + &mut candidates, + StaleMultipartUploadCandidate { + path: "sha/upload".to_string(), + initiated: OffsetDateTime::UNIX_EPOCH, + metadata: None, + }, + ); + merge_stale_multipart_candidate( + &mut candidates, + StaleMultipartUploadCandidate { + path: "sha/upload".to_string(), + initiated: OffsetDateTime::UNIX_EPOCH, + metadata: Some(HashMap::from([("k".to_string(), "v".to_string())])), + }, + ); + + assert_eq!( + candidates + .get("sha/upload") + .and_then(|candidate| candidate.metadata.as_ref()) + .and_then(|metadata| metadata.get("k")), + Some(&"v".to_string()) + ); + } + + #[tokio::test] + #[serial] + async fn stale_multipart_cleanup_uses_default_expiry_without_lifecycle() { + let (_paths, ecstore) = setup_test_env().await; + let bucket = format!("stale-default-{}", Uuid::new_v4().simple()); + let object = "default-cleanup/object.txt"; + create_test_bucket(&ecstore, &bucket).await; + + let initiated = OffsetDateTime::now_utc() - time::Duration::hours(30); + let upload = ecstore + .new_multipart_upload( + &bucket, + object, + &ObjectOptions { + mod_time: Some(initiated), + ..Default::default() + }, + ) + .await + .expect("multipart upload should be created"); + + let deleted = cleanup_stale_multipart_uploads_once_at( + ecstore.clone(), + OffsetDateTime::now_utc(), + StdDuration::from_secs(24 * 60 * 60), + ) + .await; + assert!(deleted >= 1, "expected at least one stale multipart upload to be removed"); + + let err = ecstore + .get_multipart_info(&bucket, object, &upload.upload_id, &ObjectOptions::default()) + .await + .expect_err("stale multipart upload should be removed"); + assert!(is_err_invalid_upload_id(&err)); + } + + #[tokio::test] + #[serial] + async fn stale_multipart_cleanup_applies_abort_incomplete_lifecycle_before_default_expiry() { + let (_paths, ecstore) = setup_test_env().await; + let bucket = format!("stale-lifecycle-{}", Uuid::new_v4().simple()); + let object = "logs/prefix/object.txt"; + create_test_bucket(&ecstore, &bucket).await; + set_abort_incomplete_lifecycle(&bucket, "logs/", 1).await; + + let initiated = OffsetDateTime::now_utc() - time::Duration::hours(48); + let upload = ecstore + .new_multipart_upload( + &bucket, + object, + &ObjectOptions { + mod_time: Some(initiated), + ..Default::default() + }, + ) + .await + .expect("multipart upload should be created"); + + let deleted = cleanup_stale_multipart_uploads_once_at( + ecstore.clone(), + OffsetDateTime::now_utc(), + StdDuration::from_secs(7 * 24 * 60 * 60), + ) + .await; + assert!(deleted >= 1, "expected lifecycle-driven stale multipart cleanup to run"); + + let err = ecstore + .get_multipart_info(&bucket, object, &upload.upload_id, &ObjectOptions::default()) + .await + .expect_err("multipart upload should be removed by lifecycle abort rule"); + assert!(is_err_invalid_upload_id(&err)); + } + + #[tokio::test] + #[serial] + async fn stale_multipart_cleanup_applies_abort_lifecycle_with_size_filter() { + let (_paths, ecstore) = setup_test_env().await; + let bucket = format!("stale-size-{}", Uuid::new_v4().simple()); + let object = "logs/sized/object.txt"; + create_test_bucket(&ecstore, &bucket).await; + set_abort_incomplete_lifecycle_with_size(&bucket, "logs/", 1, 5).await; + + let initiated = OffsetDateTime::now_utc() - time::Duration::hours(48); + let upload = ecstore + .new_multipart_upload( + &bucket, + object, + &ObjectOptions { + mod_time: Some(initiated), + ..Default::default() + }, + ) + .await + .expect("multipart upload should be created"); + + let mut data = PutObjReader::from_vec(vec![1, 2, 3, 4, 5, 6]); + ecstore + .put_object_part(&bucket, object, &upload.upload_id, 1, &mut data, &ObjectOptions::default()) + .await + .expect("multipart part should be uploaded"); + + let deleted = cleanup_stale_multipart_uploads_once_at( + ecstore.clone(), + OffsetDateTime::now_utc(), + StdDuration::from_secs(7 * 24 * 60 * 60), + ) + .await; + assert!(deleted >= 1, "expected lifecycle-driven stale multipart cleanup to run"); + + let err = ecstore + .get_multipart_info(&bucket, object, &upload.upload_id, &ObjectOptions::default()) + .await + .expect_err("multipart upload should be removed by size-qualified lifecycle abort rule"); + assert!(is_err_invalid_upload_id(&err)); + } + + #[tokio::test] + #[serial] + async fn multipart_info_and_list_parts_do_not_expose_internal_metadata_keys() { + let (_paths, ecstore) = setup_test_env().await; + let bucket = format!("stale-sanitize-{}", Uuid::new_v4().simple()); + let object = "sanitize/object.txt"; + create_test_bucket(&ecstore, &bucket).await; + + let upload = ecstore + .new_multipart_upload(&bucket, object, &ObjectOptions::default()) + .await + .expect("multipart upload should be created"); + + let multipart_info = ecstore + .get_multipart_info(&bucket, object, &upload.upload_id, &ObjectOptions::default()) + .await + .expect("multipart info should be readable"); + assert!(!multipart_info.user_defined.contains_key(RUSTFS_MULTIPART_BUCKET_KEY)); + assert!(!multipart_info.user_defined.contains_key(RUSTFS_MULTIPART_OBJECT_KEY)); + + let parts = ecstore + .list_object_parts(&bucket, object, &upload.upload_id, None, 0, &ObjectOptions::default()) + .await + .expect("multipart parts should be readable"); + assert!(!parts.user_defined.contains_key(RUSTFS_MULTIPART_BUCKET_KEY)); + assert!(!parts.user_defined.contains_key(RUSTFS_MULTIPART_OBJECT_KEY)); + } + + #[tokio::test] + #[serial] + async fn cleanup_removes_empty_multipart_sha_dirs() { + let (paths, ecstore) = setup_test_env().await; + let bucket = format!("stale-empty-sha-{}", Uuid::new_v4().simple()); + let object = "empty-sha/object.txt"; + let sha_dir = multipart_sha_dir(&bucket, object); + for path in &paths { + fs::create_dir_all(path.join(RUSTFS_META_MULTIPART_BUCKET).join(&sha_dir)) + .await + .expect("empty multipart sha dir should be created for cleanup"); + assert!( + path.join(RUSTFS_META_MULTIPART_BUCKET).join(&sha_dir).exists(), + "empty multipart sha dir should exist before cleanup" + ); + } + + cleanup_empty_multipart_sha_dirs_on_local_disks(&ecstore.pools[0].disk_set[0]).await; + + for path in &paths { + assert!( + !path.join(RUSTFS_META_MULTIPART_BUCKET).join(&sha_dir).exists(), + "empty multipart sha dir should be removed" + ); + } + } } diff --git a/crates/ecstore/src/bucket/lifecycle/lifecycle.rs b/crates/ecstore/src/bucket/lifecycle/lifecycle.rs index 7832d6893..395fa063a 100644 --- a/crates/ecstore/src/bucket/lifecycle/lifecycle.rs +++ b/crates/ecstore/src/bucket/lifecycle/lifecycle.rs @@ -806,6 +806,26 @@ pub fn expected_expiry_time(mod_time: OffsetDateTime, days: i32) -> OffsetDateTi t } +pub async fn abort_incomplete_multipart_upload_due( + lc: &BucketLifecycleConfiguration, + obj: &ObjectOpts, +) -> Option<(OffsetDateTime, String)> { + let initiated = obj.mod_time?; + let rules = lc.filter_rules(obj).await?; + + rules + .into_iter() + .filter_map(|rule| { + let days = rule + .abort_incomplete_multipart_upload + .as_ref()? + .days_after_initiation + .filter(|days| *days > 0)?; + Some((expected_expiry_time(initiated, days), rule.id.clone().unwrap_or_default())) + }) + .min_by_key(|(due, _)| due.unix_timestamp_nanos()) +} + #[derive(Default)] pub struct ObjectOpts { pub name: String, diff --git a/crates/ecstore/src/set_disk.rs b/crates/ecstore/src/set_disk.rs index b474d0c59..29877ced6 100644 --- a/crates/ecstore/src/set_disk.rs +++ b/crates/ecstore/src/set_disk.rs @@ -184,6 +184,13 @@ use uuid::Uuid; pub const DEFAULT_READ_BUFFER_SIZE: usize = MI_B; // 1 MiB = 1024 * 1024; pub const MAX_PARTS_COUNT: usize = 10000; +pub(crate) const RUSTFS_MULTIPART_BUCKET_KEY: &str = "x-rustfs-internal-multipart-bucket"; +pub(crate) const RUSTFS_MULTIPART_OBJECT_KEY: &str = "x-rustfs-internal-multipart-object"; + +pub(crate) fn strip_internal_multipart_metadata(metadata: &mut HashMap) { + metadata.remove(RUSTFS_MULTIPART_BUCKET_KEY); + metadata.remove(RUSTFS_MULTIPART_OBJECT_KEY); +} /// Get the duplex buffer size from environment variable or use default. /// @@ -2691,7 +2698,11 @@ impl MultipartOperations for SetDisks { storage_class, max_parts, part_number_marker, - user_defined: fi.metadata.clone(), + user_defined: { + let mut metadata = fi.metadata.clone(); + strip_internal_multipart_metadata(&mut metadata); + metadata + }, ..Default::default() }; @@ -3024,6 +3035,9 @@ impl MultipartOperations for SetDisks { ); } + user_defined.insert(RUSTFS_MULTIPART_BUCKET_KEY.to_string(), bucket.to_string()); + user_defined.insert(RUSTFS_MULTIPART_OBJECT_KEY.to_string(), object.to_string()); + let (shuffle_disks, mut parts_metadatas) = Self::shuffle_disks_and_parts_metadata(&disks, &parts_metadata, &fi); let mod_time = opts.mod_time.unwrap_or(OffsetDateTime::now_utc()); @@ -3072,7 +3086,7 @@ impl MultipartOperations for SetDisks { _opts: &ObjectOptions, ) -> Result { // TODO: nslock - let (fi, _) = self + let (mut fi, _) = self .check_upload_id_exists(bucket, object, upload_id, false) .await .map_err(|e| to_object_err(e, vec![bucket, object, upload_id]))?; @@ -3081,7 +3095,10 @@ impl MultipartOperations for SetDisks { bucket: bucket.to_owned(), object: object.to_owned(), upload_id: upload_id.to_owned(), - user_defined: fi.metadata.clone(), + user_defined: { + strip_internal_multipart_metadata(&mut fi.metadata); + fi.metadata.clone() + }, ..Default::default() }) } @@ -3372,6 +3389,7 @@ impl MultipartOperations for SetDisks { fi.metadata.remove(rustfs_rio::RUSTFS_MULTIPART_CHECKSUM); fi.metadata.remove(rustfs_rio::RUSTFS_MULTIPART_CHECKSUM_TYPE); + strip_internal_multipart_metadata(&mut fi.metadata); fi.size = object_size as i64; fi.mod_time = opts.mod_time; diff --git a/crates/ecstore/src/store/init.rs b/crates/ecstore/src/store/init.rs index e01b7bdb6..457e7a2cf 100644 --- a/crates/ecstore/src/store/init.rs +++ b/crates/ecstore/src/store/init.rs @@ -344,6 +344,7 @@ impl ECStore { init_global_bucket_monitor(num_nodes); init_background_expiry(self.clone()).await; + crate::bucket::lifecycle::bucket_lifecycle_ops::init_background_stale_multipart_upload_cleanup(self.clone()); TransitionState::init(self.clone()).await; crate::tier::tier::try_migrate_tiering_config(self.clone()).await;