// Copyright 2024 RustFS Team // // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. // You may obtain a copy of the License at // // http://www.apache.org/licenses/LICENSE-2.0 // // Unless required by applicable law or agreed to in writing, software // distributed under the License is distributed on an "AS IS" BASIS, // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. // See the License for the specific language governing permissions and // limitations under the License. use crate::disk::disk_store::{get_drive_walkdir_peek_timeout, get_drive_walkdir_stall_timeout}; use crate::disk::error::DiskError; use crate::disk::{self, DiskAPI, DiskStore, WalkDirOptions}; use metrics::counter; use rustfs_filemeta::{MetaCacheEntries, MetaCacheEntry, MetacacheReader, is_io_eof}; use std::{ collections::{HashSet, VecDeque}, future::Future, io::ErrorKind, pin::Pin, sync::{Arc, OnceLock}, task::{Context, Poll}, time::Duration, }; use tokio::io::{AsyncRead, AsyncWrite}; use tokio::sync::Mutex as TokioMutex; use tokio::task::JoinSet; use tokio::time::timeout; use tokio_util::sync::CancellationToken; use tracing::{debug, error, warn}; const LOG_COMPONENT_ECSTORE: &str = "ecstore"; const LOG_SUBSYSTEM_METACACHE: &str = "metacache"; const EVENT_METACACHE_LISTING: &str = "metacache_listing"; pub type AgreedFn = Box Pin + Send>> + Send + 'static>; pub type PartialFn = Box]) -> Pin + Send>> + Send + 'static>; type FinishedFn = Box]) -> Pin + Send>> + Send + 'static>; #[derive(Clone, Default)] pub(crate) struct FallbackClaimTracker { claimed: Arc>>, } impl FallbackClaimTracker { pub(crate) async fn claim_disk(&self, disk: &DiskStore) { self.claimed.lock().await.insert(disk.endpoint().to_string()); } pub(crate) async fn claimed_keys(&self) -> HashSet { self.claimed.lock().await.clone() } #[cfg(test)] pub(crate) async fn claim_test_fallback(&self) { let mut claimed = self.claimed.lock().await; let key = format!("test-fallback-{}", claimed.len()); claimed.insert(key); } #[cfg(test)] pub(crate) async fn contains_key(&self, key: &str) -> bool { self.claimed.lock().await.contains(key) } } #[derive(Debug)] enum PeekOutcome { Ready(Option), Error(rustfs_filemeta::Error), TimedOut, Cancelled, } async fn peek_with_timeout( cancel: &CancellationToken, reader: &mut MetacacheReader, timeout_duration: Duration, ) -> PeekOutcome { tokio::select! { biased; _ = cancel.cancelled() => PeekOutcome::Cancelled, result = timeout(timeout_duration, reader.peek()) => match result { Ok(Ok(entry)) => PeekOutcome::Ready(entry), Ok(Err(err)) => PeekOutcome::Error(err), Err(_) => PeekOutcome::TimedOut, }, } } 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(fallback_items: &Arc>>) -> Option { fallback_items.lock().await.pop_front() } fn duration_millis(duration: Duration) -> u64 { u64::try_from(duration.as_millis()).unwrap_or(u64::MAX) } fn classify_listing_quorum_failure(errors: &[DiskError]) -> DiskError { if errors.contains(&DiskError::Timeout) { return DiskError::Timeout; } if let Some(first) = errors.first() && !matches!(first, DiskError::Io(_)) && errors.iter().all(|err| err == first) { return first.clone(); } DiskError::ErasureReadQuorum } /// Returns true when a metacache listing missed quorum purely because the /// volume or path is absent on a quorum of drives, i.e. every recorded failure /// is [`DiskError::VolumeNotFound`] or [`DiskError::FileNotFound`]. /// /// This is a benign, expected outcome: the caller is handed /// `VolumeNotFound`/`FileNotFound` and decides how to react. The most common /// trigger is a startup race where the system bucket has not yet been created /// on every drive when an early reader (e.g. the IAM config loader) lists it. /// It must be distinguished from a listing that failed for a real reason (I/O, /// timeout, corruption) so the former is not surfaced at `error`. See /// rustfs/rustfs#5076. fn is_benign_not_found_listing_failure(errors: &[DiskError]) -> bool { !errors.is_empty() && errors .iter() .all(|err| matches!(err, DiskError::VolumeNotFound | DiskError::FileNotFound)) } struct PublishedBytesWriter { inner: W, published: bool, } impl PublishedBytesWriter { fn new(inner: W) -> Self { Self { inner, published: false } } fn has_published(&self) -> bool { self.published } } impl AsyncWrite for PublishedBytesWriter { fn poll_write(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &[u8]) -> Poll> { match Pin::new(&mut self.inner).poll_write(cx, buf) { Poll::Ready(Ok(written)) => { if written != 0 && !self.published { self.published = true; } Poll::Ready(Ok(written)) } other => other, } } fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { Pin::new(&mut self.inner).poll_flush(cx) } fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { Pin::new(&mut self.inner).poll_shutdown(cx) } } #[cfg(test)] #[derive(Clone)] pub(crate) enum TestReaderBehavior { Eof, Entries(Vec), Stall, StallWithSignals { entered: Arc, stopped: Arc, }, IgnoreCancel, ProducerError(DiskError), PrimaryErrorThenFallback(DiskError), PartialThenTimeout(Vec), DelayedEntries { delay: Duration, entries: Vec, }, } #[cfg(test)] struct NotifyOnDrop(Arc); #[cfg(test)] impl Drop for NotifyOnDrop { fn drop(&mut self) { self.0.notify_one(); } } #[derive(Default)] pub struct ListPathRawOptions { pub disks: Vec>, pub fallback_disks: Vec>, pub bucket: String, pub path: String, pub recursive: bool, pub incl_deleted: bool, pub filter_prefix: Option, pub forward_to: Option, pub min_disks: usize, pub report_not_found: bool, pub per_disk_limit: i32, pub skip_walkdir_total_timeout: bool, pub walkdir_timeout: Option, pub walkdir_stall_timeout: Option, pub agreed: Option, pub partial: Option, pub finished: Option, #[cfg(test)] pub(crate) test_reader_behaviors: Vec, #[cfg(test)] pub(crate) test_fallback_reader_behaviors: Vec, #[cfg(test)] pub(crate) peek_timeout: Option, // pub agreed: Option>, // pub partial: Option]) + Send + Sync>>, // pub finished: Option]) + Send + Sync>>, } impl Clone for ListPathRawOptions { fn clone(&self) -> Self { Self { disks: self.disks.clone(), fallback_disks: self.fallback_disks.clone(), bucket: self.bucket.clone(), path: self.path.clone(), recursive: self.recursive, incl_deleted: self.incl_deleted, filter_prefix: self.filter_prefix.clone(), forward_to: self.forward_to.clone(), min_disks: self.min_disks, report_not_found: self.report_not_found, per_disk_limit: self.per_disk_limit, skip_walkdir_total_timeout: self.skip_walkdir_total_timeout, walkdir_timeout: self.walkdir_timeout, walkdir_stall_timeout: self.walkdir_stall_timeout, #[cfg(test)] test_reader_behaviors: self.test_reader_behaviors.clone(), #[cfg(test)] test_fallback_reader_behaviors: self.test_fallback_reader_behaviors.clone(), #[cfg(test)] peek_timeout: self.peek_timeout, ..Default::default() } } } fn walk_dir_options(opts: &ListPathRawOptions) -> WalkDirOptions { WalkDirOptions { bucket: opts.bucket.clone(), base_dir: opts.path.clone(), recursive: opts.recursive, incl_deleted: opts.incl_deleted, report_notfound: opts.report_not_found, filter_prefix: opts.filter_prefix.clone(), forward_to: opts.forward_to.clone(), limit: opts.per_disk_limit, 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), ..Default::default() } } pub async fn list_path_raw(rx: CancellationToken, opts: ListPathRawOptions) -> disk::error::Result<()> { let rx = rx.child_token(); let _cancel_guard = rx.clone().drop_guard(); list_path_raw_inner(rx, opts, None).await } pub(crate) async fn list_path_raw_with_claim_tracker( rx: CancellationToken, opts: ListPathRawOptions, claim_tracker: FallbackClaimTracker, ) -> disk::error::Result<()> { let rx = rx.child_token(); let _cancel_guard = rx.clone().drop_guard(); list_path_raw_inner(rx, opts, Some(claim_tracker)).await } async fn list_path_raw_inner( rx: CancellationToken, opts: ListPathRawOptions, fallback_claim_tracker: Option, ) -> disk::error::Result<()> { if opts.disks.is_empty() { return Err(DiskError::ErasureReadQuorum); } if opts.min_disks > opts.disks.len() { return Err(DiskError::ErasureReadQuorum); } let log_bucket = opts.bucket.clone(); let log_path = opts.path.clone(); let mut jobs = JoinSet::new(); let mut readers = Vec::with_capacity(opts.disks.len()); let fds = Arc::new(TokioMutex::new(opts.fallback_disks.iter().flatten().cloned().collect::>())); #[cfg(test)] let test_fallbacks = Arc::new(TokioMutex::new( opts.test_fallback_reader_behaviors.iter().cloned().collect::>(), )); let max_disk_failures = opts.disks.len().saturating_sub(opts.min_disks); let producer_errs: Arc<[OnceLock]> = (0..opts.disks.len()).map(|_| OnceLock::new()).collect::>().into(); let cancel_rx = CancellationToken::new(); for (disk_idx, disk) in opts.disks.iter().enumerate() { let opdisk = disk.clone(); let opts_clone = opts.clone(); let fallback_claim_tracker = fallback_claim_tracker.clone(); let fds_clone = fds.clone(); #[cfg(test)] let test_fallbacks_clone = test_fallbacks.clone(); let cancel_rx_clone = cancel_rx.clone(); let producer_errs_clone = producer_errs.clone(); let (rd, wr) = tokio::io::duplex(64); readers.push(MetacacheReader::new(rd)); jobs.spawn(async move { let mut wr = PublishedBytesWriter::new(wr); #[cfg(test)] let test_primary_error = if let Some(behavior) = opts_clone.test_reader_behaviors.get(disk_idx).cloned() { match behavior { TestReaderBehavior::Eof => return Ok(()), TestReaderBehavior::Entries(entries) => { let mut out = rustfs_filemeta::MetacacheWriter::new(&mut wr); out.write(&entries).await.expect("test entries should be written"); out.close().await.expect("test entries should close"); return Ok(()); } TestReaderBehavior::Stall => { let _held_writer = wr; cancel_rx_clone.cancelled().await; return Ok(()); } TestReaderBehavior::StallWithSignals { entered, stopped } => { let _drop_signal = NotifyOnDrop(stopped); entered.notify_one(); let _held_writer = &mut wr; std::future::pending::>().await } TestReaderBehavior::IgnoreCancel => { let _held_writer = wr; std::future::pending::<()>().await; return Ok(()); } TestReaderBehavior::ProducerError(err) => { record_producer_error(&producer_errs_clone, disk_idx, &err); return Err(err); } TestReaderBehavior::PrimaryErrorThenFallback(err) => Some(err), TestReaderBehavior::PartialThenTimeout(entries) => { let mut out = rustfs_filemeta::MetacacheWriter::new(&mut wr); let err = DiskError::Timeout; let _ = out.write(&entries).await; drop(out); Some(err) } TestReaderBehavior::DelayedEntries { delay, entries } => { tokio::time::sleep(delay).await; let mut out = rustfs_filemeta::MetacacheWriter::new(&mut wr); out.write(&entries).await.expect("delayed test entries should be written"); out.close().await.expect("delayed test entries should close"); return Ok(()); } } } else { None }; let wakl_opts = walk_dir_options(&opts_clone); let mut need_fallback = false; let mut last_err = None; #[cfg(test)] if let Some(err) = test_primary_error { if wr.has_published() { record_producer_error(&producer_errs_clone, disk_idx, &err); return Err(err); } last_err = Some(err); need_fallback = true; } if !need_fallback && let Some(disk) = opdisk { let primary_walk_started = std::time::Instant::now(); match disk.walk_dir(wakl_opts, &mut wr).await { Ok(_res) => { rustfs_io_metrics::record_stage_duration( "metacache_walk_dir_primary", primary_walk_started.elapsed().as_secs_f64() * 1000.0, ); } Err(err) => { if err.is_metacache_output_stream_closed() { rustfs_io_metrics::record_stage_duration( "metacache_walk_dir_primary", primary_walk_started.elapsed().as_secs_f64() * 1000.0, ); return Ok(()); } rustfs_io_metrics::record_stage_duration( "metacache_walk_dir_primary_failed", primary_walk_started.elapsed().as_secs_f64() * 1000.0, ); if is_missing_path_error(&err) { debug!( event = EVENT_METACACHE_LISTING, component = LOG_COMPONENT_ECSTORE, subsystem = LOG_SUBSYSTEM_METACACHE, bucket = %opts_clone.bucket, path = %opts_clone.path, disk_index = disk_idx, state = "walk_dir_missing_path", error = ?err, "Metacache walk_dir missing path skipped" ); } else { warn!( event = EVENT_METACACHE_LISTING, component = LOG_COMPONENT_ECSTORE, subsystem = LOG_SUBSYSTEM_METACACHE, bucket = %opts_clone.bucket, path = %opts_clone.path, disk_index = disk_idx, state = "walk_dir_failed", error = ?err, "Metacache walk_dir failed" ); } if wr.has_published() { record_producer_error(&producer_errs_clone, disk_idx, &err); return Err(err); } last_err = Some(err); need_fallback = true; } } } else if !need_fallback { last_err = Some(DiskError::DiskNotFound); need_fallback = true; } if cancel_rx_clone.is_cancelled() { // warn!("list_path_raw: cancel_rx_clone.is_cancelled()"); return Ok(()); } while need_fallback { #[cfg(test)] if let Some(behavior) = take_fallback_candidate(&test_fallbacks_clone).await { if let Some(claim_tracker) = fallback_claim_tracker.as_ref() { claim_tracker.claim_test_fallback().await; } match behavior { TestReaderBehavior::Eof => { need_fallback = false; last_err = None; continue; } TestReaderBehavior::Entries(entries) => { let mut out = rustfs_filemeta::MetacacheWriter::new(&mut wr); out.write(&entries).await.expect("test fallback entries should be written"); out.close().await.expect("test fallback entries should close"); need_fallback = false; last_err = None; continue; } TestReaderBehavior::ProducerError(err) | TestReaderBehavior::PrimaryErrorThenFallback(err) => { last_err = Some(err); continue; } TestReaderBehavior::Stall | TestReaderBehavior::StallWithSignals { .. } | TestReaderBehavior::IgnoreCancel => { last_err = Some(DiskError::Timeout); continue; } TestReaderBehavior::PartialThenTimeout(entries) => { let mut out = rustfs_filemeta::MetacacheWriter::new(&mut wr); let err = DiskError::Timeout; let _ = out.write(&entries).await; drop(out); record_producer_error(&producer_errs_clone, disk_idx, &err); return Err(err); } TestReaderBehavior::DelayedEntries { delay, entries } => { tokio::time::sleep(delay).await; let mut out = rustfs_filemeta::MetacacheWriter::new(&mut wr); out.write(&entries) .await .expect("delayed test fallback entries should be written"); out.close().await.expect("delayed test fallback entries should close"); need_fallback = false; last_err = None; continue; } } } let mut disk_op = None; while let Some(disk) = take_fallback_candidate(&fds_clone).await { if disk.is_online().await { disk_op = Some(disk); break; } } let Some(disk) = disk_op else { let err = last_err.unwrap_or(DiskError::DiskNotFound); if is_missing_path_error(&err) { debug!( event = EVENT_METACACHE_LISTING, component = LOG_COMPONENT_ECSTORE, subsystem = LOG_SUBSYSTEM_METACACHE, bucket = %opts_clone.bucket, path = %opts_clone.path, disk_index = disk_idx, state = "fallback_disk_missing_for_path", error = ?err, "Metacache fallback disk unavailable for missing path" ); } else { warn!( event = EVENT_METACACHE_LISTING, component = LOG_COMPONENT_ECSTORE, subsystem = LOG_SUBSYSTEM_METACACHE, bucket = %opts_clone.bucket, path = %opts_clone.path, disk_index = disk_idx, state = "fallback_disk_missing", error = ?err, "Metacache fallback disk missing" ); } record_producer_error(&producer_errs_clone, disk_idx, &err); return Err(err); }; if let Some(claim_tracker) = fallback_claim_tracker.as_ref() { claim_tracker.claim_disk(&disk).await; } let fallback_walk_started = std::time::Instant::now(); match disk.as_ref().walk_dir(walk_dir_options(&opts_clone), &mut wr).await { Ok(_r) => { rustfs_io_metrics::record_stage_duration( "metacache_walk_dir_fallback", fallback_walk_started.elapsed().as_secs_f64() * 1000.0, ); need_fallback = false; last_err = None; } Err(err) => { if err.is_metacache_output_stream_closed() { rustfs_io_metrics::record_stage_duration( "metacache_walk_dir_fallback", fallback_walk_started.elapsed().as_secs_f64() * 1000.0, ); return Ok(()); } rustfs_io_metrics::record_stage_duration( "metacache_walk_dir_fallback_failed", fallback_walk_started.elapsed().as_secs_f64() * 1000.0, ); if is_missing_path_error(&err) { debug!( event = EVENT_METACACHE_LISTING, component = LOG_COMPONENT_ECSTORE, subsystem = LOG_SUBSYSTEM_METACACHE, bucket = %opts_clone.bucket, path = %opts_clone.path, disk_index = disk_idx, state = "fallback_walk_dir_missing_path", error = ?err, "Metacache fallback walk_dir missing path skipped" ); } else { error!( event = EVENT_METACACHE_LISTING, component = LOG_COMPONENT_ECSTORE, subsystem = LOG_SUBSYSTEM_METACACHE, bucket = %opts_clone.bucket, path = %opts_clone.path, disk_index = disk_idx, state = "fallback_walk_dir_failed", error = ?err, "Metacache fallback walk_dir failed" ); } if wr.has_published() { record_producer_error(&producer_errs_clone, disk_idx, &err); return Err(err); } last_err = Some(err); } } } if need_fallback { return Err(last_err.unwrap_or(DiskError::DiskNotFound)); } // warn!("list_path_raw: while need_fallback done"); Ok(()) }); } let revjob_rx = rx.clone(); let mut revjobs = JoinSet::new(); revjobs.spawn(async move { // Consumer-side peek timeout, and a caveat worth understanding // (rustfs/backlog#1217). // // This budget must not be stricter than the producer-side walk stall // budget. The two measure different things: // * producer stall: bounds a single drive READ inside the walk (see // `with_walk_stall_deadline` in `disk/local.rs`); // * this consumer peek: bounds the interval between two ADJACENT // entries arriving from a drive's reader (`peek_with_timeout`). // // A drive can spend one stall budget walking a dense non-listable region // before it can publish the next visible entry. Use an independent, // profile-aware consumer budget so the merge does not detach that reader // before the producer itself would have failed. let producer_stall_timeout = opts.walkdir_stall_timeout.unwrap_or_else(get_drive_walkdir_stall_timeout); let configured_peek_timeout = get_drive_walkdir_peek_timeout().max(producer_stall_timeout); #[cfg(not(test))] let peek_timeout = configured_peek_timeout; #[cfg(test)] let peek_timeout = opts.peek_timeout.unwrap_or(configured_peek_timeout); let mut errs: Vec> = Vec::with_capacity(readers.len()); for _ in 0..readers.len() { errs.push(None); } let mut pending_entries: Vec> = vec![None; readers.len()]; loop { let mut current = MetaCacheEntry::default(); // warn!( // "list_path_raw: loop start, bucket: {}, path: {}, current: {:?}", // opts.bucket, opts.path, ¤t.name // ); if revjob_rx.is_cancelled() { return Ok(()); } let mut top_entries: Vec> = vec![None; readers.len()]; let mut at_eof = 0; let mut fnf = 0; let mut vnf = 0; let mut has_err = 0; let mut agree = 0; for (i, r) in readers.iter_mut().enumerate() { if errs[i].is_some() { has_err += 1; continue; } let entry = if let Some(entry) = pending_entries[i].take() { entry } else { match peek_with_timeout(&revjob_rx, r, peek_timeout).await { PeekOutcome::Ready(res) => { if let Some(entry) = res { // info!("read entry disk: {}, name: {}", i, entry.name); entry } else { if let Some(err) = producer_error(&producer_errs, i) { has_err += 1; errs[i] = Some(err); continue; } // eof at_eof += 1; // warn!("list_path_raw: peek eof, disk: {}", i); continue; } } PeekOutcome::Error(err) => { if let Some(err) = producer_error(&producer_errs, i) { has_err += 1; errs[i] = Some(err); continue; } if err == rustfs_filemeta::Error::Unexpected { at_eof += 1; // warn!("list_path_raw: peek err eof, disk: {}", i); continue; } // warn!("list_path_raw: peek err00, err: {:?}", err); if is_io_eof(&err) { at_eof += 1; // warn!("list_path_raw: peek eof, disk: {}", i); continue; } if err == rustfs_filemeta::Error::FileNotFound { at_eof += 1; fnf += 1; // warn!("list_path_raw: peek fnf, disk: {}", i); continue; } else if err == rustfs_filemeta::Error::VolumeNotFound { at_eof += 1; fnf += 1; vnf += 1; // warn!("list_path_raw: peek vnf, disk: {}", i); continue; } else { has_err += 1; errs[i] = Some(err.into()); // warn!("list_path_raw: peek err, disk: {}", i); continue; } } PeekOutcome::TimedOut => { has_err += 1; errs[i] = Some(DiskError::Timeout); let endpoint = opts .disks .get(i) .and_then(|disk| disk.as_ref().map(|disk| disk.endpoint().to_string())) .unwrap_or_else(|| "missing".to_string()); counter!( "rustfs_list_path_raw_stall_total", "drive" => endpoint.clone() ) .increment(1); warn!( event = EVENT_METACACHE_LISTING, component = LOG_COMPONENT_ECSTORE, subsystem = LOG_SUBSYSTEM_METACACHE, drive = %endpoint, bucket = %opts.bucket, path = %opts.path, timeout_ms = peek_timeout.as_millis(), state = "peek_timed_out", "Metacache reader peek timed out" ); let (detached_rd, write_half) = tokio::io::duplex(1); drop(write_half); *r = MetacacheReader::new(detached_rd); continue; } PeekOutcome::Cancelled => return Ok(()), } }; // warn!("list_path_raw: loop entry: {:?}, disk: {}", &entry.name, i); // If no current, add it. if current.name.is_empty() { top_entries[i] = Some(entry.clone()); current = entry; agree += 1; continue; } // If exact match, we agree. if let (_, true) = current.matches(Some(&entry), true) { top_entries[i] = Some(entry); agree += 1; continue; } // If only the name matches we didn't agree, but add it for resolution. if entry.name == current.name { top_entries[i] = Some(entry); continue; } // We got different entries if entry.name > current.name { pending_entries[i] = Some(entry); continue; } for (idx, item) in top_entries.iter_mut().enumerate().take(i) { if let Some(entry) = item.take() { pending_entries[idx] = Some(entry); } } agree = 1; top_entries[i] = Some(entry.clone()); current = entry; } if vnf > 0 && vnf >= (readers.len() - opts.min_disks) { // warn!("list_path_raw: vnf > 0 && vnf >= (readers.len() - opts.min_disks) break"); return Err(DiskError::VolumeNotFound); } if fnf > 0 && fnf >= (readers.len() - opts.min_disks) { // warn!("list_path_raw: fnf > 0 && fnf >= (readers.len() - opts.min_disks) break"); return Err(DiskError::FileNotFound); } if has_err > 0 && has_err > opts.disks.len() - opts.min_disks { if let Some(finished_fn) = opts.finished.as_ref() { tokio::select! { biased; _ = revjob_rx.cancelled() => return Ok(()), _ = finished_fn(&errs) => {} } } let mut combined_err = Vec::new(); errs.iter().zip(opts.disks.iter()).for_each(|(err, disk)| match (err, disk) { (Some(err), Some(disk)) => { combined_err.push(format!("drive {} returned: {}", disk.to_string(), err)); } (Some(err), None) => { combined_err.push(err.to_string()); } _ => {} }); let failures = errs.iter().flatten().cloned().collect::>(); // A listing that misses quorum purely because the volume/path is // absent on a quorum of drives is benign and expected — the caller // receives VolumeNotFound/FileNotFound and decides how to react. // The common trigger is a startup race where the system bucket is // not yet created on every drive when an early reader (e.g. the IAM // config loader) lists it, so surfacing it at `error` is misleading // noise (rustfs/rustfs#5076). Keep `error` for listings that failed // for a real reason (I/O, timeout, corruption). if is_benign_not_found_listing_failure(&failures) { debug!( event = EVENT_METACACHE_LISTING, component = LOG_COMPONENT_ECSTORE, subsystem = LOG_SUBSYSTEM_METACACHE, bucket = %opts.bucket, path = %opts.path, state = "quorum_not_found", error = %combined_err.join(", "), "Metacache listing quorum not reached (volume/path absent)" ); } else { error!( event = EVENT_METACACHE_LISTING, component = LOG_COMPONENT_ECSTORE, subsystem = LOG_SUBSYSTEM_METACACHE, bucket = %opts.bucket, path = %opts.path, state = "quorum_failed", error = %combined_err.join(", "), "Metacache listing quorum failed" ); } return Err(classify_listing_quorum_failure(&failures)); } // Break if all at EOF or error. if at_eof + has_err == readers.len() { if has_err > 0 && let Some(finished_fn) = opts.finished.as_ref() { tokio::select! { biased; _ = revjob_rx.cancelled() => return Ok(()), _ = finished_fn(&errs) => {} } } // Tolerated reader failures, including timeouts, must not turn a // quorum EOF into a failed listing. The quorum failure branch above // is the single place that escalates too many reader errors. // error!("list_path_raw: at_eof + has_err == readers.len() break {:?}", &errs); break; } if agree == readers.len() { for r in readers.iter_mut() { let _ = r.skip(1).await; } if let Some(agreed_fn) = opts.agreed.as_ref() { // warn!("list_path_raw: agreed_fn start, current: {:?}", ¤t.name); tokio::select! { biased; _ = revjob_rx.cancelled() => return Ok(()), _ = agreed_fn(current) => {} } // warn!("list_path_raw: agreed_fn done"); } continue; } // warn!("list_path_raw: skip start, current: {:?}", ¤t.name); for (i, r) in readers.iter_mut().enumerate() { if top_entries[i].is_some() { let _ = r.skip(1).await; } } if let Some(partial_fn) = opts.partial.as_ref() { tokio::select! { biased; _ = revjob_rx.cancelled() => return Ok(()), _ = partial_fn(MetaCacheEntries(top_entries), &errs) => {} } } } Ok(()) }); let merge_started = std::time::Instant::now(); let Some(revjob) = revjobs.join_next().await else { return Err(DiskError::Unexpected); }; if let Err(err) = revjob.map_err(std::io::Error::other)? { rustfs_io_metrics::record_stage_duration("metacache_merge_failed", merge_started.elapsed().as_secs_f64() * 1000.0); if is_missing_path_error(&err) { debug!( event = EVENT_METACACHE_LISTING, component = LOG_COMPONENT_ECSTORE, subsystem = LOG_SUBSYSTEM_METACACHE, bucket = %log_bucket, path = %log_path, state = "merge_job_missing_path", error = ?err, "Metacache merge job missing path skipped" ); } else { error!( event = EVENT_METACACHE_LISTING, component = LOG_COMPONENT_ECSTORE, subsystem = LOG_SUBSYSTEM_METACACHE, bucket = %log_bucket, path = %log_path, state = "merge_job_failed", error = ?err, "Metacache merge job failed" ); } cancel_rx.cancel(); jobs.abort_all(); return Err(err); } if rx.is_cancelled() { cancel_rx.cancel(); jobs.abort_all(); return Ok(()); } rustfs_io_metrics::record_stage_duration("metacache_merge", merge_started.elapsed().as_secs_f64() * 1000.0); // The merge consumer can finish successfully before every producer finishes // (for example after reaching EOF quorum while a tolerated drive is stalled, // or after the requested listing limit is satisfied). Cancel remaining walk // jobs before aborting them so list calls do not wait for slow remote streams. cancel_rx.cancel(); jobs.abort_all(); let mut job_errs = Vec::new(); while let Some(result) = jobs.join_next().await { match result { Ok(Ok(())) => {} Ok(Err(err)) => { 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) { debug!( event = EVENT_METACACHE_LISTING, component = LOG_COMPONENT_ECSTORE, subsystem = LOG_SUBSYSTEM_METACACHE, state = "producer_missing_path", error = ?err, "Metacache producer missing path" ); } else { error!( event = EVENT_METACACHE_LISTING, component = LOG_COMPONENT_ECSTORE, subsystem = LOG_SUBSYSTEM_METACACHE, state = "producer_failed", error = ?err, "Metacache producer failed" ); } job_errs.push(err); } Err(err) => { if err.is_cancelled() { continue; } error!( event = EVENT_METACACHE_LISTING, component = LOG_COMPONENT_ECSTORE, subsystem = LOG_SUBSYSTEM_METACACHE, state = "producer_join_failed", error = ?err, "Metacache producer join failed" ); job_errs.push(err.into()); } } } if job_errs.len() > max_disk_failures { return Err(classify_listing_quorum_failure(&job_errs)); } // warn!("list_path_raw: done"); Ok(()) } #[inline] fn record_producer_error(producer_errs: &[OnceLock], idx: usize, err: &DiskError) { let _ = producer_errs[idx].set(err.clone()); } #[inline] fn producer_error(producer_errs: &[OnceLock], idx: usize) -> Option { producer_errs[idx].get().cloned() } #[cfg(test)] mod tests { use super::*; use rustfs_filemeta::MetacacheWriter; use rustfs_filemeta::{FileInfo, FileMeta, MetadataResolutionParams}; use std::sync::Mutex; use time::OffsetDateTime; use uuid::Uuid; #[test] fn benign_not_found_listing_failure_detection() { // Pure not-found quorum misses are benign (the volume/path simply does // not exist on a quorum of drives) and must not be logged at ERROR. assert!(is_benign_not_found_listing_failure(&[DiskError::VolumeNotFound])); assert!(is_benign_not_found_listing_failure( &[DiskError::VolumeNotFound, DiskError::FileNotFound,] )); // No recorded failure is not a not-found case. assert!(!is_benign_not_found_listing_failure(&[])); // Any real error must keep the failure at ERROR severity. assert!(!is_benign_not_found_listing_failure(&[DiskError::VolumeNotFound, DiskError::Timeout,])); assert!(!is_benign_not_found_listing_failure(&[DiskError::DiskNotFound])); } #[test] fn walk_dir_options_preserve_zero_total_and_inherited_stall_timeouts() { let options = walk_dir_options(&ListPathRawOptions { walkdir_timeout: Some(Duration::ZERO), walkdir_stall_timeout: None, ..Default::default() }); assert_eq!(options.timeout_ms, Some(0)); assert_eq!(options.stall_timeout_ms, None); assert!(!options.skip_total_timeout); } #[tokio::test] async fn list_path_raw_empty_disks_returns_read_quorum() { let err = list_path_raw(CancellationToken::new(), ListPathRawOptions::default()) .await .expect_err("empty drive list should fail"); assert_eq!(err, DiskError::ErasureReadQuorum); } #[tokio::test] async fn list_path_raw_rejects_impossible_min_disks() { let err = list_path_raw( CancellationToken::new(), ListPathRawOptions { disks: vec![None, None], min_disks: 3, ..Default::default() }, ) .await .expect_err("impossible listing quorum should fail before producing partial results"); assert_eq!(err, DiskError::ErasureReadQuorum); } #[tokio::test] async fn list_path_raw_returns_typed_quorum_error_for_multiple_drive_failures() { let err = list_path_raw( CancellationToken::new(), ListPathRawOptions { disks: vec![None, None, None], min_disks: 2, test_reader_behaviors: vec![ TestReaderBehavior::ProducerError(DiskError::other("/sensitive/disk-a")), TestReaderBehavior::ProducerError(DiskError::other("http://internal-node/disk-b")), TestReaderBehavior::Eof, ], ..Default::default() }, ) .await .expect_err("multiple producer failures beyond tolerance should fail listing quorum"); assert_eq!(err, DiskError::ErasureReadQuorum); assert!(!err.to_string().contains("sensitive")); assert!(!err.to_string().contains("internal-node")); } #[test] fn listing_quorum_failure_never_returns_raw_io_details() { let err = classify_listing_quorum_failure(&[DiskError::other("/sensitive/disk-a")]); assert_eq!(err, DiskError::ErasureReadQuorum); assert!(!err.to_string().contains("sensitive")); let access_denied = classify_listing_quorum_failure(&[DiskError::FileAccessDenied, DiskError::FileAccessDenied]); assert_eq!(access_denied, DiskError::FileAccessDenied); let mixed_timeout = classify_listing_quorum_failure(&[DiskError::Timeout, DiskError::FileAccessDenied, DiskError::ErasureReadQuorum]); assert_eq!(mixed_timeout, DiskError::Timeout); } #[test] fn missing_path_error_classification_excludes_actionable_failures() { assert!(is_missing_path_error(&DiskError::FileNotFound)); assert!(is_missing_path_error(&DiskError::FileVersionNotFound)); assert!(is_missing_path_error(&DiskError::VolumeNotFound)); assert!(!is_missing_path_error(&DiskError::Timeout)); assert!(!is_missing_path_error(&DiskError::DiskNotFound)); 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(); queue.push_back(1usize); let candidates = Arc::new(TokioMutex::new(queue)); let first = take_fallback_candidate(&candidates).await; let second = take_fallback_candidate(&candidates).await; assert_eq!(first, Some(1)); assert_eq!(second, None); } fn fallback_test_entry() -> MetaCacheEntry { MetaCacheEntry { name: "bucket/object".to_string(), metadata: vec![1, 2, 3], cached: None, reusable: false, } } #[tokio::test] async fn list_path_raw_does_not_reuse_one_fallback_for_multiple_failed_producers() { let err = list_path_raw( CancellationToken::new(), ListPathRawOptions { disks: vec![None, None], min_disks: 2, test_reader_behaviors: vec![ TestReaderBehavior::PrimaryErrorThenFallback(DiskError::DiskNotFound), TestReaderBehavior::PrimaryErrorThenFallback(DiskError::DiskNotFound), ], test_fallback_reader_behaviors: vec![TestReaderBehavior::Entries(vec![fallback_test_entry()])], ..Default::default() }, ) .await .expect_err("one fallback must not be counted as two failed primary producers"); assert_eq!(err, DiskError::DiskNotFound); } #[tokio::test] async fn list_path_raw_uses_distinct_fallbacks_to_restore_quorum() { let seen = Arc::new(Mutex::new(Vec::new())); let seen_clone = seen.clone(); list_path_raw( CancellationToken::new(), ListPathRawOptions { disks: vec![None, None], min_disks: 2, test_reader_behaviors: vec![ TestReaderBehavior::PrimaryErrorThenFallback(DiskError::DiskNotFound), TestReaderBehavior::PrimaryErrorThenFallback(DiskError::DiskNotFound), ], test_fallback_reader_behaviors: vec![ TestReaderBehavior::Entries(vec![fallback_test_entry()]), TestReaderBehavior::Entries(vec![fallback_test_entry()]), ], partial: Some(Box::new(move |entries: MetaCacheEntries, _: &[Option]| { let seen = seen_clone.clone(); Box::pin(async move { let matches = entries .0 .iter() .flatten() .filter(|entry| entry.name == "bucket/object") .count(); seen.lock().expect("seen mutex poisoned").push(matches); }) })), ..Default::default() }, ) .await .expect("distinct fallback producers should restore listing quorum"); assert_eq!(seen.lock().expect("seen mutex poisoned").as_slice(), &[2]); } #[tokio::test] async fn list_path_raw_records_claimed_fallback_candidates() { let claim_tracker = FallbackClaimTracker::default(); list_path_raw_with_claim_tracker( CancellationToken::new(), ListPathRawOptions { disks: vec![None], min_disks: 1, test_reader_behaviors: vec![TestReaderBehavior::PrimaryErrorThenFallback(DiskError::DiskNotFound)], test_fallback_reader_behaviors: vec![TestReaderBehavior::Entries(vec![fallback_test_entry()])], ..Default::default() }, claim_tracker.clone(), ) .await .expect("fallback producer should restore the single logical reader"); assert!(claim_tracker.contains_key("test-fallback-0").await); } #[tokio::test] async fn list_path_raw_returns_timeout_when_reader_stalls_before_completion() { let err = list_path_raw( CancellationToken::new(), ListPathRawOptions { disks: vec![None, None], min_disks: 2, test_reader_behaviors: vec![TestReaderBehavior::Stall, TestReaderBehavior::Eof], peek_timeout: Some(Duration::from_millis(20)), ..Default::default() }, ) .await .expect_err("stalled reader should fail when read quorum cannot be met"); assert_eq!(err, DiskError::Timeout); } #[tokio::test] async fn list_path_raw_waits_past_producer_stall_for_slow_progressing_reader() { let entry = MetaCacheEntry { name: "bucket/visible-object".to_string(), metadata: vec![1, 2, 3], cached: None, reusable: false, }; let seen = Arc::new(Mutex::new(Vec::new())); let seen_clone = seen.clone(); timeout( Duration::from_secs(1), list_path_raw( CancellationToken::new(), ListPathRawOptions { disks: vec![None, None], min_disks: 2, walkdir_stall_timeout: Some(Duration::from_millis(20)), test_reader_behaviors: vec![ TestReaderBehavior::DelayedEntries { delay: Duration::from_millis(60), entries: vec![entry.clone()], }, TestReaderBehavior::Entries(vec![entry]), ], partial: Some(Box::new(move |entries: MetaCacheEntries, _: &[Option]| { let seen = seen_clone.clone(); Box::pin(async move { let mut names = entries.0.iter().flatten().map(|entry| entry.name.as_str()); if let (Some(first), Some(second), None) = (names.next(), names.next(), names.next()) && first == second { seen.lock().expect("seen mutex poisoned").push(first.to_owned()); } }) })), ..Default::default() }, ), ) .await .expect("slow-progressing reader should complete inside the consumer peek budget") .expect("slow-progressing reader should not be detached by the producer stall budget"); assert_eq!( seen.lock().expect("seen mutex poisoned").as_slice(), &["bucket/visible-object".to_string()] ); } #[tokio::test] async fn list_path_raw_prefers_timeout_for_mixed_errors_beyond_quorum_budget() { let err = list_path_raw( CancellationToken::new(), ListPathRawOptions { disks: vec![None, None, None], min_disks: 2, test_reader_behaviors: vec![ TestReaderBehavior::ProducerError(DiskError::FileAccessDenied), TestReaderBehavior::ProducerError(DiskError::Timeout), TestReaderBehavior::Eof, ], ..Default::default() }, ) .await .expect_err("mixed failures beyond the tolerated budget should fail listing"); assert_eq!(err, DiskError::Timeout); } #[tokio::test] async fn list_path_raw_tolerates_stalled_reader_after_quorum_eof() { list_path_raw( CancellationToken::new(), ListPathRawOptions { disks: vec![None, None, None], min_disks: 2, test_reader_behaviors: vec![TestReaderBehavior::Eof, TestReaderBehavior::Eof, TestReaderBehavior::Stall], peek_timeout: Some(Duration::from_millis(20)), ..Default::default() }, ) .await .expect("stalled reader within the tolerated failure budget should not fail quorum EOF"); } #[tokio::test] async fn list_path_raw_completes_after_partial_quorum_when_reader_stalls() { let entry = MetaCacheEntry { name: "bucket/object".to_string(), metadata: vec![1, 2, 3], cached: None, reusable: false, }; let seen = Arc::new(Mutex::new(Vec::new())); let seen_clone = seen.clone(); list_path_raw( CancellationToken::new(), ListPathRawOptions { disks: vec![None, None, None], min_disks: 2, test_reader_behaviors: vec![ TestReaderBehavior::Entries(vec![entry.clone()]), TestReaderBehavior::Entries(vec![entry]), TestReaderBehavior::Stall, ], peek_timeout: Some(Duration::from_millis(20)), partial: Some(Box::new(move |entries: MetaCacheEntries, _: &[Option]| { let seen = seen_clone.clone(); Box::pin(async move { let mut names = entries.0.iter().flatten().map(|entry| entry.name.clone()); if let Some(name) = names.next() && names.any(|next| next == name) { seen.lock().expect("seen mutex poisoned").push(name); } }) })), ..Default::default() }, ) .await .expect("stalled reader within failure budget should not fail a quorum listing"); assert_eq!(seen.lock().expect("seen mutex poisoned").as_slice(), &["bucket/object".to_string()]); } #[tokio::test] async fn list_path_raw_aborts_unresponsive_producer_after_quorum_eof() { let result = timeout( Duration::from_millis(200), list_path_raw( CancellationToken::new(), ListPathRawOptions { disks: vec![None, None, None], min_disks: 2, test_reader_behaviors: vec![ TestReaderBehavior::Eof, TestReaderBehavior::Eof, TestReaderBehavior::IgnoreCancel, ], peek_timeout: Some(Duration::from_millis(20)), ..Default::default() }, ), ) .await; let listing = result.expect("list_path_raw should abort unresponsive producer instead of hanging"); listing.expect("unresponsive producer within failure budget should not fail quorum EOF"); } #[tokio::test] async fn list_path_raw_treats_external_cancel_as_successful_shutdown() { let cancel = CancellationToken::new(); cancel.cancel(); list_path_raw( cancel, ListPathRawOptions { disks: vec![None], min_disks: 1, test_reader_behaviors: vec![TestReaderBehavior::IgnoreCancel], ..Default::default() }, ) .await .expect("external cancellation should stop listing without a synthetic canceled error"); } #[tokio::test] async fn list_path_raw_cancels_a_blocked_peek_without_waiting_for_the_stall_timeout() { let cancel = CancellationToken::new(); let entered = Arc::new(tokio::sync::Notify::new()); let stopped = Arc::new(tokio::sync::Notify::new()); let list = tokio::spawn(list_path_raw( cancel.clone(), ListPathRawOptions { disks: vec![None], min_disks: 1, test_reader_behaviors: vec![TestReaderBehavior::StallWithSignals { entered: entered.clone(), stopped: stopped.clone(), }], peek_timeout: Some(Duration::from_secs(60)), ..Default::default() }, )); timeout(Duration::from_secs(1), entered.notified()) .await .expect("producer should block before cancellation"); cancel.cancel(); let result = timeout(Duration::from_secs(1), list) .await .expect("cancellation should interrupt the blocked metacache peek") .expect("listing task should not panic"); assert!(result.is_ok(), "cancellation should stop the listing cleanly, got {result:?}"); timeout(Duration::from_secs(1), stopped.notified()) .await .expect("cancellation should drop the blocked producer"); } #[tokio::test] async fn aborting_list_path_raw_drops_blocked_producers() { let entered = Arc::new(tokio::sync::Notify::new()); let stopped = Arc::new(tokio::sync::Notify::new()); let list = tokio::spawn(list_path_raw( CancellationToken::new(), ListPathRawOptions { disks: vec![None], min_disks: 1, test_reader_behaviors: vec![TestReaderBehavior::StallWithSignals { entered: entered.clone(), stopped: stopped.clone(), }], peek_timeout: Some(Duration::from_secs(60)), ..Default::default() }, )); timeout(Duration::from_secs(1), entered.notified()) .await .expect("producer should block before aborting the listing"); list.abort(); assert!(list.await.expect_err("listing task should be cancelled").is_cancelled()); timeout(Duration::from_secs(1), stopped.notified()) .await .expect("aborting the listing should drop the blocked producer"); } #[tokio::test] async fn list_path_raw_cancels_a_blocked_partial_callback() { let cancel = CancellationToken::new(); let callback_started = Arc::new(tokio::sync::Notify::new()); let callback_started_clone = callback_started.clone(); let entry = MetaCacheEntry { name: "bucket/object-a".to_string(), metadata: vec![1], ..Default::default() }; let later_entry = MetaCacheEntry { name: "bucket/object-b".to_string(), metadata: vec![1], ..Default::default() }; let list = tokio::spawn(list_path_raw( cancel.clone(), ListPathRawOptions { disks: vec![None, None], min_disks: 2, test_reader_behaviors: vec![ TestReaderBehavior::Entries(vec![entry]), TestReaderBehavior::Entries(vec![later_entry]), ], partial: Some(Box::new(move |_, _| { let callback_started = callback_started_clone.clone(); Box::pin(async move { callback_started.notify_one(); std::future::pending::<()>().await; }) })), ..Default::default() }, )); timeout(Duration::from_secs(1), callback_started.notified()) .await .expect("partial callback should block before cancellation"); cancel.cancel(); let result = timeout(Duration::from_secs(1), list) .await .expect("cancellation should interrupt the blocked partial callback") .expect("listing task should not panic"); assert!(result.is_ok(), "cancellation should stop the listing cleanly, got {result:?}"); } #[tokio::test] async fn list_path_raw_cancels_a_blocked_agreed_callback() { let cancel = CancellationToken::new(); let callback_started = Arc::new(tokio::sync::Notify::new()); let callback_started_clone = callback_started.clone(); let entry = MetaCacheEntry { name: "bucket/object".to_string(), metadata: vec![1], ..Default::default() }; let list = tokio::spawn(list_path_raw( cancel.clone(), ListPathRawOptions { disks: vec![None], min_disks: 1, test_reader_behaviors: vec![TestReaderBehavior::Entries(vec![entry])], agreed: Some(Box::new(move |_| { let callback_started = callback_started_clone.clone(); Box::pin(async move { callback_started.notify_one(); std::future::pending::<()>().await; }) })), ..Default::default() }, )); timeout(Duration::from_secs(1), callback_started.notified()) .await .expect("agreed callback should block before cancellation"); cancel.cancel(); let result = timeout(Duration::from_secs(1), list) .await .expect("cancellation should interrupt the blocked agreed callback") .expect("listing task should not panic"); assert!(result.is_ok(), "cancellation should stop the listing cleanly, got {result:?}"); } #[tokio::test] async fn list_path_raw_cancels_a_blocked_finished_callback_after_quorum_failure() { let cancel = CancellationToken::new(); let callback_started = Arc::new(tokio::sync::Notify::new()); let callback_started_clone = callback_started.clone(); let list = tokio::spawn(list_path_raw( cancel.clone(), ListPathRawOptions { disks: vec![None], min_disks: 1, test_reader_behaviors: vec![TestReaderBehavior::ProducerError(DiskError::FileAccessDenied)], finished: Some(Box::new(move |_| { let callback_started = callback_started_clone.clone(); Box::pin(async move { callback_started.notify_one(); std::future::pending::<()>().await; }) })), ..Default::default() }, )); timeout(Duration::from_secs(1), callback_started.notified()) .await .expect("finished callback should block before cancellation"); cancel.cancel(); let result = timeout(Duration::from_secs(1), list) .await .expect("cancellation should interrupt the blocked finished callback") .expect("listing task should not panic"); assert!(result.is_ok(), "cancellation should stop the listing cleanly, got {result:?}"); } #[tokio::test] async fn list_path_raw_cancels_a_blocked_finished_callback_after_tolerated_error() { let cancel = CancellationToken::new(); let callback_started = Arc::new(tokio::sync::Notify::new()); let callback_started_clone = callback_started.clone(); let list = tokio::spawn(list_path_raw( cancel.clone(), ListPathRawOptions { disks: vec![None, None, None], min_disks: 2, test_reader_behaviors: vec![ TestReaderBehavior::Eof, TestReaderBehavior::Eof, TestReaderBehavior::ProducerError(DiskError::FileAccessDenied), ], finished: Some(Box::new(move |_| { let callback_started = callback_started_clone.clone(); Box::pin(async move { callback_started.notify_one(); std::future::pending::<()>().await; }) })), ..Default::default() }, )); timeout(Duration::from_secs(1), callback_started.notified()) .await .expect("finished callback should block before cancellation"); cancel.cancel(); let result = timeout(Duration::from_secs(1), list) .await .expect("cancellation should interrupt the blocked finished callback") .expect("listing task should not panic"); assert!(result.is_ok(), "cancellation should stop the listing cleanly, got {result:?}"); } #[tokio::test] async fn list_path_raw_ignores_closed_output_stream_after_quorum_eof() { 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::metacache_output_stream_closed()), ], ..Default::default() }, ) .await .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())); let seen_clone = seen.clone(); let claim_tracker = FallbackClaimTracker::default(); let err = list_path_raw_with_claim_tracker( CancellationToken::new(), ListPathRawOptions { disks: vec![None], min_disks: 1, test_reader_behaviors: vec![TestReaderBehavior::PartialThenTimeout(vec![MetaCacheEntry { name: "bucket/object".to_string(), metadata: vec![1, 2, 3], cached: None, reusable: false, }])], test_fallback_reader_behaviors: vec![TestReaderBehavior::Entries(vec![fallback_test_entry()])], agreed: Some(Box::new(move |entry: MetaCacheEntry| { let seen = seen_clone.clone(); Box::pin(async move { seen.lock().expect("seen mutex poisoned").push(entry.name); }) })), ..Default::default() }, claim_tracker.clone(), ) .await .expect_err("producer timeout after partial output must fail the listing"); assert_eq!(err, DiskError::Timeout); assert_eq!(seen.lock().expect("seen mutex poisoned").as_slice(), &["bucket/object".to_string()]); assert!( claim_tracker.claimed_keys().await.is_empty(), "fallback must not append a second metacache stream after partial output" ); } #[tokio::test] async fn list_path_raw_does_not_try_second_fallback_after_partial_fallback_output() { let seen = Arc::new(Mutex::new(Vec::new())); let seen_clone = seen.clone(); let claim_tracker = FallbackClaimTracker::default(); let err = list_path_raw_with_claim_tracker( CancellationToken::new(), ListPathRawOptions { disks: vec![None], min_disks: 1, test_reader_behaviors: vec![TestReaderBehavior::PrimaryErrorThenFallback(DiskError::DiskNotFound)], test_fallback_reader_behaviors: vec![ TestReaderBehavior::PartialThenTimeout(vec![MetaCacheEntry { name: "bucket/partial-fallback".to_string(), metadata: vec![1, 2, 3], cached: None, reusable: false, }]), TestReaderBehavior::Entries(vec![fallback_test_entry()]), ], agreed: Some(Box::new(move |entry: MetaCacheEntry| { let seen = seen_clone.clone(); Box::pin(async move { seen.lock().expect("seen mutex poisoned").push(entry.name); }) })), ..Default::default() }, claim_tracker.clone(), ) .await .expect_err("a fallback timeout after partial output must fail the listing"); assert_eq!(err, DiskError::Timeout); assert_eq!( seen.lock().expect("seen mutex poisoned").as_slice(), &["bucket/partial-fallback".to_string()] ); assert_eq!(claim_tracker.claimed_keys().await.len(), 1, "a second fallback must not be claimed"); } #[tokio::test] async fn list_path_raw_continues_after_single_disk_delete_marker_gap() { fn object_entry(name: &str, version_id: &str) -> MetaCacheEntry { let mut meta = FileMeta::default(); let mut fi = FileInfo::new(name, 1, 1); fi.version_id = Some(Uuid::parse_str(version_id).expect("test version id should parse")); fi.mod_time = Some(OffsetDateTime::now_utc()); meta.add_version(fi).expect("object metadata should be valid"); MetaCacheEntry { name: name.to_string(), metadata: meta.marshal_msg().expect("object metadata should encode"), cached: Some(meta), reusable: false, } } fn delete_marker_entry(name: &str, version_id: &str) -> MetaCacheEntry { let mut meta = FileMeta::default(); meta.add_version(FileInfo { deleted: true, version_id: Some(Uuid::parse_str(version_id).expect("test version id should parse")), mod_time: Some(OffsetDateTime::now_utc()), ..Default::default() }) .expect("delete marker metadata should be valid"); MetaCacheEntry { name: name.to_string(), metadata: meta.marshal_msg().expect("delete marker metadata should encode"), cached: Some(meta), reusable: false, } } let visible_before = object_entry("object-000003", "11111111-1111-1111-1111-111111111111"); let hidden_delete = delete_marker_entry("object-000004", "22222222-2222-2222-2222-222222222222"); let visible_after = object_entry("object-000005", "33333333-3333-3333-3333-333333333333"); let seen = Arc::new(Mutex::new(Vec::new())); let agreed_seen = seen.clone(); let partial_seen = seen.clone(); list_path_raw( CancellationToken::new(), ListPathRawOptions { disks: vec![None, None, None, None], min_disks: 2, test_reader_behaviors: vec![ TestReaderBehavior::Entries(vec![visible_before.clone(), hidden_delete, visible_after.clone()]), TestReaderBehavior::Entries(vec![visible_before.clone(), visible_after.clone()]), TestReaderBehavior::Entries(vec![visible_before.clone(), visible_after.clone()]), TestReaderBehavior::Entries(vec![visible_before, visible_after]), ], agreed: Some(Box::new(move |entry| { let seen = agreed_seen.clone(); Box::pin(async move { seen.lock().expect("seen mutex poisoned").push(entry.name); }) })), partial: Some(Box::new(move |entries, _| { let seen = partial_seen.clone(); Box::pin(async move { if let Some(entry) = entries.resolve(MetadataResolutionParams { obj_quorum: 2, requested_versions: 1, bucket: "bucket".to_string(), ..Default::default() }) { seen.lock().expect("seen mutex poisoned").push(entry.name); } }) })), ..Default::default() }, ) .await .expect("single-disk delete marker gap should not end listing"); assert_eq!( seen.lock().expect("seen mutex poisoned").as_slice(), &["object-000003".to_string(), "object-000005".to_string()] ); } #[tokio::test] async fn peek_with_timeout_times_out_on_silent_reader() { let (_writer, reader) = tokio::io::duplex(64); let mut reader = MetacacheReader::new(reader); let cancel = CancellationToken::new(); let outcome = peek_with_timeout(&cancel, &mut reader, Duration::from_millis(20)).await; assert!(matches!(outcome, PeekOutcome::TimedOut)); } #[tokio::test] async fn peek_with_timeout_stops_immediately_when_cancelled() { struct PendingReader(Arc); impl AsyncRead for PendingReader { fn poll_read( self: std::pin::Pin<&mut Self>, _cx: &mut std::task::Context<'_>, _buf: &mut tokio::io::ReadBuf<'_>, ) -> std::task::Poll> { self.get_mut().0.notify_one(); std::task::Poll::Pending } } let entered = Arc::new(tokio::sync::Notify::new()); let mut reader = MetacacheReader::new(PendingReader(entered.clone())); let cancel = CancellationToken::new(); let wait_cancel = cancel.clone(); let wait = tokio::spawn(async move { peek_with_timeout(&wait_cancel, &mut reader, Duration::from_secs(60)).await }); timeout(Duration::from_secs(1), entered.notified()) .await .expect("peek should block before cancellation"); cancel.cancel(); let outcome = timeout(Duration::from_secs(1), wait) .await .expect("cancellation should interrupt the pending peek") .expect("peek task should not panic"); assert!(matches!(outcome, PeekOutcome::Cancelled)); } #[tokio::test] async fn peek_with_timeout_reads_entry_before_deadline() { let (reader, writer) = tokio::io::duplex(256); let mut metacache_reader = MetacacheReader::new(reader); let cancel = CancellationToken::new(); tokio::spawn(async move { let mut writer = MetacacheWriter::new(writer); let entry = MetaCacheEntry { name: "bucket/object".to_string(), metadata: vec![1, 2, 3], cached: None, reusable: false, }; writer.write(&[entry]).await.expect("entry should be written"); writer.close().await.expect("writer should close"); }); let outcome = peek_with_timeout(&cancel, &mut metacache_reader, Duration::from_secs(1)).await; match outcome { PeekOutcome::Ready(Some(entry)) => assert_eq!(entry.name, "bucket/object"), other => panic!("expected ready entry, got {other:?}"), } } #[tokio::test] async fn list_path_raw_propagates_producer_access_denied() { let err = list_path_raw( CancellationToken::new(), ListPathRawOptions { disks: vec![None], min_disks: 1, test_reader_behaviors: vec![TestReaderBehavior::ProducerError(DiskError::FileAccessDenied)], ..Default::default() }, ) .await .expect_err("producer access failure must not be treated as an empty listing"); assert_eq!(err, DiskError::FileAccessDenied); } }