diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index 47de63c8c..8ba420268 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/crates/ecstore/src/set_disk/mod.rs @@ -4446,7 +4446,6 @@ 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 _, @@ -8401,7 +8400,6 @@ mod tests { range_offset, range_length as i64, &mut writer, - GetObjectOutputKind::HttpResponse, fi, files, &disks, @@ -8513,7 +8511,6 @@ 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 8eb4b590f..a3c22967b 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/crates/ecstore/src/set_disk/ops/object.rs @@ -21,7 +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::set_disk::read::GetObjectDownstreamWriter; use crate::bucket::lifecycle::{ tier_delete_journal::{persist_tier_delete_journal_entry, remove_tier_delete_journal_entry}, @@ -524,7 +524,6 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks { 0, object_info.size, &mut output, - GetObjectOutputKind::Internal, fi, files, &disks, @@ -642,7 +641,7 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks { // task so it lives for the duration of the streaming read. tokio::spawn(async move { let _guard = read_lock_guard; - let mut writer = wd; + let mut writer = GetObjectDownstreamWriter::new(wd); // Do not wrap the entire read+write pipeline in `disk_read_timeout`. // `get_object_with_fileinfo` also waits on `writer`, so an outer timeout // would incorrectly treat downstream backpressure as disk-read latency. @@ -653,7 +652,6 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks { offset, length, &mut writer, - GetObjectOutputKind::HttpResponse, fi, files, &disks, @@ -3379,7 +3377,6 @@ 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 b0e2bab5a..c55d4ed60 100644 --- a/crates/ecstore/src/set_disk/read.rs +++ b/crates/ecstore/src/set_disk/read.rs @@ -59,51 +59,33 @@ use tokio::task::JoinSet; use super::core::io_primitives::*; -#[derive(Clone, Copy)] -pub(super) enum GetObjectOutputKind { - HttpResponse, - Internal, +pub(super) struct GetObjectDownstreamWriter { + inner: W, } -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 } +impl GetObjectDownstreamWriter { + pub(super) fn new(inner: W) -> Self { + Self { inner } } } -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> { +impl AsyncWrite for GetObjectDownstreamWriter { 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) + Pin::new(&mut self.inner) .poll_write(cx, buf) - .map(|result| result.map_err(|err| map_get_object_output_error(output_kind, err))) + .map(|result| result.map_err(mark_get_object_downstream_closed)) } fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - let output_kind = self.output_kind; - Pin::new(&mut *self.inner) + Pin::new(&mut self.inner) .poll_flush(cx) - .map(|result| result.map_err(|err| map_get_object_output_error(output_kind, err))) + .map(|result| result.map_err(mark_get_object_downstream_closed)) } fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - let output_kind = self.output_kind; - Pin::new(&mut *self.inner) + Pin::new(&mut self.inner) .poll_shutdown(cx) - .map(|result| result.map_err(|err| map_get_object_output_error(output_kind, err))) + .map(|result| result.map_err(mark_get_object_downstream_closed)) } } @@ -634,7 +616,6 @@ impl SetDisks { offset: usize, length: i64, writer: &mut W, - output_kind: GetObjectOutputKind, fi: FileInfo, files: Vec, disks: &[Option], @@ -1018,10 +999,9 @@ 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( - &mut output_writer, + writer, readers, part_offset, part_length, @@ -2018,7 +1998,6 @@ mod metadata_cache_tests { 2, 1, &mut output, - GetObjectOutputKind::Internal, valid_test_fileinfo(object), Vec::new(), &[], @@ -2042,7 +2021,6 @@ mod metadata_cache_tests { usize::MAX, 1, &mut output, - GetObjectOutputKind::Internal, overflow, Vec::new(), &[], @@ -2064,7 +2042,6 @@ mod metadata_cache_tests { 1, 1, &mut output, - GetObjectOutputKind::Internal, valid_test_fileinfo(object), Vec::new(), &[], @@ -2088,7 +2065,6 @@ mod metadata_cache_tests { 0, 1, &mut output, - GetObjectOutputKind::Internal, invalid_erasure, Vec::new(), &[], @@ -2132,7 +2108,6 @@ mod metadata_cache_tests { 0, 1, &mut output, - GetObjectOutputKind::Internal, fi, Vec::new(), &[], @@ -2979,27 +2954,11 @@ mod tests { 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); + async fn downstream_writer_marks_closed_duplex_reader_as_downstream_close() { + let (reader, inner) = tokio::io::duplex(64); drop(reader); - let mut writer = GetObjectOutputWriter::new(&mut inner, GetObjectOutputKind::HttpResponse); + let mut writer = GetObjectDownstreamWriter::new(inner); let err = writer .write_all(b"range payload") .await @@ -3012,22 +2971,6 @@ mod tests { ); } - #[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); @@ -4122,7 +4065,6 @@ 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 f91ff09fb..05e753533 100644 --- a/rustfs/src/app/object_usecase.rs +++ b/rustfs/src/app/object_usecase.rs @@ -1500,9 +1500,6 @@ impl GetObjectStreamingReader { { return "read_quorum"; } - if error_msg.contains("getobject output closed:") { - return "downstream_closed"; - } } match err.kind() { @@ -10604,21 +10601,6 @@ 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