diff --git a/crates/ecstore/src/disk/local.rs b/crates/ecstore/src/disk/local.rs index 9127e2c9e..6baa92a3e 100644 --- a/crates/ecstore/src/disk/local.rs +++ b/crates/ecstore/src/disk/local.rs @@ -6964,6 +6964,17 @@ impl LocalDisk { while let Some((last_name, _, _, _)) = dir_stack.last() && *last_name < name { + // A prior iteration of this same loop may have just recursed + // into a pending subdirectory and hit the page limit there. + // Popping and recursing into another one anyway would still + // scan (and emit entries for) a directory beyond where the + // page was supposed to stop - stop draining the stack the + // moment the limit is reached, same as the check below this + // loop guards against for the current entry itself. + if opts.limit > 0 && *objs_returned >= opts.limit { + return Ok(()); + } + let (pop, skip_object, dir_to_skip, scan_required) = dir_stack.pop().expect("operation should succeed"); write_metacache_obj( out, @@ -6995,6 +7006,16 @@ impl LocalDisk { } } + // The while-loop above may have just recursed into a pending + // subdirectory and hit the page limit there. `name` sorts after + // that subdirectory's entries, so emitting it now would hand the + // caller a continuation marker past the subdirectory's unscanned + // tail, permanently skipping those keys on the next page instead + // of just deferring them to it. + if opts.limit > 0 && *objs_returned >= opts.limit { + return Ok(()); + } + let mut meta = MetaCacheEntry { name, ..Default::default() @@ -17030,6 +17051,92 @@ mod test { assert!(names.contains(&"quux/thud".to_string())); } + #[test] + #[serial_test::serial] + fn scan_dir_does_not_emit_entries_past_a_limit_hit_inside_a_subdirectory() { + // Reproduces the rustfs 1.0.0-rc.5 recursive-listing data loss: a + // subdirectory ("subdir/") holds more objects than the page limit, and + // a sibling object ("zzz_after") sorts after that whole subdirectory. + // scan_dir must stop at the limit and never emit "zzz_after" once the + // recursive scan of "subdir/" has already exhausted it - emitting it + // anyway hands gather_results a continuation marker that skips past + // the still-unscanned tail of "subdir/" on the next page, losing those + // keys permanently. + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("test runtime should be created"); + + runtime.block_on(async { + 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); + const SUBDIR_OBJECTS: usize = 20; + const LIMIT: i32 = 15; + + let subdir = bucket_dir.join("subdir"); + for index in 0..SUBDIR_OBJECTS { + let object_dir = subdir.join(format!("object-{index:04}")); + fs::create_dir_all(&object_dir) + .await + .expect("object directory should be created"); + fs::write(object_dir.join(STORAGE_FORMAT_FILE), b"meta") + .await + .expect("object metadata should be written"); + } + + let after_dir = bucket_dir.join("zzz_after"); + fs::create_dir_all(&after_dir) + .await + .expect("object directory should be created"); + fs::write(after_dir.join(STORAGE_FORMAT_FILE), b"meta") + .await + .expect("object metadata should be written"); + + let endpoint = + Endpoint::try_from(dir.path().to_str().expect("tempdir path should be utf8")).expect("endpoint should parse"); + let disk = LocalDisk::new(&endpoint, false).await.expect("local disk should be created"); + + let (reader, mut writer) = tokio::io::duplex(1 << 20); + let mut out = MetacacheWriter::new(&mut writer); + let opts = WalkDirOptions { + bucket: bucket.to_string(), + base_dir: String::new(), + recursive: true, + limit: LIMIT, + ..Default::default() + }; + let mut objs_returned = 0; + + disk.scan_dir(String::new(), String::new(), &opts, &mut out, &mut objs_returned, false, None) + .await + .expect("scan_dir should succeed"); + out.close().await.expect("metacache writer should close"); + drop(out); + drop(writer); + + let mut reader = MetacacheReader::new(reader); + let names: Vec = reader + .read_all() + .await + .expect("scan output should decode") + .into_iter() + .filter(|entry| !entry.metadata.is_empty()) + .map(|entry| entry.name) + .collect(); + + assert!( + !names.contains(&"zzz_after".to_string()), + "zzz_after sorts after the truncated subdir/ and must not be emitted \ + once the page limit was hit inside subdir/, or the continuation \ + marker built from it will skip subdir/'s unscanned tail forever: {names:?}" + ); + }); + } + #[test] #[serial_test::serial] fn scan_dir_records_whole_parent_read_dir_before_page_limit() { diff --git a/rustfs/src/app/gating_test_env.rs b/rustfs/src/app/gating_test_env.rs index f8d5bfd9b..d6adec4cd 100644 --- a/rustfs/src/app/gating_test_env.rs +++ b/rustfs/src/app/gating_test_env.rs @@ -29,6 +29,7 @@ use super::storage_api::test::contract::bucket::MakeBucketOptions; use super::storage_api::test::contract::bucket::{BucketOperations, BucketOptions}; use super::storage_api::test::{ECStore, Endpoint, EndpointServerPools, Endpoints, PoolEndpoints}; use super::{context::AppContext, object_traffic_health::ObjectTrafficHealth}; +use std::future::Future; use std::path::PathBuf; use std::sync::{Arc, OnceLock}; use tempfile::TempDir; @@ -38,6 +39,27 @@ use tokio_util::sync::CancellationToken; static SHARED_GATING_ENV: OnceLock<(Vec, Arc, TempDir)> = OnceLock::new(); static SHARED_GATING_INIT: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(()); +pub(crate) fn run_large_stack_test(name: &'static str, test: F) +where + F: FnOnce() -> Fut + Send + 'static, + Fut: Future + 'static, +{ + std::thread::Builder::new() + .name(name.to_string()) + .stack_size(32 * 1024 * 1024) + .spawn(move || { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("large-stack test runtime should build"); + + runtime.block_on(test()); + }) + .expect("large-stack test thread should spawn") + .join() + .expect("large-stack test thread should finish"); +} + /// Return a shared 4-disk `ECStore` with bucket metadata initialized. /// /// The first caller creates the store and initializes the metadata system; diff --git a/rustfs/src/app/object/head.rs b/rustfs/src/app/object/head.rs index 6ed8de0cd..73fd09b39 100644 --- a/rustfs/src/app/object/head.rs +++ b/rustfs/src/app/object/head.rs @@ -1003,9 +1003,16 @@ mod tests { /// runtime configured as `head = local_only`: any HEAD that wrongly /// enters the runtime shows up as a `filtered` count, so a zero counter /// proves the gate held. - #[tokio::test] + #[test] #[serial_test::serial] - async fn execute_head_object_odm_gate_against_real_store() { + fn execute_head_object_odm_gate_against_real_store() { + crate::app::gating_test_env::run_large_stack_test( + "execute-head-object-odm-gate", + execute_head_object_odm_gate_against_real_store_inner, + ); + } + + async fn execute_head_object_odm_gate_against_real_store_inner() { use crate::app::storage_api::test::contract::bucket::{BucketOperations as _, MakeBucketOptions}; let store = crate::app::gating_test_env::shared_gating_ecstore().await; diff --git a/rustfs/src/app/object/internal_put.rs b/rustfs/src/app/object/internal_put.rs index 0455ca192..cfeca6567 100644 --- a/rustfs/src/app/object/internal_put.rs +++ b/rustfs/src/app/object/internal_put.rs @@ -829,9 +829,16 @@ mod tests { assert_eq!(err.code, S3ErrorCode::InvalidRequest); } - #[tokio::test] + #[test] #[serial_test::serial] - async fn internal_multipart_roundtrip_completes_and_abort_leaves_nothing() { + fn internal_multipart_roundtrip_completes_and_abort_leaves_nothing() { + crate::app::gating_test_env::run_large_stack_test( + "internal-multipart-roundtrip", + internal_multipart_roundtrip_completes_and_abort_leaves_nothing_inner, + ); + } + + async fn internal_multipart_roundtrip_completes_and_abort_leaves_nothing_inner() { const FIRST_PART_SIZE: usize = 5 * 1024 * 1024; let (store, bucket) = internal_put_test_bucket("internal-mpu").await; let usecase = DefaultObjectUsecase::from_global();