From fa235e9018d37786cc7b4db971e356eb565beed4 Mon Sep 17 00:00:00 2001 From: cxymds Date: Fri, 24 Jul 2026 21:28:06 +0800 Subject: [PATCH] fix(s3): list multipart uploads by bucket prefix (#5195) * fix(s3): list multipart uploads by bucket prefix * fix(s3): preserve exact-key crash reclamation --- crates/ecstore/src/bucket/utils.rs | 11 +- crates/ecstore/src/lib.rs | 1 + crates/ecstore/src/multipart_listing.rs | 102 ++++ crates/ecstore/src/set_disk/ops/multipart.rs | 488 ++++++++++++++----- crates/ecstore/src/store/multipart.rs | 103 ++-- rustfs/src/storage/s3_api/multipart.rs | 6 + 6 files changed, 556 insertions(+), 155 deletions(-) create mode 100644 crates/ecstore/src/multipart_listing.rs diff --git a/crates/ecstore/src/bucket/utils.rs b/crates/ecstore/src/bucket/utils.rs index f574408d3..a187d6111 100644 --- a/crates/ecstore/src/bucket/utils.rs +++ b/crates/ecstore/src/bucket/utils.rs @@ -310,10 +310,8 @@ pub fn check_list_multipart_args( ) -> Result<()> { check_list_objs_args(bucket, prefix, key_marker)?; - if let Some(upload_id_marker) = upload_id_marker { - if let Some(key_marker) = key_marker - && key_marker.ends_with('/') - { + if let (Some(key_marker), Some(upload_id_marker)) = (key_marker, upload_id_marker) { + if key_marker.ends_with('/') { return Err(StorageError::InvalidUploadIDKeyCombination( upload_id_marker.to_string(), key_marker.to_string(), @@ -629,6 +627,11 @@ mod tests { assert!(check_list_objs_args("INVALID", "", &None).is_err()); } + #[test] + fn test_list_multipart_upload_marker_is_ignored_without_key_marker() { + assert!(check_list_multipart_args("valid-bucket", "", &None, &Some("not-base64!".to_string()), &None,).is_ok()); + } + #[test] fn test_check_multipart_args() { assert!(check_new_multipart_args("valid-bucket", "valid-object").is_ok()); diff --git a/crates/ecstore/src/lib.rs b/crates/ecstore/src/lib.rs index a30223c03..79e779ba8 100644 --- a/crates/ecstore/src/lib.rs +++ b/crates/ecstore/src/lib.rs @@ -45,6 +45,7 @@ mod erasure; mod error; mod io_support; pub(crate) mod layout; +mod multipart_listing; mod object_api; mod runtime; mod services; diff --git a/crates/ecstore/src/multipart_listing.rs b/crates/ecstore/src/multipart_listing.rs new file mode 100644 index 000000000..e6613b693 --- /dev/null +++ b/crates/ecstore/src/multipart_listing.rs @@ -0,0 +1,102 @@ +// 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::storage_api_contracts::multipart::MultipartInfo; + +enum MultipartListingEntry { + Upload(MultipartInfo), + CommonPrefix(String), +} + +impl MultipartListingEntry { + fn key(&self) -> &str { + match self { + Self::Upload(upload) => &upload.object, + Self::CommonPrefix(prefix) => prefix, + } + } + + fn upload_id(&self) -> &str { + match self { + Self::Upload(upload) => &upload.upload_id, + Self::CommonPrefix(_) => "", + } + } +} + +pub(crate) struct MultipartListingPage { + pub(crate) uploads: Vec, + pub(crate) common_prefixes: Vec, + pub(crate) is_truncated: bool, + pub(crate) next_key_marker: Option, + pub(crate) next_upload_id_marker: Option, +} + +pub(crate) fn paginate_multipart_listing( + uploads: Vec, + common_prefixes: Vec, + key_marker: Option<&str>, + upload_id_marker: Option<&str>, + max_uploads: usize, + source_truncated: bool, +) -> MultipartListingPage { + let mut entries = uploads + .into_iter() + .map(MultipartListingEntry::Upload) + .chain(common_prefixes.into_iter().map(MultipartListingEntry::CommonPrefix)) + .filter(|entry| match key_marker { + None => true, + Some(key_marker) => match entry { + MultipartListingEntry::CommonPrefix(prefix) => prefix.as_str() > key_marker, + MultipartListingEntry::Upload(upload) => { + upload.object.as_str() > key_marker + || (upload.object == key_marker + && upload_id_marker.is_some_and(|marker| upload.upload_id.as_str() > marker)) + } + }, + }) + .collect::>(); + entries.sort_by(|left, right| { + left.key() + .cmp(right.key()) + .then_with(|| left.upload_id().cmp(right.upload_id())) + }); + + let is_truncated = source_truncated || entries.len() > max_uploads; + entries.truncate(max_uploads); + + let (next_key_marker, next_upload_id_marker) = if is_truncated { + entries.last().map_or((None, None), |entry| match entry { + MultipartListingEntry::Upload(upload) => (Some(upload.object.clone()), Some(upload.upload_id.clone())), + MultipartListingEntry::CommonPrefix(prefix) => (Some(prefix.clone()), None), + }) + } else { + (None, None) + }; + + let mut page = MultipartListingPage { + uploads: Vec::new(), + common_prefixes: Vec::new(), + is_truncated, + next_key_marker, + next_upload_id_marker, + }; + for entry in entries { + match entry { + MultipartListingEntry::Upload(upload) => page.uploads.push(upload), + MultipartListingEntry::CommonPrefix(prefix) => page.common_prefixes.push(prefix), + } + } + page +} diff --git a/crates/ecstore/src/set_disk/ops/multipart.rs b/crates/ecstore/src/set_disk/ops/multipart.rs index 922c2e532..37d4d6209 100644 --- a/crates/ecstore/src/set_disk/ops/multipart.rs +++ b/crates/ecstore/src/set_disk/ops/multipart.rs @@ -23,10 +23,14 @@ use super::super::*; use super::bitrot_self_verify::{BitrotSelfVerifyTarget, drop_failed_writer_disks, verify_written_bitrot_shards}; use crate::crash_inject::{self, CrashPoint}; +use crate::multipart_listing::paginate_multipart_listing; +use futures::{StreamExt, stream}; use std::future::Future; use std::time::Duration; use tokio::task::JoinSet; +const MULTIPART_LIST_IO_CONCURRENCY: usize = 16; + #[cfg(test)] #[derive(Clone, Copy, PartialEq, Eq)] enum MultipartCommitPause { @@ -236,6 +240,51 @@ fn paginate_upload_page(remaining: &[MultipartInfo], max_uploads: usize) -> (Vec (page, is_truncated, next_upload_id_marker) } +async fn multipart_upload_paths_on_disk(disk: DiskStore, bucket: &str) -> disk::error::Result> { + if !disk.is_online().await { + return Err(DiskError::DiskNotFound); + } + + let sha_dirs = match disk.list_dir(bucket, RUSTFS_META_MULTIPART_BUCKET, "", -1).await { + Ok(entries) => entries, + Err(DiskError::FileNotFound | DiskError::VolumeNotFound) => return Ok(Vec::new()), + Err(err) => return Err(err), + }; + + let sha_dirs = sha_dirs + .into_iter() + .map(|sha_dir| sha_dir.trim_end_matches('/').to_owned()) + .filter(|sha_dir| sha_dir.len() == 64 && sha_dir.bytes().all(|byte| byte.is_ascii_hexdigit())) + .collect::>(); + let listed_upload_dirs = stream::iter(sha_dirs) + .map(|sha_dir| { + let disk = disk.clone(); + async move { + match disk.list_dir(bucket, RUSTFS_META_MULTIPART_BUCKET, &sha_dir, -1).await { + Ok(entries) => Ok((sha_dir, entries)), + Err(DiskError::FileNotFound) => Ok((sha_dir, Vec::new())), + Err(err) => Err(err), + } + } + }) + .buffer_unordered(MULTIPART_LIST_IO_CONCURRENCY) + .collect::>() + .await; + + let mut upload_paths = Vec::new(); + for listed_upload_dirs in listed_upload_dirs { + let (sha_dir, upload_dirs) = listed_upload_dirs?; + upload_paths.reserve(upload_dirs.len()); + upload_paths.extend(upload_dirs.into_iter().filter_map(|upload_dir| { + let upload_dir = upload_dir.trim_end_matches('/'); + (!upload_dir.is_empty() && upload_dir != "." && upload_dir != ".." && !upload_dir.contains(['/', '\\'])) + .then(|| format!("{sha_dir}/{upload_dir}")) + })); + } + + Ok(upload_paths) +} + impl SetDisks { async fn acquire_multipart_upload_read_lock( &self, @@ -843,131 +892,184 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks { async fn list_multipart_uploads( &self, bucket: &str, - object: &str, + prefix: &str, key_marker: Option, upload_id_marker: Option, delimiter: Option, max_uploads: usize, ) -> Result { - let disks = { - let disks = self.get_online_local_disks().await; - if disks.is_empty() { - // TODO: getOnlineDisksWithHealing - self.get_online_disks().await - } else { - disks - } + let disks = self.disks.read().await.clone(); + if disks.is_empty() { + return Err(Error::ErasureReadQuorum); + } + let discovery_quorum = if self.default_parity_count == 0 { + disks.len() + } else { + (disks.len() / 2).max(1) }; - - let mut upload_ids: Vec = Vec::new(); - - for disk in disks.iter().flatten() { - if !disk.is_online().await { - continue; - } - - let has_uoload_ids = match disk - .list_dir( - bucket, - RUSTFS_META_MULTIPART_BUCKET, - Self::get_multipart_sha_dir(bucket, object).as_str(), - -1, - ) - .await - { - Ok(res) => Some(res), - Err(err) => { - if err == DiskError::DiskNotFound { - None - } else if err == DiskError::FileNotFound { - return Ok(ListMultipartsInfo { - key_marker: key_marker.to_owned(), - max_uploads, - prefix: object.to_owned(), - delimiter: delimiter.to_owned(), - ..Default::default() - }); - } else { - return Err(to_object_err(err.into(), vec![bucket, object])); - } - } - }; - - if let Some(ids) = has_uoload_ids { - upload_ids = ids; - break; - } - } - - let mut uploads = Vec::new(); - - let mut populated_upload_ids = HashSet::new(); - - for upload_id in upload_ids.iter() { - let upload_id = upload_id.trim_end_matches(SLASH_SEPARATOR).to_string(); - if populated_upload_ids.contains(&upload_id) { - continue; - } - - let start_time = { - let now = OffsetDateTime::now_utc(); - - let splits: Vec<&str> = upload_id.split("x").collect(); - if splits.len() == 2 { - if let Ok(unix) = splits[1].parse::() { - OffsetDateTime::from_unix_timestamp_nanos(unix)? - } else { - now - } - } else { - now - } - }; - - uploads.push(MultipartInfo { - bucket: bucket.to_owned(), - object: object.to_owned(), - upload_id: runtime_sources::deployment_upload_id(&upload_id), - initiated: Some(start_time), - ..Default::default() + let mut discovery_errors = (0..disks.len()).map(|_| Some(DiskError::DiskNotFound)).collect::>(); + let mut candidate_counts = HashMap::::new(); + let mut discovery_tasks = JoinSet::new(); + for (index, disk) in disks.iter().enumerate() { + let disk = disk.clone(); + let bucket = bucket.to_string(); + discovery_tasks.spawn(async move { + let result = match disk { + Some(disk) => multipart_upload_paths_on_disk(disk, &bucket).await, + None => Err(DiskError::DiskNotFound), + }; + (index, result) }); - - populated_upload_ids.insert(upload_id); } - uploads.sort_by_key(|a| a.initiated); - - let mut upload_idx = 0; - if let Some(upload_id_marker) = &upload_id_marker { - while upload_idx < uploads.len() { - if &uploads[upload_idx].upload_id != upload_id_marker { - upload_idx += 1; - continue; + while let Some(task_result) = discovery_tasks.join_next().await { + let Ok((index, result)) = task_result else { + continue; + }; + match result { + Ok(paths) => { + discovery_errors[index] = None; + for path in paths { + *candidate_counts.entry(path).or_insert(0) += 1; + } } - - if &uploads[upload_idx].upload_id == upload_id_marker { - upload_idx += 1; - break; - } - - upload_idx += 1; + Err(err) => discovery_errors[index] = Some(err), } } - // `upload_idx` is the post-marker offset, so `uploads[upload_idx..]` is the - // page candidate. The helper applies the `max_uploads` cap, using the first - // overflow element only as a truncation probe (never returned). - let (ret_uploads, is_truncated, next_upload_id_marker) = paginate_upload_page(&uploads[upload_idx..], max_uploads); + if let Some(err) = reduce_read_quorum_errs(&discovery_errors, OBJECT_OP_IGNORED_ERRS, discovery_quorum) { + return Err(to_object_err(err.into(), vec![bucket, prefix])); + } + + let candidate_paths = candidate_counts + .into_iter() + .filter_map(|(path, count)| (count >= discovery_quorum).then_some(path)) + .collect::>(); + let listed_uploads = stream::iter(candidate_paths) + .map(|upload_path| { + let disks = &disks; + async move { + let (sha_dir, raw_upload_id) = upload_path + .rsplit_once('/') + .filter(|(sha_dir, upload_id)| !sha_dir.is_empty() && !upload_id.is_empty()) + .ok_or(DiskError::CorruptedFormat)?; + let (parts_metadata, errs) = Self::read_all_fileinfo( + disks, + bucket, + RUSTFS_META_MULTIPART_BUCKET, + &upload_path, + "", + false, + false, + false, + ) + .await?; + let missing_metadata = errs + .iter() + .filter(|err| matches!(err, Some(DiskError::FileNotFound | DiskError::VolumeNotFound))) + .count(); + if missing_metadata >= discovery_quorum { + // Completion moves the authoritative upload metadata into the + // committed object before it removes the staging directory. A + // crash in that window intentionally leaves a reclaimable + // upload directory whose object name can still be proven for + // an exact-key listing by matching the namespace hash. + if !prefix.is_empty() && sha_dir == Self::get_multipart_sha_dir(bucket, prefix) { + let initiated = raw_upload_id + .rsplit_once('x') + .and_then(|(_, timestamp)| timestamp.parse::().ok()) + .and_then(|timestamp| OffsetDateTime::from_unix_timestamp_nanos(timestamp).ok()); + return Ok(Some(MultipartInfo { + bucket: bucket.to_owned(), + object: prefix.to_owned(), + upload_id: runtime_sources::deployment_upload_id(raw_upload_id), + initiated, + ..Default::default() + })); + } + return Ok(None); + } + let (read_quorum, _) = Self::object_quorum_from_meta(&parts_metadata, &errs, self.default_parity_count)?; + let read_quorum = usize::try_from(read_quorum).map_err(|_| DiskError::ErasureReadQuorum)?; + if let Some(err) = reduce_read_quorum_errs(&errs, OBJECT_OP_IGNORED_ERRS, read_quorum) { + return Err(err); + } + let (_, mod_time, etag) = Self::list_online_disks(disks, &parts_metadata, &errs, read_quorum); + let file_info = Self::pick_valid_fileinfo(&parts_metadata, mod_time, etag, read_quorum)?; + + let object = match ( + file_info.metadata.get(RUSTFS_MULTIPART_BUCKET_KEY), + file_info.metadata.get(RUSTFS_MULTIPART_OBJECT_KEY), + ) { + (Some(stored_bucket), Some(object)) if stored_bucket == bucket && !object.is_empty() => object.clone(), + _ => return Err(DiskError::CorruptedFormat), + }; + if !object.starts_with(prefix) { + return Ok(None); + } + + let initiated = raw_upload_id + .rsplit_once('x') + .and_then(|(_, timestamp)| timestamp.parse::().ok()) + .and_then(|timestamp| OffsetDateTime::from_unix_timestamp_nanos(timestamp).ok()) + .or(file_info.mod_time); + + Ok(Some(MultipartInfo { + bucket: bucket.to_owned(), + object, + upload_id: runtime_sources::deployment_upload_id(raw_upload_id), + initiated, + ..Default::default() + })) + } + }) + .buffer_unordered(MULTIPART_LIST_IO_CONCURRENCY) + .collect::>>>() + .await; + + let mut uploads = Vec::with_capacity(listed_uploads.len()); + for result in listed_uploads { + if let Some(upload) = result.map_err(Error::from)? { + uploads.push(upload); + } + } + + let mut common_prefixes = HashSet::new(); + let mut unfolded_uploads = Vec::with_capacity(uploads.len()); + let delimiter_value = delimiter.as_deref().filter(|delimiter| !delimiter.is_empty()); + for upload in uploads { + let Some(delimiter) = delimiter_value else { + unfolded_uploads.push(upload); + continue; + }; + let suffix = upload.object.strip_prefix(prefix).ok_or(DiskError::CorruptedFormat)?; + if let Some((common_prefix, _)) = suffix.split_once(delimiter) { + common_prefixes.insert(format!("{prefix}{common_prefix}{delimiter}")); + } else { + unfolded_uploads.push(upload); + } + } + + let page = paginate_multipart_listing( + unfolded_uploads, + common_prefixes.into_iter().collect(), + key_marker.as_deref(), + key_marker.as_ref().and(upload_id_marker.as_deref()), + max_uploads, + false, + ); Ok(ListMultipartsInfo { key_marker: key_marker.to_owned(), - next_upload_id_marker, + upload_id_marker: upload_id_marker.to_owned(), + next_key_marker: page.next_key_marker, + next_upload_id_marker: page.next_upload_id_marker, max_uploads, - is_truncated, - uploads: ret_uploads, - prefix: object.to_owned(), + is_truncated: page.is_truncated, + uploads: page.uploads, + common_prefixes: page.common_prefixes, + prefix: prefix.to_owned(), delimiter: delimiter.to_owned(), - ..Default::default() }) } @@ -3112,11 +3214,12 @@ mod tests { // Paginating one upload at a time must enumerate every upload exactly once, // with no gaps and no duplicates, and terminate. let mut seen = HashSet::new(); - let mut marker: Option = None; + let mut key_marker: Option = None; + let mut upload_id_marker: Option = None; let mut pages = 0usize; loop { let page = set_disks - .list_multipart_uploads(bucket, object, None, marker.clone(), None, 1) + .list_multipart_uploads(bucket, object, key_marker.clone(), upload_id_marker.clone(), None, 1) .await .expect("list should succeed"); assert!(page.uploads.len() <= 1, "max_uploads=1 must never return more than one upload"); @@ -3132,8 +3235,10 @@ mod tests { if !page.is_truncated { break; } - marker = page.next_upload_id_marker.clone(); - assert!(marker.is_some(), "a truncated page must carry a next marker to continue"); + key_marker = page.next_key_marker.clone(); + upload_id_marker = page.next_upload_id_marker.clone(); + assert!(key_marker.is_some(), "a truncated page must carry a next key marker"); + assert!(upload_id_marker.is_some(), "a truncated upload page must carry a next upload marker"); } assert_eq!( seen, created, @@ -3141,6 +3246,167 @@ mod tests { ); } + #[tokio::test] + async fn list_multipart_uploads_enumerates_bucket_and_prefix() { + let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; + let bucket = "multipart-prefix-list-bucket"; + for disk in &disk_stores { + disk.make_volume(bucket).await.expect("bucket volume should be created"); + } + + let mut expected = Vec::new(); + for object in ["logs/a.bin", "logs/a.bin", "logs/b.bin", "other/c.bin"] { + let upload = set_disks + .new_multipart_upload(bucket, object, &ObjectOptions::default()) + .await + .expect("multipart upload should be created"); + expected.push((object.to_string(), upload.upload_id)); + } + expected.sort(); + + let all = set_disks + .list_multipart_uploads(bucket, "", None, None, None, 1000) + .await + .expect("bucket-wide multipart listing should succeed"); + let listed = all + .uploads + .iter() + .map(|upload| (upload.object.clone(), upload.upload_id.clone())) + .collect::>(); + assert_eq!(listed, expected); + assert!(!all.is_truncated); + + let logs = set_disks + .list_multipart_uploads(bucket, "logs/", None, None, None, 1000) + .await + .expect("prefix multipart listing should succeed"); + assert_eq!(logs.uploads.len(), 3); + assert!(logs.uploads.iter().all(|upload| upload.object.starts_with("logs/"))); + + let exact = set_disks + .list_multipart_uploads(bucket, "logs/a.bin", None, None, None, 1000) + .await + .expect("exact-key multipart listing should remain supported"); + assert_eq!(exact.uploads.len(), 2); + assert!(exact.uploads.iter().all(|upload| upload.object == "logs/a.bin")); + } + + #[tokio::test] + async fn list_multipart_uploads_paginates_across_same_key_boundary() { + let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; + let bucket = "multipart-prefix-page-bucket"; + for disk in &disk_stores { + disk.make_volume(bucket).await.expect("bucket volume should be created"); + } + + let mut expected = Vec::new(); + for object in ["logs/a.bin", "logs/a.bin", "logs/b.bin"] { + let upload = set_disks + .new_multipart_upload(bucket, object, &ObjectOptions::default()) + .await + .expect("multipart upload should be created"); + expected.push((object.to_string(), upload.upload_id)); + } + expected.sort(); + + let mut key_marker = None; + let mut upload_id_marker = None; + let mut listed = Vec::new(); + for _ in 0..expected.len() { + let page = set_disks + .list_multipart_uploads(bucket, "logs/", key_marker.clone(), upload_id_marker.clone(), None, 1) + .await + .expect("multipart page should succeed"); + assert_eq!(page.uploads.len(), 1); + listed.push((page.uploads[0].object.clone(), page.uploads[0].upload_id.clone())); + if !page.is_truncated { + break; + } + key_marker = page.next_key_marker; + upload_id_marker = page.next_upload_id_marker; + assert!(key_marker.is_some()); + assert!(upload_id_marker.is_some()); + } + + assert_eq!(listed, expected); + + let key_only = set_disks + .list_multipart_uploads(bucket, "logs/", Some("logs/a.bin".to_string()), None, None, 1000) + .await + .expect("key-only marker should succeed"); + assert_eq!( + key_only + .uploads + .iter() + .map(|upload| upload.object.as_str()) + .collect::>(), + vec!["logs/b.bin"] + ); + + let upload_only = set_disks + .list_multipart_uploads(bucket, "logs/", None, Some(expected[0].1.clone()), None, 1000) + .await + .expect("an upload marker without a key marker should be ignored"); + assert_eq!(upload_only.uploads.len(), expected.len()); + } + + #[tokio::test] + async fn list_multipart_uploads_folds_delimiter_and_paginates_prefixes() { + let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; + let bucket = "multipart-delimiter-list-bucket"; + for disk in &disk_stores { + disk.make_volume(bucket).await.expect("bucket volume should be created"); + } + + for object in [ + "logs/2025", + "logs/2025/a.bin", + "logs/2026/b.bin", + "logs/root.bin", + "other/c.bin", + ] { + set_disks + .new_multipart_upload(bucket, object, &ObjectOptions::default()) + .await + .expect("multipart upload should be created"); + } + + let first = set_disks + .list_multipart_uploads(bucket, "logs/", None, None, Some("/".to_string()), 2) + .await + .expect("delimiter multipart listing should succeed"); + assert_eq!(first.uploads.len(), 1); + assert_eq!(first.uploads[0].object, "logs/2025"); + assert_eq!(first.common_prefixes, vec!["logs/2025/".to_string()]); + assert!(first.is_truncated); + assert_eq!(first.next_key_marker.as_deref(), Some("logs/2025/")); + assert!(first.next_upload_id_marker.is_none()); + + let second = set_disks + .list_multipart_uploads( + bucket, + "logs/", + first.next_key_marker, + first.next_upload_id_marker, + Some("/".to_string()), + 2, + ) + .await + .expect("delimiter continuation should succeed"); + assert_eq!(second.uploads.len(), 1); + assert_eq!(second.uploads[0].object, "logs/root.bin"); + assert_eq!(second.common_prefixes, vec!["logs/2026/".to_string()]); + assert!(!second.is_truncated); + + let exact_boundary = set_disks + .list_multipart_uploads(bucket, "logs/", None, None, Some("/".to_string()), 4) + .await + .expect("delimiter exact boundary should succeed"); + assert_eq!(exact_boundary.uploads.len(), 2); + assert_eq!(exact_boundary.common_prefixes.len(), 2); + assert!(!exact_boundary.is_truncated); + } + /// Recursively collect every file named `file_name` under the multipart /// staging bucket on a single disk. Used to observe whether a failed commit /// left the per-part metadata intact for a retry. diff --git a/crates/ecstore/src/store/multipart.rs b/crates/ecstore/src/store/multipart.rs index ad6190cae..65f0128a3 100644 --- a/crates/ecstore/src/store/multipart.rs +++ b/crates/ecstore/src/store/multipart.rs @@ -13,8 +13,10 @@ // limitations under the License. use super::*; +use crate::multipart_listing::paginate_multipart_listing; use crate::set_disk::get_lock_acquire_timeout; use crate::storage_api_contracts::multipart::MultipartOperations as _; +use std::collections::HashSet; fn map_multipart_namespace_lock_error( bucket: &str, @@ -117,6 +119,8 @@ impl ECStore { } let mut uploads = Vec::new(); + let mut common_prefixes = HashSet::new(); + let mut source_truncated = false; for pool in self.pools.iter() { if self.is_suspended(pool.pool_idx).await { @@ -133,25 +137,27 @@ impl ECStore { ) .await?; uploads.extend(res.uploads); + common_prefixes.extend(res.common_prefixes); + source_truncated |= res.is_truncated; } // Each pool caps its own page at `max_uploads`, so the concatenation is // unordered across pools and may exceed the global cap. Re-sort, re-cap, // and derive the truncation markers so a bucket whose uploads span pools // pages correctly instead of being silently reported complete. - let (uploads, is_truncated, next_key_marker, next_upload_id_marker) = merge_multipart_upload_pages(uploads, max_uploads); + let page = merge_multipart_upload_pages(uploads, common_prefixes.into_iter().collect(), max_uploads, source_truncated); Ok(ListMultipartsInfo { key_marker, upload_id_marker, - next_key_marker, - next_upload_id_marker, + next_key_marker: page.next_key_marker, + next_upload_id_marker: page.next_upload_id_marker, max_uploads, - is_truncated, - uploads, + is_truncated: page.is_truncated, + uploads: page.uploads, + common_prefixes: page.common_prefixes, prefix: prefix.to_owned(), delimiter: delimiter.to_owned(), - ..Default::default() }) } @@ -398,24 +404,12 @@ impl ECStore { /// from the first overflow element (used only as a probe, never returned) so a /// bucket whose uploads span pools can be paged without loss or duplication. fn merge_multipart_upload_pages( - mut uploads: Vec, + uploads: Vec, + common_prefixes: Vec, max_uploads: usize, -) -> (Vec, bool, Option, Option) { - uploads.sort_by(|a, b| a.object.cmp(&b.object).then_with(|| a.upload_id.cmp(&b.upload_id))); - - let is_truncated = uploads.len() > max_uploads; - uploads.truncate(max_uploads); - - let (next_key_marker, next_upload_id_marker) = if is_truncated { - match uploads.last() { - Some(last) => (Some(last.object.clone()), Some(last.upload_id.clone())), - None => (None, None), - } - } else { - (None, None) - }; - - (uploads, is_truncated, next_key_marker, next_upload_id_marker) + source_truncated: bool, +) -> crate::multipart_listing::MultipartListingPage { + paginate_multipart_listing(uploads, common_prefixes, None, None, max_uploads, source_truncated) } #[cfg(test)] @@ -462,26 +456,30 @@ mod tests { // Union of two pools, unordered and exceeding the global cap. let uploads = vec![mp("b", "u1"), mp("a", "u2"), mp("a", "u1"), mp("c", "u1"), mp("b", "u2")]; - let (page, is_truncated, next_key_marker, next_upload_id_marker) = merge_multipart_upload_pages(uploads, 3); + let page = merge_multipart_upload_pages(uploads, Vec::new(), 3, false); - assert!(is_truncated); - assert_eq!(page.len(), 3); - let ordered: Vec<(&str, &str)> = page.iter().map(|u| (u.object.as_str(), u.upload_id.as_str())).collect(); + assert!(page.is_truncated); + assert_eq!(page.uploads.len(), 3); + let ordered: Vec<(&str, &str)> = page + .uploads + .iter() + .map(|u| (u.object.as_str(), u.upload_id.as_str())) + .collect(); assert_eq!(ordered, vec![("a", "u1"), ("a", "u2"), ("b", "u1")]); - assert_eq!(next_key_marker.as_deref(), Some("b")); - assert_eq!(next_upload_id_marker.as_deref(), Some("u1")); + assert_eq!(page.next_key_marker.as_deref(), Some("b")); + assert_eq!(page.next_upload_id_marker.as_deref(), Some("u1")); } #[test] fn merge_multipart_upload_pages_reports_complete_within_cap() { let uploads = vec![mp("b", "u1"), mp("a", "u1")]; - let (page, is_truncated, next_key_marker, next_upload_id_marker) = merge_multipart_upload_pages(uploads, 3); + let page = merge_multipart_upload_pages(uploads, Vec::new(), 3, false); - assert_eq!(page.len(), 2); - assert!(!is_truncated); - assert!(next_key_marker.is_none()); - assert!(next_upload_id_marker.is_none()); + assert_eq!(page.uploads.len(), 2); + assert!(!page.is_truncated); + assert!(page.next_key_marker.is_none()); + assert!(page.next_upload_id_marker.is_none()); } #[test] @@ -509,15 +507,15 @@ mod tests { merged.extend(pool_query(pool, key_marker.as_deref(), upload_id_marker.as_deref(), max_uploads)); } - let (page, is_truncated, next_key_marker, next_upload_id_marker) = merge_multipart_upload_pages(merged, max_uploads); - assert!(page.len() <= max_uploads); - collected.extend(page.iter().map(|u| (u.object.clone(), u.upload_id.clone()))); + let page = merge_multipart_upload_pages(merged, Vec::new(), max_uploads, false); + assert!(page.uploads.len() <= max_uploads); + collected.extend(page.uploads.iter().map(|u| (u.object.clone(), u.upload_id.clone()))); - if !is_truncated { + if !page.is_truncated { break; } - key_marker = next_key_marker; - upload_id_marker = next_upload_id_marker; + key_marker = page.next_key_marker; + upload_id_marker = page.next_upload_id_marker; } assert_eq!(collected, expected, "pagination must return every upload exactly once, in sorted order"); @@ -526,6 +524,31 @@ mod tests { assert_eq!(deduped.len(), collected.len(), "pagination must not duplicate uploads"); } + #[test] + fn merge_multipart_upload_pages_includes_common_prefixes() { + let page = merge_multipart_upload_pages( + vec![mp("logs/root.bin", "u1")], + vec!["logs/2026/".to_string(), "logs/2025/".to_string()], + 2, + false, + ); + + assert!(page.is_truncated); + assert!(page.uploads.is_empty()); + assert_eq!(page.common_prefixes, vec!["logs/2025/", "logs/2026/"]); + assert_eq!(page.next_key_marker.as_deref(), Some("logs/2026/")); + assert!(page.next_upload_id_marker.is_none()); + } + + #[test] + fn merge_multipart_upload_pages_preserves_pool_truncation() { + let page = merge_multipart_upload_pages(vec![mp("a", "u1")], Vec::new(), 1, true); + + assert!(page.is_truncated); + assert_eq!(page.next_key_marker.as_deref(), Some("a")); + assert_eq!(page.next_upload_id_marker.as_deref(), Some("u1")); + } + async fn new_multipart_lock_test_store() -> ECStore { let format = FormatV3::new(1, 2); let endpoints = vec![ diff --git a/rustfs/src/storage/s3_api/multipart.rs b/rustfs/src/storage/s3_api/multipart.rs index ee2cc75b1..d74997b8b 100644 --- a/rustfs/src/storage/s3_api/multipart.rs +++ b/rustfs/src/storage/s3_api/multipart.rs @@ -164,6 +164,8 @@ pub(crate) fn build_list_multipart_uploads_output( delimiter: result.delimiter, key_marker: result.key_marker, upload_id_marker: result.upload_id_marker, + next_key_marker: result.next_key_marker, + next_upload_id_marker: result.next_upload_id_marker, max_uploads: Some(result.max_uploads as i32), is_truncated: Some(result.is_truncated), uploads: Some( @@ -285,6 +287,8 @@ mod tests { delimiter: Some("/".to_string()), key_marker: Some("key-marker".to_string()), upload_id_marker: Some("upload-id-marker".to_string()), + next_key_marker: Some("next-key-marker".to_string()), + next_upload_id_marker: Some("next-upload-id-marker".to_string()), max_uploads: 1000, is_truncated: true, uploads: vec![MultipartInfo { @@ -307,6 +311,8 @@ mod tests { assert_eq!(output.delimiter.as_deref(), Some("/")); assert_eq!(output.key_marker.as_deref(), Some("key-marker")); assert_eq!(output.upload_id_marker.as_deref(), Some("upload-id-marker")); + assert_eq!(output.next_key_marker.as_deref(), Some("next-key-marker")); + assert_eq!(output.next_upload_id_marker.as_deref(), Some("next-upload-id-marker")); assert_eq!(output.max_uploads, Some(1000)); assert_eq!(output.is_truncated, Some(true));