diff --git a/crates/ecstore/src/core/pools.rs b/crates/ecstore/src/core/pools.rs index b354165c1..6f0439101 100644 --- a/crates/ecstore/src/core/pools.rs +++ b/crates/ecstore/src/core/pools.rs @@ -5282,13 +5282,12 @@ impl ECStore { "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" - ))); + for set in &pool.disk_set { + if let Some(upload_path) = set.first_multipart_upload_path_for_decommission(bucket).await? { + return Err(Error::other(format!( + "pool {idx} still contains multipart upload `{upload_path}` for bucket `{bucket}`; resolve it before retrying decommission" + ))); + } } } diff --git a/crates/ecstore/src/set_disk/ops/multipart.rs b/crates/ecstore/src/set_disk/ops/multipart.rs index 5deb93d8d..77239f6a9 100644 --- a/crates/ecstore/src/set_disk/ops/multipart.rs +++ b/crates/ecstore/src/set_disk/ops/multipart.rs @@ -612,9 +612,9 @@ impl SetDisks { Ok((disks, candidate_paths, discovery_quorum)) } - pub(crate) async fn first_multipart_upload_path_for_decommission(&self) -> Result> { + pub(crate) async fn first_multipart_upload_path_for_decommission(&self, bucket: &str) -> Result> { let (_, paths, _) = self - .discover_multipart_upload_paths(RUSTFS_META_BUCKET, RUSTFS_META_MULTIPART_BUCKET) + .discover_multipart_upload_paths(bucket, RUSTFS_META_MULTIPART_BUCKET) .await?; Ok(paths.into_iter().next()) } diff --git a/crates/ecstore/src/store/init.rs b/crates/ecstore/src/store/init.rs index ac49529c5..aae749cf4 100644 --- a/crates/ecstore/src/store/init.rs +++ b/crates/ecstore/src/store/init.rs @@ -1631,9 +1631,10 @@ mod tests { .ensure_decommission_multipart_uploads_drained_for_test(0) .await .expect_err("an unresolved source multipart upload must block final decommission"); + let drain_error = err.to_string(); assert!( - err.to_string().contains("still contains multipart upload"), - "unexpected drain error: {err}" + drain_error.contains("still contains multipart upload") && drain_error.contains(&bucket), + "the drain error must identify both the upload path and user bucket: {drain_error}" ); let listed = store @@ -1700,6 +1701,65 @@ mod tests { shutdown.cancel(); } + #[tokio::test] + #[serial_test::serial(storage_class_env)] + async fn active_multipart_upload_routes_before_faulted_suspended_source() { + let temp_dir = tempfile::tempdir().expect("create active-first multipart routing store dir"); + let (_ctx, store, shutdown) = + without_storage_class_env(build_isolated_test_store(temp_dir.path(), "active-first-multipart-routing", &[4, 4])) + .await; + crate::bucket::metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await; + + let bucket = format!("active-first-multipart-routing-{}", uuid::Uuid::new_v4()); + let object = "target-upload.bin"; + store + .make_bucket(&bucket, &MakeBucketOptions::default()) + .await + .expect("create active-first multipart routing 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 upload = store.pools[1] + .new_multipart_upload(&bucket, object, &upload_opts) + .await + .expect("create upload in active target pool"); + drop(lifecycle_guard); + + mark_test_pool_decommissioning(&store, 0).await; + let source_set = store.pools[0].get_disks_by_key(object); + let original_source_disks = { + let mut disks = source_set.disks.write().await; + let original = disks.clone(); + disks.fill(None); + original + }; + + let source_result = store.pools[0] + .get_multipart_info(&bucket, object, &upload.upload_id, &ObjectOptions::default()) + .await; + let routed_result = store + .get_multipart_info(&bucket, object, &upload.upload_id, &ObjectOptions::default()) + .await; + *source_set.disks.write().await = original_source_disks; + + assert!( + matches!(&source_result, Err(StorageError::ErasureReadQuorum)), + "the suspended source must expose the injected hard read failure: {source_result:?}" + ); + let routed = routed_result.expect("the active target UploadID must be resolved before the faulted suspended source"); + assert_eq!(routed.upload_id, upload.upload_id); + + 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 29c5d68ec..74837e46a 100644 --- a/crates/ecstore/src/store/multipart.rs +++ b/crates/ecstore/src/store/multipart.rs @@ -196,10 +196,23 @@ 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 + async fn existing_multipart_pool_order(&self) -> Vec { + // A draining source must not hide a valid UploadID in an active target, + // while physical order within each phase preserves fail-closed errors. + let mut active = Vec::with_capacity(self.pools.len()); + let mut draining = Vec::new(); + for (idx, pool) in self.pools.iter().enumerate() { + if self.is_pool_rebalancing(pool.pool_idx).await { + continue; + } + if self.is_suspended(pool.pool_idx).await { + draining.push(idx); + } else { + active.push(idx); + } + } + active.extend(draining); + active } #[allow(clippy::too_many_arguments)] @@ -296,10 +309,8 @@ impl ECStore { .await; } - for pool in self.pools.iter() { - if !self.multipart_pool_accepts_existing_upload_operations(pool.pool_idx).await { - continue; - } + for pool_idx in self.existing_multipart_pool_order().await { + let pool = &self.pools[pool_idx]; return match pool .list_object_parts(bucket, object, upload_id, part_number_marker, max_parts, opts) .await @@ -359,10 +370,8 @@ impl ECStore { let mut common_prefixes = HashSet::new(); let mut source_truncated = false; - for pool in self.pools.iter() { - if !self.multipart_pool_accepts_existing_upload_operations(pool.pool_idx).await { - continue; - } + for pool_idx in self.existing_multipart_pool_order().await { + let pool = &self.pools[pool_idx]; let res = list_pool_multipart_uploads_for_incarnation( pool, bucket, @@ -529,10 +538,8 @@ impl ECStore { .await; } - for pool in self.pools.iter() { - if !self.multipart_pool_accepts_existing_upload_operations(pool.pool_idx).await { - continue; - } + for pool_idx in self.existing_multipart_pool_order().await { + let pool = &self.pools[pool_idx]; let err = match pool.put_object_part(bucket, object, upload_id, part_id, data, opts).await { Ok(res) => return Ok(res), Err(err) => { @@ -592,10 +599,8 @@ impl ECStore { return self.pools[0].get_multipart_info(bucket, object, upload_id, opts).await; } - for pool in self.pools.iter() { - if !self.multipart_pool_accepts_existing_upload_operations(pool.pool_idx).await { - continue; - } + for pool_idx in self.existing_multipart_pool_order().await { + let pool = &self.pools[pool_idx]; return match pool.get_multipart_info(bucket, object, upload_id, opts).await { Ok(res) => Ok(res), @@ -630,10 +635,8 @@ impl ECStore { return self.pools[0].abort_multipart_upload(bucket, object, upload_id, opts).await; } - for pool in self.pools.iter() { - if !self.multipart_pool_accepts_existing_upload_operations(pool.pool_idx).await { - continue; - } + for pool_idx in self.existing_multipart_pool_order().await { + let pool = &self.pools[pool_idx]; let err = match pool.abort_multipart_upload(bucket, object, upload_id, opts).await { Ok(_) => return Ok(()), @@ -691,10 +694,8 @@ impl ECStore { .await; } - for pool in self.pools.iter() { - if !self.multipart_pool_accepts_existing_upload_operations(pool.pool_idx).await { - continue; - } + for pool_idx in self.existing_multipart_pool_order().await { + let pool = &self.pools[pool_idx]; let pool = pool.clone(); let err = match pool