fix(ecstore): stop scan_dir emitting entries past a limit hit inside a subdirectory (#7049)

* fix(ecstore): stop scan_dir emitting entries past a limit hit inside a subdirectory

scan_dir's flush loop recurses into a pending subdirectory when the
current sibling entry's page limit is reached mid-recursion, but kept
writing the current (later-sorting) entry regardless. gather_results
then builds the next page's continuation marker from that later entry,
which permanently skips the still-unscanned tail of the subdirectory
on resume instead of just deferring it to the next page.

Add a limit re-check right after the flush loop, before the current
entry is written, so scan_dir stops cleanly at the true last-written
key. Reproduces and fixes the rc.5 recursive ListObjectsV2 data-loss
report (7826/7881 keys, contiguous 55-key block silently dropped).

Adds scan_dir_does_not_emit_entries_past_a_limit_hit_inside_a_subdirectory.

* fix(ecstore): re-check the page limit on every dir_stack flush iteration

The flush loop that drains dir_stack can pop and recurse into more than
one pending subdirectory per outer iteration (whenever more than one
stack entry sorts below the current sibling entry). The limit re-check
added in the previous commit only ran once, after the whole flush loop
exited - so if the first recursive scan_dir call already exhausted the
page limit, the loop's next pop+recurse still went ahead and scanned
(and emitted entries for) another subdirectory beyond where the page
was supposed to stop.

Confirmed against production data: a bucket with ~1.17M objects under
one prefix still cut a recursive ListObjectsV2 listing short (825 of an
expected much larger next page, IsTruncated=false) even with the first
fix deployed, at a two-level-nested subdirectory. Move the check inside
the while loop so it runs before every pop, not just once after.

---------

Co-authored-by: Claude Agent <agent@local>

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<String> = 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() {

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<PathBuf>, Arc<ECStore>, TempDir)> = OnceLock::new();
 static SHARED_GATING_INIT: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(());
 
+pub(crate) fn run_large_stack_test<F, Fut>(name: &'static str, test: F)
+where
+    F: FnOnce() -> Fut + Send + 'static,
+    Fut: Future<Output = ()> + '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;

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;

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();