ci(s3tests): omit proxy query strings

.github/workflows/e2e-s3tests.yml

@@ -90,6 +90,10 @@ on:
         description: "Optional pytest -m expression"
         required: false
         default: ""
+      testexpr:
+        description: "Optional pytest -k expression"
+        required: false
+        default: ""
   schedule:
     # Weekly full sweep (Sunday 02:00 UTC): full suite, run against BOTH the
     # single-node and the 4-node distributed topologies (matrix below).
@@ -116,6 +120,7 @@ env:
   XDIST: ${{ github.event.inputs.xdist || '4' }}
   MAXFAIL: ${{ github.event.inputs.maxfail || '0' }}
   MARKEXPR: ${{ github.event.inputs.markexpr || '' }}
+  TESTEXPR: ${{ github.event.inputs.testexpr || '' }}
   S3_SHARD_COUNT: ${{ github.event_name == 'schedule' && '4' || github.event.inputs.shard-count || '1' }}
   TEST_TIMEOUT: "300"
 
@@ -269,8 +274,13 @@ jobs:
           EOF
 
           cat > haproxy.cfg <<'EOF'
+          global
+            log stdout format raw local0 info
+
           defaults
             mode http
+            log global
+            log-format '%ci:%cp [%tr] %ft %b/%s %TR/%Tw/%Tc/%Tr/%Ta %ST %B %tsc %HM %HP'
             timeout connect 5s
             timeout client  30s
             timeout server  30s
@@ -314,6 +324,7 @@ jobs:
           XDIST="${XDIST}" \
           MAXFAIL="${MAXFAIL}" \
           MARKEXPR="${MARKEXPR}" \
+          TESTEXPR="${TESTEXPR}" \
           ./scripts/s3-tests/run.sh
 
       - name: Publish compatibility report

crates/e2e_test/src/cluster_concurrency_test.rs

@@ -15,12 +15,14 @@
 use crate::common::RustFSTestClusterEnvironment;
 use aws_sdk_s3::Client;
 use aws_sdk_s3::error::SdkError;
+use aws_sdk_s3::types::{CorsConfiguration, CorsRule};
 use bytes::Bytes;
 use std::sync::Arc;
 use tokio::sync::Barrier;
 use tracing::{info, warn};
 
 const BUCKET: &str = "conditional-put-race-bucket";
+const BUCKET_METADATA_RELOAD_BUCKET: &str = "bucket-metadata-reload-barrier";
 
 async fn cleanup_object(client: &Client, key: &str) {
     if let Err(e) = client.delete_object().bucket(BUCKET).key(key).send().await {
@@ -28,6 +30,16 @@ async fn cleanup_object(client: &Client, key: &str) {
     }
 }
 
+async fn assert_bucket_cors_missing(client: &Client) {
+    let result = client.get_bucket_cors().bucket(BUCKET_METADATA_RELOAD_BUCKET).send().await;
+    match result {
+        Err(SdkError::ServiceError(error)) => {
+            assert_eq!(error.err().meta().code(), Some("NoSuchCORSConfiguration"));
+        }
+        result => panic!("expected the peer to report a missing CORS configuration: {result:?}"),
+    }
+}
+
 async fn conditional_put(
     client: &Client,
     key: &str,
@@ -236,3 +248,48 @@ async fn test_conditional_put_basic_cluster() -> Result<(), Box<dyn std::error::
     cleanup_object(&client, test_key).await;
     Ok(())
 }
+
+#[tokio::test]
+async fn test_bucket_cors_write_is_visible_on_peer_before_response() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+    crate::common::init_logging();
+
+    let mut cluster = RustFSTestClusterEnvironment::new(2).await?;
+    cluster.start().await?;
+    cluster.create_test_bucket(BUCKET_METADATA_RELOAD_BUCKET).await?;
+
+    let writer = cluster.create_s3_client(0)?;
+    let reader = cluster.create_s3_client(1)?;
+    assert_bucket_cors_missing(&reader).await;
+
+    let rule = CorsRule::builder()
+        .allowed_methods("GET")
+        .allowed_origins("https://example.com")
+        .build()?;
+    let configuration = CorsConfiguration::builder().cors_rules(rule).build()?;
+
+    writer
+        .put_bucket_cors()
+        .bucket(BUCKET_METADATA_RELOAD_BUCKET)
+        .cors_configuration(configuration)
+        .send()
+        .await?;
+
+    let response = reader.get_bucket_cors().bucket(BUCKET_METADATA_RELOAD_BUCKET).send().await?;
+    let rules = response.cors_rules();
+    assert_eq!(
+        rules.len(),
+        1,
+        "peer should observe the committed CORS rule before the write response returns"
+    );
+    assert_eq!(rules[0].allowed_methods(), ["GET"]);
+    assert_eq!(rules[0].allowed_origins(), ["https://example.com"]);
+
+    writer
+        .delete_bucket_cors()
+        .bucket(BUCKET_METADATA_RELOAD_BUCKET)
+        .send()
+        .await?;
+    assert_bucket_cors_missing(&reader).await;
+    writer.delete_bucket().bucket(BUCKET_METADATA_RELOAD_BUCKET).send().await?;
+    Ok(())
+}

crates/ecstore/src/cluster/rpc/peer_rest_client.rs

@@ -85,6 +85,7 @@ const PEER_REST_RECOVERY_MAX_ATTEMPTS: u32 = 60;
 const PEER_REST_RECOVERY_MAX_BACKOFF: Duration = Duration::from_secs(30);
 const SCANNER_ACTIVITY_MAX_MESSAGE_SIZE: usize = 1024;
 const REPLICATION_STATS_MAX_MESSAGE_SIZE: usize = 8 * 1024 * 1024;
+const BUCKET_METADATA_RELOAD_TIMEOUT: Duration = Duration::from_secs(5);
 
 /// Error for a peer that reported `success = false` without an `error_info` payload.
 ///
