refactor(getobject): scope downstream close tagging (#5224)

Co-authored-by: heihutu <heihutu@gmail.com>

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,

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,

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) struct GetObjectDownstreamWriter<W> {
-pub(super) enum GetObjectOutputKind {
+    inner: W,
-    HttpResponse,
-    Internal,
 }
 
-struct GetObjectOutputWriter<'a, W> {
+impl<W> GetObjectDownstreamWriter<W> {
-    inner: &'a mut W,
+    pub(super) fn new(inner: W) -> Self {
-    output_kind: GetObjectOutputKind,
+        Self { inner }
-}
-
-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 {
+impl<W: AsyncWrite + Unpin> AsyncWrite for GetObjectDownstreamWriter<W> {
-    if matches!(output_kind, GetObjectOutputKind::HttpResponse) {
-        mark_get_object_downstream_closed(err)
-    } else {
-        err
-    }
-}
-
-impl<W: AsyncWrite + Unpin> AsyncWrite for GetObjectOutputWriter<'_, W> {
     fn poll_write(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &[u8]) -> Poll<std::io::Result<usize>> {
-        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<std::io::Result<()>> {
-        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<std::io::Result<()>> {
-        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<FileInfo>,
         disks: &[Option<DiskStore>],
@@ -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<std::io::Result<usize>> {
-            Poll::Ready(Err(std::io::Error::from(ErrorKind::BrokenPipe)))
-        }
-
-        fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
-            Poll::Ready(Ok(()))
-        }
-
-        fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
-            Poll::Ready(Ok(()))
-        }
-    }
-
     #[tokio::test]
-    async fn output_writer_marks_closed_duplex_reader_as_downstream_close() {
+    async fn downstream_writer_marks_closed_duplex_reader_as_downstream_close() {
-        let (reader, mut inner) = tokio::io::duplex(64);
+        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<tempfile::TempDir>, Vec<Option<crate::disk::DiskStore>>) {
         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,

rustfs/src/app/object_usecase.rs

@@ -1500,9 +1500,6 @@ impl<R> GetObjectStreamingReader<R> {
             {
                 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::<std::io::Cursor<Vec<u8>>>::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::<std::io::Cursor<Vec<u8>>>::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