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 <housemecn@gmail.com>

* 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 <housemecn@gmail.com>
Co-authored-by: houseme <housemecn@gmail.com>

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

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

scripts/scanner_abba.py

@@ -654,12 +654,9 @@ def collect_live(prepared, request, request_path, adapter):
     if request.get("evidence") == "measured":
         expected_metrics_endpoints = request["release_evidence"]["distributed"]["metrics_endpoints"]
     output = request_path.parent / "telemetry"
-    interval_seconds = 60
-    sample_count = request["duration_seconds"] // interval_seconds + 1
-    collector_timeout_seconds = (sample_count - 1) * interval_seconds + 300
     args = ["bash", str(collector), "--alias", connection["alias"], "--endpoint", connection["endpoint"],
             "--metrics-endpoints", connection["metrics_endpoints"], "--deployment", "distributed",
-            "--samples", str(sample_count), "--interval-secs", str(interval_seconds),
+            "--samples", str(request["duration_seconds"] // 60 + 1), "--interval-secs", "60",
             "--out-dir", str(output)]
     with (request_path.parent / "collector.log").open("wb") as log:
         process = OwnedCommand(args, log)
@@ -667,11 +664,10 @@ def collect_live(prepared, request, request_path, adapter):
             started = time.monotonic()
             result = invoke(adapter, "measure", request_path, request["duration_seconds"] + 300)
             require(time.monotonic() - started >= request["duration_seconds"], "measurement ended before required window")
-            require(process.wait(collector_timeout_seconds) == 0,
-                    f"scanner collector failed within {collector_timeout_seconds}s timeout")
+            require(process.wait(120) == 0, "scanner collector failed")
             require(output.joinpath("scanner-summary.csv").stat().st_size > 0, "missing collector samples")
             samples = list((output / "status").glob("scanner-status.*.json"))
-            require(len(samples) == sample_count, "missing scanner samples")
+            require(len(samples) == request["duration_seconds"] // 60 + 1, "missing scanner samples")
             for sample in samples:
                 status = read_json(sample)
                 require(isinstance(status.get("metrics"), dict) and status["metrics"], "invalid scanner status response")