From 17aa75a1759e856e93b8c1df2fd447b5ce92e30d Mon Sep 17 00:00:00 2001 From: Chris An Date: Sun, 25 Oct 2026 02:24:41 +0800 Subject: [PATCH] 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 | 65 +++++++++++++++++++++++- crates/filemeta/src/metacache.rs | 86 +++++++++++++++++++++++++++----- 2 files changed, 138 insertions(+), 13 deletions(-) diff --git a/crates/ecstore/src/disk/local.rs b/crates/ecstore/src/disk/local.rs index 84378c787..5a21c3706 100644 --- a/crates/ecstore/src/disk/local.rs +++ b/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| { + 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 diff --git a/crates/filemeta/src/metacache.rs b/crates/filemeta/src/metacache.rs index 14f0846fc..f72bc8071 100644 --- a/crates/filemeta/src/metacache.rs +++ b/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) { - 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::>(), ["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::>(), ["b", "c"]); + } + #[tokio::test] async fn test_writer() { let mut f = Cursor::new(Vec::new());