gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-23 04:39:04 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(ecstore): prioritize active multipart pools

Browse Source
This commit is contained in:
overtrue
2026-08-23 04:28:50 +08:00
parent e547820cb8
commit 71c3f9bd2c
4 changed files with 99 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/core/pools.rs
+6 -7
View File
@@ -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"
)));
}
}
}
crates/ecstore/src/set_disk/ops/multipart.rs
+2 -2
View File
@@ -612,9 +612,9 @@ impl SetDisks {
Ok((disks, candidate_paths, discovery_quorum))
}
pub(crate) async fn first_multipart_upload_path_for_decommission(&self) -> Result<Option<String>> {
pub(crate) async fn first_multipart_upload_path_for_decommission(&self, bucket: &str) -> Result<Option<String>> {
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())
}
crates/ecstore/src/store/init.rs
+62 -2
View File
@@ -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() {
crates/ecstore/src/store/multipart.rs
+29 -28
View File
@@ -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<usize> {
// 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