diff --git a/crates/ecstore/src/diagnostics/get.rs b/crates/ecstore/src/diagnostics/get.rs index b72b16cfb..cde5a5536 100644 --- a/crates/ecstore/src/diagnostics/get.rs +++ b/crates/ecstore/src/diagnostics/get.rs @@ -14,8 +14,8 @@ use crate::disk::error::DiskError; use crate::error::StorageError; -use std::io; use std::time::Instant; +use std::{error::Error as StdError, fmt, io}; pub(crate) const GET_OBJECT_PATH_CODEC_STREAMING: &str = "codec_streaming"; pub(crate) const GET_OBJECT_PATH_CODEC_STREAMING_LEGACY_ENGINE: &str = "codec_streaming_legacy_engine"; @@ -207,9 +207,42 @@ pub(crate) fn classify_disk_error(err: &DiskError) -> GetObjectFailureReason { } } -pub(crate) fn classify_io_error(err: &io::Error) -> GetObjectFailureReason { +#[derive(Debug)] +struct GetObjectDownstreamClosed { + source: io::Error, +} + +impl fmt::Display for GetObjectDownstreamClosed { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "GetObject output closed: {}", self.source) + } +} + +impl StdError for GetObjectDownstreamClosed { + fn source(&self) -> Option<&(dyn StdError + 'static)> { + Some(&self.source) + } +} + +pub(crate) fn mark_get_object_downstream_closed(err: io::Error) -> io::Error { + match err.kind() { + io::ErrorKind::BrokenPipe | io::ErrorKind::ConnectionReset | io::ErrorKind::ConnectionAborted => { + io::Error::new(err.kind(), GetObjectDownstreamClosed { source: err }) + } + _ => err, + } +} + +pub(crate) fn classify_io_error(err: &io::Error) -> GetObjectFailureReason { + let mut source = err.get_ref().map(|source| source as &(dyn StdError + 'static)); + while let Some(error) = source { + if error.is::() { + return GetObjectFailureReason::DownstreamClosed; + } + source = error.source(); + } + match err.kind() { - io::ErrorKind::BrokenPipe | io::ErrorKind::ConnectionReset => GetObjectFailureReason::DownstreamClosed, io::ErrorKind::TimedOut => GetObjectFailureReason::Timeout, io::ErrorKind::UnexpectedEof => GetObjectFailureReason::ShortRead, io::ErrorKind::InvalidInput | io::ErrorKind::InvalidData => GetObjectFailureReason::RangeOrLengthInvalid, @@ -260,6 +293,16 @@ mod tests { classify_storage_error(&StorageError::InvalidRangeSpec("bad range".to_string())), GetObjectFailureReason::RangeOrLengthInvalid ); + + let internal_broken_pipe = StorageError::Io(io::Error::from(io::ErrorKind::BrokenPipe)); + assert_eq!(classify_storage_error(&internal_broken_pipe), GetObjectFailureReason::Io); + + let downstream_closed = StorageError::Io(mark_get_object_downstream_closed(io::Error::from(io::ErrorKind::BrokenPipe))); + assert_eq!( + classify_storage_error(&downstream_closed), + GetObjectFailureReason::DownstreamClosed, + "the legacy duplex producer must suppress only explicitly tagged output closes" + ); } #[test] @@ -277,8 +320,8 @@ mod tests { #[test] fn classifies_io_errors_for_get_pipeline() { let cases = [ - (io::ErrorKind::BrokenPipe, GetObjectFailureReason::DownstreamClosed), - (io::ErrorKind::ConnectionReset, GetObjectFailureReason::DownstreamClosed), + (io::ErrorKind::BrokenPipe, GetObjectFailureReason::Io), + (io::ErrorKind::ConnectionReset, GetObjectFailureReason::Io), (io::ErrorKind::TimedOut, GetObjectFailureReason::Timeout), (io::ErrorKind::UnexpectedEof, GetObjectFailureReason::ShortRead), (io::ErrorKind::InvalidInput, GetObjectFailureReason::RangeOrLengthInvalid), @@ -292,6 +335,18 @@ mod tests { } } + #[test] + fn classifies_marked_output_close_as_downstream_closed() { + for kind in [ + io::ErrorKind::BrokenPipe, + io::ErrorKind::ConnectionReset, + io::ErrorKind::ConnectionAborted, + ] { + let err = mark_get_object_downstream_closed(io::Error::from(kind)); + assert_eq!(classify_io_error(&err), GetObjectFailureReason::DownstreamClosed, "kind={kind:?}"); + } + } + #[test] fn keeps_metric_labels_stable() { assert_eq!(GetObjectFailureReason::ReadQuorum.as_str(), "read_quorum"); diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index 8ba420268..47de63c8c 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/crates/ecstore/src/set_disk/mod.rs @@ -4446,6 +4446,7 @@ mod tests { use crate::layout::endpoints::SetupType; use crate::object_api::BLOCK_SIZE_V2; use crate::object_api::ObjectInfo; + use crate::set_disk::read::GetObjectOutputKind; use crate::storage_api_contracts::{ heal::HealOperations as _, lifecycle::TransitionedObject, list::ListOperations as _, multipart::CompletePart, namespace::NamespaceLocking as _, object::ObjectIO as _, object::ObjectOperations as _, @@ -8400,6 +8401,7 @@ mod tests { range_offset, range_length as i64, &mut writer, + GetObjectOutputKind::HttpResponse, fi, files, &disks, @@ -8511,6 +8513,7 @@ mod tests { 0, total_size as i64, &mut writer, + GetObjectOutputKind::HttpResponse, fi, files, &disks, diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index 5136cde92..8eb4b590f 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/crates/ecstore/src/set_disk/ops/object.rs @@ -21,6 +21,7 @@ use super::super::*; use super::bitrot_self_verify::{BitrotSelfVerifyTarget, drop_failed_writer_disks, verify_written_bitrot_shards}; +use crate::set_disk::read::GetObjectOutputKind; use crate::bucket::lifecycle::{ tier_delete_journal::{persist_tier_delete_journal_entry, remove_tier_delete_journal_entry}, @@ -31,6 +32,7 @@ use crate::bucket::lifecycle::{ save_transition_transaction_record, }, }; +use crate::diagnostics::get::GetObjectFailureReason; use crate::disk::OldCurrentSize; use crate::object_api::{GetObjectBodySource, get_object_body_cache_hook_suppressed}; use crate::services::tier::tier::{TierConfigMgr, TierOperationLease}; @@ -522,6 +524,7 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks { 0, object_info.size, &mut output, + GetObjectOutputKind::Internal, fi, files, &disks, @@ -650,6 +653,7 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks { offset, length, &mut writer, + GetObjectOutputKind::HttpResponse, fi, files, &disks, @@ -664,24 +668,43 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks { .await { let reason = classify_storage_error(&e); - record_get_object_pipeline_failure(GET_STAGE_EMIT, reason); - error!( - event = EVENT_SET_DISK_WRITE, - component = LOG_COMPONENT_ECSTORE, - subsystem = LOG_SUBSYSTEM_SET_DISK, - bucket, - object, - pool_index, - set_index, - offset, - requested_length = length, - skip_verify_bitrot = skip_verify, - state = "read_pipeline_failed", - stage = GET_STAGE_EMIT, - reason = reason.as_str(), - error = ?e, - "Set disk object read pipeline failed" - ); + if reason == GetObjectFailureReason::DownstreamClosed { + debug!( + event = EVENT_SET_DISK_WRITE, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_SET_DISK, + bucket, + object, + pool_index, + set_index, + offset, + requested_length = length, + state = "downstream_closed", + stage = GET_STAGE_EMIT, + reason = reason.as_str(), + error = ?e, + "Set disk object read pipeline stopped after downstream closed" + ); + } else { + record_get_object_pipeline_failure(GET_STAGE_EMIT, reason); + error!( + event = EVENT_SET_DISK_WRITE, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_SET_DISK, + bucket, + object, + pool_index, + set_index, + offset, + requested_length = length, + skip_verify_bitrot = skip_verify, + state = "read_pipeline_failed", + stage = GET_STAGE_EMIT, + reason = reason.as_str(), + error = ?e, + "Set disk object read pipeline failed" + ); + } }; }); @@ -3356,6 +3379,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { 0, cloned_fi.size, &mut writer, + GetObjectOutputKind::Internal, cloned_fi, meta_arr, &online_disks, diff --git a/crates/ecstore/src/set_disk/read.rs b/crates/ecstore/src/set_disk/read.rs index b43b9c70f..b0e2bab5a 100644 --- a/crates/ecstore/src/set_disk/read.rs +++ b/crates/ecstore/src/set_disk/read.rs @@ -34,8 +34,8 @@ use crate::diagnostics::get::{ GET_STAGE_METADATA_RESOLVE, GET_STAGE_RANGE, GET_STAGE_READER_SETUP, GET_STAGE_READER_SETUP_DROP_PENDING, GET_STAGE_READER_SETUP_SCHEDULE, GET_STAGE_READER_SETUP_WAIT_QUORUM, GET_STAGE_READER_TASK_BITROT_READER_INIT, GET_STAGE_READER_TASK_FILE_OPEN, GET_STAGE_READER_TASK_READER_CONSTRUCTION, GetObjectFailureReason, classify_disk_error, - get_stage_timer_if_enabled, record_get_object_pipeline_failure, record_get_object_pipeline_failure_for_path, - record_get_stage_duration_if_enabled, + get_stage_timer_if_enabled, mark_get_object_downstream_closed, record_get_object_pipeline_failure, + record_get_object_pipeline_failure_for_path, record_get_stage_duration_if_enabled, }; use crate::erasure::coding::BitrotReader; use crate::io_support::bitrot::{ @@ -53,12 +53,60 @@ use std::{ task::{Context, Poll}, time::{Duration, Instant}, }; -use tokio::io::{AsyncRead, ReadBuf}; +use tokio::io::{AsyncRead, AsyncWrite, ReadBuf}; use tokio::sync::RwLock; use tokio::task::JoinSet; use super::core::io_primitives::*; +#[derive(Clone, Copy)] +pub(super) enum GetObjectOutputKind { + HttpResponse, + Internal, +} + +struct GetObjectOutputWriter<'a, W> { + inner: &'a mut W, + output_kind: GetObjectOutputKind, +} + +impl<'a, W> GetObjectOutputWriter<'a, W> { + fn new(inner: &'a mut W, output_kind: GetObjectOutputKind) -> Self { + Self { inner, output_kind } + } +} + +fn map_get_object_output_error(output_kind: GetObjectOutputKind, err: std::io::Error) -> std::io::Error { + if matches!(output_kind, GetObjectOutputKind::HttpResponse) { + mark_get_object_downstream_closed(err) + } else { + err + } +} + +impl AsyncWrite for GetObjectOutputWriter<'_, W> { + fn poll_write(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &[u8]) -> Poll> { + let output_kind = self.output_kind; + Pin::new(&mut *self.inner) + .poll_write(cx, buf) + .map(|result| result.map_err(|err| map_get_object_output_error(output_kind, err))) + } + + fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + let output_kind = self.output_kind; + Pin::new(&mut *self.inner) + .poll_flush(cx) + .map(|result| result.map_err(|err| map_get_object_output_error(output_kind, err))) + } + + fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + let output_kind = self.output_kind; + Pin::new(&mut *self.inner) + .poll_shutdown(cx) + .map(|result| result.map_err(|err| map_get_object_output_error(output_kind, err))) + } +} + impl SetDisks { async fn get_object_metadata_cache_bypass_reason( &self, @@ -586,6 +634,7 @@ impl SetDisks { offset: usize, length: i64, writer: &mut W, + output_kind: GetObjectOutputKind, fi: FileInfo, files: Vec, disks: &[Option], @@ -969,9 +1018,10 @@ impl SetDisks { let unattempted_data_shards = !reader_setup.data_shards_attempted(erasure.data_shards); let readers = reader_setup.readers; let deferred_stripe_handles = reader_setup.deferred_stripe_handles; + let mut output_writer = GetObjectOutputWriter::new(writer, output_kind); let (written, err) = erasure .decode_with_stripe_handles( - writer, + &mut output_writer, readers, part_offset, part_length, @@ -989,7 +1039,7 @@ impl SetDisks { metrics_size_bucket, decode_elapsed.as_secs_f64(), ); - if decode_elapsed >= SLOW_OBJECT_READ_LOG_THRESHOLD || err.is_some() { + if decode_elapsed >= SLOW_OBJECT_READ_LOG_THRESHOLD && err.is_none() { warn!( event = EVENT_SET_DISK_READ, component = LOG_COMPONENT_ECSTORE, @@ -1005,9 +1055,8 @@ impl SetDisks { available_shards, missing_shards, elapsed_ms = decode_elapsed.as_millis(), - error = ?err, - state = if err.is_some() { "decode_failed_or_slow" } else { "decode_slow" }, - "Set disk object decode stage is slow or failed" + state = "decode_slow", + "Set disk object decode stage is slow" ); } debug!( @@ -1051,7 +1100,7 @@ impl SetDisks { if has_err { let reason = classify_disk_error(&de_err); if reason == GetObjectFailureReason::DownstreamClosed { - warn!( + debug!( bucket, object, part_index = current_part, @@ -1969,6 +2018,7 @@ mod metadata_cache_tests { 2, 1, &mut output, + GetObjectOutputKind::Internal, valid_test_fileinfo(object), Vec::new(), &[], @@ -1992,6 +2042,7 @@ mod metadata_cache_tests { usize::MAX, 1, &mut output, + GetObjectOutputKind::Internal, overflow, Vec::new(), &[], @@ -2013,6 +2064,7 @@ mod metadata_cache_tests { 1, 1, &mut output, + GetObjectOutputKind::Internal, valid_test_fileinfo(object), Vec::new(), &[], @@ -2036,6 +2088,7 @@ mod metadata_cache_tests { 0, 1, &mut output, + GetObjectOutputKind::Internal, invalid_erasure, Vec::new(), &[], @@ -2079,6 +2132,7 @@ mod metadata_cache_tests { 0, 1, &mut output, + GetObjectOutputKind::Internal, fi, Vec::new(), &[], @@ -2920,11 +2974,60 @@ mod tests { Arc, atomic::{AtomicUsize, Ordering}, }; - use tokio::io::AsyncReadExt; + use tokio::io::{AsyncReadExt, AsyncWriteExt}; const CODEC_STREAMING_TEST_BUCKET: &str = "bucket"; const CODEC_STREAMING_TEST_OBJECT: &str = "object"; + struct ClosedOutputWriter; + + impl AsyncWrite for ClosedOutputWriter { + fn poll_write(self: Pin<&mut Self>, _cx: &mut Context<'_>, _buf: &[u8]) -> Poll> { + Poll::Ready(Err(std::io::Error::from(ErrorKind::BrokenPipe))) + } + + fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } + + fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } + } + + #[tokio::test] + async fn output_writer_marks_closed_duplex_reader_as_downstream_close() { + let (reader, mut inner) = tokio::io::duplex(64); + drop(reader); + let mut writer = GetObjectOutputWriter::new(&mut inner, GetObjectOutputKind::HttpResponse); + let err = writer + .write_all(b"range payload") + .await + .expect_err("closed duplex reader must fail the output write"); + + assert_eq!( + classify_disk_error(&DiskError::from(err)), + GetObjectFailureReason::DownstreamClosed, + "only the output adapter may classify a broken pipe as a downstream close" + ); + } + + #[tokio::test] + async fn output_writer_preserves_remote_broken_pipe_as_io_failure() { + let mut inner = ClosedOutputWriter; + let mut writer = GetObjectOutputWriter::new(&mut inner, GetObjectOutputKind::Internal); + let err = writer + .write_all(b"transition payload") + .await + .expect_err("closed remote writer must fail the output write"); + + assert_eq!( + classify_disk_error(&DiskError::from(err)), + GetObjectFailureReason::Io, + "remote transition writer failures must not be mistaken for HTTP client cancellation" + ); + } + async fn local_test_disks(count: usize, bucket: &str) -> (Vec, Vec>) { let mut dirs = Vec::with_capacity(count); let mut disks = Vec::with_capacity(count); @@ -4019,6 +4122,7 @@ mod tests { 0, part_data.len() as i64, &mut output, + GetObjectOutputKind::Internal, fi, files, &disks, diff --git a/rustfs/src/app/object_usecase.rs b/rustfs/src/app/object_usecase.rs index 776114691..f91ff09fb 100644 --- a/rustfs/src/app/object_usecase.rs +++ b/rustfs/src/app/object_usecase.rs @@ -1500,11 +1500,7 @@ impl GetObjectStreamingReader { { return "read_quorum"; } - if error_msg.contains("connection reset") - || error_msg.contains("broken pipe") - || error_msg.contains("downstream") - || error_msg.contains("remote closed") - { + if error_msg.contains("getobject output closed:") { return "downstream_closed"; } } @@ -10608,6 +10604,21 @@ mod tests { assert_eq!(err.kind(), std::io::ErrorKind::TimedOut); } + #[test] + fn get_object_streaming_reader_distinguishes_internal_and_output_broken_pipes() { + let internal = std::io::Error::from(std::io::ErrorKind::BrokenPipe); + assert_eq!(GetObjectStreamingReader::>>::classify_read_error(&internal), "io"); + + let output_closed = std::io::Error::new( + std::io::ErrorKind::BrokenPipe, + std::io::Error::other("GetObject output closed: broken pipe"), + ); + assert_eq!( + GetObjectStreamingReader::>>::classify_read_error(&output_closed), + "downstream_closed" + ); + } + #[tokio::test] async fn put_object_body_read_timeout_guard_aborts_on_stall() { // Inner stream never yields and never reports EOF (a proxy that forwarded