test(heal): cover suspended owner production path

.config/nextest.toml

@@ -78,12 +78,6 @@ test-group = 'embedded-test-ports'
 filter = 'package(rustfs-ecstore) & test(manual_transition_page_checkpoint_persists_durable_job_progress)'
 test-group = 'ecstore-serial-flaky'
 
-# The durable ILM decommission regressions build isolated multi-pool stores and
-# deliberately take source or target disks offline while checking fencing.
-[[profile.default.overrides]]
-filter = 'package(rustfs-ecstore) & (test(decommission_migrates_and_verifies_registered_durable_ilm_records) | test(decommission_durable_ilm_target_read_error_is_not_masked_by_peer_success) | test(decommission_durable_ilm_terminal_receipt_recovers_failed_source_cleanup) | test(decommission_durable_ilm_receipt_pagination_fails_closed_on_second_page) | test(decommission_durable_ilm_recovery_keeps_multiple_active_sources))'
-test-group = 'ecstore-serial-flaky'
-
 # Serialize the bucket-incarnation / lifecycle-fence tests. They drive
 # init_bucket_metadata_sys and bucket_metadata_sys_of, i.e. process-global
 # OnceLock state that serial_test's #[serial] cannot protect across nextest's
@@ -196,10 +190,6 @@ test-group = 'embedded-test-ports'
 filter = 'package(rustfs-ecstore) & test(manual_transition_page_checkpoint_persists_durable_job_progress)'
 test-group = 'ecstore-serial-flaky'
 
-[[profile.ci.overrides]]
-filter = 'package(rustfs-ecstore) & (test(decommission_migrates_and_verifies_registered_durable_ilm_records) | test(decommission_durable_ilm_target_read_error_is_not_masked_by_peer_success) | test(decommission_durable_ilm_terminal_receipt_recovers_failed_source_cleanup) | test(decommission_durable_ilm_receipt_pagination_fails_closed_on_second_page) | test(decommission_durable_ilm_recovery_keeps_multiple_active_sources))'
-test-group = 'ecstore-serial-flaky'
-
 # Serialize the bucket-incarnation / lifecycle-fence tests under the ci profile
 # too (see the matching default-profile override near the top). No retries.
 [[profile.ci.overrides]]

crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs

