From 20c4ea864afe232c4102ff0481e826f907c0fcde Mon Sep 17 00:00:00 2001 From: houseme Date: Fri, 24 Jul 2026 22:05:27 +0800 Subject: [PATCH] fix(ecstore): fix wide-prefix listing stalls and decode downstream logging (#5198) * fix(ecstore): handle list stall and downstream decode logs * docs(ecstore): keep wide-directory stall context for list_dir --- crates/ecstore/src/disk/local.rs | 90 +++++++++++++++++++++++------ crates/ecstore/src/set_disk/read.rs | 42 +++++++++----- 2 files changed, 100 insertions(+), 32 deletions(-) diff --git a/crates/ecstore/src/disk/local.rs b/crates/ecstore/src/disk/local.rs index 3b7064a49..8bb29b940 100644 --- a/crates/ecstore/src/disk/local.rs +++ b/crates/ecstore/src/disk/local.rs @@ -1859,6 +1859,46 @@ where with_walk_stall_deadline(stall, fut).await? } +async fn read_dir_entries_with_walk_stall(path: &Path, count: i32, stall: Option) -> Result> { + let mut entries = with_walk_stall_deadline(stall, fs::read_dir(path)) + .await? + .map_err(to_file_error)?; + let mut names = Vec::new(); + let mut remaining = count; + + loop { + let Some(entry) = with_walk_stall_deadline(stall, entries.next_entry()) + .await? + .map_err(to_file_error)? + else { + break; + }; + let name = entry.file_name().to_string_lossy().to_string(); + + if name.is_empty() || name == "." || name == ".." { + continue; + } + + let file_type = with_walk_stall_deadline(stall, entry.file_type()) + .await? + .map_err(to_file_error)?; + if file_type.is_file() { + names.push(name); + } else if file_type.is_dir() { + names.push(format!("{name}{SLASH_SEPARATOR}")); + } else { + continue; + } + + remaining -= 1; + if remaining == 0 { + break; + } + } + + Ok(names) +} + impl FileCacheReclaimReader { fn new(inner: File, reclaim_offset: u64, reclaim_len: usize, reclaim_on_drop: bool) -> Self { #[cfg(target_os = "macos")] @@ -5163,17 +5203,24 @@ impl LocalDisk { let stall = opts.stall_timeout_duration(); - // `count = -1` enumerates the whole directory in one `list_dir` call, and - // the stall budget bounds that entire enumeration as a single unit. On a - // WIDE, FLAT directory (millions of immediate children) that one readdir - // can exceed `stall` on a healthy disk and fail the whole walk -- see the - // wide-directory stall hazard documented on `list_dir` - // (rustfs/backlog#1216, a #2999 sub-class). Mitigate operationally with a - // larger `RUSTFS_DRIVE_WALKDIR_STALL_TIMEOUT_SECS` or the high-latency - // drive-timeout profile; a streaming readdir rewrite is a separate, - // higher-risk follow-up and is intentionally not done here. + // Keep the existing in-memory sort contract, but bound each directory-entry + // read rather than treating the whole enumeration as one stalled disk + // operation. Object listing keeps using per-entry stall deadlines through + // `read_dir_entries_with_walk_stall` so wide prefixes can still be handled + // as a single logical read in API semantics. let read_dir_started = rustfs_io_metrics::get_stage_metrics_enabled().then(std::time::Instant::now); - let read_dir_result = with_walk_stall_timeout(stall, self.list_dir("", &opts.bucket, ¤t, -1)).await; + let dir_path_abs = self.get_object_path(&opts.bucket, current.trim_start_matches(SLASH_SEPARATOR))?; + let read_dir_result = match read_dir_entries_with_walk_stall(&dir_path_abs, -1, stall).await { + Err(err) if err == Error::FileNotFound && !skip_access_checks(&opts.bucket) => { + let volume_dir = self.get_bucket_path(&opts.bucket)?; + if let Err(access_err) = access(&volume_dir).await { + Err(to_access_error(access_err, DiskError::VolumeAccessDenied).into()) + } else { + Err(err) + } + } + result => result, + }; if let Some(started) = read_dir_started { rustfs_io_metrics::record_list_objects_local_read_dir(rustfs_io_metrics::ListObjectsLocalReadDirObservation { outcome: if read_dir_result.is_ok() { @@ -6590,15 +6637,20 @@ impl DiskAPI for LocalDisk { /// escalate to a quorum failure and surface to the client as a ListObjects /// 500, even though nothing is actually wrong with the drive. /// - /// This is deliberately NOT fixed here by rewriting the one-shot - /// `os::read_dir` into a streaming/batched readdir that would refresh the - /// stall deadline between chunks: that is an architecture-level change with - /// high regression surface (ordering, the `count` contract, quorum merge - /// semantics) and is tracked as a separate follow-up. The supported - /// mitigation for wide-directory deployments today is operational -- raise - /// `RUSTFS_DRIVE_WALKDIR_STALL_TIMEOUT_SECS` or run with the high-latency - /// drive-timeout profile (see `get_drive_walkdir_stall_timeout`), both of - /// which widen the budget without any code change. + /// This path keeps the legacy one-shot call contract and does not attempt + /// per-entry timeout segmentation for compatibility reasons. Wide prefix + /// listing workarounds therefore remain operational: raise + /// `RUSTFS_DRIVE_WALKDIR_STALL_TIMEOUT_SECS` or use the + /// high-latency drive-timeout profile from + /// `get_drive_walkdir_stall_timeout`. + /// + /// This is deliberately NOT fixed here by rewriting this one-shot + /// `os::read_dir` into a streaming/batched readdir with per-chunk timeout + /// refresh: that is an architecture-level change with high regression + /// surface (ordering, the `count` contract, quorum merge semantics) and is + /// tracked as a separate follow-up. Object listing now uses + /// `scan_dir` + `read_dir_entries_with_walk_stall` to keep wide-prefix + /// listing resilient while preserving existing full-enumeration behavior. #[tracing::instrument(level = "trace", skip_all)] async fn list_dir(&self, origvolume: &str, volume: &str, dir_path: &str, count: i32) -> Result> { if !origvolume.is_empty() { diff --git a/crates/ecstore/src/set_disk/read.rs b/crates/ecstore/src/set_disk/read.rs index bc7f37736..b43b9c70f 100644 --- a/crates/ecstore/src/set_disk/read.rs +++ b/crates/ecstore/src/set_disk/read.rs @@ -1050,19 +1050,35 @@ impl SetDisks { if has_err { let reason = classify_disk_error(&de_err); - error!( - bucket, - object, - part_index = current_part, - part_number, - part_offset, - part_length, - bytes_written = written, - stage = GET_STAGE_DECODE, - reason = reason.as_str(), - error = ?de_err, - "Erasure decode failed during GetObject" - ); + if reason == GetObjectFailureReason::DownstreamClosed { + warn!( + bucket, + object, + part_index = current_part, + part_number, + part_offset, + part_length, + bytes_written = written, + stage = GET_STAGE_DECODE, + reason = reason.as_str(), + error = ?de_err, + "GetObject downstream closed during erasure decode" + ); + } else { + error!( + bucket, + object, + part_index = current_part, + part_number, + part_offset, + part_length, + bytes_written = written, + stage = GET_STAGE_DECODE, + reason = reason.as_str(), + error = ?de_err, + "Erasure decode failed during GetObject" + ); + } record_get_object_pipeline_failure(GET_STAGE_DECODE, reason); return Err(de_err.into()); }