fix(ecstore): reclaim stale object prefixes (#6974)

crates/ecstore/src/bucket/metadata_sys.rs

@@ -1136,6 +1136,16 @@ pub(crate) async fn has_authoritative_never_versioned_state(bucket: &str) -> Res
     bucket_meta_sys.has_authoritative_never_versioned_state(bucket).await
 }
 
+pub(crate) async fn has_authoritative_never_versioned_state_in(
+    ctx: &crate::runtime::instance::InstanceContext,
+    bucket: &str,
+) -> Result<bool> {
+    let bucket_meta_sys_lock = bucket_metadata_sys_of(ctx)?;
+    let bucket_meta_sys = bucket_meta_sys_lock.read().await.clone();
+
+    bucket_meta_sys.has_authoritative_never_versioned_state(bucket).await
+}
+
 pub async fn get_website_config(bucket: &str) -> Result<(WebsiteConfiguration, OffsetDateTime)> {
     let bucket_meta_sys_lock = get_bucket_metadata_sys()?;
     let bucket_meta_sys = bucket_meta_sys_lock.read().await;

crates/ecstore/src/disk/local.rs

@@ -7092,6 +7092,9 @@ impl LocalDisk {
                                 .await?
                         {
                             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
@@ -17619,7 +17622,7 @@ mod test {
     }
 
     #[tokio::test]
-    async fn test_scan_dir_nonrecursive_visible_prefix_probe_cost() {
+    async fn test_scan_dir_nonrecursive_fast_path_preserves_probe_bound() {
         use rustfs_filemeta::MetacacheReader;
         use tempfile::tempdir;
 
@@ -17651,6 +17654,10 @@ mod test {
             expected_names.push(format!("{prefix}/"));
         }
 
+        fs::create_dir_all(bucket_dir.join("stale/nested/residue"))
+            .await
+            .expect("stale backing directory should be created");
+
         async fn scan_prefixes(disk: &LocalDisk, bucket: &str, skip_hidden_prefix_check: bool) -> (Vec<String>, usize) {
             let probe_count = Arc::new(AtomicUsize::new(0));
             let (reader, mut writer) = tokio::io::duplex(64 * 1024);
@@ -17693,8 +17700,11 @@ mod test {
         let (fast_path_names, fast_path_probes) = scan_prefixes(&disk, bucket, true).await;
 
         assert_eq!(conservative_names, expected_names);
-        assert_eq!(fast_path_names, expected_names);
-        assert_eq!(conservative_probes, PREFIX_COUNT * 3);
+        let mut expected_fast_path_names = expected_names.clone();
+        expected_fast_path_names.push("stale/".to_owned());
+        assert_eq!(fast_path_names, expected_fast_path_names);
+        let expected_probes = PREFIX_COUNT * 3 + 3;
+        assert_eq!(conservative_probes, expected_probes);
         assert_eq!(fast_path_probes, 0);
     }
 

crates/ecstore/src/set_disk/core/io_primitives.rs

@@ -41,22 +41,22 @@ use super::super::ENV_RUSTFS_GET_METADATA_TWO_PHASE_READ_PLAN_ENABLE;
 #[cfg(test)]
 use super::super::get_metadata_slowtail_fault_delay;
 use super::super::{
-    Bytes, CHECK_PART_DISK_NOT_FOUND, DeleteOptions, DiskError, DiskStore, EVENT_SET_DISK_RENAME_TAIL_DRAIN_FAILED,
-    EVENT_SET_DISK_WRITE, Error, FileInfo, FileMeta, FileMetaShallowVersion, GetCodecStreamingFallbackReason,
-    GetObjectMetadataCacheEntry, HTTPPreconditions, HashAlgorithm, HealAdmissionResult, HealChannelPriority, HealRequestSource,
-    LOG_COMPONENT_ECSTORE, LOG_SUBSYSTEM_SET_DISK, MultipartWriteQuorumContext, OBJECT_OP_IGNORED_ERRS, ObjectOptions,
-    ObjectPartInfo, OffsetDateTime, RUSTFS_META_BUCKET, RUSTFS_META_MULTIPART_BUCKET, RawFileInfo, ReadMultipleReq,
-    ReadMultipleResp, ReadOptions, Result, SLASH_SEPARATOR, STORAGE_FORMAT_FILE, SetDisks, SnapshotLeaseToken, StorageError,
-    UpdateMetadataOpts, Uuid, build_inline_bitrot_readers_from_refs, can_try_inline_data_shards_direct,
-    capacity_scope_from_disks, codec_streaming_rollout_applies, coding, collect_inline_data_shard_fileinfos_by_index_or_reason,
-    current_dirty_generation, debug, disk, file_info_is_valid_for_metadata, get_metadata_slowtail_fault_request, info,
-    inline_erasure_shard_file_offset, inline_erasure_shard_size, is_err_object_not_found, is_err_version_not_found,
-    is_get_metadata_data_read_early_stop_enabled, is_get_metadata_early_stop_bounded_fanout_enabled,
-    is_get_metadata_early_stop_enabled, is_get_metadata_non_inline_data_read_early_stop_enabled, is_object_dangling,
-    is_version_early_stop_enabled, issue3031_diag_enabled, join_all, join_errs, log_multipart_write_quorum_failure,
-    merge_file_meta_versions, object_fits_single_block, path_join_buf, record_global_dirty_scope, reduce_read_quorum_errs,
-    reduce_write_quorum_errs, send_heal_request_with_admission, should_prevent_write, to_object_err,
-    try_read_inline_data_shards_direct, warn,
+    Bytes, CHECK_PART_DISK_NOT_FOUND, DeleteOptions, DiskError, DiskStore, EVENT_SET_DISK_ORPHAN_PURGE_SKIPPED,
+    EVENT_SET_DISK_RENAME_TAIL_DRAIN_FAILED, EVENT_SET_DISK_WRITE, Error, FileInfo, FileMeta, FileMetaShallowVersion,
+    GetCodecStreamingFallbackReason, GetObjectMetadataCacheEntry, HTTPPreconditions, HashAlgorithm, HealAdmissionResult,
+    HealChannelPriority, HealRequestSource, LOG_COMPONENT_ECSTORE, LOG_SUBSYSTEM_SET_DISK, MultipartWriteQuorumContext,
+    OBJECT_OP_IGNORED_ERRS, ObjectOptions, ObjectPartInfo, OffsetDateTime, RUSTFS_META_BUCKET, RUSTFS_META_MULTIPART_BUCKET,
+    RawFileInfo, ReadMultipleReq, ReadMultipleResp, ReadOptions, Result, SLASH_SEPARATOR, STORAGE_FORMAT_FILE, SetDisks,
+    SnapshotLeaseToken, StorageError, UpdateMetadataOpts, Uuid, build_inline_bitrot_readers_from_refs,
+    can_try_inline_data_shards_direct, capacity_scope_from_disks, codec_streaming_rollout_applies, coding,
+    collect_inline_data_shard_fileinfos_by_index_or_reason, current_dirty_generation, debug, disk,
+    file_info_is_valid_for_metadata, get_metadata_slowtail_fault_request, info, inline_erasure_shard_file_offset,
+    inline_erasure_shard_size, is_err_object_not_found, is_err_version_not_found, is_get_metadata_data_read_early_stop_enabled,
+    is_get_metadata_early_stop_bounded_fanout_enabled, is_get_metadata_early_stop_enabled,
+    is_get_metadata_non_inline_data_read_early_stop_enabled, is_object_dangling, is_version_early_stop_enabled,
+    issue3031_diag_enabled, join_all, join_errs, log_multipart_write_quorum_failure, merge_file_meta_versions,
+    object_fits_single_block, path_join_buf, record_global_dirty_scope, reduce_read_quorum_errs, reduce_write_quorum_errs,
+    send_heal_request_with_admission, should_prevent_write, to_object_err, try_read_inline_data_shards_direct, warn,
 };
 #[cfg(test)]
 use crate::bucket::lifecycle::lifecycle::TRANSITION_COMPLETE;
@@ -3733,16 +3733,34 @@ fn dangling_delete_grace() -> time::Duration {
 /// Result of scanning one disk's copy of a directory prefix while deciding
 /// whether an orphan (metadata-less) directory tree can be safely purged.
 enum OrphanDirScan {
-    /// The subtree holds at least one regular file (object metadata or data), so
-    /// it is a real object and must not be purged.
+    /// The subtree holds object metadata or uncommitted data, so it must not be
+    /// purged.
     HasData,
-    /// The prefix exists on this disk and contains only nested empty directories.
-    /// Carries every directory path in pre-order (parents before children).
-    Empty(Vec<String>),
+    /// The prefix contains only empty directories and/or UUID data directories
+    /// carrying a committed delete marker.
+    Purgeable {
+        empty_dirs: Vec<String>,
+        committed_files: Vec<String>,
+    },
     /// The prefix does not exist on this disk.
     Missing,
 }
 
+fn is_safe_orphan_dir_entry(entry: &str) -> bool {
+    let component = entry.strip_suffix(SLASH_SEPARATOR).unwrap_or(entry);
+    !component.is_empty()
+        && component != "."
+        && component != ".."
+        && !component.contains(SLASH_SEPARATOR)
+        && !component.contains('\\')
+}
+
+fn is_committed_delete_marker(entry: &str) -> bool {
+    entry
+        .strip_prefix(DELETE_DATA_DIR_MARKER_PREFIX)
+        .is_some_and(|transaction| Uuid::parse_str(transaction).is_ok_and(|uuid| !uuid.is_nil()))
+}
+
 /// Outcome of a *post-quorum* `rename_data` commit, classifying whether the
 /// committed replicas converged so the caller can decide heal admission
 /// WITHOUT conflating "a version signature exists" with "this write needs
@@ -6125,52 +6143,151 @@ impl SetDisks {
     }
 
     /// Scan a single disk's copy of `prefix` and decide whether it is an orphan
-    /// (metadata-less) directory subtree. Walks the tree iteratively and returns
-    /// [`OrphanDirScan::HasData`] as soon as any regular file is found.
+    /// directory subtree. Only empty directories and UUID data directories with
+    /// valid committed delete markers are purgeable; every child is still scanned.
     async fn scan_orphan_dir(disk: &DiskStore, bucket: &str, prefix: &str) -> OrphanDirScan {
         let root = prefix.trim_end_matches(SLASH_SEPARATOR).to_string();
         let mut stack = vec![root.clone()];
         // Pre-order list of directories (a parent always precedes its descendants),
         // so reversing it yields a safe children-first removal order.
         let mut dirs: Vec<String> = Vec::new();
+        let mut committed_files: Vec<String> = Vec::new();
         let mut existed = false;
 
         while let Some(dir) = stack.pop() {
             let entries = match disk.list_dir("", bucket, &dir, 0).await {
                 Ok(entries) => entries,
-                Err(_) => {
-                    // The root missing (or never existing) means there is nothing to
-                    // purge on this disk. A nested directory vanishing mid-scan is a
-                    // benign race, so skip it and keep walking.
+                Err(DiskError::FileNotFound | DiskError::VolumeNotFound) => {
                     if dir == root {
                         return OrphanDirScan::Missing;
                     }
+                    // A nested directory vanishing mid-scan is a benign race.
                     continue;
                 }
+                // Classification must fail closed: committed residue is safe to
+                // remove only after every reachable child was inspected.
+                Err(_) => return OrphanDirScan::HasData,
             };
 
             existed = true;
-            dirs.push(dir.clone());
+            let mut child_dirs = Vec::new();
+            let mut files = Vec::new();
 
             for entry in entries {
+                if !is_safe_orphan_dir_entry(&entry) {
+                    return OrphanDirScan::HasData;
+                }
                 match entry.strip_suffix(SLASH_SEPARATOR) {
-                    // `read_dir` marks directories with a trailing slash; anything else
-                    // is a regular file, which means real object data lives here.
-                    Some(child) => stack.push(format!("{dir}{SLASH_SEPARATOR}{child}")),
-                    None => return OrphanDirScan::HasData,
+                    Some(child) => child_dirs.push(format!("{dir}{SLASH_SEPARATOR}{child}")),
+                    None => files.push(entry),
                 }
             }
+
+            if !files.is_empty() {
+                let data_dir_name = dir.rsplit(SLASH_SEPARATOR).next().unwrap_or_default();
+                let is_uuid_data_dir = Uuid::parse_str(data_dir_name).is_ok_and(|uuid| !uuid.is_nil());
+                let has_committed_delete = files.iter().any(|entry| is_committed_delete_marker(entry));
+
+                if !is_uuid_data_dir || !has_committed_delete || files.iter().any(|entry| entry == STORAGE_FORMAT_FILE) {
+                    return OrphanDirScan::HasData;
+                }
+
+                committed_files.extend(files.into_iter().map(|entry| path_join_buf(&[&dir, &entry])));
+                dirs.push(dir);
+                stack.extend(child_dirs);
+                continue;
+            }
+
+            dirs.push(dir);
+            stack.extend(child_dirs);
         }
 
         if existed {
-            OrphanDirScan::Empty(dirs)
+            OrphanDirScan::Purgeable {
+                empty_dirs: dirs,
+                committed_files,
+            }
         } else {
             OrphanDirScan::Missing
         }
     }
 
+    async fn delete_purgeable_orphan_entries(
+        disk: &DiskStore,
+        bucket: &str,
+        object: &str,
+        mut empty_dirs: Vec<String>,
+        committed_files: Vec<String>,
+    ) {
+        // Keep every committed marker until all ordinary residue files are gone.
+        // If any delete fails, a later request can still recognize and retry the
+        // committed cleanup instead of stranding an unmarked partial residue.
+        for delete_markers in [false, true] {
+            for file in &committed_files {
+                let is_marker = file.rsplit(SLASH_SEPARATOR).next().is_some_and(is_committed_delete_marker);
+                if is_marker != delete_markers {
+                    continue;
+                }
+                if let Err(err) = disk
+                    .delete(
+                        bucket,
+                        file,
+                        DeleteOptions {
+                            recursive: false,
+                            immediate: true,
+                            ..Default::default()
+                        },
+                    )
+                    .await
+                {
+                    debug!(
+                        event = EVENT_SET_DISK_ORPHAN_PURGE_SKIPPED,
+                        component = LOG_COMPONENT_ECSTORE,
+                        subsystem = LOG_SUBSYSTEM_SET_DISK,
+                        bucket,
+                        object,
+                        path = file,
+                        error = ?err,
+                        "Orphan prefix purge skipped"
+                    );
+                    return;
+                }
+            }
+        }
+        empty_dirs.reverse();
+        for dir in empty_dirs {
+            if let Err(err) = disk
+                .delete(
+                    bucket,
+                    &dir,
+                    DeleteOptions {
+                        recursive: false,
+                        immediate: true,
+                        ..Default::default()
+                    },
+                )
+                .await
+            {
+                // Best effort: a sibling removal may have already cleared a shared
+                // parent, or a concurrent writer repopulated the directory. Neither
+                // is fatal to purging the orphan tree.
+                debug!(
+                    event = EVENT_SET_DISK_ORPHAN_PURGE_SKIPPED,
+                    component = LOG_COMPONENT_ECSTORE,
+                    subsystem = LOG_SUBSYSTEM_SET_DISK,
+                    bucket,
+                    object,
+                    path = dir,
+                    error = ?err,
+                    "Orphan prefix purge skipped"
+                );
+            }
+        }
+    }
+
     /// Purge an orphan directory prefix — a trailing-slash key that exists on disk
-    /// as an empty directory tree with no object metadata on any disk of this set.
+    /// as empty directories or committed delete residue, with no object metadata
+    /// or uncommitted data on any disk of this set.
     /// Such prefixes are listable (see `scan_dir`) yet are not real objects, so the
     /// normal delete path returns NotFound and leaves them stranded (issue #4189).
     ///
@@ -6187,15 +6304,18 @@ impl SetDisks {
         // Phase 1: classify every online disk. Refuse to purge if ANY disk holds
         // object data under the prefix, so a degraded/healable object is never
         // destroyed.
-        let mut per_disk_dirs: Vec<(usize, Vec<String>)> = Vec::new();
+        let mut per_disk_dirs: Vec<(usize, Vec<String>, Vec<String>)> = Vec::new();
         let mut existed = false;
         for (i, disk) in disks.iter().enumerate() {
             let Some(disk) = disk else { continue };
             match Self::scan_orphan_dir(disk, bucket, object).await {
                 OrphanDirScan::HasData => return Ok(false),
-                OrphanDirScan::Empty(dirs) => {
+                OrphanDirScan::Purgeable {
+                    empty_dirs,
+                    committed_files,
+                } => {
                     existed = true;
-                    per_disk_dirs.push((i, dirs));
+                    per_disk_dirs.push((i, empty_dirs, committed_files));
                 }
                 OrphanDirScan::Missing => {}
             }
@@ -6205,32 +6325,14 @@ impl SetDisks {
             return Ok(false);
         }
 
-        // Phase 2: remove the empty directories children-first on each disk. A
-        // non-recursive delete performs an empty-only `rmdir`, so a directory that
-        // concurrently gained an object fails with DirectoryNotEmpty and is skipped —
-        // a racing PutObject is never clobbered.
-        for (i, mut dirs) in per_disk_dirs {
+        // Phase 2: remove only the files classified as committed residue, then
+        // remove directories children-first. Every directory delete is
+        // non-recursive, so a directory that concurrently gained an object fails
+        // with DirectoryNotEmpty and is skipped — a racing PutObject is never
+        // clobbered.
+        for (i, empty_dirs, committed_files) in per_disk_dirs {
             let Some(disk) = disks[i].as_ref() else { continue };
-            dirs.reverse();
-            for dir in dirs {
-                if let Err(err) = disk
-                    .delete(
-                        bucket,
-                        &dir,
-                        DeleteOptions {
-                            recursive: false,
-                            immediate: true,
-                            ..Default::default()
-                        },
-                    )
-                    .await
-                {
-                    // Best effort: a sibling removal may have already cleared a shared
-                    // parent, or a concurrent writer repopulated the directory. Neither
-                    // is fatal to purging the orphan tree.
-                    debug!(bucket, object, dir, error = ?err, "purge_orphan_dir_object: skipped non-empty/absent directory");
-                }
-            }
+            Self::delete_purgeable_orphan_entries(disk, bucket, object, empty_dirs, committed_files).await;
         }
 
         Ok(true)
@@ -7049,6 +7151,16 @@ mod tests {
     use tempfile::TempDir;
     use tokio::io::AsyncReadExt;
 
+    #[test]
+    fn orphan_dir_entries_must_be_single_relative_components() {
+        for entry in ["part.1", "child/", "delete-data.00000000-0000-0000-0000-000000000001"] {
+            assert!(is_safe_orphan_dir_entry(entry), "{entry:?} should be accepted");
+        }
+        for entry in ["", "/", ".", "..", "../", "child//", "a/b", r"a\b", "./"] {
+            assert!(!is_safe_orphan_dir_entry(entry), "{entry:?} should be rejected");
+        }
+    }
+
     #[test]
     #[serial_test::serial(codec_streaming_env)]
     fn non_inline_early_stop_is_mutually_exclusive_with_codec_rollout() {
@@ -7227,6 +7339,96 @@ mod tests {
         (dir, disk)
     }
 
+    #[tokio::test]
+    async fn orphan_cleanup_preserves_object_published_after_scan() {
+        let (dir, disk) = read_multiple_test_disk("bucket", &[]).await;
+        let transaction = Uuid::new_v4();
+        let residue = dir
+            .path()
+            .join("bucket")
+            .join("pfx")
+            .join("object")
+            .join(Uuid::new_v4().to_string());
+        tokio::fs::create_dir_all(&residue)
+            .await
+            .expect("committed data directory 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!("{DELETE_DATA_DIR_MARKER_PREFIX}{transaction}")), [])
+            .await
+            .expect("committed delete marker should be written");
+
+        let OrphanDirScan::Purgeable {
+            empty_dirs,
+            committed_files,
+        } = SetDisks::scan_orphan_dir(&disk, "bucket", "pfx/").await
+        else {
+            panic!("committed residue should be classified as purgeable");
+        };
+
+        let nested_object = residue.join("nested");
+        tokio::fs::create_dir_all(&nested_object)
+            .await
+            .expect("concurrent object directory should be created");
+        tokio::fs::write(nested_object.join(STORAGE_FORMAT_FILE), b"new metadata")
+            .await
+            .expect("concurrent object metadata should be written");
+
+        SetDisks::delete_purgeable_orphan_entries(&disk, "bucket", "pfx/", empty_dirs, committed_files).await;
+
+        assert!(
+            nested_object.join(STORAGE_FORMAT_FILE).exists(),
+            "an object published after classification must survive cleanup"
+        );
+        assert!(!residue.join("part.1").exists(), "classified stale data should be reclaimed");
+    }
+
+    #[cfg(unix)]
+    #[tokio::test]
+    async fn orphan_cleanup_keeps_commit_marker_when_residue_delete_fails() {
+        use std::os::unix::fs::PermissionsExt;
+
+        let (dir, disk) = read_multiple_test_disk("bucket", &[]).await;
+        let marker_name = format!("{DELETE_DATA_DIR_MARKER_PREFIX}{}", Uuid::new_v4());
+        let residue = dir
+            .path()
+            .join("bucket")
+            .join("pfx")
+            .join("object")
+            .join(Uuid::new_v4().to_string());
+        tokio::fs::create_dir_all(&residue)
+            .await
+            .expect("committed data directory should be created");
+        tokio::fs::write(residue.join("part.1"), b"stale")
+            .await
+            .expect("stale part should be written");
+        tokio::fs::write(residue.join(&marker_name), [])
+            .await
+            .expect("committed delete marker should be written");
+
+        let OrphanDirScan::Purgeable {
+            empty_dirs,
+            committed_files,
+        } = SetDisks::scan_orphan_dir(&disk, "bucket", "pfx/").await
+        else {
+            panic!("committed residue should be classified as purgeable");
+        };
+        tokio::fs::set_permissions(&residue, std::fs::Permissions::from_mode(0o555))
+            .await
+            .expect("residue directory should become read-only");
+
+        SetDisks::delete_purgeable_orphan_entries(&disk, "bucket", "pfx/", empty_dirs, committed_files).await;
+
+        let part_remains = residue.join("part.1").exists();
+        let marker_remains = residue.join(marker_name).exists();
+        tokio::fs::set_permissions(&residue, std::fs::Permissions::from_mode(0o755))
+            .await
+            .expect("residue directory permissions should be restored");
+        assert!(part_remains, "the injected residue delete failure should retain the part");
+        assert!(marker_remains, "the commit marker must remain so a later cleanup can retry");
+    }
+
     async fn io_primitives_test_set(disks: Vec<Option<DiskStore>>, default_parity_count: usize) -> Arc<SetDisks> {
         let set_drive_count = disks.len();
         SetDisks::new(

crates/ecstore/src/set_disk/mod.rs

@@ -329,6 +329,7 @@ const EVENT_SET_DISK_HEAL: &str = "set_disk_heal";
 const EVENT_SET_DISK_COMMIT_TAIL_SLOW: &str = "set_disk_commit_tail_slow";
 const EVENT_SET_DISK_RENAME_TAIL_DRAIN_FAILED: &str = "set_disk_rename_tail_drain_failed";
 const EVENT_SET_DISK_PUT_OBJECT_STAGE_SUMMARY: &str = "set_disk_put_object_stage_summary";
+const EVENT_SET_DISK_ORPHAN_PURGE_SKIPPED: &str = "set_disk_orphan_purge_skipped";
 const SET_DISK_COMMIT_TAIL_WARN_THRESHOLD_MS: u128 = 5_000;
 const ENV_RUSTFS_PUT_LARGE_BATCH_MIN_SIZE_BYTES: &str = "RUSTFS_PUT_LARGE_BATCH_MIN_SIZE_BYTES";
 const DEFAULT_RUSTFS_PUT_LARGE_BATCH_MIN_SIZE_BYTES: usize = 64 * 1024 * 1024;
@@ -8775,6 +8776,111 @@ mod tests {
         assert!(root.join("bucket").exists(), "bucket volume should remain");
     }
 
+    #[tokio::test]
+    async fn purge_orphan_dir_object_removes_committed_delete_residue() {
+        let (dir, disk) = make_single_local_disk().await;
+        let root = dir.path();
+        let data_dir = Uuid::new_v4();
+        let transaction = Uuid::new_v4();
+        let residue = root
+            .join("bucket")
+            .join("pfx")
+            .join("nested")
+            .join("object")
+            .join(data_dir.to_string());
+        fs::create_dir_all(&residue)
+            .await
+            .expect("committed delete residue should be created");
+        fs::write(residue.join("part.1"), b"stale")
+            .await
+            .expect("stale part should be written");
+        fs::write(
+            residue.join(format!("{}{}", crate::disk::local::DELETE_DATA_DIR_MARKER_PREFIX, transaction)),
+            [],
+        )
+        .await
+        .expect("committed delete marker should be written");
+
+        let set = make_set_disks_with(vec![Some(disk)]).await;
+        let purged = set
+            .purge_orphan_dir_object("bucket", "pfx/")
+            .await
+            .expect("purge should succeed");
+
+        assert!(purged, "committed delete residue should be purgeable");
+        assert!(!root.join("bucket").join("pfx").exists(), "prefix directory should be gone");
+    }
+
+    #[tokio::test]
+    async fn purge_orphan_dir_object_preserves_uncommitted_data_residue() {
+        let (dir, disk) = make_single_local_disk().await;
+        let root = dir.path();
+        let residue = root
+            .join("bucket")
+            .join("pfx")
+            .join("object")
+            .join(Uuid::new_v4().to_string());
+        fs::create_dir_all(&residue)
+            .await
+            .expect("uncommitted data residue should be created");
+        fs::write(residue.join("part.1"), b"possibly live")
+            .await
+            .expect("data part should be written");
+        fs::write(residue.join("delete-data.not-a-uuid"), [])
+            .await
+            .expect("malformed marker should be written");
+
+        let set = make_set_disks_with(vec![Some(disk)]).await;
+        let purged = set
+            .purge_orphan_dir_object("bucket", "pfx/")
+            .await
+            .expect("scan should succeed");
+
+        assert!(!purged, "data without a valid committed marker must be preserved");
+        assert!(residue.join("part.1").exists(), "possibly live data must remain");
+    }
+
+    #[tokio::test]
+    async fn purge_orphan_dir_object_preserves_nested_object_below_committed_residue() {
+        let (dir, disk) = make_single_local_disk().await;
+        let root = dir.path();
+        let transaction = Uuid::new_v4();
+        let residue = root
+            .join("bucket")
+            .join("pfx")
+            .join("object")
+            .join(Uuid::new_v4().to_string());
+        fs::create_dir_all(&residue)
+            .await
+            .expect("committed data directory should be created");
+        fs::write(residue.join("part.1"), b"stale")
+            .await
+            .expect("stale part should be written");
+        fs::write(
+            residue.join(format!("{}{}", crate::disk::local::DELETE_DATA_DIR_MARKER_PREFIX, transaction)),
+            [],
+        )
+        .await
+        .expect("committed delete marker should be written");
+        let nested_object = residue.join("nested");
+        fs::create_dir_all(&nested_object)
+            .await
+            .expect("nested object directory should be created");
+        fs::write(nested_object.join(STORAGE_FORMAT_FILE), b"meta")
+            .await
+            .expect("nested object metadata should be written");
+
+        let set = make_set_disks_with(vec![Some(disk)]).await;
+        let purged = set
+            .purge_orphan_dir_object("bucket", "pfx/")
+            .await
+            .expect("scan should succeed");
+
+        assert!(!purged, "nested object metadata must veto committed-residue cleanup");
+        assert!(nested_object.join(STORAGE_FORMAT_FILE).exists(), "nested object metadata must remain");
+        assert!(residue.join("part.1").exists(), "committed residue must remain when cleanup is vetoed");
+    }
+
     // issue #4189: a prefix that still anchors a real object must be left intact.
     #[tokio::test]
     async fn purge_orphan_dir_object_preserves_prefix_with_object() {

crates/ecstore/src/store/list_objects.rs

@@ -20,7 +20,9 @@ fn to_filemeta_err(err: Error) -> rustfs_filemeta::Error {
     err.narrow_to_filemeta().unwrap_or_else(rustfs_filemeta::Error::other)
 }
 
-use crate::bucket::metadata_sys::{get_versioning_config, has_authoritative_never_versioned_state};
+use crate::bucket::metadata_sys::{
+    get_versioning_config, has_authoritative_never_versioned_state, has_authoritative_never_versioned_state_in,
+};
 use crate::bucket::utils::check_list_objs_args;
 use crate::bucket::versioning::VersioningApi;
 use crate::cache_value::metacache_set::{FallbackClaimTracker, ListPathRawOptions, list_path_raw_with_claim_tracker};
@@ -314,6 +316,25 @@ async fn can_skip_hidden_prefix_check(options: &ListPathOptions) -> bool {
         .unwrap_or(false)
 }
 
+fn should_purge_empty_directory_listing(
+    prefix: &str,
+    marker: Option<&str>,
+    delimiter: Option<&str>,
+    max_keys: i32,
+    incl_deleted: bool,
+    result: &ListObjectsInfo,
+) -> bool {
+    !prefix.is_empty()
+        && prefix.ends_with(SLASH_SEPARATOR)
+        && marker.is_none()
+        && delimiter.is_none_or(str::is_empty)
+        && max_keys == 1
+        && !incl_deleted
+        && !result.is_truncated
+        && result.objects.is_empty()
+        && result.prefixes.is_empty()
+}
+
 const MARKER_TAG_VERSION: &str = "v2";
 const LEGACY_MARKER_TAG_VERSIONS: &[&str] = &["v1", MARKER_TAG_VERSION];
 const LIST_CACHE_MARKER_PREFIX: &str = "[rustfs_cache:";
@@ -3788,6 +3809,19 @@ 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)
+            {
+                self.purge_orphan_dir_object(bucket, prefix).await;
+            }
             return Ok(result);
         }
 
@@ -3818,6 +3852,7 @@ impl ECStore {
         };
 
         let mut list_result = self
+            .clone()
             .list_path(&opts)
             .await
             .unwrap_or_else(|err| MetaCacheEntriesSortedResult {
@@ -3836,7 +3871,7 @@ impl ECStore {
         }
 
         if let Some(result) = list_result.entries.as_mut() {
-            result.forward_past(opts.marker);
+            result.forward_past(opts.marker.clone());
         }
 
         // contextCanceled
@@ -3864,12 +3899,26 @@ impl ECStore {
         );
         let _ = next_version_idmarker;
 
-        Ok(ListObjectsInfo {
+        let result = ListObjectsInfo {
             is_truncated,
             next_marker,
             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)
+        {
+            self.purge_orphan_dir_object(bucket, prefix).await;
+        }
+        Ok(result)
     }
 
     pub async fn inner_list_object_versions(
@@ -6834,8 +6883,8 @@ mod test {
         LIST_CURSOR_GENERATION_LIVE, LIST_OBJECTS_INDEX_PROVIDER_PERSISTENT_KEY_ONLY,
         LIST_OBJECTS_INDEX_PROVIDER_WALKER_KEY_ONLY, ListIndexFallbackReason, ListIndexLifecycle, ListIndexLifecycleState,
         ListIndexSourceDecision, ListMetadataAuthority, ListMetadataIndexHealth, ListObjectsIndexProviderKind,
-        ListObjectsIndexProviderState, ListPathOptions, ListPathRawOptions, ListSourceMode, ListingEntryResolution,
-        ListingSupplement, ListingSupplementOptions, MAX_OBJECT_LIST, NamespaceMutationJournalBackend,
+        ListObjectsIndexProviderState, ListObjectsInfo, ListPathOptions, ListPathRawOptions, ListSourceMode,
+        ListingEntryResolution, ListingSupplement, ListingSupplementOptions, MAX_OBJECT_LIST, NamespaceMutationJournalBackend,
         NamespaceMutationJournalSnapshot, NamespaceMutationJournalStatus, PERSISTENT_KEY_ONLY_INDEX_BUCKET_HEADER,
         PERSISTENT_KEY_ONLY_INDEX_CHECKPOINT_HEADER, PERSISTENT_KEY_ONLY_INDEX_FORMAT_VERSION,
         PERSISTENT_KEY_ONLY_INDEX_GENERATION_HEADER, PERSISTENT_KEY_ONLY_INDEX_HEADER, PersistentKeyOnlyIndex,
@@ -6858,8 +6907,8 @@ mod test {
         persistent_key_only_index_health, persistent_key_only_index_matches_provider,
         reset_list_objects_mutation_sequences_for_test, resolve_agreed_listing_entry, resolve_listing_entries,
         resolve_listing_entries_with_supplement, scanner_namespace_mutation_generation, select_list_index_provider_source_mode,
-        select_list_index_source_mode, send_or_cancel, version_marker_for_entries, walk_result_from_set_errors,
-        write_namespace_mutation_journal_state, write_persistent_key_only_index_with_metadata,
+        select_list_index_source_mode, send_or_cancel, should_purge_empty_directory_listing, version_marker_for_entries,
+        walk_result_from_set_errors, write_namespace_mutation_journal_state, write_persistent_key_only_index_with_metadata,
     };
     use crate::cache_value::metacache_set::{FallbackClaimTracker, TestReaderBehavior, list_path_raw};
     use crate::disk::{DiskAPI, DiskOption, STORAGE_FORMAT_FILE, endpoint::Endpoint, error::DiskError, new_disk};
@@ -8708,6 +8757,80 @@ mod test {
         assert_eq!(scanner_namespace_mutation_generation(), 2);
     }
 
+    #[test]
+    fn empty_directory_listing_purge_requires_complete_exact_recursive_request() {
+        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));
+
+        let mut live = ListObjectsInfo::default();
+        live.objects.push(ObjectInfo::default());
+        assert!(!should_purge_empty_directory_listing("ghost/", None, None, 1, false, &live));
+
+        let truncated = ListObjectsInfo {
+            is_truncated: true,
+            ..Default::default()
+        };
+        assert!(!should_purge_empty_directory_listing("ghost/", None, None, 1, false, &truncated));
+    }
+
+    #[tokio::test]
+    async fn empty_recursive_listing_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-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("ghost")
+                .join("nested")
+                .join("object")
+                .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");
+        }
+
+        let result = store
+            .list_objects_generic(bucket, "ghost/", None, None, 1, false)
+            .await
+            .expect("empty recursive listing should succeed");
+
+        assert!(result.objects.is_empty());
+        assert!(result.prefixes.is_empty());
+        for dir in &dirs {
+            assert!(
+                !dir.path().join(bucket).join("ghost").exists(),
+                "the empty listing should reclaim its committed delete residue"
+            );
+        }
+    }
+
     #[test]
     fn list_objects_index_provider_state_uses_lifecycle_active_generation() {
         let provider = ListObjectsIndexProviderState::walker_key_only();

crates/ecstore/src/store/object.rs

@@ -3185,12 +3185,12 @@ impl ECStore {
     }
 
     /// Best-effort purge of an orphan directory prefix — an on-disk tree of empty
-    /// directories with no `xl.meta` anywhere (issue #4189). Orphan fragments can sit
+    /// directories or committed delete residue with no `xl.meta` anywhere. Orphan fragments can sit
     /// on any erasure set of any pool (they are left behind by whichever sets stored
     /// the now-deleted children), so every set is swept. Returns true when at least
     /// one set removed an orphan tree. Hard per-set failures are logged and skipped:
     /// the caller falls back to surfacing the original NotFound.
-    async fn purge_orphan_dir_object(&self, bucket: &str, object: &str) -> bool {
+    pub(super) async fn purge_orphan_dir_object(&self, bucket: &str, object: &str) -> bool {
         let prefix = decode_dir_object(object);
         let mut purged = false;
         for pool in self.pools.iter() {