From 6f7a4ff060d4a3e79c5e8a73b2c2a9efe4c213bd Mon Sep 17 00:00:00 2001 From: houseme Date: Thu, 27 Aug 2026 23:42:51 +0800 Subject: [PATCH] fix(api): preserve server-side storage error surface (#6753) Co-authored-by: heihutu --- rustfs/src/app/object/copy.rs | 89 +++++++++++++++++++++++++++++++++- rustfs/src/error.rs | 90 ++++++++++++++++++++++++++++++----- 2 files changed, 165 insertions(+), 14 deletions(-) diff --git a/rustfs/src/app/object/copy.rs b/rustfs/src/app/object/copy.rs index c9f855662..6d7ee4f80 100644 --- a/rustfs/src/app/object/copy.rs +++ b/rustfs/src/app/object/copy.rs @@ -17,6 +17,71 @@ use super::*; use crate::auth::{VerifiedPresignedRequest, reject_presigned_put_max_content_length_for_other_operation}; +use crate::error::ServerSideSourceReadError; + +struct CopySourceReadStream { + inner: R, + remaining: i64, +} + +impl CopySourceReadStream { + fn new(inner: R, expected_size: i64) -> Self { + Self { + inner, + remaining: expected_size.max(0), + } + } +} + +fn copy_source_read_stream(inner: R, expected_size: i64) -> CopySourceReadStream { + CopySourceReadStream::new(inner, expected_size) +} + +fn copy_source_read_error(source: std::io::Error) -> std::io::Error { + let kind = source.kind(); + std::io::Error::new(kind, ServerSideSourceReadError::new("CopyObject", source)) +} + +fn copy_source_incomplete_body_error(remaining: i64) -> std::io::Error { + copy_source_read_error(std::io::Error::new( + std::io::ErrorKind::UnexpectedEof, + rustfs_rio::IncompleteBody { remaining }, + )) +} + +impl AsyncRead for CopySourceReadStream +where + R: AsyncRead + Unpin, +{ + fn poll_read(self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll> { + let this = self.get_mut(); + let before = buf.filled().len(); + match Pin::new(&mut this.inner).poll_read(cx, buf) { + Poll::Pending => Poll::Pending, + Poll::Ready(Err(err)) => Poll::Ready(Err(copy_source_read_error(err))), + Poll::Ready(Ok(())) => { + let read = buf.filled().len() - before; + if read == 0 { + if this.remaining > 0 { + return Poll::Ready(Err(copy_source_incomplete_body_error(this.remaining))); + } + return Poll::Ready(Ok(())); + } + + let read = match i64::try_from(read) { + Ok(read) => read, + Err(_) => { + return Poll::Ready(Err(copy_source_read_error(std::io::Error::other( + "copy source read count exceeds i64::MAX", + )))); + } + }; + this.remaining = this.remaining.saturating_sub(read); + Poll::Ready(Ok(())) + } + } + } +} fn copy_namespace_lock_error(bucket: &str, object: &str, mode: &'static str, err: rustfs_lock::LockError) -> StorageError { match err { @@ -580,11 +645,13 @@ impl DefaultObjectUsecase { let mut write_plan = WritePlan::new(); let mut reader = if should_compress { let algorithm = CompressionAlgorithm::default(); - let hrd = HashReader::from_stream(gr.stream, length, actual_size, None, None, false).map_err(ApiError::from)?; + let hrd = HashReader::from_stream(copy_source_read_stream(gr.stream, length), length, actual_size, None, None, false) + .map_err(ApiError::from)?; write_plan = write_plan.with_compression(algorithm); hrd } else { - HashReader::from_stream(gr.stream, length, actual_size, None, None, false).map_err(ApiError::from)? + HashReader::from_stream(copy_source_read_stream(gr.stream, length), length, actual_size, None, None, false) + .map_err(ApiError::from)? }; // Give the destination object a checksum so CopyObject returns it and a later checksum-mode @@ -835,6 +902,7 @@ mod tests { use http::{HeaderValue, Method}; use s3s::dto::{ServerSideEncryptionByDefault, ServerSideEncryptionConfiguration, ServerSideEncryptionRule}; use std::sync::Arc; + use tokio::io::AsyncReadExt; // A malformed bucket-default algorithm reaches this resolution only through // corrupt or hand-edited bucket metadata (PutBucketEncryption validates the @@ -876,6 +944,23 @@ mod tests { } } + #[tokio::test] + async fn copy_source_read_stream_maps_short_eof_to_service_unavailable() { + let source = std::io::Cursor::new(b"abc".to_vec()); + let mut reader = HashReader::from_stream(copy_source_read_stream(source, 4), 4, 4, None, None, false) + .expect("copy source hash reader should build"); + + let mut output = Vec::new(); + let err = reader + .read_to_end(&mut output) + .await + .expect_err("short copy source must fail before destination write succeeds"); + let api_error = ApiError::from(err); + + assert_eq!(api_error.code, S3ErrorCode::ServiceUnavailable); + assert_ne!(api_error.code, S3ErrorCode::IncompleteBody); + } + #[tokio::test] async fn execute_copy_object_rejects_self_copy_without_replace_directive() { let input = CopyObjectInput::builder() diff --git a/rustfs/src/error.rs b/rustfs/src/error.rs index d72026ea8..cb9bafdd9 100644 --- a/rustfs/src/error.rs +++ b/rustfs/src/error.rs @@ -34,6 +34,32 @@ impl std::fmt::Display for UploadLimitExceeded { impl std::error::Error for UploadLimitExceeded {} +/// Marks a server-side object/source reader failure that must not be reported as +/// a malformed client request body. +#[derive(Debug)] +pub(crate) struct ServerSideSourceReadError { + operation: &'static str, + source: std::io::Error, +} + +impl ServerSideSourceReadError { + pub(crate) const fn new(operation: &'static str, source: std::io::Error) -> Self { + Self { operation, source } + } +} + +impl std::fmt::Display for ServerSideSourceReadError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "{} source read failed: {}", self.operation, self.source) + } +} + +impl std::error::Error for ServerSideSourceReadError { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + Some(&self.source) + } +} + #[derive(Debug)] pub struct ApiError { pub code: S3ErrorCode, @@ -302,6 +328,17 @@ impl From for ApiError { }; } + if let StorageError::Io(ref io_err) = err + && let Some(inner) = io_err.get_ref() + && error_chain_has_type::(inner) + { + return ApiError { + code: S3ErrorCode::ServiceUnavailable, + message: ApiError::error_code_to_message(&S3ErrorCode::ServiceUnavailable), + source: Some(Box::new(err)), + }; + } + if let StorageError::Io(ref io_err) = err && io_err .get_ref() @@ -340,15 +377,15 @@ impl From for ApiError { StorageError::ObjectNameInvalid(_, _) => S3ErrorCode::InvalidArgument, StorageError::BucketExists(_) => S3ErrorCode::BucketAlreadyOwnedByYou, StorageError::StorageFull => S3ErrorCode::ServiceUnavailable, - StorageError::SlowDown - | StorageError::FaultyDisk + StorageError::SlowDown => S3ErrorCode::SlowDown, + StorageError::FaultyDisk | StorageError::FaultyRemoteDisk | StorageError::DiskNotFound - | StorageError::TooManyOpenFiles => S3ErrorCode::SlowDown, + | StorageError::TooManyOpenFiles => S3ErrorCode::ServiceUnavailable, StorageError::ErasureReadQuorum | StorageError::InsufficientReadQuorum(_, _) | StorageError::ErasureWriteQuorum - | StorageError::InsufficientWriteQuorum(_, _) => S3ErrorCode::SlowDown, + | StorageError::InsufficientWriteQuorum(_, _) => S3ErrorCode::ServiceUnavailable, StorageError::NamespaceLockQuorumUnavailable { .. } => S3ErrorCode::ServiceUnavailable, StorageError::QuotaExceeded { .. } => S3ErrorCode::InvalidRequest, StorageError::Lock(_) => S3ErrorCode::ServiceUnavailable, @@ -435,6 +472,13 @@ impl From for ApiError { source: Some(Box::new(err)), }; } + if error_chain_has_type::(inner) { + return ApiError { + code: S3ErrorCode::ServiceUnavailable, + message: ApiError::error_code_to_message(&S3ErrorCode::ServiceUnavailable), + source: Some(Box::new(err)), + }; + } if error_chain_has_type::(inner) { return ApiError { code: S3ErrorCode::IncompleteBody, @@ -613,6 +657,22 @@ mod tests { } } + #[test] + fn server_side_source_read_error_maps_to_service_unavailable_before_incomplete_body() { + let short_source = IoError::new(ErrorKind::UnexpectedEof, rustfs_rio::IncompleteBody { remaining: 17 }); + let marker = ServerSideSourceReadError::new("CopyObject", short_source); + + let api_error = ApiError::from(IoError::new(ErrorKind::UnexpectedEof, marker)); + assert_eq!(api_error.code, S3ErrorCode::ServiceUnavailable); + assert_ne!(api_error.code, S3ErrorCode::IncompleteBody); + + let short_source = IoError::new(ErrorKind::UnexpectedEof, rustfs_rio::IncompleteBody { remaining: 17 }); + let marker = ServerSideSourceReadError::new("CopyObject", short_source); + let api_error = ApiError::from(StorageError::Io(IoError::new(ErrorKind::UnexpectedEof, marker))); + assert_eq!(api_error.code, S3ErrorCode::ServiceUnavailable); + assert_ne!(api_error.code, S3ErrorCode::IncompleteBody); + } + #[test] fn test_api_error_surfaces_invalid_argument_reason() { let err = StorageError::InvalidArgument( @@ -785,14 +845,20 @@ mod tests { (StorageError::BucketExists("test".into()), S3ErrorCode::BucketAlreadyOwnedByYou), (StorageError::StorageFull, S3ErrorCode::ServiceUnavailable), (StorageError::SlowDown, S3ErrorCode::SlowDown), - (StorageError::FaultyDisk, S3ErrorCode::SlowDown), - (StorageError::FaultyRemoteDisk, S3ErrorCode::SlowDown), - (StorageError::DiskNotFound, S3ErrorCode::SlowDown), - (StorageError::TooManyOpenFiles, S3ErrorCode::SlowDown), - (StorageError::ErasureReadQuorum, S3ErrorCode::SlowDown), - (StorageError::InsufficientReadQuorum("test".into(), "test".into()), S3ErrorCode::SlowDown), - (StorageError::ErasureWriteQuorum, S3ErrorCode::SlowDown), - (StorageError::InsufficientWriteQuorum("test".into(), "test".into()), S3ErrorCode::SlowDown), + (StorageError::FaultyDisk, S3ErrorCode::ServiceUnavailable), + (StorageError::FaultyRemoteDisk, S3ErrorCode::ServiceUnavailable), + (StorageError::DiskNotFound, S3ErrorCode::ServiceUnavailable), + (StorageError::TooManyOpenFiles, S3ErrorCode::ServiceUnavailable), + (StorageError::ErasureReadQuorum, S3ErrorCode::ServiceUnavailable), + ( + StorageError::InsufficientReadQuorum("test".into(), "test".into()), + S3ErrorCode::ServiceUnavailable, + ), + (StorageError::ErasureWriteQuorum, S3ErrorCode::ServiceUnavailable), + ( + StorageError::InsufficientWriteQuorum("test".into(), "test".into()), + S3ErrorCode::ServiceUnavailable, + ), ( StorageError::NamespaceLockQuorumUnavailable { mode: "write",