Fix RecordingClient request body recording for non‑streaming bodies

This fixes a bug in `RecordingClient` where recording request bodies could
corrupt real HTTP requests. For non‑streaming request bodies (where `SdkBody::bytes()`
is available), we now buffer the bytes, record them, and rebuild the request body
from the buffered data instead of swapping in a channel‑based streaming body. This
avoids timing issues that can break HTTP/2 framing and cause 400 responses from ELB.

Also preserves size hints on channel bodies by carrying over the original content length,
keeping body metadata consistent with `Content-Length`.

Author: signadou

.changelog/recordingclient-request-body.md

@@ -0,0 +1,12 @@
+---
+applies_to:
+- client
+authors:
+- signadou
+references:
+- smithy-rs#4495
+breaking: false
+new_feature: false
+bug_fix: true
+---
+Fix `RecordingClient` request-body recording for non-streaming bodies by buffering request payloads in memory before recording, avoiding channel-based timing issues.

rust-runtime/aws-smithy-http-client/src/test_util/body.rs

@@ -17,9 +17,17 @@ use tokio::sync::mpsc;
 ///
 /// Useful for sending data from another thread/task and test scenarios.
 pub(crate) fn channel_body() -> (Sender, SdkBody) {
+    channel_body_with_size_hint(None)
+}
+
+/// Create a `SdkBody` with an associated sender half and a known content length.
+///
+/// The `content_length` will be reported via the body's `size_hint()`, which is
+/// important for HTTP clients that rely on it for content-length headers or HTTP/2 framing.
+pub(crate) fn channel_body_with_size_hint(content_length: Option<u64>) -> (Sender, SdkBody) {
     let (tx, rx) = mpsc::channel(1);
     let sender = Sender { tx };
-    let ch_body = ChannelBody { rx };
+    let ch_body = ChannelBody { rx, content_length };
     (sender, SdkBody::from_body_1_x(ch_body))
 }
 
@@ -54,7 +62,8 @@ impl Sender {
 
 pin_project! {
     struct ChannelBody {
-        rx: mpsc::Receiver<Result<Frame<Bytes>, BoxError>>
+        rx: mpsc::Receiver<Result<Frame<Bytes>, BoxError>>,
+        content_length: Option<u64>,
     }
 }
 
@@ -75,7 +84,10 @@ impl http_body_1x::Body for ChannelBody {
     }
 
     fn size_hint(&self) -> SizeHint {
-        SizeHint::default()
+        match self.content_length {
+            Some(len) => SizeHint::with_exact(len),
+            None => SizeHint::default(),
+        }
     }
 }
 

rust-runtime/aws-smithy-http-client/src/test_util/dvr/record.rs

@@ -15,11 +15,12 @@ use aws_smithy_runtime_api::client::orchestrator::HttpRequest;
 use aws_smithy_runtime_api::client::runtime_components::RuntimeComponents;
 use aws_smithy_runtime_api::shared::IntoShared;
 use aws_smithy_types::body::SdkBody;
+use bytes::Bytes;
 use std::path::Path;
 use std::sync::atomic::{AtomicUsize, Ordering};
 use std::sync::{Arc, Mutex, MutexGuard};
-use std::{fs, io};
 use tokio::task::JoinHandle;
+use {std::fs, std::io};
 
 /// Recording client
 ///
@@ -28,17 +29,13 @@ use tokio::task::JoinHandle;
 /// # Example
 ///
 /// ```rust,ignore
-/// use aws_smithy_async::rt::sleep::default_async_sleep;
-/// use aws_smithy_runtime::client::http::hyper_014::default_connector;
 /// use aws_smithy_http_client::test_util::dvr::RecordingClient;
-/// use aws_smithy_runtime_api::client::http::HttpConnectorSettingsBuilder;
 /// use aws_sdk_s3::{Client, Config};
 ///
 /// #[tokio::test]
 /// async fn test_content_length_enforcement_is_not_applied_to_head_request() {
-///     let settings = HttpConnectorSettingsBuilder::default().build();
-///     let http_client = default_connector(&settings, default_async_sleep()).unwrap();
-///     let http_client = RecordingClient::new(http_client);
+///     // Create a recording client that wraps a default HTTPS connector
+///     let http_client = RecordingClient::https();
 ///
 ///     // Since we need to send a real request for this,
 ///     // you'll need to use your real credentials.
@@ -125,14 +122,57 @@ impl RecordingClient {
     }
 }
 
