fix(admin): probe set drives concurrently for storage info

The admin storage walk probed a set's drives one after another, each
bounded by the disk_info timeout, so a few drives still recovering
after a power cut pushed the local snapshot past the peer probe budget
and healthy peers rendered as unknown. Probe all drives at once so the
walk costs one timeout at most, and add a test-only probe delay hook to
pin that bound.

Cargo.lock

@@ -6139,9 +6139,9 @@ checksum = "3eaf3ede3fee6db1a4c2ee091bf8a8b4dccdc6d17f656fb07896ee72867612f2"
 
 [[package]]
 name = "libflate"
-version = "2.3.2"
+version = "2.3.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "561a8da1a50e1428d3c51321dafeca849df992a5bb67720c386131234caba82e"
+checksum = "a4da9b700e758e57152a1fd1c52cbdc5727c1aa6d8743dc1acda917398f1d76c"
 dependencies = [
  "adler32",
  "crc32fast",
@@ -10943,9 +10943,9 @@ dependencies = [
 
 [[package]]
 name = "rustfs-uring"
-version = "0.2.2"
+version = "0.2.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b29bc57b4bd62a73f4fae408b536adf578332e50e464797d09dc2382c7cb68c2"
+checksum = "0486e62d0efe25db95c00aeacb2da84368adcba299216cda99fcb11328061c84"
 dependencies = [
  "io-uring",
  "libc",
@@ -12406,7 +12406,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd"
 dependencies = [
  "fastrand",
- "getrandom 0.4.3",
+ "getrandom 0.3.4",
  "once_cell",
  "rustix",
  "windows-sys 0.59.0",

crates/ecstore/Cargo.toml

@@ -226,7 +226,7 @@ metrics = { workspace = true }
 # crates.io. The guard scripts/check_no_tokio_io_uring.sh allows an explicit
 # io-uring integration; only the tokio "io-uring" runtime feature is banned.
 [target.'cfg(target_os = "linux")'.dependencies]
-rustfs-uring = "0.2.2"
+rustfs-uring = "0.2.1"
 
 [target.'cfg(windows)'.dependencies]
 winapi-util.workspace = true

crates/ecstore/src/disk/disk_store.rs

@@ -195,6 +195,13 @@ fn resolve_drive_timeout_profile_from_env() -> DriveTimeoutProfile {
     DriveTimeoutProfile::parse(rustfs_config::DEFAULT_DRIVE_TIMEOUT_PROFILE).unwrap_or(DriveTimeoutProfile::Default)
 }
 
+#[cfg(test)]
+tokio::task_local! {
+    /// Artificial `disk_info` latency for tests that pin how the admin storage
+    /// walk composes per-drive probe time.
+    pub(crate) static DISK_INFO_PROBE_DELAY_FOR_TEST: Duration;
+}
+
 fn get_drive_timeout_profile() -> DriveTimeoutProfile {
     #[cfg(test)]
     {
@@ -2036,6 +2043,10 @@ impl DiskAPI for LocalDiskWrapper {
             .track_disk_health_with_op_and_timeout_action(
                 "disk_info",
                 || async {
+                    #[cfg(test)]
+                    if let Ok(delay) = DISK_INFO_PROBE_DELAY_FOR_TEST.try_with(|delay| *delay) {
+                        tokio::time::sleep(delay).await;
+                    }
                     let result = self.disk.disk_info(opts).await?;
 
                     if let Some(current_disk_id) = *self.disk_id.read().await

crates/ecstore/src/set_disk/mod.rs

@@ -6463,113 +6463,114 @@ pub fn should_heal_object_on_disk(
     (false, false, None)
 }
 
+/// Probe every drive of the set at once. Each live probe is bounded by the
+/// drive `disk_info` timeout, and the admin peer probe budget only covers one
+/// such timeout; a sequential walk over several stalled drives after a power
+/// cut would exceed it and make healthy peers render as unknown (#6488).
 async fn get_disks_info(disks: &[Option<DiskStore>], eps: &[Endpoint]) -> Vec<rustfs_madmin::Disk> {
-    let mut ret = Vec::new();
+    join_all(disks.iter().zip(eps).map(|(disk, ep)| disk_admin_info(disk.as_ref(), ep))).await
+}
 
-    for (i, pool) in disks.iter().enumerate() {
-        if let Some(disk) = pool {
-            let runtime_state = disk.runtime_state();
-            let offline_duration_seconds = disk.offline_duration_secs();
-            let capacity_snapshot = disk.last_capacity_snapshot();
-            let cached_disk_id = disk.cached_disk_id().await;
-            if runtime_state.should_probe_for_admin() || runtime_state == disk::health_state::RuntimeDriveHealthState::Suspect {
-                match disk
-                    .disk_info(&DiskInfoOptions {
-                        metrics: true,
-                        ..Default::default()
-                    })
-                    .await
-                {
-                    Ok(res) => {
-                        disk.record_capacity_probe(res.total, res.used, res.free);
-                        ret.push(rustfs_madmin::Disk {
-                            endpoint: eps[i].to_string(),
-                            local: eps[i].is_local,
-                            pool_index: eps[i].pool_idx,
-                            set_index: eps[i].set_idx,
-                            disk_index: eps[i].disk_idx,
-                            state: "ok".to_owned(),
+async fn disk_admin_info(disk: Option<&DiskStore>, ep: &Endpoint) -> rustfs_madmin::Disk {
+    let Some(disk) = disk else {
+        return rustfs_madmin::Disk {
+            endpoint: ep.to_string(),
+            drive_path: ep.get_file_path(),
+            local: ep.is_local,
+            pool_index: ep.pool_idx,
+            set_index: ep.set_idx,
+            disk_index: ep.disk_idx,
+            runtime_state: None,
+            offline_duration_seconds: None,
+            state: DiskError::DiskNotFound.to_string(),
+            capacity_observation_source: Some("missing".to_owned()),
+            capacity_observation_age_seconds: Some(0),
+            ..Default::default()
+        };
+    };
 
-                            root_disk: res.root_disk,
-                            drive_path: res.mount_path.clone(),
-                            healing: res.healing,
-                            scanning: res.scanning,
-                            runtime_state: Some(runtime_state.as_str().to_string()),
-                            offline_duration_seconds,
-                            capacity_observation_source: Some("live_probe".to_owned()),
-                            capacity_observation_age_seconds: Some(0),
-
-                            uuid: res.id.map_or_else(|| "".to_string(), |id| id.to_string()),
-                            major: res.major as u32,
-                            minor: res.minor as u32,
-                            model: None,
-                            total_space: res.total,
-                            used_space: res.used,
-                            available_space: res.free,
-                            physical_device_ids: (!res.physical_device_ids.is_empty()).then_some(res.physical_device_ids.clone()),
-                            utilization: utilization_percent(res.total, res.used),
-                            used_inodes: res.used_inodes,
-                            free_inodes: res.free_inodes,
-                            metrics: Some(res.metrics),
-                            ..Default::default()
-                        });
-                    }
-                    Err(err) => {
-                        let mut disk_info = rustfs_madmin::Disk {
-                            state: err.to_string(),
-                            endpoint: eps[i].to_string(),
-                            drive_path: eps[i].get_file_path(),
-                            local: eps[i].is_local,
-                            pool_index: eps[i].pool_idx,
-                            set_index: eps[i].set_idx,
-                            disk_index: eps[i].disk_idx,
-                            runtime_state: Some(runtime_state.as_str().to_string()),
-                            offline_duration_seconds,
-                            metrics: disk.metrics_snapshot(),
-                            uuid: cached_disk_id.map_or_else(String::new, |id| id.to_string()),
-                            ..Default::default()
-                        };
-                        if let Some((total, used, free, _)) = capacity_snapshot {
-                            disk_info.total_space = total;
-                            disk_info.used_space = used;
-                            disk_info.available_space = free;
-                            disk_info.utilization = utilization_percent(total, used);
-                            disk_info.capacity_observation_source = Some("snapshot".to_owned());
-                            disk_info.capacity_observation_age_seconds = capacity_snapshot
-                                .map(|(_, _, _, probe_unix_secs)| capacity_snapshot_age_seconds(probe_unix_secs));
-                        } else {
-                            disk_info.capacity_observation_source = Some("missing".to_owned());
-                            disk_info.capacity_observation_age_seconds = Some(0);
-                        }
-                        ret.push(disk_info);
-                    }
-                }
-            } else {
-                let mut disk_info =
-                    build_runtime_snapshot_disk(&eps[i], runtime_state, offline_duration_seconds, capacity_snapshot);
-                disk_info.metrics = disk.metrics_snapshot();
-                disk_info.uuid = cached_disk_id.map_or_else(String::new, |id| id.to_string());
-                ret.push(disk_info);
-            }
-        } else {
-            ret.push(rustfs_madmin::Disk {
-                endpoint: eps[i].to_string(),
-                drive_path: eps[i].get_file_path(),
-                local: eps[i].is_local,
-                pool_index: eps[i].pool_idx,
-                set_index: eps[i].set_idx,
-                disk_index: eps[i].disk_idx,
-                runtime_state: None,
-                offline_duration_seconds: None,
-                state: DiskError::DiskNotFound.to_string(),
-                capacity_observation_source: Some("missing".to_owned()),
-                capacity_observation_age_seconds: Some(0),
-                ..Default::default()
-            })
-        }
+    let runtime_state = disk.runtime_state();
+    let offline_duration_seconds = disk.offline_duration_secs();
+    let capacity_snapshot = disk.last_capacity_snapshot();
+    let cached_disk_id = disk.cached_disk_id().await;
+    if !(runtime_state.should_probe_for_admin() || runtime_state == disk::health_state::RuntimeDriveHealthState::Suspect) {
+        let mut disk_info = build_runtime_snapshot_disk(ep, runtime_state, offline_duration_seconds, capacity_snapshot);
+        disk_info.metrics = disk.metrics_snapshot();
+        disk_info.uuid = cached_disk_id.map_or_else(String::new, |id| id.to_string());
+        return disk_info;
     }
 
-    ret
+    match disk
+        .disk_info(&DiskInfoOptions {
+            metrics: true,
+            ..Default::default()
+        })
+        .await
+    {
+        Ok(res) => {
+            disk.record_capacity_probe(res.total, res.used, res.free);
+            rustfs_madmin::Disk {
+                endpoint: ep.to_string(),
+                local: ep.is_local,
+                pool_index: ep.pool_idx,
+                set_index: ep.set_idx,
+                disk_index: ep.disk_idx,
+                state: "ok".to_owned(),
+
+                root_disk: res.root_disk,
+                drive_path: res.mount_path.clone(),
+                healing: res.healing,
+                scanning: res.scanning,
+                runtime_state: Some(runtime_state.as_str().to_string()),
+                offline_duration_seconds,
+                capacity_observation_source: Some("live_probe".to_owned()),
+                capacity_observation_age_seconds: Some(0),
+
+                uuid: res.id.map_or_else(|| "".to_string(), |id| id.to_string()),
+                major: res.major as u32,
+                minor: res.minor as u32,
+                model: None,
+                total_space: res.total,
+                used_space: res.used,
+                available_space: res.free,
+                physical_device_ids: (!res.physical_device_ids.is_empty()).then_some(res.physical_device_ids.clone()),
+                utilization: utilization_percent(res.total, res.used),
+                used_inodes: res.used_inodes,
+                free_inodes: res.free_inodes,
+                metrics: Some(res.metrics),
+                ..Default::default()
+            }
+        }
+        Err(err) => {
+            let mut disk_info = rustfs_madmin::Disk {
+                state: err.to_string(),
+                endpoint: ep.to_string(),
+                drive_path: ep.get_file_path(),
+                local: ep.is_local,
+                pool_index: ep.pool_idx,
+                set_index: ep.set_idx,
+                disk_index: ep.disk_idx,
+                runtime_state: Some(runtime_state.as_str().to_string()),
+                offline_duration_seconds,
+                metrics: disk.metrics_snapshot(),
+                uuid: cached_disk_id.map_or_else(String::new, |id| id.to_string()),
+                ..Default::default()
+            };
+            if let Some((total, used, free, _)) = capacity_snapshot {
+                disk_info.total_space = total;
+                disk_info.used_space = used;
+                disk_info.available_space = free;
+                disk_info.utilization = utilization_percent(total, used);
+                disk_info.capacity_observation_source = Some("snapshot".to_owned());
+                disk_info.capacity_observation_age_seconds =
+                    capacity_snapshot.map(|(_, _, _, probe_unix_secs)| capacity_snapshot_age_seconds(probe_unix_secs));
+            } else {
+                disk_info.capacity_observation_source = Some("missing".to_owned());
+                disk_info.capacity_observation_age_seconds = Some(0);
+            }
+            disk_info
+        }
+    }
 }
 
 fn build_runtime_snapshot_disk(
@@ -10691,6 +10692,41 @@ mod tests {
         );
     }
 
+    #[tokio::test(start_paused = true)]
+    async fn test_get_disks_info_probes_drives_concurrently() {
+        use crate::disk::disk_store::DISK_INFO_PROBE_DELAY_FOR_TEST;
+
+        let format = FormatV3::new(1, 4);
+        let mut temp_dirs = Vec::new();
+        let mut endpoints = Vec::new();
+        let mut disks = Vec::new();
+        for disk_idx in 0..4 {
+            let (dir, endpoint, disk) = make_formatted_local_disk_for_info_test(disk_idx, &format).await;
+            temp_dirs.push(dir);
+            endpoints.push(endpoint);
+            disks.push(Some(disk));
+        }
+
+        let probe_delay = std::time::Duration::from_secs(2);
+        let started = tokio::time::Instant::now();
+        let info = DISK_INFO_PROBE_DELAY_FOR_TEST
+            .scope(probe_delay, get_disks_info(&disks, &endpoints))
+            .await;
+        let elapsed = started.elapsed();
+
+        assert_eq!(info.len(), 4);
+        assert!(info.iter().all(|disk| disk.state == "ok"), "every drive should still report a live probe");
+        assert_eq!(
+            info.iter().map(|disk| disk.disk_index).collect::<Vec<_>>(),
+            endpoints.iter().map(|ep| ep.disk_idx).collect::<Vec<_>>(),
+            "concurrent probes must keep endpoint order"
+        );
+        assert!(
+            elapsed < probe_delay * 2,
+            "four stalled drives must cost one probe delay, not four; took {elapsed:?}"
+        );
+    }
+
     #[tokio::test]
     async fn test_get_disks_info_preserves_remote_cached_disk_id_when_offline() {
         let (endpoint, disk) = make_remote_disk_for_info_test(0).await;

crates/heal/src/heal/mrf_queue.rs

@@ -186,11 +186,6 @@ impl MrfQueue {
         MrfQueuePushResult::Enqueued
     }
 
-    fn raise_limits_for_replay(&mut self, intents: usize, bytes: usize) {
-        self.capacity = self.capacity.max(self.pending.len().saturating_add(intents));
-        self.byte_budget = self.byte_budget.max(self.bytes.saturating_add(bytes));
-    }
-
     /// Bool compatibility adapter: only a newly executable queue item is
     /// reported as accepted; a coalesced duplicate is not durable admission.
     #[cfg(test)]
@@ -660,10 +655,9 @@ pub fn spawn_mrf_consumer(manager: Arc<HealManager>) {
 
 /// Replay the durable journal into a fresh pending queue and submit whatever
 /// it armed. Returns the number of intact intents replayed. Duplicates are
-/// merged by the manager's dedup key; the journal is retained whenever replay
-/// cannot fully hand off a successor in-memory snapshot (torn tails truncate
-/// via the per-record CRC). Public for integration tests; the live consumer
-/// invokes this through [`replay_into`] at startup.
+/// merged by the manager's dedup key; the journal file is removed once read
+/// (torn tails truncate via the per-record CRC). Public for integration tests;
+/// the live consumer invokes this through [`replay_into`] at startup.
 pub async fn replay_journal_once(manager: &Arc<HealManager>) -> usize {
     let config = MrfConsumerConfig::default();
     let mut queue = MrfQueue::new(config.queue_capacity, config.journal_max_bytes);
@@ -676,13 +670,7 @@ struct ReplayOutcome {
     journal_on_disk: bool,
 }
 
-fn replay_must_retain_journal(rearm_incomplete: bool, pending_depth: usize) -> bool {
-    rearm_incomplete || pending_depth > 0
-}
-
-/// Shared replay core: read + decode + re-arm, then drain what fits. The
-/// startup journal is removed only after every replayed record has either
-/// reached the manager or been proven redundant inside the in-memory queue.
+/// Shared replay core: read + decode + re-arm + delete, then drain what fits.
 async fn replay_into(
     manager: &Arc<HealManager>,
     queue: &mut MrfQueue,
@@ -715,26 +703,13 @@ async fn replay_into(
     }
     counter!("rustfs_heal_mrf_replayed_total").increment(u64::try_from(intents.len()).unwrap_or(u64::MAX));
     let replayed = intents.len();
-    let replay_bytes = intents
-        .iter()
-        .fold(0usize, |total, intent| total.saturating_add(intent.estimated_bytes()));
-    // The decoded journal is already resident in memory. Allow the startup
-    // queue to arm that full bounded snapshot so a later flush can become the
-    // successor anchor instead of overwriting the old journal with only a
-    // prefix.
-    queue.raise_limits_for_replay(intents.len(), replay_bytes);
-    let mut rearm_incomplete = false;
     for intent in intents {
         let result = queue.try_push_typed(intent.clone());
-        match result {
-            MrfQueuePushResult::Enqueued => {}
-            MrfQueuePushResult::Coalesced => rustfs_common::mrf_channel::release_mrf_intent(&intent),
-            MrfQueuePushResult::Rejected => {
-                rearm_incomplete = true;
-                rustfs_common::mrf_channel::release_mrf_intent(&intent);
-            }
+        if !matches!(result, MrfQueuePushResult::Enqueued) {
+            rustfs_common::mrf_channel::release_mrf_intent(&intent);
         }
     }
+    let journal_on_disk = !delete_journals().await;
 
     // Drain the replayed intents immediately; whatever the manager refuses
     // stays armed in `queue` for the consumer's retry loop.
@@ -754,11 +729,6 @@ async fn replay_into(
             }
         }
     }
-    let journal_on_disk = if replay_must_retain_journal(rearm_incomplete, queue.depth()) {
-        true
-    } else {
-        !delete_journals().await
-    };
     ReplayOutcome {
         replayed,
         journal_on_disk,
@@ -778,13 +748,13 @@ async fn run_mrf_consumer(manager: Arc<HealManager>, mut receiver: mpsc::Receive
         backoff_until: None,
     };
 
-    // Replay reads the journal and re-arms intents. The startup journal stays
-    // on disk whenever any replayed intent still needs a successor snapshot.
+    // Replay: read the journal, re-arm intents (duplicates are merged by the
+    // manager's dedup key), then drop the file so the next flush starts clean.
     let replay = replay_into(&manager, &mut runtime.queue, &mut runtime.backoff_until).await;
     runtime.journal_on_disk = replay.journal_on_disk;
-    // Anything still pending (e.g. the manager was full and backoff armed)
-    // must be re-persisted by the next flush before replay can delete the
-    // startup anchor.
+    // The replay deleted the journal file; anything still pending (e.g. the
+    // manager was full and backoff armed) must be re-persisted by the next
+    // flush or a crash before it would lose those intents.
     runtime.dirty = runtime.queue.depth() > 0;
 
     let mut flush_tick = tokio::time::interval(runtime.config.flush_interval);
@@ -920,38 +890,6 @@ mod tests {
         assert!(matches!(tick_action(false, 0, false), Idle));
     }
 
-    #[test]
-    fn replay_cleanup_retains_journal_for_unarmed_or_refused_records() {
-        assert!(
-            replay_must_retain_journal(true, 0),
-            "a rejected replay record still needs its disk anchor"
-        );
-        assert!(
-            replay_must_retain_journal(false, 1),
-            "a Full admission retry must keep the startup journal until the next snapshot"
-        );
-        assert!(
-            !replay_must_retain_journal(false, 0),
-            "only a fully consumed replay snapshot may be deleted"
-        );
-    }
-
-    #[test]
-    fn replay_can_arm_more_records_than_live_queue_budget() {
-        let mut queue = MrfQueue::new(1, intent("bucket", "object-0", 0).estimated_bytes());
-        let intents = vec![intent("bucket", "object-0", 0), intent("bucket", "object-1", 0)];
-        let bytes = intents
-            .iter()
-            .fold(0usize, |total, intent| total.saturating_add(intent.estimated_bytes()));
-
-        queue.raise_limits_for_replay(intents.len(), bytes);
-
-        for intent in intents {
-            assert_eq!(queue.try_push_typed(intent), MrfQueuePushResult::Enqueued);
-        }
-        assert_eq!(queue.depth(), 2);
-    }
-
     #[test]
     fn queue_enforces_count_and_byte_ceilings() {
         let mut queue = MrfQueue::new(2, usize::MAX);

crates/heal/tests/mrf_pipeline_test.rs

@@ -60,29 +60,6 @@ fn make_manager(storage: Arc<dyn HealStorageAPI>) -> Arc<HealManager> {
     ))
 }
 
-async fn register_local_disks(disk_paths: &[std::path::PathBuf], cmd_line: &str) {
-    let mut endpoints: Vec<Endpoint> = disk_paths
-        .iter()
-        .map(|p| Endpoint::try_from(p.to_string_lossy().as_ref()).expect("endpoint from disk path"))
-        .collect();
-    for (i, endpoint) in endpoints.iter_mut().enumerate() {
-        endpoint.set_pool_index(0);
-        endpoint.set_set_index(0);
-        endpoint.set_disk_index(i);
-    }
-    let pool = PoolEndpoints {
-        legacy: false,
-        set_count: 1,
-        drives_per_set: endpoints.len(),
-        endpoints: Endpoints::from(endpoints),
-        cmd_line: cmd_line.to_string(),
-        platform: String::new(),
-    };
-    init_local_disks(EndpointServerPools::from(vec![pool]))
-        .await
-        .expect("local disks should register");
-}
-
 /// Encode one journal record independently of the implementation, so a format
 /// drift between writer and this fixture fails loudly here.
 fn journal_record(kind: u8, bucket: &str, object: &str, version: Option<[u8; 16]>, attempts: u8) -> Vec<u8> {
@@ -174,7 +151,26 @@ async fn journal_replay_arms_intents_and_deletes_the_file() {
 
     // The journal reader resolves disks through the process-local disk map;
     // register the environment's disks the same way server startup does.
-    register_local_disks(&disk_paths, "mrf-test").await;
+    let mut endpoints: Vec<Endpoint> = disk_paths
+        .iter()
+        .map(|p| Endpoint::try_from(p.to_string_lossy().as_ref()).expect("endpoint from disk path"))
+        .collect();
+    for (i, endpoint) in endpoints.iter_mut().enumerate() {
+        endpoint.set_pool_index(0);
+        endpoint.set_set_index(0);
+        endpoint.set_disk_index(i);
+    }
+    let pool = PoolEndpoints {
+        legacy: false,
+        set_count: 1,
+        drives_per_set: endpoints.len(),
+        endpoints: Endpoints::from(endpoints),
+        cmd_line: "mrf-test".to_string(),
+        platform: String::new(),
+    };
+    init_local_disks(EndpointServerPools::from(vec![pool]))
+        .await
+        .expect("local disks should register");
 
     let mut journal = journal_record(1, "replay-bucket", "replay-object", Some([9u8; 16]), 0);
     journal.extend(journal_record(3, "replay-bucket", "partial-object", None, 1));
@@ -217,7 +213,26 @@ async fn journal_replay_arms_intents_and_deletes_the_file() {
 #[serial]
 async fn authoritative_journal_is_not_merged_with_legacy_mirror() {
     let (disk_paths, storage) = heal_env().await;
-    register_local_disks(&disk_paths, "mrf-authoritative-test").await;
+    let mut endpoints: Vec<Endpoint> = disk_paths
+        .iter()
+        .map(|p| Endpoint::try_from(p.to_string_lossy().as_ref()).expect("endpoint from disk path"))
+        .collect();
+    for (i, endpoint) in endpoints.iter_mut().enumerate() {
+        endpoint.set_pool_index(0);
+        endpoint.set_set_index(0);
+        endpoint.set_disk_index(i);
+    }
+    let pool = PoolEndpoints {
+        legacy: false,
+        set_count: 1,
+        drives_per_set: endpoints.len(),
+        endpoints: Endpoints::from(endpoints),
+        cmd_line: "mrf-authoritative-test".to_string(),
+        platform: String::new(),
+    };
+    init_local_disks(EndpointServerPools::from(vec![pool]))
+        .await
+        .expect("local disks should register");
 
     let authoritative = journal_record(1, "authoritative-bucket", "authoritative-object", None, 0);
     let legacy = journal_record(1, "legacy-bucket", "legacy-object", None, 0);
@@ -249,67 +264,3 @@ async fn authoritative_journal_is_not_merged_with_legacy_mirror() {
             && !Path::new(path).join(META_BUCKET).join(SCOPED_JOURNAL_REL).exists()
     }));
 }
-
-/// If replay reaches a full heal-manager queue, the old journal remains the
-/// durable restart anchor until a later consumer flush publishes the pending
-/// successor snapshot.
-#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
-#[serial]
-async fn journal_replay_retains_file_when_manager_is_full() {
-    let (disk_paths, storage) = heal_env().await;
-    register_local_disks(&disk_paths, "mrf-full-replay-test").await;
-
-    let mut journal = journal_record(1, "full-bucket", "first-object", None, 0);
-    journal.extend(journal_record(1, "full-bucket", "second-object", None, 0));
-    write_journal_path_to_disks(&disk_paths, SCOPED_JOURNAL_REL, &journal);
-    write_journal_path_to_disks(&disk_paths, JOURNAL_REL, &journal);
-
-    let manager = Arc::new(HealManager::new(
-        storage.clone(),
-        Some(HealConfig {
-            queue_size: 1,
-            heal_interval: Duration::from_secs(3600),
-            enable_auto_heal: false,
-            ..Default::default()
-        }),
-    ));
-    let replayed = mrf_queue::replay_journal_once(&manager).await;
-    assert_eq!(replayed, 2, "both records must be decoded before manager admission");
-    assert_eq!(
-        manager.operations_snapshot().await.queued_by_source.mrf,
-        1,
-        "only the first record can enter a one-slot manager queue"
-    );
-    assert!(
-        disk_paths
-            .iter()
-            .all(|path| Path::new(path).join(META_BUCKET).join(SCOPED_JOURNAL_REL).exists()),
-        "replay must keep the authoritative journal when a later record is pending retry"
-    );
-
-    let restarted = Arc::new(HealManager::new(
-        storage,
-        Some(HealConfig {
-            queue_size: 1,
-            heal_interval: Duration::from_secs(3600),
-            enable_auto_heal: false,
-            ..Default::default()
-        }),
-    ));
-    let replayed_after_restart = mrf_queue::replay_journal_once(&restarted).await;
-    assert_eq!(
-        replayed_after_restart, 2,
-        "retained startup journal must replay again after a process restart"
-    );
-    assert_eq!(
-        restarted.operations_snapshot().await.queued_by_source.mrf,
-        1,
-        "the restart sees the same bounded admission state instead of a lost tail"
-    );
-    assert!(
-        disk_paths
-            .iter()
-            .all(|path| Path::new(path).join(META_BUCKET).join(SCOPED_JOURNAL_REL).exists()),
-        "the anchor remains until a successor snapshot can safely replace it"
-    );
-}