fix(app): wait for peer bucket metadata reload

2026-08-22 12:26:37 +00:00 · 2026-08-22 15:49:37 +08:00
15 changed files with 176 additions and 700 deletions
@@ -585,12 +585,9 @@ impl VersionsHistogram {
    }
 }

-/// Replication statistics for a single target.
-///
-/// Renamed from `ReplicationStats`; serde field names are preserved
-/// byte-identically to maintain wire compatibility with existing snapshots.
-#[derive(Debug, Default, Clone, Serialize, Deserialize, PartialEq, Eq)]
-pub struct ReplicationTargetUsage {
+/// Replication statistics for a single target
+#[derive(Debug, Default, Clone, Serialize, Deserialize)]
+pub struct ReplicationStats {
    pub pending_size: u64,
    pub replicated_size: u64,
    pub failed_size: u64,
@@ -603,7 +600,7 @@ pub struct ReplicationTargetUsage {
    pub replicated_count: u64,
 }

-impl ReplicationTargetUsage {
+impl ReplicationStats {
    pub fn is_empty(&self) -> bool {
        let Self {
            pending_size,
@@ -639,7 +636,7 @@ impl ReplicationTargetUsage {
 /// Replication statistics for all targets
 #[derive(Debug, Default, Clone, Serialize, Deserialize)]
 pub struct ReplicationAllStats {
-    pub targets: HashMap<String, ReplicationTargetUsage>,
+    pub targets: HashMap<String, ReplicationStats>,
    pub replica_size: u64,
    pub replica_count: u64,
 }
@@ -652,7 +649,7 @@ impl ReplicationAllStats {
            targets,
        } = self;

-        *replica_size == 0 && *replica_count == 0 && targets.values().all(ReplicationTargetUsage::is_empty)
+        *replica_size == 0 && *replica_count == 0 && targets.values().all(ReplicationStats::is_empty)
    }

    #[deprecated(note = "use is_empty instead")]
@@ -2469,7 +2466,7 @@ mod tests {

    #[test]
    fn replication_stats_empty_checks_every_field() {
-        type SetField = fn(&mut ReplicationTargetUsage);
+        type SetField = fn(&mut ReplicationStats);

        let cases: [(&str, SetField); 10] = [
            ("pending_size", |stats| stats.pending_size = 1),
@@ -2484,9 +2481,9 @@ mod tests {
            ("replicated_count", |stats| stats.replicated_count = 1),
        ];

-        assert!(ReplicationTargetUsage::default().is_empty());
+        assert!(ReplicationStats::default().is_empty());
        for (field, set_nonzero) in cases {
-            let mut stats = ReplicationTargetUsage::default();
+            let mut stats = ReplicationStats::default();
            set_nonzero(&mut stats);
            assert!(!stats.is_empty(), "{field} must make replication stats non-empty");
        }
@@ -2517,17 +2514,17 @@ mod tests {
        }

        let empty_targets = ReplicationAllStats {
-            targets: HashMap::from([("arn:test:empty".to_string(), ReplicationTargetUsage::default())]),
+            targets: HashMap::from([("arn:test:empty".to_string(), ReplicationStats::default())]),
            ..Default::default()
        };
        assert!(empty_targets.is_empty(), "all-empty targets must keep aggregate stats empty");

        let stats = ReplicationAllStats {
            targets: HashMap::from([
-                ("arn:test:empty".to_string(), ReplicationTargetUsage::default()),
+                ("arn:test:empty".to_string(), ReplicationStats::default()),
                (
                    "arn:test:non-empty".to_string(),
-                    ReplicationTargetUsage {
+                    ReplicationStats {
                        pending_count: 1,
                        ..Default::default()
                    },
@@ -2568,7 +2565,7 @@ mod tests {
                replication_stats: Some(ReplicationAllStats {
                    targets: HashMap::from([(
                        "arn:test:pending".to_string(),
-                        ReplicationTargetUsage {
+                        ReplicationStats {
                            pending_count: 1,
                            ..Default::default()
                        },
@@ -2717,7 +2714,7 @@ mod tests {
                targets: HashMap::from([
                    (
                        "arn:self-only".to_string(),
-                        ReplicationTargetUsage {
+                        ReplicationStats {
                            pending_size: 7,
                            pending_count: 1,
                            ..Default::default()
@@ -2725,7 +2722,7 @@ mod tests {
                    ),
                    (
                        "arn:shared".to_string(),
-                        ReplicationTargetUsage {
+                        ReplicationStats {
                            failed_size: 3,
                            failed_count: 1,
                            missed_threshold_size: 2,
@@ -2744,7 +2741,7 @@ mod tests {
                targets: HashMap::from([
                    (
                        "arn:shared".to_string(),
-                        ReplicationTargetUsage {
+                        ReplicationStats {
                            failed_size: 5,
                            failed_count: 2,
                            after_threshold_size: 4,
@@ -2754,7 +2751,7 @@ mod tests {
                    ),
                    (
                        "arn:other-only".to_string(),
-                        ReplicationTargetUsage {
+                        ReplicationStats {
                            replicated_size: 11,
                            replicated_count: 3,
                            ..Default::default()
@@ -2996,9 +2993,7 @@ mod tests {
    fn replication_target_deserialization_preserves_large_historical_maps() {
        let mut stats = ReplicationAllStats::default();
        for index in 0..=1024 {
-            stats
-                .targets
-                .insert(format!("target-{index}"), ReplicationTargetUsage::default());
+            stats.targets.insert(format!("target-{index}"), ReplicationStats::default());
        }
        let encoded = rmp_serde::to_vec_named(&stats).expect("large replication target fixture should encode");
        let decoded = rmp_serde::from_slice::<ReplicationAllStats>(&encoded)
@@ -3007,47 +3002,6 @@ mod tests {
        assert_eq!(decoded.targets.len(), stats.targets.len());
    }

-    /// Round-trip test: encoding a [`ReplicationTargetUsage`] and decoding it back
-    /// must produce the exact same value.  This guards against accidental serde
-    /// field-name drift during the `ReplicationStats` -> `ReplicationTargetUsage`
-    /// rename.  Wire-level field names are the serialized Rust field identifiers,
-    /// which must remain byte-identical.
-    #[test]
-    fn replication_target_usage_rmp_round_trip() {
-        let original = ReplicationTargetUsage {
-            pending_size: 100,
-            replicated_size: 2_000,
-            failed_size: 50,
-            failed_count: 3,
-            pending_count: 7,
-            missed_threshold_size: 11,
-            after_threshold_size: 22,
-            missed_threshold_count: 1,
-            after_threshold_count: 2,
-            replicated_count: 99,
-        };
-
-        let buf = rmp_serde::to_vec_named(&original).expect("encode ReplicationTargetUsage to msgpack");
-        let decoded: ReplicationTargetUsage = rmp_serde::from_slice(&buf).expect("decode ReplicationTargetUsage from msgpack");
-        assert_eq!(original, decoded, "round-trip through rmp must preserve every field");
-
-        // Also verify that encoding as an unnamed sequence and then decoding
-        // with named fields produces the correct mapping (this catches reordering).
-        let named_buf = rmp_serde::to_vec_named(&original).expect("re-encode for field-name pinning");
-        // Spot-check that known field names appear in the named encoding.
-        let named_str = String::from_utf8_lossy(&named_buf);
-        assert!(named_str.contains("pending_size"), "field 'pending_size' must survive the rename");
-        assert!(named_str.contains("replicated_size"), "field 'replicated_size' must survive the rename");
-        assert!(
-            named_str.contains("missed_threshold_size"),
-            "field 'missed_threshold_size' must survive the rename"
-        );
-        assert!(
-            named_str.contains("after_threshold_count"),
-            "field 'after_threshold_count' must survive the rename"
-        );
-    }
-
    #[test]
    fn checked_merge_rejects_noncanonical_histograms_without_mutation() {
        let mut entry = DataUsageEntry {
@@ -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(())
+}
@@ -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<()> {
@@ -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, DeleteVersionsResponse, DeleteVolumeRequest, DiskInfoRequest, ListDirRequest,
-    ListVolumesRequest, MakeVolumeRequest, MakeVolumesRequest, PreparePartTransactionRequest, ReadAllRequest,
-    ReadMetadataRequest, ReadMultipleRequest, ReadMultipleResponse, ReadPartsRequest, ReadVersionRequest, ReadXlRequest,
-    RenameDataRequest, RenameFileRequest, SettlePartTransactionRequest, SnapshotLeaseReleaseRequest, SnapshotLeaseRenewRequest,
+    DeleteVersionRequest, DeleteVersionsRequest, 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,28 +112,6 @@ 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());
@@ -2428,6 +2406,8 @@ impl DiskAPI for RemoteDisk {
            return errors;
        }

+        // TODO(backlog): replace string errors with typed `StorageError` variants
+
        let result = self
            .execute_with_timeout(
                || async {
@@ -2459,7 +2439,17 @@ impl DiskAPI for RemoteDisk {
            }
            return errors;
        }
-        decode_delete_versions_errors(response, versions.len())
+        response
+            .errors
+            .iter()
+            .map(|error| {
+                if error.is_empty() {
+                    None
+                } else {
+                    Some(Error::other(error.to_string()))
+                }
+            })
+            .collect()
    }

    #[tracing::instrument(level = "trace", skip_all)]
@@ -3770,63 +3760,6 @@ 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(());
@@ -1640,60 +1640,6 @@ 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();
@@ -40,7 +40,6 @@ 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";
@@ -121,30 +120,26 @@ struct MrfRepairNoticeTarget {
    version_id: Option<[u8; 16]>,
 }

-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, PartialEq, Eq)]
 struct HealAdmissionDecision {
    result: HealAdmissionResult,
-    displaced_request: Option<HealRequest>,
+    displaced_task_id: Option<String>,
 }

 impl HealAdmissionDecision {
    const fn new(result: HealAdmissionResult) -> Self {
        Self {
            result,
-            displaced_request: None,
+            displaced_task_id: None,
        }
    }

-    fn accepted_with_displacement(displaced_request: HealRequest) -> Self {
+    fn accepted_with_displacement(displaced_task_id: String) -> Self {
        Self {
            result: HealAdmissionResult::Accepted,
-            displaced_request: Some(displaced_request),
+            displaced_task_id: Some(displaced_task_id),
        }
    }
-
-    fn displaced_task_id(&self) -> Option<&str> {
-        self.displaced_request.as_ref().map(|request| request.id.as_str())
-    }
 }

 fn lock_mrf_repair_notice_targets(
@@ -156,55 +151,6 @@ 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()
@@ -672,14 +618,6 @@ 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.
@@ -721,7 +659,6 @@ 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>>>>,
@@ -937,7 +874,7 @@ impl HealManager {
                        result = "accepted_by_displacement",
                        "Heal queue request accepted by displacement"
                    });
-                    return HealAdmissionDecision::accepted_with_displacement(displaced);
+                    return HealAdmissionDecision::accepted_with_displacement(displaced.id);
                }

                demote_to_debug_when!(per_object_request, warn, target: "rustfs::heal::manager", {
@@ -1168,7 +1105,6 @@ 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())),
@@ -1273,10 +1209,6 @@ 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();
@@ -1527,11 +1459,7 @@ 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().map(ToOwned::to_owned);
-        let displaced_terminal = admission_decision
-            .displaced_request
-            .as_ref()
-            .map(|request| record_displaced_terminal(&self.displaced_terminals, request));
+        let displaced_task_id = admission_decision.displaced_task_id;
        if matches!(admission, HealAdmissionResult::Accepted | HealAdmissionResult::Merged)
            && let Some(target) = mrf_notice_target
        {
@@ -1545,12 +1473,8 @@ impl HealManager {
        drop(queue);
        drop(active_heals);

-        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 let Some(displaced_task_id) = displaced_task_id {
+            self.remove_aliases_for_task(&displaced_task_id).await;
        }

        if should_notify {
@@ -1625,15 +1549,6 @@ 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,
@@ -1754,19 +1669,9 @@ impl HealManager {

        let mut completed_heals = self.completed_heals.lock().await;
        prune_completed_heal_statuses(&mut completed_heals);
-        if completed_heals
+        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
@@ -21,7 +21,6 @@ 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();
@@ -482,10 +481,6 @@ 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
                            {
@@ -496,16 +491,8 @@ impl HealManager {
                            }
                            drop(queue);
                            drop(config);
-                            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;
+                            if let Some(displaced_task_id) = admission_decision.displaced_task_id {
+                                remove_task_aliases_for_task(&task_aliases, &displaced_task_id).await;
                                lock_mrf_repair_notice_targets(&mrf_repair_notice_targets).remove(&displaced_task_id);
                            }
                            if matches!(admission, HealAdmissionResult::Accepted) {
@@ -21,7 +21,6 @@ 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();
@@ -54,7 +53,6 @@ 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,
@@ -73,7 +71,6 @@ 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,
@@ -101,7 +98,6 @@ impl HealManager {
            heal_queue,
            active_heals,
            completed_heals,
-            displaced_terminals,
            task_aliases,
            retrying_heals,
            mrf_repair_notice_targets,
@@ -187,7 +183,6 @@ 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();
@@ -368,7 +363,6 @@ 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();
@@ -436,14 +430,6 @@ 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,
@@ -451,18 +437,10 @@ 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().map(ToOwned::to_owned);
+                                        let displaced_task_id = admission_decision.displaced_task_id;
                                        drop(queue);
-                                        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;
+                                        if let Some(displaced_task_id) = displaced_task_id {
+                                            remove_task_aliases_for_task(&retry_task_aliases, &displaced_task_id).await;
                                            remove_mrf_repair_notice_targets(
                                                &retry_mrf_repair_notice_targets,
                                                &displaced_task_id,
@@ -84,7 +84,6 @@ 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,
@@ -2779,10 +2778,7 @@ 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,
-        Ok(HealTaskStatus::Failed { error }) if error.contains("reason=displaced")
-    ));
+    assert!(matches!(manager.get_task_status(&low_id).await, Err(Error::TaskNotFound { .. })));
    assert_eq!(
        manager
            .get_task_status(&high_id)
@@ -2792,263 +2788,6 @@ 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);
@@ -722,10 +722,6 @@ 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 {
@@ -493,9 +493,6 @@ 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 {
@@ -16,7 +16,7 @@ use super::persistence::DataUsageCacheLoadAttempt;
 use super::*;
 use crate::storage_api::scanner_io::{HTTPRangeSpec, ObjectIO};
 use crate::{ScannerGetObjectReader, ScannerPutObjReader};
-use rustfs_data_usage::{ReplicationAllStats, ReplicationTargetUsage};
+use rustfs_data_usage::{ReplicationAllStats, ReplicationStats};
 use serde_json::Value;
 use std::io::Cursor;
 use std::pin::Pin;
@@ -1636,7 +1636,7 @@ fn size_recursive_prunes_empty_and_preserves_threshold_replication_stats() {
            replication_stats: Some(ReplicationAllStats {
                targets: HashMap::from([(
                    "arn:test:threshold".to_string(),
-                    ReplicationTargetUsage {
+                    ReplicationStats {
                        after_threshold_count: 1,
                        ..Default::default()
                    },
@@ -13,7 +13,7 @@
 // limitations under the License.

 use super::*;
-use rustfs_data_usage::{ReplicationAllStats, ReplicationTargetUsage};
+use rustfs_data_usage::{ReplicationAllStats, ReplicationStats};

 const TEST_PLAN_DIGEST: DataUsageScanPlanDigest = DataUsageScanPlanDigest([7; 32]);

@@ -271,7 +271,7 @@ fn completed_data_usage_info_flattens_nested_bucket_entries() {
        replication_stats: Some(ReplicationAllStats {
            targets: HashMap::from([(
                "arn:target".to_string(),
-                ReplicationTargetUsage {
+                ReplicationStats {
                    replicated_size: 2048,
                    replicated_count: 2,
                    ..Default::default()
@@ -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),
@@ -146,29 +146,6 @@ 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
@@ -575,7 +552,6 @@ impl NodeService {
                            success: false,
                            errors: Vec::new(),
                            error: Some(DiskError::other(format!("decode FileInfoVersions failed: {err}")).into()),
-                            item_errors: Vec::new(),
                        }));
                    }
                };
@@ -587,26 +563,30 @@ impl NodeService {
                        success: false,
                        errors: Vec::new(),
                        error: Some(DiskError::other(format!("decode DeleteOptions failed: {err}")).into()),
-                        item_errors: Vec::new(),
                    }));
                }
            };

-            let (errors, item_errors) =
-                encode_delete_versions_errors(disk.delete_versions(&request.volume, versions, opts).await);
+            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();

            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(),
            }))
        }
    }
@@ -1632,8 +1612,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_delete_versions_errors, 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_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;
@@ -1652,18 +1632,6 @@ 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() {