fix(ecstore): drain multipart uploads before decommission

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::<Vec<_>>();
+        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<Self>, 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<Self>,

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<Option<DiskStore>>, Vec<String>, 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::<Vec<_>>();
+        let mut candidate_counts = HashMap::<String, usize>::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::<Vec<_>>();
+        candidate_paths.sort_unstable();
+        Ok((disks, candidate_paths, discovery_quorum))
+    }
+
+    pub(crate) async fn first_multipart_upload_path_for_decommission(&self) -> Result<Option<String>> {
+        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<Uuid>,
     ) -> Result<ListMultipartsInfo> {
-        let disks = self.disks.read().await.clone();
+        let (disks, candidate_paths, discovery_quorum) = self.discover_multipart_upload_paths(bucket, prefix).await?;
-        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::<Vec<_>>();
-        let mut candidate_counts = HashMap::<String, usize>::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::<Vec<_>>();
         let listed_uploads = stream::iter(candidate_paths)
             .map(|upload_path| {
                 let disks = &disks;

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() {

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;
             }