fix(listing): keep the limit flag monotonic and fix metacache helpers (#8906)

`LocalDisk::walk_dir` stored its own `limit_reached` into the
`producer_limit_reached` flag that every producer of a page shares, so a
walk reaching EOF after a sibling stopped at its per-disk limit cleared the
signal and the collector reported a complete page. Use `fetch_or` so the
flag answers "did any bounded walk stop early", whatever the finish order.

`MetaCacheEntry::is_in_dir` stripped the directory prefix repeatedly, so
`a/a/b` counted as a direct child of `a/`. `MetaCacheEntriesSorted::
forward_past` kept every entry when none sorted after the marker and split
on an index computed over the flattened entries. Both are masked today by
the collector's pre-filtering; fix them and pin the contracts with tests.

crates/ecstore/src/disk/local.rs

@@ -12389,7 +12389,11 @@ impl DiskAPI for LocalDisk {
             .await?;
 
         if let Some(flag) = opts.producer_limit_reached.as_ref() {
-            flag.store(limit_reached, std::sync::atomic::Ordering::Release);
+            // Every disk of every set in a page shares this flag, and the
+            // collector reads it as "did ANY bounded walk stop at its limit".
+            // A walk that reached EOF must not clear what another producer
+            // reported, whatever order they finish in.
+            flag.fetch_or(limit_reached, std::sync::atomic::Ordering::AcqRel);
         }
         out.close().await?;
         Ok(())
@@ -15934,6 +15938,65 @@ mod test {
         );
     }
 
+    // The collector decides "truncated or exhausted" from one flag shared by every
+    // producer of the page. It must stay set once any bounded walk stopped early,
+    // however the remaining producers finish.
+    #[tokio::test]
+    async fn walk_dir_keeps_shared_limit_flag_set_by_another_producer() {
+        use std::sync::Arc;
+        use std::sync::atomic::{AtomicBool, Ordering};
+        use tempfile::tempdir;
+
+        let dir = tempdir().expect("temp dir should be created");
+        let bucket = "test-bucket";
+        for idx in 0..3 {
+            let object_dir = dir.path().join(bucket).join(format!("prefix/object-{idx}"));
+            fs::create_dir_all(&object_dir).await.expect("object dir should be created");
+            fs::write(object_dir.join(STORAGE_FORMAT_FILE), b"meta")
+                .await
+                .expect("object metadata should be written");
+        }
+
+        let endpoint = Endpoint::try_from(dir.path().to_str().expect("temp dir should be utf8")).expect("endpoint should parse");
+        let disk = LocalDisk::new(&endpoint, false).await.expect("local disk should be created");
+        disk.wait_for_startup_cleanup().await;
+
+        let walk = |limit: i32, flag: Arc<AtomicBool>| {
+            let disk = &disk;
+            async move {
+                let mut sink = Vec::new();
+                disk.walk_dir(
+                    WalkDirOptions {
+                        bucket: bucket.to_string(),
+                        base_dir: "prefix/".to_string(),
+                        recursive: true,
+                        limit,
+                        producer_limit_reached: Some(flag),
+                        ..Default::default()
+                    },
+                    &mut sink,
+                )
+                .await
+                .expect("walk should succeed");
+            }
+        };
+
+        let bounded = Arc::new(AtomicBool::new(false));
+        walk(1, bounded.clone()).await;
+        assert!(
+            bounded.load(Ordering::Acquire),
+            "a walk that stopped at its limit must report it through the shared flag"
+        );
+
+        // A sibling producer already stopped at its limit; this walk reaches EOF after it.
+        let shared = Arc::new(AtomicBool::new(true));
+        walk(0, shared.clone()).await;
+        assert!(
+            shared.load(Ordering::Acquire),
+            "a walk that reached EOF must not clear a limit another producer reported"
+        );
+    }
+
     // #4644: the stall timeout is what catches a drive that stops answering.
     //
     // This exercises the bound directly rather than through `walk_dir`. A drive

crates/filemeta/src/metacache.rs

@@ -139,14 +139,13 @@ impl MetaCacheEntry {
             return idx.is_none() || idx.unwrap() == self.name.len() - separator.len();
         }
 
-        let ext = self.name.trim_start_matches(dir);
+        // Strip `dir` exactly once: a repeated strip would treat `a/a/b` as a
+        // direct child of `a/`.
+        let Some(ext) = self.name.strip_prefix(dir) else {
+            return false;
+        };
 
-        if ext.len() != self.name.len() {
-            let idx = ext.find(separator);
-            return idx.is_none() || idx.unwrap() == ext.len() - separator.len();
-        }
-
-        false
+        ext.find(separator).is_none_or(|idx| idx == ext.len() - separator.len())
     }
 
     pub fn is_object(&self) -> bool {
@@ -921,12 +920,23 @@ impl MetaCacheEntriesSorted {
         entries
     }
 
+    /// Drop every entry whose name is at or before `marker`.
+    ///
+    /// Entries are sorted by name, so this keeps the suffix starting at the
+    /// first name strictly after the marker; when no such entry exists the
+    /// page is empty. The split index is taken over the raw slots so that
+    /// `None` placeholders cannot shift it.
     pub fn forward_past(&mut self, marker: Option<String>) {
-        if let Some(val) = marker
-            && let Some(idx) = self.o.0.iter().flatten().position(|v| v.name > val)
-        {
-            self.o.0 = self.o.0.split_off(idx);
-        }
+        let Some(val) = marker else {
+            return;
+        };
+        let idx = self
+            .o
+            .0
+            .iter()
+            .position(|v| v.as_ref().is_some_and(|v| v.name > val))
+            .unwrap_or(self.o.0.len());
+        self.o.0 = self.o.0.split_off(idx);
     }
 }
 
