From bf425ca32c58d337feb99b8714a8193d21b80b90 Mon Sep 17 00:00:00 2001 From: cxymds Date: Fri, 11 Sep 2026 13:46:35 +0800 Subject: [PATCH] fix(heal): reuse API error boundary for decoding failures (#7647) * fix(heal): reuse API error boundary for decoding failures * Update heal.rs Signed-off-by: houseme * test(ecstore): synchronize batch cleanup metadata reads Acquire object read locks while observing unversioned and explicit-version batch cleanup, then release them before waiting for progress. This prevents snapshots from spanning per-disk marker removal while retaining the existing quorum, timeout, and remote delete count assertions. Validation: cargo fmt --all --check and git diff --check passed. The focused nextest test passed 20 stress iterations each with test-util and test-util,rio-v2, with retries disabled. --------- Signed-off-by: houseme Co-authored-by: houseme --- crates/ecstore/src/store/init.rs | 44 ++++++++++++++++++++++--------- rustfs/src/admin/handlers/heal.rs | 7 ++--- 2 files changed, 36 insertions(+), 15 deletions(-) diff --git a/crates/ecstore/src/store/init.rs b/crates/ecstore/src/store/init.rs index a12b887ce..26e003f84 100644 --- a/crates/ecstore/src/store/init.rs +++ b/crates/ecstore/src/store/init.rs @@ -15721,12 +15721,20 @@ mod tests { ); tokio::time::timeout(Duration::from_secs(30), async { loop { - let metadata_absent = store.pools[0] - .get_disks_by_key(causal) - .load_file_info_versions_exact(bucket, causal) - .await - .expect("causal batch cleanup metadata should remain readable") - .is_none(); + let metadata_absent = { + // Synchronize with cleanup so the snapshot cannot span per-disk marker removal. + let mut read_opts = ObjectOptions::default(); + let _guards = store + .acquire_all_physical_object_read_locks("batch_transitioned_delete_test", bucket, causal, &mut read_opts) + .await + .expect("causal batch cleanup observation should acquire object read locks"); + store.pools[0] + .get_disks_by_key(causal) + .load_file_info_versions_exact(bucket, causal) + .await + .expect("causal batch cleanup metadata should remain readable") + .is_none() + }; if metadata_absent && backend.remove_versions().await.len() >= 2 { return; } @@ -15805,12 +15813,24 @@ mod tests { ); tokio::time::timeout(Duration::from_secs(30), async { loop { - let metadata_absent = store.pools[0] - .get_disks_by_key(versioned_causal) - .load_file_info_versions_exact(bucket, versioned_causal) - .await - .expect("versioned causal batch cleanup metadata should remain readable") - .is_none(); + let metadata_absent = { + let mut read_opts = ObjectOptions::default(); + let _guards = store + .acquire_all_physical_object_read_locks( + "batch_transitioned_delete_test", + bucket, + versioned_causal, + &mut read_opts, + ) + .await + .expect("versioned causal batch cleanup observation should acquire object read locks"); + store.pools[0] + .get_disks_by_key(versioned_causal) + .load_file_info_versions_exact(bucket, versioned_causal) + .await + .expect("versioned causal batch cleanup metadata should remain readable") + .is_none() + }; if metadata_absent && backend.remove_versions().await.len() == 3 { return; } diff --git a/rustfs/src/admin/handlers/heal.rs b/rustfs/src/admin/handlers/heal.rs index 195aca9b8..22dfcd438 100644 --- a/rustfs/src/admin/handlers/heal.rs +++ b/rustfs/src/admin/handlers/heal.rs @@ -17,6 +17,7 @@ use crate::admin::router::{AdminOperation, Operation, S3Router}; use crate::admin::runtime_sources::app_context_from_req; use crate::admin::storage_api::bucket::is_reserved_or_invalid_bucket; use crate::admin::storage_api::bucket::utils::is_valid_object_prefix; +use crate::error::ApiError; use crate::server::ADMIN_PREFIX; use crate::server::RemoteAddr; use crate::storage::rpc::node_service::heal::{ @@ -37,7 +38,7 @@ use rustfs_heal_contracts::heal_channel::{ use rustfs_policy::policy::action::{Action, AdminAction}; use rustfs_scanner::scanner::{BackgroundHealInfo, read_background_heal_info}; use s3s::header::{CONTENT_LENGTH, CONTENT_TYPE}; -use s3s::{Body, S3Error, S3ErrorCode, S3Request, S3Response, S3Result, s3_error}; +use s3s::{Body, S3Request, S3Response, S3Result, s3_error}; use serde::{Deserialize, Serialize}; use std::collections::{BTreeMap, BTreeSet, HashSet}; use std::future::Future; @@ -75,11 +76,11 @@ fn extract_heal_init_params(body: &Bytes, uri: &Uri, params: Params<'_, '_>) -> let mut hip = HealInitParams { bucket: percent_decode_str(params.get("bucket").unwrap_or_default()) .decode_utf8() - .map_err(|_| S3Error::with_message(S3ErrorCode::InvalidRequest, "invalid bucket name encoding"))? + .map_err(|_| ApiError::invalid_request("invalid bucket name encoding"))? .into_owned(), obj_prefix: percent_decode_str(params.get("prefix").unwrap_or_default()) .decode_utf8() - .map_err(|_| S3Error::with_message(S3ErrorCode::InvalidRequest, "invalid object name encoding"))? + .map_err(|_| ApiError::invalid_request("invalid object name encoding"))? .into_owned(), ..Default::default() };