feat(ecstore): add stale multipart upload cleanup (#2416)

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<K, V>(&mut self, key: K, value: V)
+    where
+        K: Into<String>,
+        V: Into<String>,
+    {
+        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);
         }

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;

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<bool, Box<dyn std::error::Error + Send + Sync>> {
+    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<bool, Box<dyn std::error::Error + Send + Sync>> {
+    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<dyn std::error::Error + Send + Sync>> {
+    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<dyn std::error::Error + Send + Sync>> {
+    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(())
+}

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<RwLock<ExpiryState>> = ExpiryState::new();
@@ -686,6 +697,334 @@ pub async fn init_background_expiry(api: Arc<ECStore>) {
     ExpiryState::resize_workers(workers, api).await;
 }
 
+#[derive(Debug, Clone)]
+struct StaleMultipartUploadCandidate {
+    path: String,
+    initiated: OffsetDateTime,
+    metadata: Option<HashMap<String, String>>,
+}
+
+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>) -> OffsetDateTime {
+    upload_dir
+        .split_once('x')
+        .and_then(|(_, nanos)| nanos.parse::<i128>().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<SetDisks>, metadata: &HashMap<String, String>, upload_dir: &str) -> Option<usize> {
+    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<SetDisks>,
+    metadata: &HashMap<String, String>,
+    initiated: OffsetDateTime,
+    upload_dir: &str,
+) -> Option<OffsetDateTime> {
+    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<StaleMultipartUploadCandidate, DiskError> {
+    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<String, StaleMultipartUploadCandidate>,
+    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<SetDisks>) {
+    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<SetDisks>, 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<ECStore>, 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<ECStore>) -> 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<ECStore>) {
+    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<PathBuf>, Arc<ECStore>)> = OnceLock::new();
+
+    async fn setup_test_env() -> (Vec<PathBuf>, Arc<ECStore>) {
+        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<ECStore>, 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#"<?xml version="1.0" encoding="UTF-8"?>
+<LifecycleConfiguration>
+  <Rule>
+    <ID>abort-multipart</ID>
+    <Status>Enabled</Status>
+    <Filter>
+      <Prefix>{prefix}</Prefix>
+    </Filter>
+    <AbortIncompleteMultipartUpload>
+      <DaysAfterInitiation>{days_after_initiation}</DaysAfterInitiation>
+    </AbortIncompleteMultipartUpload>
+  </Rule>
+</LifecycleConfiguration>"#
+        );
+
+        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#"<?xml version="1.0" encoding="UTF-8"?>
+<LifecycleConfiguration>
+  <Rule>
+    <ID>abort-multipart-size</ID>
+    <Status>Enabled</Status>
+    <Filter>
+      <And>
+        <Prefix>{prefix}</Prefix>
+        <ObjectSizeGreaterThan>{object_size_greater_than}</ObjectSizeGreaterThan>
+      </And>
+    </Filter>
+    <AbortIncompleteMultipartUpload>
+      <DaysAfterInitiation>{days_after_initiation}</DaysAfterInitiation>
+    </AbortIncompleteMultipartUpload>
+  </Rule>
+</LifecycleConfiguration>"#
+        );
+
+        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"
+            );
+        }
+    }
 }

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,

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<String, String>) {
+    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<MultipartInfo> {
         // 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;

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;