@@ -1443,6 +1453,58 @@ mod tests {
     use tokio::sync::{Notify, oneshot};
     use uuid::Uuid;
 
+    fn named_entry(name: &str) -> MetaCacheEntry {
+        MetaCacheEntry {
+            name: name.to_owned(),
+            ..Default::default()
+        }
+    }
+
+    #[test]
+    fn is_in_dir_strips_the_directory_prefix_once() {
+        assert!(named_entry("a/b").is_in_dir("a/", "/"));
+        assert!(named_entry("a/b/").is_in_dir("a/", "/"));
+        assert!(named_entry("a/").is_in_dir("a/", "/"));
+        // `a/a/b` lives in `a/a/`, not directly in `a/`.
+        assert!(!named_entry("a/a/b").is_in_dir("a/", "/"));
+        assert!(!named_entry("b/c").is_in_dir("a/", "/"));
+        assert!(named_entry("top").is_in_dir("", "/"));
+        assert!(named_entry("top/").is_in_dir("", "/"));
+        assert!(!named_entry("top/leaf").is_in_dir("", "/"));
+    }
+
+    #[test]
+    fn forward_past_drops_every_entry_at_or_before_the_marker() {
+        let sorted = |names: &[&str]| MetaCacheEntriesSorted {
+            o: MetaCacheEntries(names.iter().map(|name| Some(named_entry(name))).collect()),
+            ..Default::default()
+        };
+
+        let mut entries = sorted(&["a", "b", "c"]);
+        entries.forward_past(Some("b".to_owned()));
+        assert_eq!(entries.entries().iter().map(|e| e.name.as_str()).collect::<Vec<_>>(), ["c"]);
+
+        let mut entries = sorted(&["a", "b"]);
+        entries.forward_past(Some("b".to_owned()));
+        assert!(entries.entries().is_empty(), "nothing sorts after the marker");
+
+        let mut entries = sorted(&["a", "b"]);
+        entries.forward_past(Some("z".to_owned()));
+        assert!(entries.entries().is_empty(), "a marker past every entry empties the page");
+
+        let mut entries = sorted(&["a", "b"]);
+        entries.forward_past(None);
+        assert_eq!(entries.entries().len(), 2, "no marker keeps the page intact");
+
+        // A `None` slot before the split point must not shift the retained suffix.
+        let mut entries = MetaCacheEntriesSorted {
+            o: MetaCacheEntries(vec![None, Some(named_entry("a")), Some(named_entry("b")), Some(named_entry("c"))]),
+            ..Default::default()
+        };
+        entries.forward_past(Some("a".to_owned()));
+        assert_eq!(entries.entries().iter().map(|e| e.name.as_str()).collect::<Vec<_>>(), ["b", "c"]);
+    }
+
     #[tokio::test]
     async fn test_writer() {
         let mut f = Cursor::new(Vec::new());