fix(rdma): preserve transfer diagnostics and payload lengths

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 {

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,

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<ObjectInfo> {
         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<Vec<String>>,
+        uploaded_payloads: std::sync::Mutex<Vec<Vec<u8>>>,
     }
 
     #[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]