@@ -9136,7 +9136,6 @@ mod tests {
         assert_eq!(loaded.report.scanned, 37);
         assert_eq!(loaded.report.eligible, 11);
         assert_eq!(loaded.report.enqueued, 5);
-        assert_eq!(loaded.cursor_revision, Some(1));
         assert!(loaded.lease_expires_at_unix_nanos > 0);
         let token = loaded
             .report
@@ -9155,30 +9154,6 @@ mod tests {
         assert_eq!(admission.lease_id, loaded.lease_id);
         assert_eq!(admission.lease_expires_at_unix_nanos, loaded.lease_expires_at_unix_nanos);
 
-        let mut same_marker_report = report.clone();
-        same_marker_report.scanned += 1;
-        persist_manual_transition_page_checkpoint(
-            &checkpoint_options,
-            &same_marker_report,
-            Some("logs/page-end".to_string()),
-            Some("opaque-next-version".to_string()),
-        )
-        .await
-        .expect("same-marker version checkpoint should persist through the durable progress sink");
-        let same_marker_checkpointed = load_manual_transition_job_record(ecstore.clone(), job_id)
-            .await
-            .expect("same-marker version checkpoint should reload");
-        assert_eq!(same_marker_checkpointed.cursor_revision, Some(2));
-        let (_, version_marker) = decode_manual_transition_continuation_token(
-            same_marker_checkpointed
-                .report
-                .continuation_token
-                .as_deref()
-                .expect("same-marker version checkpoint should persist a cursor"),
-        )
-        .expect("same-marker version cursor should decode");
-        assert_eq!(version_marker.as_deref(), Some("opaque-next-version"));
-
         create_test_bucket(&ecstore, &bucket).await;
         let lifecycle_xml = format!(
             r#"<?xml version="1.0" encoding="UTF-8"?>
@@ -9235,7 +9210,6 @@ mod tests {
         assert_eq!(checkpointed.report.scanned, 1000);
         assert_eq!(checkpointed.report.eligible, 1000);
         assert_eq!(checkpointed.report.dry_run_eligible, 1000);
-        assert_eq!(checkpointed.cursor_revision, Some(3));
         let token = checkpointed
             .report
             .continuation_token

crates/ecstore/src/bucket/lifecycle/manual_transition_job.rs

@@ -24,10 +24,6 @@ use crate::bucket::lifecycle::bucket_lifecycle_ops::{
     ManualTransitionQueueSnapshot, ManualTransitionRunOptions, ManualTransitionRunReport,
 };
 use crate::bucket::lifecycle::config_boundary;
-use crate::bucket::lifecycle::durable_namespace::{
-    MANUAL_TRANSITION_JOB_NAMESPACE, MANUAL_TRANSITION_SCOPE_NAMESPACE, MANUAL_TRANSITION_TASK_NAMESPACE,
-    MANUAL_TRANSITION_WORKER_RESULT_NAMESPACE,
-};
 use crate::disk::RUSTFS_META_BUCKET;
 use crate::error::{Error, Result as EcstoreResult};
 use crate::object_api::ObjectOptions;
@@ -38,10 +34,10 @@ use crate::store::ECStore;
 pub const MANUAL_TRANSITION_JOB_SCHEMA: &str = "rustfs-manual-transition-job-v1";
 pub const MANUAL_TRANSITION_TASK_SCHEMA: &str = "rustfs-manual-transition-task-v1";
 pub const MANUAL_TRANSITION_WORKER_RESULT_SCHEMA: &str = "rustfs-manual-transition-worker-result-v1";
-pub const MANUAL_TRANSITION_JOB_RECORD_PREFIX: &str = MANUAL_TRANSITION_JOB_NAMESPACE.prefix;
-pub const MANUAL_TRANSITION_SCOPE_RECORD_PREFIX: &str = MANUAL_TRANSITION_SCOPE_NAMESPACE.prefix;
-pub const MANUAL_TRANSITION_TASK_PREFIX: &str = MANUAL_TRANSITION_TASK_NAMESPACE.prefix;
-pub const MANUAL_TRANSITION_WORKER_RESULT_PREFIX: &str = MANUAL_TRANSITION_WORKER_RESULT_NAMESPACE.prefix;
+pub const MANUAL_TRANSITION_JOB_RECORD_PREFIX: &str = "ilm/manual-transition/jobs";
+pub const MANUAL_TRANSITION_SCOPE_RECORD_PREFIX: &str = "ilm/manual-transition/scopes";
+pub const MANUAL_TRANSITION_TASK_PREFIX: &str = "ilm/manual-transition/tasks";
+pub const MANUAL_TRANSITION_WORKER_RESULT_PREFIX: &str = "ilm/manual-transition/results";
 pub const MAX_MANUAL_TRANSITION_JOB_RECORD_SIZE: usize = 64 * 1024;
 pub const MAX_MANUAL_TRANSITION_TASK_RECORD_SIZE: usize = 16 * 1024;
 pub const MAX_MANUAL_TRANSITION_WORKER_RESULT_RECORD_SIZE: usize = 8 * 1024;
@@ -199,8 +195,6 @@ pub struct ManualTransitionJobRecord {
     pub updated_at_unix_nanos: i128,
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub completed_at_unix_nanos: Option<i128>,
-    #[serde(default, skip_serializing_if = "Option::is_none")]
-    pub cursor_revision: Option<u64>,
     pub report: ManualTransitionRunReport,
     pub queue_snapshot: ManualTransitionQueueSnapshot,
     #[serde(default, skip_serializing_if = "Option::is_none")]
@@ -230,7 +224,6 @@ impl ManualTransitionJobRecord {
             created_at_unix_nanos: now,
             updated_at_unix_nanos: now,
             completed_at_unix_nanos: None,
-            cursor_revision: Some(0),
             report: ManualTransitionRunReport {
                 bucket: bucket.to_string(),
                 prefix: options.prefix.clone(),
@@ -245,7 +238,7 @@ impl ManualTransitionJobRecord {
 
     pub fn complete(&mut self, report: ManualTransitionRunReport, queue_snapshot: ManualTransitionQueueSnapshot) {
         self.scan_completed = true;
-        self.merge_scan_report(&report);
+        self.report.merge_scan_report_preserving_worker(&report);
         self.queue_snapshot = queue_snapshot;
         self.error = None;
         self.mark_terminal_if_worker_drained();
@@ -323,7 +316,7 @@ impl ManualTransitionJobRecord {
             }
         }
         self.queue_snapshot = queue_snapshot;
-        self.advance_updated_at();
+        self.updated_at_unix_nanos = OffsetDateTime::now_utc().unix_timestamp_nanos();
         self.mark_terminal_if_worker_drained();
     }
 
@@ -366,14 +359,14 @@ impl ManualTransitionJobRecord {
         self.report.tier_failure = scan_tier_failure.saturating_add(transition_failed);
         self.report.tier_failure_by_reason = scan_tier_failure_by_reason;
         self.queue_snapshot = queue_snapshot;
-        self.advance_updated_at();
+        self.updated_at_unix_nanos = OffsetDateTime::now_utc().unix_timestamp_nanos();
         self.mark_terminal_if_worker_drained();
         true
     }
 
     pub fn mark_cancel_requested(&mut self) {
         self.cancel_requested = true;
-        self.advance_updated_at();
+        self.updated_at_unix_nanos = OffsetDateTime::now_utc().unix_timestamp_nanos();
     }
 
     pub fn claim_recovery_lease(&mut self, owner_id: impl Into<String>, queue_snapshot: ManualTransitionQueueSnapshot) {
@@ -387,7 +380,7 @@ impl ManualTransitionJobRecord {
     pub fn abandon_recovery_lease(&mut self, lease_id: Uuid) {
         if self.state == ManualTransitionJobState::Running && self.lease_id == lease_id {
             self.lease_expires_at_unix_nanos = 0;
-            self.advance_updated_at();
+            self.updated_at_unix_nanos = OffsetDateTime::now_utc().unix_timestamp_nanos();
         }
     }
 
@@ -405,7 +398,7 @@ impl ManualTransitionJobRecord {
 
     pub fn renew_lease(&mut self, queue_snapshot: ManualTransitionQueueSnapshot) {
         let now = OffsetDateTime::now_utc().unix_timestamp_nanos();
-        self.updated_at_unix_nanos = self.updated_at_unix_nanos.saturating_add(1).max(now);
+        self.updated_at_unix_nanos = now;
         self.lease_expires_at_unix_nanos = manual_transition_job_lease_expires_at(now);
         self.queue_snapshot = queue_snapshot;
     }
@@ -445,18 +438,11 @@ impl ManualTransitionJobRecord {
 
     pub fn update_running_progress(&mut self, report: ManualTransitionRunReport, queue_snapshot: ManualTransitionQueueSnapshot) {
         if self.state == ManualTransitionJobState::Running {
-            self.merge_scan_report(&report);
+            self.report.merge_scan_report_preserving_worker(&report);
             self.renew_lease(queue_snapshot);
         }
     }
 
-    fn merge_scan_report(&mut self, report: &ManualTransitionRunReport) {
-        if self.report.continuation_token != report.continuation_token {
-            self.cursor_revision = Some(self.cursor_revision.unwrap_or(0).saturating_add(1));
-        }
-        self.report.merge_scan_report_preserving_worker(report);
-    }
-
     pub fn mark_unknown_if_unowned(&mut self) {
         if self.state == ManualTransitionJobState::Running {
             self.state = ManualTransitionJobState::Unknown;
@@ -477,13 +463,9 @@ impl ManualTransitionJobRecord {
     }
 
     fn mark_updated_terminal(&mut self) {
-        self.advance_updated_at();
-        self.completed_at_unix_nanos = Some(self.updated_at_unix_nanos);
-    }
-
-    fn advance_updated_at(&mut self) {
         let now = OffsetDateTime::now_utc().unix_timestamp_nanos();
-        self.updated_at_unix_nanos = self.updated_at_unix_nanos.saturating_add(1).max(now);
+        self.updated_at_unix_nanos = now;
+        self.completed_at_unix_nanos = Some(now);
     }
 
     fn mark_terminal_if_worker_drained(&mut self) {
@@ -1127,8 +1109,7 @@ pub fn manual_transition_scope_record_object_name(scope_key: &str) -> Result<Str
 pub async fn save_manual_transition_job_record(api: Arc<ECStore>, job: &ManualTransitionJobRecord) -> EcstoreResult<()> {
     let object = manual_transition_job_record_object_name(job.job_id).map_err(manual_transition_job_store_error)?;
     let data = job.encode().map_err(manual_transition_job_store_error)?;
-    config_boundary::save_config(api.clone(), &object, data.clone()).await?;
-    api.record_durable_ilm_decommission_progress(&object, &data).await
+    config_boundary::save_config(api, &object, data).await
 }
 
 pub async fn load_manual_transition_job_record(api: Arc<ECStore>, job_id: Uuid) -> EcstoreResult<ManualTransitionJobRecord> {
@@ -1161,9 +1142,9 @@ pub async fn save_manual_transition_job_record_if_current(
     let object = manual_transition_job_record_object_name(job.job_id).map_err(manual_transition_job_store_error)?;
     let data = job.encode().map_err(manual_transition_job_store_error)?;
     config_boundary::save_config_with_opts_quiet(
-        api.clone(),
+        api,
         &object,
-        data.clone(),
+        data,
         &ObjectOptions {
             max_parity: true,
             http_preconditions: Some(HTTPPreconditions {
@@ -1173,8 +1154,7 @@ pub async fn save_manual_transition_job_record_if_current(
             ..Default::default()
         },
     )
-    .await?;
-    api.record_durable_ilm_decommission_progress(&object, &data).await
+    .await
 }
 
 /// Applies a job-record mutation with optimistic concurrency control.
@@ -1612,9 +1592,9 @@ pub async fn save_manual_transition_scope_admission_if_absent(
     let object = manual_transition_scope_record_object_name(&admission.scope_key).map_err(manual_transition_job_store_error)?;
     let data = serde_json::to_vec(admission).map_err(Error::other)?;
     config_boundary::save_config_with_opts(
-        api.clone(),
+        api,
         &object,
-        data.clone(),
+        data,
         &ObjectOptions {
             max_parity: true,
             http_preconditions: Some(HTTPPreconditions {
@@ -1624,8 +1604,7 @@ pub async fn save_manual_transition_scope_admission_if_absent(
             ..Default::default()
         },
     )
-    .await?;
-    api.record_durable_ilm_decommission_progress(&object, &data).await
+    .await
 }
 
 pub async fn load_manual_transition_scope_admission(
@@ -1663,9 +1642,9 @@ pub async fn save_manual_transition_scope_admission_if_current(
     let object = manual_transition_scope_record_object_name(&admission.scope_key).map_err(manual_transition_job_store_error)?;
     let data = serde_json::to_vec(admission).map_err(Error::other)?;
     match config_boundary::save_config_with_opts(
-        api.clone(),
+        api,
         &object,
-        data.clone(),
+        data,
         &ObjectOptions {
             max_parity: true,
             http_preconditions: Some(HTTPPreconditions {
@@ -1681,8 +1660,7 @@ pub async fn save_manual_transition_scope_admission_if_current(
             Err(Error::PreconditionFailed)
         }
         result => result,
-    }?;
-    api.record_durable_ilm_decommission_progress(&object, &data).await
+    }
 }
 
 pub async fn claim_manual_transition_scope_admission(
@@ -1975,15 +1953,13 @@ pub async fn delete_manual_transition_scope_admission_if_current(
     job_id: Uuid,
     lease_id: Uuid,
 ) -> EcstoreResult<bool> {
-    let (admission, etag) = match load_manual_transition_scope_admission_with_etag(api.clone(), scope_key).await {
-        Ok((admission, etag)) if admission.job_id == job_id && admission.lease_id == lease_id => (admission, etag),
+    let etag = match load_manual_transition_scope_admission_with_etag(api.clone(), scope_key).await {
+        Ok((admission, etag)) if admission.job_id == job_id && admission.lease_id == lease_id => etag,
         Ok(_) => return Ok(false),
         Err(Error::ConfigNotFound) => return Ok(true),
         Err(err) => return Err(err),
     };
     let object = manual_transition_scope_record_object_name(scope_key).map_err(manual_transition_job_store_error)?;
-    let data = serde_json::to_vec(&admission).map_err(Error::other)?;
-    api.record_durable_ilm_decommission_terminal(&object, &data).await?;
     match config_boundary::delete_config_if_match(api, &object, &etag).await {
         Ok(()) | Err(Error::ConfigNotFound) => Ok(true),
         Err(Error::PreconditionFailed) => Ok(false),
@@ -2601,25 +2577,6 @@ mod tests {
         assert!(decoded.report.tier_failure_by_reason.is_empty());
     }
 
-    #[test]
-    fn manual_transition_job_record_decodes_legacy_cursor_without_revision() {
-        let options = ManualTransitionRunOptions::default();
-        let record = ManualTransitionJobRecord::new(Uuid::new_v4(), "bucket", &options, TEST_OWNER);
-        let encoded = record.encode().expect("job record should encode");
-        let mut value: serde_json::Value = serde_json::from_slice(&encoded).expect("encoded job should be json");
-        value["job"]
-            .as_object_mut()
-            .expect("job should be object")
-            .remove("cursor_revision");
-        let record_bytes = serde_json::to_vec(&value["job"]).expect("legacy job should encode");
-        value["content_sha256"] = serde_json::Value::String(hex_sha256(&record_bytes, ToOwned::to_owned));
-        let legacy = serde_json::to_vec(&value).expect("legacy envelope should encode");
-
-        let decoded = ManualTransitionJobRecord::decode(record.job_id, &legacy).expect("legacy job should decode");
-
-        assert_eq!(decoded.cursor_revision, None);
-    }
-
     #[test]
     fn manual_transition_job_record_rejects_unknown_report_fields() {
         let options = ManualTransitionRunOptions::default();

crates/ecstore/src/bucket/lifecycle/mod.rs

@@ -16,7 +16,6 @@ pub mod bucket_lifecycle_audit;
 pub mod bucket_lifecycle_ops;
 mod config_boundary;
 pub mod core;
-mod durable_namespace;
 pub mod evaluator;
 pub mod manual_transition_job;
 mod metadata_boundary;
@@ -32,8 +31,3 @@ pub mod tier_free_version_recovery;
 pub mod tier_last_day_stats;
 pub mod tier_sweeper;
 pub mod transition_transaction;
-
-pub(crate) use durable_namespace::{
-    DurableIlmRecordCheckpoint, ILM_META_PREFIX, ValidatedDurableIlmRecord, classify_durable_ilm_record,
-    validate_durable_ilm_record,
-};

crates/ecstore/src/bucket/lifecycle/tier_delete_journal.rs

@@ -20,7 +20,6 @@ use tokio_util::sync::CancellationToken;
 use tracing::{debug, warn};
 
 use crate::bucket::lifecycle::config_boundary;
-use crate::bucket::lifecycle::durable_namespace::TIER_DELETE_JOURNAL_NAMESPACE;
 use crate::bucket::lifecycle::runtime_boundary;
 use crate::bucket::lifecycle::tier_sweeper::{
     Jentry, TierDeleteJournalState, TierDeleteSourceIdentity,
@@ -50,7 +49,7 @@ const TIER_DELETE_JOURNAL_VERSION: u8 = 2;
 const TIER_DELETE_JOURNAL_EXACT_VERSION: u8 = 3;
 const TIER_DELETE_JOURNAL_STATE_VERSION: u8 = 4;
 const TIER_DELETE_JOURNAL_TRANSACTION_VERSION: u8 = 5;
-pub(crate) const TIER_DELETE_JOURNAL_PREFIX: &str = TIER_DELETE_JOURNAL_NAMESPACE.prefix;
+pub(crate) const TIER_DELETE_JOURNAL_PREFIX: &str = "ilm/tier-delete-journal/";
 
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
 #[serde(deny_unknown_fields)]
@@ -433,21 +432,6 @@ async fn process_committed_tier_delete_journal_entry(api: Arc<ECStore>, je: &Jen
         )
         .await?;
     }
-    let path = tier_delete_journal_object_name(je);
-    let data = encode_tier_delete_journal_entry(je).map_err(std::io::Error::other)?;
-    let target_pool_indices = api
-        .record_durable_ilm_decommission_terminal_target_pools(&path, &data)
-        .await
-        .map_err(std::io::Error::other)?;
-    if let Some(target_pool_indices) = target_pool_indices {
-        for target_pool_idx in target_pool_indices {
-            match config_boundary::delete_config(api.pools[target_pool_idx].clone(), &path).await {
-                Ok(()) | Err(Error::ConfigNotFound) => {}
-                Err(err) => return Err(std::io::Error::other(err)),
-            }
-        }
-        return Ok(());
-    }
     remove_tier_delete_journal_entry(api, je).await
 }
 

crates/ecstore/src/bucket/lifecycle/transition_transaction.rs

@@ -21,7 +21,6 @@ use tracing::{debug, warn};
 use uuid::Uuid;
 
 use crate::bucket::lifecycle::config_boundary;
-use crate::bucket::lifecycle::durable_namespace::TRANSITION_TRANSACTION_NAMESPACE;
 use crate::bucket::lifecycle::lifecycle::TRANSITION_COMPLETE;
 use crate::bucket::lifecycle::tier_sweeper::{
     delete_confirmed_transition_candidate_exact_with_lease_idempotent,
@@ -43,7 +42,7 @@ const TRANSITION_TRANSACTION_RECOVERY_INTERVAL: Duration = Duration::from_secs(6
 const TRANSITION_TRANSACTION_RECOVERY_TIMEOUT: Duration = Duration::from_secs(300);
 pub const TRANSITION_TRANSACTION_SCHEMA: &str = "rustfs-transition-transaction-v1";
 pub const TRANSITION_TRANSACTION_PREFIX: &str = "ilm/transition-transactions";
-pub const TRANSITION_TRANSACTION_RECORD_PREFIX: &str = TRANSITION_TRANSACTION_NAMESPACE.prefix;
+pub const TRANSITION_TRANSACTION_RECORD_PREFIX: &str = "ilm/transition-transactions/records";
 pub const MAX_TRANSITION_TRANSACTION_SIZE: usize = 64 * 1024;
 
 pub type Result<T> = std::result::Result<T, TransitionTransactionError>;
@@ -585,8 +584,7 @@ pub(crate) async fn save_transition_transaction_record(
     let object =
         transition_transaction_record_object_name(transaction.transaction_id).map_err(transition_transaction_store_error)?;
     let data = transaction.encode().map_err(transition_transaction_store_error)?;
-    config_boundary::save_config(api.clone(), &object, data.clone()).await?;
-    api.record_durable_ilm_decommission_progress(&object, &data).await
+    config_boundary::save_config(api, &object, data).await
 }
 
 pub(crate) async fn load_transition_transaction_record(
@@ -598,14 +596,8 @@ pub(crate) async fn load_transition_transaction_record(
     TransitionTransaction::decode(transaction_id, &data).map_err(transition_transaction_store_error)
 }
 
-pub(crate) async fn delete_transition_transaction_record(
-    api: Arc<ECStore>,
-    transaction: &TransitionTransaction,
-) -> EcstoreResult<()> {
-    let object =
-        transition_transaction_record_object_name(transaction.transaction_id).map_err(transition_transaction_store_error)?;
-    let data = transaction.encode().map_err(transition_transaction_store_error)?;
-    api.record_durable_ilm_decommission_terminal(&object, &data).await?;
+pub(crate) async fn delete_transition_transaction_record(api: Arc<ECStore>, transaction_id: Uuid) -> EcstoreResult<()> {
+    let object = transition_transaction_record_object_name(transaction_id).map_err(transition_transaction_store_error)?;
     match config_boundary::delete_config(api, &object).await {
         Ok(()) | Err(Error::ConfigNotFound) => Ok(()),
         Err(err) => Err(err),
@@ -821,7 +813,7 @@ pub async fn finalize_missing_transition_transaction_for_operator(
     if probe != TransitionOperatorProbe::Missing {
         return Err(TransitionOperatorError::CandidateNotMissing(probe));
     }
-    delete_transition_transaction_record(api, &transaction)
+    delete_transition_transaction_record(api, transaction_id)
         .await
         .map_err(TransitionOperatorError::Store)
 }
@@ -857,22 +849,22 @@ pub async fn process_transition_transaction_record(
     match transaction.state {
         TransitionTransactionState::Uploaded => {
             delete_transition_remote_candidate(api.clone(), transaction).await?;
-            delete_transition_transaction_record(api, transaction).await?;
+            delete_transition_transaction_record(api, transaction.transaction_id).await?;
             Ok(TransitionTransactionRecoveryOutcome::RemoteCandidateDeleted)
         }
         TransitionTransactionState::CleanupPending => match local_commit_matches_transaction(api.clone(), transaction).await {
             Ok(true) => {
-                delete_transition_transaction_record(api, transaction).await?;
+                delete_transition_transaction_record(api, transaction.transaction_id).await?;
                 Ok(TransitionTransactionRecoveryOutcome::RecordDeleted)
             }
             Ok(false) => {
                 delete_transition_remote_candidate(api.clone(), transaction).await?;
-                delete_transition_transaction_record(api, transaction).await?;
+                delete_transition_transaction_record(api, transaction.transaction_id).await?;
                 Ok(TransitionTransactionRecoveryOutcome::RemoteCandidateDeleted)
             }
             Err(err) if transition_source_is_missing(&err) => {
                 delete_transition_remote_candidate(api.clone(), transaction).await?;
-                delete_transition_transaction_record(api, transaction).await?;
+                delete_transition_transaction_record(api, transaction.transaction_id).await?;
                 Ok(TransitionTransactionRecoveryOutcome::RemoteCandidateDeleted)
             }
             Err(err) => Err(err),
@@ -880,7 +872,7 @@ pub async fn process_transition_transaction_record(
         TransitionTransactionState::LocalCommitStarted => {
             match local_commit_matches_transaction(api.clone(), transaction).await {
                 Ok(true) => {
-                    delete_transition_transaction_record(api, transaction).await?;
+                    delete_transition_transaction_record(api, transaction.transaction_id).await?;
                     Ok(TransitionTransactionRecoveryOutcome::RecordDeleted)
                 }
                 Ok(false) => Ok(TransitionTransactionRecoveryOutcome::Retained),
@@ -889,7 +881,7 @@ pub async fn process_transition_transaction_record(
             }
         }
         TransitionTransactionState::AbortedNoRemote | TransitionTransactionState::Committed => {
-            delete_transition_transaction_record(api, transaction).await?;
+            delete_transition_transaction_record(api, transaction.transaction_id).await?;
             Ok(TransitionTransactionRecoveryOutcome::RecordDeleted)
         }
         TransitionTransactionState::UploadOutcomeUnknown => recover_unknown_upload_outcome(api, transaction).await,
@@ -915,7 +907,7 @@ async fn recover_unknown_upload_outcome(
         .map_err(Error::other)?
     {
         TransitionCandidateProbe::Missing => {
-            delete_transition_transaction_record(api, transaction).await?;
+            delete_transition_transaction_record(api, transaction.transaction_id).await?;
             Ok(TransitionTransactionRecoveryOutcome::RecordDeleted)
         }
         TransitionCandidateProbe::UnversionedPresent => {
@@ -933,7 +925,7 @@ async fn recover_unknown_upload_outcome(
             )
             .await
             .map_err(Error::other)?;
-            delete_transition_transaction_record(api, transaction).await?;
+            delete_transition_transaction_record(api, transaction.transaction_id).await?;
             Ok(TransitionTransactionRecoveryOutcome::RemoteCandidateDeleted)
         }
         TransitionCandidateProbe::VersionedPresent(version_id) => {
@@ -966,7 +958,7 @@ async fn cleanup_recovered_unknown_upload_candidate(
         .map_err(transition_transaction_store_error)?;
     save_transition_transaction_record(api.clone(), &cleanup).await?;
     delete_transition_remote_candidate(api.clone(), &cleanup).await?;
-    delete_transition_transaction_record(api, &cleanup).await?;
+    delete_transition_transaction_record(api, cleanup.transaction_id).await?;
     Ok(TransitionTransactionRecoveryOutcome::RemoteCandidateDeleted)
 }
 

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

crates/ecstore/src/config/com.rs

@@ -406,25 +406,6 @@ where
     Ok(data)
 }
 
-pub(crate) async fn read_config_limited_preserve_empty<S>(api: Arc<S>, file: &str, max_bytes: usize) -> Result<Vec<u8>>
-where
-    S: EcstoreObjectIO,
-{
-    let (data, _obj) = read_config_limited_preserve_empty_with_metadata(api, file, max_bytes).await?;
-    Ok(data)
-}
-
-pub(crate) async fn read_config_limited_preserve_empty_with_metadata<S>(
-    api: Arc<S>,
-    file: &str,
-    max_bytes: usize,
-) -> Result<(Vec<u8>, ObjectInfo)>
-where
-    S: EcstoreObjectIO,
-{
-    read_config_with_metadata_inner(api, file, &ObjectOptions::default(), true, Some(max_bytes)).await
-}
-
 /// Read an existing config object without treating an empty payload as absent.
 /// Callers that validate their own payload format need to distinguish corruption
 /// from `ConfigNotFound`.

crates/ecstore/src/set_disk/ops/object.rs

@@ -37,7 +37,7 @@ use crate::bucket::lifecycle::{
     transition_transaction::{
         TransitionRemoteVersion, TransitionSourceIdentity, TransitionSourceVersionMode, TransitionTransaction,
         TransitionTransactionInit, TransitionTransactionState, delete_transition_transaction_record,
-        load_transition_transaction_record, save_transition_transaction_record,
+        save_transition_transaction_record,
     },
 };
 use crate::bucket::quota::reservation;
@@ -4246,12 +4246,7 @@ fn record_transition_uploaded_save_attempt(transaction: &TransitionTransaction,
 
 async fn delete_transition_transaction_if_available(api: Option<&Arc<ECStore>>, transaction_id: Uuid) -> Result<()> {
     if let Some(api) = api {
-        let transaction = match load_transition_transaction_record(api.clone(), transaction_id).await {
-            Ok(transaction) => transaction,
-            Err(Error::ConfigNotFound) => return Ok(()),
-            Err(err) => return Err(err),
-        };
-        return delete_transition_transaction_record(api.clone(), &transaction).await;
+        return delete_transition_transaction_record(api.clone(), transaction_id).await;
     }
     Ok(())
 }

crates/ecstore/src/store/heal.rs

@@ -297,10 +297,16 @@ impl ECStore {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::bucket::metadata_sys;
     use crate::core::pools::{PoolDecommissionInfo, PoolStatus};
-    use crate::disk::{DiskOption, format::FormatV3, new_disk};
-    use crate::layout::endpoints::{Endpoints, PoolEndpoints};
+    use crate::disk::{DeleteOptions, DiskOption, format::FormatV3, new_disk};
+    use crate::layout::endpoints::{EndpointServerPools, Endpoints, PoolEndpoints};
+    use crate::runtime::instance::InstanceContext;
+    use crate::storage_api_contracts::bucket::{BucketOperations, MakeBucketOptions};
+    use crate::storage_api_contracts::object::{ObjectIO as _, ObjectOperations};
     use crate::store::init_format::{load_format_erasure, save_format_file};
+    use crate::store::init_local_disks_with_instance_ctx;
+    use tokio_util::sync::CancellationToken;
 
     async fn minimal_heal_pool(pool_idx: usize) -> Arc<Sets> {
         let format = FormatV3::new(1, 1);
@@ -347,6 +353,51 @@ mod tests {
         }
     }
 
+    async fn multi_pool_heal_store() -> (tempfile::TempDir, Arc<ECStore>, CancellationToken) {
+        let temp_dir = tempfile::tempdir().expect("multi-pool heal test directory should be created");
+        let mut pool_endpoints = Vec::new();
+        for pool_index in 0..2 {
+            let mut endpoints = Vec::new();
+            for disk_index in 0..4 {
+                let disk_path = temp_dir.path().join(format!("pool{pool_index}-disk{disk_index}"));
+                tokio::fs::create_dir_all(&disk_path)
+                    .await
+                    .expect("multi-pool heal test disk should be created");
+                let mut endpoint = Endpoint::try_from(disk_path.to_str().expect("disk path should be utf8"))
+                    .expect("test endpoint should parse");
+                endpoint.set_pool_index(pool_index);
+                endpoint.set_set_index(0);
+                endpoint.set_disk_index(disk_index);
+                endpoints.push(endpoint);
+            }
+            pool_endpoints.push(PoolEndpoints {
+                legacy: false,
+                set_count: 1,
+                drives_per_set: 4,
+                endpoints: Endpoints::from(endpoints),
+                cmd_line: format!("heal-owner-pool-{pool_index}"),
+                platform: "test".to_string(),
+            });
+        }
+
+        let endpoint_pools = EndpointServerPools::from(pool_endpoints);
+        let instance_ctx = Arc::new(InstanceContext::new());
+        init_local_disks_with_instance_ctx(&instance_ctx, endpoint_pools.clone())
+            .await
+            .expect("multi-pool local disks should initialize");
+        let shutdown = CancellationToken::new();
+        let store = ECStore::new_with_instance_ctx(
+            "127.0.0.1:0".parse().expect("test address should parse"),
+            endpoint_pools,
+            shutdown.clone(),
+            instance_ctx,
+        )
+        .await
+        .expect("multi-pool test store should initialize");
+        metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await;
+        (temp_dir, store, shutdown)
+    }
+
     #[tokio::test]
     async fn heal_object_pool_scope_selects_only_requested_pool() {
         let store = minimal_heal_store().await;
@@ -506,6 +557,229 @@ mod tests {
         }
     }
 
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn unscoped_heal_object_suspended_owner_semantics() {
+        let (_temp_dir, store, shutdown) = multi_pool_heal_store().await;
+        let bucket = format!("heal-owner-{}", Uuid::new_v4().simple());
+        let active_object = "active-owner";
+        let suspended_only_object = "suspended-only";
+        let duplicate_object = "duplicate-owner";
+        let marker_object = "marker-owner";
+        let quorum_object = "quorum-owner";
+        store
+            .make_bucket(&bucket, &MakeBucketOptions::default())
+            .await
+            .expect("bucket should be created in all pools");
+
+        let mut active_reader = PutObjReader::from_vec(b"active owner".to_vec());
+        store.pools[0]
+            .put_object(&bucket, active_object, &mut active_reader, &ObjectOptions::default())
+            .await
+            .expect("active owner object should be written");
+        let active_disks = store.pools[0].disk_set[0].disks.read().await.clone();
+        let missing_active_disk = active_disks[0].clone().expect("active disk should be online");
+        missing_active_disk
+            .delete(
+                &bucket,
+                active_object,
+                DeleteOptions {
+                    recursive: true,
+                    immediate: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("active owner shard should be removed for repair");
+        assert!(
+            missing_active_disk.read_xl(&bucket, active_object, false).await.is_err(),
+            "the active owner fixture must start with one missing metadata copy"
+        );
+
+        let mut suspended_reader = PutObjReader::from_vec(b"suspended owner".to_vec());
+        store.pools[1]
+            .put_object(&bucket, suspended_only_object, &mut suspended_reader, &ObjectOptions::default())
+            .await
+            .expect("suspended owner object should be written");
+        for (pool_index, mod_time) in [1_i64, 2_i64].into_iter().enumerate() {
+            let mut duplicate_reader = PutObjReader::from_vec(format!("duplicate-pool-{pool_index}").into_bytes());
+            store.pools[pool_index]
+                .put_object(
+                    &bucket,
+                    duplicate_object,
+                    &mut duplicate_reader,
+                    &ObjectOptions {
+                        mod_time: Some(OffsetDateTime::UNIX_EPOCH + time::Duration::seconds(mod_time)),
+                        ..Default::default()
+                    },
+                )
+                .await
+                .expect("duplicate owner object should be written");
+        }
+        let duplicate_missing_disk = store.pools[0].disk_set[0].disks.read().await[0]
+            .clone()
+            .expect("duplicate active owner disk should be online");
+        duplicate_missing_disk
+            .delete(
+                &bucket,
+                duplicate_object,
+                DeleteOptions {
+                    recursive: true,
+                    immediate: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("duplicate active owner shard should be removed for repair");
+        let history_version = Uuid::new_v4();
+        let mut history_reader = PutObjReader::from_vec(b"marker history".to_vec());
+        store.pools[0]
+            .put_object(
+                &bucket,
+                marker_object,
+                &mut history_reader,
+                &ObjectOptions {
+                    versioned: true,
+                    version_id: Some(history_version.to_string()),
+                    mod_time: Some(OffsetDateTime::UNIX_EPOCH + time::Duration::seconds(1)),
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("versioned marker history should be written");
+        store.pools[0]
+            .delete_object(
+                &bucket,
+                marker_object,
+                ObjectOptions {
+                    versioned: true,
+                    mod_time: Some(OffsetDateTime::UNIX_EPOCH + time::Duration::seconds(2)),
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("delete marker should be written");
+        let mut quorum_reader = PutObjReader::from_vec(b"quorum boundary".to_vec());
+        store.pools[0]
+            .put_object(&bucket, quorum_object, &mut quorum_reader, &ObjectOptions::default())
+            .await
+            .expect("quorum boundary object should be written");
+        {
+            let mut pool_meta = store.pool_meta.write().await;
+            let mut next = PoolMeta::new(&store.pools, &pool_meta);
+            next.pools[1].decommission = Some(PoolDecommissionInfo {
+                start_time: Some(OffsetDateTime::UNIX_EPOCH),
+                ..Default::default()
+            });
+            *pool_meta = next;
+        }
+
+        let (_, duplicate_owner) = store
+            .get_latest_object_info_with_idx(&bucket, duplicate_object, &ObjectOptions::default())
+            .await
+            .expect("duplicate owner should resolve");
+        assert_eq!(duplicate_owner, 1, "latest duplicate must win when all pools are eligible");
+        let (_, active_duplicate_owner) = store
+            .get_latest_object_info_with_idx(
+                &bucket,
+                duplicate_object,
+                &ObjectOptions {
+                    skip_decommissioned: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("active duplicate owner should resolve");
+        assert_eq!(
+            active_duplicate_owner, 0,
+            "suspended duplicate must be excluded from active owner selection"
+        );
+        let (duplicate_result, duplicate_err) = store
+            .handle_heal_object(&bucket, duplicate_object, "", &HealOpts::default())
+            .await
+            .expect("duplicate owner heal should complete through the production path");
+        assert_eq!(duplicate_result.object, duplicate_object);
+        assert!(duplicate_err.is_none(), "active duplicate should be repaired: {duplicate_err:?}");
+        assert!(
+            duplicate_missing_disk.read_xl(&bucket, duplicate_object, false).await.is_ok(),
+            "production heal must repair the active duplicate owner rather than the suspended owner"
+        );
+        let (marker_info, marker_owner) = store
+            .get_latest_object_info_with_idx(
+                &bucket,
+                marker_object,
+                &ObjectOptions {
+                    skip_decommissioned: true,
+                    versioned: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("latest delete marker should resolve");
+        assert_eq!(marker_owner, 0);
+        assert!(marker_info.delete_marker, "latest version must preserve delete-marker semantics");
+
+        let (active_result, active_err) = store
+            .handle_heal_object(&bucket, active_object, "", &HealOpts::default())
+            .await
+            .expect("unscoped active-owner heal should complete");
+        assert_eq!(active_result.object, active_object);
+        assert!(active_err.is_none(), "active owner must be selected even with a suspended pool");
+        assert!(
+            missing_active_disk.read_xl(&bucket, active_object, false).await.is_ok(),
+            "active owner heal must write the missing disk metadata: result={active_result:?}, err={active_err:?}"
+        );
+        assert!(
+            store.pools[1]
+                .get_object_info(&bucket, active_object, &ObjectOptions::default())
+                .await
+                .is_err(),
+            "the suspended pool must not be written for an active-owner object"
+        );
+
+        let (suspended_result, suspended_err) = store
+            .handle_heal_object(&bucket, suspended_only_object, "", &HealOpts::default())
+            .await
+            .expect("unscoped suspended-only heal should return a terminal result");
+        assert!(suspended_result.object.is_empty());
+        assert!(matches!(suspended_err, Some(Error::FileNotFound)));
+        assert!(
+            store.pools[1]
+                .get_object_info(&bucket, suspended_only_object, &ObjectOptions::default())
+                .await
+                .is_ok(),
+            "suspended-only data must remain untouched when unscoped heal reports absent"
+        );
+
+        let (_, explicit_err) = store
+            .handle_heal_object(
+                &bucket,
+                suspended_only_object,
+                "",
+                &HealOpts {
+                    pool: Some(1),
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("explicit suspended-owner heal should return a mapped error");
+        assert!(matches!(explicit_err, Some(Error::SlowDown)));
+
+        let original_quorum_disks = store.pools[0].disk_set[0].disks.read().await.clone();
+        let surviving_quorum_disk = original_quorum_disks[3].clone();
+        *store.pools[0].disk_set[0].disks.write().await = vec![None, None, None, surviving_quorum_disk];
+        let (_, quorum_err) = store
+            .handle_heal_object(&bucket, quorum_object, "", &HealOpts::default())
+            .await
+            .expect("quorum boundary heal should return a mapped result");
+        *store.pools[0].disk_set[0].disks.write().await = original_quorum_disks;
+        assert!(
+            matches!(quorum_err, Some(Error::ErasureReadQuorum)),
+            "quorum-boundary heal must preserve quorum error, got {quorum_err:?}"
+        );
+        shutdown.cancel();
+    }
+
     #[tokio::test]
     async fn handle_heal_format_continues_after_a_pool_error() {
         let canonical_format = FormatV3::new(1, 3);

crates/heal/src/heal/channel.rs

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

crates/heal/src/heal/manager.rs

@@ -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

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

@@ -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) {

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

@@ -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,

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

@@ -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);

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

@@ -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 {

crates/protos/src/node.proto

@@ -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 {

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

@@ -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() {