diff --git a/crates/ecstore/src/bucket/metadata_sys.rs b/crates/ecstore/src/bucket/metadata_sys.rs index 96b35e206..2cbb0eff4 100644 --- a/crates/ecstore/src/bucket/metadata_sys.rs +++ b/crates/ecstore/src/bucket/metadata_sys.rs @@ -1076,6 +1076,13 @@ pub async fn get_versioning_config(bucket: &str) -> Result<(VersioningConfigurat bucket_meta_sys.get_versioning_config(bucket).await } +pub(crate) async fn has_authoritative_never_versioned_state(bucket: &str) -> Result { + let bucket_meta_sys_lock = get_bucket_metadata_sys()?; + 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; @@ -1914,6 +1921,18 @@ impl BucketMetadataSys { } } + async fn has_authoritative_never_versioned_state(&self, bucket: &str) -> Result { + let BucketMetadataAuthority::Authoritative(metadata) = self.get_metadata_authority(bucket).await? else { + return Ok(false); + }; + + if metadata.versioning_config.is_none() && !metadata.versioning_config_xml.is_empty() { + return Err(Error::other("persisted bucket versioning configuration is invalid")); + } + + Ok(metadata.versioning_config.is_none() && metadata.versioning_config_xml.is_empty()) + } + pub async fn get_bucket_policy(&self, bucket: &str) -> Result<(BucketPolicy, OffsetDateTime)> { let bm = match self.get_metadata_authority(bucket).await? { BucketMetadataAuthority::Authoritative(bm) => bm, @@ -2469,6 +2488,10 @@ mod tests { sys.get_versioning_config(bucket).await.is_err(), "malformed versioning metadata must block destructive requests" ); + assert!( + sys.has_authoritative_never_versioned_state(bucket).await.is_err(), + "malformed versioning metadata must not enable listing shortcuts" + ); assert!( sys.get_replication_config(bucket).await.is_err(), "malformed replication metadata must not be reported as ConfigNotFound" @@ -2492,8 +2515,50 @@ mod tests { sys.get_object_lock_config_state("authoritative-empty").await.unwrap(), ObjectLockConfigState::ConfirmedAbsent )); + assert!( + sys.has_authoritative_never_versioned_state("authoritative-empty") + .await + .unwrap(), + "authoritative config absence should identify a never-versioned bucket" + ); assert!(matches!(sys.get_bucket_policy("authoritative-empty").await, Err(Error::ConfigNotFound))); + let mut versioned = BucketMetadata::new("authoritative-versioned"); + versioned.versioning_config_xml = b"Enabled".to_vec(); + versioned.versioning_config = Some(VersioningConfiguration { + status: Some(s3s::dto::BucketVersioningStatus::from_static(s3s::dto::BucketVersioningStatus::ENABLED)), + ..Default::default() + }); + sys.set("authoritative-versioned".to_string(), Arc::new(versioned)).await; + assert!( + !sys.has_authoritative_never_versioned_state("authoritative-versioned") + .await + .unwrap(), + "versioned buckets must retain delete-marker visibility probes" + ); + + let mut ambiguous = BucketMetadata::new("authoritative-ambiguous-versioning"); + ambiguous.versioning_config = Some(VersioningConfiguration::default()); + sys.set("authoritative-ambiguous-versioning".to_string(), Arc::new(ambiguous)) + .await; + assert!( + !sys.has_authoritative_never_versioned_state("authoritative-ambiguous-versioning") + .await + .unwrap(), + "ambiguous versioning metadata must retain delete-marker visibility probes" + ); + + sys.fabricated_metadata + .write() + .await + .insert("fabricated-versioning".to_string()); + assert!( + !sys.has_authoritative_never_versioned_state("fabricated-versioning") + .await + .unwrap(), + "fabricated metadata must retain delete-marker visibility probes" + ); + for dir in &dirs { std::fs::create_dir_all(dir.path().join("policy-only-legacy")).unwrap(); } diff --git a/crates/ecstore/src/cache_value/metacache_set.rs b/crates/ecstore/src/cache_value/metacache_set.rs index 003960bac..c3e699e61 100644 --- a/crates/ecstore/src/cache_value/metacache_set.rs +++ b/crates/ecstore/src/cache_value/metacache_set.rs @@ -218,6 +218,7 @@ pub struct ListPathRawOptions { pub path: String, pub recursive: bool, pub incl_deleted: bool, + pub skip_hidden_prefix_check: bool, pub filter_prefix: Option, pub forward_to: Option, pub min_disks: usize, @@ -249,6 +250,7 @@ impl Clone for ListPathRawOptions { path: self.path.clone(), recursive: self.recursive, incl_deleted: self.incl_deleted, + skip_hidden_prefix_check: self.skip_hidden_prefix_check, filter_prefix: self.filter_prefix.clone(), forward_to: self.forward_to.clone(), min_disks: self.min_disks, @@ -274,6 +276,7 @@ fn walk_dir_options(opts: &ListPathRawOptions) -> WalkDirOptions { base_dir: opts.path.clone(), recursive: opts.recursive, incl_deleted: opts.incl_deleted, + skip_hidden_prefix_check: opts.skip_hidden_prefix_check, report_notfound: opts.report_not_found, filter_prefix: opts.filter_prefix.clone(), forward_to: opts.forward_to.clone(), @@ -1098,11 +1101,13 @@ mod tests { #[test] fn walk_dir_options_preserve_zero_total_and_inherited_stall_timeouts() { let options = walk_dir_options(&ListPathRawOptions { + skip_hidden_prefix_check: true, walkdir_timeout: Some(Duration::ZERO), walkdir_stall_timeout: None, ..Default::default() }); + assert!(options.skip_hidden_prefix_check); assert_eq!(options.timeout_ms, Some(0)); assert_eq!(options.stall_timeout_ms, None); assert!(!options.skip_total_timeout); diff --git a/crates/ecstore/src/cluster/rpc/remote_disk.rs b/crates/ecstore/src/cluster/rpc/remote_disk.rs index 3798970be..07d121da3 100644 --- a/crates/ecstore/src/cluster/rpc/remote_disk.rs +++ b/crates/ecstore/src/cluster/rpc/remote_disk.rs @@ -6911,7 +6911,7 @@ mod tests { } #[tokio::test] - async fn test_remote_disk_walk_dir_preserves_skip_total_timeout_option() { + async fn test_remote_disk_walk_dir_preserves_control_options() { let transport = RecordingInternodeDataTransport::default(); let remote_disk = new_remote_disk_with_transport(Arc::new(transport.clone())).await; let opts = WalkDirOptions { @@ -6919,6 +6919,7 @@ mod tests { base_dir: "prefix".to_string(), recursive: true, skip_total_timeout: true, + skip_hidden_prefix_check: true, ..Default::default() }; let mut writer = Vec::new(); @@ -6935,6 +6936,7 @@ mod tests { let sent_opts: WalkDirOptions = serde_json::from_slice(&request.body).expect("walk_dir request body should deserialize"); assert!(sent_opts.skip_total_timeout); + assert!(sent_opts.skip_hidden_prefix_check); assert_eq!(request.stall_timeout, Some(get_drive_walkdir_stall_timeout())); } other => panic!("expected walk-dir transport call, got {other:?}"), diff --git a/crates/ecstore/src/disk/local.rs b/crates/ecstore/src/disk/local.rs index 0290a040d..424cd7b25 100644 --- a/crates/ecstore/src/disk/local.rs +++ b/crates/ecstore/src/disk/local.rs @@ -80,6 +80,11 @@ use uuid::Uuid; const DELETED_OBJECTS_CLEANUP_INTERVAL: Duration = Duration::from_secs(60 * 5); const STALE_TMP_OBJECT_EXPIRY: Duration = Duration::from_secs(24 * 60 * 60); + +#[cfg(test)] +tokio::task_local! { + static DIRECTORY_LISTING_ENTRY_PROBE_COUNT: Arc; +} const RUSTFS_META_TMP_OLD_BUCKET: &str = ".rustfs.sys/tmp-old"; const INLINE_METADATA_ROLLBACK_DIR_XOR: u128 = 0x7275737466735f696e6c696e655f7262; const DELETE_MARKER_ROLLBACK_FILE: &str = "xl.meta.delete-marker.rollback"; @@ -7057,6 +7062,7 @@ impl LocalDisk { meta.name.push_str(SLASH_SEPARATOR); if opts.recursive || opts.incl_deleted + || opts.skip_hidden_prefix_check || self .directory_has_listing_entry(&opts.bucket, &meta.name, opts.incl_deleted, stall) .await? @@ -7202,6 +7208,10 @@ impl LocalDisk { continue; } + #[cfg(test)] + let _previous_probe_count = + DIRECTORY_LISTING_ENTRY_PROBE_COUNT.try_with(|probe_count| probe_count.fetch_add(1, Ordering::Relaxed)); + let entries = match with_walk_stall_timeout(stall, self.list_dir("", bucket, ¤t, -1)).await { Ok(entries) => entries, Err(err) => { @@ -17511,6 +17521,86 @@ mod test { assert_eq!(objs_returned, 1); } + #[tokio::test] + async fn test_scan_dir_nonrecursive_visible_prefix_probe_cost() { + use rustfs_filemeta::MetacacheReader; + use tempfile::tempdir; + + const PREFIX_COUNT: usize = 64; + + let dir = tempdir().expect("tempdir should be created"); + let bucket = "test-bucket"; + let bucket_dir = dir.path().join(bucket); + let mut expected_names = Vec::with_capacity(PREFIX_COUNT); + + for index in 0..PREFIX_COUNT { + let prefix = format!("prefix-{index:04}"); + let object_name = format!("{prefix}/nested/object"); + let object_dir = bucket_dir.join(&object_name); + fs::create_dir_all(&object_dir) + .await + .expect("visible 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("visible metadata should be valid"); + fs::write( + object_dir.join(STORAGE_FORMAT_FILE), + metadata.marshal_msg().expect("visible metadata should encode"), + ) + .await + .expect("visible object metadata should be written"); + expected_names.push(format!("{prefix}/")); + } + + 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); + let mut output = MetacacheWriter::new(&mut writer); + let opts = WalkDirOptions { + bucket: bucket.to_string(), + skip_hidden_prefix_check, + ..Default::default() + }; + let mut objects_returned = 0; + + DIRECTORY_LISTING_ENTRY_PROBE_COUNT + .scope( + Arc::clone(&probe_count), + disk.scan_dir("".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 reader = MetacacheReader::new(reader); + let names = reader + .read_all() + .await + .expect("scan output should decode") + .into_iter() + .map(|entry| entry.name) + .collect::>(); + + (names, probe_count.load(Ordering::Relaxed)) + } + + 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"); + + let (conservative_names, conservative_probes) = scan_prefixes(&disk, bucket, false).await; + 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); + assert_eq!(fast_path_probes, 0); + } + #[tokio::test] async fn test_scan_dir_nonrecursive_skips_dirs_with_only_hidden_delete_markers() { use rustfs_filemeta::MetacacheReader; diff --git a/crates/ecstore/src/disk/mod.rs b/crates/ecstore/src/disk/mod.rs index 6a51a1e41..fb69085c4 100644 --- a/crates/ecstore/src/disk/mod.rs +++ b/crates/ecstore/src/disk/mod.rs @@ -1231,6 +1231,11 @@ pub struct WalkDirOptions { #[serde(default)] pub incl_deleted: bool, + // Skip recursive prefix visibility probes only when authoritative bucket + // metadata proves versioning was never enabled. + #[serde(default)] + pub skip_hidden_prefix_check: bool, + // ReportNotFound will return errFileNotFound if all disks reports the BaseDir cannot be found. pub report_notfound: bool, @@ -1569,6 +1574,7 @@ mod tests { base_dir: "/path/to/dir".to_string(), recursive: true, incl_deleted: false, + skip_hidden_prefix_check: false, report_notfound: false, filter_prefix: Some("prefix_".to_string()), forward_to: Some("object/path".to_string()), @@ -1583,6 +1589,7 @@ mod tests { assert_eq!(opts.base_dir, "/path/to/dir"); assert!(opts.recursive); assert!(!opts.incl_deleted); + assert!(!opts.skip_hidden_prefix_check); assert!(!opts.report_notfound); assert_eq!(opts.filter_prefix, Some("prefix_".to_string())); assert_eq!(opts.forward_to, Some("object/path".to_string())); @@ -1593,6 +1600,23 @@ mod tests { assert_eq!(opts.stall_timeout_duration(), Some(std::time::Duration::from_secs(20))); } + #[test] + fn test_walk_dir_options_default_hidden_prefix_check_for_old_peers() { + let mut encoded = serde_json::to_value(WalkDirOptions { + skip_hidden_prefix_check: true, + ..Default::default() + }) + .expect("walk options should serialize"); + encoded + .as_object_mut() + .expect("walk options should serialize as an object") + .remove("skip_hidden_prefix_check"); + + let decoded: WalkDirOptions = serde_json::from_value(encoded).expect("old peer options should deserialize"); + + assert!(!decoded.skip_hidden_prefix_check); + } + /// Test DeleteOptions structure #[test] fn test_delete_options() { diff --git a/crates/ecstore/src/store/list_objects.rs b/crates/ecstore/src/store/list_objects.rs index 0d8d238f9..5b18481d7 100644 --- a/crates/ecstore/src/store/list_objects.rs +++ b/crates/ecstore/src/store/list_objects.rs @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -use crate::bucket::metadata_sys::get_versioning_config; +use crate::bucket::metadata_sys::{get_versioning_config, has_authoritative_never_versioned_state}; 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}; @@ -261,6 +261,9 @@ pub struct ListPathOptions { // InclDeleted will keep all entries where latest version is a delete marker. pub incl_deleted: bool, + // Authoritative bucket metadata proves that delete markers cannot exist. + pub skip_hidden_prefix_check: bool, + // Scan recursively. // If false only main directory will be scanned. // Should always be true if Separator is n SlashSeparator. @@ -293,6 +296,16 @@ pub struct ListPathOptions { pub walkdir_stall_timeout: Option, } +async fn can_skip_hidden_prefix_check(options: &ListPathOptions) -> bool { + if options.recursive || options.incl_deleted || options.versioned { + return false; + } + + has_authoritative_never_versioned_state(&options.bucket) + .await + .unwrap_or(false) +} + const MARKER_TAG_VERSION: &str = "v2"; const LEGACY_MARKER_TAG_VERSIONS: &[&str] = &["v1", MARKER_TAG_VERSION]; const LIST_CACHE_MARKER_PREFIX: &str = "[rustfs_cache:"; @@ -2523,6 +2536,7 @@ struct ListingSupplementOptions { path: String, recursive: bool, incl_deleted: bool, + skip_hidden_prefix_check: bool, filter_prefix: Option, forward_to: Option, per_disk_limit: i32, @@ -2655,6 +2669,7 @@ async fn read_fallback_listing_disk( base_dir: options.path, recursive: options.recursive, incl_deleted: options.incl_deleted, + skip_hidden_prefix_check: options.skip_hidden_prefix_check, report_notfound: false, filter_prefix: options.filter_prefix, forward_to: options.forward_to, @@ -4010,6 +4025,8 @@ impl ECStore { o.recursive = true } + o.skip_hidden_prefix_check = can_skip_hidden_prefix_check(&o).await; + o.parse_marker(); if o.base_dir.is_empty() { @@ -4357,6 +4374,7 @@ impl ECStore { path: path.clone(), recursive: true, incl_deleted: !opts.latest_only, + skip_hidden_prefix_check: false, filter_prefix: Some(filter_prefix.clone()), forward_to: opts.marker.clone(), per_disk_limit: bounded_usize_to_i32(opts.limit), @@ -5288,6 +5306,8 @@ impl Sets { o.recursive = true; } + o.skip_hidden_prefix_check = can_skip_hidden_prefix_check(&o).await; + o.parse_marker(); if o.base_dir.is_empty() { @@ -5583,6 +5603,7 @@ impl Sets { path: path.clone(), recursive: true, incl_deleted: !opts.latest_only, + skip_hidden_prefix_check: false, filter_prefix: Some(filter_prefix.clone()), forward_to: opts.marker.clone(), per_disk_limit: bounded_usize_to_i32(opts.limit), @@ -6338,6 +6359,8 @@ impl SetDisks { o.recursive = true; } + o.skip_hidden_prefix_check = can_skip_hidden_prefix_check(&o).await; + o.parse_marker(); if o.base_dir.is_empty() { @@ -6562,6 +6585,7 @@ impl SetDisks { path: opts.base_dir.clone(), recursive: opts.recursive, incl_deleted: opts.incl_deleted, + skip_hidden_prefix_check: opts.skip_hidden_prefix_check, filter_prefix: opts.filter_prefix.clone(), forward_to: opts.marker.clone(), per_disk_limit: limit, @@ -6584,6 +6608,7 @@ impl SetDisks { path: opts.base_dir, recursive: opts.recursive, incl_deleted: opts.incl_deleted, + skip_hidden_prefix_check: opts.skip_hidden_prefix_check, filter_prefix: opts.filter_prefix, forward_to: opts.marker, min_disks: raw_min_disks, @@ -6960,6 +6985,7 @@ mod test { path: String::new(), recursive: true, incl_deleted: false, + skip_hidden_prefix_check: false, filter_prefix: None, forward_to: None, per_disk_limit: 100, @@ -6976,6 +7002,7 @@ mod test { path: String::new(), recursive: true, incl_deleted: false, + skip_hidden_prefix_check: false, filter_prefix: None, forward_to: None, per_disk_limit: 0, @@ -7032,6 +7059,7 @@ mod test { path: String::new(), recursive: true, incl_deleted: false, + skip_hidden_prefix_check: false, filter_prefix: None, forward_to: None, per_disk_limit: 0, diff --git a/rustfs/src/app/object_usecase.rs b/rustfs/src/app/object_usecase.rs index 61759886e..45ec8592b 100644 --- a/rustfs/src/app/object_usecase.rs +++ b/rustfs/src/app/object_usecase.rs @@ -18081,7 +18081,7 @@ mod tests { async fn observed_bucket_usage(bucket: &str) -> Option { let mut usage = rustfs_data_usage::DataUsageInfo::default(); - apply_bucket_usage_memory_overlay(&mut usage).await; + crate::app::storage_api::object_usecase::data_usage::apply_bucket_usage_memory_overlay(&mut usage).await; usage.buckets_usage.get(bucket).map(|value| value.size) }