fix(ecstore): hide and reclaim delete residue in prefix listings

The never-versioned listing fast path emitted any non-empty directory
without xl.meta as a prefix, which surfaced the data dirs deleted
versions leave behind as phantom folders that HEAD, listing and
recursive remove could not touch. Probe such directories for the
delete-residue shape before emitting them, and let any complete empty
first page of a prefix listing trigger the existing fail-closed orphan
purge so ordinary browse and remove traffic reclaims committed residue.

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/local.rs

@@ -98,6 +98,35 @@ const INLINE_METADATA_ROLLBACK_DIR_XOR: u128 = 0x7275737466735f696e6c696e655f726
 const DELETE_MARKER_ROLLBACK_FILE: &str = "xl.meta.delete-marker.rollback";
 pub(crate) const DELETE_DATA_DIR_MARKER_PREFIX: &str = "delete-data.";
 pub(crate) const RESERVED_DELETE_DATA_DIR_MARKER_PREFIX: &str = "reserve-delete-data.";
+/// Largest directory read the delete-residue probe issues before it must
+/// fall back to a complete read. Residue holds one or two data dirs, so an
+/// under-filled batch settles the common case without materializing large
+/// child sets; a full batch cannot prove no listable child hides behind it.
+const DELETE_RESIDUE_PROBE_LIMIT: i32 = 8;
+
+/// A `part.N` file with a positive part number, the shape erasure data takes
+/// inside a version data dir.
+pub(crate) fn metadata_less_part_file(entry: &str) -> bool {
+    entry
+        .strip_prefix("part.")
+        .is_some_and(|part_number| part_number.parse::<usize>().is_ok_and(|part_number| part_number > 0))
+}
+
+fn is_delete_transaction_marker(entry: &str, prefix: &str) -> bool {
+    entry
+        .strip_prefix(prefix)
+        .is_some_and(|transaction| Uuid::parse_str(transaction).is_ok_and(|uuid| !uuid.is_nil()))
+}
+
+/// Whether a `list_dir` entry inside a UUID data dir is erasure data or a
+/// delete-transaction marker. Anything else (a subdirectory, an `xl.meta`, an
+/// unknown file) means the directory is not plain delete residue.
+fn is_metadata_less_data_dir_entry(entry: &str) -> bool {
+    !entry.ends_with(SLASH_SEPARATOR)
+        && (metadata_less_part_file(entry)
+            || is_delete_transaction_marker(entry, DELETE_DATA_DIR_MARKER_PREFIX)
+            || is_delete_transaction_marker(entry, RESERVED_DELETE_DATA_DIR_MARKER_PREFIX))
+}
 const STARTUP_CLEANUP_WAIT_TIMEOUT: Duration = Duration::from_secs(2);
 const ENV_BITROT_SIZE_MISMATCH_RETRY_COUNT: &str = "RUSTFS_BITROT_SIZE_MISMATCH_RETRY_COUNT";
 const ENV_BITROT_SIZE_MISMATCH_RETRY_DELAY_MS: &str = "RUSTFS_BITROT_SIZE_MISMATCH_RETRY_DELAY_MS";
