From e547820cb86f474f58d53aa024547b764d2da9fa Mon Sep 17 00:00:00 2001 From: overtrue Date: Sun, 23 Aug 2026 03:57:25 +0800 Subject: [PATCH] fix(ecstore): drain multipart uploads before decommission --- crates/ecstore/src/core/pools.rs | 47 ++++++++ crates/ecstore/src/set_disk/ops/multipart.rs | 111 +++++++++++-------- crates/ecstore/src/store/init.rs | 111 +++++++++++++++++++ crates/ecstore/src/store/multipart.rs | 18 ++- 4 files changed, 234 insertions(+), 53 deletions(-) diff --git a/crates/ecstore/src/core/pools.rs b/crates/ecstore/src/core/pools.rs index 9d8e0d01a..b354165c1 100644 --- a/crates/ecstore/src/core/pools.rs +++ b/crates/ecstore/src/core/pools.rs @@ -5143,6 +5143,9 @@ impl ECStore { let buckets = self.get_buckets_to_decommission().await?; let pool = self.pools[idx].clone(); + self.ensure_decommission_multipart_uploads_drained(idx, &pool, &buckets) + .await?; + for (set_index, set) in pool.disk_set.iter().enumerate() { for bucket_info in &buckets { let mut lifecycle_config = None; @@ -5256,6 +5259,50 @@ impl ECStore { Ok(()) } + async fn ensure_decommission_multipart_uploads_drained( + &self, + idx: usize, + pool: &Sets, + buckets: &[DecomBucketInfo], + ) -> Result<()> { + let mut bucket_names = buckets + .iter() + .filter(|bucket| bucket.name != RUSTFS_META_BUCKET) + .map(|bucket| bucket.name.as_str()) + .collect::>(); + bucket_names.sort_unstable(); + bucket_names.dedup(); + + // Take one bucket fence at a time so cross-bucket COPY cannot form an + // ABBA cycle. Suspension prevents new source uploads after each fence. + for bucket in bucket_names { + let lifecycle_guard = self.acquire_bucket_lifecycle_write_lock(bucket).await?; + if lifecycle_guard.is_lock_lost() { + return Err(Error::other(format!( + "decommission multipart drain lost the bucket lifecycle fence for `{bucket}`" + ))); + } + } + + for set in &pool.disk_set { + if let Some(upload_path) = set.first_multipart_upload_path_for_decommission().await? { + return Err(Error::other(format!( + "pool {idx} still contains multipart upload `{upload_path}`; resolve it before retrying decommission" + ))); + } + } + + Ok(()) + } + + #[cfg(test)] + pub(crate) async fn ensure_decommission_multipart_uploads_drained_for_test(self: &Arc, idx: usize) -> Result<()> { + let buckets = self.get_buckets_to_decommission().await?; + let pool = self.pools[idx].clone(); + self.ensure_decommission_multipart_uploads_drained(idx, pool.as_ref(), &buckets) + .await + } + #[tracing::instrument(skip(self, rd))] async fn decommission_object( self: Arc, diff --git a/crates/ecstore/src/set_disk/ops/multipart.rs b/crates/ecstore/src/set_disk/ops/multipart.rs index 5aa59b158..5deb93d8d 100644 --- a/crates/ecstore/src/set_disk/ops/multipart.rs +++ b/crates/ecstore/src/set_disk/ops/multipart.rs @@ -556,6 +556,69 @@ async fn multipart_upload_paths_on_disk(disk: DiskStore, bucket: &str) -> disk:: } impl SetDisks { + async fn discover_multipart_upload_paths( + &self, + orig_bucket: &str, + error_path: &str, + ) -> Result<(Vec>, Vec, usize)> { + 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 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 orig_bucket = orig_bucket.to_string(); + discovery_tasks.spawn(async move { + let result = match disk { + Some(disk) => multipart_upload_paths_on_disk(disk, &orig_bucket).await, + None => Err(DiskError::DiskNotFound), + }; + (index, result) + }); + } + + 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; + } + } + Err(err) => discovery_errors[index] = Some(err), + } + } + + if let Some(err) = reduce_read_quorum_errs(&discovery_errors, OBJECT_OP_IGNORED_ERRS, discovery_quorum) { + return Err(to_object_err(err.into(), vec![orig_bucket, error_path])); + } + + let mut candidate_paths = candidate_counts + .into_iter() + .filter_map(|(path, count)| (count >= discovery_quorum).then_some(path)) + .collect::>(); + candidate_paths.sort_unstable(); + Ok((disks, candidate_paths, discovery_quorum)) + } + + pub(crate) async fn first_multipart_upload_path_for_decommission(&self) -> Result> { + let (_, paths, _) = self + .discover_multipart_upload_paths(RUSTFS_META_BUCKET, RUSTFS_META_MULTIPART_BUCKET) + .await?; + Ok(paths.into_iter().next()) + } + async fn acquire_multipart_upload_read_lock( &self, op: &'static str, @@ -747,53 +810,7 @@ impl SetDisks { max_uploads: usize, expected_incarnation_id: Option, ) -> Result { - 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 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) - }); - } - - 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; - } - } - Err(err) => discovery_errors[index] = Some(err), - } - } - - 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 (disks, candidate_paths, discovery_quorum) = self.discover_multipart_upload_paths(bucket, prefix).await?; let listed_uploads = stream::iter(candidate_paths) .map(|upload_path| { let disks = &disks; diff --git a/crates/ecstore/src/store/init.rs b/crates/ecstore/src/store/init.rs index 451e42242..ac49529c5 100644 --- a/crates/ecstore/src/store/init.rs +++ b/crates/ecstore/src/store/init.rs @@ -1589,6 +1589,117 @@ mod tests { shutdown.cancel(); } + #[tokio::test] + #[serial_test::serial(storage_class_env)] + async fn suspended_decommission_source_multipart_remains_operable_until_drained() { + let temp_dir = tempfile::tempdir().expect("create decommission multipart drain store dir"); + let (_ctx, store, shutdown) = + without_storage_class_env(build_isolated_test_store(temp_dir.path(), "decommission-multipart-drain", &[4, 4])).await; + crate::bucket::metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await; + + let bucket = format!("decommission-multipart-drain-{}", uuid::Uuid::new_v4()); + let complete_object = "complete.bin"; + let abort_object = "abort.bin"; + store + .make_bucket(&bucket, &MakeBucketOptions::default()) + .await + .expect("create decommission multipart drain bucket"); + + let incarnation = store.bucket_incarnation_id(&bucket).await.expect("read bucket incarnation"); + let lifecycle_guard = store + .acquire_bucket_lifecycle_read_lock(&bucket) + .await + .expect("acquire multipart creation lifecycle fence"); + let mut upload_opts = ObjectOptions { + expected_bucket_incarnation_id: Some(incarnation), + ..Default::default() + }; + upload_opts.add_bucket_lifecycle_lock_guard(&lifecycle_guard); + let complete_upload = store.pools[0] + .new_multipart_upload(&bucket, complete_object, &upload_opts) + .await + .expect("create source upload to complete"); + let abort_upload = store.pools[0] + .new_multipart_upload(&bucket, abort_object, &upload_opts) + .await + .expect("create source upload to abort"); + drop(lifecycle_guard); + + mark_test_pool_decommissioning(&store, 0).await; + + let err = store + .ensure_decommission_multipart_uploads_drained_for_test(0) + .await + .expect_err("an unresolved source multipart upload must block final decommission"); + assert!( + err.to_string().contains("still contains multipart upload"), + "unexpected drain error: {err}" + ); + + let listed = store + .list_multipart_uploads(&bucket, "", None, None, None, 100) + .await + .expect("list uploads from suspended decommission source"); + assert!( + listed + .uploads + .iter() + .any(|upload| upload.upload_id.as_str() == complete_upload.upload_id.as_str()), + "the upload selected before suspension must remain visible" + ); + store + .get_multipart_info(&bucket, complete_object, &complete_upload.upload_id, &ObjectOptions::default()) + .await + .expect("read upload metadata from suspended decommission source"); + + let mut part_reader = PutObjReader::from_vec(b"multipart body".to_vec()); + let part = store + .put_object_part( + &bucket, + complete_object, + &complete_upload.upload_id, + 1, + &mut part_reader, + &ObjectOptions::default(), + ) + .await + .expect("write part to suspended decommission source"); + let parts = store + .list_object_parts(&bucket, complete_object, &complete_upload.upload_id, None, 100, &ObjectOptions::default()) + .await + .expect("list parts from suspended decommission source"); + assert_eq!(parts.parts.len(), 1); + assert_eq!(parts.parts[0].etag.as_deref(), part.etag.as_deref()); + + store + .clone() + .complete_multipart_upload( + &bucket, + complete_object, + &complete_upload.upload_id, + vec![crate::storage_api_contracts::multipart::CompletePart { + part_num: part.part_num, + etag: part.etag, + ..Default::default() + }], + &ObjectOptions::default(), + ) + .await + .expect("complete upload on suspended decommission source"); + store + .abort_multipart_upload(&bucket, abort_object, &abort_upload.upload_id, &ObjectOptions::default()) + .await + .expect("abort upload on suspended decommission source"); + + store + .ensure_decommission_multipart_uploads_drained_for_test(0) + .await + .expect("final decommission gate should open after all source uploads are resolved"); + assert_pool_object_present(&store.pools[0], &bucket, complete_object).await; + + shutdown.cancel(); + } + #[tokio::test] #[serial_test::serial(storage_class_env)] async fn delete_objects_skips_active_rebalance_source_pool() { diff --git a/crates/ecstore/src/store/multipart.rs b/crates/ecstore/src/store/multipart.rs index 8d72287d4..29c5d68ec 100644 --- a/crates/ecstore/src/store/multipart.rs +++ b/crates/ecstore/src/store/multipart.rs @@ -196,6 +196,12 @@ async fn list_pool_multipart_uploads_for_incarnation( } impl ECStore { + // Decommission drains existing UploadIDs in place; rebalance keeps its + // established source-exclusion behavior. + async fn multipart_pool_accepts_existing_upload_operations(&self, pool_idx: usize) -> bool { + !self.is_pool_rebalancing(pool_idx).await + } + #[allow(clippy::too_many_arguments)] pub async fn list_multipart_uploads_for_bucket_incarnation( &self, @@ -291,7 +297,7 @@ impl ECStore { } for pool in self.pools.iter() { - if self.is_suspended(pool.pool_idx).await || self.is_pool_rebalancing(pool.pool_idx).await { + if !self.multipart_pool_accepts_existing_upload_operations(pool.pool_idx).await { continue; } return match pool @@ -354,7 +360,7 @@ impl ECStore { let mut source_truncated = false; for pool in self.pools.iter() { - if self.is_suspended(pool.pool_idx).await || self.is_pool_rebalancing(pool.pool_idx).await { + if !self.multipart_pool_accepts_existing_upload_operations(pool.pool_idx).await { continue; } let res = list_pool_multipart_uploads_for_incarnation( @@ -524,7 +530,7 @@ impl ECStore { } for pool in self.pools.iter() { - if self.is_suspended(pool.pool_idx).await || self.is_pool_rebalancing(pool.pool_idx).await { + if !self.multipart_pool_accepts_existing_upload_operations(pool.pool_idx).await { continue; } let err = match pool.put_object_part(bucket, object, upload_id, part_id, data, opts).await { @@ -587,7 +593,7 @@ impl ECStore { } for pool in self.pools.iter() { - if self.is_suspended(pool.pool_idx).await || self.is_pool_rebalancing(pool.pool_idx).await { + if !self.multipart_pool_accepts_existing_upload_operations(pool.pool_idx).await { continue; } @@ -625,7 +631,7 @@ impl ECStore { } for pool in self.pools.iter() { - if self.is_suspended(pool.pool_idx).await || self.is_pool_rebalancing(pool.pool_idx).await { + if !self.multipart_pool_accepts_existing_upload_operations(pool.pool_idx).await { continue; } @@ -686,7 +692,7 @@ impl ECStore { } for pool in self.pools.iter() { - if self.is_suspended(pool.pool_idx).await || self.is_pool_rebalancing(pool.pool_idx).await { + if !self.multipart_pool_accepts_existing_upload_operations(pool.pool_idx).await { continue; }