opensource/rustfs
3
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-28 19:24:58 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(api): preserve server-side storage error surface (#6753)

Browse Source 
Co-authored-by: heihutu <heihutu@gmail.com>
This commit is contained in:
houseme
2026-08-27 23:42:51 +08:00
committed by GitHub
parent 8a57632bfd
commit 6f7a4ff060
2 changed files with 165 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rustfs/src/app/object/copy.rs
+87 -2
View File
@@ -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<R> {
inner: R,
remaining: i64,
}
impl<R> CopySourceReadStream<R> {
fn new(inner: R, expected_size: i64) -> Self {
Self {
inner,
remaining: expected_size.max(0),
}
}
}
fn copy_source_read_stream<R>(inner: R, expected_size: i64) -> CopySourceReadStream<R> {
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<R> AsyncRead for CopySourceReadStream<R>
where
R: AsyncRead + Unpin,
{
fn poll_read(self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<std::io::Result<()>> {
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()
rustfs/src/error.rs
+78 -12
View File
@@ -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<StorageError> for ApiError {
};
}
if let StorageError::Io(ref io_err) = err
&& let Some(inner) = io_err.get_ref()
&& error_chain_has_type::<ServerSideSourceReadError>(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<StorageError> 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<std::io::Error> for ApiError {
source: Some(Box::new(err)),
};
}
if error_chain_has_type::<ServerSideSourceReadError>(inner) {
return ApiError {
code: S3ErrorCode::ServiceUnavailable,
message: ApiError::error_code_to_message(&S3ErrorCode::ServiceUnavailable),
source: Some(Box::new(err)),
};
}
if error_chain_has_type::<rustfs_rio::IncompleteBody>(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",