@@ -7644,15 +7673,18 @@ impl LocalDisk {
                         {
                             meta.name.push_str(SLASH_SEPARATOR);
                             // Conservative listings verify physical prefixes. Never-versioned
-                            // buckets use the bounded fast path and reclaim residue after an
-                            // exact recursive listing proves that prefix empty.
-                            if opts.recursive
-                                || opts.incl_deleted
-                                || opts.skip_hidden_prefix_check
-                                || self
-                                    .directory_has_listing_entry(&opts.bucket, &meta.name, opts.incl_deleted, stall)
+                            // buckets use the bounded fast path, which only has to rule out
+                            // the data dirs a deleted version leaves behind; an empty listing
+                            // of such a prefix then reclaims committed residue.
+                            let listable = if opts.recursive || opts.incl_deleted {
+                                true
+                            } else if opts.skip_hidden_prefix_check {
+                                !self.directory_is_delete_residue(&opts.bucket, &meta.name, stall).await?
+                            } else {
+                                self.directory_has_listing_entry(&opts.bucket, &meta.name, opts.incl_deleted, stall)
                                     .await?
-                            {
+                            };
+                            if listable {
                                 schedule_dir(&mut dir_stack, meta.name, false, None, true);
                             }
                         }
@@ -7776,6 +7808,74 @@ impl LocalDisk {
         Ok(false)
     }
 
+    /// Whether the metadata-less directory `dir_name` holds nothing but the
+    /// data dirs of deleted versions: it is itself a non-nil UUID directory of
+    /// `part.N` files and delete-transaction markers, or every child is one.
+    /// That is what an interrupted or deferred version delete leaves behind
+    /// once the `xl.meta` is gone, and it must not surface as a prefix. Real
+    /// object children are directories carrying their own `xl.meta`, so the
+    /// first non-UUID child, stray file, or subdirectory inside a UUID child
+    /// proves the directory is a genuine prefix. Reads are bounded: a
+    /// directory that vanishes mid-probe holds nothing listable.
+    async fn directory_is_delete_residue(&self, bucket: &str, dir_name: &str, stall: Option<Duration>) -> Result<bool> {
+        let dir_name = dir_name.trim_end_matches(SLASH_SEPARATOR);
+        let Some(entries) = self.read_dir_for_residue_probe(bucket, dir_name, stall).await? else {
+            return Ok(false);
+        };
+        if entries.is_empty() {
+            return Ok(false);
+        }
+
+        let is_data_dir = dir_name
+            .rsplit(SLASH_SEPARATOR)
+            .next()
+            .is_some_and(|name| Uuid::parse_str(name).is_ok_and(|uuid| !uuid.is_nil()));
+        if is_data_dir && entries.iter().all(|entry| is_metadata_less_data_dir_entry(entry)) {
+            return Ok(true);
+        }
+
+        for entry in entries {
+            let Some(child) = entry.strip_suffix(SLASH_SEPARATOR) else {
+                return Ok(false);
+            };
+            if !Uuid::parse_str(child).is_ok_and(|uuid| !uuid.is_nil()) {
+                return Ok(false);
+            }
+
+            let child_path = path_join_buf(&[dir_name, child]);
+            let Some(child_entries) = self.read_dir_for_residue_probe(bucket, &child_path, stall).await? else {
+                continue;
+            };
+            if !child_entries.iter().all(|entry| is_metadata_less_data_dir_entry(entry)) {
+                return Ok(false);
+            }
+        }
+
+        Ok(true)
+    }
+
+    /// Read `dir` with a bounded batch first and a complete read only when the
+    /// batch was full. `None` when the directory does not exist any more.
+    async fn read_dir_for_residue_probe(&self, bucket: &str, dir: &str, stall: Option<Duration>) -> Result<Option<Vec<String>>> {
+        for count in [DELETE_RESIDUE_PROBE_LIMIT, -1] {
+            let entries = match with_walk_stall_timeout(stall, self.list_dir("", bucket, dir, count)).await {
+                Ok(entries) => entries,
+                Err(err) => {
+                    if err == DiskError::VolumeNotFound || err == Error::FileNotFound {
+                        return Ok(None);
+                    }
+
+                    return Err(err);
+                }
+            };
+            if count < 0 || entries.len() < count as usize {
+                return Ok(Some(entries));
+            }
+        }
+
+        Ok(None)
+    }
+
     /// Whether anything under `dir_name` would appear in a listing. With
     /// `incl_deleted`, any `xl.meta` counts (versioned listings surface
     /// delete-marker-only objects too); otherwise the metadata must hold a
@@ -17755,6 +17855,133 @@ mod test {
         assert_eq!(fast_path_probes, 0);
     }
 
+    #[tokio::test]
+    async fn test_scan_dir_nonrecursive_fast_path_hides_delete_residue() {
+        use rustfs_filemeta::MetacacheReader;
+        use tempfile::tempdir;
+
+        let dir = tempdir().expect("tempdir should be created");
+        let bucket = "test-bucket";
+        let bucket_dir = dir.path().join(bucket);
+
+        async fn write_object(object_dir: &Path, object_name: &str) {
+            fs::create_dir_all(object_dir)
+                .await
+                .expect("object directory should be created");
+            let mut metadata = FileMeta::default();
+            let mut file_info = FileInfo::new(object_name, 1, 1);
+            file_info.mod_time = Some(OffsetDateTime::now_utc());
+            metadata.add_version(file_info).expect("metadata should be valid");
+            fs::write(
+                object_dir.join(STORAGE_FORMAT_FILE),
+                metadata.marshal_msg().expect("metadata should encode"),
+            )
+            .await
+            .expect("object metadata should be written");
+        }
+
+        // A deleted version whose data dir survived: part files only.
+        let residue = bucket_dir
+            .join("residue/2026/object.parquet")
+            .join(Uuid::new_v4().to_string());
+        fs::create_dir_all(&residue).await.expect("residue should be created");
+        fs::write(residue.join("part.1"), b"stale")
+            .await
+            .expect("stale part should be written");
+
+        // The same shape after a committed delete transaction.
+        let committed = bucket_dir.join("committed/object").join(Uuid::new_v4().to_string());
+        fs::create_dir_all(&committed)
+            .await
+            .expect("committed residue should be created");
+        fs::write(committed.join("part.1"), b"stale")
+            .await
+            .expect("stale part should be written");
+        fs::write(committed.join(format!("{DELETE_DATA_DIR_MARKER_PREFIX}{}", Uuid::new_v4())), [])
+            .await
+            .expect("delete marker should be written");
+
+        // A user prefix made of UUID-named directories holding real objects.
+        let upload = Uuid::new_v4().to_string();
+        write_object(&bucket_dir.join("uploads").join(&upload).join("file"), &format!("uploads/{upload}/file")).await;
+
+        // An object whose key is itself a UUID.
+        let named = Uuid::new_v4().to_string();
+        write_object(&bucket_dir.join("named").join(&named), &format!("named/{named}")).await;
+
+        // Residue next to a live child object.
+        let mixed_residue = bucket_dir.join("mixed").join(Uuid::new_v4().to_string());
+        fs::create_dir_all(&mixed_residue)
+            .await
+            .expect("mixed residue should be created");
+        fs::write(mixed_residue.join("part.1"), b"stale")
+            .await
+            .expect("stale part should be written");
+        write_object(&bucket_dir.join("mixed/child"), "mixed/child").await;
+
+        let endpoint =
+            Endpoint::try_from(dir.path().to_str().expect("tempdir path should be UTF-8")).expect("endpoint should parse");
+        let disk = LocalDisk::new(&endpoint, false).await.expect("local disk should initialize");
+
+        async fn scan_names(disk: &LocalDisk, bucket: &str, current: &str) -> Vec<String> {
+            let (reader, mut writer) = tokio::io::duplex(64 * 1024);
+            let mut output = MetacacheWriter::new(&mut writer);
+            let opts = WalkDirOptions {
+                bucket: bucket.to_string(),
+                base_dir: current.to_string(),
+                skip_hidden_prefix_check: true,
+                ..Default::default()
+            };
+            let mut objects_returned = 0;
+            disk.scan_dir(
+                current.to_string(),
+                "".to_string(),
+                &opts,
+                &mut output,
+                &mut objects_returned,
+                false,
+                None,
+            )
+            .await
+            .expect("scan_dir should succeed");
+            output.close().await.expect("metacache writer should close");
+            drop(output);
+            drop(writer);
+
+            let mut names = MetacacheReader::new(reader)
+                .read_all()
+                .await
+                .expect("scan output should decode")
+                .into_iter()
+                .map(|entry| entry.name)
+                .collect::<Vec<_>>();
+            names.sort();
+            names
+        }
+
+        // Directories whose only content is a deleted version's data dir are
+        // not prefixes; their ancestors stay ordinary directories until an
+        // empty listing reclaims them.
+        assert_eq!(scan_names(&disk, bucket, "residue/2026/").await, Vec::<String>::new());
+        assert_eq!(scan_names(&disk, bucket, "committed/").await, Vec::<String>::new());
+
+        // UUID-named directories holding real objects, an object keyed by a
+        // UUID, and residue beside a live child all remain visible.
+        assert_eq!(scan_names(&disk, bucket, "uploads/").await, vec![format!("uploads/{upload}/")]);
+        assert_eq!(scan_names(&disk, bucket, "named/").await, vec![format!("named/{named}")]);
+        assert_eq!(scan_names(&disk, bucket, "mixed/").await, vec!["mixed/child".to_owned()]);
+        assert_eq!(
+            scan_names(&disk, bucket, "").await,
+            vec![
+                "committed/".to_owned(),
+                "mixed/".to_owned(),
+                "named/".to_owned(),
+                "residue/".to_owned(),
+                "uploads/".to_owned(),
+            ]
+        );
+    }
+
     #[tokio::test]
     async fn test_scan_dir_nonrecursive_skips_dirs_with_only_hidden_delete_markers() {
         use rustfs_filemeta::MetacacheReader;

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

@@ -24,7 +24,7 @@ use super::super::{
 };
 use crate::disk::DataDirDeleteStatus;
 use crate::disk::DiskAPI;
-use crate::disk::local::DELETE_DATA_DIR_MARKER_PREFIX;
+use crate::disk::local::{DELETE_DATA_DIR_MARKER_PREFIX, metadata_less_part_file};
 use crate::io_support::bitrot::object_mmap_read_enabled;
 use crate::storage_api_contracts::namespace::NamespaceLocking as _;
 use rustfs_common::trace_bus::{TraceEvent, TraceFunc, TraceKind, trace_emit};
@@ -301,12 +301,6 @@ struct MetadataLessDataDirCleanup {
     touched_disks: Vec<bool>,
 }
 
-fn metadata_less_part_file(entry: &str) -> bool {
-    entry
-        .strip_prefix("part.")
-        .is_some_and(|part_number| part_number.parse::<usize>().is_ok_and(|part_number| part_number > 0))
-}
-
 #[cfg(test)]
 struct DanglingCheckPartsFailure {
     key: DanglingCheckPartsFailureKey,

crates/ecstore/src/store/list_objects.rs

@@ -316,10 +316,15 @@ async fn can_skip_hidden_prefix_check(options: &ListPathOptions) -> bool {
         .unwrap_or(false)
 }
 
+/// Whether an empty listing of `prefix` is the proof that lets the caller
+/// reclaim delete residue under it. Any first page that scanned the whole
+/// prefix without finding an object or a sub-prefix qualifies, with or without
+/// a delimiter: that is the shape clients issue when they stat, browse, or
+/// recursively remove a phantom folder. The purge itself re-verifies every
+/// directory on every disk before deleting anything.
 fn should_purge_empty_directory_listing(
     prefix: &str,
     marker: Option<&str>,
-    delimiter: Option<&str>,
     max_keys: i32,
     incl_deleted: bool,
     result: &ListObjectsInfo,
@@ -327,8 +332,7 @@ fn should_purge_empty_directory_listing(
     !prefix.is_empty()
         && prefix.ends_with(SLASH_SEPARATOR)
         && marker.is_none()
-        && delimiter.is_none_or(str::is_empty)
-        && max_keys == 1
+        && max_keys > 0
         && !incl_deleted
         && !result.is_truncated
         && result.objects.is_empty()
@@ -3847,16 +3851,10 @@ impl ECStore {
                 .list_objects_from_opt_in_key_only_provider(&opts, mode, max_keys, incl_deleted)
                 .await?
         {
-            if should_purge_empty_directory_listing(
-                prefix,
-                opts.marker.as_deref(),
-                delimiter.as_deref(),
-                max_keys,
-                incl_deleted,
-                &result,
-            ) && has_authoritative_never_versioned_state_in(&self.ctx, bucket)
-                .await
-                .unwrap_or(false)
+            if should_purge_empty_directory_listing(prefix, opts.marker.as_deref(), max_keys, incl_deleted, &result)
+                && has_authoritative_never_versioned_state_in(&self.ctx, bucket)
+                    .await
+                    .unwrap_or(false)
             {
                 self.purge_orphan_dir_object(bucket, prefix).await;
             }
@@ -3943,16 +3941,10 @@ impl ECStore {
             objects,
             prefixes,
         };
-        if should_purge_empty_directory_listing(
-            prefix,
-            opts.marker.as_deref(),
-            delimiter.as_deref(),
-            max_keys,
-            incl_deleted,
-            &result,
-        ) && has_authoritative_never_versioned_state_in(&self.ctx, bucket)
-            .await
-            .unwrap_or(false)
+        if should_purge_empty_directory_listing(prefix, opts.marker.as_deref(), max_keys, incl_deleted, &result)
+            && has_authoritative_never_versioned_state_in(&self.ctx, bucket)
+                .await
+                .unwrap_or(false)
         {
             self.purge_orphan_dir_object(bucket, prefix).await;
         }
@@ -8843,26 +8835,28 @@ mod test {
     }
 
     #[test]
-    fn empty_directory_listing_purge_requires_complete_exact_recursive_request() {
+    fn empty_directory_listing_purge_requires_complete_first_page_of_prefix() {
         let empty = ListObjectsInfo::default();
-        assert!(should_purge_empty_directory_listing("ghost/", None, None, 1, false, &empty));
-        assert!(should_purge_empty_directory_listing("ghost/", None, Some(""), 1, false, &empty));
-        assert!(!should_purge_empty_directory_listing("ghost", None, None, 1, false, &empty));
-        assert!(!should_purge_empty_directory_listing("ghost/", Some("marker"), None, 1, false, &empty));
-        assert!(!should_purge_empty_directory_listing("ghost/", None, Some("/"), 1, false, &empty));
-        assert!(!should_purge_empty_directory_listing("ghost/", None, None, 0, false, &empty));
-        assert!(!should_purge_empty_directory_listing("ghost/", None, None, 2, false, &empty));
-        assert!(!should_purge_empty_directory_listing("ghost/", None, None, 1, true, &empty));
+        assert!(should_purge_empty_directory_listing("ghost/", None, 1, false, &empty));
+        assert!(should_purge_empty_directory_listing("ghost/", None, 1000, false, &empty));
+        assert!(!should_purge_empty_directory_listing("ghost", None, 1, false, &empty));
+        assert!(!should_purge_empty_directory_listing("ghost/", Some("marker"), 1, false, &empty));
+        assert!(!should_purge_empty_directory_listing("ghost/", None, 0, false, &empty));
+        assert!(!should_purge_empty_directory_listing("ghost/", None, 1, true, &empty));
 
         let mut live = ListObjectsInfo::default();
         live.objects.push(ObjectInfo::default());
-        assert!(!should_purge_empty_directory_listing("ghost/", None, None, 1, false, &live));
+        assert!(!should_purge_empty_directory_listing("ghost/", None, 1, false, &live));
+
+        let mut prefixed = ListObjectsInfo::default();
+        prefixed.prefixes.push("ghost/child/".to_owned());
+        assert!(!should_purge_empty_directory_listing("ghost/", None, 1, false, &prefixed));
 
         let truncated = ListObjectsInfo {
             is_truncated: true,
             ..Default::default()
         };
-        assert!(!should_purge_empty_directory_listing("ghost/", None, None, 1, false, &truncated));
+        assert!(!should_purge_empty_directory_listing("ghost/", None, 1, false, &truncated));
     }
 
     #[tokio::test]
@@ -8916,6 +8910,59 @@ mod test {
         }
     }
 
+    #[tokio::test]
+    async fn empty_delimiter_listing_hides_and_purges_committed_delete_residue() {
+        use crate::bucket::metadata_sys::{init_bucket_metadata_sys, test_support::isolated_store_over_temp_disks};
+        use crate::storage_api_contracts::bucket::{BucketOperations as _, MakeBucketOptions};
+
+        let (dirs, store) = isolated_store_over_temp_disks().await;
+        let bucket = "listing-purge-delimiter-bucket";
+        init_bucket_metadata_sys(store.clone(), Vec::new()).await;
+        store
+            .make_bucket(bucket, &MakeBucketOptions::default())
+            .await
+            .expect("bucket should be created with authoritative metadata");
+        let data_dir = uuid::Uuid::new_v4();
+        let transaction = uuid::Uuid::new_v4();
+        for dir in &dirs {
+            let residue = dir
+                .path()
+                .join(bucket)
+                .join("metrics")
+                .join("2026")
+                .join("object.parquet")
+                .join(data_dir.to_string());
+            tokio::fs::create_dir_all(&residue)
+                .await
+                .expect("committed delete residue should be created");
+            tokio::fs::write(residue.join("part.1"), b"stale")
+                .await
+                .expect("stale part should be written");
+            tokio::fs::write(
+                residue.join(format!("{}{}", crate::disk::local::DELETE_DATA_DIR_MARKER_PREFIX, transaction)),
+                [],
+            )
+            .await
+            .expect("committed delete marker should be written");
+        }
+
+        // The object directory holds only a deleted version's data dir, so a
+        // console-style browse of its parent must not show it as a folder.
+        let result = store
+            .clone()
+            .list_objects_generic(bucket, "metrics/2026/", None, Some("/".to_owned()), 1000, false)
+            .await
+            .expect("delimiter listing should succeed");
+        assert!(result.objects.is_empty());
+        assert!(result.prefixes.is_empty(), "delete residue must not surface as a prefix");
+        for dir in &dirs {
+            assert!(
+                !dir.path().join(bucket).join("metrics").join("2026").exists(),
+                "the empty delimiter listing should reclaim the committed delete residue under it"
+            );
+        }
+    }
+
     #[test]
     fn list_objects_index_provider_state_uses_lifecycle_active_generation() {
         let provider = ListObjectsIndexProviderState::walker_key_only();

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"
-    );
-}