From 0f69c073de65c412dbf7ffeaf41693f6190de29a Mon Sep 17 00:00:00 2001 From: Hauser Date: Fri, 18 Sep 2026 15:41:44 +0800 Subject: [PATCH] fix(ecstore): preserve bounded listing completion reason (#7994) Co-authored-by: zhi22915 --- .../ecstore/src/cache_value/metacache_set.rs | 6 +- crates/ecstore/src/disk/local.rs | 126 ++++++++++-------- crates/ecstore/src/disk/mod.rs | 14 +- crates/ecstore/src/store/list_objects.rs | 69 +++++++++- 4 files changed, 152 insertions(+), 63 deletions(-) diff --git a/crates/ecstore/src/cache_value/metacache_set.rs b/crates/ecstore/src/cache_value/metacache_set.rs index 74cf741e8..203bcfc42 100644 --- a/crates/ecstore/src/cache_value/metacache_set.rs +++ b/crates/ecstore/src/cache_value/metacache_set.rs @@ -23,7 +23,7 @@ use std::{ future::Future, io::ErrorKind, pin::Pin, - sync::{Arc, OnceLock}, + sync::{Arc, OnceLock, atomic::AtomicBool}, task::{Context, Poll}, time::Duration, }; @@ -234,6 +234,8 @@ pub struct ListPathRawOptions { pub skip_walkdir_total_timeout: bool, pub walkdir_timeout: Option, pub walkdir_stall_timeout: Option, + /// Shared terminal state for bounded producer walks. + pub producer_limit_reached: Option>, pub agreed: Option, pub partial: Option, pub finished: Option, @@ -267,6 +269,7 @@ impl Clone for ListPathRawOptions { skip_walkdir_total_timeout: self.skip_walkdir_total_timeout, walkdir_timeout: self.walkdir_timeout, walkdir_stall_timeout: self.walkdir_stall_timeout, + producer_limit_reached: self.producer_limit_reached.clone(), #[cfg(test)] test_reader_behaviors: self.test_reader_behaviors.clone(), #[cfg(test)] @@ -292,6 +295,7 @@ fn walk_dir_options(opts: &ListPathRawOptions) -> WalkDirOptions { skip_total_timeout: opts.skip_walkdir_total_timeout, timeout_ms: opts.walkdir_timeout.map(duration_millis), stall_timeout_ms: opts.walkdir_stall_timeout.map(duration_millis), + producer_limit_reached: opts.producer_limit_reached.clone(), ..Default::default() } } diff --git a/crates/ecstore/src/disk/local.rs b/crates/ecstore/src/disk/local.rs index 12fce4480..6c279059b 100644 --- a/crates/ecstore/src/disk/local.rs +++ b/crates/ecstore/src/disk/local.rs @@ -7611,7 +7611,7 @@ impl LocalDisk { objs_returned: &mut i32, skip_current_dir_object: bool, multipart_dir_to_skip: Option>, - ) -> Result<()> + ) -> Result where W: AsyncWrite + Unpin + Send, { @@ -7629,7 +7629,7 @@ impl LocalDisk { }; if opts.limit > 0 && *objs_returned >= opts.limit { - return Ok(()); + return Ok(true); } // TODO(backlog): add directory listing lock to prevent concurrent enumeration @@ -7690,12 +7690,12 @@ impl LocalDisk { return Err(DiskError::FileNotFound); } - return Ok(()); + return Ok(false); } }; if entries.is_empty() { - return Ok(()); + return Ok(false); } current = current.trim_matches('/').to_owned(); @@ -7709,7 +7709,7 @@ impl LocalDisk { let entry = item.clone(); // check limit if opts.limit > 0 && *objs_returned >= opts.limit { - return Ok(()); + return Ok(true); } // check multipart dir if skip_current_dir_object @@ -7814,14 +7814,14 @@ impl LocalDisk { prefix = "".to_owned(); for entry in entries.iter() { - if opts.limit > 0 && *objs_returned >= opts.limit { - return Ok(()); - } - if entry.is_empty() { continue; } + if opts.limit > 0 && *objs_returned >= opts.limit { + return Ok(true); + } + let name = path_join_buf(&[current.as_str(), entry.as_str()]); while let Some((last_name, _, _, _)) = dir_stack.last() @@ -7835,7 +7835,7 @@ impl LocalDisk { // 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(()); + return Ok(true); } let (pop, skip_object, dir_to_skip, scan_required) = dir_stack.pop().expect("operation should succeed"); @@ -7849,23 +7849,25 @@ impl LocalDisk { .await?; let scan_path = pop.clone(); - if opts.recursive - && scan_required - && let Err(er) = - Box::pin(self.scan_dir(pop, prefix.clone(), opts, out, objs_returned, skip_object, dir_to_skip)).await - { - if !er.is_metacache_output_stream_closed() { - error!( - event = EVENT_DISK_LOCAL_SCAN_FAILED, - component = LOG_COMPONENT_ECSTORE, - subsystem = LOG_SUBSYSTEM_DISK_LOCAL, - path = %scan_path, - operation = "scan_dir", - error = ?er, - "Disk local scan failed" - ); + if opts.recursive && scan_required { + match Box::pin(self.scan_dir(pop, prefix.clone(), opts, out, objs_returned, skip_object, dir_to_skip)).await { + Ok(true) => return Ok(true), + Ok(false) => {} + Err(er) => { + if !er.is_metacache_output_stream_closed() { + error!( + event = EVENT_DISK_LOCAL_SCAN_FAILED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_DISK_LOCAL, + path = %scan_path, + operation = "scan_dir", + error = ?er, + "Disk local scan failed" + ); + } + return Err(er); + } } - return Err(er); } } @@ -7876,7 +7878,7 @@ impl LocalDisk { // 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(()); + return Ok(true); } let mut meta = MetaCacheEntry { @@ -8019,7 +8021,7 @@ impl LocalDisk { while let Some((dir, skip_object, dir_to_skip, scan_required)) = dir_stack.pop() { if opts.limit > 0 && *objs_returned >= opts.limit { - return Ok(()); + return Ok(true); } write_metacache_obj( @@ -8032,27 +8034,29 @@ impl LocalDisk { .await?; let scan_path = dir.clone(); - if opts.recursive - && scan_required - && let Err(er) = - Box::pin(self.scan_dir(dir, prefix.clone(), opts, out, objs_returned, skip_object, dir_to_skip)).await - { - if !er.is_metacache_output_stream_closed() { - error!( - event = EVENT_DISK_LOCAL_SCAN_FAILED, - component = LOG_COMPONENT_ECSTORE, - subsystem = LOG_SUBSYSTEM_DISK_LOCAL, - path = %scan_path, - operation = "scan_dir", - error = ?er, - "Disk local recursive scan failed" - ); + if opts.recursive && scan_required { + match Box::pin(self.scan_dir(dir, prefix.clone(), opts, out, objs_returned, skip_object, dir_to_skip)).await { + Ok(true) => return Ok(true), + Ok(false) => {} + Err(er) => { + if !er.is_metacache_output_stream_closed() { + error!( + event = EVENT_DISK_LOCAL_SCAN_FAILED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_DISK_LOCAL, + path = %scan_path, + operation = "scan_dir", + error = ?er, + "Disk local recursive scan failed" + ); + } + return Err(er); + } } - return Err(er); } } - Ok(()) + Ok(false) } /// Whether the backing directory of plain object `object_name` also holds @@ -10176,21 +10180,25 @@ impl DiskAPI for LocalDisk { } } - self.scan_dir( - opts.base_dir.clone(), - opts.filter_prefix.clone().unwrap_or_default(), - &opts, - &mut out, - &mut objs_returned, - skip_current_dir_object, - if multipart_dir_to_skip.is_empty() { - None - } else { - Some(multipart_dir_to_skip) - }, - ) - .await?; + let limit_reached = self + .scan_dir( + opts.base_dir.clone(), + opts.filter_prefix.clone().unwrap_or_default(), + &opts, + &mut out, + &mut objs_returned, + skip_current_dir_object, + if multipart_dir_to_skip.is_empty() { + None + } else { + Some(multipart_dir_to_skip) + }, + ) + .await?; + if let Some(flag) = opts.producer_limit_reached.as_ref() { + flag.store(limit_reached, std::sync::atomic::Ordering::Release); + } out.close().await?; Ok(()) } diff --git a/crates/ecstore/src/disk/mod.rs b/crates/ecstore/src/disk/mod.rs index 75f748a3a..dfaae98bb 100644 --- a/crates/ecstore/src/disk/mod.rs +++ b/crates/ecstore/src/disk/mod.rs @@ -70,7 +70,12 @@ use rustfs_filemeta::{FileInfo, ObjectPartInfo, RawFileInfo}; use rustfs_madmin::info_commands::DiskMetrics; use rustfs_rio::ChunkReaderBox; use serde::{Deserialize, Serialize}; -use std::{fmt::Debug, path::PathBuf, sync::Arc, time::Duration}; +use std::{ + fmt::Debug, + path::PathBuf, + sync::{Arc, atomic::AtomicBool}, + time::Duration, +}; use time::OffsetDateTime; use tokio::io::{AsyncRead, AsyncWrite}; use uuid::Uuid; @@ -1479,6 +1484,12 @@ pub struct WalkDirOptions { // Override the remote stream stall timeout for long background walks. #[serde(default)] pub stall_timeout_ms: Option, + + /// In-process completion state for bounded local walks. This is skipped + /// from RPC serialization; remote peers retain the legacy natural-EOF + /// behavior until they support an explicit capability. + #[serde(skip)] + pub producer_limit_reached: Option>, } impl WalkDirOptions { @@ -1802,6 +1813,7 @@ mod tests { skip_total_timeout: false, timeout_ms: Some(10_000), stall_timeout_ms: Some(20_000), + producer_limit_reached: None, }; assert_eq!(opts.bucket, "test-bucket"); diff --git a/crates/ecstore/src/store/list_objects.rs b/crates/ecstore/src/store/list_objects.rs index 9a1de1cb8..9c5342fac 100644 --- a/crates/ecstore/src/store/list_objects.rs +++ b/crates/ecstore/src/store/list_objects.rs @@ -66,7 +66,7 @@ use std::future::Future; use std::path::{Path, PathBuf}; use std::sync::{ Arc, - atomic::{AtomicU64, Ordering}, + atomic::{AtomicBool, AtomicU64, Ordering}, }; use std::time::{Duration, SystemTime, UNIX_EPOCH}; use tokio::io::duplex; @@ -341,6 +341,18 @@ pub struct ListPathOptions { pub cursor_generation: Option, pub walkdir_timeout: Option, pub walkdir_stall_timeout: Option, + /// Shared by the collector and raw producers to preserve bounded-walk + /// completion when filtering removes all entries from a batch. + pub producer_limit_reached: Option>, +} + +fn ensure_producer_limit_state(options: &mut ListPathOptions) -> Arc { + let state = options + .producer_limit_reached + .clone() + .unwrap_or_else(|| Arc::new(AtomicBool::new(false))); + options.producer_limit_reached = Some(state.clone()); + state } async fn can_skip_hidden_prefix_check(options: &ListPathOptions) -> bool { @@ -4304,6 +4316,7 @@ impl ECStore { // cancel channel let cancel = CancellationToken::new(); let _cancel_guard = cancel.clone().drop_guard(); + ensure_producer_limit_state(&mut o); let (err_tx, mut err_rx) = broadcast::channel::>(1); @@ -5047,6 +5060,15 @@ async fn gather_results( } } + // A producer can close its stream after exhausting its own scan budget + // before this collector reaches its output limit. In that case the input + // channel closing is not authoritative EOF: the caller must advertise a + // continuation page even when every remaining entry was filtered out. + let producer_limit_reached = opts + .producer_limit_reached + .as_ref() + .is_some_and(|reached| reached.load(Ordering::Acquire)); + // finish not full, return eof let filtered = scanned_entries.saturating_sub(candidate_entries); if let Some(started) = gather_started { @@ -5083,7 +5105,7 @@ async fn gather_results( o: MetaCacheEntries(entries), ..Default::default() }), - err: Some(rustfs_filemeta::Error::Unexpected), + err: (!producer_limit_reached).then_some(rustfs_filemeta::Error::Unexpected), }) .await .is_err() @@ -5734,6 +5756,7 @@ impl Sets { let (err_tx, mut err_rx) = broadcast::channel::>(1); let (sender, recv) = mpsc::channel(o.limit as usize); + ensure_producer_limit_state(&mut o); let sets = self.clone(); let opts = o.clone(); let cancel_rx1 = cancel.clone(); @@ -6777,6 +6800,7 @@ impl SetDisks { let (err_tx, mut err_rx) = broadcast::channel::>(1); let (sender, recv) = mpsc::channel(o.limit as usize); + ensure_producer_limit_state(&mut o); let set = self.clone(); let opts = o.clone(); let cancel_rx1 = cancel.clone(); @@ -7098,6 +7122,7 @@ impl SetDisks { forward_to: opts.marker, min_disks: raw_min_disks, per_disk_limit: limit, + producer_limit_reached: opts.producer_limit_reached.clone(), // A foreground listing is bounded by lack of drive progress (the walk // stall timeout) and by the page limit, never by how long a healthy // walk takes — a large prefix on slow media is not a fault (#4644). @@ -7328,6 +7353,7 @@ mod test { }; use std::collections::{HashMap, HashSet}; use std::path::PathBuf; + use std::sync::atomic::AtomicBool; use std::sync::{Arc, Mutex}; use std::time::Duration; use tokio::sync::mpsc; @@ -8347,6 +8373,45 @@ mod test { assert!(!cancel.is_cancelled()); } + #[tokio::test] + async fn list_path_gather_results_preserves_bounded_producer_after_filtering() { + let (entry_tx, entry_rx) = mpsc::channel(4); + let (result_tx, mut result_rx) = mpsc::channel(1); + let cancel = CancellationToken::new(); + let producer_limit_reached = Arc::new(AtomicBool::new(true)); + + entry_tx + .send(test_meta_entry("outside-prefix")) + .await + .expect("filtered test entry should be queued"); + drop(entry_tx); + + let handle = tokio::spawn(gather_results( + cancel, + ListPathOptions { + bucket: "bucket".to_owned(), + prefix: "requested/".to_owned(), + limit: 8, + incl_deleted: true, + producer_limit_reached: Some(producer_limit_reached), + ..Default::default() + }, + entry_rx, + result_tx, + )); + + let result = result_rx.recv().await.expect("bounded producer result should be delivered"); + assert!(result.entries.expect("entries should be present").entries().is_empty()); + assert!(result.err.is_none(), "bounded producer must not be reported as EOF"); + assert_eq!( + handle + .await + .expect("gather task should not panic") + .expect("gather should succeed"), + GatherResultsState::InputClosed + ); + } + /// A-1 guard (rustfs/backlog#1306): pin that a *successful* send is never /// misclassified as `ConsumerGone`. With the receiver alive, gather_results /// must return the real drain state and deliver the payload carrying the