From b56b9b7c18dbd6fa50fb99bb11821dd04938be76 Mon Sep 17 00:00:00 2001 From: gatewayJ <835269233@qq.com> Date: Mon, 2 Nov 2026 23:32:56 +0800 Subject: [PATCH] fix(rdma): preserve transfer diagnostics and payload lengths --- crates/core/src/data_transport.rs | 31 +++++++ crates/core/src/error.rs | 5 +- crates/s3/src/client.rs | 133 ++++++++++++++++++++++++++++-- 3 files changed, 159 insertions(+), 10 deletions(-) diff --git a/crates/core/src/data_transport.rs b/crates/core/src/data_transport.rs index 0e56226..5c6b2b2 100644 --- a/crates/core/src/data_transport.rs +++ b/crates/core/src/data_transport.rs @@ -94,6 +94,37 @@ mod tests { } } + #[test] + fn transfer_error_display_includes_optional_source_without_changing_outcome() { + for state in [ + DataTransferState::NotAccepted, + DataTransferState::ConfirmedFailure, + DataTransferState::OutcomeUnknown, + ] { + let mut error = crate::Error::DataTransfer { + state, + message: "S3 request failed".to_string(), + source: None, + }; + assert_eq!( + error.to_string(), + format!("Object data transfer {state}: S3 request failed") + ); + if let crate::Error::DataTransfer { source, .. } = &mut error { + *source = Some(Box::new(crate::Error::Auth("access denied".to_string()))); + } + assert_eq!( + error.to_string(), + format!( + "Object data transfer {state}: S3 request failed: Authentication failed: access denied" + ) + ); + assert_eq!(error.exit_code(), 4); + assert!(std::error::Error::source(&error).is_some()); + assert!(!crate::is_retryable_error(&error)); + } + } + #[test] fn authoritative_error_retains_exit_code_and_cleanup_without_retry() { let error = crate::Error::DataTransfer { diff --git a/crates/core/src/error.rs b/crates/core/src/error.rs index d0b9e4c..afb52dc 100644 --- a/crates/core/src/error.rs +++ b/crates/core/src/error.rs @@ -104,7 +104,10 @@ pub enum Error { Network(String), /// An external data operation with explicit acceptance and outcome state. - #[error("Object data transfer {state}: {message}")] + #[error( + "Object data transfer {state}: {message}{cause}", + cause = .source.as_ref().map(|source| format!(": {source}")).unwrap_or_default() + )] DataTransfer { /// Acceptance and business outcome classification. state: crate::DataTransferState, diff --git a/crates/s3/src/client.rs b/crates/s3/src/client.rs index 9ce099f..2932f27 100644 --- a/crates/s3/src/client.rs +++ b/crates/s3/src/client.rs @@ -4224,12 +4224,13 @@ impl S3Client { &self, path: &RemotePath, file_path: &std::path::Path, - file_size: u64, options: PathUploadOptions<'_>, + on_progress: impl Fn(u64) + Send, ) -> Result { let data = tokio::fs::read(file_path) .await .map_err(|e| Error::General(format!("read file '{}': {e}", file_path.display())))?; + let file_size = data.len() as u64; let requested_checksum = requested_sha256_checksum(&data, options.write.checksum.as_ref())?; let condition = match options.precondition { ObjectWritePrecondition::None => WriteCondition::None, @@ -4259,6 +4260,7 @@ impl S3Client { .await .map_err(data_transport::committed_verification_failure)?; } + on_progress(file_size); return Ok(result.object); } let body = aws_sdk_s3::primitives::ByteStream::from(data); @@ -4326,6 +4328,7 @@ impl S3Client { ) .await?; } + on_progress(file_size); Ok(info) } @@ -4849,11 +4852,8 @@ impl S3Client { self.put_object_multipart_from_path(path, file_path, file_size, options, on_progress) .await } else { - let result = self - .put_object_single_part_from_path(path, file_path, file_size, options) - .await?; - on_progress(file_size); - Ok(result) + self.put_object_single_part_from_path(path, file_path, options, on_progress) + .await } } @@ -14415,6 +14415,7 @@ mod tests { struct TestExternalTransport { behavior: ExternalBehavior, operations: std::sync::Mutex>, + uploaded_payloads: std::sync::Mutex>>, } #[async_trait] @@ -14477,6 +14478,10 @@ mod tests { .await .expect("read external source"); assert_eq!(payload.len() as u64, request.content_length); + self.uploaded_payloads + .lock() + .expect("record upload payload") + .push(payload); if matches!(self.behavior, ExternalBehavior::Unknown) { return Err(data_transport::DataTransportError { state: rc_core::DataTransferState::OutcomeUnknown, @@ -14502,9 +14507,106 @@ mod tests { Arc::new(TestExternalTransport { behavior, operations: std::sync::Mutex::new(Vec::new()), + uploaded_payloads: std::sync::Mutex::new(Vec::new()), }) } + #[tokio::test] + async fn external_single_part_upload_uses_current_file_contents() { + for payload in [b"longer payload".as_slice(), b"x", b""] { + let (client, replay) = test_s3_client_with_response_sequence(Vec::new()); + let provider = external_transport(ExternalBehavior::Success); + let client = client + .with_object_data_transport(rc_core::DataTransportMode::Rdma, provider.clone()); + let source = tempfile::NamedTempFile::new().expect("file source"); + std::fs::write(source.path(), b"original").expect("initial source"); + let initial_size = std::fs::metadata(source.path()) + .expect("initial metadata") + .len(); + std::fs::write(source.path(), payload).expect("replace source"); + assert_ne!(initial_size, payload.len() as u64); + let progress = std::sync::Mutex::new(Vec::new()); + let info = client + .put_object_single_part_from_path( + &RemotePath::new("test", "bucket", "key"), + source.path(), + PathUploadOptions { + write: &ObjectWriteOptions::default(), + precondition: ObjectWritePrecondition::None, + }, + |bytes| progress.lock().expect("progress record").push(bytes), + ) + .await + .expect("upload current file contents"); + assert_eq!(info.size_bytes, Some(payload.len() as i64)); + assert_eq!( + *progress.lock().expect("progress record"), + vec![payload.len() as u64] + ); + assert_eq!( + provider + .uploaded_payloads + .lock() + .expect("uploaded payloads") + .as_slice(), + &[payload.to_vec()] + ); + assert_eq!(replay.actual_requests().count(), 0); + } + } + + #[tokio::test] + async fn external_single_part_auto_fallback_uses_current_file_contents() { + for payload in [b"longer payload".as_slice(), b"x", b""] { + let response = http::Response::builder() + .status(200) + .header("etag", "\"http-etag\"") + .body(SdkBody::empty()) + .expect("HTTP PUT response"); + let (client, receiver) = test_s3_client(Some(response)); + let provider = external_transport(ExternalBehavior::Reject); + let client = client + .with_object_data_transport(rc_core::DataTransportMode::Auto, provider.clone()); + let source = tempfile::NamedTempFile::new().expect("file source"); + std::fs::write(source.path(), b"original").expect("initial source"); + let initial_size = std::fs::metadata(source.path()) + .expect("initial metadata") + .len(); + std::fs::write(source.path(), payload).expect("replace source"); + assert_ne!(initial_size, payload.len() as u64); + let progress = std::sync::Mutex::new(Vec::new()); + let info = client + .put_object_single_part_from_path( + &RemotePath::new("test", "bucket", "key"), + source.path(), + PathUploadOptions { + write: &ObjectWriteOptions::default(), + precondition: ObjectWritePrecondition::None, + }, + |bytes| progress.lock().expect("progress record").push(bytes), + ) + .await + .expect("HTTP fallback uploads current file contents"); + assert_eq!(info.size_bytes, Some(payload.len() as i64)); + assert_eq!( + *progress.lock().expect("progress record"), + vec![payload.len() as u64] + ); + assert_eq!(receiver.expect_request().body().bytes(), Some(payload)); + assert!( + provider + .uploaded_payloads + .lock() + .expect("uploaded payloads") + .is_empty() + ); + assert_eq!( + provider.operations.lock().expect("upload attempts").len(), + 1 + ); + } + } + #[tokio::test] async fn external_auto_rejection_preserves_http_get_selection() { let response = http::Response::builder() @@ -14660,16 +14762,26 @@ mod tests { #[test] fn external_error_source_is_redacted_without_losing_business_class() { let (client, _) = test_s3_client(None); + let key = test_sse_customer_key(); + let headers = SseCustomerHeaders::new(&key); let error = Error::DataTransfer { state: rc_core::DataTransferState::ConfirmedFailure, message: "secret-key in response".to_string(), - source: Some(Box::new(Error::Auth("access-key secret-key".to_string()))), + source: Some(Box::new(Error::Auth(format!( + "access denied: access-key secret-key {} {} 0123456789abcdef0123456789abcdef", + headers.key.as_str(), + headers.key_md5.as_str(), + )))), }; - let error = client.redact_sensitive_error(error); + let error = client.redact_sse_customer_error(error, Some(&key)); assert_eq!(error.exit_code(), 4); - let diagnostic = format!("{error:?}"); + assert!(error.to_string().contains("access denied")); + let diagnostic = format!("{error:?} {error}"); assert!(!diagnostic.contains("secret-key")); assert!(!diagnostic.contains("access-key")); + assert!(!diagnostic.contains(headers.key.as_str())); + assert!(!diagnostic.contains(headers.key_md5.as_str())); + assert!(!diagnostic.contains("0123456789abcdef0123456789abcdef")); assert!(std::error::Error::source(&error).is_some()); assert!(!rc_core::is_retryable_error(&error)); } @@ -14733,6 +14845,7 @@ mod tests { )); assert!(!rc_core::is_retryable_error(&error)); assert_eq!(receiver.expect_request().method(), "HEAD"); + assert!(error.to_string().contains("ServiceUnavailable")); } #[tokio::test] async fn external_multipart_completion_failure_does_not_replay_the_file() { @@ -14777,6 +14890,7 @@ mod tests { )); assert!(!rc_core::is_retryable_error(&error)); assert!(error.to_string().contains("upload-id")); + assert!(error.to_string().contains("ServiceUnavailable")); assert_eq!(replay.actual_requests().count(), 3); } @@ -14824,6 +14938,7 @@ mod tests { } )); assert!(!rc_core::is_retryable_error(&error)); + assert!(error.to_string().contains("ServiceUnavailable")); assert_eq!(replay.actual_requests().count(), 3); } #[tokio::test]