@@ -1328,27 +1329,38 @@ impl PeerRestClient {
     }
 
     pub async fn load_bucket_metadata(&self, bucket: &str, scanner_maintenance_change: bool) -> Result<()> {
-        self.finalize_result(
-            async {
-                let mut client = self.get_client().await?;
-                let mut request = Request::new(LoadBucketMetadataRequest {
-                    bucket: bucket.to_string(),
-                    scanner_maintenance_change,
-                });
-                set_tonic_mutation_body_digest(&mut request)?;
-
-                let response = client.load_bucket_metadata(request).await?.into_inner();
-                if !response.success {
-                    if let Some(msg) = response.error_info {
-                        return Err(Error::other(msg));
-                    }
-                    return Err(peer_failure_without_details("load_bucket_metadata", Some(bucket)));
-                }
-                Ok(())
+        let result = tokio::time::timeout(BUCKET_METADATA_RELOAD_TIMEOUT, async {
+            let result = self.load_bucket_metadata_once(bucket, scanner_maintenance_change).await;
+            if let Err(err) = &result
+                && Self::is_network_like_error(err)
+            {
+                self.prepare_retry().await;
+                return self.load_bucket_metadata_once(bucket, scanner_maintenance_change).await;
             }
-            .await,
-        )
+            result
+        })
         .await
+        .unwrap_or_else(|_| Err(Error::other(format!("load_bucket_metadata({bucket}) timed out"))));
+        self.finalize_result(result).await
+    }
+
+    async fn load_bucket_metadata_once(&self, bucket: &str, scanner_maintenance_change: bool) -> Result<()> {
+        let mut client = self.get_client().await?;
+        let mut request = Request::new(LoadBucketMetadataRequest {
+            bucket: bucket.to_string(),
+            scanner_maintenance_change,
+        });
+        set_tonic_mutation_body_digest(&mut request)?;
+        request.set_timeout(BUCKET_METADATA_RELOAD_TIMEOUT);
+
+        let response = client.load_bucket_metadata(request).await?.into_inner();
+        if !response.success {
+            if let Some(msg) = response.error_info {
+                return Err(Error::other(msg));
+            }
+            return Err(peer_failure_without_details("load_bucket_metadata", Some(bucket)));
+        }
+        Ok(())
     }
 
     pub async fn delete_bucket_metadata(&self, bucket: &str) -> Result<()> {

crates/ecstore/src/cluster/rpc/remote_disk.rs

@@ -50,10 +50,10 @@ use rustfs_protos::evict_failed_connection;
 use rustfs_protos::proto_gen::node_service::RenamePartRequest;
 use rustfs_protos::proto_gen::node_service::{
     BatchReadVersionRequest, BatchReadVersionResponse, CheckPartsRequest, DeletePathsRequest, DeleteRequest,
-    DeleteVersionRequest, DeleteVersionsRequest, DeleteVolumeRequest, DiskInfoRequest, ListDirRequest, ListVolumesRequest,
-    MakeVolumeRequest, MakeVolumesRequest, PreparePartTransactionRequest, ReadAllRequest, ReadMetadataRequest,
-    ReadMultipleRequest, ReadMultipleResponse, ReadPartsRequest, ReadVersionRequest, ReadXlRequest, RenameDataRequest,
-    RenameFileRequest, SettlePartTransactionRequest, SnapshotLeaseReleaseRequest, SnapshotLeaseRenewRequest,
+    DeleteVersionRequest, DeleteVersionsRequest, DeleteVersionsResponse, DeleteVolumeRequest, DiskInfoRequest, ListDirRequest,
+    ListVolumesRequest, MakeVolumeRequest, MakeVolumesRequest, PreparePartTransactionRequest, ReadAllRequest,
+    ReadMetadataRequest, ReadMultipleRequest, ReadMultipleResponse, ReadPartsRequest, ReadVersionRequest, ReadXlRequest,
+    RenameDataRequest, RenameFileRequest, SettlePartTransactionRequest, SnapshotLeaseReleaseRequest, SnapshotLeaseRenewRequest,
     SnapshotLeaseRequest, SnapshotLeaseResponse, StatVolumeRequest, UpdateMetadataRequest, VerifyFileRequest, WriteAllRequest,
     WriteMetadataRequest, node_service_client::NodeServiceClient,
 };
@@ -112,6 +112,28 @@ const EVENT_REMOTE_DISK_RPC: &str = "remote_disk_rpc";
 const SNAPSHOT_LEASE_PROTOCOL_VERSION: u32 = 1;
 pub const REMOTE_SNAPSHOT_LEASE_TTL: Duration = Duration::from_secs(60);
 
+fn decode_delete_versions_errors(response: DeleteVersionsResponse, expected_len: usize) -> Vec<Option<Error>> {
+    if !response.item_errors.is_empty() {
+        if response.item_errors.len() != expected_len {
+            return vec![Some(Error::other("malformed delete_versions item errors")); expected_len];
+        }
+        return response
+            .item_errors
+            .into_iter()
+            .map(|error| (error.code != 0).then(|| error.into()))
+            .collect();
+    }
+
+    if response.errors.len() != expected_len {
+        return vec![Some(Error::other("malformed delete_versions errors")); expected_len];
+    }
+    response
+        .errors
+        .into_iter()
+        .map(|error| (!error.is_empty()).then(|| Error::other(error)))
+        .collect()
+}
+
 fn snapshot_lease_token_from_response(response: SnapshotLeaseResponse) -> Result<SnapshotLeaseToken> {
     if !response.success {
         return Err(response.error.unwrap_or_default().into());
@@ -2406,8 +2428,6 @@ impl DiskAPI for RemoteDisk {
             return errors;
         }
 
-        // TODO(backlog): replace string errors with typed `StorageError` variants
-
         let result = self
             .execute_with_timeout(
                 || async {
@@ -2439,17 +2459,7 @@ impl DiskAPI for RemoteDisk {
             }
             return errors;
         }
-        response
-            .errors
-            .iter()
-            .map(|error| {
-                if error.is_empty() {
-                    None
-                } else {
-                    Some(Error::other(error.to_string()))
-                }
-            })
-            .collect()
+        decode_delete_versions_errors(response, versions.len())
     }
 
     #[tracing::instrument(level = "trace", skip_all)]
@@ -3760,6 +3770,63 @@ mod tests {
 
     static INIT: Once = Once::new();
 
+    #[test]
+    fn delete_versions_response_preserves_typed_item_errors() {
+        let errors = decode_delete_versions_errors(
+            DeleteVersionsResponse {
+                success: true,
+                errors: vec!["file not found".to_string(), String::new()],
+                error: None,
+                item_errors: vec![
+                    rustfs_protos::proto_gen::node_service::Error {
+                        code: DiskError::FileNotFound.to_u32(),
+                        error_info: "file not found".to_string(),
+                    },
+                    rustfs_protos::proto_gen::node_service::Error::default(),
+                ],
+            },
+            2,
+        );
+
+        assert!(matches!(errors.as_slice(), [Some(DiskError::FileNotFound), None]));
+    }
+
+    #[test]
+    fn delete_versions_response_accepts_legacy_string_errors() {
+        let errors = decode_delete_versions_errors(
+            DeleteVersionsResponse {
+                success: true,
+                errors: vec!["legacy error".to_string(), String::new()],
+                error: None,
+                item_errors: Vec::new(),
+            },
+            2,
+        );
+
+        assert_eq!(errors.len(), 2);
+        assert_eq!(errors[0].as_ref().map(ToString::to_string).as_deref(), Some("io error legacy error"));
+        assert!(errors[1].is_none());
+    }
+
+    #[test]
+    fn delete_versions_response_rejects_misaligned_item_errors() {
+        let errors = decode_delete_versions_errors(
+            DeleteVersionsResponse {
+                success: true,
+                errors: vec!["file not found".to_string()],
+                error: None,
+                item_errors: vec![rustfs_protos::proto_gen::node_service::Error {
+                    code: DiskError::FileNotFound.to_u32(),
+                    error_info: "file not found".to_string(),
+                }],
+            },
+            2,
+        );
+
+        assert_eq!(errors.len(), 2);
+        assert!(errors.iter().all(Option::is_some));
+    }
+
     #[test]
     fn disk_mutation_digest_marks_rolling_compatibility() {
         let mut request = Request::new(());

crates/ecstore/src/core/pools.rs

@@ -1808,7 +1808,7 @@ impl PoolMeta {
         self.load_no_lock(pool).await
     }
 
-    pub(crate) async fn load_no_lock<S>(&mut self, pool: Arc<S>) -> Result<()>
+    async fn load_no_lock<S>(&mut self, pool: Arc<S>) -> Result<()>
     where
         S: EcstoreObjectIO,
     {

crates/ecstore/src/core/sets.rs

@@ -985,11 +985,14 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for Sets {
     }
 }
 
-impl Sets {
-    pub(crate) async fn heal_format_with_fence<F>(&self, dry_run: bool, fence_lost: F) -> Result<(HealResultItem, Option<Error>)>
-    where
-        F: Fn() -> bool + Send + Sync,
-    {
+#[async_trait::async_trait]
+impl crate::storage_api_contracts::heal::HealOperations for Sets {
+    type Error = Error;
+    type HealResultItem = HealResultItem;
+    type HealOptions = HealOpts;
+
+    #[tracing::instrument(skip(self))]
+    async fn heal_format(&self, dry_run: bool) -> Result<(HealResultItem, Option<Error>)> {
         let (disks, init_errs) = init_storage_disks_with_errors(
             &self.endpoints.endpoints,
             &DiskOption {
@@ -1062,9 +1065,6 @@ impl Sets {
             // Save new formats `format.json` on unformatted disks.
             for (index, (fm, disk)) in tmp_new_formats.iter_mut().zip(disks.iter()).enumerate() {
                 if fm.is_some() && disk.is_some() {
-                    if fence_lost() {
-                        return Ok((res, Some(StorageError::SlowDown)));
-                    }
                     if let Err(err) = save_format_file(disk, fm).await {
                         if let Some(disk) = disk.as_ref() {
                             let _ = disk.close().await;
@@ -1098,18 +1098,6 @@ impl Sets {
         }
         Ok((res, None))
     }
-}
-
-#[async_trait::async_trait]
-impl crate::storage_api_contracts::heal::HealOperations for Sets {
-    type Error = Error;
-    type HealResultItem = HealResultItem;
-    type HealOptions = HealOpts;
-
-    #[tracing::instrument(skip(self))]
-    async fn heal_format(&self, dry_run: bool) -> Result<(HealResultItem, Option<Error>)> {
-        self.heal_format_with_fence(dry_run, || false).await
-    }
     #[tracing::instrument(skip(self))]
     async fn heal_bucket(&self, bucket: &str, opts: &HealOpts) -> Result<HealResultItem> {
         let mut result = HealResultItem {

crates/ecstore/src/store/heal.rs

@@ -13,12 +13,7 @@
 // limitations under the License.
 
 use super::*;
-use crate::core::pools::POOL_META_NAME;
-use crate::services::rebalance::{REBAL_META_NAME, RebalStatus};
-use crate::set_disk::get_lock_acquire_timeout;
 use crate::storage_api_contracts::heal::HealOperations as _;
-use crate::storage_api_contracts::namespace::NamespaceLocking as _;
-use rustfs_lock::NamespaceLockGuard;
 use tracing::trace;
 
 const LOG_COMPONENT_ECSTORE: &str = "ecstore";
@@ -35,119 +30,7 @@ fn invalid_heal_pool_index(pool_idx: usize, pool_count: usize) -> Error {
     )
 }
 
-#[derive(Debug, Clone, Copy)]
-enum HealFormatPoolSkip {
-    Completed,
-    Retryable,
-}
-
-fn classify_heal_format_pool(
-    pool_idx: usize,
-    pool_cmd_line: &str,
-    pool_meta: &PoolMeta,
-    rebalance_meta: Option<&RebalanceMeta>,
-) -> Option<HealFormatPoolSkip> {
-    let Some(pool) = pool_meta.pools.get(pool_idx) else {
-        return Some(HealFormatPoolSkip::Retryable);
-    };
-
-    if pool.id != pool_idx || pool_cmd_line.is_empty() || pool.cmd_line.is_empty() || pool.cmd_line != pool_cmd_line {
-        return Some(HealFormatPoolSkip::Retryable);
-    }
-
-    if let Some(decommission) = pool.decommission.as_ref() {
-        if decommission.complete {
-            return Some(HealFormatPoolSkip::Completed);
-        }
-        if decommission.failed || decommission.canceled || decommission.queued || pool_meta.is_suspended(pool_idx) {
-            return Some(HealFormatPoolSkip::Retryable);
-        }
-    }
-
-    if let Some(meta) = rebalance_meta {
-        let Some(pool_stats) = meta.pool_stats.get(pool_idx) else {
-            return Some(HealFormatPoolSkip::Retryable);
-        };
-        if pool_stats.info.stopping || (pool_stats.participating && pool_stats.info.status == RebalStatus::Started) {
-            return Some(HealFormatPoolSkip::Retryable);
-        }
-    }
-
-    None
-}
-
-fn heal_format_pool_skip_error(skip: HealFormatPoolSkip) -> Error {
-    match skip {
-        HealFormatPoolSkip::Completed => StorageError::NoHealRequired,
-        HealFormatPoolSkip::Retryable => StorageError::SlowDown,
-    }
-}
-
-fn heal_format_fence_lost_error() -> Error {
-    StorageError::SlowDown
-}
-
 impl ECStore {
-    async fn acquire_heal_format_fence(
-        &self,
-    ) -> Result<(NamespaceLockGuard, NamespaceLockGuard, PoolMeta, Option<RebalanceMeta>)> {
-        let metadata_pool = self
-            .pools
-            .first()
-            .cloned()
-            .ok_or_else(|| Error::other("heal format requires at least one storage pool"))?;
-
-        // Metadata fence order is part of the decommission/rebalance protocol:
-        // pool.bin must always be acquired before rebalance.bin.
-        let pool_lock = metadata_pool.new_ns_lock(RUSTFS_META_BUCKET, POOL_META_NAME).await?;
-        let pool_guard = pool_lock.get_write_lock(get_lock_acquire_timeout()).await?;
-        let rebalance_lock = metadata_pool.new_ns_lock(RUSTFS_META_BUCKET, REBAL_META_NAME).await?;
-        let rebalance_guard = rebalance_lock.get_write_lock(get_lock_acquire_timeout()).await?;
-
-        if pool_guard.is_lock_lost() || rebalance_guard.is_lock_lost() {
-            return Err(heal_format_fence_lost_error());
-        }
-
-        let mut pool_meta = PoolMeta::default();
-        pool_meta.load_no_lock(metadata_pool.clone()).await?;
-        if pool_meta.pools.len() != self.pools.len()
-            || pool_meta.pools.iter().enumerate().any(|(pool_idx, pool)| {
-                pool.id != pool_idx || pool.cmd_line.is_empty() || pool.cmd_line != self.pools[pool_idx].endpoints.cmd_line
-            })
-        {
-            return Err(heal_format_fence_lost_error());
-        }
-
-        let mut rebalance_meta = RebalanceMeta::new();
-        let rebalance_meta = match rebalance_meta
-            .load_with_opts(
-                metadata_pool,
-                ObjectOptions {
-                    no_lock: true,
-                    ..Default::default()
-                },
-            )
-            .await
-        {
-            Ok(()) => Some(rebalance_meta),
-            Err(Error::ConfigNotFound) => None,
-            Err(err) => return Err(err),
-        };
-
-        if rebalance_meta
-            .as_ref()
-            .is_some_and(|meta| meta.pool_stats.len() != self.pools.len())
-        {
-            return Err(heal_format_fence_lost_error());
-        }
-
-        if pool_guard.is_lock_lost() || rebalance_guard.is_lock_lost() {
-            return Err(heal_format_fence_lost_error());
-        }
-
-        Ok((pool_guard, rebalance_guard, pool_meta, rebalance_meta))
-    }
-
     fn get_pools_for_heal_object(&self, opts: &HealOpts) -> Result<Vec<Arc<Sets>>> {
         match opts.pool {
             Some(pool_idx) => Ok(vec![
@@ -169,26 +52,9 @@ impl ECStore {
         };
 
         let mut count_no_heal = 0;
-        let mut count_completed = 0;
         let mut first_error = None;
-        for (pool_idx, pool) in self.pools.iter().enumerate() {
-            let (pool_guard, rebalance_guard, pool_meta, rebalance_meta) = self.acquire_heal_format_fence().await?;
-            if pool_guard.is_lock_lost() || rebalance_guard.is_lock_lost() {
-                first_error.get_or_insert(heal_format_fence_lost_error());
-                break;
-            }
-            if let Some(skip) = classify_heal_format_pool(pool_idx, &pool.endpoints.cmd_line, &pool_meta, rebalance_meta.as_ref())
-            {
-                if matches!(skip, HealFormatPoolSkip::Completed) {
-                    count_completed += 1;
-                } else {
-                    first_error.get_or_insert(heal_format_pool_skip_error(skip));
-                }
-                continue;
-            }
-
-            let fence_lost = || pool_guard.is_lock_lost() || rebalance_guard.is_lock_lost();
-            let (mut result, err) = pool.heal_format_with_fence(dry_run, fence_lost).await?;
+        for pool in self.pools.iter() {
+            let (mut result, err) = pool.heal_format(dry_run).await?;
             if let Some(err) = err {
                 match err {
                     StorageError::NoHealRequired => {
@@ -203,18 +69,11 @@ impl ECStore {
             r.set_count += result.set_count;
             r.before.drives.append(&mut result.before.drives);
             r.after.drives.append(&mut result.after.drives);
-
-            // A lease can be lost after the final write; fail closed before
-            // reporting the pool as successfully healed.
-            if pool_guard.is_lock_lost() || rebalance_guard.is_lock_lost() {
-                first_error.get_or_insert(heal_format_fence_lost_error());
-                break;
-            }
         }
         if let Some(err) = first_error {
             return Ok((r, Some(err)));
         }
-        if count_no_heal + count_completed == self.pools.len() {
+        if count_no_heal == self.pools.len() {
             info!(
                 event = EVENT_HEAL_FORMAT_COMPLETED,
                 component = LOG_COMPONENT_ECSTORE,
@@ -441,7 +300,6 @@ mod tests {
     use crate::core::pools::{PoolDecommissionInfo, PoolStatus};
     use crate::disk::{DiskOption, format::FormatV3, new_disk};
     use crate::layout::endpoints::{Endpoints, PoolEndpoints};
-    use crate::services::rebalance::{RebalanceInfo, RebalanceStats};
     use crate::store::init_format::{load_format_erasure, save_format_file};
 
     async fn minimal_heal_pool(pool_idx: usize) -> Arc<Sets> {
@@ -489,164 +347,6 @@ mod tests {
         }
     }
 
-    fn pool_meta_with_decommission(info: PoolDecommissionInfo) -> PoolMeta {
-        PoolMeta {
-            pools: vec![PoolStatus {
-                id: 0,
-                cmd_line: "pool-0".to_string(),
-                last_update: OffsetDateTime::UNIX_EPOCH,
-                decommission: Some(info),
-            }],
-            ..Default::default()
-        }
-    }
-
-    #[test]
-    fn heal_format_pool_state_barriers_are_classified() {
-        let active = pool_meta_with_decommission(PoolDecommissionInfo {
-            start_time: Some(OffsetDateTime::UNIX_EPOCH),
-            ..Default::default()
-        });
-        assert!(matches!(
-            classify_heal_format_pool(0, "pool-0", &active, None),
-            Some(HealFormatPoolSkip::Retryable)
-        ));
-
-        for info in [
-            PoolDecommissionInfo {
-                failed: true,
-                ..Default::default()
-            },
-            PoolDecommissionInfo {
-                canceled: true,
-                ..Default::default()
-            },
-        ] {
-            assert!(matches!(
-                classify_heal_format_pool(0, "pool-0", &pool_meta_with_decommission(info), None),
-                Some(HealFormatPoolSkip::Retryable)
-            ));
-        }
-
-        let completed = pool_meta_with_decommission(PoolDecommissionInfo {
-            complete: true,
-            ..Default::default()
-        });
-        assert!(matches!(
-            classify_heal_format_pool(0, "pool-0", &completed, None),
-            Some(HealFormatPoolSkip::Completed)
-        ));
-    }
-
-    #[test]
-    fn heal_format_pool_rebalance_barriers_and_identity_are_fail_closed() {
-        let identity_meta = pool_meta_with_decommission(PoolDecommissionInfo::default());
-        let rebalance = RebalanceMeta {
-            pool_stats: vec![RebalanceStats {
-                participating: true,
-                info: RebalanceInfo {
-                    status: RebalStatus::Started,
-                    ..Default::default()
-                },
-                ..Default::default()
-            }],
-            ..Default::default()
-        };
-        assert!(matches!(
-            classify_heal_format_pool(0, "pool-0", &identity_meta, Some(&rebalance)),
-            Some(HealFormatPoolSkip::Retryable)
-        ));
-
-        let stopping = RebalanceMeta {
-            pool_stats: vec![RebalanceStats {
-                info: RebalanceInfo {
-                    stopping: true,
-                    ..Default::default()
-                },
-                ..Default::default()
-            }],
-            ..Default::default()
-        };
-        assert!(matches!(
-            classify_heal_format_pool(0, "pool-0", &identity_meta, Some(&stopping)),
-            Some(HealFormatPoolSkip::Retryable)
-        ));
-
-        let identity = pool_meta_with_decommission(PoolDecommissionInfo::default());
-        assert!(matches!(
-            classify_heal_format_pool(0, "pool-new", &identity, None),
-            Some(HealFormatPoolSkip::Retryable)
-        ));
-
-        let identity_without_decommission = PoolMeta {
-            pools: vec![PoolStatus {
-                id: 0,
-                cmd_line: "pool-0".to_string(),
-                last_update: OffsetDateTime::UNIX_EPOCH,
-                decommission: None,
-            }],
-            ..Default::default()
-        };
-        assert!(matches!(
-            classify_heal_format_pool(0, "pool-new", &identity_without_decommission, None),
-            Some(HealFormatPoolSkip::Retryable)
-        ));
-
-        assert!(matches!(
-            classify_heal_format_pool(0, "", &identity_meta, None),
-            Some(HealFormatPoolSkip::Retryable)
-        ));
-
-        assert!(matches!(
-            classify_heal_format_pool(0, "pool-0", &PoolMeta::default(), None),
-            Some(HealFormatPoolSkip::Retryable)
-        ));
-
-        let stopped = RebalanceMeta {
-            stopped_at: Some(OffsetDateTime::UNIX_EPOCH),
-            pool_stats: vec![RebalanceStats {
-                participating: true,
-                info: RebalanceInfo {
-                    status: RebalStatus::Stopped,
-                    ..Default::default()
-                },
-                ..Default::default()
-            }],
-            ..Default::default()
-        };
-        assert!(classify_heal_format_pool(0, "pool-0", &identity_meta, Some(&stopped)).is_none());
-
-        let stopping_after_stop = RebalanceMeta {
-            stopped_at: Some(OffsetDateTime::UNIX_EPOCH),
-            pool_stats: vec![RebalanceStats {
-                participating: true,
-                info: RebalanceInfo {
-                    status: RebalStatus::Started,
-                    stopping: true,
-                    ..Default::default()
-                },
-                ..Default::default()
-            }],
-            ..Default::default()
-        };
-        assert!(matches!(
-            classify_heal_format_pool(0, "pool-0", &identity_meta, Some(&stopping_after_stop)),
-            Some(HealFormatPoolSkip::Retryable)
-        ));
-    }
-
-    #[test]
-    fn skipped_heal_format_pool_is_never_reported_as_success() {
-        assert!(matches!(
-            heal_format_pool_skip_error(HealFormatPoolSkip::Retryable),
-            StorageError::SlowDown
-        ));
-        assert!(matches!(
-            heal_format_pool_skip_error(HealFormatPoolSkip::Completed),
-            StorageError::NoHealRequired
-        ));
-    }
-
     #[tokio::test]
     async fn heal_object_pool_scope_selects_only_requested_pool() {
         let store = minimal_heal_store().await;
@@ -915,18 +615,6 @@ mod tests {
             bucket_fence_registry: std::sync::Arc::default(),
         };
 
-        let err = store
-            .handle_heal_format(false)
-            .await
-            .expect_err("missing pool metadata must fail closed before format writes");
-        assert!(matches!(err, StorageError::SlowDown));
-
-        let pool_meta = PoolMeta::new(&store.pools, &PoolMeta::default());
-        pool_meta
-            .save(store.pools.clone())
-            .await
-            .expect("pool metadata should be persisted before format heal");
-
         let (result, err) = store
             .handle_heal_format(false)
             .await
@@ -940,22 +628,5 @@ mod tests {
             .await
             .expect("the later pool should be healed despite the first pool error");
         assert_eq!(healed.erasure.this, recoverable_format.erasure.sets[0][2]);
-
-        let mut completed_meta = PoolMeta::new(&store.pools, &PoolMeta::default());
-        for status in &mut completed_meta.pools {
-            status.decommission = Some(PoolDecommissionInfo {
-                complete: true,
-                ..Default::default()
-            });
-        }
-        completed_meta
-            .save(store.pools.clone())
-            .await
-            .expect("completed pool metadata should be persisted");
-        let (_, err) = store
-            .handle_heal_format(false)
-            .await
-            .expect("completed pools should be reported as a no-op");
-        assert!(matches!(err, Some(StorageError::NoHealRequired)));
     }
 }

crates/heal/src/heal/channel.rs

@@ -1640,6 +1640,60 @@ mod tests {
         assert_eq!(payload["items"].as_array().expect("items should be an array").len(), 0);
     }
 
+    #[tokio::test]
+    async fn test_process_query_request_reports_displaced_terminal_detail() {
+        let heal_manager = Arc::new(HealManager::new(
+            Arc::new(MockStorage),
+            Some(HealConfig {
+                queue_size: 1,
+                ..HealConfig::default()
+            }),
+        ));
+        let mut displaced = HealRequest::new(
+            HealType::Bucket {
+                bucket: "displaced-channel".to_string(),
+            },
+            HealOptions::default(),
+            HealPriority::Low,
+        );
+        displaced.id = "displaced-channel-task".to_string();
+        let displaced_id = displaced.id.clone();
+        heal_manager
+            .submit_heal_request(displaced)
+            .await
+            .expect("initial channel task should queue");
+        heal_manager
+            .submit_heal_request(HealRequest::new(
+                HealType::Bucket {
+                    bucket: "successor-channel".to_string(),
+                },
+                HealOptions::default(),
+                HealPriority::High,
+            ))
+            .await
+            .expect("successor channel task should displace the initial task");
+
+        let processor = HealChannelProcessor::new(heal_manager);
+        let (tx, rx) = oneshot::channel();
+        processor
+            .process_query_request("displaced-channel".to_string(), displaced_id, None, tx)
+            .await
+            .expect("displaced query should process");
+        let response = rx
+            .await
+            .expect("query response should be returned")
+            .expect("displaced query should remain successful");
+        let payload: serde_json::Value = serde_json::from_slice(response.data.as_deref().expect("status payload should exist"))
+            .expect("status payload should be json");
+        assert_eq!(payload["summary"], "stopped");
+        assert!(
+            response
+                .error
+                .as_deref()
+                .is_some_and(|detail| detail.contains("reason=displaced"))
+        );
+    }
+
     #[tokio::test]
     async fn test_process_query_request_reports_running_for_queued_task() {
         let heal_manager = create_test_heal_manager();

crates/heal/src/heal/manager.rs

@@ -40,6 +40,7 @@ use tracing::{debug, error, info, warn};
 use super::{DiskError, Endpoint, HealDiskExt as _, local_disk_map_read};
 
 const KEEP_HEAL_TASK_STATUS_DURATION: Duration = Duration::from_secs(10 * 60);
+const DISPLACED_HEAL_REASON: &str = "reason=displaced; retry_hint=submit_again";
 const LOG_COMPONENT_HEAL: &str = "heal";
 const LOG_SUBSYSTEM_DISK_SCANNER: &str = "disk_scanner";
 const LOG_SUBSYSTEM_MANAGER: &str = "manager";
@@ -120,26 +121,30 @@ struct MrfRepairNoticeTarget {
     version_id: Option<[u8; 16]>,
 }
 
-#[derive(Debug, Clone, PartialEq, Eq)]
+#[derive(Debug, Clone)]
 struct HealAdmissionDecision {
     result: HealAdmissionResult,
-    displaced_task_id: Option<String>,
+    displaced_request: Option<HealRequest>,
 }
 
 impl HealAdmissionDecision {
     const fn new(result: HealAdmissionResult) -> Self {
         Self {
             result,
-            displaced_task_id: None,
+            displaced_request: None,
         }
     }
 
-    fn accepted_with_displacement(displaced_task_id: String) -> Self {
+    fn accepted_with_displacement(displaced_request: HealRequest) -> Self {
         Self {
             result: HealAdmissionResult::Accepted,
-            displaced_task_id: Some(displaced_task_id),
+            displaced_request: Some(displaced_request),
         }
     }
+
+    fn displaced_task_id(&self) -> Option<&str> {
+        self.displaced_request.as_ref().map(|request| request.id.as_str())
+    }
 }
 
 fn lock_mrf_repair_notice_targets(
@@ -151,6 +156,55 @@ fn lock_mrf_repair_notice_targets(
     }
 }
 
+fn lock_displaced_terminals(
+    registry: &StdMutex<HashMap<String, Arc<CompletedHealStatus>>>,
+) -> StdMutexGuard<'_, HashMap<String, Arc<CompletedHealStatus>>> {
+    match registry.lock() {
+        Ok(guard) => guard,
+        Err(poisoned) => poisoned.into_inner(),
+    }
+}
+
+fn record_displaced_terminal(
+    registry: &StdMutex<HashMap<String, Arc<CompletedHealStatus>>>,
+    request: &HealRequest,
+) -> Arc<CompletedHealStatus> {
+    let terminal = Arc::new(CompletedHealStatus {
+        heal_type: request.heal_type.clone(),
+        status: HealTaskStatus::Failed {
+            error: format!("heal task displaced by a higher-priority request ({DISPLACED_HEAL_REASON})"),
+        },
+        result_items_truncated: false,
+        completed_at: SystemTime::now(),
+        seqed_items: Vec::new(),
+        next_seq: 0,
+        min_seq: 0,
+    });
+    let mut terminals = lock_displaced_terminals(registry);
+    prune_completed_heal_statuses(&mut terminals);
+    terminals.insert(request.id.clone(), Arc::clone(&terminal));
+    terminal
+}
+
+async fn remove_displaced_task_aliases(
+    aliases: &Arc<Mutex<HashMap<String, HealTaskAlias>>>,
+    terminals: &StdMutex<HashMap<String, Arc<CompletedHealStatus>>>,
+    task_id: &str,
+    terminal: &Arc<CompletedHealStatus>,
+) {
+    let mut aliases = aliases.lock().await;
+    let alias_ids = aliases
+        .iter()
+        .filter_map(|(alias_id, alias)| (alias.task_id == task_id).then_some(alias_id.clone()))
+        .collect::<Vec<_>>();
+    let mut displaced_terminals = lock_displaced_terminals(terminals);
+    prune_completed_heal_statuses(&mut displaced_terminals);
+    for alias_id in alias_ids {
+        displaced_terminals.insert(alias_id, Arc::clone(terminal));
+    }
+    aliases.retain(|alias_id, alias| alias_id != task_id && alias.task_id != task_id);
+}
+
 async fn remove_task_aliases_for_task(registry: &Arc<Mutex<HashMap<String, HealTaskAlias>>>, task_id: &str) {
     registry
         .lock()
@@ -618,6 +672,14 @@ pub struct HealManager {
     /// are shared so the lookup helper can hand a completed entry to a
     /// caller without cloning the retained result window.
     completed_heals: Arc<Mutex<HashMap<String, Arc<CompletedHealStatus>>>>,
+    /// Terminals for requests removed by priority displacement. An Accepted
+    /// task ID remains queryable for the same process lifetime and the normal
+    /// ten-minute status TTL; clients should treat `reason=displaced` as a
+    /// terminal result and submit a fresh request. This sidecar is synchronous
+    /// so admission can publish the terminal while the queue transition is
+    /// still under its lock, without awaiting another tokio lock. Queue state
+    /// is process-local, so this guarantee does not extend across restart.
+    displaced_terminals: Arc<StdMutex<HashMap<String, Arc<CompletedHealStatus>>>>,
     /// Client tokens merged into an existing task id.
     task_aliases: Arc<Mutex<HashMap<String, HealTaskAlias>>>,
     /// Heal tasks waiting for a retry backoff to expire.
@@ -659,6 +721,7 @@ struct HealQueueContext<'a> {
     heal_queue: &'a Arc<Mutex<PriorityHealQueue>>,
     active_heals: &'a Arc<Mutex<HashMap<String, Arc<HealTask>>>>,
     completed_heals: &'a Arc<Mutex<HashMap<String, Arc<CompletedHealStatus>>>>,
+    displaced_terminals: &'a Arc<StdMutex<HashMap<String, Arc<CompletedHealStatus>>>>,
     task_aliases: &'a Arc<Mutex<HashMap<String, HealTaskAlias>>>,
     retrying_heals: &'a Arc<Mutex<HashMap<String, RetryingHeal>>>,
     mrf_repair_notice_targets: &'a Arc<StdMutex<HashMap<String, Vec<MrfRepairNoticeTarget>>>>,
@@ -874,7 +937,7 @@ impl HealManager {
                         result = "accepted_by_displacement",
                         "Heal queue request accepted by displacement"
                     });
-                    return HealAdmissionDecision::accepted_with_displacement(displaced.id);
+                    return HealAdmissionDecision::accepted_with_displacement(displaced);
                 }
 
                 demote_to_debug_when!(per_object_request, warn, target: "rustfs::heal::manager", {
@@ -1105,6 +1168,7 @@ impl HealManager {
             active_heals: Arc::new(Mutex::new(HashMap::new())),
             heal_queue: Arc::new(Mutex::new(PriorityHealQueue::new())),
             completed_heals: Arc::new(Mutex::new(HashMap::new())),
+            displaced_terminals: Arc::new(StdMutex::new(HashMap::new())),
             task_aliases: Arc::new(Mutex::new(HashMap::new())),
             retrying_heals: Arc::new(Mutex::new(HashMap::new())),
             mrf_repair_notice_targets: Arc::new(StdMutex::new(HashMap::new())),
@@ -1209,6 +1273,10 @@ impl HealManager {
         active_heals.clear();
         publish_active_heal_count(&active_heals);
         self.completed_heals.lock().await.clear();
+        // Do not let the synchronous guard live across the following async lock.
+        {
+            lock_displaced_terminals(&self.displaced_terminals).clear();
+        }
         self.task_aliases.lock().await.clear();
         self.retrying_heals.lock().await.clear();
         lock_mrf_repair_notice_targets(&self.mrf_repair_notice_targets).clear();
@@ -1459,7 +1527,11 @@ impl HealManager {
             task_id = queued_id.to_owned();
         }
         let should_notify = matches!(admission, HealAdmissionResult::Accepted) && config.event_driven_scheduler_enable;
-        let displaced_task_id = admission_decision.displaced_task_id;
+        let displaced_task_id = admission_decision.displaced_task_id().map(ToOwned::to_owned);
+        let displaced_terminal = admission_decision
+            .displaced_request
+            .as_ref()
+            .map(|request| record_displaced_terminal(&self.displaced_terminals, request));
         if matches!(admission, HealAdmissionResult::Accepted | HealAdmissionResult::Merged)
             && let Some(target) = mrf_notice_target
         {
@@ -1473,8 +1545,12 @@ impl HealManager {
         drop(queue);
         drop(active_heals);
 
-        if let Some(displaced_task_id) = displaced_task_id {
-            self.remove_aliases_for_task(&displaced_task_id).await;
+        if let (Some(displaced_task_id), Some(displaced_terminal)) = (displaced_task_id, displaced_terminal) {
+            // The queue has already removed the displaced request, so the
+            // synchronous terminal sidecar was published before aliases and
+            // MRF ownership are cleaned up.
+            remove_displaced_task_aliases(&self.task_aliases, &self.displaced_terminals, &displaced_task_id, &displaced_terminal)
+                .await;
         }
 
         if should_notify {
@@ -1549,6 +1625,15 @@ impl HealManager {
             }
         }
 
+        if terminal_completed.is_none() {
+            let mut displaced_terminals = lock_displaced_terminals(&self.displaced_terminals);
+            prune_completed_heal_statuses(&mut displaced_terminals);
+            terminal_completed = displaced_terminals
+                .get(canonical_task_id)
+                .filter(|terminal| matches_path(&terminal.heal_type))
+                .cloned();
+        }
+
         match terminal_completed {
             Some(completed) => TaskStateLookup::Completed(completed),
             None => TaskStateLookup::NotFound,
@@ -1669,9 +1754,19 @@ impl HealManager {
 
         let mut completed_heals = self.completed_heals.lock().await;
         prune_completed_heal_statuses(&mut completed_heals);
-        completed_heals
+        if completed_heals
             .values()
             .any(|completed| heal_type_matches_path(&completed.heal_type, heal_path))
+        {
+            return true;
+        }
+        drop(completed_heals);
+
+        let mut displaced_terminals = lock_displaced_terminals(&self.displaced_terminals);
+        prune_completed_heal_statuses(&mut displaced_terminals);
+        displaced_terminals
+            .values()
+            .any(|terminal| heal_type_matches_path(&terminal.heal_type, heal_path))
     }
 
     /// Get task progress

crates/heal/src/heal/manager/auto_scan.rs

@@ -21,6 +21,7 @@ impl HealManager {
         let heal_queue = self.heal_queue.clone();
         let active_heals = self.active_heals.clone();
         let task_aliases = self.task_aliases.clone();
+        let displaced_terminals = self.displaced_terminals.clone();
         let mrf_repair_notice_targets = self.mrf_repair_notice_targets.clone();
         let storage = self.storage.clone();
         let replacement_recovery_anchors = self.replacement_recovery_anchors.clone();
@@ -481,6 +482,10 @@ impl HealManager {
                             let admission = admission_decision.result;
                             let should_notify =
                                 matches!(admission, HealAdmissionResult::Accepted) && config.event_driven_scheduler_enable;
+                            let displaced_terminal = admission_decision
+                                .displaced_request
+                                .as_ref()
+                                .map(|request| record_displaced_terminal(&displaced_terminals, request));
                             if matches!(admission, HealAdmissionResult::Accepted)
                                 && let Some(anchor) = recovery_anchor
                             {
@@ -491,8 +496,16 @@ impl HealManager {
                             }
                             drop(queue);
                             drop(config);
-                            if let Some(displaced_task_id) = admission_decision.displaced_task_id {
-                                remove_task_aliases_for_task(&task_aliases, &displaced_task_id).await;
+                            if let (Some(displaced_task_id), Some(displaced_terminal)) =
+                                (admission_decision.displaced_task_id().map(ToOwned::to_owned), displaced_terminal)
+                            {
+                                remove_displaced_task_aliases(
+                                    &task_aliases,
+                                    &displaced_terminals,
+                                    &displaced_task_id,
+                                    &displaced_terminal,
+                                )
+                                .await;
                                 lock_mrf_repair_notice_targets(&mrf_repair_notice_targets).remove(&displaced_task_id);
                             }
                             if matches!(admission, HealAdmissionResult::Accepted) {

crates/heal/src/heal/manager/scheduler.rs

@@ -21,6 +21,7 @@ impl HealManager {
         let heal_queue = self.heal_queue.clone();
         let active_heals = self.active_heals.clone();
         let completed_heals = self.completed_heals.clone();
+        let displaced_terminals = self.displaced_terminals.clone();
         let task_aliases = self.task_aliases.clone();
         let retrying_heals = self.retrying_heals.clone();
         let mrf_repair_notice_targets = self.mrf_repair_notice_targets.clone();
@@ -53,6 +54,7 @@ impl HealManager {
                             heal_queue: &heal_queue,
                             active_heals: &active_heals,
                             completed_heals: &completed_heals,
+                            displaced_terminals: &displaced_terminals,
                             task_aliases: &task_aliases,
                             retrying_heals: &retrying_heals,
                             mrf_repair_notice_targets: &mrf_repair_notice_targets,
@@ -71,6 +73,7 @@ impl HealManager {
                             heal_queue: &heal_queue,
                             active_heals: &active_heals,
                             completed_heals: &completed_heals,
+                            displaced_terminals: &displaced_terminals,
                             task_aliases: &task_aliases,
                             retrying_heals: &retrying_heals,
                             mrf_repair_notice_targets: &mrf_repair_notice_targets,
@@ -98,6 +101,7 @@ impl HealManager {
             heal_queue,
             active_heals,
             completed_heals,
+            displaced_terminals,
             task_aliases,
             retrying_heals,
             mrf_repair_notice_targets,
@@ -183,6 +187,7 @@ impl HealManager {
                 let active_heals_clone = active_heals.clone();
                 let heal_queue_clone = heal_queue.clone();
                 let completed_heals_clone = completed_heals.clone();
+                let displaced_terminals_clone = displaced_terminals.clone();
                 let task_aliases_clone = task_aliases.clone();
                 let retrying_heals_clone = retrying_heals.clone();
                 let mrf_repair_notice_targets_clone = mrf_repair_notice_targets.clone();
@@ -363,6 +368,7 @@ impl HealManager {
                         let retry_heal_queue = heal_queue_clone.clone();
                         let retrying_heals_for_spawn = retrying_heals_clone.clone();
                         let retry_task_aliases = task_aliases_clone.clone();
+                        let retry_displaced_terminals = displaced_terminals_clone.clone();
                         let retry_mrf_repair_notice_targets = mrf_repair_notice_targets_clone.clone();
                         let retry_completed_heals = completed_heals_clone.clone();
                         let retry_notify = notify_clone.clone();
@@ -430,6 +436,14 @@ impl HealManager {
                                 let admission = admission_decision.result;
                                 let should_notify = matches!(admission, HealAdmissionResult::Accepted)
                                     && retry_config.event_driven_scheduler_enable;
+                                // Publish the terminal synchronously while the
+                                // queue transition is protected. The subsequent
+                                // queue -> retrying handoff retains the lock order
+                                // used by operations_snapshot.
+                                let displaced_terminal = admission_decision
+                                    .displaced_request
+                                    .as_ref()
+                                    .map(|request| record_displaced_terminal(&retry_displaced_terminals, request));
                                 match admission {
                                     HealAdmissionResult::Accepted => {
                                         // Transfer ownership while holding queue -> retrying,
@@ -437,10 +451,18 @@ impl HealManager {
                                         #[cfg(test)]
                                         pause_retry_ownership_transition(&retry_request_id, true).await;
                                         retrying_heals_for_spawn.lock().await.remove(&retry_request_id);
-                                        let displaced_task_id = admission_decision.displaced_task_id;
+                                        let displaced_task_id = admission_decision.displaced_task_id().map(ToOwned::to_owned);
                                         drop(queue);
-                                        if let Some(displaced_task_id) = displaced_task_id {
-                                            remove_task_aliases_for_task(&retry_task_aliases, &displaced_task_id).await;
+                                        if let (Some(displaced_task_id), Some(displaced_terminal)) =
+                                            (displaced_task_id, displaced_terminal)
+                                        {
+                                            remove_displaced_task_aliases(
+                                                &retry_task_aliases,
+                                                &retry_displaced_terminals,
+                                                &displaced_task_id,
+                                                &displaced_terminal,
+                                            )
+                                            .await;
                                             remove_mrf_repair_notice_targets(
                                                 &retry_mrf_repair_notice_targets,
                                                 &displaced_task_id,

crates/heal/src/heal/manager/tests.rs

@@ -84,6 +84,7 @@ async fn process_manager_queue_once(manager: &HealManager) {
         heal_queue: &manager.heal_queue,
         active_heals: &manager.active_heals,
         completed_heals: &manager.completed_heals,
+        displaced_terminals: &manager.displaced_terminals,
         task_aliases: &manager.task_aliases,
         retrying_heals: &manager.retrying_heals,
         mrf_repair_notice_targets: &manager.mrf_repair_notice_targets,
@@ -2778,7 +2779,10 @@ async fn test_high_priority_request_displaces_lower_priority_when_queue_full() {
         HealAdmissionResult::Accepted
     );
     assert_eq!(manager.get_queue_length().await, 1);
-    assert!(matches!(manager.get_task_status(&low_id).await, Err(Error::TaskNotFound { .. })));
+    assert!(matches!(
+        manager.get_task_status(&low_id).await,
+        Ok(HealTaskStatus::Failed { error }) if error.contains("reason=displaced")
+    ));
     assert_eq!(
         manager
             .get_task_status(&high_id)
@@ -2788,6 +2792,263 @@ async fn test_high_priority_request_displaces_lower_priority_when_queue_full() {
     );
 }
 
+#[tokio::test]
+async fn displaced_task_remains_queryable() {
+    let manager = HealManager::new(
+        Arc::new(MockStorage),
+        Some(HealConfig {
+            queue_size: 1,
+            ..HealConfig::default()
+        }),
+    );
+    let mut displaced = HealRequest::new(
+        HealType::Bucket {
+            bucket: "displaced-bucket".to_string(),
+        },
+        HealOptions::default(),
+        HealPriority::Low,
+    );
+    displaced.id = "displaced-task".to_string();
+    let displaced_id = displaced.id.clone();
+    manager
+        .submit_heal_request(displaced)
+        .await
+        .expect("displaced request should queue");
+
+    let successor = HealRequest::new(
+        HealType::Bucket {
+            bucket: "successor-bucket".to_string(),
+        },
+        HealOptions::default(),
+        HealPriority::High,
+    );
+    manager
+        .submit_heal_request(successor)
+        .await
+        .expect("successor should displace low work");
+
+    let report = manager
+        .get_task_report(&displaced_id)
+        .await
+        .expect("displaced report should remain queryable");
+    assert!(matches!(report.status, HealTaskStatus::Failed { ref error } if error.contains("reason=displaced")));
+}
+
+#[tokio::test]
+async fn displaced_archive_failure_keeps_queryable_terminal() {
+    let manager = HealManager::new(Arc::new(MockStorage), None);
+    let mut request = HealRequest::new(
+        HealType::Bucket {
+            bucket: "archive-failure".to_string(),
+        },
+        HealOptions::default(),
+        HealPriority::Low,
+    );
+    request.id = "archive-failure-task".to_string();
+    let request_id = request.id.clone();
+    // The synchronous sidecar is the authoritative fallback when the normal
+    // completed-task archive has no entry (the failure window that must not
+    // turn an Accepted ID into NotFound).
+    record_displaced_terminal(&manager.displaced_terminals, &request);
+    assert!(manager.completed_heals.lock().await.is_empty());
+    assert!(matches!(
+        manager.get_task_status(&request_id).await,
+        Ok(HealTaskStatus::Failed { error }) if error.contains("reason=displaced")
+    ));
+}
+
+#[tokio::test]
+async fn scheduler_retry_displacement_keeps_evicted_task_queryable() {
+    let manager = Arc::new(HealManager::new(
+        Arc::new(MockStorage),
+        Some(HealConfig {
+            queue_size: 1,
+            event_driven_scheduler_enable: false,
+            ..HealConfig::default()
+        }),
+    ));
+    let mut retry_request = HealRequest::object("retry-transition".to_string(), "object".to_string(), None);
+    retry_request.priority = HealPriority::High;
+    let retry_id = retry_request.id.clone();
+    manager
+        .submit_heal_request(retry_request)
+        .await
+        .expect("retry request should queue");
+
+    // Process exactly one queue cycle so the retry task is spawned without a
+    // background scheduler consuming the filler request before the retry wakes.
+    process_manager_queue_once(&manager).await;
+    tokio::time::timeout(Duration::from_secs(1), async {
+        loop {
+            if manager.retrying_heals.lock().await.contains_key(&retry_id) {
+                break;
+            }
+            tokio::task::yield_now().await;
+        }
+    })
+    .await
+    .expect("retry request should enter backoff");
+
+    let filler = HealRequest::new(
+        HealType::Bucket {
+            bucket: "retry-displaced-filler".to_string(),
+        },
+        HealOptions::default(),
+        HealPriority::Low,
+    );
+    let filler_id = filler.id.clone();
+    manager
+        .submit_heal_request(filler)
+        .await
+        .expect("filler request should occupy the queue");
+
+    tokio::time::timeout(Duration::from_secs(5), async {
+        loop {
+            if matches!(
+                manager.get_task_status(&filler_id).await,
+                Ok(HealTaskStatus::Failed { ref error }) if error.contains("reason=displaced")
+            ) {
+                break;
+            }
+            tokio::time::sleep(Duration::from_millis(10)).await;
+        }
+    })
+    .await
+    .expect("retry admission should displace the filler request");
+    assert_eq!(manager.get_queue_length().await, 1);
+    assert_eq!(
+        manager.get_task_status(&retry_id).await.expect("retry should be queued"),
+        HealTaskStatus::Pending
+    );
+}
+
+#[tokio::test]
+async fn concurrent_displacers_produce_one_terminal_generation() {
+    let manager = Arc::new(HealManager::new(
+        Arc::new(MockStorage),
+        Some(HealConfig {
+            queue_size: 1,
+            ..HealConfig::default()
+        }),
+    ));
+    let mut displaced = HealRequest::new(
+        HealType::Bucket {
+            bucket: "concurrent-displaced".to_string(),
+        },
+        HealOptions::default(),
+        HealPriority::Low,
+    );
+    displaced.id = "concurrent-displaced-task".to_string();
+    let displaced_id = displaced.id.clone();
+    manager
+        .submit_heal_request(displaced)
+        .await
+        .expect("initial request should queue");
+
+    let first = HealRequest::new(
+        HealType::Bucket {
+            bucket: "concurrent-successor-a".to_string(),
+        },
+        HealOptions::default(),
+        HealPriority::High,
+    );
+    let second = HealRequest::new(
+        HealType::Bucket {
+            bucket: "concurrent-successor-b".to_string(),
+        },
+        HealOptions::default(),
+        HealPriority::High,
+    );
+    let (first_result, second_result) = tokio::join!(manager.submit_heal_request(first), manager.submit_heal_request(second));
+    let accepted = [&first_result, &second_result]
+        .into_iter()
+        .filter(|result| matches!(result, Ok(HealAdmissionResult::Accepted)))
+        .count();
+    assert_eq!(accepted, 1, "exactly one concurrent displacer should win the full queue");
+    assert!(
+        first_result.is_ok() && second_result.is_ok(),
+        "the losing request should receive a typed Full result"
+    );
+    let terminals = lock_displaced_terminals(&manager.displaced_terminals);
+    assert_eq!(terminals.len(), 1);
+    assert!(terminals.contains_key(&displaced_id));
+}
+
+#[tokio::test]
+async fn successor_chain_is_bounded_and_authorized() {
+    let manager = HealManager::new(
+        Arc::new(MockStorage),
+        Some(HealConfig {
+            queue_size: 1,
+            ..HealConfig::default()
+        }),
+    );
+    let mut original = HealRequest::new(
+        HealType::Bucket {
+            bucket: "authorized-original".to_string(),
+        },
+        HealOptions::default(),
+        HealPriority::Low,
+    );
+    original.id = "authorized-original-task".to_string();
+    let original_id = original.id.clone();
+    manager.submit_heal_request(original).await.expect("original should queue");
+    let mut duplicate = HealRequest::new(
+        HealType::Bucket {
+            bucket: "authorized-original".to_string(),
+        },
+        HealOptions::default(),
+        HealPriority::Low,
+    );
+    duplicate.id = "authorized-duplicate-task".to_string();
+    let duplicate_id = duplicate.id.clone();
+    manager
+        .submit_heal_request(duplicate)
+        .await
+        .expect("same-target duplicate should merge");
+    let successor = HealRequest::new(
+        HealType::Bucket {
+            bucket: "authorized-successor".to_string(),
+        },
+        HealOptions::default(),
+        HealPriority::High,
+    );
+    let successor_id = successor.id.clone();
+    manager.submit_heal_request(successor).await.expect("successor should queue");
+    assert!(manager.task_aliases.lock().await.is_empty());
+    assert!(matches!(manager.get_task_status(&original_id).await, Ok(HealTaskStatus::Failed { .. })));
+    assert!(matches!(manager.get_task_status(&duplicate_id).await, Ok(HealTaskStatus::Failed { .. })));
+    assert_eq!(
+        manager
+            .get_task_status(&successor_id)
+            .await
+            .expect("successor should remain queued"),
+        HealTaskStatus::Pending
+    );
+}
+
+#[tokio::test]
+async fn displaced_terminal_expires_after_bounded_ttl() {
+    let manager = HealManager::new(Arc::new(MockStorage), None);
+    let mut request = HealRequest::new(
+        HealType::Bucket {
+            bucket: "expires".to_string(),
+        },
+        HealOptions::default(),
+        HealPriority::Low,
+    );
+    request.id = "expires-task".to_string();
+    let request_id = request.id.clone();
+    record_displaced_terminal(&manager.displaced_terminals, &request);
+    {
+        let mut terminals = lock_displaced_terminals(&manager.displaced_terminals);
+        let entry =
+            Arc::get_mut(terminals.get_mut(&request_id).expect("terminal should be retained")).expect("test owns terminal entry");
+        entry.completed_at = SystemTime::now() - KEEP_HEAL_TASK_STATUS_DURATION - Duration::from_secs(1);
+    }
+    assert!(matches!(manager.get_task_status(&request_id).await, Err(Error::TaskNotFound { .. })));
+}
+
 #[tokio::test]
 async fn test_displacing_registered_mrf_task_drops_notice_ownership() {
     let storage: Arc<dyn HealStorageAPI> = Arc::new(MockStorage);

crates/heal/src/heal/task/heal_erasure_set.rs

@@ -231,10 +231,6 @@ impl HealTask {
                             "Heal erasure set format repair skipped because no format heal was required"
                         );
                     } else {
-                        let error = e;
-                        if error.is_recoverable_heal() {
-                            return Err(error);
-                        }
                         error!(
                             target: "rustfs::heal::task",
                             event = EVENT_HEAL_ERASURE_SET_RESULT,
@@ -243,7 +239,7 @@ impl HealTask {
                             task_id = %self.id,
                             set_disk_id,
                             result = "format_failed",
-                            error = %error,
+                            error = %e,
                             "Heal erasure set failed"
                         );
                         {
@@ -251,7 +247,7 @@ impl HealTask {
                             progress.update_progress(4, 4, 0, 0);
                         }
                         return Err(Error::TaskExecutionFailed {
-                            message: format!("Failed to heal disk format for {set_disk_id}: {error}"),
+                            message: format!("Failed to heal disk format for {set_disk_id}: {e}"),
                         });
                     }
                 } else {
@@ -288,9 +284,6 @@ impl HealTask {
             Err(Error::TaskCancelled) => return Err(Error::TaskCancelled),
             Err(Error::TaskTimeout) => return Err(Error::TaskTimeout),
             Err(e) => {
-                if e.is_recoverable_heal() {
-                    return Err(e);
-                }
                 error!(
                     target: "rustfs::heal::task",
                     event = EVENT_HEAL_ERASURE_SET_RESULT,

crates/heal/src/heal/task/tests.rs

@@ -547,7 +547,6 @@ struct MockStorage {
     heal_object_outcome: Mutex<Option<MockHealObjectOutcome>>,
     heal_object_outcomes: Mutex<HashMap<String, VecDeque<MockHealObjectOutcome>>>,
     format_no_heal_required: Mutex<bool>,
-    format_error: Mutex<Option<Error>>,
     global_format_calls: Mutex<u32>,
     replacement_format_calls: Mutex<Vec<(usize, usize, Vec<String>)>>,
     replacement_targets_ready: Mutex<bool>,
@@ -868,9 +867,6 @@ impl HealStorageAPI for MockStorage {
 
     async fn heal_format(&self, _dry_run: bool) -> Result<(HealResultItem, Option<Error>)> {
         *self.global_format_calls.lock().unwrap() += 1;
-        if let Some(error) = self.format_error.lock().unwrap().take() {
-            return Err(error);
-        }
         let no_heal_required = *self.format_no_heal_required.lock().unwrap();
         if no_heal_required {
             Ok((HealResultItem::default(), Some(Error::Storage(EcstoreError::NoHealRequired))))
@@ -2056,30 +2052,6 @@ async fn test_erasure_set_heal_continues_after_format_no_heal_required() {
     );
 }
 
-#[tokio::test]
-async fn erasure_set_format_slowdown_is_propagated() {
-    let storage = Arc::new(MockStorage {
-        format_error: Mutex::new(Some(Error::Storage(EcstoreError::SlowDown))),
-        ..Default::default()
-    });
-    let request = HealRequest::new(
-        HealType::ErasureSet {
-            buckets: Vec::new(),
-            set_disk_id: "pool_0_set_0".to_string(),
-        },
-        HealOptions::default(),
-        HealPriority::Normal,
-    );
-    let task = HealTask::from_request(request, storage);
-
-    let error = task
-        .execute()
-        .await
-        .expect_err("format SlowDown must remain recoverable for the task manager");
-
-    assert!(matches!(error, Error::Storage(EcstoreError::SlowDown)));
-}
-
 #[tokio::test]
 async fn erasure_set_bucket_prepass_failure_stops_before_object_heal() {
     let temp = TempDir::new().expect("temporary directory should be created");

crates/protos/src/generated/proto_gen/node_service.rs

@@ -722,6 +722,10 @@ pub struct DeleteVersionsResponse {
     pub errors: ::prost::alloc::vec::Vec<::prost::alloc::string::String>,
     #[prost(message, optional, tag = "3")]
     pub error: ::core::option::Option<Error>,
+    /// Senders dual-write the legacy strings and typed entries. Receivers prefer typed entries
+    /// when present and fall back to strings for peers that predate this field. Code zero means success.
+    #[prost(message, repeated, tag = "4")]
+    pub item_errors: ::prost::alloc::vec::Vec<Error>,
 }
 #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
 pub struct ReadMultipleRequest {

crates/protos/src/node.proto

@@ -493,6 +493,9 @@ message DeleteVersionsResponse {
   bool success = 1;
   repeated string errors = 2;
   optional Error error = 3;
+  // Senders dual-write the legacy strings and typed entries. Receivers prefer typed entries
+  // when present and fall back to strings for peers that predate this field. Code zero means success.
+  repeated Error item_errors = 4;
 }
 
 message ReadMultipleRequest {

crates/test-utils/src/lib.rs

@@ -245,18 +245,6 @@ impl TestECStoreEnvBuilder {
             .await
             .expect("build test ECStore");
 
-        // The production bootstrap only persists pool.bin from the elected
-        // first cluster node.  Test stores intentionally have no cluster
-        // election, but heal-format still requires that durable fence before
-        // it can write any disk format.  Materialize the validated topology
-        // here so the shared fixture models a ready single-node store.
-        let mut pool_meta = ecstore.pool_meta.read().await.clone();
-        pool_meta.dont_save = false;
-        pool_meta
-            .save(ecstore.pools.clone())
-            .await
-            .expect("persist test pool metadata");
-
         if self.init_bucket_metadata {
             let buckets_list = ecstore
                 .list_bucket(&BucketOptions {

rustfs/src/app/bucket_usecase.rs

@@ -513,13 +513,15 @@ fn sr_bucket_meta_item(bucket: String, item_type: &str) -> SRBucketMeta {
     }
 }
 
-fn notify_bucket_metadata_reload(
+async fn notify_bucket_metadata_reload(
     bucket: String,
     operation: &'static str,
     request_context: Option<request_context::RequestContext>,
     scanner_maintenance_change: bool,
 ) {
     record_local_scanner_maintenance_reload(&bucket, scanner_maintenance_change);
+    // Keep reload detached across request cancellation, but wait before a healthy peer can serve the previous config.
+    let (completed_tx, completed_rx) = tokio::sync::oneshot::channel();
     spawn_background_with_context(request_context, async move {
         if let Some(notification_sys) = current_notification_system() {
             let result = if scanner_maintenance_change {
@@ -531,7 +533,9 @@ fn notify_bucket_metadata_reload(
                 warn!(bucket = %bucket, error = %err, "failed to notify peers after {operation}");
             }
         }
+        let _ = completed_tx.send(());
     });
+    let _ = completed_rx.await;
 }
 
 fn record_local_scanner_maintenance_reload(bucket: &str, scanner_maintenance_change: bool) {
@@ -1476,7 +1480,7 @@ impl DefaultBucketUsecase {
             .await
             .map_err(ApiError::from)?;
 
-        notify_bucket_metadata_reload(bucket.clone(), "delete bucket encryption", request_context, false);
+        notify_bucket_metadata_reload(bucket.clone(), "delete bucket encryption", request_context, false).await;
 
         let item = sr_bucket_meta_item(bucket.clone(), "sse-config");
         if let Err(err) = site_replication_bucket_meta_hook(item).await {
@@ -1508,7 +1512,7 @@ impl DefaultBucketUsecase {
             .await
             .map_err(ApiError::from)?;
 
-        notify_bucket_metadata_reload(bucket.clone(), "delete bucket cors", request_context, false);
+        notify_bucket_metadata_reload(bucket.clone(), "delete bucket cors", request_context, false).await;
 
         let item = sr_bucket_meta_item(bucket.clone(), "cors-config");
         if let Err(err) = site_replication_bucket_meta_hook(item).await {
@@ -1540,7 +1544,7 @@ impl DefaultBucketUsecase {
             .await
             .map_err(ApiError::from)?;
 
-        notify_bucket_metadata_reload(bucket.clone(), "delete bucket lifecycle", request_context, true);
+        notify_bucket_metadata_reload(bucket.clone(), "delete bucket lifecycle", request_context, true).await;
 
         let item = sr_bucket_meta_item(bucket.clone(), "lc-config");
         if let Err(err) = site_replication_bucket_meta_hook(item).await {
@@ -1572,7 +1576,7 @@ impl DefaultBucketUsecase {
             .await
             .map_err(ApiError::from)?;
 
-        notify_bucket_metadata_reload(bucket.clone(), "delete bucket policy", request_context, false);
+        notify_bucket_metadata_reload(bucket.clone(), "delete bucket policy", request_context, false).await;
 
         let item = sr_bucket_meta_item(bucket.clone(), "policy");
         if let Err(err) = site_replication_bucket_meta_hook(item).await {
@@ -1630,7 +1634,7 @@ impl DefaultBucketUsecase {
         }
         drop(targets_guard);
 
-        notify_bucket_metadata_reload(bucket.clone(), "delete bucket replication", request_context, true);
+        notify_bucket_metadata_reload(bucket.clone(), "delete bucket replication", request_context, true).await;
 
         let item = sr_bucket_meta_item(bucket.clone(), "replication-config");
         if let Err(err) = site_replication_bucket_meta_hook(item).await {
@@ -1655,7 +1659,7 @@ impl DefaultBucketUsecase {
             .await
             .map_err(ApiError::from)?;
 
-        notify_bucket_metadata_reload(bucket.clone(), "delete bucket tagging", request_context, false);
+        notify_bucket_metadata_reload(bucket.clone(), "delete bucket tagging", request_context, false).await;
 
         let item = sr_bucket_meta_item(bucket.clone(), "tags");
         if let Err(err) = site_replication_bucket_meta_hook(item).await {
@@ -1688,7 +1692,7 @@ impl DefaultBucketUsecase {
             .await
             .map_err(ApiError::from)?;
 
-        notify_bucket_metadata_reload(bucket.clone(), "delete public access block", request_context, false);
+        notify_bucket_metadata_reload(bucket.clone(), "delete public access block", request_context, false).await;
 
         Ok(S3Response::with_status(DeletePublicAccessBlockOutput::default(), StatusCode::NO_CONTENT))
     }
@@ -2143,7 +2147,7 @@ impl DefaultBucketUsecase {
             .await
             .map_err(ApiError::from)?;
 
-        notify_bucket_metadata_reload(bucket.clone(), "put bucket encryption", request_context, false);
+        notify_bucket_metadata_reload(bucket.clone(), "put bucket encryption", request_context, false).await;
 
         let mut item = sr_bucket_meta_item(bucket.clone(), "sse-config");
         item.sse_config = Some(
@@ -2222,7 +2226,7 @@ impl DefaultBucketUsecase {
             .await
             .map_err(ApiError::from)?;
 
-        notify_bucket_metadata_reload(bucket.clone(), "put bucket lifecycle", request_context, true);
+        notify_bucket_metadata_reload(bucket.clone(), "put bucket lifecycle", request_context, true).await;
 
         let mut item = sr_bucket_meta_item(bucket.clone(), "lc-config");
         item.expiry_lc_config =
@@ -2307,7 +2311,7 @@ impl DefaultBucketUsecase {
             .await
             .map_err(ApiError::from)?;
 
-        notify_bucket_metadata_reload(bucket.clone(), "put bucket notification", request_context, false);
+        notify_bucket_metadata_reload(bucket.clone(), "put bucket notification", request_context, false).await;
 
         let region = resolve_notification_region(self.global_region(), request_region);
         let notify = current_notify_interface_for_context(self.context.as_deref());
@@ -2412,7 +2416,7 @@ impl DefaultBucketUsecase {
             .await
             .map_err(ApiError::from)?;
 
-        notify_bucket_metadata_reload(bucket.clone(), "put bucket policy", request_context, false);
+        notify_bucket_metadata_reload(bucket.clone(), "put bucket policy", request_context, false).await;
 
         let mut item = sr_bucket_meta_item(bucket.clone(), "policy");
         item.policy = Some(serde_json::from_str(&policy).map_err(|e| s3_error!(InvalidArgument, "parse policy failed {:?}", e))?);
@@ -2447,7 +2451,7 @@ impl DefaultBucketUsecase {
             .await
             .map_err(ApiError::from)?;
 
-        notify_bucket_metadata_reload(bucket.clone(), "put bucket cors", request_context, false);
+        notify_bucket_metadata_reload(bucket.clone(), "put bucket cors", request_context, false).await;
 
         let mut item = sr_bucket_meta_item(bucket.clone(), "cors-config");
         item.cors =
@@ -2491,7 +2495,7 @@ impl DefaultBucketUsecase {
             .map_err(ApiError::from)?;
         drop(targets_guard);
 
-        notify_bucket_metadata_reload(bucket.clone(), "put bucket replication", request_context, true);
+        notify_bucket_metadata_reload(bucket.clone(), "put bucket replication", request_context, true).await;
 
         let mut item = sr_bucket_meta_item(bucket.clone(), "replication-config");
         item.replication_config = Some(
@@ -2531,7 +2535,7 @@ impl DefaultBucketUsecase {
             .await
             .map_err(ApiError::from)?;
 
-        notify_bucket_metadata_reload(bucket.clone(), "put public access block", request_context, false);
+        notify_bucket_metadata_reload(bucket.clone(), "put public access block", request_context, false).await;
 
         Ok(S3Response::new(PutPublicAccessBlockOutput::default()))
     }
@@ -2560,7 +2564,7 @@ impl DefaultBucketUsecase {
             .await
             .map_err(ApiError::from)?;
 
-        notify_bucket_metadata_reload(bucket.clone(), "put bucket tagging", request_context, false);
+        notify_bucket_metadata_reload(bucket.clone(), "put bucket tagging", request_context, false).await;
 
         let mut item = sr_bucket_meta_item(bucket.clone(), "tags");
         item.tags = Some(serialize_config(&tagging).and_then(|bytes| String::from_utf8(bytes).map_err(to_internal_error))?);
@@ -2593,7 +2597,7 @@ impl DefaultBucketUsecase {
             .await
             .map_err(ApiError::from)?;
 
-        notify_bucket_metadata_reload(bucket.clone(), "put bucket versioning", request_context, false);
+        notify_bucket_metadata_reload(bucket.clone(), "put bucket versioning", request_context, false).await;
 
         let mut item = sr_bucket_meta_item(bucket.clone(), "version-config");
         item.versioning = Some(
@@ -3044,7 +3048,7 @@ mod tests {
                 "{method} should identify the bucket metadata operation in reload logs"
             );
             let expected_reload = format!(
-                "notify_bucket_metadata_reload(bucket.clone(), \"{operation}\", request_context, {scanner_maintenance_change});"
+                "notify_bucket_metadata_reload(bucket.clone(), \"{operation}\", request_context, {scanner_maintenance_change}).await;"
             );
             assert!(
                 body.contains(&expected_reload),

rustfs/src/storage/rpc/node_service/disk.rs

@@ -146,6 +146,29 @@ fn encode_file_info_msgpack(value: &FileInfo) -> std::result::Result<Vec<u8>, Di
     encode_msgpack_with_capacity(value, "FileInfo", FILE_INFO_MSGPACK_ENCODE_CAPACITY_HINT)
 }
 
+fn encode_delete_versions_errors(disk_errors: Vec<Option<DiskError>>) -> (Vec<String>, Vec<Error>) {
+    let mut errors = Vec::with_capacity(disk_errors.len());
+    let mut item_errors = Vec::with_capacity(disk_errors.len());
+    for error in disk_errors {
+        match error {
+            Some(error) => {
+                let code = match &error {
+                    DiskError::Io(source) if source.kind() == std::io::ErrorKind::NotFound => DiskError::FileNotFound.to_u32(),
+                    _ => error.to_u32(),
+                };
+                let error_info = error.to_string();
+                errors.push(error_info.clone());
+                item_errors.push(Error { code, error_info });
+            }
+            None => {
+                errors.push(String::new());
+                item_errors.push(Error::default());
+            }
+        }
+    }
+    (errors, item_errors)
+}
+
 fn encode_msgpack_named<T: serde::Serialize>(value: &T, value_name: &str) -> std::result::Result<Vec<u8>, DiskError> {
     let mut serializer = rmp_serde::Serializer::new(Vec::with_capacity(MSGPACK_ENCODE_CAPACITY_HINT)).with_struct_map();
     value
@@ -552,6 +575,7 @@ impl NodeService {
                             success: false,
                             errors: Vec::new(),
                             error: Some(DiskError::other(format!("decode FileInfoVersions failed: {err}")).into()),
+                            item_errors: Vec::new(),
                         }));
                     }
                 };
@@ -563,30 +587,26 @@ impl NodeService {
                         success: false,
                         errors: Vec::new(),
                         error: Some(DiskError::other(format!("decode DeleteOptions failed: {err}")).into()),
+                        item_errors: Vec::new(),
                     }));
                 }
             };
 
-            let errors = disk
-                .delete_versions(&request.volume, versions, opts)
-                .await
-                .into_iter()
-                .map(|error| match error {
-                    Some(e) => e.to_string(),
-                    None => "".to_string(),
-                })
-                .collect();
+            let (errors, item_errors) =
+                encode_delete_versions_errors(disk.delete_versions(&request.volume, versions, opts).await);
 
             Ok(Response::new(DeleteVersionsResponse {
                 success: true,
                 errors,
                 error: None,
+                item_errors,
             }))
         } else {
             Ok(Response::new(DeleteVersionsResponse {
                 success: false,
                 errors: Vec::new(),
                 error: Some(DiskError::other("cannot find disk".to_string()).into()),
+                item_errors: Vec::new(),
             }))
         }
     }
@@ -1612,8 +1632,8 @@ impl NodeService {
 mod tests {
     use super::{
         compat_response_json, decode_msgpack_or_json, decode_rename_data_request_file_info,
-        encode_batch_read_version_response_payloads, encode_file_info_msgpack, encode_msgpack, encode_msgpack_named,
-        encode_read_multiple_response_payloads, encode_rename_data_response_payloads,
+        encode_batch_read_version_response_payloads, encode_delete_versions_errors, encode_file_info_msgpack, encode_msgpack,
+        encode_msgpack_named, encode_read_multiple_response_payloads, encode_rename_data_response_payloads,
     };
     use crate::storage::rpc::node_service::make_server;
     use crate::storage::storage_api::ReadMultipleResp;
@@ -1632,6 +1652,18 @@ mod tests {
         count: u32,
     }
 
+    #[test]
+    fn delete_versions_response_dual_writes_typed_item_errors() {
+        let raw_not_found = super::DiskError::Io(std::io::Error::from(std::io::ErrorKind::NotFound));
+        let (errors, item_errors) = encode_delete_versions_errors(vec![Some(raw_not_found), None]);
+
+        assert!(errors[0].starts_with("io error "));
+        assert!(errors[1].is_empty());
+        assert_eq!(item_errors[0].code, super::DiskError::FileNotFound.to_u32());
+        assert_eq!(item_errors[0].error_info, errors[0]);
+        assert_eq!(item_errors[1].code, 0);
+    }
+
     #[tokio::test]
     #[serial]
     async fn handle_read_version_records_attribution_for_missing_disk() {