gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-09-22 02:23:45 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(ecstore): preserve bounded listing completion reason (#7994)

Browse Source 
Co-authored-by: zhi22915 <qiuzgang@gmail.com>
This commit is contained in:
Hauser
2026-09-18 15:41:44 +08:00
committed by GitHub
parent 7a5e1efa95
commit 0f69c073de
4 changed files with 152 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/cache_value/metacache_set.rs
+5 -1
View File
@@ -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<Duration>,
pub walkdir_stall_timeout: Option<Duration>,
/// Shared terminal state for bounded producer walks.
pub producer_limit_reached: Option<Arc<AtomicBool>>,
pub agreed: Option<AgreedFn>,
pub partial: Option<PartialFn>,
pub finished: Option<FinishedFn>,
@@ -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()
}
}
crates/ecstore/src/disk/local.rs
+67 -59
View File
@@ -7611,7 +7611,7 @@ impl LocalDisk {
objs_returned: &mut i32,
skip_current_dir_object: bool,
multipart_dir_to_skip: Option<HashSet<String>>,
) -> Result<()>
) -> Result<bool>
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(())
}
crates/ecstore/src/disk/mod.rs
+13 -1
View File
@@ -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<u64>,
/// 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<Arc<AtomicBool>>,
}
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");
crates/ecstore/src/store/list_objects.rs
+67 -2
View File
@@ -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<String>,
pub walkdir_timeout: Option<Duration>,
pub walkdir_stall_timeout: Option<Duration>,
/// 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<Arc<AtomicBool>>,
}
fn ensure_producer_limit_state(options: &mut ListPathOptions) -> Arc<AtomicBool> {
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::<Arc<Error>>(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::<Arc<Error>>(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::<Arc<Error>>(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