+/// Record a request body, preferring a buffered approach for non-streaming bodies.
+///
+/// For request bodies that have their content available as bytes, this buffers
+/// the body in memory to avoid timing issues with channel-based streaming.
+/// Streaming request bodies fall back to the channel-based recorder.
+fn record_request_body(
+    body: &mut SdkBody,
+    event_id: ConnectionId,
+    direction: Direction,
+    event_bus: Arc<Mutex<Vec<Event>>>,
+) {
+    if let Some(bytes) = body.bytes() {
+        let data = Bytes::copy_from_slice(bytes);
+        let mut events = event_bus.lock().unwrap();
+        events.push(Event {
+            connection_id: event_id,
+            action: Action::Data {
+                data: BodyData::from(data.clone()),
+                direction,
+            },
+        });
+        events.push(Event {
+            connection_id: event_id,
+            action: Action::Eof {
+                ok: true,
+                direction,
+            },
+        });
+        *body = SdkBody::from(data);
+    } else {
+        record_body(body, event_id, direction, event_bus);
+    }
+}
+
+/// Record body data using a channel-based streaming approach.
+///
+/// This spawns a task that reads from the original body and forwards data
+/// through a channel to a replacement body. This is used for response bodies
+/// and streaming request bodies.
 fn record_body(
     body: &mut SdkBody,
     event_id: ConnectionId,
     direction: Direction,
     event_bus: Arc<Mutex<Vec<Event>>>,
 ) -> JoinHandle<()> {
-    let (sender, output_body) = crate::test_util::body::channel_body();
+    // Capture the content length so the replacement body reports the correct size_hint()
+    let content_length = body.content_length();
+    let (sender, output_body) =
+        crate::test_util::body::channel_body_with_size_hint(content_length);
     let real_body = std::mem::replace(body, output_body);
+
     tokio::spawn(async move {
         let mut real_body = real_body;
         let mut sender = sender;
@@ -147,18 +187,17 @@ fn record_body(
                             direction,
                         },
                     });
-                    // This happens if the real connection is closed during recording.
-                    // Need to think more carefully if this is the correct thing to log in this
-                    // case.
+                    // Forward the data to the replacement body
                     if sender.send_data(data).await.is_err() {
                         event_bus.lock().unwrap().push(Event {
                             connection_id: event_id,
                             action: Action::Eof {
                                 direction: direction.opposite(),
                                 ok: false,
                             },
-                        })
-                    };
+                        });
+                        break;
+                    }
                 }
                 None => {
                     event_bus.lock().unwrap().push(Event {
@@ -190,45 +229,47 @@ fn record_body(
 impl HttpConnector for RecordingClient {
     fn call(&self, mut request: HttpRequest) -> HttpConnectorFuture {
         let event_id = self.next_id();
+
         // A request has three phases:
-        // 1. A "Request" phase. This is initial HTTP request, headers, & URI
-        // 2. A body phase. This may contain multiple data segments.
-        // 3. A finalization phase. An EOF of some sort is sent on the body to indicate that
-        // the channel should be closed.
+        // 1. A "Request" phase: initial HTTP request (headers, method, URI)
+        // 2. A body phase: may contain multiple data segments
+        // 3. A finalization phase: EOF is sent to indicate the body is complete
 
-        // Phase 1: the initial http request
+        // Phase 1: Record the initial request
         self.data.lock().unwrap().push(Event {
             connection_id: event_id,
             action: Action::Request {
                 request: Request::from(&request),
             },
         });
 
-        // Phase 2: Swap out the real request body for one that will log all traffic that passes
-        // through it
-        // This will also handle phase three when the request body runs out of data.
-        record_body(
+        // Phase 2 & 3: Record the request body
+        // Use buffered recording for request bodies to avoid race conditions
+        // with channel-based streaming that can cause HTTP/2 framing issues.
+        record_request_body(
             request.body_mut(),
             event_id,
             Direction::Request,
             self.data.clone(),
         );
+
         let events = self.data.clone();
-        // create a channel we'll use to stream the data while reading it
         let resp_fut = self.inner.call(request);
+
         let fut = async move {
             let resp = resp_fut.await;
             match resp {
                 Ok(mut resp) => {
-                    // push the initial response event
+                    // Record the initial response
                     events.lock().unwrap().push(Event {
                         connection_id: event_id,
                         action: Action::Response {
                             response: Ok(Response::from(&resp)),
                         },
                     });
 
-                    // instrument the body and record traffic
+                    // Record the response body using channel-based streaming
+                    // (this works fine for responses since we're consuming, not producing)
                     record_body(resp.body_mut(), event_id, Direction::Response, events);
                     Ok(resp)
                 }