fix(ecstore): keep unknown-size sentinel in create_bitrot_writer

SSE and compression wrap the payload so its length is unknown and
advertise HashReader::SIZE_PRESERVE_LAYER (-1). Every layer preserved
that sentinel except create_bitrot_writer, which clamped it to 0 before
calling DiskAPI::create_file. RemoteDisk forwards that size verbatim in
the put_file_stream query, so remote peers were told the body was empty.

Since the authenticated put-file trailer (#5868) the receiver used the
declared size to split body from trailer, turning the clamp into a fatal
"auth trailer has trailing data" failure for every SSE PUT on multi-node
deployments (rc.2). #6320 relaxed the receiver to only trust size > 0;
this change fixes the sender so the sentinel survives end to end and the
wire no longer conflates empty objects with unknown-length streams.

Refs #6331

crates/ecstore/src/io_support/bitrot.rs

@@ -784,6 +784,24 @@ pub(crate) fn create_deferred_bitrot_reader_with_stripe_handle(
 ///
 /// # Returns
 /// A Result containing the BitrotWriterWrapper or an error
+/// Size hint handed to `DiskAPI::create_file` for a bitrot-wrapped shard.
+///
+/// A known length is grown by one checksum per shard so the on-disk file size
+/// matches what the bitrot writer emits. A negative length is the
+/// unknown-size sentinel (`HashReader::SIZE_PRESERVE_LAYER`, used by SSE and
+/// compression) and must be preserved: `RemoteDisk::create_file` forwards it
+/// in the `put_file_stream` query, and the receiver only treats `size > 0` as
+/// a fixed body length when locating the authenticated trailer. Clamping it
+/// to `0` would claim an empty body and misframe the stream. `0` stays `0`
+/// because a genuinely empty object still means an empty body.
+fn bitrot_create_file_size(length: i64, shard_size: usize, checksum_algo: &HashAlgorithm) -> i64 {
+    if length <= 0 {
+        return length;
+    }
+    let length = length as usize;
+    (length.div_ceil(shard_size) * checksum_algo.size() + length) as i64
+}
+
 pub async fn create_bitrot_writer(
     is_inline_buffer: bool,
     disk: Option<&DiskStore>,
@@ -796,12 +814,7 @@ pub async fn create_bitrot_writer(
     let writer = if is_inline_buffer {
         CustomWriter::new_inline_buffer()
     } else if let Some(disk) = disk {
-        let length = if length > 0 {
-            let length = length as usize;
-            (length.div_ceil(shard_size) * checksum_algo.size() + length) as i64
-        } else {
-            0
-        };
+        let length = bitrot_create_file_size(length, shard_size, &checksum_algo);
 
         let file = disk.create_file("", volume, path, length).await?;
         #[cfg(feature = "hotpath")]
@@ -820,6 +833,25 @@ mod tests {
     use rustfs_rio::ChunkReader;
     use std::collections::VecDeque;
 
+    #[test]
+    fn bitrot_create_file_size_grows_known_length_by_checksums() {
+        // 10 bytes over 4-byte shards = 3 shards, each followed by a 32-byte hash.
+        assert_eq!(bitrot_create_file_size(10, 4, &HashAlgorithm::HighwayHash256), 10 + 3 * 32);
+        assert_eq!(bitrot_create_file_size(10, 4, &HashAlgorithm::None), 10);
+    }
+
+    #[test]
+    fn bitrot_create_file_size_keeps_empty_and_unknown_distinct() {
+        assert_eq!(bitrot_create_file_size(0, 4, &HashAlgorithm::HighwayHash256), 0);
+        // SSE/compression streams advertise SIZE_PRESERVE_LAYER (-1); the remote
+        // put_file_stream receiver relies on a non-positive size to parse the auth
+        // trailer from the stream tail, so the sentinel must survive untouched.
+        assert_eq!(
+            bitrot_create_file_size(rustfs_rio::HashReader::SIZE_PRESERVE_LAYER, 4, &HashAlgorithm::HighwayHash256),
+            rustfs_rio::HashReader::SIZE_PRESERVE_LAYER
+        );
+    }
+
     struct TestChunkReader {
         chunks: VecDeque<Bytes>,
     }

crates/ecstore/src/store/heal.rs

@@ -297,16 +297,10 @@ impl ECStore {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::bucket::metadata_sys;
     use crate::core::pools::{PoolDecommissionInfo, PoolStatus};
-    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::disk::{DiskOption, format::FormatV3, new_disk};
+    use crate::layout::endpoints::{Endpoints, PoolEndpoints};
     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);
@@ -353,51 +347,6 @@ 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;
@@ -557,229 +506,6 @@ 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,6 +1640,60 @@ mod tests {
         assert_eq!(payload["items"].as_array().expect("items should be an array").len(), 0);
     }
 
+    #[tokio::test]
+    async fn test_process_query_request_reports_displaced_terminal_detail() {
+        let heal_manager = Arc::new(HealManager::new(
+            Arc::new(MockStorage),
+            Some(HealConfig {
+                queue_size: 1,
+                ..HealConfig::default()
+            }),
+        ));
+        let mut displaced = HealRequest::new(
+            HealType::Bucket {
+                bucket: "displaced-channel".to_string(),
+            },
+            HealOptions::default(),
+            HealPriority::Low,
+        );
+        displaced.id = "displaced-channel-task".to_string();
+        let displaced_id = displaced.id.clone();
+        heal_manager
+            .submit_heal_request(displaced)
+            .await
+            .expect("initial channel task should queue");
+        heal_manager
+            .submit_heal_request(HealRequest::new(
+                HealType::Bucket {
+                    bucket: "successor-channel".to_string(),
+                },
+                HealOptions::default(),
+                HealPriority::High,
+            ))
+            .await
+            .expect("successor channel task should displace the initial task");
+
+        let processor = HealChannelProcessor::new(heal_manager);
+        let (tx, rx) = oneshot::channel();
+        processor
+            .process_query_request("displaced-channel".to_string(), displaced_id, None, tx)
+            .await
+            .expect("displaced query should process");
+        let response = rx
+            .await
+            .expect("query response should be returned")
+            .expect("displaced query should remain successful");
+        let payload: serde_json::Value = serde_json::from_slice(response.data.as_deref().expect("status payload should exist"))
+            .expect("status payload should be json");
+        assert_eq!(payload["summary"], "stopped");
+        assert!(
+            response
+                .error
+                .as_deref()
+                .is_some_and(|detail| detail.contains("reason=displaced"))
+        );
+    }
+
     #[tokio::test]
     async fn test_process_query_request_reports_running_for_queued_task() {
         let heal_manager = create_test_heal_manager();

crates/heal/src/heal/manager.rs

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

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

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

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

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

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

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