From 09fac9a52b242e239a4320cd6932c370e230021f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=A9=AC=E7=99=BB=E5=B1=B1?= Date: Sat, 22 Aug 2026 11:04:29 +0800 Subject: [PATCH] fix(ecstore): fail closed on unresolved decommission entries --- crates/ecstore/src/core/pools.rs | 224 +++++++++++++++++++++++++++---- 1 file changed, 200 insertions(+), 24 deletions(-) diff --git a/crates/ecstore/src/core/pools.rs b/crates/ecstore/src/core/pools.rs index fb983dc0c..142ff0873 100644 --- a/crates/ecstore/src/core/pools.rs +++ b/crates/ecstore/src/core/pools.rs @@ -36,7 +36,8 @@ use crate::disk::error::DiskError; use crate::disk::{BUCKET_META_PREFIX, RUSTFS_META_BUCKET}; use crate::error::{Error, Result}; use crate::error::{ - StorageError, is_err_bucket_exists, is_err_bucket_not_found, is_err_object_not_found, is_err_version_not_found, + StorageError, is_err_bucket_exists, is_err_bucket_not_found, is_err_object_not_found, is_err_operation_canceled, + is_err_version_not_found, }; use crate::layout::endpoints::EndpointServerPools; use crate::object_api::{GetObjectReader, ObjectOptions}; @@ -773,7 +774,75 @@ async fn load_decommission_entry_exact_versions( } fn resolve_decommission_check_after_list_result(list_result: Result<()>, entry_error: Option) -> Result<()> { - if let Some(err) = entry_error { Err(err) } else { list_result } + match list_result { + Ok(()) => entry_error.map_or(Ok(()), Err), + Err(list_err) => resolve_decommission_listing_error(Some(list_err), entry_error).map_or(Ok(()), Err), + } +} + +fn resolve_decommission_listing_error(listing_error: Option, entry_error: Option) -> Option { + match (listing_error, entry_error) { + (Some(listing_error), Some(entry_error)) if is_err_operation_canceled(&listing_error) => Some(entry_error), + (Some(listing_error), Some(entry_error)) if is_err_operation_canceled(&entry_error) => Some(listing_error), + (Some(listing_error), _) => Some(listing_error), + (None, entry_error) => entry_error, + } +} + +fn decommission_unresolved_listing_error( + bucket: &str, + prefix: &str, + candidate: Option<&str>, + candidate_count: usize, + disk_error_count: usize, + pool_index: usize, + set_index: usize, +) -> Error { + let location = candidate.unwrap_or(prefix); + Error::other(format!( + "decommission listing could not resolve metadata for {bucket}/{location} on pool {pool_index} set {set_index} ({candidate_count} candidate(s), {disk_error_count} disk error(s))" + )) +} + +fn resolve_decommission_partial_listing_entry( + entries: MetaCacheEntries, + resolver: MetadataResolutionParams, + bucket: &str, + prefix: &str, + disk_error_count: usize, + pool_index: usize, + set_index: usize, +) -> Result { + let candidate_count = entries.as_ref().iter().flatten().count(); + let candidate = entries.as_ref().iter().flatten().map(|entry| entry.name.clone()).next(); + + entries.resolve(resolver).ok_or_else(|| { + decommission_unresolved_listing_error( + bucket, + prefix, + candidate.as_deref(), + candidate_count, + disk_error_count, + pool_index, + set_index, + ) + }) +} + +async fn record_decommission_entry_error( + entry_error: &Arc>>, + rx: &CancellationToken, + err: Error, +) { + if rx.is_cancelled() { + return; + } + + let mut first_err = entry_error.lock().await; + if first_err.is_none() && !rx.is_cancelled() { + *first_err = Some(err); + rx.cancel(); + } } fn resolve_decommission_pool_meta_reload_result(result: Result<()>, stage: &str) -> Result<()> { @@ -3538,6 +3607,7 @@ impl ECStore { let rx_clone = rx.clone(); let bi = bi.clone(); let set_id = set_idx; + let listing_entry_error = entry_error.clone(); let worker = tokio::spawn(async move { let _listing_permit = listing_permit; run_decommission_listing_with_retry( @@ -3551,7 +3621,11 @@ impl ECStore { let set = set.clone(); let rx = rx_clone.clone(); let bucket = bi.clone(); - async move { set.list_objects_to_decommission(rx, bucket, callback).await } + let entry_error = listing_entry_error.clone(); + async move { + set.list_objects_to_decommission(rx, bucket, callback, entry_error.clone(), idx, set_id) + .await + } }, ) .await @@ -3581,11 +3655,7 @@ impl ECStore { wait_decommission_worker_drain(&workers, worker_limit).await?; - if let Some(err) = listing_worker_error { - return Err(err); - } - - if let Some(err) = entry_error.lock().await.clone() { + if let Some(err) = resolve_decommission_listing_error(listing_worker_error, entry_error.lock().await.clone()) { return Err(err); } @@ -4191,7 +4261,7 @@ impl ECStore { let buckets = self.get_buckets_to_decommission().await?; let pool = self.pools[idx].clone(); - for set in &pool.disk_set { + for (set_index, set) in pool.disk_set.iter().enumerate() { for bucket_info in &buckets { let mut lifecycle_config = None; let mut object_lock_config = None; @@ -4286,7 +4356,7 @@ impl ECStore { }); let list_result = set - .list_objects_to_decommission(callback_rx, bucket_info.clone(), callback) + .list_objects_to_decommission(callback_rx, bucket_info.clone(), callback, entry_error.clone(), idx, set_index) .await; let entry_error = entry_error.lock().await.clone(); resolve_decommission_check_after_list_result(list_result, entry_error)?; @@ -5021,12 +5091,15 @@ mod tests { pub type ListCallback = Arc BoxFuture<'static, ()> + Send + Sync + 'static>; impl SetDisks { - #[tracing::instrument(skip(self, rx, cb_func))] + #[tracing::instrument(skip(self, rx, cb_func, entry_error))] async fn list_objects_to_decommission( self: &Arc, rx: CancellationToken, bucket_info: DecomBucketInfo, cb_func: ListCallback, + entry_error: Arc>>, + pool_index: usize, + set_index: usize, ) -> Result<()> { let (disks, _) = self.get_online_disks_with_healing(false).await; ensure_decommission_listing_disks_available(!disks.is_empty(), &bucket_info.name)?; @@ -5041,6 +5114,12 @@ impl SetDisks { }; let cb1 = cb_func.clone(); + let unresolved_error = entry_error.clone(); + let unresolved_rx = rx.clone(); + let unresolved_bucket = bucket_info.name.clone(); + let unresolved_prefix = bucket_info.prefix.clone(); + let unresolved_pool_index = pool_index; + let unresolved_set_index = set_index; list_path_raw( rx, @@ -5053,20 +5132,51 @@ impl SetDisks { skip_walkdir_total_timeout: true, walkdir_stall_timeout: Some(DECOMMISSION_BACKGROUND_WALKDIR_STALL_TIMEOUT), agreed: Some(Box::new(move |entry: MetaCacheEntry| Box::pin(cb1(entry)))), - partial: Some(Box::new(move |entries: MetaCacheEntries, _: &[Option]| { + partial: Some(Box::new(move |entries: MetaCacheEntries, errs: &[Option]| { let resolver = resolver.clone(); let cb_func = cb_func.clone(); - match entries.resolve(resolver) { - Some(entry) => { + let bucket = unresolved_bucket.clone(); + let prefix = unresolved_prefix.clone(); + let unresolved_error = unresolved_error.clone(); + let unresolved_rx = unresolved_rx.clone(); + let pool_index = unresolved_pool_index; + let set_index = unresolved_set_index; + let disk_error_count = errs.iter().flatten().count(); + if unresolved_rx.is_cancelled() { + return Box::pin(async {}); + } + + match resolve_decommission_partial_listing_entry( + entries, + resolver, + &bucket, + &prefix, + disk_error_count, + pool_index, + set_index, + ) { + Ok(entry) => { warn!("decommission_pool: list_objects_to_decommission get {}", &entry.name); Box::pin(async move { cb_func(entry).await; }) } - None => { - warn!("decommission_pool: list_objects_to_decommission get none"); - Box::pin(async {}) - } + Err(err) => Box::pin(async move { + if unresolved_rx.is_cancelled() { + return; + } + warn!( + event = EVENT_DECOMMISSION_BUCKET, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_POOLS, + bucket = %bucket, + prefix = %prefix, + state = "unresolved_entry", + error = %err, + "Decommission listing failed closed on unresolved metadata" + ); + record_decommission_entry_error(&unresolved_error, &unresolved_rx, err).await; + }), } })), ..Default::default() @@ -5074,6 +5184,10 @@ impl SetDisks { ) .await?; + if let Some(err) = entry_error.lock().await.clone() { + return Err(err); + } + Ok(()) } } @@ -5279,11 +5393,12 @@ mod pools_tests { has_active_decommission_canceler, is_decommission_active, is_decommission_cancel_requested, load_decommission_entry_versions, local_decommission_queue_prefix, mark_decommission_bucket_done, merge_pool_status_refresh, missing_decommission_worker_prefix, observe_decommission_terminal_reload_result, - pool_meta_has_active_decommission, require_decommission_store, resolve_decommission_bucket_done_save_result, - resolve_decommission_bucket_state, resolve_decommission_check_after_list_result, - resolve_decommission_entry_cleanup_delete_result, resolve_decommission_entry_exact_versions, - resolve_decommission_entry_reload_result, resolve_decommission_listing_worker_result, - resolve_decommission_optional_bucket_config_result, resolve_decommission_pool_meta_reload_result, + pool_meta_has_active_decommission, record_decommission_entry_error, require_decommission_store, + resolve_decommission_bucket_done_save_result, resolve_decommission_bucket_state, + resolve_decommission_check_after_list_result, resolve_decommission_entry_cleanup_delete_result, + resolve_decommission_entry_exact_versions, resolve_decommission_entry_reload_result, resolve_decommission_listing_error, + resolve_decommission_listing_worker_result, resolve_decommission_optional_bucket_config_result, + resolve_decommission_partial_listing_entry, resolve_decommission_pool_meta_reload_result, resolve_decommission_preflight_heal_result, resolve_decommission_progress_save_result, resolve_decommission_spawn_failure_result, resolve_decommission_terminal_mark_after_error_result, resolve_decommission_terminal_mark_result, resolve_decommission_update_after_result, @@ -5302,7 +5417,9 @@ mod pools_tests { use crate::error::{Error, StorageError}; use crate::layout::endpoints::{EndpointServerPools, Endpoints, PoolEndpoints}; use crate::services::rebalance::{RebalStatus, RebalanceInfo, RebalanceMeta, RebalanceStats}; - use rustfs_filemeta::{FileInfo, FileInfoVersions, MetaCacheEntry, ObjectPartInfo}; + use rustfs_filemeta::{ + FileInfo, FileInfoVersions, MetaCacheEntries, MetaCacheEntry, MetadataResolutionParams, ObjectPartInfo, + }; use rustfs_rio::Index; use std::sync::{ Arc, @@ -6321,6 +6438,65 @@ mod pools_tests { assert!(matches!(err, Error::SlowDown)); } + #[test] + fn test_resolve_decommission_partial_listing_entry_rejects_unresolved_metadata() { + let err = resolve_decommission_partial_listing_entry( + MetaCacheEntries(vec![None]), + MetadataResolutionParams { + dir_quorum: 2, + obj_quorum: 2, + bucket: "bucket-a".to_string(), + ..Default::default() + }, + "bucket-a", + "prefix/", + 1, + 2, + 3, + ) + .expect_err("unresolved partial listing must fail closed"); + + let message = err.to_string(); + assert!(message.contains("decommission listing could not resolve metadata")); + assert!(message.contains("bucket-a/prefix/")); + assert!(message.contains("pool 2 set 3")); + assert!(message.contains("1 disk error(s)")); + } + + #[tokio::test] + async fn test_record_decommission_entry_error_cancels_listing_and_preserves_first_error() { + let entry_error = Arc::new(tokio::sync::Mutex::new(None)); + let rx = CancellationToken::new(); + + record_decommission_entry_error(&entry_error, &rx, Error::SlowDown).await; + record_decommission_entry_error(&entry_error, &rx, Error::OperationCanceled).await; + + assert!(rx.is_cancelled()); + assert!(matches!(*entry_error.lock().await, Some(Error::SlowDown))); + } + + #[tokio::test] + async fn test_record_decommission_entry_error_ignores_already_canceled_listing() { + let entry_error = Arc::new(tokio::sync::Mutex::new(None)); + let rx = CancellationToken::new(); + rx.cancel(); + + record_decommission_entry_error(&entry_error, &rx, Error::SlowDown).await; + + assert!(entry_error.lock().await.is_none()); + } + + #[test] + fn test_resolve_decommission_listing_error_preserves_real_listing_failure() { + let err = resolve_decommission_listing_error(Some(Error::SlowDown), Some(Error::OperationCanceled)) + .expect("listing failure should be returned"); + assert!(matches!(err, Error::SlowDown)); + + let err = resolve_decommission_listing_error(Some(Error::OperationCanceled), Some(Error::SlowDown)) + .expect("entry failure should be returned"); + assert!(matches!(err, Error::SlowDown)); + } + #[test] fn test_resolve_decommission_check_after_list_result_returns_list_result_without_entry_error() { let err = resolve_decommission_check_after_list_result(Err(Error::OperationCanceled), None)