From b18ccefd1bdf89e049362425c0fd0f91eacf3310 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=A9=AC=E7=99=BB=E5=B1=B1?= Date: Fri, 14 Aug 2026 09:30:27 +0800 Subject: [PATCH] fix(ecstore): preserve CopyObject producer errors --- crates/ecstore/src/set_disk/ops/object.rs | 164 +++++++++++++++++++++- crates/rio/src/http_reader.rs | 60 ++++++-- rustfs/src/error.rs | 17 +++ 3 files changed, 229 insertions(+), 12 deletions(-) diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index 10dda4d81..caa16bf04 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/crates/ecstore/src/set_disk/ops/object.rs @@ -882,6 +882,8 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks { let (rd, wd) = tokio::io::duplex(duplex_buffer_size); debug!(bucket, object, duplex_buffer_size, "Created duplex pipe for object data transfer"); + let (producer_terminal_tx, producer_terminal_rx) = tokio::sync::oneshot::channel(); + let rd = LegacyDuplexProducerReader::new(rd, producer_terminal_rx); let (mut reader, offset, length) = get_object_reader_with_context(&self.ctx, Box::new(rd), range, &object_info, opts, &h).await?; // Carry the hook probe result so the app layer skips its now-redundant @@ -902,7 +904,7 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks { // `get_object_with_fileinfo` also waits on `writer`, so an outer timeout // would incorrectly treat downstream backpressure as disk-read latency. // Disk read timeouts must be enforced at the actual disk I/O operations. - if let Err(e) = Self::get_object_with_fileinfo( + let producer_result = Self::get_object_with_fileinfo( &bucket, &object, offset, @@ -919,9 +921,9 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks { object_class.as_str(), size_bucket, ) - .await - { - let reason = classify_storage_error(&e); + .await; + if let Err(e) = &producer_result { + let reason = classify_storage_error(e); if reason == GetObjectFailureReason::DownstreamClosed { debug!( event = EVENT_SET_DISK_WRITE, @@ -960,6 +962,7 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks { ); } }; + let _ = producer_terminal_tx.send(producer_result.map(|_| ())); }); Ok(reader) @@ -1869,6 +1872,159 @@ impl AsyncRead for TransitionUploadReader { } } +struct LegacyDuplexProducerReader { + inner: R, + terminal: Option>>, + inner_eof: bool, +} + +impl LegacyDuplexProducerReader { + fn new(inner: R, terminal: tokio::sync::oneshot::Receiver>) -> Self { + Self { + inner, + terminal: Some(terminal), + inner_eof: false, + } + } +} + +impl AsyncRead for LegacyDuplexProducerReader { + fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll> { + if !self.inner_eof { + let before = buf.filled().len(); + match Pin::new(&mut self.inner).poll_read(cx, buf) { + Poll::Pending => return Poll::Pending, + Poll::Ready(Err(err)) => return Poll::Ready(Err(err)), + Poll::Ready(Ok(())) if buf.filled().len() > before => return Poll::Ready(Ok(())), + Poll::Ready(Ok(())) => { + self.inner_eof = true; + } + } + } + + let Some(terminal) = self.terminal.as_mut() else { + return Poll::Ready(Ok(())); + }; + match Pin::new(terminal).poll(cx) { + Poll::Pending => Poll::Pending, + Poll::Ready(Ok(Ok(()))) => { + self.terminal = None; + Poll::Ready(Ok(())) + } + Poll::Ready(Ok(Err(err))) => { + self.terminal = None; + Poll::Ready(Err(std::io::Error::other(err))) + } + Poll::Ready(Err(_)) => { + self.terminal = None; + Poll::Ready(Err(std::io::Error::other(StorageError::Unexpected))) + } + } + } +} + +#[cfg(test)] +mod legacy_duplex_producer_reader_tests { + use super::*; + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + + fn storage_error_source(error: &std::io::Error) -> &StorageError { + error + .get_ref() + .and_then(|source| source.downcast_ref::()) + .expect("legacy duplex terminal error should retain StorageError source") + } + + #[tokio::test] + async fn legacy_duplex_reader_allows_clean_completion() { + let (mut writer, reader) = tokio::io::duplex(64); + let (terminal_tx, terminal_rx) = tokio::sync::oneshot::channel(); + writer + .write_all(b"complete") + .await + .expect("duplex write should fit in buffer"); + drop(writer); + terminal_tx.send(Ok(())).expect("terminal receiver should remain installed"); + + let mut reader = LegacyDuplexProducerReader::new(reader, terminal_rx); + let mut out = Vec::new(); + reader + .read_to_end(&mut out) + .await + .expect("clean producer completion should surface clean EOF"); + + assert_eq!(out, b"complete"); + } + + #[tokio::test] + async fn legacy_duplex_reader_surfaces_terminal_error_after_partial_data() { + let (mut writer, reader) = tokio::io::duplex(64); + let (terminal_tx, terminal_rx) = tokio::sync::oneshot::channel(); + writer.write_all(b"partial").await.expect("duplex write should fit in buffer"); + drop(writer); + terminal_tx + .send(Err(StorageError::FileCorrupt)) + .expect("terminal receiver should remain installed"); + + let mut reader = LegacyDuplexProducerReader::new(reader, terminal_rx); + let mut out = Vec::new(); + let err = reader + .read_to_end(&mut out) + .await + .expect_err("terminal producer error must not become clean EOF"); + + assert_eq!(out, b"partial"); + assert!(matches!(storage_error_source(&err), StorageError::FileCorrupt)); + } + + #[tokio::test] + async fn legacy_duplex_reader_surfaces_terminal_error_after_declared_length() { + let (mut writer, reader) = tokio::io::duplex(64); + let (terminal_tx, terminal_rx) = tokio::sync::oneshot::channel(); + writer.write_all(b"exact").await.expect("duplex write should fit in buffer"); + drop(writer); + terminal_tx + .send(Err(StorageError::Io(std::io::Error::new( + std::io::ErrorKind::ConnectionReset, + "remote body reset after final byte", + )))) + .expect("terminal receiver should remain installed"); + + let reader = LegacyDuplexProducerReader::new(reader, terminal_rx); + let mut reader = + HashReader::from_stream(reader, 5, 5, None, None, false).expect("hash reader should accept exact declared length"); + let mut out = Vec::new(); + let err = reader + .read_to_end(&mut out) + .await + .expect_err("producer terminal error after the declared length must still fail"); + + assert_eq!(out, b"exact"); + assert!( + matches!(storage_error_source(&err), StorageError::Io(io_error) if io_error.kind() == std::io::ErrorKind::ConnectionReset) + ); + } + + #[tokio::test] + async fn legacy_duplex_reader_fails_closed_when_terminal_channel_closes() { + let (mut writer, reader) = tokio::io::duplex(64); + let (terminal_tx, terminal_rx) = tokio::sync::oneshot::channel::>(); + writer.write_all(b"body").await.expect("duplex write should fit in buffer"); + drop(writer); + drop(terminal_tx); + + let mut reader = LegacyDuplexProducerReader::new(reader, terminal_rx); + let mut out = Vec::new(); + let err = reader + .read_to_end(&mut out) + .await + .expect_err("producer disappearance must fail closed"); + + assert_eq!(out, b"body"); + assert!(matches!(storage_error_source(&err), StorageError::Unexpected)); + } +} + struct TransitionUploadWriter { inner: W, produced: u64, diff --git a/crates/rio/src/http_reader.rs b/crates/rio/src/http_reader.rs index 8be68cef4..96fd0971f 100644 --- a/crates/rio/src/http_reader.rs +++ b/crates/rio/src/http_reader.rs @@ -138,6 +138,12 @@ impl std::fmt::Display for InternodeHttpErrorKind { } } +#[derive(thiserror::Error, Debug, Clone, Copy, Eq, PartialEq)] +#[error("internode body stalled for {timeout:?}")] +pub struct BodyStalled { + pub timeout: Duration, +} + #[derive(Debug, Clone, Eq, PartialEq)] pub struct InternodeHttpRequestContext { method: String, @@ -271,6 +277,10 @@ pub fn internode_http_timeout_error(method: &Method, url: &str) -> io::Error { internode_kind_error(method, url, internode_rpc_operation(url), InternodeHttpErrorKind::ConnectTimeout) } +fn body_stalled_error(stall_timeout: Duration) -> io::Error { + Error::new(io::ErrorKind::TimedOut, BodyStalled { timeout: stall_timeout }) +} + /// Clone an internode HTTP I/O error while retaining its structured classification. /// /// The underlying transport source is intentionally omitted because it is not @@ -1085,10 +1095,7 @@ impl AsyncRead for HttpReader { ); record_internode_stall_timeout(*this.track_internode_metrics, *this.internode_operation); record_internode_error(*this.track_internode_metrics, *this.internode_operation); - Poll::Ready(Err(Error::new( - io::ErrorKind::TimedOut, - "HttpReader stall timeout: no data received before deadline", - ))) + Poll::Ready(Err(body_stalled_error(stall_timeout))) } else { Poll::Pending } @@ -1217,10 +1224,7 @@ impl ChunkReader for HttpChunkReader { ); record_internode_stall_timeout(*this.track_internode_metrics, *this.internode_operation); record_internode_error(*this.track_internode_metrics, *this.internode_operation); - return Poll::Ready(Err(Error::new( - io::ErrorKind::TimedOut, - "HttpReader stall timeout: no data received before deadline", - ))); + return Poll::Ready(Err(body_stalled_error(stall_timeout))); } return Poll::Pending; } @@ -2379,6 +2383,46 @@ mod tests { Err(err) => err, }; assert_eq!(err.kind(), io::ErrorKind::TimedOut); + let stalled = err + .get_ref() + .and_then(|source| source.downcast_ref::()) + .expect("stall timeout should retain typed body-stalled source"); + assert_eq!(stalled.timeout, Duration::from_millis(20)); + + handle.abort(); + } + + #[tokio::test] + async fn http_chunk_reader_stall_timeout_retains_typed_source() { + let state = TestState::default(); + let Some((base_url, handle)) = start_test_server(state).await else { + return; + }; + let url = base_url.replace("/stream", "/stall"); + let mut reader = + HttpChunkReader::new_with_stall_timeout(url, Method::GET, HeaderMap::new(), None, Some(Duration::from_millis(20))) + .await + .expect("chunk reader should open"); + + let first = std::future::poll_fn(|cx| Pin::new(&mut reader).poll_read_chunk(cx, 64)) + .await + .expect("initial body chunk should arrive") + .expect("initial body chunk should not be EOF"); + assert_eq!(first, b"hello"[..]); + + let err = tokio::time::timeout( + Duration::from_secs(1), + std::future::poll_fn(|cx| Pin::new(&mut reader).poll_read_chunk(cx, 64)), + ) + .await + .expect("stall timeout should wake chunk reader") + .expect_err("chunk reader should return a timeout error"); + assert_eq!(err.kind(), io::ErrorKind::TimedOut); + let stalled = err + .get_ref() + .and_then(|source| source.downcast_ref::()) + .expect("chunk stall timeout should retain typed body-stalled source"); + assert_eq!(stalled.timeout, Duration::from_millis(20)); handle.abort(); } diff --git a/rustfs/src/error.rs b/rustfs/src/error.rs index 49ac36319..15089c34e 100644 --- a/rustfs/src/error.rs +++ b/rustfs/src/error.rs @@ -669,6 +669,23 @@ mod tests { assert!(api_error.source.is_some()); } + #[test] + fn test_api_error_from_storage_io_copy_object_terminal_error_stays_internal() { + let io_error = IoError::other(StorageError::FileCorrupt); + let storage_error: StorageError = io_error.into(); + assert!(matches!(storage_error, StorageError::FileCorrupt)); + + let api_error: ApiError = storage_error.into(); + + assert_eq!(api_error.code, S3ErrorCode::InternalError); + let source = api_error + .source + .as_deref() + .and_then(|source| source.downcast_ref::()) + .expect("API error should retain the storage error source"); + assert!(matches!(source, StorageError::FileCorrupt)); + } + #[test] fn test_api_error_from_iam_error() { let iam_error = rustfs_iam::error::Error::other("IAM test error");