diff --git a/crates/ecstore/src/cache_value/metacache_set.rs b/crates/ecstore/src/cache_value/metacache_set.rs index 56fb0e489..ce120301f 100644 --- a/crates/ecstore/src/cache_value/metacache_set.rs +++ b/crates/ecstore/src/cache_value/metacache_set.rs @@ -280,6 +280,13 @@ async fn list_path_raw_inner( ); } 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, @@ -427,6 +434,13 @@ async fn list_path_raw_inner( 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, @@ -490,7 +504,7 @@ async fn list_path_raw_inner( // ); if rx.is_cancelled() { - return Err(DiskError::other("canceled")); + return Ok(()); } let mut top_entries: Vec> = vec![None; readers.len()]; @@ -780,6 +794,9 @@ async fn list_path_raw_inner( match result { Ok(Ok(())) => {} Ok(Err(err)) => { + if err.is_metacache_output_stream_closed() { + continue; + } if is_missing_path_error(&err) { debug!( event = EVENT_METACACHE_LISTING, @@ -1084,6 +1101,43 @@ mod tests { 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_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_returns_timeout_when_producer_fails_after_partial_entry() { let seen = Arc::new(Mutex::new(Vec::new())); diff --git a/crates/ecstore/src/disk/error.rs b/crates/ecstore/src/disk/error.rs index 630802ac3..606ec7456 100644 --- a/crates/ecstore/src/disk/error.rs +++ b/crates/ecstore/src/disk/error.rs @@ -13,6 +13,7 @@ // limitations under the License. use rustfs_rio::{InternodeHttpError, InternodeHttpErrorKind}; +use std::error::Error as StdError; use std::hash::{Hash, Hasher}; use std::io::{self}; use std::path::PathBuf; @@ -20,6 +21,8 @@ use std::path::PathBuf; pub type Error = DiskError; pub type Result = core::result::Result; +const METACACHE_OUTPUT_STREAM_CLOSED: &str = "metacache output stream closed"; + // DiskError == StorageErr #[derive(Debug, thiserror::Error)] pub enum DiskError { @@ -158,6 +161,26 @@ impl DiskError { DiskError::Io(std::io::Error::other(error)) } + pub(crate) fn metacache_output_stream_closed() -> Self { + DiskError::Io(std::io::Error::new(std::io::ErrorKind::BrokenPipe, METACACHE_OUTPUT_STREAM_CLOSED)) + } + + pub(crate) fn is_metacache_output_stream_closed(&self) -> bool { + matches!( + self, + DiskError::Io(io_error) + if io_error.kind() == std::io::ErrorKind::BrokenPipe + && io_error.to_string() == METACACHE_OUTPUT_STREAM_CLOSED + ) + } + + pub(crate) fn contains_io_error_kind(&self, kind: std::io::ErrorKind) -> bool { + match self { + DiskError::Io(io_error) => io_error_chain_contains_kind(io_error, kind), + _ => false, + } + } + pub fn is_all_not_found(errs: &[Option]) -> bool { for err in errs.iter() { if let Some(err) = err { @@ -250,7 +273,7 @@ impl DiskError { impl From for DiskError { fn from(e: rustfs_filemeta::Error) -> Self { match e { - rustfs_filemeta::Error::Io(e) => DiskError::other(e), + rustfs_filemeta::Error::Io(e) => DiskError::Io(e), rustfs_filemeta::Error::FileNotFound => DiskError::FileNotFound, rustfs_filemeta::Error::FileVersionNotFound => DiskError::FileVersionNotFound, rustfs_filemeta::Error::FileCorrupt => DiskError::FileCorrupt, @@ -260,6 +283,29 @@ impl From for DiskError { } } +fn io_error_chain_contains_kind(io_error: &std::io::Error, kind: std::io::ErrorKind) -> bool { + if io_error.kind() == kind { + return true; + } + + let mut source = StdError::source(io_error); + while let Some(err) = source { + if let Some(io_error) = err.downcast_ref::() + && io_error.kind() == kind + { + return true; + } + if let Some(disk_error) = err.downcast_ref::() + && disk_error.contains_io_error_kind(kind) + { + return true; + } + source = err.source(); + } + + false +} + impl From for DiskError { fn from(e: std::io::Error) -> Self { match e.downcast::() { @@ -900,6 +946,25 @@ mod tests { assert!(converted_io.to_string().contains("broken pipe")); } + #[test] + fn test_wrapped_io_error_kind_detection() { + let filemeta_error = rustfs_filemeta::Error::Io(std::io::Error::new(std::io::ErrorKind::BrokenPipe, "broken pipe")); + let disk_error = DiskError::from(filemeta_error); + + assert!(disk_error.contains_io_error_kind(std::io::ErrorKind::BrokenPipe)); + assert!(!disk_error.contains_io_error_kind(std::io::ErrorKind::TimedOut)); + } + + #[test] + fn test_metacache_output_stream_closed_classification_survives_clone() { + let disk_error = DiskError::metacache_output_stream_closed(); + let cloned_error = disk_error.clone(); + + assert!(disk_error.is_metacache_output_stream_closed()); + assert!(cloned_error.is_metacache_output_stream_closed()); + assert_eq!(disk_error, cloned_error); + } + #[test] fn test_error_display_preservation() { let disk_errors = vec![ diff --git a/crates/ecstore/src/disk/local.rs b/crates/ecstore/src/disk/local.rs index 2ffc0f8c0..bebaf1ecb 100644 --- a/crates/ecstore/src/disk/local.rs +++ b/crates/ecstore/src/disk/local.rs @@ -834,6 +834,22 @@ fn is_bitrot_size_mismatch_error(err: &std::io::Error) -> bool { err.to_string().contains("bitrot shard file size mismatch") } +fn metacache_write_error(err: rustfs_filemeta::Error) -> DiskError { + let err = DiskError::from(err); + if err.contains_io_error_kind(ErrorKind::BrokenPipe) { + DiskError::metacache_output_stream_closed() + } else { + err + } +} + +async fn write_metacache_obj(out: &mut MetacacheWriter, obj: &MetaCacheEntry) -> Result<()> +where + W: AsyncWrite + Unpin, +{ + out.write_obj(obj).await.map_err(metacache_write_error) +} + impl FileCacheReclaimReader { fn new(inner: File, reclaim_offset: u64, reclaim_len: usize, reclaim_on_drop: bool) -> Self { #[cfg(target_os = "macos")] @@ -3005,11 +3021,14 @@ impl LocalDisk { *objs_returned += 1; } - out.write_obj(&MetaCacheEntry { - name: name.clone(), - metadata: metadata.to_vec(), - ..Default::default() - }) + write_metacache_obj( + out, + &MetaCacheEntry { + name: name.clone(), + metadata: metadata.to_vec(), + ..Default::default() + }, + ) .await?; continue; @@ -3065,10 +3084,13 @@ impl LocalDisk { && *last_name < name { let (pop, skip_object, dir_to_skip) = dir_stack.pop().expect("operation should succeed"); - out.write_obj(&MetaCacheEntry { - name: pop.clone(), - ..Default::default() - }) + write_metacache_obj( + out, + &MetaCacheEntry { + name: pop.clone(), + ..Default::default() + }, + ) .await?; let scan_path = pop.clone(); @@ -3076,15 +3098,17 @@ impl LocalDisk { && let Err(er) = Box::pin(self.scan_dir(pop, prefix.clone(), opts, out, objs_returned, skip_object, dir_to_skip)).await { - 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 !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); } } @@ -3114,7 +3138,7 @@ impl LocalDisk { meta.metadata = res.to_vec(); - out.write_obj(&meta).await?; + write_metacache_obj(out, &meta).await?; let file_meta = if opts.limit > 0 || opts.recursive { FileMeta::load(&res).ok() @@ -3183,10 +3207,13 @@ impl LocalDisk { return Ok(()); } - out.write_obj(&MetaCacheEntry { - name: dir.clone(), - ..Default::default() - }) + write_metacache_obj( + out, + &MetaCacheEntry { + name: dir.clone(), + ..Default::default() + }, + ) .await?; let scan_path = dir.clone(); @@ -3194,15 +3221,17 @@ impl LocalDisk { && let Err(er) = Box::pin(self.scan_dir(dir, prefix.clone(), opts, out, objs_returned, skip_object, dir_to_skip)).await { - 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 !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); } } @@ -4201,7 +4230,7 @@ impl DiskAPI for LocalDisk { metadata: data.to_vec(), ..Default::default() }; - out.write_obj(&meta).await?; + write_metacache_obj(&mut out, &meta).await?; objs_returned += 1; } else { let fpath = @@ -6284,6 +6313,49 @@ mod test { assert!(!wait.await.expect("operation should succeed")); } + #[test] + fn metacache_write_obj_classifies_closed_output_stream() { + struct BrokenPipeWriter; + + impl AsyncWrite for BrokenPipeWriter { + fn poll_write( + self: std::pin::Pin<&mut Self>, + _cx: &mut std::task::Context<'_>, + _buf: &[u8], + ) -> std::task::Poll> { + std::task::Poll::Ready(Err(std::io::Error::new(std::io::ErrorKind::BrokenPipe, "closed"))) + } + + fn poll_flush( + self: std::pin::Pin<&mut Self>, + _cx: &mut std::task::Context<'_>, + ) -> std::task::Poll> { + std::task::Poll::Ready(Ok(())) + } + + fn poll_shutdown( + self: std::pin::Pin<&mut Self>, + _cx: &mut std::task::Context<'_>, + ) -> std::task::Poll> { + std::task::Poll::Ready(Ok(())) + } + } + + let mut writer = BrokenPipeWriter; + let mut out = MetacacheWriter::new(&mut writer); + + let err = futures::executor::block_on(write_metacache_obj( + &mut out, + &MetaCacheEntry { + name: "object".to_string(), + ..Default::default() + }, + )) + .expect_err("closed metacache output stream should fail"); + + assert!(err.is_metacache_output_stream_closed()); + } + #[tokio::test] async fn test_scan_dir_includes_nested_object_dirs() { use rustfs_filemeta::MetacacheReader; diff --git a/rustfs/src/app/bucket_usecase.rs b/rustfs/src/app/bucket_usecase.rs index a4fa3b78e..9040a2919 100644 --- a/rustfs/src/app/bucket_usecase.rs +++ b/rustfs/src/app/bucket_usecase.rs @@ -1766,14 +1766,21 @@ impl DefaultBucketUsecase { let replication_configuration = match metadata_sys::get_replication_config(&bucket).await { Ok((cfg, _created)) => cfg, + Err(StorageError::ConfigNotFound) => { + return Err(S3Error::with_message( + S3ErrorCode::ReplicationConfigurationNotFoundError, + "replication not found".to_string(), + )); + } Err(err) => { - error!("get_replication_config err {:?}", err); - if err == StorageError::ConfigNotFound { - return Err(S3Error::with_message( - S3ErrorCode::ReplicationConfigurationNotFoundError, - "replication not found".to_string(), - )); - } + warn!( + component = LOG_COMPONENT_APP, + subsystem = LOG_SUBSYSTEM_BUCKET, + event = "bucket_replication_config_load_failed", + bucket = %bucket, + error = ?err, + "Failed to load bucket replication configuration" + ); return Err(ApiError::from(err).into()); } };