fix(ecstore): tolerate completed metacache producers (#4531)

Co-authored-by: Henry Guo <marshawcoco@users.noreply.github.com>
This commit is contained in:
Henry Guo
2026-07-09 04:10:57 +08:00
committed by GitHub
parent c6d054245f
commit 91a23361ee
@@ -21,6 +21,7 @@ use rustfs_filemeta::{MetaCacheEntries, MetaCacheEntry, MetacacheReader, is_io_e
use std::{
collections::{HashSet, VecDeque},
future::Future,
io::ErrorKind,
pin::Pin,
sync::{Arc, OnceLock},
time::Duration,
@@ -87,6 +88,12 @@ fn is_missing_path_error(err: &DiskError) -> bool {
matches!(err, DiskError::FileNotFound | DiskError::FileVersionNotFound | DiskError::VolumeNotFound)
}
fn is_tolerated_producer_completion_error(err: &DiskError) -> bool {
matches!(err, DiskError::DiskOngoingReq)
|| err.is_metacache_output_stream_closed()
|| err.contains_io_error_kind(ErrorKind::BrokenPipe)
}
async fn take_fallback_candidate<T>(fallback_items: &Arc<TokioMutex<VecDeque<T>>>) -> Option<T> {
fallback_items.lock().await.pop_front()
}
@@ -815,7 +822,15 @@ async fn list_path_raw_inner(
match result {
Ok(Ok(())) => {}
Ok(Err(err)) => {
if err.is_metacache_output_stream_closed() {
if is_tolerated_producer_completion_error(&err) {
debug!(
event = EVENT_METACACHE_LISTING,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_METACACHE,
state = "producer_tolerated_after_merge",
error = ?err,
"Metacache producer stopped after merge completed"
);
continue;
}
if is_missing_path_error(&err) {
@@ -918,6 +933,19 @@ mod tests {
assert!(!is_missing_path_error(&DiskError::FileAccessDenied));
}
#[test]
fn tolerated_producer_completion_error_classification_excludes_actionable_failures() {
assert!(is_tolerated_producer_completion_error(&DiskError::DiskOngoingReq));
assert!(is_tolerated_producer_completion_error(&DiskError::metacache_output_stream_closed()));
assert!(is_tolerated_producer_completion_error(&DiskError::Io(std::io::Error::new(
ErrorKind::BrokenPipe,
"reader closed after merge",
))));
assert!(!is_tolerated_producer_completion_error(&DiskError::Timeout));
assert!(!is_tolerated_producer_completion_error(&DiskError::DiskNotFound));
assert!(!is_tolerated_producer_completion_error(&DiskError::FileAccessDenied));
}
#[tokio::test]
async fn fallback_candidates_are_claimed_once_across_producers() {
let mut queue = VecDeque::new();
@@ -1159,6 +1187,26 @@ mod tests {
.expect("closed metacache output after quorum EOF should not fail listing");
}
#[tokio::test]
async fn list_path_raw_tolerates_concurrent_scan_after_quorum_eof() {
let result = list_path_raw(
CancellationToken::new(),
ListPathRawOptions {
disks: vec![None, None, None],
min_disks: 2,
test_reader_behaviors: vec![
TestReaderBehavior::Eof,
TestReaderBehavior::Eof,
TestReaderBehavior::ProducerError(DiskError::DiskOngoingReq),
],
..Default::default()
},
)
.await;
assert!(result.is_ok(), "concurrent scan after quorum EOF should not fail listing");
}
#[tokio::test]
async fn list_path_raw_returns_timeout_when_producer_fails_after_partial_entry() {
let seen = Arc::new(Mutex::new(Vec::new()));