From 23ab078c56f413266a7bf15fd678af08a6320c34 Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Tue, 1 Sep 2026 18:09:00 +0800 Subject: [PATCH] fix(ecstore): reclaim stale object prefixes (#6974) --- crates/ecstore/src/bucket/metadata_sys.rs | 10 + crates/ecstore/src/disk/local.rs | 16 +- .../src/set_disk/core/io_primitives.rs | 326 ++++++++++++++---- crates/ecstore/src/set_disk/mod.rs | 106 ++++++ crates/ecstore/src/store/list_objects.rs | 139 +++++++- crates/ecstore/src/store/object.rs | 4 +- 6 files changed, 526 insertions(+), 75 deletions(-) diff --git a/crates/ecstore/src/bucket/metadata_sys.rs b/crates/ecstore/src/bucket/metadata_sys.rs index d72ee05fe..d9f1892c1 100644 --- a/crates/ecstore/src/bucket/metadata_sys.rs +++ b/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 { + 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; diff --git a/crates/ecstore/src/disk/local.rs b/crates/ecstore/src/disk/local.rs index f6f75979f..83fc9a9dc 100644 --- a/crates/ecstore/src/disk/local.rs +++ b/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, 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); } diff --git a/crates/ecstore/src/set_disk/core/io_primitives.rs b/crates/ecstore/src/set_disk/core/io_primitives.rs index 2c8064c0e..c07a013de 100644 --- a/crates/ecstore/src/set_disk/core/io_primitives.rs +++ b/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), + /// The prefix contains only empty directories and/or UUID data directories + /// carrying a committed delete marker. + Purgeable { + empty_dirs: Vec, + committed_files: Vec, + }, /// 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 = Vec::new(); + let mut committed_files: Vec = 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, + committed_files: Vec, + ) { + // 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)> = Vec::new(); + let mut per_disk_dirs: Vec<(usize, Vec, Vec)> = 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>, default_parity_count: usize) -> Arc { let set_drive_count = disks.len(); SetDisks::new( diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index 0f8cf9193..e0c3fbb63 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/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() { diff --git a/crates/ecstore/src/store/list_objects.rs b/crates/ecstore/src/store/list_objects.rs index 216a04c33..381cd1cee 100644 --- a/crates/ecstore/src/store/list_objects.rs +++ b/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(); diff --git a/crates/ecstore/src/store/object.rs b/crates/ecstore/src/store/object.rs index c1fbf3b3f..bdc9930c1 100644 --- a/crates/ecstore/src/store/object.rs +++ b/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() {