Aligning http-1x checksum updates with checksum changes from #4200 (#4204)

## Motivation and Context
<!--- Why is this change required? What problem does it solve? -->
<!--- If it fixes an open issue, please link to the issue here -->
There were some conflicts around checksum changes made in
#4200 while merging main
back into the `feature/http-1.x` branch. This PR reconciles those
changes

----

_By submitting this pull request, I confirm that you can use, modify,
copy, and redistribute this contribution, under the terms of your
choice._

Author: landonxjames

aws/rust-runtime/aws-runtime/src/content_encoding.rs

@@ -382,6 +382,17 @@ where
                         tracing::trace!(
                             "No more chunk data, writing CRLF + CHUNK_TERMINATOR to end the data, and the first trailer frame"
                         );
+
+                        // We exhausted the body data, now check if the length is correct
+                        if let Err(poll_stream_len_err) =
+                            http_1x_utils::check_for_stream_length_mismatch(
+                                *this.inner_body_bytes_read_so_far as u64,
+                                this.options.stream_length,
+                            )
+                        {
+                            return poll_stream_len_err;
+                        }
+
                         *this.state = AwsChunkedBodyState::WritingTrailers;
                         let trailers = frame.trailers_ref();
 
@@ -436,15 +447,19 @@ where
             Poll::Ready(None) => {
                 let trailers = match *this.state {
                     AwsChunkedBodyState::WritingChunk => {
-                        let actual_stream_length = *this.inner_body_bytes_read_so_far as u64;
-                        let expected_stream_length = this.options.stream_length;
-                        if actual_stream_length != expected_stream_length {
-                            let err = Box::new(AwsChunkedBodyError::StreamLengthMismatch {
-                                actual: actual_stream_length,
-                                expected: expected_stream_length,
-                            });
-                            return Poll::Ready(Some(Err(err)));
-                        };
+                        // We exhausted the body data, now check if the length is correct
+                        if let Err(poll_stream_len_err) =
+                            http_1x_utils::check_for_stream_length_mismatch(
+                                *this.inner_body_bytes_read_so_far as u64,
+                                this.options.stream_length,
+                            )
+                        {
+                            return poll_stream_len_err;
+                        }
+
+                        // Since we exhausted the body data, but are still in the WritingChunk state we did
+                        // not poll any trailer frames and we write the CRLF + Chunk terminator to begin the
+                        // trailer section plus a single final CRLF to end the (empty) trailer section
                         let mut trailers = BytesMut::with_capacity(7);
                         trailers.extend_from_slice(
                             &[CRLF_RAW, CHUNK_TERMINATOR_RAW, CRLF_RAW].concat(),
@@ -473,8 +488,10 @@ where
 }
 /// Utility functions to help with the [http_body_1x::Body] trait implementation
 mod http_1x_utils {
+    use std::task::Poll;
+
     use super::{CRLF_RAW, TRAILER_SEPARATOR};
-    use bytes::BytesMut;
+    use bytes::{Bytes, BytesMut};
     use http_1x::{HeaderMap, HeaderName};
 
     /// Writes trailers out into a `string` and then converts that `String` to a `Bytes` before
@@ -531,6 +548,25 @@ mod http_1x_utils {
             None => 0,
         }
     }
+
+    /// This is an ugly return type, but in practice it just returns `Ok(())` if the values match
+    /// and `Err(Poll::Ready(Some(Err(AwsChunkedBodyError::StreamLengthMismatch))))` if they don't
+    #[allow(clippy::type_complexity)]
+    pub(super) fn check_for_stream_length_mismatch(
+        actual_stream_length: u64,
+        expected_stream_length: u64,
+    ) -> Result<(), Poll<Option<Result<http_body_1x::Frame<Bytes>, aws_smithy_types::body::Error>>>>
+    {
+        if actual_stream_length != expected_stream_length {
+            let err = Box::new(super::AwsChunkedBodyError::StreamLengthMismatch {
+                actual: actual_stream_length,
+                expected: expected_stream_length,
+            });
+            return Err(Poll::Ready(Some(Err(err))));
+        };
+
+        Ok(())
+    }
 }
 
 /// Errors related to `AwsChunkedBody`

aws/sdk-codegen/src/main/kotlin/software/amazon/smithy/rustsdk/IntegrationTestDependencies.kt

@@ -21,7 +21,7 @@ import software.amazon.smithy.rust.codegen.core.rustlang.CargoDependency.Compani
 import software.amazon.smithy.rust.codegen.core.rustlang.CargoDependency.Companion.Hound
 import software.amazon.smithy.rust.codegen.core.rustlang.CargoDependency.Companion.Http1x
 import software.amazon.smithy.rust.codegen.core.rustlang.CargoDependency.Companion.HttpBody1x
-import software.amazon.smithy.rust.codegen.core.rustlang.CargoDependency.Companion.HttpBodyUtil
+import software.amazon.smithy.rust.codegen.core.rustlang.CargoDependency.Companion.HttpBodyUtil01x
 import software.amazon.smithy.rust.codegen.core.rustlang.CargoDependency.Companion.SerdeJson
 import software.amazon.smithy.rust.codegen.core.rustlang.CargoDependency.Companion.Smol
 import software.amazon.smithy.rust.codegen.core.rustlang.CargoDependency.Companion.TempFile
@@ -187,7 +187,7 @@ class S3TestDependencies(private val runtimeConfig: RuntimeConfig) : LibRsCustom
             addDependency(FuturesUtil.toDevDependency())
             addDependency(HdrHistogram)
             addDependency(HttpBody1x.toDevDependency().copy(optional = false))
-            addDependency(HttpBodyUtil.toDevDependency().copy(optional = false))
+            addDependency(HttpBodyUtil01x.toDevDependency().copy(optional = false))
             addDependency(Smol)
             addDependency(TempFile)
             addDependency(TracingAppender)

codegen-core/src/main/kotlin/software/amazon/smithy/rust/codegen/core/rustlang/CargoDependency.kt

@@ -370,8 +370,6 @@ data class CargoDependency(
         val HttpBodyUtil01x: CargoDependency =
             CargoDependency("http-body-util", CratesIo("0.1.3"))
 
-        val HttpBodyUtil: CargoDependency = CargoDependency("http-body-util", CratesIo("0.1.3"))
-
         fun smithyAsync(runtimeConfig: RuntimeConfig) = runtimeConfig.smithyRuntimeCrate("smithy-async")
 
         fun smithyCbor(runtimeConfig: RuntimeConfig) = runtimeConfig.smithyRuntimeCrate("smithy-cbor")

rust-runtime/aws-smithy-checksums/src/body/cache.rs

@@ -5,7 +5,7 @@
 
 //! Checksum caching functionality.
 
-use http::HeaderMap;
+use http_1x::HeaderMap;
 use std::sync::{Arc, Mutex};
 
 /// A cache for storing previously calculated checksums.

rust-runtime/aws-smithy-checksums/src/body/calculate.rs

@@ -49,6 +49,28 @@ impl ChecksumBody<SdkBody> {
             cache: Some(cache),
         }
     }
+
+    // It would be nicer if this could take &self, but I couldn't make that
+    // work out with the Pin/Projection types, so its a static method for now
+    fn extract_or_set_cached_headers(
+        maybe_cache: &Option<ChecksumCache>,
+        checksum: Box<dyn HttpChecksum>,
+    ) -> http_1x::HeaderMap {
+        let calculated_headers = checksum.headers();
+        if let Some(cache) = maybe_cache {
+            if let Some(cached_headers) = cache.get() {
+                if cached_headers != calculated_headers {
+                    warn!(cached = ?cached_headers, calculated = ?calculated_headers, "calculated checksum differs from cached checksum!");
+                }
+                cached_headers
+            } else {
+                cache.set(calculated_headers.clone());
+                calculated_headers
+            }
+        } else {
+            calculated_headers
+        }
+    }
 }
 
 impl http_body_1x::Body for ChecksumBody<SdkBody> {
@@ -72,7 +94,7 @@ impl http_body_1x::Body for ChecksumBody<SdkBody> {
                 } else {
                     // Add checksum trailer to other trailers if necessary
                     let checksum_headers = if let Some(checksum) = this.checksum.take() {
-                        checksum.headers()
+                        ChecksumBody::extract_or_set_cached_headers(this.cache, checksum)
                     } else {
                         return Poll::Ready(None);
                     };
@@ -91,7 +113,7 @@ impl http_body_1x::Body for ChecksumBody<SdkBody> {
                 // trailers on the body) we write them here
                 if !*this.written_trailers {
                     let checksum_headers = if let Some(checksum) = this.checksum.take() {
-                        checksum.headers()
+                        ChecksumBody::extract_or_set_cached_headers(this.cache, checksum)
                     } else {
                         return Poll::Ready(None);
                     };
@@ -105,46 +127,6 @@ impl http_body_1x::Body for ChecksumBody<SdkBody> {
         };
         poll_res
     }
-
-    fn poll_trailers(
-        self: Pin<&mut Self>,
-        cx: &mut Context<'_>,
-    ) -> Poll<Result<Option<HeaderMap>, Self::Error>> {
-        let this = self.project();
-        let poll_res = this.body.poll_trailers(cx);
-
-        if let Poll::Ready(Ok(maybe_inner_trailers)) = poll_res {
-            let checksum_headers = if let Some(checksum) = this.checksum.take() {
-                let calculated_headers = checksum.headers();
-
-                if let Some(cache) = this.cache {
-                    if let Some(cached_headers) = cache.get() {
-                        if cached_headers != calculated_headers {
-                            warn!(cached = ?cached_headers, calculated = ?calculated_headers, "calculated checksum differs from cached checksum!");
-                        }
-                        cached_headers
-                    } else {
-                        cache.set(calculated_headers.clone());
-                        calculated_headers
-                    }
-                } else {
-                    calculated_headers
-                }
-            } else {
-                return Poll::Ready(Ok(None));
-            };
-
-            return match maybe_inner_trailers {
-                Some(inner_trailers) => Poll::Ready(Ok(Some(append_merge_header_maps(
-                    inner_trailers,
-                    checksum_headers,
-                )))),
-                None => Poll::Ready(Ok(Some(checksum_headers))),
-            };
-        }
-
-        poll_res
-    }
 }
 
 #[cfg(test)]