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.

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
+                            // buckets use the bounded fast path, which only has to rule out
-                            // exact recursive listing proves that prefix empty.
+                            // the data dirs a deleted version leaves behind; an empty listing
-                            if opts.recursive
+                            // of such a prefix then reclaims committed residue.
-                                || opts.incl_deleted
+                            let listable = if opts.recursive || opts.incl_deleted {
-                                || opts.skip_hidden_prefix_check
+                                true
-                                || self
+                            } else if opts.skip_hidden_prefix_check {
-                                    .directory_has_listing_entry(&opts.bucket, &meta.name, opts.incl_deleted, stall)
+                                !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 > 0
-        && max_keys == 1
         && !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(
+            if should_purge_empty_directory_listing(prefix, opts.marker.as_deref(), max_keys, incl_deleted, &result)
-                prefix,
+                && has_authoritative_never_versioned_state_in(&self.ctx, bucket)
-                opts.marker.as_deref(),
+                    .await
-                delimiter.as_deref(),
+                    .unwrap_or(false)
-                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(
+        if should_purge_empty_directory_listing(prefix, opts.marker.as_deref(), max_keys, incl_deleted, &result)
-            prefix,
+            && has_authoritative_never_versioned_state_in(&self.ctx, bucket)
-            opts.marker.as_deref(),
+                .await
-            delimiter.as_deref(),
+                .unwrap_or(false)
-            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, 1, false, &empty));
-        assert!(should_purge_empty_directory_listing("ghost/", None, Some(""), 1, false, &empty));
+        assert!(should_purge_empty_directory_listing("ghost/", None, 1000, false, &empty));
-        assert!(!should_purge_empty_directory_listing("ghost", None, None, 1, false, &empty));
+        assert!(!should_purge_empty_directory_listing("ghost", None, 1, false, &empty));
-        assert!(!should_purge_empty_directory_listing("ghost/", Some("marker"), None, 1, false, &empty));
+        assert!(!should_purge_empty_directory_listing("ghost/", Some("marker"), 1, false, &empty));
-        assert!(!should_purge_empty_directory_listing("ghost/", None, Some("/"), 1, false, &empty));
+        assert!(!should_purge_empty_directory_listing("ghost/", None, 0, false, &empty));
-        assert!(!should_purge_empty_directory_listing("ghost/", None, None, 0, false, &empty));
+        assert!(!should_purge_empty_directory_listing("ghost/", None, 1, true, &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));
 
         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();